package executor import ( "bytes" "context" "io" "net/http" "net/http/httptest" "strings" "testing" "time" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" _ "github.com/router-for-me/CLIProxyAPI/v7/internal/translator" 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" "github.com/tidwall/gjson") func TestCodexExecutorExecute_EmptyStreamCompletionOutputUsesOutputItemDone(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.output_item.done\",\"item\":{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"ok\"}]},\"output_index\":0}\n")) _, _ = w.Write([]byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"created_at\":1775555723,\"status\":\"completed\",\"model\":\"gpt-5.4-mini-2026-03-17\",\"output\":[],\"usage\":{\"input_tokens\":8,\"output_tokens\":28,\"total_tokens\":36}}}\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.4-mini", Payload: []byte(`{"model":"gpt-5.4-mini","messages":[{"role":"user","content":"Say ok"}]}`), }, cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai"), Stream: false, }) if err != nil { t.Fatalf("Execute error: %v", err) } gotContent := gjson.GetBytes(resp.Payload, "choices.0.message.content").String() if gotContent != "ok" { t.Fatalf("choices.0.message.content = %q, want %q; payload=%s", gotContent, "ok", string(resp.Payload)) }} func TestCodexExecutorExecuteSurfacesTerminalStreamError(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("event: response.created\n")) _, _ = w.Write([]byte(`data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.5"}}` + "\n\n")) _, _ = w.Write([]byte("event: error\n")) _, _ = w.Write([]byte(`data: {"type":"error","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again.","param":"input"},"sequence_number":2}` + "\n\n")) _, _ = w.Write([]byte("event: response.failed\n")) _, _ = w.Write([]byte(`data: {"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."}}}` + "\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 terminal stream error, got nil") } if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) } assertCodexErrorCode(t, err.Error(), "invalid_request_error", "context_too_large") if !strings.Contains(err.Error(), "Your input exceeds the context window") { t.Fatalf("error message missing upstream context text: %v", err) }} 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") _, _ = w.Write([]byte("event: response.created\n")) _, _ = w.Write([]byte(`data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.5"}}` + "\n\n")) _, _ = w.Write([]byte("event: error\n")) _, _ = w.Write([]byte(`data: {"type":"error","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again.","param":"input"},"sequence_number":2}` + "\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 break } } if streamErr == nil { t.Fatal("missing stream terminal error") } if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, streamErr) } assertCodexErrorCode(t, streamErr.Error(), "invalid_request_error", "context_too_large")} func TestCodexTerminalStreamContextLengthErrFromResponseFailed(t *testing.T) { err, ok := codexTerminalStreamContextLengthErr([]byte(`{"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."}}}`)) if !ok { t.Fatal("expected context length terminal error") } if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) } assertCodexErrorCode(t, err.Error(), "invalid_request_error", "context_too_large")} func TestCodexTerminalStreamContextLengthErrFromTopLevelError(t *testing.T) { err, ok := codexTerminalStreamContextLengthErr([]byte(`{"type":"error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again.","sequence_number":2}`)) if !ok { t.Fatal("expected top-level context length terminal error") } if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) } assertCodexErrorCode(t, err.Error(), "invalid_request_error", "context_too_large") if !strings.Contains(err.Error(), "Your input exceeds the context window") { t.Fatalf("error message missing upstream context text: %v", err) }} func TestCodexTerminalStreamContextLengthErrIgnoresOtherTerminalErrors(t *testing.T) { _, ok := codexTerminalStreamContextLengthErr([]byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"Rate limit reached."}}`)) if ok { t.Fatal("rate limit terminal error should not be handled by context length fix") }} func TestCodexTerminalStreamErrIgnoresRateLimitTerminalErrors(t *testing.T) { _, _, ok := codexTerminalStreamErr([]byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"Rate limit reached."}}`)) if ok { t.Fatal("rate limit terminal error should not be handled by replay terminal error path") }} 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) } }) }}Lines 1–500 of 617. Load more
package executor import ( "bytes" "context" "io" "net/http" "net/http/httptest" "strings" "testing" "time" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" _ "github.com/router-for-me/CLIProxyAPI/v7/internal/translator" 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" "github.com/tidwall/gjson" ) func TestCodexExecutorExecute_EmptyStreamCompletionOutputUsesOutputItemDone(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.output_item.done\",\"item\":{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"ok\"}]},\"output_index\":0}\n")) _, _ = w.Write([]byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"created_at\":1775555723,\"status\":\"completed\",\"model\":\"gpt-5.4-mini-2026-03-17\",\"output\":[],\"usage\":{\"input_tokens\":8,\"output_tokens\":28,\"total_tokens\":36}}}\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.4-mini", Payload: []byte(`{"model":"gpt-5.4-mini","messages":[{"role":"user","content":"Say ok"}]}`), }, cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai"), Stream: false, }) if err != nil { t.Fatalf("Execute error: %v", err) } gotContent := gjson.GetBytes(resp.Payload, "choices.0.message.content").String() if gotContent != "ok" { t.Fatalf("choices.0.message.content = %q, want %q; payload=%s", gotContent, "ok", string(resp.Payload)) } } func TestCodexExecutorExecuteSurfacesTerminalStreamError(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("event: response.created\n")) _, _ = w.Write([]byte(`data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.5"}}` + "\n\n")) _, _ = w.Write([]byte("event: error\n")) _, _ = w.Write([]byte(`data: {"type":"error","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again.","param":"input"},"sequence_number":2}` + "\n\n")) _, _ = w.Write([]byte("event: response.failed\n")) _, _ = w.Write([]byte(`data: {"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."}}}` + "\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 terminal stream error, got nil") } if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) } assertCodexErrorCode(t, err.Error(), "invalid_request_error", "context_too_large") if !strings.Contains(err.Error(), "Your input exceeds the context window") { t.Fatalf("error message missing upstream context text: %v", err) } } 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") _, _ = w.Write([]byte("event: response.created\n")) _, _ = w.Write([]byte(`data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5.5"}}` + "\n\n")) _, _ = w.Write([]byte("event: error\n")) _, _ = w.Write([]byte(`data: {"type":"error","error":{"type":"invalid_request_error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again.","param":"input"},"sequence_number":2}` + "\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 break } } if streamErr == nil { t.Fatal("missing stream terminal error") } if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, streamErr) } assertCodexErrorCode(t, streamErr.Error(), "invalid_request_error", "context_too_large") } func TestCodexTerminalStreamContextLengthErrFromResponseFailed(t *testing.T) { err, ok := codexTerminalStreamContextLengthErr([]byte(`{"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."}}}`)) if !ok { t.Fatal("expected context length terminal error") } if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) } assertCodexErrorCode(t, err.Error(), "invalid_request_error", "context_too_large") } func TestCodexTerminalStreamContextLengthErrFromTopLevelError(t *testing.T) { err, ok := codexTerminalStreamContextLengthErr([]byte(`{"type":"error","code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again.","sequence_number":2}`)) if !ok { t.Fatal("expected top-level context length terminal error") } if got := statusCodeFromTestError(t, err); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d; err=%v", got, http.StatusBadRequest, err) } assertCodexErrorCode(t, err.Error(), "invalid_request_error", "context_too_large") if !strings.Contains(err.Error(), "Your input exceeds the context window") { t.Fatalf("error message missing upstream context text: %v", err) } } func TestCodexTerminalStreamContextLengthErrIgnoresOtherTerminalErrors(t *testing.T) { _, ok := codexTerminalStreamContextLengthErr([]byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"Rate limit reached."}}`)) if ok { t.Fatal("rate limit terminal error should not be handled by context length fix") } } func TestCodexTerminalStreamErrIgnoresRateLimitTerminalErrors(t *testing.T) { _, _, ok := codexTerminalStreamErr([]byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded","message":"Rate limit reached."}}`)) if ok { t.Fatal("rate limit terminal error should not be handled by replay terminal error path") } } 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) } }) } }