1package provider
2
3import (
4 "bytes"
5 "context"
6 "encoding/base64"
7 "encoding/json"
8 "fmt"
9 "io"
10 "net/http"
11 "runtime"
12 "strings"
13 "sync"
14 "time"
15)
16
17// Codex (ChatGPT subscription) client. Uses OpenAI's Responses API via
18// chatgpt.com/backend-api with the chatgpt-account-id handshake.
19//
20// Wire protocol notes:
21// - Endpoint: POST https://chatgpt.com/backend-api/codex/responses
22// - Headers: Authorization: Bearer <access_token>, chatgpt-account-id: <id>,
23// OpenAI-Beta: responses=experimental, originator: zot (required legacy value)
24// - Body: OpenAI Responses API shape (not chat/completions).
25// input: [{role, content: [{type: "input_text" | "input_image" | ... }]}]
26// instructions: <system prompt>
27// tools: [{type:"function", name, description, parameters, strict}]
28// stream: true
29// - SSE events: response.output_item.added, response.output_text.delta,
30// response.function_call_arguments.delta, response.output_item.done,
31// response.completed, response.failed, error
32//
33// The access_token comes from the OpenAI OAuth flow (auth.openai.com).
34// The accountID is parsed from the id_token JWT's `chatgpt_account_id`
35// claim; see auth/oauth.go.
36
37const codexDefaultBaseURL = "https://chatgpt.com/backend-api/codex/responses"
38
39type codexClient struct {
40 token string
41 accountID string
42 baseURL string
43 errorLabel string
44 providerName string
45 modelName func(string) string
46 disableCLIRouting bool
47 cliRoutingAll bool
48 capabilities responsesCapabilities
49 http *http.Client
50 modelOverridesMu sync.RWMutex
51 modelOverrides map[string]Model
52}
53
54// responsesCapabilities records extensions verified for the exact provider
55// client, rather than assuming every Responses-shaped endpoint accepts them.
56type responsesCapabilities struct {
57 StablePromptCacheKey bool
58 ExplicitPromptCacheBreakpoint bool
59 SessionHeader bool
60 ThreadHeader bool
61 ClientRequestHeader bool
62}
63
64// NewOpenAICodex creates a Responses client for ChatGPT's Codex endpoint
65// using a subscription OAuth access token and the user's ChatGPT account ID.
66// It prefers WebSocket mode and falls back to HTTP/SSE before visible output.
67// baseURL may be empty to use the default.
68func NewOpenAICodex(token, accountID, baseURL string) Client {
69 return newResponsesWebSocketClient(newOpenAICodexClient(token, accountID, baseURL))
70}
71
72func newOpenAICodexClient(token, accountID, baseURL string) *codexClient {
73 if baseURL == "" {
74 baseURL = codexDefaultBaseURL
75 }
76 return &codexClient{
77 token: token,
78 accountID: accountID,
79 baseURL: strings.TrimRight(baseURL, "/"),
80 errorLabel: "codex",
81 providerName: "openai-codex",
82 // The ChatGPT Codex backend load-sheds requests from client
83 // identities it does not recognize: unknown originator/user-agent
84 // pairs receive "Our servers are currently overloaded" stream
85 // errors near-deterministically while Codex CLI requests succeed.
86 // Send the Codex CLI request shape for every model on this
87 // backend.
88 cliRoutingAll: true,
89 capabilities: responsesCapabilities{
90 StablePromptCacheKey: true,
91 SessionHeader: true,
92 ThreadHeader: true,
93 ClientRequestHeader: true,
94 },
95 http: &http.Client{Timeout: 0},
96 }
97}
98
99func (c *codexClient) Name() string { return ProviderOpenAICodex }
100
101func (c *codexClient) SetModelMetadata(model Model) {
102 model = cloneModel(model)
103 c.modelOverridesMu.Lock()
104 if c.modelOverrides == nil {
105 c.modelOverrides = make(map[string]Model)
106 }
107 c.modelOverrides[model.ID] = model
108 c.modelOverridesMu.Unlock()
109}
110
111// ---- Responses API wire types (subset needed for zut's surface) ----
112
113type codexInputText struct {
114 Type string `json:"type"` // "input_text"
115 Text string `json:"text"`
116 PromptCacheBreakpoint *codexPromptCacheBreakpoint `json:"prompt_cache_breakpoint,omitempty"`
117}
118
119type codexInputImage struct {
120 Type string `json:"type"` // "input_image"
121 Detail string `json:"detail"`
122 ImageURL string `json:"image_url"`
123}
124
125type codexOutputText struct {
126 Type string `json:"type"` // "output_text"
127 Text string `json:"text"`
128 Annotations []any `json:"annotations"`
129}
130
131type codexPromptCacheOptions struct {
132 Mode string `json:"mode"`
133}
134
135type codexPromptCacheBreakpoint struct {
136 Mode string `json:"mode"`
137}
138
139type codexInputMessage struct {
140 Type string `json:"type,omitempty"` // "message" (optional for input)
141 Role string `json:"role"`
142 Content []any `json:"content"`
143}
144
145type codexOutputMessage struct {
146 Type string `json:"type"` // "message"
147 Role string `json:"role"`
148 Status string `json:"status,omitempty"`
149 ID string `json:"id,omitempty"`
150 Content []codexOutputText `json:"content"`
151}
152
153type codexFunctionCall struct {
154 Type string `json:"type"` // "function_call"
155 ID string `json:"id,omitempty"`
156 CallID string `json:"call_id"`
157 Name string `json:"name"`
158 Arguments string `json:"arguments"` // JSON string
159}
160
161type codexFunctionCallOutput struct {
162 Type string `json:"type"` // "function_call_output"
163 CallID string `json:"call_id"`
164 Output string `json:"output"` // string (or ResponseFunctionCallOutputItemList for images; v1 only uses string)
165}
166
167// codexReasoningItem mirrors the Responses API "reasoning" output item.
168// We capture it on incoming streams and replay it verbatim on follow-up
169// requests: the API rejects assistant tool-call replays without it when
170// thinking is enabled.
171type codexReasoningItem struct {
172 Type string `json:"type"` // "reasoning"
173 ID string `json:"id,omitempty"`
174 EncryptedContent string `json:"encrypted_content,omitempty"`
175 // Summary is required by the Responses API even when no summary text
176 // was streamed; encode an empty array rather than omitting the field.
177 Summary []codexReasoningSummary `json:"summary"`
178}
179
180type codexReasoningSummary struct {
181 Type string `json:"type"` // "summary_text"
182 Text string `json:"text"`
183}
184
185type codexTool struct {
186 Type string `json:"type"` // "function"
187 Name string `json:"name"`
188 Description string `json:"description,omitempty"`
189 Parameters json.RawMessage `json:"parameters"`
190 // Responses otherwise attempts to normalize schemas into strict mode,
191 // which makes non-nullable optional properties required model output.
192 Strict bool `json:"strict"`
193}
194
195type codexReasoningConfig struct {
196 Effort string `json:"effort,omitempty"`
197}
198
199type codexRequest struct {
200 Model string `json:"model"`
201 Store bool `json:"store"`
202 Stream bool `json:"stream,omitempty"`
203 Instructions string `json:"instructions,omitempty"`
204 Input []any `json:"input"`
205 Tools []codexTool `json:"tools,omitempty"`
206 ToolChoice string `json:"tool_choice,omitempty"`
207 ParallelToolCalls bool `json:"parallel_tool_calls"`
208 Include []string `json:"include,omitempty"`
209 Reasoning *codexReasoningConfig `json:"reasoning,omitempty"`
210 PromptCacheKey string `json:"prompt_cache_key,omitempty"`
211 PromptCacheOptions *codexPromptCacheOptions `json:"prompt_cache_options,omitempty"`
212 ServiceTier string `json:"service_tier,omitempty"`
213 PreviousResponseID string `json:"previous_response_id,omitempty"`
214}
215
216// ---- Request building ----
217
218func (c *codexClient) findModel(id string) (Model, error) {
219 id = strings.TrimSpace(id)
220 c.modelOverridesMu.RLock()
221 model, ok := c.modelOverrides[id]
222 if ok {
223 model = cloneModel(model)
224 }
225 c.modelOverridesMu.RUnlock()
226 if ok {
227 return model, nil
228 }
229 if c.providerName != "" && c.providerName != ProviderOpenAICodex {
230 if m, err := FindModel(c.providerName, id); err == nil {
231 return m, nil
232 }
233 }
234 if m, err := FindModel(ProviderOpenAICodex, id); err == nil {
235 return m, nil
236 }
237 m, err := FindModel(ProviderOpenAI, id)
238 if err == nil {
239 return m, nil
240 }
241 if c.providerName == ProviderOpenCodeGo {
242 if api := openCodeGoAPIForModel(id); api != "" {
243 return Model{
244 Provider: ProviderOpenCodeGo,
245 ID: id,
246 API: api,
247 ContextWindow: 128000,
248 MaxOutput: 16384,
249 Reasoning: true,
250 }, nil
251 }
252 }
253 return m, err
254}
255
256func supportsOpenAIExplicitPromptCache(model string) bool {
257 id := strings.ToLower(strings.TrimSpace(model))
258 return strings.HasPrefix(id, "gpt-5.6-") || id == "gpt-6-astra"
259}
260
261func (c *codexClient) buildRequest(req Request) (*codexRequest, error) {
262 req.Model = strings.TrimSpace(req.Model)
263 providerName := c.providerName
264 if providerName == "" {
265 providerName = ProviderOpenAICodex
266 }
267 if err := ValidateFastMode(providerName, req.FastMode); err != nil {
268 return nil, err
269 }
270 m, err := c.findModel(req.Model)
271 if err != nil {
272 return nil, err
273 }
274 reasoning := ClampReasoningForModel(m, req.Reasoning)
275
276 body := &codexRequest{
277 Model: req.Model,
278 Store: false,
279 Stream: true,
280 Instructions: req.SystemPrompt(),
281 ParallelToolCalls: true,
282 Include: []string{"reasoning.encrypted_content"},
283 }
284 if req.FastMode {
285 body.ServiceTier = fastModeServiceTier
286 }
287 if c.capabilities.ExplicitPromptCacheBreakpoint && supportsOpenAIExplicitPromptCache(req.Model) {
288 // Keep implicit mode so OpenAI can retain later conversation
289 // checkpoints as well as this stable instruction/tool boundary.
290 body.PromptCacheOptions = &codexPromptCacheOptions{Mode: "implicit"}
291 body.Input = append(body.Input, codexInputMessage{
292 Role: "developer",
293 Content: []any{codexInputText{
294 Type: "input_text",
295 Text: "[Stable prompt boundary]",
296 PromptCacheBreakpoint: &codexPromptCacheBreakpoint{Mode: "explicit"},
297 }},
298 })
299 }
300 if m.Reasoning {
301 effort := OpenAICodexReasoningEffort(reasoning, req.Model)
302 effort = reasoningEffortForModel(m, req.Reasoning, reasoning, effort)
303 if effort != "" {
304 body.Reasoning = &codexReasoningConfig{Effort: effort}
305 }
306 }
307 activeTools := activeToolDefinitions(req.Tools, req.Messages)
308 if len(activeTools) > 0 {
309 body.ToolChoice = "auto"
310 for _, t := range activeTools {
311 params := t.Schema
312 if len(params) == 0 {
313 params = json.RawMessage(`{"type":"object","properties":{}}`)
314 }
315 body.Tools = append(body.Tools, codexTool{
316 Type: "function",
317 Name: t.Name,
318 Description: t.Description,
319 Parameters: params,
320 })
321 }
322 }
323
324 msgIdx := 0
325 req.Messages = RepairOrphanedToolResults(req.Messages)
326 for _, msg := range req.Messages {
327 switch msg.Role {
328 case RoleDeveloper, RoleUser:
329 content := []any{}
330 for _, c := range msg.Content {
331 switch v := c.(type) {
332 case TextBlock:
333 if v.Text != "" {
334 content = append(content, codexInputText{Type: "input_text", Text: v.Text})
335 }
336 case ImageBlock:
337 url := "data:" + v.MimeType + ";base64," + base64.StdEncoding.EncodeToString(v.Data)
338 content = append(content, codexInputImage{Type: "input_image", Detail: "auto", ImageURL: url})
339 }
340 }
341 if len(content) == 0 {
342 continue
343 }
344 role := "user"
345 if msg.Role == RoleDeveloper {
346 role = "developer"
347 }
348 body.Input = append(body.Input, codexInputMessage{Role: role, Content: content})
349 case RoleAssistant:
350 // Emit one output_message per text block, one function_call per
351 // tool call, and one reasoning item per replayable ReasoningBlock,
352 // preserving the order so the model sees the same interleaving
353 // we captured. The reasoning replay is what keeps Responses
354 // providers from rejecting follow-up tool calls when thinking
355 // is enabled. Blocks without encrypted content are omitted:
356 // an ID or summary alone is not a self-contained reasoning
357 // item while store is false. See reasoningForReplay.
358 for _, c := range replayContent(msg, providerName, req.Model) {
359 switch v := c.(type) {
360 case ReasoningBlock:
361 item, ok := reasoningForReplay(v)
362 if !ok {
363 continue
364 }
365 body.Input = append(body.Input, item)
366 case TextBlock:
367 if v.Text == "" {
368 continue
369 }
370 msgIdx++
371 body.Input = append(body.Input, codexOutputMessage{
372 Type: "message",
373 Role: "assistant",
374 Status: "completed",
375 ID: fmt.Sprintf("msg_%d", msgIdx),
376 Content: []codexOutputText{
377 {Type: "output_text", Text: v.Text, Annotations: []any{}},
378 },
379 })
380 case ToolCallBlock:
381 args := string(v.Arguments)
382 if args == "" || !json.Valid([]byte(args)) {
383 args = "{}"
384 }
385 callID, _ := splitCallID(v.ID)
386 body.Input = append(body.Input, codexFunctionCall{
387 Type: "function_call",
388 CallID: callID,
389 Name: v.Name,
390 Arguments: args,
391 })
392 }
393 }
394 case RoleTool:
395 for _, c := range msg.Content {
396 if tr, ok := c.(ToolResultBlock); ok {
397 var text strings.Builder
398 imageCount := 0
399 for _, inner := range tr.Content {
400 switch v := inner.(type) {
401 case TextBlock:
402 if text.Len() > 0 {
403 text.WriteString("\n")
404 }
405 text.WriteString(v.Text)
406 case ImageBlock:
407 imageCount++
408 }
409 }
410 // The Responses API function_call_output only carries a
411 // string, so image bytes cannot ride along here. The agent
412 // loop mirrors any tool-result images into the following
413 // user message (which this client does serialize as
414 // input_image). Leave a short text note so an image-only
415 // result is not an empty output the API may reject, and so
416 // the model knows the image arrives next.
417 out := text.String()
418 if out == "" && imageCount > 0 {
419 if imageCount == 1 {
420 out = "[image returned; see the following message]"
421 } else {
422 out = fmt.Sprintf("[%d images returned; see the following message]", imageCount)
423 }
424 }
425 callID, _ := splitCallID(tr.CallID)
426 body.Input = append(body.Input, codexFunctionCallOutput{
427 Type: "function_call_output",
428 CallID: callID,
429 Output: out,
430 })
431 }
432 }
433 }
434 }
435
436 return body, nil
437}
438
439// reasoningForReplay decides whether a persisted ReasoningBlock carries a
440// self-contained Responses payload. Only blocks with encrypted content are
441// replayed; an ID or a readable summary alone cannot stand in for the
442// provider-issued payload when store is false. The boolean is false when
443// the block must be omitted from outgoing Responses input. Stored messages
444// are never mutated by this decision.
445func reasoningForReplay(block ReasoningBlock) (codexReasoningItem, bool) {
446 if block.Encrypted == "" {
447 return codexReasoningItem{}, false
448 }
449 // Gateway composite IDs and other providers' signatures are not native
450 // Responses items. Never rewrite an opaque ID to make it look valid.
451 for _, ch := range block.ID {
452 switch {
453 case ch >= 'a' && ch <= 'z', ch >= 'A' && ch <= 'Z',
454 ch >= '0' && ch <= '9', ch == '_', ch == '-':
455 // Responses item IDs use ASCII letters, digits, underscores, or dashes.
456 default:
457 return codexReasoningItem{}, false
458 }
459 }
460 item := codexReasoningItem{
461 Type: "reasoning",
462 ID: block.ID,
463 EncryptedContent: block.Encrypted,
464 Summary: []codexReasoningSummary{},
465 }
466 if block.Summary != "" {
467 item.Summary = []codexReasoningSummary{{Type: "summary_text", Text: block.Summary}}
468 }
469 return item, true
470}
471
472func splitCallID(id string) (string, string) {
473 if i := strings.Index(id, "|"); i >= 0 {
474 return id[:i], id[i+1:]
475 }
476 return id, ""
477}
478
479const minOpenAIPromptCacheTokens = 1024
480
481func responsesCacheDiagnostics(wire *codexRequest, transport, continuation string) CacheDiagnostics {
482 diagnostics := CacheDiagnostics{Mode: "none", Transport: transport, Continuation: continuation}
483 if wire == nil || wire.PromptCacheOptions == nil {
484 return diagnostics
485 }
486 diagnostics.Mode = wire.PromptCacheOptions.Mode
487 diagnostics.Eligible = estimatedOpenAIPromptTokens(wire) >= minOpenAIPromptCacheTokens
488 return diagnostics
489}
490
491// estimatedOpenAIPromptTokens supplies a deliberately conservative local
492// eligibility signal. OpenAI makes the final tokenization and cache decision;
493// this estimate never leaves the process and is not persisted in diagnostics.
494func estimatedOpenAIPromptTokens(wire *codexRequest) int {
495 encoded, err := json.Marshal(wire)
496 if err != nil {
497 return 0
498 }
499 return len(encoded) / 4
500}
501
502func usesCodexCLIRouting(model string) bool {
503 switch model {
504 case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro":
505 return true
506 default:
507 return false
508 }
509}
510
511// ---- Streaming ----
512
513func (c *codexClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
514 req.Model = strings.TrimSpace(req.Model)
515 wire, err := c.buildRequest(req)
516 if err != nil {
517 return nil, err
518 }
519 ReportCacheDiagnostics(req.Lifecycle, responsesCacheDiagnostics(wire, "http_sse", "full"))
520 if c.modelName != nil {
521 wire.Model = c.modelName(wire.Model)
522 }
523 useCodexCLIRouting := !c.disableCLIRouting && (c.cliRoutingAll || usesCodexCLIRouting(wire.Model))
524 cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
525 if c.capabilities.StablePromptCacheKey && cacheSessionID != "" {
526 wire.PromptCacheKey = cacheSessionID
527 }
528 body, err := json.Marshal(wire)
529 if err != nil {
530 return nil, err
531 }
532
533 newReq := func() (*http.Request, error) {
534 httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL, bytes.NewReader(body))
535 if err != nil {
536 return nil, err
537 }
538 httpReq.Header = c.responsesHeaders(req, useCodexCLIRouting, false)
539 httpReq.Header.Set("content-type", "application/json")
540 httpReq.Header.Set("accept", "text/event-stream")
541 return httpReq, nil
542 }
543
544 resp, err := doStreamWithRetry(ctx, c.http, newReq, req.Lifecycle)
545 if err != nil {
546 label := c.providerName
547 if label == "" {
548 label = ProviderOpenAICodex
549 }
550 return nil, fmt.Errorf("%s: %w", label, err)
551 }
552 if resp.StatusCode != http.StatusOK {
553 b, _ := io.ReadAll(resp.Body)
554 resp.Body.Close()
555 label := c.providerName
556 if label == "" {
557 label = ProviderOpenAICodex
558 }
559 return nil, fmt.Errorf("%s: http %d: %s", label, resp.StatusCode, strings.TrimSpace(string(b)))
560 }
561
562 out := make(chan Event, 16)
563 go c.runStream(ctx, resp, req, out)
564 return out, nil
565}
566
567func (c *codexClient) responsesHeaders(req Request, useCodexCLIRouting, websocketMode bool) http.Header {
568 headers := make(http.Header)
569 headers.Set("authorization", "Bearer "+c.token)
570 if !websocketMode || c.accountID != "" {
571 headers.Set("openai-beta", "responses=experimental")
572 }
573 if c.accountID != "" {
574 headers.Set("chatgpt-account-id", c.accountID)
575 }
576
577 cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
578 threadID := strings.TrimSpace(req.Context.ThreadID)
579 if useCodexCLIRouting {
580 // The ChatGPT Codex backend only admits (and reliably serves)
581 // requests that follow Codex CLI routing metadata; other client
582 // identities are load-shed even when capacity is available.
583 headers.Set("originator", "codex_cli_rs")
584 if c.capabilities.SessionHeader && cacheSessionID != "" {
585 headers.Set("session-id", cacheSessionID)
586 }
587 if c.capabilities.ThreadHeader && threadID != "" {
588 headers.Set("thread-id", threadID)
589 }
590 if c.capabilities.ClientRequestHeader && threadID != "" {
591 headers.Set("x-client-request-id", threadID)
592 }
593 headers.Set("user-agent", "codex_cli_rs/0.0.0")
594 setOpenCodeGoHeaders(headers, c.providerName, req.Context)
595 return headers
596 }
597
598 // ChatGPT's backend recognizes these established values. They are
599 // external provider metadata, not zut product identifiers.
600 if c.accountID != "" {
601 headers.Set("originator", "zot")
602 headers.Set("user-agent", fmt.Sprintf("zot (%s %s)", runtime.GOOS, runtime.GOARCH))
603 }
604 setOpenCodeGoHeaders(headers, c.providerName, req.Context)
605 return headers
606}
607
608func (c *codexClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
609 defer close(out)
610 defer resp.Body.Close()
611
612 raw := make(chan sseEvent, 16)
613 go readSSE(resp.Body, raw)
614 c.runResponseEvents(ctx, req, out, raw, nil)
615}
616
617// runResponseEvents translates the Responses streaming event protocol into
618// provider events. Both HTTP/SSE and Responses WebSocket mode use the same
619// event vocabulary, so keeping the state machine here prevents the transports
620// from drifting in tool-call, reasoning, and usage handling.
621func (c *codexClient) runResponseEvents(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, completed func(string)) {
622 var observe func(string, Message)
623 if completed != nil {
624 observe = func(responseID string, _ Message) { completed(responseID) }
625 }
626 c.runResponseEventsWithFirst(ctx, req, out, raw, nil, observe)
627}
628
629func (c *codexClient) runResponseEventsWithFirst(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, first *sseEvent, completed func(string, Message)) {
630 model, _ := c.findModel(req.Model)
631 providerName := c.providerName
632 if providerName == "" {
633 providerName = ProviderOpenAICodex
634 }
635 out <- EventStart{Model: req.Model, Provider: providerName}
636
637 // Accumulators. The Responses API emits output_items in order; each
638 // item is either a "message" (text) or a "function_call". We track
639 // the in-flight item by its index.
640 type itemState struct {
641 kind string // "message" | "function_call" | "reasoning"
642 callID string
643 name string
644 argsBuf strings.Builder
645 textBuf strings.Builder
646 summary strings.Builder
647 rawID string
648 encrypted string
649 announced bool
650 }
651 var (
652 items = map[int]*itemState{}
653 order []int
654 usage Usage
655 stop = StopEnd
656 finalErr error
657 )
658
659 assemble := func() Message {
660 content := []Content{}
661 for _, idx := range order {
662 it := items[idx]
663 switch it.kind {
664 case "message":
665 if it.textBuf.Len() > 0 {
666 content = append(content, TextBlock{Text: it.textBuf.String()})
667 }
668 case "function_call":
669 args := it.argsBuf.String()
670 if args == "" || !json.Valid([]byte(args)) {
671 args = "{}"
672 }
673 content = append(content, ToolCallBlock{
674 ID: it.callID, Name: it.name, Arguments: json.RawMessage(args),
675 })
676 case "reasoning":
677 if it.encrypted == "" && it.summary.Len() == 0 && it.rawID == "" {
678 continue
679 }
680 content = append(content, ReasoningBlock{
681 ID: it.rawID,
682 Summary: it.summary.String(),
683 Encrypted: it.encrypted,
684 })
685 }
686 }
687 return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
688 }
689
690 sendDone := func() {
691 usage.CostUSD = ComputeCost(model, usage)
692 out <- EventUsage{Usage: usage}
693 out <- EventDone{Stop: stop, Err: finalErr, Message: assemble()}
694 }
695
696 for {
697 var ev sseEvent
698 if first != nil {
699 ev = *first
700 first = nil
701 } else {
702 select {
703 case <-ctx.Done():
704 stop = StopAborted
705 finalErr = ctx.Err()
706 sendDone()
707 return
708 case next, ok := <-raw:
709 if !ok {
710 sendDone()
711 return
712 }
713 ev = next
714 }
715 }
716 if ev.Err != nil {
717 stop = StopError
718 finalErr = fmt.Errorf("read SSE: %w", ev.Err)
719 sendDone()
720 return
721 }
722 var head struct {
723 Type string `json:"type"`
724 }
725 if err := json.Unmarshal([]byte(ev.Data), &head); err != nil {
726 continue
727 }
728
729 switch head.Type {
730 case "response.output_item.added":
731 var p struct {
732 OutputIndex int `json:"output_index"`
733 Item struct {
734 Type string `json:"type"` // "message" | "function_call" | "reasoning"
735 ID string `json:"id"`
736 CallID string `json:"call_id"`
737 Name string `json:"name"`
738 EncryptedContent string `json:"encrypted_content"`
739 } `json:"item"`
740 }
741 _ = json.Unmarshal([]byte(ev.Data), &p)
742 it := &itemState{}
743 switch p.Item.Type {
744 case "message":
745 it.kind = "message"
746 case "function_call":
747 it.kind = "function_call"
748 it.callID = p.Item.CallID
749 it.name = p.Item.Name
750 if !it.announced {
751 it.announced = true
752 out <- EventToolStart{ID: it.callID, Name: it.name}
753 }
754 case "reasoning":
755 it.kind = "reasoning"
756 it.rawID = p.Item.ID
757 it.encrypted = p.Item.EncryptedContent
758 default:
759 continue
760 }
761 items[p.OutputIndex] = it
762 order = append(order, p.OutputIndex)
763 case "response.output_text.delta":
764 var p struct {
765 OutputIndex int `json:"output_index"`
766 Delta string `json:"delta"`
767 }
768 _ = json.Unmarshal([]byte(ev.Data), &p)
769 if it, ok := items[p.OutputIndex]; ok && it.kind == "message" {
770 it.textBuf.WriteString(p.Delta)
771 out <- EventTextDelta{Delta: p.Delta}
772 }
773 case "response.reasoning_summary_text.delta":
774 var p struct {
775 OutputIndex int `json:"output_index"`
776 Delta string `json:"delta"`
777 }
778 _ = json.Unmarshal([]byte(ev.Data), &p)
779 if it, ok := items[p.OutputIndex]; ok && it.kind == "reasoning" {
780 it.summary.WriteString(p.Delta)
781 }
782 case "response.reasoning_summary_text.done":
783 // summary text already accumulated via deltas
784 case "response.function_call_arguments.delta":
785 var p struct {
786 OutputIndex int `json:"output_index"`
787 Delta string `json:"delta"`
788 }
789 _ = json.Unmarshal([]byte(ev.Data), &p)
790 if it, ok := items[p.OutputIndex]; ok && it.kind == "function_call" {
791 it.argsBuf.WriteString(p.Delta)
792 out <- EventToolArgs{ID: it.callID, Delta: p.Delta}
793 }
794 case "response.output_item.done":
795 var p struct {
796 OutputIndex int `json:"output_index"`
797 Item struct {
798 Type string `json:"type"`
799 ID string `json:"id"`
800 EncryptedContent string `json:"encrypted_content"`
801 Summary []struct {
802 Type string `json:"type"`
803 Text string `json:"text"`
804 } `json:"summary"`
805 } `json:"item"`
806 }
807 _ = json.Unmarshal([]byte(ev.Data), &p)
808 if it, ok := items[p.OutputIndex]; ok {
809 switch it.kind {
810 case "function_call":
811 out <- EventToolEnd{ID: it.callID}
812 case "reasoning":
813 // Fill-only: earlier streamed payloads and summaries win
814 // over later duplicates from output_item.done.
815 if it.encrypted == "" && p.Item.EncryptedContent != "" {
816 it.encrypted = p.Item.EncryptedContent
817 }
818 if it.rawID == "" && p.Item.ID != "" {
819 it.rawID = p.Item.ID
820 }
821 if it.summary.Len() == 0 {
822 for _, s := range p.Item.Summary {
823 if s.Text == "" {
824 continue
825 }
826 if it.summary.Len() > 0 {
827 it.summary.WriteString("\n")
828 }
829 it.summary.WriteString(s.Text)
830 }
831 }
832 }
833 }
834 case "response.completed", "response.done":
835 var p struct {
836 Response struct {
837 ID string `json:"id"`
838 Usage struct {
839 InputTokens int `json:"input_tokens"`
840 OutputTokens int `json:"output_tokens"`
841 InputTokensDetails *openAIInputTokensDetails `json:"input_tokens_details"`
842 OutputTokensDetails *struct {
843 ReasoningTokens int `json:"reasoning_tokens"`
844 } `json:"output_tokens_details"`
845 } `json:"usage"`
846 Status string `json:"status"`
847 EndTurn *bool `json:"end_turn"`
848 Output []struct {
849 Type string `json:"type"`
850 ID string `json:"id"`
851 EncryptedContent string `json:"encrypted_content"`
852 Summary []struct {
853 Type string `json:"type"`
854 Text string `json:"text"`
855 } `json:"summary"`
856 } `json:"output"`
857 } `json:"response"`
858 }
859 _ = json.Unmarshal([]byte(ev.Data), &p)
860 // Backfill encrypted reasoning supplied only in the terminal
861 // response output. Match by nonempty provider-issued item ID,
862 // never by output-array position. Fill a missing payload from
863 // a matching terminal item without overwriting an earlier
864 // payload or duplicating streamed summaries.
865 if len(p.Response.Output) > 0 {
866 terminal := make(map[string]struct {
867 encrypted string
868 summary string
869 }, len(p.Response.Output))
870 for _, out := range p.Response.Output {
871 if out.Type != "reasoning" || out.ID == "" {
872 continue
873 }
874 if _, seen := terminal[out.ID]; seen {
875 continue
876 }
877 var summary strings.Builder
878 for _, s := range out.Summary {
879 if s.Text == "" {
880 continue
881 }
882 if summary.Len() > 0 {
883 summary.WriteString("\n")
884 }
885 summary.WriteString(s.Text)
886 }
887 terminal[out.ID] = struct {
888 encrypted string
889 summary string
890 }{encrypted: out.EncryptedContent, summary: summary.String()}
891 }
892 for _, it := range items {
893 if it == nil || it.kind != "reasoning" || it.rawID == "" {
894 continue
895 }
896 match, ok := terminal[it.rawID]
897 if !ok {
898 continue
899 }
900 if it.encrypted == "" && match.encrypted != "" {
901 it.encrypted = match.encrypted
902 }
903 if it.summary.Len() == 0 && match.summary != "" {
904 it.summary.WriteString(match.summary)
905 }
906 }
907 }
908 usage = normalizeOpenAIUsage(p.Response.Usage.InputTokens, p.Response.Usage.OutputTokens, p.Response.Usage.InputTokensDetails)
909 if details := p.Response.Usage.OutputTokensDetails; details != nil {
910 usage.ReasoningTokens = details.ReasoningTokens
911 usage.ReasoningTokensKnown = true
912 }
913
914 hadTool := false
915 for _, it := range items {
916 if it.kind == "function_call" {
917 hadTool = true
918 break
919 }
920 }
921 if hadTool {
922 stop = StopToolUse
923 } else {
924 stop = StopEnd
925 if p.Response.EndTurn != nil && !*p.Response.EndTurn {
926 stop = StopContinue
927 }
928 }
929 if completed != nil && p.Response.ID != "" {
930 completed(p.Response.ID, assemble())
931 }
932 sendDone()
933 return
934 case "response.failed":
935 var p struct {
936 Response struct {
937 Error struct {
938 Message string `json:"message"`
939 } `json:"error"`
940 } `json:"response"`
941 }
942 _ = json.Unmarshal([]byte(ev.Data), &p)
943 stop = StopError
944 finalErr = fmt.Errorf("codex: %s", p.Response.Error.Message)
945 sendDone()
946 return
947 case "error":
948 var p struct {
949 Message string `json:"message"`
950 Code string `json:"code"`
951 Error struct {
952 Message string `json:"message"`
953 Code string `json:"code"`
954 } `json:"error"`
955 }
956 _ = json.Unmarshal([]byte(ev.Data), &p)
957 msg := p.Message
958 if msg == "" {
959 msg = p.Error.Message
960 }
961 if msg == "" {
962 msg = p.Code
963 }
964 if msg == "" {
965 msg = p.Error.Code
966 }
967 if msg == "" {
968 msg = strings.TrimSpace(ev.Data)
969 }
970 stop = StopError
971 label := c.errorLabel
972 if label == "" {
973 label = "codex"
974 }
975 finalErr = fmt.Errorf("%s error: %s", label, msg)
976 sendDone()
977 return
978 }
979 }
980}