Skip to content

Commit 202074b

Browse files
rfholdrfhold
committed
fix(session-pool): stop opencode's title-gen call from racing the pool
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.
1 parent 2299bbf commit 202074b

6 files changed

Lines changed: 382 additions & 118 deletions

File tree

‎src/plugin/index.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,16 @@ export const CursorPlugin: Plugin = async (input) => {
238238
if (input.agent === "plan" && output.options["mode"] === undefined) {
239239
output.options["mode"] = "plan";
240240
}
241+
// opencode runs its own title-generation call on the same sessionID as
242+
// a session's real first turn, concurrently, with an unrelated (empty)
243+
// system prompt. Mark it ephemeral so the provider always treats it as
244+
// a side-call regardless of whether a pool record exists yet — without
245+
// this, a race between the two calls' agent-creation round-trips can
246+
// let the title call's fingerprint win and permanently overwrite the
247+
// session's pool record (see language-model.ts's `ephemeral` check).
248+
if (input.agent === "title") {
249+
output.options["ephemeral"] = true;
250+
}
241251

242252
// Dynamically re-forward MCP servers from opencode's *live* state so
243253
// mid-session enable/disable reaches the Cursor agent (the config hook

‎src/provider/language-model.ts‎

Lines changed: 143 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ import {
4141
acquireAgent,
4242
dropSessionRecord,
4343
getSessionRecord,
44+
withSessionLock,
4445
} from "./session-pool.js";
4546
import {
4647
classifyTurn,
@@ -192,132 +193,153 @@ export class CursorLanguageModel implements LanguageModelV3 {
192193

193194
// Decide create-vs-resume and whether to pool, from the turn classification.
194195
const usePool = sessionEnabled && Boolean(sessionID) && !explicitAgentId;
195-
let resumeAgentId: string | undefined = explicitAgentId;
196-
let poolKey: string | undefined;
197-
let record:
198-
| { systemHash: string; userHashes: string[]; mcpHash?: string }
199-
| undefined;
200-
// Number of new trailing user messages for a multi-message interjection
201-
// (>= 2). Stays 0 for every other turn kind. When set, and the agent is
202-
// resumed, we replay just those new messages as sequential turns instead
203-
// of a cold full-transcript replay.
204-
let multiNewUserCount = 0;
205-
if (usePool) {
206-
const classification = ephemeral
207-
? {
208-
kind: "side-call" as const,
209-
fingerprint: fingerprint(options.prompt),
210-
}
211-
: classifyTurn(getSessionRecord(sessionID!), options.prompt);
212-
switch (classification.kind) {
213-
case "continuation":
214-
case "continuation-multi": {
215-
const prev = getSessionRecord(sessionID!);
216-
// A resumed agent keeps its original MCP servers, so only resume
217-
// when the live MCP set is unchanged; otherwise create fresh so the
218-
// new server set takes effect (re-pooled under the same session).
219-
if (prev?.mcpHash === mcpHash) {
220-
resumeAgentId = prev?.agentId;
221-
}
222-
poolKey = sessionID;
223-
record = { ...classification.fingerprint, mcpHash };
224-
if (classification.kind === "continuation-multi") {
225-
multiNewUserCount = classification.newUserCount ?? 0;
196+
197+
// The whole classify -> acquire span below is wrapped in a per-session
198+
// lock (withSessionLock). opencode can run a concurrent side call (e.g.
199+
// its title-generation turn) against the SAME sessionID as a session's
200+
// real first turn; classifyTurn's side-call detection only works once a
201+
// prior pool record exists, so on turn 1 both calls can independently
202+
// classify as "new" and both write to the pool — whichever's agent
203+
// creation round-trip resolves last silently and permanently overwrites
204+
// the other's entry. Serializing per sessionID here means the second
205+
// call's classification always sees the first call's completed write.
206+
const {
207+
acquired,
208+
multiTurns,
209+
idempotencyKey,
210+
systemMode,
211+
baseAcquire,
212+
record,
213+
} = await withSessionLock(usePool ? sessionID : undefined, async () => {
214+
let resumeAgentId: string | undefined = explicitAgentId;
215+
let poolKey: string | undefined;
216+
let record:
217+
| { systemHash: string; userHashes: string[]; mcpHash?: string }
218+
| undefined;
219+
// Number of new trailing user messages for a multi-message interjection
220+
// (>= 2). Stays 0 for every other turn kind. When set, and the agent is
221+
// resumed, we replay just those new messages as sequential turns instead
222+
// of a cold full-transcript replay.
223+
let multiNewUserCount = 0;
224+
if (usePool) {
225+
const classification = ephemeral
226+
? {
227+
kind: "side-call" as const,
228+
fingerprint: fingerprint(options.prompt),
229+
}
230+
: classifyTurn(getSessionRecord(sessionID!), options.prompt);
231+
switch (classification.kind) {
232+
case "continuation":
233+
case "continuation-multi": {
234+
const prev = getSessionRecord(sessionID!);
235+
// A resumed agent keeps its original MCP servers, so only resume
236+
// when the live MCP set is unchanged; otherwise create fresh so the
237+
// new server set takes effect (re-pooled under the same session).
238+
if (prev?.mcpHash === mcpHash) {
239+
resumeAgentId = prev?.agentId;
240+
}
241+
poolKey = sessionID;
242+
record = { ...classification.fingerprint, mcpHash };
243+
if (classification.kind === "continuation-multi") {
244+
multiNewUserCount = classification.newUserCount ?? 0;
245+
}
246+
break;
226247
}
227-
break;
248+
case "new":
249+
case "divergence":
250+
poolKey = sessionID;
251+
record = { ...classification.fingerprint, mcpHash };
252+
break;
253+
case "side-call":
254+
// fresh ephemeral agent; pool left untouched.
255+
break;
256+
}
257+
if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") {
258+
const label =
259+
classification.kind === "continuation"
260+
? "resume"
261+
: classification.kind === "continuation-multi"
262+
? `resume-multi:${multiNewUserCount}`
263+
: `fresh:${classification.kind}`;
264+
pluginLog("debug", "turn classification", { label, session: sessionID });
228265
}
229-
case "new":
230-
case "divergence":
231-
poolKey = sessionID;
232-
record = { ...classification.fingerprint, mcpHash };
233-
break;
234-
case "side-call":
235-
// fresh ephemeral agent; pool left untouched.
236-
break;
237-
}
238-
if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") {
239-
const label =
240-
classification.kind === "continuation"
241-
? "resume"
242-
: classification.kind === "continuation-multi"
243-
? `resume-multi:${multiNewUserCount}`
244-
: `fresh:${classification.kind}`;
245-
pluginLog("debug", "turn classification", { label, session: sessionID });
246266
}
247-
}
248267

249-
// A multi-message interjection: two-or-more user messages were queued while
250-
// the agent was busy, forming a contiguous user-turn tail (the classifier
251-
// guarantees this shape for "continuation-multi"). On a resumed agent we
252-
// replay just those new messages as sequential turns.
253-
//
254-
// Defensive invariant check: if the recovered tail doesn't match the
255-
// classifier's count (unreachable today, but one classifier refactor away
256-
// from real), we must NOT degrade to sending only the latest message —
257-
// the session record keeps the full N-message fingerprint, so messages
258-
// 1..N-1 would be silently lost. Instead force the cold path: clear the
259-
// resume id so a FRESH agent gets the FULL transcript, which matches the
260-
// record being written and loses nothing.
261-
//
262-
// Computed before acquireAgent so a mismatched tail can clear
263-
// resumeAgentId in time to affect which agent we acquire.
264-
let multiTurns: SDKUserMessage[] | undefined;
265-
if (multiNewUserCount >= 2) {
266-
const turns = trailingUserMessages(options.prompt, multiNewUserCount);
267-
if (turns.length === multiNewUserCount) {
268-
multiTurns = turns;
269-
} else {
270-
resumeAgentId = undefined;
268+
// A multi-message interjection: two-or-more user messages were queued
269+
// while the agent was busy, forming a contiguous user-turn tail (the
270+
// classifier guarantees this shape for "continuation-multi"). On a
271+
// resumed agent we replay just those new messages as sequential turns.
272+
//
273+
// Defensive invariant check: if the recovered tail doesn't match the
274+
// classifier's count (unreachable today, but one classifier refactor
275+
// away from real), we must NOT degrade to sending only the latest
276+
// message — the session record keeps the full N-message fingerprint,
277+
// so messages 1..N-1 would be silently lost. Instead force the cold
278+
// path: clear the resume id so a FRESH agent gets the FULL transcript,
279+
// which matches the record being written and loses nothing.
280+
//
281+
// Computed before acquireAgent so a mismatched tail can clear
282+
// resumeAgentId in time to affect which agent we acquire.
283+
let multiTurns: SDKUserMessage[] | undefined;
284+
if (multiNewUserCount >= 2) {
285+
const turns = trailingUserMessages(options.prompt, multiNewUserCount);
286+
if (turns.length === multiNewUserCount) {
287+
multiTurns = turns;
288+
} else {
289+
resumeAgentId = undefined;
290+
}
271291
}
272-
}
273292

274-
const latestUser = latestUserMessage(options.prompt);
275-
const idempotencyKey = sendIdempotencyKey(
276-
sessionID,
277-
record,
278-
latestUser?.text ?? JSON.stringify(options.prompt),
279-
);
293+
const latestUser = latestUserMessage(options.prompt);
294+
const idempotencyKey = sendIdempotencyKey(
295+
sessionID,
296+
record,
297+
latestUser?.text ?? JSON.stringify(options.prompt),
298+
);
280299

281-
// In "rules" mode (default), deliver opencode's system prompt through
282-
// Cursor's authoritative rules channel instead of the user transcript.
283-
// Degrades to inline "message" delivery when the user explicitly opted
284-
// out of the "project" settings layer, when the rule file is user-owned,
285-
// or when the write fails (read-only checkout etc.).
286-
const delivery = resolveSystemDelivery({
287-
mode: this.config.systemPrompt ?? "rules",
288-
settingSources: this.config.settingSources,
289-
cwd: this.config.cwd,
290-
systemText: extractSystemText(options.prompt),
291-
warn: (message) => this.warnOnce(message),
292-
});
293-
const systemMode: SystemPromptMode = delivery.mode;
294-
const settingSources = delivery.settingSources;
300+
// In "rules" mode (default), deliver opencode's system prompt through
301+
// Cursor's authoritative rules channel instead of the user transcript.
302+
// Degrades to inline "message" delivery when the user explicitly opted
303+
// out of the "project" settings layer, when the rule file is user-owned,
304+
// or when the write fails (read-only checkout etc.).
305+
const delivery = resolveSystemDelivery({
306+
mode: this.config.systemPrompt ?? "rules",
307+
settingSources: this.config.settingSources,
308+
cwd: this.config.cwd,
309+
systemText: extractSystemText(options.prompt),
310+
warn: (message) => this.warnOnce(message),
311+
});
312+
const systemMode: SystemPromptMode = delivery.mode;
313+
const settingSources = delivery.settingSources;
295314

296-
// Shared acquire params. The retry path reuses this verbatim (minus
297-
// resumeAgentId) so a fresh agent can never drift from the first attempt's
298-
// config (sandbox, settingSources, MCP, etc.).
299-
const baseAcquire = {
300-
apiKey: this.requireApiKey(),
301-
modelSelection,
302-
mode,
303-
cwd: this.config.cwd,
304-
...(settingSources ? { settingSources } : {}),
305-
...(this.config.sandbox !== undefined
306-
? { sandbox: this.config.sandbox }
307-
: {}),
308-
...(this.config.autoReview !== undefined
309-
? { autoReview: this.config.autoReview }
310-
: {}),
311-
...(mcpServers ? { mcpServers } : {}),
312-
...(this.config.agents ? { agents: this.config.agents } : {}),
313-
...(poolKey ? { name: `opencode/${sessionID!.slice(-8)}` } : {}),
314-
...(poolKey ? { poolKey } : {}),
315-
...(record ? { record } : {}),
316-
};
315+
// Shared acquire params. The retry path reuses this verbatim (minus
316+
// resumeAgentId) so a fresh agent can never drift from the first
317+
// attempt's config (sandbox, settingSources, MCP, etc.).
318+
const baseAcquire = {
319+
apiKey: this.requireApiKey(),
320+
modelSelection,
321+
mode,
322+
cwd: this.config.cwd,
323+
...(settingSources ? { settingSources } : {}),
324+
...(this.config.sandbox !== undefined
325+
? { sandbox: this.config.sandbox }
326+
: {}),
327+
...(this.config.autoReview !== undefined
328+
? { autoReview: this.config.autoReview }
329+
: {}),
330+
...(mcpServers ? { mcpServers } : {}),
331+
...(this.config.agents ? { agents: this.config.agents } : {}),
332+
...(poolKey ? { name: `opencode/${sessionID!.slice(-8)}` } : {}),
333+
...(poolKey ? { poolKey } : {}),
334+
...(record ? { record } : {}),
335+
};
317336

318-
const acquired = await acquireAgent({
319-
...baseAcquire,
320-
...(resumeAgentId ? { resumeAgentId } : {}),
337+
const acquired = await acquireAgent({
338+
...baseAcquire,
339+
...(resumeAgentId ? { resumeAgentId } : {}),
340+
});
341+
342+
return { acquired, multiTurns, idempotencyKey, systemMode, baseAcquire, record };
321343
});
322344

323345
let yielded = false;
@@ -461,7 +483,10 @@ export class CursorLanguageModel implements LanguageModelV3 {
461483
// original resume failure as the cause for diagnosability.
462484
let retry: Awaited<ReturnType<typeof acquireAgent>>;
463485
try {
464-
retry = await acquireAgent({ ...baseAcquire });
486+
retry = await withSessionLock(
487+
usePool ? sessionID : undefined,
488+
() => acquireAgent({ ...baseAcquire }),
489+
);
465490
} catch (retryErr) {
466491
if (retryErr instanceof Error && retryErr.cause === undefined) {
467492
retryErr.cause = err;

‎src/provider/session-pool.ts‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,35 @@ export function resetSessionPoolMemory(): void {
6868
hydrated = false;
6969
}
7070

71+
/**
72+
* Per-session chain of pending lock holders, so concurrent turns for the same
73+
* opencode session serialize across the classify-then-acquire-then-pool-write
74+
* span instead of racing on the shared `pool` map. Two calls for the SAME
75+
* sessionID (e.g. opencode's forked title-generation call racing the real
76+
* first turn) can otherwise both read "no prior record", both classify as
77+
* "new", and both write to the pool — whichever's agent-creation round-trip
78+
* resolves last silently overwrites the other's entry, permanently. Calls for
79+
* different sessionIDs are unaffected and run fully concurrently.
80+
*/
81+
const sessionLocks = new Map<string, Promise<unknown>>();
82+
83+
export function withSessionLock<T>(
84+
sessionID: string | undefined,
85+
fn: () => Promise<T>,
86+
): Promise<T> {
87+
if (!sessionID) return fn();
88+
const prior = sessionLocks.get(sessionID) ?? Promise.resolve();
89+
const run = prior.then(fn, fn);
90+
// Chained promise for ordering only; errors are handled by the caller via
91+
// the returned `run`, not here.
92+
const guarded = run.catch(() => {});
93+
sessionLocks.set(sessionID, guarded);
94+
void guarded.finally(() => {
95+
if (sessionLocks.get(sessionID) === guarded) sessionLocks.delete(sessionID);
96+
});
97+
return run;
98+
}
99+
71100
export interface AcquireAgentParams {
72101
apiKey: string;
73102
modelSelection: ModelSelection;

‎test/language-model-system.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ const streamAgentTurn = vi.fn();
1919
vi.mock("../src/provider/session-pool.js", () => ({
2020
acquireAgent: (...args: unknown[]) => acquireAgent(...args),
2121
getSessionRecord: () => undefined,
22+
withSessionLock: (_sessionID: unknown, fn: () => Promise<unknown>) => fn(),
2223
}));
2324
vi.mock("../src/provider/agent-events.js", () => ({
2425
streamAgentTurn: (...args: unknown[]) => streamAgentTurn(...args),

‎test/plugin-tools.test.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,4 +136,15 @@ describe("CursorPlugin chat.params hook", () => {
136136
const options = await runHook("plan", {}, { providerID: "anthropic", modelID: "x" });
137137
expect(options).toEqual({});
138138
});
139+
140+
it("marks opencode's title-generation call as ephemeral so it never touches the session pool", async () => {
141+
const options = await runHook("title");
142+
expect(options["ephemeral"]).toBe(true);
143+
expect(options["sessionID"]).toBe("s1");
144+
});
145+
146+
it("does not mark other agents as ephemeral", async () => {
147+
const options = await runHook("build");
148+
expect(options["ephemeral"]).toBeUndefined();
149+
});
139150
});

0 commit comments

Comments
 (0)