fix(gateway,tools): multiplexed profiles keep their own max_turns, fallback chain, hooks, auth, caches and paths (salvage #56315 #56508 #63962, fixes #95685)

agent/auxiliary_client.py

@@ -910,6 +910,15 @@ def _nous_extra_body() -> dict:
 _NOUS_DEFAULT_BASE_URL = "https://inference-api.nousresearch.com/v1"
 _ANTHROPIC_DEFAULT_BASE_URL = "https://api.anthropic.com"
 _AUTH_JSON_PATH = get_hermes_home() / "auth.json"
+_AUTH_JSON_PATH_AT_IMPORT = _AUTH_JSON_PATH
+
+
+def _auth_json_path():
+    """Active profile's ``auth.json`` at call time (a patched ``_AUTH_JSON_PATH`` still wins). The
+    import-time constant is the LAUNCH profile's; under multiplexing a secondary's auxiliary calls
+    would otherwise authenticate to Nous with the default profile's token."""
+    from hermes_cli.auth import _auth_file_path
+    return _AUTH_JSON_PATH if _AUTH_JSON_PATH != _AUTH_JSON_PATH_AT_IMPORT else _auth_file_path()
 
 # Hosts exposing BOTH ``…/anthropic`` and a sibling OpenAI ``…/v1``. Matched on the URL *host*
 # only: unconditional rewrites break Anthropic-only gateways.
@@ -1859,9 +1868,10 @@ def _read_nous_auth() -> Optional[dict]:
             "source": "pool",
         }
     try:
-        if not _AUTH_JSON_PATH.is_file():
+        auth_path = _auth_json_path()
+        if not auth_path.is_file():
             return None
-        data = json.loads(_AUTH_JSON_PATH.read_text(encoding="utf-8-sig"))
+        data = json.loads(auth_path.read_text(encoding="utf-8-sig"))
         if data.get("active_provider") != "nous":
             return None
         provider = data.get("providers", {}).get("nous", {})

agent/shell_hooks.py

@@ -83,11 +83,14 @@ def _payload_fields(kwargs: Dict[str, Any]) -> Dict[str, Any]:
         cwd = str(Path.cwd())
     except OSError:
         cwd = ""
+    from hermes_cli.profiles import get_active_profile_name
     return {
         "tool_name": kwargs.get("tool_name"),
         "tool_input": kwargs.get("args") if isinstance(kwargs.get("args"), dict) else None,
         "session_id": kwargs.get("session_id") or kwargs.get("parent_session_id") or "",
         "cwd": cwd,
+        # Resolved at fire time: a multiplexed gateway's hook script must know which profile fired it.
+        "profile": get_active_profile_name(),
         "extra": {k: v for k, v in kwargs.items() if k not in _TOP_LEVEL_PAYLOAD_KEYS},
     }
 
@@ -301,11 +304,15 @@ def failed(error: str) -> Dict[str, Any]:
     # Own process group on POSIX so a timed-out hook's descendants are reaped with it (Windows: kill_process_tree
     # / taskkill /T). Hooks that finish in time keep detached helpers alive.
     popen_kwargs: Dict[str, Any] = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {"process_group": 0}
-    from agent.delegation_context import delegated_child_subprocess_env
+    # HERMES_HOME follows the routed profile (the import-time environ holds the launch profile's), and
+    # under multiplexing os.environ carries the DEFAULT profile's secrets, which a secondary's hook
+    # script must not inherit; single-profile runs keep the process env byte-for-byte as before.
+    from agent.secret_scope import is_multiplex_active
+    from tools.environments.local import build_subprocess_env
     try:
         proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                 text=True, encoding='utf-8', errors='replace', shell=False,
-                                env=delegated_child_subprocess_env(), **popen_kwargs)
+                                env=build_subprocess_env(scrub_secrets=is_multiplex_active()), **popen_kwargs)
     except Exception as exc:
         return failed(next((msg for cls, msg in _POPEN_ERRORS if isinstance(exc, cls)), str(exc)))
     try:

gateway/hooks.py

@@ -12,15 +12,28 @@
 import asyncio
 import importlib.util
 import sys
+import threading
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional
 
 import yaml
 
 from hermes_cli.config import get_hermes_home
