package executor import (	"bufio"	"bytes"	"context"	"crypto/sha256"	"encoding/hex"	"fmt"	"io"	"net/http"	"sort"	"strings"	"time" 	codexauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/codex"	internalcache "github.com/router-for-me/CLIProxyAPI/v7/internal/cache"	"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/registry"	"github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps"	"github.com/router-for-me/CLIProxyAPI/v7/internal/signature"	"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-tui/0.135.0 (Mac OS 26.5.0; arm64) iTerm.app/3.6.10 (codex-tui; 0.135.0)"	codexOriginator            = "codex-tui"	codexDefaultImageToolModel = "gpt-image-2"	codexResponsesLiteHeader   = "X-OpenAI-Internal-Codex-Responses-Lite"	codexResponsesLiteMetadata = "client_metadata.ws_request_header_x_openai_internal_codex_responses_lite") var dataTag = []byte("data:") const codexIncompleteStreamMessage = "stream error: stream disconnected before completion: stream closed before response.completed" type codexIncompleteStreamError struct {	statusErr} func newCodexIncompleteStreamError() codexIncompleteStreamError {	return codexIncompleteStreamError{statusErr: statusErr{		code: http.StatusRequestTimeout,		msg:  codexIncompleteStreamMessage,	}}} func (codexIncompleteStreamError) IsRequestScoped() bool {	return true} // 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) {	streamErr, body, ok := codexTerminalStreamErr(eventData)	if !ok || !codexTerminalErrorIsContextLength(body) {		return statusErr{}, false	}	return streamErr, true} func codexTerminalStreamErr(eventData []byte) (statusErr, []byte, bool) {	body, ok := codexTerminalFailureBody(eventData)	if !ok || !codexTerminalStreamErrShouldHandle(body) {		return statusErr{}, nil, false	}	return newCodexStatusErr(http.StatusBadRequest, body), body, true} func codexTerminalFailureErr(eventData []byte) (statusErr, []byte, bool) {	if streamErr, body, ok := codexTerminalStreamErr(eventData); ok {		return streamErr, body, true	}	body, ok := codexTerminalFailureBody(eventData)	if !ok {		return statusErr{}, nil, false	}	return newCodexStatusErr(codexTerminalFailureStatus(body), body), body, true} func codexTerminalFailureStatus(body []byte) int {	for _, path := range []string{"error.status_code", "error.status"} {		if status := int(gjson.GetBytes(body, path).Int()); status >= 400 && status <= 599 {			return status		}	} 	errorType := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.type").String()))	errorCode := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, "error.code").String()))	switch {	case errorType == "invalid_request_error", errorType == "bad_request_error":		return http.StatusBadRequest	case errorType == "authentication_error", errorCode == "invalid_api_key", errorCode == "unauthorized":		return http.StatusUnauthorized	case errorType == "permission_error", errorCode == "forbidden", errorCode == "permission_denied":		return http.StatusForbidden	case errorType == "not_found_error", errorCode == "not_found", errorCode == "model_not_found":		return http.StatusNotFound	case errorType == "rate_limit_error", errorCode == "rate_limit_exceeded":		return http.StatusTooManyRequests	default:		return http.StatusBadGateway	}} func codexTerminalFailureBody(eventData []byte) ([]byte, 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 nil, false	}	if len(body) == 0 {		body = []byte(`{"error":{"message":"upstream stream failed without error details"}}`)	}	return body, true} func codexTerminalStreamErrShouldHandle(body []byte) bool {	if codexTerminalErrorIsContextLength(body) {		return true	}	if isCodexUsageLimitError(body) || isCodexModelCapacityError(body) {		return true	}	code, _, ok := codexStatusErrorClassification(http.StatusBadRequest, body)	return ok && code == "thinking_signature_invalid"} 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" } func translateCodexRequestPair(from, to sdktranslator.Format, model string, originalPayload, payload []byte, stream bool) ([]byte, []byte) {	if bytes.Equal(originalPayload, payload) {		body := sdktranslator.TranslateRequest(from, to, model, payload, stream)		return body, body	}	originalTranslated := sdktranslator.TranslateRequest(from, to, model, originalPayload, stream)	body := sdktranslator.TranslateRequest(from, to, model, payload, stream)	return originalTranslated, body} type codexReasoningReplayScope struct {	modelName          string	sessionKey         string	requestFingerprint string} func (s codexReasoningReplayScope) valid() bool {	return strings.TrimSpace(s.modelName) != "" && strings.TrimSpace(s.sessionKey) != ""} func applyCodexReasoningReplayCache(ctx context.Context, from sdktranslator.Format, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, body []byte) ([]byte, codexReasoningReplayScope) {	updated, scope, _ := applyCodexReasoningReplayCacheRequired(ctx, from, req, opts, body)	return updated, scope} func applyCodexReasoningReplayCacheRequired(ctx context.Context, from sdktranslator.Format, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, body []byte) ([]byte, codexReasoningReplayScope, error) {	scope := codexReasoningReplayScopeFromRequest(ctx, from, req, opts, body)	if !scope.valid() {		return body, scope, nil	}	items, ok, errReplay := internalcache.GetCodexReasoningReplayItemsRequired(ctx, scope.modelName, scope.sessionKey)	if errReplay != nil || !ok {		return body, scope, errReplay	}	updated, ok := insertCodexReasoningReplayTurns(body, items)	if !ok {		return body, scope, nil	}	return updated, scope, nil} func codexReasoningReplayScopeFromRequest(ctx context.Context, from sdktranslator.Format, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, body []byte) codexReasoningReplayScope {	if !codexReasoningReplayEnabledForSource(from) {		return codexReasoningReplayScope{}	}	modelName := strings.TrimSpace(gjson.GetBytes(body, "model").String())	if modelName == "" {		modelName = thinking.ParseSuffix(req.Model).ModelName	}	inputItems := gjson.GetBytes(body, "input").Array()	return codexReasoningReplayScope{		modelName:          modelName,		sessionKey:         codexReasoningReplaySessionKey(ctx, from, req, opts, body),		requestFingerprint: codexReplayInputPrefixFingerprint(inputItems, len(inputItems)),	}} func codexReasoningReplayEnabledForSource(from sdktranslator.Format) bool {	return sourceFormatEqual(from, sdktranslator.FormatClaude)} func sourceFormatEqual(from, want sdktranslator.Format) bool {	return strings.EqualFold(strings.TrimSpace(from.String()), want.String())} func codexClaudeCodeReplaySessionKey(ctx context.Context, payload []byte, headers http.Header) string {	sessionKey, _ := helps.ClaudeCodeExecutionScope(ctx, payload, headers)	return sessionKey} func codexReasoningReplaySessionKey(ctx context.Context, from sdktranslator.Format, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, body []byte) string {	if ctx == nil {		ctx = context.Background()	}	if sourceFormatEqual(from, sdktranslator.FormatClaude) {		if sessionKey := codexClaudeCodeReplaySessionKey(ctx, req.Payload, opts.Headers); sessionKey != "" {			return sessionKey		}	}	if value := metadataString(opts.Metadata, cliproxyexecutor.ExecutionSessionMetadataKey); value != "" {		return "execution:" + value	}	if value := metadataString(req.Metadata, cliproxyexecutor.ExecutionSessionMetadataKey); value != "" {		return "execution:" + value	}	if value := codexReasoningReplaySessionKeyFromPayload(body); value != "" {		return value	}	if value := codexReasoningReplaySessionKeyFromPayload(req.Payload); value != "" {		return value	}	if value := codexReasoningReplaySessionKeyFromHeaders(opts.Headers); value != "" {		return value	}	if ginCtx, ok := ctx.Value("gin").(*gin.Context); ok && ginCtx != nil && ginCtx.Request != nil {		if value := codexReasoningReplaySessionKeyFromHeaders(ginCtx.Request.Header); value != "" {			return value		}	}	if sourceFormatEqual(from, sdktranslator.FormatOpenAI) {		if apiKey := strings.TrimSpace(helps.APIKeyFromContext(ctx)); apiKey != "" {			return "prompt-cache:" + uuid.NewSHA1(uuid.NameSpaceOID, []byte("cli-proxy-api:codex:prompt-cache:"+apiKey)).String()		}	}	return ""} func metadataString(metadata map[string]any, key string) string {	if len(metadata) == 0 {		return ""	}	raw, ok := metadata[key]	if !ok || raw == nil {		return ""	}	switch v := raw.(type) {	case string:		return strings.TrimSpace(v)	case []byte:		return strings.TrimSpace(string(v))	default:		return ""	}} func codexReasoningReplaySessionKeyFromPayload(payload []byte) string {	if len(payload) == 0 {		return ""	}	if promptCacheKey := strings.TrimSpace(gjson.GetBytes(payload, "prompt_cache_key").String()); promptCacheKey != "" {		return "prompt-cache:" + promptCacheKey	}	if windowID := strings.TrimSpace(gjson.GetBytes(payload, "client_metadata.x-codex-window-id").String()); windowID != "" {		return "window:" + windowID	}	if turnMetadata := strings.TrimSpace(gjson.GetBytes(payload, "client_metadata.x-codex-turn-metadata").String()); turnMetadata != "" {		return codexReasoningReplaySessionKeyFromTurnMetadata(turnMetadata)	}	return ""} func codexReasoningReplaySessionKeyFromHeaders(headers http.Header) string {	if headers == nil {		return ""	}	if turnMetadata := strings.TrimSpace(headers.Get("X-Codex-Turn-Metadata")); turnMetadata != "" {		if key := codexReasoningReplaySessionKeyFromTurnMetadata(turnMetadata); key != "" {			return key		}	}	if windowID := strings.TrimSpace(headerValueCaseInsensitive(headers, "X-Codex-Window-Id")); windowID != "" {		return "window:" + windowID	}	for _, headerName := range []string{"Session_id", "session_id", "Session-Id"} {		if value := strings.TrimSpace(headerValueCaseInsensitive(headers, headerName)); value != "" {			return "session-id:" + value		}	}	if conversationID := strings.TrimSpace(headerValueCaseInsensitive(headers, "Conversation_id")); conversationID != "" {		return "conversation_id:" + conversationID	}	return ""} func codexReasoningReplaySessionKeyFromTurnMetadata(turnMetadata string) string {	if promptCacheKey := strings.TrimSpace(gjson.Get(turnMetadata, "prompt_cache_key").String()); promptCacheKey != "" {		return "prompt-cache:" + promptCacheKey	}	if windowID := strings.TrimSpace(gjson.Get(turnMetadata, "window_id").String()); windowID != "" {		return "window:" + windowID	}	return ""} func codexInputHasValidReasoningEncryptedContent(body []byte) bool {	input := gjson.GetBytes(body, "input")	if !input.IsArray() {		return false	}	for _, item := range input.Array() {		if strings.TrimSpace(item.Get("type").String()) != "reasoning" {			continue		}		encryptedContent := item.Get("encrypted_content")		if encryptedContent.Type != gjson.String {			continue		}		if _, err := signature.InspectGPTReasoningSignature(encryptedContent.String()); err == nil {			return true		}	}	return false} type codexReasoningReplayTurn struct {	marked               bool	assistantFingerprint string	requestFingerprint   string	callIDs              []string	items                [][]byte} func insertCodexReasoningReplayTurns(body []byte, replayItems [][]byte) ([]byte, bool) {	input := gjson.GetBytes(body, "input")	if !input.IsArray() || len(replayItems) == 0 {		return body, false	}

Lines 1–500 of 2350. Load more