fix(kanban): trust a worker PID only inside the claim's PID namespace

hermes_cli/kanban_db_connect.py

@@ -727,8 +727,7 @@ def _init_if_needed(conn: sqlite3.Connection) -> None:
             # Idempotent; runs under _INIT_LOCK so same-process dispatcher
             # threads can't race the ALTER TABLE pass with stale PRAGMA snapshots.
             if resolved not in _INITIALIZED_PATHS:
-                conn.executescript(_kb.SCHEMA_SQL)
-                _migrate_add_optional_columns(conn)
+                migrate_schema(conn)
                 _INITIALIZED_PATHS.add(resolved)
 
         conn, _ = _open_configured(path, _init_if_needed)
@@ -837,6 +836,12 @@ def init_db(db_path: Optional[Path] = None, *, board: Optional[str] = None) -> P
     ("block_recurrences", "block_recurrences INTEGER NOT NULL DEFAULT 0"),
     # Spawn-time start fingerprint of worker_pid (PID-reuse guard; NULL = legacy row).
     ("worker_started_at", "worker_started_at INTEGER"),
+    # PID namespace of the process holding ``claim_lock`` (kanban_db_pidns).
+    # NULL on legacy rows and on claims from older writers: fails closed where
+    # the reader can identify its own namespace (Linux) — the PID is neither
+    # probed nor signalled and the claim waits for its TTL — and keeps the
+    # hostname-only PID trust only where it cannot (macOS/Windows).
+    ("claim_pidns", "claim_pidns TEXT"),
 )
 
 _NOTIFY_SUB_COLUMNS = (
@@ -852,6 +857,9 @@ def init_db(db_path: Optional[Path] = None, *, board: Optional[str] = None) -> P
 )
 
 _TASK_RUN_COLUMNS = (
+    # PID namespace that owned the run's claim. It survives the terminal task
+    # transition so the terminal-worker reaper can qualify its PID evidence.
+    ("claim_pidns", "claim_pidns TEXT"),
     # Spawn-time start fingerprint of the run's worker_pid (PID-reuse guard for the
     # terminal-worker reaper; NULL = legacy row, never signalled).
     ("worker_started_at", "worker_started_at INTEGER"),
@@ -868,6 +876,16 @@ def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
     ).fetchone() is not None
 
 
+def migrate_schema(conn: sqlite3.Connection) -> None:
+    """Bring an open connection (``row_factory = sqlite3.Row``) to this install's
+    schema: ``SCHEMA_SQL`` is ``IF NOT EXISTS`` throughout, then the additive
+    column pass. ``connect`` runs it on the first open per path; board export
+    runs it on its snapshot, which never went through ``connect`` and would
+    otherwise carry the previous release's columns into the scrub."""
+    conn.executescript(_kb.SCHEMA_SQL)
+    _migrate_add_optional_columns(conn)
+
+
 def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
     """Add columns introduced after v1 to legacy DBs (called via ``init_db``)."""
     cols = _column_names(conn, "tasks")
@@ -1031,7 +1049,7 @@ def _backfill_legacy_inflight_runs(conn: sqlite3.Connection) -> None:
         "CREATE TABLE task_runs ("
         " id INTEGER PRIMARY KEY AUTOINCREMENT,"
         " task_id TEXT NOT NULL, profile TEXT, step_key TEXT,"
-        " status TEXT NOT NULL, claim_lock TEXT, claim_expires INTEGER,"
+        " status TEXT NOT NULL, claim_lock TEXT, claim_expires INTEGER, claim_pidns TEXT,"
         " worker_pid INTEGER, worker_started_at INTEGER, max_runtime_seconds INTEGER,"
         " last_heartbeat_at INTEGER, started_at INTEGER NOT NULL,"
         " ended_at INTEGER, outcome TEXT, summary TEXT, metadata TEXT,"

hermes_cli/kanban_db_pidns.py

