e97c0cb578
* feat: GPT tool-use steering + strip budget warnings from history
Two changes to improve tool reliability, especially for OpenAI GPT models:
1. GPT tool-use enforcement prompt: Adds GPT_TOOL_USE_GUIDANCE to the
system prompt when the model name contains 'gpt' and tools are loaded.
This addresses a known behavioral pattern where GPT models describe
intended actions ('I will run the tests') instead of actually making
tool calls. Inspired by similar steering in OpenCode (beast.txt) and
Cline (GPT-5.1 variant).
2. Budget warning history stripping: Budget pressure warnings injected by
_get_budget_warning() into tool results are now stripped when
conversation history is replayed via run_conversation(). Previously,
these turn-scoped signals persisted across turns, causing models to
avoid tool calls in all subsequent messages after any turn that hit
the 70-90% iteration threshold.
* fix: replace hardcoded ~/.hermes paths with get_hermes_home() for profile support
Prep for the upcoming profiles feature — each profile is a separate
HERMES_HOME directory, so all paths must respect the env var.
Fixes:
- gateway/platforms/matrix.py: Matrix E2EE store was hardcoded to
~/.hermes/matrix/store, ignoring HERMES_HOME. Now uses
get_hermes_home() so each profile gets its own Matrix state.
- gateway/platforms/telegram.py: Two locations reading config.yaml via
Path.home()/.hermes instead of get_hermes_home(). DM topic thread_id
persistence and hot-reload would read the wrong config in a profile.
- tools/file_tools.py: Security path for hub index blocking was
hardcoded to ~/.hermes, would miss the actual profile's hub cache.
- hermes_cli/gateway.py: Service naming now uses the profile name
(hermes-gateway-coder) instead of a cryptic hash suffix. Extracted
_profile_suffix() helper shared by systemd and launchd.
- hermes_cli/gateway.py: Launchd plist path and Label now scoped per
profile (ai.hermes.gateway-coder.plist). Previously all profiles
would collide on the same plist file on macOS.
- hermes_cli/gateway.py: Launchd plist now includes HERMES_HOME in
EnvironmentVariables — was missing entirely, making custom
HERMES_HOME broken on macOS launchd (pre-existing bug).
- All launchctl commands in gateway.py, main.py, status.py updated
to use get_launchd_label() instead of hardcoded string.
Test fixes: DM topic tests now set HERMES_HOME env var alongside
Path.home() mock. Launchd test uses get_launchd_label() for expected
commands.
445 lines
17 KiB
Python
445 lines
17 KiB
Python
"""Tests for gateway service management helpers."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import hermes_cli.gateway as gateway_cli
|
|
|
|
|
|
class TestSystemdServiceRefresh:
|
|
def test_systemd_install_repairs_outdated_unit_without_force(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
|
|
|
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")
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=True, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.systemd_install()
|
|
|
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
|
assert calls[:2] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["systemctl", "--user", "enable", gateway_cli.get_service_name()],
|
|
]
|
|
|
|
def test_systemd_start_refreshes_outdated_unit(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
|
|
|
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")
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=True, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.systemd_start()
|
|
|
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
|
assert calls[:2] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["systemctl", "--user", "start", gateway_cli.get_service_name()],
|
|
]
|
|
|
|
def test_systemd_restart_refreshes_outdated_unit(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
|
|
|
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")
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=True, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.systemd_restart()
|
|
|
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
|
assert calls[:2] == [
|
|
["systemctl", "--user", "daemon-reload"],
|
|
["systemctl", "--user", "restart", gateway_cli.get_service_name()],
|
|
]
|
|
|
|
|
|
class TestGeneratedSystemdUnits:
|
|
def test_user_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self):
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
assert "ExecStart=" in unit
|
|
assert "ExecStop=" not in unit
|
|
assert "TimeoutStopSec=60" in unit
|
|
|
|
def test_user_unit_includes_resolved_node_directory_in_path(self, monkeypatch):
|
|
monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: "/home/test/.nvm/versions/node/v24.14.0/bin/node" if cmd == "node" else None)
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
assert "/home/test/.nvm/versions/node/v24.14.0/bin" in unit
|
|
|
|
def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self):
|
|
unit = gateway_cli.generate_systemd_unit(system=True)
|
|
|
|
assert "ExecStart=" in unit
|
|
assert "ExecStop=" not in unit
|
|
assert "TimeoutStopSec=60" in unit
|
|
assert "WantedBy=multi-user.target" in unit
|
|
|
|
|
|
class TestGatewayStopCleanup:
|
|
def test_stop_sweeps_manual_gateway_processes_after_service_stop(self, tmp_path, monkeypatch):
|
|
unit_path = tmp_path / "hermes-gateway.service"
|
|
unit_path.write_text("unit\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path)
|
|
|
|
service_calls = []
|
|
kill_calls = []
|
|
|
|
monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop"))
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"kill_gateway_processes",
|
|
lambda force=False: kill_calls.append(force) or 2,
|
|
)
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="stop"))
|
|
|
|
assert service_calls == ["stop"]
|
|
assert kill_calls == [False]
|
|
|
|
|
|
class TestLaunchdServiceRecovery:
|
|
def test_launchd_install_repairs_outdated_plist_without_force(self, tmp_path, monkeypatch):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text("<plist>old content</plist>", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.launchd_install()
|
|
|
|
assert "--replace" in plist_path.read_text(encoding="utf-8")
|
|
assert calls[:2] == [
|
|
["launchctl", "unload", str(plist_path)],
|
|
["launchctl", "load", str(plist_path)],
|
|
]
|
|
|
|
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")
|
|
label = gateway_cli.get_launchd_label()
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, check=False, **kwargs):
|
|
calls.append(cmd)
|
|
if cmd == ["launchctl", "start", label] and calls.count(cmd) == 1:
|
|
raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
gateway_cli.launchd_start()
|
|
|
|
assert calls == [
|
|
["launchctl", "start", label],
|
|
["launchctl", "load", str(plist_path)],
|
|
["launchctl", "start", label],
|
|
]
|
|
|
|
def test_launchd_status_reports_local_stale_plist_when_unloaded(self, tmp_path, monkeypatch, capsys):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text("<plist>old content</plist>", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(
|
|
gateway_cli.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: SimpleNamespace(returncode=113, stdout="", stderr="Could not find service"),
|
|
)
|
|
|
|
gateway_cli.launchd_status()
|
|
|
|
output = capsys.readouterr().out
|
|
assert str(plist_path) in output
|
|
assert "stale" in output.lower()
|
|
assert "not loaded" in output.lower()
|
|
|
|
|
|
class TestGatewayServiceDetection:
|
|
def test_is_service_running_checks_system_scope_when_user_scope_is_inactive(self, monkeypatch):
|
|
user_unit = SimpleNamespace(exists=lambda: True)
|
|
system_unit = SimpleNamespace(exists=lambda: True)
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: system_unit if system else user_unit,
|
|
)
|
|
|
|
def fake_run(cmd, capture_output=True, text=True, **kwargs):
|
|
if cmd == ["systemctl", "--user", "is-active", gateway_cli.get_service_name()]:
|
|
return SimpleNamespace(returncode=0, stdout="inactive\n", stderr="")
|
|
if cmd == ["systemctl", "is-active", gateway_cli.get_service_name()]:
|
|
return SimpleNamespace(returncode=0, stdout="active\n", stderr="")
|
|
raise AssertionError(f"Unexpected command: {cmd}")
|
|
|
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
|
|
|
assert gateway_cli._is_service_running() is True
|
|
|
|
|
|
class TestGatewaySystemServiceRouting:
|
|
def test_gateway_install_passes_system_flags(self, monkeypatch):
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"systemd_install",
|
|
lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)),
|
|
)
|
|
|
|
gateway_cli.gateway_command(
|
|
SimpleNamespace(gateway_command="install", force=True, system=True, run_as_user="alice")
|
|
)
|
|
|
|
assert calls == [(True, True, "alice")]
|
|
|
|
def test_gateway_status_prefers_system_service_when_only_system_unit_exists(self, monkeypatch):
|
|
user_unit = SimpleNamespace(exists=lambda: False)
|
|
system_unit = SimpleNamespace(exists=lambda: True)
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"get_systemd_unit_path",
|
|
lambda system=False: system_unit if system else user_unit,
|
|
)
|
|
|
|
calls = []
|
|
monkeypatch.setattr(gateway_cli, "systemd_status", lambda deep=False, system=False: calls.append((deep, system)))
|
|
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="status", deep=False, system=False))
|
|
|
|
assert calls == [(False, False)]
|
|
|
|
def test_gateway_restart_does_not_fallback_to_foreground_when_launchd_restart_fails(self, tmp_path, monkeypatch):
|
|
plist_path = tmp_path / "ai.hermes.gateway.plist"
|
|
plist_path.write_text("plist\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(gateway_cli, "is_linux", lambda: False)
|
|
monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
|
|
monkeypatch.setattr(gateway_cli, "get_launchd_plist_path", lambda: plist_path)
|
|
monkeypatch.setattr(
|
|
gateway_cli,
|
|
"launchd_restart",
|
|
lambda: (_ for _ in ()).throw(
|
|
gateway_cli.subprocess.CalledProcessError(5, ["launchctl", "start", "ai.hermes.gateway"])
|
|
),
|
|
)
|
|
|
|
run_calls = []
|
|
monkeypatch.setattr(gateway_cli, "run_gateway", lambda verbose=False, replace=False: run_calls.append((verbose, replace)))
|
|
monkeypatch.setattr(gateway_cli, "kill_gateway_processes", lambda force=False: 0)
|
|
|
|
try:
|
|
gateway_cli.gateway_command(SimpleNamespace(gateway_command="restart", system=False))
|
|
except SystemExit as exc:
|
|
assert exc.code == 1
|
|
else:
|
|
raise AssertionError("Expected gateway_command to exit when service restart fails")
|
|
|
|
assert run_calls == []
|
|
|
|
|
|
class TestDetectVenvDir:
|
|
"""Tests for _detect_venv_dir() virtualenv detection."""
|
|
|
|
def test_detects_active_virtualenv_via_sys_prefix(self, tmp_path, monkeypatch):
|
|
venv_path = tmp_path / "my-custom-venv"
|
|
venv_path.mkdir()
|
|
monkeypatch.setattr("sys.prefix", str(venv_path))
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == venv_path
|
|
|
|
def test_falls_back_to_dot_venv_directory(self, tmp_path, monkeypatch):
|
|
# Not inside a virtualenv
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
dot_venv = tmp_path / ".venv"
|
|
dot_venv.mkdir()
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == dot_venv
|
|
|
|
def test_falls_back_to_venv_directory(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
venv = tmp_path / "venv"
|
|
venv.mkdir()
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == venv
|
|
|
|
def test_prefers_dot_venv_over_venv(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
(tmp_path / ".venv").mkdir()
|
|
(tmp_path / "venv").mkdir()
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result == tmp_path / ".venv"
|
|
|
|
def test_returns_none_when_no_virtualenv(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("sys.prefix", "/usr")
|
|
monkeypatch.setattr("sys.base_prefix", "/usr")
|
|
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
|
|
|
|
result = gateway_cli._detect_venv_dir()
|
|
assert result is None
|
|
|
|
|
|
class TestGeneratedUnitUsesDetectedVenv:
|
|
def test_systemd_unit_uses_dot_venv_when_detected(self, tmp_path, monkeypatch):
|
|
dot_venv = tmp_path / ".venv"
|
|
dot_venv.mkdir()
|
|
(dot_venv / "bin").mkdir()
|
|
|
|
monkeypatch.setattr(gateway_cli, "_detect_venv_dir", lambda: dot_venv)
|
|
monkeypatch.setattr(gateway_cli, "get_python_path", lambda: str(dot_venv / "bin" / "python"))
|
|
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
|
|
assert f"VIRTUAL_ENV={dot_venv}" in unit
|
|
assert f"{dot_venv}/bin" in unit
|
|
# Must NOT contain a hardcoded /venv/ path
|
|
assert "/venv/" not in unit or "/.venv/" in unit
|
|
|
|
|
|
class TestGeneratedUnitIncludesLocalBin:
|
|
"""~/.local/bin must be in PATH so uvx/pipx tools are discoverable."""
|
|
|
|
def test_user_unit_includes_local_bin_in_path(self):
|
|
unit = gateway_cli.generate_systemd_unit(system=False)
|
|
home = str(Path.home())
|
|
assert f"{home}/.local/bin" in unit
|
|
|
|
def test_system_unit_includes_local_bin_in_path(self):
|
|
unit = gateway_cli.generate_systemd_unit(system=True)
|
|
# System unit uses the resolved home dir from _system_service_identity
|
|
assert "/.local/bin" in unit
|
|
|
|
|
|
class TestEnsureUserSystemdEnv:
|
|
"""Tests for _ensure_user_systemd_env() D-Bus session bus auto-detection."""
|
|
|
|
def test_sets_xdg_runtime_dir_when_missing(self, tmp_path, monkeypatch):
|
|
monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False)
|
|
monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
|
|
monkeypatch.setattr(os, "getuid", lambda: 42)
|
|
|
|
# Patch Path.exists so /run/user/42 appears to exist.
|
|
# Using a FakePath subclass breaks on Python 3.12+ where
|
|
# PosixPath.__new__ ignores the redirected path argument.
|
|
_orig_exists = gateway_cli.Path.exists
|
|
monkeypatch.setattr(
|
|
gateway_cli.Path, "exists",
|
|
lambda self: True if str(self) == "/run/user/42" else _orig_exists(self),
|
|
)
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert os.environ.get("XDG_RUNTIME_DIR") == "/run/user/42"
|
|
|
|
def test_sets_dbus_address_when_bus_socket_exists(self, tmp_path, monkeypatch):
|
|
runtime = tmp_path / "runtime"
|
|
runtime.mkdir()
|
|
bus_socket = runtime / "bus"
|
|
bus_socket.touch() # simulate the socket file
|
|
|
|
monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime))
|
|
monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
|
|
monkeypatch.setattr(os, "getuid", lambda: 99)
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert os.environ["DBUS_SESSION_BUS_ADDRESS"] == f"unix:path={bus_socket}"
|
|
|
|
def test_preserves_existing_env_vars(self, monkeypatch):
|
|
monkeypatch.setenv("XDG_RUNTIME_DIR", "/custom/runtime")
|
|
monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/custom/bus")
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert os.environ["XDG_RUNTIME_DIR"] == "/custom/runtime"
|
|
assert os.environ["DBUS_SESSION_BUS_ADDRESS"] == "unix:path=/custom/bus"
|
|
|
|
def test_no_dbus_when_bus_socket_missing(self, tmp_path, monkeypatch):
|
|
runtime = tmp_path / "runtime"
|
|
runtime.mkdir()
|
|
# no bus socket created
|
|
|
|
monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime))
|
|
monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False)
|
|
monkeypatch.setattr(os, "getuid", lambda: 99)
|
|
|
|
gateway_cli._ensure_user_systemd_env()
|
|
|
|
assert "DBUS_SESSION_BUS_ADDRESS" not in os.environ
|
|
|
|
def test_systemctl_cmd_calls_ensure_for_user_mode(self, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("called"))
|
|
|
|
result = gateway_cli._systemctl_cmd(system=False)
|
|
assert result == ["systemctl", "--user"]
|
|
assert calls == ["called"]
|
|
|
|
def test_systemctl_cmd_skips_ensure_for_system_mode(self, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("called"))
|
|
|
|
result = gateway_cli._systemctl_cmd(system=True)
|
|
assert result == ["systemctl"]
|
|
assert calls == []
|