From 5e597908246b95de4ed3e91a492163ccd4ec2721 Mon Sep 17 00:00:00 2001 From: Kyle Welsworth Date: Wed, 29 Jul 2026 19:29:06 +0700 Subject: [PATCH] fix(cursor): stop leaking agent tool errors as text Emit SSE error frame for unsupported Cursor AgentService IDE tools instead of assistant content, and drop frames after the turn finishes to avoid double-closing the stream controller. --- open-sse/executors/cursor.js | 17 ++- tests/unit/cursor-agent-exec-request.test.js | 114 +++++++++++++++++++ 2 files changed, 128 insertions(+), 3 deletions(-) create mode 100644 tests/unit/cursor-agent-exec-request.test.js diff --git a/open-sse/executors/cursor.js b/open-sse/executors/cursor.js index 5023aa45..0aefc623 100644 --- a/open-sse/executors/cursor.js +++ b/open-sse/executors/cursor.js @@ -12,7 +12,7 @@ import { import { buildCursorHeaders } from "../utils/cursorChecksum.js"; import { estimateUsage } from "../utils/usageTracking.js"; import { SSE_DONE, SSE_HEADERS } from "../utils/sseConstants.js"; -import { chatChunkSse } from "../utils/sse.js"; +import { chatChunkSse, sseChunk } from "../utils/sse.js"; import { FORMATS } from "../translator/formats.js"; import { proxyAwareFetch } from "../utils/proxyFetch.js"; import zlib from "zlib"; @@ -543,6 +543,9 @@ export class CursorExecutor extends BaseExecutor { if (done) break; pending = Buffer.concat([pending, Buffer.from(value)]); pending = decodeAgentFrames(pending, (payload) => { + // A single read can carry several frames; once the turn is over the + // rest of the batch must not reach the already-closed controller. + if (finished) return; const serverMessage = decodeMessage(payload); // agent.v1.AgentServerMessage.interaction_update @@ -570,9 +573,12 @@ export class CursorExecutor extends BaseExecutor { if (execRequest.has(10)) { session.write(createRequestContextResponse()); } else { + // Every other ExecServerMessage variant is an editor-backed tool + // (shell, read, write, …) that 9router cannot service. Fail the + // turn rather than narrating protocol state as assistant text. + debugLog(`[CURSOR AGENT] Unsupported exec request fields: ${[...execRequest.keys()].join(",")}`); finished = true; onEvent({ type: "error", value: "Cursor AgentService requested an unsupported IDE tool" }); - onEvent({ type: "done" }); } } }); @@ -630,7 +636,12 @@ export class CursorExecutor extends BaseExecutor { } else if (event.type === "thinking") { controller.enqueue(encoder.encode(chatChunkSse({ id: responseId, created, model, delta: { reasoning_content: event.value } }))); } else if (event.type === "error") { - controller.enqueue(encoder.encode(chatChunkSse({ id: responseId, created, model, delta: { content: `\n[${event.value}]` } }))); + // An SSE error frame, not a content delta: a protocol failure must not + // be rendered to the user as the assistant's reply, and downstream + // usage tracking must not record the turn as a success. + controller.enqueue(encoder.encode(sseChunk({ error: { message: event.value, type: "api_error" } }))); + controller.enqueue(encoder.encode(SSE_DONE)); + controller.close(); } else if (event.type === "done") { controller.enqueue(encoder.encode(chatChunkSse({ id: responseId, created, model, delta: {}, finishReason: "stop" }))); controller.enqueue(encoder.encode(SSE_DONE)); diff --git a/tests/unit/cursor-agent-exec-request.test.js b/tests/unit/cursor-agent-exec-request.test.js new file mode 100644 index 00000000..347e159f --- /dev/null +++ b/tests/unit/cursor-agent-exec-request.test.js @@ -0,0 +1,114 @@ +import { describe, it, expect } from "vitest"; + +import { CursorExecutor } from "../../open-sse/executors/cursor.js"; +import { encodeField, wrapConnectRPCFrame } from "../../open-sse/utils/cursorProtobuf.js"; + +const LEN = 2; + +// agent.v1.AgentServerMessage.exec_request (field 2) carrying one ExecServerMessage variant. +function execRequestFrame(execField) { + const execServerMessage = Buffer.from(encodeField(execField, LEN, new Uint8Array())); + return Buffer.from(wrapConnectRPCFrame(encodeField(2, LEN, execServerMessage))); +} + +// agent.v1.AgentServerMessage.interaction_update (field 1) → text delta. +function textFrame(text) { + const textPart = Buffer.from(encodeField(1, LEN, text)); + const update = Buffer.from(encodeField(1, LEN, textPart)); + return Buffer.from(wrapConnectRPCFrame(encodeField(1, LEN, update))); +} + +function stubAgentSession(executor, frames) { + const written = []; + const queue = [...frames]; + executor.openAgentHttp2Stream = () => ({ + responseHeaders: Promise.resolve({ ":status": 200 }), + write: (frame) => written.push(Buffer.from(frame)), + end() {}, + close() {}, + async read() { + if (!queue.length) return { value: undefined, done: true }; + return { value: queue.shift(), done: false }; + }, + }); + return written; +} + +const credentials = { + accessToken: "test-token", + providerSpecificData: { machineId: "a".repeat(64) }, +}; + +function parseSSE(text) { + return text + .split("\n\n") + .filter((chunk) => chunk.startsWith("data: ")) + .map((chunk) => chunk.slice("data: ".length)) + .filter((data) => data !== "[DONE]") + .map((data) => JSON.parse(data)); +} + +async function runAgent({ frames, stream }) { + const executor = new CursorExecutor(); + const written = stubAgentSession(executor, frames); + const result = await executor.executeAgent({ + model: "gpt-5.2", + body: { messages: [{ role: "user", content: "hi" }] }, + stream, + credentials, + }); + return { result, written }; +} + +describe("CursorExecutor AgentService exec_request handling", () => { + it("acknowledges a request-context exec request without ending the turn", async () => { + const { result, written } = await runAgent({ + frames: [execRequestFrame(10), textFrame("hello")], + stream: true, + }); + + expect(written.length).toBe(2); // run frame + request-context reply + const events = parseSSE(await result.response.text()); + const content = events.map((e) => e.choices?.[0]?.delta?.content || "").join(""); + expect(content).toBe("hello"); + }); + + it("does not render an unsupported exec request as assistant content", async () => { + const { result } = await runAgent({ + frames: [textFrame("partial answer"), execRequestFrame(2)], + stream: true, + }); + + const body = await result.response.text(); + expect(body).not.toContain("unsupported IDE tool\\n"); + const events = parseSSE(body); + const content = events.map((e) => e.choices?.[0]?.delta?.content || "").join(""); + expect(content).toBe("partial answer"); + + const errorEvent = events.find((e) => e.error); + expect(errorEvent?.error?.message).toContain("unsupported IDE tool"); + expect(events.some((e) => e.choices?.[0]?.finish_reason === "stop")).toBe(false); + }); + + it("drops frames batched behind an unsupported exec request in the same read", async () => { + const { result } = await runAgent({ + frames: [Buffer.concat([execRequestFrame(2), textFrame("late")])], + stream: true, + }); + + const body = await result.response.text(); + expect(body).toContain("unsupported IDE tool"); + expect(body).not.toContain("late"); + }); + + it("returns a non-200 error body for an unsupported exec request when not streaming", async () => { + const { result } = await runAgent({ + frames: [execRequestFrame(11)], + stream: false, + }); + + expect(result.response.status).not.toBe(200); + const payload = await result.response.json(); + expect(payload.error.message).toContain("unsupported IDE tool"); + }); +});