Address PR review feedback (#1363)

- db.ts: restrict migration-105 orphan-step cleanup to JSON arrays
  (json_type guard so json_each can't expand objects/strings)
- project-engine.ts: requestInterpreterMerge throws on null task lookup
  instead of casting null into MergeResult (seam converts to clean failure)
- executor.ts: truncate dual-observe shadow stage walk at the live terminal
  stage so healthy in-review tasks don't record a phantom merge transition

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/core/src/db.ts

@@ -4129,6 +4129,7 @@ export class Database {
             FROM task_workflow_selection sel
             JOIN json_each(sel.stepIds) je
             WHERE json_valid(sel.stepIds)
+              AND json_type(sel.stepIds) = 'array'
               AND sel.taskId NOT IN (SELECT id FROM tasks)
           );
           DELETE FROM task_workflow_selection

packages/engine/src/__tests__/interpreter-merge-seam.test.ts

@@ -84,5 +84,47 @@ describe("interpreter merge seam", () => {
       expect(onMerge).toHaveBeenCalledWith("FN-3");
       expect(result.merged).toBe(true);
     });
+
+    it("throws (never returns a null-task MergeResult) when the task lookup yields nothing", async () => {
+      // getTask returning null (deleted task / failed lookup) must not produce a
+      // MergeResult whose `task` is a null cast — callers dereference result.task.
+      const onMerge = vi.fn();
+      const fakeEngine = {
+        runtime: {
+          getTaskStore: () => ({
+            getSettings: async () => ({ autoMerge: true, globalPause: false, enginePaused: false }),
+            getTask: async () => null,
+          }),
+        },
+        allowInReviewMergeProcessing: () => true,
+        onMerge,
+      };
+
+      await expect(
+        (ProjectEngine.prototype as any).requestInterpreterMerge.call(fakeEngine, "FN-404"),
+      ).rejects.toThrow(/FN-404/);
+      expect(onMerge).not.toHaveBeenCalled();
+    });
+
+    it("throws when getTask itself rejects (lookup failure), not a null-task result", async () => {
+      const onMerge = vi.fn();
+      const fakeEngine = {
+        runtime: {
+          getTaskStore: () => ({
+            getSettings: async () => ({ autoMerge: true, globalPause: false, enginePaused: false }),
+            getTask: async () => {
+              throw new Error("store offline");
+            },
+          }),
+        },
+        allowInReviewMergeProcessing: () => true,
+        onMerge,
+      };
+
+      await expect(
+        (ProjectEngine.prototype as any).requestInterpreterMerge.call(fakeEngine, "FN-500"),
+      ).rejects.toThrow(/FN-500/);
+      expect(onMerge).not.toHaveBeenCalled();
+    });
   });
 });

packages/engine/src/__tests__/reliability-interactions/dual-observe-shadow-terminal-stage.test.ts

