From 26b5e8745782abe1fa41a2ebfc0b95e7e7d7e2b3 Mon Sep 17 00:00:00 2001 From: docooler Date: Fri, 17 Jul 2026 03:35:26 +0000 Subject: [PATCH] fix(apicompat): emit content_part events and full output in Responses stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Anthropic→Responses streaming converter omits two things the OpenAI Responses wire format requires, which breaks clients that reconstruct the response from the event stream (rather than just reading deltas). 1. response.content_part.added is never emitted. A message item is opened with content: [], and the OpenAI SDK's accumulating stream helper (client.responses.stream) only appends a content part when it sees content_part.added. Without it, the next output_text.delta indexes output.content[content_index] and raises: File "openai/lib/streaming/responses/_responses.py", line 352, in accumulate_event content = output.content[event.content_index] IndexError: list index out of range Raw iteration (responses.create(stream=True)) does not accumulate and is unaffected, which is why this went unnoticed. 2. response.completed carries Output: []ResponsesOutput{}. get_final_response() and tracing integrations parse the terminal event's response directly, so callers see an empty output_text even though the deltas streamed correctly. This one is invisible when only watching the stream render. Also carries the full text on output_text.done / content_part.done (deltas carry increments only, done events carry the whole part) and fills in content/arguments/summary on output_item.done, which had the same empty-payload issue. Reproduced against a live Anthropic-platform group with the openai Python SDK 2.46.0; Arize Phoenix's playground hits the same path. Verified before (IndexError) and after (full text via get_final_response()). Adds regression tests covering event ordering, done-event payloads, and the terminal event's output for both text and tool calls. --- .../anthropic_to_responses_response.go | 95 ++++++++- .../anthropic_to_responses_stream_test.go | 187 ++++++++++++++++++ 2 files changed, 273 insertions(+), 9 deletions(-) create mode 100644 backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go index 9ca20e1cd..b22215836 100644 --- a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go +++ b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go @@ -152,11 +152,25 @@ type AnthropicEventToResponsesState struct { // For message output: accumulate text parts ContentIndex int + // TextAccum accumulates the current text part so that output_text.done and + // content_part.done can carry the full text (deltas carry increments only). + TextAccum string // For function_call: track per-output info CurrentCallID string CurrentName string + // Content of the currently open item, folded into Outputs when it closes. + CurrentContent []ResponsesContentPart // message + CurrentArgs string // function_call + CurrentSummary string // reasoning + + // Outputs accumulates every closed output item so that response.completed + // can carry the full output list. The OpenAI SDK's get_final_response() + // parses the terminal event's response directly; without this, clients see + // an empty output_text. + Outputs []ResponsesOutput + // Usage from message_start / message_delta. InputTokens here follows // Anthropic semantics (excludes cached tokens); they are added back when // emitting the OpenAI Responses usage. @@ -293,6 +307,22 @@ func anthToResHandleContentBlockStart(evt *AnthropicStreamEvent, state *Anthropi })) } + // response.content_part.added must precede the output_text.delta events + // for that part. The message item is added with content: [], and the + // OpenAI SDK's accumulating stream helper (client.responses.stream) only + // appends a content part when it sees content_part.added. Without it the + // following output_text.delta indexes output.content[content_index] and + // raises IndexError. Raw event iteration + // (responses.create(stream=True)) does not accumulate, which is why this + // went unnoticed. + events = append(events, makeResponsesEvent(state, "response.content_part.added", &ResponsesStreamEvent{ + OutputIndex: state.OutputIndex, + ContentIndex: state.ContentIndex, + ItemID: state.CurrentItemID, + Part: &ResponsesContentPart{Type: "output_text", Text: ""}, + })) + state.TextAccum = "" + case "tool_use": // Close previous item if any events = append(events, closeCurrentResponsesItem(state)...) @@ -327,6 +357,7 @@ func anthToResHandleContentBlockDelta(evt *AnthropicStreamEvent, state *Anthropi if evt.Delta.Text == "" { return nil } + state.TextAccum += evt.Delta.Text return []ResponsesStreamEvent{makeResponsesEvent(state, "response.output_text.delta", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex, ContentIndex: state.ContentIndex, @@ -338,6 +369,7 @@ func anthToResHandleContentBlockDelta(evt *AnthropicStreamEvent, state *Anthropi if evt.Delta.Thinking == "" { return nil } + state.CurrentSummary += evt.Delta.Thinking return []ResponsesStreamEvent{makeResponsesEvent(state, "response.reasoning_summary_text.delta", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex, SummaryIndex: 0, @@ -349,6 +381,7 @@ func anthToResHandleContentBlockDelta(evt *AnthropicStreamEvent, state *Anthropi if evt.Delta.PartialJSON == "" { return nil } + state.CurrentArgs += evt.Delta.PartialJSON return []ResponsesStreamEvent{makeResponsesEvent(state, "response.function_call_arguments.delta", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex, Delta: evt.Delta.PartialJSON, @@ -393,12 +426,24 @@ func anthToResHandleContentBlockStop(evt *AnthropicStreamEvent, state *Anthropic return events case "message": - // Emit output_text.done (text block is done, but message item stays open for potential more blocks) + // Text block is done: emit output_text.done then content_part.done (the + // order OpenAI uses), both carrying the part's full text. The message + // item itself stays open since more blocks may follow. + text := state.TextAccum + state.TextAccum = "" + state.CurrentContent = append(state.CurrentContent, ResponsesContentPart{Type: "output_text", Text: text}) return []ResponsesStreamEvent{ makeResponsesEvent(state, "response.output_text.done", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex, ContentIndex: state.ContentIndex, ItemID: state.CurrentItemID, + Text: text, + }), + makeResponsesEvent(state, "response.content_part.done", &ResponsesStreamEvent{ + OutputIndex: state.OutputIndex, + ContentIndex: state.ContentIndex, + ItemID: state.CurrentItemID, + Part: &ResponsesContentPart{Type: "output_text", Text: text}, }), } } @@ -454,24 +499,48 @@ func closeCurrentResponsesItem(state *AnthropicEventToResponsesState) []Response return nil } - itemType := state.CurrentItemType - itemID := state.CurrentItemID + // Assemble the full item: both output_item.done and response.completed must + // carry its content. Emitting only {type,id,status} makes SDK-side + // accumulation produce an empty output. + item := ResponsesOutput{ + Type: state.CurrentItemType, + ID: state.CurrentItemID, + Status: "completed", + } + switch state.CurrentItemType { + case "message": + item.Role = "assistant" + item.Content = state.CurrentContent + case "function_call": + item.CallID = state.CurrentCallID + item.Name = state.CurrentName + args := state.CurrentArgs + if args == "" { + args = "{}" + } + item.Arguments = args + case "reasoning": + if state.CurrentSummary != "" { + item.Summary = []ResponsesSummary{{Type: "summary_text", Text: state.CurrentSummary}} + } + } + state.Outputs = append(state.Outputs, item) // Reset state.CurrentItemType = "" state.CurrentItemID = "" state.CurrentCallID = "" state.CurrentName = "" + state.CurrentContent = nil + state.CurrentArgs = "" + state.CurrentSummary = "" + state.TextAccum = "" state.OutputIndex++ state.ContentIndex = 0 return []ResponsesStreamEvent{makeResponsesEvent(state, "response.output_item.done", &ResponsesStreamEvent{ OutputIndex: state.OutputIndex - 1, // Use the index before increment - Item: &ResponsesOutput{ - Type: itemType, - ID: itemID, - Status: "completed", - }, + Item: &item, })} } @@ -519,6 +588,14 @@ func makeResponsesCompletedEvent( eventType = "response.incomplete" } + // Carry the output items accumulated over the stream. The SDK's + // get_final_response() reads them straight from the terminal event, so an + // empty list leaves clients with an empty result. + outputs := state.Outputs + if outputs == nil { + outputs = []ResponsesOutput{} + } + return ResponsesStreamEvent{ Type: eventType, SequenceNumber: seq, @@ -527,7 +604,7 @@ func makeResponsesCompletedEvent( Object: "response", Model: state.Model, Status: status, - Output: []ResponsesOutput{}, + Output: outputs, Usage: usage, IncompleteDetails: incompleteDetails, }, diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go b/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go new file mode 100644 index 000000000..c26e28982 --- /dev/null +++ b/backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go @@ -0,0 +1,187 @@ +package apicompat + +import "testing" + +// TestAnthropicEventToResponses_TextEmitsContentPart pins that a message text +// stream emits response.content_part.added, and that it precedes the first +// output_text.delta for that part. +// +// Why: the OpenAI SDK's accumulating stream helper (client.responses.stream) +// only appends a content part to the message item when it sees +// content_part.added. The item is added with content: [], so a missing event +// makes the following output_text.delta index output.content[content_index] and +// raise IndexError. Raw event iteration does not accumulate, so a regression +// here is easy to miss. +func TestAnthropicEventToResponses_TextEmitsContentPart(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var types []string + feed := func(evt *AnthropicStreamEvent) { + for _, out := range AnthropicEventToResponsesEvents(evt, state) { + types = append(types, out.Type) + } + } + + idx := 0 + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1", Model: "claude-sonnet-4-5"}}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &idx, ContentBlock: &AnthropicContentBlock{Type: "text"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "text_delta", Text: "Hel"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "text_delta", Text: "lo"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &idx}) + feed(&AnthropicStreamEvent{Type: "message_stop"}) + + posOf := func(target string) int { + for i, ty := range types { + if ty == target { + return i + } + } + return -1 + } + + partAdded := posOf("response.content_part.added") + firstDelta := posOf("response.output_text.delta") + + if partAdded < 0 { + t.Fatalf("response.content_part.added was not emitted; got %v", types) + } + if firstDelta < 0 { + t.Fatalf("response.output_text.delta was not emitted; got %v", types) + } + if partAdded > firstDelta { + t.Errorf("content_part.added must precede the first output_text.delta; got %v", types) + } + if posOf("response.content_part.done") < 0 { + t.Errorf("response.content_part.done was not emitted; got %v", types) + } +} + +// TestAnthropicEventToResponses_DoneEventsCarryFullText pins that done events +// carry the part's full text (deltas carry increments only). +func TestAnthropicEventToResponses_DoneEventsCarryFullText(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var events []ResponsesStreamEvent + feed := func(evt *AnthropicStreamEvent) { + events = append(events, AnthropicEventToResponsesEvents(evt, state)...) + } + + idx := 0 + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1"}}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &idx, ContentBlock: &AnthropicContentBlock{Type: "text"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "text_delta", Text: "Hello "}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "text_delta", Text: "world"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &idx}) + + const want = "Hello world" + var sawTextDone, sawPartDone bool + for _, e := range events { + switch e.Type { + case "response.output_text.done": + sawTextDone = true + if e.Text != want { + t.Errorf("output_text.done text = %q, want %q", e.Text, want) + } + case "response.content_part.done": + sawPartDone = true + if e.Part == nil || e.Part.Text != want { + t.Errorf("content_part.done part = %+v, want text %q", e.Part, want) + } + } + } + if !sawTextDone || !sawPartDone { + t.Errorf("missing done events: output_text.done=%v content_part.done=%v", sawTextDone, sawPartDone) + } +} + +// TestAnthropicEventToResponses_CompletedCarriesOutput pins that +// response.completed carries the full output list. The SDK's +// get_final_response() and tracing integrations parse the terminal event's +// response directly; an empty output leaves them with nothing (the text still +// renders from deltas, which is why this is invisible when only watching the +// stream). +func TestAnthropicEventToResponses_CompletedCarriesOutput(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var events []ResponsesStreamEvent + feed := func(evt *AnthropicStreamEvent) { + events = append(events, AnthropicEventToResponsesEvents(evt, state)...) + } + + idx := 0 + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1"}}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &idx, ContentBlock: &AnthropicContentBlock{Type: "text"}}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{Type: "text_delta", Text: "4826"}}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &idx}) + 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") + } + if len(completed.Response.Output) == 0 { + t.Fatalf("response.completed carries an empty output; clients would see no result") + } + msg := completed.Response.Output[0] + if msg.Type != "message" || len(msg.Content) == 0 { + t.Fatalf("output[0] = %+v, want a message with content", msg) + } + if msg.Content[0].Text != "4826" { + t.Errorf("output[0].content[0].text = %q, want %q", msg.Content[0].Text, "4826") + } +} + +// TestAnthropicEventToResponses_ToolCallCompletedCarriesArguments pins that a +// function call's accumulated arguments survive into output_item.done and +// response.completed. +func TestAnthropicEventToResponses_ToolCallCompletedCarriesArguments(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-5" + + var events []ResponsesStreamEvent + feed := func(evt *AnthropicStreamEvent) { + events = append(events, AnthropicEventToResponsesEvents(evt, state)...) + } + + idx := 0 + feed(&AnthropicStreamEvent{Type: "message_start", Message: &AnthropicResponse{ID: "msg_1"}}) + feed(&AnthropicStreamEvent{Type: "content_block_start", Index: &idx, ContentBlock: &AnthropicContentBlock{ + Type: "tool_use", ID: "toolu_1", Name: "get_weather", + }}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{ + Type: "input_json_delta", PartialJSON: `{"city":`, + }}) + feed(&AnthropicStreamEvent{Type: "content_block_delta", Index: &idx, Delta: &AnthropicDelta{ + Type: "input_json_delta", PartialJSON: `"SH"}`, + }}) + feed(&AnthropicStreamEvent{Type: "content_block_stop", Index: &idx}) + 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 || len(completed.Response.Output) == 0 { + t.Fatalf("response.completed carries no output") + } + fc := completed.Response.Output[0] + if fc.Type != "function_call" { + t.Fatalf("output[0].type = %q, want function_call", fc.Type) + } + if fc.Arguments != `{"city":"SH"}` { + t.Errorf("arguments = %q, want %q", fc.Arguments, `{"city":"SH"}`) + } + if fc.Name != "get_weather" { + t.Errorf("name = %q, want get_weather", fc.Name) + } +}