From d844795e070f065334c506a8b1b24d2356b94083 Mon Sep 17 00:00:00 2001 From: minhdang03 Date: Sat, 18 Jul 2026 14:49:27 +1000 Subject: [PATCH] fix(teams): recover task locks for all active teams, not only v2 (#1448) Task locking (AssignTask/ClaimTask) applies to every active team with no version check, but the recovery/stale/followup queries only ran for teams whose settings.version >= 2. Any team without that flag accumulated permanently-stuck in_progress tasks: locks were set on dispatch but never reclaimed once they expired, so the task ticker skipped them forever (including the startup force-recover). Align the maintenance scope with the locking scope by filtering recovery, stale-marking, and follow-ups to active teams regardless of settings version. Applies to both the Postgres and SQLite backends; renames the shared JOIN constant v2ActiveTeamJoin -> activeTeamJoin to match. --- internal/store/pg/teams_tasks_followup.go | 2 +- internal/store/pg/teams_tasks_progress.go | 22 +++++++------------ .../store/sqlitestore/teams_tasks_activity.go | 15 ++++--------- 3 files changed, 13 insertions(+), 26 deletions(-) diff --git a/internal/store/pg/teams_tasks_followup.go b/internal/store/pg/teams_tasks_followup.go index 72bf8a6c..2087839e 100644 --- a/internal/store/pg/teams_tasks_followup.go +++ b/internal/store/pg/teams_tasks_followup.go @@ -52,7 +52,7 @@ func (s *PGTeamStore) ListAllFollowupDueTasks(ctx context.Context) ([]store.Team rows, err := s.db.QueryContext(ctx, `SELECT `+taskSelectCols+` `+taskJoinClause+` - `+v2ActiveTeamJoin+` + `+activeTeamJoin+` WHERE t.followup_at IS NOT NULL AND t.followup_at <= $1 AND t.status = $2 diff --git a/internal/store/pg/teams_tasks_progress.go b/internal/store/pg/teams_tasks_progress.go index fcd172e3..d6c2c9ef 100644 --- a/internal/store/pg/teams_tasks_progress.go +++ b/internal/store/pg/teams_tasks_progress.go @@ -71,15 +71,14 @@ func (s *PGTeamStore) RenewTaskLock(ctx context.Context, taskID, teamID uuid.UUI } // ============================================================ -// Stale recovery (batch — all v2 active teams in one query) +// Stale recovery (batch — all active teams in one query) // ============================================================ -// v2ActiveTeamJoin is the JOIN clause that filters to v2 active teams. -const v2ActiveTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id - AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2` +// activeTeamJoin is the JOIN clause that filters to active teams. +const activeTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id + AND tm.status = 'active'` -// RecoverAllStaleTasks resets in_progress tasks with expired locks across all v2 active teams. +// RecoverAllStaleTasks resets in_progress tasks with expired locks across all active teams. func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -89,7 +88,6 @@ func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Recover followup_channel = NULL, followup_chat_id = NULL, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 AND t.lock_expires_at IS NOT NULL AND t.lock_expires_at < $2 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, @@ -102,7 +100,7 @@ func (s *PGTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Recover return scanRecoveredTaskInfoRows(rows) } -// ForceRecoverAllTasks resets ALL in_progress tasks across v2 active teams (startup). +// ForceRecoverAllTasks resets ALL in_progress tasks across active teams (startup). func (s *PGTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -112,7 +110,6 @@ func (s *PGTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.Recover followup_channel = NULL, followup_chat_id = NULL, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, store.TeamTaskStatusPending, now, store.TeamTaskStatusInProgress, @@ -148,7 +145,7 @@ func (s *PGTeamStore) ListRecoverableTasks(ctx context.Context, teamID uuid.UUID return scanTaskRowsJoined(rows) } -// MarkAllStaleTasks marks pending tasks older than olderThan as stale across all v2 active teams. +// MarkAllStaleTasks marks pending tasks older than olderThan as stale across all active teams. func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -156,7 +153,6 @@ func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time SET status = $1, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 AND t.updated_at < $4 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, store.TeamTaskStatusStale, now, store.TeamTaskStatusPending, olderThan, @@ -168,7 +164,7 @@ func (s *PGTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time.Time return scanRecoveredTaskInfoRows(rows) } -// MarkInReviewStaleTasks marks in_review tasks older than olderThan as stale across all v2 active teams. +// MarkInReviewStaleTasks marks in_review tasks older than olderThan as stale across all active teams. func (s *PGTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan time.Time) ([]store.RecoveredTaskInfo, error) { now := time.Now() rows, err := s.db.QueryContext(ctx, @@ -176,7 +172,6 @@ func (s *PGTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan time SET status = $1, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = $3 AND t.updated_at < $4 RETURNING t.id, t.team_id, t.tenant_id, t.task_number, t.subject, COALESCE(t.channel, ''), COALESCE(t.chat_id, '')`, store.TeamTaskStatusStale, now, store.TeamTaskStatusInReview, olderThan, @@ -197,7 +192,6 @@ func (s *PGTeamStore) FixOrphanedBlockedTasks(ctx context.Context) ([]store.Reco SET blocked_by = '{}', status = $1, updated_at = $2 FROM agent_teams tm WHERE t.team_id = tm.id AND tm.status = 'active' - AND COALESCE((tm.settings->>'version')::int, 0) >= 2 AND t.status = 'blocked' AND array_length(t.blocked_by, 1) > 0 AND NOT EXISTS ( diff --git a/internal/store/sqlitestore/teams_tasks_activity.go b/internal/store/sqlitestore/teams_tasks_activity.go index c76f1dfa..90ecf76e 100644 --- a/internal/store/sqlitestore/teams_tasks_activity.go +++ b/internal/store/sqlitestore/teams_tasks_activity.go @@ -352,11 +352,10 @@ func (s *SQLiteTeamStore) RenewTaskLock(ctx context.Context, taskID, teamID uuid // Stale recovery // ============================================================ -// v2ActiveTeamJoin is the JOIN that filters to v2 active teams. +// activeTeamJoin is the JOIN that filters to active teams. // SQLite uses json_extract instead of PG's ->> operator. -const v2ActiveTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id - AND tm.status = 'active' - AND COALESCE(CAST(json_extract(tm.settings, '$.version') AS INTEGER), 0) >= 2` +const activeTeamJoin = `JOIN agent_teams tm ON tm.id = t.team_id + AND tm.status = 'active'` func (s *SQLiteTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.RecoveredTaskInfo, error) { now := time.Now() @@ -368,7 +367,6 @@ func (s *SQLiteTeamStore) RecoverAllStaleTasks(ctx context.Context) ([]store.Rec WHERE status = ? AND lock_expires_at IS NOT NULL AND lock_expires_at < ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusPending, now, store.TeamTaskStatusInProgress, now, ) @@ -387,7 +385,6 @@ func (s *SQLiteTeamStore) ForceRecoverAllTasks(ctx context.Context) ([]store.Rec WHERE status = ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusPending, now, store.TeamTaskStatusInProgress, ) @@ -426,7 +423,6 @@ func (s *SQLiteTeamStore) MarkAllStaleTasks(ctx context.Context, olderThan time. WHERE status = ? AND updated_at < ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusStale, now, store.TeamTaskStatusPending, olderThan, ) @@ -443,7 +439,6 @@ func (s *SQLiteTeamStore) MarkInReviewStaleTasks(ctx context.Context, olderThan WHERE status = ? AND updated_at < ? AND team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 )`, store.TeamTaskStatusStale, now, store.TeamTaskStatusInReview, olderThan, ) @@ -462,7 +457,6 @@ func (s *SQLiteTeamStore) FixOrphanedBlockedTasks(ctx context.Context) ([]store. COALESCE(t.channel, ''), COALESCE(t.chat_id, '') FROM team_tasks t JOIN agent_teams tm ON tm.id = t.team_id AND tm.status = 'active' - AND COALESCE(CAST(json_extract(tm.settings, '$.version') AS INTEGER), 0) >= 2 WHERE t.status = 'blocked' AND t.blocked_by IS NOT NULL AND t.blocked_by != '[]' AND NOT EXISTS ( @@ -501,7 +495,6 @@ func (s *SQLiteTeamStore) queryRecoveredTasks(ctx context.Context, status string AND t.updated_at >= ? AND t.team_id IN ( SELECT id FROM agent_teams WHERE status = 'active' - AND COALESCE(CAST(json_extract(settings, '$.version') AS INTEGER), 0) >= 2 ) ORDER BY t.updated_at DESC LIMIT 200`, @@ -599,7 +592,7 @@ func (s *SQLiteTeamStore) ListAllFollowupDueTasks(ctx context.Context) ([]store. rows, err := s.db.QueryContext(ctx, `SELECT `+taskSelectCols+` `+taskJoinClause+` - `+v2ActiveTeamJoin+` + `+activeTeamJoin+` WHERE t.followup_at IS NOT NULL AND t.followup_at <= ? AND t.status = ?