| 1 | // Package anthropic implements the Anthropic Messages API provider (POST |
| 2 | // /v1/messages, SSE streaming) with a hand-written net/http client — no SDK. It |
| 3 | // self-registers under the "anthropic" kind, so any Claude model is a config |
| 4 | // instance rather than code. |
| 5 | // |
| 6 | // Two notes, both rooted in the transport-agnostic provider.Message abstraction: |
| 7 | // |
| 8 | // - Extended thinking is opt-in (provider config thinking="adaptive"). Anthropic |
| 9 | // requires the *signed* thinking block be replayed on the next turn when a tool |
| 10 | // call followed thinking, so Message carries ReasoningSignature alongside |
| 11 | // ReasoningContent and this provider replays the signed block on the next |
| 12 | // request. DeepSeek's Anthropic endpoint instead uses unsigned thinking blocks, |
| 13 | // thinking.type enabled|disabled, and output_config.effort; requests carrying |
| 14 | // tools must replay all provider reasoning. Some other compatible gateways such |
| 15 | // as LongCat use the binary toggle without output_config. (redacted_thinking |
| 16 | // blocks are not yet captured/replayed.) |
| 17 | // - Native Anthropic requests omit temperature/top_p. Current Claude models |
| 18 | // (Opus 4.8/4.7) reject sampling parameters with a 400; Anthropic steers |
| 19 | // behavior via prompting instead. DeepSeek's compatible endpoint accepts the |
| 20 | // caller's temperature, so that field is preserved only for DeepSeek. |
| 21 | package anthropic |
| 22 | |
| 23 | import ( |
| 24 | "bufio" |
| 25 | "bytes" |
| 26 | "context" |
| 27 | "encoding/json" |
| 28 | "fmt" |
| 29 | "io" |
| 30 | "net/http" |
| 31 | "strings" |
| 32 | "sync" |
| 33 | "sync/atomic" |
| 34 | "time" |
| 35 | |
| 36 | "reasonix/internal/netclient" |
| 37 | "reasonix/internal/provider" |
| 38 | "reasonix/internal/provider/openai" |
| 39 | ) |
| 40 | |
| 41 | // defaultStreamIdleTimeout caps how long a started SSE stream may go silent before |
| 42 | // it's treated as a dropped connection — a half-open TCP connection (proxy switched |
| 43 | // mid-stream) sends no RST, so scanner.Scan() would block forever. Generous on |
| 44 | // purpose; live streams emit far more often. Stored per-client (client.idleTimeout) |
| 45 | // so a test can shorten it without a shared global that races other watchdogs. |
| 46 | const defaultStreamIdleTimeout = 120 * time.Second |
| 47 | |
| 48 | const ( |
| 49 | // anthropicVersion is the required API version header value. |
| 50 | anthropicVersion = "2023-06-01" |
| 51 | // defaultBaseURL is the first-party endpoint; config may override it (e.g. a |
| 52 | // gateway). Bedrock/Vertex use a different request shape and are out of scope. |
| 53 | defaultBaseURL = "https://api.anthropic.com" |
| 54 | // defaultMaxTokens is the output ceiling used when neither the provider config |
| 55 | // nor the request supplies one. Anthropic requires max_tokens, so unlike the |
| 56 | // optional OpenAI-compatible budget it cannot be omitted. |
| 57 | defaultMaxTokens = 32768 |
| 58 | ) |
| 59 | |
| 60 | func init() { |
| 61 | provider.Register("anthropic", New) |
| 62 | } |
| 63 | |
| 64 | // New builds an Anthropic provider from a resolved config. |
| 65 | func New(cfg provider.Config) (provider.Provider, error) { |
| 66 | if cfg.Model == "" { |
| 67 | return nil, fmt.Errorf("anthropic: model is required for provider %q", cfg.Name) |
| 68 | } |
| 69 | name := cfg.Name |
| 70 | if name == "" { |
| 71 | name = "anthropic" |
| 72 | } |
| 73 | baseURL := cfg.BaseURL |
| 74 | if baseURL == "" { |
| 75 | baseURL = defaultBaseURL |
| 76 | } |
| 77 | keyEnv, _ := cfg.Extra["api_key_env"].(string) // for actionable auth errors |
| 78 | keySource, _ := cfg.Extra["api_key_source"].(string) |
| 79 | thinking, _ := cfg.Extra["thinking"].(string) |
| 80 | thinking = strings.ToLower(strings.TrimSpace(thinking)) |
| 81 | effort, _ := cfg.Extra["effort"].(string) |
| 82 | effort = strings.ToLower(strings.TrimSpace(effort)) |
| 83 | vision, _ := cfg.Extra["vision"].(bool) |
| 84 | webSearch, _ := cfg.Extra["web_search"].(bool) |
| 85 | headers, _ := cfg.Extra["headers"].(map[string]string) |
| 86 | authHeader, _ := cfg.Extra["auth_header"].(bool) |
| 87 | maxOutputTokens, _ := cfg.Extra["max_output_tokens"].(int) |
| 88 | if maxOutputTokens <= 0 { |
| 89 | // Messages requires max_tokens, so an optional-budget disable request |
| 90 | // falls back to the provider's stable mandatory default. |
| 91 | maxOutputTokens = defaultMaxTokens |
| 92 | } |
| 93 | httpClient, err := newHTTPClient(cfg) |
| 94 | if err != nil { |
| 95 | return nil, fmt.Errorf("anthropic: network: %w", err) |
| 96 | } |
| 97 | // Anthropic's API surface is at {root}/v1/messages, so c.baseURL stores |
| 98 | // the *root* — without any trailing /v1. The setup wizard, however, lets |
| 99 | // users paste a full OpenAI-compatible URL (e.g. |
| 100 | // "https://proxy.example.com/v1") because that's what /models probes |
| 101 | // expect. Stripping the trailing /v1 here makes both forms land on the |
| 102 | // same endpoint without forcing users to remember Anthropic's quirky |
| 103 | // root-vs-versioned split. Without this, a user pasting |
| 104 | // "https://proxy.example.com/v1" would probe /v1/models successfully |
| 105 | // but get the chat client concatenating onto |
| 106 | // "https://proxy.example.com/v1/v1/messages" — a 404. |
| 107 | root := strings.TrimRight(baseURL, "/") |
| 108 | root = strings.TrimSuffix(root, "/v1") |
| 109 | if root == "" { |
| 110 | root = defaultBaseURL |
| 111 | } |
| 112 | return &client{ |
| 113 | name: name, |
| 114 | apiKey: cfg.APIKey, |
| 115 | keyEnv: keyEnv, |
| 116 | keySource: keySource, |
| 117 | baseURL: root, |
| 118 | model: cfg.Model, |
| 119 | nativeAnthropic: strings.EqualFold(root, defaultBaseURL), |
| 120 | deepseek: openai.IsDeepSeek(root), |
| 121 | thinking: thinking, |
| 122 | effort: effort, |
| 123 | vision: vision, |
| 124 | mimo: provider.IsMiMoEndpoint(root), |
| 125 | webSearch: webSearch, |
| 126 | headers: cleanCustomHeaders(headers), |
| 127 | authHeader: authHeader, |
| 128 | defaultMaxTokens: maxOutputTokens, |
| 129 | http: httpClient, // no overall timeout; lifecycle is ctx-driven |
| 130 | idleTimeout: defaultStreamIdleTimeout, |
| 131 | }, nil |
| 132 | } |
| 133 | |
| 134 | func newHTTPClient(cfg provider.Config) (*http.Client, error) { |
| 135 | spec, _ := cfg.Extra["proxy_spec"].(netclient.ProxySpec) |
| 136 | return netclient.NewHTTPClient(spec, netclient.TransportOptions{}) |
| 137 | } |
| 138 | |
| 139 | type client struct { |
| 140 | name string |
| 141 | apiKey string |
| 142 | keyEnv string // api_key_env name, surfaced in auth errors |
| 143 | keySource string // source of keyEnv, surfaced in auth errors |
| 144 | baseURL string |
| 145 | model string |
| 146 | nativeAnthropic bool // first-party endpoint: documented default-5m cache-write pricing applies |
| 147 | deepseek bool // official DeepSeek Anthropic endpoint: unsigned reasoning replay + automatic cache |
| 148 | thinking string // "adaptive" enables extended thinking; "" = off (config-driven) |
| 149 | effort string // output_config.effort: low|medium|high|xhigh|max; "" = provider default |
| 150 | vision bool // model accepts image input — embed attached images as base64 image blocks |
| 151 | mimo bool // true for MiMo — upgrades legacy tuple schemas to Draft 2020-12 |
| 152 | webSearch bool // enable server-side web_search tool (DeepSeek Anthropic API) |
| 153 | headers map[string]string |
| 154 | authHeader bool // send Authorization: Bearer instead of Anthropic's x-api-key header |
| 155 | defaultMaxTokens int |
| 156 | http *http.Client |
| 157 | idleTimeout time.Duration // SSE stall watchdog window; defaultStreamIdleTimeout unless a test overrides |
| 158 | authed atomic.Bool // a request has succeeded — gate transient-401 retry |
| 159 | } |
| 160 | |
| 161 | func (c *client) Name() string { return c.name } |
| 162 | |
| 163 | func (c *client) deepSeekThinkingEnabled() bool { |
| 164 | return c != nil && c.deepseek && c.thinking != "disabled" && c.effort != "disabled" |
| 165 | } |
| 166 | |
| 167 | // deepSeekAnthropicUsesProEffortMapping mirrors DeepSeek's model routing for the |
| 168 | // Anthropic endpoint. Opus aliases route to V4 Pro; Sonnet/Haiku aliases and |
| 169 | // unsupported model names route to V4 Flash. |
| 170 | func deepSeekAnthropicUsesProEffortMapping(model string) bool { |
| 171 | model = strings.ToLower(strings.TrimSpace(model)) |
| 172 | return model == "deepseek-v4-pro" || strings.HasPrefix(model, "claude-opus") |
| 173 | } |
| 174 | |
| 175 | func normalizeDeepSeekAnthropicEffort(model, effort string) string { |
| 176 | switch effort { |
| 177 | case "low": |
| 178 | if deepSeekAnthropicUsesProEffortMapping(model) { |
| 179 | return "high" |
| 180 | } |
| 181 | return "low" |
| 182 | case "medium": |
| 183 | return "high" |
| 184 | case "xhigh": |
| 185 | if deepSeekAnthropicUsesProEffortMapping(model) { |
| 186 | return "max" |
| 187 | } |
| 188 | return "high" |
| 189 | case "high", "max": |
| 190 | return effort |
| 191 | default: |
| 192 | return "" |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | func (c *client) RequiresToolCallReasoning() bool { |
| 197 | return c.deepSeekThinkingEnabled() |
| 198 | } |
| 199 | |
| 200 | func (c *client) MissingToolCallReasoningWarningIdentity() string { |
| 201 | if c == nil { |
| 202 | return "" |
| 203 | } |
| 204 | protocol := "anthropic" |
| 205 | if c.deepseek { |
| 206 | protocol = "deepseek-anthropic" |
| 207 | } |
| 208 | return strings.Join([]string{ |
| 209 | "anthropic", strings.TrimSpace(c.name), strings.TrimSpace(c.baseURL), |
| 210 | strings.TrimSpace(c.model), protocol, strings.TrimSpace(c.thinking), strings.TrimSpace(c.effort), |
| 211 | }, "\x00") |
| 212 | } |
| 213 | |
| 214 | func (c *client) sendOpts() provider.SendOptions { |
| 215 | return provider.SendOptions{ |
| 216 | Provider: c.name, |
| 217 | KeyEnv: c.keyEnv, |
| 218 | KeySource: c.keySource, |
| 219 | KeyPresent: c.apiKey != "", |
| 220 | RetryAuth: c.authed.Load(), |
| 221 | } |
| 222 | } |
| 223 | |
| 224 | func cleanCustomHeaders(in map[string]string) map[string]string { |
| 225 | if len(in) == 0 { |
| 226 | return nil |
| 227 | } |
| 228 | out := make(map[string]string, len(in)) |
| 229 | for name, value := range in { |
| 230 | name = strings.TrimSpace(name) |
| 231 | if name == "" || reservedCustomHeader(name) { |
| 232 | continue |
| 233 | } |
| 234 | out[name] = strings.TrimSpace(value) |
| 235 | } |
| 236 | if len(out) == 0 { |
| 237 | return nil |
| 238 | } |
| 239 | return out |
| 240 | } |
| 241 | |
| 242 | func reservedCustomHeader(name string) bool { |
| 243 | switch strings.ToLower(strings.TrimSpace(name)) { |
| 244 | case "content-type", "accept", "x-api-key", "authorization", "anthropic-version": |
| 245 | return true |
| 246 | default: |
| 247 | return false |
| 248 | } |
| 249 | } |
| 250 | |
| 251 | func applyCustomHeaders(h http.Header, headers map[string]string) { |
| 252 | for name, value := range cleanCustomHeaders(headers) { |
| 253 | h.Set(name, value) |
| 254 | } |
| 255 | } |
| 256 | |
| 257 | // bufPool reuses byte buffers for JSON-marshalled request bodies, reducing GC |
| 258 | // churn from repeated alloc/free of ~10-100KB buffers per turn. |
| 259 | var bufPool = sync.Pool{ |
| 260 | New: func() any { return new(bytes.Buffer) }, |
| 261 | } |
| 262 | |
| 263 | func (c *client) Stream(ctx context.Context, req provider.Request) (<-chan provider.Chunk, error) { |
| 264 | requestCtx := provider.WithRequestAttemptCounter(ctx) |
| 265 | buf := bufPool.Get().(*bytes.Buffer) |
| 266 | buf.Reset() |
| 267 | if err := json.NewEncoder(buf).Encode(c.buildRequest(requestCtx, req)); err != nil { |
| 268 | bufPool.Put(buf) |
| 269 | return nil, fmt.Errorf("%s: marshal request: %w", c.name, err) |
| 270 | } |
| 271 | body := make([]byte, buf.Len()) |
| 272 | copy(body, buf.Bytes()) |
| 273 | bufPool.Put(buf) |
| 274 | |
| 275 | newReq := func(ctx context.Context) (*http.Request, error) { |
| 276 | httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/v1/messages", bytes.NewReader(body)) |
| 277 | if err != nil { |
| 278 | return nil, err |
| 279 | } |
| 280 | httpReq.Header.Set("Content-Type", "application/json") |
| 281 | httpReq.Header.Set("Accept", "text/event-stream") |
| 282 | if c.authHeader { |
| 283 | httpReq.Header.Set("Authorization", "Bearer "+c.apiKey) |
| 284 | } else { |
| 285 | httpReq.Header.Set("x-api-key", c.apiKey) |
| 286 | } |
| 287 | httpReq.Header.Set("anthropic-version", anthropicVersion) |
| 288 | applyCustomHeaders(httpReq.Header, c.headers) |
| 289 | return httpReq, nil |
| 290 | } |
| 291 | resp, err := provider.SendWithRetry(requestCtx, c.http, c.sendOpts(), newReq) |
| 292 | if err != nil { |
| 293 | return nil, provider.AnnotateToolSchemaError(err, req.Tools) |
| 294 | } |
| 295 | c.authed.Store(true) |
| 296 | |
| 297 | out := make(chan provider.Chunk) |
| 298 | go c.readStream(requestCtx, resp, out) |
| 299 | return out, nil |
| 300 | } |
| 301 | |
| 302 | // buildRequest converts the transport-agnostic Request into the Messages API shape: |
| 303 | // RoleSystem messages lift to the top-level `system` field; assistant tool calls |
| 304 | // become `tool_use` blocks; RoleTool results become `tool_result` blocks in a user |
| 305 | // turn. Consecutive same-role messages are coalesced because the API requires |
| 306 | // alternating user/assistant turns (tool results are user turns). |
| 307 | func (c *client) buildRequest(_ context.Context, req provider.Request) anthRequest { |
| 308 | var system []textBlock |
| 309 | var msgs []anthMessage |
| 310 | |
| 311 | // appendBlocks adds blocks under role, merging into the previous message when |
| 312 | // it shares the role (keeps user/assistant strictly alternating). |
| 313 | appendBlocks := func(role string, blocks ...contentBlock) { |
| 314 | if len(blocks) == 0 { |
| 315 | return |
| 316 | } |
| 317 | if n := len(msgs); n > 0 && msgs[n-1].Role == role { |
| 318 | msgs[n-1].Content = append(msgs[n-1].Content, blocks...) |
| 319 | return |
| 320 | } |
| 321 | msgs = append(msgs, anthMessage{Role: role, Content: blocks}) |
| 322 | } |
| 323 | |
| 324 | for _, m := range provider.SanitizeToolPairing(req.Messages) { |
| 325 | switch m.Role { |
| 326 | case provider.RoleSystem: |
| 327 | if m.Content != "" { |
| 328 | system = append(system, textBlock{Type: "text", Text: m.Content}) |
| 329 | } |
| 330 | case provider.RoleUser: |
| 331 | if m.Content != "" { |
| 332 | appendBlocks("user", contentBlock{Type: "text", Text: m.Content}) |
| 333 | } |
| 334 | if c.vision { |
| 335 | for _, url := range m.Images { |
| 336 | if mt, data, ok := provider.ParseImageDataURL(url); ok { |
| 337 | appendBlocks("user", contentBlock{Type: "image", Source: &imageSource{Type: "base64", MediaType: mt, Data: data}}) |
| 338 | } |
| 339 | } |
| 340 | } |
| 341 | case provider.RoleTool: |
| 342 | content := m.Content |
| 343 | if content == "" { |
| 344 | content = "(no output)" // tool_result content must be non-empty |
| 345 | } |
| 346 | block := contentBlock{Type: "tool_result", ToolUseID: m.ToolCallID, Content: content} |
| 347 | if c.vision { |
| 348 | if blocks := toolResultBlocks(content, m.Images); blocks != nil { |
| 349 | block.Content = blocks |
| 350 | } |
| 351 | } |
| 352 | appendBlocks("user", block) |
| 353 | case provider.RoleAssistant: |
| 354 | var blocks []contentBlock |
| 355 | // Replay provider reasoning before the tool_use it led to. DeepSeek uses |
| 356 | // unsigned thinking blocks and requires the reasoning from a tool-call |
| 357 | // turn in every subsequent request, even if the current request no longer |
| 358 | // declares tools or has since disabled thinking. Anthropic proper requires |
| 359 | // a signature, so reasoning without one cannot be replayed on that endpoint. |
| 360 | if c.deepseek && len(m.ToolCalls) > 0 && m.ReasoningContent != "" { |
| 361 | blocks = append(blocks, contentBlock{Type: "thinking", Thinking: m.ReasoningContent}) |
| 362 | } else if c.thinking == "adaptive" && m.ReasoningContent != "" && m.ReasoningSignature != "" { |
| 363 | blocks = append(blocks, contentBlock{Type: "thinking", Thinking: m.ReasoningContent, Signature: m.ReasoningSignature}) |
| 364 | } |
| 365 | if m.Content != "" { |
| 366 | blocks = append(blocks, contentBlock{Type: "text", Text: m.Content}) |
| 367 | } |
| 368 | for _, tc := range m.ToolCalls { |
| 369 | input := json.RawMessage(tc.Arguments) |
| 370 | if len(input) == 0 { |
| 371 | input = json.RawMessage("{}") // input is required, even when empty |
| 372 | } |
| 373 | blocks = append(blocks, contentBlock{Type: "tool_use", ID: tc.ID, Name: tc.Name, Input: input}) |
| 374 | } |
| 375 | appendBlocks("assistant", blocks...) |
| 376 | } |
| 377 | } |
| 378 | |
| 379 | var tools []anthTool |
| 380 | if c.webSearch { |
| 381 | tools = append(tools, anthTool{Type: "web_search_20250305", Name: "web_search"}) |
| 382 | } |
| 383 | for _, t := range req.Tools { |
| 384 | schema := t.Parameters |
| 385 | if len(schema) == 0 { |
| 386 | schema = json.RawMessage(`{"type":"object","properties":{}}`) |
| 387 | } |
| 388 | if c.mimo { |
| 389 | schema = provider.NormalizeLegacyTupleItemsForDraft202012(schema) |
| 390 | } |
| 391 | tools = append(tools, anthTool{Name: t.Name, Description: t.Description, InputSchema: schema}) |
| 392 | } |
| 393 | |
| 394 | // Prompt-cache breakpoints (ephemeral, prefix-match). DeepSeek ignores |
| 395 | // cache_control and manages prefix caching automatically, so keep those fields |
| 396 | // off its wire entirely. Render order for native Anthropic is |
| 397 | // tools → system → messages, so a marker on the last system block caches |
| 398 | // tools+system together; with no system, mark the last tool. A marker on the |
| 399 | // last block of the last message caches the conversation prefix, accruing hits |
| 400 | // incrementally as turns are appended. Max 4 breakpoints; we use ≤2. Keep |
| 401 | // Anthropic's default 5m TTL by omitting the ttl field. Besides being cheaper |
| 402 | // than the opt-in 1h write, this keeps provider-visible request bytes stable |
| 403 | // across turns, retries, and wall-clock timing. |
| 404 | if !c.deepseek { |
| 405 | if n := len(system); n > 0 { |
| 406 | system[n-1].CacheControl = ephemeral() |
| 407 | } else if n := len(tools); n > 0 { |
| 408 | tools[n-1].CacheControl = ephemeral() |
| 409 | } |
| 410 | if n := len(msgs); n > 0 { |
| 411 | if k := len(msgs[n-1].Content); k > 0 { |
| 412 | msgs[n-1].Content[k-1].CacheControl = ephemeral() |
| 413 | } |
| 414 | } |
| 415 | } |
| 416 | |
| 417 | maxTokens := req.MaxTokens |
| 418 | if maxTokens <= 0 { |
| 419 | maxTokens = c.defaultMaxTokens |
| 420 | if maxTokens <= 0 { |
| 421 | maxTokens = defaultMaxTokens |
| 422 | } |
| 423 | } |
| 424 | r := anthRequest{ |
| 425 | Model: c.model, |
| 426 | MaxTokens: maxTokens, |
| 427 | System: system, |
| 428 | Messages: msgs, |
| 429 | Tools: tools, |
| 430 | Stream: true, |
| 431 | } |
| 432 | // Extended thinking is provider-specific. DeepSeek defaults to enabled and |
| 433 | // accepts output_config.effort alongside its binary toggle. Anthropic proper |
| 434 | // uses type=adaptive plus display/output_config. LongCat-style compatible |
| 435 | // gateways use the simpler enabled|disabled knob and reject output_config. |
| 436 | if c.deepseek { |
| 437 | r.Temperature = req.Temperature |
| 438 | t := c.thinking |
| 439 | if t != "disabled" { |
| 440 | t = "enabled" |
| 441 | } |
| 442 | if c.effort == "disabled" { |
| 443 | t = "disabled" |
| 444 | } |
| 445 | r.Thinking = &thinkingConfig{Type: t} |
| 446 | if t != "disabled" { |
| 447 | effort := normalizeDeepSeekAnthropicEffort(c.model, c.effort) |
| 448 | switch effort { |
| 449 | case "low", "high", "max": |
| 450 | r.OutputConfig = &outputConfig{Effort: effort} |
| 451 | } |
| 452 | } |
| 453 | } else { |
| 454 | switch c.thinking { |
| 455 | case "adaptive": |
| 456 | r.Thinking = &thinkingConfig{Type: "adaptive", Display: "summarized"} |
| 457 | if c.effort != "" { |
| 458 | r.OutputConfig = &outputConfig{Effort: c.effort} |
| 459 | } |
| 460 | case "enabled", "disabled": |
| 461 | t := c.thinking |
| 462 | if c.effort == "enabled" || c.effort == "disabled" { |
| 463 | t = c.effort |
| 464 | } |
| 465 | r.Thinking = &thinkingConfig{Type: t} |
| 466 | } |
| 467 | } |
| 468 | return r |
| 469 | } |
| 470 | |
| 471 | // readStream parses the Messages API SSE stream into Chunks. Text deltas emit live; |
| 472 | // each tool_use content block emits a ChunkToolCallStart when its id+name are known |
| 473 | // and a complete ChunkToolCall when the block closes; usage is assembled from |
| 474 | // message_start/message_delta usage (compatible gateways may put every counter |
| 475 | // in the final delta) and emitted once before ChunkDone. |
| 476 | func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<- provider.Chunk) { |
| 477 | defer resp.Body.Close() |
| 478 | defer close(out) |
| 479 | |
| 480 | // Close the body if the stream stalls past c.idleTimeout so scanner.Scan() |
| 481 | // unblocks instead of hanging on a half-open connection. The watchdog owns the |
| 482 | // timer; the read loop only pings the buffered activity channel (no Timer.Reset |
| 483 | // race). A context cancel already unblocks the scan via the transport. |
| 484 | idleTimeout := c.idleTimeout |
| 485 | if idleTimeout <= 0 { // zero-value client (constructed without New) |
| 486 | idleTimeout = defaultStreamIdleTimeout |
| 487 | } |
| 488 | done := make(chan struct{}) |
| 489 | defer close(done) |
| 490 | activity := make(chan struct{}, 1) |
| 491 | var stalled atomic.Bool |
| 492 | go func() { |
| 493 | idle := time.NewTimer(idleTimeout) |
| 494 | defer idle.Stop() |
| 495 | for { |
| 496 | select { |
| 497 | case <-ctx.Done(): |
| 498 | resp.Body.Close() |
| 499 | return |
| 500 | case <-idle.C: |
| 501 | stalled.Store(true) |
| 502 | resp.Body.Close() |
| 503 | return |
| 504 | case <-activity: |
| 505 | if !idle.Stop() { |
| 506 | select { |
| 507 | case <-idle.C: |
| 508 | default: |
| 509 | } |
| 510 | } |
| 511 | idle.Reset(idleTimeout) |
| 512 | case <-done: |
| 513 | return |
| 514 | } |
| 515 | } |
| 516 | }() |
| 517 | |
| 518 | send := func(chunk provider.Chunk) bool { |
| 519 | return sendChunk(ctx, out, chunk) |
| 520 | } |
| 521 | |
| 522 | tools := map[int]*provider.ToolCall{} // tool_use blocks, keyed by content index |
| 523 | argBuckets := map[int]int{} // last emitted 2KB progress bucket per block |
| 524 | var inTok, outTok, cacheCreate, cacheRead int |
| 525 | var stopReason string |
| 526 | haveUsage := false |
| 527 | mergeUsage := func(usage *wireUsage) { |
| 528 | if usage == nil { |
| 529 | return |
| 530 | } |
| 531 | // The native Anthropic stream reports input/cache counters in |
| 532 | // message_start and output_tokens in message_delta. Compatible gateways |
| 533 | // such as LongCat report all counters in message_delta instead. Counters |
| 534 | // are cumulative and non-negative, so retaining the largest value also |
| 535 | // tolerates gateways that repeat partial usage in both events. |
| 536 | inTok = max(inTok, usage.InputTokens) |
| 537 | outTok = max(outTok, usage.OutputTokens) |
| 538 | cacheCreate = max(cacheCreate, usage.CacheCreationInputTokens) |
| 539 | cacheRead = max(cacheRead, usage.CacheReadInputTokens) |
| 540 | haveUsage = true |
| 541 | } |
| 542 | |
| 543 | scanner := bufio.NewScanner(resp.Body) |
| 544 | scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) |
| 545 | |
| 546 | for scanner.Scan() { |
| 547 | select { // ping the idle watchdog; non-blocking so a full buffer is fine |
| 548 | case activity <- struct{}{}: |
| 549 | default: |
| 550 | } |
| 551 | line := strings.TrimSpace(scanner.Text()) |
| 552 | // SSE carries `event:` and `data:` lines; the data JSON's own `type` field |
| 553 | // is authoritative, so we only need the data payloads. |
| 554 | if !strings.HasPrefix(line, "data:") { |
| 555 | continue |
| 556 | } |
| 557 | data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) |
| 558 | if data == "" { |
| 559 | continue |
| 560 | } |
| 561 | |
| 562 | var ev streamEvent |
| 563 | if err := json.Unmarshal([]byte(data), &ev); err != nil { |
| 564 | send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamDecodeError(c.name, data, err)}) |
| 565 | return |
| 566 | } |
| 567 | |
| 568 | switch ev.Type { |
| 569 | case "message_start": |
| 570 | if ev.Message != nil && ev.Message.Usage != nil { |
| 571 | mergeUsage(ev.Message.Usage) |
| 572 | } |
| 573 | case "content_block_start": |
| 574 | if ev.ContentBlock != nil { |
| 575 | switch ev.ContentBlock.Type { |
| 576 | case "tool_use": |
| 577 | tc := &provider.ToolCall{ID: ev.ContentBlock.ID, Name: ev.ContentBlock.Name} |
| 578 | tools[ev.Index] = tc |
| 579 | if !send(provider.Chunk{Type: provider.ChunkToolCallStart, ToolCall: &provider.ToolCall{ID: tc.ID, Name: tc.Name}}) { |
| 580 | return |
| 581 | } |
| 582 | case "web_search_tool_result": |
| 583 | // Search results are delivered inline in content_block.content as a |
| 584 | // JSON array of result objects (title, url, encrypted_content). |
| 585 | // Only the model sees the plain text; we surface titles and URLs. |
| 586 | // server_tool_use blocks (the model initiating the search) are |
| 587 | // intentionally skipped — the API executes them server-side and |
| 588 | // the results appear here. |
| 589 | formatted := formatWebSearchResults(ev.ContentBlock.Content) |
| 590 | if formatted != "" { |
| 591 | if !send(provider.Chunk{Type: provider.ChunkText, Text: formatted}) { |
| 592 | return |
| 593 | } |
| 594 | } |
| 595 | } |
| 596 | } |
| 597 | case "content_block_delta": |
| 598 | if ev.Delta == nil { |
| 599 | continue |
| 600 | } |
| 601 | switch ev.Delta.Type { |
| 602 | case "text_delta": |
| 603 | if ev.Delta.Text != "" { |
| 604 | if !send(provider.Chunk{Type: provider.ChunkText, Text: ev.Delta.Text}) { |
| 605 | return |
| 606 | } |
| 607 | } |
| 608 | case "thinking_delta": |
| 609 | if ev.Delta.Thinking != "" { |
| 610 | if !send(provider.Chunk{Type: provider.ChunkReasoning, Text: ev.Delta.Thinking}) { |
| 611 | return |
| 612 | } |
| 613 | } |
| 614 | case "signature_delta": |
| 615 | if ev.Delta.Signature != "" { |
| 616 | if !send(provider.Chunk{Type: provider.ChunkReasoning, Signature: ev.Delta.Signature}) { |
| 617 | return |
| 618 | } |
| 619 | } |
| 620 | case "input_json_delta": |
| 621 | if tc := tools[ev.Index]; tc != nil { |
| 622 | tc.Arguments += ev.Delta.PartialJSON |
| 623 | // Progress ticks for large streaming argument payloads, one |
| 624 | // per 2KB bucket (see the openai provider for rationale). |
| 625 | if bucket := len(tc.Arguments) / 2048; bucket > argBuckets[ev.Index] { |
| 626 | argBuckets[ev.Index] = bucket |
| 627 | if !send(provider.Chunk{Type: provider.ChunkToolCallArgsDelta, ToolCall: &provider.ToolCall{ID: tc.ID, Name: tc.Name}, ArgChars: len(tc.Arguments)}) { |
| 628 | return |
| 629 | } |
| 630 | } |
| 631 | } |
| 632 | } |
| 633 | case "content_block_stop": |
| 634 | if tc := tools[ev.Index]; tc != nil { |
| 635 | if !send(provider.Chunk{Type: provider.ChunkToolCall, ToolCall: tc}) { |
| 636 | return |
| 637 | } |
| 638 | delete(tools, ev.Index) |
| 639 | } |
| 640 | case "message_delta": |
| 641 | if ev.Delta != nil && ev.Delta.StopReason != "" { |
| 642 | stopReason = ev.Delta.StopReason |
| 643 | } |
| 644 | mergeUsage(ev.Usage) |
| 645 | case "message_stop": |
| 646 | // Anthropic's terminal event. Tool blocks may already have closed; |
| 647 | // without this, the attempt stays speculative and is not committed. |
| 648 | // Stop reading immediately so a post-terminal connection reset cannot |
| 649 | // reclassify a complete response as interrupted. |
| 650 | goto finalize |
| 651 | case "error": |
| 652 | msg := "stream error" |
| 653 | if ev.Error != nil && ev.Error.Message != "" { |
| 654 | msg = ev.Error.Message |
| 655 | } |
| 656 | send(provider.Chunk{Type: provider.ChunkError, Err: fmt.Errorf("%s: %s", c.name, msg)}) |
| 657 | return |
| 658 | } |
| 659 | } |
| 660 | |
| 661 | if ctx.Err() != nil { |
| 662 | return |
| 663 | } |
| 664 | if stalled.Load() { |
| 665 | err := fmt.Errorf("%s: stream stalled — no data for %s, connection likely dropped", c.name, idleTimeout) |
| 666 | send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt(err, provider.StreamInterruptIdleTimeout)}) |
| 667 | return |
| 668 | } |
| 669 | if err := scanner.Err(); err != nil { |
| 670 | wrapped := fmt.Errorf("%s: read stream: %w", c.name, err) |
| 671 | if provider.IsConnReset(err) { |
| 672 | send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt(wrapped, provider.ClassifyStreamInterrupt(err))}) |
| 673 | return |
| 674 | } |
| 675 | send(provider.Chunk{Type: provider.ChunkError, Err: wrapped}) |
| 676 | return |
| 677 | } |
| 678 | // EOF / clean close before message_stop is an uncommitted attempt. Complete |
| 679 | // ChunkToolCall blocks that arrived earlier remain speculative. |
| 680 | send(provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt( |
| 681 | fmt.Errorf("%s: stream ended before message_stop: %w", c.name, io.ErrUnexpectedEOF), |
| 682 | provider.StreamInterruptPrematureEOF, |
| 683 | )}) |
| 684 | return |
| 685 | |
| 686 | finalize: |
| 687 | if haveUsage { |
| 688 | cacheWriteBilledTokens := 0.0 |
| 689 | if cacheCreate > 0 && c.nativeAnthropic { |
| 690 | cacheWriteBilledTokens = float64(cacheCreate) * cacheWrite5MinuteInputMultiplier |
| 691 | } |
| 692 | usage := &provider.Usage{ |
| 693 | PromptTokens: inTok + cacheCreate + cacheRead, |
| 694 | CompletionTokens: outTok, |
| 695 | TotalTokens: inTok + cacheCreate + cacheRead + outTok, |
| 696 | CacheHitTokens: cacheRead, |
| 697 | CacheMissTokens: inTok + cacheCreate, |
| 698 | CacheWriteTokens: cacheCreate, |
| 699 | CacheWriteBilledTokens: cacheWriteBilledTokens, |
| 700 | FinishReason: mapStopReason(stopReason), |
| 701 | } |
| 702 | provider.ApplyRequestAttemptCount(ctx, usage) |
| 703 | if !send(provider.Chunk{Type: provider.ChunkUsage, Usage: usage}) { |
| 704 | return |
| 705 | } |
| 706 | } |
| 707 | send(provider.Chunk{Type: provider.ChunkDone}) |
| 708 | } |
| 709 | |
| 710 | func sendChunk(ctx context.Context, out chan<- provider.Chunk, chunk provider.Chunk) bool { |
| 711 | select { |
| 712 | case out <- chunk: |
| 713 | return true |
| 714 | default: |
| 715 | } |
| 716 | select { |
| 717 | case <-ctx.Done(): |
| 718 | return false |
| 719 | case out <- chunk: |
| 720 | return true |
| 721 | } |
| 722 | } |
| 723 | |
| 724 | // mapStopReason translates Anthropic stop reasons to the OpenAI-style finish |
| 725 | // reasons the agent already recognises (it surfaces abnormal ones like "length"). |
| 726 | func mapStopReason(s string) string { |
| 727 | switch s { |
| 728 | case "end_turn", "stop_sequence": |
| 729 | return "stop" |
| 730 | case "tool_use": |
| 731 | return "tool_calls" |
| 732 | case "max_tokens": |
| 733 | return "length" |
| 734 | default: |
| 735 | return s // "refusal", "pause_turn", "" — pass through |
| 736 | } |
| 737 | } |
| 738 | |
| 739 | // webSearchResult is a single result from a web_search_tool_result block. |
| 740 | type webSearchResult struct { |
| 741 | URL string `json:"url"` |
| 742 | Title string `json:"title"` |
| 743 | Text string `json:"text"` |
| 744 | SiteName string `json:"site_name"` |
| 745 | } |
| 746 | |
| 747 | // formatWebSearchResults parses a web_search_tool_result content array |
| 748 | // and formats titles and URLs as human-readable text. DeepSeek returns |
| 749 | // encrypted_content rather than plain text at the transport layer; the |
| 750 | // model still sees the original content. |
| 751 | func formatWebSearchResults(raw json.RawMessage) string { |
| 752 | if len(raw) == 0 { |
| 753 | return "" |
| 754 | } |
| 755 | var results []webSearchResult |
| 756 | if err := json.Unmarshal(raw, &results); err != nil { |
| 757 | return "" |
| 758 | } |
| 759 | var b strings.Builder |
| 760 | for _, r := range results { |
| 761 | if r.Title == "" && r.URL == "" { |
| 762 | continue |
| 763 | } |
| 764 | fmt.Fprintf(&b, "\n- **%s**", r.Title) |
| 765 | if r.URL != "" { |
| 766 | fmt.Fprintf(&b, "\n <%s>", r.URL) |
| 767 | } |
| 768 | } |
| 769 | if b.Len() == 0 { |
| 770 | return "" |
| 771 | } |
| 772 | return "\n" + b.String() + "\n" |
| 773 | } |
| 774 | |
| 775 | // --- Messages API wire protocol --- |
| 776 | |
| 777 | const cacheWrite5MinuteInputMultiplier = 1.25 |
| 778 | |
| 779 | func ephemeral() *cacheControl { return &cacheControl{Type: "ephemeral"} } |
| 780 | |
| 781 | type cacheControl struct { |
| 782 | Type string `json:"type"` |
| 783 | } |
| 784 | |
| 785 | type anthRequest struct { |
| 786 | Model string `json:"model"` |
| 787 | MaxTokens int `json:"max_tokens"` |
| 788 | System []textBlock `json:"system,omitempty"` |
| 789 | Messages []anthMessage `json:"messages"` |
| 790 | Tools []anthTool `json:"tools,omitempty"` |
| 791 | Temperature *float64 `json:"temperature,omitempty"` |
| 792 | Thinking *thinkingConfig `json:"thinking,omitempty"` |
| 793 | OutputConfig *outputConfig `json:"output_config,omitempty"` |
| 794 | Stream bool `json:"stream"` |
| 795 | } |
| 796 | |
| 797 | type thinkingConfig struct { |
| 798 | Type string `json:"type"` // "adaptive" |
| 799 | Display string `json:"display,omitempty"` // "summarized" to stream the reasoning text |
| 800 | } |
| 801 | |
| 802 | type outputConfig struct { |
| 803 | Effort string `json:"effort,omitempty"` // low | high | max |
| 804 | } |
| 805 | |
| 806 | type textBlock struct { |
| 807 | Type string `json:"type"` |
| 808 | Text string `json:"text"` |
| 809 | CacheControl *cacheControl `json:"cache_control,omitempty"` |
| 810 | } |
| 811 | |
| 812 | type anthMessage struct { |
| 813 | Role string `json:"role"` |
| 814 | Content []contentBlock `json:"content"` |
| 815 | } |
| 816 | |
| 817 | // contentBlock is the union of the block kinds we emit in a request: text, |
| 818 | // tool_use (echoing a prior assistant call), and tool_result. Unused fields are |
| 819 | // omitted so each block serialises to its canonical shape. |
| 820 | type contentBlock struct { |
| 821 | Type string `json:"type"` |
| 822 | Text string `json:"text,omitempty"` // text |
| 823 | Thinking string `json:"thinking,omitempty"` // thinking |
| 824 | Signature string `json:"signature,omitempty"` // thinking |
| 825 | ID string `json:"id,omitempty"` // tool_use |
| 826 | Name string `json:"name,omitempty"` // tool_use |
| 827 | Input json.RawMessage `json:"input,omitempty"` // tool_use |
| 828 | ToolUseID string `json:"tool_use_id,omitempty"` // tool_result |
| 829 | Content any `json:"content,omitempty"` // tool_result: string, or []contentBlock when the result carries images |
| 830 | Source *imageSource `json:"source,omitempty"` // image |
| 831 | CacheControl *cacheControl `json:"cache_control,omitempty"` |
| 832 | } |
| 833 | |
| 834 | type imageSource struct { |
| 835 | Type string `json:"type"` // "base64" |
| 836 | MediaType string `json:"media_type"` |
| 837 | Data string `json:"data"` |
| 838 | } |
| 839 | |
| 840 | // toolResultBlocks builds array content for a tool_result whose message carries |
| 841 | // images: the text first, then one image block per parseable data URL. It |
| 842 | // returns nil when nothing parses, so text-only results keep plain string |
| 843 | // content — byte-identical serialization to previous releases. |
| 844 | func toolResultBlocks(text string, images []string) []contentBlock { |
| 845 | var imgs []contentBlock |
| 846 | for _, url := range images { |
| 847 | if mt, data, ok := provider.ParseImageDataURL(url); ok { |
| 848 | imgs = append(imgs, contentBlock{Type: "image", Source: &imageSource{Type: "base64", MediaType: mt, Data: data}}) |
| 849 | } |
| 850 | } |
| 851 | if imgs == nil { |
| 852 | return nil |
| 853 | } |
| 854 | return append([]contentBlock{{Type: "text", Text: text}}, imgs...) |
| 855 | } |
| 856 | |
| 857 | type anthTool struct { |
| 858 | Type string `json:"type,omitempty"` // "web_search" for server-side search; empty for named tools |
| 859 | Name string `json:"name,omitempty"` |
| 860 | Description string `json:"description,omitempty"` |
| 861 | InputSchema json.RawMessage `json:"input_schema,omitempty"` |
| 862 | CacheControl *cacheControl `json:"cache_control,omitempty"` |
| 863 | } |
| 864 | |
| 865 | // streamEvent is the discriminated SSE event; read the fields matching Type. |
| 866 | type streamEvent struct { |
| 867 | Type string `json:"type"` |
| 868 | Index int `json:"index"` |
| 869 | Message *struct { |
| 870 | Usage *wireUsage `json:"usage"` |
| 871 | } `json:"message"` |
| 872 | ContentBlock *struct { |
| 873 | Type string `json:"type"` |
| 874 | ID string `json:"id"` |
| 875 | Name string `json:"name"` |
| 876 | ToolUseID string `json:"tool_use_id"` // web_search_tool_result |
| 877 | Content json.RawMessage `json:"content"` // web_search_tool_result: array of result objects |
| 878 | } `json:"content_block"` |
| 879 | Delta *struct { |
| 880 | Type string `json:"type"` // text_delta | thinking_delta | signature_delta | input_json_delta | web_search_tool_result_delta |
| 881 | Text string `json:"text"` // text_delta |
| 882 | Thinking string `json:"thinking"` // thinking_delta |
| 883 | Signature string `json:"signature"` // signature_delta |
| 884 | PartialJSON string `json:"partial_json"` // input_json_delta |
| 885 | StopReason string `json:"stop_reason"` // message_delta |
| 886 | WebSearchResults json.RawMessage `json:"results"` // web_search_tool_result_delta |
| 887 | } `json:"delta"` |
| 888 | Usage *wireUsage `json:"usage"` // message_delta (cumulative output_tokens) |
| 889 | Error *struct { |
| 890 | Type string `json:"type"` |
| 891 | Message string `json:"message"` |
| 892 | } `json:"error"` |
| 893 | } |
| 894 | |
| 895 | type wireUsage struct { |
| 896 | InputTokens int `json:"input_tokens"` |
| 897 | OutputTokens int `json:"output_tokens"` |
| 898 | CacheCreationInputTokens int `json:"cache_creation_input_tokens"` |
| 899 | CacheReadInputTokens int `json:"cache_read_input_tokens"` |
| 900 | } |
| 901 |