Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions agent/secret_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,18 @@


def set_multiplex_active(active: bool) -> None:
"""Mark whether the process is a profile multiplexer (get_secret fails closed)."""
"""Mark whether the process is a profile multiplexer (get_secret fails closed).

When activating multiplex mode, pins the process home to the current ``HERMES_HOME`` value so
subsequent per-turn mirrors of other profiles into that env var cannot re-label the launch home
(first pin wins; an embedding host that calls :func:`hermes_constants.pin_process_hermes_home`
explicitly before this will keep its own pin).
"""
global _MULTIPLEX_ACTIVE
_MULTIPLEX_ACTIVE = bool(active)
if active:
import hermes_constants as _hc
_hc.pin_process_hermes_home(_hc.get_process_hermes_home())


def is_multiplex_active() -> bool:
Expand All @@ -42,7 +51,9 @@ def serves_routed_profile() -> bool:
"""True when the current task runs for a profile other than the process's own: always under
multiplexing, else when a HERMES_HOME override names another home (dashboard/desktop backend,
per-profile cron ticker). The MCP registry scope and the check_fn cache key both follow this
predicate so a served profile's view never aliases the launch profile's (#111151)."""
predicate so a served profile's view never aliases the launch profile's (#111151). A host that
mirrors the turn's profile into ``HERMES_HOME`` pins its own home with
``hermes_constants.pin_process_hermes_home`` so the mirror cannot flip this predicate."""
if is_multiplex_active():
return True
from hermes_constants import get_hermes_home_override, get_process_hermes_home, hermes_home_key
Expand Down
36 changes: 36 additions & 0 deletions hermes_constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -155,16 +155,52 @@ def reset_hermes_home_key_cache() -> None:
_HOME_KEY_CACHE.clear()


# Host-pinned identity of the profile this process serves as its own (None: follow HERMES_HOME).
# Set explicitly by an embedding host via pin_process_hermes_home(), or auto-set to the
# launch-time value by set_multiplex_active(True) (first pin wins).
_PINNED_PROCESS_HERMES_HOME: str | None = None


def pin_process_hermes_home(path: "str | Path | None") -> None:
"""Pin the home this process serves as its own for all process-level home decisions.

An embedding host that mirrors the active turn's profile into ``os.environ["HERMES_HOME"]``
for legacy readers must call this once at startup with the launch profile's home; after that
per-turn mutations to ``HERMES_HOME`` do not affect any of the four launch-home decisions
(``serves_routed_profile``, ``_is_routed_home``, ``_is_process_home``,
``env_loader._process_hermes_home``). First call wins; subsequent calls with the same path are
idempotent. ``None`` clears the pin. Hosts that never mutate ``HERMES_HOME`` need not call this.
"""
global _PINNED_PROCESS_HERMES_HOME
if path is None:
_PINNED_PROCESS_HERMES_HOME = None
return
if _PINNED_PROCESS_HERMES_HOME is None:
_PINNED_PROCESS_HERMES_HOME = str(path)


def get_process_hermes_home() -> Path:
"""Hermes home of the running process, ignoring task overrides.

Returns the pinned home when one has been set (see :func:`pin_process_hermes_home`); otherwise
reads ``HERMES_HOME`` live so standalone invocations (no multiplex, no explicit pin) keep
following the env var as before.

For process-level assets (theme YAML, dashboard plugin manifests) that must stay visible while a
request is scoped to another profile (e.g. embedded ``/chat`` under ``--open-profile``).
"""
pinned = _PINNED_PROCESS_HERMES_HOME
if pinned:
return _expand_hermes_home(pinned)
val = os.environ.get("HERMES_HOME", "").strip()
return _expand_hermes_home(val) if val else _get_platform_default_hermes_home()


def get_routing_process_hermes_home() -> Path:
"""Alias for :func:`get_process_hermes_home`; kept for callers that imported it explicitly."""
return get_process_hermes_home()


# Hermes-managed runtime downloads at the root of a home (GGUF models, llama.cpp runtimes,
# managed Node): re-downloadable on demand and routinely tens to hundreds of GB. Shared by
# ``hermes backup`` (excludes them) and ``profile create --clone-all`` (skips them from the
Expand Down
157 changes: 157 additions & 0 deletions tests/agent/test_serves_routed_profile_pin.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
"""A host that mirrors the turn's profile into HERMES_HOME must not flip any launch-home decision.

Hermes WebUI serves several profiles from one process and, for legacy readers, mirrors the active
turn's profile into ``os.environ["HERMES_HOME"]`` while also installing the context-local override.
Without a pinned process home the override then equals the "process" home and all four launch-home
decisions flip: serves_routed_profile(), _is_routed_home(), _is_process_home(), and
env_loader._process_hermes_home() all read get_process_hermes_home() which follows the env var.
"""
from __future__ import annotations

import pytest

import hermes_constants
from agent.secret_scope import serves_routed_profile, set_multiplex_active


@pytest.fixture(autouse=True)
def _reset_pin(monkeypatch):
monkeypatch.setattr(hermes_constants, "_PINNED_PROCESS_HERMES_HOME", None)
yield
monkeypatch.setattr(hermes_constants, "_PINNED_PROCESS_HERMES_HOME", None)


@pytest.fixture
def homes(tmp_path, monkeypatch):
launch = tmp_path / "launch"
served = tmp_path / "profiles" / "served"
launch.mkdir()
served.mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(launch))
return launch, served


def _with_override(path, fn):
token = hermes_constants.set_hermes_home_override(path)
try:
return fn()
finally:
hermes_constants.reset_hermes_home_override(token)


# ---------------------------------------------------------------------------
# serves_routed_profile
# ---------------------------------------------------------------------------

def test_unpinned_behaviour_is_unchanged(homes, monkeypatch):
launch, served = homes
assert _with_override(served, serves_routed_profile) is True
assert _with_override(launch, serves_routed_profile) is False
# Mirroring the served home into HERMES_HOME makes it the process home (legacy semantics).
monkeypatch.setenv("HERMES_HOME", str(served))
assert _with_override(served, serves_routed_profile) is False


def test_pinned_home_survives_a_mirrored_hermes_home(homes, monkeypatch):
launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
monkeypatch.setenv("HERMES_HOME", str(served)) # the host's per-turn mirror
assert _with_override(served, serves_routed_profile) is True
assert _with_override(launch, serves_routed_profile) is False
assert serves_routed_profile() is False # no override: the process's own profile
assert hermes_constants.get_process_hermes_home() == launch
assert hermes_constants.get_routing_process_hermes_home() == launch # alias


def test_clearing_the_pin_restores_hermes_home_semantics(homes, monkeypatch):
launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
hermes_constants.pin_process_hermes_home(None)
monkeypatch.setenv("HERMES_HOME", str(served))
assert hermes_constants.get_process_hermes_home() == served
assert _with_override(served, serves_routed_profile) is False


def test_first_pin_wins(homes):
launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
hermes_constants.pin_process_hermes_home(served) # ignored — first pin wins
assert hermes_constants.get_process_hermes_home() == launch


def test_mcp_connection_key_is_profile_scoped_under_a_mirrored_home(homes, monkeypatch):
from tools.mcp_tool_scope import _server_key
from tools.registry import registry

launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
monkeypatch.setenv("HERMES_HOME", str(served))
key = _with_override(served, lambda: _server_key("atlassian"))
assert key == (hermes_constants.hermes_home_key(served), "atlassian")
assert _with_override(served, registry.current_scope_key) == key[0]
assert _with_override(launch, lambda: _server_key("atlassian")) == "atlassian"


# ---------------------------------------------------------------------------
# set_multiplex_active auto-pins the launch home
# ---------------------------------------------------------------------------

def test_set_multiplex_active_pins_launch_home(homes, monkeypatch):
from agent.secret_scope import is_multiplex_active

launch, served = homes # HERMES_HOME == launch
assert hermes_constants._PINNED_PROCESS_HERMES_HOME is None
set_multiplex_active(True)
try:
assert hermes_constants._PINNED_PROCESS_HERMES_HOME == str(launch)
# Now mirror served into HERMES_HOME — pin must hold
monkeypatch.setenv("HERMES_HOME", str(served))
assert hermes_constants.get_process_hermes_home() == launch
finally:
set_multiplex_active(False)


def test_set_multiplex_active_respects_explicit_pin(homes, monkeypatch):
launch, served = homes # HERMES_HOME == launch
# Embedding host pins before multiplex is activated.
hermes_constants.pin_process_hermes_home(launch)
monkeypatch.setenv("HERMES_HOME", str(served))
set_multiplex_active(True)
try:
# Pin should still be launch, not served.
assert hermes_constants.get_process_hermes_home() == launch
finally:
set_multiplex_active(False)


# ---------------------------------------------------------------------------
# The three other launch-home decisions that were also broken (#119242)
# ---------------------------------------------------------------------------

def test_is_routed_home_respects_pin(homes, monkeypatch):
from tools.environments.local import _is_routed_home

launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
monkeypatch.setenv("HERMES_HOME", str(served))
assert _is_routed_home(served) is True # served != pinned launch
assert _is_routed_home(launch) is False # launch == pinned launch


def test_is_process_home_respects_pin(homes, monkeypatch):
from agent.secret_scope import _is_process_home

launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
monkeypatch.setenv("HERMES_HOME", str(served))
assert _is_process_home(launch) is True # launch == pinned
assert _is_process_home(served) is False # served != pinned


def test_env_loader_process_hermes_home_respects_pin(homes, monkeypatch):
from hermes_cli.env_loader import _process_hermes_home

launch, served = homes
hermes_constants.pin_process_hermes_home(launch)
monkeypatch.setenv("HERMES_HOME", str(served))
assert _process_hermes_home() == launch