+from hermes_constants import hermes_home_key
 
 
 HOOKS_DIR = get_hermes_home() / "hooks"
+_HOOKS_DIR_AT_IMPORT = HOOKS_DIR
+
+
+def _resolve_hooks_dir() -> Path:
+    """Active profile's hooks dir at call time: the patched ``HOOKS_DIR`` when a test changed it,
+    else ``get_hermes_home()/hooks``. The import-time constant is the LAUNCH profile's; under
+    ``gateway.multiplex_profiles`` every served profile has its own ``hooks/``, and a registry
+    loaded from the launch home would run the default profile's handlers (arbitrary Python) on
+    every other profile's messages, responses and user ids."""
+    configured = Path(HOOKS_DIR)
+    return configured if configured != _HOOKS_DIR_AT_IMPORT else get_hermes_home() / "hooks"
 
 
 def _skip(name: str, reason: str) -> None:
@@ -74,11 +87,12 @@ def _register_builtin_hooks(self) -> None:
         """Extension point for always-on built-in hooks; currently none shipped."""
 
     def discover_and_load(self) -> None:
-        """Register built-in hooks, then load every valid hook dir under HOOKS_DIR."""
+        """Register built-in hooks, then load every valid hook dir under the active profile's ``hooks/``."""
         self._register_builtin_hooks()
-        if not HOOKS_DIR.exists():
+        hooks_dir = _resolve_hooks_dir()
+        if not hooks_dir.exists():
             return
-        for hook_dir in sorted(HOOKS_DIR.iterdir()):
+        for hook_dir in sorted(hooks_dir.iterdir()):
             if not hook_dir.is_dir():
                 continue
             try:
@@ -123,3 +137,45 @@ async def emit_collect(self, event_type: str, context: Optional[Dict[str, Any]]
             except Exception as e:
                 print(f"[hooks] Error in handler for '{event_type}': {e}", flush=True)
         return results
+
+
+class ProfileHookRegistries:
+    """``HookRegistry`` per served profile home, picked at emit time from the active HERMES_HOME.
+
+    The gateway holds ONE of these. Every hook emit already runs inside the routed profile's
+    ``_profile_runtime_scope`` (message handlers, /new, turn wiring), so resolving the registry by
+    ``get_hermes_home()`` there gives each profile its own ``hooks/`` and keeps the default
+    profile's handlers from seeing other profiles' messages. Each home's registry is loaded on its
+    first emit, i.e. inside that profile's scope (handler imports see its HERMES_HOME); multiplexing off
+    means a single entry for the launch home, i.e. exactly the old behaviour.
+    """
+
+    def __init__(self):
+        self._by_home: Dict[str, HookRegistry] = {}
+        self._lock = threading.Lock()
+
+    def _active(self) -> HookRegistry:
+        key = hermes_home_key(get_hermes_home())
+        registry = self._by_home.get(key)
+        if registry is None:
+            with self._lock:
+                registry = self._by_home.get(key)
+                if registry is None:
+                    registry = HookRegistry()
+                    registry.discover_and_load()
+                    self._by_home[key] = registry
+        return registry
+
+    @property
+    def loaded_hooks(self) -> List[dict]:
+        return self._active().loaded_hooks
+
+    def discover_and_load(self) -> None:
+        """Load the active home's hooks now (startup, or a secondary profile's scoped startup)."""
+        self._active()
+
+    async def emit(self, event_type: str, context: Optional[Dict[str, Any]] = None) -> None:
+        await self._active().emit(event_type, context)
+
+    async def emit_collect(self, event_type: str, context: Optional[Dict[str, Any]] = None) -> List[Any]:
+        return await self._active().emit_collect(event_type, context)

gateway/media_fetch.py

@@ -26,8 +26,7 @@
 from typing import Optional
 
 from gateway.platforms.base import (
-    _MEDIA_DELIVERY_DENIED_HOME_SUBPATHS, _MEDIA_DELIVERY_DENIED_PREFIXES, _ROOT_CREDENTIAL_PATHS,
-    _TRUTHY, MEDIA_DELIVERY_STRICT_ENV)
+    _MEDIA_DELIVERY_DENIED_HOME_SUBPATHS, _MEDIA_DELIVERY_DENIED_PREFIXES, _ROOT_CREDENTIAL_PATHS)
 
 logger = logging.getLogger(__name__)
 
