diff --git a/src/provider/agent-events.ts b/src/provider/agent-events.ts index a5efa66..aee3546 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,95 @@ 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-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, + 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 !== "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 "${result.status}"${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..7c36b1e 100644 --- a/src/provider/language-model.ts +++ b/src/provider/language-model.ts @@ -12,17 +12,30 @@ import type { McpServerConfig, SettingSource, AgentModeOption, + SDKUserMessage, } 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 +159,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 +172,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 +183,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,13 +201,40 @@ 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}`, ); } } + // 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.). @@ -212,82 +261,121 @@ 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); - 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; + // 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 + // 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; + } + await sendAgentTurnSilently(acquired.agent, multiTurns[i]!, { + mode, + abortSignal: options.abortSignal, + }); + } + 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); } - } 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", - ); + } 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; } - 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. + } 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] original resume failure (not attachable as cause):", - err, + "[cursor:debug] resumed turn failed before emitting; retrying with a fresh agent", ); } - throw retryErr; - } - try { - const replay = promptToCursorMessage(options.prompt); - yield* streamAgentTurn(retry.agent, replay, { - mode, - abortSignal: options.abortSignal, - }); - } finally { - retry.release(); + 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; } - } 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..0826cd6 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,115 @@ 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("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"; + 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..2ee1218 --- /dev/null +++ b/test/language-model-interjection.test.ts @@ -0,0 +1,243 @@ +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"; +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-")); + +/** + * 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, abortSignal?: AbortSignal) { + return { + prompt, + providerOptions: { cursor: { sessionID: SESSION_ID } }, + ...(abortSignal ? { abortSignal } : {}), + } as unknown as LanguageModelV3CallOptions; +} + +async function drain( + model: InstanceType, + prompt: LanguageModelV3Prompt, + abortSignal?: AbortSignal, +): Promise> { + const { stream } = await model.doStream(callOptions(prompt, abortSignal)); + 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("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)); + 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/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" }); + }); +}); 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", () => {