diff --git a/.changeset/FN-5770-workflow-interpreter-authoritative.md b/.changeset/FN-5770-workflow-interpreter-authoritative.md new file mode 100644 index 0000000000..59f8da2a68 --- /dev/null +++ b/.changeset/FN-5770-workflow-interpreter-authoritative.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": minor +--- + +Add a guarded interpreter-authoritative workflow cutover for coding-task lifecycle execution. The new capability stays default-off behind `experimentalFeatures.workflowInterpreterAuthoritative` and only activates when rollout-readiness checks pass, preserving legacy execution as the fallback path. diff --git a/docs/architecture.md b/docs/architecture.md index 7b9e8dccf8..3109da5b67 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1820,6 +1820,7 @@ Reliability-layer changes are in scope. Interaction regression backstops live in - FN-5741 backstop: `packages/engine/src/__tests__/reliability-interactions/merge-request-shadow-handoff.test.ts` guards Phase-1 merge-request contract shadow writes: flag OFF is a no-op, flag ON writes marker/record strictly after legacy handoff, and `autoMerge:false` remains `manual-required` without shadow running transitions. - FN-5742 backstop: `packages/engine/src/__tests__/reliability-interactions/dual-observe-merge-seam.test.ts` guards Phase-2 dual-observe invariants: legacy dependency satisfaction remains authoritative while parity diffs emit, and shadow dequeue selection never advances `manual-required` rows. - FN-5743 backstop: `packages/engine/src/__tests__/reliability-interactions/merge-request-cancel-on-hard-cancel.test.ts` and `packages/core/src/__tests__/merge-request-record.test.ts` guard Phase-3 cutover invariants: transient merge retries mutate merge-request state (no column rebound), user hard-cancel after accepted handoff cancels pending merge requests, and non-user rebounds preserve legacy fail-soft semantics. +- FN-5770 backstop: `packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts` guards the interpreter-authoritative lifecycle seam. The cutover remains opt-in (`workflowInterpreterAuthoritative` default OFF), readiness-gated by dual-observe parity evidence, reversible by flipping one flag back OFF, and must preserve file-scope, squash-overlap, `autoMerge:false`, hard-cancel, and self-healing interaction invariants. - FN-5337 backstop: `packages/engine/src/__tests__/reliability-interactions/orphan-detected-no-requeue.test.ts` locks observation-only orphan detection across FN-5279 repro metadata desync, worktree-present and worktree-missing candidates, FN-5219 ordering, FN-5147 in-review isolation, FN-5083 branch-cleared composition, lease-manager non-invocation, and per-sweep idempotent audit emission. - FN-5256 backstop: `packages/engine/src/__tests__/reliability-interactions/dependency-cycle-reconcile.test.ts` covers persisted dependency-cycle detection via `reconcileDependencyCycles`, bounded umbrella-back-edge auto-repair, ambiguous-cycle observe-only behavior, composition ordering with `reconcileSelfDefeatingDependencies`, and the post-sweep write-time guard invariant. Core write-boundary regressions (FN-5240/5241/5242 signature, indirect cycle, umbrella back-edge rejection) live in `packages/core/src/__tests__/store-dependency-cycle.test.ts`. - FN-5325 backstop: `packages/engine/src/__tests__/reliability-interactions/scheduler-overlap-priority-inversion.test.ts` covers queued-overlap priority/age deferral, equal-priority age ordering, FN-4969 fanout composition, and transition-only `scheduler:overlap-priority-inversion` audit surfacing across unchanged, changed-blocker, and clear→reappear states. diff --git a/docs/settings-reference.md b/docs/settings-reference.md index c13e065959..94712f71a3 100644 --- a/docs/settings-reference.md +++ b/docs/settings-reference.md @@ -1216,7 +1216,7 @@ The Experimental Features section in Settings shows: - Global scope indicator (features are shared across projects) - Description explaining the purpose of experimental features -Common built-in dashboard flags include: +Common built-in dashboard/runtime flags include: - `insights` - `roadmap` - `memoryView` @@ -1226,6 +1226,9 @@ Common built-in dashboard flags include: - `todoView` (enables dashboard Todo View; see [Todo View](./todo-view.md)) - `researchView` - `evalsView` (gates Evals dashboard view, Settings → Scheduled Evals section, and scheduled-eval cron execution) +- `workflowGraphExecutor` (enables the workflow-IR interpreter path) +- `workflowInterpreterDualObserve` (observe-only parity instrumentation for interpreter rollout) +- `workflowInterpreterAuthoritative` (readiness-gated authoritative interpreter lifecycle cutover; legacy remains default/fallback when OFF) - `remoteAccess` - `agentOnboarding` (enables the **AI Interview** option inside the New Agent dialog) diff --git a/docs/workflow-steps.md b/docs/workflow-steps.md index 59dc51b1c7..698ec0b53b 100644 --- a/docs/workflow-steps.md +++ b/docs/workflow-steps.md @@ -410,7 +410,7 @@ Run-audit events emitted in `database` domain: The parity contract is exported from `@fusion/core` (`compareWorkflowRunObservations`, `compareWorkflowRunAudits`) and produces deterministic drift reports shaped as `{ agree, diffs[] }`, where each diff includes field name, legacy/interpreter values, category, and severity. -This is a dual-observe stage only; interpreter-authoritative cutover is deferred to a later phase. +Dual-observe remains the rollout evidence path for the later authoritative cutover: the interpreter may only become authoritative when the separate `experimentalFeatures.workflowInterpreterAuthoritative` flag is ON **and** the cutover-readiness guard reports zero unresolved parity drift. #### Self-healing recovery for parked review tasks @@ -480,6 +480,26 @@ Traversal semantics: Parity coverage includes flag-OFF no-op behavior, lifecycle ordering parity vs legacy seams, merge/file-scope-like failure routing, and downstream halt behavior for hard-cancel/self-healing style failures. +### Interpreter-authoritative cutover + +A second default-OFF flag, `experimentalFeatures.workflowInterpreterAuthoritative`, promotes the interpreter from shadow/selected-workflow sequencing to the **authoritative** lifecycle driver for default coding tasks. + +The cutover stays opt-in, guarded, and reversible: +- **Default OFF:** legacy executor/reviewer/merger/scheduler flow remains authoritative. +- **Guarded ON:** the engine only routes through the authoritative driver when `evaluateInterpreterCutoverReadiness(...)` reports ready. The guard consumes explicit rollout evidence (cutover flag enabled, dual-observe enabled, non-empty parity observations, zero unresolved drift). +- **Rollback:** turning `workflowInterpreterAuthoritative` back OFF immediately restores the legacy path; no migration or cleanup step is required. + +When the guard passes, the runtime binds real DI seams from `TaskExecutor` into the built-in coding IR and drives `BUILTIN_CODING_WORKFLOW_IR` through `WorkflowGraphExecutor`. The interpreter does **not** reimplement lifecycle behavior: it delegates execute/review/merge to the same legacy seams already used by the imperative path. + +Reliability invariants preserved under authoritative mode: +- file-scope enforcement including `FileScopeViolationError` +- squash/file-scope overlap enforcement via `assertSquashOverlapsFileScope` +- `autoMerge: false` terminal-until-merged behavior in `in-review` +- `moveTask(in-progress → todo)` hard-cancel semantics without stray `userPaused` rebounds +- existing self-healing routing and fail-soft fallback behavior + +The interaction backstop lives in `packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts`. + ## Workflow Step APIs | Endpoint | Purpose | diff --git a/packages/core/src/__tests__/workflow-cutover.test.ts b/packages/core/src/__tests__/workflow-cutover.test.ts new file mode 100644 index 0000000000..246bc7192d --- /dev/null +++ b/packages/core/src/__tests__/workflow-cutover.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "vitest"; + +import { evaluateInterpreterCutoverReadiness } from "../workflow-cutover.js"; + +describe("workflow interpreter authoritative cutover readiness", () => { + it("is ready when the cutover flag, dual-observe, and clean parity evidence are present", () => { + const result = evaluateInterpreterCutoverReadiness({ + authoritativeFlagEnabled: true, + dualObserveEnabled: true, + paritySummary: { observed: 5, drift: 0, recentDrift: [] }, + minimumObservedRuns: 3, + }); + + expect(result).toEqual({ ready: true, reasons: [] }); + }); + + it("enumerates every failed criterion deterministically", () => { + const result = evaluateInterpreterCutoverReadiness({ + authoritativeFlagEnabled: false, + dualObserveEnabled: false, + paritySummary: null, + }); + + expect(result.ready).toBe(false); + expect(result.reasons).toEqual([ + "experimentalFeatures.workflowInterpreterAuthoritative is disabled", + "experimentalFeatures.workflowInterpreterDualObserve is disabled", + "workflow parity summary unavailable", + ]); + }); + + it("blocks when parity drift is present in the summary or unresolved drift reports", () => { + const result = evaluateInterpreterCutoverReadiness({ + authoritativeFlagEnabled: true, + dualObserveEnabled: true, + paritySummary: { + observed: 8, + drift: 2, + recentDrift: [ + { taskId: "FN-1", timestamp: "2026-06-01T00:00:00.000Z", diffs: [] }, + ], + }, + unresolvedDriftReports: [ + { agree: false, diffs: [{ field: "terminalColumn", legacy: "done", interpreter: "in-review", category: "lifecycle", severity: "error" }] }, + ], + }); + + expect(result.ready).toBe(false); + expect(result.reasons).toEqual([ + "workflow parity drift above zero (2 drift events)", + "workflow parity has unresolved drift reports (1)", + ]); + }); + + it("normalizes minimum observed runs to at least one", () => { + const result = evaluateInterpreterCutoverReadiness({ + authoritativeFlagEnabled: true, + dualObserveEnabled: true, + paritySummary: { observed: 0, drift: 0, recentDrift: [] }, + minimumObservedRuns: 0, + }); + + expect(result.ready).toBe(false); + expect(result.reasons).toEqual([ + "workflow parity observation window too small (0/1 observed)", + ]); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 90802a24c8..8b4589aaba 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -1446,6 +1446,14 @@ export type { WorkflowColumnsGraduationReport, GraduationReportInputs, } from "./workflow-parity.js"; +export { + WORKFLOW_INTERPRETER_AUTHORITATIVE_FLAG, + evaluateInterpreterCutoverReadiness, +} from "./workflow-cutover.js"; +export type { + InterpreterCutoverReadinessInput, + InterpreterCutoverReadinessResult, +} from "./workflow-cutover.js"; export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js"; export type { ResolvedResearchSettings } from "./research-settings.js"; export { isEvalsExperimentalEnabled, resolveEvalSettings } from "./eval-settings.js"; diff --git a/packages/core/src/workflow-cutover.ts b/packages/core/src/workflow-cutover.ts new file mode 100644 index 0000000000..8b793eed74 --- /dev/null +++ b/packages/core/src/workflow-cutover.ts @@ -0,0 +1,81 @@ +import type { WorkflowParityDriftReport, WorkflowParitySummary } from "./workflow-parity.js"; + +/** + * Opt-in authoritative cutover flag for routing the coding lifecycle through the + * workflow interpreter. The cutover is guarded by rollout-readiness checks and + * remains reversible by disabling the flag. + */ +export const WORKFLOW_INTERPRETER_AUTHORITATIVE_FLAG = "workflowInterpreterAuthoritative" as const; + +export interface InterpreterCutoverReadinessInput { + /** Explicit operator opt-in; default runtime remains legacy when false. */ + authoritativeFlagEnabled: boolean; + /** The dual-observe rollout must stay enabled/proven before cutover. */ + dualObserveEnabled: boolean; + /** Aggregated parity signal from the audit trail (for example `store.getWorkflowParitySummary()`). */ + paritySummary?: Pick | null; + /** Optional unresolved drift reports surfaced directly by the caller. */ + unresolvedDriftReports?: readonly Pick[] | null; + /** Minimum observed parity runs required before cutover may proceed. Default: 1. */ + minimumObservedRuns?: number; +} + +export interface InterpreterCutoverReadinessResult { + ready: boolean; + reasons: string[]; +} + +function normalizeMinimumObservedRuns(value: number | undefined): number { + if (!Number.isFinite(value)) return 1; + return Math.max(1, Math.floor(value!)); +} + +function countUnresolvedDriftReports( + reports: readonly Pick[] | null | undefined, +): number { + if (!reports || reports.length === 0) return 0; + return reports.filter((report) => report.agree === false || report.diffs.length > 0).length; +} + +/** + * Pure rollout-readiness guard for the interpreter-authoritative cutover. + * Callers supply explicit parity evidence; this function performs no I/O. + */ +export function evaluateInterpreterCutoverReadiness( + input: InterpreterCutoverReadinessInput, +): InterpreterCutoverReadinessResult { + const reasons: string[] = []; + const minimumObservedRuns = normalizeMinimumObservedRuns(input.minimumObservedRuns); + + if (!input.authoritativeFlagEnabled) { + reasons.push("experimentalFeatures.workflowInterpreterAuthoritative is disabled"); + } + + if (!input.dualObserveEnabled) { + reasons.push("experimentalFeatures.workflowInterpreterDualObserve is disabled"); + } + + const paritySummary = input.paritySummary; + if (!paritySummary) { + reasons.push("workflow parity summary unavailable"); + } else { + if (paritySummary.observed < minimumObservedRuns) { + reasons.push( + `workflow parity observation window too small (${paritySummary.observed}/${minimumObservedRuns} observed)`, + ); + } + if (paritySummary.drift > 0) { + reasons.push(`workflow parity drift above zero (${paritySummary.drift} drift events)`); + } + } + + const unresolvedDriftCount = countUnresolvedDriftReports(input.unresolvedDriftReports); + if (unresolvedDriftCount > 0) { + reasons.push(`workflow parity has unresolved drift reports (${unresolvedDriftCount})`); + } + + return { + ready: reasons.length === 0, + reasons, + }; +} diff --git a/packages/engine/src/__tests__/openclaw-runtime-e2e.test.ts b/packages/engine/src/__tests__/openclaw-runtime-e2e.test.ts index d6202e8bed..960b4e2fe9 100644 --- a/packages/engine/src/__tests__/openclaw-runtime-e2e.test.ts +++ b/packages/engine/src/__tests__/openclaw-runtime-e2e.test.ts @@ -42,6 +42,7 @@ vi.mock("../pi.js", () => ({ })); vi.mock("node:child_process", () => ({ + execFile: vi.fn(), spawn: (...args: unknown[]) => mockSpawn(...args), })); diff --git a/packages/engine/src/__tests__/pi-layers-wiring.test.ts b/packages/engine/src/__tests__/pi-layers-wiring.test.ts index 7dd78f0fb7..8c242edbab 100644 --- a/packages/engine/src/__tests__/pi-layers-wiring.test.ts +++ b/packages/engine/src/__tests__/pi-layers-wiring.test.ts @@ -64,7 +64,7 @@ vi.mock("node:child_process", () => { } }); }); - return { execSync: execSyncFn, exec: execFn }; + return { execSync: execSyncFn, exec: execFn, execFile: vi.fn() }; }); vi.mock("node:fs", async () => { diff --git a/packages/engine/src/__tests__/reliability-interactions/branch-recovery-live-zero-commits.test.ts b/packages/engine/src/__tests__/reliability-interactions/branch-recovery-live-zero-commits.test.ts index 2e37d2f4a0..82d0fbc6d1 100644 --- a/packages/engine/src/__tests__/reliability-interactions/branch-recovery-live-zero-commits.test.ts +++ b/packages/engine/src/__tests__/reliability-interactions/branch-recovery-live-zero-commits.test.ts @@ -12,7 +12,7 @@ vi.mock("node:child_process", async () => { .catch((err: Error) => callback?.(err, "", err.message)); }; execFn[promisify.custom] = (cmd: string, opts?: any) => execMock(cmd, opts).then((stdout: string) => ({ stdout, stderr: "" })); - return { exec: execFn, execSync: vi.fn() }; + return { exec: execFn, execSync: vi.fn(), execFile: vi.fn() }; }); import { SelfHealingManager } from "../../self-healing.js"; diff --git a/packages/engine/src/__tests__/reliability-interactions/branch-recovery-stale-cached-base.test.ts b/packages/engine/src/__tests__/reliability-interactions/branch-recovery-stale-cached-base.test.ts index f745d77ed7..366e79ccef 100644 --- a/packages/engine/src/__tests__/reliability-interactions/branch-recovery-stale-cached-base.test.ts +++ b/packages/engine/src/__tests__/reliability-interactions/branch-recovery-stale-cached-base.test.ts @@ -12,7 +12,7 @@ vi.mock("node:child_process", async () => { .catch((err: Error) => callback?.(err, "", err.message)); }; execFn[promisify.custom] = (cmd: string, opts?: any) => execMock(cmd, opts).then((stdout: string) => ({ stdout, stderr: "" })); - return { exec: execFn, execSync: vi.fn() }; + return { exec: execFn, execSync: vi.fn(), execFile: vi.fn() }; }); import { SelfHealingManager } from "../../self-healing.js"; diff --git a/packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts b/packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts new file mode 100644 index 0000000000..74d3a45e76 --- /dev/null +++ b/packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts @@ -0,0 +1,306 @@ +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { DEFAULT_SETTINGS, type Settings, type TaskDetail } from "@fusion/core"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { assertSquashOverlapsFileScope, FileScopeViolationError } from "../../merger.js"; +import type { WorkflowLegacySeams } from "../../workflow-node-handlers.js"; +import { WorkflowAuthoritativeDriver } from "../../workflow-authoritative-driver.js"; +import { git, hasGit, makeReliabilityFixture } from "./_helpers.js"; + +const readyParity = { + observed: 5, + agreed: 5, + drift: 0, + agreeRate: 1, + driftFieldCounts: {}, + recentDrift: [], +}; + +const baseTask = { + id: "FN-5770", + column: "in-progress", + steps: [], + review: null, + mergeDetails: null, +} as unknown as TaskDetail; + +function settingsWith(flags: Record, overrides: Partial = {}): Settings { + return { + ...DEFAULT_SETTINGS, + ...overrides, + experimentalFeatures: { + ...(DEFAULT_SETTINGS.experimentalFeatures ?? {}), + ...flags, + }, + } as Settings; +} + +function createStore(options: { + settings?: Settings; + selection?: { workflowId: string; stepIds: string[] } | undefined; + task?: TaskDetail; + paritySummary?: typeof readyParity | undefined; +} = {}) { + return { + getSettings: vi.fn(async () => options.settings ?? settingsWith({ workflowInterpreterAuthoritative: true, workflowInterpreterDualObserve: true })), + getTask: vi.fn(async () => options.task ?? baseTask), + getTaskWorkflowSelection: vi.fn(() => options.selection), + getWorkflowParitySummary: vi.fn(() => options.paritySummary ?? readyParity), + }; +} + +function createExecutor(seams: WorkflowLegacySeams) { + return { + createAuthoritativeWorkflowSeams: vi.fn(() => seams), + }; +} + +describe("workflow interpreter authoritative cutover", () => { + it("is a strict no-op when the cutover flag is off", async () => { + const store = createStore({ + settings: settingsWith({ workflowInterpreterAuthoritative: false, workflowInterpreterDualObserve: true }), + }); + const executor = createExecutor({ + planning: vi.fn(async () => ({ outcome: "success" as const })), + execute: vi.fn(async () => ({ outcome: "success" as const })), + review: vi.fn(async () => ({ outcome: "success" as const })), + merge: vi.fn(async () => ({ outcome: "success" as const })), + schedule: vi.fn(async () => ({ outcome: "success" as const })), + }); + + const result = await new WorkflowAuthoritativeDriver({ store, executor }).maybeRun(baseTask as any); + + expect(result.handled).toBe(false); + expect(result.disposition).toBe("fell-back"); + expect(executor.createAuthoritativeWorkflowSeams).not.toHaveBeenCalled(); + }); + + it("falls back when readiness fails even if the cutover flag is on", async () => { + const store = createStore({ + paritySummary: { ...readyParity, observed: 4, drift: 1 }, + }); + const executor = createExecutor({ + planning: vi.fn(async () => ({ outcome: "success" as const })), + execute: vi.fn(async () => ({ outcome: "success" as const })), + review: vi.fn(async () => ({ outcome: "success" as const })), + merge: vi.fn(async () => ({ outcome: "success" as const })), + schedule: vi.fn(async () => ({ outcome: "success" as const })), + }); + + const result = await new WorkflowAuthoritativeDriver({ store, executor }).maybeRun(baseTask as any); + + expect(result.handled).toBe(false); + expect(result.reason).toMatch(/drift above zero/); + expect(executor.createAuthoritativeWorkflowSeams).not.toHaveBeenCalled(); + }); + + it("drives execute → review → merge through authoritative seams on a clean run", async () => { + const calls: string[] = []; + const executor = createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => { + calls.push("execute"); + return { outcome: "success" as const }; + }, + review: async () => { + calls.push("review"); + return { outcome: "success" as const }; + }, + merge: async () => { + calls.push("merge"); + return { outcome: "success" as const }; + }, + schedule: async () => ({ outcome: "success" as const }), + }); + + const result = await new WorkflowAuthoritativeDriver({ store: createStore(), executor }).maybeRun(baseTask as any); + + expect(result.handled).toBe(true); + expect(result.disposition).toBe("completed"); + expect(calls).toEqual(["execute", "review", "merge"]); + }); + + it("keeps autoMerge:false tasks terminal in review by stopping before merge", async () => { + const merge = vi.fn(async () => ({ outcome: "success" as const })); + const executor = createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => ({ outcome: "success" as const }), + review: async () => ({ outcome: "failure" as const, value: "manual-merge-required" }), + merge, + schedule: async () => ({ outcome: "success" as const }), + }); + + const result = await new WorkflowAuthoritativeDriver({ + store: createStore({ settings: settingsWith({ workflowInterpreterAuthoritative: true, workflowInterpreterDualObserve: true }, { autoMerge: false }) }), + executor, + }).maybeRun(baseTask as any); + + expect(result.handled).toBe(true); + expect(result.disposition).toBe("failed"); + expect(merge).not.toHaveBeenCalled(); + }); + + it("preserves moveTask hard-cancel semantics by halting downstream seams without setting userPaused", async () => { + const review = vi.fn(async () => ({ outcome: "success" as const })); + const merge = vi.fn(async () => ({ outcome: "success" as const })); + const task = { ...baseTask, userPaused: undefined } as TaskDetail; + const executor = createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => ({ outcome: "failure" as const, value: "hard-cancel" }), + review, + merge, + schedule: async () => ({ outcome: "success" as const }), + }); + + const result = await new WorkflowAuthoritativeDriver({ + store: createStore({ task }), + executor, + }).maybeRun(task as any); + + expect(result.handled).toBe(true); + expect(result.disposition).toBe("failed"); + expect(review).not.toHaveBeenCalled(); + expect(merge).not.toHaveBeenCalled(); + expect(task.userPaused).toBeUndefined(); + }); + + it("routes self-healing style execute failures without divergent downstream lifecycle mutations", async () => { + const review = vi.fn(async () => ({ outcome: "success" as const })); + const merge = vi.fn(async () => ({ outcome: "success" as const })); + const executor = createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => ({ outcome: "failure" as const, value: "recoverable" }), + review, + merge, + schedule: async () => ({ outcome: "success" as const }), + }); + + const result = await new WorkflowAuthoritativeDriver({ store: createStore(), executor }).maybeRun(baseTask as any); + + expect(result.handled).toBe(true); + expect(result.disposition).toBe("failed"); + expect(review).not.toHaveBeenCalled(); + expect(merge).not.toHaveBeenCalled(); + }); + + it("immediately rolls back to legacy when the cutover flag is flipped back off", async () => { + let settings = settingsWith({ workflowInterpreterAuthoritative: true, workflowInterpreterDualObserve: true }); + const store = createStore(); + store.getSettings.mockImplementation(async () => settings); + const executor = createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => ({ outcome: "success" as const }), + review: async () => ({ outcome: "success" as const }), + merge: async () => ({ outcome: "success" as const }), + schedule: async () => ({ outcome: "success" as const }), + }); + const driver = new WorkflowAuthoritativeDriver({ store, executor }); + + const first = await driver.maybeRun(baseTask as any); + settings = settingsWith({ workflowInterpreterAuthoritative: false, workflowInterpreterDualObserve: true }); + const second = await driver.maybeRun(baseTask as any); + + expect(first.handled).toBe(true); + expect(second.handled).toBe(false); + expect(executor.createAuthoritativeWorkflowSeams).toHaveBeenCalledTimes(1); + }); + + it("defers to existing selected custom workflows instead of double-driving", async () => { + const executor = createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => ({ outcome: "success" as const }), + review: async () => ({ outcome: "success" as const }), + merge: async () => ({ outcome: "success" as const }), + schedule: async () => ({ outcome: "success" as const }), + }); + + const result = await new WorkflowAuthoritativeDriver({ + store: createStore({ selection: { workflowId: "WF-123", stepIds: [] } }), + executor, + }).maybeRun(baseTask as any); + + expect(result.handled).toBe(false); + expect(result.reason).toContain("workflow selection already present"); + expect(executor.createAuthoritativeWorkflowSeams).not.toHaveBeenCalled(); + }); +}); + +const describeIfGit = hasGit ? describe : describe.skip; + +describeIfGit("workflow interpreter authoritative cutover + file-scope invariants", () => { + const fixtures: Array>> = []; + + afterEach(async () => { + while (fixtures.length) { + await fixtures.pop()!.cleanup(); + } + }); + + async function createRealDriverFixture() { + const fx = await makeReliabilityFixture({ + taskId: "FN-5770-FS", + task: { + column: "in-progress", + }, + }); + fixtures.push(fx); + vi.spyOn(fx.store, "parseFileScopeFromPrompt").mockResolvedValue(["packages/engine/src/**"]); + let mergeChecked = false; + + const driver = new WorkflowAuthoritativeDriver({ + store: { + getSettings: async () => settingsWith({ workflowInterpreterAuthoritative: true, workflowInterpreterDualObserve: true }), + getTask: (taskId) => fx.store.getTask(taskId) as Promise, + getTaskWorkflowSelection: () => undefined, + getWorkflowParitySummary: () => readyParity, + }, + executor: createExecutor({ + planning: async () => ({ outcome: "success" as const }), + execute: async () => ({ outcome: "success" as const }), + review: async () => ({ outcome: "success" as const }), + merge: async () => { + mergeChecked = true; + await assertSquashOverlapsFileScope({ + store: fx.store, + rootDir: fx.rootDir, + taskId: fx.task.id, + task: await fx.store.getTask(fx.task.id) as any, + }); + return { outcome: "success" as const }; + }, + schedule: async () => ({ outcome: "success" as const }), + }), + }); + + return { fx, driver, wasMergeChecked: () => mergeChecked }; + } + + it("trips FileScopeViolationError under interpreter authority for off-scope staged changes", async () => { + const { fx, driver, wasMergeChecked } = await createRealDriverFixture(); + await mkdir(join(fx.rootDir, "packages/core/src"), { recursive: true }); + await writeFile(join(fx.rootDir, "packages/core/src/offscope.txt"), "x\n", "utf-8"); + git(fx.rootDir, "git add packages/core/src/offscope.txt"); + + const result = await driver.maybeRun(fx.task as any); + + expect(result.handled).toBe(true); + expect(result.disposition).toBe("failed"); + expect(result.graphResult?.outcome).toBe("failure"); + expect(wasMergeChecked()).toBe(true); + }); + + it("preserves the squash/merge contract when staged changes stay inside file scope", async () => { + const { fx, driver, wasMergeChecked } = await createRealDriverFixture(); + await mkdir(join(fx.rootDir, "packages/engine/src"), { recursive: true }); + await writeFile(join(fx.rootDir, "packages/engine/src/inscope.txt"), "ok\n", "utf-8"); + git(fx.rootDir, "git add packages/engine/src/inscope.txt"); + + const result = await driver.maybeRun(fx.task as any); + + expect(result.handled).toBe(true); + expect(result.disposition).toBe("completed"); + expect(wasMergeChecked()).toBe(true); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-completion-fanout.test.ts b/packages/engine/src/__tests__/self-healing-completion-fanout.test.ts index faecf0b24b..dfd4b5c1cc 100644 --- a/packages/engine/src/__tests__/self-healing-completion-fanout.test.ts +++ b/packages/engine/src/__tests__/self-healing-completion-fanout.test.ts @@ -6,7 +6,7 @@ const { execMock, existsSyncMock } = vi.hoisted(() => ({ execMock: vi.fn(), existsSyncMock: vi.fn(() => false), })); -vi.mock("node:child_process", () => ({ exec: execMock, execSync: vi.fn() })); +vi.mock("node:child_process", () => ({ exec: execMock, execSync: vi.fn(), execFile: vi.fn() })); vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); diff --git a/packages/engine/src/__tests__/self-healing-ghost-branch-recovery.test.ts b/packages/engine/src/__tests__/self-healing-ghost-branch-recovery.test.ts index 7b2ef2f2dd..b5f6f008d2 100644 --- a/packages/engine/src/__tests__/self-healing-ghost-branch-recovery.test.ts +++ b/packages/engine/src/__tests__/self-healing-ghost-branch-recovery.test.ts @@ -14,7 +14,7 @@ vi.mock("node:child_process", async () => { }; execFn[promisify.custom] = (cmd: string, opts?: any) => execMock(cmd, opts).then((stdout: string) => ({ stdout, stderr: "" })); - return { exec: execFn, execSync: vi.fn() }; + return { exec: execFn, execSync: vi.fn(), execFile: vi.fn() }; }); import { SelfHealingManager } from "../self-healing.js"; diff --git a/packages/engine/src/__tests__/self-healing-orphan-only-scope.test.ts b/packages/engine/src/__tests__/self-healing-orphan-only-scope.test.ts index 39f6c5482e..0c4df8e381 100644 --- a/packages/engine/src/__tests__/self-healing-orphan-only-scope.test.ts +++ b/packages/engine/src/__tests__/self-healing-orphan-only-scope.test.ts @@ -28,7 +28,7 @@ vi.mock("node:child_process", async () => { } }); }); - return { execSync: execSyncFn, exec: execFn }; + return { execSync: execSyncFn, exec: execFn, execFile: vi.fn() }; }); import { EventEmitter } from "node:events"; diff --git a/packages/engine/src/__tests__/self-healing-reclaim-live-zero-commits.test.ts b/packages/engine/src/__tests__/self-healing-reclaim-live-zero-commits.test.ts index a64414cca5..4c407e0f3c 100644 --- a/packages/engine/src/__tests__/self-healing-reclaim-live-zero-commits.test.ts +++ b/packages/engine/src/__tests__/self-healing-reclaim-live-zero-commits.test.ts @@ -14,7 +14,7 @@ vi.mock("node:child_process", async () => { }; execFn[promisify.custom] = (cmd: string, opts?: any) => execMock(cmd, opts).then((stdout: string) => ({ stdout, stderr: "" })); - return { exec: execFn, execSync: vi.fn() }; + return { exec: execFn, execSync: vi.fn(), execFile: vi.fn() }; }); import { SelfHealingManager } from "../self-healing.js"; diff --git a/packages/engine/src/__tests__/self-healing-stale-merger-status.test.ts b/packages/engine/src/__tests__/self-healing-stale-merger-status.test.ts index bf4e2cbcf8..26daa07096 100644 --- a/packages/engine/src/__tests__/self-healing-stale-merger-status.test.ts +++ b/packages/engine/src/__tests__/self-healing-stale-merger-status.test.ts @@ -5,7 +5,7 @@ import type { Settings, Task, TaskStore } from "@fusion/core"; const { execMock } = vi.hoisted(() => ({ execMock: vi.fn(), })); -vi.mock("node:child_process", () => ({ exec: execMock, execSync: vi.fn() })); +vi.mock("node:child_process", () => ({ exec: execMock, execSync: vi.fn(), execFile: vi.fn() })); const { logger } = vi.hoisted(() => ({ logger: { log: vi.fn(), warn: vi.fn(), error: vi.fn() }, diff --git a/packages/engine/src/__tests__/step-session-executor.test.ts b/packages/engine/src/__tests__/step-session-executor.test.ts index 001656622f..c69f488f75 100644 --- a/packages/engine/src/__tests__/step-session-executor.test.ts +++ b/packages/engine/src/__tests__/step-session-executor.test.ts @@ -685,7 +685,7 @@ vi.mock("node:child_process", async () => { } }); }); - return { execSync: execSyncFn, exec: execFn }; + return { execSync: execSyncFn, exec: execFn, execFile: vi.fn() }; }); vi.mock("node:fs", () => ({ existsSync: vi.fn().mockReturnValue(true), diff --git a/packages/engine/src/__tests__/worktree-admin-entry-prune.test.ts b/packages/engine/src/__tests__/worktree-admin-entry-prune.test.ts index 78c21d6d79..882cb1c0cd 100644 --- a/packages/engine/src/__tests__/worktree-admin-entry-prune.test.ts +++ b/packages/engine/src/__tests__/worktree-admin-entry-prune.test.ts @@ -21,7 +21,7 @@ describe("worktree prune wiring", () => { execMock.mockRejectedValueOnce(new Error("create failed")).mockResolvedValueOnce({ stdout: "", stderr: "" }); - vi.doMock("node:child_process", () => ({ exec: execMock })); + vi.doMock("node:child_process", () => ({ exec: execMock, execFile: vi.fn() })); vi.doMock("../worktree-prune.js", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, pruneWorktreeAdminEntries: pruneSpy }; @@ -61,7 +61,7 @@ describe("worktree prune wiring", () => { (execMock as any)[Symbol.for("nodejs.util.promisify.custom")] = execMock; execMock.mockRejectedValueOnce(new Error("create failed")).mockResolvedValueOnce({ stdout: "", stderr: "" }); - vi.doMock("node:child_process", () => ({ exec: execMock })); + vi.doMock("node:child_process", () => ({ exec: execMock, execFile: vi.fn() })); vi.doMock("../worktree-prune.js", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, pruneWorktreeAdminEntries: pruneSpy }; @@ -91,7 +91,7 @@ describe("worktree prune wiring", () => { (execMock as any)[Symbol.for("nodejs.util.promisify.custom")] = execMock; execMock.mockResolvedValueOnce({ stdout: "", stderr: "" }).mockResolvedValueOnce({ stdout: "", stderr: "" }); - vi.doMock("node:child_process", () => ({ exec: execMock })); + vi.doMock("node:child_process", () => ({ exec: execMock, execFile: vi.fn() })); vi.doMock("../worktree-prune.js", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, pruneWorktreeAdminEntries: pruneSpy }; @@ -122,7 +122,7 @@ describe("pruneWorktreeAdminEntries helper", () => { execMock.mockRejectedValue(new Error("boom")); vi.doMock("node:child_process", async () => { const actual = await vi.importActual("node:child_process"); - return { ...actual, exec: execMock }; + return { ...actual, exec: execMock, execFile: vi.fn() }; }); vi.unmock("../worktree-prune.js"); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 66e83d3db2..8d33579a77 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1129,6 +1129,10 @@ export interface TaskExecutorOptions { onStart?: (task: Task, worktreePath: string) => void; onComplete?: (task: Task) => void; onError?: (task: Task, error: Error) => void; + /** Optional runtime-owned dispatch seam that lets a flag-gated workflow + * interpreter own the authoritative lifecycle for default coding tasks. + * Return true when the task was fully handled and legacy execute() should stop. */ + workflowAuthoritativeDispatch?: (task: Task) => Promise; onAgentText?: (taskId: string, delta: string) => void; onAgentTool?: (taskId: string, toolName: string) => void; autoRecoveryDispatcher?: AutoRecoveryDispatcher; @@ -3704,7 +3708,7 @@ export class TaskExecutor { const runner = new WorkflowGraphTaskRunner({ store: this.store, runId: resolvedRunId, - seams: this.createGraphSeams(settings), + seams: this.createAuthoritativeWorkflowSeams(settings), runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings, resolveBindingForNode(node.id)), onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`), @@ -4543,8 +4547,9 @@ export class TaskExecutor { return only; } - /** Seam implementations delegating to the legacy engine (KTD-1: delegate, never reimplement). */ - private createGraphSeams(_settings: Settings): WorkflowLegacySeams { + /** Public authoritative-driver seam factory: exposes the same real lifecycle + * seams the internal graph runner uses, without changing legacy behavior. */ + public createAuthoritativeWorkflowSeams(_settings: Settings): WorkflowLegacySeams { return { // Built-in triage/spec generation runs upstream of the interpreter today, // so planning is a no-op for already-specified tasks. Custom planning @@ -5568,6 +5573,8 @@ export class TaskExecutor { executorLog.log(`execute() called for ${task.id} while graph routing is active — skipping duplicate`); return; } + const authoritativeOwned = await this.options.workflowAuthoritativeDispatch?.(task); + if (authoritativeOwned) return; const graphOwned = await this.maybeExecuteWorkflowGraph(task); if (graphOwned) return; } diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 8f5642569c..1110472ac9 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -201,6 +201,12 @@ export { type WorkflowParityObserverLegacyRunResult, type WorkflowParityObserverShadowRunResult, } from "./workflow-parity-observer.js"; +export { + WorkflowAuthoritativeDriver, + type WorkflowAuthoritativeDriverDeps, + type WorkflowAuthoritativeDriverResult, + type WorkflowAuthoritativeDriverStore, +} from "./workflow-authoritative-driver.js"; export { auditSquashMerge, formatSquashAuditReport, diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 19f14f163c..c83aa10650 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -17,6 +17,7 @@ import { Scheduler } from "../scheduler.js"; import type { PrMonitor, PrComment } from "../pr-monitor.js"; import type { PrInfo } from "@fusion/core"; import { TaskExecutor, type TaskExecutorOptions } from "../executor.js"; +import { WorkflowAuthoritativeDriver } from "../workflow-authoritative-driver.js"; import { buildPrNodeDeps } from "../pr-nodes.js"; import { isExperimentalFeatureEnabled } from "@fusion/core"; import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "../cli-agent/runtime.js"; @@ -474,6 +475,7 @@ export class InProcessRuntime } const prNodeGithubOps = this.config.prNodeGithubOps; + const workflowAuthoritativeDriverRef: { current?: WorkflowAuthoritativeDriver } = {}; const executorOptions: TaskExecutorOptions = { semaphore: this.globalSemaphore, pool: this.worktreePool, @@ -494,6 +496,8 @@ export class InProcessRuntime onSliceComplete: (slice) => { void this.scheduler.onSliceComplete(slice); }, + workflowAuthoritativeDispatch: async (task) => + (await workflowAuthoritativeDriverRef.current?.maybeRun(task))?.handled ?? false, onStart: (task, worktreePath) => { this.recordActivity(); runtimeLog.log(`Started executing task ${task.id} in ${worktreePath}`); @@ -536,6 +540,10 @@ export class InProcessRuntime this.config.workingDirectory, executorOptions ); + workflowAuthoritativeDriverRef.current = new WorkflowAuthoritativeDriver({ + store: this.taskStore, + executor: this.executor, + }); if (this.mergeRequester) { this.executor.setMergeRequester(this.mergeRequester); } diff --git a/packages/engine/src/workflow-authoritative-driver.ts b/packages/engine/src/workflow-authoritative-driver.ts new file mode 100644 index 0000000000..77c115981f --- /dev/null +++ b/packages/engine/src/workflow-authoritative-driver.ts @@ -0,0 +1,154 @@ +import { + BUILTIN_CODING_WORKFLOW_IR, + WORKFLOW_INTERPRETER_AUTHORITATIVE_FLAG, + evaluateInterpreterCutoverReadiness, + isExperimentalFeatureEnabled, + type Settings, + type Task, + type TaskDetail, + type WorkflowDefinition, + type WorkflowParitySummary, +} from "@fusion/core"; + +import type { TaskExecutor } from "./executor.js"; +import { executorLog } from "./logger.js"; +import { WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js"; +import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js"; + +const AUTHORITATIVE_WORKFLOW_ID = "builtin:workflow-interpreter-authoritative"; + +export interface WorkflowAuthoritativeDriverStore { + getSettings(): Promise; + getTask(taskId: string): Promise; + getTaskWorkflowSelection?(taskId: string): { workflowId: string; stepIds: string[] } | undefined; + getWorkflowParitySummary?(options?: { since?: string; limit?: number }): WorkflowParitySummary; +} + +export interface WorkflowAuthoritativeDriverDeps { + store: WorkflowAuthoritativeDriverStore; + executor: Pick; + minimumObservedRuns?: number; +} + +export interface WorkflowAuthoritativeDriverResult { + handled: boolean; + disposition: "completed" | "failed" | "fell-back"; + reason?: string; + readinessReasons: string[]; + graphResult?: WorkflowGraphTaskRunResult; +} + +function buildAuthoritativeSettings(settings: Settings): Settings { + return { + ...settings, + experimentalFeatures: { + ...(settings.experimentalFeatures ?? {}), + workflowGraphExecutor: true, + [WORKFLOW_INTERPRETER_AUTHORITATIVE_FLAG]: true, + }, + }; +} + +export class WorkflowAuthoritativeDriver { + public constructor(private readonly deps: WorkflowAuthoritativeDriverDeps) {} + + public async maybeRun(task: Task): Promise { + let settings: Settings; + let paritySummary: WorkflowParitySummary | undefined; + try { + settings = await this.deps.store.getSettings(); + paritySummary = this.deps.store.getWorkflowParitySummary?.(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + executorLog.warn(`[workflow-authoritative] ${task.id}: readiness probe failed — falling back to legacy (${message})`); + return { + handled: false, + disposition: "fell-back", + reason: `store-unavailable: ${message}`, + readinessReasons: ["workflow parity readiness probe unavailable"], + }; + } + const authoritativeFlagEnabled = isExperimentalFeatureEnabled( + settings, + WORKFLOW_INTERPRETER_AUTHORITATIVE_FLAG, + ); + const dualObserveEnabled = isExperimentalFeatureEnabled( + settings, + WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG, + ); + const readiness = evaluateInterpreterCutoverReadiness({ + authoritativeFlagEnabled, + dualObserveEnabled, + paritySummary, + minimumObservedRuns: this.deps.minimumObservedRuns, + }); + if (!readiness.ready) { + return { + handled: false, + disposition: "fell-back", + reason: readiness.reasons.join("; "), + readinessReasons: readiness.reasons, + }; + } + + const existingSelection = this.deps.store.getTaskWorkflowSelection?.(task.id); + if (existingSelection) { + return { + handled: false, + disposition: "fell-back", + reason: `workflow selection already present (${existingSelection.workflowId})`, + readinessReasons: [], + }; + } + + let liveTask: TaskDetail; + try { + liveTask = await this.deps.store.getTask(task.id); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + executorLog.warn(`[workflow-authoritative] ${task.id}: failed to load live task — falling back to legacy (${message})`); + return { + handled: false, + disposition: "fell-back", + reason: `task-load-failed: ${message}`, + readinessReasons: [], + }; + } + const runner = new WorkflowGraphTaskRunner({ + store: { + getTaskWorkflowSelection: () => ({ workflowId: AUTHORITATIVE_WORKFLOW_ID, stepIds: [] }), + getWorkflowDefinition: async () => ({ + id: AUTHORITATIVE_WORKFLOW_ID, + name: "Workflow interpreter authoritative cutover", + ir: BUILTIN_CODING_WORKFLOW_IR, + } satisfies Pick as WorkflowDefinition), + }, + seams: this.deps.executor.createAuthoritativeWorkflowSeams(settings), + runCustomNode: async (node) => { + throw new Error(`unexpected custom node in builtin authoritative workflow: ${node.id}`); + }, + onEvent: (event) => { + executorLog.log(`[workflow-authoritative] ${event.type} ${event.taskId}: ${event.detail}`); + }, + }); + + const graphResult = await runner.run(liveTask, buildAuthoritativeSettings(settings)); + if (graphResult.disposition === "fell-back") { + return { + handled: false, + disposition: "fell-back", + reason: graphResult.reason, + readinessReasons: [], + graphResult, + }; + } + + return { + handled: true, + disposition: graphResult.disposition, + reason: graphResult.reason, + readinessReasons: [], + graphResult, + }; + } +}