feat(core): workflow parity observation builders (CU-U5 #1)

Add buildWorkflowObservationFromTask (legacy authoritative side, from a task's
terminal column/status/review/mergeDetails + recorded column history) and
buildWorkflowObservation (interpreter/shadow side, from explicit parts), plus
deriveStageTransitions (maps the task-move column history to execute/review/
merge stages) and DEFAULT_WORKFLOW_INVARIANTS. These let both sides of the
dual-observe seam produce a comparable WorkflowRunObservation without
hand-rolling the shape. Covered by workflow-parity.test.ts.

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

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

@@ -2,6 +2,10 @@ import { describe, expect, it } from "vitest";
 import {
   compareWorkflowRunAudits,
   compareWorkflowRunObservations,
+  buildWorkflowObservationFromTask,
+  buildWorkflowObservation,
+  deriveStageTransitions,
+  DEFAULT_WORKFLOW_INVARIANTS,
   type RunAuditEvent,
   type WorkflowRunObservation,
 } from "../index.js";
@@ -141,3 +145,53 @@ describe("workflow parity", () => {
     );
   });
 });
+
+describe("observation builders (CU-U5)", () => {
+  it("deriveStageTransitions maps columns to stages and collapses repeats", () => {
+    expect(deriveStageTransitions(["todo", "in-progress", "in-progress", "in-review", "done"]))
+      .toEqual(["triage", "execute", "review", "merge"]);
+    expect(deriveStageTransitions([])).toEqual([]);
+  });
+
+  it("buildWorkflowObservationFromTask reads terminal lifecycle + derives stages from columnSequence", () => {
+    const obs = buildWorkflowObservationFromTask(
+      { column: "done", status: null, review: { verdict: "APPROVE" }, mergeDetails: { outcome: "merged" } },
+      { columnSequence: ["todo", "in-progress", "in-review", "done"] },
+    );
+    expect(obs.stageTransitions).toEqual(["triage", "execute", "review", "merge"]);
+    expect(obs.terminalColumn).toBe("done");
+    expect(obs.reviewVerdict).toBe("APPROVE");
+    expect(obs.mergeOutcome).toBe("merged");
+    expect(obs.invariants).toEqual(DEFAULT_WORKFLOW_INVARIANTS);
+  });
+
+  it("buildWorkflowObservationFromTask infers merged from terminal column when mergeDetails absent", () => {
+    const obs = buildWorkflowObservationFromTask({ column: "done" });
+    expect(obs.mergeOutcome).toBe("merged");
+    expect(obs.stageTransitions).toEqual(["merge"]); // terminal-only fallback
+  });
+
+  it("a task and an equivalent interpreter parts observation compare as agree", () => {
+    const legacy = buildWorkflowObservationFromTask(
+      { column: "done", review: { verdict: "APPROVE" }, mergeDetails: { outcome: "merged" } },
+      { columnSequence: ["in-progress", "in-review", "done"] },
+    );
+    const interpreter = buildWorkflowObservation({
+      stageTransitions: ["execute", "review", "merge"],
+      terminalColumn: "done",
+      reviewVerdict: "APPROVE",
+      mergeOutcome: "merged",
+    });
+    expect(compareWorkflowRunObservations(legacy, interpreter).agree).toBe(true);
+  });
+
+  it("a divergent stage sequence surfaces an error-severity lifecycle drift", () => {
+    const legacy = buildWorkflowObservationFromTask({ column: "done" }, { columnSequence: ["in-progress", "in-review", "done"] });
+    const interpreter = buildWorkflowObservation({ stageTransitions: ["execute", "merge"], terminalColumn: "done", mergeOutcome: "merged" });
+    const report = compareWorkflowRunObservations(legacy, interpreter);
+    expect(report.agree).toBe(false);
+    expect(report.diffs).toEqual(
+      expect.arrayContaining([expect.objectContaining({ field: "stageTransitions", category: "lifecycle", severity: "error" })]),
+    );
+  });
+});

packages/core/src/index.ts

@@ -1131,6 +1131,10 @@ export {
   compareWorkflowRunAudits,
   compareWorkflowRunObservations,
   extractWorkflowAuditObservations,
+  DEFAULT_WORKFLOW_INVARIANTS,
+  deriveStageTransitions,
+  buildWorkflowObservationFromTask,
+  buildWorkflowObservation,
 } from "./workflow-parity.js";
 export type {
   WorkflowAuditObservation,
@@ -1141,6 +1145,9 @@ export type {
   WorkflowReliabilityInvariantSignals,
   WorkflowRunObservation,
   WorkflowStage,
+  WorkflowObservationTaskInput,
+  WorkflowObservationBuildOptions,
+  WorkflowObservationParts,
 } from "./workflow-parity.js";
 export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js";
 export type { ResolvedResearchSettings } from "./research-settings.js";

packages/core/src/workflow-parity.ts

@@ -203,3 +203,99 @@ export function compareWorkflowRunAudits(
     diffs,
   };
 }
