diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 82f6a5c7706..24b458935c1 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -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)") diff --git a/hermes_cli/gateway_windows.py b/hermes_cli/gateway_windows.py index 590778ac80a..77ea60d9b39 100644 --- a/hermes_cli/gateway_windows.py +++ b/hermes_cli/gateway_windows.py @@ -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 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 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})") diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 8f24360aadd..5d0f57b20f0 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -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( diff --git a/tests/hermes_cli/test_gateway.py b/tests/hermes_cli/test_gateway.py index a655b88b110..d78dcc131af 100644 --- a/tests/hermes_cli/test_gateway.py +++ b/tests/hermes_cli/test_gateway.py @@ -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): diff --git a/tests/hermes_cli/test_gateway_windows.py b/tests/hermes_cli/test_gateway_windows.py index 2c464c8ca74..65d05d63679 100644 --- a/tests/hermes_cli/test_gateway_windows.py +++ b/tests/hermes_cli/test_gateway_windows.py @@ -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 \ No newline at end of file