1861 lines
83 KiB
Python
1861 lines
83 KiB
Python
"""Windows gateway service backend (Scheduled Task + Startup-folder fallback).
|
||
|
||
Mirrors the ``launchd_*`` / ``systemd_*`` contract. ``schtasks /Create ... /RL LIMITED`` runs at the
|
||
CURRENT USER's next logon without elevation. Manual starts and ``install --start-now`` use the direct
|
||
hidden-console launcher instead of ``schtasks /Run`` so start/restart behavior is consistent.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import json
|
||
import locale
|
||
import logging
|
||
import os
|
||
import re
|
||
import shlex
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from xml.etree import ElementTree
|
||
from xml.sax.saxutils import escape
|
||
|
||
from hermes_cli._subprocess_compat import (
|
||
_WINDOWS_GATEWAY_BREAKAWAY_ENV,
|
||
windows_detach_flags,
|
||
windows_detach_flags_without_breakaway,
|
||
windows_hide_flags,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Short timeouts: schtasks occasionally wedges and we don't want to hang forever.
|
||
_SCHTASKS_TIMEOUT_S = 15
|
||
# Patterns in schtasks stderr that mean "fall back to the Startup folder".
|
||
# schtasks' localized "access is denied" (en/es/cs/zh-Hans/zh-Hant/ja/ko): one vocabulary for both
|
||
# the elevated-install offer and the Startup-folder fallback.
|
||
_ACCESS_DENIED_WORDS = (
|
||
r"access is denied|acceso denegado|přístup byl odepřen|拒绝访问|拒絕存取|アクセスが拒否されました|"
|
||
r"액세스가 거부되었습니다"
|
||
)
|
||
_FALLBACK_PATTERNS = re.compile(
|
||
rf"({_ACCESS_DENIED_WORDS}|schtasks timed out|schtasks produced no output)", re.IGNORECASE
|
||
)
|
||
_ACCESS_DENIED_PATTERN = re.compile(rf"({_ACCESS_DENIED_WORDS})", re.IGNORECASE)
|
||
|
||
# Set by _spawn_detached() when the breakaway spawn failed and it retried WITHOUT
|
||
# CREATE_BREAKAWAY_FROM_JOB — the child stays in the parent's Job Object and may be killed when this
|
||
# shell exits. Dict (not bare bool) so the flag is mutable without ``global``.
|
||
_LAST_SPAWN_BREAKAWAY_FALLBACK: dict = {"fallback": False}
|
||
|
||
_TASK_NAME_DEFAULT = "Hermes_Gateway"
|
||
_TASK_DESCRIPTION = "Hermes Agent Gateway - Messaging Platform Integration"
|
||
_TASK_LOGON_DELAY = "PT30S"
|
||
_TASK_RESTART_INTERVAL = "PT1M"
|
||
_TASK_RESTART_COUNT = 999
|
||
|
||
_GATEWAY_ENV = (("PYTHONIOENCODING", "utf-8"), ("HERMES_GATEWAY_DETACHED", "1"), ("HERMES_SUPERVISED_CHILD", "1"))
|
||
|
||
|
||
def _schtasks_encoding() -> str:
|
||
"""Console encoding for ``schtasks.exe`` output: localized Windows emits the OEM/ANSI code page,
|
||
not UTF-8, and decoding with the wrong codec raised UnicodeDecodeError in subprocess' reader
|
||
threads. Prefer the locale's preferred encoding, fall back to UTF-8."""
|
||
try:
|
||
return locale.getpreferredencoding(False) or "utf-8"
|
||
except Exception:
|
||
return "utf-8"
|
||
|
||
|
||
def _windows_console_encodings() -> list[str]:
|
||
"""Code pages a console tool such as ``schtasks.exe`` writes to a pipe, most likely first: the
|
||
console output code page (65001 once ``configure_windows_stdio`` ran, else the OEM page), the OEM
|
||
page, then the ANSI page. Read from kernel32, so Python's UTF-8 mode cannot disguise them."""
|
||
kernel32 = getattr(getattr(ctypes, "windll", None), "kernel32", None)
|
||
if kernel32 is None:
|
||
return [_schtasks_encoding()]
|
||
pages: list[str] = []
|
||
for getter in ("GetConsoleOutputCP", "GetOEMCP", "GetACP"):
|
||
try:
|
||
code_page = int(getattr(kernel32, getter)())
|
||
except (AttributeError, OSError, ValueError):
|
||
continue
|
||
if code_page > 0 and f"cp{code_page}" not in pages:
|
||
pages.append(f"cp{code_page}")
|
||
return pages or [_schtasks_encoding()]
|
||
|
||
|
||
def _decode_schtasks_output(data: bytes) -> str:
|
||
"""Decode captured ``schtasks.exe`` bytes. schtasks writes the console/OEM code page even when this
|
||
process runs in UTF-8 mode (the launcher sets ``PYTHONUTF8=1``), so ``locale.getpreferredencoding``
|
||
is the wrong codec on a non-ASCII account path: ``C:\\Users\\方舟`` came back as ``<60><><EFBFBD><EFBFBD>`` and the
|
||
Scheduled-Task drift check could never settle (#116193). Strict UTF-8 first (ASCII and genuine
|
||
UTF-8 output pass; legacy multi-byte text fails loudly), then the native code pages, then a lossy
|
||
fallback so a reader thread never raises."""
|
||
for encoding in ("utf-8", *_windows_console_encodings()):
|
||
try:
|
||
return data.decode(encoding)
|
||
except (LookupError, UnicodeDecodeError):
|
||
continue
|
||
return data.decode("utf-8", errors="replace")
|
||
|
||
|
||
def _assert_windows() -> None:
|
||
if sys.platform != "win32":
|
||
raise RuntimeError("gateway_windows is Windows-only")
|
||
|
||
|
||
def _hermes_home() -> Path:
|
||
from hermes_cli.config import get_hermes_home
|
||
|
||
return Path(get_hermes_home())
|
||
|
||
|
||
def hermes_service_roots() -> tuple[str, ...]:
|
||
"""Directories a Hermes-owned SCM service binary lives under: the checkout (its ``venv`` included),
|
||
the running interpreter's ``Scripts`` dir (``hermes.exe`` shim) and the ``gateway-service`` launcher dir."""
|
||
project_root = Path(__file__).resolve().parent.parent
|
||
return (str(project_root), str(Path(sys.executable).parent), str(_hermes_home() / "gateway-service"))
|
||
|
||
|
||
def _normalize_windows_path(value: str) -> str:
|
||
return value.strip().lstrip('"').replace("\\", "/").rstrip("/").casefold()
|
||
|
||
|
||
def hermes_owns_windows_service(name: str, binpath: str, hermes_roots: tuple[str, ...]) -> bool:
|
||
"""Positive ownership of an SCM service: Hermes-named (``hermes*``) or its binary path starts under a
|
||
Hermes root. Pure so it is testable off-Windows. A Scheduled-Task-launched gateway descends from
|
||
``svchost.exe`` hosting ``Schedule``; without this gate the updater took Task Scheduler for the
|
||
gateway's supervisor and ``sc.exe stop Schedule`` aborted every update (#97208)."""
|
||
normalized_name = "".join(char for char in name.casefold() if char.isalnum())
|
||
if normalized_name.startswith("hermes"):
|
||
return True
|
||
candidate = _normalize_windows_path(binpath)
|
||
return any(candidate.startswith(_normalize_windows_path(root) + "/") for root in hermes_roots if root)
|
||
|
||
|
||
def _preserve_hermes_home_path(path: str | Path) -> str:
|
||
r"""Render Hermes-owned paths under the configured HERMES_HOME spelling.
|
||
|
||
``%LOCALAPPDATA%\hermes`` may be a symlink/junction to another drive; launcher files must not
|
||
bake in the resolved target for paths under HERMES_HOME.
|
||
"""
|
||
candidate = Path(path)
|
||
try:
|
||
home = _hermes_home()
|
||
resolved_home = home.resolve()
|
||
resolved_candidate = candidate.resolve()
|
||
home_key = os.path.normcase(str(resolved_home))
|
||
candidate_key = os.path.normcase(str(resolved_candidate))
|
||
if os.path.commonpath([home_key, candidate_key]) == home_key:
|
||
return str(home / os.path.relpath(str(resolved_candidate), str(resolved_home)))
|
||
except Exception:
|
||
pass
|
||
return str(candidate)
|
||
|
||
|
||
# ── Quoting helpers. cmd.exe (.cmd body), VBScript literals and schtasks /TR are three DIFFERENT
|
||
# parsers — never reuse one helper for another. The task XML path avoids /TR quoting entirely.
|
||
|
||
def _quote_cmd_script_arg(value: str) -> str:
|
||
"""Quote one argument INSIDE a .cmd file for cmd.exe: split on spaces/tabs outside double quotes,
|
||
embedded quotes doubled. Line breaks are refused — they'd end the logical command line."""
|
||
if "\r" in value or "\n" in value:
|
||
raise ValueError(f"refusing to quote value containing newline: {value!r}")
|
||
if not value:
|
||
return '""'
|
||
if not re.search(r'[ \t"]', value):
|
||
return value
|
||
return '"' + value.replace('"', '""') + '"'
|
||
|
||
|
||
def _quote_vbs_string(value: str) -> str:
|
||
"""VBScript double-quoted literal (embedded quote doubled; newline refused)."""
|
||
if "\r" in value or "\n" in value:
|
||
raise ValueError(f"refusing to quote VBScript value containing newline: {value!r}")
|
||
return '"' + value.replace('"', '""') + '"'
|
||
|
||
|
||
# ── schtasks.exe wrapper
|
||
|
||
def _exec_schtasks(args: list[str]) -> tuple[int, str, str]:
|
||
"""Run ``schtasks.exe`` with a hard timeout. Return (code, stdout, stderr); a wedge returns
|
||
code=124 with a synthetic stderr so the fallback regex matches."""
|
||
_assert_windows()
|
||
schtasks = shutil.which("schtasks")
|
||
if schtasks is None:
|
||
return (1, "", "schtasks.exe not found on PATH")
|
||
try:
|
||
# Bytes, decoded by _decode_schtasks_output: a non-UTF-8 status line must never surface a
|
||
# UnicodeDecodeError from subprocess' reader threads. CREATE_NO_WINDOW: no flashing console under a TUI.
|
||
proc = subprocess.run(
|
||
[schtasks, *args], capture_output=True, text=False,
|
||
timeout=_SCHTASKS_TIMEOUT_S, creationflags=windows_hide_flags(),
|
||
)
|
||
return (
|
||
proc.returncode,
|
||
_decode_schtasks_output(proc.stdout or b""),
|
||
_decode_schtasks_output(proc.stderr or b""),
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
return (124, "", f"schtasks timed out after {_SCHTASKS_TIMEOUT_S}s")
|
||
except OSError as e:
|
||
return (1, "", f"schtasks invocation failed: {e}")
|
||
|
||
|
||
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. The child is console
|
||
``python.exe`` with ``SW_HIDE``: it owns one hidden console its subprocesses (schtasks, taskkill)
|
||
inherit — no visible window and no per-descendant conhost flashes (the console-less pythonw.exe
|
||
alternative re-created #54220/#56747 for every descendant).
|
||
|
||
All operator decisions are already collected in the parent shell before this point. See #54220, #56747.
|
||
"""
|
||
_assert_windows()
|
||
args = ["-m", "hermes_cli.main", *_current_profile_cli_args(), "gateway", command, *(extra_args or [])]
|
||
params = subprocess.list2cmdline(args)
|
||
cwd = str(Path(__file__).resolve().parent.parent)
|
||
try:
|
||
result = ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, params, cwd, 0) # 0 = SW_HIDE
|
||
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."""
|
||
overrides = {"HERMES_GATEWAY_ELEVATED_HANDOFF": "1"}
|
||
extra_args = ["--elevated-handoff"]
|
||
if force:
|
||
extra_args.append("--force")
|
||
for choice, env_key, flag in (
|
||
(start_now, "HERMES_GATEWAY_INSTALL_START_NOW", "start-now"),
|
||
(start_on_login, "HERMES_GATEWAY_INSTALL_START_ON_LOGIN", "start-on-login"),
|
||
):
|
||
if choice is not None:
|
||
overrides[env_key] = "1" if choice else "0"
|
||
extra_args.append(f"--{flag}" if choice else f"--no-{flag}")
|
||
saved = {key: os.environ.get(key) for key in overrides}
|
||
try:
|
||
os.environ.update(overrides)
|
||
return _launch_elevated_gateway_command("install", extra_args)
|
||
finally:
|
||
for key, old in saved.items():
|
||
if old is None:
|
||
os.environ.pop(key, None)
|
||
else:
|
||
os.environ[key] = old
|
||
|
||
|
||
# ── Paths: where we stash our task script and where Startup lives
|
||
|
||
def get_task_name() -> str:
|
||
"""Scheduled Task name, scoped per profile."""
|
||
_assert_windows()
|
||
from hermes_cli.gateway import _profile_suffix # local: avoids circular init during boot
|
||
|
||
suffix = _profile_suffix()
|
||
return f"{_TASK_NAME_DEFAULT}_{suffix}" if suffix else _TASK_NAME_DEFAULT
|
||
|
||
|
||
def _sanitize_filename(value: str) -> str:
|
||
"""Remove characters illegal in Windows filenames."""
|
||
return re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", value)
|
||
|
||
|
||
def get_task_script_path() -> Path:
|
||
"""The generated ``gateway.cmd`` wrapper under ``<HERMES_HOME>/gateway-service/`` (per-profile
|
||
installs stay self-contained); the VBS launcher lives beside it."""
|
||
_assert_windows()
|
||
script_dir = _hermes_home() / "gateway-service"
|
||
script_dir.mkdir(parents=True, exist_ok=True)
|
||
return script_dir / f"{_sanitize_filename(get_task_name())}.cmd"
|
||
|
||
|
||
def _startup_dir() -> Path:
|
||
appdata = os.environ.get("APPDATA", "").strip()
|
||
if appdata:
|
||
return Path(appdata) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup"
|
||
userprofile = os.environ.get("USERPROFILE", "").strip() or os.environ.get("HOME", "").strip()
|
||
if not userprofile:
|
||
raise RuntimeError("neither APPDATA nor USERPROFILE is set — cannot resolve Startup folder")
|
||
return Path(userprofile).joinpath("AppData", "Roaming", "Microsoft", "Windows", "Start Menu", "Programs", "Startup")
|
||
|
||
|
||
def get_startup_entry_path() -> Path:
|
||
_assert_windows()
|
||
return _startup_dir() / f"{_sanitize_filename(get_task_name())}.vbs"
|
||
|
||
|
||
def _legacy_startup_entry_path() -> Path:
|
||
_assert_windows()
|
||
return _startup_dir() / f"{_sanitize_filename(get_task_name())}.cmd"
|
||
|
||
|
||
def _startup_staging_path() -> Path:
|
||
"""The Startup-folder staging file; also the debris a pre-fix failed swap left behind (#114093)."""
|
||
return get_startup_entry_path().with_suffix(".tmp")
|
||
|
||
|
||
def _stable_gateway_working_dir(project_root: Path) -> str:
|
||
"""Stable cwd for detached/startup runs: anchor at HERMES_HOME when it exists (mirrors the POSIX
|
||
service invariant) so a moved checkout/worktree can't fail the ``cd`` step; else the checkout."""
|
||
from hermes_cli.config import get_hermes_home
|
||
|
||
try:
|
||
home = get_hermes_home()
|
||
if home and Path(home).is_dir():
|
||
return str(Path(home))
|
||
except Exception:
|
||
pass
|
||
return str(project_root)
|
||
|
||
|
||
# ── Script rendering
|
||
|
||
def _gateway_run_argv(python_exe: str, profile_arg: str) -> list[str]:
|
||
"""``python -m hermes_cli.main [--profile X] gateway run`` — shared by every launcher renderer."""
|
||
argv = [python_exe, "-m", "hermes_cli.main"]
|
||
if profile_arg:
|
||
argv.extend(profile_arg.split())
|
||
argv.extend(["gateway", "run"])
|
||
return argv
|
||
|
||
|
||
def _launcher_settings(home: Path | None = None) -> tuple[str, str, str, str]:
|
||
"""Return (python_path, working_dir, hermes_home, profile_arg) for generated launchers.
|
||
``home`` targets another profile's HERMES_HOME (per-profile cold-start, #110959)."""
|
||
from hermes_cli.gateway import PROJECT_ROOT, _profile_arg, get_python_path # avoid circular init
|
||
|
||
hermes_home = str(home if home is not None else _hermes_home())
|
||
return (
|
||
_preserve_hermes_home_path(get_python_path()),
|
||
_stable_gateway_working_dir(PROJECT_ROOT),
|
||
hermes_home,
|
||
_profile_arg(hermes_home),
|
||
)
|
||
|
||
|
||
def _launcher_pythonpath_entries(extra_pythonpath: list[str]) -> list[str]:
|
||
return [
|
||
_preserve_hermes_home_path(Path(__file__).resolve().parent.parent),
|
||
*[_preserve_hermes_home_path(entry) for entry in extra_pythonpath],
|
||
]
|
||
|
||
|
||
def _build_gateway_cmd_script(python_path: str, working_dir: str, hermes_home: str, profile_arg: str) -> str:
|
||
"""Build the ``gateway.cmd`` wrapper (CRLF-terminated). No PATH overrides (rewriting PATH breaks
|
||
Homebrew/nvm-style installs), no ``start`` (extra wrapper process muddles lifecycle/status), no
|
||
``--replace`` (repeated /Run calls must be idempotent, not takeover loops)."""
|
||
python_exe_path, venv_dir, extra_pythonpath = _resolve_detached_python(python_path)
|
||
pythonpath = ";".join([*_launcher_pythonpath_entries(extra_pythonpath), "%PYTHONPATH%"])
|
||
lines = [
|
||
"@echo off",
|
||
f"rem {_TASK_DESCRIPTION}",
|
||
f"cd /d {_quote_cmd_script_arg(working_dir)}",
|
||
f'set "HERMES_HOME={hermes_home}"',
|
||
*[f'set "{k}={v}"' for k, v in _GATEWAY_ENV],
|
||
# VIRTUAL_ENV lets the gateway's own python detection find the venv.
|
||
f'set "VIRTUAL_ENV={_preserve_hermes_home_path(venv_dir)}"',
|
||
f'set "PYTHONPATH={pythonpath}"',
|
||
" ".join(_quote_cmd_script_arg(a) for a in _gateway_run_argv(python_exe_path, profile_arg)),
|
||
"exit /b 0",
|
||
]
|
||
return "\r\n".join(lines) + "\r\n"
|
||
|
||
|
||
def _build_gateway_vbs_script(python_path: str, working_dir: str, hermes_home: str, profile_arg: str) -> str:
|
||
"""Build the hidden-console ``gateway.vbs`` launcher (CRLF-terminated).
|
||
|
||
Run via ``wscript.exe``, not ``cmd.exe``: at logon Windows broadcasts CTRL_CLOSE_EVENT to console
|
||
groups, killing a cmd-hosted gateway with STATUS_CONTROL_C_EXIT, which Task Scheduler treats as a
|
||
user cancel (``RestartOnFailure`` never fires). wscript has no console; python.exe runs with window
|
||
style 0 so descendants inherit one hidden console instead of flashing their own (#54220/#56747).
|
||
|
||
Why: issue #45599 root cause #1.
|
||
``wscript.exe`` is a GUI-subsystem executable with no console, so this launcher receives no console
|
||
control events. It ``Run``s the console ``python.exe`` with window style 0 (hidden): the gateway owns a
|
||
single hidden console — never shown, never CTRL_CLOSE'd at logon, and inherited by every
|
||
console-subsystem descendant (git, gh, node, …) so none of them allocate a visible flashing conhost
|
||
(#54220/#56747; the previous console-less pythonw.exe gateway forced exactly that per-descendant flash).
|
||
No cmd.exe anywhere in the chain. Mirrors ``_build_gateway_cmd_script`` (same env + argv via
|
||
``_resolve_detached_python``).
|
||
"""
|
||
python_exe_path, venv_dir, extra_pythonpath = _resolve_detached_python(python_path)
|
||
# list2cmdline gives CreateProcess-correct quoting for WScript.Shell.Run.
|
||
command_line = subprocess.list2cmdline(_gateway_run_argv(python_exe_path, profile_arg))
|
||
static_pythonpath = os.pathsep.join(_launcher_pythonpath_entries(extra_pythonpath))
|
||
q = _quote_vbs_string
|
||
lines = [
|
||
f"' {_TASK_DESCRIPTION}",
|
||
"Option Explicit",
|
||
"Dim sh, env, existing_pp",
|
||
'Set sh = CreateObject("WScript.Shell")',
|
||
'Set env = sh.Environment("PROCESS")',
|
||
f"env.Item({q('HERMES_HOME')}) = {q(hermes_home)}",
|
||
*[f"env.Item({q(k)}) = {q(v)}" for k, v in _GATEWAY_ENV],
|
||
f"env.Item({q('VIRTUAL_ENV')}) = {q(_preserve_hermes_home_path(venv_dir))}",
|
||
# Mirror the cmd wrapper's ``PYTHONPATH=<static>;%PYTHONPATH%`` at runtime.
|
||
f"existing_pp = env.Item({q('PYTHONPATH')})",
|
||
"If Len(existing_pp) > 0 Then",
|
||
f" env.Item({q('PYTHONPATH')}) = {q(static_pythonpath + os.pathsep)} & existing_pp",
|
||
"Else",
|
||
f" env.Item({q('PYTHONPATH')}) = {q(static_pythonpath)}",
|
||
"End If",
|
||
f"sh.CurrentDirectory = {q(working_dir)}",
|
||
# Window style 0 = hidden; bWaitOnReturn False = detached/async.
|
||
f"sh.Run {q(command_line)}, 0, False",
|
||
]
|
||
return "\r\n".join(lines) + "\r\n"
|
||
|
||
|
||
def _build_startup_launcher(script_path: Path) -> str:
|
||
"""The tiny Startup-folder .vbs that chains hidden. Quits silently if the target is gone so a
|
||
stale entry doesn't error on every login."""
|
||
target = str(script_path.with_suffix(".vbs"))
|
||
command = subprocess.list2cmdline(["wscript.exe", target])
|
||
lines = [
|
||
f"' {_TASK_DESCRIPTION}",
|
||
"Option Explicit",
|
||
"Dim fso, sh, target",
|
||
f"target = {_quote_vbs_string(target)}",
|
||
'Set fso = CreateObject("Scripting.FileSystemObject")',
|
||
"If Not fso.FileExists(target) Then WScript.Quit 0",
|
||
'Set sh = CreateObject("WScript.Shell")',
|
||
f"sh.Run {_quote_vbs_string(command)}, 0, False",
|
||
]
|
||
return "\r\n".join(lines) + "\r\n"
|
||
|
||
|
||
def _write_task_script() -> Path:
|
||
"""Generate the gateway.cmd wrapper (kept as a compatibility artifact) and the console-less .vbs
|
||
launcher used by the Scheduled Task and Startup fallback. Return the .cmd path."""
|
||
_assert_windows()
|
||
settings = _launcher_settings()
|
||
script_path = get_task_script_path()
|
||
_atomic_write(script_path, _build_gateway_cmd_script(*settings), script_path.with_suffix(".tmp"))
|
||
# Also render the console-less .vbs launcher used by Scheduled Task and the Startup-folder fallback via
|
||
# wscript.exe (issue #45599 fix A). The .cmd wrapper stays as a generated helper/compatibility artifact.
|
||
vbs_path = script_path.with_suffix(".vbs")
|
||
_atomic_write(vbs_path, _build_gateway_vbs_script(*settings), vbs_path.with_name(vbs_path.name + ".tmp"))
|
||
return script_path
|
||
|
||
|
||
def _atomic_write(path: Path, content: str, tmp: Path) -> None:
|
||
"""Write ``content`` verbatim (no newline translation) via ``tmp`` then rename over ``path``.
|
||
|
||
The staging file is removed even when the rename fails: the Startup-folder caller stages
|
||
inside the Startup folder itself, and Windows opens every file there at login — a leftover
|
||
``Hermes_Gateway.tmp`` pops up in Notepad after every sign-in (#114093).
|
||
"""
|
||
try:
|
||
tmp.write_text(content, encoding="utf-8", newline="")
|
||
tmp.replace(path)
|
||
finally:
|
||
try:
|
||
tmp.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# ── Install / uninstall
|
||
|
||
def _resolve_task_user() -> str | None:
|
||
"""Return ``DOMAIN\\USER`` if available, else bare USERNAME, else None."""
|
||
username = os.environ.get("USERNAME") or os.environ.get("USER") or os.environ.get("LOGNAME")
|
||
if not username:
|
||
return None
|
||
if "\\" in username:
|
||
return username
|
||
domain = os.environ.get("USERDOMAIN")
|
||
return f"{domain}\\{username}" if domain else username
|
||
|
||
|
||
def _build_scheduled_task_xml(task_name: str, launcher_path: Path, user: str | None) -> str:
|
||
"""Task Scheduler XML with safe long-running defaults. ``launcher_path`` is the console-less
|
||
``.vbs`` run via ``wscript.exe`` (see ``_build_gateway_vbs_script`` for why not cmd.exe).
|
||
|
||
See #45599.
|
||
"""
|
||
user_principal = f"\n <UserId>{escape(user)}</UserId>" if user else ""
|
||
return f"""<?xml version="1.0" encoding="UTF-16"?>
|
||
<Task version="1.4" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
|
||
<RegistrationInfo>
|
||
<Description>{escape(_TASK_DESCRIPTION)}</Description>
|
||
</RegistrationInfo>
|
||
<Triggers>
|
||
<LogonTrigger>
|
||
<Enabled>true</Enabled>
|
||
<Delay>{_TASK_LOGON_DELAY}</Delay>
|
||
</LogonTrigger>
|
||
</Triggers>
|
||
<Principals>
|
||
<Principal id="Author">{user_principal}
|
||
<LogonType>InteractiveToken</LogonType>
|
||
<RunLevel>LeastPrivilege</RunLevel>
|
||
</Principal>
|
||
</Principals>
|
||
<Settings>
|
||
<MultipleInstancesPolicy>IgnoreNew</MultipleInstancesPolicy>
|
||
<DisallowStartIfOnBatteries>false</DisallowStartIfOnBatteries>
|
||
<StopIfGoingOnBatteries>false</StopIfGoingOnBatteries>
|
||
<AllowHardTerminate>true</AllowHardTerminate>
|
||
<StartWhenAvailable>true</StartWhenAvailable>
|
||
<RunOnlyIfNetworkAvailable>false</RunOnlyIfNetworkAvailable>
|
||
<IdleSettings>
|
||
<StopOnIdleEnd>false</StopOnIdleEnd>
|
||
<RestartOnIdle>false</RestartOnIdle>
|
||
</IdleSettings>
|
||
<AllowStartOnDemand>true</AllowStartOnDemand>
|
||
<Enabled>true</Enabled>
|
||
<Hidden>false</Hidden>
|
||
<RunOnlyIfIdle>false</RunOnlyIfIdle>
|
||
<WakeToRun>false</WakeToRun>
|
||
<ExecutionTimeLimit>PT0S</ExecutionTimeLimit>
|
||
<Priority>7</Priority>
|
||
<RestartOnFailure>
|
||
<Interval>{_TASK_RESTART_INTERVAL}</Interval>
|
||
<Count>{_TASK_RESTART_COUNT}</Count>
|
||
</RestartOnFailure>
|
||
</Settings>
|
||
<Actions Context="Author">
|
||
<Exec>
|
||
<Command>wscript.exe</Command>
|
||
<Arguments>//B //Nologo "{escape(str(launcher_path))}"</Arguments>
|
||
</Exec>
|
||
</Actions>
|
||
</Task>
|
||
"""
|
||
|
||
|
||
def _install_scheduled_task(task_name: str, script_path: Path) -> tuple[bool, str]:
|
||
"""Create or replace the Scheduled Task. Returns (success, detail). Always delete+create, never
|
||
``/Change``: it preserves stale repeat/restart settings that relaunch the gateway every minute."""
|
||
delete_code, delete_out, delete_err = _exec_schtasks(["/Delete", "/F", "/TN", task_name])
|
||
delete_detail = (delete_err or delete_out or "").strip()
|
||
if "cannot find" in delete_detail.lower():
|
||
delete_detail = ""
|
||
if delete_code != 0 and delete_detail and _is_access_denied(delete_detail):
|
||
return (False, f"schtasks /Delete failed (code {delete_code}): {delete_detail}")
|
||
# Other /Delete failures are non-fatal: /Create /F may still replace it; keep the detail.
|
||
user = _resolve_task_user()
|
||
launcher_path = script_path.with_suffix(".vbs") # the task launches the console-less .vbs
|
||
xml_path = launcher_path.with_suffix(".task.xml")
|
||
xml_path.write_text(_build_scheduled_task_xml(task_name, launcher_path, user), encoding="utf-16", newline="")
|
||
# Immediate manual starts use _spawn_detached(). See #45599.
|
||
base = ["/Create", "/F", "/TN", task_name, "/XML", str(xml_path)]
|
||
variants = [[*base, "/RU", user, "/NP", "/IT"], base] if user else [base]
|
||
last_code, last_err = 1, ""
|
||
try:
|
||
for argv in variants:
|
||
code, out, err = _exec_schtasks(argv)
|
||
if code == 0:
|
||
return (True, f"Created Scheduled Task {task_name!r}")
|
||
last_code, last_err = code, (err or out or "")
|
||
finally:
|
||
try:
|
||
xml_path.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
if delete_detail:
|
||
last_err = f"{last_err.strip()} (delete detail: {delete_detail})"
|
||
return (False, f"schtasks /Create failed (code {last_code}): {last_err.strip()}")
|
||
|
||
|
||
def _install_startup_entry(script_path: Path) -> Path:
|
||
"""Write the Startup-folder fallback launcher. Returns its path."""
|
||
entry = get_startup_entry_path()
|
||
entry.parent.mkdir(parents=True, exist_ok=True)
|
||
_atomic_write(entry, _build_startup_launcher(script_path), _startup_staging_path())
|
||
legacy_entry = _legacy_startup_entry_path()
|
||
try:
|
||
if legacy_entry.exists():
|
||
legacy_entry.unlink()
|
||
except OSError:
|
||
pass
|
||
return entry
|
||
|
||
|
||
def _remove_startup_entries() -> tuple[list[str], list[str]]:
|
||
"""Unlink the Startup-folder entries (``.vbs`` fallback + legacy ``.cmd``); ``(done, warnings)``.
|
||
|
||
A failure (file locked, access denied) is reported rather than swallowed so callers warn
|
||
instead of claiming a single autostart mechanism.
|
||
"""
|
||
done: list[str] = []
|
||
warnings: list[str] = []
|
||
for path in (get_startup_entry_path(), _legacy_startup_entry_path()):
|
||
try:
|
||
path.unlink()
|
||
done.append(f"Removed redundant Windows login item: {path}")
|
||
except FileNotFoundError:
|
||
pass
|
||
except OSError:
|
||
warnings.append(f"Could not remove redundant Windows login item: {path} (locked or access denied; it still fires at logon)")
|
||
return done, warnings
|
||
|
||
|
||
def redundant_autostart_entries() -> list[Path]:
|
||
"""Startup-folder entries that fire the gateway a second time at logon: every entry beside a
|
||
registered Scheduled Task, or a legacy ``.cmd`` beside the ``.vbs`` fallback."""
|
||
entries = [p for p in (get_startup_entry_path(), _legacy_startup_entry_path()) if p.exists()]
|
||
if is_task_registered():
|
||
return entries
|
||
return entries[1:]
|
||
|
||
|
||
def reconcile_autostart_launchers() -> tuple[list[str], list[str]]:
|
||
"""Converge gateway logon persistence to ONE mechanism; returns ``(done, warnings)`` messages.
|
||
|
||
The Scheduled Task and the Startup-folder entry are alternatives, but a successful task install
|
||
never removed an earlier fallback and pre-#45610 installs left a ``cmd.exe`` launcher behind, so
|
||
logon could fire the launcher twice (#80569). Task registered: remove the Startup entries. No
|
||
task but a legacy ``.cmd``: rewrite it as the console-less ``.vbs`` fallback. File operations
|
||
only (no schtasks mutation, no elevation), so install, update and doctor can all run it.
|
||
"""
|
||
if is_task_registered():
|
||
return _remove_startup_entries()
|
||
legacy = _legacy_startup_entry_path()
|
||
if legacy.exists():
|
||
entry = _install_startup_entry(_write_task_script())
|
||
if legacy.exists(): # _install_startup_entry swallows the unlink failure; both would fire at logon
|
||
return [], [f"Could not remove legacy Windows login item: {legacy} (locked or access denied; it still fires at logon beside {entry})"]
|
||
return [f"Migrated legacy Windows login item to: {entry}"], []
|
||
return [], []
|
||
|
||
|
||
def _resolve_detached_python(python_exe: str) -> tuple[str, Path, list[str]]:
|
||
"""Return (hidden_console_python, venv_dir, extra_pythonpath) for detached runs. ``extra_pythonpath``
|
||
is always empty now; the tuple shape is kept so every call site stays unchanged.
|
||
|
||
Returns the venv's **console** ``python.exe`` — deliberately NOT ``pythonw.exe``. Every detached launch
|
||
path pairs this interpreter with a hidden-console mechanism (``CREATE_NO_WINDOW`` creationflags, or
|
||
``WScript.Shell.Run`` window style 0), so the daemon owns a single hidden console that all of its
|
||
console-subsystem descendants (git, gh, cmd, node, wmic, powershell, …) inherit instead of each
|
||
allocating a visible flashing one. A GUI-subsystem ``pythonw.exe`` daemon has NO console, which is what
|
||
made every descendant spawn flash (#54220/#56747) and forced the endless per-call-site CREATE_NO_WINDOW
|
||
sweep. Root cause isolated + A/B verified on Windows 11 by the desktop backend fix (commit aa2ae36c3f).
|
||
- uv venv launcher: ``venv\\Scripts\\python.exe`` under ``CREATE_NO_WINDOW`` re-execs the base
|
||
interpreter *windowless* — the child inherits the shim's hidden console, so no conhost flashes (the
|
||
#52239 concern). The historical "CREATE_NO_WINDOW cannot suppress the second window" observations were
|
||
made while ``DETACHED_PROCESS`` was in the flag bundle, where MSDN specifies CREATE_NO_WINDOW is IGNORED
|
||
— the hide bit was dead, not ineffective. The base-interpreter + PYTHONPATH-overlay detour is therefore
|
||
unnecessary; the venv shim resolves imports itself. - Console python restores stdout/stderr, so daemon
|
||
logs flow normally.
|
||
Legacy normalization: launchers and argv snapshots from pre-aa2ae36c3f installs lead with
|
||
``pythonw.exe``. When the sibling console ``python.exe`` exists, swap to it so respawns and regenerated
|
||
launchers get the hidden-console design instead of resurrecting the console-less daemon (the
|
||
#54220/#56747 flash class, plus the ``sys.stderr is None`` startup-crash class from #71671).
|
||
"""
|
||
p = Path(python_exe)
|
||
if p.name.lower() in ("pythonw.exe", "pythonw"):
|
||
sibling = p.with_name("python.exe" if p.suffix else "python")
|
||
try:
|
||
if sibling.exists():
|
||
p = sibling
|
||
python_exe = str(sibling)
|
||
except OSError:
|
||
# Can't stat the sibling — keep the original interpreter: a console-less gateway is
|
||
# worse than a hidden-console one, but a failed respawn is worse still.
|
||
pass
|
||
return (python_exe, p.parent.parent, [])
|
||
|
||
|
||
def _prepend_pythonpath(env_overlay: dict[str, str], entries: list[str]) -> None:
|
||
clean_entries = [entry for entry in entries if entry]
|
||
if not clean_entries:
|
||
return
|
||
existing = os.environ.get("PYTHONPATH", "")
|
||
if existing:
|
||
clean_entries.append(existing)
|
||
env_overlay["PYTHONPATH"] = os.pathsep.join(clean_entries)
|
||
|
||
|
||
def _build_gateway_argv(home: Path | None = None) -> tuple[list[str], str, dict[str, str]]:
|
||
"""Build (argv, working_dir, env_overlay) for the gateway subprocess — the same logical command
|
||
as gateway.cmd, assembled as a native argv so no cmd.exe layer sits in between."""
|
||
_assert_windows()
|
||
from hermes_cli.gateway import PROJECT_ROOT
|
||
|
||
python_path, working_dir, hermes_home, profile_arg = _launcher_settings(home)
|
||
python_exe, venv_dir, extra_pythonpath = _resolve_detached_python(python_path)
|
||
env_overlay = {"HERMES_HOME": hermes_home, **dict(_GATEWAY_ENV), "VIRTUAL_ENV": _preserve_hermes_home_path(venv_dir)}
|
||
_prepend_pythonpath(env_overlay, [_preserve_hermes_home_path(p) for p in (PROJECT_ROOT, *extra_pythonpath)])
|
||
return _gateway_run_argv(python_exe, profile_arg), working_dir, env_overlay
|
||
|
||
|
||
def windowless_gateway_restart_spec(run_argv: list[str]) -> tuple[list[str], str, dict[str, str]]:
|
||
"""(argv, cwd, env overlay) for a hidden-console gateway respawn; arguments after the interpreter
|
||
are preserved verbatim. Non-Windows or a non-python argv[0] → argv unchanged, empty overlay.
|
||
|
||
The post-update restart paths build their respawn command from ``get_python_path()`` (the venv's console
|
||
``python.exe``). That is the right interpreter: the watcher launches it with ``CREATE_NO_WINDOW`` detach
|
||
flags, so the respawned gateway owns a single hidden console that all of its descendants inherit —
|
||
nothing flashes (#54220/#56747; the old pythonw.exe rewrite here produced a console-less gateway whose
|
||
every console-subsystem child allocated a visible conhost). This helper now only normalizes the
|
||
interpreter via ``_resolve_detached_python`` and supplies the stable cwd + env overlay (HERMES_HOME,
|
||
VIRTUAL_ENV, PYTHONPATH) so the respawn doesn't depend on the watcher's transient working directory.
|
||
"""
|
||
if not run_argv or sys.platform != "win32":
|
||
return run_argv, "", {}
|
||
from hermes_cli.gateway import PROJECT_ROOT
|
||
|
||
try:
|
||
hidden_console_python, venv_dir, extra_pythonpath = _resolve_detached_python(run_argv[0])
|
||
except Exception:
|
||
return run_argv, "", {}
|
||
|
||
try:
|
||
hermes_home = str(_hermes_home().resolve())
|
||
except Exception:
|
||
hermes_home = ""
|
||
env_overlay: dict[str, str] = {"PYTHONIOENCODING": "utf-8", "HERMES_GATEWAY_DETACHED": "1", "VIRTUAL_ENV": str(venv_dir)}
|
||
if hermes_home:
|
||
env_overlay["HERMES_HOME"] = hermes_home
|
||
_prepend_pythonpath(env_overlay, [str(PROJECT_ROOT), *extra_pythonpath])
|
||
return [hidden_console_python, *run_argv[1:]], _stable_gateway_working_dir(PROJECT_ROOT), env_overlay
|
||
|
||
|
||
def _spawn_detached(script_path: Path | None = None, home: Path | None = None) -> int:
|
||
"""Launch the gateway as a fully detached background process (``script_path`` is ignored; kept
|
||
for API symmetry; ``home`` spawns another profile's gateway — ``--profile`` is derived from it). Spawns python.exe directly — a cmd.exe shim inherits the parent console and
|
||
gets reaped when the shell exits. Flags: CREATE_NEW_PROCESS_GROUP (no Ctrl+C from our group),
|
||
CREATE_NO_WINDOW (hidden console descendants inherit, so nothing flashes — #54220/#56747; the old
|
||
DETACHED_PROCESS made every descendant spawn flash), CREATE_BREAKAWAY_FROM_JOB
|
||
(escape a parent Job Object — some Windows Terminal versions wrap children in one).
|
||
|
||
With ``CREATE_NO_WINDOW`` the gateway gets its OWN hidden console instead of inheriting ours, so it
|
||
survives our shell closing, and every console-subsystem descendant it spawns inherits that hidden
|
||
console instead of flashing a visible one (#54220/#56747 — this is why we don't use console-less
|
||
pythonw.exe here). Combined with CREATE_NEW_PROCESS_GROUP + DEVNULL stdin + a fresh env, the resulting
|
||
process is independent of whichever shell started it.
|
||
"""
|
||
_assert_windows()
|
||
argv, working_dir, env_overlay = _build_gateway_argv(home)
|
||
from tools.environments.local import served_profile_child_env
|
||
# home=None is this process's own gateway, not a forced jump to the default root.
|
||
# served_profile_child_env overlays that home's secrets instead of os.environ.copy().
|
||
target = home if home is not None else _hermes_home()
|
||
env = {**served_profile_child_env(target_home=target, inherit_credentials=True), **env_overlay}
|
||
|
||
# Stray print()/native stderr goes to a sidecar log; real gateway logs still land in gateway.log
|
||
# via the logging FileHandler.
|
||
log_dir = _hermes_home() / "logs"
|
||
log_dir.mkdir(parents=True, exist_ok=True)
|
||
stray_log = log_dir / "gateway-stdio.log"
|
||
|
||
def _popen(breakaway: str, flags: int):
|
||
with open(stray_log, "ab", buffering=0) as log_fh:
|
||
return subprocess.Popen(
|
||
argv, cwd=working_dir, env={**env, _WINDOWS_GATEWAY_BREAKAWAY_ENV: breakaway}, creationflags=flags,
|
||
close_fds=True, stdin=subprocess.DEVNULL, stdout=log_fh, stderr=log_fh,
|
||
)
|
||
|
||
try:
|
||
proc = _popen("1", windows_detach_flags())
|
||
_LAST_SPAWN_BREAKAWAY_FALLBACK["fallback"] = False
|
||
except OSError as exc:
|
||
# CREATE_BREAKAWAY_FROM_JOB fails with "access denied" when the parent's job object forbids
|
||
# breakaway (some Windows Terminal configs). Retry without it — the hidden-console
|
||
# CREATE_NO_WINDOW spawn is usually enough on its own.
|
||
error_code = getattr(exc, "winerror", None)
|
||
if error_code is None:
|
||
error_code = exc.errno
|
||
logger.warning("Gateway breakaway spawn failed (error=%s); retrying without CREATE_BREAKAWAY_FROM_JOB", error_code)
|
||
proc = _popen("0", windows_detach_flags_without_breakaway())
|
||
_LAST_SPAWN_BREAKAWAY_FALLBACK["fallback"] = True
|
||
return proc.pid
|
||
|
||
|
||
def _stdin_is_interactive(*, isatty: bool, console_mode_ok: bool | None) -> bool:
|
||
"""A human can answer a prompt only on a real console. The Windows CRT reports isatty()==True for
|
||
every character device — the NUL device included (`hermes gateway start < NUL`, stdin=DEVNULL) — so
|
||
isatty must be confirmed by GetConsoleMode accepting the handle (#113977). ``console_mode_ok`` is
|
||
None where that fact does not exist (not Windows) and isatty alone decides."""
|
||
return isatty and console_mode_ok is not False
|
||
|
||
|
||
def _stdout_isatty() -> bool:
|
||
"""The question is printed to stdout. When stdout is captured, nobody sees it. Desktop update
|
||
hand-offs before #122234 captured each step's stdout while leaving it the console's stdin, so a
|
||
prompt there waited forever for an answer to a question nobody saw."""
|
||
return sys.stdout is not None and sys.stdout.isatty()
|
||
|
||
|
||
def _stdin_console_mode_ok() -> bool | None:
|
||
if sys.platform != "win32":
|
||
return None
|
||
kernel32 = ctypes.windll.kernel32
|
||
handle = kernel32.GetStdHandle(-10) # STD_INPUT_HANDLE
|
||
return bool(kernel32.GetConsoleMode(handle, ctypes.byref(ctypes.c_ulong())))
|
||
|
||
|
||
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."""
|
||
if start_now is None:
|
||
start_now = _install_choice_from_env("HERMES_GATEWAY_INSTALL_START_NOW")
|
||
if start_on_login is None:
|
||
start_on_login = _install_choice_from_env("HERMES_GATEWAY_INSTALL_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 _report_already_running(running_pids: list[int]) -> None:
|
||
print(f"✓ Gateway already running (PID: {', '.join(map(str, running_pids))})")
|
||
|
||
|
||
def _start_or_report_running(running_pids: list[int] | None = None) -> None:
|
||
"""Spawn the gateway unless one is already running for this profile."""
|
||
if running_pids is None:
|
||
running_pids = _gateway_pids()
|
||
if running_pids:
|
||
_report_already_running(running_pids)
|
||
else:
|
||
pid = _spawn_detached()
|
||
_report_gateway_start("direct spawn")
|
||
|
||
|
||
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")
|
||
if is_task_registered():
|
||
# An earlier task survives (UAC declined, access denied on re-create) and still fires at
|
||
# logon; adding the fallback beside it would start the gateway twice (#80569).
|
||
print("⚠ Scheduled Task is still registered — skipped the Startup fallback to avoid a duplicate autostart.")
|
||
print(" If that task is disabled or broken, run 'hermes gateway uninstall', then install again.")
|
||
else:
|
||
entry = _install_startup_entry(script_path)
|
||
print(f"✓ Installed Windows login item: {entry}")
|
||
print(f" Task script: {script_path}")
|
||
|
||
# Re-running install must be safe: the fallback only installs login persistence; starting is
|
||
# controlled by the pre-UAC start_now answer so every user decision precedes elevation.
|
||
running_pids = _gateway_pids()
|
||
if running_pids or start_now:
|
||
_start_or_report_running(running_pids)
|
||
else:
|
||
from hermes_cli.gateway import _profile_arg
|
||
|
||
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 _offer_elevated_install(headline: str, force: bool, start_now: bool, start_on_login: bool) -> bool:
|
||
"""Offer the UAC prompt for a Scheduled Task install. True when handed off to an elevated child."""
|
||
from hermes_cli.setup import prompt_yes_no
|
||
|
||
print(headline)
|
||
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 True
|
||
print("⚠ Falling back to Startup folder because elevation was unavailable or cancelled.")
|
||
else:
|
||
print(" Skipped elevation. Falling back to Startup folder.")
|
||
return False
|
||
|
||
|
||
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 — we
|
||
always reconcile; ``force`` exists for API parity with launchd/systemd."""
|
||
_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:
|
||
_start_or_report_running()
|
||
else:
|
||
print("ℹ Gateway not started and no auto-start service installed.")
|
||
print(" Run in the foreground later with: hermes gateway run")
|
||
return
|
||
|
||
task_name = get_task_name()
|
||
script_path = _write_task_script()
|
||
# A pre-fix install that failed its Startup-folder swap left `Hermes_Gateway.tmp` there, and the
|
||
# Scheduled Task path below never touches that folder — sweep it so a re-run clears the debris.
|
||
try:
|
||
_startup_staging_path().unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
if force:
|
||
# Pre-suffix strays (task ``Hermes_Gateway``, Startup ``Hermes_Gateway.vbs``) are unreachable by
|
||
# the current names, so a plain reconcile never heals them (#116157).
|
||
from hermes_cli.gateway_windows_legacy import remove_legacy_launchers
|
||
remove_legacy_launchers()
|
||
|
||
# On locked-down accounts schtasks can sit for the full timeout before returning Access Denied.
|
||
# All intent questions were asked above, so ask for UAC before touching schtasks.
|
||
if not _is_running_as_admin() and not elevated_handoff:
|
||
if _offer_elevated_install(
|
||
"↻ Scheduled Task install may need administrator approval on this Windows account.",
|
||
force, start_now, start_on_login,
|
||
):
|
||
return
|
||
_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}")
|
||
print("ℹ Gateway auto-start installed for Windows login.")
|
||
# A Startup-folder entry from an earlier fallback install would fire alongside the task (#80569).
|
||
done, warnings = _remove_startup_entries()
|
||
for message in done:
|
||
print(f"✓ {message}")
|
||
for message in warnings:
|
||
print(f"⚠ {message}")
|
||
if start_now:
|
||
_start_or_report_running()
|
||
else:
|
||
print("ℹ Gateway not started now.")
|
||
print(" Start manually with: hermes gateway start")
|
||
_print_next_steps()
|
||
return
|
||
|
||
# Prefer a real Scheduled Task over the Startup fallback when elevation is the only blocker.
|
||
if _is_access_denied(detail) and not _is_running_as_admin() and _offer_elevated_install(
|
||
f"↻ Scheduled Task install needs administrator approval ({detail.splitlines()[0]})",
|
||
force, start_now, start_on_login,
|
||
):
|
||
return
|
||
|
||
if _should_fall_back(1, detail):
|
||
_install_startup_fallback(script_path, start_now, detail)
|
||
return
|
||
|
||
raise RuntimeError(f"Windows gateway install failed: {detail}")
|
||
|
||
|
||
def _live_gateway_pids(
|
||
all_profiles: bool = False, home: Path | None = None, pid_filter=None
|
||
) -> list[int]:
|
||
"""Live gateway PIDs for the readiness poll. ``home`` scopes the probe to ONE profile's identity
|
||
files (a still-running sibling must not vouch for a per-profile spawn, #110959); otherwise the
|
||
process-table discovery for the active profile or the whole fleet."""
|
||
if home is not None:
|
||
from gateway.status import get_running_pid
|
||
|
||
pid = get_running_pid(home / "gateway.pid", cleanup_stale=False)
|
||
pids = [pid] if pid else []
|
||
else:
|
||
from hermes_cli.gateway import find_gateway_pids
|
||
pids = list(find_gateway_pids(all_profiles=all_profiles))
|
||
return list(pid_filter(pids)) if pid_filter is not None else pids
|
||
|
||
|
||
def _confirm_gateway_stable(
|
||
initial_pids: list[int], confirm_s: float, interval_s: float, all_profiles: bool = False,
|
||
home: Path | None = None, pid_filter=None,
|
||
) -> list[int]:
|
||
"""Re-check a freshly detected gateway for ``confirm_s`` seconds: one process-table hit proves
|
||
the child was *created*, not that it survived startup (or a parent Job Object teardown).
|
||
|
||
A single process-table hit only proves the child was *created*, not that it survived startup — a gateway
|
||
that crashes moments after spawn (or is reaped by the parent shell's Job Object, #91675/#84185) passes a
|
||
first-hit poll and then dies. Require the gateway to stay visible for the whole confirmation window
|
||
before we vouch for it. Returns the last observed PID list, or ``[]`` if the gateway vanished
|
||
mid-window.
|
||
"""
|
||
if confirm_s <= 0:
|
||
return initial_pids
|
||
pids = initial_pids
|
||
confirm_deadline = time.monotonic() + confirm_s
|
||
while time.monotonic() < confirm_deadline:
|
||
time.sleep(interval_s)
|
||
pids = _live_gateway_pids(all_profiles, home, pid_filter)
|
||
if not pids:
|
||
return []
|
||
return pids
|
||
|
||
|
||
def _wait_for_gateway_ready(
|
||
timeout_s: float = 6.0, interval_s: float = 0.4, confirm_s: float = 2.0, all_profiles: bool = False,
|
||
home: Path | None = None, pid_filter=None,
|
||
) -> list[int]:
|
||
"""Poll for a live gateway for up to ``timeout_s``; a first hit is provisional until the gateway
|
||
stays visible for ``confirm_s`` more seconds (a child that dies right after spawn earns no ✓)."""
|
||
deadline = time.monotonic() + timeout_s
|
||
while time.monotonic() < deadline:
|
||
pids = _live_gateway_pids(all_profiles, home, pid_filter)
|
||
if pids:
|
||
confirmed = _confirm_gateway_stable(
|
||
pids, confirm_s, interval_s, all_profiles=all_profiles, home=home, pid_filter=pid_filter
|
||
)
|
||
if confirmed:
|
||
return confirmed
|
||
continue # died during confirmation — keep polling until deadline
|
||
time.sleep(interval_s)
|
||
return []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Start attestation — honest reporting for deaths AFTER the liveness poll
|
||
#
|
||
# The poll cannot observe a death after this CLI process exits (the parent shell's Job Object tears
|
||
# the gateway down on CLI exit). So every ✓ persists a marker recording which PIDs we vouched for;
|
||
# the NEXT gateway CLI invocation checks it: PIDs gone with no clean exit in the lifecycle ledger
|
||
# means the earlier ✓ was a lie, and we say so — once — with the schtasks recovery hint.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_START_ATTESTATION_RELATIVE = ("state", "gateway.start-attestation.json")
|
||
|
||
|
||
def _start_attestation_path(home: Path | None = None) -> Path:
|
||
"""Marker path; ``home`` addresses another profile's marker (per-profile cold-start, #110959)."""
|
||
return (home if home is not None else _hermes_home()).joinpath(*_START_ATTESTATION_RELATIVE)
|
||
|
||
|
||
def _write_start_attestation(pids: list[int], via: str, home: Path | None = None) -> None:
|
||
"""Persist the PIDs a ✓ vouched for. Best-effort, never raises.
|
||
|
||
``generation`` identifies this marker instance: the update resume token records the generation
|
||
whose death authorized a cold-start, so execution consumes exactly that marker and never a
|
||
newer one written by a concurrent ``hermes gateway start`` (#110020 review)."""
|
||
try:
|
||
path = _start_attestation_path(home)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
from hermes_cli.process_identity import _process_create_time
|
||
|
||
payload = {
|
||
"pids": [int(p) for p in pids], "via": via, "ts": datetime.now(timezone.utc).isoformat(),
|
||
"generation": uuid.uuid4().hex,
|
||
}
|
||
# Bind each PID to its incarnation (#110020 review): the ledger sentinel is matched by PID
|
||
# only otherwise, so a stale marker would be re-read against whatever lifecycle wrote last.
|
||
create_times = {str(int(p)): _process_create_time(int(p)) for p in pids}
|
||
payload["create_times"] = {k: v for k, v in create_times.items() if v is not None}
|
||
tmp = path.with_suffix(".json.tmp")
|
||
tmp.write_text(json.dumps(payload), encoding="utf-8")
|
||
tmp.replace(path)
|
||
except Exception:
|
||
logger.debug("Failed to write gateway start attestation", exc_info=True)
|
||
|
||
|
||
def _clear_start_attestation(home: Path | None = None) -> None:
|
||
try:
|
||
_start_attestation_path(home).unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# A start attestation older than this is no authority (#110020 review (d)): the marker is a one-shot
|
||
# meant to bridge the seconds between a ✓ and the next ``hermes gateway status``/``update``; a
|
||
# historical marker must never later override Desktop ownership into a duplicate gateway (#76129).
|
||
START_ATTESTATION_MAX_AGE_S = 24 * 3600
|
||
# Same slack process_identity uses for psutil create_time comparisons (PID reuse disambiguation).
|
||
_CREATE_TIME_TOLERANCE_S = 2.0
|
||
# A backwards clock step (NTP) between write and read must not kill a fresh marker.
|
||
_ATTESTATION_CLOCK_SLACK_S = 60.0
|
||
|
||
|
||
def _attestation_within_horizon(data: object) -> bool:
|
||
"""False for a marker whose ``ts`` is missing, unparsable or older than the horizon (fail closed)."""
|
||
try:
|
||
ts = datetime.fromisoformat(str(data["ts"])) if isinstance(data, dict) else None
|
||
if ts is None:
|
||
return False
|
||
if ts.tzinfo is None:
|
||
ts = ts.replace(tzinfo=timezone.utc)
|
||
age = time.time() - ts.timestamp()
|
||
return -_ATTESTATION_CLOCK_SLACK_S <= age <= START_ATTESTATION_MAX_AGE_S
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _attestation_generation(data: object) -> str | None:
|
||
"""The marker instance identity, or ``None`` for a marker that carries none."""
|
||
return str(data["generation"]) if isinstance(data, dict) and data.get("generation") else None
|
||
|
||
|
||
def _consume_start_attestation(generation: str, home: Path | None = None) -> None:
|
||
"""Clear the marker only while it is still the ``generation`` that was acted on; a newer
|
||
marker belongs to a gateway start this caller knows nothing about and keeps its own report."""
|
||
# Best-effort read-then-unlink: a marker written in between loses one post-start report, never authority.
|
||
if _attestation_generation(_read_start_attestation(home)) == generation:
|
||
_clear_start_attestation(home)
|
||
|
||
|
||
def _read_start_attestation(home: Path | None = None) -> object | None:
|
||
"""Parsed attestation payload (any JSON type), or ``None`` when absent/unreadable. Never raises."""
|
||
try:
|
||
return json.loads(_start_attestation_path(home).read_text(encoding="utf-8-sig"))
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
|
||
def _attested_pids_from(data: object) -> list[int]:
|
||
"""PID list from an attestation payload; empty for anything malformed."""
|
||
if not isinstance(data, dict):
|
||
return []
|
||
pids = data.get("pids")
|
||
# Fail closed: a null/malformed marker must never authorize a cold start (or raise on iteration).
|
||
# Exact positive ints only — ``isinstance(True, int)`` holds, and 0 / negatives are not PIDs; one
|
||
# bad item taints the whole list because the writer never emits such values.
|
||
if not isinstance(pids, list) or not all(type(p) is int and p > 0 for p in pids):
|
||
return []
|
||
return list(pids)
|
||
|
||
|
||
def _attested_create_time(data: object, pid: int) -> float | None:
|
||
"""The process create time the marker bound ``pid`` to, or ``None`` (older marker / psutil silent)."""
|
||
times = data.get("create_times") if isinstance(data, dict) else None
|
||
value = times.get(str(pid)) if isinstance(times, dict) else None
|
||
return float(value) if type(value) in (int, float) else None
|
||
|
||
|
||
def _attested_pid_exited_cleanly(pid: int, create_time: float | None = None, home: Path | None = None) -> bool:
|
||
"""True when the lifecycle ledger shows a clean exit for ``pid`` — or, for a marker that bound
|
||
``pid`` to a ``create_time``, whenever the sentinel cannot be shown to describe THAT incarnation
|
||
(#110020 review): a sentinel for another PID or another start time means an unrelated lifecycle
|
||
has run since and the marker is stale; "unknown" must never read as "dead". A missing sentinel
|
||
still reads as dead (the attested process never booted far enough to claim it)."""
|
||
try:
|
||
from gateway.lifecycle_ledger import get_lifecycle_sentinel_path
|
||
|
||
sentinel = get_lifecycle_sentinel_path(home if home is not None else _hermes_home())
|
||
data = json.loads(sentinel.read_text(encoding="utf-8-sig"))
|
||
except OSError:
|
||
return False
|
||
except Exception:
|
||
return create_time is not None
|
||
if not isinstance(data, dict):
|
||
return create_time is not None
|
||
if create_time is not None:
|
||
if data.get("pid") != pid:
|
||
return True
|
||
sentinel_birth = data.get("create_time")
|
||
# A sentinel from a gateway older than the identity stamp cannot be told apart: PID-only rule.
|
||
if type(sentinel_birth) in (int, float) and abs(float(sentinel_birth) - create_time) > _CREATE_TIME_TOLERANCE_S:
|
||
return True
|
||
return data.get("phase") == "exited" and data.get("pid") == pid
|
||
|
||
|
||
def _attested_dead(
|
||
attested: list[int], current_pids: list[int], data: object = None, home: Path | None = None
|
||
) -> bool:
|
||
"""The liveness rule shared by the consuming and read-only probes: attested PIDs are dead when
|
||
no gateway runs now and the lifecycle ledger shows no clean exit for any of them."""
|
||
return not current_pids and not any(
|
||
_attested_pid_exited_cleanly(pid, _attested_create_time(data, pid), home) for pid in attested
|
||
)
|
||
|
||
|
||
def attested_death_generation(current_pids: list[int], home: Path | None = None) -> str | None:
|
||
"""The generation of a start attestation that vouches for gateway PID(s) gone without a clean exit,
|
||
or ``None``.
|
||
|
||
Read-only twin of :func:`check_start_attestation` for callers that must not consume the
|
||
one-shot marker — ``hermes update`` consults it to decide whether a Desktop-owned install
|
||
still owes a gateway cold-start (#109538) and records the generation in its resume token so the
|
||
execution step consumes exactly the marker it was authorized by. Callers pass the liveness they
|
||
already established (``[]`` after their own discovery came back empty) so the process table is
|
||
not scanned twice. ``None`` for anything undecidable (no marker, no generation, a clean ledger
|
||
exit): "unknown" must never read as "dead". ``home`` probes another profile's marker and ledger
|
||
(the updater evaluates every profile that is not running, #110959)."""
|
||
data = _read_start_attestation(home)
|
||
attested = _attested_pids_from(data)
|
||
if not attested or not _attestation_within_horizon(data) or not _attested_dead(attested, current_pids, data, home):
|
||
return None
|
||
return _attestation_generation(data)
|
||
|
||
|
||
def check_start_attestation(current_pids: list[int] | None = None) -> str | None:
|
||
"""Surface (once) a gateway that died after a ✓ was printed for it. Never raises. Gateway running
|
||
or a clean-exit ledger record: clear silently; otherwise return a warning and consume the marker."""
|
||
data = _read_start_attestation()
|
||
if data is None:
|
||
return None
|
||
attested = _attested_pids_from(data)
|
||
if not attested:
|
||
_clear_start_attestation()
|
||
return None
|
||
|
||
if current_pids is None:
|
||
try:
|
||
from hermes_cli.gateway import find_gateway_pids
|
||
|
||
current_pids = list(find_gateway_pids())
|
||
except Exception:
|
||
return None
|
||
|
||
_clear_start_attestation()
|
||
if not _attested_dead(attested, current_pids, data):
|
||
return None
|
||
return _format_attestation_warning(attested, data)
|
||
|
||
|
||
def _format_attestation_warning(attested: list[int], data: dict) -> str:
|
||
via = data.get("via") or "direct spawn"
|
||
ts = data.get("ts") or "unknown time"
|
||
lines = [
|
||
f"⚠ The previous gateway start ({via}, {ts}) reported success, but the "
|
||
f"process (PID {', '.join(map(str, attested))}) died without a clean "
|
||
"shutdown record.",
|
||
" This usually means the shell that ran `hermes gateway start` was inside "
|
||
"a Windows Job Object that killed the gateway on exit (#91675).",
|
||
]
|
||
hint = _task_run_hint(" Recovery: schtasks /Run /TN {} (Task Scheduler starts the gateway outside any Job Object)")
|
||
if hint:
|
||
lines.append(hint)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _task_run_hint(fmt: str) -> str | None:
|
||
"""``fmt`` with the task name filled in, when a Scheduled Task is registered. Never raises."""
|
||
try:
|
||
if is_task_registered():
|
||
return fmt.format(get_task_name())
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _print_task_run_hint(fmt: str) -> None:
|
||
hint = _task_run_hint(fmt)
|
||
if hint:
|
||
print(hint)
|
||
|
||
|
||
def _print_start_attestation_warning() -> None:
|
||
"""Print the stale-attestation warning if one is pending. Never raises."""
|
||
try:
|
||
warning = check_start_attestation()
|
||
except Exception:
|
||
return
|
||
if warning:
|
||
print(warning)
|
||
|
||
|
||
def _report_gateway_start(via: str) -> None:
|
||
pids = _wait_for_gateway_ready()
|
||
if pids:
|
||
print(f"✓ Gateway started via {via} (PID: {', '.join(map(str, pids))})")
|
||
if _LAST_SPAWN_BREAKAWAY_FALLBACK.get("fallback"):
|
||
print("⚠ The gateway could not break away from this shell's Job Object; it may be killed when this shell exits.")
|
||
_print_task_run_hint(" If it dies, start it with: schtasks /Run /TN {}")
|
||
_write_start_attestation(pids, via)
|
||
else:
|
||
print(f"✗ Gateway start via {via} FAILED — no stable gateway process detected within the verification window.")
|
||
print(" (The process may have been created and then killed — e.g. by a parent Job Object, #91675.)")
|
||
print(f" Check the log for startup errors:\n type {_hermes_home()}\\logs\\gateway.log\n type {_hermes_home()}\\logs\\gateway-stdio.log")
|
||
_print_task_run_hint(" Recovery: schtasks /Run /TN {} (starts the gateway outside any Job Object)")
|
||
|
||
|
||
def _print_next_steps() -> None:
|
||
print("\nNext steps:\n hermes gateway status # Check status")
|
||
print(f" type {_hermes_home()}\\logs\\gateway.log # View logs")
|
||
|
||
|
||
def uninstall() -> None:
|
||
"""Remove both the Scheduled Task and the Startup-folder fallback, if present."""
|
||
_assert_windows()
|
||
task_name = get_task_name()
|
||
script_path = get_task_script_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_gateway_command("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}: {detail}")
|
||
|
||
for path, label in (
|
||
(get_startup_entry_path(), "Windows login item"), (_legacy_startup_entry_path(), "legacy Windows login item"),
|
||
(_startup_staging_path(), "Windows login item staging file"),
|
||
(script_path, "Task script"), (script_path.with_suffix(".vbs"), "Task launcher"),
|
||
):
|
||
try:
|
||
path.unlink()
|
||
print(f"✓ Removed {label}: {path}")
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
from hermes_cli.gateway_windows_legacy import remove_legacy_launchers
|
||
remove_legacy_launchers()
|
||
|
||
if is_task_registered() and not scheduled_task_removed:
|
||
print(f"⚠ Scheduled Task still registered: {task_name}")
|
||
|
||
|
||
# ── Status / start / stop / restart
|
||
|
||
def is_task_registered() -> bool:
|
||
code, _out, _err = _exec_schtasks(["/Query", "/TN", get_task_name()])
|
||
return code == 0
|
||
|
||
|
||
def is_startup_entry_installed() -> bool:
|
||
return get_startup_entry_path().exists() or _legacy_startup_entry_path().exists()
|
||
|
||
|
||
def _query_scheduled_task_xml(task_name: str) -> str | None:
|
||
"""Return a registered task's XML, or ``None`` when it cannot be inspected (fail open: a
|
||
localized ``schtasks`` failure is not evidence about an otherwise working task)."""
|
||
code, out, err = _exec_schtasks(["/Query", "/TN", task_name, "/XML"])
|
||
if code != 0 or not out.strip():
|
||
logger.debug("Could not query Scheduled Task XML for %r: %s", task_name, (err or out).strip())
|
||
return None
|
||
return out
|
||
|
||
|
||
def _task_xml_leaf_values(xml: str) -> dict[str, str] | None:
|
||
"""Namespace-agnostic ``Task/Settings/...`` leaf-path → text map, or ``None`` for invalid XML.
|
||
The root ``version`` attribute is exposed as ``Task@version``."""
|
||
try:
|
||
root = ElementTree.fromstring(xml)
|
||
except ElementTree.ParseError:
|
||
return None
|
||
values: dict[str, str] = {"Task@version": root.attrib.get("version", "")}
|
||
|
||
def visit(element: ElementTree.Element, path: tuple[str, ...]) -> None:
|
||
current_path = (*path, element.tag.rsplit("}", 1)[-1])
|
||
children = list(element)
|
||
if not children:
|
||
values["/".join(current_path)] = " ".join((element.text or "").split())
|
||
for child in children:
|
||
visit(child, current_path)
|
||
|
||
visit(root, ())
|
||
return values
|
||
|
||
|
||
# Allowlist of template leaves whose absence/mismatch on the live task means it predates the current
|
||
# template (#113670). Never a full-leaf compare: schtasks exports <UserId> as a SID while the template
|
||
# writes DOMAIN\user, so equality would flag every healthy registration.
|
||
_TASK_DRIFT_LEAVES = {
|
||
"Task/Settings/RestartOnFailure/Interval": "RestartOnFailure",
|
||
"Task/Triggers/LogonTrigger/Delay": "LogonTrigger Delay",
|
||
"Task/Actions/Exec/Arguments": "launcher arguments",
|
||
}
|
||
|
||
|
||
def compare_scheduled_task_drift(registered_xml: str, template_xml: str) -> list[str]:
|
||
"""Human-readable drift fragments between a registered task export and the current template,
|
||
over ``_TASK_DRIFT_LEAVES`` plus the Task ``version``. Empty when aligned or when either side
|
||
does not parse (fail open)."""
|
||
live = _task_xml_leaf_values(registered_xml)
|
||
want = _task_xml_leaf_values(template_xml)
|
||
if live is None or want is None:
|
||
return []
|
||
missing = [label for path, label in _TASK_DRIFT_LEAVES.items() if path in want and path not in live]
|
||
differs = [label for path, label in _TASK_DRIFT_LEAVES.items() if path in want and path in live and live[path] != want[path]]
|
||
drift = []
|
||
if missing:
|
||
drift.append(f"missing: {', '.join(missing)}")
|
||
drift.extend(f"{label} differs" for label in differs)
|
||
if live["Task@version"] != want["Task@version"]:
|
||
drift.append(f"version {live['Task@version']} vs {want['Task@version']}")
|
||
return drift
|
||
|
||
|
||
def scheduled_task_drift(task_name: str) -> list[str]:
|
||
"""Drift fragments between the registered task and ``_build_scheduled_task_xml``; empty when
|
||
aligned or when the task cannot be queried."""
|
||
registered = _query_scheduled_task_xml(task_name)
|
||
if registered is None:
|
||
return []
|
||
template = _build_scheduled_task_xml(task_name, get_task_script_path().with_suffix(".vbs"), _resolve_task_user())
|
||
return compare_scheduled_task_drift(registered, template)
|
||
|
||
|
||
def _print_scheduled_task_drift(task_name: str) -> None:
|
||
"""Warn when the registered task predates the current template (status is read-only; the
|
||
repair runs from ``start()`` / ``hermes update`` via ``reconcile_scheduled_task``)."""
|
||
drift = scheduled_task_drift(task_name)
|
||
if drift:
|
||
print(f"⚠ Scheduled Task registration predates the current template ({'; '.join(drift)})")
|
||
print(" Repair: hermes gateway start (or: hermes gateway install)")
|
||
|
||
|
||
def reconcile_scheduled_task(task_name: str) -> bool:
|
||
"""Re-register the task from the current template when it drifts (#113670) — the Windows sibling
|
||
of ``gateway.py::refresh_systemd_unit_if_needed``. Template hardening (``RestartOnFailure``, logon
|
||
``Delay``) otherwise only ever reaches fresh installs. False when aligned/unqueryable or when
|
||
``schtasks`` refused (typically Access Denied — the elevating ``hermes gateway install`` is the fallback)."""
|
||
drift = scheduled_task_drift(task_name)
|
||
if not drift:
|
||
return False
|
||
print(f"↻ Repairing outdated Scheduled Task registration ({'; '.join(drift)})")
|
||
ok, detail = _install_scheduled_task(task_name, _write_task_script())
|
||
print(f"{'✓' if ok else '⚠'} {detail}")
|
||
if not ok:
|
||
print(" Repair manually: hermes gateway install")
|
||
return ok
|
||
|
||
|
||
def is_installed() -> bool:
|
||
"""True when either the schtasks entry or the Startup fallback is present."""
|
||
return is_task_registered() or is_startup_entry_installed()
|
||
|
||
|
||
def query_task_status() -> dict[str, str]:
|
||
"""Parse ``schtasks /Query /V /FO LIST`` and pull the interesting keys."""
|
||
code, out, err = _exec_schtasks(["/Query", "/TN", get_task_name(), "/V", "/FO", "LIST"])
|
||
if code != 0:
|
||
return {}
|
||
info: dict[str, str] = {}
|
||
for raw in out.splitlines():
|
||
line = raw.strip()
|
||
if not line or ":" not in line:
|
||
continue
|
||
key, _, value = line.partition(":")
|
||
key = key.strip().lower()
|
||
value = value.strip()
|
||
# Some Windows locales emit "Last Result" instead of "Last Run Result".
|
||
if key == "last result":
|
||
info.setdefault("last run result", value)
|
||
elif key in {"status", "last run time", "last run result"}:
|
||
info[key] = value
|
||
return info
|
||
|
||
|
||
def _gateway_pids() -> list[int]:
|
||
"""Reuse the cross-platform PID scanner in gateway.py."""
|
||
from hermes_cli.gateway import find_gateway_pids
|
||
|
||
return list(find_gateway_pids())
|
||
|
||
|
||
def _probe(index: int, ok: bool, message: str) -> None:
|
||
print(f" [{index}] {'PASS' if ok else 'FAIL':4s} {message}")
|
||
|
||
|
||
def _probe_missing(index: int, path: Path, label: str) -> bool:
|
||
if path.exists():
|
||
return False
|
||
_probe(index, False, f"{label} missing: {path}")
|
||
return True
|
||
|
||
|
||
def _probe_pid_file(pid_path: Path) -> int | None:
|
||
if _probe_missing(1, pid_path, "PID file"):
|
||
return None
|
||
try:
|
||
data = json.loads(pid_path.read_text(encoding="utf-8-sig"))
|
||
pid_value = int(data.get("pid")) if data.get("pid") is not None else None
|
||
_probe(1, True, f"PID file present: {pid_path} (pid={pid_value})")
|
||
return pid_value
|
||
except Exception as exc:
|
||
_probe(1, False, f"PID file present but unreadable: {exc}")
|
||
return None
|
||
|
||
|
||
def _probe_lock_file(lock_path: Path) -> None:
|
||
if _probe_missing(2, lock_path, "Lock file"):
|
||
return
|
||
try:
|
||
from gateway.status import is_gateway_runtime_lock_active
|
||
|
||
_probe(2, is_gateway_runtime_lock_active(lock_path), f"Lock file held by a live process: {lock_path}")
|
||
except Exception as exc:
|
||
_probe(2, False, f"Could not probe lock: {exc}")
|
||
|
||
|
||
def _probe_running_pid() -> int | None:
|
||
try:
|
||
from gateway.status import get_running_pid
|
||
|
||
running_pid = get_running_pid(cleanup_stale=False)
|
||
_probe(3, running_pid is not None, f"get_running_pid() => {running_pid}")
|
||
return running_pid
|
||
except Exception as exc:
|
||
_probe(3, False, f"get_running_pid() raised: {exc!r}")
|
||
return None
|
||
|
||
|
||
def _probe_pid_exists(candidate_pid: int | None) -> None:
|
||
if candidate_pid is None:
|
||
_probe(4, False, "No candidate PID to verify")
|
||
return
|
||
try:
|
||
from gateway.status import _pid_exists
|
||
|
||
alive = bool(_pid_exists(candidate_pid))
|
||
_probe(4, alive, f"_pid_exists({candidate_pid}) => {alive}")
|
||
except Exception as exc:
|
||
_probe(4, False, f"_pid_exists raised: {exc!r}")
|
||
|
||
|
||
def _probe_state_file(state_path: Path) -> None:
|
||
if _probe_missing(5, state_path, "gateway_state.json"):
|
||
return
|
||
try:
|
||
state_data = json.loads(state_path.read_text(encoding="utf-8-sig"))
|
||
gateway_state = state_data.get("gateway_state")
|
||
updated_at = state_data.get("updated_at")
|
||
age_str = ""
|
||
if updated_at:
|
||
try:
|
||
updated_dt = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
|
||
age_seconds = int((datetime.now(timezone.utc) - updated_dt).total_seconds())
|
||
age_str = f" (updated {age_seconds}s ago)"
|
||
except Exception:
|
||
pass
|
||
_probe(5, gateway_state in ("running", "degraded"), f"gateway_state.json state={gateway_state!r}{age_str}")
|
||
except Exception as exc:
|
||
_probe(5, False, f"gateway_state.json present but unreadable: {exc}")
|
||
|
||
|
||
def _probe_exit_diag(diag_path: Path) -> None:
|
||
if _probe_missing(6, diag_path, "exit-diag log"):
|
||
return
|
||
try:
|
||
with open(diag_path, "rb") as fh:
|
||
fh.seek(0, 2) # last ~4KB; one event is well under 500 bytes
|
||
size = fh.tell()
|
||
fh.seek(max(0, size - 4096))
|
||
tail = fh.read().decode("utf-8", errors="replace").splitlines()
|
||
last_event = next((ln for ln in reversed(tail) if ln.strip()), "")
|
||
if not last_event:
|
||
_probe(6, False, f"exit-diag log empty: {diag_path}")
|
||
return
|
||
try:
|
||
event = json.loads(last_event)
|
||
tag = event.get("tag", "?")
|
||
_probe(6, tag in ("gateway.start",), f"Last lifecycle event: tag={tag} pid={event.get('pid', '?')} ts={event.get('ts', '?')}")
|
||
except Exception:
|
||
_probe(6, False, f"Last lifecycle line not JSON: {last_event[:120]}")
|
||
except Exception as exc:
|
||
_probe(6, False, f"exit-diag log unreadable: {exc}")
|
||
|
||
|
||
def _print_deep_probes() -> None:
|
||
"""Print PASS/FAIL per individual liveness signal, so when the collapsed ✓/✗ summary disagrees
|
||
with reality the user can see exactly which signal is wrong."""
|
||
home = _hermes_home()
|
||
print("\nDeep probes:")
|
||
pid_value = _probe_pid_file(home / "gateway.pid")
|
||
_probe_lock_file(home / "gateway.lock")
|
||
running_pid = _probe_running_pid()
|
||
_probe_pid_exists(running_pid if running_pid is not None else pid_value)
|
||
_probe_state_file(home / "gateway_state.json")
|
||
_probe_exit_diag(home / "logs" / "gateway-exit-diag.log")
|
||
|
||
|
||
def status(deep: bool = False) -> None:
|
||
"""Print a status report for the Windows gateway service."""
|
||
_assert_windows()
|
||
_print_start_attestation_warning() # once: a gateway that died after a previous ✓
|
||
task_name = get_task_name()
|
||
task_installed = is_task_registered()
|
||
startup_installed = is_startup_entry_installed()
|
||
pids = _gateway_pids()
|
||
|
||
if task_installed:
|
||
print(f"✓ Scheduled Task registered: {task_name}")
|
||
info = query_task_status()
|
||
for key in ("status", "last run time", "last run result"):
|
||
if key in info:
|
||
print(f" {key.title()}: {info[key]}")
|
||
_print_scheduled_task_drift(task_name)
|
||
elif startup_installed:
|
||
entry = get_startup_entry_path()
|
||
print(f"✓ Windows login item installed: {entry if entry.exists() else _legacy_startup_entry_path()}")
|
||
else:
|
||
print("✗ Gateway service not installed")
|
||
from hermes_cli.gateway_windows_legacy import warn_legacy_launchers
|
||
warn_legacy_launchers()
|
||
|
||
print(f"✓ Gateway process running (PID: {', '.join(map(str, pids))})" if pids else "✗ No gateway process detected")
|
||
|
||
if deep:
|
||
print()
|
||
print(f" Task name: {task_name}")
|
||
print(f" Task script: {get_task_script_path()}")
|
||
print(f" Startup entry: {get_startup_entry_path()}")
|
||
_print_deep_probes()
|
||
|
||
if not task_installed and not startup_installed and not pids:
|
||
print("\nTo install:\n hermes gateway install")
|
||
|
||
|
||
def start() -> None:
|
||
"""Start the gateway using the canonical detached Windows launch path."""
|
||
_assert_windows()
|
||
_print_start_attestation_warning() # once: the LAST start's ✓ turned out to be false
|
||
running_pids = _gateway_pids()
|
||
if running_pids:
|
||
_report_already_running(running_pids)
|
||
return
|
||
|
||
if not is_task_registered() and not is_startup_entry_installed():
|
||
# Login persistence is a lasting system change: a bare ``start`` installs it only on an explicit
|
||
# answer — the HERMES_GATEWAY_INSTALL_START_ON_LOGIN override or a real TTY prompt — never on a
|
||
# non-TTY default (#113977). Declining still starts the gateway; the command is ``start``.
|
||
start_on_login = _install_choice_from_env("HERMES_GATEWAY_INSTALL_START_ON_LOGIN")
|
||
if start_on_login is None:
|
||
from hermes_cli.setup import is_interactive_stdin, is_noninteractive, prompt_yes_no
|
||
|
||
print("✗ Gateway service is not installed")
|
||
if is_noninteractive() or not _stdout_isatty() or not _stdin_is_interactive(
|
||
isatty=is_interactive_stdin(), console_mode_ok=_stdin_console_mode_ok()
|
||
):
|
||
start_on_login = False
|
||
else:
|
||
start_on_login = prompt_yes_no(" Install it now so the gateway starts on login?", True)
|
||
if start_on_login:
|
||
# install() starts the gateway itself (start_now) and reports the outcome — including a UAC
|
||
# hand-off to an elevated child — so there is nothing left to spawn or to warn about here.
|
||
install(force=False, start_now=True, start_on_login=True)
|
||
return
|
||
print("ℹ Login auto-start not installed; add it later with: hermes gateway install")
|
||
elif is_task_registered():
|
||
reconcile_scheduled_task(get_task_name()) # like systemd's regenerate-on-stale before a start
|
||
|
||
# Manual starts use the same console-less direct spawn as restart() and install --start-now;
|
||
# Scheduled Task / Startup entries are only login persistence.
|
||
pid = _spawn_detached()
|
||
_report_gateway_start("direct spawn")
|
||
|
||
|
||
def _drain_gateway_pid(pid: int, drain_timeout: float) -> bool:
|
||
"""Write the planned-stop marker and wait for the PID to exit. Windows can't deliver POSIX signals
|
||
to an asyncio loop, so the marker is the ONLY way to ask the gateway to drain and persist."""
|
||
if pid <= 0:
|
||
return False
|
||
try:
|
||
from gateway.status import write_planned_stop_marker, _pid_exists
|
||
except ImportError:
|
||
return False
|
||
|
||
try:
|
||
write_planned_stop_marker(pid)
|
||
except Exception:
|
||
pass # best-effort; caller escalates to a hard kill
|
||
|
||
deadline = time.monotonic() + max(drain_timeout, 1.0)
|
||
while time.monotonic() < deadline:
|
||
if not _pid_exists(pid):
|
||
return True
|
||
time.sleep(0.5)
|
||
return False
|
||
|
||
|
||
def _windows_stop_drain_timeout() -> float:
|
||
"""Bounded stop grace period: a real graceful-drain window, but the CLI must never wedge."""
|
||
try:
|
||
from hermes_cli.gateway import _get_restart_drain_timeout
|
||
|
||
configured = float(_get_restart_drain_timeout() or 30.0)
|
||
except Exception:
|
||
configured = 30.0
|
||
return max(1.0, min(configured, 30.0))
|
||
|
||
|
||
def _gateway_pid_identities(pids: list[int]) -> dict[int, int | None]:
|
||
"""``{pid: start_time}`` fingerprints, captured BEFORE any drain/wait so a later force-kill can
|
||
detect that the PID was recycled meanwhile."""
|
||
try:
|
||
from gateway.status import get_process_start_time
|
||
except ImportError:
|
||
return {pid: None for pid in pids}
|
||
return {pid: get_process_start_time(pid) for pid in pids}
|
||
|
||
|
||
def _force_terminate_known_gateway_pids(identities: dict[int, int | None]) -> int:
|
||
"""Force-kill known gateway PIDs without a broad process sweep. ``identities`` maps each PID to
|
||
the start time observed when it was identified as a gateway (``_gateway_pid_identities``);
|
||
``terminate_pid`` refuses the kill when the live process no longer matches. Re-reading the start
|
||
time here would compare the process with itself and taskkill whatever now owns the PID."""
|
||
try:
|
||
from gateway.status import _pid_exists, terminate_pid
|
||
except ImportError:
|
||
return 0
|
||
|
||
own_pid = os.getpid()
|
||
killed = 0
|
||
for pid, expected_start_time in identities.items():
|
||
if pid <= 0 or pid == own_pid:
|
||
continue
|
||
try:
|
||
if not _pid_exists(pid):
|
||
continue
|
||
terminate_pid(pid, force=True, expected_start_time=expected_start_time)
|
||
killed += 1
|
||
except ProcessLookupError:
|
||
continue
|
||
except PermissionError:
|
||
print(f"⚠ Permission denied to kill PID {pid}")
|
||
except OSError as exc:
|
||
print(f"Failed to kill PID {pid}: {exc}")
|
||
return killed
|
||
|
||
|
||
def _collect_gateway_stop_pids(primary_pid: int | None = None) -> list[int]:
|
||
"""Collect gateway PIDs for the active profile, preserving primary first."""
|
||
pids: list[int] = []
|
||
if primary_pid is not None and primary_pid > 0:
|
||
pids.append(primary_pid)
|
||
try:
|
||
for pid in _gateway_pids():
|
||
if pid > 0 and pid not in pids:
|
||
pids.append(pid)
|
||
except Exception:
|
||
pass
|
||
return pids
|
||
|
||
|
||
def stop() -> None:
|
||
"""Stop the gateway: planned-stop marker first so it can drain in-flight agents and persist
|
||
``resume_pending`` (Windows asyncio can't receive SIGTERM — the marker is our only IPC), then
|
||
``schtasks /End``, then a bounded hard-kill of known PIDs."""
|
||
_assert_windows()
|
||
from gateway.status import get_running_pid
|
||
|
||
# A user-initiated stop is a planned death: don't later report it as a silent crash.
|
||
_clear_start_attestation()
|
||
|
||
pid = get_running_pid()
|
||
stop_pids = _collect_gateway_stop_pids(pid)
|
||
# Fingerprint before the drain: the kill below must refuse a PID recycled during the wait.
|
||
identities = _gateway_pid_identities(stop_pids)
|
||
drained = pid is not None and _drain_gateway_pid(pid, _windows_stop_drain_timeout())
|
||
|
||
stopped_any = drained
|
||
if is_task_registered():
|
||
code, _out, err = _exec_schtasks(["/End", "/TN", get_task_name()])
|
||
# schtasks returns nonzero when the task isn't currently running — not an error.
|
||
if code == 0:
|
||
stopped_any = True
|
||
elif "not running" not in (err or "").lower():
|
||
print(f"⚠ schtasks /End returned code {code}: {err.strip()}")
|
||
|
||
# No generic process sweep: starts are profile-scoped and stop must stay bounded even if wedged.
|
||
late_pids = [pid for pid in _collect_gateway_stop_pids() if pid not in identities]
|
||
identities.update(_gateway_pid_identities(late_pids))
|
||
killed = _force_terminate_known_gateway_pids(identities)
|
||
if killed:
|
||
stopped_any = True
|
||
print(f"✓ Killed {killed} gateway process(es)")
|
||
if stopped_any:
|
||
print("✓ Gateway stopped (drained cleanly)" if drained else "✓ Gateway stopped")
|
||
else:
|
||
print("✗ No gateway was running")
|
||
|
||
|
||
def _wait_for_gateway_absent(timeout_s: float = 30.0, interval_s: float = 0.5) -> bool:
|
||
"""Block until no gateway is detectable (authoritative ``get_running_pid()`` plus the strict
|
||
``_gateway_pids()`` scan) or the timeout elapses, so a relaunch never races a draining process."""
|
||
from gateway.status import get_running_pid
|
||
|
||
def _absent() -> bool:
|
||
return get_running_pid() is None and not _gateway_pids()
|
||
|
||
deadline = time.monotonic() + max(timeout_s, interval_s)
|
||
while time.monotonic() < deadline:
|
||
if _absent():
|
||
return True
|
||
time.sleep(interval_s)
|
||
return _absent()
|
||
|
||
|
||
def restart() -> None:
|
||
"""Stop then start. Waits for the old gateway to be authoritatively gone first; otherwise
|
||
``start()``'s "already running" guard sees the draining process and no-ops, and nothing
|
||
replaces it when it exits (a silent outage). Fails loudly on either side."""
|
||
_assert_windows()
|
||
|
||
stop()
|
||
|
||
if not _wait_for_gateway_absent(timeout_s=30.0):
|
||
print("⚠ Gateway still present after stop; forcing termination before restart...")
|
||
_force_terminate_known_gateway_pids(_gateway_pid_identities(_collect_gateway_stop_pids()))
|
||
if not _wait_for_gateway_absent(timeout_s=10.0):
|
||
raise RuntimeError(
|
||
"Gateway process still detected after force kill; refusing to "
|
||
"start a duplicate. Investigate stray PIDs before retrying."
|
||
)
|
||
|
||
from hermes_cli.gateway import _wait_for_api_server_port_free # avoid circular init
|
||
|
||
_wait_for_api_server_port_free()
|
||
start()
|
||
|
||
if not _wait_for_gateway_ready(timeout_s=15.0):
|
||
raise RuntimeError(
|
||
"Gateway restart did not produce a running gateway process. "
|
||
"Check logs/gateway.log and run `hermes gateway status`."
|
||
)
|