From 73b09860f953b190c92a42e6bf73671a91ec4ef0 Mon Sep 17 00:00:00 2001 From: Dmitrii Balabanov Date: Sun, 8 Mar 2026 00:54:40 +0200 Subject: [PATCH] Fix task plan delivery and Telegram chunk edits --- pkg/agent/loop.go | 29 ++++-- pkg/agent/loop_test.go | 72 ++++++++++++++ pkg/channels/manager.go | 19 ++-- pkg/channels/manager_test.go | 34 +++++++ pkg/channels/telegram/telegram.go | 131 ++++++++++++++++++------- pkg/channels/telegram/telegram_test.go | 28 +++++- pkg/tools/tasktool.go | 107 ++++++++++++++------ 7 files changed, 331 insertions(+), 89 deletions(-) diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 588d55c50..986b86b88 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -909,6 +909,7 @@ func (al *AgentLoop) runLLMIteration( ) (string, int, error) { iteration := 0 var finalContent string + var directToolOutputs []string // Determine effective model tier for this conversation turn. // selectCandidates evaluates routing once and the decision is sticky for @@ -1241,17 +1242,21 @@ func (al *AgentLoop) runLLMIteration( // Process results in original order (send to user, save to session) for _, r := range agentResults { // Send ForUser content to user immediately if not Silent - if !r.result.Silent && r.result.ForUser != "" && opts.SendResponse { - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: r.result.ForUser, - }) - logger.DebugCF("agent", "Sent tool result to user", - map[string]any{ - "tool": r.tc.Name, - "content_len": len(r.result.ForUser), + if !r.result.Silent && r.result.ForUser != "" { + if opts.SendResponse { + al.bus.PublishOutbound(ctx, bus.OutboundMessage{ + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: r.result.ForUser, }) + logger.DebugCF("agent", "Sent tool result to user", + map[string]any{ + "tool": r.tc.Name, + "content_len": len(r.result.ForUser), + }) + } else { + directToolOutputs = append(directToolOutputs, r.result.ForUser) + } } // If tool returned media refs, publish them as outbound media @@ -1293,6 +1298,10 @@ func (al *AgentLoop) runLLMIteration( } } + if strings.TrimSpace(finalContent) == "" && len(directToolOutputs) > 0 { + finalContent = strings.Join(directToolOutputs, "\n\n") + } + return finalContent, iteration, nil } diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 2e456fa60..5625dafba 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -342,6 +342,48 @@ func (m *countingMockProvider) GetDefaultModel() string { return "counting-mock-model" } +type taskToolPlanMockProvider struct { + calls int +} + +func (m *taskToolPlanMockProvider) Chat( + ctx context.Context, + messages []providers.Message, + tools []providers.ToolDefinition, + model string, + opts map[string]any, +) (*providers.LLMResponse, error) { + m.calls++ + if m.calls == 1 { + return &providers.LLMResponse{ + ToolCalls: []providers.ToolCall{ + { + ID: "call_tasktool", + Name: "tasktool", + Arguments: map[string]any{ + "action": "create_plan", + "tasks": []any{ + map[string]any{ + "id": "step_1", + "description": "Inspect direct mode", + }, + }, + }, + }, + }, + }, nil + } + + return &providers.LLMResponse{ + Content: "", + ToolCalls: []providers.ToolCall{}, + }, nil +} + +func (m *taskToolPlanMockProvider) GetDefaultModel() string { + return "tasktool-mock-model" +} + // mockCustomTool is a simple mock tool for registration testing type mockCustomTool struct{} @@ -659,6 +701,36 @@ func TestToolResult_UserFacingToolDoesSendMessage(t *testing.T) { } } +func TestTaskTool_DirectModeWithoutChannelManagerReturnsPlan(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.DefaultConfig() + cfg.Agents.Defaults.Workspace = tmpDir + cfg.Agents.Defaults.Model = "test-model" + cfg.Agents.Defaults.MaxTokens = 4096 + cfg.Agents.Defaults.MaxToolIterations = 4 + + msgBus := bus.NewMessageBus() + provider := &taskToolPlanMockProvider{} + al := NewAgentLoop(cfg, msgBus, provider) + + response, err := al.ProcessDirect(context.Background(), "make a plan", "cli:default") + if err != nil { + t.Fatalf("ProcessDirect failed: %v", err) + } + + if !strings.Contains(response, "Execution Plan") { + t.Fatalf("expected direct-mode response to include the execution plan, got: %q", response) + } + if !strings.Contains(response, "Inspect direct mode") { + t.Fatalf("expected direct-mode response to include the task description, got: %q", response) + } +} + // failFirstMockProvider fails on the first N calls with a specific error type failFirstMockProvider struct { failures int diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 65c05b46f..1fb319823 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -836,7 +836,7 @@ func (m *Manager) SendToChannel(ctx context.Context, channelName, chatID, conten // SendMessageWithID sends a message synchronously via the channel's native API if supported, // returning the platform-specific message ID. If the channel does not support SyncSender, -// it falls back to the async bus and returns an error. +// it falls back to the async bus and returns an empty message ID with a nil error. func (m *Manager) SendMessageWithID(ctx context.Context, msg bus.OutboundMessage) (string, error) { ch, ok := m.GetChannel(msg.Channel) if !ok { @@ -845,21 +845,20 @@ func (m *Manager) SendMessageWithID(ctx context.Context, msg bus.OutboundMessage if syncSender, ok := ch.(SyncSender); ok { msgID, err := syncSender.SendMessageWithID(ctx, msg) - if err == nil && msgID != "" { - return msgID, nil + if err != nil { + logger.ErrorCF("manager", "SendMessageWithID failed", map[string]any{"error": err, "msgID": msgID}) + return "", err } - logger.ErrorCF("manager", "SendMessageWithID failed", map[string]any{"error": err, "msgID": msgID}) - if err == nil { - err = fmt.Errorf("sync sender returned empty message ID") - } - return "", err + return msgID, nil } logger.WarnCF("manager", "channel does not implement SyncSender", map[string]any{"channel": msg.Channel}) logger.WarnCF("manager", "falling back to bus publish", nil) - m.bus.PublishOutbound(ctx, msg) + if err := m.bus.PublishOutbound(ctx, msg); err != nil { + return "", err + } - return "", fmt.Errorf("channel does not support returning message ID") + return "", nil } // EditMessage synchronously edits an existing message if the channel supports MessageEditor. diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index f09ecfe2f..7d9ea700e 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -35,6 +35,40 @@ func newTestManager() *Manager { } } +func TestSendMessageWithID_FallsBackToBusWithoutError(t *testing.T) { + msgBus := bus.NewMessageBus() + m := &Manager{ + channels: map[string]Channel{ + "test": &mockChannel{ + sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil }, + }, + }, + workers: make(map[string]*channelWorker), + bus: msgBus, + } + + msgID, err := m.SendMessageWithID(context.Background(), bus.OutboundMessage{ + Channel: "test", + ChatID: "1", + Content: "hello", + }) + + if err != nil { + t.Fatalf("expected nil error for async fallback, got %v", err) + } + if msgID != "" { + t.Fatalf("expected empty message ID for async fallback, got %q", msgID) + } + + outbound, ok := msgBus.SubscribeOutbound(context.Background()) + if !ok { + t.Fatal("expected fallback message to be queued on the outbound bus") + } + if outbound.Content != "hello" { + t.Fatalf("expected fallback content %q, got %q", "hello", outbound.Content) + } +} + func TestSendWithRetry_Success(t *testing.T) { m := newTestManager() var callCount int diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 4b74f70e4..73f0cd2db 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -183,42 +183,20 @@ func (c *TelegramChannel) SendMessageWithID(ctx context.Context, msg bus.Outboun return "", nil } - // The Manager already splits messages to ≤4000 chars (WithMaxMessageLength), - // so msg.Content is guaranteed to be within that limit. We still need to - // check if HTML expansion pushes it beyond Telegram's 4096-char API limit. - queue := []string{msg.Content} - var lastMsgID int - - for len(queue) > 0 { - chunk := queue[0] - queue = queue[1:] - - htmlContent := markdownToTelegramHTML(chunk) - - if len([]rune(htmlContent)) > 4096 { - ratio := float64(len([]rune(chunk))) / float64(len([]rune(htmlContent))) - smallerLen := int(float64(4096) * ratio * 0.95) // 5% safety margin - if smallerLen < 100 { - smallerLen = 100 - } - // Push sub-chunks back to the front of the queue for - // re-validation instead of sending them blindly. - subChunks := channels.SplitMessage(chunk, smallerLen) - queue = append(subChunks, queue...) - continue - } - - msgID, err := c.sendHTMLChunk(ctx, cid, htmlContent, chunk) + chunks := telegramMessageChunks(msg.Content) + ids := make([]string, 0, len(chunks)) + for _, chunk := range chunks { + msgID, err := c.sendHTMLChunk(ctx, cid, chunk.HTML, chunk.Markdown) if err != nil { return "", err } - lastMsgID = msgID + ids = append(ids, strconv.Itoa(msgID)) } - if lastMsgID == 0 { + if len(ids) == 0 { return "", nil } - return fmt.Sprintf("%d", lastMsgID), nil + return strings.Join(ids, ","), nil } // sendHTMLChunk sends a single HTML message, falling back to the original @@ -278,15 +256,22 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag if err != nil { return err } - mid, err := strconv.Atoi(messageID) + messageIDs, err := parseTelegramMessageIDs(messageID) if err != nil { return err } - htmlContent := markdownToTelegramHTML(content) - editMsg := tu.EditMessageText(tu.ID(cid), mid, htmlContent) - editMsg.ParseMode = telego.ModeHTML - _, err = c.bot.EditMessageText(ctx, editMsg) - return err + chunks := telegramMessageChunks(content) + if len(messageIDs) != len(chunks) { + return fmt.Errorf("telegram edit: chunk count changed from %d to %d", len(messageIDs), len(chunks)) + } + + for i, mid := range messageIDs { + if err := c.editHTMLChunk(ctx, cid, mid, chunks[i].HTML, chunks[i].Markdown); err != nil { + return err + } + } + + return nil } // SendPlaceholder implements channels.PlaceholderCapable. @@ -604,6 +589,82 @@ func parseChatID(chatIDStr string) (int64, error) { return id, err } +type telegramMessageChunk struct { + Markdown string + HTML string +} + +func telegramMessageChunks(content string) []telegramMessageChunk { + queue := []string{content} + chunks := make([]telegramMessageChunk, 0, 1) + + for len(queue) > 0 { + chunk := queue[0] + queue = queue[1:] + + htmlContent := markdownToTelegramHTML(chunk) + if len([]rune(htmlContent)) > 4096 { + ratio := float64(len([]rune(chunk))) / float64(len([]rune(htmlContent))) + smallerLen := int(float64(4096) * ratio * 0.95) // 5% safety margin + if smallerLen < 100 { + smallerLen = 100 + } + subChunks := channels.SplitMessage(chunk, smallerLen) + queue = append(subChunks, queue...) + continue + } + + chunks = append(chunks, telegramMessageChunk{ + Markdown: chunk, + HTML: htmlContent, + }) + } + + return chunks +} + +func parseTelegramMessageIDs(messageID string) ([]int, error) { + parts := strings.Split(messageID, ",") + ids := make([]int, 0, len(parts)) + + for _, part := range parts { + part = strings.TrimSpace(part) + if part == "" { + continue + } + + id, err := strconv.Atoi(part) + if err != nil { + return nil, err + } + ids = append(ids, id) + } + + if len(ids) == 0 { + return nil, fmt.Errorf("telegram edit: empty message ID") + } + + return ids, nil +} + +func (c *TelegramChannel) editHTMLChunk(ctx context.Context, chatID int64, messageID int, htmlContent, mdFallback string) error { + editMsg := tu.EditMessageText(tu.ID(chatID), messageID, htmlContent) + editMsg.ParseMode = telego.ModeHTML + + if _, err := c.bot.EditMessageText(ctx, editMsg); err != nil { + logger.ErrorCF("telegram", "HTML edit failed, falling back to plain text", map[string]any{ + "error": err.Error(), + }) + editMsg.Text = mdFallback + editMsg.ParseMode = "" + if _, err = c.bot.EditMessageText(ctx, editMsg); err != nil { + return fmt.Errorf("telegram edit: %w", channels.ErrTemporary) + } + } + + return nil +} + func markdownToTelegramHTML(text string) string { if text == "" { return "" diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 0ff5df5f2..ffb9bbad0 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -50,8 +50,12 @@ func (s *stubConstructor) MultipartRequest( // successResponse returns a ta.Response that telego will treat as a successful SendMessage. func successResponse(t *testing.T) *ta.Response { + return successResponseWithID(t, 1) +} + +func successResponseWithID(t *testing.T, id int) *ta.Response { t.Helper() - msg := &telego.Message{MessageID: 1} + msg := &telego.Message{MessageID: id} b, err := json.Marshal(msg) require.NoError(t, err) return &ta.Response{Ok: true, Result: b} @@ -199,9 +203,11 @@ func TestSendMessageWithID_LongMessage_HTMLFallback_StopsOnError(t *testing.T) { } func TestSendMessageWithID_MarkdownShortButHTMLLong_MultipleCalls(t *testing.T) { + callCount := 0 caller := &stubCaller{ callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { - return successResponse(t), nil + callCount++ + return successResponseWithID(t, callCount), nil }, } ch := newTestChannel(t, caller) @@ -212,8 +218,24 @@ func TestSendMessageWithID_MarkdownShortButHTMLLong_MultipleCalls(t *testing.T) msgID, err := ch.SendMessageWithID(context.Background(), bus.OutboundMessage{ChatID: "12345", Content: markdownContent}) assert.NoError(t, err) - assert.Equal(t, "1", msgID) assert.Greater(t, len(caller.calls), 1, "markdown-short but HTML-long message should be split into multiple SendMessage calls") + assert.Equal(t, "1,2", msgID) +} + +func TestEditMessage_MultipleChunkIDs(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + + content := strings.Repeat("**a** ", 600) + + err := ch.EditMessage(context.Background(), "12345", "1,2", content) + + assert.NoError(t, err) + assert.Len(t, caller.calls, 2, "multi-part edit should update every tracked message") } func TestSendMessageWithID_NotRunning(t *testing.T) { diff --git a/pkg/tools/tasktool.go b/pkg/tools/tasktool.go index 9885ac4c5..13cd240e6 100644 --- a/pkg/tools/tasktool.go +++ b/pkg/tools/tasktool.go @@ -159,20 +159,23 @@ func (t *TaskTool) handleCreatePlan(ctx context.Context, sessionKey, channel, ch st := t.taskManager.CreatePlan(sessionKey, parsedTasks) content := t.formatPlanMessage(st.Tasks) + summary := fmt.Sprintf("Plan created with %d tasks.\nTasks: %s", len(parsedTasks), mustMarshalJSON(parsedTasks)) - // Send message through callback if available + delivered := false + var deliveryErr error if t.sendPlaceholder != nil { msgID, err := t.sendPlaceholder(ctx, channel, chatID, content) - if err == nil && msgID != "" { - t.taskManager.SetMessageID(sessionKey, msgID) + if err == nil { + delivered = true + if msgID != "" { + t.taskManager.SetMessageID(sessionKey, msgID) + } + } else { + deliveryErr = err } } - tasksJSON, _ := json.Marshal(parsedTasks) - return &ToolResult{ - ForLLM: fmt.Sprintf("Plan created with %d tasks.\nTasks: %s", len(parsedTasks), string(tasksJSON)), - Silent: true, // We already sent the message via callback - } + return t.newPlanResult(summary, content, delivered, deliveryErr) } func (t *TaskTool) handleListPlan(sessionKey string) *ToolResult { @@ -185,11 +188,14 @@ func (t *TaskTool) handleListPlan(sessionKey string) *ToolResult { } content := t.formatPlanMessage(st.Tasks) - tasksJSON, _ := json.Marshal(st.Tasks) + summary := fmt.Sprintf("Current plan state:\n%s\n\nRaw JSON:\n%s", content, mustMarshalJSON(st.Tasks)) + // list_plan does not send anything on its own, so always expose the plan to + // direct callers through ForUser as well. return &ToolResult{ - ForLLM: fmt.Sprintf("Current plan state:\n%s\n\nRaw JSON:\n%s", content, string(tasksJSON)), - Silent: true, + ForLLM: summary, + ForUser: content, + Silent: false, } } @@ -204,26 +210,26 @@ func (t *TaskTool) handleResendPlan(ctx context.Context, sessionKey, channel, ch content := t.formatPlanMessage(st.Tasks) + delivered := false + var deliveryErr error if t.sendPlaceholder != nil { msgID, err := t.sendPlaceholder(ctx, channel, chatID, content) if err == nil { + delivered = true if msgID != "" { t.taskManager.SetMessageID(sessionKey, msgID) } - // If err == nil but msgID == "", the channel delivered the message - // (or is async) but doesn't support returning IDs. We consider this a success. } else { - return &ToolResult{ForLLM: fmt.Sprintf("Failed to resend message: %v", err), IsError: true} + deliveryErr = err } - } else { - return &ToolResult{ForLLM: "tasktool: channel sending callbacks are not configured", IsError: true} } - tasksJSON, _ := json.Marshal(st.Tasks) - return &ToolResult{ - ForLLM: fmt.Sprintf("Plan successfully resent as a new message.\nTasks: %s", string(tasksJSON)), - Silent: true, + summary := fmt.Sprintf("Plan content prepared for resend.\nTasks: %s", mustMarshalJSON(st.Tasks)) + if delivered { + summary = fmt.Sprintf("Plan successfully resent as a new message.\nTasks: %s", mustMarshalJSON(st.Tasks)) } + + return t.newPlanResult(summary, content, delivered, deliveryErr) } func (t *TaskTool) handleUpdateTask(ctx context.Context, sessionKey, channel, chatID string, args map[string]any) *ToolResult { @@ -245,8 +251,10 @@ func (t *TaskTool) handleUpdateTask(ctx context.Context, sessionKey, channel, ch } content := t.formatPlanMessage(st.Tasks) + summary := fmt.Sprintf("Task '%s' updated to '%s'. Current plan:\n%s", taskID, statusStr, mustMarshalJSON(st.Tasks)) - // Edit message through callback if available + delivered := false + var deliveryErr error if t.editMessage != nil && st.MessageID != "" { if err := t.editMessage(ctx, channel, chatID, st.MessageID, content); err != nil { logger.WarnCF("tasktool", "Failed to edit task message", map[string]any{ @@ -255,20 +263,27 @@ func (t *TaskTool) handleUpdateTask(ctx context.Context, sessionKey, channel, ch "message_id": st.MessageID, "error": err.Error(), }) - } - } else if t.sendPlaceholder != nil && st.MessageID == "" { - // Fallback: send new progress message if we didn't have one - msgID, err := t.sendPlaceholder(ctx, channel, chatID, content) - if err == nil && msgID != "" { - t.taskManager.SetMessageID(sessionKey, msgID) + deliveryErr = err + } else { + delivered = true } } - tasksJSON, _ := json.Marshal(st.Tasks) - return &ToolResult{ - ForLLM: fmt.Sprintf("Task '%s' updated to '%s'. Current plan:\n%s", taskID, statusStr, string(tasksJSON)), - Silent: true, + if !delivered && t.sendPlaceholder != nil { + // Fallback: send new progress message if we didn't have one + msgID, err := t.sendPlaceholder(ctx, channel, chatID, content) + if err == nil { + delivered = true + deliveryErr = nil + if msgID != "" { + t.taskManager.SetMessageID(sessionKey, msgID) + } + } else if deliveryErr == nil { + deliveryErr = err + } } + + return t.newPlanResult(summary, content, delivered, deliveryErr) } func (t *TaskTool) formatPlanMessage(tasks []session.Task) string { @@ -303,3 +318,33 @@ func (t *TaskTool) formatPlanMessage(tasks []session.Task) string { return sb.String() } + +func (t *TaskTool) newPlanResult(summary, content string, delivered bool, deliveryErr error) *ToolResult { + if delivered { + return &ToolResult{ + ForLLM: summary, + Silent: true, + } + } + + forLLM := summary + if deliveryErr != nil { + forLLM = fmt.Sprintf("%s\nAutomatic delivery failed (%v). Respond to the user with the following plan content:\n\n%s", summary, deliveryErr, content) + } else { + forLLM = fmt.Sprintf("%s\nAutomatic delivery is unavailable in this context. Respond to the user with the following plan content:\n\n%s", summary, content) + } + + return &ToolResult{ + ForLLM: forLLM, + ForUser: content, + Silent: false, + } +} + +func mustMarshalJSON(v any) string { + data, err := json.Marshal(v) + if err != nil { + return "[]" + } + return string(data) +}