diff --git a/pi-worker/src/asana/client.ts b/pi-worker/src/asana/client.ts index f17067d..13240ad 100644 --- a/pi-worker/src/asana/client.ts +++ b/pi-worker/src/asana/client.ts @@ -16,20 +16,37 @@ export class AsanaClient { }; } - private async request(path: string, options: RequestInit = {}): Promise { + private async request(path: string, options: RequestInit = {}, retries = 3): 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}`); + for (let attempt = 1; attempt <= retries; attempt++) { + const res = await fetch(url, { + ...options, + headers: { ...this.headers, ...options.headers }, + }); + + if (res.status === 429) { + const retryAfter = parseInt(res.headers.get("Retry-After") || "30"); + logger.warn("asana", `Rate limited, retrying in ${retryAfter}s (attempt ${attempt}/${retries})`); + await new Promise((r) => setTimeout(r, retryAfter * 1000)); + continue; + } + + if (!res.ok) { + const body = await res.text(); + if (attempt < retries && res.status >= 500) { + logger.warn("asana", `Server error ${res.status}, retrying (attempt ${attempt}/${retries})`); + await new Promise((r) => setTimeout(r, 2000 * attempt)); + continue; + } + throw new Error(`Asana API ${res.status}: ${body}`); + } + + const json = (await res.json()) as { data: T }; + return json.data; } - const json = (await res.json()) as { data: T }; - return json.data; + throw new Error(`Asana API failed after ${retries} retries`); } // Get all workspaces diff --git a/pi-worker/src/scheduler/task-loop.ts b/pi-worker/src/scheduler/task-loop.ts index 909bb05..7a2bab1 100644 --- a/pi-worker/src/scheduler/task-loop.ts +++ b/pi-worker/src/scheduler/task-loop.ts @@ -3,13 +3,12 @@ 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"; +import type { AsanaTask, AsanaSection } from "../asana/types.js"; -// Section names in Asana board (searched in order, partial case-insensitive match) const SECTIONS = { TODO: ["Pi Worker", "To Do", "New Bugs", "Ready for Development", "Recently Assigned", "Awaiting More Info"], IN_PROGRESS: ["In Progress", "Under Review"], - DONE: ["Done", "Resolved"], + DONE: ["Done", "Resolved", "Complete", "Completed"], }; export class TaskLoop { @@ -22,6 +21,14 @@ export class TaskLoop { private userTaskListGid: string | null = null; private workspaceGid: string = ""; + // Cache: track tasks we already verified as "not done" to avoid re-checking + private verifiedNotDone: Map = new Map(); // gid -> timestamp + private readonly VERIFY_CACHE_TTL = 30 * 60 * 1000; // 30 min + + // Cache: sections per project + private sectionCache: Map = new Map(); + private readonly SECTION_CACHE_TTL = 5 * 60 * 1000; // 5 min + constructor(config: Config) { this.config = config; this.asana = new AsanaClient(config.asanaAccessToken); @@ -36,12 +43,10 @@ export class TaskLoop { 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); } @@ -66,22 +71,18 @@ export class TaskLoop { 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); + logger.info("task-loop", `Workspace "${ws.name}" has ${projects.length} project(s)`); if (projects.length > 0 && !this.config.asanaProjectGid) { - // Use first project as default this.config.asanaProjectGid = projects[0].gid; + this.workspaceGid = ws.gid; logger.info("task-loop", `Auto-selected project: ${projects[0].name} (${projects[0].gid})`); } - - this.workspaceGid = ws.gid; } - // Cache user task list for "My Tasks" / "Recently Assigned" sweep + // Cache user task list try { const me = await this.asana.getMe(); if (this.workspaceGid) { @@ -100,18 +101,20 @@ export class TaskLoop { private async poll() { if (!this.running || !this.config.asanaProjectGid) return; if (!this.lock.acquire()) { - logger.debug("task-loop", "Skipping poll - another execution in progress"); + logger.debug("task-loop", "Skipping poll — another execution in progress"); return; } try { - // First: sweep all sections for tasks that are already done - await this.sweepDoneTasks(); + // Step 1: Quick cleanup - move completed My Tasks to Done sections + await this.cleanupCompletedMyTasks(); - // Then: pick up new tasks to execute + // Step 2: Sweep incomplete tasks for already-done work (with caching) + await this.sweepIncompleteTasks(); + + // Step 3: Pick up new tasks to execute logger.info("task-loop", "Polling for new Asana tasks..."); - - let tasks = await this.getTasksFromSections(this.config.asanaProjectGid, SECTIONS.TODO); + const tasks = await this.getTasksFromSections(this.config.asanaProjectGid, SECTIONS.TODO); if (tasks.length === 0) { logger.info("task-loop", "No pending tasks found"); @@ -119,10 +122,8 @@ export class TaskLoop { 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 }); @@ -131,67 +132,172 @@ export class TaskLoop { } } - private async findFirstSection(candidates: string[]) { - for (const name of candidates) { - const section = await this.asana.findSection(this.config.asanaProjectGid, name); - if (section) return section; + // Quick: just move completed tasks out of Recently Assigned. No Pi calls. + private async cleanupCompletedMyTasks() { + if (!this.userTaskListGid) return; + + try { + const myTasks = await this.asana.getUserTasks(this.userTaskListGid); + const completedTasks = myTasks.filter((t) => t.completed); + + if (completedTasks.length === 0) return; + + logger.info("task-loop", `Cleaning ${completedTasks.length} completed tasks from My Tasks`); + + for (const task of completedTasks) { + try { + // Move to Done section in the task's project + if (task.memberships && task.memberships.length > 0) { + for (const membership of task.memberships) { + const projectGid = membership.project?.gid; + const currentSection = membership.section?.name?.toLowerCase() || ""; + + if (currentSection.includes("done") || currentSection.includes("resolved") || currentSection.includes("complete") || currentSection.includes("won't")) { + continue; + } + + if (projectGid) { + const doneSection = await this.findDoneSection(projectGid); + if (doneSection) { + await this.asana.moveTaskToSection(task.gid, doneSection.gid); + logger.info("task-loop", `Moved to ${doneSection.name}: ${task.name}`); + } + } + } + } + + // Move out of Recently Assigned + await this.asana.updateTask(task.gid, { assignee_status: "later" }); + } catch (e: any) { + logger.debug("task-loop", `Could not clean task ${task.name}: ${e.message}`); + } + } + } catch (error: any) { + logger.warn("task-loop", `My Tasks cleanup error: ${error.message}`); + } + } + + // Sweep: verify incomplete tasks - are they already done? Uses cache to skip recently checked. + private async sweepIncompleteTasks() { + try { + const sweepSections = [...SECTIONS.TODO, ...SECTIONS.IN_PROGRESS]; + const allTasks: AsanaTask[] = []; + + for (const sectionName of sweepSections) { + const section = await this.findSectionCached(this.config.asanaProjectGid, sectionName); + if (!section) continue; + const tasks = await this.asana.getSectionTasks(section.gid); + allTasks.push(...tasks.filter((t) => !t.completed)); + } + + // Also get incomplete My Tasks + if (this.userTaskListGid) { + try { + const myTasks = await this.asana.getUserTasks(this.userTaskListGid); + allTasks.push(...myTasks.filter((t) => !t.completed)); + } catch {} + } + + // Deduplicate + const seen = new Set(); + 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) + const now = Date.now(); + const tasksToVerify = uniqueTasks.filter((t) => { + const lastCheck = this.verifiedNotDone.get(t.gid); + if (lastCheck && now - lastCheck < this.VERIFY_CACHE_TTL) return false; + return true; + }); + + if (tasksToVerify.length === 0) { + 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}`); } - return null; } private async executeTask(task: AsanaTask) { try { - // STEP 1: Check if the work is already done + // Step 1: Quick pre-verify const alreadyDone = await this.verifyTaskAlreadyDone(task); if (alreadyDone.done) { - logger.info("task-loop", `Task already done: ${task.name}`, { evidence: alreadyDone.evidence }); - - await this.asana.addComment( - task.gid, - `✅ Pi Worker verified this task is **already completed** on the server.\n\nEvidence:\n${alreadyDone.evidence}` - ); - - // Move to Done and mark complete - const doneSection = await this.findFirstSection(SECTIONS.DONE); - if (doneSection) { - await this.asana.moveTaskToSection(task.gid, doneSection.gid); - } + 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 }); - logger.info("task-loop", `Task marked done (was already complete): ${task.name}`); return; } - // STEP 2: Move to "In Progress" - const inProgressSection = await this.findFirstSection(SECTIONS.IN_PROGRESS); - if (inProgressSection) { - await this.asana.moveTaskToSection(task.gid, inProgressSection.gid); - } + // 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); - // Add starting comment - await this.asana.addComment(task.gid, `🤖 Pi Worker picked up this task.\nVerification: Not yet done on server.\nStarting execution at ${new Date().toISOString()}`); + 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); - // STEP 3: Execute with Pi agent + // Step 3: Execute const result = await this.pi.execute(task.name + (task.notes ? `\n\nDetails:\n${task.notes}` : ""), context); - // STEP 4: After execution, verify the work was actually done + // Step 4: Post-verify let verified = false; if (result.success && result.toolCalls > 0) { const postCheck = await this.verifyTaskAlreadyDone(task); verified = postCheck.done; - - if (!verified) { - logger.warn("task-loop", `Task executed but verification failed: ${task.name}`); - } } - // Add completion comment const statusEmoji = result.success ? (verified ? "✅" : "⚠️") : "❌"; 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 = [ @@ -210,18 +316,13 @@ export class TaskLoop { await this.asana.addComment(task.gid, comment); if (result.success) { - const doneSection = await this.findFirstSection(SECTIONS.DONE); - if (doneSection) { - await this.asana.moveTaskToSection(task.gid, doneSection.gid); - } + 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 }); logger.info("task-loop", `Task completed: ${task.name}`, { verified }); } else { - // Move back to To Do on failure - const todoSection = await this.findFirstSection(SECTIONS.TODO); - if (todoSection) { - await this.asana.moveTaskToSection(task.gid, todoSection.gid); - } + const todoSection = await this.findFirstSection(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) { @@ -250,8 +351,8 @@ RESPOND EXACTLY: DONE: true or false EVIDENCE: one line why`; - const result = await this.pi.execute(verifyPrompt, undefined, 120_000); // 2 min timeout for verify - + const result = await this.pi.execute(verifyPrompt, undefined, 120_000); + if (!result.success) { return { done: false, evidence: "Verification check failed to run" }; } @@ -263,8 +364,6 @@ EVIDENCE: one line why`; const isDone = doneMatch ? doneMatch[1].toLowerCase() === "true" : false; const evidence = evidenceMatch ? evidenceMatch[1].trim().substring(0, 1000) : output.substring(0, 1000); - logger.info("task-loop", `Verification result for "${task.name}": done=${isDone}`, { evidence: evidence.substring(0, 200) }); - return { done: isDone, evidence }; } catch (error: any) { logger.error("task-loop", `Verification error: ${error.message}`); @@ -272,137 +371,40 @@ EVIDENCE: one line why`; } } - private async sweepDoneTasks() { - try { - // Get all incomplete tasks from all active sections (not just TODO) - const sweepSections = [...SECTIONS.TODO, ...SECTIONS.IN_PROGRESS]; - const allTasks: AsanaTask[] = []; + // Cached section lookup per project + private async findSectionCached(projectGid: string, sectionName: string): Promise { + const cached = this.sectionCache.get(projectGid); + let sections: AsanaSection[]; - for (const sectionName of sweepSections) { - const section = await this.asana.findSection(this.config.asanaProjectGid, sectionName); - if (!section) continue; - const tasks = await this.asana.getSectionTasks(section.gid); - const incomplete = tasks.filter((t) => !t.completed); - allTasks.push(...incomplete); - } - - if (allTasks.length === 0) return; - - // Deduplicate by GID - const seen = new Set(); - const uniqueTasks = allTasks.filter((t) => { - if (seen.has(t.gid)) return false; - seen.add(t.gid); - return true; - }); - - logger.info("task-loop", `Sweeping ${uniqueTasks.length} incomplete tasks for already-done work`); - - for (const task of uniqueTasks) { - 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** on the server.\n\nEvidence:\n${check.evidence}` - ); - - const doneSection = await this.findFirstSection(SECTIONS.DONE); - 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 { - logger.debug("task-loop", `Sweep: "${task.name}" not yet done`, { evidence: check.evidence }); - } - } - - // Also sweep user's "My Tasks" (Recently Assigned) - if (this.userTaskListGid) { - try { - const myTasks = await this.asana.getUserTasks(this.userTaskListGid); - - // First: clear completed tasks from "Recently Assigned" - // Move to "later" AND move to Done/Resolved section in their project - const completedInbox = myTasks.filter((t) => t.completed); - if (completedInbox.length > 0) { - logger.info("task-loop", `Clearing ${completedInbox.length} completed tasks from Recently Assigned`); - for (const task of completedInbox) { - try { - // Move to Done/Resolved section in the task's project - if (task.memberships && task.memberships.length > 0) { - for (const membership of task.memberships) { - const projectGid = membership.project?.gid; - const currentSection = membership.section?.name?.toLowerCase() || ""; - - if (currentSection.includes("done") || currentSection.includes("resolved") || currentSection.includes("won't")) { - continue; - } - - if (projectGid) { - const doneNames = ["Done", "Resolved", "Complete", "Completed"]; - let doneSection = null; - for (const name of doneNames) { - doneSection = await this.asana.findSection(projectGid, name); - if (doneSection) break; - } - - if (doneSection) { - await this.asana.moveTaskToSection(task.gid, doneSection.gid); - logger.info("task-loop", `Moved to ${doneSection.name}: ${task.name} (project: ${membership.project?.name})`); - } - } - } - } - - logger.info("task-loop", `Cleared completed task: ${task.name}`); - } catch (e: any) { - logger.debug("task-loop", `Could not clear task ${task.name}: ${e.message}`); - } - } - } - - // Then: check incomplete tasks to see if they're already done - const incompleteMyTasks = myTasks.filter((t) => !t.completed); - - if (incompleteMyTasks.length > 0) { - logger.info("task-loop", `Sweeping ${incompleteMyTasks.length} incomplete tasks from My Tasks`); - - for (const task of incompleteMyTasks) { - if (seen.has(task.gid)) continue; - - const check = await this.verifyTaskAlreadyDone(task); - - if (check.done) { - logger.info("task-loop", `My Tasks 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}` - ); - - await this.asana.updateTask(task.gid, { completed: true }); - logger.info("task-loop", `My Tasks sweep: Marked done: ${task.name}`); - } else { - logger.debug("task-loop", `My Tasks sweep: "${task.name}" not yet done`); - } - } - } - } catch (error: any) { - logger.warn("task-loop", `My Tasks sweep error: ${error.message}`); - } - } - } catch (error: any) { - logger.error("task-loop", `Sweep error: ${error.message}`); + if (cached && Date.now() - cached.ts < this.SECTION_CACHE_TTL) { + sections = cached.sections; + } else { + sections = await this.asana.getSections(projectGid); + this.sectionCache.set(projectGid, { sections, ts: Date.now() }); } + + return sections.find((s) => s.name.toLowerCase().includes(sectionName.toLowerCase())) || null; + } + + private async findDoneSection(projectGid: string): Promise { + for (const name of SECTIONS.DONE) { + const section = await this.findSectionCached(projectGid, name); + if (section) return section; + } + return null; + } + + private async findFirstSection(projectGid: string, names: string[]): Promise { + for (const name of names) { + const section = await this.findSectionCached(projectGid, name); + if (section) return section; + } + return null; } private async getTasksFromSections(projectGid: string, sectionNames: string[]): Promise { for (const name of sectionNames) { - const section = await this.asana.findSection(projectGid, name); + const section = await this.findSectionCached(projectGid, name); if (!section) continue; const tasks = await this.asana.getSectionTasks(section.gid); const incomplete = tasks.filter((t) => !t.completed); @@ -416,7 +418,7 @@ EVIDENCE: one line why`; `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(", ")}` : "", + task.tags?.length > 0 ? `Tags: ${task.tags.map((t) => t.name).join(", ")}` : "", ]; return parts.filter(Boolean).join("\n"); }