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) + } +}