Merge branch 'main' into dev/3.1

2026-08-28 19:01:32 +08:00 · 2026-08-24 20:02:57 +08:00
parent 70e7fd1a80 17c8744de9
commit f2360ac046
21 changed files with 1179 additions and 55 deletions
@@ -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",
@@ -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<string, string>;
+  method?: string;
+  url: string;
+};
+
+export type ResponsesSessionAffinityInput = {
+  request?: {
+    body?: unknown;
+    headers?: Record<string, HeaderValue>;
+  };
+  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<string, unknown> = {};
+  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<string, HeaderValue> | 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<string, HeaderValue> | 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;
+}
@@ -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)
+        };
+      }
    }]
  };
 }
@@ -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<string, unknown>): { 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 = "";
@@ -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 = "";
@@ -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) {
@@ -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<string, unknown>, 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<string, unknown>, 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<string | undefined> {
+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<ImageInputToUrlResult> {
  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|<svg)/i;
+
+/**
+ * Validate and repair a base64 image payload. Returns undefined when the bytes are
+ * not usable: non-base64 characters, an empty decode, or a length mod 4 of 1, which
+ * means the image was cut off mid-stream and padding cannot restore it. A remainder
+ * of 2 or 3 is fixed by adding `=` padding. SVG/XML payloads are flagged from the
+ * decoded head so the caller can reject them with a precise reason, and the raster
+ * format is sniffed so the caller can label the data URL with a supported type.
+ */
+function normalizeImagePayload(value: string): { payload: string; svg: boolean; sniffedType?: string } | undefined {
+  if (!value || !base64PayloadPattern.test(value) || value.length % 4 === 1) {
+    return undefined;
+  }
+  const padded = value.padEnd(value.length + ((4 - value.length % 4) % 4), "=");
+  const buffer = Buffer.from(padded, "base64");
+  if (buffer.byteLength === 0) {
+    return undefined;
+  }
+  return {
+    payload: padded,
+    svg: svgHeadPattern.test(buffer.subarray(0, 64).toString("utf8")),
+    sniffedType: sniffImageType(buffer)
+  };
+}
+
+/** Detect a raster format from its leading bytes; only the supported types are returned. */
+function sniffImageType(buffer: Buffer): string | undefined {
+  if (buffer.length >= 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";
@@ -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;
@@ -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<UsageSnapshot>(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<UsageSnapshot>(bodyUsage, {
+          path: usagePath,
+          source: "responseBody"
+        })
+      ) ?? {};
      if (hasUsageNumbers(usage)) {
        const inputTokens = normalizeCount(usage.inputTokens);
        const outputTokens = normalizeCount(usage.outputTokens);
@@ -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 = {
@@ -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<T extends UsageTokenAccounting>(
  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<T extends UsageTokenAccounting>(

 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;
@@ -220,11 +220,20 @@ export class UsageStore {
  async recordCapture(input: UsageCaptureInput): Promise<void> {
    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);
@@ -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, '<svg xmlns="http://www.w3.org/2000/svg" width="8" height="8"/>', "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) => {
@@ -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;
@@ -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");
+});
@@ -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,
+    "🎉"
+  );
+});
@@ -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();
+  }
+});
@@ -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(
@@ -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;
@@ -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)) ?? "";
@@ -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/);
+});