tiennm99/goclaw
1
0 Fork 0
mirror of https://github.com/tiennm99/goclaw.git synced 2026-10-20 00:15:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(teams): recover task locks for all active teams, not only v2 (#1448)

Browse Source 
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.
This commit is contained in:
minhdang03 authored and GitHub committed 2026-07-18 11:49:27 +07:00
1 parent 21ad48de16
commit d844795e07
3 files changed
+13 -26
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/store/pg/teams_tasks_followup.go
+1 -1
View File
@@ -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
internal/store/pg/teams_tasks_progress.go
+8 -14
View File
@@ -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 (
internal/store/sqlitestore/teams_tasks_activity.go
+4 -11
View File
@@ -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 = ?