package executor import ( "bytes" "context" "fmt" "io" "net/http" "net/http/httptest" "testing" "github.com/gorilla/websocket" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" 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 TestCodexNativeStreamFidelity(t *testing.T) { for _, source := range []sdktranslator.Format{sdktranslator.FormatCodex, sdktranslator.FormatOpenAIResponse, sdktranslator.FormatClaude, sdktranslator.FormatOpenAI} { t.Run(source.String(), func(t *testing.T) { testCodexNativeStreamFidelity(t, source) }) }} func testCodexNativeStreamFidelity(t *testing.T, source sdktranslator.Format) { t.Helper() for _, transport := range []string{"http", "websocket"} { for _, lite := range []string{"", "header", "metadata"} { for _, buffering := range []bool{false, true} { t.Run(fmt.Sprintf("%s/lite=%s/buffering=%t", transport, lite, buffering), func(t *testing.T) { metadata := `{"type":"codex.response.metadata","headers":{"x-models-etag":"models-v1","x-codex-turn-state":"turn-1","x-codex-safety-buffering-enabled":"true","x-codex-safety-buffering-faster-model":"fixture-model"},"future":{"ok":true}}` completed := `{"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[],"future":{"ok":true},"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}` events := []string{metadata, `{"type":"response.output_item.done","output_index":0,"item":{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"ok"}]}}`, completed} captured := make(chan []byte, 1) capturedHeaders := make(chan http.Header, 1) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { capturedHeaders <- r.Header.Clone() if transport == "websocket" { upgrader := websocket.Upgrader{} conn, err := upgrader.Upgrade(w, r, nil) if err != nil { t.Error(err) return } defer func() { _ = conn.Close() }() _, body, errRead := conn.ReadMessage() if errRead != nil { t.Error(errRead) return } captured <- body for _, event := range events { if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(event)); errWrite != nil { t.Error(errWrite) } } return } body, errRead := io.ReadAll(r.Body) if errRead != nil { t.Error(errRead) } captured <- body w.Header().Set("Content-Type", "text/event-stream") for _, event := range events { _, _ = fmt.Fprintf(w, "data: %s\n\n", event) } })) defer server.Close() payload := []byte(`{"model":"gpt-5.6-sol","input":[],"parallel_tool_calls":false}`) headers := http.Header{"Session-Id": {"session-1"}, "Thread-Id": {"thread-1"}} if lite == "header" { headers.Set(codexResponsesLiteHeader, "true") } else if lite == "metadata" { payload = []byte(`{"model":"gpt-5.6-sol","input":[],"parallel_tool_calls":false,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}}`) } cfg := &config.Config{Codex: config.CodexConfig{StreamBootstrapBuffering: buffering, DisableCodexCloaking: true}} executeStream := NewCodexExecutor(cfg).ExecuteStream if transport == "websocket" { executeStream = NewCodexWebsocketsExecutor(cfg).ExecuteStream } result, err := executeStream(context.Background(), codexTestAuth(server.URL), cliproxyexecutor.Request{Model: "gpt-5.6-sol", Payload: payload}, cliproxyexecutor.Options{SourceFormat: source, ResponseFormat: sdktranslator.FormatCodex, Headers: headers, Stream: true}) if err != nil { t.Fatal(err) } var terminal []byte var metadataEvents []string for chunk := range result.Chunks { if chunk.Err != nil { t.Fatal(chunk.Err) } for _, line := range bytes.Split(chunk.Payload, []byte("\n")) { data := bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data:"))) if gjson.GetBytes(data, "type").String() == "codex.response.metadata" { metadataEvents = append(metadataEvents, string(data)) } if gjson.GetBytes(data, "type").String() == "response.completed" { terminal = bytes.Clone(data) } } } body := <-captured upstreamHeaders := <-capturedHeaders native := lite != "" && (source == sdktranslator.FormatCodex || source == sdktranslator.FormatOpenAIResponse) if transport == "websocket" { wantLiteHeader := "" if native && lite == "header" { wantLiteHeader = "true" } if got := upstreamHeaders.Get(codexResponsesLiteHeader); got != wantLiteHeader { t.Errorf("upstream Lite header = %q, want %q", got, wantLiteHeader) } alias := headerValueCaseInsensitive(upstreamHeaders, "session_id") t.Logf("upstream session alias: %q", alias) if (alias == "") != native { t.Errorf("session alias = %q, native = %t", alias, native) } } t.Logf("downstream metadata: %q", metadataEvents) if len(metadataEvents) != 1 || metadataEvents[0] != metadata { t.Errorf("metadata event changed or duplicated: %q", metadataEvents) } t.Logf("upstream request: %s; downstream completion: %s", body, terminal) if native { if gjson.GetBytes(body, "instructions").Exists() { t.Errorf("native request gained instructions: %s", body) } if string(terminal) != completed { t.Errorf("native completion changed: %s", terminal) } } else if gjson.GetBytes(body, "instructions").Type != gjson.String || gjson.GetBytes(terminal, "response.output.0.id").String() != "msg_1" { t.Errorf("compatibility normalization/backfill lost: %s; %s", body, terminal) } }) } } }} func TestCodexWebsocketLiteHeaderWithoutSessionHeaders(t *testing.T) { for _, native := range []bool{false, true} { for _, disableCloaking := range []bool{false, true} { cfg := &config.Config{Codex: config.CodexConfig{DisableCodexCloaking: disableCloaking}} headers := http.Header{} headers.Set(codexResponsesLiteHeader, "true") got := applyCodexWebsocketHeaders(context.Background(), nil, nil, "fixture-token", cfg, native, headers) want := "" if native { want = "true" } if value := got.Get(codexResponsesLiteHeader); value != want { t.Errorf("native=%t disableCloaking=%t: Lite header = %q, want %q", native, disableCloaking, value, want) } } }}
package executor
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"net/http/httptest"
"testing"
"github.com/gorilla/websocket"
"github.com/router-for-me/CLIProxyAPI/v7/internal/config"
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 TestCodexNativeStreamFidelity(t *testing.T) {
for _, source := range []sdktranslator.Format{sdktranslator.FormatCodex, sdktranslator.FormatOpenAIResponse, sdktranslator.FormatClaude, sdktranslator.FormatOpenAI} {
t.Run(source.String(), func(t *testing.T) { testCodexNativeStreamFidelity(t, source) })
}
}
func testCodexNativeStreamFidelity(t *testing.T, source sdktranslator.Format) {
t.Helper()
for _, transport := range []string{"http", "websocket"} {
for _, lite := range []string{"", "header", "metadata"} {
for _, buffering := range []bool{false, true} {
t.Run(fmt.Sprintf("%s/lite=%s/buffering=%t", transport, lite, buffering), func(t *testing.T) {
metadata := `{"type":"codex.response.metadata","headers":{"x-models-etag":"models-v1","x-codex-turn-state":"turn-1","x-codex-safety-buffering-enabled":"true","x-codex-safety-buffering-faster-model":"fixture-model"},"future":{"ok":true}}`
completed := `{"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[],"future":{"ok":true},"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}`
events := []string{metadata, `{"type":"response.output_item.done","output_index":0,"item":{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"ok"}]}}`, completed}
captured := make(chan []byte, 1)
capturedHeaders := make(chan http.Header, 1)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
capturedHeaders <- r.Header.Clone()
if transport == "websocket" {
upgrader := websocket.Upgrader{}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
t.Error(err)
return
}
defer func() { _ = conn.Close() }()
_, body, errRead := conn.ReadMessage()
if errRead != nil {
t.Error(errRead)
return
}
captured <- body
for _, event := range events {
if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(event)); errWrite != nil {
t.Error(errWrite)
}
}
return
}
body, errRead := io.ReadAll(r.Body)
if errRead != nil {
t.Error(errRead)
}
captured <- body
w.Header().Set("Content-Type", "text/event-stream")
for _, event := range events {
_, _ = fmt.Fprintf(w, "data: %s\n\n", event)
}
}))
defer server.Close()
payload := []byte(`{"model":"gpt-5.6-sol","input":[],"parallel_tool_calls":false}`)
headers := http.Header{"Session-Id": {"session-1"}, "Thread-Id": {"thread-1"}}
if lite == "header" {
headers.Set(codexResponsesLiteHeader, "true")
} else if lite == "metadata" {
payload = []byte(`{"model":"gpt-5.6-sol","input":[],"parallel_tool_calls":false,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}}`)
}
cfg := &config.Config{Codex: config.CodexConfig{StreamBootstrapBuffering: buffering, DisableCodexCloaking: true}}
executeStream := NewCodexExecutor(cfg).ExecuteStream
if transport == "websocket" {
executeStream = NewCodexWebsocketsExecutor(cfg).ExecuteStream
}
result, err := executeStream(context.Background(), codexTestAuth(server.URL), cliproxyexecutor.Request{Model: "gpt-5.6-sol", Payload: payload}, cliproxyexecutor.Options{SourceFormat: source, ResponseFormat: sdktranslator.FormatCodex, Headers: headers, Stream: true})
if err != nil {
t.Fatal(err)
}
var terminal []byte
var metadataEvents []string
for chunk := range result.Chunks {
if chunk.Err != nil {
t.Fatal(chunk.Err)
}
for _, line := range bytes.Split(chunk.Payload, []byte("\n")) {
data := bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data:")))
if gjson.GetBytes(data, "type").String() == "codex.response.metadata" {
metadataEvents = append(metadataEvents, string(data))
}
if gjson.GetBytes(data, "type").String() == "response.completed" {
terminal = bytes.Clone(data)
}
}
}
body := <-captured
upstreamHeaders := <-capturedHeaders
native := lite != "" && (source == sdktranslator.FormatCodex || source == sdktranslator.FormatOpenAIResponse)
if transport == "websocket" {
wantLiteHeader := ""
if native && lite == "header" {
wantLiteHeader = "true"
}
if got := upstreamHeaders.Get(codexResponsesLiteHeader); got != wantLiteHeader {
t.Errorf("upstream Lite header = %q, want %q", got, wantLiteHeader)
}
alias := headerValueCaseInsensitive(upstreamHeaders, "session_id")
t.Logf("upstream session alias: %q", alias)
if (alias == "") != native {
t.Errorf("session alias = %q, native = %t", alias, native)
}
}
t.Logf("downstream metadata: %q", metadataEvents)
if len(metadataEvents) != 1 || metadataEvents[0] != metadata {
t.Errorf("metadata event changed or duplicated: %q", metadataEvents)
}
t.Logf("upstream request: %s; downstream completion: %s", body, terminal)
if native {
if gjson.GetBytes(body, "instructions").Exists() {
t.Errorf("native request gained instructions: %s", body)
}
if string(terminal) != completed {
t.Errorf("native completion changed: %s", terminal)
}
} else if gjson.GetBytes(body, "instructions").Type != gjson.String || gjson.GetBytes(terminal, "response.output.0.id").String() != "msg_1" {
t.Errorf("compatibility normalization/backfill lost: %s; %s", body, terminal)
}
})
}
}
}
}
func TestCodexWebsocketLiteHeaderWithoutSessionHeaders(t *testing.T) {
for _, native := range []bool{false, true} {
for _, disableCloaking := range []bool{false, true} {
cfg := &config.Config{Codex: config.CodexConfig{DisableCodexCloaking: disableCloaking}}
headers := http.Header{}
headers.Set(codexResponsesLiteHeader, "true")
got := applyCodexWebsocketHeaders(context.Background(), nil, nil, "fixture-token", cfg, native, headers)
want := ""
if native {
want = "true"
}
if value := got.Get(codexResponsesLiteHeader); value != want {
t.Errorf("native=%t disableCloaking=%t: Lite header = %q, want %q", native, disableCloaking, value, want)
}
}
}
}