mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 20:17:20 +00:00
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.
This commit is contained in:
1 parent
9897dd77ed
commit
27f2c0d80a
4 files changed
+87
No files matched your search
@@ -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() {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user