@@ -0,0 +1,130 @@
+"""PID-namespace identity for kanban worker claims (the claim-provenance owner).
+
+A claim lock is ``<hostname>:<pid>`` (``kanban_db._claimer_id``) and the reclaim
+passes historically used the hostname alone to decide that a PID they could not
+see was a dead worker. Hostname and PID namespace are independent in
+containers: Compose services joined with ``network_mode: "service:x"`` share
+one hostname, and unless ``pid:`` is set too each keeps its own PID namespace,
+where the same PID number is an unrelated process. ``/proc/<pid>`` only answers
+for the caller's namespace, so "no such PID" is not evidence about a worker in
+a container we cannot see into.
+
+``tasks.claim_pidns`` records the namespace a claim was made in
+(``kanban_db._claim_and_open_run``) and leaves with the lock on release. This
+module owns the identity resolution (``_local_pid_namespace``) and the guard
+every PID probe and every SIGTERM goes through (``_claim_pid_checkable``). It
+is a leaf: it imports nothing from the other ``kanban_db_*`` modules at load
+time, so those facades import it without a cycle.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+from typing import NamedTuple, Optional
+
+
+class LocalPidNamespace(NamedTuple):
+    """What this process knows about its own PID namespace.
+
+    Three states, and :func:`_claim_pid_checkable` treats each differently:
+    ``supported`` False — the platform has no namespace identity at all (no
+    ``/proc/self/ns/pid``: macOS, Windows); ``supported`` True with an ``id``
+    — Linux, resolved; ``supported`` True with ``id None`` — Linux, but the
+    lookup failed (restricted or unmounted ``/proc``). The last is unknown
+    authority, not the compatibility case: absence of our own provenance
+    cannot become provenance because the read failed.
+    """
+
+    id: Optional[str]
+    supported: bool
+
+
+_UNSUPPORTED = LocalPidNamespace(None, False)
+
+# Cached once there is a definite answer: a process cannot move itself into
+# another PID namespace (``setns`` applies to its future children), so a
+# successful read never changes for us. A failed read is NOT cached — it may
+# be transient, and while it lasts every host-local claim is unverifiable.
+_LOCAL_PID_NS: Optional[LocalPidNamespace] = None
+
+
+def _parse_pid_namespace_link(link: str) -> Optional[str]:
+    """``"pid:[4026532534]"`` -> ``"4026532534"``; None when unparseable.
+
+    Split out from the resolver so the parsing can be tested as the pure
+    string -> string mapping it is, on any host.
+    """
+    match = re.search(r"\[(\d+)\]", link or "")
+    return match.group(1) if match else None
+
+
+def _resolve_local_pid_namespace() -> LocalPidNamespace:
+    if sys.platform != "linux":
+        return _UNSUPPORTED
+    try:
+        link = os.readlink("/proc/self/ns/pid")
+    except OSError:
+        return LocalPidNamespace(None, True)
+    return LocalPidNamespace(_parse_pid_namespace_link(link), True)
+
+
+def _local_pid_namespace() -> LocalPidNamespace:
+    """This process' PID-namespace identity (see :class:`LocalPidNamespace`)."""
+    global _LOCAL_PID_NS
+    if _LOCAL_PID_NS is None:
+        local = _resolve_local_pid_namespace()
+        if local.id is not None or not local.supported:
+            _LOCAL_PID_NS = local
+        return local
+    return _LOCAL_PID_NS
+
+
+def _pid_namespace_id() -> Optional[str]:
+    """What a claim records: the ``/proc/self/ns/pid`` inode on Linux, ``None``
+    where there is no namespace identity or the lookup failed. Stored as NULL
+    either way — a claim never asserts a namespace it cannot prove."""
+    return _local_pid_namespace().id
+
+
+def _claim_pid_checkable(
+    claim_lock: Optional[str], claim_pidns: Optional[str], *,
+    host_prefix: Optional[str] = None,
+) -> bool:
+    """True when ``worker_pid`` from this claim means anything to *this* process.
+
+    A PID is checkable only when the lock's host prefix is ours AND the
+    claim's PID namespace is provably ours. The matrix:
+
+    * host prefix differs -> False (as always);
+    * this platform has no namespace identity (macOS/Windows) -> True: the
+      hostname is the only evidence there is, unchanged from before this
+      column existed, and an unknown claim namespace stays harmless there;
+    * this platform has one but our own lookup failed -> False: unknown
+      authority, see :class:`LocalPidNamespace`;
+    * the claim recorded no namespace (pre-upgrade row, or a claim written by
+      an older binary that never stamps it) -> False: an older writer sharing
+      our hostname is indistinguishable from a sibling container sharing it,
+      so hostname is not authority across the mixed-version boundary either;
+    * both known -> equality.
+
+    Every False leaves the claim to its TTL: never probed, never signalled.
+
+    ``host_prefix`` lets a sweep hoist ``kanban_db._host_prefix`` out of its
+    row loop (one ``gethostname`` per pass). Otherwise it is read at call time
+    — ``kanban_db`` imports this module, so importing it at load would be a
+    cycle, and the late lookup keeps ``kanban_db._host_prefix`` the patch
+    point for tests.
+    """
+    if host_prefix is None:
+        from hermes_cli.kanban_db import _host_prefix
+        host_prefix = _host_prefix()
+    if not str(claim_lock or "").startswith(host_prefix):
+        return False
+    local = _local_pid_namespace()
+    if not local.supported:
+        return True
+    if local.id is None or claim_pidns is None:
+        return False
+    return str(claim_pidns) == str(local.id)

hermes_cli/kanban_swarm.py

