diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index eb23839e..791cbea9 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -386,6 +386,9 @@ func wireManagedExtras( // Register team tools (team_tasks + team_message) if team store is available. if stores.Teams != nil && stores.Agents != nil { teamMgr := tools.NewTeamToolManager(stores.Teams, stores.Agents, msgBus) + if delegateMgr != nil { + teamMgr.SetDelegateManager(delegateMgr) + } toolsReg.Register(tools.NewTeamTasksTool(teamMgr)) toolsReg.Register(tools.NewTeamMessageTool(teamMgr)) diff --git a/internal/bootstrap/seed_store.go b/internal/bootstrap/seed_store.go index f0245a73..74d17a6f 100644 --- a/internal/bootstrap/seed_store.go +++ b/internal/bootstrap/seed_store.go @@ -81,10 +81,10 @@ var userSeedFilesOpen = []string{ } // userSeedFilesPredefined is the set of files seeded per-user for predefined agents. -// Includes BOOTSTRAP.md for user onboarding (uses BOOTSTRAP_PREDEFINED.md template). +// Only USER.md — predefined agents already have full context (SOUL.md, IDENTITY.md, AGENTS.md) +// and don't need a bootstrap onboarding ritual. They just need to learn the user's profile. var userSeedFilesPredefined = []string{ UserFile, - BootstrapFile, } // SeedUserFiles seeds embedded templates into user_context_files for a new user. diff --git a/internal/store/pg/teams_tasks.go b/internal/store/pg/teams_tasks.go index 7f65bbd9..17c6b2b0 100644 --- a/internal/store/pg/teams_tasks.go +++ b/internal/store/pg/teams_tasks.go @@ -159,9 +159,12 @@ func (s *PGTeamStore) CompleteTask(ctx context.Context, taskID, teamID uuid.UUID } // Unblock dependent tasks: remove this taskID from their blocked_by arrays. - // Tasks with empty blocked_by after removal become claimable. + // Tasks whose blocked_by becomes empty transition from blocked→pending so they can be claimed. _, err = tx.ExecContext(ctx, - `UPDATE team_tasks SET blocked_by = array_remove(blocked_by, $1), updated_at = $2 + `UPDATE team_tasks SET + blocked_by = array_remove(blocked_by, $1), + status = CASE WHEN status = 'blocked' AND blocked_by = ARRAY[$1]::uuid[] THEN 'pending' ELSE status END, + updated_at = $2 WHERE $1 = ANY(blocked_by)`, taskID, time.Now(), ) @@ -172,6 +175,48 @@ func (s *PGTeamStore) CompleteTask(ctx context.Context, taskID, teamID uuid.UUID return tx.Commit() } +func (s *PGTeamStore) CancelTask(ctx context.Context, taskID, teamID uuid.UUID, reason string) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + // Cancel the task (guard: only non-completed tasks) + now := time.Now() + res, err := tx.ExecContext(ctx, + `UPDATE team_tasks SET status = $1, result = $2, updated_at = $3 + WHERE id = $4 AND status != $5 AND team_id = $6`, + store.TeamTaskStatusCompleted, "CANCELLED: "+reason, now, + taskID, store.TeamTaskStatusCompleted, teamID, + ) + if err != nil { + return err + } + n, err := res.RowsAffected() + if err != nil { + return err + } + if n == 0 { + return fmt.Errorf("task not found, already completed, or wrong team") + } + + // Unblock dependent tasks (same logic as CompleteTask) + _, err = tx.ExecContext(ctx, + `UPDATE team_tasks SET + blocked_by = array_remove(blocked_by, $1), + status = CASE WHEN status = 'blocked' AND blocked_by = ARRAY[$1]::uuid[] THEN 'pending' ELSE status END, + updated_at = $2 + WHERE $1 = ANY(blocked_by)`, + taskID, now, + ) + if err != nil { + return err + } + + return tx.Commit() +} + func scanTaskRowsJoined(rows *sql.Rows) ([]store.TeamTaskData, error) { var tasks []store.TeamTaskData for rows.Next() { diff --git a/internal/store/team_store.go b/internal/store/team_store.go index 060b6440..318076c6 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -197,6 +197,11 @@ type TeamStore interface { // teamID is validated in the WHERE clause to prevent cross-team task completion. CompleteTask(ctx context.Context, taskID, teamID uuid.UUID, result string) error + // CancelTask marks a non-completed task as cancelled (status=completed, result="CANCELLED: ..."), + // unblocks dependent tasks, and transitions blocked→pending when all blockers are resolved. + // Returns error if the task is already completed or not found. + CancelTask(ctx context.Context, taskID, teamID uuid.UUID, reason string) error + // Delegation history SaveDelegationHistory(ctx context.Context, record *DelegationHistoryData) error ListDelegationHistory(ctx context.Context, opts DelegationHistoryListOpts) ([]DelegationHistoryData, int, error) diff --git a/internal/tools/delegate_state.go b/internal/tools/delegate_state.go index 4aadfb92..5a8bbd42 100644 --- a/internal/tools/delegate_state.go +++ b/internal/tools/delegate_state.go @@ -54,6 +54,31 @@ func (dm *DelegateManager) Cancel(delegationID string) bool { return true } +// CancelByTeamTaskID cancels a running delegation associated with a team task. +// Returns true if a delegation was found and cancelled. +func (dm *DelegateManager) CancelByTeamTaskID(teamTaskID uuid.UUID) bool { + found := false + dm.active.Range(func(key, val any) bool { + t := val.(*DelegationTask) + if t.TeamTaskID == teamTaskID && t.Status == "running" { + if t.cancelFunc != nil { + t.cancelFunc() + } + t.Status = "cancelled" + now := time.Now() + t.CompletedAt = &now + dm.active.Delete(key) + dm.emitEvent("delegation.cancelled", t) + slog.Info("delegation cancelled by team task cancel", + "delegation_id", t.ID, "team_task_id", teamTaskID, "target", t.TargetAgentKey) + found = true + return false // stop iteration + } + return true + }) + return found +} + // ListActive returns all active delegations for a source agent. func (dm *DelegateManager) ListActive(sourceAgentID uuid.UUID) []*DelegationTask { var tasks []*DelegationTask diff --git a/internal/tools/team_tasks_tool.go b/internal/tools/team_tasks_tool.go index 379fa36f..bf512fcd 100644 --- a/internal/tools/team_tasks_tool.go +++ b/internal/tools/team_tasks_tool.go @@ -305,6 +305,11 @@ func (t *TeamTasksTool) executeClaim(ctx context.Context, args map[string]interf } func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]interface{}) *Result { + // Delegate agents cannot complete tasks — autoCompleteTeamTask handles it. + if ToolChannelFromCtx(ctx) == "delegate" { + return ErrorResult("delegate agents cannot complete team tasks directly — results are auto-completed when delegation finishes") + } + team, agentID, err := t.manager.resolveTeam(ctx) if err != nil { return ErrorResult(err.Error()) @@ -342,6 +347,11 @@ func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]int } func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]interface{}) *Result { + // Delegate agents cannot cancel tasks — only lead/user-facing agents can. + if ToolChannelFromCtx(ctx) == "delegate" { + return ErrorResult("delegate agents cannot cancel team tasks directly") + } + team, _, err := t.manager.resolveTeam(ctx) if err != nil { return ErrorResult(err.Error()) @@ -358,20 +368,23 @@ func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]inter reason, _ := args["reason"].(string) if reason == "" { - reason = "Cancelled: task was not executed" + reason = "Cancelled by agent" } - if err := t.manager.teamStore.UpdateTask(ctx, taskID, map[string]any{ - "status": store.TeamTaskStatusCompleted, - "result": "CANCELLED: " + reason, - }); err != nil { + // CancelTask: guards against completed tasks, unblocks dependents, transitions blocked→pending. + if err := t.manager.teamStore.CancelTask(ctx, taskID, team.ID, reason); err != nil { return ErrorResult("failed to cancel task: " + err.Error()) } + // Cancel any running delegation for this task. + if t.manager.delegateMgr != nil { + t.manager.delegateMgr.CancelByTeamTaskID(taskID) + } + t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, map[string]string{ "team_id": team.ID.String(), "task_id": taskIDStr, }) - return NewResult(fmt.Sprintf("Task %s cancelled.", taskIDStr)) + return NewResult(fmt.Sprintf("Task %s cancelled. Any running delegation has been stopped and dependent tasks unblocked.", taskIDStr)) } diff --git a/internal/tools/team_tool_manager.go b/internal/tools/team_tool_manager.go index 0fd30c6e..31bf924f 100644 --- a/internal/tools/team_tool_manager.go +++ b/internal/tools/team_tool_manager.go @@ -25,16 +25,22 @@ type teamCacheEntry struct { // the team store, agent store, and message bus. // Includes a TTL cache for team data to avoid DB queries on every tool call. type TeamToolManager struct { - teamStore store.TeamStore - agentStore store.AgentStore - msgBus *bus.MessageBus - teamCache sync.Map // agentID (uuid.UUID) → *teamCacheEntry + teamStore store.TeamStore + agentStore store.AgentStore + msgBus *bus.MessageBus + delegateMgr *DelegateManager // optional: enables delegation cancellation on task cancel + teamCache sync.Map // agentID (uuid.UUID) → *teamCacheEntry } func NewTeamToolManager(teamStore store.TeamStore, agentStore store.AgentStore, msgBus *bus.MessageBus) *TeamToolManager { return &TeamToolManager{teamStore: teamStore, agentStore: agentStore, msgBus: msgBus} } +// SetDelegateManager enables delegation cancellation when team tasks are cancelled. +func (m *TeamToolManager) SetDelegateManager(dm *DelegateManager) { + m.delegateMgr = dm +} + // resolveTeam returns the team that the calling agent belongs to. // Uses a TTL cache to avoid repeated DB queries. Access control // (user/channel) is checked on every call regardless of cache hit.