mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-09-09 17:42:08 +08:00
ba7e55836d
- Check `error.message` and `message` for retry advice on upstream `server_error` responses. - Treat server errors indicating the request can be retried as eligible overload bootstrap failures.
488 lines
21 KiB
Go
488 lines
21 KiB
Go
package executor
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
|
|
"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"
|
|
)
|
|
|
|
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}`
|
|
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}`
|
|
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 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)
|
|
}
|
|
}
|
|
|
|
// 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, "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, codexBootstrapMaxBufferedEvents+2)
|
|
for i := 0; i < codexBootstrapMaxBufferedEvents+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")
|
|
}
|
|
}
|
|
|
|
// Buffered handshake events must be replayed in upstream order ahead of the first generated event.
|
|
func TestCodexExecutor_BootstrapBuffering_FlushesInOrderOnFirstOutput(t *testing.T) {
|
|
server := codexSSEServer(codexCreatedEvent, codexInProgressEvent, codexOutputAddedEvent, codexCompletedEventBody)
|
|
defer server.Close()
|
|
|
|
req, opts := codexTestRequest()
|
|
result, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts)
|
|
if err != nil {
|
|
t.Fatalf("unexpected ExecuteStream error: %v", err)
|
|
}
|
|
|
|
combined, streamErr := drainChunks(result)
|
|
if streamErr != nil {
|
|
t.Fatalf("unexpected chunk error: %v", streamErr)
|
|
}
|
|
createdAt := strings.Index(combined, "response.created")
|
|
addedAt := strings.Index(combined, "response.output_item.added")
|
|
if createdAt < 0 || addedAt < 0 {
|
|
t.Fatalf("missing handshake or first generated event: %s", combined)
|
|
}
|
|
if createdAt > addedAt {
|
|
t.Fatalf("buffered handshake must be replayed before the first generated event: %s", combined)
|
|
}
|
|
}
|
|
|
|
// With the feature disabled the overload rejection keeps its legacy in-stream delivery.
|
|
func TestCodexExecutor_BootstrapBuffering_DefaultDisabledPassthrough(t *testing.T) {
|
|
server := codexSSEServer(codexCreatedEvent, codexOverloadEvent)
|
|
defer server.Close()
|
|
|
|
req, opts := codexTestRequest()
|
|
result, err := NewCodexExecutor(&config.Config{}).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts)
|
|
if err != nil {
|
|
t.Fatalf("default unbuffered ExecuteStream returned error at call time: %v", err)
|
|
}
|
|
if result == nil {
|
|
t.Fatal("expected non-nil result in default unbuffered mode")
|
|
}
|
|
_, streamErr := drainChunks(result)
|
|
if streamErr == nil {
|
|
t.Fatal("expected stream error in chunks for default unbuffered mode")
|
|
}
|
|
// Disabling the feature must restore the previous behaviour exactly, status classification
|
|
// included: the 503 restoration is scoped to the buffered failover path, so an unbuffered
|
|
// overload still classifies as a bad gateway and keeps its old cooldown treatment.
|
|
if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadGateway {
|
|
t.Fatalf("status code = %d, want %d while buffering is disabled", got, http.StatusBadGateway)
|
|
}
|
|
}
|
|
|
|
// A cancelled downstream request must surface the context error rather than being recorded as an
|
|
// upstream failure that penalises the credential.
|
|
func TestCodexExecutor_BootstrapBuffering_ContextCancelDuringBootstrap(t *testing.T) {
|
|
server := codexSSEServer(codexCreatedEvent)
|
|
defer server.Close()
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
req, opts := codexTestRequest()
|
|
_, err := NewCodexExecutor(codexBufferingConfig(true)).ExecuteStream(ctx, codexTestAuth(server.URL), req, opts)
|
|
if err == nil {
|
|
t.Fatal("expected an error for a cancelled bootstrap")
|
|
}
|
|
if !strings.Contains(err.Error(), context.Canceled.Error()) {
|
|
t.Fatalf("expected the context cancellation to surface, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestCodexWebsocketsExecutor_BootstrapBuffering_OverloadFailsAttempt(t *testing.T) {
|
|
server := codexWebsocketServer(t, codexCreatedEvent, codexInProgressEvent, codexOverloadEvent)
|
|
defer server.Close()
|
|
|
|
req, opts := codexWebsocketRequest()
|
|
result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts)
|
|
|
|
if err == nil {
|
|
t.Fatal("expected ExecuteStream to fail the attempt on a websocket overload rejection")
|
|
}
|
|
if result != nil {
|
|
t.Fatal("expected nil result so no buffered handshake frame can reach the client")
|
|
}
|
|
if got := statusCodeFromTestError(t, err); got != http.StatusServiceUnavailable {
|
|
t.Fatalf("status code = %d, want %d", got, http.StatusServiceUnavailable)
|
|
}
|
|
}
|
|
|
|
// The websocket transport prefixes response events with private metadata frames. Frame order
|
|
// below matches live wire capture: codex.rate_limits and codex.response.metadata both arrive
|
|
// *before* response.created, making the first generated event the fifth frame. They must be
|
|
// treated as handshake events, otherwise a fixed 3-event window would release the stream at
|
|
// response.created and never observe the rejection.
|
|
func TestCodexWebsocketsExecutor_BootstrapBuffering_PrivateHandshakeFramesDoNotExhaustWindow(t *testing.T) {
|
|
server := codexWebsocketServer(t,
|
|
`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1}}}`,
|
|
`{"type":"codex.response.metadata","metadata":{"conversation_id":"conv_1"}}`,
|
|
codexCreatedEvent,
|
|
codexInProgressEvent,
|
|
codexOverloadEvent,
|
|
)
|
|
defer server.Close()
|
|
|
|
req, opts := codexWebsocketRequest()
|
|
result, err := NewCodexWebsocketsExecutor(codexBufferingConfig(true)).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts)
|
|
|
|
if err == nil {
|
|
t.Fatal("expected the overload rejection to be caught past the private handshake frames")
|
|
}
|
|
if result != nil {
|
|
t.Fatal("expected nil result so no buffered frame can reach the client")
|
|
}
|
|
if got := statusCodeFromTestError(t, err); got != http.StatusServiceUnavailable {
|
|
t.Fatalf("status code = %d, want %d", got, http.StatusServiceUnavailable)
|
|
}
|
|
}
|
|
|
|
func TestCodexWebsocketsExecutor_BootstrapBuffering_NonOverloadStaysInStream(t *testing.T) {
|
|
server := codexWebsocketServer(t, codexCreatedEvent, codexInProgressEvent, codexInvalidEvent)
|
|
defer server.Close()
|
|
|
|
req, opts := codexWebsocketRequest()
|
|
result, err := NewCodexWebsocketsExecutor(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, "response.created") {
|
|
t.Fatalf("buffered handshake must be flushed before the in-stream error: %s", combined)
|
|
}
|
|
}
|
|
|
|
func TestCodexWebsocketsExecutor_BootstrapBuffering_FlushesInOrderOnFirstOutput(t *testing.T) {
|
|
server := codexWebsocketServer(t,
|
|
codexCreatedEvent,
|
|
codexInProgressEvent,
|
|
codexOutputAddedEvent,
|
|
`{"type":"response.completed","response":{"id":"resp_1","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}`,
|
|
)
|
|
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)
|
|
}
|
|
|
|
combined, streamErr := drainChunks(result)
|
|
if streamErr != nil {
|
|
t.Fatalf("unexpected chunk error: %v", streamErr)
|
|
}
|
|
createdAt := strings.Index(combined, "response.created")
|
|
addedAt := strings.Index(combined, "response.output_item.added")
|
|
if createdAt < 0 || addedAt < 0 {
|
|
t.Fatalf("missing handshake or first generated event: %s", combined)
|
|
}
|
|
if createdAt > addedAt {
|
|
t.Fatalf("buffered handshake must be replayed before the first generated event: %s", combined)
|
|
}
|
|
}
|
|
|
|
func TestCodexWebsocketsExecutor_BootstrapBuffering_DefaultDisabledPassthrough(t *testing.T) {
|
|
server := codexWebsocketServer(t, codexCreatedEvent, codexOverloadEvent)
|
|
defer server.Close()
|
|
|
|
req, opts := codexWebsocketRequest()
|
|
result, err := NewCodexWebsocketsExecutor(&config.Config{}).ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts)
|
|
if err != nil {
|
|
t.Fatalf("default unbuffered ExecuteStream returned error at call time: %v", err)
|
|
}
|
|
if result == nil {
|
|
t.Fatal("expected non-nil result in default unbuffered mode")
|
|
}
|
|
_, streamErr := drainChunks(result)
|
|
if streamErr == nil {
|
|
t.Fatal("expected stream error in chunks for default unbuffered mode")
|
|
}
|
|
if got := statusCodeFromTestError(t, streamErr); got != http.StatusBadGateway {
|
|
t.Fatalf("status code = %d, want %d while buffering is disabled", got, http.StatusBadGateway)
|
|
}
|
|
}
|
|
|
|
// The 503 restoration is scoped to the buffered failover path, so this only covers which
|
|
// rejections are eligible to replace the whole attempt.
|
|
func TestIsCodexOverloadBootstrapFailureRejectsRequestFaults(t *testing.T) {
|
|
notOverload := []string{
|
|
`{"error":{"type":"invalid_request_error","code":"invalid_value"}}`,
|
|
`{"error":{"type":"authentication_error","code":"invalid_api_key"}}`,
|
|
`{"error":{"type":"upstream_error","code":"unknown"}}`,
|
|
`{"error":{"type":"server_error","code":"server_error","message":"An internal error occurred without retry advice"}}`,
|
|
}
|
|
for _, body := range notOverload {
|
|
if isCodexOverloadBootstrapFailure([]byte(body)) {
|
|
t.Fatalf("request-level fault must not trigger bootstrap failover: %s", body)
|
|
}
|
|
}
|
|
if !isCodexOverloadBootstrapFailure([]byte(`{"error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`)) {
|
|
t.Fatal("rate limit rejections should be eligible for bootstrap failover")
|
|
}
|
|
serverErrorFull := `{"error":{"type":"server_error","code":"server_error","message":"An error occurred while processing your request. You can retry your request, or contact us through our help center at help.openai.com if the error persists. Please include the request ID 2f5014c9-7cfe-4fb7-813e-ecb447da3edd in your message."}}`
|
|
if !isCodexOverloadBootstrapFailure([]byte(serverErrorFull)) {
|
|
t.Fatal("server_error with 'You can retry your request' should be eligible for bootstrap failover")
|
|
}
|
|
serverErrorShort := `{"error":{"type":"server_error","code":"server_error","message":"You can retry your request"}}`
|
|
if !isCodexOverloadBootstrapFailure([]byte(serverErrorShort)) {
|
|
t.Fatal("short server_error with 'You can retry your request' should be eligible for bootstrap failover")
|
|
}
|
|
}
|
|
|
|
// codexWebsocketServerHoldingConnection behaves like codexWebsocketServer but keeps the upstream
|
|
// connection open after writing the frames, so the executor's own teardown path is the only
|
|
// source of session invalidation. With the plain helper the connection closes immediately, the
|
|
// reader goroutine observes EOF first and reports upstream_disconnected, which both masks the
|
|
// path under test and can make a disconnect assertion pass for the wrong reason.
|
|
func codexWebsocketServerHoldingConnection(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))
|
|
}
|
|
for {
|
|
if _, _, errRead := conn.ReadMessage(); errRead != nil {
|
|
return
|
|
}
|
|
}
|
|
}))
|
|
}
|
|
|
|
// executeWebsocketStreamInSession runs ExecuteStream bound to a named execution session and
|
|
// reports whether the upstream teardown was signalled to the downstream handler.
|
|
//
|
|
// The downstream Responses WebSocket handler subscribes to UpstreamDisconnectChan and closes
|
|
// the client connection as soon as a disconnect is published. A bootstrap overload is retried
|
|
// on another credential, so publishing there would tear down the client connection before the
|
|
// retry can deliver anything, and the client would observe an abnormal close with zero frames.
|
|
func executeWebsocketStreamInSession(t *testing.T, frames ...string) (notified bool, err error) {
|
|
t.Helper()
|
|
|
|
server := codexWebsocketServerHoldingConnection(t, frames...)
|
|
defer server.Close()
|
|
|
|
exec := NewCodexWebsocketsExecutor(codexBufferingConfig(true))
|
|
exec.store = &codexWebsocketSessionStore{sessions: make(map[string]*codexWebsocketSession)}
|
|
|
|
const sessionID = "bootstrap-session"
|
|
disconnectCh := exec.UpstreamDisconnectChan(sessionID)
|
|
if disconnectCh == nil {
|
|
t.Fatal("expected a disconnect channel")
|
|
}
|
|
|
|
req, opts := codexWebsocketRequest()
|
|
opts.Metadata = map[string]any{cliproxyexecutor.ExecutionSessionMetadataKey: sessionID}
|
|
_, err = exec.ExecuteStream(context.Background(), codexTestAuth(server.URL), req, opts)
|
|
|
|
select {
|
|
case <-disconnectCh:
|
|
notified = true
|
|
default:
|
|
}
|
|
return notified, err
|
|
}
|
|
|
|
func TestCodexWebsocketsExecutor_BootstrapOverload_DoesNotNotifyDownstreamDisconnect(t *testing.T) {
|
|
notified, err := executeWebsocketStreamInSession(t, codexCreatedEvent, codexInProgressEvent, codexOverloadEvent)
|
|
|
|
if err == nil {
|
|
t.Fatal("expected the overload rejection to fail the attempt")
|
|
}
|
|
if got := statusCodeFromTestError(t, err); got != http.StatusServiceUnavailable {
|
|
t.Fatalf("status code = %d, want %d", got, http.StatusServiceUnavailable)
|
|
}
|
|
if notified {
|
|
t.Fatal("bootstrap overload must not signal a downstream disconnect: the conductor still has to retry on another credential, and signalling closes the client connection with zero frames delivered")
|
|
}
|
|
}
|
|
|
|
// A non-overload terminal failure is delivered in-stream and genuinely ends the session, so it
|
|
// must keep signalling the disconnect exactly as it did before buffering existed.
|
|
func TestCodexWebsocketsExecutor_BootstrapNonOverload_StillNotifiesDownstreamDisconnect(t *testing.T) {
|
|
notified, err := executeWebsocketStreamInSession(t, codexCreatedEvent, codexInProgressEvent, codexInvalidEvent)
|
|
|
|
if err != nil {
|
|
t.Fatalf("non-overload failures stay in-stream, got err = %v", err)
|
|
}
|
|
if !notified {
|
|
t.Fatal("a terminal failure that is delivered in-stream must still signal the downstream disconnect")
|
|
}
|
|
}
|