From d0e505b4f8ba57eb2d80cba41791ba5c2aee3c6c Mon Sep 17 00:00:00 2001 From: left-to-right <2201218482@qq.com> Date: Thu, 13 Aug 2026 01:33:24 +0800 Subject: [PATCH] fix(ai): rotate Cursor conversationId after a poisoned conversation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cursor's backend can pin a per-conversation rejection (bare resource_exhausted, zero tokens) to one wire conversationId forever, making the session unusable until /fork mints a new id. On the first such failure the id is rotated once and the cached state migrates, so the caller's retry loop starts a fresh conversation — the same recovery /fork performs. Repeated failures keep the rotated id so a genuine account-level exhaustion is not hidden (#8345). The regex intentionally drops the global flag: test() would otherwise share lastIndex across calls and silently miss every second message. Covered by an end-to-end HTTP/2 fixture: a real local server rejects the first conversation and accepts the next, asserting the wire conversationId differs; a second fixture fails every request and asserts the rotation happens exactly once. Co-Authored-By: Claude Opus 4.7 --- packages/ai/src/providers/cursor.ts | 48 ++++- .../test/cursor-conversation-rotate.test.ts | 200 ++++++++++++++++++ 2 files changed, 244 insertions(+), 4 deletions(-) create mode 100644 packages/ai/test/cursor-conversation-rotate.test.ts diff --git a/packages/ai/src/providers/cursor.ts b/packages/ai/src/providers/cursor.ts index ca44cdc12..de00062c8 100644 --- a/packages/ai/src/providers/cursor.ts +++ b/packages/ai/src/providers/cursor.ts @@ -293,11 +293,23 @@ const CURSOR_PROXY_TUNNEL_TIMEOUT_MS = 30_000; * model reads it and should route around the capability, not retry the call. */ const NOT_IMPLEMENTED_SUFFIX = "not implemented by this client"; +/** Bare gRPC `resource_exhausted` end-streams (also inside a Connect error message). */ +const RESOURCE_EXHAUSTED_PATTERN = /resource.?exhausted/i; const NOT_IMPLEMENTED = `Not implemented by this client`; const conversationStateCache = new Map(); const conversationBlobStores = new Map>(); const warnedCursorKimiK3ReplayMessages = new Set(); +/** + * Base conversation id → rotated wire id (#8345). Cursor's backend can pin a + * per-conversation rejection (bare `resource_exhausted`, zero tokens) to one + * conversationId forever; the session is then unusable until /fork mints a + * new id. On the first such failure the id is rotated once and the cached + * state migrates, so the retry loop's next attempt starts a fresh + * conversation — the same recovery /fork performs. Keyed by the base id the + * caller derived, so a failed rotation is never repeated. + */ +const rotatedConversationIds = new Map(); export interface CursorOptions extends StreamOptions { customSystemPrompt?: string; @@ -587,13 +599,19 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( h2Completion.resolve(); }; + // Hoisted out of the try block: the #8345 rotation in the catch path + // needs both ids, and the catch block cannot see try-scoped consts. + let baseConversationId: string | undefined; + let conversationId: string | undefined; + let usageState: UsageState | undefined; try { const apiKey = options?.apiKey; if (!apiKey) { throw new AIError.MissingApiKeyError(undefined, "Cursor API key (access token) is required"); } - const conversationId = options?.conversationId ?? options?.sessionId ?? crypto.randomUUID(); + baseConversationId = options?.conversationId ?? options?.sessionId ?? crypto.randomUUID(); + conversationId = rotatedConversationIds.get(baseConversationId) ?? baseConversationId; const blobStore = conversationBlobStores.get(conversationId) ?? new Map(); conversationBlobStores.set(conversationId, blobStore); const cachedState = conversationStateCache.get(conversationId); @@ -667,7 +685,7 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( let currentThinkingBlock: (ThinkingContent & { [kStreamingBlockIndex]: number }) | null = null; let currentToolCall: ToolCallState | null = null; const resolvedMcpToolCallIds = new Set(); - const usageState: UsageState = { sawTokenDelta: false }; + usageState = { sawTokenDelta: false }; const state: BlockState = { get currentTextBlock() { @@ -702,7 +720,7 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( openBlockState = state; const onConversationCheckpoint = (checkpoint: ConversationStateStructure) => { - conversationStateCache.set(conversationId, checkpoint); + conversationStateCache.set(conversationId!, checkpoint); }; h2Request.on("response", headers => { @@ -759,7 +777,7 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( h2Request!, options?.execHandlers, options?.onToolResult, - usageState, + usageState!, requestContextTools, onConversationCheckpoint, ).catch(error => { @@ -873,6 +891,28 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( flushOpenToolCalls(output, stream, openBlockState); } const result = await AIError.finalize(error, { api: model.api, signal: options?.signal }); + // #8345: a server-side per-conversation rejection surfaces as a bare + // resource_exhausted with zero tokens — the conversation is poisoned, + // not the account (sibling conversations keep working). Rotate the + // wire id once and migrate the cached state so the next attempt (the + // caller's retry loop) starts a fresh conversation, exactly like + // /fork. Only the first failure rotates; repeated failures keep the + // rotated id so a genuine account-level exhaustion is not hidden. + if ( + conversationId !== undefined && + baseConversationId !== undefined && + usageState !== undefined && + !usageState.sawTokenDelta && + RESOURCE_EXHAUSTED_PATTERN.test(result.message) && + !rotatedConversationIds.has(baseConversationId) + ) { + const rotated = crypto.randomUUID(); + rotatedConversationIds.set(baseConversationId, rotated); + const state = conversationStateCache.get(conversationId); + if (state) conversationStateCache.set(rotated, state); + const blobs = conversationBlobStores.get(conversationId); + if (blobs) conversationBlobStores.set(rotated, blobs); + } output.stopReason = result.stopReason; output.errorStatus = result.status; output.errorId = result.id; diff --git a/packages/ai/test/cursor-conversation-rotate.test.ts b/packages/ai/test/cursor-conversation-rotate.test.ts new file mode 100644 index 000000000..aa478188c --- /dev/null +++ b/packages/ai/test/cursor-conversation-rotate.test.ts @@ -0,0 +1,200 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import * as http2 from "node:http2"; +import { create, fromBinary, toBinary } from "@bufbuild/protobuf"; +import { streamCursor } from "@oh-my-pi/pi-ai/providers/cursor"; +import type { Context, Model } from "@oh-my-pi/pi-ai/types"; +import { buildModel } from "@oh-my-pi/pi-catalog/build"; +import { + AgentClientMessageSchema, + AgentServerMessageSchema, + InteractionUpdateSchema, + TextDeltaUpdateSchema, + TurnEndedUpdateSchema, +} from "@oh-my-pi/pi-catalog/discovery/cursor-gen/agent_pb"; + +// #8345: a server-side per-conversation rejection (bare resource_exhausted, +// zero tokens) poisons the wire conversationId; the next attempt must rotate +// to a fresh id and succeed, instead of failing forever until /fork. + +let server: http2.Http2Server | undefined; +const sessions = new Set(); + +function frameConnectMessage(data: Uint8Array, flags = 0): Buffer { + const frame = Buffer.alloc(5 + data.length); + frame[0] = flags; + frame.writeUInt32BE(data.length, 1); + frame.set(data, 5); + return frame; +} + +function textDeltaFrame(text: string): Buffer { + const message = create(AgentServerMessageSchema, { + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "textDelta", value: create(TextDeltaUpdateSchema, { text }) }, + }), + }, + }); + return frameConnectMessage(toBinary(AgentServerMessageSchema, message)); +} + +function turnEndedFrame(): Buffer { + const message = create(AgentServerMessageSchema, { + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "turnEnded", value: create(TurnEndedUpdateSchema, {}) }, + }), + }, + }); + return frameConnectMessage(toBinary(AgentServerMessageSchema, message)); +} + +/** Decode the wire conversationId from the first client frame of a request. */ +function decodeConversationId(chunk: Buffer): string | undefined { + const msg = fromBinary(AgentClientMessageSchema, chunk.subarray(5)); + if (msg.message.case !== "runRequest") return undefined; + return msg.message.value.conversationId; +} + +/** First request ends with a bare resource_exhausted; later ones turn normally. */ +async function startServer(seenConversationIds: string[]): Promise { + server = http2.createServer(); + server.on("session", session => { + sessions.add(session); + session.on("close", () => sessions.delete(session)); + }); + let requestCount = 0; + server.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("data", (chunk: Buffer) => { + const conversationId = decodeConversationId(chunk); + if (conversationId !== undefined) seenConversationIds.push(conversationId); + requestCount++; + if (requestCount === 1) { + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }, { waitForTrailers: true }); + stream.once("wantTrailers", () => { + stream.sendTrailers({ "grpc-status": "8", "grpc-message": "resource_exhausted" }); + }); + stream.end(); + } else { + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + stream.write(textDeltaFrame("recovered")); + stream.write(turnEndedFrame()); + stream.end(); + } + }); + }); + + const listening = Promise.withResolvers(); + server.once("error", listening.reject); + server.listen(0, "127.0.0.1", listening.resolve); + await listening.promise; + const address = server.address(); + if (!address || typeof address === "string") throw new Error("expected the fixture server to bind a tcp port"); + return `http://127.0.0.1:${address.port}`; +} + +async function stopServer(): Promise { + for (const session of sessions) session.destroy(); + sessions.clear(); + if (!server) return; + const closing = server; + server = undefined; + const closed = Promise.withResolvers(); + closing.close(error => (error ? closed.reject(error) : closed.resolve())); + await closed.promise; +} + +function makeModel(baseUrl: string): Model<"cursor-agent"> { + return buildModel({ + id: "cursor-rotation-fixture", + name: "Cursor rotation fixture", + api: "cursor-agent", + provider: "cursor", + baseUrl, + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 1, + maxTokens: 1, + }); +} + +const context: Context = { messages: [{ role: "user", content: "hello", timestamp: 1 }] }; + +/** Drain a stream and return its terminal event (done / error). */ +async function runToEnd( + baseUrl: string, + sessionId: string, +): Promise<{ type: "done" | "error"; message?: string }> { + const stream = streamCursor(makeModel(baseUrl), context, { apiKey: "test-token", sessionId }); + let terminal: { type: "done" | "error"; message?: string } = { type: "done" }; + for await (const event of stream) { + if (event.type === "error") { + terminal = { type: "error", message: event.error.errorMessage }; + } + } + await stream.result().catch(() => {}); + return terminal; +} + +afterEach(async () => { + await stopServer(); +}); + +describe("Cursor conversationId rotation (issue #8345)", () => { + it("rotates the poisoned conversationId and recovers on the next attempt", async () => { + const seenConversationIds: string[] = []; + const baseUrl = await startServer(seenConversationIds); + + const first = await runToEnd(baseUrl, "sess-poisoned"); + expect(first.type).toBe("error"); + expect(first.message).toMatch(/resource.?exhausted/i); + + const second = await runToEnd(baseUrl, "sess-poisoned"); + expect(second.type).toBe("done"); + + expect(seenConversationIds).toHaveLength(2); + expect(seenConversationIds[0]).toBe("sess-poisoned"); + expect(seenConversationIds[1]).not.toBe(seenConversationIds[0]); + }); + + it("keeps the rotated id when the new conversation is also rejected", async () => { + const seenConversationIds: string[] = []; + // Fail every request: rotation must happen exactly once. + server = http2.createServer(); + server.on("session", session => { + sessions.add(session); + session.on("close", () => sessions.delete(session)); + }); + server.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.on("data", (chunk: Buffer) => { + const conversationId = decodeConversationId(chunk); + if (conversationId !== undefined) seenConversationIds.push(conversationId); + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }, { waitForTrailers: true }); + stream.once("wantTrailers", () => { + stream.sendTrailers({ "grpc-status": "8", "grpc-message": "resource_exhausted" }); + }); + stream.end(); + }); + }); + const listening = Promise.withResolvers(); + server.once("error", listening.reject); + server.listen(0, "127.0.0.1", listening.resolve); + await listening.promise; + const address = server.address(); + if (!address || typeof address === "string") throw new Error("expected the fixture server to bind a tcp port"); + const baseUrl = `http://127.0.0.1:${address.port}`; + + const r1 = await runToEnd(baseUrl, "sess-sticky"); + const r2 = await runToEnd(baseUrl, "sess-sticky"); + const r3 = await runToEnd(baseUrl, "sess-sticky"); + console.log("[test] results:", JSON.stringify([r1.type, r1.message, r2.type, r2.message, r3.type, r3.message]), "seen:", JSON.stringify(seenConversationIds)); + + expect(seenConversationIds).toHaveLength(3); + expect(seenConversationIds[0]).toBe("sess-sticky"); + expect(seenConversationIds[1]).toBe(seenConversationIds[2]); + expect(seenConversationIds[1]).not.toBe("sess-sticky"); + }); +});