From 8669b06ba382384e1ea906595226ea757b45acca Mon Sep 17 00:00:00 2001 From: Kirill <17173528+0xff23@users.noreply.github.com> Date: Fri, 25 Sep 2026 11:25:51 -0700 Subject: [PATCH] Avoid messages for response-only updates --- agent/agent.go | 2 +- agent/agent_test.go | 43 +++++++++ agent/response.go | 49 +++++++---- agent/response_test.go | 102 ++++++++++++++++++++++ provider/a2aprovider/a2a.go | 12 ++- provider/a2aprovider/a2a_test.go | 24 ++++- provider/aguiprovider/agui.go | 2 - provider/aguiprovider/agui_test.go | 11 +-- provider/openaiprovider/responses.go | 19 ++-- provider/openaiprovider/responses_test.go | 47 +++++++++- 10 files changed, 264 insertions(+), 47 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index e7d552d5..35a46ed6 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -476,7 +476,7 @@ func (a *Agent) setAuthor(update *ResponseUpdate) { if update.AgentID == "" { update.AgentID = a.id } - if update.AuthorName == "" { + if update.AuthorName == "" && responseUpdateHasMessageData(update) { update.AuthorName = a.name } } diff --git a/agent/agent_test.go b/agent/agent_test.go index f79ab332..c06211eb 100644 --- a/agent/agent_test.go +++ b/agent/agent_test.go @@ -226,6 +226,49 @@ func newGenericTestAgent(runFn func(context.Context, []*message.Message, ...agen }) } +func TestAgent_Collect_ResponseMetadataOnlyUpdateDoesNotCreateMessage(t *testing.T) { + run := func(context.Context, []*message.Message, ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] { + return func(yield func(*agent.ResponseUpdate, error) bool) { + yield(&agent.ResponseUpdate{ + ContinuationToken: "next", + }, nil) + } + } + a := newGenericTestAgent(run, nil) + + resp, err := a.RunText(t.Context(), "hello").Collect() + if err != nil { + t.Fatalf("Collect() error = %v", err) + } + if len(resp.Messages) != 0 { + t.Fatalf("len(Messages) = %d, want 0", len(resp.Messages)) + } + if resp.ContinuationToken == "" { + t.Fatal("ContinuationToken is empty") + } +} + +func TestAgent_Collect_RawOnlyUpdateDoesNotCreateMessage(t *testing.T) { + raw := struct{ Kind string }{Kind: "task-status"} + run := func(context.Context, []*message.Message, ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] { + return func(yield func(*agent.ResponseUpdate, error) bool) { + yield(&agent.ResponseUpdate{RawRepresentation: raw}, nil) + } + } + a := newGenericTestAgent(run, nil) + + resp, err := a.RunText(t.Context(), "hello").Collect() + if err != nil { + t.Fatalf("Collect() error = %v", err) + } + if len(resp.Messages) != 0 { + t.Fatalf("len(Messages) = %d, want 0", len(resp.Messages)) + } + if resp.RawRepresentation != raw { + t.Fatalf("RawRepresentation = %#v, want %#v", resp.RawRepresentation, raw) + } +} + func TestNew_IgnoresNilMiddleware(t *testing.T) { var providerCalls, agentMiddlewareCalls, providerMiddlewareCalls int run := func(context.Context, []*message.Message, ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] { diff --git a/agent/response.go b/agent/response.go index 52511fc9..df5618fa 100644 --- a/agent/response.go +++ b/agent/response.go @@ -146,6 +146,12 @@ func (resp *Response) ToUpdates() []*ResponseUpdate { if createdAt.IsZero() { createdAt = resp.CreatedAt } + // A role distinguishes a metadata-only message update from the trailing + // response-level metadata update emitted below. + role := msg.Role + if role == "" { + role = message.RoleAssistant + } updates = append(updates, &ResponseUpdate{ RawRepresentation: msg.RawRepresentation, AdditionalProperties: msg.AdditionalProperties, @@ -154,7 +160,7 @@ func (resp *Response) ToUpdates() []*ResponseUpdate { ResponseID: resp.ID, FinishReason: resp.FinishReason, AuthorName: msg.AuthorName, - Role: msg.Role, + Role: role, CreatedAt: createdAt, Contents: msg.Contents, }) @@ -179,25 +185,28 @@ func (resp *Response) Update(update *ResponseUpdate) { if update == nil { return } - msg := resp.targetMessage(update) - // Some members on ResponseUpdate map to members of Message. - // Incorporate those into the latest message; in cases where the message - // stores a single value, prefer the latest update's value over anything - // stored in the message. - msg.AuthorName = cmp.Or(update.AuthorName, msg.AuthorName) - msg.Role = cmp.Or(update.Role, msg.Role) - msg.ID = cmp.Or(update.MessageID, msg.ID) - if !isValidCreatedAt(msg.CreatedAt) && isValidCreatedAt(update.CreatedAt) { - msg.CreatedAt = update.CreatedAt - } - msg.Contents = append(msg.Contents, update.Contents...) - if update.AdditionalProperties != nil { - if msg.AdditionalProperties == nil { - msg.AdditionalProperties = make(map[string]any) + // A response-level metadata update must not create an empty message. + if responseUpdateHasMessageData(update) { + msg := resp.targetMessage(update) + // Some members on ResponseUpdate map to members of Message. + // Incorporate those into the latest message; in cases where the message + // stores a single value, prefer the latest update's value over anything + // stored in the message. + msg.AuthorName = cmp.Or(update.AuthorName, msg.AuthorName) + msg.Role = cmp.Or(update.Role, msg.Role) + msg.ID = cmp.Or(update.MessageID, msg.ID) + if !isValidCreatedAt(msg.CreatedAt) && isValidCreatedAt(update.CreatedAt) { + msg.CreatedAt = update.CreatedAt } - maps.Copy(msg.AdditionalProperties, update.AdditionalProperties) + msg.Contents = append(msg.Contents, update.Contents...) + if update.AdditionalProperties != nil { + if msg.AdditionalProperties == nil { + msg.AdditionalProperties = make(map[string]any) + } + maps.Copy(msg.AdditionalProperties, update.AdditionalProperties) + } + msg.RawRepresentation = appendRawRepresentation(msg.RawRepresentation, update.RawRepresentation) } - msg.RawRepresentation = appendRawRepresentation(msg.RawRepresentation, update.RawRepresentation) // Other members on a ResponseUpdate map to members of the response. // Update the response object with those, preferring the values from later updates. @@ -319,6 +328,10 @@ type ResponseUpdate struct { Contents message.Contents `json:",omitzero"` } +func responseUpdateHasMessageData(update *ResponseUpdate) bool { + return update != nil && (update.MessageID != "" || update.AuthorName != "" || update.Role != "" || len(update.Contents) > 0) +} + // String returns the concatenated text contents of this update. func (r *ResponseUpdate) String() string { if r == nil { diff --git a/agent/response_test.go b/agent/response_test.go index 7d45feee..6737af87 100644 --- a/agent/response_test.go +++ b/agent/response_test.go @@ -562,6 +562,19 @@ func TestResponse_Update_RawRepresentation(t *testing.T) { } } +func TestResponse_Update_RawRepresentationOnlyDoesNotCreateMessage(t *testing.T) { + resp := &agent.Response{} + + resp.Update(&agent.ResponseUpdate{RawRepresentation: "raw-lifecycle-event"}) + + if len(resp.Messages) != 0 { + t.Fatalf("expected no messages for raw-only response update, got %d", len(resp.Messages)) + } + if resp.RawRepresentation != "raw-lifecycle-event" { + t.Fatalf("expected raw representation on response, got %#v", resp.RawRepresentation) + } +} + func TestResponse_ToUpdates_RoundTripPreservesRawRepresentationWithContinuationToken(t *testing.T) { // A response with a message raw representation and a continuation token emits // a trailing metadata-only update (RawRepresentation nil). Collecting the @@ -963,6 +976,95 @@ func TestResponse_ToUpdates_WithAdditionalPropertiesOnlyProducesSingleUpdate(t * } } +func TestResponse_ToUpdates_RoundTripDoesNotCreateMessageForResponseMetadata(t *testing.T) { + tests := []struct { + name string + response *agent.Response + wantToken string + wantProperty any + }{ + { + name: "continuation token", + response: &agent.Response{ContinuationToken: "token-123"}, + wantToken: "token-123", + }, + { + name: "additional properties", + response: &agent.Response{AdditionalProperties: map[string]any{"key": "value"}}, + wantProperty: "value", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var collected agent.Response + for _, update := range tt.response.ToUpdates() { + collected.Update(update) + } + + if len(collected.Messages) != 0 { + t.Fatalf("expected no messages after round trip, got %d", len(collected.Messages)) + } + if collected.ContinuationToken != tt.wantToken { + t.Errorf("continuation token = %q, want %q", collected.ContinuationToken, tt.wantToken) + } + if got := collected.AdditionalProperties["key"]; got != tt.wantProperty { + t.Errorf("additional property = %v, want %v", got, tt.wantProperty) + } + }) + } +} + +func TestResponse_ToUpdates_RoundTripPreservesMetadataOnlyMessages(t *testing.T) { + createdAt := time.Date(2026, time.September, 20, 12, 0, 0, 0, time.UTC) + tests := []struct { + name string + message *message.Message + }{ + { + name: "additional properties", + message: &message.Message{ + AdditionalProperties: map[string]any{"key": "value"}, + }, + }, + { + name: "created at", + message: &message.Message{ + CreatedAt: createdAt, + }, + }, + { + name: "raw representation", + message: &message.Message{ + RawRepresentation: "raw-message", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + original := &agent.Response{Messages: []*message.Message{tt.message}} + var collected agent.Response + for _, update := range original.ToUpdates() { + collected.Update(update) + } + + if len(collected.Messages) != 1 { + t.Fatalf("message count = %d, want 1", len(collected.Messages)) + } + if got := collected.Messages[0].AdditionalProperties["key"]; got != tt.message.AdditionalProperties["key"] { + t.Errorf("additional property = %v, want %v", got, tt.message.AdditionalProperties["key"]) + } + if got := collected.Messages[0].CreatedAt; !got.Equal(tt.message.CreatedAt) { + t.Errorf("created at = %v, want %v", got, tt.message.CreatedAt) + } + if got := collected.Messages[0].RawRepresentation; got != tt.message.RawRepresentation { + t.Errorf("raw representation = %v, want %v", got, tt.message.RawRepresentation) + } + }) + } +} + func TestResponse_String(t *testing.T) { msg := func(texts ...string) *message.Message { var contents message.Contents diff --git a/provider/a2aprovider/a2a.go b/provider/a2aprovider/a2a.go index 679209c6..8708795f 100644 --- a/provider/a2aprovider/a2a.go +++ b/provider/a2aprovider/a2a.go @@ -270,7 +270,11 @@ func sendMsg(session *agent.Session, seq iter.Seq2[a2a.Event, error], stream boo } } } - update := newResponseUpdate(e, e.Metadata, string(e.TaskID), messageID, message.RoleAssistant, contents) + role := message.Role("") + if messageID != "" || len(contents) > 0 { + role = message.RoleAssistant + } + update := newResponseUpdate(e, e.Metadata, string(e.TaskID), messageID, role, contents) update.FinishReason = finishReasonForTaskState(e.Status.State) if !yield(update, nil) { return @@ -407,7 +411,11 @@ func yieldTask(yield func(*agent.ResponseUpdate, error) bool, task *a2a.Task, sp return false } } - update := newResponseUpdate(task, cloneMetadata(task.Metadata), string(task.ID), messageID, message.RoleAssistant, contents) + role := message.Role("") + if messageID != "" || len(contents) > 0 { + role = message.RoleAssistant + } + update := newResponseUpdate(task, cloneMetadata(task.Metadata), string(task.ID), messageID, role, contents) update.ContinuationToken = continuationToken update.FinishReason = finishReason return yield(update, nil) diff --git a/provider/a2aprovider/a2a_test.go b/provider/a2aprovider/a2a_test.go index 92a08a81..09e74646 100644 --- a/provider/a2aprovider/a2a_test.go +++ b/provider/a2aprovider/a2a_test.go @@ -1843,8 +1843,8 @@ func TestRunStreamingWithTaskStatusUpdateEvent(t *testing.T) { } update := updates[0] - if update.Role != message.RoleAssistant { - t.Errorf("update.Role = %q, want %q", update.Role, message.RoleAssistant) + if update.Role != "" { + t.Errorf("update.Role = %q, want empty for lifecycle-only status", update.Role) } if update.ResponseID != taskID { t.Errorf("update.ResponseID = %q, want %q", update.ResponseID, taskID) @@ -2153,3 +2153,23 @@ func TestRunStreamingWithTaskArtifactUpdateEvent(t *testing.T) { t.Errorf("session.TaskID = %q, want %q", got, taskID) } } + +func TestAgentRunStreamingLifecycleOnlyTaskDoesNotCreateMessage(t *testing.T) { + transport := &mockA2ATransport{streamingResponseToReturn: &a2a.Task{ + ID: "task-raw", + ContextID: "ctx-raw", + Status: a2a.TaskStatus{State: a2a.TaskStateSubmitted}, + }} + a := newTestAgent(transport, agent.Config{}) + + resp, err := a.RunText(t.Context(), "start", agent.Stream(true)).Collect() + if err != nil { + t.Fatal(err) + } + if len(resp.Messages) != 0 { + t.Fatalf("len(Messages) = %d, want 0", len(resp.Messages)) + } + if resp.RawRepresentation == nil { + t.Fatal("RawRepresentation is nil") + } +} diff --git a/provider/aguiprovider/agui.go b/provider/aguiprovider/agui.go index 43530117..992069b0 100644 --- a/provider/aguiprovider/agui.go +++ b/provider/aguiprovider/agui.go @@ -397,7 +397,6 @@ func (a *toolCallAccumulator) onEvent(evt aguiEvents.Event) ([]*agent.ResponseUp switch e := evt.(type) { case *aguiEvents.RunStartedEvent: return []*agent.ResponseUpdate{{ - Role: message.RoleAssistant, ResponseID: e.RunID(), CreatedAt: eventTime(evt), AdditionalProperties: map[string]any{ @@ -418,7 +417,6 @@ func (a *toolCallAccumulator) onEvent(evt aguiEvents.Event) ([]*agent.ResponseUp props["result"] = e.Result } return []*agent.ResponseUpdate{{ - Role: message.RoleAssistant, ResponseID: e.RunID(), CreatedAt: eventTime(evt), FinishReason: "stop", diff --git a/provider/aguiprovider/agui_test.go b/provider/aguiprovider/agui_test.go index 8577df24..5eee3af0 100644 --- a/provider/aguiprovider/agui_test.go +++ b/provider/aguiprovider/agui_test.go @@ -220,7 +220,7 @@ func TestAGUIAgentRun_ConfigInstructionsBecomeSystemMessage(t *testing.T) { } } -func TestAGUIAgentRun_WithEmptyEventStream_EmitsMetadataUpdate(t *testing.T) { +func TestAGUIAgentRun_WithEmptyEventStream_DoesNotCreateMessage(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") writeSSE(t, w, aguiEvents.NewRunStartedEvent("thread-1", "run-1")) @@ -233,13 +233,8 @@ func TestAGUIAgentRun_WithEmptyEventStream_EmitsMetadataUpdate(t *testing.T) { if err != nil { t.Fatalf("run error: %v", err) } - if len(resp.Messages) == 0 { - t.Fatal("expected at least one metadata message") - } - for _, msg := range resp.Messages { - if msg.Role != message.RoleAssistant { - t.Fatalf("message role = %q, want %q", msg.Role, message.RoleAssistant) - } + if len(resp.Messages) != 0 { + t.Fatalf("messages length = %d, want 0", len(resp.Messages)) } } diff --git a/provider/openaiprovider/responses.go b/provider/openaiprovider/responses.go index 26e59ab4..b740649b 100644 --- a/provider/openaiprovider/responses.go +++ b/provider/openaiprovider/responses.go @@ -1075,7 +1075,6 @@ func responsesProcessResponse(resp *responses.Response, seqNum int64, yield func ResponseID: resp.ID, FinishReason: finishReason, CreatedAt: time.Unix(int64(resp.CreatedAt), 0), - Role: message.RoleAssistant, AdditionalProperties: responsesPopulateAdditionalProperties(resp), } // Only set ContinuationToken if it's not empty @@ -1390,7 +1389,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, var u *agent.ResponseUpdate switch event := update.AsAny().(type) { case responses.ResponseCreatedEvent: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) u.CreatedAt = time.Unix(int64(event.Response.CreatedAt), 0) u.ResponseID = event.Response.ID u.AdditionalProperties = responsesPopulateAdditionalProperties(&event.Response) @@ -1399,7 +1398,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, } case responses.ResponseQueuedEvent: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) u.CreatedAt = time.Unix(int64(event.Response.CreatedAt), 0) u.ResponseID = event.Response.ID u.AdditionalProperties = responsesPopulateAdditionalProperties(&event.Response) @@ -1408,7 +1407,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, } case responses.ResponseInProgressEvent: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) u.CreatedAt = time.Unix(int64(event.Response.CreatedAt), 0) u.ResponseID = event.Response.ID u.AdditionalProperties = responsesPopulateAdditionalProperties(&event.Response) @@ -1428,13 +1427,13 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, state.anyFunctions = true state.role = message.RoleAssistant } - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) if contToken := createContinuationToken(responseID, event.SequenceNumber, responses.ResponseStatusInProgress, isBackground); contToken != "" { u.ContinuationToken = contToken } case responses.ResponseCompletedEvent: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) u.CreatedAt = time.Unix(int64(event.Response.CreatedAt), 0) u.ResponseID = event.Response.ID u.FinishReason = responsesFinishReason(&event.Response) @@ -1448,7 +1447,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, } case responses.ResponseIncompleteEvent: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) u.CreatedAt = time.Unix(int64(event.Response.CreatedAt), 0) u.ResponseID = event.Response.ID u.FinishReason = responsesFinishReason(&event.Response) @@ -1458,7 +1457,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, } case responses.ResponseFailedEvent: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) u.CreatedAt = time.Unix(int64(event.Response.CreatedAt), 0) u.ResponseID = event.Response.ID u.AdditionalProperties = responsesPopulateAdditionalProperties(&event.Response) @@ -1571,7 +1570,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, case responses.ResponseOutputItemDoneEvent: // Create update for all output item done events - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) if contToken := createContinuationToken(responseID, event.SequenceNumber, responses.ResponseStatusInProgress, isBackground); contToken != "" { u.ContinuationToken = contToken } @@ -1679,7 +1678,7 @@ func responsesProcessStreamingUpdate(update responses.ResponseStreamEventUnion, u.ContinuationToken = contToken } default: - u = createUpdate(message.RoleAssistant, nil) + u = createUpdate("", nil) if contToken := createContinuationToken(responseID, update.SequenceNumber, responses.ResponseStatusInProgress, isBackground); contToken != "" { u.ContinuationToken = contToken } diff --git a/provider/openaiprovider/responses_test.go b/provider/openaiprovider/responses_test.go index 62c04ea0..53e7711b 100644 --- a/provider/openaiprovider/responses_test.go +++ b/provider/openaiprovider/responses_test.go @@ -7014,8 +7014,8 @@ func TestResponsesBackgroundResponses_FirstCall(t *testing.T) { t.Fatalf("error = %v", err) } - if len(resp.Messages) != 1 { - t.Errorf("expected 1 message (for continuation token), got %d", len(resp.Messages)) + if len(resp.Messages) != 0 { + t.Errorf("expected no messages for continuation-only response, got %d", len(resp.Messages)) } if resp.ContinuationToken == "" { @@ -7101,8 +7101,8 @@ func testResponsesBackgroundPolling(t *testing.T, status string) { t.Error("expected ContinuationToken to be set for queued/in_progress status") } - if len(resp.Messages) != 1 { - t.Errorf("expected 1 message for %s status, got %d", status, len(resp.Messages)) + if len(resp.Messages) != 0 { + t.Errorf("expected no messages for %s status, got %d", status, len(resp.Messages)) } case "completed": @@ -7928,3 +7928,42 @@ func TestResponsesFunctionCallUsesToolCallsFinishReason_NonStreaming(t *testing. t.Fatalf("FinishReason = %q, want tool_calls", resp.FinishReason) } } + +func TestResponsesBackgroundResponses_StreamingLifecycleOnlyDoesNotCreateMessage(t *testing.T) { + const input = `{"model":"gpt-4o-2024-08-06","background":true,"input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"hello"}]}],"stream":true}` + const output = `event: response.created + +data: {"type":"response.created","sequence_number":0,"response":{"id":"resp_lifecycle","object":"response","created_at":1758724519,"status":"queued","background":true,"model":"gpt-4o-2024-08-06","output":[]}} + +event: response.queued + +data: {"type":"response.queued","sequence_number":1,"response":{"id":"resp_lifecycle","object":"response","created_at":1758724519,"status":"queued","background":true,"model":"gpt-4o-2024-08-06","output":[]}} + +event: response.in_progress + +data: {"type":"response.in_progress","sequence_number":2,"response":{"id":"resp_lifecycle","object":"response","created_at":1758724519,"status":"in_progress","background":true,"model":"gpt-4o-2024-08-06","output":[]}} + +` + server := newTestResponsesServerStreaming(t, input, output) + defer server.Close() + a := newTestResponsesClient(server, "gpt-4o-2024-08-06") + session, err := a.CreateSession(t.Context()) + if err != nil { + t.Fatal(err) + } + + resp, err := a.RunText( + t.Context(), "hello", agent.Stream(true), + agent.AllowBackgroundResponses(true), + agent.WithSession(session), + ).Collect() + if err != nil { + t.Fatal(err) + } + if len(resp.Messages) != 0 { + t.Fatalf("len(Messages) = %d, want 0", len(resp.Messages)) + } + if resp.RawRepresentation == nil { + t.Fatal("RawRepresentation is nil") + } +}