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.
This commit is contained in:
Teknium
2026-09-11 19:19:44 -07:00
parent fc71fb63e5
commit 284d220ba4
18 changed files with 565 additions and 42 deletions
+26
View File
@@ -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)
+2 -1
View File
@@ -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):
+12 -10
View File
@@ -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):
+2 -1
View File
@@ -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.
+3 -1
View File
@@ -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:
+2 -1
View File
@@ -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)
+18 -4
View File
@@ -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
+6 -2
View File
@@ -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:
+4 -4
View File
@@ -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)
+30 -16
View File
@@ -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
+24 -1
View File
@@ -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
+13 -1
View File
@@ -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.
@@ -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"
@@ -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
@@ -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
+23
View File
@@ -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
+6
View File
@@ -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)",
@@ -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