Compare commits

...

2 commits

Author SHA1 Message Date
Wolfram Ravenwolf
1df3834d26
fix(gateway): keep macOS launchd runtime paths logical
Consolidates these related Amy fork patches:
- 9b6df28be fix(gateway): include Amy bin paths in macOS launchd PATH
- c1b50030d fix(gateway): defer macOS launchd reloads from gateway process
- 86a326fbe fix(gateway): preserve logical amy home in launchd
- 6271302ba fix(gateway): drop macOS launchd legacy root paths
- d1616cac4 fix(gateway): keep launchd Hermes paths logical
2026-06-16 06:09:01 -07:00
Wolfram Ravenwolf
a6bb43362c
feat(gateway): add macOS app-wrapper launchd identity 2026-06-16 06:09:01 -07:00
3 changed files with 1117 additions and 139 deletions

View file

@ -7,10 +7,12 @@ Handles: hermes gateway [run|start|stop|restart|status|install|uninstall|setup]
import asyncio
import logging
import os
import plistlib
import shutil
import signal
import subprocess
import sys
import tempfile
import textwrap
from dataclasses import dataclass
from pathlib import Path
@ -2169,6 +2171,278 @@ def get_launchd_plist_path() -> Path:
return _launchd_user_home() / "Library" / "LaunchAgents" / f"{name}.plist"
MACOS_APP_WRAPPER_DISPLAY_NAME = "Hermes Agent"
MACOS_APP_WRAPPER_ENV_KEY = "HERMES_LAUNCHD_APP_WRAPPER"
MACOS_APP_WRAPPER_SOURCE_INFO = "HermesPythonSource.plist"
def get_launchd_bundle_identifier() -> str:
"""Return the bundle identifier associated with the launchd gateway job."""
return get_launchd_label()
def get_launchd_app_wrapper_path() -> Path:
"""Return the macOS app bundle path used for the optional launchd wrapper."""
suffix = _profile_suffix()
bundle_name = (
f"{MACOS_APP_WRAPPER_DISPLAY_NAME} ({suffix}).app"
if suffix
else f"{MACOS_APP_WRAPPER_DISPLAY_NAME}.app"
)
return get_hermes_home() / "macos" / bundle_name
def get_launchd_app_wrapper_executable_path() -> Path:
"""Return the executable path inside the optional macOS app wrapper."""
return (
get_launchd_app_wrapper_path()
/ "Contents"
/ "MacOS"
/ MACOS_APP_WRAPPER_DISPLAY_NAME
)
def _python_home_from_path(python_path: str | Path | None = None, venv: Path | None = None) -> Path:
"""Return the base Python distribution root for a venv/base executable."""
if venv is not None:
pyvenv_cfg = venv / "pyvenv.cfg"
if pyvenv_cfg.exists():
for line in pyvenv_cfg.read_text(encoding="utf-8", errors="ignore").splitlines():
key, sep, value = line.partition("=")
if sep and key.strip().lower() == "home" and value.strip():
home = Path(value.strip()).expanduser()
if home.name in {"bin", "Scripts"}:
return home.parent
return home
resolved = Path(python_path or get_python_path()).resolve()
return resolved.parent.parent
def _venv_site_packages_path(venv: Path | None) -> Path | None:
"""Return the best site-packages path for a virtualenv, if known."""
if venv is None:
return None
lib_dir = venv / "lib"
if lib_dir.exists():
candidates = sorted(lib_dir.glob("python*/site-packages"))
if candidates:
return candidates[0]
return venv / "lib" / f"python{sys.version_info.major}.{sys.version_info.minor}" / "site-packages"
def _launchd_child_python_path(venv: Path | None) -> str:
"""Return the real interpreter path child processes should use."""
if venv is not None:
candidate = venv / ("Scripts/python.exe" if is_windows() else "bin/python")
if candidate.exists():
return str(candidate)
return get_python_path()
def _launchd_app_wrapper_pythonpath(venv: Path | None) -> str:
"""Build a stable PYTHONPATH for the copied Python executable inside the app wrapper."""
parts: list[str] = []
site_packages = _venv_site_packages_path(venv)
if site_packages is not None:
parts.append(str(site_packages))
parts.append(str(PROJECT_ROOT))
return ":".join(dict.fromkeys(parts))
def _launchd_app_wrapper_info() -> dict:
"""Return Info.plist metadata for the optional macOS launchd app wrapper."""
from hermes_cli import __version__ as hermes_version
return {
"CFBundleIdentifier": get_launchd_bundle_identifier(),
"CFBundleName": MACOS_APP_WRAPPER_DISPLAY_NAME,
"CFBundleDisplayName": MACOS_APP_WRAPPER_DISPLAY_NAME,
"CFBundleExecutable": MACOS_APP_WRAPPER_DISPLAY_NAME,
"CFBundlePackageType": "APPL",
"CFBundleVersion": hermes_version,
"CFBundleShortVersionString": hermes_version,
# No Dock icon/window for this background helper when LaunchServices
# encounters it. launchd still executes the bundled binary directly.
"LSBackgroundOnly": True,
}
def _launchd_app_wrapper_source_info(source_python: Path | None = None) -> dict:
"""Return stable source metadata for deciding whether the wrapper is fresh."""
source = Path(source_python or get_python_path()).resolve()
stat = source.stat()
return {
"SourcePython": str(source),
"SourceSize": stat.st_size,
"SourceMTimeNs": stat.st_mtime_ns,
}
def _launchd_app_wrapper_signature_is_valid(app_path: Path | None = None) -> bool:
"""Return True when the macOS app wrapper has a valid code signature."""
target = app_path or get_launchd_app_wrapper_path()
try:
result = subprocess.run(
["codesign", "--verify", "--deep", "--strict", str(target)],
capture_output=True,
text=True,
timeout=30,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
return result.returncode == 0
def launchd_app_wrapper_is_current() -> bool:
"""Return True when the optional macOS app wrapper matches this install."""
app_path = get_launchd_app_wrapper_path()
executable_path = get_launchd_app_wrapper_executable_path()
info_path = app_path / "Contents" / "Info.plist"
source_info_path = app_path / "Contents" / "Resources" / MACOS_APP_WRAPPER_SOURCE_INFO
source_python = Path(get_python_path()).resolve()
if (
not executable_path.exists()
or not info_path.exists()
or not source_info_path.exists()
or not source_python.exists()
):
return False
try:
info = plistlib.loads(info_path.read_bytes())
source_info = plistlib.loads(source_info_path.read_bytes())
expected_source_info = _launchd_app_wrapper_source_info(source_python)
except Exception:
return False
expected = _launchd_app_wrapper_info()
return (
all(info.get(key) == value for key, value in expected.items())
and all(source_info.get(key) == value for key, value in expected_source_info.items())
and _launchd_app_wrapper_signature_is_valid(app_path)
)
def install_launchd_app_wrapper(force: bool = False) -> Path:
"""Install/update the optional macOS app bundle used as launchd executable.
The bundle contains a *copy* of the active Python executable named
``Hermes Agent``. launchd then starts that bundle executable instead of the
generic ``python3.13`` binary, giving macOS/TCC a Hermes-specific path and
bundle identity while still running Hermes through the existing venv.
"""
app_path = get_launchd_app_wrapper_path()
if app_path.exists() and not force and launchd_app_wrapper_is_current():
return app_path
source_python = Path(get_python_path()).resolve()
if not source_python.exists():
raise FileNotFoundError(f"Python executable not found: {source_python}")
detected_venv = _detect_venv_dir()
app_parent = app_path.parent
app_parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix=f".{app_path.name}.", dir=app_parent) as tmpdir:
staging_app_path = Path(tmpdir) / app_path.name
contents_dir = staging_app_path / "Contents"
macos_dir = contents_dir / "MacOS"
resources_dir = contents_dir / "Resources"
macos_dir.mkdir(parents=True, exist_ok=True)
resources_dir.mkdir(parents=True, exist_ok=True)
executable_path = macos_dir / get_launchd_app_wrapper_executable_path().name
shutil.copy2(source_python, executable_path)
executable_path.chmod(executable_path.stat().st_mode | 0o755)
info_path = contents_dir / "Info.plist"
info_path.write_bytes(plistlib.dumps(_launchd_app_wrapper_info(), sort_keys=False))
source_info_path = resources_dir / MACOS_APP_WRAPPER_SOURCE_INFO
source_info_path.write_bytes(
plistlib.dumps(_launchd_app_wrapper_source_info(source_python), sort_keys=False)
)
subprocess.run(
["codesign", "--force", "--deep", "--sign", "-", str(staging_app_path)],
check=True,
timeout=30,
)
subprocess.run(
["codesign", "--verify", "--deep", "--strict", str(staging_app_path)],
check=True,
timeout=30,
)
smoke_env = os.environ.copy()
smoke_env.update(
{
"PYTHONHOME": str(_python_home_from_path(source_python, detected_venv)),
"PYTHONPATH": _launchd_app_wrapper_pythonpath(detected_venv),
"PYTHONEXECUTABLE": _launchd_child_python_path(detected_venv),
"VIRTUAL_ENV": str(detected_venv) if detected_venv else "",
"HERMES_HOME": str(get_hermes_home().resolve()),
MACOS_APP_WRAPPER_ENV_KEY: "1",
}
)
if detected_venv:
smoke_env["PATH"] = f"{detected_venv / 'bin'}:{smoke_env.get('PATH', '')}"
subprocess.run(
[str(executable_path), "-c", "import encodings, sys; print(sys.executable)"],
check=True,
capture_output=True,
text=True,
timeout=30,
env=smoke_env,
)
backup_path = app_parent / f".{app_path.name}.previous"
if backup_path.exists():
shutil.rmtree(backup_path)
try:
if app_path.exists():
app_path.rename(backup_path)
staging_app_path.rename(app_path)
except Exception:
if not app_path.exists() and backup_path.exists():
backup_path.rename(app_path)
raise
finally:
if backup_path.exists():
shutil.rmtree(backup_path)
return app_path
def _installed_launchd_plist_uses_app_wrapper(plist_path: Path | None = None) -> bool:
"""Return True when an installed launchd plist already targets the app wrapper."""
path = plist_path or get_launchd_plist_path()
if not path.exists():
return False
try:
data = plistlib.loads(path.read_bytes())
except Exception:
return False
if not isinstance(data, dict):
return False
env = data.get("EnvironmentVariables") or {}
if env.get(MACOS_APP_WRAPPER_ENV_KEY) == "1":
return True
program = data.get("Program")
args = data.get("ProgramArguments") or []
first_arg = args[0] if args else None
app_exe = str(get_launchd_app_wrapper_executable_path())
return program == app_exe or first_arg == app_exe
def _resolve_launchd_app_wrapper_mode(app_wrapper: bool | None = None) -> bool:
"""Resolve target app-wrapper mode, preserving existing wrapper installs."""
if app_wrapper is not None:
return app_wrapper
return _installed_launchd_plist_uses_app_wrapper()
def _drop_launchd_app_wrapper_python_env() -> None:
"""Remove wrapper-only Python startup env so child processes inherit clean env."""
if os.environ.get(MACOS_APP_WRAPPER_ENV_KEY) == "1":
os.environ.pop("PYTHONHOME", None)
os.environ.pop("PYTHONPATH", None)
os.environ.pop("PYTHONEXECUTABLE", None)
def _detect_venv_dir() -> Path | None:
"""Detect the active virtualenv directory.
@ -3340,169 +3614,342 @@ def _launchd_fallback_to_detached(reason: str, *, exit_on_failure: bool = True)
return False
def generate_launchd_plist() -> str:
python_path = get_python_path()
# Stable cwd anchor — never the volatile source checkout. See
# _stable_service_working_dir() for the rationale (same rot risk applies
# to launchd's WorkingDirectory as to systemd's).
working_dir = _stable_service_working_dir()
hermes_home = str(get_hermes_home().resolve())
def _launchd_target() -> str:
return f"{_launchd_domain()}/{get_launchd_label()}"
def _launchd_reload_pending_path(plist_path: Path | None = None) -> Path:
"""Return the marker path used when a launchd plist reload was deferred."""
path = plist_path or get_launchd_plist_path()
return path.with_name(f"{path.name}.reload-pending")
def _mark_launchd_reload_pending(plist_path: Path | None = None) -> None:
marker = _launchd_reload_pending_path(plist_path)
marker.parent.mkdir(parents=True, exist_ok=True)
marker.write_text("launchd plist reload pending\n", encoding="utf-8")
def _clear_launchd_reload_pending(plist_path: Path | None = None) -> None:
marker = _launchd_reload_pending_path(plist_path)
try:
marker.unlink()
except FileNotFoundError:
pass
def _launchd_reload_is_pending(plist_path: Path | None = None) -> bool:
return _launchd_reload_pending_path(plist_path).exists()
def _launchd_loaded_job_pid() -> int | None:
"""Return the PID launchd has loaded for this gateway label, if known."""
try:
result = subprocess.run(
["launchctl", "print", _launchd_target()],
capture_output=True,
text=True,
timeout=5,
)
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
return None
if result.returncode != 0:
return None
for line in (result.stdout or "").splitlines():
key, sep, value = line.strip().partition("=")
if sep and key.strip().lower() == "pid":
try:
pid = int(value.strip().rstrip(";"))
except ValueError:
return None
return pid if pid > 0 else None
return None
def _reload_launchd_plist_now(plist_path: Path, label: str | None = None, *, check_bootstrap: bool = False) -> bool:
"""Force launchd to re-read *plist_path* with bootout/bootstrap."""
target = f"{_launchd_domain()}/{label or get_launchd_label()}"
subprocess.run(["launchctl", "bootout", target], check=False, timeout=90)
result = subprocess.run(
["launchctl", "bootstrap", _launchd_domain(), str(plist_path)],
check=check_bootstrap,
timeout=30,
)
if result.returncode == 0:
_clear_launchd_reload_pending(plist_path)
return True
print(f"⚠ launchd bootstrap failed with exit {result.returncode}; reload marker preserved")
return False
def _is_running_inside_gateway_process_tree() -> bool:
"""Return True when this command is executing under the live gateway process.
launchd ``bootout``/``kickstart -k`` kills the gateway process tree. When a
tool subprocess invokes service management from inside that same tree, doing
the reload inline can kill the command before it has a chance to bootstrap
the updated plist again.
"""
try:
from gateway.status import get_running_pid
pid = get_running_pid()
except Exception:
return False
if pid is None or not _is_pid_ancestor_of_current_process(pid):
return False
if is_macos():
launchd_pid = _launchd_loaded_job_pid()
if launchd_pid is not None and launchd_pid != pid:
return False
return True
def _launchd_path_entry_exists(path: str) -> bool:
"""Return True when a PATH entry exists and can be inherited by launchd."""
try:
return Path(path).exists()
except (OSError, ValueError):
return False
def _launchd_logical_hermes_path(path: str | Path | None) -> Path | None:
"""Map physical paths under HERMES_HOME back to the configured logical home."""
if path is None:
return None
hermes_home = get_hermes_home()
candidate = Path(path).expanduser()
candidate_paths = [candidate]
if not candidate.is_absolute():
candidate_paths.append(candidate.resolve())
try:
candidate_paths.append(candidate.resolve())
except OSError:
pass
home_paths = [hermes_home]
try:
home_paths.append(hermes_home.resolve())
except OSError:
pass
for candidate_path in candidate_paths:
for home_path in home_paths:
try:
relative = candidate_path.relative_to(home_path)
except ValueError:
continue
return hermes_home / relative
return candidate
def _launchd_should_keep_inherited_path_entry(path: str) -> bool:
"""Return True for inherited PATH entries worth preserving in launchd."""
entry = path.strip()
if not entry:
return False
lowered = entry.lower()
stale_markers = (
"/.codex/tmp/",
"/codex.system/bootstrap/",
"/applications/codex.app/",
"/opt/pkg/env/",
"/opt/pmk/env/",
)
if any(marker in lowered for marker in stale_markers):
return False
if entry == "/config" or entry.startswith("/config/"):
return False
if entry == "/share" or entry.startswith("/share/"):
return False
return _launchd_path_entry_exists(entry)
def _append_launchd_path_entry(entries: list[str], path: str | Path | None) -> None:
"""Append an existing launchd PATH entry once, preserving order."""
if path is None:
return
entry = str(path)
if _launchd_path_entry_exists(entry) and entry not in entries:
entries.append(entry)
def generate_launchd_plist(app_wrapper: bool = False) -> str:
detected_venv = _detect_venv_dir()
launchd_project_root = _launchd_logical_hermes_path(PROJECT_ROOT) or PROJECT_ROOT
launchd_venv = _launchd_logical_hermes_path(detected_venv)
python_path = (
str(get_launchd_app_wrapper_executable_path())
if app_wrapper
else str(_launchd_logical_hermes_path(get_python_path()) or get_python_path())
)
# Preserve the configured logical home in the launchd environment. If
# HERMES_HOME points at a symlinked or synthetic path, resolving it here
# would silently replace the user-facing runtime path with its physical
# target.
hermes_home_path = get_hermes_home()
hermes_home = str(_launchd_logical_hermes_path(hermes_home_path) or hermes_home_path)
# Stable cwd anchor — prefer logical HERMES_HOME over the source checkout so
# launchd does not depend on a volatile worktree path.
working_dir = hermes_home if _launchd_path_entry_exists(hermes_home) else str(launchd_project_root)
log_dir = get_hermes_home() / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
label = get_launchd_label()
profile_arg = _profile_arg(hermes_home)
# Build a sane PATH for the launchd plist. launchd provides only a
# minimal default (/usr/bin:/bin:/usr/sbin:/sbin) which misses Homebrew,
# nvm, cargo, etc. We prepend venv/bin and node_modules/.bin (matching
# the systemd unit), then capture the user's full shell PATH so every
# user-installed tool (node, ffmpeg, …) is reachable.
detected_venv = _detect_venv_dir()
venv_dir = str(detected_venv) if detected_venv else str(PROJECT_ROOT / "venv")
# Build a sane PATH for the launchd plist. launchd starts with only a
# minimal system PATH, while an interactive/dev environment may contain
# stale Codex/bootstrap entries. Prioritize Hermes' own venv and
# HERMES_HOME-local bin directories first, then keep only inherited entries
# that still exist.
venv_bin = str(launchd_venv / "bin") if launchd_venv else str(launchd_project_root / "venv" / "bin")
venv_dir = str(launchd_venv) if launchd_venv else str(launchd_project_root / "venv")
node_bin = str(launchd_project_root / "node_modules" / ".bin")
priority_dirs: list[str] = []
for candidate in (
venv_bin,
hermes_home_path / "bin",
hermes_home_path / ".go" / "bin",
node_bin,
):
_append_launchd_path_entry(priority_dirs, candidate)
inherited_dirs = [
p for p in os.environ.get("PATH", "").split(":")
if _launchd_should_keep_inherited_path_entry(p)
]
# Resolve the directory containing the node binary (e.g. Homebrew, nvm)
# so it's explicitly in PATH even if the user's shell PATH changes later.
priority_dirs = _build_service_path_dirs()
# Validate the resolved parent through the same stale-entry filter used for
# inherited PATH entries; otherwise a stale Codex/bootstrap node can sneak
# back into the launchd plist before the inherited PATH is filtered.
resolved_node = shutil.which("node")
if resolved_node:
resolved_node_dir = str(Path(resolved_node).resolve().parent)
if resolved_node_dir not in priority_dirs:
priority_dirs.append(resolved_node_dir)
sane_path = ":".join(
dict.fromkeys(
priority_dirs + [p for p in os.environ.get("PATH", "").split(":") if p]
if _launchd_should_keep_inherited_path_entry(resolved_node_dir):
_append_launchd_path_entry(priority_dirs, resolved_node_dir)
system_fallback_dirs = ["/usr/local/bin", "/usr/bin", "/bin", "/usr/sbin", "/sbin"]
system_dirs = [p for p in system_fallback_dirs if _launchd_path_entry_exists(p)]
sane_path = ":".join(dict.fromkeys(priority_dirs + inherited_dirs + system_dirs))
environment = {
"PATH": sane_path,
"VIRTUAL_ENV": venv_dir,
"HERMES_HOME": hermes_home,
}
if app_wrapper:
environment.update(
{
"PYTHONHOME": str(_python_home_from_path(get_python_path(), detected_venv)),
"PYTHONPATH": _launchd_app_wrapper_pythonpath(detected_venv),
"PYTHONEXECUTABLE": _launchd_child_python_path(detected_venv),
MACOS_APP_WRAPPER_ENV_KEY: "1",
}
)
)
# Build ProgramArguments array, including --profile when using a named profile
prog_args = [
f"<string>{python_path}</string>",
"<string>-m</string>",
"<string>hermes_cli.main</string>",
]
prog_args = [python_path, "-m", "hermes_cli.main"]
if profile_arg:
for part in profile_arg.split():
prog_args.append(f"<string>{part}</string>")
prog_args.extend(
[
"<string>gateway</string>",
"<string>run</string>",
"<string>--replace</string>",
]
)
prog_args_xml = "\n ".join(prog_args)
prog_args.extend(profile_arg.split())
prog_args.extend(["gateway", "run", "--replace"])
return f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>{label}</string>
plist_data = {
"Label": label,
"ProgramArguments": prog_args,
"WorkingDirectory": working_dir,
"EnvironmentVariables": environment,
"LimitLoadToSessionType": ["Aqua", "Background"],
"RunAtLoad": True,
"KeepAlive": True,
"StandardOutPath": f"{log_dir}/gateway.log",
"StandardErrorPath": f"{log_dir}/gateway.error.log",
}
if app_wrapper:
plist_data["AssociatedBundleIdentifiers"] = get_launchd_bundle_identifier()
<key>ProgramArguments</key>
<array>
{prog_args_xml}
</array>
<key>WorkingDirectory</key>
<string>{working_dir}</string>
<key>EnvironmentVariables</key>
<dict>
<key>PATH</key>
<string>{sane_path}</string>
<key>VIRTUAL_ENV</key>
<string>{venv_dir}</string>
<key>HERMES_HOME</key>
<string>{hermes_home}</string>
</dict>
<key>LimitLoadToSessionType</key>
<array>
<string>Aqua</string>
<string>Background</string>
</array>
<key>RunAtLoad</key>
<true/>
<key>KeepAlive</key>
<true/>
<key>StandardOutPath</key>
<string>{log_dir}/gateway.log</string>
<key>StandardErrorPath</key>
<string>{log_dir}/gateway.error.log</string>
</dict>
</plist>
"""
return plistlib.dumps(plist_data, sort_keys=False).decode("utf-8")
def launchd_plist_is_current() -> bool:
def launchd_plist_is_current(app_wrapper: bool | None = None) -> bool:
"""Check if the installed launchd plist matches the currently generated one."""
plist_path = get_launchd_plist_path()
if not plist_path.exists():
return False
target_app_wrapper = _resolve_launchd_app_wrapper_mode(app_wrapper)
if target_app_wrapper and not launchd_app_wrapper_is_current():
return False
installed = plist_path.read_text(encoding="utf-8")
expected = generate_launchd_plist()
return _normalize_launchd_plist_for_comparison(
installed
) == _normalize_launchd_plist_for_comparison(expected)
expected = generate_launchd_plist(app_wrapper=target_app_wrapper)
return _normalize_launchd_plist_for_comparison(installed) == _normalize_launchd_plist_for_comparison(expected)
def refresh_launchd_plist_if_needed() -> bool:
"""Rewrite the installed launchd plist when the generated definition has changed.
Unlike systemd, launchd picks up plist changes on the next ``launchctl kill``/
``launchctl kickstart`` cycle no daemon-reload is needed. We still bootout/
bootstrap to make launchd re-read the updated plist immediately.
"""
def refresh_launchd_plist_if_needed(app_wrapper: bool | None = None) -> bool:
"""Rewrite/reload the installed launchd plist when its definition changed."""
plist_path = get_launchd_plist_path()
if not plist_path.exists() or launchd_plist_is_current():
if not plist_path.exists():
return False
new_plist = generate_launchd_plist()
if _refuse_temp_home_service_write(new_plist, "launchd plist"):
return False
target_app_wrapper = _resolve_launchd_app_wrapper_mode(app_wrapper)
if target_app_wrapper and not launchd_app_wrapper_is_current():
install_launchd_app_wrapper(force=True)
plist_path.write_text(new_plist, encoding="utf-8")
label = get_launchd_label()
# Bootout/bootstrap so launchd picks up the new definition
subprocess.run(
["launchctl", "bootout", f"{_launchd_domain()}/{label}"],
check=False,
timeout=90,
)
subprocess.run(
["launchctl", "bootstrap", _launchd_domain(), str(plist_path)],
check=False,
timeout=30,
)
print(
"↻ Updated gateway launchd service definition to match the current Hermes install"
)
reload_pending = _launchd_reload_is_pending(plist_path)
plist_current = launchd_plist_is_current(app_wrapper=target_app_wrapper)
if plist_current and not reload_pending:
return False
if not plist_current:
new_plist = generate_launchd_plist(app_wrapper=target_app_wrapper)
if _refuse_temp_home_service_write(new_plist, "launchd plist"):
return False
plist_path.write_text(new_plist, encoding="utf-8")
reload_pending = True
if _is_running_inside_gateway_process_tree():
if reload_pending:
_mark_launchd_reload_pending(plist_path)
print("↻ Updated gateway launchd service definition to match the current Hermes install")
print("⚠ launchd reload deferred because this command is running inside the gateway process tree")
print(" Run 'hermes gateway start' from an external shell to make launchd re-read the plist")
return True
# Bootout/bootstrap so launchd picks up the new definition. This also
# consumes reload-pending markers from previous in-gateway refreshes.
_reload_launchd_plist_now(plist_path, label)
print("↻ Updated gateway launchd service definition to match the current Hermes install")
return True
def launchd_install(force: bool = False):
def launchd_install(force: bool = False, app_wrapper: bool = False):
plist_path = get_launchd_plist_path()
target_app_wrapper = app_wrapper or (
plist_path.exists() and not force and _installed_launchd_plist_uses_app_wrapper(plist_path)
)
if target_app_wrapper:
install_launchd_app_wrapper(force=force or not launchd_app_wrapper_is_current())
if plist_path.exists() and not force:
if not launchd_plist_is_current():
if not launchd_plist_is_current(app_wrapper=target_app_wrapper):
print(f"↻ Repairing outdated launchd service at: {plist_path}")
refresh_launchd_plist_if_needed()
refresh_launchd_plist_if_needed(app_wrapper=target_app_wrapper)
print("✓ Service definition updated")
return
print(f"Service already installed at: {plist_path}")
if target_app_wrapper:
print(f"Using macOS app wrapper: {get_launchd_app_wrapper_path()}")
print("Use --force to reinstall")
return
plist_path.parent.mkdir(parents=True, exist_ok=True)
new_plist = generate_launchd_plist()
new_plist = generate_launchd_plist(app_wrapper=target_app_wrapper)
if _refuse_temp_home_service_write(new_plist, "launchd plist"):
return
print(f"Installing launchd service to: {plist_path}")
plist_path.write_text(new_plist)
if target_app_wrapper:
print(f"Using macOS app wrapper: {get_launchd_app_wrapper_path()}")
plist_path.write_text(new_plist, encoding="utf-8")
try:
subprocess.run(
@ -3554,12 +4001,13 @@ def launchd_start():
print("↻ launchd plist missing; regenerating service definition")
plist_path.parent.mkdir(parents=True, exist_ok=True)
plist_path.write_text(new_plist, encoding="utf-8")
if _is_running_inside_gateway_process_tree():
_mark_launchd_reload_pending(plist_path)
print("⚠ launchd bootstrap deferred because this command is running inside the gateway process tree")
print(" Run 'hermes gateway start' from an external shell to load the regenerated plist")
return
try:
subprocess.run(
["launchctl", "bootstrap", _launchd_domain(), str(plist_path)],
check=True,
timeout=30,
)
_reload_launchd_plist_now(plist_path, label, check_bootstrap=True)
subprocess.run(
["launchctl", "kickstart", f"{_launchd_domain()}/{label}"],
check=True,
@ -3573,7 +4021,13 @@ def launchd_start():
print("✓ Service started")
return
refresh_launchd_plist_if_needed()
refreshed = refresh_launchd_plist_if_needed()
if _is_running_inside_gateway_process_tree():
if not refreshed:
print("✓ Gateway service is already running")
print("⚠ Not kickstarting launchd from inside the gateway process tree")
print(" Run 'hermes gateway start' from an external shell if launchd must be reloaded")
return
try:
subprocess.run(
["launchctl", "kickstart", f"{_launchd_domain()}/{label}"],
@ -3690,6 +4144,8 @@ def _wait_for_gateway_exit(
def launchd_restart():
if _installed_launchd_plist_uses_app_wrapper() and not launchd_app_wrapper_is_current():
install_launchd_app_wrapper(force=True)
label = get_launchd_label()
target = f"{_launchd_domain()}/{label}"
drain_timeout = _get_restart_drain_timeout()
@ -3934,6 +4390,8 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo
_guard_existing_gateway_process_conflict(replace=replace)
sys.path.insert(0, str(PROJECT_ROOT))
_drop_launchd_app_wrapper_python_env()
# Detached Windows gateway runs must ignore console-control broadcasts
# from sibling CLI processes, but foreground `hermes gateway run` still
# needs to obey the banner's "Press Ctrl+C to stop" contract.
@ -6493,6 +6951,10 @@ def _gateway_command_inner(args):
force = getattr(args, "force", False)
system = getattr(args, "system", False)
run_as_user = getattr(args, "run_as_user", None)
macos_app_wrapper = getattr(args, "macos_app_wrapper", False)
if macos_app_wrapper and not is_macos():
print_error("--macos-app-wrapper is only supported on macOS launchd services")
sys.exit(1)
if is_termux():
print("Gateway service installation is not supported on Termux.")
print("Run manually: hermes gateway")
@ -6520,7 +6982,7 @@ def _gateway_command_inner(args):
if start_now:
systemd_start(system=system)
elif is_macos():
launchd_install(force)
launchd_install(force=force, app_wrapper=macos_app_wrapper)
elif is_windows():
from hermes_cli import gateway_windows

View file

@ -194,6 +194,15 @@ def build_gateway_parser(subparsers, *, cmd_gateway: Callable, cmd_proxy: Callab
action="store_true",
help=argparse.SUPPRESS,
)
gateway_install.add_argument(
"--macos-app-wrapper",
dest="macos_app_wrapper",
action="store_true",
help=(
"macOS only: run launchd through a Hermes Agent.app wrapper so "
"privacy prompts show Hermes instead of python"
),
)
# gateway uninstall
gateway_uninstall = gateway_subparsers.add_parser(

View file

@ -1,6 +1,7 @@
"""Tests for gateway service management helpers."""
import os
import plistlib
import subprocess
from pathlib import Path
from types import SimpleNamespace
@ -68,6 +69,7 @@ class TestSystemdServiceRefresh:
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kwargs: None)
calls = []
@ -91,6 +93,7 @@ class TestSystemdServiceRefresh:
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kwargs: None)
calls = []
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
@ -549,6 +552,407 @@ class TestGatewayStopCleanup:
assert kill_calls == [False]
class TestLaunchdMacOSAppWrapper:
def test_generate_launchd_plist_prioritizes_hermes_home_bins_and_filters_stale_path_entries(self, tmp_path, monkeypatch):
home = tmp_path / "hermes-home"
repo = home / "hermes-agent"
venv = repo / ".venv"
node_dir = tmp_path / "homebrew" / "bin"
stale_codex = tmp_path / ".codex" / "tmp" / "arg0" / "codex-arg0dead"
codex_app = tmp_path / "Applications" / "Codex.app" / "Contents" / "Resources"
missing_dir = tmp_path / "missing" / "bin"
for path in [
venv / "bin",
home / "bin",
home / ".go" / "bin",
repo / "node_modules" / ".bin",
node_dir,
codex_app,
]:
path.mkdir(parents=True)
(venv / "bin" / "python").write_text("venv-python", encoding="utf-8")
(node_dir / "node").write_text("node", encoding="utf-8")
real_path_exists = Path.exists
def fake_path_exists(path):
if str(path) in {"/config/hermes/bin", "/config/.go/bin", "/share/hermes/bin"}:
return True
return real_path_exists(path)
monkeypatch.setattr(Path, "exists", fake_path_exists)
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: venv)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(venv / "bin" / "python"))
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: str(node_dir / "node") if cmd == "node" else None)
monkeypatch.setenv(
"PATH",
":".join([
str(stale_codex),
str(codex_app),
str(missing_dir),
"/config/hermes/bin",
"/config/.go/bin",
"/share/hermes/bin",
str(node_dir),
str(home / "bin"),
]),
)
plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode("utf-8"))
path_parts = plist["EnvironmentVariables"]["PATH"].split(":")
expected_prefix = [
str(venv / "bin"),
str(home / "bin"),
str(home / ".go" / "bin"),
str(repo / "node_modules" / ".bin"),
str(node_dir),
]
assert path_parts[: len(expected_prefix)] == expected_prefix
assert str(stale_codex) not in path_parts
assert str(codex_app) not in path_parts
assert str(missing_dir) not in path_parts
assert path_parts.count(str(home / "bin")) == 1
assert "/config/hermes/bin" not in path_parts
assert "/config/.go/bin" not in path_parts
assert "/share/hermes/bin" not in path_parts
assert not any(part.startswith("/share") for part in path_parts)
def test_generate_launchd_plist_preserves_logical_hermes_home(self, tmp_path, monkeypatch):
physical_home = tmp_path / "physical-home"
logical_home = tmp_path / "logical-home"
physical_home.mkdir()
logical_home.symlink_to(physical_home, target_is_directory=True)
physical_repo = physical_home / "hermes-agent"
logical_repo = logical_home / "hermes-agent"
physical_venv = physical_repo / ".venv"
logical_venv = logical_repo / ".venv"
node_bin = physical_repo / "node_modules" / ".bin"
(physical_venv / "bin").mkdir(parents=True)
(physical_venv / "bin" / "python").write_text("venv-python", encoding="utf-8")
node_bin.mkdir(parents=True)
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: logical_home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", physical_repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: physical_venv)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(physical_venv / "bin" / "python"))
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode("utf-8"))
env = plist["EnvironmentVariables"]
path_parts = env["PATH"].split(":")
assert env["HERMES_HOME"] == str(logical_home)
assert env["HERMES_HOME"] != str(physical_home.resolve())
assert env["VIRTUAL_ENV"] == str(logical_venv)
assert plist["WorkingDirectory"] == str(logical_home)
assert plist["ProgramArguments"][0] == str(logical_venv / "bin" / "python")
assert str(logical_repo / "node_modules" / ".bin") in path_parts
assert not any(str(physical_home) in part for part in path_parts)
def test_generate_launchd_plist_rejects_stale_resolved_node_parent(self, tmp_path, monkeypatch):
home = tmp_path / "hermes-home"
repo = home / "hermes-agent"
stale_codex = tmp_path / ".codex" / "tmp" / "arg0" / "codex-arg0dead"
stale_codex.mkdir(parents=True)
(stale_codex / "node").write_text("stale-node", encoding="utf-8")
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: None)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: "/usr/bin/python3")
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: str(stale_codex / "node") if cmd == "node" else None)
monkeypatch.setenv("PATH", str(stale_codex))
plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode("utf-8"))
path_parts = plist["EnvironmentVariables"]["PATH"].split(":")
assert str(stale_codex) not in path_parts
def test_generate_launchd_plist_keeps_system_path_fallback_when_inherited_path_is_empty(self, tmp_path, monkeypatch):
home = tmp_path / "hermes-home"
repo = home / "hermes-agent"
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: None)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: "/usr/bin/python3")
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
monkeypatch.setenv("PATH", "")
plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode("utf-8"))
path_parts = plist["EnvironmentVariables"]["PATH"].split(":")
assert "/usr/bin" in path_parts
assert "/bin" in path_parts
def test_generate_launchd_plist_with_app_wrapper_uses_named_bundle_executable(self, tmp_path, monkeypatch):
home = tmp_path / "home"
repo = tmp_path / "repo"
venv = repo / ".venv"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = python_home / "bin" / "python3.13"
source_python.parent.mkdir(parents=True)
source_python.write_text("python", encoding="utf-8")
(venv / "bin").mkdir(parents=True)
(venv / "bin" / "python").write_text("venv-python", encoding="utf-8")
(venv / "lib" / "python3.13" / "site-packages").mkdir(parents=True)
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: venv)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
plist = plistlib.loads(gateway_cli.generate_launchd_plist(app_wrapper=True).encode("utf-8"))
app_exe = home / "macos" / "Hermes Agent.app" / "Contents" / "MacOS" / "Hermes Agent"
assert plist["ProgramArguments"][:3] == [str(app_exe), "-m", "hermes_cli.main"]
assert plist["AssociatedBundleIdentifiers"] == "ai.hermes.gateway"
env = plist["EnvironmentVariables"]
assert env["PYTHONHOME"] == str(python_home)
assert env["PYTHONEXECUTABLE"] == str(venv / "bin" / "python")
assert str(venv / "lib" / "python3.13" / "site-packages") in env["PYTHONPATH"].split(":")
assert str(repo) in env["PYTHONPATH"].split(":")
def test_generate_launchd_plist_with_app_wrapper_uses_pyvenv_home_for_copy_venvs(self, tmp_path, monkeypatch):
home = tmp_path / "home"
repo = tmp_path / "repo"
venv = repo / ".venv"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = venv / "bin" / "python"
source_python.parent.mkdir(parents=True)
source_python.write_text("copied-python", encoding="utf-8")
(venv / "lib" / "python3.13" / "site-packages").mkdir(parents=True)
(venv / "pyvenv.cfg").write_text(f"home = {python_home / 'bin'}\n", encoding="utf-8")
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: venv)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
plist = plistlib.loads(gateway_cli.generate_launchd_plist(app_wrapper=True).encode("utf-8"))
assert plist["EnvironmentVariables"]["PYTHONHOME"] == str(python_home)
def test_install_launchd_app_wrapper_copies_python_and_writes_bundle_metadata(self, tmp_path, monkeypatch):
home = tmp_path / "home"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = python_home / "bin" / "python3.13"
source_python.parent.mkdir(parents=True)
source_python.write_bytes(b"fake-macho-python")
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
app_path = gateway_cli.install_launchd_app_wrapper(force=True)
app_exe = app_path / "Contents" / "MacOS" / "Hermes Agent"
info = plistlib.loads((app_path / "Contents" / "Info.plist").read_bytes())
assert app_path == home / "macos" / "Hermes Agent.app"
assert app_exe.read_bytes() == b"fake-macho-python"
assert info["CFBundleIdentifier"] == "ai.hermes.gateway"
assert info["CFBundleDisplayName"] == "Hermes Agent"
assert gateway_cli.launchd_app_wrapper_is_current() is True
assert any(cmd[:5] == ["codesign", "--force", "--deep", "--sign", "-"] for cmd in calls)
assert any(cmd[:4] == ["codesign", "--verify", "--deep", "--strict"] for cmd in calls)
assert any(cmd[1:2] == ["-c"] and Path(cmd[0]).name == "Hermes Agent" for cmd in calls)
def test_install_launchd_app_wrapper_keeps_existing_bundle_when_validation_fails(self, tmp_path, monkeypatch):
home = tmp_path / "home"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = python_home / "bin" / "python3.13"
source_python.parent.mkdir(parents=True)
source_python.write_bytes(b"new-python")
app_exe = home / "macos" / "Hermes Agent.app" / "Contents" / "MacOS" / "Hermes Agent"
app_exe.parent.mkdir(parents=True)
app_exe.write_bytes(b"old-python")
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
def fail_codesign(cmd, **kwargs):
if cmd[:2] == ["codesign", "--force"]:
raise gateway_cli.subprocess.CalledProcessError(1, cmd, stderr="sign failed")
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fail_codesign)
with pytest.raises(gateway_cli.subprocess.CalledProcessError):
gateway_cli.install_launchd_app_wrapper(force=True)
assert app_exe.read_bytes() == b"old-python"
def test_launchd_app_wrapper_current_survives_codesign_binary_mutation(self, tmp_path, monkeypatch):
home = tmp_path / "home"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = python_home / "bin" / "python3.13"
source_python.parent.mkdir(parents=True)
source_python.write_bytes(b"fake-macho-python")
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
def fake_codesign(cmd, **kwargs):
if cmd and cmd[0] == "codesign":
target_app = Path(cmd[-1])
app_exe = target_app / "Contents" / "MacOS" / "Hermes Agent"
app_exe.write_bytes(app_exe.read_bytes() + b"-signed")
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_codesign)
gateway_cli.install_launchd_app_wrapper(force=True)
assert gateway_cli.launchd_app_wrapper_is_current() is True
def test_launchd_app_wrapper_current_requires_valid_codesign_verification(self, tmp_path, monkeypatch):
home = tmp_path / "home"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = python_home / "bin" / "python3.13"
source_python.parent.mkdir(parents=True)
source_python.write_bytes(b"fake-macho-python")
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
monkeypatch.setattr(
gateway_cli.subprocess,
"run",
lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="", stderr=""),
)
gateway_cli.install_launchd_app_wrapper(force=True)
monkeypatch.setattr(
gateway_cli.subprocess,
"run",
lambda *args, **kwargs: SimpleNamespace(returncode=1, stdout="", stderr="invalid signature"),
)
assert gateway_cli.launchd_app_wrapper_is_current() is False
def test_launchd_plist_is_stale_when_app_wrapper_bundle_missing(self, tmp_path, monkeypatch):
home = tmp_path / "home"
plist_path = tmp_path / "ai.hermes.gateway.plist"
app_exe = home / "macos" / "Hermes Agent.app" / "Contents" / "MacOS" / "Hermes Agent"
plist_path.write_text(
plistlib.dumps(
{
"Label": "ai.hermes.gateway",
"ProgramArguments": [str(app_exe), "-m", "hermes_cli.main", "gateway", "run", "--replace"],
"EnvironmentVariables": {"HERMES_LAUNCHD_APP_WRAPPER": "1"},
}
).decode("utf-8"),
encoding="utf-8",
)
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
assert gateway_cli.launchd_plist_is_current() is False
def test_drop_launchd_app_wrapper_python_env_removes_runtime_overrides(self, monkeypatch):
monkeypatch.setenv("HERMES_LAUNCHD_APP_WRAPPER", "1")
monkeypatch.setenv("PYTHONHOME", "/private/hermes-python")
monkeypatch.setenv("PYTHONPATH", "/private/hermes-site")
monkeypatch.setenv("PYTHONEXECUTABLE", "/private/hermes-venv/bin/python")
gateway_cli._drop_launchd_app_wrapper_python_env()
assert "PYTHONHOME" not in os.environ
assert "PYTHONPATH" not in os.environ
assert "PYTHONEXECUTABLE" not in os.environ
def test_refresh_launchd_plist_preserves_existing_app_wrapper_mode(self, tmp_path, monkeypatch):
home = tmp_path / "home"
repo = tmp_path / "repo"
venv = repo / ".venv"
python_home = tmp_path / "cpython-3.13.13-macos-aarch64-none"
source_python = python_home / "bin" / "python3.13"
source_python.parent.mkdir(parents=True)
source_python.write_bytes(b"fake-macho-python")
(venv / "bin").mkdir(parents=True)
(venv / "lib" / "python3.13" / "site-packages").mkdir(parents=True)
plist_path = tmp_path / "ai.hermes.gateway.plist"
app_exe = home / "macos" / "Hermes Agent.app" / "Contents" / "MacOS" / "Hermes Agent"
plist_path.write_text(
plistlib.dumps(
{
"Label": "ai.hermes.gateway",
"ProgramArguments": [str(app_exe), "-m", "hermes_cli.main", "gateway", "run", "--replace"],
"EnvironmentVariables": {"HERMES_LAUNCHD_APP_WRAPPER": "1"},
}
).decode("utf-8"),
encoding="utf-8",
)
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
monkeypatch.setattr(gateway_cli, "_profile_suffix", lambda: "")
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", repo)
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: venv)
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(source_python))
def fake_generate_launchd_plist(app_wrapper=False):
env = {"HERMES_HOME": "/Users/alice/.hermes"}
program = str(app_exe) if app_wrapper else str(source_python)
if app_wrapper:
env["HERMES_LAUNCHD_APP_WRAPPER"] = "1"
return plistlib.dumps(
{
"Label": "ai.hermes.gateway",
"ProgramArguments": [
program,
"-m",
"hermes_cli.main",
"gateway",
"run",
"--replace",
],
"EnvironmentVariables": env,
},
sort_keys=False,
).decode("utf-8")
monkeypatch.setattr(gateway_cli, "generate_launchd_plist", fake_generate_launchd_plist)
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
assert gateway_cli.refresh_launchd_plist_if_needed() is True
refreshed = plistlib.loads(plist_path.read_bytes())
assert refreshed["ProgramArguments"][0] == str(app_exe)
assert refreshed["EnvironmentVariables"]["HERMES_LAUNCHD_APP_WRAPPER"] == "1"
assert (home / "macos" / "Hermes Agent.app" / "Contents" / "MacOS" / "Hermes Agent").exists()
class TestLaunchdServiceRecovery:
def test_get_restart_drain_timeout_prefers_env_then_config_then_default(self, monkeypatch):
monkeypatch.delenv("HERMES_RESTART_DRAIN_TIMEOUT", raising=False)
@ -586,7 +990,7 @@ class TestLaunchdServiceRecovery:
monkeypatch.setattr(
gateway_cli,
"generate_launchd_plist",
lambda: (
lambda app_wrapper=False: (
"<plist>--replace\n<key>HERMES_HOME</key>"
"<string>/Users/alice/.hermes</string></plist>"
),
@ -613,6 +1017,112 @@ class TestLaunchdServiceRecovery:
["launchctl", "bootstrap", domain, str(plist_path)],
]
def test_refresh_launchd_plist_defers_launchctl_when_running_inside_gateway_process_tree(self, tmp_path, monkeypatch):
plist_path = tmp_path / "ai.hermes.gateway.plist"
plist_path.write_text("old plist\n", encoding="utf-8")
expected = plistlib.dumps({"Label": "ai.hermes.gateway", "ProgramArguments": ["python"]}).decode("utf-8")
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
monkeypatch.setattr(gateway_cli, "generate_launchd_plist", lambda app_wrapper=False: expected)
monkeypatch.setattr(gateway_cli, "_is_running_inside_gateway_process_tree", lambda: True, raising=False)
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
assert gateway_cli.refresh_launchd_plist_if_needed() is True
assert plist_path.read_text(encoding="utf-8") == expected
assert calls == []
def test_launchd_start_defers_kickstart_after_self_context_refresh(self, tmp_path, monkeypatch):
plist_path = tmp_path / "ai.hermes.gateway.plist"
plist_path.write_text("old plist\n", encoding="utf-8")
expected = plistlib.dumps({"Label": "ai.hermes.gateway", "ProgramArguments": ["python"]}).decode("utf-8")
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
monkeypatch.setattr(gateway_cli, "generate_launchd_plist", lambda app_wrapper=False: expected)
monkeypatch.setattr(gateway_cli, "_is_running_inside_gateway_process_tree", lambda: True, raising=False)
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
gateway_cli.launchd_start()
assert plist_path.read_text(encoding="utf-8") == expected
assert calls == []
def test_launchd_start_reloads_pending_definition_from_external_shell(self, tmp_path, monkeypatch):
plist_path = tmp_path / "ai.hermes.gateway.plist"
plist_path.write_text("old plist\n", encoding="utf-8")
expected = plistlib.dumps({"Label": "ai.hermes.gateway", "ProgramArguments": ["python"]}).decode("utf-8")
label = gateway_cli.get_launchd_label()
domain = gateway_cli._launchd_domain()
target = f"{domain}/{label}"
inside_gateway = True
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
monkeypatch.setattr(gateway_cli, "generate_launchd_plist", lambda app_wrapper=False: expected)
monkeypatch.setattr(gateway_cli, "_is_running_inside_gateway_process_tree", lambda: inside_gateway, raising=False)
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
gateway_cli.launchd_start()
assert calls == []
inside_gateway = False
gateway_cli.launchd_start()
assert calls == [
["launchctl", "bootout", target],
["launchctl", "bootstrap", domain, str(plist_path)],
["launchctl", "kickstart", target],
]
def test_launchd_start_defers_missing_plist_bootstrap_inside_gateway_tree(self, tmp_path, monkeypatch):
plist_path = tmp_path / "ai.hermes.gateway.plist"
expected = plistlib.dumps({"Label": "ai.hermes.gateway", "ProgramArguments": ["python"]}).decode("utf-8")
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
monkeypatch.setattr(gateway_cli, "generate_launchd_plist", lambda app_wrapper=False: expected)
monkeypatch.setattr(gateway_cli, "_is_running_inside_gateway_process_tree", lambda: True, raising=False)
calls = []
def fake_run(cmd, **kwargs):
calls.append(cmd)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
gateway_cli.launchd_start()
assert plist_path.read_text(encoding="utf-8") == expected
assert calls == []
def test_running_inside_gateway_process_tree_requires_matching_launchd_job_pid(self, monkeypatch):
monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: 123)
monkeypatch.setattr(gateway_cli, "_is_pid_ancestor_of_current_process", lambda pid: pid == 123)
def fake_run(cmd, **kwargs):
assert cmd == ["launchctl", "print", f"{gateway_cli._launchd_domain()}/{gateway_cli.get_launchd_label()}"]
return SimpleNamespace(returncode=0, stdout="pid = 999\n", stderr="")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
assert gateway_cli._is_running_inside_gateway_process_tree() is False
def test_launchd_start_reloads_unloaded_job_and_retries(self, tmp_path, monkeypatch):
plist_path = tmp_path / "ai.hermes.gateway.plist"
plist_path.write_text(gateway_cli.generate_launchd_plist(), encoding="utf-8")
@ -892,7 +1402,7 @@ class TestLaunchdServiceRecovery:
monkeypatch.setattr(
gateway_cli,
"generate_launchd_plist",
lambda: (
lambda app_wrapper=False: (
"<plist><key>HERMES_HOME</key>"
"<string>/Users/alice/.hermes</string></plist>"
),
@ -1143,6 +1653,7 @@ class TestGatewaySystemServiceRouting:
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kwargs: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: calls.append(("refresh", system)))
monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 12.0)
monkeypatch.setattr(
@ -1188,6 +1699,7 @@ class TestGatewaySystemServiceRouting:
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kwargs: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 10.0)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
@ -1247,6 +1759,7 @@ class TestGatewaySystemServiceRouting:
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kwargs: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
monkeypatch.setattr(gateway_cli, "_recover_pending_systemd_restart", lambda system=False, previous_pid=None: False)
@ -1277,6 +1790,7 @@ class TestGatewaySystemServiceRouting:
def test_systemd_restart_recovers_failed_planned_restart(self, monkeypatch, capsys):
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "_preflight_user_systemd", lambda **kwargs: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
monkeypatch.setattr(
"gateway.status.read_runtime_status",
@ -2069,10 +2583,8 @@ class TestProfileArg:
def test_launchd_plist_supports_aqua_and_background_sessions(self):
# macOS 26+ only loads the agent in non-Aqua sessions when the plist
# opts into Background as well (issue #23387).
plist = gateway_cli.generate_launchd_plist()
assert "<key>LimitLoadToSessionType</key>" in plist
assert "<string>Aqua</string>" in plist
assert "<string>Background</string>" in plist
plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode("utf-8"))
assert plist["LimitLoadToSessionType"] == ["Aqua", "Background"]
def test_launchd_plist_path_uses_real_user_home_not_profile_home(self, tmp_path, monkeypatch):
profile_dir = tmp_path / ".hermes" / "profiles" / "orcha"
@ -3036,12 +3548,7 @@ class TestServiceWorkingDirIsStable:
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
plist = gateway_cli.generate_launchd_plist()
plist = plistlib.loads(gateway_cli.generate_launchd_plist().encode("utf-8"))
# Scalar <true/> must be present immediately after the KeepAlive key
assert "<key>KeepAlive</key>" in plist
# The unconditional form
assert "<key>KeepAlive</key>\n <true/>" in plist
# The old conditional dict form must NOT appear
assert "SuccessfulExit" not in plist
assert "<key>KeepAlive</key>\n <dict>" not in plist
assert plist["KeepAlive"] is True
assert "SuccessfulExit" not in str(plist.get("KeepAlive"))