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