Wait for quiet after exact replay completion

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

@@ -29,7 +29,7 @@ function writeLane(
   );
 }
 
-it("waits for the exact planned tail before closing bridge stdin", async () => {
+it("waits for the exact planned tail and a quiet period before closing", async () => {
   const dir = mkdtempSync(join(tmpdir(), "bb-parity-tail-test-"));
   const bridgePath = join(dir, "delayed-tail-bridge.mjs");
   const identity = {
@@ -55,6 +55,18 @@ it("waits for the exact planned tail before closing bridge stdin", async () => {
     },
     { type: "turn/completed", ...identity, scope, status: "completed" },
   ];
+  const extraEvents: ThreadEvent[] = [
+    {
+      type: "thread/contextWindowUsage/updated",
+      ...identity,
+      scope: { kind: "thread" },
+      contextWindowUsage: {
+        usedTokens: 42,
+        modelContextWindow: 1_000,
+        estimated: false,
+      },
+    },
+  ];
   const delta = (events: readonly ThreadEvent[]): string =>
     JSON.stringify({
       jsonrpc: "2.0",
@@ -83,6 +95,7 @@ it("waits for the exact planned tail before closing bridge stdin", async () => {
       [
         `const prefix = ${JSON.stringify(prefixEvents)};`,
         `const tail = ${JSON.stringify(tailEvents)};`,
+        `const extra = ${JSON.stringify(extraEvents)};`,
         "const delta = (events) => JSON.stringify({ jsonrpc: '2.0', method: 'thread/delta', params: { events } });",
         "let pending = '';",
         "let tailTimer = null;",
@@ -100,7 +113,10 @@ it("waits for the exact planned tail before closing bridge stdin", async () => {
         "    } else if (message.method === 'thread/start') {",
         "      process.stdout.write(delta(prefix) + '\\n');",
         "      process.stdout.write(JSON.stringify({ jsonrpc: '2.0', id: message.id, result: {} }) + '\\n');",
-        "      tailTimer = setTimeout(() => process.stdout.write(delta(tail) + '\\n'), 100);",
+        "      tailTimer = setTimeout(() => {",
+        "        process.stdout.write(delta(tail) + '\\n');",
+        "        tailTimer = setTimeout(() => process.stdout.write(delta(extra) + '\\n'), 40);",
+        "      }, 100);",
         "    }",
         "  }",
         "});",
@@ -135,13 +151,17 @@ it("waits for the exact planned tail before closing bridge stdin", async () => {
         },
       }),
       planFromCurrentLane: true,
-      settleMs: 20,
+      settleMs: 60,
       timeoutMs: 1_000,
     });
 
     expect(run.stalls).toEqual([]);
-    expect(run.events).toEqual([...prefixEvents, ...tailEvents]);
-    expect(run.events.at(-1)?.type).toBe("turn/completed");
+    expect(run.grammarViolations).toEqual([]);
+    expect(run.events).toEqual([
+      ...prefixEvents,
+      ...tailEvents,
+      ...extraEvents,
+    ]);
   } finally {
     rmSync(dir, { recursive: true, force: true });
   }

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

@@ -301,8 +301,8 @@ export interface ReplayRecordingOptions {
    */
   orderTimeoutMs?: number;
   /**
-   * Quiet period after the last request before a non-exact replay is closed.
-   * An exact current-lane replay waits for every planned event instead.
+   * Quiet period after the last bridge output before the replay is closed.
+   * An exact current-lane replay first waits for every planned event.
    */
   settleMs?: number;
   /**
@@ -760,21 +760,19 @@ export async function replayRecording(options: ReplayRecordingOptions): Promise<
   }
   setCursor("end");
   await waitFor("the last responses", () => sentRequestIds.every((id) => answeredIds.has(id)));
-  // An exact plan is this bridge's own recorded lane, so its complete accepted
-  // event count is the deterministic end of replay. Output silence is not:
-  // the provider child and bridge can be alive in provider-internal control
-  // flow without emitting runtime deltas, especially under scheduler load.
+  // An exact plan is this bridge's own recorded lane, so wait for its complete
+  // accepted event count before considering the stream settled. Output silence
+  // alone is not enough while planned events are still missing.
   if (plannedEventCount !== null) {
     await waitFor(
       `all ${plannedEventCount} planned events before closing the bridge`,
       () => events.length >= plannedEventCount,
     );
-  } else {
-    // A bridge compared with a lane from another version may deliberately
-    // emit fewer events, so only that non-exact comparison needs the quiet
-    // fallback to terminate and report its parity diff.
-    await waitFor("the stream to settle", () => Date.now() - lastOutputAt >= settleMs);
   }
+  // Reaching the planned count is only a lower bound: another valid event may
+  // still be in flight. Non-exact replays retain this same divergent-version
+  // fallback because they may deliberately emit fewer events than the plan.
+  await waitFor("the stream to settle", () => Date.now() - lastOutputAt >= settleMs);
   child.stdin?.end();
   const exitCode = await Promise.race([
     exited,