root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-5768: add dual-observe workflow parity instrumentation

Browse Source 
Add default-off workflow interpreter dual-observe parity instrumentation across core and engine.

- add workflow parity comparison primitives and exports in @fusion/core
- add engine parity observer seam with experimental flag and public exports
- add regression coverage for parity contracts and dual-observe reliability behavior
- document the dual-observe parity mode, events, and contract in workflow docs
- add a patch changeset for @runfusion/fusion release notes

Files changed:
 .changeset/fn-5768-workflow-parity-observer.md     |   9 +
 docs/workflow-steps.md                             |  18 ++
 .../core/src/__tests__/workflow-parity.test.ts     | 143 ++++++++++++++
 packages/core/src/index.ts                         |  18 ++
 packages/core/src/workflow-parity.ts               | 205 +++++++++++++++++++++
 .../workflow-interpreter-dual-observe.test.ts      | 144 +++++++++++++++
 packages/engine/src/index.ts                       |   7 +
 packages/engine/src/workflow-parity-observer.ts    | 116 ++++++++++++
 8 files changed, 660 insertions(+)

Fusion-Task-Id: FN-5768

Fusion-Task-Lineage: fb4f8111-f48d-482c-b1d9-9106c43c829a
This commit is contained in:
gsxdsm
2026-05-31 07:55:18 -07:00
parent 5c33ab133e
commit 5b4eecb5d4
8 changed files with 660 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
packages/core/src/__tests__/workflow-parity.test.ts Normal file
View File

@@ -0,0 +1,143 @@
import { describe, expect, it } from "vitest";
import {
compareWorkflowRunAudits,
compareWorkflowRunObservations,
type RunAuditEvent,
type WorkflowRunObservation,
} from "../index.js";
function observation(overrides: Partial<WorkflowRunObservation> = {}): WorkflowRunObservation {
return {
stageTransitions: ["triage", "execute", "review", "merge"],
terminalColumn: "done",
terminalStatus: null,
reviewVerdict: "APPROVE",
mergeOutcome: "merged",
invariants: {
fileScopeGuardOutcome: "pass",
squashMergeContractOutcome: "pass",
autoMergeTerminalUntilMergedRespected: true,
moveTaskHardCancelRespected: true,
},
...overrides,
};
}
function auditEvent(mutationType: string, target: string, phase: string): RunAuditEvent {
return {
id: `${mutationType}-${target}`,
timestamp: new Date().toISOString(),
taskId: "FN-1",
agentId: "executor",
runId: "run-1",
domain: "database",
mutationType,
target,
metadata: { phase },
};
}
describe("workflow parity", () => {
it("agrees for identical observations", () => {
const report = compareWorkflowRunObservations(observation(), observation());
expect(report).toEqual({ agree: true, diffs: [] });
});
it("reports lifecycle transition drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ stageTransitions: ["triage", "execute", "merge"] }),
);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([expect.objectContaining({ field: "stageTransitions", category: "lifecycle" })]),
);
});
it("reports terminal status and review verdict drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ terminalColumn: "in-review", reviewVerdict: "REVISE" }),
);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "terminalColumn" }),
expect.objectContaining({ field: "reviewVerdict" }),
]),
);
});
it("reports file-scope guard invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ invariants: { ...observation().invariants, fileScopeGuardOutcome: "fail" } }),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.fileScopeGuardOutcome", category: "invariant" }),
]),
);
});
it("reports squash merge invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ invariants: { ...observation().invariants, squashMergeContractOutcome: "blocked" } }),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.squashMergeContractOutcome", category: "invariant" }),
]),
);
});
it("reports auto-merge terminal invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({
invariants: { ...observation().invariants, autoMergeTerminalUntilMergedRespected: false },
}),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.autoMergeTerminalUntilMergedRespected", category: "invariant" }),
]),
);
});
it("reports moveTask hard-cancel invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ invariants: { ...observation().invariants, moveTaskHardCancelRespected: false } }),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.moveTaskHardCancelRespected", category: "invariant" }),
]),
);
});
it("agrees on identical comparable run-audit slices", () => {
const events = [
auditEvent("task:move", "FN-1", "execute"),
auditEvent("task:update", "FN-1", "review"),
];
const report = compareWorkflowRunAudits(events, events);
expect(report).toEqual({ agree: true, diffs: [] });
});
it("reports run-audit drift", () => {
const legacy = [auditEvent("task:move", "FN-1", "execute")];
const interpreter = [auditEvent("task:update", "FN-2", "review")];
const report = compareWorkflowRunAudits(legacy, interpreter);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "audit[0].mutationType", category: "audit" }),
expect.objectContaining({ field: "audit[0].target", category: "audit" }),
expect.objectContaining({ field: "audit[0].phase", category: "audit" }),
]),
);
});
});

