/** * Anthropic Messages inbound (/v1/messages + /v1/messages/count_tokens) for Claude Code. * * Translate-and-replay (devlog/260711_claude_inbound/010): the Anthropic request is * converted to a /v1/responses body and replayed through handleResponses on an * internal Request, so routing/OAuth/account-pool/failover/sidecars are inherited * unchanged. The Responses output (SSE or JSON) is converted back to Anthropic shape. */ import { FORWARD_HEADERS } from "../adapters/openai-responses"; import { admissionModelDeniedResponse, AdmissionModelDeniedError, assertRouteAllowedByScope, resolveAdmissionModelScope, } from "./admission-model-scope"; import { jsonUtf8Bytes } from "../lib/json-byte-size"; import { redactSecretString } from "../lib/redact"; import { enforceAnthropicImageLimits } from "../adapters/anthropic-image-guard"; import { normalizeAnthropicImages } from "../adapters/anthropic-image-normalize"; import { openAIChatSerializesThinking } from "../adapters/openai-chat/messages"; import { messagesToResponsesTranslation } from "../protocols/codecs/messages"; import { messagesDirectPlanHolds, messagesDirectRequestOption, messagesDirectTraceDecline, planMessagesAntigravityDirect, type MessagesDirectPlan, } from "./messages-antigravity-direct"; import { AnthropicRequestError, callerCacheRetention, DesktopModelMappingUnavailableError, extractOcxEffortDirective, extractOcxRouteDirective, nativeAnthropicProjection, resolveInboundModel, type ClaudeCacheKeySource } from "../claude/inbound"; import { nativeReasoningTag, type NativeReasoningOwner } from "../responses/reasoning-replay-cache"; import { isKnownDesktop3pModelId, resolveDesktop3pAlias } from "../claude/desktop-3p"; import { resolveAlias, claudeCodeNativeAlias, legacyAliasForNative } from "../claude/alias"; import { recordDesktopRequest } from "../claude/desktop-health"; import { stripOneMillionMarker } from "../claude/context-windows"; import { captureClaudeInbound } from "../claude/inbound-debug"; import { declaredAnthropicServerTools, replayedAnthropicServerBlocks } from "../claude/inbound-content-options"; import { claudeCodeForIngress } from "../claude/intercept/model-bindings"; import { classifyInterceptClient } from "../claude/intercept/client-class"; import { analyzeClaudeCompatibility, isClaudeCompatibilityMode } from "../claude/compatibility"; import { carriesMessageThread, messageThreadUnsupportedResponse } from "../claude/message-threads"; import { applyReplayRefusalClientHeaders, carryReplayRefusal, isReplayRefusalResponse, isTransientUpstreamStatus, REPLAY_REFUSED_STATUS, UPSTREAM_RESET_REPLAY_REFUSED_CODE, } from "../lib/upstream-retry"; import { isClientClosedMessage, recognizedUpstreamError, UNDECLARED_TOOL_CALL_ERROR_CODE } from "../lib/errors"; import { resolveClientRetryAfter } from "../lib/retry-after"; import { anthropicRateLimitHeaders } from "./anthropic-rate-limit-headers"; import { upstreamMessagesRequestIdHeaders } from "./messages-response-headers"; import { anthropicErrorBody, anthropicErrorResponse, collectAnthropicMessage, responsesJsonToAnthropicMessage, responsesSseToAnthropicSse, } from "../claude/outbound"; import { clearableDeadline, idleDeadline } from "../lib/abort"; import { CLAUDE_NATIVE_THINKING, type ClaudeThinkingProjection, } from "../lib/claude-request-projection"; import { captureRouteStaticPolicy, NoEligiblePolicyCandidateError, previewRouteModel, routedProviderConfig, UnknownRoutingPolicyError, routeModel, type RouteResult } from "../router"; import { evidenceFromBody } from "../routing/request-evidence"; import { resolveWireProtocolOverride } from "./adapter-resolve"; import type { OcxConfig } from "../types"; import { readJsonRequestBody, resolveInboundBodyLimitBytes } from "./request-decompress"; import { addFinalRequestLog, httpStatusForRequestLogTerminal, recordFirstOutput, type RequestLogContext } from "./request-log"; import { createFinalRequestLog } from "./inference/final-log"; import { conversationIdFromClaudeMetadata, getOrAllocateRequestSessionLane, linkRequestSessionLane, normalizeLogConversationId, sessionLaneIdFromRequest, } from "./request-log-conversation"; import { responseWithDeferredRequestLog } from "./relay"; import { clientWireOf } from "./inference/client-wire"; import { directEncodersApply } from "./inference/client-encoder-delivery"; import type { ClientEncoderOption } from "./responses/core-options"; import { handleResponses } from "./responses"; import { previewXaiOauthWireModel } from "./responses/core-normalize"; import { upstreamWireForAdapter } from "../protocols/contract"; import { createProtocolEnvelope, type ProtocolEnvelope } from "../protocols/envelope"; import { featuresFromMessagesBody, type ProtocolFeature } from "../protocols/features"; import { credentialDomainFor, messagesBodyHasOpaqueState } from "../protocols/opaque-state"; import { anthropicRoutingFor, anthropicSessionKeyFromParts } from "../oauth/anthropic-routing"; import { configuredAnthropicInstance } from "../providers/anthropic-instance"; import { messagesSelectorTargetsSecondaryInstance, messagesSecondaryInstanceUnavailable } from "./messages-native-selector"; import { checkRepresentable, unrepresentableMessage } from "../protocols/guard"; import { requestPathForLane } from "../protocols/path"; import { resolveApiSurfaceSettings, resolveProtocolSettings } from "../protocols/settings"; import { markProtocolBlocked, markProtocolEntry } from "../protocols/trace"; import { recordProtocolShadowPlan } from "../protocols/shadow-plan"; import { captureAnthropicClientIdentity } from "../adapters/anthropic/client-identity"; import { nativeMessagesDeclineReason, type NativeMessagesSelector } from "./messages-native-eligibility"; import { isApiAuthRequired, isDataPlaneAdmissionSecret, isProxyAdmissionSecret, resolveContextPrincipal, type RequestPolicyView, type DataPlaneAdmission, } from "./auth-cors"; import type { AdmissionLease } from "../lib/admission"; import { tryClaimNativeMainProfileForTurn } from "../codex/native-main-admission"; import { CODEX_MAIN_PROFILE_MAINTENANCE_MESSAGE } from "../codex/auth-context"; import { createTranslatorBudget, finalizeTranslatorBudgetResponse, isTranslatorBudgetExceededError, type TranslatorBudget, } from "../lib/translator-budget"; import { parseRequestEffortRowId, type ParsedEffortRowId, } from "./effort-row"; import { parseFastOnlyRowId, parseSyntheticRowId, type ParsedFastRowId, } from "./fast-row"; type Rec = Record; /** Which listener a Claude Messages request arrived on. Only the intercept ingress honours bindings. */ export interface ClaudeIngressOptions { claudeIntercept?: boolean; } /** * Decode a Claude selector that may carry the fast marker. * * The exact form is tried first, so a real model whose alias genuinely ends in the marker * keeps winning. Only then is the marker treated as synthetic and the bare base decoded: * a Desktop 3P alias is a HASH registered WITHOUT the marker, so an exact lookup can never * resolve a synthetic one. */ function decodeClaudeFastSelector(raw: string, cc?: OcxConfig["claudeCode"]): string { const model = stripOneMillionMarker(raw); const exact = resolveInboundModel(model, cc); if (!model.endsWith("--fast")) return exact; const fullMapping = cc?.modelMap?.[model]; if (resolveAlias(model) || isKnownDesktop3pModelId(model) || (typeof fullMapping === "string" && fullMapping.length > 0)) return exact; const bare = model.slice(0, -"--fast".length); // A classifier fallback is not an exact match for a registered Desktop base. // Preserve established non-Desktop fallback behavior while decoding that base first. if (exact !== model && !resolveDesktop3pAlias(bare)) return exact; const decodedBase = resolveInboundModel(bare, cc); return decodedBase === bare ? exact : `${decodedBase}--fast`; } /** Restore reversible native Claude picker aliases before Anthropic passthrough checks. */ function decodeNativeClaudePickerAlias(raw: string, cc?: OcxConfig["claudeCode"]): string { const decoded = resolveInboundModel(raw, cc); if (!decoded.startsWith("claude-")) return raw; // A picker value saved before the ocx-claude spelling keeps the native passthrough too. return claudeCodeNativeAlias(decoded) === raw || legacyAliasForNative(decoded) === raw ? decoded : raw; } function isRec(v: unknown): v is Rec { return !!v && typeof v === "object" && !Array.isArray(v); } function desktopMappingUnavailableResponse(error: DesktopModelMappingUnavailableError): Response { const response = anthropicErrorResponse(503, error.message, "api_error", "desktop_model_mapping_unavailable"); response.headers.set("Retry-After", "1"); return response; } /** Resolve Claude-only sidecar overrides without mutating the shared server config. */ export function buildClaudeReplayConfig(config: OcxConfig): OcxConfig { return { ...config, webSearchSidecar: { ...config.webSearchSidecar, ...config.claudeCode?.webSearchSidecar, }, visionSidecar: { ...config.visionSidecar, ...config.claudeCode?.visionSidecar, }, }; } /** * Messages exposure, shared by /v1/messages and /v1/messages/count_tokens so the two can never * disagree. `resolveApiSurfaceSettings` is the only reader: an explicit * `apiSurfaces.messages.enabled` wins, a malformed one closes the surface, and absence inherits * `claudeCode.enabled`. */ function claudeInboundDisabled(config: OcxConfig): Response | null { const messages = resolveApiSurfaceSettings(config).messages; if (messages.enabled) return null; const detail = messages.source === "invalid" ? "config.apiSurfaces.messages is not a valid setting, so the surface stays closed" : "GUI: API page Messages toggle / config.apiSurfaces.messages.enabled / config.claudeCode.enabled"; return anthropicErrorResponse(403, `Messages API is disabled (${detail})`, "permission_error"); } async function readAnthropicBody(req: Request, budget: TranslatorBudget, maxBytes: number): Promise { try { return await readJsonRequestBody(req, budget, maxBytes); } catch (err) { if (isTranslatorBudgetExceededError(err)) throw err; throw new AnthropicRequestError(err instanceof Error && err.message ? err.message : "Invalid JSON body"); } } // ── Native Anthropic passthrough (subscription OAuth pierce) ────────────────────── // When Claude Code runs with ONLY ANTHROPIC_BASE_URL set (subscription mode — the // connectors warning stays off), it sends its OWN claude.ai OAuth Bearer to us. // Requests for genuine claude/anthropic models that no alias/modelMap claims are // forwarded VERBATIM to api.anthropic.com with the caller's credential and all // end-to-end headers, so betas/thinking signatures/billing identity stay native. // (Evidence: teamclaude --no-mitm + Vercel gateway docs, devlog 003/060.) const PASSTHROUGH_STRIP_HEADERS = new Set([ "connection", "keep-alive", "transfer-encoding", "upgrade", "te", "trailer", "proxy-authenticate", "proxy-authorization", "host", "content-length", "accept-encoding", "x-opencodex-api-key", "origin", ]); function singleCredentialToken(name: "authorization" | "x-api-key", value: string | null): string | null { const raw = value?.trim() ?? ""; // Fetch Headers comma-joins duplicate fields. Neither Anthropic credential format permits a // comma, so treating a joined value as one token could hide an admission secret behind a real // provider credential. Ambiguous credential headers fail closed. if (!raw || raw.includes(",")) return null; if (name === "authorization") { const match = /^Bearer\s+(.+)$/i.exec(raw); return match?.[1]?.trim() || null; } return raw; } function hasAnthropicNativeCredential(req: Request, config: OcxConfig): boolean { const bearer = singleCredentialToken("authorization", req.headers.get("authorization")); const apiKey = singleCredentialToken("x-api-key", req.headers.get("x-api-key")); return (!!bearer && bearer.startsWith("sk-ant-") && !isProxyAdmissionSecret(bearer, config)) || (!!apiKey && apiKey.startsWith("sk-ant-") && !isProxyAdmissionSecret(apiKey, config)); } function wantsNativePassthrough( req: Request, config: OcxConfig, requestPolicy: RequestPolicyView, model: unknown, cc: OcxConfig["claudeCode"] = config.claudeCode, ): model is string { if (cc?.nativePassthrough === false) return false; if (typeof model !== "string" || !/^(claude|anthropic)/i.test(model)) return false; if (messagesSelectorTargetsSecondaryInstance(config, model, cc)) return false; // Authorization and x-api-key both belong to the upstream on this branch. An exposed listener // therefore requires the dedicated admission header even though the routed Messages surface // keeps accepting all three legacy admission forms. if (isApiAuthRequired(requestPolicy)) { const dedicated = req.headers.get("x-opencodex-api-key")?.trim() ?? ""; if (!isDataPlaneAdmissionSecret(dedicated, config)) return false; } if (!hasAnthropicNativeCredential(req, config)) return false; // An alias or modelMap hit means the user asked for a ROUTED model: translate instead. // `cc` carries first-party intercept bindings for requests on the claude-intercept ingress. return resolveInboundModel(model, cc) === model; } function shouldForwardNativeHeader(name: string, value: string, config: OcxConfig): boolean { const lowerName = name.toLowerCase(); if (PASSTHROUGH_STRIP_HEADERS.has(lowerName)) return false; if (lowerName !== "authorization" && lowerName !== "x-api-key") return true; const token = singleCredentialToken(lowerName, value); return !!token && !isProxyAdmissionSecret(token, config); } /** Format a 32-hex cache key as a uuid-shaped session id (version/variant nibbles forced). */ function uuidFromHex(hex32: string): string { const h = (hex32 + "0".repeat(32)).slice(0, 32); return `${h.slice(0, 8)}-${h.slice(8, 12)}-4${h.slice(13, 16)}-8${h.slice(17, 20)}-${h.slice(20, 32)}`; } /** * The passthrough accounting layer, shared with the managed native lane. It lives in * messages-shared.ts so that lane can take it without importing this module, and this * module without importing that lane. Re-exported here because this file was its only * home and callers still resolve it from this module. */ import { anthropicUsageToOcx, estimateClaudeRequestTokens, resolvePassthroughBodyGuard, sanitizePassthroughToolCallIds, tapAnthropicSseForLog, type PassthroughBodyGuard, type PassthroughCloseReason, } from "./messages-shared"; export { anthropicUsageToOcx, estimateClaudeRequestTokens, resolvePassthroughBodyGuard, sanitizePassthroughToolCallIds, tapAnthropicSseForLog, type PassthroughBodyGuard, type PassthroughCloseReason, type PassthroughFinalizeMeta, } from "./messages-shared"; /** Upstream error types a non-stream passthrough failure may report in history. */ const PASSTHROUGH_ERROR_TYPES = [ "invalid_request_error", "authentication_error", "permission_error", "not_found_error", "rate_limit_error", "api_error", "overloaded_error", "request_too_large", ] as const; const PASSTHROUGH_ERROR_DETAIL_MAX_CHARS = 64 * 1024; async function anthropicNativePassthrough( req: Request, config: OcxConfig, logCtx: RequestLogContext, logIds: { requestId: string; start: number } | undefined, body: Rec, pathname: string, ): Promise { const model = typeof body.model === "string" ? body.model : "unknown"; logCtx.model = model; logCtx.provider = "anthropic-native"; logCtx.requestedModel = model; const finalize = createFinalRequestLog(logIds, logCtx).finish; // Locally generated diagnostics are fixed text plus validated guard limits. Runtime // fetch errors retain their existing client response but use a fixed stored reason. const fail = (status: number, closeReason: PassthroughCloseReason | "non_stream", message: string, type: string, storedReason = message) => { const safeMessage = redactSecretString(message); logCtx.upstreamError = storedReason.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; // Native passthrough bypasses the anthropic adapter, so the generous image pipeline // (devlog/260714_image_normalization_pipeline/040) must run here: tier-normalize then // guard the already-Anthropic-wire messages before serialization. Applies to // count_tokens too — counts must match what the real send will contain, and the 32MB // body cap applies to it equally. Non-message bodies pass through untouched. if (Array.isArray(body.messages)) { await normalizeAnthropicImages(body.messages, { abortSignal: req.signal }); enforceAnthropicImageLimits(body.messages); sanitizePassthroughToolCallIds(body.messages); } const headers = new Headers(); req.headers.forEach((value, name) => { if (shouldForwardNativeHeader(name, value, config)) headers.set(name, value); }); headers.set("content-type", "application/json"); const result = await fetchWithHeaderDeadline( `${base}${pathname}${search}`, { method: "POST", headers, body: JSON.stringify(body) }, config.connectTimeoutMs ?? 200_000, req.signal, ); if (result.kind === "timeout") { return fail(504, "non_stream", "anthropic passthrough timed out waiting for response headers", "timeout_error"); } if (result.kind === "error") { const err = result.error; return fail(502, "non_stream", `anthropic passthrough failed: ${err instanceof Error ? err.message : String(err)}`, "api_error", "anthropic passthrough failed: upstream connection error"); } const upstream = result.upstream; const contentType = upstream.headers.get("content-type") ?? "application/json"; const rateLimitHeaders = { ...anthropicRateLimitHeaders(upstream.headers), ...(pathname === "/v1/messages" ? upstreamMessagesRequestIdHeaders(upstream.headers) : {}), }; const bodyGuard = resolvePassthroughBodyGuard(config, req.signal); if (upstream.ok && contentType.includes("text/event-stream") && upstream.body) { return new Response(tapAnthropicSseForLog(upstream.body, logCtx, finalize, bodyGuard), { status: upstream.status, headers: { ...rateLimitHeaders, "Content-Type": contentType, "Cache-Control": "no-cache", "Connection": "keep-alive", }, }); } // Non-stream (count_tokens, errors, stream:false): relay verbatim under the same // 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") { return fail(499, "client_cancel", "client closed request during anthropic passthrough", "api_error"); } if (bodyResult.kind === "stall") { 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") { return fail(502, "body_overflow", `anthropic passthrough body exceeded ${bodyGuard.maxBytes} bytes`, "api_error"); } const text = bodyResult.text; if (upstream.ok) { try { 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) { // Upstream text is untrusted: retain only a closed type vocabulary in history. // Bound parsing work independently of the larger response-relay body limit. const statusReason = `Provider error ${upstream.status}`; logCtx.upstreamError = statusReason; if (text.length <= PASSTHROUGH_ERROR_DETAIL_MAX_CHARS) { try { const parsed: unknown = JSON.parse(text); if (isRec(parsed) && parsed.type === "error" && isRec(parsed.error)) { const upstreamType = parsed.error.type; const knownType = PASSTHROUGH_ERROR_TYPES.find(type => type === upstreamType); if (knownType) logCtx.upstreamError = `${statusReason}: ${knownType}`; } } catch { /* Malformed errors keep the fixed status-only diagnostic. */ } } } finalize(upstream.status, { closeReason: "non_stream" }); const retryAfter = upstream.headers.get("retry-after"); return new Response(text, { status: upstream.status, headers: { ...rateLimitHeaders, "Content-Type": contentType, ...(retryAfter ? { "Retry-After": retryAfter } : {}) }, }); } type BoundedPassthroughBody = | { kind: "ok"; text: string } | { kind: "stall" } | { kind: "overflow" } | { kind: "client_cancel" }; /** * Bounded replacement for `await upstream.text()` on the non-stream passthrough * branch: same idle-only + size-cap semantics as the SSE tap. NOTE: reader.cancel() * resolves a pending read as done rather than rejecting, so the stalled flag is * re-checked after every read settlement (audit round 3). */ export async function readBoundedPassthroughBody( upstream: Response, guard: PassthroughBodyGuard, ): Promise { if (!upstream.body) return { kind: "ok", text: await upstream.text() }; const reader = upstream.body.getReader(); const decoder = new TextDecoder(); let text = ""; let bytes = 0; let stalled = false; let aborted = false; const idle = idleDeadline(guard.stallMs, () => { stalled = true; reader.cancel(new DOMException("anthropic passthrough body stalled", "TimeoutError")).catch(() => {}); }); // Deterministic client-abort classification (audit round 4): Bun may surface the // abort as a read rejection OR a resolved done, so we cancel the reader ourselves // and classify via the flag rather than the read's settlement shape. const signal = guard.reqSignal; const onAbort = () => { aborted = true; reader.cancel(signal?.reason).catch(() => {}); }; if (signal?.aborted) onAbort(); else signal?.addEventListener("abort", onAbort, { once: true }); try { while (true) { idle.reset(); let result: Awaited>; try { result = await reader.read(); } catch (err) { if (aborted) return { kind: "client_cancel" }; if (stalled) return { kind: "stall" }; throw err; } finally { idle.pause(); } if (aborted) return { kind: "client_cancel" }; if (stalled) return { kind: "stall" }; if (result.done) break; if (result.value.byteLength === 0) continue; bytes += result.value.byteLength; if (guard.maxBytes > 0 && bytes > guard.maxBytes) { reader.cancel(new DOMException("anthropic passthrough body exceeded byte cap", "QuotaExceededError")).catch(() => {}); return { kind: "overflow" }; } text += decoder.decode(result.value, { stream: true }); } text += decoder.decode(); return { kind: "ok", text }; } finally { idle.cancel(); signal?.removeEventListener("abort", onAbort); } } /** * Header-phase fetch guarded by a clearable deadline (PR #136 follow-up hardening). * * The deadline covers ONLY the wait for response headers; once `fetch` settles — * fulfilled OR rejected — the timer must die. The `finally` block guarantees * `clear()` on every path (success, upstream reject, deadline expiry), fixing the * timer leak where a rejected fetch left the deadline running until expiry. * `didExpire()` stays truthful after `clear()` (see src/lib/abort.ts), so timeout * classification inside the catch is unaffected by the finally cleanup. * * `makeDeadline`/`fetchImpl` are injectable for deterministic unit tests. */ export type HeaderDeadlineFetchResult = | { kind: "response"; upstream: Response } | { kind: "timeout" } | { kind: "error"; error: unknown }; export async function fetchWithHeaderDeadline( input: string | URL, init: RequestInit, timeoutMs: number, parent?: AbortSignal, makeDeadline: typeof clearableDeadline = clearableDeadline, fetchImpl: typeof fetch = fetch, ): Promise { const deadline = makeDeadline(timeoutMs, parent); try { const upstream = await fetchImpl(input, { ...init, redirect: "manual", signal: deadline.signal, timeout: 0 }); return { kind: "response", upstream }; } catch (error) { if (deadline.didExpire()) return { kind: "timeout" }; return { kind: "error", error }; } finally { deadline.clear(); } } export async function handleClaudeMessages( req: Request, config: OcxConfig, logCtx: RequestLogContext, logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease; admission?: DataPlaneAdmission }, requestPolicy: RequestPolicyView = config, ingress: ClaudeIngressOptions = {}, ): Promise { const translatorBudget = createTranslatorBudget(); try { return finalizeTranslatorBudgetResponse( await handleClaudeMessagesWithBudget(req, config, logCtx, translatorBudget, logIds, requestPolicy, ingress), translatorBudget, req.signal, ); } catch (error) { translatorBudget.dispose(); throw error; } } /** The client answer for a Messages body the translation (or the direct decode) refused. */ function translationFailure(err: unknown, logCtx: RequestLogContext, logIds: { requestId: string; start: number } | undefined): Response { const overflow = isTranslatorBudgetExceededError(err); const unavailable = err instanceof DesktopModelMappingUnavailableError; const status = overflow ? 413 : unavailable ? 503 : err instanceof AnthropicRequestError ? 400 : 500; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, { closeReason: "non_stream" }); if (unavailable) return desktopMappingUnavailableResponse(err); return anthropicErrorResponse( status, overflow ? "request translation buffer exceeded the safe limit" : err instanceof Error ? err.message : String(err), overflow ? "request_too_large" : undefined, overflow ? "translation_buffer_limit" : undefined, ); } /** * Translate a Claude Messages request, route it through the Responses pipeline, * and translate the reply back. Runs under a translator budget owned by the * caller; Go session affinity is derived here and handed to the final Go * transport out of band rather than through replay headers. */ async function handleClaudeMessagesWithBudget( req: Request, config: OcxConfig, logCtx: RequestLogContext, translatorBudget: TranslatorBudget, logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease; admission?: DataPlaneAdmission }, requestPolicy: RequestPolicyView = config, ingress: ClaudeIngressOptions = {}, ): Promise { logCtx.surface = "claude"; const disabled = claudeInboundDisabled(config); if (disabled) { markProtocolBlocked(logCtx, { inbound: "messages", reasonCodes: ["surface-disabled"] }); if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 403, { closeReason: "non_stream" }); return disabled; } // Model resolution reads this view; every other claudeCode setting keeps reading `config`. const cc = claudeCodeForIngress(config.claudeCode, ingress.claudeIntercept === true); let anthropicBody: unknown; let internalBody: Rec; let cacheKeySource: ClaudeCacheKeySource = null; let nativeReasoningReplay: ReadonlyMap | undefined; const nativeReasoningMint: { owner?: NativeReasoningOwner } = {}; let nativeProjectionOnlyError: AnthropicRequestError | undefined; let effortOverride: string | null = null; let effortRow: ParsedEffortRowId | null = null; let fastRow: ParsedFastRowId | null = null; let requestedModel = ""; // Antigravity direct lane (off by default): planned before translation, kept only if routing // settles on the route the plan was made for. let directPlan: MessagesDirectPlan | undefined; /** The retained charge for the settings-only body, handed back if the plan is left. */ let directSettingsBodyBytes = 0; // Built only under the reject policy; the legacy default leaves this request untouched. let envelope: ProtocolEnvelope | undefined; let messagesFeatures: () => Iterable = () => []; try { anthropicBody = await readAnthropicBody(req, translatorBudget, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)); // Defensive [1m] strip (devlog 138): clients normally remove the context-variant // marker themselves; the 1M signal we act on is the anthropic-beta header. // Case-insensitive — the CLI matches /\[1m\]/i (audit 021 #7). if (isRec(anthropicBody) && typeof anthropicBody.model === "string") { anthropicBody.model = stripOneMillionMarker(anthropicBody.model); } // ocx-route override (devlog 072): injected agent bodies pin their model via a // system-prompt directive because 2.1.207 ignores custom ids in agent // frontmatter. Must run BEFORE the native-passthrough branch — the CLI sends // these subagent turns under a fallback claude model id. if (isRec(anthropicBody)) { const routeOverride = extractOcxRouteDirective(anthropicBody); if (routeOverride && typeof anthropicBody.model === "string") { anthropicBody.model = stripOneMillionMarker(routeOverride); effortOverride = extractOcxEffortDirective(anthropicBody); } } if (isRec(anthropicBody) && typeof anthropicBody.model === "string") { anthropicBody.model = decodeNativeClaudePickerAlias(anthropicBody.model, cc); } if (isRec(anthropicBody) && typeof anthropicBody.model === "string") { requestedModel = anthropicBody.model; // Decode for Fast only. A Claude alias is `claude-ocx---`, so it // already uses `--` as its own separator: stripping the marker off the RAW alias would // turn `claude-ocx-p--foo--fast` into `claude-ocx-p--foo` and route a DIFFERENT model. // Effort parsing keeps the raw selector, so its behaviour is untouched. ({ fastRow, effortRow } = parseSyntheticRowId( requestedModel, config, () => decodeClaudeFastSelector(requestedModel, cc), )); if (effortRow) { anthropicBody.model = effortRow.baseId; effortOverride = effortRow.effort; } if (fastRow) anthropicBody.model = fastRow.baseId; } // Debug capture (opt-in allowlist scalars) BEFORE the passthrough branch so // native, routed, and disabled-alias paths are all observable (devlog 130 B1). captureClaudeInbound( "messages", anthropicBody, isRec(anthropicBody) && typeof anthropicBody.model === "string" ? resolveInboundModel(anthropicBody.model, cc) : undefined, req.headers.get("anthropic-beta") ?? undefined, ); // Client surface discrimination: Desktop 3P aliases resolve through the // desktop registry; Code uses readable aliases or direct model names. The CLI's first-party // picker also offers registry aliases, so a CLI-classified User-Agent stays the Code surface. if (isRec(anthropicBody) && typeof anthropicBody.model === "string" && resolveDesktop3pAlias(anthropicBody.model) && classifyInterceptClient(req.headers.get("user-agent")) !== "cli") { logCtx.surface = "claude-desktop"; recordDesktopRequest(); } // Correlate before native passthrough so Anthropic-credential turns still filter/total (#330 / #522). if (isRec(anthropicBody)) { const claudeConversationId = conversationIdFromClaudeMetadata( isRec(anthropicBody.metadata) ? anthropicBody.metadata : undefined, ); if (claudeConversationId) logCtx.conversationId = claudeConversationId; } // A fast row blocks passthrough, unlike the chat case: native passthrough forwards the // caller's body with the caller's credential and never runs the Anthropic adapter, so the // proxy-owned `speed` + beta (anthropic-speed wire) and its usage.speed observation would be // silently skipped. Translation reaches the adapter, which owns both. const messagesBody = anthropicBody; if (isRec(messagesBody) && resolveProtocolSettings(config).unrepresentable === "reject") { envelope = createProtocolEnvelope({ inbound: "messages", body: messagesBody, translatorBudget }); } const sourceEnvelope = envelope; // The bridge entry mark below reads these before an effort override rewrites `thinking`, // which also fixes the envelope's cached features on the caller's own settings. // Scanned once, like the envelope's cache, so a later mark reports the same features. let scannedFeatures: ReadonlySet | undefined; messagesFeatures = sourceEnvelope ? () => sourceEnvelope.features() : () => (scannedFeatures ??= featuresFromMessagesBody(messagesBody)); if (!effortRow && !fastRow && isRec(anthropicBody) && wantsNativePassthrough(req, config, requestPolicy, anthropicBody.model, cc)) { markProtocolEntry(logCtx, { inbound: "messages", lane: "native", features: messagesFeatures }); recordProtocolShadowPlan(logCtx, config, { inbound: "messages", model: requestedModel }); let projectedBody: Rec; try { projectedBody = nativeAnthropicProjection(anthropicBody, translatorBudget); } catch (err) { if (!isTranslatorBudgetExceededError(err)) throw err; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); return anthropicErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); } return await anthropicNativePassthrough(req, config, logCtx, logIds, projectedBody, "/v1/messages"); } if (isRec(anthropicBody) && typeof anthropicBody.model === "string" && messagesSecondaryInstanceUnavailable(config, anthropicBody.model, cc)) { if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 401, { closeReason: "non_stream" }); return anthropicErrorResponse(401, "Configured Anthropic OAuth instance is unavailable", "authentication_error"); } // Only Anthropic holds message-thread state; the error makes Claude Code resend the full turn. if (carriesMessageThread(anthropicBody)) { logCtx.errorCode = "claude_thread_unsupported"; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 400, { closeReason: "non_stream" }); return messageThreadUnsupportedResponse(); } // Capture source semantics before effort rewriting or translation drops fields. // This policy is uniform across translated targets, including later fallback attempts. const compatibilityMode: unknown = config.claudeCode?.compatibility; if (compatibilityMode !== undefined) { if (!isClaudeCompatibilityMode(compatibilityMode)) { logCtx.errorCode = "claude_compatibility_configuration"; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 503, { closeReason: "non_stream" }); return anthropicErrorResponse(503, "Invalid claudeCode.compatibility setting", "api_error"); } const compatibility = analyzeClaudeCompatibility(anthropicBody, { mode: compatibilityMode, anthropicBeta: req.headers.get("anthropic-beta") ?? undefined, }); if (compatibility.decision === "reject") { markProtocolBlocked(logCtx, { inbound: "messages", reasonCodes: ["compatibility-reject"], features: messagesFeatures }); logCtx.errorCode = "claude_compatibility_unsupported"; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 400, { closeReason: "non_stream" }); return anthropicErrorResponse(400, compatibility.reason!, "invalid_request_error"); } if (compatibility.decision === "shadow") { logCtx.claudeCompatibility = { decision: "shadow", featureCodes: compatibility.featureCodes, reason: compatibility.reason, }; } } // Features are read here, before an effort override rewrites the thinking settings. markProtocolEntry(logCtx, { inbound: "messages", lane: "bridge", reasonCodes: effortRow ? ["effort-row"] : fastRow ? ["fast-row"] : [], features: messagesFeatures, }); recordProtocolShadowPlan(logCtx, config, { inbound: "messages", model: requestedModel }); if (isRec(anthropicBody) && effortOverride) { anthropicBody.output_config = { ...(isRec(anthropicBody.output_config) ? anthropicBody.output_config : {}), effort: effortOverride, }; delete anthropicBody.thinking; } // On the direct lane the core receives only the request settings; the conversation is // already decoded into the IR. Without a plan the translation below runs as before. directPlan = planMessagesAntigravityDirect(anthropicBody, cc, config, translatorBudget); let translation = directPlan && { body: directPlan.settingsBody, cacheKeySource: directPlan.cacheKeySource, nativeReasoningReplay: directPlan.conversation.nativeReasoningReplay, }; try { translation ??= messagesToResponsesTranslation(anthropicBody, cc, translatorBudget); } catch (err) { // A native Messages route can discard an undecodable proxy envelope. Keep the // original error for any route that ultimately translates to Responses. if (!(err instanceof AnthropicRequestError) || err.message !== "malformed ocxr1 reasoning signature" || !isRec(anthropicBody)) throw err; const projected = nativeAnthropicProjection(anthropicBody, translatorBudget); if (projected === anthropicBody) throw err; translation = messagesToResponsesTranslation(projected, cc, translatorBudget); nativeProjectionOnlyError = err; } internalBody = translation.body; // The Anthropic translator builds its body from model/input/store/stream plus sampling // fields only, so the caller intent is applied to the TRANSLATED body rather than the // inbound one. if (fastRow) internalBody.service_tier = "priority"; const internalBodyBytes = jsonUtf8Bytes(internalBody); translatorBudget.chargeRetained(internalBodyBytes, { kind: "request_copies" }); if (directPlan) directSettingsBodyBytes = internalBodyBytes; cacheKeySource = translation.cacheKeySource; nativeReasoningReplay = translation.nativeReasoningReplay; } catch (err) { return translationFailure(err, logCtx, logIds); } if (!requestedModel) requestedModel = (anthropicBody as Rec).model as string; const stream = internalBody.stream === true; /** * This proxy's count of the prompt it is about to forward, computed at most once per wire. * * Two readers want it and they want it under different rules. The usage log takes it as a * floor only for estimated-usage adapters, because its merge is `max(reported, estimate)` and * would otherwise overwrite real usage. `message_start` takes it whenever the upstream sent * no confirmed usage before the first frame, where nothing is merged and the terminal * `message_delta` still corrects it (#4857). */ let requestTokenFloor: number | undefined; /** * The `text|signature|redacted` triple `requestTokenFloor` was measured under. * * The count is not a constant for the request: a combo re-picks its child at dispatch and a * retry can rotate the wire, so the pre-dispatch callers and the post-dispatch translator can * legitimately want different projections. Keying the memo re-measures when they disagree * instead of handing the translator the pre-dispatch measurement, and the estimator still runs * at most twice for one request. At most, because a request has one settled wire per phase and * the key collapses every repeat of the same answer. */ let requestTokenFloorKey: string | undefined; // Routed adapters only support streamed turns; always stream internally and fold // the translated Anthropic SSE into a message JSON for non-streaming clients. internalBody.stream = true; let clientEncoder: ClientEncoderOption | undefined; // Native ChatGPT passthrough (openai-responses forward) accepts only Codex-shaped // bodies: it 400s on sampling params (verified live 2026-07-11). Every openai-responses // route drops temperature/top_p/stop/user below; max_output_tokens is left to // stripUnsupportedForwardParams (canonical-forward.ts), which drops it only for that backend. let settledRoute: RouteResult | undefined; /** * Which replayed thinking fields the send that actually happened will serialize. * * Read lazily: routing settles the ingress wire, core may then re-pick it (a combo child, a * rotated retry), and this is called both before and after that happens. It therefore reads the * physical attempt when one exists and the ingress route only until then. */ const claudeThinkingProjection = (): ClaudeThinkingProjection => thinkingProjectionForDispatch(config, settledRoute, logCtx); const claudeRequestTokenFloor = (): number => { const thinking = claudeThinkingProjection(); const key = `${thinking.text}|${thinking.signature}|${thinking.redacted}`; if (requestTokenFloor === undefined || requestTokenFloorKey !== key) { requestTokenFloor = estimateClaudeRequestTokens(anthropicBody as Rec, requestedModel, thinking); requestTokenFloorKey = key; } return requestTokenFloor; }; /** Leave the direct plan: the core gets the full bridge translation, as without the switch. */ const leaveDirectPlan = (): unknown => { // Hand back everything the plan retained: its decode and the settings-only body it replaces. directPlan?.releaseDecodeCharges(); directPlan = undefined; nativeReasoningReplay = undefined; translatorBudget.releaseRetained(directSettingsBodyBytes, { kind: "request_copies" }); directSettingsBodyBytes = 0; try { const translation = messagesToResponsesTranslation(anthropicBody, cc, translatorBudget); internalBody = translation.body; if (fastRow) internalBody.service_tier = "priority"; internalBody.stream = true; translatorBudget.chargeRetained(jsonUtf8Bytes(internalBody), { kind: "request_copies" }); cacheKeySource = translation.cacheKeySource; nativeReasoningReplay = translation.nativeReasoningReplay; return undefined; } catch (err) { return err; } }; let directLeaveError: unknown; try { const route = routeModel(config, internalBody.model as string, directPlan ? directPlan.evidence : evidenceFromBody(internalBody)); // Settle the wire once so the sampling decision below reads the effective // adapter rather than the provider-wide default (#404). route.staticPolicy = captureRouteStaticPolicy( route.providerName, route.modelId, route.provider, route.staticPolicy.effectiveAlias, "anthropic", ); // Keep native dispatch scoped while translated xAI OAuth requests preview // the same billed Fast lane as their final Responses scope check. assertRouteAllowedByScope( resolveAdmissionModelScope(config, logIds?.admission), String(internalBody.model ?? ""), { providerName: route.providerName, modelId: previewXaiOauthWireModel({ options: { serviceTier: typeof internalBody.service_tier === "string" ? internalBody.service_tier : undefined, } }, route, config, "anthropic") }, ); route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, "anthropic", route.staticPolicy); logCtx.routeDecision = route.routeDecision; settledRoute = route; // Before any route-specific shaping, so a request that leaves the lane is shaped as the bridge's. if (directPlan && !messagesDirectPlanHolds(directPlan, route, anthropicBody as Rec, config)) directLeaveError = leaveDirectPlan(); if (directEncodersApply(config, route)) { clientEncoder = { protocol: "messages", stream, model: requestedModel, inputTokenFloor: claudeRequestTokenFloor() }; } if (route.provider.adapter === "openai-responses") { // max_output_tokens is the caller's spend cap: the adapter strips it only for the canonical // ChatGPT backend that rejects it (#5753), so other Responses gateways keep the cap. delete internalBody.temperature; delete internalBody.top_p; delete internalBody.stop; delete internalBody.user; } // Estimated-usage adapters (cursor/kiro) report no per-turn input tokens; stash a // request-side estimate so the log's in:0 rows get a floor. NEVER set this for // accurate-usage adapters — the request-log merge is max(reported, estimate) and // would overwrite real usage (audit 133 R1#7). if (route.provider.adapter === "cursor" || route.provider.adapter === "kiro") { logCtx.usageLogInputTokens = claudeRequestTokenFloor(); } // Effort safety valve (devlog 136 B6, audit 139 R2#2): opus-shaped aliases make // every routed model look like a reasoning model to Claude clients, so a forced // effort (CLAUDE_CODE_ALWAYS_ENABLE_EFFORT) would leak reasoning params to routes // that affirmatively expose NO effort control. Strip only on a definitive [] from // supportedLadderFor; unknown (undefined) passes through untouched. if (internalBody.reasoning !== undefined) { const { supportedLadderFor } = await import("./effort-policy"); const ladder = supportedLadderFor({ provider: route.provider, modelId: route.modelId }); if (ladder !== undefined && ladder.length === 0) delete internalBody.reasoning; } } catch (err) { if (err instanceof AdmissionModelDeniedError) { logCtx.requestedModel = requestedModel; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 403, { closeReason: "non_stream" }); return admissionModelDeniedResponse(err); } if (err instanceof UnknownRoutingPolicyError) { logCtx.requestedModel = requestedModel; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 404, { closeReason: "non_stream" }); return anthropicErrorResponse(404, err.message, "invalid_request_error"); } if (err instanceof NoEligiblePolicyCandidateError) { logCtx.routeDecision = err.trace; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 404, { closeReason: "non_stream" }); return anthropicErrorResponse(404, err.message, "invalid_request_error"); } /* unknown model: let handleResponses shape the 404 */ } if (directPlan && !settledRoute) directLeaveError ??= leaveDirectPlan(); if (directLeaveError !== undefined) return translationFailure(directLeaveError, logCtx, logIds); const directRoute = directPlan && settledRoute ? settledRoute : undefined; // The direct lane answers through the direct encoder whatever `directEncoders` says: neither // direction passes through the internal Responses body. if (directRoute) { clientEncoder = { protocol: "messages", stream, model: requestedModel, inputTokenFloor: claudeRequestTokenFloor(), requestLane: "direct" }; } // PF-08: a managed-key Anthropic route sends its Messages body natively. The caller-forward // passthrough above was decided on the caller's own credential and never reaches this point. const settledInstance = configuredAnthropicInstance(config, settledRoute?.providerName); if (settledRoute?.providerName === "anthropic2" && settledRoute.provider.authMode === "oauth" && !settledInstance) { if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 401, { closeReason: "non_stream" }); return anthropicErrorResponse(401, "Configured Anthropic OAuth instance is unavailable", "authentication_error"); } const nativeSelector: NativeMessagesSelector = { effortRow: !!effortRow, fastRow: !!fastRow, routeSelector: String(internalBody.model ?? ""), claudeCode: cc, // PF-10: a stored second OAuth account turns on the bridge's rotation, so it stays there. ...(settledRoute?.provider.authMode === "oauth" && settledInstance ? { oauthFailoverQuorum: anthropicRoutingFor(settledInstance).hasAnthropicFailoverQuorum() } : {}), }; const nativeDecline = settledRoute && isRec(anthropicBody) ? nativeMessagesDeclineReason(settledRoute, anthropicBody, config, nativeSelector) : "rollout-disabled"; const nativeMessagesRoute = settledRoute && nativeDecline === undefined ? settledRoute : undefined; if (nativeProjectionOnlyError && !nativeMessagesRoute) { if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 400, { closeReason: "non_stream" }); return anthropicErrorResponse(400, nativeProjectionOnlyError.message, "invalid_request_error"); } // With the switch on, a declined route says why on its bridge mark, as native Chat does. const antigravityDecline = settledRoute && !nativeMessagesRoute && !directRoute ? messagesDirectTraceDecline(settledRoute, anthropicBody, String(internalBody.model ?? ""), config) : undefined; if (directRoute) { markProtocolEntry(logCtx, { inbound: "messages", lane: "direct", features: messagesFeatures }); } else if ((nativeDecline && nativeDecline !== "rollout-disabled") || antigravityDecline) { markProtocolEntry(logCtx, { inbound: "messages", lane: "bridge", reasonCodes: [ ...(effortRow ? ["effort-row" as const] : fastRow ? ["fast-row" as const] : []), ...(nativeDecline && nativeDecline !== "rollout-disabled" ? [nativeDecline] : []), ...(antigravityDecline ? [antigravityDecline] : []), ], features: messagesFeatures, }); } // Combo and policy children are judged per candidate (PF-07); an unknown model has no route. if (envelope && settledRoute && !settledRoute.combo && settledRoute.routeKind !== "policy") { const verdict = checkRepresentable({ inbound: "messages", requestPath: requestPathForLane( "messages", nativeMessagesRoute ? "native" : directRoute ? "direct" : "bridge", upstreamWireForAdapter(settledRoute.provider.adapter), ), features: envelope.features(), policy: "reject", }); if (!verdict.ok) { markProtocolBlocked(logCtx, { inbound: "messages", reasonCodes: verdict.reasonCodes, features: verdict.features }); logCtx.errorCode = "unsupported_feature"; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 400, { closeReason: "non_stream" }); return anthropicErrorResponse(400, unrepresentableMessage(verdict.features), "invalid_request_error"); } } // PF-10: opaque thinking state reaches only first-party Anthropic; under `reject` a native // route to anyone else refuses the request instead of sending it without that state. if (envelope && nativeMessagesRoute && !credentialDomainFor(nativeMessagesRoute.provider)?.firstPartyAnthropic && messagesBodyHasOpaqueState(anthropicBody as Rec)) { markProtocolBlocked(logCtx, { inbound: "messages", reasonCodes: ["feature-unrepresentable", "opaque-state-stripped"], features: messagesFeatures }); logCtx.errorCode = "unsupported_feature"; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 400, { closeReason: "non_stream" }); return anthropicErrorResponse(400, "The selected route cannot carry thinking signatures or redacted_thinking blocks", "invalid_request_error"); } if (nativeMessagesRoute) { markProtocolEntry(logCtx, { inbound: "messages", lane: "native", features: messagesFeatures }); let nativeBody: Rec; try { // Built from the source envelope when there is one, after the managed-client steps above. nativeBody = nativeAnthropicProjection(envelope ? envelope.freshBody() : anthropicBody as Rec, translatorBudget); } catch (err) { if (!isTranslatorBudgetExceededError(err)) throw err; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); return anthropicErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); } const { handleNativeMessages } = await import("./messages-native"); return await handleNativeMessages({ req, config, logCtx, ...(logIds ? { logIds } : {}), route: nativeMessagesRoute, body: nativeBody, requestedModel, translatorBudget, selector: nativeSelector, modelScope: resolveAdmissionModelScope(config, logIds?.admission), // Compatibility identity is an opaque request-local handle, separate from credentials. clientIdentity: captureAnthropicClientIdentity(req.headers), callerAnthropicBeta: req.headers.get("anthropic-beta"), sessionKey: anthropicSessionKeyFromParts({ sessionIdHeader: req.headers.get("session_id")?.trim() || req.headers.get("x-claude-code-session-id"), threadIdHeader: req.headers.get("thread_id"), clientThreadId: conversationIdFromClaudeMetadata(isRec(nativeBody.metadata) ? nativeBody.metadata : undefined), promptCacheKey: typeof internalBody.prompt_cache_key === "string" ? internalBody.prompt_cache_key : null, promptCacheKeyIsSharedCohort: cacheKeySource === "system", }), }); } const headers = new Headers({ "content-type": "application/json" }); let trustedClaudeMainAuth: { authorization: string; chatgptAccountId?: string } | undefined; for (const name of FORWARD_HEADERS) { // The caller's bearer is the proxy admission token (ocx claude placeholder), never a // ChatGPT credential — forwarding it upstream turns into {"detail":"Unauthorized"}. if (name === "authorization") continue; const value = req.headers.get(name); if (value) headers.set(name, value); } // Routed replays need main ChatGPT auth so OpenAI-backed sidecars remain reachable; // native replays have no caller ChatGPT credential. This enrichment is optional: // auth-context later rejects a real physical-main selection, while routed/pool // traffic continues without reading native credentials during a fence/recovery. if (tryClaimNativeMainProfileForTurn(logIds?.turnAdmissionLease)) { const { getMainAccountToken } = await import("../codex/main-account"); const token = getMainAccountToken(); if (token) { const authorization = `Bearer ${token.accessToken}`; headers.set("authorization", authorization); headers.set("chatgpt-account-id", token.chatgptAccountId); trustedClaudeMainAuth = { authorization, ...(token.chatgptAccountId ? { chatgptAccountId: token.chatgptAccountId } : {}), }; } } // Carry Go identity out of band: a combo's preflight target may differ from its // actual dispatch/fallback target. Never add Go-only identity to replay headers. const claudeNativeSessionId = cacheKeySource === "metadata" && typeof internalBody.prompt_cache_key === "string" && isRec(anthropicBody) && conversationIdFromClaudeMetadata(isRec(anthropicBody.metadata) ? anthropicBody.metadata : undefined) !== undefined ? uuidFromHex(internalBody.prompt_cache_key) : undefined; const metadataGoLane = normalizeLogConversationId(claudeNativeSessionId); // Without any valid conversation identity, fall back to the request-scoped lane // allocated on the admitted client request (#4172): stable across retries and // route reconstruction, distinct per request, and never derived from a shared // system-prompt cache key or from a later synthesized native session_id header. const claudeGoSessionLane = sessionLaneIdFromRequest(headers) ?? normalizeLogConversationId(req.headers.get("x-opencode-session")) ?? metadataGoLane ?? getOrAllocateRequestSessionLane(req); let internalReq: Request; try { // The UTF-16 JSON string and the Request's UTF-8 body coexist until dispatch. const bodyBytes = jsonUtf8Bytes(internalBody); const reservation = translatorBudget.reserveTransient(3 * bodyBytes, { kind: "request_copies" }); try { internalReq = new Request("http://localhost/v1/responses", { method: "POST", headers, body: JSON.stringify(internalBody), }); linkRequestSessionLane(req, internalReq); } finally { reservation.release(); } translatorBudget.chargeRetained(bodyBytes, { kind: "request_copies" }); } catch (err) { if (!isTranslatorBudgetExceededError(err)) throw err; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); return anthropicErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); } // Request-log wiring mirrors the /v1/responses route: native passthrough finalizes // via the terminal callbacks; routed streams get the Responses-vocabulary log tap // BEFORE translation (the translated Anthropic stream has no response.completed // frame, so tapping it records a bogus 502 with no usage/cache detail). const finalizeNativeLog = createFinalRequestLog(logIds, logCtx).finish; const upstream = await handleResponses(internalReq, { ...buildClaudeReplayConfig(config), // Operator config wins; otherwise the caller's explicit TTLs (all 1h, or a pinned 5m) carry // over, and TTL-less breakpoints take the destination default. cacheRetention: config.cacheRetention ?? callerCacheRetention(anthropicBody), }, logCtx, { // Routing keeps Claude-only sidecar overrides; admission policy must follow the live owner. codexAuthPolicy: config, ...(logIds?.admission ? { admission: logIds.admission } : {}), // The translated request never forwards the caller credential; resolve the principal first. replayPrincipal: { clientPrincipalId: resolveContextPrincipal(req, config, logIds?.admission) }, ...(logIds?.turnAdmissionLease ? { turnAdmissionLease: logIds.turnAdmissionLease } : {}), abortSignal: req.signal, promptCacheKeyIsSharedCohort: cacheKeySource === "system", // The body is Responses-shaped by now, but the client spoke Anthropic Messages. // Without this the replay would look native and a Responses-scoped wire default // would fire, disagreeing with the pre-flight decision above. inboundWire: "anthropic", // An explicit empty `tools` array is a deny-all declaration the translation drops with // the rest of the catalog; carried separately so the undeclared-tool guard can still // enforce it while partial and absent catalogs stay deferred. clientDeclaredEmptyToolCatalog: isRec(anthropicBody) && Array.isArray(anthropicBody.tools) && anthropicBody.tools.length === 0, nativeReasoningReplay, nativeReasoningMint, // Out of body like the replay above: the core hands these to an Anthropic destination only. anthropicServerTools: declaredAnthropicServerTools(anthropicBody), anthropicServerReplay: replayedAnthropicServerBlocks(anthropicBody), // Preserve the original catalog through late fallback; only the physical adapter projects it. anthropicToolLoadingSource: cacheKeySource === "metadata" && isRec(anthropicBody) && Array.isArray(cc?.appendOnlyToolLoadingModels) && cc.appendOnlyToolLoadingModels.length > 0 ? { tools: anthropicBody.tools, messages: anthropicBody.messages, tool_choice: anthropicBody.tool_choice } : undefined, claudeGoAffinity: { sessionLane: claudeGoSessionLane }, claudeNativeSessionId, stripClaudeMainAuthForNoncanonicalForward: true, ...(trustedClaudeMainAuth ? { trustedClaudeMainAuth } : {}), // Claude's internal stored-main enrichment is not an original caller credential. nativeCallerAuth: null, callerDirectAuth: null, translatorBudget, ...(logIds ? { onFirstOutput: () => recordFirstOutput(logCtx, logIds.start) } : {}), onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForRequestLogTerminal(status, logCtx), { terminalStatus: status, closeReason: "terminal" }), onNativePassthroughCancel: () => finalizeNativeLog(499, { closeReason: "client_cancel" }), ...(clientEncoder ? { clientEncoder } : {}), ...(directRoute && directPlan ? { directRequest: messagesDirectRequestOption(directPlan, anthropicBody, cc, translatorBudget) } : {}), }); const response = logIds ? responseWithDeferredRequestLog(upstream, logIds.requestId, logIds.start, logCtx) : upstream; // Already in the Messages wire (direct encoder): no conversion. if (clientWireOf(upstream) === "messages") return response; if (!response.ok) { // Read the shared provenance verdict before consuming and re-wrapping the body. A refusal // and an ordinary provider rate limit are both 429, so the status cannot distinguish them. const replayRefusal = isReplayRefusalResponse(response); // Re-shape the OpenAI-style error envelope into the Anthropic one, preserving status. let message = `upstream error (${response.status})`; let contextError = false; let catalogRefusal = false; try { const text = await response.text(); try { const parsed = JSON.parse(text) as { error?: { message?: string; type?: string; code?: unknown } | string; message?: string }; contextError = typeof parsed?.error === "object" && parsed.error?.code === "context_length_exceeded"; catalogRefusal = typeof parsed?.error === "object" && parsed.error?.code === UNDECLARED_TOOL_CALL_ERROR_CODE; const nested = typeof parsed?.error === "object" && parsed.error ? parsed.error.message : undefined; const flat = typeof parsed?.error === "string" ? parsed.error : parsed?.message; message = nested || flat || (text ? `upstream error (${response.status}): ${text.slice(0, 400)}` : message); } catch { if (text) message = `upstream error (${response.status}): ${text.slice(0, 400)}`; } } catch { /* keep fallback message */ } const upstreamRetryAfter = response.headers.get("retry-after"); const retryAfter = replayRefusal || contextError ? undefined : resolveClientRetryAfter({ status: response.status, message, upstreamRetryAfter, }) // Instant-retry "0" is a valid client directive but rejected by cooldown parsers. // Preserve it so it still wins over the transient "2" fallback (claude-529 mapping). ?? (upstreamRetryAfter?.trim() === "0" ? "0" : undefined); // Transient upstream 5xx (already retried pre-stream, 010): reclassify as Anthropic // 529 overloaded_error so the Claude Code client applies its built-in backoff retry // instead of dying on a fatal api_error (260716 sol-builder incident). The request // log keeps the upstream status (captured in the deferred-log closure before this // rewrite): log = upstream truth, client = retry signal. // Retryable 429s also get Retry-After (#507) so Codex-shaped clients and Claude Code // share a backoff hint when the upstream omitted the header. const nativeMainFence = response.status === 503 && upstreamRetryAfter?.trim() === "1" && message === CODEX_MAIN_PROFILE_MAINTENANCE_MESSAGE; // The undeclared-tool guard's refusal is deterministic: the same request produces the // same verdict, so it must not arrive as a retryable overload like a transient 5xx. const transient = !replayRefusal && !nativeMainFence && !contextError && !catalogRefusal && isTransientUpstreamStatus(response.status); const outStatus = replayRefusal ? REPLAY_REFUSED_STATUS : nativeMainFence ? 503 : contextError ? 400 : transient ? 529 : response.status; const outHeaders = new Headers({ "Content-Type": "application/json" }); if (retryAfter) outHeaders.set("Retry-After", retryAfter); else if (transient) outHeaders.set("Retry-After", "2"); if (replayRefusal) applyReplayRefusalClientHeaders(outHeaders); const out = new Response(JSON.stringify(anthropicErrorBody( outStatus, message, undefined, replayRefusal ? UPSTREAM_RESET_REPLAY_REFUSED_CODE : contextError ? "context_length_exceeded" : catalogRefusal ? UNDECLARED_TOOL_CALL_ERROR_CODE : undefined, )), { status: outStatus, headers: outHeaders, }); return carryReplayRefusal(response, out); } const contentType = response.headers.get("content-type") ?? ""; if (contentType.includes("text/event-stream") && response.body) { const anthropicSse = responsesSseToAnthropicSse(response.body, requestedModel, { translatorBudget, nativeReasoningTagFor: blob => nativeReasoningTag(nativeReasoningMint.owner, blob), // Only a floor, and only for the first frame: an upstream that reports usage early wins // over it inside the translator, and the terminal `message_delta` carries the // authoritative count either way (#4857). inputTokenFloor: claudeRequestTokenFloor(), }); if (stream) { return new Response(anthropicSse, { status: 200, headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache", "Connection": "keep-alive", }, }); } let message: Rec; try { message = await collectAnthropicMessage(anthropicSse, requestedModel, translatorBudget); } catch (error) { if (isTranslatorBudgetExceededError(error)) { return anthropicErrorResponse(413, error.message, "request_too_large", error.code); } return anthropicErrorResponse(502, error instanceof Error ? error.message : String(error), "api_error"); } const isError = (message as Rec).type === "error"; const translatedError = isError && typeof (message as Rec).error === "object" ? (message as { error: { code?: unknown; message?: unknown } }).error : undefined; if (translatedError?.code === "translation_buffer_limit") { return anthropicErrorResponse( 413, typeof translatedError.message === "string" ? translatedError.message : "upstream translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit", ); } // A declared client refusal keeps its 4xx: as a 502 Claude Code retries a verdict that cannot // change. Server classes and rate limits keep the 502 this path has always answered. A 400 // worded as a client close answers the 499 httpStatusFromTerminalError logs for it. const refusalStatus = recognizedUpstreamError(translatedError ?? {})?.httpStatus; return new Response(JSON.stringify(message), { status: isError ? translatedError?.code === "context_length_exceeded" ? 400 : refusalStatus !== undefined && refusalStatus < 429 ? refusalStatus === 400 && isClientClosedMessage(String(translatedError?.message ?? "")) ? 499 : refusalStatus : 502 : 200, headers: { "Content-Type": "application/json" }, }); } // Defensive: some passthrough paths may answer JSON despite stream:true. let json: unknown; try { json = await response.json(); } catch { return anthropicErrorResponse(502, "internal replay returned a non-JSON response", "api_error"); } const status = (json as Rec)?.status; if (status === "failed") { const error = (json as { error?: { message?: string; code?: string } }).error; if (error?.code === "translation_buffer_limit") { return anthropicErrorResponse( 413, error.message ?? "upstream translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit", ); } if (error?.code === "context_length_exceeded") { return anthropicErrorResponse(400, error.message ?? "upstream context limit exceeded", "invalid_request_error", error.code); } return anthropicErrorResponse(502, error?.message ?? "upstream request failed", "api_error"); } let message: Rec; try { message = responsesJsonToAnthropicMessage(json, requestedModel, translatorBudget, blob => nativeReasoningTag(nativeReasoningMint.owner, blob)); } catch (err) { if (!isTranslatorBudgetExceededError(err)) throw err; return anthropicErrorResponse(413, "upstream translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); } if ((message as Rec).type === "error") { return new Response(JSON.stringify(message), { status: 529, headers: { "Content-Type": "application/json", "Retry-After": "2" }, }); } if (!stream) { return new Response(JSON.stringify(message), { status: 200, headers: { "Content-Type": "application/json" } }); } // Streaming client + JSON upstream: synthesize a minimal valid Anthropic stream. const encoder = new TextEncoder(); const frames: string[] = []; const emit = (name: string, data: Rec) => frames.push(`event: ${name}\ndata: ${JSON.stringify(data)}\n\n`); emit("message_start", { type: "message_start", message: { ...message, content: [], stop_reason: null, usage: { input_tokens: 0, output_tokens: 0 } } }); const blocks = Array.isArray((message as Rec).content) ? (message as Rec).content as Rec[] : []; blocks.forEach((block, index) => { emit("content_block_start", { type: "content_block_start", index, content_block: block }); emit("content_block_stop", { type: "content_block_stop", index }); }); emit("message_delta", { type: "message_delta", delta: { stop_reason: (message as Rec).stop_reason ?? "end_turn", stop_sequence: null }, usage: (message as Rec).usage ?? {} }); emit("message_stop", { type: "message_stop" }); return new Response(encoder.encode(frames.join("")), { status: 200, headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache" }, }); } /** * The projection for a route that has already settled. * * Only the OpenAI-shaped Chat adapter discards replayed thinking; every other settled wire * forwards the body it was given. An unknown route keeps the full body. */ function thinkingProjectionForRoute(route: RouteResult | undefined): ClaudeThinkingProjection { if (!route || route.provider.adapter !== "openai-chat") return CLAUDE_NATIVE_THINKING; return openAIChatSerializesThinking(route.provider, route.modelId); } /** * The projection for the wire that will physically carry the request. * * The ingress route is not the last word on that wire. A combo re-picks its child at dispatch, * and a retry can rotate the adapter mid-turn, so the ingress pick can name a different provider * — and therefore a different body — than the one that is sent. `logCtx.activeAttempt` is the * send that actually happened (`sealRequestAttemptIdentity` keeps its adapter current), so it * wins once it exists; before the first send the ingress route is the only authority there is. * * A Chat identity is re-derived through `routedProviderConfig`, not read off the raw config row: * `preserveReasoningContentModels` is registry-merged, so a row that omits it would otherwise * price a preserve-listed model as if the wire dropped its reasoning. */ function thinkingProjectionForDispatch( config: OcxConfig, route: RouteResult | undefined, logCtx: Pick, ): ClaudeThinkingProjection { const attempt = logCtx.activeAttempt; const adapter = attempt?.adapter ?? logCtx.providerAdapter; // No send to describe yet: the ingress route is the best available answer, and for the // Anthropic-native wire (which drops nothing) it is already the right one. if (adapter === undefined) return thinkingProjectionForRoute(route); if (adapter !== "openai-chat") return CLAUDE_NATIVE_THINKING; const providerName = attempt?.provider; const modelId = attempt?.model ?? route?.modelId; const provider = providerName !== undefined && Object.hasOwn(config.providers, providerName) ? config.providers[providerName] : undefined; // A Chat wire whose destination cannot be named keeps the route's own answer: over-counting on // a path that publishes nothing is harmless, under-counting a real prompt is not. if (providerName === undefined || modelId === undefined || provider === undefined) { return thinkingProjectionForRoute(route); } try { return openAIChatSerializesThinking(routedProviderConfig(providerName, provider), modelId); } catch { return thinkingProjectionForRoute(route); } } /** * The projection for a route resolved only to MEASURE a body this handler never sends. * * `previewRouteModel` is the read-only resolver: it advances no combo round-robin state, so a * count request cannot steer where the next real turn goes. An unresolvable model keeps the * full body, matching the old behavior for models routing cannot place. * * The wire is settled exactly as the turn path settles it. Routing fills in the provider's * registry adapter, and a per-model `modelAdapters` override or a pinned wire is applied later by * `resolveWireProtocolOverride` — so skipping it here would price the count against a body the * routed adapter never sends. */ function thinkingProjectionForPreview(config: OcxConfig, modelId: string): ClaudeThinkingProjection { try { const route = previewRouteModel(config, modelId); route.staticPolicy = captureRouteStaticPolicy( route.providerName, route.modelId, route.provider, route.staticPolicy.effectiveAlias, "anthropic", ); route.provider = resolveWireProtocolOverride( route.providerName, route.modelId, route.provider, "anthropic", route.staticPolicy, ); return thinkingProjectionForRoute(route); } catch { return CLAUDE_NATIVE_THINKING; } } export async function handleClaudeCountTokens( req: Request, config: OcxConfig, requestPolicy: RequestPolicyView = config, ingress: ClaudeIngressOptions = {}, ): Promise { const disabled = claudeInboundDisabled(config); if (disabled) return disabled; const cc = claudeCodeForIngress(config.claudeCode, ingress.claudeIntercept === true); let body: unknown; const translatorBudget = createTranslatorBudget(); try { body = await readAnthropicBody(req, translatorBudget, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)); } catch (err) { if (err instanceof DesktopModelMappingUnavailableError) return desktopMappingUnavailableResponse(err); if (err instanceof AnthropicRequestError) return anthropicErrorResponse(400, err.message); return anthropicErrorResponse(500, err instanceof Error ? err.message : String(err)); } finally { translatorBudget.dispose(); } if (!body || typeof body !== "object" || Array.isArray(body)) { return anthropicErrorResponse(400, "request body must be a JSON object"); } const raw = body as Rec; if (typeof raw.model !== "string" || raw.model.length === 0) { return anthropicErrorResponse(400, "model is required"); } try { let model = raw.model; // Case-insensitive [1m] strip (audit 021 #7 — the CLI matches /\[1m\]/i). const stripped = stripOneMillionMarker(model); if (stripped !== model) { model = stripped; raw.model = model; } // ocx-route override (devlog 072): keep count_tokens consistent with messages. const countRoute = extractOcxRouteDirective(raw); if (countRoute) { model = stripOneMillionMarker(countRoute); raw.model = model; } model = decodeNativeClaudePickerAlias(model, cc); raw.model = model; // Fast-only: count_tokens never parsed an effort row, so it must not start. It returns a // token estimate and sends no tier, so only the IDENTITY is corrected - without this the // synthetic id reaches native passthrough as a model Anthropic has never heard of. const countFastRow = parseFastOnlyRowId( config, () => decodeClaudeFastSelector(model, cc), ); if (countFastRow) { model = countFastRow.baseId; raw.model = model; } captureClaudeInbound("count_tokens", raw, resolveInboundModel(model, cc), req.headers.get("anthropic-beta") ?? undefined); if (wantsNativePassthrough(req, config, requestPolicy, model, cc)) { return await anthropicNativePassthrough(req, config, { model, provider: "anthropic-native", surface: "claude" }, undefined, raw, "/v1/messages/count_tokens"); } // A thread delta would undercount; refuse it exactly as the translated Messages path does. if (carriesMessageThread(raw)) return messageThreadUnsupportedResponse(); // PF-08: an eligible managed-key route counts the body the native lane would send. const nativeCountBody = (await import("./messages-native")).nativeMessagesCountBody( config, cc, raw, { fastRow: countFastRow !== null }, captureAnthropicClientIdentity(req.headers), ); // A count answers for the prompt a real turn from this model would forward, so it projects // the same unserialized content that turn's `message_start` floor does. Counting the raw // caller body instead reported replayed thinking this route never sends (#4857 family). const inputTokens = estimateClaudeRequestTokens( nativeCountBody ?? raw, model, thinkingProjectionForPreview(config, model), ); return new Response(JSON.stringify({ input_tokens: inputTokens }), { status: 200, headers: { "Content-Type": "application/json" }, }); } catch (error) { if (error instanceof DesktopModelMappingUnavailableError) return desktopMappingUnavailableResponse(error); if (error instanceof AnthropicRequestError) return anthropicErrorResponse(400, error.message); throw error; } }