feat: live task terminal, session lock, reply-based intervention

- Fix session key collision: honor any pre-set session key, not just
  "agent:" prefix, so cron jobs use their own session
- Add per-session semaphore lock to prevent concurrent access on the
  same session key
- Replace simple "🔧 exec (3/20)" status with rich terminal view
  showing a sliding window of last 5 tool calls with timing, project
  name, and error details
- Add IsTaskStatus/TaskID to OutboundMessage for background task
  status tracking (send new message, then edit in-place)
- Telegram: track task status messages, detect reply-to for
  intervention routing
- Channel manager: dispatch IsTaskStatus messages to channels
- Reply-based intervention: reply "stop/cancel/abort/停止/中止/やめて"
  to cancel, or any other text to inject into the LLM context
- Background task notifications: auto-notify user's last active
  channel when cron/cli tasks start and complete
- Smart arg snippets: strip "cd <workspace> && " from exec commands,
  show relative paths for file tools

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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 <workspace> && ".
+// 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,15 +1196,37 @@ func (al *AgentLoop) runLLMIteration(
 				"iteration": iteration,
 			})
 
-		// Publish status update for channels that support placeholder editing
-		if !constants.IsInternalChannel(opts.Channel) {
+		// 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:  fmt.Sprintf("🔧 %s (%d/%d)", strings.Join(toolNames, ", "), iteration, maxIter),
+					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
 		assistantMsg := providers.Message{
@@ -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.

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

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

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,

pkg/channels/telegram.go

@@ -33,6 +33,8 @@ type TelegramChannel struct {
 	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
 }