From 71da3d7889666a1fea7680bfbc29aea6ded80995 Mon Sep 17 00:00:00 2001
From: viettranx <viettranx@gmail.com>
Date: Wed, 4 Mar 2026 08:06:03 +0700
Subject: [PATCH] feat(delegation): proactive progress notification +
 in-progress task reminder

Send a "still working" message to users after 45s if async delegation is
still running, using PublishOutbound directly. Also expand the team task
reminder to include in-progress tasks so lead agents wait for delegates.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 internal/agent/loop.go     | 26 +++++++++++++++++++++-----
 internal/tools/delegate.go | 28 ++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 5 deletions(-)

diff --git a/internal/agent/loop.go b/internal/agent/loop.go
index 959fd431..73a099cd 100644
--- a/internal/agent/loop.go
+++ b/internal/agent/loop.go
@@ -537,27 +537,43 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
 		}
 	}
 
-	// 2b. Cross-session recovery: notify team leads about orphaned pending tasks.
+	// 2b. Cross-session recovery: notify team leads about orphaned pending tasks
+	// and in-progress tasks being handled by delegates.
 	// Safe because Bước 1 (early ClaimTask) ensures running tasks are in_progress,
 	// so only truly un-spawned tasks remain pending.
 	if l.teamStore != nil && l.agentUUID != uuid.Nil {
 		if team, _ := l.teamStore.GetTeamForAgent(ctx, l.agentUUID); team != nil && team.LeadAgentID == l.agentUUID {
 			if tasks, err := l.teamStore.ListTasks(ctx, team.ID, "newest", "", req.UserID); err == nil {
 				var stale []string
+				var inProgress []string
 				for _, t := range tasks {
 					if t.Status == store.TeamTaskStatusPending {
 						age := time.Since(t.CreatedAt).Truncate(time.Minute)
 						stale = append(stale, fmt.Sprintf("- %s: \"%s\" (pending %s)", t.ID, t.Subject, age))
 					}
+					if t.Status == store.TeamTaskStatusInProgress {
+						age := time.Since(t.UpdatedAt).Truncate(time.Minute)
+						inProgress = append(inProgress, fmt.Sprintf("- %s: \"%s\" (in progress %s)", t.ID, t.Subject, age))
+					}
 				}
+				var parts []string
 				if len(stale) > 0 {
-					reminder := fmt.Sprintf(
-						"[System] You have %d pending team task(s) that were never spawned:\n%s\n"+
+					parts = append(parts, fmt.Sprintf(
+						"You have %d pending team task(s) that were never spawned:\n%s\n"+
 							"Spawn each one, or cancel with team_tasks action=cancel if no longer needed.",
-						len(stale), strings.Join(stale, "\n"))
+						len(stale), strings.Join(stale, "\n")))
+				}
+				if len(inProgress) > 0 {
+					parts = append(parts, fmt.Sprintf(
+						"You have %d in-progress team task(s) being handled by delegates:\n%s\n"+
+							"Wait for their results before acting on these tasks.",
+						len(inProgress), strings.Join(inProgress, "\n")))
+				}
+				if len(parts) > 0 {
+					reminder := "[System] " + strings.Join(parts, "\n\n")
 					messages = append(messages,
 						providers.Message{Role: "user", Content: reminder},
-						providers.Message{Role: "assistant", Content: "I see the pending tasks. Let me handle them."},
+						providers.Message{Role: "assistant", Content: "I see the task status. Let me handle accordingly."},
 					)
 				}
 			}
diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go
index 2d3ffe7d..57bd4056 100644
--- a/internal/tools/delegate.go
+++ b/internal/tools/delegate.go
@@ -17,6 +17,7 @@ import (
 )
 
 const defaultMaxDelegationLoad = 5
+const delegationProgressDelay = 45 * time.Second
 
 // DelegationTask tracks an active delegation for concurrency control and cancellation.
 type DelegationTask struct {
@@ -254,6 +255,12 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts)
 			dm.active.Delete(task.ID)
 		}()
 
+		// Progress notification timer — fires once after delay if still running
+		progressTimer := time.AfterFunc(delegationProgressDelay, func() {
+			dm.sendProgressNotification(task)
+		})
+		defer progressTimer.Stop()
+
 		startTime := time.Now()
 		result, runErr := dm.runAgent(taskCtx, opts.TargetAgentKey, runReq)
 		duration := time.Since(startTime)
@@ -577,6 +584,27 @@ func (dm *DelegateManager) injectDependencyResults(ctx context.Context, opts *De
 	}
 }
 
+// sendProgressNotification sends a proactive "still working" message to the user
+// when an async delegation takes longer than delegationProgressDelay.
+func (dm *DelegateManager) sendProgressNotification(task *DelegationTask) {
+	if dm.msgBus == nil || task.OriginChannel == "" || task.OriginChatID == "" {
+		return
+	}
+	elapsed := time.Since(task.CreatedAt).Round(time.Second)
+	content := fmt.Sprintf("⏳ %s is still working on the task... (%s elapsed)",
+		task.TargetAgentKey, elapsed)
+
+	dm.msgBus.PublishOutbound(bus.OutboundMessage{
+		Channel: task.OriginChannel,
+		ChatID:  task.OriginChatID,
+		Content: content,
+		Metadata: map[string]string{
+			"local_key": task.OriginLocalKey,
+			"peer_kind": task.OriginPeerKind,
+		},
+	})
+}
+
 func buildDelegateMessage(opts DelegateOpts) string {
 	if opts.Context != "" {
 		return fmt.Sprintf("[Additional Context]\n%s\n\n[Task]\n%s", opts.Context, opts.Task)