diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 3ecc77e07..6e75c9b51 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -123,10 +123,61 @@ SERVICE_NAME = "hermes-gateway" SERVICE_DESCRIPTION = "Hermes Agent Gateway - Messaging Platform Integration" -def get_systemd_unit_path() -> Path: +def get_systemd_unit_path(system: bool = False) -> Path: + if system: + return Path("/etc/systemd/system") / f"{SERVICE_NAME}.service" return Path.home() / ".config" / "systemd" / "user" / f"{SERVICE_NAME}.service" +def _systemctl_cmd(system: bool = False) -> list[str]: + return ["systemctl"] if system else ["systemctl", "--user"] + + +def _journalctl_cmd(system: bool = False) -> list[str]: + return ["journalctl"] if system else ["journalctl", "--user"] + + +def _service_scope_label(system: bool = False) -> str: + return "system" if system else "user" + + +def _require_root_for_system_service(action: str) -> None: + if os.geteuid() != 0: + print(f"System gateway {action} requires root. Re-run with sudo.") + sys.exit(1) + + +def _system_service_identity(run_as_user: str | None = None) -> tuple[str, str, str]: + import getpass + import grp + import pwd + + username = (run_as_user or os.getenv("SUDO_USER") or os.getenv("USER") or os.getenv("LOGNAME") or getpass.getuser()).strip() + if not username: + raise ValueError("Could not determine which user the gateway service should run as") + if username == "root": + raise ValueError("Refusing to install the gateway system service as root; pass --run-as USER") + + try: + user_info = pwd.getpwnam(username) + except KeyError as e: + raise ValueError(f"Unknown user: {username}") from e + + group_name = grp.getgrgid(user_info.pw_gid).gr_name + return username, group_name, user_info.pw_dir + + +def _read_systemd_user_from_unit(unit_path: Path) -> str | None: + if not unit_path.exists(): + return None + + for line in unit_path.read_text(encoding="utf-8").splitlines(): + if line.startswith("User="): + value = line.split("=", 1)[1].strip() + return value or None + return None + + def get_systemd_linger_status() -> tuple[bool | None, str]: """Return whether systemd user lingering is enabled for the current user. @@ -216,8 +267,9 @@ def get_hermes_cli_path() -> str: # Systemd (Linux) # ============================================================================= -def generate_systemd_unit() -> str: +def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) -> str: import shutil + python_path = get_python_path() working_dir = str(PROJECT_ROOT) venv_dir = str(PROJECT_ROOT / "venv") @@ -226,8 +278,38 @@ def generate_systemd_unit() -> str: # Build a PATH that includes the venv, node_modules, and standard system dirs sane_path = f"{venv_bin}:{node_bin}:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" - hermes_cli = shutil.which("hermes") or f"{python_path} -m hermes_cli.main" + + if system: + username, group_name, home_dir = _system_service_identity(run_as_user) + return f"""[Unit] +Description={SERVICE_DESCRIPTION} +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User={username} +Group={group_name} +ExecStart={python_path} -m hermes_cli.main gateway run --replace +WorkingDirectory={working_dir} +Environment="HOME={home_dir}" +Environment="USER={username}" +Environment="LOGNAME={username}" +Environment="PATH={sane_path}" +Environment="VIRTUAL_ENV={venv_dir}" +Restart=on-failure +RestartSec=10 +KillMode=mixed +KillSignal=SIGTERM +TimeoutStopSec=15 +StandardOutput=journal +StandardError=journal + +[Install] +WantedBy=multi-user.target +""" + return f"""[Unit] Description={SERVICE_DESCRIPTION} After=network.target @@ -255,26 +337,28 @@ def _normalize_service_definition(text: str) -> str: return "\n".join(line.rstrip() for line in text.strip().splitlines()) -def systemd_unit_is_current() -> bool: - unit_path = get_systemd_unit_path() +def systemd_unit_is_current(system: bool = False) -> bool: + unit_path = get_systemd_unit_path(system=system) if not unit_path.exists(): return False installed = unit_path.read_text(encoding="utf-8") - expected = generate_systemd_unit() + expected_user = _read_systemd_user_from_unit(unit_path) if system else None + expected = generate_systemd_unit(system=system, run_as_user=expected_user) return _normalize_service_definition(installed) == _normalize_service_definition(expected) -def refresh_systemd_unit_if_needed() -> bool: - """Rewrite the installed user unit when the generated definition has changed.""" - unit_path = get_systemd_unit_path() - if not unit_path.exists() or systemd_unit_is_current(): +def refresh_systemd_unit_if_needed(system: bool = False) -> bool: + """Rewrite the installed systemd unit when the generated definition has changed.""" + unit_path = get_systemd_unit_path(system=system) + if not unit_path.exists() or systemd_unit_is_current(system=system): return False - unit_path.write_text(generate_systemd_unit(), encoding="utf-8") - subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) - print("↻ Updated gateway service definition to match the current Hermes install") + expected_user = _read_systemd_user_from_unit(unit_path) if system else None + unit_path.write_text(generate_systemd_unit(system=system, run_as_user=expected_user), encoding="utf-8") + subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True) + print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install") return True @@ -337,93 +421,131 @@ def _ensure_linger_enabled() -> None: _print_linger_enable_warning(username, detail or linger_detail) -def systemd_install(force: bool = False): - unit_path = get_systemd_unit_path() - +def _select_systemd_scope(system: bool = False) -> bool: + if system: + return True + return get_systemd_unit_path(system=True).exists() and not get_systemd_unit_path(system=False).exists() + + +def systemd_install(force: bool = False, system: bool = False, run_as_user: str | None = None): + if system: + _require_root_for_system_service("install") + + unit_path = get_systemd_unit_path(system=system) + scope_flag = " --system" if system else "" + if unit_path.exists() and not force: print(f"Service already installed at: {unit_path}") print("Use --force to reinstall") return - + unit_path.parent.mkdir(parents=True, exist_ok=True) - print(f"Installing systemd service to: {unit_path}") - unit_path.write_text(generate_systemd_unit()) - - subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) - subprocess.run(["systemctl", "--user", "enable", SERVICE_NAME], check=True) - + print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}") + unit_path.write_text(generate_systemd_unit(system=system, run_as_user=run_as_user), encoding="utf-8") + + subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True) + subprocess.run(_systemctl_cmd(system) + ["enable", SERVICE_NAME], check=True) + print() - print("✓ Service installed and enabled!") + print(f"✓ {_service_scope_label(system).capitalize()} service installed and enabled!") print() print("Next steps:") - print(f" hermes gateway start # Start the service") - print(f" hermes gateway status # Check status") - print(f" journalctl --user -u {SERVICE_NAME} -f # View logs") + print(f" {'sudo ' if system else ''}hermes gateway start{scope_flag} # Start the service") + print(f" {'sudo ' if system else ''}hermes gateway status{scope_flag} # Check status") + print(f" {'journalctl' if system else 'journalctl --user'} -u {SERVICE_NAME} -f # View logs") print() - _ensure_linger_enabled() -def systemd_uninstall(): - subprocess.run(["systemctl", "--user", "stop", SERVICE_NAME], check=False) - subprocess.run(["systemctl", "--user", "disable", SERVICE_NAME], check=False) - - unit_path = get_systemd_unit_path() + if system: + configured_user = _read_systemd_user_from_unit(unit_path) + if configured_user: + print(f"Configured to run as: {configured_user}") + else: + _ensure_linger_enabled() + + +def systemd_uninstall(system: bool = False): + system = _select_systemd_scope(system) + if system: + _require_root_for_system_service("uninstall") + + subprocess.run(_systemctl_cmd(system) + ["stop", SERVICE_NAME], check=False) + subprocess.run(_systemctl_cmd(system) + ["disable", SERVICE_NAME], check=False) + + unit_path = get_systemd_unit_path(system=system) if unit_path.exists(): unit_path.unlink() print(f"✓ Removed {unit_path}") - - subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) - print("✓ Service uninstalled") -def systemd_start(): - refresh_systemd_unit_if_needed() - subprocess.run(["systemctl", "--user", "start", SERVICE_NAME], check=True) - print("✓ Service started") + subprocess.run(_systemctl_cmd(system) + ["daemon-reload"], check=True) + print(f"✓ {_service_scope_label(system).capitalize()} service uninstalled") -def systemd_stop(): - subprocess.run(["systemctl", "--user", "stop", SERVICE_NAME], check=True) - print("✓ Service stopped") +def systemd_start(system: bool = False): + system = _select_systemd_scope(system) + if system: + _require_root_for_system_service("start") + refresh_systemd_unit_if_needed(system=system) + subprocess.run(_systemctl_cmd(system) + ["start", SERVICE_NAME], check=True) + print(f"✓ {_service_scope_label(system).capitalize()} service started") -def systemd_restart(): - refresh_systemd_unit_if_needed() - subprocess.run(["systemctl", "--user", "restart", SERVICE_NAME], check=True) - print("✓ Service restarted") + +def systemd_stop(system: bool = False): + system = _select_systemd_scope(system) + if system: + _require_root_for_system_service("stop") + subprocess.run(_systemctl_cmd(system) + ["stop", SERVICE_NAME], check=True) + print(f"✓ {_service_scope_label(system).capitalize()} service stopped") -def systemd_status(deep: bool = False): - # Check if service unit file exists - unit_path = get_systemd_unit_path() + +def systemd_restart(system: bool = False): + system = _select_systemd_scope(system) + if system: + _require_root_for_system_service("restart") + refresh_systemd_unit_if_needed(system=system) + subprocess.run(_systemctl_cmd(system) + ["restart", SERVICE_NAME], check=True) + print(f"✓ {_service_scope_label(system).capitalize()} service restarted") + + + +def systemd_status(deep: bool = False, system: bool = False): + system = _select_systemd_scope(system) + unit_path = get_systemd_unit_path(system=system) + scope_flag = " --system" if system else "" + if not unit_path.exists(): print("✗ Gateway service is not installed") - print(" Run: hermes gateway install") + print(f" Run: {'sudo ' if system else ''}hermes gateway install{scope_flag}") return - if not systemd_unit_is_current(): + if not systemd_unit_is_current(system=system): print("⚠ Installed gateway service definition is outdated") - print(" Run: hermes gateway restart # auto-refreshes the unit") + print(f" Run: {'sudo ' if system else ''}hermes gateway restart{scope_flag} # auto-refreshes the unit") print() - - # Show detailed status first + subprocess.run( - ["systemctl", "--user", "status", SERVICE_NAME, "--no-pager"], - capture_output=False + _systemctl_cmd(system) + ["status", SERVICE_NAME, "--no-pager"], + capture_output=False, ) - # Check if service is active result = subprocess.run( - ["systemctl", "--user", "is-active", SERVICE_NAME], + _systemctl_cmd(system) + ["is-active", SERVICE_NAME], capture_output=True, - text=True + text=True, ) status = result.stdout.strip() if status == "active": - print("✓ Gateway service is running") + print(f"✓ {_service_scope_label(system).capitalize()} gateway service is running") else: - print("✗ Gateway service is stopped") - print(" Run: hermes gateway start") + print(f"✗ {_service_scope_label(system).capitalize()} gateway service is stopped") + print(f" Run: {'sudo ' if system else ''}hermes gateway start{scope_flag}") + + configured_user = _read_systemd_user_from_unit(unit_path) if system else None + if configured_user: + print(f"Configured to run as: {configured_user}") runtime_lines = _runtime_health_lines() if runtime_lines: @@ -432,7 +554,9 @@ def systemd_status(deep: bool = False): for line in runtime_lines: print(f" {line}") - if deep: + if system: + print("✓ System service starts at boot without requiring systemd linger") + elif deep: print_systemd_linger_guidance() else: linger_enabled, _ = get_systemd_linger_status() @@ -445,10 +569,7 @@ def systemd_status(deep: bool = False): if deep: print() print("Recent logs:") - subprocess.run([ - "journalctl", "--user", "-u", SERVICE_NAME, - "-n", "20", "--no-pager" - ]) + subprocess.run(_journalctl_cmd(system) + ["-u", SERVICE_NAME, "-n", "20", "--no-pager"]) # ============================================================================= @@ -895,7 +1016,7 @@ def _setup_whatsapp(): def _is_service_installed() -> bool: """Check if the gateway is installed as a system service.""" if is_linux(): - return get_systemd_unit_path().exists() + return get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists() elif is_macos(): return get_launchd_plist_path().exists() return False @@ -903,12 +1024,19 @@ def _is_service_installed() -> bool: def _is_service_running() -> bool: """Check if the gateway service is currently running.""" - if is_linux() and get_systemd_unit_path().exists(): - result = subprocess.run( - ["systemctl", "--user", "is-active", SERVICE_NAME], - capture_output=True, text=True - ) - return result.stdout.strip() == "active" + if is_linux(): + if get_systemd_unit_path(system=False).exists(): + result = subprocess.run( + _systemctl_cmd(False) + ["is-active", SERVICE_NAME], + capture_output=True, text=True + ) + return result.stdout.strip() == "active" + if get_systemd_unit_path(system=True).exists(): + result = subprocess.run( + _systemctl_cmd(True) + ["is-active", SERVICE_NAME], + capture_output=True, text=True + ) + return result.stdout.strip() == "active" elif is_macos() and get_launchd_plist_path().exists(): result = subprocess.run( ["launchctl", "list", "ai.hermes.gateway"], @@ -1183,8 +1311,10 @@ def gateway_command(args): # Service management commands if subcmd == "install": force = getattr(args, 'force', False) + system = getattr(args, 'system', False) + run_as_user = getattr(args, 'run_as_user', None) if is_linux(): - systemd_install(force) + systemd_install(force=force, system=system, run_as_user=run_as_user) elif is_macos(): launchd_install(force) else: @@ -1193,8 +1323,9 @@ def gateway_command(args): sys.exit(1) elif subcmd == "uninstall": + system = getattr(args, 'system', False) if is_linux(): - systemd_uninstall() + systemd_uninstall(system=system) elif is_macos(): launchd_uninstall() else: @@ -1202,8 +1333,9 @@ def gateway_command(args): sys.exit(1) elif subcmd == "start": + system = getattr(args, 'system', False) if is_linux(): - systemd_start() + systemd_start(system=system) elif is_macos(): launchd_start() else: @@ -1213,10 +1345,11 @@ def gateway_command(args): elif subcmd == "stop": # Try service first, then sweep any stray/manual gateway processes. service_available = False + system = getattr(args, 'system', False) - if is_linux() and get_systemd_unit_path().exists(): + if is_linux() and (get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()): try: - systemd_stop() + systemd_stop(system=system) service_available = True except subprocess.CalledProcessError: pass # Fall through to process kill @@ -1239,10 +1372,11 @@ def gateway_command(args): elif subcmd == "restart": # Try service first, fall back to killing and restarting service_available = False + system = getattr(args, 'system', False) - if is_linux() and get_systemd_unit_path().exists(): + if is_linux() and (get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()): try: - systemd_restart() + systemd_restart(system=system) service_available = True except subprocess.CalledProcessError: pass @@ -1268,10 +1402,11 @@ def gateway_command(args): elif subcmd == "status": deep = getattr(args, 'deep', False) + system = getattr(args, 'system', False) # Check for service first - if is_linux() and get_systemd_unit_path().exists(): - systemd_status(deep) + if is_linux() and (get_systemd_unit_path(system=False).exists() or get_systemd_unit_path(system=True).exists()): + systemd_status(deep, system=system) elif is_macos() and get_launchd_plist_path().exists(): launchd_status(deep) else: @@ -1289,6 +1424,7 @@ def gateway_command(args): print() print("To install as a service:") print(" hermes gateway install") + print(" sudo hermes gateway install --system") else: print("✗ Gateway is not running") runtime_lines = _runtime_health_lines() @@ -1300,4 +1436,5 @@ def gateway_command(args): print() print("To start:") print(" hermes gateway # Run in foreground") - print(" hermes gateway install # Install as service") + print(" hermes gateway install # Install as user service") + print(" sudo hermes gateway install --system # Install as boot-time system service") diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 429c8b593..1238d9b6a 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -2481,23 +2481,30 @@ For more help on a command: # gateway start gateway_start = gateway_subparsers.add_parser("start", help="Start gateway service") + gateway_start.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service") # gateway stop gateway_stop = gateway_subparsers.add_parser("stop", help="Stop gateway service") + gateway_stop.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service") # gateway restart gateway_restart = gateway_subparsers.add_parser("restart", help="Restart gateway service") + gateway_restart.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service") # gateway status gateway_status = gateway_subparsers.add_parser("status", help="Show gateway status") gateway_status.add_argument("--deep", action="store_true", help="Deep status check") + gateway_status.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service") # gateway install gateway_install = gateway_subparsers.add_parser("install", help="Install gateway as service") gateway_install.add_argument("--force", action="store_true", help="Force reinstall") + gateway_install.add_argument("--system", action="store_true", help="Install as a Linux system-level service (starts at boot)") + gateway_install.add_argument("--run-as-user", dest="run_as_user", help="User account the Linux system service should run as") # gateway uninstall gateway_uninstall = gateway_subparsers.add_parser("uninstall", help="Uninstall gateway service") + gateway_uninstall.add_argument("--system", action="store_true", help="Target the Linux system-level gateway service") # gateway setup gateway_setup = gateway_subparsers.add_parser("setup", help="Configure messaging platforms") diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py index ad987d575..d3f4bb9e8 100644 --- a/tests/hermes_cli/test_gateway.py +++ b/tests/hermes_cli/test_gateway.py @@ -35,7 +35,7 @@ def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys unit_path = tmp_path / "hermes-gateway.service" unit_path.write_text("[Unit]\n") - monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path) + monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) monkeypatch.setattr(gateway, "get_systemd_linger_status", lambda: (False, "")) def fake_run(cmd, capture_output=False, text=False, check=False): @@ -50,7 +50,7 @@ def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys gateway.systemd_status(deep=False) out = capsys.readouterr().out - assert "Gateway service is running" in out + assert "gateway service is running" in out assert "Systemd linger is disabled" in out assert "loginctl enable-linger" in out @@ -58,7 +58,7 @@ def test_systemd_status_warns_when_linger_disabled(monkeypatch, tmp_path, capsys def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys): unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service" - monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path) + monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) calls = [] helper_calls = [] @@ -79,4 +79,39 @@ def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys): ["systemctl", "--user", "enable", gateway.SERVICE_NAME], ] assert helper_calls == [True] - assert "Service installed and enabled" in out + assert "User service installed and enabled" in out + + +def test_systemd_install_system_scope_skips_linger_and_uses_systemctl(monkeypatch, tmp_path, capsys): + unit_path = tmp_path / "etc" / "systemd" / "system" / "hermes-gateway.service" + + monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) + monkeypatch.setattr( + gateway, + "generate_systemd_unit", + lambda system=False, run_as_user=None: f"scope={system} user={run_as_user}\n", + ) + monkeypatch.setattr(gateway, "_require_root_for_system_service", lambda action: None) + + calls = [] + helper_calls = [] + + def fake_run(cmd, check=False, **kwargs): + calls.append((cmd, check)) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(gateway.subprocess, "run", fake_run) + monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True)) + + gateway.systemd_install(force=False, system=True, run_as_user="alice") + + out = capsys.readouterr().out + assert unit_path.exists() + assert unit_path.read_text(encoding="utf-8") == "scope=True user=alice\n" + assert [cmd for cmd, _ in calls] == [ + ["systemctl", "daemon-reload"], + ["systemctl", "enable", gateway.SERVICE_NAME], + ] + assert helper_calls == [] + assert "Configured to run as: alice" not in out # generated test unit has no User= line + assert "System service installed and enabled" in out diff --git a/tests/hermes_cli/test_gateway_linger.py b/tests/hermes_cli/test_gateway_linger.py index f1341d068..cdc07f95f 100644 --- a/tests/hermes_cli/test_gateway_linger.py +++ b/tests/hermes_cli/test_gateway_linger.py @@ -96,7 +96,7 @@ class TestEnsureLingerEnabled: def test_systemd_install_calls_linger_helper(monkeypatch, tmp_path, capsys): unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service" - monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda: unit_path) + monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path) calls = [] @@ -117,4 +117,4 @@ def test_systemd_install_calls_linger_helper(monkeypatch, tmp_path, capsys): ["systemctl", "--user", "enable", gateway.SERVICE_NAME], ] assert helper_calls == [True] - assert "Service installed and enabled" in out + assert "User service installed and enabled" in out diff --git a/tests/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py index 4f8eb39a2..1cc0968da 100644 --- a/tests/hermes_cli/test_gateway_service.py +++ b/tests/hermes_cli/test_gateway_service.py @@ -10,8 +10,8 @@ class TestSystemdServiceRefresh: 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: unit_path) - monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda: "new unit\n") + 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 = [] @@ -33,8 +33,8 @@ class TestSystemdServiceRefresh: 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: unit_path) - monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda: "new unit\n") + 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 = [] @@ -60,12 +60,12 @@ class TestGatewayStopCleanup: 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: unit_path) + monkeypatch.setattr(gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path) service_calls = [] kill_calls = [] - monkeypatch.setattr(gateway_cli, "systemd_stop", lambda: service_calls.append("stop")) + monkeypatch.setattr(gateway_cli, "systemd_stop", lambda system=False: service_calls.append("stop")) monkeypatch.setattr( gateway_cli, "kill_gateway_processes", @@ -76,3 +76,41 @@ class TestGatewayStopCleanup: assert service_calls == ["stop"] assert kill_calls == [False] + + +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)]