fix(kanban): use one-shot worker launch nonces

agent/delegation_context.py

@@ -41,8 +41,14 @@
 
 DELEGATED_CHILD_ENV_MARKER = "HERMES_DELEGATED_CHILD_CONTEXT"
 
+# One-shot native launch proof emitted by the Kanban spawner. It prevents
+# accidental/passive lifecycle inheritance; it is not authentication against a
+# malicious same-user process that can inspect another process's launch state.
+KANBAN_WORKER_LAUNCH_MARKER = "HERMES_KANBAN_WORKER_LAUNCH"
+
 KANBAN_ENV_KEYS: tuple[str, ...] = (
     "HERMES_KANBAN_TASK",
+    KANBAN_WORKER_LAUNCH_MARKER,
     "HERMES_KANBAN_RUN_ID",
     "HERMES_KANBAN_WORKSPACE",
     "HERMES_KANBAN_WORKSPACES_ROOT",
@@ -57,15 +63,13 @@
 # makes ``kanban_complete`` default to the parent card.
 KANBAN_LIFECYCLE_OWNERSHIP_KEYS: tuple[str, ...] = (
     "HERMES_KANBAN_TASK",
+    KANBAN_WORKER_LAUNCH_MARKER,
     "HERMES_KANBAN_RUN_ID",
     "HERMES_KANBAN_WORKSPACE",
     "HERMES_KANBAN_WORKSPACES_ROOT",
     "HERMES_KANBAN_CLAIM_LOCK",
 )
 
-_BOARD_WORKER_QUERY_PREFIX = "work kanban task "
-
-
 @contextmanager
 def delegated_child_context(session_id: str | None = None) -> Iterator[None]:
     """Mark child execution and isolate its task-local session identity.
@@ -178,51 +182,63 @@ def scrub_kanban_lifecycle_ownership(
 
 def is_explicit_board_worker_launch(
     *,
-    query: str | None,
     source: str | None,
     task_id: str | None,
+    launch_marker: str | None,
+    launch_arg: str | None,
 ) -> bool:
-    """True only for the dispatcher worker argv/env contract.
+    """True only for the native dispatcher's one-shot launch contract.
 
-    ``_default_spawn`` launches ``hermes chat -q "work kanban task <id>"``
-    with ``HERMES_SESSION_SOURCE=kanban``. Any other child chat — including
-    ``hermes chat --source tool`` used by Browser Use benchmarks — is not
-    the board worker, even if it inherited the parent's env.
+    Query text is deliberately irrelevant: a human phrase cannot authenticate
+    a worker. Source and task id must be present, while the fresh env nonce and
+    hidden argv nonce must agree exactly. The public task id is never proof.
     """
-    tid = (task_id or "").strip()
-    if not tid:
+    tid = task_id or ""
+    if not tid or tid != tid.strip():
         return False
-    if (source or "").strip() != "kanban":
+    if (source or "") != "kanban":
         return False
-    return (query or "").strip() == f"{_BOARD_WORKER_QUERY_PREFIX}{tid}"
+    marker = launch_marker or ""
+    arg = launch_arg or ""
+    if not marker or marker != marker.strip() or marker == tid:
+        return False
+    return marker == arg
 
 
 def drop_inherited_kanban_lifecycle_if_not_board_worker(
     *,
-    query: str | None = None,
     source: str | None = None,
+    launch_arg: str | None = None,
     environ: MutableMapping[str, str] | None = None,
 ) -> bool:
     """Drop inherited lifecycle ownership unless this process is the worker.
 
-    Returns True when vars were removed. Mutates *environ* (default
-    ``os.environ``) so a child ``hermes chat`` that slipped past spawn-time
-    scrubbing still cannot ``kanban_complete`` the parent card.
+    Returns True when lifecycle authority was rejected and removed. A valid
+    worker keeps its runtime ownership, consumes only the one-shot launch
+    marker, and returns False. Mutates *environ* (default ``os.environ``) so a
+    child ``hermes chat`` that slipped past spawn-time scrubbing still cannot
+    ``kanban_complete`` the parent card.
     """
     import os
 
     env = os.environ if environ is None else environ
-    task = (env.get("HERMES_KANBAN_TASK") or "").strip()
-    if not task:
-        return False
+    task = env.get("HERMES_KANBAN_TASK") or ""
     resolved_source = source if source is not None else env.get("HERMES_SESSION_SOURCE")
     if is_explicit_board_worker_launch(
-        query=query, source=resolved_source, task_id=task
+        source=resolved_source,
+        task_id=task,
+        launch_marker=env.get(KANBAN_WORKER_LAUNCH_MARKER),
+        launch_arg=launch_arg,
     ):
+        # Consume the one-shot env proof immediately. Runtime ownership keys
+        # remain for the real worker, but no raw-env child can inherit enough
+        # material to mint itself as another worker.
+        env.pop(KANBAN_WORKER_LAUNCH_MARKER, None)
         return False
+    removed = any(key in env for key in KANBAN_LIFECYCLE_OWNERSHIP_KEYS)
     for key in KANBAN_LIFECYCLE_OWNERSHIP_KEYS:
         env.pop(key, None)
-    return True
+    return removed
 
 
 def delegated_child_subprocess_env(

hermes_cli/_parser.py

@@ -314,6 +314,12 @@ def build_top_level_parser():
             "verbatim. Mutually exclusive with -q."
         ),
     )
