diff --git a/internal/agent/loop.go b/internal/agent/loop.go index d57fcd45..40a0383c 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -107,6 +107,9 @@ type Loop struct { // Group writer cache for system prompt injection (managed mode) groupWriterCache *store.GroupWriterCache + + // Team store for cross-session pending task detection (managed mode) + teamStore store.TeamStore } // AgentEvent is emitted during agent execution for WS broadcasting. @@ -175,6 +178,9 @@ type LoopConfig struct { // Group writer cache for system prompt injection (managed mode) GroupWriterCache *store.GroupWriterCache + + // Team store for cross-session pending task detection (managed mode) + TeamStore store.TeamStore } func NewLoop(cfg LoopConfig) *Loop { @@ -235,6 +241,7 @@ func NewLoop(cfg LoopConfig) *Loop { builtinToolSettings: cfg.BuiltinToolSettings, thinkingLevel: cfg.ThinkingLevel, groupWriterCache: cfg.GroupWriterCache, + teamStore: cfg.TeamStore, } } @@ -530,6 +537,33 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } } + // 2b. Cross-session recovery: notify team leads about orphaned pending tasks. + // 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", "", ""); err == nil { + var stale []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 len(stale) > 0 { + reminder := fmt.Sprintf( + "[System] 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")) + messages = append(messages, + providers.Message{Role: "user", Content: reminder}, + providers.Message{Role: "assistant", Content: "I see the pending tasks. Let me handle them."}, + ) + } + } + } + } + // 3. Buffer new messages — write to session only AFTER the run completes. // This prevents concurrent runs from seeing each other's in-progress messages. // NOTE: pendingMsgs stores TEXT ONLY (no images) to avoid bloating session storage. diff --git a/internal/agent/resolver.go b/internal/agent/resolver.go index 70e48a68..bcab647b 100644 --- a/internal/agent/resolver.go +++ b/internal/agent/resolver.go @@ -297,6 +297,7 @@ func NewManagedResolver(deps ResolverDeps) ResolverFunc { BuiltinToolSettings: builtinSettings, ThinkingLevel: ag.ParseThinkingLevel(), GroupWriterCache: deps.GroupWriterCache, + TeamStore: deps.TeamStore, }) slog.Info("resolved agent from DB", "agent", agentKey, "model", ag.Model, "provider", ag.Provider) @@ -459,7 +460,8 @@ func buildTeamMD(team *store.TeamData, members []store.TeamMemberData, selfID uu sb.WriteString("- action=list, status=all → all tasks including completed\n") sb.WriteString("- action=get, task_id= → full task detail with result\n") sb.WriteString("- action=search, query= → search tasks by subject/description\n") - sb.WriteString("- action=complete, task_id=, result= → manually complete a task\n\n") + sb.WriteString("- action=complete, task_id=, result= → manually complete a task\n") + sb.WriteString("- action=cancel, task_id=, reason= → cancel a pending task that is no longer needed\n\n") sb.WriteString("Use `team_message` to send updates to team members.\n\n") sb.WriteString("For simple questions about team composition, answer directly from the member list above.\n") } else { diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index 104518b1..6f5f647d 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -388,43 +388,36 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO return nil, nil, err } + // Resolve team once — used for task enforcement, validation, and access checks. + var team *store.TeamData + if dm.teamStore != nil { + team, _ = dm.teamStore.GetTeamForAgent(ctx, sourceAgentID) + } + // Enforce team_task_id for team members: every delegation must be tracked. - if dm.teamStore != nil && opts.TeamTaskID == uuid.Nil { - if team, _ := dm.teamStore.GetTeamForAgent(ctx, sourceAgentID); team != nil { - // List pending tasks so the LLM can retry with a correct task_id - // (common case: LLM called team_tasks create + spawn in parallel, - // hallucinated the task_id, uuid.Parse failed → uuid.Nil). - hint := "" - if tasks, err := dm.teamStore.ListTasks(ctx, team.ID, "newest", "", ""); err == nil { - var pendingIDs []string - for _, t := range tasks { - if t.Status == store.TeamTaskStatusPending { - pendingIDs = append(pendingIDs, fmt.Sprintf("%s (%s)", t.ID, t.Subject)) - } - } - if len(pendingIDs) > 0 { - hint = fmt.Sprintf(" Pending tasks you already created: %s", strings.Join(pendingIDs, ", ")) - } - } - return nil, nil, fmt.Errorf( - "spawn requires a valid team_task_id (UUID). "+ - "Create a task first with team_tasks action=create, "+ - "then pass the returned task id as team_task_id.%s", - hint) - } + if team != nil && opts.TeamTaskID == uuid.Nil { + hint := dm.pendingTasksHint(ctx, team.ID) + return nil, nil, fmt.Errorf( + "spawn requires a valid team_task_id (UUID). "+ + "Create a task first with team_tasks action=create, "+ + "then pass the returned task id as team_task_id.%s", + hint) } // Validate that team_task_id belongs to the agent's team (prevent cross-team task completion). if dm.teamStore != nil && opts.TeamTaskID != uuid.Nil { teamTask, err := dm.teamStore.GetTask(ctx, opts.TeamTaskID) if err != nil { - return nil, nil, fmt.Errorf("team_task_id not found: %w", err) + hint := "" + if team != nil { + hint = dm.pendingTasksHint(ctx, team.ID) + } + return nil, nil, fmt.Errorf("team_task_id %s not found.%s", opts.TeamTaskID, hint) } - if team, _ := dm.teamStore.GetTeamForAgent(ctx, sourceAgentID); team != nil { + if team != nil { if teamTask.TeamID != team.ID { return nil, nil, fmt.Errorf("team_task_id does not belong to your team") } - // Check team access for the end-user/channel userID := store.UserIDFromContext(ctx) channel := ToolChannelFromCtx(ctx) if err := checkTeamAccess(team.Settings, userID, channel); err != nil { @@ -440,6 +433,12 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO "description": opts.Task, }) } + + // Claim task early so status moves to in_progress immediately. + // This prevents the pending reminder from re-triggering spawns for + // tasks that are already running. The ClaimTask in autoCompleteTeamTask() + // will harmlessly fail (WHERE status='pending' won't match). + _ = dm.teamStore.ClaimTask(ctx, opts.TeamTaskID, targetAgent.ID, teamTask.TeamID) } linkCount := dm.ActiveCountForLink(sourceAgentID, targetAgent.ID) @@ -498,6 +497,25 @@ func buildDelegateMessage(opts DelegateOpts) string { return opts.Task } +// pendingTasksHint returns a formatted hint listing pending team tasks, +// so the LLM can self-correct a hallucinated team_task_id. +func (dm *DelegateManager) pendingTasksHint(ctx context.Context, teamID uuid.UUID) string { + tasks, err := dm.teamStore.ListTasks(ctx, teamID, "newest", "", "") + if err != nil { + return "" + } + var ids []string + for _, t := range tasks { + if t.Status == store.TeamTaskStatusPending { + ids = append(ids, fmt.Sprintf("%s (%s)", t.ID, t.Subject)) + } + } + if len(ids) == 0 { + return "" + } + return fmt.Sprintf(" Pending tasks: %s", strings.Join(ids, ", ")) +} + func (dm *DelegateManager) buildRunRequest(task *DelegationTask, message string) DelegateRunRequest { return DelegateRunRequest{ SessionKey: task.SessionKey, diff --git a/internal/tools/team_tasks_tool.go b/internal/tools/team_tasks_tool.go index af863808..379fa36f 100644 --- a/internal/tools/team_tasks_tool.go +++ b/internal/tools/team_tasks_tool.go @@ -24,7 +24,7 @@ func NewTeamTasksTool(manager *TeamToolManager) *TeamTasksTool { func (t *TeamTasksTool) Name() string { return "team_tasks" } func (t *TeamTasksTool) Description() string { - return "Manage the shared team task list. Actions: list (active tasks overview), get (full task detail with result), create, claim, complete, search. See TEAM.md for your team context." + return "Manage the shared team task list. Actions: list (active tasks overview), get (full task detail with result), create, claim, complete, cancel, search. See TEAM.md for your team context." } func (t *TeamTasksTool) Parameters() map[string]interface{} { @@ -33,7 +33,7 @@ func (t *TeamTasksTool) Parameters() map[string]interface{} { "properties": map[string]interface{}{ "action": map[string]interface{}{ "type": "string", - "description": "'list', 'get', 'create', 'claim', 'complete', or 'search'", + "description": "'list', 'get', 'create', 'claim', 'complete', 'cancel', or 'search'", }, "status": map[string]interface{}{ "type": "string", @@ -62,12 +62,16 @@ func (t *TeamTasksTool) Parameters() map[string]interface{} { }, "task_id": map[string]interface{}{ "type": "string", - "description": "Task ID (required for action=get, claim, complete)", + "description": "Task ID (required for action=get, claim, complete, cancel)", }, "result": map[string]interface{}{ "type": "string", "description": "Task result summary (required for action=complete)", }, + "reason": map[string]interface{}{ + "type": "string", + "description": "Cancellation reason (optional for action=cancel)", + }, }, "required": []string{"action"}, } @@ -87,10 +91,12 @@ func (t *TeamTasksTool) Execute(ctx context.Context, args map[string]interface{} return t.executeClaim(ctx, args) case "complete": return t.executeComplete(ctx, args) + case "cancel": + return t.executeCancel(ctx, args) case "search": return t.executeSearch(ctx, args) default: - return ErrorResult(fmt.Sprintf("unknown action: %s (use list, get, create, claim, complete, or search)", action)) + return ErrorResult(fmt.Sprintf("unknown action: %s (use list, get, create, claim, complete, cancel, or search)", action)) } } @@ -334,3 +340,38 @@ func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]int return NewResult(fmt.Sprintf("Task %s completed. Dependent tasks have been unblocked.", taskIDStr)) } + +func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]interface{}) *Result { + team, _, err := t.manager.resolveTeam(ctx) + if err != nil { + return ErrorResult(err.Error()) + } + + taskIDStr, _ := args["task_id"].(string) + if taskIDStr == "" { + return ErrorResult("task_id is required for cancel action") + } + taskID, err := uuid.Parse(taskIDStr) + if err != nil { + return ErrorResult("invalid task_id") + } + + reason, _ := args["reason"].(string) + if reason == "" { + reason = "Cancelled: task was not executed" + } + + if err := t.manager.teamStore.UpdateTask(ctx, taskID, map[string]any{ + "status": store.TeamTaskStatusCompleted, + "result": "CANCELLED: " + reason, + }); err != nil { + return ErrorResult("failed to cancel task: " + err.Error()) + } + + t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, map[string]string{ + "team_id": team.ID.String(), + "task_id": taskIDStr, + }) + + return NewResult(fmt.Sprintf("Task %s cancelled.", taskIDStr)) +}