diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 06751f81f..efaabf917 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -10,6 +10,7 @@ import ( "context" "encoding/json" "fmt" + "regexp" "strconv" "strings" "sync" @@ -31,6 +32,41 @@ import ( "github.com/sipeed/picoclaw/pkg/utils" ) +// activeTask tracks a running agent task for live status and intervention. +type activeTask struct { + Description string + Iteration int + MaxIter int + StartedAt time.Time + cancel context.CancelFunc + interrupt chan string // buffered 1, for user message injection + toolLog []toolLogEntry + mu sync.Mutex +} + +// toolLogEntry records a single tool call for the live terminal view. +type toolLogEntry struct { + Name string + ArgsSnip string // first ~80 chars of args + Result string // "✓ 4.9s" or "✗ 3.2s" + ErrDetail string // non-empty on error — e.g. "Exit code: exit status 1" +} + +// maxToolLogEntries limits the sliding window of tool log entries +// kept in memory and displayed in status messages. +const maxToolLogEntries = 5 + +// sessionSemaphore is a per-session mutex using a buffered channel. +type sessionSemaphore struct { + ch chan struct{} +} + +func newSessionSemaphore() *sessionSemaphore { + s := &sessionSemaphore{ch: make(chan struct{}, 1)} + s.ch <- struct{}{} // initially unlocked + return s +} + type AgentLoop struct { bus *bus.MessageBus cfg *config.Config @@ -43,6 +79,8 @@ type AgentLoop struct { channelManager *channels.Manager providerCache map[string]providers.LLMProvider planStartPending bool // set by /plan start to trigger LLM execution + sessionLocks sync.Map // sessionKey → *sessionSemaphore + activeTasks sync.Map // sessionKey → *activeTask } // processOptions configures how a message is processed @@ -56,6 +94,7 @@ type processOptions struct { EnableSummary bool // Whether to trigger summarization SendResponse bool // Whether to send response via bus NoHistory bool // If true, don't load session history (for heartbeat) + TaskID string // Unique task ID for background task status tracking } func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers.LLMProvider, enableStats ...bool) *AgentLoop { @@ -353,6 +392,9 @@ func (al *AgentLoop) ProcessDirectWithChannel( ChatID: chatID, Content: content, SessionKey: sessionKey, + Metadata: map[string]string{ + "background": "true", + }, } return al.processMessage(ctx, msg) @@ -390,6 +432,44 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) "session_key": msg.SessionKey, }) + // Handle reply-based intervention for active tasks + if taskID, ok := msg.Metadata["task_id"]; ok && taskID != "" { + if val, found := al.activeTasks.Load(taskID); found { + task := val.(*activeTask) + content := strings.TrimSpace(msg.Content) + lower := strings.ToLower(content) + + // Check for stop keywords + stopKeywords := []string{"stop", "cancel", "abort", "停止", "中止", "やめて"} + isStop := false + for _, kw := range stopKeywords { + if lower == kw { + isStop = true + break + } + } + + if isStop { + task.cancel() + logger.InfoCF("agent", "Task cancelled by user intervention", + map[string]any{"task_id": taskID}) + return "Task cancelled.", nil + } + + // Inject message into interrupt channel for the tool loop + select { + case task.interrupt <- content: + logger.InfoCF("agent", "User intervention queued", + map[string]any{"task_id": taskID, "content": utils.Truncate(content, 80)}) + default: + logger.WarnCF("agent", "Interrupt channel full, message dropped", + map[string]any{"task_id": taskID}) + } + return "Intervention sent to running task.", nil + } + // Task not found — fall through to normal processing + } + // Route system messages to processSystemMessage if msg.Channel == "system" { return al.processSystemMessage(ctx, msg) @@ -428,9 +508,9 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) agent = al.registry.GetDefaultAgent() } - // Use routed session key, but honor pre-set agent-scoped keys (for ProcessDirect/cron) + // Use routed session key, but honor ANY pre-set session key (for ProcessDirect/cron) sessionKey := route.SessionKey - if msg.SessionKey != "" && strings.HasPrefix(msg.SessionKey, "agent:") { + if msg.SessionKey != "" { sessionKey = msg.SessionKey } @@ -509,8 +589,100 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe }) } +// acquireSessionLock gets or creates a per-session semaphore and acquires it. +// Returns false if the context is cancelled before the lock is acquired. +func (al *AgentLoop) acquireSessionLock(ctx context.Context, sessionKey string) bool { + val, _ := al.sessionLocks.LoadOrStore(sessionKey, newSessionSemaphore()) + sem := val.(*sessionSemaphore) + select { + case <-sem.ch: + return true + case <-ctx.Done(): + return false + } +} + +// releaseSessionLock releases the per-session semaphore. +func (al *AgentLoop) releaseSessionLock(sessionKey string) { + if val, ok := al.sessionLocks.Load(sessionKey); ok { + sem := val.(*sessionSemaphore) + sem.ch <- struct{}{} + } +} + // runAgentLoop is the core message processing logic. func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opts processOptions) (string, error) { + // -1. Acquire per-session lock to prevent concurrent access on the same session + if !al.acquireSessionLock(ctx, opts.SessionKey) { + return "", fmt.Errorf("context cancelled while waiting for session lock") + } + defer al.releaseSessionLock(opts.SessionKey) + + // -0. Create cancellable child context and register active task + taskCtx, taskCancel := context.WithCancel(ctx) + defer taskCancel() + + task := &activeTask{ + Description: utils.Truncate(opts.UserMessage, 80), + MaxIter: agent.MaxIterations, + StartedAt: time.Now(), + cancel: taskCancel, + interrupt: make(chan string, 1), + } + + // For background tasks (cron/cli), generate a TaskID and resolve notification channel + isBackgroundTask := constants.IsInternalChannel(opts.Channel) && al.state != nil + if isBackgroundTask && opts.TaskID == "" { + opts.TaskID = fmt.Sprintf("task-%s-%d", opts.SessionKey, time.Now().UnixMilli()) + + // Resolve user's last active channel for notifications + if lastChannel := al.state.GetLastChannel(); lastChannel != "" { + // lastChannel format: "channel:chatID" + if idx := strings.Index(lastChannel, ":"); idx > 0 { + notifyChannel := lastChannel[:idx] + notifyChatID := lastChannel[idx+1:] + + // Override opts channel/chatID for status updates + opts.Channel = notifyChannel + opts.ChatID = notifyChatID + + // Send initial task notification + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: notifyChannel, + ChatID: notifyChatID, + Content: fmt.Sprintf("\U0001F916 Background task started\n%s", task.Description), + IsTaskStatus: true, + TaskID: opts.TaskID, + }) + } + } + } + + // Use TaskID as key if available (for background tasks), else sessionKey + taskKey := opts.SessionKey + if opts.TaskID != "" { + taskKey = opts.TaskID + } + al.activeTasks.Store(taskKey, task) + defer func() { + al.activeTasks.Delete(taskKey) + + // Publish final task status on completion for background tasks + if opts.TaskID != "" && !constants.IsInternalChannel(opts.Channel) { + elapsed := time.Since(task.StartedAt) + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: fmt.Sprintf("\u2705 Task completed (%.1fs)\n%s", elapsed.Seconds(), task.Description), + IsTaskStatus: true, + TaskID: opts.TaskID, + }) + } + }() + + // Replace ctx with the cancellable child context + ctx = taskCtx + // 0. Record last channel for heartbeat notifications (skip internal channels) if opts.Channel != "" && opts.ChatID != "" { // Don't record internal channels (cli, system, subagent) @@ -571,7 +743,7 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt } // 5. Run LLM iteration loop - finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts) + finalContent, iteration, err := al.runLLMIteration(ctx, agent, messages, opts, task) if err != nil { return "", err } @@ -734,12 +906,111 @@ func buildPlanReminder(planStatus string) (providers.Message, bool) { return providers.Message{Role: "user", Content: content}, true } +// cdPrefixPattern matches "cd /some/path && " at the start of a shell command. +var cdPrefixPattern = regexp.MustCompile(`^cd\s+\S+\s*&&\s*`) + +// buildArgsSnippet produces a human-friendly snippet for the tool log. +// For exec: extracts the command and strips the leading "cd && ". +// For file tools: extracts the path and strips the workspace prefix. +// Falls back to raw JSON truncation. +func buildArgsSnippet(toolName string, args map[string]interface{}, workspace string) string { + switch toolName { + case "exec": + cmd, _ := args["command"].(string) + if cmd == "" { + break + } + cmd = cdPrefixPattern.ReplaceAllString(cmd, "") + return utils.Truncate(cmd, 80) + + case "read_file", "write_file", "edit_file", "append_file", "list_dir": + path, _ := args["path"].(string) + if path == "" { + break + } + if workspace != "" { + path = strings.TrimPrefix(path, workspace) + path = strings.TrimPrefix(path, "/") + } + extra := "" + if toolName == "edit_file" { + if old, ok := args["old_text"].(string); ok && old != "" { + extra = " old:" + utils.Truncate(old, 30) + } + } + return utils.Truncate(path, 60) + extra + } + + // Default: raw JSON truncated + argsJSON, _ := json.Marshal(args) + return utils.Truncate(string(argsJSON), 80) +} + +// buildRichStatus builds a terminal-like status display from the active task's tool log. +func buildRichStatus(task *activeTask, isBackground bool, workspace string) string { + task.mu.Lock() + defer task.mu.Unlock() + + var sb strings.Builder + fmt.Fprintf(&sb, "\U0001F504 Task in progress (%d/%d)\n", task.Iteration, task.MaxIter) + // Show project directory name so user knows which workspace is active + if workspace != "" { + // Extract last path component as project name + project := workspace + if idx := strings.LastIndex(workspace, "/"); idx >= 0 { + project = workspace[idx+1:] + } else if idx := strings.LastIndex(workspace, "\\"); idx >= 0 { + project = workspace[idx+1:] + } + if project != "" { + fmt.Fprintf(&sb, "\U0001F4C2 %s\n", project) + } + } + sb.WriteString("\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\n") + + // Sliding window: show only the last maxToolLogEntries entries + entries := task.toolLog + if len(entries) > maxToolLogEntries { + entries = entries[len(entries)-maxToolLogEntries:] + } + for _, entry := range entries { + isErr := strings.HasPrefix(entry.Result, "\u2717") + if isErr { + // Error: multi-line block with decoration + if entry.ArgsSnip != "" { + fmt.Fprintf(&sb, "%s %s %s\n", entry.Name, entry.ArgsSnip, entry.Result) + } else { + fmt.Fprintf(&sb, "%s %s\n", entry.Name, entry.Result) + } + if entry.ErrDetail != "" { + for _, line := range strings.Split(entry.ErrDetail, "\n") { + fmt.Fprintf(&sb, "\u2502 %s\n", line) + } + } + } else { + // Success/pending: compact one-liner + if entry.ArgsSnip != "" { + fmt.Fprintf(&sb, "%s %s %s\n", entry.Name, utils.Truncate(entry.ArgsSnip, 40), entry.Result) + } else { + fmt.Fprintf(&sb, "%s %s\n", entry.Name, entry.Result) + } + } + } + + sb.WriteString("\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\u2501\n") + if isBackground { + sb.WriteString("\u21A9\uFE0F Reply to intervene") + } + return sb.String() +} + // runLLMIteration executes the LLM call loop with tool handling. func (al *AgentLoop) runLLMIteration( ctx context.Context, agent *AgentInstance, messages []providers.Message, opts processOptions, + task *activeTask, ) (string, int, error) { iteration := 0 var finalContent string @@ -747,9 +1018,33 @@ func (al *AgentLoop) runLLMIteration( maxIter := agent.MaxIterations + // Determine if this is a background task (cron, heartbeat, etc.) + isBackground := opts.TaskID != "" + for iteration < maxIter { iteration++ + // Update active task iteration + if task != nil { + task.mu.Lock() + task.Iteration = iteration + task.mu.Unlock() + } + + // Check for user intervention via interrupt channel + if task != nil { + select { + case msg := <-task.interrupt: + messages = append(messages, providers.Message{ + Role: "user", + Content: "[User Intervention] " + msg, + }) + logger.InfoCF("agent", "User intervention injected", + map[string]any{"agent_id": agent.ID, "iteration": iteration}) + default: + } + } + logger.DebugCF("agent", "LLM iteration", map[string]any{ "agent_id": agent.ID, @@ -901,14 +1196,36 @@ func (al *AgentLoop) runLLMIteration( "iteration": iteration, }) - // Publish status update for channels that support placeholder editing - if !constants.IsInternalChannel(opts.Channel) { - al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: opts.Channel, - ChatID: opts.ChatID, - Content: fmt.Sprintf("🔧 %s (%d/%d)", strings.Join(toolNames, ", "), iteration, maxIter), - IsStatus: true, - }) + // Publish rich status update + if !constants.IsInternalChannel(opts.Channel) && task != nil { + // Add pending entries to tool log for the current tool calls + task.mu.Lock() + for _, tc := range normalizedToolCalls { + task.toolLog = append(task.toolLog, toolLogEntry{ + Name: fmt.Sprintf("[%d] %s", iteration, tc.Name), + ArgsSnip: buildArgsSnippet(tc.Name, tc.Arguments, agent.Workspace), + Result: "\u23F3", + }) + } + task.mu.Unlock() + + statusContent := buildRichStatus(task, isBackground, agent.Workspace) + if isBackground { + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: statusContent, + IsTaskStatus: true, + TaskID: opts.TaskID, + }) + } else { + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: statusContent, + IsStatus: true, + }) + } } // Build assistant message with tool calls @@ -945,7 +1262,7 @@ func (al *AgentLoop) runLLMIteration( // Execute tool calls var lastBlocker string - for _, tc := range normalizedToolCalls { + for tcIdx, tc := range normalizedToolCalls { argsJSON, _ := json.Marshal(tc.Arguments) argsPreview := utils.Truncate(string(argsJSON), 200) logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), @@ -973,12 +1290,43 @@ func (al *AgentLoop) runLLMIteration( // Block non-allowed tools during plan interview mode. // Only read-type tools and MEMORY.md writes are permitted. + toolStart := time.Now() var toolResult *tools.ToolResult if isPlanPreExecution(agent.ContextBuilder.GetPlanStatus()) && !isToolAllowedDuringInterview(tc.Name, tc.Arguments) { toolResult = tools.ErrorResult("Interview mode: only read tools and MEMORY.md edits are allowed. Focus on asking questions and updating the plan.") } else { toolResult = agent.Tools.ExecuteWithContext(ctx, tc.Name, tc.Arguments, opts.Channel, opts.ChatID, asyncCallback) } + toolDuration := time.Since(toolStart) + + // Update tool log entry with result + if task != nil { + task.mu.Lock() + // Find the matching pending entry (added earlier in this iteration) + logIdx := len(task.toolLog) - len(normalizedToolCalls) + tcIdx + if logIdx >= 0 && logIdx < len(task.toolLog) { + if toolResult.IsError || toolResult.Err != nil { + task.toolLog[logIdx].Result = fmt.Sprintf("\u2717 %.1fs", toolDuration.Seconds()) + // Extract error detail for block display + if toolResult.Err != nil { + task.toolLog[logIdx].ErrDetail = utils.Truncate(toolResult.Err.Error(), 120) + } else if toolResult.ForLLM != "" { + // exec returns IsError with exit info in ForLLM, not Err + // Show last few lines (stderr / exit code) + lines := strings.Split(strings.TrimSpace(toolResult.ForLLM), "\n") + start := len(lines) - 3 + if start < 0 { + start = 0 + } + task.toolLog[logIdx].ErrDetail = utils.Truncate( + strings.Join(lines[start:], "\n"), 200) + } + } else { + task.toolLog[logIdx].Result = fmt.Sprintf("\u2713 %.1fs", toolDuration.Seconds()) + } + } + task.mu.Unlock() + } // Send ForUser content to user immediately if not Silent if !toolResult.Silent && toolResult.ForUser != "" && opts.SendResponse { @@ -1016,6 +1364,15 @@ func (al *AgentLoop) runLLMIteration( agent.Sessions.AddFullMessage(opts.SessionKey, toolResultMsg) } + // Trim tool log sliding window to prevent unbounded growth + if task != nil { + task.mu.Lock() + if len(task.toolLog) > maxToolLogEntries { + task.toolLog = task.toolLog[len(task.toolLog)-maxToolLogEntries:] + } + task.mu.Unlock() + } + // Inject ephemeral task reminder to prevent focus drift. // Remove previous reminder and re-append at the tail so it stays // close to the LLM's attention window. diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index a507351e5..2fa7d7256 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -1409,3 +1409,171 @@ func TestIsToolAllowedDuringInterview_FuzzyNames(t *testing.T) { } } } + +func TestBuildArgsSnippet_ExecStripsCD(t *testing.T) { + tests := []struct { + name string + tool string + args map[string]interface{} + workspace string + wantSnip string + }{ + { + name: "exec strips cd prefix", + tool: "exec", + args: map[string]interface{}{"command": "cd /home/user/workspace/project/my-projects && pytest tests/test_integration.py"}, + workspace: "/home/user/workspace", + wantSnip: "pytest tests/test_integration.py", + }, + { + name: "exec no cd prefix", + tool: "exec", + args: map[string]interface{}{"command": "ls -la"}, + workspace: "/ws", + wantSnip: "ls -la", + }, + { + name: "exec empty command", + tool: "exec", + args: map[string]interface{}{}, + workspace: "/ws", + wantSnip: "{}", + }, + { + name: "read_file strips workspace", + tool: "read_file", + args: map[string]interface{}{"path": "/home/user/workspace/src/main.go"}, + workspace: "/home/user/workspace", + wantSnip: "src/main.go", + }, + { + name: "edit_file with old_text", + tool: "edit_file", + args: map[string]interface{}{"path": "/ws/config.json", "old_text": "old value here"}, + workspace: "/ws", + wantSnip: "config.json old:old value here", + }, + { + name: "unknown tool shows raw JSON", + tool: "web_search", + args: map[string]interface{}{"query": "hello"}, + workspace: "/ws", + wantSnip: `{"query":"hello"}`, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := buildArgsSnippet(tt.tool, tt.args, tt.workspace) + if got != tt.wantSnip { + t.Errorf("buildArgsSnippet(%q) = %q, want %q", tt.tool, got, tt.wantSnip) + } + }) + } +} + +func TestBuildRichStatus(t *testing.T) { + task := &activeTask{ + Iteration: 3, + MaxIter: 20, + toolLog: []toolLogEntry{ + {Name: "[1] exec", ArgsSnip: "ls -la", Result: "✓ 1.2s"}, + {Name: "[2] exec", ArgsSnip: "pytest tests/", Result: "✓ 5.0s"}, + {Name: "[3] read_file", ArgsSnip: "src/main.go", Result: "⏳"}, + }, + } + + got := buildRichStatus(task, false, "/home/user/my-projects") + + mustContain := []string{ + "Task in progress (3/20)", + "my-projects", + "[1] exec", + "pytest tests/", + "[3] read_file", + "src/main.go", + } + for _, s := range mustContain { + if !strings.Contains(got, s) { + t.Errorf("expected output to contain %q, got:\n%s", s, got) + } + } + + // Non-background: should NOT have reply prompt + if strings.Contains(got, "Reply to intervene") { + t.Error("non-background task should not have reply prompt") + } + + // Background: should have reply prompt + bgGot := buildRichStatus(task, true, "/home/user/my-projects") + if !strings.Contains(bgGot, "Reply to intervene") { + t.Error("background task should have reply prompt") + } +} + +func TestBuildRichStatus_ShowsLast5(t *testing.T) { + task := &activeTask{ + Iteration: 8, + MaxIter: 20, + } + for i := 1; i <= 8; i++ { + task.toolLog = append(task.toolLog, toolLogEntry{ + Name: fmt.Sprintf("[%d] exec", i), + Result: "✓ 1.0s", + }) + } + + got := buildRichStatus(task, false, "/home/user/my-projects") + + // Should contain entries 4-8 but not 1-3 + if strings.Contains(got, "[3] exec") { + t.Error("should not contain entry [3] (only last 5)") + } + if !strings.Contains(got, "[4] exec") { + t.Error("should contain entry [4]") + } + if !strings.Contains(got, "[8] exec") { + t.Error("should contain entry [8]") + } +} + +func TestBuildRichStatus_ErrorBlock(t *testing.T) { + task := &activeTask{ + Iteration: 3, + MaxIter: 10, + toolLog: []toolLogEntry{ + {Name: "[1] exec", ArgsSnip: "ls -la", Result: "✓ 0.5s"}, + { + Name: "[2] exec", + ArgsSnip: "pytest tests/test_auth.py", + Result: "✗ 3.2s", + ErrDetail: "FAILED tests/test_auth.py::test_login\nExit code: exit status 1", + }, + {Name: "[3] read_file", ArgsSnip: "src/auth.py", Result: "⏳"}, + }, + } + + got := buildRichStatus(task, false, "/ws/my-project") + + // Success entry: compact one-liner + if !strings.Contains(got, "[1] exec ls -la ✓ 0.5s") { + t.Errorf("expected compact success line, got:\n%s", got) + } + + // Error entry: command + mark on first line + if !strings.Contains(got, "[2] exec pytest tests/test_auth.py ✗ 3.2s") { + t.Errorf("expected error header line, got:\n%s", got) + } + + // Error detail: │-prefixed lines + if !strings.Contains(got, "│ FAILED tests/test_auth.py::test_login") { + t.Errorf("expected │-prefixed error detail, got:\n%s", got) + } + if !strings.Contains(got, "│ Exit code: exit status 1") { + t.Errorf("expected │-prefixed exit code line, got:\n%s", got) + } + + // Pending entry: compact + if !strings.Contains(got, "[3] read_file src/auth.py ⏳") { + t.Errorf("expected compact pending line, got:\n%s", got) + } +} diff --git a/pkg/bus/types.go b/pkg/bus/types.go index debe1fc57..a6b404a33 100644 --- a/pkg/bus/types.go +++ b/pkg/bus/types.go @@ -15,6 +15,8 @@ type OutboundMessage struct { ChatID string `json:"chat_id"` Content string `json:"content"` IsStatus bool `json:"is_status,omitempty"` + IsTaskStatus bool `json:"is_task_status,omitempty"` + TaskID string `json:"task_id,omitempty"` SkipPlaceholder bool `json:"skip_placeholder,omitempty"` } diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 421684fa3..91c68acdc 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -312,6 +312,21 @@ func (m *Manager) dispatchOutbound(ctx context.Context) { continue } + if msg.IsTaskStatus { + if tc, ok := channel.(interface { + EditTaskStatus(context.Context, bus.OutboundMessage) error + }); ok { + if err := tc.EditTaskStatus(ctx, msg); err != nil { + logger.DebugCF("channels", "EditTaskStatus failed", map[string]interface{}{ + "channel": msg.Channel, + "task_id": msg.TaskID, + "error": err.Error(), + }) + } + } + continue + } + if err := channel.Send(ctx, msg); err != nil { logger.ErrorCF("channels", "Error sending message to channel", map[string]any{ "channel": msg.Channel, diff --git a/pkg/channels/telegram.go b/pkg/channels/telegram.go index 75a9c0d71..62bbb1419 100644 --- a/pkg/channels/telegram.go +++ b/pkg/channels/telegram.go @@ -26,13 +26,15 @@ import ( type TelegramChannel struct { *BaseChannel - bot *telego.Bot - commands TelegramCommander - config *config.Config - chatIDs map[string]int64 - transcriber *voice.GroqTranscriber - placeholders sync.Map // chatID -> messageID - stopThinking sync.Map // chatID -> thinkingCancel + bot *telego.Bot + commands TelegramCommander + config *config.Config + chatIDs map[string]int64 + transcriber *voice.GroqTranscriber + placeholders sync.Map // chatID -> messageID + stopThinking sync.Map // chatID -> thinkingCancel + taskStatuses sync.Map // taskID -> messageID (int) + taskStatusReverse sync.Map // messageID (int) -> taskID (string) } type thinkingCancel struct { @@ -182,6 +184,47 @@ func (c *TelegramChannel) EditStatus(ctx context.Context, msg bus.OutboundMessag return err } +func (c *TelegramChannel) EditTaskStatus(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() || msg.TaskID == "" { + return nil + } + chatID, err := parseChatID(msg.ChatID) + if err != nil { + return nil + } + + // Check if we already have a message for this task + if existingMsgID, ok := c.taskStatuses.Load(msg.TaskID); ok { + // Edit existing task status message + editMsg := tu.EditMessageText(tu.ID(chatID), existingMsgID.(int), msg.Content) + _, err = c.bot.EditMessageText(ctx, editMsg) + if err != nil { + logger.DebugCF("telegram", "EditTaskStatus edit failed", map[string]interface{}{ + "error": err.Error(), + "task_id": msg.TaskID, + }) + } + return err + } + + // First task status message: send a new message and track it + tgMsg := tu.Message(tu.ID(chatID), msg.Content) + sent, err := c.bot.SendMessage(ctx, tgMsg) + if err != nil { + return err + } + c.taskStatuses.Store(msg.TaskID, sent.MessageID) + c.taskStatusReverse.Store(sent.MessageID, msg.TaskID) + return nil +} + +// CleanupTaskStatus removes tracking entries for a completed task. +func (c *TelegramChannel) CleanupTaskStatus(taskID string) { + if msgID, ok := c.taskStatuses.LoadAndDelete(taskID); ok { + c.taskStatusReverse.Delete(msgID) + } +} + func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { if !c.IsRunning() { return fmt.Errorf("telegram bot not running") @@ -437,6 +480,16 @@ func (c *TelegramChannel) handleMessage(ctx context.Context, message *telego.Mes "peer_id": peerID, } + // Detect reply-to messages for task intervention + if message.ReplyToMessage != nil { + replyToID := message.ReplyToMessage.MessageID + metadata["reply_to_message_id"] = fmt.Sprintf("%d", replyToID) + // Check if replying to a task status message + if taskID, ok := c.taskStatusReverse.Load(replyToID); ok { + metadata["task_id"] = taskID.(string) + } + } + c.HandleMessage(fmt.Sprintf("%d", user.ID), fmt.Sprintf("%d", chatID), content, mediaPaths, metadata) return nil }