diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 45eb39e9585..f9b80943a41 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -15,6 +15,7 @@ import { } from "./admission-model-scope"; import { jsonUtf8Bytes } from "../lib/json-byte-size"; import { redactSecretString } from "../lib/redact"; +import { formatAnthropicErrorBody } from "../adapters/anthropic"; import { sseFieldValue } from "../lib/sse-decoder"; import { enforceAnthropicImageLimits, sniffImageDimensions } from "../adapters/anthropic-image-guard"; import { normalizeAnthropicImages } from "../adapters/anthropic-image-normalize"; @@ -555,6 +556,15 @@ async function anthropicNativePassthrough( logCtx.provider = "anthropic-native"; logCtx.requestedModel = model; const finalize = createFinalRequestLog(logIds, logCtx).finish; + // Every local answer carries its reason into the row, as the managed native lane's + // finishLog does; without it a failed passthrough row has no upstreamError at all. Like + // that lane's fail(), the client and the log both get the redacted message. + const fail = (status: number, closeReason: PassthroughCloseReason | "non_stream", message: string, type: string) => { + const safeMessage = redactSecretString(message); + logCtx.upstreamError = safeMessage.slice(0, 500); + finalize(status, { closeReason }); + return anthropicErrorResponse(status, safeMessage, type); + }; const base = (config.claudeCode?.anthropicBaseUrl ?? "https://api.anthropic.com").replace(/\/$/, ""); const search = new URL(req.url).search; @@ -581,13 +591,11 @@ async function anthropicNativePassthrough( req.signal, ); if (result.kind === "timeout") { - finalize(504, { closeReason: "non_stream" }); - return anthropicErrorResponse(504, "anthropic passthrough timed out waiting for response headers", "timeout_error"); + return fail(504, "non_stream", "anthropic passthrough timed out waiting for response headers", "timeout_error"); } if (result.kind === "error") { const err = result.error; - finalize(502, { closeReason: "non_stream" }); - return anthropicErrorResponse(502, `anthropic passthrough failed: ${err instanceof Error ? err.message : String(err)}`, "api_error"); + return fail(502, "non_stream", `anthropic passthrough failed: ${err instanceof Error ? err.message : String(err)}`, "api_error"); } const upstream = result.upstream; @@ -607,16 +615,13 @@ async function anthropicNativePassthrough( // idle/size bounds — headers are NOT yet sent here, so real statuses are available. const bodyResult = await readBoundedPassthroughBody(upstream, bodyGuard); if (bodyResult.kind === "client_cancel") { - finalize(499, { closeReason: "client_cancel" }); - return anthropicErrorResponse(499, "client closed request during anthropic passthrough", "api_error"); + return fail(499, "client_cancel", "client closed request during anthropic passthrough", "api_error"); } if (bodyResult.kind === "stall") { - finalize(504, { closeReason: "body_stall" }); - return anthropicErrorResponse(504, `anthropic passthrough body stalled: no upstream bytes for ${Math.round(bodyGuard.stallMs / 1000)}s`, "timeout_error"); + return fail(504, "body_stall", `anthropic passthrough body stalled: no upstream bytes for ${Math.round(bodyGuard.stallMs / 1000)}s`, "timeout_error"); } if (bodyResult.kind === "overflow") { - finalize(502, { closeReason: "body_overflow" }); - return anthropicErrorResponse(502, `anthropic passthrough body exceeded ${bodyGuard.maxBytes} bytes`, "api_error"); + return fail(502, "body_overflow", `anthropic passthrough body exceeded ${bodyGuard.maxBytes} bytes`, "api_error"); } const text = bodyResult.text; if (upstream.ok) { @@ -624,6 +629,16 @@ async function anthropicNativePassthrough( const parsed = JSON.parse(text) as { usage?: Rec }; if (isRec(parsed?.usage)) logCtx.usage = anthropicUsageToOcx(parsed.usage); } catch { /* count_tokens etc. */ } + } else if (upstream.status >= 400) { + // The body is relayed verbatim; the row gets a "Provider error" reason from the same + // redacting formatAnthropicErrorBody the managed lane falls back to. Only a body of at + // most 64 Ki characters (the order of that lane's 64 KiB error read bound) is parsed for + // it: the body cap here is far larger, and JSON.parse plus redaction over a huge body + // would block the event loop. Characters, not bytes, are what that cost scales with. + const detail = text.length <= PASSTHROUGH_ERROR_DETAIL_MAX_CHARS + ? formatAnthropicErrorBody(upstream.status, upstream.headers, text) + : ""; + logCtx.upstreamError = detail ? `Provider error ${upstream.status}: ${detail}` : `Provider error ${upstream.status}`; } finalize(upstream.status, { closeReason: "non_stream" }); const retryAfter = upstream.headers.get("retry-after"); @@ -633,6 +648,7 @@ async function anthropicNativePassthrough( }); } +const PASSTHROUGH_ERROR_DETAIL_MAX_CHARS = 64 * 1024; const DEFAULT_BODY_STALL_SEC = 90; const DEFAULT_BODY_MAX_BYTES = 64 * 1024 * 1024; diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index e256089a2a8..a500fc71b0a 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -466,4 +466,6 @@ Native Anthropic passthrough in `src/server/claude-messages.ts` forwards the cal `tapAnthropicSseForLog` in `src/server/claude-messages.ts` relays the streamed body of both the native passthrough and the managed native Messages lane (`src/server/messages-native.ts`). The response headers are already sent, so a stall, a byte-cap overflow, or an upstream read failure ends the body with an Anthropic `event: error` frame after a blank-line boundary and a clean close: `timeout_error` for an idle stall, `api_error` for the byte cap, and `api_error` when an upstream read fails mid-stream (a socket reset). The mid-stream reset is logged like the Responses relay's read error: status 502, `terminalStatus: "failed"`, `closeReason: "terminal"`, `transportPhase: "mid_stream"`, a synthetic terminal source, the attempt marked `streamAborted`, the redacted reason in `upstreamError`, and the usage seen before the reset. The non-streaming fold in the managed lane closes its row with the tap's meta for a reset, a stall or the byte cap, so its row matches the streaming lane's. The request is not replayed. The logged status is not uniform across these frames: a stall and the byte cap keep status 200 with `closeReason` `body_stall` or `body_overflow`, which classify as `incomplete`, while a reset is a 502 that classifies as `failed`. Some read failures are not upstream failures. When the cancel signal is already aborted, the rejection is a `499` client cancel, because Bun can reject the read before it dispatches the abort listener. The managed lane passes its upstream controller's signal, so shutdown and turn release count as cancels too. When `message_stop` or an upstream `error` event has been seen, including one still in the buffer without its blank-line delimiter, the turn is complete: it logs 200 and closes with no error frame. A terminal found only in that unterminated tail gets its blank line restored, because an SSE parser drops an event that EOF cuts off. A translator budget overflow is a local cap, so it still errors the stream, and the non-streaming fold answers it with 413. `tests/claude-integration/claude-native-passthrough.test.ts` and `tests/claude-integration/messages-native.test.ts` cover both lanes against an upstream that resets after a partial or a complete body, plus both cancel paths. +A native passthrough answered without a stream records its reason in `upstreamError` as well, so the row and the failure diagnostics in usage.jsonl name the cause. An upstream error response (status 400 or above) is relayed verbatim and logged as `Provider error : : `, built by the same redacting, bounded `formatAnthropicErrorBody` in `src/adapters/anthropic.ts` that the managed lane falls back to. A body without an Anthropic error, or one over 64 Ki characters (the order of the managed lane's 64 KiB error read bound), logs `Provider error `. A header timeout, fetch failure, body stall, byte-cap overflow or client cancel logs the message the client receives; both are redacted, as in the managed lane. With a reason present, `requestLogErrorCode` classifies the row the way it does for the other lanes; for example, an authentication-shaped 403 becomes `invalid_api_key` instead of `permission_denied`. `tests/claude-integration/claude-native-passthrough.test.ts` covers each case. + Linked-machine data uses the [connection-bound relay contract](../remote-link.md#connection-bound-relay-authentication); client-local credentials and routing policy remain unchanged. diff --git a/tests/claude-integration/claude-native-passthrough.test.ts b/tests/claude-integration/claude-native-passthrough.test.ts index c01425f744e..5f147b032b3 100644 --- a/tests/claude-integration/claude-native-passthrough.test.ts +++ b/tests/claude-integration/claude-native-passthrough.test.ts @@ -12,6 +12,7 @@ import type { OcxConfig } from "../../src/types"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { startTruncatedSseUpstream } from "../helpers/truncated-sse-upstream"; +import { readRecentUsageEntries } from "../../src/usage/log"; import { tapAnthropicSseForLog } from "../../src/server/claude-messages"; import type { RequestLogContext } from "../../src/server/request-log"; import { TranslatorBudgetExceededError } from "../../src/lib/translator-budget"; @@ -968,3 +969,238 @@ test("a terminal frame still in the buffer when the read fails counts as seen", expect(calls).toEqual([{ status: 200, closeReason: "terminal" }]); expect(logCtx.usage).toEqual(expect.objectContaining({ inputTokens: 12, outputTokens: 5 })); }); + +// --- A failed passthrough keeps its reason in the request log --- + +async function nativeFailureRow(upstreamBase: string, extraClaude?: Record, overrides?: Partial) { + const { clearRequestLogsForTests } = await import("../../src/server/request-log"); + clearRequestLogsForTests(); + saveConfig({ ...cfg(upstreamBase, extraClaude), ...overrides }); + const server = startServer(0); + try { + const res = await fetch(new URL("/v1/messages?beta=true", server.url), { + method: "POST", + headers: OAUTH_HEADERS, + body: JSON.stringify({ ...claudeBody(), stream: false }), + }); + const text = await res.text(); + const logs = logsFromApiBody<{ status?: number; closeReason?: string; upstreamError?: string; errorCode?: string }>( + await (await fetch(new URL("/api/logs?tail=1", server.url))).json(), + ); + expect(logs).toHaveLength(1); + return { res, text, row: logs[0]! }; + } finally { + await server.stop(true); + } +} + +test("an upstream error response is relayed verbatim and its reason reaches the request log and usage.jsonl", async () => { + const errorBody = { type: "error", error: { type: "invalid_request_error", message: "prompt is too long: 213000 tokens > 200000 maximum" }, request_id: "req_fixture" }; + const upstream = Bun.serve({ port: 0, fetch: () => Response.json(errorBody, { status: 400 }) }); + try { + const { res, text, row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, "")); + expect(res.status).toBe(400); + expect(JSON.parse(text)).toEqual(errorBody); + expect(row.status).toBe(400); + expect(row.upstreamError).toBe("Provider error 400: invalid_request_error: prompt is too long: 213000 tokens > 200000 maximum"); + // usage.jsonl carries failure diagnostics for >= 400 rows; this row used to persist none. + expect(readRecentUsageEntries(1)[0]?.upstreamError).toBe(row.upstreamError); + } finally { + upstream.stop(true); + } +}); + +test("a non-JSON upstream error still logs its status as the reason", async () => { + const upstream = Bun.serve({ port: 0, fetch: () => new Response("bad gateway", { status: 502, headers: { "content-type": "text/html" } }) }); + try { + const { res, row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, "")); + expect(res.status).toBe(502); + expect(row.upstreamError).toBe("Provider error 502"); + } finally { + upstream.stop(true); + } +}); + +test("a credential echoed in an upstream error message is redacted in the log", async () => { + const leaked = "sk-ant-api03-" + "Z".repeat(40); + const upstream = Bun.serve({ port: 0, fetch: () => Response.json({ type: "error", error: { type: "authentication_error", message: `invalid x-api-key ${leaked}` } }, { status: 401 }) }); + try { + const { row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, "")); + expect(row.status).toBe(401); + expect(row.upstreamError).toStartWith("Provider error 401: authentication_error: invalid x-api-key "); + expect(row.upstreamError).not.toContain(leaked); + } finally { + upstream.stop(true); + } +}); + +test("an unreachable upstream logs the fetch failure as the reason", async () => { + // Stubbed like the reject-path activation test in claude-messages-endpoint.test.ts: a real + // refusal can take longer than cfg()'s 250 ms header deadline on some platforms. + const refusedOrigin = "http://127.0.0.1:1"; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(typeof input === "string" ? input : input instanceof URL ? input.href : input.url); + if (url.origin === refusedOrigin) throw Object.assign(new TypeError("connect ECONNREFUSED"), { code: "ECONNREFUSED" }); + return originalFetch(input, init); + }) as typeof globalThis.fetch; + try { + const { res, text, row } = await nativeFailureRow(refusedOrigin); + expect(res.status).toBe(502); + expect(row.status).toBe(502); + expect(row.upstreamError).toBe("anthropic passthrough failed: connect ECONNREFUSED"); + expect(JSON.parse(text).error.message).toBe(row.upstreamError); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("a credential in a local failure message is redacted for the client as well as the log", async () => { + // The managed lane's fail() redacts both; a future runtime could put a URL or key in a fetch error. + const refusedOrigin = "http://127.0.0.1:1"; + const leaked = "sk-ant-api03-" + "Q".repeat(40); + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(typeof input === "string" ? input : input instanceof URL ? input.href : input.url); + if (url.origin === refusedOrigin) throw new TypeError(`proxy rejected key ${leaked}`); + return originalFetch(input, init); + }) as typeof globalThis.fetch; + try { + const { res, text, row } = await nativeFailureRow(refusedOrigin); + expect(res.status).toBe(502); + expect(text).not.toContain(leaked); + expect(row.upstreamError).toStartWith("anthropic passthrough failed: proxy rejected key "); + expect(row.upstreamError).not.toContain(leaked); + expect(JSON.parse(text).error.message).toBe(row.upstreamError); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("an upstream that never sends headers logs the header timeout as the reason", async () => { + const silent = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() { /* accept, never answer */ } } }); + try { + const { res, row } = await nativeFailureRow(`http://127.0.0.1:${silent.port}`); + expect(res.status).toBe(504); + expect(row.status).toBe(504); + expect(row.upstreamError).toBe("anthropic passthrough timed out waiting for response headers"); + } finally { + silent.stop(true); + } +}); + +test("a non-stream body over the byte cap logs the cap as the reason", async () => { + const upstream = Bun.serve({ port: 0, fetch: () => Response.json({ type: "message", content: [{ type: "text", text: "x".repeat(4096) }] }) }); + try { + const { res, row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, ""), { bodyMaxBytes: 1024 }); + expect(res.status).toBe(502); + expect(row).toMatchObject({ status: 502, closeReason: "body_overflow" }); + expect(row.upstreamError).toBe("anthropic passthrough body exceeded 1024 bytes"); + } finally { + upstream.stop(true); + } +}); + +test("an oversized upstream error body is relayed verbatim but not parsed for the log reason", async () => { + // 64 Ki characters: the order of the managed native lane's 64 KiB error read bound. + const errorBody = { type: "error", error: { type: "invalid_request_error", message: "x".repeat(70 * 1024) } }; + const upstream = Bun.serve({ port: 0, fetch: () => Response.json(errorBody, { status: 400 }) }); + try { + const { res, text, row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, "")); + expect(res.status).toBe(400); + expect(JSON.parse(text)).toEqual(errorBody); + expect(row.upstreamError).toBe("Provider error 400"); + } finally { + upstream.stop(true); + } +}); + +test("a 3xx answer is not logged with an error reason", async () => { + const upstream = Bun.serve({ port: 0, fetch: () => new Response("choose", { status: 300 }) }); + try { + const { res, row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, "")); + expect(res.status).toBe(300); + expect(row.status).toBe(300); + expect(row.upstreamError).toBeUndefined(); + } finally { + upstream.stop(true); + } +}); + +test("a 403 reason now drives the row's error code like the other lanes", async () => { + const upstream = Bun.serve({ port: 0, fetch: () => Response.json({ type: "error", error: { type: "permission_error", message: "OAuth authentication is currently not supported." } }, { status: 403 }) }); + try { + const { row } = await nativeFailureRow(upstream.url.toString().replace(/\/$/, "")); + expect(row.status).toBe(403); + expect(row.upstreamError).toBe("Provider error 403: permission_error: OAuth authentication is currently not supported."); + // Without a reason every passthrough 403 was permission_denied. + expect(row.errorCode).toBe("invalid_api_key"); + } finally { + upstream.stop(true); + } +}); + +/** + * Sends a 200 status line and headers promising a JSON body, then nothing. Bun.serve holds the + * headers of a streamed body until its first chunk, which would trip the header deadline instead. + */ +function startHeadersOnlyUpstream(onRequest?: () => void): { port: number; stop: () => void } { + const listener = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + data(socket) { + onRequest?.(); + socket.write("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 64\r\n\r\n"); + }, + }, + }); + return { port: listener.port, stop: () => listener.stop(true) }; +} + +test("a non-stream body that stalls logs the stall as the reason", async () => { + const upstream = startHeadersOnlyUpstream(); + try { + const { res, row } = await nativeFailureRow(`http://127.0.0.1:${upstream.port}`, { bodyStallSec: 1 }); + expect(res.status).toBe(504); + expect(row).toMatchObject({ status: 504, closeReason: "body_stall" }); + expect(row.upstreamError).toBe("anthropic passthrough body stalled: no upstream bytes for 1s"); + } finally { + upstream.stop(); + } +}); + +test("a client that leaves while a non-stream body is pending logs the cancel as the reason", async () => { + const { clearRequestLogsForTests } = await import("../../src/server/request-log"); + clearRequestLogsForTests(); + let reached!: () => void; + const upstreamReached = new Promise(resolve => { reached = resolve; }); + const upstream = startHeadersOnlyUpstream(() => reached()); + saveConfig(cfg(`http://127.0.0.1:${upstream.port}`)); + const server = startServer(0); + try { + const client = new AbortController(); + const pending = fetch(new URL("/v1/messages?beta=true", server.url), { + method: "POST", + headers: OAUTH_HEADERS, + body: JSON.stringify({ ...claudeBody(), stream: false }), + signal: client.signal, + }).catch(() => undefined); + await upstreamReached; + await Bun.sleep(50); + client.abort(); + await pending; + let row: { status?: number; closeReason?: string; upstreamError?: string } | undefined; + for (let i = 0; i < 100 && !row; i++) { + row = logsFromApiBody<{ status?: number; closeReason?: string; upstreamError?: string }>( + await (await fetch(new URL("/api/logs?tail=1", server.url))).json(), + )[0]; + if (!row) await Bun.sleep(20); + } + expect(row).toMatchObject({ status: 499, closeReason: "client_cancel" }); + expect(row!.upstreamError).toBe("client closed request during anthropic passthrough"); + } finally { + await server.stop(true); + upstream.stop(); + } +});