fix: major cleanup - verify cache, section cache, rate limit retry, split sweep from cleanup

- Separate cleanupCompletedMyTasks (fast, no Pi) from sweepIncompleteTasks (with Pi verify)
- Cache verified-not-done tasks for 30min to avoid re-checking every cycle
- Cache project sections for 5min to reduce API calls
- Add rate limit (429) retry with Retry-After header
- Add server error (5xx) retry
- Remove assignee:null - only use assignee_status:later
- Keep tasks assigned to user always

pi-worker/src/asana/client.ts

@@ -16,20 +16,37 @@ export class AsanaClient {
     };
   }
 
-  private async request<T>(path: string, options: RequestInit = {}): Promise<T> {
+  private async request<T>(path: string, options: RequestInit = {}, retries = 3): Promise<T> {
     const url = `${ASANA_BASE}${path}`;
-    const res = await fetch(url, {
-      ...options,
-      headers: { ...this.headers, ...options.headers },
-    });
 
-    if (!res.ok) {
+    for (let attempt = 1; attempt <= retries; attempt++) {
-      const body = await res.text();
+      const res = await fetch(url, {
-      throw new Error(`Asana API ${res.status}: ${body}`);
+        ...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 };
+    throw new Error(`Asana API failed after ${retries} retries`);
-    return json.data;
   }
 
   // Get all workspaces

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<string, number> = new Map(); // gid -> timestamp
+  private readonly VERIFY_CACHE_TTL = 30 * 60 * 1000; // 30 min
+
+  // Cache: sections per project
+  private sectionCache: Map<string, { sections: AsanaSection[]; ts: number }> = 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
+      // Step 1: Quick cleanup - move completed My Tasks to Done sections
-      await this.sweepDoneTasks();
+      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...");
-
+      const tasks = await this.getTasksFromSections(this.config.asanaProjectGid, SECTIONS.TODO);
-      let 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[]) {
+  // Quick: just move completed tasks out of Recently Assigned. No Pi calls.
-    for (const name of candidates) {
+  private async cleanupCompletedMyTasks() {
-      const section = await this.asana.findSection(this.config.asanaProjectGid, name);
+    if (!this.userTaskListGid) return;
-      if (section) return section;
+
+    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<string>();
+      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 });
+        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}`);
-        await this.asana.addComment(
+        const doneSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.DONE);
-          task.gid,
+        if (doneSection) await this.asana.moveTaskToSection(task.gid, doneSection.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);
-        }
         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"
+      // Step 2: Move to In Progress
-      const inProgressSection = await this.findFirstSection(SECTIONS.IN_PROGRESS);
+      const inProgressSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.IN_PROGRESS);
-      if (inProgressSection) {
+      if (inProgressSection) await this.asana.moveTaskToSection(task.gid, inProgressSection.gid);
-        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()}`);
-      await this.asana.addComment(task.gid, `🤖 Pi Worker picked up this task.\nVerification: Not yet done on server.\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);
+        const doneSection = await this.findFirstSection(this.config.asanaProjectGid, SECTIONS.DONE);
-        if (doneSection) {
+        if (doneSection) await this.asana.moveTaskToSection(task.gid, doneSection.gid);
-          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(this.config.asanaProjectGid, SECTIONS.TODO);
-        const todoSection = await this.findFirstSection(SECTIONS.TODO);
+        if (todoSection) await this.asana.moveTaskToSection(task.gid, todoSection.gid);
-        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() {
+  // Cached section lookup per project
-    try {
+  private async findSectionCached(projectGid: string, sectionName: string): Promise<AsanaSection | null> {
-      // Get all incomplete tasks from all active sections (not just TODO)
+    const cached = this.sectionCache.get(projectGid);
-      const sweepSections = [...SECTIONS.TODO, ...SECTIONS.IN_PROGRESS];
+    let sections: AsanaSection[];
-      const allTasks: AsanaTask[] = [];
 
-      for (const sectionName of sweepSections) {
+    if (cached && Date.now() - cached.ts < this.SECTION_CACHE_TTL) {
-        const section = await this.asana.findSection(this.config.asanaProjectGid, sectionName);
+      sections = cached.sections;
-        if (!section) continue;
+    } else {
-        const tasks = await this.asana.getSectionTasks(section.gid);
+      sections = await this.asana.getSections(projectGid);
-        const incomplete = tasks.filter((t) => !t.completed);
+      this.sectionCache.set(projectGid, { sections, ts: Date.now() });
-        allTasks.push(...incomplete);
-      }
-
-      if (allTasks.length === 0) return;
-
-      // Deduplicate by GID
-      const seen = new Set<string>();
-      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}`);
     }
+
+    return sections.find((s) => s.name.toLowerCase().includes(sectionName.toLowerCase())) || null;
+  }
+
+  private async findDoneSection(projectGid: string): Promise<AsanaSection | null> {
+    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<AsanaSection | null> {
+    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<AsanaTask[]> {
     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");
   }