package executor import ( "bytes" "context" "fmt" "io" "net/http" "strings" "time" codexauth "github.com/router-for-me/CLIProxyAPI/v6/internal/auth/codex" "github.com/router-for-me/CLIProxyAPI/v6/internal/config" "github.com/router-for-me/CLIProxyAPI/v6/internal/thinking" "github.com/router-for-me/CLIProxyAPI/v6/internal/util" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/auth" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/executor" sdktranslator "github.com/router-for-me/CLIProxyAPI/v6/sdk/translator" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" "github.com/tidwall/sjson" "github.com/tiktoken-go/tokenizer" ) // CodexExecutorRefactored is a stateless executor for Codex (OpenAI Responses API). type CodexExecutorRefactored struct { cfg *config.Config base *BaseExecutor } // NewCodexExecutorRefactored creates a new Codex executor instance. func NewCodexExecutorRefactored(cfg *config.Config) *CodexExecutorRefactored { provider := NewCodexProvider(cfg) return &CodexExecutorRefactored{ cfg: cfg, base: NewBaseExecutor(cfg, provider), } } // Identifier returns the executor identifier. func (e *CodexExecutorRefactored) Identifier() string { return "codex" } // PrepareRequest injects Codex credentials into the outgoing HTTP request. func (e *CodexExecutorRefactored) 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 *CodexExecutorRefactored) 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 := newProxyAwareHTTPClient(ctx, e.cfg, auth, 0) return httpClient.Do(httpReq) } // Execute performs a non-streaming request to the Codex API. func (e *CodexExecutorRefactored) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if cfg := parseCodexAgentConfig(req.Payload); cfg.Enabled() && opts.Alt == "responses/compact" { return cliproxyexecutor.Response{}, statusErr{code: http.StatusBadRequest, msg: "codex agent_mode is not supported for /responses/compact"} } if opts.Alt == "responses/compact" { return e.executeCompact(ctx, auth, req, opts) } // Codex always uses streaming internally, then converts to non-streaming return e.executeViaStream(ctx, auth, req, opts) } // executeViaStream executes a non-streaming request by consuming a stream func (e *CodexExecutorRefactored) executeViaStream(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 = codexDefaultURL } reporter := 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 := bytes.Clone(originalPayloadSource) originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, bytes.Clone(req.Payload), false) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) if err != nil { return resp, err } requestedModel := payloadRequestedModel(opts, req.Model) body = applyPayloadConfigWithRoot(e.cfg, baseModel, to.String(), "", body, originalTranslated, requestedModel) agentCfg := resolveCodexAgentConfigForNonStream(body, req.Payload) body, err = normalizeCodexRequestBody(body, baseModel, true) if err != nil { return resp, err } var pass codexCompletedPassResult switch agentCfg.Mode { case codexAgentModePlannerReviewer: pass, err = e.executeCodexPlannerReviewerPipeline(ctx, auth, apiKey, baseURL, from, req, body) default: url := strings.TrimSuffix(baseURL, "/") + "/responses" pass, err = e.executeCodexCompletedHTTPPass(ctx, auth, apiKey, url, from, req, body, true) } if err != nil { return resp, err } if pass.UsageOK { reporter.publish(ctx, pass.Usage) } reqBodyForTranslation := body if len(pass.Request) > 0 { reqBodyForTranslation = pass.Request } var param any out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, originalPayload, reqBodyForTranslation, pass.Completed, ¶m) resp = cliproxyexecutor.Response{Payload: []byte(out), Headers: pass.Headers} return resp, nil } // executeCompact handles the /responses/compact endpoint func (e *CodexExecutorRefactored) 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 = codexDefaultURL } reporter := 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 := bytes.Clone(originalPayloadSource) originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, bytes.Clone(req.Payload), false) body, err = thinking.ApplyThinking(body, req.Model, from.String(), to.String(), e.Identifier()) if err != nil { return resp, err } requestedModel := payloadRequestedModel(opts, req.Model) body = applyPayloadConfigWithRoot(e.cfg, baseModel, to.String(), "", body, originalTranslated, requestedModel) body, err = normalizeCodexRequestBody(body, baseModel, false) if err != nil { return resp, err } body, _ = sjson.DeleteBytes(body, "stream") url := strings.TrimSuffix(baseURL, "/") + "/responses/compact" httpReq, err := e.cacheHelper(ctx, from, url, req, body) if err != nil { return resp, err } provider := NewCodexProvider(e.cfg) provider.ApplyHeaders(httpReq, auth, apiKey, false) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } recordAPIRequest(ctx, e.cfg, upstreamRequestLog{ URL: url, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := newProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { recordAPIResponseError(ctx, e.cfg, err) return resp, err } defer httpResp.Body.Close() recordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) appendAPIResponseChunk(ctx, e.cfg, b) logWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, summarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) return resp, statusErr{code: httpResp.StatusCode, msg: string(b)} } data, err := io.ReadAll(httpResp.Body) if err != nil { recordAPIResponseError(ctx, e.cfg, err) return resp, err } appendAPIResponseChunk(ctx, e.cfg, data) reporter.publish(ctx, parseOpenAIUsage(data)) reporter.ensurePublished(ctx) var param any out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, originalPayload, body, data, ¶m) resp = cliproxyexecutor.Response{Payload: []byte(out), Headers: httpResp.Header.Clone()} return resp, nil } // ExecuteStream performs a streaming request to the Codex API. func (e *CodexExecutorRefactored) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (*cliproxyexecutor.StreamResult, error) { if opts.Alt == "responses/compact" { return nil, statusErr{code: http.StatusBadRequest, msg: "streaming not supported for /responses/compact"} } if cfg := parseCodexAgentConfig(req.Payload); cfg.Enabled() { return nil, statusErr{code: http.StatusBadRequest, msg: "codex agent_mode is not supported for streaming requests"} } return e.base.ExecuteStream(ctx, auth, req, opts) } // CountTokens estimates token count for Codex requests. func (e *CodexExecutorRefactored) CountTokens(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (cliproxyexecutor.Response, error) { baseModel := thinking.ParseSuffix(req.Model).ModelName from := opts.SourceFormat to := sdktranslator.FromString("codex") 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 cliproxyexecutor.Response{}, err } body, err = normalizeCodexRequestBody(body, baseModel, false) if err != nil { return cliproxyexecutor.Response{}, err } enc, err := tokenizerForCodexModel(baseModel) if err != nil { return cliproxyexecutor.Response{}, fmt.Errorf("codex executor: tokenizer init failed: %w", err) } count, err := countCodexInputTokens(enc, body) if err != nil { return cliproxyexecutor.Response{}, fmt.Errorf("codex executor: token counting failed: %w", err) } usageJSON := fmt.Sprintf(`{"response":{"usage":{"input_tokens":%d,"output_tokens":0,"total_tokens":%d}}}`, count, count) translated := sdktranslator.TranslateTokenCount(ctx, to, from, count, []byte(usageJSON)) return cliproxyexecutor.Response{Payload: []byte(translated)}, nil } // Refresh refreshes the Codex authentication tokens. func (e *CodexExecutorRefactored) Refresh(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) { log.Debugf("codex executor: refresh called") if auth == nil { return nil, statusErr{code: 500, msg: "codex executor: auth is nil"} } var refreshToken string if auth.Metadata != nil { if v, ok := auth.Metadata["refresh_token"].(string); ok && v != "" { refreshToken = v } } if refreshToken == "" { return auth, nil } svc := codexauth.NewCodexAuth(e.cfg) td, err := svc.RefreshTokensWithRetry(ctx, refreshToken, 3) if err != nil { return nil, err } if auth.Metadata == nil { auth.Metadata = make(map[string]any) } auth.Metadata["id_token"] = td.IDToken auth.Metadata["access_token"] = td.AccessToken if td.RefreshToken != "" { auth.Metadata["refresh_token"] = td.RefreshToken } if td.AccountID != "" { auth.Metadata["account_id"] = td.AccountID } auth.Metadata["email"] = td.Email auth.Metadata["expired"] = td.Expire auth.Metadata["type"] = "codex" now := time.Now().Format(time.RFC3339) auth.Metadata["last_refresh"] = now return auth, nil } // cacheHelper applies cache-related logic to the request. func (e *CodexExecutorRefactored) cacheHelper(ctx context.Context, from sdktranslator.Format, url string, req cliproxyexecutor.Request, rawJSON []byte) (*http.Request, error) { rawJSON, _, cacheHeaders := applyCodexPromptCache(e.cfg, from, req.Payload, req.Model, rawJSON, false) httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(rawJSON)) if err != nil { return nil, err } for k, values := range cacheHeaders { for i := range values { httpReq.Header.Add(k, values[i]) } } return httpReq, nil } func tokenizerForCodexModel(model string) (tokenizer.Codec, error) { sanitized := strings.ToLower(strings.TrimSpace(model)) switch { case sanitized == "": return tokenizer.Get(tokenizer.Cl100kBase) case strings.HasPrefix(sanitized, "gpt-5"): return tokenizer.ForModel(tokenizer.GPT5) case strings.HasPrefix(sanitized, "gpt-4.1"): return tokenizer.ForModel(tokenizer.GPT41) case strings.HasPrefix(sanitized, "gpt-4o"): return tokenizer.ForModel(tokenizer.GPT4o) case strings.HasPrefix(sanitized, "gpt-4"): return tokenizer.ForModel(tokenizer.GPT4) case strings.HasPrefix(sanitized, "gpt-3.5"), strings.HasPrefix(sanitized, "gpt-3"): return tokenizer.ForModel(tokenizer.GPT35Turbo) default: return tokenizer.Get(tokenizer.Cl100kBase) } } func countCodexInputTokens(enc tokenizer.Codec, body []byte) (int64, error) { if enc == nil { return 0, fmt.Errorf("encoder is nil") } if len(body) == 0 { return 0, nil } root := gjson.ParseBytes(body) var segments []string if inst := strings.TrimSpace(root.Get("instructions").String()); inst != "" { segments = append(segments, inst) } inputItems := root.Get("input") if inputItems.IsArray() { arr := inputItems.Array() for i := range arr { item := arr[i] switch item.Get("type").String() { case "message": content := item.Get("content") if content.IsArray() { parts := content.Array() for j := range parts { part := parts[j] if text := strings.TrimSpace(part.Get("text").String()); text != "" { segments = append(segments, text) } } } case "function_call": if name := strings.TrimSpace(item.Get("name").String()); name != "" { segments = append(segments, name) } if args := strings.TrimSpace(item.Get("arguments").String()); args != "" { segments = append(segments, args) } case "function_call_output": if out := strings.TrimSpace(item.Get("output").String()); out != "" { segments = append(segments, out) } default: if text := strings.TrimSpace(item.Get("text").String()); text != "" { segments = append(segments, text) } } } } tools := root.Get("tools") if tools.IsArray() { tarr := tools.Array() for i := range tarr { tool := tarr[i] if name := strings.TrimSpace(tool.Get("name").String()); name != "" { segments = append(segments, name) } if desc := strings.TrimSpace(tool.Get("description").String()); desc != "" { segments = append(segments, desc) } if params := tool.Get("parameters"); params.Exists() { val := params.Raw if params.Type == gjson.String { val = params.String() } if trimmed := strings.TrimSpace(val); trimmed != "" { segments = append(segments, trimmed) } } } } textFormat := root.Get("text.format") if textFormat.Exists() { if name := strings.TrimSpace(textFormat.Get("name").String()); name != "" { segments = append(segments, name) } if schema := textFormat.Get("schema"); schema.Exists() { val := schema.Raw if schema.Type == gjson.String { val = schema.String() } if trimmed := strings.TrimSpace(val); trimmed != "" { segments = append(segments, trimmed) } } } text := strings.Join(segments, "\n") if text == "" { return 0, nil } count, err := enc.Count(text) if err != nil { return 0, err } return int64(count), nil } func codexCreds(a *cliproxyauth.Auth) (apiKey, baseURL string) { if a == nil { return "", "" } if a.Attributes != nil { apiKey = a.Attributes["api_key"] baseURL = a.Attributes["base_url"] } if apiKey == "" && a.Metadata != nil { if v, ok := a.Metadata["access_token"].(string); ok { apiKey = v } } return }