wushenghua/sub2api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(apicompat): emit content_part events and full output in Responses stream

Browse Source 
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.
This commit is contained in:
docooler
2026-07-17 03:35:26 +00:00
parent c2c19a7cbe
commit 26b5e87457
2 changed files with 273 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/pkg/apicompat/anthropic_to_responses_response.go
+86 -9
View File
@@ -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,
},
backend/internal/pkg/apicompat/anthropic_to_responses_stream_test.go
+187
View File
@@ -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)
}
}