diff --git a/agent/delegation_context.py b/agent/delegation_context.py index 9b8bce97597a..992bbb3c5e10 100644 --- a/agent/delegation_context.py +++ b/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", diff --git a/agent/shell_hooks.py b/agent/shell_hooks.py index 965bbcd3d019..fe0e33d8c6a8 100644 --- a/agent/shell_hooks.py +++ b/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 [] diff --git a/cli.py b/cli.py index 1d09a4742a79..d2b17921a9ae 100644 --- a/cli.py +++ b/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", diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 05d6de052eca..b2739f5e5e1d 100644 --- a/hermes_cli/env_loader.py +++ b/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 diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index ce29dbe7e200..e920afdcb0f0 100644 --- a/hermes_cli/kanban_db.py +++ b/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([ diff --git a/hermes_cli/kanban_worker_scope.py b/hermes_cli/kanban_worker_scope.py new file mode 100644 index 000000000000..2ce29b888117 --- /dev/null +++ b/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 + ) diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 93758d0c0089..8918bcdf15ac 100644 --- a/hermes_cli/main.py +++ b/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 diff --git a/hermes_cli/mcp_startup.py b/hermes_cli/mcp_startup.py index c3688054055a..62bcd326b94d 100644 --- a/hermes_cli/mcp_startup.py +++ b/hermes_cli/mcp_startup.py @@ -39,6 +39,11 @@ def start_background_mcp_discovery(*, logger, thread_name: str) -> None: """ global _mcp_discovery_started, _mcp_discovery_thread + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return + with _mcp_discovery_lock: if _mcp_discovery_started: thread = _mcp_discovery_thread diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index ac1246e8109e..d4c96edc2e91 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -3773,6 +3773,11 @@ def discover_and_load(self, force: bool = False) -> None: sessions without requiring a full agent restart. """ with self._discovery_lock, _plugin_home_scope(self.home_path): + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + self._discovered = True + return if self._discovered and not force: return if force: diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 9aaef475cdd8..685dd95a2c55 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -2456,6 +2456,17 @@ def _get_platform_tools( skip = configurable_keys | plugin_ts_keys | platform_default_keys skip |= {k for k in TOOLSETS if k.startswith("hermes-")} skip |= set(_DEFAULT_OFF_TOOLSETS) - {platform} + # This security boundary is opt-in by profile and must never be inferred + # merely because its four tools are a subset of the broader Kanban surface. + # Otherwise every normal worker profile that includes Kanban gets silently + # collapsed to lifecycle-only by the dispatcher. + skip.add("kanban_lifecycle") + # A deliberately minimal worker profile satisfies lifecycle requirements + # with ``kanban_lifecycle``. Recovering the broader ``kanban`` toolset here + # would silently re-enable child creation, attachments, and URL fetching. + if "kanban_lifecycle" in toolset_names: + enabled_toolsets.add("kanban_lifecycle") + skip.add("kanban") for ts_key, ts_def in TOOLSETS.items(): if ts_key in skip: continue diff --git a/model_tools.py b/model_tools.py index 241c862faec1..1c50226357c2 100644 --- a/model_tools.py +++ b/model_tools.py @@ -81,6 +81,27 @@ def _is_dispatcher_owned_worker() -> bool: return True +def _lifecycle_worker_dispatch_error(function_name: str) -> Optional[str]: + """Return a fail-closed error for forged out-of-scope tool dispatch.""" + if not os.environ.get("HERMES_KANBAN_TASK"): + return None + raw_scope = str(os.environ.get("HERMES_KANBAN_WORKER_SCOPE") or "").strip() + if not raw_scope: + return None + try: + from hermes_cli.kanban_worker_scope import ( + LIFECYCLE_TOOL_NAMES, + is_lifecycle_only_worker, + ) + + scoped = is_lifecycle_only_worker() + except Exception as exc: + return f"Kanban worker tool dispatch denied: invalid worker scope ({exc})" + if scoped and function_name not in LIFECYCLE_TOOL_NAMES: + return f"Tool '{function_name}' is unavailable in lifecycle-only Kanban workers" + return None + + # ============================================================================= # Async Bridging (single source of truth -- used by registry.dispatch too) # ============================================================================= @@ -244,8 +265,12 @@ def _run_in_worker(): # Plugin tool discovery (user/project/pip plugins) try: - from hermes_cli.plugins import discover_plugins - discover_plugins() + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if not is_lifecycle_only_worker(): + from hermes_cli.plugins import discover_plugins + + discover_plugins() except Exception as e: logger.debug("Plugin discovery failed: %s", e) @@ -370,6 +395,7 @@ def get_tool_definitions( registry._generation, cfg_fp, bool(os.environ.get("HERMES_KANBAN_TASK")), + os.environ.get("HERMES_KANBAN_WORKER_SCOPE", ""), bool(skip_tool_search_assembly), _is_delegated_child_context(), _is_dispatcher_owned_worker(), @@ -430,13 +456,12 @@ def _compute_tool_definitions( os.environ.get("HERMES_KANBAN_TASK") and not _is_delegated_child_context() and _is_dispatcher_owned_worker() - and "kanban" not in effective_enabled_toolsets + and not {"kanban", "kanban_lifecycle"}.intersection(effective_enabled_toolsets) ): - # Dispatcher-spawned workers are scoped by HERMES_KANBAN_TASK and - # must always receive the lifecycle handoff tools. Assignee - # profiles may intentionally restrict their normal chat toolsets - # (for token/cost reasons), but that should not strip the kanban - # worker's completion/block/heartbeat surface. + # Dispatcher-spawned workers must always receive a lifecycle + # handoff surface. A profile may deliberately select the minimal + # ``kanban_lifecycle`` subset; never widen that boundary back to the + # full board/attachment API. effective_enabled_toolsets.append("kanban") for toolset_name in effective_enabled_toolsets: if validate_toolset(toolset_name): @@ -506,9 +531,46 @@ def _compute_tool_definitions( # needed; plugins respect enabled_toolsets / disabled_toolsets like any # other toolset. + from hermes_cli.kanban_worker_scope import ( + LIFECYCLE_TOOL_NAMES, + is_lifecycle_only_worker, + ) + + lifecycle_only = is_lifecycle_only_worker() + if lifecycle_only: + # The process scope is authoritative. A stale CLI/profile argument may + # never broaden a lifecycle worker after the dispatcher pins it. + tools_to_include = set(LIFECYCLE_TOOL_NAMES) + # Ask the registry for schemas (only returns tools whose check_fn passes) filtered_tools = registry.get_definitions(tools_to_include, quiet=quiet_mode) + if lifecycle_only: + import copy + + narrowed = [] + for tool_def in filtered_tools: + cloned = copy.deepcopy(tool_def) + function = cloned.get("function", {}) + name = function.get("name") + parameters = function.get("parameters", {}) + properties = parameters.get("properties") + if isinstance(properties, dict): + properties.pop("task_id", None) + properties.pop("board", None) + if name == "kanban_complete": + properties.pop("created_cards", None) + properties.pop("artifacts", None) + required = parameters.get("required") + if isinstance(required, list): + parameters["required"] = [ + item + for item in required + if item not in {"task_id", "board", "created_cards", "artifacts"} + ] + narrowed.append(cloned) + filtered_tools = narrowed + # The set of tool names that actually passed check_fn filtering. # Use this (not tools_to_include) for any downstream schema that references # other tools by name — otherwise the model sees tools mentioned in @@ -1230,6 +1292,10 @@ def handle_function_call( Returns: Function result as a JSON string. """ + dispatch_error = _lifecycle_worker_dispatch_error(function_name) + if dispatch_error is not None: + return json.dumps({"error": dispatch_error}, ensure_ascii=False) + # Coerce string arguments to their schema-declared types (e.g. "42"→42) function_args = coerce_tool_args(function_name, function_args) if not isinstance(function_args, dict): diff --git a/plugins/context_engine/__init__.py b/plugins/context_engine/__init__.py index 906ade4a34cb..2d429855343d 100644 --- a/plugins/context_engine/__init__.py +++ b/plugins/context_engine/__init__.py @@ -30,6 +30,19 @@ _CONTEXT_ENGINE_PLUGINS_DIR = Path(__file__).parent +def _context_engine_extensions_disabled_for_scoped_worker() -> bool: + """Keep configurable engine code outside dispatcher-scoped workers.""" + try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + return is_lifecycle_only_worker() + except (ImportError, ValueError): + # An unavailable scope guard and unknown dispatcher scopes must not + # fall through to an extension loader that executes Python before the + # first agent turn. + return True + + def discover_context_engines() -> List[Tuple[str, str, bool]]: """Scan plugins/context_engine/ for available engines. @@ -37,6 +50,9 @@ def discover_context_engines() -> List[Tuple[str, str, bool]]: Does NOT import the engines — just reads plugin.yaml for metadata and does a lightweight availability check. """ + if _context_engine_extensions_disabled_for_scoped_worker(): + return [] + results = [] if not _CONTEXT_ENGINE_PLUGINS_DIR.is_dir(): return results @@ -81,6 +97,9 @@ def load_context_engine(name: str) -> Optional["ContextEngine"]: Returns None if the engine is not found or fails to load. """ + if _context_engine_extensions_disabled_for_scoped_worker(): + return None + engine_dir = _CONTEXT_ENGINE_PLUGINS_DIR / name if not engine_dir.is_dir(): logger.debug("Context engine '%s' not found in %s", name, _CONTEXT_ENGINE_PLUGINS_DIR) @@ -104,6 +123,9 @@ def _load_engine_from_dir(engine_dir: Path) -> Optional["ContextEngine"]: - A register(ctx) function (plugin-style) — we simulate a ctx - A top-level class that extends ContextEngine — we instantiate it """ + if _context_engine_extensions_disabled_for_scoped_worker(): + return None + name = engine_dir.name module_name = f"plugins.context_engine.{name}" init_file = engine_dir / "__init__.py" diff --git a/plugins/memory/__init__.py b/plugins/memory/__init__.py index a24029a0de31..00cbfdef0c40 100644 --- a/plugins/memory/__init__.py +++ b/plugins/memory/__init__.py @@ -372,6 +372,16 @@ def _load_provider_from_entry_point( register_skills: bool = True, ) -> Optional["MemoryProvider"]: """Import a provider entry point and extract the MemoryProvider instance.""" + # ``entry_point.load()`` executes third-party package code. Keep the same + # fail-closed boundary as directory providers, including unknown scopes. + try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return None + except (ImportError, ValueError): + return None + from agent.memory_provider import MemoryProvider loaded = entry_point.load() @@ -427,6 +437,18 @@ def _load_provider_from_dir( - A register(ctx) function (plugin-style) — we simulate a ctx - A top-level class that extends MemoryProvider — we instantiate it """ + # This is the single import/exec chokepoint for bundled and user-installed + # memory providers. Lifecycle-only workers must not execute extension code + # before (or lazily after) their first model turn. Unknown pinned scopes also + # fail closed here. + try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return None + except (ImportError, ValueError): + return None + name = provider_dir.name # Use a separate namespace for user-installed plugins so they don't # collide with bundled providers in sys.modules. @@ -697,6 +719,14 @@ def discover_plugin_cli_commands() -> List[dict]: any provider is loaded. """ results: List[dict] = [] + try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return results + except (ImportError, ValueError): + return results + if not _MEMORY_PLUGINS_DIR.is_dir(): return results diff --git a/providers/__init__.py b/providers/__init__.py index 011e84afa723..243c85dd03d7 100644 --- a/providers/__init__.py +++ b/providers/__init__.py @@ -179,6 +179,17 @@ def _discover_entry_point_providers() -> None: ``register_provider()`` — a pip package cannot hijack a first-party provider name. """ + # Pip entry points execute third-party code during ``load()``. A + # dispatcher-scoped lifecycle worker must reach its first model turn + # without touching that extension surface; unknown scopes fail closed. + try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return + except (ImportError, ValueError): + return + try: import importlib.metadata as _md except Exception: # pragma: no cover — importlib.metadata always present ≥3.8 @@ -308,9 +319,18 @@ def _discover_providers() -> None: _import_plugin_dir(child, "bundled") # 2. User plugins — under $HERMES_HOME/plugins/model-providers//. - # These can override any bundled profile of the same name (last-writer-wins - # in register_provider()). - user_dir = _user_plugins_dir() + # These can override any bundled profile of the same name for normal + # agents. Dispatcher-scoped lifecycle workers must not execute profile + # extension code; unknown pinned scopes fail closed as well. + allow_user_plugins = True + try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + allow_user_plugins = not is_lifecycle_only_worker() + except (ImportError, ValueError): + allow_user_plugins = False + + user_dir = _user_plugins_dir() if allow_user_plugins else None if user_dir is not None: for child in sorted(user_dir.iterdir()): if not child.is_dir() or child.name.startswith(("_", ".")): @@ -318,25 +338,26 @@ def _discover_providers() -> None: _import_plugin_dir(child, "user") # 3. Legacy single-file profiles at providers/.py. Kept for - # back-compat — if someone drops a ``providers/foo.py`` into an - # editable install, it still works without the plugin layout. - try: - import pkgutil - - import providers as _pkg - - for _importer, modname, _ispkg in pkgutil.iter_modules(_pkg.__path__): - if modname.startswith("_") or modname == "base": - continue - try: - importlib.import_module(f"providers.{modname}") - except ImportError as exc: - logger.warning( - "Failed to import legacy provider module %s: %s", modname, exc - ) - except Exception: - pass - - # (Pip entry-point providers are discovered in step 0, before the - # filesystem plugins, so first-party profiles always win on name - # collision — see _discover_entry_point_providers.) + # back-compat for normal agents. In an editable install these files are + # another user-extension surface, so scoped workers skip them together + # with $HERMES_HOME provider plugins. + if allow_user_plugins: + try: + import pkgutil + + import providers as _pkg + + for _importer, modname, _ispkg in pkgutil.iter_modules(_pkg.__path__): + if modname.startswith("_") or modname == "base": + continue + try: + importlib.import_module(f"providers.{modname}") + except ImportError as exc: + logger.warning( + "Failed to import legacy provider module %s: %s", modname, exc + ) + except Exception: + pass + + # Pip entry-point providers are discovered in step 0, before filesystem + # plugins, so first-party profiles always win on name collision. diff --git a/tests/hermes_cli/test_kanban_lifecycle_startup.py b/tests/hermes_cli/test_kanban_lifecycle_startup.py new file mode 100644 index 000000000000..fe9700173857 --- /dev/null +++ b/tests/hermes_cli/test_kanban_lifecycle_startup.py @@ -0,0 +1,847 @@ +"""Startup isolation contracts for lifecycle-only Kanban workers.""" + +from __future__ import annotations + +from argparse import Namespace +import os +from pathlib import Path +import subprocess +import sys +import types + +import pytest + + +def _lifecycle_env( + monkeypatch: pytest.MonkeyPatch, + scope: str = "lifecycle-only", +) -> None: + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_lifecycle_startup") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", scope) + + +def _clear_lifecycle_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False) + monkeypatch.delenv("HERMES_KANBAN_WORKER_SCOPE", raising=False) + monkeypatch.delenv("HERMES_TUI", raising=False) + + +def _agent_args(command: str = "chat") -> Namespace: + return Namespace( + accept_hooks=True, + command=command, + cron_command=None, + gateway_command=None, + mcp_action="serve" if command == "mcp" else None, + safe_mode=False, + tui=False, + yolo=False, + ) + + +def test_profile_dotenv_cannot_override_dispatcher_identity( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from gateway.session_context import _VAR_MAP + from hermes_cli import env_loader + + home = tmp_path / ".hermes" + home.mkdir() + foreign_values = [ + "HERMES_KANBAN_WORKER_SCOPE=", + "HERMES_KANBAN_TASK=foreign-task", + "HERMES_KANBAN_RUN_ID=foreign-run", + "HERMES_KANBAN_WORKSPACE=/foreign/workspace", + "HERMES_KANBAN_BOARD=foreign-board", + "HERMES_HOME=/foreign/profile", + "HERMES_TENANT=foreign-tenant", + "HERMES_TUI=1", + "HERMES_ACCEPT_HOOKS=1", + ] + foreign_values.extend(f"{key}=foreign-route" for key in sorted(_VAR_MAP)) + (home / ".env").write_text( + "\n".join(foreign_values) + "\n", + encoding="utf-8", + ) + dispatcher_identity = { + "HERMES_KANBAN_WORKER_SCOPE": "lifecycle-only", + "HERMES_KANBAN_TASK": "owned-task", + "HERMES_KANBAN_RUN_ID": "owned-run", + "HERMES_KANBAN_WORKSPACE": "/owned/workspace", + "HERMES_KANBAN_BOARD": "owned-board", + "HERMES_HOME": "/owned/profile", + "HERMES_TENANT": "owned-tenant", + "HERMES_SESSION_SOURCE": "kanban", + } + for key, value in dispatcher_identity.items(): + monkeypatch.setenv(key, value) + dispatcher_absences = ( + set(_VAR_MAP) - {"HERMES_SESSION_SOURCE"} + ) | {"HERMES_TUI", "HERMES_ACCEPT_HOOKS"} + for key in dispatcher_absences: + monkeypatch.delenv(key, raising=False) + monkeypatch.setattr(env_loader, "_apply_external_secret_sources", lambda _home: None) + monkeypatch.setattr(env_loader, "_apply_managed_env", lambda: None) + + loaded = env_loader.load_hermes_dotenv(hermes_home=home) + + assert loaded == [home / ".env"] + assert { + key: os.environ.get(key) for key in dispatcher_identity + } == dispatcher_identity + assert dispatcher_absences.isdisjoint(os.environ) + + +def test_pinned_worker_authority_covers_gateway_routing() -> None: + from gateway.session_context import _VAR_MAP + from hermes_cli.kanban_worker_scope import PINNED_WORKER_ENV_KEYS + + assert set(_VAR_MAP) <= set(PINNED_WORKER_ENV_KEYS) + + +@pytest.mark.parametrize("failing_stage", ["dotenv", "managed", "config"]) +def test_dispatcher_authority_is_restored_when_startup_stage_raises( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + failing_stage: str, +) -> None: + from hermes_cli import env_loader + from hermes_cli.kanban_worker_scope import PINNED_WORKER_ENV_KEYS + + home = tmp_path / ".hermes" + home.mkdir() + (home / ".env").write_text("ORDINARY_SETTING=loaded\n", encoding="utf-8") + + for index, key in enumerate(PINNED_WORKER_ENV_KEYS): + if key in {"HERMES_TUI", "HERMES_ACCEPT_HOOKS"}: + monkeypatch.delenv(key, raising=False) + elif key == "HERMES_KANBAN_WORKER_SCOPE": + monkeypatch.setenv(key, "lifecycle-only") + elif key == "HERMES_KANBAN_TASK": + monkeypatch.setenv(key, "owned-task") + else: + monkeypatch.setenv(key, f"owned-{index}") + expected = {key: os.environ.get(key) for key in PINNED_WORKER_ENV_KEYS} + + def mutate_and_raise(*_args, **_kwargs) -> None: + for key in PINNED_WORKER_ENV_KEYS: + os.environ[key] = "foreign" + raise RuntimeError(f"{failing_stage} bridge failed") + + monkeypatch.setattr(env_loader, "_apply_managed_env", lambda: None) + monkeypatch.setattr(env_loader, "_reapply_terminal_config_bridge", lambda _home: None) + if failing_stage == "dotenv": + monkeypatch.setattr(env_loader, "_load_dotenv_with_fallback", mutate_and_raise) + elif failing_stage == "managed": + monkeypatch.setattr(env_loader, "_apply_managed_env", mutate_and_raise) + else: + monkeypatch.setattr(env_loader, "_reapply_terminal_config_bridge", mutate_and_raise) + + with pytest.raises(RuntimeError, match=f"{failing_stage} bridge failed"): + env_loader.load_hermes_dotenv(hermes_home=home) + + assert { + key: os.environ.get(key) for key in PINNED_WORKER_ENV_KEYS + } == expected + + +def test_dispatcher_identity_is_restored_before_later_startup_bridges( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from hermes_cli import env_loader + + home = tmp_path / ".hermes" + home.mkdir() + (home / ".env").write_text( + "HERMES_KANBAN_WORKER_SCOPE=\nHERMES_KANBAN_TASK=foreign-task\n", + encoding="utf-8", + ) + _lifecycle_env(monkeypatch) + monkeypatch.setenv("HERMES_HOME", "/owned/profile") + monkeypatch.setenv("HERMES_TENANT", "owned-tenant") + monkeypatch.delenv("HERMES_TUI", raising=False) + monkeypatch.delenv("HERMES_ACCEPT_HOOKS", raising=False) + observations: list[tuple[str | None, ...]] = [] + + def observe() -> tuple[str | None, ...]: + return tuple( + os.environ.get(key) + for key in ( + "HERMES_KANBAN_WORKER_SCOPE", + "HERMES_KANBAN_TASK", + "HERMES_HOME", + "HERMES_TENANT", + "HERMES_TUI", + "HERMES_ACCEPT_HOOKS", + ) + ) + + def managed_bridge() -> None: + observations.append(observe()) + os.environ["HERMES_KANBAN_TASK"] = "managed-task" + os.environ["HERMES_HOME"] = "/managed/profile" + os.environ["HERMES_TENANT"] = "managed-tenant" + os.environ["HERMES_TUI"] = "1" + os.environ["HERMES_ACCEPT_HOOKS"] = "1" + + def terminal_bridge(_home: Path) -> None: + observations.append(observe()) + os.environ["HERMES_KANBAN_TASK"] = "terminal-task" + os.environ["HERMES_HOME"] = "/terminal/profile" + os.environ["HERMES_TENANT"] = "terminal-tenant" + os.environ["HERMES_TUI"] = "1" + os.environ["HERMES_ACCEPT_HOOKS"] = "1" + + monkeypatch.setattr(env_loader, "_apply_managed_env", managed_bridge) + monkeypatch.setattr(env_loader, "_reapply_terminal_config_bridge", terminal_bridge) + monkeypatch.setattr( + env_loader, + "_apply_external_secret_sources", + lambda _home: pytest.fail("scoped worker must skip external secret sources"), + ) + + env_loader.load_hermes_dotenv(hermes_home=home) + + assert observations == [ + ( + "lifecycle-only", + "t_lifecycle_startup", + "/owned/profile", + "owned-tenant", + None, + None, + ), + ( + "lifecycle-only", + "t_lifecycle_startup", + "/owned/profile", + "owned-tenant", + None, + None, + ), + ] + assert observe() == ( + "lifecycle-only", + "t_lifecycle_startup", + "/owned/profile", + "owned-tenant", + None, + None, + ) + + +@pytest.mark.parametrize("scope", ["lifecycle-only", "future-worker-scope"]) +def test_scoped_worker_skips_external_secret_sources( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + scope: str, +) -> None: + from hermes_cli import env_loader + + home = tmp_path / ".hermes" + home.mkdir() + _lifecycle_env(monkeypatch, scope) + calls: list[Path] = [] + monkeypatch.setattr( + env_loader, + "_apply_external_secret_sources", + lambda resolved_home: calls.append(resolved_home), + ) + monkeypatch.setattr(env_loader, "_apply_managed_env", lambda: None) + + env_loader.load_hermes_dotenv(hermes_home=home) + + assert calls == [] + + +def test_normal_worker_runs_external_secret_sources( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from hermes_cli import env_loader + + home = tmp_path / ".hermes" + home.mkdir() + _clear_lifecycle_env(monkeypatch) + calls: list[Path] = [] + monkeypatch.setattr( + env_loader, + "_apply_external_secret_sources", + lambda resolved_home: calls.append(resolved_home), + ) + monkeypatch.setattr(env_loader, "_apply_managed_env", lambda: None) + + env_loader.load_hermes_dotenv(hermes_home=home) + + assert calls == [home] + + +def test_normal_dispatcher_worker_keeps_authority_and_runs_secret_sources( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from hermes_cli import env_loader + + home = tmp_path / ".hermes" + home.mkdir() + (home / ".env").write_text( + "HERMES_KANBAN_TASK=foreign-task\n" + "HERMES_KANBAN_BOARD=foreign-board\n" + "HERMES_SESSION_SOURCE=telegram\n" + "HERMES_SESSION_CHAT_ID=foreign-chat\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_KANBAN_TASK", "owned-task") + monkeypatch.setenv("HERMES_KANBAN_BOARD", "owned-board") + monkeypatch.setenv("HERMES_SESSION_SOURCE", "kanban") + monkeypatch.delenv("HERMES_KANBAN_WORKER_SCOPE", raising=False) + monkeypatch.delenv("HERMES_SESSION_CHAT_ID", raising=False) + calls: list[Path] = [] + monkeypatch.setattr( + env_loader, + "_apply_external_secret_sources", + lambda resolved_home: calls.append(resolved_home), + ) + monkeypatch.setattr(env_loader, "_apply_managed_env", lambda: None) + + env_loader.load_hermes_dotenv(hermes_home=home) + + assert os.environ["HERMES_KANBAN_TASK"] == "owned-task" + assert os.environ["HERMES_KANBAN_BOARD"] == "owned-board" + assert os.environ["HERMES_SESSION_SOURCE"] == "kanban" + assert "HERMES_SESSION_CHAT_ID" not in os.environ + assert calls == [home] + + +@pytest.mark.parametrize("scope", ["lifecycle-only", "future-worker-scope"]) +def test_scoped_worker_does_not_execute_user_memory_provider( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + scope: str, +) -> None: + _lifecycle_env(monkeypatch, scope) + marker = tmp_path / "memory-provider-executed" + plugins_dir = tmp_path / "plugins" + provider_dir = plugins_dir / "adversarial_probe" + provider_dir.mkdir(parents=True) + (provider_dir / "__init__.py").write_text( + "\n".join( + [ + "from pathlib import Path", + f"Path({str(marker)!r}).write_text('executed', encoding='utf-8')", + "class MemoryProvider:", + " pass", + ] + ) + + "\n", + encoding="utf-8", + ) + + from plugins import memory as memory_plugins + + monkeypatch.setattr(memory_plugins, "_get_user_plugins_dir", lambda: plugins_dir) + module_name = "_hermes_user_memory.adversarial_probe" + sys.modules.pop(module_name, None) + try: + assert memory_plugins.load_memory_provider("adversarial_probe") is None + finally: + sys.modules.pop(module_name, None) + + assert not marker.exists() + + +@pytest.mark.parametrize("scope", ["lifecycle-only", "future-worker-scope"]) +def test_scoped_worker_does_not_execute_memory_provider_cli( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + scope: str, +) -> None: + _lifecycle_env(monkeypatch, scope) + marker = tmp_path / "memory-provider-cli-executed" + plugins_dir = tmp_path / "plugins" + provider_dir = plugins_dir / "adversarial_cli" + provider_dir.mkdir(parents=True) + (provider_dir / "__init__.py").write_text( + "# MemoryProvider plugin probe\n", + encoding="utf-8", + ) + (provider_dir / "cli.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed', encoding='utf-8')\n" + "def register_cli(_parser):\n" + " return None\n", + encoding="utf-8", + ) + + from plugins import memory as memory_plugins + + monkeypatch.setattr(memory_plugins, "_get_user_plugins_dir", lambda: plugins_dir) + monkeypatch.setattr( + memory_plugins, + "_get_active_memory_provider", + lambda: "adversarial_cli", + ) + module_name = "_hermes_user_memory.adversarial_cli.cli" + sys.modules.pop(module_name, None) + try: + assert memory_plugins.discover_plugin_cli_commands() == [] + finally: + sys.modules.pop(module_name, None) + sys.modules.pop("_hermes_user_memory.adversarial_cli", None) + + assert not marker.exists() + + +@pytest.mark.parametrize("scope", ["lifecycle-only", "future-worker-scope"]) +def test_scoped_worker_does_not_execute_user_model_provider( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + scope: str, +) -> None: + _lifecycle_env(monkeypatch, scope) + marker = tmp_path / "model-provider-executed" + plugins_dir = tmp_path / "model-providers" + provider_dir = plugins_dir / "adversarial_probe" + provider_dir.mkdir(parents=True) + (provider_dir / "__init__.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed', encoding='utf-8')\n", + encoding="utf-8", + ) + + import pkgutil + import providers + + monkeypatch.setattr(providers, "_discovered", False) + monkeypatch.setattr(providers, "_REGISTRY", {}) + monkeypatch.setattr(providers, "_ALIASES", {}) + monkeypatch.setattr(providers, "_PROVIDER_LIST_CACHE", None) + monkeypatch.setattr(providers, "_BUNDLED_PLUGINS_DIR", tmp_path / "bundled") + monkeypatch.setattr(providers, "_user_plugins_dir", lambda: plugins_dir) + monkeypatch.setattr(pkgutil, "iter_modules", lambda *_args, **_kwargs: []) + module_name = "_hermes_user_provider_adversarial_probe" + sys.modules.pop(module_name, None) + try: + assert providers.list_providers() == [] + finally: + sys.modules.pop(module_name, None) + + assert not marker.exists() + + +@pytest.mark.parametrize("scope", ["lifecycle-only", "future-worker-scope"]) +def test_scoped_worker_does_not_import_legacy_provider_extensions( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + scope: str, +) -> None: + _lifecycle_env(monkeypatch, scope) + + import pkgutil + import providers + + monkeypatch.setattr(providers, "_discovered", False) + monkeypatch.setattr(providers, "_REGISTRY", {}) + monkeypatch.setattr(providers, "_ALIASES", {}) + monkeypatch.setattr(providers, "_PROVIDER_LIST_CACHE", None) + monkeypatch.setattr(providers, "_BUNDLED_PLUGINS_DIR", tmp_path / "bundled") + monkeypatch.setattr(providers, "_user_plugins_dir", lambda: None) + monkeypatch.setattr( + pkgutil, + "iter_modules", + lambda *_args, **_kwargs: [(None, "adversarial_legacy", False)], + ) + imported: list[str] = [] + monkeypatch.setattr( + providers.importlib, + "import_module", + lambda name: imported.append(name), + ) + + assert providers.list_providers() == [] + assert imported == [] + + +@pytest.mark.parametrize( + ("scope", "should_execute"), + [ + ("lifecycle-only", False), + ("future-worker-scope", False), + (None, True), + ], +) +def test_legacy_provider_execution_boundary_in_fresh_process( + tmp_path: Path, + scope: str | None, + should_execute: bool, +) -> None: + marker = tmp_path / "legacy-provider-executed" + legacy_dir = tmp_path / "providers" + legacy_dir.mkdir() + (legacy_dir / "adversarial_legacy.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed', encoding='utf-8')\n", + encoding="utf-8", + ) + code = """ +import os +from pathlib import Path +import providers +providers.__path__ = [os.environ["LEGACY_PROVIDER_DIR"]] +providers._BUNDLED_PLUGINS_DIR = Path(os.environ["EMPTY_BUNDLED_DIR"]) +providers._REGISTRY.clear() +providers._ALIASES.clear() +providers._PROVIDER_LIST_CACHE = None +providers._discovered = False +providers.list_providers() +print("MARKER=" + str(Path(os.environ["LEGACY_MARKER"]).exists())) +""" + env = os.environ.copy() + env.update( + { + "LEGACY_PROVIDER_DIR": str(legacy_dir), + "EMPTY_BUNDLED_DIR": str(tmp_path / "bundled"), + "LEGACY_MARKER": str(marker), + } + ) + env.pop("HERMES_KANBAN_TASK", None) + env.pop("HERMES_KANBAN_WORKER_SCOPE", None) + if scope is not None: + env["HERMES_KANBAN_TASK"] = "t_legacy_probe" + env["HERMES_KANBAN_WORKER_SCOPE"] = scope + + result = subprocess.run( + [sys.executable, "-c", code], + cwd=Path(__file__).resolve().parents[2], + env=env, + capture_output=True, + text=True, + timeout=30, + check=True, + ) + + assert f"MARKER={should_execute}" in result.stdout + + +@pytest.mark.parametrize("scope", ["lifecycle-only", "future-worker-scope"]) +@pytest.mark.parametrize("operation", ["load", "discover"]) +def test_scoped_worker_does_not_execute_context_engine_extensions( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + scope: str, + operation: str, +) -> None: + _lifecycle_env(monkeypatch, scope) + marker = tmp_path / "context-engine-executed" + plugins_dir = tmp_path / "context_engine" + engine_dir = plugins_dir / "adversarial_context" + engine_dir.mkdir(parents=True) + (engine_dir / "__init__.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed', encoding='utf-8')\n", + encoding="utf-8", + ) + + import plugins.context_engine as context_engine + + monkeypatch.setattr(context_engine, "_CONTEXT_ENGINE_PLUGINS_DIR", plugins_dir) + sys.modules.pop("plugins.context_engine.adversarial_context", None) + if operation == "load": + assert context_engine.load_context_engine("adversarial_context") is None + else: + assert context_engine.discover_context_engines() == [] + + assert not marker.exists() + + +def test_context_engine_guard_fails_closed_when_scope_import_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import builtins + import plugins.context_engine as context_engine + + real_import = builtins.__import__ + + def fail_scope_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "hermes_cli.kanban_worker_scope": + raise ImportError("scope guard unavailable") + return real_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fail_scope_import) + + assert context_engine._context_engine_extensions_disabled_for_scoped_worker() + + +def test_normal_worker_can_execute_configured_context_engine( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _clear_lifecycle_env(monkeypatch) + marker = tmp_path / "normal-context-engine-executed" + plugins_dir = tmp_path / "context_engine" + engine_dir = plugins_dir / "normal_context" + engine_dir.mkdir(parents=True) + (engine_dir / "__init__.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed', encoding='utf-8')\n", + encoding="utf-8", + ) + + import plugins.context_engine as context_engine + + monkeypatch.setattr(context_engine, "_CONTEXT_ENGINE_PLUGINS_DIR", plugins_dir) + sys.modules.pop("plugins.context_engine.normal_context", None) + try: + assert context_engine.load_context_engine("normal_context") is None + finally: + sys.modules.pop("plugins.context_engine.normal_context", None) + + assert marker.exists() + + +def _install_startup_spies(monkeypatch: pytest.MonkeyPatch) -> dict[str, int]: + calls = {"plugin": 0, "mcp_background": 0, "mcp_inline": 0, "hook": 0} + + plugins = types.ModuleType("hermes_cli.plugins") + plugins.start_background_plugin_discovery = lambda: calls.__setitem__( + "plugin", calls["plugin"] + 1 + ) + monkeypatch.setitem(sys.modules, "hermes_cli.plugins", plugins) + + mcp_startup = types.ModuleType("hermes_cli.mcp_startup") + mcp_startup.start_background_mcp_discovery = lambda **_kwargs: calls.__setitem__( + "mcp_background", calls["mcp_background"] + 1 + ) + monkeypatch.setitem(sys.modules, "hermes_cli.mcp_startup", mcp_startup) + + mcp_tool = types.ModuleType("tools.mcp_tool") + mcp_tool.discover_mcp_tools = lambda: calls.__setitem__( + "mcp_inline", calls["mcp_inline"] + 1 + ) + monkeypatch.setitem(sys.modules, "tools.mcp_tool", mcp_tool) + + config = types.ModuleType("hermes_cli.config") + config.load_config = lambda: { + "hooks": {"on_session_start": [{"command": "adversarial-hook"}]}, + "mcp_servers": {"adversarial": {"url": "https://invalid.test/mcp"}}, + } + monkeypatch.setitem(sys.modules, "hermes_cli.config", config) + + shell_hooks = types.ModuleType("agent.shell_hooks") + shell_hooks.register_from_config = lambda *_args, **_kwargs: calls.__setitem__( + "hook", calls["hook"] + 1 + ) + monkeypatch.setitem(sys.modules, "agent.shell_hooks", shell_hooks) + return calls + + +@pytest.mark.parametrize("command", ["chat", "mcp"]) +def test_prepare_agent_startup_lifecycle_worker_calls_no_extension_surface( + monkeypatch: pytest.MonkeyPatch, + command: str, +) -> None: + from hermes_cli import main as main_mod + + _lifecycle_env(monkeypatch) + calls = _install_startup_spies(monkeypatch) + + main_mod._prepare_agent_startup(_agent_args(command)) + + assert calls == { + "plugin": 0, + "mcp_background": 0, + "mcp_inline": 0, + "hook": 0, + } + + +def test_prepare_agent_startup_normal_worker_preserves_extension_startup( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from hermes_cli import main as main_mod + + _clear_lifecycle_env(monkeypatch) + calls = _install_startup_spies(monkeypatch) + + main_mod._prepare_agent_startup(_agent_args("chat")) + assert calls == { + "plugin": 1, + "mcp_background": 1, + "mcp_inline": 0, + "hook": 1, + } + + main_mod._prepare_agent_startup(_agent_args("mcp")) + assert calls == { + "plugin": 2, + "mcp_background": 1, + "mcp_inline": 1, + "hook": 2, + } + + +def test_deferred_cli_startup_lifecycle_worker_starts_nothing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import cli as cli_mod + + _lifecycle_env(monkeypatch) + monkeypatch.setenv("HERMES_DEFER_AGENT_STARTUP", "1") + monkeypatch.setattr(cli_mod, "_deferred_agent_startup_done", False) + calls = _install_startup_spies(monkeypatch) + + cli_mod._prepare_deferred_agent_startup() + + assert cli_mod._deferred_agent_startup_done is True + assert calls == { + "plugin": 0, + "mcp_background": 0, + "mcp_inline": 0, + "hook": 0, + } + + +def test_model_tools_import_skips_plugin_discovery_only_for_lifecycle_worker( + tmp_path: Path, +) -> None: + code = """ +import hermes_cli.plugins as plugins +calls = [] +plugins.discover_plugins = lambda *args, **kwargs: calls.append('plugin') +import model_tools +print('PLUGIN_DISCOVERY_CALLS=' + str(len(calls))) +""" + base_env = os.environ.copy() + base_env["HERMES_HOME"] = str(tmp_path / ".hermes") + base_env["HERMES_QUIET"] = "1" + lifecycle_env = dict(base_env) + lifecycle_env.update( + { + "HERMES_KANBAN_TASK": "t_lifecycle_startup", + "HERMES_KANBAN_WORKER_SCOPE": "lifecycle-only", + } + ) + + scoped = subprocess.run( + [sys.executable, "-c", code], + cwd=Path(__file__).resolve().parents[2], + env=lifecycle_env, + capture_output=True, + text=True, + timeout=30, + check=True, + ) + assert "PLUGIN_DISCOVERY_CALLS=0" in scoped.stdout + + normal_env = dict(base_env) + normal_env.pop("HERMES_KANBAN_TASK", None) + normal_env.pop("HERMES_KANBAN_WORKER_SCOPE", None) + normal = subprocess.run( + [sys.executable, "-c", code], + cwd=Path(__file__).resolve().parents[2], + env=normal_env, + capture_output=True, + text=True, + timeout=30, + check=True, + ) + assert "PLUGIN_DISCOVERY_CALLS=1" in normal.stdout + + +def test_lifecycle_worker_blocks_lazy_plugin_mcp_and_hook_init( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _lifecycle_env(monkeypatch) + + from agent import shell_hooks + from hermes_cli import mcp_startup + from hermes_cli.plugins import PluginManager + from tools import mcp_tool + + plugin_calls: list[str] = [] + manager = PluginManager() + monkeypatch.setattr( + manager, + "_discover_and_load_inner", + lambda: plugin_calls.append("discovered"), + ) + manager.discover_and_load(force=True) + assert plugin_calls == [] + assert manager._discovered is True + + monkeypatch.setattr(mcp_startup, "_mcp_discovery_started", False) + monkeypatch.setattr(mcp_startup, "_mcp_discovery_thread", None) + monkeypatch.setattr( + mcp_startup, + "_has_configured_mcp_servers", + lambda: pytest.fail("lifecycle worker must not probe MCP configuration"), + ) + mcp_startup.start_background_mcp_discovery( + logger=types.SimpleNamespace(debug=lambda *_args, **_kwargs: None), + thread_name="lifecycle-worker-mcp", + ) + assert mcp_startup._mcp_discovery_started is False + assert mcp_startup._mcp_discovery_thread is None + + monkeypatch.setattr( + mcp_tool, + "_load_mcp_config", + lambda: pytest.fail("lifecycle worker must not read MCP config"), + ) + assert mcp_tool.discover_mcp_tools() == [] + assert mcp_tool.register_mcp_servers( + {"adversarial": {"url": "https://invalid.test/mcp"}} + ) == [] + + hook_calls: list[str] = [] + monkeypatch.setattr( + shell_hooks, + "_resolve_effective_accept", + lambda *_args, **_kwargs: hook_calls.append("resolved") or True, + ) + assert shell_hooks.register_from_config( + {"hooks": {"on_session_start": [{"command": "adversarial-hook"}]}}, + accept_hooks=True, + ) == [] + assert hook_calls == [] + + +def test_lazy_extension_guards_preserve_normal_behavior( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _clear_lifecycle_env(monkeypatch) + + from agent import shell_hooks + from hermes_cli.plugins import PluginManager + from tools import mcp_tool + + plugin_calls: list[str] = [] + manager = PluginManager() + monkeypatch.setattr( + manager, + "_discover_and_load_inner", + lambda: plugin_calls.append("discovered"), + ) + manager.discover_and_load() + assert plugin_calls == ["discovered"] + + mcp_config_calls: list[str] = [] + monkeypatch.setattr(mcp_tool, "_MCP_AVAILABLE", True) + monkeypatch.setattr( + mcp_tool, + "_load_mcp_config", + lambda: mcp_config_calls.append("loaded") or {}, + ) + assert mcp_tool.discover_mcp_tools() == [] + assert mcp_config_calls == ["loaded"] + + hook_calls: list[str] = [] + monkeypatch.setattr( + shell_hooks, + "_resolve_effective_accept", + lambda *_args, **_kwargs: hook_calls.append("resolved") or False, + ) + assert shell_hooks.register_from_config({}, accept_hooks=False) == [] + assert hook_calls == ["resolved"] diff --git a/tests/hermes_cli/test_kanban_scope_import_fail_closed.py b/tests/hermes_cli/test_kanban_scope_import_fail_closed.py new file mode 100644 index 000000000000..0f2a4a8e3403 --- /dev/null +++ b/tests/hermes_cli/test_kanban_scope_import_fail_closed.py @@ -0,0 +1,73 @@ +"""Fail closed when the dispatcher worker-scope oracle cannot be imported.""" + +from __future__ import annotations + +import builtins +from pathlib import Path + + +def test_extension_surfaces_stay_closed_when_scope_oracle_import_fails( + monkeypatch, + tmp_path: Path, +) -> None: + import plugins.memory as memory_plugins + import providers + + real_import = builtins.__import__ + + def fail_scope_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "hermes_cli.kanban_worker_scope": + raise ImportError("scope oracle unavailable") + return real_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fail_scope_import) + + entry_point_loaded = False + + class HostileEntryPoint: + name = "hostile" + + def load(self): + nonlocal entry_point_loaded + entry_point_loaded = True + raise AssertionError("entry point must not execute") + + imported_provider_dirs: list[tuple[Path, str]] = [] + user_provider_dir = tmp_path / "user-provider" + user_provider_dir.mkdir() + monkeypatch.setattr(providers, "_discovered", False) + monkeypatch.setattr(providers, "_BUNDLED_PLUGINS_DIR", tmp_path / "missing") + monkeypatch.setattr(providers, "_user_plugins_dir", lambda: user_provider_dir) + monkeypatch.setattr( + providers, + "_import_plugin_dir", + lambda path, source: imported_provider_dirs.append((path, source)), + ) + + providers._discover_entry_point_providers() + providers._discover_providers() + + memory_provider_dir = tmp_path / "memory-provider" + memory_provider_dir.mkdir() + (memory_provider_dir / "__init__.py").write_text( + "raise AssertionError('memory plugin must not execute')\n", + encoding="utf-8", + ) + + assert ( + memory_plugins._load_provider_from_entry_point( + HostileEntryPoint(), + register_skills=False, + ) + is None + ) + assert ( + memory_plugins._load_provider_from_dir( + memory_provider_dir, + register_skills=False, + ) + is None + ) + assert memory_plugins.discover_plugin_cli_commands() == [] + assert entry_point_loaded is False + assert imported_provider_dirs == [] diff --git a/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py b/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py index 14654b5168c4..52e5fbe86fd7 100644 --- a/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py +++ b/tests/hermes_cli/test_kanban_worker_spawn_toolsets.py @@ -2,6 +2,8 @@ import subprocess +import pytest + def _make_task(kb, *, assignee: str): return kb.Task( @@ -83,6 +85,8 @@ def fake_popen(cmd, *args, **kwargs): assert pid == 4242 assert captured["env"]["HERMES_HOME"] == str(profile) assert captured["env"]["HERMES_KANBAN_TASK"] == "t_spawn_tools" + assert "HERMES_KANBAN_WORKER_SCOPE" not in captured["env"] + assert "--accept-hooks" in captured["cmd"] assert "--toolsets" in captured["cmd"] pinned = captured["cmd"][captured["cmd"].index("--toolsets") + 1].split(",") for required in ("terminal", "web", "file", "skills", "code_execution", "delegation"): @@ -161,3 +165,171 @@ def test_resolve_worker_cli_toolsets_uses_profile_home_not_parent_config(monkeyp assert "web" in resolved assert "kanban" in resolved # recovered worker lifecycle surface assert resolved != ["kanban"] + + +def test_lifecycle_only_worker_surface_excludes_broader_kanban_tools(monkeypatch): + """A minimal worker profile must not be widened back to the full Kanban API.""" + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_lifecycle_only") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "lifecycle-only") + + from model_tools import _clear_tool_defs_cache, get_tool_definitions + + _clear_tool_defs_cache() + try: + definitions = get_tool_definitions( + enabled_toolsets=["kanban_lifecycle"], + disabled_toolsets=[], + quiet_mode=True, + ) + by_name = { + item.get("function", {}).get("name"): item.get("function", {}) + for item in definitions + if item.get("function", {}).get("name") + } + names = set(by_name) + assert names == { + "kanban_show", + "kanban_complete", + "kanban_block", + "kanban_heartbeat", + } + assert set(by_name["kanban_show"]["parameters"]["properties"]) == set() + assert set(by_name["kanban_complete"]["parameters"]["properties"]) == { + "summary", "metadata", "result", + } + assert set(by_name["kanban_block"]["parameters"]["properties"]) == { + "reason", "kind", + } + assert set(by_name["kanban_heartbeat"]["parameters"]["properties"]) == { + "note", + } + finally: + _clear_tool_defs_cache() + + +def test_resolve_worker_cli_toolsets_preserves_lifecycle_only_profile(monkeypatch, tmp_path): + root = tmp_path / ".hermes" + profile = root / "profiles" / "dashboardcontrol" + profile.mkdir(parents=True) + root.joinpath("config.yaml").write_text("{}\n", encoding="utf-8") + profile.joinpath("config.yaml").write_text( + """ +platform_toolsets: + cli: + - kanban_lifecycle +toolsets: + - kanban_lifecycle +agent: + disabled_toolsets: [] +""".lstrip(), + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(root)) + + from hermes_cli import kanban_db as kb + + resolved = kb._resolve_worker_cli_toolsets(str(profile)) + + assert resolved == ["kanban_lifecycle"] + + +def test_resolve_worker_cli_toolsets_fails_closed_on_config_error( + monkeypatch, tmp_path +): + profile = tmp_path / "profile" + profile.mkdir() + + from hermes_cli import kanban_db as kb + from hermes_cli import config as hermes_config + + monkeypatch.setattr( + hermes_config, + "load_config", + lambda: (_ for _ in ()).throw(ValueError("invalid profile config")), + ) + + with pytest.raises(RuntimeError, match="refusing spawn"): + kb._resolve_worker_cli_toolsets(str(profile)) + + +def test_default_spawn_does_not_start_process_when_toolset_resolution_fails( + monkeypatch, tmp_path +): + root = tmp_path / ".hermes" + profile = root / "profiles" / "dashboardcontrol" + profile.mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(root)) + + from hermes_cli import kanban_db as kb + + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: ["hermes"]) + monkeypatch.setattr( + kb, + "_resolve_worker_cli_toolsets", + lambda _home: (_ for _ in ()).throw(RuntimeError("resolution failed")), + ) + monkeypatch.setattr( + subprocess, + "Popen", + lambda *_args, **_kwargs: pytest.fail("worker process must not start"), + ) + workspace = tmp_path / "workspace" + workspace.mkdir() + + with pytest.raises(RuntimeError, match="resolution failed"): + kb._default_spawn( + _make_task(kb, assignee="dashboardcontrol"), + str(workspace), + ) + + +def test_default_spawn_pins_lifecycle_process_scope_and_suppresses_hooks( + monkeypatch, tmp_path +): + root = tmp_path / ".hermes" + profile = root / "profiles" / "dashboardcontrol" + profile.mkdir(parents=True) + root.joinpath("config.yaml").write_text("{}\n", encoding="utf-8") + profile.joinpath("config.yaml").write_text( + """ +platform_toolsets: + cli: + - kanban_lifecycle +toolsets: + - kanban_lifecycle +""".lstrip(), + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(root)) + monkeypatch.setenv("HERMES_ACCEPT_HOOKS", "1") + monkeypatch.setenv("HERMES_TUI", "1") + monkeypatch.setenv("HERMES_TENANT", "stale-parent-tenant") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "inherited-invalid") + + from hermes_cli import kanban_db as kb + + monkeypatch.setattr(kb, "_resolve_hermes_argv", lambda: ["hermes"]) + captured = {} + + class FakeProc: + pid = 4246 + + def fake_popen(cmd, *args, **kwargs): + captured["cmd"] = list(cmd) + captured["env"] = dict(kwargs.get("env") or {}) + return FakeProc() + + monkeypatch.setattr(subprocess, "Popen", fake_popen) + workspace = tmp_path / "workspace" + workspace.mkdir() + + assert kb._default_spawn( + _make_task(kb, assignee="dashboardcontrol"), str(workspace) + ) == 4246 + assert captured["env"]["HERMES_KANBAN_WORKER_SCOPE"] == "lifecycle-only" + assert "HERMES_ACCEPT_HOOKS" not in captured["env"] + assert "HERMES_TUI" not in captured["env"] + assert "HERMES_TENANT" not in captured["env"] + assert "--accept-hooks" not in captured["cmd"] + pinned = captured["cmd"][captured["cmd"].index("--toolsets") + 1] + assert pinned == "kanban_lifecycle" diff --git a/tests/tools/test_delegate_kanban_isolation.py b/tests/tools/test_delegate_kanban_isolation.py index cc12f62ea0df..914baf172edf 100644 --- a/tests/tools/test_delegate_kanban_isolation.py +++ b/tests/tools/test_delegate_kanban_isolation.py @@ -144,6 +144,7 @@ def test_delegate_child_execute_code_env_bridges_contextvar_and_scrubs_kanban( home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setenv("HERMES_KANBAN_TASK", "t_parent") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "lifecycle-only") monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "123") monkeypatch.setenv("HERMES_KANBAN_DB", str(home / "kanban.db")) monkeypatch.setenv("HERMES_KANBAN_WORKSPACE", str(tmp_path / "parent-workspace")) @@ -164,6 +165,7 @@ def test_delegate_child_execute_code_env_bridges_contextvar_and_scrubs_kanban( assert env["HERMES_HOME"] == str(home) assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1" assert "HERMES_KANBAN_TASK" not in env + assert "HERMES_KANBAN_WORKER_SCOPE" not in env assert "HERMES_KANBAN_RUN_ID" not in env assert "HERMES_KANBAN_DB" not in env assert "HERMES_KANBAN_WORKSPACE" not in env diff --git a/tests/tools/test_hermes_subprocess_env.py b/tests/tools/test_hermes_subprocess_env.py index 9838f10cb590..2670ca104381 100644 --- a/tests/tools/test_hermes_subprocess_env.py +++ b/tests/tools/test_hermes_subprocess_env.py @@ -149,6 +149,7 @@ def test_delegated_child_context_scrubs_parent_kanban_keys_and_sets_marker(self) { **_SAFE_SAMPLE, "HERMES_KANBAN_TASK": "t_parent", + "HERMES_KANBAN_WORKER_SCOPE": "lifecycle-only", "HERMES_KANBAN_RUN_ID": "123", "HERMES_KANBAN_DB": "/tmp/parent-kanban.db", "HERMES_KANBAN_WORKSPACE": "/tmp/parent-workspace", @@ -160,6 +161,7 @@ def test_delegated_child_context_scrubs_parent_kanban_keys_and_sets_marker(self) assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1" assert "HERMES_KANBAN_TASK" not in env + assert "HERMES_KANBAN_WORKER_SCOPE" not in env assert "HERMES_KANBAN_RUN_ID" not in env assert "HERMES_KANBAN_DB" not in env assert "HERMES_KANBAN_WORKSPACE" not in env diff --git a/tests/tools/test_kanban_lifecycle_scope.py b/tests/tools/test_kanban_lifecycle_scope.py new file mode 100644 index 000000000000..102747dd60cc --- /dev/null +++ b/tests/tools/test_kanban_lifecycle_scope.py @@ -0,0 +1,181 @@ +"""Fail-closed contracts for lifecycle-only Kanban worker processes.""" + +from __future__ import annotations + +import json + +import pytest + + +@pytest.fixture +def lifecycle_task(monkeypatch: pytest.MonkeyPatch, tmp_path): + home = tmp_path / ".hermes" + home.mkdir() + workspace = tmp_path / "workspace" + workspace.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_PROFILE", "dashboardcontrol") + monkeypatch.setenv("HERMES_KANBAN_BOARD", "default") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "lifecycle-only") + monkeypatch.delenv("HERMES_SESSION_ID", raising=False) + + from hermes_cli import kanban_db as kb + + kb._INITIALIZED_PATHS.clear() + kb.init_db() + conn = kb.connect() + try: + task_id = kb.create_task( + conn, + title="lifecycle task", + assignee="dashboardcontrol", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + foreign_id = kb.create_task( + conn, + title="foreign task", + assignee="other", + ) + kb.claim_task(conn, task_id) + finally: + conn.close() + monkeypatch.setenv("HERMES_KANBAN_TASK", task_id) + return task_id, foreign_id, workspace + + +def test_lifecycle_scope_forces_exact_surface_and_narrows_schemas( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_lifecycle_schema") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "lifecycle-only") + + from model_tools import _clear_tool_defs_cache, get_tool_definitions + + _clear_tool_defs_cache() + try: + definitions = get_tool_definitions( + enabled_toolsets=["hermes-cli", "terminal", "kanban"], + disabled_toolsets=[], + quiet_mode=True, + ) + finally: + _clear_tool_defs_cache() + + by_name = {item["function"]["name"]: item["function"] for item in definitions} + assert set(by_name) == { + "kanban_show", + "kanban_complete", + "kanban_block", + "kanban_heartbeat", + } + for function in by_name.values(): + properties = function["parameters"].get("properties", {}) + assert "task_id" not in properties + assert "board" not in properties + complete_properties = by_name["kanban_complete"]["parameters"]["properties"] + assert "created_cards" not in complete_properties + assert "artifacts" not in complete_properties + + +@pytest.mark.parametrize( + "name,args", + [ + ("terminal", {"command": "pwd"}), + ("kanban_create", {"title": "forged"}), + ("kanban_comment", {"task_id": "t_deadbeef", "body": "forged"}), + ("tool_call", {"name": "terminal", "arguments": {"command": "pwd"}}), + ], +) +def test_forged_dispatch_is_denied_before_any_tool_path( + monkeypatch: pytest.MonkeyPatch, + name: str, + args: dict, +) -> None: + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_lifecycle_dispatch") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "lifecycle-only") + + from model_tools import handle_function_call + + result = json.loads( + handle_function_call( + name, + args, + skip_pre_tool_call_hook=True, + skip_tool_request_middleware=True, + skip_tool_execution_middleware=True, + ) + ) + assert "unavailable in lifecycle-only" in result["error"] + + +def test_unknown_nonempty_scope_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("HERMES_KANBAN_TASK", "t_lifecycle_dispatch") + monkeypatch.setenv("HERMES_KANBAN_WORKER_SCOPE", "typo-broad") + + from model_tools import handle_function_call + + result = json.loads(handle_function_call("kanban_show", {})) + assert "invalid worker scope" in result["error"] + + +def test_lifecycle_handler_rejects_foreign_task_board_and_attachment_fields( + lifecycle_task, +) -> None: + task_id, foreign_id, _workspace = lifecycle_task + from tools import kanban_tools as kt + + foreign = json.loads(kt._handle_show({"task_id": foreign_id})) + assert "scoped to task" in foreign["error"] + + wrong_board = json.loads(kt._handle_show({"board": "other-board"})) + assert "refusing board override" in wrong_board["error"] + + for forbidden in ( + {"task_id": task_id}, + {"board": "default"}, + {"created_cards": []}, + {"artifacts": []}, + ): + result = json.loads( + kt._handle_complete({"summary": "bounded", **forbidden}) + ) + assert "unavailable in lifecycle-only" in result["error"] + + metadata_artifact = json.loads( + kt._handle_complete( + {"summary": "bounded", "metadata": {"artifacts": []}} + ) + ) + assert "metadata.artifacts is unavailable" in metadata_artifact["error"] + + +def test_lifecycle_completion_persists_structured_output_without_artifacts( + lifecycle_task, +) -> None: + task_id, _foreign_id, workspace = lifecycle_task + decoy = workspace / "decoy.json" + decoy.write_text('{"not":"worker output"}\n', encoding="utf-8") + + from hermes_cli import kanban_db as kb + from tools import kanban_tools as kt + + payload = {"schema": "control-output/v1", "status": "ready"} + result = json.loads( + kt._handle_complete( + { + "summary": f"done; legacy path {decoy}", + "metadata": {"control_output": payload}, + } + ) + ) + assert result["ok"] is True + + conn = kb.connect() + try: + run = kb.latest_run(conn, task_id) + assert run.outcome == "completed" + assert run.metadata == {"control_output": payload} + assert all(event.kind != "attached" for event in kb.list_events(conn, task_id)) + finally: + conn.close() diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index d49b53a2212b..6eed83869e20 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -35,6 +35,7 @@ from agent.redact import redact_sensitive_text from hermes_cli.goals import judge_goal +from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker from tools.registry import registry, tool_error from hermes_cli.config import cfg_get, load_config @@ -224,6 +225,18 @@ def _connect(board: Optional[str] = None): the env-pinned active board without restarting Hermes. """ from hermes_cli import kanban_db as kb + + if os.environ.get("HERMES_KANBAN_TASK"): + pinned_raw = str(os.environ.get("HERMES_KANBAN_BOARD") or "").strip() + if board is not None and pinned_raw: + requested = kb._normalize_board_slug(board) + pinned = kb._normalize_board_slug(pinned_raw) + if requested != pinned: + raise ValueError( + f"worker is pinned to board {pinned}; refusing board override {requested}" + ) + if pinned_raw: + board = pinned_raw return kb, kb.connect(board=board) @@ -522,6 +535,10 @@ def _handle_show(args: dict, **kw) -> str: return tool_error( "task_id is required (or set HERMES_KANBAN_TASK in the env)" ) + if is_lifecycle_only_worker(): + ownership_err = _enforce_worker_task_ownership(tid) + if ownership_err: + return ownership_err board = args.get("board") try: kb, conn = _connect(board=board) @@ -665,6 +682,14 @@ def _handle_complete(args: dict, **kw) -> str: ownership_err = _enforce_worker_task_ownership(tid) if ownership_err: return ownership_err + lifecycle_only = is_lifecycle_only_worker() + if lifecycle_only and any( + key in args for key in ("task_id", "board", "created_cards", "artifacts") + ): + return tool_error( + "task_id, board, created_cards, and artifacts are unavailable " + "in lifecycle-only worker posture" + ) summary = args.get("summary") metadata = args.get("metadata") result = args.get("result") @@ -673,6 +698,10 @@ def _handle_complete(args: dict, **kw) -> str: if result: result = redact_sensitive_text(str(result), force=True) if metadata is not None and isinstance(metadata, dict): + if lifecycle_only and "artifacts" in metadata: + return tool_error( + "metadata.artifacts is unavailable in lifecycle-only worker posture" + ) meta_json = json.dumps(metadata) meta_json = redact_sensitive_text(meta_json, force=True) try: @@ -771,6 +800,7 @@ def _handle_complete(args: dict, **kw) -> str: result=result, summary=summary, metadata=metadata, created_cards=created_cards, expected_run_id=_worker_run_id(tid), + allow_artifacts=not lifecycle_only, ) except kb.ArtifactPreservationError as artifact_err: return tool_error( diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index f380b8cea4ff..c262957fbbd7 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -5104,6 +5104,10 @@ def _load_mcp_config() -> Dict[str, dict]: ``os.environ`` (which includes ``~/.hermes/.env`` loaded at startup). """ try: + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return {} from hermes_cli.config import load_config from utils import env_var_enabled as _env_enabled @@ -6750,6 +6754,10 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: Returns: List of all currently registered MCP tool names. """ + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return [] if not _ensure_mcp_sdk(): logger.debug("MCP SDK not available -- skipping explicit MCP registration") return [] @@ -6964,6 +6972,10 @@ def discover_mcp_tools() -> List[str]: Returns: List of all registered MCP tool names. """ + from hermes_cli.kanban_worker_scope import is_lifecycle_only_worker + + if is_lifecycle_only_worker(): + return [] servers = _load_mcp_config() if not servers: logger.debug("No MCP servers configured") diff --git a/toolsets.py b/toolsets.py index 8bb86024f5f5..64589f15d03b 100644 --- a/toolsets.py +++ b/toolsets.py @@ -311,6 +311,19 @@ "includes": [] }, + "kanban_lifecycle": { + "description": ( + "Minimal task lifecycle surface for locked-down dispatcher workers: " + "inspect the assigned task, heartbeat, block, or complete. Excludes " + "board routing, child creation, comments, attachments, network fetches, " + "and every non-Kanban tool." + ), + "tools": [ + "kanban_show", "kanban_complete", "kanban_block", "kanban_heartbeat", + ], + "includes": [], + }, + "kanban": { "description": ( "Kanban multi-agent coordination — only active when the agent "