Kanban task/run status desync bug and fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50366 fix(kanban): make reclaim claim-lock-aware to stop task/run status desync

  merged to main

  files: hermes_cli/kanban_db.py, tests/hermes_cli/test_kanban_reclaim_claim_lock_guard.py

  Show the diff

  • 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 —

  • 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
    +
    +
    [redacted:email]
    +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
    +
    +
    [redacted:email]
    +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")