@@ -79,7 +78,8 @@ def _active_remote_env():
 def fetch_remote_media(path: str) -> Optional[str]:
     """Host path of a validated copy of sandbox file ``path``, or None (never raises). Only fires
     when a remote backend is active; the caller has already failed local validation."""
-    if os.environ.get(MEDIA_DELIVERY_STRICT_ENV, "0").strip().lower() in _TRUTHY:
+    from gateway.media_policy import media_delivery_strict
+    if media_delivery_strict():
         return None
     env = _active_remote_env()
     if env is None:

gateway/media_policy.py

@@ -19,6 +19,55 @@
 
 _FLAG_ENVS = (("strict", "HERMES_MEDIA_DELIVERY_STRICT"), ("trust_recent_files", "HERMES_MEDIA_TRUST_RECENT_FILES"))
 _ALLOW_DIRS_ENV = "HERMES_MEDIA_ALLOW_DIRS"
+_TRUST_RECENT_SECONDS_ENV = "HERMES_MEDIA_TRUST_RECENT_SECONDS"
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+
+
+def _routed_gateway_cfg() -> Optional[Dict[str, Any]]:
+    """``gateway`` section of the ROUTED profile's config when a HERMES_HOME override is active
+    (multiplexed turn), else None. The env bridge is one process-wide copy of the launch profile's
+    policy, so a secondary's deliveries must read their own config instead of ``os.environ``."""
+    from hermes_constants import get_hermes_home_override
+    if not get_hermes_home_override():
+        return None
+    try:
+        from hermes_cli.config import load_config_readonly
+        gateway_cfg = load_config_readonly().get("gateway")
+    except Exception:
+        return {}
+    return gateway_cfg if isinstance(gateway_cfg, dict) else {}
+
+
+def media_delivery_strict() -> bool:
+    cfg = _routed_gateway_cfg()
+    if cfg is not None:
+        return bool(cfg.get("strict", False))
+    return os.environ.get(_FLAG_ENVS[0][1], "0").strip().lower() in _TRUTHY
+
+
+def media_delivery_allow_dirs() -> str:
+    """Operator allowlist as the ``os.pathsep``-joined string the validator splits."""
+    cfg = _routed_gateway_cfg()
+    if cfg is not None:
+        return _allow_dirs_str(cfg.get("media_delivery_allow_dirs"))
+    return os.environ.get(_ALLOW_DIRS_ENV, "")
+
+
+def media_delivery_trust_recent() -> bool:
+    cfg = _routed_gateway_cfg()
+    if cfg is not None:
+        return bool(cfg.get("trust_recent_files", True))
+    return os.environ.get(_FLAG_ENVS[1][1], "1").strip().lower() not in ("0", "false", "no", "off", "")
+
+
+def media_delivery_trust_recent_seconds() -> str:
+    """Raw recency window (``""`` = validator default); the caller parses/floors it."""
+    cfg = _routed_gateway_cfg()
+    if cfg is not None:
+        raw = cfg.get("trust_recent_files_seconds")
+        return "" if raw is None else str(raw)
+    return os.environ.get(_TRUST_RECENT_SECONDS_ENV, "")
+
 
 
 def _load_gateway_cfg(config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:

gateway/platforms/base.py

@@ -844,8 +844,9 @@ def _kanban_attachment_roots() -> List[Path]:
 
 def _media_delivery_allowed_roots() -> List[Path]:
     """Return roots from which model-emitted local media may be delivered."""
+    from gateway.media_policy import media_delivery_allow_dirs
     operator_roots = (
-        root for chunk in os.environ.get(MEDIA_DELIVERY_ALLOW_DIRS_ENV, "").split(os.pathsep)
+        root for chunk in media_delivery_allow_dirs().split(os.pathsep)
         for raw_root in chunk.split(",")
         if (root := Path(os.path.expanduser(raw_root.strip()))).is_absolute())
     return [*map(Path, MEDIA_DELIVERY_SAFE_ROOTS), *_profile_cache_roots(),
@@ -854,10 +855,10 @@ def _media_delivery_allowed_roots() -> List[Path]:
 
 def _media_delivery_recency_seconds() -> float:
     """Recency window (seconds) for trusting fresh files; 0 = pure-allowlist mode."""
-    raw = os.environ.get(MEDIA_DELIVERY_TRUST_RECENT_ENV, "1").strip().lower()
-    if raw in ("0", "false", "no", "off", ""):
+    from gateway.media_policy import media_delivery_trust_recent, media_delivery_trust_recent_seconds
+    if not media_delivery_trust_recent():
         return 0.0
-    custom = os.environ.get(MEDIA_DELIVERY_TRUST_RECENT_SECONDS_ENV, "").strip()
+    custom = media_delivery_trust_recent_seconds().strip()
     default = float(_MEDIA_DELIVERY_TRUST_RECENT_DEFAULT_SECONDS)
     return _or_default(lambda: max(0.0, float(custom)) if custom else default, default)
 
@@ -1125,7 +1126,8 @@ def validate_media_delivery_path(path: str, session_key: str = "") -> Optional[s
         if resolved_root is not None and _path_is_within(resolved, resolved_root):
             return str(resolved)
     # Non-strict (default): anything not denylisted (/etc, /proc, ~/.ssh, Hermes-root secrets).
-    if os.environ.get(MEDIA_DELIVERY_STRICT_ENV, "0").strip().lower() not in _TRUTHY:
+    from gateway.media_policy import media_delivery_strict
+    if not media_delivery_strict():
         return None if _path_under_denied_prefix(resolved) else str(resolved)
     # Strict: recency trust for fresh files (pandoc -o /tmp/x.pdf); denylist still applies.
     window = _media_delivery_recency_seconds()

gateway/run.py

@@ -1620,9 +1620,20 @@ def _bridge_max_turns_from_config(home: "Path") -> None:
 def _current_max_iterations() -> int:
     """Return the per-turn iteration budget after runtime env refresh; ``resolve_turn_limit`` maps
     ``agent.max_turns: none``/``unlimited`` (bridged as a string) to the unlimited sentinel, not an
-    ``int()`` crash."""
+    ``int()`` crash. A routed profile (HERMES_HOME override, multiplexed turns) reads ITS
+    ``agent.max_turns`` straight from config: the ``HERMES_MAX_ITERATIONS`` bridge is one process-wide
+    slot holding the launch profile's value, so every secondary would inherit the default's budget."""
     _reload_runtime_env_preserving_config_authority()
     from hermes_cli.config import resolve_turn_limit as _resolve_turn_limit
+    override = get_hermes_home_override()
+    if override:
+        config_path = Path(override) / 'config.yaml'
+        try:
+            cfg = _load_bridge_config(config_path) if config_path.exists() else {}
+        except Exception:
+            cfg = {}
+        agent_cfg = cfg.get("agent")
+        return _resolve_turn_limit(agent_cfg.get("max_turns") if isinstance(agent_cfg, dict) else None)
     return _resolve_turn_limit(os.getenv("HERMES_MAX_ITERATIONS"))
 
 
@@ -3671,10 +3682,11 @@ def _init_registries_and_clocks(self) -> None:
         # ``pairing_store``: global/default store (CLI, callers without profile context); ``pairing_stores``:
         # per-profile map ``authz_mixin._is_user_authorized`` routes through (one whitelist per profile).
         from gateway.pairing import PairingStore
-        from gateway.hooks import HookRegistry
+        from gateway.hooks import ProfileHookRegistries
         self.pairing_store = PairingStore()
         self.pairing_stores: Dict[str, "PairingStore"] = {}
-        self.hooks = HookRegistry()
+        # One HookRegistry per served profile home, resolved from the active scope at emit time.
+        self.hooks = ProfileHookRegistries()
         # Per-chat voice reply mode: "off" | "voice_only" | "all"
         self._voice_mode: Dict[str, str] = self._load_voice_modes()
         # Per-(guild,user) transcript dedup: the voice/STT pipeline can emit one utterance twice.

gateway/run_config_loaders.py

@@ -60,7 +60,7 @@ def _load_prefill_messages() -> List[Dict[str, Any]]:
         HERMES_PREFILL_MESSAGES_FILE env wins, then top-level prefill_messages_file in config.yaml,
         then legacy agent.prefill_messages_file. Relative paths resolve from ~/.hermes/.
         """
-        from gateway.run import _hermes_home, _load_gateway_runtime_config
+        from gateway.run import _gateway_config_home, _load_gateway_runtime_config
         file_path = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "")
         if not file_path:
             cfg = _load_gateway_runtime_config()
@@ -71,7 +71,7 @@ def _load_prefill_messages() -> List[Dict[str, Any]]:
             return []
         path = Path(file_path).expanduser()
         if not path.is_absolute():
-            path = _hermes_home / path
+            path = _gateway_config_home() / path
         if not path.exists():
             logger.warning("Prefill messages file not found: %s", path)
             return []
@@ -426,13 +426,23 @@ def _refresh_fallback_model(self) -> list | None:
         ``self._fallback_model`` at process start, so a chain configured (or changed) after ``hermes
         gateway`` was running never reached messaging sessions even though the same process's cron jobs fell
         back correctly. Fixes #60955.
+
+        Reads the ACTIVE gateway home (the routed profile under multiplexing, else the launch home)
+        and keeps one last-known-good chain per home: a single runner-wide slot filled from the launch
+        home handed every secondary profile the default profile's fallback chain.
         """
-        from gateway.run import _hermes_home
+        from gateway.run import _gateway_config_home
+        from hermes_constants import hermes_home_key
+        home = _gateway_config_home()
+        by_home = getattr(self, "_fallback_model_by_home", None)
+        if by_home is None:
+            by_home = self._fallback_model_by_home = {}
+        home_key = hermes_home_key(home)
         try:
             from hermes_cli.config import read_user_config_raw
-            cfg_path = _hermes_home / "config.yaml"
+            cfg_path = home / "config.yaml"
             if not cfg_path.exists():
-                self._fallback_model = None
+                by_home[home_key] = self._fallback_model = None
                 return self._fallback_model
             # Raw primitive (raises on parse failure) is required here: the canonical fail-open
             # loader would return {} on a torn mid-edit write and WIPE the last known-good chain.
@@ -448,8 +458,9 @@ def _refresh_fallback_model(self) -> list | None:
                     cfg = expanded
         except Exception:
             logger.debug("fallback_providers refresh: config.yaml read failed; keeping last known-good chain", exc_info=True)
+            self._fallback_model = by_home.get(home_key, self._fallback_model)
             return self._fallback_model
-        self._fallback_model = get_fallback_chain(cfg) or None
+        by_home[home_key] = self._fallback_model = get_fallback_chain(cfg) or None
         return self._fallback_model
 
     @staticmethod

gateway/sticker_cache.py

@@ -6,12 +6,21 @@
 
 import json
 import time
+from pathlib import Path
 from typing import Optional
 
 from hermes_cli.config import get_hermes_home
 from utils import atomic_json_write
 
 CACHE_PATH = get_hermes_home() / "sticker_cache.json"
+_CACHE_PATH_AT_IMPORT = CACHE_PATH
+
+
+def _resolve_cache_path() -> Path:
+    """Active profile's cache file at call time: the patched ``CACHE_PATH`` when a test changed
+    it, else live profile-scoped HERMES_HOME — under the multiplexed gateway one process serves
+    every profile, so the import-time constant would pin every profile to the launch home."""
+    return CACHE_PATH if CACHE_PATH != _CACHE_PATH_AT_IMPORT else get_hermes_home() / "sticker_cache.json"
 
 # Kept concise to save tokens.
 STICKER_VISION_PROMPT = (
@@ -22,13 +31,13 @@
 
 def _load_cache() -> dict:
     try:
-        return json.loads(CACHE_PATH.read_text(encoding="utf-8"))
+        return json.loads(_resolve_cache_path().read_text(encoding="utf-8"))
     except (FileNotFoundError, json.JSONDecodeError, OSError):
         return {}
 
 
 def _save_cache(cache: dict) -> None:
-    atomic_json_write(CACHE_PATH, cache)
+    atomic_json_write(_resolve_cache_path(), cache)
 
 
 def get_cached_description(file_unique_id: str) -> Optional[dict]: