diff --git a/.changeset/fn-7232-workflow-merge-boundary.md b/.changeset/fn-7232-workflow-merge-boundary.md new file mode 100644 index 0000000000..4b03c75fac --- /dev/null +++ b/.changeset/fn-7232-workflow-merge-boundary.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep workflow merge nodes moving even when a workflow skips a review handoff. +category: fix +dev: Workflow merge primitives now establish the in-review merge boundary before requesting merge; non-gate skill output no longer requires a verdict. diff --git a/packages/engine/src/__tests__/ce-workflow-step-executor.test.ts b/packages/engine/src/__tests__/ce-workflow-step-executor.test.ts index 85ff33b7e3..69a5c1f8f4 100644 --- a/packages/engine/src/__tests__/ce-workflow-step-executor.test.ts +++ b/packages/engine/src/__tests__/ce-workflow-step-executor.test.ts @@ -355,6 +355,104 @@ describe("CE workflow-step executor integration", () => { expect(store.updateTask).not.toHaveBeenCalledWith("FN-CE-1", expect.objectContaining({ column: "in-progress" })); }); + it("moves direct-to-merge workflow tasks into in-review before requesting merge", async () => { + const store = createMockStore(); + let live = baseStepTask({ + column: "in-progress", + steps: [{ name: "Implement", status: "done" }], + }); + store.getTask.mockImplementation(async () => live as any); + store.moveTask.mockImplementation(async (_id: string, column: string) => { + live = { ...live, column }; + return live as any; + }); + const { executor } = makeExecutor(store); + const mergeRequester = vi.fn(async () => ({ + task: live, + branch: "fusion/fn-ce-1", + merged: false, + noOp: false, + reason: "queued", + })); + executor.setMergeRequester(mergeRequester as any); + const settings = await store.getSettings(); + const primitives = (executor as any).createAuthoritativeWorkflowPrimitives(settings); + + const result = await primitives.requestMerge( + { + run: { runId: "run-merge", taskId: "FN-CE-1", workflowId: "builtin:quick-fix" }, + node: { node: { id: "merge", kind: "prompt", column: "in-review", config: { seam: "merge" } }, context: {} }, + }, + live, + ); + + expect(result).toEqual({ + outcome: "failure", + value: "queued", + data: { status: "failed", reason: "queued" }, + }); + expect(store.moveTask).toHaveBeenCalledWith( + "FN-CE-1", + "in-review", + expect.objectContaining({ + preserveProgress: true, + moveSource: "engine", + workflowMoveSource: "workflow-graph", + workflowMoveMetadata: expect.objectContaining({ + reason: "workflow-merge-boundary", + nodeId: "merge", + workflowId: "builtin:quick-fix", + runId: "run-merge", + }), + }), + ); + expect(mergeRequester).toHaveBeenCalledWith("FN-CE-1", expect.objectContaining({ signal: expect.any(AbortSignal) })); + expect(live.column).toBe("in-review"); + }); + + it("clears stale workflow input markers when a resumed graph restarts before the original node", async () => { + const store = createMockStore(); + let live = baseStepTask({ + paused: false, + status: null, + pausedReason: "workflow-input:commit-pr@1782751605619: Should I rewrite the PR?", + steeringComments: [{ text: "Yes", createdAt: "2026-06-29T16:47:05.075Z" }], + }); + store.getTask.mockImplementation(async () => live as any); + store.updateTask.mockImplementation(async (_id: string, patch: Record) => { + live = { ...live, ...patch }; + return live as any; + }); + const { executor } = makeExecutor(store); + vi.spyOn(executor as any, "executeWorkflowStep").mockResolvedValue({ success: true, output: "ok" }); + + const result = await (executor as any).runGraphCustomNode( + { + id: "plan", + kind: "prompt", + column: "in-progress", + config: { executor: "skill", skillName: "compound-engineering:ce-plan", prompt: "Plan the work." }, + }, + live, + {}, + undefined, + ); + + expect(result.outcome).toBe("success"); + expect(store.updateTask).toHaveBeenCalledWith( + "FN-CE-1", + { status: null, pausedReason: null }, + undefined, + ); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-CE-1", + "Workflow input marker 'commit-pr' already has a reply — clearing stale marker before step 'plan'", + undefined, + undefined, + ); + expect(live.pausedReason).toBeNull(); + }); + it("treats terminal graph step projection as success when the legacy pass rejects", async () => { const store = createMockStore(); store.getTask.mockResolvedValue(baseStepTask({ diff --git a/packages/engine/src/__tests__/reliability-interactions/merge-node-paused-abort-retryable.test.ts b/packages/engine/src/__tests__/reliability-interactions/merge-node-paused-abort-retryable.test.ts index 9095f1a901..ea2e05a0ec 100644 --- a/packages/engine/src/__tests__/reliability-interactions/merge-node-paused-abort-retryable.test.ts +++ b/packages/engine/src/__tests__/reliability-interactions/merge-node-paused-abort-retryable.test.ts @@ -212,7 +212,7 @@ describe("merge-node paused-abort retry classification (FN-6735)", () => { ); }); - it("does not retry merge-confirmed partial landing evidence", async () => { + it("finalizes merge-confirmed partial landing evidence without retrying merge", async () => { const { store, task, executor, mergeRequester } = makeHarness({ mergeDetails: { mergeConfirmed: true } as any }); await invokeGraphFailure(executor, task, "merge"); @@ -220,8 +220,7 @@ describe("merge-node paused-abort retry classification (FN-6735)", () => { expect(mergeRequester).not.toHaveBeenCalled(); expect(store.updateTask).toHaveBeenCalledWith( task.id, - expect.objectContaining({ status: "failed", error: expect.stringContaining("operator action required") }), - undefined, + expect.objectContaining({ status: null, error: null, paused: false }), ); }); }); diff --git a/packages/engine/src/__tests__/workflow-malformed-verdict-gate.test.ts b/packages/engine/src/__tests__/workflow-malformed-verdict-gate.test.ts index e0c218d0c2..85035cc6cf 100644 --- a/packages/engine/src/__tests__/workflow-malformed-verdict-gate.test.ts +++ b/packages/engine/src/__tests__/workflow-malformed-verdict-gate.test.ts @@ -44,6 +44,7 @@ describe("workflow malformed-verdict gate", () => { notes: "", }); expect(parseWorkflowStepOutput("lorem ipsum")).toEqual({ output: "lorem ipsum", malformed: true }); + expect(parseWorkflowStepOutput("native skill output", { requireVerdict: false })).toEqual({ output: "native skill output" }); }); it("keeps a malformed blocking graph gate from producing a passing outcome", async () => { diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index cf9b57dd91..4406aeaaa8 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1164,6 +1164,18 @@ export function parseWorkflowStepOutput(rawOutput: string): { verdict?: "APPROVE" | "APPROVE_WITH_NOTES" | "REVISE"; notes?: string; malformed?: boolean; +}; +export function parseWorkflowStepOutput(rawOutput: string, options: { requireVerdict: false }): { + output: string; + verdict?: "APPROVE" | "APPROVE_WITH_NOTES" | "REVISE"; + notes?: string; + malformed?: boolean; +}; +export function parseWorkflowStepOutput(rawOutput: string, options: { requireVerdict?: boolean } = {}): { + output: string; + verdict?: "APPROVE" | "APPROVE_WITH_NOTES" | "REVISE"; + notes?: string; + malformed?: boolean; } { const trimmed = rawOutput.trim(); const parsed = parseWorkflowStepVerdict(trimmed); @@ -1184,6 +1196,10 @@ export function parseWorkflowStepOutput(rawOutput: string): { }; } + if (options.requireVerdict === false) { + return { output: trimmed }; + } + return { output: trimmed, malformed: true }; } @@ -5764,6 +5780,12 @@ export class TaskExecutor { if (!this.mergeRequester) { return { outcome: "failure", value: "merge-unavailable", data: { status: "failed", reason: "merge-unavailable" } }; } + const mergeTask = await this.ensureWorkflowMergeBoundaryTask(task, { + reason: "workflow-merge-boundary", + nodeId: ctx.node.node.id, + workflowId: ctx.run.workflowId, + runId: ctx.run.runId, + }); const GRAPH_MERGE_TIMEOUT_MS = 30 * 60 * 1000; const controller = new AbortController(); let timeoutHandle: ReturnType | undefined; @@ -5775,9 +5797,9 @@ export class TaskExecutor { timeoutHandle.unref?.(); }); try { - const result = await Promise.race([this.mergeRequester(task.id, { signal: controller.signal }), timeout]); + const result = await Promise.race([this.mergeRequester(mergeTask.id, { signal: controller.signal }), timeout]); if (result === "timeout") { - executorLog.warn(`${task.id}: workflow merge primitive timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`); + executorLog.warn(`${mergeTask.id}: workflow merge primitive timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`); return { outcome: "failure", value: "merge-timeout", data: { status: "timeout" } }; } if (result.merged || result.noOp) { @@ -5787,14 +5809,14 @@ export class TaskExecutor { */ const finalization = await finalizeProvenAutoMergeTask({ store: this.store, - taskId: task.id, + taskId: mergeTask.id, result, rootDir: this.rootDir, audit: createRunAuditor(this.store, { runId: ctx.run.runId, agentId: "executor", - taskId: task.id, - taskLineageId: task.lineageId, + taskId: mergeTask.id, + taskLineageId: mergeTask.lineageId, phase: "workflow-merge", }), auditAgentId: "executor", @@ -5822,7 +5844,7 @@ export class TaskExecutor { }; } finally { if (timeoutHandle) clearTimeout(timeoutHandle); - await logAudit(task.id, { + await logAudit(mergeTask.id, { type: "merge-requested", message: `Workflow node ${ctx.node.node.id} requested merge`, }); @@ -5844,6 +5866,38 @@ export class TaskExecutor { }; } + private async ensureWorkflowMergeBoundaryTask( + task: TaskDetail, + metadata: { reason: string; nodeId: string; workflowId: string; runId: string }, + ): Promise { + const live = await this.store.getTask(task.id); + if (!live) return task; + if (live.column === "in-review" || live.column === "done") return live; + if (live.paused || live.userPaused) return live; + + /* + FNXC:WorkflowMerge 2026-06-29-10:15: + User-authored workflows may legitimately route execution directly to a merge node without an explicit review node. Reaching that node is the workflow-owned merge boundary, so the engine must establish the durable in-review/merge lifecycle handoff before requesting merge instead of assuming a prior node already moved the card. + */ + const moveOptions = { + preserveProgress: true, + moveSource: "engine" as const, + workflowMoveSource: "workflow-graph", + workflowMoveMetadata: metadata, + }; + const storeWithMove = this.store as typeof this.store & { + moveTask?: (id: string, column: string, options?: unknown) => Promise; + }; + if (typeof storeWithMove.moveTask === "function") { + const moved = await storeWithMove.moveTask(live.id, "in-review", moveOptions); // handoff-invariant-violation-allowlist: workflow merge node owns the merge lifecycle boundary for custom workflows. + await this.store.logEntry(live.id, "Workflow merge boundary moved task to in-review before requesting merge", undefined, this.getRunContextFor(live.id)); + return moved ?? { ...live, column: "in-review" }; + } + await this.store.updateTask(live.id, { column: "in-review" } as Partial, this.getRunContextFor(live.id)); + await this.store.logEntry(live.id, "Workflow merge boundary moved task to in-review before requesting merge", undefined, this.getRunContextFor(live.id)); + return { ...live, column: "in-review" }; + } + public createAuthoritativeWorkflowSeams(_settings: Settings): WorkflowLegacySeams { return { // Built-in triage/spec generation runs upstream of the interpreter today, @@ -5905,6 +5959,12 @@ export class TaskExecutor { if (!this.mergeRequester) { return { outcome: "failure", value: "merge-unavailable" }; } + const mergeTask = await this.ensureWorkflowMergeBoundaryTask(seamTask, { + reason: "workflow-merge-boundary", + nodeId: "legacy-merge-seam", + workflowId: "legacy-seams", + runId: this.getRunContextFor(seamTask.id)?.runId ?? "legacy-seam", + }); // Bound the wait: a wedged merge queue must not strand the graph walk // holding the routing claim. On timeout the run fails cleanly and the // task is parked for human review; the queue can still finish later. @@ -5915,9 +5975,9 @@ export class TaskExecutor { timeoutHandle.unref?.(); }); try { - const result = await Promise.race([this.mergeRequester(seamTask.id), timeout]); + const result = await Promise.race([this.mergeRequester(mergeTask.id), timeout]); if (result === "timeout") { - executorLog.warn(`${seamTask.id}: graph merge seam timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`); + executorLog.warn(`${mergeTask.id}: graph merge seam timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`); return { outcome: "failure", value: "merge-timeout" }; } if (result.merged || result.noOp) { @@ -6660,7 +6720,11 @@ export class TaskExecutor { columnBinding?: WorkflowColumnAgent, ): Promise { const cfg = node.config ?? {}; - const live = await this.store.getTask(nodeTask.id); + let live = await this.store.getTask(nodeTask.id); + + const staleInput = await this.resolveWorkflowInputMarkerForGraphNode(live, node.id); + if (staleInput === "waiting") return { outcome: "failure", value: "awaiting-user-input" }; + if (staleInput === "clear") live = await this.store.getTask(nodeTask.id); // Await-input nodes never run a session — they pause for the user. if (cfg.awaitInput === true) { @@ -7584,7 +7648,13 @@ export class TaskExecutor { executorLog.warn(`${live.id}: ${message}`); await this.store.logEntry(live.id, message, undefined, this.getRunContextFor(live.id)); try { - await this.mergeRequester(live.id); + const mergeTask = await this.ensureWorkflowMergeBoundaryTask(live, { + reason: "workflow-merge-retry-boundary", + nodeId: failedNode, + workflowId: result.context?.["workflow:id"] as string | undefined ?? "workflow-graph", + runId: this.getRunContextFor(live.id)?.runId ?? "graph-merge-retry", + }); + await this.mergeRequester(mergeTask.id); } catch (error) { executorLog.warn(`${live.id}: bounded auto-merge retry request failed after graph merge failure: ${error instanceof Error ? error.message : String(error)}`); } @@ -13550,6 +13620,49 @@ ${failureFeedback} return parseWorkflowStepOutput(rawOutput); } + private workflowInputRepliesAfterWatermark(task: TaskDetail, marker: string): Array<{ createdAt?: string }> { + const pausedReason = task.pausedReason ?? ""; + const watermark = (() => { + const match = pausedReason.slice(marker.length).match(/^@(\d+)/); + const parsed = match ? Number(match[1]) : NaN; + return Number.isFinite(parsed) ? parsed : undefined; + })(); + const steering = Array.isArray(task.steeringComments) ? task.steeringComments : []; + return watermark === undefined + ? steering + : steering.filter((comment) => { + const created = Date.parse((comment as { createdAt?: string }).createdAt ?? ""); + return Number.isFinite(created) ? created >= watermark : false; + }); + } + + private async resolveWorkflowInputMarkerForGraphNode(live: TaskDetail, nodeId: string): Promise<"clear" | "waiting" | "none"> { + const pausedReason = live.pausedReason ?? ""; + if (!pausedReason.startsWith("workflow-input:")) return "none"; + const markerMatch = /^workflow-input:([^:@\s]+)(?:@\d+)?[:]/.exec(pausedReason); + if (!markerMatch) return "none"; + const marker = `workflow-input:${markerMatch[1]}`; + const replies = this.workflowInputRepliesAfterWatermark(live, marker); + if (live.paused || replies.length === 0) { + await this.store.updateTask(live.id, { status: "awaiting-user-input", paused: true }, this.getRunContextFor(live.id)); + return "waiting"; + } + /* + * FNXC:WorkflowInput 2026-06-29-10:00: + * A workflow graph can restart at an earlier node after pause/resume recovery while the durable pausedReason still points at the later skill node that asked the question. If the user already supplied a post-watermark reply, clear that stale marker before any node executes so Compound Engineering cannot loop at Plan while Commit & open PR's answered question remains attached. + */ + await this.store.updateTask(live.id, { status: null, pausedReason: null }, this.getRunContextFor(live.id)); + await this.store.logEntry( + live.id, + marker === `workflow-input:${nodeId}` + ? `Workflow input received for step '${nodeId}' — resuming` + : `Workflow input marker '${markerMatch[1]}' already has a reply — clearing stale marker before step '${nodeId}'`, + undefined, + this.getRunContextFor(live.id), + ); + return "clear"; + } + /** * Execute a single workflow step by spawning an agent with the step's prompt. * Returns structured outcome with support for revision requests. @@ -13949,7 +14062,7 @@ You have access to the file system to review changes.${verdictBlock}`; session.dispose(); await agentLogger.flush(); - const parsed = this.parseWorkflowStepOutput(output); + const parsed = requireVerdict ? parseWorkflowStepOutput(output) : parseWorkflowStepOutput(output, { requireVerdict: false }); if (parsed.verdict) { const revisionRequested = parsed.verdict === "REVISE"; if (workflowStep.requiresBrowser === true) {