Skip to content

fix(session-pool): stop opencode's title-gen call from racing the pool #84

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
justin-carper merged 1 commit into from
Jul 30, 2026
Merged

fix(session-pool): stop opencode's title-gen call from racing the pool #84

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/plugin/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
261 changes: 143 additions & 118 deletions src/provider/language-model.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import {
acquireAgent,
dropSessionRecord,
getSessionRecord,
withSessionLock,
} from "./session-pool.js";
import {
classifyTurn,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -461,7 +483,10 @@ export class CursorLanguageModel implements LanguageModelV3 {
// original resume failure as the cause for diagnosability.
let retry: Awaited<ReturnType<typeof acquireAgent>>;
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;
Expand Down
29 changes: 29 additions & 0 deletions src/provider/session-pool.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, Promise<unknown>>();

export function withSessionLock<T>(
sessionID: string | undefined,
fn: () => Promise<T>,
): Promise<T> {
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;
Expand Down
1 change: 1 addition & 0 deletions test/language-model-system.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<unknown>) => fn(),
}));
vi.mock("../src/provider/agent-events.js", () => ({
streamAgentTurn: (...args: unknown[]) => streamAgentTurn(...args),
Expand Down
11 changes: 11 additions & 0 deletions test/plugin-tools.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
});
});
Loading