diff --git a/pkg/agent/context.go b/pkg/agent/context.go index d03e488b6..667d714a5 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -611,7 +611,7 @@ func (cb *ContextBuilder) BuildMessages( // so the LLM can navigate thread structure from persisted sessions. for _, msg := range history { annotated := msg - if prefix := messageThreadAnnotation(msg); prefix != "" { + if prefix := messageHistoryAnnotation(msg); prefix != "" { annotated.Content = prefix + msg.Content } messages = append(messages, annotated) @@ -865,9 +865,62 @@ func (cb *ContextBuilder) GetSkillsInfo() map[string]any { } } +func messageHistoryAnnotation(msg providers.Message) string { + parts := make([]string, 0, 2) + if sender := messageSenderAnnotation(msg.Sender); sender != "" { + parts = append(parts, sender) + } + if thread := messageThreadAnnotationBody(msg); thread != "" { + parts = append(parts, thread) + } + if len(parts) == 0 { + return "" + } + return fmt.Sprintf("[%s] ", strings.Join(parts, ", ")) +} + +func messageSenderAnnotation(sender *providers.MessageSender) string { + if sender == nil { + return "" + } + + nameParts := make([]string, 0, 2) + if first := strings.TrimSpace(sender.FirstName); first != "" { + nameParts = append(nameParts, first) + } + if last := strings.TrimSpace(sender.LastName); last != "" { + nameParts = append(nameParts, last) + } + name := strings.TrimSpace(strings.Join(nameParts, " ")) + + username := strings.TrimSpace(sender.Username) + if username != "" && !strings.HasPrefix(username, "@") { + username = "@" + username + } + + switch { + case name != "" && username != "": + return fmt.Sprintf("from:%s (%s)", name, username) + case name != "": + return fmt.Sprintf("from:%s", name) + case username != "": + return fmt.Sprintf("from:%s", username) + default: + return "" + } +} + // messageThreadAnnotation returns the thread annotation prefix for a message, // e.g. "[msg:#5, reply_to:#3] " or "" if the message has no threading IDs. func messageThreadAnnotation(msg providers.Message) string { + body := messageThreadAnnotationBody(msg) + if body == "" { + return "" + } + return fmt.Sprintf("[%s] ", body) +} + +func messageThreadAnnotationBody(msg providers.Message) string { msgIDs := msg.MessageIDs formattedIDs := strings.Join(msgIDs, ",#") if formattedIDs != "" { @@ -875,15 +928,15 @@ func messageThreadAnnotation(msg providers.Message) string { } switch { case len(msgIDs) > 1 && msg.ReplyToMessageID != "": - return fmt.Sprintf("[msgs:%s, reply_to:#%s] ", formattedIDs, msg.ReplyToMessageID) + return fmt.Sprintf("msgs:%s, reply_to:#%s", formattedIDs, msg.ReplyToMessageID) case len(msgIDs) > 1: - return fmt.Sprintf("[msgs:%s] ", formattedIDs) + return fmt.Sprintf("msgs:%s", formattedIDs) case len(msgIDs) == 1 && msg.ReplyToMessageID != "": - return fmt.Sprintf("[msg:%s, reply_to:#%s] ", formattedIDs, msg.ReplyToMessageID) + return fmt.Sprintf("msg:%s, reply_to:#%s", formattedIDs, msg.ReplyToMessageID) case len(msgIDs) == 1: - return fmt.Sprintf("[msg:%s] ", formattedIDs) + return fmt.Sprintf("msg:%s", formattedIDs) case msg.ReplyToMessageID != "": - return fmt.Sprintf("[reply_to:#%s] ", msg.ReplyToMessageID) + return fmt.Sprintf("reply_to:#%s", msg.ReplyToMessageID) default: return "" } diff --git a/pkg/agent/context_test.go b/pkg/agent/context_test.go index 0d7948eef..6a6ec24b2 100644 --- a/pkg/agent/context_test.go +++ b/pkg/agent/context_test.go @@ -213,6 +213,38 @@ func TestSanitizeHistoryForProvider_DuplicateToolResults(t *testing.T) { } } +func TestMessageHistoryAnnotation_IncludesSenderAndThreading(t *testing.T) { + msg := providers.Message{ + Role: "user", + Content: "hello", + MessageIDs: []string{"m1"}, + ReplyToMessageID: "p0", + Sender: &providers.MessageSender{ + Username: "alice", + FirstName: "Alice", + LastName: "Example", + }, + } + + if got := messageHistoryAnnotation(msg); got != "[from:Alice Example (@alice), msg:#m1, reply_to:#p0] " { + t.Fatalf("messageHistoryAnnotation() = %q", got) + } +} + +func TestMessageHistoryAnnotation_UsesUsernameWhenNameMissing(t *testing.T) { + msg := providers.Message{ + Role: "user", + Content: "hello", + Sender: &providers.MessageSender{ + Username: "alice", + }, + } + + if got := messageHistoryAnnotation(msg); got != "[from:@alice] " { + t.Fatalf("messageHistoryAnnotation() = %q", got) + } +} + func roles(msgs []providers.Message) []string { r := make([]string, len(msgs)) for i, m := range msgs { diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 4ba5368c6..77e13632e 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -494,8 +494,6 @@ func (al *AgentLoop) Run(ctx context.Context) error { ChatID: msg.ChatID, } } - finalResponse := response.Content - target, targetErr := al.buildContinuationTarget(msg) if targetErr != nil { logger.WarnCF("agent", "Failed to build steering continuation target", @@ -513,12 +511,8 @@ func (al *AgentLoop) Run(ctx context.Context) error { return } - responsePersisted := false - continuedOnce := false - if al.pendingSteeringCountForScope(target.SessionKey) > 0 && - response.Content != "" && response.OnDelivered != nil { - response.OnDelivered(nil) - responsePersisted = true + if response.Content != "" { + al.publishAgentResponseIfNeeded(ctx, response, target.Channel, target.ChatID) } for al.pendingSteeringCountForScope(target.SessionKey) > 0 { @@ -530,7 +524,7 @@ func (al *AgentLoop) Run(ctx context.Context) error { "queue_depth": al.pendingSteeringCountForScope(target.SessionKey), }) - continued, continueErr := al.Continue(ctx, target.SessionKey, target.Channel, target.ChatID) + continued, continueErr := al.continueResponse(ctx, target.SessionKey, target.Channel, target.ChatID) if continueErr != nil { logger.WarnCF("agent", "Failed to continue queued steering", map[string]any{ @@ -540,21 +534,14 @@ func (al *AgentLoop) Run(ctx context.Context) error { }) return } - if continued == "" { + if continued.Content == "" { return } - - finalResponse = continued - continuedOnce = true + al.publishAgentResponseIfNeeded(ctx, continued, target.Channel, target.ChatID) } cancelDrain() - if al.pendingSteeringCountForScope(target.SessionKey) > 0 && - !responsePersisted && response.Content != "" && response.OnDelivered != nil { - response.OnDelivered(nil) - } - for al.pendingSteeringCountForScope(target.SessionKey) > 0 { logger.InfoCF("agent", "Draining steering queued during turn shutdown", map[string]any{ @@ -564,7 +551,7 @@ func (al *AgentLoop) Run(ctx context.Context) error { "queue_depth": al.pendingSteeringCountForScope(target.SessionKey), }) - continued, continueErr := al.Continue(ctx, target.SessionKey, target.Channel, target.ChatID) + continued, continueErr := al.continueResponse(ctx, target.SessionKey, target.Channel, target.ChatID) if continueErr != nil { logger.WarnCF("agent", "Failed to continue queued steering after shutdown drain", map[string]any{ @@ -574,20 +561,10 @@ func (al *AgentLoop) Run(ctx context.Context) error { }) return } - if continued == "" { + if continued.Content == "" { break } - - finalResponse = continued - continuedOnce = true - } - - if finalResponse != "" { - if continuedOnce { - al.publishResponseIfNeeded(ctx, target.Channel, target.ChatID, finalResponse) - } else { - al.publishAgentResponseIfNeeded(ctx, response, target.Channel, target.ChatID) - } + al.publishAgentResponseIfNeeded(ctx, continued, target.Channel, target.ChatID) } }() default: @@ -655,10 +632,17 @@ func (al *AgentLoop) drainBusToSteering(ctx context.Context, activeScope, active "scope": activeScope, }) + replyToMessageID := msg.ReplyToMessageID + if replyToMessageID == "" { + replyToMessageID = inboundMetadata(msg, metadataKeyReplyToMessage) + } if err := al.enqueueSteeringMessage(activeScope, activeAgentID, providers.Message{ - Role: "user", - Content: msg.Content, - Media: append([]string(nil), msg.Media...), + Role: "user", + Content: msg.Content, + Media: append([]string(nil), msg.Media...), + MessageIDs: singleMessageIDs(msg.MessageID), + ReplyToMessageID: replyToMessageID, + Sender: messageSenderFromInbound(msg.Sender), }); err != nil { logger.WarnCF("agent", "Failed to steer message, will be lost", map[string]any{ @@ -1448,9 +1432,12 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) EnableSummary: true, SendResponse: false, MessageID: msg.MessageID, - ReplyToMessageID: inboundMetadata(msg, metadataKeyReplyToMessage), + ReplyToMessageID: msg.ReplyToMessageID, Sender: messageSenderFromInbound(msg.Sender), } + if opts.ReplyToMessageID == "" { + opts.ReplyToMessageID = inboundMetadata(msg, metadataKeyReplyToMessage) + } // context-dependent commands check their own Runtime fields and report // "unavailable" when the required capability is nil. diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 01ad4c3c1..b61527993 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -467,6 +467,59 @@ func TestProcessMessage_AssistantSavedOnDelivered(t *testing.T) { } } +func TestProcessMessage_SavesReplyToMessageIDFromInboundField(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, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &recordingProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + + sessionKey := "agent:test-reply-to" + _, err = al.processMessage(context.Background(), bus.InboundMessage{ + Channel: "telegram", + SenderID: "telegram:123", + ChatID: "chat-1", + Content: "hello", + SessionKey: sessionKey, + MessageID: "in-42", + ReplyToMessageID: "in-41", + }) + if err != nil { + t.Fatalf("processMessage() error = %v", err) + } + + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("No default agent found") + } + + history := defaultAgent.Sessions.GetHistory(sessionKey) + if len(history) != 1 { + t.Fatalf("expected only user message before delivery, got %d", len(history)) + } + if len(history[0].MessageIDs) != 1 || history[0].MessageIDs[0] != "in-42" { + t.Fatalf("expected user message_ids [in-42], got %v", history[0].MessageIDs) + } + if history[0].ReplyToMessageID != "in-41" { + t.Fatalf("expected ReplyToMessageID in-41, got %q", history[0].ReplyToMessageID) + } +} + func TestRecordLastChannel(t *testing.T) { al, cfg, msgBus, provider, cleanup := newTestAgentLoop(t) defer cleanup() diff --git a/pkg/agent/steering.go b/pkg/agent/steering.go index 3f95a6d74..d3b08f91a 100644 --- a/pkg/agent/steering.go +++ b/pkg/agent/steering.go @@ -291,7 +291,7 @@ func (al *AgentLoop) continueWithSteeringMessages( agent *AgentInstance, sessionKey, channel, chatID string, steeringMsgs []providers.Message, -) (string, error) { +) (agentResponse, error) { response, err := al.runAgentLoop(ctx, agent, processOptions{ SessionKey: sessionKey, Channel: channel, @@ -303,12 +303,42 @@ func (al *AgentLoop) continueWithSteeringMessages( SkipInitialSteeringPoll: true, }) if err != nil { - return "", err + return agentResponse{}, err } - if response.OnDelivered != nil { - response.OnDelivered(nil) + return response, nil +} + +func (al *AgentLoop) continueResponse( + ctx context.Context, + sessionKey, channel, chatID string, +) (agentResponse, error) { + if active := al.GetActiveTurn(); active != nil { + return agentResponse{}, fmt.Errorf("turn %s is still active", active.TurnID) } - return response.Content, nil + if err := al.ensureHooksInitialized(ctx); err != nil { + return agentResponse{}, err + } + if err := al.ensureMCPInitialized(ctx); err != nil { + return agentResponse{}, err + } + + steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(sessionKey) + if len(steeringMsgs) == 0 { + return agentResponse{}, nil + } + + agent := al.agentForSession(sessionKey) + if agent == nil { + return agentResponse{}, fmt.Errorf("no agent available for session %q", sessionKey) + } + + if tool, ok := agent.Tools.Get("message"); ok { + if resetter, ok := tool.(interface{ ResetSentInRound() }); ok { + resetter.ResetSentInRound() + } + } + + return al.continueWithSteeringMessages(ctx, agent, sessionKey, channel, chatID, steeringMsgs) } func (al *AgentLoop) agentForSession(sessionKey string) *AgentInstance { @@ -333,33 +363,14 @@ func (al *AgentLoop) agentForSession(sessionKey string) *AgentInstance { // // If no steering messages are pending, it returns an empty string. func (al *AgentLoop) Continue(ctx context.Context, sessionKey, channel, chatID string) (string, error) { - if active := al.GetActiveTurn(); active != nil { - return "", fmt.Errorf("turn %s is still active", active.TurnID) - } - if err := al.ensureHooksInitialized(ctx); err != nil { + response, err := al.continueResponse(ctx, sessionKey, channel, chatID) + if err != nil { return "", err } - if err := al.ensureMCPInitialized(ctx); err != nil { - return "", err + if response.OnDelivered != nil { + response.OnDelivered(nil) } - - steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(sessionKey) - if len(steeringMsgs) == 0 { - return "", nil - } - - agent := al.agentForSession(sessionKey) - if agent == nil { - return "", fmt.Errorf("no agent available for session %q", sessionKey) - } - - if tool, ok := agent.Tools.Get("message"); ok { - if resetter, ok := tool.(interface{ ResetSentInRound() }); ok { - resetter.ResetSentInRound() - } - } - - return al.continueWithSteeringMessages(ctx, agent, sessionKey, channel, chatID, steeringMsgs) + return response.Content, nil } func (al *AgentLoop) InterruptGraceful(hint string) error { diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index 75ba9861d..71fe51dd6 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -431,6 +431,129 @@ func TestDrainBusToSteering_RequeuesDifferentScopeMessage(t *testing.T) { } } +func TestDrainBusToSteering_PreservesSenderAndThreadingMetadata(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, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + Session: config.SessionConfig{ + DMScope: "per-peer", + }, + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, &mockProvider{}) + + activeMsg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "telegram:100", + Sender: bus.SenderInfo{DisplayName: "Alice", Username: "alice"}, + ChatID: "chat1", + Content: "follow up", + MessageID: "in-2", + ReplyToMessageID: "in-1", + Peer: bus.Peer{ + Kind: "direct", + ID: "100", + }, + } + activeScope, activeAgentID, ok := al.resolveSteeringTarget(activeMsg) + if !ok { + t.Fatal("expected active message to resolve to a steering scope") + } + + if err := msgBus.PublishInbound(context.Background(), activeMsg); err != nil { + t.Fatalf("PublishInbound failed: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + al.drainBusToSteering(ctx, activeScope, activeAgentID) + + msgs := al.dequeueSteeringMessagesForScope(activeScope) + if len(msgs) != 1 { + t.Fatalf("expected 1 steering message, got %d", len(msgs)) + } + if len(msgs[0].MessageIDs) != 1 || msgs[0].MessageIDs[0] != "in-2" { + t.Fatalf("expected steering message_ids [in-2], got %v", msgs[0].MessageIDs) + } + if msgs[0].ReplyToMessageID != "in-1" { + t.Fatalf("expected ReplyToMessageID in-1, got %q", msgs[0].ReplyToMessageID) + } + if msgs[0].Sender == nil || msgs[0].Sender.FirstName != "Alice" || msgs[0].Sender.Username != "alice" { + t.Fatalf("expected sender to be preserved, got %+v", msgs[0].Sender) + } +} + +func TestContinueWithSteeringMessages_ReturnsTrackedAssistantDelivery(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, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + al := NewAgentLoop(cfg, msgBus, &simpleMockProvider{response: "continued response"}) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil { + t.Fatal("expected default agent") + } + + sessionKey := "agent:test-continue-delivery" + response, err := al.continueWithSteeringMessages(context.Background(), defaultAgent, sessionKey, "test", "chat1", []providers.Message{ + {Role: "user", Content: "new direction"}, + }) + if err != nil { + t.Fatalf("continueWithSteeringMessages failed: %v", err) + } + if response.Content != "continued response" { + t.Fatalf("expected continued response, got %q", response.Content) + } + if response.OnDelivered == nil { + t.Fatal("expected OnDelivered callback for continued response") + } + + history := defaultAgent.Sessions.GetHistory(sessionKey) + if len(history) != 1 || history[0].Role != "user" { + t.Fatalf("expected only steering user message before delivery, got %#v", history) + } + + response.OnDelivered([]string{"out-continue"}) + + history = defaultAgent.Sessions.GetHistory(sessionKey) + if len(history) != 2 { + t.Fatalf("expected 2 messages after delivery, got %d", len(history)) + } + if history[1].Role != "assistant" { + t.Fatalf("expected assistant message, got %+v", history[1]) + } + if len(history[1].MessageIDs) != 1 || history[1].MessageIDs[0] != "out-continue" { + t.Fatalf("expected assistant message_ids [out-continue], got %v", history[1].MessageIDs) + } +} + // slowTool simulates a tool that takes some time to execute. type slowTool struct { name string diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 0366e89df..30746615c 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -18,17 +18,18 @@ type SenderInfo struct { } type InboundMessage struct { - Channel string `json:"channel"` - SenderID string `json:"sender_id"` - Sender SenderInfo `json:"sender"` - ChatID string `json:"chat_id"` - Content string `json:"content"` - Media []string `json:"media,omitempty"` - Peer Peer `json:"peer"` // routing peer - MessageID string `json:"message_id,omitempty"` // platform message ID - MediaScope string `json:"media_scope,omitempty"` // media lifecycle scope - SessionKey string `json:"session_key"` - Metadata map[string]string `json:"metadata,omitempty"` + Channel string `json:"channel"` + SenderID string `json:"sender_id"` + Sender SenderInfo `json:"sender"` + ChatID string `json:"chat_id"` + Content string `json:"content"` + Media []string `json:"media,omitempty"` + Peer Peer `json:"peer"` // routing peer + MessageID string `json:"message_id,omitempty"` // platform message ID + ReplyToMessageID string `json:"reply_to_message_id,omitempty"` // parent platform message ID + MediaScope string `json:"media_scope,omitempty"` // media lifecycle scope + SessionKey string `json:"session_key"` + Metadata map[string]string `json:"metadata,omitempty"` } type OutboundMessage struct { diff --git a/pkg/channels/README.md b/pkg/channels/README.md index 7f238ece5..e16d364c3 100644 --- a/pkg/channels/README.md +++ b/pkg/channels/README.md @@ -881,6 +881,7 @@ type InboundMessage struct { Media []string // Media reference list (media://...) Peer Peer // Routing peer (first-class field) MessageID string // Platform message ID (first-class field) + ReplyToMessageID string // Parent platform message ID (first-class field) MediaScope string // Media lifecycle scope SessionKey string // Session key Metadata map[string]string // Only for channel-specific extensions @@ -1191,10 +1192,11 @@ Timeout configuration: ReadTimeout = 30s, WriteTimeout = 30s **Do NOT put the following information in Metadata anymore**: - `peer_kind` / `peer_id` → Use `InboundMessage.Peer` - `message_id` → Use `InboundMessage.MessageID` +- `reply_to_message_id` → Use `InboundMessage.ReplyToMessageID` - `sender_platform` / `sender_username` → Use `InboundMessage.Sender` **Metadata should only be used for**: -- Channel-specific extension information (e.g., Telegram's `reply_to_message_id`) +- Channel-specific extension information that has no structured field yet - Temporary information that doesn't fit into structured fields ### 5.3 Concurrency Safety Conventions @@ -1380,4 +1382,4 @@ agentLoop.Stop() // Stop Agent 7. **PlaceholderConfig vs implementation**: `PlaceholderConfig` appears in 6 channel configs (Telegram, Discord, Slack, LINE, OneBot, Pico), but only channels that implement both `PlaceholderCapable` + `MessageEditor` (Telegram, Discord, Pico) can actually use placeholder message editing. The rest are reserved fields. -8. **ReasoningChannelID**: Most channel configs include a `reasoning_channel_id` field to route LLM reasoning/thinking output to a designated channel (WhatsApp, Telegram, Feishu, Discord, MaixCam, QQ, DingTalk, Slack, LINE, OneBot, WeCom). Note: `PicoConfig` does not currently expose this field. `BaseChannel` exposes this via the `WithReasoningChannelID` option and `ReasoningChannelID()` method. +8. **ReasoningChannelID**: Most channel configs include a `reasoning_channel_id` field to route LLM reasoning/thinking output to a designated channel (WhatsApp, Telegram, Feishu, Discord, MaixCam, QQ, DingTalk, Slack, LINE, OneBot, WeCom, WeComApp). Note: `PicoConfig` does not currently expose this field. `BaseChannel` exposes this via the `WithReasoningChannelID` option and `ReasoningChannelID()` method. diff --git a/pkg/channels/README.zh.md b/pkg/channels/README.zh.md index 8bc8c8dbc..7e122ae1b 100644 --- a/pkg/channels/README.zh.md +++ b/pkg/channels/README.zh.md @@ -880,6 +880,7 @@ type InboundMessage struct { Media []string // 媒体引用列表（media://...） Peer Peer // 路由对等体（一等字段） MessageID string // 平台消息 ID（一等字段） + ReplyToMessageID string // 父消息 ID（一等字段） MediaScope string // 媒体生命周期作用域 SessionKey string // 会话键 Metadata map[string]string // 仅用于 channel 特有扩展 @@ -1190,10 +1191,11 @@ Manager 创建单一 `http.Server`，自动发现和注册： **不要再把以下信息放入 Metadata**： - `peer_kind` / `peer_id` → 使用 `InboundMessage.Peer` - `message_id` → 使用 `InboundMessage.MessageID` +- `reply_to_message_id` → 使用 `InboundMessage.ReplyToMessageID` - `sender_platform` / `sender_username` → 使用 `InboundMessage.Sender` **Metadata 仅用于**： -- Channel 特有的扩展信息（如 Telegram 的 `reply_to_message_id`） +- 尚未有结构化字段承载的 Channel 特有扩展信息 - 不适合放入结构化字段的临时信息 ### 5.3 并发安全约定 @@ -1379,4 +1381,4 @@ agentLoop.Stop() // 停止 Agent 7. **PlaceholderConfig 的配置与实现**：`PlaceholderConfig` 出现在 6 个 channel config 中（Telegram、Discord、Slack、LINE、OneBot、Pico），但只有实现了 `PlaceholderCapable` + `MessageEditor` 的 channel（Telegram、Discord、Pico）能真正使用占位消息编辑功能。其余 channel 的 `PlaceholderConfig` 为预留字段。 -8. **ReasoningChannelID**：大多数 channel config 都包含 `reasoning_channel_id` 字段，用于将 LLM 的思维链（reasoning/thinking）路由到指定 channel（WhatsApp、Telegram、Feishu、Discord、MaixCam、QQ、DingTalk、Slack、LINE、OneBot、WeCom）。注意：`PicoConfig` 目前不包含该字段。`BaseChannel` 通过 `WithReasoningChannelID` 选项和 `ReasoningChannelID()` 方法暴露此配置。 +8. **ReasoningChannelID**：大多数 channel config 都包含 `reasoning_channel_id` 字段，用于将 LLM 的思维链（reasoning/thinking）路由到指定 channel（WhatsApp、Telegram、Feishu、Discord、MaixCam、QQ、DingTalk、Slack、LINE、OneBot、WeCom、WeComApp）。注意：`PicoConfig` 目前不包含该字段。`BaseChannel` 通过 `WithReasoningChannelID` 选项和 `ReasoningChannelID()` 方法暴露此配置。 diff --git a/pkg/channels/base.go b/pkg/channels/base.go index 882e72d08..f50e2abef 100644 --- a/pkg/channels/base.go +++ b/pkg/channels/base.go @@ -259,18 +259,23 @@ func (c *BaseChannel) HandleMessage( } scope := BuildMediaScope(c.name, chatID, messageID) + replyToMessageID := "" + if metadata != nil { + replyToMessageID = metadata["reply_to_message_id"] + } msg := bus.InboundMessage{ - Channel: c.name, - SenderID: resolvedSenderID, - Sender: sender, - ChatID: chatID, - Content: content, - Media: media, - Peer: peer, - MessageID: messageID, - MediaScope: scope, - Metadata: metadata, + Channel: c.name, + SenderID: resolvedSenderID, + Sender: sender, + ChatID: chatID, + Content: content, + Media: media, + Peer: peer, + MessageID: messageID, + ReplyToMessageID: replyToMessageID, + MediaScope: scope, + Metadata: metadata, } // Auto-trigger typing indicator, message reaction, and placeholder before publishing. diff --git a/pkg/channels/base_test.go b/pkg/channels/base_test.go index 6132b8bf9..3f462704a 100644 --- a/pkg/channels/base_test.go +++ b/pkg/channels/base_test.go @@ -1,6 +1,7 @@ package channels import ( + "context" "testing" "github.com/sipeed/picoclaw/pkg/bus" @@ -263,3 +264,32 @@ func TestIsAllowedSender(t *testing.T) { }) } } + +func TestBaseChannelHandleMessage_PopulatesReplyToMessageID(t *testing.T) { + msgBus := bus.NewMessageBus() + defer msgBus.Close() + + ch := NewBaseChannel("test", nil, msgBus, nil) + ch.HandleMessage( + context.Background(), + bus.Peer{Kind: "direct", ID: "user1"}, + "msg-2", + "user1", + "chat1", + "hello", + nil, + map[string]string{"reply_to_message_id": "msg-1"}, + ) + + select { + case got := <-msgBus.InboundChan(): + if got.MessageID != "msg-2" { + t.Fatalf("expected MessageID msg-2, got %q", got.MessageID) + } + if got.ReplyToMessageID != "msg-1" { + t.Fatalf("expected ReplyToMessageID msg-1, got %q", got.ReplyToMessageID) + } + case <-context.Background().Done(): + t.Fatal("expected inbound message") + } +}