fix(cursor): state the T04 re-arming contract on an injectable clock

src/adapters/cursor/live-transport.ts

@@ -87,7 +87,7 @@ import {
   type BackgroundShellTerminationReport,
 } from "./native-exec-shell";
 import type { CursorClientMessage, CursorRunRequest, CursorServerMessage } from "./types";
-import type { CursorTransport, CursorTransportFactoryInput } from "./transport";
+import type { CursorStreamHealthClock, CursorTransport, CursorTransportFactoryInput } from "./transport";
 import { CursorHttp1BidiConnection } from "./http1-bidi";
 import { isPinnedHttp1 } from "../../lib/upstream-http-version";
 
@@ -107,6 +107,16 @@ const CURSOR_STREAM_SILENCE_FAIL_MS = 30_000;
  * Reset on every decoded frame that is not liveness-only.
  */
 const CURSOR_STREAM_HEARTBEAT_ONLY_FAIL_MS = 90_000;
+/**
+ * The production T04 clock. The watchdog reads time and schedules its one timer through this
+ * object so a test can supply a clock it advances by hand; nothing else about the transport's
+ * timers moves with it.
+ */
+const REAL_STREAM_HEALTH_CLOCK: CursorStreamHealthClock = {
+  now: () => Date.now(),
+  setTimeout: (callback, ms) => setTimeout(callback, ms),
+  clearTimeout: timer => clearTimeout(timer),
+};
 /**
  * After `turnEnded` is decoded, the application turn is complete. A server that keeps
  * HTTP/2 open past this point cannot hold the turn hostage (senpi #1062): we close our side
@@ -494,6 +504,8 @@ class LiveCursorTransport implements CursorTransport {
   private lastInboundFrameAt = 0;
   private lastMeaningfulFrameAt = 0;
   private streamHealthFail?: (error: Error) => void;
+  /** Time source for the T04 watchdog only; the globals unless a test injects one. */
+  private readonly streamHealthClock: CursorStreamHealthClock;
   private committed = false;
   private expectedClose = false;
   /**
@@ -542,6 +554,7 @@ class LiveCursorTransport implements CursorTransport {
     // so the transport-level race test can drive it deterministically.
     this.clientToolFinalizeGraceMs = input.clientToolFinalizeGraceMs ?? CLIENT_TOOL_FINALIZE_GRACE_MS;
     this.activeClientToolFinalizeGraceMs = this.clientToolFinalizeGraceMs;
+    this.streamHealthClock = input.streamHealthClock ?? REAL_STREAM_HEALTH_CLOCK;
     // Desktop (computer-use / record-screen) executors are available even with no MCP servers.
     this.desktopDeps = desktopDepsFromConfig(input.provider.desktopExecutor);
     this.execContext = {
@@ -847,7 +860,7 @@ class LiveCursorTransport implements CursorTransport {
 
   private clearStreamHealthTimer(): void {
     if (this.streamHealthTimer) {
-      clearTimeout(this.streamHealthTimer);
+      this.streamHealthClock.clearTimeout(this.streamHealthTimer);
       this.streamHealthTimer = undefined;
     }
     this.streamHealthFail = undefined;
@@ -858,24 +871,28 @@ class LiveCursorTransport implements CursorTransport {
    * failAndClear; the timer owns nothing else. Never armed before the first decoded
    * frame (the first-frame timer covers dial + first response), and disarmed by
    * every settle / expected-close path alongside the other timers.
+   *
+   * Both clocks and the timer go through `streamHealthClock`, which is the globals in
+   * production. The `elapsedMs` diagnostic below deliberately stays on the wall clock,
+   * because `turnStartedAt` is stamped there and the pair has to subtract coherently.
    */
   private armStreamHealthTimer(fail: (error: Error) => void): void {
-    if (this.streamHealthTimer) clearTimeout(this.streamHealthTimer);
+    if (this.streamHealthTimer) this.streamHealthClock.clearTimeout(this.streamHealthTimer);
     if (this.expectedClose) return;
     this.streamHealthFail = fail;
     const silenceMs = this.input.streamSilenceFailMs ?? CURSOR_STREAM_SILENCE_FAIL_MS;
     const heartbeatOnlyMs = this.input.streamHeartbeatOnlyFailMs ?? CURSOR_STREAM_HEARTBEAT_ONLY_FAIL_MS;
-    const now = Date.now();
+    const now = this.streamHealthClock.now();
     const deadline = Math.min(
       this.lastInboundFrameAt + silenceMs,
       this.lastMeaningfulFrameAt + heartbeatOnlyMs,
     );
-    this.streamHealthTimer = setTimeout(() => {
+    this.streamHealthTimer = this.streamHealthClock.setTimeout(() => {
       this.streamHealthTimer = undefined;
       const failFn = this.streamHealthFail;
       if (!failFn || this.expectedClose) return;
-      const stalledFor = Date.now() - this.lastInboundFrameAt;
-      const meaningfulStalledFor = Date.now() - this.lastMeaningfulFrameAt;
+      const stalledFor = this.streamHealthClock.now() - this.lastInboundFrameAt;
+      const meaningfulStalledFor = this.streamHealthClock.now() - this.lastMeaningfulFrameAt;
       if (stalledFor < silenceMs && meaningfulStalledFor < heartbeatOnlyMs) {
         // A frame landed between arming and firing — re-arm for the fresh deadline.
         this.armStreamHealthTimer(failFn);
@@ -906,7 +923,7 @@ class LiveCursorTransport implements CursorTransport {
    * heartbeat-only threshold.
    */
   private noteInboundFrame(livenessOnly: boolean): void {
-    const now = Date.now();
+    const now = this.streamHealthClock.now();
     this.lastInboundFrameAt = now;
     if (!livenessOnly) this.lastMeaningfulFrameAt = now;
     if (this.streamHealthFail) this.armStreamHealthTimer(this.streamHealthFail);
@@ -1288,7 +1305,7 @@ class LiveCursorTransport implements CursorTransport {
       const decodedUpdate = decoded.message.case === "interactionUpdate" ? decoded.message.value.message?.case : undefined;
       const livenessOnly = decodedUpdate === "heartbeat" || decoded.message.case === "conversationCheckpointUpdate";
       if (!this.streamHealthFail) {
-        const now = Date.now();
+        const now = this.streamHealthClock.now();
         this.lastInboundFrameAt = now;
         this.lastMeaningfulFrameAt = now;
         this.streamHealthFail = failAndClear;

src/adapters/cursor/transport.ts

@@ -55,6 +55,25 @@ export interface CursorTransportFactoryInput {
    * client thread. Distinct from the per-transport native-exec/shell owner.
    */
   sessionId?: string;
+  /**
+   * Test-only clock for the T04 inbound stream-health watchdog. Production omits it and the
+   * transport uses the global timers; injecting here must not change scheduling semantics.
+   *
+   * It exists because the watchdog's contract — a meaningful frame re-arms both deadlines — can
+   * only be asserted against real timers by keeping a synthetic server delivering frames faster
+   * than the silence budget for several multiples of it. That is an assertion about how busy the
+   * machine is, and it failed in the unsharded macOS lane for exactly that reason while the
+   * watchdog was working correctly. With the clock injected, virtual time advances only where
+   * the test says it does, so no amount of runner contention can expire a deadline.
+   */
+  streamHealthClock?: CursorStreamHealthClock;
+}
+
+/** The three time primitives the T04 watchdog uses. Production binds all three to the globals. */
+export interface CursorStreamHealthClock {
+  now(): number;
+  setTimeout(callback: () => void, ms: number): ReturnType<typeof setTimeout>;
+  clearTimeout(timer: ReturnType<typeof setTimeout>): void;
 }
 
 export type CursorTransportFactory = (input: CursorTransportFactoryInput) => CursorTransport;

structure/providers/cursor.md

@@ -219,3 +219,28 @@ Canonical Responses identity sanitation and narrowly scoped pre-output combo rec
 Upstream API-key usage follows the [physical-attempt account attribution contract](../gui-and-management-api.md#upstream-key-account-attribution), independently of subscription quota observations.
 
 Unicode pattern normalization uses [copy-on-write traversal](../transports/byte-accounting.md#unicode-pattern-normalization) while preserving the existing schema and wire semantics.
+
+## Inbound stream-health clock ownership
+
+The T04 inbound stream-health watchdog in `src/adapters/cursor/live-transport.ts` fails a turn that
+received its first frame and then went silent for `CURSOR_STREAM_SILENCE_FAIL_MS` (30s) or
+produced only liveness frames for `CURSOR_STREAM_HEARTBEAT_ONLY_FAIL_MS` (90s), instead of waiting
+out the 300s bridge stall watchdog. One timer covers both thresholds and re-arms on every
+decoded frame, so the deadline is always recomputed from the newest frame.
+
+Those two budgets are the production contract and a test does not move them to make itself
+pass. What a test may replace is the watchdog's time source: `streamHealthClock` on
+`CursorTransportFactoryInput` supplies `now`, `setTimeout` and `clearTimeout`, defaulting to the
+globals, and the seam is deliberately scoped to T04 alone — the first-frame timer, the
+turn-ended close grace, the client-tool finalize grace and the outbound heartbeat all stay on
+the global timers, as does the `elapsedMs` diagnostic, whose `turnStartedAt` is stamped on the
+wall clock.
+
+The seam exists because the re-arming half of the contract cannot be stated against real
+timers without also asserting that the machine keeps up: showing that a deadline did NOT expire
+means keeping a synthetic server ahead of the silence budget for several multiples of it, which
+is what failed in the unsharded macOS lane while the watchdog was correct. Scaling the budget
+lengthens the window rather than shrinking the exposure. The firing half needs no seam and is
+still covered on real timers in `tests/providers/cursor/cursor-stream-health.test.ts`: silence
+after the first frame, heartbeat-only traffic reaching the progress threshold, and `turnEnded`
+cancelling the watchdog while the server holds the stream open.

tests/providers/cursor/cursor-stream-health.test.ts

@@ -100,6 +100,52 @@ function runRequest(): CursorRunRequest {
   } as CursorRunRequest;
 }
 
+/**
+ * A clock the test advances by hand, for the T04 watchdog only.
+ *
+ * The watchdog's re-arming contract cannot be stated against real timers without also
+ * asserting that the machine keeps up: the only way to show a deadline did NOT expire is to
+ * keep a synthetic server delivering frames faster than the silence budget for several
+ * multiples of it, and a loaded runner that pauses longer than one budget fails the assertion
+ * while the watchdog is behaving correctly. Virtual time removes that term. Timers fire only
+ * from `advanceTo`, so contention can delay a frame's arrival — which the test waits for —
+ * without any deadline passing.
+ */
+function manualStreamHealthClock() {
+  type Pending = { at: number; callback: () => void };
+  const pending = new Map<number, Pending>();
+  let now = 0;
+  let nextHandle = 1;
+  return {
+    clock: {
+      now: () => now,
+      setTimeout: (callback: () => void, ms: number) => {
+        const handle = nextHandle++;
+        pending.set(handle, { at: now + Math.max(0, ms), callback });
+        return handle as unknown as ReturnType<typeof setTimeout>;
+      },
+      clearTimeout: (timer: ReturnType<typeof setTimeout>) => {
+        pending.delete(timer as unknown as number);
+      },
+    },
+    /** Move virtual time to `target`, running every timer due at or before it in deadline order. */
+    advanceTo(target: number): void {
+      for (;;) {
+        const due = [...pending.entries()]
+          .filter(([, timer]) => timer.at <= target)
+          .sort((left, right) => left[1].at - right[1].at)[0];
+        if (!due) break;
+        pending.delete(due[0]);
+        now = Math.max(now, due[1].at);
+        due[1].callback();
+      }
+      now = Math.max(now, target);
+    },
+    now: () => now,
+    armed: () => pending.size,
+  };
+}
+
 async function drain(
   baseUrl: string,
   knobs: { streamSilenceFailMs?: number; streamHeartbeatOnlyFailMs?: number },
@@ -138,7 +184,6 @@ describe("Cursor inbound stream-health watchdog (T04)", () => {
   // would collapse the two clocks to the same value in CI.
   const silenceMs = isolationBudgetMs(1_000);
   const heartbeatOnlyMs = 2 * silenceMs;
-  const progressDurationMs = 3 * silenceMs;
   // Include the existing two-second first-frame allowance and leave time for cleanup.
   const fixtureLimitMs = 4 * silenceMs + 2_000;
   const timeoutMs = Math.max(watchdogMs(15_000), fixtureLimitMs + silenceMs);
@@ -189,46 +234,82 @@ describe("Cursor inbound stream-health watchdog (T04)", () => {
   }, timeoutMs);
 
   test("meaningful frames keep resetting both clocks; turnEnded finishes cleanly", async () => {
-    let firstTextReceivedAt: number | undefined;
-    let completedProgressSpan = false;
+    // Virtual budgets: nothing here is scaled for CI, because no real interval has to beat them.
+    const virtualSilenceMs = 1_000;
+    const virtualHeartbeatOnlyMs = 2 * virtualSilenceMs;
+    const timing = manualStreamHealthClock();
+    const connected = Promise.withResolvers<http2.ServerHttp2Stream>();
+    const messages: CursorServerMessage[] = [];
+    const armedAfterFrame: number[] = [];
+    let armedAfterTurnEnded: number | undefined;
+    let failure: Error | undefined;
     await withH2Server(stream => {
       stream.on("error", () => {});
       stream.respond({ ":status": 200, "content-type": "application/connect+proto" });
-      stream.write(Buffer.from(textDeltaFrame("part-0")));
-      const latestEndAt = performance.now() + fixtureLimitMs;
-      let count = 0;
-      const tick = setInterval(() => {
-        count += 1;
-        try {
-          const now = performance.now();
-          const progressComplete = firstTextReceivedAt !== undefined
-            && now - firstTextReceivedAt >= progressDurationMs;
-          if (progressComplete || now >= latestEndAt) {
-            completedProgressSpan = progressComplete;
-            stream.write(Buffer.from(turnEndedFrame()));
-            stream.end();
-            clearInterval(tick);
-          } else {
-            stream.write(Buffer.from(textDeltaFrame(`part-${count}`)));
+      connected.resolve(stream);
+    }, async baseUrl => {
+      const transport = createLiveCursorTransport({
+        provider: { adapter: "cursor", baseUrl, apiKey: "test-token" },
+        translatorBudget: createTestTranslatorBudget(),
+        // Stays on the real clock: it owns dial and first response, not this contract. Generous
+        // so a slow local dial cannot pre-empt the case; a real hang is bounded by the timeout.
+        firstFrameTimeoutMs: watchdogMs(15_000),
+        streamSilenceFailMs: virtualSilenceMs,
+        streamHeartbeatOnlyFailMs: virtualHeartbeatOnlyMs,
+        streamHealthClock: timing.clock,
+      });
+      const iterator = transport.run(runRequest())[Symbol.asyncIterator]();
+      try {
+        // The first next() dials and puts the request on the wire.
+        const opened = iterator.next();
+        const server = await connected.promise;
+        const textFrame = async (label: string, pending?: Promise<IteratorResult<CursorServerMessage>>) => {
+          server.write(Buffer.from(textDeltaFrame(label)));
+          let result = await (pending ?? iterator.next());
+          while (!result.done && result.value.type !== "text") {
+            messages.push(result.value);
+            result = await iterator.next();
           }
-        } catch {
-          clearInterval(tick);
-          stream.destroy();
+          if (result.done) throw new Error(`stream ended before the ${label} text arrived`);
+          messages.push(result.value);
+          // Awaiting the yielded message is the synchronization: noteInboundFrame has run by the
+          // time the frame it decoded reaches the consumer, so the re-armed timer is observable.
+          armedAfterFrame.push(timing.armed());
+        };
+        await textFrame("part-0", opened);
+        // Every frame lands just under the deadline the previous one set, and the four of them
+        // carry virtual time past both non-resetting deadlines (S and 2S). That is the whole
+        // claim: each deadline is recomputed from the newest frame, not from the first one.
+        const landings = [900, 1_800, 2_700];
+        for (let index = 0; index < landings.length; index += 1) {
+          timing.advanceTo(landings[index]!);
+          await textFrame(`part-${index + 1}`);
         }
-      }, 100);
-      stream.on("close", () => clearInterval(tick));
-    }, async baseUrl => {
-      // Observe progress for 3S after receipt: both non-resetting deadlines (S and 2S)
-      // would expire before turnEnded, even when the first text reaches us late.
-      const { messages, failure } = await drain(baseUrl, {
-        streamSilenceFailMs: silenceMs,
-        streamHeartbeatOnlyFailMs: heartbeatOnlyMs,
-      }, () => { firstTextReceivedAt = performance.now(); });
-      expect(failure).toBeUndefined();
-      expect(completedProgressSpan).toBe(true);
-      expect(messages.some(message => message.type === "text")).toBe(true);
-      expect(messages.some(message => message.type === "done")).toBe(true);
+        server.write(Buffer.from(turnEndedFrame()));
+        server.end();
+        for (;;) {
+          const result = await iterator.next();
+          if (result.done) break;
+          messages.push(result.value);
+        }
+        // Read before close(), which would clear the timer on its own.
+        armedAfterTurnEnded = timing.armed();
+      } catch (err) {
+        failure = err instanceof Error ? err : new Error(String(err));
+      } finally {
+        await transport.close?.();
+      }
     });
+    expect(failure).toBeUndefined();
+    // One timer after every frame: the previous one was cleared rather than left stacked.
+    expect(armedAfterFrame).toEqual([1, 1, 1, 1]);
+    expect(timing.now()).toBeGreaterThan(virtualHeartbeatOnlyMs);
+    expect(messages.some(message => message.type === "text")).toBe(true);
+    expect(messages.some(message => message.type === "done")).toBe(true);
+    // turnEnded disarmed the watchdog, so no deadline survives the turn to fail it later.
+    expect(armedAfterTurnEnded).toBe(0);
+    timing.advanceTo(timing.now() + 10 * virtualHeartbeatOnlyMs);
+    expect(failure).toBeUndefined();
   }, timeoutMs);
 
   test("turnEnded disarms the watchdog even when the server holds the stream open", async () => {