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 Phase string `json:"phase,omitempty"`
130}
131
132type codexPromptCacheOptions struct {
133 Mode string `json:"mode"`
134}
135
136type codexPromptCacheBreakpoint struct {
137 Mode string `json:"mode"`
138}
139
140type codexInputMessage struct {
141 Type string `json:"type,omitempty"` // "message" (optional for input)
142 Role string `json:"role"`
143 Content []any `json:"content"`
144}
145
146type codexOutputMessage struct {
147 Type string `json:"type"` // "message"
148 Role string `json:"role"`
149 Status string `json:"status,omitempty"`
150 ID string `json:"id,omitempty"`
151 Content []codexOutputText `json:"content"`
152}
153
154type codexFunctionCall struct {
155 Type string `json:"type"` // "function_call"
156 ID string `json:"id,omitempty"`
157 CallID string `json:"call_id"`
158 Name string `json:"name"`
159 Arguments string `json:"arguments"` // JSON string
160}
161
162type codexFunctionCallOutput struct {
163 Type string `json:"type"` // "function_call_output"
164 CallID string `json:"call_id"`
165 Output string `json:"output"` // string (or ResponseFunctionCallOutputItemList for images; v1 only uses string)
166}
167
168// codexReasoningItem mirrors the Responses API "reasoning" output item.
169// We capture it on incoming streams and replay it verbatim on follow-up
170// requests: the API rejects assistant tool-call replays without it when
171// thinking is enabled.
172type codexReasoningItem struct {
173 Type string `json:"type"` // "reasoning"
174 ID string `json:"id,omitempty"`
175 EncryptedContent string `json:"encrypted_content,omitempty"`
176 // Summary is required by the Responses API even when no summary text
177 // was streamed; encode an empty array rather than omitting the field.
178 Summary []codexReasoningSummary `json:"summary"`
179}
180
181type codexReasoningSummary struct {
182 Type string `json:"type"` // "summary_text"
183 Text string `json:"text"`
184}
185
186type codexTool struct {
187 Type string `json:"type"` // "function"
188 Name string `json:"name"`
189 Description string `json:"description,omitempty"`
190 Parameters json.RawMessage `json:"parameters"`
191 // Responses otherwise attempts to normalize schemas into strict mode,
192 // which makes non-nullable optional properties required model output.
193 Strict bool `json:"strict"`
194}
195
196type codexReasoningConfig struct {
197 Effort string `json:"effort,omitempty"`
198}
199
200// supportsStrictToolSchema reports whether the top-level object already meets
201// Responses strict-mode requirements. Keep this deliberately narrow: schemas
202// with nested containers or composition remain non-strict rather than risking
203// provider rejection or changing optional fields into required ones.
204func supportsStrictToolSchema(raw json.RawMessage) bool {
205 var schema struct {
206 Type string `json:"type"`
207 Properties map[string]json.RawMessage `json:"properties"`
208 Required []string `json:"required"`
209 AdditionalProperties *bool `json:"additionalProperties"`
210 }
211 if json.Unmarshal(raw, &schema) != nil || schema.Type != "object" || schema.AdditionalProperties == nil || *schema.AdditionalProperties {
212 return false
213 }
214 required := make(map[string]struct{}, len(schema.Required))
215 for _, name := range schema.Required {
216 required[name] = struct{}{}
217 }
218 if len(required) != len(schema.Properties) {
219 return false
220 }
221 for name, rawProperty := range schema.Properties {
222 if _, ok := required[name]; !ok {
223 return false
224 }
225 var property struct {
226 Type string `json:"type"`
227 }
228 if json.Unmarshal(rawProperty, &property) != nil {
229 return false
230 }
231 switch property.Type {
232 case "string", "number", "integer", "boolean":
233 default:
234 return false
235 }
236 }
237 return true
238}
239
240type codexRequest struct {
241 Model string `json:"model"`
242 Store bool `json:"store"`
243 Stream bool `json:"stream,omitempty"`
244 Instructions string `json:"instructions,omitempty"`
245 Input []any `json:"input"`
246 Tools []codexTool `json:"tools,omitempty"`
247 ToolChoice string `json:"tool_choice,omitempty"`
248 ParallelToolCalls bool `json:"parallel_tool_calls"`
249 Include []string `json:"include,omitempty"`
250 Reasoning *codexReasoningConfig `json:"reasoning,omitempty"`
251 PromptCacheKey string `json:"prompt_cache_key,omitempty"`
252 PromptCacheOptions *codexPromptCacheOptions `json:"prompt_cache_options,omitempty"`
253 ServiceTier string `json:"service_tier,omitempty"`
254 PreviousResponseID string `json:"previous_response_id,omitempty"`
255}
256
257// ---- Request building ----
258
259func (c *codexClient) findModel(id string) (Model, error) {
260 id = strings.TrimSpace(id)
261 c.modelOverridesMu.RLock()
262 model, ok := c.modelOverrides[id]
263 if ok {
264 model = cloneModel(model)
265 }
266 c.modelOverridesMu.RUnlock()
267 if ok {
268 return model, nil
269 }
270 if c.providerName != "" && c.providerName != ProviderOpenAICodex {
271 if m, err := FindModel(c.providerName, id); err == nil {
272 return m, nil
273 }
274 }
275 if m, err := FindModel(ProviderOpenAICodex, id); err == nil {
276 return m, nil
277 }
278 m, err := FindModel(ProviderOpenAI, id)
279 if err == nil {
280 return m, nil
281 }
282 if c.providerName == ProviderOpenCodeGo {
283 if api := openCodeGoAPIForModel(id); api != "" {
284 return Model{
285 Provider: ProviderOpenCodeGo,
286 ID: id,
287 API: api,
288 ContextWindow: 128000,
289 MaxOutput: 16384,
290 Reasoning: true,
291 }, nil
292 }
293 }
294 return m, err
295}
296
297func supportsOpenAIExplicitPromptCache(model string) bool {
298 id := strings.ToLower(strings.TrimSpace(model))
299 return strings.HasPrefix(id, "gpt-5.6-") || id == "gpt-6-astra"
300}
301
302func (c *codexClient) buildRequest(req Request) (*codexRequest, error) {
303 req.Model = strings.TrimSpace(req.Model)
304 providerName := c.providerName
305 if providerName == "" {
306 providerName = ProviderOpenAICodex
307 }
308 if err := ValidateFastMode(providerName, req.FastMode); err != nil {
309 return nil, err
310 }
311 m, err := c.findModel(req.Model)
312 if err != nil {
313 return nil, err
314 }
315 reasoning := ClampReasoningForModel(m, req.Reasoning)
316
317 body := &codexRequest{
318 Model: req.Model,
319 Store: false,
320 Stream: true,
321 Instructions: req.SystemPrompt(),
322 ParallelToolCalls: true,
323 Include: []string{"reasoning.encrypted_content"},
324 }
325 if req.FastMode {
326 body.ServiceTier = fastModeServiceTier
327 }
328 if c.capabilities.ExplicitPromptCacheBreakpoint && supportsOpenAIExplicitPromptCache(req.Model) {
329 // Keep implicit mode so OpenAI can retain later conversation
330 // checkpoints as well as this stable instruction/tool boundary.
331 body.PromptCacheOptions = &codexPromptCacheOptions{Mode: "implicit"}
332 body.Input = append(body.Input, codexInputMessage{
333 Role: "developer",
334 Content: []any{codexInputText{
335 Type: "input_text",
336 Text: "[Stable prompt boundary]",
337 PromptCacheBreakpoint: &codexPromptCacheBreakpoint{Mode: "explicit"},
338 }},
339 })
340 }
341 if m.Reasoning {
342 effort := OpenAICodexReasoningEffort(reasoning, req.Model)
343 effort = reasoningEffortForModel(m, req.Reasoning, reasoning, effort)
344 if effort != "" {
345 body.Reasoning = &codexReasoningConfig{Effort: effort}
346 }
347 }
348 activeTools := activeToolDefinitions(req.Tools, req.Messages)
349 if len(activeTools) > 0 {
350 body.ToolChoice = "auto"
351 for _, t := range activeTools {
352 params := t.Schema
353 if len(params) == 0 {
354 params = json.RawMessage(`{"type":"object","properties":{}}`)
355 }
356 body.Tools = append(body.Tools, codexTool{
357 Type: "function",
358 Name: t.Name,
359 Description: t.Description,
360 Parameters: params,
361 Strict: providerName == ProviderOpenAICodex && supportsStrictToolSchema(params),
362 })
363 }
364 }
365
366 msgIdx := 0
367 req.Messages = RepairOrphanedToolResults(req.Messages)
368 for _, msg := range req.Messages {
369 switch msg.Role {
370 case RoleDeveloper, RoleUser:
371 content := []any{}
372 for _, c := range msg.Content {
373 switch v := c.(type) {
374 case TextBlock:
375 if v.Text != "" {
376 content = append(content, codexInputText{Type: "input_text", Text: v.Text})
377 }
378 case ImageBlock:
379 url := "data:" + v.MimeType + ";base64," + base64.StdEncoding.EncodeToString(v.Data)
380 content = append(content, codexInputImage{Type: "input_image", Detail: "auto", ImageURL: url})
381 }
382 }
383 if len(content) == 0 {
384 continue
385 }
386 role := "user"
387 if msg.Role == RoleDeveloper {
388 role = "developer"
389 }
390 body.Input = append(body.Input, codexInputMessage{Role: role, Content: content})
391 case RoleAssistant:
392 // Emit one output_message per text block, one function_call per
393 // tool call, and one reasoning item per replayable ReasoningBlock,
394 // preserving the order so the model sees the same interleaving
395 // we captured. The reasoning replay is what keeps Responses
396 // providers from rejecting follow-up tool calls when thinking
397 // is enabled. Blocks without encrypted content are omitted:
398 // an ID or summary alone is not a self-contained reasoning
399 // item while store is false. See reasoningForReplay.
400 for _, c := range replayContent(msg, providerName, req.Model) {
401 switch v := c.(type) {
402 case ReasoningBlock:
403 item, ok := reasoningForReplay(v)
404 if !ok {
405 continue
406 }
407 body.Input = append(body.Input, item)
408 case TextBlock:
409 if v.Text == "" {
410 continue
411 }
412 msgIdx++
413 out := codexOutputText{Type: "output_text", Text: v.Text, Annotations: []any{}}
414 if IsRecognizedTextPhase(v.Phase) {
415 out.Phase = v.Phase
416 }
417 body.Input = append(body.Input, codexOutputMessage{
418 Type: "message",
419 Role: "assistant",
420 Status: "completed",
421 ID: fmt.Sprintf("msg_%d", msgIdx),
422 Content: []codexOutputText{out},
423 })
424 case ToolCallBlock:
425 args := string(v.Arguments)
426 if args == "" || !json.Valid([]byte(args)) {
427 args = "{}"
428 }
429 callID, _ := splitCallID(v.ID)
430 body.Input = append(body.Input, codexFunctionCall{
431 Type: "function_call",
432 CallID: callID,
433 Name: v.Name,
434 Arguments: args,
435 })
436 }
437 }
438 case RoleTool:
439 for _, c := range msg.Content {
440 if tr, ok := c.(ToolResultBlock); ok {
441 var text strings.Builder
442 imageCount := 0
443 for _, inner := range tr.Content {
444 switch v := inner.(type) {
445 case TextBlock:
446 if text.Len() > 0 {
447 text.WriteString("\n")
448 }
449 text.WriteString(v.Text)
450 case ImageBlock:
451 imageCount++
452 }
453 }
454 // The Responses API function_call_output only carries a
455 // string, so image bytes cannot ride along here. The agent
456 // loop mirrors any tool-result images into the following
457 // user message (which this client does serialize as
458 // input_image). Leave a short text note so an image-only
459 // result is not an empty output the API may reject, and so
460 // the model knows the image arrives next.
461 out := text.String()
462 if out == "" && imageCount > 0 {
463 if imageCount == 1 {
464 out = "[image returned; see the following message]"
465 } else {
466 out = fmt.Sprintf("[%d images returned; see the following message]", imageCount)
467 }
468 }
469 callID, _ := splitCallID(tr.CallID)
470 body.Input = append(body.Input, codexFunctionCallOutput{
471 Type: "function_call_output",
472 CallID: callID,
473 Output: out,
474 })
475 }
476 }
477 }
478 }
479
480 return body, nil
481}
482
483// reasoningForReplay decides whether a persisted ReasoningBlock carries a
484// self-contained Responses payload. Only blocks with encrypted content are
485// replayed; an ID or a readable summary alone cannot stand in for the
486// provider-issued payload when store is false. The boolean is false when
487// the block must be omitted from outgoing Responses input. Stored messages
488// are never mutated by this decision.
489func reasoningForReplay(block ReasoningBlock) (codexReasoningItem, bool) {
490 if block.Encrypted == "" {
491 return codexReasoningItem{}, false
492 }
493 // Gateway composite IDs and other providers' signatures are not native
494 // Responses items. Never rewrite an opaque ID to make it look valid.
495 for _, ch := range block.ID {
496 switch {
497 case ch >= 'a' && ch <= 'z', ch >= 'A' && ch <= 'Z',
498 ch >= '0' && ch <= '9', ch == '_', ch == '-':
499 // Responses item IDs use ASCII letters, digits, underscores, or dashes.
500 default:
501 return codexReasoningItem{}, false
502 }
503 }
504 item := codexReasoningItem{
505 Type: "reasoning",
506 ID: block.ID,
507 EncryptedContent: block.Encrypted,
508 Summary: []codexReasoningSummary{},
509 }
510 if block.Summary != "" {
511 item.Summary = []codexReasoningSummary{{Type: "summary_text", Text: block.Summary}}
512 }
513 return item, true
514}
515
516func splitCallID(id string) (string, string) {
517 if i := strings.Index(id, "|"); i >= 0 {
518 return id[:i], id[i+1:]
519 }
520 return id, ""
521}
522
523// messageItemPhase extracts the Responses message phase from either the
524// item-level phase field or the first nonempty phase on its output_text
525// content parts. The return is fill-only input for itemState.phase: empty
526// means absent, never a reason to erase an observed value. Unrecognized
527// values are preserved so storage stays faithful; IsRecognizedTextPhase
528// gates interpretation and replay.
529func messageItemPhase(itemPhase string, content []struct {
530 Type string `json:"type"`
531 Text string `json:"text"`
532 Phase string `json:"phase"`
533}) string {
534 if itemPhase != "" {
535 return itemPhase
536 }
537 for _, part := range content {
538 if part.Phase != "" {
539 return part.Phase
540 }
541 }
542 return ""
543}
544
545const minOpenAIPromptCacheTokens = 1024
546
547func responsesCacheDiagnostics(wire *codexRequest, transport, continuation string) CacheDiagnostics {
548 diagnostics := CacheDiagnostics{Mode: "none", Transport: transport, Continuation: continuation}
549 if wire == nil || wire.PromptCacheOptions == nil {
550 return diagnostics
551 }
552 diagnostics.Mode = wire.PromptCacheOptions.Mode
553 diagnostics.Eligible = estimatedOpenAIPromptTokens(wire) >= minOpenAIPromptCacheTokens
554 return diagnostics
555}
556
557// estimatedOpenAIPromptTokens supplies a deliberately conservative local
558// eligibility signal. OpenAI makes the final tokenization and cache decision;
559// this estimate never leaves the process and is not persisted in diagnostics.
560func estimatedOpenAIPromptTokens(wire *codexRequest) int {
561 encoded, err := json.Marshal(wire)
562 if err != nil {
563 return 0
564 }
565 return len(encoded) / 4
566}
567
568func usesCodexCLIRouting(model string) bool {
569 switch model {
570 case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro":
571 return true
572 default:
573 return false
574 }
575}
576
577// ---- Streaming ----
578
579func (c *codexClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
580 req.Model = strings.TrimSpace(req.Model)
581 wire, err := c.buildRequest(req)
582 if err != nil {
583 return nil, err
584 }
585 ReportCacheDiagnostics(req.Lifecycle, responsesCacheDiagnostics(wire, "http_sse", "full"))
586 if c.modelName != nil {
587 wire.Model = c.modelName(wire.Model)
588 }
589 useCodexCLIRouting := !c.disableCLIRouting && (c.cliRoutingAll || usesCodexCLIRouting(wire.Model))
590 cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
591 if c.capabilities.StablePromptCacheKey && cacheSessionID != "" {
592 wire.PromptCacheKey = cacheSessionID
593 }
594 body, err := json.Marshal(wire)
595 if err != nil {
596 return nil, err
597 }
598
599 newReq := func() (*http.Request, error) {
600 httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL, bytes.NewReader(body))
601 if err != nil {
602 return nil, err
603 }
604 httpReq.Header = c.responsesHeaders(req, useCodexCLIRouting, false)
605 httpReq.Header.Set("content-type", "application/json")
606 httpReq.Header.Set("accept", "text/event-stream")
607 return httpReq, nil
608 }
609
610 resp, err := doStreamWithRetry(ctx, c.http, newReq, req.Lifecycle)
611 if err != nil {
612 label := c.providerName
613 if label == "" {
614 label = ProviderOpenAICodex
615 }
616 return nil, fmt.Errorf("%s: %w", label, err)
617 }
618 if resp.StatusCode != http.StatusOK {
619 b, _ := io.ReadAll(resp.Body)
620 resp.Body.Close()
621 label := c.providerName
622 if label == "" {
623 label = ProviderOpenAICodex
624 }
625 return nil, fmt.Errorf("%s: http %d: %s", label, resp.StatusCode, strings.TrimSpace(string(b)))
626 }
627
628 out := make(chan Event, 16)
629 go c.runStream(ctx, resp, req, out)
630 return out, nil
631}
632
633func (c *codexClient) responsesHeaders(req Request, useCodexCLIRouting, websocketMode bool) http.Header {
634 headers := make(http.Header)
635 headers.Set("authorization", "Bearer "+c.token)
636 if !websocketMode || c.accountID != "" {
637 headers.Set("openai-beta", "responses=experimental")
638 }
639 if c.accountID != "" {
640 headers.Set("chatgpt-account-id", c.accountID)
641 }
642
643 cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
644 threadID := strings.TrimSpace(req.Context.ThreadID)
645 if useCodexCLIRouting {
646 // The ChatGPT Codex backend only admits (and reliably serves)
647 // requests that follow Codex CLI routing metadata; other client
648 // identities are load-shed even when capacity is available.
649 headers.Set("originator", "codex_cli_rs")
650 if c.capabilities.SessionHeader && cacheSessionID != "" {
651 headers.Set("session-id", cacheSessionID)
652 }
653 if c.capabilities.ThreadHeader && threadID != "" {
654 headers.Set("thread-id", threadID)
655 }
656 if c.capabilities.ClientRequestHeader && threadID != "" {
657 headers.Set("x-client-request-id", threadID)
658 }
659 headers.Set("user-agent", "codex_cli_rs/0.0.0")
660 setOpenCodeGoHeaders(headers, c.providerName, req.Context)
661 return headers
662 }
663
664 // ChatGPT's backend recognizes these established values. They are
665 // external provider metadata, not zut product identifiers.
666 if c.accountID != "" {
667 headers.Set("originator", "zot")
668 headers.Set("user-agent", fmt.Sprintf("zot (%s %s)", runtime.GOOS, runtime.GOARCH))
669 }
670 setOpenCodeGoHeaders(headers, c.providerName, req.Context)
671 return headers
672}
673
674func (c *codexClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
675 defer close(out)
676 defer resp.Body.Close()
677
678 raw := make(chan sseEvent, 16)
679 go readSSE(resp.Body, raw)
680 c.runResponseEvents(ctx, req, out, raw, nil)
681}
682
683// runResponseEvents translates the Responses streaming event protocol into
684// provider events. Both HTTP/SSE and Responses WebSocket mode use the same
685// event vocabulary, so keeping the state machine here prevents the transports
686// from drifting in tool-call, reasoning, and usage handling.
687func (c *codexClient) runResponseEvents(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, completed func(string)) {
688 var observe func(string, Message)
689 if completed != nil {
690 observe = func(responseID string, _ Message) { completed(responseID) }
691 }
692 c.runResponseEventsWithFirst(ctx, req, out, raw, nil, observe)
693}
694
695func (c *codexClient) runResponseEventsWithFirst(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, first *sseEvent, completed func(string, Message)) {
696 model, _ := c.findModel(req.Model)
697 providerName := c.providerName
698 if providerName == "" {
699 providerName = ProviderOpenAICodex
700 }
701 out <- EventStart{Model: req.Model, Provider: providerName}
702
703 // Accumulators. The Responses API emits output_items in order; each
704 // item is either a "message" (text) or a "function_call". We track
705 // the in-flight item by its index.
706 type itemState struct {
707 kind string // "message" | "function_call" | "reasoning"
708 callID string
709 name string
710 argsBuf strings.Builder
711 textBuf strings.Builder
712 summary strings.Builder
713 rawID string
714 encrypted string
715 // phase is the Responses output phase observed for a message item.
716 // It is fill-only: an absent later field must never erase an
717 // observed value. Any observed value (including unrecognized ones)
718 // is retained in storage; only recognized values are replayed.
719 phase string
720 announced bool
721 }
722 var (
723 items = map[int]*itemState{}
724 order []int
725 usage Usage
726 stop = StopEnd
727 finalErr error
728 )
729
730 assemble := func() Message {
731 content := []Content{}
732 for _, idx := range order {
733 it := items[idx]
734 switch it.kind {
735 case "message":
736 if it.textBuf.Len() > 0 {
737 content = append(content, TextBlock{Text: it.textBuf.String(), Phase: it.phase})
738 }
739 case "function_call":
740 args := it.argsBuf.String()
741 if args == "" || !json.Valid([]byte(args)) {
742 args = "{}"
743 }
744 content = append(content, ToolCallBlock{
745 ID: it.callID, Name: it.name, Arguments: json.RawMessage(args),
746 })
747 case "reasoning":
748 if it.encrypted == "" && it.summary.Len() == 0 && it.rawID == "" {
749 continue
750 }
751 content = append(content, ReasoningBlock{
752 ID: it.rawID,
753 Summary: it.summary.String(),
754 Encrypted: it.encrypted,
755 })
756 }
757 }
758 return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
759 }
760
761 sendDone := func() {
762 usage.CostUSD = ComputeCost(model, usage)
763 out <- EventUsage{Usage: usage}
764 out <- EventDone{Stop: stop, Err: finalErr, Message: assemble()}
765 }
766
767 for {
768 var ev sseEvent
769 if first != nil {
770 ev = *first
771 first = nil
772 } else {
773 select {
774 case <-ctx.Done():
775 stop = StopAborted
776 finalErr = ctx.Err()
777 sendDone()
778 return
779 case next, ok := <-raw:
780 if !ok {
781 sendDone()
782 return
783 }
784 ev = next
785 }
786 }
787 if ev.Err != nil {
788 stop = StopError
789 finalErr = fmt.Errorf("read SSE: %w", ev.Err)
790 sendDone()
791 return
792 }
793 var head struct {
794 Type string `json:"type"`
795 }
796 if err := json.Unmarshal([]byte(ev.Data), &head); err != nil {
797 continue
798 }
799
800 switch head.Type {
801 case "response.output_item.added":
802 var p struct {
803 OutputIndex int `json:"output_index"`
804 Item struct {
805 Type string `json:"type"` // "message" | "function_call" | "reasoning"
806 ID string `json:"id"`
807 CallID string `json:"call_id"`
808 Name string `json:"name"`
809 EncryptedContent string `json:"encrypted_content"`
810 Phase string `json:"phase"`
811 Content []struct {
812 Type string `json:"type"`
813 Text string `json:"text"`
814 Phase string `json:"phase"`
815 } `json:"content"`
816 } `json:"item"`
817 }
818 _ = json.Unmarshal([]byte(ev.Data), &p)
819 it := &itemState{}
820 switch p.Item.Type {
821 case "message":
822 it.kind = "message"
823 it.phase = messageItemPhase(p.Item.Phase, p.Item.Content)
824 case "function_call":
825 it.kind = "function_call"
826 it.callID = p.Item.CallID
827 it.name = p.Item.Name
828 if !it.announced {
829 it.announced = true
830 out <- EventToolStart{ID: it.callID, Name: it.name}
831 }
832 case "reasoning":
833 it.kind = "reasoning"
834 it.rawID = p.Item.ID
835 it.encrypted = p.Item.EncryptedContent
836 default:
837 continue
838 }
839 items[p.OutputIndex] = it
840 order = append(order, p.OutputIndex)
841 case "response.output_text.delta":
842 var p struct {
843 OutputIndex int `json:"output_index"`
844 Delta string `json:"delta"`
845 }
846 _ = json.Unmarshal([]byte(ev.Data), &p)
847 if it, ok := items[p.OutputIndex]; ok && it.kind == "message" {
848 it.textBuf.WriteString(p.Delta)
849 out <- EventTextDelta{Delta: p.Delta}
850 }
851 case "response.reasoning_summary_text.delta":
852 var p struct {
853 OutputIndex int `json:"output_index"`
854 Delta string `json:"delta"`
855 }
856 _ = json.Unmarshal([]byte(ev.Data), &p)
857 if it, ok := items[p.OutputIndex]; ok && it.kind == "reasoning" {
858 it.summary.WriteString(p.Delta)
859 }
860 case "response.reasoning_summary_text.done":
861 // summary text already accumulated via deltas
862 case "response.function_call_arguments.delta":
863 var p struct {
864 OutputIndex int `json:"output_index"`
865 Delta string `json:"delta"`
866 }
867 _ = json.Unmarshal([]byte(ev.Data), &p)
868 if it, ok := items[p.OutputIndex]; ok && it.kind == "function_call" {
869 it.argsBuf.WriteString(p.Delta)
870 out <- EventToolArgs{ID: it.callID, Delta: p.Delta}
871 }
872 case "response.output_item.done":
873 var p struct {
874 OutputIndex int `json:"output_index"`
875 Item struct {
876 Type string `json:"type"`
877 ID string `json:"id"`
878 EncryptedContent string `json:"encrypted_content"`
879 Phase string `json:"phase"`
880 Content []struct {
881 Type string `json:"type"`
882 Text string `json:"text"`
883 Phase string `json:"phase"`
884 } `json:"content"`
885 Summary []struct {
886 Type string `json:"type"`
887 Text string `json:"text"`
888 } `json:"summary"`
889 } `json:"item"`
890 }
891 _ = json.Unmarshal([]byte(ev.Data), &p)
892 if it, ok := items[p.OutputIndex]; ok {
893 switch it.kind {
894 case "function_call":
895 out <- EventToolEnd{ID: it.callID}
896 case "message":
897 // Fill-only: an absent phase on item-done must not
898 // erase a phase already observed on item-added.
899 if it.phase == "" {
900 it.phase = messageItemPhase(p.Item.Phase, p.Item.Content)
901 }
902 case "reasoning":
903 // Fill-only: earlier streamed payloads and summaries win
904 // over later duplicates from output_item.done.
905 if it.encrypted == "" && p.Item.EncryptedContent != "" {
906 it.encrypted = p.Item.EncryptedContent
907 }
908 if it.rawID == "" && p.Item.ID != "" {
909 it.rawID = p.Item.ID
910 }
911 if it.summary.Len() == 0 {
912 for _, s := range p.Item.Summary {
913 if s.Text == "" {
914 continue
915 }
916 if it.summary.Len() > 0 {
917 it.summary.WriteString("\n")
918 }
919 it.summary.WriteString(s.Text)
920 }
921 }
922 }
923 }
924 case "response.completed", "response.done":
925 var p struct {
926 Response struct {
927 ID string `json:"id"`
928 Usage struct {
929 InputTokens int `json:"input_tokens"`
930 OutputTokens int `json:"output_tokens"`
931 InputTokensDetails *openAIInputTokensDetails `json:"input_tokens_details"`
932 OutputTokensDetails *struct {
933 ReasoningTokens int `json:"reasoning_tokens"`
934 } `json:"output_tokens_details"`
935 } `json:"usage"`
936 Status string `json:"status"`
937 EndTurn *bool `json:"end_turn"`
938 Output []struct {
939 Type string `json:"type"`
940 ID string `json:"id"`
941 EncryptedContent string `json:"encrypted_content"`
942 Phase string `json:"phase"`
943 Content []struct {
944 Type string `json:"type"`
945 Text string `json:"text"`
946 Phase string `json:"phase"`
947 } `json:"content"`
948 Summary []struct {
949 Type string `json:"type"`
950 Text string `json:"text"`
951 } `json:"summary"`
952 } `json:"output"`
953 } `json:"response"`
954 }
955 _ = json.Unmarshal([]byte(ev.Data), &p)
956 // Backfill encrypted reasoning supplied only in the terminal
957 // response output. Match by nonempty provider-issued item ID,
958 // never by output-array position. Fill a missing payload from
959 // a matching terminal item without overwriting an earlier
960 // payload or duplicating streamed summaries. Message phase is
961 // reconciled the same fill-only way, but by output-array
962 // position: terminal text content without a phase must not
963 // erase a phase observed during streaming.
964 if len(p.Response.Output) > 0 {
965 terminal := make(map[string]struct {
966 encrypted string
967 summary string
968 }, len(p.Response.Output))
969 terminalPhases := make([]string, 0, len(p.Response.Output))
970 for _, out := range p.Response.Output {
971 if out.Type == "message" {
972 terminalPhases = append(terminalPhases, messageItemPhase(out.Phase, out.Content))
973 }
974 if out.Type != "reasoning" || out.ID == "" {
975 continue
976 }
977 if _, seen := terminal[out.ID]; seen {
978 continue
979 }
980 var summary strings.Builder
981 for _, s := range out.Summary {
982 if s.Text == "" {
983 continue
984 }
985 if summary.Len() > 0 {
986 summary.WriteString("\n")
987 }
988 summary.WriteString(s.Text)
989 }
990 terminal[out.ID] = struct {
991 encrypted string
992 summary string
993 }{encrypted: out.EncryptedContent, summary: summary.String()}
994 }
995 for _, it := range items {
996 if it == nil || it.kind != "reasoning" || it.rawID == "" {
997 continue
998 }
999 match, ok := terminal[it.rawID]
1000 if !ok {
1001 continue
1002 }
1003 if it.encrypted == "" && match.encrypted != "" {
1004 it.encrypted = match.encrypted
1005 }
1006 if it.summary.Len() == 0 && match.summary != "" {
1007 it.summary.WriteString(match.summary)
1008 }
1009 }
1010 // Reconcile message phases by output-array position: the
1011 // n-th terminal message corresponds to the n-th streamed
1012 // message item. Fill only; never erase streamed phases.
1013 pos := 0
1014 for _, idx := range order {
1015 it := items[idx]
1016 if it == nil || it.kind != "message" {
1017 continue
1018 }
1019 if pos >= len(terminalPhases) {
1020 break
1021 }
1022 if it.phase == "" {
1023 it.phase = terminalPhases[pos]
1024 }
1025 pos++
1026 }
1027 }
1028 usage = normalizeOpenAIUsage(p.Response.Usage.InputTokens, p.Response.Usage.OutputTokens, p.Response.Usage.InputTokensDetails)
1029 if details := p.Response.Usage.OutputTokensDetails; details != nil {
1030 usage.ReasoningTokens = details.ReasoningTokens
1031 usage.ReasoningTokensKnown = true
1032 }
1033
1034 hadTool := false
1035 for _, it := range items {
1036 if it.kind == "function_call" {
1037 hadTool = true
1038 break
1039 }
1040 }
1041 if hadTool {
1042 stop = StopToolUse
1043 } else {
1044 stop = StopEnd
1045 if p.Response.EndTurn != nil && !*p.Response.EndTurn {
1046 stop = StopContinue
1047 }
1048 }
1049 if completed != nil && p.Response.ID != "" {
1050 completed(p.Response.ID, assemble())
1051 }
1052 sendDone()
1053 return
1054 case "response.failed":
1055 var p struct {
1056 Response struct {
1057 Error struct {
1058 Message string `json:"message"`
1059 } `json:"error"`
1060 } `json:"response"`
1061 }
1062 _ = json.Unmarshal([]byte(ev.Data), &p)
1063 stop = StopError
1064 finalErr = fmt.Errorf("codex: %s", p.Response.Error.Message)
1065 sendDone()
1066 return
1067 case "error":
1068 var p struct {
1069 Message string `json:"message"`
1070 Code string `json:"code"`
1071 Error struct {
1072 Message string `json:"message"`
1073 Code string `json:"code"`
1074 } `json:"error"`
1075 }
1076 _ = json.Unmarshal([]byte(ev.Data), &p)
1077 msg := p.Message
1078 if msg == "" {
1079 msg = p.Error.Message
1080 }
1081 if msg == "" {
1082 msg = p.Code
1083 }
1084 if msg == "" {
1085 msg = p.Error.Code
1086 }
1087 if msg == "" {
1088 msg = strings.TrimSpace(ev.Data)
1089 }
1090 stop = StopError
1091 label := c.errorLabel
1092 if label == "" {
1093 label = "codex"
1094 }
1095 finalErr = fmt.Errorf("%s error: %s", label, msg)
1096 sendDone()
1097 return
1098 }
1099 }
1100}