[for #2153] parity: rerecord lanes against a v3 assembler + replay timing fixes

docs/provider-bridge-protocol.md

@@ -390,6 +390,16 @@ the pins and replays every cell through the current bridge on each commit,
 and `UPDATE_PARITY_ROW_COUNTS=1` rewrites the pins deliberately. Raw
 recordings stay out of git.
 
+A recording is never rewritten. When a bridge change alters what the bridge
+emits for a recording, `pnpm --filter @bb/provider-parity rerecord
+[--plan-with <recording-time checkout>]` writes the bridge's current output
+to `bridge→runtime.current.ndjson` beside the recorded lane; the self-suite
+pins and compares against that file when it exists, while `pnpm parity`
+still paces a pre-migration leg from the recorded lane (and the current leg
+from the current one). `pnpm parity --dump-dir <dir>` writes both legs'
+normalized event and row lists per cell, for allowlist entries that must
+name a list index.
+
 The conformance kit runs the same recordings as its recorded-traffic
 scenario set: `replayRecordedCells` replays a bridge's cells and
 `checkRecordedCellReplay` reports `recorded/<cell>/{replays,

packages/provider-bridge-protocol/src/testing/parity.ts

@@ -36,6 +36,7 @@ import {
   COMMITTED_RECORDINGS_ROOT,
   listRecordedCells,
   readBridgeRecording,
+  withCurrentBridgeLane,
   type BridgeRecording,
   type RecordedCell,
 } from "./recording.js";
@@ -348,6 +349,20 @@ export interface ReplayRecordingOptions {
   recordingDir: string;
   bridge: ParityBridgeSpec;
   createAssembler: CreateParityAssembler;
+  /**
+   * The assembler that plans the replay's gates from the recorded
+   * `bridge→runtime` lane; defaults to `createAssembler`. A re-recording run
+   * on a checkout whose grammar no longer accepts the whole recorded lane
+   * plans with the recording-time checkout's assembler instead.
+   */
+  createPlanAssembler?: CreateParityAssembler;
+  /**
+   * Plan the replay's gates from the cell's current bridge lane
+   * (`bridge→runtime.current.ndjson`, see `withCurrentBridgeLane`) when one
+   * exists, instead of the recorded lane. The leg whose bridge wrote that
+   * lane parses all of it; the recording-time leg parses the recorded lane.
+   */
+  planFromCurrentLane?: boolean;
   /** Per-wait timeout for a gate or a response. */
   timeoutMs?: number;
   /**
@@ -358,6 +373,14 @@ export interface ReplayRecordingOptions {
   orderTimeoutMs?: number;
   /** Quiet period after the last request before the bridge is closed. */
   settleMs?: number;
+  /**
+   * Quiet period a request waits for once the gates are met. The replay child
+   * plays every provider line before the request's cursor point a couple of
+   * milliseconds apart, so a short silence means the bridge has emitted all
+   * that the pre-request stream produces; without it a request the bridge
+   * acknowledges at once (a steer) lands at a load-dependent point.
+   */
+  drainMs?: number;
   /** Mirror the bridge's stderr (and the replay child's logs) here. */
   onStderr?: (text: string) => void;
 }
@@ -375,6 +398,12 @@ export interface ParityRun {
   lines: string[];
   /** When each line arrived, ms since the replay started (diagnostics). */
   lineTimes: number[];
+  /**
+   * For each line, the recorded `runtime→bridge` entry written last before
+   * it arrived (null before any was sent) — where the line sits in the
+   * recording's wire order, for a lane re-recorded through this bridge.
+   */
+  lineAfter: Array<{ run: number; seq: number; ts: number } | null>;
   /** Assembled events, minus the ones the grammar dropped (as the runtime does). */
   events: ThreadEvent[];
   grammarViolations: ParityGrammarViolation[];
@@ -502,6 +531,7 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
   // waits this long, while a slow CI runner must never trip it for a healthy one.
   const orderTimeoutMs = options.orderTimeoutMs ?? 5_000;
   const settleMs = options.settleMs ?? 750;
+  const drainMs = options.drainMs ?? 300;
   const providerId = options.bridge.providerId;
   const profile = resolveReplayProfile(providerId);
   const recording = readBridgeRecording(options.recordingDir);
@@ -557,14 +587,19 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
   const startedAt = Date.now();
   const lines: string[] = [];
   const lineTimes: number[] = [];
+  const lineAfter: ParityRun["lineAfter"] = [];
+  let lastSentRuntimeEntry: { run: number; seq: number; ts: number } | null = null;
   const events: ThreadEvent[] = [];
   const grammarViolations: ParityGrammarViolation[] = [];
   const stalls: string[] = [];
   let stderr = "";
   const grammar = new ThreadEventGrammar();
   const liveAssembler = options.createAssembler(providerId);
-  const planAssembler = options.createAssembler(providerId);
-  const steps = planRuntimeSteps(recording, planAssembler);
+  const planAssembler = (options.createPlanAssembler ?? options.createAssembler)(providerId);
+  const steps = planRuntimeSteps(
+    options.planFromCurrentLane === true ? withCurrentBridgeLane(recording) : recording,
+    planAssembler,
+  );
 
   const answeredIds = new Set<string>();
   const pendingBridgeRequests: { id: string | number; method: string }[] = [];
@@ -619,6 +654,7 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
       lastOutputAt = Date.now();
       lines.push(line);
       lineTimes.push(lastOutputAt - startedAt);
+      lineAfter.push(lastSentRuntimeEntry);
       const message = parseWire(line);
       if (message === null) return;
       if (isResponse(message)) {
@@ -698,6 +734,7 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
       // Responses are replayed on demand when the bridge asks; notifications
       // go straight through.
       if (step.message !== null && !isResponse(step.message)) {
+        lastSentRuntimeEntry = { run: step.entry.run, seq: step.entry.seq, ts: step.entry.ts };
         write(step.entry.line);
       }
       continue;
@@ -730,6 +767,12 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
       timeoutMs,
       false,
     );
+    await waitFor(
+      `the stream to drain before ${method}`,
+      () => Date.now() - lastOutputAt >= drainMs,
+      timeoutMs,
+      false,
+    );
     if (child.exitCode !== null) break;
     if (
       method === "thread/stop" &&
@@ -746,6 +789,7 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
       profile.rewriteRuntimeLine === undefined
         ? rewritten
         : profile.rewriteRuntimeLine(rewritten, { replayCommand });
+    lastSentRuntimeEntry = { run: step.entry.run, seq: step.entry.seq, ts: step.entry.ts };
     write(line);
     sentRequestIds.push(String(request.id));
     // Release the provider lines up to the next runtime request.
@@ -775,6 +819,7 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
     recordingDir: options.recordingDir,
     lines,
     lineTimes,
+    lineAfter,
     events,
     grammarViolations,
     stalls,
@@ -1085,15 +1130,19 @@ export async function replayRecordedCells(
   );
   return Promise.all(
     cells.map(async (cell): Promise<RecordedCellReplay> => {
+      // The expectation is this checkout's current bridge lane when a bridge
+      // change wrote one (`pnpm rerecord`), else the recorded lane; the
+      // replay paces itself from the same lane.
       const recorded = assembleRecordedEvents(
-        readBridgeRecording(cell.dir),
+        withCurrentBridgeLane(readBridgeRecording(cell.dir)),
         options.createAssembler,
         cell.provider,
       );
       const run = await replayRecording({
         recordingDir: cell.dir,
         bridge: { checkoutRoot, providerId: cell.provider },
         createAssembler: options.createAssembler,
+        planFromCurrentLane: true,
         ...(options.timeoutMs !== undefined ? { timeoutMs: options.timeoutMs } : {}),
         ...(options.onStderr !== undefined ? { onStderr: options.onStderr } : {}),
       });

packages/provider-bridge-protocol/src/testing/recording.ts

@@ -92,6 +92,51 @@ export function readBridgeRecordingLane(
   return entries;
 }
 
+/**
+ * The file a bridge change re-records its side of the wire into
+ * (`pnpm rerecord`): the `bridge→runtime` lane as THIS checkout's bridge
+ * emits it for the recording's provider and runtime lanes. The recorded lane
+ * itself is never rewritten — it is the recording, and a pre-migration
+ * checkout paces its replay from it — so the current expectation lives
+ * beside it. Absent until a bridge change first needs one.
+ */
+export const CURRENT_BRIDGE_LANE_FILE = "bridge→runtime.current.ndjson";
+
+export function readCurrentBridgeLane(dir: string): BridgeRecordingEntry[] | null {
+  const file = join(dir, CURRENT_BRIDGE_LANE_FILE);
+  if (!existsSync(file)) {
+    return null;
+  }
+  const entries: BridgeRecordingEntry[] = [];
+  const lines = readFileSync(file, "utf8").split("\n");
+  for (const [index, raw] of lines.entries()) {
+    if (raw.length === 0) continue;
+    const entry = parseEntry(raw, file, index + 1);
+    if (entry.dir !== "bridge→runtime") {
+      throw new Error(`${file}:${index + 1}: entry direction ${entry.dir} in the current bridge lane`);
+    }
+    entries.push(entry);
+  }
+  return entries;
+}
+
+/**
+ * The recording with its `bridge→runtime` lane replaced by the current
+ * expectation when one exists: what the self-suite pins and compares.
+ */
+export function withCurrentBridgeLane(recording: BridgeRecording): BridgeRecording {
+  const current = readCurrentBridgeLane(recording.dir);
+  if (current === null) {
+    return recording;
+  }
+  const entries = [
+    ...recording.entries.filter((entry) => entry.dir !== "bridge→runtime"),
+    ...current,
+  ];
+  entries.sort(compareRecordingEntries);
+  return { ...recording, entries };
+}
+
 export function readBridgeRecording(dir: string): BridgeRecording {
   const manifestPath = join(dir, "manifest.json");
   const manifest = existsSync(manifestPath)

packages/provider-bridge-protocol/src/testing/replay-provider-child.mjs

@@ -53,6 +53,15 @@ const CURSOR_POLL_MS = 5;
  * them, or the replay reorders what the recording had in order.
  */
 const EMIT_GAP_MS = 2;
+/**
+ * Gap after a response. The bridge continues its request's continuation in
+ * a microtask once the line loop yields; a notification read in the same
+ * chunk is handled first, so under load two milliseconds let a steer's ack
+ * (emitted after `await request("turn/steer")`) land after the next
+ * notification instead of before it, as the recording had it. A response
+ * is rare, so the longer gap costs nothing measurable.
+ */
+const RESPONSE_GAP_MS = 50;
 /** A request that opens or addresses a provider session; see segment release. */
 const SESSION_DEFINING_KEY =
   /^(thread|session)\/(start|resume|fork|new|load|archive|unarchive|name\/set)$/;
@@ -333,12 +342,12 @@ function main() {
     return null;
   }
 
-  function scheduleAdvance() {
+  function scheduleAdvance(gapMs = EMIT_GAP_MS) {
     if (emitTimer !== null) return;
     emitTimer = setTimeout(() => {
       emitTimer = null;
       advance();
-    }, EMIT_GAP_MS);
+    }, gapMs);
   }
 
   function advance() {
@@ -383,7 +392,9 @@ function main() {
         }
         emitRecorded(step);
         position += 1;
-        scheduleAdvance();
+        scheduleAdvance(
+          step.classified.kind === "response" ? RESPONSE_GAP_MS : EMIT_GAP_MS,
+        );
         return;
       }
       // An expectation of what the bridge writes.

packages/provider-parity/package.json

@@ -15,7 +15,8 @@
     "clean": "rimraf dist tsconfig.tsbuildinfo",
     "typecheck": "tsc --noEmit",
     "test": "vitest run --config vitest.config.ts",
-    "parity": "node --conditions=source --import tsx src/cli.ts"
+    "parity": "node --conditions=source --import tsx src/cli.ts",
+    "rerecord": "node --conditions=source --import tsx src/rerecord.ts"
   },
   "dependencies": {
     "@bb/agent-runtime": "workspace:*",

packages/provider-parity/src/cli.ts

@@ -1,15 +1,22 @@
 #!/usr/bin/env node
 /**
  * `pnpm parity --old <checkout> --new . [--provider <id>] [--cell <name>]
- *              [--recordings <dir>] [--allowlist <file>] [--timeout <ms>] [--verbose]`
+ *              [--recordings <dir>] [--allowlist <file>] [--timeout <ms>] [--verbose]
+ *              [--dump-dir <dir>]`
+ *
+ * `--dump-dir` writes each leg's normalized event and row lists per cell
+ * (`<provider>-<cell>.<old|new>.<events|rows>.json`), so an allowlist entry
+ * can name the exact list index of a diff the 160-character CLI rendering
+ * cannot show in full.
  *
  * Replay every committed recording through the bridge of two checkouts and
  * diff the assembled events and projected rows against the allowlist. Exit
  * 1 on any unallowed diff or stale allowlist entry, so the run doubles as a
  * gate. With `--old` equal to `--new` the diff must be empty: that is the
  * harness's own acceptance test.
  */
-import { resolve } from "node:path";
+import { mkdirSync, writeFileSync } from "node:fs";
+import { join, resolve } from "node:path";
 import {
   ALLOWLIST_PATH,
   RECORDINGS_ROOT,
@@ -26,7 +33,11 @@ import {
   type RecordedCell,
 } from "./index.js";
 import { loadParityLeg } from "./leg.js";
-import { describeParityValue } from "@bb/provider-bridge-protocol/testing/parity";
+import {
+  describeParityValue,
+  normalizeParityEvents,
+  normalizeParityRows,
+} from "@bb/provider-bridge-protocol/testing/parity";
 
 interface CliArgs {
   oldRoot: string;
@@ -37,11 +48,12 @@ interface CliArgs {
   allowlist: string;
   timeoutMs: number | undefined;
   verbose: boolean;
+  dumpDir: string | null;
 }
 
 function usage(): never {
   process.stderr.write(
-    "usage: pnpm parity --old <checkout> --new <checkout> [--provider <id>] [--cell <name>] [--recordings <dir>] [--allowlist <file>] [--timeout <ms>] [--verbose]\n",
+    "usage: pnpm parity --old <checkout> --new <checkout> [--provider <id>] [--cell <name>] [--recordings <dir>] [--allowlist <file>] [--timeout <ms>] [--verbose] [--dump-dir <dir>]\n",
   );
   process.exit(2);
 }
@@ -59,6 +71,7 @@ function parseArgs(argv: string[]): CliArgs {
     allowlist: ALLOWLIST_PATH,
     timeoutMs: undefined,
     verbose: false,
+    dumpDir: null,
   };
   for (let index = 0; index < argv.length; index += 1) {
     const flag = argv[index];
@@ -95,6 +108,10 @@ function parseArgs(argv: string[]): CliArgs {
       case "--verbose":
         args.verbose = true;
         break;
+      case "--dump-dir":
+        args.dumpDir = resolve(callerCwd, value ?? usage());
+        index += 1;
+        break;
       default:
         usage();
     }
@@ -167,10 +184,21 @@ async function main(): Promise<void> {
     const onStderr = args.verbose
       ? (text: string) => process.stderr.write(text)
       : undefined;
+    // Each leg paces itself from the lane its own assembler parses whole:
+    // the old (recording-time) leg from the recorded lane, the new leg from
+    // the current lane its bridge wrote, when there is one.
     const [oldInputs, newInputs] = await Promise.all([
       replayCell(cell, { ...oldLeg, timeoutMs: args.timeoutMs, onStderr }),
-      replayCell(cell, { ...newLeg, timeoutMs: args.timeoutMs, onStderr }),
+      replayCell(cell, { ...newLeg, timeoutMs: args.timeoutMs, onStderr, planFromCurrentLane: true }),
     ]);
+    if (args.dumpDir !== null) {
+      mkdirSync(args.dumpDir, { recursive: true });
+      const prefix = join(args.dumpDir, `${cell.provider}-${cell.cell}`);
+      writeFileSync(`${prefix}.old.events.json`, JSON.stringify(normalizeParityEvents(oldInputs.events), null, 2));
+      writeFileSync(`${prefix}.new.events.json`, JSON.stringify(normalizeParityEvents(newInputs.events), null, 2));
+      writeFileSync(`${prefix}.old.rows.json`, JSON.stringify(normalizeParityRows(oldInputs.rows), null, 2));
+      writeFileSync(`${prefix}.new.rows.json`, JSON.stringify(normalizeParityRows(newInputs.rows), null, 2));
+    }
     const comparison = compareCell(cell, oldInputs, newInputs, allowlist);
     const oldCounts = countCellInputs(oldInputs);
     const newCounts = countCellInputs(newInputs);