From f9c8b30f8d681490013e09ffc6fd6629db4be18e Mon Sep 17 00:00:00 2001 From: Alix-007 <267018309+Alix-007@users.noreply.github.com> Date: Sun, 29 Mar 2026 16:53:30 +0800 Subject: [PATCH] feat(agent): annotate user turns with thread metadata --- pkg/agent/loop.go | 66 +++++++++++++++++++++++++++ pkg/agent/loop_test.go | 100 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 166 insertions(+) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index ef2951365..cfab995d5 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -76,7 +76,10 @@ type processOptions struct { Channel string // Target channel for tool execution ChatID string // Target chat ID for tool execution SenderID string // Current sender ID for dynamic context + SenderUsername string // Current sender username for thread annotations SenderDisplayName string // Current sender display name for dynamic context + MessageID string // Current inbound platform message ID + ReplyToMessageID string // Current inbound reply target message ID UserMessage string // User message content (may include prefix) ForcedSkills []string // Skills explicitly requested for this message SystemPromptOverride string // Override the default system prompt (Used by SubTurns) @@ -106,6 +109,7 @@ const ( metadataKeyTeamID = "team_id" metadataKeyParentPeerKind = "parent_peer_kind" metadataKeyParentPeerID = "parent_peer_id" + metadataKeyReplyToMessage = "reply_to_message_id" ) func NewAgentLoop( @@ -1316,7 +1320,10 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) Channel: msg.Channel, ChatID: msg.ChatID, SenderID: msg.SenderID, + SenderUsername: msg.Sender.Username, SenderDisplayName: msg.Sender.DisplayName, + MessageID: msg.MessageID, + ReplyToMessageID: inboundMetadata(msg, metadataKeyReplyToMessage), UserMessage: msg.Content, Media: msg.Media, DefaultResponse: defaultResponse, @@ -1469,6 +1476,15 @@ func (al *AgentLoop) runAgentLoop( agent *AgentInstance, opts processOptions, ) (string, error) { + opts.UserMessage = formatUserMessageWithThreadMetadata( + opts.UserMessage, + opts.SenderDisplayName, + opts.SenderUsername, + opts.SenderID, + opts.MessageID, + opts.ReplyToMessageID, + ) + // Record last channel for heartbeat notifications (skip internal channels and cli) if opts.Channel != "" && opts.ChatID != "" && !constants.IsInternalChannel(opts.Channel) { channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID) @@ -3539,6 +3555,56 @@ func inboundMetadata(msg bus.InboundMessage, key string) string { return msg.Metadata[key] } +func formatUserMessageWithThreadMetadata( + content string, + senderDisplayName string, + senderUsername string, + senderID string, + messageID string, + replyToMessageID string, +) string { + messageID = strings.TrimSpace(messageID) + replyToMessageID = strings.TrimSpace(replyToMessageID) + if messageID == "" && replyToMessageID == "" { + return content + } + + metaParts := make([]string, 0, 3) + from := formatThreadSender(senderDisplayName, senderUsername, senderID) + if from != "" { + metaParts = append(metaParts, fmt.Sprintf("from:%s", from)) + } + if messageID != "" { + metaParts = append(metaParts, fmt.Sprintf("msg:#%s", messageID)) + } + if replyToMessageID != "" { + metaParts = append(metaParts, fmt.Sprintf("reply_to:#%s", replyToMessageID)) + } + + annotation := fmt.Sprintf("[%s]", strings.Join(metaParts, "; ")) + if strings.TrimSpace(content) == "" { + return annotation + } + return annotation + "\n" + content +} + +func formatThreadSender(displayName string, username string, senderID string) string { + displayName = strings.TrimSpace(displayName) + username = strings.TrimPrefix(strings.TrimSpace(username), "@") + senderID = strings.TrimSpace(senderID) + + switch { + case displayName != "" && username != "": + return fmt.Sprintf("%s (@%s)", displayName, username) + case displayName != "": + return displayName + case username != "": + return fmt.Sprintf("@%s", username) + default: + return senderID + } +} + // extractParentPeer extracts the parent peer (reply-to) from inbound message metadata. func extractParentPeer(msg bus.InboundMessage) *routing.RoutePeer { parentKind := inboundMetadata(msg, metadataKeyParentPeerKind) diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 25d20c689..db0b90d38 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -168,6 +168,106 @@ func TestProcessMessage_IncludesCurrentSenderInDynamicContext(t *testing.T) { } } +func TestFormatUserMessageWithThreadMetadata(t *testing.T) { + t.Run("no message IDs keeps original content", func(t *testing.T) { + got := formatUserMessageWithThreadMetadata("hello", "Alice", "alice", "discord:1", "", "") + if got != "hello" { + t.Fatalf("formatUserMessageWithThreadMetadata() = %q, want %q", got, "hello") + } + }) + + t.Run("includes from, message and reply IDs", func(t *testing.T) { + got := formatUserMessageWithThreadMetadata("ping", "Alice", "@alice", "discord:1", "123", "120") + want := "[from:Alice (@alice); msg:#123; reply_to:#120]\nping" + if got != want { + t.Fatalf("formatUserMessageWithThreadMetadata() = %q, want %q", got, want) + } + }) + + t.Run("empty content returns annotation only", func(t *testing.T) { + got := formatUserMessageWithThreadMetadata("", "", "", "discord:1", "123", "") + want := "[from:discord:1; msg:#123]" + if got != want { + t.Fatalf("formatUserMessageWithThreadMetadata() = %q, want %q", got, want) + } + }) +} + +func TestProcessMessage_AnnotatesThreadMetadataInPromptAndHistory(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &recordingProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + + msg := bus.InboundMessage{ + Channel: "discord", + SenderID: "discord:123", + ChatID: "group-1", + Content: "hello", + MessageID: "123", + Sender: bus.SenderInfo{ + DisplayName: "Alice", + Username: "@alice", + }, + Peer: bus.Peer{ + Kind: "direct", + ID: "discord:123", + }, + Metadata: map[string]string{ + metadataKeyReplyToMessage: "120", + }, + } + + response, err := al.processMessage(context.Background(), msg) + if err != nil { + t.Fatalf("processMessage() error = %v", err) + } + if response != "Mock response" { + t.Fatalf("processMessage() response = %q, want %q", response, "Mock response") + } + if len(provider.lastMessages) == 0 { + t.Fatal("provider did not receive any messages") + } + + wantAnnotated := "[from:Alice (@alice); msg:#123; reply_to:#120]\nhello" + lastMessage := provider.lastMessages[len(provider.lastMessages)-1] + if lastMessage.Role != "user" || lastMessage.Content != wantAnnotated { + t.Fatalf("last provider message = %+v, want user annotation %q", lastMessage, wantAnnotated) + } + + route := al.registry.ResolveRoute(routing.RouteInput{ + Channel: msg.Channel, + Peer: extractPeer(msg), + }) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("No default agent found") + } + history := defaultAgent.Sessions.GetHistory(route.SessionKey) + if len(history) != 2 { + t.Fatalf("expected history len=2, got %d", len(history)) + } + if history[0].Role != "user" || history[0].Content != wantAnnotated { + t.Fatalf("history user message = %+v, want %q", history[0], wantAnnotated) + } +} + func TestProcessMessage_UseCommandLoadsRequestedSkill(t *testing.T) { tmpDir := t.TempDir() skillDir := filepath.Join(tmpDir, "skills", "shell")