diff --git a/packages/core/src/agents/local-providers/claude-code.ts b/packages/core/src/agents/local-providers/claude-code.ts index ef0e210b..9b13f7f3 100644 --- a/packages/core/src/agents/local-providers/claude-code.ts +++ b/packages/core/src/agents/local-providers/claude-code.ts @@ -55,10 +55,10 @@ const percentLimitMapping = (id: string, label: string, path: string, window: st const claudeCodeAccountMapping: ProviderAccountMappingConfig = { meters: [ percentLimitMapping("claude_five_hour_quota", "5h quota", "$.five_hour", "5h"), - percentLimitMapping("claude_seven_day_quota", "7d quota", "$.seven_day", "7d"), - percentLimitMapping("claude_oauth_apps_quota", "OAuth apps quota", "$.seven_day_oauth_apps", "7d"), - percentLimitMapping("claude_opus_quota", "Opus quota", "$.seven_day_opus", "7d"), - percentLimitMapping("claude_sonnet_quota", "Sonnet quota", "$.seven_day_sonnet", "7d"), + percentLimitMapping("claude_seven_day_quota", "7d quota", "$.seven_day", "weekly"), + percentLimitMapping("claude_oauth_apps_quota", "OAuth apps quota", "$.seven_day_oauth_apps", "weekly"), + percentLimitMapping("claude_opus_quota", "Opus quota", "$.seven_day_opus", "weekly"), + percentLimitMapping("claude_sonnet_quota", "Sonnet quota", "$.seven_day_sonnet", "weekly"), { id: "claude_extra_usage", kind: "quota", diff --git a/packages/core/src/gateway/core-runtime/responses-session-affinity.ts b/packages/core/src/gateway/core-runtime/responses-session-affinity.ts new file mode 100644 index 00000000..4b8ffa48 --- /dev/null +++ b/packages/core/src/gateway/core-runtime/responses-session-affinity.ts @@ -0,0 +1,104 @@ +import { isRecord, stringValue } from "@ccr/core/gateway/internal/value"; + +type HeaderValue = string | string[] | undefined; + +type UpstreamRequest = { + body?: unknown; + bodyEncoding?: "bytes" | "form" | "json" | "none" | "text"; + headers?: Record; + method?: string; + url: string; +}; + +export type ResponsesSessionAffinityInput = { + request?: { + body?: unknown; + headers?: Record; + }; + targetProviderConfig?: { + type?: string; + }; + upstreamRequest: UpstreamRequest; +}; + +const sessionIdHeaderNames = ["x-claude-code-session-id", "x-claude-session-id"]; + +/** + * Copies the Claude Code session identity onto outbound OpenAI Responses + * bodies. The protocol conversion emits neither `prompt_cache_key` nor + * `metadata.user_id`, so multi-channel Responses upstreams that pin sessions + * on body fields hash each turn onto a different channel and the next hop + * rejects channel-bound `encrypted_content` continuations. A caller-supplied + * non-empty `prompt_cache_key` always wins; other protocols and non-JSON + * bodies pass through untouched. + */ +export function applyResponsesSessionAffinity(input: ResponsesSessionAffinityInput): UpstreamRequest { + const upstreamRequest = input.upstreamRequest; + const providerType = input.targetProviderConfig?.type?.trim().toLowerCase(); + if (providerType !== "openai_responses") { + return upstreamRequest; + } + const body = upstreamRequest.body; + if ((upstreamRequest.bodyEncoding ?? "json") !== "json" || !isRecord(body)) { + return upstreamRequest; + } + + const inboundUserId = inboundMetadataUserId(input.request?.body); + const changes: Record = {}; + if (!stringValue(body.prompt_cache_key)) { + const sessionKey = resolveResponsesSessionKey(input.request?.headers, inboundUserId); + if (sessionKey) { + changes.prompt_cache_key = sessionKey; + } + } + if (inboundUserId && body.metadata === undefined) { + changes.metadata = { user_id: inboundUserId }; + } + if (Object.keys(changes).length === 0) { + return upstreamRequest; + } + return { + ...upstreamRequest, + body: { ...body, ...changes } + }; +} + +/** + * First non-empty of the Claude Code session headers (case-insensitive), + * falling back to the inbound Anthropic `metadata.user_id`. + */ +export function resolveResponsesSessionKey( + headers: Record | undefined, + inboundUserId: string | undefined +): string | undefined { + for (const name of sessionIdHeaderNames) { + const value = readHeaderValue(headers, name); + if (value) { + return value; + } + } + return inboundUserId; +} + +function inboundMetadataUserId(body: unknown): string | undefined { + if (!isRecord(body) || !isRecord(body.metadata)) { + return undefined; + } + return stringValue(body.metadata.user_id); +} + +function readHeaderValue(headers: Record | undefined, name: string): string | undefined { + for (const [headerName, headerValue] of Object.entries(headers ?? {})) { + if (headerName.trim().toLowerCase() !== name) { + continue; + } + const values = Array.isArray(headerValue) ? headerValue : [headerValue]; + for (const value of values) { + const normalized = stringValue(value); + if (normalized) { + return normalized; + } + } + } + return undefined; +} diff --git a/packages/core/src/gateway/core-runtime/upstream-header-sanitizer.ts b/packages/core/src/gateway/core-runtime/upstream-header-sanitizer.ts index 47ea9b4f..bbcde520 100644 --- a/packages/core/src/gateway/core-runtime/upstream-header-sanitizer.ts +++ b/packages/core/src/gateway/core-runtime/upstream-header-sanitizer.ts @@ -1,3 +1,6 @@ +import { applyResponsesSessionAffinity } from "@ccr/core/gateway/core-runtime/responses-session-affinity"; +import type { ResponsesSessionAffinityInput } from "@ccr/core/gateway/core-runtime/responses-session-affinity"; + type UpstreamRequest = { body: unknown; bodyEncoding?: "bytes" | "form" | "json" | "none" | "text"; @@ -196,6 +199,14 @@ export function createGatewayPlugin() { } }; } + }, { + key: "ccr-responses-session-affinity", + transformRequest(input: ResponsesSessionAffinityInput) { + return { + ok: true as const, + value: applyResponsesSessionAffinity(input) + }; + } }] }; } diff --git a/packages/core/src/gateway/features/codex-multi-agent-bridge.ts b/packages/core/src/gateway/features/codex-multi-agent-bridge.ts index 57f15fa3..3914cbc9 100644 --- a/packages/core/src/gateway/features/codex-multi-agent-bridge.ts +++ b/packages/core/src/gateway/features/codex-multi-agent-bridge.ts @@ -1,5 +1,6 @@ import type { IncomingHttpHeaders } from "node:http"; import { Readable, Transform } from "node:stream"; +import { StringDecoder } from "node:string_decoder"; import type { AppConfig } from "@ccr/core/contracts/app"; import { normalizeRouteSelector } from "@ccr/core/routing/model-registry"; import { isRecord, rawStringValue, stringValue } from "@ccr/core/gateway/internal/value"; @@ -345,9 +346,17 @@ function transformMultiAgentFunctionCall(item: Record): { value }; } +type CodexMultiAgentBridgeSseTransform = Transform & { + __ccrCodexMultiAgentBridgeSseDecoder?: StringDecoder; + __ccrCodexMultiAgentBridgeSsePending?: string; +}; + function transformSseChunk(stream: Transform, chunk: Buffer | string): void { - const state = stream as Transform & { __ccrCodexMultiAgentBridgeSsePending?: string }; - state.__ccrCodexMultiAgentBridgeSsePending = (state.__ccrCodexMultiAgentBridgeSsePending ?? "") + chunk.toString(); + const state = stream as CodexMultiAgentBridgeSseTransform; + const decoder = state.__ccrCodexMultiAgentBridgeSseDecoder ?? new StringDecoder("utf8"); + state.__ccrCodexMultiAgentBridgeSseDecoder = decoder; + state.__ccrCodexMultiAgentBridgeSsePending = (state.__ccrCodexMultiAgentBridgeSsePending ?? "") + + decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); while (state.__ccrCodexMultiAgentBridgeSsePending) { const match = /\r?\n\r?\n/.exec(state.__ccrCodexMultiAgentBridgeSsePending); if (!match || match.index === undefined) { @@ -361,7 +370,9 @@ function transformSseChunk(stream: Transform, chunk: Buffer | string): void { } function flushSseTransform(stream: Transform): void { - const state = stream as Transform & { __ccrCodexMultiAgentBridgeSsePending?: string }; + const state = stream as CodexMultiAgentBridgeSseTransform; + state.__ccrCodexMultiAgentBridgeSsePending = (state.__ccrCodexMultiAgentBridgeSsePending ?? "") + + (state.__ccrCodexMultiAgentBridgeSseDecoder?.end() ?? ""); if (state.__ccrCodexMultiAgentBridgeSsePending) { stream.push(transformCodexMultiAgentBridgeSseEvent(state.__ccrCodexMultiAgentBridgeSsePending)); state.__ccrCodexMultiAgentBridgeSsePending = ""; diff --git a/packages/core/src/gateway/features/codex-patch-bridge.ts b/packages/core/src/gateway/features/codex-patch-bridge.ts index a79be245..c3aa1d88 100644 --- a/packages/core/src/gateway/features/codex-patch-bridge.ts +++ b/packages/core/src/gateway/features/codex-patch-bridge.ts @@ -3,6 +3,7 @@ */ import type { IncomingHttpHeaders } from "node:http"; import { Readable, Transform } from "node:stream"; +import { StringDecoder } from "node:string_decoder"; import type { AppConfig } from "@ccr/core/contracts/app"; import { normalizeRouteSelector } from "@ccr/core/routing/model-registry"; import { isRecord, rawStringValue, stringValue } from "@ccr/core/gateway/internal/value"; @@ -416,9 +417,17 @@ function patchInputFromVirtualApplyPatchArguments(value: unknown): string | unde } +type CodexPatchBridgeSseTransform = Transform & { + __ccrCodexPatchBridgeSseDecoder?: StringDecoder; + __ccrCodexPatchBridgeSsePending?: string; +}; + function transformSseChunk(stream: Transform, chunk: Buffer | string): void { - const state = stream as Transform & { __ccrCodexPatchBridgeSsePending?: string }; - state.__ccrCodexPatchBridgeSsePending = (state.__ccrCodexPatchBridgeSsePending ?? "") + chunk.toString(); + const state = stream as CodexPatchBridgeSseTransform; + const decoder = state.__ccrCodexPatchBridgeSseDecoder ?? new StringDecoder("utf8"); + state.__ccrCodexPatchBridgeSseDecoder = decoder; + state.__ccrCodexPatchBridgeSsePending = (state.__ccrCodexPatchBridgeSsePending ?? "") + + decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); while (state.__ccrCodexPatchBridgeSsePending) { const match = /\r?\n\r?\n/.exec(state.__ccrCodexPatchBridgeSsePending); if (!match || match.index === undefined) { @@ -433,7 +442,9 @@ function transformSseChunk(stream: Transform, chunk: Buffer | string): void { function flushSseTransform(stream: Transform): void { - const state = stream as Transform & { __ccrCodexPatchBridgeSsePending?: string }; + const state = stream as CodexPatchBridgeSseTransform; + state.__ccrCodexPatchBridgeSsePending = (state.__ccrCodexPatchBridgeSsePending ?? "") + + (state.__ccrCodexPatchBridgeSseDecoder?.end() ?? ""); if (state.__ccrCodexPatchBridgeSsePending) { stream.push(transformCodexApplyPatchBridgeSseEvent(state.__ccrCodexPatchBridgeSsePending)); state.__ccrCodexPatchBridgeSsePending = ""; diff --git a/packages/core/src/gateway/features/hosted-web-search/response-transform.ts b/packages/core/src/gateway/features/hosted-web-search/response-transform.ts index 0074cb05..0397402c 100644 --- a/packages/core/src/gateway/features/hosted-web-search/response-transform.ts +++ b/packages/core/src/gateway/features/hosted-web-search/response-transform.ts @@ -1,4 +1,5 @@ import { Readable, Transform } from "node:stream"; +import { StringDecoder } from "node:string_decoder"; import type { GatewayProviderProtocol } from "@ccr/core/contracts/app"; import { isRecord, numberValue, stringValue } from "@ccr/core/gateway/internal/value"; import { formatError } from "@ccr/core/gateway/http/io"; @@ -65,6 +66,7 @@ function hostedWebSearchProtocolSseStream( const recordsPromise = context.records?.length ? Promise.resolve(context.records) : selectHostedWebSearchProtocolRecords(context, integration); + const decoder = new StringDecoder("utf8"); let records: BrowserWebSearchProtocolRecord[] | undefined; let pending = ""; let passThrough = false; @@ -84,7 +86,7 @@ function hostedWebSearchProtocolSseStream( return input.pipe(new Transform({ transform(chunk, _encoding, callback) { - const text = chunk.toString(); + const text = decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); const rawText = pending + text; void (async () => { await ensureRecords(); @@ -103,6 +105,7 @@ function hostedWebSearchProtocolSseStream( }).finally(() => callback()); }, flush(callback) { + pending += decoder.end(); void (async () => { await ensureRecords(); if (passThrough || !records) { @@ -263,6 +266,7 @@ function anthropicHostedWebSearchProtocolSseStream( const recordsPromise = context.records?.length ? Promise.resolve(context.records) : selectHostedWebSearchProtocolRecords(context, integration); + const decoder = new StringDecoder("utf8"); let records: BrowserWebSearchProtocolRecord[] | undefined; let pending = ""; let passThrough = false; @@ -286,7 +290,7 @@ function anthropicHostedWebSearchProtocolSseStream( return input.pipe(new Transform({ transform(chunk, _encoding, callback) { - const text = chunk.toString(); + const text = decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); const rawText = pending + text; void (async () => { await ensureRecords(); @@ -305,6 +309,7 @@ function anthropicHostedWebSearchProtocolSseStream( }).finally(() => callback()); }, flush(callback) { + pending += decoder.end(); void (async () => { await ensureRecords(); if (passThrough || !records) { diff --git a/packages/core/src/mcp/fusion-vision-mcp.ts b/packages/core/src/mcp/fusion-vision-mcp.ts index e747e84b..7c812018 100644 --- a/packages/core/src/mcp/fusion-vision-mcp.ts +++ b/packages/core/src/mcp/fusion-vision-mcp.ts @@ -83,7 +83,7 @@ const visionTool = { detail: { enum: ["auto", "low", "high"], type: "string" }, imageBase64: { description: "Single raw base64 image payload or data URL.", type: "string" }, imagePath: { description: "Single local image path.", type: "string" }, - imageUrl: { description: "Single HTTP(S) image URL or data URL.", type: "string" }, + imageUrl: { description: "Single HTTP(S) image URL, data URL, or bare base64 payload.", type: "string" }, images: { items: objectSchema({ base64: { type: "string" }, @@ -827,9 +827,11 @@ async function buildImageParts(args: Record, detail: "auto" | " } const parts: JsonValue[] = []; + const skipped: string[] = []; for (const input of inputs) { - const url = await imageInputToUrl(input); - if (!url) { + const result = await imageInputToUrl(input); + if ("skip" in result) { + skipped.push(result.skip); continue; } if (input.label) { @@ -838,36 +840,182 @@ async function buildImageParts(args: Record, detail: "auto" | " parts.push({ image_url: { detail, - url + url: result.url }, type: "image_url" }); } + // Dropping bad images must not look like dropping the argument: surface every + // reason to the caller instead of silently proceeding without the image. + if (parts.length === 0) { + if (skipped.length === 0) { + throw new Error(`${toolName} requires imageUrl, imagePath, imageBase64, or images.`); + } + throw new Error(`${toolName} found no usable image. Skipped: ${skipped.join("; ")}.`); + } return parts; } -async function imageInputToUrl(input: { base64?: string; mimeType?: string; path?: string; url?: string }): Promise { +type ImageInputToUrlResult = { url: string } | { skip: string }; + +/** + * Turn one image input into a data URL ready for the upstream, or say why it cannot + * be used. The upstream (commonly litellm in front of a strict vision provider) + * rejects any image whose payload is not strictly valid base64 and reports that as + * a 400 which surfaces to the caller as a raw provider error -- so anything that + * cannot be made into a well-formed data URL is dropped here instead of forwarded. + * + * Failure shapes seen in production: a local file path passed as imageUrl, a bare + * [media_ref:...] id passed verbatim, a base64 payload whose length mod 4 is 1 + * (irreparably truncated mid-image), and an XML/SVG payload -- which the typical + * upstream rejects outright (supported formats are jpeg/png/gif/webp), so + * relabeling it buys nothing. A remainder of 2 or 3 is repairable by padding. + * Local files (imagePath/images[].path) go through the same content checks. + */ +async function imageInputToUrl(input: { base64?: string; mimeType?: string; path?: string; url?: string }): Promise { if (input.url) { - return input.url; + const url = input.url.trim(); + if (/^https?:\/\//i.test(url)) { + return { url }; + } + if (url.startsWith("data:")) { + return dataUrlResult(url, input.mimeType); + } + // Not an HTTP(S) URL and not a data URL. Either a bare base64 payload (the + // virtual-model tool loop's usual shape; wrapped here because strict gateways + // reject it as an invalid URL) or garbage that must not be forwarded as-is. + const normalized = normalizeImagePayload(url); + if (!normalized) { + return { skip: `imageUrl is neither an HTTP(S) URL, a data URL, nor base64 (${preview(url)})` }; + } + return imageDataUrlOrSkip(normalized, "image/png"); } if (input.base64) { - return toDataUrl(input.base64, input.mimeType || "image/png"); + const value = input.base64.trim(); + if (value.startsWith("data:")) { + // The schema documents imageBase64 as "Single raw base64 image payload or + // data URL"; both shapes must behave alike, so a data URL takes the same + // path as imageUrl above. + return dataUrlResult(value, input.mimeType); + } + const normalized = normalizeImagePayload(value); + if (!normalized) { + return { skip: "imageBase64 is not usable base64" }; + } + return imageDataUrlOrSkip(normalized, input.mimeType); } if (!input.path) { - return undefined; + return { skip: "image entry has no url, base64, or path" }; } const buffer = await readFile(input.path); if (buffer.byteLength > maxLocalImageBytes) { throw new Error(`Local image exceeds ${maxLocalImageBytes} bytes: ${input.path}`); } - return toDataUrl(buffer.toString("base64"), input.mimeType || mimeTypeFromPath(input.path)); + // File contents get the same checks as every other input: a .svg on disk must + // not be forwarded as a fake raster data URL. + const normalized = normalizeImagePayload(buffer.toString("base64")); + if (!normalized) { + return { skip: `file at ${preview(input.path)} is not usable image data` }; + } + return imageDataUrlOrSkip(normalized, input.mimeType || mimeTypeFromPath(input.path)); } -function toDataUrl(value: string, mimeType: string): string { - return value.startsWith("data:") ? value : `data:${mimeType};base64,${value}`; +/** Only these media types are ever emitted on a data URL; strict upstreams validate them. */ +const supportedImageMimeTypes = new Set(["image/jpeg", "image/png", "image/gif", "image/webp"]); + +/** + * Parse and validate a full `data:...;base64,...` URL. Shared by the imageUrl and + * imageBase64 fields so both accept the same shapes. Non-base64 data URLs are + * rejected; a header media type outside the supported set is dropped and the + * payload's sniffed type is used instead. + */ +function dataUrlResult(value: string, fallbackMimeType: string | undefined): ImageInputToUrlResult { + const comma = value.indexOf(","); + if (comma < 1) { + return { skip: `malformed data URL (${preview(value)})` }; + } + const header = value.slice(5, comma); + if (!/;base64/i.test(header)) { + return { skip: `data URL is not base64 (${preview(value)})` }; + } + const headerMimeType = header.split(";")[0] || undefined; + const normalized = normalizeImagePayload(value.slice(comma + 1)); + if (!normalized) { + return { skip: `data URL payload is not usable base64 (${preview(value)})` }; + } + return imageDataUrlOrSkip(normalized, headerMimeType || fallbackMimeType); } -function mimeTypeFromPath(path: string): string { +const base64PayloadPattern = /^[A-Za-z0-9+/]*={0,2}$/; +const svgHeadPattern = /^(?:\uFEFF)?\s*(?:<\?xml|= 3 && buffer[0] === 0xff && buffer[1] === 0xd8 && buffer[2] === 0xff) { + return "image/jpeg"; + } + if (buffer.length >= 8 && buffer[0] === 0x89 && buffer[1] === 0x50 && buffer[2] === 0x4e && buffer[3] === 0x47) { + return "image/png"; + } + if (buffer.length >= 6 && buffer[0] === 0x47 && buffer[1] === 0x49 && buffer[2] === 0x46 && buffer[3] === 0x38 && (buffer[4] === 0x37 || buffer[4] === 0x39) && buffer[5] === 0x61) { + return "image/gif"; + } + if (buffer.length >= 12 && buffer.subarray(0, 4).toString("latin1") === "RIFF" && buffer.subarray(8, 12).toString("latin1") === "WEBP") { + return "image/webp"; + } + return undefined; +} + +/** + * Pick the media-type label for the data URL. Only the supported raster types are + * ever emitted: an explicit label in that set wins, otherwise the sniffed format, + * otherwise image/png. The label is validated for its own sake -- it is part of the + * data URL a strict upstream checks -- while the bytes decide whether the image is + * decodable at all. + */ +function imageLabel(mimeType: string | undefined, image: { sniffedType?: string }): string { + const explicit = mimeType?.trim().toLowerCase(); + if (explicit && supportedImageMimeTypes.has(explicit)) { + return explicit; + } + return image.sniffedType ?? "image/png"; +} + +function imageDataUrlOrSkip(image: { payload: string; svg: boolean; sniffedType?: string }, mimeType: string | undefined): ImageInputToUrlResult { + if (image.svg) { + return { skip: "SVG/XML image payload is not supported by the vision upstream (supported: image/jpeg, image/png, image/gif, image/webp)" }; + } + return { url: `data:${imageLabel(mimeType, image)};base64,${image.payload}` }; +} + +/** Short, quoted head of a value for skip reasons and error messages. */ +function preview(value: string): string { + return JSON.stringify(value.length > 40 ? `${value.slice(0, 37)}…` : value); +}function mimeTypeFromPath(path: string): string { const ext = extname(path).toLowerCase(); if (ext === ".jpg" || ext === ".jpeg") return "image/jpeg"; if (ext === ".webp") return "image/webp"; diff --git a/packages/core/src/media/executors.ts b/packages/core/src/media/executors.ts index ac62d117..fd09e645 100644 --- a/packages/core/src/media/executors.ts +++ b/packages/core/src/media/executors.ts @@ -116,15 +116,21 @@ export class GatewayMediaExecutor { if (attempt < 3) await delay(attempt * 500, undefined, { signal }); } if (!response) throw lastError ?? mediaError("artifact_download_failed", "Failed to download generated artifact.", true); - if (!response.ok) throw mediaError("artifact_download_failed", `Failed to download generated artifact: HTTP ${response.status}.`, true); + if (!response.ok) { + await response.body?.cancel(); + throw mediaError("artifact_download_failed", `Failed to download generated artifact: HTTP ${response.status}.`, true); + } const declaredLength = Number(response.headers.get("content-length") ?? 0); - if (declaredLength > maxApiArtifactBytes) throw mediaError("artifact_too_large", "Generated artifact exceeds the 250 MB limit.", false); + if (declaredLength > maxApiArtifactBytes) { + await response.body?.cancel(); + throw mediaError("artifact_too_large", "Generated artifact exceeds the 250 MB limit.", false); + } if (!response.body) throw mediaError("artifact_download_failed", "Generated artifact response has no body.", true); const temporary = path.join(os.tmpdir(), `ccr-media-${randomUUID()}.download`); const file = openSync(temporary, "wx", 0o600); + const reader = response.body.getReader(); let size = 0; try { - const reader = response.body.getReader(); while (true) { const chunk = await reader.read(); if (chunk.done) break; @@ -137,6 +143,7 @@ export class GatewayMediaExecutor { writeSync(file, buffer); } } catch (error) { + await reader.cancel().catch(() => undefined); closeSync(file); rmSync(temporary, { force: true }); throw error; diff --git a/packages/core/src/observability/request-log-store.ts b/packages/core/src/observability/request-log-store.ts index c43be3e8..b5306681 100644 --- a/packages/core/src/observability/request-log-store.ts +++ b/packages/core/src/observability/request-log-store.ts @@ -556,11 +556,19 @@ export class RequestLogStore { const responseError = normalizeFilterValue(input.error) ?? detectSseError(responseBodyText, headerValue(responseHeaders, "content-type")); const bodyUsage = extractUsageFromBody(responseBodyText); - const usage: UsageSnapshot = normalizeUsageInputTokens(mergeUsageSnapshots(extractUsageFromBillingHeaders(input.responseHeaders), bodyUsage), { - path: input.path, - providerProtocol: input.providerProtocol, - usageHint: bodyUsage - }) ?? {}; + // Each source carries its own cache-inclusion convention; normalize before + // merging (see UsageConventionSource). + const usage: UsageSnapshot = mergeUsageSnapshots( + normalizeUsageInputTokens(extractUsageFromBillingHeaders(input.responseHeaders), { + path: input.path, + providerProtocol: input.providerProtocol, + source: "providerBilling" + }), + normalizeUsageInputTokens(bodyUsage, { + path: input.path, + source: "responseBody" + }) + ) ?? {}; const route = splitRequestLogRouteSelector(input.fallbackModel); const bodyModel = requestLogRequestedModel(input.requestBody, input.path); const requestModel = normalizeFilterValue(input.model) ?? bodyModel; @@ -891,10 +899,20 @@ export class RequestLogStore { const bodyUsage = input.responseBodyText === undefined ? undefined : extractUsageFromBody(input.responseBodyText); - const usage: UsageSnapshot = normalizeUsageInputTokens(mergeUsageSnapshots(extractUsageFromBillingHeaders(responseHeaders), bodyUsage), { - path: usagePath, - usageHint: bodyUsage - }) ?? {}; + // As in record(), each source is normalized under its own convention. + // Raw-trace updates carry no provider protocol — it is not part of the + // gateway's raw-trace sync contract — so the billing headers fall back to + // the request path, which is only a proxy for the upstream's convention. + const usage: UsageSnapshot = mergeUsageSnapshots( + normalizeUsageInputTokens(extractUsageFromBillingHeaders(responseHeaders), { + path: usagePath, + source: "providerBilling" + }), + normalizeUsageInputTokens(bodyUsage, { + path: usagePath, + source: "responseBody" + }) + ) ?? {}; if (hasUsageNumbers(usage)) { const inputTokens = normalizeCount(usage.inputTokens); const outputTokens = normalizeCount(usage.outputTokens); diff --git a/packages/core/src/usage/billing-sync.ts b/packages/core/src/usage/billing-sync.ts index 990e656a..c992e3ba 100644 --- a/packages/core/src/usage/billing-sync.ts +++ b/packages/core/src/usage/billing-sync.ts @@ -166,8 +166,11 @@ export class GatewayBillingSynchronizer { outputTokens: numberValue(usage.output_tokens), totalTokens: numberValue(usage.total_tokens) }, { + // The gateway's own billing event, so these are the upstream provider's + // counters in the upstream's convention. path, - providerProtocol + providerProtocol, + source: "providerBilling" }); const reportedCost = finiteNumber(cost.total); const input: UsageEventInput = { diff --git a/packages/core/src/usage/normalization.ts b/packages/core/src/usage/normalization.ts index e2355b8c..34377fb9 100644 --- a/packages/core/src/usage/normalization.ts +++ b/packages/core/src/usage/normalization.ts @@ -7,13 +7,25 @@ export type UsageTokenAccounting = { inputTokens?: number; }; +/** + * Which artefact the counts were read from. The two disagree whenever the + * gateway translates between formats — an Anthropic response served from an + * OpenAI upstream carries billing headers in the upstream's cache-inclusive + * convention next to a body in Anthropic's cache-exclusive one — so each has + * to be normalized under its own rule before the two are merged. + */ +export type UsageConventionSource = "providerBilling" | "responseBody"; + +export type UsageNormalizationOptions = { + path?: string; + providerProtocol?: GatewayProviderProtocol; + source?: UsageConventionSource; + usageHint?: UsageTokenAccounting; +}; + export function normalizeUsageInputTokens( usage: T | undefined, - options: { - path?: string; - providerProtocol?: GatewayProviderProtocol; - usageHint?: UsageTokenAccounting; - } = {} + options: UsageNormalizationOptions = {} ): T | undefined { if (!usage) { return undefined; @@ -37,12 +49,26 @@ export function normalizeUsageInputTokens( function inputIncludesCacheTokens( usage: UsageTokenAccounting, - options: { - path?: string; - providerProtocol?: GatewayProviderProtocol; - usageHint?: UsageTokenAccounting; - } + options: UsageNormalizationOptions ): boolean | undefined { + if (options.source === "responseBody") { + // A body describes its own wire format, and the field names we parsed are + // the direct evidence of it. The upstream protocol is deliberately not + // consulted: it describes what the gateway talked to, not what it emitted, + // and reading a translated body under the upstream's rule subtracts the + // cached prefix a second time — clamping input tokens to zero on any turn + // where the cache hit exceeds the new input, which is most of them. + if (usage.inputIncludesCacheTokens !== undefined) { + return usage.inputIncludesCacheTokens; + } + if (options.usageHint?.inputIncludesCacheTokens !== undefined) { + return options.usageHint.inputIncludesCacheTokens; + } + return inputIncludesCacheTokensForPath(options.path); + } + + // Billing headers and billing events restate the upstream provider's own + // counters verbatim, so the upstream protocol governs them. const protocolValue = inputIncludesCacheTokensForProtocol(options.providerProtocol); if (protocolValue !== undefined) { return protocolValue; diff --git a/packages/core/src/usage/store.ts b/packages/core/src/usage/store.ts index 68d9c4e3..3e8d215d 100644 --- a/packages/core/src/usage/store.ts +++ b/packages/core/src/usage/store.ts @@ -220,11 +220,20 @@ export class UsageStore { async recordCapture(input: UsageCaptureInput): Promise { const headersUsage = extractUsageFromBillingHeaders(input.responseHeaders); const bodyUsage = extractUsageFromBody(input.bodyText); - const usage = normalizeUsageInputTokens(mergeUsageSnapshots(headersUsage, bodyUsage), { - path: input.path, - providerProtocol: input.providerProtocol, - usageHint: bodyUsage - }); + // Normalize each source under its own convention before merging them: on a + // translated response the billing headers and the body state input tokens + // differently, and one shared rule is wrong for one of them. + const usage = mergeUsageSnapshots( + normalizeUsageInputTokens(headersUsage, { + path: input.path, + providerProtocol: input.providerProtocol, + source: "providerBilling" + }), + normalizeUsageInputTokens(bodyUsage, { + path: input.path, + source: "responseBody" + }) + ); const fallbackAttribution = resolveUsageModelAttribution(input.config, input.fallbackModel); const responseAttribution = resolveUsageResponseModelAttribution(input.config, bodyUsage?.model); const route = splitRouteSelector(input.fallbackModel); diff --git a/packages/core/test/integration/mcp/fusion-vision-mcp.test.mjs b/packages/core/test/integration/mcp/fusion-vision-mcp.test.mjs index 28e2c924..cb4ac833 100644 --- a/packages/core/test/integration/mcp/fusion-vision-mcp.test.mjs +++ b/packages/core/test/integration/mcp/fusion-vision-mcp.test.mjs @@ -1,9 +1,13 @@ import assert from "node:assert/strict"; import { spawn } from "node:child_process"; import http from "node:http"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; import path from "node:path"; import test from "node:test"; +const pngA = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="; + test("Fusion vision MCP sends the core API key and retries a body-free lightweight usage event", async (t) => { const seen = { providerAuthorization: "", @@ -401,6 +405,173 @@ test("Fusion vision MCP preserves slash-containing model IDs for external runtim assert.deepEqual(seen.usageBody?.target, { model }); }); +test("Fusion vision MCP accepts a full data URL in imageBase64", async (t) => { + const provider = await serveVision(t); + if (!provider) { + return; + } + const child = spawnVision(t, provider.port); + const response = await sendJsonRpc(child, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { imageBase64: `data:image/png;base64,${pngA}`, prompt: "Read it." }, + name: "vision_understand" + } + }); + + assert.equal(response.error, undefined); + assert.equal(response.result?.isError, undefined); + assert.equal(provider.requests, 1); + assert.equal(provider.lastBody?.messages?.[0]?.content?.[1]?.image_url?.url, `data:image/png;base64,${pngA}`); +}); + +test("Fusion vision MCP refuses a local SVG file instead of forwarding it as a fake PNG", async (t) => { + const provider = await serveVision(t); + if (!provider) { + return; + } + const dir = await mkdtemp(path.join(os.tmpdir(), "fusion-vision-svg-")); + t.after(() => rm(dir, { force: true, recursive: true })); + const svgPath = path.join(dir, "logo.svg"); + await writeFile(svgPath, '', "utf8"); + + const child = spawnVision(t, provider.port); + const response = await sendJsonRpc(child, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { images: [{ path: svgPath }], prompt: "Read it." }, + name: "vision_understand" + } + }); + + assert.equal(response.result?.isError, true); + assert.match(response.result?.content?.[0]?.text, /SVG\/XML/); + assert.equal(provider.requests, 0, "the SVG must never reach the upstream"); +}); + +test("Fusion vision MCP restricts the forwarded media-type label to supported raster types", async (t) => { + const provider = await serveVision(t); + if (!provider) { + return; + } + const child = spawnVision(t, provider.port); + + const octetStream = await sendJsonRpc(child, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { imageBase64: pngA, mimeType: "application/octet-stream", prompt: "Read it." }, + name: "vision_understand" + } + }); + assert.equal(provider.lastBody?.messages?.[0]?.content?.[1]?.image_url?.url, `data:image/png;base64,${pngA}`); + + const textPlain = await sendJsonRpc(child, { + id: 2, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { imageUrl: `data:text/plain;base64,${pngA}`, prompt: "Read it." }, + name: "vision_understand" + } + }); + assert.equal(provider.lastBody?.messages?.[0]?.content?.[1]?.image_url?.url, `data:image/png;base64,${pngA}`); + + const webp = await sendJsonRpc(child, { + id: 3, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { imageUrl: `data:image/webp;base64,${pngA}`, prompt: "Read it." }, + name: "vision_understand" + } + }); + assert.equal(provider.lastBody?.messages?.[0]?.content?.[1]?.image_url?.url, `data:image/webp;base64,${pngA}`); +}); + +test("Fusion vision MCP forwards a local PNG file with a supported label", async (t) => { + const provider = await serveVision(t); + if (!provider) { + return; + } + const dir = await mkdtemp(path.join(os.tmpdir(), "fusion-vision-png-")); + t.after(() => rm(dir, { force: true, recursive: true })); + const pngPath = path.join(dir, "img.png"); + await writeFile(pngPath, Buffer.from(pngA, "base64")); + + const child = spawnVision(t, provider.port); + const response = await sendJsonRpc(child, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { images: [{ path: pngPath }], prompt: "Read it." }, + name: "vision_understand" + } + }); + + assert.equal(response.error, undefined); + assert.equal(response.result?.isError, undefined); + assert.equal(provider.requests, 1); + assert.equal(provider.lastBody?.messages?.[0]?.content?.[1]?.image_url?.url, `data:image/png;base64,${pngA}`); +}); + +async function serveVision(t, respond) { + let requests = 0; + let lastBody; + const server = http.createServer(async (request, response) => { + const body = readRequestBody(request); + requests += 1; + lastBody = JSON.parse(await body); + if (respond) { + respond(lastBody); + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ + choices: [{ message: { content: "vision ok" } }], + usage: { completion_tokens: 3, prompt_tokens: 10, total_tokens: 13 } + })); + }); + try { + await listen(server); + } catch (error) { + if (isLocalListenUnavailable(error)) { + t.skip(`Local HTTP listen is unavailable: ${formatError(error)}`); + return undefined; + } + throw error; + } + t.after(() => server.close()); + assert.ok(server.address() && typeof server.address() === "object"); + return { port: server.address().port, get requests() { return requests; }, get lastBody() { return lastBody; } }; +} + +function spawnVision(t, port, env = {}) { + const child = spawn(process.execPath, [path.join(process.cwd(), ".test-dist", "core", "runtime", "fusion-vision-mcp.js")], { + env: { + ...process.env, + ELECTRON_RUN_AS_NODE: "1", + FUSION_BUILTIN_TOOL_KIND: "vision", + FUSION_TOOL_NAME: "vision_understand", + VISION_API_KEY: "external-key", + VISION_BASE_URL: `http://127.0.0.1:${port}/v1`, + VISION_MODEL: "test-vision", + ...env + }, + stdio: ["pipe", "pipe", "pipe"] + }); + t.after(() => { + if (!child.killed) { + child.kill(); + } + }); + return child; +} function listen(server) { return new Promise((resolve, reject) => { const onError = (error) => { diff --git a/packages/core/test/unit/gateway/local-agent-auth-provider-hook.test.mjs b/packages/core/test/unit/gateway/local-agent-auth-provider-hook.test.mjs index 2ece12b8..45ddc968 100644 --- a/packages/core/test/unit/gateway/local-agent-auth-provider-hook.test.mjs +++ b/packages/core/test/unit/gateway/local-agent-auth-provider-hook.test.mjs @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import os from "node:os"; import path from "node:path"; import test from "node:test"; @@ -105,6 +105,57 @@ test("Grok local agent auth hook refreshes live login state before authenticatin }); }); +// Regression for musistudio/claude-code-router#1628: the imported plugin's +// auth.headers carry a token snapshot from import time, but the hook must +// resolve the current on-disk token on every call so a rotation picked up by +// the interactive Claude Code CLI is honored without a gateway restart. +test("Claude Code local agent auth hook re-reads the on-disk access token on every request", { skip: process.platform === "win32" }, async () => { + await withClaudeCodeHome(async (home) => { + await withPlatform("darwin", async () => { + await withFakeSecurityFailure(async () => { + writeClaudeCredentials(home, { + accessToken: "stale-imported-access-token", + refreshToken: "stale-refresh-token" + }); + + const [hook] = createGatewayPlugin({ + config: { + providerPlugins: [claudeCodeOauthProviderPlugin()] + } + }).providerHooks; + assert.equal(hook.key, "config:ccr-local-agent-claude-code-api-claude-code-oauth"); + + const upstreamRequest = { + headers: { + "content-type": "application/json", + "x-api-key": "client-key" + }, + method: "POST", + url: "https://api.anthropic.com/v1/messages" + }; + + const staleAuth = await hook.authenticate({ upstreamRequest }); + assert.equal(staleAuth.ok, true); + assert.equal(staleAuth.value.headers.authorization, "Bearer stale-imported-access-token"); + assert.equal(staleAuth.value.headers["x-api-key"], undefined); + assert.equal(upstreamRequest.headers["x-api-key"], "client-key"); + + // Simulate the interactive Claude Code CLI rotating the shared token + // family on disk -- no gateway restart, no re-import. + writeClaudeCredentials(home, { + accessToken: "rotated-access-token", + refreshToken: "rotated-refresh-token" + }); + + const rotatedAuth = await hook.authenticate({ upstreamRequest }); + assert.equal(rotatedAuth.ok, true); + assert.equal(rotatedAuth.value.headers.authorization, "Bearer rotated-access-token"); + assert.equal(rotatedAuth.value.headers["anthropic-beta"], "oauth-2025-04-20"); + }); + }); + }); +}); + test("Grok local agent request hook removes unsupported Responses tools and stale tool choice", () => { const [hook] = createGatewayPlugin({ config: { @@ -232,6 +283,78 @@ function grokOauthProviderPlugin() { }; } +function claudeCodeOauthProviderPlugin() { + return { + auth: { + headers: { + authorization: "Bearer stale-imported-access-token", + "anthropic-beta": "oauth-2025-04-20" + }, + removeHeaders: ["x-api-key"], + strict: true + }, + key: "ccr-local-agent-claude-code-api-claude-code-oauth", + providerName: "Claude Code API" + }; +} + +async function withClaudeCodeHome(run) { + const home = mkdtempSync(path.join(os.tmpdir(), "ccr-claude-code-hook-test-")); + const previousHome = process.env.HOME; + process.env.HOME = home; + try { + await run(home); + } finally { + restoreEnv("HOME", previousHome); + rmSync(home, { force: true, recursive: true }); + } +} + +async function withPlatform(platform, run) { + const descriptor = Object.getOwnPropertyDescriptor(process, "platform"); + Object.defineProperty(process, "platform", { + configurable: true, + value: platform + }); + try { + await run(); + } finally { + Object.defineProperty(process, "platform", descriptor); + } +} + +// Forces the macOS Keychain lookup to miss so the scan falls back to the file +// credentials this test controls, regardless of the host's real Keychain state. +async function withFakeSecurityFailure(run) { + await withFakeSecurityScript("exit 44\n", run); +} + +async function withFakeSecurityScript(body, run) { + const binDir = mkdtempSync(path.join(os.tmpdir(), "ccr-claude-code-security-bin-")); + const securityPath = path.join(binDir, "security"); + const previousPath = process.env.PATH; + const previousUser = process.env.USER; + writeFileSync(securityPath, `#!/bin/sh\n${body}`); + chmodSync(securityPath, 0o755); + process.env.PATH = `${binDir}${path.delimiter}${previousPath ?? ""}`; + process.env.USER = "ccr-test-user"; + try { + await run(); + } finally { + restoreEnv("PATH", previousPath); + restoreEnv("USER", previousUser); + rmSync(binDir, { force: true, recursive: true }); + } +} + +function writeClaudeCredentials(home, credentials) { + const directory = path.join(home, ".claude"); + const credentialFile = path.join(directory, ".credentials.json"); + mkdirSync(directory, { recursive: true }); + writeFileSync(credentialFile, JSON.stringify(credentials, null, 2)); + return credentialFile; +} + async function withGrokHome(t, run) { const previousGrokHome = process.env.GROK_HOME; const previousGrokAuthFile = process.env.GROK_AUTH_FILE; diff --git a/packages/core/test/unit/gateway/responses-session-affinity.test.mjs b/packages/core/test/unit/gateway/responses-session-affinity.test.mjs new file mode 100644 index 00000000..20b58ac4 --- /dev/null +++ b/packages/core/test/unit/gateway/responses-session-affinity.test.mjs @@ -0,0 +1,142 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { applyResponsesSessionAffinity, resolveResponsesSessionKey } from "@ccr/core/gateway/core-runtime/responses-session-affinity.ts"; +import { createGatewayPlugin } from "@ccr/core/gateway/core-runtime/upstream-header-sanitizer.ts"; + +function responsesInput(overrides = {}) { + return { + request: { + body: { + messages: [{ content: "hello", role: "user" }], + metadata: { user_id: "user_abc123_account__session_11112222" }, + model: "claude-sonnet-4-5" + }, + headers: { + "x-claude-code-session-id": "session-1111-2222" + } + }, + targetProviderConfig: { + name: "multi-channel::openai_responses", + type: "openai_responses" + }, + upstreamRequest: { + body: { + input: [], + instructions: "system prompt", + max_output_tokens: 32000, + model: "gpt-5.1-codex", + stream: true + }, + bodyEncoding: "json", + headers: { "content-type": "application/json" }, + method: "POST", + url: "https://provider.example/v1/responses" + }, + ...overrides + }; +} + +test("openai_responses bodies gain prompt_cache_key from the Claude Code session header", () => { + const input = responsesInput(); + + const result = applyResponsesSessionAffinity(input); + + assert.equal(result.body.prompt_cache_key, "session-1111-2222"); + assert.equal(result.body.model, "gpt-5.1-codex"); + assert.equal(input.upstreamRequest.body.prompt_cache_key, undefined); +}); + +test("inbound metadata.user_id is carried onto the outbound Responses body", () => { + const result = applyResponsesSessionAffinity(responsesInput()); + + assert.deepEqual(result.body.metadata, { user_id: "user_abc123_account__session_11112222" }); +}); + +test("caller-supplied prompt_cache_key is never overwritten", () => { + const input = responsesInput(); + input.upstreamRequest.body.prompt_cache_key = "caller-key"; + + const result = applyResponsesSessionAffinity(input); + + assert.equal(result.body.prompt_cache_key, "caller-key"); +}); + +test("empty prompt_cache_key is treated as missing", () => { + const input = responsesInput(); + input.upstreamRequest.body.prompt_cache_key = " "; + + const result = applyResponsesSessionAffinity(input); + + assert.equal(result.body.prompt_cache_key, "session-1111-2222"); +}); + +test("session headers resolve case-insensitively and prefer x-claude-code-session-id", () => { + const input = responsesInput(); + input.request.headers = { + "X-Claude-Code-Session-ID": "code-session", + "x-claude-session-id": "legacy-session" + }; + + const result = applyResponsesSessionAffinity(input); + + assert.equal(result.body.prompt_cache_key, "code-session"); +}); + +test("x-claude-session-id and inbound metadata.user_id are fallback key sources", () => { + assert.equal( + resolveResponsesSessionKey({ "x-claude-session-id": "legacy-session" }, "metadata-user"), + "legacy-session" + ); + assert.equal(resolveResponsesSessionKey({}, "metadata-user"), "metadata-user"); + assert.equal(resolveResponsesSessionKey(undefined, undefined), undefined); +}); + +test("non-Responses providers and non-JSON bodies pass through untouched", () => { + const chatInput = responsesInput({ + targetProviderConfig: { type: "openai_chat_completions" } + }); + assert.equal(applyResponsesSessionAffinity(chatInput), chatInput.upstreamRequest); + + const bytesInput = responsesInput(); + bytesInput.upstreamRequest = { + ...bytesInput.upstreamRequest, + body: Buffer.from("{}"), + bodyEncoding: "bytes" + }; + assert.equal(applyResponsesSessionAffinity(bytesInput), bytesInput.upstreamRequest); +}); + +test("requests without any session key source pass through untouched", () => { + const input = responsesInput({ + request: { + body: { messages: [] }, + headers: { "content-type": "application/json" } + } + }); + + const result = applyResponsesSessionAffinity(input); + + assert.equal(result, input.upstreamRequest); +}); + +test("outbound metadata supplied by the caller is preserved", () => { + const input = responsesInput(); + input.upstreamRequest.body.metadata = { user_id: "caller-user" }; + input.upstreamRequest.body.prompt_cache_key = "caller-key"; + + const result = applyResponsesSessionAffinity(input); + + assert.equal(result, input.upstreamRequest); +}); + +test("gateway boundary plugin registers the session affinity hook", async () => { + const hooks = createGatewayPlugin().providerHooks; + const affinityHook = hooks.find((hook) => hook.key === "ccr-responses-session-affinity"); + assert.ok(affinityHook); + + const input = responsesInput(); + const result = await affinityHook.transformRequest(input); + + assert.equal(result.ok, true); + assert.equal(result.value.body.prompt_cache_key, "session-1111-2222"); +}); diff --git a/packages/core/test/unit/gateway/sse-utf8-chunk-boundary.test.mjs b/packages/core/test/unit/gateway/sse-utf8-chunk-boundary.test.mjs new file mode 100644 index 00000000..b71d99fc --- /dev/null +++ b/packages/core/test/unit/gateway/sse-utf8-chunk-boundary.test.mjs @@ -0,0 +1,114 @@ +import assert from "node:assert/strict"; +import { Readable } from "node:stream"; +import test from "node:test"; +import { codexMultiAgentBridgeResponseStream } from "@ccr/core/gateway/features/codex-multi-agent-bridge.ts"; +import { codexApplyPatchBridgeResponseStream } from "@ccr/core/gateway/features/codex-patch-bridge.ts"; +import { hostedWebSearchProtocolResponseStream } from "@ccr/core/gateway/features/hosted-web-search/index.ts"; + +const sseHeaders = () => new Headers({ "content-type": "text/event-stream" }); + +const webSearchRecords = [{ + engine: "test", + query: "北京天气", + results: [{ content: "多云", snippet: "多云", title: "天气", url: "https://example.test/weather" }], + searchUrl: "https://example.test/search" +}]; + +function hostedWebSearchContext(protocol) { + return { + maxUses: 1, + protocol, + queryHint: "北京天气", + records: webSearchRecords, + requestId: "req-utf8", + sinceMs: 0, + toolName: "web_search" + }; +} + +async function streamText(stream) { + const chunks = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + return Buffer.concat(chunks).toString("utf8"); +} + +// Feeds `text` through `makeStream` once per interior byte offset of the first +// multi-byte character, so at least one chunk boundary always lands inside it. +async function assertSurvivesEveryByteSplit(makeStream, text, marker) { + const bytes = Buffer.from(text, "utf8"); + const markerBytes = Buffer.from(marker, "utf8"); + const markerStart = bytes.indexOf(markerBytes); + assert.ok(markerStart >= 0, "marker must be present in the event"); + assert.ok(markerBytes.length > 1, "marker must be a multi-byte character"); + + for (let offset = 1; offset < markerBytes.length; offset += 1) { + const splitAt = markerStart + offset; + const output = await streamText(makeStream( + Readable.from([bytes.subarray(0, splitAt), bytes.subarray(splitAt)]) + )); + assert.equal( + output.includes("�"), + false, + `split at byte ${splitAt} produced U+FFFD: ${JSON.stringify(output)}` + ); + assert.equal(output, text, `split at byte ${splitAt} changed the event`); + } +} + +test("Codex multi-agent bridge SSE keeps multi-byte characters split across chunks", async () => { + await assertSurvivesEveryByteSplit( + (input) => codexMultiAgentBridgeResponseStream(input, sseHeaders()), + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"北京今天多云"}\n\n', + "北" + ); +}); + +test("Codex apply_patch bridge SSE keeps multi-byte characters split across chunks", async () => { + await assertSurvivesEveryByteSplit( + (input) => codexApplyPatchBridgeResponseStream(input, sseHeaders()), + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"北京今天多云"}\n\n', + "北" + ); +}); + +test("Hosted web search Anthropic SSE keeps multi-byte characters split across chunks", async () => { + await assertSurvivesEveryByteSplit( + (input) => hostedWebSearchProtocolResponseStream( + input, + sseHeaders(), + hostedWebSearchContext("anthropic_messages"), + undefined + ), + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"北京今天多云"}}\n\n', + "北" + ); +}); + +test("Hosted web search OpenAI Responses SSE keeps multi-byte characters split across chunks", async () => { + await assertSurvivesEveryByteSplit( + (input) => hostedWebSearchProtocolResponseStream( + input, + sseHeaders(), + hostedWebSearchContext("openai_responses"), + undefined + ), + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","output_index":0,"delta":"北京今天多云"}\n\n', + "北" + ); +}); + +test("Codex bridges keep emoji surrogate pairs split across chunks", async () => { + const event = 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"done 🎉"}\n\n'; + await assertSurvivesEveryByteSplit( + (input) => codexMultiAgentBridgeResponseStream(input, sseHeaders()), + event, + "🎉" + ); + await assertSurvivesEveryByteSplit( + (input) => codexApplyPatchBridgeResponseStream(input, sseHeaders()), + event, + "🎉" + ); +}); diff --git a/packages/core/test/unit/media/media-executor-download.test.mjs b/packages/core/test/unit/media/media-executor-download.test.mjs new file mode 100644 index 00000000..34eb302b --- /dev/null +++ b/packages/core/test/unit/media/media-executor-download.test.mjs @@ -0,0 +1,76 @@ +import assert from "node:assert/strict"; +import http from "node:http"; +import test from "node:test"; +import { GatewayMediaExecutor } from "@ccr/core/media/executors.ts"; + +// Starts a loopback server that reports a large content-length but never ends +// the body, so the client keeps the connection open unless it explicitly +// cancels the response body. `closed.fired` flips once the upstream socket is +// torn down, which only happens when `download()` releases the body. +function stalledArtifactServer(statusCode) { + return new Promise((resolve) => { + const closed = { fired: false }; + const server = http.createServer((request, response) => { + request.on("close", () => { + closed.fired = true; + }); + response.writeHead(statusCode, { + "content-length": String(300 * 1024 * 1024), + "content-type": "application/octet-stream" + }); + response.write(Buffer.alloc(16)); + }); + server.listen(0, "127.0.0.1", () => { + resolve({ closed, port: server.address().port, server }); + }); + }); +} + +function loopbackExecutor(port) { + return new GatewayMediaExecutor( + { + model: "test-model", + protocol: "openai", + providerBaseUrl: `http://127.0.0.1:${port}`, + providerName: "test-provider" + }, + {} + ); +} + +async function connectionClosedWithin(closed, timeoutMs) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (closed.fired) return true; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return closed.fired; +} + +test("download releases the response body when the declared artifact is too large", async () => { + const { closed, port, server } = await stalledArtifactServer(200); + const executor = loopbackExecutor(port); + try { + await assert.rejects( + executor.download({ fileName: "artifact.bin", remoteUrl: `http://127.0.0.1:${port}/artifact` }, new AbortController().signal), + /exceeds the 250 MB limit/ + ); + assert.equal(await connectionClosedWithin(closed, 2000), true, "expected the upstream response body to be cancelled"); + } finally { + server.close(); + } +}); + +test("download releases the response body on a non-ok status", async () => { + const { closed, port, server } = await stalledArtifactServer(404); + const executor = loopbackExecutor(port); + try { + await assert.rejects( + executor.download({ fileName: "artifact.bin", remoteUrl: `http://127.0.0.1:${port}/artifact` }, new AbortController().signal), + /HTTP 404/ + ); + assert.equal(await connectionClosedWithin(closed, 2000), true, "expected the upstream response body to be cancelled"); + } finally { + server.close(); + } +}); diff --git a/packages/core/test/unit/usage/usage-normalization.test.mjs b/packages/core/test/unit/usage/usage-normalization.test.mjs index 419f1ce2..5f0eee5e 100644 --- a/packages/core/test/unit/usage/usage-normalization.test.mjs +++ b/packages/core/test/unit/usage/usage-normalization.test.mjs @@ -38,6 +38,96 @@ test("normalizeUsageInputTokens keeps Anthropic input tokens unchanged", () => { }); }); +test("normalizeUsageInputTokens ignores the upstream protocol for a translated body", () => { + // Anthropic response served from an OpenAI upstream: the body already excludes + // the cached prefix, so the upstream's cache-inclusive rule must not apply. + const usage = normalizeUsageInputTokens( + { + cacheReadTokens: 3584, + inputIncludesCacheTokens: false, + inputTokens: 250, + outputTokens: 40 + }, + { + path: "/v1/messages", + providerProtocol: "openai_chat_completions", + source: "responseBody" + } + ); + + assert.equal(usage?.inputTokens, 250); +}); + +test("normalizeUsageInputTokens does not clamp a translated body to zero", () => { + // The failure this guards against: subtracting an already-excluded cached + // prefix drives input tokens negative, and the clamp reports it as 0. + const usage = normalizeUsageInputTokens( + { + cacheReadTokens: 260608, + inputIncludesCacheTokens: false, + inputTokens: 1888 + }, + { + path: "/v1/messages", + providerProtocol: "openai_chat_completions", + source: "responseBody" + } + ); + + assert.equal(usage?.inputTokens, 1888); +}); + +test("normalizeUsageInputTokens subtracts on billing headers from the same response", () => { + // Same response, other source: the gateway restates the upstream's own + // counters, which do include the cached prefix. + const usage = normalizeUsageInputTokens( + { + cacheReadTokens: 3584, + inputTokens: 3834, + outputTokens: 40 + }, + { + path: "/v1/messages", + providerProtocol: "openai_chat_completions", + source: "providerBilling" + } + ); + + assert.equal(usage?.inputTokens, 250); +}); + +test("normalizeUsageInputTokens still subtracts for a same-format body", () => { + // No translation: an OpenAI body from an OpenAI upstream is cache-inclusive, + // and the body's own fields say so. + const usage = normalizeUsageInputTokens( + { + cacheReadTokens: 20, + inputIncludesCacheTokens: true, + inputTokens: 100 + }, + { path: "/v1/chat/completions", source: "responseBody" } + ); + + assert.equal(usage?.inputTokens, 80); +}); + +test("normalizeUsageInputTokens uses the path when a body declares no convention", () => { + assert.equal( + normalizeUsageInputTokens( + { cacheReadTokens: 8, inputTokens: 50 }, + { path: "/v1/messages", source: "responseBody" } + )?.inputTokens, + 50 + ); + assert.equal( + normalizeUsageInputTokens( + { cacheReadTokens: 8, inputTokens: 50 }, + { path: "/v1/chat/completions", source: "responseBody" } + )?.inputTokens, + 42 + ); +}); + test("normalizeUsageInputTokens falls back to path and usage hints", () => { assert.equal( normalizeUsageInputTokens( diff --git a/packages/ui/src/pages/home/shared/log-body-worker-protocol.ts b/packages/ui/src/pages/home/shared/log-body-worker-protocol.ts index 6bbbebfc..e731d581 100644 --- a/packages/ui/src/pages/home/shared/log-body-worker-protocol.ts +++ b/packages/ui/src/pages/home/shared/log-body-worker-protocol.ts @@ -114,7 +114,16 @@ export function formatLogBodyForWorker( } { const large = isLargeLogBody(body, largeTextThreshold); const preview = large && mode !== "full"; - const formattedBodyView = formatLogBodyView(body); + // Large-body preview: only skip the full JSON parse/pretty-print when the actual + // text is over-length. Otherwise the worker does JSON.parse on a huge object graph + // and pretty-prints + structured-clones it back, blowing up memory/CPU and surfacing + // "Body formatter worker failed." (issue #1694). + // The judge is the real body.text length, not sizeBytes (sizeBytes may be inflated + // storage metadata; a preview:true lightweight JSON should still show its JSON tree). + const previewText: FormattedLogBody | undefined = preview && (body?.text?.length ?? 0) > previewTextLimit + ? { text: createLogBodyPreviewText(body, previewTextLimit) } + : undefined; + const formattedBodyView = previewText ?? formatLogBodyView(body); const bodyView = preview && formattedBodyView.json === undefined ? { text: createLogBodyPreviewText(body, previewTextLimit) } : formattedBodyView; diff --git a/packages/ui/src/pages/home/shared/providers.ts b/packages/ui/src/pages/home/shared/providers.ts index b3d8730e..8a40e0d8 100644 --- a/packages/ui/src/pages/home/shared/providers.ts +++ b/packages/ui/src/pages/home/shared/providers.ts @@ -20,6 +20,7 @@ import type { ProviderAccountConfig, ProviderAccountConnectorConfig, ProviderAccountHttpJsonConnectorConfig, + ProviderAccountMappedMeterConfig, ProviderAccountStandardConnectorConfig, ProviderAccountWebContentJsonConnectorConfig, ProviderCredentialConfig, @@ -1215,7 +1216,7 @@ export function createProviderAccountDraftFromConfig(account: ProviderAccountCon accountRefreshIntervalMs: account.refreshIntervalMs ? String(account.refreshIntervalMs) : "" }; } - if (jsonConnector.parser) { + if (jsonConnector.parser || flatDraftDropsMeters(jsonConnector.mapping.meters)) { return { ...base, accountConnectorsText: JSON.stringify(connectors, null, 2), @@ -1259,6 +1260,22 @@ export function createProviderAccountDraftFromConfig(account: ProviderAccountCon }; } +function isBalanceDraftMeter(meter: ProviderAccountMappedMeterConfig): boolean { + return meter.kind === "balance" || meter.id === "balance"; +} + +function isSubscriptionDraftMeter(meter: ProviderAccountMappedMeterConfig): boolean { + return meter.kind === "subscription" || meter.id === "subscription" || meter.kind === "quota" || meter.kind === "tokens" || meter.kind === "time_window"; +} + +function flatDraftDropsMeters(meters: ProviderAccountMappedMeterConfig[]): boolean { + const balanceMeters = meters.filter(isBalanceDraftMeter); + const subscriptionMeters = meters.filter((meter) => !isBalanceDraftMeter(meter) && isSubscriptionDraftMeter(meter)); + return balanceMeters.length > 1 + || subscriptionMeters.length > 1 + || balanceMeters.length + subscriptionMeters.length !== meters.length; +} + function mappedStringDraftValue(value: string | string[] | undefined): string { if (Array.isArray(value)) { return value.find((item) => stringValue(item)) ?? ""; diff --git a/packages/ui/test/unit/network-format.test.ts b/packages/ui/test/unit/network-format.test.ts index 4d6ab59b..adf08716 100644 --- a/packages/ui/test/unit/network-format.test.ts +++ b/packages/ui/test/unit/network-format.test.ts @@ -50,3 +50,32 @@ test("request log preview bodies still format parseable JSON as JSON", () => { }); assert.match(view.text, /"model": "test-model"/); }); + +test("preview mode truncates an over-large JSON text body instead of parsing it", () => { + const hugeText = JSON.stringify({ + model: "test-model", + messages: Array.from({ length: 4000 }, (_, i) => ({ + role: i % 2 === 0 ? "user" : "assistant", + content: `message body number ${i} `.repeat(20) + })) + }); + // Far beyond previewTextLimit, simulating a real ~308KB request body (issue #1694) + assert.ok(hugeText.length > 300 * 1024, "fixture should exceed 300KB"); + + const body: RequestLogBody = { + bodyRef: "large-json-body", + contentType: "application/json", + encoding: "utf8", + preview: true, + sizeBytes: hugeText.length, + text: hugeText, + truncated: false + }; + + const view = formatLogBodyForWorker(body, "preview", 256 * 1024, 160 * 1024); + assert.equal(view.preview, true); + // Over-length text is truncated instead of fully parsed/pretty-printed, avoiding the worker crash + assert.equal(view.json, undefined); + assert.ok(view.text.length < hugeText.length, "should be truncated"); + assert.match(view.text, /characters omitted from preview/); +});