From 66808413df18bc1af96c8832648de6c1ebf33122 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Thu, 20 Aug 2026 20:10:54 +0800 Subject: [PATCH] fix(openai): drop orphan replay tool calls --- .../openai_ws_forwarder_ingress_test.go | 85 +++++++++++++++++++ .../service/openai_ws_forwarder_payload.go | 35 ++++++++ .../service/openai_ws_http_bridge_test.go | 10 ++- 3 files changed, 128 insertions(+), 2 deletions(-) diff --git a/backend/internal/service/openai_ws_forwarder_ingress_test.go b/backend/internal/service/openai_ws_forwarder_ingress_test.go index a7bdf31b7d..1f00d0dce6 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress_test.go +++ b/backend/internal/service/openai_ws_forwarder_ingress_test.go @@ -852,6 +852,91 @@ func TestBuildOpenAIWSReplayInputSequence(t *testing.T) { require.Equal(t, "world", gjson.GetBytes(items[1], "text").String()) }) + t.Run("previous_response_id_filters_orphan_historical_custom_tool_call", func(t *testing.T) { + previousFull := []json.RawMessage{ + json.RawMessage(`{"type":"input_text","text":"hello"}`), + json.RawMessage(`{"type":"custom_tool_call","id":"item_orphan","call_id":"call_orphan","name":"exec","input":"pwd"}`), + } + items, exists, err := buildOpenAIWSReplayInputSequence( + previousFull, + true, + []byte(`{"previous_response_id":"resp_1","input":[{"role":"user","content":"continue"}]}`), + true, + ) + require.NoError(t, err) + require.True(t, exists) + require.Len(t, items, 2) + require.Equal(t, "hello", gjson.GetBytes(items[0], "text").String()) + require.Equal(t, "user", gjson.GetBytes(items[1], "role").String()) + }) + + t.Run("previous_response_id_preserves_paired_historical_function_call", func(t *testing.T) { + previousFull := []json.RawMessage{ + json.RawMessage(`{"type":"function_call","id":"item_1","call_id":"call_1","name":"lookup","arguments":"{}"}`), + json.RawMessage(`{"type":"function_call_output","call_id":"call_1","output":"ok"}`), + } + items, exists, err := buildOpenAIWSReplayInputSequence( + previousFull, + true, + []byte(`{"previous_response_id":"resp_1","input":[{"role":"user","content":"continue"}]}`), + true, + ) + require.NoError(t, err) + require.True(t, exists) + require.Len(t, items, 3) + require.Equal(t, "function_call", gjson.GetBytes(items[0], "type").String()) + require.Equal(t, "function_call_output", gjson.GetBytes(items[1], "type").String()) + }) + + t.Run("previous_response_id_preserves_paired_historical_custom_tool_call", func(t *testing.T) { + previousFull := []json.RawMessage{ + json.RawMessage(`{"type":"custom_tool_call","id":"item_1","call_id":"call_1","name":"exec","input":"pwd"}`), + json.RawMessage(`{"type":"custom_tool_call_output","call_id":"call_1","output":"/tmp"}`), + } + items, exists, err := buildOpenAIWSReplayInputSequence( + previousFull, + true, + []byte(`{"previous_response_id":"resp_1","input":[{"role":"user","content":"continue"}]}`), + true, + ) + require.NoError(t, err) + require.True(t, exists) + require.Len(t, items, 3) + require.Equal(t, "custom_tool_call", gjson.GetBytes(items[0], "type").String()) + require.Equal(t, "custom_tool_call_output", gjson.GetBytes(items[1], "type").String()) + }) + + t.Run("item_reference_does_not_complete_historical_call", func(t *testing.T) { + previousFull := []json.RawMessage{ + json.RawMessage(`{"type":"custom_tool_call","id":"item_1","call_id":"call_1","name":"exec","input":"pwd"}`), + } + items, exists, err := buildOpenAIWSReplayInputSequence( + previousFull, + true, + []byte(`{"previous_response_id":"resp_1","input":[{"type":"item_reference","id":"call_1"},{"role":"user","content":"continue"}]}`), + true, + ) + require.NoError(t, err) + require.True(t, exists) + require.Len(t, items, 2) + require.Equal(t, "item_reference", gjson.GetBytes(items[0], "type").String()) + require.Equal(t, "user", gjson.GetBytes(items[1], "role").String()) + }) + + t.Run("previous_response_id_preserves_current_orphan_custom_tool_call", func(t *testing.T) { + items, exists, err := buildOpenAIWSReplayInputSequence( + lastFull, + true, + []byte(`{"previous_response_id":"resp_1","input":[{"type":"custom_tool_call","id":"item_live","call_id":"call_live","name":"exec","input":"pwd"}]}`), + true, + ) + require.NoError(t, err) + require.True(t, exists) + require.Len(t, items, 2) + require.Equal(t, "custom_tool_call", gjson.GetBytes(items[1], "type").String()) + require.Equal(t, "call_live", gjson.GetBytes(items[1], "call_id").String()) + }) + t.Run("previous_response_id_full_input_replace", func(t *testing.T) { items, exists, err := buildOpenAIWSReplayInputSequence( lastFull, diff --git a/backend/internal/service/openai_ws_forwarder_payload.go b/backend/internal/service/openai_ws_forwarder_payload.go index 3d6fef10be..c4bd00e3f4 100644 --- a/backend/internal/service/openai_ws_forwarder_payload.go +++ b/backend/internal/service/openai_ws_forwarder_payload.go @@ -604,6 +604,40 @@ func openAIWSRawItemsHaveToolCallContextForOutputs(items []json.RawMessage) bool return true } +func sanitizeOpenAIWSHistoricalReplayToolCalls( + previousItems []json.RawMessage, + currentItems []json.RawMessage, +) []json.RawMessage { + if len(previousItems) == 0 { + return cloneOpenAIWSRawMessages(previousItems) + } + outputCallIDs := make(map[string]struct{}) + collectOutputCallIDs := func(items []json.RawMessage) { + for _, item := range items { + if !isCodexToolCallOutputItemType(gjson.GetBytes(item, "type").String()) { + continue + } + if callID := strings.TrimSpace(gjson.GetBytes(item, "call_id").String()); callID != "" { + outputCallIDs[callID] = struct{}{} + } + } + } + collectOutputCallIDs(previousItems) + collectOutputCallIDs(currentItems) + + sanitized := make([]json.RawMessage, 0, len(previousItems)) + for _, item := range previousItems { + if isCodexToolCallContextItemType(gjson.GetBytes(item, "type").String()) { + callID := strings.TrimSpace(gjson.GetBytes(item, "call_id").String()) + if _, paired := outputCallIDs[callID]; !paired { + continue + } + } + sanitized = append(sanitized, append(json.RawMessage(nil), item...)) + } + return sanitized +} + func openAIWSRawPayloadHasToolCallOutput(payload []byte) bool { if len(payload) == 0 { return false @@ -642,6 +676,7 @@ func buildOpenAIWSReplayInputSequence( if !previousFullInputExists { return cloneOpenAIWSRawMessages(currentItems), currentExists, nil } + previousFullInput = sanitizeOpenAIWSHistoricalReplayToolCalls(previousFullInput, currentItems) if !currentExists || len(currentItems) == 0 { return cloneOpenAIWSRawMessages(previousFullInput), true, nil } diff --git a/backend/internal/service/openai_ws_http_bridge_test.go b/backend/internal/service/openai_ws_http_bridge_test.go index 74838026c6..d224123585 100644 --- a/backend/internal/service/openai_ws_http_bridge_test.go +++ b/backend/internal/service/openai_ws_http_bridge_test.go @@ -423,6 +423,7 @@ func TestOpenAIWSHTTPBridgeFullCustomToolHistoryWithoutPreviousResponseIDDoesNot {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(completed("resp_1", `[{"type":"custom_tool_call","id":"item_1","call_id":"call_1","name":"exec","input":"pwd"}]`)))}, {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(completed("resp_2", `[]`)))}, {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(completed("resp_3", `[]`)))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(completed("resp_4", `[]`)))}, }} cfg := &config.Config{} cfg.Security.URLAllowlist.Enabled = false @@ -485,6 +486,7 @@ func TestOpenAIWSHTTPBridgeFullCustomToolHistoryWithoutPreviousResponseIDDoesNot } writeAndRead(`{"type":"response.create","model":"gpt-5.1","input":"run pwd"}`) + writeAndRead(`{"type":"response.create","model":"gpt-5.1","previous_response_id":"resp_1","input":[{"role":"user","content":"continue without tool output"}]}`) fullContext := `{"type":"response.create","model":"gpt-5.1","input":[{"type":"custom_tool_call","id":"item_1","call_id":"call_1","name":"exec","input":"pwd"},{"type":"custom_tool_call_output","call_id":"call_1","output":"/tmp"},{"role":"user","content":"continue"}]}` writeAndRead(fullContext) writeAndRead(fullContext) @@ -497,8 +499,12 @@ func TestOpenAIWSHTTPBridgeFullCustomToolHistoryWithoutPreviousResponseIDDoesNot t.Fatal("timed out waiting for websocket bridge proxy to finish") } - require.Len(t, upstream.bodies, 3) - for _, body := range upstream.bodies[1:] { + require.Len(t, upstream.bodies, 4) + orphanInput := gjson.GetBytes(upstream.bodies[1], "input").Array() + require.Len(t, orphanInput, 2) + require.Equal(t, "run pwd", orphanInput[0].String()) + require.Equal(t, "user", orphanInput[1].Get("role").String()) + for _, body := range upstream.bodies[2:] { input := gjson.GetBytes(body, "input").Array() require.Len(t, input, 3) require.Equal(t, "custom_tool_call", input[0].Get("type").String())