package providers import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "log/slog" "strings" ) func (p *OpenAIProvider) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) { model := p.resolveModel(req.Model) body := p.buildRequestBody(model, req, false) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(model, req)) chatFn := p.chatRequestFn(ctx, body, req.Tools) resp, err := RetryDo(ctx, p.retryConfig, chatFn) // Auto-clamp max_tokens and retry once if the model rejects the value if err != nil { if clamped := clampMaxTokensFromError(err, body); clamped { slog.Info("max_tokens clamped, retrying", "model", model, "limit", clampedLimit(body)) resp, err = RetryDo(ctx, p.retryConfig, chatFn) } } // Drop user-visible reasoning for models flagged as leakers (e.g. Kimi, // DeepSeek-Reasoner). Usage.ThinkingTokens is preserved so billing stays // correct (Phase 1 depends on this). if resp != nil { if strip, _ := req.Options[OptStripThinking].(bool); strip { resp.Thinking = "" } } return resp, err } // chatRequestFn returns a closure that performs a single non-streaming chat request. // Shared between initial attempt and post-clamp retry to avoid duplication. func (p *OpenAIProvider) chatRequestFn(ctx context.Context, body map[string]any, tools []ToolDefinition) func() (*ChatResponse, error) { return func() (*ChatResponse, error) { respBody, err := p.doRequest(ctx, body) if err != nil { return nil, err } defer respBody.Close() var oaiResp openAIResponse if err := json.NewDecoder(respBody).Decode(&oaiResp); err != nil { return nil, fmt.Errorf("%s: decode response: %w", p.name, err) } return p.parseResponse(&oaiResp, tools), nil } } func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, error) { model := p.resolveModel(req.Model) // stripThinking suppresses user-visible reasoning while leaving // Usage.ThinkingTokens untouched (the usage chunk below still records it). stripThinking, _ := req.Options[OptStripThinking].(bool) body := p.buildRequestBody(model, req, true) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(model, req)) // Retry only the connection phase; once streaming starts, no retry. respBody, err := RetryDo(ctx, p.retryConfig, func() (io.ReadCloser, error) { return p.doRequest(ctx, body) }) // Auto-clamp max_tokens and retry once if the model rejects the value if err != nil { if clamped := clampMaxTokensFromError(err, body); clamped { slog.Info("max_tokens clamped, retrying stream", "model", model, "limit", clampedLimit(body)) respBody, err = RetryDo(ctx, p.retryConfig, func() (io.ReadCloser, error) { return p.doRequest(ctx, body) }) } } if err != nil { return nil, err } // Wrap respBody so ctx cancellation closes the socket, unblocking bufio.Scanner. cb := NewCtxBody(ctx, respBody) defer cb.Close() result := &ChatResponse{FinishReason: "stop"} accumulators := make(map[int]*toolCallAccumulator) normalizer := newControlOutputNormalizer(req.Tools) var textToolCalls []ToolCall sse := NewSSEScanner(cb) for sse.Next() { data := sse.Data() var chunk openAIStreamChunk if err := json.Unmarshal([]byte(data), &chunk); err != nil { continue } // Usage chunk often has empty choices — extract usage before skipping. // When stream_options.include_usage is true, the final chunk contains // usage data but choices is typically an empty array. if chunk.Usage != nil { result.Usage = &Usage{ PromptTokens: chunk.Usage.PromptTokens, CompletionTokens: chunk.Usage.CompletionTokens, TotalTokens: chunk.Usage.TotalTokens, RequestCount: 1, } if chunk.Usage.PromptTokensDetails != nil { result.Usage.CacheReadTokens = chunk.Usage.PromptTokensDetails.CachedTokens result.Usage.CacheCreationTokens = chunk.Usage.PromptTokensDetails.CacheWriteTokens + chunk.Usage.PromptTokensDetails.CacheCreationInputTokens result.Usage.PromptTokensIncludeCachedSegments = true } if chunk.Usage.CompletionTokensDetails != nil && chunk.Usage.CompletionTokensDetails.ReasoningTokens > 0 { result.Usage.ThinkingTokens = chunk.Usage.CompletionTokensDetails.ReasoningTokens } if chunk.Usage.ServerToolUse != nil { result.Usage.WebSearchCount = chunk.Usage.ServerToolUse.WebSearchRequests } } if len(chunk.Choices) == 0 { continue } delta := chunk.Choices[0].Delta reasoning := delta.ReasoningContent if reasoning == "" { reasoning = delta.Reasoning } if reasoning != "" && !stripThinking { result.Thinking += reasoning if onChunk != nil { onChunk(StreamChunk{Thinking: reasoning}) } } if delta.Content != "" { normalized := normalizer.Append(delta.Content) emitNormalizedControlChunk(result, normalized, stripThinking, onChunk) textToolCalls = append(textToolCalls, normalized.ToolCalls...) } // Accumulate images from delta.images[]. // Each chunk may carry one or more image parts; we collect all into result.Images. // Malformed data URLs are skipped with a warning — they don't abort the stream. for _, img := range delta.Images { mimeType, b64Data, err := parseDataURL(img.ImageURL.URL) if err != nil { slog.Warn("openai_stream: skipping malformed image data URL", "type", img.Type, "url_len", len(img.ImageURL.URL), "error", err) continue } result.Images = append(result.Images, ImageContent{ MimeType: mimeType, Data: b64Data, }) } // Accumulate streamed tool calls for _, tc := range delta.ToolCalls { acc, ok := accumulators[tc.Index] if !ok { acc = &toolCallAccumulator{ ToolCall: ToolCall{ID: tc.ID, Name: strings.TrimSpace(tc.Function.Name)}, } accumulators[tc.Index] = acc } if tc.Function.Name != "" { acc.Name = strings.TrimSpace(tc.Function.Name) } acc.rawArgs += tc.Function.Arguments if tc.Function.ThoughtSignature != "" { acc.thoughtSig = tc.Function.ThoughtSignature } } if chunk.Choices[0].FinishReason != "" { result.FinishReason = chunk.Choices[0].FinishReason } } normalized := normalizer.Finish() emitNormalizedControlChunk(result, normalized, stripThinking, onChunk) textToolCalls = append(textToolCalls, normalized.ToolCalls...) // Check for scanner errors (timeout, connection reset, etc.) if err := sse.Err(); err != nil { return result, fmt.Errorf("%s: stream read error: %w", p.name, err) } // Parse accumulated tool call arguments. // Iterate the map directly (not 0..len(accumulators)-1): providers are not // guaranteed to emit contiguous zero-based tc.Index values in delta.ToolCalls, // so indexing by position can miss populated slots and hit a nil accumulator, // causing a nil-pointer panic (observed with the point-p1/9router provider). for _, acc := range accumulators { if acc == nil { continue } args := make(map[string]any) if err := json.Unmarshal([]byte(acc.rawArgs), &args); err != nil && acc.rawArgs != "" { slog.Warn("openai_stream: failed to parse tool call arguments", "tool", acc.Name, "raw_len", len(acc.rawArgs), "error", err) acc.ParseError = fmt.Sprintf("malformed JSON (%d chars): %v", len(acc.rawArgs), err) } acc.Arguments = args if acc.thoughtSig != "" { acc.Metadata = map[string]string{"thought_signature": acc.thoughtSig} } result.ToolCalls = append(result.ToolCalls, acc.ToolCall) } result.ToolCalls = append(result.ToolCalls, textToolCalls...) // Only override finish_reason when stream wasn't truncated. // Preserve "length" so agent loop can detect truncation and retry. if len(result.ToolCalls) > 0 && result.FinishReason != "length" { result.FinishReason = "tool_calls" } if onChunk != nil { onChunk(StreamChunk{Done: true}) } return result, nil } const maxToolCallIDLen = 40 // normalizeMistralToolCallID deterministically maps any tool call ID to a // 9-character alphanumeric string required by the Mistral API. // Uses SHA-256 of the full ID to avoid prefix-dependent collisions. func normalizeMistralToolCallID(id string) string { h := sha256.Sum256([]byte(id)) return hex.EncodeToString(h[:])[:9] } // wireToolCallID dispatches to Mistral-specific normalization (9-char alnum) // or the standard OpenAI truncation (40-char max) based on the provider. func (p *OpenAIProvider) wireToolCallID(id string) string { if p.name == "mistral" || p.providerType == "mistral" { return normalizeMistralToolCallID(id) } return truncateToolCallID(id) } // truncateToolCallID deterministically fits tool call IDs into OpenAI's 40-char // limit. Prefix truncation can alias distinct legacy IDs that only diverge after // byte 40, so we hash the full original ID when shortening is needed. // // Fresh tool calls from the agent loop already go through uniquifyToolCallIDs // (which produces 40-char hashed IDs), so this is a no-op for those. This // function catches replayed/legacy history entries that bypassed uniquification. func truncateToolCallID(id string) string { if len(id) <= maxToolCallIDLen { return id } hash := sha256.Sum256([]byte(id)) return "call_" + hex.EncodeToString(hash[:])[:maxToolCallIDLen-len("call_")] }