@@ -0,0 +1,98 @@
+import { describe, expect, it } from "vitest";
+import type { TaskDetail, WorkflowRunObservation } from "@fusion/core";
+import {
+  BUILTIN_CODING_WORKFLOW_IR,
+  buildWorkflowObservationFromTask,
+} from "@fusion/core";
+
+import { TaskExecutor } from "../../executor.js";
+import { WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "../../workflow-parity-observer.js";
+
+// `buildShadowObservation` is private; exercise it via the prototype with a
+// minimal `this` that only supplies the store surface the method touches.
+const def = { ir: BUILTIN_CODING_WORKFLOW_IR } as const;
+const fakeStore = {
+  getTaskWorkflowSelection: () => ({ workflowId: "builtin-coding-workflow", stepIds: [] }),
+  getWorkflowDefinition: async () => ({ id: "builtin-coding-workflow", ir: BUILTIN_CODING_WORKFLOW_IR }),
+};
+const settings = {
+  experimentalFeatures: {
+    workflowGraphExecutor: true,
+    [WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG]: true,
+  },
+} as any;
+
+function buildShadow(live: TaskDetail, legacy: WorkflowRunObservation): Promise<WorkflowRunObservation> {
+  return (TaskExecutor.prototype as any).buildShadowObservation.call(
+    { store: fakeStore },
+    live,
+    def,
+    settings,
+    legacy,
+  );
+}
+
+describe("FN-5768 dual-observe shadow terminal-stage truncation", () => {
+  it("stops the shadow walk at review for a healthy in-review task (no phantom merge stage)", async () => {
+    const live = {
+      id: "FN-IR",
+      column: "in-review",
+      status: null,
+      review: { verdict: "APPROVE" },
+      mergeDetails: null,
+    } as unknown as TaskDetail;
+
+    const legacy = buildWorkflowObservationFromTask(
+      { column: "in-review", status: null, review: { verdict: "APPROVE" }, mergeDetails: null },
+      { columnSequence: ["in-progress", "in-review"] },
+    );
+
+    const shadow = await buildShadow(live, legacy);
+
+    // The graph walker visits the merge node before invoking its seam, so the
+    // raw walk would record ["execute","review","merge"]; truncation at the live
+    // terminal stage must drop the phantom merge so it matches the legacy side.
+    expect(shadow.stageTransitions).toEqual(["execute", "review"]);
+    expect(shadow.stageTransitions).toEqual([...legacy.stageTransitions]);
+    expect(shadow.mergeOutcome).toBeNull();
+  });
+
+  it("keeps the merge stage for a merged (done) task", async () => {
+    const live = {
+      id: "FN-DONE",
+      column: "done",
+      status: null,
+      review: { verdict: "APPROVE" },
+      mergeDetails: { outcome: "merged" },
+    } as unknown as TaskDetail;
+
+    const legacy = buildWorkflowObservationFromTask(
+      { column: "done", status: null, review: { verdict: "APPROVE" }, mergeDetails: { outcome: "merged" } },
+      { columnSequence: ["in-progress", "in-review", "done"] },
+    );
+
+    const shadow = await buildShadow(live, legacy);
+
+    expect(shadow.stageTransitions).toEqual(["execute", "review", "merge"]);
+    expect(shadow.mergeOutcome).toBe("merged");
+  });
+
+  it("stops at execute for a task still in-progress", async () => {
+    const live = {
+      id: "FN-WIP",
+      column: "in-progress",
+      status: null,
+      review: null,
+      mergeDetails: null,
+    } as unknown as TaskDetail;
+
+    const legacy = buildWorkflowObservationFromTask(
+      { column: "in-progress", status: null, review: null, mergeDetails: null },
+      { columnSequence: ["in-progress"] },
+    );
+
+    const shadow = await buildShadow(live, legacy);
+
+    expect(shadow.stageTransitions).toEqual(["execute"]);
+  });
+});

packages/engine/src/executor.ts

@@ -3377,10 +3377,20 @@ export class TaskExecutor {
         stageByNodeId.set(node.id, seam);
       }
     }
+    // Stop the shadow walk at the live terminal seam. The graph walker visits a
+    // node *before* invoking its seam, so even a failing merge seam (the case
+    // when the live task is parked in-review with autoMerge off) still records a
+    // "merge" stage. The legacy side never reports merge for an in-review task,
+    // so that phantom stage manufactures stageTransitions drift on healthy runs.
+    // Truncate the visited-stage sequence at the stage the live task actually
+    // reached: merged → merge, reachedReview → review, else → execute.
+    const terminalStage: WorkflowStage = merged ? "merge" : reachedReview ? "review" : "execute";
     const stages: WorkflowStage[] = [];
     for (const nodeId of result.visitedNodeIds) {
       const stage = stageByNodeId.get(nodeId);
-      if (stage && stages[stages.length - 1] !== stage) stages.push(stage);
+      if (!stage || stages[stages.length - 1] === stage) continue;
+      stages.push(stage);
+      if (stage === terminalStage) break;
     }
 
     return buildWorkflowObservation({

packages/engine/src/project-engine.ts

@@ -1065,10 +1065,17 @@ export class ProjectEngine {
       && this.allowInReviewMergeProcessing(task, settings)
       && !(task.paused && !task.mergeDetails?.mergeConfirmed);
     if (!eligible) {
+      // A null task means the lookup failed or the task was deleted; never hand
+      // back a MergeResult with `task` cast from null — callers dereference
+      // result.task. Throw so the merge seam (which converts seam throws into a
+      // clean "failure" outcome) parks the task for human review.
+      if (!task) {
+        throw new Error(`Interpreter merge for ${taskId} aborted: task not found (deleted or lookup failed)`);
+      }
       runtimeLog.log(`Interpreter merge for ${taskId} not auto-eligible (autoMerge off / not ready) — manual merge required`);
       return {
-        task: task as Task,
-        branch: task?.branch ?? "",
+        task,
+        branch: task.branch ?? "",
         merged: false,
         worktreeRemoved: false,
         branchDeleted: false,