From 27f2c0d80a52dd05e2535cc0d984331e3ae1831b Mon Sep 17 00:00:00 2001 From: viettranx Date: Sat, 7 Mar 2026 23:20:01 +0700 Subject: [PATCH] fix(delegation): auto-fail team task when async delegation fails When an async delegation failed, the team task stayed in_progress forever because only the success path called autoCompleteTeamTask(). Now autoFailTeamTask() marks the team task as failed, stores the error message, unblocks dependents, and logs an audit trail message. --- internal/store/pg/teams_tasks.go | 41 ++++++++++++++++++++++++++++++++ internal/store/team_store.go | 5 ++++ internal/tools/delegate_async.go | 2 ++ internal/tools/delegate_state.go | 39 ++++++++++++++++++++++++++++++ 4 files changed, 87 insertions(+) diff --git a/internal/store/pg/teams_tasks.go b/internal/store/pg/teams_tasks.go index 17c6b2b0..10badeeb 100644 --- a/internal/store/pg/teams_tasks.go +++ b/internal/store/pg/teams_tasks.go @@ -217,6 +217,47 @@ func (s *PGTeamStore) CancelTask(ctx context.Context, taskID, teamID uuid.UUID, return tx.Commit() } +func (s *PGTeamStore) FailTask(ctx context.Context, taskID, teamID uuid.UUID, errMsg string) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + 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.TeamTaskStatusFailed, "FAILED: "+errMsg, now, + taskID, store.TeamTaskStatusInProgress, teamID, + ) + if err != nil { + return err + } + n, err := res.RowsAffected() + if err != nil { + return err + } + if n == 0 { + return fmt.Errorf("task not in progress or not found") + } + + // Unblock dependent tasks (same logic as CompleteTask/CancelTask) + _, 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 318076c6..b486f4f4 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -26,6 +26,7 @@ const ( TeamTaskStatusInProgress = "in_progress" TeamTaskStatusCompleted = "completed" TeamTaskStatusBlocked = "blocked" + TeamTaskStatusFailed = "failed" ) // Team task list filter constants (for ListTasks statusFilter parameter). @@ -202,6 +203,10 @@ type TeamStore interface { // Returns error if the task is already completed or not found. CancelTask(ctx context.Context, taskID, teamID uuid.UUID, reason string) error + // FailTask marks an in_progress task as failed and stores the error message. + // Unblocks dependent tasks so they aren't stuck. + FailTask(ctx context.Context, taskID, teamID uuid.UUID, errMsg 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_async.go b/internal/tools/delegate_async.go index e430b8d2..48ed8106 100644 --- a/internal/tools/delegate_async.go +++ b/internal/tools/delegate_async.go @@ -214,12 +214,14 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if runErr != nil { task.Status = "failed" + dm.autoFailTeamTask(task, runErr.Error()) dm.emitDelegationEventWithError(task, runErr) dm.saveDelegationHistory(task, "", runErr, duration) } else { // Apply quality gates before marking completed. if result, runErr = dm.applyQualityGates(taskCtx, task, opts, result); runErr != nil { task.Status = "failed" + dm.autoFailTeamTask(task, runErr.Error()) dm.emitDelegationEventWithError(task, runErr) dm.saveDelegationHistory(task, "", runErr, duration) } else { diff --git a/internal/tools/delegate_state.go b/internal/tools/delegate_state.go index 08adb577..9ce83b85 100644 --- a/internal/tools/delegate_state.go +++ b/internal/tools/delegate_state.go @@ -222,6 +222,45 @@ func (dm *DelegateManager) autoCompleteTeamTask(task *DelegationTask, resultCont } } +// autoFailTeamTask marks the associated team task as failed. +// Called after a delegation fails. Errors are logged but not fatal. +func (dm *DelegateManager) autoFailTeamTask(task *DelegationTask, errMsg string) { + if dm.teamStore == nil || task.TeamTaskID == uuid.Nil { + return + } + + // Truncate error message for storage + if len(errMsg) > 2000 { + errMsg = errMsg[:2000] + "..." + } + + _ = dm.teamStore.ClaimTask(context.Background(), task.TeamTaskID, task.TargetAgentID, task.TeamID) + if err := dm.teamStore.FailTask(context.Background(), task.TeamTaskID, task.TeamID, errMsg); err != nil { + slog.Warn("delegate: failed to auto-fail team task", + "task_id", task.TeamTaskID, "delegation_id", task.ID, "error", err) + } else { + slog.Info("delegate: auto-failed team task", + "task_id", task.TeamTaskID, "delegation_id", task.ID) + + // Persist delegation failure as team message for audit trail + if task.TeamID != uuid.Nil { + summary := errMsg + if len(summary) > 500 { + summary = summary[:500] + "..." + } + taskID := task.TeamTaskID + _ = dm.teamStore.SendMessage(context.Background(), &store.TeamMessageData{ + TeamID: task.TeamID, + FromAgentID: task.TargetAgentID, + ToAgentID: &task.SourceAgentID, + Content: fmt.Sprintf("[Delegation failed] %s", summary), + MessageType: store.TeamMessageTypeChat, + TaskID: &taskID, + }) + } + } +} + // saveDelegationHistory persists a delegation record to the database. // Called after delegation completes (success, fail, or cancel). Errors are logged, not fatal. func (dm *DelegateManager) saveDelegationHistory(task *DelegationTask, resultContent string, delegateErr error, duration time.Duration) {