From 18ba5c0c5931f4218a3a8594d7a69c81122dc2a7 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 21:02:01 -0700 Subject: [PATCH] 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) --- packages/core/src/db.ts | 1 + .../__tests__/interpreter-merge-seam.test.ts | 42 ++++++++ ...dual-observe-shadow-terminal-stage.test.ts | 98 +++++++++++++++++++ packages/engine/src/executor.ts | 12 ++- packages/engine/src/project-engine.ts | 11 ++- 5 files changed, 161 insertions(+), 3 deletions(-) create mode 100644 packages/engine/src/__tests__/reliability-interactions/dual-observe-shadow-terminal-stage.test.ts diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 5a0bd6ea02..1a19f33a42 100644 --- a/packages/core/src/db.ts +++ b/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 diff --git a/packages/engine/src/__tests__/interpreter-merge-seam.test.ts b/packages/engine/src/__tests__/interpreter-merge-seam.test.ts index b4a42c1c1f..90a28c05a0 100644 --- a/packages/engine/src/__tests__/interpreter-merge-seam.test.ts +++ b/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(); + }); }); }); diff --git a/packages/engine/src/__tests__/reliability-interactions/dual-observe-shadow-terminal-stage.test.ts b/packages/engine/src/__tests__/reliability-interactions/dual-observe-shadow-terminal-stage.test.ts new file mode 100644 index 0000000000..198fd7933b --- /dev/null +++ b/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 { + 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"]); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index cde1425eb6..b2d88c30b7 100644 --- a/packages/engine/src/executor.ts +++ b/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({ diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 6ec6d18594..4cf1b162c9 100644 --- a/packages/engine/src/project-engine.ts +++ b/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,