fix(multiplex): cron, kanban, /loop and completion paths for a served profile match its standalone gateway

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
  adf23550f5) 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

@@ -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)

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):

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 <name>` 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):

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.

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:

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.<config_key>`` 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)

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
 

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:

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)

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.<p>`` 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

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

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.

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 <name> 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"

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

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

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

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)",

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 <name> …` 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 <name> 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