From 261f4467ddfbaffb8b32ccf0ed613edc14d3b93d Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 21 Jun 2026 12:36:18 -0700 Subject: [PATCH] fix(kanban): make reclaim claim-lock-aware to stop task/run status desync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After a worker crash + reclaim + respawn, the board could show a task in the Ready lane while its task_run was 'running' and the new worker was actively executing (#36910). The dispatcher could then treat live work as available and double-assign. Root cause: the three reclaim paths (detect_crashed_workers, release_stale_claims heartbeat-stale backstop, enforce_max_runtime) each snapshot a task's worker_pid/claim_lock, do liveness work, then reset tasks.status back to 'ready' with only a 'WHERE status=running' guard. If the task was reclaimed AND re-claimed by a NEW worker in between (new run, new claim_lock, live pid), the stale UPDATE clobbered the live task: status flipped to 'ready' while the fresh run stayed 'running'. claim_task is the only writer that sets status='running', so nothing put it back — permanent desync. Fix: gate each reset on the snapshot's claim_lock (and worker_pid where available) so it only fires when the task is still owned by the worker the reclaim was computed for. A stale reclaim now no-ops (rowcount 0) instead of desyncing a re-claimed task. Genuine crashes (lock still matches) reclaim exactly as before. This is the same race class the in-gateway dispatch lock (single-writer ticks) mitigates, closed at the row level so a single dispatcher's fast reclaim->respawn across two ticks is also safe. Closes #36910. --- hermes_cli/kanban_db.py | 15 ++- .../test_kanban_reclaim_claim_lock_guard.py | 113 ++++++++++++++++++ 2 files changed, 122 insertions(+), 6 deletions(-) create mode 100644 tests/hermes_cli/test_kanban_reclaim_claim_lock_guard.py diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 721403892c9c..236c600e31cb 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -5662,8 +5662,9 @@ def enforce_max_runtime( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " "claim_expires = NULL, worker_pid = NULL, " "last_heartbeat_at = NULL " - "WHERE id = ? AND status = 'running'", - (tid,), + "WHERE id = ? AND status = 'running' " + " AND worker_pid = ? AND claim_lock IS ?", + (tid, pid, row["claim_lock"]), ) if cur.rowcount == 1: payload = { @@ -5787,8 +5788,9 @@ def detect_stale_running( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " "claim_expires = NULL, worker_pid = NULL, " "last_heartbeat_at = NULL " - "WHERE id = ? AND status = 'running'", - (tid,), + "WHERE id = ? AND status = 'running' " + " AND claim_lock IS ?", + (tid, row["claim_lock"]), ) if cur.rowcount != 1: continue @@ -5960,8 +5962,9 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " "claim_expires = NULL, worker_pid = NULL " - "WHERE id = ? AND status = 'running'", - (row["id"],), + "WHERE id = ? AND status = 'running' " + " AND worker_pid = ? AND claim_lock IS ?", + (row["id"], pid, row["claim_lock"]), ) if cur.rowcount == 1: # Rate-limited requeues are a clean release, not a crash — diff --git a/tests/hermes_cli/test_kanban_reclaim_claim_lock_guard.py b/tests/hermes_cli/test_kanban_reclaim_claim_lock_guard.py new file mode 100644 index 000000000000..40ca86a741f0 --- /dev/null +++ b/tests/hermes_cli/test_kanban_reclaim_claim_lock_guard.py @@ -0,0 +1,113 @@ +"""Tests: reclaim paths are claim-lock-aware so they can't desync a re-claimed +task (issue #36910). + +A stale crash/stale-claim/max-runtime reclaim, computed from a snapshot of an +OLD worker, used to reset ``tasks.status`` back to ``ready`` with only a +``WHERE status='running'`` guard. If the task had since been reclaimed AND +re-claimed by a NEW worker (new run, new claim_lock, live pid), that stale +UPDATE clobbered the live task: ``tasks.status='ready'`` while the new +``task_runs.status='running'`` and the worker kept executing — the board showed +the task in the Ready lane and the dispatcher could treat live work as +available. The reset is now gated on the snapshot's ``claim_lock`` (and pid), +so it only fires when the task is still owned by the worker the reclaim was +computed for. +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_KANBAN_HOME", str(home)) + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") + monkeypatch.setattr(Path, "home", lambda: tmp_path) + db_path = kb.kanban_db_path(board="default") + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + kb.init_db() + return home + + +@pytest.fixture +def conn(kanban_home): + with kb.connect() as c: + yield c + + +def test_stale_crash_reset_rejected_for_reclaimed_task(conn): + """A reset carrying an OLD worker's claim_lock must NOT clobber a task + that has since been re-claimed by a new worker.""" + host = kb._claimer_id().split(":", 1)[0] + tid = kb.create_task(conn, title="desync", assignee="w") + + # Worker A claims, then dies. + kb.claim_task(conn, tid, claimer=f"{host}:A") + dead = subprocess.Popen(["true"]) + dead.wait() + kb._set_worker_pid(conn, tid, dead.pid) + old = conn.execute( + "SELECT claim_lock, worker_pid FROM tasks WHERE id=?", (tid,) + ).fetchone() + + # Reclaim + re-claim by worker B (alive). + conn.execute( + "UPDATE tasks SET status='ready', claim_lock=NULL, claim_expires=NULL, " + "worker_pid=NULL, current_run_id=NULL WHERE id=?", + (tid,), + ) + conn.commit() + kb.claim_task(conn, tid, claimer=f"{host}:B") + sleeper = subprocess.Popen(["sleep", "30"]) + try: + kb._set_worker_pid(conn, tid, sleeper.pid) + + # The stale reset for worker A — same shape as the guarded UPDATE in + # detect_crashed_workers — must reject (rowcount 0) because B owns it. + cur = conn.execute( + "UPDATE tasks SET status='ready', claim_lock=NULL, " + "claim_expires=NULL, worker_pid=NULL " + "WHERE id=? AND status='running' AND worker_pid=? AND claim_lock IS ?", + (tid, old["worker_pid"], old["claim_lock"]), + ) + conn.commit() + assert cur.rowcount == 0, "stale reclaim wrongly clobbered the re-claimed task" + + final = conn.execute( + "SELECT status, claim_lock FROM tasks WHERE id=?", (tid,) + ).fetchone() + assert final["status"] == "running" + assert final["claim_lock"] == f"{host}:B" + finally: + sleeper.terminate() + + +def test_genuine_crash_still_reclaims(conn): + """When the claim_lock still matches the dead worker, the crash reclaim + fires normally — the guard must not break the legitimate path.""" + host = kb._claimer_id().split(":", 1)[0] + tid = kb.create_task(conn, title="legit", assignee="w") + kb.claim_task(conn, tid, claimer=f"{host}:A") + dead = subprocess.Popen(["true"]) + dead.wait() + kb._set_worker_pid(conn, tid, dead.pid) + # Rewind started_at so the launch grace window doesn't skip the check. + conn.execute("UPDATE tasks SET started_at = started_at - 9999 WHERE id=?", (tid,)) + conn.execute( + "UPDATE task_runs SET started_at = started_at - 9999 WHERE task_id=?", (tid,) + ) + conn.commit() + kb._record_worker_exit(dead.pid, 1 << 8) # nonzero exit → crash + + crashed = kb.detect_crashed_workers(conn) + assert tid in crashed + final = conn.execute("SELECT status FROM tasks WHERE id=?", (tid,)).fetchone() + assert final["status"] in ("ready", "blocked", "todo")