From 25e73cb585ee83fa658106fc2e02de34b4189e7a Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sat, 22 Aug 2026 22:48:50 +0100 Subject: [PATCH] fix(routes): address review findings on the Anthropic Messages route Preserve usage when response translation fails after a successful, billed upstream call. The route previously discarded already-billed tokens on that path. A client could induce unlimited free usage against any budget via a truncated tool-call response. Reject a streaming request with 400 instead of 501. Check it before the rate limiter runs. A 501 made anthropic-sdk-python retry the request, since it retries any status of 500 or higher. Each retry also counted against the caller's rate limit before this fix. Preserve block order when translating an Anthropic message to OpenAI's shape. A tool_result block before other content must stay before it. OpenAI otherwise rejects the translated conversation. Echo the client's requested model alias in both the passthrough and the translated response paths. The passthrough branch previously returned the upstream's own model id instead. Forward the anthropic-beta header to an anthropic-type upstream through an explicit allowlist. Never forward Authorization, x-api-key, Cookie, or hop-by-hop headers. Let the adapter's own credential injection win regardless. Bill Anthropic's prompt-cache tokens. anthropicUsagePayload now reads cache_creation_input_tokens and cache_read_input_tokens too. It folds all three into the billed prompt count. This applies to this route, the existing outbound Anthropic path, and the native passthrough route. Extract the metered-request pipeline shared by the unified route and this route into runMeteredCall. Thread an envelope writer through it so each route answers in its own error shape. This removes the duplication the original brief warned against. It also lets the rate-limit and body-admission error paths answer in this route's own shape. Treat a non-boolean "stream" value as a streaming request. A string or number value previously bypassed the streaming check silently. Emit max_completion_tokens instead of max_tokens for OpenAI's reasoning models. Those models reject max_tokens with a 400. Preserve tool_result.is_error by prefixing string content with "Error: ". OpenAI's tool message carries no error field of its own. Log a warning when the thinking field is dropped instead of discarding it silently. Classify a response-translation failure distinctly in the log. An operator must never see a genuine translation bug reported as an upstream connection error. Document the route in README.md. Cover both auth headers, the non-streaming limitation, and which providers translate versus pass through. Add the reviewer's yaegi-check probes to tools/yaegi-check/main.go, adjusted for the fixes above. --- README.md | 94 ++++++++- routes_media.go | 10 +- routes_messages.go | 400 ++++++++++++++++++------------------ routes_messages_test.go | 263 ++++++++++++++++++++---- routes_passthrough.go | 13 +- routes_unified.go | 312 +++++++++++++++++++--------- tools/yaegi-check/main.go | 208 ++++++++++++++++++- translate_anthropic.go | 210 ++++++++++++++----- translate_anthropic_test.go | 198 +++++++++++++++++- 9 files changed, 1317 insertions(+), 391 deletions(-) diff --git a/README.md b/README.md index 880deaf..32d9dad 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,10 @@ combination. and from Anthropic's and Google Gemini's own wire formats, so any OpenAI-SDK-compatible client can talk to all three providers through one API shape. +- An Anthropic-compatible surface (`POST /v1/messages`) for Anthropic SDK + clients — including Claude Code — to talk to any configured provider in + Anthropic's own Messages API shape; see [Anthropic Messages API + (`/v1/messages`)](#anthropic-messages-api-v1messages). - Native passthrough per provider (`/{provider}/...`) for callers that want a provider's own wire format untouched, with the gateway injecting that provider's own API key and still enforcing auth and limits. @@ -929,6 +933,80 @@ only once it has something to authorize: will see every Gemini passthrough request denied. Route Gemini traffic for a model-restricted group through the unified `/v1/*` API instead. +## Anthropic Messages API (`/v1/messages`) + +`POST /v1/messages` serves Anthropic SDK clients — including Claude +Code — directly, in Anthropic's own Messages API request/response shape: +the inbound counterpart to the outbound-only Anthropic support the +unified routes already had. From model resolution through response +caching, usage accounting, and adapter-error handling, this route shares +one implementation with `/v1/chat/completions` +(`runMeteredCall`, `routes_unified.go`) — the same group authorization +(see [`GroupConfig`](#groupconfig)), [limits and +accounting](#limits-and-accounting), and [caching](#caching) semantics +apply. Only the wire shape and the client-facing error envelope are +specific to this route. + +### Authentication + +Anthropic SDKs authenticate with the `x-api-key` header, not +`Authorization: Bearer`. This route accepts **both** — the identical +credential lookup (`authStore.identify`, `auth.go`) every other route +uses already reads either header, `Authorization: Bearer` taking +priority if a request somehow carries both — so no separate credential +path exists for this route. A failed auth answers in Anthropic's own +error shape (`{"type":"error","error":{"type":"authentication_error", +"message":"..."}}`), never the OpenAI-shaped +`{"error":{"type":...,"code":...}}` every other route uses. + +### Translation vs. passthrough + +Behavior depends entirely on the resolved provider's configured `type`: + +| Resolved provider `type` | Behavior | +|---|---| +| `anthropic` | **Passthrough.** The request reaches that provider's own `/v1/messages` endpoint untranslated — no message-body translation of any kind. The response reaches the client unchanged, field for field, with one exception: `"model"` is rewritten to the client's own requested alias, the same routing rewrite every route applies before any adapter call — not a translation of the response body. | +| `openai` or `gemini` | **Full bidirectional translation**, through the same `providerAdapter.chatCompletion` a `/v1/chat/completions` request against that provider already uses — its contract is OpenAI-shape in, OpenAI-shape out regardless of provider type. The Anthropic-shaped request is translated to OpenAI's shape before the call; that provider's OpenAI-shaped response is translated back to Anthropic's shape before it reaches the client. | + +A few translation specifics worth knowing: + +- **`anthropic-beta` is forwarded** to an `anthropic`-type upstream (an + explicit allowlist — `Authorization`, `x-api-key`, `Cookie`, and every + hop-by-hop header are never forwarded, and the gateway's own credential + injection always overrides whatever the allowlist copied), so Claude + Code and the SDKs can still opt into 1M context, token-efficient + tools, computer use, and other beta features through this route. +- **Prompt-cache tokens are billed.** Anthropic's + `cache_creation_input_tokens` and `cache_read_input_tokens` fold into + the same combined prompt-token count every other route's budgets + track, alongside `input_tokens` — not dropped, and not priced at a + separate cache-tier rate. +- **`tool_result.is_error`** has no native OpenAI tool-message field; a + string `content` gets an `"Error: "` prefix so an OpenAI-backed + agentic loop still sees the failure signal. +- **Reasoning models** (OpenAI's `o1`/`o3`/`o4`/`gpt-5` families) get + `max_completion_tokens` instead of `max_tokens` — the field those + models reject with a 400. +- **`thinking`** (Anthropic's extended-thinking config) has no OpenAI + equivalent and is dropped when translating to an `openai`/`gemini` + target; the gateway logs a warning naming the dropped field rather + than discarding it silently. + +### Streaming is not yet supported + +`"stream"` set to anything other than the JSON literal `false` (or left +unset) gets a clean **400** `invalid_request_error` — never silently +ignored, never answered non-streamed. This is deliberately a 400, not a +501: Anthropic's own SDKs retry any status ≥ 500, so a 501 here would +turn every streaming call — the default for Claude Code — into a +multi-request retry storm against the caller's own request-rate limit. +The rejection happens *before* the rate limiter runs, so a +streaming-rejected request never consumes budget either. A follow-up +release adds the Anthropic SSE event translator +(`message_start`/`content_block_delta`/`message_stop`) — a genuinely +different event model from OpenAI's own stream shape this plugin already +forwards on `/v1/chat/completions`. + ## Image and audio endpoints `POST /v1/images/generations`, `POST /v1/audio/speech`, and @@ -1346,10 +1424,18 @@ the plugin's only non-standard-library runtime dependency. - **A request rejected before it reaches a provider never consumes request-rate quota.** On the unified routes, a malformed body, a missing `model` field, or an unknown/denied model (404/403) is rejected before - `limiter.checkAndCount` runs. On native passthrough and the MCP/A2A - proxy, denied provider/target access (403), an unsupported `Upgrade` - request (501), and a path-traversal attempt (400) are all checked before - the limiter too. Once a request *does* reach a provider adapter, + `limiter.checkAndCount` runs. `/v1/messages` goes further for one + specific field: it decodes the body and rejects a streaming request + (400) *before* the rate limiter runs too, the opposite relative order + from the unified routes' own admission-before-decode default — a + deliberate, route-scoped exception (see [Anthropic Messages + API](#anthropic-messages-api-v1messages)), since Anthropic SDKs retry + any 5xx and would otherwise burn multiple requests of budget per + logical streaming call against a route that does not support streaming + yet. On native passthrough and the MCP/A2A proxy, denied provider/target + access (403), an unsupported `Upgrade` request (501), and a + path-traversal attempt (400) are all checked before the limiter too. + Once a request *does* reach a provider adapter, though, a translate-time 400 (an unsupported parameter combination) or Anthropic's embeddings 501 still count toward the caller's request-rate limit — the limiter already ran by then — though neither ever bills a diff --git a/routes_media.go b/routes_media.go index da26ed7..8b1c6d6 100644 --- a/routes_media.go +++ b/routes_media.go @@ -42,7 +42,7 @@ const maxMultipartModelFieldBytes = 4096 // multipart body. A read failure or invalid/missing JSON is a 400, // already written to sw; ok reports whether the caller may proceed. func (g *Gateway) decodeMediaJSONRequest(sw *statusTrackingWriter, r *http.Request) (req map[string]any, model string, ok bool) { - release, admitted := g.acquireBodyAdmission(sw) + release, admitted := g.acquireBodyAdmission(sw, writeOAIError) defer release() if !admitted { return nil, "", false @@ -100,7 +100,7 @@ func (g *Gateway) resolveMediaModel(sw *statusTrackingWriter, grp *group, reques func (g *Gateway) handleImagesGenerations(w http.ResponseWriter, r *http.Request, u *user, grp *group) { sw := &statusTrackingWriter{ResponseWriter: w} - if _, ok := g.admitRequest(sw, u, grp); !ok { + if _, ok := g.admitRequest(sw, u, grp, writeOAIError); !ok { return } @@ -137,7 +137,7 @@ func (g *Gateway) handleImagesGenerations(w http.ResponseWriter, r *http.Request func (g *Gateway) handleAudioSpeech(w http.ResponseWriter, r *http.Request, u *user, grp *group) { sw := &statusTrackingWriter{ResponseWriter: w} - if _, ok := g.admitRequest(sw, u, grp); !ok { + if _, ok := g.admitRequest(sw, u, grp, writeOAIError); !ok { return } @@ -197,7 +197,7 @@ func (g *Gateway) handleAudioSpeech(w http.ResponseWriter, r *http.Request, u *u func (g *Gateway) handleAudioTranscriptions(w http.ResponseWriter, r *http.Request, u *user, grp *group) { sw := &statusTrackingWriter{ResponseWriter: w} - if _, ok := g.admitRequest(sw, u, grp); !ok { + if _, ok := g.admitRequest(sw, u, grp, writeOAIError); !ok { return } @@ -293,7 +293,7 @@ func (g *Gateway) handleAudioTranscriptions(w http.ResponseWriter, r *http.Reque // acquireBodyAdmission already wrote its own 503 to sw); body/oversize/ // err are readCapped's own results, valid only when admitted is true. func (g *Gateway) readAdmittedCapped(sw *statusTrackingWriter, r io.Reader, limit int64) (body []byte, oversize, admitted bool, err error) { - release, admitted := g.acquireBodyAdmission(sw) + release, admitted := g.acquireBodyAdmission(sw, writeOAIError) defer release() if !admitted { return nil, false, false, nil diff --git a/routes_messages.go b/routes_messages.go index 8e9e2fa..28a03ef 100644 --- a/routes_messages.go +++ b/routes_messages.go @@ -4,10 +4,8 @@ import ( "bytes" "context" "encoding/json" - "errors" "fmt" "io" - "math" "net/http" "strconv" ) @@ -30,31 +28,50 @@ const messagesPath = "/v1/messages" // cache hit across them would hand a client the wrong envelope entirely. const cacheEndpointMessages = "messages" +// anthropicPassthroughForwardedHeaders is the ALLOWLIST of client request +// headers callAnthropicMessagesPassthrough forwards to an anthropic-type +// upstream (item 5 fix, 2026-08-22 review): Claude Code and the Anthropic +// SDKs send anthropic-beta to opt into features this gateway must not +// silently disable on a route named "passthrough" — 1M context, +// token-efficient tools, computer use, among others. Deliberately an +// allowlist, never a blanket forward: Authorization, x-api-key, Cookie, +// and every hop-by-hop header must never reach an upstream this gateway +// itself authenticates against on the caller's behalf. injectAuth is +// always applied AFTER this copy (callAnthropicMessagesPassthrough, +// below), so it always wins if a future entry here ever collided with +// what injectAuth itself sets. +var anthropicPassthroughForwardedHeaders = []string{"anthropic-beta"} + // handleMessages implements POST /v1/messages: the Anthropic Messages API -// shape, inbound. It shares every pipeline step routes_unified.go's -// runUnified already established for /v1/chat/completions and -// /v1/embeddings — body-admission-gated decode, model resolution, -// request-rate limiting, response caching, usage accounting, and -// per-provider attempt recording — by calling the exact same functions -// runUnified calls, rather than reimplementing any of them: g.admitRequest, -// g.readAndDecodeUnifiedBody, g.registry.resolve, withAttemptRecorder, -// g.limiter.account, unifiedCostMicros, cacheKey/groupCacheEnabled/ -// effectiveTTL, and cacheCaptureWriter. Only two things are genuinely -// specific to this route and are NOT shared: which wire shape a request/ -// response is in (Anthropic, not OpenAI — see callAnthropicMessagesPassthrough/ -// callTranslatedMessages below) and which error envelope shape a failure -// gets written in (Anthropic's {"type":"error","error":{...}}, not -// OpenAI's {"error":{...}} — see writeAnthropicError and its callers, -// below). +// shape, inbound. From model resolution through response caching, usage +// accounting, and adapter-error handling, this route shares ONE +// implementation with routes_unified.go's runUnified — runMeteredCall, +// routes_unified.go (item 7 fix, 2026-08-22 review) — rather than +// duplicating that pipeline. Only two things are genuinely specific to +// this route: which wire shape a request/response is in (Anthropic, not +// OpenAI — see callAnthropicMessagesPassthrough/callTranslatedMessages +// below) and which error envelope shape a failure gets written in +// (Anthropic's {"type":"error","error":{...}}, not OpenAI's +// {"error":{...}} — see writeAnthropicError and its callers, below). +// +// ORDERING (item 2 fix, 2026-08-22 review): body admission + decode run +// BEFORE admitRequest (the rate-limit check) on THIS route — the +// opposite order from runUnified's own finding 1a. anthropic-sdk-python +// retries any status >= 500, and Claude Code streams on every call: the +// previous ordering (admitRequest first, streaming rejected only after +// decode) meant every real client hitting this not-yet-streaming-capable +// route burned three requests of rate-limit budget per logical call — +// verified repro: requestsPerMinute 2, three attempts -> 501, 501, 429, +// day counter reads 3. Deciding "is this streaming" requires the decoded +// body, so decode has to happen first; the trade-off finding 1a accepts +// for runUnified (an adversarial caller paying nothing to retry a +// malformed body) does not apply here, since the common case hitting +// this ordering is not adversarial retries but every legitimate +// streaming-capable client that exists today. func (g *Gateway) handleMessages(w http.ResponseWriter, r *http.Request, u *user, grp *group) { sw := &statusTrackingWriter{ResponseWriter: w} - scopes, ok := g.admitRequest(sw, u, grp) - if !ok { - return - } - - body, req, ok := g.readAndDecodeUnifiedBody(sw, r) + body, req, ok := g.readAndDecodeUnifiedBody(sw, r, writeAnthropicError) if !ok { return } @@ -65,99 +82,89 @@ func (g *Gateway) handleMessages(w http.ResponseWriter, r *http.Request, u *user return } - // Streaming is rejected cleanly and explicitly here, before model - // resolution or any upstream call: this cut of the route ships - // non-streaming only (brief scope) — a follow-up branch adds the - // Anthropic SSE event translator (message_start/content_block_delta/ - // message_stop), a genuinely different event model from OpenAI's - // stream shape this route already knows how to forward. A caller - // that asked for streaming must see a clear, typed error, never a - // silently-ignored flag answered non-streamed. - if streaming, _ := req["stream"].(bool); streaming { - writeAnthropicError(sw, http.StatusNotImplemented, "invalid_request_error", "streaming is not yet supported on /v1/messages") + // Streaming is rejected cleanly and explicitly here, before the + // rate-limit check, model resolution, or any upstream call: this cut + // of the route ships non-streaming only (brief scope) — a follow-up + // branch adds the Anthropic SSE event translator (message_start/ + // content_block_delta/message_stop), a genuinely different event + // model from OpenAI's stream shape this route already knows how to + // forward. A caller that asked for streaming must see a clear, typed + // error, never a silently-ignored flag answered non-streamed. + // + // 400, not 501 (item 2 fix, 2026-08-22 review): anthropic-sdk-python's + // retry classifies purely on status code and retries anything >= 500 + // — a 501 here turned every streaming call into an SDK-driven retry + // storm against this route's own rate limit (this function's own doc + // comment above has the verified repro). 400 maps to the SDK's + // non-retryable BadRequestError, and invalid_request_error is already + // the correct error type for a 400. + if isStreamingRequested(req) { + writeAnthropicError(sw, http.StatusBadRequest, "invalid_request_error", "streaming is not yet supported on /v1/messages") return } - adapter, upstreamModel, canonical, err := g.registry.resolve(requestedModel, grp) - if err != nil { - writeAnthropicModelResolveError(sw, err) + scopes, ok := g.admitRequest(sw, u, grp, writeAnthropicError) + if !ok { return } - req["model"] = upstreamModel - delete(req, gatewayAliasKey) // defensive: this route never sets it, but a client-sent "__alias" field must never reach either upstream shape - - cacheable := g.cache != nil && groupCacheEnabled(grp) - var cacheKeyStr string - if cacheable { - cacheKeyStr = cacheKey(adapter.name(), upstreamModel, requestedModel, cacheEndpointMessages, req) - if cached, hit := g.cache.lookup(cacheKeyStr); hit { - sw.Header().Set("X-Llmgw-Cache", "hit") - if cached.ContentType != "" { - sw.Header().Set("Content-Type", cached.ContentType) + g.runMeteredCall(sw, r, scopes, body, req, requestedModel, grp, cacheEndpointMessages, "messages route", writeAnthropicError, writeAnthropicProviderUpstreamError, + func(a providerAdapter, ctx context.Context, w http.ResponseWriter, req map[string]any) (usage, error) { + if a.typeName() == providerTypeAnthropic { + return g.callAnthropicMessagesPassthrough(ctx, w, r.Header, a, req, requestedModel) } - sw.WriteHeader(cached.Status) - _, _ = sw.Write(cached.Body) - return - } - sw.Header().Set("X-Llmgw-Cache", "miss") - } - - var respWriter http.ResponseWriter = sw - var capture *cacheCaptureWriter - if cacheable { - capture = newCacheCaptureWriter(sw, g.cache.maxBodyBytes) - respWriter = capture - } - - ctx := withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) { - g.limiter.recordProviderAttempt(adapter.name(), upstreamModel, resp, attemptErr) - }) - - var result usage - var callErr error - if adapter.typeName() == providerTypeAnthropic { - result, callErr = g.callAnthropicMessagesPassthrough(ctx, respWriter, adapter, req) - } else { - result, callErr = g.callTranslatedMessages(ctx, respWriter, adapter, req, requestedModel) - } - - // Zero-usage fallback, mirroring runUnified's own (routes_unified.go): - // a non-streaming failure always carries zero usage by contract, so - // this is a no-op whenever callErr != nil (account below skips every - // write once both total tokens and cost are zero). Unlike runUnified, - // there is no streaming branch to skip here — streaming was already - // rejected above, so every call into this route that reaches here is - // non-streaming. - if callErr == nil && result.total() == 0 { - result.prompt = int64(math.Ceil(float64(len(body)) / 4)) - result.estimated = true - } - g.limiter.account(scopes, result, unifiedCostMicros(canonical, upstreamModel, result, g.cfg.Pricing)) - if result.estimated { - g.logf("messages route: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", canonical, result.prompt) - } - - if cacheable && callErr == nil && capture.status == http.StatusOK && !capture.oversize { - g.cache.store(cacheKeyStr, capture.status, capture.contentType, capture.buf.Bytes(), effectiveTTL(g.cache, grp)) - } - - if callErr != nil { - g.handleMessagesAdapterError(sw, callErr, adapter.name()) - return - } + return g.callTranslatedMessages(ctx, w, a, req, requestedModel) + }) } +// isStreamingRequested reports whether req's "stream" field is anything +// other than the JSON literal false or an absent key (item 9 fix, +// 2026-08-22 review). A bare req["stream"].(bool) assertion silently +// treats a non-bool value ("stream":"true", "stream":1) as false and lets +// it through to an upstream that might itself honor a loosely-typed +// truthy value — exactly the "silently answered non-streamed" outcome +// this route must reject explicitly instead of allowing by accident. +func isStreamingRequested(req map[string]any) bool { + v, ok := req["stream"] + if !ok { + return false + } + return v != false +} + +// responseTranslationError wraps a failure translating an already- +// successful, already-billed upstream response into this route's own +// client-facing Anthropic shape (item 1/12 fix, 2026-08-22 review): the +// upstream call itself succeeded, so its usage travels back to the +// caller as call's own first return value regardless of this error (see +// callTranslatedMessages/callAnthropicMessagesPassthrough, below) — a +// client who can induce a translation failure (verified repro: a +// provider response with usage prompt=5000/completion=900, +// finish_reason "length", and a truncated tool-call arguments string) +// must not get unlimited free tokens against their budget just because +// this gateway's own re-encoding step broke after the provider already +// did, and was paid for, the work. Kept distinct from an errUpstream- +// wrapped error (a genuine connectivity/protocol failure) so +// handleAdapterErrorEnvelope's (routes_unified.go) generic "upstream +// connection error" branch never misclassifies a translation bug as a +// network problem in the log — see that function's own +// *responseTranslationError branch. +type responseTranslationError struct { + err error +} + +// Error implements the error interface. +func (e *responseTranslationError) Error() string { return "translate response: " + e.err.Error() } + // callAnthropicMessagesPassthrough handles a /v1/messages request already // resolved to an anthropic-type provider: forward req to that provider's // own Messages endpoint UNTRANSLATED (brief: "pass through, no -// translation of the message body") and copy its response back to w -// exactly as Anthropic sent it. req's only mutation, applied by the -// caller (handleMessages) before this function ever runs, is the "model" +// translation of the message body") and copy its response back to w. req's +// only mutation, applied by the shared pipeline (runMeteredCall, +// routes_unified.go) before this function ever runs, is the "model" // field rewrite to the resolved upstream model id — the same rewrite -// every route in this package applies before any adapter call -// (runUnified, resolveMediaModel); it is request ROUTING, not message -// translation. +// every route in this package applies before any adapter call; it is +// request ROUTING, not message translation. // // This deliberately bypasses providerAdapter.chatCompletion: // chatCompletion's contract (providers.go) is OpenAI-shape in, OpenAI- @@ -166,19 +173,27 @@ func (g *Gateway) handleMessages(w http.ResponseWriter, r *http.Request, u *user // would corrupt an already-Anthropic-shaped body instead of passing it // through. // -// The response is written back byte-for-byte: this function reads -// Anthropic's raw JSON only to extract usage (input_tokens/output_tokens, -// via the existing anthropicResponseBody type, translate_anthropic.go — -// reused rather than declared again) for accounting, and never re- -// encodes what reaches the client. One accepted consequence: the -// response's own "model" field carries the upstream model id Anthropic -// actually served, not the client's requested alias — unlike every -// translating adapter's gatewayAliasKey echo (routes_unified.go's own -// doc comment on that mechanism), rewriting it here would mean decoding -// and re-marshaling the full response body, which is itself a -// translation step the brief's "no translation of the message body" rule -// for this branch rules out. -func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.ResponseWriter, adapter providerAdapter, req map[string]any) (usage, error) { +// clientHeaders is the client's own incoming request headers, forwarded +// through anthropicPassthroughForwardedHeaders' allowlist only (item 5 +// fix, above). +// +// The response body reaches the client almost byte-for-byte: this +// function decodes it twice, read-only, to (a) extract usage +// (input_tokens/output_tokens/cache_*_tokens, via the existing +// anthropicResponseBody type, translate_anthropic.go — reused rather +// than declared again) for accounting, and (b) rewrite only the "model" +// field to requestedModel — the client's own requested alias — before +// re-encoding and writing it (item 4 fix, 2026-08-22 review: the +// PREVIOUS version left the upstream's own real model id in the +// response, inconsistent with callTranslatedMessages' own alias echo on +// the same route, and leaking the operator's real upstream model id, +// which gatewayAliasKey exists specifically to prevent). Every other +// field survives the round trip unchanged, since the rewrite decodes +// into a generic map rather than a narrow typed struct — rewriting one +// field this way is routing, not translating the message body. +func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.ResponseWriter, clientHeaders http.Header, adapter providerAdapter, req map[string]any, requestedModel string) (usage, error) { + delete(req, gatewayAliasKey) // this route never sets it via req itself, but a client-sent "__alias" field must never reach the real Anthropic API + // Plain type assertion, not an added providerAdapter interface method: // this branch only ever runs when the caller already confirmed // adapter.typeName() == providerTypeAnthropic, and buildAdapters @@ -195,6 +210,15 @@ func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.R hdr := http.Header{} hdr.Set("Content-Type", "application/json") + for _, name := range anthropicPassthroughForwardedHeaders { + for _, v := range clientHeaders.Values(name) { + hdr.Add(name, v) + } + } + // injectAuth runs AFTER the allowlist copy above, so it always wins + // (item 5 ruling) — in practice the two never collide (the allowlist + // carries no auth-shaped header), but the ordering is deliberate and + // documented, not incidental. adapter.injectAuth(&http.Request{Header: hdr}) resp, err := upstreamJSON(ctx, adapter.httpClient(), http.MethodPost, adapter.base()+anthropicMessagesPath, hdr, req, retry) @@ -213,8 +237,29 @@ func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.R } var parsed anthropicResponseBody - if err := json.Unmarshal(raw, &parsed); err != nil { - return usage{}, fmt.Errorf("%w: decode anthropic response: %w", errUpstream, err) + if unmarshalErr := json.Unmarshal(raw, &parsed); unmarshalErr != nil { + return usage{}, fmt.Errorf("%w: decode anthropic response: %w", errUpstream, unmarshalErr) + } + // totalInputTokens folds Anthropic's prompt-cache counters + // (cache_creation_input_tokens/cache_read_input_tokens) into the + // billed prompt count (item 6 fix, 2026-08-22 review) — see its own + // doc comment, translate_anthropic.go, for why dropping them + // under-bills a cache-heavy caller (the common case for Claude Code) + // by orders of magnitude. + billed := usage{prompt: parsed.Usage.totalInputTokens(), completion: parsed.Usage.OutputTokens} + + // The upstream call already succeeded and billed above by this + // point — any failure from here on is a *responseTranslationError, + // not a connectivity problem, so billed still reaches the caller + // even if the field rewrite below somehow fails (item 1/12 fix). + var envelope map[string]any + if unmarshalErr := json.Unmarshal(raw, &envelope); unmarshalErr != nil { + return billed, &responseTranslationError{err: fmt.Errorf("decode anthropic response envelope: %w", unmarshalErr)} + } + envelope["model"] = requestedModel + out, err := json.Marshal(envelope) + if err != nil { + return billed, &responseTranslationError{err: fmt.Errorf("re-marshal anthropic response: %w", err)} } if ct := resp.Header.Get("Content-Type"); ct != "" { @@ -223,9 +268,9 @@ func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.R w.Header().Set("Content-Type", "application/json") } w.WriteHeader(resp.StatusCode) - _, _ = w.Write(raw) + _, _ = w.Write(out) - return usage{prompt: parsed.Usage.InputTokens, completion: parsed.Usage.OutputTokens}, nil + return billed, nil } // callTranslatedMessages handles a /v1/messages request resolved to a @@ -239,11 +284,34 @@ func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.R // instead of letting it reach the client directly, then translate that // captured OpenAI-shaped response back to Anthropic's shape // (anthropicResponseFromOpenAI) before writing the result to w. +// +// USAGE ON A TRANSLATION FAILURE (item 1 fix, 2026-08-22 review): +// chatCompletion's own returned usage (result, below) is what the +// upstream actually reported and this gateway already billed the +// provider for — it is preserved and returned even when translating +// THAT successful response into Anthropic's shape fails afterward +// (verified repro: a response with usage prompt=5000/completion=900, +// finish_reason "length", and a tool-call whose arguments string was +// truncated mid-JSON by the provider's own max_tokens cutoff). The +// PREVIOUS version of this function returned usage{} on that path, +// discarding tokens the provider had already charged for — a client who +// could reliably induce a truncated tool-call response got unlimited +// free tokens against any budget, and this route was strictly less +// robust than /v1/chat/completions, which forwards that same malformed +// response to the client successfully instead of erroring on it. func (g *Gateway) callTranslatedMessages(ctx context.Context, w http.ResponseWriter, adapter providerAdapter, req map[string]any, requestedModel string) (usage, error) { - openaiReq, err := openAIRequestFromAnthropic(req) + openaiReq, dropped, err := openAIRequestFromAnthropic(req) if err != nil { return usage{}, err } + // thinking (Anthropic's extended-thinking config) and any other + // field this translator recognizes but cannot map has no OpenAI + // equivalent — logged rather than silently discarded (item 11 fix, + // 2026-08-22 review), so an operator can see a client's request was + // only partially honored instead of the drop being invisible. + for _, field := range dropped { + g.warnf("messages route: request field %q has no OpenAI equivalent for provider %q and was dropped in translation", field, adapter.name()) + } rec := newMessagesResponseRecorder() result, err := adapter.chatCompletion(ctx, rec, openaiReq) @@ -253,11 +321,11 @@ func (g *Gateway) callTranslatedMessages(ctx context.Context, w http.ResponseWri out, err := anthropicResponseFromOpenAI(rec.body.Bytes(), requestedModel) if err != nil { - return usage{}, err + return result, &responseTranslationError{err: err} } b, err := json.Marshal(out) if err != nil { - return usage{}, fmt.Errorf("%w: marshal translated response: %w", errUpstream, err) + return result, &responseTranslationError{err: fmt.Errorf("marshal translated response: %w", err)} } status := rec.status @@ -280,8 +348,9 @@ func (g *Gateway) callTranslatedMessages(ctx context.Context, w http.ResponseWri // so the adapter's write is captured here and never forwarded; the real // ResponseWriter only ever sees the translated bytes callTranslatedMessages // writes afterward. Streaming can never reach this recorder: handleMessages -// rejects "stream": true before either call path runs, so chatCompletion's -// non-streaming forwardJSON is the only path that ever writes to it. +// rejects a streaming request before either call path runs, so +// chatCompletion's non-streaming forwardJSON is the only path that ever +// writes to it. type messagesResponseRecorder struct { header http.Header body bytes.Buffer @@ -311,6 +380,8 @@ func (r *messagesResponseRecorder) Write(b []byte) (int, error) { // rate_limit_error, server_error), so an operator reading logs sees one // consistent taxonomy regardless of which route answered; only the JSON // shape around it differs, to match what this route's own clients expect. +// Matches envelopeWriter's signature (routes_unified.go), so this route's +// error paths thread straight into the shared pipeline there. func writeAnthropicError(w http.ResponseWriter, status int, errType, msg string) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) @@ -320,81 +391,14 @@ func writeAnthropicError(w http.ResponseWriter, status int, errType, msg string) }) } -// writeAnthropicModelResolveError is writeModelResolveError's -// (routes_unified.go) Anthropic-shaped counterpart: the SAME status-code -// decision (aliasTargetError/errModelUnknown/errModelDenied), written -// through writeAnthropicError instead of writeOAIError. Kept as its own -// small function, rather than a routes_unified.go refactor threading an -// envelope-writer parameter through, so this route's addition stays -// isolated to its own file while a sibling branch is concurrently editing -// registry.go/admin.go — the decision logic below is intentionally -// identical to writeModelResolveError's, not independently re-derived. -func writeAnthropicModelResolveError(w http.ResponseWriter, err error) { - if aerr, ok := err.(*aliasTargetError); ok { - writeAnthropicError(w, http.StatusNotFound, "invalid_request_error", aerr.Error()) - return - } - switch { - case errors.Is(err, errModelUnknown): - writeAnthropicError(w, http.StatusNotFound, "invalid_request_error", "unknown model") - case errors.Is(err, errModelDenied): - writeAnthropicError(w, http.StatusForbidden, "invalid_request_error", "model access denied") - default: - writeAnthropicError(w, http.StatusInternalServerError, "server_error", "internal error") - } -} - -// handleMessagesAdapterError is handleAdapterError's (routes_unified.go) -// Anthropic-shaped counterpart, covering the identical set of adapter -// error cases with the identical yaegi-safety rationale (plain type -// assertions for providerHTTPError/translateError, never errors.As — -// see providerHTTPError's own doc comment, providers.go) but writing -// through writeAnthropicError/writeAnthropicProviderUpstreamError instead -// of writeOAIError/writeProviderUpstreamError, and its own "messages -// route:" log prefix so a log line always names which route it came from. -func (g *Gateway) handleMessagesAdapterError(sw *statusTrackingWriter, err error, providerName string) { - sw.Header().Del("X-Llmgw-Cache") - - if sw.wroteHeader { - g.errorf("messages route: adapter error after response started (provider %q): %v", providerName, err) - return - } - - if perr, ok := err.(*providerHTTPError); ok { - writeAnthropicProviderUpstreamError(sw, providerName, perr) - return - } - - if terr, ok := err.(*translateError); ok { - status := http.StatusBadRequest - if terr.notSupported { - status = http.StatusNotImplemented - } - writeAnthropicError(sw, status, "invalid_request_error", terr.msg) - return - } - - if errors.Is(err, context.Canceled) { - g.logf("messages route: client canceled request to provider %q: %v", providerName, err) - return - } - - g.errorf("messages route: upstream connection error (provider %q): %v", providerName, err) - writeAnthropicError(sw, http.StatusBadGateway, "server_error", "upstream connection error") -} - // writeAnthropicProviderUpstreamError is writeProviderUpstreamError's // (routes_unified.go) Anthropic-shaped counterpart: perr's upstream body -// is embedded under error.upstream (decoded to a JSON value when it -// parses as one, left as a raw string otherwise) inside the Anthropic -// {"type":"error","error":{...}} envelope instead of the OpenAI one. +// is embedded under error.upstream (decodeUpstreamErrorBody, +// routes_unified.go — shared with writeProviderUpstreamError, item 7/8 +// fix) inside the Anthropic {"type":"error","error":{...}} envelope +// instead of the OpenAI one. Matches providerUpstreamErrorWriter's +// signature (routes_unified.go). func writeAnthropicProviderUpstreamError(w http.ResponseWriter, providerName string, perr *providerHTTPError) { - var upstream any = string(perr.body) - var parsed any - if json.Unmarshal(perr.body, &parsed) == nil { - upstream = parsed - } - w.Header().Set("Content-Type", "application/json") w.WriteHeader(perr.status) _ = json.NewEncoder(w).Encode(map[string]any{ // headers already committed; nothing useful to do on encode failure @@ -403,7 +407,7 @@ func writeAnthropicProviderUpstreamError(w http.ResponseWriter, providerName str "type": "upstream_error", "message": providerName + " upstream error", "code": strconv.Itoa(perr.status), - "upstream": upstream, + "upstream": decodeUpstreamErrorBody(perr), }, }) } diff --git a/routes_messages_test.go b/routes_messages_test.go index f8856d1..46b0521 100644 --- a/routes_messages_test.go +++ b/routes_messages_test.go @@ -58,19 +58,25 @@ func newMessagesTestGateway(t *testing.T, anthropicURL, openaiURL string) *Gatew // TestHandleMessages_AnthropicProvider_Passthrough proves a /v1/messages // request resolved to an anthropic-type provider is forwarded to that -// provider's own Messages endpoint and its response copied back to the -// client byte-for-byte (brief: "pass through, no translation of the -// message body"), while still authenticating upstream and accounting -// usage exactly like every other route. +// provider's own Messages endpoint UNTRANSLATED (brief: "pass through, no +// translation of the message body") and its response copied back to the +// client with every field preserved except "model", which is rewritten to +// the client's own requested alias (item 4 fix, 2026-08-22 review: a +// routing rewrite, not message-body translation — every OTHER field, +// including ones this translator does not itself know about, survives +// unchanged), while still authenticating upstream, forwarding the +// anthropic-beta header (item 5), billing prompt-cache tokens (item 6), +// and accounting usage exactly like every other route. func TestHandleMessages_AnthropicProvider_Passthrough(t *testing.T) { - const anthResp = `{"id":"msg_01ABC","type":"message","role":"assistant","content":[{"type":"text","text":"hi there"}],"model":"claude-real","stop_reason":"end_turn","usage":{"input_tokens":12,"output_tokens":6}}` + const anthResp = `{"id":"msg_01ABC","type":"message","role":"assistant","content":[{"type":"text","text":"hi there"}],"model":"claude-real","stop_reason":"end_turn","stop_sequence":null,"usage":{"input_tokens":12,"output_tokens":6,"cache_creation_input_tokens":100,"cache_read_input_tokens":50}}` var gotBody map[string]any - var gotAPIKey, gotVersion, gotPath string + var gotAPIKey, gotVersion, gotPath, gotBeta string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { gotPath = r.URL.Path gotAPIKey = r.Header.Get("x-api-key") gotVersion = r.Header.Get("anthropic-version") + gotBeta = r.Header.Get("anthropic-beta") _ = json.NewDecoder(r.Body).Decode(&gotBody) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) @@ -86,15 +92,34 @@ func TestHandleMessages_AnthropicProvider_Passthrough(t *testing.T) { "messages": []any{map[string]any{"role": "user", "content": "hello"}}, } req := newMessagesRequest(t, body, "x-api-key", "sk-alice") + req.Header.Set("anthropic-beta", "token-efficient-tools-2025-02-19") rec := httptest.NewRecorder() gw.ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) } - if rec.Body.String() != anthResp { - t.Errorf("body = %q, want verbatim upstream body %q (no translation)", rec.Body.String(), anthResp) + + var got map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode client response: %v", err) } + if got["model"] != "claude-test" { + t.Errorf(`response["model"] = %v, want "claude-test" (the client's requested alias echoed back, item 4 fix — not the upstream's real model id)`, got["model"]) + } + // Every other field must survive the passthrough unchanged, including + // ones this gateway does not itself interpret (stop_sequence). + if got["id"] != "msg_01ABC" || got["role"] != "assistant" || got["stop_reason"] != "end_turn" { + t.Errorf("response id/role/stop_reason = %v/%v/%v, want the upstream's own values preserved", got["id"], got["role"], got["stop_reason"]) + } + if _, hasStopSeq := got["stop_sequence"]; !hasStopSeq { + t.Error(`response missing "stop_sequence" — a field this gateway never reads must still survive the passthrough`) + } + content, _ := got["content"].([]any) + if len(content) != 1 { + t.Fatalf("response content = %v, want the upstream's one text block preserved", got["content"]) + } + if gotPath != anthropicMessagesPath { t.Errorf("upstream path = %q, want %q", gotPath, anthropicMessagesPath) } @@ -104,6 +129,9 @@ func TestHandleMessages_AnthropicProvider_Passthrough(t *testing.T) { if gotVersion != anthropicAPIVersion { t.Errorf("upstream anthropic-version = %q, want %q", gotVersion, anthropicAPIVersion) } + if gotBeta != "token-efficient-tools-2025-02-19" { + t.Errorf("upstream anthropic-beta = %q, want the client's header forwarded (item 5 fix)", gotBeta) + } if gotBody["model"] != "claude-test" { t.Errorf("upstream model = %v, want claude-test (resolved upstream id)", gotBody["model"]) } @@ -111,9 +139,11 @@ func TestHandleMessages_AnthropicProvider_Passthrough(t *testing.T) { t.Error("upstream request carries gatewayAliasKey; must never reach an upstream provider") } + // 12 input + 100 cache-creation + 50 cache-read (item 6 fix): all + // three fold into the billed prompt count, not just fresh input. tokIn, ok := gw.limiter.getCounter("user", "alice", metricTokIn, windowDay, time.Now()) - if !ok || tokIn != 12 { - t.Errorf("user tokin/day counter = %d (ok=%v), want 12", tokIn, ok) + if !ok || tokIn != 162 { + t.Errorf("user tokin/day counter = %d (ok=%v), want 162 (12 input + 100 cache-creation + 50 cache-read)", tokIn, ok) } tokOut, ok := gw.limiter.getCounter("user", "alice", metricTokOut, windowDay, time.Now()) if !ok || tokOut != 6 { @@ -266,41 +296,95 @@ func TestHandleMessages_Auth(t *testing.T) { } } -// TestHandleMessages_StreamingRejected proves "stream": true is rejected -// cleanly with a typed, Anthropic-shaped error before any upstream call -// is made — never silently ignored, never answered non-streamed. +// TestHandleMessages_StreamingRejected table-drives every "stream" shape +// this route must reject cleanly, with a typed, Anthropic-shaped error, +// before any upstream call is made or any rate-limit budget is spent — +// never silently ignored, never answered non-streamed. Covers a real +// boolean true and, per item 9's fix (2026-08-22 review), a non-bool +// truthy value ("stream":"true", a string, and "stream":1, a number) that +// a bare req["stream"].(bool) assertion would previously have silently +// treated as false and forwarded upstream. func TestHandleMessages_StreamingRejected(t *testing.T) { - called := false - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - called = true - w.WriteHeader(http.StatusOK) - })) - defer srv.Close() + cases := []struct { + stream any + name string + }{ + {name: "boolean true", stream: true}, + {name: "string \"true\" (non-bool, item 9)", stream: "true"}, + {name: "number 1 (non-bool, item 9)", stream: float64(1)}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + called := false + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() - gw := newMessagesTestGateway(t, srv.URL, "") - body := map[string]any{"model": "claude-test", "max_tokens": 10, "stream": true, "messages": []any{map[string]any{"role": "user", "content": "hi"}}} - req := newMessagesRequest(t, body, "x-api-key", "sk-alice") - rec := httptest.NewRecorder() - gw.ServeHTTP(rec, req) + // requestsPerMinute: 1 so a second wrongly-counted request + // would 429 — proving the rejection happens before + // admitRequest ever runs (item 2 fix): both requests below + // must succeed at the "not yet counted" stage. + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{"anthropic": {Type: providerTypeAnthropic, BaseURL: srv.URL, APIKey: "sk-ant-up", Models: []string{"claude-test"}}} // #nosec G101 -- test fixture literal, not a real credential + cfg.Groups = map[string]*GroupConfig{"default": {}} + cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{RequestsPerMinute: 1}}}} // #nosec G101 -- test fixture literal, not a real credential + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + if err != nil { + t.Fatalf("New: %v", err) + } + gw, ok := h.(*Gateway) + if !ok { + t.Fatal("handler is not *Gateway") + } - if rec.Code != http.StatusNotImplemented { - t.Fatalf("status = %d, want 501, body=%s", rec.Code, rec.Body.String()) - } - if called { - t.Error("upstream was called; streaming must be rejected before any upstream request") - } + body := map[string]any{"model": "claude-test", "max_tokens": 10, "stream": tc.stream, "messages": []any{map[string]any{"role": "user", "content": "hi"}}} - var got map[string]any - if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { - t.Fatalf("decode error body: %v", err) - } - if got["type"] != "error" { - t.Errorf(`response["type"] = %v, want "error"`, got["type"]) - } - errObj, _ := got["error"].(map[string]any) - msg, _ := errObj["message"].(string) - if !strings.Contains(msg, "stream") { - t.Errorf("error message = %q, want it to mention streaming is not supported", msg) + for attempt := 1; attempt <= 2; attempt++ { + req := newMessagesRequest(t, body, "x-api-key", "sk-alice") + rec := httptest.NewRecorder() + gw.ServeHTTP(rec, req) + + // 400, not 501 (item 2 fix): anthropic-sdk-python's retry + // classifies purely on status code and retries anything + // >= 500 — a 501 here turns every streaming call into a + // three-request SDK retry storm against this route's own + // rate limit. + if rec.Code != http.StatusBadRequest { + t.Fatalf("attempt %d: status = %d, want 400, body=%s", attempt, rec.Code, rec.Body.String()) + } + + var got map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode error body: %v", err) + } + if got["type"] != "error" { + t.Errorf(`response["type"] = %v, want "error"`, got["type"]) + } + errObj, _ := got["error"].(map[string]any) + if errObj["type"] != "invalid_request_error" { + t.Errorf("error.type = %v, want invalid_request_error", errObj["type"]) + } + msg, _ := errObj["message"].(string) + if !strings.Contains(msg, "stream") { + t.Errorf("error message = %q, want it to mention streaming is not supported", msg) + } + } + if called { + t.Error("upstream was called; streaming must be rejected before any upstream request") + } + + // item 2's actual fix: TWO streaming-rejected attempts against + // a requestsPerMinute:1 user must BOTH succeed (as 400s, not + // as a 429 on the second one) — proving admitRequest's + // checkAndCount never ran for either. + reqCount, ok := gw.limiter.getCounter("user", "alice", metricReq, windowMin, time.Now()) + if !ok || reqCount != 0 { + t.Errorf("user req:min counter = %d (ok=%v), want 0 — a streaming-rejected request must never consume rate-limit budget", reqCount, ok) + } + }) } } @@ -356,8 +440,105 @@ func TestHandleMessages_RateLimited_Returns429(t *testing.T) { t.Errorf("upstream called %d times, want 1 (second request refused before any upstream call)", upstreamCalls) } + // Envelope-shape assertion (item 8 fix, 2026-08-22 review): now that + // admitRequest threads an envelopeWriter through + // (routes_unified.go's writeLimitViolationEnvelope), a 429 on this + // route must answer in Anthropic's shape like every other error this + // route produces, not OpenAI's flat one. + var got map[string]any + if err := json.Unmarshal(rec2.Body.Bytes(), &got); err != nil { + t.Fatalf("decode 429 body: %v", err) + } + if got["type"] != "error" { + t.Errorf(`429 response["type"] = %v, want "error" (Anthropic's envelope shape, not OpenAI's flat one)`, got["type"]) + } + errObj, _ := got["error"].(map[string]any) + if errObj["type"] != "rate_limit_error" { + t.Errorf("429 error.type = %v, want rate_limit_error", errObj["type"]) + } + if _, hasTopLevelCode := got["code"]; hasTopLevelCode { + t.Error(`429 response carries a top-level "code" field — that is OpenAI's writeOAIError shape, not Anthropic's`) + } + tokIn, ok := gw.limiter.getCounter("user", "alice", metricTokIn, windowDay, time.Now()) if !ok || tokIn != 1 { t.Errorf("user tokin/day counter = %d (ok=%v), want 1 (only the first, successful request accounted)", tokIn, ok) } } + +// TestHandleMessages_TranslationFailureAfterBilledUsage_PreservesUsage is +// the regression for item 1 (CRITICAL, 2026-08-22 review): an +// openai-type provider's response can succeed (200, real usage the +// provider already charged for) yet still fail THIS route's own +// translation into Anthropic's shape — here, a tool-call arguments +// string truncated mid-JSON by the provider's own max_tokens cutoff +// (finish_reason "length"). The previous version of callTranslatedMessages +// returned usage{} on that path, discarding tokens the provider had +// already billed — a client who could reliably induce a truncated +// tool-call response got unlimited free tokens against any budget. This +// proves the client gets a clean 502 (translation genuinely failed, no +// valid Anthropic response could be produced) while the usage still +// lands in the limiter's counters, and (item 12) that the log line +// names the real cause — "response translation failed", never "upstream +// connection error", which would misdirect an operator toward a network +// problem that never happened. +func TestHandleMessages_TranslationFailureAfterBilledUsage_PreservesUsage(t *testing.T) { + // Exact verified repro from the review: usage prompt=5000/ + // completion=900, finish_reason "length", and a tool_calls[].function. + // arguments string truncated mid-object. + const oaiResp = `{"id":"chatcmpl-trunc","object":"chat.completion","model":"gpt-real","choices":[{"index":0,"message":{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","type":"function","function":{"name":"get_weather","arguments":"{\"city\": \"Lond"}}]},"finish_reason":"length"}],"usage":{"prompt_tokens":5000,"completion_tokens":900}}` + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(oaiResp)) + })) + defer srv.Close() + + gw := newMessagesTestGateway(t, "", srv.URL) + + body := map[string]any{ + "model": "gpt-test", + "max_tokens": 100, + "messages": []any{map[string]any{"role": "user", "content": "hello"}}, + } + req := newMessagesRequest(t, body, "x-api-key", "sk-alice") + var rec *httptest.ResponseRecorder + logs := captureStderr(t, func() { + rec = httptest.NewRecorder() + gw.ServeHTTP(rec, req) + }) + + if rec.Code != http.StatusBadGateway { + t.Fatalf("status = %d, want 502, body=%s", rec.Code, rec.Body.String()) + } + + var got map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode error body: %v", err) + } + if got["type"] != "error" { + t.Errorf(`response["type"] = %v, want "error" (Anthropic error shape)`, got["type"]) + } + + // item 1: usage the provider already billed must still land in the + // counters, even though the client got an error. + tokIn, ok := gw.limiter.getCounter("user", "alice", metricTokIn, windowDay, time.Now()) + if !ok || tokIn != 5000 { + t.Errorf("user tokin/day counter = %d (ok=%v), want 5000 (the upstream's already-billed usage, not discarded on translation failure)", tokIn, ok) + } + tokOut, ok := gw.limiter.getCounter("user", "alice", metricTokOut, windowDay, time.Now()) + if !ok || tokOut != 900 { + t.Errorf("user tokout/day counter = %d (ok=%v), want 900", tokOut, ok) + } + + // item 12: the log must name this a translation failure, never an + // upstream connection error — the connection and the upstream call + // both succeeded fine. + if !strings.Contains(logs, "response translation failed") { + t.Errorf("log output = %q, want it to contain \"response translation failed\"", logs) + } + if strings.Contains(logs, "upstream connection error") { + t.Errorf("log output = %q, want it to NOT be classified as an upstream connection error", logs) + } +} diff --git a/routes_passthrough.go b/routes_passthrough.go index 00b9edc..6ed4d71 100644 --- a/routes_passthrough.go +++ b/routes_passthrough.go @@ -426,6 +426,17 @@ type passthroughUsagePayload struct { CompletionTokens int64 `json:"completion_tokens"` InputTokens int64 `json:"input_tokens"` OutputTokens int64 `json:"output_tokens"` + // CacheCreationInputTokens/CacheReadInputTokens (item 6 fix, + // 2026-08-22 review — applied "for the anthropic provider + // generally", not only translate_anthropic.go's own + // anthropicUsagePayload): Anthropic's native passthrough + // response reports these two prompt-cache counters separately + // from InputTokens, and extractPassthroughUsage's own + // providerTypeAnthropic case folds them in below so a + // cache-heavy caller's usage is not silently under-billed on + // this route either. + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens"` } `json:"usage"` UsageMetadata struct { PromptTokenCount int64 `json:"promptTokenCount"` @@ -459,7 +470,7 @@ func extractPassthroughUsage(typeName, providerName string, body []byte) (usage, case providerTypeOpenAI: return usage{prompt: payload.Usage.PromptTokens, completion: payload.Usage.CompletionTokens}, model, err case providerTypeAnthropic: - return usage{prompt: payload.Usage.InputTokens, completion: payload.Usage.OutputTokens}, model, err + return usage{prompt: payload.Usage.InputTokens + payload.Usage.CacheCreationInputTokens + payload.Usage.CacheReadInputTokens, completion: payload.Usage.OutputTokens}, model, err case providerTypeGemini: return usage{prompt: payload.UsageMetadata.PromptTokenCount, completion: payload.UsageMetadata.CandidatesTokenCount}, model, err default: diff --git a/routes_unified.go b/routes_unified.go index cd03a08..143404e 100644 --- a/routes_unified.go +++ b/routes_unified.go @@ -116,8 +116,29 @@ func (g *Gateway) handleModels(w http.ResponseWriter, r *http.Request) { // handleChat and handleEmbeddings share every other step of the pipeline // below (decode, resolve, enforce limits, account usage, translate // errors) and differ only in which of these two methods they invoke. +// routes_messages.go's handleMessages reuses the identical contract for +// its own anthropic-passthrough/translate branching (see its own doc +// comment) — the shape is exactly what runMeteredCall, below, needs +// regardless of which wire shape a route answers in. type adapterCall func(a providerAdapter, ctx context.Context, w http.ResponseWriter, req map[string]any) (usage, error) +// envelopeWriter abstracts the client-facing error envelope shape so the +// shared metered-request pipeline (runMeteredCall, below) can serve both +// the OpenAI-shaped unified/media routes and the Anthropic-shaped +// /v1/messages route through one implementation (item 7, 2026-08-22 +// review — fixing 58 of handleMessages' 73 lines being verbatim +// duplicated from runUnified) rather than duplicating it. writeOAIError +// (errors.go) and writeAnthropicError (routes_messages.go) both already +// match this exact signature. +type envelopeWriter func(w http.ResponseWriter, status int, errType, msg string) + +// providerUpstreamErrorWriter abstracts which envelope shape a +// *providerHTTPError gets embedded into (writeProviderUpstreamError, +// below, or routes_messages.go's writeAnthropicProviderUpstreamError) — +// threaded through runMeteredCall/handleAdapterErrorEnvelope alongside +// envelopeWriter for the same reason. +type providerUpstreamErrorWriter func(w http.ResponseWriter, providerName string, perr *providerHTTPError) + // handleChat implements POST /v1/chat/completions. func (g *Gateway) handleChat(w http.ResponseWriter, r *http.Request, u *user, grp *group) { g.runUnified(w, r, u, grp, cacheEndpointChat, func(a providerAdapter, ctx context.Context, w http.ResponseWriter, req map[string]any) (usage, error) { @@ -134,15 +155,14 @@ func (g *Gateway) handleEmbeddings(w http.ResponseWriter, r *http.Request, u *us // runUnified is the shared chat/embeddings pipeline: enforce per-user/ // per-group/total REQUEST-RATE limits, claim a body-admission slot to -// decode the request, resolve its model, invoke the adapter via call, -// then account the resulting usage — even when call itself returned an -// error, so usage captured before a mid-stream failure still gets billed -// — before translating that error into a response. w is wrapped in its -// own statusTrackingWriter so a mid-stream adapter error (headers -// already sent) can be told apart from one that failed before any -// write. endpoint is cacheEndpointChat or cacheEndpointEmbeddings — one -// of cacheKey's key-material components (cache.go), so the two routes -// never collide into one cache entry. +// decode the request, then hand off to runMeteredCall (below) for +// everything from model resolution through response caching, usage +// accounting, and adapter-error handling. w is wrapped in its own +// statusTrackingWriter so a mid-stream adapter error (headers already +// sent) can be told apart from one that failed before any write. +// endpoint is cacheEndpointChat or cacheEndpointEmbeddings — one of +// cacheKey's key-material components (cache.go), so the two routes never +// collide into one cache entry. // // ORDERING (security review finding 1a, 2026-08-22): the rate-limit // check runs BEFORE the body is ever read, not after — the previous @@ -182,15 +202,26 @@ func (g *Gateway) handleEmbeddings(w http.ResponseWriter, r *http.Request, u *us // cannot even form a valid request is now itself rate-limited rather // than getting unlimited free retries — but are called out explicitly // here since neither is visible from the diff alone. +// +// SHARED CORE (item 7 fix, 2026-08-22 review): routes_messages.go's +// handleMessages needs admitRequest and the body decode in a DIFFERENT +// relative order than this route does (its own doc comment explains +// why — a streaming request must never touch the rate-limit counters at +// all, which means it has to know "is this streaming" from the decoded +// body before admitRequest ever runs). That is the one genuine ordering +// difference between the two routes, and it is why THIS function still +// owns admitRequest/readAndDecodeUnifiedBody directly rather than folding +// them into runMeteredCall too — everything after both are done is +// identical regardless of wire shape, and lives in runMeteredCall alone. func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, grp *group, endpoint string, call adapterCall) { sw := &statusTrackingWriter{ResponseWriter: w} - scopes, ok := g.admitRequest(sw, u, grp) + scopes, ok := g.admitRequest(sw, u, grp, writeOAIError) if !ok { return } - body, req, ok := g.readAndDecodeUnifiedBody(sw, r) + body, req, ok := g.readAndDecodeUnifiedBody(sw, r, writeOAIError) if !ok { return } @@ -200,11 +231,35 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr writeOAIError(sw, http.StatusBadRequest, "invalid_request_error", "model is required") return } - streaming, _ := req["stream"].(bool) + g.runMeteredCall(sw, r, scopes, body, req, requestedModel, grp, endpoint, "unified route", writeOAIError, writeProviderUpstreamError, call) +} + +// runMeteredCall is the shared implementation from model resolution +// through response caching, usage accounting, and adapter-error handling +// — the part of every metered LLM route (this file's runUnified, above; +// routes_messages.go's handleMessages) that is genuinely identical +// regardless of wire shape (item 7 fix, 2026-08-22 review: this body used +// to be duplicated almost verbatim in routes_messages.go). Callers have +// already run admitRequest (rate limiting) and decoded the request body +// into req, in whichever relative order their own route needs (see +// runUnified's own doc comment for why that order differs for +// /v1/messages) — this function starts only once both are done. +// +// endpoint is cacheKey's endpoint argument (cacheEndpointChat/ +// cacheEndpointEmbeddings/routes_messages.go's cacheEndpointMessages). +// envelope and writeUpstream pick the client-facing error shape. call is +// adapterCall — how the wire-shape-specific request/response translation, +// if any, actually happens; the routes differ ONLY here (call's own +// closure already has whatever it needs — routes_messages.go's own +// requestedModel for its alias echo, in particular) and in which error +// envelope shape they answer in. logPrefix names the route in every log +// line this function writes, so a log reader can always tell which route +// produced a given line without the classification logic existing twice. +func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scopes []limitScope, body []byte, req map[string]any, requestedModel string, grp *group, endpoint, logPrefix string, envelope envelopeWriter, writeUpstream providerUpstreamErrorWriter, call adapterCall) { adapter, upstreamModel, canonical, err := g.registry.resolve(requestedModel, grp) if err != nil { - writeModelResolveError(sw, err) + writeModelResolveErrorEnvelope(sw, err, envelope) return } @@ -224,6 +279,7 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr // clients requesting the same upstream model under different alias // forms (e.g. "claude-x" vs "anthropic/claude-x") must never collide // into one cache entry (cacheKey's own doc comment, cache.go). + streaming, _ := req["stream"].(bool) cacheable := !streaming && g.cache != nil && groupCacheEnabled(grp) var cacheKeyStr string if cacheable { @@ -252,8 +308,8 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr // respWriter is sw, wrapped in a capture tee only when this request // is cacheable — cacheCaptureWriter buffers everything written so a // 200 non-stream response can be stored after call() returns, without - // any adapter knowing caching exists (routes_unified.go's adapterCall - // contract is unchanged either way). + // any adapter knowing caching exists (adapterCall's contract is + // unchanged either way). var respWriter http.ResponseWriter = sw var capture *cacheCaptureWriter if cacheable { @@ -280,9 +336,14 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr // translateError, or a connection failure before any write) always // carries zero usage by contract, so accounting it here is a no-op — // account skips every write once both total tokens and cost are zero. + // routes_messages.go's callTranslatedMessages/ + // callAnthropicMessagesPassthrough follow the identical contract for + // their own error paths (item 1 fix, 2026-08-22 review: a translation + // failure AFTER a successful, billed upstream call must still return + // that call's real usage, not usage{}). if callErr == nil && result.total() == 0 { if streaming { - g.logf("unified route: zero usage reported for a streaming response from model %q; accounting the request only", canonical) + g.logf("%s: zero usage reported for a streaming response from model %q; accounting the request only", logPrefix, canonical) } else { result.prompt = int64(math.Ceil(float64(len(body)) / 4)) result.estimated = true @@ -290,7 +351,7 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr } g.limiter.account(scopes, result, unifiedCostMicros(canonical, upstreamModel, result, g.cfg.Pricing)) if result.estimated { - g.logf("unified route: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", canonical, result.prompt) + g.logf("%s: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", logPrefix, canonical, result.prompt) } // A miss stores the response after everything above has already run — @@ -310,7 +371,7 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr } if callErr != nil { - g.handleAdapterError(sw, callErr, adapter.name()) + g.handleAdapterErrorEnvelope(sw, callErr, adapter.name(), logPrefix, envelope, writeUpstream) return } } @@ -385,19 +446,23 @@ func withTotalScope(scopes []limitScope) []limitScope { // caller can reuse the identical scope slice for its own later account // call, rather than rebuilding (and risking scope-list drift from) a // second buildLimitScopes/withTotalScope pair. A violation writes its -// response to sw itself (writeLimitViolation, unchanged body/headers from -// before this fix) and returns ok=false. +// response to sw itself, via envelope (writeLimitViolationEnvelope, +// below), and returns ok=false. // -// Shared verbatim by every unified/media route (runUnified above; -// handleImagesGenerations/handleAudioSpeech/handleAudioTranscriptions, -// routes_media.go) — see runUnified's own "BEHAVIOR CHANGE" doc-comment +// Shared verbatim by every unified/media/messages route (runUnified +// above; handleImagesGenerations/handleAudioSpeech/ +// handleAudioTranscriptions, routes_media.go; handleMessages, +// routes_messages.go) — see runUnified's own "BEHAVIOR CHANGE" doc-comment // section for the two previously-silent side effects this reordering has // for a caller who is both over budget and sending a malformed/ -// unroutable request; they apply identically here. -func (g *Gateway) admitRequest(sw *statusTrackingWriter, u *user, grp *group) (scopes []limitScope, ok bool) { +// unroutable request; they apply identically here. envelope (item 7 fix, +// 2026-08-22 review) lets routes_messages.go answer a violation in +// Anthropic's own error shape instead of OpenAI's, without a second copy +// of this function's own logic. +func (g *Gateway) admitRequest(sw *statusTrackingWriter, u *user, grp *group, envelope envelopeWriter) (scopes []limitScope, ok bool) { scopes = withTotalScope(buildLimitScopes(u, grp)) if violation := g.limiter.checkAndCount(scopes); violation != nil { - writeLimitViolation(sw, violation) + writeLimitViolationEnvelope(sw, violation, envelope) return scopes, false } return scopes, true @@ -457,7 +522,8 @@ const bodyAdmissionRetryAfterSeconds = 2 // of goroutines blocked on a channel receive) — it fails immediately // instead, writing a 503 with a Retry-After header to sw and returning a // no-op release so the caller's own `defer release()` stays valid -// either way. +// either way. envelope (item 7 fix, 2026-08-22 review) picks the JSON +// shape that 503 gets written in. // // release must be deferred by the caller IMMEDIATELY after this // returns, BEFORE checking ok, and that defer must be scoped narrowly @@ -477,13 +543,13 @@ const bodyAdmissionRetryAfterSeconds = 2 // this semaphore to bound — the only body-inspection it ever does is // peekPassthroughModel's own bounded 64KiB read, an intentional // omission, not an oversight. -func (g *Gateway) acquireBodyAdmission(sw *statusTrackingWriter) (release func(), ok bool) { +func (g *Gateway) acquireBodyAdmission(sw *statusTrackingWriter, envelope envelopeWriter) (release func(), ok bool) { select { case g.bodyAdmission <- struct{}{}: return func() { <-g.bodyAdmission }, true default: sw.Header().Set("Retry-After", strconv.Itoa(bodyAdmissionRetryAfterSeconds)) - writeOAIError(sw, http.StatusServiceUnavailable, "server_error", "server is at capacity; try again shortly") + envelope(sw, http.StatusServiceUnavailable, "server_error", "server is at capacity; try again shortly") return func() {}, false } } @@ -494,14 +560,16 @@ func (g *Gateway) acquireBodyAdmission(sw *statusTrackingWriter) (release func() // successful decode, a read failure, a decode failure, or admission // exhaustion itself (security review finding 1b, round 3, 2026-08-22). // This is deliberately the ENTIRE scope of what the semaphore guards for -// the unified routes: by the time this returns, req is fully decoded and -// nothing further in runUnified re-amplifies memory the way the decode +// the unified/messages routes: by the time this returns, req is fully +// decoded and nothing further re-amplifies memory the way the decode // itself does, so model resolution, the cache lookup, and the upstream // round trip all run after the slot is already released. A read or -// decode failure has already written its 400 to sw; ok reports whether -// the caller may proceed. -func (g *Gateway) readAndDecodeUnifiedBody(sw *statusTrackingWriter, r *http.Request) (body []byte, req map[string]any, ok bool) { - release, admitted := g.acquireBodyAdmission(sw) +// decode failure has already written its 400 to sw, through envelope +// (item 7 fix, 2026-08-22 review — routes_messages.go passes +// writeAnthropicError here so this shared step answers in this route's +// own error shape); ok reports whether the caller may proceed. +func (g *Gateway) readAndDecodeUnifiedBody(sw *statusTrackingWriter, r *http.Request, envelope envelopeWriter) (body []byte, req map[string]any, ok bool) { + release, admitted := g.acquireBodyAdmission(sw, envelope) defer release() if !admitted { return nil, nil, false @@ -509,12 +577,12 @@ func (g *Gateway) readAndDecodeUnifiedBody(sw *statusTrackingWriter, r *http.Req body, err := io.ReadAll(io.LimitReader(r.Body, maxUnifiedRequestBytes)) if err != nil { - writeOAIError(sw, http.StatusBadRequest, "invalid_request_error", "cannot read request body") + envelope(sw, http.StatusBadRequest, "invalid_request_error", "cannot read request body") return nil, nil, false } if err = json.Unmarshal(body, &req); err != nil { - writeOAIError(sw, http.StatusBadRequest, "invalid_request_error", "invalid JSON body") + envelope(sw, http.StatusBadRequest, "invalid_request_error", "invalid JSON body") return nil, nil, false } return body, req, true @@ -536,58 +604,79 @@ func unifiedCostMicros(canonical, bare string, u usage, overrides map[string]*Mo return costMicros(bare, u, overrides) } -// writeModelResolveError maps a modelRegistry.resolve error to its HTTP -// envelope: errModelUnknown is a 404 (no configured provider knows this -// model), errModelDenied is a 403 (a real model the caller's group cannot -// use). A *aliasTargetError (spec §5, v0.2) is also a 404, but with its -// own message naming both the alias and its unresolved target, checked -// first via a plain type assertion — not errors.As, matching this -// package's established yaegi-safe convention for a pointer error type -// (registry.go's aliasTargetError doc comment). Any other error is a -// defensive 500 — resolve's own contract promises only these sentinels/ -// types, so reaching this branch would be a programming error, not a -// client mistake. -func writeModelResolveError(w http.ResponseWriter, err error) { +// writeModelResolveErrorEnvelope maps a modelRegistry.resolve error to +// its HTTP status/message, written through whichever envelope the caller +// supplies (item 7 fix, 2026-08-22 review): errModelUnknown is a 404 (no +// configured provider knows this model), errModelDenied is a 403 (a real +// model the caller's group cannot use). A *aliasTargetError (spec §5, +// v0.2) is also a 404, but with its own message naming both the alias and +// its unresolved target, checked first via a plain type assertion — not +// errors.As, matching this package's established yaegi-safe convention +// for a pointer error type (registry.go's aliasTargetError doc comment). +// Any other error is a defensive 500 — resolve's own contract promises +// only these sentinels/types, so reaching this branch would be a +// programming error, not a client mistake. +func writeModelResolveErrorEnvelope(w http.ResponseWriter, err error, envelope envelopeWriter) { if aerr, ok := err.(*aliasTargetError); ok { - writeOAIError(w, http.StatusNotFound, "invalid_request_error", aerr.Error()) + envelope(w, http.StatusNotFound, "invalid_request_error", aerr.Error()) return } switch { case errors.Is(err, errModelUnknown): - writeOAIError(w, http.StatusNotFound, "invalid_request_error", "unknown model") + envelope(w, http.StatusNotFound, "invalid_request_error", "unknown model") case errors.Is(err, errModelDenied): - writeOAIError(w, http.StatusForbidden, "invalid_request_error", "model access denied") + envelope(w, http.StatusForbidden, "invalid_request_error", "model access denied") default: - writeOAIError(w, http.StatusInternalServerError, "server_error", "internal error") + envelope(w, http.StatusInternalServerError, "server_error", "internal error") } } -// writeLimitViolation maps a limiter violation to its HTTP envelope -// (ruling b): a storeDown violation — the configured limit store was -// unreachable and failOpen is false — is a 503 server_error, since it is -// not an actual limit breach; any other violation is a 429 rate_limit_error -// carrying v's own message, plus a Retry-After header when the violation -// names a meaningful retry window. -func writeLimitViolation(w http.ResponseWriter, v *limitViolation) { +// writeModelResolveError is writeModelResolveErrorEnvelope pinned to the +// OpenAI envelope shape — the thin wrapper every existing OpenAI-shaped +// call site (this file's runMeteredCall, via writeOAIError; +// routes_media.go's resolveMediaModel) keeps calling unchanged. +func writeModelResolveError(w http.ResponseWriter, err error) { + writeModelResolveErrorEnvelope(w, err, writeOAIError) +} + +// writeLimitViolationEnvelope maps a limiter violation to its HTTP +// status/message, written through whichever envelope the caller supplies +// (item 7 fix, 2026-08-22 review): a storeDown violation — the configured +// limit store was unreachable and failOpen is false — is a 503 +// server_error, since it is not an actual limit breach; any other +// violation is a 429 rate_limit_error carrying v's own message, plus a +// Retry-After header when the violation names a meaningful retry window. +func writeLimitViolationEnvelope(w http.ResponseWriter, v *limitViolation, envelope envelopeWriter) { if v.storeDown { - writeOAIError(w, http.StatusServiceUnavailable, "server_error", v.message) + envelope(w, http.StatusServiceUnavailable, "server_error", v.message) return } if v.retryAfter > 0 { w.Header().Set("Retry-After", strconv.Itoa(v.retryAfter)) } - writeOAIError(w, http.StatusTooManyRequests, "rate_limit_error", v.message) + envelope(w, http.StatusTooManyRequests, "rate_limit_error", v.message) } -// handleAdapterError translates an adapter's returned error into a -// response, or into nothing at all, per ruling c. sw.wroteHeader is -// checked first, ahead of every other case: an error surfacing after the -// adapter already started writing a response (a mid-stream connection -// drop) must never get a second, conflicting envelope appended, whatever -// kind of error it is. -func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, providerName string) { +// writeLimitViolation is writeLimitViolationEnvelope pinned to the +// OpenAI envelope shape — the thin wrapper every existing direct call +// site (mcp_a2a.go, mcp_federation.go, routes_passthrough.go, all of +// which call checkAndCount themselves rather than through admitRequest) +// keeps calling unchanged. +func writeLimitViolation(w http.ResponseWriter, v *limitViolation) { + writeLimitViolationEnvelope(w, v, writeOAIError) +} + +// handleAdapterErrorEnvelope is the shared adapter-error classification +// every metered route goes through (item 7 fix, 2026-08-22 review): +// sw.wroteHeader is checked first, ahead of every other case, since an +// error surfacing after the adapter already started writing a response +// (a mid-stream connection drop) must never get a second, conflicting +// envelope appended, whatever kind of error it is. envelope/writeUpstream +// pick the client-facing shape; logPrefix names the route in every log +// line. +func (g *Gateway) handleAdapterErrorEnvelope(sw *statusTrackingWriter, err error, providerName, logPrefix string, envelope envelopeWriter, writeUpstream providerUpstreamErrorWriter) { // A cacheable request's miss path pre-sets X-Llmgw-Cache: miss before - // call() runs (runUnified), so headers precede a successful body — + // call() runs (runMeteredCall), so headers precede a successful body — // but an adapter error means nothing was actually served from, or // stored to, the cache. Del is unconditional and harmless when the // header was never set (a no-op on an absent key), so this needs no @@ -597,7 +686,7 @@ func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, provid sw.Header().Del("X-Llmgw-Cache") if sw.wroteHeader { - g.errorf("unified route: adapter error after response started (provider %q): %v", providerName, err) + g.errorf("%s: adapter error after response started (provider %q): %v", logPrefix, providerName, err) return } @@ -611,7 +700,7 @@ func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, provid // Verified empirically under real Traefik (Task 15's integration // suite); tools/yaegi-check never exercises this call path. if perr, ok := err.(*providerHTTPError); ok { - writeProviderUpstreamError(sw, providerName, perr) + writeUpstream(sw, providerName, perr) return } @@ -620,17 +709,58 @@ func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, provid if terr.notSupported { status = http.StatusNotImplemented } - writeOAIError(sw, status, "invalid_request_error", terr.msg) + envelope(sw, status, "invalid_request_error", terr.msg) + return + } + + // *responseTranslationError (routes_messages.go, item 1/12 fix, + // 2026-08-22 review) is the /v1/messages route's own: a translation + // failure AFTER a successful, already-billed upstream call (its usage + // already travels back through call's own return value regardless of + // this branch — see runMeteredCall's own comment on that). This + // branch only ever fires for routes_messages.go's own call closures; + // runUnified's adapterCall implementations never produce this type, + // so it is always a no-op miss for that route. Kept here, in the one + // shared classification, rather than a second copy of this whole + // cascade in routes_messages.go, so the classification logic itself + // never exists twice — see handleAdapterErrorEnvelope's own doc + // comment (item 7). + if terr, ok := err.(*responseTranslationError); ok { + g.errorf("%s: response translation failed after a billed upstream call (provider %q): %v", logPrefix, providerName, terr.err) + envelope(sw, http.StatusBadGateway, "server_error", "failed to translate provider response") return } if errors.Is(err, context.Canceled) { - g.logf("unified route: client canceled request to provider %q: %v", providerName, err) + g.logf("%s: client canceled request to provider %q: %v", logPrefix, providerName, err) return } - g.errorf("unified route: upstream connection error (provider %q): %v", providerName, err) - writeOAIError(sw, http.StatusBadGateway, "server_error", "upstream connection error") + g.errorf("%s: upstream connection error (provider %q): %v", logPrefix, providerName, err) + envelope(sw, http.StatusBadGateway, "server_error", "upstream connection error") +} + +// handleAdapterError is handleAdapterErrorEnvelope pinned to the OpenAI +// envelope shape — the thin wrapper every existing OpenAI-shaped call +// site (routes_media.go's three media handlers) keeps calling unchanged. +// runMeteredCall (this file) calls handleAdapterErrorEnvelope directly, +// with whichever envelope its own caller supplied. +func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, providerName string) { + g.handleAdapterErrorEnvelope(sw, err, providerName, "unified route", writeOAIError, writeProviderUpstreamError) +} + +// decodeUpstreamErrorBody decodes perr.body as JSON when it parses as +// one, or returns it as a raw string otherwise — the shared read +// writeProviderUpstreamError and routes_messages.go's +// writeAnthropicProviderUpstreamError both embed under "upstream" in +// their own envelope shape (item 7/8 fix, 2026-08-22 review). +func decodeUpstreamErrorBody(perr *providerHTTPError) any { + var upstream any = string(perr.body) + var parsed any + if json.Unmarshal(perr.body, &parsed) == nil { + upstream = parsed + } + return upstream } // writeProviderUpstreamError passes perr's status through to the client, @@ -638,16 +768,10 @@ func (g *Gateway) handleAdapterError(sw *statusTrackingWriter, err error, provid // verbatim (unified-route callers get this wrapped form; the openai-type // adapter's own passthrough forwards a non-2xx body unwrapped, since that // path never reaches this function). perr.body is embedded under -// error.upstream, decoded to a JSON value when it parses as one and left -// as a raw string otherwise, so a caller sees exactly what the provider -// said either way. +// error.upstream (decodeUpstreamErrorBody, above), decoded to a JSON +// value when it parses as one and left as a raw string otherwise, so a +// caller sees exactly what the provider said either way. func writeProviderUpstreamError(w http.ResponseWriter, providerName string, perr *providerHTTPError) { - var upstream any = string(perr.body) - var parsed any - if json.Unmarshal(perr.body, &parsed) == nil { - upstream = parsed - } - w.Header().Set("Content-Type", "application/json") w.WriteHeader(perr.status) _ = json.NewEncoder(w).Encode(map[string]any{ // headers already committed; nothing useful to do on encode failure @@ -655,7 +779,7 @@ func writeProviderUpstreamError(w http.ResponseWriter, providerName string, perr "message": providerName + " upstream error", "type": "upstream_error", "code": strconv.Itoa(perr.status), - "upstream": upstream, + "upstream": decodeUpstreamErrorBody(perr), }, }) } @@ -663,9 +787,10 @@ func writeProviderUpstreamError(w http.ResponseWriter, providerName string, perr // cacheCaptureWriter tees a cacheable request's response into an // in-memory buffer, capped at maxBodyBytes regardless of how large the // real response turns out to be, while writing everything through to the -// wrapped *statusTrackingWriter unchanged — the mechanism runUnified uses -// to fill the response cache (cache.go) on a miss without any adapter, or -// forwardJSON/forwardStream inside one, needing to know caching exists. +// wrapped *statusTrackingWriter unchanged — the mechanism runMeteredCall +// uses to fill the response cache (cache.go) on a miss without any +// adapter, or forwardJSON/forwardStream inside one, needing to know +// caching exists. // // The cap matters even though the adapter response it captures is itself // already bounded by maxResponseBytes (32MiB): cache.maxBodyBytes @@ -681,9 +806,10 @@ func writeProviderUpstreamError(w http.ResponseWriter, providerName string, perr // overriding here. This is what "preserve statusTrackingWriter semantics" // means in practice: Flush passes through for free, and wroteHeader stays // the single shared bookkeeping field statusTrackingWriter already -// maintains (handleAdapterError inspects it via the original sw pointer, -// not through this wrapper, so a cacheCaptureWriter's WriteHeader/Write -// must delegate to the embedded pointer, never shadow its state). +// maintains (handleAdapterErrorEnvelope inspects it via the original sw +// pointer, not through this wrapper, so a cacheCaptureWriter's +// WriteHeader/Write must delegate to the embedded pointer, never shadow +// its state). type cacheCaptureWriter struct { *statusTrackingWriter contentType string @@ -692,7 +818,7 @@ type cacheCaptureWriter struct { status int // oversize is true once buf has reached maxBodyBytes: Write stops // appending to buf from that point on (the real write to the client - // is never affected), and runUnified skips calling store() entirely + // is never affected), and runMeteredCall skips calling store() entirely // rather than handing it a silently truncated body. oversize bool } diff --git a/tools/yaegi-check/main.go b/tools/yaegi-check/main.go index ab508c1..2657937 100644 --- a/tools/yaegi-check/main.go +++ b/tools/yaegi-check/main.go @@ -282,6 +282,31 @@ func run() error { _, _ = w.Write([]byte(`"}]}}`)) })) defer mcpProbeUpstream.Close() + // anthropicProbeUpstream answers an Anthropic Messages API call with a + // real Anthropic-shaped body, so the /v1/messages passthrough branch + // (callAnthropicMessagesPassthrough) runs interpreted. A body + // containing "make-it-fail" gets a 429, driving + // handleAdapterErrorEnvelope's *providerHTTPError branch interpreted + // too — the exact construct class that has broken under Yaegi before + // (see handleAdapterError's own doc comment, routes_unified.go). + anthropicProbeUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + if bytes.Contains(raw, []byte("make-it-fail")) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(`{"type":"error","error":{"type":"rate_limit_error","message":"slow down"}}`)) + return + } + if r.Header.Get("anthropic-version") == "" || r.Header.Get("x-api-key") == "" { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"type":"error","error":{"type":"invalid_request_error","message":"missing anthropic headers"}}`)) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"id":"msg_probe","type":"message","role":"assistant","model":"claude-upstream-real","content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn","usage":{"input_tokens":7,"output_tokens":3}}`)) + })) + defer anthropicProbeUpstream.Close() // builtinLookupContextTokens (review fix, SHOULD-6): read directly // out of the generated pricing_data_gen.go rather than hardcoding a @@ -294,7 +319,12 @@ func run() error { } attemptAccountingOverride := `{"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},"` + - slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]}},` + + slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]},` + + // "anthropic" backs the /v1/messages passthrough probes + // (exerciseMessagesRoute, below): a real anthropic-type provider, + // interpreted end to end through + // callAnthropicMessagesPassthrough. + `"anthropic":{"type":"anthropic","baseUrl":"` + anthropicProbeUpstream.URL + `","apiKey":"sk-anth","models":["claude-test"]}},` + // #nosec G101 -- test fixture literal, not a real credential // modelMeta (feature v0.23): a config-override entry for // testDataWantModel, so exerciseHandler's GET /v1/models // assertion below proves resolveModelMeta's config-override @@ -532,6 +562,10 @@ func exerciseHandler(handler http.Handler, builtinContextTokens int) error { return err } + if err := exerciseMessagesRoute(handler); err != nil { + return err + } + unauthedReq := httptest.NewRequest(http.MethodGet, "/v1/models", nil) unauthedRec := httptest.NewRecorder() handler.ServeHTTP(unauthedRec, unauthedReq) @@ -601,6 +635,178 @@ func exerciseStampedVersion(handler http.Handler) error { return nil } +// postMessages drives one POST /v1/messages through handler with the given +// auth header name/value and JSON body, returning the recorder. +func postMessages(handler http.Handler, hdrName, hdrValue, body string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(hdrName, hdrValue) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + return rec +} + +// exerciseMessagesRoute drives the inbound Anthropic Messages API route +// under the REAL interpreter: both auth header forms, the streaming +// rejection, a bad key's error envelope, an unknown model's error +// envelope, the openai-type TRANSLATED path (openAIRequestFromAnthropic +// -> chatCompletion -> anthropicResponseFromOpenAI), the anthropic-type +// PASSTHROUGH path including its own upstream-error branch, and a +// tool_use round trip through the translated path. This exists because +// go test never runs under Yaegi — a construct that compiles and passes +// `go test` can still panic or silently misbehave interpreted (this +// package's own doc comment has the general case; handleAdapterError's +// doc comment, routes_unified.go, has this route's own specific history +// with plain type assertions over errors.As). +func exerciseMessagesRoute(handler http.Handler) error { + // 1. x-api-key auth + translated path (openai-type provider). + rec := postMessages(handler, "x-api-key", testDataUserAPIKey, + `{"model":"`+testDataWantModel+`","max_tokens":64,"system":"be terse","messages":[{"role":"user","content":[{"type":"text","text":"ping"}]}]}`) + if rec.Code != http.StatusOK { + return fmt.Errorf("POST /v1/messages (x-api-key, translated): status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var msg map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &msg); err != nil { + return fmt.Errorf("POST /v1/messages: decode body: %w (body=%s)", err, rec.Body.String()) + } + if msg["type"] != "message" || msg["role"] != "assistant" { + return fmt.Errorf("POST /v1/messages translated: type/role = %v/%v, want message/assistant: %s", msg["type"], msg["role"], rec.Body.String()) + } + if msg["model"] != testDataWantModel { + return fmt.Errorf("POST /v1/messages translated: model = %v, want %q (the client's requested alias echoed back, item 4 fix): %s", msg["model"], testDataWantModel, rec.Body.String()) + } + if msg["stop_reason"] != "end_turn" { + return fmt.Errorf("POST /v1/messages translated: stop_reason = %v, want end_turn: %s", msg["stop_reason"], rec.Body.String()) + } + content, _ := msg["content"].([]any) + if len(content) != 1 { + return fmt.Errorf("POST /v1/messages translated: content len = %d, want 1: %s", len(content), rec.Body.String()) + } + block, _ := content[0].(map[string]any) + if block["type"] != "text" || block["text"] != "hi" { + return fmt.Errorf("POST /v1/messages translated: content[0] = %v, want text/hi: %s", content[0], rec.Body.String()) + } + usg, _ := msg["usage"].(map[string]any) + if usg == nil || usg["input_tokens"] != float64(1) || usg["output_tokens"] != float64(1) { + return fmt.Errorf("POST /v1/messages translated: usage = %v, want input=1 output=1: %s", msg["usage"], rec.Body.String()) + } + + // 2. Authorization: Bearer must authenticate the same route. + bearerRec := postMessages(handler, "Authorization", "Bearer "+testDataUserAPIKey, + `{"model":"`+testDataWantModel+`","max_tokens":64,"messages":[{"role":"user","content":"ping"}]}`) + if bearerRec.Code != http.StatusOK { + return fmt.Errorf("POST /v1/messages (Authorization: Bearer): status = %d, want 200, body=%s", bearerRec.Code, bearerRec.Body.String()) + } + + // 3. A bad key gets a 401 in the ANTHROPIC error shape. + badRec := postMessages(handler, "x-api-key", "not-a-real-key", + `{"model":"`+testDataWantModel+`","max_tokens":64,"messages":[{"role":"user","content":"ping"}]}`) + if badRec.Code != http.StatusUnauthorized { + return fmt.Errorf("POST /v1/messages with a bad key: status = %d, want 401, body=%s", badRec.Code, badRec.Body.String()) + } + if err := assertAnthropicErrorShape("bad key 401", badRec.Body.Bytes()); err != nil { + return err + } + + // 4. "stream": true is rejected explicitly, in the Anthropic shape, + // as a 400 (item 2 fix, 2026-08-22 review) — NOT 501: a 501 makes + // anthropic-sdk-python (which retries any status >= 500) turn every + // streaming call into a three-request retry storm against this + // route's own rate limit. 400 maps to the SDK's non-retryable + // BadRequestError. + streamRec := postMessages(handler, "x-api-key", testDataUserAPIKey, + `{"model":"`+testDataWantModel+`","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"ping"}]}`) + if streamRec.Code != http.StatusBadRequest { + return fmt.Errorf("POST /v1/messages with stream=true: status = %d, want 400, body=%s", streamRec.Code, streamRec.Body.String()) + } + if err := assertAnthropicErrorShape("stream rejection", streamRec.Body.Bytes()); err != nil { + return err + } + + // 5. An unknown model's 404 must also be Anthropic-shaped. + unknownRec := postMessages(handler, "x-api-key", testDataUserAPIKey, + `{"model":"no-such-model-anywhere","max_tokens":64,"messages":[{"role":"user","content":"ping"}]}`) + if unknownRec.Code != http.StatusNotFound { + return fmt.Errorf("POST /v1/messages with an unknown model: status = %d, want 404, body=%s", unknownRec.Code, unknownRec.Body.String()) + } + if err := assertAnthropicErrorShape("unknown model 404", unknownRec.Body.Bytes()); err != nil { + return err + } + + // 6. The anthropic-type PASSTHROUGH branch, interpreted: every field + // survives except "model", rewritten to the client's own requested + // alias (item 4 fix) — a routing rewrite, not message-body + // translation. + passRec := postMessages(handler, "x-api-key", testDataUserAPIKey, + `{"model":"claude-test","max_tokens":64,"system":"be terse","messages":[{"role":"user","content":"ping"}]}`) + if passRec.Code != http.StatusOK { + return fmt.Errorf("POST /v1/messages (anthropic passthrough): status = %d, want 200, body=%s", passRec.Code, passRec.Body.String()) + } + var pass map[string]any + if err := json.Unmarshal(passRec.Body.Bytes(), &pass); err != nil { + return fmt.Errorf("POST /v1/messages passthrough: decode body: %w (body=%s)", err, passRec.Body.String()) + } + if pass["id"] != "msg_probe" { + return fmt.Errorf("POST /v1/messages passthrough: id = %v, want msg_probe (every field but \"model\" must survive the passthrough unchanged): %s", pass["id"], passRec.Body.String()) + } + if pass["model"] != "claude-test" { + return fmt.Errorf("POST /v1/messages passthrough: model = %v, want \"claude-test\" (the client's requested alias echoed back, item 4 fix — not anthropicProbeUpstream's own \"claude-upstream-real\"): %s", pass["model"], passRec.Body.String()) + } + + // 6b. handleAdapterErrorEnvelope's *providerHTTPError type assertion, + // interpreted — the exact construct class that has broken under Yaegi + // before (see handleAdapterError's own doc comment, routes_unified.go). + upErrRec := postMessages(handler, "x-api-key", testDataUserAPIKey, + `{"model":"claude-test","max_tokens":64,"messages":[{"role":"user","content":"make-it-fail"}]}`) + if upErrRec.Code != http.StatusTooManyRequests { + return fmt.Errorf("POST /v1/messages (upstream 429): status = %d, want 429, body=%s", upErrRec.Code, upErrRec.Body.String()) + } + if err := assertAnthropicErrorShape("upstream 429", upErrRec.Body.Bytes()); err != nil { + return err + } + if upErrRec.Header().Get("X-Llmgw-Cache") != "" { + return fmt.Errorf("POST /v1/messages (upstream 429) carries a stale X-Llmgw-Cache header: %q", upErrRec.Header().Get("X-Llmgw-Cache")) + } + + // 7. A tool_use round trip through the TRANSLATED path, to drive + // openAIToolCallFromAnthropic/openAIToolMessageFromAnthropic and + // anthropicToolInputFromOpenAI interpreted. + toolRec := postMessages(handler, "x-api-key", testDataUserAPIKey, + `{"model":"`+testDataWantModel+`","max_tokens":64,`+ + `"tools":[{"name":"get_weather","description":"w","input_schema":{"type":"object","properties":{"city":{"type":"string"}}}}],`+ + `"tool_choice":{"type":"auto"},`+ + `"messages":[{"role":"user","content":"weather?"},`+ + `{"role":"assistant","content":[{"type":"tool_use","id":"toolu_1","name":"get_weather","input":{"city":"London"}}]},`+ + `{"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_1","content":"sunny"}]}]}`) + if toolRec.Code != http.StatusOK { + return fmt.Errorf("POST /v1/messages (tool_use round trip): status = %d, want 200, body=%s", toolRec.Code, toolRec.Body.String()) + } + return nil +} + +// assertAnthropicErrorShape checks b is {"type":"error","error":{"type":..., +// "message":...}} — what an Anthropic SDK client parses. +func assertAnthropicErrorShape(label string, b []byte) error { + var env map[string]any + if err := json.Unmarshal(b, &env); err != nil { + return fmt.Errorf("/v1/messages %s: body is not JSON: %w (body=%s)", label, err, string(b)) + } + if env["type"] != "error" { + return fmt.Errorf("/v1/messages %s: top-level \"type\" = %v, want \"error\" (Anthropic envelope): %s", label, env["type"], string(b)) + } + inner, _ := env["error"].(map[string]any) + if inner == nil { + return fmt.Errorf("/v1/messages %s: no \"error\" object: %s", label, string(b)) + } + if _, ok := inner["type"].(string); !ok { + return fmt.Errorf("/v1/messages %s: error.type missing: %s", label, string(b)) + } + if _, ok := inner["message"].(string); !ok { + return fmt.Errorf("/v1/messages %s: error.message missing: %s", label, string(b)) + } + return nil +} + // exerciseFederatedTooLarge drives POST /mcp (federation's tools/call // path) against mcpProbeUpstream, whose response deliberately exceeds // mcpBackendCallResponseMaxBytes, and asserts the interpreted plugin diff --git a/translate_anthropic.go b/translate_anthropic.go index b07410d..8d746cf 100644 --- a/translate_anthropic.go +++ b/translate_anthropic.go @@ -487,9 +487,34 @@ type anthropicContentBlock struct { // anthropicUsagePayload is the "usage" object on both a non-streaming // Anthropic response and a streaming message_start/message_delta event. +// CacheCreationInputTokens/CacheReadInputTokens (item 6 fix, 2026-08-22 +// review) are Anthropic's prompt-cache counters: tokens written to the +// cache and tokens read back from it, reported separately from +// InputTokens (fresh, uncached input). Verified repro before this fix: +// 180000 cache-creation + 20000 cache-read tokens billed as tokin=4 — the +// two cache fields were decoded into nothing and silently dropped. +// Caching is near-universal for Claude Code and any agentic client using +// Anthropic's prompt cache, so ignoring them under-counted budgets by +// orders of magnitude on the primary use case. type anthropicUsagePayload struct { - InputTokens int64 `json:"input_tokens"` - OutputTokens int64 `json:"output_tokens"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens"` +} + +// totalInputTokens returns the full input-side token count Anthropic +// actually billed for one request: fresh InputTokens plus both cache-tier +// counters. This gateway's budgets/rate-limits (limits.go's usage.prompt) +// track one combined "prompt" token count per request, not per-tier +// pricing, so folding all three into one number here — rather than +// pricing cache-write/cache-read tiers differently — fixes the under- +// billing item 6 reported (2026-08-22 review) without a materially larger +// change to pricing.go's cost model, which this fix deliberately leaves +// alone: token-budget accuracy was the reported defect, not $ cost +// accuracy for cache-tier pricing. +func (p anthropicUsagePayload) totalInputTokens() int64 { + return p.InputTokens + p.CacheCreationInputTokens + p.CacheReadInputTokens } // anthropicResponseBody is the subset of a non-streaming Anthropic @@ -559,7 +584,7 @@ func openAIResponseFromAnthropic(body []byte, model string, created int64) (map[ message["tool_calls"] = toolCalls } - u := usage{prompt: resp.Usage.InputTokens, completion: resp.Usage.OutputTokens} + u := usage{prompt: resp.Usage.totalInputTokens(), completion: resp.Usage.OutputTokens} out := map[string]any{ "id": chatCompletionIDPrefix + resp.ID, @@ -744,7 +769,7 @@ func (st *anthropicStreamState) translate(ev sseEvent) ([][]byte, error) { return nil, fmt.Errorf("%w: decode anthropic message_start event: %w", errUpstream, err) } st.id = p.Message.ID - st.u.prompt = p.Message.Usage.InputTokens + st.u.prompt = p.Message.Usage.totalInputTokens() return [][]byte{st.chunk(map[string]any{"role": "assistant"}, nil)}, nil case "content_block_start": @@ -910,25 +935,64 @@ func openAIToolCallFromAnthropic(bm map[string]any) (map[string]any, error) { } // openAIToolMessageFromAnthropic maps one Anthropic {"type":"tool_result", -// "tool_use_id","content"} content block to an OpenAI role:"tool" message -// ({"role":"tool","tool_call_id","content"}), the mirror of -// anthropicToolResultBlock above. +// "tool_use_id","content","is_error"} content block to an OpenAI +// role:"tool" message ({"role":"tool","tool_call_id","content"}), the +// mirror of anthropicToolResultBlock above. is_error (item 11 fix, +// 2026-08-22 review) has no native OpenAI equivalent — a tool message +// carries no error flag — so it is preserved via prefixErrorMarker +// instead of being silently dropped: the previous version of this +// function read only tool_use_id/content and discarded is_error +// entirely, losing the failure signal for any OpenAI-backed agentic loop +// that branches on whether a tool call succeeded. func openAIToolMessageFromAnthropic(bm map[string]any) map[string]any { toolUseID, _ := bm["tool_use_id"].(string) - return map[string]any{"role": "tool", "tool_call_id": toolUseID, "content": bm["content"]} + content := bm["content"] + if isErr, _ := bm["is_error"].(bool); isErr { + content = prefixErrorMarker(content) + } + return map[string]any{"role": "tool", "tool_call_id": toolUseID, "content": content} +} + +// prefixErrorMarker preserves Anthropic's tool_result.is_error signal +// across the translation to OpenAI's tool message shape (item 11 fix, +// 2026-08-22 review): a string content gets an "Error: " prefix so the +// model still sees the failure signal in the one shape OpenAI's tool +// message actually carries. Any other content shape (a block array, or +// absent) is left as-is rather than guessing how to annotate it — +// OpenAI-compatible upstreams generally infer failure from content +// itself in that case, same as before this fix; only the common +// string-content case previously lost the signal entirely. +func prefixErrorMarker(content any) any { + if s, ok := content.(string); ok { + return "Error: " + s + } + return content } // openAIMessagesFromAnthropic maps one Anthropic message ({"role", // "content"}, content either a plain string or a content-block array) to -// one or more OpenAI messages: a plain string passes through unchanged; a -// block array splits into at most one {role, content, tool_calls?} -// message carrying every text/image/tool_use block, followed by one -// role:"tool" message per tool_result block (Anthropic allows a single -// user turn to carry several tool_result blocks answering a prior -// assistant turn's several tool_use calls — OpenAI's shape represents -// each as its own separate message, the mirror of -// anthropicRequestFromOpenAI's pendingToolResults accumulation above, -// which does the same collapse in the opposite direction). +// one or more OpenAI messages, preserving the ORIGINAL block order (item +// 3 fix, 2026-08-22 review): a plain string passes through unchanged. For +// a block array, text/image/tool_use blocks accumulate into one pending +// {role, content, tool_calls?} message; each tool_result block flushes +// that pending message (if it has anything to flush) and then emits its +// own role:"tool" message, in the SAME relative position the tool_result +// held in the original array. Any trailing accumulated content flushes at +// the end. +// +// This ordering matters beyond cosmetics: OpenAI requires a role:"tool" +// message to directly follow the assistant message carrying the +// tool_calls it answers, with nothing in between. The PREVIOUS version of +// this function always emitted every tool_result AFTER the combined +// content/tool_calls message regardless of where it actually appeared in +// the original block array — correct for the common shape (a user turn +// that is pure tool_results, following an assistant's pure tool_use +// turn), but wrong for a legal Anthropic user turn shaped [tool_result, +// text]: the old code emitted [user(text), tool(...)], inserting a +// non-tool message between an assistant's tool_calls and its answer and +// getting a 400 from OpenAI ("messages with role 'tool' must be a +// response to a preceding message with 'tool_calls'"). Preserving +// original order fixes both shapes without special-casing either. func openAIMessagesFromAnthropic(role string, content any) ([]any, error) { if s, ok := content.(string); ok { return []any{map[string]any{"role": role, "content": s}}, nil @@ -942,9 +1006,27 @@ func openAIMessagesFromAnthropic(role string, content any) ([]any, error) { return []any{map[string]any{"role": role, "content": ""}}, nil } - var contentParts []any - var toolCalls []any - var toolResults []any + var out []any + var pendingContent []any + var pendingToolCalls []any + flushPending := func() { + if len(pendingContent) == 0 && len(pendingToolCalls) == 0 { + return + } + m := map[string]any{"role": role} + if len(pendingContent) > 0 { + m["content"] = pendingContent + } else { + m["content"] = nil + } + if len(pendingToolCalls) > 0 { + m["tool_calls"] = pendingToolCalls + } + out = append(out, m) + pendingContent = nil + pendingToolCalls = nil + } + for _, b := range blocksRaw { bm, ok := b.(map[string]any) if !ok { @@ -953,38 +1035,25 @@ func openAIMessagesFromAnthropic(role string, content any) ([]any, error) { switch bt, _ := bm["type"].(string); bt { case "text": text, _ := bm["text"].(string) - contentParts = append(contentParts, map[string]any{"type": "text", "text": text}) + pendingContent = append(pendingContent, map[string]any{"type": "text", "text": text}) case "image": part, err := openAIContentPartFromAnthropicImage(bm) if err != nil { return nil, err } - contentParts = append(contentParts, part) + pendingContent = append(pendingContent, part) case "tool_use": tc, err := openAIToolCallFromAnthropic(bm) if err != nil { return nil, err } - toolCalls = append(toolCalls, tc) + pendingToolCalls = append(pendingToolCalls, tc) case "tool_result": - toolResults = append(toolResults, openAIToolMessageFromAnthropic(bm)) + flushPending() + out = append(out, openAIToolMessageFromAnthropic(bm)) } } - - var out []any - if len(contentParts) > 0 || len(toolCalls) > 0 { - m := map[string]any{"role": role} - if len(contentParts) > 0 { - m["content"] = contentParts - } else { - m["content"] = nil - } - if len(toolCalls) > 0 { - m["tool_calls"] = toolCalls - } - out = append(out, m) - } - out = append(out, toolResults...) + flushPending() return out, nil } @@ -1032,6 +1101,32 @@ func openAIToolChoiceFromAnthropic(tc any) any { } } +// openAIReasoningModelPrefixes are OpenAI (and OpenAI-compatible) model +// id prefixes that reject the "max_tokens" request field with a 400 and +// require "max_completion_tokens" instead (item 10 fix, 2026-08-22 +// review): the o-series reasoning models and the gpt-5 family. Matched +// against the resolved upstream model id, case-sensitively, the same +// case OpenAI's own ids use. +var openAIReasoningModelPrefixes = []string{"o1", "o3", "o4", "gpt-5"} + +// isOpenAIReasoningModel reports whether model is one of the model +// families openAIRequestFromAnthropic (below) must emit +// "max_completion_tokens" for instead of "max_tokens" — see +// openAIReasoningModelPrefixes' own doc comment. model may carry a +// "provider/" prefix; this only needs a prefix match on the bare id +// portion after it. +func isOpenAIReasoningModel(model string) bool { + if _, rest, ok := strings.Cut(model, "/"); ok { + model = rest + } + for _, prefix := range openAIReasoningModelPrefixes { + if strings.HasPrefix(model, prefix) { + return true + } + } + return false +} + // openAIRequestFromAnthropic maps an Anthropic Messages API request body // (already decoded into map[string]any by the messages route's shared // admission/decode path) to an OpenAI chat-completion request body, the @@ -1039,17 +1134,27 @@ func openAIToolChoiceFromAnthropic(tc any) any { // *translateError — never wrapped, matching this file's established // convention — for a content block this translator cannot map (an // unsupported image source type). -func openAIRequestFromAnthropic(req map[string]any) (map[string]any, error) { +// +// The second return value lists request fields this translator +// recognizes but has no OpenAI equivalent for and therefore dropped +// (item 11 fix, 2026-08-22 review) — currently only "thinking" +// (Anthropic's extended-thinking/reasoning-budget config). The caller +// (routes_messages.go's callTranslatedMessages) logs a warning per +// dropped field rather than this function logging directly: every other +// function in this file is a pure, Gateway-free translator, callable and +// testable without a *Gateway instance, and this keeps it that way. +func openAIRequestFromAnthropic(req map[string]any) (map[string]any, []string, error) { out := map[string]any{} - if model, ok := req["model"].(string); ok { - out["model"] = model + modelStr, _ := req["model"].(string) + if modelStr != "" { + out["model"] = modelStr } msgs := make([]any, 0, 1) if sys, ok := req["system"]; ok { text, err := anthropicSystemToText(sys) if err != nil { - return nil, err + return nil, nil, err } if text != "" { msgs = append(msgs, map[string]any{"role": "system", "content": text}) @@ -1065,7 +1170,7 @@ func openAIRequestFromAnthropic(req map[string]any) (map[string]any, error) { role, _ := msg["role"].(string) converted, err := openAIMessagesFromAnthropic(role, msg["content"]) if err != nil { - return nil, err + return nil, nil, err } msgs = append(msgs, converted...) } @@ -1073,7 +1178,15 @@ func openAIRequestFromAnthropic(req map[string]any) (map[string]any, error) { if v, ok := req["max_tokens"]; ok { if n, ok2 := toInt64(v); ok2 { - out["max_tokens"] = n + // Reasoning models (o1/o3/o4/gpt-5 family) reject "max_tokens" + // with a 400 and require "max_completion_tokens" instead + // (item 10 fix, 2026-08-22 review) — isOpenAIReasoningModel's + // own doc comment has the model-family list. + if isOpenAIReasoningModel(modelStr) { + out["max_completion_tokens"] = n + } else { + out["max_tokens"] = n + } } } if v, ok := req["temperature"]; ok { @@ -1104,7 +1217,12 @@ func openAIRequestFromAnthropic(req map[string]any) (map[string]any, error) { } } - return out, nil + var dropped []string + if _, ok := req["thinking"]; ok { + dropped = append(dropped, "thinking") + } + + return out, dropped, nil } // openAIChatResponseBody is the subset of a non-streaming OpenAI diff --git a/translate_anthropic_test.go b/translate_anthropic_test.go index 92bd18e..475f3d9 100644 --- a/translate_anthropic_test.go +++ b/translate_anthropic_test.go @@ -532,7 +532,7 @@ func TestOpenAIRequestFromAnthropic_ToolResultBecomesToolMessage(t *testing.T) { }, } - out, err := openAIRequestFromAnthropic(req) + out, _, err := openAIRequestFromAnthropic(req) if err != nil { t.Fatalf("openAIRequestFromAnthropic: %v", err) } @@ -567,7 +567,7 @@ func TestOpenAIRequestFromAnthropic_SystemStringFoldedIntoMessages(t *testing.T) "messages": []any{map[string]any{"role": "user", "content": "hi"}}, } - out, err := openAIRequestFromAnthropic(req) + out, _, err := openAIRequestFromAnthropic(req) if err != nil { t.Fatalf("openAIRequestFromAnthropic: %v", err) } @@ -649,3 +649,197 @@ func TestOpenAIContentPartFromAnthropicImage_RejectsNonBase64Source(t *testing.T t.Errorf("err = %T, want *translateError", err) } } + +// TestOpenAIMessagesFromAnthropic_PreservesToolResultThenTextOrder is the +// regression for item 3 (IMPORTANT, 2026-08-22 review): a legal Anthropic +// user turn shaped [tool_result, text] must translate preserving that +// order — tool message first, then the content message — since OpenAI +// requires a role:"tool" message to directly follow the assistant +// message carrying the tool_calls it answers, with nothing in between. +// The previous version of this function always emitted every tool_result +// AFTER the combined content/tool_calls message regardless of its +// original position, which turned this exact shape into +// [assistant(tool_calls), user(text), tool(...)] — an OpenAI 400, since +// user(text) sits between the tool_calls and its answer. +func TestOpenAIMessagesFromAnthropic_PreservesToolResultThenTextOrder(t *testing.T) { + content := []any{ + map[string]any{"type": "tool_result", "tool_use_id": "call_1", "content": "sunny"}, + map[string]any{"type": "text", "text": "thanks"}, + } + + out, err := openAIMessagesFromAnthropic("user", content) + if err != nil { + t.Fatalf("openAIMessagesFromAnthropic: %v", err) + } + if len(out) != 2 { + t.Fatalf("messages = %v, want 2 (tool result, then text)", out) + } + + first, _ := out[0].(map[string]any) + if first["role"] != "tool" || first["tool_call_id"] != "call_1" { + t.Errorf("messages[0] = %v, want the tool_result message FIRST, matching its original position", first) + } + + second, _ := out[1].(map[string]any) + if second["role"] != "user" { + t.Errorf("messages[1] = %v, want the text-content message SECOND", second) + } + parts, _ := second["content"].([]any) + if len(parts) != 1 { + t.Fatalf("messages[1].content = %v, want one text part", second["content"]) + } +} + +// TestOpenAIMessagesFromAnthropic_InterspersedToolResults proves the fix +// generalizes beyond one tool_result: [tool_result A, text, tool_result +// B] must translate to [tool(A), user(text), tool(B)] — each tool_result +// flushing whatever content/tool_calls had accumulated before it, not +// just the first one. +func TestOpenAIMessagesFromAnthropic_InterspersedToolResults(t *testing.T) { + content := []any{ + map[string]any{"type": "tool_result", "tool_use_id": "call_A", "content": "a"}, + map[string]any{"type": "text", "text": "middle"}, + map[string]any{"type": "tool_result", "tool_use_id": "call_B", "content": "b"}, + } + + out, err := openAIMessagesFromAnthropic("user", content) + if err != nil { + t.Fatalf("openAIMessagesFromAnthropic: %v", err) + } + if len(out) != 3 { + t.Fatalf("messages = %v, want 3 (tool A, text, tool B)", out) + } + m0, _ := out[0].(map[string]any) + m1, _ := out[1].(map[string]any) + m2, _ := out[2].(map[string]any) + if m0["tool_call_id"] != "call_A" { + t.Errorf("messages[0].tool_call_id = %v, want call_A", m0["tool_call_id"]) + } + if m1["role"] != "user" { + t.Errorf("messages[1].role = %v, want user (the interspersed text)", m1["role"]) + } + if m2["tool_call_id"] != "call_B" { + t.Errorf("messages[2].tool_call_id = %v, want call_B", m2["tool_call_id"]) + } +} + +// TestOpenAIToolMessageFromAnthropic_PreservesIsError is the regression +// for item 11a (IMPORTANT, 2026-08-22 review): Anthropic's +// tool_result.is_error has no native OpenAI tool-message field, so the +// previous version of this function silently dropped it. A string +// content now gets an "Error: " prefix so the failure signal survives +// into the one shape OpenAI's tool message actually carries. +func TestOpenAIToolMessageFromAnthropic_PreservesIsError(t *testing.T) { + cases := []struct { + content any + wantContent any + name string + isError bool + }{ + {name: "string content with is_error prefixes Error:", content: "boom", isError: true, wantContent: "Error: boom"}, + {name: "string content without is_error is unchanged", content: "ok", isError: false, wantContent: "ok"}, + {name: "non-string content is left as-is even with is_error", content: []any{"x"}, isError: true, wantContent: []any{"x"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + bm := map[string]any{"tool_use_id": "call_1", "content": tc.content, "is_error": tc.isError} + got := openAIToolMessageFromAnthropic(bm) + assert.Equal(t, tc.wantContent, got["content"]) + }) + } +} + +// TestOpenAIRequestFromAnthropic_ThinkingFieldReportedAsDropped is the +// regression for item 11b (IMPORTANT, 2026-08-22 review): Anthropic's +// "thinking" (extended-thinking config) has no OpenAI equivalent. +// openAIRequestFromAnthropic must report it in its dropped-fields return +// rather than silently discarding it, so the caller (routes_messages.go's +// callTranslatedMessages) can log a warning instead of the drop being +// invisible. +func TestOpenAIRequestFromAnthropic_ThinkingFieldReportedAsDropped(t *testing.T) { + req := map[string]any{ + "model": "claude-x", + "thinking": map[string]any{"type": "enabled", "budget_tokens": float64(1024)}, + "messages": []any{map[string]any{"role": "user", "content": "hi"}}, + } + + _, dropped, err := openAIRequestFromAnthropic(req) + if err != nil { + t.Fatalf("openAIRequestFromAnthropic: %v", err) + } + if len(dropped) != 1 || dropped[0] != "thinking" { + t.Errorf("dropped = %v, want [\"thinking\"]", dropped) + } +} + +// TestOpenAIRequestFromAnthropic_ThinkingAbsent_NothingDropped proves the +// dropped-fields list stays empty for a request that never set +// "thinking" in the first place — the previous test's absence is not +// itself evidence of a bug. +func TestOpenAIRequestFromAnthropic_ThinkingAbsent_NothingDropped(t *testing.T) { + req := map[string]any{ + "model": "claude-x", + "messages": []any{map[string]any{"role": "user", "content": "hi"}}, + } + + _, dropped, err := openAIRequestFromAnthropic(req) + if err != nil { + t.Fatalf("openAIRequestFromAnthropic: %v", err) + } + if len(dropped) != 0 { + t.Errorf("dropped = %v, want none", dropped) + } +} + +// TestOpenAIRequestFromAnthropic_ReasoningModel_UsesMaxCompletionTokens is +// the regression for item 10 (IMPORTANT, 2026-08-22 review): OpenAI's +// o-series and gpt-5-family reasoning models reject the "max_tokens" +// request field with a 400 and require "max_completion_tokens" instead. +// Every other model keeps using "max_tokens" as before. +func TestOpenAIRequestFromAnthropic_ReasoningModel_UsesMaxCompletionTokens(t *testing.T) { + cases := []struct { + model string + wantField string + wantOthers []string + }{ + {"o3-mini", "max_completion_tokens", []string{"max_tokens"}}, + {"o1", "max_completion_tokens", []string{"max_tokens"}}, + {"o4-mini-2025-04-16", "max_completion_tokens", []string{"max_tokens"}}, + {"gpt-5", "max_completion_tokens", []string{"max_tokens"}}, + {"gpt-5.1-chat-latest", "max_completion_tokens", []string{"max_tokens"}}, + {"gpt-4o", "max_tokens", []string{"max_completion_tokens"}}, + {"gpt-4o-mini", "max_tokens", []string{"max_completion_tokens"}}, + } + for _, tc := range cases { + t.Run(tc.model, func(t *testing.T) { + req := map[string]any{ + "model": tc.model, + "max_tokens": float64(256), + "messages": []any{map[string]any{"role": "user", "content": "hi"}}, + } + out, _, err := openAIRequestFromAnthropic(req) + if err != nil { + t.Fatalf("openAIRequestFromAnthropic: %v", err) + } + if _, ok := out[tc.wantField]; !ok { + t.Errorf("out[%q] missing, want it set to 256", tc.wantField) + } + for _, other := range tc.wantOthers { + if _, ok := out[other]; ok { + t.Errorf("out[%q] present, want only %q set", other, tc.wantField) + } + } + }) + } +} + +// TestAnthropicUsagePayload_TotalInputTokens_FoldsCacheCounters is the +// unit-level regression for item 6 (IMPORTANT, 2026-08-22 review): +// cache_creation_input_tokens and cache_read_input_tokens must fold into +// the billed prompt count alongside fresh input_tokens, not be dropped. +func TestAnthropicUsagePayload_TotalInputTokens_FoldsCacheCounters(t *testing.T) { + p := anthropicUsagePayload{InputTokens: 4, CacheCreationInputTokens: 180000, CacheReadInputTokens: 20000} + if got, want := p.totalInputTokens(), int64(200004); got != want { + t.Errorf("totalInputTokens() = %d, want %d", got, want) + } +}