From 994e5ba87145fa55b9af41f7fb382de20635b4d8 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sat, 22 Aug 2026 12:46:16 +0900 Subject: [PATCH] fix(cursor): fail silent and heartbeat-only streams at the transport instead of the 300s bridge watchdog T04 (devlog 260822_senpi_cursor_transfer/110, senpi #1062 second half): after the first decoded frame, 30s of inbound-frame silence or 90s of heartbeat/checkpoint-only traffic fails the turn with a typed stall error. Decoded frames reset the silence clock; only non-liveness frames reset the progress clock, so TLS keepalive noise cannot defeat the watchdog and a pinging-but-stuck server still fails at 90s. Disarmed by turnEnded (the T03 grace close owns the socket from there), client-tool suspend, protocol complete, and every settle path. Addresses the #2210 stall class. --- src/adapters/cursor/live-transport.ts | 118 ++++++++++++++- src/adapters/cursor/transport.ts | 10 ++ tests/cursor-stream-health.test.ts | 210 ++++++++++++++++++++++++++ 3 files changed, 336 insertions(+), 2 deletions(-) create mode 100644 tests/cursor-stream-health.test.ts diff --git a/src/adapters/cursor/live-transport.ts b/src/adapters/cursor/live-transport.ts index 9913b7defed..ad48ec67134 100644 --- a/src/adapters/cursor/live-transport.ts +++ b/src/adapters/cursor/live-transport.ts @@ -91,6 +91,18 @@ const CURSOR_RUN_PATH = "/agent.v1.AgentService/Run"; const CURSOR_CLIENT_VERSION = "cli-2026.07.08-0c04a8a"; const HEARTBEAT_MS = 5_000; const CURSOR_FIRST_FRAME_TIMEOUT_MS = 30_000; +/** + * T04 (senpi #1062 second half): after the first frame, a turn with NO inbound decoded + * frames for this long is failed instead of waiting for the 300s bridge stall watchdog + * (issue #2210). Reset on every decoded AgentServerMessage. + */ +const CURSOR_STREAM_SILENCE_FAIL_MS = 30_000; +/** + * A stream that produces ONLY liveness frames (server heartbeat / conversationCheckpointUpdate) + * for this long is equally stuck — the server is alive but the turn is not progressing. + * Reset on every decoded frame that is not liveness-only. + */ +const CURSOR_STREAM_HEARTBEAT_ONLY_FAIL_MS = 90_000; /** * 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 @@ -421,6 +433,17 @@ class LiveCursorTransport implements CursorTransport { private heartbeat?: ReturnType; private firstFrameTimer?: ReturnType; private turnEndedCloseTimer?: ReturnType; + /** + * T04 inbound stream-health watchdog. Armed after the request is on the wire, reset by + * every DECODED frame (raw chunks deliberately do not count — TLS keepalive noise must not + * defeat it), disarmed by any settle/expected-close path. One timer covers both thresholds: + * it always fires at min(lastInbound + silence, lastMeaningful + heartbeatOnly) and re-arms + * when neither deadline has actually elapsed. + */ + private streamHealthTimer?: ReturnType; + private lastInboundFrameAt = 0; + private lastMeaningfulFrameAt = 0; + private streamHealthFail?: (error: Error) => void; private committed = false; private expectedClose = false; /** @@ -760,6 +783,73 @@ class LiveCursorTransport implements CursorTransport { } } + private clearStreamHealthTimer(): void { + if (this.streamHealthTimer) { + clearTimeout(this.streamHealthTimer); + this.streamHealthTimer = undefined; + } + this.streamHealthFail = undefined; + } + + /** + * T04: arm (or re-arm) the inbound stream-health watchdog. `fail` is the turn's + * 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. + */ + private armStreamHealthTimer(fail: (error: Error) => void): void { + if (this.streamHealthTimer) 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 deadline = Math.min( + this.lastInboundFrameAt + silenceMs, + this.lastMeaningfulFrameAt + heartbeatOnlyMs, + ); + this.streamHealthTimer = 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; + if (stalledFor < silenceMs && meaningfulStalledFor < heartbeatOnlyMs) { + // A frame landed between arming and firing — re-arm for the fresh deadline. + this.armStreamHealthTimer(failFn); + return; + } + const heartbeatOnly = stalledFor < silenceMs; + debugProviderDiagnostic("cursor", "stream-health-timeout", { + stalledMs: stalledFor, + meaningfulStalledMs: meaningfulStalledFor, + heartbeatOnly, + framesReceived: this.framesReceived, + elapsedMs: Date.now() - this.turnStartedAt, + }); + const reason = heartbeatOnly + ? `Cursor stream stalled: heartbeat-only traffic for ${Math.round(meaningfulStalledFor / 1000)}s without turn progress` + : `Cursor stream stalled: no inbound frames for ${Math.round(stalledFor / 1000)}s before turnEnded`; + failFn(new Error(reason)); + try { this.stream?.close(); } catch { this.stream?.destroy(); } + this.session?.close(); + this.http1Connection?.close(); + }, Math.max(0, deadline - now)); + } + + /** + * T04: record a decoded inbound frame. Liveness-only frames (server heartbeat, + * conversationCheckpointUpdate) keep the silence clock fresh but not the progress + * clock — matching senpi's split so a server that only pings still fails at the + * heartbeat-only threshold. + */ + private noteInboundFrame(livenessOnly: boolean): void { + const now = Date.now(); + this.lastInboundFrameAt = now; + if (!livenessOnly) this.lastMeaningfulFrameAt = now; + if (this.streamHealthFail) this.armStreamHealthTimer(this.streamHealthFail); + } + /** * A clean Connect END_STREAM owns the turn terminal even when Cursor keeps the * HTTP body open or tears it down with an abort/reset immediately afterward. @@ -774,6 +864,7 @@ class LiveCursorTransport implements CursorTransport { this.heartbeat = undefined; } this.clearFirstFrameTimer(); + this.clearStreamHealthTimer(); } private startShellCleanup(): Promise { @@ -785,6 +876,7 @@ class LiveCursorTransport implements CursorTransport { if (this.turnEndedCloseTimer) clearTimeout(this.turnEndedCloseTimer); this.clearPendingFinalize(); this.clearFirstFrameTimer(); + this.clearStreamHealthTimer(); this.stream?.close(); this.session?.close(); this.http1Connection?.close(); @@ -800,6 +892,7 @@ class LiveCursorTransport implements CursorTransport { this.clearPendingFinalize(); if (this.heartbeat) clearInterval(this.heartbeat); this.clearFirstFrameTimer(); + this.clearStreamHealthTimer(); if (this.http1Connection) { this.http1Connection.close(); } else { @@ -825,6 +918,10 @@ class LiveCursorTransport implements CursorTransport { */ private closeAfterTurnEnded(): void { if (this.turnEndedCloseTimer) return; + // The application turn is over: the T03 grace timer owns the socket from here. + // The T04 watchdog must disarm NOW, not at the grace close — a watchdog shorter + // than the grace would otherwise fail a completed turn. + this.clearStreamHealthTimer(); this.turnEndedCloseTimer = setTimeout(() => { this.turnEndedCloseTimer = undefined; // Only expectedClose (client-tool suspend cancel) blocks the close. @@ -839,6 +936,7 @@ class LiveCursorTransport implements CursorTransport { }); this.expectedClose = true; this.clearFirstFrameTimer(); + this.clearStreamHealthTimer(); if (this.heartbeat) clearInterval(this.heartbeat); if (this.http1Connection) { this.http1Connection.close(); @@ -962,7 +1060,10 @@ class LiveCursorTransport implements CursorTransport { const settler = createTerminalSettler({ fail, finish, - clearTimer: () => this.clearFirstFrameTimer(), + clearTimer: () => { + this.clearFirstFrameTimer(); + this.clearStreamHealthTimer(); + }, }); const failAndClear = (error: Error) => { releaseBacklogLease(); @@ -1093,7 +1194,20 @@ class LiveCursorTransport implements CursorTransport { settler.settleFinish(); return; } - await this.handleServerMessage(fromBinary(AgentServerMessageSchema, frame.payload), state, push); + const decoded = fromBinary(AgentServerMessageSchema, frame.payload); + // T04: every decoded frame refreshes the silence clock; only non-liveness frames + // refresh the progress clock. First decoded frame arms the watchdog (the first-frame + // timer owned everything before this point). + 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(); + this.lastInboundFrameAt = now; + this.lastMeaningfulFrameAt = now; + this.streamHealthFail = failAndClear; + } + this.noteInboundFrame(livenessOnly); + await this.handleServerMessage(decoded, state, push); }; const drainPendingFrames = () => { const availableSlots = CURSOR_MAX_PENDING_FRAMES - this.pendingTransportFrames; diff --git a/src/adapters/cursor/transport.ts b/src/adapters/cursor/transport.ts index 81b924c90ba..79f241ca0e4 100644 --- a/src/adapters/cursor/transport.ts +++ b/src/adapters/cursor/transport.ts @@ -29,6 +29,16 @@ export interface CursorTransportFactoryInput { firstFrameTimeoutMs?: number; /** Grace (ms) between close() and the force-destroy fallback after a first-frame timeout. Defaults to 1s. */ timeoutDestroyGraceMs?: number; + /** + * T04 watchdog: maximum inbound decoded-frame silence (ms) after the first frame before the + * turn is failed. Defaults to 30s. + */ + streamSilenceFailMs?: number; + /** + * T04 watchdog: maximum heartbeat/checkpoint-only traffic (ms) without turn progress before + * the turn is failed. Defaults to 90s. + */ + streamHeartbeatOnlyFailMs?: number; /** * Grace window (ms) before a drained client-tool turn is finalized, so a sibling tool call * announced in a later receive chunk can revoke a premature finalize. Defaults to 50ms. diff --git a/tests/cursor-stream-health.test.ts b/tests/cursor-stream-health.test.ts new file mode 100644 index 00000000000..d59c1147ae6 --- /dev/null +++ b/tests/cursor-stream-health.test.ts @@ -0,0 +1,210 @@ +import http2 from "node:http2"; +import { create, toBinary } from "@bufbuild/protobuf"; +import { describe, expect, test } from "bun:test"; +import { + AgentServerMessageSchema, + ConversationStateStructureSchema, + HeartbeatUpdateSchema, + InteractionUpdateSchema, + TextDeltaUpdateSchema, + TurnEndedUpdateSchema, +} from "../src/adapters/cursor/gen/agent_pb"; +import { encodeConnectFrame } from "../src/adapters/cursor/framing"; +import { createLiveCursorTransport } from "../src/adapters/cursor/live-transport"; +import { createTestTranslatorBudget } from "./helpers/translator-budget"; +import type { CursorRunRequest, CursorServerMessage } from "../src/adapters/cursor/types"; + +/** + * T04 (devlog 260822_senpi_cursor_transfer/110): inbound stream-health watchdog. + * A turn that received its first frame but then goes silent (or heartbeat-only) + * must fail at the transport with a typed stall error instead of waiting for the + * 300s bridge stall watchdog (issue #2210 class). + */ + +function agentFrame(message: Parameters>[1]): Uint8Array { + return encodeConnectFrame(toBinary(AgentServerMessageSchema, create(AgentServerMessageSchema, message))); +} + +function textDeltaFrame(textValue: string): Uint8Array { + return agentFrame({ + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "textDelta", value: create(TextDeltaUpdateSchema, { text: textValue }) }, + }), + }, + }); +} + +function heartbeatFrame(): Uint8Array { + return agentFrame({ + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "heartbeat", value: create(HeartbeatUpdateSchema, {}) }, + }), + }, + }); +} + +function checkpointFrame(): Uint8Array { + return agentFrame({ + message: { + case: "conversationCheckpointUpdate", + value: create(ConversationStateStructureSchema, {}), + }, + }); +} + +function turnEndedFrame(): Uint8Array { + return agentFrame({ + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "turnEnded", value: create(TurnEndedUpdateSchema, {}) }, + }), + }, + }); +} + +async function withH2Server( + handler: (stream: http2.ServerHttp2Stream) => void, + run: (baseUrl: string) => Promise, +): Promise { + const server = http2.createServer(); + server.on("stream", handler); + await new Promise((resolve, reject) => { + const onError = (error: Error) => reject(error); + server.once("error", onError); + server.listen(0, "127.0.0.1", () => { + server.off("error", onError); + resolve(); + }); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("HTTP/2 fixture did not bind a TCP port"); + try { + return await run(`http://127.0.0.1:${address.port}`); + } finally { + await new Promise(resolve => server.close(() => resolve())); + } +} + +function runRequest(): CursorRunRequest { + return { + modelId: "composer-2", + conversationId: "cursor_stream_health_test", + system: [], + messages: [{ role: "user", content: "hello" }], + } as CursorRunRequest; +} + +async function drain(baseUrl: string, knobs: { streamSilenceFailMs?: number; streamHeartbeatOnlyFailMs?: number }): Promise<{ + messages: CursorServerMessage[]; + failure?: Error; +}> { + const transport = createLiveCursorTransport({ + provider: { adapter: "cursor", baseUrl, apiKey: "test-token" }, + translatorBudget: createTestTranslatorBudget(), + firstFrameTimeoutMs: 2_000, + ...knobs, + }); + const messages: CursorServerMessage[] = []; + let failure: Error | undefined; + try { + for await (const message of transport.run(runRequest())) messages.push(message); + } catch (err) { + failure = err instanceof Error ? err : new Error(String(err)); + } finally { + await transport.close?.(); + } + return { messages, failure }; +} + +describe("Cursor inbound stream-health watchdog (T04)", () => { + test("silence after the first frame fails the turn with the stall error", async () => { + 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: 300, streamHeartbeatOnlyFailMs: 10_000 }); + expect(failure).toBeDefined(); + expect(failure!.message).toContain("no inbound frames"); + }); + }, 15_000); + + test("heartbeat-only traffic survives the silence threshold but fails at the heartbeat-only threshold", async () => { + await withH2Server(stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + stream.write(Buffer.from(textDeltaFrame("hi"))); + const ping = setInterval(() => { + try { + stream.write(Buffer.from(heartbeatFrame())); + stream.write(Buffer.from(checkpointFrame())); + } catch { clearInterval(ping); } + }, 100); + stream.on("close", () => clearInterval(ping)); + }, async baseUrl => { + const { failure } = await drain(baseUrl, { streamSilenceFailMs: 400, streamHeartbeatOnlyFailMs: 900 }); + expect(failure).toBeDefined(); + expect(failure!.message).toContain("heartbeat-only"); + }); + }, 15_000); + + test("meaningful frames keep resetting both clocks; turnEnded finishes cleanly", async () => { + await withH2Server(stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + let count = 0; + const tick = setInterval(() => { + count += 1; + try { + if (count < 6) { + stream.write(Buffer.from(textDeltaFrame(`part-${count}`))); + } else { + stream.write(Buffer.from(turnEndedFrame())); + stream.end(); + clearInterval(tick); + } + } catch { clearInterval(tick); } + }, 150); + stream.on("close", () => clearInterval(tick)); + }, async baseUrl => { + // Each 150ms text delta must reset the 400ms silence clock: six ticks ≈ 900ms total, + // far past a NON-resetting 400ms deadline. + const { messages, failure } = await drain(baseUrl, { streamSilenceFailMs: 400, streamHeartbeatOnlyFailMs: 10_000 }); + expect(failure).toBeUndefined(); + expect(messages.some(message => message.type === "text")).toBe(true); + expect(messages.some(message => message.type === "done")).toBe(true); + }); + }, 15_000); + + test("turnEnded disarms the watchdog even when the server holds the stream open", async () => { + await withH2Server(stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + stream.write(Buffer.from(textDeltaFrame("hi"))); + stream.write(Buffer.from(turnEndedFrame())); + // hold open: the T03 turnEnded close owns this case; the watchdog must not fire first + }, async baseUrl => { + const { messages, failure } = await drain(baseUrl, { streamSilenceFailMs: 300, streamHeartbeatOnlyFailMs: 10_000 }); + expect(failure).toBeUndefined(); + expect(messages.some(message => message.type === "done")).toBe(true); + }); + }, 15_000); + + test("no watchdog before the first frame: the first-frame timeout still owns dial silence", async () => { + await withH2Server(stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + // no frames at all + }, async baseUrl => { + const { failure } = await drain(baseUrl, { streamSilenceFailMs: 60_000, streamHeartbeatOnlyFailMs: 60_000 }); + expect(failure).toBeDefined(); + expect(failure!.message).toContain("before first response"); + }); + }, 15_000); +});