From 072271a6d358b96e79118efef7c5994ce24878cc Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 19 Sep 2026 17:12:43 +0900 Subject: [PATCH] fix(cursor): state the T04 re-arming contract on an injectable clock tests/providers/cursor/cursor-stream-health.test.ts asserted that meaningful frames keep resetting both stream-health deadlines by running a synthetic HTTP/2 server for three times the silence budget and requiring that no deadline expire. That is an assertion about how busy the machine is. In the unsharded macOS control lane it fails with "Cursor stream stalled: no inbound frames for 5s before turnEnded" while the watchdog is behaving correctly: under contention the fixture's own frames do not arrive inside the window. The budget is already CI-scaled, and scaling is what makes it worse. With the silence budget floored at 5s the case needs 15 consecutive seconds during which no 5s gap in frame decoding occurs, so raising the floor lengthens the exposure rather than reducing it. Add a T04-only clock seam (now, setTimeout, clearTimeout), matching StageLeaseClock in src/cli/account-api.ts. Production omits it and keeps the globals; the 30s and 90s budgets, the first-frame timer, the turn-ended grace and the outbound heartbeat are untouched. The test now advances virtual time by hand between frames whose arrival it awaits, so runner contention can delay a frame without any deadline passing, and it additionally asserts that exactly one timer is armed after each frame and none survives turnEnded. No production timeout changed, no budget widened, no retry added and no platform skipped. --- src/adapters/cursor/live-transport.ts | 35 ++-- src/adapters/cursor/transport.ts | 19 +++ structure/providers/cursor.md | 25 +++ .../cursor/cursor-stream-health.test.ts | 151 ++++++++++++++---- 4 files changed, 186 insertions(+), 44 deletions(-) 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 () => {