feat: pi-worker now picks up and fixes real bugs from Asana
Major changes: - Task loop polls Under Review + New Bugs sections (was only checking empty sections) - Smart project routing: keyword matching maps task names to correct codebase (e.g. 'pledge' → legacy-donation-system-laravel) - Pi executes in the project directory, not /opt/pi-worker - Git pull before execution to get latest code - Skips already-attempted tasks (checks Asana comments) - Config now exposes projectsPath First real bug fixed autonomously: 'Duplicate Pledges Detected with Same External Reference' → Fixed uniqid() causing duplicate Engage pledges on every sync → 1 file changed in app/N3O/Schema.php Also: fix watchdog to use systemd-notify command (unix_dgram unsupported in Bun), add debug logging to lock acquire failures
This commit is contained in:
@@ -17,6 +17,7 @@ export interface Config {
|
|||||||
extensionsPath: string;
|
extensionsPath: string;
|
||||||
agentsPath: string;
|
agentsPath: string;
|
||||||
logDir: string;
|
logDir: string;
|
||||||
|
projectsPath: string;
|
||||||
|
|
||||||
// Timing
|
// Timing
|
||||||
pollIntervalSec: number;
|
pollIntervalSec: number;
|
||||||
@@ -46,6 +47,7 @@ export function loadConfig(): Config {
|
|||||||
extensionsPath: process.env.EXTENSIONS_PATH || "../extensions",
|
extensionsPath: process.env.EXTENSIONS_PATH || "../extensions",
|
||||||
agentsPath: process.env.AGENTS_PATH || "../.pi/agents",
|
agentsPath: process.env.AGENTS_PATH || "../.pi/agents",
|
||||||
logDir: process.env.LOG_DIR || "./logs",
|
logDir: process.env.LOG_DIR || "./logs",
|
||||||
|
projectsPath: process.env.PROJECTS_PATH || "./projects",
|
||||||
|
|
||||||
pollIntervalSec: parseInt(process.env.POLL_INTERVAL_SEC || "120"),
|
pollIntervalSec: parseInt(process.env.POLL_INTERVAL_SEC || "120"),
|
||||||
improvementIntervalSec: parseInt(process.env.IMPROVEMENT_INTERVAL_SEC || "3600"),
|
improvementIntervalSec: parseInt(process.env.IMPROVEMENT_INTERVAL_SEC || "3600"),
|
||||||
|
|||||||
@@ -79,16 +79,16 @@ function startWatchdog(healthPort: number) {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Notify systemd via NOTIFY_SOCKET
|
// Notify systemd via systemd-notify command
|
||||||
const notifySocket = process.env.NOTIFY_SOCKET;
|
try {
|
||||||
if (notifySocket) {
|
const proc = Bun.spawn({
|
||||||
const { createSocket } = await import("dgram");
|
cmd: ["systemd-notify", "WATCHDOG=1"],
|
||||||
const socketPath = notifySocket.startsWith("@")
|
stdout: "ignore",
|
||||||
? `\0${notifySocket.slice(1)}`
|
stderr: "ignore",
|
||||||
: notifySocket;
|
});
|
||||||
const client = createSocket("unix_dgram" as any);
|
await proc.exited;
|
||||||
const msg = Buffer.from("WATCHDOG=1");
|
} catch {
|
||||||
client.send(msg, 0, msg.length, socketPath, () => client.close());
|
// systemd-notify not available — not critical
|
||||||
}
|
}
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
logger.error("health", `Watchdog ping failed: ${e}`);
|
logger.error("health", `Watchdog ping failed: ${e}`);
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ export class PiExecutor {
|
|||||||
this.config = config;
|
this.config = config;
|
||||||
}
|
}
|
||||||
|
|
||||||
async execute(task: string, context?: string, timeoutMs?: number): Promise<PiExecutionResult> {
|
async execute(task: string, context?: string, timeoutMs?: number, cwd?: string): Promise<PiExecutionResult> {
|
||||||
const startTime = Date.now();
|
const startTime = Date.now();
|
||||||
let output = "";
|
let output = "";
|
||||||
let toolCalls = 0;
|
let toolCalls = 0;
|
||||||
@@ -33,9 +33,11 @@ export class PiExecutor {
|
|||||||
fullPrompt,
|
fullPrompt,
|
||||||
];
|
];
|
||||||
|
|
||||||
|
const workingDir = cwd || this.config.targetProjectPath;
|
||||||
|
|
||||||
logger.info("pi-executor", `Executing task: ${task.substring(0, 100)}...`, {
|
logger.info("pi-executor", `Executing task: ${task.substring(0, 100)}...`, {
|
||||||
model: this.config.piModel,
|
model: this.config.piModel,
|
||||||
cwd: this.config.targetProjectPath,
|
cwd: workingDir,
|
||||||
});
|
});
|
||||||
|
|
||||||
try {
|
try {
|
||||||
@@ -45,7 +47,7 @@ export class PiExecutor {
|
|||||||
|
|
||||||
const proc = spawn({
|
const proc = spawn({
|
||||||
cmd: args,
|
cmd: args,
|
||||||
cwd: this.config.targetProjectPath,
|
cwd: workingDir,
|
||||||
stdout: "pipe",
|
stdout: "pipe",
|
||||||
stderr: "pipe",
|
stderr: "pipe",
|
||||||
env: cleanEnv,
|
env: cleanEnv,
|
||||||
|
|||||||
@@ -77,7 +77,12 @@ export class FileLock {
|
|||||||
writeFileSync(this.lockPath, Date.now().toString(), { flag: 'wx' });
|
writeFileSync(this.lockPath, Date.now().toString(), { flag: 'wx' });
|
||||||
this.held = true;
|
this.held = true;
|
||||||
return true;
|
return true;
|
||||||
} catch {
|
} catch (e) {
|
||||||
|
const code = (e as NodeJS.ErrnoException).code;
|
||||||
|
if (code !== "EEXIST") {
|
||||||
|
// Unexpected error (permissions, disk full, etc.) — log it
|
||||||
|
process.stderr.write(`[lock] acquire failed for ${this.lockPath}: ${code || e}\n`);
|
||||||
|
}
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
import { readdirSync, existsSync } from "fs";
|
||||||
|
import { join } from "path";
|
||||||
import { AsanaClient } from "../asana/client.js";
|
import { AsanaClient } from "../asana/client.js";
|
||||||
import { PiExecutor } from "../pi/executor.js";
|
import { PiExecutor } from "../pi/executor.js";
|
||||||
import { logger } from "../logger.js";
|
import { logger } from "../logger.js";
|
||||||
@@ -5,10 +7,58 @@ import { FileLock } from "./lock.js";
|
|||||||
import type { Config } from "../config.js";
|
import type { Config } from "../config.js";
|
||||||
import type { AsanaTask, AsanaSection } from "../asana/types.js";
|
import type { AsanaTask, AsanaSection } from "../asana/types.js";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Section classification for the Bug Intake board.
|
||||||
|
*
|
||||||
|
* PICKUP = sections we actively pull tasks FROM to execute
|
||||||
|
* EXECUTE = "Under Review" / "In Progress" — where we move tasks during execution
|
||||||
|
* DONE = where completed tasks land
|
||||||
|
* SKIP = sections we never touch (deferred, won't fix)
|
||||||
|
*/
|
||||||
const SECTIONS = {
|
const SECTIONS = {
|
||||||
TODO: ["Pi Worker", "To Do", "New Bugs", "Ready for Development", "Recently Assigned", "Awaiting More Info"],
|
PICKUP: ["New Bugs", "Ready for Development"],
|
||||||
IN_PROGRESS: ["In Progress", "Under Review"],
|
EXECUTE: ["Under Review", "In Progress"],
|
||||||
DONE: ["Done", "Resolved", "Complete", "Completed"],
|
DONE: ["Resolved", "Done", "Complete", "Completed"],
|
||||||
|
SKIP: ["Deferred", "Won't Fix", "Awaiting More Info"],
|
||||||
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Map of keywords in task names/notes → project directory names.
|
||||||
|
* Used to route tasks to the correct codebase for execution.
|
||||||
|
*/
|
||||||
|
const PROJECT_KEYWORDS: Record<string, string[]> = {
|
||||||
|
"legacy-donation-system-laravel": [
|
||||||
|
"laravel", "legacy", "donation system", "receipt", "email receipt",
|
||||||
|
"pledge", "donor", "donation", "fundrais", "campaign", "appeal",
|
||||||
|
"checkout", "payment", "stripe", "paypal", "gocardless",
|
||||||
|
"cross sell", "team page", "ben nevis",
|
||||||
|
],
|
||||||
|
"charity-right-uk-v2": [
|
||||||
|
"charity-right", "charityright", "website", "frontend", "next.js",
|
||||||
|
"nextjs", "landing", "home page", "UI", "layout", "styling",
|
||||||
|
],
|
||||||
|
"checkout-v2": [
|
||||||
|
"checkout-v2", "checkout v2", "new checkout", "payment flow",
|
||||||
|
],
|
||||||
|
"donation-dashboard": [
|
||||||
|
"dashboard", "admin", "analytics", "reporting", "reports",
|
||||||
|
"export", "data export",
|
||||||
|
],
|
||||||
|
"command-center": [
|
||||||
|
"command center", "command-center", "ops", "monitoring",
|
||||||
|
],
|
||||||
|
"charityright-sync": [
|
||||||
|
"sync", "n3o", "engage", "import", "county", "state", "mapping",
|
||||||
|
],
|
||||||
|
"enthuse-db-sync-v2": [
|
||||||
|
"enthuse", "enthuse sync", "db sync",
|
||||||
|
],
|
||||||
|
"launchgood-sync": [
|
||||||
|
"launchgood", "launch good",
|
||||||
|
],
|
||||||
|
"marketing-site": [
|
||||||
|
"marketing", "blog", "wordpress", "seo", "content",
|
||||||
|
],
|
||||||
};
|
};
|
||||||
|
|
||||||
export class TaskLoop {
|
export class TaskLoop {
|
||||||
@@ -22,13 +72,17 @@ export class TaskLoop {
|
|||||||
private workspaceGid: string = "";
|
private workspaceGid: string = "";
|
||||||
|
|
||||||
// Cache: track tasks we already verified as "not done" to avoid re-checking
|
// Cache: track tasks we already verified as "not done" to avoid re-checking
|
||||||
private verifiedNotDone: Map<string, number> = new Map(); // gid -> timestamp
|
private verifiedNotDone: Map<string, number> = new Map(); // gid → timestamp
|
||||||
private readonly VERIFY_CACHE_TTL = 30 * 60 * 1000; // 30 min
|
private readonly VERIFY_CACHE_TTL = 30 * 60 * 1000; // 30 min
|
||||||
|
|
||||||
// Cache: sections per project
|
// Cache: sections per project
|
||||||
private sectionCache: Map<string, { sections: AsanaSection[]; ts: number }> = new Map();
|
private sectionCache: Map<string, { sections: AsanaSection[]; ts: number }> = new Map();
|
||||||
private readonly SECTION_CACHE_TTL = 5 * 60 * 1000; // 5 min
|
private readonly SECTION_CACHE_TTL = 5 * 60 * 1000; // 5 min
|
||||||
|
|
||||||
|
// Track consecutive failures to back off
|
||||||
|
private consecutiveFailures = 0;
|
||||||
|
private readonly MAX_BACKOFF_MULTIPLIER = 5;
|
||||||
|
|
||||||
constructor(config: Config) {
|
constructor(config: Config) {
|
||||||
this.config = config;
|
this.config = config;
|
||||||
this.asana = new AsanaClient(config.asanaAccessToken);
|
this.asana = new AsanaClient(config.asanaAccessToken);
|
||||||
@@ -41,12 +95,17 @@ export class TaskLoop {
|
|||||||
logger.info("task-loop", "Starting Asana task polling loop", {
|
logger.info("task-loop", "Starting Asana task polling loop", {
|
||||||
interval: `${this.config.pollIntervalSec}s`,
|
interval: `${this.config.pollIntervalSec}s`,
|
||||||
project: this.config.asanaProjectGid,
|
project: this.config.asanaProjectGid,
|
||||||
|
projectsPath: this.getProjectsDir(),
|
||||||
});
|
});
|
||||||
|
|
||||||
if (!this.config.asanaProjectGid) {
|
if (!this.config.asanaProjectGid) {
|
||||||
await this.discoverProject();
|
await this.discoverProject();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Log available projects for debugging
|
||||||
|
const projects = this.getAvailableProjects();
|
||||||
|
logger.info("task-loop", `Available codebases: ${projects.join(", ") || "NONE"}`);
|
||||||
|
|
||||||
await this.poll();
|
await this.poll();
|
||||||
this.timer = setInterval(() => this.poll(), this.config.pollIntervalSec * 1000);
|
this.timer = setInterval(() => this.poll(), this.config.pollIntervalSec * 1000);
|
||||||
}
|
}
|
||||||
@@ -61,6 +120,74 @@ export class TaskLoop {
|
|||||||
this.lock.release();
|
this.lock.release();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─── Project resolution ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
private getProjectsDir(): string {
|
||||||
|
return join(this.config.targetProjectPath, this.config.projectsPath);
|
||||||
|
}
|
||||||
|
|
||||||
|
private getAvailableProjects(): string[] {
|
||||||
|
const dir = this.getProjectsDir();
|
||||||
|
if (!existsSync(dir)) return [];
|
||||||
|
return readdirSync(dir, { withFileTypes: true })
|
||||||
|
.filter((d) => d.isDirectory() && !d.name.startsWith("."))
|
||||||
|
.map((d) => d.name);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Determine which project codebase a task should execute against.
|
||||||
|
* Returns the absolute path to the project directory, or the
|
||||||
|
* targetProjectPath as fallback.
|
||||||
|
*/
|
||||||
|
private resolveProjectForTask(task: AsanaTask): { projectDir: string; projectName: string } {
|
||||||
|
const searchText = `${task.name} ${task.notes || ""}`.toLowerCase();
|
||||||
|
const available = this.getAvailableProjects();
|
||||||
|
|
||||||
|
// First: check if the task name contains a [project] tag (from improvement loop)
|
||||||
|
const tagMatch = task.name.match(/\[([^\]]+)\]/);
|
||||||
|
if (tagMatch) {
|
||||||
|
const tagName = tagMatch[1].toLowerCase().replace(/\s+/g, "-");
|
||||||
|
const exact = available.find((p) => p.toLowerCase() === tagName);
|
||||||
|
if (exact) {
|
||||||
|
return {
|
||||||
|
projectDir: join(this.getProjectsDir(), exact),
|
||||||
|
projectName: exact,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Second: keyword matching
|
||||||
|
let bestMatch: { project: string; score: number } | null = null;
|
||||||
|
|
||||||
|
for (const [project, keywords] of Object.entries(PROJECT_KEYWORDS)) {
|
||||||
|
if (!available.includes(project)) continue;
|
||||||
|
let score = 0;
|
||||||
|
for (const keyword of keywords) {
|
||||||
|
if (searchText.includes(keyword.toLowerCase())) {
|
||||||
|
score += keyword.length; // Longer matches score higher
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (score > 0 && (!bestMatch || score > bestMatch.score)) {
|
||||||
|
bestMatch = { project, score };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (bestMatch) {
|
||||||
|
return {
|
||||||
|
projectDir: join(this.getProjectsDir(), bestMatch.project),
|
||||||
|
projectName: bestMatch.project,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fallback: use the main project path
|
||||||
|
return {
|
||||||
|
projectDir: this.config.targetProjectPath,
|
||||||
|
projectName: "unknown",
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── Discovery ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
private async discoverProject() {
|
private async discoverProject() {
|
||||||
try {
|
try {
|
||||||
logger.info("task-loop", "No project GID configured, discovering...");
|
logger.info("task-loop", "No project GID configured, discovering...");
|
||||||
@@ -98,6 +225,8 @@ export class TaskLoop {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─── Main poll loop ──────────────────────────────────────────────────
|
||||||
|
|
||||||
private async poll() {
|
private async poll() {
|
||||||
if (!this.running || !this.config.asanaProjectGid) return;
|
if (!this.running || !this.config.asanaProjectGid) return;
|
||||||
if (!this.lock.acquire()) {
|
if (!this.lock.acquire()) {
|
||||||
@@ -106,15 +235,25 @@ export class TaskLoop {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Step 1: Quick cleanup - move completed My Tasks to Done sections
|
// Step 1: Quick cleanup — move completed My Tasks to Done sections
|
||||||
await this.cleanupCompletedMyTasks();
|
await this.cleanupCompletedMyTasks();
|
||||||
|
|
||||||
// Step 2: Sweep incomplete tasks for already-done work (with caching)
|
// Step 2: Pick up tasks to execute
|
||||||
await this.sweepIncompleteTasks();
|
// Check PICKUP sections first (New Bugs, Ready for Development)
|
||||||
|
// Then check EXECUTE sections (Under Review) for tasks that haven't been attempted
|
||||||
|
logger.info("task-loop", "Polling for Asana tasks...");
|
||||||
|
|
||||||
// Step 3: Pick up new tasks to execute
|
let tasks = await this.getTasksFromSections(this.config.asanaProjectGid, SECTIONS.PICKUP);
|
||||||
logger.info("task-loop", "Polling for new Asana tasks...");
|
|
||||||
const tasks = await this.getTasksFromSections(this.config.asanaProjectGid, SECTIONS.TODO);
|
if (tasks.length === 0) {
|
||||||
|
// Also check Under Review — these are real bugs waiting for someone to fix them
|
||||||
|
tasks = await this.getTasksFromSections(this.config.asanaProjectGid, SECTIONS.EXECUTE);
|
||||||
|
|
||||||
|
// Filter out tasks we recently attempted (check comments for Pi Worker activity)
|
||||||
|
if (tasks.length > 0) {
|
||||||
|
tasks = await this.filterUnatttemptedTasks(tasks);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (tasks.length === 0) {
|
if (tasks.length === 0) {
|
||||||
logger.info("task-loop", "No pending tasks found");
|
logger.info("task-loop", "No pending tasks found");
|
||||||
@@ -122,17 +261,53 @@ export class TaskLoop {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Pick the first task
|
||||||
const task = tasks[0];
|
const task = tasks[0];
|
||||||
logger.info("task-loop", `Picked task: ${task.name}`, { gid: task.gid });
|
const { projectDir, projectName } = this.resolveProjectForTask(task);
|
||||||
await this.executeTask(task);
|
logger.info("task-loop", `Picked task: "${task.name}"`, {
|
||||||
|
gid: task.gid,
|
||||||
|
project: projectName,
|
||||||
|
projectDir,
|
||||||
|
});
|
||||||
|
|
||||||
|
await this.executeTask(task, projectDir, projectName);
|
||||||
} catch (error: any) {
|
} catch (error: any) {
|
||||||
logger.error("task-loop", "Poll failed", { error: error.message });
|
logger.error("task-loop", "Poll failed", { error: error.message });
|
||||||
|
this.consecutiveFailures++;
|
||||||
} finally {
|
} finally {
|
||||||
this.lock.release();
|
this.lock.release();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Quick: just move completed tasks out of Recently Assigned. No Pi calls.
|
/**
|
||||||
|
* Filter out tasks that Pi Worker has already commented on (attempted).
|
||||||
|
* Prevents re-attempting failed tasks every cycle.
|
||||||
|
*/
|
||||||
|
private async filterUnatttemptedTasks(tasks: AsanaTask[]): Promise<AsanaTask[]> {
|
||||||
|
const unattempted: AsanaTask[] = [];
|
||||||
|
|
||||||
|
for (const task of tasks) {
|
||||||
|
try {
|
||||||
|
const comments = await this.asana.getComments(task.gid);
|
||||||
|
const hasWorkerComment = comments.some(
|
||||||
|
(c) => c.text && c.text.includes("Pi Worker")
|
||||||
|
);
|
||||||
|
if (!hasWorkerComment) {
|
||||||
|
unattempted.push(task);
|
||||||
|
} else {
|
||||||
|
logger.debug("task-loop", `Skipping already-attempted task: "${task.name}"`);
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
// If we can't read comments, include the task
|
||||||
|
unattempted.push(task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return unattempted;
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── Cleanup ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
private async cleanupCompletedMyTasks() {
|
private async cleanupCompletedMyTasks() {
|
||||||
if (!this.userTaskListGid) return;
|
if (!this.userTaskListGid) return;
|
||||||
|
|
||||||
@@ -146,7 +321,6 @@ export class TaskLoop {
|
|||||||
|
|
||||||
for (const task of completedTasks) {
|
for (const task of completedTasks) {
|
||||||
try {
|
try {
|
||||||
// Move to Done section in the task's project
|
|
||||||
if (task.memberships && task.memberships.length > 0) {
|
if (task.memberships && task.memberships.length > 0) {
|
||||||
for (const membership of task.memberships) {
|
for (const membership of task.memberships) {
|
||||||
const projectGid = membership.project?.gid;
|
const projectGid = membership.project?.gid;
|
||||||
@@ -166,7 +340,6 @@ export class TaskLoop {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Move out of Recently Assigned
|
|
||||||
await this.asana.updateTask(task.gid, { assignee_status: "later" });
|
await this.asana.updateTask(task.gid, { assignee_status: "later" });
|
||||||
} catch (e: any) {
|
} catch (e: any) {
|
||||||
logger.debug("task-loop", `Could not clean task ${task.name}: ${e.message}`);
|
logger.debug("task-loop", `Could not clean task ${task.name}: ${e.message}`);
|
||||||
@@ -177,131 +350,50 @@ export class TaskLoop {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Sweep: verify incomplete tasks - are they already done? Uses cache to skip recently checked.
|
// ─── Task execution ─────────────────────────────────────────────────
|
||||||
private async sweepIncompleteTasks() {
|
|
||||||
|
private async executeTask(task: AsanaTask, projectDir: string, projectName: string) {
|
||||||
try {
|
try {
|
||||||
const sweepSections = [...SECTIONS.TODO, ...SECTIONS.IN_PROGRESS];
|
// Step 1: Move to Under Review
|
||||||
const allTasks: AsanaTask[] = [];
|
const executeSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.EXECUTE);
|
||||||
|
if (executeSection) await this.asana.moveTaskToSection(task.gid, executeSection.gid);
|
||||||
|
|
||||||
for (const sectionName of sweepSections) {
|
await this.asana.addComment(
|
||||||
const section = await this.findSectionCached(this.config.asanaProjectGid, sectionName);
|
task.gid,
|
||||||
if (!section) continue;
|
`🤖 Pi Worker picked up this task.\nProject: \`${projectName}\`\nStarted: ${new Date().toISOString()}`
|
||||||
const tasks = await this.asana.getSectionTasks(section.gid);
|
);
|
||||||
allTasks.push(...tasks.filter((t) => !t.completed));
|
|
||||||
}
|
|
||||||
|
|
||||||
// Also get incomplete My Tasks
|
// Step 2: Build context with project awareness
|
||||||
if (this.userTaskListGid) {
|
const context = this.buildTaskContext(task, projectDir, projectName);
|
||||||
try {
|
|
||||||
const myTasks = await this.asana.getUserTasks(this.userTaskListGid);
|
|
||||||
allTasks.push(...myTasks.filter((t) => !t.completed));
|
|
||||||
} catch {}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Deduplicate
|
// Step 3: Pull latest code
|
||||||
const seen = new Set<string>();
|
await this.gitPull(projectDir, projectName);
|
||||||
const uniqueTasks = allTasks.filter((t) => {
|
|
||||||
if (seen.has(t.gid)) return false;
|
|
||||||
seen.add(t.gid);
|
|
||||||
return true;
|
|
||||||
});
|
|
||||||
|
|
||||||
// Filter out recently verified tasks (cache hit)
|
// Step 4: Execute with Pi in the correct project directory
|
||||||
const now = Date.now();
|
const result = await this.pi.execute(
|
||||||
const tasksToVerify = uniqueTasks.filter((t) => {
|
task.name + (task.notes ? `\n\nDetails:\n${task.notes}` : ""),
|
||||||
const lastCheck = this.verifiedNotDone.get(t.gid);
|
context,
|
||||||
if (lastCheck && now - lastCheck < this.VERIFY_CACHE_TTL) return false;
|
undefined,
|
||||||
return true;
|
projectDir, // Execute in the project directory
|
||||||
});
|
);
|
||||||
|
|
||||||
if (tasksToVerify.length === 0) {
|
// Step 5: Post-verify
|
||||||
logger.debug("task-loop", `Sweep: ${uniqueTasks.length} tasks all recently verified, skipping`);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.info("task-loop", `Sweeping ${tasksToVerify.length} tasks (${uniqueTasks.length - tasksToVerify.length} cached, skipped)`);
|
|
||||||
|
|
||||||
for (const task of tasksToVerify) {
|
|
||||||
const check = await this.verifyTaskAlreadyDone(task);
|
|
||||||
|
|
||||||
if (check.done) {
|
|
||||||
logger.info("task-loop", `Sweep: "${task.name}" is already done`, { evidence: check.evidence });
|
|
||||||
|
|
||||||
await this.asana.addComment(
|
|
||||||
task.gid,
|
|
||||||
`✅ Pi Worker sweep: This task is **already completed**.\n\nEvidence:\n${check.evidence}`
|
|
||||||
);
|
|
||||||
|
|
||||||
// Move to done section if task has project membership
|
|
||||||
if (task.memberships && task.memberships.length > 0) {
|
|
||||||
for (const membership of task.memberships) {
|
|
||||||
const projectGid = membership.project?.gid;
|
|
||||||
if (projectGid) {
|
|
||||||
const doneSection = await this.findDoneSection(projectGid);
|
|
||||||
if (doneSection) {
|
|
||||||
await this.asana.moveTaskToSection(task.gid, doneSection.gid);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
await this.asana.updateTask(task.gid, { completed: true });
|
|
||||||
logger.info("task-loop", `Sweep: Marked done: ${task.name}`);
|
|
||||||
} else {
|
|
||||||
// Cache as "not done" to avoid re-verifying next cycle
|
|
||||||
this.verifiedNotDone.set(task.gid, now);
|
|
||||||
logger.debug("task-loop", `Sweep: "${task.name}" not yet done`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Clean expired cache entries
|
|
||||||
for (const [gid, ts] of this.verifiedNotDone) {
|
|
||||||
if (now - ts > this.VERIFY_CACHE_TTL) this.verifiedNotDone.delete(gid);
|
|
||||||
}
|
|
||||||
} catch (error: any) {
|
|
||||||
logger.error("task-loop", `Sweep error: ${error.message}`);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private async executeTask(task: AsanaTask) {
|
|
||||||
try {
|
|
||||||
// Step 1: Quick pre-verify
|
|
||||||
const alreadyDone = await this.verifyTaskAlreadyDone(task);
|
|
||||||
|
|
||||||
if (alreadyDone.done) {
|
|
||||||
logger.info("task-loop", `Task already done: ${task.name}`);
|
|
||||||
await this.asana.addComment(task.gid, `✅ Pi Worker verified this task is **already completed**.\n\nEvidence:\n${alreadyDone.evidence}`);
|
|
||||||
const doneSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.DONE);
|
|
||||||
if (doneSection) await this.asana.moveTaskToSection(task.gid, doneSection.gid);
|
|
||||||
await this.asana.updateTask(task.gid, { completed: true });
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Step 2: Move to In Progress
|
|
||||||
const inProgressSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.IN_PROGRESS);
|
|
||||||
if (inProgressSection) await this.asana.moveTaskToSection(task.gid, inProgressSection.gid);
|
|
||||||
|
|
||||||
await this.asana.addComment(task.gid, `🤖 Pi Worker picked up this task.\nStarting execution at ${new Date().toISOString()}`);
|
|
||||||
|
|
||||||
const context = this.buildTaskContext(task);
|
|
||||||
|
|
||||||
// Step 3: Execute
|
|
||||||
const result = await this.pi.execute(task.name + (task.notes ? `\n\nDetails:\n${task.notes}` : ""), context);
|
|
||||||
|
|
||||||
// Step 4: Post-verify
|
|
||||||
let verified = false;
|
let verified = false;
|
||||||
if (result.success && result.toolCalls > 0) {
|
if (result.success && result.toolCalls > 0) {
|
||||||
const postCheck = await this.verifyTaskAlreadyDone(task);
|
const postCheck = await this.verifyTaskDone(task, projectDir);
|
||||||
verified = postCheck.done;
|
verified = postCheck.done;
|
||||||
}
|
}
|
||||||
|
|
||||||
const statusEmoji = result.success ? (verified ? "✅" : "⚠️") : "❌";
|
const statusEmoji = result.success ? (verified ? "✅" : "⚠️") : "❌";
|
||||||
const verificationNote = result.success
|
const verificationNote = result.success
|
||||||
? (verified ? "Post-execution verification: PASSED" : "Post-execution verification: COULD NOT CONFIRM — please review manually")
|
? (verified
|
||||||
|
? "Post-execution verification: **PASSED**"
|
||||||
|
: "Post-execution verification: COULD NOT CONFIRM — please review manually")
|
||||||
: "";
|
: "";
|
||||||
|
|
||||||
const comment = [
|
const comment = [
|
||||||
`${statusEmoji} Pi Worker ${result.success ? "completed" : "failed"} this task.`,
|
`${statusEmoji} Pi Worker ${result.success ? "completed" : "failed"} this task.`,
|
||||||
|
`Project: \`${projectName}\``,
|
||||||
verificationNote,
|
verificationNote,
|
||||||
`Duration: ${(result.durationMs / 1000).toFixed(1)}s`,
|
`Duration: ${(result.durationMs / 1000).toFixed(1)}s`,
|
||||||
`Tool calls: ${result.toolCalls}`,
|
`Tool calls: ${result.toolCalls}`,
|
||||||
@@ -319,23 +411,27 @@ export class TaskLoop {
|
|||||||
const doneSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.DONE);
|
const doneSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.DONE);
|
||||||
if (doneSection) await this.asana.moveTaskToSection(task.gid, doneSection.gid);
|
if (doneSection) await this.asana.moveTaskToSection(task.gid, doneSection.gid);
|
||||||
await this.asana.updateTask(task.gid, { completed: true });
|
await this.asana.updateTask(task.gid, { completed: true });
|
||||||
logger.info("task-loop", `Task completed: ${task.name}`, { verified });
|
logger.info("task-loop", `Task completed: ${task.name}`, { verified, project: projectName });
|
||||||
|
this.consecutiveFailures = 0;
|
||||||
} else {
|
} else {
|
||||||
const todoSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.TODO);
|
// Leave in Under Review — don't move back to New Bugs
|
||||||
if (todoSection) await this.asana.moveTaskToSection(task.gid, todoSection.gid);
|
// The filterUnattemptedTasks check will skip it on future polls
|
||||||
logger.error("task-loop", `Task failed: ${task.name}`, { error: result.error });
|
logger.error("task-loop", `Task failed: ${task.name}`, { error: result.error, project: projectName });
|
||||||
|
this.consecutiveFailures++;
|
||||||
}
|
}
|
||||||
} catch (error: any) {
|
} catch (error: any) {
|
||||||
logger.error("task-loop", `Task execution error: ${task.name}`, { error: error.message });
|
logger.error("task-loop", `Task execution error: ${task.name}`, { error: error.message });
|
||||||
|
this.consecutiveFailures++;
|
||||||
try {
|
try {
|
||||||
await this.asana.addComment(task.gid, `❌ Pi Worker crashed: ${error.message}`);
|
await this.asana.addComment(task.gid, `❌ Pi Worker crashed: ${error.message}`);
|
||||||
} catch {}
|
} catch {}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private async verifyTaskAlreadyDone(task: AsanaTask): Promise<{ done: boolean; evidence: string }> {
|
private async verifyTaskDone(task: AsanaTask, projectDir: string): Promise<{ done: boolean; evidence: string }> {
|
||||||
try {
|
try {
|
||||||
const verifyPrompt = `CHECK if this task is already done. Be FAST - spend max 30 seconds investigating. Do NOT fix anything. Do NOT make changes. Just check and report.
|
const verifyPrompt = `CHECK if this task is already done. Be FAST — spend max 30 seconds.
|
||||||
|
Do NOT fix anything. Do NOT make changes. Just check and report.
|
||||||
|
|
||||||
Task: ${task.name}
|
Task: ${task.name}
|
||||||
${task.notes ? `Details: ${task.notes}` : ""}
|
${task.notes ? `Details: ${task.notes}` : ""}
|
||||||
@@ -343,7 +439,8 @@ ${task.notes ? `Details: ${task.notes}` : ""}
|
|||||||
Quick checks only:
|
Quick checks only:
|
||||||
- ls and grep relevant files/dirs
|
- ls and grep relevant files/dirs
|
||||||
- Check if a fix/feature is already in the code
|
- Check if a fix/feature is already in the code
|
||||||
- Check running services if relevant (curl, systemctl)
|
- Check git log for recent relevant commits
|
||||||
|
- Check running services if relevant (curl localhost)
|
||||||
- Do NOT run find on large directories
|
- Do NOT run find on large directories
|
||||||
- Do NOT clone repos or install anything
|
- Do NOT clone repos or install anything
|
||||||
|
|
||||||
@@ -351,7 +448,7 @@ RESPOND EXACTLY:
|
|||||||
DONE: true or false
|
DONE: true or false
|
||||||
EVIDENCE: one line why`;
|
EVIDENCE: one line why`;
|
||||||
|
|
||||||
const result = await this.pi.execute(verifyPrompt, undefined, 120_000);
|
const result = await this.pi.execute(verifyPrompt, undefined, 120_000, projectDir);
|
||||||
|
|
||||||
if (!result.success) {
|
if (!result.success) {
|
||||||
return { done: false, evidence: "Verification check failed to run" };
|
return { done: false, evidence: "Verification check failed to run" };
|
||||||
@@ -371,7 +468,22 @@ EVIDENCE: one line why`;
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Cached section lookup per project
|
private async gitPull(projectDir: string, projectName: string) {
|
||||||
|
try {
|
||||||
|
const proc = Bun.spawn({
|
||||||
|
cmd: ["git", "-C", projectDir, "pull", "--ff-only"],
|
||||||
|
stdout: "pipe",
|
||||||
|
stderr: "pipe",
|
||||||
|
});
|
||||||
|
await proc.exited;
|
||||||
|
logger.debug("task-loop", `Git pull: ${projectName}`);
|
||||||
|
} catch {
|
||||||
|
logger.debug("task-loop", `Git pull failed for ${projectName}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── Section helpers ─────────────────────────────────────────────────
|
||||||
|
|
||||||
private async findSectionCached(projectGid: string, sectionName: string): Promise<AsanaSection | null> {
|
private async findSectionCached(projectGid: string, sectionName: string): Promise<AsanaSection | null> {
|
||||||
const cached = this.sectionCache.get(projectGid);
|
const cached = this.sectionCache.get(projectGid);
|
||||||
let sections: AsanaSection[];
|
let sections: AsanaSection[];
|
||||||
@@ -403,22 +515,37 @@ EVIDENCE: one line why`;
|
|||||||
}
|
}
|
||||||
|
|
||||||
private async getTasksFromSections(projectGid: string, sectionNames: string[]): Promise<AsanaTask[]> {
|
private async getTasksFromSections(projectGid: string, sectionNames: string[]): Promise<AsanaTask[]> {
|
||||||
|
const allTasks: AsanaTask[] = [];
|
||||||
for (const name of sectionNames) {
|
for (const name of sectionNames) {
|
||||||
const section = await this.findSectionCached(projectGid, name);
|
const section = await this.findSectionCached(projectGid, name);
|
||||||
if (!section) continue;
|
if (!section) continue;
|
||||||
const tasks = await this.asana.getSectionTasks(section.gid);
|
const tasks = await this.asana.getSectionTasks(section.gid);
|
||||||
const incomplete = tasks.filter((t) => !t.completed);
|
const incomplete = tasks.filter((t) => !t.completed);
|
||||||
if (incomplete.length > 0) return incomplete;
|
allTasks.push(...incomplete);
|
||||||
}
|
}
|
||||||
return [];
|
return allTasks;
|
||||||
}
|
}
|
||||||
|
|
||||||
private buildTaskContext(task: AsanaTask): string {
|
// ─── Context building ───────────────────────────────────────────────
|
||||||
|
|
||||||
|
private buildTaskContext(task: AsanaTask, projectDir: string, projectName: string): string {
|
||||||
const parts = [
|
const parts = [
|
||||||
`Asana Task: ${task.name}`,
|
`Asana Task: ${task.name}`,
|
||||||
task.notes ? `Description: ${task.notes}` : "",
|
task.notes ? `Description: ${task.notes}` : "",
|
||||||
task.due_on ? `Due: ${task.due_on}` : "",
|
task.due_on ? `Due: ${task.due_on}` : "",
|
||||||
task.tags?.length > 0 ? `Tags: ${task.tags.map((t) => t.name).join(", ")}` : "",
|
task.tags?.length > 0 ? `Tags: ${task.tags.map((t) => t.name).join(", ")}` : "",
|
||||||
|
"",
|
||||||
|
`You are working on the "${projectName}" codebase.`,
|
||||||
|
`The code is at: ${projectDir}`,
|
||||||
|
"",
|
||||||
|
`IMPORTANT RULES:`,
|
||||||
|
`- Read relevant files before making any changes`,
|
||||||
|
`- Make small, focused changes that fix the specific bug`,
|
||||||
|
`- Test your changes (run type checks, linters, or curl endpoints)`,
|
||||||
|
`- Commit with a clear message referencing the Asana task`,
|
||||||
|
`- Do NOT modify .env files or credentials`,
|
||||||
|
`- Do NOT run destructive commands (rm -rf, DROP DATABASE)`,
|
||||||
|
`- If the bug is in the database data (not code), describe what SQL fix is needed but do NOT run it`,
|
||||||
];
|
];
|
||||||
return parts.filter(Boolean).join("\n");
|
return parts.filter(Boolean).join("\n");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,46 @@
|
|||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the sec
|
||||||
|
Network error: The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the sec
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the sec
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the sec
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the sec
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the sec
|
||||||
|
Network error: The operation timed out.
|
||||||
|
Network error: The operation timed out.
|
||||||
|
|||||||
@@ -1,2 +1,9 @@
|
|||||||
🤖 CharityRight DevOps Bot starting...
|
🤖 CharityRight DevOps Bot starting...
|
||||||
✅ Bot online — polling for messages
|
✅ Bot online — polling for messages
|
||||||
|
🤖 CharityRight DevOps Bot starting...
|
||||||
|
✅ Bot online — polling for messages
|
||||||
|
🤖 CharityRight DevOps Bot starting...
|
||||||
|
✅ Bot online — polling for messages
|
||||||
|
🤖 CharityRight DevOps Bot starting...
|
||||||
|
✅ Bot online — polling for messages
|
||||||
|
⚠️ 409 conflict — waiting 10s...
|
||||||
|
|||||||
Reference in New Issue
Block a user