Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions hermes_cli/kanban_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 —
Expand Down
113 changes: 113 additions & 0 deletions tests/hermes_cli/test_kanban_reclaim_claim_lock_guard.py
Original file line number Diff line number Diff line change
@@ -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")
Loading