From 202074bedccda2c9e2896005ab8989d9d561bf0c Mon Sep 17 00:00:00 2001 From: rfhold Date: Wed, 29 Jul 2026 15:41:12 -0400 Subject: [PATCH] fix(session-pool): stop opencode's title-gen call from racing the pool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit opencode forks a title-generation call on the exact same sessionID as a session's real first turn, concurrently, with an empty system prompt. classifyTurn's side-call detection only fires once a prior pool record exists, so on turn 1 both calls could independently classify as "new" and both write to the pool — whichever agent-creation round-trip resolved last silently and permanently overwrote the other's entry, poisoning the session's fingerprint (matching the reported symptom: a session behaving as if it only ever had the title prompt). Two changes, both needed: - Wire up the plugin's chat.params hook to mark opencode's "title" agent call as providerOptions.cursor.ephemeral = true. The provider already supported this flag (added in df220e8) but nothing ever set it, so it was dead code. - Add withSessionLock (per-sessionID async lock) in session-pool.ts and wrap agentRun's classify-then-acquire span in it, so concurrent turns for the same session always serialize: the second call's classify always observes the first call's completed pool write. This closes the race structurally, not just for the title-agent case. --- src/plugin/index.ts | 10 ++ src/provider/language-model.ts | 261 ++++++++++++++++------------- src/provider/session-pool.ts | 29 ++++ test/language-model-system.test.ts | 1 + test/plugin-tools.test.ts | 11 ++ test/session-pool.test.ts | 188 +++++++++++++++++++++ 6 files changed, 382 insertions(+), 118 deletions(-) diff --git a/src/plugin/index.ts b/src/plugin/index.ts index aeb9e39..6a2db07 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -238,6 +238,16 @@ export const CursorPlugin: Plugin = async (input) => { if (input.agent === "plan" && output.options["mode"] === undefined) { output.options["mode"] = "plan"; } + // opencode runs its own title-generation call on the same sessionID as + // a session's real first turn, concurrently, with an unrelated (empty) + // system prompt. Mark it ephemeral so the provider always treats it as + // a side-call regardless of whether a pool record exists yet — without + // this, a race between the two calls' agent-creation round-trips can + // let the title call's fingerprint win and permanently overwrite the + // session's pool record (see language-model.ts's `ephemeral` check). + if (input.agent === "title") { + output.options["ephemeral"] = true; + } // Dynamically re-forward MCP servers from opencode's *live* state so // mid-session enable/disable reaches the Cursor agent (the config hook diff --git a/src/provider/language-model.ts b/src/provider/language-model.ts index 9d31eae..34de4b7 100644 --- a/src/provider/language-model.ts +++ b/src/provider/language-model.ts @@ -41,6 +41,7 @@ import { acquireAgent, dropSessionRecord, getSessionRecord, + withSessionLock, } from "./session-pool.js"; import { classifyTurn, @@ -192,132 +193,153 @@ export class CursorLanguageModel implements LanguageModelV3 { // Decide create-vs-resume and whether to pool, from the turn classification. const usePool = sessionEnabled && Boolean(sessionID) && !explicitAgentId; - let resumeAgentId: string | undefined = explicitAgentId; - let poolKey: string | undefined; - let record: - | { systemHash: string; userHashes: string[]; mcpHash?: string } - | undefined; - // Number of new trailing user messages for a multi-message interjection - // (>= 2). Stays 0 for every other turn kind. When set, and the agent is - // resumed, we replay just those new messages as sequential turns instead - // of a cold full-transcript replay. - let multiNewUserCount = 0; - if (usePool) { - const classification = ephemeral - ? { - kind: "side-call" as const, - fingerprint: fingerprint(options.prompt), - } - : classifyTurn(getSessionRecord(sessionID!), options.prompt); - switch (classification.kind) { - case "continuation": - case "continuation-multi": { - const prev = getSessionRecord(sessionID!); - // A resumed agent keeps its original MCP servers, so only resume - // when the live MCP set is unchanged; otherwise create fresh so the - // new server set takes effect (re-pooled under the same session). - if (prev?.mcpHash === mcpHash) { - resumeAgentId = prev?.agentId; - } - poolKey = sessionID; - record = { ...classification.fingerprint, mcpHash }; - if (classification.kind === "continuation-multi") { - multiNewUserCount = classification.newUserCount ?? 0; + + // The whole classify -> acquire span below is wrapped in a per-session + // lock (withSessionLock). opencode can run a concurrent side call (e.g. + // its title-generation turn) against the SAME sessionID as a session's + // real first turn; classifyTurn's side-call detection only works once a + // prior pool record exists, so on turn 1 both calls can independently + // classify as "new" and both write to the pool — whichever's agent + // creation round-trip resolves last silently and permanently overwrites + // the other's entry. Serializing per sessionID here means the second + // call's classification always sees the first call's completed write. + const { + acquired, + multiTurns, + idempotencyKey, + systemMode, + baseAcquire, + record, + } = await withSessionLock(usePool ? sessionID : undefined, async () => { + let resumeAgentId: string | undefined = explicitAgentId; + let poolKey: string | undefined; + let record: + | { systemHash: string; userHashes: string[]; mcpHash?: string } + | undefined; + // Number of new trailing user messages for a multi-message interjection + // (>= 2). Stays 0 for every other turn kind. When set, and the agent is + // resumed, we replay just those new messages as sequential turns instead + // of a cold full-transcript replay. + let multiNewUserCount = 0; + if (usePool) { + const classification = ephemeral + ? { + kind: "side-call" as const, + fingerprint: fingerprint(options.prompt), + } + : classifyTurn(getSessionRecord(sessionID!), options.prompt); + switch (classification.kind) { + case "continuation": + case "continuation-multi": { + const prev = getSessionRecord(sessionID!); + // A resumed agent keeps its original MCP servers, so only resume + // when the live MCP set is unchanged; otherwise create fresh so the + // new server set takes effect (re-pooled under the same session). + if (prev?.mcpHash === mcpHash) { + resumeAgentId = prev?.agentId; + } + poolKey = sessionID; + record = { ...classification.fingerprint, mcpHash }; + if (classification.kind === "continuation-multi") { + multiNewUserCount = classification.newUserCount ?? 0; + } + break; } - break; + case "new": + case "divergence": + poolKey = sessionID; + record = { ...classification.fingerprint, mcpHash }; + break; + case "side-call": + // fresh ephemeral agent; pool left untouched. + break; + } + if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") { + const label = + classification.kind === "continuation" + ? "resume" + : classification.kind === "continuation-multi" + ? `resume-multi:${multiNewUserCount}` + : `fresh:${classification.kind}`; + pluginLog("debug", "turn classification", { label, session: sessionID }); } - case "new": - case "divergence": - poolKey = sessionID; - record = { ...classification.fingerprint, mcpHash }; - break; - case "side-call": - // fresh ephemeral agent; pool left untouched. - break; - } - if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") { - const label = - classification.kind === "continuation" - ? "resume" - : classification.kind === "continuation-multi" - ? `resume-multi:${multiNewUserCount}` - : `fresh:${classification.kind}`; - pluginLog("debug", "turn classification", { label, session: sessionID }); } - } - // A multi-message interjection: two-or-more user messages were queued while - // the agent was busy, forming a contiguous user-turn tail (the classifier - // guarantees this shape for "continuation-multi"). On a resumed agent we - // replay just those new messages as sequential turns. - // - // Defensive invariant check: if the recovered tail doesn't match the - // classifier's count (unreachable today, but one classifier refactor away - // from real), we must NOT degrade to sending only the latest message — - // the session record keeps the full N-message fingerprint, so messages - // 1..N-1 would be silently lost. Instead force the cold path: clear the - // resume id so a FRESH agent gets the FULL transcript, which matches the - // record being written and loses nothing. - // - // Computed before acquireAgent so a mismatched tail can clear - // resumeAgentId in time to affect which agent we acquire. - let multiTurns: SDKUserMessage[] | undefined; - if (multiNewUserCount >= 2) { - const turns = trailingUserMessages(options.prompt, multiNewUserCount); - if (turns.length === multiNewUserCount) { - multiTurns = turns; - } else { - resumeAgentId = undefined; + // A multi-message interjection: two-or-more user messages were queued + // while the agent was busy, forming a contiguous user-turn tail (the + // classifier guarantees this shape for "continuation-multi"). On a + // resumed agent we replay just those new messages as sequential turns. + // + // Defensive invariant check: if the recovered tail doesn't match the + // classifier's count (unreachable today, but one classifier refactor + // away from real), we must NOT degrade to sending only the latest + // message — the session record keeps the full N-message fingerprint, + // so messages 1..N-1 would be silently lost. Instead force the cold + // path: clear the resume id so a FRESH agent gets the FULL transcript, + // which matches the record being written and loses nothing. + // + // Computed before acquireAgent so a mismatched tail can clear + // resumeAgentId in time to affect which agent we acquire. + let multiTurns: SDKUserMessage[] | undefined; + if (multiNewUserCount >= 2) { + const turns = trailingUserMessages(options.prompt, multiNewUserCount); + if (turns.length === multiNewUserCount) { + multiTurns = turns; + } else { + resumeAgentId = undefined; + } } - } - const latestUser = latestUserMessage(options.prompt); - const idempotencyKey = sendIdempotencyKey( - sessionID, - record, - latestUser?.text ?? JSON.stringify(options.prompt), - ); + const latestUser = latestUserMessage(options.prompt); + const idempotencyKey = sendIdempotencyKey( + sessionID, + record, + latestUser?.text ?? JSON.stringify(options.prompt), + ); - // In "rules" mode (default), deliver opencode's system prompt through - // Cursor's authoritative rules channel instead of the user transcript. - // Degrades to inline "message" delivery when the user explicitly opted - // out of the "project" settings layer, when the rule file is user-owned, - // or when the write fails (read-only checkout etc.). - const delivery = resolveSystemDelivery({ - mode: this.config.systemPrompt ?? "rules", - settingSources: this.config.settingSources, - cwd: this.config.cwd, - systemText: extractSystemText(options.prompt), - warn: (message) => this.warnOnce(message), - }); - const systemMode: SystemPromptMode = delivery.mode; - const settingSources = delivery.settingSources; + // In "rules" mode (default), deliver opencode's system prompt through + // Cursor's authoritative rules channel instead of the user transcript. + // Degrades to inline "message" delivery when the user explicitly opted + // out of the "project" settings layer, when the rule file is user-owned, + // or when the write fails (read-only checkout etc.). + const delivery = resolveSystemDelivery({ + mode: this.config.systemPrompt ?? "rules", + settingSources: this.config.settingSources, + cwd: this.config.cwd, + systemText: extractSystemText(options.prompt), + warn: (message) => this.warnOnce(message), + }); + const systemMode: SystemPromptMode = delivery.mode; + const settingSources = delivery.settingSources; - // Shared acquire params. The retry path reuses this verbatim (minus - // resumeAgentId) so a fresh agent can never drift from the first attempt's - // config (sandbox, settingSources, MCP, etc.). - const baseAcquire = { - apiKey: this.requireApiKey(), - modelSelection, - mode, - cwd: this.config.cwd, - ...(settingSources ? { settingSources } : {}), - ...(this.config.sandbox !== undefined - ? { sandbox: this.config.sandbox } - : {}), - ...(this.config.autoReview !== undefined - ? { autoReview: this.config.autoReview } - : {}), - ...(mcpServers ? { mcpServers } : {}), - ...(this.config.agents ? { agents: this.config.agents } : {}), - ...(poolKey ? { name: `opencode/${sessionID!.slice(-8)}` } : {}), - ...(poolKey ? { poolKey } : {}), - ...(record ? { record } : {}), - }; + // Shared acquire params. The retry path reuses this verbatim (minus + // resumeAgentId) so a fresh agent can never drift from the first + // attempt's config (sandbox, settingSources, MCP, etc.). + const baseAcquire = { + apiKey: this.requireApiKey(), + modelSelection, + mode, + cwd: this.config.cwd, + ...(settingSources ? { settingSources } : {}), + ...(this.config.sandbox !== undefined + ? { sandbox: this.config.sandbox } + : {}), + ...(this.config.autoReview !== undefined + ? { autoReview: this.config.autoReview } + : {}), + ...(mcpServers ? { mcpServers } : {}), + ...(this.config.agents ? { agents: this.config.agents } : {}), + ...(poolKey ? { name: `opencode/${sessionID!.slice(-8)}` } : {}), + ...(poolKey ? { poolKey } : {}), + ...(record ? { record } : {}), + }; - const acquired = await acquireAgent({ - ...baseAcquire, - ...(resumeAgentId ? { resumeAgentId } : {}), + const acquired = await acquireAgent({ + ...baseAcquire, + ...(resumeAgentId ? { resumeAgentId } : {}), + }); + + return { acquired, multiTurns, idempotencyKey, systemMode, baseAcquire, record }; }); let yielded = false; @@ -461,7 +483,10 @@ export class CursorLanguageModel implements LanguageModelV3 { // original resume failure as the cause for diagnosability. let retry: Awaited>; try { - retry = await acquireAgent({ ...baseAcquire }); + retry = await withSessionLock( + usePool ? sessionID : undefined, + () => acquireAgent({ ...baseAcquire }), + ); } catch (retryErr) { if (retryErr instanceof Error && retryErr.cause === undefined) { retryErr.cause = err; diff --git a/src/provider/session-pool.ts b/src/provider/session-pool.ts index c7c08b8..d700d32 100644 --- a/src/provider/session-pool.ts +++ b/src/provider/session-pool.ts @@ -68,6 +68,35 @@ export function resetSessionPoolMemory(): void { hydrated = false; } +/** + * Per-session chain of pending lock holders, so concurrent turns for the same + * opencode session serialize across the classify-then-acquire-then-pool-write + * span instead of racing on the shared `pool` map. Two calls for the SAME + * sessionID (e.g. opencode's forked title-generation call racing the real + * first turn) can otherwise both read "no prior record", both classify as + * "new", and both write to the pool — whichever's agent-creation round-trip + * resolves last silently overwrites the other's entry, permanently. Calls for + * different sessionIDs are unaffected and run fully concurrently. + */ +const sessionLocks = new Map>(); + +export function withSessionLock( + sessionID: string | undefined, + fn: () => Promise, +): Promise { + if (!sessionID) return fn(); + const prior = sessionLocks.get(sessionID) ?? Promise.resolve(); + const run = prior.then(fn, fn); + // Chained promise for ordering only; errors are handled by the caller via + // the returned `run`, not here. + const guarded = run.catch(() => {}); + sessionLocks.set(sessionID, guarded); + void guarded.finally(() => { + if (sessionLocks.get(sessionID) === guarded) sessionLocks.delete(sessionID); + }); + return run; +} + export interface AcquireAgentParams { apiKey: string; modelSelection: ModelSelection; diff --git a/test/language-model-system.test.ts b/test/language-model-system.test.ts index 8a79077..ed38205 100644 --- a/test/language-model-system.test.ts +++ b/test/language-model-system.test.ts @@ -19,6 +19,7 @@ const streamAgentTurn = vi.fn(); vi.mock("../src/provider/session-pool.js", () => ({ acquireAgent: (...args: unknown[]) => acquireAgent(...args), getSessionRecord: () => undefined, + withSessionLock: (_sessionID: unknown, fn: () => Promise) => fn(), })); vi.mock("../src/provider/agent-events.js", () => ({ streamAgentTurn: (...args: unknown[]) => streamAgentTurn(...args), diff --git a/test/plugin-tools.test.ts b/test/plugin-tools.test.ts index 16adfaa..81ce297 100644 --- a/test/plugin-tools.test.ts +++ b/test/plugin-tools.test.ts @@ -136,4 +136,15 @@ describe("CursorPlugin chat.params hook", () => { const options = await runHook("plan", {}, { providerID: "anthropic", modelID: "x" }); expect(options).toEqual({}); }); + + it("marks opencode's title-generation call as ephemeral so it never touches the session pool", async () => { + const options = await runHook("title"); + expect(options["ephemeral"]).toBe(true); + expect(options["sessionID"]).toBe("s1"); + }); + + it("does not mark other agents as ephemeral", async () => { + const options = await runHook("build"); + expect(options["ephemeral"]).toBeUndefined(); + }); }); diff --git a/test/session-pool.test.ts b/test/session-pool.test.ts index fade2a8..17d352c 100644 --- a/test/session-pool.test.ts +++ b/test/session-pool.test.ts @@ -20,7 +20,11 @@ const { getPooledAgentId, getSessionRecord, resetSessionPoolMemory, + withSessionLock, } = await import("../src/provider/session-pool.js"); +const { classifyTurn, fingerprint } = await import( + "../src/provider/transcript-fingerprint.js" +); function fakeAgent(agentId: string) { return { agentId, close: vi.fn() }; @@ -196,3 +200,187 @@ describe("acquireAgent", () => { ).toBe(true); }); }); + +describe("withSessionLock", () => { + /** A promise you can resolve/reject from the outside. */ + function deferred() { + let resolve!: (value: T) => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; + } + + it("serializes concurrent calls for the same sessionID", async () => { + const order: string[] = []; + const first = deferred(); + + const p1 = withSessionLock("s1", async () => { + order.push("first-start"); + await first.promise; + order.push("first-end"); + }); + const p2 = withSessionLock("s1", async () => { + order.push("second-start"); + order.push("second-end"); + }); + + // Give the microtask queue a chance to run anything that isn't gated + // on `first`. If the lock didn't serialize, "second-start" would + // already be here. + await Promise.resolve(); + await Promise.resolve(); + expect(order).toEqual(["first-start"]); + + first.resolve(); + await Promise.all([p1, p2]); + expect(order).toEqual([ + "first-start", + "first-end", + "second-start", + "second-end", + ]); + }); + + it("runs calls for different sessionIDs fully concurrently", async () => { + const order: string[] = []; + const a = deferred(); + + const p1 = withSessionLock("s1", async () => { + order.push("s1-start"); + await a.promise; + order.push("s1-end"); + }); + const p2 = withSessionLock("s2", async () => { + order.push("s2-start"); + order.push("s2-end"); + }); + + await p2; + // s2 must have completed even though s1 is still blocked on `a`. + expect(order).toEqual(["s1-start", "s2-start", "s2-end"]); + + a.resolve(); + await p1; + expect(order).toEqual(["s1-start", "s2-start", "s2-end", "s1-end"]); + }); + + it("does not let an error in one call block the next call for the same key", async () => { + const order: string[] = []; + await expect( + withSessionLock("s1", async () => { + order.push("first"); + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + + await withSessionLock("s1", async () => { + order.push("second"); + }); + expect(order).toEqual(["first", "second"]); + }); + + it("runs immediately, unserialized, when sessionID is undefined", async () => { + const order: string[] = []; + const p1 = withSessionLock(undefined, async () => { + order.push("a-start"); + await Promise.resolve(); + order.push("a-end"); + }); + const p2 = withSessionLock(undefined, async () => { + order.push("b"); + }); + await Promise.all([p1, p2]); + // No lock for undefined sessionID: "b" runs before "a" finishes. + expect(order).toEqual(["a-start", "b", "a-end"]); + }); + + it("propagates the return value of the wrapped function", async () => { + const result = await withSessionLock("s1", async () => 42); + expect(result).toBe(42); + }); +}); + +describe("withSessionLock + classifyTurn + acquireAgent (title-gen race regression)", () => { + // Reproduces the bug this fix closes: opencode forks a title-generation + // call on the exact same sessionID as a session's real first turn, + // concurrently, with an unrelated (empty) system prompt. Before the pool's + // classify-then-acquire span was serialized per session, both calls could + // see "no prior record" and both write to the pool; whichever finished + // last silently and permanently overwrote the other's entry. + const realPrompt = [ + { role: "system" as const, content: "Real system prompt" }, + { + role: "user" as const, + content: [{ type: "text" as const, text: "hello" }], + }, + ]; + const titlePrompt = [ + { + role: "user" as const, + content: [ + { + type: "text" as const, + text: "Generate a title for this conversation:\n", + }, + ], + }, + ]; + + it("serializes the classify+acquire span so the side-call never corrupts the real turn's pool record", async () => { + const realFp = fingerprint(realPrompt); + const titleFp = fingerprint(titlePrompt); + expect(realFp.systemHash).not.toBe(titleFp.systemHash); + + let titleSawPriorRecord: unknown; + let titleClassification: string | undefined; + + create.mockResolvedValueOnce(fakeAgent("a1")); + create.mockResolvedValueOnce(fakeAgent("title-gen")); + + // Fire both "turns" concurrently — neither is awaited before the other + // starts, mirroring the real race between opencode's forked title call + // and the actual chat turn. + const real = withSessionLock("s1", async () => { + const classification = classifyTurn(getSessionRecord("s1"), realPrompt); + return acquireAgent({ + ...base, + poolKey: "s1", + record: classification.fingerprint, + }); + }); + const title = withSessionLock("s1", async () => { + titleSawPriorRecord = getSessionRecord("s1"); + const classification = classifyTurn( + getSessionRecord("s1"), + titlePrompt, + ); + titleClassification = classification.kind; + if (classification.kind === "side-call") { + return acquireAgent({ ...base }); + } + // If this branch were ever hit, it would reproduce the bug: a + // misclassified side-call re-pooling on top of the real turn. + return acquireAgent({ + ...base, + poolKey: "s1", + record: classification.fingerprint, + }); + }); + + await Promise.all([real, title]); + + // The lock guarantees the title call's classify only runs after the + // real call's pool write has landed, so it correctly sees the real + // turn's record and classifies as a side-call. + expect(titleSawPriorRecord).toMatchObject({ agentId: "a1", ...realFp }); + expect(titleClassification).toBe("side-call"); + expect(getPooledAgentId("s1")).toBe("a1"); + expect(getSessionRecord("s1")).toMatchObject({ + agentId: "a1", + ...realFp, + }); + }); +});