From d62fd830e0aca7469acfa34abc319c0e3a664b85 Mon Sep 17 00:00:00 2001 From: viettranx Date: Wed, 4 Mar 2026 08:29:53 +0700 Subject: [PATCH] feat(delegation): grouped progress notification + configurable estimated_duration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Progress notifications are now grouped per source agent — one combined message lists all active delegations instead of spamming per-task. Agents can set estimated_duration (seconds) when spawning to control the notification delay (default 90s). Uses dedup via progressSent map. Co-Authored-By: Claude Opus 4.6 --- internal/tools/delegate.go | 66 +++++++++++++++++++++------ internal/tools/subagent_spawn_tool.go | 20 ++++++-- 2 files changed, 66 insertions(+), 20 deletions(-) diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index 57bd4056..d2ab782a 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -17,7 +17,7 @@ import ( ) const defaultMaxDelegationLoad = 5 -const delegationProgressDelay = 45 * time.Second +const defaultProgressDelay = 90 * time.Second // DelegationTask tracks an active delegation for concurrency control and cancellation. type DelegationTask struct { @@ -25,7 +25,8 @@ type DelegationTask struct { SourceAgentID uuid.UUID `json:"source_agent_id"` SourceAgentKey string `json:"source_agent_key"` TargetAgentID uuid.UUID `json:"target_agent_id"` - TargetAgentKey string `json:"target_agent_key"` + TargetAgentKey string `json:"target_agent_key"` + TargetDisplayName string `json:"-"` UserID string `json:"user_id"` Task string `json:"task"` Status string `json:"status"` // "running", "completed", "failed", "cancelled" @@ -53,11 +54,12 @@ type DelegationTask struct { // DelegateOpts configures a single delegation call. type DelegateOpts struct { - TargetAgentKey string - Task string - Context string // optional extra context - Mode string // "sync" (default) or "async" - TeamTaskID uuid.UUID // optional: auto-complete this team task on success + TargetAgentKey string + Task string + Context string // optional extra context + Mode string // "sync" (default) or "async" + TeamTaskID uuid.UUID // optional: auto-complete this team task on success + EstimatedDuration time.Duration // optional: progress notification fires after this delay (default 90s) } // DelegateRunRequest is the request passed to the AgentRunFunc callback. @@ -134,6 +136,7 @@ type DelegateManager struct { active sync.Map // delegationID → *DelegationTask pendingArtifacts sync.Map // sourceAgentID string → *DelegateArtifacts + progressSent sync.Map // "sourceAgentID:chatID" → true (dedup grouped notifications) completedMu sync.Mutex completedSessions []string // session keys pending cleanup } @@ -256,7 +259,11 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) }() // Progress notification timer — fires once after delay if still running - progressTimer := time.AfterFunc(delegationProgressDelay, func() { + progressDelay := opts.EstimatedDuration + if progressDelay <= 0 { + progressDelay = defaultProgressDelay + } + progressTimer := time.AfterFunc(progressDelay, func() { dm.sendProgressNotification(task) }) defer progressTimer.Stop() @@ -303,6 +310,9 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) slog.Info("delegation announce suppressed (siblings still running)", "id", task.ID, "target", task.TargetAgentKey, "siblings", siblingCount) } else { + // Last completion: clear progress dedup so next batch gets fresh notifications. + dm.progressSent.Delete(task.SourceAgentID.String() + ":" + task.OriginChatID) + // Last completion: collect all accumulated artifacts + own result artifacts := dm.collectArtifacts(task.SourceAgentID) if result != nil { @@ -518,7 +528,8 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO SourceAgentID: sourceAgentID, SourceAgentKey: sourceAgent.AgentKey, TargetAgentID: targetAgent.ID, - TargetAgentKey: opts.TargetAgentKey, + TargetAgentKey: opts.TargetAgentKey, + TargetDisplayName: targetAgent.DisplayName, UserID: userID, Task: opts.Task, Status: "running", @@ -584,15 +595,40 @@ 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. +// sendProgressNotification sends a grouped "still working" message listing all +// active delegations from the same source agent. Uses progressSent to dedup — +// only the first timer that fires sends the notification; subsequent timers skip. func (dm *DelegateManager) sendProgressNotification(task *DelegationTask) { - if dm.msgBus == nil || task.OriginChannel == "" || task.OriginChatID == "" { + // Skip internal/delegate channels — only notify on real user-facing channels. + if dm.msgBus == nil || task.OriginChannel == "" || task.OriginChatID == "" || + task.OriginChannel == "delegate" || task.OriginChannel == "system" { return } - elapsed := time.Since(task.CreatedAt).Round(time.Second) - content := fmt.Sprintf("⏳ %s is still working on the task... (%s elapsed)", - task.TargetAgentKey, elapsed) + + // Dedup: one grouped notification per source agent per chat. + dedupKey := task.SourceAgentID.String() + ":" + task.OriginChatID + if _, loaded := dm.progressSent.LoadOrStore(dedupKey, true); loaded { + return + } + + // Collect all active delegations from same source agent. + active := dm.ListActive(task.SourceAgentID) + if len(active) == 0 { + dm.progressSent.Delete(dedupKey) + return + } + + var lines []string + for _, t := range active { + elapsed := time.Since(t.CreatedAt).Round(time.Second) + label := t.TargetAgentKey + if t.TargetDisplayName != "" { + label = fmt.Sprintf("%s (%s)", t.TargetDisplayName, t.TargetAgentKey) + } + lines = append(lines, fmt.Sprintf("- %s: %s", label, elapsed)) + } + + content := fmt.Sprintf("⏳ Your team is working on it...\n%s", strings.Join(lines, "\n")) dm.msgBus.PublishOutbound(bus.OutboundMessage{ Channel: task.OriginChannel, diff --git a/internal/tools/subagent_spawn_tool.go b/internal/tools/subagent_spawn_tool.go index ebe3b074..fdeb19e9 100644 --- a/internal/tools/subagent_spawn_tool.go +++ b/internal/tools/subagent_spawn_tool.go @@ -92,6 +92,10 @@ func (t *SpawnTool) Parameters() map[string]interface{} { "type": "string", "description": "Team task ID to auto-complete when task finishes (for team workflows)", } + props["estimated_duration"] = map[string]interface{}{ + "type": "number", + "description": "Estimated task duration in seconds. A progress notification is sent to the user if the task takes longer than this. Default 90.", + } } return map[string]interface{}{ @@ -220,12 +224,18 @@ func (t *SpawnTool) executeDelegation(ctx context.Context, args map[string]inter teamTaskID, _ = uuid.Parse(ttID) } + var estimatedDuration time.Duration + if ed, ok := args["estimated_duration"].(float64); ok && ed > 0 { + estimatedDuration = time.Duration(ed) * time.Second + } + opts := DelegateOpts{ - TargetAgentKey: agentKey, - Task: task, - Context: extraContext, - Mode: mode, - TeamTaskID: teamTaskID, + TargetAgentKey: agentKey, + Task: task, + Context: extraContext, + Mode: mode, + TeamTaskID: teamTaskID, + EstimatedDuration: estimatedDuration, } if mode == "async" {