diff --git a/packages/engine/src/__tests__/executor-live-overseer-retry-gate.test.ts b/packages/engine/src/__tests__/executor-live-overseer-retry-gate.test.ts new file mode 100644 index 0000000000..25f9ef0ac7 --- /dev/null +++ b/packages/engine/src/__tests__/executor-live-overseer-retry-gate.test.ts @@ -0,0 +1,213 @@ +/** + * FNXC:PlannerOversight 2026-07-21-22:56: + * Unit coverage for FN-8471 prevention helpers: live-session detection used by + * overseer retry_step, and execute-family failure park that must not stamp + * status=failed over a peer live session. + * + * FNXC:PlannerOversight 2026-07-21-23:20: + * Surface Enumeration: overseer liveness covers session + handoff + executing + * ownership; handleGraphFailure preserve is scoped to SEPARATE session surfaces + * (coding/step/CLI/workflow) — not the dying graph's own executing/graphRouting + * markers (CodeRabbit #2393). + */ +import { describe, expect, it } from "vitest"; +import "./executor-test-helpers.js"; +import { TaskExecutor } from "../executor.js"; +import { createMockStore, resetExecutorMocks } from "./executor-test-helpers.js"; +import type { TaskDetail } from "@fusion/core"; + +const now = "2026-07-21T22:56:00.000Z"; + +function task(overrides: Partial = {}): TaskDetail { + return { + id: "FN-LIVE", + title: "Live session gate", + description: "coverage", + column: "in-progress", + dependencies: [], + steps: [{ name: "Implement", status: "in-progress" }], + currentStep: 0, + log: [], + branch: "fusion/fn-live", + baseBranch: "main", + worktree: "/tmp/fusion-fn-live", + status: null, + error: null, + paused: false, + userPaused: false, + autoMerge: true, + createdAt: now, + updatedAt: now, + ...overrides, + } as TaskDetail; +} + +type Surface = { + name: string; + install: (executor: TaskExecutor, taskId: string) => void; + clear: (executor: TaskExecutor, taskId: string) => void; +}; + +/** Surfaces that mark isTaskLiveForOverseerRetry (includes handoff/executing). */ +const OVERSEER_LIVE_SURFACES: Surface[] = [ + { + name: "coding session", + install: (e, id) => { + (e as any).activeSessions.set(id, { id: "s1" }); + }, + clear: (e, id) => { + (e as any).activeSessions.delete(id); + }, + }, + { + name: "step executor", + install: (e, id) => { + (e as any).activeStepExecutors.set(id, {}); + }, + clear: (e, id) => { + (e as any).activeStepExecutors.delete(id); + }, + }, + { + name: "CLI session", + install: (e, id) => { + (e as any).activeCliTaskSessions.set(id, {}); + }, + clear: (e, id) => { + (e as any).activeCliTaskSessions.delete(id); + }, + }, + { + name: "workflow step session", + install: (e, id) => { + (e as any).activeWorkflowStepSessions.set(id, {}); + }, + clear: (e, id) => { + (e as any).activeWorkflowStepSessions.delete(id); + }, + }, + { + name: "resumingUnpaused handoff", + install: (e, id) => { + (e as any).resumingUnpaused.add(id); + }, + clear: (e, id) => { + (e as any).resumingUnpaused.delete(id); + }, + }, + { + name: "executing ownership", + install: (e, id) => { + (e as any).executing.add(id); + }, + clear: (e, id) => { + (e as any).executing.delete(id); + }, + }, +]; + +/** + * Peer session surfaces only — handleGraphFailure preserve deliberately excludes + * the dying run's own executing/graphRouting so a pure graph fail still parks. + */ +const PEER_SESSION_SURFACES: Surface[] = OVERSEER_LIVE_SURFACES.filter((s) => + ["coding session", "step executor", "CLI session", "workflow step session"].includes(s.name), +); + +function settingsForStore() { + return { + autoMerge: true, + maxAutoMergeRetries: 3, + maxConcurrent: 2, + maxWorktrees: 4, + pollIntervalMs: 15000, + executorToolFailureRetryCount: 0, + }; +} + +describe("isTaskLiveForOverseerRetry", () => { + it("is false when no sessions or execution claims exist", () => { + resetExecutorMocks(); + const executor = new TaskExecutor(createMockStore(), "/tmp/test"); + expect(executor.isTaskLiveForOverseerRetry("FN-1")).toBe(false); + }); + + it.each(OVERSEER_LIVE_SURFACES.map((s) => [s.name, s] as const))( + "is true for %s ownership", + (_name, surface) => { + resetExecutorMocks(); + const executor = new TaskExecutor(createMockStore(), "/tmp/test"); + surface.install(executor, "FN-1"); + expect(executor.isTaskLiveForOverseerRetry("FN-1")).toBe(true); + surface.clear(executor, "FN-1"); + }, + ); +}); + +describe("handleGraphFailure execute-family live session preserve", () => { + it.each(PEER_SESSION_SURFACES.map((s) => [s.name, s] as const))( + "does not park status=failed for step-execute when peer %s is live", + async (_name, surface) => { + resetExecutorMocks(); + const store = createMockStore(); + const live = task({ column: "in-progress", status: null }); + store.getTask.mockResolvedValue(live); + store.getSettings.mockResolvedValue(settingsForStore()); + const executor = new TaskExecutor(store, "/tmp/test"); + surface.install(executor, live.id); + (executor as any).graphRouting.add(live.id); + try { + await (executor as any).handleGraphFailure(live, { + disposition: "failed", + outcome: "failure", + visitedNodeIds: ["steps#1:step-execute"], + context: { "node:steps#1:step-execute:value": "step-failed" }, + }); + } finally { + (executor as any).graphRouting.delete(live.id); + surface.clear(executor, live.id); + } + + expect(store.logEntry).toHaveBeenCalledWith( + live.id, + expect.stringContaining("while a live agent session is still executing"), + undefined, + undefined, + ); + // Match actual updateTask signature (run context is undefined on this path). + expect(store.updateTask).not.toHaveBeenCalledWith( + live.id, + expect.objectContaining({ status: "failed" }), + undefined, + ); + }, + ); + + it("still parks status=failed for merge-region failure even when a live session exists", async () => { + resetExecutorMocks(); + const store = createMockStore(); + const live = task({ column: "in-progress", status: null }); + store.getTask.mockResolvedValue(live); + store.getSettings.mockResolvedValue(settingsForStore()); + const executor = new TaskExecutor(store, "/tmp/test"); + (executor as any).activeSessions.set(live.id, { id: "peer-session" }); + (executor as any).graphRouting.add(live.id); + try { + await (executor as any).handleGraphFailure(live, { + disposition: "failed", + outcome: "failure", + visitedNodeIds: ["merge-attempt"], + context: { "node:merge-attempt:value": "merge-failed" }, + }); + } finally { + (executor as any).graphRouting.delete(live.id); + (executor as any).activeSessions.delete(live.id); + } + + expect(store.updateTask).toHaveBeenCalledWith( + live.id, + expect.objectContaining({ status: "failed", error: expect.stringContaining("merge-attempt") }), + undefined, + ); + }); +}); diff --git a/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts b/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts index 361910b4bf..6249c771ff 100644 --- a/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts +++ b/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts @@ -43,10 +43,22 @@ interface EngineOverseerInternals { } /** Extracts the real `ProjectEngine` prototype methods FN-7551 wired without running its heavy constructor/`start()`. */ -function makeEngineInternals(): EngineOverseerInternals { - const engineLike = Object.create(ProjectEngine.prototype) as unknown as EngineOverseerInternals; +function makeEngineInternals(opts?: { + isTaskLiveForOverseerRetry?: (taskId: string) => boolean; +}): EngineOverseerInternals & { runtime: { getExecutor: () => { isTaskLiveForOverseerRetry?: (id: string) => boolean } | undefined } } { + const engineLike = Object.create(ProjectEngine.prototype) as unknown as EngineOverseerInternals & { + runtime: { getExecutor: () => { isTaskLiveForOverseerRetry?: (id: string) => boolean } | undefined }; + }; engineLike.plannerObservationEmitDedup = new Map(); engineLike.plannerEscalationEmitDedup = new Set(); + (engineLike as { plannerLiveRetrySkipLogDedup: Set }).plannerLiveRetrySkipLogDedup = new Set(); + // FNXC:PlannerOversight 2026-07-21-22:56: retry_step reads runtime.getExecutor for the live-session gate. + engineLike.runtime = { + getExecutor: () => + opts?.isTaskLiveForOverseerRetry + ? { isTaskLiveForOverseerRetry: opts.isTaskLiveForOverseerRetry } + : { isTaskLiveForOverseerRetry: () => false }, + }; return engineLike; } @@ -157,6 +169,41 @@ pgDescribe("FN-7551 — overseer decision points populate the intervention timel expect(retryEntry?.attemptLimit).toBe(3); }); + /* + FNXC:PlannerOversight 2026-07-21-22:56: + FN-8471: when a live coding/step session still owns the task, retry_step must + not moveTask→todo (hard-cancel thrash) and must not burn the recovery budget. + */ + it("failed executor with a live session skips retry_step bounce and does not burn attempt budget", async () => { + const task = await seedTask("in-progress"); + const internals = makeEngineInternals({ + isTaskLiveForOverseerRetry: (id) => id === task.id, + }); + const handlers = internals.buildPlannerRecoveryHandlers(store); + const controller = new PlannerRecoveryController({ + snapshotProvider: { + getSnapshot: () => + observation({ taskId: task.id, stage: "executor", signal: "failed", sources: [] }), + }, + handlers, + }); + + const decision = await controller.tick(task); + expect(decision?.action).toBe("retry_step"); + + const refreshed = await store.getTask(task.id); + expect(refreshed?.column).toBe("in-progress"); + + const timeline = await getPlannerInterventionTimeline(store, task.id); + expect(timeline.find((e) => e.action === "retry")).toBeUndefined(); + // Exact budget invariant — two burns would still leave room under limit 3. + expect(controller.getAttemptCount(task.id, "executor")).toBe(0); + const second = await controller.tick(task); + expect(second?.action).toBe("retry_step"); + expect(second?.exhausted).not.toBe(true); + expect(controller.getAttemptCount(task.id, "executor")).toBe(0); + }); + it("failed executor WITH an error source (failed-check) dispatches request_targeted_fix and emits a request-fix entry", async () => { const task = await seedTask("in-progress"); const { controllerWithSnapshot } = wireRealEngineOverseer(store); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 1e267408c0..f325d68a4b 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -2534,6 +2534,20 @@ export class TaskExecutor { ); } + /* + FNXC:PlannerOversight 2026-07-21-22:56: + Overseer retry_step must not hard-cancel a live agent (FN-8471 thrash: status=failed from a raced graph park while step-execute still held a session, then overseer moveTask→todo aborted the live work three times). True when any in-process graph claim, coding/step/CLI session, or unpause-resume handoff still owns the task — broader than isTaskActive so step/workflow/CLI surfaces are covered. + */ + isTaskLiveForOverseerRetry(taskId: string): boolean { + // isTaskActive covers executing/graphRouting/coding session/recoveringCompleted; + // hasLiveTaskSessionSurface adds step/workflow/CLI surfaces; resumingUnpaused is the unpause handoff gap. + return ( + this.isTaskActive(taskId) + || this.hasLiveTaskSessionSurface(taskId) + || this.resumingUnpaused.has(taskId) + ); + } + /** * FNXC:ExecutorBinding 2026-06-19-00:00: * FN-6736 gives self-healing a narrow escape hatch for phantom in-memory executor bindings after the liveness gate proves the owner is dead. Never use this as a general task stopper: it refuses to detach observable live session surfaces, then clears only stale bookkeeping (`executing`, resume/recovery sets, process-wide graph routing, activeWorktrees, activeSessionRegistry paths, and executingTaskLock) so the scheduler can re-dispatch the preserved worktree. @@ -2910,6 +2924,9 @@ export class TaskExecutor { /* FNXC:ExecutorResume 2026-07-14-15:31: A terminal failed in-progress task must not be resurrected by an unrelated `task:updated` event. Planner oversight steering comments emit that event; treating it as an unpause cleared the failure and restarted the same missing-credential execution every 45 seconds. Explicit Retry/Unpause routes clear `status` before emitting their update, while startup orphan recovery has its own bounded path, so keep failed rows parked here for operator action. + + FNXC:ExecutorResume 2026-07-21-22:56: + Claim resumingUnpaused BEFORE any await so concurrent task:updated handlers cannot both pass the gate, both await getExecutionPauseLabel, and both log "Resuming execution after unpause" (FN-8471 multi-resume race). Also treat process-wide graphRouting as already-owned work. */ if (task.status === "failed") { return false; @@ -2922,44 +2939,75 @@ export class TaskExecutor { || this.activeSessions.has(task.id) || this.activeStepExecutors.has(task.id) || this.activeWorkflowStepSessions.has(task.id) + || this.graphRouting.has(task.id) ) { return false; } - const pauseLabel = await this.getExecutionPauseLabel(); - if (pauseLabel) { - executorLog.log(`Skipping unpause resume for ${task.id} — ${pauseLabel} active`); - return false; - } - - this.approvalSuspended.delete(task.id); - if (this.isTaskWorkComplete(task) && !task.mergeDetails) { - this.recoveringCompleted.add(task.id); - executorLog.log(`${task.id} unpaused with completed work and no session — recovering directly to in-review`); - void this.recoverCompletedTask(task) - .catch((err) => executorLog.error(`Failed to recover completed unpaused task ${task.id}:`, err)) - .finally(() => this.recoveringCompleted.delete(task.id)); - return true; - } - + // Synchronous single-flight claim before any await (TOCTOU fix). this.resumingUnpaused.add(task.id); - executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`); + let handoffOwnsClaim = false; try { - await this.clearResumeFailureState(task); - await this.store.updateTask(task.id, { - resumeLimboCount: 0, - resumeLimboTipSha: null, - resumeLimboStepSignature: null, - }); - await this.store.logEntry(task.id, "Resuming execution after unpause", undefined, this.getRunContextFor(task.id)); - await this.recoverApprovedStepsOnResume(task.id); - } catch (clearErr) { - executorLog.warn(`${task.id} clearResumeFailureState failed during unpause: ${clearErr instanceof Error ? clearErr.message : String(clearErr)}`); + const pauseLabel = await this.getExecutionPauseLabel(); + if (pauseLabel) { + executorLog.log(`Skipping unpause resume for ${task.id} — ${pauseLabel} active`); + return false; + } + + // Re-check after await: a concurrent graph claim may have won meanwhile. + if ( + this.executing.has(task.id) + || this.recoveringCompleted.has(task.id) + || this.activeSessions.has(task.id) + || this.activeStepExecutors.has(task.id) + || this.activeWorkflowStepSessions.has(task.id) + || this.graphRouting.has(task.id) + ) { + return false; + } + + this.approvalSuspended.delete(task.id); + if (this.isTaskWorkComplete(task) && !task.mergeDetails) { + /* + FNXC:ExecutorResume 2026-07-21-23:06: + recoverCompletedTask refuses when resumingUnpaused still holds the id. + Transfer ownership: clear the unpause claim before the recovery path runs, + then own the flight via recoveringCompleted (FN-8471 early-claim fix). + */ + this.resumingUnpaused.delete(task.id); + this.recoveringCompleted.add(task.id); + handoffOwnsClaim = true; // prevent finally from double-deleting a already-cleared claim + executorLog.log(`${task.id} unpaused with completed work and no session — recovering directly to in-review`); + void this.recoverCompletedTask(task) + .catch((err) => executorLog.error(`Failed to recover completed unpaused task ${task.id}:`, err)) + .finally(() => this.recoveringCompleted.delete(task.id)); + return true; + } + + executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`); + try { + await this.clearResumeFailureState(task); + await this.store.updateTask(task.id, { + resumeLimboCount: 0, + resumeLimboTipSha: null, + resumeLimboStepSignature: null, + }); + await this.store.logEntry(task.id, "Resuming execution after unpause", undefined, this.getRunContextFor(task.id)); + await this.recoverApprovedStepsOnResume(task.id); + } catch (clearErr) { + executorLog.warn(`${task.id} clearResumeFailureState failed during unpause: ${clearErr instanceof Error ? clearErr.message : String(clearErr)}`); + } + handoffOwnsClaim = true; + this.execute(task) + .catch((err) => executorLog.error(`Failed to resume unpaused ${task.id}:`, err)) + .finally(() => this.resumingUnpaused.delete(task.id)); + // execute().finally owns resumingUnpaused release from here. + return true; + } finally { + if (!handoffOwnsClaim) { + this.resumingUnpaused.delete(task.id); + } } - this.execute(task) - .catch((err) => executorLog.error(`Failed to resume unpaused ${task.id}:`, err)) - .finally(() => this.resumingUnpaused.delete(task.id)); - return true; } private async resumeApprovalAfterUnwindIfNeeded(taskId: string): Promise { @@ -5515,10 +5563,13 @@ export class TaskExecutor { * Returns true when the graph owned the task to a terminal disposition * (completed or failed); false when the legacy pipeline should run. */ - private async executeWorkflowGraph(task: Task): Promise { + private async executeWorkflowGraph(task: Task, opts?: { alreadyClaimed?: boolean }): Promise { // Claim synchronously before any await so concurrent execute() calls for // the same task cannot both enter graph routing (mirrors executingTaskLock). - this.graphRouting.add(task.id); + // executeCore may already have claimed before its pre-graph awaits (FN-8471). + if (!opts?.alreadyClaimed) { + this.graphRouting.add(task.id); + } let graphAbortController: AbortController | undefined; /* FNXC:GlobalConcurrencyControls 2026-07-14-18:30: @@ -10416,19 +10467,29 @@ export class TaskExecutor { /* FNXC:WorkflowRemediation 2026-07-01-23:40: Do NOT flag a still-executing task as failed. A `pre-merge-remediation` / `plan-replan` node (e.g. `code-review-remediation`) is a fire-and-forget async scheduler with no `failure` out-edge, so a failed re-arm (missing rehydrated failureContext after restart, remediation-not-scheduled, or an exhausted rework budget) bubbles out as the terminal graph outcome here. When a SEPARATE live agent session surface is still registered for this task, the previously-scheduled fix/reviewer is genuinely mid-flight — parking `status:"failed"` would surface a spurious "Task Failed" over live work. Preserve the row and let the live session drive its own terminal handoff instead. Scoped strictly to remediation nodes + a live session surface so genuine execute/merge terminal failures (and remediation failures with NO live session, e.g. a truly exhausted budget) still park exactly as before. + + FNXC:WorkflowRemediation 2026-07-21-22:56: + Extend the same preserve rule to execute-family nodes when a SEPARATE live session surface exists. A losing raced graph (duplicate resume after plan-review) can terminate at steps#N:step-execute while a peer session still owns coding work; stamping status=failed arms overseer retry_step hard-cancels (FN-8471). Merge-region failures still park — they are not execute-family. */ - if (this.hasLiveTaskSessionSurface(task.id) && await this.isRemediationGraphNode(task.id, failedNode)) { - const benignMessage = `Workflow graph ended at remediation node '${failedNode ?? "unknown"}' while a live agent session is still executing — not flagging as failed; live session preserved`; - executorLog.warn(`${task.id}: ${benignMessage}`); - await this.store.logEntry(task.id, benignMessage, undefined, this.getRunContextFor(task.id)); - await this.persistTokenUsage(task.id); - return; + const isExecuteFamilyNode = + failedNode === "execute" + || failedNode === "step-execute" + || failedNode?.endsWith(":step-execute") === true; + if (this.hasLiveTaskSessionSurface(task.id)) { + const isRemediation = await this.isRemediationGraphNode(task.id, failedNode); + if (isRemediation || isExecuteFamilyNode) { + const kind = isRemediation ? "remediation" : "execute"; + const benignMessage = `Workflow graph ended at ${kind} node '${failedNode ?? "unknown"}' while a live agent session is still executing — not flagging as failed; live session preserved`; + executorLog.warn(`${task.id}: ${benignMessage}`); + await this.store.logEntry(task.id, benignMessage, undefined, this.getRunContextFor(task.id)); + await this.persistTokenUsage(task.id); + return; + } } const message = `Workflow graph terminated with failure at node '${failedNode ?? "unknown"}'`; const settings = await this.store.getSettings(); const maxToolFailureRetries = resolveMaxConsecutiveToolFailureRetries(settings); - const isExecuteFailure = failedNode === "execute" || failedNode?.endsWith(":step-execute") === true || failedNode === "step-execute"; - if (maxToolFailureRetries > 0 && isExecuteFailure && !live.paused && !live.userPaused && !live.deletedAt && live.column === "in-progress") { + if (maxToolFailureRetries > 0 && isExecuteFamilyNode && !live.paused && !live.userPaused && !live.deletedAt && live.column === "in-progress") { // Prefer the execution-local boundary; recovery paths refetch durable state rather than use the stale failure snapshot. const cursor = this.graphToolFailureRunCursors.get(task.id) ?? (await this.store.getTask(task.id))?.toolFailureDetectorLogCursor; const threshold = resolveConsecutiveToolFailureThreshold(settings); @@ -10916,7 +10977,6 @@ export class TaskExecutor { */ private async executeCore(task: Task): Promise { this.completionFinalizedTaskIds.delete(task.id); - await this.clearStalePauseAbortBeforeDispatch(task); /* FNXC:ExecutorSoftDelete 2026-07-20-23:30: Soft-delete refuse belongs in routing, not only inside runImplementation. After U10b the @@ -10931,41 +10991,59 @@ export class TaskExecutor { dropPreHeldExecutorSlot(task.id, this.options.semaphore); return; } + /* + FNXC:WorkflowExecution 2026-07-21-22:56: + Claim graphRouting BEFORE any await. The previous check-then-await-then-claim + window let concurrent execute() calls (task:moved + unpause resume after plan-review) + both pass the graphRouting.has gate, both enter executeWorkflowGraph, and one park + status=failed while the other still owned work (FN-8471 overseer thrash). + */ if (this.graphRouting.has(task.id)) { // Duplicate dispatch while the graph runner owns this task — drop it, // mirroring the executingTaskLock duplicate-invocation behavior. executorLog.log(`execute() called for ${task.id} while graph routing is active — skipping duplicate`); return; } - if (await this.blockOuterDispatchWhenDependenciesUnmet(task)) { - // FNXC:GlobalConcurrencyControls 2026-07-14-18:30: release any scheduler pre-held slot when outer dispatch aborts before agent work starts. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); - return; - } - // FNXC:EphemeralAgents 2026-07-01-00:00: gate ALL workflow dispatch paths - // (graph/authoritative/work-engine) on ephemeralAgentsEnabled before any of - // them can claim the task, so the single check covers all three entry points. - if (await this.blockOuterDispatchWhenEphemeralDisabled(task)) { - dropPreHeldExecutorSlot(task.id, this.options.semaphore); - return; - } - /* - FNXC:WorkflowExecution 2026-07-19-10:40: - U10 (R9) — the `workflowAuthoritativeDispatch` branch is DELETED along with - WorkflowAuthoritativeDriver. It was the pre-graph "authoritative" runtime: a second - in-process execution path that could claim a task between the graph and the legacy - implementation. The graph is now the sole orchestrator, so a second claimant is not a - fallback, it is a race. + this.graphRouting.add(task.id); + let graphRunnerOwnsClaim = false; + try { + await this.clearStalePauseAbortBeforeDispatch(task); + if (await this.blockOuterDispatchWhenDependenciesUnmet(task)) { + // FNXC:GlobalConcurrencyControls 2026-07-14-18:30: release any scheduler pre-held slot when outer dispatch aborts before agent work starts. + dropPreHeldExecutorSlot(task.id, this.options.semaphore); + return; + } + // FNXC:EphemeralAgents 2026-07-01-00:00: gate ALL workflow dispatch paths + // (graph/authoritative/work-engine) on ephemeralAgentsEnabled before any of + // them can claim the task, so the single check covers all three entry points. + if (await this.blockOuterDispatchWhenEphemeralDisabled(task)) { + dropPreHeldExecutorSlot(task.id, this.options.semaphore); + return; + } + /* + FNXC:WorkflowExecution 2026-07-19-10:40: + U10 (R9) — the `workflowAuthoritativeDispatch` branch is DELETED along with + WorkflowAuthoritativeDriver. It was the pre-graph "authoritative" runtime: a second + in-process execution path that could claim a task between the graph and the legacy + implementation. The graph is now the sole orchestrator, so a second claimant is not a + fallback, it is a race. - FNXC:WorkflowExecution 2026-07-19-17:45 (U10b / R9): - The trailing `await this.runImplementation(task)` is DELETED too, and - `maybeExecuteWorkflowGraph` is now `executeWorkflowGraph` returning void. The old boolean - meant "did the graph claim this task"; with the legacy fallback gone the answer is always - yes, so a bare `runImplementation` call with NO `graphCompletion` — an implementation pass - that nothing owns the completion of — is unreachable by construction rather than by - convention. That is what makes `graphCompletion` a required parameter below. - */ - await this.executeWorkflowGraph(task); + FNXC:WorkflowExecution 2026-07-19-17:45 (U10b / R9): + The trailing `await this.runImplementation(task)` is DELETED too, and + `maybeExecuteWorkflowGraph` is now `executeWorkflowGraph` returning void. The old boolean + meant "did the graph claim this task"; with the legacy fallback gone the answer is always + yes, so a bare `runImplementation` call with NO `graphCompletion` — an implementation pass + that nothing owns the completion of — is unreachable by construction rather than by + convention. That is what makes `graphCompletion` a required parameter below. + */ + graphRunnerOwnsClaim = true; + await this.executeWorkflowGraph(task, { alreadyClaimed: true }); + } finally { + // executeWorkflowGraph's finally releases the claim when it owns the run. + if (!graphRunnerOwnsClaim) { + this.graphRouting.delete(task.id); + } + } } /* diff --git a/packages/engine/src/planner-recovery-controller.ts b/packages/engine/src/planner-recovery-controller.ts index 6beff92b8c..88f9bf6363 100644 --- a/packages/engine/src/planner-recovery-controller.ts +++ b/packages/engine/src/planner-recovery-controller.ts @@ -74,7 +74,13 @@ export interface PlannerRecoveryContext { */ export interface PlannerRecoveryHandlers { injectGuidance?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; - retryStep?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; + /** + * FNXC:PlannerOversight 2026-07-21-22:56: + * Return `false` when the bounce was intentionally skipped (e.g. live executor + * session) so `tick()` does not burn the bounded recovery attempt budget. + * Void/`true`/undefined still counts as dispatched. + */ + retryStep?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; requestTargetedFix?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; /** * FN-7513: records/surfaces a pending `PlannerConfirmationRequest` for a @@ -333,8 +339,9 @@ export class PlannerRecoveryController { } if (decision.action === "retry_step") { if (!this.handlers.retryStep) return false; - await this.handlers.retryStep(task, decision, ctx); - return true; + const result = await this.handlers.retryStep(task, decision, ctx); + // Explicit false = intentional skip (live session); do not burn budget. + return result !== false; } if (decision.action === "request_targeted_fix") { if (!this.handlers.requestTargetedFix) return false; diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index e48719dd64..4a425988c6 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -421,6 +421,12 @@ export class ProjectEngine { * subsequent polls emits exactly one `escalate` entry, not one per poll. */ private readonly plannerEscalationEmitDedup = new Set(); + /** + * FNXC:PlannerOversight 2026-07-21-22:56: + * Dedup keys for durable "retry_step skipped — live session" task-log lines so + * the 45s overseer poll does not flood FN-8471-class live-skip conditions. + */ + private readonly plannerLiveRetrySkipLogDedup = new Set(); private prReconciler?: PrReconciler; private prCommentHandler?: PrCommentHandler; private notifier?: NtfyNotifier; @@ -1642,6 +1648,7 @@ export class ProjectEngine { // dedup keys from before the stop. this.plannerObservationEmitDedup.delete(taskId); this.clearPlannerEscalationDedup(taskId); + this.clearPlannerLiveRetrySkipLogDedup(taskId); return { applied: true, reason: "stopped", task: updatedTask }; } catch (err) { @@ -1778,6 +1785,22 @@ export class ProjectEngine { } } + /** + * FNXC:PlannerOversight 2026-07-21-23:20: + * Clear live-retry skip-log dedup keys when oversight stops, is disabled, or the + * task leaves the in-flight set — same lifetime as observation/escalation dedup. + * Without this, a later live-skip episode on the same (taskId, stage) would never + * emit its durable log (Greptile/CodeRabbit on #2393). + */ + private clearPlannerLiveRetrySkipLogDedup(taskId: string): void { + const prefix = `${taskId}::`; + for (const key of [...this.plannerLiveRetrySkipLogDedup]) { + if (key.startsWith(prefix)) { + this.plannerLiveRetrySkipLogDedup.delete(key); + } + } + } + /** * FNXC:PlannerOversight 2026-07-04-12:00: * Concrete FN-7512 handler wiring — ONLY reuses existing mechanisms: @@ -1810,6 +1833,33 @@ export class ProjectEngine { ); }, retryStep: async (task, decision) => { + /* + FNXC:PlannerOversight 2026-07-21-22:56: + Never hard-cancel a live executor to "retry" incomplete work (FN-8471). + moveTask(in-progress→todo) aborts agent/graph sessions via task:moved. + When a coding/step/CLI session or graph claim is still live, skip the bounce + and return false so PlannerRecoveryController does not burn the attempt + budget. Mirror self-healing FN-7566 live-session refusal before reclaim. + Durable skip log is deduped per (taskId, stage) so 45s polls do not flood the task log. + */ + const executor = this.runtime.getExecutor?.(); + if (executor?.isTaskLiveForOverseerRetry?.(task.id) === true) { + const stage = (decision.watchedStage ?? "executor") as string; + const skipKey = `${task.id}::${stage}`; + if (!this.plannerLiveRetrySkipLogDedup.has(skipKey)) { + this.plannerLiveRetrySkipLogDedup.add(skipKey); + runtimeLog.log( + `[planner-oversight] retry_step skipped for ${task.id} — live executor/session still active (refusing hard-cancel thrash)`, + ); + await store.logEntry( + task.id, + `[planner] stage=${stage} signal=retry-skipped: live session active — not bouncing to todo`, + ).catch(() => undefined); + } + return false; + } + // Live surface cleared — allow a fresh skip log if work goes live again later. + this.plannerLiveRetrySkipLogDedup.delete(`${task.id}::${decision.watchedStage ?? "executor"}`); await store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine" } as Parameters[2]); // FN-7551: the attempt just dispatched — record it as attemptCount + 1 // (decision.attemptCount is the count BEFORE this dispatch). @@ -1824,6 +1874,7 @@ export class ProjectEngine { sourceLinks: this.toInterventionSourceLinks(decision.sourceLinks), }), ); + return true; }, requestTargetedFix: async (task, decision) => { const sourceRef = decision.sourceLinks[0]?.ref; @@ -2889,6 +2940,7 @@ export class ProjectEngine { this.sessionAdvisorLogCursor.delete(task.id); this.plannerObservationEmitDedup.delete(task.id); this.clearPlannerEscalationDedup(task.id); + this.clearPlannerLiveRetrySkipLogDedup(task.id); continue; } // FN-7743: resolve the executor-stall threshold from the task's @@ -2951,6 +3003,7 @@ export class ProjectEngine { this.sessionAdvisorLogCursor.delete(taskId); this.plannerObservationEmitDedup.delete(taskId); this.clearPlannerEscalationDedup(taskId); + this.clearPlannerLiveRetrySkipLogDedup(taskId); } } } catch {