From 09da52ad509e2c18e7b9540db3b98c2214c280aa Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Thu, 16 Jul 2026 04:38:19 +0800 Subject: [PATCH] feat(translator): improve error handling and response conversion for incomplete statuses - Enhanced response handling to support `response.incomplete` events across translator components and executors. - Refactored `resultErrorFromError` to centralize error conversion and consolidate repeated logic for constructing custom errors. - Updated handling of `max_output_tokens` and similar terminal reasons in OpenAI and Gemini translator workflows. - Introduced `IsRequestScoped` and `IsRequestScopedError` to differentiate between request-scoped and non-request-scoped errors. - Added comprehensive test cases for incomplete responses, terminal failures, and error propagation in both streaming and non-streaming conditions. Closes: #3055 --- internal/runtime/executor/codex_executor.go | 125 ++++-- .../codex_executor_stream_output_test.go | 359 ++++++++++++++++++ .../codex/gemini/codex_gemini_response.go | 33 +- .../gemini/codex_gemini_response_test.go | 19 + .../interactions_codex_response.go | 8 +- .../interactions/interactions_codex_test.go | 18 + .../chat-completions/codex_openai_response.go | 46 ++- .../codex_openai_response_test.go | 29 ++ .../codex_openai-responses_response.go | 5 +- .../codex_openai-responses_response_test.go | 21 + sdk/cliproxy/auth/conductor.go | 89 +++-- sdk/cliproxy/auth/conductor_overrides_test.go | 127 +++++++ sdk/cliproxy/auth/errors.go | 8 + sdk/cliproxy/auth/errors_compat_test.go | 16 + sdk/cliproxy/executor/types.go | 8 + 15 files changed, 817 insertions(+), 94 deletions(-) create mode 100644 internal/translator/codex/openai/responses/codex_openai-responses_response_test.go create mode 100644 sdk/cliproxy/auth/errors_compat_test.go diff --git a/internal/runtime/executor/codex_executor.go b/internal/runtime/executor/codex_executor.go index d9ac0c2d0..c4922bec3 100644 --- a/internal/runtime/executor/codex_executor.go +++ b/internal/runtime/executor/codex_executor.go @@ -44,6 +44,23 @@ const ( 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. @@ -116,6 +133,50 @@ func codexTerminalStreamContextLengthErr(eventData []byte) (statusErr, bool) { } 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 { @@ -130,15 +191,12 @@ func codexTerminalStreamErr(eventData []byte) (statusErr, []byte, bool) { body = codexTerminalErrorBody(eventData, "error") } default: - return statusErr{}, nil, false + return nil, false } if len(body) == 0 { - return statusErr{}, nil, false + body = []byte(`{"error":{"message":"upstream stream failed without error details"}}`) } - if !codexTerminalStreamErrShouldHandle(body) { - return statusErr{}, nil, false - } - return newCodexStatusErr(http.StatusBadRequest, body), body, true + return body, true } func codexTerminalStreamErrShouldHandle(body []byte) bool { @@ -854,11 +912,7 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re err = newCodexStatusErr(httpResp.StatusCode, b) return resp, err } - data, err := io.ReadAll(httpResp.Body) - if err != nil { - helps.RecordAPIResponseError(ctx, e.cfg, err) - return resp, err - } + data, errRead := io.ReadAll(httpResp.Body) upstreamData := applyCodexIdentityConfuseResponsePayload(data, identityState) helps.AppendAPIResponseChunk(ctx, e.cfg, upstreamData) @@ -873,7 +927,7 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re eventData := bytes.TrimSpace(line[5:]) eventType := gjson.GetBytes(eventData, "type").String() - if streamErr, terminalBody, ok := codexTerminalStreamErr(eventData); ok { + if streamErr, terminalBody, ok := codexTerminalFailureErr(eventData); ok { if errClearReplay := clearCodexReasoningReplayOnInvalidSignature(ctx, replayScope, streamErr.StatusCode(), terminalBody); errClearReplay != nil { return resp, errClearReplay } @@ -895,7 +949,7 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re continue } - if eventType != "response.completed" { + if eventType != "response.completed" && eventType != "response.incomplete" { continue } @@ -926,7 +980,9 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re } completedData = completedDataPatched } - cacheCodexReasoningReplayFromCompleted(replayScope, completedData) + if eventType == "response.completed" { + cacheCodexReasoningReplayFromCompleted(replayScope, completedData) + } var param any clientCompletedData := applyCodexIdentityExposeResponsePayload(completedData, identityState) @@ -934,7 +990,15 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil } - err = statusErr{code: 408, msg: "stream error: stream disconnected before completion: stream closed before response.completed"} + if errRead != nil { + if errCtx := ctx.Err(); errCtx != nil { + helps.RecordAPIResponseError(ctx, e.cfg, errCtx) + err = errCtx + return resp, err + } + helps.RecordAPIResponseError(ctx, e.cfg, errRead) + } + err = newCodexIncompleteStreamError() return resp, err } @@ -1159,10 +1223,12 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au line := applyCodexIdentityConfuseResponsePayload(scanner.Bytes(), identityState) helps.AppendAPIResponseChunk(ctx, e.cfg, line) translatedLine := bytes.Clone(line) + terminalSuccess := false if bytes.HasPrefix(line, dataTag) { data := bytes.TrimSpace(line[5:]) - if streamErr, terminalBody, ok := codexTerminalStreamErr(data); ok { + eventType := gjson.GetBytes(data, "type").String() + if streamErr, terminalBody, ok := codexTerminalFailureErr(data); ok { if errClearReplay := clearCodexReasoningReplayOnInvalidSignature(ctx, replayScope, streamErr.StatusCode(), terminalBody); errClearReplay != nil { helps.RecordAPIResponseError(ctx, e.cfg, errClearReplay) reporter.PublishFailure(ctx, errClearReplay) @@ -1180,16 +1246,19 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au } return } - switch gjson.GetBytes(data, "type").String() { + switch eventType { case "response.output_item.done": collectCodexOutputItemDone(data, outputItemsByIndex, &outputItemsFallback) - case "response.completed": + case "response.completed", "response.incomplete": + terminalSuccess = true if detail, ok := helps.ParseCodexUsage(data); ok { reporter.Publish(ctx, detail) } publishCodexImageToolUsage(ctx, reporter, body, data) data = patchCodexCompletedOutput(data, outputItemsByIndex, outputItemsFallback) - cacheCodexReasoningReplayFromCompleted(replayScope, data) + if eventType == "response.completed" { + cacheCodexReasoningReplayFromCompleted(replayScope, data) + } translatedLine = append([]byte("data: "), data...) } } @@ -1203,14 +1272,22 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au return } } + if terminalSuccess { + return + } } if errScan := scanner.Err(); errScan != nil { - helps.RecordAPIResponseError(ctx, e.cfg, errScan) - reporter.PublishFailure(ctx, errScan) - select { - case out <- cliproxyexecutor.StreamChunk{Err: errScan}: - case <-ctx.Done(): + if ctx.Err() != nil { + return } + helps.RecordAPIResponseError(ctx, e.cfg, errScan) + } + streamErr := newCodexIncompleteStreamError() + helps.RecordAPIResponseError(ctx, e.cfg, streamErr) + reporter.PublishFailure(ctx, streamErr) + select { + case out <- cliproxyexecutor.StreamChunk{Err: streamErr}: + case <-ctx.Done(): } }() return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil diff --git a/internal/runtime/executor/codex_executor_stream_output_test.go b/internal/runtime/executor/codex_executor_stream_output_test.go index f495d3c1e..477a9800a 100644 --- a/internal/runtime/executor/codex_executor_stream_output_test.go +++ b/internal/runtime/executor/codex_executor_stream_output_test.go @@ -3,6 +3,7 @@ package executor import ( "bytes" "context" + "io" "net/http" "net/http/httptest" "strings" @@ -85,6 +86,296 @@ func TestCodexExecutorExecuteSurfacesTerminalStreamError(t *testing.T) { } } +func TestCodexExecutorExecuteIncompleteResponseIsSuccessful(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(`data: {"type":"response.incomplete","response":{"id":"resp_1","model":"gpt-5.5","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[],"usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}` + "\n\n")) + })) + defer server.Close() + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": server.URL, + "api_key": "test", + }} + + resp, err := executor.Execute(context.Background(), auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","messages":[{"role":"user","content":"hello"}]}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("claude"), + Stream: false, + }) + if err != nil { + t.Fatalf("Execute error: %v", err) + } + if got := gjson.GetBytes(resp.Payload, "stop_reason").String(); got != "max_tokens" { + t.Fatalf("stop_reason = %q, want %q; payload=%s", got, "max_tokens", resp.Payload) + } +} + +func TestCodexExecutorExecuteExplicitTerminalFailureIsNotRequestScoped(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(`data: {"type":"error","error":{"type":"invalid_request_error","code":"invalid_value","message":"Invalid input."}}` + "\n\n")) + })) + defer server.Close() + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": server.URL, + "api_key": "test", + }} + + _, err := executor.Execute(context.Background(), auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Stream: false, + }) + if err == nil { + t.Fatal("expected explicit terminal failure, got nil") + } + if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) + } + assertNotRequestScopedTestError(t, err) +} + +func TestCodexExecutorExecuteMissingCompletionIsRequestScoped(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5.5\"}}\n\n")) + })) + defer server.Close() + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": server.URL, + "api_key": "test", + }} + + _, err := executor.Execute(context.Background(), auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Stream: false, + }) + if err == nil { + t.Fatal("expected missing-completion error, got nil") + } + if got := statusCodeFromTestError(t, err); got != http.StatusRequestTimeout { + t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusRequestTimeout, err) + } + assertRequestScopedTestError(t, err) +} + +func TestCodexExecutorExecuteStreamMissingCompletionIsRequestScoped(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5.5\"}}\n\n")) + })) + defer server.Close() + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": server.URL, + "api_key": "test", + }} + + result, err := executor.ExecuteStream(context.Background(), auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Stream: true, + }) + if err != nil { + t.Fatalf("ExecuteStream error: %v", err) + } + + var streamErr error + for chunk := range result.Chunks { + if chunk.Err != nil { + streamErr = chunk.Err + } + } + if streamErr == nil { + t.Fatal("expected missing-completion stream error, got nil") + } + if got := statusCodeFromTestError(t, streamErr); got != http.StatusRequestTimeout { + t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusRequestTimeout, streamErr) + } + assertRequestScopedTestError(t, streamErr) +} + +func TestCodexExecutorExecuteStreamExplicitTerminalFailureIsNotSuccessful(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5.5\"}}\n\n")) + _, _ = w.Write([]byte(`data: {"type":"error","error":{"type":"invalid_request_error","code":"invalid_value","message":"Invalid input."}}` + "\n\n")) + })) + defer server.Close() + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": server.URL, + "api_key": "test", + }} + + result, err := executor.ExecuteStream(context.Background(), auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Stream: true, + }) + if err != nil { + t.Fatalf("ExecuteStream error: %v", err) + } + + var streamErr error + for chunk := range result.Chunks { + if chunk.Err != nil { + streamErr = chunk.Err + } + } + if streamErr == nil { + t.Fatal("expected explicit terminal stream error, got nil") + } + if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, streamErr) + } + assertNotRequestScopedTestError(t, streamErr) +} + +func TestCodexExecutorTransportFailureBeforeTerminalIsRequestScoped(t *testing.T) { + tests := []struct { + name string + stream bool + }{ + {name: "non-streaming"}, + {name: "streaming", stream: true}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + created := []byte("data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5.5\"}}\n\n") + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", roundTripperFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(io.MultiReader(bytes.NewReader(created), unexpectedEOFReader{})), + Request: req, + }, nil + })) + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": "http://codex.test", + "api_key": "test", + }} + req := cliproxyexecutor.Request{Model: "gpt-5.5", Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`)} + opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai-response"), Stream: tc.stream} + + var terminalErr error + if tc.stream { + result, errStream := executor.ExecuteStream(ctx, auth, req, opts) + if errStream != nil { + t.Fatalf("ExecuteStream error: %v", errStream) + } + for chunk := range result.Chunks { + if chunk.Err != nil { + terminalErr = chunk.Err + } + } + } else { + _, terminalErr = executor.Execute(ctx, auth, req, opts) + } + if terminalErr == nil { + t.Fatal("expected transport failure before terminal event") + } + if got := statusCodeFromTestError(t, terminalErr); got != http.StatusRequestTimeout { + t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusRequestTimeout, terminalErr) + } + assertRequestScopedTestError(t, terminalErr) + }) + } +} + +func TestCodexExecutorExecuteIgnoresTransportErrorAfterCompletion(t *testing.T) { + completed := []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5.5\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n") + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", roundTripperFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(io.MultiReader(bytes.NewReader(completed), unexpectedEOFReader{})), + Request: req, + }, nil + })) + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": "http://codex.test", + "api_key": "test", + }} + + resp, err := executor.Execute(ctx, auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Stream: false, + }) + if err != nil { + t.Fatalf("unexpected error after response.completed: %v", err) + } + if got := gjson.GetBytes(resp.Payload, "id").String(); got != "resp_1" { + t.Fatalf("response id = %q, want resp_1; payload=%s", got, resp.Payload) + } +} + +func TestCodexExecutorExecuteStreamIgnoresTransportErrorAfterCompletion(t *testing.T) { + completed := []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5.5\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n") + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", roundTripperFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": {"text/event-stream"}}, + Body: io.NopCloser(io.MultiReader(bytes.NewReader(completed), unexpectedEOFReader{})), + Request: req, + }, nil + })) + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{Attributes: map[string]string{ + "base_url": "http://codex.test", + "api_key": "test", + }} + + result, err := executor.ExecuteStream(ctx, auth, cliproxyexecutor.Request{ + Model: "gpt-5.5", + Payload: []byte(`{"model":"gpt-5.5","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Stream: true, + }) + if err != nil { + t.Fatalf("ExecuteStream error: %v", err) + } + + var streamErr error + for chunk := range result.Chunks { + if chunk.Err != nil { + streamErr = chunk.Err + } + } + if streamErr != nil { + t.Fatalf("unexpected error after response.completed: %v", streamErr) + } +} + func TestCodexExecutorExecuteStreamSurfacesTerminalStreamError(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") @@ -167,6 +458,47 @@ func TestCodexTerminalStreamErrIgnoresRateLimitTerminalErrors(t *testing.T) { } } +func TestCodexTerminalFailureErrClassifiesStatus(t *testing.T) { + tests := []struct { + name string + event string + wantStatus int + }{ + { + name: "invalid request", + event: `{"type":"error","error":{"type":"invalid_request_error","code":"invalid_value","message":"Invalid input."}}`, + wantStatus: http.StatusBadRequest, + }, + { + name: "authentication", + event: `{"type":"response.failed","response":{"error":{"type":"authentication_error","code":"invalid_api_key","message":"Invalid token."}}}`, + wantStatus: http.StatusUnauthorized, + }, + { + name: "rate limit", + event: `{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"Rate limit reached."}}`, + wantStatus: http.StatusTooManyRequests, + }, + { + name: "unknown upstream failure", + event: `{"type":"response.failed","response":{"error":{"type":"upstream_error","code":"unknown","message":"Upstream failed."}}}`, + wantStatus: http.StatusBadGateway, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + streamErr, _, ok := codexTerminalFailureErr([]byte(tc.event)) + if !ok { + t.Fatal("expected terminal failure to be handled") + } + if got := streamErr.StatusCode(); got != tc.wantStatus { + t.Fatalf("status code = %d, want %d; err=%v", got, tc.wantStatus, streamErr) + } + }) + } +} + func TestCodexTerminalStreamErrHandlesUsageLimitErrorEvent(t *testing.T) { streamErr, _, ok := codexTerminalStreamErr([]byte(`{"type":"error","error":{"type":"usage_limit_reached","message":"You've hit your usage limit.","resets_in_seconds":300}}`)) if !ok { @@ -207,6 +539,33 @@ func statusCodeFromTestError(t *testing.T, err error) int { return statusErr.StatusCode() } +func assertRequestScopedTestError(t *testing.T, err error) { + t.Helper() + + requestErr, ok := err.(interface{ IsRequestScoped() bool }) + if !ok { + t.Fatalf("error %T does not expose IsRequestScoped(): %v", err, err) + } + if !requestErr.IsRequestScoped() { + t.Fatalf("error %T is not request-scoped: %v", err, err) + } +} + +func assertNotRequestScopedTestError(t *testing.T, err error) { + t.Helper() + + requestErr, ok := err.(interface{ IsRequestScoped() bool }) + if ok && requestErr.IsRequestScoped() { + t.Fatalf("error %T is unexpectedly request-scoped: %v", err, err) + } +} + +type unexpectedEOFReader struct{} + +func (unexpectedEOFReader) Read([]byte) (int, error) { + return 0, io.ErrUnexpectedEOF +} + func TestCodexExecutorExecuteStream_EmptyStreamCompletionOutputUsesOutputItemDone(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") diff --git a/internal/translator/codex/gemini/codex_gemini_response.go b/internal/translator/codex/gemini/codex_gemini_response.go index a5144ea63..6500de568 100644 --- a/internal/translator/codex/gemini/codex_gemini_response.go +++ b/internal/translator/codex/gemini/codex_gemini_response.go @@ -210,11 +210,14 @@ func ConvertCodexResponseToGemini(_ context.Context, modelName string, originalR return [][]byte{template} } return [][]byte{} - } else if typeStr == "response.completed" { // Handle response completion with usage metadata + } else if typeStr == "response.completed" || typeStr == "response.incomplete" { // Handle response completion with usage metadata template, _ = sjson.SetBytes(template, "usageMetadata.promptTokenCount", rootResult.Get("response.usage.input_tokens").Int()) template, _ = sjson.SetBytes(template, "usageMetadata.candidatesTokenCount", rootResult.Get("response.usage.output_tokens").Int()) totalTokens := rootResult.Get("response.usage.input_tokens").Int() + rootResult.Get("response.usage.output_tokens").Int() template, _ = sjson.SetBytes(template, "usageMetadata.totalTokenCount", totalTokens) + if typeStr == "response.incomplete" { + template, _ = sjson.SetBytes(template, "candidates.0.finishReason", codexGeminiIncompleteFinishReason(rootResult.Get("response.incomplete_details.reason").String())) + } } else { return [][]byte{} } @@ -243,8 +246,9 @@ func ConvertCodexResponseToGemini(_ context.Context, modelName string, originalR func ConvertCodexResponseToGeminiNonStream(_ context.Context, modelName string, originalRequestRawJSON, requestRawJSON, rawJSON []byte, _ *any) []byte { rootResult := gjson.ParseBytes(rawJSON) - // Verify this is a response.completed event - if rootResult.Get("type").String() != "response.completed" { + // Verify this is a terminal response event. + responseType := rootResult.Get("type").String() + if responseType != "response.completed" && responseType != "response.incomplete" { return []byte{} } @@ -257,6 +261,9 @@ func ConvertCodexResponseToGeminiNonStream(_ context.Context, modelName string, // Set response metadata from the completed response responseData := rootResult.Get("response") if responseData.Exists() { + if responseType == "response.incomplete" { + template, _ = sjson.SetBytes(template, "candidates.0.finishReason", codexGeminiIncompleteFinishReason(responseData.Get("incomplete_details.reason").String())) + } // Set response ID if responseId := responseData.Get("id"); responseId.Exists() { template, _ = sjson.SetBytes(template, "responseId", responseId.String()) @@ -279,7 +286,6 @@ func ConvertCodexResponseToGeminiNonStream(_ context.Context, modelName string, } // Process output content to build parts array - hasToolCall := false var pendingFunctionCalls [][]byte flushPendingFunctionCalls := func() { @@ -344,7 +350,6 @@ func ConvertCodexResponseToGeminiNonStream(_ context.Context, modelName string, case "function_call": // Collect function call for potential merging with consecutive ones - hasToolCall = true functionCall := []byte(`{"functionCall":{"args":{},"name":""}}`) { n := value.Get("name").String() @@ -372,13 +377,6 @@ func ConvertCodexResponseToGeminiNonStream(_ context.Context, modelName string, // Handle any remaining pending function calls at the end flushPendingFunctionCalls() } - - // Set finish reason based on whether there were tool calls - if hasToolCall { - template, _ = sjson.SetBytes(template, "candidates.0.finishReason", "STOP") - } else { - template, _ = sjson.SetBytes(template, "candidates.0.finishReason", "STOP") - } } return template } @@ -423,6 +421,17 @@ func setGeminiFunctionCallID(functionCall []byte, item gjson.Result) []byte { return functionCall } +func codexGeminiIncompleteFinishReason(reason string) string { + switch reason { + case "max_tokens", "max_output_tokens": + return "MAX_TOKENS" + case "content_filter": + return "SAFETY" + default: + return "OTHER" + } +} + func GeminiTokenCount(ctx context.Context, count int64) []byte { return translatorcommon.GeminiTokenCountJSON(count) } diff --git a/internal/translator/codex/gemini/codex_gemini_response_test.go b/internal/translator/codex/gemini/codex_gemini_response_test.go index 55b135290..5dda9cd58 100644 --- a/internal/translator/codex/gemini/codex_gemini_response_test.go +++ b/internal/translator/codex/gemini/codex_gemini_response_test.go @@ -7,6 +7,25 @@ import ( "github.com/tidwall/gjson" ) +func TestConvertCodexResponseToGemini_IncompleteTerminal(t *testing.T) { + ctx := context.Background() + terminal := []byte(`{"type":"response.incomplete","response":{"id":"resp_1","model":"gpt-5.5","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[],"usage":{"input_tokens":1,"output_tokens":2,"total_tokens":3}}}`) + + var param any + streamOut := ConvertCodexResponseToGemini(ctx, "gemini-2.5-pro", nil, nil, append([]byte("data: "), terminal...), ¶m) + if len(streamOut) != 1 { + t.Fatalf("expected 1 streaming terminal chunk, got %d", len(streamOut)) + } + if got := gjson.GetBytes(streamOut[0], "candidates.0.finishReason").String(); got != "MAX_TOKENS" { + t.Fatalf("stream finishReason = %q, want MAX_TOKENS; payload=%s", got, streamOut[0]) + } + + nonStreamOut := ConvertCodexResponseToGeminiNonStream(ctx, "gemini-2.5-pro", nil, nil, terminal, nil) + if got := gjson.GetBytes(nonStreamOut, "candidates.0.finishReason").String(); got != "MAX_TOKENS" { + t.Fatalf("non-stream finishReason = %q, want MAX_TOKENS; payload=%s", got, nonStreamOut) + } +} + func TestConvertCodexResponseToGemini_StreamEmptyOutputUsesOutputItemDoneMessageFallback(t *testing.T) { ctx := context.Background() originalRequest := []byte(`{"tools":[]}`) diff --git a/internal/translator/codex/interactions/interactions_codex_response.go b/internal/translator/codex/interactions/interactions_codex_response.go index dec2b28aa..cea921a52 100644 --- a/internal/translator/codex/interactions/interactions_codex_response.go +++ b/internal/translator/codex/interactions/interactions_codex_response.go @@ -68,7 +68,7 @@ func ConvertCodexResponseToInteractions(ctx context.Context, modelName string, o return codexFunctionArgumentsDeltaToInteractions(st, root) case "response.output_item.done": return codexOutputItemDoneToInteractions(st, root.Get("item")) - case "response.completed": + case "response.completed", "response.incomplete": out := appendCodexInteractionsCreated(nil, st, root.Get("response")) out = appendCodexInteractionsStepStop(out, st) out = appendCodexInteractionsCompleted(out, st, root.Get("response")) @@ -88,6 +88,9 @@ func ConvertCodexResponseToInteractionsNonStream(ctx context.Context, modelName response = root } out := []byte(`{"id":"","object":"interaction","status":"completed","model":"","steps":[]}`) + if status := response.Get("status").String(); status != "" { + out, _ = sjson.SetBytes(out, "status", status) + } id := response.Get("id").String() if id == "" { id = fmt.Sprintf("interaction_%d", time.Now().UnixNano()) @@ -168,6 +171,9 @@ func appendCodexInteractionsCompleted(out [][]byte, st *codexToInteractionsStrea completed, _ = sjson.SetBytes(completed, "interaction.created", created.Format(time.RFC3339)) completed, _ = sjson.SetBytes(completed, "interaction.updated", time.Now().UTC().Format(time.RFC3339)) completed, _ = sjson.SetBytes(completed, "interaction.model", st.Model) + if status := response.Get("status").String(); status != "" { + completed, _ = sjson.SetBytes(completed, "interaction.status", status) + } completed = setCodexInteractionsUsage(completed, "interaction.usage", response.Get("usage"), true) out = append(out, translatorcommon.SSEEventData("interaction.completed", completed)) st.Completed = true diff --git a/internal/translator/codex/interactions/interactions_codex_test.go b/internal/translator/codex/interactions/interactions_codex_test.go index 34a3fecda..5c6b38eb5 100644 --- a/internal/translator/codex/interactions/interactions_codex_test.go +++ b/internal/translator/codex/interactions/interactions_codex_test.go @@ -84,6 +84,24 @@ func TestConvertInteractionsRequestToCodexFunctionDeclarations(t *testing.T) { } } +func TestConvertCodexResponseToInteractionsIncompleteTerminal(t *testing.T) { + raw := []byte(`{"type":"response.incomplete","response":{"id":"resp_1","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[],"usage":{"input_tokens":1,"output_tokens":2,"total_tokens":3}}}`) + nonStreamOut := ConvertCodexResponseToInteractionsNonStream(context.Background(), "codex-test", nil, nil, raw, nil) + if got := gjson.GetBytes(nonStreamOut, "status").String(); got != "incomplete" { + t.Fatalf("non-stream status = %q, want incomplete. Output: %s", got, nonStreamOut) + } + + var param any + streamOut := ConvertCodexResponseToInteractions(context.Background(), "codex-test", nil, nil, append([]byte("data: "), raw...), ¶m) + payload := findCodexInteractionsEventPayload(streamOut, "interaction.completed") + if len(payload) == 0 { + t.Fatalf("stream incomplete event did not terminate interaction: %q", streamOut) + } + if got := gjson.GetBytes(payload, "interaction.status").String(); got != "incomplete" { + t.Fatalf("stream status = %q, want incomplete. Payload: %s", got, payload) + } +} + func TestConvertCodexResponseToInteractionsNonStream(t *testing.T) { raw := []byte(`{"type":"response.completed","response":{"id":"resp_1","created_at":1700000000,"usage":{"input_tokens":3,"output_tokens":2},"output":[{"type":"message","content":[{"type":"output_text","text":"ok"}]},{"type":"reasoning","content":"thinking"},{"type":"function_call","call_id":"call_1","name":"lookup","arguments":"{\"q\":\"x\"}"}]}}`) out := ConvertCodexResponseToInteractionsNonStream(context.Background(), "codex-test", nil, nil, raw, nil) diff --git a/internal/translator/codex/openai/chat-completions/codex_openai_response.go b/internal/translator/codex/openai/chat-completions/codex_openai_response.go index 864472098..89bac12d5 100644 --- a/internal/translator/codex/openai/chat-completions/codex_openai_response.go +++ b/internal/translator/codex/openai/chat-completions/codex_openai_response.go @@ -163,13 +163,23 @@ func ConvertCodexResponseToOpenAI(_ context.Context, modelName string, originalR template, _ = sjson.SetBytes(template, "choices.0.delta.role", "assistant") template, _ = sjson.SetRawBytes(template, "choices.0.delta.images.-1", imagePayload) - } else if dataType == "response.completed" { + } else if dataType == "response.completed" || dataType == "response.incomplete" { finishReason := "stop" - if (*param).(*ConvertCliToOpenAIParams).FunctionCallIndex != -1 { + nativeFinishReason := finishReason + if dataType == "response.incomplete" { + nativeFinishReason = rootResult.Get("response.incomplete_details.reason").String() + switch nativeFinishReason { + case "max_tokens", "max_output_tokens": + finishReason = "length" + case "content_filter": + finishReason = "content_filter" + } + } else if (*param).(*ConvertCliToOpenAIParams).FunctionCallIndex != -1 { finishReason = "tool_calls" + nativeFinishReason = finishReason } template, _ = sjson.SetBytes(template, "choices.0.finish_reason", finishReason) - template, _ = sjson.SetBytes(template, "choices.0.native_finish_reason", finishReason) + template, _ = sjson.SetBytes(template, "choices.0.native_finish_reason", nativeFinishReason) } else if dataType == "response.output_item.added" { itemResult := rootResult.Get("item") if !itemResult.Exists() || itemResult.Get("type").String() != "function_call" { @@ -318,8 +328,9 @@ func ConvertCodexResponseToOpenAI(_ context.Context, modelName string, originalR // - []byte: An OpenAI-compatible JSON response containing all message content and metadata func ConvertCodexResponseToOpenAINonStream(_ context.Context, _ string, originalRequestRawJSON, requestRawJSON, rawJSON []byte, _ *any) []byte { rootResult := gjson.ParseBytes(rawJSON) - // Verify this is a response.completed event - if rootResult.Get("type").String() != "response.completed" { + // Verify this is a terminal response event. + responseType := rootResult.Get("type").String() + if responseType != "response.completed" && responseType != "response.incomplete" { return []byte{} } @@ -475,16 +486,33 @@ func ConvertCodexResponseToOpenAINonStream(_ context.Context, _ string, original } } - // Extract and set the finish reason based on status + // Extract and set the finish reason based on status. if statusResult := responseResult.Get("status"); statusResult.Exists() { status := statusResult.String() - if status == "completed" { - finishReason := "stop" + finishReason := "" + nativeFinishReason := "" + switch status { + case "completed": + finishReason = "stop" + nativeFinishReason = finishReason if len(toolCalls) > 0 { finishReason = "tool_calls" + nativeFinishReason = finishReason } + case "incomplete": + nativeFinishReason = responseResult.Get("incomplete_details.reason").String() + switch nativeFinishReason { + case "max_tokens", "max_output_tokens": + finishReason = "length" + case "content_filter": + finishReason = "content_filter" + default: + finishReason = "stop" + } + } + if finishReason != "" { template, _ = sjson.SetBytes(template, "choices.0.finish_reason", finishReason) - template, _ = sjson.SetBytes(template, "choices.0.native_finish_reason", finishReason) + template, _ = sjson.SetBytes(template, "choices.0.native_finish_reason", nativeFinishReason) } } diff --git a/internal/translator/codex/openai/chat-completions/codex_openai_response_test.go b/internal/translator/codex/openai/chat-completions/codex_openai_response_test.go index 4de746090..663ee2334 100644 --- a/internal/translator/codex/openai/chat-completions/codex_openai_response_test.go +++ b/internal/translator/codex/openai/chat-completions/codex_openai_response_test.go @@ -7,6 +7,35 @@ import ( "github.com/tidwall/gjson" ) +func TestConvertCodexResponseToOpenAI_IncompleteTerminal(t *testing.T) { + ctx := context.Background() + terminal := []byte(`{"type":"response.incomplete","response":{"id":"resp_1","model":"gpt-5.5","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[],"usage":{"input_tokens":1,"output_tokens":2,"total_tokens":3}}}`) + + var param any + streamOut := ConvertCodexResponseToOpenAI(ctx, "gpt-5.5", nil, nil, append([]byte("data: "), terminal...), ¶m) + if len(streamOut) != 1 { + t.Fatalf("expected 1 streaming terminal chunk, got %d", len(streamOut)) + } + if got := gjson.GetBytes(streamOut[0], "choices.0.finish_reason").String(); got != "length" { + t.Fatalf("stream finish_reason = %q, want length; payload=%s", got, streamOut[0]) + } + if got := gjson.GetBytes(streamOut[0], "choices.0.native_finish_reason").String(); got != "max_output_tokens" { + t.Fatalf("stream native_finish_reason = %q, want max_output_tokens; payload=%s", got, streamOut[0]) + } + + var toolParam any + _ = ConvertCodexResponseToOpenAI(ctx, "gpt-5.5", nil, nil, []byte(`data: {"type":"response.output_item.added","item":{"type":"function_call","call_id":"call_1","name":"lookup"}}`), &toolParam) + toolStreamOut := ConvertCodexResponseToOpenAI(ctx, "gpt-5.5", nil, nil, append([]byte("data: "), terminal...), &toolParam) + if got := gjson.GetBytes(toolStreamOut[0], "choices.0.finish_reason").String(); got != "length" { + t.Fatalf("tool stream finish_reason = %q, want length; payload=%s", got, toolStreamOut[0]) + } + + nonStreamOut := ConvertCodexResponseToOpenAINonStream(ctx, "gpt-5.5", nil, nil, terminal, nil) + if got := gjson.GetBytes(nonStreamOut, "choices.0.finish_reason").String(); got != "length" { + t.Fatalf("non-stream finish_reason = %q, want length; payload=%s", got, nonStreamOut) + } +} + func TestConvertCodexResponseToOpenAI_StreamSetsModelFromResponseCreated(t *testing.T) { ctx := context.Background() var param any diff --git a/internal/translator/codex/openai/responses/codex_openai-responses_response.go b/internal/translator/codex/openai/responses/codex_openai-responses_response.go index 968c11631..83ef2834c 100644 --- a/internal/translator/codex/openai/responses/codex_openai-responses_response.go +++ b/internal/translator/codex/openai/responses/codex_openai-responses_response.go @@ -25,8 +25,9 @@ func ConvertCodexResponseToOpenAIResponses(_ context.Context, _ string, _, _, ra // from a non-streaming OpenAI Chat Completions response. func ConvertCodexResponseToOpenAIResponsesNonStream(_ context.Context, _ string, _, _, rawJSON []byte, _ *any) []byte { rootResult := gjson.ParseBytes(rawJSON) - // Verify this is a response.completed event - if rootResult.Get("type").String() != "response.completed" { + // Verify this is a terminal response event. + responseType := rootResult.Get("type").String() + if responseType != "response.completed" && responseType != "response.incomplete" { return []byte{} } responseResult := rootResult.Get("response") diff --git a/internal/translator/codex/openai/responses/codex_openai-responses_response_test.go b/internal/translator/codex/openai/responses/codex_openai-responses_response_test.go new file mode 100644 index 000000000..80b349365 --- /dev/null +++ b/internal/translator/codex/openai/responses/codex_openai-responses_response_test.go @@ -0,0 +1,21 @@ +package responses + +import ( + "context" + "testing" + + "github.com/tidwall/gjson" +) + +func TestConvertCodexResponseToOpenAIResponsesNonStreamIncomplete(t *testing.T) { + raw := []byte(`{"type":"response.incomplete","response":{"id":"resp_1","status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[],"usage":{"input_tokens":1,"output_tokens":2,"total_tokens":3}}}`) + + out := ConvertCodexResponseToOpenAIResponsesNonStream(context.Background(), "gpt-5.5", nil, nil, raw, nil) + + if got := gjson.GetBytes(out, "status").String(); got != "incomplete" { + t.Fatalf("status = %q, want incomplete; payload=%s", got, out) + } + if got := gjson.GetBytes(out, "incomplete_details.reason").String(); got != "max_output_tokens" { + t.Fatalf("incomplete reason = %q, want max_output_tokens; payload=%s", got, out) + } +} diff --git a/sdk/cliproxy/auth/conductor.go b/sdk/cliproxy/auth/conductor.go index 8164749bd..fedf49518 100644 --- a/sdk/cliproxy/auth/conductor.go +++ b/sdk/cliproxy/auth/conductor.go @@ -1785,10 +1785,7 @@ func (m *Manager) wrapStreamResult(ctx context.Context, auth *Auth, provider, re emit := func(chunk cliproxyexecutor.StreamChunk) bool { if chunk.Err != nil && !failed { failed = true - rerr := &Error{Message: chunk.Err.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](chunk.Err); ok && se != nil { - rerr.HTTPStatus = se.StatusCode() - } + rerr := resultErrorFromError(chunk.Err) m.MarkResult(ctx, Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr}) } if !forward { @@ -1885,10 +1882,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi } } if errStream != nil { - rerr := &Error{Message: errStream.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errStream); ok && se != nil { - rerr.HTTPStatus = se.StatusCode() - } + rerr := resultErrorFromError(errStream) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr} result.RetryAfter = retryAfterFromError(errStream) m.MarkResult(ctx, result) @@ -1924,10 +1918,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi } if bootstrapErr != nil { if isRequestInvalidError(bootstrapErr) { - rerr := &Error{Message: bootstrapErr.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](bootstrapErr); ok && se != nil { - rerr.HTTPStatus = se.StatusCode() - } + rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr} result.RetryAfter = retryAfterFromError(bootstrapErr) m.MarkResult(ctx, result) @@ -1935,10 +1926,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi return nil, bootstrapErr } if idx < len(execModels)-1 { - rerr := &Error{Message: bootstrapErr.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](bootstrapErr); ok && se != nil { - rerr.HTTPStatus = se.StatusCode() - } + rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr} result.RetryAfter = retryAfterFromError(bootstrapErr) m.MarkResult(ctx, result) @@ -1946,10 +1934,7 @@ func (m *Manager) executeStreamWithModelPool(ctx context.Context, executor Provi lastErr = bootstrapErr continue } - rerr := &Error{Message: bootstrapErr.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](bootstrapErr); ok && se != nil { - rerr.HTTPStatus = se.StatusCode() - } + rerr := resultErrorFromError(bootstrapErr) result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: false, Error: rerr} result.RetryAfter = retryAfterFromError(bootstrapErr) m.MarkResult(ctx, result) @@ -2597,10 +2582,7 @@ func (m *Manager) executeMixedOnce(ctx context.Context, providers []string, req var errPrepare error auth, errPrepare = m.prepareRequestAuth(execCtx, executor, auth) if errPrepare != nil { - result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: &Error{Message: errPrepare.Error()}} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errPrepare); ok && se != nil { - result.Error.HTTPStatus = se.StatusCode() - } + result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare)} m.MarkResult(execCtx, result) lastErr = errPrepare continue @@ -2634,10 +2616,7 @@ func (m *Manager) executeMixedOnce(ctx context.Context, providers []string, req } result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: errExec == nil} if errExec != nil { - result.Error = &Error{Message: errExec.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errExec); ok && se != nil { - result.Error.HTTPStatus = se.StatusCode() - } + result.Error = resultErrorFromError(errExec) if ra := retryAfterFromError(errExec); ra != nil { result.RetryAfter = ra } @@ -2716,10 +2695,7 @@ func (m *Manager) executeCountMixedOnce(ctx context.Context, providers []string, var errPrepare error auth, errPrepare = m.prepareRequestAuth(execCtx, executor, auth) if errPrepare != nil { - result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: &Error{Message: errPrepare.Error()}} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errPrepare); ok && se != nil { - result.Error.HTTPStatus = se.StatusCode() - } + result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare)} m.MarkResult(execCtx, result) lastErr = errPrepare continue @@ -2753,10 +2729,7 @@ func (m *Manager) executeCountMixedOnce(ctx context.Context, providers []string, } result := Result{AuthID: auth.ID, Provider: provider, Model: resultModel, Success: errExec == nil} if errExec != nil { - result.Error = &Error{Message: errExec.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errExec); ok && se != nil { - result.Error.HTTPStatus = se.StatusCode() - } + result.Error = resultErrorFromError(errExec) if ra := retryAfterFromError(errExec); ra != nil { result.RetryAfter = ra } @@ -2833,10 +2806,7 @@ func (m *Manager) executeStreamMixedOnce(ctx context.Context, providers []string var errPrepare error auth, errPrepare = m.prepareRequestAuth(execCtx, executor, auth) if errPrepare != nil { - result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: &Error{Message: errPrepare.Error()}} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errPrepare); ok && se != nil { - result.Error.HTTPStatus = se.StatusCode() - } + result := Result{AuthID: auth.ID, Provider: provider, Model: routeModel, Success: false, Error: resultErrorFromError(errPrepare)} m.MarkResult(execCtx, result) lastErr = errPrepare continue @@ -3774,7 +3744,7 @@ func (m *Manager) MarkResult(ctx context.Context, result Result) { } } else { if result.Model != "" { - if !isRequestScopedNotFoundResultError(result.Error) { + if !isRequestScopedResultError(result.Error) { disableCooling := m.cooldownDisabledForAuth(auth) state := ensureModelState(auth, result.Model) state.Unavailable = true @@ -4110,6 +4080,28 @@ func statusCodeFromError(err error) int { return 0 } +func isRequestScopedError(err error) bool { + if err == nil { + return false + } + requestErr, ok := errors.AsType[cliproxyexecutor.RequestScopedError](err) + return ok && requestErr != nil && requestErr.IsRequestScoped() +} + +func resultErrorFromError(err error) *Error { + if err == nil { + return nil + } + resultErr := &Error{ + Message: err.Error(), + HTTPStatus: statusCodeFromError(err), + } + if isRequestScopedError(err) || isRequestInvalidError(err) { + resultErr.Code = requestScopedErrorCode + } + return resultErr +} + func isUnauthorizedError(err error) bool { if err == nil { return false @@ -4296,6 +4288,10 @@ func isRequestScopedNotFoundResultError(err *Error) bool { return isRequestScopedNotFoundMessage(err.Message) } +func isRequestScopedResultError(err *Error) bool { + return err != nil && (err.IsRequestScoped() || isRequestScopedNotFoundResultError(err)) +} + // isRequestInvalidError returns true if the error represents a client request // error that should not be retried. Specifically, it treats 400 responses with // "invalid_request_error", request-scoped 404 item misses caused by `store=false`, @@ -4306,6 +4302,9 @@ func isRequestInvalidError(err error) bool { if err == nil { return false } + if isRequestScopedError(err) { + return true + } if isCloudflareChallengeError(err) { return false } @@ -4320,6 +4319,7 @@ func isRequestInvalidError(err error) bool { case http.StatusBadRequest: msg := err.Error() return strings.Contains(msg, "invalid_request_error") || + strings.Contains(msg, "bad_request_error") || strings.Contains(msg, "INVALID_ARGUMENT") || strings.Contains(msg, "FAILED_PRECONDITION") case http.StatusNotFound: @@ -4339,7 +4339,7 @@ func applyAuthFailureState(auth *Auth, resultErr *Error, retryAfter *time.Durati if auth == nil { return } - if isRequestScopedNotFoundResultError(resultErr) { + if isRequestScopedResultError(resultErr) { return } auth.Unavailable = true @@ -5498,10 +5498,7 @@ func (m *Manager) tryAntigravityCreditsExecute(ctx context.Context, req cliproxy resp, errExec := c.executor.Execute(creditsCtx, c.auth, execReq, creditsOpts) result := Result{AuthID: c.auth.ID, Provider: c.provider, Model: resultModel, Success: errExec == nil} if errExec != nil { - result.Error = &Error{Message: errExec.Error()} - if se, ok := errors.AsType[cliproxyexecutor.StatusError](errExec); ok && se != nil { - result.Error.HTTPStatus = se.StatusCode() - } + result.Error = resultErrorFromError(errExec) if ra := retryAfterFromError(errExec); ra != nil { result.RetryAfter = ra } diff --git a/sdk/cliproxy/auth/conductor_overrides_test.go b/sdk/cliproxy/auth/conductor_overrides_test.go index 3123e32d4..1cb0b2490 100644 --- a/sdk/cliproxy/auth/conductor_overrides_test.go +++ b/sdk/cliproxy/auth/conductor_overrides_test.go @@ -230,6 +230,29 @@ type retryAfterStatusError struct { retryAfter time.Duration } +type requestScopedStatusError struct { + status int + message string +} + +func (e *requestScopedStatusError) Error() string { + if e == nil { + return "" + } + return e.message +} + +func (e *requestScopedStatusError) StatusCode() int { + if e == nil { + return 0 + } + return e.status +} + +func (e *requestScopedStatusError) IsRequestScoped() bool { + return e != nil +} + func (e *retryAfterStatusError) Error() string { if e == nil { return "" @@ -1104,6 +1127,110 @@ func TestManager_Execute_DisableCooling_RetriesAfter429RetryAfter(t *testing.T) } } +func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth(t *testing.T) { + incompleteErr := &requestScopedStatusError{ + status: http.StatusRequestTimeout, + message: "stream error: stream disconnected before completion: stream closed before response.completed", + } + invalidRequestErr := &Error{ + HTTPStatus: http.StatusBadRequest, + Message: `{"error":{"type":"invalid_request_error","code":"invalid_value","message":"Invalid input."}}`, + } + badRequestErr := &Error{ + HTTPStatus: http.StatusBadRequest, + Message: `{"error":{"type":"bad_request_error","code":"invalid_value","message":"Bad input."}}`, + } + tests := []struct { + name string + stream bool + err error + wantStatus int + }{ + {name: "non-streaming incomplete", err: incompleteErr, wantStatus: http.StatusRequestTimeout}, + {name: "streaming incomplete", stream: true, err: incompleteErr, wantStatus: http.StatusRequestTimeout}, + {name: "non-streaming invalid request", err: invalidRequestErr, wantStatus: http.StatusBadRequest}, + {name: "streaming invalid request", stream: true, err: invalidRequestErr, wantStatus: http.StatusBadRequest}, + {name: "non-streaming bad request", err: badRequestErr, wantStatus: http.StatusBadRequest}, + {name: "streaming bad request", stream: true, err: badRequestErr, wantStatus: http.StatusBadRequest}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + m := NewManager(nil, nil, nil) + m.SetRetryConfig(2, 30*time.Second, 0) + + executor := &authFallbackExecutor{id: "codex"} + if tc.stream { + executor.streamFirstErrors = map[string]error{"aa-bad-auth": tc.err} + } else { + executor.executeErrors = map[string]error{"aa-bad-auth": tc.err} + } + m.RegisterExecutor(executor) + + model := "gpt-5.5" + badAuth := &Auth{ID: "aa-bad-auth", Provider: "codex"} + goodAuth := &Auth{ID: "bb-good-auth", Provider: "codex"} + + reg := registry.GetGlobalRegistry() + reg.RegisterClient(badAuth.ID, badAuth.Provider, []*registry.ModelInfo{{ID: model}}) + reg.RegisterClient(goodAuth.ID, goodAuth.Provider, []*registry.ModelInfo{{ID: model}}) + t.Cleanup(func() { + reg.UnregisterClient(badAuth.ID) + reg.UnregisterClient(goodAuth.ID) + }) + + if _, errRegister := m.Register(context.Background(), badAuth); errRegister != nil { + t.Fatalf("register bad auth: %v", errRegister) + } + if _, errRegister := m.Register(context.Background(), goodAuth); errRegister != nil { + t.Fatalf("register good auth: %v", errRegister) + } + + var errExecute error + if tc.stream { + result, errStream := m.ExecuteStream(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{Stream: true}) + if result != nil { + for range result.Chunks { + } + } + errExecute = errStream + } else { + _, errExecute = m.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}) + } + if errExecute == nil { + t.Fatal("expected request-scoped stream error") + } + if got := statusCodeFromError(errExecute); got != tc.wantStatus { + t.Fatalf("status = %d, want %d", got, tc.wantStatus) + } + + var calls []string + if tc.stream { + calls = executor.StreamCalls() + } else { + calls = executor.ExecuteCalls() + } + if len(calls) != 1 || calls[0] != badAuth.ID { + t.Fatalf("credential calls = %v, want [%s]", calls, badAuth.ID) + } + + updatedBad, ok := m.GetByID(badAuth.ID) + if !ok || updatedBad == nil { + t.Fatal("expected bad auth to remain registered") + } + if updatedBad.Unavailable { + t.Fatal("expected request-scoped error to keep auth available") + } + if !updatedBad.NextRetryAfter.IsZero() { + t.Fatalf("expected auth cooldown to remain unset, got %v", updatedBad.NextRetryAfter) + } + if state := updatedBad.ModelStates[model]; state != nil { + t.Fatalf("expected request-scoped error to avoid model cooldown state, got %#v", state) + } + }) + } +} + func TestManager_MarkResult_RequestScopedNotFoundDoesNotCooldownAuth(t *testing.T) { m := NewManager(nil, nil, nil) diff --git a/sdk/cliproxy/auth/errors.go b/sdk/cliproxy/auth/errors.go index 72bca1fcf..85529cb51 100644 --- a/sdk/cliproxy/auth/errors.go +++ b/sdk/cliproxy/auth/errors.go @@ -1,5 +1,7 @@ package auth +const requestScopedErrorCode = "request_scoped" + // Error describes an authentication related failure in a provider agnostic format. type Error struct { // Code is a short machine readable identifier. @@ -30,3 +32,9 @@ func (e *Error) StatusCode() int { } return e.HTTPStatus } + +// IsRequestScoped reports whether the failure is tied to the current request +// rather than the selected credential. +func (e *Error) IsRequestScoped() bool { + return e != nil && e.Code == requestScopedErrorCode +} diff --git a/sdk/cliproxy/auth/errors_compat_test.go b/sdk/cliproxy/auth/errors_compat_test.go new file mode 100644 index 000000000..db058ab67 --- /dev/null +++ b/sdk/cliproxy/auth/errors_compat_test.go @@ -0,0 +1,16 @@ +package auth_test + +import ( + "net/http" + "testing" + + cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" +) + +func TestErrorLegacyUnkeyedLiteralCompatibility(t *testing.T) { + err := cliproxyauth.Error{"code", "message", false, http.StatusRequestTimeout} + + if err.Code != "code" || err.Message != "message" || err.Retryable || err.HTTPStatus != http.StatusRequestTimeout { + t.Fatalf("unexpected error fields: %#v", err) + } +} diff --git a/sdk/cliproxy/executor/types.go b/sdk/cliproxy/executor/types.go index c9e9a346a..d114e33d8 100644 --- a/sdk/cliproxy/executor/types.go +++ b/sdk/cliproxy/executor/types.go @@ -152,3 +152,11 @@ type StatusError interface { error StatusCode() int } + +// RequestScopedError identifies a failure tied to the current request rather +// than the selected credential. Auth managers should not retry these errors +// across credentials or change credential availability because of them. +type RequestScopedError interface { + error + IsRequestScoped() bool +}