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