From 8e9a3ac4cecbce40945716f2c2a3f040faa7a0a4 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 19:02:37 -0700 Subject: [PATCH] feat(engine): wire dual-observe parity at the post-execute handoff (CU-U5 #2) maybeObserveWorkflowParity compares the selected workflow graph's routing against the legacy authoritative run for the same task and records workflow:parity-observed / -drift audit events (aggregated by getWorkflowParitySummary). The shadow walks the graph via WorkflowGraphTaskRunner with no-side-effect seams whose outcomes mirror the legacy task's reality (execute reached review?, review verdict, merged?), then maps visited seam nodes to execute/review/merge stages. Hooked in handoffTaskToReview, gated by workflowInterpreterDualObserve (off by default), wrapped so it never affects the authoritative handoff. Scope: execute->review->merge routing parity; full execution-fidelity shadow is future graduation work. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/engine/src/executor.ts | 123 ++++++++++++++++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 0e2ce55cc4..62ea97fcc6 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -10,7 +10,14 @@ import { existsSync, realpathSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core"; import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled } from "@fusion/core"; +import { + buildWorkflowObservationFromTask, + buildWorkflowObservation, + type WorkflowStage, + type WorkflowRunObservation, +} from "@fusion/core"; import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js"; +import { observeWorkflowParity, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js"; import type { WorkflowLegacySeams } from "./workflow-node-handlers.js"; import type { WorkflowNodeResult } from "./workflow-graph-executor.js"; import { @@ -1455,6 +1462,10 @@ export class TaskExecutor { }); } + // Dual-observe parity (CU-U5): post-execute observation point. Flag-gated + // and fully isolated — never affects the authoritative handoff result. + await this.maybeObserveWorkflowParity(task.id, settings); + return handedOff; } @@ -3268,6 +3279,118 @@ export class TaskExecutor { } } + /** + * Dual-observe parity (CU-U5): for a workflow-selected task, compare the + * selected graph's routing against the legacy authoritative run for the SAME + * task and record the result as workflow:parity-observed / -drift audit + * events. Observe-only — the shadow walks the graph with no-side-effect seams + * driven by the legacy task's actual outcomes, so it never mutates anything. + * Gated by workflowInterpreterDualObserve (off by default) and fully isolated + * (never throws into the caller). Hooked at the post-execute handoff point. + * + * Scope: this validates execute→review→merge ROUTING parity. Full + * execution-fidelity parity (a real isolated shadow run) is future work. + */ + private async maybeObserveWorkflowParity(taskId: string, settings: Settings): Promise { + if (!isExperimentalFeatureEnabled(settings, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG)) return; + if (typeof this.store.getTaskWorkflowSelection !== "function") return; + try { + const selection = this.store.getTaskWorkflowSelection(taskId); + if (!selection) return; + const def = await this.store.getWorkflowDefinition?.(selection.workflowId); + if (!def) return; + const live = await this.store.getTask(taskId); + + const legacyObs = buildWorkflowObservationFromTask( + { + column: live.column, + status: live.status ?? null, + review: live.review as { verdict?: string } | null, + mergeDetails: live.mergeDetails as { outcome?: string } | null, + }, + { columnSequence: this.inferLegacyColumnSequence(live.column) }, + ); + const legacyAudit = typeof this.store.getRunAuditEvents === "function" + ? this.store.getRunAuditEvents({ taskId }) + : []; + + await observeWorkflowParity({ + settings, + store: this.store, + agentId: "workflow-shadow", + legacy: { taskId, observation: legacyObs, auditEvents: legacyAudit }, + runShadow: async () => ({ + observation: await this.buildShadowObservation(live, def, settings, legacyObs), + auditEvents: [], + }), + }); + } catch (err) { + executorLog.warn( + `${taskId}: dual-observe parity skipped: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + /** Canonical column path a legacy run took to reach its terminal column + * (excluding the pre-execute todo/triage prefix so it lines up with the + * graph's execute→review→merge seam nodes). */ + private inferLegacyColumnSequence(terminalColumn: string): string[] { + switch (terminalColumn) { + case "done": return ["in-progress", "in-review", "done"]; + case "in-review": return ["in-progress", "in-review"]; + case "in-progress": return ["in-progress"]; + default: return [terminalColumn]; + } + } + + /** Build the interpreter-side observation by walking the selected graph with + * no-side-effect seams whose outcomes mirror the legacy task's reality. */ + private async buildShadowObservation( + live: TaskDetail, + def: { ir: { nodes: Array<{ id: string; kind: string; config?: Record }> } }, + settings: Settings, + legacyObs: WorkflowRunObservation, + ): Promise { + const reachedReview = live.column === "in-review" || live.column === "done"; + const merged = live.column === "done"; + const verdict = (live.review as { verdict?: string } | undefined)?.verdict; + const outcome = (ok: boolean): WorkflowNodeResult => ({ outcome: ok ? "success" : "failure" }); + const seams: WorkflowLegacySeams = { + planning: async () => outcome(true), + execute: async () => outcome(reachedReview || merged), + review: async () => outcome(verdict !== "REVISE"), + merge: async () => outcome(merged), + schedule: async () => outcome(true), + }; + const runner = new WorkflowGraphTaskRunner({ + store: this.store, + seams, + runCustomNode: async () => outcome(true), + }); + const result = await runner.run(live, settings); + + const stageByNodeId = new Map(); + for (const node of def.ir.nodes) { + const seam = typeof node.config?.seam === "string" ? node.config.seam : undefined; + if (seam === "execute" || seam === "review" || seam === "merge") { + stageByNodeId.set(node.id, seam); + } + } + const stages: WorkflowStage[] = []; + for (const nodeId of result.visitedNodeIds) { + const stage = stageByNodeId.get(nodeId); + if (stage && stages[stages.length - 1] !== stage) stages.push(stage); + } + + return buildWorkflowObservation({ + stageTransitions: stages, + terminalColumn: result.disposition === "completed" ? (merged ? "done" : "in-review") : live.column, + terminalStatus: live.status ?? null, + reviewVerdict: legacyObs.reviewVerdict, + mergeOutcome: merged ? "merged" : null, + }); + } + /** * Run ONLY the implementation phase of execute() for a graph-driven task — * full legacy setup plus the agent session up to fn_task_done. The registered