From 876a278afdffbc07189738a2fb23c4017cd7e8b4 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 20 Jul 2026 00:18:06 -0700 Subject: [PATCH] fix(engine): prevent workflow boundary restart loops (#2360) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Workflow tasks no longer restart or become stranded in Planning when the graph moves through replan and review boundaries. The executor now distinguishes its own synchronous column transition from an external cancellation, while preserving the existing hard-cancel behavior for user and unrelated engine moves. Existing advanced tasks left in Planning are recovered from durable worktree and graph-pin evidence: completed work advances through the normal review handoff, and incomplete remediation resumes at its pinned execution column. A shared synchronous reservation keeps Planning and recovery mutually exclusive, and Planning excludes advanced rows so they cannot consume capacity in a repeated claim/skip loop. ## Validation - 238 affected engine tests passed, including graph-boundary cancellation, planner eligibility, ownership races, and advanced-task recovery coverage. - `pnpm --filter @fusion/engine typecheck` - `pnpm verify:fast` — workspace build, CLI bundle, and real `/api/health` boot smoke passed. ## Summary by CodeRabbit * **Bug Fixes** * Workflow graph tasks now continue running correctly when crossing workflow column boundaries. * Improved recovery of interrupted advanced-triage tasks, including completed and in-progress work. * Prevented duplicate triage dispatches and protected tasks from competing recovery and planning actions. * Added safeguards for task state changes during recovery and maintenance operations. --- ...or-archive-releases-active-session.test.ts | 51 ++++++ .../self-healing-advanced-triage.test.ts | 167 ++++++++++++++++++ .../self-healing-db-corruption.test.ts | 1 + packages/engine/src/__tests__/triage.test.ts | 41 +++++ packages/engine/src/executor.ts | 32 +++- .../engine/src/runtimes/in-process-runtime.ts | 3 +- packages/engine/src/self-healing.ts | 109 ++++++++++++ packages/engine/src/triage.ts | 39 +++- 8 files changed, 433 insertions(+), 10 deletions(-) create mode 100644 packages/engine/src/__tests__/self-healing-advanced-triage.test.ts diff --git a/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts b/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts index b67931032c..0f815b5b16 100644 --- a/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts +++ b/packages/engine/src/__tests__/executor-archive-releases-active-session.test.ts @@ -165,3 +165,54 @@ describe("archiving a task releases its active-session registry entries (FN-7717 ).not.toThrow(); }); }); + +describe("workflow graph column boundaries do not abort their own run", () => { + it.each(["triage", "in-review"] as const)( + "keeps the graph runner alive across in-progress → %s", + async (to) => { + const { executor, store } = makeExecutor(); + const task = { id: `TASK-GRAPH-${to}`, column: to } as any; + const abortSpy = vi + .spyOn(executor as any, "awaitAbortInFlightTaskWork") + .mockResolvedValue(undefined); + (store as any).moveTask = vi.fn(async (_taskId: string, column: string) => { + store.emit("task:moved", { + task: { ...task, column }, + from: "in-progress", + to: column, + source: "engine", + }); + }); + + (executor as any).graphRouting.add(task.id); + try { + await (executor as any).buildColumnBoundaryHooks(task).moveTask(to, { + fromColumn: "in-progress", + nodeId: to === "triage" ? "plan-replan" : "review", + }); + await Promise.resolve(); + + expect(abortSpy).not.toHaveBeenCalled(); + } finally { + (executor as any).graphRouting.delete(task.id); + } + }, + ); + + it("still aborts an external engine move away from in-progress", async () => { + const { executor, store } = makeExecutor(); + const abortSpy = vi + .spyOn(executor as any, "awaitAbortInFlightTaskWork") + .mockResolvedValue(undefined); + + store.emit("task:moved", { + task: { id: "TASK-EXTERNAL", column: "todo" }, + from: "in-progress", + to: "todo", + source: "engine", + }); + await Promise.resolve(); + + expect(abortSpy).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-advanced-triage.test.ts b/packages/engine/src/__tests__/self-healing-advanced-triage.test.ts new file mode 100644 index 0000000000..8faf1234b2 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-advanced-triage.test.ts @@ -0,0 +1,167 @@ +import { EventEmitter } from "node:events"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Settings, Task, TaskStore } from "@fusion/core"; +import { activeSessionRegistry } from "../active-session-registry.js"; +import { SelfHealingManager } from "../self-healing.js"; + +function task(id: string, overrides: Partial = {}): Task { + return { + id, + title: id, + description: id, + column: "triage", + status: null, + paused: false, + worktree: `/tmp/${id}`, + workflowIrPinNodeId: "code-review-remediation", + workflowIrPinColumnId: "in-progress", + dependencies: [], + steps: [{ name: "Fix review", status: "pending" }], + currentStep: 0, + log: [], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, + } as Task; +} + +function storeFor(tasks: Task[]): TaskStore & EventEmitter { + const rows = new Map(tasks.map((entry) => [entry.id, entry])); + return Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false }) as Settings), + listTasks: vi.fn(async () => [...rows.values()]), + getTask: vi.fn(async (id: string) => rows.get(id)), + moveTask: vi.fn(async (id: string, column: Task["column"]) => { + const current = rows.get(id)!; + rows.set(id, { ...current, column }); + return rows.get(id); + }), + moveTaskIf: vi.fn(async ( + id: string, + column: Task["column"], + predicate: (live: Task) => boolean | Promise, + ) => { + const current = rows.get(id)!; + if (!await predicate(current)) return { task: current, moved: false }; + const moved = { ...current, column } as Task; + rows.set(id, moved); + return { task: moved, moved: true }; + }), + logEntry: vi.fn(async () => undefined), + }) as unknown as TaskStore & EventEmitter; +} + +describe("advanced workflow tasks stranded in triage", () => { + beforeEach(() => activeSessionRegistry.clear()); + + it("resumes incomplete remediation at its durable pinned column", async () => { + const stranded = task("FN-INCOMPLETE"); + const store = storeFor([stranded]); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + getExecutingTaskIds: () => new Set(), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(1); + expect(store.moveTaskIf).toHaveBeenCalledWith( + stranded.id, + "in-progress", + expect.any(Function), + expect.objectContaining({ + recoveryRehome: true, + preserveProgress: true, + preserveWorktree: true, + workflowMoveSource: "self-healing-advanced-triage", + }), + ); + }); + + it("promotes completed pinned work through the normal completion recovery seam", async () => { + const stranded = task("FN-COMPLETE", { + workflowIrPinNodeId: "merge", + workflowIrPinColumnId: undefined, + steps: [{ name: "Implement", status: "done" }], + }); + const store = storeFor([stranded]); + const recoverCompletedTask = vi.fn(async () => true); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + recoverCompletedTask, + getExecutingTaskIds: () => new Set(), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(1); + expect(recoverCompletedTask).toHaveBeenCalledWith(stranded); + expect(store.moveTaskIf).not.toHaveBeenCalled(); + }); + + it("leaves ordinary planning rows and actively-owned graph runs untouched", async () => { + const ordinary = task("FN-ORDINARY", { worktree: undefined, workflowIrPinNodeId: undefined }); + const active = task("FN-ACTIVE"); + const store = storeFor([ordinary, active]); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + getExecutingTaskIds: () => new Set([active.id]), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(0); + expect(store.moveTaskIf).not.toHaveBeenCalled(); + }); + + it("does not rehome a task claimed after discovery but before the atomic move", async () => { + const stranded = task("FN-RACING-CLAIM"); + const store = storeFor([stranded]); + let claimed = false; + const moveTaskIfMock = vi.mocked(store.moveTaskIf); + const moveTaskIf = moveTaskIfMock.getMockImplementation()!; + moveTaskIfMock.mockImplementation(async (...args) => { + claimed = true; + return moveTaskIf(...args); + }); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + getExecutingTaskIds: () => claimed ? new Set([stranded.id]) : new Set(), + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(0); + expect(store.moveTaskIf).toHaveBeenCalledOnce(); + }); + + it("does not promote completed work when planning wins the ownership reservation", async () => { + const stranded = task("FN-COMPLETED-RACING-CLAIM", { + workflowIrPinNodeId: "merge", + workflowIrPinColumnId: undefined, + steps: [{ name: "Implement", status: "done" }], + }); + const store = storeFor([stranded]); + const recoverCompletedTask = vi.fn(async () => true); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + recoverCompletedTask, + getExecutingTaskIds: () => new Set(), + reserveAdvancedTriageRecovery: () => undefined, + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(0); + expect(recoverCompletedTask).not.toHaveBeenCalled(); + }); + + it("releases the planning fence after completed recovery", async () => { + const stranded = task("FN-COMPLETED-RESERVED", { + workflowIrPinNodeId: "merge", + workflowIrPinColumnId: undefined, + steps: [{ name: "Implement", status: "done" }], + }); + const store = storeFor([stranded]); + const release = vi.fn(); + const manager = new SelfHealingManager(store, { + rootDir: "/repo", + recoverCompletedTask: vi.fn(async () => true), + getExecutingTaskIds: () => new Set(), + reserveAdvancedTriageRecovery: () => release, + }); + + expect(await manager.recoverAdvancedTriageTasks()).toBe(1); + expect(release).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-db-corruption.test.ts b/packages/engine/src/__tests__/self-healing-db-corruption.test.ts index ab07bc9f4a..156bfe8f9e 100644 --- a/packages/engine/src/__tests__/self-healing-db-corruption.test.ts +++ b/packages/engine/src/__tests__/self-healing-db-corruption.test.ts @@ -49,6 +49,7 @@ const BATCH1_METHODS = [ const BATCH2_METHODS = [ "recoverCompletedTasks", "recoverStrandedCompletedTodoTasks", + "recoverAdvancedTriageTasks", "recoverStaleIncompleteReviewTasks", "recoverReviewTasksWithFailedPreMergeSteps", "recoverInterruptedMergingTasks", diff --git a/packages/engine/src/__tests__/triage.test.ts b/packages/engine/src/__tests__/triage.test.ts index 9f87c2143d..2d576e8b0d 100644 --- a/packages/engine/src/__tests__/triage.test.ts +++ b/packages/engine/src/__tests__/triage.test.ts @@ -1536,6 +1536,20 @@ describe("TriageProcessor", () => { expect(processor).toBeInstanceOf(TriageProcessor); }); + it("uses a synchronous reservation to keep planning and advanced recovery mutually exclusive", async () => { + const task = createTriageTask({ id: "FN-RECOVERY-RESERVED" }); + const release = processor.tryReserveAdvancedRecovery(task.id); + expect(release).toBeTypeOf("function"); + expect(processor.getProcessingTaskIds()).toContain(task.id); + expect(processor.getPlanningTaskIds()).not.toContain(task.id); + + await processor.specifyTask(task); + expect(store.getTask).not.toHaveBeenCalled(); + + release?.(); + expect(processor.getProcessingTaskIds()).not.toContain(task.id); + }); + /* FNXC:OriginalDescriptionInPrompt 2026-07-14-23:35: finalizeApprovedTask must inject ## Original Description with the task description @@ -1730,6 +1744,33 @@ Planner rewrote mission without the raw request. expect(specifySpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-200" })); }); + it("does not repeatedly dispatch a triage row that already has executor advancement evidence", async () => { + const tasks: Task[] = [ + createTriageTask({ id: "FN-ADVANCED", worktree: "/tmp/fusion-fn-advanced" }), + createTriageTask({ id: "FN-UNPLANNED" }), + ]; + const triageStore = createMockStore({ + listTasks: vi.fn().mockResolvedValue(tasks), + getSettings: vi.fn().mockResolvedValue({ + maxConcurrent: 10, + maxTriageConcurrent: 10, + pollIntervalMs: 10_000, + groupOverlappingFiles: false, + autoMerge: true, + }), + }); + const triageProcessor = new TriageProcessor(triageStore, rootDir); + const specifySpy = vi + .spyOn(triageProcessor, "specifyTask") + .mockResolvedValue(undefined); + + (triageProcessor as any).running = true; + await (triageProcessor as any).poll(); + + expect(specifySpy).toHaveBeenCalledOnce(); + expect(specifySpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-UNPLANNED" })); + }); + /* FNXC:GlobalConcurrencyControls 2026-07-14-18:30: When an in-progress executor already counts toward the live running-agent total, triage must leave room under the global cap instead of filling maxTriageConcurrent purely from semaphore.availableCount. diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 226d74ff56..f3721eaa4a 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1660,6 +1660,13 @@ export class TaskExecutor { * Prevents the task:moved handler from dispatching execute() before the * bounce finishes its own dispatch. */ private workflowRerunPending = new Set(); + /** + * Task ids whose current `task:moved` event is being emitted by this + * executor's workflow column-boundary hook. The store emits synchronously, + * so this narrowly distinguishes a graph's own transition from an external + * engine/user move that must still hard-cancel the active run. + */ + private workflowBoundaryMovesInFlight = new Set(); /** FN-5256: in-flight session-disposal promises keyed by taskId. The * task:moved (away from in-progress) and task:deleted listeners populate * this so a fast re-dispatch (task:moved → in-progress) awaits the prior @@ -3038,6 +3045,12 @@ export class TaskExecutor { }), ); } else if (from === "in-progress") { + if (this.workflowBoundaryMovesInFlight.has(task.id) && this.graphRouting.has(task.id)) { + executorLog.log( + `[event:task:moved] Preserving graph run for ${task.id} across its own ${from} → ${to} boundary`, + ); + return; + } this.trackTaskDisposal( task.id, this.awaitAbortInFlightTaskWork(task.id, `parent moved from in-progress to ${to}`, { @@ -5823,13 +5836,18 @@ export class TaskExecutor { // row fields so an ordinary requeue re-resolves the CURRENT IR fresh. clearPin: pinPersistence.clearPin, moveTask: async (toColumn, ctx) => { - await this.store.moveTask(task.id, toColumn, { - moveSource: "engine", - workflowMoveSource: "workflow-graph", - bypassGuards: true, - preserveProgress: true, - workflowMoveMetadata: { fromColumn: ctx.fromColumn, nodeId: ctx.nodeId }, - }); + this.workflowBoundaryMovesInFlight.add(task.id); + try { + await this.store.moveTask(task.id, toColumn, { + moveSource: "engine", + workflowMoveSource: "workflow-graph", + bypassGuards: true, + preserveProgress: true, + workflowMoveMetadata: { fromColumn: ctx.fromColumn, nodeId: ctx.nodeId }, + }); + } finally { + this.workflowBoundaryMovesInFlight.delete(task.id); + } }, emitAudit: async (event) => { await this.store.recordRunAuditEvent?.({ diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 6a1602785b..9207b76024 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -1049,7 +1049,8 @@ export class InProcessRuntime clearPhantomExecutorBinding: (taskId: string, options?: { preserveWorktrees?: boolean }) => this.executor?.clearPhantomExecutorBinding(taskId, options), listWorktreeHolders: () => this.executor?.listWorktreeHolders() ?? [], recoverApprovedTriageTask: (task) => this.triageProcessor?.recoverApprovedTask(task) ?? Promise.resolve(false), - getPlanningTaskIds: () => this.triageProcessor?.getProcessingTaskIds() ?? new Set(), + getPlanningTaskIds: () => this.triageProcessor?.getPlanningTaskIds() ?? new Set(), + reserveAdvancedTriageRecovery: (taskId) => this.triageProcessor?.tryReserveAdvancedRecovery(taskId), evictStaleTriageProcessing: () => this.triageProcessor?.evictStaleProcessing() ?? new Set(), enqueueMerge: this.mergeEnqueuer ? (taskId: string) => this.mergeEnqueuer?.(taskId) ?? false : undefined, requeueForAutoMerge: this.mergeEnqueuer ? (taskId: string) => this.mergeEnqueuer?.(taskId) ?? false : undefined, diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index b27cb8e6c1..731001c82f 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -296,6 +296,8 @@ export interface SelfHealingOptions { * Used to avoid recovering active triage sessions. */ getPlanningTaskIds?: () => Set; + /** Atomically fence planner ownership while advanced triage recovery runs. */ + reserveAdvancedTriageRecovery?: (taskId: string) => (() => void) | undefined; /** * Evict tasks from the triage processor's `processing` set that have been * there longer than the staleness threshold (hung promises from stuck kills). @@ -1377,6 +1379,7 @@ export class SelfHealingManager { { name: "no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures().then(() => undefined) }, { name: "completed-tasks", fn: () => this.recoverCompletedTasks().then(() => undefined) }, { name: "recover-stranded-completed-todo", fn: () => this.recoverStrandedCompletedTodoTasks().then(() => undefined) }, + { name: "recover-advanced-triage", fn: () => this.recoverAdvancedTriageTasks().then(() => undefined) }, { name: "stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks().then(() => undefined) }, { name: "failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps().then(() => undefined) }, { name: "missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures().then(() => undefined) }, @@ -2672,6 +2675,7 @@ export class SelfHealingManager { }, { name: "recover-completed-tasks", fn: () => this.recoverCompletedTasks() }, { name: "recover-stranded-completed-todo", fn: () => this.recoverStrandedCompletedTodoTasks() }, + { name: "recover-advanced-triage", fn: () => this.recoverAdvancedTriageTasks() }, { name: "recover-stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks() }, { name: "recover-failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps() }, { name: "recover-missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures() }, @@ -3009,6 +3013,111 @@ export class SelfHealingManager { } } + /** + * Re-home workflow-graph tasks stranded in the planner column after the + * executor aborted its own column-boundary move. A worktree plus a durable + * graph pin is the proof that this is advanced execution state, not an + * ordinary triage card. Completed work goes through the normal review + * recovery seam; incomplete remediation resumes at its pinned column. + */ + async recoverAdvancedTriageTasks(): Promise { + try { + const settings = await this.store.getSettings(); + if (settings.globalPause || settings.enginePaused) return 0; + + const tasks = await this.store.listTasks({ column: "triage", slim: true }); + const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const planningIds = this.options.getPlanningTaskIds?.() ?? new Set(); + const candidates = tasks.filter((task) => + task.column === "triage" + && task.status == null + && !task.paused + && !task.error + && Boolean(task.worktree) + && Boolean(task.workflowIrPinNodeId) + && !executingIds.has(task.id) + && !planningIds.has(task.id) + && this.options.isTaskActive?.(task.id) !== true + && !(task.worktree && activeSessionRegistry.isPathActive(task.worktree)), + ); + + let recovered = 0; + for (const snapshot of candidates) { + const releaseReservation = this.options.reserveAdvancedTriageRecovery?.(snapshot.id); + if (this.options.reserveAdvancedTriageRecovery && !releaseReservation) continue; + try { + const live = await this.store.getTask(snapshot.id); + if ( + live.column !== "triage" + || live.status != null + || live.paused + || live.error + || !live.worktree + || !live.workflowIrPinNodeId + || (this.options.getExecutingTaskIds?.() ?? new Set()).has(live.id) + || this.options.isTaskActive?.(live.id) === true + || activeSessionRegistry.isPathActive(live.worktree) + ) { + continue; + } + + const steps = live.steps ?? []; + const complete = steps.length > 0 + && steps.every((step) => step.status === "done" || step.status === "skipped"); + if (complete) { + if (!this.options.recoverCompletedTask) continue; + if (await this.options.recoverCompletedTask(live)) recovered++; + continue; + } + + const resumeColumn = live.workflowIrPinColumnId; + if (!resumeColumn || resumeColumn === "triage") continue; + const moved = await this.store.moveTaskIf(live.id, resumeColumn, (current) => + current.column === "triage" + && current.status == null + && !current.paused + && !current.error + && current.worktree === live.worktree + && current.workflowIrPinNodeId === live.workflowIrPinNodeId + && current.workflowIrPinColumnId === resumeColumn + && !(this.options.getExecutingTaskIds?.() ?? new Set()).has(current.id) + && !(this.options.getPlanningTaskIds?.() ?? new Set()).has(current.id) + && this.options.isTaskActive?.(current.id) !== true + && !(current.worktree && activeSessionRegistry.isPathActive(current.worktree)), + { + moveSource: "engine", + workflowMoveSource: "self-healing-advanced-triage", + workflowMoveMetadata: { + reason: "graph-boundary-self-abort-recovery", + pinnedNodeId: live.workflowIrPinNodeId, + }, + recoveryRehome: true, + bypassGuards: true, + preserveProgress: true, + preserveWorktree: true, + preserveResumeState: true, + }); + if (!moved.moved) continue; + await this.store.logEntry( + live.id, + `Auto-recovered workflow task stranded in triage — resumed at pinned ${resumeColumn} column`, + ); + recovered++; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.warn(`Failed to recover advanced triage task ${snapshot.id}: ${errorMessage}`); + } finally { + releaseReservation?.(); + } + } + return recovered; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Advanced triage recovery failed: ${errorMessage}`); + return 0; + } + } + /** * Clear stale transient merge statuses when no active merger owns the task. * diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index e780b4098d..bc67b306a2 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -215,6 +215,8 @@ export class TriageProcessor { /** The interval (ms) of the currently active `setInterval` timer. */ private activePollMs: number | null = null; private processing = new Set(); + /** Synchronous ownership fence shared with advanced-triage self-healing. */ + private advancedRecoveryReservations = new Set(); /** Timestamps when tasks entered the `processing` set, for staleness detection. */ private processingSince = new Map(); private wasGlobalPaused = false; @@ -647,6 +649,16 @@ export class TriageProcessor { * still completing Plan Review → todo after a stuck-kill of the main session. */ getProcessingTaskIds(): Set { + const ids = this.getPlanningTaskIds(); + for (const taskId of this.advancedRecoveryReservations) ids.add(taskId); + return ids; + } + + /** + * Return tasks owned by actual planner work, excluding recovery reservations. + * Recovery uses this narrower view when revalidating its own reserved task. + */ + getPlanningTaskIds(): Set { const ids = new Set(this.processing); for (const taskId of this.finalizing) ids.add(taskId); for (const taskId of this.activeSubagentSessions.keys()) { @@ -656,6 +668,23 @@ export class TriageProcessor { return ids; } + /** + * Reserve a task for advanced-state recovery unless planning already owns it. + * The check-and-add is synchronous, as is specifyTask's reciprocal guard, so + * neither side can slip between ownership inspection and acquisition. + */ + tryReserveAdvancedRecovery(taskId: string): (() => void) | undefined { + if ( + this.advancedRecoveryReservations.has(taskId) + || this.processing.has(taskId) + || this.hasLivePlanningWork(taskId) + ) { + return undefined; + } + this.advancedRecoveryReservations.add(taskId); + return () => this.advancedRecoveryReservations.delete(taskId); + } + /** True when this processor still owns live work for `taskId` (main, subagent, or finalize). */ private hasLivePlanningWork(taskId: string): boolean { if (this.finalizing.has(taskId)) return true; @@ -1042,7 +1071,9 @@ export class TriageProcessor { } const eligibleTriageTasks = allTasks.filter( - (t) => t.column === "triage" && !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused + (t) => t.column === "triage" && isTaskStillInPlanningStage(t) + && !this.advancedRecoveryReservations.has(t.id) + && !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused // Skip tasks awaiting manual plan approval — they should not be auto-discovered && t.status !== "awaiting-approval" // Skip failed specifications until the user explicitly retries them. @@ -1197,7 +1228,11 @@ export class TriageProcessor { `processing` was cleared by a stuck-kill eviction race. Concurrent claim is what leaves `status:"planning"` on a todo card after Plan Review APPROVE. */ - if (this.processing.has(task.id) || this.hasLivePlanningWork(task.id)) return; + if ( + this.advancedRecoveryReservations.has(task.id) + || this.processing.has(task.id) + || this.hasLivePlanningWork(task.id) + ) return; this.processing.add(task.id); this.processingSince.set(task.id, Date.now());