mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 02:15:57 +00:00
feat(teams): add CancelTask with dependency unblocking and delegation cancellation
- Add CancelTask store method: guards against completed tasks, unblocks dependents, transitions blocked→pending - Cancel running delegation when team task is cancelled via CancelByTeamTaskID - Prevent delegate agents from completing/cancelling tasks directly - Wire DelegateManager into TeamToolManager for task-delegation lifecycle - Remove BOOTSTRAP.md from predefined agent seed (predefined agents already have full context) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
1 parent
1841a7097c
commit
4da116292a
7 files changed
+111
-14
No files matched your search
@@ -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))
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user