feat: add system gateway service mode

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")

hermes_cli/main.py

@@ -2433,23 +2433,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")

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

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

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)]