From 49c8ec69d01e5aad244a33523e7233637ea2be8a Mon Sep 17 00:00:00 2001 From: canxin121 Date: Mon, 23 Feb 2026 12:52:25 +0800 Subject: [PATCH] fix(openai): emit valid responses stream error chunks When /v1/responses streaming fails after headers are sent, we now emit a type=error chunk instead of an HTTP-style {error:{...}} payload, preventing AI SDK chunk validation errors. --- .../openai/openai_responses_handlers.go | 4 +- ...ai_responses_handlers_stream_error_test.go | 43 +++++++ .../handlers/openai_responses_stream_error.go | 119 ++++++++++++++++++ .../openai_responses_stream_error_test.go | 48 +++++++ 4 files changed, 212 insertions(+), 2 deletions(-) create mode 100644 sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go create mode 100644 sdk/api/handlers/openai_responses_stream_error.go create mode 100644 sdk/api/handlers/openai_responses_stream_error_test.go diff --git a/sdk/api/handlers/openai/openai_responses_handlers.go b/sdk/api/handlers/openai/openai_responses_handlers.go index 1cd7e04f..3bca75f9 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers.go +++ b/sdk/api/handlers/openai/openai_responses_handlers.go @@ -265,8 +265,8 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesStream(c *gin.Context, flush if errMsg.Error != nil && errMsg.Error.Error() != "" { errText = errMsg.Error.Error() } - body := handlers.BuildErrorResponseBody(status, errText) - _, _ = fmt.Fprintf(c.Writer, "\nevent: error\ndata: %s\n\n", string(body)) + chunk := handlers.BuildOpenAIResponsesStreamErrorChunk(status, errText, 0) + _, _ = fmt.Fprintf(c.Writer, "\nevent: error\ndata: %s\n\n", string(chunk)) }, WriteDone: func() { _, _ = c.Writer.Write([]byte("\n")) 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 new file mode 100644 index 00000000..dce73807 --- /dev/null +++ b/sdk/api/handlers/openai/openai_responses_handlers_stream_error_test.go @@ -0,0 +1,43 @@ +package openai + +import ( + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/router-for-me/CLIProxyAPI/v6/internal/interfaces" + "github.com/router-for-me/CLIProxyAPI/v6/sdk/api/handlers" + sdkconfig "github.com/router-for-me/CLIProxyAPI/v6/sdk/config" +) + +func TestForwardResponsesStreamTerminalErrorUsesResponsesErrorChunk(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) + + flusher, ok := c.Writer.(http.Flusher) + if !ok { + t.Fatalf("expected gin writer to implement http.Flusher") + } + + data := make(chan []byte) + errs := make(chan *interfaces.ErrorMessage, 1) + errs <- &interfaces.ErrorMessage{StatusCode: http.StatusInternalServerError, Error: errors.New("unexpected EOF")} + close(errs) + + h.forwardResponsesStream(c, flusher, func(error) {}, data, errs) + body := recorder.Body.String() + if !strings.Contains(body, `"type":"error"`) { + t.Fatalf("expected responses error chunk, got: %q", body) + } + if strings.Contains(body, `"error":{`) { + t.Fatalf("expected streaming error chunk (top-level type), got HTTP error body: %q", body) + } +} diff --git a/sdk/api/handlers/openai_responses_stream_error.go b/sdk/api/handlers/openai_responses_stream_error.go new file mode 100644 index 00000000..e7760bd0 --- /dev/null +++ b/sdk/api/handlers/openai_responses_stream_error.go @@ -0,0 +1,119 @@ +package handlers + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" +) + +type openAIResponsesStreamErrorChunk struct { + Type string `json:"type"` + Code string `json:"code"` + Message string `json:"message"` + SequenceNumber int `json:"sequence_number"` +} + +func openAIResponsesStreamErrorCode(status int) string { + switch status { + case http.StatusUnauthorized: + return "invalid_api_key" + case http.StatusForbidden: + return "insufficient_quota" + case http.StatusTooManyRequests: + return "rate_limit_exceeded" + case http.StatusNotFound: + return "model_not_found" + case http.StatusRequestTimeout: + return "request_timeout" + default: + if status >= http.StatusInternalServerError { + return "internal_server_error" + } + if status >= http.StatusBadRequest { + return "invalid_request_error" + } + return "unknown_error" + } +} + +// BuildOpenAIResponsesStreamErrorChunk builds an OpenAI Responses streaming error chunk. +// +// Important: OpenAI's HTTP error bodies are shaped like {"error":{...}}; those are valid for +// non-streaming responses, but streaming clients validate SSE `data:` payloads against a union +// of chunks that requires a top-level `type` field. +func BuildOpenAIResponsesStreamErrorChunk(status int, errText string, sequenceNumber int) []byte { + if status <= 0 { + status = http.StatusInternalServerError + } + if sequenceNumber < 0 { + sequenceNumber = 0 + } + + message := strings.TrimSpace(errText) + if message == "" { + message = http.StatusText(status) + } + + code := openAIResponsesStreamErrorCode(status) + + trimmed := strings.TrimSpace(errText) + if trimmed != "" && json.Valid([]byte(trimmed)) { + var payload map[string]any + if err := json.Unmarshal([]byte(trimmed), &payload); err == nil { + if t, ok := payload["type"].(string); ok && strings.TrimSpace(t) == "error" { + if m, ok := payload["message"].(string); ok && strings.TrimSpace(m) != "" { + message = strings.TrimSpace(m) + } + if v, ok := payload["code"]; ok && v != nil { + if c, ok := v.(string); ok && strings.TrimSpace(c) != "" { + code = strings.TrimSpace(c) + } else { + code = strings.TrimSpace(fmt.Sprint(v)) + } + } + if v, ok := payload["sequence_number"].(float64); ok && sequenceNumber == 0 { + sequenceNumber = int(v) + } + } + if e, ok := payload["error"].(map[string]any); ok { + if m, ok := e["message"].(string); ok && strings.TrimSpace(m) != "" { + message = strings.TrimSpace(m) + } + if v, ok := e["code"]; ok && v != nil { + if c, ok := v.(string); ok && strings.TrimSpace(c) != "" { + code = strings.TrimSpace(c) + } else { + code = strings.TrimSpace(fmt.Sprint(v)) + } + } + } + } + } + + if strings.TrimSpace(code) == "" { + code = "unknown_error" + } + + data, err := json.Marshal(openAIResponsesStreamErrorChunk{ + Type: "error", + Code: code, + Message: message, + SequenceNumber: sequenceNumber, + }) + if err == nil { + return data + } + + // Extremely defensive fallback. + data, _ = json.Marshal(openAIResponsesStreamErrorChunk{ + Type: "error", + Code: "internal_server_error", + Message: message, + SequenceNumber: sequenceNumber, + }) + if len(data) > 0 { + return data + } + return []byte(`{"type":"error","code":"internal_server_error","message":"internal error","sequence_number":0}`) +} diff --git a/sdk/api/handlers/openai_responses_stream_error_test.go b/sdk/api/handlers/openai_responses_stream_error_test.go new file mode 100644 index 00000000..90b2c667 --- /dev/null +++ b/sdk/api/handlers/openai_responses_stream_error_test.go @@ -0,0 +1,48 @@ +package handlers + +import ( + "encoding/json" + "net/http" + "testing" +) + +func TestBuildOpenAIResponsesStreamErrorChunk(t *testing.T) { + chunk := BuildOpenAIResponsesStreamErrorChunk(http.StatusInternalServerError, "unexpected EOF", 0) + var payload map[string]any + if err := json.Unmarshal(chunk, &payload); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if payload["type"] != "error" { + t.Fatalf("type = %v, want %q", payload["type"], "error") + } + if payload["code"] != "internal_server_error" { + t.Fatalf("code = %v, want %q", payload["code"], "internal_server_error") + } + if payload["message"] != "unexpected EOF" { + t.Fatalf("message = %v, want %q", payload["message"], "unexpected EOF") + } + if payload["sequence_number"] != float64(0) { + t.Fatalf("sequence_number = %v, want %v", payload["sequence_number"], 0) + } +} + +func TestBuildOpenAIResponsesStreamErrorChunkExtractsHTTPErrorBody(t *testing.T) { + chunk := BuildOpenAIResponsesStreamErrorChunk( + http.StatusInternalServerError, + `{"error":{"message":"oops","type":"server_error","code":"internal_server_error"}}`, + 0, + ) + var payload map[string]any + if err := json.Unmarshal(chunk, &payload); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if payload["type"] != "error" { + t.Fatalf("type = %v, want %q", payload["type"], "error") + } + if payload["code"] != "internal_server_error" { + t.Fatalf("code = %v, want %q", payload["code"], "internal_server_error") + } + if payload["message"] != "oops" { + t.Fatalf("message = %v, want %q", payload["message"], "oops") + } +}