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

Fix task plan delivery and Telegram chunk edits

Browse source
This commit is contained in:
Dmitrii Balabanov 2026-03-08 00:54:40 +02:00
parent 96a7125879
commit 73b09860f9
7 changed files with 331 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
pkg/agent/loop.go
View file

@ -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,7 +1242,8 @@ 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 {
if !r.result.Silent && r.result.ForUser != "" {
if opts.SendResponse {
al.bus.PublishOutbound(ctx, bus.OutboundMessage{
Channel: opts.Channel,
ChatID: opts.ChatID,
@ -1252,6 +1254,9 @@ func (al *AgentLoop) runLLMIteration(
"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
}

72
pkg/agent/loop_test.go
View file

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

17
pkg/channels/manager.go
View file

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

34
pkg/channels/manager_test.go
View file

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

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

@ -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,16 +256,23 @@ 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)
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.
// It sends a placeholder message (e.g. "Thinking... 💭") that will later be
@ -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 ""

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

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

105
pkg/tools/tasktool.go
View file

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