| 1 | // Package openai implements the OpenAI-compatible /chat/completions provider. |
| 2 | // It self-registers under the "openai" kind, so DeepSeek, MiMo, MiniMax-M3, and |
| 3 | // any other OpenAI-compatible endpoint are just config instances rather than |
| 4 | // code. Each instance picks the wire shape from its base URL: |
| 5 | // - api.deepseek.com → emits thinking.type=enabled (DeepSeek-flavor CoT) plus |
| 6 | // reasoning_effort as a depth hint. |
| 7 | // - api.minimaxi.com → emits thinking.type=adaptive|disabled (M3's binary |
| 8 | // knob) instead of reasoning_effort, since M3 has no level scale. |
| 9 | // - open.bigmodel.cn / api.z.ai (Zhipu GLM) → emits thinking.type=enabled| |
| 10 | // disabled instead of reasoning_effort, which Zhipu silently ignores. |
| 11 | // - api.longcat.chat → emits thinking.type=enabled|disabled and omits |
| 12 | // reasoning_effort, matching LongCat's OpenAI-compatible API. |
| 13 | // - ollama.com → accepts hosted Ollama Cloud's reasoning_effort scale, |
| 14 | // including max, and omits the field for none/disabled. |
| 15 | // - official Kimi API + kimi-k3 preserves complete assistant messages and |
| 16 | // uses K3's fixed-sampling/max_completion_tokens request shape. |
| 17 | // - everything else (MiMo and other OpenAI-compatible gateways) uses the |
| 18 | // vanilla reasoning_effort scale (low/medium/high), unless its config |
| 19 | // declares a custom supported_efforts validation contract. |
| 20 | // |
| 21 | // See docs/REASONING_PROVIDERS.md for the per-backend protocol reference. |
| 22 | package openai |
| 23 | |
| 24 | import ( |
| 25 | "bufio" |
| 26 | "bytes" |
| 27 | "context" |
| 28 | "encoding/json" |
| 29 | "errors" |
| 30 | "fmt" |
| 31 | "io" |
| 32 | "net/http" |
| 33 | "sort" |
| 34 | "strings" |
| 35 | "sync" |
| 36 | "sync/atomic" |
| 37 | "time" |
| 38 | |
| 39 | "reasonix/internal/netclient" |
| 40 | "reasonix/internal/provider" |
| 41 | ) |
| 42 | |
| 43 | // defaultStreamIdleTimeout caps how long a started SSE stream may go without any |
| 44 | // bytes before it's treated as a dropped connection. A half-open TCP connection |
| 45 | // (e.g. a proxy switched mid-stream) sends no RST, so scanner.Scan() would block |
| 46 | // forever; this turns that hang into a recoverable error. Generous on purpose — |
| 47 | // live streams emit tokens/keepalives far more often. Stored per-client |
| 48 | // (client.idleTimeout) so a test can shorten it without a shared global that |
| 49 | // would race other streams' watchdogs. |
| 50 | const defaultStreamIdleTimeout = 120 * time.Second |
| 51 | |
| 52 | // maxPrefixContinuations keeps automatic recovery bounded. A second length |
| 53 | // finish is surfaced through the existing truncation notice instead of opening |
| 54 | // an unbounded (and billable) continuation loop against the Beta endpoint. |
| 55 | const maxPrefixContinuations = 1 |
| 56 | |
| 57 | func init() { |
| 58 | provider.Register("openai", New) |
| 59 | } |
| 60 | |
| 61 | // New builds an OpenAI-compatible provider from a resolved config. |
| 62 | func New(cfg provider.Config) (provider.Provider, error) { |
| 63 | if cfg.BaseURL == "" { |
| 64 | return nil, fmt.Errorf("openai: base_url is required for provider %q", cfg.Name) |
| 65 | } |
| 66 | if cfg.Model == "" { |
| 67 | return nil, fmt.Errorf("openai: model is required for provider %q", cfg.Name) |
| 68 | } |
| 69 | name := cfg.Name |
| 70 | if name == "" { |
| 71 | name = "openai" |
| 72 | } |
| 73 | keyEnv, _ := cfg.Extra["api_key_env"].(string) // for actionable auth errors |
| 74 | keySource, _ := cfg.Extra["api_key_source"].(string) |
| 75 | effort, _ := cfg.Extra["effort"].(string) |
| 76 | effort = strings.ToLower(strings.TrimSpace(effort)) |
| 77 | if effort == "auto" { |
| 78 | effort = "" |
| 79 | } |
| 80 | supportedEfforts, _ := cfg.Extra["supported_efforts"].([]string) |
| 81 | // A meaningful explicit list is the endpoint's declared effort vocabulary; |
| 82 | // auto remains implicit and is therefore ignored here. |
| 83 | hasExplicitEfforts := hasExplicitSupportedEfforts(supportedEfforts) |
| 84 | protocol, _ := cfg.Extra["reasoning_protocol"].(string) |
| 85 | protocol = normalizeReasoningProtocol(protocol) |
| 86 | chatURL, _ := cfg.Extra["chat_url"].(string) |
| 87 | chatURL = normalizeChatURL(cfg.BaseURL, chatURL) |
| 88 | prefixChatURL := deepSeekPrefixChatURL(chatURL) |
| 89 | headers, _ := cfg.Extra["headers"].(map[string]string) |
| 90 | extraBody, _ := cfg.Extra["extra_body"].(map[string]any) |
| 91 | vision, _ := cfg.Extra["vision"].(bool) |
| 92 | explicitModelVision, _ := cfg.Extra["vision_model_explicit"].(bool) |
| 93 | officialDeepSeek := IsDeepSeek(cfg.BaseURL) |
| 94 | // DeepSeek's official chat API accepts string message content only. Keep |
| 95 | // this provider-boundary guard even though config capability resolution |
| 96 | // normally prevents image attachments from reaching this layer. A positive |
| 97 | // model-scoped capability can opt in without letting stale provider-wide |
| 98 | // vision=true settings affect current text-only models. |
| 99 | vision = vision && (!officialDeepSeek || explicitModelVision) |
| 100 | visionDetail, _ := cfg.Extra["vision_detail"].(string) |
| 101 | visionDetail = strings.ToLower(strings.TrimSpace(visionDetail)) |
| 102 | if visionDetail != "low" && visionDetail != "high" { |
| 103 | visionDetail = "" // auto — omit the field |
| 104 | } |
| 105 | deepseek := protocol == "deepseek" || (protocol == "" && officialDeepSeek) |
| 106 | maxOutputTokens, _ := cfg.Extra["max_output_tokens"].(int) |
| 107 | deepseekV4Flash := strings.EqualFold(strings.TrimSpace(cfg.Model), "deepseek-v4-flash") |
| 108 | minimax := protocol == "" && IsMiniMax(cfg.BaseURL) |
| 109 | zhipu := protocol == "glm" || (protocol == "" && IsZhipu(cfg.BaseURL)) |
| 110 | longcat := protocol == "" && IsLongCat(cfg.BaseURL) |
| 111 | ollamaCloud := protocol == "" && IsOllamaCloud(cfg.BaseURL) |
| 112 | kimiK3 := IsKimiAPI(cfg.BaseURL) && strings.EqualFold(strings.TrimSpace(cfg.Model), "kimi-k3") |
| 113 | // Optional explicit `thinking` config field — a vendor-agnostic escape hatch |
| 114 | // (credit @eghrhegpe, #5063) for OpenAI-compatible providers we don't |
| 115 | // auto-detect (e.g. opencode.ai). "enabled"/"disabled" drive thinking.type; |
| 116 | // anything else is ignored so an unknown value never breaks a request. |
| 117 | thinkingType, _ := cfg.Extra["thinking"].(string) |
| 118 | thinkingType = strings.ToLower(strings.TrimSpace(thinkingType)) |
| 119 | if thinkingType != "enabled" && thinkingType != "disabled" { |
| 120 | thinkingType = "" |
| 121 | } |
| 122 | switch { |
| 123 | case protocol == "none": |
| 124 | effort = "" |
| 125 | case deepseek: |
| 126 | if thinkingType == "disabled" { |
| 127 | effort = "" |
| 128 | break |
| 129 | } |
| 130 | switch effort { |
| 131 | case "", "off": // "off" is a retired level (disabled thinking); fall back to the default depth |
| 132 | effort = "high" |
| 133 | case "disabled": |
| 134 | if hasExplicitEfforts && !supportsEffort(supportedEfforts, effort) { |
| 135 | return nil, fmt.Errorf("openai: provider %q: effort %q is not listed in supported_efforts: %v", name, effort, supportedEfforts) |
| 136 | } |
| 137 | // DeepSeek can turn thinking off too; route through thinking.type and |
| 138 | // drop the depth hint so the wire carries thinking.type=disabled only. |
| 139 | effort = "" |
| 140 | thinkingType = "disabled" |
| 141 | default: |
| 142 | if hasExplicitEfforts { |
| 143 | // A provider that declares supported_efforts defines the endpoint's |
| 144 | // complete effort vocabulary. Honor that list for compatible DeepSeek |
| 145 | // request shapes instead of applying the built-in official scale. |
| 146 | if !supportsEffort(supportedEfforts, effort) { |
| 147 | return nil, fmt.Errorf("openai: provider %q: effort %q is not listed in supported_efforts: %v", name, effort, supportedEfforts) |
| 148 | } |
| 149 | break |
| 150 | } |
| 151 | switch effort { |
| 152 | case "low": |
| 153 | if !deepseekV4Flash { |
| 154 | return nil, fmt.Errorf("openai: provider %q uses DeepSeek thinking; effort low requires deepseek-v4-flash or explicit supported_efforts", name) |
| 155 | } |
| 156 | case "high", "max": |
| 157 | default: |
| 158 | return nil, fmt.Errorf("openai: provider %q uses DeepSeek thinking; effort must be low, high, max, or disabled", name) |
| 159 | } |
| 160 | } |
| 161 | case minimax: |
| 162 | // M3's knob is binary. The config effort layer normalises user input |
| 163 | // to "adaptive", "disabled", or "" (== auto). We keep "high"/"max" |
| 164 | // (legacy DeepSeek) and "low"/"medium" (Anthropic) out — config-level |
| 165 | // NormalizeEffort remaps them to "adaptive" already, so anything |
| 166 | // reaching here is expected to be one of: "", "adaptive", "disabled". |
| 167 | effort = strings.ToLower(strings.TrimSpace(effort)) |
| 168 | switch effort { |
| 169 | case "": // auto — leave empty so the wire emits thinking.type=adaptive |
| 170 | case "adaptive", "disabled": |
| 171 | default: |
| 172 | return nil, fmt.Errorf("openai: provider %q uses MiniMax thinking; effort must be adaptive or disabled", name) |
| 173 | } |
| 174 | case zhipu: |
| 175 | // Zhipu GLM gates chain-of-thought through `thinking.type` |
| 176 | // (enabled|disabled) and silently ignores reasoning_effort, so /effort |
| 177 | // mirrors that binary knob. The config effort layer normalises depth |
| 178 | // levels onto one of these; "" means auto == the GLM default (thinking on). |
| 179 | switch effort { |
| 180 | case "", "enabled", "disabled": |
| 181 | default: |
| 182 | return nil, fmt.Errorf("openai: provider %q uses Zhipu thinking; effort must be enabled or disabled", name) |
| 183 | } |
| 184 | case longcat: |
| 185 | // LongCat exposes a binary thinking knob on its OpenAI-compatible endpoint: |
| 186 | // thinking.type=enabled|disabled. It documents reasoning text via |
| 187 | // reasoning_content, but not the generic reasoning_effort scale. |
| 188 | switch effort { |
| 189 | case "", "enabled", "disabled": |
| 190 | default: |
| 191 | return nil, fmt.Errorf("openai: provider %q uses LongCat thinking; effort must be enabled or disabled", name) |
| 192 | } |
| 193 | case ollamaCloud: |
| 194 | // Hosted Ollama Cloud uses top-level reasoning_effort. "none" and the |
| 195 | // legacy/off aliases intentionally omit the field, which lets the model |
| 196 | // run without thinking. Local Ollama is not auto-detected because its |
| 197 | // model/version support varies. |
| 198 | switch effort { |
| 199 | case "", "none", "disabled", "off": |
| 200 | effort = "" |
| 201 | case "xhigh", "max": |
| 202 | effort = "max" |
| 203 | case "low", "medium", "high": |
| 204 | default: |
| 205 | return nil, fmt.Errorf("openai: provider %q uses Ollama Cloud thinking; effort must be none, low, medium, high, or max", name) |
| 206 | } |
| 207 | case effort != "": |
| 208 | if hasExplicitEfforts { |
| 209 | // Explicit endpoint metadata overrides the generic OpenAI enum and its |
| 210 | // legacy max-to-high compatibility clamp. |
| 211 | if !supportsEffort(supportedEfforts, effort) { |
| 212 | return nil, fmt.Errorf("openai: provider %q: effort %q is not listed in supported_efforts: %v", name, effort, supportedEfforts) |
| 213 | } |
| 214 | break |
| 215 | } |
| 216 | // Non-DeepSeek backends use OpenAI's reasoning_effort scale (low/medium/ |
| 217 | // high) by default. Without an explicit provider vocabulary, max remains |
| 218 | // clamped to the OpenAI ceiling because MiMo and similar backends reject it. |
| 219 | switch effort { |
| 220 | case "max": |
| 221 | effort = "high" |
| 222 | case "low", "medium", "high": |
| 223 | default: |
| 224 | return nil, fmt.Errorf("openai: provider %q: effort must be low, medium, or high", name) |
| 225 | } |
| 226 | } |
| 227 | // The automatic cap protects DeepSeek reasoning, not ordinary long-form |
| 228 | // output. Preserve an explicit user budget in either mode, but leave a |
| 229 | // thinking-disabled request uncapped unless the user configured one. |
| 230 | if maxOutputTokens == 0 && officialDeepSeek && thinkingType != "disabled" { |
| 231 | maxOutputTokens = provider.DefaultReasoningOutputTokens |
| 232 | } |
| 233 | httpClient, err := newHTTPClient(cfg) |
| 234 | if err != nil { |
| 235 | return nil, fmt.Errorf("openai: network: %w", err) |
| 236 | } |
| 237 | return &client{ |
| 238 | name: name, |
| 239 | apiKey: cfg.APIKey, |
| 240 | keyEnv: keyEnv, |
| 241 | keySource: keySource, |
| 242 | baseURL: strings.TrimRight(cfg.BaseURL, "/"), |
| 243 | chatURL: chatURL, |
| 244 | prefixChatURL: prefixChatURL, |
| 245 | headers: cleanCustomHeaders(headers), |
| 246 | extraBody: cleanExtraBody(extraBody), |
| 247 | model: normalizeModelID(cfg.BaseURL, cfg.Model), |
| 248 | deepseek: deepseek, |
| 249 | minimax: minimax, |
| 250 | zhipu: zhipu, |
| 251 | longcat: longcat, |
| 252 | kimiK3: kimiK3, |
| 253 | mimo: IsMiMo(cfg.BaseURL), |
| 254 | thinkingType: thinkingType, |
| 255 | vision: vision, |
| 256 | visionDetail: visionDetail, |
| 257 | maxOutputTokens: maxOutputTokens, |
| 258 | effort: effort, |
| 259 | http: httpClient, |
| 260 | idleTimeout: defaultStreamIdleTimeout, |
| 261 | }, nil |
| 262 | } |
| 263 | |
| 264 | func supportsEffort(levels []string, want string) bool { |
| 265 | want = strings.ToLower(strings.TrimSpace(want)) |
| 266 | for _, level := range levels { |
| 267 | if strings.ToLower(strings.TrimSpace(level)) == want { |
| 268 | return true |
| 269 | } |
| 270 | } |
| 271 | return false |
| 272 | } |
| 273 | |
| 274 | func hasExplicitSupportedEfforts(levels []string) bool { |
| 275 | for _, level := range levels { |
| 276 | level = strings.ToLower(strings.TrimSpace(level)) |
| 277 | if level != "" && level != "auto" { |
| 278 | return true |
| 279 | } |
| 280 | } |
| 281 | return false |
| 282 | } |
| 283 | |
| 284 | func newHTTPClient(cfg provider.Config) (*http.Client, error) { |
| 285 | spec, _ := cfg.Extra["proxy_spec"].(netclient.ProxySpec) |
| 286 | return netclient.NewHTTPClient(spec, netclient.TransportOptions{ |
| 287 | DialTimeout: 30 * time.Second, |
| 288 | KeepAlive: 30 * time.Second, |
| 289 | TLSHandshakeTimeout: 15 * time.Second, |
| 290 | ResponseHeaderTimeout: 120 * time.Second, // models can think for a while before the first token |
| 291 | }) |
| 292 | } |
| 293 | |
| 294 | type client struct { |
| 295 | name string |
| 296 | apiKey string |
| 297 | keyEnv string // api_key_env name, surfaced in auth errors |
| 298 | keySource string // source of keyEnv, surfaced in auth errors |
| 299 | baseURL string |
| 300 | chatURL string |
| 301 | prefixChatURL string // official DeepSeek Beta endpoint; empty for custom gateways |
| 302 | headers map[string]string |
| 303 | extraBody map[string]any |
| 304 | model string |
| 305 | http *http.Client |
| 306 | deepseek bool |
| 307 | minimax bool // true for api.minimaxi.com — emits MiniMax-M3's thinking knob instead of reasoning_effort |
| 308 | zhipu bool // true for Zhipu GLM (bigmodel.cn / z.ai) — gates thinking via thinking.type, ignores reasoning_effort |
| 309 | longcat bool // true for LongCat — gates thinking via thinking.type, ignores reasoning_effort |
| 310 | kimiK3 bool // true only for kimi-k3 on Moonshot's official direct API hosts |
| 311 | mimo bool // true for MiMo — upgrades legacy tuple schemas to Draft 2020-12 |
| 312 | thinkingType string // explicit `thinking` config override (enabled|disabled); "" = no override |
| 313 | vision bool // model accepts image input — embed attached images as image_url parts |
| 314 | visionDetail string // image_url detail hint (low|high); "" = auto/omit |
| 315 | maxOutputTokens int // configured/default total output budget; <=0 omits the optional field |
| 316 | effort string // reasoning_effort for OpenAI; thinking.type for MiniMax; "" = auto/provider default |
| 317 | idleTimeout time.Duration // SSE stall watchdog window; defaultStreamIdleTimeout unless a test overrides |
| 318 | authed atomic.Bool // a request has succeeded — gate transient-401 retry |
| 319 | } |
| 320 | |
| 321 | func (c *client) Name() string { return c.name } |
| 322 | |
| 323 | func (c *client) RequiresToolCallReasoning() bool { |
| 324 | return c != nil && c.deepseek && c.thinkingType != "disabled" |
| 325 | } |
| 326 | |
| 327 | func (c *client) RequiresReasoningRoundTrip() bool { |
| 328 | return c != nil && (c.kimiK3 || c.glmThinkingEnabled()) |
| 329 | } |
| 330 | |
| 331 | func (c *client) WarnOnMissingToolCallReasoning() bool { |
| 332 | return c.RequiresToolCallReasoning() && expectsDeepSeekToolCallReasoning(c.model, c.thinkingType) |
| 333 | } |
| 334 | |
| 335 | func (c *client) glmThinkingEnabled() bool { |
| 336 | if c == nil || !c.zhipu { |
| 337 | return false |
| 338 | } |
| 339 | t := c.effort |
| 340 | if c.thinkingType != "" { |
| 341 | t = c.thinkingType |
| 342 | } |
| 343 | return t != "disabled" |
| 344 | } |
| 345 | |
| 346 | func expectsDeepSeekToolCallReasoning(model, thinkingType string) bool { |
| 347 | if strings.EqualFold(strings.TrimSpace(thinkingType), "enabled") { |
| 348 | return true |
| 349 | } |
| 350 | model = strings.ToLower(strings.TrimSpace(model)) |
| 351 | return strings.Contains(model, "deepseek-v4-flash") || |
| 352 | strings.Contains(model, "deepseek-v4-pro") || |
| 353 | strings.Contains(model, "deepseek-v3.2") || |
| 354 | strings.Contains(model, "deepseek-reasoner") || |
| 355 | strings.Contains(model, "deepseek-r1") |
| 356 | } |
| 357 | |
| 358 | func (c *client) MissingToolCallReasoningWarningIdentity() string { |
| 359 | if c == nil { |
| 360 | return "" |
| 361 | } |
| 362 | protocol := "openai" |
| 363 | if c.deepseek { |
| 364 | protocol = "deepseek" |
| 365 | } |
| 366 | return strings.Join([]string{ |
| 367 | "openai", strings.TrimSpace(c.name), strings.TrimSpace(c.baseURL), |
| 368 | strings.TrimSpace(c.model), protocol, strings.TrimSpace(c.thinkingType), strings.TrimSpace(c.effort), |
| 369 | }, "\x00") |
| 370 | } |
| 371 | |
| 372 | func (c *client) sendOpts() provider.SendOptions { |
| 373 | return provider.SendOptions{ |
| 374 | Provider: c.name, |
| 375 | KeyEnv: c.keyEnv, |
| 376 | KeySource: c.keySource, |
| 377 | KeyPresent: c.apiKey != "", |
| 378 | RetryAuth: c.authed.Load(), |
| 379 | } |
| 380 | } |
| 381 | |
| 382 | func normalizeReasoningProtocol(raw string) string { |
| 383 | switch strings.ToLower(strings.TrimSpace(raw)) { |
| 384 | case "deepseek", "glm", "openai", "none": |
| 385 | return strings.ToLower(strings.TrimSpace(raw)) |
| 386 | default: |
| 387 | return "" |
| 388 | } |
| 389 | } |
| 390 | |
| 391 | func normalizeChatURL(baseURL, chatURL string) string { |
| 392 | if trimmed := strings.TrimRight(strings.TrimSpace(chatURL), "/"); trimmed != "" { |
| 393 | return trimmed |
| 394 | } |
| 395 | return strings.TrimRight(strings.TrimSpace(baseURL), "/") + "/chat/completions" |
| 396 | } |
| 397 | |
| 398 | func cleanCustomHeaders(in map[string]string) map[string]string { |
| 399 | if len(in) == 0 { |
| 400 | return nil |
| 401 | } |
| 402 | out := make(map[string]string, len(in)) |
| 403 | for rawName, rawValue := range in { |
| 404 | name := strings.TrimSpace(rawName) |
| 405 | value := strings.TrimSpace(rawValue) |
| 406 | if name == "" || value == "" || reservedCustomHeader(name) { |
| 407 | continue |
| 408 | } |
| 409 | out[name] = value |
| 410 | } |
| 411 | if len(out) == 0 { |
| 412 | return nil |
| 413 | } |
| 414 | return out |
| 415 | } |
| 416 | |
| 417 | func applyCustomHeaders(h http.Header, headers map[string]string) { |
| 418 | for name, value := range cleanCustomHeaders(headers) { |
| 419 | h.Set(name, value) |
| 420 | } |
| 421 | } |
| 422 | |
| 423 | func applyAPIKeyHeader(h http.Header, baseURL, apiKey string) { |
| 424 | apiKey = strings.TrimSpace(apiKey) |
| 425 | if apiKey == "" { |
| 426 | return |
| 427 | } |
| 428 | if IsMiMo(baseURL) { |
| 429 | h.Set("api-key", apiKey) |
| 430 | return |
| 431 | } |
| 432 | h.Set("Authorization", "Bearer "+apiKey) |
| 433 | } |
| 434 | |
| 435 | func cleanExtraBody(in map[string]any) map[string]any { |
| 436 | if len(in) == 0 { |
| 437 | return nil |
| 438 | } |
| 439 | out := make(map[string]any, len(in)) |
| 440 | for rawName, value := range in { |
| 441 | name := strings.TrimSpace(rawName) |
| 442 | if name == "" || reservedExtraBodyField(name) { |
| 443 | continue |
| 444 | } |
| 445 | out[name] = value |
| 446 | } |
| 447 | if len(out) == 0 { |
| 448 | return nil |
| 449 | } |
| 450 | return out |
| 451 | } |
| 452 | |
| 453 | func reservedExtraBodyField(name string) bool { |
| 454 | switch strings.ToLower(strings.TrimSpace(name)) { |
| 455 | case "model", "messages", "tools", "stream", "stream_options", "temperature", "max_tokens", "max_completion_tokens", "max_output_tokens", "reasoning_effort", "thinking": |
| 456 | return true |
| 457 | default: |
| 458 | return false |
| 459 | } |
| 460 | } |
| 461 | |
| 462 | func reservedCustomHeader(name string) bool { |
| 463 | switch strings.ToLower(strings.TrimSpace(name)) { |
| 464 | case "authorization", "content-type", "accept", "host": |
| 465 | return true |
| 466 | default: |
| 467 | return false |
| 468 | } |
| 469 | } |
| 470 | |
| 471 | // bufPool reuses byte buffers for JSON-marshalled request bodies. Each turn |
| 472 | // allocates a buffer, marshals the request, and sends it — pooling avoids the |
| 473 | // GC churn from repeated alloc/free of ~10-100KB buffers. The pool is |
| 474 | // provider-level (not global) so OpenAI and Anthropic don't compete. |
| 475 | var bufPool = sync.Pool{ |
| 476 | New: func() any { return new(bytes.Buffer) }, |
| 477 | } |
| 478 | |
| 479 | func (c *client) Stream(ctx context.Context, req provider.Request) (<-chan provider.Chunk, error) { |
| 480 | stream, err := c.openStream(ctx, c.chatURL, c.buildRequest(req), req.Tools) |
| 481 | if err != nil { |
| 482 | return nil, err |
| 483 | } |
| 484 | if c.prefixChatURL == "" { |
| 485 | return stream, nil |
| 486 | } |
| 487 | |
| 488 | out := make(chan provider.Chunk) |
| 489 | go c.streamWithPrefixContinuation(ctx, req, stream, out) |
| 490 | return out, nil |
| 491 | } |
| 492 | |
| 493 | func (c *client) openStream(ctx context.Context, targetURL string, wireReq chatRequest, tools []provider.ToolSchema) (<-chan provider.Chunk, error) { |
| 494 | requestCtx := provider.WithRequestAttemptCounter(ctx) |
| 495 | buf := bufPool.Get().(*bytes.Buffer) |
| 496 | buf.Reset() |
| 497 | if err := json.NewEncoder(buf).Encode(wireReq); err != nil { |
| 498 | bufPool.Put(buf) |
| 499 | return nil, fmt.Errorf("%s: marshal request: %w", c.name, err) |
| 500 | } |
| 501 | body := make([]byte, buf.Len()) |
| 502 | copy(body, buf.Bytes()) |
| 503 | bufPool.Put(buf) |
| 504 | |
| 505 | newReq := func(ctx context.Context) (*http.Request, error) { |
| 506 | httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(body)) |
| 507 | if err != nil { |
| 508 | return nil, err |
| 509 | } |
| 510 | httpReq.Header.Set("Content-Type", "application/json") |
| 511 | applyAPIKeyHeader(httpReq.Header, c.baseURL, c.apiKey) |
| 512 | httpReq.Header.Set("Accept", "text/event-stream") |
| 513 | applyCustomHeaders(httpReq.Header, c.headers) |
| 514 | return httpReq, nil |
| 515 | } |
| 516 | resp, err := provider.SendWithRetry(requestCtx, c.http, c.sendOpts(), newReq) |
| 517 | if err != nil { |
| 518 | return nil, provider.AnnotateToolSchemaError(err, tools) |
| 519 | } |
| 520 | c.authed.Store(true) |
| 521 | |
| 522 | out := make(chan provider.Chunk) |
| 523 | // Body-phase stream cuts surface as StreamInterruptedError so the Agent |
| 524 | // can replay the exact frozen request. Connection+header retries stay in |
| 525 | // SendWithRetry; providers must not stack a second body-retry budget. |
| 526 | go c.streamOnce(requestCtx, resp, out) |
| 527 | return out, nil |
| 528 | } |
| 529 | |
| 530 | // streamWithPrefixContinuation makes a DeepSeek Beta continuation look like one |
| 531 | // ordinary provider stream. Text/reasoning stays live, while usage is folded |
| 532 | // across both requests so cost and cache accounting remain truthful. If the |
| 533 | // Beta request fails before emitting anything, the original truncated response |
| 534 | // is kept and its finish_reason=length reaches the agent's existing warning. |
| 535 | func (c *client) streamWithPrefixContinuation(ctx context.Context, req provider.Request, current <-chan provider.Chunk, out chan<- provider.Chunk) { |
| 536 | defer close(out) |
| 537 | |
| 538 | var fullText, fullReasoning strings.Builder |
| 539 | var totalUsage *provider.Usage |
| 540 | continuations := 0 |
| 541 | |
| 542 | for { |
| 543 | var currentUsage *provider.Usage |
| 544 | currentHadTool := false |
| 545 | currentEmitted := false |
| 546 | |
| 547 | for chunk := range current { |
| 548 | switch chunk.Type { |
| 549 | case provider.ChunkText: |
| 550 | fullText.WriteString(chunk.Text) |
| 551 | currentEmitted = currentEmitted || chunk.Text != "" |
| 552 | if !sendChunk(ctx, out, chunk) { |
| 553 | return |
| 554 | } |
| 555 | case provider.ChunkReasoning: |
| 556 | fullReasoning.WriteString(chunk.Text) |
| 557 | currentEmitted = currentEmitted || chunk.Text != "" |
| 558 | if !sendChunk(ctx, out, chunk) { |
| 559 | return |
| 560 | } |
| 561 | case provider.ChunkToolCallStart, provider.ChunkToolCallArgsDelta, provider.ChunkToolCall: |
| 562 | currentHadTool = true |
| 563 | currentEmitted = true |
| 564 | if !sendChunk(ctx, out, chunk) { |
| 565 | return |
| 566 | } |
| 567 | case provider.ChunkUsage: |
| 568 | currentUsage = mergeUsage(currentUsage, chunk.Usage, false) |
| 569 | case provider.ChunkDone: |
| 570 | // The wrapper emits one final Done after any continuation. |
| 571 | case provider.ChunkError: |
| 572 | // A Beta failure before any continuation bytes is a safe fallback: |
| 573 | // the already-streamed first response remains visible and its |
| 574 | // length finish reason triggers the normal truncation warning. |
| 575 | if continuations > 0 && !currentEmitted && ctx.Err() == nil { |
| 576 | emitUsageAndDone(ctx, out, totalUsage) |
| 577 | return |
| 578 | } |
| 579 | _ = sendChunk(ctx, out, chunk) |
| 580 | return |
| 581 | default: |
| 582 | if !sendChunk(ctx, out, chunk) { |
| 583 | return |
| 584 | } |
| 585 | } |
| 586 | } |
| 587 | |
| 588 | totalUsage = mergeUsage(totalUsage, currentUsage, true) |
| 589 | if continuations >= maxPrefixContinuations || |
| 590 | currentUsage == nil || currentUsage.FinishReason != "length" || |
| 591 | currentHadTool || |
| 592 | (fullText.Len() == 0 && (c.thinkingType == "disabled" || fullReasoning.Len() == 0)) { |
| 593 | emitUsageAndDone(ctx, out, totalUsage) |
| 594 | return |
| 595 | } |
| 596 | |
| 597 | prefixReq := c.buildPrefixRequest(req, fullText.String(), fullReasoning.String()) |
| 598 | next, err := c.openStream(ctx, c.prefixChatURL, prefixReq, req.Tools) |
| 599 | if err != nil { |
| 600 | emitUsageAndDone(ctx, out, totalUsage) |
| 601 | return |
| 602 | } |
| 603 | continuations++ |
| 604 | current = next |
| 605 | } |
| 606 | } |
| 607 | |
| 608 | func emitUsageAndDone(ctx context.Context, out chan<- provider.Chunk, usage *provider.Usage) { |
| 609 | if usage != nil && !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkUsage, Usage: usage}) { |
| 610 | return |
| 611 | } |
| 612 | _ = sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkDone}) |
| 613 | } |
| 614 | |
| 615 | // mergeUsage folds token counters. countRequests is false for multiple usage |
| 616 | // chunks from one HTTP stream (keep its request count), and true when combining |
| 617 | // distinct prefix-continuation requests (sum their request counts). |
| 618 | func mergeUsage(total, next *provider.Usage, countRequests bool) *provider.Usage { |
| 619 | if next == nil { |
| 620 | return total |
| 621 | } |
| 622 | if total == nil { |
| 623 | clone := *next |
| 624 | return &clone |
| 625 | } |
| 626 | totalRequests := usageRequestCount(total) |
| 627 | nextRequests := usageRequestCount(next) |
| 628 | total.PromptTokens += next.PromptTokens |
| 629 | total.CompletionTokens += next.CompletionTokens |
| 630 | total.TotalTokens += next.TotalTokens |
| 631 | total.CacheHitTokens += next.CacheHitTokens |
| 632 | total.CacheMissTokens += next.CacheMissTokens |
| 633 | total.CacheWriteTokens += next.CacheWriteTokens |
| 634 | total.CacheWriteBilledTokens += next.CacheWriteBilledTokens |
| 635 | total.ReasoningTokens += next.ReasoningTokens |
| 636 | if countRequests { |
| 637 | total.RequestCount = totalRequests + nextRequests |
| 638 | } else if nextRequests > totalRequests { |
| 639 | total.RequestCount = nextRequests |
| 640 | } else { |
| 641 | total.RequestCount = totalRequests |
| 642 | } |
| 643 | total.FinishReason = next.FinishReason |
| 644 | return total |
| 645 | } |
| 646 | |
| 647 | func usageRequestCount(usage *provider.Usage) int { |
| 648 | if usage != nil && usage.RequestCount > 0 { |
| 649 | return usage.RequestCount |
| 650 | } |
| 651 | return 1 |
| 652 | } |
| 653 | |
| 654 | // streamOnce drives a single body read. Mid-stream transport cuts become |
| 655 | // StreamInterruptedError so the Agent can commit-or-replay; providers no longer |
| 656 | // replay the body themselves (that would stack retry budgets with the Agent). |
| 657 | func (c *client) streamOnce(ctx context.Context, resp *http.Response, out chan<- provider.Chunk) { |
| 658 | defer close(out) |
| 659 | _, err := c.readStream(ctx, resp, out) |
| 660 | if err == nil { |
| 661 | return |
| 662 | } |
| 663 | if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { |
| 664 | sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkError, Err: err}) |
| 665 | return |
| 666 | } |
| 667 | if provider.IsConnReset(err) { |
| 668 | sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkError, Err: provider.StreamInterrupt(err, provider.ClassifyStreamInterrupt(err))}) |
| 669 | return |
| 670 | } |
| 671 | sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkError, Err: err}) |
| 672 | } |
| 673 | |
| 674 | func sendChunk(ctx context.Context, out chan<- provider.Chunk, chunk provider.Chunk) bool { |
| 675 | select { |
| 676 | case out <- chunk: |
| 677 | return true |
| 678 | default: |
| 679 | } |
| 680 | select { |
| 681 | case <-ctx.Done(): |
| 682 | return false |
| 683 | case out <- chunk: |
| 684 | return true |
| 685 | } |
| 686 | } |
| 687 | |
| 688 | func (c *client) buildRequest(req provider.Request) chatRequest { |
| 689 | // Repair tool-call pairing before sending: an interrupted/resumed history can |
| 690 | // carry an assistant tool_calls turn whose results never landed, which DeepSeek |
| 691 | // rejects with a 400 ("must be followed by tool messages …"). |
| 692 | src := provider.SanitizeToolPairing(req.Messages) |
| 693 | msgs := make([]chatMessage, 0, len(src)) |
| 694 | // Images returned by tool calls can't ride in the tool message itself — the |
| 695 | // OpenAI API accepts only text content parts under role "tool" — so they are |
| 696 | // carried by a synthetic user message injected after the turn's full run of |
| 697 | // tool results, before the next non-tool message (splitting a tool-result |
| 698 | // run would break the API's tool-call pairing validation). |
| 699 | var pendingToolImages []string |
| 700 | flushToolImages := func() { |
| 701 | if len(pendingToolImages) == 0 { |
| 702 | return |
| 703 | } |
| 704 | msgs = append(msgs, chatMessage{ |
| 705 | Role: "user", |
| 706 | Content: imageContentParts("Images returned by the preceding tool call(s):", pendingToolImages, c.visionDetail), |
| 707 | }) |
| 708 | pendingToolImages = nil |
| 709 | } |
| 710 | for _, m := range src { |
| 711 | if m.Role != provider.RoleTool { |
| 712 | flushToolImages() |
| 713 | } |
| 714 | cm := chatMessage{ |
| 715 | Role: string(m.Role), |
| 716 | ToolCallID: m.ToolCallID, |
| 717 | } |
| 718 | if m.Role == provider.RoleTool { |
| 719 | // Always send the tool message's name, even when empty: strict |
| 720 | // backends (MiMo) 400 a tool result without the key (#4711). |
| 721 | name := m.Name |
| 722 | cm.Name = &name |
| 723 | } |
| 724 | // DeepSeek thinking mode 400s an assistant tool_calls turn whose |
| 725 | // reasoning_content KEY is absent from the request JSON ("reasoning_content |
| 726 | // … must be passed back"). The API accepts an empty string, and only |
| 727 | // validates turns after the last user message, but emitting the field on |
| 728 | // every tool_calls turn is uniform and verified accepted — so always send |
| 729 | // it (empty included) rather than fail the request when reasoning was lost |
| 730 | // upstream (e.g. a gateway renamed the field). With thinking disabled the |
| 731 | // API tolerates every shape, so keep the exact pre-fix bytes there: send |
| 732 | // the key only when a thinking-mode round left reasoning in the history |
| 733 | // (dropping it would invalidate the prompt-cache prefix of mixed |
| 734 | // thinking-on→off sessions for no gain). |
| 735 | if m.Role == provider.RoleAssistant { |
| 736 | switch { |
| 737 | case c.kimiK3 && (m.ReasoningContent != "" || len(m.ToolCalls) > 0): |
| 738 | // Kimi K3 requires the complete assistant message on multi-turn |
| 739 | // and tool-call requests, including provider-issued reasoning. |
| 740 | cm.ReasoningContent = &m.ReasoningContent |
| 741 | case c.deepseek && len(m.ToolCalls) > 0: |
| 742 | if c.RequiresToolCallReasoning() || m.ReasoningContent != "" { |
| 743 | cm.ReasoningContent = &m.ReasoningContent |
| 744 | } |
| 745 | case c.zhipu && m.ReasoningContent != "": |
| 746 | // GLM interleaved and preserved thinking require provider-issued |
| 747 | // reasoning content to be returned unchanged in later history. Keep |
| 748 | // an existing value even after thinking is turned off so an |
| 749 | // enabled→disabled session retains its valid history bytes. |
| 750 | cm.ReasoningContent = &m.ReasoningContent |
| 751 | } |
| 752 | } |
| 753 | for _, tc := range m.ToolCalls { |
| 754 | wire := chatToolCall{ID: tc.ID, Type: "function"} |
| 755 | wire.Function.Name = tc.Name |
| 756 | wire.Function.Arguments = tc.Arguments |
| 757 | if tc.ThoughtSignature != "" && usesGeminiThoughtSignatures(c.baseURL, c.model) { |
| 758 | // Gemini's current OpenAI compatibility schema carries the |
| 759 | // opaque signature beside the function payload. Keep the |
| 760 | // legacy function.thought_signature field decode-only below so |
| 761 | // older gateways remain readable without sending an unknown |
| 762 | // function parameter to current Google endpoints. |
| 763 | wire.ExtraContent = &chatToolCallExtraContent{} |
| 764 | wire.ExtraContent.Google.ThoughtSignature = tc.ThoughtSignature |
| 765 | } |
| 766 | cm.ToolCalls = append(cm.ToolCalls, wire) |
| 767 | } |
| 768 | switch { |
| 769 | case c.vision && m.Role == provider.RoleUser && len(m.Images) > 0: |
| 770 | cm.Content = imageContentParts(m.Content, m.Images, c.visionDetail) |
| 771 | case m.Role != provider.RoleAssistant || len(cm.ToolCalls) == 0 || m.Content != "": |
| 772 | cm.Content = m.Content |
| 773 | } |
| 774 | msgs = append(msgs, cm) |
| 775 | if c.vision && m.Role == provider.RoleTool { |
| 776 | pendingToolImages = append(pendingToolImages, m.Images...) |
| 777 | } |
| 778 | } |
| 779 | flushToolImages() |
| 780 | |
| 781 | var tools []chatTool |
| 782 | for _, t := range req.Tools { |
| 783 | parameters := t.Parameters |
| 784 | if len(parameters) == 0 { |
| 785 | parameters = provider.CanonicalizeSchema(nil) |
| 786 | } |
| 787 | if c.mimo { |
| 788 | parameters = provider.NormalizeLegacyTupleItemsForDraft202012(parameters) |
| 789 | } |
| 790 | tools = append(tools, chatTool{ |
| 791 | Type: "function", |
| 792 | Function: chatFunction{Name: t.Name, Description: t.Description, Parameters: parameters}, |
| 793 | }) |
| 794 | } |
| 795 | |
| 796 | maxOutputTokens := req.MaxTokens |
| 797 | if maxOutputTokens == 0 { |
| 798 | maxOutputTokens = c.maxOutputTokens |
| 799 | } |
| 800 | if maxOutputTokens < 0 { |
| 801 | maxOutputTokens = 0 |
| 802 | } |
| 803 | out := chatRequest{ |
| 804 | Model: c.model, |
| 805 | Messages: msgs, |
| 806 | Tools: tools, |
| 807 | Stream: true, |
| 808 | StreamOptions: &streamOptions{IncludeUsage: true}, |
| 809 | Temperature: req.Temperature, |
| 810 | MaxTokens: maxOutputTokens, |
| 811 | ReasoningEffort: c.effort, |
| 812 | ExtraBody: c.extraBody, |
| 813 | } |
| 814 | switch { |
| 815 | case c.kimiK3: |
| 816 | // K3 fixes its sampling values and recommends omitting them. It also |
| 817 | // names the output budget max_completion_tokens rather than max_tokens. |
| 818 | out.Temperature = nil |
| 819 | out.MaxTokens = 0 |
| 820 | out.MaxCompletionTokens = maxOutputTokens |
| 821 | out.ExtraBody = omitExtraBodyFields(out.ExtraBody, |
| 822 | "temperature", "top_p", "n", "presence_penalty", "frequency_penalty", "max_completion_tokens") |
| 823 | case IsOpenAI(c.baseURL): |
| 824 | // OpenAI's current Chat Completions contract replaces max_tokens with |
| 825 | // max_completion_tokens, which includes visible and reasoning tokens and |
| 826 | // is required by o-series models. Compatible gateways retain max_tokens. |
| 827 | out.MaxTokens = 0 |
| 828 | out.MaxCompletionTokens = maxOutputTokens |
| 829 | case c.deepseek: |
| 830 | // DeepSeek's CoT is controlled by `thinking` plus `reasoning_effort` for |
| 831 | // depth. Thinking is on by default but can be turned off via |
| 832 | // effort=disabled / thinking=disabled (credit @eghrhegpe, #5063). |
| 833 | if c.thinkingType == "disabled" { |
| 834 | out.Thinking = &thinkingMode{Type: "disabled"} |
| 835 | } else { |
| 836 | out.Thinking = &thinkingMode{Type: "enabled"} |
| 837 | } |
| 838 | case c.minimax: |
| 839 | // M3 uses a single `thinking.type` field with two valid values: |
| 840 | // "adaptive" (default, thinking on) and "disabled" (off). Reasoning |
| 841 | // depth is not a knob on M3, so reasoning_effort is omitted entirely. |
| 842 | t := c.effort |
| 843 | if t == "" { |
| 844 | t = "adaptive" // /effort auto == the M3 model default |
| 845 | } |
| 846 | out.Thinking = &thinkingMode{Type: t} |
| 847 | out.ReasoningEffort = "" |
| 848 | case c.zhipu: |
| 849 | // Zhipu GLM's binary thinking knob: "enabled" (default, thinking on) or |
| 850 | // "disabled". reasoning_effort is silently ignored by the endpoint, so we |
| 851 | // omit it and drive chain-of-thought purely through thinking.type. |
| 852 | t := c.effort |
| 853 | if t == "" { |
| 854 | t = "enabled" // auto == the GLM default (thinking on) |
| 855 | } |
| 856 | if c.thinkingType != "" { |
| 857 | t = c.thinkingType // explicit `thinking` config overrides the effort knob |
| 858 | } |
| 859 | out.Thinking = &thinkingMode{Type: t} |
| 860 | out.ReasoningEffort = "" |
| 861 | case c.longcat: |
| 862 | // LongCat's binary thinking knob: "enabled" (default, thinking on) or |
| 863 | // "disabled". The API documents reasoning_content in OpenAI responses but |
| 864 | // not reasoning_effort, so keep depth out of the request. |
| 865 | t := c.effort |
| 866 | if t == "" { |
| 867 | t = c.thinkingType |
| 868 | } |
| 869 | if t == "" { |
| 870 | t = "enabled" |
| 871 | } |
| 872 | out.Thinking = &thinkingMode{Type: t} |
| 873 | out.ReasoningEffort = "" |
| 874 | case c.thinkingType != "": |
| 875 | // Generic OpenAI-compatible provider with an explicit `thinking` config |
| 876 | // field (e.g. opencode.ai) — emit thinking.type; reasoning_effort, if any, |
| 877 | // is left untouched for backends that also honour it. |
| 878 | out.Thinking = &thinkingMode{Type: c.thinkingType} |
| 879 | } |
| 880 | return out |
| 881 | } |
| 882 | |
| 883 | func (c *client) buildPrefixRequest(req provider.Request, content, reasoning string) chatRequest { |
| 884 | out := c.buildRequest(req) |
| 885 | prefix := chatMessage{Role: "assistant", Content: content, Prefix: true} |
| 886 | if c.deepseek && c.thinkingType != "disabled" { |
| 887 | prefix.ReasoningContent = &reasoning |
| 888 | } |
| 889 | out.Messages = append(out.Messages, prefix) |
| 890 | return out |
| 891 | } |
| 892 | |
| 893 | // readStream parses one SSE response into chunks: text deltas stream live, |
| 894 | // tool-call fragments accumulate by index and emit complete on [DONE], and a |
| 895 | // ChunkToolCallStart fires the moment a call's name is known. It returns whether |
| 896 | // any model output was forwarded (so the caller can decide a replay is safe) and |
| 897 | // the first fatal error — a nil error means the stream reached [DONE]. |
| 898 | func (c *client) readStream(ctx context.Context, resp *http.Response, out chan<- provider.Chunk) (emitted bool, _ error) { |
| 899 | defer resp.Body.Close() |
| 900 | |
| 901 | // Close the response body when the context is canceled (user interrupt) or the |
| 902 | // stream stalls past c.idleTimeout, so scanner.Scan() unblocks instead of |
| 903 | // hanging on a half-open connection. done lets the watchdog exit on a normal |
| 904 | // return — otherwise it outlives the call and blocks forever on a non-cancellable |
| 905 | // context whose Done() is nil. The watchdog owns the timer; the read loop only |
| 906 | // pings the buffered activity channel, so there's no Timer.Reset race. |
| 907 | idleTimeout := c.idleTimeout |
| 908 | if idleTimeout <= 0 { // zero-value client (constructed without New) |
| 909 | idleTimeout = defaultStreamIdleTimeout |
| 910 | } |
| 911 | done := make(chan struct{}) |
| 912 | defer close(done) |
| 913 | activity := make(chan struct{}, 1) |
| 914 | var stalled atomic.Bool |
| 915 | go func() { |
| 916 | idle := time.NewTimer(idleTimeout) |
| 917 | defer idle.Stop() |
| 918 | for { |
| 919 | select { |
| 920 | case <-ctx.Done(): |
| 921 | resp.Body.Close() |
| 922 | return |
| 923 | case <-idle.C: |
| 924 | stalled.Store(true) |
| 925 | resp.Body.Close() |
| 926 | return |
| 927 | case <-activity: |
| 928 | if !idle.Stop() { |
| 929 | select { |
| 930 | case <-idle.C: |
| 931 | default: |
| 932 | } |
| 933 | } |
| 934 | idle.Reset(idleTimeout) |
| 935 | case <-done: |
| 936 | return |
| 937 | } |
| 938 | } |
| 939 | }() |
| 940 | |
| 941 | acc := map[int]*provider.ToolCall{} |
| 942 | started := map[int]bool{} |
| 943 | argBucket := map[int]int{} |
| 944 | var order []int |
| 945 | var lastFinishReason string |
| 946 | var sawDone bool |
| 947 | var think thinkSplitter |
| 948 | |
| 949 | scanner := bufio.NewScanner(resp.Body) |
| 950 | scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) |
| 951 | |
| 952 | for scanner.Scan() { |
| 953 | select { // ping the idle watchdog; non-blocking so a full buffer is fine |
| 954 | case activity <- struct{}{}: |
| 955 | default: |
| 956 | } |
| 957 | line := strings.TrimSpace(scanner.Text()) |
| 958 | if line == "" || !strings.HasPrefix(line, "data:") { |
| 959 | continue |
| 960 | } |
| 961 | data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) |
| 962 | if data == "[DONE]" { |
| 963 | sawDone = true |
| 964 | break |
| 965 | } |
| 966 | if data == "" { |
| 967 | continue |
| 968 | } |
| 969 | |
| 970 | var sr streamResponse |
| 971 | if err := json.Unmarshal([]byte(data), &sr); err != nil { |
| 972 | return emitted, provider.StreamDecodeError(c.name, data, err) |
| 973 | } |
| 974 | if sr.Error != nil { |
| 975 | return emitted, fmt.Errorf("%s: %s", c.name, sr.Error.Message) |
| 976 | } |
| 977 | if len(sr.Choices) > 0 && sr.Choices[0].FinishReason != nil && *sr.Choices[0].FinishReason != "" { |
| 978 | lastFinishReason = *sr.Choices[0].FinishReason |
| 979 | } |
| 980 | if sr.Usage != nil { |
| 981 | u := normaliseUsage(sr.Usage) |
| 982 | u.FinishReason = lastFinishReason |
| 983 | provider.ApplyRequestAttemptCount(ctx, u) |
| 984 | emitted = true |
| 985 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkUsage, Usage: u}) { |
| 986 | return emitted, ctx.Err() |
| 987 | } |
| 988 | } |
| 989 | if len(sr.Choices) == 0 { |
| 990 | continue |
| 991 | } |
| 992 | |
| 993 | delta := sr.Choices[0].Delta |
| 994 | reasoningDelta := delta.ReasoningContent |
| 995 | if reasoningDelta == "" { |
| 996 | reasoningDelta = delta.Reasoning |
| 997 | } |
| 998 | if reasoningDelta != "" { |
| 999 | emitted = true |
| 1000 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkReasoning, Text: reasoningDelta}) { |
| 1001 | return emitted, ctx.Err() |
| 1002 | } |
| 1003 | } |
| 1004 | if delta.Content != "" { |
| 1005 | r, txt := think.push(delta.Content) |
| 1006 | if r != "" { |
| 1007 | emitted = true |
| 1008 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkReasoning, Text: r}) { |
| 1009 | return emitted, ctx.Err() |
| 1010 | } |
| 1011 | } |
| 1012 | if txt != "" { |
| 1013 | emitted = true |
| 1014 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkText, Text: txt}) { |
| 1015 | return emitted, ctx.Err() |
| 1016 | } |
| 1017 | } |
| 1018 | } |
| 1019 | for _, tc := range delta.ToolCalls { |
| 1020 | cur, ok := acc[tc.Index] |
| 1021 | if !ok { |
| 1022 | cur = &provider.ToolCall{} |
| 1023 | acc[tc.Index] = cur |
| 1024 | order = append(order, tc.Index) |
| 1025 | } |
| 1026 | if tc.ID != "" { |
| 1027 | cur.ID = tc.ID |
| 1028 | } |
| 1029 | if tc.Function.Name != "" { |
| 1030 | cur.Name = tc.Function.Name |
| 1031 | } |
| 1032 | cur.Arguments += tc.Function.Arguments |
| 1033 | thoughtSignature := "" |
| 1034 | if tc.ExtraContent != nil { |
| 1035 | thoughtSignature = tc.ExtraContent.Google.ThoughtSignature |
| 1036 | } |
| 1037 | if thoughtSignature == "" { |
| 1038 | // Early Gemini OpenAI-compatible responses placed the field in |
| 1039 | // function. Accept that shape when replaying older sessions and |
| 1040 | // when talking to compatibility gateways that still emit it. |
| 1041 | thoughtSignature = tc.Function.ThoughtSignature |
| 1042 | } |
| 1043 | if thoughtSignature != "" { |
| 1044 | cur.ThoughtSignature = thoughtSignature |
| 1045 | } |
| 1046 | // Signal the call's start the moment its name is known, so a frontend |
| 1047 | // can show the tool card immediately rather than only after its |
| 1048 | // (possibly large) arguments finish streaming. |
| 1049 | if !started[tc.Index] && cur.Name != "" { |
| 1050 | started[tc.Index] = true |
| 1051 | emitted = true |
| 1052 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkToolCallStart, ToolCall: &provider.ToolCall{ID: cur.ID, Name: cur.Name}}) { |
| 1053 | return emitted, ctx.Err() |
| 1054 | } |
| 1055 | } |
| 1056 | // Progress ticks while a large argument payload streams (a 30KB |
| 1057 | // write_file body can take a minute-plus): one chunk per 2KB bucket |
| 1058 | // so the consumer can show liveness without per-delta spam. |
| 1059 | if started[tc.Index] { |
| 1060 | if bucket := len(cur.Arguments) / 2048; bucket > argBucket[tc.Index] { |
| 1061 | argBucket[tc.Index] = bucket |
| 1062 | emitted = true |
| 1063 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkToolCallArgsDelta, ToolCall: &provider.ToolCall{ID: cur.ID, Name: cur.Name}, ArgChars: len(cur.Arguments)}) { |
| 1064 | return emitted, ctx.Err() |
| 1065 | } |
| 1066 | } |
| 1067 | } |
| 1068 | } |
| 1069 | } |
| 1070 | |
| 1071 | if err := ctx.Err(); err != nil { |
| 1072 | return emitted, err |
| 1073 | } |
| 1074 | if stalled.Load() { |
| 1075 | // Idle stall is a body-phase cut: wrap so the Agent can replay the |
| 1076 | // frozen request. Providers no longer reconnect here. |
| 1077 | return emitted, fmt.Errorf("%s: stream stalled — no data for %s, connection likely dropped: %w", c.name, idleTimeout, io.ErrUnexpectedEOF) |
| 1078 | } |
| 1079 | if err := scanner.Err(); err != nil { |
| 1080 | return emitted, fmt.Errorf("%s: read stream: %w", c.name, err) |
| 1081 | } |
| 1082 | // A proxy that idle-closes with a clean FIN ends the scan with no error. Without |
| 1083 | // this check the turn would be committed as complete — including half-streamed |
| 1084 | // tool-call arguments, which then 400 on every replay (#3953). OpenAI Chat |
| 1085 | // accepts either [DONE] or a legal finish_reason as a complete terminal. |
| 1086 | if !sawDone && lastFinishReason == "" { |
| 1087 | return emitted, fmt.Errorf("%s: stream ended before completion: %w", c.name, io.ErrUnexpectedEOF) |
| 1088 | } |
| 1089 | |
| 1090 | if r, txt := think.flush(); r != "" || txt != "" { |
| 1091 | if r != "" { |
| 1092 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkReasoning, Text: r}) { |
| 1093 | return emitted, ctx.Err() |
| 1094 | } |
| 1095 | } |
| 1096 | if txt != "" { |
| 1097 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkText, Text: txt}) { |
| 1098 | return emitted, ctx.Err() |
| 1099 | } |
| 1100 | } |
| 1101 | } |
| 1102 | |
| 1103 | sort.Ints(order) |
| 1104 | for _, idx := range order { |
| 1105 | tc := acc[idx] |
| 1106 | if tc.ID == "" { |
| 1107 | // Some OpenAI-compatible gateways stream tool calls by index with no id. |
| 1108 | // Synthesize a stable one so the result can be paired back to its call — |
| 1109 | // an empty tool_call_id collapses multi-tool turns downstream. |
| 1110 | tc.ID = fmt.Sprintf("call_%d", idx) |
| 1111 | } |
| 1112 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkToolCall, ToolCall: tc}) { |
| 1113 | return emitted, ctx.Err() |
| 1114 | } |
| 1115 | } |
| 1116 | if !sendChunk(ctx, out, provider.Chunk{Type: provider.ChunkDone}) { |
| 1117 | return emitted, ctx.Err() |
| 1118 | } |
| 1119 | return emitted, nil |
| 1120 | } |
| 1121 | |
| 1122 | // normaliseUsage folds the cache shapes used by OpenAI-compatible providers into |
| 1123 | // a single Usage. DeepSeek reports prompt_cache_{hit,miss}_tokens at the top of |
| 1124 | // usage; OpenAI and MiMo put cache hits under prompt_tokens_details; some |
| 1125 | // compatible gateways return Anthropic-style input/cache counters instead. |
| 1126 | // Reasoning tokens land in completion_tokens_details on thinking-mode models. |
| 1127 | func normaliseUsage(u *wireUsage) *provider.Usage { |
| 1128 | prompt := u.PromptTokens |
| 1129 | anthropicPrompt := prompt == 0 && |
| 1130 | (u.InputTokens != 0 || u.CacheCreationInputTokens != 0 || u.CacheReadInputTokens != 0) |
| 1131 | if anthropicPrompt { |
| 1132 | // Anthropic-style input_tokens excludes both cache reads and cache |
| 1133 | // writes, while Reasonix PromptTokens represents the complete input. |
| 1134 | prompt = u.InputTokens + u.CacheCreationInputTokens + u.CacheReadInputTokens |
| 1135 | } |
| 1136 | completion := u.CompletionTokens |
| 1137 | if completion == 0 { |
| 1138 | completion = u.OutputTokens |
| 1139 | } |
| 1140 | total := u.TotalTokens |
| 1141 | if total == 0 && (prompt != 0 || completion != 0) { |
| 1142 | total = prompt + completion |
| 1143 | } |
| 1144 | |
| 1145 | hit := u.PromptCacheHitTokens |
| 1146 | miss := u.PromptCacheMissTokens |
| 1147 | if hit == 0 && u.PromptTokensDetails != nil { |
| 1148 | hit = u.PromptTokensDetails.CachedTokens |
| 1149 | } |
| 1150 | if hit == 0 { |
| 1151 | hit = u.CacheReadInputTokens |
| 1152 | } |
| 1153 | if miss == 0 { |
| 1154 | switch { |
| 1155 | case anthropicPrompt: |
| 1156 | // Cache writes are still uncached input for Reasonix pricing and |
| 1157 | // cache-ratio accounting. |
| 1158 | miss = u.InputTokens + u.CacheCreationInputTokens |
| 1159 | case hit > 0 && prompt > hit: |
| 1160 | miss = prompt - hit |
| 1161 | } |
| 1162 | } |
| 1163 | reasoning := 0 |
| 1164 | if u.CompletionTokensDetails != nil { |
| 1165 | reasoning = u.CompletionTokensDetails.ReasoningTokens |
| 1166 | } |
| 1167 | return &provider.Usage{ |
| 1168 | PromptTokens: prompt, |
| 1169 | CompletionTokens: completion, |
| 1170 | TotalTokens: total, |
| 1171 | CacheHitTokens: hit, |
| 1172 | CacheMissTokens: miss, |
| 1173 | ReasoningTokens: reasoning, |
| 1174 | } |
| 1175 | } |
| 1176 | |
| 1177 | // --- OpenAI-compatible wire protocol --- |
| 1178 | |
| 1179 | type chatRequest struct { |
| 1180 | Model string `json:"model"` |
| 1181 | Messages []chatMessage `json:"messages"` |
| 1182 | Tools []chatTool `json:"tools,omitempty"` |
| 1183 | Stream bool `json:"stream"` |
| 1184 | StreamOptions *streamOptions `json:"stream_options,omitempty"` |
| 1185 | Temperature *float64 `json:"temperature,omitempty"` |
| 1186 | MaxTokens int `json:"max_tokens,omitempty"` |
| 1187 | MaxCompletionTokens int `json:"max_completion_tokens,omitempty"` |
| 1188 | ReasoningEffort string `json:"reasoning_effort,omitempty"` |
| 1189 | Thinking *thinkingMode `json:"thinking,omitempty"` |
| 1190 | ExtraBody map[string]any `json:"-"` |
| 1191 | } |
| 1192 | |
| 1193 | func omitExtraBodyFields(in map[string]any, names ...string) map[string]any { |
| 1194 | if len(in) == 0 { |
| 1195 | return nil |
| 1196 | } |
| 1197 | omit := make(map[string]struct{}, len(names)) |
| 1198 | for _, name := range names { |
| 1199 | omit[strings.ToLower(strings.TrimSpace(name))] = struct{}{} |
| 1200 | } |
| 1201 | out := make(map[string]any, len(in)) |
| 1202 | for name, value := range in { |
| 1203 | if _, blocked := omit[strings.ToLower(strings.TrimSpace(name))]; !blocked { |
| 1204 | out[name] = value |
| 1205 | } |
| 1206 | } |
| 1207 | if len(out) == 0 { |
| 1208 | return nil |
| 1209 | } |
| 1210 | return out |
| 1211 | } |
| 1212 | |
| 1213 | func (r chatRequest) MarshalJSON() ([]byte, error) { |
| 1214 | type wire chatRequest |
| 1215 | baseReq := wire(r) |
| 1216 | baseReq.ExtraBody = nil |
| 1217 | raw, err := json.Marshal(baseReq) |
| 1218 | if err != nil { |
| 1219 | return nil, err |
| 1220 | } |
| 1221 | if len(r.ExtraBody) == 0 { |
| 1222 | return raw, nil |
| 1223 | } |
| 1224 | var body map[string]any |
| 1225 | if err := json.Unmarshal(raw, &body); err != nil { |
| 1226 | return nil, err |
| 1227 | } |
| 1228 | for key, value := range cleanExtraBody(r.ExtraBody) { |
| 1229 | body[key] = value |
| 1230 | } |
| 1231 | return json.Marshal(body) |
| 1232 | } |
| 1233 | |
| 1234 | type thinkingMode struct { |
| 1235 | Type string `json:"type"` |
| 1236 | } |
| 1237 | |
| 1238 | type streamOptions struct { |
| 1239 | IncludeUsage bool `json:"include_usage"` |
| 1240 | } |
| 1241 | |
| 1242 | type chatMessage struct { |
| 1243 | Role string `json:"role"` |
| 1244 | // content is always present (never omitted): DeepSeek's strict deserializer |
| 1245 | // rejects a message missing the field. A pure tool_calls assistant turn |
| 1246 | // serializes as null (nil here); a string for every other text message |
| 1247 | // (empty included — null is rejected by some backends for a tool message); |
| 1248 | // and a []chatContentPart array for a vision user turn carrying images. |
| 1249 | Content any `json:"content"` |
| 1250 | // Prefix is wire-only and is set exclusively on an automatically recovered |
| 1251 | // DeepSeek assistant tail. omitempty keeps every ordinary request byte-stable. |
| 1252 | Prefix bool `json:"prefix,omitempty"` |
| 1253 | // A pointer so the field can serialize as an empty string: DeepSeek thinking |
| 1254 | // mode requires the reasoning_content key to be PRESENT on assistant |
| 1255 | // tool_calls turns (an empty value passes; a missing key 400s), while every |
| 1256 | // other message must keep omitting it. |
| 1257 | ReasoningContent *string `json:"reasoning_content,omitempty"` |
| 1258 | ToolCalls []chatToolCall `json:"tool_calls,omitempty"` |
| 1259 | ToolCallID string `json:"tool_call_id,omitempty"` |
| 1260 | // Name is the role=tool message's function name. A pointer so ordinary |
| 1261 | // messages omit the key (byte-stable prefix), while tool messages always |
| 1262 | // serialize it — even empty: strict OpenAI-compatible backends (MiMo, per |
| 1263 | // its error table) reject a tool message whose `name` key is absent |
| 1264 | // ("name is not set"), and OpenAI's spec requires the field on role=tool. |
| 1265 | Name *string `json:"name,omitempty"` |
| 1266 | } |
| 1267 | |
| 1268 | type chatContentPart struct { |
| 1269 | Type string `json:"type"` |
| 1270 | Text string `json:"text,omitempty"` |
| 1271 | ImageURL *chatImageURL `json:"image_url,omitempty"` |
| 1272 | } |
| 1273 | |
| 1274 | type chatImageURL struct { |
| 1275 | URL string `json:"url"` |
| 1276 | Detail string `json:"detail,omitempty"` |
| 1277 | } |
| 1278 | |
| 1279 | func imageContentParts(text string, images []string, detail string) []chatContentPart { |
| 1280 | parts := make([]chatContentPart, 0, len(images)+1) |
| 1281 | if text != "" { |
| 1282 | parts = append(parts, chatContentPart{Type: "text", Text: text}) |
| 1283 | } |
| 1284 | for _, url := range images { |
| 1285 | parts = append(parts, chatContentPart{Type: "image_url", ImageURL: &chatImageURL{URL: url, Detail: detail}}) |
| 1286 | } |
| 1287 | return parts |
| 1288 | } |
| 1289 | |
| 1290 | type chatTool struct { |
| 1291 | Type string `json:"type"` |
| 1292 | Function chatFunction `json:"function"` |
| 1293 | } |
| 1294 | |
| 1295 | type chatFunction struct { |
| 1296 | Name string `json:"name"` |
| 1297 | Description string `json:"description,omitempty"` |
| 1298 | Parameters json.RawMessage `json:"parameters,omitempty"` |
| 1299 | } |
| 1300 | |
| 1301 | type chatToolCall struct { |
| 1302 | Index int `json:"index,omitempty"` |
| 1303 | ID string `json:"id,omitempty"` |
| 1304 | Type string `json:"type,omitempty"` |
| 1305 | ExtraContent *chatToolCallExtraContent `json:"extra_content,omitempty"` |
| 1306 | Function struct { |
| 1307 | Name string `json:"name"` |
| 1308 | Arguments string `json:"arguments"` |
| 1309 | // Decode compatibility for the early Gemini OpenAI shape. New requests |
| 1310 | // use extra_content.google.thought_signature. |
| 1311 | ThoughtSignature string `json:"thought_signature,omitempty"` |
| 1312 | } `json:"function"` |
| 1313 | } |
| 1314 | |
| 1315 | type chatToolCallExtraContent struct { |
| 1316 | Google struct { |
| 1317 | ThoughtSignature string `json:"thought_signature,omitempty"` |
| 1318 | } `json:"google"` |
| 1319 | } |
| 1320 | |
| 1321 | type streamResponse struct { |
| 1322 | Choices []struct { |
| 1323 | Delta struct { |
| 1324 | Content string `json:"content"` |
| 1325 | ReasoningContent string `json:"reasoning_content"` |
| 1326 | Reasoning string `json:"reasoning"` |
| 1327 | ToolCalls []chatToolCall `json:"tool_calls"` |
| 1328 | } `json:"delta"` |
| 1329 | FinishReason *string `json:"finish_reason"` |
| 1330 | } `json:"choices"` |
| 1331 | Usage *wireUsage `json:"usage"` |
| 1332 | Error *struct { |
| 1333 | Message string `json:"message"` |
| 1334 | } `json:"error"` |
| 1335 | } |
| 1336 | |
| 1337 | // wireUsage covers DeepSeek's top-level cache fields, OpenAI/MiMo's nested |
| 1338 | // details, and Anthropic-style fallbacks returned by compatible gateways. |
| 1339 | type wireUsage struct { |
| 1340 | PromptTokens int `json:"prompt_tokens"` |
| 1341 | CompletionTokens int `json:"completion_tokens"` |
| 1342 | TotalTokens int `json:"total_tokens"` |
| 1343 | InputTokens int `json:"input_tokens"` |
| 1344 | OutputTokens int `json:"output_tokens"` |
| 1345 | PromptCacheHitTokens int `json:"prompt_cache_hit_tokens"` |
| 1346 | PromptCacheMissTokens int `json:"prompt_cache_miss_tokens"` |
| 1347 | CacheCreationInputTokens int `json:"cache_creation_input_tokens"` |
| 1348 | CacheReadInputTokens int `json:"cache_read_input_tokens"` |
| 1349 | PromptTokensDetails *struct { |
| 1350 | CachedTokens int `json:"cached_tokens"` |
| 1351 | } `json:"prompt_tokens_details"` |
| 1352 | CompletionTokensDetails *struct { |
| 1353 | ReasoningTokens int `json:"reasoning_tokens"` |
| 1354 | } `json:"completion_tokens_details"` |
| 1355 | } |
| 1356 |