From 435a35ddaefbdc4d812bb802ba2094e54eb49ae3 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Mon, 5 Oct 2026 07:11:58 +0000 Subject: [PATCH 1/3] fix(provider): bound screenshot cache by active references --- packages/opencode/src/provider/image-files.ts | 545 +++++++++++++----- .../provider/image-files-lifecycle.test.ts | 424 ++++++++++++++ .../test/provider/image-files.test.ts | 4 +- 3 files changed, 832 insertions(+), 141 deletions(-) create mode 100644 packages/opencode/test/provider/image-files-lifecycle.test.ts diff --git a/packages/opencode/src/provider/image-files.ts b/packages/opencode/src/provider/image-files.ts index 850fb8b83..9062a67ce 100644 --- a/packages/opencode/src/provider/image-files.ts +++ b/packages/opencode/src/provider/image-files.ts @@ -2,9 +2,17 @@ import type { LanguageModelV3, LanguageModelV3CallOptions } from "@ai-sdk/provid type Uploaded = { file_id: string; expires_at: number; signature: string } type Options = { baseURL: string; apiKey: string; fetch?: typeof fetch; now?: () => number } +type Lease = { keys: Set; files: Set } type RunCache = { + leases: Set + lock: Promise + expiresAt: number + chain: boolean + chainKeys: Set owners: Set cache: Map> + ready: Map + busy: boolean refreshed: Set retired: Uploaded[] waiting: Array<() => void> @@ -15,8 +23,15 @@ const runCaches = new Map() export function withImageFiles(model: LanguageModelV3, options: Options) { let state: RunCache = { + leases: new Set(), + lock: Promise.resolve(), + expiresAt: 0, + chain: false, + chainKeys: new Set(), owners: new Set(), cache: new Map(), + ready: new Map(), + busy: false, refreshed: new Set(), retired: [], waiting: [], @@ -28,6 +43,7 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { const now = options.now ?? Date.now const endpoint = options.baseURL.replace(/\/$/, "") + "/image-files" const send = options.fetch ?? fetch + let cacheID: string | undefined let scope: string | undefined let routeHeaders: Record = {} const anthropic = /\/anthropic\/v1\/?$/.test(new URL(options.baseURL).pathname) @@ -37,8 +53,104 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { new URL(options.baseURL).pathname, ) - async function prepare(params: LanguageModelV3CallOptions): Promise { + async function locked(call: () => Promise): Promise { + const prior = state.lock + const next = Promise.withResolvers() + state.lock = next.promise + await prior + state.busy = true + try { + return await call() + } finally { + state.busy = false + next.resolve() + } + } + + // Retry only explicit transient HTTP failures. A network failure after an upload + // has an unknown outcome, so blindly retrying it could create orphaned files. + async function request(url: string, init: RequestInit) { + for (let attempt = 0; ; attempt++) { + init.signal?.throwIfAborted() + const response = await send(url, init) + if (![429, 502, 503, 504].includes(response.status) || attempt === 2) return response + await response.body?.cancel() + await new Promise((resolve, reject) => { + const abort = () => { + clearTimeout(timer) + reject(init.signal?.reason) + } + const timer = setTimeout( + () => { + init.signal?.removeEventListener("abort", abort) + resolve() + }, + 100 * 2 ** attempt, + ) + init.signal?.addEventListener("abort", abort, { once: true }) + if (init.signal?.aborted) abort() + }) + } + } + + function identity(data: string | Uint8Array | URL, mediaType: string) { + const raw = data instanceof URL ? data.toString().split(",", 2)[1] : data + const bytes = typeof raw === "string" ? Buffer.from(raw, "base64") : raw + if (bytes.byteLength > 5 * 1024 * 1024) throw new Error("Screenshot exceeds upload limit") + return { bytes, key: `${mediaType}:${new Bun.CryptoHasher("sha256").update(bytes).digest("hex")}` } + } + + async function cleanup(signal?: AbortSignal) { + const pinned = new Set([...state.leases].flatMap((lease) => [...lease.files])) + if (state.closed) { + for (const [key, pending] of state.cache) { + if ([...state.leases].some((lease) => lease.keys.has(key))) continue + if (state.retired.length >= 300) break + const file = await pending.catch(() => undefined) + state.cache.delete(key) + state.ready.delete(key) + state.refreshed.delete(key) + if (file) state.retired.push(file) + } + } + // Cleanup is best effort with one total deadline, including retries. A + // deletion outage must not hold the request lock for one timeout per file. + const deadline = AbortSignal.any([AbortSignal.timeout(1000), ...(signal ? [signal] : [])]) + const removed = new Set() + let attempted = 0 + for (const file of state.retired) { + if (pinned.has(file.file_id) || (state.chain && !state.closed)) continue + if (file.expires_at <= now() / 1000) { + removed.add(file) + continue + } + if (deadline.aborted || attempted++ >= (state.closed ? 300 : 8)) break + const response = await request( + `${endpoint}/${encodeURIComponent(file.file_id)}?model=${encodeURIComponent(model.modelId)}`, + { + method: "DELETE", + headers: { ...routeHeaders, "x-bu-image-scope": scope! }, + signal: deadline, + }, + ).catch(() => undefined) + if (!response?.ok && response?.status !== 404) break + removed.add(file) + } + state.retired = state.retired.filter((file) => !removed.has(file)) + if (state.closed && !state.cache.size && !state.retired.length && cacheID && runCaches.get(cacheID) === state) + runCaches.delete(cacheID) + } + + async function prepare(params: LanguageModelV3CallOptions, lease: Lease): Promise { if (!enabled) return params + const stateful = + !anthropic && + (typeof params.providerOptions?.openai?.previousResponseId === "string" || + params.providerOptions?.openai?.store === true) + if (stateful) { + state.chain = true + for (const key of state.cache.keys()) state.chainKeys.add(key) + } const hasImages = params.prompt.some( (message) => (message.role === "user" && @@ -74,152 +186,253 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { ...(legacy ? { "x-bu-openai-reasoning": "1" } : {}), } const signal = AbortSignal.any([...(params.abortSignal ? [params.abortSignal] : []), AbortSignal.timeout(30_000)]) - const capability = await send(`${endpoint}/capability?model=${encodeURIComponent(model.modelId)}`, { + const capability = await request(`${endpoint}/capability?model=${encodeURIComponent(model.modelId)}`, { headers: currentHeaders, signal, }) // Older gateways have no upload contract. Never fall back after selecting reference mode. if (capability.status === 404 && !scope) return params if (!capability.ok) throw new Error(`Screenshot capability failed (${capability.status})`) - const route = (await capability.json()) as { supported: boolean; scope?: string } + const route = (await capability.json()) as { + supported: boolean + scope?: string + max_references?: number + proof_transport?: string + } if (!route.supported && !scope) return params if (!route.supported || !route.scope || (scope && scope !== route.scope)) throw new Error("Screenshot route or account changed; start a new run") if (closed || state.closed) throw new Error("Screenshot file cache is closed") scope = route.scope - if (!runCaches.has(scope) && runCaches.size >= 128) throw new Error("Too many active screenshot caches") - state = runCaches.get(scope) ?? state + cacheID = `${endpoint}:${model.modelId}:${scope}` + for (const [id, cached] of runCaches) { + if (cached.closed && !cached.leases.size && cached.expiresAt <= now() / 1000) runCaches.delete(id) + } + if (!runCaches.has(cacheID) && runCaches.size >= 128) throw new Error("Too many active screenshot caches") + state = runCaches.get(cacheID) ?? state + if (state.closed) throw new Error("Screenshot file cache is closing; retry cleanup first") + if (stateful) { + state.chain = true + for (const key of state.cache.keys()) state.chainKeys.add(key) + } state.owners.add(owner) - runCaches.set(scope, state) + runCaches.set(cacheID, state) routeHeaders = currentHeaders - const headers = { ...currentHeaders, "x-bu-image-scope": scope } - const proofs: Record = {} - async function upload(data: string | Uint8Array | URL, mediaType: string) { - if (state.active >= 2) await new Promise((resolve) => state.waiting.push(resolve)) - else state.active++ - try { - signal.throwIfAborted() - if (closed || state.closed) throw new Error("Screenshot file cache is closed") - const raw = data instanceof URL ? data.toString().split(",", 2)[1] : data - const bytes = typeof raw === "string" ? Buffer.from(raw, "base64") : raw - if (bytes.byteLength > 5 * 1024 * 1024) throw new Error("Screenshot exceeds upload limit") - const hash = new Bun.CryptoHasher("sha256").update(bytes).digest("hex") - const key = `${scope}:${mediaType}:${hash}` - let existing = state.cache.get(key) - if (existing) { - const file = await existing - if (state.cache.get(key) !== existing) existing = state.cache.get(key) - else if (file.expires_at <= now() / 1000 + 30) { - if (state.refreshed.has(key)) throw new Error("Screenshot reference expired again; start a new run") - state.refreshed.add(key) - state.retired.push(file) - existing = undefined + const limit = route.proof_transport === "body" && route.max_references === 300 ? 300 : 64 + const keys = new Set() + for (const message of params.prompt) { + if (message.role === "user") + for (const part of message.content) { + if ( + part.type === "file" && + part.mediaType.startsWith("image/") && + (!(part.data instanceof URL) || part.data.protocol === "data:") + ) + keys.add(identity(part.data, part.mediaType).key) + } + if (message.role === "tool") + for (const part of message.content) { + if (part.type === "tool-result" && part.output.type === "content") + for (const item of part.output.value) { + if (item.type === "image-data") keys.add(identity(item.data, item.mediaType).key) + } + } + } + const cached = [...keys].every((key) => (state.ready.get(key)?.expires_at ?? 0) > now() / 1000 + 30) + const work = async () => { + signal.throwIfAborted() + if (closed || state.closed) throw new Error("Screenshot file cache is closed") + if (keys.size > limit) + throw new Error( + `Screenshot request needs ${keys.size} distinct images; gateway limit is ${limit}. Compact history before retrying`, + ) + const pinned = new Set([...state.chainKeys, ...[...state.leases].flatMap((item) => [...item.keys])]) + if (new Set([...pinned, ...keys]).size > 300) + throw new Error( + "Screenshot cache has 300 images in active requests or a stateful response chain; finish active requests or start a new chain", + ) + lease.keys = keys + state.leases.add(lease) + if (!cached) await cleanup(signal) + const missing = [...keys].filter((key) => !state.cache.has(key)).length + for (const key of state.cache.keys()) { + if (state.cache.size + missing <= 300) break + if (keys.has(key) || state.chainKeys.has(key) || [...state.leases].some((item) => item.keys.has(key))) continue + if (state.retired.length >= 300) + throw new Error("Screenshot cleanup backlog is full; retry cleanup before uploading") + state.retired.push(state.ready.get(key)!) + state.cache.delete(key) + state.ready.delete(key) + state.refreshed.delete(key) + } + if (state.cache.size + missing > 300) + throw new Error("Screenshot cache has 300 pinned images; retry after active requests finish") + // Provider-side previousResponseId history may still depend on omitted + // files. Keep explicit stateful chains until close; never assume a copy. + if (state.chain) for (const key of keys) state.chainKeys.add(key) + lease.keys = keys + state.leases.add(lease) + if (!cached) await cleanup(signal) + const headers = { ...currentHeaders, "x-bu-image-scope": scope! } + const proofs: Record = {} + async function upload(data: string | Uint8Array | URL, mediaType: string) { + const { bytes, key } = identity(data, mediaType) + const ready = state.ready.get(key) + if (ready && ready.expires_at > now() / 1000 + 30) { + lease.files.add(ready.file_id) + proofs[ready.file_id] = { expires_at: ready.expires_at, signature: ready.signature } + return ready.file_id + } + if (state.active >= 2) await new Promise((resolve) => state.waiting.push(resolve)) + else state.active++ + try { + signal.throwIfAborted() + if (closed || state.closed) throw new Error("Screenshot file cache is closed") + let existing = state.cache.get(key) + if (existing) { + const file = await existing + if (state.cache.get(key) !== existing) existing = state.cache.get(key) + else if (file.expires_at <= now() / 1000 + 30) { + if (state.refreshed.has(key)) throw new Error("Screenshot reference expired again; start a new run") + if (state.retired.length >= 300) throw new Error("Screenshot cleanup backlog is full") + state.refreshed.add(key) + state.retired.push(file) + state.ready.delete(key) + existing = undefined + } } + const pending = + existing ?? + (async () => { + const response = await request(`${endpoint}?model=${encodeURIComponent(model.modelId)}`, { + method: "POST", + headers: { ...headers, "content-type": mediaType }, + body: new Uint8Array(bytes), + signal, + }) + if (!response.ok) throw new Error(`Screenshot upload failed (${response.status})`) + const uploaded = (await response.json()) as Uploaded + if (!uploaded.file_id?.startsWith(anthropic ? "file_" : "file-")) + throw new Error("Invalid screenshot file ID") + if (!(uploaded.expires_at > now() / 1000 + 30)) throw new Error("Screenshot reference expired") + state.expiresAt = Math.max(state.expiresAt, uploaded.expires_at) + return uploaded + })() + state.cache.set(key, pending) + const uploaded = await pending.catch((error) => { + if (state.cache.get(key) === pending) { + state.cache.delete(key) + state.ready.delete(key) + if (!state.refreshed.has(key)) state.chainKeys.delete(key) + state.refreshed.delete(key) + } + throw error + }) + state.ready.set(key, uploaded) + if (uploaded.expires_at <= now() / 1000 + 30) throw new Error("Screenshot reference expired; start a new run") + signal.throwIfAborted() + if (closed || state.closed) throw new Error("Screenshot file cache is closed") + lease.files.add(uploaded.file_id) + proofs[uploaded.file_id] = { expires_at: uploaded.expires_at, signature: uploaded.signature } + return uploaded.file_id + } finally { + const next = state.waiting.shift() + if (next) next() + else state.active-- } - if (!existing && !state.cache.has(key) && state.cache.size >= 64) - throw new Error("Screenshot file limit reached; start a new run") - const pending = - existing ?? - (async () => { - const response = await send(`${endpoint}?model=${encodeURIComponent(model.modelId)}`, { - method: "POST", - headers: { ...headers, "content-type": mediaType }, - body: new Uint8Array(bytes), - signal, - }) - if (!response.ok) throw new Error(`Screenshot upload failed (${response.status})`) - const uploaded = (await response.json()) as Uploaded - if (!uploaded.file_id?.startsWith(anthropic ? "file_" : "file-")) - throw new Error("Invalid screenshot file ID") - if (!(uploaded.expires_at > now() / 1000 + 30)) throw new Error("Screenshot reference expired") - return uploaded - })() - state.cache.set(key, pending) - const uploaded = await pending.catch((error) => { - if (state.cache.get(key) === pending) state.cache.delete(key) - throw error - }) - if (uploaded.expires_at <= now() / 1000 + 30) throw new Error("Screenshot reference expired; start a new run") - signal.throwIfAborted() - if (closed || state.closed) throw new Error("Screenshot file cache is closed") - proofs[uploaded.file_id] = { expires_at: uploaded.expires_at, signature: uploaded.signature } - return uploaded.file_id - } finally { - const next = state.waiting.shift() - if (next) next() - else state.active-- } - } - const prompt = await Promise.all( - params.prompt.map(async (message) => { - if (message.role === "tool") + const prompt = await settled( + params.prompt.map(async (message) => { + if (message.role === "tool") + return { + ...message, + content: await settled( + message.content.map(async (part) => { + if (part.type !== "tool-result" || part.output.type !== "content") return part + return { + ...part, + output: { + ...part.output, + value: await settled( + part.output.value.map(async (item) => { + if (item.type !== "image-data") return item + return { + type: "image-url" as const, + url: `bu-${anthropic ? "anthropic" : "openai"}-file:${await upload(item.data, item.mediaType)}`, + providerOptions: item.providerOptions, + } + }), + ), + }, + } + }), + ), + } + if (message.role !== "user") return message return { ...message, - content: await Promise.all( + content: await settled( message.content.map(async (part) => { - if (part.type !== "tool-result" || part.output.type !== "content") return part - return { - ...part, - output: { - ...part.output, - value: await Promise.all( - part.output.value.map(async (item) => { - if (item.type !== "image-data") return item - return { - type: "image-url" as const, - url: `bu-${anthropic ? "anthropic" : "openai"}-file:${await upload(item.data, item.mediaType)}`, - providerOptions: item.providerOptions, - } - }), - ), - }, - } + if ( + part.type !== "file" || + !part.mediaType.startsWith("image/") || + (part.data instanceof URL && part.data.protocol !== "data:") + ) + return part + const id = await upload(part.data, part.mediaType) + return { ...part, data: anthropic ? new URL(`bu-anthropic-file:${id}`) : id } }), ), } - if (message.role !== "user") return message - return { - ...message, - content: await Promise.all( - message.content.map(async (part) => { - if ( - part.type !== "file" || - !part.mediaType.startsWith("image/") || - (part.data instanceof URL && part.data.protocol !== "data:") - ) - return part - const id = await upload(part.data, part.mediaType) - return { ...part, data: anthropic ? new URL(`bu-anthropic-file:${id}`) : id } - }), - ), - } - }), - ) - return { - ...params, - prompt, - headers: { ...params.headers, "x-bu-image-scope": scope, "x-bu-image-proofs": JSON.stringify(proofs) }, + }), + ) + const encoded = JSON.stringify(proofs) + if (route.proof_transport !== "body" && Buffer.byteLength(encoded) > 12 * 1024) + throw new Error("Screenshot proofs exceed legacy header limit; upgrade the gateway") + return { + ...params, + prompt, + headers: { + ...params.headers, + "x-bu-image-scope": scope!, + "x-bu-image-proofs": encoded, + ...(route.proof_transport === "body" ? { "x-bu-image-proof-transport": "body" } : {}), + }, + } } + return cached ? work() : locked(work) + } + + async function release(lease: Lease) { + state.leases.delete(lease) + // A cache-hit call must not queue its completion behind unrelated uploads. + // The current lock holder performs deferred cleanup when it releases its lease. + if (!state.busy) await locked(() => cleanup()) } async function invoke( params: LanguageModelV3CallOptions, + lease: Lease, call: (prepared: LanguageModelV3CallOptions) => PromiseLike, ): Promise { try { - return await call(await prepare(params)) + return await call(await prepare(params, lease)) } catch (error) { const message = error instanceof Error ? error.message : "" if (!/file.*(not found|does not exist|expired|deleted)/i.test(message)) throw error - const entries = await Promise.all([...state.cache].map(async ([key, pending]) => ({ key, file: await pending }))) - const missing = entries.filter(({ file }) => message.includes(file.file_id)) - if (!missing.length || missing.some(({ key }) => state.refreshed.has(key))) throw error - for (const { key, file } of missing) { - state.refreshed.add(key) - state.retired.push(file) - state.cache.delete(key) - } - return await call(await prepare(params)) + await locked(async () => { + const entries = await settled([...state.cache].map(async ([key, pending]) => ({ key, file: await pending }))) + const missing = entries.filter(({ file }) => lease.files.has(file.file_id) && message.includes(file.file_id)) + if (!missing.length || missing.some(({ key }) => state.refreshed.has(key))) throw error + if (state.retired.length + missing.length > 300) throw error + for (const { key, file } of missing) { + state.refreshed.add(key) + state.retired.push(file) + state.cache.delete(key) + state.ready.delete(key) + } + }) + return await call(await prepare(params, lease)) } } @@ -229,42 +442,97 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { modelId: model.modelId, supportedUrls: model.supportedUrls, async doGenerate(params: LanguageModelV3CallOptions) { - return invoke(params, (prepared) => model.doGenerate(prepared)) + const lease: Lease = { keys: new Set(), files: new Set() } + try { + return await invoke(params, lease, (prepared) => model.doGenerate(prepared)) + } finally { + await release(lease) + } }, async doStream(params: LanguageModelV3CallOptions) { - return invoke(params, (prepared) => model.doStream(prepared)) + const lease: Lease = { keys: new Set(), files: new Set() } + try { + const result = await invoke(params, lease, (prepared) => model.doStream(prepared)) + const reader = result.stream.getReader() + let finished = false + const finish = async () => { + if (finished) return + finished = true + params.abortSignal?.removeEventListener("abort", abort) + await release(lease) + } + const abort = () => { + void reader + .cancel(params.abortSignal?.reason) + .finally(finish) + .catch(() => {}) + } + params.abortSignal?.addEventListener("abort", abort, { once: true }) + if (params.abortSignal?.aborted) abort() + return { + ...result, + stream: new ReadableStream({ + async pull(controller) { + try { + const item = await reader.read() + if (item.done) { + await finish() + controller.close() + return + } + controller.enqueue(item.value) + } catch (error) { + await finish() + controller.error(error) + } + }, + async cancel(reason) { + try { + await reader.cancel(reason) + } finally { + await finish() + } + }, + }), + } + } catch (error) { + await release(lease) + throw error + } }, async closeImageFiles() { - if (closed) return closed = true - state.owners.delete(owner) - if (state.owners.size) return - state.closed = true - if (scope && runCaches.get(scope) === state) runCaches.delete(scope) - const entries = await Promise.allSettled([ - ...state.cache.values(), - ...state.retired.map((file) => Promise.resolve(file)), - ]) - await Promise.allSettled( - entries - .filter((entry) => entry.status === "fulfilled") - .map((entry) => - send(`${endpoint}/${encodeURIComponent(entry.value.file_id)}?model=${encodeURIComponent(model.modelId)}`, { - method: "DELETE", - headers: { ...routeHeaders, "x-bu-image-scope": scope! }, - signal: AbortSignal.timeout(10_000), - }), - ), - ) - state.cache.clear() + await locked(async () => { + state.owners.delete(owner) + if (state.owners.size) return + state.closed = true + await cleanup() + }) }, } } +async function settled(items: Promise[]): Promise { + const results = await Promise.allSettled(items) + const failed = results.find((item) => item.status === "rejected") + if (failed?.status === "rejected") throw failed.reason + return results.map((item) => (item as PromiseFulfilledResult).value) +} + // The pinned SDK lacks tool-result file IDs; only small placeholders enter its JSON serializer. export function imageFileRequest(init?: RequestInit): RequestInit | undefined { if (!init || !new Headers(init.headers).has("x-bu-image-scope") || typeof init.body !== "string") return init const body = JSON.parse(init.body) + const headers = new Headers(init.headers) + if (headers.get("x-bu-image-proof-transport") === "body") { + const proofs = headers.get("x-bu-image-proofs") ?? "{}" + if (Buffer.byteLength(proofs) > 128 * 1024 || Object.keys(JSON.parse(proofs)).length > 300) + throw new Error("Screenshot proof body limit exceeded") + body._image_file_proofs = JSON.parse(proofs) + body._image_file_scope = headers.get("x-bu-image-scope") + headers.delete("x-bu-image-proofs") + } + headers.delete("x-bu-image-proof-transport") if (Array.isArray(body.input)) body.input = body.input.map((item: Record) => { if (!item || typeof item !== "object") return item @@ -298,12 +566,11 @@ export function imageFileRequest(init?: RequestInit): RequestInit | undefined { body.messages = body.messages.map((message: Record) => Array.isArray(message.content) ? { ...message, content: rewrite(message.content) } : message, ) - const headers = new Headers(init.headers) headers.set( "anthropic-beta", [...new Set([...(headers.get("anthropic-beta")?.split(",") ?? []), "files-api-2025-04-14"])].join(","), ) return { ...init, headers, body: JSON.stringify(body) } } - return { ...init, body: JSON.stringify(body) } + return { ...init, headers, body: JSON.stringify(body) } } diff --git a/packages/opencode/test/provider/image-files-lifecycle.test.ts b/packages/opencode/test/provider/image-files-lifecycle.test.ts new file mode 100644 index 000000000..56fea2563 --- /dev/null +++ b/packages/opencode/test/provider/image-files-lifecycle.test.ts @@ -0,0 +1,424 @@ +import { expect, test } from "bun:test" +import type { LanguageModelV3, LanguageModelV3CallOptions, LanguageModelV3StreamPart } from "@ai-sdk/provider" +import { imageFileRequest, withImageFiles } from "../../src/provider/image-files" + +function images(start: number, count: number): LanguageModelV3CallOptions { + return { + prompt: [ + { + role: "user", + content: Array.from({ length: count }, (_, i) => ({ + type: "file", + mediaType: "image/png", + data: Buffer.from(`screenshot-${start + i}`), + })), + }, + ], + } +} +function fixture(options: { legacy?: boolean; anthropic?: boolean } = {}) { + const state = { + now: Date.now(), + scope: crypto.randomUUID(), + deleteStatus: 0, + hangDeletes: false, + uploadGate: undefined as Promise | undefined, + uploads: 0, + capabilities: 0, + uploadFailures: 0, + capabilityFailures: 0, + deleteFailures: 0, + calls: [] as LanguageModelV3CallOptions[], + deleted: [] as string[], + live: new Set(), + gate: undefined as Promise | undefined, + stream: undefined as ReadableStreamDefaultController | undefined, + } + const server = Bun.serve({ + port: 0, + async fetch(req) { + if (req.url.includes("/capability")) { + state.capabilities++ + if (state.capabilityFailures-- > 0) return new Response("unavailable", { status: 503 }) + return Response.json({ + supported: true, + scope: state.scope, + ...(options.legacy ? {} : { max_references: 300, proof_transport: "body" }), + }) + } + if (req.method === "DELETE") { + if (state.hangDeletes) return new Promise(() => {}) + if (state.deleteStatus) return new Response("failed", { status: state.deleteStatus }) + if (state.deleteFailures-- > 0) return new Response("unavailable", { status: 503 }) + const id = new URL(req.url).pathname.split("/").at(-1)! + state.deleted.push(id) + state.live.delete(id) + return Response.json({ deleted: true }) + } + state.uploads++ + const upload = state.uploads + await state.uploadGate + if (state.uploadFailures-- > 0) return new Response("unavailable", { status: 503 }) + const id = `${options.anthropic ? "file_" : "file-"}${upload}` + state.live.add(id) + return Response.json({ file_id: id, expires_at: Math.floor(state.now / 1000) + 3600, signature: "a".repeat(64) }) + }, + }) + const native: LanguageModelV3 = { + specificationVersion: "v3", + provider: options.anthropic ? "anthropic" : "openai", + modelId: "test", + supportedUrls: {}, + async doGenerate(params) { + state.calls.push(params) + await state.gate + return { + content: [], + finishReason: { unified: "stop", raw: "stop" }, + usage: { + inputTokens: { total: 0, noCache: 0, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 0, text: 0, reasoning: 0 }, + }, + warnings: [], + } + }, + async doStream(params) { + state.calls.push(params) + return { + stream: new ReadableStream({ + start(controller) { + state.stream = controller + }, + }), + warnings: [], + } + }, + } + const baseURL = `${server.url}api/v4/llm/${native.provider}/v1` + const wrap = (modelId = native.modelId) => + withImageFiles({ ...native, modelId }, { baseURL, apiKey: "test", now: () => state.now }) + const model = wrap() + return { + state, + model, + wrap, + async close() { + await model.closeImageFiles() + server.stop(true) + }, + } +} + +for (const anthropic of [false, true]) { + test(`200 distinct screenshots and retained history upload once (anthropic=${anthropic})`, async () => { + const f = fixture({ anthropic }) + try { + await f.model.doGenerate(images(0, 200)) + await f.model.doGenerate(images(0, 201)) + expect(f.state.uploads).toBe(201) + expect(f.state.calls[1].prompt[0].content).toHaveLength(201) + expect(JSON.stringify(f.state.calls)).not.toContain(Buffer.from("screenshot-0").toString("base64")) + } finally { + await f.close() + } + }) +} +test("more than 300 lifetime images with bounded history evict only unused files", async () => { + const f = fixture() + try { + await f.model.doGenerate(images(0, 300)) + const message = f.state.calls[0].prompt[0] + if (message.role !== "user") throw new Error("Expected user images") + const ids = message.content.map((part) => (part.type === "file" ? String(part.data) : "")) + await f.model.doGenerate(images(1, 300)) + expect(f.state.uploads).toBe(301) + expect(f.state.deleted).toEqual([ids[0]]) + expect(f.state.live.size).toBe(300) + await f.model.doGenerate(images(2, 300)) + expect(f.state.uploads).toBe(302) + expect(f.state.deleted).toEqual([ids[0], ids[1]]) + } finally { + await f.close() + } +}) +test("301 simultaneous images fail before any upload without discarding content", async () => { + const f = fixture({ anthropic: true }) + try { + await expect(f.model.doGenerate(images(0, 301))).rejects.toThrow("300") + expect(f.state.uploads).toBe(0) + expect(f.state.calls).toHaveLength(0) + } finally { + await f.close() + } +}) +test("in-flight generation pins its files against eviction and close", async () => { + const f = fixture() + try { + await f.model.doGenerate(images(0, 300)) + const gate = Promise.withResolvers() + f.state.gate = gate.promise + const pending = f.model.doGenerate(images(0, 1)) + while (f.state.calls.length < 2) await Bun.sleep(1) + const part = f.state.calls[1].prompt[0].content[0] + const pinned = typeof part !== "string" && part.type === "file" ? String(part.data) : "missing" + f.state.gate = undefined + await f.model.doGenerate(images(300, 1)) + expect(f.state.deleted).not.toContain(pinned) + await f.model.closeImageFiles() + expect(f.state.deleted).not.toContain(pinned) + gate.resolve() + await pending + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } +}) +test("stream retains references until cancellation then permits cleanup", async () => { + const f = fixture() + try { + const result = await f.model.doStream(images(0, 1)) + await f.model.closeImageFiles() + expect(f.state.deleted).toEqual([]) + await result.stream.cancel() + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } +}) +test("transient capability and upload failures retry within one call", async () => { + const f = fixture() + try { + f.state.capabilityFailures = 1 + f.state.uploadFailures = 1 + await f.model.doGenerate(images(0, 1)) + expect(f.state.capabilities).toBe(2) + expect(f.state.uploads).toBe(2) + expect(f.state.calls).toHaveLength(1) + } finally { + await f.close() + } +}) +test("failed uploads and cleanup can recover on explicit retry", async () => { + const f = fixture() + try { + f.state.uploadFailures = 10 + await expect(f.model.doGenerate(images(0, 1))).rejects.toThrow("upload failed") + expect(f.state.uploads).toBe(3) + f.state.uploadFailures = 0 + await f.model.doGenerate(images(0, 1)) + f.state.deleteFailures = 10 + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(1) + f.state.deleteFailures = 0 + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } +}) +test("negotiated 300 proofs move into bounded body metadata before HTTP", async () => { + const f = fixture() + try { + await f.model.doGenerate(images(0, 300)) + const result = imageFileRequest({ + headers: f.state.calls[0].headers as Record, + body: JSON.stringify({ input: [] }), + })! + const headers = new Headers(result.headers) + const body = JSON.parse(result.body as string) + expect(headers.has("x-bu-image-proofs")).toBe(false) + expect(Object.keys(body._image_file_proofs)).toHaveLength(300) + expect(JSON.stringify([...headers]).length).toBeLessThan(1024) + } finally { + await f.close() + } +}) + +test("legacy gateway keeps 64-reference contract and header proofs", async () => { + const f = fixture({ legacy: true }) + try { + await f.model.doGenerate(images(0, 64)) + const result = imageFileRequest({ + headers: f.state.calls[0].headers as Record, + body: JSON.stringify({ input: [] }), + })! + expect(new Headers(result.headers).has("x-bu-image-proofs")).toBe(true) + expect(JSON.parse(result.body as string)._image_file_proofs).toBeUndefined() + await expect(f.model.doGenerate(images(0, 65))).rejects.toThrow("64") + expect(f.state.uploads).toBe(64) + } finally { + await f.close() + } +}) +for (const ending of ["complete", "error", "abort"] as const) { + test(`stream ${ending} releases pins after close`, async () => { + const f = fixture() + try { + const abort = new AbortController() + const result = await f.model.doStream({ ...images(0, 1), abortSignal: abort.signal }) + const reader = result.stream.getReader() + await f.model.closeImageFiles() + expect(f.state.deleted).toEqual([]) + if (ending === "complete") f.state.stream!.close() + if (ending === "error") f.state.stream!.error(new Error("stream failed")) + if (ending === "abort") abort.abort() + if (ending === "error") await expect(reader.read()).rejects.toThrow("stream failed") + else expect((await reader.read()).done).toBe(true) + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } + }) +} +test("overlapping active requests cannot evict each other's full working set", async () => { + const f = fixture() + const gate = Promise.withResolvers() + try { + f.state.gate = gate.promise + const pending = f.model.doGenerate(images(0, 300)) + while (!f.state.calls.length) await Bun.sleep(1) + await expect(f.model.doGenerate(images(300, 1))).rejects.toThrow("active requests") + expect(f.state.deleted).toEqual([]) + f.state.gate = undefined + gate.resolve() + await pending + await f.model.doGenerate(images(300, 1)) + expect(f.state.uploads).toBe(301) + } finally { + gate.resolve() + await f.close() + } +}) +test("abort during transient retry stops uploading and leaves later call usable", async () => { + const f = fixture() + try { + f.state.uploadFailures = 10 + const abort = new AbortController() + const pending = f.model.doGenerate({ ...images(0, 1), abortSignal: abort.signal }) + while (!f.state.uploads) await Bun.sleep(1) + abort.abort() + await expect(pending).rejects.toThrow() + expect(f.state.uploads).toBe(1) + expect(f.state.calls).toHaveLength(0) + f.state.uploadFailures = 0 + await f.model.doGenerate(images(0, 1)) + expect(f.state.uploads).toBe(2) + } finally { + await f.close() + } +}) + +test("hung deletes have a total cleanup deadline and remain retryable", async () => { + const f = fixture() + try { + await f.model.doGenerate(images(0, 20)) + f.state.hangDeletes = true + const started = performance.now() + await f.model.closeImageFiles() + expect(performance.now() - started).toBeLessThan(2500) + expect(f.state.live.size).toBe(20) + f.state.hangDeletes = false + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } +}) +test("expired closed caches cannot exhaust the global run slots after failed deletes", async () => { + const f = fixture() + try { + f.state.deleteStatus = 400 + for (let i = 0; i < 130; i++) { + f.state.scope = crypto.randomUUID() + const model = f.wrap() + await model.doGenerate(images(i, 1)) + await model.closeImageFiles() + f.state.now += 3601_000 + } + expect(f.state.calls).toHaveLength(130) + } finally { + await f.close() + } +}) +test("delete outage bounds retired files and permits recovery without losing pins", async () => { + const f = fixture() + try { + f.state.deleteStatus = 400 + await f.model.doGenerate(images(0, 300)) + await f.model.doGenerate(images(300, 300)) + expect(f.state.live.size).toBe(600) + await expect(f.model.doGenerate(images(600, 1))).rejects.toThrow("cleanup backlog") + expect(f.state.uploads).toBe(600) + f.state.deleteStatus = 0 + await f.model.doGenerate(images(600, 1)) + expect(f.state.uploads).toBe(601) + await f.model.closeImageFiles() + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } +}) + +test("different models never share uploaded files even with an identical gateway scope", async () => { + const f = fixture() + const sibling = f.wrap("another-model") + try { + await f.model.doGenerate(images(0, 1)) + await sibling.doGenerate(images(0, 1)) + expect(f.state.uploads).toBe(2) + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(1) + await sibling.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await sibling.closeImageFiles() + await f.close() + } +}) + +for (const store of [false, true]) { + test(`stateful Responses keep omitted files until close (explicit store=${store})`, async () => { + const f = fixture() + try { + await f.model.doGenerate({ + ...images(0, 300), + ...(store ? { providerOptions: { openai: { store: true } } } : {}), + }) + await f.model.doGenerate({ prompt: [], providerOptions: { openai: { previousResponseId: "resp-prior" } } }) + await expect(f.model.doGenerate(images(300, 1))).rejects.toThrow("stateful response chain") + expect(f.state.uploads).toBe(300) + expect(f.state.deleted).toEqual([]) + await f.model.closeImageFiles() + expect(f.state.live.size).toBe(0) + } finally { + await f.close() + } + }) +} + +test("cached image requests do not wait behind unrelated slow uploads", async () => { + const f = fixture() + const gate = Promise.withResolvers() + try { + await f.model.doGenerate(images(0, 1)) + f.state.uploadGate = gate.promise + const uploading = f.model.doGenerate(images(1, 1)) + while (f.state.uploads < 2) await Bun.sleep(1) + await Promise.race([ + f.model.doGenerate(images(0, 1)), + Bun.sleep(500).then(() => { + throw new Error("cached request blocked by upload") + }), + ]) + gate.resolve() + await uploading + expect(f.state.uploads).toBe(2) + } finally { + gate.resolve() + await f.close() + } +}) diff --git a/packages/opencode/test/provider/image-files.test.ts b/packages/opencode/test/provider/image-files.test.ts index ff10e70ec..d6971c3d6 100644 --- a/packages/opencode/test/provider/image-files.test.ts +++ b/packages/opencode/test/provider/image-files.test.ts @@ -364,7 +364,7 @@ test("repeated invalid references and failed upload stay bounded", async () => { try { await expect(failed.model.doGenerate({ prompt })).rejects.toThrow("upload failed") expect(failed.state.responses).toBe(0) - expect(failed.state.uploads).toBe(1) + expect(failed.state.uploads).toBe(3) } finally { failed.stop() } @@ -492,7 +492,7 @@ test("transient upload rejection permits a later explicit retry", async () => { expect(f.state.responses).toBe(0) options.uploadStatus = undefined await f.model.doGenerate({ prompt }) - expect(f.state.uploads).toBe(2) + expect(f.state.uploads).toBe(4) expect(f.state.responses).toBe(1) } finally { await f.model.closeImageFiles() From 9b2e71a23484c492e7e86db3304989af12fb1667 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Mon, 5 Oct 2026 07:18:04 +0000 Subject: [PATCH 2/3] fix(provider): avoid duplicate screenshot lease registration --- packages/opencode/src/provider/image-files.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/opencode/src/provider/image-files.ts b/packages/opencode/src/provider/image-files.ts index 9062a67ce..a930436e3 100644 --- a/packages/opencode/src/provider/image-files.ts +++ b/packages/opencode/src/provider/image-files.ts @@ -270,8 +270,7 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { // Provider-side previousResponseId history may still depend on omitted // files. Keep explicit stateful chains until close; never assume a copy. if (state.chain) for (const key of keys) state.chainKeys.add(key) - lease.keys = keys - state.leases.add(lease) + // Reclaim files newly retired by eviction, after the pre-eviction backlog pass. if (!cached) await cleanup(signal) const headers = { ...currentHeaders, "x-bu-image-scope": scope! } const proofs: Record = {} From 2dfd70b4f179b464dc0d1c27b6bc49895ed0a446 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Mon, 5 Oct 2026 07:39:54 +0000 Subject: [PATCH 3/3] fix(provider): recover deferred and closed-cache cleanup --- packages/opencode/src/provider/image-files.ts | 45 +++++++++++++++---- .../provider/image-files-lifecycle.test.ts | 42 +++++++++++++++++ 2 files changed, 79 insertions(+), 8 deletions(-) diff --git a/packages/opencode/src/provider/image-files.ts b/packages/opencode/src/provider/image-files.ts index a930436e3..66fb0ac6b 100644 --- a/packages/opencode/src/provider/image-files.ts +++ b/packages/opencode/src/provider/image-files.ts @@ -13,6 +13,7 @@ type RunCache = { cache: Map> ready: Map busy: boolean + cleanupQueued: boolean refreshed: Set retired: Uploaded[] waiting: Array<() => void> @@ -32,6 +33,7 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { cache: new Map(), ready: new Map(), busy: false, + cleanupQueued: false, refreshed: new Set(), retired: [], waiting: [], @@ -100,7 +102,7 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { return { bytes, key: `${mediaType}:${new Bun.CryptoHasher("sha256").update(bytes).digest("hex")}` } } - async function cleanup(signal?: AbortSignal) { + async function cleanup(signal?: AbortSignal, preserveRegistry = false) { const pinned = new Set([...state.leases].flatMap((lease) => [...lease.files])) if (state.closed) { for (const [key, pending] of state.cache) { @@ -137,7 +139,14 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { removed.add(file) } state.retired = state.retired.filter((file) => !removed.has(file)) - if (state.closed && !state.cache.size && !state.retired.length && cacheID && runCaches.get(cacheID) === state) + if ( + !preserveRegistry && + state.closed && + !state.cache.size && + !state.retired.length && + cacheID && + runCaches.get(cacheID) === state + ) runCaches.delete(cacheID) } @@ -169,7 +178,7 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { )), ) if (!hasImages) return params - if (closed || state.closed) throw new Error("Screenshot file cache is closed") + if (closed) throw new Error("Screenshot file cache is closed") const previous = params.providerOptions?.openai?.previousResponseId const legacy = params.prompt.some( (message) => @@ -202,7 +211,7 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { if (!route.supported && !scope) return params if (!route.supported || !route.scope || (scope && scope !== route.scope)) throw new Error("Screenshot route or account changed; start a new run") - if (closed || state.closed) throw new Error("Screenshot file cache is closed") + if (closed) throw new Error("Screenshot file cache is closed") scope = route.scope cacheID = `${endpoint}:${model.modelId}:${scope}` for (const [id, cached] of runCaches) { @@ -210,7 +219,18 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { } if (!runCaches.has(cacheID) && runCaches.size >= 128) throw new Error("Too many active screenshot caches") state = runCaches.get(cacheID) ?? state - if (state.closed) throw new Error("Screenshot file cache is closing; retry cleanup first") + routeHeaders = currentHeaders + if (state.closed) + await locked(async () => { + if (!state.closed) return + await cleanup(signal, true) + if (state.cache.size || state.retired.length || state.leases.size) + throw new Error("Screenshot file cache is closing; retry cleanup first") + state.closed = false + state.chain = false + state.chainKeys.clear() + }) + if (closed) throw new Error("Screenshot file cache is closed") if (stateful) { state.chain = true for (const key of state.cache.keys()) state.chainKeys.add(key) @@ -404,9 +424,18 @@ export function withImageFiles(model: LanguageModelV3, options: Options) { async function release(lease: Lease) { state.leases.delete(lease) - // A cache-hit call must not queue its completion behind unrelated uploads. - // The current lock holder performs deferred cleanup when it releases its lease. - if (!state.busy) await locked(() => cleanup()) + if (state.cleanupQueued) return + // Cache hits complete promptly, but the final lease may end during close. + // Coalesce a follow-up pass rather than relying on another future release. + if (state.busy) { + state.cleanupQueued = true + void locked(async () => { + state.cleanupQueued = false + await cleanup() + }).catch(() => {}) + return + } + await locked(() => cleanup()) } async function invoke( diff --git a/packages/opencode/test/provider/image-files-lifecycle.test.ts b/packages/opencode/test/provider/image-files-lifecycle.test.ts index 56fea2563..9a2d6ca37 100644 --- a/packages/opencode/test/provider/image-files-lifecycle.test.ts +++ b/packages/opencode/test/provider/image-files-lifecycle.test.ts @@ -21,6 +21,8 @@ function fixture(options: { legacy?: boolean; anthropic?: boolean } = {}) { now: Date.now(), scope: crypto.randomUUID(), deleteStatus: 0, + deleteGate: undefined as Promise | undefined, + deleteStarted: 0, hangDeletes: false, uploadGate: undefined as Promise | undefined, uploads: 0, @@ -47,6 +49,8 @@ function fixture(options: { legacy?: boolean; anthropic?: boolean } = {}) { }) } if (req.method === "DELETE") { + state.deleteStarted++ + await state.deleteGate if (state.hangDeletes) return new Promise(() => {}) if (state.deleteStatus) return new Response("failed", { status: state.deleteStatus }) if (state.deleteFailures-- > 0) return new Response("unavailable", { status: 503 }) @@ -422,3 +426,41 @@ test("cached image requests do not wait behind unrelated slow uploads", async () await f.close() } }) + +test("fresh wrapper retries cleanup left by a closed owner", async () => { + const f = fixture() + const sibling = f.wrap() + try { + await f.model.doGenerate(images(0, 1)) + f.state.deleteStatus = 400 + await f.model.closeImageFiles() + await expect(sibling.doGenerate(images(0, 1))).rejects.toThrow("closing") + f.state.deleteStatus = 0 + await sibling.doGenerate(images(0, 1)) + expect(f.state.uploads).toBe(2) + expect(f.state.deleted).toHaveLength(1) + } finally { + await sibling.closeImageFiles() + await f.close() + } +}) + +test("last stream ending during close cleanup schedules a final cleanup pass", async () => { + const f = fixture() + const gate = Promise.withResolvers() + try { + const result = await f.model.doStream(images(0, 1)) + await f.model.doGenerate(images(1, 1)) + f.state.deleteGate = gate.promise + const closing = f.model.closeImageFiles() + while (!f.state.deleteStarted) await Bun.sleep(1) + await result.stream.cancel() + gate.resolve() + await closing + for (let i = 0; i < 100 && f.state.live.size; i++) await Bun.sleep(5) + expect(f.state.live.size).toBe(0) + } finally { + gate.resolve() + await f.close() + } +})