diff --git a/internal/provider/anthropic.go b/internal/provider/anthropic.go index e25d0fd55..a9cfbbd40 100644 --- a/internal/provider/anthropic.go +++ b/internal/provider/anthropic.go @@ -26,10 +26,19 @@ type AnthropicProvider struct { cap *adaptiveCap transport *headerInjectingTransport // kept for runtime header updates calibrator *tokenCountCalibrator // periodic real-API token calibration - reasoningEffort string // "", "low", "medium", "high" — maps to thinking budget + reasoningEffort string // "", "low", "medium", "high", "xhigh", "max" — maps to thinking budget toolChoice string // "", "auto", "required", "none" — maps to Anthropic tool_choice temperature float64 // 0 = provider default topP float64 // 0 = provider default + + // Top-level effort carrier (output_config.effort, GA effort parameter). + // Cache-aware per Anthropic's 2026 effort guidance: a top-level effort + // change does not preserve cached prefixes, so the carrier is attached + // only once a level stabilizes across consecutive requests — per-turn + // adaptive-effort oscillation never touches it (see beginEffortTracking). + effortCarrier atomic.Bool // true until the endpoint rejects output_config + lastCallEffort string // effort level observed on the previous request + conversationEffort string // effort level established for the cached prefix } // ModelName returns the current model name used by this provider. @@ -37,7 +46,7 @@ func (p *AnthropicProvider) ModelName() string { return p.model } // CloneWithModel returns a shallow copy of this provider with a different model. func (p *AnthropicProvider) CloneWithModel(model string) Provider { - return &AnthropicProvider{ + clone := &AnthropicProvider{ client: p.client, model: model, maxTokens: p.maxTokens, @@ -52,11 +61,29 @@ func (p *AnthropicProvider) CloneWithModel(model string) Provider { temperature: p.temperature, topP: p.topP, } + // Inherit the endpoint capability latch (an endpoint that rejected + // output_config stays off), but reset the per-conversation stability + // window: the clone re-learns effort stabilization for its own cache + // prefix over its first two requests. + clone.effortCarrier.Store(p.effortCarrier.Load()) + return clone +} + +// SetReasoningEffort sets the reasoning effort. It maps to Anthropic's +// extended thinking budget_tokens parameter ("low" ~5K, "medium" ~16K, +// "high" ~32K) and, once the level stabilizes, to the top-level +// output_config.effort carrier ("xhigh"/"max" are carrier-only). Empty +// string disables both. +func (p *AnthropicProvider) SetReasoningEffort(effort string) { + effort = strings.ToLower(strings.TrimSpace(effort)) + switch effort { + case "", "low", "medium", "high", "xhigh", "max": + p.reasoningEffort = effort + } } -// SetReasoningEffort sets the reasoning effort, which maps to Anthropic's -// extended thinking budget_tokens parameter. Effort levels: "low" (~5K), -// "medium" (~16K), "high" (~32K). Empty string disables thinking. +func (p *AnthropicProvider) ReasoningEffort() string { return p.reasoningEffort } + // SetMaxTokens implements provider.MaxTokensSetter (#1592-A). func (p *AnthropicProvider) SetMaxTokens(n int) { if n > 0 { @@ -64,16 +91,41 @@ func (p *AnthropicProvider) SetMaxTokens(n int) { } } -func (p *AnthropicProvider) SetReasoningEffort(effort string) { - effort = strings.ToLower(strings.TrimSpace(effort)) - switch effort { - case "", "low", "medium", "high": - p.reasoningEffort = effort +// beginEffortTracking updates the cache-aware effort-carrier bookkeeping +// for this request. Returns whether buildParams should attach +// output_config.effort. +// +// Hysteresis (Anthropic effort guidance, 2026): hold the top-level effort +// constant within a conversation. A level is adopted as the carrier only +// after TWO consecutive requests at the same level — the adaptive-effort +// adapter sets a level before each call and restores the previous level +// right after, so its oscillation never stabilizes and never reaches the +// carrier (budget_tokens alone modulates per-turn thinking, which is +// cache-neutral). A user switch (/effort, config) persists across calls +// and re-establishes the carrier on the second call: one deliberate, +// one-time cache rewrite instead of a storm. +func (p *AnthropicProvider) beginEffortTracking() bool { + if !p.effortCarrier.Load() { + return false + } + effort := strings.ToLower(strings.TrimSpace(p.reasoningEffort)) + if effort == "" { + // Effort off (or adaptive restored its previous level): reset the + // stability window so oscillation can never establish a carrier. + p.lastCallEffort = "" + return false + } + if p.lastCallEffort != effort { + p.lastCallEffort = effort + return false } + if p.conversationEffort != effort { + debug.Log("anthropic", "effort carrier established: %s (top-level output_config changes restart the prompt cache)", effort) + p.conversationEffort = effort + } + return true } -func (p *AnthropicProvider) ReasoningEffort() string { return p.reasoningEffort } - // SetToolChoice sets the tool_choice parameter: "auto" (model decides), // "required" (force tool use), "none" (disable tools), or "" (API default). func (p *AnthropicProvider) SetToolChoice(choice string) { @@ -208,6 +260,38 @@ var thinkingErrorAnchors = []string{ "max_tokens must be greater than budget_tokens", } +// effortErrorAnchors are phrases from real output_config/effort rejection +// errors (Anthropic direct and gateway-stringified variants). +var effortErrorAnchors = []string{ + "output_config", // unknown-parameter and field-error shapes + "per-turn effort", + "per-message effort", + "does not support effort", + "effort is not supported", +} + +// isEffortError reports whether an API error is a genuine rejection of the +// top-level output_config effort carrier — e.g. an Anthropic-compatible +// gateway that predates the parameter. Mirrors isThinkingError: status must +// be a 4xx parameter class and the message must contain an anchored phrase. +func isEffortError(err error) bool { + if err == nil { + return false + } + if sc, ok := asStatusCode(err); ok { + if sc != 400 && sc != 404 && sc != 422 { + return false + } + } + msg := strings.ToLower(err.Error()) + for _, anchor := range effortErrorAnchors { + if strings.Contains(msg, anchor) { + return true + } + } + return false +} + // asStatusCode extracts an HTTP status code from a provider error when // the concrete type exposes one (the anthropic SDK's apierror.Error does). func asStatusCode(err error) (int, bool) { @@ -315,6 +399,7 @@ func (p *AnthropicProvider) SetSessionID(sessionID string) { func (p *AnthropicProvider) Chat(ctx context.Context, messages []Message, tools []ToolDefinition) (*ChatResponse, error) { debug.Log("anthropic", "Chat START model=%s msgs=%d tools=%d", p.model, len(messages), len(tools)) + p.beginEffortTracking() params := p.buildParams(messages, tools) var resp *anthropic.Message @@ -333,6 +418,18 @@ func (p *AnthropicProvider) Chat(ctx context.Context, messages []Message, tools return callErr }, providerRetryAttempts) } + // Retry once without the effort carrier if the endpoint rejects + // output_config (Anthropic-compatible gateways predating the parameter). + // The latch keeps effort semantics on budget_tokens for the session. + if err != nil && isEffortError(err) && p.effortCarrier.CompareAndSwap(true, false) { + debug.Log("anthropic", "Chat: retrying without output_config (endpoint rejected the effort carrier)") + params.OutputConfig = anthropic.OutputConfigParam{} + err = retryWithBackoffCtx(ctx, func() error { + var callErr error + resp, callErr = p.client.Messages.New(ctx, params) + return callErr + }, providerRetryAttempts) + } if err != nil { if rejected, parsed := maxTokensRejection(err); rejected { p.cap.OnRejected(parsed) @@ -363,6 +460,7 @@ func (p *AnthropicProvider) Chat(ctx context.Context, messages []Message, tools func (p *AnthropicProvider) ChatStream(ctx context.Context, messages []Message, tools []ToolDefinition) (<-chan StreamEvent, error) { debug.Log("anthropic", "ChatStream START model=%s msgs=%d tools=%d", p.model, len(messages), len(tools)) + p.beginEffortTracking() params := p.buildParams(messages, tools) ch := make(chan StreamEvent, 64) @@ -545,6 +643,14 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, messages []Message, retry = true return } + // Retry without the effort carrier if the endpoint rejects it. + if !emitted && isEffortError(err) && p.effortCarrier.CompareAndSwap(true, false) { + debug.Log("anthropic", "Stream: retrying without output_config (endpoint rejected the effort carrier)") + ch <- StreamEvent{Type: StreamEventSystem, Text: "[Endpoint rejected output_config effort - retrying without it] "} + params.OutputConfig = anthropic.OutputConfigParam{} + retry = true + return + } // Retry if no content has been emitted yet and the error is retryable. if !emitted && isRetryableForContext(ctx, err) && attempt < providerRetryAttempts-1 { // Notify user about retry @@ -990,6 +1096,21 @@ func (p *AnthropicProvider) buildParams(messages []Message, tools []ToolDefiniti params.Thinking = anthropic.ThinkingConfigParamOfEnabled(budget) } + // Effort via the top-level output_config (GA effort parameter, 2026): + // attach only the established conversation level (see beginEffortTracking) + // so the request prefix stays constant across calls. Composed with + // budget_tokens this is the documented best practice for models that + // support effort alongside extended thinking: effort governs total token + // volume, the budget caps explicit thinking. The per-message + // output_config marker (cache-preserving mid-conversation switching) is + // not expressible in SDK v1.68 typed params; when the SDK gains it, flip + // re-establishments to the marker form. + if p.effortCarrier.Load() && p.conversationEffort != "" && p.lastCallEffort == p.conversationEffort { + params.OutputConfig = anthropic.OutputConfigParam{ + Effort: anthropic.OutputConfigEffort(p.conversationEffort), + } + } + if len(tools) > 0 { toolParams := make([]anthropic.ToolUnionParam, len(tools)) for i, t := range tools { diff --git a/internal/provider/zz_effort_carrier_test.go b/internal/provider/zz_effort_carrier_test.go new file mode 100644 index 000000000..ca8f3b2d6 --- /dev/null +++ b/internal/provider/zz_effort_carrier_test.go @@ -0,0 +1,213 @@ +package provider + +// Effort carrier (top-level output_config.effort) tests. +// +// Cache contract (Anthropic effort guidance, 2026): a top-level effort +// change does not preserve cached prefixes, so the carrier must only be +// attached for user-established, stable effort levels — never for the +// per-turn oscillation produced by the adaptive-effort adapter. + +import ( + "errors" + "testing" + + anthropic "github.com/anthropics/anthropic-sdk-go" +) + +// newCarrierProvider returns a provider with the carrier enabled and no +// stability history. +func newCarrierProvider() *AnthropicProvider { + p := &AnthropicProvider{maxTokens: 64000} + p.SetReasoningEffort("high") + p.effortCarrier.Store(true) + return p +} + +func TestSetReasoningEffortAcceptsXhighAndMax(t *testing.T) { + p := &AnthropicProvider{maxTokens: 64000} + for _, level := range []string{"xhigh", "max", "XHIGH", " High "} { + p.SetReasoningEffort(level) + if p.ReasoningEffort() != normalizeEffortForTest(level) { + t.Errorf("SetReasoningEffort(%q) = %q, want %q", level, p.ReasoningEffort(), normalizeEffortForTest(level)) + } + } + // Invalid levels are still ignored. + p.SetReasoningEffort("turbo") + if p.ReasoningEffort() != "high" { + t.Errorf("invalid effort overwrote existing level: %q", p.ReasoningEffort()) + } +} + +func normalizeEffortForTest(s string) string { + return map[string]string{"xhigh": "xhigh", "max": "max", "XHIGH": "xhigh", " High ": "high"}[s] +} + +func TestEffortCarrierHysteresis(t *testing.T) { + t.Run("disabled by default", func(t *testing.T) { + p := &AnthropicProvider{maxTokens: 64000} + p.SetReasoningEffort("high") + for i := 0; i < 3; i++ { + if p.beginEffortTracking() { + t.Fatalf("call %d: carrier must stay off when disabled", i) + } + } + }) + + t.Run("user-established level stabilizes on second call", func(t *testing.T) { + p := newCarrierProvider() + if p.beginEffortTracking() { + t.Fatal("first call must not attach the carrier") + } + if !p.beginEffortTracking() { + t.Fatal("second consecutive call at the same level must attach the carrier") + } + if !p.beginEffortTracking() { + t.Fatal("third call must keep attaching the carrier") + } + }) + + t.Run("constant applied+restored level stabilizes", func(t *testing.T) { + // Adaptive-effort pattern with base "": apply high before each call, + // restore "" after. Every REQUEST is at high, so the top-level + // carrier is constant across requests — attaching is cache-correct. + p := newCarrierProvider() + for i := 0; i < 4; i++ { + p.SetReasoningEffort("high") + attached := p.beginEffortTracking() + if i == 0 && attached { + t.Fatal("first request must open the window, not attach") + } + if i > 0 && !attached { + t.Fatalf("round %d: constant request level must attach", i) + } + p.SetReasoningEffort("") + } + }) + + t.Run("alternating request levels never stabilize", func(t *testing.T) { + p := newCarrierProvider() + for i := 0; i < 10; i++ { + level := "high" + if i%2 == 1 { + level = "low" + } + p.SetReasoningEffort(level) + if p.beginEffortTracking() { + t.Fatalf("round %d (%s): alternating levels attached the carrier", i, level) + } + } + }) + + t.Run("user switch re-establishes after one suppressed call", func(t *testing.T) { + p := newCarrierProvider() + p.beginEffortTracking() + p.beginEffortTracking() // established at high + p.SetReasoningEffort("medium") + if p.beginEffortTracking() { + t.Fatal("first call after a switch must not attach the new level") + } + if !p.beginEffortTracking() { + t.Fatal("second call after a switch must re-establish the carrier") + } + }) + + t.Run("rejection latch disables permanently", func(t *testing.T) { + p := newCarrierProvider() + p.beginEffortTracking() + if !p.effortCarrier.CompareAndSwap(true, false) { + t.Fatal("latch pre-state must be true") + } + if p.beginEffortTracking() { + t.Fatal("carrier must stay off after the endpoint rejected output_config") + } + }) +} + +func TestBuildParamsOutputConfigCarrier(t *testing.T) { + t.Run("attached for established level", func(t *testing.T) { + p := newCarrierProvider() + p.beginEffortTracking() + p.beginEffortTracking() + params := p.buildParams(nil, nil) + if params.OutputConfig.Effort != anthropic.OutputConfigEffortHigh { + t.Errorf("OutputConfig.Effort = %q, want %q", params.OutputConfig.Effort, anthropic.OutputConfigEffortHigh) + } + }) + + t.Run("suppressed during stability window", func(t *testing.T) { + p := newCarrierProvider() + p.beginEffortTracking() // first call: window opens + params := p.buildParams(nil, nil) + if params.OutputConfig.Effort != "" { + t.Errorf("first call attached carrier: %q", params.OutputConfig.Effort) + } + }) + + t.Run("suppressed when disabled", func(t *testing.T) { + p := &AnthropicProvider{maxTokens: 64000} + p.SetReasoningEffort("high") + p.lastCallEffort = "high" + p.conversationEffort = "high" + params := p.buildParams(nil, nil) + if params.OutputConfig.Effort != "" { + t.Errorf("disabled provider attached carrier: %q", params.OutputConfig.Effort) + } + }) +} + +func TestIsEffortError(t *testing.T) { + trueCases := []struct { + name string + err error + }{ + {"unknown parameter", errors.New("[400] unexpected field output_config.effort")}, + {"gateway unknown param", errors.New("output_config is not a valid parameter")}, + {"per-turn rejection", errors.New("[400] per-turn effort is not accepted by this endpoint")}, + {"no status, anchored", errors.New("output_config not supported by this provider")}, + {"typed 422", statusErr{422, "output_config: unrecognized request field"}}, + } + for _, tc := range trueCases { + if !isEffortError(tc.err) { + t.Errorf("%s: expected true, got false", tc.name) + } + } + falseCases := []struct { + name string + err error + }{ + // Bare errors carry no extractable status, so the anchored-phrase + // fallback applies (same design as isThinkingError); the negative + // cases that matter are typed-status errors, which production SDK + // errors always are. + {"typed 500 with anchor", statusErr{500, "output_config internal error"}}, + {"typed 429 quota", statusErr{429, "output_config quota exceeded"}}, + {"unrelated 400", errors.New("[400] messages: field required")}, + {"nil", nil}, + } + for _, tc := range falseCases { + if isEffortError(tc.err) { + t.Errorf("%s: expected false, got true", tc.name) + } + } +} + +func TestCloneWithModelInheritsLatchResetsWindow(t *testing.T) { + parent := newCarrierProvider() + parent.beginEffortTracking() + parent.beginEffortTracking() // established at high + + clone := parent.CloneWithModel("claude-sonnet-4-5").(*AnthropicProvider) + if !clone.effortCarrier.Load() { + t.Fatal("clone must inherit the enabled latch") + } + if clone.lastCallEffort != "" || clone.conversationEffort != "" { + t.Fatalf("clone must reset the stability window, got last=%q conv=%q", clone.lastCallEffort, clone.conversationEffort) + } + + // A rejected-output_config endpoint keeps the latch off across clones. + parent.effortCarrier.Store(false) + clone2 := parent.CloneWithModel("claude-opus-4-6").(*AnthropicProvider) + if clone2.effortCarrier.Load() { + t.Fatal("clone must inherit the disabled latch") + } +}