From 3522e481aa7baa68f47cb712d7c0f90919b69bf0 Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Sun, 9 Aug 2026 03:47:01 +0800 Subject: [PATCH] fix(openai): emit `response.failed` stream errors for Codex requests Closes: #4854 --- .../optimize_multi_agent_v2.go | 7 +- .../openai/openai_responses_handlers.go | 23 ++++++ ...ai_responses_handlers_stream_error_test.go | 37 ++++++++++ .../handlers/openai_responses_stream_error.go | 71 +++++++++++++++++++ .../openai_responses_stream_error_test.go | 42 +++++++++++ 5 files changed, 179 insertions(+), 1 deletion(-) diff --git a/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go b/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go index fbfde0073..807748837 100644 --- a/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go +++ b/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go @@ -166,7 +166,8 @@ func headerValueCaseInsensitive(headers http.Header, name string) string { return "" } -func isCodexMultiAgentClient(userAgent string) bool { +// IsCodexClientUserAgent reports whether a request uses an official Codex client identity. +func IsCodexClientUserAgent(userAgent string) bool { userAgent = strings.TrimSpace(userAgent) return strings.HasPrefix(userAgent, "Codex Desktop/") || strings.HasPrefix(userAgent, "codex-tui/") || @@ -174,6 +175,10 @@ func isCodexMultiAgentClient(userAgent string) bool { strings.HasPrefix(userAgent, "codex_cli_rs/") } +func isCodexMultiAgentClient(userAgent string) bool { + return IsCodexClientUserAgent(userAgent) +} + func codexSpawnAgentModelsForRequest(ctx context.Context, headers http.Header, homeEnabled bool) []codexSpawnAgentModel { availableModels := registry.GetGlobalRegistry().GetAvailableModels("openai") if homeEnabled { diff --git a/sdk/api/handlers/openai/openai_responses_handlers.go b/sdk/api/handlers/openai/openai_responses_handlers.go index ba78a4470..405d47f0a 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers.go +++ b/sdk/api/handlers/openai/openai_responses_handlers.go @@ -14,6 +14,7 @@ import ( "io" "net/http" "sort" + "strings" "github.com/gin-gonic/gin" "github.com/router-for-me/CLIProxyAPI/v7/internal/client/codex/optimize-multi-agent-v2" @@ -573,6 +574,23 @@ func (h *OpenAIResponsesAPIHandler) handleStreamingResponse(c *gin.Context, rawJ } } +// isCodexResponsesClientRequest limits the alternate terminal event to official Codex clients. +func isCodexResponsesClientRequest(c *gin.Context) bool { + if c == nil || c.Request == nil { + return false + } + if multiagentv2.IsCodexClientUserAgent(c.GetHeader("User-Agent")) { + return true + } + + switch originator := strings.ToLower(strings.TrimSpace(c.GetHeader("Originator"))); originator { + case "codex desktop", "codex-tui", "codex_cli_rs": + return true + default: + return strings.HasPrefix(originator, "codex desktop/") || strings.HasPrefix(originator, "codex-tui/") || strings.HasPrefix(originator, "codex_cli_rs/") + } +} + func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flusher http.Flusher, cancel func(error), data <-chan []byte, errs <-chan *interfaces.ErrorMessage, framer *responsesSSEFramer) { if framer == nil { framer = &responsesSSEFramer{} @@ -594,6 +612,11 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flush if errMsg.Error != nil && errMsg.Error.Error() != "" { errText = errMsg.Error.Error() } + if isCodexResponsesClientRequest(c) { + chunk := handlers.BuildOpenAIResponsesStreamFailedChunk(status, errText, 0) + _, _ = fmt.Fprintf(c.Writer, "\nevent: response.failed\ndata: %s\n\n", string(chunk)) + return + } chunk := handlers.BuildOpenAIResponsesStreamErrorChunk(status, errText, 0) _, _ = fmt.Fprintf(c.Writer, "\nevent: error\ndata: %s\n\n", string(chunk)) }, diff --git a/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go b/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go index f416dcd04..038ced455 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go +++ b/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go @@ -88,3 +88,40 @@ func TestForwardResponsesStreamExposesOnlyClientErrors(t *testing.T) { }) } } + +func TestForwardResponsesStreamUsesResponseFailedForCodex(t *testing.T) { + gin.SetMode(gin.TestMode) + + base := handlers.NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, nil) + h := NewOpenAIResponsesAPIHandler(base) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + c.Request.Header.Set("User-Agent", "Codex Desktop/26.803.41515") + + flusher, ok := c.Writer.(http.Flusher) + if !ok { + t.Fatal("expected gin writer to implement http.Flusher") + } + + data := make(chan []byte) + errs := make(chan *interfaces.ErrorMessage, 1) + errs <- &interfaces.ErrorMessage{ + StatusCode: http.StatusBadRequest, + Error: errors.New(`{"error":{"type":"invalid_request","code":"cyber_policy","message":"blocked"}}`), + } + close(errs) + + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs, nil) + body := recorder.Body.String() + if !strings.Contains(body, "event: response.failed") { + t.Fatalf("missing response.failed event: %q", body) + } + if strings.Contains(body, "event: error") { + t.Fatalf("unexpected legacy error event for Codex: %q", body) + } + if !strings.Contains(body, `"type":"invalid_request"`) || !strings.Contains(body, `"code":"cyber_policy"`) { + t.Fatalf("missing nested Codex error detail: %q", body) + } +} diff --git a/sdk/api/handlers/openai_responses_stream_error.go b/sdk/api/handlers/openai_responses_stream_error.go index e7760bd09..a3c3c7e32 100644 --- a/sdk/api/handlers/openai_responses_stream_error.go +++ b/sdk/api/handlers/openai_responses_stream_error.go @@ -14,6 +14,17 @@ type openAIResponsesStreamErrorChunk struct { SequenceNumber int `json:"sequence_number"` } +type openAIResponsesStreamFailedChunk struct { + Type string `json:"type"` + SequenceNumber int `json:"sequence_number"` + Response openAIResponsesStreamFailedResponse `json:"response"` +} + +type openAIResponsesStreamFailedResponse struct { + Status string `json:"status"` + Error map[string]any `json:"error"` +} + func openAIResponsesStreamErrorCode(status int) string { switch status { case http.StatusUnauthorized: @@ -117,3 +128,63 @@ func BuildOpenAIResponsesStreamErrorChunk(status int, errText string, sequenceNu } return []byte(`{"type":"error","code":"internal_server_error","message":"internal error","sequence_number":0}`) } + +func openAIResponsesStreamFailedErrorDetail(status int, errText, code, message string) map[string]any { + var payload map[string]any + if errUnmarshal := json.Unmarshal([]byte(strings.TrimSpace(errText)), &payload); errUnmarshal == nil { + if errorDetail, ok := payload["error"].(map[string]any); ok { + return errorDetail + } + if response, ok := payload["response"].(map[string]any); ok { + if errorDetail, ok := response["error"].(map[string]any); ok { + return errorDetail + } + } + } + + errorType := "invalid_request_error" + if status >= http.StatusInternalServerError { + errorType = "server_error" + } + return map[string]any{ + "type": errorType, + "code": code, + "message": message, + } +} + +// BuildOpenAIResponsesStreamFailedChunk builds the terminal Responses event used by official Codex clients. +// It is intentionally separate from BuildOpenAIResponsesStreamErrorChunk so existing clients keep the legacy shape. +func BuildOpenAIResponsesStreamFailedChunk(status int, errText string, sequenceNumber int) []byte { + if status <= 0 { + status = http.StatusInternalServerError + } + if sequenceNumber < 0 { + sequenceNumber = 0 + } + + legacyChunk := BuildOpenAIResponsesStreamErrorChunk(status, errText, sequenceNumber) + var legacyPayload openAIResponsesStreamErrorChunk + if errUnmarshal := json.Unmarshal(legacyChunk, &legacyPayload); errUnmarshal != nil { + legacyPayload.Code = openAIResponsesStreamErrorCode(status) + legacyPayload.Message = http.StatusText(status) + legacyPayload.SequenceNumber = sequenceNumber + } + if sequenceNumber == 0 && legacyPayload.SequenceNumber > 0 { + sequenceNumber = legacyPayload.SequenceNumber + } + + data, errMarshal := json.Marshal(openAIResponsesStreamFailedChunk{ + Type: "response.failed", + SequenceNumber: sequenceNumber, + Response: openAIResponsesStreamFailedResponse{ + Status: "failed", + Error: openAIResponsesStreamFailedErrorDetail(status, errText, legacyPayload.Code, legacyPayload.Message), + }, + }) + if errMarshal == nil { + return data + } + + return []byte(`{"type":"response.failed","sequence_number":0,"response":{"status":"failed","error":{"type":"server_error","code":"internal_server_error","message":"internal error"}}}`) +} diff --git a/sdk/api/handlers/openai_responses_stream_error_test.go b/sdk/api/handlers/openai_responses_stream_error_test.go index 90b2c6678..c6dfd25ef 100644 --- a/sdk/api/handlers/openai_responses_stream_error_test.go +++ b/sdk/api/handlers/openai_responses_stream_error_test.go @@ -46,3 +46,45 @@ func TestBuildOpenAIResponsesStreamErrorChunkExtractsHTTPErrorBody(t *testing.T) t.Fatalf("message = %v, want %q", payload["message"], "oops") } } + +func TestBuildOpenAIResponsesStreamFailedChunkPreservesNestedError(t *testing.T) { + chunk := BuildOpenAIResponsesStreamFailedChunk( + http.StatusBadRequest, + `{"error":{"type":"invalid_request","code":"cyber_policy","message":"blocked","param":null}}`, + 0, + ) + + var payload struct { + Type string `json:"type"` + SequenceNumber int `json:"sequence_number"` + Response struct { + Status string `json:"status"` + Error struct { + Type string `json:"type"` + Code string `json:"code"` + Message string `json:"message"` + } `json:"error"` + } `json:"response"` + } + if err := json.Unmarshal(chunk, &payload); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if payload.Type != "response.failed" { + t.Fatalf("type = %q, want %q", payload.Type, "response.failed") + } + if payload.SequenceNumber != 0 { + t.Fatalf("sequence_number = %d, want 0", payload.SequenceNumber) + } + if payload.Response.Status != "failed" { + t.Fatalf("response.status = %q, want %q", payload.Response.Status, "failed") + } + if payload.Response.Error.Type != "invalid_request" { + t.Fatalf("response.error.type = %q, want %q", payload.Response.Error.Type, "invalid_request") + } + if payload.Response.Error.Code != "cyber_policy" { + t.Fatalf("response.error.code = %q, want %q", payload.Response.Error.Code, "cyber_policy") + } + if payload.Response.Error.Message != "blocked" { + t.Fatalf("response.error.message = %q, want %q", payload.Response.Error.Message, "blocked") + } +}