FN-5768: add dual-observe workflow parity instrumentation

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

.changeset/fn-5768-workflow-parity-observer.md

@@ -0,0 +1,9 @@
+---
+"@runfusion/fusion": minor
+---
+
+Add workflow interpreter dual-observe parity instrumentation surfaces for phased rollout.
+
+- Export pure workflow parity comparison helpers from `@fusion/core` (`compareWorkflowRunObservations`, `compareWorkflowRunAudits`) with structured drift reports.
+- Add `observeWorkflowParity` in the engine as a default-OFF, fail-soft observer gated by `experimentalFeatures.workflowInterpreterDualObserve`.
+- Emit run-audit parity events (`workflow:parity-observed`, `workflow:parity-drift`) for shadow agreement/drift visibility without changing authoritative legacy execution.

docs/workflow-steps.md

@@ -310,6 +310,24 @@ For pre-merge workflow hard failures, executor behavior is (gate-mode steps):
 
 Tasks are not parked in `in-review` for this remediable path unless additional terminal failures occur.
 
+## Workflow Interpreter Dual-Observe (parity instrumentation)
+
+Fusion now includes a **default-OFF** experimental parity seam for the workflow interpreter rollout.
+
+- **Flag:** `experimentalFeatures.workflowInterpreterDualObserve`
+- **Mode:** observe-only shadow run; legacy executor/reviewer/merger/scheduler path remains authoritative
+- **Behavior when OFF (default):** strict no-op (no shadow run, no parity audit records)
+- **Behavior when ON:** compare legacy and interpreter observations plus comparable run-audit slices
+
+Run-audit events emitted in `database` domain:
+
+- `workflow:parity-observed` — always emitted for an enabled parity check with `metadata.agree`
+- `workflow:parity-drift` — emitted when parity differs (or shadow execution fails), carrying `metadata.diffs`
+
+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.
+
 #### Self-healing recovery for parked review tasks
 
 If a task is found in `in-review` with failed pre-merge workflow results and no active executor, self-healing can auto-revive it (bounded by `maxPostReviewFixes`) by replaying the same remediation send-back flow.

packages/core/src/__tests__/workflow-parity.test.ts

@@ -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" }),
+      ]),
+    );
+  });
+});

packages/core/src/index.ts

@@ -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";

packages/core/src/workflow-parity.ts

@@ -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,
+  };
+}

packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-dual-observe.test.ts

@@ -0,0 +1,144 @@
+import { describe, expect, it, vi } from "vitest";
+import {
+  observeWorkflowParity,
+  WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG,
+} from "../../workflow-parity-observer.js";
+import type { WorkflowRunObservation } from "@fusion/core";
+
+const baseObservation: WorkflowRunObservation = {
+  stageTransitions: ["triage", "execute", "review", "merge"],
+  terminalColumn: "done",
+  terminalStatus: null,
+  reviewVerdict: "APPROVE",
+  mergeOutcome: "merged",
+  invariants: {
+    fileScopeGuardOutcome: "pass",
+    squashMergeContractOutcome: "pass",
+    autoMergeTerminalUntilMergedRespected: true,
+    moveTaskHardCancelRespected: true,
+  },
+};
+
+describe("FN-5768 workflow interpreter dual-observe", () => {
+  it("is strict no-op when flag is off", async () => {
+    const recordRunAuditEvent = vi.fn();
+    const runShadow = vi.fn();
+
+    await observeWorkflowParity({
+      settings: { experimentalFeatures: {} },
+      store: { recordRunAuditEvent },
+      agentId: "executor",
+      legacy: {
+        taskId: "FN-1",
+        observation: baseObservation,
+        auditEvents: [],
+      },
+      runShadow,
+    });
+
+    expect(runShadow).not.toHaveBeenCalled();
+    expect(recordRunAuditEvent).not.toHaveBeenCalled();
+  });
+
+  it("records parity-observed agree=true when observations match", async () => {
+    const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
+
+    await observeWorkflowParity({
+      settings: { experimentalFeatures: { [WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG]: true } },
+      store: { recordRunAuditEvent },
+      agentId: "executor",
+      legacy: {
+        taskId: "FN-2",
+        observation: baseObservation,
+        auditEvents: [],
+      },
+      runShadow: async () => ({ observation: baseObservation, auditEvents: [] }),
+    });
+
+    expect(recordRunAuditEvent).toHaveBeenCalledTimes(1);
+    expect(recordRunAuditEvent).toHaveBeenCalledWith(
+      expect.objectContaining({
+        mutationType: "workflow:parity-observed",
+        metadata: expect.objectContaining({ agree: true }),
+      }),
+    );
+  });
+
+  it("records parity drift and keeps authoritative result unchanged", async () => {
+    const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
+    const legacyResult = { authoritative: true };
+
+    await observeWorkflowParity({
+      settings: { experimentalFeatures: { [WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG]: true } },
+      store: { recordRunAuditEvent },
+      agentId: "executor",
+      legacy: {
+        taskId: "FN-3",
+        observation: baseObservation,
+        auditEvents: [],
+      },
+      runShadow: async () => ({
+        observation: {
+          ...baseObservation,
+          terminalColumn: "in-review",
+        },
+        auditEvents: [],
+      }),
+    });
+
+    expect(legacyResult).toEqual({ authoritative: true });
+    expect(recordRunAuditEvent).toHaveBeenCalledTimes(2);
+    expect(recordRunAuditEvent).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ mutationType: "workflow:parity-observed" }),
+    );
+    expect(recordRunAuditEvent).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({
+        mutationType: "workflow:parity-drift",
+        metadata: expect.objectContaining({
+          agree: false,
+          diffs: expect.arrayContaining([expect.objectContaining({ field: "terminalColumn" })]),
+        }),
+      }),
+    );
+  });
+
+  it("captures shadow errors fail-soft without rethrow", async () => {
+    const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
+
+    await expect(
+      observeWorkflowParity({
+        settings: { experimentalFeatures: { [WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG]: true } },
+        store: { recordRunAuditEvent },
+        agentId: "executor",
+        legacy: {
+          taskId: "FN-4",
+          observation: baseObservation,
+          auditEvents: [],
+        },
+        runShadow: async () => {
+          throw new Error("shadow exploded");
+        },
+      }),
+    ).resolves.toBeUndefined();
+
+    expect(recordRunAuditEvent).toHaveBeenCalledTimes(2);
+    expect(recordRunAuditEvent).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({
+        mutationType: "workflow:parity-observed",
+        metadata: expect.objectContaining({ agree: false }),
+      }),
+    );
+    expect(recordRunAuditEvent).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({
+        mutationType: "workflow:parity-drift",
+        metadata: expect.objectContaining({
+          diffs: expect.arrayContaining([expect.objectContaining({ field: "shadow.error" })]),
+        }),
+      }),
+    );
+  });
+});

