From abb17febbf8ed9fb0888305f6c78b44ac054f8a9 Mon Sep 17 00:00:00 2001 From: tzzs Date: Mon, 21 Sep 2026 00:36:43 +0800 Subject: [PATCH 1/4] fix(streaming): read every usage token field through the provider mappers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The core layer kept parsing provider payloads itself — a raw `total_tokens`/`cached_tokens` read in streaming/common.ts — against the CLAUDE.md architecture rule. Both field lists move into providers/usage/openai.ts, the capture stores the mapped total, and the dead withCacheTokens/argsOf paths are removed. --- src/providers/usage/index.ts | 6 ++-- src/providers/usage/openai.ts | 3 +- src/streaming/common.ts | 58 ++++++++--------------------------- src/streaming/to-anthropic.ts | 5 ++- 4 files changed, 20 insertions(+), 52 deletions(-) diff --git a/src/providers/usage/index.ts b/src/providers/usage/index.ts index 9788b5a..73156b7 100644 --- a/src/providers/usage/index.ts +++ b/src/providers/usage/index.ts @@ -5,9 +5,9 @@ import { extractResponsesUsage, extractChatUsage, mapResponsesUsage, mapChatUsag import { extractAnthropicUsage, mapAnthropicUsage } from "./anthropic.js"; export type { TokenUsage, UsageContext } from "../../usage/types.js"; -// Bare-usage-object field mappers, re-exported so src/streaming/common.ts can -// derive cache/reasoning tokens from a raw usage chunk without keeping its -// own copy of these field names. +// Bare-usage-object field mappers, re-exported so the streaming pipes in +// src/streaming/common.ts can read token fields from raw usage chunks through +// the same field lists as the non-streaming path. export { mapResponsesUsage, mapChatUsage, mapAnthropicUsage }; export function extractUsage(response: JsonRecord, model: ProviderModel, ctx: UsageContext): TokenUsage | null { diff --git a/src/providers/usage/openai.ts b/src/providers/usage/openai.ts index 0548435..d345c4f 100644 --- a/src/providers/usage/openai.ts +++ b/src/providers/usage/openai.ts @@ -30,7 +30,8 @@ function mapUsage(usage: JsonRecord | undefined, ctx: UsageContext, names: { result.inputTokens = recCount(usage, names.input) ?? 0; result.outputTokens = recCount(usage, names.output) ?? 0; result.totalTokens = recCount(usage, "total_tokens") ?? (result.inputTokens + result.outputTokens); - const cached = recCount(recObj(usage, names.cached), "cached_tokens"); + // Some providers flatten the details objects into a bare top-level cached_tokens; the nested shape is read first. + const cached = recCount(recObj(usage, names.cached), "cached_tokens") ?? recCount(usage, "cached_tokens"); if (cached !== undefined) result.cachedInputTokens = cached; const reasoning = recCount(recObj(usage, names.reasoning), "reasoning_tokens"); if (reasoning !== undefined) result.reasoningTokens = reasoning; diff --git a/src/streaming/common.ts b/src/streaming/common.ts index 4d40bb4..b6ff753 100644 --- a/src/streaming/common.ts +++ b/src/streaming/common.ts @@ -2,7 +2,7 @@ import type { ServerResponse } from "node:http"; import type { TokenUsage } from "../usage/types.js"; import type { ProviderProtocol } from "../providers/types.js"; import type { JsonRecord } from "../json.js"; -import { recCount, recObj, recObjs, recStr } from "../json.js"; +import { recObj, recObjs, recStr } from "../json.js"; import { mapAnthropicUsage, mapChatUsage, mapResponsesUsage } from "../providers/usage/index.js"; export const SSE_HEADERS = { "content-type": "text/event-stream", "cache-control": "no-cache", connection: "keep-alive" }; @@ -121,42 +121,12 @@ export async function drain(reader: ReadableStreamDefaultReader, con } } -/** - * Cache-token fields shared by chat-completions, Responses, and Anthropic - * usage payloads. A stream pipe doesn't always know in advance which shape - * `usage` will turn out to be (pipeResponsesStream in to-anthropic.ts serves - * whichever of the two non-Anthropic upstream protocols is actually - * configured, and can even be called with no options at all — see its - * message_delta construction), so this tries every protocol's field mapper - * and keeps the first hit; each mapper's own field list is the single source - * of truth (providers/usage/*.ts), not duplicated here. - */ -export function cacheTokensOf(usage: JsonRecord | undefined): { cached?: number; reasoning?: number } { - const chat = mapChatUsage(usage, {}); - const responses = mapResponsesUsage(usage, {}); - const anthropic = mapAnthropicUsage(usage, {}); - const cached = chat?.cachedInputTokens ?? responses?.cachedInputTokens ?? anthropic?.cachedInputTokens ?? recCount(usage, "cached_tokens"); - const reasoning = chat?.reasoningTokens ?? responses?.reasoningTokens; - return { - ...(cached === undefined ? {} : { cached }), - ...(reasoning === undefined ? {} : { reasoning }), - }; -} - -/** Attach captured cache/reasoning tokens to a usage record when present. */ -export function withCacheTokens(usage: TokenUsage, source: JsonRecord | undefined): TokenUsage { - const { cached, reasoning } = cacheTokensOf(source); - if (cached !== undefined) usage.cachedInputTokens = cached; - if (reasoning !== undefined) usage.reasoningTokens = reasoning; - return usage; -} - /** Per-request token counters every pipe tracks; the single store pipes read when emitting and reporting usage. */ export interface UsageCapture { input: number; output: number; - /** The most recent raw upstream usage payload, for protocol-shape-dependent emission (e.g. Anthropic cache_read). */ - raw?: JsonRecord; + /** Upstream total tokens, mapped from the latest raw usage by the provider field mappers. */ + total?: number; sawUsage: boolean; /** Cache/reasoning fields mapped from the latest raw usage via the provider field mappers. */ cached?: number; @@ -169,9 +139,10 @@ export function newUsageCapture(): UsageCapture { } function mapRawUsage(raw: JsonRecord | undefined, protocol: ProviderProtocol): TokenUsage | null { + // The bare-usage field lists — including each protocol's total-token field — + // live in providers/usage/*; the core never parses provider payloads + // itself (CLAUDE.md architecture rule). switch (protocol) { - // Bare-usage field lists live in providers/usage/*; the core never parses - // provider payloads itself (CLAUDE.md architecture rule). case "anthropic": return mapAnthropicUsage(raw, {}); case "responses": return mapResponsesUsage(raw, {}); case "chat-completions": return mapChatUsage(raw, {}); @@ -182,9 +153,8 @@ function mapRawUsage(raw: JsonRecord | undefined, protocol: ProviderProtocol): T * Bind usage capture to the upstream protocol so token fields are read via * the same provider field mappers as the non-streaming path. `usage()` merges * a raw usage object into the capture; `total()` prefers the upstream's own - * total when it carries one. `enabled=false` (a pipe called with no options) - * keeps raw payloads as estimation hints without claiming real usage, which - * `reportUsage`'s fallback then reports. + * total as mapped by that protocol's field mapper. `enabled=false` (a pipe + * called with no options) captures nothing — there is no client to report to. */ export function usageCapture(capture: UsageCapture, protocol: ProviderProtocol, enabled = true): { usage(raw: JsonRecord | undefined): void; @@ -192,19 +162,19 @@ export function usageCapture(capture: UsageCapture, protocol: ProviderProtocol, } { return { usage(raw) { + if (!enabled) return; const mapped = mapRawUsage(raw, protocol); if (!mapped) return; - capture.raw = raw; - if (!enabled) return; capture.sawUsage = true; if (mapped.inputTokens) capture.input = mapped.inputTokens; if (mapped.outputTokens) capture.output = mapped.outputTokens; + if (mapped.totalTokens) capture.total = mapped.totalTokens; if (mapped.cachedInputTokens !== undefined) capture.cached = mapped.cachedInputTokens; if (mapped.reasoningTokens !== undefined) capture.reasoning = mapped.reasoningTokens; if (mapped.cacheWriteTokens !== undefined) capture.cacheWrite = mapped.cacheWriteTokens; }, total() { - return recCount(capture.raw, "total_tokens") ?? capture.input + capture.output; + return capture.total ?? capture.input + capture.output; }, }; } @@ -307,8 +277,7 @@ export function functionCallItem(call: { id: string; name: string; arguments: st /** * Finalize the turn for Responses-facing clients: text/reasoning done events, * per-call argument-done + item-done events, then the completed response — - * Codex strictly requires total_tokens there. Chat-style `calls` carry no - * `arguments`, so the caller supplies them via `argsOf`. + * Codex strictly requires total_tokens there. */ export function emitResponsesCompleted( response: ServerResponse, @@ -321,7 +290,6 @@ export function emitResponsesCompleted( msgId?: string; reasoning?: ReasoningState; calls: Iterable<[number | string, { id: string; name: string; arguments?: string }]>; - argsOf?(index: number | string): string; }, ) { const { id, model, capture, truncated } = args; @@ -337,7 +305,7 @@ export function emitResponsesCompleted( if (args.reasoning?.id) output.push(reasoningItem(args.reasoning)); if (args.text) output.push(messageItem(args.msgId ?? "", args.text)); for (const [index, call] of args.calls) { - const callArgs = call.arguments ?? args.argsOf?.(index) ?? ""; + const callArgs = call.arguments ?? ""; event(response, "response.function_call_arguments.done", { type: "response.function_call_arguments.done", item_id: `fc_${index}`, call_id: call.id, arguments: callArgs }); event(response, "response.output_item.done", { type: "response.output_item.done", output_index: index, item: functionCallItem({ ...call, arguments: callArgs }, index) }); output.push(functionCallItem({ ...call, arguments: callArgs }, index)); diff --git a/src/streaming/to-anthropic.ts b/src/streaming/to-anthropic.ts index b6547a4..94aead3 100644 --- a/src/streaming/to-anthropic.ts +++ b/src/streaming/to-anthropic.ts @@ -2,7 +2,7 @@ import type { ServerResponse } from "node:http"; import { chatResponseFailure } from "../convert/index.js"; import { jsonRecord, recNum, recObj, recObjs, recStr } from "../json.js"; import { - cacheTokensOf, dataLine, drain, emitAnthropicError, event, failureMessage, newUsageCapture, reasoningDeltaOf, + dataLine, drain, emitAnthropicError, event, failureMessage, newUsageCapture, reasoningDeltaOf, reportUsage, usageCapture, withSsePipe, UpstreamFailure, type StreamUsageOptions, } from "./common.js"; @@ -126,8 +126,7 @@ export async function pipeResponsesStream(upstream: Response, response: ServerRe if (!blockStarted) startText(); stopBlock(); const stopReason = truncated ? "max_tokens" : toolStop ? "tool_use" : "end_turn"; - const { cached } = cacheTokensOf(capture.raw); - event(response, "message_delta", { type: "message_delta", delta: { stop_reason: stopReason, stop_sequence: null }, usage: { output_tokens: capture.output, input_tokens: capture.input, cache_creation_input_tokens: 0, cache_read_input_tokens: cached ?? 0 } }); + event(response, "message_delta", { type: "message_delta", delta: { stop_reason: stopReason, stop_sequence: null }, usage: { output_tokens: capture.output, input_tokens: capture.input, cache_creation_input_tokens: 0, cache_read_input_tokens: capture.cached ?? 0 } }); event(response, "message_stop", { type: "message_stop" }); } catch (error) { // Match Anthropic semantics: a terminal error event closes the stream. From daf933c3f0749f4090415a4a9e0154c02c341d90 Mon Sep 17 00:00:00 2001 From: tzzs Date: Mon, 21 Sep 2026 00:37:05 +0800 Subject: [PATCH 2/4] fix(convert): keep tool-result images usable on Anthropic upstreams MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit toAnthropicRequest and toAnthropicRequestFromChat still JSON.stringify'd tool-output block arrays — the exact ~1.37M-character screenshot mistake toolResultContent() documents for the other direction. A new toolResultBlocks() maps Chat/Responses text and image parts back to Anthropic tool_result blocks. The count_tokens token estimator moves from server.ts into convert/shared.ts alongside the payload logic it reads. --- src/convert/anthropic.ts | 12 +++---- src/convert/shared.ts | 68 +++++++++++++++++++++++++++++++++++ test/server-endpoints.test.ts | 3 +- test/tools.test.ts | 34 +++++++++++++++++- 4 files changed, 109 insertions(+), 8 deletions(-) diff --git a/src/convert/anthropic.ts b/src/convert/anthropic.ts index 9ecda0e..0fc424f 100644 --- a/src/convert/anthropic.ts +++ b/src/convert/anthropic.ts @@ -6,8 +6,8 @@ */ import type { AnthropicMessage, AnthropicRequest, AnthropicThinking } from "./shared.js"; import type { JsonRecord, JsonValue } from "../json.js"; -import { asRecords, isRecord, parse, recNum, recObj, recObjs, recStr } from "../json.js"; -import { anthropicThinking, anthropicToolChoice, collapseAnthropicContent, textOfBlocks, toAnthropicImageSource } from "./shared.js"; +import { asRecords, isRecord, parseJson, recNum, recObj, recObjs, recStr } from "../json.js"; +import { anthropicThinking, anthropicToolChoice, collapseAnthropicContent, textOfBlocks, toAnthropicImageSource, toolResultBlocks } from "./shared.js"; /** Responses message content (string or input_text/input_image parts) to Anthropic message content. */ function toAnthropicContent(content: JsonValue): JsonValue { @@ -44,10 +44,10 @@ export function toAnthropicRequest(input: JsonRecord, model: string): AnthropicR const flushAssistant = () => { if (pendingAssistantBlocks.length) { messages.push({ role: "assistant", content: pendingAssistantBlocks }); pendingAssistantBlocks = []; } }; for (const item of items) { if (item.type === "reasoning") continue; // no signature to echo upstream; dropped like other local-only reasoning echoes - if (item.type === "function_call") { pendingAssistantBlocks.push({ type: "tool_use", id: recStr(item, "call_id") ?? recStr(item, "id"), name: recStr(item, "name"), input: parse(item.arguments) }); continue; } + if (item.type === "function_call") { pendingAssistantBlocks.push({ type: "tool_use", id: recStr(item, "call_id") ?? recStr(item, "id"), name: recStr(item, "name"), input: parseJson(item.arguments) }); continue; } if (item.type === "function_call_output") { flushAssistant(); - messages.push({ role: "user", content: [{ type: "tool_result", tool_use_id: item.call_id, content: typeof item.output === "string" ? item.output : JSON.stringify(item.output ?? "") }] }); + messages.push({ role: "user", content: [{ type: "tool_result", tool_use_id: item.call_id, content: toolResultBlocks(item.output) }] }); continue; } if (item.role === "assistant") { const text = responsesItemText(item.content); if (text) pendingAssistantBlocks.push({ type: "text", text }); continue; } @@ -160,7 +160,7 @@ export function toAnthropicRequestFromChat(input: JsonRecord, model: string): An const role = recStr(message, "role"); if (role === "system") { const content = recStr(message, "content"); if (content) systemParts.push(content); continue; } if (role === "tool") { - messages.push({ role: "user", content: [{ type: "tool_result", tool_use_id: message.tool_call_id, content: typeof message.content === "string" ? message.content : JSON.stringify(message.content ?? "") }] }); + messages.push({ role: "user", content: [{ type: "tool_result", tool_use_id: message.tool_call_id, content: toolResultBlocks(message.content) }] }); continue; } if (typeof role !== "string") continue; @@ -168,7 +168,7 @@ export function toAnthropicRequestFromChat(input: JsonRecord, model: string): An const converted = chatContentToAnthropic(message.content); if (Array.isArray(converted)) blocks.push(...asRecords(converted)); else if (converted) blocks.push({ type: "text", text: converted }); - for (const call of recObjs(message, "tool_calls")) blocks.push({ type: "tool_use", id: recStr(call, "id"), name: recStr(recObj(call, "function"), "name"), input: parse(recStr(recObj(call, "function"), "arguments")) }); + for (const call of recObjs(message, "tool_calls")) blocks.push({ type: "tool_use", id: recStr(call, "id"), name: recStr(recObj(call, "function"), "name"), input: parseJson(recStr(recObj(call, "function"), "arguments")) }); if (blocks.length) messages.push({ role, content: blocks }); } const rawStop = input.stop; diff --git a/src/convert/shared.ts b/src/convert/shared.ts index f2a9a2a..fd9eaa1 100644 --- a/src/convert/shared.ts +++ b/src/convert/shared.ts @@ -263,3 +263,71 @@ export function toolResultText(text: string, imageCount: number, forwarded: bool export function acceptsImageInput(provider?: { modalities?: string[] }): boolean { return provider?.modalities === undefined || provider.modalities.includes("image"); } + +/** + * Anthropic `tool_result` content built from a Chat Completions or Responses + * tool-output value — the inbound counterpart of `toolResultContent()`. Both + * dialects can carry images in a tool output (`image_url` / `input_image` + * parts), and Anthropic accepts image blocks inside a tool_result directly, + * so they are mapped to blocks instead of stringified into the text — the + * exact mistake `toolResultContent()` documents for the other direction. + * Text-only results collapse back to a plain string. + */ +export function toolResultBlocks(value: unknown): JsonValue { + if (typeof value === "string") return value; + if (value === undefined || value === null) return ""; + if (!Array.isArray(value)) return JSON.stringify(value); + const parts: JsonRecord[] = []; + for (const raw of value as JsonValue[]) { + if (isRecord(raw)) { + if (raw.type === "text" || raw.type === "input_text" || raw.type === "output_text") { + const text = recStr(raw, "text"); + if (text !== undefined) parts.push({ type: "text", text }); + continue; + } + const url = recStr(raw, "image_url") ?? recStr(recObj(raw, "image_url"), "url"); + if ((raw.type === "input_image" || raw.type === "image_url") && url) { + const source = toAnthropicImageSource(url); + if (source) parts.push({ type: "image", source }); + continue; + } + } + // Anything Anthropic does not accept as a tool_result block is kept as + // text rather than dropped, so an unknown part still reaches the model. + if (raw !== undefined && raw !== null) parts.push({ type: "text", text: typeof raw === "string" ? raw : JSON.stringify(raw) }); + } + // Text-only results collapse to a string, newline-joined like + // `toolResultContent()` does for the other direction. + if (parts.length && parts.every((part) => part.type === "text")) return parts.map((part) => recStr(part, "text") ?? "").join("\n"); + return collapseAnthropicContent(parts); +} + +/** + * Flat per-image token allowance for the estimator below. Anthropic bills an + * image by its pixel area, which a request body does not carry, so a single + * mid-sized-screenshot figure stands in for it — far closer than counting the + * base64 payload as if it were prose. + */ +const IMAGE_TOKEN_ESTIMATE = 1_600; + +/** + * Approximate the input tokens of an Anthropic Messages request. Used by + * `/v1/messages/count_tokens` when the upstream protocol has no equivalent + * endpoint to ask. Four characters per token is the conventional English + * approximation; this is explicitly an estimate, but one made from the actual + * request, unlike the blind fallback a 404 used to force on the client. + */ +export function estimateInputTokens(input: JsonRecord): number { + let chars = 0; + const walk = (value: unknown): void => { + if (typeof value === "string") { chars += value.length; return; } + if (Array.isArray(value)) { value.forEach(walk); return; } + if (!isRecord(value)) return; + // An image block's base64 payload is not prose; charge it a flat rate + // instead of walking into the data string. + if (value.type === "image") { chars += IMAGE_TOKEN_ESTIMATE * 4; return; } + Object.values(value).forEach(walk); + }; + walk(input.system); walk(input.messages); walk(input.tools); + return Math.max(1, Math.ceil(chars / 4)); +} diff --git a/test/server-endpoints.test.ts b/test/server-endpoints.test.ts index f87ca8e..a4d0bd5 100644 --- a/test/server-endpoints.test.ts +++ b/test/server-endpoints.test.ts @@ -1,6 +1,7 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { createConcurrencyGate, estimateInputTokens, startAdapter } from "../src/server.js"; +import { createConcurrencyGate, startAdapter } from "../src/server.js"; +import { estimateInputTokens } from "../src/convert/shared.js"; import { createUsageStore } from "../src/usage/storage.js"; import { registerCustomProvider, unregisterCustomProvider } from "../src/providers/registry.js"; import type { Config } from "../src/config.js"; diff --git a/test/tools.test.ts b/test/tools.test.ts index 4695c26..d6f8d39 100644 --- a/test/tools.test.ts +++ b/test/tools.test.ts @@ -1,6 +1,6 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { fromChatResponse, fromResponsesResponse, toChatRequest, toResponsesRequest } from "../src/convert/index.js"; +import { fromChatResponse, fromResponsesResponse, toAnthropicRequest, toAnthropicRequestFromChat, toChatRequest, toResponsesRequest } from "../src/convert/index.js"; test("converts tools and tool results", () => { const result = toResponsesRequest({ tools: [{ name: "bash", description: "Run a command", input_schema: { type: "object" } }], messages: [{ role: "assistant", content: [{ type: "tool_use", id: "call-1", name: "bash", input: { command: "pwd" } }] }, { role: "user", content: [{ type: "tool_result", tool_use_id: "call-1", content: "/tmp" }] }] }, "gpt-5.6-luna"); @@ -97,6 +97,38 @@ test("flattens a text-only tool_result block array instead of forwarding its JSO assert.equal((toChatRequest(blocks as any, "glm-5") as any).messages[1].content, "line one\nline two"); }); +// Anthropic accepts image blocks inside a tool_result directly, so the two +// inbound directions must map them back to blocks instead of stringifying the +// part array (which entered the context as the base64 rendered as text). + +test("restores tool-result images as Anthropic tool_result blocks from Responses output", () => { + const result = toAnthropicRequest({ input: [ + { type: "function_call", call_id: "call-1", name: "read", arguments: "{}" }, + { type: "function_call_output", call_id: "call-1", output: [{ type: "input_text", text: "read ok" }, { type: "input_image", image_url: "data:image/png;base64,AAA" }] }, + ] }, "claude-y") as any; + assert.deepEqual(result.messages[1].content[0], { + type: "tool_result", tool_use_id: "call-1", + content: [{ type: "text", text: "read ok" }, { type: "image", source: { type: "base64", media_type: "image/png", data: "AAA" } }], + }); +}); + +test("restores tool-result images as Anthropic tool_result blocks from chat content parts", () => { + const result = toAnthropicRequestFromChat({ messages: [ + { role: "tool", tool_call_id: "call-1", content: [{ type: "text", text: "ok" }, { type: "image_url", image_url: { url: "https://x.invalid/a.png" } }] }, + ] }, "claude-y") as any; + assert.deepEqual(result.messages[0].content[0], { + type: "tool_result", tool_use_id: "call-1", + content: [{ type: "text", text: "ok" }, { type: "image", source: { type: "url", url: "https://x.invalid/a.png" } }], + }); +}); + +test("keeps a text-only tool output a plain string for Anthropic upstreams", () => { + const result = toAnthropicRequest({ input: [ + { type: "function_call_output", call_id: "call-1", output: [{ type: "input_text", text: "line one" }, { type: "input_text", text: "line two" }] }, + ] }, "claude-y") as any; + assert.equal(result.messages[0].content[0].content, "line one\nline two"); +}); + test("maps an unrecognized finish_reason to end_turn rather than max_tokens", () => { const ended = fromChatResponse({ choices: [{ message: { content: "hi" }, finish_reason: "something_new" }] }, "glm-5") as any; assert.equal(ended.stop_reason, "end_turn"); From e6b82753fe2350d0884149e6940a276e2335abb8 Mon Sep 17 00:00:00 2001 From: tzzs Date: Mon, 21 Sep 2026 00:37:07 +0800 Subject: [PATCH 3/4] refactor(server): pick endpoint translators from protocol maps The per-endpoint finish/failure ternary cascades become protocol-keyed lookup tables (failures shared via one map), the resolved spec reads as endpointSpec, and json.ts's parse accessor is renamed parseJson to match its call sites. Docs and comments (retry status set, new endpoints, concurrency gate) are synced with the shipped behavior. --- CLAUDE.md | 4 +- docs/architecture.md | 14 +++--- src/convert/chat.ts | 4 +- src/convert/responses.ts | 4 +- src/json.ts | 2 +- src/server.ts | 96 +++++++++++++++++----------------------- test/json.test.ts | 12 ++--- 7 files changed, 60 insertions(+), 76 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7ed3f01..0520cd2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -25,7 +25,7 @@ npm run typecheck # tsc --noEmit，同时检查 src/ 与 test/ 本包声明了 `"type": "module"` 并使用 `rootDir: "."` 编译，因此**所有 import 都使用 `.js` 后缀，且所有路径以仓库根目录为基准**（例如 `import { loadConfig } from "./config.js"`，测试位于 `test/` 下，导入如 `"../src/catalog.js"`）。 -`src/` 中不使用 `any`：客户端请求、Provider 响应与 SSE 事件这类无法预先验证的 JSON 一律建模为 `src/json.ts` 的 `JsonRecord`，通过 `recStr`/`recNum`/`recCount`/`recObj`/`recObjs`/`parse` 读取。字段名写错或漏掉非对象检查会直接编译失败，而不是变成一次运行时 undefined。tsconfig 开启了 `noUncheckedIndexedAccess` 等严格标志，ESLint 只保留 tsc 看不见的规则（见 `eslint.config.js`）。 +`src/` 中不使用 `any`：客户端请求、Provider 响应与 SSE 事件这类无法预先验证的 JSON 一律建模为 `src/json.ts` 的 `JsonRecord`，通过 `recStr`/`recNum`/`recCount`/`recObj`/`recObjs`/`parseJson` 读取。字段名写错或漏掉非对象检查会直接编译失败，而不是变成一次运行时 undefined。tsconfig 开启了 `noUncheckedIndexedAccess` 等严格标志，ESLint 只保留 tsc 看不见的规则（见 `eslint.config.js`）。 ### 两套协议转换路径 @@ -38,7 +38,7 @@ npm run typecheck # tsc --noEmit，同时检查 src/ 与 test/ 转换函数集中在 `src/convert/`，按**上游协议**（而非转换方向）拆分为四个文件： -- `shared.ts` — 跨方向 helper：图片/effort/thinking/三套 tool-choice 转换、采样参数（JSON 字段的读取统一走 `src/json.ts`） +- `shared.ts` — 跨方向 helper：图片/effort/thinking/三套 tool-choice 转换、采样参数、tool_result 媒体的双向拆分/回填、`count_tokens` 的本地 token 估算（JSON 字段的读取统一走 `src/json.ts`） - `chat.ts` — 上游 = Chat Completions 的全部方向：`toChatRequest` / `fromChatResponse`（Anthropic ↔ Chat Completions）、`toChatCompletionsRequest` / `fromChatResponseToResponses`（Responses ↔ Chat Completions） - `responses.ts` — 上游 = Responses：`toResponsesRequest` / `fromResponsesResponse`（Anthropic ↔ Responses）；`toResponsesRequestFromChat` / `fromResponsesResponseToChat`（本地 Chat Completions ↔ Responses；分别复用 `toResponsesRequest`/`fromResponsesResponse` 加 `anthropic.ts` 的两个新函数拼出，不重复写转换逻辑） - `anthropic.ts` — 上游 = Anthropic（自定义 Provider 声明 `protocol: "anthropic"` 时专属）：`toAnthropicRequest` / `fromAnthropicResponse`（Responses ↔ Anthropic；Codex 只会看到本地 Responses 端点，仍需经此转换才能到达一个原生 Anthropic 上游）；`toAnthropicRequestFromChat` / `fromAnthropicResponseToChat`（本地 Chat Completions ↔ Anthropic，供 `/v1/chat/completions` 使用，只映射主流字段，不映射 DeepSeek 的 `thinking`/`reasoning_effort` 扩展） diff --git a/docs/architecture.md b/docs/architecture.md index 8d5b318..7c17b6c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -35,8 +35,8 @@ cli.ts (claude/codex/proxy/exec) │ ├─ server.ts startAdapter() → 生成本地随机 token，绑定回环端口 │ ├─ /health - │ ├─ /v1/models - │ ├─ /v1/messages (Anthropic) + │ ├─ /v1/models 与 /v1/models/{id} + │ ├─ /v1/messages (Anthropic；含 POST /v1/messages/count_tokens) │ ├─ /v1/responses (OpenAI Responses) │ └─ /v1/chat/completions (OpenAI Chat Completions) │ @@ -54,7 +54,7 @@ cli.ts (claude/codex/proxy/exec) Claude Code (Anthropic Messages API) │ POST /v1/messages Bearer <本地随机 token> ▼ - server.ts (校验本地 token，不转发给上游；按 config.retry 对 429/502/503/504 做指数退避重试) + server.ts (校验本地 token，不转发给上游；按 config.retry 对 408/429/502/503/504 做指数退避重试；`--max-concurrency` 限制在途上游请求数) │ ├─ catalog.ts providerFor（按 model 选路由；纯路由模块，不含转换函数） │ @@ -73,9 +73,9 @@ Claude Code (Anthropic Messages API) └─ 流式: streaming/anthropic-passthrough.ts, streaming/anthropic-to-responses.ts ``` -`src/convert/` 是转换函数所在的模块化目录：`shared.ts`（跨方向 helper：图片/effort/thinking/tool-choice 映射、采样参数）、`chat.ts`（上游 = Chat Completions 的全部方向）、`responses.ts`（上游 = Responses）、`anthropic.ts`（上游 = Anthropic，自定义 Provider 专属）、`index.ts`（barrel 导出，公共函数名不变）。`src/catalog.ts` 只保留 `providers`/`providerFor`/`honorRequestedModel` 路由函数。 +`src/convert/` 是转换函数所在的模块化目录：`shared.ts`（跨方向 helper：图片/effort/thinking/tool-choice 映射、采样参数、tool_result 媒体的双向拆分/回填、`count_tokens` 的本地 token 估算）、`chat.ts`（上游 = Chat Completions 的全部方向）、`responses.ts`（上游 = Responses）、`anthropic.ts`（上游 = Anthropic，自定义 Provider 专属）、`index.ts`（barrel 导出，公共函数名不变）。`src/catalog.ts` 只保留 `providers`/`providerFor`/`honorRequestedModel` 路由函数。 -所有 wire payload（客户端请求、Provider 响应、SSE 事件）经 `src/json.ts` 的 `JsonRecord` + 字段访问器（`recStr`/`recNum`/`recCount`/`recObj`/`recObjs`/`parse`）读取，代码库里不使用 `any`：字段名写错或漏掉非对象检查由编译器拦下，而不是留成一次 undefined 运行时错误。 +所有 wire payload（客户端请求、Provider 响应、SSE 事件）经 `src/json.ts` 的 `JsonRecord` + 字段访问器（`recStr`/`recNum`/`recCount`/`recObj`/`recObjs`/`parseJson`）读取，代码库里不使用 `any`：字段名写错或漏掉非对象检查由编译器拦下，而不是留成一次 undefined 运行时错误。 `src/streaming/` 同样是拆分后的模块化目录（`common.ts` 收敛公共 SSE 写入/heartbeat/usage capture 逻辑，每条协议转换路径各占一个文件），不是单一的 `streaming.ts`。 @@ -108,7 +108,7 @@ Claude Code (Anthropic Messages API) - **凭据隔离**：真实 Key 只留在 Adapter 内，客户端只拿到每次启动随机生成的本地 token - **工具调用**：只做协议转换（`tool_use`↔`function_call`、`tool_result`↔`function_call_output`），不在 Adapter 内执行 - **模型路由**：没有隐式 `auto` 路由（已移除）；运行时必须解析为具体模型，解析优先级见 README「Configuration」 -- **上游重试**：`server.ts` 的 `forwardWithRetry` 对网络失败和 429/502/503/504 做指数退避重试（`--retry`/`AGENTX_RETRY`，默认 3，0 禁用），仅发生在流式传输开始之前 +- **上游重试**：`server.ts` 的 `forwardWithRetry` 对网络失败和 408/429/502/503/504 做指数退避重试（`--retry`/`AGENTX_RETRY`，默认 3，0 禁用），仅发生在流式传输开始之前 - **命令覆盖**：`claude`/`codex`/`proxy`/`exec`/`auth`/`usage`/`quota`/`doctor`/`forget`/`version`/`help` - **可选集成**：`agentx quota --provider ` 查询 DeepSeek/OpenRouter 额度（OpenCode 返回"不支持"；`usage --provider` 为过渡期别名）；凭据只来自环境变量（`AGENTX__API_KEY`，旧的无前缀变量兼容） @@ -118,7 +118,7 @@ Claude Code (Anthropic Messages API) |---|---| | `src/cli.ts` | 命令分发、参数解析、编排（`runAuthCommand`/`runDoctorCommand`/`runClientLaunch` 等具名函数） | | `src/config.ts` | 配置加载与优先级 | -| `src/server.ts` | HTTP Adapter、认证、路由、重试、端口回退 | +| `src/server.ts` | HTTP Adapter、认证、路由、重试、并发闸（`--max-concurrency`）、端口回退 | | `src/catalog.ts` | 模型路由（`providers`/`providerFor`/`honorRequestedModel`），不含转换函数 | | `src/convert/` | 协议转换函数，按上游协议拆分：`shared.ts`（跨方向 helper）、`chat.ts`（上游 = Chat Completions）、`responses.ts`（上游 = Responses）、`anthropic.ts`（上游 = Anthropic，自定义 Provider 专属）、`index.ts`（barrel） | | `src/json.ts` | wire payload 的类型词汇表：`JsonRecord` 与字段访问器，替代 `any` | diff --git a/src/convert/chat.ts b/src/convert/chat.ts index b9e2dc7..c0953c7 100644 --- a/src/convert/chat.ts +++ b/src/convert/chat.ts @@ -5,7 +5,7 @@ */ import type { AnthropicMessage, AnthropicRequest } from "./shared.js"; import type { JsonRecord, JsonValue } from "../json.js"; -import { isRecord, parse, recNum, recObj, recObjs, recStr } from "../json.js"; +import { isRecord, parseJson, recNum, recObj, recObjs, recStr } from "../json.js"; import { acceptsImageInput, chatControlParams, collapseAnthropicContent, imageDataUri, samplingParams, toolResultContent, toolResultText, TOOL_RESULT_MEDIA_PROMPT } from "./shared.js"; import type { ProviderModel } from "../providers/types.js"; import { isDeepSeekLongContextModel } from "../providers/registry.js"; @@ -112,7 +112,7 @@ export function fromChatResponse(response: JsonRecord, model: string): JsonRecor const text = chatText(message.content); if (text) content.push({ type: "text", text }); const toolCalls = recObjs(message, "tool_calls"); - for (const call of toolCalls) content.push({ type: "tool_use", id: recStr(call, "id"), name: recStr(recObj(call, "function"), "name"), input: parse(recStr(recObj(call, "function"), "arguments")) }); + for (const call of toolCalls) content.push({ type: "tool_use", id: recStr(call, "id"), name: recStr(recObj(call, "function"), "name"), input: parseJson(recStr(recObj(call, "function"), "arguments")) }); const usage = recObj(response, "usage"); return { id: recStr(response, "id") ?? `msg_${crypto.randomUUID()}`, type: "message", role: "assistant", model, content, stop_reason: anthropicStopReason(choice.finish_reason, toolCalls.length > 0), stop_sequence: null, usage: { input_tokens: recNum(usage, "prompt_tokens") ?? 0, output_tokens: recNum(usage, "completion_tokens") ?? 0, cache_creation_input_tokens: 0, cache_read_input_tokens: chatUsageDetails(response, "usage", "prompt_tokens_details", "cached_tokens") ?? 0 } }; } diff --git a/src/convert/responses.ts b/src/convert/responses.ts index 5ab37fc..88d8068 100644 --- a/src/convert/responses.ts +++ b/src/convert/responses.ts @@ -1,7 +1,7 @@ /** Anthropic Messages API <-> Responses API: the direction whose upstream speaks the Responses protocol. */ import type { AnthropicMessage, AnthropicRequest, ResponsesItem } from "./shared.js"; import type { JsonRecord, JsonValue } from "../json.js"; -import { asRecords, isRecord, parse, recNum, recObj, recObjs, recStr } from "../json.js"; +import { asRecords, isRecord, parseJson, recNum, recObj, recObjs, recStr } from "../json.js"; import { acceptsImageInput, chatThinking, imageDataUri, reasoningEffort, responsesToolChoice, toolResultContent, toolResultText } from "./shared.js"; import type { ProviderModel } from "../providers/types.js"; import { fromAnthropicResponseToChat, toAnthropicRequestFromChat } from "./anthropic.js"; @@ -101,7 +101,7 @@ export function fromResponsesResponse(response: JsonRecord, model: string): Json .flatMap((item) => asRecords(item.content)) .filter((part) => part.type === "output_text") .map((part) => recStr(part, "text") ?? "").join(""); - const toolUses = output.filter((item) => item.type === "function_call").map((item) => ({ type: "tool_use", id: recStr(item, "call_id") ?? recStr(item, "id"), name: recStr(item, "name"), input: parse(recStr(item, "arguments")) })); + const toolUses = output.filter((item) => item.type === "function_call").map((item) => ({ type: "tool_use", id: recStr(item, "call_id") ?? recStr(item, "id"), name: recStr(item, "name"), input: parseJson(recStr(item, "arguments")) })); const thinking = reasoningText(output); const usage = recObj(response, "usage"); const cached = recNum(recObj(usage, "input_tokens_details"), "cached_tokens"); diff --git a/src/json.ts b/src/json.ts index ce9877e..c8fa056 100644 --- a/src/json.ts +++ b/src/json.ts @@ -56,6 +56,6 @@ export function recCount(obj: JsonRecord | undefined, key: string): number | und } /** Best-effort JSON parse of a wire payload; malformed or absent input becomes `{}` rather than throwing. */ -export function parse(value: unknown): JsonValue { +export function parseJson(value: unknown): JsonValue { try { return (typeof value === "string" ? JSON.parse(value) : value ?? {}) as JsonValue; } catch { return {}; } } diff --git a/src/server.ts b/src/server.ts index 920ebe6..9cc77ad 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,9 +1,9 @@ import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; import { randomBytes, randomUUID, timingSafeEqual } from "node:crypto"; import type { Config } from "./config.js"; -import { asAnthropicRequest, chatResponseFailure, fromAnthropicResponse, fromAnthropicResponseToChat, fromChatResponse, fromChatResponseToResponses, fromResponsesResponse, fromResponsesResponseToChat, responsesResponseFailure, toAnthropicRequest, toAnthropicRequestFromChat, toChatCompletionsRequest, toChatRequest, toResponsesRequest, toResponsesRequestFromChat } from "./convert/index.js"; +import { asAnthropicRequest, chatResponseFailure, estimateInputTokens, fromAnthropicResponse, fromAnthropicResponseToChat, fromChatResponse, fromChatResponseToResponses, fromResponsesResponse, fromResponsesResponseToChat, responsesResponseFailure, toAnthropicRequest, toAnthropicRequestFromChat, toChatCompletionsRequest, toChatRequest, toResponsesRequest, toResponsesRequestFromChat } from "./convert/index.js"; import type { JsonRecord } from "./json.js"; -import { isRecord, jsonRecord, recNum, recStr } from "./json.js"; +import { jsonRecord, recNum, recStr } from "./json.js"; import { pipeAnthropicPassthrough, pipeAnthropicStreamToChat, pipeAnthropicStreamToResponses, pipeChatPassthrough, pipeChatStreamToResponses, pipeResponsesPassthrough, pipeResponsesStream, pipeResponsesStreamToChat, type StreamUsageOptions } from "./streaming/index.js"; import type { ProviderModel, ProviderProtocol } from "./providers/types.js"; import type { TokenUsage } from "./usage/types.js"; @@ -126,7 +126,7 @@ function retryDelayMs(attempt: number, retryAfterHeader: string | null): number /** * Wrap `forward()` with bounded retry for transient upstream failures: a * network-level failure (surfaced by `forward()` as a status:502 Error) or a - * 429/502/503/504 response. Other statuses — including other 4xx — return + * 408/429/502/503/504 response. Other statuses — including other 4xx — return * immediately; retrying a client/config error wastes time without changing * the outcome. This must be the only place that calls `forward()`: retries * only ever happen before any response body reaches the client (this @@ -156,36 +156,6 @@ export async function forwardWithRetry( throw new Error("forwardWithRetry: unreachable"); } -/** - * Flat per-image token allowance for the estimator below. Anthropic bills an - * image by its pixel area, which a request body does not carry, so a single - * mid-sized-screenshot figure stands in for it — far closer than counting the - * base64 payload as if it were prose. - */ -const IMAGE_TOKEN_ESTIMATE = 1_600; - -/** - * Approximate the input tokens of an Anthropic Messages request. Used by - * `/v1/messages/count_tokens` when the upstream protocol has no equivalent - * endpoint to ask. Four characters per token is the conventional English - * approximation; this is explicitly an estimate, but one made from the actual - * request, unlike the blind fallback a 404 used to force on the client. - */ -export function estimateInputTokens(input: JsonRecord): number { - let chars = 0; - const walk = (value: unknown): void => { - if (typeof value === "string") { chars += value.length; return; } - if (Array.isArray(value)) { value.forEach(walk); return; } - if (!isRecord(value)) return; - // An image block's base64 payload is not prose; charge it a flat rate - // instead of walking into the data string. - if (value.type === "image") { chars += IMAGE_TOKEN_ESTIMATE * 4; return; } - Object.values(value).forEach(walk); - }; - walk(input.system); walk(input.messages); walk(input.tools); - return Math.max(1, Math.ceil(chars / 4)); -} - /** * Ask a native Anthropic upstream for an exact count. Any failure (endpoint * absent, network, malformed body) resolves to `undefined` so the caller @@ -300,9 +270,9 @@ async function proxyRequest( } /** - * Per client-facing endpoint: how to build the upstream payload for each - * provider protocol, which pipe serves each streaming combination, and how - * to translate a non-streaming upstream JSON back to this endpoint's format. + * Per client-facing endpoint: for each provider protocol, how to build the + * upstream payload, which pipe serves the stream, and how to translate a + * non-streaming upstream JSON back to this endpoint's format. * Invariant: the pipe/translator selected always matches the protocol whose * payload builder ran (e.g. a 200 with no body on an Anthropic upstream goes * to the Anthropic pipe, never the Responses one). @@ -310,12 +280,21 @@ async function proxyRequest( interface EndpointSpec { payloads: Record unknown>; pipes: Record Promise>; - /** Non-stream JSON translation; an upstream-reported failure maps to a 502 instead. */ - finish(upstream: JsonRecord, provider: ProviderModel, model: string): unknown; - /** In-band failure message carried by a non-OK-shaped 200 payload, if any. */ - failure?(upstream: JsonRecord, provider: ProviderModel): string | undefined; + /** Non-stream JSON translation; same-protocol entries pass the payload through. */ + finish: Record unknown>; } +/** + * In-band failure message carried by a non-OK-shaped 200 payload, per upstream + * protocol. Identical across the client-facing endpoints; Anthropic reports + * failures as non-200 responses instead, so it has no in-band form here. + */ +const FAILURE_BY_PROTOCOL: Record string | undefined) | undefined> = { + "chat-completions": chatResponseFailure, + "responses": responsesResponseFailure, + "anthropic": undefined, +}; + const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = { responses: { payloads: { @@ -328,10 +307,11 @@ const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = "responses": pipeResponsesPassthrough, "anthropic": pipeAnthropicStreamToResponses, }, - finish: (upstream, provider, model) => provider.protocol === "anthropic" - ? fromAnthropicResponse(upstream, model) - : provider.protocol === "chat-completions" ? fromChatResponseToResponses(upstream, model) : upstream, - failure: (upstream, provider) => provider.protocol === "chat-completions" ? chatResponseFailure(upstream) : provider.protocol === "responses" ? responsesResponseFailure(upstream) : undefined, + finish: { + "responses": (upstream) => upstream, + "chat-completions": (upstream, _provider, model) => fromChatResponseToResponses(upstream, model), + "anthropic": (upstream, _provider, model) => fromAnthropicResponse(upstream, model), + }, }, chat: { payloads: { @@ -344,10 +324,11 @@ const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = "responses": pipeResponsesStreamToChat, "anthropic": pipeAnthropicStreamToChat, }, - finish: (upstream, provider, model) => provider.protocol === "anthropic" - ? fromAnthropicResponseToChat(upstream, model) - : provider.protocol === "responses" ? fromResponsesResponseToChat(upstream, model) : upstream, - failure: (upstream, provider) => provider.protocol === "responses" ? responsesResponseFailure(upstream) : provider.protocol === "chat-completions" ? chatResponseFailure(upstream) : undefined, + finish: { + "chat-completions": (upstream) => upstream, + "responses": (upstream, _provider, model) => fromResponsesResponseToChat(upstream, model), + "anthropic": (upstream, _provider, model) => fromAnthropicResponseToChat(upstream, model), + }, }, messages: { payloads: { @@ -362,8 +343,11 @@ const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = "anthropic": pipeAnthropicPassthrough, }, // Anthropic upstreams are already Anthropic-shaped; others convert back. - finish: (upstream, provider) => provider.protocol === "anthropic" ? upstream : provider.protocol === "responses" ? fromResponsesResponse(upstream, provider.model) : fromChatResponse(upstream, provider.model), - failure: (upstream, provider) => provider.protocol === "chat-completions" ? chatResponseFailure(upstream) : provider.protocol === "responses" ? responsesResponseFailure(upstream) : undefined, + finish: { + "anthropic": (upstream) => upstream, + "responses": (upstream, provider) => fromResponsesResponse(upstream, provider.model), + "chat-completions": (upstream, provider) => fromChatResponse(upstream, provider.model), + }, }, }; @@ -419,11 +403,11 @@ export async function startAdapter(config: Config, options: AdapterOptions = {}) return json(response, 200, { input_tokens: exact ?? estimateInputTokens(input) }); } catch (error) { return respondError(response, error); } } - const specFor = pathname === "/v1/responses" ? ENDPOINT_SPECS.responses + const endpointSpec = pathname === "/v1/responses" ? ENDPOINT_SPECS.responses : pathname === "/v1/chat/completions" ? ENDPOINT_SPECS.chat : pathname === "/v1/messages" ? ENDPOINT_SPECS.messages : undefined; - if (!specFor || request.method !== "POST") return json(response, 404, { error: { message: "Not found", type: "not_found" } }); + if (!endpointSpec || request.method !== "POST") return json(response, 404, { error: { message: "Not found", type: "not_found" } }); // Authentication comes first: a request that is going to be rejected must // not sit in the queue ahead of a legitimate one. if (!authorized(request, token)) return unauthorized(response); @@ -436,16 +420,16 @@ export async function startAdapter(config: Config, options: AdapterOptions = {}) // The endpoint decides which model id counts as "configured"; clients that // echo a preferred model (Codex sending OPENAI_MODEL=auto, Claude Code's // haiku background lane) are honored per honorRequestedModel's rules. - const proxied = await proxyRequest(config, request, response, config.model, (input, model, provider) => specFor.payloads[provider.protocol](input, model, provider), maxBodyBytes); + const proxied = await proxyRequest(config, request, response, config.model, (input, model, provider) => endpointSpec.payloads[provider.protocol](input, model, provider), maxBodyBytes); if (!proxied) return; const { input, model, provider, watched } = proxied; const usageOptions = streamOptions(config, provider, sessionFor(input), safeRecord); - if (input.stream) return specFor.pipes[provider.protocol](watched, response, model, usageOptions); + if (input.stream) return endpointSpec.pipes[provider.protocol](watched, response, model, usageOptions); if (!watched.body) { response.end(); return; } const value = await watched.json(); recordUsage(value, provider, sessionFor(input)); - const failure = specFor.failure?.(value, provider); + const failure = FAILURE_BY_PROTOCOL[provider.protocol]?.(value); if (failure) return json(response, 502, { error: { message: failure, type: "upstream_error" } }); - return json(response, watched.status, specFor.finish(value, provider, model)); + return json(response, watched.status, endpointSpec.finish[provider.protocol](value, provider, model)); }; const server = createServer((request, response) => { // Last-resort guard: a rejection escaping the per-route handling (an diff --git a/test/json.test.ts b/test/json.test.ts index a35d434..4ef3bdc 100644 --- a/test/json.test.ts +++ b/test/json.test.ts @@ -1,6 +1,6 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { asRecords, isRecord, jsonRecord, parse, recCount, recNum, recObj, recObjs, recStr } from "../src/json.js"; +import { asRecords, isRecord, jsonRecord, parseJson, recCount, recNum, recObj, recObjs, recStr } from "../src/json.js"; test("isRecord accepts objects and rejects arrays, null and primitives", () => { assert.equal(isRecord({}), true); @@ -57,9 +57,9 @@ test("object accessors read nested payloads and drop the wrong shapes", () => { assert.deepEqual(recObjs(payload, "usage"), []); }); -test("parse is best-effort and never throws", () => { - assert.deepEqual(parse('{"a":1}'), { a: 1 }); - assert.deepEqual(parse({ a: 1 }), { a: 1 }); - assert.deepEqual(parse("not json"), {}); - assert.deepEqual(parse(undefined), {}); +test("parseJson is best-effort and never throws", () => { + assert.deepEqual(parseJson('{"a":1}'), { a: 1 }); + assert.deepEqual(parseJson({ a: 1 }), { a: 1 }); + assert.deepEqual(parseJson("not json"), {}); + assert.deepEqual(parseJson(undefined), {}); }); From 4ce917db04a315e51c728b63c8b3f744a67b9901 Mon Sep 17 00:00:00 2001 From: tzzs Date: Mon, 21 Sep 2026 00:43:34 +0800 Subject: [PATCH 4/4] refactor(server): carry one request context instead of three loose fields input/model/provider traveled together through proxyRequest and all nine endpoint payload builders, so every signature re-stated the same route. Binding them into a RequestContext lets the payload seam pass one value and proxyRequest hand its caller back the same context it resolved. --- src/server.ts | 39 +++++++++++++++++++++++---------------- 1 file changed, 23 insertions(+), 16 deletions(-) diff --git a/src/server.ts b/src/server.ts index 9cc77ad..4478211 100644 --- a/src/server.ts +++ b/src/server.ts @@ -227,12 +227,18 @@ function watchedBody(stream: ReadableStream | null, progress: () => return stream.pipeThrough(new TransformStream({ transform(chunk, controller) { progress(); controller.enqueue(chunk); } })); } +/** + * The parsed client request plus its resolved route — the three fields every + * payload builder and downstream handler travel with, carried as one value. + */ +interface RequestContext { input: JsonRecord; model: string; provider: ProviderModel } + /** * Shared proxy pipeline for one request: parse the body, resolve * provider/model/key, convert the payload to the upstream protocol, forward it * with the idle watchdog, and map failures to errors. Returns `undefined` * after answering the request directly (non-OK upstream status); otherwise - * yields the parsed request and the watched upstream response. + * yields the request context and the watched upstream response. * * Callers authenticate before reaching this: the route table rejects a bad * token ahead of the concurrency queue so an unauthorized request cannot make @@ -243,9 +249,9 @@ async function proxyRequest( request: IncomingMessage, response: ServerResponse, fallbackModel: string, - payload: (input: JsonRecord, model: string, provider: ProviderModel) => unknown, + payload: (context: RequestContext) => unknown, maxBodyBytes = MAX_BODY_BYTES, -): Promise<{ input: JsonRecord; model: string; provider: ProviderModel; watched: Response } | undefined> { +): Promise<(RequestContext & { watched: Response }) | undefined> { try { const input = jsonRecord(JSON.parse(await body(request, maxBodyBytes))); // The endpoint decides which model id counts as "configured"; clients @@ -254,14 +260,15 @@ async function proxyRequest( const model = honorRequestedModel(recStr(input, "model"), fallbackModel, config.provider); debug(config, `POST ${request.url} model=${model}`); const provider = providerFor(model, config.provider); const apiKey = apiKeyFor(provider, config.apiKey); - const payloadBody = payload(input, model, provider); + const context: RequestContext = { input, model, provider }; + const payloadBody = payload(context); const session = upstreamSession(response, config); const upstream = await forwardWithRetry(config, provider, apiKey, payloadBody, session.signal, config.retry); session.progress(); debug(config, `provider status=${upstream.status}`); if (!upstream.ok) { await upstreamError(response, providerDisplayName(provider.provider), upstream, upstream.status); return undefined; } const watched = new Response(watchedBody(upstream.body, session.progress), { status: upstream.status }); - return { input, model, provider, watched }; + return { ...context, watched }; } catch (error) { debug(config, `proxy error=${error instanceof Error ? error.message : "unknown"}`); respondError(response, error); @@ -278,7 +285,7 @@ async function proxyRequest( * to the Anthropic pipe, never the Responses one). */ interface EndpointSpec { - payloads: Record unknown>; + payloads: Record unknown>; pipes: Record Promise>; /** Non-stream JSON translation; same-protocol entries pass the payload through. */ finish: Record unknown>; @@ -298,9 +305,9 @@ const FAILURE_BY_PROTOCOL: Record s const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = { responses: { payloads: { - "responses": (input, model) => ({ ...input, model }), - "chat-completions": (input, model, provider) => toChatCompletionsRequest(input, model, provider.provider), - "anthropic": (input, model) => toAnthropicRequest(input, model), + "responses": ({ input, model }) => ({ ...input, model }), + "chat-completions": ({ input, model, provider }) => toChatCompletionsRequest(input, model, provider.provider), + "anthropic": ({ input, model }) => toAnthropicRequest(input, model), }, pipes: { "chat-completions": pipeChatStreamToResponses, @@ -315,9 +322,9 @@ const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = }, chat: { payloads: { - "responses": (input, model) => toResponsesRequestFromChat(input, model), - "chat-completions": (input, model) => ({ ...input, model }), - "anthropic": (input, model) => toAnthropicRequestFromChat(input, model), + "responses": ({ input, model }) => toResponsesRequestFromChat(input, model), + "chat-completions": ({ input, model }) => ({ ...input, model }), + "anthropic": ({ input, model }) => toAnthropicRequestFromChat(input, model), }, pipes: { "chat-completions": pipeChatPassthrough, @@ -332,10 +339,10 @@ const ENDPOINT_SPECS: Record<"responses" | "chat" | "messages", EndpointSpec> = }, messages: { payloads: { - "responses": (input, model, provider) => toResponsesRequest(asAnthropicRequest(input), model, provider), - "chat-completions": (input, model, provider) => toChatRequest(asAnthropicRequest(input), model, provider), + "responses": ({ input, model, provider }) => toResponsesRequest(asAnthropicRequest(input), model, provider), + "chat-completions": ({ input, model, provider }) => toChatRequest(asAnthropicRequest(input), model, provider), // already Anthropic-shaped; zero conversion - "anthropic": (input, model) => ({ ...input, model }), + "anthropic": ({ input, model }) => ({ ...input, model }), }, pipes: { "chat-completions": pipeResponsesStream, @@ -420,7 +427,7 @@ export async function startAdapter(config: Config, options: AdapterOptions = {}) // The endpoint decides which model id counts as "configured"; clients that // echo a preferred model (Codex sending OPENAI_MODEL=auto, Claude Code's // haiku background lane) are honored per honorRequestedModel's rules. - const proxied = await proxyRequest(config, request, response, config.model, (input, model, provider) => endpointSpec.payloads[provider.protocol](input, model, provider), maxBodyBytes); + const proxied = await proxyRequest(config, request, response, config.model, (context) => endpointSpec.payloads[context.provider.protocol](context), maxBodyBytes); if (!proxied) return; const { input, model, provider, watched } = proxied; const usageOptions = streamOptions(config, provider, sessionFor(input), safeRecord);