mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 16:17:25 +00:00
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>
This commit is contained in:
1 parent
10361b36f8
commit
71da3d7889
2 files changed
+49
-5
No files matched your search
+21
-5
@@ -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."},
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user