fix(FN-7232): enforce workflow merge boundary

.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.

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<string, unknown>) => {
+        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({

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") }),
+      expect.objectContaining({ status: null, error: null, paused: false }),
-      undefined,
     );
   });
 });

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 () => {

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<typeof setTimeout> | 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,
+                taskId: mergeTask.id,
-                taskLineageId: task.lineageId,
+                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<TaskDetail> {
+    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<TaskDetail | undefined>;
+    };
+    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<TaskDetail>, 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<WorkflowNodeResult> {
     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) {