fix(kanban): guard request_review against live-claim theft

request_review on a running task under a live claim now requires the
caller to prove ownership (expected_run_id, the unchanged worker path)
or pass an explicit force=True override (CLI --force; dashboard human
actions pass force=True) instead of silently clearing claim_lock /
worker_pid of a live run.

Failures now carry distinct diagnostic reasons via with_reason=True
(mirroring request_changes' tuple pattern): live-claim refusal,
malformed re-review provenance, unsatisfied parents, unknown task, and
CAS miss. Tool/CLI handlers surface the specific reason instead of the
generic 'unknown id or not in running/ready'.

Regression tests: live-claim refusal + force/worker paths; malformed
provenance gets a distinct reason and explicit reviewer= recovers.

hermes_cli/kanban.py

@@ -672,6 +672,13 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu
         "--metadata", default=None,
         help="JSON object with structured reviewer handoff facts.",
     )
+    p_request_review.add_argument(
+        "--force", action="store_true",
+        help=(
+            "Override the live-claim guard: move a running, claimed task to "
+            "review even without owning its run (clears the worker's claim)."
+        ),
+    )
 
     p_request_changes = sub.add_parser(
         "request-changes",
@@ -2415,16 +2422,20 @@ def _cmd_request_review(args: argparse.Namespace) -> int:
                 file=sys.stderr,
             )
             return 1
-        if not kb.request_review(
+        ok, reason = kb.request_review(
             conn,
             tid,
             summary=summary,
             metadata=metadata,
             reviewer=reviewer,
             expected_run_id=_worker_run_id_for(tid),
-        ):
+            force=bool(getattr(args, "force", False)),
+            with_reason=True,
+        )
+        if not ok:
+            detail = reason or "not running/ready?"
             print(
-                f"cannot request review for {tid} (not running/ready?)",
+                f"cannot request review for {tid}: {detail}",
                 file=sys.stderr,
             )
             return 1

hermes_cli/kanban_db.py

@@ -6131,7 +6131,9 @@ def request_review(
     metadata: Optional[dict] = None,
     reviewer: Optional[str] = None,
     expected_run_id: Optional[int] = None,
-) -> bool:
+    force: bool = False,
+    with_reason: bool = False,
+):
     """Transition implementation work into the first-class review phase.
 
     Unlike :func:`block_task`, this transition never touches block recurrence
@@ -6140,17 +6142,46 @@ def request_review(
     right profile.  Supplying ``reviewer`` reassigns the task before it is
     exposed to the review dispatcher.  On re-review, omitting it reuses the
     reviewer provenance persisted by the latest ``changes_requested`` event.
+
+    When the task is ``running`` under a live claim, a caller that supplies no
+    ``expected_run_id`` must pass ``force=True`` (explicit human/CLI override)
+    — otherwise the request is refused instead of silently clearing the live
+    worker's ``claim_lock``/``worker_pid``. Workers prove ownership by passing
+    their own run id as ``expected_run_id`` (unchanged).
+
+    Returns ``bool`` by default. With ``with_reason=True`` returns
+    ``(ok, reason)`` mirroring :func:`request_changes` — ``reason`` is a
+    diagnostic string on failure, ``None`` on success.
     """
+
+    def _ret(ok: bool, reason: Optional[str] = None):
+        return (ok, reason) if with_reason else ok
+
     summary = redact_review_value(summary)
     metadata = redact_review_value(metadata)
     with write_txn(conn):
         if not _parents_satisfied(conn, task_id):
-            return False
+            return _ret(False, "parent dependencies are not satisfied")
         trow = conn.execute(
-            "SELECT assignee FROM tasks WHERE id = ?", (task_id,),
+            "SELECT assignee, status, claim_lock, current_run_id "
+            "FROM tasks WHERE id = ?", (task_id,),
         ).fetchone()
         if trow is None:
-            return False
+            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
+        ):
+            return _ret(
+                False,
+                "task is running under a live claim; pass expected_run_id "
+                "(worker ownership) or force=True (explicit operator "
+                "override) instead of clearing the live run's claim",
+            )
         implementer = trow["assignee"]
         if reviewer is None:
             changes_run = conn.execute(
@@ -6183,7 +6214,12 @@ def request_review(
             )
             if changes_run is not None:
                 if not isinstance(prior_reviewer, str) or not prior_reviewer.strip():
-                    return False
+                    return _ret(
+                        False,
+                        "re-review has no durable reviewer provenance (the "
+                        "latest changes_requested event is missing or "
+                        "malformed); pass reviewer= explicitly",
+                    )
                 reviewer = prior_reviewer
         reviewer = _canonical_assignee(reviewer) if reviewer is not None else None
         assignee_sql = ", assignee = ?" if reviewer is not None else ""
@@ -6212,7 +6248,11 @@ def request_review(
             params,
         )
         if cur.rowcount != 1:
-            return False
+            return _ret(
+                False,
+                "task is not in running/ready (or expected_run_id did not "
+                "match the current run)",
+            )
         run_id = _end_run(
             conn,
             task_id,
@@ -6242,7 +6282,7 @@ def request_review(
             },
             run_id=run_id,
         )
-    return True
+    return _ret(True)
 
 
 def request_changes(

plugins/kanban/dashboard/plugin_api.py

@@ -916,6 +916,9 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu
                     conn, task_id, summary=payload.summary,
                     metadata=payload.metadata,
                     reviewer=(payload.assignee or None),
+                    # Dashboard PATCH is an explicit human action — allowed
+                    # to override a live worker claim (M1 guard).
+                    force=True,
                 )
                 if ok and review_assignee_deferred and not payload.assignee:
                     ok = kanban_db.assign_task(conn, task_id, None)
@@ -1380,6 +1383,8 @@ def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)):
                             conn, tid, summary=payload.summary,
                             metadata=payload.metadata,
                             reviewer=(payload.assignee or None),
