1package provider
2
3import (
4 "bytes"
5 "context"
6 "encoding/base64"
7 "encoding/json"
8 "fmt"
9 "io"
10 "net/http"
11 "regexp"
12 "strings"
13 "sync"
14 "time"
15)
16
17const openaiDefaultBaseURL = "https://api.openai.com"
18
19// versionSegmentSuffix matches a trailing API version segment such as
20// "/v1" or Z.AI's "/v4".
21var versionSegmentSuffix = regexp.MustCompile(`/v\d+$`)
22
23// chatCompletionsURL builds the chat-completions endpoint for an
24// OpenAI-compatible base URL. A base that already carries an API
25// version segment gets "/chat/completions" appended directly; a bare
26// host (e.g. api.openai.com) gets the conventional "/v1/chat/completions".
27//
28// Matching any "/vN" segment (not just "/v1") keeps Z.AI's coding-plan
29// base, which ends in "/paas/v4", from getting a spurious "/v1" that
30// yields ".../paas/v4/v1/chat/completions" and a 404.
31func chatCompletionsURL(baseURL string) string {
32 if versionSegmentSuffix.MatchString(baseURL) {
33 return baseURL + "/chat/completions"
34 }
35 return baseURL + "/v1/chat/completions"
36}
37
38type openaiClient struct {
39 apiKey string
40 baseURL string
41 chatCompletionsPath string
42 name string
43 oauth bool // when true, apiKey actually holds an OAuth access token
44 headers map[string]string
45 http *http.Client
46 modelOverridesMu sync.RWMutex
47 modelOverrides map[string]Model
48}
49
50// NewOpenAI creates an OpenAI client using an API key. baseURL may be empty.
51func NewOpenAI(apiKey, baseURL string) Client {
52 if baseURL == "" {
53 baseURL = openaiDefaultBaseURL
54 }
55 return &openaiClient{
56 apiKey: apiKey,
57 baseURL: strings.TrimRight(baseURL, "/"),
58 name: "openai",
59 http: &http.Client{Timeout: 0},
60 }
61}
62
63// NewOpenAIOAuth creates an OpenAI client using a subscription OAuth access token.
64// The token is sent as an HTTP Bearer credential on the standard chat/completions endpoint.
65func NewOpenAIOAuth(accessToken, baseURL string) Client {
66 if baseURL == "" {
67 baseURL = openaiDefaultBaseURL
68 }
69 return &openaiClient{
70 apiKey: accessToken,
71 baseURL: strings.TrimRight(baseURL, "/"),
72 name: "openai",
73 oauth: true,
74 http: &http.Client{Timeout: 0},
75 }
76}
77
78// NewKimi creates a Kimi/Moonshot client. Kimi's chat API is OpenAI-compatible.
79func NewKimi(apiKey, baseURL string) Client {
80 return NewKimiWithHeaders(apiKey, baseURL, nil)
81}
82
83// NewDeepSeek creates a DeepSeek client. DeepSeek's chat API is
84// OpenAI-compatible at https://api.deepseek.com/v1.
85func NewDeepSeek(apiKey, baseURL string) Client {
86 if baseURL == "" {
87 baseURL = "https://api.deepseek.com/v1"
88 }
89 return &openaiClient{
90 apiKey: apiKey,
91 baseURL: strings.TrimRight(baseURL, "/"),
92 name: "deepseek",
93 http: &http.Client{Timeout: 0},
94 }
95}
96
97// NewKimiWithHeaders creates a Kimi/Moonshot client with extra headers.
98// Subscription tokens from Kimi Code need the official CLI's X-Msh-* headers.
99func NewKimiWithHeaders(apiKey, baseURL string, headers map[string]string) Client {
100 if baseURL == "" {
101 baseURL = "https://api.kimi.com/coding/v1"
102 }
103 return &openaiClient{
104 apiKey: apiKey,
105 baseURL: strings.TrimRight(baseURL, "/"),
106 name: "kimi",
107 headers: headers,
108 http: &http.Client{Timeout: 0},
109 }
110}
111
112func (c *openaiClient) Name() string {
113 if c.name != "" {
114 return c.name
115 }
116 return "openai"
117}
118
119func (c *openaiClient) SetModelMetadata(model Model) {
120 model = cloneModel(model)
121 c.modelOverridesMu.Lock()
122 if c.modelOverrides == nil {
123 c.modelOverrides = make(map[string]Model)
124 }
125 c.modelOverrides[model.ID] = model
126 c.modelOverridesMu.Unlock()
127}
128
129// ---- wire types ----
130
131type oaiContentText struct {
132 Type string `json:"type"` // "text"
133 Text string `json:"text"`
134}
135
136type oaiContentImage struct {
137 Type string `json:"type"` // "image_url"
138 ImageURL struct {
139 URL string `json:"url"`
140 } `json:"image_url"`
141}
142
143type oaiToolCallFn struct {
144 Name string `json:"name"`
145 Arguments string `json:"arguments"` // JSON string
146}
147
148type oaiToolCall struct {
149 ID string `json:"id"`
150 Type string `json:"type"` // "function"
151 Function oaiToolCallFn `json:"function"`
152}
153
154type oaiMessage struct {
155 Role string `json:"role"`
156 Content interface{} `json:"content,omitempty"` // string or []block
157 Name string `json:"name,omitempty"`
158 ToolCalls []oaiToolCall `json:"tool_calls,omitempty"`
159 ToolCallID string `json:"tool_call_id,omitempty"`
160 Tools []oaiTool `json:"tools,omitempty"`
161 // ReasoningContent carries the model's chain-of-thought summary
162 // alongside an assistant tool-call message. Required by Kimi's
163 // chat completions endpoint when thinking is enabled and the
164 // assistant message contains a tool call; OpenAI ignores it.
165 ReasoningContent string `json:"reasoning_content,omitempty"`
166}
167
168type oaiTool struct {
169 Type string `json:"type"` // "function"
170 Function struct {
171 Name string `json:"name"`
172 Description string `json:"description,omitempty"`
173 Parameters json.RawMessage `json:"parameters"`
174 } `json:"function"`
175}
176
177type oaiStreamOptions struct {
178 IncludeUsage bool `json:"include_usage"`
179}
180
181type oaiRequest struct {
182 Model string `json:"model"`
183 Messages []oaiMessage `json:"messages"`
184 Tools []oaiTool `json:"tools,omitempty"`
185 ToolChoice string `json:"tool_choice,omitempty"`
186 Stream bool `json:"stream"`
187 StreamOptions *oaiStreamOptions `json:"stream_options,omitempty"`
188 Temperature *float32 `json:"temperature,omitempty"`
189 MaxTokens *int `json:"max_tokens,omitempty"`
190 MaxCompletionTok *int `json:"max_completion_tokens,omitempty"`
191 ReasoningEffort string `json:"reasoning_effort,omitempty"`
192 ServiceTier string `json:"service_tier,omitempty"`
193}
194
195// ---- request building ----
196
197func (c *openaiClient) modelForRequest(id string) (Model, error) {
198 id = strings.TrimSpace(id)
199 c.modelOverridesMu.RLock()
200 model, ok := c.modelOverrides[id]
201 if ok {
202 model = cloneModel(model)
203 }
204 c.modelOverridesMu.RUnlock()
205 if ok {
206 return model, nil
207 }
208 model, err := FindModel(c.Name(), id)
209 if err != nil {
210 model, err = FindModel("", id)
211 }
212 return model, err
213}
214
215func (c *openaiClient) buildRequest(req Request) (*oaiRequest, error) {
216 req.Model = strings.TrimSpace(req.Model)
217 if err := ValidateFastMode(c.Name(), req.FastMode); err != nil {
218 return nil, err
219 }
220 // The OpenAI wire client is shared by many providers, so prefer its own
221 // provider namespace before falling back to a provider-agnostic lookup.
222 m, err := c.modelForRequest(req.Model)
223 if err != nil {
224 // Unknown model: use sensible defaults so local/custom
225 // models still work without a catalog entry.
226 m = Model{
227 ID: req.Model,
228 ContextWindow: 32768,
229 MaxOutput: 8192,
230 }
231 }
232 reasoning := ClampReasoningForModel(m, req.Reasoning)
233 out := &oaiRequest{
234 Model: req.Model,
235 Stream: true,
236 StreamOptions: &oaiStreamOptions{IncludeUsage: true},
237 Temperature: req.Temperature,
238 }
239 if req.FastMode {
240 out.ServiceTier = fastModeServiceTier
241 }
242
243 maxTok := req.MaxTokens
244 if maxTok <= 0 {
245 maxTok = m.MaxOutput
246 }
247 // Clamp max_tokens so output plus a minimum input reservation fits
248 // within the context window. Some providers (OpenRouter) enforce
249 // input + max_output <= context_length and reject requests where the
250 // total exceeds it. Reserving headroom guarantees the system prompt,
251 // first message, and tool definitions have room.
252 //
253 // The cap is derived from the context window, never from MaxOutput:
254 // MaxOutput is already the output ceiling, so subtracting from it
255 // would shrink every model's budget even when its output comfortably
256 // fits the window. We only lower maxTok when ContextWindow - reserve
257 // is actually tighter than the requested budget, which is exactly the
258 // pathological case (e.g. a model whose MaxOutput equals its window).
259 //
260 // The reserve is proportional (window/8, capped at 4096) rather than a
261 // flat 4096 so small-window models aren't over-penalized: a flat 4096
262 // would halve gpt-4's 8192 budget, while window/8 reserves a sensible
263 // 1024 there and still tops out at 4096 for large contexts.
264 //
265 // Some providers (OpenRouter) report inflated model-level context
266 // windows (e.g. 1000000) while the serving provider enforces a much
267 // tighter limit (e.g. 262144). Discovery already prefers the serving
268 // provider's smaller context_length, so m.ContextWindow is the real
269 // limit by the time we get here.
270 if m.ContextWindow > 0 {
271 reserve := m.ContextWindow / 8
272 const maxReserve = 4096
273 if reserve > maxReserve {
274 reserve = maxReserve
275 }
276 clamped := m.ContextWindow - reserve
277 if clamped < 1 {
278 clamped = 1
279 }
280 if maxTok > clamped {
281 maxTok = clamped
282 }
283 }
284 if m.Reasoning {
285 if maxTok > 0 {
286 out.MaxCompletionTok = &maxTok
287 }
288 effort := OpenAIReasoningEffort(reasoning)
289 if usesAdaptiveThinking(m) {
290 // Some gateways expose adaptive-thinking Anthropic models through
291 // the OpenAI-compatible chat-completions wire. They accept the
292 // same reasoning_effort knob, including the top "xhigh" tier;
293 // don't clamp zut's "maximum" to "high" for those models.
294 effort = OpenAICompatAnthropicEffort(reasoning)
295 }
296 // Explicit model mappings describe the endpoint's actual effort
297 // values and take precedence over conservative protocol defaults.
298 effort = reasoningEffortForModel(m, req.Reasoning, reasoning, effort)
299 if effort != "" {
300 out.ReasoningEffort = effort
301 }
302 } else if maxTok > 0 {
303 // Omit max_tokens when unknown (some discovered models don't
304 // advertise an output cap) so the server applies its own default
305 // instead of receiving an invalid max_tokens: 0.
306 out.MaxTokens = &maxTok
307 }
308
309 if system := req.SystemPrompt(); system != "" {
310 out.Messages = append(out.Messages, oaiMessage{Role: "system", Content: system})
311 }
312
313 deferredMode := supportsDeferredTools(c.Name()) && isKimiDeferredModel(req.Model)
314 toolByName := make(map[string]Tool, len(req.Tools))
315 for _, t := range req.Tools {
316 toolByName[t.Name] = t
317 }
318 activatedTools := activatedToolNames(req.Messages)
319
320 // DeepSeek's chat-completions API rejects the multimodal content
321 // schema (parts arrays containing image_url). Force every user/tool
322 // message to a plain string and silently drop image blocks for
323 // this provider so historical sessions with screenshots still replay.
324 // The same quirk applies to DeepSeek models routed through opencode-go.
325 textOnly := c.isDeepSeekRoute(req.Model)
326
327 req.Messages = RepairOrphanedToolResults(req.Messages)
328 for msgIndex := 0; msgIndex < len(req.Messages); msgIndex++ {
329 msg := req.Messages[msgIndex]
330 addedToolNames := append([]string(nil), msg.AddedToolNames...)
331 switch msg.Role {
332 case RoleDeveloper:
333 content := buildOAIUserContent(msg.Content, textOnly)
334 role := "developer"
335 if c.isDeepSeekRoute(req.Model) {
336 role = "system"
337 }
338 out.Messages = append(out.Messages, oaiMessage{Role: role, Content: content})
339 case RoleUser:
340 content := buildOAIUserContent(msg.Content, textOnly)
341 out.Messages = append(out.Messages, oaiMessage{Role: "user", Content: content})
342 case RoleAssistant:
343 am := oaiMessage{Role: "assistant"}
344 var text strings.Builder
345 var reasoning strings.Builder
346 for _, b := range msg.Content {
347 switch v := b.(type) {
348 case TextBlock:
349 if strings.TrimSpace(v.Text) == "" {
350 continue
351 }
352 if text.Len() > 0 {
353 text.WriteString("\n")
354 }
355 text.WriteString(v.Text)
356 case ToolCallBlock:
357 args := v.Arguments
358 if len(args) == 0 || !json.Valid(args) {
359 args = json.RawMessage("{}")
360 }
361 am.ToolCalls = append(am.ToolCalls, oaiToolCall{
362 ID: v.ID,
363 Type: "function",
364 Function: oaiToolCallFn{
365 Name: v.Name,
366 Arguments: string(args),
367 },
368 })
369 case ReasoningBlock:
370 if v.Summary != "" {
371 if reasoning.Len() > 0 {
372 reasoning.WriteString("\n")
373 }
374 reasoning.WriteString(v.Summary)
375 }
376 }
377 }
378 if text.Len() > 0 {
379 am.Content = text.String()
380 }
381 if reasoning.Len() > 0 && len(am.ToolCalls) > 0 {
382 am.ReasoningContent = reasoning.String()
383 }
384 // Kimi rejects assistant messages with neither visible text nor
385 // tool calls ("assistant must not be empty"). This can happen when
386 // a previous stream produced only reasoning_content, which zut keeps
387 // internally for provider replay but cannot send back as standalone
388 // assistant content on OpenAI-compatible chat-completions APIs.
389 if am.Content == nil && len(am.ToolCalls) == 0 {
390 continue
391 }
392 out.Messages = append(out.Messages, am)
393 case RoleTool:
394 // Chat Completions tool messages only carry text. Keep every
395 // tool result paired with its call, then deliver any images in
396 // one user message after the complete batch of tool results.
397 // Strict OpenAI-compatible APIs reject image_url blocks on a
398 // role=tool message.
399 var images []Content
400 groupEnd := msgIndex
401 for groupEnd < len(req.Messages) && req.Messages[groupEnd].Role == RoleTool {
402 toolMessage := req.Messages[groupEnd]
403 if groupEnd > msgIndex {
404 addedToolNames = append(addedToolNames, toolMessage.AddedToolNames...)
405 }
406 for _, b := range toolMessage.Content {
407 if tr, ok := b.(ToolResultBlock); ok {
408 content := buildOAIToolContent(tr.Content, tr.IsError, textOnly)
409 out.Messages = append(out.Messages, oaiMessage{
410 Role: "tool",
411 ToolCallID: tr.CallID,
412 Content: content,
413 })
414 if !textOnly {
415 for _, inner := range tr.Content {
416 if image, ok := inner.(ImageBlock); ok {
417 images = append(images, image)
418 }
419 }
420 }
421 }
422 }
423 groupEnd++
424 }
425 // The agent loop persists an image mirror as the next user
426 // message so Responses-routed models can consume it. Reuse
427 // that message instead of sending the image twice.
428 hasPersistedMirror := groupEnd < len(req.Messages) && isOpenAIToolImageMirror(req.Messages[groupEnd])
429 if len(images) > 0 && !hasPersistedMirror {
430 content := append([]Content{TextBlock{Text: "Tool output included the following image content:"}}, images...)
431 out.Messages = append(out.Messages, oaiMessage{
432 Role: "user",
433 Content: buildOAIContentBlocks(content, false),
434 })
435 }
436 msgIndex = groupEnd - 1
437 }
438 if deferredMode {
439 var loaded []oaiTool
440 for _, name := range addedToolNames {
441 if t, ok := toolByName[name]; ok && t.Deferred {
442 loaded = append(loaded, makeOAITool(t))
443 }
444 }
445 if len(loaded) > 0 {
446 out.Messages = append(out.Messages, oaiMessage{Role: "system", Tools: loaded})
447 }
448 }
449 }
450
451 for _, t := range req.Tools {
452 if t.Deferred && (deferredMode || !activatedTools[t.Name]) {
453 continue
454 }
455 out.Tools = append(out.Tools, makeOAITool(t))
456 }
457 if len(out.Tools) > 0 {
458 out.ToolChoice = "auto"
459 }
460
461 return out, nil
462}
463
464func isKimiDeferredModel(model string) bool {
465 model = strings.ToLower(model)
466 return model == "kimi-k3" || strings.HasSuffix(model, "/kimi-k3")
467}
468
469// supportsDeferredTools reports whether a provider speaks Moonshot's
470// deferred-tools handshake, where a tool that becomes available
471// mid-conversation is delivered on a system message carrying a `tools`
472// array. That message is a Moonshot API extension, not OpenAI wire format:
473// it has no content, which a strict chat-completions schema rejects
474// outright. Other providers take the standard path instead, where an
475// activated deferred tool joins the top-level tools array.
476func supportsDeferredTools(provider string) bool {
477 return provider == "moonshotai" || provider == "moonshotai-cn"
478}
479
480func (c *openaiClient) isDeepSeekRoute(model string) bool {
481 return c.name == ProviderDeepSeek || (c.name == ProviderOpenCodeGo && isDeepSeekModel(model))
482}
483
484func isDeepSeekModel(model string) bool {
485 model = strings.ToLower(strings.TrimSpace(model))
486 return strings.HasPrefix(model, "deepseek-") || strings.Contains(model, "/deepseek-")
487}
488
489func makeOAITool(t Tool) oaiTool {
490 var tool oaiTool
491 tool.Type = "function"
492 tool.Function.Name = t.Name
493 tool.Function.Description = t.Description
494 tool.Function.Parameters = t.Schema
495 return tool
496}
497
498func buildOAIUserContent(blocks []Content, textOnly bool) interface{} {
499 hasImage := false
500 for _, b := range blocks {
501 if _, ok := b.(ImageBlock); ok {
502 hasImage = true
503 break
504 }
505 }
506 if textOnly || !hasImage {
507 // Keep multiple text blocks structured. The core provider projection
508 // adds message-time annotations as a separate block; collapsing that
509 // block with the user text changes the wire representation of the
510 // original prompt and can make consumers misclassify the request.
511 if !textOnly && len(blocks) > 1 {
512 return buildOAIContentBlocks(blocks, false)
513 }
514 var sb strings.Builder
515 for _, b := range blocks {
516 if tb, ok := b.(TextBlock); ok {
517 if sb.Len() > 0 {
518 sb.WriteString("\n")
519 }
520 sb.WriteString(tb.Text)
521 }
522 }
523 return sb.String()
524 }
525 return buildOAIContentBlocks(blocks, false)
526}
527
528func buildOAIToolContent(blocks []Content, isError, textOnly bool) string {
529 var sb strings.Builder
530 hasImage := false
531 for _, b := range blocks {
532 switch v := b.(type) {
533 case TextBlock:
534 if sb.Len() > 0 {
535 sb.WriteString("\n")
536 }
537 sb.WriteString(v.Text)
538 case ImageBlock:
539 hasImage = true
540 }
541 }
542 if sb.Len() == 0 && hasImage && !textOnly {
543 sb.WriteString("(see attached image)")
544 }
545 if isError && sb.Len() > 0 {
546 sb.WriteString(" [error]")
547 }
548 return sb.String()
549}
550
551// isOpenAIToolImageMirror reports whether msg is the synthetic user message
552// the agent loop persists after a tool result carrying images (see
553// mirrorToolImagesAsUser). The image bytes are already available once
554// through that message, so buildRequest must not emit a second copy.
555func isOpenAIToolImageMirror(msg Message) bool {
556 if msg.Role != RoleUser || len(msg.Content) < 2 {
557 return false
558 }
559 text, ok := msg.Content[0].(TextBlock)
560 if !ok || text.Text != "Tool output included the following image content:" {
561 return false
562 }
563 for _, content := range msg.Content[1:] {
564 if _, ok := content.(ImageBlock); ok {
565 return true
566 }
567 }
568 return false
569}
570
571func buildOAIContentBlocks(blocks []Content, isError bool) []interface{} {
572 var arr []interface{}
573 for _, b := range blocks {
574 switch v := b.(type) {
575 case TextBlock:
576 arr = append(arr, oaiContentText{Type: "text", Text: v.Text})
577 case ImageBlock:
578 var img oaiContentImage
579 img.Type = "image_url"
580 img.ImageURL.URL = "data:" + v.MimeType + ";base64," + base64.StdEncoding.EncodeToString(v.Data)
581 arr = append(arr, img)
582 }
583 }
584 if isError {
585 arr = append(arr, oaiContentText{Type: "text", Text: "[error]"})
586 }
587 return arr
588}
589
590// ---- streaming ----
591
592func (c *openaiClient) chatCompletionsURL() string {
593 if c.chatCompletionsPath != "" {
594 return strings.TrimRight(c.baseURL, "/") + "/" + strings.TrimLeft(c.chatCompletionsPath, "/")
595 }
596 return chatCompletionsURL(c.baseURL)
597}
598
599func (c *openaiClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
600 req.Model = strings.TrimSpace(req.Model)
601 endpoint := c.chatCompletionsURL()
602 wire, err := c.buildRequest(req)
603 if err != nil {
604 return nil, err
605 }
606 body, err := json.Marshal(wire)
607 if err != nil {
608 return nil, err
609 }
610 newReq := func() (*http.Request, error) {
611 httpReq, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(body))
612 if err != nil {
613 return nil, err
614 }
615 httpReq.Header.Set("content-type", "application/json")
616 httpReq.Header.Set("accept", "text/event-stream")
617 if c.apiKey != "" {
618 // Keyless custom endpoints send no Authorization header at all.
619 httpReq.Header.Set("authorization", "Bearer "+c.apiKey)
620 }
621 for k, v := range c.headers {
622 httpReq.Header.Set(k, v)
623 }
624 setOpenCodeGoHeaders(httpReq.Header, c.Name(), req.Context)
625 return httpReq, nil
626 }
627
628 resp, err := doStreamWithRetry(ctx, c.http, newReq, req.Lifecycle)
629 if err != nil {
630 return nil, fmt.Errorf("%s: %w", c.Name(), err)
631 }
632 if resp.StatusCode != http.StatusOK {
633 b, _ := io.ReadAll(resp.Body)
634 resp.Body.Close()
635 return nil, fmt.Errorf("%s: http %d: %s", c.Name(), resp.StatusCode, strings.TrimSpace(string(b)))
636 }
637
638 out := make(chan Event, 16)
639 go c.runStream(ctx, resp, req, out)
640 return out, nil
641}
642
643func (c *openaiClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
644 defer close(out)
645 defer resp.Body.Close()
646
647 model, _ := c.modelForRequest(req.Model)
648 out <- EventStart{Model: req.Model, Provider: c.Name()}
649
650 raw := make(chan sseEvent, 16)
651 go readSSE(resp.Body, raw)
652
653 // Interleaved block tracking: text and tool_calls preserve their
654 // emission order so the assistant message renders in the same order
655 // the model produced it. The builder fires one kind of block at a
656 // time — incoming text deltas after a tool_call split into a fresh
657 // text block; subsequent tool_calls each get their own slot.
658 type blockEntry struct {
659 kind string // "text" | "tool_use"
660 textBuf strings.Builder
661 toolID string
662 toolName string
663 toolArgs strings.Builder
664 announced bool
665 }
666 var (
667 blocks []*blockEntry
668 currentText *blockEntry // most-recent text block, nil if none
669 toolByIdx = map[int]*blockEntry{} // openai tool_call index -> block
670 reasoningBuf strings.Builder
671 usage Usage
672 stop = StopEnd
673 finalErr error
674 )
675
676 appendText := func(delta string) {
677 if currentText == nil {
678 currentText = &blockEntry{kind: "text"}
679 blocks = append(blocks, currentText)
680 }
681 currentText.textBuf.WriteString(delta)
682 }
683
684 getOrCreateTool := func(idx int) *blockEntry {
685 if t, ok := toolByIdx[idx]; ok {
686 return t
687 }
688 t := &blockEntry{kind: "tool_use"}
689 toolByIdx[idx] = t
690 blocks = append(blocks, t)
691 // A new tool block breaks the current text block. Subsequent text
692 // deltas will start a fresh text block after this tool.
693 currentText = nil
694 return t
695 }
696
697 assembleMsg := func() Message {
698 content := []Content{}
699 for _, b := range blocks {
700 switch b.kind {
701 case "text":
702 if b.textBuf.Len() > 0 {
703 content = append(content, TextBlock{Text: b.textBuf.String()})
704 }
705 case "tool_use":
706 content = append(content, ToolCallBlock{
707 ID: b.toolID, Name: b.toolName, Arguments: normalizeToolArgs(b.toolArgs.String()),
708 })
709 }
710 }
711 if reasoningBuf.Len() > 0 {
712 content = append(content, ReasoningBlock{Summary: reasoningBuf.String()})
713 }
714 return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
715 }
716
717 sendDone := func() {
718 usage.CostUSD = ComputeCost(model, usage)
719 out <- EventUsage{Usage: usage}
720 out <- EventDone{Stop: stop, Err: finalErr, Message: assembleMsg()}
721 }
722
723 for {
724 select {
725 case <-ctx.Done():
726 stop = StopAborted
727 finalErr = ctx.Err()
728 sendDone()
729 return
730 case ev, ok := <-raw:
731 if !ok {
732 sendDone()
733 return
734 }
735 if ev.Err != nil {
736 stop = StopError
737 finalErr = fmt.Errorf("read SSE: %w", ev.Err)
738 sendDone()
739 return
740 }
741 if ev.Data == "[DONE]" {
742 sendDone()
743 return
744 }
745 var chunk struct {
746 Choices []struct {
747 Index int `json:"index"`
748 Delta struct {
749 Content string `json:"content"`
750 ReasoningContent string `json:"reasoning_content"`
751 ToolCalls []struct {
752 Index int `json:"index"`
753 ID string `json:"id"`
754 Type string `json:"type"`
755 Function struct {
756 Name string `json:"name"`
757 Arguments string `json:"arguments"`
758 } `json:"function"`
759 } `json:"tool_calls"`
760 } `json:"delta"`
761 FinishReason string `json:"finish_reason"`
762 } `json:"choices"`
763 Usage *struct {
764 PromptTokens int `json:"prompt_tokens"`
765 PromptCacheHitTokens *int `json:"prompt_cache_hit_tokens"`
766 PromptCacheMissTokens *int `json:"prompt_cache_miss_tokens"`
767 CompletionTokens int `json:"completion_tokens"`
768 PromptTokensDetails *openAIInputTokensDetails `json:"prompt_tokens_details"`
769 CompletionTokensDetails *struct {
770 ReasoningTokens int `json:"reasoning_tokens"`
771 } `json:"completion_tokens_details"`
772 } `json:"usage"`
773 Error *struct {
774 Message string `json:"message"`
775 Type string `json:"type"`
776 } `json:"error"`
777 }
778 if err := json.Unmarshal([]byte(ev.Data), &chunk); err != nil {
779 continue
780 }
781 if chunk.Error != nil {
782 stop = StopError
783 finalErr = fmt.Errorf("openai: %s", chunk.Error.Message)
784 sendDone()
785 return
786 }
787 if chunk.Usage != nil {
788 if c.isDeepSeekRoute(req.Model) && (chunk.Usage.PromptCacheHitTokens != nil || chunk.Usage.PromptCacheMissTokens != nil) {
789 usage = normalizeDeepSeekUsage(chunk.Usage.PromptTokens, chunk.Usage.CompletionTokens, chunk.Usage.PromptCacheHitTokens, chunk.Usage.PromptCacheMissTokens)
790 } else {
791 usage = normalizeOpenAIUsage(chunk.Usage.PromptTokens, chunk.Usage.CompletionTokens, chunk.Usage.PromptTokensDetails)
792 }
793 if details := chunk.Usage.CompletionTokensDetails; details != nil {
794 usage.ReasoningTokens = details.ReasoningTokens
795 usage.ReasoningTokensKnown = true
796 }
797 }
798 for _, ch := range chunk.Choices {
799 if ch.Delta.ReasoningContent != "" {
800 reasoningBuf.WriteString(ch.Delta.ReasoningContent)
801 }
802 if ch.Delta.Content != "" {
803 appendText(ch.Delta.Content)
804 out <- EventTextDelta{Delta: ch.Delta.Content}
805 }
806 for _, tc := range ch.Delta.ToolCalls {
807 t := getOrCreateTool(tc.Index)
808 if tc.ID != "" {
809 t.toolID = tc.ID
810 }
811 if tc.Function.Name != "" {
812 t.toolName = tc.Function.Name
813 }
814 if !t.announced && t.toolID != "" && t.toolName != "" {
815 t.announced = true
816 out <- EventToolStart{ID: t.toolID, Name: t.toolName}
817 }
818 if tc.Function.Arguments != "" {
819 t.toolArgs.WriteString(tc.Function.Arguments)
820 if t.announced {
821 out <- EventToolArgs{ID: t.toolID, Delta: tc.Function.Arguments}
822 }
823 }
824 }
825 switch ch.FinishReason {
826 case "stop":
827 stop = StopEnd
828 case "length":
829 stop = StopLength
830 case "tool_calls", "function_call":
831 stop = StopToolUse
832 for _, b := range blocks {
833 if b.kind == "tool_use" && b.announced {
834 out <- EventToolEnd{ID: b.toolID}
835 }
836 }
837 }
838 }
839 }
840 }
841}