feat: surface dual-observe flag + parity summary (CU-U5 #3)

- Settings → Experimental gains a 'Workflow Graph Engine — dual-observe parity
  (diagnostic)' toggle for the workflowInterpreterDualObserve flag.
- store.getWorkflowParitySummary() aggregates the workflow:parity-observed /
  workflow:parity-drift run-audit events into the graduation signal: agree-rate,
  per-field drift counts, and recent drift samples. Covered by
  workflow-parity-summary.test.ts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

@@ -0,0 +1,69 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import {
+  WORKFLOW_PARITY_OBSERVED_MUTATION,
+  WORKFLOW_PARITY_DRIFT_MUTATION,
+} from "../workflow-parity.js";
+import { createTaskStoreTestHarness } from "./store-test-helpers.js";
+
+describe("getWorkflowParitySummary (CU-U5)", () => {
+  const harness = createTaskStoreTestHarness();
+  let store: ReturnType<typeof harness.store>;
+
+  beforeEach(async () => {
+    await harness.beforeEach();
+    store = harness.store();
+  });
+  afterEach(async () => {
+    await harness.afterEach();
+  });
+
+  function observe(taskId: string, agree: boolean, diffs?: unknown[]): void {
+    store.recordRunAuditEvent({
+      taskId,
+      agentId: "store",
+      runId: `parity:${taskId}`,
+      domain: "database",
+      mutationType: WORKFLOW_PARITY_OBSERVED_MUTATION as never,
+      target: taskId,
+      metadata: { agree },
+    });
+    if (!agree && diffs) {
+      store.recordRunAuditEvent({
+        taskId,
+        agentId: "store",
+        runId: `parity-drift:${taskId}`,
+        domain: "database",
+        mutationType: WORKFLOW_PARITY_DRIFT_MUTATION as never,
+        target: taskId,
+        metadata: { agree, diffs },
+      });
+    }
+  }
+
+  it("returns zeros when no parity events recorded", () => {
+    const summary = store.getWorkflowParitySummary();
+    expect(summary).toMatchObject({ observed: 0, agreed: 0, drift: 0, agreeRate: 0 });
+    expect(summary.driftFieldCounts).toEqual({});
+  });
+
+  it("computes agree-rate and per-field drift counts", () => {
+    observe("FN-1", true);
+    observe("FN-2", true);
+    observe("FN-3", false, [
+      { field: "stageTransitions", legacy: [], interpreter: [], category: "lifecycle", severity: "error" },
+      { field: "mergeOutcome", legacy: "merged", interpreter: null, category: "lifecycle", severity: "error" },
+    ]);
+    observe("FN-4", false, [
+      { field: "stageTransitions", legacy: [], interpreter: [], category: "lifecycle", severity: "error" },
+    ]);
+
+    const summary = store.getWorkflowParitySummary();
+    expect(summary.observed).toBe(4);
+    expect(summary.agreed).toBe(2);
+    expect(summary.drift).toBe(2);
+    expect(summary.agreeRate).toBeCloseTo(0.5, 5);
+    expect(summary.driftFieldCounts).toEqual({ stageTransitions: 2, mergeOutcome: 1 });
+    expect(summary.recentDrift.length).toBe(2);
+    expect(summary.recentDrift[0].diffs.length).toBeGreaterThan(0);
+  });
+});

packages/core/src/index.ts

@@ -1148,6 +1148,7 @@ export type {
   WorkflowObservationTaskInput,
   WorkflowObservationBuildOptions,
   WorkflowObservationParts,
+  WorkflowParitySummary,
 } from "./workflow-parity.js";
 export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js";
 export type { ResolvedResearchSettings } from "./research-settings.js";

packages/core/src/store.ts

@@ -16,6 +16,12 @@ import type {
 } from "./workflow-definition-types.js";
 import { compileWorkflowToSteps } from "./workflow-compiler.js";
 import { BUILTIN_WORKFLOWS, getBuiltinWorkflow, isBuiltinWorkflowId } from "./builtin-workflows.js";
+import {
+  WORKFLOW_PARITY_OBSERVED_MUTATION,
+  WORKFLOW_PARITY_DRIFT_MUTATION,
+  type WorkflowParityDiff,
+  type WorkflowParitySummary,
+} from "./workflow-parity.js";
 
 /** Tags WorkflowStep rows materialized by compiling a workflow so they can be
  *  filtered out of the user-facing step manager and cleaned up on re-selection. */
@@ -7280,6 +7286,55 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     return rows.map((row) => this.rowToRunAuditEvent(row));
   }
 
