fix(kanban): request_review shares complete_task's live-worker fence

The PR docstring said complete_task applies "the same fence request_review
applies", but the two disagreed: complete_task keyed on a live worker
process while request_review still refused any running task with a
claim_lock, so a claim whose worker is gone (or a CLI/library claim that
never spawned one) could be completed but not sent to review without
force. Factor the liveness test into _claim_is_live and use it in both.

TTL expiry is deliberately not part of "live": reclaim_stale_tasks extends
(not reclaims) the claim of a live worker, so the process stays the
liveness authority.

Test: claim -> request_review without a worker PID now succeeds (red
before); a live worker's claim is still refused without expected_run_id.

hermes_cli/kanban_db.py

@@ -2614,6 +2614,20 @@ class LiveClaimError(ValueError):
         )
 
 
+def _claim_is_live(trow) -> bool:
+    """True when a ``running`` task's claim still protects a run: the worker process
+    it spawned exists (PID + start-time fingerprint). A claim whose worker is gone,
+    or a library/CLI claim that never spawned one, has no run to protect. TTL expiry
+    is deliberately not consulted: ``reclaim_stale_tasks`` extends, not reclaims, the
+    claim of a live worker, so the process is the liveness authority here too."""
+    return bool(
+        trow["status"] == "running"
+        and trow["claim_lock"] is not None
+        and trow["worker_pid"]
+        and _worker_alive(trow["worker_pid"], trow["worker_started_at"])
+    )
+
+
 def complete_task(
     conn: sqlite3.Connection, task_id: str, *, result: Optional[str] = None,
     summary: Optional[str] = None, metadata: Optional[dict] = None,
@@ -2659,18 +2673,9 @@ def complete_task(
         ).fetchone()
         prior_status = trow["status"] if trow else None
         # Refuse to close a LIVE worker's run without proof of ownership
-        # (expected_run_id) or an explicit human override (force=True). "Live"
-        # means the spawned worker process still exists: a claim whose worker
-        # is gone (or a library claim that never spawned one) has no run to
-        # protect, so manual completion keeps working there.
-        if (
-            expected_run_id is None
-            and not force
-            and prior_status == "running"
-            and trow["claim_lock"] is not None
-            and trow["worker_pid"]
-            and _worker_alive(trow["worker_pid"], trow["worker_started_at"])
-        ):
+        # (expected_run_id) or an explicit human override (force=True); see
+        # _claim_is_live for what "live" means.
+        if expected_run_id is None and not force and trow and _claim_is_live(trow):
             raise LiveClaimError(task_id)
         sql = """
                 UPDATE tasks
@@ -3164,19 +3169,15 @@ def request_review(
             if not _parents_satisfied(conn, task_id):
                 return _ret(False, "parent dependencies are not satisfied")
             trow = conn.execute(
-                "SELECT assignee, status, claim_lock, current_run_id "
-                "FROM tasks WHERE id = ?", (task_id,),
+                "SELECT assignee, status, claim_lock, current_run_id, worker_pid, "
+                "worker_started_at FROM tasks WHERE id = ?", (task_id,),
             ).fetchone()
             if trow is None:
                 return _ret(False, "task not found")
             # Refuse to clear a live worker's claim without proof of ownership
-            # (expected_run_id) or an explicit human override (force=True).
-            if (
-                expected_run_id is None
-                and not force
-                and trow["status"] == "running"
-                and trow["claim_lock"] is not None
-            ):
+            # (expected_run_id) or an explicit human override (force=True);
+            # the same fence as complete_task (_claim_is_live).
+            if expected_run_id is None and not force and _claim_is_live(trow):
                 return _ret(
                     False, "task is running under a live claim; pass expected_run_id "
                     "(worker ownership) or force=True (explicit operator "

tests/hermes_cli/test_kanban_complete_live_claim_guard.py

@@ -75,3 +75,17 @@ def test_claimless_complete_of_unclaimed_card_unchanged(conn):
     tid = kb.create_task(conn, title="admin", assignee="coder")
     assert kb.complete_task(conn, tid, result="done") is True
     assert conn.execute("SELECT status FROM tasks WHERE id = ?", (tid,)).fetchone()["status"] == "done"
+
+
+def test_request_review_shares_the_live_worker_fence(conn):
+    """``request_review`` keys on the same liveness as ``complete_task``: a claim
+    without a live worker process is not a live claim (the human/library flow
+    ``claim`` -> ``request_review`` works), a live worker's claim still is."""
+    tid, _ = _claimed_running_task(conn, live_worker=False)
+    assert kb.request_review(conn, tid, summary="handoff") is True
+    assert kb.get_task(conn, tid).status == "review"
+
+    tid2, run2 = _claimed_running_task(conn)
+    ok, reason = kb.request_review(conn, tid2, summary="steal", with_reason=True)
+    assert ok is False and "live claim" in reason
+    assert kb.request_review(conn, tid2, summary="own", expected_run_id=run2) is True

tests/hermes_cli/test_kanban_review_lifecycle.py

@@ -21,6 +21,7 @@ down:
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -201,6 +202,8 @@ def test_request_review_refuses_to_clear_live_claim_without_ownership(
         tid = kb.create_task(conn, title="live claim", assignee="worker")
         claimed = kb.claim_task(conn, tid)
         assert claimed is not None
+        # This process stands in for the spawned worker: alive, fingerprinted.
+        kbd._set_worker_pid(conn, tid, os.getpid())
 
         # 1) No run id, no force -> refused with a distinct reason.
         ok, reason = kb.request_review(conn, tid, with_reason=True)