packages/engine/src/index.ts

@@ -84,6 +84,13 @@ export {
 export {
   generateSyntheticRunId,
 } from "./run-audit.js";
+export {
+  observeWorkflowParity,
+  WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG,
+  type WorkflowParityObserverInput,
+  type WorkflowParityObserverLegacyRunResult,
+  type WorkflowParityObserverShadowRunResult,
+} from "./workflow-parity-observer.js";
 export {
   auditSquashMerge,
   formatSquashAuditReport,

packages/engine/src/workflow-parity-observer.ts

@@ -0,0 +1,116 @@
+import {
+  compareWorkflowRunAudits,
+  compareWorkflowRunObservations,
+  isExperimentalFeatureEnabled,
+  WORKFLOW_PARITY_DRIFT_MUTATION,
+  WORKFLOW_PARITY_OBSERVED_MUTATION,
+  type RunAuditEvent,
+  type Settings,
+  type TaskStore,
+  type WorkflowParityDiff,
+  type WorkflowRunObservation,
+} from "@fusion/core";
+import { generateSyntheticRunId } from "./run-audit.js";
+
+export const WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG = "workflowInterpreterDualObserve" as const;
+
+export interface WorkflowParityObserverLegacyRunResult {
+  taskId: string;
+  observation: WorkflowRunObservation;
+  auditEvents: RunAuditEvent[];
+}
+
+export interface WorkflowParityObserverShadowRunResult {
+  observation: WorkflowRunObservation;
+  auditEvents: RunAuditEvent[];
+}
+
+export interface WorkflowParityObserverInput {
+  settings: Pick<Settings, "experimentalFeatures"> | undefined;
+  store: Pick<TaskStore, "recordRunAuditEvent">;
+  agentId: string;
+  legacy: WorkflowParityObserverLegacyRunResult;
+  runShadow: () => Promise<WorkflowParityObserverShadowRunResult>;
+}
+
+function buildErrorDiff(error: unknown): WorkflowParityDiff {
+  return {
+    field: "shadow.error",
+    legacy: null,
+    interpreter: error instanceof Error ? error.message : String(error),
+    category: "audit",
+    severity: "error",
+  };
+}
+
+/**
+ * Observe-only parity seam. Never mutates/blocks authoritative legacy behavior.
+ */
+export async function observeWorkflowParity(input: WorkflowParityObserverInput): Promise<void> {
+  if (!isExperimentalFeatureEnabled(input.settings, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG)) {
+    return;
+  }
+
+  const { store, agentId, legacy } = input;
+  const runId = generateSyntheticRunId("workflow-shadow", legacy.taskId);
+
+  try {
+    const shadow = await input.runShadow();
+    const observationReport = compareWorkflowRunObservations(legacy.observation, shadow.observation);
+    const auditReport = compareWorkflowRunAudits(legacy.auditEvents, shadow.auditEvents);
+    const diffs = [...observationReport.diffs, ...auditReport.diffs];
+    const agree = diffs.length === 0;
+
+    await store.recordRunAuditEvent?.({
+      taskId: legacy.taskId,
+      agentId,
+      runId,
+      domain: "database",
+      mutationType: WORKFLOW_PARITY_OBSERVED_MUTATION,
+      target: legacy.taskId,
+      metadata: {
+        agree,
+      },
+    });
+
+    if (!agree) {
+      await store.recordRunAuditEvent?.({
+        taskId: legacy.taskId,
+        agentId,
+        runId,
+        domain: "database",
+        mutationType: WORKFLOW_PARITY_DRIFT_MUTATION,
+        target: legacy.taskId,
+        metadata: {
+          agree,
+          diffs,
+        },
+      });
+    }
+  } catch (error) {
+    await store.recordRunAuditEvent?.({
+      taskId: legacy.taskId,
+      agentId,
+      runId,
+      domain: "database",
+      mutationType: WORKFLOW_PARITY_OBSERVED_MUTATION,
+      target: legacy.taskId,
+      metadata: {
+        agree: false,
+      },
+    });
+
+    await store.recordRunAuditEvent?.({
+      taskId: legacy.taskId,
+      agentId,
+      runId,
+      domain: "database",
+      mutationType: WORKFLOW_PARITY_DRIFT_MUTATION,
+      target: legacy.taskId,
+      metadata: {
+        agree: false,
+        diffs: [buildErrorDiff(error)],
+      },
+    });
+  }
+}