mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-06-02 07:11:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

fix(gateway): harden Windows gateway install lifecycle

Browse source 
Preserve Windows profile install decisions across UAC handoff, avoid visible console windows by launching via pythonw, make repeated install/start idempotent, recreate stale Scheduled Tasks, and separate start-now from login auto-start behavior. Add Windows gateway regression coverage and systemd setup tests for the shared install flow.
This commit is contained in:
nekwo 2026-05-16 13:22:52 -04:00 • committed by Teknium
parent 95683c0283
commit d948de39e9
5 changed files with 914 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

65
hermes_cli/gateway.py
View file

@ -1839,7 +1839,7 @@ def prompt_linux_gateway_install_scope() -> str | None:
return {0: "user", 1: "system", 2: None}[choice]
def install_linux_gateway_from_setup(force: bool = False) -> tuple[str | None, bool]:
def install_linux_gateway_from_setup(force: bool = False, enable_on_startup: bool = True) -> tuple[str | None, bool]:
scope = prompt_linux_gateway_install_scope()
if scope is None:
return None, False
@ -1863,10 +1863,10 @@ def install_linux_gateway_from_setup(force: bool = False) -> tuple[str | None, b
break
print_error(" Enter a username.")
systemd_install(force=force, system=True, run_as_user=run_as_user)
systemd_install(force=force, system=True, run_as_user=run_as_user, enable_on_startup=enable_on_startup)
return scope, True
systemd_install(force=force, system=False)
systemd_install(force=force, system=False, enable_on_startup=enable_on_startup)
return scope, True
@ -2437,7 +2437,12 @@ def _get_restart_drain_timeout() -> float:
return parse_restart_drain_timeout(raw)
def systemd_install(force: bool = False, system: bool = False, run_as_user: str | None = None):
def systemd_install(
force: bool = False,
system: bool = False,
run_as_user: str | None = None,
enable_on_startup: bool = True,
):
if system:
_require_root_for_system_service("install")
@ -2461,7 +2466,8 @@ def systemd_install(force: bool = False, system: bool = False, run_as_user: str
if not systemd_unit_is_current(system=system):
print(f"↻ Repairing outdated {_service_scope_label(system)} systemd service at: {unit_path}")
refresh_systemd_unit_if_needed(system=system)
_run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
if enable_on_startup:
_run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
print(f"✓ {_service_scope_label(system).capitalize()} service definition updated")
return
print(f"Service already installed at: {unit_path}")
@ -2473,10 +2479,12 @@ def systemd_install(force: bool = False, system: bool = False, run_as_user: str
unit_path.write_text(generate_systemd_unit(system=system, run_as_user=run_as_user), encoding="utf-8")
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
_run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
if enable_on_startup:
_run_systemctl(["enable", get_service_name()], system=system, check=True, timeout=30)
print()
print(f"✓ {_service_scope_label(system).capitalize()} service installed and enabled!")
enable_label = "installed and enabled" if enable_on_startup else "installed"
print(f"✓ {_service_scope_label(system).capitalize()} service {enable_label}!")
print()
print("Next steps:")
print(f" {'sudo ' if system else ''}hermes gateway start{scope_flag} # Start the service")
@ -4947,31 +4955,37 @@ def gateway_setup():
else:
platform_name = "Scheduled Task"
wsl_note = " (note: services may not survive WSL restarts)" if is_wsl() else ""
if prompt_yes_no(f" Install the gateway as a {platform_name} service?{wsl_note} (runs in background, starts on boot)", True):
start_now = prompt_yes_no(" Start the gateway now?", True)
start_on_login = prompt_yes_no(
f" Start the gateway automatically on login/boot as a {platform_name} service?{wsl_note}",
True,
)
if start_now or start_on_login:
try:
installed_scope = None
did_install = False
started_inline = False
if supports_systemd_services():
installed_scope, did_install = install_linux_gateway_from_setup(force=False)
installed_scope, did_install = install_linux_gateway_from_setup(
force=False,
enable_on_startup=start_on_login,
)
elif is_macos():
launchd_install(force=False)
did_install = True
else:
# gateway_windows.install() registers the Scheduled
# Task AND starts it (schtasks /Run or direct-spawn
# fallback), so no separate start prompt is needed.
from hermes_cli import gateway_windows
gateway_windows.install(force=False)
did_install = True
started_inline = True
print()
if did_install and not started_inline and prompt_yes_no(" Start the service now?", True):
if did_install and start_now:
try:
if supports_systemd_services():
systemd_start(system=installed_scope == "system")
else:
elif is_macos():
launchd_start()
elif is_windows():
from hermes_cli import gateway_windows
gateway_windows.start()
except UserSystemdUnavailableError as e:
print_error(" Start failed — user systemd not reachable:")
for line in str(e).splitlines():
@ -4982,6 +4996,7 @@ def gateway_setup():
print_error(f" Install failed: {e}")
print_info(" You can try manually: hermes gateway install")
else:
print_info(" Skipped start and auto-start setup.")
print_info(" You can install later: hermes gateway install")
if supports_systemd_services():
print_info(" Or as a boot-time service: sudo hermes gateway install --system")
@ -5064,12 +5079,26 @@ def _gateway_command_inner(args):
print_info(" Consider running in foreground instead: hermes gateway run")
print_info(" Or use tmux/screen for persistence: tmux new -s hermes 'hermes gateway run'")
print()
systemd_install(force=force, system=system, run_as_user=run_as_user)
start_now = prompt_yes_no("Start the gateway now after installing the service?", True)
start_on_login = prompt_yes_no("Start the gateway automatically on login/boot with systemd?", True)
systemd_install(
force=force,
system=system,
run_as_user=run_as_user,
enable_on_startup=start_on_login,
)
if start_now:
systemd_start(system=system)
elif is_macos():
launchd_install(force)
elif is_windows():
from hermes_cli import gateway_windows
gateway_windows.install(force=force)
gateway_windows.install(
force=force,
start_now=getattr(args, 'start_now', None),
start_on_login=getattr(args, 'start_on_login', None),
elevated_handoff=getattr(args, 'elevated_handoff', False),
)
elif is_wsl():
print("WSL detected but systemd is not running.")
print("Either enable systemd (add systemd=true to /etc/wsl.conf and restart WSL)")

363
hermes_cli/gateway_windows.py
View file

@ -28,6 +28,7 @@ Design notes
from __future__ import annotations
import ctypes
import os
import re
import shlex
@ -45,6 +46,7 @@ _FALLBACK_PATTERNS = re.compile(
r"(access is denied|acceso denegado|přístup byl odepřen|schtasks timed out|schtasks produced no output)",
re.IGNORECASE,
)
_ACCESS_DENIED_PATTERN = re.compile(r"(access is denied|acceso denegado)", re.IGNORECASE)
_TASK_NAME_DEFAULT = "Hermes_Gateway"
_TASK_DESCRIPTION = "Hermes Agent Gateway - Messaging Platform Integration"
@ -127,6 +129,100 @@ def _should_fall_back(code: int, detail: str) -> bool:
return code == 124 or bool(_FALLBACK_PATTERNS.search(detail or ""))
def _is_access_denied(detail: str) -> bool:
return bool(_ACCESS_DENIED_PATTERN.search(detail or ""))
def _is_running_as_admin() -> bool:
"""Return True when the current Windows process is elevated."""
_assert_windows()
try:
return bool(ctypes.windll.shell32.IsUserAnAdmin())
except Exception:
return False
def _current_profile_cli_args() -> list[str]:
"""Return CLI args that preserve the current Hermes profile."""
from hermes_cli.gateway import _profile_arg
profile_arg = _profile_arg()
return shlex.split(profile_arg) if profile_arg else []
def _launch_elevated_gateway_command(command: str, extra_args: list[str] | None = None) -> bool:
"""Launch an elevated gateway subcommand via UAC and return True on handoff.
Use pythonw.exe for the elevated child so approving UAC does not leave a
second elevated console window sitting open after the handoff. All operator
decisions are already collected in the parent shell before this point.
"""
_assert_windows()
args = ["-m", "hermes_cli.main", *_current_profile_cli_args(), "gateway", command]
if extra_args:
args.extend(extra_args)
params = subprocess.list2cmdline(args)
cwd = str(Path(__file__).resolve().parent.parent)
elevated_python = _derive_venv_pythonw(sys.executable)
try:
result = ctypes.windll.shell32.ShellExecuteW(
None,
"runas",
elevated_python,
params,
cwd,
0, # SW_HIDE: pythonw child should not create a visible console.
)
except Exception as exc:
print(f"⚠ Could not launch elevated gateway {command} prompt: {exc}")
return False
if result <= 32:
print(f"⚠ Elevated gateway {command} prompt was not started (ShellExecuteW={result})")
return False
return True
def _launch_elevated_install(
force: bool = False,
*,
start_now: bool | None = None,
start_on_login: bool | None = None,
) -> bool:
"""Launch an elevated gateway install via UAC and return True on handoff."""
old_start_now = os.environ.get("HERMES_GATEWAY_INSTALL_START_NOW")
old_start_on_login = os.environ.get("HERMES_GATEWAY_INSTALL_START_ON_LOGIN")
old_handoff = os.environ.get("HERMES_GATEWAY_ELEVATED_HANDOFF")
try:
if start_now is not None:
os.environ["HERMES_GATEWAY_INSTALL_START_NOW"] = "1" if start_now else "0"
if start_on_login is not None:
os.environ["HERMES_GATEWAY_INSTALL_START_ON_LOGIN"] = "1" if start_on_login else "0"
os.environ["HERMES_GATEWAY_ELEVATED_HANDOFF"] = "1"
extra_args = ["--elevated-handoff"]
if force:
extra_args.append("--force")
if start_now is not None:
extra_args.append("--start-now" if start_now else "--no-start-now")
if start_on_login is not None:
extra_args.append("--start-on-login" if start_on_login else "--no-start-on-login")
return _launch_elevated_gateway_command("install", extra_args)
finally:
for key, old in (
("HERMES_GATEWAY_INSTALL_START_NOW", old_start_now),
("HERMES_GATEWAY_INSTALL_START_ON_LOGIN", old_start_on_login),
("HERMES_GATEWAY_ELEVATED_HANDOFF", old_handoff),
):
if old is None:
os.environ.pop(key, None)
else:
os.environ[key] = old
def _launch_elevated_uninstall() -> bool:
"""Launch an elevated gateway uninstall via UAC and return True on handoff."""
return _launch_elevated_gateway_command("uninstall")
# ---------------------------------------------------------------------------
# Paths: where we stash our task script and where Startup lives
# ---------------------------------------------------------------------------
@ -206,7 +302,8 @@ def _build_gateway_cmd_script(
The script:
- cd's into the project directory
- exports HERMES_HOME, PYTHONIOENCODING, VIRTUAL_ENV
- invokes ``python -m hermes_cli.main [--profile X] gateway run --replace``
- invokes ``pythonw -m hermes_cli.main [--profile X] gateway run``
directly so the wrapper cmd.exe exits without a visible gateway console
We intentionally do NOT inline PATH overrides here — cmd.exe inherits
the per-user PATH the Scheduled Task was created with, and forcibly
@ -222,11 +319,19 @@ def _build_gateway_cmd_script(
venv_dir = str(Path(python_path).resolve().parent.parent)
lines.append(f'set "VIRTUAL_ENV={venv_dir}"')
prog_args = [python_path, "-m", "hermes_cli.main"]
pythonw_path = _derive_venv_pythonw(python_path)
prog_args = [pythonw_path, "-m", "hermes_cli.main"]
if profile_arg:
prog_args.extend(profile_arg.split())
prog_args.extend(["gateway", "run", "--replace"])
prog_args.extend(["gateway", "run"])
# `pythonw.exe` is a GUI-subsystem executable: cmd.exe launches it and
# returns immediately, so the Scheduled Task action finishes without a
# visible console window. Do NOT use `start` here; that creates an extra
# wrapper process and made gateway lifecycle/status harder to reason about.
# Do NOT use `--replace` for service-managed starts; repeated /Run calls
# should be idempotent, not churn parent/child takeover loops.
lines.append(" ".join(_quote_cmd_script_arg(a) for a in prog_args))
lines.append("exit /b 0")
return "\r\n".join(lines) + "\r\n"
@ -280,17 +385,22 @@ def _resolve_task_user() -> str | None:
def _install_scheduled_task(task_name: str, script_path: Path) -> tuple[bool, str]:
"""Create or update the Scheduled Task. Returns (success, detail)."""
quoted_script = _quote_schtasks_arg(str(script_path))
# First try /Change in case the task already exists — keeps the existing
# trigger + settings intact and just repoints /TR.
change_code, _out, change_err = _exec_schtasks(
["/Change", "/TN", task_name, "/TR", quoted_script]
)
if change_code == 0:
return (True, f"Updated existing Scheduled Task {task_name!r}")
"""Create or replace the Scheduled Task. Returns (success, detail).
# Create fresh. Start with the "current user, interactive, no stored
Always recreate instead of ``/Change``. Older Hermes builds and failed
experiments may have left repeat/restart settings on the task; ``/Change``
preserves those stale triggers and can make the gateway relaunch every
minute. Delete+create gives us a clean ONLOGON task every install.
"""
quoted_script = _quote_schtasks_arg(str(script_path))
delete_code, delete_out, delete_err = _exec_schtasks(["/Delete", "/F", "/TN", task_name])
delete_detail = (delete_err or delete_out or "").strip()
if delete_code != 0 and delete_detail and "cannot find" not in delete_detail.lower():
if _is_access_denied(delete_detail):
return (False, f"schtasks /Delete failed (code {delete_code}): {delete_detail}")
# Non-fatal: /Create /F below may still replace it. Keep the detail in
# the final error if creation also fails.
# password" variant; if that fails, retry without /RU /NP /IT.
base = [
"/Create",
@ -317,6 +427,8 @@ def _install_scheduled_task(task_name: str, script_path: Path) -> tuple[bool, st
if code == 0:
return (True, f"Created Scheduled Task {task_name!r}")
last_code, last_err = code, (err or out or "")
if delete_detail and "cannot find" not in delete_detail.lower():
last_err = f"{last_err.strip()} (delete detail: {delete_detail})"
return (False, f"schtasks /Create failed (code {last_code}): {last_err.strip()}")
@ -417,7 +529,7 @@ def _build_gateway_argv() -> tuple[list[str], str, dict[str, str]]:
argv = [python_exe, "-m", "hermes_cli.main"]
if profile_arg:
argv.extend(profile_arg.split())
argv.extend(["gateway", "run", "--replace"])
argv.extend(["gateway", "run"])
env_overlay = {
"HERMES_HOME": hermes_home,
@ -432,7 +544,7 @@ def _build_gateway_argv() -> tuple[list[str], str, dict[str, str]]:
def _spawn_detached(script_path: Path | None = None) -> int:
"""Launch the gateway as a fully detached background process.
We spawn ``pythonw.exe -m hermes_cli.main gateway run --replace``
We spawn ``pythonw.exe -m hermes_cli.main gateway run``
directly — NOT through a cmd.exe shim — because on Windows a cmd.exe
child inherits the parent session's console handle and tends to get
reaped when the spawning shell exits. pythonw.exe has no console, and
@ -505,7 +617,78 @@ def _spawn_detached(script_path: Path | None = None) -> int:
return proc.pid
def install(force: bool = False) -> None:
def _install_choice_from_env(name: str) -> bool | None:
raw = os.environ.get(name)
if raw is None:
return None
value = raw.strip().lower()
if value in {"1", "true", "yes", "y", "on"}:
return True
if value in {"0", "false", "no", "n", "off"}:
return False
return None
def _prompt_install_choices(
start_now: bool | None = None,
start_on_login: bool | None = None,
) -> tuple[bool, bool]:
"""Return (start_now, start_on_login), asking before any UAC escalation."""
env_start_now = _install_choice_from_env("HERMES_GATEWAY_INSTALL_START_NOW")
env_start_on_login = _install_choice_from_env("HERMES_GATEWAY_INSTALL_START_ON_LOGIN")
if start_now is None:
start_now = env_start_now
if start_on_login is None:
start_on_login = env_start_on_login
if start_now is not None and start_on_login is not None:
return start_now, start_on_login
from hermes_cli.setup import prompt_yes_no
if start_now is None:
start_now = prompt_yes_no("Start the gateway now after install?", True)
if start_on_login is None:
start_on_login = prompt_yes_no(
"Start the gateway automatically on Windows login with a Scheduled Task?",
True,
)
return start_now, start_on_login
def _install_startup_fallback(script_path: Path, start_now: bool, detail: str) -> None:
"""Install the Startup-folder fallback and optionally start once."""
print(f"↻ Scheduled Task install blocked ({detail.splitlines()[0]}) — using Startup folder fallback")
entry = _install_startup_entry(script_path)
print(f"✓ Installed Windows login item: {entry}")
print(f" Task script: {script_path}")
# Re-running `hermes -p <profile> gateway install` must be safe.
# Startup-folder fallback only installs login persistence. Starting is
# controlled by the pre-UAC start_now answer so all user decisions happen
# before any elevation prompt.
from hermes_cli.gateway import find_gateway_pids, _profile_arg
running_pids = list(find_gateway_pids())
if running_pids:
print(f"✓ Gateway already running (PID: {', '.join(map(str, running_pids))})")
elif start_now:
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")
else:
profile_arg = _profile_arg()
start_cmd = f"hermes {profile_arg} gateway start" if profile_arg else "hermes gateway start"
print("ℹ Startup fallback installed; gateway not started now.")
print(f" Start manually with: {start_cmd}")
_print_next_steps()
def install(
force: bool = False,
*,
start_now: bool | None = None,
start_on_login: bool | None = None,
elevated_handoff: bool = False,
) -> None:
"""Install the gateway as a Windows Scheduled Task (with Startup fallback).
Idempotent: re-running updates the task to point at the current python/
@ -513,35 +696,111 @@ def install(force: bool = False) -> None:
/ ``systemd_install`` but isn't needed — we always reconcile.
"""
_assert_windows()
start_now, start_on_login = _prompt_install_choices(start_now, start_on_login)
if not start_on_login:
print("ℹ Skipped Windows login auto-start install.")
if start_now:
running_pids = _gateway_pids()
if running_pids:
print(f"✓ Gateway already running (PID: {', '.join(map(str, running_pids))})")
else:
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")
else:
print("ℹ Gateway not started and no auto-start service installed.")
print(" Run later with: hermes gateway start")
return
task_name = get_task_name()
script_path = _write_task_script()
# On machines where the current user's scheduled-task ACL is locked down,
# schtasks /Create or /Change can sit for the timeout before returning
# Access Denied. We already collected all intent questions above, so avoid
# a mysterious post-question pause: ask for UAC before touching schtasks.
if not _is_running_as_admin() and not elevated_handoff:
from hermes_cli.setup import prompt_yes_no
print("↻ Scheduled Task install may need administrator approval on this Windows account.")
print(" UAC is Windows' admin approval prompt; it is needed to create/update the Scheduled Task.")
if prompt_yes_no(" Open the UAC prompt now?", False):
if _launch_elevated_install(force=force, start_now=start_now, start_on_login=start_on_login):
print("✓ Launched elevated Hermes gateway install prompt.")
if start_now:
print(" Approve the Windows UAC prompt; the elevated install will start the gateway afterwards.")
else:
print(" Approve the Windows UAC prompt, then run: hermes gateway status")
return
print("⚠ Falling back to Startup folder because elevation was unavailable or cancelled.")
else:
print(" Skipped elevation. Falling back to Startup folder.")
_install_startup_fallback(script_path, start_now, "administrator approval was not used")
return
ok, detail = _install_scheduled_task(task_name, script_path)
if ok:
print(f"✓ {detail}")
print(f" Task script: {script_path}")
# Start it now so the user doesn't have to log off/on.
run_code, _out, run_err = _exec_schtasks(["/Run", "/TN", task_name])
if run_code == 0:
_report_gateway_start("Scheduled Task")
print("ℹ Gateway auto-start installed for Windows login.")
if start_now:
running_pids = _gateway_pids()
if running_pids:
print(f"✓ Gateway already running (PID: {', '.join(map(str, running_pids))})")
else:
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")
else:
# Scheduled Task was created but /Run failed (e.g. the task's
# action is malformed). Spawn directly as a backstop.
pid = _spawn_detached(script_path)
_report_gateway_start(
f"direct spawn (PID {pid}; schtasks /Run said: {run_err.strip()})"
)
print("ℹ Gateway not started now.")
print(" Start manually with: hermes gateway start")
_print_next_steps()
return
# schtasks create didn't work. Prefer a real Scheduled Task over the
# Startup-folder fallback when the only blocker is elevation. This gives
# users a UAC prompt instead of silently installing a less reliable login
# item, and keeps the fallback for locked-down boxes / cancelled prompts.
if _is_access_denied(detail) and not _is_running_as_admin():
from hermes_cli.setup import prompt_yes_no
print(f"↻ Scheduled Task install needs administrator approval ({detail.splitlines()[0]})")
print(" UAC is Windows' admin approval prompt; it is needed to create/update the Scheduled Task.")
if prompt_yes_no(" Open the UAC prompt now?", False):
if _launch_elevated_install(force=force, start_now=start_now, start_on_login=start_on_login):
print("✓ Launched elevated Hermes gateway install prompt.")
if start_now:
print(" Approve the Windows UAC prompt; the elevated install will start the gateway afterwards.")
else:
print(" Approve the Windows UAC prompt, then run: hermes gateway status")
return
print("⚠ Falling back to Startup folder because elevation was unavailable or cancelled.")
else:
print(" Skipped elevation. Falling back to Startup folder.")
# schtasks create didn't work. See if it's a "fall back to startup" case.
if _should_fall_back(1, detail):
print(f"↻ Scheduled Task install blocked ({detail.splitlines()[0]}) — using Startup folder fallback")
entry = _install_startup_entry(script_path)
pid = _spawn_detached(script_path)
print(f"✓ Installed Windows login item: {entry}")
print(f" Task script: {script_path}")
_report_gateway_start(f"direct spawn (PID {pid})")
# Re-running `hermes -p <profile> gateway install` must be safe.
# Startup-folder fallback only installs login persistence. Starting is
# controlled by the pre-UAC start_now answer so all user decisions happen
# before any elevation prompt.
from hermes_cli.gateway import find_gateway_pids, _profile_arg
running_pids = list(find_gateway_pids())
if running_pids:
print(f"✓ Gateway already running (PID: {', '.join(map(str, running_pids))})")
elif start_now:
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")
else:
profile_arg = _profile_arg()
start_cmd = f"hermes {profile_arg} gateway start" if profile_arg else "hermes gateway start"
print("ℹ Startup fallback installed; gateway not started now.")
print(f" Start manually with: {start_cmd}")
_print_next_steps()
return
@ -595,12 +854,28 @@ def uninstall() -> None:
script_path = get_task_script_path()
startup_entry = get_startup_entry_path()
scheduled_task_removed = False
if is_task_registered():
code, _out, err = _exec_schtasks(["/Delete", "/F", "/TN", task_name])
detail = err.strip()
if code == 0:
scheduled_task_removed = True
print(f"✓ Removed Scheduled Task {task_name!r}")
elif _is_access_denied(detail) and not _is_running_as_admin():
from hermes_cli.setup import prompt_yes_no
print(f"↻ Scheduled Task uninstall needs administrator approval ({detail or 'access denied'})")
print(" UAC is Windows' admin approval prompt; it is needed to remove the Scheduled Task.")
if prompt_yes_no(" Open the UAC prompt now?", False):
if _launch_elevated_uninstall():
print("✓ Launched elevated Hermes gateway uninstall prompt.")
print(" Approve the Windows UAC prompt, then run: hermes gateway status")
return
print("⚠ Elevated uninstall prompt was unavailable or cancelled.")
else:
print(" Skipped elevation. Scheduled Task was not removed.")
else:
print(f"⚠ schtasks /Delete returned code {code}: {err.strip()}")
print(f"⚠ schtasks /Delete returned code {code}: {detail}")
for path, label in [(startup_entry, "Windows login item"), (script_path, "Task script")]:
try:
@ -609,6 +884,9 @@ def uninstall() -> None:
except FileNotFoundError:
pass
if is_task_registered() and not scheduled_task_removed:
print(f"⚠ Scheduled Task still registered: {task_name}")
# ---------------------------------------------------------------------------
# Status / start / stop / restart
@ -697,14 +975,37 @@ def status(deep: bool = False) -> None:
def start() -> None:
"""Start the gateway. Prefers /Run on the scheduled task if present."""
_assert_windows()
if is_task_registered():
running_pids = _gateway_pids()
if running_pids:
print(f"✓ Gateway already running (PID: {', '.join(map(str, running_pids))})")
return
task_installed = is_task_registered()
startup_installed = is_startup_entry_installed()
if not task_installed and not startup_installed:
from hermes_cli.setup import prompt_yes_no
print("✗ Gateway service is not installed")
if not prompt_yes_no(" Install it now so the gateway starts on login?", True):
print(" Run: hermes gateway install")
return
install(force=False)
task_installed = is_task_registered()
startup_installed = is_startup_entry_installed()
if not task_installed and not startup_installed:
print("⚠ Gateway install did not complete in this process.")
print(" If a UAC prompt opened, approve it, then run: hermes gateway start")
return
if task_installed:
code, _out, err = _exec_schtasks(["/Run", "/TN", get_task_name()])
if code == 0:
_report_gateway_start(f"Scheduled Task {get_task_name()!r}")
return
print(f"⚠ schtasks /Run failed (code {code}): {err.strip()} — falling back to direct spawn")
# Direct spawn — no script_path needed with the new argv-based spawner.
# Startup fallback or failed /Run: direct spawn one foreground-detached gateway.
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")

32
hermes_cli/main.py
View file

@ -10553,6 +10553,38 @@ def main():
dest="run_as_user",
help="User account the Linux system service should run as",
)
gateway_install.add_argument(
"--start-now",
dest="start_now",
action="store_true",
default=None,
help=argparse.SUPPRESS,
)
gateway_install.add_argument(
"--no-start-now",
dest="start_now",
action="store_false",
help=argparse.SUPPRESS,
)
gateway_install.add_argument(
"--start-on-login",
dest="start_on_login",
action="store_true",
default=None,
help=argparse.SUPPRESS,
)
gateway_install.add_argument(
"--no-start-on-login",
dest="start_on_login",
action="store_false",
help=argparse.SUPPRESS,
)
gateway_install.add_argument(
"--elevated-handoff",
dest="elevated_handoff",
action="store_true",
help=argparse.SUPPRESS,
)
# gateway uninstall
gateway_uninstall = gateway_subparsers.add_parser(

85
tests/hermes_cli/test_gateway.py
View file

@ -237,11 +237,13 @@ def test_gateway_install_in_container_with_operational_systemd_uses_systemd(monk
monkeypatch.setattr(gateway, "is_managed", lambda: False)
calls = []
monkeypatch.setattr(gateway, "prompt_yes_no", lambda question, default=True: calls.append(("prompt", question, default)) or True)
monkeypatch.setattr(
gateway,
"systemd_install",
lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)),
lambda force=False, system=False, run_as_user=None, enable_on_startup=True: calls.append(("install", force, system, run_as_user, enable_on_startup)),
)
monkeypatch.setattr(gateway, "systemd_start", lambda system=False: calls.append(("start", system)))
args = SimpleNamespace(
gateway_command="install",
@ -251,7 +253,12 @@ def test_gateway_install_in_container_with_operational_systemd_uses_systemd(monk
)
gateway.gateway_command(args)
assert calls == [(False, False, None)]
assert calls == [
("prompt", "Start the gateway now after installing the service?", True),
("prompt", "Start the gateway automatically on login/boot with systemd?", True),
("install", False, False, None, True),
("start", False),
]
def test_gateway_start_in_container_with_operational_systemd_uses_systemd(monkeypatch):
@ -386,6 +393,34 @@ def test_systemd_install_checks_linger_status(monkeypatch, tmp_path, capsys):
assert "User service installed and enabled" in out
def test_systemd_install_can_skip_enable_on_startup(monkeypatch, tmp_path, capsys):
unit_path = tmp_path / "systemd" / "user" / "hermes-gateway.service"
monkeypatch.setattr(gateway, "get_systemd_unit_path", lambda system=False: unit_path)
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_user_systemd_env", lambda: None)
monkeypatch.setattr(gateway, "_ensure_linger_enabled", lambda: helper_calls.append(True))
gateway.systemd_install(force=False, enable_on_startup=False)
out = capsys.readouterr().out
assert unit_path.exists()
assert [cmd for cmd, _ in calls] == [
["systemctl", "--user", "daemon-reload"],
]
assert helper_calls == [True]
assert "User service installed!" in out
assert "installed and enabled" not 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"
@ -466,13 +501,55 @@ def test_install_linux_gateway_from_setup_system_choice_as_root_installs(monkeyp
monkeypatch.setattr(
gateway,
"systemd_install",
lambda force=False, system=False, run_as_user=None: calls.append((force, system, run_as_user)),
lambda force=False, system=False, run_as_user=None, enable_on_startup=True: calls.append((force, system, run_as_user, enable_on_startup)),
)
scope, did_install = gateway.install_linux_gateway_from_setup(force=True)
assert (scope, did_install) == ("system", True)
assert calls == [(True, True, "alice")]
assert calls == [(True, True, "alice", True)]
def test_install_linux_gateway_from_setup_passes_startup_choice(monkeypatch):
monkeypatch.setattr(gateway, "prompt_linux_gateway_install_scope", lambda: "user")
calls = []
monkeypatch.setattr(
gateway,
"systemd_install",
lambda force=False, system=False, run_as_user=None, enable_on_startup=True: calls.append((force, system, run_as_user, enable_on_startup)),
)
scope, did_install = gateway.install_linux_gateway_from_setup(force=False, enable_on_startup=False)
assert (scope, did_install) == ("user", True)
assert calls == [(False, False, None, False)]
def test_gateway_install_can_decline_start_now_and_startup(monkeypatch):
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: True)
monkeypatch.setattr(gateway, "is_wsl", lambda: False)
monkeypatch.setattr(gateway, "is_macos", lambda: False)
monkeypatch.setattr(gateway, "is_managed", lambda: False)
answers = iter([False, False])
calls = []
monkeypatch.setattr(gateway, "prompt_yes_no", lambda question, default=True: calls.append(("prompt", question, default)) or next(answers))
monkeypatch.setattr(
gateway,
"systemd_install",
lambda force=False, system=False, run_as_user=None, enable_on_startup=True: calls.append(("install", force, system, run_as_user, enable_on_startup)),
)
monkeypatch.setattr(gateway, "systemd_start", lambda system=False: calls.append(("start", system)))
args = SimpleNamespace(gateway_command="install", force=True, system=False, run_as_user=None)
gateway.gateway_command(args)
assert calls == [
("prompt", "Start the gateway now after installing the service?", True),
("prompt", "Start the gateway automatically on login/boot with systemd?", True),
("install", True, False, None, False),
]
def test_find_gateway_pids_falls_back_to_pid_file_when_process_scan_fails(monkeypatch):

423
tests/hermes_cli/test_gateway_windows.py
View file

@ -1,8 +1,12 @@
"""Tests for the Windows gateway backend."""
"""Tests for hermes_cli.gateway_windows."""
from pathlib import Path
import pytest
import hermes_cli.gateway as gateway
import hermes_cli.gateway_windows as gateway_windows
import hermes_cli.setup as setup
@pytest.mark.parametrize(
@ -61,3 +65,420 @@ def test_build_gateway_argv_uses_base_pythonw_for_uv_venv_launcher(monkeypatch,
assert env_overlay["VIRTUAL_ENV"] == str(project / "venv")
assert str(project) in env_overlay["PYTHONPATH"].split(gateway_windows.os.pathsep)
assert str(site_packages) in env_overlay["PYTHONPATH"].split(gateway_windows.os.pathsep)
def _arrange_startup_fallback(monkeypatch, tmp_path, running_pids):
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
startup_entry = tmp_path / "Startup" / "Hermes_Gateway_alice.cmd"
calls = []
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "_write_task_script", lambda: script_path)
monkeypatch.setattr(
gateway_windows,
"_install_scheduled_task",
lambda task_name, script_path: (
False,
"schtasks /Create failed (code 1): ERROR: Access is denied.",
),
)
monkeypatch.setattr(gateway_windows, "_should_fall_back", lambda code, detail: True)
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: True)
monkeypatch.setattr(
gateway_windows,
"_launch_elevated_install",
lambda force=False, start_now=None, start_on_login=None: calls.append(("elevate", force, start_now, start_on_login)) or True,
)
def fake_install_startup_entry(path: Path) -> Path:
calls.append(("install_startup", path))
return startup_entry
monkeypatch.setattr(gateway_windows, "_install_startup_entry", fake_install_startup_entry)
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda path: calls.append(("spawn", path)) or 12345)
monkeypatch.setattr(gateway_windows, "_report_gateway_start", lambda via: calls.append(("report_start", via)))
monkeypatch.setattr(gateway_windows, "_print_next_steps", lambda: calls.append(("next_steps", None)))
monkeypatch.setattr(gateway, "find_gateway_pids", lambda: running_pids)
monkeypatch.setattr(gateway, "_profile_arg", lambda: "--profile alice")
return script_path, calls
def test_gateway_cmd_script_uses_pythonw_without_replace_or_start_churn(monkeypatch):
"""Scheduled Task wrapper should launch pythonw once and avoid replace loops."""
monkeypatch.setattr(gateway_windows, "_derive_venv_pythonw", lambda exe: exe.replace("python.exe", "pythonw.exe"))
content = gateway_windows._build_gateway_cmd_script(
r"C:\\Hermes\\hermes-agent\\venv\\Scripts\\python.exe",
r"C:\\Hermes\\hermes-agent",
r"C:\\HermesHome\\profiles\\alice",
"--profile alice",
)
assert "pythonw.exe" in content
assert "gateway run" in content
assert "--replace" not in content
assert "start \"\"" not in content
assert "exit /b 0" in content
def test_elevated_gateway_command_uses_pythonw_hidden_console(monkeypatch):
"""UAC handoff should not leave a second elevated cmd.exe window open."""
calls = []
class FakeShell32:
def ShellExecuteW(self, hwnd, verb, executable, params, cwd, show):
calls.append((hwnd, verb, executable, params, cwd, show))
return 33
class FakeWindll:
shell32 = FakeShell32()
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "_current_profile_cli_args", lambda: ["--profile", "alice"])
monkeypatch.setattr(gateway_windows, "_derive_venv_pythonw", lambda exe: exe.replace("python.exe", "pythonw.exe"))
monkeypatch.setattr(gateway_windows.sys, "executable", r"C:\Hermes\venv\Scripts\python.exe")
monkeypatch.setattr(gateway_windows.ctypes, "windll", FakeWindll())
assert gateway_windows._launch_elevated_gateway_command("install", ["--start-now", "--elevated-handoff"])
assert len(calls) == 1
_hwnd, verb, executable, params, cwd, show = calls[0]
assert verb == "runas"
assert executable.endswith("pythonw.exe")
assert "--profile alice gateway install --start-now --elevated-handoff" in params
assert show == 0
assert cwd
def test_install_scheduled_task_recreates_instead_of_change(monkeypatch, tmp_path):
"""Install must delete+create so stale minute-repeat task settings are not preserved."""
calls = []
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
def fake_schtasks(args):
calls.append(tuple(args))
if args[0] == "/Delete":
return (0, "SUCCESS", "")
if args[0] == "/Create":
return (0, "SUCCESS", "")
raise AssertionError(f"unexpected schtasks args: {args}")
monkeypatch.setattr(gateway_windows, "_exec_schtasks", fake_schtasks)
ok, detail = gateway_windows._install_scheduled_task("Hermes_Gateway_alice", script_path)
assert ok is True
assert "/Change" not in [arg for call in calls for arg in call]
assert calls[0][:4] == ("/Delete", "/F", "/TN", "Hermes_Gateway_alice")
assert calls[1][0] == "/Create"
assert "/SC" in calls[1]
assert "ONLOGON" in calls[1]
def test_install_scheduled_task_success_start_now_uses_direct_spawn_not_task_run(monkeypatch, tmp_path, capsys):
"""Install start-now should not /Run the task; that preserved old restart loops."""
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
calls = []
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (True, True))
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: True)
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "_write_task_script", lambda: script_path)
monkeypatch.setattr(
gateway_windows,
"_install_scheduled_task",
lambda task_name, script_path: (True, "Created Scheduled Task 'Hermes_Gateway_alice'"),
)
monkeypatch.setattr(gateway_windows, "_gateway_pids", lambda: [])
monkeypatch.setattr(gateway_windows, "_exec_schtasks", lambda args: calls.append(("schtasks", tuple(args))) or (0, "", ""))
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda path=None: calls.append(("spawn", path)) or 12345)
monkeypatch.setattr(gateway_windows, "_report_gateway_start", lambda via: calls.append(("report_start", via)))
monkeypatch.setattr(gateway_windows, "_print_next_steps", lambda: calls.append(("next_steps", None)))
gateway_windows.install(force=False)
assert not any(call[0] == "schtasks" and "/Run" in call[1] for call in calls)
assert ("spawn", None) in calls
assert any(call[0] == "report_start" for call in calls)
out = capsys.readouterr().out
assert "auto-start installed for Windows login" in out
def test_install_scheduled_task_success_does_not_auto_start(monkeypatch, tmp_path, capsys):
"""Install should register/update the task only; start is explicit."""
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
calls = []
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: True)
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "_write_task_script", lambda: script_path)
monkeypatch.setattr(
gateway_windows,
"_install_scheduled_task",
lambda task_name, script_path: (True, "Created Scheduled Task 'Hermes_Gateway_alice'"),
)
monkeypatch.setattr(gateway_windows, "_exec_schtasks", lambda args: calls.append(("schtasks", tuple(args))) or (0, "", ""))
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda path=None: calls.append(("spawn", path)) or 12345)
monkeypatch.setattr(gateway_windows, "_report_gateway_start", lambda via: calls.append(("report_start", via)))
monkeypatch.setattr(gateway_windows, "_print_next_steps", lambda: calls.append(("next_steps", None)))
gateway_windows.install(force=False)
assert not any(call[0] == "schtasks" and "/Run" in call[1] for call in calls)
assert not any(call[0] == "spawn" for call in calls)
assert not any(call[0] == "report_start" for call in calls)
assert ("next_steps", None) in calls
out = capsys.readouterr().out
assert "auto-start installed for Windows login" in out
def test_install_access_denied_launches_elevated_install_before_startup_fallback(monkeypatch, tmp_path, capsys):
"""Non-admin Scheduled Task access denied should hand off to UAC elevation."""
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
calls = []
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "_write_task_script", lambda: script_path)
monkeypatch.setattr(
gateway_windows,
"_install_scheduled_task",
lambda task_name, script_path: (
False,
"schtasks /Create failed (code 1): ERROR: Access is denied.",
),
)
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: False)
monkeypatch.setattr(
gateway_windows,
"_launch_elevated_install",
lambda force=False, start_now=None, start_on_login=None: calls.append(("elevate", force, start_now, start_on_login)) or True,
)
monkeypatch.setattr(setup, "prompt_yes_no", lambda prompt, default=True: calls.append(("prompt", prompt, default)) or True)
monkeypatch.setattr(gateway_windows, "_install_startup_entry", lambda path: calls.append(("install_startup", path)) or path)
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda path=None: calls.append(("spawn", path)) or 12345)
gateway_windows.install(force=True)
assert calls == [("prompt", " Open the UAC prompt now?", False), ("elevate", True, False, True)]
out = capsys.readouterr().out
assert "administrator approval" in out
assert "UAC is Windows' admin approval prompt" in out
assert "Launched elevated Hermes gateway install prompt" in out
def test_install_prompts_start_choices_before_uac(monkeypatch, tmp_path, capsys):
"""Windows install asks start-now and auto-start before any UAC handoff."""
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
calls = []
answers = iter([True, True, True])
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "_write_task_script", lambda: script_path)
monkeypatch.setattr(
gateway_windows,
"_install_scheduled_task",
lambda task_name, script_path: (
False,
"schtasks /Create failed (code 1): ERROR: Access is denied.",
),
)
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: False)
monkeypatch.setattr(setup, "prompt_yes_no", lambda prompt, default=True: calls.append(("prompt", prompt, default)) or next(answers))
monkeypatch.setattr(
gateway_windows,
"_launch_elevated_install",
lambda force=False, start_now=None, start_on_login=None: calls.append(("elevate", force, start_now, start_on_login)) or True,
)
gateway_windows.install(force=False)
assert calls == [
("prompt", "Start the gateway now after install?", True),
("prompt", "Start the gateway automatically on Windows login with a Scheduled Task?", True),
("prompt", " Open the UAC prompt now?", False),
("elevate", False, True, True),
]
out = capsys.readouterr().out
assert "elevated install will start the gateway afterwards" in out
def test_install_start_now_without_login_autostart_never_escalates(monkeypatch, capsys):
"""If auto-start is declined, install can start directly without touching schtasks/UAC."""
calls = []
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (True, False))
monkeypatch.setattr(gateway_windows, "_gateway_pids", lambda: [])
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda path=None: calls.append(("spawn", path)) or 12345)
monkeypatch.setattr(gateway_windows, "_report_gateway_start", lambda via: calls.append(("report_start", via)))
monkeypatch.setattr(gateway_windows, "_install_scheduled_task", lambda *args, **kwargs: calls.append(("install_task", args)) or (True, "should not happen"))
monkeypatch.setattr(gateway_windows, "_launch_elevated_install", lambda *args, **kwargs: calls.append(("elevate", args, kwargs)) or True)
gateway_windows.install(force=False)
assert not any(call[0] in {"install_task", "elevate"} for call in calls)
assert ("spawn", None) in calls
assert any(call[0] == "report_start" for call in calls)
out = capsys.readouterr().out
assert "Skipped Windows login auto-start install" in out
def test_start_noops_when_gateway_already_running(monkeypatch, capsys):
"""Repeated start should not invoke schtasks /Run or spawn another process."""
calls = []
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "_gateway_pids", lambda: [27128])
monkeypatch.setattr(gateway_windows, "is_task_registered", lambda: calls.append("task_check") or True)
monkeypatch.setattr(gateway_windows, "_exec_schtasks", lambda args: calls.append(("schtasks", tuple(args))) or (0, "", ""))
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda path=None: calls.append(("spawn", path)) or 12345)
gateway_windows.start()
assert calls == []
out = capsys.readouterr().out
assert "already running" in out
assert "27128" in out
def test_install_startup_fallback_does_not_spawn_when_gateway_already_running(monkeypatch, tmp_path, capsys):
"""Repeated Windows fallback installs should not spawn duplicate gateways."""
script_path, calls = _arrange_startup_fallback(monkeypatch, tmp_path, [24476])
gateway_windows.install(force=False)
assert ("install_startup", script_path) in calls
assert not any(call[0] == "spawn" for call in calls)
assert not any(call[0] == "report_start" for call in calls)
assert ("next_steps", None) in calls
out = capsys.readouterr().out
assert "already running" in out
assert "24476" in out
def test_install_startup_fallback_does_not_auto_spawn_when_gateway_stopped(monkeypatch, tmp_path, capsys):
"""Startup fallback install should only install login item, not launch pythonw."""
script_path, calls = _arrange_startup_fallback(monkeypatch, tmp_path, [])
gateway_windows.install(force=False)
assert ("install_startup", script_path) in calls
assert not any(call[0] == "spawn" for call in calls)
assert not any(call[0] == "report_start" for call in calls)
assert ("next_steps", None) in calls
out = capsys.readouterr().out
assert "gateway not started now" in out
assert "hermes --profile alice gateway start" in out
def test_install_access_denied_declined_elevation_uses_startup_fallback(monkeypatch, tmp_path, capsys):
"""Install should ask before UAC; declining keeps the non-jarring fallback path."""
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
calls = []
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "_write_task_script", lambda: script_path)
monkeypatch.setattr(
gateway_windows,
"_install_scheduled_task",
lambda task_name, script_path: (
False,
"schtasks /Create failed (code 1): ERROR: Access is denied.",
),
)
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: False)
monkeypatch.setattr(setup, "prompt_yes_no", lambda prompt, default=True: calls.append(("prompt", prompt, default)) or False)
monkeypatch.setattr(
gateway_windows,
"_launch_elevated_install",
lambda force=False, start_now=None, start_on_login=None: calls.append(("elevate", force, start_now, start_on_login)) or True,
)
monkeypatch.setattr(gateway_windows, "_install_startup_entry", lambda path: calls.append(("install_startup", path)) or path)
monkeypatch.setattr(gateway, "find_gateway_pids", lambda: [])
monkeypatch.setattr(gateway, "_profile_arg", lambda: "--profile alice")
monkeypatch.setattr(gateway_windows, "_print_next_steps", lambda: calls.append(("next_steps", None)))
gateway_windows.install(force=False)
assert ("prompt", " Open the UAC prompt now?", False) in calls
assert not any(call[0] == "elevate" for call in calls)
assert ("install_startup", script_path) in calls
out = capsys.readouterr().out
assert "Skipped elevation" in out
assert "UAC is Windows' admin approval prompt" in out
def test_uninstall_access_denied_prompts_before_elevating(monkeypatch, tmp_path, capsys):
"""Uninstall should hand off to an elevated uninstall only after user consent."""
calls = []
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
startup_entry = tmp_path / "Startup" / "Hermes_Gateway_alice.cmd"
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "get_task_script_path", lambda: script_path)
monkeypatch.setattr(gateway_windows, "get_startup_entry_path", lambda: startup_entry)
monkeypatch.setattr(gateway_windows, "is_task_registered", lambda: True)
monkeypatch.setattr(
gateway_windows,
"_exec_schtasks",
lambda args: calls.append(("schtasks", tuple(args))) or (1, "", "ERROR: Access is denied."),
)
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: False)
monkeypatch.setattr(setup, "prompt_yes_no", lambda prompt, default=True: calls.append(("prompt", prompt, default)) or True)
monkeypatch.setattr(gateway_windows, "_launch_elevated_uninstall", lambda: calls.append(("elevate_uninstall", None)) or True)
gateway_windows.uninstall()
assert ("prompt", " Open the UAC prompt now?", False) in calls
assert ("elevate_uninstall", None) in calls
out = capsys.readouterr().out
assert "uninstall needs administrator approval" in out
assert "UAC is Windows' admin approval prompt" in out
assert "Launched elevated Hermes gateway uninstall prompt" in out
def test_uninstall_access_denied_declined_keeps_task_and_cleans_files(monkeypatch, tmp_path, capsys):
"""Declining UAC should not surprise the user, but should still remove user-writable artifacts."""
calls = []
script_path = tmp_path / "Hermes_Gateway_alice.cmd"
startup_entry = tmp_path / "Startup" / "Hermes_Gateway_alice.cmd"
startup_entry.parent.mkdir(parents=True)
script_path.write_text("task", encoding="utf-8")
startup_entry.write_text("startup", encoding="utf-8")
monkeypatch.setattr(gateway_windows, "_prompt_install_choices", lambda *args, **kwargs: (False, True))
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_name", lambda: "Hermes_Gateway_alice")
monkeypatch.setattr(gateway_windows, "get_task_script_path", lambda: script_path)
monkeypatch.setattr(gateway_windows, "get_startup_entry_path", lambda: startup_entry)
monkeypatch.setattr(gateway_windows, "is_task_registered", lambda: True)
monkeypatch.setattr(
gateway_windows,
"_exec_schtasks",
lambda args: calls.append(("schtasks", tuple(args))) or (1, "", "ERROR: Access is denied."),
)
monkeypatch.setattr(gateway_windows, "_is_running_as_admin", lambda: False)
monkeypatch.setattr(setup, "prompt_yes_no", lambda prompt, default=True: calls.append(("prompt", prompt, default)) or False)
monkeypatch.setattr(gateway_windows, "_launch_elevated_uninstall", lambda: calls.append(("elevate_uninstall", None)) or True)
gateway_windows.uninstall()
assert not any(call[0] == "elevate_uninstall" for call in calls)
assert not script_path.exists()
assert not startup_entry.exists()
out = capsys.readouterr().out
assert "Skipped elevation" in out
assert "UAC is Windows' admin approval prompt" in out
assert "Scheduled Task still registered" in out