test(cursor): state the T04 heartbeat-only contract on the injectable clock

structure/providers/cursor.md

@@ -240,7 +240,13 @@ The seam exists because the re-arming half of the contract cannot be stated agai
 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.
+lengthens the window rather than shrinking the exposure. Every claim of that shape therefore lives
+under the seam in `tests/providers/cursor/cursor-stream-health.test.ts`: that meaningful frames
+re-arm both clocks, that liveness-only frames refresh the silence clock while the progress clock
+still expires, and that the silence deadline is the one that fires when it is the earlier of the
+two. That last one is load-bearing: a watchdog that dropped the `min()` and read only the progress
+deadline would relax silence detection from 30s to 90s while every real-timer case stayed green,
+because a later deadline still produces the same message. The firing half needs no seam and stays
+on real timers in the same file: silence after the first frame, the progress budget alone failing a
+turn when the silence budget is out of reach, and `turnEnded` cancelling the watchdog while the
+server holds the stream open.

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

@@ -12,7 +12,7 @@ import {
 import { encodeConnectFrame } from "../../../src/adapters/cursor/framing";
 import { createLiveCursorTransport } from "../../../src/adapters/cursor/live-transport";
 import { createTestTranslatorBudget } from "../../helpers/translator-budget";
-import { isolationBudgetMs, watchdogMs } from "../../helpers/ci-watchdog";
+import { watchdogMs } from "../../helpers/ci-watchdog";
 import type { CursorRunRequest, CursorServerMessage } from "../../../src/adapters/cursor/types";
 
 /**
@@ -130,7 +130,13 @@ function manualStreamHealthClock() {
     },
     /** Move virtual time to `target`, running every timer due at or before it in deadline order. */
     advanceTo(target: number): void {
-      for (;;) {
+      // The bound is deliberate. A watchdog that re-armed an ALREADY expired deadline would keep
+      // scheduling a zero-delay timer, and this loop is synchronous, so Bun's per-test timeout could
+      // never interrupt it: one mutation would wedge the whole lane instead of reddening one case.
+      for (let fired = 0; ; fired += 1) {
+        if (fired > 1_000) {
+          throw new Error("stream-health clock fired 1000 timers without draining: an expired deadline is being re-armed");
+        }
         const due = [...pending.entries()]
           .filter(([, timer]) => timer.at <= target)
           .sort((left, right) => left[1].at - right[1].at)[0];
@@ -180,13 +186,9 @@ async function drain(
 }
 
 describe("Cursor inbound stream-health watchdog (T04)", () => {
-  // Scale once: the load helper applies a floor, so scaling each deadline separately
-  // would collapse the two clocks to the same value in CI.
-  const silenceMs = isolationBudgetMs(1_000);
-  const heartbeatOnlyMs = 2 * 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);
+  // Bounds a hung case; no assertion here is stated against elapsed real time, so this
+  // never has to cover a synthetic server outrunning a deadline.
+  const caseTimeoutMs = watchdogMs(15_000);
 
   test("silence after the first frame fails the turn with the stall error", async () => {
     await withH2Server(stream => {
@@ -201,37 +203,185 @@ describe("Cursor inbound stream-health watchdog (T04)", () => {
     });
   }, 15_000);
 
-  test("heartbeat-only traffic survives the silence threshold but fails at the heartbeat-only threshold", async () => {
+  test("the silence deadline wins when it is the earlier of the two", async () => {
+    // The case above proves a genuine stall fails the turn on real timers; it cannot prove WHICH
+    // deadline did it, because a longer one arriving later still produces the same message inside
+    // the case timeout. That distinction is the `min(silence, progress)` rule, and dropping it
+    // would relax production silence detection from 30s to 90s while every real-timer case stayed
+    // green. Virtual time pins it: the turn must fail exactly at S, with nothing left armed.
+    const virtualSilenceMs = 1_000;
+    const virtualHeartbeatOnlyMs = 10 * virtualSilenceMs;
+    const timing = manualStreamHealthClock();
+    const connected = Promise.withResolvers<http2.ServerHttp2Stream>();
+    let armedBeforeDeadline: number | undefined;
+    let armedAtDeadline: 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("hi")));
-      // Frequent heartbeats/checkpoints keep the silence clock fresh while the
-      // longer heartbeat-only clock must still expire under a loaded test runner.
-      const ping = setInterval(() => {
-        try {
-          stream.write(Buffer.from(heartbeatFrame()));
-          stream.write(Buffer.from(checkpointFrame()));
-        } catch { clearInterval(ping); }
-      }, 40);
-      const limit = setTimeout(() => stream.close(), fixtureLimitMs);
-      stream.on("close", () => {
-        clearInterval(ping);
-        clearTimeout(limit);
+      connected.resolve(stream);
+    }, async baseUrl => {
+      const transport = createLiveCursorTransport({
+        provider: { adapter: "cursor", baseUrl, apiKey: "test-token" },
+        translatorBudget: createTestTranslatorBudget(),
+        firstFrameTimeoutMs: watchdogMs(15_000),
+        streamSilenceFailMs: virtualSilenceMs,
+        streamHeartbeatOnlyFailMs: virtualHeartbeatOnlyMs,
+        streamHealthClock: timing.clock,
       });
+      const iterator = transport.run(runRequest())[Symbol.asyncIterator]();
+      try {
+        const opened = iterator.next();
+        const server = await connected.promise;
+        server.write(Buffer.from(textDeltaFrame("hi")));
+        let first = await opened;
+        while (!first.done && first.value.type !== "text") first = await iterator.next();
+        if (first.done) throw new Error("stream ended before the first text arrived");
+        // then: silence. One frame stamped both clocks, so S is strictly the earlier deadline.
+        timing.advanceTo(virtualSilenceMs - 1);
+        armedBeforeDeadline = timing.armed();
+        timing.advanceTo(virtualSilenceMs);
+        armedAtDeadline = timing.armed();
+        // Cross the progress deadline too, so a watchdog that ignored S reports itself instead of
+        // leaving this case to hang to its own timeout.
+        timing.advanceTo(virtualHeartbeatOnlyMs + 1);
+        for (;;) {
+          const result = await iterator.next();
+          if (result.done) break;
+        }
+      } catch (err) {
+        failure = err instanceof Error ? err : new Error(String(err));
+      } finally {
+        await transport.close?.();
+      }
+    });
+    expect(failure).toBeDefined();
+    expect(failure!.message).toContain("no inbound frames");
+    expect(failure!.message).not.toContain("heartbeat-only");
+    // The watchdog identifies itself: the two branches share this prefix.
+    expect(failure!.message).toContain("Cursor stream stalled");
+    // Armed one tick before S and gone at S: the deadline was S, not the progress budget.
+    expect(armedBeforeDeadline).toBe(1);
+    expect(armedAtDeadline).toBe(0);
+  }, caseTimeoutMs);
+
+  test("liveness-only traffic keeps the silence clock fresh and still fails at the heartbeat-only threshold", async () => {
+    // Which clock expires is the contract. Stating it against real timers also states that a real
+    // interval outran a real deadline: the case had to keep liveness frames arriving with no gap
+    // longer than the silence budget for the whole heartbeat-only window, and a runner that pauses
+    // longer than one budget made the SILENCE watchdog win while both watchdogs behaved correctly.
+    // Virtual time removes that term — timers fire only from `advanceTo`, so contention can delay a
+    // frame's arrival (which this case waits for) without any deadline passing. Same seam #5131 used
+    // for the re-arming case below; production budgets and every other timer are untouched.
+    const virtualSilenceMs = 1_000;
+    const virtualHeartbeatOnlyMs = 2 * virtualSilenceMs;
+    const timing = manualStreamHealthClock();
+    const connected = Promise.withResolvers<http2.ServerHttp2Stream>();
+    const messages: CursorServerMessage[] = [];
+    const armedAfterLiveness: number[] = [];
+    let armedAfterDeadline: number | undefined;
+    let failure: Error | undefined;
+    await withH2Server(stream => {
+      stream.on("error", () => {});
+      stream.respond({ ":status": 200, "content-type": "application/connect+proto" });
+      connected.resolve(stream);
     }, async baseUrl => {
-      const { failure } = await drain(baseUrl, {
-        streamSilenceFailMs: silenceMs,
-        streamHeartbeatOnlyFailMs: heartbeatOnlyMs,
+      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.
+        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 nextOfType = async (
+          type: CursorServerMessage["type"],
+          pending?: Promise<IteratorResult<CursorServerMessage>>,
+        ) => {
+          let result = await (pending ?? iterator.next());
+          while (!result.done && result.value.type !== type) {
+            messages.push(result.value);
+            result = await iterator.next();
+          }
+          if (result.done) throw new Error(`stream ended before a ${type} message arrived`);
+          messages.push(result.value);
+        };
+        // One meaningful frame stamps both clocks and arms the watchdog.
+        server.write(Buffer.from(textDeltaFrame("hi")));
+        await nextOfType("text", opened);
+        // Advance BEFORE writing so each pair lands at that virtual instant, just under the silence
+        // deadline the previous frame set. A checkpoint is a progress frame, so it yields a
+        // `heartbeat` message; the bare heartbeat frame yields nothing outward. The frame chain is
+        // serialized, so awaiting the checkpoint's message proves both were decoded and the
+        // watchdog re-armed from the later of them.
+        for (const landing of [900, 1_800]) {
+          timing.advanceTo(landing);
+          server.write(Buffer.from(heartbeatFrame()));
+          server.write(Buffer.from(checkpointFrame()));
+          await nextOfType("heartbeat");
+          armedAfterLiveness.push(timing.armed());
+        }
+        // Silence was refreshed at 1800 and the progress clock never was, so 2S can only be the
+        // heartbeat-only deadline.
+        timing.advanceTo(virtualHeartbeatOnlyMs);
+        armedAfterDeadline = timing.armed();
+        // Had liveness frames wrongly refreshed the progress clock, nothing fires at 2S and the only
+        // surviving deadline is silence at 1800 + S. Cross it so this case reports which watchdog won
+        // instead of hanging to its own timeout.
+        timing.advanceTo(1_800 + virtualSilenceMs + 100);
+        for (;;) {
+          const result = await iterator.next();
+          if (result.done) break;
+          messages.push(result.value);
+        }
+      } catch (err) {
+        failure = err instanceof Error ? err : new Error(String(err));
+      } finally {
+        await transport.close?.();
+      }
+    });
+    expect(failure).toBeDefined();
+    // Say which watchdog won. A bare toContain reported only the expected substring, which reads as
+    // "the heartbeat-only watchdog is broken" when the real story was the silence watchdog firing first.
+    expect(failure!.message).toContain("heartbeat-only");
+    expect(failure!.message).not.toContain("no inbound frames");
+    expect(failure!.message).toContain("Cursor stream stalled");
+    // The heartbeat-only deadline fired at exactly 2S: nothing was left armed behind it.
+    expect(armedAfterDeadline).toBe(0);
+    // Liveness frames refreshed the silence clock and left one timer armed, never a stacked pair.
+    expect(armedAfterLiveness).toEqual([1, 1]);
+    expect(messages.some(message => message.type === "heartbeat")).toBe(true);
+  }, caseTimeoutMs);
+
+  test("the progress clock fires on real timers when the silence budget is out of reach", async () => {
+    // The firing half needs no seam. With a silence budget two orders of magnitude beyond the
+    // progress budget, the only deadline in reach is the progress one, so ordinary load delays this
+    // case rather than changing its outcome: a pause long enough to cross 60s of silence as well,
+    // and so be reported as silence instead, has already blown the case's own 15s limit. The
+    // production default clock (Date.now plus the global timers) stays on the heartbeat-only path.
+    // What this pins is that the progress budget alone can fail a turn through that clock, and that
+    // the reported branch is selected by which budget was crossed
+    // rather than by whether liveness frames were seen. What it cannot state is stated under the
+    // injected clock above: that the deadline is the minimum of both clocks, and that liveness
+    // frames refresh the silence clock. Both of those are "a deadline did not expire" claims.
+    await withH2Server(stream => {
+      stream.on("error", () => {});
+      stream.respond({ ":status": 200, "content-type": "application/connect+proto" });
+      stream.write(Buffer.from(textDeltaFrame("hi")));
+      // then: silence — never end the stream
+    }, async baseUrl => {
+      const { failure } = await drain(baseUrl, { streamSilenceFailMs: 60_000, streamHeartbeatOnlyFailMs: 600 });
       expect(failure).toBeDefined();
-      // Assert on the message, and say which watchdog won when the wrong one does.
-      // A bare toContain here reported only the expected substring, which reads as
-      // "the heartbeat-only watchdog is broken" when the real story is that the
-      // silence watchdog fired first on a loaded runner.
       expect(failure!.message).toContain("heartbeat-only");
+      expect(failure!.message).not.toContain("no inbound frames");
     });
-  }, timeoutMs);
+  }, 15_000);
 
   test("meaningful frames keep resetting both clocks; turnEnded finishes cleanly", async () => {
     // Virtual budgets: nothing here is scaled for CI, because no real interval has to beat them.
@@ -310,7 +460,7 @@ describe("Cursor inbound stream-health watchdog (T04)", () => {
     expect(armedAfterTurnEnded).toBe(0);
     timing.advanceTo(timing.now() + 10 * virtualHeartbeatOnlyMs);
     expect(failure).toBeUndefined();
-  }, timeoutMs);
+  }, caseTimeoutMs);
 
   test("turnEnded disarms the watchdog even when the server holds the stream open", async () => {
     await withH2Server(stream => {