diff --git a/src/adapters/cursor/live-transport.ts b/src/adapters/cursor/live-transport.ts index f5ad317d051..e0b6b527c42 100644 --- a/src/adapters/cursor/live-transport.ts +++ b/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; diff --git a/src/adapters/cursor/transport.ts b/src/adapters/cursor/transport.ts index 79f241ca0e4..5cb3c099c8a 100644 --- a/src/adapters/cursor/transport.ts +++ b/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; + clearTimeout(timer: ReturnType): void; } export type CursorTransportFactory = (input: CursorTransportFactoryInput) => CursorTransport; diff --git a/structure/providers/cursor.md b/structure/providers/cursor.md index 8be236cd916..af22677eb55 100644 --- a/structure/providers/cursor.md +++ b/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. diff --git a/tests/providers/cursor/cursor-stream-health.test.ts b/tests/providers/cursor/cursor-stream-health.test.ts index 27a6cdac432..56c829e5184 100644 --- a/tests/providers/cursor/cursor-stream-health.test.ts +++ b/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(); + 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; + }, + clearTimeout: (timer: ReturnType) => { + 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(); + 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>) => { + 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 () => {