hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix(gateway): a service definition is written only by the home it pins

Browse Source 
Two homes can resolve to one gateway service name: `<root>/profiles/<name>`
and `~/.hermes/profiles/<name>` both take `hermes-gateway-<name>`
(`ai.hermes.gateway-<name>` on launchd). Every gateway boot, start and
restart calls `refresh_systemd_unit_if_needed` / `refresh_launchd_plist_if_needed`,
which rewrote whatever definition sat at that name with no check of the
HERMES_HOME it pins. A scratch/E2E gateway started from a custom root
therefore repointed the real install's unit at the scratch directory; the
real gateway failed on its next restart (200/CHDIR once the scratch dir was
cleaned up). `systemd_install --force` and `launchd_install --force` did
the same. Uninstall already refused a unit pinning another home
(`_systemd_unit_belongs_to_current_home`); the writers did not.

- hermes_cli/gateway_service_owner.py (new sibling): `pinned_home()` reads
  the HERMES_HOME a unit or plist pins; `definition_belongs_to_home()`
  refuses a write from any other home; `refuse_foreign_home_install()`
  gates `gateway install` for a home outside the platform-default tree.
- Writers refuse a definition pinned to another home: the systemd and
  launchd refreshes (boot/start/restart) and both installers, `--force`
  included. The system-scope check runs on the caller's own home, before
  `systemd_unit_is_current` adopts the unit's home into os.environ (else
  the check compares the unit with itself); a sudo-stripped caller with no
  HERMES_HOME still acts for the unit's pinned home, as before.
- `gateway install` refuses a HERMES_HOME outside the account's Hermes
  tree that owns no bare service name and has no service installed yet: a
  service registered for a scratch home outlives the run.
- New `gateway install --force-unit-path` overrides both refusals (an
  explicit repoint of the definition at this home).
- Refused installs exit 1, so `gateway install` no longer goes on to start
  the other home's service; the setup wizard reports the refusal and
  continues, as ensure_gateway_service already did.
- Existing tests whose sandbox HERMES_HOME is a tmp dir opt in to the
  admission check (5 files, one monkeypatch each).

Tests: tests/hermes_cli/test_gateway_service_owner.py, 18 tests. On
e36a818033 (sources reverted, tests kept): 11 failed, 7 passed (foreign
unit/plist rewritten on refresh and --force, refused install starts the
service, scratch install proceeds, no --force-unit-path). With this change:
18 passed. test_colliding_profile_names_share_a_unit pins the premise on
the real name resolution.
This commit is contained in:
ang-fleet-workers[bot]
2026-10-05 12:17:56 -07:00
committed by Teknium
parent 8ac5c74432
commit b6ab666b6d
11 changed files with 418 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
hermes_cli/gateway.py
View File

