Merge pull request #4339 from wp-a/codex/read-tool-stream-finalization

fix(apicompat): sanitize complete Read tool argument streams

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)
 	}

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)
 }