fix(teams): prevent leader self-assignment causing dual-session loop

Block team lead from being assigned as task owner across all dispatch
paths. When a leader assigns a task to itself, it gets dispatched to a
team session while still running on the main session — causing race
conditions, duplicate work, and potential infinite announce loops.

Guards added:
- executeCreate: reject assigneeID == team.LeadAgentID
- dispatchTaskToAgent: safety net auto-fail (retry/ticker edge cases)
- DispatchUnblockedTasks: auto-fail leader-owned pending tasks
- ProcessPendingTasks: skip before AssignTask
- executeRetry: explicit error before dispatch
- TeamsMethods.dispatchTaskToAgent: guard for WS/dashboard path

Also removed redundant GetTeamForAgent DB call in dispatchTaskToAgent
by passing *store.TeamData directly from callers.
This commit is contained in:
viettranx committed 2026-03-23 22:57:06 +07:00
1 parent 6e80a1af61
commit 5b749faa35
5 files changed
+68 -9

No files matched your search

+11
View File
@@ -778,6 +778,17 @@ func (m *TeamsMethods) handleTaskDeleteBulk(ctx context.Context, client *gateway
// gateway consumer picks it up and runs the assigned agent, then auto-completes
// the task on success or auto-fails on error.
func (m *TeamsMethods) dispatchTaskToAgent(ctx context.Context, task *store.TeamTaskData, taskID, teamID, agentID uuid.UUID, userID string) {
// Block dispatch to the lead agent — causes dual-session loop.
if team, err := m.teamStore.GetTeam(ctx, teamID); err == nil && team != nil && agentID == team.LeadAgentID {
slog.Warn("teams.tasks.dispatch: blocked dispatch to lead agent",
"task_id", taskID, "agent_id", agentID, "team_id", teamID)
_ = m.teamStore.UpdateTask(ctx, taskID, map[string]any{
"status": store.TeamTaskStatusFailed,
"result": "Cannot dispatch task to the team lead — reassign to a team member",
})
return
}
ag, err := m.agentStore.GetByID(ctx, agentID)
if err != nil {
slog.Warn("teams.tasks.dispatch: cannot resolve agent", "agent_id", agentID, "error", err)
+5 -1
View File
@@ -126,6 +126,10 @@ func (t *TeamTasksTool) executeRetry(ctx context.Context, args map[string]any) *
if task.OwnerAgentID == nil {
return ErrorResult("task has no assignee — assign it first via update")
}
// Block retry to the lead agent — would cause self-dispatch loop.
if *task.OwnerAgentID == team.LeadAgentID {
return ErrorResult("cannot retry task assigned to the team lead — reassign to a team member first via update")
}
// Reset status to pending first (AssignTask only transitions from pending).
if err := t.manager.teamStore.ResetTaskStatus(ctx, taskID, team.ID); err != nil {
@@ -152,7 +156,7 @@ func (t *TeamTasksTool) executeRetry(ctx context.Context, args map[string]any) *
})
// Dispatch immediately (retry is an explicit action, not during a turn).
t.manager.dispatchTaskToAgent(ctx, task, team.ID, *task.OwnerAgentID)
t.manager.dispatchTaskToAgent(ctx, task, team, *task.OwnerAgentID)
assignee := t.manager.agentKeyFromID(ctx, *task.OwnerAgentID)
return NewResult(fmt.Sprintf("Task #%d \"%s\" (id: %s) retried and dispatched to %s. The assignee will receive the task with your recent comments.", task.TaskNumber, task.Subject, taskID, assignee))
+7 -1
View File
@@ -119,6 +119,12 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any)
if !isMember {
return ErrorResult(fmt.Sprintf("agent %q is not a member of this team", assigneeKey))
}
// Prevent lead from self-assigning — causes dual-session execution + loop.
// buildCreateHint already hides the lead from the member list hint,
// but weaker models may still attempt self-assignment.
if assigneeID == team.LeadAgentID {
return ErrorResult("team lead cannot assign tasks to itself — delegate to a team member instead. You are the team lead; handle this work directly or assign to one of your members.")
}
requireApproval, _ := args["require_approval"].(bool)
status := store.TeamTaskStatusPending
@@ -269,7 +275,7 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]any)
ActorType: "system",
ActorID: "fallback_dispatch",
})
t.manager.dispatchTaskToAgent(ctx, task, team.ID, assigneeID)
t.manager.dispatchTaskToAgent(ctx, task, team, assigneeID)
}
}
}
+33 -6
View File
@@ -28,10 +28,24 @@ const maxTaskDispatches = 3
// falling back to ctx only for initial dispatch when the task is created and dispatched
// in the same call. This ensures correct routing even when called from
// DispatchUnblockedTasks (where ctx is the member agent's context, not the lead's).
func (m *TeamToolManager) dispatchTaskToAgent(ctx context.Context, task *store.TeamTaskData, teamID, agentID uuid.UUID) {
func (m *TeamToolManager) dispatchTaskToAgent(ctx context.Context, task *store.TeamTaskData, team *store.TeamData, agentID uuid.UUID) {
if m.msgBus == nil {
return
}
teamID := team.ID
// Safety net: never dispatch to the lead agent — causes dual-session loop.
// Self-assignment is blocked at create time, but catch edge cases
// from retry, ticker recovery, or manual DB edits.
if agentID == team.LeadAgentID {
slog.Warn("team_tasks.dispatch: blocked dispatch to lead agent",
"task_id", task.ID, "agent_id", agentID, "team_id", teamID)
_ = m.teamStore.UpdateTask(ctx, task.ID, map[string]any{
"status": store.TeamTaskStatusFailed,
"result": "Cannot dispatch task to the team lead — reassign to a team member",
})
return
}
// Circuit breaker: auto-fail tasks that have been dispatched too many times.
dispatchCount := 0
@@ -100,10 +114,8 @@ func (m *TeamToolManager) dispatchTaskToAgent(ctx context.Context, task *store.T
}
// Resolve lead agent key for completion announce routing.
fromAgent := ToolAgentKeyFromCtx(ctx)
if team, err := m.teamStore.GetTeamForAgent(ctx, store.AgentIDFromContext(ctx)); err == nil && team != nil {
if leadAg, err := m.cachedGetAgentByID(ctx, team.LeadAgentID); err == nil {
fromAgent = leadAg.AgentKey
}
if leadAg, err := m.cachedGetAgentByID(ctx, team.LeadAgentID); err == nil {
fromAgent = leadAg.AgentKey
}
// Resolve user ID: prefer context (available during leader's turn),
@@ -290,6 +302,10 @@ func (m *TeamToolManager) restoreTraceContext(ctx context.Context, task *store.T
// Called after task completion/cancellation to start newly-unblocked work
// instead of waiting for the ticker (up to 5 min delay).
func (m *TeamToolManager) DispatchUnblockedTasks(ctx context.Context, teamID uuid.UUID) {
team, err := m.teamStore.GetTeam(ctx, teamID)
if err != nil || team == nil {
return
}
tasks, err := m.teamStore.ListRecoverableTasks(ctx, teamID)
if err != nil {
return
@@ -304,6 +320,17 @@ func (m *TeamToolManager) DispatchUnblockedTasks(ctx context.Context, teamID uui
continue
}
ownerID := *task.OwnerAgentID
// Auto-fail tasks assigned to the lead agent — would cause self-dispatch loop.
// Don't just skip: pending lead-owned tasks would stay stuck until stale timeout.
if ownerID == team.LeadAgentID {
slog.Warn("DispatchUnblockedTasks: auto-failing lead-owned task",
"task_id", task.ID, "team_id", teamID)
_ = m.teamStore.UpdateTask(ctx, task.ID, map[string]any{
"status": store.TeamTaskStatusFailed,
"result": "Cannot dispatch task to the team lead — reassign to a team member",
})
continue
}
if dispatched[ownerID] {
continue // skip — this owner already has a higher-priority task dispatched
}
@@ -341,6 +368,6 @@ func (m *TeamToolManager) DispatchUnblockedTasks(ctx context.Context, teamID uui
// Restore leader's trace context (stored in task metadata during creation)
// so the member agent's trace links back to the leader, not the completing member.
dispatchCtx := m.restoreTraceContext(ctx, task)
m.dispatchTaskToAgent(dispatchCtx, task, teamID, ownerID)
m.dispatchTaskToAgent(dispatchCtx, task, team, ownerID)
}
}
+12 -1
View File
@@ -76,6 +76,13 @@ func (m *TeamToolManager) ProcessPendingTasks(ctx context.Context, teamID uuid.U
m.failCycledTasks(ctx, teamID, cycled, taskMap)
}
// Fetch team once for lead-agent guard in dispatchTaskToAgent.
team, err := m.teamStore.GetTeam(ctx, teamID)
if err != nil || team == nil {
slog.Warn("post_turn: cannot fetch team", "team_id", teamID, "error", err)
return fmt.Errorf("cannot fetch team %s: %w", teamID, err)
}
// Dispatch pending assigned tasks (not blocked, not failed).
for _, task := range tasks {
if _, isCycled := cycled[task.ID]; isCycled {
@@ -87,6 +94,10 @@ func (m *TeamToolManager) ProcessPendingTasks(ctx context.Context, teamID uuid.U
if task.Status != store.TeamTaskStatusPending || task.OwnerAgentID == nil {
continue
}
// Skip tasks assigned to the lead agent — would cause self-dispatch loop.
if *task.OwnerAgentID == team.LeadAgentID {
continue
}
if err := m.teamStore.AssignTask(ctx, task.ID, *task.OwnerAgentID, teamID); err != nil {
slog.Warn("post_turn: assign failed", "task_id", task.ID, "error", err)
continue
@@ -107,7 +118,7 @@ func (m *TeamToolManager) ProcessPendingTasks(ctx context.Context, teamID uuid.U
// Restore leader's trace context from task metadata (ctx here is the
// consumer goroutine context which has no trace after the turn ends).
dispatchCtx := m.restoreTraceContext(ctx, task)
m.dispatchTaskToAgent(dispatchCtx, task, teamID, *task.OwnerAgentID)
m.dispatchTaskToAgent(dispatchCtx, task, team, *task.OwnerAgentID)
}
slog.Info("post_turn: processed pending tasks",