diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 8061801c2df39..acd30808f738c 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -892,6 +892,13 @@ def from_row(cls, row: sqlite3.Row) -> "Event": project_id TEXT, claim_lock TEXT, claim_expires INTEGER, + -- PID namespace the claim was made in (_pid_namespace_id): the caller's + -- /proc only answers for its own namespace, so a worker_pid from another + -- one must never be read as "dead". NULL = unknown (pre-upgrade rows, + -- non-Linux); kanban_db_pidns._claim_pid_checkable fails closed on it + -- when the local platform can identify namespaces, and keeps the + -- hostname-only fallback only where it cannot (macOS/Windows). + claim_pidns TEXT, tenant TEXT, result TEXT, idempotency_key TEXT, @@ -1014,6 +1021,10 @@ def from_row(cls, row: sqlite3.Row) -> "Event": -- status: running | done | blocked | crashed | timed_out | failed | released claim_lock TEXT, claim_expires INTEGER, + -- PID namespace that owned this run's claim. Retained after tasks clears + -- its live claim state so terminal-worker cleanup has the same authority + -- fence as every running-claim mutation. + claim_pidns TEXT, worker_pid INTEGER, -- Spawn-time start fingerprint of worker_pid (see tasks.worker_started_at). Retained with -- worker_pid after the run ends so a worker that outlives its terminal transition can @@ -1031,6 +1042,17 @@ def from_row(cls, row: sqlite3.Row) -> "Event": error TEXT ); +-- Mixed-version safety: pre-claim_pidns clients clear claim_lock but do not +-- know to clear the provenance column. A stale namespace must never bless a +-- later old-client claim, so clearing any live lock clears its provenance at +-- the database boundary older clients cannot skip. +CREATE TRIGGER IF NOT EXISTS clear_claim_pidns_when_lock_released +AFTER UPDATE OF claim_lock ON tasks +WHEN NEW.claim_lock IS NULL AND OLD.claim_lock IS NOT NULL AND NEW.claim_pidns IS NOT NULL +BEGIN + UPDATE tasks SET claim_pidns = NULL WHERE id = NEW.id; +END; + -- Files attached to a task (PDFs, images, source documents). The blob -- lives on disk under ``attachments_root(board)//``; -- this row carries metadata + the absolute ``stored_path`` so the @@ -1090,7 +1112,15 @@ def _new_task_id() -> str: def _claimer_id() -> str: - """Return a ``host:pid`` string that identifies this claimer.""" + """Return a ``host:pid`` string that identifies this claimer. + + Deliberately does NOT carry the PID namespace (``kanban_db_pidns._pid_namespace_id``): + the string is compared for equality by workers (``HERMES_KANBAN_CLAIM_LOCK``) + and shown in events and dashboards, and a stable hostname is what lets a + recreated container recognise its predecessor's claims. Whether a PID from + such a claim can be *probed* is a separate question — see + ``kanban_db_pidns._claim_pid_checkable``. + """ import socket try: host = socket.gethostname() or "unknown" @@ -2224,19 +2254,28 @@ def _claim_and_open_run( *, event_extra: Optional[dict] = None, ) -> Optional[int]: """CAS ``source_status -> running``, open a run row, emit ``claimed``; None - when the CAS lost. Caller holds the txn.""" + when the CAS lost. Caller holds the txn. + + Also stamps ``claim_pidns`` (``kanban_db_pidns._pid_namespace_id``) beside the lock: + only a process in that namespace can read ``worker_pid`` as evidence. It is + claim state and is cleared wherever ``claim_lock`` is, so a claim written + later by a binary that does not know the column starts from NULL instead of + inheriting the previous claimer's namespace. + """ + pidns = _kbpidns._pid_namespace_id() cur = conn.execute( f""" UPDATE tasks SET status = 'running', claim_lock = ?, claim_expires = ?, + claim_pidns = ?, started_at = COALESCE(started_at, ?) WHERE id = ? AND status = '{source_status}' AND claim_lock IS NULL """, - (lock, expires, now, task_id), + (lock, expires, pidns, now, task_id), ) if cur.rowcount != 1: return None @@ -2248,20 +2287,24 @@ def _claim_and_open_run( """ INSERT INTO task_runs ( task_id, profile, step_key, status, - claim_lock, claim_expires, max_runtime_seconds, + claim_lock, claim_expires, claim_pidns, max_runtime_seconds, started_at - ) VALUES (?, ?, ?, 'running', ?, ?, ?, ?) + ) VALUES (?, ?, ?, 'running', ?, ?, ?, ?, ?) """, ( task_id, trow["assignee"] if trow else None, trow["current_step_key"] if trow else None, - lock, expires, trow["max_runtime_seconds"] if trow else None, now, + lock, expires, pidns, trow["max_runtime_seconds"] if trow else None, now, ), ) run_id = run_cur.lastrowid conn.execute("UPDATE tasks SET current_run_id = ? WHERE id = ?", (run_id, task_id)) _append_event( conn, task_id, "claimed", - {"lock": lock, "expires": expires, "run_id": run_id, **(event_extra or {})}, run_id=run_id, + { + "lock": lock, "expires": expires, "run_id": run_id, "pidns": pidns, + **(event_extra or {}), + }, + run_id=run_id, ) return run_id @@ -2445,7 +2488,7 @@ def release_stale_claims( reclaimed = 0 host_prefix = _host_prefix() stale = conn.execute( - "SELECT id, claim_lock, worker_pid, worker_started_at, claim_expires, last_heartbeat_at, " + "SELECT id, claim_lock, claim_pidns, worker_pid, worker_started_at, claim_expires, last_heartbeat_at, " " assignee " "FROM tasks " "WHERE status = 'running' AND claim_expires IS NOT NULL " @@ -2453,18 +2496,23 @@ def release_stale_claims( ).fetchall() for row in stale: host_local = (row["claim_lock"] or "").startswith(host_prefix) + # Host-local is not enough to read this PID: see kanban_db_pidns._claim_pid_checkable. + pid_checkable = _kbpidns._claim_pid_checkable( + row["claim_lock"], row["claim_pidns"], host_prefix=host_prefix, + ) hb = row["last_heartbeat_at"] # Backstop: a heartbeat older than the max-stale threshold means no # observable progress — reclaim even if the PID is alive (logic loop). heartbeat_stale = hb is not None and (now - int(hb)) > DEFAULT_CLAIM_HEARTBEAT_MAX_STALE_SECONDS started_at = _row_get(row, "worker_started_at") - if (host_local and row["worker_pid"] and _worker_alive(row["worker_pid"], started_at) + if (pid_checkable and row["worker_pid"] and _worker_alive(row["worker_pid"], started_at) and not heartbeat_stale): _extend_live_stale_claim(conn, row, now) continue termination = _terminate_reclaimed_worker( - row["worker_pid"], row["claim_lock"], signal_fn=signal_fn, started_at=started_at, + row["worker_pid"], row["claim_lock"], + claim_pidns=row["claim_pidns"], signal_fn=signal_fn, started_at=started_at, ) # A live worker of ours must keep its claim (else a duplicate spawns beside it). if _worker_survived_termination(termination): @@ -2477,7 +2525,7 @@ def release_stale_claims( retry_status = _retry_status_for_run(conn, row["id"]) cur = conn.execute( "UPDATE tasks SET status = ?, claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL " "WHERE id = ? AND status = 'running' AND claim_lock IS ? " "AND claim_expires IS NOT NULL AND claim_expires < ? " # A worker that registered its own pid since the SELECT keeps its claim. @@ -2496,6 +2544,7 @@ def release_stale_claims( "last_heartbeat_at": _opt_int(row["last_heartbeat_at"]), "now": now, "host_local": host_local, + "pid_checkable": bool(pid_checkable), "heartbeat_stale": bool(heartbeat_stale), "retry_status": retry_status, }, @@ -2524,12 +2573,21 @@ def _record_reclaim( conn: sqlite3.Connection, task_id: str, termination: dict, *, error: str, payload: dict, ) -> Optional[int]: """Close the active run as ``reclaimed`` and emit the ``reclaimed`` event - (payload merged with the termination report). Caller holds the txn.""" + (payload merged with the termination report). Caller holds the txn. + + The caller's ``payload`` wins on key collisions. ``host_local`` and + ``pid_checkable`` exist in both dicts with different meanings: in the + termination report they answer "did we get far enough to signal", which is + hard ``False`` whenever there was no PID to signal at all, while the caller + computed them from the claim itself. Letting the report overwrite them made + a reclaimed row with ``worker_pid = NULL`` report ``host_local: false`` for + a claim that was plainly this host's — the opposite of what the diagnostic + is for. + """ run_id = _end_run( conn, task_id, outcome="reclaimed", status="reclaimed", error=error, metadata=termination, ) - payload.update(termination) - _append_event(conn, task_id, "reclaimed", payload, run_id=run_id) + _append_event(conn, task_id, "reclaimed", {**termination, **payload}, run_id=run_id) return run_id @@ -2569,7 +2627,7 @@ def reclaim_task( """Operator reclaim regardless of TTL: release the claim, restore the source phase, reset the failure counter. False when not running.""" row = conn.execute( - "SELECT status, claim_lock, worker_pid, worker_started_at FROM tasks WHERE id = ?", (task_id,), + "SELECT status, claim_lock, claim_pidns, worker_pid, worker_started_at FROM tasks WHERE id = ?", (task_id,), ).fetchone() if not row: return False @@ -2577,13 +2635,17 @@ def reclaim_task( # Nothing to reclaim — already ready / blocked / done. return False prev_lock = row["claim_lock"] + # ``claim_pidns``: the operator's reclaim releases the claim either way, but + # it must not SIGTERM a PID number it cannot verify — in another PID + # namespace that number belongs to an unrelated process. termination = _terminate_reclaimed_worker( - row["worker_pid"], prev_lock, signal_fn=signal_fn, started_at=row["worker_started_at"]) + row["worker_pid"], prev_lock, claim_pidns=row["claim_pidns"], signal_fn=signal_fn, + started_at=row["worker_started_at"]) with write_txn(conn): retry_status = _retry_status_for_run(conn, task_id) cur = conn.execute( "UPDATE tasks SET status = ?, claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL " "WHERE id = ? AND status IN ('running', 'ready', 'blocked') " "AND claim_lock IS ?", (retry_status, task_id, prev_lock), ) @@ -2790,6 +2852,7 @@ def complete_task( claim_lock = NULL, claim_expires= NULL, worker_pid = NULL, + claim_pidns = NULL, block_kind = NULL, block_recurrences = 0 WHERE id = ? @@ -3284,6 +3347,7 @@ def block_task( claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, + claim_pidns = NULL, {set_sql} WHERE id = ? AND status IN ('running', 'ready') @@ -3444,7 +3508,8 @@ def _ret(ok: bool, reason: Optional[str] = None): SET status = 'review', claim_lock = NULL, claim_expires = NULL, - worker_pid = NULL + worker_pid = NULL, + claim_pidns = NULL """ + assignee_sql + """ WHERE id = ? AND status IN ('running', 'ready') @@ -3545,7 +3610,7 @@ def request_changes( assignee = COALESCE(?, assignee), claim_lock = NULL, claim_expires = NULL, - worker_pid = NULL, worker_started_at = NULL + worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL WHERE id = ? AND status = 'running' AND current_run_id = ? """, (new_status, implementer, task_id, int(current_run_id)), @@ -3713,7 +3778,7 @@ def reopen_review_task(conn: sqlite3.Connection, task_id: str) -> bool: # consecutive_failures deliberately PRESERVED: review reopen is not # a success signal; only complete_task resets the breaker (#35072). "UPDATE tasks SET status = ?, current_run_id = NULL, " - "claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL " + "claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL " + (", assignee = ?" if implementer else "") + " WHERE id = ? AND status = 'review'", params, @@ -3746,12 +3811,15 @@ def invalidate_descendants_for_parent_reopen( action), the opposite of :func:`reopen_review_task`. Returns ``{"invalidated": [{id, prior_status, new_status, resume_status}], - "terminations": [(worker_pid, claim_lock, worker_started_at)]}``. + "terminations": [(worker_pid, claim_lock, claim_pidns, worker_started_at)]}`` — the + namespace and the spawn-time fingerprint ride along so the caller's + post-commit kill can tell a PID it may signal from one it merely shares a + hostname with, and from a recycled PID. """ caller_owns_txn = bool(conn.in_transaction) now = int(time.time()) invalidated: list[dict[str, Any]] = [] - terminations: list[tuple[Optional[int], Optional[str], Optional[int]]] = [] + terminations: list[tuple[Optional[int], Optional[str], Optional[str], Optional[int]]] = [] with write_txn(conn, allow_nested=True): rows = conn.execute( """ @@ -3762,7 +3830,8 @@ def invalidate_descendants_for_parent_reopen( FROM task_links l JOIN descendants d ON d.id = l.parent_id ) - SELECT t.id, t.status, t.current_run_id, t.worker_pid, t.claim_lock, t.worker_started_at + SELECT t.id, t.status, t.current_run_id, t.worker_pid, t.claim_lock, + t.claim_pidns, t.worker_started_at FROM descendants d JOIN tasks t ON t.id = d.id ORDER BY t.id @@ -3779,7 +3848,9 @@ def invalidate_descendants_for_parent_reopen( resume_status = "review" elif previous_status == "running": resume_status = _retry_status_for_run(conn, row["id"], row["current_run_id"]) - terminations.append((row["worker_pid"], row["claim_lock"], row["worker_started_at"])) + terminations.append( + (row["worker_pid"], row["claim_lock"], row["claim_pidns"], row["worker_started_at"]) + ) run_id = _end_run( conn, row["id"], outcome="reclaimed", status="todo", summary=f"ancestor {task_id} reopened", @@ -3788,7 +3859,7 @@ def invalidate_descendants_for_parent_reopen( # docstring for why this diverges from reopen_review_task. conn.execute( "UPDATE tasks SET status = 'todo', completed_at = NULL, " - "claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, " + "claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL, " "current_run_id = NULL, consecutive_failures = 0 WHERE id = ?", (row["id"],), ) entry = { @@ -3819,8 +3890,9 @@ def invalidate_descendants_for_parent_reopen( if not caller_owns_txn: # Standalone: committed above, audit trail durable, safe to kill now. # Composed calls leave this to the caller post-commit. - for pid, claim_lock, started_at in terminations: - _terminate_reclaimed_worker(pid, claim_lock, started_at=started_at) + for pid, claim_lock, claim_pidns, started_at in terminations: + _terminate_reclaimed_worker( + pid, claim_lock, claim_pidns=claim_pidns, started_at=started_at) return {"invalidated": invalidated, "terminations": terminations} @@ -3897,16 +3969,18 @@ def archive_task(conn: sqlite3.Connection, task_id: str, *, signal_fn=None) -> b """ with write_txn(conn): row = conn.execute( - "SELECT status, claim_lock, worker_pid, worker_started_at FROM tasks WHERE id = ?", + "SELECT status, claim_lock, claim_pidns, worker_pid, worker_started_at FROM tasks WHERE id = ?", (task_id,), ).fetchone() if not row: return False was_running = row["status"] == "running" - prev_pid, prev_lock, prev_started = row["worker_pid"], row["claim_lock"], row["worker_started_at"] + prev_pid, prev_lock, prev_pidns, prev_started = ( + row["worker_pid"], row["claim_lock"], row["claim_pidns"], row["worker_started_at"], + ) cur = conn.execute( "UPDATE tasks SET status = 'archived', " - " claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL " + " claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL " "WHERE id = ? AND status != 'archived'", (task_id,), ) if cur.rowcount != 1: @@ -3918,7 +3992,9 @@ def archive_task(conn: sqlite3.Connection, task_id: str, *, signal_fn=None) -> b ) _append_event(conn, task_id, "archived", None, run_id=run_id) if was_running: - termination = _terminate_reclaimed_worker(prev_pid, prev_lock, signal_fn=signal_fn, started_at=prev_started) + termination = _terminate_reclaimed_worker( + prev_pid, prev_lock, claim_pidns=prev_pidns, signal_fn=signal_fn, started_at=prev_started, + ) with write_txn(conn): _append_event(conn, task_id, "archive_worker_termination", termination, run_id=run_id) # ``archived`` parents no longer block children; promote them now. @@ -3970,7 +4046,8 @@ def schedule_task( SET status = 'scheduled', claim_lock = NULL, claim_expires= NULL, - worker_pid = NULL + worker_pid = NULL, + claim_pidns = NULL WHERE id = ? AND status IN ('todo', 'ready', 'running', 'blocked') """ @@ -4487,6 +4564,7 @@ def current_run_started_ats(conn: sqlite3.Connection, task_ids: Iterable[str]) - # --- Split modules (imported at the tail: they import this module as ``_kb``) --- +from hermes_cli import kanban_db_pidns as _kbpidns # noqa: E402 from hermes_cli.kanban_db_connect import ( # noqa: E402 _INITIALIZED_PATHS, init_db, diff --git a/hermes_cli/kanban_db_connect.py b/hermes_cli/kanban_db_connect.py index 0886ef472558e..be7e43c1de1c0 100644 --- a/hermes_cli/kanban_db_connect.py +++ b/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," diff --git a/hermes_cli/kanban_db_dispatch.py b/hermes_cli/kanban_db_dispatch.py index 0fc321320c3c9..e17b8fa7031ad 100644 --- a/hermes_cli/kanban_db_dispatch.py +++ b/hermes_cli/kanban_db_dispatch.py @@ -444,27 +444,54 @@ def _terminate_reclaimed_worker( pid: Optional[int], claim_lock: Optional[str], *, + claim_pidns: Optional[str], signal_fn=None, started_at=None, ) -> dict[str, Any]: - """Best-effort host-local worker termination for reclaim paths. ``started_at`` is the spawn-time - fingerprint: when the live process no longer matches it, the PID was recycled and nothing is - signalled — the worker is gone, which is what the reclaim wanted (``terminated`` = True). An - UNVERIFIED spawn (fingerprint capture failed) that is still live is never signalled either, but - it is reported as surviving (``signal_refused``) so the reclaim holds the claim instead of - spawning a duplicate beside it.""" + """Best-effort worker termination for reclaim paths. + + Signals only a PID this process can actually resolve to that worker: + ``claim_lock`` must carry our host prefix AND the claim's PID namespace must + be provably ours (:func:`_kbp._claim_pid_checkable`). A hostname match + alone is not enough — containers sharing a network namespace share the + hostname, and there the same PID number is an unrelated process, so a + SIGTERM would kill a bystander. ``claim_pidns`` is required for the same + reason: ``None`` means "the claim recorded no namespace", which on Linux is + no signal at all (the claim is left to its TTL), not hostname-only trust — + so every caller passes the row's column rather than an implicit unknown. + + ``started_at`` is the spawn-time fingerprint: when the live process no + longer matches it, the PID was recycled and nothing is signalled — the + worker is gone, which is what the reclaim wanted (``terminated`` = True). + An UNVERIFIED spawn (fingerprint capture failed) that is still live is + never signalled either, but it is reported as surviving + (``signal_refused``) so the reclaim holds the claim instead of spawning a + duplicate beside it. + + The returned ``host_local`` / ``pid_checkable`` describe how far this call + got, not the claim: both stay ``False`` when there was no PID to signal at + all. A caller that reports claim provenance computes it from the claim (see + :func:`_kb._record_reclaim`, whose merge lets the caller's values win). + """ info: dict[str, Any] = { "prev_pid": int(pid) if pid else None, "host_local": False, + "pid_checkable": False, "termination_attempted": False, "terminated": False, "sigkill": False, } if not pid or pid <= 0 or not claim_lock: return info - if not str(claim_lock).startswith(_kb._host_prefix()): + host_prefix = _kb._host_prefix() + if not str(claim_lock).startswith(host_prefix): return info info["host_local"] = True + if not _kbp._claim_pid_checkable(claim_lock, claim_pidns, host_prefix=host_prefix): + # Another PID namespace: the claim is left to the claim TTL rather than + # signalling a PID number that means something else here. + return info + info["pid_checkable"] = True kill = _kill_fn(signal_fn) if kill is None: @@ -513,7 +540,7 @@ def reap_terminal_workers(conn: sqlite3.Connection, *, signal_fn=None) -> list[s One row's failure (signal, /proc probe) is logged and skips only that row. Returns the task ids whose worker was terminated.""" rows = conn.execute( - "SELECT id, task_id, worker_pid, worker_started_at, claim_lock FROM task_runs " + "SELECT id, task_id, worker_pid, worker_started_at, claim_lock, claim_pidns FROM task_runs " "WHERE ended_at IS NOT NULL AND ended_at <= ? " "AND worker_pid IS NOT NULL AND worker_started_at IS NOT NULL", (int(time.time()) - TERMINAL_WORKER_REAP_GRACE_SECONDS,), @@ -533,7 +560,12 @@ def reap_terminal_workers(conn: sqlite3.Connection, *, signal_fn=None) -> list[s def _reap_terminal_worker_row(conn, row, host_prefix: str, signal_fn, reaped: list[str]) -> None: pid, fingerprint = int(row["worker_pid"]), row["worker_started_at"] - if pid == os.getpid() or not str(row["claim_lock"] or "").startswith(host_prefix): + claim_lock, claim_pidns = row["claim_lock"], row["claim_pidns"] + if pid == os.getpid() or not str(claim_lock or "").startswith(host_prefix): + return + # A closed run owns its PID evidence. Never probe, clear, or signal it + # unless its retained namespace proves this process can interpret it. + if not _kbp._claim_pid_checkable(claim_lock, claim_pidns, host_prefix=host_prefix): return if fingerprint == UNVERIFIED_WORKER_FINGERPRINT and _kb._pid_alive(pid): return # unproven identity: never signalled; its evidence is cleared once the pid is gone @@ -541,7 +573,7 @@ def _reap_terminal_worker_row(conn, row, host_prefix: str, signal_fn, reaped: li termination = None if alive: termination = _terminate_reclaimed_worker( - pid, row["claim_lock"], signal_fn=signal_fn, started_at=fingerprint) + pid, claim_lock, claim_pidns=claim_pidns, signal_fn=signal_fn, started_at=fingerprint) if not termination["terminated"]: return # still alive: try again next tick with _kb.write_txn(conn): @@ -564,8 +596,9 @@ def _worker_survived_termination(termination: dict) -> bool: Reclaiming then would release the claim and spawn a second worker while the first still runs — the duplication loop. Only host-local workers we actually - signalled count; a non-local lock or no-op attempt (no ``os.kill``) must fall - through to the normal release path since we cannot manage that worker anyway. + signalled count; a non-local lock, a lock from another PID namespace, or a + no-op attempt (no ``os.kill``) must fall through to the normal release path + since we cannot manage that worker anyway. """ return bool( termination.get("host_local") @@ -650,26 +683,28 @@ def enforce_max_runtime(conn: sqlite3.Connection, *, signal_fn=None) -> list[str SIGTERM, short grace, then SIGKILL. Emits ``timed_out`` and restores the task's source phase so the next tick re-spawns the same kind of worker — - unless the circuit breaker already gave up, leaving it blocked. Host-local - only (same reasoning as ``detect_crashed_workers``). ``signal_fn`` is a test hook. + unless the circuit breaker already gave up, leaving it blocked. Only claims + whose PID this process can resolve (:func:`_kbp._claim_pid_checkable`, same + reasoning as ``detect_crashed_workers``). ``signal_fn`` is a test hook. """ timed_out: list[str] = [] now = int(time.time()) - host_prefix = _kb._host_prefix() rows = conn.execute( "SELECT t.id, t.worker_pid, t.worker_started_at, " " COALESCE(r.started_at, t.started_at) AS active_started_at, " - " t.max_runtime_seconds, t.claim_lock " + " t.max_runtime_seconds, t.claim_lock, t.claim_pidns " "FROM tasks t " "LEFT JOIN task_runs r ON r.id = t.current_run_id " "WHERE t.status = 'running' AND t.max_runtime_seconds IS NOT NULL " " AND COALESCE(r.started_at, t.started_at) IS NOT NULL " " AND t.worker_pid IS NOT NULL" ).fetchall() + host_prefix = _kb._host_prefix() for row in rows: - lock = row["claim_lock"] or "" - if not lock.startswith(host_prefix): + if not _kbp._claim_pid_checkable( + row["claim_lock"], row["claim_pidns"], host_prefix=host_prefix, + ): continue # Runtime is per attempt: ``tasks.started_at`` records the FIRST start, # so retries must be measured from the active task_runs row. @@ -705,7 +740,7 @@ def enforce_max_runtime(conn: sqlite3.Connection, *, signal_fn=None) -> list[str retry_status = _kb._retry_status_for_run(conn, tid) cur = conn.execute( "UPDATE tasks SET status = ?, claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL, " "last_heartbeat_at = NULL " "WHERE id = ? AND status = 'running' " " AND worker_pid = ? AND claim_lock IS ?", @@ -756,7 +791,8 @@ def detect_stale_running( Stale = running longer than ``stale_timeout_seconds`` (active run's ``started_at``, else ``tasks.started_at``) AND ``last_heartbeat_at`` NULL or older than ``_STALE_HEARTBEAT_GAP_SECONDS``. Task returns to its source - phase, run closes ``outcome='stale'``, a live host-local worker is killed. + phase, run closes ``outcome='stale'``, a live worker whose PID this process + can resolve is killed. ``0`` disables the check; ``signal_fn`` is a test hook. Deliberately NOT counted via ``_record_task_failure``: an absent heartbeat is not a worker failure, and counting it would let long-running tasks trip the breaker. @@ -768,7 +804,7 @@ def detect_stale_running( reclaimed: list[str] = [] rows = conn.execute( - "SELECT t.id, t.worker_pid, t.worker_started_at, t.last_heartbeat_at, t.claim_lock, " + "SELECT t.id, t.worker_pid, t.worker_started_at, t.last_heartbeat_at, t.claim_lock, t.claim_pidns, " " COALESCE(r.started_at, t.started_at) AS active_started_at " "FROM tasks t " "LEFT JOIN task_runs r ON r.id = t.current_run_id " @@ -792,7 +828,9 @@ def detect_stale_running( lock = row["claim_lock"] or "" termination = _kb._terminate_reclaimed_worker( - pid, lock, signal_fn=signal_fn, started_at=_kb._row_get(row, "worker_started_at")) + pid, lock, claim_pidns=row["claim_pidns"], signal_fn=signal_fn, + started_at=_kb._row_get(row, "worker_started_at"), + ) # Never release a claim while our own worker is still alive: that would # spawn a duplicate beside it. Hold the claim and retry next tick. @@ -807,7 +845,7 @@ def detect_stale_running( retry_status = _kb._retry_status_for_run(conn, tid) cur = conn.execute( "UPDATE tasks SET status = ?, claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL, " "last_heartbeat_at = NULL " "WHERE id = ? AND status = 'running' " " AND claim_lock IS ?", @@ -872,7 +910,7 @@ def reconcile_orphaned_running(conn: sqlite3.Connection) -> list[str]: with _kb.write_txn(conn): cur = conn.execute( "UPDATE tasks SET status = 'ready', claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL, " "last_heartbeat_at = NULL " "WHERE id = ? AND status = 'running' " " AND claim_lock IS ? AND claim_expires IS ?", @@ -1135,18 +1173,29 @@ class _CrashSweep: def _reclaim_dead_workers(conn: sqlite3.Connection, board: Optional[str] = None) -> _CrashSweep: - """Release every host-local ``running`` task whose worker PID is dead.""" + """Release every ``running`` task whose worker PID is dead *and probeable*. + + "Dead" is only knowable for a claim made on this host AND in this PID + namespace (:func:`_kbp._claim_pid_checkable`). ``/proc`` answers for the + caller's namespace, so a claim from a sibling container that shares our + hostname would otherwise look like four dead workers when they are alive. + Such claims are left to the claim TTL. + + ``worker_started_at`` rides in the SELECT because the liveness probe is a + fingerprint check, not a bare PID lookup (PID-reuse guard). + """ sweep = _CrashSweep() with _kb.write_txn(conn): rows = conn.execute( - "SELECT id, worker_pid, worker_started_at, claim_lock, started_at, assignee " + "SELECT id, worker_pid, worker_started_at, claim_lock, claim_pidns, started_at, assignee " "FROM tasks " "WHERE status = 'running' AND worker_pid IS NOT NULL" ).fetchall() host_prefix = _kb._host_prefix() for row in rows: - lock = row["claim_lock"] or "" - if not lock.startswith(host_prefix): + if not _kbp._claim_pid_checkable( + row["claim_lock"], row["claim_pidns"], host_prefix=host_prefix, + ): continue # Launch-window grace so a freshly-spawned worker isn't reclaimed # before its PID is visible on /proc. @@ -1162,7 +1211,7 @@ def _reclaim_dead_workers(conn: sqlite3.Connection, board: Optional[str] = None) dead.event_payload["retry_status"] = retry_status cur = conn.execute( "UPDATE tasks SET status = ?, claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL " "WHERE id = ? AND status = 'running' " " AND worker_pid = ? AND claim_lock IS ?", (retry_status, row["id"], pid, row["claim_lock"]), @@ -1288,7 +1337,9 @@ def detect_crashed_workers(conn: sqlite3.Connection, board: Optional[str] = None """Reclaim ``running`` tasks whose worker PID is no longer alive. Restores the source phase immediately (no waiting for the claim TTL), for - tasks claimed by *this host* only — other hosts' PIDs are meaningless. + tasks whose claim was made on this host AND in this PID namespace only — + any other PID is meaningless here, whether it belongs to another host or to + a container that merely shares our hostname (``_kbp._claim_pid_checkable``). Clean exit while ``running`` is a protocol violation with a bounded violation-only retry budget; ``KANBAN_RATE_LIMIT_EXIT_CODE`` is a quota wall, released WITHOUT counting a failure and surfaced via the @@ -1393,7 +1444,7 @@ def _record_task_failure( # Spawn path: restore the claimed source phase + clear claim. conn.execute( "UPDATE tasks SET status = ?, claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, " + "claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL, " "consecutive_failures = ?, last_failure_error = ? " "WHERE id = ? AND status = 'running'", (retry_status, failures, error, task_id), @@ -1419,7 +1470,7 @@ def _record_task_failure( # state; the timeout/crash path already did. conn.execute( "UPDATE tasks SET status = 'blocked', " - + ("claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, " + + ("claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, worker_started_at = NULL, claim_pidns = NULL, " if release_claim else "") + "consecutive_failures = ?, last_failure_error = ? " "WHERE id = ? AND status IN ('running', 'ready', 'review')", @@ -2988,6 +3039,7 @@ def _handle(_signum, _frame): # Late-bound origin namespace (see module docstring); imported LAST so this # module is fully populated before ``kanban_db`` imports from it. +from hermes_cli import kanban_db_pidns as _kbp # noqa: E402 from hermes_cli import kanban_db as _kb # noqa: E402 from hermes_cli import kanban_db_connect as _kbc # noqa: E402 from hermes_cli import kanban_db_workspace as _kbw # noqa: E402 diff --git a/hermes_cli/kanban_db_pidns.py b/hermes_cli/kanban_db_pidns.py new file mode 100644 index 0000000000000..fe151a54515dc --- /dev/null +++ b/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 ``:`` (``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/`` 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) diff --git a/hermes_cli/kanban_swarm.py b/hermes_cli/kanban_swarm.py index 1a807f17c8464..becf750c007f0 100644 --- a/hermes_cli/kanban_swarm.py +++ b/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' """, diff --git a/hermes_cli/kanban_transfer.py b/hermes_cli/kanban_transfer.py index cd1dca4c13886..6ea8789a4ba74 100644 --- a/hermes_cli/kanban_transfer.py +++ b/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) diff --git a/plugins/kanban/dashboard/plugin_api.py b/plugins/kanban/dashboard/plugin_api.py index 6995d49be81cc..b01a1af8fcd31 100644 --- a/plugins/kanban/dashboard/plugin_api.py +++ b/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,9 +758,10 @@ 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 @@ -763,7 +769,7 @@ def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str) 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) diff --git a/tests/hermes_cli/test_kanban_claim_pid_namespace.py b/tests/hermes_cli/test_kanban_claim_pid_namespace.py new file mode 100644 index 0000000000000..dc430ec433ae8 --- /dev/null +++ b/tests/hermes_cli/test_kanban_claim_pid_namespace.py @@ -0,0 +1,695 @@ +"""Tests: a worker PID is only trusted inside the claimer's PID namespace. + +A claim lock is ``:`` (``kanban_db._claimer_id``) and every +reclaim pass used the hostname alone to decide that a PID it 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. + +Measured consequence: a dispatch from a sibling container found four ``running`` +claims carrying its own hostname, saw none of the worker PIDs, and closed all +four runs as crashed "pid N not alive" while the workers were alive. + +``tasks.claim_pidns`` records the namespace the claim was made in, and +``_claim_pid_checkable`` is the guard every PID probe and every SIGTERM now goes +through. A missing stored namespace fails closed when the local platform can +identify its own (Linux): the PID is unverifiable, so the claim is left to its +TTL. So does a failed lookup of our *own* namespace on Linux. The hostname-only +fallback survives only where the local platform has no namespace identity at +all (macOS/Windows, no ``/proc``). These tests patch +``kanban_db_pidns._local_pid_namespace`` — platform support and id as data — +so both sides are knowable and deterministic on any host. +""" + +from __future__ import annotations + +import importlib.util +import os +import signal +import sqlite3 +import sys +import time +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb +from hermes_cli import kanban_db_pidns as kbp +from hermes_cli import kanban_db_connect as kbc +from hermes_cli import kanban_db_dispatch as kbd + + +@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)) + # 0 = no launch-window grace, so a recorded PID is probed on the first pass. + 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 kbc.connect() as c: + yield c + + +@pytest.fixture +def local_pidns(monkeypatch): + """Make this process' PID namespace identity readable and switchable. + + ``ns["id"]`` is what ``_pid_namespace_id()`` reports; assigning a different + value mid-test is "the same board, read from a container with its own PID + namespace". ``ns["supported"]`` is whether the platform has a namespace + identity at all: ``id None`` with it ``True`` is a Linux lookup that + failed, with it ``False`` it is macOS/Windows. + """ + ns = {"id": "4026532111", "supported": True} + monkeypatch.setattr( + kbp, "_local_pid_namespace", + lambda: kbp.LocalPidNamespace(ns["id"], ns["supported"]), + ) + return ns + + +@pytest.fixture +def signals(monkeypatch): + """Record every signal any reclaim path would deliver, and deliver none.""" + sent: list[tuple[int, int]] = [] + + def _recorder(pid, sig): + sent.append((int(pid), int(sig))) + + monkeypatch.setattr(kbd, "_kill_fn", lambda signal_fn=None: signal_fn or _recorder) + return sent + + +def _host() -> str: + return kb._claimer_id().split(":", 1)[0] + + +def _running_claim(conn, *, title: str = "work", pid: int = 4242, + max_runtime_seconds=None) -> str: + """A ``running`` task claimed by this host/namespace with a recorded PID.""" + tid = kb.create_task( + conn, title=title, assignee="w", max_runtime_seconds=max_runtime_seconds, + ) + assert kb.claim_task(conn, tid, claimer=f"{_host()}:{title}") is not None + kbd._set_worker_pid(conn, tid, pid) + # Legacy-row fingerprint (worker_started_at NULL): these tests exercise the + # PID-namespace gate, not the PID-reuse guard, and a fabricated PID has no + # real spawn fingerprint to register — upstream's own reclaim tests stamp + # the same legacy shape. + conn.execute("UPDATE tasks SET worker_started_at = NULL WHERE id = ?", (tid,)) + return tid + + +def _status(conn, tid: str) -> str: + return conn.execute("SELECT status FROM tasks WHERE id = ?", (tid,)).fetchone()["status"] + + +def _pidns_of(conn, tid: str): + return conn.execute("SELECT claim_pidns FROM tasks WHERE id = ?", (tid,)).fetchone()["claim_pidns"] + + +def _events(conn, tid: str) -> list[str]: + return [e.kind for e in kb.list_events(conn, tid)] + + +def _payload(conn, tid: str, kind: str) -> dict: + for event in kb.list_events(conn, tid): + if event.kind == kind: + return event.payload or {} + raise AssertionError(f"no {kind!r} event on {tid}: {_events(conn, tid)}") + + +def _expire_claim(conn, tid: str) -> None: + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", (int(time.time()) - 600, tid), + ) + + +def _backdate_start(conn, tid: str, seconds: int) -> None: + then = int(time.time()) - seconds + with kb.write_txn(conn): + conn.execute("UPDATE tasks SET started_at = ? WHERE id = ?", (then, tid)) + conn.execute( + "UPDATE task_runs SET started_at = ? WHERE task_id = ? AND ended_at IS NULL", + (then, tid), + ) + + +# --- The namespace identity itself ---------------------------------------- + +def test_parse_pid_namespace_link_extracts_the_inode(): + """``readlink`` text in, inode out — a pure mapping, testable anywhere.""" + assert kbp._parse_pid_namespace_link("pid:[4026532534]") == "4026532534" + assert kbp._parse_pid_namespace_link("pid:[1]") == "1" + assert kbp._parse_pid_namespace_link("") is None + assert kbp._parse_pid_namespace_link("pid:[]") is None + assert kbp._parse_pid_namespace_link("not a namespace link") is None + + +@pytest.mark.platforms("linux") +def test_pid_namespace_id_reads_proc_self_ns_pid(monkeypatch): + monkeypatch.setattr(kbp, "_LOCAL_PID_NS", None) + local = kbp._local_pid_namespace() + assert local.supported is True + assert local.id is not None and local.id.isdigit() + assert kbp._pid_namespace_id() == local.id + # Same process, same namespace — and the cached answer is stable. + assert kbp._local_pid_namespace() is local + + +@pytest.mark.platforms("linux") +def test_linux_namespace_lookup_failure_is_not_the_unsupported_platform(monkeypatch): + """``/proc/self/ns/pid`` unreadable (restricted or unmounted ``/proc``). + + Linux *has* a namespace identity; failing to read ours is unknown + authority, not "no such thing here" — so no hostname-only fallback, and + the failure is not cached: the next call asks again. + """ + monkeypatch.setattr(kbp, "_LOCAL_PID_NS", None) + real_readlink = os.readlink + proc = {"readable": False} + + def _readlink(path, *args, **kwargs): + if str(path) == "/proc/self/ns/pid" and not proc["readable"]: + raise PermissionError(13, "Permission denied", path) + return real_readlink(path, *args, **kwargs) + + monkeypatch.setattr(os, "readlink", _readlink) + assert kbp._local_pid_namespace() == kbp.LocalPidNamespace(None, True) + assert kbp._pid_namespace_id() is None + host_lock = f"{_host()}:1" + assert kbp._claim_pid_checkable(host_lock, "4026532534") is False + assert kbp._claim_pid_checkable(host_lock, None) is False + + proc["readable"] = True + assert kbp._local_pid_namespace().id is not None + + +def _assert_no_namespace_identity_keeps_hostname_semantics(monkeypatch): + """Off Linux there is no namespace identity, and that must stay harmless.""" + monkeypatch.setattr(kbp, "_LOCAL_PID_NS", None) + assert kbp._local_pid_namespace() == kbp.LocalPidNamespace(None, False) + assert kbp._pid_namespace_id() is None + # No local identity => today's hostname-only semantics, unchanged. + assert kbp._claim_pid_checkable(f"{_host()}:1", "4026532534") is True + assert kbp._claim_pid_checkable(f"{_host()}:1", None) is True + assert kbp._claim_pid_checkable("other-host:1", None) is False + + +@pytest.mark.platforms("macos") +def test_pid_namespace_unsupported_on_macos(monkeypatch): + _assert_no_namespace_identity_keeps_hostname_semantics(monkeypatch) + + +@pytest.mark.platforms("windows") +def test_pid_namespace_unsupported_on_windows(monkeypatch): + _assert_no_namespace_identity_keeps_hostname_semantics(monkeypatch) + + +def test_claim_pid_checkable_matrix(local_pidns): + host_lock = f"{_host()}:123" + assert kbp._claim_pid_checkable(host_lock, local_pidns["id"]) is True + # Local namespace knowable + stored namespace missing (pre-upgrade row or + # an older writer): fail closed — the PID is unverifiable. + assert kbp._claim_pid_checkable(host_lock, None) is False # rollout NULL + assert kbp._claim_pid_checkable(host_lock, "4026531836") is False # other namespace + assert kbp._claim_pid_checkable("other-host:123", local_pidns["id"]) is False + assert kbp._claim_pid_checkable(None, None) is False + + +def test_claim_pid_checkable_without_namespace_support_keeps_hostname_semantics(local_pidns): + """macOS/Windows (no /proc): hostname-only fallback survives intact. + + When the local platform has no namespace identity at all, a NULL stored + namespace must not freeze recovery — the old semantics are all we have. + """ + host_lock = f"{_host()}:123" + local_pidns["id"], local_pidns["supported"] = None, False + assert kbp._claim_pid_checkable(host_lock, None) is True # rollout NULL + assert kbp._claim_pid_checkable(host_lock, "4026532534") is True # any claim ns + assert kbp._claim_pid_checkable("other-host:1", None) is False # host still gates + + +def test_claim_pid_checkable_fails_closed_when_own_lookup_failed(local_pidns): + """Linux, but ``/proc/self/ns/pid`` could not be read: unknown authority. + + Absence of our own provenance cannot become provenance. Nothing host-local + is checkable until the lookup succeeds; the TTL path recovers the claims. + """ + host_lock = f"{_host()}:123" + local_pidns["id"] = None # lookup failed; supported stays True + assert kbp._claim_pid_checkable(host_lock, "4026532534") is False + assert kbp._claim_pid_checkable(host_lock, None) is False + assert kbp._claim_pid_checkable("other-host:1", "4026532534") is False + + +# --- Recording the namespace on the claim --------------------------------- + +def test_claim_records_pid_namespace(conn, local_pidns): + tid = kb.create_task(conn, title="stamped", assignee="w") + assert kb.claim_task(conn, tid) is not None + + assert _pidns_of(conn, tid) == local_pidns["id"] + assert _payload(conn, tid, "claimed")["pidns"] == local_pidns["id"] + + +# --- detect_crashed_workers ------------------------------------------------ + +def test_dead_pid_in_another_namespace_is_not_crashed(conn, local_pidns, monkeypatch, signals): + """The measured incident: four live workers closed as crashed.""" + tid = _running_claim(conn, title="live-elsewhere") + # Now read the board from a container that shares the hostname only. + local_pidns["id"] = "4026531836" + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + + assert kbd.detect_crashed_workers(conn) == [] + assert _status(conn, tid) == "running" + assert "crashed" not in _events(conn, tid) + run = conn.execute( + "SELECT status, ended_at FROM task_runs WHERE task_id = ?", (tid,) + ).fetchone() + assert run["status"] == "running" and run["ended_at"] is None + assert signals == [] + + +def test_dead_pid_in_the_same_namespace_still_crashes(conn, local_pidns, monkeypatch, signals): + """Control: the recovery this guard must not break.""" + tid = _running_claim(conn, title="really-dead") + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + + assert kbd.detect_crashed_workers(conn) == [tid] + assert _status(conn, tid) == "ready" + assert "crashed" in _events(conn, tid) + # The stamp is claim state: it leaves with the lock. + assert _pidns_of(conn, tid) is None + + +def _null_pidns(conn, tid: str) -> None: + with kb.write_txn(conn): + conn.execute("UPDATE tasks SET claim_pidns = NULL WHERE id = ?", (tid,)) + + +def test_migrated_null_pidns_claim_is_not_crashed_and_released_by_ttl( + conn, local_pidns, monkeypatch, signals, +): + """Rollout (a): a pre-upgrade row survives with ``claim_pidns = NULL``. + + The migration adds the column without backfilling, so every live claim + from before the upgrade reads NULL. Read from a sibling container that + shares our hostname: the crash sweep must neither probe the PID nor + signal it (``/proc`` there is another world), and the row is released by + the claim TTL instead. + """ + tid = _running_claim(conn, title="migrated") + _null_pidns(conn, tid) + local_pidns["id"] = "4026531836" # reader is a sibling container + probes: list[int] = [] + monkeypatch.setattr(kb, "_pid_alive", lambda pid: probes.append(int(pid)) or False) + + assert kbd.detect_crashed_workers(conn) == [] + assert _status(conn, tid) == "running" + assert "crashed" not in _events(conn, tid) + assert probes == [] and signals == [] + + # TTL-safe release: once the claim expires, the sweep reclaims without + # probing or signalling — an unknown namespace cannot certify the PID. + _expire_claim(conn, tid) + assert kb.release_stale_claims(conn) == 1 + assert _status(conn, tid) == "ready" + assert probes == [] and signals == [] + payload = _payload(conn, tid, "reclaimed") + assert payload["pid_checkable"] is False + assert payload["termination_attempted"] is False + + +def _claim_like_an_older_binary(conn, tid: str, *, pid: int) -> None: + """What a pre-``claim_pidns`` ``_claim_and_open_run`` writes: status, + lock, expiry, start — and nothing about the namespace column it does not + know exists.""" + now = int(time.time()) + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET status = 'running', claim_lock = ?, claim_expires = ?, " + "started_at = COALESCE(started_at, ?), worker_pid = ? " + "WHERE id = ? AND claim_lock IS NULL", + (f"{_host()}:older", now + 900, now, pid, tid), + ) + conn.execute( + "INSERT INTO task_runs (task_id, status, claim_lock, claim_expires, " + "worker_pid, started_at) VALUES (?, 'running', ?, ?, ?, ?)", + (tid, f"{_host()}:older", now + 900, pid, now), + ) + run_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0] + conn.execute("UPDATE tasks SET current_run_id = ? WHERE id = ?", (run_id, tid)) + + +def test_older_writer_null_pidns_claim_is_never_signalled( + conn, local_pidns, monkeypatch, signals, +): + """Rollout (b): an older CLI keeps authoring NULL claims post-migration. + + The real sequence, not a hand-nulled column: this binary claims and + releases the task (the release must clear ``claim_pidns`` with the lock), + then an older binary sharing the board claims it without touching the + column. Same hostname, even the same namespace as the reader — but the + claim does not say so, and hostname is not authority across the + mixed-version boundary. The PID must not be probed or signalled; an + operator reclaim still releases the claim, it just cannot certify the kill. + """ + probes: list[int] = [] + monkeypatch.setattr(kb, "_pid_alive", lambda pid: probes.append(int(pid)) or False) + tid = _running_claim(conn, title="older-writer", pid=1111) + assert _pidns_of(conn, tid) == local_pidns["id"] + assert kb.reclaim_task(conn, tid, reason="hand back") is True + # Same namespace, stamped: that kill is legitimate and not what is under test. + assert signals == [(1111, int(signal.SIGTERM))] + assert _pidns_of(conn, tid) is None, "release must not leave a stale namespace behind" + signals.clear() + probes.clear() + + _claim_like_an_older_binary(conn, tid, pid=4242) + assert _pidns_of(conn, tid) is None + + assert kbd.detect_crashed_workers(conn) == [] + assert _status(conn, tid) == "running" + assert probes == [] and signals == [] + + # Operator reclaim releases the claim either way, but must not SIGTERM a + # PID number it cannot verify. + assert kb.reclaim_task(conn, tid, reason="operator asked") is True + assert _status(conn, tid) == "ready" + assert probes == [] and signals == [] + + +def test_older_release_clears_pidns_before_an_older_claim(conn, local_pidns, monkeypatch, signals): + """The DB trigger fences the old-release -> old-claim mixed-version path. + + A pre-column release only clears ``claim_lock``. Its successor cannot be + trusted with the prior claimant's namespace stamp, even when the reader is + back in that former namespace. + """ + probes: list[int] = [] + monkeypatch.setattr(kb, "_pid_alive", lambda pid: probes.append(int(pid)) or False) + tid = _running_claim(conn, title="old-release", pid=1111) + assert _pidns_of(conn, tid) == local_pidns["id"] + + # Pre-claim_pidns release SQL: it cannot name the new column. The trigger + # clears the stale stamp at the only boundary this older writer does touch. + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET status = 'ready', claim_lock = NULL, claim_expires = NULL, " + "worker_pid = NULL, worker_started_at = NULL, current_run_id = NULL WHERE id = ?", + (tid,), + ) + assert _pidns_of(conn, tid) is None + + _claim_like_an_older_binary(conn, tid, pid=4242) + assert _pidns_of(conn, tid) is None + assert kbd.detect_crashed_workers(conn) == [] + assert probes == [] and signals == [] + + +def test_archive_running_claim_uses_snapshot_namespace(conn, local_pidns, monkeypatch, signals): + """Archive snapshots provenance before clearing tasks and still cleans up.""" + monkeypatch.setattr(kb, "_pid_alive", lambda _pid: False) + local = local_pidns["id"] + local_tid = _running_claim(conn, title="archive-local", pid=1111) + assert kb.archive_task(conn, local_tid) is True + assert signals == [(1111, int(signal.SIGTERM))] + local_payload = _payload(conn, local_tid, "archive_worker_termination") + assert local_payload["pid_checkable"] is True + + signals.clear() + local_pidns["id"] = local + foreign_tid = _running_claim(conn, title="archive-foreign", pid=2222) + local_pidns["id"] = "4026531836" + assert kb.archive_task(conn, foreign_tid) is True + assert signals == [] + foreign_payload = _payload(conn, foreign_tid, "archive_worker_termination") + assert foreign_payload["pid_checkable"] is False + assert _status(conn, foreign_tid) == "archived" + + +def test_failed_local_namespace_lookup_neither_probes_nor_signals( + conn, local_pidns, monkeypatch, signals, +): + """A Linux reader whose own ``/proc/self/ns/pid`` lookup failed. + + The claim is fully stamped; it is *our* provenance that is missing. That + must not re-open hostname-only trust: no probe, no signal, TTL release. + """ + tid = _running_claim(conn, title="unreadable-proc") + local_pidns["id"] = None # supported stays True: Linux, lookup failed + probes: list[int] = [] + monkeypatch.setattr(kb, "_pid_alive", lambda pid: probes.append(int(pid)) or False) + + assert kbd.detect_crashed_workers(conn) == [] + assert _status(conn, tid) == "running" + assert probes == [] and signals == [] + + _expire_claim(conn, tid) + assert kb.release_stale_claims(conn) == 1 + assert _status(conn, tid) == "ready" + assert probes == [] and signals == [] + assert _payload(conn, tid, "reclaimed")["pid_checkable"] is False + + +# --- release_stale_claims -------------------------------------------------- + +def test_stale_claim_from_another_namespace_is_released_at_ttl_without_signals( + conn, local_pidns, monkeypatch, signals, +): + """Not extended as "pid alive", not signalled — released by the TTL.""" + tid = _running_claim(conn, title="foreign-stale") + _expire_claim(conn, tid) + local_pidns["id"] = "4026531836" + # A PID number that happens to exist in THIS namespace says nothing about + # the worker, so it must neither extend the claim nor be killed. + monkeypatch.setattr(kb, "_pid_alive", lambda pid: True) + + assert kb.release_stale_claims(conn) == 1 + assert _status(conn, tid) == "ready" + assert "claim_extended" not in _events(conn, tid) + assert signals == [] + payload = _payload(conn, tid, "reclaimed") + assert payload["host_local"] is True + assert payload["pid_checkable"] is False + + +def test_reclaim_payload_reports_the_claim_not_the_kill( + conn, local_pidns, monkeypatch, signals, +): + """``host_local`` / ``pid_checkable`` describe the claim, not the signal. + + The termination report carries keys of the same name meaning "did we get far + enough to signal", so a merge in the wrong order made this payload lie. + """ + tid = _running_claim(conn, title="same-ns-dead") + _expire_claim(conn, tid) + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + + assert kb.release_stale_claims(conn) == 1 + payload = _payload(conn, tid, "reclaimed") + assert payload["host_local"] is True + assert payload["pid_checkable"] is True + # The termination keys still report the signal that did happen. + assert payload["termination_attempted"] is True + assert signals == [(4242, int(signal.SIGTERM))] + assert _pidns_of(conn, tid) is None + + +def test_reclaim_payload_without_a_worker_pid_still_reports_the_claim( + conn, local_pidns, signals, +): + """No PID to signal must not read as "this claim was not ours".""" + tid = _running_claim(conn, title="no-pid") + with kb.write_txn(conn): + conn.execute("UPDATE tasks SET worker_pid = NULL WHERE id = ?", (tid,)) + _expire_claim(conn, tid) + + assert kb.release_stale_claims(conn) == 1 + payload = _payload(conn, tid, "reclaimed") + assert payload["host_local"] is True + assert payload["pid_checkable"] is True + assert payload["worker_pid"] is None + assert payload["termination_attempted"] is False + assert signals == [] + + +def test_stale_claim_in_the_same_namespace_is_extended_while_alive( + conn, local_pidns, monkeypatch, signals, +): + """Control: a live local worker still gets its claim extended, not reclaimed.""" + tid = _running_claim(conn, title="local-slow") + _expire_claim(conn, tid) + monkeypatch.setattr(kb, "_pid_alive", lambda pid: True) + + assert kb.release_stale_claims(conn) == 0 + assert _status(conn, tid) == "running" + assert "claim_extended" in _events(conn, tid) + assert signals == [] + + +# --- enforce_max_runtime / detect_stale_running --------------------------- + +def test_max_runtime_does_not_signal_another_namespace(conn, local_pidns, signals): + tid = _running_claim(conn, title="overrun", max_runtime_seconds=1) + _backdate_start(conn, tid, 600) + local_pidns["id"] = "4026531836" + + assert kbd.enforce_max_runtime(conn) == [] + assert _status(conn, tid) == "running" + assert signals == [] + + +def test_max_runtime_still_kills_in_the_same_namespace(conn, local_pidns, monkeypatch, signals): + tid = _running_claim(conn, title="overrun-local", max_runtime_seconds=1) + _backdate_start(conn, tid, 600) + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + + assert kbd.enforce_max_runtime(conn) == [tid] + assert signals == [(4242, int(signal.SIGTERM))] + + +def test_stale_running_reclaims_but_never_signals_another_namespace( + conn, local_pidns, signals, +): + """The heartbeat sweep is namespace-independent; only its kill is withheld.""" + tid = _running_claim(conn, title="no-heartbeat") + _backdate_start(conn, tid, 7200) + local_pidns["id"] = "4026531836" + + assert kbd.detect_stale_running(conn, stale_timeout_seconds=60) == [tid] + assert _status(conn, tid) == "ready" + assert signals == [] + + +# --- Operator surfaces: hermes kanban reclaim / dashboard ----------------- + +def test_operator_reclaim_releases_without_signalling_another_namespace( + conn, local_pidns, signals, +): + tid = _running_claim(conn, title="operator") + local_pidns["id"] = "4026531836" + + assert kb.reclaim_task(conn, tid, reason="operator asked") is True + assert _status(conn, tid) == "ready" + assert signals == [] + + +def test_operator_reclaim_still_kills_its_own_worker(conn, local_pidns, monkeypatch, signals): + tid = _running_claim(conn, title="operator-local") + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + + assert kb.reclaim_task(conn, tid, reason="operator asked") is True + assert signals == [(4242, int(signal.SIGTERM))] + + +def _load_dashboard_plugin(): + """Import plugins/kanban/dashboard/plugin_api.py as a module.""" + repo_root = Path(__file__).resolve().parents[2] + plugin_file = repo_root / "plugins" / "kanban" / "dashboard" / "plugin_api.py" + spec = importlib.util.spec_from_file_location( + "hermes_dashboard_plugin_kanban_pidns_test", plugin_file, + ) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +def test_dashboard_status_change_does_not_signal_another_namespace( + conn, local_pidns, signals, +): + """Dragging a card out of Running from the dashboard container. + + The dashboard usually runs beside the gateway, not in it (its own PID + namespace), so the ``worker_pid`` it reads is another process' number here. + """ + plugin_api = _load_dashboard_plugin() + tid = _running_claim(conn, title="dragged") + local_pidns["id"] = "4026531836" + + assert plugin_api._set_status_direct(conn, tid, "ready") is True + assert _status(conn, tid) == "ready" + assert signals == [] + assert _pidns_of(conn, tid) is None, "release must not leave a stale namespace behind" + + +def test_dashboard_status_change_still_kills_its_own_worker( + conn, local_pidns, monkeypatch, signals, +): + plugin_api = _load_dashboard_plugin() + tid = _running_claim(conn, title="dragged-local") + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + + assert plugin_api._set_status_direct(conn, tid, "ready") is True + assert signals == [(4242, int(signal.SIGTERM))] + assert _pidns_of(conn, tid) is None + + +# --- Schema migration ------------------------------------------------------ + +def test_migration_adds_claim_pidns_to_a_legacy_board(tmp_path, kanban_home, local_pidns): + """A pre-column board gains ``claim_pidns`` and keeps working.""" + db_path = tmp_path / "legacy-kanban.db" + legacy = sqlite3.connect(str(db_path)) + legacy.execute( + """ + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, + title TEXT NOT NULL, + body TEXT, + assignee TEXT, + status TEXT NOT NULL, + priority INTEGER NOT NULL DEFAULT 0, + created_by TEXT, + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER, + workspace_kind TEXT NOT NULL DEFAULT 'scratch', + workspace_path TEXT, + claim_lock TEXT, + claim_expires INTEGER + ) + """ + ) + legacy.execute( + """ + CREATE TABLE task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + kind TEXT NOT NULL, + payload TEXT, + created_at INTEGER NOT NULL + ) + """ + ) + legacy.execute( + "INSERT INTO tasks (id, title, status, created_at) " + "VALUES ('legacy', 'old board task', 'ready', 1)" + ) + legacy.commit() + legacy.close() + + with kbc.connect(db_path) as migrated: + columns = {row["name"] for row in migrated.execute("PRAGMA table_info(tasks)")} + assert "claim_pidns" in columns + # And the migrated row is claimable, stamping the namespace as usual. + assert kb.claim_task(migrated, "legacy") is not None + assert migrated.execute( + "SELECT claim_pidns FROM tasks WHERE id = 'legacy'" + ).fetchone()["claim_pidns"] == local_pidns["id"] diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index c82d6ff4531cb..c8813887db82c 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -1121,11 +1121,14 @@ def test_reclaim_task_resets_running_to_ready(kanban_home, monkeypatch): import time import secrets import hermes_cli.kanban_db as _kb + import hermes_cli.kanban_db_pidns as _kbp conn = kbc.connect() try: t = kb.create_task(conn, title="stuck", assignee="broken") - # Simulate a live claim (not expired). + # Simulate a live claim (not expired), stamped the way a real claimer + # stamps it — including the PID namespace, so the PID is checkable. lock = f"{_kb._claimer_id().split(':', 1)[0]}:{secrets.token_hex(8)}" + pidns = _kbp._pid_namespace_id() future = int(time.time()) + 3600 killed: list[int] = [] state = {"alive": True} @@ -1138,8 +1141,8 @@ def _signal(pid, sig): monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: state["alive"]) conn.execute( "UPDATE tasks SET status='running', claim_lock=?, claim_expires=?, " - "worker_pid=? WHERE id=?", - (lock, future, 12345, t), + "worker_pid=?, claim_pidns=? WHERE id=?", + (lock, future, 12345, pidns, t), ) conn.execute( "INSERT INTO task_runs (task_id, status, claim_lock, claim_expires, " diff --git a/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py b/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py index 7fbd660c4959b..5deb1220bebf3 100644 --- a/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py +++ b/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py @@ -105,7 +105,7 @@ def test_running_descendant_event_precedes_termination_via_reclaim_helper( kills: list[tuple] = [] - def fake_terminate(pid, claim_lock, started_at=None, **kwargs): + def fake_terminate(pid, claim_lock, *, claim_pidns=None, started_at=None, **kwargs): # The audit trail must already be durable when the kill fires: # standalone calls commit before terminating. side = kbc.connect(tmp_path / "kanban.db") @@ -114,7 +114,10 @@ def fake_terminate(pid, claim_lock, started_at=None, **kwargs): finally: side.close() assert "descendant_invalidated" in kinds - kills.append((pid, claim_lock, started_at)) + # The claim's PID namespace and the spawn-time fingerprint travel with + # the PID so the callee can tell whether that number is probeable here + # at all and whether it was recycled. + kills.append((pid, claim_lock, claim_pidns, started_at)) return {"terminated": True} monkeypatch.setattr(kb, "_terminate_reclaimed_worker", fake_terminate) diff --git a/tests/hermes_cli/test_kanban_terminal_worker_reaper.py b/tests/hermes_cli/test_kanban_terminal_worker_reaper.py index bf621e8362146..9eea6cbe7b1e4 100644 --- a/tests/hermes_cli/test_kanban_terminal_worker_reaper.py +++ b/tests/hermes_cli/test_kanban_terminal_worker_reaper.py @@ -22,6 +22,7 @@ from hermes_cli import kanban_db as kb from hermes_cli import kanban_db_connect as kbc from hermes_cli import kanban_db_dispatch as kbd +from hermes_cli import kanban_db_pidns as kbp @pytest.fixture @@ -147,3 +148,34 @@ def signal_fn(pid, sig): for p in (broken, healthy): p.kill() p.wait() + + +def test_foreign_namespace_terminal_worker_keeps_run_evidence(conn, monkeypatch): + """A sibling namespace cannot probe, clear, or signal a closed run's PID.""" + proc = _sleeper() + try: + tid, run_id = _completed_card_with_worker(conn, proc) + run = conn.execute( + "SELECT claim_pidns, worker_pid FROM task_runs WHERE id = ?", (run_id,), + ).fetchone() + assert run["claim_pidns"] is not None and run["worker_pid"] == proc.pid + monkeypatch.setattr( + kbp, "_local_pid_namespace", lambda: kbp.LocalPidNamespace("foreign", True), + ) + monkeypatch.setattr( + kbd, "_worker_alive", lambda *_args: pytest.fail("foreign PID was probed"), + ) + signals = [] + + assert kbd.reap_terminal_workers( + conn, signal_fn=lambda pid, sig: signals.append((pid, sig)), + ) == [] + run = conn.execute( + "SELECT worker_pid, worker_started_at FROM task_runs WHERE id = ?", (run_id,), + ).fetchone() + assert run["worker_pid"] == proc.pid and run["worker_started_at"] is not None + assert signals == [] + finally: + if proc.poll() is None: + proc.kill() + proc.wait() diff --git a/tests/hermes_cli/test_kanban_transfer.py b/tests/hermes_cli/test_kanban_transfer.py index 9e437d8ed243e..b70ba63d899a3 100644 --- a/tests/hermes_cli/test_kanban_transfer.py +++ b/tests/hermes_cli/test_kanban_transfer.py @@ -17,6 +17,7 @@ from __future__ import annotations import json +import sqlite3 import sys import tarfile import time @@ -131,6 +132,31 @@ def test_round_trip_preserves_content(kanban_root, tmp_path): assert tasks["scratch task"]["assignee"] == "coder" +def test_export_migrates_a_board_this_release_has_not_opened(kanban_root, tmp_path): + """``boards export`` runs before any other command touches the board. + + Nothing on the export path opens the source through ``connect``, so the + snapshot carries the previous release's schema. The scrub writes columns + added since (``claim_pidns``); the snapshot has to be migrated first, or + the first post-upgrade backup dies with "no such column". + """ + _seed_board() + kb._INITIALIZED_PATHS.clear() + # The board as the previous release left it: one additive column short. + # Its schema also predates the trigger that guards that column. + with sqlite3.connect(str(kb.kanban_db_path("alpha"))) as raw: + raw.execute("DROP TRIGGER IF EXISTS clear_claim_pidns_when_lock_released") + raw.execute("ALTER TABLE tasks DROP COLUMN claim_pidns") + + archive = kt.export_board("alpha", str(tmp_path / "alpha"))["archive"] + + kanban_root("target") + result = kt.import_board(archive) + assert result["counts"]["tasks"] == 2 + tasks = _tasks_by_title(result["board"]) + assert all(row["claim_pidns"] is None for row in tasks.values()) + + def test_attachment_blob_travels_and_is_readable(kanban_root, tmp_path): _seed_board() archive = kt.export_board("alpha", str(tmp_path / "alpha"))["archive"] diff --git a/website/docs/user-guide/docker.md b/website/docs/user-guide/docker.md index 525d3028a3d13..29ce6b9edd4e3 100644 --- a/website/docs/user-guide/docker.md +++ b/website/docs/user-guide/docker.md @@ -175,7 +175,7 @@ If no provider is registered and the bind is non-loopback, the dashboard **fails An unauthenticated public dashboard was the entry point for the June 2026 MCP-config persistence campaign: internet scanners reached exposed dashboards (and OpenAI API servers) and drove the agent into planting an SSH-key backdoor. The auth gate is now mandatory on every non-loopback bind. For a trusted-LAN / homelab box, the bundled username/password provider (`HERMES_DASHBOARD_BASIC_AUTH_USERNAME` + `_PASSWORD`) is the zero-infra way to satisfy it. ::: -Running the dashboard as a separate container **is** supported when that container shares the host PID and network namespace (e.g. `network_mode: host`, as the repo's own `docker-compose.yml` does — see its `dashboard` service). Its gateway-liveness detection requires a shared PID namespace with the gateway process, so the limitation only applies to dashboards run in isolated bridge-network containers without a shared PID namespace. +Running the dashboard as a separate container **is** supported when that container shares the host PID and network namespace (e.g. `network_mode: host`, as the repo's own `docker-compose.yml` does — see its `dashboard` service). Its gateway-liveness detection requires a shared PID namespace with the gateway process, so the limitation only applies to dashboards run in isolated bridge-network containers without a shared PID namespace. The same sharing (`pid: "service:gateway"` in Compose) is what lets the dashboard's dispatch nudge reclaim the gateway's crashed workers on the first tick; without it, kanban treats the gateway's worker PIDs as unverifiable from the dashboard and leaves those claims to the claim TTL rather than guessing. ## Running interactively (CLI chat) diff --git a/website/docs/user-guide/features/kanban-worker-lanes.md b/website/docs/user-guide/features/kanban-worker-lanes.md index 6507462f9e226..4d7494e1d8c80 100644 --- a/website/docs/user-guide/features/kanban-worker-lanes.md +++ b/website/docs/user-guide/features/kanban-worker-lanes.md @@ -128,8 +128,8 @@ The historical issue for this is [#19931](https://github.com/NousResearch/hermes So lane authors don't have to reimplement these: -- **Stale claim TTL** — a worker that claims and then never heartbeats / completes / blocks gets reclaimed after `DEFAULT_CLAIM_TTL_SECONDS` (15 min default) — but only if the worker process has actually died. A live worker (slow model spending 20+ min in one tool-free LLM call) gets the claim *extended* instead of killed; only a dead PID is reclaimed. -- **Crashed worker** — a worker whose host-local PID has vanished is detected by `detect_crashed_workers` and reaped; the task increments `consecutive_failures` and may auto-block when the breaker trips. +- **Stale claim TTL** — a worker that claims and then never heartbeats / completes / blocks gets reclaimed after `DEFAULT_CLAIM_TTL_SECONDS` (15 min default) — but only if the worker process has actually died. A live worker (slow model spending 20+ min in one tool-free LLM call) gets the claim *extended* instead of killed; only a dead PID is reclaimed. A PID the dispatcher cannot verify (claim from another host or PID namespace, or one with no recorded namespace) is neither extended nor signalled — the claim is released at the TTL. +- **Crashed worker** — a worker whose PID has vanished from the dispatcher's own host *and* PID namespace is detected by `detect_crashed_workers` and reaped; the task increments `consecutive_failures` and may auto-block when the breaker trips. - **Run-level retry** — when a task is retried (post-block, post-crash, post-reclaim), the worker can use the `expected_run_id` parameter on terminating tools to fail fast if its own run was already superseded. - **Per-task max runtime** — `task.max_runtime_seconds` hard-caps wall-clock time per run, regardless of PID liveness. Catches genuinely-deadlocked workers that the live-PID extension would otherwise keep running. - **Stranded-task detection** — a ready task whose assignee never produces a claim within `kanban.stranded_threshold_seconds` (default 30 min) shows up in `hermes kanban diagnostics` as a `stranded_in_ready` warning. Severity escalates to error at 2x the threshold and critical at 6x. Catches typo'd assignees, deleted profiles, and down external worker pools in one signal — identity-agnostic, no per-board allowlist to curate. diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index b682fb89176df..4bcc8d5e7548b 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -1395,7 +1395,7 @@ Every transition appends a row to `task_events`. Each row carries an optional `r |---|---|---| | `created` | `{assignee, status, parents, tenant}` | Task inserted. `run_id` is `NULL`. | | `promoted` | — | `todo → ready` because all parents hit `done`. `run_id` is `NULL`. | -| `claimed` | `{lock, expires, run_id}` | Dispatcher atomically claimed a `ready` task for spawn. | +| `claimed` | `{lock, expires, run_id, pidns}` | Dispatcher atomically claimed a `ready` task for spawn. `pidns` is the claimer's PID namespace (Linux: the `/proc/self/ns/pid` inode; `null` where there is none) — only a reader in that namespace may treat the worker PID as evidence of liveness or death. | | `completed` | `{result_len, summary?}` | Worker wrote `--result` / `--summary` and task hit `done`. `summary` is the first-line handoff (400-char cap); full version lives on the run row. If `complete_task` is called on a never-claimed task with handoff fields, a zero-duration run is synthesized so `run_id` still points at something. | | `blocked` | `{reason, kind, recurrences}` | Worker or human flipped the task to `blocked`. `kind` is the typed block reason (`needs_input`, `capability`, `transient`, or `null` for a generic block); `recurrences` is the unblock-loop counter. A `kind=dependency` block with no incomplete parent lands here as `needs_input` (payload adds `requested_kind: dependency`, `rekind_reason: no_open_parent`) because `todo` would only get it re-promoted and respawned on the next dispatch tick. Synthesizes a zero-duration run when called on a never-claimed task with `--reason`. | | `dependency_wait` | `{reason, kind}` or `{reason: parent_not_done, demoted: true, parent}` | Worker blocked with `kind=dependency` while at least one parent is still open — the task is only waiting on another task, so it routes to `todo` (parent-gated, auto-promoted) instead of `blocked` and no recurrence is counted. No human needed. Also emitted when `link`/`kanban_link` puts a `ready` child under a parent that is not `done`: the child drops back to `todo` and this event records why (the `ready → running` claim re-checks parents, so nothing can run it until the parent completes or the link is removed with `hermes kanban unlink`). | @@ -1419,7 +1419,7 @@ Every transition appends a row to `task_events`. Each row carries an optional `r | `spawned` | `{pid}` | Dispatcher successfully started a worker process. | | `worker_registered` | `{pid, started_at}` | The dispatcher died after starting the worker but before recording its pid, so the worker recorded it itself before its first model call. Liveness checks then see it and an expired claim is extended instead of spawning a second worker. A worker whose run was reclaimed before it got that far exits without working the card. | | `heartbeat` | `{note?}` | Worker called `hermes kanban heartbeat $TASK` to signal liveness during long operations. | -| `reclaimed` | `{stale_lock}` | Claim TTL expired without a completion; task goes back to `ready`. An automatic reclaim counts as one non-successful attempt toward the `gave_up` breaker (a claim that never spawned a worker would otherwise loop claim → reclaim → claim forever); an operator `reclaim` resets the counter instead. | +| `reclaimed` | `{stale_lock, worker_pid, host_local, pid_checkable, heartbeat_stale, retry_status, …}` | Claim TTL expired without a completion; task goes back to its source phase (`retry_status`: `ready`, or `review` for a reviewer run). An automatic reclaim counts as one non-successful attempt toward the `gave_up` breaker (a claim that never spawned a worker would otherwise loop claim → reclaim → claim forever); an operator `reclaim` resets the counter instead. `pid_checkable: false` means the sweep could not verify the worker PID — the claim came from another host, from another PID namespace on this host, or recorded no namespace (pre-upgrade row, older writer), or this reader could not read its own — so it neither probed nor signalled the PID and waited for the TTL instead. | | `crashed` | `{pid, claimer, exit_kind?, exit_code?, worker_output?}` | Worker PID no longer alive but TTL hadn't expired yet. `worker_output` is the tail of the worker's own log (its final response or the rendered provider error, chrome stripped, ≤ 400 chars) and is also appended to the task's `last_failure_error`, so the board shows *why* instead of only the exit code. | | `timed_out` | `{pid, elapsed_seconds, limit_seconds, sigkill}` | `max_runtime_seconds` exceeded; dispatcher SIGTERM'd (then SIGKILL'd after 5 s grace) and re-queued. | | `stale` | `{elapsed_seconds, last_heartbeat_at, heartbeat_age_seconds, timeout_seconds, pid, terminated}` | Task ran longer than `kanban.dispatch_stale_timeout_seconds` (default 4 h) AND no `kanban_heartbeat` arrived in the last hour. Dispatcher SIGTERM'd the host-local worker (if any), reset the task to `ready` for re-dispatch. Does NOT tick the failure counter (stale is dispatcher-side absence detection, not a worker fault). Workers running long operations should call `kanban_heartbeat` at least once an hour to avoid this. | @@ -1433,4 +1433,6 @@ Every transition appends a row to `task_events`. Each row carries an optional `r ## Out of scope -Kanban is deliberately single-host. `~/.hermes/kanban.db` is a local SQLite file and the dispatcher spawns workers on the same machine. Running a shared board across two hosts is not supported — there's no coordination primitive for "worker X on host A, worker Y on host B," and the crash-detection path assumes PIDs are host-local. If you need multi-host, run an independent board per host and use `delegate_task` / a message queue to bridge them. +Kanban is deliberately single-host. `~/.hermes/kanban.db` is a local SQLite file and the dispatcher spawns workers on the same machine. Running a shared board across two hosts is not supported — there's no coordination primitive for "worker X on host A, worker Y on host B," and the crash-detection path only trusts a worker PID recorded on this host **and in this PID namespace** (`claim_pidns`, stamped on every claim). If you need multi-host, run an independent board per host and use `delegate_task` / a message queue to bridge them. + +The namespace rule matters inside one Docker host too. Compose services that join one network namespace (`network_mode: "service:gateway"`) share a hostname but not a PID namespace, so a dispatch run from the dashboard container cannot see the gateway's worker PIDs; it leaves those claims alone instead of closing live workers as crashed, and a claim it cannot verify (including the claims of a recreated gateway container, whose namespace is gone) is released only when its TTL expires (`claim_expires`, 15 min default) with `pid_checkable: false`. Add `pid: "service:gateway"` to the dashboard service to share the PID namespace and get first-tick crash recovery back from either container.