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

feat(session): preserve threaded message history metadata

Browse source
This commit is contained in:
Dmitrii Balabanov 2026-03-23 10:01:45 +02:00
parent 056b9f5845
commit 8647d56b3d
19 changed files with 726 additions and 214 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
pkg/agent/context.go
View file

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

12
pkg/agent/eventbus_test.go
View file

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

12
pkg/agent/hook_process_test.go
View file

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

12
pkg/agent/hooks_test.go
View file

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

235
pkg/agent/loop.go
View file

@ -88,6 +88,9 @@ type processOptions struct {
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,6 +99,46 @@ 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."
@ -106,6 +149,7 @@ const (
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 != "" {
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{
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) {
@ -1677,12 +1827,11 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er
Role: "user",
Content: ts.userMessage,
Media: append([]string(nil), ts.media...),
MessageIDs: singleMessageIDs(ts.opts.MessageID),
ReplyToMessageID: ts.opts.ReplyToMessageID,
Sender: ts.opts.Sender,
}
if len(rootMsg.Media) > 0 {
ts.agent.Sessions.AddFullMessage(ts.sessionKey, rootMsg)
} else {
ts.agent.Sessions.AddMessage(ts.sessionKey, rootMsg.Role, rootMsg.Content)
}
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,
}
}

83
pkg/agent/loop_test.go
View file

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

9
pkg/agent/steering.go
View file

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

10
pkg/bus/bus.go
View file

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

5
pkg/bus/types.go
View file

@ -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 {
@ -34,6 +36,7 @@ type OutboundMessage struct {
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.

51
pkg/channels/discord/discord.go
View file

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

6
pkg/channels/interfaces.go
View file

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

89
pkg/channels/manager.go
View file

@ -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,15 +620,32 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker)
if !ok {
return
}
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()
}
// 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 {
@ -636,22 +653,24 @@ func (m *Manager) runWorker(ctx context.Context, name string, w *channelWorker)
}
}
}
// 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
var messageIDs []string
for _, chunk := range chunks {
chunkMsg := msg
chunkMsg.Content = chunk
m.sendWithRetry(ctx, name, w, chunkMsg)
chunkMsg.OnDelivered = nil
chunkIDs, delivered := m.sendWithRetry(ctx, name, w, chunkMsg)
if !delivered {
return nil, false
}
case <-ctx.Done():
return
if len(chunkIDs) > 0 {
messageIDs = append(messageIDs, chunkIDs...)
}
}
return messageIDs, true
}
// splitByLength splits content by maxLen if needed, otherwise returns single chunk.
@ -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++ {
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
}

141
pkg/channels/manager_test.go
View file

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

24
pkg/channels/qq/qq.go
View file

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

16
pkg/channels/slack/slack.go
View file

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

45
pkg/channels/telegram/telegram.go
View file

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

28
pkg/channels/telegram/telegram_test.go
View file

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

12
pkg/providers/protocoltypes/types.go
View file

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

1
pkg/providers/types.go
View file

@ -19,6 +19,7 @@ type (
GoogleExtra = protocoltypes.GoogleExtra
ContentBlock = protocoltypes.ContentBlock
CacheControl = protocoltypes.CacheControl
MessageSender = protocoltypes.MessageSender
)
type LLMProvider interface {