feat(kanban): enforce lifecycle-only worker boundaries

agent/delegation_context.py

@@ -35,6 +35,7 @@
 DELEGATED_CHILD_ENV_MARKER = "HERMES_DELEGATED_CHILD_CONTEXT"
 
 KANBAN_ENV_KEYS: tuple[str, ...] = (
+    "HERMES_KANBAN_WORKER_SCOPE",
     "HERMES_KANBAN_TASK",
     "HERMES_KANBAN_RUN_ID",
     "HERMES_KANBAN_WORKSPACE",

agent/shell_hooks.py

@@ -259,6 +259,10 @@ def register_from_config(
     up on the plugin manager.  Skipped entries (unknown events, malformed,
     not allowlisted, already registered) are logged but not returned.
     """
+    from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker
+
+    if is_lifecycle_only_worker():
+        return []
     if not isinstance(cfg, dict):
         return []
 

cli.py

@@ -1011,6 +1011,10 @@ def _prepare_deferred_agent_startup() -> None:
     if os.environ.get("HERMES_DEFER_AGENT_STARTUP") != "1":
         return
     _deferred_agent_startup_done = True
+    from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker
+
+    if is_lifecycle_only_worker():
+        return
     _accept_hooks = os.environ.get("HERMES_ACCEPT_HOOKS", "").lower() in {
         "1",
         "true",

hermes_cli/env_loader.py

@@ -482,6 +482,34 @@ def load_hermes_dotenv(
     """
     loaded: list[Path] = []
 
+    # The Kanban dispatcher pins task ownership and execution location in the
+    # child process before profile startup. User/profile .env files normally
+    # override stale shell values, but these values are process authority, not
+    # configuration. Snapshot the complete pinned set (including absent keys)
+    # so dotenv, managed scope, secret sources, and config bridges cannot alter
+    # it during startup.
+    from hermes_cli.kanban_worker_scope import (
+        PINNED_WORKER_ENV_KEYS,
+        WORKER_SCOPE_ENV,
+    )
+
+    pinned_worker_env: dict[str, str | None] | None = None
+    dispatcher_worker = bool(os.environ.get("HERMES_KANBAN_TASK"))
+    scoped_worker = dispatcher_worker and bool(os.environ.get(WORKER_SCOPE_ENV))
+    if dispatcher_worker:
+        pinned_worker_env = {
+            key: os.environ.get(key) for key in PINNED_WORKER_ENV_KEYS
+        }
+
+    def restore_pinned_worker_env() -> None:
+        if pinned_worker_env is None:
+            return
+        for key, value in pinned_worker_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
     home_path = Path(hermes_home or os.getenv("HERMES_HOME", Path.home() / ".hermes"))
     user_env = home_path / ".env"
     project_env_path = Path(project_env) if project_env else None
@@ -492,30 +520,33 @@ def load_hermes_dotenv(
     if project_env_path and project_env_path.exists():
         _sanitize_env_file_if_needed(project_env_path)
 
-    if user_env.exists():
-        _load_dotenv_with_fallback(user_env, override=True)
-        loaded.append(user_env)
-        # Mirror reload_env() known-key cleanup so inherited Hermes keys
-        # absent from this profile's .env do not leak into the runtime.
-        _clear_known_keys_missing_from_dotenv(user_env)
-
-    # Load .op.env AFTER .env so that .env values win, but the bootstrap
-    # token (OP_SERVICE_ACCOUNT_TOKEN) becomes available for
-    # apply_onepassword_secrets() even in cron / subprocess environments
-    # that inherit no shell state (no systemd EnvironmentFile, no op run).
-    # .op.env is gitignored — the service-account token never enters the
-    # committed .env file.
-    # Users on systemd can alternatively use:
-    #   EnvironmentFile=-/path/to/.hermes/.op.env
-    # in their gateway unit, which takes precedence (override=False below
-    # ensures .op.env never clobbers a token already in the environment).
-    op_env = home_path / ".op.env"
-    if op_env.exists() and not os.environ.get("OP_SERVICE_ACCOUNT_TOKEN"):
-        _load_dotenv_with_fallback(op_env, override=False)
-
-    if project_env_path and project_env_path.exists():
-        _load_dotenv_with_fallback(project_env_path, override=not loaded)
-        loaded.append(project_env_path)
+    try:
+        if user_env.exists():
+            _load_dotenv_with_fallback(user_env, override=True)
+            loaded.append(user_env)
+            # Mirror reload_env() known-key cleanup so inherited Hermes keys
+            # absent from this profile's .env do not leak into the runtime.
+            _clear_known_keys_missing_from_dotenv(user_env)
+
+        # Load .op.env AFTER .env so that .env values win, but the bootstrap
+        # token (OP_SERVICE_ACCOUNT_TOKEN) becomes available for
+        # apply_onepassword_secrets() even in cron / subprocess environments
+        # that inherit no shell state (no systemd EnvironmentFile, no op run).
+        # .op.env is gitignored — the service-account token never enters the
+        # committed .env file.
+        # Users on systemd can alternatively use:
+        #   EnvironmentFile=-/path/to/.hermes/.op.env
+        # in their gateway unit, which takes precedence (override=False below
+        # ensures .op.env never clobbers a token already in the environment).
+        op_env = home_path / ".op.env"
+        if op_env.exists() and not os.environ.get("OP_SERVICE_ACCOUNT_TOKEN"):
+            _load_dotenv_with_fallback(op_env, override=False)
+
+        if project_env_path and project_env_path.exists():
+            _load_dotenv_with_fallback(project_env_path, override=not loaded)
+            loaded.append(project_env_path)
+    finally:
+        restore_pinned_worker_env()
 
     # A fresh ``hermes update`` retry may have completed a deferred dependency
     # install before importing this module.  Do not remap native secret-source
@@ -524,9 +555,19 @@ def load_hermes_dotenv(
     # only external source resolution is unnecessary for the updater.
     from hermes_cli import _early_recovery
 
-    if not _early_recovery._should_skip_external_secret_sources():
-        _apply_external_secret_sources(home_path)
-    _apply_managed_env()
+    # Dispatcher-pinned workers must reach the first turn without executing
+    # profile-configured command/plugin secret sources. Their profile carries
+    # the provider credentials needed for the run; external source discovery is
+    # an extension surface, not part of lifecycle execution.
+    try:
+        if (
+            not scoped_worker
+            and not _early_recovery._should_skip_external_secret_sources()
+        ):
+            _apply_external_secret_sources(home_path)
+        _apply_managed_env()
+    finally:
+        restore_pinned_worker_env()
 
     # config.yaml is the documented source of truth for terminal.* settings,
     # but the dotenv loads above run with override=True — so a stale
@@ -540,7 +581,10 @@ def load_hermes_dotenv(
     # the documented config path always wins. Runs after _apply_managed_env()
     # so the merged config (which already carries the managed overlay) is
     # what lands in the env.
-    _reapply_terminal_config_bridge(home_path)
+    try:
+        _reapply_terminal_config_bridge(home_path)
+    finally:
+        restore_pinned_worker_env()
 
     return loaded
 

hermes_cli/kanban_db.py

@@ -5358,6 +5358,7 @@ def complete_task(
     metadata: Optional[dict] = None,
     created_cards: Optional[Iterable[str]] = None,
     expected_run_id: Optional[int] = None,
+    allow_artifacts: bool = True,
     fire_lifecycle_hook: bool = True,
 ) -> bool:
     """Transition ``running|ready|blocked|review -> done`` and record ``result``.
@@ -5425,9 +5426,14 @@ def complete_task(
     else:
         verified_cards = []
 
-    metadata = _merge_completion_prose_artifacts(
-        conn, task_id, metadata, summary=summary, result=result,
-    )
+    if allow_artifacts:
+        metadata = _merge_completion_prose_artifacts(
+            conn, task_id, metadata, summary=summary, result=result,
+        )
+    elif isinstance(metadata, dict) and "artifacts" in metadata:
+        raise ArtifactPreservationError(
+            "completion artifacts are disabled for this worker posture"
+        )
     with write_txn(conn):
         # Parent completion is a hard invariant even for direct human review
         # approval. A parent may have been reopened after this task entered
@@ -5476,7 +5482,7 @@ def complete_task(
             )
         if cur.rowcount != 1:
             return False
-        if isinstance(metadata, dict):
+        if allow_artifacts and isinstance(metadata, dict):
             _persist_scratch_completion_artifacts(conn, task_id, metadata)
             for stored_path in metadata.pop("_staged_artifacts", []):
                 path = Path(stored_path)
@@ -10304,12 +10310,10 @@ def _resolve_worker_cli_toolsets(hermes_home: Optional[str]) -> Optional[list[st
             reset_hermes_home_override(token)
         return toolsets or None
     except Exception as exc:
-        _log.debug(
-            "kanban worker: could not resolve CLI toolsets for HERMES_HOME=%r (%s)",
-            hermes_home,
-            exc,
-        )
-        return None
+        raise RuntimeError(
+            "kanban worker: refusing spawn because CLI toolsets could not be "
+            f"resolved for HERMES_HOME={hermes_home!r}"
+        ) from exc
 
 
 _retagged_workspace_roots: set[str] = set()
@@ -10391,6 +10395,8 @@ def _default_spawn(
         # This only happens in test fixtures where the isolated
         # HERMES_HOME never had profiles created.
         pass
+    # A tenantless task must not inherit a stale tenant from the dispatcher.
+    env.pop("HERMES_TENANT", None)
     if task.tenant:
         env["HERMES_TENANT"] = task.tenant
     env["HERMES_KANBAN_TASK"] = task.id
@@ -10462,6 +10468,15 @@ def _default_spawn(
     # attributed correctly regardless of how the child loads config.
     env["HERMES_PROFILE"] = profile_arg
 
+    worker_toolsets = _resolve_worker_cli_toolsets(env.get("HERMES_HOME"))
+    from hermes_cli.kanban_worker_scope import (
+        LIFECYCLE_SCOPE,
+        WORKER_SCOPE_ENV,
+        pin_worker_scope,
+    )
+
+    worker_toolsets = pin_worker_scope(env, worker_toolsets)
+
     # A worker must NEVER boot the interactive TUI: an inherited HERMES_TUI=1
     # or a `display.interface: tui` in the profile's config would send the
     # quiet chat run into the Ink TUI, whose no-TTY bail-out exits 0 without
@@ -10470,16 +10485,14 @@ def _default_spawn(
     # older hermes builds on PATH that predate the flag's precedence.
     env.pop("HERMES_TUI", None)
 
-    cmd = [
-        *_resolve_hermes_argv(),
-        "-p", profile_arg,
-        "--cli",
-        # Worker subprocesses switch to a profile-scoped HERMES_HOME above,
-        # so they see that profile's shell-hook allowlist instead of the
-        # dispatcher's root allowlist. Pass --accept-hooks explicitly so
-        # profile-local worker sessions still register configured hooks.
-        "--accept-hooks",
-    ]
+    cmd = [*_resolve_hermes_argv(), "-p", profile_arg, "--cli"]
+    if env.get(WORKER_SCOPE_ENV) == LIFECYCLE_SCOPE:
+        # A lifecycle-only child must reach its first model turn without
+        # executing profile-supplied hooks, plugins, or startup integrations.
+        env.pop("HERMES_ACCEPT_HOOKS", None)
+    else:
+        # Preserve the existing behavior for normal workers.
+        cmd.append("--accept-hooks")
     # Per-task force-loaded skills. Each name goes in its own
     # `--skills X` pair rather than a single comma-joined arg: the CLI
     # accepts both forms (action='append' + comma-split), but
@@ -10502,7 +10515,6 @@ def _default_spawn(
     # branch, not a nested one.
     if task.reasoning_effort:
         cmd.extend(["--reasoning", task.reasoning_effort])
-    worker_toolsets = _resolve_worker_cli_toolsets(env.get("HERMES_HOME"))
     if worker_toolsets:
         cmd.extend(["--toolsets", ",".join(worker_toolsets)])
     cmd.extend([

hermes_cli/kanban_worker_scope.py

@@ -0,0 +1,116 @@
+"""Fail-closed capability boundary for lifecycle-only Kanban workers.
+
+The assignee profile selects the public ``kanban_lifecycle`` toolset.  The
+dispatcher resolves that profile before spawning the child and pins the
+selection into an internal process scope so startup and dispatch code can
+enforce the same boundary before the first model turn.
+"""
+from __future__ import annotations
+
+import os
+from typing import MutableMapping, Sequence
+
+
+LIFECYCLE_TOOLSET = "kanban_lifecycle"
+WORKER_SCOPE_ENV = "HERMES_KANBAN_WORKER_SCOPE"
+LIFECYCLE_SCOPE = "lifecycle-only"
+LIFECYCLE_TOOL_NAMES = frozenset(
+    {
+        "kanban_show",
+        "kanban_complete",
+        "kanban_block",
+        "kanban_heartbeat",
+    }
+)
+
+# Session-routing values scrubbed by ``kanban_db._default_spawn`` before the
+# child starts. Their absence is dispatcher authority too: a profile dotenv
+# must not reconnect a detached worker to a gateway/cron delivery target.
+# Keep this in sync with ``gateway.session_context._VAR_MAP``; the startup
+# isolation tests enforce parity without importing gateway code here.
+DISPATCHER_SESSION_ENV_KEYS = (
+    "HERMES_SESSION_PLATFORM",
+    "HERMES_SESSION_SOURCE",
+    "HERMES_SESSION_CHAT_ID",
+    "HERMES_SESSION_CHAT_TYPE",
+    "HERMES_SESSION_CHAT_NAME",
+    "HERMES_SESSION_THREAD_ID",
+    "HERMES_SESSION_USER_ID",
+    "HERMES_SESSION_USER_NAME",
+    "HERMES_SESSION_KEY",
+    "HERMES_SESSION_ID",
+    "HERMES_UI_SESSION_ID",
+    "HERMES_SESSION_MESSAGE_ID",
+    "HERMES_SESSION_PROFILE",
+    "HERMES_SESSION_SCOPE_ID",
+    "HERMES_SESSION_USER_ID_ALT",
+    "HERMES_CRON_SESSION",
+    "HERMES_CRON_AUTO_DELIVER_PLATFORM",
+    "HERMES_CRON_AUTO_DELIVER_CHAT_ID",
+    "HERMES_CRON_AUTO_DELIVER_THREAD_ID",
+)
+
+# Values pinned by the dispatcher before the profile process starts. Profile
+# dotenv/config reloads may populate ordinary user configuration, but they must
+# never replace this process authority or execution location. Missing values
+# are authority too: lifecycle workers must not acquire an interactive UI,
+# hook consent, or tenant from a later startup bridge.
+PINNED_WORKER_ENV_KEYS = (
+    WORKER_SCOPE_ENV,
+    "HERMES_KANBAN_TASK",
+    "HERMES_KANBAN_RUN_ID",
+    "HERMES_KANBAN_WORKSPACE",
+    "HERMES_KANBAN_WORKSPACES_ROOT",
+    "HERMES_KANBAN_CLAIM_LOCK",
+    "HERMES_KANBAN_BOARD",
+    "HERMES_KANBAN_DB",
+    "HERMES_KANBAN_BRANCH",
+    "HERMES_KANBAN_GOAL_MODE",
+    "HERMES_KANBAN_GOAL_MAX_TURNS",
+    "HERMES_PROFILE",
+    "HERMES_HOME",
+    "HERMES_TENANT",
+    *DISPATCHER_SESSION_ENV_KEYS,
+    "HERMES_TUI",
+    "HERMES_ACCEPT_HOOKS",
+    "TERMINAL_CWD",
+    "TERMINAL_TIMEOUT",
+    "TERMINAL_MAX_FOREGROUND_TIMEOUT",
+)
+
+
+def pin_worker_scope(
+    env: MutableMapping[str, str],
+    toolsets: Sequence[str] | None,
+) -> list[str] | None:
+    """Pin a resolved lifecycle profile into the child and return its toolsets.
+
+    The value is internal process state, not user-facing configuration.  Clear
+    inherited state on every spawn so one task/profile can never broaden or
+    narrow another by environment leakage.
+    """
+    env.pop(WORKER_SCOPE_ENV, None)
+    if not toolsets:
+        return None
+    resolved = [str(item) for item in toolsets]
+    if LIFECYCLE_TOOLSET in resolved:
+        env[WORKER_SCOPE_ENV] = LIFECYCLE_SCOPE
+        return [LIFECYCLE_TOOLSET]
+    return resolved
+
+
+def current_worker_scope() -> str | None:
+    """Return the canonical process scope, failing closed on unknown values."""
+    raw = str(os.environ.get(WORKER_SCOPE_ENV) or "").strip()
+    if not raw:
+        return None
+    if raw != LIFECYCLE_SCOPE:
+        raise ValueError(f"invalid Kanban worker scope: {raw!r}")
+    return raw
+
+
+def is_lifecycle_only_worker() -> bool:
+    """True only for a dispatcher-owned task process pinned lifecycle-only."""
+    return bool(os.environ.get("HERMES_KANBAN_TASK")) and (
+        current_worker_scope() == LIFECYCLE_SCOPE
+    )

hermes_cli/main.py

@@ -11167,6 +11167,11 @@ def _prepare_agent_startup(args) -> None:
     ):
         return
 
+    from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker
+
+    if is_lifecycle_only_worker():
+        return
+
     _accept_hooks = bool(getattr(args, "accept_hooks", False))
     if not _is_tui_chat_launch(args):
         # The TUI backend process does its own plugin discovery; the launcher