From 8f5451587ebbb986169f4d4d1158efde9531246c Mon Sep 17 00:00:00 2001 From: li Date: Mon, 31 Aug 2026 16:15:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(apicompat):=20Anthropic=E2=86=92Responses?= =?UTF-8?q?=20=E6=B5=81=E5=BC=8F=E8=BD=AC=E6=8D=A2=E5=9C=A8=20thinking=20?= =?UTF-8?q?=E5=9D=97=E5=89=8D=E5=85=B3=E9=97=AD=20message=20item=20?= =?UTF-8?q?=E5=B9=B6=E6=8E=A8=E8=BF=9B=20content=5Findex?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit thinking 块开始时不关闭已打开的 message item，直接覆盖 CurrentItemType/ CurrentItemID，累积的助手文本既不进 state.Outputs 也拿不到 output_item.done， response.completed 只带 reasoning——客户端收到「成功但无输出」。同 switch 的 tool_use 分支本来就先调 closeCurrentResponsesItem，只有 thinking 漏了。 content_index 只在 item 打开时置 0，从不推进。同一 message item 里第二个 text 块会再发一次 content_part.added(content_index=0)，与第一个 part 撞在同一下标， SDK 累积式 helper 按 content[content_index] 写入时后者覆盖前者。反方向 responses_to_anthropic.go 有三处 ContentBlockIndex++，正方向一处都没有。 新增用例覆盖两条坏顺序，外加一条生命周期不变式用例（output_item.added 与 output_item.done 必须配平、terminal 事件按announce 顺序逐项携带）。 --- .../anthropic_to_responses_response.go | 27 ++- .../anthropic_to_responses_stream_test.go | 209 ++++++++++++++++++ 2 files changed, 234 insertions(+), 2 deletions(-) diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go index d5a7057036..43542dfaaf 100644 --- a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go +++ b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go @@ -280,6 +280,16 @@ func anthToResHandleContentBlockStart(evt *AnthropicStreamEvent, state *Anthropi switch evt.ContentBlock.Type { case "thinking": + // 开新 item 前必须先关掉在开的那个，与下面的 tool_use 分支一致。 + // 一个 message item 在它的 text 块 content_block_stop 时是刻意保持打开的 + // （同一 item 里可能还有后续 text 块），所以 thinking 块到来时它仍然开着： + // 不关就直接被 CurrentItemType/CurrentItemID 覆盖，累积在 CurrentContent + // 里的助手文本既不会进 state.Outputs，也拿不到 output_item.done， + // response.completed 于是只带 reasoning——客户端看到的是「成功但无输出」。 + // 交错思考（interleaved-thinking，本仓库在 anthropic-beta 透传里明确支持） + // 会稳定产生 text → thinking 这个顺序。 + events = append(events, closeCurrentResponsesItem(state)...) + state.CurrentItemID = generateItemID() state.CurrentItemType = "reasoning" state.ContentIndex = 0 @@ -295,6 +305,11 @@ func anthToResHandleContentBlockStart(evt *AnthropicStreamEvent, state *Anthropi case "text": // If we don't have an open message item, open one if state.CurrentItemType != "message" { + // 走到这里时 CurrentItemType 只可能是 ""（前一个 reasoning/function_call + // 已在自己的 content_block_stop 里关闭）；保留这次关闭是为了让三个分支 + // 的「开新 item 前先关旧的」保持同一条不变式，而不是留一个仅 text 例外。 + events = append(events, closeCurrentResponsesItem(state)...) + state.CurrentItemID = generateItemID() state.CurrentItemType = "message" state.ContentIndex = 0 @@ -434,17 +449,25 @@ func anthToResHandleContentBlockStop(evt *AnthropicStreamEvent, state *Anthropic // item itself stays open since more blocks may follow. text := state.TextAccum state.TextAccum = "" + contentIndex := state.ContentIndex state.CurrentContent = append(state.CurrentContent, ResponsesContentPart{Type: "output_text", Text: text}) + // 关掉一个 part 就推进 content_index：上面那句注释说的「item 保持打开， + // 因为后面可能还有块」正是这里的触发条件。不推进的话，同一 item 里第二个 + // text 块会再发一次 content_part.added(content_index=0)，与第一个 part 撞在 + // 同一下标上——SDK 的累积式 stream helper 按 content[content_index] 写入， + // 后一个 part 直接覆盖前一个，可见文本丢失。 + // 只在这里推进：新 item 的 content_index 由 closeCurrentResponsesItem 归 0。 + state.ContentIndex++ return []ResponsesStreamEvent{ makeResponsesEvent(state, "response.output_text.done", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex, - ContentIndex: state.ContentIndex, + ContentIndex: contentIndex, ItemID: state.CurrentItemID, Text: text, }), makeResponsesEvent(state, "response.content_part.done", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex, - ContentIndex: state.ContentIndex, + ContentIndex: contentIndex, ItemID: state.CurrentItemID, Part: &ResponsesContentPart{Type: "output_text", Text: text}, }), diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go b/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go index c26e289822..e21f4175ef 100644 --- a/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go +++ b/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go @@ -185,3 +185,212 @@ func TestAnthropicEventToResponses_ToolCallCompletedCarriesArguments(t *testing. t.Errorf("name = %q, want get_weather", fc.Name) } } + +// TestAnthropicEventToResponses_ThinkingAfterTextKeepsMessageOutput pins that a +// thinking block arriving after a text block closes the open message item +// instead of silently replacing it. +// +// Why: a message item is deliberately left open when its text block stops +// (more text blocks may follow in the same item). content_block_start therefore +// has to close whatever is open before starting a new item — tool_use already +// did, thinking did not, so it overwrote CurrentItemType/CurrentItemID and the +// accumulated CurrentContent never reached state.Outputs. response.completed +// then carried only the reasoning item and the client saw a successful response +// with no assistant text. Interleaved thinking (anthropic-beta +// interleaved-thinking-2025-05-14, forwarded by this gateway) produces exactly +// this text → thinking ordering. +func TestAnthropicEventToResponses_ThinkingAfterTextKeepsMessageOutput(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var events []ResponsesStreamEvent + feed := func(evt *AnthropicStreamEvent) { + events = append(events, AnthropicEventToResponsesEvents(evt, state)...) + } + + i0, i1 := 0, 1 + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1"}}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &i0, ContentBlock: &AnthropicContentBlock{Type: "text"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &i0, Delta: &AnthropicDelta{Type: "text_delta", Text: "answer"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &i0}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &i1, ContentBlock: &AnthropicContentBlock{Type: "thinking"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &i1, Delta: &AnthropicDelta{Type: "thinking_delta", Thinking: "hmm"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &i1}) + feed(&AnthropicStreamEvent{Type: "message_stop"}) + + var completed *ResponsesStreamEvent + for i := range events { + if events[i].Type == "response.completed" { + completed = &events[i] + } + } + if completed == nil || completed.Response == nil { + t.Fatalf("response.completed was not emitted") + } + outputs := completed.Response.Output + if len(outputs) != 2 { + t.Fatalf("response.completed carries %d output items, want 2 (message + reasoning): %+v", len(outputs), outputs) + } + if outputs[0].Type != "message" { + t.Fatalf("output[0].type = %q, want message", outputs[0].Type) + } + if len(outputs[0].Content) != 1 || outputs[0].Content[0].Text != "answer" { + t.Errorf("assistant text lost: output[0].content = %+v", outputs[0].Content) + } + if outputs[1].Type != "reasoning" { + t.Errorf("output[1].type = %q, want reasoning", outputs[1].Type) + } + + // The two items must also occupy distinct output_index values; sharing one + // index is how the reasoning item used to land on top of the message item. + var addedIndexes []int + for _, evt := range events { + if evt.Type == "response.output_item.added" { + addedIndexes = append(addedIndexes, evt.OutputIndex) + } + } + if len(addedIndexes) != 2 || addedIndexes[0] == addedIndexes[1] { + t.Errorf("output_item.added indexes = %v, want two distinct values", addedIndexes) + } +} + +// TestAnthropicEventToResponses_MultipleTextBlocksAdvanceContentIndex pins that +// each text block within one message item gets its own content_index. +// +// Why: content_index was assigned when the item opened and never advanced, so a +// second text block re-emitted content_part.added at index 0. The OpenAI SDK's +// accumulating stream helper writes parts at output.content[content_index], so +// the second part overwrote the first and the earlier text disappeared from the +// accumulated response. +func TestAnthropicEventToResponses_MultipleTextBlocksAdvanceContentIndex(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var events []ResponsesStreamEvent + feed := func(evt *AnthropicStreamEvent) { + events = append(events, AnthropicEventToResponsesEvents(evt, state)...) + } + + i0, i1 := 0, 1 + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1"}}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &i0, ContentBlock: &AnthropicContentBlock{Type: "text"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &i0, Delta: &AnthropicDelta{Type: "text_delta", Text: "first"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &i0}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &i1, ContentBlock: &AnthropicContentBlock{Type: "text"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &i1, Delta: &AnthropicDelta{Type: "text_delta", Text: "second"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &i1}) + feed(&AnthropicStreamEvent{Type: "message_stop"}) + + var partAdded []int + for _, evt := range events { + if evt.Type == "response.content_part.added" { + partAdded = append(partAdded, evt.ContentIndex) + } + } + if len(partAdded) != 2 { + t.Fatalf("content_part.added emitted %d times, want 2", len(partAdded)) + } + if partAdded[0] != 0 || partAdded[1] != 1 { + t.Errorf("content_part.added indexes = %v, want [0 1]", partAdded) + } + + // Every delta/done for a part must carry that part's index, otherwise the + // SDK appends the text to the wrong slot. + byIndex := map[int]string{} + for _, evt := range events { + if evt.Type == "response.output_text.delta" { + byIndex[evt.ContentIndex] += evt.Delta + } + } + if byIndex[0] != "first" || byIndex[1] != "second" { + t.Errorf("output_text.delta grouped by content_index = %v, want {0:first 1:second}", byIndex) + } +} + +// TestAnthropicEventToResponses_ItemLifecycleIsBalanced is the invariant behind +// both regressions above: the converter must never leave an item that was +// announced with response.output_item.added without a matching +// response.output_item.done, and the terminal event must carry one output entry +// per announced item. Any future block type that forgets to close the open item +// breaks this, whatever the symptom looks like. +func TestAnthropicEventToResponses_ItemLifecycleIsBalanced(t *testing.T) { + for _, tc := range []struct { + name string + blocks []*AnthropicContentBlock + }{ + {"text then thinking", []*AnthropicContentBlock{{Type: "text"}, {Type: "thinking"}}}, + {"thinking then text", []*AnthropicContentBlock{{Type: "thinking"}, {Type: "text"}}}, + {"text then tool_use", []*AnthropicContentBlock{{Type: "text"}, {Type: "tool_use", ID: "toolu_1", Name: "t"}}}, + {"text thinking text", []*AnthropicContentBlock{{Type: "text"}, {Type: "thinking"}, {Type: "text"}}}, + {"thinking text tool_use", []*AnthropicContentBlock{{Type: "thinking"}, {Type: "text"}, {Type: "tool_use", ID: "toolu_2", Name: "t"}}}, + } { + t.Run(tc.name, func(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var events []ResponsesStreamEvent + feed := func(evt *AnthropicStreamEvent) { + events = append(events, AnthropicEventToResponsesEvents(evt, state)...) + } + + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1"}}) + for i, block := range tc.blocks { + idx := i + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &idx, ContentBlock: block}) + switch block.Type { + case "text": + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "text_delta", Text: "t"}}) + case "thinking": + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "thinking_delta", Thinking: "r"}}) + case "tool_use": + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "input_json_delta", PartialJSON: "{}"}}) + } + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &idx}) + } + feed(&AnthropicStreamEvent{Type: "message_stop"}) + + openIDs := map[string]int{} + var order []string + for _, evt := range events { + switch evt.Type { + case "response.output_item.added": + if evt.Item == nil { + t.Fatalf("output_item.added without item") + } + openIDs[evt.Item.ID]++ + order = append(order, evt.Item.ID) + case "response.output_item.done": + if evt.Item == nil { + t.Fatalf("output_item.done without item") + } + openIDs[evt.Item.ID]-- + } + } + for id, n := range openIDs { + if n != 0 { + t.Errorf("item %s: output_item.added/done imbalance %+d", id, n) + } + } + + var completed *ResponsesStreamEvent + for i := range events { + if events[i].Type == "response.completed" { + completed = &events[i] + } + } + if completed == nil || completed.Response == nil { + t.Fatalf("response.completed was not emitted") + } + if len(completed.Response.Output) != len(order) { + t.Fatalf("response.completed carries %d outputs, want %d (one per announced item)", + len(completed.Response.Output), len(order)) + } + for i, id := range order { + if completed.Response.Output[i].ID != id { + t.Errorf("output[%d].id = %q, want %q (announcement order must be preserved)", + i, completed.Response.Output[i].ID, id) + } + } + }) + } +}