Skip to content

fix(responses): preserve streamed tool call history #2038

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
asuan-dev wants to merge 3 commits into
base: unstable
Choose a base branch
from
Open

fix(responses): preserve streamed tool call history #2038

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
88876db
fix(responses): preserve streamed tool call history
Jul 19, 2026
458cb2a
fix(responses): accept authoritative terminal tool payloads
Jul 19, 2026
e1abec8
refactor(responses): centralize streamed tool state
Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions llm/tools.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"`

Expand Down Expand Up @@ -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"`
}
197 changes: 127 additions & 70 deletions llm/transformer/openai/responses/aggregator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,118 @@ func (a *streamAggregator) getItemForEvent(outputIndex int, itemID *string) *agg
return a.lastItemByOutputIndex(outputIndex)
}

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) *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 {
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)
}
Comment on lines +232 to +234

@greptile-apps greptile-apps Bot Jul 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 response.completed cannot rescue a function_call that output_item.done marked done with empty arguments

The early return fires for every caller, including applyResponseSnapshot (fromCompletedResponse=true). When a provider sends output_item.done with arguments:"" and there are no prior argument deltas, finalItem marks the item OutputItemDone=true but leaves item.Arguments.Len() == 0. A subsequent response.completed snapshot carrying the full arguments hits this early return before applying them, so toolItem() — which requires Arguments.Len() > 0 for function_call — silently drops the call.

The fromCompletedResponse parameter was introduced to distinguish the two callers, but the guard does not consult it. Guarding with && !fromCompletedResponse would allow response.completed to fill in missing payload while still letting output_item.done win whenever it carries a non-empty value (since if src.Arguments != "" { ... } only overwrites when the snapshot has real data).


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 && (*src.Input != "" || item.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...)
}
}
}

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
}

return item
}

func applyDoneText(dst *strings.Builder, doneText string) {
if doneText == "" {
return
Expand Down Expand Up @@ -458,72 +570,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)

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:
Expand Down Expand Up @@ -551,6 +603,10 @@ func (a *streamAggregator) applyResponseSnapshot(response *Response) {
return
}

for outputIndex := range response.Output {
a.finalItem(outputIndex, &response.Output[outputIndex])
}

if response.ID != "" {
a.responseID = response.ID
}
Expand Down Expand Up @@ -634,12 +690,13 @@ func (a *streamAggregator) buildResponse() *Response {

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,
ID: item.ID,
Type: item.Type,
Status: lo.ToPtr(item.Status),
CallID: item.CallID,
Name: item.Name,
Namespace: item.Namespace,
Input: item.Input,
})

case "reasoning":
Expand Down
53 changes: 53 additions & 0 deletions llm/transformer/openai/responses/aggregator_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -883,3 +883,56 @@ 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_EmptyFinalCustomInputDoesNotEraseStreamedInput(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.Equal(t, "patch", *resp.Output[0].Input)
}
Loading