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-") || strings.HasPrefix(id, "gpt-6-")
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
568// unsupportedPromptCacheRetention reports whether a 400 response is the
569// backend rejecting the prompt_cache_key retention the Codex CLI identity
570// selects. Only that structured error retries with the native identity;
571// every other status, param, or code is returned as-is.
572func unsupportedPromptCacheRetention(status int, body []byte) bool {
573 if status != http.StatusBadRequest {
574 return false
575 }
576 var payload struct {
577 Error struct {
578 Message string `json:"message"`
579 Param string `json:"param"`
580 Code string `json:"code"`
581 } `json:"error"`
582 }
583 if json.Unmarshal(body, &payload) != nil || payload.Error.Param != "prompt_cache_retention" {
584 return false
585 }
586 message := strings.ToLower(payload.Error.Message)
587 return payload.Error.Code == "invalid_parameter" || strings.Contains(message, "not supported")
588}
589
590func usesCodexCLIRouting(model string) bool {
591 switch model {
592 case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro":
593 return true
594 default:
595 return false
596 }
597}
598
599// ---- Streaming ----
600
601func (c *codexClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
602 req.Model = strings.TrimSpace(req.Model)
603 wire, err := c.buildRequest(req)
604 if err != nil {
605 return nil, err
606 }
607 ReportCacheDiagnostics(req.Lifecycle, responsesCacheDiagnostics(wire, "http_sse", "full"))
608 if c.modelName != nil {
609 wire.Model = c.modelName(wire.Model)
610 }
611 useCodexCLIRouting := !c.disableCLIRouting && (c.cliRoutingAll || usesCodexCLIRouting(wire.Model))
612 cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
613 if c.capabilities.StablePromptCacheKey && cacheSessionID != "" {
614 wire.PromptCacheKey = cacheSessionID
615 }
616 body, err := json.Marshal(wire)
617 if err != nil {
618 return nil, err
619 }
620
621 newReq := func(useCodexIdentity bool) func() (*http.Request, error) {
622 return func() (*http.Request, error) {
623 httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL, bytes.NewReader(body))
624 if err != nil {
625 return nil, err
626 }
627 httpReq.Header = c.responsesHeaders(req, useCodexIdentity, false)
628 if !useCodexIdentity {
629 // The native identity shape carries no session/request
630 // correlation headers; keep the values selected under the
631 // CLI identity so the retry stays on the same session and
632 // request correlation.
633 cliHeaders := c.responsesHeaders(req, true, false)
634 for _, key := range []string{"session-id", "thread-id", "x-client-request-id"} {
635 if value := cliHeaders.Get(key); value != "" {
636 httpReq.Header.Set(key, value)
637 }
638 }
639 }
640 httpReq.Header.Set("content-type", "application/json")
641 httpReq.Header.Set("accept", "text/event-stream")
642 return httpReq, nil
643 }
644 }
645
646 resp, err := doStreamWithRetry(ctx, c.http, newReq(useCodexCLIRouting), req.Lifecycle)
647 if err != nil {
648 label := c.providerName
649 if label == "" {
650 label = ProviderOpenAICodex
651 }
652 return nil, fmt.Errorf("%s: %w", label, err)
653 }
654 if resp.StatusCode != http.StatusOK {
655 status := resp.StatusCode
656 responseBody, _ := io.ReadAll(resp.Body)
657 resp.Body.Close()
658 if !useCodexCLIRouting || !unsupportedPromptCacheRetention(status, responseBody) {
659 label := c.providerName
660 if label == "" {
661 label = ProviderOpenAICodex
662 }
663 return nil, fmt.Errorf("%s: http %d: %s", label, status, strings.TrimSpace(string(responseBody)))
664 }
665 // The Codex CLI identity selects cached-prompt retention the model
666 // does not support. Retry once with the native identity on the same
667 // session/request correlation and body; anything else is returned.
668 resp, err = doStreamWithRetry(ctx, c.http, newReq(false), req.Lifecycle)
669 if err != nil {
670 label := c.providerName
671 if label == "" {
672 label = ProviderOpenAICodex
673 }
674 return nil, fmt.Errorf("%s: cache compatibility retry: %w", label, err)
675 }
676 if resp.StatusCode != http.StatusOK {
677 status = resp.StatusCode
678 responseBody, _ = io.ReadAll(resp.Body)
679 resp.Body.Close()
680 label := c.providerName
681 if label == "" {
682 label = ProviderOpenAICodex
683 }
684 return nil, fmt.Errorf("%s: http %d: %s", label, status, strings.TrimSpace(string(responseBody)))
685 }
686 }
687
688 out := make(chan Event, 16)
689 go c.runStream(ctx, resp, req, out)
690 return out, nil
691}
692
693func (c *codexClient) responsesHeaders(req Request, useCodexCLIRouting, websocketMode bool) http.Header {
694 headers := make(http.Header)
695 headers.Set("authorization", "Bearer "+c.token)
696 if !websocketMode || c.accountID != "" {
697 headers.Set("openai-beta", "responses=experimental")
698 }
699 if c.accountID != "" {
700 headers.Set("chatgpt-account-id", c.accountID)
701 }
702
703 cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
704 threadID := strings.TrimSpace(req.Context.ThreadID)
705 if useCodexCLIRouting {
706 // The ChatGPT Codex backend only admits (and reliably serves)
707 // requests that follow Codex CLI routing metadata; other client
708 // identities are load-shed even when capacity is available.
709 headers.Set("originator", "codex_cli_rs")
710 if c.capabilities.SessionHeader && cacheSessionID != "" {
711 headers.Set("session-id", cacheSessionID)
712 }
713 if c.capabilities.ThreadHeader && threadID != "" {
714 headers.Set("thread-id", threadID)
715 }
716 if c.capabilities.ClientRequestHeader && threadID != "" {
717 headers.Set("x-client-request-id", threadID)
718 }
719 headers.Set("user-agent", "codex_cli_rs/0.0.0")
720 setOpenCodeGoHeaders(headers, c.providerName, req.Context)
721 return headers
722 }
723
724 // ChatGPT's backend recognizes these established values. They are
725 // external provider metadata, not zut product identifiers.
726 if c.accountID != "" {
727 headers.Set("originator", "zot")
728 headers.Set("user-agent", fmt.Sprintf("zot (%s %s)", runtime.GOOS, runtime.GOARCH))
729 }
730 setOpenCodeGoHeaders(headers, c.providerName, req.Context)
731 return headers
732}
733
734func (c *codexClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
735 defer close(out)
736 defer resp.Body.Close()
737
738 raw := make(chan sseEvent, 16)
739 go readSSE(resp.Body, raw)
740 c.runResponseEvents(ctx, req, out, raw, nil)
741}
742
743// runResponseEvents translates the Responses streaming event protocol into
744// provider events. Both HTTP/SSE and Responses WebSocket mode use the same
745// event vocabulary, so keeping the state machine here prevents the transports
746// from drifting in tool-call, reasoning, and usage handling.
747func (c *codexClient) runResponseEvents(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, completed func(string)) {
748 var observe func(string, Message)
749 if completed != nil {
750 observe = func(responseID string, _ Message) { completed(responseID) }
751 }
752 c.runResponseEventsWithFirst(ctx, req, out, raw, nil, observe)
753}
754
755func (c *codexClient) runResponseEventsWithFirst(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, first *sseEvent, completed func(string, Message)) {
756 model, _ := c.findModel(req.Model)
757 providerName := c.providerName
758 if providerName == "" {
759 providerName = ProviderOpenAICodex
760 }
761 out <- EventStart{Model: req.Model, Provider: providerName}
762
763 // Accumulators. The Responses API emits output_items in order; each
764 // item is either a "message" (text) or a "function_call". We track
765 // the in-flight item by its index.
766 type itemState struct {
767 kind string // "message" | "function_call" | "reasoning"
768 callID string
769 name string
770 argsBuf strings.Builder
771 textBuf strings.Builder
772 summary strings.Builder
773 rawID string
774 encrypted string
775 // phase is the Responses output phase observed for a message item.
776 // It is fill-only: an absent later field must never erase an
777 // observed value. Any observed value (including unrecognized ones)
778 // is retained in storage; only recognized values are replayed.
779 phase string
780 announced bool
781 }
782 var (
783 items = map[int]*itemState{}
784 order []int
785 usage Usage
786 stop = StopEnd
787 finalErr error
788 )
789
790 assemble := func() Message {
791 content := []Content{}
792 for _, idx := range order {
793 it := items[idx]
794 switch it.kind {
795 case "message":
796 if it.textBuf.Len() > 0 {
797 content = append(content, TextBlock{Text: it.textBuf.String(), Phase: it.phase})
798 }
799 case "function_call":
800 args := it.argsBuf.String()
801 if args == "" || !json.Valid([]byte(args)) {
802 args = "{}"
803 }
804 content = append(content, ToolCallBlock{
805 ID: it.callID, Name: it.name, Arguments: json.RawMessage(args),
806 })
807 case "reasoning":
808 if it.encrypted == "" && it.summary.Len() == 0 && it.rawID == "" {
809 continue
810 }
811 content = append(content, ReasoningBlock{
812 ID: it.rawID,
813 Summary: it.summary.String(),
814 Encrypted: it.encrypted,
815 })
816 }
817 }
818 return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
819 }
820
821 sendDone := func() {
822 usage.CostUSD = ComputeCost(model, usage)
823 out <- EventUsage{Usage: usage}
824 out <- EventDone{Stop: stop, Err: finalErr, Message: assemble()}
825 }
826
827 for {
828 var ev sseEvent
829 if first != nil {
830 ev = *first
831 first = nil
832 } else {
833 select {
834 case <-ctx.Done():
835 stop = StopAborted
836 finalErr = ctx.Err()
837 sendDone()
838 return
839 case next, ok := <-raw:
840 if !ok {
841 sendDone()
842 return
843 }
844 ev = next
845 }
846 }
847 if ev.Err != nil {
848 stop = StopError
849 finalErr = fmt.Errorf("read SSE: %w", ev.Err)
850 sendDone()
851 return
852 }
853 var head struct {
854 Type string `json:"type"`
855 }
856 if err := json.Unmarshal([]byte(ev.Data), &head); err != nil {
857 continue
858 }
859
860 switch head.Type {
861 case "response.output_item.added":
862 var p struct {
863 OutputIndex int `json:"output_index"`
864 Item struct {
865 Type string `json:"type"` // "message" | "function_call" | "reasoning"
866 ID string `json:"id"`
867 CallID string `json:"call_id"`
868 Name string `json:"name"`
869 EncryptedContent string `json:"encrypted_content"`
870 Phase string `json:"phase"`
871 Content []struct {
872 Type string `json:"type"`
873 Text string `json:"text"`
874 Phase string `json:"phase"`
875 } `json:"content"`
876 } `json:"item"`
877 }
878 _ = json.Unmarshal([]byte(ev.Data), &p)
879 it := &itemState{}
880 switch p.Item.Type {
881 case "message":
882 it.kind = "message"
883 it.phase = messageItemPhase(p.Item.Phase, p.Item.Content)
884 case "function_call":
885 it.kind = "function_call"
886 it.callID = p.Item.CallID
887 it.name = p.Item.Name
888 if !it.announced {
889 it.announced = true
890 out <- EventToolStart{ID: it.callID, Name: it.name}
891 }
892 case "reasoning":
893 it.kind = "reasoning"
894 it.rawID = p.Item.ID
895 it.encrypted = p.Item.EncryptedContent
896 default:
897 continue
898 }
899 items[p.OutputIndex] = it
900 order = append(order, p.OutputIndex)
901 case "response.output_text.delta":
902 var p struct {
903 OutputIndex int `json:"output_index"`
904 Delta string `json:"delta"`
905 }
906 _ = json.Unmarshal([]byte(ev.Data), &p)
907 if it, ok := items[p.OutputIndex]; ok && it.kind == "message" {
908 it.textBuf.WriteString(p.Delta)
909 out <- EventTextDelta{Delta: p.Delta}
910 }
911 case "response.reasoning_summary_text.delta":
912 var p struct {
913 OutputIndex int `json:"output_index"`
914 Delta string `json:"delta"`
915 }
916 _ = json.Unmarshal([]byte(ev.Data), &p)
917 if it, ok := items[p.OutputIndex]; ok && it.kind == "reasoning" {
918 it.summary.WriteString(p.Delta)
919 }
920 case "response.reasoning_summary_text.done":
921 // summary text already accumulated via deltas
922 case "response.function_call_arguments.delta":
923 var p struct {
924 OutputIndex int `json:"output_index"`
925 Delta string `json:"delta"`
926 }
927 _ = json.Unmarshal([]byte(ev.Data), &p)
928 if it, ok := items[p.OutputIndex]; ok && it.kind == "function_call" {
929 it.argsBuf.WriteString(p.Delta)
930 out <- EventToolArgs{ID: it.callID, Delta: p.Delta}
931 }
932 case "response.output_item.done":
933 var p struct {
934 OutputIndex int `json:"output_index"`
935 Item struct {
936 Type string `json:"type"`
937 ID string `json:"id"`
938 EncryptedContent string `json:"encrypted_content"`
939 Phase string `json:"phase"`
940 Content []struct {
941 Type string `json:"type"`
942 Text string `json:"text"`
943 Phase string `json:"phase"`
944 } `json:"content"`
945 Summary []struct {
946 Type string `json:"type"`
947 Text string `json:"text"`
948 } `json:"summary"`
949 } `json:"item"`
950 }
951 _ = json.Unmarshal([]byte(ev.Data), &p)
952 if it, ok := items[p.OutputIndex]; ok {
953 switch it.kind {
954 case "function_call":
955 out <- EventToolEnd{ID: it.callID}
956 case "message":
957 // Fill-only: an absent phase on item-done must not
958 // erase a phase already observed on item-added.
959 if it.phase == "" {
960 it.phase = messageItemPhase(p.Item.Phase, p.Item.Content)
961 }
962 case "reasoning":
963 // Fill-only: earlier streamed payloads and summaries win
964 // over later duplicates from output_item.done.
965 if it.encrypted == "" && p.Item.EncryptedContent != "" {
966 it.encrypted = p.Item.EncryptedContent
967 }
968 if it.rawID == "" && p.Item.ID != "" {
969 it.rawID = p.Item.ID
970 }
971 if it.summary.Len() == 0 {
972 for _, s := range p.Item.Summary {
973 if s.Text == "" {
974 continue
975 }
976 if it.summary.Len() > 0 {
977 it.summary.WriteString("\n")
978 }
979 it.summary.WriteString(s.Text)
980 }
981 }
982 }
983 }
984 case "response.completed", "response.done":
985 var p struct {
986 Response struct {
987 ID string `json:"id"`
988 Usage struct {
989 InputTokens int `json:"input_tokens"`
990 OutputTokens int `json:"output_tokens"`
991 InputTokensDetails *openAIInputTokensDetails `json:"input_tokens_details"`
992 OutputTokensDetails *struct {
993 ReasoningTokens int `json:"reasoning_tokens"`
994 } `json:"output_tokens_details"`
995 } `json:"usage"`
996 Status string `json:"status"`
997 EndTurn *bool `json:"end_turn"`
998 Output []struct {
999 Type string `json:"type"`
1000 ID string `json:"id"`
1001 EncryptedContent string `json:"encrypted_content"`
1002 Phase string `json:"phase"`
1003 Content []struct {
1004 Type string `json:"type"`
1005 Text string `json:"text"`
1006 Phase string `json:"phase"`
1007 } `json:"content"`
1008 Summary []struct {
1009 Type string `json:"type"`
1010 Text string `json:"text"`
1011 } `json:"summary"`
1012 } `json:"output"`
1013 } `json:"response"`
1014 }
1015 _ = json.Unmarshal([]byte(ev.Data), &p)
1016 // Backfill encrypted reasoning supplied only in the terminal
1017 // response output. Match by nonempty provider-issued item ID,
1018 // never by output-array position. Fill a missing payload from
1019 // a matching terminal item without overwriting an earlier
1020 // payload or duplicating streamed summaries. Message phase is
1021 // reconciled the same fill-only way, but by output-array
1022 // position: terminal text content without a phase must not
1023 // erase a phase observed during streaming.
1024 if len(p.Response.Output) > 0 {
1025 terminal := make(map[string]struct {
1026 encrypted string
1027 summary string
1028 }, len(p.Response.Output))
1029 terminalPhases := make([]string, 0, len(p.Response.Output))
1030 for _, out := range p.Response.Output {
1031 if out.Type == "message" {
1032 terminalPhases = append(terminalPhases, messageItemPhase(out.Phase, out.Content))
1033 }
1034 if out.Type != "reasoning" || out.ID == "" {
1035 continue
1036 }
1037 if _, seen := terminal[out.ID]; seen {
1038 continue
1039 }
1040 var summary strings.Builder
1041 for _, s := range out.Summary {
1042 if s.Text == "" {
1043 continue
1044 }
1045 if summary.Len() > 0 {
1046 summary.WriteString("\n")
1047 }
1048 summary.WriteString(s.Text)
1049 }
1050 terminal[out.ID] = struct {
1051 encrypted string
1052 summary string
1053 }{encrypted: out.EncryptedContent, summary: summary.String()}
1054 }
1055 for _, it := range items {
1056 if it == nil || it.kind != "reasoning" || it.rawID == "" {
1057 continue
1058 }
1059 match, ok := terminal[it.rawID]
1060 if !ok {
1061 continue
1062 }
1063 if it.encrypted == "" && match.encrypted != "" {
1064 it.encrypted = match.encrypted
1065 }
1066 if it.summary.Len() == 0 && match.summary != "" {
1067 it.summary.WriteString(match.summary)
1068 }
1069 }
1070 // Reconcile message phases by output-array position: the
1071 // n-th terminal message corresponds to the n-th streamed
1072 // message item. Fill only; never erase streamed phases.
1073 pos := 0
1074 for _, idx := range order {
1075 it := items[idx]
1076 if it == nil || it.kind != "message" {
1077 continue
1078 }
1079 if pos >= len(terminalPhases) {
1080 break
1081 }
1082 if it.phase == "" {
1083 it.phase = terminalPhases[pos]
1084 }
1085 pos++
1086 }
1087 }
1088 usage = normalizeOpenAIUsage(p.Response.Usage.InputTokens, p.Response.Usage.OutputTokens, p.Response.Usage.InputTokensDetails)
1089 if details := p.Response.Usage.OutputTokensDetails; details != nil {
1090 usage.ReasoningTokens = details.ReasoningTokens
1091 usage.ReasoningTokensKnown = true
1092 }
1093
1094 hadTool := false
1095 for _, it := range items {
1096 if it.kind == "function_call" {
1097 hadTool = true
1098 break
1099 }
1100 }
1101 if hadTool {
1102 stop = StopToolUse
1103 } else {
1104 stop = StopEnd
1105 if p.Response.EndTurn != nil && !*p.Response.EndTurn {
1106 stop = StopContinue
1107 }
1108 }
1109 if completed != nil && p.Response.ID != "" {
1110 completed(p.Response.ID, assemble())
1111 }
1112 sendDone()
1113 return
1114 case "response.failed":
1115 var p struct {
1116 Response struct {
1117 Error struct {
1118 Message string `json:"message"`
1119 } `json:"error"`
1120 } `json:"response"`
1121 }
1122 _ = json.Unmarshal([]byte(ev.Data), &p)
1123 stop = StopError
1124 finalErr = fmt.Errorf("codex: %s", p.Response.Error.Message)
1125 sendDone()
1126 return
1127 case "error":
1128 var p struct {
1129 Message string `json:"message"`
1130 Code string `json:"code"`
1131 Error struct {
1132 Message string `json:"message"`
1133 Code string `json:"code"`
1134 } `json:"error"`
1135 }
1136 _ = json.Unmarshal([]byte(ev.Data), &p)
1137 msg := p.Message
1138 if msg == "" {
1139 msg = p.Error.Message
1140 }
1141 if msg == "" {
1142 msg = p.Code
1143 }
1144 if msg == "" {
1145 msg = p.Error.Code
1146 }
1147 if msg == "" {
1148 msg = strings.TrimSpace(ev.Data)
1149 }
1150 stop = StopError
1151 label := c.errorLabel
1152 if label == "" {
1153 label = "codex"
1154 }
1155 finalErr = fmt.Errorf("%s error: %s", label, msg)
1156 sendDone()
1157 return
1158 }
1159 }
1160}