smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge PR #8349: fix(ai): rotate Cursor conversationId after a poisoned conversation (@zhang17-24)

Browse Source
This commit is contained in:
can1357
2026-08-16 02:43:34 +02:00
parent 44e50af907 d0e505b4f8
commit 9c2c17b3e2
2 changed files with 244 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/src/providers/cursor.ts
+44 -4
View File
@@ -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<string, ConversationStateStructure>();
const conversationBlobStores = new Map<string, Map<string, Uint8Array>>();
const warnedCursorKimiK3ReplayMessages = new Set<string>();
/**
* 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<string, string>();
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<string, Uint8Array>();
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<string>();
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;
packages/ai/test/cursor-conversation-rotate.test.ts
+200
View File
@@ -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<http2.Http2Session>();
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<string> {
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<void>();
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<void> {
for (const session of sessions) session.destroy();
sessions.clear();
if (!server) return;
const closing = server;
server = undefined;
const closed = Promise.withResolvers<void>();
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<void>();
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");
});
});