From 19420cc01d5ad72afbb48f7c4406b50e3467b983 Mon Sep 17 00:00:00 2001 From: Azreen Jamal Date: Fri, 6 Mar 2026 15:26:13 +0800 Subject: [PATCH] feat: add pi-worker service - autonomous Asana task executor & project improvement agent - Asana REST API client for task polling, section management, and commenting - Pi CLI executor for running tasks with specialist agent teams - Task loop: polls Asana To Do/Pi Worker sections, executes tasks, reports results - Improvement loop: rotates through 8 improvement categories (code quality, security, docs, testing, etc) - Health server on :8787 - File-based mutex locks with stale lock detection - Docker + docker-compose deployment config - deploy.sh for one-command deployment to cr-server-new via SSH/Incus --- pi-worker/.env.sample | 22 ++ pi-worker/Dockerfile | 23 ++ pi-worker/README.md | 50 ++++ pi-worker/bun.lock | 19 ++ pi-worker/deploy.sh | 46 ++++ pi-worker/docker-compose.yml | 24 ++ pi-worker/logs/.gitkeep | 0 pi-worker/logs/pi-worker-2026-03-06.log | 11 + pi-worker/package.json | 13 ++ pi-worker/src/asana/client.ts | 126 ++++++++++ pi-worker/src/asana/types.ts | 38 +++ pi-worker/src/config.ts | 54 +++++ pi-worker/src/health/server.ts | 51 ++++ pi-worker/src/index.ts | 60 +++++ pi-worker/src/logger.ts | 36 +++ pi-worker/src/pi/executor.ts | 247 ++++++++++++++++++++ pi-worker/src/pi/types.ts | 32 +++ pi-worker/src/scheduler/improvement-loop.ts | 165 +++++++++++++ pi-worker/src/scheduler/lock.ts | 47 ++++ pi-worker/src/scheduler/task-loop.ts | 186 +++++++++++++++ pi-worker/tsconfig.json | 15 ++ 21 files changed, 1265 insertions(+) create mode 100644 pi-worker/.env.sample create mode 100644 pi-worker/Dockerfile create mode 100644 pi-worker/README.md create mode 100644 pi-worker/bun.lock create mode 100755 pi-worker/deploy.sh create mode 100644 pi-worker/docker-compose.yml create mode 100644 pi-worker/logs/.gitkeep create mode 100644 pi-worker/logs/pi-worker-2026-03-06.log create mode 100644 pi-worker/package.json create mode 100644 pi-worker/src/asana/client.ts create mode 100644 pi-worker/src/asana/types.ts create mode 100644 pi-worker/src/config.ts create mode 100644 pi-worker/src/health/server.ts create mode 100644 pi-worker/src/index.ts create mode 100644 pi-worker/src/logger.ts create mode 100644 pi-worker/src/pi/executor.ts create mode 100644 pi-worker/src/pi/types.ts create mode 100644 pi-worker/src/scheduler/improvement-loop.ts create mode 100644 pi-worker/src/scheduler/lock.ts create mode 100644 pi-worker/src/scheduler/task-loop.ts create mode 100644 pi-worker/tsconfig.json diff --git a/pi-worker/.env.sample b/pi-worker/.env.sample new file mode 100644 index 0000000..50863c1 --- /dev/null +++ b/pi-worker/.env.sample @@ -0,0 +1,22 @@ +# Asana Integration +ASANA_ACCESS_TOKEN=your_asana_token +ASANA_PROJECT_GID= # Leave empty for auto-discovery +ASANA_WORKSPACE_GID= # Leave empty for auto-discovery + +# Pi Agent +ANTHROPIC_API_KEY=your_anthropic_key +PI_BIN=pi # Path to pi binary +PI_MODEL=claude-sonnet-4-6 # Model to use + +# Project +TARGET_PROJECT_PATH=/opt/charityright # Path to project codebase +EXTENSIONS_PATH=../extensions # Path to pi extensions +AGENTS_PATH=../.pi/agents # Path to agent definitions +LOG_DIR=./logs # Log directory + +# Timing +POLL_INTERVAL_SEC=120 # Check Asana every 2 minutes +IMPROVEMENT_INTERVAL_SEC=3600 # Run improvements every hour + +# Health +HEALTH_PORT=8787 # Health check endpoint port diff --git a/pi-worker/Dockerfile b/pi-worker/Dockerfile new file mode 100644 index 0000000..62a341c --- /dev/null +++ b/pi-worker/Dockerfile @@ -0,0 +1,23 @@ +FROM oven/bun:1-alpine AS base +WORKDIR /app + +# Install pi CLI and dependencies +RUN apk add --no-cache git curl openssh-client + +# Copy package files +COPY package.json bun.lock* ./ +RUN bun install --frozen-lockfile || bun install + +# Copy source +COPY . . + +# Create logs directory +RUN mkdir -p logs + +# Health check +HEALTHCHECK --interval=30s --timeout=10s --retries=3 \ + CMD curl -f http://localhost:8787/health || exit 1 + +EXPOSE 8787 + +CMD ["bun", "run", "src/index.ts"] diff --git a/pi-worker/README.md b/pi-worker/README.md new file mode 100644 index 0000000..987f51a --- /dev/null +++ b/pi-worker/README.md @@ -0,0 +1,50 @@ +# 🤖 Pi Worker + +Autonomous agent that runs inside `cr-server-new`, picks up Asana tasks, and performs automated project improvements using the Pi improved agent. + +## What It Does + +1. **Asana Task Execution**: Polls Asana for tasks in "To Do" or "Pi Worker" sections, executes them using the Pi CLI agent, and reports results back as comments +2. **Autonomous Improvements**: Periodically analyzes the project codebase and makes improvements (code quality, security, docs, testing, performance, type safety) + +## Architecture + +``` +Asana Board ←→ Pi Worker (cr-server-new) ←→ Pi CLI Agent ←→ Codebase + ↕ + Health Server (:8787) +``` + +## Setup + +1. Copy `.env.sample` to `.env` and fill in credentials +2. `bun install` +3. `bun run dev` (local) or `bash deploy.sh` (production) + +## Configuration + +| Variable | Description | Default | +|----------|-------------|---------| +| `ASANA_ACCESS_TOKEN` | Asana personal access token | Required | +| `ASANA_PROJECT_GID` | Project to poll (auto-discovered if empty) | Auto | +| `PI_MODEL` | AI model to use | `claude-sonnet-4-6` | +| `POLL_INTERVAL_SEC` | Asana polling interval | `120` | +| `IMPROVEMENT_INTERVAL_SEC` | Auto-improvement interval | `3600` | + +## Asana Board Setup + +Create these sections in your Asana project: +- **To Do** — Tasks for Pi Worker to pick up +- **In Progress** — Currently being executed +- **Done** — Completed tasks +- **Pi Worker** (optional) — Dedicated section for Pi Worker tasks + +## Health Check + +```bash +curl http://localhost:8787/health +``` + +## Logs + +Logs are written to `./logs/pi-worker-YYYY-MM-DD.log` and stdout. diff --git a/pi-worker/bun.lock b/pi-worker/bun.lock new file mode 100644 index 0000000..61ee9f4 --- /dev/null +++ b/pi-worker/bun.lock @@ -0,0 +1,19 @@ +{ + "lockfileVersion": 1, + "configVersion": 1, + "workspaces": { + "": { + "name": "pi-worker", + "devDependencies": { + "bun-types": "^1.3.10", + }, + }, + }, + "packages": { + "@types/node": ["@types/node@25.3.5", "", { "dependencies": { "undici-types": "~7.18.0" } }, "sha512-oX8xrhvpiyRCQkG1MFchB09f+cXftgIXb3a7UUa4Y3wpmZPw5tyZGTLWhlESOLq1Rq6oDlc8npVU2/9xiCuXMA=="], + + "bun-types": ["bun-types@1.3.10", "", { "dependencies": { "@types/node": "*" } }, "sha512-tcpfCCl6XWo6nCVnpcVrxQ+9AYN1iqMIzgrSKYMB/fjLtV2eyAVEg7AxQJuCq/26R6HpKWykQXuSOq/21RYcbg=="], + + "undici-types": ["undici-types@7.18.2", "", {}, "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w=="], + } +} diff --git a/pi-worker/deploy.sh b/pi-worker/deploy.sh new file mode 100755 index 0000000..e876861 --- /dev/null +++ b/pi-worker/deploy.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Deploy pi-worker to cr-server-new +SERVER="root@159.195.60.33" +CONTAINER="cr-server-new" +DEPLOY_PATH="/opt/pi-worker" + +echo "🚀 Deploying Pi Worker to cr-server-new..." + +# Sync files to server +echo "📦 Syncing files..." +rsync -avz --exclude=node_modules --exclude=logs --exclude=.env \ + ./ ${SERVER}:${DEPLOY_PATH}/ + +# Copy .env if it exists locally +if [ -f .env ]; then + echo "🔐 Syncing .env..." + rsync -avz .env ${SERVER}:${DEPLOY_PATH}/.env +fi + +# Push into container and build +echo "🔨 Building inside cr-server-new..." +ssh ${SERVER} << 'EOF' + incus file push -r /root/pi-worker cr-server-new/opt/ + incus exec cr-server-new -- bash -c " + cd /opt/pi-worker + bun install + echo '✅ Dependencies installed' + " +EOF + +echo "🏃 Starting Pi Worker..." +ssh ${SERVER} << 'EOF' + incus exec cr-server-new -- bash -c " + cd /opt/pi-worker + # Stop existing instance + pkill -f 'bun.*pi-worker' || true + # Start in background + nohup bun run src/index.ts > /var/log/pi-worker.log 2>&1 & + echo '✅ Pi Worker started (PID: $!)' + " +EOF + +echo "🎉 Deployment complete!" +echo "Health check: ssh ${SERVER} 'incus exec ${CONTAINER} -- curl -s http://localhost:8787/health'" diff --git a/pi-worker/docker-compose.yml b/pi-worker/docker-compose.yml new file mode 100644 index 0000000..317881a --- /dev/null +++ b/pi-worker/docker-compose.yml @@ -0,0 +1,24 @@ +version: "3.8" + +services: + pi-worker: + build: . + container_name: pi-worker + restart: unless-stopped + env_file: .env + ports: + - "8787:8787" + volumes: + - ./logs:/app/logs + - pi-sessions:/app/sessions + environment: + - NODE_ENV=production + deploy: + resources: + limits: + memory: 2G + reservations: + memory: 512M + +volumes: + pi-sessions: diff --git a/pi-worker/logs/.gitkeep b/pi-worker/logs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/pi-worker/logs/pi-worker-2026-03-06.log b/pi-worker/logs/pi-worker-2026-03-06.log new file mode 100644 index 0000000..4ddaa49 --- /dev/null +++ b/pi-worker/logs/pi-worker-2026-03-06.log @@ -0,0 +1,11 @@ +[2026-03-06T07:25:24.241Z] [INFO] [main] Configuration loaded {"model":"claude-sonnet-4-6","pollInterval":"120s","improvementInterval":"3600s","targetProject":"/Users/azreenjamal/pi-vs-claude-code","healthPort":8787} +[2026-03-06T07:25:24.249Z] [INFO] [health] Health server listening on :8787 +[2026-03-06T07:25:24.249Z] [INFO] [task-loop] Starting Asana task polling loop {"interval":"120s","project":""} +[2026-03-06T07:25:24.249Z] [INFO] [task-loop] No project GID configured, discovering... +[2026-03-06T07:25:24.892Z] [INFO] [task-loop] Found 1 workspace(s) [{"gid":"342607773496276","resource_type":"workspace","name":"charityright.org.uk"}] +[2026-03-06T07:25:25.419Z] [INFO] [task-loop] Workspace "charityright.org.uk" has 6 project(s) [{"gid":"1208646974991820","name":"Bug Intake"},{"gid":"1208713370972799","name":"Omair's Todos"},{"gid":"1211205155206401","name":"Automation of event fundraising (Team365)"},{"gid":"1211205155206405","name":"CRM housekeeping"},{"gid":"1211205155206409","name":"Rebuild website & donation platform"},{"gid":"1211205155206413","name":"Agentic AI content mechanism (SEO/AI discoverability)"}] +[2026-03-06T07:25:25.419Z] [INFO] [task-loop] Auto-selected project: Bug Intake (1208646974991820) +[2026-03-06T07:25:25.421Z] [INFO] [task-loop] Polling for Asana tasks... +[2026-03-06T07:25:26.194Z] [INFO] [task-loop] No pending tasks found +[2026-03-06T07:25:26.196Z] [INFO] [improvement-loop] Starting autonomous improvement loop {"interval":"3600s","categories":8} +[2026-03-06T07:25:26.197Z] [INFO] [main] Pi Worker is fully operational 🚀 diff --git a/pi-worker/package.json b/pi-worker/package.json new file mode 100644 index 0000000..97aef96 --- /dev/null +++ b/pi-worker/package.json @@ -0,0 +1,13 @@ +{ + "name": "pi-worker", + "version": "1.0.0", + "private": true, + "type": "module", + "scripts": { + "start": "bun run src/index.ts", + "dev": "bun --watch src/index.ts" + }, + "devDependencies": { + "bun-types": "^1.3.10" + } +} diff --git a/pi-worker/src/asana/client.ts b/pi-worker/src/asana/client.ts new file mode 100644 index 0000000..15179dc --- /dev/null +++ b/pi-worker/src/asana/client.ts @@ -0,0 +1,126 @@ +import type { AsanaTask, AsanaSection, AsanaProject, AsanaComment } from "./types.js"; +import { logger } from "../logger.js"; + +const ASANA_BASE = "https://app.asana.com/api/1.0"; + +export class AsanaClient { + private token: string; + private headers: Record; + + constructor(token: string) { + this.token = token; + this.headers = { + Authorization: `Bearer ${this.token}`, + "Content-Type": "application/json", + Accept: "application/json", + }; + } + + private async request(path: string, options: RequestInit = {}): Promise { + const url = `${ASANA_BASE}${path}`; + const res = await fetch(url, { + ...options, + headers: { ...this.headers, ...options.headers }, + }); + + if (!res.ok) { + const body = await res.text(); + throw new Error(`Asana API ${res.status}: ${body}`); + } + + const json = (await res.json()) as { data: T }; + return json.data; + } + + // Get all workspaces + async getWorkspaces(): Promise> { + return this.request("/workspaces"); + } + + // Get all projects in a workspace + async getProjects(workspaceGid: string): Promise { + return this.request(`/workspaces/${workspaceGid}/projects?opt_fields=name`); + } + + // Get all sections in a project + async getSections(projectGid: string): Promise { + return this.request(`/projects/${projectGid}/sections`); + } + + // Get tasks in a project + async getProjectTasks(projectGid: string): Promise { + return this.request( + `/projects/${projectGid}/tasks?opt_fields=name,notes,completed,assignee.name,tags.name,custom_fields,due_on,created_at,modified_at,memberships.project.name,memberships.section.name` + ); + } + + // Get tasks in a section + async getSectionTasks(sectionGid: string): Promise { + return this.request( + `/sections/${sectionGid}/tasks?opt_fields=name,notes,completed,assignee.name,tags.name,custom_fields,due_on,created_at,modified_at,memberships.project.name,memberships.section.name` + ); + } + + // Get single task details + async getTask(taskGid: string): Promise { + return this.request( + `/tasks/${taskGid}?opt_fields=name,notes,completed,assignee.name,tags.name,custom_fields,due_on,created_at,modified_at,memberships.project.name,memberships.section.name` + ); + } + + // Move task to a section + async moveTaskToSection(taskGid: string, sectionGid: string): Promise { + await this.request(`/sections/${sectionGid}/addTask`, { + method: "POST", + body: JSON.stringify({ data: { task: taskGid } }), + }); + } + + // Update task (mark complete, change name, etc) + async updateTask(taskGid: string, updates: Record): Promise { + return this.request(`/tasks/${taskGid}`, { + method: "PUT", + body: JSON.stringify({ data: updates }), + }); + } + + // Add a comment to a task + async addComment(taskGid: string, text: string): Promise { + return this.request(`/tasks/${taskGid}/stories`, { + method: "POST", + body: JSON.stringify({ data: { text } }), + }); + } + + // Get comments on a task + async getComments(taskGid: string): Promise { + return this.request(`/tasks/${taskGid}/stories?opt_fields=text,created_at`); + } + + // Create a new task + async createTask(projectGid: string, data: { name: string; notes?: string; due_on?: string }): Promise { + return this.request("/tasks", { + method: "POST", + body: JSON.stringify({ + data: { + ...data, + projects: [projectGid], + }, + }), + }); + } + + // Helper: Find section by name + async findSection(projectGid: string, sectionName: string): Promise { + const sections = await this.getSections(projectGid); + return sections.find((s) => s.name.toLowerCase().includes(sectionName.toLowerCase())) || null; + } + + // Helper: Get incomplete tasks from a section by name + async getIncompleteTasks(projectGid: string, sectionName: string): Promise { + const section = await this.findSection(projectGid, sectionName); + if (!section) return []; + const tasks = await this.getSectionTasks(section.gid); + return tasks.filter((t) => !t.completed); + } +} diff --git a/pi-worker/src/asana/types.ts b/pi-worker/src/asana/types.ts new file mode 100644 index 0000000..1a8bc41 --- /dev/null +++ b/pi-worker/src/asana/types.ts @@ -0,0 +1,38 @@ +export interface AsanaTask { + gid: string; + name: string; + notes: string; + completed: boolean; + assignee: { gid: string; name: string } | null; + projects: Array<{ gid: string; name: string }>; + tags: Array<{ gid: string; name: string }>; + custom_fields: Array<{ + gid: string; + name: string; + display_value: string | null; + enum_value: { name: string } | null; + }>; + due_on: string | null; + created_at: string; + modified_at: string; + memberships: Array<{ + project: { gid: string; name: string }; + section: { gid: string; name: string }; + }>; +} + +export interface AsanaSection { + gid: string; + name: string; +} + +export interface AsanaProject { + gid: string; + name: string; +} + +export interface AsanaComment { + gid: string; + text: string; + created_at: string; +} diff --git a/pi-worker/src/config.ts b/pi-worker/src/config.ts new file mode 100644 index 0000000..7218045 --- /dev/null +++ b/pi-worker/src/config.ts @@ -0,0 +1,54 @@ +// Pi Worker Configuration +// Loads and validates environment variables + +export interface Config { + // Asana + asanaAccessToken: string; + asanaProjectGid: string; + asanaWorkspaceGid: string; + + // Pi Agent + piBin: string; + piModel: string; + anthropicApiKey: string; + + // Paths + targetProjectPath: string; + extensionsPath: string; + agentsPath: string; + logDir: string; + + // Timing + pollIntervalSec: number; + improvementIntervalSec: number; + healthPort: number; +} + +function requireEnv(key: string): string { + const value = process.env[key]; + if (!value) { + throw new Error(`Missing required environment variable: ${key}`); + } + return value; +} + +export function loadConfig(): Config { + return { + asanaAccessToken: requireEnv("ASANA_ACCESS_TOKEN"), + asanaProjectGid: process.env.ASANA_PROJECT_GID || "", + asanaWorkspaceGid: process.env.ASANA_WORKSPACE_GID || "", + + piBin: process.env.PI_BIN || "pi", + piModel: process.env.PI_MODEL || "claude-sonnet-4-6", + anthropicApiKey: requireEnv("ANTHROPIC_API_KEY"), + + targetProjectPath: process.env.TARGET_PROJECT_PATH || process.cwd(), + extensionsPath: process.env.EXTENSIONS_PATH || "../extensions", + agentsPath: process.env.AGENTS_PATH || "../.pi/agents", + logDir: process.env.LOG_DIR || "./logs", + + pollIntervalSec: parseInt(process.env.POLL_INTERVAL_SEC || "120"), + improvementIntervalSec: parseInt(process.env.IMPROVEMENT_INTERVAL_SEC || "3600"), + healthPort: parseInt(process.env.HEALTH_PORT || "8787"), + }; +} diff --git a/pi-worker/src/health/server.ts b/pi-worker/src/health/server.ts new file mode 100644 index 0000000..68616e1 --- /dev/null +++ b/pi-worker/src/health/server.ts @@ -0,0 +1,51 @@ +import { logger } from "../logger.js"; + +interface HealthStatus { + status: "ok" | "degraded" | "error"; + uptime: number; + startedAt: string; + taskLoop: boolean; + improvementLoop: boolean; + lastPoll?: string; +} + +let healthStatus: HealthStatus = { + status: "ok", + uptime: 0, + startedAt: new Date().toISOString(), + taskLoop: false, + improvementLoop: false, +}; + +export function updateHealth(updates: Partial) { + healthStatus = { ...healthStatus, ...updates }; +} + +export function startHealthServer(port: number) { + const startTime = Date.now(); + + const server = Bun.serve({ + port, + fetch(req) { + const url = new URL(req.url); + + if (url.pathname === "/health") { + return Response.json({ + ...healthStatus, + uptime: Math.floor((Date.now() - startTime) / 1000), + }); + } + + if (url.pathname === "/") { + return new Response("Pi Worker is running. GET /health for status.", { + headers: { "Content-Type": "text/plain" }, + }); + } + + return new Response("Not Found", { status: 404 }); + }, + }); + + logger.info("health", `Health server listening on :${port}`); + return server; +} diff --git a/pi-worker/src/index.ts b/pi-worker/src/index.ts new file mode 100644 index 0000000..9c33b54 --- /dev/null +++ b/pi-worker/src/index.ts @@ -0,0 +1,60 @@ +import { loadConfig } from "./config.js"; +import { initLogger, logger } from "./logger.js"; +import { TaskLoop } from "./scheduler/task-loop.js"; +import { ImprovementLoop } from "./scheduler/improvement-loop.js"; +import { startHealthServer, updateHealth } from "./health/server.js"; + +async function main() { + console.log(` +╔══════════════════════════════════════╗ +║ 🤖 Pi Worker v1.0.0 ║ +║ Autonomous Asana Task Executor ║ +║ & Project Improvement Agent ║ +╚══════════════════════════════════════╝ +`); + + // Load config + const config = loadConfig(); + initLogger(config.logDir); + + logger.info("main", "Configuration loaded", { + model: config.piModel, + pollInterval: `${config.pollIntervalSec}s`, + improvementInterval: `${config.improvementIntervalSec}s`, + targetProject: config.targetProjectPath, + healthPort: config.healthPort, + }); + + // Start health server + startHealthServer(config.healthPort); + updateHealth({ status: "ok" }); + + // Start task loop (Asana polling) + const taskLoop = new TaskLoop(config); + await taskLoop.start(); + updateHealth({ taskLoop: true }); + + // Start improvement loop (autonomous project fixes) + const improvementLoop = new ImprovementLoop(config); + await improvementLoop.start(); + updateHealth({ improvementLoop: true }); + + logger.info("main", "Pi Worker is fully operational 🚀"); + + // Graceful shutdown + const shutdown = () => { + logger.info("main", "Shutting down..."); + taskLoop.stop(); + improvementLoop.stop(); + updateHealth({ status: "degraded" }); + process.exit(0); + }; + + process.on("SIGINT", shutdown); + process.on("SIGTERM", shutdown); +} + +main().catch((error) => { + console.error("Fatal error:", error); + process.exit(1); +}); diff --git a/pi-worker/src/logger.ts b/pi-worker/src/logger.ts new file mode 100644 index 0000000..90ad3d1 --- /dev/null +++ b/pi-worker/src/logger.ts @@ -0,0 +1,36 @@ +import { mkdirSync, appendFileSync, existsSync } from "fs"; +import { join } from "path"; + +type LogLevel = "INFO" | "WARN" | "ERROR" | "DEBUG"; + +let logDir = "./logs"; + +export function initLogger(dir: string) { + logDir = dir; + if (!existsSync(logDir)) { + mkdirSync(logDir, { recursive: true }); + } +} + +function formatLog(level: LogLevel, component: string, message: string, data?: any): string { + const ts = new Date().toISOString(); + const base = `[${ts}] [${level}] [${component}] ${message}`; + return data ? `${base} ${JSON.stringify(data)}` : base; +} + +function log(level: LogLevel, component: string, message: string, data?: any) { + const line = formatLog(level, component, message, data); + console.log(line); + + try { + const logFile = join(logDir, `pi-worker-${new Date().toISOString().split("T")[0]}.log`); + appendFileSync(logFile, line + "\n"); + } catch {} +} + +export const logger = { + info: (component: string, message: string, data?: any) => log("INFO", component, message, data), + warn: (component: string, message: string, data?: any) => log("WARN", component, message, data), + error: (component: string, message: string, data?: any) => log("ERROR", component, message, data), + debug: (component: string, message: string, data?: any) => log("DEBUG", component, message, data), +}; diff --git a/pi-worker/src/pi/executor.ts b/pi-worker/src/pi/executor.ts new file mode 100644 index 0000000..c2b2070 --- /dev/null +++ b/pi-worker/src/pi/executor.ts @@ -0,0 +1,247 @@ +import { spawn, type Subprocess } from "bun"; +import type { PiExecutionResult, PiEvent } from "./types.js"; +import { logger } from "../logger.js"; +import type { Config } from "../config.js"; + +const PI_TIMEOUT_MS = 10 * 60 * 1000; // 10 minutes max per task + +export class PiExecutor { + private config: Config; + private runningProcess: Subprocess | null = null; + + constructor(config: Config) { + this.config = config; + } + + async execute(task: string, context?: string): Promise { + const startTime = Date.now(); + let output = ""; + let toolCalls = 0; + + const fullPrompt = context ? `${context}\n\nTask: ${task}` : task; + + const args = [ + this.config.piBin, + "--mode", "json", + "-p", + "--no-extensions", + "--model", this.config.piModel, + "--tools", "bash,read,write,edit,grep,find,ls", + "--thinking", "off", + "--append-system-prompt", this.buildSystemPrompt(), + fullPrompt, + ]; + + logger.info("pi-executor", `Executing task: ${task.substring(0, 100)}...`, { + model: this.config.piModel, + cwd: this.config.targetProjectPath, + }); + + try { + const proc = spawn({ + cmd: args, + cwd: this.config.targetProjectPath, + stdout: "pipe", + stderr: "pipe", + env: { + ...process.env, + ANTHROPIC_API_KEY: this.config.anthropicApiKey, + }, + }); + + this.runningProcess = proc; + + // Set timeout + const timeout = setTimeout(() => { + logger.warn("pi-executor", "Task timed out, killing process"); + proc.kill(); + }, PI_TIMEOUT_MS); + + // Read stdout line by line + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() || ""; + + for (const line of lines) { + if (!line.trim()) continue; + try { + const event: PiEvent = JSON.parse(line); + + if (event.type === "message_update" || event.type === "message_end") { + output += event.content || ""; + } + if (event.type === "tool_execution_start") { + toolCalls++; + } + } catch { + // Not JSON, append as raw output + output += line + "\n"; + } + } + } + + clearTimeout(timeout); + const exitCode = await proc.exited; + this.runningProcess = null; + + // Read all stderr + let stderr = ""; + const stderrReader = proc.stderr.getReader(); + while (true) { + const { done: stderrDone, value: stderrValue } = await stderrReader.read(); + if (stderrDone) break; + stderr += decoder.decode(stderrValue, { stream: true }); + } + + if (exitCode !== 0) { + return { + success: false, + output: output || stderr, + toolCalls, + durationMs: Date.now() - startTime, + error: `Pi exited with code ${exitCode}: ${stderr}`, + }; + } + + return { + success: true, + output: output.substring(0, 10000), // Truncate output + toolCalls, + durationMs: Date.now() - startTime, + }; + } catch (error: any) { + this.runningProcess = null; + return { + success: false, + output: "", + toolCalls, + durationMs: Date.now() - startTime, + error: error.message, + }; + } + } + + // Execute with the full agent-team (dispatch model) + async executeWithAgentTeam(task: string, context?: string): Promise { + const startTime = Date.now(); + let output = ""; + let toolCalls = 0; + + const fullPrompt = context ? `${context}\n\nTask: ${task}` : task; + + const args = [ + this.config.piBin, + "--mode", "json", + "-p", + "-e", `${this.config.extensionsPath}/agent-team.ts`, + "-e", `${this.config.extensionsPath}/theme-cycler.ts`, + "--model", this.config.piModel, + "--thinking", "off", + fullPrompt, + ]; + + logger.info("pi-executor", `Executing with agent-team: ${task.substring(0, 100)}...`); + + try { + const proc = spawn({ + cmd: args, + cwd: this.config.targetProjectPath, + stdout: "pipe", + stderr: "pipe", + env: { + ...process.env, + ANTHROPIC_API_KEY: this.config.anthropicApiKey, + }, + }); + + this.runningProcess = proc; + + const timeout = setTimeout(() => { + proc.kill(); + }, PI_TIMEOUT_MS); + + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() || ""; + + for (const line of lines) { + if (!line.trim()) continue; + try { + const event: PiEvent = JSON.parse(line); + if (event.type === "message_update" || event.type === "message_end") { + output += event.content || ""; + } + if (event.type === "tool_execution_start") { + toolCalls++; + } + } catch { + output += line + "\n"; + } + } + } + + clearTimeout(timeout); + const exitCode = await proc.exited; + this.runningProcess = null; + + return { + success: exitCode === 0, + output: output.substring(0, 10000), + toolCalls, + durationMs: Date.now() - startTime, + error: exitCode !== 0 ? `Exited with code ${exitCode}` : undefined, + }; + } catch (error: any) { + this.runningProcess = null; + return { + success: false, + output: "", + toolCalls, + durationMs: Date.now() - startTime, + error: error.message, + }; + } + } + + abort() { + if (this.runningProcess) { + this.runningProcess.kill(); + this.runningProcess = null; + } + } + + private buildSystemPrompt(): string { + return `You are Pi Worker — an autonomous DevOps and development agent running inside cr-server-new. +You work on the CharityRight ecosystem projects. You have full bash access. + +RULES: +- Always read relevant files before making changes +- Make small, focused changes +- Test your changes when possible (run linters, type checks) +- Never modify .env files or credentials +- Never run destructive commands (rm -rf, DROP DATABASE, etc) +- Commit your changes with clear commit messages +- If a task is unclear, document what you understood and what you did + +AVAILABLE PROJECTS: +- The current working directory contains the project code +- Use git to track all changes +- Write clear, maintainable code`; + } +} diff --git a/pi-worker/src/pi/types.ts b/pi-worker/src/pi/types.ts new file mode 100644 index 0000000..73a60ea --- /dev/null +++ b/pi-worker/src/pi/types.ts @@ -0,0 +1,32 @@ +export interface PiEvent { + type: string; + [key: string]: any; +} + +export interface PiMessageUpdate extends PiEvent { + type: "message_update"; + content: string; +} + +export interface PiToolCall extends PiEvent { + type: "tool_execution_start"; + tool_name: string; + tool_input: any; +} + +export interface PiAgentEnd extends PiEvent { + type: "agent_end"; +} + +export interface PiMessageEnd extends PiEvent { + type: "message_end"; + content: string; +} + +export interface PiExecutionResult { + success: boolean; + output: string; + toolCalls: number; + durationMs: number; + error?: string; +} diff --git a/pi-worker/src/scheduler/improvement-loop.ts b/pi-worker/src/scheduler/improvement-loop.ts new file mode 100644 index 0000000..5d9b320 --- /dev/null +++ b/pi-worker/src/scheduler/improvement-loop.ts @@ -0,0 +1,165 @@ +import { PiExecutor } from "../pi/executor.js"; +import { logger } from "../logger.js"; +import { FileLock } from "./lock.js"; +import type { Config } from "../config.js"; + +// Categories of autonomous improvements +const IMPROVEMENT_PROMPTS = [ + { + name: "code-quality", + prompt: `Analyze the project codebase for code quality issues. Look for: +- Unused imports or variables +- Inconsistent naming conventions +- Missing error handling +- Functions that are too long or complex +- Duplicated code that could be refactored +Fix the top 2-3 most impactful issues you find. Make small, focused changes.`, + }, + { + name: "documentation", + prompt: `Review the project documentation: +- Check if README.md is up to date +- Look for functions/modules missing JSDoc comments +- Check if API endpoints are documented +- Look for outdated or incorrect docs +Fix the most impactful documentation gap you find.`, + }, + { + name: "security", + prompt: `Perform a security review of the codebase: +- Check for hardcoded secrets or credentials +- Look for SQL injection vulnerabilities +- Check for missing input validation +- Review authentication/authorization logic +- Check for insecure dependencies +Report your findings and fix any non-breaking security issues.`, + }, + { + name: "type-safety", + prompt: `Review TypeScript type safety: +- Look for 'any' types that should be properly typed +- Check for missing null checks +- Look for type assertions that could be avoided +- Check for missing return type annotations +Fix the top 2-3 type safety issues.`, + }, + { + name: "error-handling", + prompt: `Review error handling across the project: +- Look for unhandled promise rejections +- Check for missing try/catch blocks +- Look for swallowed errors (empty catch blocks) +- Check if errors are properly logged +- Look for missing error boundaries in React components +Fix the most critical error handling gaps.`, + }, + { + name: "performance", + prompt: `Look for performance issues: +- N+1 query patterns in database calls +- Missing database indexes (check Prisma schema) +- Unnecessary re-renders in React components +- Large bundle imports that could be lazy-loaded +- Missing caching opportunities +Fix 1-2 performance issues if found.`, + }, + { + name: "testing", + prompt: `Review the test coverage: +- Identify critical untested code paths +- Check for test files with failing or skipped tests +- Look for integration test gaps +Write 1-2 new tests for the most critical untested functionality.`, + }, + { + name: "dependency-health", + prompt: `Check project dependencies: +- Look for outdated packages with known vulnerabilities +- Check for unused dependencies +- Look for duplicate dependencies +- Check if lockfile is in sync +Report findings and fix any safe-to-fix issues.`, + }, +]; + +export class ImprovementLoop { + private pi: PiExecutor; + private config: Config; + private lock: FileLock; + private running = false; + private timer: ReturnType | null = null; + private initialTimeout: ReturnType | null = null; + private currentIndex = 0; + + constructor(config: Config) { + this.config = config; + this.pi = new PiExecutor(config); + this.lock = new FileLock("improvement-loop"); + } + + async start() { + this.running = true; + logger.info("improvement-loop", "Starting autonomous improvement loop", { + interval: `${this.config.improvementIntervalSec}s`, + categories: IMPROVEMENT_PROMPTS.length, + }); + + // Wait a bit before first improvement (let task-loop take priority) + this.initialTimeout = setTimeout(() => { + if (this.running) this.runImprovement(); + }, 30_000); + + this.timer = setInterval( + () => this.runImprovement(), + this.config.improvementIntervalSec * 1000 + ); + } + + stop() { + this.running = false; + if (this.initialTimeout) { + clearTimeout(this.initialTimeout); + this.initialTimeout = null; + } + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + this.pi.abort(); + this.lock.release(); + } + + private async runImprovement() { + if (!this.running) return; + if (!this.lock.acquire()) { + logger.debug("improvement-loop", "Skipping — another improvement in progress"); + return; + } + + const improvement = IMPROVEMENT_PROMPTS[this.currentIndex % IMPROVEMENT_PROMPTS.length]; + this.currentIndex++; + + try { + logger.info("improvement-loop", `Running improvement: ${improvement.name}`); + + const result = await this.pi.execute(improvement.prompt); + + if (result.success) { + logger.info("improvement-loop", `Improvement completed: ${improvement.name}`, { + toolCalls: result.toolCalls, + durationMs: result.durationMs, + }); + } else { + logger.warn("improvement-loop", `Improvement failed: ${improvement.name}`, { + error: result.error, + }); + } + } catch (error: any) { + logger.error("improvement-loop", `Improvement error: ${improvement.name}`, { + error: error.message, + }); + } finally { + this.lock.release(); + } + } +} diff --git a/pi-worker/src/scheduler/lock.ts b/pi-worker/src/scheduler/lock.ts new file mode 100644 index 0000000..5f8ab49 --- /dev/null +++ b/pi-worker/src/scheduler/lock.ts @@ -0,0 +1,47 @@ +import { writeFileSync, unlinkSync, readFileSync, existsSync } from "fs"; +import { join } from "path"; + +export class FileLock { + private lockPath: string; + + constructor(name: string, dir: string = "/tmp") { + this.lockPath = join(dir, `pi-worker-${name}.lock`); + } + + acquire(): boolean { + // Check for stale lock first + if (existsSync(this.lockPath)) { + try { + const content = readFileSync(this.lockPath, "utf-8"); + const lockTime = parseInt(content); + if (Date.now() - lockTime > 15 * 60 * 1000) { + // Stale lock, remove it + try { unlinkSync(this.lockPath); } catch {} + } else { + return false; + } + } catch { + // Corrupted lock file, remove it + try { unlinkSync(this.lockPath); } catch {} + } + } + + // Atomic create + try { + writeFileSync(this.lockPath, Date.now().toString(), { flag: 'wx' }); + return true; + } catch { + return false; // Another process got it first + } + } + + release() { + try { + unlinkSync(this.lockPath); + } catch {} + } + + isLocked(): boolean { + return existsSync(this.lockPath); + } +} diff --git a/pi-worker/src/scheduler/task-loop.ts b/pi-worker/src/scheduler/task-loop.ts new file mode 100644 index 0000000..f749642 --- /dev/null +++ b/pi-worker/src/scheduler/task-loop.ts @@ -0,0 +1,186 @@ +import { AsanaClient } from "../asana/client.js"; +import { PiExecutor } from "../pi/executor.js"; +import { logger } from "../logger.js"; +import { FileLock } from "./lock.js"; +import type { Config } from "../config.js"; +import type { AsanaTask } from "../asana/types.js"; + +// Section names in Asana board +const SECTIONS = { + TODO: "To Do", + IN_PROGRESS: "In Progress", + DONE: "Done", + PI_WORKER: "Pi Worker", // Optional dedicated section +}; + +export class TaskLoop { + private asana: AsanaClient; + private pi: PiExecutor; + private config: Config; + private lock: FileLock; + private running = false; + private timer: ReturnType | null = null; + + constructor(config: Config) { + this.config = config; + this.asana = new AsanaClient(config.asanaAccessToken); + this.pi = new PiExecutor(config); + this.lock = new FileLock("task-loop"); + } + + async start() { + this.running = true; + logger.info("task-loop", "Starting Asana task polling loop", { + interval: `${this.config.pollIntervalSec}s`, + project: this.config.asanaProjectGid, + }); + + // Initial discovery if no project GID configured + if (!this.config.asanaProjectGid) { + await this.discoverProject(); + } + + // Run immediately, then on interval + await this.poll(); + this.timer = setInterval(() => this.poll(), this.config.pollIntervalSec * 1000); + } + + stop() { + this.running = false; + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + this.pi.abort(); + this.lock.release(); + } + + private async discoverProject() { + try { + logger.info("task-loop", "No project GID configured, discovering..."); + const workspaces = await this.asana.getWorkspaces(); + + if (workspaces.length === 0) { + logger.error("task-loop", "No Asana workspaces found"); + return; + } + + logger.info("task-loop", `Found ${workspaces.length} workspace(s)`, workspaces); + + for (const ws of workspaces) { + const projects = await this.asana.getProjects(ws.gid); + logger.info("task-loop", `Workspace "${ws.name}" has ${projects.length} project(s)`, projects); + + if (projects.length > 0 && !this.config.asanaProjectGid) { + // Use first project as default + this.config.asanaProjectGid = projects[0].gid; + logger.info("task-loop", `Auto-selected project: ${projects[0].name} (${projects[0].gid})`); + } + } + } catch (error: any) { + logger.error("task-loop", "Failed to discover Asana project", { error: error.message }); + } + } + + private async poll() { + if (!this.running || !this.config.asanaProjectGid) return; + if (!this.lock.acquire()) { + logger.debug("task-loop", "Skipping poll — another execution in progress"); + return; + } + + try { + logger.info("task-loop", "Polling for Asana tasks..."); + + // Try dedicated "Pi Worker" section first, then "To Do" + let tasks = await this.asana.getIncompleteTasks(this.config.asanaProjectGid, SECTIONS.PI_WORKER); + + if (tasks.length === 0) { + tasks = await this.asana.getIncompleteTasks(this.config.asanaProjectGid, SECTIONS.TODO); + } + + if (tasks.length === 0) { + logger.info("task-loop", "No pending tasks found"); + this.lock.release(); + return; + } + + // Pick the first (highest priority) task + const task = tasks[0]; + logger.info("task-loop", `Picked task: ${task.name}`, { gid: task.gid }); + + await this.executeTask(task); + } catch (error: any) { + logger.error("task-loop", "Poll failed", { error: error.message }); + } finally { + this.lock.release(); + } + } + + private async executeTask(task: AsanaTask) { + try { + // Move to "In Progress" + const inProgressSection = await this.asana.findSection(this.config.asanaProjectGid, SECTIONS.IN_PROGRESS); + if (inProgressSection) { + await this.asana.moveTaskToSection(task.gid, inProgressSection.gid); + } + + // Add starting comment + await this.asana.addComment(task.gid, `🤖 Pi Worker picked up this task.\nStarting execution at ${new Date().toISOString()}`); + + // Build context from task + const context = this.buildTaskContext(task); + + // Execute with Pi agent + const result = await this.pi.execute(task.name + (task.notes ? `\n\nDetails:\n${task.notes}` : ""), context); + + // Add completion comment + const statusEmoji = result.success ? "✅" : "❌"; + const comment = [ + `${statusEmoji} Pi Worker ${result.success ? "completed" : "failed"} this task.`, + `Duration: ${(result.durationMs / 1000).toFixed(1)}s`, + `Tool calls: ${result.toolCalls}`, + result.error ? `Error: ${result.error}` : "", + "", + "Output (truncated):", + "```", + result.output.substring(0, 3000), + "```", + ].filter(Boolean).join("\n"); + + await this.asana.addComment(task.gid, comment); + + if (result.success) { + // Move to "Done" and mark complete + const doneSection = await this.asana.findSection(this.config.asanaProjectGid, SECTIONS.DONE); + if (doneSection) { + await this.asana.moveTaskToSection(task.gid, doneSection.gid); + } + await this.asana.updateTask(task.gid, { completed: true }); + logger.info("task-loop", `Task completed: ${task.name}`); + } else { + // Move back to To Do on failure + const todoSection = await this.asana.findSection(this.config.asanaProjectGid, SECTIONS.TODO); + if (todoSection) { + await this.asana.moveTaskToSection(task.gid, todoSection.gid); + } + logger.error("task-loop", `Task failed: ${task.name}`, { error: result.error }); + } + } catch (error: any) { + logger.error("task-loop", `Task execution error: ${task.name}`, { error: error.message }); + try { + await this.asana.addComment(task.gid, `❌ Pi Worker crashed: ${error.message}`); + } catch {} + } + } + + private buildTaskContext(task: AsanaTask): string { + const parts = [ + `Asana Task: ${task.name}`, + task.notes ? `Description: ${task.notes}` : "", + task.due_on ? `Due: ${task.due_on}` : "", + task.tags.length > 0 ? `Tags: ${task.tags.map((t) => t.name).join(", ")}` : "", + ]; + return parts.filter(Boolean).join("\n"); + } +} diff --git a/pi-worker/tsconfig.json b/pi-worker/tsconfig.json new file mode 100644 index 0000000..9143401 --- /dev/null +++ b/pi-worker/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "target": "ESNext", + "module": "ESNext", + "moduleResolution": "bundler", + "types": ["bun-types"], + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "outDir": "dist", + "rootDir": "src", + "resolveJsonModule": true + }, + "include": ["src/**/*"] +}