From 96c05446bbdd7833c039c80a1f95e87d42abff0d Mon Sep 17 00:00:00 2001 From: Justin Carper Date: Mon, 6 Jul 2026 13:53:31 -0500 Subject: [PATCH 1/3] feat(session): warm resume for multi-message interjections Two-or-more user messages queued while the agent is busy previously classified as divergence -> fresh agent + full transcript replay, discarding the pooled Cursor agent's warm context. Now a strict-prefix transcript whose new messages form a contiguous trailing run of user turns classifies as continuation-multi: the pooled agent is resumed and each new message is sent sequentially - leading ones silently (no deltas), only the final run streams back. Mirrors opencode's coalesced-loop model where interjected messages fold into one visible turn. Safety: - Interleaved assistant turns in the tail (abort-then-interject) stay divergence: a tail-only replay would silently drop earlier messages. - Pool record is written optimistically before delivery, so an error or abort mid-sequence drops the session record; the next turn replays fresh instead of resuming on top of undelivered messages. - Edit/revert (non-prefix) unchanged: divergence -> full replay. Known trade-off: silent turns' token usage is not reported (no onDelta observer), so multi-message turns slightly undercount usage. --- src/provider/agent-events.ts | 89 ++++++++-- src/provider/language-model.ts | 210 +++++++++++++++-------- src/provider/message-map.ts | 47 ++++- src/provider/session-pool.ts | 12 ++ src/provider/transcript-fingerprint.ts | 41 ++++- test/agent-events.test.ts | 77 +++++++++ test/language-model-interjection.test.ts | 197 +++++++++++++++++++++ test/message-map.test.ts | 55 ++++++ test/session-pool.test.ts | 14 ++ test/transcript-fingerprint.test.ts | 31 +++- 10 files changed, 670 insertions(+), 103 deletions(-) create mode 100644 test/language-model-interjection.test.ts diff --git a/src/provider/agent-events.ts b/src/provider/agent-events.ts index a5efa66..16e17d4 100644 --- a/src/provider/agent-events.ts +++ b/src/provider/agent-events.ts @@ -111,21 +111,8 @@ export async function* streamAgentTurn( }; options.abortSignal?.addEventListener("abort", onAbort); - // A previous opencode/CLI crash (or a second instance racing on the same - // agent store) can leave a persisted run wedged; the SDK then rejects new - // sends with AgentBusyError. Retry once with the SDK's documented recovery - // path (local.force expires the wedged run) instead of failing the turn. - const sendTurn = async (): Promise => { - try { - return await agent.send(message, { mode: options.mode, onDelta }); - } catch (err) { - if (err instanceof Error && err.name === "AgentBusyError") { - if (debug) console.error("[cursor:debug] agent busy; retrying send with local.force"); - return agent.send(message, { mode: options.mode, onDelta, local: { force: true } }); - } - throw err; - } - }; + const sendTurn = (): Promise => + sendWithBusyRetry(agent, message, { mode: options.mode, onDelta }, debug); // Kick off the turn. Resolve text from run.wait() for models that don't emit // incremental text deltas. @@ -176,3 +163,75 @@ export async function* streamAgentTurn( options.abortSignal?.removeEventListener("abort", onAbort); } } + +/** + * Send a message on an agent, retrying once with the SDK's documented recovery + * path on `AgentBusyError`. A previous opencode/CLI crash (or a second instance + * racing on the same agent store) can leave a persisted run wedged; the SDK then + * rejects new sends with `AgentBusyError`. `local.force` expires the wedged run + * instead of failing the turn. Shared by streaming and silent sends. + */ +async function sendWithBusyRetry( + agent: AgentLike, + message: SDKUserMessage, + sendOptions: { + mode: AgentModeOption; + onDelta?: (args: { update: { type: string } & Record }) => void; + }, + debug: boolean, +): Promise { + try { + return await agent.send(message, sendOptions); + } catch (err) { + if (err instanceof Error && err.name === "AgentBusyError") { + if (debug) console.error("[cursor:debug] agent busy; retrying send with local.force"); + return agent.send(message, { ...sendOptions, local: { force: true } }); + } + throw err; + } +} + +/** + * Send a single turn on an already-acquired agent WITHOUT streaming anything + * back. Used to replay the leading messages of a multi-message interjection + * (two-or-more user messages queued while the agent was busy): messages + * `1..N-1` are sent silently and awaited, and only the final message streams + * via {@link streamAgentTurn}. This mirrors opencode's own model, where + * interjected messages fold into a single visible turn. + * + * Honors `options.abortSignal`: an abort cancels the in-flight run so the + * caller can stop before sending the next queued message. + * + * Known trade-off: token usage from silent turns is not reported (no onDelta + * means the `turn-ended` usage update is never observed), so opencode slightly + * undercounts usage on multi-message turns. + */ +export async function sendAgentTurnSilently( + agent: AgentLike, + message: SDKUserMessage, + options: StreamAgentTurnOptions, +): Promise { + // Already aborted: don't start a turn just to cancel it. + if (options.abortSignal?.aborted) return; + const debug = process.env.OPENCODE_CURSOR_DEBUG === "1"; + const runHolder: { run?: AgentRunLike } = {}; + const onAbort = () => { + void Promise.resolve(runHolder.run?.cancel()).catch(() => {}); + }; + options.abortSignal?.addEventListener("abort", onAbort); + try { + const run = await sendWithBusyRetry(agent, message, { mode: options.mode }, debug); + runHolder.run = run; + // The signal may have fired while send() was in flight (before runHolder + // was populated, so onAbort had nothing to cancel); cancel now. + if (options.abortSignal?.aborted) void Promise.resolve(run.cancel()).catch(() => {}); + const result = await run.wait(); + if (result.status === "error") { + throw new Error( + `Cursor run ended with status "error"${result.result ? `: ${result.result}` : ""}`, + ); + } + } finally { + options.abortSignal?.removeEventListener("abort", onAbort); + } +} diff --git a/src/provider/language-model.ts b/src/provider/language-model.ts index 9d40559..c5ff4a8 100644 --- a/src/provider/language-model.ts +++ b/src/provider/language-model.ts @@ -14,15 +14,27 @@ import type { AgentModeOption, } from "@cursor/sdk"; import { resolveCursorApiKey } from "../api-key.js"; -import { latestUserMessage, promptToCursorMessage } from "./message-map.js"; -import { streamAgentTurn, type CursorEvent } from "./agent-events.js"; +import { + latestUserMessage, + promptToCursorMessage, + trailingUserMessages, +} from "./message-map.js"; +import { + sendAgentTurnSilently, + streamAgentTurn, + type CursorEvent, +} from "./agent-events.js"; import { cursorEventsToContent, cursorEventsToStream, type ToolDisplay, } from "./stream-map.js"; import { resolveControls } from "./controls.js"; -import { acquireAgent, getSessionRecord } from "./session-pool.js"; +import { + acquireAgent, + dropSessionRecord, + getSessionRecord, +} from "./session-pool.js"; import { classifyTurn, fingerprint, @@ -146,6 +158,11 @@ export class CursorLanguageModel implements LanguageModelV3 { 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 ? { @@ -154,7 +171,8 @@ export class CursorLanguageModel implements LanguageModelV3 { } : classifyTurn(getSessionRecord(sessionID!), options.prompt); switch (classification.kind) { - case "continuation": { + 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 @@ -164,6 +182,9 @@ export class CursorLanguageModel implements LanguageModelV3 { } poolKey = sessionID; record = { ...classification.fingerprint, mcpHash }; + if (classification.kind === "continuation-multi") { + multiNewUserCount = classification.newUserCount ?? 0; + } break; } case "new": @@ -179,7 +200,9 @@ export class CursorLanguageModel implements LanguageModelV3 { const label = classification.kind === "continuation" ? "resume" - : `fresh:${classification.kind}`; + : classification.kind === "continuation-multi" + ? `resume-multi:${multiNewUserCount}` + : `fresh:${classification.kind}`; console.error( `[cursor:debug] turn classification=${label} session=${sessionID}`, ); @@ -212,82 +235,127 @@ export class CursorLanguageModel implements LanguageModelV3 { ...(resumeAgentId ? { resumeAgentId } : {}), }); - // A resumed agent already remembers the prior conversation, so send only the - // new turn; otherwise send the full transcript. - const message = acquired.resumed - ? (latestUserMessage(options.prompt) ?? - promptToCursorMessage(options.prompt)) - : promptToCursorMessage(options.prompt); + // 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, + // replay just those new messages as sequential turns — send the leading + // ones silently and stream only the final one. If the tail can't be + // recovered (defensive; shouldn't happen post-classifier), skip the multi + // path: on a resumed agent that means sending only the latest message, so + // the empty/short check below treats it as a normal single continuation. + const multiTurns = + acquired.resumed && multiNewUserCount >= 2 + ? trailingUserMessages(options.prompt, multiNewUserCount) + : undefined; let yielded = false; let releasedOriginal = false; try { - try { - for await (const event of streamAgentTurn(acquired.agent, message, { - mode, - abortSignal: options.abortSignal, - })) { - yielded = true; - yield event; - } - } catch (err) { - // Resume-aware retry: a resumed agent can pass resume() yet fail the - // actual send when Cursor's server has already expired the agent (its - // server-side retention is shorter than our local 7-day reuse window, - // and not documented). If nothing has been emitted downstream yet and - // the user hasn't aborted, transparently re-create a fresh agent and - // replay the full transcript — self-healing, no context loss. The - // fresh agent re-pools under the same session (overwriting the dead - // agentId) via acquireAgent's existing pooling path. + if (multiTurns && multiTurns.length === multiNewUserCount) { + // A multi-message interjection on a resumed agent: send the leading + // messages silently and stream only the final one. // - // Deliberate tradeoff: the retry fires on ANY error class (including - // rate-limit or network failures) because Cursor's status:"error" - // carries no machine-readable class to discriminate on. Bounded to a - // single attempt, so the worst case is one extra create. - if ( - acquired.resumed && - !yielded && - !options.abortSignal?.aborted - ) { - if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") { - console.error( - "[cursor:debug] resumed turn failed before emitting; retrying with a fresh agent", - ); - } - acquired.release(); - releasedOriginal = true; - // A fresh create (no resumeAgentId) re-pools under the same - // session, overwriting the dead agentId. If re-acquiring itself - // fails (e.g. transient create error), surface that but keep the - // original resume failure as the cause for diagnosability. - let retry: Awaited>; - try { - retry = await acquireAgent({ ...baseAcquire }); - } catch (retryErr) { - if (retryErr instanceof Error && retryErr.cause === undefined) { - retryErr.cause = err; - } else if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") { - // Non-Error throw or pre-existing cause: the original resume - // failure can't ride along as `cause`, so log it instead of - // dropping it silently. - console.error( - "[cursor:debug] original resume failure (not attachable as cause):", - err, - ); + // The pool record was written optimistically with the FULL new + // fingerprint before any send. If delivery stops partway (error or + // abort), drop the record so the next turn classifies fresh instead + // of resuming on top of messages the agent never received. + let delivered = false; + try { + let aborted = false; + for (let i = 0; i < multiTurns.length - 1; i++) { + if (options.abortSignal?.aborted) { + aborted = true; + break; } - throw retryErr; - } - try { - const replay = promptToCursorMessage(options.prompt); - yield* streamAgentTurn(retry.agent, replay, { + await sendAgentTurnSilently(acquired.agent, multiTurns[i]!, { mode, abortSignal: options.abortSignal, }); - } finally { - retry.release(); } - } else { - throw err; + if (!aborted && !options.abortSignal?.aborted) { + for await (const event of streamAgentTurn( + acquired.agent, + multiTurns[multiTurns.length - 1]!, + { mode, abortSignal: options.abortSignal }, + )) { + yielded = true; + yield event; + } + delivered = true; + } + } finally { + if (!delivered && sessionID) dropSessionRecord(sessionID); + } + } else { + // A resumed agent already remembers the prior conversation, so send + // only the new turn; otherwise send the full transcript. + const message = acquired.resumed + ? (latestUserMessage(options.prompt) ?? + promptToCursorMessage(options.prompt)) + : promptToCursorMessage(options.prompt); + try { + for await (const event of streamAgentTurn(acquired.agent, message, { + mode, + abortSignal: options.abortSignal, + })) { + yielded = true; + yield event; + } + } catch (err) { + // Resume-aware retry: a resumed agent can pass resume() yet fail the + // actual send when Cursor's server has already expired the agent (its + // server-side retention is shorter than our local 7-day reuse window, + // and not documented). If nothing has been emitted downstream yet and + // the user hasn't aborted, transparently re-create a fresh agent and + // replay the full transcript — self-healing, no context loss. The + // fresh agent re-pools under the same session (overwriting the dead + // agentId) via acquireAgent's existing pooling path. + // + // Deliberate tradeoff: the retry fires on ANY error class (including + // rate-limit or network failures) because Cursor's status:"error" + // carries no machine-readable class to discriminate on. Bounded to a + // single attempt, so the worst case is one extra create. + if (acquired.resumed && !yielded && !options.abortSignal?.aborted) { + if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") { + console.error( + "[cursor:debug] resumed turn failed before emitting; retrying with a fresh agent", + ); + } + acquired.release(); + releasedOriginal = true; + // A fresh create (no resumeAgentId) re-pools under the same + // session, overwriting the dead agentId. If re-acquiring itself + // fails (e.g. transient create error), surface that but keep the + // original resume failure as the cause for diagnosability. + let retry: Awaited>; + try { + retry = await acquireAgent({ ...baseAcquire }); + } catch (retryErr) { + if (retryErr instanceof Error && retryErr.cause === undefined) { + retryErr.cause = err; + } else if (process.env["OPENCODE_CURSOR_DEBUG"] === "1") { + // Non-Error throw or pre-existing cause: the original resume + // failure can't ride along as `cause`, so log it instead of + // dropping it silently. + console.error( + "[cursor:debug] original resume failure (not attachable as cause):", + err, + ); + } + throw retryErr; + } + try { + const replay = promptToCursorMessage(options.prompt); + yield* streamAgentTurn(retry.agent, replay, { + mode, + abortSignal: options.abortSignal, + }); + } finally { + retry.release(); + } + } else { + throw err; + } } } } finally { diff --git a/src/provider/message-map.ts b/src/provider/message-map.ts index 36fdab6..199fe5b 100644 --- a/src/provider/message-map.ts +++ b/src/provider/message-map.ts @@ -146,6 +146,23 @@ function fileUrlToPath(url: string | URL): string { } } +/** + * Map one AI-SDK user turn into a Cursor `SDKUserMessage`. File parts (images + * and other files) can't be forwarded to the local Cursor agent, so they're + * noted as text via {@link fileNote} instead of attached natively. + */ +function userTurnToCursorMessage( + message: Extract, +): SDKUserMessage { + const text: string[] = []; + for (const part of message.content) { + if (part.type === "text") text.push(part.text); + else if (part.type === "file") text.push(fileNote(part)); + } + + return { text: text.join("\n") }; +} + /** * Extract only the final user turn as a Cursor message. Used when resuming a * pooled agent that already remembers the prior conversation, so we send just @@ -157,12 +174,30 @@ export function latestUserMessage( ): SDKUserMessage | undefined { const last = prompt[prompt.length - 1]; if (!last || last.role !== "user") return undefined; + return userTurnToCursorMessage(last); +} - const text: string[] = []; - for (const part of last.content) { - if (part.type === "text") text.push(part.text); - else if (part.type === "file") text.push(fileNote(part)); +/** + * Extract the last `count` CONTIGUOUS trailing user turns as Cursor messages, + * in conversation order (oldest of the tail first, newest last). Used when + * resuming a pooled agent after two-or-more user messages were queued while it + * was busy: we replay just those new messages as sequential turns. + * + * Only the contiguous trailing run of user turns is considered; if the final + * message isn't a user turn the result is empty. Returns fewer than `count` + * entries when the trailing run is shorter (caller should fall back to a full + * transcript replay when the array is empty or shorter than expected). + */ +export function trailingUserMessages( + prompt: LanguageModelV3Prompt, + count: number, +): SDKUserMessage[] { + if (count <= 0) return []; + const collected: SDKUserMessage[] = []; + for (let i = prompt.length - 1; i >= 0 && collected.length < count; i--) { + const message = prompt[i]; + if (!message || message.role !== "user") break; + collected.push(userTurnToCursorMessage(message)); } - - return { text: text.join("\n") }; + return collected.reverse(); } diff --git a/src/provider/session-pool.ts b/src/provider/session-pool.ts index 6cc3e47..98332a0 100644 --- a/src/provider/session-pool.ts +++ b/src/provider/session-pool.ts @@ -40,6 +40,18 @@ export function getSessionRecord( return pool.get(sessionID); } +/** + * Drop a session's pooled record so the NEXT turn classifies as "new" (fresh + * agent + full transcript replay). Called when a multi-message replay fails or + * aborts mid-sequence: the record was written optimistically with the full new + * fingerprint before delivery, so leaving it in place would let a later + * "continuation" resume on top of messages the agent never received. + */ +export function dropSessionRecord(sessionID: string): void { + hydrate(); + if (pool.delete(sessionID)) saveSessionRecords(pool); +} + /** Test/diagnostic helpers. */ export function getPooledAgentId(sessionID: string): string | undefined { hydrate(); diff --git a/src/provider/transcript-fingerprint.ts b/src/provider/transcript-fingerprint.ts index c683c68..4c7ccd3 100644 --- a/src/provider/transcript-fingerprint.ts +++ b/src/provider/transcript-fingerprint.ts @@ -38,13 +38,21 @@ export type TurnKind = | "side-call" /** Prior user sequence is a strict prefix + exactly one new trailing user msg. */ | "continuation" - /** Edit / revert / compaction / multiple queued msgs — prior prefix no longer holds. */ + /** Prior user sequence is a strict prefix + two-or-more new trailing user messages. */ + | "continuation-multi" + /** Edit / revert / compaction — prior prefix no longer holds. */ | "divergence"; export interface TurnClassification { kind: TurnKind; /** Fingerprint of the CURRENT prompt, to store when (re)pooling. */ fingerprint: { systemHash: string; userHashes: string[] }; + /** + * Number of new trailing user messages relative to the prior record. Set on + * `continuation` (always 1) and `continuation-multi` (>= 2); undefined for + * other kinds. Lets the caller know how many tail messages to send on resume. + */ + newUserCount?: number; } function sha(input: string): string { @@ -108,7 +116,12 @@ function isStrictPrefix(prefix: string[], full: string[]): boolean { * 2. system prompt changed -> "side-call" (don't touch the pool) * 3. prior user hashes are a strict prefix AND exactly one new trailing user * message AND the last prompt message is a user turn -> "continuation" - * 4. otherwise -> "divergence" (edit/revert/compaction/queued) + * 4. prior user hashes are a strict prefix AND two-or-more new user messages + * forming a CONTIGUOUS user-turn tail of the prompt -> "continuation-multi" + * (interleaved assistant turns — e.g. an aborted reply between queued + * messages — disqualify, because a resumed replay of only the contiguous + * tail would silently drop the earlier queued message) + * 5. otherwise -> "divergence" (edit/revert/compaction) * * Worst case on any misclassification is a single wasted full replay that * self-heals on the next turn — never worse than the `session: false` default. @@ -123,13 +136,23 @@ export function classifyTurn( return { kind: "side-call", fingerprint: fp }; const lastIsUser = prompt[prompt.length - 1]?.role === "user"; - const exactlyOneNew = fp.userHashes.length === prev.userHashes.length + 1; - if ( - lastIsUser && - exactlyOneNew && - isStrictPrefix(prev.userHashes, fp.userHashes) - ) { - return { kind: "continuation", fingerprint: fp }; + const newUserCount = fp.userHashes.length - prev.userHashes.length; + const isPrefix = isStrictPrefix(prev.userHashes, fp.userHashes); + if (lastIsUser && isPrefix && newUserCount === 1) { + return { kind: "continuation", fingerprint: fp, newUserCount: 1 }; + } + // The N new messages must be the prompt's contiguous trailing user turns. + // When an assistant turn sits between them (aborted reply between queued + // interjections), a resumed tail-only replay would drop the earlier queued + // message — so that shape must take the divergence full-replay path. + const tailContiguous = + newUserCount >= 2 && + prompt.length >= newUserCount && + prompt + .slice(prompt.length - newUserCount) + .every((m) => m.role === "user"); + if (lastIsUser && isPrefix && tailContiguous) { + return { kind: "continuation-multi", fingerprint: fp, newUserCount }; } return { kind: "divergence", fingerprint: fp }; } diff --git a/test/agent-events.test.ts b/test/agent-events.test.ts index a6cf820..21938af 100644 --- a/test/agent-events.test.ts +++ b/test/agent-events.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import type { Run, SDKUserMessage } from "@cursor/sdk"; import { + sendAgentTurnSilently, streamAgentTurn, type CursorEvent, } from "../src/provider/agent-events.js"; @@ -117,6 +118,82 @@ describe("streamAgentTurn busy-agent recovery", () => { }); }); +describe("sendAgentTurnSilently", () => { + it("sends the message with no onDelta and awaits completion", async () => { + const sendCalls: Array | undefined> = []; + const agent = fakeAgent({ + updates: [{ type: "text-delta", text: "should-not-surface" }], + result: { status: "finished", result: "done" }, + sendCalls, + }); + + await sendAgentTurnSilently(agent, MESSAGE, { mode: "agent" }); + + expect(sendCalls).toHaveLength(1); + expect(sendCalls[0]?.["onDelta"]).toBeUndefined(); + }); + + it("throws when the run ends with status 'error'", async () => { + const agent = fakeAgent({ result: { status: "error", result: "boom" } }); + await expect( + sendAgentTurnSilently(agent, MESSAGE, { mode: "agent" }), + ).rejects.toThrow(/error/i); + }); + + it("retries with local.force on AgentBusyError", async () => { + const busy = new Error("agent busy"); + busy.name = "AgentBusyError"; + const sendCalls: Array | undefined> = []; + const agent = fakeAgent({ + rejectFirst: { error: busy, times: 1 }, + result: { status: "finished", result: "" }, + sendCalls, + }); + + await sendAgentTurnSilently(agent, MESSAGE, { mode: "agent" }); + + expect(sendCalls).toHaveLength(2); + expect(sendCalls[1]?.["local"]).toMatchObject({ force: true }); + }); + + it("cancels the run when the abort signal fires", async () => { + let cancelled = false; + const controller = new AbortController(); + const agent = { + agentId: "agent-test", + send: async ( + _message: SDKUserMessage, + _sendOptions?: Record, + ) => { + const run: Partial = { + wait: () => + new Promise((resolve) => { + // Resolve only after abort triggers cancel(). + const check = setInterval(() => { + if (cancelled) { + clearInterval(check); + resolve({ status: "cancelled" } as never); + } + }, 1); + }), + cancel: async () => { + cancelled = true; + }, + }; + return run as Run; + }, + } as unknown as AgentLike; + + const promise = sendAgentTurnSilently(agent, MESSAGE, { + mode: "agent", + abortSignal: controller.signal, + }); + controller.abort(); + await promise; + expect(cancelled).toBe(true); + }); +}); + describe("streamAgentTurn MCP error surfacing", () => { it("marks an MCP tool result as error when its success value carries isError", async () => { const agent = fakeAgent({ diff --git a/test/language-model-interjection.test.ts b/test/language-model-interjection.test.ts new file mode 100644 index 0000000..0c1bf6c --- /dev/null +++ b/test/language-model-interjection.test.ts @@ -0,0 +1,197 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { LanguageModelV3Prompt } from "@ai-sdk/provider"; + +// Sandbox the on-disk session store away from the user's real cache dir. +process.env.XDG_CACHE_HOME = mkdtempSync(join(tmpdir(), "cursor-lm-test-")); + +/** + * A fake Cursor agent. Records every send() so tests can assert how many turns + * were replayed and in what order. Each send drives its onDelta (if any) with a + * single text delta echoing the sent text, then resolves wait(). + */ +interface SentTurn { + text: string; + streamed: boolean; +} + +function makeFakeAgent( + agentId: string, + sent: SentTurn[], + opts?: { failOnText?: string }, +) { + return { + agentId, + close: vi.fn(), + send: async ( + message: { text: string }, + sendOptions?: Record, + ) => { + const onDelta = sendOptions?.["onDelta"] as + | ((a: { update: { type: string } & Record }) => void) + | undefined; + sent.push({ text: message.text, streamed: Boolean(onDelta) }); + const failed = opts?.failOnText === message.text; + onDelta?.({ update: { type: "text-delta", text: message.text } }); + return { + id: "run", + wait: async () => + failed + ? { status: "error", result: "silent send blew up" } + : { status: "finished", result: message.text }, + cancel: async () => {}, + }; + }, + }; +} + +const sentTurns: SentTurn[] = []; +const create = vi.fn(); +const resume = vi.fn(); + +vi.mock("../src/cursor-runtime.js", () => ({ + loadCursorSdk: async () => ({ Agent: { create, resume } }), +})); + +const { CursorLanguageModel } = await import( + "../src/provider/language-model.js" +); +const { clearAgentPool, getSessionRecord } = await import( + "../src/provider/session-pool.js" +); + +const SESSION_ID = "sess-interjection"; + +function model() { + return new CursorLanguageModel("cursor/auto", { + providerName: "cursor", + apiKey: "k", + cwd: "/tmp", + mode: "agent", + }); +} + +function callOptions(prompt: LanguageModelV3Prompt) { + return { + prompt, + providerOptions: { cursor: { sessionID: SESSION_ID } }, + } as never; +} + +async function drain( + model: InstanceType, + prompt: LanguageModelV3Prompt, +): Promise> { + const { stream } = await model.doStream(callOptions(prompt)); + const reader = stream.getReader(); + const parts: Array<{ type: string }> = []; + // eslint-disable-next-line no-constant-condition + while (true) { + const { done, value } = await reader.read(); + if (done) break; + parts.push(value as { type: string }); + } + return parts; +} + +const sys = { role: "system" as const, content: "S" }; +const user = (text: string): LanguageModelV3Prompt[number] => ({ + role: "user", + content: [{ type: "text", text }], +}); +const assistant = (text: string): LanguageModelV3Prompt[number] => ({ + role: "assistant", + content: [{ type: "text", text }], +}); + +afterEach(() => { + create.mockReset(); + resume.mockReset(); + sentTurns.length = 0; + clearAgentPool(); +}); + +describe("multi-message interjection (continuation-multi)", () => { + it("resumes and replays each queued message in order, streaming only the last", async () => { + // Turn 1: establishes the pooled agent + fingerprint (single user msg). + create.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + await drain(model(), [sys, user("a")]); + expect(create).toHaveBeenCalledOnce(); + + // Turn 2: two new user messages were queued while busy -> continuation-multi. + resume.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + sentTurns.length = 0; + await drain(model(), [sys, user("a"), assistant("x"), user("b"), user("c")]); + + // Resumed the pooled agent (not a fresh create). + expect(resume).toHaveBeenCalledWith("a1", expect.anything()); + // Two sequential sends: "b" silent, "c" streamed. + expect(sentTurns).toEqual([ + { text: "b", streamed: false }, + { text: "c", streamed: true }, + ]); + }); + + it("drops the session record when a silent send fails mid-sequence, so the next turn replays fresh", async () => { + // Turn 1: pool the agent. + create.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + await drain(model(), [sys, user("a")]); + expect(getSessionRecord(SESSION_ID)).toBeDefined(); + + // Turn 2: continuation-multi, but the FIRST (silent) send errors. + resume.mockResolvedValue( + makeFakeAgent("a1", sentTurns, { failOnText: "b" }), + ); + sentTurns.length = 0; + const parts = await drain(model(), [ + sys, + user("a"), + assistant("x"), + user("b"), + user("c"), + ]); + + // The failure surfaced as an error stream part, message "c" never sent. + expect(parts.some((p) => p.type === "error")).toBe(true); + expect(sentTurns).toEqual([{ text: "b", streamed: false }]); + // Record rolled back: next turn must NOT resume on top of undelivered msgs. + expect(getSessionRecord(SESSION_ID)).toBeUndefined(); + + // Turn 3 (same prompt retry): classifies "new" -> fresh agent, full replay. + create.mockClear(); + create.mockResolvedValue(makeFakeAgent("a2", sentTurns)); + resume.mockClear(); + sentTurns.length = 0; + await drain(model(), [sys, user("a"), assistant("x"), user("b"), user("c")]); + expect(create).toHaveBeenCalledOnce(); + expect(resume).not.toHaveBeenCalled(); + expect(sentTurns).toHaveLength(1); // one full-transcript send + expect(sentTurns[0]?.streamed).toBe(true); + }); + + it("keeps the session record when the full multi sequence delivers", async () => { + create.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + await drain(model(), [sys, user("a")]); + resume.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + await drain(model(), [sys, user("a"), user("b"), user("c")]); + expect(getSessionRecord(SESSION_ID)).toMatchObject({ agentId: "a1" }); + }); + + it("falls back to a full transcript replay when trailing user msgs are insufficient", async () => { + // Prime the pool. + create.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + await drain(model(), [sys, user("a")]); + + // A transcript whose fingerprint claims 2 new msgs but the tail is broken by + // an assistant turn -> classifier yields divergence, so full replay (fresh). + create.mockResolvedValue(makeFakeAgent("a2", sentTurns)); + sentTurns.length = 0; + await drain(model(), [sys, user("a"), user("b"), assistant("mid")]); + + // Last message not a user turn -> divergence -> fresh full replay, one send. + expect(sentTurns).toHaveLength(1); + expect(sentTurns[0]?.streamed).toBe(true); + }); +}); diff --git a/test/message-map.test.ts b/test/message-map.test.ts index b382546..19e567e 100644 --- a/test/message-map.test.ts +++ b/test/message-map.test.ts @@ -7,6 +7,7 @@ import type { LanguageModelV3Prompt } from "@ai-sdk/provider"; import { latestUserMessage, promptToCursorMessage, + trailingUserMessages, } from "../src/provider/message-map.js"; /** Write a temp file and return its absolute path + file:// URL string. */ @@ -369,3 +370,57 @@ describe("latestUserMessage", () => { expect(msg?.text).toContain("application/x-directory"); }); }); + +describe("trailingUserMessages", () => { + it("returns the last N user turns in conversation order", () => { + const prompt: LanguageModelV3Prompt = [ + { role: "system", content: "S" }, + { role: "user", content: [{ type: "text", text: "a" }] }, + { role: "assistant", content: [{ type: "text", text: "x" }] }, + { role: "user", content: [{ type: "text", text: "b" }] }, + { role: "user", content: [{ type: "text", text: "c" }] }, + ]; + expect(trailingUserMessages(prompt, 2)).toEqual([ + { text: "b" }, + { text: "c" }, + ]); + }); + + it("notes files on trailing turns the same way latestUserMessage does", () => { + const prompt: LanguageModelV3Prompt = [ + { role: "user", content: [{ type: "text", text: "first" }] }, + { + role: "user", + content: [ + { type: "text", text: "look" }, + { + type: "file", + data: "https://x/a.png", + mediaType: "image/png", + }, + ], + }, + ]; + const [msg] = trailingUserMessages(prompt, 1); + expect(msg?.images).toBeUndefined(); + expect(msg?.text).toContain("look"); + expect(msg?.text).toContain( + "[attached file: https://x/a.png (image/png) — not forwarded to Cursor]", + ); + }); + + it("returns only the available user turns when fewer than N exist", () => { + const prompt: LanguageModelV3Prompt = [ + { role: "user", content: [{ type: "text", text: "only" }] }, + ]; + expect(trailingUserMessages(prompt, 3)).toEqual([{ text: "only" }]); + }); + + it("returns an empty array when the final turn is not a user message", () => { + const prompt: LanguageModelV3Prompt = [ + { role: "user", content: [{ type: "text", text: "hi" }] }, + { role: "assistant", content: [{ type: "text", text: "bye" }] }, + ]; + expect(trailingUserMessages(prompt, 2)).toEqual([]); + }); +}); diff --git a/test/session-pool.test.ts b/test/session-pool.test.ts index 9c5b04f..61cd71b 100644 --- a/test/session-pool.test.ts +++ b/test/session-pool.test.ts @@ -16,6 +16,7 @@ vi.mock("../src/cursor-runtime.js", () => ({ const { acquireAgent, clearAgentPool, + dropSessionRecord, getPooledAgentId, getSessionRecord, resetSessionPoolMemory, @@ -160,6 +161,19 @@ describe("acquireAgent", () => { expect(r.agent.close).toHaveBeenCalled(); }); + it("dropSessionRecord removes the record from memory and disk", async () => { + create.mockResolvedValue(fakeAgent("a1")); + await acquireAgent({ ...base, poolKey: "s1", record: rec }); + expect(getSessionRecord("s1")).toBeDefined(); + + dropSessionRecord("s1"); + expect(getSessionRecord("s1")).toBeUndefined(); + + // The delete must persist: rehydration from disk must not resurrect it. + resetSessionPoolMemory(); + expect(getSessionRecord("s1")).toBeUndefined(); + }); + it("does not touch the pool when poolKey is omitted (side-call)", async () => { create.mockResolvedValueOnce(fakeAgent("a1")); await acquireAgent({ ...base, poolKey: "s1", record: rec }); diff --git a/test/transcript-fingerprint.test.ts b/test/transcript-fingerprint.test.ts index dc0fc31..d375ab6 100644 --- a/test/transcript-fingerprint.test.ts +++ b/test/transcript-fingerprint.test.ts @@ -41,6 +41,7 @@ describe("classifyTurn", () => { const c = classifyTurn(prev, turn2); expect(c.kind).toBe("continuation"); expect(c.fingerprint.userHashes).toHaveLength(2); + expect(c.newUserCount).toBe(1); }); it("returns 'side-call' when the system prompt changes (e.g. title gen)", () => { @@ -66,10 +67,36 @@ describe("classifyTurn", () => { expect(classifyTurn(prev, reverted).kind).toBe("divergence"); }); - it("returns 'divergence' when more than one new user message is queued", () => { + it("returns 'continuation-multi' when >=2 new user messages are appended to a strict prefix", () => { const prev = record([sys("S"), user("a")]); const queued = [sys("S"), user("a"), user("b"), user("c")]; - expect(classifyTurn(prev, queued).kind).toBe("divergence"); + const c = classifyTurn(prev, queued); + expect(c.kind).toBe("continuation-multi"); + expect(c.newUserCount).toBe(2); + expect(c.fingerprint.userHashes).toHaveLength(3); + }); + + it("returns 'divergence' when new user msgs are interleaved with assistant turns (non-contiguous tail)", () => { + // Interject -> abort mid-response -> interject again: opencode keeps the + // aborted assistant turn, so the new user msgs are NOT a contiguous tail. + // Resuming here would silently drop the earlier queued message, so this + // must fall back to divergence (fresh agent + full replay). + const prev = record([sys("S"), user("a")]); + const interleaved = [ + sys("S"), + user("a"), + assistant("x"), + user("b"), + assistant("partial"), + user("c"), + ]; + expect(classifyTurn(prev, interleaved).kind).toBe("divergence"); + }); + + it("returns 'divergence' when an earlier message is edited AND multiple new msgs are queued", () => { + const prev = record([sys("S"), user("a")]); + const editedPlusQueued = [sys("S"), user("EDITED"), user("b"), user("c")]; + expect(classifyTurn(prev, editedPlusQueued).kind).toBe("divergence"); }); it("returns 'divergence' when the last message is not a user turn", () => { From 9b3e224eef119ff59b04d57abe05e6cb4c0a3108 Mon Sep 17 00:00:00 2001 From: Justin Carper Date: Mon, 6 Jul 2026 14:37:47 -0500 Subject: [PATCH 2/3] test(session): cover abort between silent multi-sends Adds the one M4 gap the reviewer flagged that wasn't yet covered at the agentRun level: when the abort signal fires between sequential silent sends, the loop must stop before the next send, never stream the final turn, drop the session record, and still finish the stream cleanly with no error part. --- test/language-model-interjection.test.ts | 52 ++++++++++++++++++++++-- 1 file changed, 49 insertions(+), 3 deletions(-) diff --git a/test/language-model-interjection.test.ts b/test/language-model-interjection.test.ts index 0c1bf6c..2ee1218 100644 --- a/test/language-model-interjection.test.ts +++ b/test/language-model-interjection.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { LanguageModelV3Prompt } from "@ai-sdk/provider"; +import type { LanguageModelV3CallOptions } from "@ai-sdk/provider"; // Sandbox the on-disk session store away from the user's real cache dir. process.env.XDG_CACHE_HOME = mkdtempSync(join(tmpdir(), "cursor-lm-test-")); @@ -73,18 +74,20 @@ function model() { }); } -function callOptions(prompt: LanguageModelV3Prompt) { +function callOptions(prompt: LanguageModelV3Prompt, abortSignal?: AbortSignal) { return { prompt, providerOptions: { cursor: { sessionID: SESSION_ID } }, - } as never; + ...(abortSignal ? { abortSignal } : {}), + } as unknown as LanguageModelV3CallOptions; } async function drain( model: InstanceType, prompt: LanguageModelV3Prompt, + abortSignal?: AbortSignal, ): Promise> { - const { stream } = await model.doStream(callOptions(prompt)); + const { stream } = await model.doStream(callOptions(prompt, abortSignal)); const reader = stream.getReader(); const parts: Array<{ type: string }> = []; // eslint-disable-next-line no-constant-condition @@ -179,6 +182,49 @@ describe("multi-message interjection (continuation-multi)", () => { expect(getSessionRecord(SESSION_ID)).toMatchObject({ agentId: "a1" }); }); + it("stops sending and drops the record when abort fires between silent sends", async () => { + // Turn 1: pool the agent. + create.mockResolvedValue(makeFakeAgent("a1", sentTurns)); + await drain(model(), [sys, user("a")]); + expect(getSessionRecord(SESSION_ID)).toBeDefined(); + + // Turn 2: continuation-multi with THREE new msgs (b, c, d). The abort fires + // as soon as the first silent send ("b") resolves, so the loop's pre-send + // check breaks before "c" is ever sent and the final "d" never streams. + const controller = new AbortController(); + const abortingAgent = { + agentId: "a1", + close: vi.fn(), + send: async ( + message: { text: string }, + _sendOptions?: Record, + ) => { + sentTurns.push({ text: message.text, streamed: false }); + if (message.text === "b") controller.abort(); + return { + id: "run", + wait: async () => ({ status: "finished", result: message.text }), + cancel: async () => {}, + }; + }, + }; + resume.mockResolvedValue(abortingAgent); + sentTurns.length = 0; + + const parts = await drain( + model(), + [sys, user("a"), user("b"), user("c"), user("d")], + controller.signal, + ); + + // Only "b" was sent; "c"/"d" skipped once the signal was observed. + expect(sentTurns).toEqual([{ text: "b", streamed: false }]); + // No error part: an aborted multi turn finishes cleanly (empty stream). + expect(parts.some((p) => p.type === "error")).toBe(false); + // Record dropped: next turn must replay fresh, not resume on undelivered msgs. + expect(getSessionRecord(SESSION_ID)).toBeUndefined(); + }); + it("falls back to a full transcript replay when trailing user msgs are insufficient", async () => { // Prime the pool. create.mockResolvedValue(makeFakeAgent("a1", sentTurns)); From 84d916d2642a0ec6f7505e534e415032d9d40cd0 Mon Sep 17 00:00:00 2001 From: Justin Carper Date: Mon, 6 Jul 2026 14:55:46 -0500 Subject: [PATCH 3/3] fix(session): force cold replay instead of degrading multi-turn resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR #57 review findings: - language-model: on a count/tail mismatch (classifier invariant broken), never degrade to sending only the latest message — that silently drops messages 1..N-1 while the record keeps the full fingerprint. Clear the resume id pre-acquire so a fresh agent gets the full transcript instead. - agent-events: sendAgentTurnSilently now treats any terminal status other than "finished" as non-delivery and throws, except cancellation caused by our own abort signal (caller drops the record on abort). - document silent-turn trade-offs: tool invisibility, serial latency, and why concatenating queued messages was rejected (message fidelity). --- src/provider/agent-events.ts | 30 +++- src/provider/language-model.ts | 54 ++++--- test/agent-events.test.ts | 33 +++++ .../language-model-multiturn-fallback.test.ts | 140 ++++++++++++++++++ 4 files changed, 235 insertions(+), 22 deletions(-) create mode 100644 test/language-model-multiturn-fallback.test.ts diff --git a/src/provider/agent-events.ts b/src/provider/agent-events.ts index 16e17d4..aee3546 100644 --- a/src/provider/agent-events.ts +++ b/src/provider/agent-events.ts @@ -202,9 +202,22 @@ async function sendWithBusyRetry( * Honors `options.abortSignal`: an abort cancels the in-flight run so the * caller can stop before sending the next queued message. * - * Known trade-off: token usage from silent turns is not reported (no onDelta - * means the `turn-ended` usage update is never observed), so opencode slightly - * undercounts usage on multi-message turns. + * Known trade-offs of silent turns being FULL agent runs: + * - Tool invisibility: the agent may execute tools (shell, edits, MCP) during + * a silent turn with zero streamed output or tool display — the user sees + * nothing until the final message streams. Accepted because interjections + * are typically short course-corrections, and opencode itself folds + * interjected messages into one visible turn. + * - Serial latency: each silent turn is awaited to completion before the next + * send, so an N-message interjection costs N sequential agent runs. + * - Usage undercount: no onDelta means the `turn-ended` usage update is never + * observed, so opencode slightly undercounts tokens on multi-message turns. + * + * Concatenating the queued messages into one Cursor message was rejected for + * message fidelity: each interjection must land as a distinct user turn in the + * agent's conversation memory (mirroring opencode's transcript), so the model + * sees the same message boundaries the user created and later fingerprint + * classification stays aligned turn-for-turn. */ export async function sendAgentTurnSilently( agent: AgentLike, @@ -226,9 +239,16 @@ export async function sendAgentTurnSilently( // was populated, so onAbort had nothing to cancel); cancel now. if (options.abortSignal?.aborted) void Promise.resolve(run.cancel()).catch(() => {}); const result = await run.wait(); - if (result.status === "error") { + if (result.status !== "finished") { + // Our own abort cancelled the run mid-flight: expected, not a failure. + // The caller's abort check stops the multi-send sequence and drops the + // session record, so this partial turn is never counted as delivered. + if (options.abortSignal?.aborted) return; + // Anything else ("error", an external "cancelled", unknown states) means + // the message was NOT delivered; treating it as success would leave the + // session record claiming the agent saw a message it never received. throw new Error( - `Cursor run ended with status "error"${result.result ? `: ${result.result}` : ""}`, + `Cursor run ended with status "${result.status}"${result.result ? `: ${result.result}` : ""}`, ); } } finally { diff --git a/src/provider/language-model.ts b/src/provider/language-model.ts index c5ff4a8..7c36b1e 100644 --- a/src/provider/language-model.ts +++ b/src/provider/language-model.ts @@ -12,6 +12,7 @@ import type { McpServerConfig, SettingSource, AgentModeOption, + SDKUserMessage, } from "@cursor/sdk"; import { resolveCursorApiKey } from "../api-key.js"; import { @@ -209,6 +210,31 @@ export class CursorLanguageModel implements LanguageModelV3 { } } + // 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; + } + } + // 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.). @@ -235,26 +261,20 @@ export class CursorLanguageModel implements LanguageModelV3 { ...(resumeAgentId ? { resumeAgentId } : {}), }); - // 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, - // replay just those new messages as sequential turns — send the leading - // ones silently and stream only the final one. If the tail can't be - // recovered (defensive; shouldn't happen post-classifier), skip the multi - // path: on a resumed agent that means sending only the latest message, so - // the empty/short check below treats it as a normal single continuation. - const multiTurns = - acquired.resumed && multiNewUserCount >= 2 - ? trailingUserMessages(options.prompt, multiNewUserCount) - : undefined; - let yielded = false; let releasedOriginal = false; try { - if (multiTurns && multiTurns.length === multiNewUserCount) { - // A multi-message interjection on a resumed agent: send the leading - // messages silently and stream only the final one. - // + // Replay the queued messages as sequential turns: leading ones silent, + // only the final one streamed. Note silent turns are FULL agent runs — + // tools may execute with nothing surfaced until the last turn streams, + // and each run is awaited serially (see sendAgentTurnSilently for the + // trade-offs and why concatenation was rejected). + // + // Guard on acquired.resumed: multiTurns is computed before acquire (so a + // mismatched tail can clear resumeAgentId), but the silent-replay path + // only makes sense against a resumed agent. A fresh agent falls through + // to the full-transcript replay below. + if (acquired.resumed && multiTurns) { // The pool record was written optimistically with the FULL new // fingerprint before any send. If delivery stops partway (error or // abort), drop the record so the next turn classifies fresh instead diff --git a/test/agent-events.test.ts b/test/agent-events.test.ts index 21938af..0826cd6 100644 --- a/test/agent-events.test.ts +++ b/test/agent-events.test.ts @@ -140,6 +140,39 @@ describe("sendAgentTurnSilently", () => { ).rejects.toThrow(/error/i); }); + it("throws when the run ends 'cancelled' without our abort (message never delivered)", async () => { + // Cancellation we did NOT request (external cancel, CLI kill, …) means the + // silent turn was not delivered; treating it as success would let the + // caller keep a session record for a message the agent never received. + const agent = fakeAgent({ result: { status: "cancelled" } }); + await expect( + sendAgentTurnSilently(agent, MESSAGE, { mode: "agent" }), + ).rejects.toThrow(/cancelled/); + }); + + it("throws when the run ends with an unknown terminal status", async () => { + const agent = fakeAgent({ result: { status: "expired" } }); + await expect( + sendAgentTurnSilently(agent, MESSAGE, { mode: "agent" }), + ).rejects.toThrow(/expired/); + }); + + it("does not throw on 'cancelled' when our own abort signal caused it", async () => { + const controller = new AbortController(); + controller.abort(); + // Already-aborted signal: returns early without sending at all. + const sendCalls: Array | undefined> = []; + const agent = fakeAgent({ + result: { status: "cancelled" }, + sendCalls, + }); + await sendAgentTurnSilently(agent, MESSAGE, { + mode: "agent", + abortSignal: controller.signal, + }); + expect(sendCalls).toHaveLength(0); + }); + it("retries with local.force on AgentBusyError", async () => { const busy = new Error("agent busy"); busy.name = "AgentBusyError"; diff --git a/test/language-model-multiturn-fallback.test.ts b/test/language-model-multiturn-fallback.test.ts new file mode 100644 index 0000000..00938f5 --- /dev/null +++ b/test/language-model-multiturn-fallback.test.ts @@ -0,0 +1,140 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { + LanguageModelV3CallOptions, + LanguageModelV3Prompt, +} from "@ai-sdk/provider"; + +// Sandbox the on-disk session store away from the user's real cache dir. +process.env.XDG_CACHE_HOME = mkdtempSync(join(tmpdir(), "cursor-lm-fb-test-")); + +interface SentTurn { + text: string; + streamed: boolean; +} + +const sentTurns: SentTurn[] = []; +const create = vi.fn(); +const resume = vi.fn(); + +vi.mock("../src/cursor-runtime.js", () => ({ + loadCursorSdk: async () => ({ Agent: { create, resume } }), +})); + +// The classifier guarantees `continuation-multi` prompts end in a contiguous +// user tail of exactly `newUserCount` messages, so the count/tail mismatch the +// model guards against is unreachable through the public API. Construct it +// artificially: make trailingUserMessages return one message fewer than asked, +// as a stand-in for a future refactor breaking the classifier invariant. +vi.mock("../src/provider/message-map.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + trailingUserMessages: (prompt: LanguageModelV3Prompt, count: number) => + actual.trailingUserMessages(prompt, count).slice(1), + }; +}); + +const { CursorLanguageModel } = await import( + "../src/provider/language-model.js" +); +const { clearAgentPool, getSessionRecord } = await import( + "../src/provider/session-pool.js" +); + +const SESSION_ID = "sess-fallback"; + +function makeFakeAgent(agentId: string) { + return { + agentId, + close: vi.fn(), + send: async ( + message: { text: string }, + sendOptions?: Record, + ) => { + sentTurns.push({ + text: message.text, + streamed: Boolean(sendOptions?.["onDelta"]), + }); + return { + id: "run", + wait: async () => ({ status: "finished", result: message.text }), + cancel: async () => {}, + }; + }, + }; +} + +function model() { + return new CursorLanguageModel("cursor/auto", { + providerName: "cursor", + apiKey: "k", + cwd: "/tmp", + mode: "agent", + }); +} + +async function drain(prompt: LanguageModelV3Prompt): Promise { + const options = { + prompt, + providerOptions: { cursor: { sessionID: SESSION_ID } }, + } as unknown as LanguageModelV3CallOptions; + const { stream } = await model().doStream(options); + const reader = stream.getReader(); + // eslint-disable-next-line no-constant-condition + while (true) { + const { done } = await reader.read(); + if (done) break; + } +} + +const sys = { role: "system" as const, content: "S" }; +const user = (text: string): LanguageModelV3Prompt[number] => ({ + role: "user", + content: [{ type: "text", text }], +}); +const assistant = (text: string): LanguageModelV3Prompt[number] => ({ + role: "assistant", + content: [{ type: "text", text }], +}); + +afterEach(() => { + create.mockReset(); + resume.mockReset(); + sentTurns.length = 0; + clearAgentPool(); +}); + +describe("multi-turn tail mismatch (defensive fallback)", () => { + it("forces a cold full-transcript replay instead of degrading to the latest message", async () => { + // Turn 1: pool the agent with a single-user fingerprint. + create.mockResolvedValue(makeFakeAgent("a1")); + await drain([sys, user("a")]); + expect(getSessionRecord(SESSION_ID)).toBeDefined(); + + // Turn 2: continuation-multi (2 new user msgs), but the mocked + // trailingUserMessages recovers only 1 of them — the mismatch case. + // The model must NOT resume-and-send-only-"c" (which silently drops "b"); + // it must fall back to a fresh agent + full transcript so nothing is lost. + create.mockClear(); + create.mockResolvedValue(makeFakeAgent("a2")); + resume.mockResolvedValue(makeFakeAgent("a1")); + sentTurns.length = 0; + await drain([sys, user("a"), assistant("x"), user("b"), user("c")]); + + expect(resume).not.toHaveBeenCalled(); + expect(create).toHaveBeenCalledOnce(); + expect(sentTurns).toHaveLength(1); + expect(sentTurns[0]?.streamed).toBe(true); + // Full transcript: every queued message is present, none dropped. + expect(sentTurns[0]?.text).toContain("b"); + expect(sentTurns[0]?.text).toContain("c"); + + // The record now reflects the fully delivered transcript: the NEXT turn + // is a clean continuation on the fresh agent. + expect(getSessionRecord(SESSION_ID)).toMatchObject({ agentId: "a2" }); + }); +});