From 51933ecd09b5ec4c5829152c6588633a88c1abb5 Mon Sep 17 00:00:00 2001 From: Dmitrii Balabanov Date: Mon, 9 Mar 2026 16:39:09 +0200 Subject: [PATCH 1/2] Add Telegram reply routing and group placeholder handling --- pkg/agent/context.go | 39 +++++ pkg/agent/context_cache_test.go | 38 ++++- pkg/agent/loop.go | 158 ++++++++++++++---- pkg/agent/loop_test.go | 141 +++++++++++++++- pkg/bus/types.go | 14 +- pkg/channels/interfaces.go | 6 + pkg/channels/manager.go | 17 +- pkg/channels/manager_test.go | 93 +++++++++++ pkg/channels/telegram/telegram.go | 77 ++++++++- .../telegram/telegram_dispatch_test.go | 40 +++++ pkg/channels/telegram/telegram_test.go | 45 ++++- pkg/tools/base.go | 31 +++- pkg/tools/message.go | 73 +++++++- pkg/tools/message_test.go | 140 +++++++++++++--- pkg/tools/registry_test.go | 9 +- 15 files changed, 838 insertions(+), 83 deletions(-) diff --git a/pkg/agent/context.go b/pkg/agent/context.go index 719b0cb6d..665496781 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -438,12 +438,45 @@ func (cb *ContextBuilder) buildDynamicContext(channel, chatID string) string { return sb.String() } +type ReplyContextInfo struct { + CurrentMessageID string + ParentMessageID string +} + +func buildReplyRoutingContext(channel string, replyCtx *ReplyContextInfo) string { + if channel != "telegram" || replyCtx == nil || strings.TrimSpace(replyCtx.CurrentMessageID) == "" { + return "" + } + + parentID := strings.TrimSpace(replyCtx.ParentMessageID) + if parentID == "" { + parentID = "(none)" + } + + return fmt.Sprintf( + "## Reply Routing\n"+ + "Current inbound message ID: %s\n"+ + "Parent message ID: %s\n\n"+ + "To control Telegram reply threading through the final answer, you may put exactly one hidden directive on the first line of your final response:\n"+ + "- `[[reply:chat]]` posts a normal chat message\n"+ + "- `[[reply:current]]` replies to the current inbound message\n"+ + "- `[[reply:parent]]` replies to the parent/replied-to message when there is one\n"+ + "- `[[reply:message_id=123]]` replies to a specific known message ID\n\n"+ + "After the directive, add a blank line and then the user-visible message.\n"+ + "If you do not need special routing, answer normally without a directive.\n"+ + "Never mention the directive in the visible message body.", + replyCtx.CurrentMessageID, + parentID, + ) +} + func (cb *ContextBuilder) BuildMessages( history []providers.Message, summary string, currentMessage string, media []string, channel, chatID string, + replyCtx *ReplyContextInfo, ) []providers.Message { messages := []providers.Message{} @@ -460,6 +493,7 @@ func (cb *ContextBuilder) BuildMessages( // Build short dynamic context (time, runtime, session) — changes per request dynamicCtx := cb.buildDynamicContext(channel, chatID) + replyRoutingCtx := buildReplyRoutingContext(channel, replyCtx) // Compose a single system message: static (cached) + dynamic + optional summary. // Keeping all system content in one message ensures every provider adapter can @@ -477,6 +511,11 @@ func (cb *ContextBuilder) BuildMessages( {Type: "text", Text: dynamicCtx}, } + if replyRoutingCtx != "" { + stringParts = append(stringParts, replyRoutingCtx) + contentBlocks = append(contentBlocks, providers.ContentBlock{Type: "text", Text: replyRoutingCtx}) + } + if summary != "" { summaryText := fmt.Sprintf( "CONTEXT_SUMMARY: The following is an approximate summary of prior conversation "+ diff --git a/pkg/agent/context_cache_test.go b/pkg/agent/context_cache_test.go index 707510820..d2e605ca0 100644 --- a/pkg/agent/context_cache_test.go +++ b/pkg/agent/context_cache_test.go @@ -82,7 +82,7 @@ func TestSingleSystemMessage(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - msgs := cb.BuildMessages(tt.history, tt.summary, tt.message, nil, "test", "chat1") + msgs := cb.BuildMessages(tt.history, tt.summary, tt.message, nil, "test", "chat1", nil) systemCount := 0 for _, m := range msgs { @@ -126,6 +126,38 @@ func TestSingleSystemMessage(t *testing.T) { } } +func TestBuildMessages_TelegramReplyRoutingContext(t *testing.T) { + tmpDir := setupWorkspace(t, map[string]string{ + "IDENTITY.md": "# Identity\nTest agent.", + }) + defer os.RemoveAll(tmpDir) + + cb := NewContextBuilder(tmpDir) + msgs := cb.BuildMessages( + nil, + "", + "hello", + nil, + "telegram", + "chat1", + &ReplyContextInfo{ + CurrentMessageID: "910", + ParentMessageID: "905", + }, + ) + + sys := msgs[0].Content + if !strings.Contains(sys, "## Reply Routing") { + t.Fatal("system prompt missing reply routing section") + } + if !strings.Contains(sys, "Current inbound message ID: 910") { + t.Fatal("system prompt missing current message ID") + } + if !strings.Contains(sys, "[[reply:current]]") { + t.Fatal("system prompt missing final reply directive guidance") + } +} + // TestMtimeAutoInvalidation verifies that the cache detects source file changes // via mtime without requiring explicit InvalidateCache(). // Fix: original implementation had no auto-invalidation — edits to bootstrap files, @@ -576,7 +608,7 @@ func TestConcurrentBuildSystemPromptWithCache(t *testing.T) { } // Also exercise BuildMessages concurrently - msgs := cb.BuildMessages(nil, "", "hello", nil, "test", "chat") + msgs := cb.BuildMessages(nil, "", "hello", nil, "test", "chat", nil) if len(msgs) < 2 { errs <- "BuildMessages returned fewer than 2 messages" return @@ -664,6 +696,6 @@ func BenchmarkBuildMessagesWithCache(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { - _ = cb.BuildMessages(history, "summary", "new message", nil, "cli", "test") + _ = cb.BuildMessages(history, "summary", "new message", nil, "cli", "test", nil) } } diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 23dd079ac..f2b9c1131 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -63,6 +63,21 @@ type processOptions struct { EnableSummary bool // Whether to trigger summarization SendResponse bool // Whether to send response via bus NoHistory bool // If true, don't load session history (for heartbeat) + ReplyContext *ReplyContextInfo +} + +type agentResponse struct { + Content string + ReplyToMessageID string +} + +func (r agentResponse) outboundMessage(channel, chatID string) bus.OutboundMessage { + return bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: r.Content, + ReplyToMessageID: r.ReplyToMessageID, + } } const ( @@ -73,6 +88,7 @@ const ( metadataKeyTeamID = "team_id" metadataKeyParentPeerKind = "parent_peer_kind" metadataKeyParentPeerID = "parent_peer_id" + metadataKeyReplyToMessage = "reply_to_message_id" metadataKeyRouteAgentID = "route_agent_id" metadataKeyRouteMatchedBy = "route_matched_by" ) @@ -177,14 +193,10 @@ func registerSharedTools( // Message tool if cfg.Tools.IsToolEnabled("message") { messageTool := tools.NewMessageTool() - messageTool.SetSendCallback(func(channel, chatID, content string) error { + messageTool.SetSendCallback(func(msg bus.OutboundMessage) error { pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second) defer pubCancel() - return msgBus.PublishOutbound(pubCtx, bus.OutboundMessage{ - Channel: channel, - ChatID: chatID, - Content: content, - }) + return msgBus.PublishOutbound(pubCtx, msg) }) agent.Tools.Register(messageTool) } @@ -344,10 +356,10 @@ func (al *AgentLoop) Run(ctx context.Context) error { response, err := al.processMessage(ctx, msg) if err != nil { - response = fmt.Sprintf("Error processing message: %v", err) + response = agentResponse{Content: fmt.Sprintf("Error processing message: %v", err)} } - if response != "" { + if response.Content != "" { // Check if the message tool already sent a response during this round. // If so, skip publishing to avoid duplicate messages to the user. // Use default agent's tools to check (message tool is shared). @@ -362,16 +374,12 @@ func (al *AgentLoop) Run(ctx context.Context) error { } if !alreadySent { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Channel: msg.Channel, - ChatID: msg.ChatID, - Content: response, - }) + al.bus.PublishOutbound(ctx, response.outboundMessage(msg.Channel, msg.ChatID)) logger.InfoCF("agent", "Published outbound response", map[string]any{ "channel": msg.Channel, "chat_id": msg.ChatID, - "content_len": len(response), + "content_len": len(response.Content), }) } else { logger.DebugCF( @@ -565,7 +573,8 @@ func (al *AgentLoop) ProcessDirectWithChannel( SessionKey: sessionKey, } - return al.processMessage(ctx, msg) + response, err := al.processMessage(ctx, msg) + return response.Content, err } // ProcessHeartbeat processes a heartbeat request without session history. @@ -578,7 +587,7 @@ func (al *AgentLoop) ProcessHeartbeat( if agent == nil { return "", fmt.Errorf("no default agent for heartbeat") } - return al.runAgentLoop(ctx, agent, processOptions{ + response, err := al.runAgentLoop(ctx, agent, processOptions{ SessionKey: "heartbeat", Channel: channel, ChatID: chatID, @@ -588,9 +597,10 @@ func (al *AgentLoop) ProcessHeartbeat( SendResponse: false, NoHistory: true, // Don't load session history for heartbeat }) + return response.Content, err } -func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) { +func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (agentResponse, error) { // Add message preview to log (show full content for error messages) var logContent string if strings.Contains(msg.Content, "Error:") || strings.Contains(msg.Content, "error") { @@ -618,7 +628,14 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) route, agent, routeErr := al.resolveMessageRoute(msg) if routeErr != nil { - return "", routeErr + // Commands are checked before requiring a successful route. + // Global commands (/help, /show, /switch) work even when routing fails; + // context-dependent commands check their own Runtime fields and report + // "unavailable" when the required capability is nil. + if response, handled := al.handleCommand(ctx, msg, agent, nil); handled { + return agentResponse{Content: response}, nil + } + return agentResponse{}, routeErr } // Reset message-tool state for this round so we don't skip publishing due to a previous round. @@ -651,12 +668,16 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) DefaultResponse: defaultResponse, EnableSummary: true, SendResponse: false, + ReplyContext: &ReplyContextInfo{ + CurrentMessageID: msg.MessageID, + ParentMessageID: inboundMetadata(msg, metadataKeyReplyToMessage), + }, } // context-dependent commands check their own Runtime fields and report // "unavailable" when the required capability is nil. if response, handled := al.handleCommand(ctx, msg, agent, &opts); handled { - return response, nil + return agentResponse{Content: response}, nil } return al.runAgentLoop(ctx, agent, opts) @@ -695,9 +716,9 @@ func resolveScopeKey(route routing.ResolvedRoute, msgSessionKey string) string { func (al *AgentLoop) processSystemMessage( ctx context.Context, msg bus.InboundMessage, -) (string, error) { +) (agentResponse, error) { if msg.Channel != "system" { - return "", fmt.Errorf( + return agentResponse{}, fmt.Errorf( "processSystemMessage called with non-system message channel: %s", msg.Channel, ) @@ -734,13 +755,13 @@ func (al *AgentLoop) processSystemMessage( "content_len": len(content), "channel": originChannel, }) - return "", nil + return agentResponse{}, nil } // Use default agent for system messages agent := al.registry.GetDefaultAgent() if agent == nil { - return "", fmt.Errorf("no default agent for system message") + return agentResponse{}, fmt.Errorf("no default agent for system message") } // Use the origin session for context @@ -762,7 +783,7 @@ func (al *AgentLoop) runAgentLoop( ctx context.Context, agent *AgentInstance, opts processOptions, -) (string, error) { +) (agentResponse, error) { // 0. Record last channel for heartbeat notifications (skip internal channels and cli) if opts.Channel != "" && opts.ChatID != "" { if !constants.IsInternalChannel(opts.Channel) { @@ -791,6 +812,7 @@ func (al *AgentLoop) runAgentLoop( opts.Media, opts.Channel, opts.ChatID, + opts.ReplyContext, ) // Resolve media:// refs to base64 data URLs (streaming) @@ -803,9 +825,16 @@ func (al *AgentLoop) runAgentLoop( // 3. Run LLM iteration loop // Inject session key so tools (e.g. tasktool) can look it up from context. ctx = tools.WithToolSessionKey(ctx, opts.SessionKey) + if opts.ReplyContext != nil { + ctx = tools.WithToolReplyContext( + ctx, + opts.ReplyContext.CurrentMessageID, + opts.ReplyContext.ParentMessageID, + ) + } finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts) if err != nil { - return "", err + return agentResponse{}, err } // If last tool had ForUser content and we already sent it, we might not need to send final response @@ -815,9 +844,13 @@ func (al *AgentLoop) runAgentLoop( if finalContent == "" { finalContent = opts.DefaultResponse } + response := resolveFinalResponse(opts.Channel, opts.ReplyContext, finalContent) + if response.Content == "" { + response.Content = opts.DefaultResponse + } // 5. Save final assistant message to session - agent.Sessions.AddMessage(opts.SessionKey, "assistant", finalContent) + agent.Sessions.AddMessage(opts.SessionKey, "assistant", response.Content) agent.Sessions.Save(opts.SessionKey) // 6. Optional: summarization @@ -827,24 +860,79 @@ func (al *AgentLoop) runAgentLoop( // 7. Optional: send response via bus if opts.SendResponse { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: finalContent, - }) + al.bus.PublishOutbound(ctx, response.outboundMessage(opts.Channel, opts.ChatID)) } // 8. Log response - responsePreview := utils.Truncate(finalContent, 120) + responsePreview := utils.Truncate(response.Content, 120) logger.InfoCF("agent", fmt.Sprintf("Response: %s", responsePreview), map[string]any{ "agent_id": agent.ID, "session_key": opts.SessionKey, "iterations": iteration, - "final_length": len(finalContent), + "final_length": len(response.Content), }) - return finalContent, nil + return response, nil +} + +func resolveFinalResponse( + channel string, + replyCtx *ReplyContextInfo, + rawContent string, +) agentResponse { + content, replyToMessageID := parseFinalReplyDirective(channel, replyCtx, rawContent) + return agentResponse{ + Content: content, + ReplyToMessageID: replyToMessageID, + } +} + +func parseFinalReplyDirective( + channel string, + replyCtx *ReplyContextInfo, + rawContent string, +) (content, replyToMessageID string) { + content = rawContent + if channel != "telegram" { + return content, "" + } + + firstLine, rest, hasRest := strings.Cut(rawContent, "\n") + directive := strings.TrimSpace(firstLine) + if !strings.HasPrefix(directive, "[[reply:") || !strings.HasSuffix(directive, "]]") { + return content, "" + } + + body := "" + if hasRest { + body = strings.TrimLeft(rest, "\n") + } + + mode := strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(directive, "[[reply:"), "]]")) + switch { + case mode == "chat": + return body, "" + case mode == "current": + if replyCtx != nil && strings.TrimSpace(replyCtx.CurrentMessageID) != "" { + return body, strings.TrimSpace(replyCtx.CurrentMessageID) + } + case mode == "parent": + if replyCtx != nil && strings.TrimSpace(replyCtx.ParentMessageID) != "" { + return body, strings.TrimSpace(replyCtx.ParentMessageID) + } + case strings.HasPrefix(mode, "message_id="): + id := strings.TrimSpace(strings.TrimPrefix(mode, "message_id=")) + if id != "" { + return body, id + } + } + + logger.WarnCF("agent", "Ignoring invalid final reply directive", map[string]any{ + "channel": channel, + "directive": directive, + }) + return body, "" } func (al *AgentLoop) targetReasoningChannelID(channelName string) (chatID string) { @@ -1061,7 +1149,7 @@ func (al *AgentLoop) runLLMIteration( newSummary := agent.Sessions.GetSummary(opts.SessionKey) messages = agent.ContextBuilder.BuildMessages( newHistory, newSummary, "", - nil, opts.Channel, opts.ChatID, + nil, opts.Channel, opts.ChatID, opts.ReplyContext, ) continue } diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index acd81bef1..eaafbaa81 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -526,7 +526,7 @@ func (h testHelper) executeAndGetResponse(tb testing.TB, ctx context.Context, ms if err != nil { tb.Fatalf("processMessage failed: %v", err) } - return response + return response.Content } const responseTimeout = 3 * time.Second @@ -587,6 +587,145 @@ func TestProcessMessage_UsesRouteSessionKey(t *testing.T) { } } +func TestParseFinalReplyDirective(t *testing.T) { + content, replyTo := parseFinalReplyDirective( + "telegram", + &ReplyContextInfo{ + CurrentMessageID: "910", + ParentMessageID: "905", + }, + "[[reply:parent]]\n\nThreaded answer", + ) + + if content != "Threaded answer" { + t.Fatalf("content=%q", content) + } + if replyTo != "905" { + t.Fatalf("replyTo=%q", replyTo) + } +} + +func TestProcessMessage_TelegramFinalDirectiveSetsReplyTarget(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 := &simpleMockProvider{response: "[[reply:parent]]\n\nThreaded answer"} + al := NewAgentLoop(cfg, msgBus, provider) + + msg := bus.InboundMessage{ + Channel: "telegram", + SenderID: "user1", + ChatID: "chat1", + Content: "hello", + MessageID: "910", + Metadata: map[string]string{ + "reply_to_message_id": "905", + }, + Peer: bus.Peer{ + Kind: "direct", + ID: "user1", + }, + } + + response, err := al.processMessage(context.Background(), msg) + if err != nil { + t.Fatalf("processMessage failed: %v", err) + } + if response.Content != "Threaded answer" { + t.Fatalf("content=%q", response.Content) + } + if response.ReplyToMessageID != "905" { + t.Fatalf("reply_to_message_id=%q", response.ReplyToMessageID) + } +} + +func TestRun_PublishesTelegramReplyTargetFromFinalDirective(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 := &simpleMockProvider{response: "[[reply:parent]]\n\nThreaded answer"} + al := NewAgentLoop(cfg, msgBus, provider) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- al.Run(ctx) + }() + defer func() { + al.Stop() + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("agent loop did not stop in time") + } + }() + + inbound := bus.InboundMessage{ + Channel: "telegram", + SenderID: "user1", + ChatID: "chat1", + Content: "hello", + MessageID: "910", + Metadata: map[string]string{ + "reply_to_message_id": "905", + }, + Peer: bus.Peer{ + Kind: "direct", + ID: "user1", + }, + } + if err := msgBus.PublishInbound(context.Background(), inbound); err != nil { + t.Fatalf("publish inbound: %v", err) + } + + outCtx, outCancel := context.WithTimeout(context.Background(), time.Second) + defer outCancel() + + outbound, ok := msgBus.SubscribeOutbound(outCtx) + if !ok { + t.Fatal("expected outbound message") + } + if outbound.Content != "Threaded answer" { + t.Fatalf("content=%q", outbound.Content) + } + if outbound.ReplyToMessageID != "905" { + t.Fatalf("reply_to_message_id=%q", outbound.ReplyToMessageID) + } +} + func TestProcessMessage_CommandOutcomes(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") if err != nil { diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 7ad8f0417..98927dfbd 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -30,9 +30,10 @@ type InboundMessage struct { } type OutboundMessage struct { - Channel string `json:"channel"` - ChatID string `json:"chat_id"` - Content string `json:"content"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Content string `json:"content"` + ReplyToMessageID string `json:"reply_to_message_id,omitempty"` } // MediaPart describes a single media attachment to send. @@ -46,7 +47,8 @@ type MediaPart struct { // OutboundMediaMessage carries media attachments from Agent to channels via the bus. type OutboundMediaMessage struct { - Channel string `json:"channel"` - ChatID string `json:"chat_id"` - Parts []MediaPart `json:"parts"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Parts []MediaPart `json:"parts"` + ReplyToMessageID string `json:"reply_to_message_id,omitempty"` } diff --git a/pkg/channels/interfaces.go b/pkg/channels/interfaces.go index 4ca975356..3986cd8b8 100644 --- a/pkg/channels/interfaces.go +++ b/pkg/channels/interfaces.go @@ -20,6 +20,12 @@ type MessageEditor interface { EditMessage(ctx context.Context, chatID string, messageID string, content string) error } +// MessageDeleter — channels that can delete an existing message. +// messageID is always string; channels convert platform-specific types internally. +type MessageDeleter interface { + DeleteMessage(ctx context.Context, chatID string, messageID string) error +} + // ReactionCapable — channels that can add a reaction (e.g. 👀) to an inbound message. // ReactToMessage adds a reaction and returns an undo function to remove it. // The undo function MUST be idempotent and safe to call multiple times. diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index ed2afdda2..5c199849e 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -134,9 +134,24 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess } } - // 3. Try editing placeholder + // 3. Try editing placeholder. + // Reply-targeted outbound messages must remain new sends so the transport can + // attach platform reply metadata; editing a placeholder would lose that target. if v, loaded := m.placeholders.LoadAndDelete(key); loaded { if entry, ok := v.(placeholderEntry); ok && entry.id != "" { + if msg.ReplyToMessageID != "" { + if deleter, ok := ch.(MessageDeleter); ok { + if err := deleter.DeleteMessage(ctx, msg.ChatID, entry.id); err != nil { + logger.WarnCF("manager", "Failed to delete placeholder before reply-targeted send", map[string]any{ + "channel": name, + "chat_id": msg.ChatID, + "placeholder_id": entry.id, + "error": err.Error(), + }) + } + } + return false + } if editor, ok := ch.(MessageEditor); ok { if err := editor.EditMessage(ctx, msg.ChatID, entry.id, msg.Content); err == nil { return true // edited successfully, skip Send diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index 7d9ea700e..1045502af 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -461,6 +461,15 @@ func (m *mockMessageEditor) EditMessage(ctx context.Context, chatID, messageID, return m.editFn(ctx, chatID, messageID, content) } +type mockMessageEditorDeleter struct { + mockMessageEditor + deleteFn func(ctx context.Context, chatID, messageID string) error +} + +func (m *mockMessageEditorDeleter) DeleteMessage(ctx context.Context, chatID, messageID string) error { + return m.deleteFn(ctx, chatID, messageID) +} + func TestPreSend_PlaceholderEditSuccess(t *testing.T) { m := newTestManager() var sendCalled bool @@ -529,6 +538,90 @@ func TestPreSend_PlaceholderEditFails_FallsThrough(t *testing.T) { } } +func TestPreSend_ReplyTargetSkipsPlaceholderEdit(t *testing.T) { + m := newTestManager() + var editCalled bool + var deleteCalled bool + + ch := &mockMessageEditorDeleter{ + mockMessageEditor: mockMessageEditor{ + mockChannel: mockChannel{ + sendFn: func(_ context.Context, _ bus.OutboundMessage) error { + return nil + }, + }, + editFn: func(_ context.Context, _, _, _ string) error { + editCalled = true + return nil + }, + }, + deleteFn: func(_ context.Context, chatID, messageID string) error { + deleteCalled = true + if chatID != "123" { + t.Fatalf("expected chatID 123, got %s", chatID) + } + if messageID != "456" { + t.Fatalf("expected messageID 456, got %s", messageID) + } + return nil + }, + } + + m.RecordPlaceholder("test", "123", "456") + + msg := bus.OutboundMessage{ + Channel: "test", + ChatID: "123", + Content: "hello", + ReplyToMessageID: "99", + } + edited := m.preSend(context.Background(), "test", msg, ch) + + if edited { + t.Fatal("expected preSend to fall through for reply-targeted outbound") + } + if editCalled { + t.Fatal("expected placeholder edit to be skipped when reply target is set") + } + if !deleteCalled { + t.Fatal("expected placeholder delete to be attempted for reply-targeted outbound") + } +} + +func TestPreSend_ReplyTargetWithoutDeleterStillSkipsEdit(t *testing.T) { + m := newTestManager() + var editCalled bool + + ch := &mockMessageEditor{ + mockChannel: mockChannel{ + sendFn: func(_ context.Context, _ bus.OutboundMessage) error { + return nil + }, + }, + editFn: func(_ context.Context, _, _, _ string) error { + editCalled = true + return nil + }, + } + + m.RecordPlaceholder("test", "123", "456") + + msg := bus.OutboundMessage{ + Channel: "test", + ChatID: "123", + Content: "hello", + ReplyToMessageID: "99", + } + edited := m.preSend(context.Background(), "test", msg, ch) + + if edited { + t.Fatal("expected preSend to fall through for reply-targeted outbound") + } + if editCalled { + t.Fatal("expected placeholder edit to be skipped when reply target is set") + } +} + func TestPreSend_TypingStopCalled(t *testing.T) { m := newTestManager() var stopCalled bool diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index cbc1664d2..60a89be04 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -186,7 +186,13 @@ func (c *TelegramChannel) SendMessageWithID(ctx context.Context, msg bus.Outboun chunks := telegramMessageChunks(msg.Content) ids := make([]string, 0, len(chunks)) for _, chunk := range chunks { - msgID, err := c.sendHTMLChunk(ctx, target, chunk.HTML, chunk.Markdown) + msgID, err := c.sendHTMLChunk( + ctx, + target, + msg.ReplyToMessageID, + chunk.HTML, + chunk.Markdown, + ) if err != nil { return "", err } @@ -204,6 +210,7 @@ func (c *TelegramChannel) SendMessageWithID(ctx context.Context, msg bus.Outboun func (c *TelegramChannel) sendHTMLChunk( ctx context.Context, target telegramTarget, + replyToMessageID string, htmlContent, mdFallback string, ) (int, error) { tgMsg := tu.Message(tu.ID(target.ChatID), htmlContent) @@ -211,6 +218,9 @@ func (c *TelegramChannel) sendHTMLChunk( if threadID, ok := target.messageThreadIDForSend(); ok { tgMsg.MessageThreadID = threadID } + if replyParams, ok := telegramReplyParameters(replyToMessageID); ok { + tgMsg.ReplyParameters = replyParams + } msg, err := c.bot.SendMessage(ctx, tgMsg) if err != nil { @@ -227,6 +237,23 @@ func (c *TelegramChannel) sendHTMLChunk( return msg.MessageID, nil } +func telegramReplyParameters(replyToMessageID string) (*telego.ReplyParameters, bool) { + replyToMessageID = strings.TrimSpace(replyToMessageID) + if replyToMessageID == "" { + return nil, false + } + + id, err := strconv.Atoi(replyToMessageID) + if err != nil || id <= 0 { + return nil, false + } + + return &telego.ReplyParameters{ + MessageID: id, + AllowSendingWithoutReply: true, + }, true +} + // StartTyping implements channels.TypingCapable. // It sends ChatAction(typing) immediately and then repeats every 4 seconds // (Telegram's typing indicator expires after ~5s) in a background goroutine. @@ -292,6 +319,26 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag return nil } +// DeleteMessage implements channels.MessageDeleter. +func (c *TelegramChannel) DeleteMessage(ctx context.Context, chatID string, messageID string) error { + target, err := parseTelegramTarget(chatID) + if err != nil { + return err + } + messageIDs, err := parseTelegramMessageIDs(messageID) + if err != nil { + return err + } + + for _, mid := range messageIDs { + if err := c.bot.DeleteMessage(ctx, tu.Delete(tu.ID(target.ChatID), mid)); err != nil { + return fmt.Errorf("telegram delete: %w", channels.ErrTemporary) + } + } + + return nil +} + // SendPlaceholder implements channels.PlaceholderCapable. // It sends a placeholder message (e.g. "Thinking... 💭") that will later be // edited to the actual response via EditMessage (channels.MessageEditor). @@ -304,15 +351,18 @@ func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (s return "", nil } - text := phCfg.Text - if text == "" { - text = "Thinking... 💭" - } - target, err := parseTelegramTarget(chatID) if err != nil { return "", err } + if target.ChatID < 0 { + return "", nil + } + + text := phCfg.Text + if text == "" { + text = "Thinking... 💭" + } params := tu.Message(tu.ID(target.ChatID), text) if threadID, ok := target.messageThreadIDForSend(); ok { @@ -371,6 +421,9 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe if threadID, ok := target.messageThreadIDForSend(); ok { params.MessageThreadID = threadID } + if replyParams, ok := telegramReplyParameters(msg.ReplyToMessageID); ok { + params.ReplyParameters = replyParams + } _, err = c.bot.SendPhoto(ctx, params) case "audio": params := &telego.SendAudioParams{ @@ -381,6 +434,9 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe if threadID, ok := target.messageThreadIDForSend(); ok { params.MessageThreadID = threadID } + if replyParams, ok := telegramReplyParameters(msg.ReplyToMessageID); ok { + params.ReplyParameters = replyParams + } _, err = c.bot.SendAudio(ctx, params) case "video": params := &telego.SendVideoParams{ @@ -391,6 +447,9 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe if threadID, ok := target.messageThreadIDForSend(); ok { params.MessageThreadID = threadID } + if replyParams, ok := telegramReplyParameters(msg.ReplyToMessageID); ok { + params.ReplyParameters = replyParams + } _, err = c.bot.SendVideo(ctx, params) default: // "file" or unknown types params := &telego.SendDocumentParams{ @@ -401,6 +460,9 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe if threadID, ok := target.messageThreadIDForSend(); ok { params.MessageThreadID = threadID } + if replyParams, ok := telegramReplyParameters(msg.ReplyToMessageID); ok { + params.ReplyParameters = replyParams + } _, err = c.bot.SendDocument(ctx, params) } @@ -575,6 +637,9 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes "is_forum": fmt.Sprintf("%t", message.Chat.IsForum), "chat_id": fmt.Sprintf("%d", chatID), } + if message.ReplyToMessage != nil { + metadata["reply_to_message_id"] = strconv.Itoa(message.ReplyToMessage.MessageID) + } if hasTopic { metadata["thread_id"] = strconv.Itoa(threadID) metadata["parent_peer_kind"] = "group" diff --git a/pkg/channels/telegram/telegram_dispatch_test.go b/pkg/channels/telegram/telegram_dispatch_test.go index 963eb146b..dcfc165c8 100644 --- a/pkg/channels/telegram/telegram_dispatch_test.go +++ b/pkg/channels/telegram/telegram_dispatch_test.go @@ -162,3 +162,43 @@ func TestHandleMessage_NonForumGroup_IgnoresThreadID(t *testing.T) { t.Fatalf("unexpected thread_id metadata=%q", inbound.Metadata["thread_id"]) } } + +func TestHandleMessage_CapturesReplyToMessageID(t *testing.T) { + messageBus := bus.NewMessageBus() + ch := &TelegramChannel{ + BaseChannel: channels.NewBaseChannel("telegram", nil, messageBus, nil), + chatIDs: make(map[string]int64), + ctx: context.Background(), + } + + msg := &telego.Message{ + Text: "replying here", + MessageID: 15, + Chat: telego.Chat{ + ID: -1001234567890, + Type: "supergroup", + }, + ReplyToMessage: &telego.Message{ + MessageID: 11, + }, + From: &telego.User{ + ID: 42, + FirstName: "Alice", + }, + } + + if err := ch.handleMessage(context.Background(), msg); err != nil { + t.Fatalf("handleMessage error: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + inbound, ok := messageBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected inbound message to be forwarded") + } + if inbound.Metadata["reply_to_message_id"] != "11" { + t.Fatalf("reply_to_message_id=%q", inbound.Metadata["reply_to_message_id"]) + } +} diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index e65c8eba7..e522b5f4b 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -178,6 +178,30 @@ func TestSendMessageWithID_ForumTopic_UsesThreadID(t *testing.T) { assert.Equal(t, float64(42), body["message_thread_id"]) } +func TestSendMessageWithID_ReplyToMessage_UsesReplyParameters(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + msgID, err := ch.SendMessageWithID(context.Background(), bus.OutboundMessage{ + ChatID: "12345", + Content: "Hello, thread!", + ReplyToMessageID: "99", + }) + + assert.NoError(t, err) + assert.Equal(t, "1", msgID) + require.Len(t, caller.calls, 1) + body := decodeCallBody(t, caller.calls[0]) + replyParams, ok := body["reply_parameters"].(map[string]any) + require.True(t, ok) + assert.Equal(t, float64(99), replyParams["message_id"]) + assert.Equal(t, true, replyParams["allow_sending_without_reply"]) +} + func TestSendMessageWithID_GeneralTopic_OmitsThreadID(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { @@ -339,7 +363,7 @@ func TestStartTyping_GeneralTopic_KeepsThreadID(t *testing.T) { assert.Equal(t, float64(1), body["message_thread_id"]) } -func TestSendPlaceholder_ForumTopic_UsesThreadID(t *testing.T) { +func TestSendPlaceholder_GroupSkipsPlaceholder(t *testing.T) { caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { return successResponse(t), nil @@ -352,10 +376,27 @@ func TestSendPlaceholder_ForumTopic_UsesThreadID(t *testing.T) { msgID, err := ch.SendPlaceholder(context.Background(), "-1001234567890:topic:42") require.NoError(t, err) + assert.Empty(t, msgID) + assert.Empty(t, caller.calls) +} + +func TestSendPlaceholder_PrivateChatSendsMessage(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + ch.config = config.DefaultConfig() + ch.config.Channels.Telegram.Placeholder.Enabled = true + ch.config.Channels.Telegram.Placeholder.Text = "Thinking" + + msgID, err := ch.SendPlaceholder(context.Background(), "12345") + require.NoError(t, err) assert.Equal(t, "1", msgID) require.Len(t, caller.calls, 1) body := decodeCallBody(t, caller.calls[0]) - assert.Equal(t, float64(42), body["message_thread_id"]) + assert.Equal(t, "Thinking", body["text"]) } func TestSendMedia_ForumTopic_UsesThreadID(t *testing.T) { diff --git a/pkg/tools/base.go b/pkg/tools/base.go index f61316667..6f0c6c880 100644 --- a/pkg/tools/base.go +++ b/pkg/tools/base.go @@ -21,9 +21,11 @@ type Tool interface { type toolCtxKey struct{ name string } var ( - ctxKeyChannel = &toolCtxKey{"channel"} - ctxKeyChatID = &toolCtxKey{"chatID"} - ctxKeySessionKey = &toolCtxKey{"sessionKey"} + ctxKeyChannel = &toolCtxKey{"channel"} + ctxKeyChatID = &toolCtxKey{"chatID"} + ctxKeySessionKey = &toolCtxKey{"sessionKey"} + ctxKeyCurrentMessageID = &toolCtxKey{"currentMessageID"} + ctxKeyParentMessageID = &toolCtxKey{"parentMessageID"} ) // WithToolContext returns a child context carrying channel and chatID. @@ -38,6 +40,17 @@ func WithToolSessionKey(ctx context.Context, sessionKey string) context.Context return context.WithValue(ctx, ctxKeySessionKey, sessionKey) } +// WithToolReplyContext returns a child context carrying the current and parent +// inbound platform message IDs for reply routing decisions. +func WithToolReplyContext( + ctx context.Context, + currentMessageID, parentMessageID string, +) context.Context { + ctx = context.WithValue(ctx, ctxKeyCurrentMessageID, currentMessageID) + ctx = context.WithValue(ctx, ctxKeyParentMessageID, parentMessageID) + return ctx +} + // ToolChannel extracts the channel from ctx, or "" if unset. func ToolChannel(ctx context.Context) string { v, _ := ctx.Value(ctxKeyChannel).(string) @@ -56,6 +69,18 @@ func ToolSessionKey(ctx context.Context) string { return v } +// ToolCurrentMessageID extracts the current inbound platform message ID. +func ToolCurrentMessageID(ctx context.Context) string { + v, _ := ctx.Value(ctxKeyCurrentMessageID).(string) + return v +} + +// ToolParentMessageID extracts the parent/replied-to inbound platform message ID. +func ToolParentMessageID(ctx context.Context) string { + v, _ := ctx.Value(ctxKeyParentMessageID).(string) + return v +} + // AsyncCallback is a function type that async tools use to notify completion. // When an async tool finishes its work, it calls this callback with the result. // diff --git a/pkg/tools/message.go b/pkg/tools/message.go index 438ceeddd..6f896fa90 100644 --- a/pkg/tools/message.go +++ b/pkg/tools/message.go @@ -3,10 +3,19 @@ package tools import ( "context" "fmt" + "strings" "sync/atomic" + + "github.com/sipeed/picoclaw/pkg/bus" ) -type SendCallback func(channel, chatID, content string) error +type SendCallback func(msg bus.OutboundMessage) error + +const ( + replyModeChat = "chat" + replyModeCurrent = "current" + replyModeParent = "parent" +) type MessageTool struct { sendCallback SendCallback @@ -22,7 +31,7 @@ func (t *MessageTool) Name() string { } func (t *MessageTool) Description() string { - return "Send a message to user on a chat channel. Use this when you want to communicate something." + return "Send a message to the user on a chat channel. Use this when you want to communicate something or explicitly control reply threading." } func (t *MessageTool) Parameters() map[string]any { @@ -41,6 +50,15 @@ func (t *MessageTool) Parameters() map[string]any { "type": "string", "description": "Optional: target chat/user ID", }, + "reply_mode": map[string]any{ + "type": "string", + "enum": []string{replyModeChat, replyModeCurrent, replyModeParent}, + "description": "Optional: threading mode. chat sends a normal message, current replies to the current inbound message, parent replies to the parent/replied-to inbound message.", + }, + "reply_to_message_id": map[string]any{ + "type": "string", + "description": "Optional: explicit platform message ID to reply to. Overrides reply_mode when provided.", + }, }, "required": []string{"content"}, } @@ -81,11 +99,26 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *ToolRes return &ToolResult{ForLLM: "No target channel/chat specified", IsError: true} } + replyToMessageID, err := resolveReplyTarget(ctx, args) + if err != nil { + return &ToolResult{ + ForLLM: err.Error(), + IsError: true, + Err: err, + } + } + if t.sendCallback == nil { return &ToolResult{ForLLM: "Message sending not configured", IsError: true} } - if err := t.sendCallback(channel, chatID, content); err != nil { + msg := bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: content, + ReplyToMessageID: replyToMessageID, + } + if err := t.sendCallback(msg); err != nil { return &ToolResult{ ForLLM: fmt.Sprintf("sending message: %v", err), IsError: true, @@ -95,8 +128,40 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *ToolRes t.sentInRound.Store(true) // Silent: user already received the message directly + status := fmt.Sprintf("Message sent to %s:%s", channel, chatID) + if replyToMessageID != "" { + status = fmt.Sprintf("%s in reply to %s", status, replyToMessageID) + } return &ToolResult{ - ForLLM: fmt.Sprintf("Message sent to %s:%s", channel, chatID), + ForLLM: status, Silent: true, } } + +func resolveReplyTarget(ctx context.Context, args map[string]any) (string, error) { + replyToMessageID, _ := args["reply_to_message_id"].(string) + replyToMessageID = strings.TrimSpace(replyToMessageID) + if replyToMessageID != "" { + return replyToMessageID, nil + } + + replyMode, _ := args["reply_mode"].(string) + replyMode = strings.ToLower(strings.TrimSpace(replyMode)) + + switch replyMode { + case "", replyModeChat: + return "", nil + case replyModeCurrent: + if id := strings.TrimSpace(ToolCurrentMessageID(ctx)); id != "" { + return id, nil + } + return "", fmt.Errorf("reply_mode=current requested but current message id is unavailable") + case replyModeParent: + if id := strings.TrimSpace(ToolParentMessageID(ctx)); id != "" { + return id, nil + } + return "", fmt.Errorf("reply_mode=parent requested but parent message id is unavailable") + default: + return "", fmt.Errorf("unsupported reply_mode %q", replyMode) + } +} diff --git a/pkg/tools/message_test.go b/pkg/tools/message_test.go index 05630972e..ccae1e518 100644 --- a/pkg/tools/message_test.go +++ b/pkg/tools/message_test.go @@ -4,16 +4,16 @@ import ( "context" "errors" "testing" + + "github.com/sipeed/picoclaw/pkg/bus" ) func TestMessageTool_Execute_Success(t *testing.T) { tool := NewMessageTool() - var sentChannel, sentChatID, sentContent string - tool.SetSendCallback(func(channel, chatID, content string) error { - sentChannel = channel - sentChatID = chatID - sentContent = content + var sent bus.OutboundMessage + tool.SetSendCallback(func(msg bus.OutboundMessage) error { + sent = msg return nil }) @@ -25,14 +25,17 @@ func TestMessageTool_Execute_Success(t *testing.T) { result := tool.Execute(ctx, args) // Verify message was sent with correct parameters - if sentChannel != "test-channel" { - t.Errorf("Expected channel 'test-channel', got '%s'", sentChannel) + if sent.Channel != "test-channel" { + t.Errorf("Expected channel 'test-channel', got '%s'", sent.Channel) } - if sentChatID != "test-chat-id" { - t.Errorf("Expected chatID 'test-chat-id', got '%s'", sentChatID) + if sent.ChatID != "test-chat-id" { + t.Errorf("Expected chatID 'test-chat-id', got '%s'", sent.ChatID) } - if sentContent != "Hello, world!" { - t.Errorf("Expected content 'Hello, world!', got '%s'", sentContent) + if sent.Content != "Hello, world!" { + t.Errorf("Expected content 'Hello, world!', got '%s'", sent.Content) + } + if sent.ReplyToMessageID != "" { + t.Errorf("Expected no reply target, got '%s'", sent.ReplyToMessageID) } // Verify ToolResult meets US-011 criteria: @@ -60,10 +63,9 @@ func TestMessageTool_Execute_Success(t *testing.T) { func TestMessageTool_Execute_WithCustomChannel(t *testing.T) { tool := NewMessageTool() - var sentChannel, sentChatID string - tool.SetSendCallback(func(channel, chatID, content string) error { - sentChannel = channel - sentChatID = chatID + var sent bus.OutboundMessage + tool.SetSendCallback(func(msg bus.OutboundMessage) error { + sent = msg return nil }) @@ -77,11 +79,11 @@ func TestMessageTool_Execute_WithCustomChannel(t *testing.T) { result := tool.Execute(ctx, args) // Verify custom channel/chatID were used instead of defaults - if sentChannel != "custom-channel" { - t.Errorf("Expected channel 'custom-channel', got '%s'", sentChannel) + if sent.Channel != "custom-channel" { + t.Errorf("Expected channel 'custom-channel', got '%s'", sent.Channel) } - if sentChatID != "custom-chat-id" { - t.Errorf("Expected chatID 'custom-chat-id', got '%s'", sentChatID) + if sent.ChatID != "custom-chat-id" { + t.Errorf("Expected chatID 'custom-chat-id', got '%s'", sent.ChatID) } if !result.Silent { @@ -96,7 +98,7 @@ func TestMessageTool_Execute_SendFailure(t *testing.T) { tool := NewMessageTool() sendErr := errors.New("network error") - tool.SetSendCallback(func(channel, chatID, content string) error { + tool.SetSendCallback(func(msg bus.OutboundMessage) error { return sendErr }) @@ -149,7 +151,7 @@ func TestMessageTool_Execute_NoTargetChannel(t *testing.T) { tool := NewMessageTool() // No WithToolContext — channel/chatID are empty - tool.SetSendCallback(func(channel, chatID, content string) error { + tool.SetSendCallback(func(msg bus.OutboundMessage) error { return nil }) @@ -189,6 +191,86 @@ func TestMessageTool_Execute_NotConfigured(t *testing.T) { } } +func TestMessageTool_Execute_ReplyToCurrent(t *testing.T) { + tool := NewMessageTool() + + var sent bus.OutboundMessage + tool.SetSendCallback(func(msg bus.OutboundMessage) error { + sent = msg + return nil + }) + + ctx := WithToolReplyContext( + WithToolContext(context.Background(), "telegram", "chat-1"), + "910", + "905", + ) + result := tool.Execute(ctx, map[string]any{ + "content": "Threaded answer", + "reply_mode": "current", + }) + + if result.IsError { + t.Fatalf("expected success, got error %q", result.ForLLM) + } + if sent.ReplyToMessageID != "910" { + t.Fatalf("reply_to_message_id=%q, want %q", sent.ReplyToMessageID, "910") + } + if result.ForLLM != "Message sent to telegram:chat-1 in reply to 910" { + t.Fatalf("ForLLM=%q", result.ForLLM) + } +} + +func TestMessageTool_Execute_ReplyToParentRequiresParentID(t *testing.T) { + tool := NewMessageTool() + tool.SetSendCallback(func(msg bus.OutboundMessage) error { return nil }) + + ctx := WithToolReplyContext( + WithToolContext(context.Background(), "telegram", "chat-1"), + "910", + "", + ) + result := tool.Execute(ctx, map[string]any{ + "content": "Reply upward", + "reply_mode": "parent", + }) + + if !result.IsError { + t.Fatal("expected error when parent message id is unavailable") + } + if result.ForLLM != "reply_mode=parent requested but parent message id is unavailable" { + t.Fatalf("ForLLM=%q", result.ForLLM) + } +} + +func TestMessageTool_Execute_ExplicitReplyTargetOverridesMode(t *testing.T) { + tool := NewMessageTool() + + var sent bus.OutboundMessage + tool.SetSendCallback(func(msg bus.OutboundMessage) error { + sent = msg + return nil + }) + + ctx := WithToolReplyContext( + WithToolContext(context.Background(), "telegram", "chat-1"), + "910", + "905", + ) + result := tool.Execute(ctx, map[string]any{ + "content": "Specific reply", + "reply_mode": "chat", + "reply_to_message_id": "777", + }) + + if result.IsError { + t.Fatalf("expected success, got error %q", result.ForLLM) + } + if sent.ReplyToMessageID != "777" { + t.Fatalf("reply_to_message_id=%q, want %q", sent.ReplyToMessageID, "777") + } +} + func TestMessageTool_Name(t *testing.T) { tool := NewMessageTool() if tool.Name() != "message" { @@ -251,4 +333,20 @@ func TestMessageTool_Parameters(t *testing.T) { if chatIDProp["type"] != "string" { t.Error("Expected chat_id type to be 'string'") } + + replyModeProp, ok := props["reply_mode"].(map[string]any) + if !ok { + t.Error("Expected 'reply_mode' property") + } + if replyModeProp["type"] != "string" { + t.Error("Expected reply_mode type to be 'string'") + } + + replyToProp, ok := props["reply_to_message_id"].(map[string]any) + if !ok { + t.Error("Expected 'reply_to_message_id' property") + } + if replyToProp["type"] != "string" { + t.Error("Expected reply_to_message_id type to be 'string'") + } } diff --git a/pkg/tools/registry_test.go b/pkg/tools/registry_test.go index 243c1bab5..0f5fe93d8 100644 --- a/pkg/tools/registry_test.go +++ b/pkg/tools/registry_test.go @@ -171,7 +171,8 @@ func TestToolRegistry_ExecuteWithContext_InjectsToolContext(t *testing.T) { } r.Register(ct) - r.ExecuteWithContext(context.Background(), "ctx_tool", nil, "telegram", "chat-42", nil) + ctx := WithToolReplyContext(context.Background(), "910", "905") + r.ExecuteWithContext(ctx, "ctx_tool", nil, "telegram", "chat-42", nil) if ct.lastCtx == nil { t.Fatal("expected Execute to be called") @@ -182,6 +183,12 @@ func TestToolRegistry_ExecuteWithContext_InjectsToolContext(t *testing.T) { if got := ToolChatID(ct.lastCtx); got != "chat-42" { t.Errorf("expected chatID 'chat-42', got %q", got) } + if got := ToolCurrentMessageID(ct.lastCtx); got != "910" { + t.Errorf("expected current message ID '910', got %q", got) + } + if got := ToolParentMessageID(ct.lastCtx); got != "905" { + t.Errorf("expected parent message ID '905', got %q", got) + } } func TestToolRegistry_ExecuteWithContext_EmptyContext(t *testing.T) { From a20ec7fe1ea17acb4e06304c2cff6c80a6667407 Mon Sep 17 00:00:00 2001 From: Dmitrii Balabanov Date: Mon, 9 Mar 2026 17:01:33 +0200 Subject: [PATCH 2/2] Clarify reply routing and add diagnostics --- pkg/agent/context.go | 1 + pkg/agent/context_cache_test.go | 3 ++ pkg/agent/loop.go | 50 ++++++++++++++++++++++++++++++--- pkg/tools/message.go | 46 +++++++++++++++++++++++------- pkg/tools/message_test.go | 20 +++++-------- 5 files changed, 93 insertions(+), 27 deletions(-) diff --git a/pkg/agent/context.go b/pkg/agent/context.go index 665496781..0d44d451c 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -463,6 +463,7 @@ func buildReplyRoutingContext(channel string, replyCtx *ReplyContextInfo) string "- `[[reply:parent]]` replies to the parent/replied-to message when there is one\n"+ "- `[[reply:message_id=123]]` replies to a specific known message ID\n\n"+ "After the directive, add a blank line and then the user-visible message.\n"+ + "Do not use the `message` tool for the normal reply in this chat; use the final answer plus a directive when you need reply routing.\n"+ "If you do not need special routing, answer normally without a directive.\n"+ "Never mention the directive in the visible message body.", replyCtx.CurrentMessageID, diff --git a/pkg/agent/context_cache_test.go b/pkg/agent/context_cache_test.go index d2e605ca0..5a15efa9a 100644 --- a/pkg/agent/context_cache_test.go +++ b/pkg/agent/context_cache_test.go @@ -156,6 +156,9 @@ func TestBuildMessages_TelegramReplyRoutingContext(t *testing.T) { if !strings.Contains(sys, "[[reply:current]]") { t.Fatal("system prompt missing final reply directive guidance") } + if !strings.Contains(sys, "Do not use the `message` tool for the normal reply in this chat") { + t.Fatal("system prompt missing guidance to avoid message tool for normal replies") + } } // TestMtimeAutoInvalidation verifies that the cache detects source file changes diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f2b9c1131..da41ca095 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -377,15 +377,21 @@ func (al *AgentLoop) Run(ctx context.Context) error { al.bus.PublishOutbound(ctx, response.outboundMessage(msg.Channel, msg.ChatID)) logger.InfoCF("agent", "Published outbound response", map[string]any{ - "channel": msg.Channel, - "chat_id": msg.ChatID, - "content_len": len(response.Content), + "channel": msg.Channel, + "chat_id": msg.ChatID, + "content_len": len(response.Content), + "reply_to_message_id": response.ReplyToMessageID, }) } else { logger.DebugCF( "agent", "Skipped outbound (message tool already sent)", - map[string]any{"channel": msg.Channel}, + map[string]any{ + "channel": msg.Channel, + "chat_id": msg.ChatID, + "content_len": len(response.Content), + "reply_to_message_id": response.ReplyToMessageID, + }, ) } } @@ -882,6 +888,42 @@ func resolveFinalResponse( rawContent string, ) agentResponse { content, replyToMessageID := parseFinalReplyDirective(channel, replyCtx, rawContent) + if channel == "telegram" { + firstLine, _, _ := strings.Cut(rawContent, "\n") + directive := strings.TrimSpace(firstLine) + hasDirective := strings.HasPrefix(directive, "[[reply:") && strings.HasSuffix(directive, "]]") + directiveMode := "" + if hasDirective { + directiveMode = strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(directive, "[[reply:"), "]]")) + } + directiveStatus := "none" + switch { + case !hasDirective: + directiveStatus = "none" + case directiveMode == "chat": + directiveStatus = "applied_chat" + case replyToMessageID != "": + directiveStatus = "applied_reply" + default: + directiveStatus = "dropped" + } + + fields := map[string]any{ + "directive_status": directiveStatus, + "reply_to_message_id": replyToMessageID, + "raw_content_len": len(rawContent), + "final_content_len": len(content), + } + if hasDirective { + fields["directive"] = directive + fields["directive_mode"] = directiveMode + } + if replyCtx != nil { + fields["current_message_id"] = strings.TrimSpace(replyCtx.CurrentMessageID) + fields["parent_message_id"] = strings.TrimSpace(replyCtx.ParentMessageID) + } + logger.DebugCF("agent", "Resolved final reply routing", fields) + } return agentResponse{ Content: content, ReplyToMessageID: replyToMessageID, diff --git a/pkg/tools/message.go b/pkg/tools/message.go index 6f896fa90..98da6ba06 100644 --- a/pkg/tools/message.go +++ b/pkg/tools/message.go @@ -7,6 +7,7 @@ import ( "sync/atomic" "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/logger" ) type SendCallback func(msg bus.OutboundMessage) error @@ -31,7 +32,7 @@ func (t *MessageTool) Name() string { } func (t *MessageTool) Description() string { - return "Send a message to the user on a chat channel. Use this when you want to communicate something or explicitly control reply threading." + return "Send an out-of-band message to a chat channel. Do not use this for the normal final reply in the current conversation." } func (t *MessageTool) Parameters() map[string]any { @@ -50,15 +51,6 @@ func (t *MessageTool) Parameters() map[string]any { "type": "string", "description": "Optional: target chat/user ID", }, - "reply_mode": map[string]any{ - "type": "string", - "enum": []string{replyModeChat, replyModeCurrent, replyModeParent}, - "description": "Optional: threading mode. chat sends a normal message, current replies to the current inbound message, parent replies to the parent/replied-to inbound message.", - }, - "reply_to_message_id": map[string]any{ - "type": "string", - "description": "Optional: explicit platform message ID to reply to. Overrides reply_mode when provided.", - }, }, "required": []string{"content"}, } @@ -99,6 +91,32 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *ToolRes return &ToolResult{ForLLM: "No target channel/chat specified", IsError: true} } + currentChannel := ToolChannel(ctx) + currentChatID := ToolChatID(ctx) + replyMode, _ := args["reply_mode"].(string) + replyMode = strings.ToLower(strings.TrimSpace(replyMode)) + explicitReplyTo, _ := args["reply_to_message_id"].(string) + explicitReplyTo = strings.TrimSpace(explicitReplyTo) + + if replyMode != "" || explicitReplyTo != "" { + logger.WarnCF("tool", "Message tool received deprecated reply routing args", map[string]any{ + "channel": channel, + "chat_id": chatID, + "reply_mode": replyMode, + "reply_to_message_id": explicitReplyTo, + }) + } + if currentChannel != "" && currentChatID != "" && channel == currentChannel && chatID == currentChatID { + logger.InfoCF("tool", "Message tool targeting current conversation", map[string]any{ + "channel": channel, + "chat_id": chatID, + "content_len": len(content), + "reply_mode": replyMode, + "same_target": true, + "session_key": ToolSessionKey(ctx), + }) + } + replyToMessageID, err := resolveReplyTarget(ctx, args) if err != nil { return &ToolResult{ @@ -127,6 +145,14 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *ToolRes } t.sentInRound.Store(true) + logger.InfoCF("tool", "Message tool sent outbound message", map[string]any{ + "channel": channel, + "chat_id": chatID, + "content_len": len(content), + "reply_to_message_id": replyToMessageID, + "same_target": currentChannel != "" && currentChatID != "" && channel == currentChannel && chatID == currentChatID, + }) + // Silent: user already received the message directly status := fmt.Sprintf("Message sent to %s:%s", channel, chatID) if replyToMessageID != "" { diff --git a/pkg/tools/message_test.go b/pkg/tools/message_test.go index ccae1e518..6e19f3345 100644 --- a/pkg/tools/message_test.go +++ b/pkg/tools/message_test.go @@ -284,6 +284,9 @@ func TestMessageTool_Description(t *testing.T) { if desc == "" { t.Error("Description should not be empty") } + if desc == "Send a message to the user on a chat channel. Use this when you want to communicate something or explicitly control reply threading." { + t.Fatal("description still advertises reply threading") + } } func TestMessageTool_Parameters(t *testing.T) { @@ -334,19 +337,10 @@ func TestMessageTool_Parameters(t *testing.T) { t.Error("Expected chat_id type to be 'string'") } - replyModeProp, ok := props["reply_mode"].(map[string]any) - if !ok { - t.Error("Expected 'reply_mode' property") + if _, ok := props["reply_mode"]; ok { + t.Error("Did not expect 'reply_mode' property in advertised schema") } - if replyModeProp["type"] != "string" { - t.Error("Expected reply_mode type to be 'string'") - } - - replyToProp, ok := props["reply_to_message_id"].(map[string]any) - if !ok { - t.Error("Expected 'reply_to_message_id' property") - } - if replyToProp["type"] != "string" { - t.Error("Expected reply_to_message_id type to be 'string'") + if _, ok := props["reply_to_message_id"]; ok { + t.Error("Did not expect 'reply_to_message_id' property in advertised schema") } }