diff --git a/structure/providers/cursor.md b/structure/providers/cursor.md index af22677eb55..5278df9ead2 100644 --- a/structure/providers/cursor.md +++ b/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. diff --git a/tests/providers/cursor/cursor-stream-health.test.ts b/tests/providers/cursor/cursor-stream-health.test.ts index 56c829e5184..b36bb036646 100644 --- a/tests/providers/cursor/cursor-stream-health.test.ts +++ b/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(); + 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(); + 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>, + ) => { + 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 => {