package executor import ( "bufio" "bytes" "context" "fmt" "io" "net/http" "sort" "strings" "time" codexauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/codex" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" "github.com/router-for-me/CLIProxyAPI/v7/internal/misc" "github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps" "github.com/router-for-me/CLIProxyAPI/v7/internal/thinking" "github.com/router-for-me/CLIProxyAPI/v7/internal/util" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" "github.com/tidwall/sjson" "github.com/tiktoken-go/tokenizer" "github.com/gin-gonic/gin" "github.com/google/uuid") const ( codexUserAgent = "codex_cli_rs/0.118.0 (Mac OS 26.3.1; arm64) iTerm.app/3.6.9" codexOriginator = "codex_cli_rs" codexDefaultImageToolModel = "gpt-image-2") var dataTag = []byte("data:") // Streamed Codex responses may emit response.output_item.done events while leaving// response.completed.response.output empty. Keep the stream path aligned with the// already-patched non-stream path by reconstructing response.output from those items.func collectCodexOutputItemDone(eventData []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback *[][]byte) { itemResult := gjson.GetBytes(eventData, "item") if !itemResult.Exists() || itemResult.Type != gjson.JSON { return } outputIndexResult := gjson.GetBytes(eventData, "output_index") if outputIndexResult.Exists() { outputItemsByIndex[outputIndexResult.Int()] = []byte(itemResult.Raw) return } *outputItemsFallback = append(*outputItemsFallback, []byte(itemResult.Raw))} func patchCodexCompletedOutput(eventData []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback [][]byte) []byte { outputResult := gjson.GetBytes(eventData, "response.output") shouldPatchOutput := (!outputResult.Exists() || !outputResult.IsArray() || len(outputResult.Array()) == 0) && (len(outputItemsByIndex) > 0 || len(outputItemsFallback) > 0) if !shouldPatchOutput { return eventData } indexes := make([]int64, 0, len(outputItemsByIndex)) for idx := range outputItemsByIndex { indexes = append(indexes, idx) } sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] }) items := make([][]byte, 0, len(outputItemsByIndex)+len(outputItemsFallback)) for _, idx := range indexes { items = append(items, outputItemsByIndex[idx]) } items = append(items, outputItemsFallback...) outputArray := []byte("[]") if len(items) > 0 { var buf bytes.Buffer totalLen := 2 for _, item := range items { totalLen += len(item) } if len(items) > 1 { totalLen += len(items) - 1 } buf.Grow(totalLen) buf.WriteByte('[') for i, item := range items { if i > 0 { buf.WriteByte(',') } buf.Write(item) } buf.WriteByte(']') outputArray = buf.Bytes() } completedDataPatched, _ := sjson.SetRawBytes(eventData, "response.output", outputArray) return completedDataPatched} func codexTerminalStreamContextLengthErr(eventData []byte) (statusErr, bool) { eventType := gjson.GetBytes(eventData, "type").String() var body []byte switch eventType { case "error": body = codexTerminalErrorBody(eventData, "error") if len(body) == 0 { body = codexTerminalTopLevelErrorBody(eventData) } case "response.failed": body = codexTerminalErrorBody(eventData, "response.error") if len(body) == 0 { body = codexTerminalErrorBody(eventData, "error") } default: return statusErr{}, false } if len(body) == 0 { return statusErr{}, false } if !codexTerminalErrorIsContextLength(body) { return statusErr{}, false } return newCodexStatusErr(http.StatusBadRequest, body), true} func codexTerminalErrorBody(eventData []byte, path string) []byte { errorResult := gjson.GetBytes(eventData, path) if !errorResult.Exists() { return nil } body := []byte(`{"error":{}}`) if errorResult.Type == gjson.JSON { body, _ = sjson.SetRawBytes(body, "error", []byte(errorResult.Raw)) } else if message := strings.TrimSpace(errorResult.String()); message != "" { body, _ = sjson.SetBytes(body, "error.message", message) } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if message := strings.TrimSpace(gjson.GetBytes(eventData, "response.error.message").String()); message != "" { body, _ = sjson.SetBytes(body, "error.message", message) } } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if code := strings.TrimSpace(gjson.GetBytes(body, "error.code").String()); code != "" { body, _ = sjson.SetBytes(body, "error.message", code) } } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if errorType := strings.TrimSpace(gjson.GetBytes(body, "error.type").String()); errorType != "" { body, _ = sjson.SetBytes(body, "error.message", errorType) } } return body} func codexTerminalTopLevelErrorBody(eventData []byte) []byte { message := strings.TrimSpace(gjson.GetBytes(eventData, "message").String()) code := strings.TrimSpace(gjson.GetBytes(eventData, "code").String()) errorType := strings.TrimSpace(gjson.GetBytes(eventData, "error_type").String()) param := strings.TrimSpace(gjson.GetBytes(eventData, "param").String()) if message == "" && code == "" && errorType == "" && param == "" { return nil } body := []byte(`{"error":{}}`) if message != "" { body, _ = sjson.SetBytes(body, "error.message", message) } if code != "" { body, _ = sjson.SetBytes(body, "error.code", code) } if errorType != "" { body, _ = sjson.SetBytes(body, "error.type", errorType) } if param != "" { body, _ = sjson.SetBytes(body, "error.param", param) } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if code != "" { body, _ = sjson.SetBytes(body, "error.message", code) } else if errorType != "" { body, _ = sjson.SetBytes(body, "error.message", errorType) } } return body} func codexTerminalErrorIsContextLength(body []byte) bool { errorCode := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.code").String())) message := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.message").String())) return errorCode == "context_length_exceeded" || errorCode == "context_too_large" || strings.Contains(message, "context window") || strings.Contains(message, "context length") || strings.Contains(message, "too many tokens")} // CodexExecutor is a stateless executor for Codex (OpenAI Responses API entrypoint).// If api_key is unavailable on auth, it falls back to legacy via ClientAdapter.type CodexExecutor struct { cfg *config.Config} func NewCodexExecutor(cfg *config.Config) *CodexExecutor { return &CodexExecutor{cfg: cfg} } func (e *CodexExecutor) Identifier() string { return "codex" } // PrepareRequest injects Codex credentials into the outgoing HTTP request.func (e *CodexExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if req == nil { return nil } apiKey, _ := codexCreds(auth) if strings.TrimSpace(apiKey) != "" { req.Header.Set("Authorization", "Bearer "+apiKey) } var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(req, attrs) return nil} // HttpRequest injects Codex credentials into the request and executes it.func (e *CodexExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) { if req == nil { return nil, fmt.Errorf("codex executor: request is nil") } if ctx == nil { ctx = req.Context() } httpReq := req.WithContext(ctx) if err := e.PrepareRequest(httpReq, auth); err != nil { return nil, err } httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) return httpClient.Do(httpReq)} func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return e.executeCompact(ctx, auth, req, opts) } if isCodexOpenAIImageRequest(opts) { return e.executeOpenAIImage(ctx, auth, req, opts) } baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := codexCreds(auth) if baseURL == "" { baseURL = "https://chatgpt.com/backend-api/codex" } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("codex") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) if err != nil { return resp, err } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, _ = sjson.SetBytes(body, "model", baseModel) body, _ = sjson.SetBytes(body, "stream", true) body, _ = sjson.DeleteBytes(body, "previous_response_id") body, _ = sjson.DeleteBytes(body, "prompt_cache_retention") body, _ = sjson.DeleteBytes(body, "safety_identifier") body, _ = sjson.DeleteBytes(body, "stream_options") body = normalizeCodexInstructions(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth) } url := strings.TrimSuffix(baseURL, "/") + "/responses" httpReq, err := e.cacheHelper(ctx, from, url, req, body) if err != nil { return resp, err } applyCodexHeaders(httpReq, auth, apiKey, true, e.cfg) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: url, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("codex executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) err = newCodexStatusErr(httpResp.StatusCode, b) return resp, err } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) lines := bytes.Split(data, []byte("\n")) outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte for _, line := range lines { if !bytes.HasPrefix(line, dataTag) { continue } eventData := bytes.TrimSpace(line[5:]) eventType := gjson.GetBytes(eventData, "type").String() if streamErr, ok := codexTerminalStreamContextLengthErr(eventData); ok { err = streamErr return resp, err } if eventType == "response.output_item.done" { itemResult := gjson.GetBytes(eventData, "item") if !itemResult.Exists() || itemResult.Type != gjson.JSON { continue } outputIndexResult := gjson.GetBytes(eventData, "output_index") if outputIndexResult.Exists() { outputItemsByIndex[outputIndexResult.Int()] = []byte(itemResult.Raw) } else { outputItemsFallback = append(outputItemsFallback, []byte(itemResult.Raw)) } continue } if eventType != "response.completed" { continue } if detail, ok := helps.ParseCodexUsage(eventData); ok { reporter.Publish(ctx, detail) } publishCodexImageToolUsage(ctx, reporter, body, eventData) completedData := eventData outputResult := gjson.GetBytes(completedData, "response.output") shouldPatchOutput := (!outputResult.Exists() || !outputResult.IsArray() || len(outputResult.Array()) == 0) && (len(outputItemsByIndex) > 0 || len(outputItemsFallback) > 0) if shouldPatchOutput { completedDataPatched := completedData completedDataPatched, _ = sjson.SetRawBytes(completedDataPatched, "response.output", []byte(`[]`)) indexes := make([]int64, 0, len(outputItemsByIndex)) for idx := range outputItemsByIndex { indexes = append(indexes, idx) } sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] }) for _, idx := range indexes { completedDataPatched, _ = sjson.SetRawBytes(completedDataPatched, "response.output.-1", outputItemsByIndex[idx]) } for _, item := range outputItemsFallback { completedDataPatched, _ = sjson.SetRawBytes(completedDataPatched, "response.output.-1", item) } completedData = completedDataPatched } var param any out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, originalPayload, body, completedData, ¶m) resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil } err = statusErr{code: 408, msg: "stream error: stream disconnected before completion: stream closed before response.completed"} return resp, err} func (e *CodexExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := codexCreds(auth) if baseURL == "" { baseURL = "https://chatgpt.com/backend-api/codex" } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("openai-response") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) if err != nil { return resp, err } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, _ = sjson.SetBytes(body, "model", baseModel) body, _ = sjson.DeleteBytes(body, "stream") body = normalizeCodexInstructions(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth) } url := strings.TrimSuffix(baseURL, "/") + "/responses/compact" httpReq, err := e.cacheHelper(ctx, from, url, req, body) if err != nil { return resp, err } applyCodexHeaders(httpReq, auth, apiKey, false, e.cfg) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: url, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("codex executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) err = newCodexStatusErr(httpResp.StatusCode, b) return resp, err } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) reporter.EnsurePublished(ctx) var param any out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, originalPayload, body, data, ¶m) resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil} Lines 1–500 of 1169. Load more
package executor import ( "bufio" "bytes" "context" "fmt" "io" "net/http" "sort" "strings" "time" codexauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/codex" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" "github.com/router-for-me/CLIProxyAPI/v7/internal/misc" "github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps" "github.com/router-for-me/CLIProxyAPI/v7/internal/thinking" "github.com/router-for-me/CLIProxyAPI/v7/internal/util" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" "github.com/tidwall/sjson" "github.com/tiktoken-go/tokenizer" "github.com/gin-gonic/gin" "github.com/google/uuid" ) const ( codexUserAgent = "codex_cli_rs/0.118.0 (Mac OS 26.3.1; arm64) iTerm.app/3.6.9" codexOriginator = "codex_cli_rs" codexDefaultImageToolModel = "gpt-image-2" ) var dataTag = []byte("data:") // Streamed Codex responses may emit response.output_item.done events while leaving // response.completed.response.output empty. Keep the stream path aligned with the // already-patched non-stream path by reconstructing response.output from those items. func collectCodexOutputItemDone(eventData []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback *[][]byte) { itemResult := gjson.GetBytes(eventData, "item") if !itemResult.Exists() || itemResult.Type != gjson.JSON { return } outputIndexResult := gjson.GetBytes(eventData, "output_index") if outputIndexResult.Exists() { outputItemsByIndex[outputIndexResult.Int()] = []byte(itemResult.Raw) return } *outputItemsFallback = append(*outputItemsFallback, []byte(itemResult.Raw)) } func patchCodexCompletedOutput(eventData []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback [][]byte) []byte { outputResult := gjson.GetBytes(eventData, "response.output") shouldPatchOutput := (!outputResult.Exists() || !outputResult.IsArray() || len(outputResult.Array()) == 0) && (len(outputItemsByIndex) > 0 || len(outputItemsFallback) > 0) if !shouldPatchOutput { return eventData } indexes := make([]int64, 0, len(outputItemsByIndex)) for idx := range outputItemsByIndex { indexes = append(indexes, idx) } sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] }) items := make([][]byte, 0, len(outputItemsByIndex)+len(outputItemsFallback)) for _, idx := range indexes { items = append(items, outputItemsByIndex[idx]) } items = append(items, outputItemsFallback...) outputArray := []byte("[]") if len(items) > 0 { var buf bytes.Buffer totalLen := 2 for _, item := range items { totalLen += len(item) } if len(items) > 1 { totalLen += len(items) - 1 } buf.Grow(totalLen) buf.WriteByte('[') for i, item := range items { if i > 0 { buf.WriteByte(',') } buf.Write(item) } buf.WriteByte(']') outputArray = buf.Bytes() } completedDataPatched, _ := sjson.SetRawBytes(eventData, "response.output", outputArray) return completedDataPatched } func codexTerminalStreamContextLengthErr(eventData []byte) (statusErr, bool) { eventType := gjson.GetBytes(eventData, "type").String() var body []byte switch eventType { case "error": body = codexTerminalErrorBody(eventData, "error") if len(body) == 0 { body = codexTerminalTopLevelErrorBody(eventData) } case "response.failed": body = codexTerminalErrorBody(eventData, "response.error") if len(body) == 0 { body = codexTerminalErrorBody(eventData, "error") } default: return statusErr{}, false } if len(body) == 0 { return statusErr{}, false } if !codexTerminalErrorIsContextLength(body) { return statusErr{}, false } return newCodexStatusErr(http.StatusBadRequest, body), true } func codexTerminalErrorBody(eventData []byte, path string) []byte { errorResult := gjson.GetBytes(eventData, path) if !errorResult.Exists() { return nil } body := []byte(`{"error":{}}`) if errorResult.Type == gjson.JSON { body, _ = sjson.SetRawBytes(body, "error", []byte(errorResult.Raw)) } else if message := strings.TrimSpace(errorResult.String()); message != "" { body, _ = sjson.SetBytes(body, "error.message", message) } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if message := strings.TrimSpace(gjson.GetBytes(eventData, "response.error.message").String()); message != "" { body, _ = sjson.SetBytes(body, "error.message", message) } } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if code := strings.TrimSpace(gjson.GetBytes(body, "error.code").String()); code != "" { body, _ = sjson.SetBytes(body, "error.message", code) } } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if errorType := strings.TrimSpace(gjson.GetBytes(body, "error.type").String()); errorType != "" { body, _ = sjson.SetBytes(body, "error.message", errorType) } } return body } func codexTerminalTopLevelErrorBody(eventData []byte) []byte { message := strings.TrimSpace(gjson.GetBytes(eventData, "message").String()) code := strings.TrimSpace(gjson.GetBytes(eventData, "code").String()) errorType := strings.TrimSpace(gjson.GetBytes(eventData, "error_type").String()) param := strings.TrimSpace(gjson.GetBytes(eventData, "param").String()) if message == "" && code == "" && errorType == "" && param == "" { return nil } body := []byte(`{"error":{}}`) if message != "" { body, _ = sjson.SetBytes(body, "error.message", message) } if code != "" { body, _ = sjson.SetBytes(body, "error.code", code) } if errorType != "" { body, _ = sjson.SetBytes(body, "error.type", errorType) } if param != "" { body, _ = sjson.SetBytes(body, "error.param", param) } if strings.TrimSpace(gjson.GetBytes(body, "error.message").String()) == "" { if code != "" { body, _ = sjson.SetBytes(body, "error.message", code) } else if errorType != "" { body, _ = sjson.SetBytes(body, "error.message", errorType) } } return body } func codexTerminalErrorIsContextLength(body []byte) bool { errorCode := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.code").String())) message := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.message").String())) return errorCode == "context_length_exceeded" || errorCode == "context_too_large" || strings.Contains(message, "context window") || strings.Contains(message, "context length") || strings.Contains(message, "too many tokens") } // CodexExecutor is a stateless executor for Codex (OpenAI Responses API entrypoint). // If api_key is unavailable on auth, it falls back to legacy via ClientAdapter. type CodexExecutor struct { cfg *config.Config } func NewCodexExecutor(cfg *config.Config) *CodexExecutor { return &CodexExecutor{cfg: cfg} } func (e *CodexExecutor) Identifier() string { return "codex" } // PrepareRequest injects Codex credentials into the outgoing HTTP request. func (e *CodexExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if req == nil { return nil } apiKey, _ := codexCreds(auth) if strings.TrimSpace(apiKey) != "" { req.Header.Set("Authorization", "Bearer "+apiKey) } var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(req, attrs) return nil } // HttpRequest injects Codex credentials into the request and executes it. func (e *CodexExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) { if req == nil { return nil, fmt.Errorf("codex executor: request is nil") } if ctx == nil { ctx = req.Context() } httpReq := req.WithContext(ctx) if err := e.PrepareRequest(httpReq, auth); err != nil { return nil, err } httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) return httpClient.Do(httpReq) } func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return e.executeCompact(ctx, auth, req, opts) } if isCodexOpenAIImageRequest(opts) { return e.executeOpenAIImage(ctx, auth, req, opts) } baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := codexCreds(auth) if baseURL == "" { baseURL = "https://chatgpt.com/backend-api/codex" } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("codex") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) if err != nil { return resp, err } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, _ = sjson.SetBytes(body, "model", baseModel) body, _ = sjson.SetBytes(body, "stream", true) body, _ = sjson.DeleteBytes(body, "previous_response_id") body, _ = sjson.DeleteBytes(body, "prompt_cache_retention") body, _ = sjson.DeleteBytes(body, "safety_identifier") body, _ = sjson.DeleteBytes(body, "stream_options") body = normalizeCodexInstructions(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth) } url := strings.TrimSuffix(baseURL, "/") + "/responses" httpReq, err := e.cacheHelper(ctx, from, url, req, body) if err != nil { return resp, err } applyCodexHeaders(httpReq, auth, apiKey, true, e.cfg) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: url, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("codex executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) err = newCodexStatusErr(httpResp.StatusCode, b) return resp, err } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) lines := bytes.Split(data, []byte("\n")) outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte for _, line := range lines { if !bytes.HasPrefix(line, dataTag) { continue } eventData := bytes.TrimSpace(line[5:]) eventType := gjson.GetBytes(eventData, "type").String() if streamErr, ok := codexTerminalStreamContextLengthErr(eventData); ok { err = streamErr return resp, err } if eventType == "response.output_item.done" { itemResult := gjson.GetBytes(eventData, "item") if !itemResult.Exists() || itemResult.Type != gjson.JSON { continue } outputIndexResult := gjson.GetBytes(eventData, "output_index") if outputIndexResult.Exists() { outputItemsByIndex[outputIndexResult.Int()] = []byte(itemResult.Raw) } else { outputItemsFallback = append(outputItemsFallback, []byte(itemResult.Raw)) } continue } if eventType != "response.completed" { continue } if detail, ok := helps.ParseCodexUsage(eventData); ok { reporter.Publish(ctx, detail) } publishCodexImageToolUsage(ctx, reporter, body, eventData) completedData := eventData outputResult := gjson.GetBytes(completedData, "response.output") shouldPatchOutput := (!outputResult.Exists() || !outputResult.IsArray() || len(outputResult.Array()) == 0) && (len(outputItemsByIndex) > 0 || len(outputItemsFallback) > 0) if shouldPatchOutput { completedDataPatched := completedData completedDataPatched, _ = sjson.SetRawBytes(completedDataPatched, "response.output", []byte(`[]`)) indexes := make([]int64, 0, len(outputItemsByIndex)) for idx := range outputItemsByIndex { indexes = append(indexes, idx) } sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] }) for _, idx := range indexes { completedDataPatched, _ = sjson.SetRawBytes(completedDataPatched, "response.output.-1", outputItemsByIndex[idx]) } for _, item := range outputItemsFallback { completedDataPatched, _ = sjson.SetRawBytes(completedDataPatched, "response.output.-1", item) } completedData = completedDataPatched } var param any out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, originalPayload, body, completedData, ¶m) resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil } err = statusErr{code: 408, msg: "stream error: stream disconnected before completion: stream closed before response.completed"} return resp, err } func (e *CodexExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := codexCreds(auth) if baseURL == "" { baseURL = "https://chatgpt.com/backend-api/codex" } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("openai-response") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) if err != nil { return resp, err } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, _ = sjson.SetBytes(body, "model", baseModel) body, _ = sjson.DeleteBytes(body, "stream") body = normalizeCodexInstructions(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth) } url := strings.TrimSuffix(baseURL, "/") + "/responses/compact" httpReq, err := e.cacheHelper(ctx, from, url, req, body) if err != nil { return resp, err } applyCodexHeaders(httpReq, auth, apiKey, false, e.cfg) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: url, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("codex executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) err = newCodexStatusErr(httpResp.StatusCode, b) return resp, err } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) reporter.EnsurePublished(ctx) var param any out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, originalPayload, body, data, ¶m) resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil }