diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py
index 689dbfca61dba..8d882f7cf0b41 100644
--- a/hermes_cli/kanban.py
+++ b/hermes_cli/kanban.py
@@ -790,6 +790,13 @@ def _cmd_attach_rm(args: argparse.Namespace) -> int:
     return 0
 
 
+def _worker_claim_lock_for(task_id: str) -> Optional[str]:
+    """This worker's dispatcher claim lock, only when it is scoped to ``task_id``."""
+    if os.environ.get("HERMES_KANBAN_TASK") != task_id:
+        return None
+    return os.environ.get("HERMES_KANBAN_CLAIM_LOCK") or None
+
+
 def _worker_run_id_for(task_id: str) -> Optional[int]:
     env_tid = os.environ.get("HERMES_KANBAN_TASK")
     if env_tid and env_tid != task_id:
@@ -881,7 +888,12 @@ def op(tid):
                 return False
             fail_msg[tid] = f"cannot complete {tid} (unknown id or terminal state)"
             return kb.complete_task(conn, tid, result=args.result, summary=summary, metadata=metadata,
-                                    expected_run_id=_worker_run_id_for(tid))
+                                    expected_run_id=_worker_run_id_for(tid),
+                                    expected_claim_lock=_worker_claim_lock_for(tid),
+                                    # os.getpid(), never an environment variable: a pid read from the
+                                    # environment is inherited exactly like the claim lock and would
+                                    # rebuild the same hole one layer down.
+                                    expected_worker_pid=os.getpid())
 
         return _bulk_apply(ids, op, lambda tid: f"Completed {tid}", fail_msg.__getitem__)
 
diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py
index 91d6dc724f7c3..9f87bd960fba0 100644
--- a/hermes_cli/kanban_db.py
+++ b/hermes_cli/kanban_db.py
@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import contextlib
+import hashlib
 import json
 import os
 import re
@@ -2537,6 +2538,7 @@ def complete_task(
     conn: sqlite3.Connection, task_id: str, *, result: Optional[str] = None,
     summary: Optional[str] = None, metadata: Optional[dict] = None,
     created_cards: Optional[Iterable[str]] = None, expected_run_id: Optional[int] = None,
+    expected_claim_lock: Optional[str] = None, expected_worker_pid: Optional[int] = None,
     fire_lifecycle_hook: bool = True,
 ) -> bool:
     """``running|ready|blocked|review -> done``; records ``result``.
@@ -2548,6 +2550,10 @@ def complete_task(
     ``created_cards`` are verified first — a phantom id raises
     :class:`HallucinatedCardsError` after an auditable event; afterwards the
     prose is scanned for unresolvable ``t_<hex>`` refs (advisory event only).
+
+    ``expected_claim_lock`` fences completion to the worker that owns the claim
+    (see :func:`_claim_fence_verdict`); a retry after a successful commit still
+    reports success instead of repeating the completion side effects.
     """
     now = int(time.time())
     # Cheap pre-check; re-checked inside the txn to close the parent-reopen race.
@@ -2570,6 +2576,11 @@ def complete_task(
         if acceptance is not None and not record_acceptance(conn, task_id, acceptance):
             return False
         prior_status = _task_status(conn, task_id)
+        if expected_claim_lock is not None:
+            verdict = _claim_fence_verdict(
+                conn, task_id, expected_claim_lock, expected_worker_pid, expected_run_id)
+            if verdict is not None:
+                return verdict
         sql = """
                 UPDATE tasks
                    SET status       = 'done',
@@ -2587,6 +2598,11 @@ def complete_task(
         if expected_run_id is not None:
             sql += " AND current_run_id = ?"
             params = (*params, int(expected_run_id))
+        if expected_claim_lock is not None:
+            # Re-assert ownership in the UPDATE itself: the pre-check above read the row in this
+            # same txn, but the predicate is what makes the fence atomic rather than TOCTOU.
+            sql += " AND claim_lock = ? AND (? IS NULL OR worker_pid IS NULL OR worker_pid = ?)"
+            params = (*params, expected_claim_lock, expected_worker_pid, expected_worker_pid)
         if conn.execute(sql, params).rowcount != 1:
             return False
         if isinstance(metadata, dict):
@@ -2609,7 +2625,8 @@ def complete_task(
             event_summary = _REVIEW_APPROVED_NOTE
         _append_event(
             conn, task_id, "completed",
-            _completed_event_payload(result, event_summary, verified_cards, metadata),
+            _completed_event_payload(
+                result, event_summary, verified_cards, metadata, claim_lock=expected_claim_lock),
             run_id=run_id,
         )
     _flag_phantom_prose_refs(conn, task_id, run_id, summary, result, verified_cards)
@@ -2676,8 +2693,66 @@ def _cleaned_artifact_paths(metadata: Any) -> list[str]:
     return [str(p).strip() for p in raw if isinstance(p, str) and str(p).strip()]
 
 
+def _claim_fence_verdict(
+    conn: sqlite3.Connection, task_id: str, expected_claim_lock: str,
+    expected_worker_pid: Optional[int], expected_run_id: Optional[int],
+) -> Optional[bool]:
+    """Ownership fence for a claimed completion. ``None`` = proceed, else the value to return.
+
+    ``True`` is reserved for the idempotent retry: the task is already ``done`` and the recorded
+    completion carries this very claim lock, so an earlier call by this worker did commit and it
+    must not be told the completion failed.
+    """
+    row = conn.execute(
+        "SELECT status, claim_lock, current_run_id, worker_pid FROM tasks WHERE id = ?", (task_id,),
+    ).fetchone()
+    if row is None:
+        return False
+    if row["status"] == "done":
+        return _completed_by_claim(conn, task_id, expected_claim_lock, expected_run_id)
+    if row["claim_lock"] != expected_claim_lock:
+        return False
+    # The claim lock reaches a nested CLI through the environment, so a child process can present a
+    # matching one and complete its parent's card. The pid of the process actually making the call
+    # cannot be inherited that way, which is what makes it worth checking. A row with no recorded
+    # pid is left alone on purpose: reporting a pid from spawn_fn is a crash-detection nicety, not a
+    # contract, so a deployment whose spawn returns none never stamps one — refusing those would
+    # reject the legitimate worker finishing its own task. Where no pid was recorded this fence is
+    # no weaker than before; where one was, it is strictly stronger.
+    if (expected_worker_pid is not None and row["worker_pid"] is not None
+            and row["worker_pid"] != int(expected_worker_pid)):
+        return False
+    if expected_run_id is not None and row["current_run_id"] != int(expected_run_id):
+        return False
+    return None
+
+
+def _completed_by_claim(
+    conn: sqlite3.Connection, task_id: str, expected_claim_lock: str, expected_run_id: Optional[int],
+) -> bool:
+    """Whether the recorded completion was made by this claim (retry-after-commit)."""
+    sql = "SELECT payload FROM task_events WHERE task_id = ? AND kind = 'completed'"
+    params: list[Any] = [task_id]
+    if expected_run_id is not None:
+        sql += " AND run_id = ?"
+        params.append(int(expected_run_id))
+    sql += " ORDER BY id DESC LIMIT 1"
+    row = conn.execute(sql, tuple(params)).fetchone()
+    try:
+        payload = json.loads(row["payload"]) if row and row["payload"] else {}
+    except (TypeError, ValueError, json.JSONDecodeError):
+        payload = {}
+    return payload.get("completion_claim_lock_sha256") == _claim_digest(expected_claim_lock)
+
+
+def _claim_digest(claim_lock: str) -> str:
+    """Digest, not the lock itself: the event log is readable by anything that can read the board."""
+    return hashlib.sha256(claim_lock.encode("utf-8")).hexdigest()
+
+
 def _completed_event_payload(
     result: Optional[str], event_summary: Optional[str], verified_cards: list[str], metadata: Any,
+    *, claim_lock: Optional[str] = None,
 ) -> dict:
     """``completed`` event payload: first summary line (400 chars) so gateway
     notifiers / dashboard WS render without a second round-trip; verified
@@ -2698,6 +2773,8 @@ def _completed_event_payload(
         cleaned = _cleaned_artifact_paths(metadata)
         if cleaned:
             payload["artifacts"] = cleaned
+    if claim_lock:
+        payload["completion_claim_lock_sha256"] = _claim_digest(claim_lock)
     return payload
 
 
diff --git a/tests/hermes_cli/test_kanban_claim_lock.py b/tests/hermes_cli/test_kanban_claim_lock.py
new file mode 100644
index 0000000000000..2df3b09317eff
--- /dev/null
+++ b/tests/hermes_cli/test_kanban_claim_lock.py
@@ -0,0 +1,313 @@
+"""Regression tests for claim-owned kanban completion."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import kanban_db as kb
+from hermes_cli.kanban_db_connect import connect
+from hermes_cli.kanban_db_dispatch import _set_worker_pid
+
+
+@pytest.fixture
+def kanban_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+    return home
+
+
+def test_complete_task_rejects_matching_claim_lock_in_triage(
+    kanban_home: Path,
+) -> None:
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="owned work", assignee="worker")
+        claimed = kb.claim_task(conn, task_id, claimer="worker:current")
+        assert claimed is not None
+        conn.execute(
+            "UPDATE tasks SET status = 'triage' WHERE id = ?",
+            (task_id,),
+        )
+        conn.commit()
+
+        assert not kb.complete_task(
+            conn,
+            task_id,
+            result="finished",
+            expected_claim_lock="worker:current",
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "triage"
+
+
+def test_complete_task_matching_claim_lock_preserves_non_completable_state(
+    kanban_home: Path,
+) -> None:
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="human-routed work", assignee="worker")
+        claimed = kb.claim_task(conn, task_id, claimer="worker:current")
+        assert claimed is not None
+        conn.execute(
+            "UPDATE tasks SET status = 'todo' WHERE id = ?",
+            (task_id,),
+        )
+        conn.commit()
+
+        # A matching lock must not bypass the set of completable task states.
+        assert not kb.complete_task(
+            conn,
+            task_id,
+            result="finished",
+            expected_claim_lock="worker:current",
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "todo"
+        assert task.result is None
+        assert task.claim_lock == "worker:current"
+
+
+def test_complete_task_rejects_stale_claim_lock(kanban_home: Path) -> None:
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="reassigned work", assignee="worker")
+        first = kb.claim_task(conn, task_id, claimer="worker:first")
+        assert first is not None
+        assert kb.reclaim_task(conn, task_id, signal_fn=lambda *_args: None)
+        second = kb.claim_task(conn, task_id, claimer="worker:second")
+        assert second is not None
+
+        assert not kb.complete_task(
+            conn,
+            task_id,
+            result="stale result",
+            expected_claim_lock="worker:first",
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "running"
+        assert task.result is None
+        assert task.claim_lock == "worker:second"
+        assert task.current_run_id == second.current_run_id
+
+
+def test_complete_task_matching_claim_lock_retry_is_idempotent(
+    kanban_home: Path,
+) -> None:
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="retry completion", assignee="worker")
+        claimed = kb.claim_task(conn, task_id, claimer="worker:retry")
+        assert claimed is not None
+
+        assert kb.complete_task(
+            conn,
+            task_id,
+            result="original",
+            summary="original summary",
+            expected_run_id=claimed.current_run_id,
+            expected_claim_lock="worker:retry",
+        )
+        assert kb.complete_task(
+            conn,
+            task_id,
+            result="replacement",
+            summary="replacement summary",
+            expected_run_id=claimed.current_run_id,
+            expected_claim_lock="worker:retry",
+        )
+
+        task = kb.get_task(conn, task_id)
+        completed = [
+            event
+            for event in kb.list_events(conn, task_id)
+            if event.kind == "completed"
+        ]
+        assert task is not None
+        assert task.result == "original"
+        assert len(completed) == 1
+
+
+def test_complete_task_without_claim_lock_preserves_legacy_behavior(
+    kanban_home: Path,
+) -> None:
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="manual completion")
+
+        assert kb.complete_task(conn, task_id, result="done")
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "done"
+        assert task.result == "done"
+
+
+@pytest.mark.parametrize("status", ["running", "ready", "blocked"])
+def test_complete_task_accepts_matching_claim_lock_on_completable_status(
+    kanban_home: Path, status: str
+) -> None:
+    """The happy path of the fenced branch: right lock, every completable status.
+
+    The rejection tests only pin what the fence rejects. Parametrized over the
+    full IN list because a single positive case (say ``running``) would stay
+    green if the predicate silently dropped ``ready`` or ``blocked``.
+    """
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="owned work", assignee="worker")
+        claimed = kb.claim_task(conn, task_id, claimer="worker:current")
+        assert claimed is not None
+        conn.execute("UPDATE tasks SET status = ? WHERE id = ?", (status, task_id))
+        conn.commit()
+
+        assert kb.complete_task(
+            conn,
+            task_id,
+            result="finished",
+            expected_claim_lock="worker:current",
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "done"
+        assert task.result == "finished"
+        assert task.claim_lock is None
+
+
+def test_complete_task_rejects_inherited_claim_lock_from_another_pid(
+    kanban_home: Path,
+) -> None:
+    """A nested CLI inherits the claim lock but not the worker's identity.
+
+    Reported on NousResearch/hermes-agent#71175: a nested Hermes CLI inherits
+    HERMES_KANBAN_TASK and HERMES_KANBAN_CLAIM_LOCK from its parent, so it
+    presents a claim lock that matches and completes the parent's card from a
+    different pid. Every other component of the identity travels down to a
+    child the same way, which is why the pid of the calling process is the one
+    that can tell them apart.
+    """
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="worker task", assignee="worker")
+        claim = kb.claim_task(conn, task_id, claimer="worker:live")
+        assert claim is not None
+
+        # claim_task does not stamp the pid; the dispatcher does it separately
+        # via _set_worker_pid, so a test that wants a live worker has to say so.
+        live_pid = 424242
+        _set_worker_pid(conn, task_id, live_pid)
+
+        assert not kb.complete_task(
+            conn,
+            task_id,
+            result="completed by a nested CLI",
+            expected_claim_lock="worker:live",
+            expected_worker_pid=live_pid + 1,
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "running"
+        assert task.result is None
+        assert task.claim_lock == "worker:live"
+        assert task.worker_pid == live_pid
+
+
+def test_complete_task_accepts_matching_worker_pid(kanban_home: Path) -> None:
+    """The real worker still completes its own card.
+
+    The bypass test above passes for a gate that refuses everything, so this is
+    the half that keeps it honest.
+    """
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="worker task", assignee="worker")
+        claim = kb.claim_task(conn, task_id, claimer="worker:live")
+        assert claim is not None
+
+        # claim_task does not stamp the pid; the dispatcher does it separately
+        # via _set_worker_pid, so a test that wants a live worker has to say so.
+        live_pid = 424242
+        _set_worker_pid(conn, task_id, live_pid)
+
+        assert kb.complete_task(
+            conn,
+            task_id,
+            result="completed by its own worker",
+            expected_claim_lock="worker:live",
+            expected_worker_pid=live_pid,
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "done"
+        assert task.result == "completed by its own worker"
+
+
+def test_tool_complete_rejects_inherited_claim_lock_from_another_pid(
+    kanban_home: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The agent tool path must carry the caller's pid, not just the claim lock.
+
+    This is the path the bypass on #71175 travels: a nested CLI inherits
+    HERMES_KANBAN_TASK and HERMES_KANBAN_CLAIM_LOCK and completes its parent's
+    card. A database that refuses a mismatched pid proves nothing if the tool
+    never sends one, so this pins the wiring rather than the check.
+    """
+    from tools import kanban_tools
+
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="worker task", assignee="worker")
+        claim = kb.claim_task(conn, task_id, claimer="worker:live")
+        assert claim is not None
+        live_pid = 424242
+        _set_worker_pid(conn, task_id, live_pid)
+
+    # The nested process inherits the parent's task and claim lock verbatim.
+    monkeypatch.setenv("HERMES_KANBAN_TASK", task_id)
+    monkeypatch.setenv("HERMES_KANBAN_CLAIM_LOCK", "worker:live")
+    # ...but runs under its own pid, which is the one thing it cannot inherit.
+    monkeypatch.setattr(kanban_tools.os, "getpid", lambda: live_pid + 1)
+
+    kanban_tools._handle_complete({"id": task_id, "result": "from a nested CLI"})
+
+    with connect() as conn:
+        task = kb.get_task(conn, task_id)
+    assert task is not None
+    assert task.status == "running"
+    assert task.result is None
+    assert task.worker_pid == live_pid
+
+
+def test_complete_task_allows_worker_whose_spawn_reported_no_pid(
+    kanban_home: Path,
+) -> None:
+    """A deployment that never stamps a pid must keep working.
+
+    Reporting a pid from spawn_fn is a crash-detection nicety rather than a
+    contract, so a custom spawn that returns none leaves worker_pid NULL
+    forever. Tightening the fence to refuse those rejected the legitimate
+    worker finishing its own task — caught in review after the first attempt
+    did exactly that. Where no pid was recorded this fence is no weaker than
+    before; where one was, it is strictly stronger.
+    """
+    with connect() as conn:
+        task_id = kb.create_task(conn, title="pidless worker", assignee="worker")
+        claim = kb.claim_task(conn, task_id, claimer="worker:live")
+        assert claim is not None
+        assert kb.get_task(conn, task_id).worker_pid is None
+
+        assert kb.complete_task(
+            conn,
+            task_id,
+            result="completed without a recorded pid",
+            expected_claim_lock="worker:live",
+            expected_worker_pid=4242,
+        )
+
+        task = kb.get_task(conn, task_id)
+        assert task is not None
+        assert task.status == "done"
diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py
index 6a3eb5aa91251..8cf1b4a9ce417 100644
--- a/tools/kanban_tools.py
+++ b/tools/kanban_tools.py
@@ -167,6 +167,11 @@ def _own_task_env(task_id: str, var: str) -> Optional[str]:
     return os.environ.get(var) if os.environ.get("HERMES_KANBAN_TASK") == task_id else None
 
 
+def _worker_claim_lock(task_id: str) -> Optional[str]:
+    """This worker's dispatcher claim lock when it is scoped to ``task_id``."""
+    return _own_task_env(task_id, "HERMES_KANBAN_CLAIM_LOCK") or None
+
+
 def _worker_run_id(task_id: str) -> Optional[int]:
     """This worker's dispatcher run id when it is scoped to task_id."""
     raw = _own_task_env(task_id, "HERMES_KANBAN_RUN_ID")
@@ -586,7 +591,12 @@ def _handle_complete(args: dict, **kw) -> str:
         try:
             ok = kb.complete_task(
                 conn, tid, result=result, summary=summary, metadata=metadata,
-                created_cards=created_cards, expected_run_id=_worker_run_id(tid))
+                created_cards=created_cards, expected_run_id=_worker_run_id(tid),
+                expected_claim_lock=_worker_claim_lock(tid),
+                # The tool handoff is the path the reported bypass actually travels: a nested CLI
+                # inherits the claim lock and completes its parent's card from here. os.getpid() is
+                # the one input that cannot be inherited along with it.
+                expected_worker_pid=os.getpid())
         except kb.ArtifactPreservationError as artifact_err:
             # Structured rejection — surface the phantom ids so the worker can retry with a corrected list
             # or drop the field. Audit event already landed in the DB. The task itself was NOT mutated (the