diff --git a/llm/tools.go b/llm/tools.go index 89834e786..53880b9e5 100644 --- a/llm/tools.go +++ b/llm/tools.go @@ -65,6 +65,10 @@ type FunctionCall struct { type ToolCall struct { ID string `json:"id,omitempty"` + // ResponseItemID is the Responses API item id. It is distinct from ID, + // which carries the function/custom tool call_id. + ResponseItemID string `json:"response_item_id,omitempty"` + // The type of the tool. Currently, only `function` is supported. Type string `json:"type,omitempty"` @@ -272,6 +276,9 @@ type ResponseCustomToolCall struct { CallID string `json:"call_id"` // Name is the name of the custom tool being called. Name string `json:"name"` + // Namespace is the namespace qualifier for tools registered as part of a + // namespace tool group (e.g. an MCP server). + Namespace string `json:"namespace,omitempty"` // Input is the freeform input for the custom tool call generated by the model. Input string `json:"input"` } diff --git a/llm/transformer/openai/responses/aggregator.go b/llm/transformer/openai/responses/aggregator.go index d018d3f43..0ab93bd0a 100644 --- a/llm/transformer/openai/responses/aggregator.go +++ b/llm/transformer/openai/responses/aggregator.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "strings" "github.com/samber/lo" @@ -33,19 +34,23 @@ type streamAggregator struct { // Terminal response details responseError *Error incompleteDetails *ResponseIncompleteDetails + err error } // aggregatedItem holds the accumulated state for an output item. type aggregatedItem struct { - ID string - Type string - Status string - Role string - CallID string - Name string - Namespace string - Arguments *strings.Builder - EncryptedContent *string + ID string + Type string + Status string + OutputItemDone bool + ToolPayloadDone bool + ToolPayloadDeltas []string + Role string + CallID string + Name string + Namespace string + Arguments *strings.Builder + EncryptedContent *string // For custom_tool_call type Input *string @@ -152,20 +157,162 @@ func (a *streamAggregator) lastItemByOutputIndex(outputIndex int) *aggregatedIte func (a *streamAggregator) getItemForEvent(outputIndex int, itemID *string) *aggregatedItem { if itemID != nil && *itemID != "" { if item, ok := a.outputItemsByID[*itemID]; ok { + if item.OutputItemDone { + return nil + } return item } // Some upstream implementations might use call_id as item_id in delta events. - for _, items := range a.outputItems { - for _, it := range items { - if it.CallID == *itemID { - return it - } + if item := a.findItemByCallID(*itemID); item != nil { + if item.OutputItemDone { + return nil + } + return item + } + } + + items := a.outputItems[outputIndex] + if len(items) > 1 { + a.rejectAmbiguousOutputIndex(outputIndex) + return nil + } + if len(items) == 0 || items[0].OutputItemDone { + return nil + } + + return items[0] +} + +func (a *streamAggregator) findItemByCallID(callID string) *aggregatedItem { + if callID == "" { + return nil + } + + for _, items := range a.outputItems { + for _, item := range items { + if item.CallID == callID { + return item + } + } + } + + return nil +} + +func (a *streamAggregator) finalItem(outputIndex int, src *Item, fromCompletedResponse bool) *aggregatedItem { + if src == nil { + return nil + } + + var item *aggregatedItem + if src.ID != "" { + item = a.outputItemsByID[src.ID] + } + if item == nil { + item = a.findItemByCallID(src.CallID) + } + if item == nil { + candidates := a.outputItems[outputIndex] + if len(candidates) > 1 { + a.rejectAmbiguousOutputIndex(outputIndex) + return nil + } + candidate := a.lastItemByOutputIndex(outputIndex) + if candidate != nil && + (src.ID == "" || candidate.ID == "" || candidate.ID == src.ID) && + (src.CallID == "" || candidate.CallID == "" || candidate.CallID == src.CallID) && + (src.Type == "" || candidate.Type == "" || candidate.Type == src.Type) { + item = candidate + } + } + if item == nil { + item = newAggregatedItem() + a.outputItems[outputIndex] = append(a.outputItems[outputIndex], item) + } else if item.OutputItemDone { + return item + } + + if src.ID != "" { + item.ID = src.ID + a.outputItemsByID[src.ID] = item + } + if src.Type != "" { + item.Type = src.Type + } + if src.Role != "" { + item.Role = src.Role + } + if src.Status != nil { + item.Status = *src.Status + } + if item.Status == "" { + item.Status = "completed" + } + if src.CallID != "" { + item.CallID = src.CallID + } + if src.Name != "" { + item.Name = src.Name + } + if src.Namespace != "" { + item.Namespace = src.Namespace + } + if src.Arguments != "" { + item.Arguments.Reset() + item.Arguments.WriteString(src.Arguments) + } + if src.Input != nil { + item.Input = lo.ToPtr(*src.Input) + } + + if src.Content != nil { + for idx, contentItem := range src.Content.Items { + part := ensureContentPart(item, idx) + if part == nil { + continue + } + if contentItem.Type != "" { + part.Type = contentItem.Type + } + if contentItem.Text != nil { + applyDoneText(part.Text, *contentItem.Text) + } + if contentItem.Annotations != nil { + part.Annotations = append([]Annotation(nil), contentItem.Annotations...) } } } - return a.lastItemByOutputIndex(outputIndex) + if len(src.Summary) > 0 { + for idx, summary := range src.Summary { + part := ensureSummaryPart(item, idx) + part.Type = summary.Type + applyDoneText(part.Text, summary.Text) + part.Final = true + } + } + + if src.EncryptedContent != nil { + item.EncryptedContent = src.EncryptedContent + } + if src.Result != nil { + item.Result = src.Result + } + if src.Type == "function_call" || src.Type == "custom_tool_call" { + item.ToolPayloadDone = true + } + if !fromCompletedResponse { + item.OutputItemDone = true + } + + return item +} + +func (a *streamAggregator) rejectAmbiguousOutputIndex(outputIndex int) { + if a.err == nil { + a.err = fmt.Errorf("ambiguous tool call output_index %d", outputIndex) + } } func applyDoneText(dst *strings.Builder, doneText string) { @@ -207,6 +354,9 @@ func AggregateStreamChunks(_ context.Context, chunks []*httpclient.StreamEvent) } agg.processEvent(&ev) + if agg.err != nil { + return nil, llm.ResponseMeta{}, agg.err + } } resp := agg.buildResponse() @@ -264,8 +414,14 @@ func (a *streamAggregator) processEvent(ev *StreamEvent) { item.Name = ev.Item.Name item.Namespace = ev.Item.Namespace item.Arguments.WriteString(ev.Item.Arguments) + if ev.Item.Arguments != "" { + item.ToolPayloadDeltas = append(item.ToolPayloadDeltas, ev.Item.Arguments) + } item.EncryptedContent = ev.Item.EncryptedContent item.Input = ev.Item.Input + if ev.Item.Input != nil && *ev.Item.Input != "" { + item.ToolPayloadDeltas = append(item.ToolPayloadDeltas, *ev.Item.Input) + } if len(ev.Item.Summary) > 0 { for idx, s := range ev.Item.Summary { @@ -318,6 +474,7 @@ func (a *streamAggregator) processEvent(ev *StreamEvent) { if ev.ItemID != nil { if item := a.getItemForEvent(ev.OutputIndex, ev.ItemID); item != nil { item.Arguments.WriteString(ev.Delta) + item.ToolPayloadDeltas = append(item.ToolPayloadDeltas, ev.Delta) } } @@ -338,6 +495,7 @@ func (a *streamAggregator) processEvent(ev *StreamEvent) { item.Arguments.Reset() item.Arguments.WriteString(ev.Arguments) } + item.ToolPayloadDone = true } } @@ -347,6 +505,7 @@ func (a *streamAggregator) processEvent(ev *StreamEvent) { if item := a.getItemForEvent(ev.OutputIndex, ev.ItemID); item != nil { current := lo.FromPtr(item.Input) item.Input = lo.ToPtr(current + ev.Delta) + item.ToolPayloadDeltas = append(item.ToolPayloadDeltas, ev.Delta) } } @@ -354,9 +513,8 @@ func (a *streamAggregator) processEvent(ev *StreamEvent) { // Finalize custom tool call input if ev.ItemID != nil { if item := a.getItemForEvent(ev.OutputIndex, ev.ItemID); item != nil { - if ev.Input != "" { - item.Input = lo.ToPtr(ev.Input) - } + item.Input = lo.ToPtr(ev.Input) + item.ToolPayloadDone = true } } @@ -458,72 +616,12 @@ func (a *streamAggregator) processEvent(ev *StreamEvent) { part.Final = true case StreamEventTypeOutputItemDone: - // Mark item as completed and update with final data - if ev.Item != nil { - item := a.outputItemsByID[ev.Item.ID] - if item == nil { - item = a.lastItemByOutputIndex(ev.OutputIndex) - } - - if item != nil { - if ev.Item.Status != nil { - item.Status = *ev.Item.Status - } - - if item.Status == "" { - item.Status = "completed" - } - - // Update with final data if provided - if ev.Item.Arguments != "" { - item.Arguments.Reset() - item.Arguments.WriteString(ev.Item.Arguments) - } - - if ev.Item.Content != nil { - for idx, contentItem := range ev.Item.Content.Items { - part := ensureContentPart(item, idx) - if part == nil { - continue - } - if contentItem.Type != "" { - part.Type = contentItem.Type - } - if contentItem.Text != nil { - applyDoneText(part.Text, *contentItem.Text) - } - if contentItem.Annotations != nil { - part.Annotations = append([]Annotation(nil), contentItem.Annotations...) - } - } - } - - if len(ev.Item.Summary) > 0 { - for idx, s := range ev.Item.Summary { - part := ensureSummaryPart(item, idx) - part.Type = s.Type - applyDoneText(part.Text, s.Text) - part.Final = true - } - } - - if ev.Item.EncryptedContent != nil { - item.EncryptedContent = ev.Item.EncryptedContent - } - - if ev.Item.Result != nil { - item.Result = ev.Item.Result - } - } - } + a.finalItem(ev.OutputIndex, ev.Item, false) case StreamEventTypeResponseCompleted: - a.status = "completed" - if ev.Response != nil { - a.previousResponseID = ev.Response.PreviousResponseID - if ev.Response.Usage != nil { - a.usage = ev.Response.Usage - } + a.applyResponseSnapshot(ev.Response) + if ev.Response == nil || ev.Response.Status == nil { + a.status = "completed" } case StreamEventTypeResponseFailed: @@ -551,6 +649,10 @@ func (a *streamAggregator) applyResponseSnapshot(response *Response) { return } + for outputIndex := range response.Output { + a.finalItem(outputIndex, &response.Output[outputIndex], true) + } + if response.ID != "" { a.responseID = response.ID } @@ -577,133 +679,179 @@ func (a *streamAggregator) applyResponseSnapshot(response *Response) { } } -// buildResponse builds the final Response object from aggregated state. -// This is used by responsesInboundStream to build the response.completed event. -func (a *streamAggregator) buildResponse() *Response { - // Build output items - output := make([]Item, 0, len(a.outputItems)) +type aggregatedToolCall struct { + Item Item + Deltas []string +} - // Sort by output index - maxIndex := 0 +func (a *streamAggregator) orderedItems() []*aggregatedItem { + maxIndex := -1 for idx := range a.outputItems { if idx > maxIndex { maxIndex = idx } } - for i := 0; i <= maxIndex; i++ { - items, ok := a.outputItems[i] - if !ok || len(items) == 0 { + items := make([]*aggregatedItem, 0, len(a.outputItems)) + for outputIndex := 0; outputIndex <= maxIndex; outputIndex++ { + items = append(items, a.outputItems[outputIndex]...) + } + return items +} + +func (item *aggregatedItem) toolItem() (Item, bool) { + if !item.ToolPayloadDone || item.CallID == "" || item.Name == "" { + return Item{}, false + } + + switch item.Type { + case "function_call": + if item.Arguments.Len() == 0 || !json.Valid([]byte(item.Arguments.String())) { + return Item{}, false + } + return Item{ + ID: item.ID, + Type: item.Type, + Status: lo.ToPtr(item.Status), + CallID: item.CallID, + Name: item.Name, + Namespace: item.Namespace, + Arguments: item.Arguments.String(), + }, true + case "custom_tool_call": + if item.Input == nil { + return Item{}, false + } + return Item{ + ID: item.ID, + Type: item.Type, + Status: lo.ToPtr(item.Status), + CallID: item.CallID, + Name: item.Name, + Namespace: item.Namespace, + Input: item.Input, + }, true + default: + return Item{}, false + } +} + +func (a *streamAggregator) completedToolCalls() []aggregatedToolCall { + if a.status != "completed" { + return nil + } + + toolCalls := make([]aggregatedToolCall, 0) + for _, item := range a.orderedItems() { + toolItem, ok := item.toolItem() + if !ok { continue } + toolCalls = append(toolCalls, aggregatedToolCall{ + Item: toolItem, + Deltas: append([]string(nil), item.ToolPayloadDeltas...), + }) + } + return toolCalls +} - for _, item := range items { - switch item.Type { - case "message": - // Convert aggregated content parts to []Item for Content.Items - contentItems := make([]Item, 0, len(item.Content)) - for _, cp := range item.Content { - text := cp.Text.String() - contentItems = append(contentItems, Item{ - Type: cp.Type, - Text: &text, - Annotations: append([]Annotation(nil), cp.Annotations...), - }) - } - - output = append(output, Item{ - ID: item.ID, - Type: item.Type, - Role: item.Role, - Status: lo.ToPtr(item.Status), - Content: &Input{ - Items: contentItems, - }, - }) +// buildResponse builds the final Response object from aggregated state. +// This is used by responsesInboundStream to build the response.completed event. +func (a *streamAggregator) buildResponse() *Response { + // Build output items + output := make([]Item, 0, len(a.outputItems)) - case "function_call": - output = append(output, Item{ - ID: item.ID, - Type: item.Type, - Status: lo.ToPtr(item.Status), - CallID: item.CallID, - Name: item.Name, - Namespace: item.Namespace, - Arguments: item.Arguments.String(), + for _, item := range a.orderedItems() { + switch item.Type { + case "message": + // Convert aggregated content parts to []Item for Content.Items + contentItems := make([]Item, 0, len(item.Content)) + for _, cp := range item.Content { + text := cp.Text.String() + contentItems = append(contentItems, Item{ + Type: cp.Type, + Text: &text, + Annotations: append([]Annotation(nil), cp.Annotations...), }) + } - case "custom_tool_call": - output = append(output, Item{ - ID: item.ID, - Type: item.Type, - Status: lo.ToPtr(item.Status), - CallID: item.CallID, - Name: item.Name, - Input: item.Input, - }) + output = append(output, Item{ + ID: item.ID, + Type: item.Type, + Role: item.Role, + Status: lo.ToPtr(item.Status), + Content: &Input{ + Items: contentItems, + }, + }) + + case "function_call", "custom_tool_call": + if a.status != "completed" { + continue + } + toolItem, ok := item.toolItem() + if !ok { + continue + } + output = append(output, toolItem) + + case "reasoning": + var summary []ReasoningSummary + if len(item.SummaryParts) > 0 { + maxSummaryIndex := -1 + for idx := range item.SummaryParts { + if idx > maxSummaryIndex { + maxSummaryIndex = idx + } + } - case "reasoning": - // ...existing reasoning handling... - { - var summary []ReasoningSummary - - if len(item.SummaryParts) > 0 { - maxSummaryIndex := -1 - for idx := range item.SummaryParts { - if idx > maxSummaryIndex { - maxSummaryIndex = idx - } - } - - summary = make([]ReasoningSummary, 0, maxSummaryIndex+1) - for idx := 0; idx <= maxSummaryIndex; idx++ { - sp, ok := item.SummaryParts[idx] - if !ok || sp == nil { - summary = append(summary, ReasoningSummary{Type: "summary_text", Text: ""}) - continue - } - - summaryType := sp.Type - if summaryType == "" { - summaryType = "summary_text" - } - - var text string - if sp.Text != nil { - text = sp.Text.String() - } - - summary = append(summary, ReasoningSummary{Type: summaryType, Text: text}) - } + summary = make([]ReasoningSummary, 0, maxSummaryIndex+1) + for idx := 0; idx <= maxSummaryIndex; idx++ { + sp, ok := item.SummaryParts[idx] + if !ok || sp == nil { + summary = append(summary, ReasoningSummary{Type: "summary_text", Text: ""}) + continue } - output = append(output, Item{ - ID: item.ID, - Type: item.Type, - Status: lo.ToPtr(item.Status), - Summary: summary, - EncryptedContent: item.EncryptedContent, - }) - } + summaryType := sp.Type + if summaryType == "" { + summaryType = "summary_text" + } - case "image_generation_call": - output = append(output, Item{ - ID: item.ID, - Type: item.Type, - Status: lo.ToPtr(item.Status), - CallID: item.CallID, - Result: item.Result, - }) + var text string + if sp.Text != nil { + text = sp.Text.String() + } - default: - // Generic item - output = append(output, Item{ - ID: item.ID, - Type: item.Type, - Status: lo.ToPtr(item.Status), - Role: item.Role, - }) + summary = append(summary, ReasoningSummary{Type: summaryType, Text: text}) + } } + + output = append(output, Item{ + ID: item.ID, + Type: item.Type, + Status: lo.ToPtr(item.Status), + Summary: summary, + EncryptedContent: item.EncryptedContent, + }) + + case "image_generation_call": + output = append(output, Item{ + ID: item.ID, + Type: item.Type, + Status: lo.ToPtr(item.Status), + CallID: item.CallID, + Result: item.Result, + }) + + default: + // Generic item + output = append(output, Item{ + ID: item.ID, + Type: item.Type, + Status: lo.ToPtr(item.Status), + Role: item.Role, + }) } } diff --git a/llm/transformer/openai/responses/aggregator_test.go b/llm/transformer/openai/responses/aggregator_test.go index ff7b4a65c..62bcda6cd 100644 --- a/llm/transformer/openai/responses/aggregator_test.go +++ b/llm/transformer/openai/responses/aggregator_test.go @@ -883,3 +883,150 @@ func TestAggregateStreamChunks_ImageGenerationCall(t *testing.T) { require.NotNil(t, resp.Output[0].Result) require.Equal(t, "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==", *resp.Output[0].Result) } + +func TestAggregateStreamChunks_FinalOnlyOutputItemDonePreservesFunctionCall(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_aggregate_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_aggregate_done","type":"function_call","status":"completed","call_id":"call_aggregate_done","name":"spawn_agent","namespace":"collaboration","arguments":"{\"description\":\"persist me\"}"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_aggregate_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Len(t, resp.Output, 1) + require.Equal(t, "fc_aggregate_done", resp.Output[0].ID) + require.Equal(t, "call_aggregate_done", resp.Output[0].CallID) + require.Equal(t, "spawn_agent", resp.Output[0].Name) + require.Equal(t, "collaboration", resp.Output[0].Namespace) + require.Equal(t, `{"description":"persist me"}`, resp.Output[0].Arguments) +} + +func TestAggregateStreamChunks_FinalOnlyCompletedSnapshotPreservesCustomTool(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_aggregate_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_aggregate_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_aggregate_completed","type":"custom_tool_call","status":"completed","call_id":"call_aggregate_completed","name":"apply_patch","namespace":"mcp__codex","input":"patch"}]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Len(t, resp.Output, 1) + require.Equal(t, "ctc_aggregate_completed", resp.Output[0].ID) + require.Equal(t, "call_aggregate_completed", resp.Output[0].CallID) + require.Equal(t, "apply_patch", resp.Output[0].Name) + require.Equal(t, "mcp__codex", resp.Output[0].Namespace) + require.NotNil(t, resp.Output[0].Input) + require.Equal(t, "patch", *resp.Output[0].Input) +} + +func TestAggregateStreamChunks_EmptyFinalCustomInputClearsStreamedInput(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_aggregate_empty_final","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_aggregate_empty_final","type":"custom_tool_call","status":"in_progress","call_id":"call_aggregate_empty_final","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_aggregate_empty_final","output_index":0,"delta":"patch"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_aggregate_empty_final","type":"custom_tool_call","status":"completed","call_id":"call_aggregate_empty_final","name":"apply_patch","input":""}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_aggregate_empty_final","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Len(t, resp.Output, 1) + require.NotNil(t, resp.Output[0].Input) + require.Empty(t, *resp.Output[0].Input) +} + +func TestAggregateStreamChunks_EmptyCustomInputDoneClearsStreamedInput(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_aggregate_empty_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_aggregate_empty_done","type":"custom_tool_call","status":"in_progress","call_id":"call_aggregate_empty_done","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_aggregate_empty_done","output_index":0,"delta":"patch"}`)}, + {Type: "response.custom_tool_call_input.done", Data: []byte(`{"type":"response.custom_tool_call_input.done","item_id":"ctc_aggregate_empty_done","output_index":0,"input":""}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_aggregate_empty_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Len(t, resp.Output, 1) + require.NotNil(t, resp.Output[0].Input) + require.Empty(t, *resp.Output[0].Input) +} + +func TestAggregateStreamChunks_OutputItemDoneRemainsAuthoritative(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_aggregate_terminal_priority","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_aggregate_terminal_priority","type":"function_call","status":"in_progress","call_id":"call_aggregate_terminal_priority","name":"search","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_aggregate_terminal_priority","output_index":0,"delta":"{}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_aggregate_terminal_priority","type":"function_call","status":"completed","call_id":"call_aggregate_terminal_priority","name":"search","arguments":"{\"source\":\"output-item\"}"}}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_aggregate_terminal_priority","output_index":0,"arguments":"{\"source\":\"late-done\"}"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_aggregate_terminal_priority","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_aggregate_terminal_priority","type":"function_call","status":"completed","call_id":"call_aggregate_terminal_priority","name":"search","arguments":"{\"source\":\"completed\"}"}]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Len(t, resp.Output, 1) + require.Equal(t, `{"source":"output-item"}`, resp.Output[0].Arguments) +} + +func TestAggregateStreamChunks_DoesNotGuessAcrossAmbiguousOutputIndex(t *testing.T) { + _, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_aggregate_ambiguous_index","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_aggregate_first","type":"function_call","status":"in_progress","call_id":"call_aggregate_first","name":"first","arguments":""}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_aggregate_second","type":"function_call","status":"in_progress","call_id":"call_aggregate_second","name":"second","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_unknown","output_index":0,"delta":"misrouted"}`)}, + }) + require.ErrorContains(t, err, "ambiguous tool call output_index 0") +} + +func TestAggregateStreamChunks_DoesNotPersistToolsFromUnsuccessfulResponses(t *testing.T) { + tests := []struct { + name string + terminalType string + status string + }{ + {name: "failed", terminalType: "response.failed", status: "failed"}, + {name: "incomplete", terminalType: "response.incomplete", status: "incomplete"}, + {name: "cancelled", terminalType: "response.cancelled", status: "cancelled"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_unsuccessful","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: tt.terminalType, Data: []byte(`{"type":"` + tt.terminalType + `","response":{"id":"resp_unsuccessful","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"` + tt.status + `","output":[{"id":"fc_unsuccessful","type":"function_call","status":"completed","call_id":"call_unsuccessful","name":"search","arguments":"{}"}]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Empty(t, resp.Output) + }) + } +} + +func TestAggregateStreamChunks_DropsInvalidTerminalToolPayloads(t *testing.T) { + tests := []struct { + name string + item string + }{ + {name: "invalid function JSON", item: `{"id":"fc_invalid","type":"function_call","status":"completed","call_id":"call_invalid","name":"search","arguments":"not-json"}`}, + {name: "missing custom input", item: `{"id":"ctc_invalid","type":"custom_tool_call","status":"completed","call_id":"call_invalid","name":"apply_patch"}`}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resultBytes, _, err := AggregateStreamChunks(t.Context(), []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_invalid_tool","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_invalid_tool","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[` + tt.item + `]}}`)}, + }) + require.NoError(t, err) + + var resp Response + require.NoError(t, json.Unmarshal(resultBytes, &resp)) + require.Empty(t, resp.Output) + }) + } +} diff --git a/llm/transformer/openai/responses/inbound.go b/llm/transformer/openai/responses/inbound.go index 75066f1d2..25d1f455a 100644 --- a/llm/transformer/openai/responses/inbound.go +++ b/llm/transformer/openai/responses/inbound.go @@ -417,8 +417,9 @@ func convertReasoningWithFollowing(items []Item, startIdx int) (*llm.Message, in case "function_call": // Merge function_call into the same assistant message msg.ToolCalls = append(msg.ToolCalls, llm.ToolCall{ - ID: nextItem.CallID, - Type: "function", + ID: nextItem.CallID, + ResponseItemID: nextItem.ID, + Type: "function", Function: llm.FunctionCall{ Name: nextItem.Name, Namespace: nextItem.Namespace, @@ -435,12 +436,14 @@ func convertReasoningWithFollowing(items []Item, startIdx int) (*llm.Message, in } msg.ToolCalls = append(msg.ToolCalls, llm.ToolCall{ - ID: nextItem.CallID, - Type: llm.ToolTypeResponsesCustomTool, + ID: nextItem.CallID, + ResponseItemID: nextItem.ID, + Type: llm.ToolTypeResponsesCustomTool, ResponseCustomToolCall: &llm.ResponseCustomToolCall{ - CallID: nextItem.CallID, - Name: nextItem.Name, - Input: inputStr, + CallID: nextItem.CallID, + Name: nextItem.Name, + Namespace: nextItem.Namespace, + Input: inputStr, }, }) consumed++ @@ -522,8 +525,9 @@ func convertItemToMessage(item *Item) (*llm.Message, error) { Role: "assistant", ToolCalls: []llm.ToolCall{ { - ID: item.CallID, - Type: "function", + ID: item.CallID, + ResponseItemID: item.ID, + Type: "function", Function: llm.FunctionCall{ Name: item.Name, Namespace: item.Namespace, @@ -544,12 +548,14 @@ func convertItemToMessage(item *Item) (*llm.Message, error) { Role: "assistant", ToolCalls: []llm.ToolCall{ { - ID: item.CallID, - Type: llm.ToolTypeResponsesCustomTool, + ID: item.CallID, + ResponseItemID: item.ID, + Type: llm.ToolTypeResponsesCustomTool, ResponseCustomToolCall: &llm.ResponseCustomToolCall{ - CallID: item.CallID, - Name: item.Name, - Input: inputStr, + CallID: item.CallID, + Name: item.Name, + Namespace: item.Namespace, + Input: inputStr, }, }, }, @@ -561,6 +567,7 @@ func convertItemToMessage(item *Item) (*llm.Message, error) { } // Function call output - convert to tool message msg := &llm.Message{ + ID: item.ID, Role: "tool", ToolCallID: lo.ToPtr(item.CallID), Content: convertToMessageContent(*item.Output), @@ -577,6 +584,7 @@ func convertItemToMessage(item *Item) (*llm.Message, error) { } // Custom tool call output - convert to tool message msg := &llm.Message{ + ID: item.ID, Role: "tool", ToolCallID: lo.ToPtr(item.CallID), Content: convertToMessageContent(*item.Output), @@ -909,7 +917,7 @@ func convertToResponsesAPIResponse(chatResp *llm.Response) *Response { messageItemID := message.ID if messageItemID == "" { - messageItemID = generateItemID() + messageItemID = generateItemID("msg") } // Handle reasoning content @@ -921,17 +929,26 @@ func convertToResponsesAPIResponse(chatResp *llm.Response) *Response { if len(message.ToolCalls) > 0 { for _, toolCall := range message.ToolCalls { if toolCall.ResponseCustomToolCall != nil { + itemID := toolCall.ResponseItemID + if itemID == "" { + itemID = generateItemID("ctc") + } resp.Output = append(resp.Output, Item{ - ID: toolCall.ID, - Type: "custom_tool_call", - CallID: toolCall.ResponseCustomToolCall.CallID, - Name: toolCall.ResponseCustomToolCall.Name, - Input: lo.ToPtr(toolCall.ResponseCustomToolCall.Input), - Status: lo.ToPtr("completed"), + ID: itemID, + Type: "custom_tool_call", + CallID: toolCall.ResponseCustomToolCall.CallID, + Name: toolCall.ResponseCustomToolCall.Name, + Namespace: toolCall.ResponseCustomToolCall.Namespace, + Input: lo.ToPtr(toolCall.ResponseCustomToolCall.Input), + Status: lo.ToPtr("completed"), }) } else { + itemID := toolCall.ResponseItemID + if itemID == "" { + itemID = generateItemID("fc") + } resp.Output = append(resp.Output, Item{ - ID: toolCall.ID, + ID: itemID, Type: "function_call", CallID: toolCall.ID, Name: toolCall.Function.Name, @@ -978,7 +995,7 @@ func convertToResponsesAPIResponse(chatResp *llm.Response) *Response { // Handle image output if part.ImageURL != nil { imageItem := Item{ - ID: generateItemID(), + ID: generateItemID("ig"), Type: "image_generation_call", Role: "assistant", Result: lo.ToPtr(xurl.ExtractBase64FromDataURL(part.ImageURL.URL)), @@ -1029,7 +1046,7 @@ func convertToResponsesAPIResponse(chatResp *llm.Response) *Response { emptyText := "" resp.Output = []Item{ { - ID: generateItemID(), + ID: generateItemID("msg"), Type: "message", Role: "assistant", Content: &Input{ @@ -1050,8 +1067,8 @@ func convertToResponsesAPIResponse(chatResp *llm.Response) *Response { } // generateItemID generates a unique item ID for output items. -func generateItemID() string { - return fmt.Sprintf("item_%s", lo.RandomString(16, lo.AlphanumericCharset)) +func generateItemID(prefix string) string { + return fmt.Sprintf("%s_%s", prefix, lo.RandomString(16, lo.AlphanumericCharset)) } // buildReasoningItem creates a reasoning Item from a message's reasoning content and signature. @@ -1073,7 +1090,7 @@ func buildReasoningItem(msg llm.Message) (Item, bool) { } return Item{ - ID: generateItemID(), + ID: generateItemID("rs"), Type: "reasoning", Status: lo.ToPtr("completed"), Summary: summary, diff --git a/llm/transformer/openai/responses/inbound_integration_test.go b/llm/transformer/openai/responses/inbound_integration_test.go index 053b516c8..cbf3cb1fe 100644 --- a/llm/transformer/openai/responses/inbound_integration_test.go +++ b/llm/transformer/openai/responses/inbound_integration_test.go @@ -3,6 +3,7 @@ package responses import ( "encoding/json" "net/http" + "strings" "testing" "github.com/google/go-cmp/cmp" @@ -313,12 +314,14 @@ func TestInboundTransformer_TransformResponse_WithTestData(t *testing.T) { // First tool call output0 := resp.Output[0] require.Equal(t, "function_call", output0.Type) - require.Equal(t, "call_eda8722c71944fe394a8893c0de8146a", output0.ID) + require.True(t, strings.HasPrefix(output0.ID, "fc_")) + require.NotEqual(t, output0.CallID, output0.ID) // Second tool call output1 := resp.Output[1] require.Equal(t, "function_call", output1.Type) - require.Equal(t, "call_bd313747960f44af8bef50dc27f0f07e", output1.ID) + require.True(t, strings.HasPrefix(output1.ID, "fc_")) + require.NotEqual(t, output1.CallID, output1.ID) }, }, { diff --git a/llm/transformer/openai/responses/inbound_stream.go b/llm/transformer/openai/responses/inbound_stream.go index ca5ea32d1..9b90fd472 100644 --- a/llm/transformer/openai/responses/inbound_stream.go +++ b/llm/transformer/openai/responses/inbound_stream.go @@ -68,7 +68,8 @@ type responsesInboundStream struct { toolCalls map[int]*llm.ToolCall currentToolCallIdx int toolCallItemStarted map[int]bool - toolCallOutputIndex map[int]int // Maps tool call index to output index + toolCallOutputIndex map[int]int // Maps tool call index to output index + toolCallItemIDs map[int]string // Maps tool call index to Responses item id // Response accumulation using streamAggregator usage *llm.Usage @@ -107,6 +108,9 @@ func (s *responsesInboundStream) enqueueEvent(ev *StreamEvent) error { } s.aggregator.processEvent(ev) + if s.aggregator.err != nil { + return s.aggregator.err + } return nil } @@ -454,7 +458,7 @@ func (s *responsesInboundStream) ensureReasoningItemStarted(sourceID string) err s.currentItemID = sourceID if s.currentItemID == "" { - s.currentItemID = generateItemID() + s.currentItemID = generateItemID("rs") } item := &Item{ ID: s.currentItemID, @@ -487,7 +491,7 @@ func (s *responsesInboundStream) handleTextContent(content *string) error { if !s.hasMessageItemStarted { s.hasMessageItemStarted = true - s.currentItemID = generateItemID() + s.currentItemID = generateItemID("msg") err := s.enqueueEvent(&StreamEvent{ Type: StreamEventTypeOutputItemAdded, @@ -609,11 +613,19 @@ func (s *responsesInboundStream) initToolCall(tc llm.ToolCall) error { return err } + var customToolCall *llm.ResponseCustomToolCall + if tc.ResponseCustomToolCall != nil { + customToolCallCopy := *tc.ResponseCustomToolCall + customToolCallCopy.Input = "" + customToolCall = &customToolCallCopy + } + s.toolCalls[toolCallIndex] = &llm.ToolCall{ Index: toolCallIndex, ID: tc.ID, + ResponseItemID: tc.ResponseItemID, Type: tc.Type, - ResponseCustomToolCall: tc.ResponseCustomToolCall, + ResponseCustomToolCall: customToolCall, Function: llm.FunctionCall{ Name: tc.Function.Name, Namespace: tc.Function.Namespace, @@ -621,20 +633,18 @@ func (s *responsesInboundStream) initToolCall(tc llm.ToolCall) error { }, } - itemID := tc.ID - if itemID == "" { - itemID = generateItemID() - } + itemID := s.resolveToolCallItemID(toolCallIndex, tc) switch { case tc.ResponseCustomToolCall != nil: item := &Item{ - ID: itemID, - Type: "custom_tool_call", - Status: lo.ToPtr("in_progress"), - CallID: tc.ResponseCustomToolCall.CallID, - Name: tc.ResponseCustomToolCall.Name, - Input: lo.ToPtr(""), + ID: itemID, + Type: "custom_tool_call", + Status: lo.ToPtr("in_progress"), + CallID: tc.ResponseCustomToolCall.CallID, + Name: tc.ResponseCustomToolCall.Name, + Namespace: tc.ResponseCustomToolCall.Namespace, + Input: lo.ToPtr(""), } err := s.enqueueEvent(&StreamEvent{ @@ -674,15 +684,48 @@ func (s *responsesInboundStream) initToolCall(tc llm.ToolCall) error { return nil } +func (s *responsesInboundStream) resolveToolCallItemID(toolCallIndex int, tc llm.ToolCall) string { + if s.toolCallItemIDs == nil { + s.toolCallItemIDs = make(map[int]string) + } + if itemID := s.toolCallItemIDs[toolCallIndex]; itemID != "" { + return itemID + } + + itemID := tc.ResponseItemID + if itemID == "" { + prefix := "fc" + if tc.ResponseCustomToolCall != nil { + prefix = "ctc" + } + itemID = generateItemID(prefix) + } + s.toolCallItemIDs[toolCallIndex] = itemID + return itemID +} + +func (s *responsesInboundStream) toolCallItemID(toolCallIndex int, tc *llm.ToolCall) string { + if itemID := s.toolCallItemIDs[toolCallIndex]; itemID != "" { + return itemID + } + if tc != nil && tc.ResponseItemID != "" { + return tc.ResponseItemID + } + return s.currentItemID +} + func (s *responsesInboundStream) handleFunctionCallDelta(tc llm.ToolCall) error { toolCallIndex := tc.Index + if tc.Function.Name != "" { + s.toolCalls[toolCallIndex].Function.Name = tc.Function.Name + } + if tc.Function.Namespace != "" { + s.toolCalls[toolCallIndex].Function.Namespace = tc.Function.Namespace + } s.toolCalls[toolCallIndex].Function.Arguments += tc.Function.Arguments if tc.Function.Arguments != "" { - itemID := s.toolCalls[toolCallIndex].ID - if itemID == "" { - itemID = s.currentItemID - } + itemID := s.toolCallItemID(toolCallIndex, s.toolCalls[toolCallIndex]) err := s.enqueueEvent(&StreamEvent{ Type: StreamEventTypeFunctionCallArgumentsDelta, @@ -701,13 +744,16 @@ func (s *responsesInboundStream) handleFunctionCallDelta(tc llm.ToolCall) error func (s *responsesInboundStream) handleCustomToolCallDelta(tc llm.ToolCall) error { toolCallIndex := tc.Index + if tc.ResponseCustomToolCall.Name != "" { + s.toolCalls[toolCallIndex].ResponseCustomToolCall.Name = tc.ResponseCustomToolCall.Name + } + if tc.ResponseCustomToolCall.Namespace != "" { + s.toolCalls[toolCallIndex].ResponseCustomToolCall.Namespace = tc.ResponseCustomToolCall.Namespace + } s.toolCalls[toolCallIndex].ResponseCustomToolCall.Input += tc.ResponseCustomToolCall.Input if tc.ResponseCustomToolCall.Input != "" { - itemID := s.toolCalls[toolCallIndex].ID - if itemID == "" { - itemID = s.currentItemID - } + itemID := s.toolCallItemID(toolCallIndex, s.toolCalls[toolCallIndex]) err := s.enqueueEvent(&StreamEvent{ Type: StreamEventTypeCustomToolCallInputDelta, @@ -918,10 +964,7 @@ func (s *responsesInboundStream) closeCurrentOutputItem() error { continue } - itemID := tc.ID - if itemID == "" { - itemID = s.currentItemID - } + itemID := s.toolCallItemID(idx, tc) switch { case tc.ResponseCustomToolCall != nil: @@ -939,12 +982,13 @@ func (s *responsesInboundStream) closeCurrentOutputItem() error { } item := Item{ - ID: itemID, - Type: "custom_tool_call", - Status: lo.ToPtr("completed"), - CallID: tc.ResponseCustomToolCall.CallID, - Name: tc.ResponseCustomToolCall.Name, - Input: lo.ToPtr(fullInput), + ID: itemID, + Type: "custom_tool_call", + Status: lo.ToPtr("completed"), + CallID: tc.ResponseCustomToolCall.CallID, + Name: tc.ResponseCustomToolCall.Name, + Namespace: tc.ResponseCustomToolCall.Namespace, + Input: lo.ToPtr(fullInput), } err = s.enqueueEvent(&StreamEvent{ diff --git a/llm/transformer/openai/responses/outbound_convert.go b/llm/transformer/openai/responses/outbound_convert.go index b0c11fbe8..f3471b780 100644 --- a/llm/transformer/openai/responses/outbound_convert.go +++ b/llm/transformer/openai/responses/outbound_convert.go @@ -226,13 +226,16 @@ func convertAssistantMessage(msg llm.Message) []Item { for _, tc := range msg.ToolCalls { if tc.ResponseCustomToolCall != nil { toolCallItems = append(toolCallItems, Item{ - Type: "custom_tool_call", - CallID: tc.ResponseCustomToolCall.CallID, - Name: tc.ResponseCustomToolCall.Name, - Input: lo.ToPtr(tc.ResponseCustomToolCall.Input), + ID: tc.ResponseItemID, + Type: "custom_tool_call", + CallID: tc.ResponseCustomToolCall.CallID, + Name: tc.ResponseCustomToolCall.Name, + Namespace: tc.ResponseCustomToolCall.Namespace, + Input: lo.ToPtr(tc.ResponseCustomToolCall.Input), }) } else { toolCallItems = append(toolCallItems, Item{ + ID: tc.ResponseItemID, Type: "function_call", CallID: tc.ID, Name: tc.Function.Name, @@ -316,6 +319,7 @@ func convertToolMessageWithType(msg llm.Message, itemType string) Item { } return Item{ + ID: msg.ID, Type: itemType, CallID: lo.FromPtr(msg.ToolCallID), Output: &output, @@ -654,8 +658,9 @@ func convertOutputToMessage(output []Item, transformerMetadata map[string]any) l annotations = appendOutputText(&textContent, &visibleTextRuneCount, annotations, outputItem) case "function_call": toolCalls = append(toolCalls, llm.ToolCall{ - ID: outputItem.CallID, - Type: "function", + ID: outputItem.CallID, + ResponseItemID: outputItem.ID, + Type: "function", Function: llm.FunctionCall{ Name: outputItem.Name, Namespace: outputItem.Namespace, @@ -669,12 +674,14 @@ func convertOutputToMessage(output []Item, transformerMetadata map[string]any) l } toolCalls = append(toolCalls, llm.ToolCall{ - ID: outputItem.CallID, - Type: llm.ToolTypeResponsesCustomTool, + ID: outputItem.CallID, + ResponseItemID: outputItem.ID, + Type: llm.ToolTypeResponsesCustomTool, ResponseCustomToolCall: &llm.ResponseCustomToolCall{ - CallID: outputItem.CallID, - Name: outputItem.Name, - Input: inputStr, + CallID: outputItem.CallID, + Name: outputItem.Name, + Namespace: outputItem.Namespace, + Input: inputStr, }, }) case "reasoning": diff --git a/llm/transformer/openai/responses/outbound_stream.go b/llm/transformer/openai/responses/outbound_stream.go index 5fb4ff69b..2e0dd874d 100644 --- a/llm/transformer/openai/responses/outbound_stream.go +++ b/llm/transformer/openai/responses/outbound_stream.go @@ -36,8 +36,9 @@ func (t *OutboundTransformer) TransformStream( // responsesOutboundStream wraps a stream and maintains state during processing. type responsesOutboundStream struct { - stream streams.Stream[*httpclient.StreamEvent] - state *outboundStreamState + stream streams.Stream[*httpclient.StreamEvent] + state *outboundStreamState + aggregator *streamAggregator // Event queue eventQueue []*llm.Response @@ -56,15 +57,6 @@ type outboundStreamState struct { usage *llm.Usage created int64 - // Content accumulation - textContent strings.Builder - reasoningContent strings.Builder - - // Tool call tracking - toolCalls map[string]*llm.ToolCall // callID -> tool call - itemToCallID map[string]string // item.id -> call_id mapping - toolCallIndex map[string]int // callID -> index in the output - // Reasoning signature tracking pendingReasoningEncryptedContent map[string]*string @@ -75,11 +67,9 @@ type outboundStreamState struct { func newResponsesOutboundStream(stream streams.Stream[*httpclient.StreamEvent]) *responsesOutboundStream { return &responsesOutboundStream{ - stream: stream, + stream: stream, + aggregator: newStreamAggregator(), state: &outboundStreamState{ - toolCalls: make(map[string]*llm.ToolCall), - itemToCallID: make(map[string]string), - toolCallIndex: make(map[string]int), pendingReasoningEncryptedContent: make(map[string]*string), transformerMetadata: make(map[string]any), }, @@ -152,6 +142,13 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE if slog.Default().Enabled(context.Background(), slog.LevelDebug) { slog.DebugContext(context.Background(), "received response stream event", slog.Any("event", streamEvent)) } + if s.responseCompleted { + return nil + } + s.aggregator.processEvent(&streamEvent) + if s.aggregator.err != nil { + return s.aggregator.err + } // Build base response resp := &llm.Response{ @@ -225,175 +222,20 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE s.state.pendingReasoningEncryptedContent[item.ID] = shared.EncodeOpenAIEncryptedContent(item.EncryptedContent) return nil - case "function_call": - // Initialize tool call tracking - toolCallIdx := len(s.state.toolCalls) - s.state.toolCalls[item.CallID] = &llm.ToolCall{ - ID: item.CallID, - Type: "function", - Function: llm.FunctionCall{ - Name: item.Name, - Namespace: item.Namespace, - Arguments: "", - }, - } - // Map item.id to call_id for later lookup - s.state.itemToCallID[item.ID] = item.CallID - s.state.toolCallIndex[item.CallID] = toolCallIdx - - resp.Choices = []llm.Choice{ - { - Index: 0, - Delta: &llm.Message{ - ToolCalls: []llm.ToolCall{ - { - ID: item.CallID, - Type: "function", - Index: toolCallIdx, - Function: llm.FunctionCall{ - Name: item.Name, - Namespace: item.Namespace, - }, - }, - }, - }, - }, - } - - case "custom_tool_call": - // Custom tool call - initialize tracking, input will be streamed via delta events - toolCallIdx := len(s.state.toolCalls) - s.state.toolCalls[item.CallID] = &llm.ToolCall{ - ID: item.CallID, - Type: llm.ToolTypeResponsesCustomTool, - ResponseCustomToolCall: &llm.ResponseCustomToolCall{ - CallID: item.CallID, - Name: item.Name, - Input: "", - }, - } - s.state.itemToCallID[item.ID] = item.CallID - s.state.toolCallIndex[item.CallID] = toolCallIdx - - resp.Choices = []llm.Choice{ - { - Index: 0, - Delta: &llm.Message{ - ToolCalls: []llm.ToolCall{ - { - ID: item.CallID, - Type: llm.ToolTypeResponsesCustomTool, - Index: toolCallIdx, - ResponseCustomToolCall: &llm.ResponseCustomToolCall{ - CallID: item.CallID, - Name: item.Name, - }, - }, - }, - }, - }, - } + case "function_call", "custom_tool_call": + return nil default: // For other item types (e.g., message), skip - no meaningful content to emit return nil // Intentionally skip this event } - case StreamEventTypeFunctionCallArgumentsDelta: - // Function call arguments delta - if streamEvent.ItemID != nil { - // Look up call_id from item_id mapping - callID, ok := s.state.itemToCallID[*streamEvent.ItemID] - if !ok { - // Fallback: item_id might be the call_id itself - callID = *streamEvent.ItemID - } - - if tc, ok := s.state.toolCalls[callID]; ok { - tc.Function.Arguments += streamEvent.Delta - toolCallIdx := s.state.toolCallIndex[callID] - - resp.Choices = []llm.Choice{ - { - Index: 0, - Delta: &llm.Message{ - ToolCalls: []llm.ToolCall{ - { - Index: toolCallIdx, - Function: llm.FunctionCall{ - Arguments: streamEvent.Delta, - }, - }, - }, - }, - }, - } - } - } - - case StreamEventTypeFunctionCallArgumentsDone: - // Function call completed - update state but don't emit an event - if streamEvent.CallID != "" { - if tc, ok := s.state.toolCalls[streamEvent.CallID]; ok { - if streamEvent.Name != "" { - tc.Function.Name = streamEvent.Name - } - if streamEvent.Namespace != "" { - tc.Function.Namespace = streamEvent.Namespace - } - tc.Function.Arguments = streamEvent.Arguments - } - } - - return nil // Intentionally skip this event - - case StreamEventTypeCustomToolCallInputDelta: - // Custom tool call input delta - accumulate and emit as tool call delta - if streamEvent.ItemID != nil { - callID, ok := s.state.itemToCallID[*streamEvent.ItemID] - if !ok { - callID = *streamEvent.ItemID - } - - if tc, ok := s.state.toolCalls[callID]; ok { - tc.ResponseCustomToolCall.Input += streamEvent.Delta - toolCallIdx := s.state.toolCallIndex[callID] - - resp.Choices = []llm.Choice{ - { - Index: 0, - Delta: &llm.Message{ - ToolCalls: []llm.ToolCall{ - { - Index: toolCallIdx, - Type: llm.ToolTypeResponsesCustomTool, - ResponseCustomToolCall: &llm.ResponseCustomToolCall{ - CallID: callID, - Name: tc.ResponseCustomToolCall.Name, - Input: streamEvent.Delta, - }, - }, - }, - }, - }, - } - } - } - - case StreamEventTypeCustomToolCallInputDone: - // Custom tool call input completed - update state but don't emit an event - if streamEvent.ItemID != nil { - callID, ok := s.state.itemToCallID[*streamEvent.ItemID] - if !ok { - callID = *streamEvent.ItemID - } - - if tc, ok := s.state.toolCalls[callID]; ok { - tc.ResponseCustomToolCall.Input = streamEvent.Input - } - } - - return nil // Intentionally skip this event + case StreamEventTypeFunctionCallArgumentsDelta, + StreamEventTypeFunctionCallArgumentsDone, + StreamEventTypeCustomToolCallInputDelta, + StreamEventTypeCustomToolCallInputDone: + // streamAggregator owns tool identity and terminal payload reconciliation. + return nil case StreamEventTypeContentPartAdded: // Content part added - skip, no meaningful content to emit @@ -401,8 +243,6 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE case StreamEventTypeOutputTextDelta: // Text content delta - s.state.textContent.WriteString(streamEvent.Delta) - resp.Choices = []llm.Choice{ { Index: 0, @@ -416,8 +256,6 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE case StreamEventTypeReasoningSummaryTextDelta: // Reasoning content delta - s.state.reasoningContent.WriteString(streamEvent.Delta) - resp.Choices = []llm.Choice{ { Index: 0, @@ -439,6 +277,9 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE if streamEvent.Item == nil { return nil // Intentionally skip this event } + if streamEvent.Item.Type == "function_call" || streamEvent.Item.Type == "custom_tool_call" { + return nil + } if streamEvent.Item.Type == "web_search_call" { appendResponseWebSearchCallMetadata(s.state.transformerMetadata, *streamEvent.Item) return nil // Intentionally skip this event @@ -502,18 +343,21 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE case StreamEventTypeResponseCompleted: // Response completed - emit two events: one with finish_reason, one with usage - s.responseCompleted = true + if !s.beginTerminalEvent() { + return nil + } if streamEvent.Response != nil { s.state.previousResponseID = streamEvent.Response.PreviousResponseID resp.PreviousResponseID = s.state.previousResponseID } + toolCallCount := s.enqueueAggregatedToolCalls() if len(s.state.transformerMetadata) > 0 && !s.state.transformerMetadataEmitted { resp.TransformerMetadata = s.state.transformerMetadata s.state.transformerMetadataEmitted = true } finishReason := "stop" - if len(s.state.toolCalls) > 0 { + if toolCallCount > 0 { finishReason = "tool_calls" } @@ -547,7 +391,9 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE case StreamEventTypeResponseFailed: // Response failed - s.responseCompleted = true + if !s.beginTerminalEvent() { + return nil + } finishReason := "error" resp.Choices = []llm.Choice{ { @@ -558,7 +404,9 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE case StreamEventTypeResponseIncomplete: // Response incomplete (e.g., max tokens) - s.responseCompleted = true + if !s.beginTerminalEvent() { + return nil + } finishReason := "length" resp.Choices = []llm.Choice{ { @@ -569,7 +417,9 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE case StreamEventTypeResponseCancelled: // Response cancelled - s.responseCompleted = true + if !s.beginTerminalEvent() { + return nil + } finishReason := "cancelled" resp.Choices = []llm.Choice{ { @@ -630,6 +480,100 @@ func (s *responsesOutboundStream) transformStreamChunk(event *httpclient.StreamE return nil } +func (s *responsesOutboundStream) beginTerminalEvent() bool { + if s.responseCompleted { + return false + } + s.responseCompleted = true + return true +} + +func (s *responsesOutboundStream) enqueueAggregatedToolCalls() int { + toolCalls := s.aggregator.completedToolCalls() + for index, aggregated := range toolCalls { + message := convertOutputToMessage([]Item{aggregated.Item}, nil) + if len(message.ToolCalls) != 1 { + continue + } + + toolCall := message.ToolCalls[0] + toolCall.Index = index + finalPayload := toolCall.Function.Arguments + if toolCall.ResponseCustomToolCall != nil { + finalPayload = toolCall.ResponseCustomToolCall.Input + } + + deltas := reconcileToolPayload(aggregated.Deltas, finalPayload) + identity := toolCall + if len(aggregated.Deltas) > 0 { + if identity.ResponseCustomToolCall != nil { + customCall := *identity.ResponseCustomToolCall + customCall.Input = "" + identity.ResponseCustomToolCall = &customCall + } else { + identity.Function.Arguments = "" + } + } + s.enqueueToolCallChunk(identity) + + for _, delta := range deltas { + toolCallDelta := llm.ToolCall{Index: index} + if toolCall.ResponseCustomToolCall != nil { + toolCallDelta.Type = llm.ToolTypeResponsesCustomTool + toolCallDelta.ResponseCustomToolCall = &llm.ResponseCustomToolCall{ + CallID: toolCall.ID, + Name: toolCall.ResponseCustomToolCall.Name, + Namespace: toolCall.ResponseCustomToolCall.Namespace, + Input: delta, + } + } else { + toolCallDelta.Function.Arguments = delta + } + s.enqueueToolCallChunk(toolCallDelta) + } + } + + return len(toolCalls) +} + +func reconcileToolPayload(deltas []string, finalPayload string) []string { + if len(deltas) == 0 { + return nil + } + + streamedPayload := strings.Join(deltas, "") + switch { + case finalPayload == streamedPayload: + return deltas + case strings.HasPrefix(finalPayload, streamedPayload): + if suffix := finalPayload[len(streamedPayload):]; suffix != "" { + return append(append([]string(nil), deltas...), suffix) + } + return deltas + default: + return []string{finalPayload} + } +} + +func (s *responsesOutboundStream) enqueueToolCallChunk(toolCall llm.ToolCall) { + resp := s.newResponseChunk() + resp.Choices = []llm.Choice{{ + Index: 0, + Delta: &llm.Message{ToolCalls: []llm.ToolCall{toolCall}}, + }} + s.enqueue(resp) +} + +func (s *responsesOutboundStream) newResponseChunk() *llm.Response { + return &llm.Response{ + Object: "chat.completion.chunk", + ID: s.state.responseID, + Model: s.state.responseModel, + Created: s.state.created, + PreviousResponseID: s.state.previousResponseID, + } +} + func (s *responsesOutboundStream) Current() *llm.Response { if s.queueIndex < len(s.eventQueue) { event := s.eventQueue[s.queueIndex] diff --git a/llm/transformer/openai/responses/stream_tool_cross_protocol_test.go b/llm/transformer/openai/responses/stream_tool_cross_protocol_test.go new file mode 100644 index 000000000..2d6408ae4 --- /dev/null +++ b/llm/transformer/openai/responses/stream_tool_cross_protocol_test.go @@ -0,0 +1,139 @@ +package responses_test + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/looplj/axonhub/llm" + "github.com/looplj/axonhub/llm/httpclient" + "github.com/looplj/axonhub/llm/streams" + "github.com/looplj/axonhub/llm/transformer/anthropic" + "github.com/looplj/axonhub/llm/transformer/openai" + "github.com/looplj/axonhub/llm/transformer/openai/responses" +) + +func TestResponsesStreamCorrectedFunctionArgumentsRemainValidAcrossInboundFormats(t *testing.T) { + const finalArguments = `{"description":"different"}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_cross_protocol","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_cross_protocol","type":"function_call","status":"in_progress","call_id":"call_cross_protocol","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_cross_protocol","output_index":0,"delta":"{\"description\":\"first\"}"}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_cross_protocol","output_index":0,"arguments":"{\"description\":\"function-done\"}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_cross_protocol","type":"function_call","status":"completed","call_id":"call_cross_protocol","name":"spawn_agent","arguments":"{\"description\":\"different\"}"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_cross_protocol","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_cross_protocol","type":"function_call","status":"completed","call_id":"call_cross_protocol","name":"spawn_agent","arguments":"{\"description\":\"different\"}"}]}}`)}, + } + + t.Run("OpenAI Chat", func(t *testing.T) { + canonical := responsesCanonicalStream(t, upstreamEvents) + client, err := openai.NewInboundTransformer().TransformStream(t.Context(), canonical) + require.NoError(t, err) + + var arguments strings.Builder + for client.Next() { + if string(client.Current().Data) == "[DONE]" { + continue + } + + var event openai.Response + require.NoError(t, json.Unmarshal(client.Current().Data, &event)) + for _, choice := range event.Choices { + if choice.Delta == nil { + continue + } + for _, toolCall := range choice.Delta.ToolCalls { + arguments.WriteString(toolCall.Function.Arguments) + } + } + } + require.NoError(t, client.Err()) + require.Equal(t, finalArguments, arguments.String()) + }) + + t.Run("Anthropic Messages", func(t *testing.T) { + canonical := responsesCanonicalStream(t, upstreamEvents) + client, err := anthropic.NewInboundTransformer().TransformStream(t.Context(), canonical) + require.NoError(t, err) + + var arguments strings.Builder + for client.Next() { + var event struct { + Delta *struct { + PartialJSON *string `json:"partial_json"` + } `json:"delta"` + } + require.NoError(t, json.Unmarshal(client.Current().Data, &event)) + if event.Delta != nil && event.Delta.PartialJSON != nil { + arguments.WriteString(*event.Delta.PartialJSON) + } + } + require.NoError(t, client.Err()) + require.Equal(t, finalArguments, arguments.String()) + }) +} + +func TestResponsesStreamMultipleToolsRemainContiguousForAnthropic(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_multiple_tools","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_multiple_tools_0","type":"function_call","status":"in_progress","call_id":"call_multiple_tools_0","name":"first","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_multiple_tools_0","output_index":0,"delta":"{\"value\":"}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":1,"item":{"id":"fc_multiple_tools_1","type":"function_call","status":"in_progress","call_id":"call_multiple_tools_1","name":"second","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_multiple_tools_1","output_index":1,"delta":"{\"value\":2}"}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_multiple_tools_0","output_index":0,"name":"first","arguments":"{\"value\":1}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_multiple_tools_0","type":"function_call","status":"completed","call_id":"call_multiple_tools_0","name":"first","arguments":"{\"value\":1}"}}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_multiple_tools_1","output_index":1,"name":"second","arguments":"{\"value\":2}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":1,"item":{"id":"fc_multiple_tools_1","type":"function_call","status":"completed","call_id":"call_multiple_tools_1","name":"second","arguments":"{\"value\":2}"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_multiple_tools","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_multiple_tools_0","type":"function_call","status":"completed","call_id":"call_multiple_tools_0","name":"first","arguments":"{\"value\":1}"},{"id":"fc_multiple_tools_1","type":"function_call","status":"completed","call_id":"call_multiple_tools_1","name":"second","arguments":"{\"value\":2}"}]}}`)}, + } + + canonical := responsesCanonicalStream(t, upstreamEvents) + client, err := anthropic.NewInboundTransformer().TransformStream(t.Context(), canonical) + require.NoError(t, err) + + toolNameByBlock := map[int]string{} + argumentsByTool := map[string]*strings.Builder{} + for client.Next() { + var event struct { + Type string `json:"type"` + Index *int `json:"index"` + ContentBlock *struct { + Type string `json:"type"` + Name *string `json:"name"` + } `json:"content_block"` + Delta *struct { + PartialJSON *string `json:"partial_json"` + } `json:"delta"` + } + require.NoError(t, json.Unmarshal(client.Current().Data, &event)) + if event.Index == nil { + continue + } + if event.Type == "content_block_start" && event.ContentBlock != nil && event.ContentBlock.Name != nil { + toolNameByBlock[*event.Index] = *event.ContentBlock.Name + argumentsByTool[*event.ContentBlock.Name] = &strings.Builder{} + } + if event.Type == "content_block_delta" && event.Delta != nil && event.Delta.PartialJSON != nil { + name := toolNameByBlock[*event.Index] + if argumentsByTool[name] != nil { + argumentsByTool[name].WriteString(*event.Delta.PartialJSON) + } + } + } + require.NoError(t, client.Err()) + require.Equal(t, `{"value":1}`, argumentsByTool["first"].String()) + require.Equal(t, `{"value":2}`, argumentsByTool["second"].String()) +} + +func responsesCanonicalStream(t *testing.T, upstreamEvents []*httpclient.StreamEvent) streams.Stream[*llm.Response] { + t.Helper() + + upstream, err := responses.NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := upstream.TransformStream(t.Context(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + + return canonical +} diff --git a/llm/transformer/openai/responses/stream_tool_roundtrip_test.go b/llm/transformer/openai/responses/stream_tool_roundtrip_test.go new file mode 100644 index 000000000..6c6aab48d --- /dev/null +++ b/llm/transformer/openai/responses/stream_tool_roundtrip_test.go @@ -0,0 +1,934 @@ +package responses + +import ( + "context" + "encoding/json" + "strconv" + "strings" + "testing" + + "github.com/samber/lo" + "github.com/stretchr/testify/require" + + "github.com/looplj/axonhub/llm" + "github.com/looplj/axonhub/llm/httpclient" + "github.com/looplj/axonhub/llm/streams" +) + +func TestResponsesStreamRoundTrip_PreservesFunctionCallArgumentsFromFinalEvents(t *testing.T) { + const arguments = `{"description":"Inspect the repository and report the relevant files.","model":"gpt-5.5"}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_spawn_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_spawn_1","type":"function_call","status":"in_progress","call_id":"call_spawn_1","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_spawn_1","output_index":0,"arguments":` + strconv.Quote(arguments) + `}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_spawn_1","type":"function_call","status":"completed","call_id":"call_spawn_1","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_spawn_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_spawn_1","type":"function_call","status":"completed","call_id":"call_spawn_1","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, "function_call", completed.Output[0].Type) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_OutputItemDoneCarriesCompleteFunctionCall(t *testing.T) { + const arguments = `{"description":"Persist the complete tool call."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_done_item","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_done_item","type":"function_call","status":"in_progress","call_id":"call_done_item","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_done_item","output_index":0,"arguments":` + strconv.Quote(arguments) + `}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_done_item","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_done_item","type":"function_call","status":"completed","call_id":"call_done_item","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + var doneItem *Item + for _, event := range roundTripResponseEvents(t, upstreamEvents) { + if event.Type == StreamEventTypeOutputItemDone && event.Item != nil && event.Item.Type == "function_call" { + doneItem = event.Item + } + } + require.NotNil(t, doneItem) + require.Equal(t, "fc_done_item", doneItem.ID) + require.Equal(t, "call_done_item", doneItem.CallID) + require.Equal(t, "spawn_agent", doneItem.Name) + require.Equal(t, "collaboration", doneItem.Namespace) + require.Equal(t, arguments, doneItem.Arguments) +} + +func TestResponsesStreamRoundTrip_PreservesFunctionCallArgumentsWhenArgumentsDoneIsMissing(t *testing.T) { + const arguments = `{"description":"Audit the stream transformer."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_spawn_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_spawn_2","type":"function_call","status":"in_progress","call_id":"call_spawn_2","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_spawn_2","type":"function_call","status":"completed","call_id":"call_spawn_2","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_spawn_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_spawn_2","type":"function_call","status":"completed","call_id":"call_spawn_2","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_PreservesFunctionCallArgumentsFromCompletedSnapshot(t *testing.T) { + const arguments = `{"description":"Use the terminal response snapshot."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_spawn_3","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_spawn_3","type":"function_call","status":"in_progress","call_id":"call_spawn_3","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_spawn_3","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_spawn_3","type":"function_call","status":"completed","call_id":"call_spawn_3","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_PreservesFunctionCallFromOutputItemDoneWithoutAdded(t *testing.T) { + const arguments = `{"description":"Recover the final tool item."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_final_only_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_final_only_done","type":"function_call","status":"completed","call_id":"call_final_only_done","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_final_only_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, "fc_final_only_done", completed.Output[0].ID) + require.Equal(t, "call_final_only_done", completed.Output[0].CallID) + require.Equal(t, "spawn_agent", completed.Output[0].Name) + require.Equal(t, "collaboration", completed.Output[0].Namespace) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_PreservesCustomToolFromCompletedSnapshotWithoutAdded(t *testing.T) { + const input = "patch" + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_final_only_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_final_only_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_final_only_completed","type":"custom_tool_call","status":"completed","call_id":"call_final_only_completed","name":"apply_patch","namespace":"mcp__codex","input":` + strconv.Quote(input) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, "ctc_final_only_completed", completed.Output[0].ID) + require.Equal(t, "call_final_only_completed", completed.Output[0].CallID) + require.Equal(t, "apply_patch", completed.Output[0].Name) + require.Equal(t, "mcp__codex", completed.Output[0].Namespace) + require.NotNil(t, completed.Output[0].Input) + require.Equal(t, input, *completed.Output[0].Input) +} + +func TestResponsesOutboundStream_CreatesToolCallFromFinalItemWithoutAdded(t *testing.T) { + const arguments = `{"description":"Recover outbound state."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_outbound_final_only","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_outbound_final_only","type":"function_call","status":"completed","call_id":"call_outbound_final_only","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_outbound_final_only","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + } + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformStream(t.Context(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + chunks, err := streams.All(canonical) + require.NoError(t, err) + + var toolCall *llm.ToolCall + for _, chunk := range chunks { + for _, choice := range chunk.Choices { + if choice.Delta == nil || len(choice.Delta.ToolCalls) == 0 { + continue + } + candidate := choice.Delta.ToolCalls[0] + toolCall = &candidate + } + } + require.NotNil(t, toolCall) + require.Equal(t, "call_outbound_final_only", toolCall.ID) + require.Equal(t, "fc_outbound_final_only", toolCall.ResponseItemID) + require.Equal(t, "spawn_agent", toolCall.Function.Name) + require.Equal(t, "collaboration", toolCall.Function.Namespace) + require.Equal(t, arguments, toolCall.Function.Arguments) +} + +func TestResponsesStreamRoundTrip_GeneratesStableItemIDWhenAddedOmitsIt(t *testing.T) { + const arguments = `{"description":"Recover item identity."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_missing_item_id","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","status":"in_progress","call_id":"call_missing_item_id","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_missing_item_id","type":"function_call","status":"completed","call_id":"call_missing_item_id","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_missing_item_id","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_missing_item_id","type":"function_call","status":"completed","call_id":"call_missing_item_id","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + var completed *Response + var addedItemID, doneItemID string + var addedCount, doneCount int + for _, event := range roundTripResponseEvents(t, upstreamEvents) { + if event.Type == StreamEventTypeOutputItemAdded && event.Item != nil && event.Item.Type == "function_call" { + addedCount++ + addedItemID = event.Item.ID + require.True(t, strings.HasPrefix(addedItemID, "fc_")) + require.NotEqual(t, "call_missing_item_id", addedItemID) + require.Equal(t, "call_missing_item_id", event.Item.CallID) + } + if event.Type == StreamEventTypeOutputItemDone && event.Item != nil && event.Item.Type == "function_call" { + doneCount++ + doneItemID = event.Item.ID + require.Equal(t, "call_missing_item_id", event.Item.CallID) + } + if event.Type == StreamEventTypeResponseCompleted { + completed = event.Response + } + } + require.Equal(t, 1, addedCount) + require.Equal(t, 1, doneCount) + require.Equal(t, addedItemID, doneItemID) + require.NotNil(t, completed) + require.Len(t, completed.Output, 1) + require.Equal(t, addedItemID, completed.Output[0].ID) + require.Equal(t, "call_missing_item_id", completed.Output[0].CallID) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_AssociatesFunctionDeltaByOutputIndexWhenAddedOmitsItemID(t *testing.T) { + const arguments = `{"description":"Keep the streamed arguments."}` + + completed := roundTripCompletedResponse(t, []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_partial_function_identity","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","status":"in_progress","call_id":"call_partial_function_identity","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_partial_function_identity","output_index":0,"delta":` + strconv.Quote(arguments) + `}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_partial_function_identity","type":"function_call","status":"completed","call_id":"call_partial_function_identity","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_partial_function_identity","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + + require.Len(t, completed.Output, 1) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_AssociatesCustomDeltaByOutputIndexWhenAddedOmitsItemID(t *testing.T) { + const input = "patch" + + completed := roundTripCompletedResponse(t, []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_partial_custom_identity","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"custom_tool_call","status":"in_progress","call_id":"call_partial_custom_identity","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_partial_custom_identity","output_index":0,"delta":"patch"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_partial_custom_identity","type":"custom_tool_call","status":"completed","call_id":"call_partial_custom_identity","name":"apply_patch"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_partial_custom_identity","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + + require.Len(t, completed.Output, 1) + require.NotNil(t, completed.Output[0].Input) + require.Equal(t, input, *completed.Output[0].Input) +} + +func TestResponsesStreamRoundTrip_DoesNotMergeDifferentToolTypesThatReuseOutputIndex(t *testing.T) { + completed := roundTripCompletedResponse(t, []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_reused_index_types","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","status":"in_progress","call_id":"call_reused_function","name":"wait","arguments":""}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_reused_custom","type":"custom_tool_call","status":"in_progress","name":"apply_patch","input":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_reused_function","type":"function_call","status":"completed","call_id":"call_reused_function","name":"wait","arguments":"{}"}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_reused_custom","type":"custom_tool_call","status":"completed","call_id":"call_reused_custom","name":"apply_patch","input":"patch"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_reused_index_types","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + + require.Len(t, completed.Output, 2) + itemsByCallID := map[string]Item{} + for _, item := range completed.Output { + itemsByCallID[item.CallID] = item + } + require.Equal(t, "function_call", itemsByCallID["call_reused_function"].Type) + require.Equal(t, "{}", itemsByCallID["call_reused_function"].Arguments) + require.Equal(t, "custom_tool_call", itemsByCallID["call_reused_custom"].Type) + require.Equal(t, "patch", lo.FromPtr(itemsByCallID["call_reused_custom"].Input)) +} + +func TestResponsesStreamRoundTrip_DoesNotMergeSameTypeToolsThatReuseOutputIndex(t *testing.T) { + completed := roundTripCompletedResponse(t, []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_reused_index_functions","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","status":"in_progress","call_id":"call_reused_function_a","name":"wait","arguments":""}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_reused_function_b","type":"function_call","status":"in_progress","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_reused_function_a","type":"function_call","status":"completed","call_id":"call_reused_function_a","name":"wait","arguments":"{}"}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_reused_function_b","type":"function_call","status":"completed","call_id":"call_reused_function_b","name":"spawn_agent","arguments":"{\"description\":\"second\"}"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_reused_index_functions","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + }) + + require.Len(t, completed.Output, 2) + itemsByCallID := map[string]Item{} + for _, item := range completed.Output { + itemsByCallID[item.CallID] = item + } + require.Equal(t, "{}", itemsByCallID["call_reused_function_a"].Arguments) + require.Equal(t, `{"description":"second"}`, itemsByCallID["call_reused_function_b"].Arguments) +} + +func TestResponsesOutboundStream_RejectsAmbiguousLateItemIDAfterOutputIndexReuse(t *testing.T) { + upstream, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := upstream.TransformStream(t.Context(), nil, streams.SliceStream([]*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_ambiguous_index","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","status":"in_progress","call_id":"call_ambiguous_a","name":"wait","arguments":""}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","status":"in_progress","call_id":"call_ambiguous_b","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_ambiguous_a","output_index":0,"delta":"{}"}`)}, + })) + require.NoError(t, err) + _, err = streams.All(canonical) + require.ErrorContains(t, err, "ambiguous tool call output_index 0") +} + +func TestResponsesStreamRoundTrip_FinalItemSuppliesMissingCallID(t *testing.T) { + const arguments = `{"description":"Recover call identity."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_missing_call_id","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_missing_call_id","type":"function_call","status":"in_progress","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_missing_call_id","type":"function_call","status":"completed","call_id":"call_missing_call_id","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_missing_call_id","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_missing_call_id","type":"function_call","status":"completed","call_id":"call_missing_call_id","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + var completed *Response + var addedCount, doneCount int + for _, event := range roundTripResponseEvents(t, upstreamEvents) { + if event.Type == StreamEventTypeOutputItemAdded && event.Item != nil && event.Item.Type == "function_call" { + addedCount++ + require.Equal(t, "fc_missing_call_id", event.Item.ID) + require.Equal(t, "call_missing_call_id", event.Item.CallID) + } + if event.Type == StreamEventTypeOutputItemDone && event.Item != nil && event.Item.Type == "function_call" { + doneCount++ + require.Equal(t, "fc_missing_call_id", event.Item.ID) + require.Equal(t, "call_missing_call_id", event.Item.CallID) + } + if event.Type == StreamEventTypeResponseCompleted { + completed = event.Response + } + } + require.Equal(t, 1, addedCount) + require.Equal(t, 1, doneCount) + require.NotNil(t, completed) + require.Len(t, completed.Output, 1) + require.Equal(t, "fc_missing_call_id", completed.Output[0].ID) + require.Equal(t, "call_missing_call_id", completed.Output[0].CallID) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_EmptyFinalArgumentsDoNotEraseStreamedArguments(t *testing.T) { + const arguments = `{"description":"Keep streamed arguments."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_empty_final","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_empty_final","type":"function_call","status":"in_progress","call_id":"call_empty_final","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_empty_final","output_index":0,"delta":` + strconv.Quote(arguments) + `}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_empty_final","output_index":0,"arguments":""}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_empty_final","type":"function_call","status":"completed","call_id":"call_empty_final","name":"spawn_agent","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_empty_final","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_AcceptsConflictingFinalFunctionArguments(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_conflict","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_conflict","type":"function_call","status":"in_progress","call_id":"call_conflict","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_conflict","output_index":0,"delta":"{\"description\":\"first\"}"}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_conflict","output_index":0,"arguments":"{\"description\":\"function-done\"}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_conflict","type":"function_call","status":"completed","call_id":"call_conflict","name":"spawn_agent","arguments":"{\"description\":\"different\"}"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_conflict","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_conflict","type":"function_call","status":"completed","call_id":"call_conflict","name":"spawn_agent","arguments":"{\"description\":\"different\"}"}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, `{"description":"different"}`, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_CompletedSnapshotCorrectsStreamedFunctionArguments(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_completed_correction","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_completed_correction","type":"function_call","status":"in_progress","call_id":"call_completed_correction","name":"search","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_completed_correction","output_index":0,"delta":"{}"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_completed_correction","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_completed_correction","type":"function_call","status":"completed","call_id":"call_completed_correction","name":"search","arguments":"{\"query\":\"x\"}"}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, `{"query":"x"}`, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_AcceptsConflictingFinalCustomToolInput(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_custom_conflict","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_conflict","type":"custom_tool_call","status":"in_progress","call_id":"call_custom_conflict","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_conflict","output_index":0,"delta":"draft"}`)}, + {Type: "response.custom_tool_call_input.done", Data: []byte(`{"type":"response.custom_tool_call_input.done","item_id":"ctc_conflict","output_index":0,"input":"final"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_conflict","type":"custom_tool_call","status":"completed","call_id":"call_custom_conflict","name":"apply_patch","input":"output-item-final"}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_custom_conflict","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_conflict","type":"custom_tool_call","status":"completed","call_id":"call_custom_conflict","name":"apply_patch","input":"output-item-final"}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.NotNil(t, completed.Output[0].Input) + require.Equal(t, "output-item-final", *completed.Output[0].Input) +} + +func TestResponsesStreamRoundTrip_OutputItemDoneOverridesLateFunctionEvents(t *testing.T) { + const finalArguments = `{"source":"output-item"}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_late_function_events","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_late_function_events","type":"function_call","status":"in_progress","call_id":"call_late_function_events","name":"search","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_late_function_events","output_index":0,"delta":"{}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_late_function_events","type":"function_call","status":"completed","call_id":"call_late_function_events","name":"search","arguments":` + strconv.Quote(finalArguments) + `}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_late_function_events","output_index":0,"delta":"late"}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_late_function_events","output_index":0,"name":"search","arguments":"{\"source\":\"late-done\"}"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_late_function_events","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_late_function_events","type":"function_call","status":"completed","call_id":"call_late_function_events","name":"search","arguments":"{\"source\":\"completed\"}"}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, finalArguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_OutputItemDoneOverridesLateCustomToolEvents(t *testing.T) { + const finalInput = "output-item" + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_late_custom_events","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_late_custom_events","type":"custom_tool_call","status":"in_progress","call_id":"call_late_custom_events","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_late_custom_events","output_index":0,"delta":"draft"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_late_custom_events","type":"custom_tool_call","status":"completed","call_id":"call_late_custom_events","name":"apply_patch","input":"output-item"}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_late_custom_events","output_index":0,"delta":"late"}`)}, + {Type: "response.custom_tool_call_input.done", Data: []byte(`{"type":"response.custom_tool_call_input.done","item_id":"ctc_late_custom_events","output_index":0,"input":"late-done"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_late_custom_events","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_late_custom_events","type":"custom_tool_call","status":"completed","call_id":"call_late_custom_events","name":"apply_patch","input":"completed"}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.NotNil(t, completed.Output[0].Input) + require.Equal(t, finalInput, *completed.Output[0].Input) +} + +func TestResponsesStreamRoundTrip_ExplicitEmptyFinalCustomToolInputClearsStreamedInput(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_custom_empty_final","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_custom_empty_final","type":"custom_tool_call","status":"in_progress","call_id":"call_custom_empty_final","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_custom_empty_final","output_index":0,"delta":"draft"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_custom_empty_final","type":"custom_tool_call","status":"completed","call_id":"call_custom_empty_final","name":"apply_patch","input":""}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_custom_empty_final","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_custom_empty_final","type":"custom_tool_call","status":"completed","call_id":"call_custom_empty_final","name":"apply_patch","input":""}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.NotNil(t, completed.Output[0].Input) + require.Empty(t, *completed.Output[0].Input) +} + +func TestResponsesStreamRoundTrip_CompletedSnapshotExplicitEmptyCustomInputClearsStreamedInput(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_custom_empty_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_custom_empty_completed","type":"custom_tool_call","status":"in_progress","call_id":"call_custom_empty_completed","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.delta", Data: []byte(`{"type":"response.custom_tool_call_input.delta","item_id":"ctc_custom_empty_completed","output_index":0,"delta":"draft"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_custom_empty_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_custom_empty_completed","type":"custom_tool_call","status":"completed","call_id":"call_custom_empty_completed","name":"apply_patch","input":""}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.NotNil(t, completed.Output[0].Input) + require.Empty(t, *completed.Output[0].Input) +} + +func TestResponsesOutboundStream_DoesNotEmitProvisionalToolCallOnCompletedResponse(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_provisional_tool","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_provisional_tool","type":"function_call","status":"in_progress","call_id":"call_provisional_tool","name":"search","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_provisional_tool","output_index":0,"delta":"{}"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_provisional_tool","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[]}}`)}, + } + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformStream(t.Context(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + chunks, err := streams.All(canonical) + require.NoError(t, err) + + for _, chunk := range chunks { + for _, choice := range chunk.Choices { + if choice.Delta != nil { + require.Empty(t, choice.Delta.ToolCalls) + } + } + } +} + +func TestResponsesOutboundStream_DoesNotEmitToolCallWhenResponseFails(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_failed_tool","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_failed_tool","type":"function_call","status":"in_progress","call_id":"call_failed_tool","name":"search","arguments":""}}`)}, + {Type: "response.function_call_arguments.delta", Data: []byte(`{"type":"response.function_call_arguments.delta","item_id":"fc_failed_tool","output_index":0,"delta":"{}"}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_failed_tool","type":"function_call","status":"completed","call_id":"call_failed_tool","name":"search","arguments":"{}"}}`)}, + {Type: "response.failed", Data: []byte(`{"type":"response.failed","response":{"id":"resp_failed_tool","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"failed","output":[]}}`)}, + } + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformStream(t.Context(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + chunks, err := streams.All(canonical) + require.NoError(t, err) + + for _, chunk := range chunks { + for _, choice := range chunk.Choices { + if choice.Delta != nil { + require.Empty(t, choice.Delta.ToolCalls) + } + } + } +} + +func TestResponsesOutboundStream_DuplicateCompletedIsIdempotent(t *testing.T) { + completedEvent := []byte(`{"type":"response.completed","response":{"id":"resp_duplicate_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[],"usage":{"input_tokens":10,"output_tokens":2,"total_tokens":12}}}`) + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_duplicate_completed","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.completed", Data: completedEvent}, + {Type: "response.completed", Data: completedEvent}, + } + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformStream(t.Context(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + chunks, err := streams.All(canonical) + require.NoError(t, err) + + var finishCount, usageCount int + for _, chunk := range chunks { + for _, choice := range chunk.Choices { + if choice.FinishReason != nil { + finishCount++ + } + } + if chunk.Usage != nil { + usageCount++ + } + } + require.Equal(t, 1, finishCount) + require.Equal(t, 1, usageCount) +} + +func TestResponsesStreamRoundTrip_PreservesMultipleFinalOnlyTools(t *testing.T) { + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_multiple_final_only","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_multiple_final_only","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_multiple_1","type":"function_call","status":"completed","call_id":"call_multiple_1","name":"spawn_agent","namespace":"collaboration","arguments":"{\"description\":\"one\"}"},{"id":"ctc_multiple_2","type":"custom_tool_call","status":"completed","call_id":"call_multiple_2","name":"apply_patch","namespace":"mcp__codex","input":"patch"}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 2) + require.Equal(t, "fc_multiple_1", completed.Output[0].ID) + require.Equal(t, "call_multiple_1", completed.Output[0].CallID) + require.Equal(t, "ctc_multiple_2", completed.Output[1].ID) + require.Equal(t, "call_multiple_2", completed.Output[1].CallID) +} + +func TestResponsesOutboundStream_DuplicateOutputItemDoneDoesNotRepeatArguments(t *testing.T) { + const arguments = `{"description":"emit once"}` + finalItem := `{"id":"fc_duplicate_done","type":"function_call","status":"completed","call_id":"call_duplicate_done","name":"spawn_agent","arguments":` + strconv.Quote(arguments) + `}` + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_duplicate_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_duplicate_done","type":"function_call","status":"in_progress","call_id":"call_duplicate_done","name":"spawn_agent","arguments":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":` + finalItem + `}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":` + finalItem + `}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_duplicate_done","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[` + finalItem + `]}}`)}, + } + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformStream(t.Context(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + chunks, err := streams.All(canonical) + require.NoError(t, err) + + var argumentChunks int + var streamedArguments strings.Builder + for _, chunk := range chunks { + for _, choice := range chunk.Choices { + if choice.Delta == nil { + continue + } + for _, toolCall := range choice.Delta.ToolCalls { + if toolCall.Function.Arguments != "" { + argumentChunks++ + streamedArguments.WriteString(toolCall.Function.Arguments) + } + } + } + } + require.Equal(t, 1, argumentChunks) + require.Equal(t, arguments, streamedArguments.String()) +} + +func TestResponsesStreamRoundTrip_PreservesCustomToolInputFromFinalEvents(t *testing.T) { + const input = "*** Begin Patch\n*** Update File: main.go\n@@\n-old\n+new\n*** End Patch" + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_patch_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_patch_1","type":"custom_tool_call","status":"in_progress","call_id":"call_patch_1","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.done", Data: []byte(`{"type":"response.custom_tool_call_input.done","item_id":"ctc_patch_1","output_index":0,"input":` + strconv.Quote(input) + `}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_patch_1","type":"custom_tool_call","status":"completed","call_id":"call_patch_1","name":"apply_patch","input":` + strconv.Quote(input) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_patch_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_patch_1","type":"custom_tool_call","status":"completed","call_id":"call_patch_1","name":"apply_patch","input":` + strconv.Quote(input) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, "custom_tool_call", completed.Output[0].Type) + require.NotNil(t, completed.Output[0].Input) + require.Equal(t, input, *completed.Output[0].Input) +} + +func TestResponsesStreamRoundTrip_PreservesFunctionCallMetadataFromFinalEvents(t *testing.T) { + const arguments = `{"description":"Run the delegated task."}` + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_metadata_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_metadata_1","type":"function_call","status":"in_progress","call_id":"call_metadata_1","name":"","namespace":"","arguments":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"fc_metadata_1","type":"function_call","status":"completed","call_id":"call_metadata_1","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_metadata_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_metadata_1","type":"function_call","status":"completed","call_id":"call_metadata_1","name":"spawn_agent","namespace":"collaboration","arguments":` + strconv.Quote(arguments) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, "spawn_agent", completed.Output[0].Name) + require.Equal(t, "collaboration", completed.Output[0].Namespace) + require.Equal(t, arguments, completed.Output[0].Arguments) +} + +func TestResponsesStreamRoundTrip_PreservesCustomToolMetadataFromFinalEvents(t *testing.T) { + const input = "patch" + + upstreamEvents := []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_metadata_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_metadata_1","type":"custom_tool_call","status":"in_progress","call_id":"call_metadata_2","name":"","namespace":"","input":""}}`)}, + {Type: "response.output_item.done", Data: []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_metadata_1","type":"custom_tool_call","status":"completed","call_id":"call_metadata_2","name":"apply_patch","namespace":"mcp__codex","input":` + strconv.Quote(input) + `}}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_metadata_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_metadata_1","type":"custom_tool_call","status":"completed","call_id":"call_metadata_2","name":"apply_patch","namespace":"mcp__codex","input":` + strconv.Quote(input) + `}]}}`)}, + } + + completed := roundTripCompletedResponse(t, upstreamEvents) + require.Len(t, completed.Output, 1) + require.Equal(t, "custom_tool_call", completed.Output[0].Type) + require.Equal(t, "apply_patch", completed.Output[0].Name) + require.Equal(t, "mcp__codex", completed.Output[0].Namespace) + require.NotNil(t, completed.Output[0].Input) + require.Equal(t, input, *completed.Output[0].Input) +} + +func TestResponsesStreamRoundTrip_PreservesToolItemIDsSeparatelyFromCallIDs(t *testing.T) { + tests := []struct { + name string + itemID string + callID string + streamData []*httpclient.StreamEvent + }{ + { + name: "function call", + itemID: "fc_identity_1", + callID: "call_identity_1", + streamData: []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_identity_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"fc_identity_1","type":"function_call","status":"in_progress","call_id":"call_identity_1","name":"wait","namespace":"collaboration","arguments":""}}`)}, + {Type: "response.function_call_arguments.done", Data: []byte(`{"type":"response.function_call_arguments.done","item_id":"fc_identity_1","output_index":0,"arguments":"{}"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_identity_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_identity_1","type":"function_call","status":"completed","call_id":"call_identity_1","name":"wait","namespace":"collaboration","arguments":"{}"}]}}`)}, + }, + }, + { + name: "custom tool call", + itemID: "ctc_identity_1", + callID: "call_identity_2", + streamData: []*httpclient.StreamEvent{ + {Type: "response.created", Data: []byte(`{"type":"response.created","response":{"id":"resp_identity_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"in_progress","output":[]}}`)}, + {Type: "response.output_item.added", Data: []byte(`{"type":"response.output_item.added","output_index":0,"item":{"id":"ctc_identity_1","type":"custom_tool_call","status":"in_progress","call_id":"call_identity_2","name":"apply_patch","input":""}}`)}, + {Type: "response.custom_tool_call_input.done", Data: []byte(`{"type":"response.custom_tool_call_input.done","item_id":"ctc_identity_1","output_index":0,"input":"patch"}`)}, + {Type: "response.completed", Data: []byte(`{"type":"response.completed","response":{"id":"resp_identity_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_identity_1","type":"custom_tool_call","status":"completed","call_id":"call_identity_2","name":"apply_patch","input":"patch"}]}}`)}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + completed := roundTripCompletedResponse(t, tt.streamData) + require.Len(t, completed.Output, 1) + require.Equal(t, tt.itemID, completed.Output[0].ID) + require.Equal(t, tt.callID, completed.Output[0].CallID) + }) + } +} + +func roundTripCompletedResponse(t *testing.T, upstreamEvents []*httpclient.StreamEvent) *Response { + t.Helper() + + var completed *Response + for _, event := range roundTripResponseEvents(t, upstreamEvents) { + if event.Type == StreamEventTypeResponseCompleted { + completed = event.Response + } + } + require.NotNil(t, completed) + return completed +} + +func roundTripResponseEvents(t *testing.T, upstreamEvents []*httpclient.StreamEvent) []StreamEvent { + t.Helper() + + upstream, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := upstream.TransformStream(context.Background(), nil, streams.SliceStream(upstreamEvents)) + require.NoError(t, err) + + client, err := NewInboundTransformer().TransformStream(context.Background(), canonical) + require.NoError(t, err) + + var events []StreamEvent + for client.Next() { + var event StreamEvent + require.NoError(t, json.Unmarshal(client.Current().Data, &event)) + events = append(events, event) + } + require.NoError(t, client.Err()) + return events +} + +func TestResponsesInboundStream_GeneratesProtocolValidToolItemIDs(t *testing.T) { + tests := []struct { + name string + toolCall llm.ToolCall + wantPrefix string + }{ + { + name: "function call", + toolCall: llm.ToolCall{ + ID: "call_generated_1", + Type: "function", + Index: 0, + Function: llm.FunctionCall{ + Name: "wait", + Arguments: `{"timeout_ms":1000}`, + }, + }, + wantPrefix: "fc_", + }, + { + name: "custom tool call", + toolCall: llm.ToolCall{ + ID: "call_generated_2", + Type: llm.ToolTypeResponsesCustomTool, + Index: 0, + ResponseCustomToolCall: &llm.ResponseCustomToolCall{ + CallID: "call_generated_2", + Name: "apply_patch", + Input: "patch", + }, + }, + wantPrefix: "ctc_", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + stream, err := NewInboundTransformer().TransformStream(t.Context(), streams.SliceStream([]*llm.Response{ + { + Object: "chat.completion.chunk", + ID: "resp_generated_ids", + Model: "test-model", + Created: 1700000000, + Choices: []llm.Choice{{ + Index: 0, + Delta: &llm.Message{ + Role: "assistant", + ToolCalls: []llm.ToolCall{tt.toolCall}, + }, + }}, + }, + { + Object: "chat.completion.chunk", + ID: "resp_generated_ids", + Model: "test-model", + Created: 1700000000, + Choices: []llm.Choice{{ + Index: 0, + Delta: &llm.Message{}, + FinishReason: lo.ToPtr("tool_calls"), + }}, + }, + })) + require.NoError(t, err) + + var itemID string + for stream.Next() { + var event StreamEvent + require.NoError(t, json.Unmarshal(stream.Current().Data, &event)) + if event.Type == StreamEventTypeOutputItemAdded && event.Item != nil { + itemID = event.Item.ID + } + } + require.NoError(t, stream.Err()) + require.True(t, strings.HasPrefix(itemID, tt.wantPrefix), "generated item id %q must use %q", itemID, tt.wantPrefix) + require.NotEqual(t, tt.toolCall.ID, itemID) + }) + } +} + +func TestResponsesInboundStream_DoesNotDuplicateInitialCustomToolInput(t *testing.T) { + const input = "patch" + + stream, err := NewInboundTransformer().TransformStream(t.Context(), streams.SliceStream([]*llm.Response{ + { + Object: "chat.completion.chunk", + ID: "resp_custom_input", + Model: "test-model", + Created: 1700000000, + Choices: []llm.Choice{{ + Index: 0, + Delta: &llm.Message{ + Role: "assistant", + ToolCalls: []llm.ToolCall{{ + ID: "call_custom_input", + ResponseItemID: "ctc_custom_input", + Type: llm.ToolTypeResponsesCustomTool, + Index: 0, + ResponseCustomToolCall: &llm.ResponseCustomToolCall{ + CallID: "call_custom_input", + Name: "apply_patch", + Input: input, + }, + }}, + }, + }}, + }, + { + Object: "chat.completion.chunk", + ID: "resp_custom_input", + Model: "test-model", + Created: 1700000000, + Choices: []llm.Choice{{ + Index: 0, + Delta: &llm.Message{}, + FinishReason: lo.ToPtr("tool_calls"), + }}, + }, + })) + require.NoError(t, err) + + var doneItem *Item + for stream.Next() { + var event StreamEvent + require.NoError(t, json.Unmarshal(stream.Current().Data, &event)) + if event.Type == StreamEventTypeOutputItemDone && event.Item != nil && event.Item.Type == "custom_tool_call" { + doneItem = event.Item + } + } + require.NoError(t, stream.Err()) + require.NotNil(t, doneItem) + require.NotNil(t, doneItem.Input) + require.Equal(t, input, *doneItem.Input) +} + +func TestResponsesNonStreamRoundTrip_PreservesToolItemIDsSeparatelyFromCallIDs(t *testing.T) { + tests := []struct { + name string + body string + itemID string + callID string + }{ + { + name: "function call", + body: `{"id":"resp_nonstream_1","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"fc_nonstream_1","type":"function_call","status":"completed","call_id":"call_nonstream_1","name":"wait","namespace":"collaboration","arguments":"{}"}]}`, + itemID: "fc_nonstream_1", + callID: "call_nonstream_1", + }, + { + name: "custom tool call", + body: `{"id":"resp_nonstream_2","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_nonstream_1","type":"custom_tool_call","status":"completed","call_id":"call_nonstream_2","name":"apply_patch","input":"patch"}]}`, + itemID: "ctc_nonstream_1", + callID: "call_nonstream_2", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformResponse(t.Context(), &httpclient.Response{ + StatusCode: 200, + Body: []byte(tt.body), + }) + require.NoError(t, err) + require.Len(t, canonical.Choices, 1) + require.NotNil(t, canonical.Choices[0].Message) + require.Len(t, canonical.Choices[0].Message.ToolCalls, 1) + require.Equal(t, tt.itemID, canonical.Choices[0].Message.ToolCalls[0].ResponseItemID) + + clientResponse, err := NewInboundTransformer().TransformResponse(t.Context(), canonical) + require.NoError(t, err) + var roundTripped Response + require.NoError(t, json.Unmarshal(clientResponse.Body, &roundTripped)) + require.Len(t, roundTripped.Output, 1) + require.Equal(t, tt.itemID, roundTripped.Output[0].ID) + require.Equal(t, tt.callID, roundTripped.Output[0].CallID) + }) + } +} + +func TestResponsesNonStreamRoundTrip_PreservesCustomToolNamespace(t *testing.T) { + body := `{"id":"resp_nonstream_namespace","object":"response","created_at":1700000000,"model":"gpt-5.5","status":"completed","output":[{"id":"ctc_nonstream_namespace","type":"custom_tool_call","status":"completed","call_id":"call_nonstream_namespace","name":"apply_patch","namespace":"mcp__codex","input":"patch"}]}` + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + canonical, err := outbound.TransformResponse(t.Context(), &httpclient.Response{ + StatusCode: 200, + Body: []byte(body), + }) + require.NoError(t, err) + require.Len(t, canonical.Choices, 1) + require.NotNil(t, canonical.Choices[0].Message) + require.Len(t, canonical.Choices[0].Message.ToolCalls, 1) + require.NotNil(t, canonical.Choices[0].Message.ToolCalls[0].ResponseCustomToolCall) + require.Equal(t, "mcp__codex", canonical.Choices[0].Message.ToolCalls[0].ResponseCustomToolCall.Namespace) + + inbound, err := NewInboundTransformer().TransformResponse(t.Context(), canonical) + require.NoError(t, err) + var roundTripped Response + require.NoError(t, json.Unmarshal(inbound.Body, &roundTripped)) + require.Len(t, roundTripped.Output, 1) + require.Equal(t, "mcp__codex", roundTripped.Output[0].Namespace) +} + +func TestResponsesRequestRoundTrip_PreservesToolItemIDsSeparatelyFromCallIDs(t *testing.T) { + requestBody := []byte(`{"model":"gpt-5.5","input":[{"id":"fc_request_1","type":"function_call","call_id":"call_request_1","name":"wait","namespace":"collaboration","arguments":"{}"},{"id":"fco_request_1","type":"function_call_output","call_id":"call_request_1","output":"done"},{"id":"ctc_request_1","type":"custom_tool_call","call_id":"call_request_2","name":"apply_patch","namespace":"mcp__codex","input":"patch"},{"id":"ctco_request_1","type":"custom_tool_call_output","call_id":"call_request_2","output":"done"}]}`) + + inbound, err := NewInboundTransformer().TransformRequest(t.Context(), &httpclient.Request{Body: requestBody}) + require.NoError(t, err) + require.Len(t, inbound.Messages, 4) + require.Equal(t, "fc_request_1", inbound.Messages[0].ToolCalls[0].ResponseItemID) + require.Equal(t, "fco_request_1", inbound.Messages[1].ID) + require.Equal(t, "ctc_request_1", inbound.Messages[2].ToolCalls[0].ResponseItemID) + require.Equal(t, "ctco_request_1", inbound.Messages[3].ID) + + outbound, err := NewOutboundTransformer("https://api.openai.com", "test-api-key") + require.NoError(t, err) + replayed, err := outbound.TransformRequest(t.Context(), inbound) + require.NoError(t, err) + + var replayedRequest Request + require.NoError(t, json.Unmarshal(replayed.Body, &replayedRequest)) + require.Len(t, replayedRequest.Input.Items, 4) + require.Equal(t, "fc_request_1", replayedRequest.Input.Items[0].ID) + require.Equal(t, "fco_request_1", replayedRequest.Input.Items[1].ID) + require.Equal(t, "ctc_request_1", replayedRequest.Input.Items[2].ID) + require.Equal(t, "mcp__codex", replayedRequest.Input.Items[2].Namespace) + require.Equal(t, "ctco_request_1", replayedRequest.Input.Items[3].ID) +}