diff --git a/.changeset/fn-7214-workflow-graph-node-pause-resume.md b/.changeset/fn-7214-workflow-graph-node-pause-resume.md new file mode 100644 index 0000000000..80c11471cb --- /dev/null +++ b/.changeset/fn-7214-workflow-graph-node-pause-resume.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Workflow graph nodes now resume cleanly after engine pause-aborts. +category: fix +dev: Distinguishes engine-internal in-flight node aborts from genuine workflow node failures, re-enters the node through a bounded graph resume path, and emits a run-audit event for the recovery. diff --git a/docs/architecture.md b/docs/architecture.md index b4ba31e0b0..6dd0c53a49 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -683,6 +683,7 @@ Runtime action-gate flow (v1): - `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`. - `recoverPausedAbortFailures()` clears executor pause/resume abort parks only when the durable row is safe to recover. `todo`/`in-progress` rows are requeued for normal scheduling, while clean `in-review` rows (completed steps, not paused/user-paused/executing, auto-merge eligible, no confirmed or terminal merge evidence) have `status`/`error` cleared in place so review progression can continue. User hard-cancel, global/user pause, `autoMerge:false`, terminal merge, and live-execution guards remain operator-actionable. Successful recovery emits `task:auto-recover-paused-abort-park` with `preservedInReview` metadata. + - Workflow graph pause/resume is node-reentrant for typed engine-internal interruptions. When `WorkflowGraphExecutor` sees the graph abort signal or a node returns `value: "aborted"`, it stamps the interrupted node and `engine-pause` abort kind into graph context. `TaskExecutor` then uses the existing bounded `graphResumeRetryCount` budget to clear the transient abort, suppress failure notification with an `Auto-recovered:` task log, and re-enter the graph/task only under the same safety guards: no user/active global pause, no merge/finalize provenance, no genuine node failure, no terminal merge value, no `autoMerge:false` protected review row, and no active execution owner. Global-pause provenance from the graph-controller abort is re-entrant once the global pause has been lifted because it represents the same in-flight node interruption. Legacy pause-abort parks without the typed node marker remain operator-action failures. - `reattach-orphaned-assigned-executions` is a forward-resume safety net for durable-agent assignments. During startup recovery and periodic maintenance, after orphaned-agent and stale-heartbeat-run repairs, self-healing finds `in-progress` tasks with an `assignedAgentId` whose agent has no active heartbeat run and no active executor session after the orphan grace window. It re-dispatches in place via `executor.resumeTaskForAgent(agentId)` (the same seam used by clean `HeartbeatMonitor.onRunCompleted` and guarded by executor double-execution checks), emits `task:reattach-orphaned-execution`, and never moves the task backward. This complements engine-start `executor.resumeOrphaned()` and leaves unassigned/role-based execution recovery to the existing startup/limbo/stuck-task paths. - Durable `Agent.taskId` is a running assignment for parked `todo`/`triage` task rows only when the agent has live proof: a fresh active heartbeat run or an executor-active/tracked heartbeat signal. Scheduler overlap requeues, task move sync, self-healing, and Reports Health Check share this invariant: stale durable links are cleared or rendered as stale while `status: "queued"` and `overlapBlockedBy` remain on the task row so file-scope lease blocking is not weakened. `fn_list_agents` and `fn_agent_show` render the linked task column next to `Current Task` (for example `Current Task: FN-1234 (triage)` or `Current Task: FN-1234 (not active — done)`) so parked-column planning ownership is not misread as in-progress execution drift. - Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`. @@ -737,6 +738,7 @@ Guardrails: this routine does **not** retry merges, does **not** apply to mixed/ ### Observability and reflection - `AgentLogger` (`agent-logger.ts`) — structured per-agent run logging - `RunAudit` (`run-audit.ts`) — mutation audit tracking (DB/git/filesystem) + - FN-7214: `task:reenter-paused-aborted-workflow-node` records executor re-entry after a typed workflow graph node was interrupted by engine pause/resume. Metadata includes `nodeId`, `fromColumn`, retry `attempt`/`maxAttempts`, `abortProvenance`, whether the task was preserved in `in-review`, and the re-entry `mode`. - FN-6782/FN-6796: `task:auto-recover-paused-abort-park` records self-healing recovery of pause-abort operator parks. Metadata includes the source column and whether recovery preserved a clean `in-review` row instead of requeueing to `todo`. - FN-7069: `task:reconcile-phantom-committed-reservation` records task-store startup or self-healing cleanup of committed-reservation-without-task phantoms. Metadata includes `reservationStatus: "committed"` plus pruned `activityLog` and `agents` counts; `runAuditEvents` and the committed reservation are intentionally retained for auditability and ID permanence. - FN-7074: `task:reservation-commit-rolled-back` records preventive create-path rollback when a distributed reservation was committed with the task-row insert but a later create materialization step failed. Metadata includes `{ reservationId, nodeId, reason: "failed-create", error }`; the task row/partial directory are removed and the reservation is moved to `aborted` so FN-7069 should not need to clean up a new phantom. @@ -1301,7 +1303,7 @@ The columns/traits track moved *board* policy (transitions, capacity, hold, merg - A `parse-steps` node reads a workflow-declared **artifact** (PROMPT.md is just the default workflow's declared `step-source` artifact) and runs a registry **parser** (`step-headings`, `json-steps`, or a plugin-contributed parser) to write `Task.steps[]`. It is the only graph-side step-list writer and must dominate any `foreach`. Parsers fail closed to a routable `outcome:parse-error`. - A `foreach(source:"task-steps")` node instantiates an inline template subgraph once per planned step, with `mode` (sequential/parallel) and `isolation` (shared/worktree) as explicit axes and per-instance run-state pinned + persisted for crash-safe resume. - Resume-limbo graph failures are retried only through a narrow persisted counter (`Task.graphResumeRetryCount`, max 2). The executor classifies a failure as transient only when it happens immediately after the engine restart/unpause resume log marker, reports no graph `reason`, has no completed step progress, and the task has no durable `lastError`/`failureReason`; it clears transient `status`/`error`, logs the auto-retry, and schedules one more graph execution. Any explicit graph reason, completed step progress, durable task error, missing resume marker, or exhausted counter remains a genuine `status:"failed"` disposition and goes to review handoff, preserving the FN-5704 anti-loop contract. -- Paused graph exits are benign only while the task is still in `in-progress`; that is the user-pause/engine-pause state where preserving the pause without requeueing is intentional. If the graph reports a pause/abort exit after the task has already advanced to another live column (for example `in-review` after an unpause/resume race), `TaskExecutor.handleGraphFailure()` surfaces the boundary as operator-actionable failure evidence (`status:"failed"`/`error` when no failure is already present, plus a task-log entry) and does **not** move, rewind, or auto-merge the task. The exceptions are completed/no-commit finalize-to-review teardown (FN-6625/FN-6644/FN-6647) and benign merge-seam pause/resume aborts (FN-6735). For completed finalize handoff, once the persisted task row proves a completed finalize handoff (non-`in-progress`, all steps done/skipped, no live pause/status/error, and the finalize-to-review log entry), a trailing graph abort resolves as an already-advanced benign graph exit even if volatile completion markers were cleared by teardown/restart and later abort provenance was re-marked from `completion-finalize` to `hard-cancel`. For merge-seam aborts, `in-review` tasks with no persisted status/error and no confirmed merge may re-enter bounded auto-merge retry only when the failed graph node is a merge/request-merge seam, the graph value is not conflict/contamination/foreign/retry-exhaustion evidence, project settings allow auto-merge processing (or the task is a shared-branch local integration member), and the merge retry budget is not exhausted. Genuine `userPaused`/global-pause exits, active-execution hard-cancels, terminal merge failure values, `autoMerge:false` human-gated tasks, retry-exhausted tasks, and already-confirmed merges still use the operator-action path. `done` and `archived` remain terminal and keep their column/status, while existing failure details are preserved. +- Paused graph exits are benign only while the task is still in `in-progress`; that is the user-pause/engine-pause state where preserving the pause without requeueing is intentional. If the graph reports a pause/abort exit after the task has already advanced to another live column (for example `in-review` after an unpause/resume race), `TaskExecutor.handleGraphFailure()` surfaces the boundary as operator-actionable failure evidence (`status:"failed"`/`error` when no failure is already present, plus a task-log entry) and does **not** move, rewind, or auto-merge the task unless the graph result carries the typed interrupted-node marker. The exceptions are typed in-flight node pause aborts (FN-7214), completed/no-commit finalize-to-review teardown (FN-6625/FN-6644/FN-6647), and benign merge-seam pause/resume aborts (FN-6735). For FN-7214 node aborts, `hard-cancel` and lifted `global-pause` provenance can re-enter the interrupted node through the bounded `graphResumeRetryCount` path; explicit `userPaused`, active global pause, merge/finalize provenance, genuine node failures, `autoMerge:false` human-gated review rows, retry-exhausted tasks, and already-confirmed merges still use the protected operator-action path. For completed finalize handoff, once the persisted task row proves a completed finalize handoff (non-`in-progress`, all steps done/skipped, no live pause/status/error, and the finalize-to-review log entry), a trailing graph abort resolves as an already-advanced benign graph exit even if volatile completion markers were cleared by teardown/restart and later abort provenance was re-marked from `completion-finalize` to `hard-cancel`. For merge-seam aborts, `in-review` tasks with no persisted status/error and no confirmed merge may re-enter bounded auto-merge retry only when the failed graph node is a merge/request-merge seam, the graph value is not conflict/contamination/foreign/retry-exhaustion evidence, project settings allow auto-merge processing (or the task is a shared-branch local integration member), and the merge retry budget is not exhausted. `done` and `archived` remain terminal and keep their column/status, while existing failure details are preserved. - A `step-review` node surfaces reviewer verdicts (APPROVE/REVISE/RETHINK/UNAVAILABLE) as outcome edges; `rework` edges (the only legal graph cycles, bounded per instance) route REVISE/RETHINK back to `step-execute`, with RETHINK traversal triggering the reset seam. - A `code` node runs sandboxed TypeScript (esbuild + child process, clamped timeout, no store handle) for arbitrary computed routing/field logic — the same trust tier as project-local script steps. diff --git a/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts b/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts index a8237d0755..34c0f5f585 100644 --- a/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts +++ b/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts @@ -47,17 +47,23 @@ function makeHarness( autoMerge: true, maxAutoMergeRetries: 3, }); + store.recordRunAuditEvent = vi.fn(); const executor = new TaskExecutor(store, "/tmp/test", {}); (executor as any).markPausedAborted(task.id, provenance); return { store, task, executor }; } -async function invokeGraphFailure(executor: TaskExecutor, task: TaskDetail) { +async function invokeGraphFailure( + executor: TaskExecutor, + task: TaskDetail, + resultOverrides: Record = {}, +) { await (executor as any).handleGraphFailure(task, { disposition: "failed", outcome: "failure", visitedNodeIds: ["plan", "execute"], context: {}, + ...resultOverrides, }); } @@ -263,7 +269,7 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => { expect(clearedFailure).toBe(true); }); - it("STILL parks a non-todo (in-review) pause-abort as operator-action failed", async () => { + it("STILL parks a non-todo (in-review) pause-abort without a typed interrupted-node marker", async () => { const { store, task, executor } = makeHarness({ column: "in-review" }); await invokeGraphFailure(executor, task); @@ -274,4 +280,225 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => { expect(parkedFailed).toBe(true); expect(logText(store)).toContain("operator action required"); }); + + it("auto-recovers an in-review paused-aborted in-flight workflow node without operator-action parking", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { + "node:plan:value": "aborted", + "node:plan:abortKind": "engine-pause", + "workflow:interruptedNodeId": "plan", + "workflow:interruptedNodeAbortKind": "engine-pause", + }, + }); + + const parkedFailed = store.updateTask.mock.calls.some( + (call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed", + ); + expect(parkedFailed).toBe(false); + expect(logText(store)).toContain("Auto-recovered: re-entering paused-aborted workflow graph node 'plan'"); + expect(logText(store)).not.toContain("operator action required"); + const bumpedRetry = store.updateTask.mock.calls.some( + (call: unknown[]) => { + const patch = call[1] as { graphResumeRetryCount?: number; status?: unknown; error?: unknown } | undefined; + return patch?.graphResumeRetryCount === 1 && patch?.status === null && patch?.error === null; + }, + ); + expect(bumpedRetry).toBe(true); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reenter-paused-aborted-workflow-node", + metadata: expect.objectContaining({ + nodeId: "plan", + fromColumn: "in-review", + attempt: 1, + maxAttempts: 2, + abortProvenance: "hard-cancel", + preservedInReview: true, + mode: "preserved-in-review", + }), + })); + await flushScheduledRetry(); + expect(graphSpy).toHaveBeenCalledTimes(1); + }); + + it("fire-time guard skips in-review graph re-entry when a graph run is already active", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" }, + }); + + (executor as any).activeWorkflowGraphAbortControllers.set(task.id, new AbortController()); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); + + it("does NOT auto-recover an explicit user pause even with an interrupted-node marker", async () => { + const { store, task, executor } = makeHarness({ column: "in-review", userPaused: true }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("operator action required"); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); + + it("auto-recovers an in-review paused-aborted execute node through graph re-entry", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "execute", + interruptedAbortKind: "engine-pause", + context: { "node:execute:value": "aborted", "node:execute:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("Auto-recovered: re-entering paused-aborted workflow graph node 'execute'"); + expect(logText(store)).not.toContain("operator action required"); + await flushScheduledRetry(); + expect(graphSpy).toHaveBeenCalledTimes(1); + }); + + it("auto-recovers a todo paused-aborted execute node by re-executing the task", async () => { + const { store, task, executor } = makeHarness({ column: "todo" }); + const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "execute", + interruptedAbortKind: "engine-pause", + context: { "node:execute:value": "aborted", "node:execute:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("Auto-recovered: re-entering paused-aborted workflow graph node 'execute'"); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reenter-paused-aborted-workflow-node", + metadata: expect.objectContaining({ nodeId: "execute", fromColumn: "todo", mode: "reexecuted-from-todo" }), + })); + await flushScheduledRetry(); + expect(executeSpy).toHaveBeenCalledTimes(1); + }); + + it("STILL parks a genuine in-review node failure with no paused-node audit", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + visitedNodeIds: ["plan"], + context: { "node:plan:value": "REVISE" }, + }); + + expect(logText(store)).toContain("operator action required"); + expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reenter-paused-aborted-workflow-node", + })); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); + + it("auto-recovers a global-pause in-review interrupted node after global resume", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }, "global-pause"); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("Auto-recovered: re-entering paused-aborted workflow graph node 'plan'"); + expect(logText(store)).not.toContain("operator action required"); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reenter-paused-aborted-workflow-node", + metadata: expect.objectContaining({ + nodeId: "plan", + fromColumn: "in-review", + abortProvenance: "global-pause", + mode: "preserved-in-review", + }), + })); + await flushScheduledRetry(); + expect(graphSpy).toHaveBeenCalledTimes(1); + }); + + it("does NOT auto-recover a global-pause in-review failure without an interrupted-node marker", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }, "global-pause"); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + visitedNodeIds: ["plan"], + context: { "node:plan:value": "aborted" }, + }); + + expect(logText(store)).toContain("operator action required"); + expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reenter-paused-aborted-workflow-node", + })); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); + + it("does NOT auto-recover a global-pause interrupted node while global pause is still active", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }, "global-pause"); + store.getSettings.mockResolvedValue({ + maxConcurrent: 2, + maxWorktrees: 4, + pollIntervalMs: 15000, + autoMerge: true, + globalPause: true, + maxAutoMergeRetries: 3, + }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("operator action required"); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); + + it("does NOT auto-recover an autoMerge:false in-review interrupted node", async () => { + const { store, task, executor } = makeHarness({ column: "in-review", autoMerge: false }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("operator action required"); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); + + it("does NOT auto-recover an exhausted in-review interrupted node", async () => { + const { store, task, executor } = makeHarness({ column: "in-review", graphResumeRetryCount: 2 }); + const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true); + + await invokeGraphFailure(executor, task, { + interruptedNodeId: "plan", + interruptedAbortKind: "engine-pause", + context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" }, + }); + + expect(logText(store)).toContain("operator action required"); + await flushScheduledRetry(); + expect(graphSpy).not.toHaveBeenCalled(); + }); }); diff --git a/packages/engine/src/__tests__/workflow-graph-paused-node-resume.test.ts b/packages/engine/src/__tests__/workflow-graph-paused-node-resume.test.ts new file mode 100644 index 0000000000..fc9c23b889 --- /dev/null +++ b/packages/engine/src/__tests__/workflow-graph-paused-node-resume.test.ts @@ -0,0 +1,148 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Settings, TaskDetail, WorkflowDefinition, WorkflowIr } from "@fusion/core"; + +import { + WORKFLOW_INTERRUPTED_NODE_ABORT_KIND_CONTEXT_KEY, + WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY, + WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND, + WorkflowGraphExecutor, + type WorkflowNodeResult, +} from "../workflow-graph-executor.js"; +import { WorkflowGraphTaskRunner, type WorkflowGraphRunnerStore } from "../workflow-graph-task-runner.js"; + +const now = "2026-06-28T18:15:00.000Z"; +const flagOn = { experimentalFeatures: { workflowGraphExecutor: true } } as unknown as Pick; + +function makeTask(overrides: Partial = {}): TaskDetail { + return { + id: "FN-7214-T", + title: "paused graph node resume", + description: "Reproduces node-level pause abort re-entry", + column: "in-review", + dependencies: [], + steps: [{ name: "Implement", status: "pending" }], + currentStep: 0, + log: [], + branch: null, + baseBranch: "main", + worktree: "/tmp/fusion-fn-7214-t", + status: null, + error: null, + paused: false, + userPaused: false, + autoMerge: true, + mergeRetries: 0, + createdAt: now, + updatedAt: now, + ...overrides, + } as TaskDetail; +} + +function planExecuteIr(): WorkflowIr { + return { + version: "v1", + name: "paused-node", + nodes: [ + { id: "start", kind: "start" }, + { id: "plan", kind: "prompt", config: { seam: "planning" } }, + { id: "execute", kind: "prompt", config: { seam: "execute" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "plan" }, + { from: "plan", to: "execute", condition: "success" }, + { from: "execute", to: "end", condition: "success" }, + ], + }; +} + +function definition(ir: WorkflowIr = planExecuteIr()): WorkflowDefinition { + return { + id: "WF-7214", + name: "Paused node workflow", + description: "", + kind: "workflow", + ir, + layout: {}, + createdAt: now, + updatedAt: now, + }; +} + +function storeWith(def: WorkflowDefinition | undefined): WorkflowGraphRunnerStore { + return { + getTaskWorkflowSelection: () => (def ? { workflowId: def.id, stepIds: [] } : undefined), + getWorkflowDefinition: async () => def, + }; +} + +function seamResult(result: WorkflowNodeResult) { + return async () => result; +} + +describe("workflow graph paused node resume contract (FN-7214)", () => { + it("stamps the interrupted node when a top-level graph signal aborts an in-flight node", async () => { + const controller = new AbortController(); + const executor = new WorkflowGraphExecutor({ + signal: controller.signal, + handlers: { + prompt: async () => { + controller.abort(); + return { outcome: "failure", value: "aborted" }; + }, + }, + }); + + const result = await executor.run(makeTask(), flagOn, planExecuteIr()); + + expect(result.outcome).toBe("failure"); + expect(result.visitedNodeIds).toEqual(["start", "plan"]); + expect(result.context["node:plan:value"]).toBe("aborted"); + expect(result.context["node:plan:abortKind"]).toBe(WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND); + expect(result.context[WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY]).toBe("plan"); + expect(result.context[WORKFLOW_INTERRUPTED_NODE_ABORT_KIND_CONTEXT_KEY]).toBe(WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND); + }); + + it("returns interrupted node metadata from the task runner for aborted planning seams", async () => { + const runner = new WorkflowGraphTaskRunner({ + store: storeWith(definition()), + seams: { + planning: seamResult({ outcome: "failure", value: "aborted" }), + execute: seamResult({ outcome: "success" }), + workflowStep: seamResult({ outcome: "success" }), + review: seamResult({ outcome: "success" }), + merge: seamResult({ outcome: "success" }), + schedule: seamResult({ outcome: "success" }), + }, + runCustomNode: vi.fn(async () => ({ outcome: "success" as const })), + }); + + const result = await runner.run(makeTask(), flagOn); + + expect(result.disposition).toBe("failed"); + expect(result.interruptedNodeId).toBe("plan"); + expect(result.interruptedAbortKind).toBe(WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND); + }); + + it("does not mark genuine node failures as paused-aborted interruptions", async () => { + const runner = new WorkflowGraphTaskRunner({ + store: storeWith(definition()), + seams: { + planning: seamResult({ outcome: "failure", value: "REVISE" }), + execute: seamResult({ outcome: "success" }), + workflowStep: seamResult({ outcome: "success" }), + review: seamResult({ outcome: "success" }), + merge: seamResult({ outcome: "success" }), + schedule: seamResult({ outcome: "success" }), + }, + runCustomNode: vi.fn(async () => ({ outcome: "success" as const })), + }); + + const result = await runner.run(makeTask(), flagOn); + + expect(result.disposition).toBe("failed"); + expect(result.interruptedNodeId).toBeUndefined(); + expect(result.interruptedAbortKind).toBeUndefined(); + expect(result.context?.[WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY]).toBeUndefined(); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 22878ae075..7159b58cc5 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -37,7 +37,7 @@ import { type ForeachActiveContext, type WorkflowLegacySeams, } from "./workflow-node-handlers.js"; -import { MERGE_REGION_KINDS } from "./workflow-graph-executor.js"; +import { MERGE_REGION_KINDS, WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND } from "./workflow-graph-executor.js"; import type { WorkflowNodeResult } from "./workflow-graph-executor.js"; import type { AuditPrimitiveInput, @@ -1575,6 +1575,8 @@ export class TaskExecutor { private activeWorkflowStepSessionSeenSteeringIds = new Map>(); /** Active configured-command abort controllers keyed by task. */ private activeConfiguredCommandControllers = new Map>(); + /** Active workflow-graph runner abort controllers keyed by task. */ + private activeWorkflowGraphAbortControllers = new Map(); /** * Active CLI agent task sessions per task (U7). Mirrors activeSessions for the * cli-agent executor kind so the hard-cancel / abort path can SIGKILL the PTY @@ -2304,6 +2306,12 @@ export class TaskExecutor { controller.abort(); } } + const claimedWorkflowGraphController = this.activeWorkflowGraphAbortControllers.get(taskId); + if (claimedWorkflowGraphController) { + hadActiveSurface = true; + this.activeWorkflowGraphAbortControllers.delete(taskId); + claimedWorkflowGraphController.abort(); + } const claimedSubagents = this.activeSubagentSessions.has(taskId); if (claimedSubagents) { hadActiveSurface = true; @@ -2383,6 +2391,7 @@ export class TaskExecutor { ...this.activeStepExecutors.keys(), ...this.activeWorkflowStepSessions.keys(), ...this.activeConfiguredCommandControllers.keys(), + ...this.activeWorkflowGraphAbortControllers.keys(), ...this.activeSubagentSessions.keys(), ...this.activeCliTaskSessions.keys(), ]); @@ -3017,6 +3026,16 @@ export class TaskExecutor { this.spawnedAgents.delete(taskId); this.stuckAborted.delete(taskId); } + for (const [taskId, controller] of this.activeWorkflowGraphAbortControllers) { + executorLog.log(`Global pause — aborting workflow graph runner for ${taskId}`); + this.markPausedAborted(taskId, "global-pause"); + this.options.stuckTaskDetector?.untrackTask(taskId); + controller.abort(); + this.activeWorkflowGraphAbortControllers.delete(taskId); + this.loopRecoveryState.delete(taskId); + this.spawnedAgents.delete(taskId); + this.stuckAborted.delete(taskId); + } } }); @@ -4313,6 +4332,7 @@ export class TaskExecutor { // 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); + let graphAbortController: AbortController | undefined; try { let settings: Settings; try { @@ -4431,6 +4451,8 @@ export class TaskExecutor { // the finally below clears it. this.graphUnattendedRuns.delete(task.id); + graphAbortController = new AbortController(); + this.activeWorkflowGraphAbortControllers.set(task.id, graphAbortController); const runner = new WorkflowGraphTaskRunner({ store: { ...this.store, @@ -4460,6 +4482,7 @@ export class TaskExecutor { }); }, onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`), + signal: graphAbortController.signal, // Wire SQLite-backed per-branch persistence in production (#1407): the // executor writes each branch's currentNodeId/status to // workflow_run_branches so fan-out crash-resume and the U9 badges have @@ -4583,6 +4606,9 @@ export class TaskExecutor { } catch (err) { executorLog.warn(`terminateAllChildren failed for graph task ${task.id}: ${err instanceof Error ? err.message : String(err)}`); } + if (graphAbortController && this.activeWorkflowGraphAbortControllers.get(task.id) === graphAbortController) { + this.activeWorkflowGraphAbortControllers.delete(task.id); + } this.graphRouting.delete(task.id); // Clear per-run step-inversion pins (KTD-8: pinned only for the run's life). this.graphStepSessionPinned.delete(task.id); @@ -6897,6 +6923,131 @@ export class TaskExecutor { return true; } + private async isReentrantPausedAbortedInFlightNode( + live: TaskDetail, + result: WorkflowGraphTaskRunResult, + abortProvenance: "global-pause" | "merge-seam" | "hard-cancel" | "completion-finalize" | undefined, + pausedAborted: boolean, + userCanceled: boolean, + ): Promise { + /* + FNXC:WorkflowLifecycle 2026-06-28-18:32: + FN-7214 makes engine-internal pause aborts re-entrant only when the workflow graph reports a typed in-flight node interruption. User pauses, active global pauses, merge/finalize aborts, genuine node failures, autoMerge:false review rows, and exhausted retry budgets must continue through the existing protected failure paths. + + FNXC:WorkflowLifecycle 2026-06-28-21:39: + A global engine pause aborts active workflow graph controllers with `global-pause` provenance; after the global pause is lifted, the typed interrupted-node marker is sufficient to re-enter that node. Only active global-pause settings and explicit task/user pauses remain terminal so resume never runs behind an operator-controlled pause. + */ + if (!pausedAborted) return false; + if (abortProvenance !== "hard-cancel" && abortProvenance !== "global-pause") return false; + if (userCanceled) return false; + if (live.paused || live.userPaused === true) return false; + if (live.status != null || live.error != null) return false; + if (live.column === "done" || live.column === "archived") return false; + if (result.interruptedAbortKind !== WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND) return false; + if (!result.interruptedNodeId) return false; + if (this.isMergeGraphFailure(result.interruptedNodeId)) return false; + if (this.isTerminalMergeGraphFailureValue(this.graphFailureValue(result))) return false; + if ((live.graphResumeRetryCount ?? 0) >= MAX_TRANSIENT_GRAPH_RESUME_RETRIES) return false; + let settings: Settings | undefined; + if (abortProvenance === "global-pause" || live.column === "in-review") { + try { + settings = await this.store.getSettings(); + } catch { + return false; + } + if (settings.globalPause === true) return false; + } + if (live.column === "in-review") { + if (live.autoMerge === false) return false; + if (!settings) return false; + const sharedBranchMember = isSharedBranchGroupMemberIntegration(live); + if (!sharedBranchMember && !allowsAutoMergeProcessing(live, settings)) return false; + if (live.mergeDetails?.mergeConfirmed === true) return false; + } + return live.column === "todo" || live.column === "in-review" || live.column === "in-progress"; + } + + private async reenterPausedAbortedWorkflowNode( + live: TaskDetail, + result: WorkflowGraphTaskRunResult, + abortProvenance: "global-pause" | "merge-seam" | "hard-cancel" | "completion-finalize" | undefined, + ): Promise { + const nodeId = result.interruptedNodeId ?? result.visitedNodeIds[result.visitedNodeIds.length - 1] ?? "unknown"; + const priorRetries = live.graphResumeRetryCount ?? 0; + if (priorRetries >= MAX_TRANSIENT_GRAPH_RESUME_RETRIES) return false; + const nextRetries = priorRetries + 1; + const preservedInReview = live.column === "in-review"; + this.clearPausedAborted(live.id); + this.activeWorktrees.delete(live.id); + const message = `Workflow graph node '${nodeId}' was interrupted by engine pause/resume — re-entering workflow graph (${nextRetries}/${MAX_TRANSIENT_GRAPH_RESUME_RETRIES})`; + executorLog.log(`${live.id}: ${message}`); + await this.store.logEntry(live.id, message, undefined, this.getRunContextFor(live.id)); + await this.store.logEntry(live.id, `Auto-recovered: re-entering paused-aborted workflow graph node '${nodeId}' — failure notification suppressed`, undefined, this.getRunContextFor(live.id)); + await this.store.updateTask(live.id, { graphResumeRetryCount: nextRetries, status: null, error: null }, this.getRunContextFor(live.id)); + try { + await this.store.recordRunAuditEvent?.({ + taskId: live.id, + agentId: "executor", + runId: generateSyntheticRunId("workflow-node-reentry", live.id), + domain: "database", + mutationType: "task:reenter-paused-aborted-workflow-node", + target: live.id, + metadata: { + nodeId, + fromColumn: live.column, + attempt: nextRetries, + maxAttempts: MAX_TRANSIENT_GRAPH_RESUME_RETRIES, + abortProvenance: abortProvenance ?? "unknown", + preservedInReview, + mode: preservedInReview ? "preserved-in-review" : live.column === "todo" ? "reexecuted-from-todo" : "reentered-graph", + }, + }); + } catch (error) { + executorLog.warn(`${live.id}: failed to record paused-node graph re-entry audit: ${error instanceof Error ? error.message : String(error)}`); + } + await this.persistTokenUsage(live.id); + + const scheduleRetry = () => { + void (async () => { + try { + const resumeTask = await this.store.getTask(live.id); + if ( + resumeTask.deletedAt + || resumeTask.paused + || resumeTask.userPaused + || resumeTask.status != null + || resumeTask.error != null + || (preservedInReview ? resumeTask.column !== "in-review" : resumeTask.column !== "todo" && resumeTask.column !== "in-progress") + || this.activeSessions.has(live.id) + || this.activeStepExecutors.has(live.id) + || this.activeWorkflowStepSessions.has(live.id) + || this.activeWorkflowGraphAbortControllers.has(live.id) + || TaskExecutor.processWideGraphRouting.has(live.id) + ) { + executorLog.log(`${live.id}: skipping paused-node graph re-entry — task is no longer in a safe resume state`); + return; + } + if (preservedInReview) { + await this.maybeExecuteWorkflowGraph(resumeTask); + } else if (resumeTask.column === "todo") { + await this.execute(resumeTask); + } else { + await this.maybeExecuteWorkflowGraph(resumeTask); + } + } catch (err) { + executorLog.error(`Failed paused-node graph re-entry for ${live.id}:`, err); + } + })(); + }; + if (TRANSIENT_GRAPH_RESUME_RETRY_BACKOFF_MS > 0) { + const handle = setTimeout(scheduleRetry, TRANSIENT_GRAPH_RESUME_RETRY_BACKOFF_MS); + handle.unref?.(); + } else { + setTimeout(scheduleRetry, 0).unref?.(); + } + return true; + } + private async routeGraphMergeFailureToRetry( live: TaskDetail, result: WorkflowGraphTaskRunResult, @@ -6995,6 +7146,11 @@ export class TaskExecutor { || (live.paused && !mergeSeamAborted && !suppressFinalizedCompletionAbort) || (pausedAborted && !mergeSeamAborted && !completionFinalizeAborted && !suppressFinalizedCompletionAbort), ); + if (genuinePauseAbort && await this.isReentrantPausedAbortedInFlightNode(live, result, abortProvenance, pausedAborted, this.userCanceledTaskIds.has(task.id))) { + if (await this.reenterPausedAbortedWorkflowNode(live, result, abortProvenance)) { + return; + } + } if (genuinePauseAbort && await this.isRetryableBenignMergePauseAbort(live, result, abortProvenance, pausedAborted)) { if (await this.routeGraphMergeFailureToRetry(live, result, abortProvenance)) { return; diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 88062b0595..6662c4692e 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -43,6 +43,12 @@ import { runLoop, runOptionalGroup } from "./workflow-graph-loop.js"; export type WorkflowNodeOutcome = "success" | "failure"; +export type WorkflowNodeAbortKind = "engine-pause"; + +export const WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY = "workflow:interruptedNodeId"; +export const WORKFLOW_INTERRUPTED_NODE_ABORT_KIND_CONTEXT_KEY = "workflow:interruptedNodeAbortKind"; +export const WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND: WorkflowNodeAbortKind = "engine-pause"; + export interface WorkflowNodeResult { outcome: WorkflowNodeOutcome; value?: string; @@ -685,7 +691,7 @@ export class WorkflowGraphExecutor { return await traverseChildren(node, result); } - const result = await this.executeNodeWithRetries(node, task, settings, context, ir); + const result = await this.executeNodeWithRetries(node, task, settings, context, ir, this.deps.signal); if (result.contextPatch) Object.assign(context, result.contextPatch); context[`node:${node.id}:outcome`] = result.outcome; if (result.value !== undefined) context[`node:${node.id}:value`] = result.value; @@ -749,6 +755,7 @@ export class WorkflowGraphExecutor { settings, context, ir, + this.deps.signal, ); if (result.contextPatch) Object.assign(context, result.contextPatch); context[`node:${syntheticMergeNode.id}:outcome`] = result.outcome; @@ -996,23 +1003,34 @@ export class WorkflowGraphExecutor { let lastError: unknown; for (let attempt = 0; attempt < maxAttempts; attempt++) { - // Fail-fast cancellation: a branch aborted mid-retry stops re-trying. - if (signal?.aborted) return { outcome: "failure", value: "aborted" }; + // Fail-fast cancellation: a branch or top-level graph abort mid-retry stops re-trying. + if (signal?.aborted) return this.withEnginePauseAbortContext(node, { outcome: "failure", value: "aborted" }); try { const pluginResult = await this.executePluginNodeHandler(node, task, workflow, context, signal); if (pluginResult) { - return await this.publishTaskProjectionFromResult(task.id, node, pluginResult); + const projected = await this.publishTaskProjectionFromResult(task.id, node, pluginResult); + return signal?.aborted || this.isAbortNodeResult(projected) + ? this.withEnginePauseAbortContext(node, projected) + : projected; } if (!handler) { throw new WorkflowIrError(`No handler registered for node kind: ${node.kind}`); } const result = await handler(node, { task, settings, context, signal }); - return await this.publishTaskProjectionFromResult(task.id, node, result); + const projected = await this.publishTaskProjectionFromResult(task.id, node, result); + return signal?.aborted || this.isAbortNodeResult(projected) + ? this.withEnginePauseAbortContext(node, projected) + : projected; } catch (error) { + if (signal?.aborted) return this.withEnginePauseAbortContext(node, { outcome: "failure", value: "aborted" }); lastError = error; } } + if (signal?.aborted) { + return this.withEnginePauseAbortContext(node, { outcome: "failure", value: "aborted" }); + } + return { outcome: "failure", value: "exception", @@ -1022,6 +1040,28 @@ export class WorkflowGraphExecutor { }; } + private isAbortNodeResult(result: WorkflowNodeResult): boolean { + return result.outcome === "failure" && result.value === "aborted"; + } + + private withEnginePauseAbortContext(node: WorkflowIrNode, result: WorkflowNodeResult): WorkflowNodeResult { + /* + FNXC:WorkflowLifecycle 2026-06-28-18:15: + FN-7214 requires engine-pause aborts of in-flight workflow nodes to be re-entrant at the node boundary. Stamp a typed abort marker on the node result so executor recovery can re-run the graph without conflating this interruption with genuine node failures such as REVISE, projection errors, or exceptions. + */ + return { + ...result, + outcome: "failure", + value: result.value ?? "aborted", + contextPatch: { + ...(result.contextPatch ?? {}), + [`node:${node.id}:abortKind`]: WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND, + [WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY]: node.id, + [WORKFLOW_INTERRUPTED_NODE_ABORT_KIND_CONTEXT_KEY]: WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND, + }, + }; + } + private async publishTaskProjectionFromResult( taskId: string, node: WorkflowIrNode, diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index 166e8a92ad..691c705c50 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -7,7 +7,16 @@ import { WorkflowCompileError, } from "@fusion/core"; -import { WorkflowGraphExecutor, type WorkflowGraphExecutorDeps, type WorkflowNodeOutcome, type WorkflowTaskProjection } from "./workflow-graph-executor.js"; +import { + WORKFLOW_INTERRUPTED_NODE_ABORT_KIND_CONTEXT_KEY, + WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY, + WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND, + WorkflowGraphExecutor, + type WorkflowGraphExecutorDeps, + type WorkflowNodeAbortKind, + type WorkflowNodeOutcome, + type WorkflowTaskProjection, +} from "./workflow-graph-executor.js"; import type { CodeNodeRunner, ForeachActiveContext, @@ -47,6 +56,10 @@ export interface WorkflowGraphTaskRunResult { reason?: string; /** Shared graph context after the run (node outcomes/values). */ context?: Record; + /** Node that was executing when an engine pause/abort interrupted the graph. */ + interruptedNodeId?: string; + /** Typed abort provenance for the interrupted node; absent for genuine node failures. */ + interruptedAbortKind?: WorkflowNodeAbortKind; } /** The minimal store surface the runner needs — keeps tests fake-friendly. */ @@ -63,6 +76,8 @@ export interface WorkflowGraphTaskRunnerDeps { maxRetriesPerNode?: number; /** Optional diagnostics hook (audit/log emission). Never throws into the run. */ onEvent?: (event: { type: "start" | "terminal" | "fallback"; taskId: string; detail: string }) => void; + /** Top-level graph abort signal used to classify engine pause/abort interruptions at node boundaries. */ + signal?: AbortSignal; /** Per-branch run-state persistence + resume (U13). Additive; in-memory without it. */ branchPersistence?: WorkflowBranchPersistence; /** Bounds concurrent branch-node execution (U13); omit when the semaphore is @@ -270,6 +285,7 @@ export class WorkflowGraphTaskRunner { recordWorkflowStepResult: this.deps.recordWorkflowStepResult, requestPreMergeOptionalStepFix: this.deps.requestPreMergeOptionalStepFix, publishTaskProjection: this.deps.publishTaskProjection, + signal: this.deps.signal, publishTouchedFiles: this.deps.publishTouchedFiles, // Single source of truth (KTD-6): prefer the caller-threaded run id so the // executor's persistence deps probe/flip rows under the SAME id; fall back @@ -290,11 +306,19 @@ export class WorkflowGraphTaskRunner { } const disposition: WorkflowGraphRunDisposition = result.outcome === "success" ? "completed" : "failed"; this.emit("terminal", task.id, `${definition.id}:${disposition}`); + const interruptedNodeId = typeof result.context[WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY] === "string" + ? result.context[WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY] + : undefined; + const interruptedAbortKind = result.context[WORKFLOW_INTERRUPTED_NODE_ABORT_KIND_CONTEXT_KEY] === WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND + ? WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND + : undefined; return { disposition, outcome: result.outcome, visitedNodeIds: result.visitedNodeIds, context: result.context, + interruptedNodeId, + interruptedAbortKind, }; } catch (err) { const reason = `interpreter-error: ${err instanceof Error ? err.message : String(err)}`;