fix(triage): clear specifying status on pause/stuck and add StuckTaskDetector support

- Fix pause-abort handler using updateTask({status: undefined}) which was a no-op, leaving tasks stuck in 'specifying' forever after a pause interrupts a session post-APPROVE. Changed to status: null to actually clear the field. - Apply same fix to transient-error retry and general error catch paths. - Wire StuckTaskDetector into TriageProcessor: trackTask/untrackTask/recordActivity on session lifecycle, markStuckAborted to prevent stuck kills from being reported as errors, and clear status to null on stuck-kill for next-poll retry. - Update dashboard.ts to pass stuckTaskDetector to TriageProcessor and call triageRef.current?.markStuckAborted in the shared onStuck callback. - Add 3 tests covering pause-abort status clearing and stuck detector wiring. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-07 21:51:09 -07:00
parent a11167081f
commit ec20ab1038
3 changed files with 193 additions and 20 deletions
--- a/packages/cli/src/commands/dashboard.ts
+++ b/packages/cli/src/commands/dashboard.ts
@@ -578,15 +578,6 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:

  // Start the AI engine (unless in dev mode)
  if (!opts.dev) {
-    const triage = new TriageProcessor(store, cwd, {
-      semaphore,
-      usageLimitPauser,
-      agentStore,
-      onSpecifyStart: (t) => console.log(`[engine] Specifying ${t.id}...`),
-      onSpecifyComplete: (t) => console.log(`[engine] ✓ ${t.id} → todo`),
-      onSpecifyError: (t, e) => console.log(`[engine] ✗ ${t.id}: ${e.message}`),
-    });
-
    // ── Self-healing: auto-unpause, stuck kill budgets, maintenance ─────
    const selfHealing = new SelfHealingManager(store, {
      rootDir: cwd,
@@ -595,14 +586,18 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
    });

    // ── Stuck task detector: monitors agent sessions for stagnation ────
-    // Created before the executor so it can be passed in options.
-    // The onStuck callback is wired to executor.markStuckAborted after
-    // executor creation (late-binding via closure on executorRef).
+    // Created before triage/executor so it can be passed in options.
+    // The onStuck callback is wired via late-binding closures on triageRef
+    // and executorRef to avoid circular construction order dependencies.
    const executorRef: { current: TaskExecutor | null } = { current: null };
+    const triageRef: { current: TriageProcessor | null } = { current: null };
    const stuckTaskDetector = new StuckTaskDetector(store, {
      beforeRequeue: (taskId) => selfHealing.checkStuckBudget(taskId),
      onLoopDetected: (event) => executorRef.current?.handleLoopDetected(event) ?? Promise.resolve(false),
      onStuck: (event) => {
+        // Notify whichever component owns this task (triage or executor).
+        // Both check their own tracking sets so only the owner acts.
+        triageRef.current?.markStuckAborted(event.taskId);
        executorRef.current?.markStuckAborted(event.taskId, event.shouldRequeue);
        console.log(
          `[engine] ⚠ ${event.taskId} stuck (${event.reason}) — ` +
@@ -613,6 +608,17 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
      },
    });

+    const triage = new TriageProcessor(store, cwd, {
+      semaphore,
+      usageLimitPauser,
+      stuckTaskDetector,
+      agentStore,
+      onSpecifyStart: (t) => console.log(`[engine] Specifying ${t.id}...`),
+      onSpecifyComplete: (t) => console.log(`[engine] ✓ ${t.id} → todo`),
+      onSpecifyError: (t, e) => console.log(`[engine] ✗ ${t.id}: ${e.message}`),
+    });
+    triageRef.current = triage;
+
    const executor = new TaskExecutor(store, cwd, {
      semaphore,
      pool,
--- a/packages/engine/src/triage.test.ts
+++ b/packages/engine/src/triage.test.ts
@@ -1734,3 +1734,137 @@ describe("stale approval detection", () => {
    await rm(rootDir, { recursive: true, force: true });
  });
 });
+
+describe("pause-abort status clearing (bug fix)", () => {
+  it("clears specifying status to null on global pause (not a no-op)", async () => {
+    const settingsListeners: Array<(e: any) => void> = [];
+
+    const store = {
+      on: vi.fn((event: string, cb: (e: any) => void) => {
+        if (event === "settings:updated") settingsListeners.push(cb);
+      }),
+      getTask: vi.fn().mockResolvedValue({ ...mockTaskDetail }),
+      getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, pollIntervalMs: 10000, groupOverlappingFiles: false, autoMerge: true } as Settings),
+      listTasks: vi.fn().mockResolvedValue([]),
+      updateTask: vi.fn().mockResolvedValue(undefined),
+      logEntry: vi.fn().mockResolvedValue(undefined),
+      appendAgentLog: vi.fn().mockResolvedValue(undefined),
+      parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]),
+    } as unknown as TaskStore;
+
+    let resolveDispose: () => void;
+    const disposePromise = new Promise<void>((r) => { resolveDispose = r; });
+    mockCreateKbAgent.mockResolvedValue({
+      session: {
+        state: {},
+        sessionManager: {},
+        prompt: vi.fn().mockReturnValue(disposePromise),
+        dispose: vi.fn().mockImplementation(() => resolveDispose()),
+        navigateTree: vi.fn(),
+      },
+    });
+
+    const task: Task = { id: "FN-001", description: "test", column: "triage", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "", updatedAt: "" };
+    const processor = new TriageProcessor(store, "/tmp/root");
+    const specifyPromise = processor.specifyTask(task);
+
+    await new Promise((r) => setTimeout(r, 20));
+
+    for (const fn of settingsListeners) {
+      fn({ settings: { globalPause: true }, previous: { globalPause: false } });
+    }
+
+    await specifyPromise;
+
+    // Status must be set to null so the next poll can retry (old bug: undefined was a no-op)
+    const nullStatusCall = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls
+      .find((c) => c[1]?.status === null);
+    expect(nullStatusCall).toBeDefined();
+
+    const undefinedStatusCall = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls
+      .find((c) => "status" in c[1] && c[1].status === undefined);
+    expect(undefinedStatusCall).toBeUndefined();
+  });
+});
+
+describe("stuck task detector integration", () => {
+  it("markStuckAborted clears specifying status to null for retry", async () => {
+    const store = {
+      on: vi.fn(),
+      getTask: vi.fn().mockResolvedValue({ ...mockTaskDetail }),
+      getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, pollIntervalMs: 10000, groupOverlappingFiles: false, autoMerge: true } as Settings),
+      listTasks: vi.fn().mockResolvedValue([]),
+      updateTask: vi.fn().mockResolvedValue(undefined),
+      logEntry: vi.fn().mockResolvedValue(undefined),
+      appendAgentLog: vi.fn().mockResolvedValue(undefined),
+      parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]),
+    } as unknown as TaskStore;
+
+    let resolveDispose: () => void;
+    let mockDispose: ReturnType<typeof vi.fn>;
+    const disposePromise = new Promise<void>((r) => { resolveDispose = r; });
+    mockDispose = vi.fn().mockImplementation(() => resolveDispose());
+    mockCreateKbAgent.mockResolvedValue({
+      session: {
+        state: {},
+        sessionManager: {},
+        prompt: vi.fn().mockReturnValue(disposePromise),
+        dispose: mockDispose,
+        navigateTree: vi.fn(),
+      },
+    });
+
+    const task: Task = { id: "FN-001", description: "test", column: "triage", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "", updatedAt: "" };
+    const processor = new TriageProcessor(store, "/tmp/root");
+    const specifyPromise = processor.specifyTask(task);
+
+    await new Promise((r) => setTimeout(r, 20));
+
+    // Stuck detector marks task then disposes the session (simulating StuckTaskDetector.killAndRetry)
+    processor.markStuckAborted("FN-001");
+    mockDispose();
+
+    await specifyPromise;
+
+    // Status cleared to null so next poll retries
+    const nullStatusCall = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls
+      .find((c) => c[1]?.status === null);
+    expect(nullStatusCall).toBeDefined();
+  });
+
+  it("tracks and untracks sessions with stuckTaskDetector", async () => {
+    const trackTask = vi.fn();
+    const untrackTask = vi.fn();
+    const recordActivity = vi.fn();
+    const mockDetector = { trackTask, untrackTask, recordActivity } as any;
+
+    const store = {
+      on: vi.fn(),
+      getTask: vi.fn().mockResolvedValue({ ...mockTaskDetail }),
+      getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, pollIntervalMs: 10000, groupOverlappingFiles: false, autoMerge: true } as Settings),
+      listTasks: vi.fn().mockResolvedValue([]),
+      updateTask: vi.fn().mockResolvedValue(undefined),
+      logEntry: vi.fn().mockResolvedValue(undefined),
+      appendAgentLog: vi.fn().mockResolvedValue(undefined),
+      parseDependenciesFromPrompt: vi.fn().mockResolvedValue([]),
+    } as unknown as TaskStore;
+
+    mockCreateKbAgent.mockResolvedValue({
+      session: {
+        state: {},
+        sessionManager: {},
+        prompt: vi.fn().mockResolvedValue(undefined),
+        dispose: vi.fn(),
+        navigateTree: vi.fn(),
+      },
+    });
+
+    const task: Task = { id: "FN-001", description: "test", column: "triage", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "", updatedAt: "" };
+    const processor = new TriageProcessor(store, "/tmp/root", { stuckTaskDetector: mockDetector });
+    await processor.specifyTask(task);
+
+    expect(trackTask).toHaveBeenCalledWith("FN-001", expect.objectContaining({ dispose: expect.any(Function) }));
+    expect(untrackTask).toHaveBeenCalledWith("FN-001");
+    expect(recordActivity).toHaveBeenCalled();
+  });
+});
--- a/packages/engine/src/triage.ts
+++ b/packages/engine/src/triage.ts
@@ -26,6 +26,7 @@ import {
 import { isTransientError, isSilentTransientError } from "./transient-error-detector.js";
 import { withRateLimitRetry } from "./rate-limit-retry.js";
 import { computeRecoveryDecision, formatDelay, MAX_RECOVERY_RETRIES } from "./recovery-policy.js";
+import type { StuckTaskDetector } from "./stuck-task-detector.js";

 export const TRIAGE_SYSTEM_PROMPT = `You are a task specification agent for "kb", an AI-orchestrated task board.

@@ -233,6 +234,8 @@ export interface TriageProcessorOptions {
  semaphore?: AgentSemaphore;
  /** Usage limit pauser — triggers global pause when API limits are detected. */
  usageLimitPauser?: UsageLimitPauser;
+  /** Stuck task detector — monitors triage sessions for stagnation and triggers recovery. */
+  stuckTaskDetector?: StuckTaskDetector;
  onSpecifyStart?: (task: Task) => void;
  onSpecifyComplete?: (task: Task) => void;
  onSpecifyError?: (task: Task, error: Error) => void;
@@ -264,6 +267,8 @@ export class TriageProcessor {
  private activeSessions = new Map<string, { dispose: () => void }>();
  /** Tasks aborted due to globalPause (to avoid reporting as errors). */
  private pauseAborted = new Set<string>();
+  /** Tasks killed by the stuck task detector (to avoid reporting as errors). */
+  private stuckAborted = new Set<string>();

  /**
   * @param store — Task store instance (also used to listen for `settings:updated` events)
@@ -288,6 +293,7 @@ export class TriageProcessor {
            `Global pause — terminating triage session for ${taskId}`,
          );
          this.pauseAborted.add(taskId);
+          this.options.stuckTaskDetector?.untrackTask(taskId);
          session.dispose();
        }
      }
@@ -342,6 +348,15 @@ export class TriageProcessor {
    triageLog.log("Processor stopped");
  }

+  /**
+   * Mark a task as stuck-aborted so the catch block knows not to treat
+   * the disposed session as a genuine failure.
+   * Called by the stuck task detector's onStuck callback.
+   */
+  markStuckAborted(taskId: string): void {
+    this.stuckAborted.add(taskId);
+  }
+
  /**
   * If `newIntervalMs` differs from the currently active timer, restart
   * the `setInterval` so the new cadence takes effect immediately.
@@ -451,14 +466,18 @@ export class TriageProcessor {
        // tasks waiting in the queue don't appear as "specifying".
        await this.store.updateTask(task.id, { status: "specifying" });

+        const stuckDetector = this.options.stuckTaskDetector;
+
        const agentLogger = new AgentLogger({
          store: this.store,
          taskId: task.id,
          agent: "triage",
-          onAgentText: this.options.onAgentText
-            ? (id, delta) => this.options.onAgentText!(id, delta)
-            : undefined,
+          onAgentText: (id, delta) => {
+            stuckDetector?.recordActivity(task.id);
+            this.options.onAgentText?.(id, delta);
+          },
          onAgentTool: (_id, name) => {
+            stuckDetector?.recordActivity(task.id);
            triageLog.log(`${task.id} tool: ${name}`);
          },
        });
@@ -561,6 +580,10 @@ export class TriageProcessor {
        // Register session so the global pause listener can terminate it
        this.activeSessions.set(task.id, session);

+        // Register with stuck task detector for heartbeat monitoring
+        stuckDetector?.trackTask(task.id, session);
+        stuckDetector?.recordActivity(task.id);
+
        try {
          // Read attachment contents for inlining in prompt
          const { attachmentContents, imageContents } =
@@ -734,6 +757,7 @@ export class TriageProcessor {
          }
        } finally {
          this.activeSessions.delete(task.id);
+          stuckDetector?.untrackTask(task.id);
          await agentLogger.flush();
          session.dispose();
        }
@@ -761,8 +785,16 @@ export class TriageProcessor {
        // Pause (global or engine) — clear specifying status without reporting an error
        this.pauseAborted.delete(task.id);
        triageLog.log(`${task.id} aborted by pause — clearing status`);
-        // For re-specification, restore needs-respecify status
-        const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : undefined;
+        // For re-specification, restore needs-respecify status; otherwise clear to null
+        // so the next poll can re-pick this task up.
+        const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
+        await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
+      } else if (this.stuckAborted.has(task.id)) {
+        // Stuck task detector killed this session — clear specifying status so the
+        // next poll retries the task from scratch without reporting an error.
+        this.stuckAborted.delete(task.id);
+        triageLog.log(`${task.id} killed by stuck detector — clearing status for retry`);
+        const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
        await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
      } else {
        // Check if the error is a usage-limit error and trigger global pause
@@ -787,7 +819,7 @@ export class TriageProcessor {
              triageLog.warn(`⚡ ${task.id} transient error during triage — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}: ${err.message}`);
              await this.store.logEntry(task.id, `Transient error during specification (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`).catch(() => {});
            }
-            const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : undefined;
+            const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
            await this.store.updateTask(task.id, {
              status: restoreStatus,
              recoveryRetryCount: decision.nextState.recoveryRetryCount,
@@ -807,8 +839,9 @@ export class TriageProcessor {
          this.options.onSpecifyError?.(task, err);
          return;
        }
-        // For re-specification, restore needs-respecify status so it can be retried
-        const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : undefined;
+        // For re-specification, restore needs-respecify status so it can be retried;
+        // otherwise clear to null so the next poll can re-pick the task up.
+        const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : null;
        await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
        triageLog.error(`✗ ${task.id} specification failed:`, err.message);
        this.options.onSpecifyError?.(task, err);