+  /**
+   * Aggregate the dual-observe parity audit events (CU-U5) into the graduation
+   * signal: how often the interpreter's shadow observation agreed with the
+   * legacy authoritative run, and which fields drift when it doesn't.
+   */
+  getWorkflowParitySummary(options: { since?: string; limit?: number } = {}): WorkflowParitySummary {
+    const limit = options.limit ?? 1000;
+    const observed = this.getRunAuditEvents({
+      domain: "database",
+      mutationType: WORKFLOW_PARITY_OBSERVED_MUTATION as unknown as RunAuditEvent["mutationType"],
+      startTime: options.since,
+      limit,
+    });
+    const driftEvents = this.getRunAuditEvents({
+      domain: "database",
+      mutationType: WORKFLOW_PARITY_DRIFT_MUTATION as unknown as RunAuditEvent["mutationType"],
+      startTime: options.since,
+      limit,
+    });
+
+    let agreed = 0;
+    for (const event of observed) {
+      if (event.metadata?.agree === true) agreed += 1;
+    }
+
+    const driftFieldCounts: Record<string, number> = {};
+    const recentDrift: WorkflowParitySummary["recentDrift"] = [];
+    for (const event of driftEvents) {
+      const diffs = Array.isArray(event.metadata?.diffs)
+        ? (event.metadata.diffs as WorkflowParityDiff[])
+        : [];
+      for (const diff of diffs) {
+        driftFieldCounts[diff.field] = (driftFieldCounts[diff.field] ?? 0) + 1;
+      }
+      if (recentDrift.length < 20) {
+        recentDrift.push({ taskId: event.taskId ?? event.target, timestamp: event.timestamp, diffs });
+      }
+    }
+
+    return {
+      observed: observed.length,
+      agreed,
+      drift: driftEvents.length,
+      agreeRate: observed.length > 0 ? agreed / observed.length : 0,
+      driftFieldCounts,
+      recentDrift,
+    };
+  }
+
   enqueueMergeQueue(taskId: string, opts: MergeQueueEnqueueOptions = {}): MergeQueueEntry {
     let invalidColumn: Column | null = null;
     const entry = this.db.transactionImmediate(() => {

packages/core/src/workflow-parity.ts

@@ -276,6 +276,22 @@ export function buildWorkflowObservationFromTask(
   };
 }
 
+/** Aggregate of dual-observe parity audit events — the graduation signal. */
+export interface WorkflowParitySummary {
+  /** Total `workflow:parity-observed` events in scope. */
+  observed: number;
+  /** Of those, how many reported agree=true. */
+  agreed: number;
+  /** Total `workflow:parity-drift` events in scope. */
+  drift: number;
+  /** agreed / observed in [0,1]; 0 when nothing observed yet. */
+  agreeRate: number;
+  /** Count of drift occurrences per observation field, most-divergent first. */
+  driftFieldCounts: Record<string, number>;
+  /** Most recent drift events (capped) for inspection. */
+  recentDrift: Array<{ taskId: string; timestamp: string; diffs: WorkflowParityDiff[] }>;
+}
+
 export interface WorkflowObservationParts {
   stageTransitions: readonly WorkflowStage[];
   terminalColumn?: string | null;