FN-5770: promote workflow interpreter to guarded lifecycle driver
Route default coding tasks through the workflow interpreter only when parity readiness checks pass. - add a core cutover-readiness evaluator and exports for the workflowInterpreterAuthoritative flag - wire a WorkflowAuthoritativeDriver into runtime dispatch and reuse authoritative lifecycle seams with legacy fallback behavior - cover the cutover with reliability tests, docs updates, and a published CLI changeset Files changed: .../FN-5770-workflow-interpreter-authoritative.md | 5 + docs/architecture.md | 1 + docs/settings-reference.md | 5 +- docs/workflow-steps.md | 22 +- .../core/src/__tests__/workflow-cutover.test.ts | 68 +++++ packages/core/src/index.ts | 8 + packages/core/src/workflow-cutover.ts | 81 ++++++ .../src/__tests__/openclaw-runtime-e2e.test.ts | 1 + .../engine/src/__tests__/pi-layers-wiring.test.ts | 2 +- .../branch-recovery-live-zero-commits.test.ts | 2 +- .../branch-recovery-stale-cached-base.test.ts | 2 +- .../workflow-interpreter-cutover.test.ts | 306 +++++++++++++++++++++ .../self-healing-completion-fanout.test.ts | 2 +- .../self-healing-ghost-branch-recovery.test.ts | 2 +- .../self-healing-orphan-only-scope.test.ts | 2 +- .../self-healing-reclaim-live-zero-commits.test.ts | 2 +- .../self-healing-stale-merger-status.test.ts | 2 +- .../src/__tests__/step-session-executor.test.ts | 2 +- .../__tests__/worktree-admin-entry-prune.test.ts | 8 +- packages/engine/src/executor.ts | 13 +- packages/engine/src/index.ts | 6 + packages/engine/src/runtimes/in-process-runtime.ts | 8 + .../engine/src/workflow-authoritative-driver.ts | 154 +++++++++++ 23 files changed, 686 insertions(+), 18 deletions(-) Fusion-Task-Id: FN-5770 Fusion-Task-Lineage: dee668b5-10e0-4a3c-b025-9cc43a26286a
This commit is contained in:
5
.changeset/FN-5770-workflow-interpreter-authoritative.md
Normal file
5
.changeset/FN-5770-workflow-interpreter-authoritative.md
Normal file
@@ -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.
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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 |
|
||||
|
||||
68
packages/core/src/__tests__/workflow-cutover.test.ts
Normal file
68
packages/core/src/__tests__/workflow-cutover.test.ts
Normal file
@@ -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)",
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -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";
|
||||
|
||||
81
packages/core/src/workflow-cutover.ts
Normal file
81
packages/core/src/workflow-cutover.ts
Normal file
@@ -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<WorkflowParitySummary, "observed" | "drift" | "recentDrift"> | null;
|
||||
/** Optional unresolved drift reports surfaced directly by the caller. */
|
||||
unresolvedDriftReports?: readonly Pick<WorkflowParityDriftReport, "agree" | "diffs">[] | 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<WorkflowParityDriftReport, "agree" | "diffs">[] | 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,
|
||||
};
|
||||
}
|
||||
@@ -42,6 +42,7 @@ vi.mock("../pi.js", () => ({
|
||||
}));
|
||||
|
||||
vi.mock("node:child_process", () => ({
|
||||
execFile: vi.fn(),
|
||||
spawn: (...args: unknown[]) => mockSpawn(...args),
|
||||
}));
|
||||
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<string, boolean>, overrides: Partial<Settings> = {}): 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<Awaited<ReturnType<typeof makeReliabilityFixture>>> = [];
|
||||
|
||||
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<TaskDetail>,
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<typeof import("node:fs")>();
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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() },
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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<typeof import("../worktree-prune.js")>();
|
||||
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<typeof import("../worktree-prune.js")>();
|
||||
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<typeof import("../worktree-prune.js")>();
|
||||
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<typeof import("node:child_process")>("node:child_process");
|
||||
return { ...actual, exec: execMock };
|
||||
return { ...actual, exec: execMock, execFile: vi.fn() };
|
||||
});
|
||||
|
||||
vi.unmock("../worktree-prune.js");
|
||||
|
||||
@@ -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<boolean>;
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
154
packages/engine/src/workflow-authoritative-driver.ts
Normal file
154
packages/engine/src/workflow-authoritative-driver.ts
Normal file
@@ -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<Settings>;
|
||||
getTask(taskId: string): Promise<TaskDetail>;
|
||||
getTaskWorkflowSelection?(taskId: string): { workflowId: string; stepIds: string[] } | undefined;
|
||||
getWorkflowParitySummary?(options?: { since?: string; limit?: number }): WorkflowParitySummary;
|
||||
}
|
||||
|
||||
export interface WorkflowAuthoritativeDriverDeps {
|
||||
store: WorkflowAuthoritativeDriverStore;
|
||||
executor: Pick<TaskExecutor, "createAuthoritativeWorkflowSeams">;
|
||||
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<WorkflowAuthoritativeDriverResult> {
|
||||
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<WorkflowDefinition, "id" | "name" | "ir"> 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,
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user