+
+// ── Observation builders (CU-U5) ─────────────────────────────────────────────
+// Construct a WorkflowRunObservation from real run data so the legacy and
+// interpreter sides can be compared without either side hand-rolling the shape.
+
+/** Conservative defaults: a run that didn't signal an invariant is assumed to
+ *  have respected the terminal/cancel contracts (the common, non-drift case). */
+export const DEFAULT_WORKFLOW_INVARIANTS: WorkflowReliabilityInvariantSignals = {
+  fileScopeGuardOutcome: null,
+  squashMergeContractOutcome: null,
+  autoMergeTerminalUntilMergedRespected: true,
+  moveTaskHardCancelRespected: true,
+};
+
+const COLUMN_TO_STAGE: Record<string, WorkflowStage | undefined> = {
+  triage: "triage",
+  todo: "triage",
+  "in-progress": "execute",
+  "in-review": "review",
+  done: "merge",
+};
+
+/**
+ * Map the ordered list of columns a run passed through to workflow stages,
+ * collapsing consecutive repeats. This is how the legacy side derives its
+ * stageTransitions — from the real task-move history rather than a guess.
+ */
+export function deriveStageTransitions(columnSequence: readonly string[]): WorkflowStage[] {
+  const stages: WorkflowStage[] = [];
+  for (const column of columnSequence) {
+    const stage = COLUMN_TO_STAGE[column];
+    if (stage && stages[stages.length - 1] !== stage) stages.push(stage);
+  }
+  return stages;
+}
+
+export interface WorkflowObservationTaskInput {
+  column: string;
+  status?: string | null;
+  review?: { verdict?: string } | null;
+  mergeDetails?: { outcome?: string } | null;
+}
+
+export interface WorkflowObservationBuildOptions {
+  /** Explicit stage sequence (wins over columnSequence). */
+  stageTransitions?: readonly WorkflowStage[];
+  /** Ordered columns the run passed through; mapped to stages when stageTransitions is absent. */
+  columnSequence?: readonly string[];
+  invariants?: Partial<WorkflowReliabilityInvariantSignals>;
+}
+
+/**
+ * Build a parity observation from a task's terminal persisted state (the legacy
+ * authoritative side). stageTransitions come from the caller's recorded column
+ * history when available, else fall back to the terminal column alone.
+ */
+export function buildWorkflowObservationFromTask(
+  task: WorkflowObservationTaskInput,
+  options?: WorkflowObservationBuildOptions,
+): WorkflowRunObservation {
+  const stageTransitions = options?.stageTransitions
+    ? [...options.stageTransitions]
+    : deriveStageTransitions(options?.columnSequence ?? [task.column]);
+  return {
+    stageTransitions,
+    terminalColumn: task.column ?? null,
+    terminalStatus: task.status ?? null,
+    reviewVerdict: task.review?.verdict ?? null,
+    mergeOutcome: task.mergeDetails?.outcome ?? (task.column === "done" ? "merged" : null),
+    invariants: { ...DEFAULT_WORKFLOW_INVARIANTS, ...options?.invariants },
+  };
+}
+
+export interface WorkflowObservationParts {
+  stageTransitions: readonly WorkflowStage[];
+  terminalColumn?: string | null;
+  terminalStatus?: string | null;
+  reviewVerdict?: string | null;
+  mergeOutcome?: string | null;
+  invariants?: Partial<WorkflowReliabilityInvariantSignals>;
+}
+
+/**
+ * Build a parity observation from explicit parts (the interpreter/shadow side
+ * assembles these from its graph-walk result).
+ */
+export function buildWorkflowObservation(parts: WorkflowObservationParts): WorkflowRunObservation {
+  return {
+    stageTransitions: [...parts.stageTransitions],
+    terminalColumn: parts.terminalColumn ?? null,
+    terminalStatus: parts.terminalStatus ?? null,
+    reviewVerdict: parts.reviewVerdict ?? null,
+    mergeOutcome: parts.mergeOutcome ?? null,
+    invariants: { ...DEFAULT_WORKFLOW_INVARIANTS, ...parts.invariants },
+  };
+}