From dbb8adeff15a4aa3e7e40f9b8ed50fbc7d1e87e1 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 18:54:39 -0700 Subject: [PATCH] feat: surface dual-observe flag + parity summary (CU-U5 #3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- .../__tests__/workflow-parity-summary.test.ts | 69 +++++++++++++++++++ packages/core/src/index.ts | 1 + packages/core/src/store.ts | 55 +++++++++++++++ packages/core/src/workflow-parity.ts | 16 +++++ .../app/components/SettingsModal.tsx | 1 + 5 files changed, 142 insertions(+) create mode 100644 packages/core/src/__tests__/workflow-parity-summary.test.ts diff --git a/packages/core/src/__tests__/workflow-parity-summary.test.ts b/packages/core/src/__tests__/workflow-parity-summary.test.ts new file mode 100644 index 0000000000..f2893ab8bc --- /dev/null +++ b/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; + + 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); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 2f65e2e5ab..3ec649baf2 100644 --- a/packages/core/src/index.ts +++ b/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"; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 443d3c9e2c..4d9dd58a5d 100644 --- a/packages/core/src/store.ts +++ b/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 { 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 = {}; + 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(() => { diff --git a/packages/core/src/workflow-parity.ts b/packages/core/src/workflow-parity.ts index ec55262a74..56bc4a92f7 100644 --- a/packages/core/src/workflow-parity.ts +++ b/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; + /** Most recent drift events (capped) for inspection. */ + recentDrift: Array<{ taskId: string; timestamp: string; diffs: WorkflowParityDiff[] }>; +} + export interface WorkflowObservationParts { stageTransitions: readonly WorkflowStage[]; terminalColumn?: string | null; diff --git a/packages/dashboard/app/components/SettingsModal.tsx b/packages/dashboard/app/components/SettingsModal.tsx index 2fa8729550..b58a39a0d9 100644 --- a/packages/dashboard/app/components/SettingsModal.tsx +++ b/packages/dashboard/app/components/SettingsModal.tsx @@ -345,6 +345,7 @@ const KNOWN_EXPERIMENTAL_FEATURES: Record = { chatRooms: "Chat Rooms", agentOnboarding: "Planning-style Agent Onboarding", workflowGraphExecutor: "Workflow Graph Engine (run custom workflows)", + workflowInterpreterDualObserve: "Workflow Graph Engine — dual-observe parity (diagnostic)", }; const EXPERIMENTAL_FEATURE_LEGACY_ALIASES: Record = {