Skip to content

Commit 6889861

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 5c65825 commit 6889861

6 files changed

Lines changed: 384 additions & 120 deletions

File tree

‎src/plugin/index.ts‎

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

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

‎src/provider/language-model.ts‎

Lines changed: 145 additions & 120 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ import {
4040
acquireAgent,
4141
dropSessionRecord,
4242
getSessionRecord,
43+
withSessionLock,
4344
} from "./session-pool.js";
4445
import {
4546
classifyTurn,
@@ -190,134 +191,155 @@ export class CursorLanguageModel implements LanguageModelV3 {
190191

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