@@ -88,7 +88,8 @@ def _activate_root_inline(
                completed_at = ?,
                claim_lock   = NULL,
                claim_expires= NULL,
-               worker_pid   = NULL
+               worker_pid   = NULL,
+               claim_pidns  = NULL
          WHERE id = ?
            AND status = 'blocked'
         """,

hermes_cli/kanban_transfer.py

@@ -77,6 +77,7 @@ def _scrub_local_state(conn: sqlite3.Connection) -> None:
         UPDATE tasks
            SET claim_lock           = NULL,
                claim_expires        = NULL,
+               claim_pidns          = NULL,
                worker_pid           = NULL,
                current_run_id       = NULL,
                last_heartbeat_at    = NULL,
@@ -140,6 +141,11 @@ def export_board(
         # The snapshot is a private file with no other writers, so plain
         # commit/close is enough — no need for the board DB's WAL dance.
         with contextlib.closing(sqlite3.connect(str(staged / "kanban.db"))) as snapshot:
+            # Nothing above opened the source through ``connect``, so a board
+            # this release has not touched yet still has the previous release's
+            # schema — and the scrub writes columns added since.
+            snapshot.row_factory = sqlite3.Row
+            kbc.migrate_schema(snapshot)
             _scrub_local_state(snapshot)
             snapshot.commit()
             counts = _count_rows(snapshot)

plugins/kanban/dashboard/plugin_api.py

@@ -730,13 +730,18 @@ def _parents_blocking_ready(conn: sqlite3.Connection, task_id: str) -> list:
 def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str) -> bool:
     """Direct status write for drag-drop moves without a structured verb (todo<->ready,
     running<->ready) + a ``status`` event. Leaving ``running`` closes the run as 'reclaimed'
-    so attempt history isn't orphaned; the worker is killed only AFTER the txn commits."""
-    terminations: list[tuple[Optional[int], Optional[str], Optional[int]]] = []
+    so attempt history isn't orphaned; the worker is killed only AFTER the txn commits.
+
+    Each termination carries the claim's PID namespace and the spawn-time
+    fingerprint as well as its lock: the dashboard often runs in a different
+    container from the worker, and there the recorded ``worker_pid`` is a
+    different process (see ``kanban_db_pidns._claim_pid_checkable``)."""
+    terminations: list[tuple[Optional[int], Optional[str], Optional[str], Optional[int]]] = []
     effective_status = new_status
     with kanban_db.write_txn(conn):
         prev = conn.execute(
-            "SELECT status, current_run_id, worker_pid, claim_lock, worker_started_at FROM tasks WHERE id = ?",
-            (task_id,)).fetchone()
+            "SELECT status, current_run_id, worker_pid, claim_lock, claim_pidns, worker_started_at "
+            "FROM tasks WHERE id = ?", (task_id,)).fetchone()
         if prev is None:
             return False
         if prev["status"] == "running" and new_status == "ready":
@@ -753,17 +758,18 @@ def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str)
             "UPDATE tasks SET status = ?, "
             "  claim_lock = CASE WHEN ? = 'running' THEN claim_lock ELSE NULL END, "
             "  claim_expires = CASE WHEN ? = 'running' THEN claim_expires ELSE NULL END, "
-            "  worker_pid = CASE WHEN ? = 'running' THEN worker_pid ELSE NULL END "
+            "  worker_pid = CASE WHEN ? = 'running' THEN worker_pid ELSE NULL END, "
+            "  claim_pidns = CASE WHEN ? = 'running' THEN claim_pidns ELSE NULL END "
             "WHERE id = ?",
-            (effective_status,) * 4 + (task_id,))
+            (effective_status,) * 5 + (task_id,))
         if cur.rowcount != 1:
             return False
         run_id = None
         if was_running and effective_status != "running" and prev["current_run_id"]:
             run_id = kanban_db._end_run(
                 conn, task_id, outcome="reclaimed", status="reclaimed",
                 summary=f"status changed to {effective_status} (dashboard/direct)")
-            terminations.append((prev["worker_pid"], prev["claim_lock"], prev["worker_started_at"]))
+            terminations.append((prev["worker_pid"], prev["claim_lock"], prev["claim_pidns"], prev["worker_started_at"]))
         conn.execute(
             "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) VALUES (?, ?, 'status', ?, ?)",
             (task_id, run_id, json.dumps({"status": effective_status, "requested_status": new_status}), int(time.time())))
@@ -772,8 +778,9 @@ def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str)
             # back worker terminations to perform post-commit.
             result = kanban_db.invalidate_descendants_for_parent_reopen(conn, task_id, author="dashboard")
             terminations.extend(result["terminations"])
-    for pid, claim_lock, started_at in terminations:
-        kanban_db._terminate_reclaimed_worker(pid, claim_lock, started_at=started_at)
+    for pid, claim_lock, claim_pidns, started_at in terminations:
+        kanban_db._terminate_reclaimed_worker(
+            pid, claim_lock, claim_pidns=claim_pidns, started_at=started_at)
     # Re-opening something may have made children stale.
     if effective_status in {"done", "ready", "review"}:
         kanban_db.recompute_ready(conn)