server/agui: unique user messages based on requestID in AGUI MessagesSnapshot (#616)

In the agenttool scenario, user messages may appear multiple times in a
session to provide context to the agenttool, causing messagesnapshot to
send multiple user messages to the frontend. Therefore, it is necessary
to deduplicate user messages based on requestID.

Author: Flash-LHR

server/agui/runner/messagessnapshot.go

@@ -118,20 +118,26 @@ func (r *runner) convertToMessagesSnapshotEvent(ctx context.Context, userID stri
 	if len(events) == 0 {
 		return aguievents.NewMessagesSnapshotEvent(messages), nil
 	}
+	lastRequestID := ""
 	for _, event := range events {
 		event, err := r.handleBeforeTranslate(ctx, &event)
 		if err != nil {
 			return nil, fmt.Errorf("handle before translate: %w", err)
 		}
-		if event == nil || event.Response == nil || len(event.Response.Choices) == 0 {
+		if r.ignoreEvent(event) {
 			continue
 		}
 		for _, choice := range event.Response.Choices {
 			switch choice.Message.Role {
 			case model.RoleSystem:
 				messages = append(messages, *r.convertToSystemMessage(event.ID, choice))
 			case model.RoleUser:
-				messages = append(messages, *r.convertToUserMessage(event.ID, userID, choice))
+				if lastRequestID != event.RequestID {
+					// User message may be repeated multiple times in multiagent scenario.
+					// Only the first message should be included in the snapshot.
+					lastRequestID = event.RequestID
+					messages = append(messages, *r.convertToUserMessage(event.ID, userID, choice))
+				}
 			case model.RoleAssistant:
 				messages = append(messages, *r.convertToAssistantMessage(event.ID, choice))
 			case model.RoleTool:
@@ -144,6 +150,25 @@ func (r *runner) convertToMessagesSnapshotEvent(ctx context.Context, userID stri
 	return aguievents.NewMessagesSnapshotEvent(messages), nil
 }
 
+func (r *runner) ignoreEvent(event *event.Event) bool {
+	if event == nil || event.Response == nil || len(event.Response.Choices) == 0 {
+		return true
+	}
+	switch event.Response.Object {
+	// Model response event.
+	case model.ObjectTypeChatCompletion:
+		return false
+	// Tool response event.
+	case model.ObjectTypeToolResponse:
+		return false
+	// User message event.
+	case "":
+		return false
+	default:
+		return true
+	}
+}
+
 // convertToSystemMessage converts system events to AG-UI Message.
 func (r *runner) convertToSystemMessage(id string, choice model.Choice) *aguievents.Message {
 	return &aguievents.Message{

server/agui/runner/messagessnapshot_test.go

@@ -15,10 +15,10 @@ import (
 	"testing"
 
 	aguievents "github.com/ag-ui-protocol/ag-ui/sdks/community/go/pkg/core/events"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"trpc.group/trpc-go/trpc-agent-go/agent"
-	agentevent "trpc.group/trpc-go/trpc-agent-go/event"
-	eventpkg "trpc.group/trpc-go/trpc-agent-go/event"
+	"trpc.group/trpc-go/trpc-agent-go/event"
 	"trpc.group/trpc-go/trpc-agent-go/model"
 	"trpc.group/trpc-go/trpc-agent-go/server/agui/adapter"
 	"trpc.group/trpc-go/trpc-agent-go/server/agui/translator"
@@ -58,7 +58,7 @@ func TestMessagesSnapshotRequiresSessionService(t *testing.T) {
 }
 
 func TestMessagesSnapshotHappyPath(t *testing.T) {
-	events := []eventpkg.Event{
+	events := []event.Event{
 		newResponse(model.RoleUser, "hello", nil),
 		newResponse(model.RoleSystem, "system", nil),
 		newResponse(model.RoleAssistant, "reply", func(m *model.Message) {
@@ -111,7 +111,7 @@ func TestMessagesSnapshotHappyPath(t *testing.T) {
 }
 
 func TestMessagesSnapshotUnknownRole(t *testing.T) {
-	events := []eventpkg.Event{
+	events := []event.Event{
 		newResponse(model.RoleUser, "hello", nil),
 		newResponse(model.Role("unknown"), "?", nil),
 	}
@@ -173,7 +173,7 @@ func TestMessagesSnapshotUserIDResolverError(t *testing.T) {
 		userIDResolver:    userIDResolver,
 		runAgentInputHook: NewOptions().RunAgentInputHook,
 		appName:           "demo",
-		sessionService:    &testSessionService{events: []eventpkg.Event{newResponse(model.RoleUser, "hello", nil)}},
+		sessionService:    &testSessionService{events: []event.Event{newResponse(model.RoleUser, "hello", nil)}},
 	}
 
 	stream, err := r.MessagesSnapshot(
@@ -252,8 +252,8 @@ func TestMessagesSnapshotRunAgentInputHookError(t *testing.T) {
 type noopBaseRunner struct{}
 
 func (noopBaseRunner) Run(ctx context.Context, userID string, sessionID string, message model.Message,
-	_ ...agent.RunOption) (<-chan *agentevent.Event, error) {
-	ch := make(chan *agentevent.Event)
+	_ ...agent.RunOption) (<-chan *event.Event, error) {
+	ch := make(chan *event.Event)
 	close(ch)
 	return ch, nil
 }
@@ -271,7 +271,7 @@ func TestGetSessionEventsNilSession(t *testing.T) {
 // TestConvertToMessagesSnapshotEventSkipsNilResponse ensures nil response events are ignored.
 func TestConvertToMessagesSnapshotEventSkipsNilResponse(t *testing.T) {
 	r := &runner{}
-	snapshot, err := r.convertToMessagesSnapshotEvent(context.Background(), "user-id", []eventpkg.Event{{}})
+	snapshot, err := r.convertToMessagesSnapshotEvent(context.Background(), "user-id", []event.Event{{}})
 	assert.NoError(t, err)
 	assert.NotNil(t, snapshot)
 	assert.Len(t, snapshot.Messages, 0)
@@ -280,7 +280,7 @@ func TestConvertToMessagesSnapshotEventSkipsNilResponse(t *testing.T) {
 func TestConvertToMessagesSnapshotEventIncludesUserIDName(t *testing.T) {
 	r := &runner{}
 	snapshot, err := r.convertToMessagesSnapshotEvent(context.Background(),
-		"user-id", []eventpkg.Event{newResponse(model.RoleUser, "hello", nil)})
+		"user-id", []event.Event{newResponse(model.RoleUser, "hello", nil)})
 	assert.NoError(t, err)
 	assert.NotNil(t, snapshot)
 	assert.Len(t, snapshot.Messages, 1)
@@ -291,7 +291,7 @@ func TestConvertToMessagesSnapshotEventIncludesUserIDName(t *testing.T) {
 // TestConvertToMessagesSnapshotEventBeforeCallbackMutates ensures before callbacks update messages.
 func TestConvertToMessagesSnapshotEventBeforeCallbackMutates(t *testing.T) {
 	callbacks := translator.NewCallbacks().
-		RegisterBeforeTranslate(func(ctx context.Context, evt *eventpkg.Event) (*eventpkg.Event, error) {
+		RegisterBeforeTranslate(func(ctx context.Context, evt *event.Event) (*event.Event, error) {
 			if evt.Response != nil && len(evt.Response.Choices) > 0 {
 				evt.Response.Choices[0].Message.Content = "patched"
 			}
@@ -301,29 +301,196 @@ func TestConvertToMessagesSnapshotEventBeforeCallbackMutates(t *testing.T) {
 		translateCallbacks: callbacks,
 	}
 	snapshot, err := r.convertToMessagesSnapshotEvent(context.Background(),
-		"user-id", []eventpkg.Event{newResponse(model.RoleUser, "hello", nil)})
+		"user-id", []event.Event{newResponse(model.RoleUser, "hello", nil)})
 	assert.NoError(t, err)
 	assert.NotNil(t, snapshot)
 	assert.Len(t, snapshot.Messages, 1)
 	assert.Equal(t, "patched", *snapshot.Messages[0].Content)
 }
 
+func TestConvertToMessagesSnapshotEventDeduplicatesUserMessages(t *testing.T) {
+	r := &runner{}
+	sharedRequestID := "req-shared"
+	events := []event.Event{
+		newResponseWithRequestID(model.RoleUser, "hello", sharedRequestID, nil),
+		newResponseWithRequestID(model.RoleUser, "hello again", sharedRequestID, nil),
+		newResponseWithRequestID(model.RoleUser, "next", "req-next", nil),
+	}
+	snapshot, err := r.convertToMessagesSnapshotEvent(context.Background(), "user-id", events)
+	assert.NoError(t, err)
+	assert.NotNil(t, snapshot)
+	assert.Len(t, snapshot.Messages, 2)
+	assert.Equal(t, "hello", *snapshot.Messages[0].Content)
+	assert.Equal(t, "next", *snapshot.Messages[1].Content)
+}
+
+func TestIgnoreEvent(t *testing.T) {
+	tests := []struct {
+		name string
+		evt  *event.Event
+		want bool
+	}{
+		{
+			name: "nil event",
+			evt:  nil,
+			want: true,
+		},
+		{
+			name: "nil response",
+			evt:  &event.Event{Response: nil},
+			want: true,
+		},
+		{
+			name: "nil choices",
+			evt:  &event.Event{Response: &model.Response{Choices: nil}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypeChatCompletion,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeChatCompletion,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleUser, Content: "hello"}}},
+			}},
+			want: false,
+		},
+		{
+			name: model.ObjectTypeToolResponse,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeToolResponse,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleTool, Content: "hello"}}},
+			}},
+			want: false,
+		},
+		{
+			name: "",
+			evt: &event.Event{Response: &model.Response{
+				Object:  "",
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleUser, Content: "hello"}}},
+			}},
+			want: false,
+		},
+		{
+			name: model.ObjectTypeError,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeError,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePreprocessingBasic,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePreprocessingBasic,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePreprocessingContent,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePreprocessingContent,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePreprocessingIdentity,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePreprocessingIdentity,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePreprocessingInstruction,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePreprocessingInstruction,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePreprocessingPlanning,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePreprocessingPlanning,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePostprocessingPlanning,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePostprocessingPlanning,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypePostprocessingCodeExecution,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypePostprocessingCodeExecution,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypeTransfer,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeTransfer,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypeRunnerCompletion,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeRunnerCompletion,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypeStateUpdate,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeStateUpdate,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+		{
+			name: model.ObjectTypeChatCompletionChunk,
+			evt: &event.Event{Response: &model.Response{
+				Object:  model.ObjectTypeChatCompletionChunk,
+				Choices: []model.Choice{{Message: model.Message{Role: model.RoleAssistant, Content: "hello"}}},
+			}},
+			want: true,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			r := &runner{}
+			got := r.ignoreEvent(test.evt)
+			assert.Equal(t, test.want, got)
+		})
+	}
+}
+
 // TestConvertToMessagesSnapshotEventBeforeCallbackError ensures errors bubble up.
 func TestConvertToMessagesSnapshotEventBeforeCallbackError(t *testing.T) {
 	callbacks := translator.NewCallbacks().
-		RegisterBeforeTranslate(func(ctx context.Context, evt *eventpkg.Event) (*eventpkg.Event, error) {
+		RegisterBeforeTranslate(func(ctx context.Context, evt *event.Event) (*event.Event, error) {
 			return nil, errors.New("fail")
 		})
 	r := &runner{
 		translateCallbacks: callbacks,
 	}
 	snapshot, err := r.convertToMessagesSnapshotEvent(context.Background(),
-		"user-id", []eventpkg.Event{newResponse(model.RoleUser, "hello", nil)})
+		"user-id", []event.Event{newResponse(model.RoleUser, "hello", nil)})
 	assert.Nil(t, snapshot)
 	assert.Error(t, err)
 }
 
-func newResponse(role model.Role, content string, mutate func(*model.Message)) eventpkg.Event {
+func newResponse(role model.Role, content string, mutate func(*model.Message)) event.Event {
 	msg := model.Message{Role: role, Content: content}
 	if mutate != nil {
 		mutate(&msg)
@@ -332,12 +499,20 @@ func newResponse(role model.Role, content string, mutate func(*model.Message)) e
 		ID:      "id-" + string(role) + content,
 		Choices: []model.Choice{{Message: msg}},
 	}
-	evt := agentevent.NewResponseEvent("invocation", string(role), resp)
+	evt := event.NewResponseEvent("invocation", string(role), resp)
+	evt.RequestID = uuid.NewString()
 	return *evt
 }
 
+func newResponseWithRequestID(role model.Role, content, requestID string,
+	mutate func(*model.Message)) event.Event {
+	evt := newResponse(role, content, mutate)
+	evt.RequestID = requestID
+	return evt
+}
+
 type testSessionService struct {
-	events    []eventpkg.Event
+	events    []event.Event
 	getErr    error
 	returnNil bool
 }
@@ -357,7 +532,7 @@ func (s *testSessionService) GetSession(ctx context.Context, key session.Key,
 	}
 	sess := &session.Session{AppName: key.AppName, UserID: key.UserID, ID: key.SessionID}
 	if len(s.events) > 0 {
-		sess.Events = append([]eventpkg.Event(nil), s.events...)
+		sess.Events = append([]event.Event(nil), s.events...)
 	}
 	return sess, nil
 }
@@ -395,7 +570,7 @@ func (s *testSessionService) DeleteUserState(ctx context.Context, key session.Us
 	return nil
 }
 
-func (s *testSessionService) AppendEvent(ctx context.Context, sess *session.Session, evt *eventpkg.Event,
+func (s *testSessionService) AppendEvent(ctx context.Context, sess *session.Session, evt *event.Event,
 	opts ...session.Option) error {
 	return nil
 }