From 8647d56b3dd345c8c3cc63e1bb35c436b3f45855 Mon Sep 17 00:00:00 2001 From: Dmitrii Balabanov Date: Mon, 23 Mar 2026 10:01:45 +0200 Subject: [PATCH] feat(session): preserve threaded message history metadata --- pkg/agent/context.go | 35 ++- pkg/agent/eventbus_test.go | 12 +- pkg/agent/hook_process_test.go | 12 +- pkg/agent/hooks_test.go | 12 +- pkg/agent/loop.go | 301 ++++++++++++++++++------- pkg/agent/loop_test.go | 83 ++++++- pkg/agent/steering.go | 9 +- pkg/bus/bus.go | 10 + pkg/bus/types.go | 13 +- pkg/channels/discord/discord.go | 51 +++-- pkg/channels/interfaces.go | 6 + pkg/channels/manager.go | 129 ++++++----- pkg/channels/manager_test.go | 141 +++++++++++- pkg/channels/qq/qq.go | 24 +- pkg/channels/slack/slack.go | 16 +- pkg/channels/telegram/telegram.go | 45 ++-- pkg/channels/telegram/telegram_test.go | 28 ++- pkg/providers/protocoltypes/types.go | 12 + pkg/providers/types.go | 1 + 19 files changed, 726 insertions(+), 214 deletions(-) diff --git a/pkg/agent/context.go b/pkg/agent/context.go index c3fcc9fff..d03e488b6 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -607,8 +607,15 @@ func (cb *ContextBuilder) BuildMessages( SystemParts: contentBlocks, }) - // Add conversation history - messages = append(messages, history...) + // Add conversation history, annotating messages that have threading IDs + // so the LLM can navigate thread structure from persisted sessions. + for _, msg := range history { + annotated := msg + if prefix := messageThreadAnnotation(msg); prefix != "" { + annotated.Content = prefix + msg.Content + } + messages = append(messages, annotated) + } // Add current user message if strings.TrimSpace(currentMessage) != "" { @@ -857,3 +864,27 @@ func (cb *ContextBuilder) GetSkillsInfo() map[string]any { "names": skillNames, } } + +// 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 { + msgIDs := msg.MessageIDs + formattedIDs := strings.Join(msgIDs, ",#") + if formattedIDs != "" { + formattedIDs = "#" + formattedIDs + } + switch { + case len(msgIDs) > 1 && msg.ReplyToMessageID != "": + return fmt.Sprintf("[msgs:%s, reply_to:#%s] ", formattedIDs, msg.ReplyToMessageID) + case len(msgIDs) > 1: + return fmt.Sprintf("[msgs:%s] ", formattedIDs) + case len(msgIDs) == 1 && msg.ReplyToMessageID != "": + return fmt.Sprintf("[msg:%s, reply_to:#%s] ", formattedIDs, msg.ReplyToMessageID) + case len(msgIDs) == 1: + return fmt.Sprintf("[msg:%s] ", formattedIDs) + case msg.ReplyToMessageID != "": + return fmt.Sprintf("[reply_to:#%s] ", msg.ReplyToMessageID) + default: + return "" + } +} diff --git a/pkg/agent/eventbus_test.go b/pkg/agent/eventbus_test.go index 19a1ea9eb..6c458b648 100644 --- a/pkg/agent/eventbus_test.go +++ b/pkg/agent/eventbus_test.go @@ -140,8 +140,8 @@ func TestAgentLoop_EmitsMinimalTurnEvents(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if response != "done" { - t.Fatalf("expected final response 'done', got %q", response) + if response.Content != "done" { + t.Fatalf("expected final response 'done', got %q", response.Content) } events := collectEventStream(sub.C) @@ -396,8 +396,8 @@ func TestAgentLoop_EmitsContextCompressEventOnRetry(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if resp != "Recovered from context error" { - t.Fatalf("expected retry success, got %q", resp) + if resp.Content != "Recovered from context error" { + t.Fatalf("expected retry success, got %q", resp.Content) } events := collectEventStream(sub.C) @@ -551,8 +551,8 @@ func TestAgentLoop_EmitsFollowUpQueuedEvent(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if resp != "async launched" { - t.Fatalf("expected final response 'async launched', got %q", resp) + if resp.Content != "async launched" { + t.Fatalf("expected final response 'async launched', got %q", resp.Content) } select { diff --git a/pkg/agent/hook_process_test.go b/pkg/agent/hook_process_test.go index 50f89811f..829c7f899 100644 --- a/pkg/agent/hook_process_test.go +++ b/pkg/agent/hook_process_test.go @@ -52,8 +52,8 @@ func TestAgentLoop_MountProcessHook_LLMAndObserver(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if resp != "provider content|ipc" { - t.Fatalf("expected process-hooked llm content, got %q", resp) + if resp.Content != "provider content|ipc" { + t.Fatalf("expected process-hooked llm content, got %q", resp.Content) } provider.mu.Lock() @@ -92,8 +92,8 @@ func TestAgentLoop_MountProcessHook_ToolRewrite(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if resp != "ipc:ipc" { - t.Fatalf("expected rewritten process-hook tool result, got %q", resp) + if resp.Content != "ipc:ipc" { + t.Fatalf("expected rewritten process-hook tool result, got %q", resp.Content) } } @@ -160,8 +160,8 @@ func TestAgentLoop_MountProcessHook_ApprovalDeny(t *testing.T) { } expected := "Tool execution denied by approval hook: blocked by ipc hook" - if resp != expected { - t.Fatalf("expected %q, got %q", expected, resp) + if resp.Content != expected { + t.Fatalf("expected %q, got %q", expected, resp.Content) } events := collectEventStream(sub.C) diff --git a/pkg/agent/hooks_test.go b/pkg/agent/hooks_test.go index 49e1b1784..9bb2126b4 100644 --- a/pkg/agent/hooks_test.go +++ b/pkg/agent/hooks_test.go @@ -159,8 +159,8 @@ func TestAgentLoop_Hooks_ObserverAndLLMInterceptor(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if resp != "hooked content" { - t.Fatalf("expected hooked content, got %q", resp) + if resp.Content != "hooked content" { + t.Fatalf("expected hooked content, got %q", resp.Content) } provider.mu.Lock() @@ -286,8 +286,8 @@ func TestAgentLoop_Hooks_ToolInterceptorCanRewrite(t *testing.T) { if err != nil { t.Fatalf("runAgentLoop failed: %v", err) } - if resp != "after:modified" { - t.Fatalf("expected rewritten tool result, got %q", resp) + if resp.Content != "after:modified" { + t.Fatalf("expected rewritten tool result, got %q", resp.Content) } } @@ -326,8 +326,8 @@ func TestAgentLoop_Hooks_ToolApproverCanDeny(t *testing.T) { t.Fatalf("runAgentLoop failed: %v", err) } expected := "Tool execution denied by approval hook: blocked" - if resp != expected { - t.Fatalf("expected %q, got %q", expected, resp) + if resp.Content != expected { + t.Fatalf("expected %q, got %q", expected, resp.Content) } events := collectEventStream(sub.C) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 81b979490..0b7102c11 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -72,22 +72,25 @@ type AgentLoop struct { // processOptions configures how a message is processed type processOptions struct { - SessionKey string // Session identifier for history/context - Channel string // Target channel for tool execution - ChatID string // Target chat ID for tool execution - SenderID string // Current sender ID for dynamic context - SenderDisplayName string // Current sender display name for dynamic context - UserMessage string // User message content (may include prefix) - ForcedSkills []string // Skills explicitly requested for this message - SystemPromptOverride string // Override the default system prompt (Used by SubTurns) - Media []string // media:// refs from inbound message - InitialSteeringMessages []providers.Message // Steering messages from refactor/agent - DefaultResponse string // Response when LLM returns empty - EnableSummary bool // Whether to trigger summarization - SendResponse bool // Whether to send response via bus - SuppressToolFeedback bool // Whether to suppress inline tool feedback messages - NoHistory bool // If true, don't load session history (for heartbeat) - SkipInitialSteeringPoll bool // If true, skip the steering poll at loop start (used by Continue) + SessionKey string // Session identifier for history/context + Channel string // Target channel for tool execution + ChatID string // Target chat ID for tool execution + SenderID string // Current sender ID for dynamic context + SenderDisplayName string // Current sender display name for dynamic context + UserMessage string // User message content (may include prefix) + ForcedSkills []string // Skills explicitly requested for this message + SystemPromptOverride string // Override the default system prompt (Used by SubTurns) + Media []string // media:// refs from inbound message + InitialSteeringMessages []providers.Message // Steering messages from refactor/agent + DefaultResponse string // Response when LLM returns empty + EnableSummary bool // Whether to trigger summarization + SendResponse bool // Whether to send response via bus + SuppressToolFeedback bool // Whether to suppress inline tool feedback messages + NoHistory bool // If true, don't load session history (for heartbeat) + SkipInitialSteeringPoll bool // If true, skip the steering poll at loop start (used by Continue) + MessageID string // Inbound platform message ID (for threading) + ReplyToMessageID string // Parent message ID from inbound (for threading) + Sender *providers.MessageSender // Author identity (nil for system/automated messages) } type continuationTarget struct { @@ -96,16 +99,57 @@ type continuationTarget struct { ChatID string } +type agentResponse struct { + Content string + Channel string + ChatID string + OnDelivered func(msgIDs []string) +} + +func (r agentResponse) outboundMessage(defaultChannel, defaultChatID string) bus.OutboundMessage { + channel := r.Channel + if channel == "" { + channel = defaultChannel + } + chatID := r.ChatID + if chatID == "" { + chatID = defaultChatID + } + return bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: r.Content, + OnDelivered: r.OnDelivered, + } +} + +func singleMessageIDs(msgID string) []string { + if msgID == "" { + return nil + } + return []string{msgID} +} + +func cloneMessageIDs(msgIDs []string) []string { + if len(msgIDs) == 0 { + return nil + } + cloned := make([]string, len(msgIDs)) + copy(cloned, msgIDs) + return cloned +} + const ( - defaultResponse = "The model returned an empty response. This may indicate a provider error or token limit." - toolLimitResponse = "I've reached `max_tool_iterations` without a final response. Increase `max_tool_iterations` in config.json if this task needs more tool steps." + defaultResponse = "The model returned an empty response. This may indicate a provider error or token limit." + toolLimitResponse = "I've reached `max_tool_iterations` without a final response. Increase `max_tool_iterations` in config.json if this task needs more tool steps." handledToolResponseSummary = "Requested output delivered via tool attachment." - sessionKeyAgentPrefix = "agent:" - metadataKeyAccountID = "account_id" - metadataKeyGuildID = "guild_id" - metadataKeyTeamID = "team_id" - metadataKeyParentPeerKind = "parent_peer_kind" - metadataKeyParentPeerID = "parent_peer_id" + sessionKeyAgentPrefix = "agent:" + metadataKeyAccountID = "account_id" + metadataKeyGuildID = "guild_id" + metadataKeyTeamID = "team_id" + metadataKeyParentPeerKind = "parent_peer_kind" + metadataKeyParentPeerID = "parent_peer_id" + metadataKeyReplyToMessage = "reply_to_message_id" ) func NewAgentLoop( @@ -444,9 +488,13 @@ 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), + Channel: msg.Channel, + ChatID: msg.ChatID, + } } - finalResponse := response + finalResponse := response.Content target, targetErr := al.buildContinuationTarget(msg) if targetErr != nil { @@ -459,12 +507,20 @@ func (al *AgentLoop) Run(ctx context.Context) error { } if target == nil { cancelDrain() - if finalResponse != "" { - al.publishResponseIfNeeded(ctx, msg.Channel, msg.ChatID, finalResponse) + if response.Content != "" { + al.publishAgentResponseIfNeeded(ctx, response, msg.Channel, msg.ChatID) } return } + responsePersisted := false + continuedOnce := false + if al.pendingSteeringCountForScope(target.SessionKey) > 0 && + response.Content != "" && response.OnDelivered != nil { + response.OnDelivered(nil) + responsePersisted = true + } + for al.pendingSteeringCountForScope(target.SessionKey) > 0 { logger.InfoCF("agent", "Continuing queued steering after turn end", map[string]any{ @@ -489,10 +545,17 @@ func (al *AgentLoop) Run(ctx context.Context) error { } finalResponse = continued + continuedOnce = true } cancelDrain() + if al.pendingSteeringCountForScope(target.SessionKey) > 0 && + !responsePersisted && response.Content != "" && response.OnDelivered != nil { + response.OnDelivered(nil) + responsePersisted = true + } + for al.pendingSteeringCountForScope(target.SessionKey) > 0 { logger.InfoCF("agent", "Draining steering queued during turn shutdown", map[string]any{ @@ -517,10 +580,15 @@ func (al *AgentLoop) Run(ctx context.Context) error { } finalResponse = continued + continuedOnce = true } if finalResponse != "" { - al.publishResponseIfNeeded(ctx, target.Channel, target.ChatID, finalResponse) + if continuedOnce { + al.publishResponseIfNeeded(ctx, target.Channel, target.ChatID, finalResponse) + } else { + al.publishAgentResponseIfNeeded(ctx, response, target.Channel, target.ChatID) + } } }() default: @@ -643,6 +711,47 @@ func (al *AgentLoop) publishResponseIfNeeded(ctx context.Context, channel, chatI }) } +func (al *AgentLoop) publishAgentResponseIfNeeded( + ctx context.Context, + response agentResponse, + defaultChannel, defaultChatID string, +) { + if response.Content == "" { + return + } + + alreadySent := false + defaultAgent := al.GetRegistry().GetDefaultAgent() + if defaultAgent != nil { + if tool, ok := defaultAgent.Tools.Get("message"); ok { + if mt, ok := tool.(*tools.MessageTool); ok { + alreadySent = mt.HasSentInRound() + } + } + } + + if alreadySent { + if response.OnDelivered != nil { + response.OnDelivered(nil) + } + logger.DebugCF( + "agent", + "Skipped outbound (message tool already sent)", + map[string]any{"channel": response.outboundMessage(defaultChannel, defaultChatID).Channel}, + ) + return + } + + outbound := response.outboundMessage(defaultChannel, defaultChatID) + al.bus.PublishOutbound(ctx, outbound) + logger.InfoCF("agent", "Published outbound response", + map[string]any{ + "channel": outbound.Channel, + "chat_id": outbound.ChatID, + "content_len": len(response.Content), + }) +} + func (al *AgentLoop) buildContinuationTarget(msg bus.InboundMessage) (*continuationTarget, error) { if msg.Channel == "system" { return nil, nil @@ -1222,7 +1331,14 @@ func (al *AgentLoop) ProcessDirectWithChannel( SessionKey: sessionKey, } - return al.processMessage(ctx, msg) + response, err := al.processMessage(ctx, msg) + if err != nil { + return "", err + } + if response.OnDelivered != nil { + response.OnDelivered(nil) + } + return response.Content, nil } // ProcessHeartbeat processes a heartbeat request without session history. @@ -1242,7 +1358,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, @@ -1253,9 +1369,16 @@ func (al *AgentLoop) ProcessHeartbeat( SuppressToolFeedback: true, NoHistory: true, // Don't load session history for heartbeat }) + if err != nil { + return "", err + } + if response.OnDelivered != nil { + response.OnDelivered(nil) + } + return response.Content, nil } -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") { @@ -1290,7 +1413,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) route, agent, routeErr := al.resolveMessageRoute(msg) if routeErr != nil { - return "", routeErr + return agentResponse{}, routeErr } // Reset message-tool state for this round so we don't skip publishing due to a previous round. @@ -1325,12 +1448,19 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) DefaultResponse: defaultResponse, EnableSummary: true, SendResponse: false, + MessageID: msg.MessageID, + ReplyToMessageID: inboundMetadata(msg, metadataKeyReplyToMessage), + Sender: messageSenderFromInbound(msg.Sender), } // 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, + Channel: opts.Channel, + ChatID: opts.ChatID, + }, nil } if pending := al.takePendingSkills(opts.SessionKey); len(pending) > 0 { @@ -1403,9 +1533,9 @@ func (al *AgentLoop) requeueInboundMessage(msg bus.InboundMessage) error { 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, ) @@ -1442,13 +1572,13 @@ func (al *AgentLoop) processSystemMessage( "content_len": len(content), "channel": originChannel, }) - return "", nil + return agentResponse{}, nil } // Use default agent for system messages agent := al.GetRegistry().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 @@ -1471,7 +1601,7 @@ func (al *AgentLoop) runAgentLoop( ctx context.Context, agent *AgentInstance, opts processOptions, -) (string, error) { +) (agentResponse, error) { // Record last channel for heartbeat notifications (skip internal channels and cli) if opts.Channel != "" && opts.ChatID != "" && !constants.IsInternalChannel(opts.Channel) { channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID) @@ -1487,10 +1617,10 @@ func (al *AgentLoop) runAgentLoop( ts := newTurnState(agent, opts, al.newTurnEventScope(agent.ID, opts.SessionKey)) result, err := al.runTurn(ctx, ts) if err != nil { - return "", err + return agentResponse{}, err } if result.status == TurnEndStatusAborted { - return "", nil + return agentResponse{}, nil } for _, followUp := range result.followUps { @@ -1503,12 +1633,32 @@ func (al *AgentLoop) runAgentLoop( } } - if opts.SendResponse && result.finalContent != "" { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: result.finalContent, - }) + response := agentResponse{ + Content: result.finalContent, + Channel: opts.Channel, + ChatID: opts.ChatID, + } + + if !opts.NoHistory && result.finalContent != "" { + response.OnDelivered = func(msgIDs []string) { + assistantMsg := providers.Message{ + Role: "assistant", + Content: result.finalContent, + MessageIDs: cloneMessageIDs(msgIDs), + } + agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg) + if saveErr := agent.Sessions.Save(opts.SessionKey); saveErr != nil { + logger.WarnCF("agent", "Failed to save delivered assistant message", + map[string]any{ + "session_key": opts.SessionKey, + "error": saveErr.Error(), + }) + return + } + if opts.EnableSummary { + al.maybeSummarize(agent, opts.SessionKey, ts.scope) + } + } } if result.finalContent != "" { @@ -1522,7 +1672,7 @@ func (al *AgentLoop) runAgentLoop( }) } - return result.finalContent, nil + return response, nil } func (al *AgentLoop) targetReasoningChannelID(channelName string) (chatID string) { @@ -1674,15 +1824,14 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er // Save user message to session (from Incoming) if !ts.opts.NoHistory && (strings.TrimSpace(ts.userMessage) != "" || len(ts.media) > 0) { rootMsg := providers.Message{ - Role: "user", - Content: ts.userMessage, - Media: append([]string(nil), ts.media...), - } - if len(rootMsg.Media) > 0 { - ts.agent.Sessions.AddFullMessage(ts.sessionKey, rootMsg) - } else { - ts.agent.Sessions.AddMessage(ts.sessionKey, rootMsg.Role, rootMsg.Content) + Role: "user", + Content: ts.userMessage, + Media: append([]string(nil), ts.media...), + MessageIDs: singleMessageIDs(ts.opts.MessageID), + ReplyToMessageID: ts.opts.ReplyToMessageID, + Sender: ts.opts.Sender, } + ts.agent.Sessions.AddFullMessage(ts.sessionKey, rootMsg) ts.recordPersistedMessage(rootMsg) } @@ -2676,27 +2825,6 @@ turnLoop: ts.setPhase(TurnPhaseFinalizing) ts.setFinalContent(finalContent) - if !ts.opts.NoHistory { - finalMsg := providers.Message{Role: "assistant", Content: finalContent} - ts.agent.Sessions.AddMessage(ts.sessionKey, finalMsg.Role, finalMsg.Content) - ts.recordPersistedMessage(finalMsg) - if err := ts.agent.Sessions.Save(ts.sessionKey); err != nil { - turnStatus = TurnEndStatusError - al.emitEvent( - EventKindError, - ts.eventMeta("runTurn", "turn.error"), - ErrorPayload{ - Stage: "session_save", - Message: err.Error(), - }, - ) - return turnResult{}, err - } - } - - if ts.opts.EnableSummary { - al.maybeSummarize(ts.agent, ts.sessionKey, ts.scope) - } ts.setPhase(TurnPhaseCompleted) return turnResult{ @@ -3577,3 +3705,22 @@ func extractProvider(registry *AgentRegistry) (providers.LLMProvider, bool) { } return defaultAgent.Provider, true } + +// messageSenderFromInbound converts bus.SenderInfo to providers.MessageSender. +// Returns nil if no meaningful identity is present. +func messageSenderFromInbound(s bus.SenderInfo) *providers.MessageSender { + if s.Username == "" && s.FirstName == "" && s.LastName == "" && s.DisplayName == "" { + return nil + } + username := s.Username + firstName := s.FirstName + lastName := s.LastName + if firstName == "" && lastName == "" && s.DisplayName != "" { + firstName = s.DisplayName + } + return &providers.MessageSender{ + Username: username, + FirstName: firstName, + LastName: lastName, + } +} diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 2366b1277..01ad4c3c1 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -149,8 +149,8 @@ func TestProcessMessage_IncludesCurrentSenderInDynamicContext(t *testing.T) { if err != nil { t.Fatalf("processMessage() error = %v", err) } - if response != "Mock response" { - t.Fatalf("processMessage() response = %q, want %q", response, "Mock response") + if response.Content != "Mock response" { + t.Fatalf("processMessage() response = %q, want %q", response.Content, "Mock response") } if len(provider.lastMessages) == 0 { t.Fatal("provider did not receive any messages") @@ -205,8 +205,8 @@ func TestProcessMessage_UseCommandLoadsRequestedSkill(t *testing.T) { if err != nil { t.Fatalf("processMessage() error = %v", err) } - if response != "Mock response" { - t.Fatalf("processMessage() response = %q, want %q", response, "Mock response") + if response.Content != "Mock response" { + t.Fatalf("processMessage() response = %q, want %q", response.Content, "Mock response") } if len(provider.lastMessages) == 0 { t.Fatal("provider did not receive any messages") @@ -295,8 +295,8 @@ func TestProcessMessage_UseCommandArmsSkillForNextMessage(t *testing.T) { if err != nil { t.Fatalf("processMessage() arm error = %v", err) } - if !strings.Contains(response, `Skill "shell" is armed for your next message.`) { - t.Fatalf("arm response = %q, want armed confirmation", response) + if !strings.Contains(response.Content, `Skill "shell" is armed for your next message.`) { + t.Fatalf("arm response = %q, want armed confirmation", response.Content) } response, err = al.processMessage(context.Background(), bus.InboundMessage{ @@ -308,8 +308,8 @@ func TestProcessMessage_UseCommandArmsSkillForNextMessage(t *testing.T) { if err != nil { t.Fatalf("processMessage() follow-up error = %v", err) } - if response != "Mock response" { - t.Fatalf("follow-up response = %q, want %q", response, "Mock response") + if response.Content != "Mock response" { + t.Fatalf("follow-up response = %q, want %q", response.Content, "Mock response") } if len(provider.lastMessages) == 0 { t.Fatal("provider did not receive any messages") @@ -405,6 +405,68 @@ func TestApplyExplicitSkillCommand_InlineMessageMutatesOptions(t *testing.T) { } } +func TestProcessMessage_AssistantSavedOnDelivered(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-delivery" + response, err := al.processMessage(context.Background(), bus.InboundMessage{ + Channel: "telegram", + SenderID: "telegram:123", + ChatID: "chat-1", + Content: "hello", + SessionKey: sessionKey, + MessageID: "in-42", + }) + 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 response.OnDelivered == nil { + t.Fatal("expected OnDelivered callback") + } + response.OnDelivered([]string{"out-99"}) + + 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-99" { + t.Fatalf("expected assistant message_ids [out-99], got %v", history[1].MessageIDs) + } +} + func TestRecordLastChannel(t *testing.T) { al, cfg, msgBus, provider, cleanup := newTestAgentLoop(t) defer cleanup() @@ -1305,7 +1367,10 @@ func (h testHelper) executeAndGetResponse(tb testing.TB, ctx context.Context, ms if err != nil { tb.Fatalf("processMessage failed: %v", err) } - return response + if response.OnDelivered != nil { + response.OnDelivered(nil) + } + return response.Content } const responseTimeout = 3 * time.Second diff --git a/pkg/agent/steering.go b/pkg/agent/steering.go index ad6613e8c..3f95a6d74 100644 --- a/pkg/agent/steering.go +++ b/pkg/agent/steering.go @@ -292,7 +292,7 @@ func (al *AgentLoop) continueWithSteeringMessages( sessionKey, channel, chatID string, steeringMsgs []providers.Message, ) (string, error) { - return al.runAgentLoop(ctx, agent, processOptions{ + response, err := al.runAgentLoop(ctx, agent, processOptions{ SessionKey: sessionKey, Channel: channel, ChatID: chatID, @@ -302,6 +302,13 @@ func (al *AgentLoop) continueWithSteeringMessages( InitialSteeringMessages: steeringMsgs, SkipInitialSteeringPoll: true, }) + if err != nil { + return "", err + } + if response.OnDelivered != nil { + response.OnDelivered(nil) + } + return response.Content, nil } func (al *AgentLoop) agentForSession(sessionKey string) *AgentInstance { diff --git a/pkg/bus/bus.go b/pkg/bus/bus.go index 37fcb74c5..339fc0550 100644 --- a/pkg/bus/bus.go +++ b/pkg/bus/bus.go @@ -95,6 +95,16 @@ func (mb *MessageBus) OutboundChan() <-chan OutboundMessage { return mb.outbound } +// SubscribeOutbound waits for the next outbound message or until ctx is done. +func (mb *MessageBus) SubscribeOutbound(ctx context.Context) (OutboundMessage, bool) { + select { + case msg, ok := <-mb.outbound: + return msg, ok + case <-ctx.Done(): + return OutboundMessage{}, false + } +} + func (mb *MessageBus) PublishOutboundMedia(ctx context.Context, msg OutboundMediaMessage) error { return publish(ctx, mb, mb.outboundMedia, msg) } diff --git a/pkg/bus/types.go b/pkg/bus/types.go index 12da3f1dd..0366e89df 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -12,7 +12,9 @@ type SenderInfo struct { PlatformID string `json:"platform_id,omitempty"` // raw platform ID, e.g. "123456" CanonicalID string `json:"canonical_id,omitempty"` // "platform:id" format Username string `json:"username,omitempty"` // username (e.g. @alice) - DisplayName string `json:"display_name,omitempty"` // display name + DisplayName string `json:"display_name,omitempty"` // display name (used when first/last are not available) + FirstName string `json:"first_name,omitempty"` // given name (preferred over DisplayName when set) + LastName string `json:"last_name,omitempty"` // family name } type InboundMessage struct { @@ -30,10 +32,11 @@ type InboundMessage struct { } type OutboundMessage struct { - Channel string `json:"channel"` - ChatID string `json:"chat_id"` - Content string `json:"content"` - ReplyToMessageID string `json:"reply_to_message_id,omitempty"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Content string `json:"content"` + ReplyToMessageID string `json:"reply_to_message_id,omitempty"` + OnDelivered func(msgIDs []string) `json:"-"` } // MediaPart describes a single media attachment to send. diff --git a/pkg/channels/discord/discord.go b/pkg/channels/discord/discord.go index 2385544a6..de7e7be8c 100644 --- a/pkg/channels/discord/discord.go +++ b/pkg/channels/discord/discord.go @@ -129,20 +129,30 @@ func (c *DiscordChannel) Stop(ctx context.Context) error { } func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + _, err := c.SendMessageWithIDs(ctx, msg) + return err +} + +// SendMessageWithIDs implements channels.MessageIDsSender. +func (c *DiscordChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } channelID := msg.ChatID if channelID == "" { - return fmt.Errorf("channel ID is empty") + return nil, fmt.Errorf("channel ID is empty") } if len([]rune(msg.Content)) == 0 { - return nil + return nil, nil } - return c.sendChunk(ctx, channelID, msg.Content, msg.ReplyToMessageID) + msgID, err := c.sendChunk(ctx, channelID, msg.Content, msg.ReplyToMessageID) + if err != nil { + return nil, err + } + return []string{msgID}, nil } // SendMedia implements the channels.MediaSender interface. @@ -264,18 +274,25 @@ func (c *DiscordChannel) SendPlaceholder(ctx context.Context, chatID string) (st return msg.ID, nil } -func (c *DiscordChannel) sendChunk(ctx context.Context, channelID, content, replyToID string) error { +func (c *DiscordChannel) sendChunk(ctx context.Context, channelID, content, replyToID string) (string, error) { // Use the passed ctx for timeout control sendCtx, cancel := context.WithTimeout(ctx, sendTimeout) defer cancel() - done := make(chan error, 1) + type sendResult struct { + id string + err error + } + done := make(chan sendResult, 1) go func() { - var err error + var ( + msg *discordgo.Message + err error + ) // If we have an ID, we send the message as "Reply" if replyToID != "" { - _, err = c.session.ChannelMessageSendComplex(channelID, &discordgo.MessageSend{ + msg, err = c.session.ChannelMessageSendComplex(channelID, &discordgo.MessageSend{ Content: content, Reference: &discordgo.MessageReference{ MessageID: replyToID, @@ -284,20 +301,24 @@ func (c *DiscordChannel) sendChunk(ctx context.Context, channelID, content, repl }) } else { // Otherwise, we send a normal message - _, err = c.session.ChannelMessageSend(channelID, content) + msg, err = c.session.ChannelMessageSend(channelID, content) } - done <- err + if err != nil { + done <- sendResult{err: err} + return + } + done <- sendResult{id: msg.ID} }() select { - case err := <-done: - if err != nil { - return fmt.Errorf("discord send: %w", channels.ErrTemporary) + case result := <-done: + if result.err != nil { + return "", fmt.Errorf("discord send: %w", channels.ErrTemporary) } - return nil + return result.id, nil case <-sendCtx.Done(): - return sendCtx.Err() + return "", sendCtx.Err() } } diff --git a/pkg/channels/interfaces.go b/pkg/channels/interfaces.go index 0cfd435b0..e4388664a 100644 --- a/pkg/channels/interfaces.go +++ b/pkg/channels/interfaces.go @@ -62,6 +62,12 @@ type PlaceholderRecorder interface { RecordReactionUndo(channel, chatID string, undo func()) } +// MessageIDsSender is implemented by channels that can return the platform +// message IDs for a delivered outbound text message. +type MessageIDsSender interface { + SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) (messageIDs []string, err error) +} + // CommandRegistrarCapable is implemented by channels that can register // command menus with their upstream platform (e.g. Telegram BotCommand). // Channels that do not support platform-level command menus can ignore it. diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 7bcb933ce..8b2cc5cbd 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -158,8 +158,8 @@ func (m *Manager) RecordReactionUndo(channel, chatID string, undo func()) { } // preSend handles typing stop, reaction undo, and placeholder editing before sending a message. -// Returns true if the message was already delivered (skip Send). -func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMessage, ch Channel) bool { +// Returns the delivered message IDs and true when delivery completed before a normal Send. +func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMessage, ch Channel) ([]string, bool) { key := name + ":" + msg.ChatID // 1. Stop typing @@ -188,7 +188,7 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess } } } - return true + return nil, true } // 4. Try editing placeholder @@ -196,14 +196,14 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess if entry, ok := v.(placeholderEntry); ok && entry.id != "" { 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 + return []string{entry.id}, true } // edit failed → fall through to normal Send } } } - return false + return nil, false } // preSendMedia handles typing stop, reaction undo, and placeholder cleanup @@ -620,40 +620,59 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker) if !ok { return } - maxLen := 0 - if mlp, ok := w.ch.(MessageLengthProvider); ok { - maxLen = mlp.MaxMessageLength() - } - - // Collect all message chunks to send - var chunks []string - - // Step 1: Try marker-based splitting if enabled - if m.config != nil && m.config.Agents.Defaults.SplitOnMarker { - if markerChunks := SplitByMarker(msg.Content); len(markerChunks) > 1 { - for _, chunk := range markerChunks { - chunks = append(chunks, splitByLength(chunk, maxLen)...) - } - } - } - - // Step 2: Fallback to length-based splitting if no chunks from marker - if len(chunks) == 0 { - chunks = splitByLength(msg.Content, maxLen) - } - - // Step 3: Send all chunks - for _, chunk := range chunks { - chunkMsg := msg - chunkMsg.Content = chunk - m.sendWithRetry(ctx, name, w, chunkMsg) - } + m.deliverOutbound(ctx, name, w, msg) case <-ctx.Done(): return } } } +func (m *Manager) deliverOutbound(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMessage) { + msgIDs, delivered := m.sendOutbound(ctx, name, w, msg) + if delivered && msg.OnDelivered != nil { + msg.OnDelivered(msgIDs) + } +} + +func (m *Manager) sendOutbound( + ctx context.Context, + name string, + w *channelWorker, + msg bus.OutboundMessage, +) ([]string, bool) { + maxLen := 0 + if mlp, ok := w.ch.(MessageLengthProvider); ok { + maxLen = mlp.MaxMessageLength() + } + + var chunks []string + if m.config != nil && m.config.Agents.Defaults.SplitOnMarker { + if markerChunks := SplitByMarker(msg.Content); len(markerChunks) > 1 { + for _, chunk := range markerChunks { + chunks = append(chunks, splitByLength(chunk, maxLen)...) + } + } + } + if len(chunks) == 0 { + chunks = splitByLength(msg.Content, maxLen) + } + + var messageIDs []string + for _, chunk := range chunks { + chunkMsg := msg + chunkMsg.Content = chunk + chunkMsg.OnDelivered = nil + chunkIDs, delivered := m.sendWithRetry(ctx, name, w, chunkMsg) + if !delivered { + return nil, false + } + if len(chunkIDs) > 0 { + messageIDs = append(messageIDs, chunkIDs...) + } + } + return messageIDs, true +} + // splitByLength splits content by maxLen if needed, otherwise returns single chunk. func splitByLength(content string, maxLen int) []string { if maxLen > 0 && len([]rune(content)) > maxLen { @@ -667,23 +686,35 @@ func splitByLength(content string, maxLen int) []string { // - ErrNotRunning / ErrSendFailed: permanent, no retry // - ErrRateLimit: fixed delay retry // - ErrTemporary / unknown: exponential backoff retry -func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWorker, msg bus.OutboundMessage) { +func (m *Manager) sendWithRetry( + ctx context.Context, + name string, + w *channelWorker, + msg bus.OutboundMessage, +) ([]string, bool) { // Rate limit: wait for token if err := w.limiter.Wait(ctx); err != nil { // ctx canceled, shutting down - return + return nil, false } // Pre-send: stop typing and try to edit placeholder - if m.preSend(ctx, name, msg, w.ch) { - return // placeholder was edited successfully, skip Send + if msgIDs, handled := m.preSend(ctx, name, msg, w.ch); handled { + return msgIDs, true } var lastErr error + var msgIDs []string + sender, hasMessageIDsSender := w.ch.(MessageIDsSender) for attempt := 0; attempt <= maxRetries; attempt++ { - lastErr = w.ch.Send(ctx, msg) + msgIDs = nil + if hasMessageIDsSender { + msgIDs, lastErr = sender.SendMessageWithIDs(ctx, msg) + } else { + lastErr = w.ch.Send(ctx, msg) + } if lastErr == nil { - return + return msgIDs, true } // Permanent failures — don't retry @@ -702,7 +733,7 @@ func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWork case <-time.After(rateLimitDelay): continue case <-ctx.Done(): - return + return nil, false } } @@ -711,7 +742,7 @@ func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWork select { case <-time.After(backoff): case <-ctx.Done(): - return + return nil, false } } @@ -722,6 +753,8 @@ func (m *Manager) sendWithRetry(ctx context.Context, name string, w *channelWork "error": lastErr.Error(), "retries": maxRetries, }) + + return nil, false } func dispatchLoop[M any]( @@ -1077,19 +1110,7 @@ func (m *Manager) SendMessage(ctx context.Context, msg bus.OutboundMessage) erro return fmt.Errorf("channel %s has no active worker", msg.Channel) } - maxLen := 0 - if mlp, ok := w.ch.(MessageLengthProvider); ok { - maxLen = mlp.MaxMessageLength() - } - if maxLen > 0 && len([]rune(msg.Content)) > maxLen { - for _, chunk := range SplitMessage(msg.Content, maxLen) { - chunkMsg := msg - chunkMsg.Content = chunk - m.sendWithRetry(ctx, msg.Channel, w, chunkMsg) - } - } else { - m.sendWithRetry(ctx, msg.Channel, w, msg) - } + m.deliverOutbound(ctx, msg.Channel, w, msg) return nil } diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index b4fd2ba3d..5b3ee3bf8 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -19,6 +19,7 @@ import ( type mockChannel struct { BaseChannel sendFn func(ctx context.Context, msg bus.OutboundMessage) error + sendWithIDsFn func(ctx context.Context, msg bus.OutboundMessage) ([]string, error) sentMessages []bus.OutboundMessage placeholdersSent int editedMessages int @@ -30,6 +31,17 @@ func (m *mockChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { return m.sendFn(ctx, msg) } +func (m *mockChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { + m.sentMessages = append(m.sentMessages, msg) + if m.sendWithIDsFn == nil { + if m.sendFn == nil { + return nil, nil + } + return nil, m.sendFn(ctx, msg) + } + return m.sendWithIDsFn(ctx, msg) +} + func (m *mockChannel) Start(ctx context.Context) error { return nil } func (m *mockChannel) Stop(ctx context.Context) error { return nil } @@ -137,6 +149,114 @@ func TestSendWithRetry_TemporaryThenSuccess(t *testing.T) { } } +func TestDeliverOutbound_CallsOnDeliveredWithMessageIDs(t *testing.T) { + m := newTestManager() + ch := &mockChannel{ + sendFn: nil, + sendWithIDsFn: func(_ context.Context, _ bus.OutboundMessage) ([]string, error) { + return []string{"msg-123"}, nil + }, + } + w := &channelWorker{ + ch: ch, + limiter: rate.NewLimiter(rate.Inf, 1), + } + + var deliveredIDs []string + msg := bus.OutboundMessage{ + Channel: "test", + ChatID: "1", + Content: "hello", + OnDelivered: func(msgIDs []string) { + deliveredIDs = append([]string(nil), msgIDs...) + }, + } + + m.deliverOutbound(context.Background(), "test", w, msg) + + if len(deliveredIDs) != 1 || deliveredIDs[0] != "msg-123" { + t.Fatalf("expected delivered IDs [msg-123], got %v", deliveredIDs) + } +} + +func TestDeliverOutbound_CallsOnDeliveredWithPlaceholderID(t *testing.T) { + m := newTestManager() + ch := &mockMessageEditor{ + mockChannel: mockChannel{ + sendFn: func(_ context.Context, _ bus.OutboundMessage) error { + t.Fatal("Send should not be called when placeholder edit succeeds") + return nil + }, + }, + editFn: func(_ context.Context, _, _, _ string) error { + return nil + }, + } + w := &channelWorker{ + ch: ch, + limiter: rate.NewLimiter(rate.Inf, 1), + } + + m.RecordPlaceholder("test", "123", "ph-456") + + var deliveredIDs []string + msg := bus.OutboundMessage{ + Channel: "test", + ChatID: "123", + Content: "hello", + OnDelivered: func(msgIDs []string) { + deliveredIDs = append([]string(nil), msgIDs...) + }, + } + + m.deliverOutbound(context.Background(), "test", w, msg) + + if len(deliveredIDs) != 1 || deliveredIDs[0] != "ph-456" { + t.Fatalf("expected delivered IDs [ph-456], got %v", deliveredIDs) + } +} + +func TestDeliverOutbound_CallsOnDeliveredWithAllSplitMessageIDs(t *testing.T) { + m := newTestManager() + callCount := 0 + ch := &mockChannel{ + sendWithIDsFn: func(_ context.Context, msg bus.OutboundMessage) ([]string, error) { + callCount++ + return []string{fmt.Sprintf("id-%d", callCount)}, nil + }, + } + w := &channelWorker{ + ch: ch, + limiter: rate.NewLimiter(rate.Inf, 1), + } + ch.BaseChannel = *NewBaseChannel("test", nil, nil, nil, WithMaxMessageLength(5)) + + var deliveredIDs []string + msg := bus.OutboundMessage{ + Channel: "test", + ChatID: "1", + Content: "hello world", + OnDelivered: func(msgIDs []string) { + deliveredIDs = append([]string(nil), msgIDs...) + }, + } + + m.deliverOutbound(context.Background(), "test", w, msg) + + if len(deliveredIDs) <= 1 { + t.Fatalf("expected multiple delivered IDs for split outbound, got %v", deliveredIDs) + } + if len(deliveredIDs) != callCount { + t.Fatalf("expected %d delivered IDs, got %v", callCount, deliveredIDs) + } + for i, deliveredID := range deliveredIDs { + expected := fmt.Sprintf("id-%d", i+1) + if deliveredID != expected { + t.Fatalf("expected delivered IDs in order, got %v", deliveredIDs) + } + } +} + func TestSendWithRetry_PermanentFailure(t *testing.T) { m := newTestManager() var callCount int @@ -628,11 +748,14 @@ func TestPreSend_PlaceholderEditSuccess(t *testing.T) { m.RecordPlaceholder("test", "123", "456") msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"} - edited := m.preSend(context.Background(), "test", msg, ch) + msgIDs, edited := m.preSend(context.Background(), "test", msg, ch) if !edited { t.Fatal("expected preSend to return true (placeholder edited)") } + if len(msgIDs) != 1 || msgIDs[0] != "456" { + t.Fatalf("expected placeholder IDs [456], got %v", msgIDs) + } if !editCalled { t.Fatal("expected EditMessage to be called") } @@ -658,7 +781,7 @@ func TestPreSend_PlaceholderEditFails_FallsThrough(t *testing.T) { m.RecordPlaceholder("test", "123", "456") msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"} - edited := m.preSend(context.Background(), "test", msg, ch) + _, edited := m.preSend(context.Background(), "test", msg, ch) if edited { t.Fatal("expected preSend to return false when edit fails") @@ -717,7 +840,7 @@ func TestPreSend_TypingStopCalled(t *testing.T) { }) msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"} - m.preSend(context.Background(), "test", msg, ch) + _, _ = m.preSend(context.Background(), "test", msg, ch) if !stopCalled { t.Fatal("expected typing stop func to be called") @@ -734,7 +857,7 @@ func TestPreSend_NoRegisteredState(t *testing.T) { } msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"} - edited := m.preSend(context.Background(), "test", msg, ch) + _, edited := m.preSend(context.Background(), "test", msg, ch) if edited { t.Fatal("expected preSend to return false with no registered state") @@ -764,7 +887,7 @@ func TestPreSend_TypingAndPlaceholder(t *testing.T) { m.RecordPlaceholder("test", "123", "456") msg := bus.OutboundMessage{Channel: "test", ChatID: "123", Content: "hello"} - edited := m.preSend(context.Background(), "test", msg, ch) + msgIDs, edited := m.preSend(context.Background(), "test", msg, ch) if !stopCalled { t.Fatal("expected typing stop to be called") @@ -775,6 +898,9 @@ func TestPreSend_TypingAndPlaceholder(t *testing.T) { if !edited { t.Fatal("expected preSend to return true") } + if len(msgIDs) != 1 || msgIDs[0] != "456" { + t.Fatalf("expected placeholder IDs [456], got %v", msgIDs) + } } func TestRecordPlaceholder_ConcurrentSafe(t *testing.T) { @@ -1025,7 +1151,7 @@ func TestPreSendStillWorksWithWrappedTypes(t *testing.T) { m.RecordPlaceholder("test", "chat1", "ph_id") msg := bus.OutboundMessage{Channel: "test", ChatID: "chat1", Content: "response"} - edited := m.preSend(context.Background(), "test", msg, ch) + msgIDs, edited := m.preSend(context.Background(), "test", msg, ch) if !stopCalled { t.Fatal("expected typing stop to be called via wrapped type") @@ -1036,6 +1162,9 @@ func TestPreSendStillWorksWithWrappedTypes(t *testing.T) { if !edited { t.Fatal("expected preSend to return true") } + if len(msgIDs) != 1 || msgIDs[0] != "ph_id" { + t.Fatalf("expected placeholder IDs [ph_id], got %v", msgIDs) + } } // --- Lazy worker creation tests (Step 6) --- diff --git a/pkg/channels/qq/qq.go b/pkg/channels/qq/qq.go index 4ea71f6df..38808238e 100644 --- a/pkg/channels/qq/qq.go +++ b/pkg/channels/qq/qq.go @@ -201,8 +201,14 @@ func (c *QQChannel) getChatKind(chatID string) string { } func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + _, err := c.SendMessageWithIDs(ctx, msg) + return err +} + +// SendMessageWithIDs implements channels.MessageIDsSender. +func (c *QQChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } chatKind := c.getChatKind(msg.ChatID) @@ -236,11 +242,14 @@ func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { } // Route to group or C2C. - var err error + var ( + sentMsg *dto.Message + err error + ) if chatKind == "group" { - _, err = c.api.PostGroupMessage(ctx, msg.ChatID, msgToCreate) + sentMsg, err = c.api.PostGroupMessage(ctx, msg.ChatID, msgToCreate) } else { - _, err = c.api.PostC2CMessage(ctx, msg.ChatID, msgToCreate) + sentMsg, err = c.api.PostC2CMessage(ctx, msg.ChatID, msgToCreate) } if err != nil { @@ -249,10 +258,13 @@ func (c *QQChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { "chat_kind": chatKind, "error": err.Error(), }) - return fmt.Errorf("qq send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("qq send: %w", channels.ErrTemporary) } - return nil + if sentMsg == nil { + return nil, nil + } + return []string{sentMsg.ID}, nil } // StartTyping implements channels.TypingCapable. diff --git a/pkg/channels/slack/slack.go b/pkg/channels/slack/slack.go index f03283ea4..5e2cecec0 100644 --- a/pkg/channels/slack/slack.go +++ b/pkg/channels/slack/slack.go @@ -109,13 +109,19 @@ func (c *SlackChannel) Stop(ctx context.Context) error { } func (c *SlackChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + _, err := c.SendMessageWithIDs(ctx, msg) + return err +} + +// SendMessageWithIDs implements channels.MessageIDsSender. +func (c *SlackChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } channelID, threadTS := parseSlackChatID(msg.ChatID) if channelID == "" { - return fmt.Errorf("invalid slack chat ID: %s", msg.ChatID) + return nil, fmt.Errorf("invalid slack chat ID: %s", msg.ChatID) } opts := []slack.MsgOption{ @@ -130,9 +136,9 @@ func (c *SlackChannel) Send(ctx context.Context, msg bus.OutboundMessage) error opts = append(opts, slack.MsgOptionTS(threadTS)) } - _, _, err := c.api.PostMessageContext(ctx, channelID, opts...) + _, ts, err := c.api.PostMessageContext(ctx, channelID, opts...) if err != nil { - return fmt.Errorf("slack send: %w", channels.ErrTemporary) + return nil, fmt.Errorf("slack send: %w", channels.ErrTemporary) } if ref, ok := c.pendingAcks.LoadAndDelete(msg.ChatID); ok { @@ -148,7 +154,7 @@ func (c *SlackChannel) Send(ctx context.Context, msg bus.OutboundMessage) error "thread_ts": threadTS, }) - return nil + return []string{ts}, nil } // SendMedia implements the channels.MediaSender interface. diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 5adb40a7e..6ca612cb0 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -169,19 +169,25 @@ func (c *TelegramChannel) Stop(ctx context.Context) error { } func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + _, err := c.SendMessageWithIDs(ctx, msg) + return err +} + +// SendMessageWithIDs implements channels.MessageIDsSender. +func (c *TelegramChannel) SendMessageWithIDs(ctx context.Context, msg bus.OutboundMessage) ([]string, error) { if !c.IsRunning() { - return channels.ErrNotRunning + return nil, channels.ErrNotRunning } useMarkdownV2 := c.config.Channels.Telegram.UseMarkdownV2 chatID, threadID, err := parseTelegramChatID(msg.ChatID) if err != nil { - return fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) + return nil, fmt.Errorf("invalid chat ID %s: %w", msg.ChatID, channels.ErrSendFailed) } if msg.Content == "" { - return nil + return nil, nil } // The Manager already splits messages to ≤4000 chars (WithMaxMessageLength), @@ -189,6 +195,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err // check if HTML expansion pushes it beyond Telegram's 4096-char API limit. replyToID := msg.ReplyToMessageID queue := []string{msg.Content} + var messageIDs []string for len(queue) > 0 { chunk := queue[0] queue = queue[1:] @@ -206,16 +213,18 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err } if smallerLen <= 0 { - if err := c.sendChunk(ctx, sendChunkParams{ + msgID, err := c.sendChunk(ctx, sendChunkParams{ chatID: chatID, threadID: threadID, content: content, replyToID: replyToID, mdFallback: chunk, useMarkdownV2: useMarkdownV2, - }); err != nil { - return err + }) + if err != nil { + return nil, err } + messageIDs = append(messageIDs, msgID) replyToID = "" continue } @@ -244,21 +253,23 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err continue } - if err := c.sendChunk(ctx, sendChunkParams{ + msgID, err := c.sendChunk(ctx, sendChunkParams{ chatID: chatID, threadID: threadID, content: content, replyToID: replyToID, mdFallback: chunk, useMarkdownV2: useMarkdownV2, - }); err != nil { - return err + }) + if err != nil { + return nil, err } + messageIDs = append(messageIDs, msgID) // Only the first chunk should be a reply; subsequent chunks are normal messages. replyToID = "" } - return nil + return messageIDs, nil } type sendChunkParams struct { @@ -275,7 +286,7 @@ type sendChunkParams struct { func (c *TelegramChannel) sendChunk( ctx context.Context, params sendChunkParams, -) error { +) (string, error) { tgMsg := tu.Message(tu.ID(params.chatID), params.content) tgMsg.MessageThreadID = params.threadID if params.useMarkdownV2 { @@ -292,17 +303,19 @@ func (c *TelegramChannel) sendChunk( } } - if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil { + msg, err := c.bot.SendMessage(ctx, tgMsg) + if err != nil { logParseFailed(err, params.useMarkdownV2) tgMsg.Text = params.mdFallback tgMsg.ParseMode = "" - if _, err = c.bot.SendMessage(ctx, tgMsg); err != nil { - return fmt.Errorf("telegram send: %w", channels.ErrTemporary) + msg, err = c.bot.SendMessage(ctx, tgMsg) + if err != nil { + return "", fmt.Errorf("telegram send: %w", channels.ErrTemporary) } } - return nil + return strconv.Itoa(msg.MessageID), nil } // maxTypingDuration limits how long the typing indicator can run. @@ -547,6 +560,8 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes CanonicalID: identity.BuildCanonicalID("telegram", platformID), Username: user.Username, DisplayName: user.FirstName, + FirstName: user.FirstName, + LastName: user.LastName, } // check allowlist to avoid downloading attachments for rejected users diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index fd189d9a7..3deaa18d3 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -98,7 +98,12 @@ func (s *multipartRecordingConstructor) MultipartRequest( // successResponse returns a ta.Response that telego will treat as a successful SendMessage. func successResponse(t *testing.T) *ta.Response { t.Helper() - msg := &telego.Message{MessageID: 1} + return successResponseWithMessageID(t, 1) +} + +func successResponseWithMessageID(t *testing.T, messageID int) *ta.Response { + t.Helper() + msg := &telego.Message{MessageID: messageID} b, err := json.Marshal(msg) require.NoError(t, err) return &ta.Response{Ok: true, Result: b} @@ -280,6 +285,27 @@ func TestSend_LongMessage_SingleCall(t *testing.T) { assert.Len(t, caller.calls, 1, "pre-split message within limit should result in one SendMessage call") } +func TestSendMessageWithIDs_ReturnsAllChunkIDsAfterHTMLResplit(t *testing.T) { + caller := &stubCaller{} + caller.callFn = func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponseWithMessageID(t, len(caller.calls)), nil + } + ch := newTestChannel(t, caller) + + chunk := "[x](https://example.com/" + strings.Repeat("a", 20) + ") " + content := strings.Repeat(chunk, 120) + + ids, err := ch.SendMessageWithIDs(context.Background(), bus.OutboundMessage{ + ChatID: "12345", + Content: content, + }) + + require.NoError(t, err) + require.Len(t, ids, 2) + assert.Equal(t, []string{"1", "2"}, ids) + assert.Len(t, caller.calls, 2) +} + func TestSend_HTMLFallback_PerChunk(t *testing.T) { callCount := 0 caller := &stubCaller{ diff --git a/pkg/providers/protocoltypes/types.go b/pkg/providers/protocoltypes/types.go index 194c1aa6f..331acac71 100644 --- a/pkg/providers/protocoltypes/types.go +++ b/pkg/providers/protocoltypes/types.go @@ -62,6 +62,15 @@ type ContentBlock struct { CacheControl *CacheControl `json:"cache_control,omitempty"` } +// MessageSender carries author identity for a user message. +// Stored alongside the message in history so the LLM can address +// participants by name in multi-user conversations. +type MessageSender struct { + Username string `json:"username,omitempty"` // e.g. "@alice" (platform handle) + FirstName string `json:"first_name,omitempty"` // given name + LastName string `json:"last_name,omitempty"` // family name +} + type Message struct { Role string `json:"role"` Content string `json:"content"` @@ -70,6 +79,9 @@ type Message struct { SystemParts []ContentBlock `json:"system_parts,omitempty"` // structured system blocks for cache-aware adapters ToolCalls []ToolCall `json:"tool_calls,omitempty"` ToolCallID string `json:"tool_call_id,omitempty"` + MessageIDs []string `json:"message_ids,omitempty"` // Platform message IDs + ReplyToMessageID string `json:"reply_to_message_id,omitempty"` // Parent message ID (for threading) + Sender *MessageSender `json:"sender,omitempty"` // Author identity (user messages only) } type ToolDefinition struct { diff --git a/pkg/providers/types.go b/pkg/providers/types.go index f98ae9243..3ea94fd1a 100644 --- a/pkg/providers/types.go +++ b/pkg/providers/types.go @@ -19,6 +19,7 @@ type ( GoogleExtra = protocoltypes.GoogleExtra ContentBlock = protocoltypes.ContentBlock CacheControl = protocoltypes.CacheControl + MessageSender = protocoltypes.MessageSender ) type LLMProvider interface {