package executor import ( "bytes" "context" "fmt" "net/http" "net/http/httptest" "strings" "sync" "testing" "time" "github.com/gorilla/websocket" "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") const ( codexOverloadEvent = `{"type":"error","error":{"type":"service_unavailable_error","code":"server_is_overloaded","message":"Our servers are currently overloaded. Please try again later.","param":null},"sequence_number":2}` codexCapacityEvent = `{"type":"error","error":{"message":"Selected model is at capacity. Please try a different model."},"sequence_number":2}` codexInvalidEvent = `{"type":"error","error":{"type":"invalid_request_error","code":"invalid_value","message":"Invalid input."},"sequence_number":2}` codexCreatedEvent = `{"type":"response.created","response":{"id":"resp_1","model":"gpt-5.6-terra"}}` codexInProgressEvent = `{"type":"response.in_progress","response":{"id":"resp_1"}}` codexOutputAddedEvent = `{"type":"response.output_item.added","item":{"id":"msg_1","type":"message","role":"assistant","content":[]},"output_index":0}` codexKeepaliveEvent = `{"type":"keepalive","sequence_number":1}` codexOutputDeltaEvent = `{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"hi"}` codexCompletedEventBody = `{"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"hello"}]}],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}`) func codexBufferingConfig(enabled bool) *config.Config { return &config.Config{Codex: config.CodexConfig{StreamBootstrapBuffering: enabled}}} func codexBufferingConfigWithTimeout(enabled bool, timeout string) *config.Config { return &config.Config{Codex: config.CodexConfig{StreamBootstrapBuffering: enabled, StreamBootstrapTimeout: timeout}}} func codexTestAuth(baseURL string) *cliproxyauth.Auth { return &cliproxyauth.Auth{Attributes: map[string]string{"base_url": baseURL, "api_key": "test"}}} func codexTestRequest() (cliproxyexecutor.Request, cliproxyexecutor.Options) { return cliproxyexecutor.Request{ Model: "gpt-5.6-terra", Payload: []byte(`{"model":"gpt-5.6-terra","input":"hello"}`), }, cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai-response"), Stream: true, }} // codexSSEServer streams the supplied event payloads as an HTTP 200 SSE response.func codexSSEServer(events ...string) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") for _, event := range events { eventType := "message" if parsed := strings.SplitN(event, `"type":"`, 2); len(parsed) == 2 { eventType = strings.SplitN(parsed[1], `"`, 2)[0] } _, _ = w.Write([]byte("event: " + eventType + "\n")) _, _ = w.Write([]byte("data: " + event + "\n\n")) } }))} // codexWebsocketServer echoes the supplied frames after receiving the client request frame.func codexWebsocketServer(t *testing.T, frames ...string) *httptest.Server { t.Helper() upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) if err != nil { t.Errorf("upgrade websocket: %v", err) return } defer func() { _ = conn.Close() }() if _, _, errRead := conn.ReadMessage(); errRead != nil { t.Errorf("read websocket message: %v", errRead) return } for _, frame := range frames { _ = conn.WriteMessage(websocket.TextMessage, []byte(frame)) } }))} func codexWebsocketRequest() (cliproxyexecutor.Request, cliproxyexecutor.Options) { return cliproxyexecutor.Request{ Model: "gpt-5.6-terra", Payload: []byte(`{"model":"gpt-5.6-terra","input":[{"type":"message","role":"user","content":"hello"}]}`), }, cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai-response"), }} // drainChunks collects every payload and the first error from a stream result.func drainChunks(result *cliproxyexecutor.StreamResult) (string, error) { var payloads [][]byte var streamErr error for chunk := range result.Chunks { if chunk.Err != nil { if streamErr == nil { streamErr = chunk.Err } continue } payloads = append(payloads, chunk.Payload) } return string(bytes.Join(payloads, []byte("\n"))), streamErr} // An overload rejection smuggled into an HTTP 200 stream must fail the whole attempt before any// downstream chunk escapes, so the conductor can retry on another credential. A nil StreamResult// is the invariant: with no channel there is no way for the buffered handshake to reach the client.func TestCodexExecutor_BootstrapBuffering_OverloadFailsAttemptWithoutLeakingHandshake(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOverloadEvent) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err == nil { t.Fatal("expected ExecuteStream to fail the attempt on an overload rejection") } if result != nil { t.Fatal("expected nil result so no buffered handshake chunk can reach the client") } if got := statusCodeFromTestError(t, err); got != http.StatusServiceUnavailable { t.Fatalf("status code = %d, want %d (upstream hides 503 behind HTTP 200)", got, http.StatusServiceUnavailable) }} func TestCodexExecutor_BootstrapBuffering_CapacityFailsAttemptWithoutLeakingHandshake(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexCapacityEvent) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err == nil { t.Fatal("expected ExecuteStream to fail the attempt on a capacity rejection") } if result != nil { t.Fatal("expected nil result so no buffered handshake chunk can reach the client") } if got := statusCodeFromTestError(t, err); got != http.StatusTooManyRequests { t.Fatalf("status code = %d, want %d", got, http.StatusTooManyRequests) }} // A non-overload terminal failure must keep the original in-stream delivery semantics: the// buffered handshake is flushed first and the error arrives as a stream chunk, so the conductor// sees a committed stream and does not burn another credential on a request-level fault.func TestCodexExecutor_BootstrapBuffering_NonOverloadStaysInStream(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexInvalidEvent) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("non-overload failure must not fail the attempt synchronously: %v", err) } if result == nil { t.Fatal("expected a stream result for in-stream error delivery") } combined, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the invalid-request failure to arrive as an in-stream chunk error") } if !strings.Contains(combined, `"type":"response.created"`) { t.Fatalf("buffered handshake must be flushed before the in-stream error: %s", combined) } if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d", got, http.StatusBadRequest) }} // Once the buffer limit is exceeded the stream is released and overload probing stops, which// bounds how long the downstream response headers can stay uncommitted.func TestCodexExecutor_BootstrapBuffering_BufferLimitReleasesStream(t *testing.T) { events := make([]string, 0, codexBootstrapMaxBufferedFrames+2) for i := 0; i < codexBootstrapMaxBufferedFrames+1; i++ { events = append(events, fmt.Sprintf(`{"type":"response.in_progress","response":{"id":"resp_%d"}}`, i)) } events = append(events, codexOverloadEvent) server := codexSSEServer(events...) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("expected the stream to be released once the buffer limit is hit: %v", err) } if result == nil { t.Fatal("expected a stream result after the buffer limit released the stream") } _, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the overload error to be delivered in-stream after the limit was hit") }} // codexWebsocketRawServer writes the supplied frames and then closes the connection, so a test can// drive an exact frame sequence rather than the canonical handshake.func codexWebsocketRawServer(t *testing.T, frames []string, tail ...string) *httptest.Server { t.Helper() upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) if err != nil { return } defer func() { _ = conn.Close() }() if _, _, errRead := conn.ReadMessage(); errRead != nil { return } for _, frame := range frames { if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(frame)); errWrite != nil { return } } for _, frame := range tail { _ = conn.WriteMessage(websocket.TextMessage, []byte(frame)) } }))} // The websocket frame budget must bound allow-listed frames too, not only the skippable ones.func TestCodexWebsocketsExecutor_BootstrapBuffering_FrameBudgetReleasesStream(t *testing.T) { frames := make([]string, 0, codexBootstrapMaxBufferedFrames+1) for i := 0; i < codexBootstrapMaxBufferedFrames+1; i++ { frames = append(frames, codexInProgressEvent) } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("the frame budget must release the stream before the overload arrives: %v", err) } if result == nil { t.Fatal("expected a stream result once the frame budget released the stream") } drainChunks(result)} // The websocket budget is pinned on both sides, the same way the SSE one is.func TestCodexWebsocketsExecutor_BootstrapBuffering_BudgetBoundaryIsExact(t *testing.T) { send := func(n int) error { frames := make([]string, n) for i := range frames { frames[i] = codexInProgressEvent } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if result != nil { drainChunks(result) } return err } if err := send(codexBootstrapMaxBufferedFrames - 1); err == nil { t.Fatalf("%d messages are inside the budget and must still fail over", codexBootstrapMaxBufferedFrames-1) } // The last message the window still admits. Without this case a budget one frame too small // looks identical, since the overload is recognised before the window is consulted and so the // max-1 case fails over either way. if err := send(codexBootstrapMaxBufferedFrames); err == nil { t.Fatalf("%d messages exactly fill the budget and must still fail over", codexBootstrapMaxBufferedFrames) } if err := send(codexBootstrapMaxBufferedFrames + 1); err != nil { t.Fatalf("%d messages exceed the budget and must release: %v", codexBootstrapMaxBufferedFrames+1, err) }} // The websocket byte budget has the same two properties as the SSE one: a message larger than the// cap is refused admission rather than taken on the strength of an empty buffer, and the budget is// seeded from the upstream message so it still advances for a downstream that renders nothing.func TestCodexWebsocketsExecutor_BootstrapBuffering_ByteCapOrderingAndSeed(t *testing.T) { t.Run("oversized message is not admitted", func(t *testing.T) { oversized := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("q", codexBootstrapMaxBufferedBytes*2) + `"}}` server := codexWebsocketRawServer(t, []string{oversized}, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("a message larger than the cap must be released, not admitted: %v", err) } if result == nil { t.Fatal("expected a stream result") } drainChunks(result) }) t.Run("budget counts the upstream message", func(t *testing.T) { frame := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("w", codexBootstrapMaxBufferedBytes/8) + `"}}` frames := make([]string, 10) for i := range frames { frames[i] = frame } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, _ := codexWebsocketRequest() opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai")} result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("ten messages of cap/8 must exhaust the byte budget: %v", err) } if result == nil { t.Fatal("expected a stream result") } drainChunks(result) })} // The frame that trips the budget must still be delivered. Releasing the window is a decision about// what to do with later frames, not a licence to drop the one in hand: on a slow turn that frame is// the first token of the answer.func TestCodexWebsocketsExecutor_BootstrapBuffering_BudgetFrameIsNotDropped(t *testing.T) { frames := make([]string, 0, codexBootstrapMaxBufferedFrames+2) for i := 0; i < codexBootstrapMaxBufferedFrames; i++ { frames = append(frames, codexInProgressEvent) } frames = append(frames, `{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"FIRSTTOKEN"}`) server := codexWebsocketRawServer(t, frames, codexCompletedEventBody) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("unexpected ExecuteStream error: %v", err) } if result == nil { t.Fatal("expected a stream result") } combined, _ := drainChunks(result) if !strings.Contains(combined, "FIRSTTOKEN") { t.Fatalf("the frame that tripped the budget was dropped: %s", combined) }} // The websocket bound must be enforced where the frame is counted. A peer that sends only frames the// bootstrap loop skips never reaches the buffering branch, so a release decided there would never// run and the downstream headers would stay uncommitted for as long as the peer keeps sending.func TestCodexWebsocketsExecutor_BootstrapBuffering_SkippedFramesExhaustTheWindow(t *testing.T) { cases := []struct { name string frame string }{ {"whitespace only frames", " \n\t "}, {"empty frames", ""}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { frames := make([]string, codexBootstrapMaxBufferedFrames+2) for i := range frames { frames[i] = tc.frame } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("the window must release before the overload arrives, got failover: %v", err) } if result == nil { t.Fatal("expected a stream result once the window was exhausted") } drainChunks(result) }) }} // The window only protects anything if it is small. Asserting it relative to itself, as the framing// tests below must, cannot catch a budget that has grown to a useless size.func TestCodexBootstrapBudgetsStaySmall(t *testing.T) { // Pinned exactly, because the framing tests below express their expectations as multiples of // these constants and so cannot notice the budgets growing. // TestCodexExecutor_BootstrapBuffering_HeartbeatArithmeticPerFraming states the resulting // heartbeat counts in absolute numbers, which is what config.example.yaml quotes to operators. if codexBootstrapMaxBufferedFrames != 48 { t.Fatalf("frame budget = %d, want 48; changing it changes how long the headers are held, so update config.example.yaml too", codexBootstrapMaxBufferedFrames) } if codexBootstrapMaxBufferedBytes != 1<<20 { t.Fatalf("byte budget = %d, want 1MiB", codexBootstrapMaxBufferedBytes) }} // In-stream delivery depends on the held frames rendering to at least one downstream chunk: the// conductor commits a stream only once it has seen a non-empty payload, so against a format that// renders the handshake as nothing there is nothing to commit and the attempt still fails over.func TestCodexExecutor_BootstrapBuffering_InStreamDeliveryNeedsRenderedFrames(t *testing.T) { incomplete := `{"type":"response.incomplete","response":{"id":"resp_1","output":[],"usage":{"input_tokens":1,"output_tokens":0,"total_tokens":1}}}` server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOutputAddedEvent, incomplete) defer server.Close() req, _ := codexTestRequest() opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai"), Stream: true} result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("the executor still returns a stream; it is the conductor that cannot commit it: %v", err) } if result == nil { t.Fatal("expected a stream result") } combined, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the failure to arrive as an in-stream chunk error") } if combined != "" { t.Fatalf("Chat Completions renders these frames as nothing, so no payload precedes the error: %q", combined) }} // An empty response.incomplete is not an overload rejection, so it keeps its in-stream delivery:// the buffered frames are flushed and the error arrives as a chunk rather than burning another// credential. The websocket executor already routes this condition that way; both transports agree// on it. A truncated stream is deliberately left alone - both transports fail that one over.func TestCodexExecutor_BootstrapBuffering_NonOverloadTerminalStaysInStream(t *testing.T) { cases := []struct{ name, tail string }{ {"empty incomplete", `{"type":"response.incomplete","response":{"id":"resp_1","output":[],"usage":{"input_tokens":1,"output_tokens":0,"total_tokens":1}}}`}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOutputAddedEvent, tc.tail) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("a non-overload terminal failure must not fail the attempt over: %v", err) } if result == nil { t.Fatal("expected a stream result carrying the buffered frames and the error") } combined, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the failure to arrive as an in-stream chunk error") } if !strings.Contains(combined, `"type":"response.created"`) { t.Fatalf("buffered frames must be flushed before the in-stream error: %s", combined) } }) }} // The cap is consulted before a frame is taken, not after. A frame that alone exceeds it is refused// admission and released, so the rejection that follows it immediately - with no line in between to// trigger a late release - is delivered in-stream rather than failing the attempt over.func TestCodexExecutor_BootstrapBuffering_OversizedFrameIsNotAdmitted(t *testing.T) { oversized := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("q", codexBootstrapMaxBufferedBytes*2) + `"}}` body := "data: " + oversized + "\n" + "data: " + codexOverloadEvent + "\n\n" server := codexSSERawServer(body) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("a frame larger than the cap must be released, not admitted: %v", err) } if result == nil { t.Fatal("expected a stream result once the oversized frame released the stream") } drainChunks(result)} // The byte budget is seeded from the upstream frame, not only from the chunks it translates into.// Against a Chat Completions downstream these frames render as zero chunks, so an accounting that// measured chunks alone would never advance and only the frame budget would remain - which these ten// frames stay well inside.func TestCodexExecutor_BootstrapBuffering_ByteCapCountsUpstreamFrames(t *testing.T) { frame := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("w", codexBootstrapMaxBufferedBytes/8) + `"}}` body := strings.Repeat("data: "+frame+"\n\n", 10) + "data: " + codexOverloadEvent + "\n\n" server := codexSSERawServer(body) defer server.Close() req, _ := codexTestRequest() opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai"), Stream: true} result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("ten frames of cap/8 must exhaust the byte budget: %v", err) } if result == nil { t.Fatal("expected a stream result once the byte budget released the stream") }Lines 1–500 of 1709. Load more
package executor import ( "bytes" "context" "fmt" "net/http" "net/http/httptest" "strings" "sync" "testing" "time" "github.com/gorilla/websocket" "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" ) const ( codexOverloadEvent = `{"type":"error","error":{"type":"service_unavailable_error","code":"server_is_overloaded","message":"Our servers are currently overloaded. Please try again later.","param":null},"sequence_number":2}` codexCapacityEvent = `{"type":"error","error":{"message":"Selected model is at capacity. Please try a different model."},"sequence_number":2}` codexInvalidEvent = `{"type":"error","error":{"type":"invalid_request_error","code":"invalid_value","message":"Invalid input."},"sequence_number":2}` codexCreatedEvent = `{"type":"response.created","response":{"id":"resp_1","model":"gpt-5.6-terra"}}` codexInProgressEvent = `{"type":"response.in_progress","response":{"id":"resp_1"}}` codexOutputAddedEvent = `{"type":"response.output_item.added","item":{"id":"msg_1","type":"message","role":"assistant","content":[]},"output_index":0}` codexKeepaliveEvent = `{"type":"keepalive","sequence_number":1}` codexOutputDeltaEvent = `{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"hi"}` codexCompletedEventBody = `{"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"hello"}]}],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}` ) func codexBufferingConfig(enabled bool) *config.Config { return &config.Config{Codex: config.CodexConfig{StreamBootstrapBuffering: enabled}} } func codexBufferingConfigWithTimeout(enabled bool, timeout string) *config.Config { return &config.Config{Codex: config.CodexConfig{StreamBootstrapBuffering: enabled, StreamBootstrapTimeout: timeout}} } func codexTestAuth(baseURL string) *cliproxyauth.Auth { return &cliproxyauth.Auth{Attributes: map[string]string{"base_url": baseURL, "api_key": "test"}} } func codexTestRequest() (cliproxyexecutor.Request, cliproxyexecutor.Options) { return cliproxyexecutor.Request{ Model: "gpt-5.6-terra", Payload: []byte(`{"model":"gpt-5.6-terra","input":"hello"}`), }, cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai-response"), Stream: true, } } // codexSSEServer streams the supplied event payloads as an HTTP 200 SSE response. func codexSSEServer(events ...string) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") for _, event := range events { eventType := "message" if parsed := strings.SplitN(event, `"type":"`, 2); len(parsed) == 2 { eventType = strings.SplitN(parsed[1], `"`, 2)[0] } _, _ = w.Write([]byte("event: " + eventType + "\n")) _, _ = w.Write([]byte("data: " + event + "\n\n")) } })) } // codexWebsocketServer echoes the supplied frames after receiving the client request frame. func codexWebsocketServer(t *testing.T, frames ...string) *httptest.Server { t.Helper() upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) if err != nil { t.Errorf("upgrade websocket: %v", err) return } defer func() { _ = conn.Close() }() if _, _, errRead := conn.ReadMessage(); errRead != nil { t.Errorf("read websocket message: %v", errRead) return } for _, frame := range frames { _ = conn.WriteMessage(websocket.TextMessage, []byte(frame)) } })) } func codexWebsocketRequest() (cliproxyexecutor.Request, cliproxyexecutor.Options) { return cliproxyexecutor.Request{ Model: "gpt-5.6-terra", Payload: []byte(`{"model":"gpt-5.6-terra","input":[{"type":"message","role":"user","content":"hello"}]}`), }, cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai-response"), } } // drainChunks collects every payload and the first error from a stream result. func drainChunks(result *cliproxyexecutor.StreamResult) (string, error) { var payloads [][]byte var streamErr error for chunk := range result.Chunks { if chunk.Err != nil { if streamErr == nil { streamErr = chunk.Err } continue } payloads = append(payloads, chunk.Payload) } return string(bytes.Join(payloads, []byte("\n"))), streamErr } // An overload rejection smuggled into an HTTP 200 stream must fail the whole attempt before any // downstream chunk escapes, so the conductor can retry on another credential. A nil StreamResult // is the invariant: with no channel there is no way for the buffered handshake to reach the client. func TestCodexExecutor_BootstrapBuffering_OverloadFailsAttemptWithoutLeakingHandshake(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOverloadEvent) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err == nil { t.Fatal("expected ExecuteStream to fail the attempt on an overload rejection") } if result != nil { t.Fatal("expected nil result so no buffered handshake chunk can reach the client") } if got := statusCodeFromTestError(t, err); got != http.StatusServiceUnavailable { t.Fatalf("status code = %d, want %d (upstream hides 503 behind HTTP 200)", got, http.StatusServiceUnavailable) } } func TestCodexExecutor_BootstrapBuffering_CapacityFailsAttemptWithoutLeakingHandshake(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexCapacityEvent) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err == nil { t.Fatal("expected ExecuteStream to fail the attempt on a capacity rejection") } if result != nil { t.Fatal("expected nil result so no buffered handshake chunk can reach the client") } if got := statusCodeFromTestError(t, err); got != http.StatusTooManyRequests { t.Fatalf("status code = %d, want %d", got, http.StatusTooManyRequests) } } // A non-overload terminal failure must keep the original in-stream delivery semantics: the // buffered handshake is flushed first and the error arrives as a stream chunk, so the conductor // sees a committed stream and does not burn another credential on a request-level fault. func TestCodexExecutor_BootstrapBuffering_NonOverloadStaysInStream(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexInvalidEvent) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("non-overload failure must not fail the attempt synchronously: %v", err) } if result == nil { t.Fatal("expected a stream result for in-stream error delivery") } combined, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the invalid-request failure to arrive as an in-stream chunk error") } if !strings.Contains(combined, `"type":"response.created"`) { t.Fatalf("buffered handshake must be flushed before the in-stream error: %s", combined) } if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadRequest { t.Fatalf("status code = %d, want %d", got, http.StatusBadRequest) } } // Once the buffer limit is exceeded the stream is released and overload probing stops, which // bounds how long the downstream response headers can stay uncommitted. func TestCodexExecutor_BootstrapBuffering_BufferLimitReleasesStream(t *testing.T) { events := make([]string, 0, codexBootstrapMaxBufferedFrames+2) for i := 0; i < codexBootstrapMaxBufferedFrames+1; i++ { events = append(events, fmt.Sprintf(`{"type":"response.in_progress","response":{"id":"resp_%d"}}`, i)) } events = append(events, codexOverloadEvent) server := codexSSEServer(events...) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("expected the stream to be released once the buffer limit is hit: %v", err) } if result == nil { t.Fatal("expected a stream result after the buffer limit released the stream") } _, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the overload error to be delivered in-stream after the limit was hit") } } // codexWebsocketRawServer writes the supplied frames and then closes the connection, so a test can // drive an exact frame sequence rather than the canonical handshake. func codexWebsocketRawServer(t *testing.T, frames []string, tail ...string) *httptest.Server { t.Helper() upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) if err != nil { return } defer func() { _ = conn.Close() }() if _, _, errRead := conn.ReadMessage(); errRead != nil { return } for _, frame := range frames { if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(frame)); errWrite != nil { return } } for _, frame := range tail { _ = conn.WriteMessage(websocket.TextMessage, []byte(frame)) } })) } // The websocket frame budget must bound allow-listed frames too, not only the skippable ones. func TestCodexWebsocketsExecutor_BootstrapBuffering_FrameBudgetReleasesStream(t *testing.T) { frames := make([]string, 0, codexBootstrapMaxBufferedFrames+1) for i := 0; i < codexBootstrapMaxBufferedFrames+1; i++ { frames = append(frames, codexInProgressEvent) } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("the frame budget must release the stream before the overload arrives: %v", err) } if result == nil { t.Fatal("expected a stream result once the frame budget released the stream") } drainChunks(result) } // The websocket budget is pinned on both sides, the same way the SSE one is. func TestCodexWebsocketsExecutor_BootstrapBuffering_BudgetBoundaryIsExact(t *testing.T) { send := func(n int) error { frames := make([]string, n) for i := range frames { frames[i] = codexInProgressEvent } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if result != nil { drainChunks(result) } return err } if err := send(codexBootstrapMaxBufferedFrames - 1); err == nil { t.Fatalf("%d messages are inside the budget and must still fail over", codexBootstrapMaxBufferedFrames-1) } // The last message the window still admits. Without this case a budget one frame too small // looks identical, since the overload is recognised before the window is consulted and so the // max-1 case fails over either way. if err := send(codexBootstrapMaxBufferedFrames); err == nil { t.Fatalf("%d messages exactly fill the budget and must still fail over", codexBootstrapMaxBufferedFrames) } if err := send(codexBootstrapMaxBufferedFrames + 1); err != nil { t.Fatalf("%d messages exceed the budget and must release: %v", codexBootstrapMaxBufferedFrames+1, err) } } // The websocket byte budget has the same two properties as the SSE one: a message larger than the // cap is refused admission rather than taken on the strength of an empty buffer, and the budget is // seeded from the upstream message so it still advances for a downstream that renders nothing. func TestCodexWebsocketsExecutor_BootstrapBuffering_ByteCapOrderingAndSeed(t *testing.T) { t.Run("oversized message is not admitted", func(t *testing.T) { oversized := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("q", codexBootstrapMaxBufferedBytes*2) + `"}}` server := codexWebsocketRawServer(t, []string{oversized}, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("a message larger than the cap must be released, not admitted: %v", err) } if result == nil { t.Fatal("expected a stream result") } drainChunks(result) }) t.Run("budget counts the upstream message", func(t *testing.T) { frame := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("w", codexBootstrapMaxBufferedBytes/8) + `"}}` frames := make([]string, 10) for i := range frames { frames[i] = frame } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, _ := codexWebsocketRequest() opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai")} result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("ten messages of cap/8 must exhaust the byte budget: %v", err) } if result == nil { t.Fatal("expected a stream result") } drainChunks(result) }) } // The frame that trips the budget must still be delivered. Releasing the window is a decision about // what to do with later frames, not a licence to drop the one in hand: on a slow turn that frame is // the first token of the answer. func TestCodexWebsocketsExecutor_BootstrapBuffering_BudgetFrameIsNotDropped(t *testing.T) { frames := make([]string, 0, codexBootstrapMaxBufferedFrames+2) for i := 0; i < codexBootstrapMaxBufferedFrames; i++ { frames = append(frames, codexInProgressEvent) } frames = append(frames, `{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"FIRSTTOKEN"}`) server := codexWebsocketRawServer(t, frames, codexCompletedEventBody) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("unexpected ExecuteStream error: %v", err) } if result == nil { t.Fatal("expected a stream result") } combined, _ := drainChunks(result) if !strings.Contains(combined, "FIRSTTOKEN") { t.Fatalf("the frame that tripped the budget was dropped: %s", combined) } } // The websocket bound must be enforced where the frame is counted. A peer that sends only frames the // bootstrap loop skips never reaches the buffering branch, so a release decided there would never // run and the downstream headers would stay uncommitted for as long as the peer keeps sending. func TestCodexWebsocketsExecutor_BootstrapBuffering_SkippedFramesExhaustTheWindow(t *testing.T) { cases := []struct { name string frame string }{ {"whitespace only frames", " \n\t "}, {"empty frames", ""}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { frames := make([]string, codexBootstrapMaxBufferedFrames+2) for i := range frames { frames[i] = tc.frame } server := codexWebsocketRawServer(t, frames, codexOverloadEvent) defer server.Close() req, opts := codexWebsocketRequest() result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("the window must release before the overload arrives, got failover: %v", err) } if result == nil { t.Fatal("expected a stream result once the window was exhausted") } drainChunks(result) }) } } // The window only protects anything if it is small. Asserting it relative to itself, as the framing // tests below must, cannot catch a budget that has grown to a useless size. func TestCodexBootstrapBudgetsStaySmall(t *testing.T) { // Pinned exactly, because the framing tests below express their expectations as multiples of // these constants and so cannot notice the budgets growing. // TestCodexExecutor_BootstrapBuffering_HeartbeatArithmeticPerFraming states the resulting // heartbeat counts in absolute numbers, which is what config.example.yaml quotes to operators. if codexBootstrapMaxBufferedFrames != 48 { t.Fatalf("frame budget = %d, want 48; changing it changes how long the headers are held, so update config.example.yaml too", codexBootstrapMaxBufferedFrames) } if codexBootstrapMaxBufferedBytes != 1<<20 { t.Fatalf("byte budget = %d, want 1MiB", codexBootstrapMaxBufferedBytes) } } // In-stream delivery depends on the held frames rendering to at least one downstream chunk: the // conductor commits a stream only once it has seen a non-empty payload, so against a format that // renders the handshake as nothing there is nothing to commit and the attempt still fails over. func TestCodexExecutor_BootstrapBuffering_InStreamDeliveryNeedsRenderedFrames(t *testing.T) { incomplete := `{"type":"response.incomplete","response":{"id":"resp_1","output":[],"usage":{"input_tokens":1,"output_tokens":0,"total_tokens":1}}}` server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOutputAddedEvent, incomplete) defer server.Close() req, _ := codexTestRequest() opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai"), Stream: true} result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("the executor still returns a stream; it is the conductor that cannot commit it: %v", err) } if result == nil { t.Fatal("expected a stream result") } combined, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the failure to arrive as an in-stream chunk error") } if combined != "" { t.Fatalf("Chat Completions renders these frames as nothing, so no payload precedes the error: %q", combined) } } // An empty response.incomplete is not an overload rejection, so it keeps its in-stream delivery: // the buffered frames are flushed and the error arrives as a chunk rather than burning another // credential. The websocket executor already routes this condition that way; both transports agree // on it. A truncated stream is deliberately left alone - both transports fail that one over. func TestCodexExecutor_BootstrapBuffering_NonOverloadTerminalStaysInStream(t *testing.T) { cases := []struct{ name, tail string }{ {"empty incomplete", `{"type":"response.incomplete","response":{"id":"resp_1","output":[],"usage":{"input_tokens":1,"output_tokens":0,"total_tokens":1}}}`}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOutputAddedEvent, tc.tail) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("a non-overload terminal failure must not fail the attempt over: %v", err) } if result == nil { t.Fatal("expected a stream result carrying the buffered frames and the error") } combined, streamErr := drainChunks(result) if streamErr == nil { t.Fatal("expected the failure to arrive as an in-stream chunk error") } if !strings.Contains(combined, `"type":"response.created"`) { t.Fatalf("buffered frames must be flushed before the in-stream error: %s", combined) } }) } } // The cap is consulted before a frame is taken, not after. A frame that alone exceeds it is refused // admission and released, so the rejection that follows it immediately - with no line in between to // trigger a late release - is delivered in-stream rather than failing the attempt over. func TestCodexExecutor_BootstrapBuffering_OversizedFrameIsNotAdmitted(t *testing.T) { oversized := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("q", codexBootstrapMaxBufferedBytes*2) + `"}}` body := "data: " + oversized + "\n" + "data: " + codexOverloadEvent + "\n\n" server := codexSSERawServer(body) defer server.Close() req, opts := codexTestRequest() result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("a frame larger than the cap must be released, not admitted: %v", err) } if result == nil { t.Fatal("expected a stream result once the oversized frame released the stream") } drainChunks(result) } // The byte budget is seeded from the upstream frame, not only from the chunks it translates into. // Against a Chat Completions downstream these frames render as zero chunks, so an accounting that // measured chunks alone would never advance and only the frame budget would remain - which these ten // frames stay well inside. func TestCodexExecutor_BootstrapBuffering_ByteCapCountsUpstreamFrames(t *testing.T) { frame := `{"type":"response.in_progress","response":{"id":"` + strings.Repeat("w", codexBootstrapMaxBufferedBytes/8) + `"}}` body := strings.Repeat("data: "+frame+"\n\n", 10) + "data: " + codexOverloadEvent + "\n\n" server := codexSSERawServer(body) defer server.Close() req, _ := codexTestRequest() opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("openai"), Stream: true} result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts) if err != nil { t.Fatalf("ten frames of cap/8 must exhaust the byte budget: %v", err) } if result == nil { t.Fatal("expected a stream result once the byte budget released the stream") }