diff --git a/src/adapters/cursor/live-transport.ts b/src/adapters/cursor/live-transport.ts index 83658837bf..61f2e6e0a8 100644 --- a/src/adapters/cursor/live-transport.ts +++ b/src/adapters/cursor/live-transport.ts @@ -753,6 +753,22 @@ class LiveCursorTransport implements CursorTransport { } } + /** + * 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. + * Stop client-side liveness work and classify that later transport close as + * expected without actively sending an RST_STREAM back to Cursor. + */ + private markProtocolComplete(): void { + this.expectedClose = true; + this.clearPendingFinalize(); + if (this.heartbeat) { + clearInterval(this.heartbeat); + this.heartbeat = undefined; + } + this.clearFirstFrameTimer(); + } + private startShellCleanup(): Promise { return this.shellCleanup ??= terminateBackgroundShellsForSession(this.shellOwnerId); } @@ -1000,7 +1016,38 @@ class LiveCursorTransport implements CursorTransport { framesReceived: this.framesReceived, elapsedMs: Date.now() - this.turnStartedAt, } : { framesReceived: this.framesReceived, elapsedMs: Date.now() - this.turnStartedAt }); - if (endError) failAndClear(endError); + if (endError) { + failAndClear(endError); + return; + } + // Connect's clean END_STREAM envelope is the protocol terminal. Cursor's RunSSE body can + // remain open after this frame (or close through an AbortError), so waiting for HTTP EOF + // strands an otherwise completed turn until the outer bridge stall watchdog fires. + // + // Earlier frames in this serialized frameWork chain have already run. Preserve their real + // turnEnded terminal when present; otherwise finalize the clean protocol end once so open + // tool calls still fail closed, a text-only turn receives its normal done event, and a + // drained client-tool turn does not lose the pending terminal when protocol cleanup clears + // its grace timer. + const hasPendingClientToolFinalization = this.pendingFinalize !== undefined; + if ( + !this.expectedClose + && !state.terminated + && !this.emittedTerminal + && ( + state.openToolCalls.size > 0 + || this.sawAssistantText + || hasPendingClientToolFinalization + ) + ) { + const terminal = hasPendingClientToolFinalization && state.openToolCalls.size === 0 + ? finalizeAfterDrain(state) + : finalizeTurnEvents(state); + for (const event of terminal) push(event); + } + this.markProtocolComplete(); + releaseBacklogLease(); + settler.settleFinish(); return; } await this.handleServerMessage(fromBinary(AgentServerMessageSchema, frame.payload), state, push); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index b4df17b72a..4fa7eb994e 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -101,6 +101,14 @@ alone never opt a gateway in. and before the `/v1/*` guard. Unknown `/v1/*` paths return JSON 404 errors instead of falling through to GUI static serving. +[Decision Log] +- 목적과 의도: Complete Cursor turns at the protocol terminal instead of waiting for a separate HTTP-body EOF that may never arrive. +- 기존 구현 및 제약 조건: Cursor can send turnEnded followed by a clean Connect END_STREAM envelope while RunSSE remains open or later closes through an abort-shaped transport error. The adapter logged the clean envelope but did not settle its terminal owner, so a completed-looking turn could remain open until the Responses stall watchdog. +- 검토한 주요 대안: Shorten the global stall timeout; treat every later abort as success; settle only when the HTTP stream emits end; make the clean Connect envelope authoritative. +- 선택한 방식: Process preceding frames in order, preserve an already-emitted terminal, run any already-armed drained client-tool finalizer before protocol cleanup clears its grace timer only while the call set is still drained, otherwise finalize once through the existing fail-closed tool-call logic, and settle the transport successfully on a clean Connect END_STREAM. +- 다른 대안 대신 이 방식을 선택한 이유: The protocol envelope is upstream's explicit terminal signal. Timeout changes only hide the race, and globally swallowing aborts would mask genuine mid-turn cancellation. +- 장점, 단점 및 영향: Completed Cursor responses no longer wait for the 300-second watchdog when the HTTP body stays open; incomplete tool calls still emit their existing truncation error, and error-bearing Connect terminals remain failures. + A replayed compaction item carries an `encrypted_content` blob only its minting backend can decode, and the client replays it on every later turn. The proxy's own `ocx1:` envelopes are transparent base64, so they always lower to plain user messages. A native blob is relayed only when there is no diff --git a/tests/cursor-eof-terminal.test.ts b/tests/cursor-eof-terminal.test.ts index a2a59b2262..18d4da50fa 100644 --- a/tests/cursor-eof-terminal.test.ts +++ b/tests/cursor-eof-terminal.test.ts @@ -3,6 +3,7 @@ import { create, toBinary } from "@bufbuild/protobuf"; import { describe, expect, test } from "bun:test"; import { AgentServerMessageSchema, + ExecServerMessageSchema, InteractionUpdateSchema, McpArgsSchema, McpToolCallSchema, @@ -63,6 +64,29 @@ function toolCallStartedFrame(callId: string, toolName: string): Uint8Array { return encodeConnectFrame(toBinary(AgentServerMessageSchema, message)); } +function clientToolArgsFrame(callId: string, toolName: string, argText: string): Uint8Array { + const message = create(AgentServerMessageSchema, { + message: { + case: "execServerMessage", + value: create(ExecServerMessageSchema, { + id: 1, + execId: `exec-${callId}`, + message: { + case: "mcpArgs", + value: create(McpArgsSchema, { + name: toolName, + toolName, + toolCallId: callId, + providerIdentifier: PROVIDER, + args: { text: new TextEncoder().encode(JSON.stringify(argText)) }, + }), + }, + }), + }, + }); + return encodeConnectFrame(toBinary(AgentServerMessageSchema, message)); +} + function turnEndedFrame(): Uint8Array { const message = create(AgentServerMessageSchema, { message: { @@ -79,6 +103,10 @@ function emptyFrame(): Uint8Array { return encodeConnectFrame(toBinary(AgentServerMessageSchema, create(AgentServerMessageSchema, {}))); } +function cleanConnectEndFrame(): Uint8Array { + return encodeConnectFrame(new TextEncoder().encode("{}"), { endStream: true }); +} + function runRequest(tools?: CursorRunRequest["tools"]): CursorRunRequest { return { modelId: "composer-2", @@ -96,6 +124,17 @@ const APPLY_PATCH_TOOL = [{ freeform: true, }] as unknown as CursorRunRequest["tools"]; +const ECHO_TOOL = [{ + name: "echo_a", + description: "echo text", + parameters: { type: "object", properties: { text: { type: "string" } }, required: ["text"] }, +}] as unknown as CursorRunRequest["tools"]; + +const ECHO_AND_APPLY_PATCH_TOOLS = [ + ...(ECHO_TOOL ?? []), + ...(APPLY_PATCH_TOOL ?? []), +] as CursorRunRequest["tools"]; + async function drain(baseUrl: string, request: CursorRunRequest): Promise<{ messages: CursorServerMessage[]; failure?: Error; @@ -153,6 +192,80 @@ describe("Cursor clean-EOF terminal gate", () => { }); }); + test("clean Connect END_STREAM finishes before a held-open HTTP body (#2300)", async () => { + let fallback: ReturnType | undefined; + const startedAt = Date.now(); + await withH2Server(stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + stream.write(Buffer.from(turnEndedFrame())); + stream.write(Buffer.from(cleanConnectEndFrame())); + // Model Cursor's observed shape: the protocol has ended, but the HTTP body has not. The + // fallback keeps the pre-fix test bounded; correct code returns well before it fires. + fallback = setTimeout(() => { + try { stream.end(); } catch { /* transport already closed */ } + }, 500); + }, async baseUrl => { + const { messages, failure } = await drain(baseUrl, runRequest()); + expect(failure).toBeUndefined(); + expect(messages.filter(message => message.type === "done")).toHaveLength(1); + }); + if (fallback) clearTimeout(fallback); + expect(Date.now() - startedAt).toBeLessThan(450); + }); + + test("clean Connect END_STREAM wins over an immediate abort-shaped body teardown (#2300)", async () => { + await withH2Server(stream => { + stream.on("error", () => {}); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + stream.write(Buffer.from(turnEndedFrame())); + stream.write(Buffer.from(cleanConnectEndFrame())); + setImmediate(() => { + const abort = new Error("The operation was aborted"); + abort.name = "AbortError"; + stream.destroy(abort); + }); + }, async baseUrl => { + const { messages, failure } = await drain(baseUrl, runRequest()); + expect(failure).toBeUndefined(); + expect(messages.filter(message => message.type === "done")).toHaveLength(1); + expect(messages.some(message => message.type === "error")).toBe(false); + }); + }); + + test("clean Connect END_STREAM preserves a drained client-tool terminal before its grace timer", async () => { + await withH2Server(respondWith([ + toolCallStartedFrame("call_client_1", "echo_a"), + clientToolArgsFrame("call_client_1", "echo_a", "A"), + cleanConnectEndFrame(), + ]), async baseUrl => { + const { messages, failure } = await drain(baseUrl, runRequest(ECHO_TOOL)); + + expect(failure).toBeUndefined(); + expect(messages.filter(message => message.type === "tool_call_end")).toHaveLength(1); + expect(messages.filter(message => message.type === "done")).toHaveLength(1); + expect(messages.some(message => message.type === "error")).toBe(false); + }); + }); + + test("clean Connect END_STREAM keeps a later open sibling fail-closed after a client-tool drain", async () => { + await withH2Server(respondWith([ + toolCallStartedFrame("call_client_2", "echo_a"), + clientToolArgsFrame("call_client_2", "echo_a", "A"), + toolCallStartedFrame("call_open_2", "apply_patch"), + cleanConnectEndFrame(), + ]), async baseUrl => { + const { messages, failure } = await drain(baseUrl, runRequest(ECHO_AND_APPLY_PATCH_TOOLS)); + + expect(failure).toBeUndefined(); + expect(messages.filter(message => message.type === "tool_call_end")).toHaveLength(1); + const terminal = messages.at(-1); + expect(terminal?.type).toBe("error"); + expect((terminal as { message?: string }).message).toContain("call_open_2"); + expect(messages.some(message => message.type === "done")).toBe(false); + }); + }); + test("EOF with no open tool call keeps its existing graceful finish", async () => { await withH2Server(respondWith([emptyFrame()]), async baseUrl => { const { failure } = await drain(baseUrl, runRequest());