18
packages/core/src/index.ts
View File

@@ -1076,6 +1076,24 @@ export type {
} from "./research-types.js";
export { isExperimentalFeatureEnabled } from "./experimental-features.js";
export {
WORKFLOW_COMPARABLE_AUDIT_MUTATIONS,
WORKFLOW_PARITY_OBSERVED_MUTATION,
WORKFLOW_PARITY_DRIFT_MUTATION,
compareWorkflowRunAudits,
compareWorkflowRunObservations,
extractWorkflowAuditObservations,
} from "./workflow-parity.js";
export type {
WorkflowAuditObservation,
WorkflowParityDiff,
WorkflowParityDiffCategory,
WorkflowParityDiffSeverity,
WorkflowParityDriftReport,
WorkflowReliabilityInvariantSignals,
WorkflowRunObservation,
WorkflowStage,
} from "./workflow-parity.js";
export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js";
export type { ResolvedResearchSettings } from "./research-settings.js";
export { isEvalsExperimentalEnabled, resolveEvalSettings } from "./eval-settings.js";

205
packages/core/src/workflow-parity.ts Normal file
View File

@@ -0,0 +1,205 @@
import type { RunAuditEvent } from "./types.js";
export const WORKFLOW_PARITY_OBSERVED_MUTATION = "workflow:parity-observed" as const;
export const WORKFLOW_PARITY_DRIFT_MUTATION = "workflow:parity-drift" as const;
export type WorkflowStage = "triage" | "execute" | "review" | "merge";
export type WorkflowParityDiffCategory = "lifecycle" | "invariant" | "audit";
export type WorkflowParityDiffSeverity = "info" | "warning" | "error";
export interface WorkflowReliabilityInvariantSignals {
fileScopeGuardOutcome: string | null;
squashMergeContractOutcome: string | null;
autoMergeTerminalUntilMergedRespected: boolean;
moveTaskHardCancelRespected: boolean;
}
/**
* Observe-only workflow snapshot used for parity checks.
* Legacy remains authoritative; interpreter observations are advisory diagnostics only.
*/
export interface WorkflowRunObservation {
stageTransitions: WorkflowStage[];
terminalColumn: string | null;
terminalStatus: string | null;
reviewVerdict: string | null;
mergeOutcome: string | null;
invariants: WorkflowReliabilityInvariantSignals;
}
export interface WorkflowParityDiff {
field: string;
legacy: unknown;
interpreter: unknown;
category: WorkflowParityDiffCategory;
severity: WorkflowParityDiffSeverity;
}
export interface WorkflowParityDriftReport {
agree: boolean;
diffs: WorkflowParityDiff[];
}
function isEqualScalarArray(left: readonly string[], right: readonly string[]): boolean {
if (left.length !== right.length) return false;
return left.every((value, index) => value === right[index]);
}
function pushDiff(
diffs: WorkflowParityDiff[],
field: string,
legacy: unknown,
interpreter: unknown,
category: WorkflowParityDiffCategory,
severity: WorkflowParityDiffSeverity = "warning",
): void {
diffs.push({ field, legacy, interpreter, category, severity });
}
/**
* Pure observation comparison contract for dual-observe shadow checks.
* Legacy observation is authoritative; interpreter drift is diagnostics only.
*/
export function compareWorkflowRunObservations(
legacy: WorkflowRunObservation,
interpreter: WorkflowRunObservation,
): WorkflowParityDriftReport {
const diffs: WorkflowParityDiff[] = [];
if (!isEqualScalarArray(legacy.stageTransitions, interpreter.stageTransitions)) {
pushDiff(
diffs,
"stageTransitions",
legacy.stageTransitions,
interpreter.stageTransitions,
"lifecycle",
"error",
);
}
const lifecycleChecks: Array<[field: string, legacyValue: unknown, interpreterValue: unknown]> = [
["terminalColumn", legacy.terminalColumn, interpreter.terminalColumn],
["terminalStatus", legacy.terminalStatus, interpreter.terminalStatus],
["reviewVerdict", legacy.reviewVerdict, interpreter.reviewVerdict],
["mergeOutcome", legacy.mergeOutcome, interpreter.mergeOutcome],
];
for (const [field, legacyValue, interpreterValue] of lifecycleChecks) {
if (legacyValue !== interpreterValue) {
pushDiff(diffs, field, legacyValue, interpreterValue, "lifecycle", "error");
}
}
const invariantChecks: Array<[field: string, legacyValue: unknown, interpreterValue: unknown]> = [
[
"invariants.fileScopeGuardOutcome",
legacy.invariants.fileScopeGuardOutcome,
interpreter.invariants.fileScopeGuardOutcome,
],
[
"invariants.squashMergeContractOutcome",
legacy.invariants.squashMergeContractOutcome,
interpreter.invariants.squashMergeContractOutcome,
],
[
"invariants.autoMergeTerminalUntilMergedRespected",
legacy.invariants.autoMergeTerminalUntilMergedRespected,
interpreter.invariants.autoMergeTerminalUntilMergedRespected,
],
[
"invariants.moveTaskHardCancelRespected",
legacy.invariants.moveTaskHardCancelRespected,
interpreter.invariants.moveTaskHardCancelRespected,
],
];
for (const [field, legacyValue, interpreterValue] of invariantChecks) {
if (legacyValue !== interpreterValue) {
pushDiff(diffs, field, legacyValue, interpreterValue, "invariant", "error");
}
}
return {
agree: diffs.length === 0,
diffs,
};
}
export const WORKFLOW_COMPARABLE_AUDIT_MUTATIONS = [
"task:move",
"task:update",
"task:pause",
"task:unpause",
"task:dependency:add",
"merge:request-enqueued",
"merge:dependency-parity-diff",
"merge:lease-parity-diff",
] as const;
const WORKFLOW_COMPARABLE_AUDIT_MUTATION_SET = new Set<string>(WORKFLOW_COMPARABLE_AUDIT_MUTATIONS);
export interface WorkflowAuditObservation {
mutationType: string;
target: string;
phase: string | null;
}
export function extractWorkflowAuditObservations(events: readonly RunAuditEvent[]): WorkflowAuditObservation[] {
return events
.filter(
(event) =>
event.domain === "database"
&& WORKFLOW_COMPARABLE_AUDIT_MUTATION_SET.has(String(event.mutationType)),
)
.map((event) => ({
mutationType: String(event.mutationType),
target: event.target,
phase: typeof event.metadata?.phase === "string" ? event.metadata.phase : null,
}));
}
export function compareWorkflowRunAudits(
legacyEvents: readonly RunAuditEvent[],
interpreterEvents: readonly RunAuditEvent[],
): WorkflowParityDriftReport {
const legacy = extractWorkflowAuditObservations(legacyEvents);
const interpreter = extractWorkflowAuditObservations(interpreterEvents);
const diffs: WorkflowParityDiff[] = [];
if (legacy.length !== interpreter.length) {
pushDiff(diffs, "audit.length", legacy.length, interpreter.length, "audit");
}
const count = Math.max(legacy.length, interpreter.length);
for (let index = 0; index < count; index += 1) {
const left = legacy[index];
const right = interpreter[index];
if (!left || !right) {
pushDiff(diffs, `audit[${index}]`, left ?? null, right ?? null, "audit");
continue;
}
if (left.mutationType !== right.mutationType) {
pushDiff(
diffs,
`audit[${index}].mutationType`,
left.mutationType,
right.mutationType,
"audit",
);
}
if (left.target !== right.target) {
pushDiff(diffs, `audit[${index}].target`, left.target, right.target, "audit");
}
if (left.phase !== right.phase) {
pushDiff(diffs, `audit[${index}].phase`, left.phase, right.phase, "audit");
}
}
return {
agree: diffs.length === 0,
diffs,
};
}