diff --git a/package.json b/package.json index 4d2832d..054e025 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "claude-opencode-proxy", - "version": "0.1.0", + "version": "0.1.1", "description": "Anthropic Messages API proxy that translates Claude Code traffic to OpenCode Zen/Go/Free wire formats", "type": "module", "license": "MIT", diff --git a/spec.md b/spec.md index 2af5334..7aaf91a 100644 --- a/spec.md +++ b/spec.md @@ -187,7 +187,7 @@ function loadConfig(env: NodeJS.ProcessEnv): Config; // throws on invalid | `GET` | `/healthz` | liveness → `200 {"status":"ok"}` | | `GET` | `/ready` | readiness → `200` once config + registry loaded | | `GET` | `/` | info JSON: version, backend, model count | -| `OPTIONS` | `*` | CORS preflight → `204` with `Access-Control-Allow-*` | +| `OPTIONS` | `*` | not handled → `404`; no CORS headers are sent (see §13.3) | ### 5.1 `GET /v1/models` response shape @@ -743,8 +743,10 @@ class ProxyError extends Error { - Build `hono` app; register routes per §5. - Global error middleware: catch `ProxyError` → Anthropic envelope; catch unknown → `500` envelope; log. -- CORS middleware: `OPTIONS *` → `204` with `Access-Control-Allow-Origin: *`, - `-Methods: POST, GET, OPTIONS`, `-Headers: *`. +- No CORS middleware. The clients are CLIs and do not need it, while the proxy + has no authentication of its own: `Access-Control-Allow-Origin: *` would let + any page open in the user's browser POST to the local port and spend the + configured OpenCode key. - `GET /` info: `{ name, version, backend, modelCount }`. ### 13.4 `router.ts` diff --git a/src/auth.ts b/src/auth.ts index df35fc4..b75c565 100644 --- a/src/auth.ts +++ b/src/auth.ts @@ -1,5 +1,3 @@ -import type { Format } from "./translate/types.js"; - /** * Extract the OpenCode key from a client request (spec §7.4): * 1. `x-api-key` @@ -18,22 +16,12 @@ export function extractApiKey(headers: Headers): string | null { } /** - * Inject the key into upstream headers in the format-correct position - * (spec §7.4). `apiKey === undefined` (free backend) → remove all auth - * headers. + * Strip every auth header from an upstream request (spec §7.4). The free + * backend sends no credential at all, and the provider helpers unconditionally + * write one, so this runs after them. */ -export function injectAuth(headers: Headers, format: Format, apiKey?: string): void { - if (apiKey === undefined) { - headers.delete("x-api-key"); - headers.delete("authorization"); - headers.delete("x-goog-api-key"); - return; - } - if (format === "anthropic" || format === "google") { - headers.set("x-api-key", apiKey); - headers.delete("authorization"); - } else { - headers.set("authorization", `Bearer ${apiKey}`); - headers.delete("x-api-key"); - } -} \ No newline at end of file +export function clearAuth(headers: Headers): void { + headers.delete("x-api-key"); + headers.delete("authorization"); + headers.delete("x-goog-api-key"); +} diff --git a/src/capability.ts b/src/capability.ts index cbf679c..bdb94c3 100644 --- a/src/capability.ts +++ b/src/capability.ts @@ -47,6 +47,24 @@ export function applyReasoningEffort( if (options.toggle) upstreamBody.thinking = { type: "enabled" }; } +/** + * DeepSeek-family thinking mode rejects any request whose history contains an + * assistant message without `reasoning_content` once tools are in play + * ("The `reasoning_content` in the thinking mode must be passed back to the + * API.", HTTP 400). Claude Code drops thinking blocks from older turns, so the + * trace is genuinely gone by then — an empty string satisfies the check. + * + * Only fills messages that carry none; real traces translated from `thinking` + * blocks are left untouched. Mutates `upstreamBody` in place. + */ +export function backfillReasoningContent(upstreamBody: Record): void { + if (!Array.isArray(upstreamBody.messages)) return; + for (const msg of upstreamBody.messages) { + if (msg?.role !== "assistant") continue; + if (typeof msg.reasoning_content !== "string") msg.reasoning_content = ""; + } +} + /** * Pick the advertised effort value closest to the requested budget. The * thresholds mirror Claude Code's three tiers; `none`/`minimal` are never diff --git a/src/data/models.static.ts b/src/data/models.static.ts index d5501c6..94c2f0f 100644 --- a/src/data/models.static.ts +++ b/src/data/models.static.ts @@ -87,6 +87,10 @@ const CTX: Record = { "mimo-v2-omni": [1_048_576, 131_072], "hy3": [256_000, 64_000], "hy3-preview": [256_000, 64_000], + // Free-lane ids are deliberately absent: their windows differ from the paid + // sibling's (deepseek-v4-flash is 1M paid, 200K free) and change without + // notice, so they come from `providers.opencode.models..limit` at + // refresh, never from this table. DEFAULT_CONTEXT covers the cold start. }; /** Free-tier display names (user-facing in the Claude Code picker). */ diff --git a/src/index.ts b/src/index.ts index f994443..357d31f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -50,6 +50,7 @@ async function refreshModels(): Promise { baseUrl: config.baseUrl, cacheFile: config.modelCacheFile, logger, + maxCacheAgeSeconds: config.modelCacheTtl, }); } catch (err) { logger.warn(`model refresh failed: ${(err as Error).message}`); @@ -57,12 +58,15 @@ async function refreshModels(): Promise { } void refreshModels(); -const refreshTimer = setInterval(refreshModels, config.modelCacheTtl * 1000); -refreshTimer.unref(); +// TTL 0 means "startup refresh only". Scheduling it would be setInterval(…, 0), +// which hammers the discovery and catalog endpoints in a tight loop. +const refreshTimer = + config.modelCacheTtl > 0 ? setInterval(refreshModels, config.modelCacheTtl * 1000) : undefined; +refreshTimer?.unref(); function shutdown(signal: string): void { logger.info(`received ${signal}, shutting down`); - clearInterval(refreshTimer); + if (refreshTimer) clearInterval(refreshTimer); server.close(() => process.exit(0)); // Force-exit if in-flight requests refuse to drain. setTimeout(() => process.exit(0), 5000).unref(); diff --git a/src/modelRegistry.ts b/src/modelRegistry.ts index 3eb15e1..1e3eafc 100644 --- a/src/modelRegistry.ts +++ b/src/modelRegistry.ts @@ -47,6 +47,11 @@ export interface RegistryRefreshOptions { logger: Logger; /** Per-fetch timeout in ms (default 3000). */ timeoutMs?: number; + /** + * Skip the network when the cache on disk is younger than this many seconds. + * Omit to always refresh. + */ + maxCacheAgeSeconds?: number; } export interface ModelRegistry { @@ -104,6 +109,8 @@ interface CacheFile { } interface CatalogMeta { + /** Catalog prices this model at zero, i.e. it is served on the free lane. */ + free?: boolean; contextWindow?: number; maxOutput?: number; capabilities?: Partial; @@ -150,7 +157,13 @@ export function createRegistry(backend: Backend): ModelRegistry { let entry = entries.get(base); if (!entry) { const real = fromAliasId(base); - if (real) entry = entries.get(real); + if (real) { + const candidate = entries.get(real); + // The alias carries the wire format. Honouring an alias whose format + // disagrees with the registry would route the request through the + // wrong translator and produce a body the upstream cannot parse. + if (candidate && aliasFormat(base) === candidate.format) entry = candidate; + } } if (!entry) return undefined; return { entry, contextVariant }; @@ -160,6 +173,14 @@ export function createRegistry(backend: Backend): ModelRegistry { return `${ALIAS_PREFIX}${entry.format}--${entry.id}`; } + /** Format segment of an alias id, or undefined when it is not an alias. */ + function aliasFormat(alias: string): string | undefined { + if (!alias.startsWith(ALIAS_PREFIX)) return undefined; + const rest = alias.slice(ALIAS_PREFIX.length); + const sep = rest.indexOf("--"); + return sep === -1 ? undefined : rest.slice(0, sep); + } + function fromAliasId(alias: string): string | undefined { if (!alias.startsWith(ALIAS_PREFIX)) return undefined; const rest = alias.slice(ALIAS_PREFIX.length); @@ -168,14 +189,15 @@ export function createRegistry(backend: Backend): ModelRegistry { return rest.slice(sep + 2); } - /** Load the cache file; returns entries or undefined on any failure. */ - async function loadCache(cacheFile: string): Promise { + /** Load the cache file; returns it or undefined on any failure. */ + async function loadCache(cacheFile: string): Promise { try { const raw = await readFile(expandHome(cacheFile), "utf8"); const parsed = JSON.parse(raw) as CacheFile; if (parsed.version !== CACHE_VERSION || parsed.backend !== backend) return undefined; if (!Array.isArray(parsed.models)) return undefined; - return parsed.models; + if (typeof parsed.fetchedAt !== "number") return undefined; + return parsed; } catch { return undefined; } @@ -295,7 +317,9 @@ export function createRegistry(backend: Backend): ModelRegistry { const limit = (v.limit ?? {}) as Record; const modalities = (v.modalities ?? {}) as Record; const input = Array.isArray(modalities.input) ? modalities.input : []; + const cost = (v.cost ?? {}) as Record; return { + free: cost.input === 0 && cost.output === 0, contextWindow: limit.context, maxOutput: limit.output, reasoningOptions: parseReasoningOptions(v.reasoning_options), @@ -334,9 +358,16 @@ export function createRegistry(backend: Backend): ModelRegistry { // the static snapshot so checked-in capability metadata survives; the // cache fills in last-known context/output and adds discovered ids. const cached = await loadCache(cacheFile); - if (cached && cached.length > 0) { - applyCache(cached); - logger.debug(`model cache loaded (${cached.length} models)`); + if (cached && cached.models.length > 0) { + applyCache(cached.models); + logger.debug(`model cache loaded (${cached.models.length} models)`); + // `fetchedAt` exists so a restart inside the TTL does not re-download + // discovery plus the multi-megabyte catalog for an answer it already has. + const age = Math.floor(Date.now() / 1000) - cached.fetchedAt; + if (opts.maxCacheAgeSeconds !== undefined && age >= 0 && age < opts.maxCacheAgeSeconds) { + logger.debug(`model cache is ${age}s old, skipping refresh`); + return; + } } // 2. Live discovery + catalog metadata. @@ -348,10 +379,22 @@ export function createRegistry(backend: Backend): ModelRegistry { // fall through with whatever we have } - if (liveIds.length > 0) { + // The free backend shares the Zen base URL, so discovery hands back every + // paid model too. Serving those would put ids in the client's picker that + // the free lane answers with a 401, since it sends no key at all — keep + // only what the catalog prices at zero. + const servable = + backend === "free" + // Drop only what the catalog positively prices as paid. An id the + // catalog has not caught up with yet stays — hiding a model we cannot + // classify is worse than listing one that might 401. + ? liveIds.filter((id) => catalog.get(id)?.free !== false) + : liveIds; + + if (servable.length > 0) { // Live ids win: keep existing metadata where known, else defaults. const merged: Array & { capabilities?: Partial }> = []; - for (const id of liveIds) { + for (const id of servable) { const existing = entries.get(id); const cat = catalog.get(id); merged.push({ @@ -367,12 +410,16 @@ export function createRegistry(backend: Backend): ModelRegistry { capabilities: { ...(existing?.capabilities ?? {}), ...(cat?.capabilities ?? {}) }, }); } - // Static-only models not seen live are kept (docs may lag the API). + // Snapshot models not seen live are kept (docs may lag the API). Only + // the snapshot — carrying over everything currently in `entries` would + // resurrect ids a stale cache added, including ones discovery just + // filtered out. + const snapshot = new Set(STATIC_MODELS[backend].map((sm) => sm.id)); for (const [id, e] of entries) { - if (!merged.some((m) => m.id === id)) merged.push(e); + if (snapshot.has(id) && !merged.some((m) => m.id === id)) merged.push(e); } mergeEntries(merged); - logger.info(`model discovery: ${liveIds.length} live ids, ${entries.size} total`); + logger.info(`model discovery: ${liveIds.length} live ids, ${servable.length} servable, ${entries.size} total`); } else if (catalog.size > 0) { // Discovery unavailable (offline, 404, auth): still take catalog // metadata for the ids we already know, so reasoning options and diff --git a/src/router.ts b/src/router.ts index 9bec884..4907aa8 100644 --- a/src/router.ts +++ b/src/router.ts @@ -1,6 +1,6 @@ import type { Context } from "hono"; -import { extractApiKey } from "./auth.js"; -import { applyReasoningEffort, stripUnsupported } from "./capability.js"; +import { clearAuth, extractApiKey } from "./auth.js"; +import { applyReasoningEffort, backfillReasoningContent, stripUnsupported } from "./capability.js"; import type { Config } from "./config.js"; import { ProxyError } from "./errors.js"; import type { Logger } from "./logging.js"; @@ -84,6 +84,7 @@ export async function handleMessages(c: Context, deps: RouterDeps): Promise = { + "content-type": upstream.headers.get("content-type") ?? "application/json", + }; + // Claude Code waits out a 429 based on this; dropping it makes it guess. + const retryAfter = upstream.headers.get("retry-after"); + if (retryAfter) errorHeaders["retry-after"] = retryAfter; + return new Response(bodyText, { status: upstream.status, headers: errorHeaders }); } if (isStream) { @@ -210,11 +210,7 @@ export async function handleCountTokens(c: Context, deps: RouterDeps): Promise= 1_000_000 || resolved.contextVariant === "1m", }); provider.modifyHeaders(headers, apiKey ?? "", ""); - if (apiKey === undefined) { - headers.delete("x-api-key"); - headers.delete("authorization"); - headers.delete("x-goog-api-key"); - } + if (apiKey === undefined) clearAuth(headers); const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), config.requestTimeoutMs); try { @@ -260,6 +256,10 @@ function localEstimate(body: any): Response { else if (obj.input && typeof obj.input === "object") { chars += JSON.stringify(obj.input).length; } + // tool_result nests its payload under `content`, as a string or as + // further blocks. In an agent session that payload is most of the + // context, so skipping it made the estimate useless. + if (obj.content !== undefined) countText(obj.content); } }; countText(body?.system); @@ -333,8 +333,13 @@ async function fetchWithRetry( }); if (res.ok || (res.status < 500 && res.status !== 429)) return res; // Transient upstream failure: drain the body so the socket can be reused. - await res.text().catch(() => undefined); - if (attempt >= maxRetries) return res; + const drained = await res.text().catch(() => ""); + // Out of attempts: hand back the text we drained. Returning `res` itself + // would give the caller a consumed body, and reading it again throws — + // turning a 429 the client knows how to back off from into a 500. + if (attempt >= maxRetries) { + return new Response(drained, { status: res.status, headers: res.headers }); + } logger.warn(`upstream ${res.status}, retrying (${attempt + 1}/${maxRetries})`); } catch (err) { lastErr = err as Error; diff --git a/src/server.ts b/src/server.ts index 75d6001..1bc50a2 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,6 +1,5 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; import { Hono } from "hono"; -import { cors } from "hono/cors"; import type { Config } from "./config.js"; import { anthropicError, ProxyError } from "./errors.js"; import { registerHealth } from "./health.js"; @@ -24,15 +23,10 @@ export function createApp(deps: ServerDeps): Hono { const { config, logger } = deps; const app = new Hono(); - app.use( - "*", - cors({ - origin: "*", - allowMethods: ["POST", "GET", "OPTIONS"], - allowHeaders: ["*"], - maxAge: 86_400, - }), - ); + // No CORS headers on purpose. The clients are CLIs, which do not need them, + // and this proxy has no authentication of its own: with `Access-Control- + // Allow-Origin: *` any page in the user's browser could POST to + // 127.0.0.1:8787 and spend the configured OpenCode key. registerHealth(app, deps); diff --git a/src/stream.ts b/src/stream.ts index 5bac7a4..940287a 100644 --- a/src/stream.ts +++ b/src/stream.ts @@ -25,9 +25,17 @@ export function pumpStream(opts: PumpOptions): ReadableStream { const sameFormat = upstreamFormat === clientFormat; const usageParser = getProvider(upstreamFormat, { model: "", providerModel: "" }).createUsageParser(); + // Latched by cancel() (client hung up) and by the first failed enqueue, so + // both start() and the keep-alive timer can see it. + let closed = false; + // Held for cancel(): start() locks the upstream body, and cancelling a + // locked stream throws. The reader is the only handle that can release it. + let upstreamReader: ReadableStreamDefaultReader | undefined; + return new ReadableStream({ async start(controller) { const reader = upstream.body?.getReader(); + upstreamReader = reader; if (!reader) { controller.close(); return; @@ -38,18 +46,30 @@ export function pumpStream(opts: PumpOptions): ReadableStream { let sawMessageStop = false; let sawError = false; + // A cancelled stream (client hung up) closes the controller while the + // read loop is still awaiting upstream, so every enqueue after that + // point throws. Inside the interval callback that throw is uncaught and + // takes the process down, so all writes go through this guard. + const write = (text: string): void => { + if (closed) return; + try { + controller.enqueue(encoder.encode(text)); + lastWrite = Date.now(); + } catch { + closed = true; + } + }; + const keepAlive = setInterval(() => { - if (upstreamDone) return; + if (upstreamDone || closed) return; if (Date.now() - lastWrite >= KEEP_ALIVE_MS) { - controller.enqueue(encoder.encode(`event: ping\ndata: {"type":"ping"}\n\n`)); - lastWrite = Date.now(); + write(`event: ping\ndata: {"type":"ping"}\n\n`); } }, KEEP_ALIVE_MS); const emit = (block: string): void => { if (!block) return; - controller.enqueue(encoder.encode(block + "\n\n")); - lastWrite = Date.now(); + write(block + "\n\n"); }; const processBlock = (block: string): void => { @@ -57,7 +77,9 @@ export function pumpStream(opts: PumpOptions): ReadableStream { if (ev === "message_stop") sawMessageStop = true; // Mid-stream upstream error: forward verbatim and suppress the synthetic // `message_stop` so we close the stream as the spec requires (§9.2.3). - if (ev === "error") sawError = true; + // oa-compat upstreams have no event line — they just send an error + // envelope as data, so the payload has to be inspected too. + if (ev === "error" || isErrorPayload(block)) sawError = true; if (upstreamFormat === "oa-compat" && block.trim() === "data: [DONE]") return; usageParser.parse(block); if (sameFormat) { @@ -84,7 +106,7 @@ export function pumpStream(opts: PumpOptions): ReadableStream { } if (clientFormat === "anthropic" && !sawMessageStop && !sawError) { - controller.enqueue(encoder.encode(`event: message_stop\ndata: {"type":"message_stop"}\n\n`)); + write(`event: message_stop\ndata: {"type":"message_stop"}\n\n`); } // Cost ping (spec §9.2.7): after the final chunk, from normalized usage. @@ -92,17 +114,21 @@ export function pumpStream(opts: PumpOptions): ReadableStream { const usage = usageParser.retrieve(); if (usage) { const normalized = getProvider(upstreamFormat, { model: "", providerModel: "" }).normalizeUsage(usage); - controller.enqueue( - encoder.encode( - `event: ping\ndata: ${JSON.stringify({ type: "ping", cost: normalized })}\n\n`, - ), - ); + write(`event: ping\ndata: ${JSON.stringify({ type: "ping", cost: normalized })}\n\n`); } } - controller.close(); + if (!closed) { + closed = true; + controller.close(); + } }, cancel() { - upstream.body?.cancel(); + closed = true; + // Abort the upstream fetch so it does not keep streaming into nothing. + // This runs inside a stream callback, where a throw is fatal, and both + // calls can reject on an already-finished body. + const pending = upstreamReader ? upstreamReader.cancel() : upstream.body?.cancel(); + void Promise.resolve(pending).catch(() => undefined); }, }); } @@ -129,6 +155,20 @@ function createSseSplitter(): { push: (text: string) => string[]; flush: () => s }; } +/** True when an SSE block's data payload is an error envelope. */ +function isErrorPayload(block: string): boolean { + const line = block.split("\n").find((l) => l.startsWith("data:")); + if (!line) return false; + const payload = line.slice(5).trim(); + if (payload === "[DONE]" || !payload.startsWith("{")) return false; + try { + const json = JSON.parse(payload) as { error?: unknown }; + return json?.error !== undefined && json.error !== null; + } catch { + return false; + } +} + function eventType(block: string): string { const m = block.match(/^event:\s*(\S+)/m); return m ? (m[1] ?? "message") : "message"; diff --git a/src/translate/anthropic.ts b/src/translate/anthropic.ts index 73fa5ae..22b5120 100644 --- a/src/translate/anthropic.ts +++ b/src/translate/anthropic.ts @@ -9,6 +9,23 @@ type Usage = { output_tokens?: number; }; +/** + * Add one beta flag to `anthropic-beta` without discarding the ones the client + * already asked for (token-efficient tools, fine-grained streaming, …) — the + * header is a comma-separated list, and overwriting it silently turns those + * features off. + */ +function addBeta(headers: Headers, flag: string): void { + const current = headers.get("anthropic-beta"); + if (!current) { + headers.set("anthropic-beta", flag); + return; + } + const flags = current.split(",").map((f) => f.trim()).filter(Boolean); + if (flags.includes(flag)) return; + headers.set("anthropic-beta", [...flags, flag].join(",")); +} + /** Anthropic Messages provider helper (spec §7.2). */ export function anthropicHelper(opts: ProviderHelperOptions): ProviderHelper { // 1M-context support is derived from the registry entry / `[1m]` variant @@ -24,7 +41,7 @@ export function anthropicHelper(opts: ProviderHelperOptions): ProviderHelper { modifyHeaders: (headers: Headers, apiKey: string, _stickyId: string) => { headers.set("x-api-key", apiKey); headers.set("anthropic-version", headers.get("anthropic-version") ?? "2023-06-01"); - if (supports1m) headers.set("anthropic-beta", "context-1m-2025-08-07"); + if (supports1m) addBeta(headers, "context-1m-2025-08-07"); }, modifyBody: (body: Record) => body, createBinaryStreamDecoder: () => undefined, @@ -32,11 +49,13 @@ export function anthropicHelper(opts: ProviderHelperOptions): ProviderHelper { let usage: Usage | undefined; return { parse: (chunk: string) => { - const data = chunk.split("\n").find((l) => l.startsWith("data: ")); + // `data:{...}` without the optional space is equally valid SSE; + // matching only "data: " silently dropped usage from such upstreams. + const data = chunk.split("\n").find((l) => l.startsWith("data:")); if (!data) return; let json: any; try { - json = JSON.parse(data.slice(6)); + json = JSON.parse(data.slice(5).trim()); } catch { return; } diff --git a/src/translate/converters/anthropic.ts b/src/translate/converters/anthropic.ts index ce73ac6..4dd9f27 100644 --- a/src/translate/converters/anthropic.ts +++ b/src/translate/converters/anthropic.ts @@ -522,11 +522,13 @@ export function createFromAnthropicChunk(): (part: string) => string { export function createToAnthropicChunk(): (chunk: CommonChunk) => string { let started = false; let blockIndex = 0; - let toolBlockIndex = -1; + /** oa-compat tool_call index → anthropic content block index. */ + const toolBlocks = new Map(); let textBlockIndex = -1; let thinkingBlockIndex = -1; let sawFinish = false; let sawUsage = false; + let stopReason: string | null = null; return (chunk: CommonChunk): string => { const events: string[] = []; @@ -583,10 +585,10 @@ export function createToAnthropicChunk(): (chunk: CommonChunk) => string { thinkingBlockIndex = -1; } if (delta?.content) { - if (toolBlockIndex !== -1) { - events.push(sse("content_block_stop", { type: "content_block_stop", index: toolBlockIndex })); - toolBlockIndex = -1; + for (const target of toolBlocks.values()) { + events.push(sse("content_block_stop", { type: "content_block_stop", index: target })); } + toolBlocks.clear(); // One text block spanning every consecutive text delta. Opening and // closing a block per delta is legal SSE but renders as one content // block per token, which the client lays out as separate lines. @@ -610,6 +612,11 @@ export function createToAnthropicChunk(): (chunk: CommonChunk) => string { } for (const tc of delta?.tool_calls ?? []) { + // Parallel tool calls stream interleaved, keyed by the oa-compat + // `index`. Each one owns its own anthropic block for the whole stream — + // a single "current tool block" would route call 0's argument deltas + // into call 1's block and hand the model's input to the wrong tool. + const callIndex = tc.index ?? 0; if (tc.function?.name) { // A tool call ends the text block that preceded it. if (textBlockIndex !== -1) { @@ -628,27 +635,34 @@ export function createToAnthropicChunk(): (chunk: CommonChunk) => string { }, }), ); - toolBlockIndex = blockIndex; + toolBlocks.set(callIndex, blockIndex); blockIndex++; } if (tc.function?.arguments) { - events.push( - sse("content_block_delta", { - type: "content_block_delta", - index: toolBlockIndex, - delta: { type: "input_json_delta", partial_json: tc.function.arguments }, - }), - ); + const target = toolBlocks.get(callIndex); + // Arguments before any name: the upstream never opened the call, so + // there is no block to attach them to. Dropping beats corrupting a + // sibling call's input. + if (target !== undefined) { + events.push( + sse("content_block_delta", { + type: "content_block_delta", + index: target, + delta: { type: "input_json_delta", partial_json: tc.function.arguments }, + }), + ); + } } } const finish = choice?.finish_reason; if (finish && !sawFinish) { sawFinish = true; - if (toolBlockIndex !== -1) { - events.push(sse("content_block_stop", { type: "content_block_stop", index: toolBlockIndex })); - toolBlockIndex = -1; + stopReason = mapFinishReason(finish); + for (const target of toolBlocks.values()) { + events.push(sse("content_block_stop", { type: "content_block_stop", index: target })); } + toolBlocks.clear(); if (textBlockIndex !== -1) { events.push(sse("content_block_stop", { type: "content_block_stop", index: textBlockIndex })); textBlockIndex = -1; @@ -656,7 +670,7 @@ export function createToAnthropicChunk(): (chunk: CommonChunk) => string { events.push( sse("message_delta", { type: "message_delta", - delta: { stop_reason: mapFinishReason(finish), stop_sequence: null }, + delta: { stop_reason: stopReason, stop_sequence: null }, usage: { output_tokens: 0 }, }), ); @@ -667,7 +681,10 @@ export function createToAnthropicChunk(): (chunk: CommonChunk) => string { events.push( sse("message_delta", { type: "message_delta", - delta: { stop_reason: null, stop_sequence: null }, + // Usage lands in its own message_delta because oa-compat only sends + // it after the finish chunk. Repeat the stop reason rather than + // sending null, which reads as "undo the stop reason I just gave". + delta: { stop_reason: stopReason, stop_sequence: null }, usage: { input_tokens: chunk.usage.prompt_tokens ?? 0, output_tokens: chunk.usage.completion_tokens ?? 0, diff --git a/src/translate/converters/index.ts b/src/translate/converters/index.ts index ae2a862..a378e1c 100644 --- a/src/translate/converters/index.ts +++ b/src/translate/converters/index.ts @@ -202,7 +202,10 @@ function toTarget(chunk: CommonChunk, to: Format): string { function openaiChunkToSse(chunk: CommonChunk): string { const events: string[] = []; - const delta = chunk.choices[0].delta; + // `stream_options.include_usage` produces a final chunk with `choices: []`. + const choice = chunk.choices[0]; + if (!choice) return ""; + const delta = choice.delta; if (delta.content) { events.push( `event: response.output_text.delta\ndata: ${JSON.stringify({ type: "response.output_text.delta", delta: delta.content })}`, @@ -228,7 +231,7 @@ function openaiChunkToSse(chunk: CommonChunk): string { ); } } - if (chunk.choices[0].finish_reason) { + if (choice.finish_reason) { events.push( `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", @@ -241,7 +244,8 @@ function openaiChunkToSse(chunk: CommonChunk): string { function googleChunkFromCommon(chunk: CommonChunk): Record { const parts: Array> = []; - const delta = chunk.choices[0].delta; + const choice = chunk.choices[0]; + const delta = choice?.delta ?? {}; if (delta.content) parts.push({ text: delta.content }); for (const tc of delta.tool_calls ?? []) { if (tc.function?.name) parts.push({ functionCall: { name: tc.function.name, args: {} } }); @@ -249,16 +253,14 @@ function googleChunkFromCommon(chunk: CommonChunk): Record { return { candidates: [ { - index: chunk.choices[0].index, + index: choice?.index ?? 0, content: { role: "model", parts }, - finishReason: chunk.choices[0].finish_reason - ? chunk.choices[0].finish_reason === "stop" - ? "STOP" - : chunk.choices[0].finish_reason === "tool_calls" - ? "TOOL_CALLS" - : chunk.choices[0].finish_reason === "length" - ? "MAX_TOKENS" - : "STOP" + finishReason: choice?.finish_reason + ? choice.finish_reason === "tool_calls" + ? "TOOL_CALLS" + : choice.finish_reason === "length" + ? "MAX_TOKENS" + : "STOP" : undefined, }, ], diff --git a/src/translate/google.ts b/src/translate/google.ts index 32997f8..43bf2b5 100644 --- a/src/translate/google.ts +++ b/src/translate/google.ts @@ -26,11 +26,13 @@ export function googleHelper(opts: ProviderHelperOptions): ProviderHelper { let usage: Usage | undefined; return { parse: (chunk: string) => { - const data = chunk.split("\n").find((l) => l.startsWith("data: ")); + // `data:{...}` without the optional space is equally valid SSE; + // matching only "data: " silently dropped usage from such upstreams. + const data = chunk.split("\n").find((l) => l.startsWith("data:")); if (!data) return; let json: any; try { - json = JSON.parse(data.slice(6)); + json = JSON.parse(data.slice(5).trim()); } catch { return; } diff --git a/src/translate/openai-compat.ts b/src/translate/openai-compat.ts index 865b24f..032d457 100644 --- a/src/translate/openai-compat.ts +++ b/src/translate/openai-compat.ts @@ -33,11 +33,13 @@ export function oaCompatHelper(opts: ProviderHelperOptions): ProviderHelper { let usage: Usage | undefined; return { parse: (chunk: string) => { - const data = chunk.split("\n").find((l) => l.startsWith("data: ")); + // `data:{...}` without the optional space is equally valid SSE; + // matching only "data: " silently dropped usage from such upstreams. + const data = chunk.split("\n").find((l) => l.startsWith("data:")); if (!data) return; let json: any; try { - json = JSON.parse(data.slice(6)); + json = JSON.parse(data.slice(5).trim()); } catch { return; } diff --git a/src/translate/openai.ts b/src/translate/openai.ts index 0c25759..beef408 100644 --- a/src/translate/openai.ts +++ b/src/translate/openai.ts @@ -25,10 +25,10 @@ export function openaiHelper(_opts: ProviderHelperOptions): ProviderHelper { parse: (chunk: string) => { const [event, data] = chunk.split("\n"); if (event !== "event: response.completed") return; - if (!data?.startsWith("data: ")) return; + if (!data?.startsWith("data:")) return; let json: any; try { - json = JSON.parse(data.slice(6)); + json = JSON.parse(data.slice(5).trim()); } catch { return; } diff --git a/test/auth.test.ts b/test/auth.test.ts index 0515455..f0e7782 100644 --- a/test/auth.test.ts +++ b/test/auth.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { extractApiKey, injectAuth } from "../src/auth.js"; +import { clearAuth, extractApiKey } from "../src/auth.js"; describe("extractApiKey", () => { it("reads x-api-key", () => { @@ -19,34 +19,12 @@ describe("extractApiKey", () => { }); }); -describe("injectAuth", () => { - it("sets x-api-key for anthropic", () => { - const h = new Headers(); - injectAuth(h, "anthropic", "k"); - expect(h.get("x-api-key")).toBe("k"); - expect(h.get("authorization")).toBeNull(); - }); - - it("sets x-api-key for google", () => { - const h = new Headers(); - injectAuth(h, "google", "k"); - expect(h.get("x-api-key")).toBe("k"); - }); - - it("sets Bearer for oa-compat and openai", () => { - for (const format of ["oa-compat", "openai"] as const) { - const h = new Headers(); - injectAuth(h, format, "k"); - expect(h.get("authorization")).toBe("Bearer k"); - expect(h.get("x-api-key")).toBeNull(); - } - }); - - it("removes all auth headers when key is undefined", () => { - const h = new Headers({ "x-api-key": "a", authorization: "Bearer b", "x-goog-api-key": "c" }); - injectAuth(h, "anthropic", undefined); +describe("clearAuth", () => { + it("removes every auth header", () => { + const h = new Headers({ "x-api-key": "k", authorization: "Bearer k", "x-goog-api-key": "k" }); + clearAuth(h); expect(h.get("x-api-key")).toBeNull(); expect(h.get("authorization")).toBeNull(); expect(h.get("x-goog-api-key")).toBeNull(); }); -}); \ No newline at end of file +}); diff --git a/test/capability.test.ts b/test/capability.test.ts index 3ea35f0..1ec3695 100644 --- a/test/capability.test.ts +++ b/test/capability.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { applyReasoningEffort, stripUnsupported } from "../src/capability.js"; +import { applyReasoningEffort, backfillReasoningContent, stripUnsupported } from "../src/capability.js"; import { createLogger } from "../src/logging.js"; const logger = createLogger("error"); @@ -135,3 +135,21 @@ describe("applyReasoningEffort", () => { expect(b).toEqual({}); }); }); + +describe("backfillReasoningContent", () => { + it("gives every assistant message a reasoning_content, keeping real traces", () => { + const body: Record = { + messages: [ + { role: "user", content: "hi" }, + { role: "assistant", content: "a", tool_calls: [{ id: "1" }] }, + { role: "tool", tool_call_id: "1", content: "ok" }, + { role: "assistant", content: "b", reasoning_content: "kept" }, + ], + }; + backfillReasoningContent(body); + expect(body.messages[1].reasoning_content).toBe(""); + expect(body.messages[3].reasoning_content).toBe("kept"); + expect(body.messages[0].reasoning_content).toBeUndefined(); + expect(body.messages[2].reasoning_content).toBeUndefined(); + }); +}); diff --git a/test/converters.test.ts b/test/converters.test.ts index df68b8d..19fd3fc 100644 --- a/test/converters.test.ts +++ b/test/converters.test.ts @@ -418,6 +418,30 @@ describe("streamed text blocks", () => { expect(out.match(/"content_block_delta","index":(\d+)/g)?.every((m) => m.endsWith(":0"))).toBe(true); }); + it("keeps parallel tool calls in separate blocks", () => { + const to = createStreamPartConverter("oa-compat", "anthropic")!; + const chunk = (delta: unknown, finish: string | null = null) => + to( + `data: ${JSON.stringify({ + id: "c", + object: "chat.completion.chunk", + created: 0, + model: "m", + choices: [{ index: 0, delta, finish_reason: finish }], + })}`, + ); + chunk({ tool_calls: [{ index: 0, id: "a", type: "function", function: { name: "Read", arguments: "" } }] }); + chunk({ tool_calls: [{ index: 1, id: "b", type: "function", function: { name: "Grep", arguments: "" } }] }); + // Arguments for call 0 arrive after call 1 opened: they belong to block 0. + const out = chunk({ tool_calls: [{ index: 0, function: { arguments: '{"p":1}' } }] }); + expect(out).toContain('"content_block_delta","index":0'); + expect(out).not.toContain('"content_block_delta","index":1'); + // Both blocks are closed at the finish. + const end = chunk({}, "tool_calls"); + expect(end).toContain('"content_block_stop","index":0'); + expect(end).toContain('"content_block_stop","index":1'); + }); + it("closes the text block before opening a tool block", () => { const to = createStreamPartConverter("oa-compat", "anthropic")!; const chunk = (delta: unknown) => @@ -437,3 +461,16 @@ describe("streamed text blocks", () => { expect(out.indexOf("content_block_stop")).toBeLessThan(out.indexOf('"type":"tool_use"')); }); }); + +describe("anthropic provider headers", () => { + it("adds context-1m without dropping the client's other betas", async () => { + const { getProvider } = await import("../src/translate/provider.js"); + const p = getProvider("anthropic", { model: "claude-sonnet-5", providerModel: "claude-sonnet-5", supports1m: true }); + const headers = new Headers({ "anthropic-beta": "token-efficient-tools-2024-11-01" }); + p.modifyHeaders(headers, "k", ""); + expect(headers.get("anthropic-beta")).toBe("token-efficient-tools-2024-11-01,context-1m-2025-08-07"); + // Idempotent: a second pass does not duplicate the flag. + p.modifyHeaders(headers, "k", ""); + expect(headers.get("anthropic-beta")).toBe("token-efficient-tools-2024-11-01,context-1m-2025-08-07"); + }); +}); diff --git a/test/countTokens.test.ts b/test/countTokens.test.ts index fb52be0..9a80da2 100644 --- a/test/countTokens.test.ts +++ b/test/countTokens.test.ts @@ -113,4 +113,25 @@ describe("handleCountTokens (POST /v1/messages/count_tokens)", () => { await new Promise((r) => server.close(() => r())); } }); -}); \ No newline at end of file +}); +describe("local estimate", () => { + it("counts tool_result payloads", async () => { + const deps = makeDeps("free", "http://127.0.0.1:1/v1"); + const body = { + model: "deepseek-v4-flash-free", + messages: [ + { + role: "user", + content: [ + { type: "tool_result", tool_use_id: "t1", content: "x".repeat(400) }, + { type: "tool_result", tool_use_id: "t2", content: [{ type: "text", text: "y".repeat(400) }] }, + ], + }, + ], + }; + const res = await handleCountTokens(mockContext(body), deps); + const json = (await res.json()) as { input_tokens: number }; + // 800 chars / 4 — previously 0, because tool_result nests under `content`. + expect(json.input_tokens).toBe(200); + }); +}); diff --git a/test/modelRegistry.test.ts b/test/modelRegistry.test.ts index a79c49e..0389355 100644 --- a/test/modelRegistry.test.ts +++ b/test/modelRegistry.test.ts @@ -29,6 +29,14 @@ describe("createRegistry", () => { expect(m?.contextVariant).toBe("1m"); }); + it("rejects an alias whose format does not match the registry", () => { + const r = createRegistry("free"); + // deepseek-v4-flash-free is oa-compat; an anthropic-flavoured alias for it + // would otherwise resolve and be sent through the wrong translator. + expect(r.resolveModel("claude-ocx-oa-compat--deepseek-v4-flash-free")).toBeDefined(); + expect(r.resolveModel("claude-ocx-anthropic--deepseek-v4-flash-free")).toBeUndefined(); + }); + it("returns undefined for unknown ids", () => { const r = createRegistry("free"); expect(r.resolveModel("does-not-exist")).toBeUndefined(); @@ -62,4 +70,82 @@ describe("static capability metadata", () => { expect(m?.entry.capabilities.structuredOutput).toBe(false); expect(m?.entry.capabilities.promptCaching).toBe(false); }); -}); \ No newline at end of file +}); +describe("catalog refresh drives context windows", () => { + it("takes the free lane's own window from the catalog, not the static default", async () => { + const catalog = { + models: { "deepseek/deepseek-v4-flash": { limit: { context: 1_000_000, output: 384_000 } } }, + providers: { + opencode: { + models: { + // Same model, free lane: separately capped. The static snapshot has + // no entry, so this is the only source of truth. + "deepseek-v4-flash-free": { limit: { context: 200_000, output: 128_000 }, cost: { input: 0, output: 0 } }, + "longcat-2.0-free": { limit: { context: 1_000_000, output: 131_072 }, cost: { input: 0, output: 0 } }, + "claude-opus-5": { limit: { context: 1_000_000, output: 128_000 }, cost: { input: 5, output: 25 } }, + }, + }, + }, + }; + const original = globalThis.fetch; + globalThis.fetch = (async (url: any) => + new Response( + JSON.stringify( + String(url).includes("catalog.json") + ? catalog + : { data: [{ id: "deepseek-v4-flash-free" }, { id: "longcat-2.0-free" }, { id: "claude-opus-5" }] }, + ), + { status: 200 }, + )) as typeof fetch; + try { + const r = createRegistry("free"); + await r.refresh({ + baseUrl: "https://example.invalid/v1", + cacheFile: `/tmp/registry-test-${Date.now()}.json`, + logger: { debug() {}, info() {}, warn() {}, error() {} } as never, + }); + expect(r.resolveModel("deepseek-v4-flash-free")?.entry.contextWindow).toBe(200_000); + // A paid model returned by the shared Zen /models endpoint is not + // servable on the free lane, which sends no key. + expect(r.resolveModel("claude-opus-5")).toBeUndefined(); + expect(r.resolveModel("longcat-2.0-free")?.entry.contextWindow).toBe(1_000_000); + } finally { + globalThis.fetch = original; + } + }); +}); + +describe("cache freshness", () => { + it("skips the network while the cache is younger than the TTL", async () => { + const { writeFile } = await import("node:fs/promises"); + const cacheFile = `/tmp/registry-fresh-${Date.now()}.json`; + await writeFile( + cacheFile, + JSON.stringify({ + version: 1, + backend: "free", + fetchedAt: Math.floor(Date.now() / 1000), + models: [{ id: "cached-only-model", format: "oa-compat", contextWindow: 1234, maxOutput: 10 }], + }), + ); + const original = globalThis.fetch; + let calls = 0; + globalThis.fetch = (async () => { + calls++; + return new Response("{}", { status: 200 }); + }) as typeof fetch; + try { + const r = createRegistry("free"); + await r.refresh({ + baseUrl: "https://example.invalid/v1", + cacheFile, + logger: { debug() {}, info() {}, warn() {}, error() {} } as never, + maxCacheAgeSeconds: 86_400, + }); + expect(calls).toBe(0); + expect(r.resolveModel("cached-only-model")?.entry.contextWindow).toBe(1234); + } finally { + globalThis.fetch = original; + } + }); +}); diff --git a/test/router.test.ts b/test/router.test.ts index 5dbf4cb..81c75da 100644 --- a/test/router.test.ts +++ b/test/router.test.ts @@ -48,6 +48,40 @@ function makeDeps(baseUrl: string) { }; } +describe("handleMessages upstream errors", () => { + let server: Server | undefined; + + afterEach(async () => { + if (server) { + await new Promise((r) => server!.close(() => r())); + server = undefined; + } + }); + + it("forwards a retried-out 429 with its body and retry-after", async () => { + server = createServer((req, res) => { + req.resume(); + req.on("end", () => { + res.writeHead(429, { "content-type": "application/json", "retry-after": "7" }); + res.end('{"type":"error","error":{"message":"slow down"}}'); + }); + }); + await new Promise((r) => server!.listen(0, "127.0.0.1", r)); + const port = (server!.address() as { port: number }).port; + + const deps = makeDeps(`http://127.0.0.1:${port}/v1`); + deps.config = { ...deps.config, maxRetries: 1 } as never; + const res = await handleMessages( + mockContext({ model: "deepseek-v4-flash-free", max_tokens: 16, messages: [{ role: "user", content: "hi" }] }), + deps, + ); + expect(res.status).toBe(429); + // Body survives the retry drain instead of throwing "Body is unusable". + expect(await res.text()).toContain("slow down"); + expect(res.headers.get("retry-after")).toBe("7"); + }); +}); + describe("handleMessages provider body modification", () => { let server: Server | undefined; diff --git a/test/server.test.ts b/test/server.test.ts index 059f169..430c50d 100644 --- a/test/server.test.ts +++ b/test/server.test.ts @@ -30,18 +30,14 @@ function makeApp() { } describe("createApp", () => { - it("answers CORS preflight with 204 and allow headers", async () => { - const app = makeApp(); - const res = await app.request("/v1/messages", { method: "OPTIONS" }); - expect(res.status).toBe(204); - expect(res.headers.get("access-control-allow-origin")).toBe("*"); - expect(res.headers.get("access-control-allow-methods")).toContain("POST"); - }); - - it("adds CORS headers to normal responses", async () => { + it("sends no CORS headers, so browser pages cannot reach the proxy", async () => { const app = makeApp(); + // The proxy has no auth of its own and holds the OpenCode key; a wildcard + // origin would let any open page spend it. const res = await app.request("/v1/models"); - expect(res.headers.get("access-control-allow-origin")).toBe("*"); + expect(res.headers.get("access-control-allow-origin")).toBeNull(); + const preflight = await app.request("/v1/messages", { method: "OPTIONS" }); + expect(preflight.headers.get("access-control-allow-origin")).toBeNull(); }); it("serves health endpoints", async () => { @@ -66,4 +62,51 @@ describe("createApp", () => { const res = await app.request("/nope"); expect(res.status).toBe(404); }); -}); \ No newline at end of file +}); +describe("pumpStream error handling", () => { + it("does not append message_stop after an oa-compat error envelope", async () => { + const { pumpStream } = await import("../src/stream.js"); + const body = new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: {"error":{"message":"boom"}}\n\n')); + c.close(); + }, + }); + const out = pumpStream({ + upstream: new Response(body), + upstreamFormat: "oa-compat", + clientFormat: "anthropic", + }); + const text = await new Response(out).text(); + expect(text).toContain("boom"); + expect(text).not.toContain("message_stop"); + }); +}); + +describe("pumpStream cancellation", () => { + it("cancels a locked upstream body without throwing", async () => { + const { pumpStream } = await import("../src/stream.js"); + let cancelled = false; + const body = new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: {"choices":[{"index":0,"delta":{"content":"hi"}}]}\n\n')); + // Never closed: the client hangs up while upstream is still streaming. + }, + cancel() { + cancelled = true; + }, + }); + const out = pumpStream({ + upstream: new Response(body), + upstreamFormat: "oa-compat", + clientFormat: "anthropic", + }); + const reader = out.getReader(); + await reader.read(); + // start() holds a reader on the upstream body, so cancelling it through + // `upstream.body.cancel()` would throw "ReadableStream is locked". + await reader.cancel(); + await new Promise((r) => setTimeout(r, 10)); + expect(cancelled).toBe(true); + }); +});