From 284d220ba48e25f2e3623b3afe72db8f24a4c2db Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 11 Sep 2026 19:19:44 -0700 Subject: [PATCH] fix(multiplex): cron, kanban, /loop and completion paths for a served profile match its standalone gateway MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under gateway.multiplex_profiles a secondary profile X is ticked, dispatched and notified from the default profile's process, where os.environ holds the DEFAULT profile's .env and X's values live only in the per-turn secret scope / HERMES_HOME override. Every remaining read that skipped that scope made X behave differently from `hermes -p X gateway run`: - cron: HERMES_CRON_TIMEOUT, HERMES_MODEL (job/preflight fallback), HERMES_CRON_MAX_PARALLEL, inflight allowance, prefill file and the script timeout were bare os.getenv → the default profile's values; a job without a model silently ran on the default's HERMES_MODEL instead of refusing. cron/env_settings.py::cron_env_setting reads the scope (fire) or the ticked home's .env (tick thread), plain environ when multiplexing is off. - child env: the restart-safe cron worker, the Bot Chat delivery child and the kanban worker inherited the launch profile's non-credential .env settings and bridged TERMINAL_* policy (TERMINAL_ENV=docker, default's image, HERMES_MODEL) — X's worker ran in the default's docker image on the default's model. tools/environments/local.py::strip_launch_profile_env drops them when the child targets another served profile. - kanban: the worker --toolsets pin was silently dropped for every served assignee (toolset probes call get_secret without a scope → swallowed UnscopedSecretError); notifier pings, artifact uploads and the wake text ran under the default's media policy / display language (only wake() was scoped). - /loop: _post_turn_loop_completion hopped to the executor without contextvars, writing the completed tick into the DEFAULT profile's state.db and leaving X's row awaiting_response forever; the --until judge ran with the default's aux credentials. - background processes: a secondary's processes.json (scope-relative since adf23550f589) was never read at startup; its processes were not re-adopted and notify_on_complete notices were lost. Startup recovers every served home under its scope; recovery adopts each session once. - completion delivery: background_process_notifications was evaluated once per drain for the ambient profile (default's mode for everyone; X's `off` dropped a sibling's `all` event), recovered watchers used the default's mode, HERMES_BACKGROUND_NOTIFICATIONS was read raw from environ; _deliver_platform_notice used the default's GatewayConfig so a secondary's notice_delivery: private went public. Not changed: gateway/run.py and tools/async_delegation.py (PR #106742 rewrites both). Known residue left for the env-bridge lane: HERMES_SESSION_STALL_TIMEOUT is bridged once from the launch config. scheduler bug, not a parity gap; unchanged here. --- cron/env_settings.py | 26 +++ cron/jobs.py | 3 +- cron/scheduler.py | 22 +-- cron/scheduler_delivery.py | 3 +- cron/scheduler_preflight.py | 4 +- cron/scheduler_script.py | 3 +- gateway/kanban_watchers_notifier.py | 22 ++- gateway/run_config_loaders.py | 8 +- gateway/run_goals.py | 8 +- gateway/run_notifications.py | 46 +++-- gateway/run_startup.py | 25 ++- hermes_cli/kanban_db_dispatch.py | 14 +- ...st_cron_multiplex_served_profile_parity.py | 104 +++++++++++ ..._kanban_multiplex_served_profile_parity.py | 124 +++++++++++++ .../test_multiplex_served_profile_parity.py | 163 ++++++++++++++++++ tools/environments/local.py | 23 +++ tools/process_registry_checkpoint.py | 6 + .../docs/user-guide/multi-profile-gateways.md | 3 + 18 files changed, 565 insertions(+), 42 deletions(-) create mode 100644 cron/env_settings.py create mode 100644 tests/cron/test_cron_multiplex_served_profile_parity.py create mode 100644 tests/gateway/test_kanban_multiplex_served_profile_parity.py create mode 100644 tests/gateway/test_multiplex_served_profile_parity.py diff --git a/cron/env_settings.py b/cron/env_settings.py new file mode 100644 index 0000000000..6f713d6fd5 --- /dev/null +++ b/cron/env_settings.py @@ -0,0 +1,26 @@ +"""Profile-scoped reads of the ``HERMES_*`` tuning settings cron honours from ``.env``. + +A standalone ``hermes -p X gateway run`` loads X's ``.env`` into ``os.environ``, so a bare +``os.getenv("HERMES_CRON_TIMEOUT")`` is X's value. Under ``gateway.multiplex_profiles`` the same +tick runs inside the default profile's process, where ``os.environ`` holds the DEFAULT profile's +``.env``. With a secret scope installed (job run + delivery) the scope is authoritative; the tick +loop itself (due-job scan, pool sizing) runs under the profile's home override only, so the +setting is read from that home's ``.env``. Outside multiplex the read is the plain environ. +""" + +from __future__ import annotations + +import os + +from agent.secret_scope import current_secret_scope, is_multiplex_active, load_env_file +from hermes_constants import get_hermes_home + + +def cron_env_setting(name: str, default: str = "") -> str: + if not is_multiplex_active(): + return os.getenv(name) or default + scope = current_secret_scope() + if scope is None: + scope = load_env_file(get_hermes_home() / ".env") + value = scope.get(name) + return default if value is None else str(value) diff --git a/cron/jobs.py b/cron/jobs.py index 27b7166342..d24acc0b19 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -28,6 +28,7 @@ except ImportError: # pragma: no cover - non-Windows from datetime import datetime, timedelta from pathlib import Path from hermes_constants import get_hermes_home +from cron.env_settings import cron_env_setting from typing import Optional, Dict, List, Any, Callable, Set, Tuple, Union, Collection logger = logging.getLogger(__name__) @@ -164,7 +165,7 @@ _DEFAULT_CRON_INACTIVITY_TIMEOUT = 600.0 def _oneshot_run_claim_ttl_seconds() -> float: """One-shot running-claim TTL from ``HERMES_CRON_TIMEOUT``: unset/invalid → 600s → 1800s; ``0`` (unlimited) → the fixed floor; positive N → ``max(N * headroom, floor)``.""" - raw = os.getenv("HERMES_CRON_TIMEOUT", "").strip() + raw = cron_env_setting("HERMES_CRON_TIMEOUT").strip() try: timeout = float(raw) if raw else _DEFAULT_CRON_INACTIVITY_TIMEOUT except (ValueError, TypeError): diff --git a/cron/scheduler.py b/cron/scheduler.py index 04dcdce88b..7941a73e29 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -36,6 +36,7 @@ from typing import Any, Callable, List, Optional, Protocol sys.path.insert(0, str(Path(__file__).parent.parent)) from hermes_constants import get_hermes_home +from cron.env_settings import cron_env_setting from hermes_cli._subprocess_compat import windows_hide_flags from hermes_cli.config import ( _expand_env_vars, load_config, resolve_cron_model_drift_defaults) @@ -605,7 +606,7 @@ def _inflight_min_allowance_minutes() -> float: val = float(_cfg_val) if val > 0: return val - raw = os.getenv("HERMES_CRON_INFLIGHT_MAX_MINUTES", "").strip() + raw = cron_env_setting("HERMES_CRON_INFLIGHT_MAX_MINUTES").strip() if raw: try: val = float(raw) @@ -936,7 +937,7 @@ def _cron_inactivity_seconds() -> float: """Parse HERMES_CRON_TIMEOUT (seconds). 0 = unlimited; bad input = 600. Shared by the inactivity monitor and the cwd-lock bound so they can't drift: the lock bound must stay >= the inactivity limit or waiters fail while a healthy holder runs.""" - raw = os.getenv("HERMES_CRON_TIMEOUT", "").strip() + raw = cron_env_setting("HERMES_CRON_TIMEOUT").strip() if not raw: return 600.0 try: @@ -1355,7 +1356,7 @@ def _load_cron_job_config(job: dict, job_id: str, job_name: str) -> _CronJobConf """Load config.yaml and resolve the run's model: per-job override > cron.model (fleet default) > creation snapshot > HERMES_MODEL > config ``model:``. Re-read every tick (no cache) so ``hermes cron edit --model`` applies next tick.""" - model = job.get("model") or os.getenv("HERMES_MODEL") or "" + model = job.get("model") or cron_env_setting("HERMES_MODEL") or "" _cron_default_provider = "" _cfg: dict = {} _model_cfg: Any = {} @@ -1380,7 +1381,8 @@ def _load_cron_job_config(job: dict, job_id: str, job_name: str) -> _CronJobConf if _cron_default_model: model = _cron_default_model else: - _, _global_model = resolve_cron_model_drift_defaults(_cfg) + _, _global_model = resolve_cron_model_drift_defaults( + _cfg, environ={"HERMES_MODEL": cron_env_setting("HERMES_MODEL")}) model = _snapshot_pin(job, "model", _global_model, job_id) or _global_model or model except Exception as e: logger.warning("Job '%s': failed to load config.yaml, using defaults: %s", job_id, e) @@ -1391,7 +1393,7 @@ def _load_cron_job_config(job: dict, job_id: str, job_name: str) -> _CronJobConf raise RuntimeError( f"Cron job '{job_name}' has no model configured " f"(job.model={job.get('model')!r}, " - f"HERMES_MODEL={os.getenv('HERMES_MODEL', '')!r}, " + f"HERMES_MODEL={cron_env_setting('HERMES_MODEL')!r}, " "config.yaml model.default missing or empty). " f"Set a per-job model via " f"`hermes cron edit {job_id} --model ` or set a " @@ -1410,7 +1412,7 @@ def _load_prefill_messages(cfg: dict, job_id: str) -> Optional[list]: """Prefill messages from env or config.yaml (top-level key canonical; agent.* is legacy).""" agent_cfg = cfg.get("agent", {}) if isinstance(cfg.get("agent", {}), dict) else {} prefill_file = ( - os.getenv("HERMES_PREFILL_MESSAGES_FILE", "") + cron_env_setting("HERMES_PREFILL_MESSAGES_FILE") or cfg.get("prefill_messages_file", "") or agent_cfg.get("prefill_messages_file", "") ) @@ -3072,7 +3074,7 @@ def _launch_external_cron_worker(job: dict) -> bool: set_secret_scope, ) from hermes_cli.env_loader import hydrate_profile_secret_sources - from tools.environments.local import build_subprocess_env + from tools.environments.local import build_subprocess_env, strip_launch_profile_env from tools.process_registry import ( restart_safe_gateway_child_argv, systemd_user_bus_env, @@ -3117,11 +3119,11 @@ def _launch_external_cron_worker(job: dict) -> bool: hydrate_profile_secret_sources(profile_home) secret_token = set_secret_scope(build_profile_secret_scope(profile_home)) try: - worker_env = build_subprocess_env( + worker_env = strip_launch_profile_env(build_subprocess_env( scrub_secrets=multiplex_active, inherit_profile_home=True, extra={"HERMES_HOME": str(profile_home)}, - ) + )) finally: reset_secret_scope(secret_token) worker_env = systemd_user_bus_env(worker_env) @@ -3527,7 +3529,7 @@ def _sweep_stale_inflight_for_tick(due_jobs: list) -> None: def _resolve_max_parallel_workers() -> Optional[int]: """Max workers: env > config.yaml > unbounded (HERMES_CRON_MAX_PARALLEL=1 restores serial).""" try: - _env_par = os.getenv("HERMES_CRON_MAX_PARALLEL", "").strip() + _env_par = cron_env_setting("HERMES_CRON_MAX_PARALLEL").strip() if _env_par: return int(_env_par) or None except (ValueError, TypeError): diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index d2589cebab..90c02b331c 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -733,7 +733,8 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str] return msg from agent.delegation_context import delegated_child_subprocess_env - env = delegated_child_subprocess_env(os.environ) + from tools.environments.local import strip_launch_profile_env + env = strip_launch_profile_env(delegated_child_subprocess_env(os.environ)) if profile: argv += ["-p", profile] # -p owns profile resolution; this scheduler's HERMES_HOME must not shadow it. diff --git a/cron/scheduler_preflight.py b/cron/scheduler_preflight.py index 21d524a937..208c0057b4 100644 --- a/cron/scheduler_preflight.py +++ b/cron/scheduler_preflight.py @@ -14,6 +14,8 @@ import logging import os from typing import Optional +from cron.env_settings import cron_env_setting + # Log-record parity with the origin module. logger = logging.getLogger("cron.scheduler") @@ -94,7 +96,7 @@ def _preflight_check_provider_key(job: dict, cfg: dict) -> Optional[str]: _cron_cfg = cfg.get("cron") if isinstance(cfg.get("cron"), dict) else {} requested = ( job.get("provider") or str((_cron_cfg or {}).get("model_provider") or "").strip() or None) - model = job.get("model") or os.getenv("HERMES_MODEL") or "" + model = job.get("model") or cron_env_setting("HERMES_MODEL") or "" from hermes_cli.auth import AuthError try: diff --git a/cron/scheduler_script.py b/cron/scheduler_script.py index 04e66ae8ef..074f4fa987 100644 --- a/cron/scheduler_script.py +++ b/cron/scheduler_script.py @@ -18,6 +18,7 @@ import subprocess import sys import threading import time +from cron.env_settings import cron_env_setting from cron.jobs import _ensure_cron_dir from pathlib import Path from typing import Any, Callable, Optional, TYPE_CHECKING @@ -41,7 +42,7 @@ def _timeout_from_env_or_config( env_var: str, config_key: str, parse: Callable[[Any], Any], label: str): """Shared env → ``cron.`` resolution. ``parse`` returns the value or None to keep looking; a parse error on the env var WARNs, on config DEBUGs. None when neither yields.""" - env_value = os.getenv(env_var, "").strip() + env_value = cron_env_setting(env_var).strip() if env_value: try: value = parse(env_value) diff --git a/gateway/kanban_watchers_notifier.py b/gateway/kanban_watchers_notifier.py index 68debcaf2a..0d94f7b4d3 100644 --- a/gateway/kanban_watchers_notifier.py +++ b/gateway/kanban_watchers_notifier.py @@ -7,6 +7,7 @@ per-subscription delivery (``_KanbanNotification``) live here. from __future__ import annotations +import contextlib import re from functools import partial from pathlib import Path @@ -486,6 +487,16 @@ class _KanbanNotification: logger.info("kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", self.task_id, self.platform_str, self.sub["chat_id"], self.sub_profile or "default", self.wake_kinds) + def _owner_scope(self): + """Runtime scope of the subscription's profile under multiplex, else a no-op context.""" + runner = self.runner + if not (self.sub_profile and getattr(getattr(runner, "config", None), "multiplex_profiles", False)): + return contextlib.nullcontext() + from gateway.run import _async_profile_runtime_scope + from gateway.session import SessionSource + source = SessionSource(platform=self.plat, chat_id=self.sub["chat_id"], profile=self.sub_profile) + return _async_profile_runtime_scope(runner._resolve_profile_home_for_source(source)) + async def wake(self) -> None: """Wake the creator session (raises on failure): push adapters get a full SessionSource, non-push a raw self-post.""" from gateway.wake import deliver_wake @@ -601,10 +612,13 @@ class _KanbanNotification: from gateway.wake import adapter_supports_push self.is_push_adapter = adapter_supports_push(adapter) - if not await self._send_pings(): - return - # All text pings delivered (or skipped for non-push / wake-only). - self.build_wake_text() + # Pings, artifact uploads (media policy) and the wake text (display.language) all read the + # SUBSCRIBER profile's config; the notifier thread itself runs in the launch profile's scope. + async with self._owner_scope(): + if not await self._send_pings(): + return + # All text pings delivered (or skipped for non-push / wake-only). + self.build_wake_text() wake_kinds, is_push = self.wake_kinds, self.is_push_adapter from gateway.wake import WakeNotAccepted diff --git a/gateway/run_config_loaders.py b/gateway/run_config_loaders.py index 7523d82026..7a02647966 100644 --- a/gateway/run_config_loaders.py +++ b/gateway/run_config_loaders.py @@ -381,9 +381,13 @@ class GatewayConfigLoadersMixin: @staticmethod def _load_background_notifications_mode() -> str: - """Background process notification mode from env/config (default ``concise``).""" + """Background process notification mode from env/config (default ``concise``), resolved for + the AMBIENT profile — callers deciding for another profile's event enter its scope first + (``_completion_event_scope``). The env override reads through the secret scope so a served + secondary sees its own ``.env`` value, not the launch profile's ``os.environ``.""" from gateway.run import _load_gateway_runtime_config - mode = os.getenv("HERMES_BACKGROUND_NOTIFICATIONS", "") + from gateway.authz_mixin import _platform_gate_env + mode = _platform_gate_env("HERMES_BACKGROUND_NOTIFICATIONS") if not mode: raw = cfg_get(_load_gateway_runtime_config(), "display", "background_process_notifications") if raw is False: diff --git a/gateway/run_goals.py b/gateway/run_goals.py index 7e4688b9aa..b87f006438 100644 --- a/gateway/run_goals.py +++ b/gateway/run_goals.py @@ -357,10 +357,10 @@ class GatewayGoalsMixin: state = mgr.state if mgr is not None else None if state is None or not state.awaiting_response: return - # The --until judge is a sync aux-LLM call — keep it off the event loop. - decision = await asyncio.get_running_loop().run_in_executor( - None, mgr.complete_tick, final_response or "" - ) + # The --until judge is a sync aux-LLM call — keep it off the event loop, but carry the + # contextvars: a bare executor hop drops the profile HERMES_HOME override and secret scope, + # so a served secondary's tick would be written into the DEFAULT profile's state.db. + decision = await self._run_in_executor_with_context(mgr.complete_tick, final_response or "") msg = decision.get("message") or "" if msg and source is not None: await self._defer_goal_status_notice_after_delivery(source, msg) diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index 2329a58a69..c3bd778e64 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -113,10 +113,18 @@ class GatewayNotificationsMixin: if config and getattr(source, "platform", None) == Platform.SLACK and _is_slack_ignored_channel(config, chat_id, adapter): logger.info("Skipping Slack platform notice for configured ignored channel %s", chat_id) return - notice_delivery = ( - config.get_notice_delivery(source.platform) if config and hasattr(config, "get_notice_delivery") - else "public" - ) + # The routed adapter carries ITS profile's ``platforms.

`` block; ``self.config`` is the + # launch profile's, so a served secondary's ``notice_delivery: private`` would be ignored. + adapter_config = getattr(adapter, "config", None) + adapter_extra = getattr(adapter_config, "extra", None) + if isinstance(adapter_extra, dict) and "notice_delivery" in adapter_extra: + from gateway.config import _normalize_choice + notice_delivery = _normalize_choice(adapter_extra.get("notice_delivery"), {"public", "private"}, "public") + else: + notice_delivery = ( + config.get_notice_delivery(source.platform) if config and hasattr(config, "get_notice_delivery") + else "public" + ) metadata = self._thread_metadata_for_source(source) if notice_delivery == "private" and getattr(source, "user_id", None): with _log_suppressed( @@ -948,23 +956,26 @@ class GatewayNotificationsMixin: """Consume queued watch events and inject them when notifications are enabled. The queue is ALWAYS drained (so watch events don't rot or requeue-spin) but injection is - skipped entirely when ``display.background_process_notifications`` is ``off``. + skipped when the OWNING profile's ``display.background_process_notifications`` is ``off`` + — one shared queue carries every served profile's events, so the gate is evaluated per + event inside its profile scope, never once for the ambient (launch) profile. See #9290. """ from gateway.run import _drain_gateway_watch_events, _format_gateway_process_notification watch_events = _drain_gateway_watch_events(completion_queue) - if self._load_background_notifications_mode() == "off": - return for evt in watch_events: - synth_text = _format_gateway_process_notification(evt) - if not synth_text: - continue - try: - delivered = await self._inject_watch_notification(synth_text, evt) - except Exception: - logger.exception("Watch notification injection error") - delivered = False + async with self._completion_event_scope(evt): + if self._load_background_notifications_mode() == "off": + continue + synth_text = _format_gateway_process_notification(evt) + if not synth_text: + continue + try: + delivered = await self._inject_watch_notification(synth_text, evt) + except Exception: + logger.exception("Watch notification injection error") + delivered = False if delivered is False: completion_queue.put(evt) @@ -1719,7 +1730,10 @@ class GatewayNotificationsMixin: chat_id = watcher.get("chat_id", "") thread_id = watcher.get("thread_id", "") agent_notify = watcher.get("notify_on_complete", False) - notify_mode = self._load_background_notifications_mode() + # The mode belongs to the profile that started the process; recovered watchers run in the + # root context, so resolve it under the owning profile's scope (no-op for the default). + async with self._completion_event_scope(watcher): + notify_mode = self._load_background_notifications_mode() logger.debug("Process watcher started: %s (every %ss, notify=%s, agent_notify=%s)", session_id, interval, notify_mode, agent_notify) silent = notify_mode == "off" and not agent_notify diff --git a/gateway/run_startup.py b/gateway/run_startup.py index 77825c3cdb..48fe825963 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -16,6 +16,7 @@ import signal import time from contextlib import suppress from datetime import datetime +from pathlib import Path from gateway.config import Platform from gateway.delivery import looks_like_telegram_private_chat_id from gateway.platforms.base import BasePlatformAdapter @@ -910,15 +911,37 @@ class GatewayStartupMixin: except Exception: logger.log(level, fail_fmt, *fail_args, exc_info=True) + def _recover_secondary_process_checkpoints(self, process_registry) -> int: + """Replay every SERVED secondary profile's ``processes.json`` under its own scope. + The launch profile's file was already read by ``recover_from_checkpoint`` above.""" + if not getattr(self.config, "multiplex_profiles", False): + return 0 + from gateway.run import _multiplex_profile_homes, _profile_runtime_scope + from hermes_constants import get_hermes_home + launch_home = get_hermes_home().resolve() + recovered = 0 + for profile_name, profile_home in _multiplex_profile_homes(self.config): + if Path(profile_home).resolve() == launch_home: + continue + try: + with _profile_runtime_scope(Path(profile_home), {}): + recovered += process_registry.recover_from_checkpoint() + except Exception: + logger.warning("Process checkpoint recovery for profile %r failed", profile_name, exc_info=True) + return recovered + async def _start_recover_previous_run(self) -> None: """Plugins, relay, hooks, then crash/clean-exit recovery of processes and sessions.""" from gateway.run import _hermes_home self._start_register_plugins_relay_hooks() self.hooks.discover_and_load() - # Recover background processes from checkpoint (crash recovery) + # Recover background processes from checkpoint (crash recovery). ``_checkpoint_path`` is + # scope-relative, so a served secondary's turn wrote ITS home's processes.json; recover each + # served profile's file under its scope or those processes are never re-adopted. with _log_suppressed(logging.WARNING, "Process checkpoint recovery: %s"): from tools.process_registry import process_registry recovered = process_registry.recover_from_checkpoint() + recovered += self._recover_secondary_process_checkpoints(process_registry) if recovered: logger.info("Recovered %s background process(es) from previous run", recovered) # Recover sessions active at last exit (exact turn markers + 120s recency fallback for diff --git a/hermes_cli/kanban_db_dispatch.py b/hermes_cli/kanban_db_dispatch.py index 5876caaa16..407ab91f24 100644 --- a/hermes_cli/kanban_db_dispatch.py +++ b/hermes_cli/kanban_db_dispatch.py @@ -2037,15 +2037,24 @@ def _resolve_worker_cli_toolsets(hermes_home: Optional[str]) -> Optional[list[st if not hermes_home: return None try: + from agent.secret_scope import ( + build_profile_secret_scope, is_multiplex_active, reset_secret_scope, set_secret_scope) from hermes_constants import reset_hermes_home_override, set_hermes_home_override from hermes_cli.config import load_config from hermes_cli.tools_config import _get_platform_tools token = set_hermes_home_override(hermes_home) + # Toolset availability probes read credentials (``get_secret``); under multiplex an + # unscoped read raises and the pin was silently dropped for every worker. + secret_token = ( + set_secret_scope(build_profile_secret_scope(Path(hermes_home))) + if is_multiplex_active() else None) try: cfg = load_config() toolsets = sorted(_get_platform_tools(cfg, "cli")) finally: + if secret_token is not None: + reset_secret_scope(secret_token) reset_hermes_home_override(token) return toolsets or None except Exception as exc: @@ -2177,7 +2186,7 @@ def _default_spawn(task: Task, workspace: str, *, board: Optional[str] = None) - profile_arg = normalize_profile_name(task.assignee) from agent.secret_scope import is_multiplex_active - from tools.environments.local import build_subprocess_env + from tools.environments.local import build_subprocess_env, strip_launch_profile_env env = build_subprocess_env( scrub_secrets=is_multiplex_active(), @@ -2195,6 +2204,9 @@ def _default_spawn(task: Task, workspace: str, *, board: Optional[str] = None) - # hermes_constants is imported. try: env["HERMES_HOME"] = resolve_profile_env(profile_arg) + # A multiplexer dispatching for another profile must not hand it the launch + # profile's .env settings / TERMINAL_* policy — a standalone dispatcher never would. + strip_launch_profile_env(env, env["HERMES_HOME"]) except FileNotFoundError: # No profile dir (isolated test fixtures) — the CLI resolves it from # HERMES_PROFILE (set below) instead. diff --git a/tests/cron/test_cron_multiplex_served_profile_parity.py b/tests/cron/test_cron_multiplex_served_profile_parity.py new file mode 100644 index 0000000000..0f79125f0a --- /dev/null +++ b/tests/cron/test_cron_multiplex_served_profile_parity.py @@ -0,0 +1,104 @@ +"""Standalone-vs-served parity for the cron subsystem under ``gateway.multiplex_profiles``. + +A profile served by the default multiplexer runs its ticks inside the default profile's process: +``os.environ`` holds the DEFAULT profile's ``.env`` and the served profile's values exist only +in its secret scope / home override. Every knob cron reads from ``.env`` and every child env it +builds must resolve exactly as it would under a standalone ``hermes -p gateway run``. +""" + +from pathlib import Path + +import pytest + +from agent.secret_scope import ( + build_profile_secret_scope, reset_secret_scope, set_multiplex_active, set_secret_scope, +) +from hermes_constants import reset_hermes_home_override, set_hermes_home_override + + +@pytest.fixture +def two_homes(tmp_path, monkeypatch): + root = tmp_path / "hermes" + alpha = root / "profiles" / "alpha" + for home in (root, alpha): + (home / "cron").mkdir(parents=True) + (root / ".env").write_text( + "HERMES_CRON_TIMEOUT=111\nHERMES_MODEL=default-model\nTERMINAL_ENV=docker\n" + "TERMINAL_DOCKER_IMAGE=default-only-image\nHERMES_LANGUAGE=en\n") + (alpha / ".env").write_text("HERMES_CRON_TIMEOUT=222\nHERMES_LANGUAGE=zh\n") + # The launch (default) profile's .env is what the multiplexer process loaded into os.environ. + monkeypatch.setenv("HERMES_HOME", str(root)) + for key, val in (("HERMES_CRON_TIMEOUT", "111"), ("HERMES_MODEL", "default-model"), + ("TERMINAL_ENV", "docker"), ("TERMINAL_DOCKER_IMAGE", "default-only-image"), + ("HERMES_LANGUAGE", "en")): + monkeypatch.setenv(key, val) + set_multiplex_active(True) + home_token = set_hermes_home_override(str(alpha)) + try: + yield root, alpha + finally: + reset_hermes_home_override(home_token) + set_multiplex_active(False) + + +def test_cron_env_settings_resolve_from_the_served_profile(two_homes): + """``HERMES_CRON_TIMEOUT`` / ``HERMES_MODEL`` are alpha's (or absent) both with the fire-time + secret scope installed and on the bare tick thread — never the default profile's environ.""" + import cron.scheduler as sched + from cron.jobs import _oneshot_run_claim_ttl_seconds + from cron.scheduler_preflight import _preflight_check_provider_key + + root, alpha = two_homes + captured = {} + + def fake_resolve(**kw): + captured.update(kw) + return {} + + import hermes_cli.runtime_provider as rp + original = rp.resolve_runtime_provider + rp.resolve_runtime_provider = fake_resolve + try: + # Tick thread: home override only (due-job scan, pool sizing, claim TTL). + assert sched._cron_inactivity_seconds() == 222.0 + assert _oneshot_run_claim_ttl_seconds() == 1800.0 # 222*3 < floor -> floor, from alpha's value + # Fire: secret scope installed as _run_one_job_body does. + token = set_secret_scope(build_profile_secret_scope(alpha)) + try: + assert sched._cron_inactivity_seconds() == 222.0 + _preflight_check_provider_key({"id": "j"}, {"cron": {}}) + assert captured["target_model"] == "" # alpha has no HERMES_MODEL; default's must not leak + with pytest.raises(RuntimeError, match="no model configured"): + sched._load_cron_job_config({"id": "j", "name": "j", "prompt": "x"}, "j", "j") + finally: + reset_secret_scope(token) + finally: + rp.resolve_runtime_provider = original + + +def test_child_env_for_served_profile_drops_launch_profile_settings(two_homes): + """A worker/bot-chat child spawned for served alpha must not inherit the default profile's + non-credential ``.env`` settings or bridged ``TERMINAL_*`` policy — a standalone alpha never + had them. Alpha's own home pin and non-profile env survive.""" + from tools.environments.local import build_subprocess_env, strip_launch_profile_env + + root, alpha = two_homes + token = set_secret_scope(build_profile_secret_scope(alpha)) + try: + env = strip_launch_profile_env(build_subprocess_env(scrub_secrets=True, inherit_profile_home=True)) + finally: + reset_secret_scope(token) + for key in ("HERMES_MODEL", "TERMINAL_ENV", "TERMINAL_DOCKER_IMAGE", "HERMES_LANGUAGE", "HERMES_CRON_TIMEOUT"): + assert key not in env, key + assert env["HERMES_HOME"] == str(alpha) + assert "PATH" in env + + # No-op when the scope IS the launch profile (standalone / default profile's own children). + reset_hermes_home_override(set_hermes_home_override(None)) + home_token = set_hermes_home_override(str(root)) + try: + env = strip_launch_profile_env(build_subprocess_env(scrub_secrets=True, inherit_profile_home=True)) + finally: + reset_hermes_home_override(home_token) + assert env["TERMINAL_ENV"] == "docker" + assert env["HERMES_MODEL"] == "default-model" diff --git a/tests/gateway/test_kanban_multiplex_served_profile_parity.py b/tests/gateway/test_kanban_multiplex_served_profile_parity.py new file mode 100644 index 0000000000..928e84ceb0 --- /dev/null +++ b/tests/gateway/test_kanban_multiplex_served_profile_parity.py @@ -0,0 +1,124 @@ +"""Standalone-vs-served parity for the kanban dispatcher and notifier under +``gateway.multiplex_profiles``: a worker spawned for served profile X gets the env a standalone +``hermes -p X`` dispatcher would build, and X's notifications are rendered/filtered under X's +config. +""" + +import asyncio +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from agent.secret_scope import set_multiplex_active +from gateway.config import Platform +from gateway.run import GatewayRunner +from hermes_cli import kanban_db as kb +from hermes_cli import kanban_db_connect as kbc +from hermes_cli import kanban_db_dispatch as kbd +from hermes_cli import kanban_db_notify as kbn +from hermes_constants import get_hermes_home + + +@pytest.fixture +def served(tmp_path, monkeypatch): + root = tmp_path / ".hermes" + alpha = root / "profiles" / "alpha" + alpha.mkdir(parents=True) + (root / ".env").write_text("HERMES_MODEL=default-model\nTERMINAL_ENV=docker\n") + (root / "config.yaml").write_text("gateway:\n multiplex_profiles: true\nterminal:\n backend: docker\n") + (alpha / ".env").write_text("") + (alpha / "config.yaml").write_text("display:\n language: zh\n") + monkeypatch.setenv("HERMES_HOME", str(root)) + monkeypatch.setenv("HERMES_MODEL", "default-model") + monkeypatch.setenv("TERMINAL_ENV", "docker") + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "board.db")) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setattr("hermes_constants.get_default_hermes_root", lambda: root) + set_multiplex_active(True) + try: + yield SimpleNamespace(root=root, alpha=alpha) + finally: + set_multiplex_active(False) + + +def test_worker_for_served_profile_gets_its_own_env_and_toolset_pin(served, monkeypatch): + """The dispatcher (root context) spawns alpha's worker: no launch-profile settings leak into the + child and the ``--toolsets`` pin (whose probes read credentials) is resolved under alpha's scope.""" + kb.init_db() + conn = kbc.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alpha") + task = kb.get_task(conn, tid) + finally: + conn.close() + + spawned = {} + + def fake_popen(argv, **kwargs): + spawned["argv"], spawned["env"] = argv, kwargs["env"] + return SimpleNamespace(pid=4242) + + monkeypatch.setattr(subprocess, "Popen", fake_popen) + monkeypatch.setattr(kbd, "_open_worker_log", lambda task, board: open("/dev/null", "w")) + monkeypatch.setattr(kbd, "_hermes_argv", lambda: ["hermes"], raising=False) + kbd._default_spawn(task, str(served.alpha), board=None) + + env = spawned["env"] + assert env["HERMES_HOME"] == str(served.alpha) + assert "HERMES_MODEL" not in env and "TERMINAL_ENV" not in env + assert "--toolsets" in spawned["argv"] + + +class RecordingAdapter: + def __init__(self): + self.sent = [] + + async def send(self, chat_id, text, metadata=None): + from gateway.media_policy import media_delivery_strict + self.sent.append({"text": text, "home": str(get_hermes_home()), "strict": media_delivery_strict()}) + return SimpleNamespace(success=True, error=None) + + async def handle_message(self, event): + event._gateway_accepted = True + + +def test_notifier_pings_run_under_the_subscribers_profile(served, monkeypatch): + """A subscription owned by served alpha is pinged with alpha's home active, so its media policy + and display language apply — alpha's own adapter was already selected before this fix.""" + (served.alpha / "config.yaml").write_text("display:\n language: zh\ngateway:\n strict: true\n") + kb.init_db() + conn = kbc.connect() + try: + tid = kb.create_task(conn, title="notify parity", assignee="alpha") + kbn.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="1001", notifier_profile="alpha") + kb.complete_task(conn, tid, summary="done") + finally: + conn.close() + + runner = GatewayRunner.__new__(GatewayRunner) + runner._running = True + runner.config = SimpleNamespace(multiplex_profiles=True, profile_routes=[]) + alpha_adapter = RecordingAdapter() + runner.adapters = {Platform.TELEGRAM: RecordingAdapter()} + runner._profile_adapters = {"alpha": {Platform.TELEGRAM: alpha_adapter}} + runner._primary_profile_name = "default" + runner._kanban_sub_fail_counts = {} + runner._kanban_dispatcher_lock_handle = object() + runner._profile_failed_platforms = {} + + real_sleep = asyncio.sleep + + async def fake_sleep(delay): + if delay == 5: + return None + runner._running = False + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + asyncio.run(runner._kanban_notifier_watcher(interval=1)) + + assert len(alpha_adapter.sent) == 1 + assert alpha_adapter.sent[0]["home"] == str(served.alpha) + assert alpha_adapter.sent[0]["strict"] is True diff --git a/tests/gateway/test_multiplex_served_profile_parity.py b/tests/gateway/test_multiplex_served_profile_parity.py new file mode 100644 index 0000000000..f0c833c838 --- /dev/null +++ b/tests/gateway/test_multiplex_served_profile_parity.py @@ -0,0 +1,163 @@ +"""Standalone-vs-served parity for goals/loops, completion notices and process recovery under +``gateway.multiplex_profiles``: every read that decides FOR a served profile must run inside that +profile's runtime scope, and every store the served profile wrote must be recovered under it. +""" + +import asyncio +import json +import threading +from collections import OrderedDict +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from agent.secret_scope import set_multiplex_active +from gateway.config import Platform, PlatformConfig +from gateway.run import GatewayRunner, _profile_runtime_scope +from gateway.session import SessionSource +from hermes_constants import get_hermes_home + + +class RecordingAdapter: + def __init__(self, notice_delivery=None): + self.config = PlatformConfig(enabled=True, extra={"notice_delivery": notice_delivery} if notice_delivery else {}) + self.platform = Platform.TELEGRAM + self.calls = [] + self._active_sessions, self._pending_messages, self._session_tasks = {}, {}, {} + + async def send(self, chat_id, content, metadata=None, **kw): + self.calls.append(("send", str(get_hermes_home()))) + return SimpleNamespace(success=True, error=None) + + async def send_private_notice(self, chat_id, user_id, content, metadata=None, **kw): + self.calls.append(("send_private_notice", str(get_hermes_home()))) + return SimpleNamespace(success=True, error=None) + + async def handle_message(self, event): + self.calls.append(("handle_message", str(get_hermes_home()))) + event._gateway_accepted = True + + +@pytest.fixture +def served(tmp_path, monkeypatch): + """Default host + served profile ``alpha``; the runner is the default multiplexer.""" + root = tmp_path / "hermes" + alpha = root / "profiles" / "alpha" + alpha.mkdir(parents=True) + (root / "config.yaml").write_text( + "gateway:\n multiplex_profiles: true\ndisplay:\n background_process_notifications: concise\n") + (alpha / "config.yaml").write_text("display:\n background_process_notifications: 'off'\n") + (root / ".env").write_text("") + (alpha / ".env").write_text("") + monkeypatch.setenv("HERMES_HOME", str(root)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setattr("hermes_constants.get_default_hermes_root", lambda: root) + set_multiplex_active(True) + + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = SimpleNamespace(multiplex_profiles=True, profile_routes=[], get_notice_delivery=lambda p: "public") + runner._running = True + default_adapter, alpha_adapter = RecordingAdapter(), RecordingAdapter(notice_delivery="private") + runner.adapters = {Platform.TELEGRAM: default_adapter} + runner._profile_adapters = {"alpha": {Platform.TELEGRAM: alpha_adapter}} + runner._primary_profile_name = "default" + runner._session_source_cache = {} + runner.session_store = SimpleNamespace(_ensure_loaded=lambda: None, _entries={}) + runner._completion_delivery_lock = threading.Lock() + runner._completion_deliveries_inflight = set() + runner._completion_deliveries_delivered = OrderedDict() + runner._completion_delivery_retention = 2048 + runner._background_tasks = set() + runner._profile_failed_platforms = {} + try: + yield SimpleNamespace(root=root, alpha=alpha, runner=runner, alpha_adapter=alpha_adapter, + default_adapter=default_adapter) + finally: + set_multiplex_active(False) + + +def _alpha_source(): + return SessionSource(platform=Platform.TELEGRAM, chat_id="1001", chat_type="dm", user_id="u1", profile="alpha") + + +def test_platform_notice_honours_the_served_profiles_notice_delivery(served): + """alpha's ``platforms.telegram.notice_delivery: private`` (carried by ITS adapter) wins over the + launch profile's GatewayConfig — as a standalone alpha gateway would behave.""" + runner = served.runner + runner._thread_metadata_for_source = lambda source: None + with _profile_runtime_scope(served.alpha): + asyncio.run(runner._deliver_platform_notice(_alpha_source(), "notice")) + assert [op for op, _ in served.alpha_adapter.calls] == ["send_private_notice"] + assert served.default_adapter.calls == [] + + +def test_loop_completion_persists_into_the_served_profiles_store(served): + """The post-turn /loop completion hop carries the profile contextvars: the completed tick lands + in alpha's state.db, not the default profile's.""" + from hermes_cli.goals import _get_session_db + from hermes_cli.loops import LoopManager + + runner = served.runner + entry = SimpleNamespace(session_id="sess-alpha-1", session_key="agent:alpha:telegram:dm:1001") + with _profile_runtime_scope(served.alpha): + _get_session_db() + mgr = LoopManager(session_id=entry.session_id) + mgr.set("check", interval_seconds=300, route={"platform": "telegram", "chat_id": "1001", "profile": "alpha"}) + assert mgr.fire_tick() + asyncio.run(runner._post_turn_loop_completion( + session_entry=entry, source=_alpha_source(), final_response="done")) + + def loop_row(home): + import sqlite3 + db = home / "state.db" + if not db.exists(): + return None + con = sqlite3.connect(db) + try: + row = con.execute("SELECT value FROM state_meta WHERE key=?", (f"loop:{entry.session_id}",)).fetchone() + finally: + con.close() + return json.loads(row[0]) if row else None + + assert loop_row(served.alpha)["awaiting_response"] is False + assert loop_row(served.root) is None + + +def test_watch_event_gate_uses_the_owning_profiles_mode(served): + """alpha has notifications ``off``; its watch event is drained silently even when the shared + queue is drained from the root context, exactly as a standalone alpha gateway would.""" + from tools.process_registry import ProcessRegistry + + runner = served.runner + registry = ProcessRegistry() + evt = {"type": "watch_match", "session_id": "p1", "session_key": "agent:alpha:telegram:dm:1001", + "platform": "telegram", "chat_type": "dm", "chat_id": "1001", "pattern": "DONE", + "output": "DONE", "command": "x"} + registry.completion_queue.put(evt) + asyncio.run(runner._drain_watch_notifications(registry.completion_queue)) + assert registry.completion_queue.qsize() == 0 + assert served.alpha_adapter.calls == [] + assert served.default_adapter.calls == [] + + +def test_served_profile_process_checkpoint_is_recovered_at_startup(served, monkeypatch): + """A background process checkpointed during alpha's turn (alpha/processes.json) is re-adopted by + the multiplexer's startup recovery, once, with its watcher re-armed.""" + from tools.process_registry import ProcessRegistry + + runner = served.runner + import os + entry = {"session_id": "proc_alpha01", "pid": os.getpid(), "pid_scope": "host", "command": "sleep 1", + "started_at": 1.0, "watcher_interval": 5, "notify_on_complete": True, + "session_key": "agent:alpha:telegram:dm:1001"} + (served.alpha / "processes.json").write_text(json.dumps([entry])) + registry = ProcessRegistry() + monkeypatch.setattr(registry, "_host_pid_is_ours", lambda pid, start: True) + + recovered = registry.recover_from_checkpoint() + recovered += runner._recover_secondary_process_checkpoints(registry) + assert recovered == 1 + assert [w["session_id"] for w in registry.pending_watchers] == ["proc_alpha01"] + # Idempotent across homes: the process-global registry already tracks it. + assert runner._recover_secondary_process_checkpoints(registry) == 0 diff --git a/tools/environments/local.py b/tools/environments/local.py index ecc9d6f3ba..b9ffb48aa5 100644 --- a/tools/environments/local.py +++ b/tools/environments/local.py @@ -337,6 +337,29 @@ def build_subprocess_env( return delegated_child_subprocess_env(env) +def strip_launch_profile_env(env: dict, target_home: "str | Path | None" = None) -> dict: + """Drop the LAUNCH profile's residue from a child env built for another served profile. + ``os.environ`` holds the default profile's ``.env`` and its bridged ``TERMINAL_*`` settings; + the secret scrub removes credentials but not settings (``HERMES_MODEL``, ``TERMINAL_ENV``, + ``HERMES_LANGUAGE``...), so a standalone ``hermes -p X`` worker and a served one saw different + envs. The child re-loads X's own ``.env`` and bridges X's config itself. ``target_home`` + defaults to the active home override; no-op outside multiplex or when the target IS the + launch profile.""" + from agent.secret_scope import _is_global_env, is_multiplex_active, load_env_file + from hermes_constants import get_hermes_home_override, get_process_hermes_home + target = target_home or get_hermes_home_override() + if not is_multiplex_active() or not target: + return env + launch_home = get_process_hermes_home() + if Path(target).resolve() == launch_home.resolve(): + return env + from hermes_cli.config import TERMINAL_CONFIG_ENV_MAP + for key in set(load_env_file(launch_home / ".env")) | set(TERMINAL_CONFIG_ENV_MAP.values()): + if not _is_global_env(key) or key.startswith("TERMINAL_"): + env.pop(key, None) + return env + + # --- Shell discovery --- def _windows_bash_candidates(custom: "str | None") -> list[str]: """Ordered bash.exe candidates on Windows: HERMES_GIT_BASH_PATH, our portable Git diff --git a/tools/process_registry_checkpoint.py b/tools/process_registry_checkpoint.py index 74c9f22fff..a12dfe749c 100644 --- a/tools/process_registry_checkpoint.py +++ b/tools/process_registry_checkpoint.py @@ -64,6 +64,12 @@ class ProcessCheckpointMixin: pid, pid_scope = entry.get("pid"), entry.get("pid_scope", "host") if not pid: continue + # The registry is process-global, so every profile's checkpoint carries every live + # process; a multiplexer recovering several homes must adopt each session once. + with self._lock: + already_tracked = entry.get("session_id") in self._running + if already_tracked: + continue if pid_scope != "host": # in-sandbox PIDs mean nothing once the env handle is gone logger.info( "Skipping recovery for non-host process: %s (pid=%s, scope=%s)", diff --git a/website/docs/user-guide/multi-profile-gateways.md b/website/docs/user-guide/multi-profile-gateways.md index 0d4bdad508..251fdebba4 100644 --- a/website/docs/user-guide/multi-profile-gateways.md +++ b/website/docs/user-guide/multi-profile-gateways.md @@ -356,6 +356,9 @@ profile and never shares with the default or any sibling: | Platform proxies (`TELEGRAM_PROXY`, `DISCORD_PROXY`, `HTTPS_PROXY`, …) | The profile's own `.env` | Direct connection — never the default profile's proxy | | MCP discovery in the Desktop/dashboard backend | Once per served profile home | A profile selected after another has already built an agent still discovers its own `mcp_servers` | | Dashboard actions (`hermes -p …` spawned by the Desktop/dashboard) | A scrubbed child env pinned to that profile's `HERMES_HOME` | The child loads its own `.env`; the dashboard profile's tokens and ports are not inherited | +| Cron `.env` tuning (`HERMES_CRON_TIMEOUT`, `HERMES_MODEL` fallback, `HERMES_CRON_MAX_PARALLEL`, prefill file), worker / Bot Chat child env | The profile's own `.env`; children never inherit the default profile's `.env` settings or bridged `TERMINAL_*` policy | Cron defaults / model refusal, exactly as a standalone `hermes -p gateway run` | +| Kanban workers and notifications for a profile's tasks | The assignee's `.env` + `config.yaml` (toolset pin, terminal backend, media policy, display language) | — | +| `/loop` ticks, `background_process_notifications` gate, `notice_delivery`, background-process checkpoint recovery | The owning profile's `state.db` / `config.yaml` / `processes.json` | — | What is **shared** by design: the process, its PID/lock and `gateway_state.json` (default home), the one HTTP listener, and the `profile_routes` table (declared