+                            # Bulk dashboard action: explicit human override.
+                            force=True,
                         )
                     elif s == "ready":
                         cur = kanban_db.get_task(conn, tid)

tests/hermes_cli/test_kanban_review_lifecycle.py

@@ -182,6 +182,88 @@ def test_request_review_unknown_task_returns_false(kanban_home: Path) -> None:
         assert kb.request_review(conn, "t_deadbeefcafe") is False
 
 
+def test_request_review_refuses_to_clear_live_claim_without_ownership(
+    kanban_home: Path,
+) -> None:
+    """M1 regression: a run-id-less caller must not steal a live worker's claim.
+
+    ``request_review`` on a running+claimed task without ``expected_run_id``
+    fails with a distinct reason instead of silently NULLing claim_lock /
+    worker_pid. ``force=True`` (explicit human override) and the worker path
+    (``expected_run_id=<own run>``) both still work.
+    """
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="live claim", assignee="worker")
+        claimed = kb.claim_task(conn, tid)
+        assert claimed is not None
+
+        # 1) No run id, no force -> refused with a distinct reason.
+        ok, reason = kb.request_review(conn, tid, with_reason=True)
+        assert ok is False
+        assert reason is not None and "live claim" in reason
+        row = conn.execute(
+            "SELECT status, claim_lock, current_run_id FROM tasks WHERE id = ?",
+            (tid,),
+        ).fetchone()
+        assert row["status"] == "running"
+        assert row["claim_lock"] is not None  # live claim untouched
+        # bool-mode caller sees plain False.
+        assert kb.request_review(conn, tid) is False
+
+        # 2) Worker path: proving ownership via expected_run_id works.
+        assert kb.request_review(
+            conn, tid, summary="done", expected_run_id=claimed.current_run_id,
+        ) is True
+        assert kb.get_task(conn, tid).status == "review"
+
+    # 3) force=True: explicit human override on a fresh live-claimed task.
+    with kb.connect() as conn:
+        tid2 = kb.create_task(conn, title="forced", assignee="worker")
+        assert kb.claim_task(conn, tid2) is not None
+        assert kb.request_review(conn, tid2, summary="override", force=True) is True
+        assert kb.get_task(conn, tid2).status == "review"
+
+
+def test_request_review_malformed_provenance_gets_distinct_reason(
+    kanban_home: Path,
+) -> None:
+    """M1 regression: malformed re-review provenance is a named failure, not
+    the generic 'unknown id or not in running/ready'."""
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="provenance", assignee="builder")
+        claimed = kb.claim_task(conn, tid)
+        assert kb.request_review(
+            conn, tid, summary="v1", reviewer="reviewer",
+            expected_run_id=claimed.current_run_id,
+        )
+        review = kb.claim_review_task(conn, tid)
+        assert review is not None
+        assert kb.request_changes(
+            conn, tid, reason="fix", expected_run_id=review.current_run_id,
+        ) == (True, "builder")
+        # Corrupt the changes_requested payload so re-review cannot recover
+        # the prior reviewer.
+        with kb.write_txn(conn):
+            conn.execute(
+                "UPDATE task_events SET payload = '{\"reviewer\": 42}' "
+                "WHERE task_id = ? AND kind = 'changes_requested'",
+                (tid,),
+            )
+        retry = kb.claim_task(conn, tid, claimer="builder:retry")
+        assert retry is not None
+        ok, reason = kb.request_review(
+            conn, tid, summary="v2",
+            expected_run_id=retry.current_run_id, with_reason=True,
+        )
+        assert ok is False
+        assert reason is not None and "provenance" in reason
+        # Passing reviewer explicitly recovers, as the reason instructs.
+        assert kb.request_review(
+            conn, tid, summary="v2", reviewer="reviewer",
+            expected_run_id=retry.current_run_id,
+        ) is True
+
+
 @pytest.mark.parametrize("blank", ["   ", "\n", "\t\n  "])
 def test_request_review_whitespace_only_summary_does_not_crash(
     kanban_home: Path, blank: str

tools/kanban_tools.py

@@ -944,17 +944,18 @@ def _handle_request_review(args: dict, **kw) -> str:
                     "Provide acceptance evidence matching the card before "
                     "requesting review."
                 )
-            ok = kb.request_review(
+            ok, fail_reason = kb.request_review(
                 conn, tid,
                 summary=summary,
                 metadata=metadata,
                 reviewer=reviewer,
                 expected_run_id=_worker_run_id(tid),
+                with_reason=True,
             )
             if not ok:
+                detail = fail_reason or "unknown id or not in running/ready"
                 return tool_error(
-                    f"could not request review for {tid} (unknown id or not "
-                    f"in running/ready)"
+                    f"could not request review for {tid}: {detail}"
                 )
             run = kb.latest_run(conn, tid)
             landed = kb.get_task(conn, tid)