opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix message sender, threading, and continuation delivery

Browse source
This commit is contained in:
Dmitrii Balabanov 2026-03-23 10:25:12 +02:00
parent f15f844ad1
commit 6cd3fdc18f
11 changed files with 394 additions and 95 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
pkg/agent/context.go
View file

@ -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 ""
}

32
pkg/agent/context_test.go
View file

@ -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 {

51
pkg/agent/loop.go
View file

@ -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...),
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.

53
pkg/agent/loop_test.go
View file

@ -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()

69
pkg/agent/steering.go
View file

@ -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
}
return response.Content, 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)
}
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 {

123
pkg/agent/steering_test.go
View file

@ -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

1
pkg/bus/types.go
View file

@ -26,6 +26,7 @@ type InboundMessage struct {
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"`

6
pkg/channels/README.md
View file

@ -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.

6
pkg/channels/README.zh.md
View file

@ -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()` 方法暴露此配置。

5
pkg/channels/base.go
View file

@ -259,6 +259,10 @@ 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,
@ -269,6 +273,7 @@ func (c *BaseChannel) HandleMessage(
Media: media,
Peer: peer,
MessageID: messageID,
ReplyToMessageID: replyToMessageID,
MediaScope: scope,
Metadata: metadata,
}

30
pkg/channels/base_test.go
View file

@ -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")
}
}