diff --git a/agent-schema.json b/agent-schema.json index eab152f6a..2a7c3a66a 100644 --- a/agent-schema.json +++ b/agent-schema.json @@ -1756,7 +1756,7 @@ "additionalProperties": true } }, - "description": "Provider-specific options. Anthropic: thinking_prefix_mismatch (error or drop_block; defaults to drop_block on Claude models whose server checks thinking prefixes: Haiku 5.5, Opus 5.5, Sonnet 5.5 and Fable 5.1, plus Fable 5.2+ by name), strict_tools (true for eligible schemas or a tool-name array requiring strict mode), cache_diagnostics (boolean; Claude API only), native_compaction (boolean; synchronous whole-history signed compaction on Claude API), cache_preserving_updates (boolean; native system/tool/effort changes with turn-scoped reminders on supported models), thinking_display: updates (progress updates beta on Fable 5+ and Mythos 5.1). Sampling parameters: top_k (integer, supported by anthropic, google, amazon-bedrock, and custom OpenAI-compatible providers like vLLM/Ollama), repetition_penalty (float, forwarded to custom OpenAI-compatible providers), min_p (float, forwarded to custom providers), seed (integer, forwarded to OpenAI). Lifecycle: unload_api (string) overrides the unload endpoint inherited from the provider config (relative path resolved against base_url's scheme+host, or an absolute URL); used by the runtime's `unload` on_agent_switch builtin hook to release model resources between agent switches. Infrastructure options: http_headers (map of string to string, adds custom HTTP headers to every request; used for OpenAI-compatible providers like github-copilot which requires Copilot-Integration-Id). dmr: runtime_flags; supports_images (boolean) and supports_pdf (boolean) declare which document attachment types the local model accepts. DMR-hosted models are not in the models.dev catalog, so attachment capabilities cannot be detected automatically and default to text-only; set supports_images: true for a vision model so image attachments are forwarded instead of silently dropped. anthropic/amazon-bedrock (Claude): interleaved_thinking (boolean, default true). anthropic: thinking_display ('summarized', 'omitted', 'updates', or 'display') controls whether thinking blocks are returned in responses when thinking is enabled. Claude Opus 4.7 hides thinking by default ('omitted'); set thinking_display: summarized to receive thinking blocks. 'display' (full thinking blocks) is only accepted by pre-4.6 token-thinking models (e.g. Sonnet 4.5, Haiku 4.5); models from the adaptive-thinking generation onward (Opus/Sonnet 4.6+, Sonnet 5, Fable 5) reject it, and docker-agent fails fast with a configuration error (fallbacks models are validated too); fallbacks (\"default\" for the server's recommended routing, or an array of up to three distinct model ID strings in priority order; enables the server-side-fallback-2026-07-01 beta so refused requests are retried in a single round trip; Claude API only, rejected on Vertex AI, Bedrock, gateways and other base_url hosts; a list is rejected for Claude Haiku 5.5; rejected when the primary or any fallback is a Claude 5.x model until issue #4584). OpenAI-compatible APIs: service_tier (string, forwarded unchanged to Chat Completions and Responses, including WebSocket and reranking; 'fast' or 'priority' enables Fast mode at premium pricing on supported models; omitted or empty leaves the API default, non-string values are ignored). openai/chatgpt: preserve_reasoning (boolean, default false) preserves ordered Responses output and encrypted reasoning for stateless continuation; native_tool_search also enables output replay. Any OpenAI-compatible destination: replay_reasoning (boolean) replays the reasoning stored on assistant turns back as a reasoning_content field on later Chat Completions requests, so a reasoning model sees its own earlier reasoning instead of re-deriving it after each tool call (distinct from preserve_reasoning, which replays encrypted reasoning items on the Responses API and does not apply to Chat Completions); default true for Chat Completions; set replay_reasoning to false for an endpoint that rejects or must not receive the non-standard field; a non-boolean value also disables replay. openai: cache_diagnostics (boolean, default false) logs Responses prompt-cache comparison results with --debug on GPT-5.6+; comparisons are scoped to the current session and resolved model; ignored for chatgpt and custom base_url endpoints. transport ('sse' or 'websocket') to choose between SSE and WebSocket streaming for the Responses API; native_tool_search (boolean, default false; also enables ordered output replay) declares the agent's deferred tools with defer_loading and a server-executed tool_search tool so OpenAI's hosted tool search discovers them (only openai models supporting deferred tools, gpt-5.4+; ignored on chatgpt, OpenAI-compatible endpoints including a custom base_url, and Chat Completions requests). openai/anthropic/google: rerank_prompt (string) to fully override the system prompt used for RAG reranking (advanced - prefer using results.reranking.criteria for domain-specific guidance). Google: google_search (boolean) enables Google Search grounding, google_maps (boolean) enables Google Maps grounding, code_execution (boolean) enables server-side code execution, url_context (boolean) enables reading public URLs during generation, output_dimensionality (positive integer) truncates Gemini embedding vectors to that size when the model is used as a RAG embedding_model; service_tier (string, forwarded unchanged as the top-level serviceTier of every generateContent request, including reranking, title generation, and compaction, on the Gemini API, Vertex AI, and the Docker models gateway; 'flex' for discounted, queued processing that may wait up to 15 minutes before the first byte, 'standard' for the default tier, 'priority' for premium processing; omitted or empty leaves the API default, non-string values are ignored; cost estimates are not tier-adjusted, set the model's cost override).", + "description": "Provider-specific options. Anthropic: thinking_prefix_mismatch (error or drop_block; defaults to drop_block on Claude models whose server checks thinking prefixes: Haiku 5.5, Opus 5.5, Sonnet 5.5 and Fable 5.1, plus Fable 5.2+ by name), strict_tools (true for eligible schemas or a tool-name array requiring strict mode), cache_diagnostics (boolean; Claude API only), native_compaction (boolean; synchronous whole-history signed compaction on Claude API), cache_preserving_updates (boolean; native system/tool/effort changes with turn-scoped reminders on supported models), thinking_display: updates (progress updates beta on Fable 5+ and Mythos 5.1). Sampling parameters: top_k (integer, supported by anthropic, google, amazon-bedrock, and custom OpenAI-compatible providers like vLLM/Ollama), repetition_penalty (float, forwarded to custom OpenAI-compatible providers), min_p (float, forwarded to custom providers), seed (integer, forwarded to OpenAI). Lifecycle: unload_api (string) overrides the unload endpoint inherited from the provider config (relative path resolved against base_url's scheme+host, or an absolute URL); used by the runtime's `unload` on_agent_switch builtin hook to release model resources between agent switches. Infrastructure options: http_headers (map of string to string, adds custom HTTP headers to every request; used for OpenAI-compatible providers like github-copilot which requires Copilot-Integration-Id). dmr: runtime_flags; supports_images (boolean) and supports_pdf (boolean) declare which document attachment types the local model accepts. DMR-hosted models are not in the models.dev catalog, so attachment capabilities cannot be detected automatically and default to text-only; set supports_images: true for a vision model so image attachments are forwarded instead of silently dropped. anthropic/amazon-bedrock (Claude): interleaved_thinking (boolean, default true). anthropic: thinking_display ('summarized', 'omitted', 'updates', or 'display') controls whether thinking blocks are returned in responses when thinking is enabled. Claude Opus 4.7 hides thinking by default ('omitted'); set thinking_display: summarized to receive thinking blocks. 'display' (full thinking blocks) is only accepted by pre-4.6 token-thinking models (e.g. Sonnet 4.5, Haiku 4.5); models from the adaptive-thinking generation onward (Opus/Sonnet 4.6+, Sonnet 5, Fable 5) reject it, and docker-agent fails fast with a configuration error (fallbacks models are validated too); fallbacks (\"default\" for the server's recommended routing, or an array of up to three distinct model ID strings in priority order; enables the server-side-fallback-2026-07-01 beta so refused requests are retried in a single round trip; Claude API only, rejected on Vertex AI, Bedrock, gateways and other base_url hosts; a list is rejected for Claude Haiku 5.5, which has no server-side fallback). OpenAI-compatible APIs: service_tier (string, forwarded unchanged to Chat Completions and Responses, including WebSocket and reranking; 'fast' or 'priority' enables Fast mode at premium pricing on supported models; omitted or empty leaves the API default, non-string values are ignored). openai/chatgpt: preserve_reasoning (boolean, default false) preserves ordered Responses output and encrypted reasoning for stateless continuation; native_tool_search also enables output replay. Any OpenAI-compatible destination: replay_reasoning (boolean) replays the reasoning stored on assistant turns back as a reasoning_content field on later Chat Completions requests, so a reasoning model sees its own earlier reasoning instead of re-deriving it after each tool call (distinct from preserve_reasoning, which replays encrypted reasoning items on the Responses API and does not apply to Chat Completions); default true for Chat Completions; set replay_reasoning to false for an endpoint that rejects or must not receive the non-standard field; a non-boolean value also disables replay. openai: cache_diagnostics (boolean, default false) logs Responses prompt-cache comparison results with --debug on GPT-5.6+; comparisons are scoped to the current session and resolved model; ignored for chatgpt and custom base_url endpoints. transport ('sse' or 'websocket') to choose between SSE and WebSocket streaming for the Responses API; native_tool_search (boolean, default false; also enables ordered output replay) declares the agent's deferred tools with defer_loading and a server-executed tool_search tool so OpenAI's hosted tool search discovers them (only openai models supporting deferred tools, gpt-5.4+; ignored on chatgpt, OpenAI-compatible endpoints including a custom base_url, and Chat Completions requests). openai/anthropic/google: rerank_prompt (string) to fully override the system prompt used for RAG reranking (advanced - prefer using results.reranking.criteria for domain-specific guidance). Google: google_search (boolean) enables Google Search grounding, google_maps (boolean) enables Google Maps grounding, code_execution (boolean) enables server-side code execution, url_context (boolean) enables reading public URLs during generation, output_dimensionality (positive integer) truncates Gemini embedding vectors to that size when the model is used as a RAG embedding_model; service_tier (string, forwarded unchanged as the top-level serviceTier of every generateContent request, including reranking, title generation, and compaction, on the Gemini API, Vertex AI, and the Docker models gateway; 'flex' for discounted, queued processing that may wait up to 15 minutes before the first byte, 'standard' for the default tier, 'priority' for premium processing; omitted or empty leaves the API default, non-string values are ignored; cost estimates are not tier-adjusted, set the model's cost override).", "additionalProperties": true }, "track_usage": { diff --git a/docs/providers/anthropic/index.md b/docs/providers/anthropic/index.md index 559c08511..085221fba 100644 --- a/docs/providers/anthropic/index.md +++ b/docs/providers/anthropic/index.md @@ -125,11 +125,10 @@ make two endpoints interchangeable. | Bedrock runtime | `amazon-bedrock`; **`ConverseStream`** | Regional Bedrock runtime endpoint | AWS SigV4 credentials or Bedrock bearer token | Exact AWS model or inference-profile ID | | OpenAI-compatible aggregators | OpenAI client, Chat Completions for Claude | Alias/custom endpoint (e.g. OpenRouter, Vercel, Azure, OpenCode) | Host-specific key | Host's catalogue ID, not necessarily a Claude API ID | -These are existing adapter paths, not a guarantee of complete Claude 5.x -compatibility. Model-specific request rules, signed history, thinking and fallback -compatibility remain the subject of the planned [core work (#4583)](https://github.com/docker/docker-agent/issues/4583) -and [fallback work (#4584)](https://github.com/docker/docker-agent/issues/4584). -This guide does not mark those changes as implemented. +These are the adapter paths Docker Agent uses. Claude 5.x request rules, signed +history and thinking ([#4583](https://github.com/docker/docker-agent/issues/4583)) +and server-side fallback ([#4584](https://github.com/docker/docker-agent/issues/4584)) +are described in the sections below. - The built-in `opencode` alias is OpenAI-compatible; the [custom OpenCode Messages provider](../opencode-zen/index.md#anthropic-compatible-messages-api) @@ -262,7 +261,7 @@ Docker Agent recognizes nine Claude 5.x models by ID: `claude-haiku-5-5`, `claud - An integer token budget becomes adaptive thinking with the model's default effort; `max_tokens` is not raised. - `temperature`, `top_p` and `top_k` are never sent. - `thinking_budget: none` (or `0`) sends `thinking: disabled` on Haiku 5.5, Opus 5 and Sonnet 5. The other six cannot disable thinking: they get adaptive thinking with `low` effort, and Docker Agent logs a warning at startup. -- Server-side `fallbacks` involving any of these models are rejected at startup until [#4584](https://github.com/docker/docker-agent/issues/4584). +- Server-side `fallbacks` work with these models on the Claude API, except Haiku 5.5, which has none (see [Server-Side Fallbacks](#server-side-fallbacks)). - A request that ends on an assistant message (a response prefill) is rejected before it is sent. Tool-mode structured output therefore fails after the first plain-text answer instead of retrying with a reminder. Other Claude IDs keep the existing name-based rules. @@ -369,10 +368,12 @@ Or list up to three distinct models, tried in order: - claude-sonnet-4-6 ``` -> [!WARNING] -> **Not yet available with Claude 5.x** -> -> A fallback can change the serving model mid-response, which Docker Agent does not handle yet. `fallbacks` is rejected at startup when the primary model or any listed fallback is a Claude 5.x model (Haiku 5.5, Opus 5/5.5, Sonnet 5/5.5, Fable 5/5.1, Mythos 5/5.1). Support is tracked in [#4584](https://github.com/docker/docker-agent/issues/4584). +When a model declines mid-response, the turn continues on the fallback model. +Docker Agent never runs a tool call the declining model made, stores the turn +under the model that served it, and replays it on the next request as the API +requires. The turn's cost adds up each attempt at its own model's rates; a +declined attempt that produced no output is billed only in the refusal +categories Anthropic bills. Docker Agent sends the `server-side-fallback-2026-07-01` beta header, which accepts both forms, through the beta Messages API. The response's `model` @@ -386,15 +387,22 @@ Rules checked at startup: Vertex AI, Bedrock, a gateway, or a base URL (`base_url` or `ANTHROPIC_BASE_URL`) whose host isn't `api.anthropic.com`. It isn't available on the Message Batches API either. -- Claude Haiku 5.5 has no server-side fallback. Once Claude 5.x chains are - supported, a list will be rejected and `default` will only log that it has - no effect. +- Claude Haiku 5.5 has no server-side fallback: a list is rejected, and + `default` only logs that it has no effect. +- The request must be valid for every listed model, or the API rejects it up + front. A configuration is rejected when a listed fallback (such as a Claude + 5.x model) refuses what the request sends: `temperature` or `top_p` (sent + only with thinking off), `top_k`, or token-budget thinking, including an + effort level that an older model without adaptive thinking turns into a + token budget. Fallback models receive the exact same request as the primary model (thinking configuration, task budget, beta features, ...), so list only models that accept the same request shape. With `default` the server picks the fallback, so per-model options such as `thinking_display` are checked -against the primary model only. +against the primary model only. The API makes one exception itself: when a Claude +Sonnet 5.5 request with `between_tools` thinking falls back to Claude Sonnet 5, +that attempt runs with thinking disabled ([fallback list rules](https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback#naming-your-own-fallback-models)). See [Refusals and fallback](https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback#server-side-fallback). diff --git a/examples/anthropic-fallbacks.yaml b/examples/anthropic-fallbacks.yaml index a0d521774..c57342007 100644 --- a/examples/anthropic-fallbacks.yaml +++ b/examples/anthropic-fallbacks.yaml @@ -3,9 +3,6 @@ # model in a single round trip. `fallbacks: default` uses the model Anthropic # recommends for the refusal's category; a list (up to three) is tried in # order. -# -# Fallback chains that include a Claude 5.x model are rejected until -# https://github.com/docker/docker-agent/issues/4584 lands. models: opus: provider: anthropic diff --git a/pkg/chat/chat.go b/pkg/chat/chat.go index 3e1302737..3e3507c5e 100644 --- a/pkg/chat/chat.go +++ b/pkg/chat/chat.go @@ -225,6 +225,9 @@ type MessageStreamResponse struct { Model string `json:"model"` Choices []MessageStreamChoice `json:"choices"` Usage *Usage `json:"usage,omitempty"` + // UsageAttempts lists each model that ran when a server fallback spread + // the response across models; Usage describes only the last one. + UsageAttempts []UsageAttempt `json:"usage_attempts,omitempty"` } type Usage struct { @@ -237,6 +240,25 @@ type Usage struct { ServiceTier string `json:"service_tier,omitempty"` } +// UsageAttempt is one attempt of a response a server fallback spread across +// models, billed at that model's rates unless Billed is false. +type UsageAttempt struct { + Model string `json:"model"` + Billed bool `json:"billed"` + // Dispatched marks the attempt of the requested model, whose reported ID + // may be a resolved form of the configured one. + Dispatched bool `json:"dispatched,omitempty"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + CachedInputTokens int64 `json:"cached_input_tokens"` + CacheWriteTokens int64 `json:"cached_write_tokens"` +} + +// Usage returns the attempt's token counts as a Usage. +func (a UsageAttempt) Usage() *Usage { + return &Usage{InputTokens: a.InputTokens, OutputTokens: a.OutputTokens, CachedInputTokens: a.CachedInputTokens, CacheWriteTokens: a.CacheWriteTokens} +} + // PromptTokens sums the disjoint fresh, cache-read, and cache-write input buckets. func (u *Usage) PromptTokens() int64 { return u.InputTokens + u.CachedInputTokens + u.CacheWriteTokens diff --git a/pkg/model/provider/anthropic/anthropictest/server.go b/pkg/model/provider/anthropic/anthropictest/server.go index 836b94213..2320d6754 100644 --- a/pkg/model/provider/anthropic/anthropictest/server.go +++ b/pkg/model/provider/anthropic/anthropictest/server.go @@ -113,12 +113,18 @@ func CompactionReply(id, summary, signature string, iterations ...[2]int64) stri ) } -// FallbackBeforeOutputReply is a turn that from declined before any output: -// message_start already names to, and the fallback marker comes first. -func FallbackBeforeOutputReply(id, from, to, text string) string { - events := append([]sse{messageStart(id, to, 0)}, fallbackBlock(0, from, to)...) +// FallbackBeforeOutputReply is a turn that from declined before any output, +// in the refusal category: message_start already names to, the fallback +// marker comes first, and the declined attempt reports no output tokens. +func FallbackBeforeOutputReply(id, from, to, category, text string) string { + events := append([]sse{messageStart(id, to, 0)}, fallbackBlock(0, from, to, category)...) events = append(events, textBlock(1, text)...) - events = append(events, messageDelta("end_turn", fallbackUsage(from, to)), event("message_stop", nil)) + declined := iteration("message", from) + declined["output_tokens"] = 0 + events = append(events, messageDelta("end_turn", map[string]any{ + "input_tokens": 0, "output_tokens": 7, + "iterations": []any{declined, iteration("fallback_message", to)}, + }), event("message_stop", nil)) return body(events...) } @@ -139,7 +145,7 @@ func FallbackMidOutputReply(id, from, to, before, after, toolID string) string { event("content_block_delta", map[string]any{"index": 2, "delta": map[string]any{"type": "input_json_delta", "partial_json": "{}"}}), event("content_block_stop", map[string]any{"index": 2}), ) - events = append(events, fallbackBlock(3, from, to)...) + events = append(events, fallbackBlock(3, from, to, "cyber")...) events = append(events, textBlock(4, after)...) events = append(events, messageDelta("end_turn", fallbackUsage(from, to)), event("message_stop", nil)) return body(events...) @@ -159,7 +165,7 @@ func FallbackChainReply(id string, models ...string) string { break } iterations = append(iterations, iteration("message", model)) - events = append(events, fallbackBlock(index, model, models[i+1])...) + events = append(events, fallbackBlock(index, model, models[i+1], "cyber")...) index++ } events = append(events, messageDelta("end_turn", map[string]any{"input_tokens": 0, "output_tokens": 7, "iterations": iterations}), event("message_stop", nil)) @@ -179,11 +185,11 @@ func StickyReply(id, to, text string) string { } // fallbackBlock is a fallback marker: a block start and stop with no deltas. -func fallbackBlock(index int, from, to string) []sse { +func fallbackBlock(index int, from, to, category string) []sse { return []sse{ event("content_block_start", map[string]any{"index": index, "content_block": map[string]any{ "type": "fallback", "from": map[string]any{"model": from}, "to": map[string]any{"model": to}, - "trigger": map[string]any{"type": "refusal", "category": "cyber"}, + "trigger": map[string]any{"type": "refusal", "category": category}, }}), event("content_block_stop", map[string]any{"index": index}), } @@ -205,7 +211,7 @@ func fallbackUsage(from, to string) map[string]any { } func iteration(typ, model string) map[string]any { - return map[string]any{"type": typ, "model": model, "input_tokens": 10, "output_tokens": 7, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0} + return map[string]any{"type": typ, "model": model, "input_tokens": 10, "output_tokens": 7, "cache_read_input_tokens": 3, "cache_creation_input_tokens": 4} } type sse struct { diff --git a/pkg/model/provider/anthropic/anthropictest/server_test.go b/pkg/model/provider/anthropic/anthropictest/server_test.go index 1260b75b3..8e1a2eac2 100644 --- a/pkg/model/provider/anthropic/anthropictest/server_test.go +++ b/pkg/model/provider/anthropic/anthropictest/server_test.go @@ -44,7 +44,7 @@ func TestFallbackRepliesAccumulate(t *testing.T) { reply string markers int }{ - "before output": {reply: anthropictest.FallbackBeforeOutputReply("m1", "claude-a", "claude-b", "hi"), markers: 1}, + "before output": {reply: anthropictest.FallbackBeforeOutputReply("m1", "claude-a", "claude-b", "cyber", "hi"), markers: 1}, "mid output": {reply: anthropictest.FallbackMidOutputReply("m2", "claude-a", "claude-b", "before", "after", "t1"), markers: 1}, "sticky": {reply: anthropictest.StickyReply("m3", "claude-b", "hi")}, } { diff --git a/pkg/model/provider/anthropic/beta_adapter.go b/pkg/model/provider/anthropic/beta_adapter.go index bd20ca066..cb34ad8b0 100644 --- a/pkg/model/provider/anthropic/beta_adapter.go +++ b/pkg/model/provider/anthropic/beta_adapter.go @@ -37,6 +37,9 @@ type betaStreamAdapter struct { // (cancellation, idle timeout). config base.Config served atomic.Pointer[string] + // declinedCategories holds each fallback marker's refusal category, in + // order: one per declined attempt. + declinedCategories []string } // newBetaStreamAdapter creates a new Beta stream adapter @@ -92,6 +95,7 @@ func (a *betaStreamAdapter) Recv() (chat.MessageStreamResponse, error) { // The declined model's tool calls are superseded. a.toolCall = false a.served.Store(&block.To.Model) + a.declinedCategories = append(a.declinedCategories, string(block.Trigger.Category)) response.Choices[0].Delta.ServerFallback = &chat.ServerFallback{From: block.From.Model, To: block.To.Model} case anthropic.BetaThinkingBlock: if block.Thinking != "" { @@ -138,6 +142,7 @@ func (a *betaStreamAdapter) Recv() (chat.MessageStreamResponse, error) { } if a.trackUsage { response.Usage = betaUsageFromMessage(a.message.Usage) + response.UsageAttempts = serverFallbackAttempts(a.message.Usage.Iterations, a.declinedCategories) } case anthropic.BetaRawMessageStopEvent: for _, dropped := range thinkingTransformations(a.message.InputTransformations) { @@ -202,3 +207,47 @@ func betaUsageFromMessage(u anthropic.BetaUsage) *chat.Usage { ReasoningTokens: u.OutputTokensDetails.ThinkingTokens, } } + +// billedBeforeOutput lists the refusal categories billed even when the model +// declined before any output, as of September 2026; Anthropic may change it. +// https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback#how-refusals-are-billed +var billedBeforeOutput = map[string]bool{"bio": true, "frontier_llm": true, "reasoning_extraction": true} + +// serverFallbackAttempts lists the attempts of a response a server fallback +// spread across models, from usage.iterations: each declined hop ("message") +// and the attempt that served it ("fallback_message"). A declined hop with no +// output is billed only in a billed category; when the markers don't line up +// with the declined hops, every hop counts as billed. Nil without a fallback. +func serverFallbackAttempts(iterations []anthropic.BetaIterationsUsageItemUnion, categories []string) []chat.UsageAttempt { + if !slices.ContainsFunc(iterations, func(it anthropic.BetaIterationsUsageItemUnion) bool { return it.Type == "fallback_message" }) { + return nil + } + declined := 0 + for _, it := range iterations { + if it.Type == "message" { + declined++ + } + } + knownCategories := declined == len(categories) + var attempts []chat.UsageAttempt + hop := 0 + for _, it := range iterations { + if it.Type != "message" && it.Type != "fallback_message" { + continue + } + billed := true + if it.Type == "message" { + // An empty category isn't a known unbilled refusal: bill it. + if knownCategories && it.OutputTokens == 0 && categories[hop] != "" && !billedBeforeOutput[categories[hop]] { + billed = false + } + hop++ + } + attempts = append(attempts, chat.UsageAttempt{ + Model: it.Model, Billed: billed, Dispatched: len(attempts) == 0 && it.Type == "message", + InputTokens: it.InputTokens, OutputTokens: it.OutputTokens, + CachedInputTokens: it.CacheReadInputTokens, CacheWriteTokens: it.CacheCreationInputTokens, + }) + } + return attempts +} diff --git a/pkg/model/provider/anthropic/fallbacks.go b/pkg/model/provider/anthropic/fallbacks.go index 66cb43fa7..691c6f20b 100644 --- a/pkg/model/provider/anthropic/fallbacks.go +++ b/pkg/model/provider/anthropic/fallbacks.go @@ -13,6 +13,7 @@ import ( "github.com/anthropics/anthropic-sdk-go/shared/constant" "github.com/docker/docker-agent/pkg/config/latest" + "github.com/docker/docker-agent/pkg/model/provider/base" "github.com/docker/docker-agent/pkg/modelinfo" ) @@ -130,6 +131,22 @@ func validateFallbackSpec(cfg *latest.ModelConfig) error { if slices.Contains(spec.Models, cfg.Model) { return fmt.Errorf("anthropic: fallback %q is the same as the model", cfg.Model) } + // The request must be valid for every listed model, or the API rejects it + // up front. Claude 5.x and newer Opus models refuse sampling settings and + // token thinking budgets that older primaries send. Check what the request + // builder actually sends, after it resolves the thinking budget. + sent := sentRequestShape(cfg) + for _, m := range spec.Models { + if sent.sampling && rejectsSampling(m) { + return fmt.Errorf("anthropic: the request sends temperature or top_p (thinking is off), which fallback %q rejects; remove them or that fallback", m) + } + if sent.topK && rejectsSampling(m) { + return fmt.Errorf("anthropic: the request sends top_k, which fallback %q rejects; remove it or that fallback", m) + } + if sent.tokenThinking && modelinfo.RejectsTokenThinking(m) { + return fmt.Errorf("anthropic: %s thinks with a token budget, which fallback %q rejects; turn thinking off or remove that fallback", cfg.Model, m) + } + } if m, ok := modelinfo.LookupClaude5(cfg.Model); ok && m.NoServerSideFallback { if len(spec.Models) > 0 { return fmt.Errorf("anthropic: %s has no server-side fallback; remove the fallbacks list", cfg.Model) @@ -141,19 +158,26 @@ func validateFallbackSpec(cfg *latest.ModelConfig) error { return nil } -// validateFallbacks rejects server-side fallback when any model in the chain -// is Claude 5.x: a fallback can change the serving model mid-response, which -// the client does not handle yet. Malformed values are reported by -// [validateFallbackSpec]. -func validateFallbacks(cfg *latest.ModelConfig) error { - spec, _ := parseFallbacks(cfg.ProviderOpts) - if !spec.Default && len(spec.Models) == 0 { - return nil - } - for _, model := range append([]string{cfg.Model}, spec.Models...) { - if _, ok := modelinfo.LookupClaude5(model); ok { - return fmt.Errorf("anthropic: server-side fallback involving Claude 5.x model %q is not supported yet (see https://github.com/docker/docker-agent/issues/4584)", model) - } +// requestShape is what a request built from a config sends, for the fields a +// fallback model may refuse. +type requestShape struct { + sampling, topK, tokenThinking bool +} + +// sentRequestShape mirrors the request builder: temperature and top_p only go +// out with thinking off, top_k whenever set, and the thinking budget after +// resolution, which turns an effort level into tokens on models without +// adaptive thinking. +func sentRequestShape(cfg *latest.ModelConfig) requestShape { + c := &Client{Config: base.Config{ModelConfig: *cfg}} + budget := c.resolveThinkingBudget() + thinking := budget != nil && !budget.IsDisabled() + _, adaptive := anthropicThinkingEffort(budget) + _, topK := cfg.ProviderOpts["top_k"] + primaryRejects := rejectsSampling(cfg.Model) + return requestShape{ + sampling: !thinking && !primaryRejects && (cfg.Temperature != nil || cfg.TopP != nil), + topK: topK && !primaryRejects, + tokenThinking: thinking && !adaptive && budget.Tokens > 0, } - return nil } diff --git a/pkg/model/provider/anthropic/fallbacks_test.go b/pkg/model/provider/anthropic/fallbacks_test.go index acc3e5fe8..be2ccd04a 100644 --- a/pkg/model/provider/anthropic/fallbacks_test.go +++ b/pkg/model/provider/anthropic/fallbacks_test.go @@ -11,52 +11,37 @@ import ( "github.com/docker/docker-agent/pkg/model/provider/options" ) -func TestValidateFallbacks(t *testing.T) { +// Claude 5.x models accept server-side fallback, except that Haiku 5.5 has +// none: a list is a 400 there, and default has no effect. +func TestNewClient_Claude5Fallbacks(t *testing.T) { t.Parallel() - cases := []struct { - name string + env := environment.NewMapEnvProvider(map[string]string{"ANTHROPIC_API_KEY": "k"}) + for name, tc := range map[string]struct { model string fallbacks any - wantModel string + wantErr string }{ - {name: "5.x primary", model: "claude-fable-5-1", fallbacks: []any{"claude-opus-4-8"}, wantModel: "claude-fable-5-1"}, - {name: "legacy primary, 5.x target", model: "claude-opus-4-8", fallbacks: []any{"claude-sonnet-5-5"}, wantModel: "claude-sonnet-5-5"}, - {name: "gateway target", model: "claude-opus-4-8", fallbacks: []any{"vendor/claude-opus-5-5"}, wantModel: "vendor/claude-opus-5-5"}, - {name: "vertex target", model: "claude-opus-4-8", fallbacks: []any{"claude-sonnet-5-5@20261001"}, wantModel: "claude-sonnet-5-5@20261001"}, - {name: "absent", model: "claude-fable-5-1"}, - {name: "empty", model: "claude-fable-5-1", fallbacks: []any{}}, - {name: "no 5.x model", model: "claude-opus-4-8", fallbacks: []any{"claude-opus-4-7"}}, - {name: "default on 5.x primary", model: "claude-fable-5-1", fallbacks: "default", wantModel: "claude-fable-5-1"}, - {name: "default on legacy primary", model: "claude-opus-4-8", fallbacks: "default"}, - } - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { + "fable 5.1 default": {model: "claude-fable-5-1", fallbacks: "default"}, + "fable 5.1 list": {model: "claude-fable-5-1", fallbacks: []any{"claude-opus-5-5"}}, + "legacy to 5.x": {model: "claude-opus-4-8", fallbacks: []any{"claude-sonnet-5-5"}}, + "sonnet 5.5 to 5": {model: "claude-sonnet-5-5", fallbacks: []any{"claude-sonnet-5"}}, + "haiku 5.5 default": {model: "claude-haiku-5-5", fallbacks: "default"}, + "haiku 5.5 list": {model: "claude-haiku-5-5", fallbacks: []any{"claude-sonnet-4-6"}, wantErr: "no server-side fallback"}, + } { + t.Run(name, func(t *testing.T) { t.Parallel() - cfg := &latest.ModelConfig{Provider: "anthropic", Model: tc.model} - if tc.fallbacks != nil { - cfg.ProviderOpts = map[string]any{"fallbacks": tc.fallbacks} - } - err := validateFallbacks(cfg) - if tc.wantModel == "" { - assert.NoError(t, err) + cfg := &latest.ModelConfig{Provider: "anthropic", Model: tc.model, ProviderOpts: map[string]any{"fallbacks": tc.fallbacks}} + _, err := NewClient(t.Context(), cfg, env) + if tc.wantErr == "" { + require.NoError(t, err) return } - require.Error(t, err) - assert.Contains(t, err.Error(), `"`+tc.wantModel+`"`) - assert.Contains(t, err.Error(), "issues/4584") + require.ErrorContains(t, err, tc.wantErr) }) } } -func TestNewClient_RejectsClaude5Fallbacks(t *testing.T) { - t.Parallel() - - cfg := &latest.ModelConfig{Provider: "anthropic", Model: "claude-opus-4-8", ProviderOpts: map[string]any{"fallbacks": []any{"claude-fable-5"}}} - _, err := NewClient(t.Context(), cfg, environment.NewMapEnvProvider(map[string]string{"ANTHROPIC_API_KEY": "k"})) - require.ErrorContains(t, err, "issues/4584") -} - func TestParseFallbacks(t *testing.T) { t.Parallel() @@ -181,3 +166,58 @@ func TestNewClient_FallbacksCheckEnvBaseURL(t *testing.T) { _, err = NewClient(t.Context(), cfg, env) require.NoError(t, err, "an explicit base_url takes precedence over the environment") } + +// The request must be valid for every listed model: a Claude 5.x fallback +// can't take the sampling settings or token budget an older primary sends. +func TestValidateFallbackRequestShape(t *testing.T) { + t.Parallel() + + for name, tc := range map[string]struct { + cfg latest.ModelConfig + wantErr string + }{ + "temperature": { + cfg: latest.ModelConfig{Model: "claude-opus-4-6", Temperature: new(0.7)}, + wantErr: "sends temperature or top_p", + }, + "top_k": { + cfg: latest.ModelConfig{Model: "claude-opus-4-6", ProviderOpts: map[string]any{"top_k": 5}}, + wantErr: "sends top_k", + }, + // With thinking on, the request builder doesn't send temperature. + "temperature with thinking on": { + cfg: latest.ModelConfig{Model: "claude-opus-4-6", Temperature: new(0.7), ThinkingBudget: &latest.ThinkingBudget{Effort: "high"}}, + }, + // Sonnet 4.5 has no adaptive thinking: its effort level goes out as a + // token budget, which a Claude 5.x fallback refuses. + "effort sent as tokens": { + cfg: latest.ModelConfig{Model: "claude-sonnet-4-5", ThinkingBudget: &latest.ThinkingBudget{Effort: "high"}}, + wantErr: "thinks with a token budget", + }, + "token budget": { + cfg: latest.ModelConfig{Model: "claude-sonnet-4-5", ThinkingBudget: &latest.ThinkingBudget{Tokens: 8000}}, + wantErr: "thinks with a token budget", + }, + "effort": {cfg: latest.ModelConfig{Model: "claude-opus-4-6", ThinkingBudget: &latest.ThinkingBudget{Effort: "high"}}}, + "no sampling": {cfg: latest.ModelConfig{Model: "claude-opus-4-6"}}, + // The primary already refuses them, so they're never sent. + "both refuse sampling": {cfg: latest.ModelConfig{Model: "claude-fable-5-1", Temperature: new(0.7)}}, + "both refuse token budget": {cfg: latest.ModelConfig{Model: "claude-opus-4-6", ThinkingBudget: &latest.ThinkingBudget{Tokens: 8000}}}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + cfg := tc.cfg + cfg.Provider = "anthropic" + if cfg.ProviderOpts == nil { + cfg.ProviderOpts = map[string]any{} + } + cfg.ProviderOpts["fallbacks"] = []any{"claude-sonnet-5-5"} + err := validateFallbackSpec(&cfg) + if tc.wantErr == "" { + require.NoError(t, err) + return + } + require.ErrorContains(t, err, tc.wantErr) + }) + } +} diff --git a/pkg/model/provider/anthropic/features_integration_test.go b/pkg/model/provider/anthropic/features_integration_test.go index a82c8351a..ba35cf744 100644 --- a/pkg/model/provider/anthropic/features_integration_test.go +++ b/pkg/model/provider/anthropic/features_integration_test.go @@ -436,9 +436,7 @@ func TestThinkingDisplayUpdates(t *testing.T) { for name, cfg := range map[string]*latest.ModelConfig{ "sonnet 5": {Provider: "anthropic", Model: "claude-sonnet-5", ProviderOpts: map[string]any{"thinking_display": "updates"}}, "mythos 5": {Provider: "anthropic", Model: "claude-mythos-5", ProviderOpts: map[string]any{"thinking_display": "updates"}}, - // Fallback chains with a Claude 5.x model are rejected earlier (#4584), - // so the display check is exercised with IDs outside the 5.x table. - "fallback": {Provider: "anthropic", Model: "claude-fable-5-2", ProviderOpts: map[string]any{ + "fallback": {Provider: "anthropic", Model: "claude-fable-5-1", ProviderOpts: map[string]any{ "thinking_display": "updates", "fallbacks": []any{"claude-opus-4-8"}, }}, } { @@ -448,7 +446,7 @@ func TestThinkingDisplayUpdates(t *testing.T) { _, err := NewClient(t.Context(), &latest.ModelConfig{Provider: "anthropic", Model: "claude-fable-5-1", ProviderOpts: map[string]any{ "thinking_display": "updates", "fallbacks": []any{"claude-mythos-5-1"}, }}, env) - require.ErrorContains(t, err, "issues/4584", "a 5.x fallback chain is rejected before the display check") + require.NoError(t, err, "every model in the chain supports updates") ok, err := NewClient(t.Context(), &latest.ModelConfig{Provider: "anthropic", Model: "claude-fable-5-2", ProviderOpts: map[string]any{ "thinking_display": "updates", "fallbacks": []any{"claude-fable-5-3"}, }}, env) diff --git a/pkg/model/provider/anthropic/server_fallback_test.go b/pkg/model/provider/anthropic/server_fallback_test.go index aab311534..cb84e5111 100644 --- a/pkg/model/provider/anthropic/server_fallback_test.go +++ b/pkg/model/provider/anthropic/server_fallback_test.go @@ -56,7 +56,7 @@ func TestServerFallbackStream(t *testing.T) { served string }{ "before output": { - reply: anthropictest.FallbackBeforeOutputReply("m", from, to, "hi"), + reply: anthropictest.FallbackBeforeOutputReply("m", from, to, "cyber", "hi"), markers: []chat.ServerFallback{{From: from, To: to}}, served: to, }, "mid output": { @@ -336,3 +336,91 @@ func TestServedByWhileStreaming(t *testing.T) { } } } + +// finalAttempts drains a stream and returns the attempts it reported. +func finalAttempts(t *testing.T, stream chat.MessageStream) []chat.UsageAttempt { + t.Helper() + var attempts []chat.UsageAttempt + for { + resp, err := stream.Recv() + if errors.Is(err, io.EOF) { + return attempts + } + require.NoError(t, err) + if resp.UsageAttempts != nil { + attempts = resp.UsageAttempts + } + } +} + +// Each attempt is billed at its own model; a declined attempt that produced +// no output is billed only in the categories Anthropic bills. +func TestServerFallbackAttempts(t *testing.T) { + t.Parallel() + + const from, to = "claude-opus-4-8", "claude-sonnet-4-6" + type attempt struct { + model string + billed bool + } + for name, tc := range map[string]struct { + reply string + want []attempt + }{ + "mid output": { + reply: anthropictest.FallbackMidOutputReply("m", from, to, "before", "after", "t1"), + want: []attempt{{from, true}, {to, true}}, + }, + "before output, unbilled category": { + reply: anthropictest.FallbackBeforeOutputReply("m", from, to, "cyber", "hi"), + want: []attempt{{from, false}, {to, true}}, + }, + "before output, billed category": { + reply: anthropictest.FallbackBeforeOutputReply("m", from, to, "bio", "hi"), + want: []attempt{{from, true}, {to, true}}, + }, + "two hops": { + reply: anthropictest.FallbackChainReply("m", from, "claude-opus-4-7", to), + want: []attempt{{from, true}, {"claude-opus-4-7", true}, {to, true}}, + }, + "sticky routing": { + reply: anthropictest.StickyReply("m", to, "hi"), + want: []attempt{{to, true}}, + }, + // Without a marker for the declined hop, its category is unknown: bill it. + "marker missing": { + reply: func() string { + r := anthropictest.FallbackBeforeOutputReply("m", from, to, "cyber", "hi") + start := strings.Index(r, "event: content_block_start") + end := strings.Index(r, "event: content_block_stop") + end += strings.Index(r[end:], "\n\n") + 2 + return r[:start] + r[end:] + }(), + want: []attempt{{from, true}, {to, true}}, + }, + "no fallback": {reply: anthropictest.TextReply("m", "hi", 10)}, + "other iterations only": { + reply: strings.Replace(anthropictest.TextReply("m", "hi", 10), `"output_tokens":7`, + `"output_tokens":7,"iterations":[{"type":"compaction","model":"`+from+`","input_tokens":5,"output_tokens":1}]`, 1), + }, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + stream, err := fallbackClient(newScriptedServer(t, tc.reply)).CreateChatCompletionStream(t.Context(), []chat.Message{user("q")}, nil) + require.NoError(t, err) + defer stream.Close() + attempts := finalAttempts(t, stream) + var got []attempt + for _, a := range attempts { + got = append(got, attempt{a.Model, a.Billed}) + // The scripted iterations report 10 fresh, 3 cache-read and 4 + // cache-write input tokens, and 7 output tokens. + assert.Equal(t, chat.Usage{InputTokens: 10, CachedInputTokens: 3, CacheWriteTokens: 4, OutputTokens: a.OutputTokens}, *a.Usage()) + } + assert.Equal(t, tc.want, got) + for i, a := range attempts { + assert.Equal(t, i == 0 && a.Model == from, a.Dispatched, "only the requested model's attempt is dispatched") + } + }) + } +} diff --git a/pkg/model/provider/anthropic/thinking_binding.go b/pkg/model/provider/anthropic/thinking_binding.go index 0368b9b31..7964f8122 100644 --- a/pkg/model/provider/anthropic/thinking_binding.go +++ b/pkg/model/provider/anthropic/thinking_binding.go @@ -10,11 +10,6 @@ import ( ) func validateThinkingOptions(cfg *latest.ModelConfig) error { - // The temporary Claude 5.x block goes first: it covers every rule - // validateFallbackSpec has for Claude 5.x models. - if err := validateFallbacks(cfg); err != nil { - return err - } if err := validateFallbackSpec(cfg); err != nil { return err } diff --git a/pkg/model/provider/providers/fallback_requests_test.go b/pkg/model/provider/providers/fallback_requests_test.go index b8f61f73b..8e248c2c9 100644 --- a/pkg/model/provider/providers/fallback_requests_test.go +++ b/pkg/model/provider/providers/fallback_requests_test.go @@ -74,3 +74,20 @@ func TestServerFallbackKeepsLegacyRequestShape(t *testing.T) { assert.Nil(t, req.body["temperature"], "no sampling with thinking on") }) } + +// A Claude 5.x chain goes out as configured: the server, not the client, +// adapts the request for each fallback attempt. +func TestServerFallbackClaude5Request(t *testing.T) { + t.Parallel() + + req := captureFinalRequest(t, latest.ModelConfig{ + Provider: "anthropic", Model: "claude-sonnet-5-5", + ProviderOpts: map[string]any{"fallbacks": []any{"claude-sonnet-5"}}, + }, map[string]string{"ANTHROPIC_API_KEY": "test-key"}) + + assert.Equal(t, "beta=true", req.query) + assert.Contains(t, strings.Join(req.header.Values("anthropic-beta"), ","), "server-side-fallback-2026-07-01") + assert.Equal(t, []any{map[string]any{"model": "claude-sonnet-5"}}, req.body["fallbacks"]) + assert.Equal(t, "drop_block", field(req.body, "thinking", "block_binding", "prefix_mismatch_behavior")) + assert.Equal(t, "adaptive", field(req.body, "thinking", "type"), "no client-side rewrite for the fallback") +} diff --git a/pkg/runtime/attempt_cost.go b/pkg/runtime/attempt_cost.go new file mode 100644 index 000000000..0c83dad0f --- /dev/null +++ b/pkg/runtime/attempt_cost.go @@ -0,0 +1,40 @@ +package runtime + +import ( + "context" + + "github.com/docker/docker-agent/pkg/chat" + "github.com/docker/docker-agent/pkg/model/provider/base" +) + +// attemptsCost prices a response a server fallback spread across models: each +// billed attempt at its own model's rates, never the top-level usage, which +// only repeats the last attempt. The dispatched attempt keeps its config, +// display alias and cost override; other models are priced from the +// catalogue. Nil when any billed attempt cannot be priced. +func attemptsCost(ctx context.Context, attempts []chat.UsageAttempt, dispatched base.Config, store ModelStore) *float64 { + var total float64 + for _, attempt := range attempts { + if !attempt.Billed { + continue + } + cfg := dispatched + if !attempt.Dispatched { + cfg.ModelConfig.Model = attempt.Model + cfg.ModelConfig.DisplayModel = "" + cfg.ModelConfig.Cost = nil + } + id := cfg.ID() + m, err := store.GetModel(ctx, id) + if err != nil { + m = nil + } + usage := attempt.Usage() + cost := computeMessageCost(usage, applyModelCost(m, id, usage, cfg)) + if cost == nil { + return nil + } + total += *cost + } + return &total +} diff --git a/pkg/runtime/attempt_cost_test.go b/pkg/runtime/attempt_cost_test.go new file mode 100644 index 000000000..a68732b21 --- /dev/null +++ b/pkg/runtime/attempt_cost_test.go @@ -0,0 +1,99 @@ +package runtime + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/chat" + "github.com/docker/docker-agent/pkg/config/latest" + "github.com/docker/docker-agent/pkg/model/provider" + "github.com/docker/docker-agent/pkg/model/provider/base" + "github.com/docker/docker-agent/pkg/modelsdev" +) + +// catalogStore prices each model ID from its own table; others are unknown. +type catalogStore struct { + ModelStore + + costs map[string]modelsdev.Cost +} + +func (s catalogStore) GetModel(_ context.Context, id modelsdev.ID) (*modelsdev.Model, error) { + c, ok := s.costs[id.String()] + if !ok { + return nil, errors.New("not in catalogue") + } + return &modelsdev.Model{Cost: &c}, nil +} + +func TestAttemptsCost(t *testing.T) { + t.Parallel() + + store := catalogStore{costs: map[string]modelsdev.Cost{ + "anthropic/claude-opus-4-8": {Input: 10, Output: 50}, + "anthropic/claude-sonnet-4-6": {Input: 3, Output: 15}, + }} + dispatched := base.Config{ModelConfig: latest.ModelConfig{Provider: "anthropic", Model: "claude-opus-4-8"}} + declined := chat.UsageAttempt{Model: "claude-opus-4-8", Billed: true, Dispatched: true, InputTokens: 1_000_000, OutputTokens: 100_000} + served := chat.UsageAttempt{Model: "claude-sonnet-4-6", Billed: true, InputTokens: 1_000_000, OutputTokens: 200_000} + + for name, tc := range map[string]struct { + attempts []chat.UsageAttempt + dispatched base.Config + want *float64 + }{ + // $10 + $5 at Opus rates, $3 + $3 at Sonnet rates. + "each attempt at its own rates": {attempts: []chat.UsageAttempt{declined, served}, dispatched: dispatched, want: new(21.0)}, + "an unbilled declined attempt costs nothing": { + attempts: []chat.UsageAttempt{{Model: "claude-opus-4-8", InputTokens: 1_000_000}, served}, + dispatched: dispatched, want: new(6.0), + }, + "an uncatalogued attempt leaves the turn unpriced": { + attempts: []chat.UsageAttempt{declined, {Model: "claude-unknown", Billed: true, InputTokens: 1}}, + dispatched: dispatched, + }, + // The API may report a resolved, dated ID for the requested model. + "the dispatched attempt keeps its config under a resolved ID": { + attempts: []chat.UsageAttempt{{Model: "claude-opus-4-8-20260901", Billed: true, Dispatched: true, InputTokens: 1_000_000, OutputTokens: 100_000}, served}, + dispatched: base.Config{ModelConfig: latest.ModelConfig{ + Provider: "anthropic", Model: "claude-opus-4-8", Cost: &latest.CostConfig{Input: 1, Output: 1}, + }}, + want: new(7.1), + }, + "the dispatched model keeps its alias and cost override": { + attempts: []chat.UsageAttempt{declined, served}, + dispatched: base.Config{ModelConfig: latest.ModelConfig{ + Provider: "anthropic", Model: "claude-opus-4-8", DisplayModel: "opus", + Cost: &latest.CostConfig{Input: 1, Output: 1}, + }}, + // $1 + $0.1 from the override, $6 at Sonnet rates. + want: new(7.1), + }, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + got := attemptsCost(t.Context(), tc.attempts, tc.dispatched, store) + if tc.want == nil { + assert.Nil(t, got) + return + } + require.NotNil(t, got) + assert.InDelta(t, *tc.want, *got, 1e-9) + }) + } +} + +// Behind a router, the requested model's attempt is priced with the leaf the +// router dispatched to, not the router's own config. +func TestDispatchedConfig(t *testing.T) { + t.Parallel() + + router := base.Config{ModelConfig: latest.ModelConfig{Provider: "anthropic", Model: "router"}} + leaf := provider.ServingModel{Config: base.Config{ModelConfig: latest.ModelConfig{Provider: "anthropic", Model: "claude-sonnet-5-5"}}} + assert.Equal(t, router, dispatchedConfig(streamResult{}, router)) + assert.Equal(t, leaf.Config, dispatchedConfig(streamResult{Dispatched: &leaf}, router)) +} diff --git a/pkg/runtime/claude5_compaction_lifecycle_test.go b/pkg/runtime/claude5_compaction_lifecycle_test.go index 0b65e3adf..3a750e2fc 100644 --- a/pkg/runtime/claude5_compaction_lifecycle_test.go +++ b/pkg/runtime/claude5_compaction_lifecycle_test.go @@ -28,12 +28,13 @@ import ( // registry, against server, with a 40k-token context window. func claude5Runtime(t *testing.T, server *anthropictest.Server, providerOpts map[string]any, store session.Store) *LocalRuntime { t.Helper() - return anthropicRuntime(t, server, "claude-sonnet-5-5", providerOpts, store) + return anthropicRuntime(t, server, "claude-sonnet-5-5", providerOpts, store, + mockModelStoreWithCostAndLimit{limit: 40_000, cost: modelsdev.Cost{Input: 1, Output: 2}}) } // anthropicRuntime runs a real Anthropic client for model, built by the -// provider registry, against server, with a 40k-token context window. -func anthropicRuntime(t *testing.T, server *anthropictest.Server, modelName string, providerOpts map[string]any, store session.Store, agentOpts ...agent.Opt) *LocalRuntime { +// provider registry, against server, priced and sized by models. +func anthropicRuntime(t *testing.T, server *anthropictest.Server, modelName string, providerOpts map[string]any, store session.Store, models ModelStore, agentOpts ...agent.Opt) *LocalRuntime { t.Helper() target, err := url.Parse(server.URL()) require.NoError(t, err) @@ -52,7 +53,7 @@ func anthropicRuntime(t *testing.T, server *anthropictest.Server, modelName stri root := agent.New("root", "You are terse.", append([]agent.Opt{agent.WithModel(model)}, agentOpts...)...) rt, err := NewLocalRuntime(t.Context(), team.New(team.WithAgents(root)), WithSessionStore(store), - WithModelStore(mockModelStoreWithCostAndLimit{limit: 40_000, cost: modelsdev.Cost{Input: 1, Output: 2}})) + WithModelStore(models)) require.NoError(t, err) t.Cleanup(func() { assert.NoError(t, rt.Close()) }) return rt diff --git a/pkg/runtime/claude5_server_fallback_test.go b/pkg/runtime/claude5_server_fallback_test.go index 00f6b0882..f7c3c113b 100644 --- a/pkg/runtime/claude5_server_fallback_test.go +++ b/pkg/runtime/claude5_server_fallback_test.go @@ -13,6 +13,7 @@ import ( "github.com/docker/docker-agent/pkg/agent" "github.com/docker/docker-agent/pkg/chat" "github.com/docker/docker-agent/pkg/model/provider/anthropic/anthropictest" + "github.com/docker/docker-agent/pkg/modelsdev" "github.com/docker/docker-agent/pkg/session" "github.com/docker/docker-agent/pkg/tools" ) @@ -67,7 +68,11 @@ func TestServerFallbackLifecycle(t *testing.T) { }}) opts := map[string]any{"fallbacks": []any{fallback}} dbPath := filepath.Join(t.TempDir(), "sessions.db") - rt := anthropicRuntime(t, server, primary, opts, openSessionStore(t, dbPath), record) + models := catalogStore{costs: map[string]modelsdev.Cost{ + "anthropic/" + primary: {Input: 10, Output: 50}, + "anthropic/" + fallback: {Input: 3, Output: 15}, + }} + rt := anthropicRuntime(t, server, primary, opts, openSessionStore(t, dbPath), models, record) sess := session.New(session.WithUserMessage("u1"), session.WithToolsApproved(true)) requireNoErrorEvents(t, drainRunStream(t, rt, sess)) @@ -78,16 +83,24 @@ func TestServerFallbackLifecycle(t *testing.T) { loaded, err := openSessionStore(t, dbPath).GetSession(t.Context(), sess.ID) require.NoError(t, err) - var models []string + var served []string + var costs []float64 for _, item := range loaded.Messages { if item.Message != nil && item.Message.Message.Role == chat.MessageRoleAssistant { - models = append(models, item.Message.Message.Model) + served = append(served, item.Message.Message.Model) + costs = append(costs, item.Message.Message.Cost) } } - assert.Equal(t, []string{"anthropic/" + fallback, "anthropic/" + fallback}, models, "the marker and sticky routing both name the serving model") + assert.Equal(t, []string{"anthropic/" + fallback, "anthropic/" + fallback}, served, "the marker and sticky routing both name the serving model") + // Each scripted attempt reports 10 input and 7 output tokens. The handoff + // turn bills the declined Opus attempt and the Sonnet one; the sticky turn + // bills Sonnet only. + require.Len(t, costs, 2) + assert.InDelta(t, (10*10+7*50+10*3+7*15)/1e6, costs[0], 1e-12) + assert.InDelta(t, (10*3+7*15)/1e6, costs[1], 1e-12) loaded.AddMessage(session.UserMessage("u3")) - reloaded := anthropicRuntime(t, server, primary, opts, openSessionStore(t, dbPath), record) + reloaded := anthropicRuntime(t, server, primary, opts, openSessionStore(t, dbPath), models, record) requireNoErrorEvents(t, drainRunStream(t, reloaded, loaded)) requests := server.Requests() diff --git a/pkg/runtime/fallback.go b/pkg/runtime/fallback.go index cff0e7f5e..0a5490158 100644 --- a/pkg/runtime/fallback.go +++ b/pkg/runtime/fallback.go @@ -11,6 +11,7 @@ import ( "github.com/docker/docker-agent/pkg/backoff" "github.com/docker/docker-agent/pkg/chat" "github.com/docker/docker-agent/pkg/model/provider" + "github.com/docker/docker-agent/pkg/model/provider/base" "github.com/docker/docker-agent/pkg/modelerrors" "github.com/docker/docker-agent/pkg/modelsdev" "github.com/docker/docker-agent/pkg/session" @@ -391,6 +392,7 @@ func (e *fallbackExecutor) execute( idleTimeout := streamIdleTimeoutFor(modelEntry.provider) res, err := handleStream(streamCtx, streamCancel, stream, a, attemptTools, sess, events, idleTimeout) res.Serving = servingModelFrom(stream) + res.Dispatched = dispatchedModelFrom(stream) streamCancel(nil) // always release the child context if err != nil { lastErr = err @@ -431,6 +433,7 @@ func (e *fallbackExecutor) execute( announceServing(stream) res, err = handleStream(streamCtx, streamCancel, stream, a, attemptTools, sess, events, idleTimeout) res.Serving = servingModelFrom(stream) + res.Dispatched = dispatchedModelFrom(stream) } streamCancel(nil) if err == nil { @@ -588,3 +591,20 @@ func servingModelFrom(stream chat.MessageStream) *provider.ServingModel { } return nil } + +// dispatchedModelFrom reports the endpoint a stream was sent to, ignoring a +// server fallback's handoff; nil when the stream doesn't say. +func dispatchedModelFrom(stream chat.MessageStream) *provider.ServingModel { + if stream, ok := stream.(provider.ServingStream); ok { + return new(stream.ServingModel()) + } + return nil +} + +// dispatchedConfig is the config that prices the requested model's attempt. +func dispatchedConfig(res streamResult, fallback base.Config) base.Config { + if res.Dispatched != nil { + return res.Dispatched.Config + } + return fallback +} diff --git a/pkg/runtime/loop.go b/pkg/runtime/loop.go index 4a30e44e6..78ee9d4e4 100644 --- a/pkg/runtime/loop.go +++ b/pkg/runtime/loop.go @@ -969,6 +969,9 @@ func (r *LocalRuntime) runTurn( // the turn cannot be priced (no usage, or a model with no pricing // table); see computeMessageCost. msgCost := computeMessageCost(res.Usage, m) + if len(res.UsageAttempts) > 0 && usedModel != nil { + msgCost = attemptsCost(ctx, res.UsageAttempts, dispatchedConfig(res, usedModel.BaseConfig()), r.modelsStore) + } if res.Usage != nil { modelName := "unknown" if m != nil { diff --git a/pkg/runtime/streaming.go b/pkg/runtime/streaming.go index 44e9c779a..78b556e01 100644 --- a/pkg/runtime/streaming.go +++ b/pkg/runtime/streaming.go @@ -75,7 +75,13 @@ type streamResult struct { Stopped bool FinishReason chat.FinishReason Usage *chat.Usage - Serving *provider.ServingModel + // UsageAttempts lists each model that ran when a server fallback spread + // the response across models; Usage describes only the last one. + UsageAttempts []chat.UsageAttempt + Serving *provider.ServingModel + // Dispatched is the endpoint the request went to (e.g. a router's leaf), + // before a server fallback handed it to another model. + Dispatched *provider.ServingModel } // handleStream reads a chat.MessageStream to completion, emitting streaming @@ -141,6 +147,7 @@ func handleStream(ctx context.Context, cancelStream context.CancelCauseFunc, str var toolCalls []tools.ToolCall var media []chat.MediaDelta var messageUsage *chat.Usage + var usageAttempts []chat.UsageAttempt var providerFinishReason chat.FinishReason var responseStarted bool @@ -264,6 +271,9 @@ mainLoop: // token counts, so the last value is the most accurate. messageUsage = response.Usage } + if len(response.UsageAttempts) > 0 { + usageAttempts = response.UsageAttempts + } if len(response.Choices) == 0 { continue @@ -416,6 +426,7 @@ mainLoop: Stopped: len(toolCalls) == 0, // stop only when there are no tool calls to execute FinishReason: finishReason, Usage: messageUsage, + UsageAttempts: usageAttempts, ResponseStarted: responseStarted, }, nil } @@ -517,6 +528,7 @@ mainLoop: Stopped: stoppedNoToolCalls, FinishReason: finishReason, Usage: messageUsage, + UsageAttempts: usageAttempts, ResponseStarted: responseStarted, }, nil }