Merge pull request #6925 from reed-yang/fix/recover-responses-text

fix(apicompat): recover Responses text from done and terminal events

backend/internal/pkg/apicompat/responses_to_anthropic.go

@@ -170,6 +170,12 @@ func sanitizeAnthropicToolUseInput(name string, raw string) json.RawMessage {
 // Streaming: ResponsesStreamEvent → []AnthropicStreamEvent (stateful converter)
 // ---------------------------------------------------------------------------
 
+// responsesTextPart identifies one output_text part of a streamed response.
+type responsesTextPart struct {
+	OutputIndex  int
+	ContentIndex int
+}
+
 // ResponsesEventToAnthropicState tracks state for converting a sequence of
 // Responses SSE events directly into Anthropic SSE events.
 type ResponsesEventToAnthropicState struct {
@@ -190,6 +196,13 @@ type ResponsesEventToAnthropicState struct {
 	// OutputIndexToBlockIdx maps Responses output_index → Anthropic content block index.
 	OutputIndexToBlockIdx map[int]int
 
+	// textByPart records the text already delivered for each output_text part
+	// so that a done payload can be reconciled against it. It outlives the
+	// content block; closeCurrentBlock must not reset it.
+	textByPart map[responsesTextPart]*strings.Builder
+	// textDelivered records whether any assistant text reached the client.
+	textDelivered bool
+
 	InputTokens              int
 	OutputTokens             int
 	CacheReadInputTokens     int
@@ -204,6 +217,7 @@ type ResponsesEventToAnthropicState struct {
 func NewResponsesEventToAnthropicState() *ResponsesEventToAnthropicState {
 	return &ResponsesEventToAnthropicState{
 		OutputIndexToBlockIdx: make(map[int]int),
+		textByPart:            make(map[responsesTextPart]*strings.Builder),
 		Created:               time.Now().Unix(),
 	}
 }
@@ -222,7 +236,7 @@ func ResponsesEventToAnthropicEvents(
 	case "response.output_text.delta":
 		return resToAnthHandleTextDelta(evt, state)
 	case "response.output_text.done":
-		return resToAnthHandleBlockDone(state)
+		return resToAnthHandleTextDone(evt, state)
 	case "response.function_call_arguments.delta",
 		// custom/freeform 工具的输入增量与 function_call 参数增量同形。
 		"response.custom_tool_call_input.delta":
@@ -390,7 +404,18 @@ func resToAnthHandleOutputItemAdded(evt *ResponsesStreamEvent, state *ResponsesE
 }
 
 func resToAnthHandleTextDelta(evt *ResponsesStreamEvent, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
-	if evt.Delta == "" {
+	return resToAnthEmitText(evt.Delta, resToAnthTextPartOf(evt), state)
+}
+
+func resToAnthTextPartOf(evt *ResponsesStreamEvent) responsesTextPart {
+	return responsesTextPart{OutputIndex: evt.OutputIndex, ContentIndex: evt.ContentIndex}
+}
+
+// resToAnthEmitText opens a text block when needed, emits text, and records it
+// against its part so that a later payload for the same part is reconciled
+// against what the client already received.
+func resToAnthEmitText(text string, part responsesTextPart, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
+	if text == "" {
 		return nil
 	}
 
@@ -413,18 +438,63 @@ func resToAnthHandleTextDelta(evt *ResponsesStreamEvent, state *ResponsesEventTo
 		})
 	}
 
+	delivered, ok := state.textByPart[part]
+	if !ok {
+		delivered = &strings.Builder{}
+		state.textByPart[part] = delivered
+	}
+	_, _ = delivered.WriteString(text)
+	state.textDelivered = true
+
 	idx := state.ContentBlockIndex
 	events = append(events, AnthropicStreamEvent{
 		Type:  "content_block_delta",
 		Index: &idx,
 		Delta: &AnthropicDelta{
 			Type: "text_delta",
-			Text: evt.Delta,
+			Text: text,
 		},
 	})
 	return events
 }
 
+// resToAnthRecoverText emits the tail of a finished text payload that never
+// reached the client. Streamed text cannot be recalled, so a payload that does
+// not extend what was already delivered is left alone.
+func resToAnthRecoverText(text string, part responsesTextPart, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
+	builder, known := state.textByPart[part]
+	if !known && state.textDelivered {
+		// The payload is indexed differently from every delta seen so far, so
+		// which part it finishes cannot be established. Recovering it could
+		// repeat an answer the client already has, which is worse than leaving
+		// a partially delivered one alone.
+		return nil
+	}
+
+	var delivered string
+	if known {
+		delivered = builder.String()
+	}
+	if text == delivered || !strings.HasPrefix(text, delivered) {
+		return nil
+	}
+	return resToAnthEmitText(text[len(delivered):], part, state)
+}
+
+// resToAnthHandleTextDone recovers text that upstream carried only on the done
+// event before closing the block, which some streams use instead of sending
+// output_text.delta at all.
+func resToAnthHandleTextDone(evt *ResponsesStreamEvent, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
+	if state.MessageStopSent {
+		// The message is already terminated; a late payload cannot be delivered
+		// without emitting a content block after message_stop.
+		return resToAnthHandleBlockDone(state)
+	}
+
+	events := resToAnthRecoverText(evt.Text, resToAnthTextPartOf(evt), state)
+	return append(events, resToAnthHandleBlockDone(state)...)
+}
+
 func resToAnthHandleFuncArgsDelta(evt *ResponsesStreamEvent, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
 	if evt.Delta == "" {
 		return nil
@@ -622,6 +692,38 @@ func resToAnthHandleWebSearchDone(evt *ResponsesStreamEvent, state *ResponsesEve
 	return events
 }
 
+// resToAnthRecoverTerminalText emits assistant text that only ever appeared in
+// the terminal response payload, which some streams populate without sending
+// any output_text event.
+//
+// It only runs when no text at all reached the client. Streamed events and the
+// terminal output array carry no guaranteed common identity, so reconciling
+// them part by part risks repeating an answer the client already has, which is
+// worse than leaving a partially streamed response as it is.
+func resToAnthRecoverTerminalText(evt *ResponsesStreamEvent, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
+	if state.textDelivered || evt.Response == nil {
+		return nil
+	}
+
+	var events []AnthropicStreamEvent
+	for outputIndex, item := range evt.Response.Output {
+		if item.Type != "message" {
+			continue
+		}
+		for contentIndex, content := range item.Content {
+			if content.Type != "output_text" {
+				continue
+			}
+			part := responsesTextPart{OutputIndex: outputIndex, ContentIndex: contentIndex}
+			events = append(events, resToAnthEmitText(content.Text, part, state)...)
+		}
+	}
+	if len(events) > 0 {
+		events = append(events, closeCurrentBlock(state)...)
+	}
+	return events
+}
+
 func resToAnthHandleCompleted(evt *ResponsesStreamEvent, state *ResponsesEventToAnthropicState) []AnthropicStreamEvent {
 	if state.MessageStopSent {
 		return nil
@@ -629,6 +731,7 @@ func resToAnthHandleCompleted(evt *ResponsesStreamEvent, state *ResponsesEventTo
 
 	var events []AnthropicStreamEvent
 	events = append(events, closeCurrentBlock(state)...)
+	events = append(events, resToAnthRecoverTerminalText(evt, state)...)
 
 	stopReason := "end_turn"
 	if evt.Usage != nil {

backend/internal/pkg/apicompat/responses_to_anthropic_text_recovery_test.go

@@ -0,0 +1,532 @@
+package apicompat
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// feedResponsesEvents converts a whole event sequence with a single state.
+func feedResponsesEvents(events ...*ResponsesStreamEvent) []AnthropicStreamEvent {
+	state := NewResponsesEventToAnthropicState()
+	var out []AnthropicStreamEvent
+	for _, evt := range events {
+		out = append(out, ResponsesEventToAnthropicEvents(evt, state)...)
+	}
+	return out
+}
+
+// collectAnthropicText accumulates the visible assistant text a Claude client
+// reconstructs from a converted event stream.
+func collectAnthropicText(events []AnthropicStreamEvent) string {
+	text := ""
+	for _, event := range events {
+		if event.Type == "content_block_delta" && event.Delta != nil && event.Delta.Type == "text_delta" {
+			text += event.Delta.Text
+		}
+	}
+	return text
+}
+
+// requireAnthropicBlockLifecycle checks the block framing a strict client
+// relies on: indices start at zero and increase by one, every block is closed,
+// and no delta arrives outside an open block.
+func requireAnthropicBlockLifecycle(t *testing.T, events []AnthropicStreamEvent) {
+	t.Helper()
+
+	open := false
+	next := 0
+	for _, event := range events {
+		switch event.Type {
+		case "content_block_start":
+			require.False(t, open, "content_block_start while a block is open")
+			require.NotNil(t, event.Index)
+			require.Equal(t, next, *event.Index, "content block indices must increase by one")
+			open = true
+		case "content_block_delta":
+			require.True(t, open, "content_block_delta outside an open block")
+			require.NotNil(t, event.Index)
+			require.Equal(t, next, *event.Index)
+		case "content_block_stop":
+			require.True(t, open, "content_block_stop without an open block")
+			require.NotNil(t, event.Index)
+			require.Equal(t, next, *event.Index)
+			open = false
+			next++
+		}
+	}
+	require.False(t, open, "stream ended with an unclosed content block")
+}
+
+func responsesCreated() *ResponsesStreamEvent {
+	return &ResponsesStreamEvent{
+		Type:     "response.created",
+		Response: &ResponsesResponse{ID: "resp_text_recovery", Model: "gpt-5.2"},
+	}
+}
+
+func responsesMessageOutput(texts ...string) []ResponsesOutput {
+	content := make([]ResponsesContentPart, 0, len(texts))
+	for _, text := range texts {
+		content = append(content, ResponsesContentPart{Type: "output_text", Text: text})
+	}
+	return []ResponsesOutput{{Type: "message", Role: "assistant", Content: content}}
+}
+
+// Some streams carry the finished text only on response.output_text.done and
+// send no output_text.delta at all. Dropping it leaves a protocol-valid message
+// with no text, which clients report as an empty response.
+func TestResponsesEventToAnthropicEvents_RecoversTextFromDoneEvent(t *testing.T) {
+	const answer = "recovered from done"
+
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_item.added", Item: &ResponsesOutput{Type: "message"}},
+		&ResponsesStreamEvent{Type: "response.output_text.done", Text: answer},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, answer, collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+	require.NotEmpty(t, events)
+	assert.Equal(t, "message_stop", events[len(events)-1].Type)
+}
+
+// When nothing streamed at all, the terminal event still carries the finished
+// output. Dropping it produces the same empty turn.
+func TestResponsesEventToAnthropicEvents_RecoversTextFromTerminalOutput(t *testing.T) {
+	const answer = "recovered from terminal"
+
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput(answer),
+		}},
+	)
+
+	assert.Equal(t, answer, collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+	require.NotEmpty(t, events)
+	assert.Equal(t, "message_stop", events[len(events)-1].Type)
+}
+
+// The normal path delivers text through deltas. Neither the done payload nor
+// the terminal payload may repeat it.
+func TestResponsesEventToAnthropicEvents_DoesNotDuplicateTextDeliveredByDeltas(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", Delta: "Hel"},
+		&ResponsesStreamEvent{Type: "response.output_text.delta", Delta: "lo"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", Text: "Hello"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput("Hello"),
+		}},
+	)
+
+	assert.Equal(t, "Hello", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Text recovered from a done event must not be repeated by the terminal
+// payload that also carries it.
+func TestResponsesEventToAnthropicEvents_DoesNotDuplicateTextRecoveredFromDoneEvent(t *testing.T) {
+	const answer = "recovered once"
+
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.done", Text: answer},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput(answer),
+		}},
+	)
+
+	assert.Equal(t, answer, collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Deduplication is per output_text part, not per content block: an event that
+// closes the block between the deltas and the matching done must not make the
+// done payload look undelivered.
+func TestResponsesEventToAnthropicEvents_DoesNotDuplicateTextWhenBlockClosedBeforeDone(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", Delta: "Hello"},
+		&ResponsesStreamEvent{
+			Type:        "response.output_item.added",
+			OutputIndex: 1,
+			Item:        &ResponsesOutput{Type: "function_call", CallID: "call_1", Name: "lookup"},
+		},
+		&ResponsesStreamEvent{Type: "response.output_text.done", Text: "Hello"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, "Hello", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// When only part of the text streamed, the done payload is authoritative and
+// the missing suffix must still reach the client.
+func TestResponsesEventToAnthropicEvents_AppendsOnlySuffixMissingFromDeltas(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", Delta: "Hel"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", Text: "Hello world"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput("Hello world"),
+		}},
+	)
+
+	assert.Equal(t, "Hello world", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Streamed text cannot be recalled, so a payload that contradicts what was
+// already delivered is left alone rather than appended.
+func TestResponsesEventToAnthropicEvents_IgnoresDonePayloadDivergingFromDeltas(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", Delta: "Hello"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", Text: "Goodbye"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, "Hello", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Streamed events and the terminal output array share no guaranteed identity,
+// so once any text has been delivered the terminal payload is left alone. An
+// upstream that numbers its streamed items differently from its terminal array
+// must not make a delivered answer look undelivered.
+func TestResponsesEventToAnthropicEvents_DoesNotDuplicateWhenTerminalOutputIsIndexedDifferently(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{
+			Type:        "response.output_item.added",
+			OutputIndex: 0,
+			Item:        &ResponsesOutput{Type: "reasoning", ID: "rs_1"},
+		},
+		&ResponsesStreamEvent{Type: "response.output_text.delta", OutputIndex: 1, Delta: "Hello"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			// The reasoning item is absent here, so the message sits at array
+			// position 0 while it streamed at output_index 1.
+			Output: responsesMessageOutput("Hello"),
+		}},
+	)
+
+	assert.Equal(t, "Hello", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Partially streamed responses keep exactly what streamed. Recovering the rest
+// would mean matching two index spaces that upstreams do not keep aligned.
+func TestResponsesEventToAnthropicEvents_LeavesTerminalOutputAloneOnceTextStreamed(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", ContentIndex: 0, Delta: "first"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", ContentIndex: 0, Text: "first"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput("first", "second"),
+		}},
+	)
+
+	assert.Equal(t, "first", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// When nothing streamed, every output_text part of the terminal payload is
+// recovered, merged into one block the way contiguous streamed text is.
+func TestResponsesEventToAnthropicEvents_RecoversEveryTerminalPartWhenNothingStreamed(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput("first", "second"),
+		}},
+	)
+
+	assert.Equal(t, "firstsecond", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+
+	starts := 0
+	for _, event := range events {
+		if event.Type == "content_block_start" {
+			starts++
+		}
+	}
+	assert.Equal(t, 1, starts, "contiguous recovered text belongs in one block")
+}
+
+// A turn that produced only a tool call must stay text-free: the terminal
+// payload carries no message item and nothing may be synthesised.
+func TestResponsesEventToAnthropicEvents_LeavesToolOnlyTurnTextFree(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{
+			Type:        "response.output_item.added",
+			OutputIndex: 0,
+			Item:        &ResponsesOutput{Type: "function_call", CallID: "call_1", Name: "lookup"},
+		},
+		&ResponsesStreamEvent{
+			Type:        "response.function_call_arguments.done",
+			OutputIndex: 0,
+			Arguments:   `{"id":1}`,
+		},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: []ResponsesOutput{{Type: "function_call", CallID: "call_1", Name: "lookup", Arguments: `{"id":1}`}},
+		}},
+	)
+
+	assert.Empty(t, collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Recovery must behave the same on every terminal alias the converter accepts,
+// not only on response.completed.
+func TestResponsesEventToAnthropicEvents_RecoversTextOnEveryTerminalAlias(t *testing.T) {
+	for _, terminal := range []string{
+		"response.completed",
+		"response.done",
+		"response.incomplete",
+		"response.failed",
+	} {
+		t.Run(terminal, func(t *testing.T) {
+			const answer = "recovered on a terminal alias"
+
+			events := feedResponsesEvents(
+				responsesCreated(),
+				&ResponsesStreamEvent{Type: terminal, Response: &ResponsesResponse{
+					Status: "completed",
+					Output: responsesMessageOutput(answer),
+				}},
+			)
+
+			assert.Equal(t, answer, collectAnthropicText(events))
+			requireAnthropicBlockLifecycle(t, events)
+			require.NotEmpty(t, events)
+			assert.Equal(t, "message_stop", events[len(events)-1].Type)
+		})
+	}
+}
+
+// A done payload indexed differently from every delta seen so far cannot be
+// matched to a part. Emitting it anyway would repeat an answer the client
+// already has, which is the failure the terminal guard also exists to prevent.
+func TestResponsesEventToAnthropicEvents_DoesNotDuplicateWhenDoneIsIndexedDifferently(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		done *ResponsesStreamEvent
+	}{
+		{"content index diverges", &ResponsesStreamEvent{Type: "response.output_text.done", ContentIndex: 1, Text: "Hello world"}},
+		{"output index diverges", &ResponsesStreamEvent{Type: "response.output_text.done", OutputIndex: 1, Text: "Hello world"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			events := feedResponsesEvents(
+				responsesCreated(),
+				&ResponsesStreamEvent{Type: "response.output_text.delta", Delta: "Hello world"},
+				tc.done,
+				&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+			)
+
+			assert.Equal(t, "Hello world", collectAnthropicText(events))
+			requireAnthropicBlockLifecycle(t, events)
+		})
+	}
+}
+
+// The same rule holds when only part of the text streamed: an unmatchable
+// payload is left alone rather than appended on top of what was delivered.
+func TestResponsesEventToAnthropicEvents_LeavesUnmatchableDoneAloneAfterPartialDeltas(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", ContentIndex: 0, Delta: "Hello"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", ContentIndex: 1, Text: "Hello world"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, "Hello", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Delivered text is keyed by output index and content index together. Keying on
+// either alone would make one of these parts look already delivered.
+func TestResponsesEventToAnthropicEvents_KeysDeliveredTextByBothIndices(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.output_text.delta", OutputIndex: 0, ContentIndex: 0, Delta: "first"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", OutputIndex: 0, ContentIndex: 0, Text: "first"},
+		&ResponsesStreamEvent{Type: "response.output_text.delta", OutputIndex: 0, ContentIndex: 1, Delta: "second"},
+		&ResponsesStreamEvent{Type: "response.output_text.done", OutputIndex: 0, ContentIndex: 1, Text: "second-tail"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, "firstsecond-tail", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// A stray done event after the terminal one must not resurrect a stopped
+// message. The baseline dropped it outright; recovery has to keep that, or a
+// client sees a content block after message_stop.
+func TestResponsesEventToAnthropicEvents_IgnoresDoneEventAfterMessageStop(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: responsesMessageOutput("the answer"),
+		}},
+		&ResponsesStreamEvent{Type: "response.output_text.done", OutputIndex: 7, Text: "late text"},
+	)
+
+	assert.Equal(t, "the answer", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+	require.NotEmpty(t, events)
+	assert.Equal(t, "message_stop", events[len(events)-1].Type, "no content may follow message_stop")
+}
+
+// The tool path reads arguments, never text. A tool done event carrying a
+// stray text field must not produce a text block.
+func TestResponsesEventToAnthropicEvents_LeavesToolArgumentsDoneTextFree(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{
+			Type:        "response.output_item.added",
+			OutputIndex: 0,
+			Item:        &ResponsesOutput{Type: "function_call", CallID: "call_1", Name: "lookup"},
+		},
+		&ResponsesStreamEvent{
+			Type:        "response.function_call_arguments.done",
+			OutputIndex: 0,
+			Arguments:   `{"id":1}`,
+			Text:        "must not be emitted",
+		},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Empty(t, collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Recovery closes whatever block is open through the shared helper, which owns
+// signature emission. A thinking block must still get its signature_delta
+// before its stop.
+func TestResponsesEventToAnthropicEvents_PreservesThinkingSignatureWhenRecovering(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{
+			Type:        "response.output_item.added",
+			OutputIndex: 0,
+			Item:        &ResponsesOutput{Type: "reasoning", EncryptedContent: "sig-abc"},
+		},
+		&ResponsesStreamEvent{Type: "response.output_text.done", OutputIndex: 1, Text: "recovered"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, "recovered", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+
+	var signatures, stopsBefore int
+	for _, event := range events {
+		if event.Type == "content_block_delta" && event.Delta != nil && event.Delta.Type == "signature_delta" {
+			assert.Equal(t, "sig-abc", event.Delta.Signature)
+			signatures++
+			assert.Equal(t, 0, stopsBefore, "signature_delta must precede the thinking block stop")
+		}
+		if event.Type == "content_block_stop" {
+			stopsBefore++
+		}
+	}
+	assert.Equal(t, 1, signatures, "the thinking signature must survive recovery")
+}
+
+// A stream that ends without any terminal event exits through the synthetic
+// finalizer, which must still close a block recovery opened.
+func TestResponsesEventToAnthropicEvents_RecoveredTextSurvivesTheSyntheticFinalizer(t *testing.T) {
+	state := NewResponsesEventToAnthropicState()
+	var events []AnthropicStreamEvent
+	for _, evt := range []*ResponsesStreamEvent{
+		responsesCreated(),
+		{Type: "response.output_text.done", Text: "recovered before the stream died"},
+	} {
+		events = append(events, ResponsesEventToAnthropicEvents(evt, state)...)
+	}
+	events = append(events, FinalizeResponsesAnthropicStream(state)...)
+
+	assert.Equal(t, "recovered before the stream died", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+	require.NotEmpty(t, events)
+	assert.Equal(t, "message_stop", events[len(events)-1].Type)
+}
+
+// The terminal walk only takes output_text parts of message items.
+func TestResponsesEventToAnthropicEvents_DeclinesTerminalRecoveryWithoutMessageText(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		response *ResponsesResponse
+	}{
+		{"no response payload", nil},
+		{"no message item", &ResponsesResponse{Status: "completed", Output: []ResponsesOutput{{Type: "reasoning"}}}},
+		{"message without output_text", &ResponsesResponse{Status: "completed", Output: []ResponsesOutput{{
+			Type:    "message",
+			Role:    "assistant",
+			Content: []ResponsesContentPart{{Type: "refusal", Text: "refused"}},
+		}}}},
+		{"empty output_text", &ResponsesResponse{Status: "completed", Output: responsesMessageOutput("")}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			events := feedResponsesEvents(
+				responsesCreated(),
+				&ResponsesStreamEvent{Type: "response.completed", Response: tc.response},
+			)
+
+			assert.Empty(t, collectAnthropicText(events))
+			requireAnthropicBlockLifecycle(t, events)
+		})
+	}
+}
+
+// Every message item of the terminal payload is recovered, and non-message
+// items between them are skipped without breaking the merged block.
+func TestResponsesEventToAnthropicEvents_RecoversEveryTerminalMessageItem(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{
+			Status: "completed",
+			Output: []ResponsesOutput{
+				{Type: "reasoning"},
+				{Type: "message", Role: "assistant", Content: []ResponsesContentPart{{Type: "output_text", Text: "first"}}},
+				{Type: "function_call", CallID: "call_1", Name: "lookup"},
+				{Type: "message", Role: "assistant", Content: []ResponsesContentPart{{Type: "output_text", Text: "second"}}},
+			},
+		}},
+	)
+
+	assert.Equal(t, "firstsecond", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}
+
+// Text that upstream delivers only on a done event still reaches the client
+// when a tool block is open, which the baseline dropped. The tool block is
+// closed first, so later block indices shift by one.
+func TestResponsesEventToAnthropicEvents_EmitsDoneTextWhileAToolBlockIsOpen(t *testing.T) {
+	events := feedResponsesEvents(
+		responsesCreated(),
+		&ResponsesStreamEvent{
+			Type:        "response.output_item.added",
+			OutputIndex: 0,
+			Item:        &ResponsesOutput{Type: "function_call", CallID: "call_1", Name: "lookup"},
+		},
+		&ResponsesStreamEvent{Type: "response.output_text.done", OutputIndex: 1, Text: "spoken after the call"},
+		&ResponsesStreamEvent{Type: "response.completed", Response: &ResponsesResponse{Status: "completed"}},
+	)
+
+	assert.Equal(t, "spoken after the call", collectAnthropicText(events))
+	requireAnthropicBlockLifecycle(t, events)
+}