From 81d70f5d9f3fdb39a6290ed9c917ff0c6f27ca30 Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Sat, 18 Jul 2026 03:58:03 +0800 Subject: [PATCH] feat(executor): add normalization for parallel tool calls in Codex executors - Introduced `normalizeCodexWebsocketParallelToolCalls` and `normalizeCodexParallelToolCalls` to enforce consistent `parallel_tool_calls` handling. - Updated WebSocket and HTTP executor logic to support headers during parallel tool normalization. - Refactored redundant logic by consolidating normalization routines for improved clarity and maintainability. --- internal/runtime/executor/codex_executor.go | 14 +++- .../executor/codex_executor_imagegen_test.go | 51 ++++++++++++ ...codex_executor_parallel_tool_calls_test.go | 25 ++++++ .../executor/codex_websockets_executor.go | 10 +++ .../codex_websockets_executor_test.go | 83 ++++++++++++++++++- 5 files changed, 178 insertions(+), 5 deletions(-) diff --git a/internal/runtime/executor/codex_executor.go b/internal/runtime/executor/codex_executor.go index 7ed57b315..7abbbc63f 100644 --- a/internal/runtime/executor/codex_executor.go +++ b/internal/runtime/executor/codex_executor.go @@ -1147,7 +1147,7 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex executor", body) - body = normalizeCodexParallelToolCallsForTools(body) + body = normalizeCodexParallelToolCalls(body, opts.Headers) body, replayScope, errReplay := applyCodexReasoningReplayCacheRequired(ctx, from, req, opts, body) if errReplay != nil { return resp, errReplay @@ -1327,7 +1327,7 @@ func (e *CodexExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.A body, _ = sjson.DeleteBytes(body, "stream") body = normalizeCodexInstructions(body) body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex executor", body) - body = normalizeCodexParallelToolCallsForTools(body) + body = normalizeCodexParallelToolCalls(body, opts.Headers) reporter.SetTranslatedReasoningEffort(body, to.String()) url := strings.TrimSuffix(baseURL, "/") + "/responses/compact" @@ -1438,7 +1438,7 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex executor", body) - body = normalizeCodexParallelToolCallsForTools(body) + body = normalizeCodexParallelToolCalls(body, opts.Headers) body, replayScope, errReplay := applyCodexReasoningReplayCacheRequired(ctx, from, req, opts, body) if errReplay != nil { return nil, errReplay @@ -2191,6 +2191,14 @@ func ensureImageGenerationTool(body []byte, baseModel string, auth *cliproxyauth return body } +func normalizeCodexParallelToolCalls(body []byte, headers http.Header) []byte { + if isCodexResponsesLiteRequest(body, headers) { + body, _ = sjson.SetBytes(body, "parallel_tool_calls", false) + return body + } + return normalizeCodexParallelToolCallsForTools(body) +} + func normalizeCodexParallelToolCallsForTools(body []byte) []byte { if !gjson.GetBytes(body, "parallel_tool_calls").Exists() { return body diff --git a/internal/runtime/executor/codex_executor_imagegen_test.go b/internal/runtime/executor/codex_executor_imagegen_test.go index dfb50584c..10fc36e7e 100644 --- a/internal/runtime/executor/codex_executor_imagegen_test.go +++ b/internal/runtime/executor/codex_executor_imagegen_test.go @@ -52,6 +52,57 @@ func TestCodexExecutorExecuteResponsesLiteHeaderDoesNotInjectImageGenerationTool if tools := gjson.GetBytes(gotBody, "tools"); tools.Exists() { t.Fatalf("unexpected tools in responses-lite upstream payload: %s", tools.Raw) } + parallelToolCalls := gjson.GetBytes(gotBody, "parallel_tool_calls") + if !parallelToolCalls.Exists() || parallelToolCalls.Bool() { + t.Fatalf("responses-lite parallel_tool_calls should be false: %s", gotBody) + } +} + +func TestCodexExecutorExecuteStreamResponsesLiteHeaderForcesParallelToolCallsFalse(t *testing.T) { + var gotBody []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, errRead := io.ReadAll(r.Body) + if errRead != nil { + t.Fatalf("read request body: %v", errRead) + } + gotBody = body + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":0,\"output_tokens\":0,\"total_tokens\":0}}}\n\n")) + })) + defer server.Close() + + executor := NewCodexExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Provider: "codex", + Attributes: map[string]string{ + "api_key": "test", + "base_url": server.URL, + "plan_type": "pro", + }, + } + headers := make(http.Header) + headers.Set(codexResponsesLiteHeader, "true") + + result, errExecute := executor.ExecuteStream(context.Background(), auth, cliproxyexecutor.Request{ + Model: "gpt-5.6-luna", + Payload: []byte(`{"model":"gpt-5.6-luna","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + Headers: headers, + }) + if errExecute != nil { + t.Fatalf("ExecuteStream() error = %v", errExecute) + } + for chunk := range result.Chunks { + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + } + + parallelToolCalls := gjson.GetBytes(gotBody, "parallel_tool_calls") + if !parallelToolCalls.Exists() || parallelToolCalls.Bool() { + t.Fatalf("responses-lite parallel_tool_calls should be false: %s", gotBody) + } } func TestEnsureImageGenerationTool_ResponsesLiteMetadataDoesNotInjectTool(t *testing.T) { diff --git a/internal/runtime/executor/codex_executor_parallel_tool_calls_test.go b/internal/runtime/executor/codex_executor_parallel_tool_calls_test.go index d1f4f8e17..f64d23294 100644 --- a/internal/runtime/executor/codex_executor_parallel_tool_calls_test.go +++ b/internal/runtime/executor/codex_executor_parallel_tool_calls_test.go @@ -1,6 +1,7 @@ package executor import ( + "net/http" "testing" "github.com/tidwall/gjson" @@ -38,3 +39,27 @@ func TestNormalizeCodexParallelToolCallsForTools_PreservesWhenToolsPresent(t *te t.Fatalf("parallel_tool_calls should be preserved when tools are present: %s", string(out)) } } + +func TestNormalizeCodexParallelToolCalls_ResponsesLiteMetadataForcesFalse(t *testing.T) { + body := []byte(`{"model":"gpt-5.6-luna","tools":[{"type":"function","name":"lookup"}],"parallel_tool_calls":true,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"},"input":"hi"}`) + + out := normalizeCodexParallelToolCalls(body, nil) + + parallelToolCalls := gjson.GetBytes(out, "parallel_tool_calls") + if !parallelToolCalls.Exists() || parallelToolCalls.Bool() { + t.Fatalf("responses-lite parallel_tool_calls should be false: %s", string(out)) + } +} + +func TestNormalizeCodexParallelToolCalls_ResponsesLiteHeaderForcesFalse(t *testing.T) { + body := []byte(`{"model":"gpt-5.6-luna","parallel_tool_calls":true,"input":"hi"}`) + headers := make(http.Header) + headers.Set(codexResponsesLiteHeader, "true") + + out := normalizeCodexParallelToolCalls(body, headers) + + parallelToolCalls := gjson.GetBytes(out, "parallel_tool_calls") + if !parallelToolCalls.Exists() || parallelToolCalls.Bool() { + t.Fatalf("responses-lite parallel_tool_calls should be false: %s", string(out)) + } +} diff --git a/internal/runtime/executor/codex_websockets_executor.go b/internal/runtime/executor/codex_websockets_executor.go index 64eff7452..5cbd4f01b 100644 --- a/internal/runtime/executor/codex_websockets_executor.go +++ b/internal/runtime/executor/codex_websockets_executor.go @@ -275,6 +275,7 @@ func (e *CodexWebsocketsExecutor) Execute(ctx context.Context, auth *cliproxyaut body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex websockets executor", body) + body = normalizeCodexWebsocketParallelToolCalls(body, opts.Headers) body, replayScope, errReplay := applyCodexReasoningReplayCacheRequired(ctx, from, req, opts, body) if errReplay != nil { return resp, errReplay @@ -520,6 +521,7 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex websockets executor", body) + body = normalizeCodexWebsocketParallelToolCalls(body, opts.Headers) body, replayScope, errReplay := applyCodexReasoningReplayCacheRequired(ctx, from, req, opts, body) if errReplay != nil { return nil, errReplay @@ -856,6 +858,14 @@ func mapCodexWebsocketReadError(err error) error { return err } +func normalizeCodexWebsocketParallelToolCalls(body []byte, headers http.Header) []byte { + if !isCodexResponsesLiteRequest(body, headers) { + return body + } + body, _ = sjson.SetBytes(body, "parallel_tool_calls", false) + return body +} + func buildCodexWebsocketRequestBody(body []byte) []byte { if len(body) == 0 { return nil diff --git a/internal/runtime/executor/codex_websockets_executor_test.go b/internal/runtime/executor/codex_websockets_executor_test.go index 2b3c10dd5..79cba3f12 100644 --- a/internal/runtime/executor/codex_websockets_executor_test.go +++ b/internal/runtime/executor/codex_websockets_executor_test.go @@ -199,7 +199,7 @@ func TestCodexWebsocketsExecuteResponsesLiteDoesNotInjectImageGenerationTool(t * } req := cliproxyexecutor.Request{ Model: "gpt-5.6-sol", - Payload: []byte(`{"model":"gpt-5.6-sol","input":[{"type":"additional_tools","role":"developer","tools":[{"type":"custom","name":"exec"}]},{"role":"user","content":"hello"}],"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}}`), + Payload: []byte(`{"model":"gpt-5.6-sol","input":[{"type":"additional_tools","role":"developer","tools":[{"type":"custom","name":"exec"}]},{"role":"user","content":"hello"}],"parallel_tool_calls":true,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}}`), } opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("codex")} @@ -218,6 +218,81 @@ func TestCodexWebsocketsExecuteResponsesLiteDoesNotInjectImageGenerationTool(t * if got := gjson.GetBytes(payload, "client_metadata.ws_request_header_x_openai_internal_codex_responses_lite").String(); got != "true" { t.Fatalf("responses-lite metadata = %q, want true; payload=%s", got, payload) } + parallelToolCalls := gjson.GetBytes(payload, "parallel_tool_calls") + if !parallelToolCalls.Exists() || parallelToolCalls.Bool() { + t.Fatalf("responses-lite parallel_tool_calls should be false: %s", payload) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for upstream websocket payload") + } +} + +func TestCodexWebsocketsExecuteStreamResponsesLiteForcesParallelToolCallsFalse(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + capturedPayload := make(chan []byte, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, errUpgrade := upgrader.Upgrade(w, r, nil) + if errUpgrade != nil { + t.Errorf("upgrade websocket: %v", errUpgrade) + return + } + defer func() { _ = conn.Close() }() + + _, payload, errRead := conn.ReadMessage() + if errRead != nil { + t.Errorf("read upstream websocket message: %v", errRead) + return + } + capturedPayload <- bytes.Clone(payload) + + completed := []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}`) + if errWrite := conn.WriteMessage(websocket.TextMessage, completed); errWrite != nil { + t.Errorf("write completed websocket message: %v", errWrite) + } + })) + defer server.Close() + + exec := NewCodexWebsocketsExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Provider: "codex", + Attributes: map[string]string{ + "api_key": "sk-test", + "base_url": server.URL, + "plan_type": "pro", + }, + } + req := cliproxyexecutor.Request{ + Model: "gpt-5.6-luna", + Payload: []byte(`{"model":"gpt-5.6-luna","input":[{"type":"additional_tools","role":"developer","tools":[{"type":"custom","name":"exec"}]},{"role":"user","content":"hello"}],"parallel_tool_calls":true,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}}`), + } + opts := cliproxyexecutor.Options{SourceFormat: sdktranslator.FromString("codex")} + + result, errExecute := exec.ExecuteStream(context.Background(), auth, req, opts) + if errExecute != nil { + t.Fatalf("ExecuteStream() error = %v", errExecute) + } + streamComplete := false + for !streamComplete { + select { + case chunk, ok := <-result.Chunks: + if !ok { + streamComplete = true + continue + } + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for websocket stream completion") + } + } + + select { + case payload := <-capturedPayload: + parallelToolCalls := gjson.GetBytes(payload, "parallel_tool_calls") + if !parallelToolCalls.Exists() || parallelToolCalls.Bool() { + t.Fatalf("responses-lite parallel_tool_calls should be false: %s", payload) + } case <-time.After(5 * time.Second): t.Fatal("timed out waiting for upstream websocket payload") } @@ -311,7 +386,7 @@ func TestCodexWebsocketsExecuteStreamPassesThroughUpstreamWebsocketPayloadForDow auth := &cliproxyauth.Auth{Attributes: map[string]string{"api_key": "sk-test", "base_url": server.URL}} req := cliproxyexecutor.Request{ Model: "gpt-5-codex", - Payload: []byte(`{"model":"prolite/gpt-5-codex","input":[{"type":"message","role":"user","content":"hello"}]}`), + Payload: []byte(`{"model":"prolite/gpt-5-codex","input":[{"type":"additional_tools","role":"developer","tools":[{"type":"custom","name":"exec"}]},{"type":"message","role":"user","content":"hello"}],"parallel_tool_calls":true}`), } opts := cliproxyexecutor.Options{ SourceFormat: sdktranslator.FromString("openai-response"), @@ -344,6 +419,10 @@ func TestCodexWebsocketsExecuteStreamPassesThroughUpstreamWebsocketPayloadForDow if got := gjson.GetBytes(payload, "model").String(); got != "gpt-5-codex" { t.Fatalf("upstream model = %s, want gpt-5-codex; payload=%s", got, payload) } + parallelToolCalls := gjson.GetBytes(payload, "parallel_tool_calls") + if !parallelToolCalls.Exists() || !parallelToolCalls.Bool() { + t.Fatalf("non-lite parallel_tool_calls should be preserved: %s", payload) + } case <-time.After(5 * time.Second): t.Fatal("timed out waiting for upstream websocket payload") }