@@ -3452,6 +3452,12 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
if not unit_path.exists():
return False
# Runs on every gateway boot/start/restart. A home whose service name collides with another install's
# (``<root>/profiles/x`` vs ``~/.hermes/profiles/x``) must not rewrite that install's unit. Checked on
# the caller's own home, before systemd_unit_is_current adopts the unit's home into os.environ.
from hermes_cli.gateway_service_owner import definition_belongs_to_home, service_home_for_unit
if not definition_belongs_to_home(unit_path, service_home_for_unit(unit_path, system), "rewrite"):
return False
# systemd_unit_is_current is the HERMES_HOME-sync chokepoint; its env mutation persists for the regenerate below.
current = systemd_unit_is_current(system=system)
if _retire_hermes_replace_dropin(system=system):
@@ -3640,10 +3646,25 @@ def systemd_install(
run_as_user: str | None = None,
enable_on_startup: bool = True,
non_interactive: bool = False,
force_unit_path: bool = False,
):
if system:
_require_root_for_system_service("install")
# A unit pinning another home is another install's gateway: refuse before any removal, sync or write,
# --force included. --force-unit-path is the explicit repoint.
unit_path = get_systemd_unit_path(system=system)
if not force_unit_path:
from hermes_cli.gateway_service_owner import definition_belongs_to_home, service_home_for_unit
target_home = service_home_for_unit(unit_path, system)
if system and run_as_user:
try:
target_home = Path(_hermes_home_for_target_user(_system_service_identity(run_as_user)[2]))
except ValueError:
pass # unknown account: generate_systemd_unit refuses it below with the real error
if not definition_belongs_to_home(unit_path, target_home, "overwrite"):
sys.exit(1)
# Offer to remove legacy units first: alongside the new unit they flap-fight for the bot token.
if has_legacy_hermes_units():
print()
@@ -3653,15 +3674,15 @@ def systemd_install(
remove_legacy_hermes_units(interactive=False)
print()
unit_path = get_systemd_unit_path(system=system)
scope_label = _service_scope_label(system)
sudo, scope_flag, user_flag = _systemd_cli_bits(system)
# Existing system units already pin HERMES_HOME; adopt it before any regenerate.
if unit_path.exists():
# Existing system units already pin HERMES_HOME; adopt it before any regenerate. A repoint keeps the
# caller's home (adopting the old one would regenerate the unit unchanged).
if unit_path.exists() and not force_unit_path:
_sync_hermes_home_from_systemd_unit(system=system)
if unit_path.exists() and not force:
if unit_path.exists() and not (force or force_unit_path):
if not systemd_unit_is_current(system=system):
print(f"↻ Repairing outdated {scope_label} systemd service at: {unit_path}")
refresh_systemd_unit_if_needed(system=system)
@@ -5246,6 +5267,7 @@ def _install_systemd_from_cli(args, *, force: bool, system: bool, run_as_user) -
systemd_install(
force=force, system=system, run_as_user=run_as_user,
enable_on_startup=start_on_login, non_interactive=non_interactive,
force_unit_path=getattr(args, "force_unit_path", False),
)
if start_now:
systemd_start(system=system)
@@ -5266,12 +5288,18 @@ def _cmd_install(args):
if _service_mgmt_blocked():
_no_backend_exit("install", "termux")
backend = _service_backend()
if backend == "systemd":
if refuses_container_user_scope_install(system):
force_unit_path = getattr(args, "force_unit_path", False)
if backend == "systemd" and refuses_container_user_scope_install(system):
sys.exit(1)
if backend in ("systemd", "launchd"):
from hermes_cli.gateway_service_owner import refuse_foreign_home_install
if refuse_foreign_home_install(get_hermes_home(), force_unit_path):
sys.exit(1)
if backend == "systemd":
_install_systemd_from_cli(args, force=force, system=system, run_as_user=run_as_user)
elif backend == "launchd":
launchd_install(force, start_now=getattr(args, "start_now", None) is not False)
launchd_install(force, start_now=getattr(args, "start_now", None) is not False,
force_unit_path=force_unit_path)
elif backend == "windows":
_gw_windows().install(
force=force,

12
hermes_cli/gateway_launchd.py
View File

@@ -540,6 +540,10 @@ def refresh_launchd_plist_if_needed() -> bool:
plist_path = _gw().get_launchd_plist_path()
if not plist_path.exists() or _gw().launchd_plist_is_current():
return False
# Runs on every gateway start/restart: a plist pinning another home is another install's gateway.
from hermes_cli.gateway_service_owner import definition_belongs_to_home
if not definition_belongs_to_home(plist_path, _gw().get_hermes_home(), "rewrite"):
return False
new_plist = _gw().generate_launchd_plist()
if _gw()._refuse_temp_home_service_write(new_plist, "launchd plist"):
@@ -599,14 +603,18 @@ def refresh_launchd_plist_if_needed() -> bool:
return True
def launchd_install(force: bool = False, *, start_now: bool = True):
def launchd_install(force: bool = False, *, start_now: bool = True, force_unit_path: bool = False):
plist_path = _gw().get_launchd_plist_path()
label = _gw().get_launchd_label()
# A plist pinning another home is another install's gateway, --force included; --force-unit-path repoints.
from hermes_cli.gateway_service_owner import definition_belongs_to_home
if not force_unit_path and not definition_belongs_to_home(plist_path, _gw().get_hermes_home(), "overwrite"):
sys.exit(1)
# Loading the plist starts the gateway (RunAtLoad), so a no-start install writes it without
# loading it. A gateway that launchd already runs is still reloaded; this install did not start it.
load = start_now or _gw()._launchctl_label_supervising_process(label)
if plist_path.exists() and not force:
if plist_path.exists() and not (force or force_unit_path):
if _gw().launchd_plist_is_current():
print(f"Service already installed at: {plist_path}")
print("Use --force to reinstall")

122
hermes_cli/gateway_service_owner.py Normal file
View File

@@ -0,0 +1,122 @@
"""Who may write a gateway service definition (systemd unit, launchd plist).
A service definition pins the ``HERMES_HOME`` its gateway runs with. Service names are derived from the
home (bare / profile name / path hash), and two homes can still resolve to the same name: a custom root's
``<root>/profiles/<name>`` takes the same ``hermes-gateway-<name>`` unit as ``~/.hermes/profiles/<name>``.
Every gateway boot refreshes "its" unit, so a scratch or E2E gateway started from such a home rewrote the
real install's unit to point at the scratch directory, and the real gateway failed on its next restart.
The rule: a definition that pins a home is written only by a process acting for that home. Uninstall
already refused a unit pinning another home (``_systemd_unit_belongs_to_current_home``); the writers now
refuse the same way. ``hermes gateway install --force-unit-path`` is the explicit repoint.
"""
from __future__ import annotations
import os
import plistlib
from pathlib import Path
def pinned_home(definition_path: Path) -> str | None:
"""``HERMES_HOME`` pinned by the systemd unit or launchd plist at *definition_path*, or None."""
if definition_path.suffix == ".plist":
try:
with definition_path.open("rb") as fh:
data = plistlib.load(fh)
except Exception: # unreadable / not a plist (hand-edited, truncated): pins nobody
return None
env = data.get("EnvironmentVariables") if isinstance(data, dict) else None
value = env.get("HERMES_HOME") if isinstance(env, dict) else None
return (str(value).strip() or None) if value is not None else None
from hermes_cli import gateway as gw
return gw._hermes_home_pinned_by_unit(definition_path)
def _resolve(raw: str | Path) -> Path | None:
try:
return Path(raw).expanduser().resolve()
except (OSError, RuntimeError, ValueError):
return None
def service_home_for_unit(unit_path: Path, system: bool) -> Path:
"""The ``HERMES_HOME`` this process acts for when it writes the systemd unit at *unit_path*.
User scope: its own home. System scope: an explicit ``HERMES_HOME`` remapped to the unit's ``User=``
account the way ``generate_systemd_unit`` pins it; with none (``sudo`` strips it, HOME=/root), the
unit's own pinned home, which is the only thing naming the install being operated on. Pure: it never
adopts the unit's home into ``os.environ``, so a check on this value compares caller and unit.
"""
from hermes_cli import gateway as gw
if not system:
return gw.get_hermes_home()
if not os.environ.get("HERMES_HOME", "").strip():
pinned = pinned_home(unit_path) if unit_path.exists() else None
return Path(pinned).expanduser() if pinned else gw.get_hermes_home()
user = gw._read_systemd_user_from_unit(unit_path)
if not user:
return gw.get_hermes_home()
try:
home_dir = gw._system_service_identity(run_as_user=user)[2]
except ValueError: # unknown User=: generate_systemd_unit refuses the same way
return gw.get_hermes_home()
return Path(gw._hermes_home_for_target_user(home_dir))
def definition_belongs_to_home(definition_path: Path, home: Path, action: str) -> bool:
"""False (with guidance printed) when the existing definition pins a ``HERMES_HOME`` other than *home*.
A missing definition, or one that pins no home (hand-written, pre-pinning), belongs to nobody else.
"""
if not definition_path.exists():
return True
raw = pinned_home(definition_path)
if raw is None or _resolve(raw) == _resolve(home):
return True
print(f"✗ Refusing to {action} {definition_path}: it runs HERMES_HOME={raw}, "
f"but this process has HERMES_HOME={home}.")
print(" That file is another install's gateway. Use that install, or pass --force-unit-path")
print(" to `hermes gateway install` if you really mean to repoint it at this home.")
return False
def _installed_definition_pins(home: Path) -> bool:
"""True when a definition at this home's service path already pins *home* (a re-install)."""
from hermes_cli import gateway as gw
paths = [gw.get_systemd_unit_path(system=False), gw.get_systemd_unit_path(system=True)]
if gw.is_macos():
paths.append(gw.get_launchd_plist_path())
for path in paths:
raw = pinned_home(path) if path.exists() else None
if raw is not None and _resolve(raw) == home:
return True
return False
def home_may_install_service(home: Path) -> bool:
"""True for a home that may own a host gateway service: inside the platform-default Hermes tree
(``~/.hermes`` and its profiles, the sudo invoker's included), the bare-name owner, or a home whose
installed definition already pins it."""
from hermes_cli import gateway as gw
resolved = _resolve(home)
if resolved is None:
return False
if any(resolved == root or root in resolved.parents for root in gw._native_service_homes()):
return True
return gw._home_owns_bare_service_name(resolved) or _installed_definition_pins(resolved)
def refuse_foreign_home_install(home: Path, force_unit_path: bool) -> bool:
"""True (with guidance printed) when ``hermes gateway install`` must not register a service for *home*."""
if force_unit_path or home_may_install_service(home):
return False
print(f"✗ Refusing to install a gateway service for HERMES_HOME={home}.")
print(" It is outside this account's Hermes tree and no service is installed for it yet, so it looks")
print(" like a scratch / test / E2E home. A service installed for it would outlive that run and restart")
print(" a gateway on this host forever. Run that gateway in the foreground (`hermes gateway run`),")
print(" or pass --force-unit-path if this custom home really is meant to own a host service.")
return True

5
hermes_cli/gateway_setup_wizard.py
View File

@@ -872,6 +872,11 @@ def _wizard_install_service(backend: str) -> None:
except subprocess.CalledProcessError as e:
_gw().print_error(f" Install failed: {e}")
_gw().print_info(" You can try manually: hermes gateway install")
except SystemExit:
# The backend installers exit non-zero on a refused write (a unit pinning another home);
# setup reports it and goes on rather than ending the wizard.
_gw().print_warning(" Gateway service install did not complete.")
_gw().print_info(" You can try manually: hermes gateway install")
def _wizard_post_setup() -> None:

3
hermes_cli/subcommands/gateway.py
View File

@@ -133,6 +133,9 @@ def build_gateway_parser(
help="Do not enable the service to start on login/boot")
_flag(gateway_install, "--if-missing", dest="if_missing",
help="Do nothing when a gateway service is already installed")
_flag(gateway_install, "--force-unit-path", dest="force_unit_path",
help="Install for a HERMES_HOME outside this account's Hermes tree, or repoint a service "
"definition that pins another home")
_flag(gateway_install, "--elevated-handoff", dest="elevated_handoff", help=argparse.SUPPRESS)
gateway_uninstall = gateway_subparsers.add_parser("uninstall", help="Uninstall gateway service")

5
tests/hermes_cli/test_gateway.py
View File

@@ -1136,7 +1136,10 @@ def test_install_if_missing_only_installs_when_no_service_exists(monkeypatch, in
monkeypatch.setattr(gateway, "_guard_named_profile_under_multiplexer", lambda force: None)
monkeypatch.setattr(gateway, "_service_mgmt_blocked", lambda: False)
monkeypatch.setattr(gateway, "_service_backend", lambda: "launchd")
monkeypatch.setattr(gateway, "launchd_install", lambda force, start_now: installs.append(force))
# The sandbox HERMES_HOME is a scratch home, which `gateway install` refuses by design.
monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True)
monkeypatch.setattr(gateway, "launchd_install",
lambda force, start_now, force_unit_path=False: installs.append(force))
gateway._cmd_install(SimpleNamespace(if_missing=True, force=False, system=False, run_as_user=None))

2
tests/hermes_cli/test_gateway_multiplex_served_record.py
View File

@@ -161,6 +161,8 @@ def test_service_verbs_do_not_start_a_second_gateway(served_root, monkeypatch, v
monkeypatch.setattr(gw, "_installed_service_kind_for", lambda *a, **k: "systemd")
monkeypatch.setattr(gw, "is_managed", lambda: False)
monkeypatch.setattr(gw, "is_termux", lambda: False)
# The served root is a tmp dir, which `gateway install` refuses as a scratch home by design.
monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True)
fn = getattr(gw, f"_cmd_{verb}")
ns = argparse.Namespace(system=False, all=False, force=False, run_as_user=None)
if verb == "restart":

2
tests/hermes_cli/test_gateway_no_new_standalone_profile.py
View File

@@ -39,6 +39,8 @@ def quiet_host(tmp_path, monkeypatch):
monkeypatch.setattr(gw, "_dispatch_via_service_manager_if_s6", lambda v: False)
monkeypatch.setattr(gw, "is_managed", lambda: False)
monkeypatch.setattr(gw, "is_termux", lambda: False)
# tmp root is outside the platform-default tree, which `gateway install` refuses by design.
monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True)
calls: list[str] = []
monkeypatch.setattr(gw, "_install_systemd_from_cli", lambda *a, **k: calls.append("install"))
monkeypatch.setattr(gw, "_service_call", lambda backend, v, system: calls.append(v))

2
tests/hermes_cli/test_gateway_service.py
View File

@@ -2071,6 +2071,8 @@ class TestDockerAwareGateway:
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
monkeypatch.setattr(gateway_cli, "is_container", lambda: True)
# The sandbox HERMES_HOME is a scratch home, which `gateway install` refuses by design.
monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True)
calls = []
monkeypatch.setattr(
gateway_cli,

231
tests/hermes_cli/test_gateway_service_owner.py Normal file
View File

@@ -0,0 +1,231 @@
"""Gateway service-definition writes belong to the home the definition pins.
Two homes can resolve to one service name: ``<root>/profiles/<name>`` and ``~/.hermes/profiles/<name>``
both take ``hermes-gateway-<name>``. Every gateway boot refreshes "its" unit, so a scratch or E2E gateway
started from such a home rewrote the real install's unit (WorkingDirectory and HERMES_HOME pointed at the
scratch dir) and the real gateway failed on its next restart. The collision is simulated here by pointing
the unit/plist path at a file that pins another home; ``test_colliding_profile_names_share_a_unit`` proves
the collision itself on the real name resolution.
"""
import argparse
import plistlib
from pathlib import Path
from types import SimpleNamespace
import pytest
import hermes_cli.gateway as gw
from hermes_cli import gateway_launchd, gateway_service_owner
from hermes_cli.subcommands.gateway import build_gateway_parser
REAL_UNIT = (
"[Service]\n"
"ExecStart=/home/ace/.hermes/hermes-agent/venv/bin/python -m hermes_cli.main gateway run\n"
"WorkingDirectory={home}\n"
'Environment="HERMES_HOME={home}"\n'
)
@pytest.fixture
def homes(tmp_path, monkeypatch):
real = tmp_path / "account" / ".hermes" / "profiles" / "coder"
scratch = tmp_path / "ci-scratch" / "run-1" / "profiles" / "coder"
real.mkdir(parents=True)
scratch.mkdir(parents=True)
# The process is the scratch gateway. A scratch dir is not always under a temp root (CI scratch
# volumes), so the temp-home guard is out of the way: the ownership check is what is under test.
monkeypatch.setenv("HERMES_HOME", str(scratch))
monkeypatch.setattr(gw, "_refuse_temp_home_service_write", lambda definition, kind: False)
return SimpleNamespace(real=real, scratch=scratch)
def test_colliding_profile_names_share_a_unit(homes, tmp_path, monkeypatch):
"""The premise: a custom root's profile resolves to the same unit as the account's profile."""
monkeypatch.setattr(gw, "_bare_unit_pinned_home", lambda: None)
monkeypatch.setattr(gw, "user_systemd_unit_dir", lambda: tmp_path / "systemd-user")
monkeypatch.setattr(gw, "_native_service_homes", lambda: {(tmp_path / "account" / ".hermes").resolve()})
scratch_unit = gw.get_systemd_unit_path(system=False)
monkeypatch.setenv("HERMES_HOME", str(homes.real))
assert gw.get_systemd_unit_path(system=False) == scratch_unit
@pytest.fixture
def systemd_unit(tmp_path, homes, monkeypatch):
unit = tmp_path / "systemd" / "hermes-gateway-coder.service"
unit.parent.mkdir()
unit.write_text(REAL_UNIT.format(home=homes.real), encoding="utf-8")
calls = []
monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: unit)
monkeypatch.setattr(gw, "systemd_unit_is_current", lambda system=False: False)
monkeypatch.setattr(gw, "_retire_hermes_replace_dropin", lambda system=False: False)
monkeypatch.setattr(gw, "_prepare_service_launcher", lambda system=False, run_as_user=None: None)
monkeypatch.setattr(gw, "_run_systemctl", lambda args, **kw: calls.append(tuple(args)) or
SimpleNamespace(returncode=0, stdout="", stderr=""))
monkeypatch.setattr(gw, "has_legacy_hermes_units", lambda: False)
monkeypatch.setattr(gw, "_ensure_linger_enabled", lambda *a, **k: True)
monkeypatch.setattr(gw, "_ensure_system_service_linger", lambda *a, **k: None)
monkeypatch.setattr(gw, "print_systemd_scope_conflict_warning", lambda: None)
monkeypatch.setattr(gw, "print_legacy_unit_warning", lambda: None)
monkeypatch.setattr(gw, "generate_systemd_unit", lambda system=False, run_as_user=None:
REAL_UNIT.format(home=Path(gw.get_hermes_home())))
return SimpleNamespace(path=unit, calls=calls, original=unit.read_text(encoding="utf-8"))
class TestSystemdWriters:
def test_boot_refresh_leaves_a_unit_pinned_to_another_home_untouched(self, systemd_unit, capsys):
assert gw.refresh_systemd_unit_if_needed(system=False) is False
assert systemd_unit.path.read_text(encoding="utf-8") == systemd_unit.original
assert ("daemon-reload",) not in systemd_unit.calls
assert "Refusing to rewrite" in capsys.readouterr().out
@pytest.fixture
def marker_free_unit(self, monkeypatch):
# refresh's own test belt refuses a generated unit naming a pytest tmpdir; any marker-free body works.
monkeypatch.setattr(gw, "generate_systemd_unit", lambda system=False, run_as_user=None: "ExecStart=new\n")
def test_boot_refresh_still_rewrites_its_own_stale_unit(self, systemd_unit, homes, marker_free_unit,
monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(homes.real))
assert gw.refresh_systemd_unit_if_needed(system=False) is True
assert systemd_unit.path.read_text(encoding="utf-8") == "ExecStart=new\n"
assert ("daemon-reload",) in systemd_unit.calls
def test_unpinned_unit_is_nobody_elses(self, systemd_unit, marker_free_unit):
systemd_unit.path.write_text("[Service]\nExecStart=/usr/bin/hermes gateway run\n", encoding="utf-8")
assert gw.refresh_systemd_unit_if_needed(system=False) is True
def test_force_install_does_not_overwrite_another_homes_unit(self, systemd_unit):
with pytest.raises(SystemExit) as exc:
gw.systemd_install(force=True, non_interactive=True)
assert exc.value.code == 1
assert systemd_unit.path.read_text(encoding="utf-8") == systemd_unit.original
assert systemd_unit.calls == []
def test_refused_install_never_starts_the_other_homes_service(self, systemd_unit, monkeypatch):
started = []
monkeypatch.setattr(gw, "systemd_start", lambda system=False: started.append(system))
args = SimpleNamespace(start_now=True, start_on_login=True, force_unit_path=False)
with pytest.raises(SystemExit):
gw._install_systemd_from_cli(args, force=False, system=False, run_as_user=None)
assert started == []
def test_force_unit_path_repoints_the_unit(self, systemd_unit, homes):
gw.systemd_install(force_unit_path=True, non_interactive=True)
assert f"HERMES_HOME={homes.scratch}" in systemd_unit.path.read_text(encoding="utf-8")
def test_system_refresh_checks_the_callers_home_before_adopting_the_units(
self, systemd_unit, homes, monkeypatch):
# systemd_unit_is_current adopts a system unit's HERMES_HOME into os.environ (sudo strips it);
# checking after that adoption compared the unit with itself.
monkeypatch.setattr(gw, "_read_systemd_user_from_unit", lambda path: None)
def is_current(system=False):
gw._sync_hermes_home_from_systemd_unit(system=system)
return False
monkeypatch.setattr(gw, "systemd_unit_is_current", is_current)
assert gw.refresh_systemd_unit_if_needed(system=True) is False
assert systemd_unit.path.read_text(encoding="utf-8") == systemd_unit.original
assert Path(gw.get_hermes_home()) == homes.scratch
def test_sudo_stripped_home_still_refreshes_the_system_unit(self, systemd_unit, homes, marker_free_unit,
monkeypatch):
# Under sudo there is no explicit HERMES_HOME: the unit's own pinned home names the install.
monkeypatch.delenv("HERMES_HOME")
monkeypatch.setattr(gw, "_read_systemd_user_from_unit", lambda path: None)
assert gw.refresh_systemd_unit_if_needed(system=True) is True
class TestLaunchdWriters:
@pytest.fixture
def plist(self, tmp_path, homes, monkeypatch):
path = tmp_path / "LaunchAgents" / "ai.hermes.gateway-coder.plist"
path.parent.mkdir()
data = {"Label": "ai.hermes.gateway-coder", "EnvironmentVariables": {"HERMES_HOME": str(homes.real)}}
path.write_bytes(plistlib.dumps(data))
monkeypatch.setattr(gw, "get_launchd_plist_path", lambda: path)
monkeypatch.setattr(gw, "get_launchd_label", lambda: "ai.hermes.gateway-coder")
monkeypatch.setattr(gw, "launchd_plist_is_current", lambda: False)
monkeypatch.setattr(gw, "generate_launchd_plist", lambda: "<plist>new</plist>")
monkeypatch.setattr(gw, "_prepare_service_launcher", lambda *a, **k: None)
monkeypatch.setattr(gw, "_launchctl_label_supervising_process", lambda label: None)
monkeypatch.setattr(gateway_launchd.subprocess, "run", lambda *a, **k: SimpleNamespace(returncode=0))
return SimpleNamespace(path=path, original=path.read_bytes())
def test_refresh_leaves_a_plist_pinned_to_another_home_untouched(self, plist):
assert gateway_launchd.refresh_launchd_plist_if_needed() is False
assert plist.path.read_bytes() == plist.original
def test_force_install_does_not_overwrite_another_homes_plist(self, plist):
with pytest.raises(SystemExit):
gateway_launchd.launchd_install(force=True, start_now=False)
assert plist.path.read_bytes() == plist.original
def test_force_unit_path_repoints_the_plist(self, plist):
gateway_launchd.launchd_install(start_now=False, force_unit_path=True)
assert plist.path.read_text(encoding="utf-8") == "<plist>new</plist>"
class TestInstallAdmission:
"""`gateway install` registers a persistent host service; a scratch/E2E home must not get one."""
@pytest.fixture
def install_cli(self, homes, tmp_path, monkeypatch):
installs = []
monkeypatch.setattr(gw, "is_managed", lambda: False)
monkeypatch.setattr(gw, "_service_mgmt_blocked", lambda: False)
monkeypatch.setattr(gw, "_guard_named_profile_under_multiplexer", lambda force: None)
monkeypatch.setattr(gw, "_service_backend", lambda: "launchd")
monkeypatch.setattr(gw, "is_macos", lambda: True)
monkeypatch.setattr(gw, "_native_service_homes", lambda: {(tmp_path / "account" / ".hermes").resolve()})
monkeypatch.setattr(gw, "_bare_unit_pinned_home", lambda: None)
monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: tmp_path / "none.service")
monkeypatch.setattr(gw, "get_launchd_plist_path", lambda: tmp_path / "none.plist")
monkeypatch.setattr(gw, "launchd_install", lambda force, start_now, force_unit_path=False:
installs.append(force_unit_path))
return installs
@staticmethod
def _args(**kw):
return SimpleNamespace(if_missing=False, force=False, system=False, run_as_user=None, **kw)
def test_scratch_home_is_refused(self, install_cli):
with pytest.raises(SystemExit) as exc:
gw._cmd_install(self._args(force_unit_path=False))
assert exc.value.code == 1
assert install_cli == []
def test_force_unit_path_admits_a_custom_home(self, install_cli):
gw._cmd_install(self._args(force_unit_path=True))
assert install_cli == [True]
def test_account_profile_is_admitted(self, install_cli, homes, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(homes.real))
gw._cmd_install(self._args(force_unit_path=False))
assert install_cli == [False]
def test_custom_home_with_its_own_installed_service_is_admitted(self, install_cli, homes, tmp_path,
monkeypatch):
own = tmp_path / "own.plist"
own.write_bytes(plistlib.dumps({"EnvironmentVariables": {"HERMES_HOME": str(homes.scratch)}}))
monkeypatch.setattr(gw, "get_launchd_plist_path", lambda: own)
gw._cmd_install(self._args(force_unit_path=False))
assert install_cli == [False]
def test_parser_accepts_force_unit_path(self):
parser = argparse.ArgumentParser()
build_gateway_parser(parser.add_subparsers(dest="command"), cmd_gateway=lambda a: None,
cmd_proxy=lambda a: None, cmd_gateway_enroll=lambda a: None)
assert parser.parse_args(["gateway", "install", "--force-unit-path"]).force_unit_path is True
def test_pinned_home_reads_both_definition_kinds(tmp_path):
unit = tmp_path / "g.service"
unit.write_text('[Service]\nEnvironment="HERMES_HOME=/srv/a"\n', encoding="utf-8")
plist = tmp_path / "g.plist"
plist.write_bytes(plistlib.dumps({"EnvironmentVariables": {"HERMES_HOME": "/srv/b"}}))
broken = tmp_path / "broken.plist"
broken.write_text("not a plist", encoding="utf-8")
assert gateway_service_owner.pinned_home(unit) == "/srv/a"
assert gateway_service_owner.pinned_home(plist) == "/srv/b"
assert gateway_service_owner.pinned_home(broken) is None

2
tests/hermes_cli/test_launchd_install_start_now.py
View File

@@ -41,6 +41,8 @@ def launchd(tmp_path, monkeypatch):
monkeypatch.setattr(gateway_cli, "get_launchd_label", lambda: LABEL)
monkeypatch.setattr(gateway_cli, "_launchd_domain", lambda: DOMAIN)
monkeypatch.setattr(gateway_cli, "_refuse_temp_home_service_write", lambda *a: False)
# The sandbox HERMES_HOME is a scratch home, which `gateway install` refuses by design.
monkeypatch.setattr("hermes_cli.gateway_service_owner.home_may_install_service", lambda home: True)
monkeypatch.setattr(gateway_cli, "_clear_launchd_unsupported_marker", lambda: None)
monkeypatch.setattr(gateway_cli, "_launchctl_supervised_pid", lambda label: state.supervised_pid)
monkeypatch.setattr(gateway_cli, "_launchctl_bootstrap", lambda *a, **k: state.bootstraps.append(a))