fix(apicompat): Anthropic→Responses 流式转换在 thinking 块前关闭 message item 并推进 content_index

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 顺序逐项携带)。
This commit is contained in:
li
2026-08-31 16:15:08 +08:00
parent 85b593fd2f
commit 8f5451587e
2 changed files with 234 additions and 2 deletions
@@ -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},
}),
@@ -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)
}
}
})
}
}