Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
26 kB · 660 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661package executor
import ( "bytes" "context" "fmt" "net/http" "strings"
"github.com/gorilla/websocket" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" "github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps" "github.com/router-for-me/CLIProxyAPI/v7/internal/thinking" 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")
func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { log.Debugf("Executing Codex Websockets stream request with auth ID: %s, model: %s", auth.ID, req.Model) if ctx == nil { ctx = context.Background() } if opts.Alt == "responses/compact" { return nil, statusErr{code: http.StatusBadRequest, msg: "streaming not supported for /responses/compact"} }
baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := codexCreds(auth) if baseURL == "" { baseURL = "https://chatgpt.com/backend-api/codex" }
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) to := sdktranslator.FromString("codex") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated, body := translateCodexRequestPair(from, to, baseModel, originalPayload, req.Payload, true)
body, err = helps.ApplyRequestThinking(body, req, opts, from.String(), to.String(), e.Identifier()) if err != nil { return nil, 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 = helps.SetStringIfDifferent(body, "model", baseModel) body = normalizeCodexInstructions(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex websockets executor", body) body = normalizeCodexWebsocketParallelToolCalls(body, opts.Headers) multiAgentV2Conflict := helps.HasCodexMultiAgentV2NamespaceConflict(body) body, optimizeMultiAgentV2 := helps.OptimizeCodexMultiAgentV2RequestForAuth(ctx, opts.Headers, body, e.cfg, auth, baseModel) body, replayScope, errReplay := applyCodexReasoningReplayCacheRequired(ctx, from, req, opts, body) if errReplay != nil { return nil, errReplay }
httpURL := strings.TrimSuffix(baseURL, "/") + "/responses" wsURL, err := buildCodexResponsesWebsocketURL(httpURL) if err != nil { return nil, err }
body, wsHeaders, errPromptCache := applyCodexPromptCacheHeadersWithContext(ctx, from, req, body, opts.Headers) if errPromptCache != nil { return nil, errPromptCache } clientBody := body var identityState codexIdentityConfuseState upstreamBody, identityState := applyCodexIdentityConfuseBody(e.cfg, auth, originalPayloadSource, body) reporter.SetTranslatedReasoningEffort(clientBody, to.String()) wsHeaders = applyCodexWebsocketHeaders(ctx, wsHeaders, auth, apiKey, e.cfg, opts.Headers) applyModelHeaderOverrides(wsHeaders, baseModel) applyCodexIdentityConfuseHeaders(wsHeaders, &identityState)
var authID, authLabel, authType, authValue string authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo()
executionSessionID := executionSessionIDFromOptions(opts) var sess *codexWebsocketSession if executionSessionID != "" { sess = e.getOrCreateSession(executionSessionID) if sess != nil { sess.reqMu.Lock() } } streamSessionLocked := sess != nil unlockStreamSession := func() { if sess != nil && streamSessionLocked { sess.reqMu.Unlock() streamSessionLocked = false } }
wsReqBody := buildCodexWebsocketRequestBody(upstreamBody) wsReqLog := helps.UpstreamRequestLog{ URL: wsURL, Method: "WEBSOCKET", Headers: wsHeaders.Clone(), Body: wsReqBody, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, } helps.RecordAPIWebsocketRequest(ctx, e.cfg, wsReqLog)
var conn *websocket.Conn var closer *websocketConnectionCloser var respHS *http.Response var errDial error dialCtx := ctx if cliproxyexecutor.RequiredUpstreamWebsocket(ctx) { conn, closer = existingWebsocketSessionConn(sess, authID, wsURL) if conn == nil { if sess != nil { sess.reqMu.Unlock() } return nil, cliproxyexecutor.NewUpstreamWebsocketReplayRequiredError() } } else { dialCtx = cliproxyexecutor.WithUpstreamAttemptTracker(ctx) conn, closer, respHS, errDial = e.ensureUpstreamConn(dialCtx, auth, sess, authID, wsURL, wsHeaders) } var upstreamHeaders http.Header if respHS != nil { upstreamHeaders = respHS.Header.Clone() } if errDial != nil { bodyErr := websocketHandshakeBody(respHS) if respHS != nil { helps.RecordAPIWebsocketUpgradeRejection(ctx, e.cfg, websocketUpgradeRequestLog(wsReqLog), respHS.StatusCode, respHS.Header.Clone(), bodyErr) } if respHS != nil && respHS.StatusCode == http.StatusUpgradeRequired { if sess != nil { sess.reqMu.Unlock() } if opts.ExecutionLifecycle == nil && !cliproxyexecutor.DownstreamWebsocket(ctx) { return e.CodexExecutor.ExecuteStream(ctx, auth, req, opts) } if cliproxyexecutor.UpstreamAttempted(dialCtx) { cliproxyexecutor.MarkUpstreamAttempt(ctx) } return nil, statusErr{code: respHS.StatusCode, msg: string(bodyErr)} } if cliproxyexecutor.UpstreamAttempted(dialCtx) { cliproxyexecutor.MarkUpstreamAttempt(ctx) } if respHS != nil && respHS.StatusCode > 0 { if sess != nil { sess.reqMu.Unlock() } return nil, newCodexStatusErr(respHS.StatusCode, bodyErr) } helps.RecordAPIWebsocketError(ctx, e.cfg, "dial", errDial) if sess != nil { sess.reqMu.Unlock() } return nil, errDial } if errBind := sess.bindExecutionLifecycle(opts, conn, closer, req.Model); errBind != nil { if sess != nil { sess.reqMu.Unlock() } closeWebsocketAfterBindFailure(sess, conn, closer) return nil, errBind } recordAPIWebsocketHandshake(ctx, e.cfg, respHS) reporter.StartResponseTTFT()
if sess == nil { logCodexWebsocketConnected(executionSessionID, authID, wsURL) }
var readCh chan codexWebsocketRead if sess != nil { readCh = sess.activate(conn) } restoreMultiAgentV2 := !multiAgentV2Conflict && (optimizeMultiAgentV2 || sess.isMultiAgentV2Optimized(conn))
cliproxyexecutor.MarkUpstreamAttempt(ctx) if errSend := writeCodexWebsocketMessage(sess, conn, wsReqBody); errSend != nil { errSend = mapCodexWebsocketWriteError(sess, conn, errSend) helps.RecordAPIWebsocketError(ctx, e.cfg, "send", errSend) if sess != nil { if cliproxyexecutor.RequiredUpstreamWebsocket(ctx) { e.invalidateUpstreamConnWithoutDisconnectNotify(sess, conn, "send_error", errSend) sess.clearActive(conn, readCh) sess.reqMu.Unlock() if !shouldRetryCodexWebsocketSend(errSend) { return nil, errSend } return nil, cliproxyexecutor.NewUpstreamWebsocketReplayRequiredError() } e.invalidateUpstreamConn(sess, conn, "send_error", errSend) if !shouldRetryCodexWebsocketSend(errSend) { sess.clearActive(conn, readCh) sess.reqMu.Unlock() return nil, errSend }
// Retry once with a new websocket connection for the same execution session. connRetry, closerRetry, respHSRetry, errDialRetry := e.ensureUpstreamConn(ctx, auth, sess, authID, wsURL, wsHeaders) if errDialRetry != nil || connRetry == nil { closeHTTPResponseBody(respHSRetry, "codex websockets executor: close handshake response body error") helps.RecordAPIWebsocketError(ctx, e.cfg, "dial_retry", errDialRetry) sess.clearActive(conn, readCh) sess.reqMu.Unlock() return nil, errDialRetry } previousConn, previousReadCh := conn, readCh conn = connRetry closer = closerRetry if errBind := sess.bindExecutionLifecycle(opts, conn, closer, req.Model); errBind != nil { clearRetryActiveState(sess, previousConn, previousReadCh) sess.reqMu.Unlock() closeWebsocketAfterBindFailure(sess, conn, closer) return nil, errBind } readCh = sess.activate(conn) restoreMultiAgentV2 = !multiAgentV2Conflict && (optimizeMultiAgentV2 || sess.isMultiAgentV2Optimized(conn)) wsReqBodyRetry := buildCodexWebsocketRequestBody(upstreamBody) helps.RecordAPIWebsocketRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: wsURL, Method: "WEBSOCKET", Headers: wsHeaders.Clone(), Body: wsReqBodyRetry, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) recordAPIWebsocketHandshake(ctx, e.cfg, respHSRetry) reporter.StartResponseTTFT() cliproxyexecutor.MarkUpstreamAttempt(ctx) if errSendRetry := writeCodexWebsocketMessage(sess, conn, wsReqBodyRetry); errSendRetry != nil { errSendRetry = mapCodexWebsocketWriteError(sess, conn, errSendRetry) helps.RecordAPIWebsocketError(ctx, e.cfg, "send_retry", errSendRetry) e.invalidateUpstreamConn(sess, conn, "send_error", errSendRetry) sess.clearActive(conn, readCh) sess.reqMu.Unlock() return nil, errSendRetry } wsReqBody = wsReqBodyRetry } else { logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, "send_error", errSend) if errClose := closer.Close(); errClose != nil { log.Errorf("codex websockets executor: close websocket error: %v", errClose) } return nil, errSend } }
if optimizeMultiAgentV2 || multiAgentV2Conflict { sess.setMultiAgentV2Optimized(conn, optimizeMultiAgentV2 && !multiAgentV2Conflict) }
buffering := e.cfg != nil && e.cfg.Codex.StreamBootstrapBuffering
claudeInputTokens := helps.NewClaudeInputTokenState(from, to, responseFormat, originalPayload) var param any outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte
var bufferedChunks [][]byte var initialChunks [][]byte immediateTerminal := false // bootstrapTerminalErr holds a non-overload terminal failure seen while buffering. It is // delivered as an in-stream chunk after the buffered handshake so downstream behaviour stays // identical to the unbuffered path instead of silently turning into a credential failover. var bootstrapTerminalErr error
if buffering { for { if ctx != nil && ctx.Err() != nil { if sess != nil { sess.clearActive(conn, readCh) unlockStreamSession() } else { _ = closer.Close() } return nil, ctx.Err() } msgType, payload, errRead := readCodexWebsocketMessage(ctx, sess, conn, readCh) if errRead != nil { mappedErr := mapCodexWebsocketReadError(errRead) if sess != nil { e.invalidateUpstreamConn(sess, conn, "read_error", mappedErr) sess.clearActive(conn, readCh) unlockStreamSession() } else { logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, "read_error", mappedErr) _ = closer.Close() } helps.RecordAPIWebsocketError(ctx, e.cfg, "read", mappedErr) reporter.PublishFailure(ctx, mappedErr) return nil, mappedErr } if msgType != websocket.TextMessage { if msgType == websocket.BinaryMessage { errBinary := fmt.Errorf("codex websockets executor: unexpected binary message") if sess != nil { e.invalidateUpstreamConn(sess, conn, "unexpected_binary", errBinary) sess.clearActive(conn, readCh) unlockStreamSession() } else { logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, "unexpected_binary", errBinary) _ = closer.Close() } helps.RecordAPIWebsocketError(ctx, e.cfg, "unexpected_binary", errBinary) reporter.PublishFailure(ctx, errBinary) return nil, errBinary } continue }
payload = bytes.TrimSpace(payload) if len(payload) == 0 { continue } observeCodexTokenEvent(reporter, payload) payload = applyCodexIdentityConfuseResponsePayload(payload, identityState) payload = restoreCodexMultiAgentToolNamespace(payload, originalPayload, clientBody) helps.AppendCodexAPIWebsocketResponse(ctx, e.cfg, payload) helps.EmitWebSocketResponseEvent(ctx, opts, auth, e.Identifier(), req.Model, payload) payload = helps.RestoreCodexMultiAgentV2Response(payload, restoreMultiAgentV2)
if wsErr, ok := parseCodexWebsocketError(payload); ok { if sess != nil { e.invalidateUpstreamConn(sess, conn, "upstream_error", wsErr) sess.clearActive(conn, readCh) unlockStreamSession() } else { logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, "upstream_error", wsErr) _ = closer.Close() } if errClearReplay := clearCodexReasoningReplayOnWebsocketError(ctx, replayScope, payload); errClearReplay != nil { helps.RecordAPIWebsocketError(ctx, e.cfg, "replay_clear_error", errClearReplay) reporter.PublishFailure(ctx, errClearReplay) return nil, errClearReplay } helps.RecordAPIWebsocketError(ctx, e.cfg, "upstream_error", wsErr) reporter.PublishFailure(ctx, wsErr) return nil, wsErr } if streamErr, terminalBody, ok := codexTerminalFailureErr(payload); ok { // A transient capacity rejection is retried on another credential, so the // downstream websocket session must survive this upstream teardown. Notifying // the disconnect here would close the client connection before the retry can // deliver anything. Every other terminal failure is forwarded in-stream and // legitimately terminates the session, so it keeps the notifying variant. failoverPending := isCodexOverloadBootstrapFailure(terminalBody) if sess != nil { unlockStreamSession() if failoverPending { e.invalidateUpstreamConnWithoutDisconnectNotify(sess, conn, "terminal_failure", streamErr) } else { e.invalidateUpstreamConn(sess, conn, "terminal_failure", streamErr) } sess.clearActive(conn, readCh) } else { logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, "terminal_failure", streamErr) _ = closer.Close() } if errClearReplay := clearCodexReasoningReplayOnInvalidSignature(ctx, replayScope, streamErr.StatusCode(), terminalBody); errClearReplay != nil { helps.RecordAPIWebsocketError(ctx, e.cfg, "replay_clear_error", errClearReplay) reporter.PublishFailure(ctx, errClearReplay) return nil, errClearReplay } helps.RecordAPIWebsocketError(ctx, e.cfg, "upstream_error", streamErr) reporter.PublishFailure(ctx, streamErr) if failoverPending { // Fail the attempt before the downstream headers are committed so the // conductor can transparently retry on another credential, and report the // status the upstream refused to put on the wire. helps.LogWithRequestID(ctx).Debugf("codex websockets executor: bootstrap overload rejection after %d buffered handshake events, failing over", len(bufferedChunks)) return nil, newCodexBootstrapOverloadErr(terminalBody) } bootstrapTerminalErr = streamErr break }
eventType := gjson.GetBytes(payload, "type").String() isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" || eventType == "response.failed" || eventType == "error" if eventType == "response.output_item.done" { collectCodexOutputItemDone(payload, outputItemsByIndex, &outputItemsFallback) } completedPayload := payload if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { completedPayload = normalizeCodexWebsocketCompletion(completedPayload) completedPayload = patchCodexCompletedOutput(completedPayload, outputItemsByIndex, outputItemsFallback) if eventType != "response.incomplete" { cacheCodexReasoningReplayFromCompleted(replayScope, completedPayload) } if detail, ok := helps.ParseCodexUsage(completedPayload); ok { reporter.Publish(ctx, detail) } else { reporter.EnsurePublished(ctx) } }
var currentChunks [][]byte if cliproxyexecutor.DownstreamWebsocket(ctx) { if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { payload = completedPayload } clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) downstreamPayload := helps.EnsureResponsesUsageDetails(clientPayload) currentChunks = [][]byte{downstreamPayload} } else { payload = normalizeCodexWebsocketCompletion(payload) if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { payload = completedPayload } clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) line := encodeCodexWebsocketAsSSE(clientPayload) currentChunks = helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, originalPayload, clientBody, line, ¶m, claudeInputTokens) }
if isCodexHandshakeMetadataEvent(eventType) && !isTerminalEvent { if len(bufferedChunks) < codexBootstrapMaxBufferedEvents { bufferedChunks = append(bufferedChunks, currentChunks...) continue } helps.LogWithRequestID(ctx).Debugf("codex websockets executor: bootstrap buffer limit %d reached, releasing stream without overload probing", codexBootstrapMaxBufferedEvents) }
initialChunks = currentChunks if isTerminalEvent { immediateTerminal = true } break } }
chanCapacity := len(bufferedChunks) + len(initialChunks) if bootstrapTerminalErr != nil { chanCapacity++ } out := make(chan cliproxyexecutor.StreamChunk, chanCapacity) for _, chunk := range bufferedChunks { out <- cliproxyexecutor.StreamChunk{Payload: chunk} } for _, chunk := range initialChunks { out <- cliproxyexecutor.StreamChunk{Payload: chunk} } if bootstrapTerminalErr != nil { // The upstream connection was already invalidated and released in the terminal-failure // branch above, so only the buffered payloads plus the in-stream error remain to emit. out <- cliproxyexecutor.StreamChunk{Err: bootstrapTerminalErr} close(out) return &cliproxyexecutor.StreamResult{Headers: upstreamHeaders, Chunks: out}, nil } if immediateTerminal { if sess != nil { sess.clearActive(conn, readCh) unlockStreamSession() } else { logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, "completed", nil) if errClose := closer.Close(); errClose != nil { log.Errorf("codex websockets executor: close websocket error: %v", errClose) } } close(out) return &cliproxyexecutor.StreamResult{Headers: upstreamHeaders, Chunks: out}, nil }
go func() { terminateReason := "completed" var terminateErr error
defer close(out) defer func() { if sess != nil { sess.clearActive(conn, readCh) unlockStreamSession() return } logCodexWebsocketDisconnected(executionSessionID, authID, wsURL, terminateReason, terminateErr) if errClose := closer.Close(); errClose != nil { log.Errorf("codex websockets executor: close websocket error: %v", errClose) } }()
send := func(chunk cliproxyexecutor.StreamChunk) bool { if ctx == nil { out <- chunk return true } select { case out <- chunk: return true case <-ctx.Done(): return false } }
for { if ctx != nil && ctx.Err() != nil { terminateReason = "context_done" terminateErr = ctx.Err() _ = send(cliproxyexecutor.StreamChunk{Err: ctx.Err()}) return } msgType, payload, errRead := readCodexWebsocketMessage(ctx, sess, conn, readCh) if errRead != nil { if sess != nil && ctx != nil && ctx.Err() != nil { terminateReason = "context_done" terminateErr = ctx.Err() _ = send(cliproxyexecutor.StreamChunk{Err: ctx.Err()}) return } mappedErr := mapCodexWebsocketReadError(errRead) terminateReason = "read_error" terminateErr = mappedErr helps.RecordAPIWebsocketError(ctx, e.cfg, "read", mappedErr) reporter.PublishFailure(ctx, mappedErr) _ = send(cliproxyexecutor.StreamChunk{Err: mappedErr}) return } if msgType != websocket.TextMessage { if msgType == websocket.BinaryMessage { unexpectedErr := fmt.Errorf("codex websockets executor: unexpected binary message") terminateReason = "unexpected_binary" terminateErr = unexpectedErr helps.RecordAPIWebsocketError(ctx, e.cfg, "unexpected_binary", unexpectedErr) reporter.PublishFailure(ctx, unexpectedErr) if sess != nil { e.invalidateUpstreamConn(sess, conn, "unexpected_binary", unexpectedErr) } _ = send(cliproxyexecutor.StreamChunk{Err: unexpectedErr}) return } continue }
payload = bytes.TrimSpace(payload) if len(payload) == 0 { continue } observeCodexTokenEvent(reporter, payload) payload = applyCodexIdentityConfuseResponsePayload(payload, identityState) helps.AppendCodexAPIWebsocketResponse(ctx, e.cfg, payload) helps.EmitWebSocketResponseEvent(ctx, opts, auth, e.Identifier(), req.Model, payload) payload = helps.RestoreCodexMultiAgentV2Response(payload, restoreMultiAgentV2)
if wsErr, ok := parseCodexWebsocketError(payload); ok { terminateReason = "upstream_error" terminateErr = wsErr if sess != nil { e.invalidateUpstreamConn(sess, conn, "upstream_error", wsErr) } if errClearReplay := clearCodexReasoningReplayOnWebsocketError(ctx, replayScope, payload); errClearReplay != nil { terminateErr = errClearReplay helps.RecordAPIWebsocketError(ctx, e.cfg, "replay_clear_error", errClearReplay) reporter.PublishFailure(ctx, errClearReplay) _ = send(cliproxyexecutor.StreamChunk{Err: errClearReplay}) return } helps.RecordAPIWebsocketError(ctx, e.cfg, "upstream_error", wsErr) reporter.PublishFailure(ctx, wsErr) _ = send(cliproxyexecutor.StreamChunk{Err: wsErr}) return } if streamErr, terminalBody, ok := codexTerminalFailureErr(payload); ok { terminateReason = "upstream_error" terminateErr = streamErr if sess != nil { unlockStreamSession() e.invalidateUpstreamConn(sess, conn, "terminal_failure", streamErr) } if errClearReplay := clearCodexReasoningReplayOnInvalidSignature(ctx, replayScope, streamErr.StatusCode(), terminalBody); errClearReplay != nil { terminateErr = errClearReplay helps.RecordAPIWebsocketError(ctx, e.cfg, "replay_clear_error", errClearReplay) reporter.PublishFailure(ctx, errClearReplay) _ = send(cliproxyexecutor.StreamChunk{Err: errClearReplay}) return } helps.RecordAPIWebsocketError(ctx, e.cfg, "upstream_error", streamErr) reporter.PublishFailure(ctx, streamErr) _ = send(cliproxyexecutor.StreamChunk{Err: streamErr}) return }
eventType := gjson.GetBytes(payload, "type").String() isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" || eventType == "response.failed" || eventType == "error" if eventType == "response.output_item.done" { collectCodexOutputItemDone(payload, outputItemsByIndex, &outputItemsFallback) } completedPayload := payload if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { completedPayload = normalizeCodexWebsocketCompletion(completedPayload) completedPayload = patchCodexCompletedOutput(completedPayload, outputItemsByIndex, outputItemsFallback) if eventType != "response.incomplete" { cacheCodexReasoningReplayFromCompleted(replayScope, completedPayload) } if detail, ok := helps.ParseCodexUsage(completedPayload); ok { reporter.Publish(ctx, detail) } else { reporter.EnsurePublished(ctx) } }
clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) if cliproxyexecutor.DownstreamWebsocket(ctx) { if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { clientPayload = applyCodexIdentityExposeResponsePayload(completedPayload, identityState) } downstreamPayload := helps.EnsureResponsesUsageDetails(clientPayload) if !send(cliproxyexecutor.StreamChunk{Payload: downstreamPayload}) { terminateReason = "context_done" terminateErr = ctx.Err() return } if isTerminalEvent { return } continue }
payload = normalizeCodexWebsocketCompletion(payload) if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { payload = completedPayload } eventType = gjson.GetBytes(payload, "type").String() clientPayload = applyCodexIdentityExposeResponsePayload(payload, identityState) line := encodeCodexWebsocketAsSSE(clientPayload) chunks := helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, originalPayload, clientBody, line, ¶m, claudeInputTokens) for i := range chunks { if !send(cliproxyexecutor.StreamChunk{Payload: chunks[i]}) { terminateReason = "context_done" terminateErr = ctx.Err() return } } if isTerminalEvent || eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { return } } }()
return &cliproxyexecutor.StreamResult{Headers: upstreamHeaders, Chunks: out}, nil}