From 8a8ed84ab1828105e0c29d0d414418d9e64ba05c Mon Sep 17 00:00:00 2001 From: Ingwannu Date: Sat, 1 Aug 2026 07:09:44 +0000 Subject: [PATCH] bound Cursor blob store memory --- src/adapters/cursor/live-transport.ts | 66 ++++---- src/adapters/cursor/native-exec.ts | 197 ++++++++++++++++++++---- src/adapters/cursor/protobuf-request.ts | 94 +++++++---- structure/04_transports-and-sidecars.md | 15 ++ tests/cursor-blob.test.ts | 96 +++++++++++- tests/cursor-live-transport.test.ts | 40 ++++- 6 files changed, 418 insertions(+), 90 deletions(-) diff --git a/src/adapters/cursor/live-transport.ts b/src/adapters/cursor/live-transport.ts index 78ac830ce..c88fff887 100644 --- a/src/adapters/cursor/live-transport.ts +++ b/src/adapters/cursor/live-transport.ts @@ -522,44 +522,48 @@ class LiveCursorTransport implements CursorTransport { const prepared = prepareCursorRunRequest(request, { estimateInputTokens: contextUsage.carryForwardTokens === undefined, }); - state = createCursorProtobufEventState({ - clientToolNames: clientToolDefs.map(tool => tool.toolName || tool.name), - parallelToolCalls: request.parallelToolCalls, - toolSchemas, - cursorToolNameMap, - contextUsage, - ...(prepared.estimatedInputTokens !== undefined - ? { estimatedInputTokens: prepared.estimatedInputTokens } - : {}), - }); + try { + state = createCursorProtobufEventState({ + clientToolNames: clientToolDefs.map(tool => tool.toolName || tool.name), + parallelToolCalls: request.parallelToolCalls, + toolSchemas, + cursorToolNameMap, + contextUsage, + ...(prepared.estimatedInputTokens !== undefined + ? { estimatedInputTokens: prepared.estimatedInputTokens } + : {}), + }); - this.open(prepared.bytes, signal, state, push, err => { - failure = err; - wake(); - }, () => { - done = true; - wake(); - }); + this.open(prepared.bytes, signal, state, push, err => { + failure = err; + wake(); + }, () => { + done = true; + wake(); + }); - while (!done || queue.length > 0) { - while (queue.length > 0) { - const message = queue.shift(); - if (message) yield message; + while (!done || queue.length > 0) { + while (queue.length > 0) { + const message = queue.shift(); + if (message) yield message; + } + if (failure) { + // A CANCEL is benign only on the client-tool suspend path (expectedClose); an + // unexpected server-side NGHTTP2_CANCEL must surface as a real transport error. + if (this.expectedClose && isCursorBenignCancelError(failure)) return; + throw attachPartialUsage(summarizeFailure(failure), state); + } + if (done) break; + await new Promise(resolve => { + notify = resolve; + }); } if (failure) { - // A CANCEL is benign only on the client-tool suspend path (expectedClose); an - // unexpected server-side NGHTTP2_CANCEL must surface as a real transport error. if (this.expectedClose && isCursorBenignCancelError(failure)) return; throw attachPartialUsage(summarizeFailure(failure), state); } - if (done) break; - await new Promise(resolve => { - notify = resolve; - }); - } - if (failure) { - if (this.expectedClose && isCursorBenignCancelError(failure)) return; - throw attachPartialUsage(summarizeFailure(failure), state); + } finally { + prepared.releaseBlobs(); } } diff --git a/src/adapters/cursor/native-exec.ts b/src/adapters/cursor/native-exec.ts index f5ae53aef..e035ad1a7 100644 --- a/src/adapters/cursor/native-exec.ts +++ b/src/adapters/cursor/native-exec.ts @@ -3,6 +3,7 @@ import { create } from "@bufbuild/protobuf"; import { DiagnosticsErrorSchema, DiagnosticsResultSchema, + ErrorSchema, GetBlobResultSchema, KvClientMessageSchema, McpErrorSchema, @@ -77,47 +78,133 @@ export function cursorUnsafeNativeLocalExecEnabled(input: Pick(); +const BLOB_MAX_ENTRY_BYTES = 16 * 1024 * 1024; +const BLOB_MAX_TOTAL_BYTES = 64 * 1024 * 1024; +const BLOB_SWEEP_INTERVAL_MS = 60 * 1000; -function evictStaleBlobs(now: number): void { - if (blobs.size <= BLOB_MAX_ENTRIES) { - // TTL sweep only when the map has any chance of stale entries; Map iterates insertion order. +interface BlobEntry { + data: Uint8Array; + storedAt: number; + pins: number; +} + +interface BlobLimits { + maxEntries: number; + maxEntryBytes: number; + maxTotalBytes: number; +} + +export interface CursorBlobScope { + readonly keys: Set; + released: boolean; +} + +export class CursorBlobAdmissionError extends Error { + readonly code = "cursor_blob_admission_failed"; + constructor(message: string) { + super(message); + this.name = "CursorBlobAdmissionError"; + } +} + +const DEFAULT_BLOB_LIMITS: BlobLimits = { + maxEntries: BLOB_MAX_ENTRIES, + maxEntryBytes: BLOB_MAX_ENTRY_BYTES, + maxTotalBytes: BLOB_MAX_TOTAL_BYTES, +}; +const blobs = new Map(); +let blobBytes = 0; +let lastBlobSweepAt = 0; +let blobLimits = { ...DEFAULT_BLOB_LIMITS }; + +function deleteBlob(k: string): void { + const entry = blobs.get(k); + if (!entry) return; + blobBytes -= entry.data.byteLength; + if (blobBytes < 0) blobBytes = 0; + blobs.delete(k); +} + +function sweepStaleBlobs(now: number, force = false, protectedKey?: string): void { + if (!force && now - lastBlobSweepAt < BLOB_SWEEP_INTERVAL_MS) return; + lastBlobSweepAt = now; + for (const [k, entry] of blobs) { + if (k !== protectedKey && entry.pins === 0 && now - entry.storedAt > BLOB_TTL_MS) deleteBlob(k); + } +} + +function ensureBlobCapacity( + additionalBytes: number, + additionalEntries: number, + now: number, + protectedKey?: string, +): boolean { + const pressured = blobs.size + additionalEntries > blobLimits.maxEntries + || blobBytes + additionalBytes > blobLimits.maxTotalBytes; + sweepStaleBlobs(now, pressured, protectedKey); + while ( + blobs.size + additionalEntries > blobLimits.maxEntries + || blobBytes + additionalBytes > blobLimits.maxTotalBytes + ) { + let victim: string | undefined; for (const [k, entry] of blobs) { - if (now - entry.storedAt <= BLOB_TTL_MS) break; - blobs.delete(k); + if (k !== protectedKey && entry.pins === 0) { + victim = k; + break; + } } - return; - } - // Over cap: drop oldest entries first (insertion order approximates recency because re-stores - // delete+set to refresh their position). - const excess = blobs.size - BLOB_MAX_ENTRIES; - let dropped = 0; - for (const k of blobs.keys()) { - if (dropped >= excess) break; - blobs.delete(k); - dropped++; + if (victim === undefined) return false; + deleteBlob(victim); } + return true; } -function setBlob(k: string, data: Uint8Array): void { +function setBlob(k: string, data: Uint8Array, contentAddressed = false): { ok: true } | { ok: false; error: string } { + if (data.byteLength > blobLimits.maxEntryBytes) { + return { ok: false, error: `blob exceeds the ${blobLimits.maxEntryBytes}-byte per-entry limit` }; + } const now = Date.now(); - blobs.delete(k); // refresh insertion order so live sessions stay newest - blobs.set(k, { data, storedAt: now }); - evictStaleBlobs(now); + const existing = blobs.get(k); + if (existing && contentAddressed && sameBytes(existing.data, data)) { + existing.storedAt = now; + blobs.delete(k); + blobs.set(k, existing); + return { ok: true }; + } + if (existing?.pins) { + return { ok: false, error: "blob is pinned by an active Cursor request" }; + } + const additionalBytes = data.byteLength - (existing?.data.byteLength ?? 0); + if (!ensureBlobCapacity(Math.max(0, additionalBytes), existing ? 0 : 1, now, k)) { + return { ok: false, error: "blob store is full of active request data" }; + } + if (existing) deleteBlob(k); + blobs.set(k, { data, storedAt: now, pins: 0 }); + blobBytes += data.byteLength; + return { ok: true }; +} + +function sameBytes(a: Uint8Array, b: Uint8Array): boolean { + if (a.byteLength !== b.byteLength) return false; + for (let i = 0; i < a.byteLength; i++) if (a[i] !== b[i]) return false; + return true; } function getBlob(k: string): Uint8Array | undefined { const entry = blobs.get(k); if (!entry) return undefined; - if (Date.now() - entry.storedAt > BLOB_TTL_MS) { - blobs.delete(k); + if (entry.pins === 0 && Date.now() - entry.storedAt > BLOB_TTL_MS) { + deleteBlob(k); return undefined; } + blobs.delete(k); + blobs.set(k, entry); return entry.data; } @@ -130,12 +217,61 @@ function key(bytes: Uint8Array): string { * blob id. Cursor's `rootPromptMessagesJson`/turn entries are blob IDS, not inline content — the * server fetches the bytes back via `getBlobArgs`. Mirrors jawcode `createBlobId`/`storeCursorBlob`. */ -export function storeCursorBlob(data: Uint8Array): Uint8Array { +export function createCursorBlobScope(): CursorBlobScope { + return { keys: new Set(), released: false }; +} + +export function releaseCursorBlobScope(scope: CursorBlobScope): void { + if (scope.released) return; + scope.released = true; + for (const k of scope.keys) { + const entry = blobs.get(k); + if (entry && entry.pins > 0) entry.pins -= 1; + } + scope.keys.clear(); + ensureBlobCapacity(0, 0, Date.now()); +} + +function pinCursorBlob(k: string, scope: CursorBlobScope): void { + if (scope.released || scope.keys.has(k)) return; + const entry = blobs.get(k); + if (!entry) throw new CursorBlobAdmissionError("Cursor blob disappeared before request pinning"); + entry.pins += 1; + scope.keys.add(k); +} + +export function storeCursorBlob(data: Uint8Array, scope?: CursorBlobScope): Uint8Array { const blobId = new Uint8Array(createHash("sha256").update(data).digest()); - setBlob(key(blobId), data); + const blobKey = key(blobId); + const stored = setBlob(blobKey, data, true); + if (!stored.ok) throw new CursorBlobAdmissionError(`Cursor blob admission failed: ${stored.error}`); + if (scope) pinCursorBlob(blobKey, scope); return blobId; } +export function cursorBlobMetricsForTests(): { + count: number; + totalBytes: number; + pinnedEntries: number; + pinnedBytes: number; +} { + let pinnedEntries = 0; + let pinnedBytes = 0; + for (const entry of blobs.values()) { + if (entry.pins === 0) continue; + pinnedEntries += 1; + pinnedBytes += entry.data.byteLength; + } + return { count: blobs.size, totalBytes: blobBytes, pinnedEntries, pinnedBytes }; +} + +export function setCursorBlobLimitsForTests(overrides: Partial | null): void { + blobs.clear(); + blobBytes = 0; + lastBlobSweepAt = 0; + blobLimits = overrides ? { ...DEFAULT_BLOB_LIMITS, ...overrides } : { ...DEFAULT_BLOB_LIMITS }; +} + export async function handleCursorNativeExec(execMsg: ExecServerMessage, deps: CursorNativeExecContext = {}): Promise { const execCase = execMsg.message.case; if (execCase === "requestContextArgs") { @@ -213,13 +349,18 @@ export function handleCursorNativeKv(kvMsg: KvServerMessage): Uint8Array { }); } if (kvMsg.message.case === "setBlobArgs") { - setBlob(key(kvMsg.message.value.blobId), kvMsg.message.value.blobData); + const stored = setBlob(key(kvMsg.message.value.blobId), kvMsg.message.value.blobData); return clientBytes({ message: { case: "kvClientMessage", value: create(KvClientMessageSchema, { id: kvMsg.id, - message: { case: "setBlobResult", value: create(SetBlobResultSchema, {}) }, + message: { + case: "setBlobResult", + value: create(SetBlobResultSchema, stored.ok ? {} : { + error: create(ErrorSchema, { message: `Cursor blob rejected: ${stored.error}` }), + }), + }, }), }, }); diff --git a/src/adapters/cursor/protobuf-request.ts b/src/adapters/cursor/protobuf-request.ts index c7030bf5b..43c1d40e4 100644 --- a/src/adapters/cursor/protobuf-request.ts +++ b/src/adapters/cursor/protobuf-request.ts @@ -6,7 +6,12 @@ import { namespacedToolName } from "../../types"; import type { CursorRunRequest } from "./types"; import { isCursorExternalWireModel } from "./discovery"; import { debugProviderDiagnostic } from "../../lib/debug"; -import { storeCursorBlob } from "./native-exec"; +import { + createCursorBlobScope, + releaseCursorBlobScope, + storeCursorBlob, + type CursorBlobScope, +} from "./native-exec"; import { estimateTokens } from "../../lib/token-estimate"; import { AgentClientMessageSchema, @@ -83,12 +88,12 @@ function jsonBlob(value: unknown): { data: Uint8Array; serialized: string } { } type StoredRootBlob = { - id: Uint8Array; + data: Uint8Array; byteLength: number; /** - * The exact JSON handed to storeCursorBlob(). Retained so a token estimate can read - * what the wire actually carries without re-serializing — and without drifting from - * it after pruning or truncation (#373). + * The exact candidate JSON. Retained so the selected suffix can be stored once and a token + * estimate can read what the wire actually carries without re-serializing or drifting after + * pruning/truncation (#373). */ serialized: string; role: "system" | "user" | "assistant" | "toolResult"; @@ -104,7 +109,7 @@ function storedRootBlob( ): StoredRootBlob { const { data, serialized } = jsonBlob(value); return { - id: storeCursorBlob(data), + data, byteLength: data.byteLength, serialized, role, @@ -169,7 +174,7 @@ function assistantRootText( // because it travels in the action. Tool results are rendered as user-role text with a marker, and // each entry is a SHA-256 blob ID (Cursor fetches the bytes back via getBlobArgs). Mirrors the // danger-pi reference buildRootPromptMessagesJson. -function rootPromptMessages(request: CursorRunRequest): { +function rootPromptMessages(request: CursorRunRequest, scope: CursorBlobScope): { ids: Uint8Array[]; byteLength: number; historyMessageStart: number; @@ -181,7 +186,7 @@ function rootPromptMessages(request: CursorRunRequest): { const messages = request.rawMessages; if (!messages?.length) { return { - ids: entries.map(entry => entry.id), + ids: entries.map(entry => storeCursorBlob(entry.data, scope)), byteLength: entries.reduce((sum, entry) => sum + entry.byteLength, 0), historyMessageStart: 0, serialized: entries.map(entry => entry.serialized), @@ -298,7 +303,9 @@ function rootPromptMessages(request: CursorRunRequest): { } return { - ids: selected.map(entry => entry.id), + // Store only the selected roots. Pre-storing every candidate retained discarded history until + // TTL eviction and made the external replay count/byte budget ineffective as a RAM bound. + ids: selected.map(entry => storeCursorBlob(entry.data, scope)), byteLength: selected.reduce((sum, entry) => sum + entry.byteLength, 0), historyMessageStart, serialized: selected.map(entry => entry.serialized), @@ -345,7 +352,11 @@ function argBytes(value: unknown): Uint8Array { } } -function toolCallStep(part: Extract, result?: OcxToolResultMessage): Uint8Array { +function toolCallStep( + part: Extract, + scope: CursorBlobScope, + result?: OcxToolResultMessage, +): Uint8Array { const args: Record = {}; for (const [key, value] of Object.entries(part.arguments ?? {})) args[key] = argBytes(value); const toolName = namespacedToolName(part.namespace, part.name); @@ -368,7 +379,7 @@ function toolCallStep(part: Extract>(); const flush = () => { if (!current) return; - for (const part of pendingToolCalls.values()) current.steps.push(toolCallStep(part)); + for (const part of pendingToolCalls.values()) current.steps.push(toolCallStep(part, scope)); turns.push(storeCursorBlob(toBinary(ConversationTurnStructureSchema, create(ConversationTurnStructureSchema, { turn: { case: "agentConversationTurn", value: create(AgentConversationTurnStructureSchema, current), }, - })))); + })), scope)); current = undefined; pendingToolCalls.clear(); }; @@ -451,7 +466,7 @@ function conversationTurns(request: CursorRunRequest, historyMessageStart = 0): case: "assistantMessage", value: create(AssistantMessageSchema, { text: part.text }), }, - })))); + })), scope)); } continue; } @@ -459,7 +474,7 @@ function conversationTurns(request: CursorRunRequest, historyMessageStart = 0): pendingToolCalls.set(part.id, part); continue; } - const step = assistantStep(part); + const step = assistantStep(part, scope); if (step) current.steps.push(step); } continue; @@ -473,12 +488,12 @@ function conversationTurns(request: CursorRunRequest, historyMessageStart = 0): case: "assistantMessage", value: create(AssistantMessageSchema, { text: `${prefix}\n${contentToText(message.content)}` }), }, - })))); + })), scope)); continue; } const priorCall = pendingToolCalls.get(message.toolCallId); if (priorCall) { - current.steps.push(toolCallStep(priorCall, message)); + current.steps.push(toolCallStep(priorCall, scope, message)); pendingToolCalls.delete(message.toolCallId); } else { current.steps.push(storeCursorBlob(toBinary(ConversationStepSchema, create(ConversationStepSchema, { @@ -486,7 +501,7 @@ function conversationTurns(request: CursorRunRequest, historyMessageStart = 0): case: "assistantMessage", value: create(AssistantMessageSchema, { text: toolResultToText(message) }), }, - })))); + })), scope)); } continue; } @@ -495,7 +510,7 @@ function conversationTurns(request: CursorRunRequest, historyMessageStart = 0): userMessage: storeCursorBlob(toBinary(UserMessageSchema, create(UserMessageSchema, { text: contentText(message), messageId: crypto.randomUUID(), - }))), + })), scope), steps: [], }; } @@ -537,6 +552,8 @@ function modelVisibleToolText(definition: McpToolDefinition): string { export interface PreparedCursorRunRequest { bytes: Uint8Array; + /** Release request pins on every terminal path; idempotent for generator cancellation/cleanup races. */ + releaseBlobs: () => void; /** Only present when the caller asked for it; see prepareCursorRunRequest(). */ estimatedInputTokens?: number; } @@ -555,6 +572,20 @@ export interface PreparedCursorRunRequest { export function prepareCursorRunRequest( request: CursorRunRequest, options?: { estimateInputTokens?: boolean }, +): PreparedCursorRunRequest { + const scope = createCursorBlobScope(); + try { + return prepareCursorRunRequestWithScope(request, options, scope); + } catch (error) { + releaseCursorBlobScope(scope); + throw error; + } +} + +function prepareCursorRunRequestWithScope( + request: CursorRunRequest, + options: { estimateInputTokens?: boolean } | undefined, + scope: CursorBlobScope, ): PreparedCursorRunRequest { const rawText = activePromptText(request); const lastRole = request.messages.at(-1)?.role; @@ -585,9 +616,9 @@ export function prepareCursorRunRequest( }), }, }); - const rootPromptMessagesState = rootPromptMessages(request); + const rootPromptMessagesState = rootPromptMessages(request, scope); const rootPromptMessageIds = rootPromptMessagesState.ids; - const turnIds = conversationTurns(request, rootPromptMessagesState.historyMessageStart); + const turnIds = conversationTurns(request, rootPromptMessagesState.historyMessageStart, scope); // Hoisted out of the mcp_tools spread below so the estimate can read the same // filtered definitions the wire carries. Both helpers are pure. const visibleTools = cursorToolsForActivePrompt(request.tools, rawText, request.toolChoice); @@ -661,7 +692,8 @@ export function prepareCursorRunRequest( message: { case: "runRequest", value: runRequest }, }); const bytes = toBinary(AgentClientMessageSchema, message); - if (!options?.estimateInputTokens) return { bytes }; + const releaseBlobs = () => releaseCursorBlobScope(scope); + if (!options?.estimateInputTokens) return { bytes, releaseBlobs }; // Same instances that produced `bytes`, so the estimate cannot count history or // tools the payload dropped — the defect that blocked PR #376. @@ -672,11 +704,17 @@ export function prepareCursorRunRequest( ]; return { bytes, + releaseBlobs, estimatedInputTokens: estimateTokens(modelVisibleParts.join("\n"), request.modelId), }; } /** Back-compat wrapper: callers that only need the wire bytes. */ export function encodeCursorRunRequest(request: CursorRunRequest): Uint8Array { - return prepareCursorRunRequest(request).bytes; + const prepared = prepareCursorRunRequest(request); + try { + return prepared.bytes; + } finally { + prepared.releaseBlobs(); + } } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 5a39c59e2..9120819c5 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -302,6 +302,21 @@ pre-compaction checkpoint is not persisted for later carry-forward. ## OpenRouter provider routing +Cursor prompt and conversation blobs use a process-wide content-addressed store capped at 16 MiB +per blob and 64 MiB total. Blobs referenced by a prepared request are pinned until that request +finishes, fails, or its async generator is cancelled; admission evicts only expired or +least-recently-used unpinned entries. External replay candidates are stored only after pruning, so +dropped history never enters the shared store. A server `setBlobArgs` that cannot fit receives a +protobuf error instead of a false success acknowledgement. + +[Decision Log] +- 목적과 의도: Bound Cursor blob RAM without evicting data that an active request has already advertised to the server. +- 기존 구현 및 제약 조건: The store was limited only to 4,096 entries, had no byte limits, and root candidates were stored before replay pruning; naive LRU eviction could break an in-flight blob fetch. +- 검토한 주요 대안: Count-only eviction; global byte LRU with no pins; request-scoped stores; shared byte LRU with request leases. +- 선택한 방식: Keep deduplicating content-addressed storage, add per-entry/global bytes, pin every advertised root/step/turn for the request lifetime, and evict only unpinned LRU entries. +- 다른 대안 대신 이 방식을 선택한 이유: Request-local stores duplicate shared history, while unpinned global eviction can turn a valid request into a mid-stream `Blob not found` failure. +- 장점, 단점 및 영향: Long-running heap is bounded and active references remain valid; a request whose own unique blobs cannot fit 64 MiB fails preparation explicitly rather than silently dropping state. + The canonical OpenRouter `openai-chat` transport may carry optional provider-routing preferences from `OcxProviderConfig.openRouterRouting`, with exact model-id replacements in `modelOpenRouterRouting`. The adapter maps camel-case config to OpenRouter's request wire diff --git a/tests/cursor-blob.test.ts b/tests/cursor-blob.test.ts index e25e0c925..b82df20c0 100644 --- a/tests/cursor-blob.test.ts +++ b/tests/cursor-blob.test.ts @@ -1,7 +1,15 @@ -import { describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, test } from "bun:test"; import { createHash } from "node:crypto"; import { create, fromBinary } from "@bufbuild/protobuf"; -import { handleCursorNativeKv, storeCursorBlob } from "../src/adapters/cursor/native-exec"; +import { + createCursorBlobScope, + CursorBlobAdmissionError, + cursorBlobMetricsForTests, + handleCursorNativeKv, + releaseCursorBlobScope, + setCursorBlobLimitsForTests, + storeCursorBlob, +} from "../src/adapters/cursor/native-exec"; import { CURSOR_EXTERNAL_ROOT_BYTE_LIMIT, CURSOR_EXTERNAL_ROOT_BLOB_LIMIT, @@ -16,8 +24,11 @@ import { ConversationTurnStructureSchema, GetBlobArgsSchema, KvServerMessageSchema, + SetBlobArgsSchema, } from "../src/adapters/cursor/gen/agent_pb"; +afterEach(() => setCursorBlobLimitsForTests(null)); + function sha256(data: Uint8Array): Uint8Array { return new Uint8Array(createHash("sha256").update(data).digest()); } @@ -62,6 +73,82 @@ describe("Cursor blob handshake", () => { expect(Array.from(id)).toEqual(Array.from(sha256(data))); }); + test("rejects a blob above the per-entry byte limit without retaining it", () => { + setCursorBlobLimitsForTests({ maxEntryBytes: 4, maxTotalBytes: 8 }); + expect(() => storeCursorBlob(Uint8Array.of(1, 2, 3, 4, 5))) + .toThrow(CursorBlobAdmissionError); + expect(cursorBlobMetricsForTests()) + .toEqual({ count: 0, totalBytes: 0, pinnedEntries: 0, pinnedBytes: 0 }); + }); + + test("unpinned blobs use LRU eviction under the total byte budget", () => { + setCursorBlobLimitsForTests({ maxEntryBytes: 4, maxTotalBytes: 8 }); + const first = storeCursorBlob(Uint8Array.of(1, 1, 1, 1)); + const second = storeCursorBlob(Uint8Array.of(2, 2, 2, 2)); + expect(blobData(first)).toEqual(Uint8Array.of(1, 1, 1, 1)); // touch first + const third = storeCursorBlob(Uint8Array.of(3, 3, 3, 3)); + + expect(blobData(first)).toEqual(Uint8Array.of(1, 1, 1, 1)); + expect(blobData(second)?.byteLength ?? 0).toBe(0); + expect(blobData(third)).toEqual(Uint8Array.of(3, 3, 3, 3)); + expect(cursorBlobMetricsForTests().totalBytes).toBe(8); + }); + + test("active request pins prevent eviction and release idempotently", () => { + setCursorBlobLimitsForTests({ maxEntryBytes: 4, maxTotalBytes: 8 }); + const scope = createCursorBlobScope(); + storeCursorBlob(Uint8Array.of(1, 1, 1, 1), scope); + storeCursorBlob(Uint8Array.of(2, 2, 2, 2), scope); + expect(cursorBlobMetricsForTests()).toMatchObject({ pinnedEntries: 2, pinnedBytes: 8 }); + expect(() => storeCursorBlob(Uint8Array.of(3, 3, 3, 3))) + .toThrow(CursorBlobAdmissionError); + + releaseCursorBlobScope(scope); + releaseCursorBlobScope(scope); + expect(cursorBlobMetricsForTests().pinnedBytes).toBe(0); + expect(() => storeCursorBlob(Uint8Array.of(3, 3, 3, 3))).not.toThrow(); + expect(cursorBlobMetricsForTests().totalBytes).toBeLessThanOrEqual(8); + }); + + test("setBlobArgs returns a protobuf error when admission is refused", () => { + setCursorBlobLimitsForTests({ maxEntryBytes: 4, maxTotalBytes: 8 }); + const reply = fromBinary(AgentClientMessageSchema, handleCursorNativeKv(create(KvServerMessageSchema, { + id: 7, + message: { + case: "setBlobArgs", + value: create(SetBlobArgsSchema, { + blobId: Uint8Array.of(9), + blobData: Uint8Array.of(1, 2, 3, 4, 5), + }), + }, + }))); + expect(reply.message.case).toBe("kvClientMessage"); + const result = reply.message.value.message; + expect(result.case).toBe("setBlobResult"); + expect(result.value.error?.message).toContain("per-entry limit"); + }); + + test("prepared requests pin advertised blobs until their release callback", () => { + const prepared = prepareCursorRunRequest({ + modelId: "composer-2.5", + conversationId: "c-pinned", + system: ["system"], + messages: [{ role: "user", content: "hello" }], + }); + expect(cursorBlobMetricsForTests().pinnedEntries).toBeGreaterThan(0); + prepared.releaseBlobs(); + prepared.releaseBlobs(); + expect(cursorBlobMetricsForTests().pinnedEntries).toBe(0); + + encodeCursorRunRequest({ + modelId: "composer-2.5", + conversationId: "c-backcompat-release", + system: ["system"], + messages: [{ role: "user", content: "hello" }], + }); + expect(cursorBlobMetricsForTests().pinnedEntries).toBe(0); + }); + test("encodeCursorRunRequest sends rootPromptMessagesJson as blob IDs, not inline JSON", () => { const bytes = encodeCursorRunRequest({ modelId: "claude-4.6-opus-high", @@ -113,6 +200,11 @@ describe("Cursor blob handshake", () => { expect((roots[1] as { role?: string }).role).toBe("user"); expect(JSON.stringify(roots)).toContain("assistant-209"); expect(JSON.stringify(roots)).not.toContain("user-0"); + const discardedRoot = sha256(new TextEncoder().encode(JSON.stringify({ + role: "user", + content: [{ type: "text", text: "user-0" }], + }))); + expect(blobData(discardedRoot)?.byteLength ?? 0).toBe(0); }); test("caps external root replay by serialized bytes", () => { diff --git a/tests/cursor-live-transport.test.ts b/tests/cursor-live-transport.test.ts index f3efee05c..af719baaa 100644 --- a/tests/cursor-live-transport.test.ts +++ b/tests/cursor-live-transport.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test"; import { createLiveCursorTransport, CursorMissingCredentialError, parseConnectEndStreamError, resolveCursorToken } from "../src/adapters/cursor/live-transport"; +import { cursorBlobMetricsForTests, setCursorBlobLimitsForTests } from "../src/adapters/cursor/native-exec"; import { prepareCursorRunRequest } from "../src/adapters/cursor/protobuf-request"; describe("Cursor live transport", () => { @@ -54,6 +55,39 @@ describe("Cursor live transport", () => { await expect(iterator.next()).rejects.toThrow("Cursor MCP preparation failed: fixture discovery failed"); await transport.close?.(); }); + + test("releases prepared blob pins when the consumer cancels the async generator", async () => { + setCursorBlobLimitsForTests(null); + const transport = createLiveCursorTransport({ + provider: { adapter: "cursor", baseUrl: "https://api2.cursor.sh", apiKey: "test-token" }, + headers: new Headers(), + }); + const internals = transport as unknown as { + open( + encoded: Uint8Array, + signal: AbortSignal | undefined, + state: unknown, + push: (message: { type: "heartbeat" }) => void, + ): void; + }; + internals.open = (_encoded, _signal, _state, push) => push({ type: "heartbeat" }); + + try { + const iterator = transport.run({ + modelId: "composer-2.5", + conversationId: "cancel-releases-blobs", + system: ["system"], + messages: [{ role: "user", content: "hello" }], + })[Symbol.asyncIterator](); + expect((await iterator.next()).value).toEqual({ type: "heartbeat" }); + expect(cursorBlobMetricsForTests().pinnedEntries).toBeGreaterThan(0); + await iterator.return?.(); + expect(cursorBlobMetricsForTests().pinnedEntries).toBe(0); + } finally { + await transport.close?.(); + setCursorBlobLimitsForTests(null); + } + }); }); describe("Cursor end-stream classification", () => { @@ -170,7 +204,11 @@ describe("Cursor live transport context estimate wiring (#373)", () => { expect(estimate).toBeGreaterThan(0); // And it must match what the same payload produces on its own. const prepared = prepareCursorRunRequest(baseRequest as never, { estimateInputTokens: true }); - expect(estimate).toBe(prepared.estimatedInputTokens); + try { + expect(estimate).toBe(prepared.estimatedInputTokens); + } finally { + prepared.releaseBlobs(); + } }); test("the estimate is skipped when the conversation carries a checkpoint forward", async () => {