+    chat_parser.add_argument(
+        "--kanban-worker-launch",
+        metavar="TASK_ID",
+        default=None,
+        help=argparse.SUPPRESS,
+    )
     chat_parser.add_argument(
         "--image", help="Optional local image path to attach to a single query"
     )

hermes_cli/kanban_db.py

@@ -10724,6 +10724,7 @@ def _default_spawn(
     vars all resolve to the same board the dispatcher claimed the task
     from. Workers cannot accidentally see other boards.
     """
+    import secrets
     import subprocess
     if not task.assignee:
         raise ValueError(f"task {task.id} has no assignee")
@@ -10762,6 +10763,13 @@ def _default_spawn(
     if task.tenant:
         env["HERMES_TENANT"] = task.tenant
     env["HERMES_KANBAN_TASK"] = task.id
+    # One-shot native launch proof paired with the hidden CLI argument below.
+    # A fresh unpredictable value prevents accidental/passive inheritance from
+    # being reconstructed from the public task id. This is not authentication
+    # against a malicious same-user process that can inspect launch state.
+    from agent.delegation_context import KANBAN_WORKER_LAUNCH_MARKER
+    worker_launch_proof = secrets.token_urlsafe(32)
+    env[KANBAN_WORKER_LAUNCH_MARKER] = worker_launch_proof
     env["HERMES_KANBAN_WORKSPACE"] = workspace
     # Tag the worker's session so it lands in state.db as `kanban`, not as an
     # untitled `cli` row. A worker is a dispatcher-owned run whose transcript is
@@ -10875,6 +10883,7 @@ def _default_spawn(
         cmd.extend(["--toolsets", ",".join(worker_toolsets)])
     cmd.extend([
         "chat",
+        "--kanban-worker-launch", worker_launch_proof,
         "-q", prompt,
     ])
     if task.goal_mode:

hermes_cli/main.py

@@ -2867,22 +2867,36 @@ def _pin_kanban_board_env() -> None:
 
 
 def _drop_inherited_kanban_lifecycle(args) -> None:
-    """Child chats must not inherit dispatcher run ownership.
+    """Validate and consume the native worker launch proof.
 
     A Kanban worker that shells out to ``hermes chat --source tool`` (Browser
     Use benchmarks, local-model evals) used to keep ``HERMES_KANBAN_TASK``
-    and could ``kanban_complete`` the parent card. The dispatcher worker
-    itself is identified by source ``kanban`` plus ``work kanban task <id>``.
+    and could ``kanban_complete`` the parent card. A real worker needs matching
+    source, a task id, and the same one-shot nonce in env and hidden CLI launch
+    argument; query text and the public task id are never launch proof.
     """
     try:
         from agent.delegation_context import (
             drop_inherited_kanban_lifecycle_if_not_board_worker,
         )
-    except Exception:
+    except ImportError:
+        # Security boundary: validator unavailability must fail closed. Keep
+        # BOARD / DB routing pins so ordinary ``hermes kanban`` shell-outs stay
+        # on the selected board; remove only lifecycle ownership and launch
+        # proof. Other import-time exceptions are deliberately not swallowed.
+        for key in (
+            "HERMES_KANBAN_TASK",
+            "HERMES_KANBAN_RUN_ID",
+            "HERMES_KANBAN_WORKSPACE",
+            "HERMES_KANBAN_WORKSPACES_ROOT",
+            "HERMES_KANBAN_CLAIM_LOCK",
+            "HERMES_KANBAN_WORKER_LAUNCH",
+        ):
+            os.environ.pop(key, None)
         return
     drop_inherited_kanban_lifecycle_if_not_board_worker(
-        query=getattr(args, "query", None),
         source=os.environ.get("HERMES_SESSION_SOURCE"),
+        launch_arg=getattr(args, "kanban_worker_launch", None),
     )
 
 
@@ -2951,6 +2965,13 @@ def _resolve_use_tui(args) -> bool:
 
 def cmd_chat(args):
     """Run interactive chat CLI."""
+    # Validate the one-shot Kanban worker launch proof before any startup work
+    # can spawn a child that inherits process env. An explicit --source wins
+    # over an inherited source exactly as it does for session tagging below.
+    if getattr(args, "source", None):
+        os.environ["HERMES_SESSION_SOURCE"] = args.source
+    _drop_inherited_kanban_lifecycle(args)
+
     use_tui = _resolve_use_tui(args)
 
     _apply_safe_mode(args)
@@ -3200,16 +3221,6 @@ def _skills_sync_bg() -> None:
     if getattr(args, "ignore_rules", False):
         os.environ["HERMES_IGNORE_RULES"] = "1"
 
-    # --source: tag session source for filtering (e.g. 'tool' for third-party integrations)
-    if getattr(args, "source", None):
-        os.environ["HERMES_SESSION_SOURCE"] = args.source
-
-    # Defer the drop when the query is still on disk: a --query-file board
-    # worker would otherwise look like an interactive child and lose ownership
-    # before the prompt is loaded.
-    if getattr(args, "query", None) or not getattr(args, "query_file", None):
-        _drop_inherited_kanban_lifecycle(args)
-
     _pin_kanban_board_env()
     _confirm_startup_expensive_model_override(args)
 
@@ -3260,9 +3271,6 @@ def _skills_sync_bg() -> None:
             print(f"Error: --query-file {_qfile} is empty", file=sys.stderr)
             sys.exit(2)
 
-    # Query may have arrived via --query-file after the first ownership check.
-    _drop_inherited_kanban_lifecycle(args)
-
     # Build kwargs from args
     kwargs = {
         "model": args.model,

tests/conftest.py

@@ -310,6 +310,7 @@ def _looks_like_credential(name: str) -> bool:
     "HERMES_KANBAN_WORKSPACES_ROOT",
     "HERMES_KANBAN_LOGS_ROOT",
     "HERMES_KANBAN_TASK",
+    "HERMES_KANBAN_WORKER_LAUNCH",
     "HERMES_KANBAN_WORKSPACE",
     "HERMES_KANBAN_RUN_ID",
     "HERMES_KANBAN_CLAIM_LOCK",
@@ -1393,7 +1394,7 @@ def _guarded_killpg(pgid, sig, *args, **kwargs):
                 return real_killpg(pgid, sig, *args, **kwargs)
             raise RuntimeError(
                 f"tests/conftest.py live-system guard: blocked "
-                f"os.killpg({pgid}, {sig}) — PGID is outside the test "
+                f"os.killpg({pgid}, {sig}) — PGID is outside the test "  # windows-footgun: ok — diagnostic text
                 "process group. See _live_system_guard for the why."
             )