From 9b4a7d5d47af2d960364b8830e02d931cf4732a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E9=B9=8F?= <2829624376@qq.com> Date: Wed, 15 Jul 2026 14:08:07 +0800 Subject: [PATCH] fix(apicompat): sanitize complete Read tool argument streams --- .../pkg/apicompat/responses_to_anthropic.go | 25 ++++ .../responses_to_anthropic_read_tool_test.go | 115 +++++++++++------- 2 files changed, 98 insertions(+), 42 deletions(-) diff --git a/backend/internal/pkg/apicompat/responses_to_anthropic.go b/backend/internal/pkg/apicompat/responses_to_anthropic.go index 2863fe7302..455dd1a444 100644 --- a/backend/internal/pkg/apicompat/responses_to_anthropic.go +++ b/backend/internal/pkg/apicompat/responses_to_anthropic.go @@ -413,6 +413,28 @@ func resToAnthHandleFuncArgsDelta(evt *ResponsesStreamEvent, state *ResponsesEve return nil } + if state.CurrentBlockType == "tool_use" && state.CurrentToolName == "Read" { + state.CurrentToolArgs += evt.Delta + if state.CurrentToolHadDelta || !json.Valid([]byte(state.CurrentToolArgs)) { + return nil + } + + blockIdx, ok := state.OutputIndexToBlockIdx[evt.OutputIndex] + if !ok { + return nil + } + state.CurrentToolHadDelta = true + sanitized := sanitizeAnthropicToolUseInput(state.CurrentToolName, state.CurrentToolArgs) + return []AnthropicStreamEvent{{ + Type: "content_block_delta", + Index: &blockIdx, + Delta: &AnthropicDelta{ + Type: "input_json_delta", + PartialJSON: string(sanitized), + }, + }} + } + if state.CurrentBlockType == "tool_use" { state.CurrentToolHadDelta = true } @@ -433,6 +455,9 @@ func resToAnthHandleFuncArgsDelta(evt *ResponsesStreamEvent, state *ResponsesEve } func resToAnthHandleFuncArgsDone(evt *ResponsesStreamEvent, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent { + if !state.ContentBlockOpen { + return nil + } if state.CurrentBlockType != "tool_use" { return resToAnthHandleBlockDone(state) } diff --git a/backend/internal/pkg/apicompat/responses_to_anthropic_read_tool_test.go b/backend/internal/pkg/apicompat/responses_to_anthropic_read_tool_test.go index 72b60099fe..0fc45b7249 100644 --- a/backend/internal/pkg/apicompat/responses_to_anthropic_read_tool_test.go +++ b/backend/internal/pkg/apicompat/responses_to_anthropic_read_tool_test.go @@ -7,63 +7,93 @@ import ( "github.com/stretchr/testify/require" ) -func TestResToAnthFuncArgsDelta_ReadToolStreamsDeltas(t *testing.T) { +func TestResToAnthFuncArgsDelta_ReadToolWaitsForCompleteJSON(t *testing.T) { state := NewResponsesEventToAnthropicState() state.MessageStartSent = true - state.CurrentBlockType = "tool_use" - state.CurrentToolName = "Read" - state.OutputIndexToBlockIdx = map[int]int{0: 0} - - evt := &ResponsesStreamEvent{ - Type: "response.function_call_arguments.delta", - OutputIndex: 0, - Delta: `{"file_path":"/tmp/test.go"}`, - } - - events := ResponsesEventToAnthropicEvents(evt, state) - - require.Len(t, events, 1, "Read tool delta must produce content_block_delta") - assert.Equal(t, "content_block_delta", events[0].Type) - assert.Equal(t, "input_json_delta", events[0].Delta.Type) - assert.Equal(t, `{"file_path":"/tmp/test.go"}`, events[0].Delta.PartialJSON) - assert.True(t, state.CurrentToolHadDelta, "Read deltas should set CurrentToolHadDelta") -} - -func TestResToAnthFuncArgsDelta_ReadToolWithoutDone(t *testing.T) { - state := NewResponsesEventToAnthropicState() - state.MessageStartSent = true - state.ContentBlockIndex = 0 state.ContentBlockOpen = true state.CurrentBlockType = "tool_use" state.CurrentToolName = "Read" state.OutputIndexToBlockIdx = map[int]int{0: 0} - delta := &ResponsesStreamEvent{ + events := ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ Type: "response.function_call_arguments.delta", OutputIndex: 0, - Delta: `{"file_path":"/tmp/test.go"}`, - } - events := ResponsesEventToAnthropicEvents(delta, state) - require.Len(t, events, 1, "delta should be streamed") + Delta: `{"file_path":"/tmp/te`, + }, state) + assert.Empty(t, events, "partial Read JSON must wait for sanitization") + assert.False(t, state.CurrentToolHadDelta) - completed := &ResponsesStreamEvent{ + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.delta", + OutputIndex: 0, + Delta: `st.go","pages":""}`, + }, state) + require.Len(t, events, 1) + assert.Equal(t, "content_block_delta", events[0].Type) + assert.Equal(t, "input_json_delta", events[0].Delta.Type) + assert.JSONEq(t, `{"file_path":"/tmp/test.go"}`, events[0].Delta.PartialJSON) + assert.Equal(t, `{"file_path":"/tmp/test.go","pages":""}`, state.CurrentToolArgs) + assert.True(t, state.CurrentToolHadDelta) + + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", + OutputIndex: 0, + Arguments: `{"file_path":"/tmp/test.go","pages":""}`, + }, state) + require.Len(t, events, 1) + assert.Equal(t, "content_block_stop", events[0].Type) + + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.done", + OutputIndex: 0, + Arguments: `{"file_path":"/tmp/test.go","pages":""}`, + }, state) + assert.Empty(t, events, "duplicate done must be idempotent") +} + +func TestResponsesEventToAnthropicEvents_ReadToolWithoutArgumentsDoneClosesOnCompleted(t *testing.T) { + state := NewResponsesEventToAnthropicState() + + events := ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.created", + Response: &ResponsesResponse{ID: "resp_read", Model: "gpt-5.5"}, + }, state) + require.Len(t, events, 1) + assert.Equal(t, "message_start", events[0].Type) + + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.output_item.added", + OutputIndex: 0, + Item: &ResponsesOutput{Type: "function_call", CallID: "call_read", Name: "Read"}, + }, state) + require.Len(t, events, 1) + assert.Equal(t, "content_block_start", events[0].Type) + + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ + Type: "response.function_call_arguments.delta", + OutputIndex: 0, + Delta: `{"file_path":"/tmp/test.go","pages":""}`, + }, state) + require.Len(t, events, 1) + assert.Equal(t, "content_block_delta", events[0].Type) + assert.Equal(t, "input_json_delta", events[0].Delta.Type) + assert.JSONEq(t, `{"file_path":"/tmp/test.go"}`, events[0].Delta.PartialJSON) + + events = ResponsesEventToAnthropicEvents(&ResponsesStreamEvent{ Type: "response.completed", Response: &ResponsesResponse{ Status: "completed", }, - } - events = ResponsesEventToAnthropicEvents(completed, state) - - hasStop := false - for _, e := range events { - if e.Type == "content_block_stop" { - hasStop = true - } - } - assert.True(t, hasStop, "block should be closed even without .done event") + }, state) + require.Len(t, events, 3) + assert.Equal(t, "content_block_stop", events[0].Type) + assert.Equal(t, "message_delta", events[1].Type) + assert.Equal(t, "tool_use", events[1].Delta.StopReason) + assert.Equal(t, "message_stop", events[2].Type) + assert.Empty(t, FinalizeResponsesAnthropicStream(state), "terminal event already finalized the stream") } -func TestResToAnthFuncArgsDelta_NonReadToolUnchanged(t *testing.T) { +func TestResToAnthFuncArgsDelta_NonReadToolStreamsPartialJSONImmediately(t *testing.T) { state := NewResponsesEventToAnthropicState() state.MessageStartSent = true state.CurrentBlockType = "tool_use" @@ -73,12 +103,13 @@ func TestResToAnthFuncArgsDelta_NonReadToolUnchanged(t *testing.T) { evt := &ResponsesStreamEvent{ Type: "response.function_call_arguments.delta", OutputIndex: 0, - Delta: `{"file_path":"/tmp/out.txt","content":"hello"}`, + Delta: `{"file_path":"/tmp/out`, } events := ResponsesEventToAnthropicEvents(evt, state) require.Len(t, events, 1) assert.Equal(t, "content_block_delta", events[0].Type) + assert.Equal(t, `{"file_path":"/tmp/out`, events[0].Delta.PartialJSON) assert.True(t, state.CurrentToolHadDelta) }