KrustyPlanet/hermes-agent
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(windows): suppress console flashes and harden gateway restarts

Browse source
This commit is contained in:
Gille 2026-06-24 23:31:52 -06:00 • committed by Teknium
parent c4ba4770eb
commit e7d2f0b93c
12 changed files with 617 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

18
hermes_cli/gateway.py
View file

@ -723,6 +723,10 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool:
import subprocess
import sys
import time
from hermes_cli._subprocess_compat import (
windows_detach_flags,
windows_detach_flags_without_breakaway,
)
pid = int(sys.argv[1])
cmd = sys.argv[2:]
@ -747,18 +751,8 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool:
"stderr": subprocess.DEVNULL,
}
if sys.platform == "win32":
_CREATE_NEW_PROCESS_GROUP = 0x00000200
_DETACHED_PROCESS = 0x00000008
_CREATE_NO_WINDOW = 0x08000000
_CREATE_BREAKAWAY_FROM_JOB = 0x01000000
_flags = (
_CREATE_NEW_PROCESS_GROUP
| _DETACHED_PROCESS
| _CREATE_NO_WINDOW
| _CREATE_BREAKAWAY_FROM_JOB
)
try:
_popen_kwargs["creationflags"] = _flags
_popen_kwargs["creationflags"] = windows_detach_flags()
subprocess.Popen(cmd, **_popen_kwargs)
except OSError:
# CREATE_BREAKAWAY_FROM_JOB can be rejected with
@ -766,7 +760,7 @@ def _spawn_gateway_restart_watcher(old_pid: int, run_argv: list[str]) -> bool:
# breakaway. Retry without it — DETACHED_PROCESS et al.
# alone are enough in most setups. Mirrors the canonical
# fallback in gateway_windows._spawn_detached.
_popen_kwargs["creationflags"] = _flags & ~_CREATE_BREAKAWAY_FROM_JOB
_popen_kwargs["creationflags"] = windows_detach_flags_without_breakaway()
subprocess.Popen(cmd, **_popen_kwargs)
else:
_popen_kwargs["start_new_session"] = True

260
hermes_cli/gateway_windows.py
View file

@ -3,21 +3,20 @@
This mirrors the contract exposed by ``launchd_install`` / ``launchd_start`` /
``launchd_status`` etc. on macOS and ``systemd_install`` / ``systemd_start`` on
Linux. It uses ``schtasks`` under the hood with ``/SC ONLOGON`` and restart-on-
failure XML settings, and falls back to a ``%APPDATA%\\...\\Startup\\<name>.cmd``
failure XML settings, and falls back to a ``%APPDATA%\\...\\Startup\\<name>.vbs``
dropper when Scheduled Task creation is denied (locked-down corporate boxes).
Design notes
------------
* ``schtasks /Create /SC ONLOGON /RL LIMITED`` means the task runs at the
CURRENT USER's next logon without any elevation prompt. We also
``schtasks /Run`` immediately after install so the gateway starts right
away without waiting for the next logon.
* We write two files: a shared ``gateway.cmd`` wrapper script (cwd + env + the
actual ``python -m hermes_cli.main gateway run --replace`` invocation) and
EITHER a schtasks entry pointing at it OR a Startup-folder ``.cmd`` that
spawns it detached.
CURRENT USER's next logon without any elevation prompt. Manual starts and
install ``--start-now`` use the direct detached ``pythonw`` launcher instead
of ``schtasks /Run`` so start/restart behavior is consistent.
* We write a shared ``gateway.cmd`` wrapper plus a console-less ``gateway.vbs``
launcher. Scheduled Task and Startup-folder persistence both route through
VBS/wscript; immediate manual starts route through direct ``subprocess`` spawn.
* Status = merge of "is the schtasks entry registered?" + "is the startup
.cmd present?" + "is there a gateway process running?" so the status
login item present?" + "is there a gateway process running?" so the status
command keeps working regardless of which install path was taken.
* Quoting is tricky: schtasks parses ``/TR`` itself and cmd.exe parses the
generated ``gateway.cmd``. Those are DIFFERENT parsers. We keep two
@ -40,6 +39,12 @@ import time
from pathlib import Path
from xml.sax.saxutils import escape
from hermes_cli._subprocess_compat import (
windows_detach_flags,
windows_detach_flags_without_breakaway,
windows_hide_flags,
)
# Short timeouts: schtasks occasionally wedges and we don't want to hang forever.
_SCHTASKS_TIMEOUT_S = 15
_SCHTASKS_NO_OUTPUT_TIMEOUT_S = 30
@ -80,6 +85,31 @@ def _assert_windows() -> None:
raise RuntimeError("gateway_windows is Windows-only")
def _preserve_hermes_home_path(path: str | Path) -> str:
"""Render Hermes-owned paths under the configured HERMES_HOME spelling.
Windows installs may keep ``%LOCALAPPDATA%\\hermes`` as a symlink/junction to
another drive. Runtime state should still identify itself by the configured
AppData path, so launcher files must not bake in the resolved target when a
path lives under HERMES_HOME.
"""
candidate = Path(path)
try:
from hermes_cli.config import get_hermes_home
home = Path(get_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:
rel = os.path.relpath(str(resolved_candidate), str(resolved_home))
return str(home / rel)
except Exception:
pass
return str(candidate)
# ---------------------------------------------------------------------------
# Quoting helpers (two DIFFERENT parsers — do not mix)
# ---------------------------------------------------------------------------
@ -141,7 +171,7 @@ def _exec_schtasks(args: list[str]) -> tuple[int, str, str]:
# CREATE_NO_WINDOW avoids a flashing console window when the CLI
# is itself hosted in a TUI. See tools/browser_tool.py for the
# same pattern and the windows-subprocess-sigint-storm.md ref.
creationflags=0x08000000, # CREATE_NO_WINDOW
creationflags=windows_hide_flags(),
)
return (proc.returncode, proc.stdout or "", proc.stderr or "")
except subprocess.TimeoutExpired:
@ -274,7 +304,7 @@ def _sanitize_filename(value: str) -> str:
def get_task_script_path() -> Path:
"""The generated ``gateway.cmd`` wrapper that the schtasks entry invokes.
"""The generated ``gateway.cmd`` wrapper kept beside the VBS launcher.
Lives under ``%LOCALAPPDATA%\\hermes\\gateway-service\\<task_name>.cmd``
(or ``<HERMES_HOME>/gateway-service/<task_name>.cmd`` so per-profile
@ -308,6 +338,11 @@ def _startup_dir() -> Path:
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"
@ -322,14 +357,18 @@ def _stable_gateway_working_dir(project_root: Path) -> str:
Mirror the POSIX service invariant: anchor at ``HERMES_HOME`` whenever it
exists so Scheduled Task / Startup launches do not fail at the ``cd`` step
after a transient checkout or worktree is moved away. Fall back to the
source checkout only if ``HERMES_HOME`` cannot be resolved yet.
source checkout only if ``HERMES_HOME`` cannot be used yet. Preserve the
configured spelling instead of resolving symlinks so AppData installs backed
by a junction/symlink still identify themselves as AppData.
"""
from hermes_cli.config import get_hermes_home
try:
home = get_hermes_home()
if home and Path(home).is_dir():
return str(Path(home).resolve())
if home:
home_path = Path(home)
if home_path.is_dir():
return str(home_path)
except Exception:
pass
return str(project_root)
@ -365,8 +404,11 @@ def _build_gateway_cmd_script(
pythonw_path, venv_dir, extra_pythonpath = _resolve_detached_python(python_path)
# VIRTUAL_ENV lets the gateway's own python detection find the venv
# if someone imports hermes_constants-based logic during startup.
lines.append(f'set "VIRTUAL_ENV={venv_dir}"')
pythonpath_entries = [str(Path(__file__).resolve().parent.parent), *extra_pythonpath]
lines.append(f'set "VIRTUAL_ENV={_preserve_hermes_home_path(venv_dir)}"')
pythonpath_entries = [
_preserve_hermes_home_path(Path(__file__).resolve().parent.parent),
*[_preserve_hermes_home_path(entry) for entry in extra_pythonpath],
]
lines.append(f'set "PYTHONPATH={";".join([*pythonpath_entries, "%PYTHONPATH%"])}"')
prog_args = [pythonw_path, "-m", "hermes_cli.main"]
@ -427,8 +469,10 @@ def _build_gateway_vbs_script(
# list2cmdline gives CreateProcess-correct quoting for WScript.Shell.Run.
command_line = subprocess.list2cmdline(prog_args)
repo_root = str(Path(__file__).resolve().parent.parent)
static_pythonpath = os.pathsep.join([repo_root, *extra_pythonpath])
repo_root = _preserve_hermes_home_path(Path(__file__).resolve().parent.parent)
static_pythonpath = os.pathsep.join(
[repo_root, *[_preserve_hermes_home_path(entry) for entry in extra_pythonpath]]
)
lines = [
f"' {_TASK_DESCRIPTION}",
@ -439,7 +483,7 @@ def _build_gateway_vbs_script(
f"env.Item({_quote_vbs_string('HERMES_HOME')}) = {_quote_vbs_string(hermes_home)}",
f"env.Item({_quote_vbs_string('PYTHONIOENCODING')}) = {_quote_vbs_string('utf-8')}",
f"env.Item({_quote_vbs_string('HERMES_GATEWAY_DETACHED')}) = {_quote_vbs_string('1')}",
f"env.Item({_quote_vbs_string('VIRTUAL_ENV')}) = {_quote_vbs_string(str(venv_dir))}",
f"env.Item({_quote_vbs_string('VIRTUAL_ENV')}) = {_quote_vbs_string(_preserve_hermes_home_path(venv_dir))}",
# Mirror the cmd wrapper's ``PYTHONPATH=<static>;%PYTHONPATH%``: chain onto
# whatever PYTHONPATH the task environment already carries, at runtime.
f"existing_pp = env.Item({_quote_vbs_string('PYTHONPATH')})",
@ -457,26 +501,25 @@ def _build_gateway_vbs_script(
def _build_startup_launcher(script_path: Path) -> str:
"""The tiny .cmd that goes in the Startup folder. Just minimizes and chains.
"""The tiny .vbs that goes in the Startup folder and chains hidden.
Defense-in-depth: bail out silently if the target script is gone. Test
fixtures historically wrote Startup entries pointing at pytest tmp_path
directories that vanish after the test session. Without the existence
guard, every subsequent Windows login flashes a cmd.exe window that
fails to find the target. The check + ``exit /b 0`` keeps that case
silent.
guard, every subsequent Windows login could attempt a stale launcher. The
check + ``WScript.Quit 0`` keeps that case silent.
"""
quoted_target = _quote_cmd_script_arg(str(script_path))
target = str(script_path.with_suffix(".vbs"))
command = subprocess.list2cmdline(["wscript.exe", target])
lines = [
"@echo off",
f"rem {_TASK_DESCRIPTION}",
# If the wrapper script is gone (typical for stale entries from
# uninstalled/migrated installs), silently no-op instead of
# flashing a cmd window with a "file not found" error.
f"if not exist {quoted_target} exit /b 0",
# ``start "" /min`` detaches with a minimized console window.
# ``/d /c`` on cmd.exe skips AUTORUN and runs the target script once.
f'start "" /min cmd.exe /d /c {quoted_target}',
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"
@ -492,9 +535,9 @@ def _write_task_script() -> Path:
get_python_path,
)
python_path = get_python_path()
python_path = _preserve_hermes_home_path(get_python_path())
working_dir = _stable_gateway_working_dir(PROJECT_ROOT)
hermes_home = str(Path(get_hermes_home()).resolve())
hermes_home = str(Path(get_hermes_home()))
profile_arg = _profile_arg(hermes_home)
content = _build_gateway_cmd_script(python_path, working_dir, hermes_home, profile_arg)
@ -503,9 +546,9 @@ def _write_task_script() -> Path:
tmp.write_text(content, encoding="utf-8", newline="")
tmp.replace(script_path)
# Also render the console-less .vbs launcher the Scheduled Task runs via
# wscript.exe (issue #45599 fix A). The .cmd above stays for the
# Startup-folder fallback and direct /Run paths.
# 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_content = _build_gateway_vbs_script(python_path, working_dir, hermes_home, profile_arg)
vbs_path = script_path.with_suffix(".vbs")
vbs_tmp = vbs_path.with_name(vbs_path.name + ".tmp")
@ -614,7 +657,7 @@ def _install_scheduled_task(task_name: str, script_path: Path) -> tuple[bool, st
# the final error if creation also fails.
user = _resolve_task_user()
# The Scheduled Task launches the console-less .vbs (issue #45599 fix A), not
# the .cmd. The .cmd stays for the Startup-folder fallback and direct /Run.
# the .cmd. Immediate manual starts use _spawn_detached().
launcher_path = script_path.with_suffix(".vbs")
xml_path = _write_scheduled_task_xml(task_name, launcher_path, user)
base = ["/Create", "/F", "/TN", task_name, "/XML", str(xml_path)]
@ -648,6 +691,12 @@ def _install_startup_entry(script_path: Path) -> Path:
tmp = entry.with_suffix(".tmp")
tmp.write_text(_build_startup_launcher(script_path), encoding="utf-8", newline="")
tmp.replace(entry)
legacy_entry = _legacy_startup_entry_path()
try:
if legacy_entry.exists():
legacy_entry.unlink()
except OSError:
pass
return entry
@ -732,10 +781,12 @@ def _build_gateway_argv() -> tuple[list[str], str, dict[str, str]]:
get_python_path,
)
python_exe, venv_dir, extra_pythonpath = _resolve_detached_python(get_python_path())
project_root = str(PROJECT_ROOT)
python_exe, venv_dir, extra_pythonpath = _resolve_detached_python(
_preserve_hermes_home_path(get_python_path())
)
project_root = _preserve_hermes_home_path(PROJECT_ROOT)
working_dir = _stable_gateway_working_dir(PROJECT_ROOT)
hermes_home = str(Path(get_hermes_home()).resolve())
hermes_home = str(Path(get_hermes_home()))
profile_arg = _profile_arg(hermes_home)
argv = [python_exe, "-m", "hermes_cli.main"]
@ -747,9 +798,14 @@ def _build_gateway_argv() -> tuple[list[str], str, dict[str, str]]:
"HERMES_HOME": hermes_home,
"PYTHONIOENCODING": "utf-8",
"HERMES_GATEWAY_DETACHED": "1",
"VIRTUAL_ENV": str(venv_dir),
"VIRTUAL_ENV": _preserve_hermes_home_path(venv_dir),
}
_prepend_pythonpath(env_overlay, [project_root, *extra_pythonpath] if extra_pythonpath else [project_root])
_prepend_pythonpath(
env_overlay,
[project_root, *[_preserve_hermes_home_path(entry) for entry in extra_pythonpath]]
if extra_pythonpath
else [project_root],
)
return argv, working_dir, env_overlay
@ -785,7 +841,7 @@ def _spawn_detached(script_path: Path | None = None) -> int:
# job teardown from reaping us;
# some Windows Terminal versions
# wrap their children in a job).
flags = 0x00000008 | 0x00000200 | 0x08000000 | 0x01000000
flags = windows_detach_flags()
# Redirect any stray stdout/stderr output to a sidecar log. Python's
# logging module writes to gateway.log through a FileHandler, so the
@ -814,7 +870,7 @@ def _spawn_detached(script_path: Path | None = None) -> int:
# parent's job object doesn't permit breakaway (some Windows
# Terminal configs). Retry without the breakaway flag — in most
# setups pythonw.exe + DETACHED_PROCESS is enough on its own.
flags_no_breakaway = flags & ~0x01000000
flags_no_breakaway = windows_detach_flags_without_breakaway()
with open(stray_log, "ab", buffering=0) as log_fh:
proc = subprocess.Popen(
argv,
@ -1045,14 +1101,14 @@ def _report_gateway_start(via: str) -> None:
print(f"⚠ Launched gateway via {via}, but no process detected after 6s.")
print(" Check the log for startup errors:")
from hermes_cli.config import get_hermes_home
print(f" type {Path(get_hermes_home()).resolve()}\\logs\\gateway.log")
print(f" type {Path(get_hermes_home()).resolve()}\\logs\\gateway-stdio.log")
print(f" type {Path(get_hermes_home())}\\logs\\gateway.log")
print(f" type {Path(get_hermes_home())}\\logs\\gateway-stdio.log")
def _print_next_steps() -> None:
from hermes_cli.config import get_hermes_home
hermes_home = Path(get_hermes_home()).resolve()
hermes_home = Path(get_hermes_home())
print()
print("Next steps:")
print(" hermes gateway status # Check status")
@ -1064,7 +1120,9 @@ def uninstall() -> None:
_assert_windows()
task_name = get_task_name()
script_path = get_task_script_path()
vbs_script_path = script_path.with_suffix(".vbs")
startup_entry = get_startup_entry_path()
legacy_startup_entry = _legacy_startup_entry_path()
scheduled_task_removed = False
if is_task_registered():
@ -1091,7 +1149,9 @@ def uninstall() -> None:
for path, label in [
(startup_entry, "Windows login item"),
(legacy_startup_entry, "legacy Windows login item"),
(script_path, "Task script"),
(vbs_script_path, "Task launcher"),
]:
try:
path.unlink()
@ -1113,7 +1173,7 @@ def is_task_registered() -> bool:
def is_startup_entry_installed() -> bool:
return get_startup_entry_path().exists()
return get_startup_entry_path().exists() or _legacy_startup_entry_path().exists()
def is_installed() -> bool:
@ -1171,7 +1231,7 @@ def _print_deep_probes() -> None:
from hermes_cli.config import get_hermes_home
home = Path(get_hermes_home()).resolve()
home = Path(get_hermes_home())
pid_path = home / "gateway.pid"
lock_path = home / "gateway.lock"
state_path = home / "gateway_state.json"
@ -1299,7 +1359,10 @@ def status(deep: bool = False) -> None:
if key in info:
print(f" {key.title()}: {info[key]}")
elif startup_installed:
print(f"✓ Windows login item installed: {get_startup_entry_path()}")
entry = get_startup_entry_path()
if not entry.exists():
entry = _legacy_startup_entry_path()
print(f"✓ Windows login item installed: {entry}")
else:
print("✗ Gateway service not installed")
@ -1324,7 +1387,7 @@ def status(deep: bool = False) -> None:
def start() -> None:
"""Start the gateway. Prefers /Run on the scheduled task if present."""
"""Start the gateway using the canonical detached Windows launch path."""
_assert_windows()
running_pids = _gateway_pids()
if running_pids:
@ -1349,14 +1412,9 @@ def start() -> None:
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")
# Startup fallback or failed /Run: direct spawn one foreground-detached gateway.
# Manual starts use the same console-less direct spawn path as restart()
# and install --start-now. Scheduled Task / Startup entries are only login
# persistence mechanisms.
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")
@ -1397,6 +1455,61 @@ def _drain_gateway_pid(pid: int, drain_timeout: float) -> bool:
return False
def _windows_stop_drain_timeout() -> float:
"""Return a bounded Windows gateway stop grace period."""
try:
from hermes_cli.gateway import _get_restart_drain_timeout
configured = float(_get_restart_drain_timeout() or 30.0)
except Exception:
configured = 30.0
# Windows CLI stop must not wedge forever. Give the gateway a real
# graceful-drain window, then escalate to the known PID.
return max(1.0, min(configured, 30.0))
def _force_terminate_known_gateway_pids(pids: list[int]) -> int:
"""Force-kill known gateway PIDs without a broad process sweep."""
try:
from gateway.status import _pid_exists, terminate_pid
except ImportError:
return 0
own_pid = os.getpid()
killed = 0
seen: set[int] = set()
for pid in pids:
if pid <= 0 or pid == own_pid or pid in seen:
continue
seen.add(pid)
try:
if not _pid_exists(pid):
continue
terminate_pid(pid, force=True)
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.
@ -1404,11 +1517,10 @@ def stop() -> None:
in-flight agents and persist ``resume_pending`` before exit (the
gateway's marker-watcher thread picks this up — Windows asyncio
can't deliver SIGTERM to the loop, so the marker is our only IPC).
Then escalates: ``schtasks /End`` (kills the scheduled-task tree)
+ ``kill_gateway_processes(force=True)`` for any strays.
Then escalates with bounded Windows process termination against the
known gateway PID(s).
"""
_assert_windows()
from hermes_cli.gateway import kill_gateway_processes, _get_restart_drain_timeout
from gateway.status import get_running_pid
# Phase 1: ask the running gateway (if any) to drain itself by writing
@ -1416,13 +1528,10 @@ def stop() -> None:
# On clean exit, sessions land with resume_pending=True and the next
# boot will auto-resume them.
pid = get_running_pid()
stop_pids = _collect_gateway_stop_pids(pid)
drained = False
if pid is not None:
try:
drain_timeout = float(_get_restart_drain_timeout() or 30.0)
except Exception:
drain_timeout = 30.0
drained = _drain_gateway_pid(pid, drain_timeout)
drained = _drain_gateway_pid(pid, _windows_stop_drain_timeout())
stopped_any = drained
if is_task_registered():
@ -1433,11 +1542,11 @@ def stop() -> None:
elif "not running" not in (err or "").lower():
print(f"⚠ schtasks /End returned code {code}: {err.strip()}")
# Phase 3: hard-kill any strays. When drain succeeded this is a no-op;
# when drain timed out this is the escalation that ensures the PID
# actually exits. Use force=True on Windows so taskkill /T /F walks
# the descendant tree (browser helpers, etc.).
killed = kill_gateway_processes(all_profiles=False, force=not drained)
# Phase 3: hard-kill any still-known gateway processes. Avoid the generic
# process sweep here: Windows direct-spawn starts are profile-scoped, and a
# stop command must be bounded even if the scanner or shutdown path is wedged.
stop_pids.extend(pid for pid in _collect_gateway_stop_pids() if pid not in stop_pids)
killed = _force_terminate_known_gateway_pids(stop_pids)
if killed:
stopped_any = True
print(f"✓ Killed {killed} gateway process(es)")
@ -1479,13 +1588,12 @@ def restart() -> None:
doesn't produce a running gateway.
"""
_assert_windows()
from hermes_cli.gateway import kill_gateway_processes
stop()
if not _wait_for_gateway_absent(timeout_s=30.0):
print("⚠ Gateway still present after stop; forcing termination before restart...")
kill_gateway_processes(all_profiles=False, force=True)
_force_terminate_known_gateway_pids(_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 "

74
hermes_cli/web_server.py
View file

@ -33,6 +33,8 @@ import threading
import time
import urllib.error
import urllib.parse
from hermes_cli._subprocess_compat import windows_detach_flags, windows_hide_flags
import urllib.request
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
@ -1228,12 +1230,17 @@ def _fs_default_cwd() -> str:
def _fs_git_branch(cwd: str) -> str:
try:
run_kwargs: Dict[str, Any] = {
"capture_output": True,
"text": True,
"timeout": 2,
"check": False,
}
if sys.platform == "win32":
run_kwargs["creationflags"] = windows_hide_flags()
result = subprocess.run(
["git", "-C", cwd, "branch", "--show-current"],
capture_output=True,
text=True,
timeout=2,
check=False,
**run_kwargs,
)
return result.stdout.strip() if result.returncode == 0 else ""
except Exception:
@ -2387,6 +2394,18 @@ def _record_completed_action(name: str, message: str, exit_code: int = 1) -> Non
_ACTION_RESULTS[name] = {"exit_code": exit_code, "pid": None}
def _dashboard_spawn_executable() -> str:
"""Prefer pythonw.exe for detached dashboard actions on Windows."""
if sys.platform != "win32":
return sys.executable
exe = sys.executable
if exe.lower().endswith("python.exe"):
pythonw = os.path.join(os.path.dirname(exe), "pythonw.exe")
if os.path.isfile(pythonw):
return pythonw
return exe
def _spawn_hermes_action(subcommand: List[str], name: str) -> subprocess.Popen:
"""Spawn ``hermes <subcommand>`` detached and record the Popen handle.
@ -2401,7 +2420,7 @@ def _spawn_hermes_action(subcommand: List[str], name: str) -> subprocess.Popen:
f"\n=== {name} started {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n".encode()
)
cmd = [sys.executable, "-m", "hermes_cli.main", *subcommand]
cmd = [_dashboard_spawn_executable(), "-m", "hermes_cli.main", *subcommand]
popen_kwargs: Dict[str, Any] = {
"cwd": str(PROJECT_ROOT),
@ -2411,10 +2430,7 @@ def _spawn_hermes_action(subcommand: List[str], name: str) -> subprocess.Popen:
"env": {**os.environ, "HERMES_NONINTERACTIVE": "1"},
}
if sys.platform == "win32":
popen_kwargs["creationflags"] = (
subprocess.CREATE_NEW_PROCESS_GROUP # type: ignore[attr-defined]
| getattr(subprocess, "DETACHED_PROCESS", 0)
)
popen_kwargs["creationflags"] = windows_detach_flags()
else:
popen_kwargs["start_new_session"] = True
@ -12943,6 +12959,45 @@ def _read_bound_port(server: "uvicorn.Server", fallback: int) -> int:
return fallback
def _write_dashboard_ready_file(actual_port: int) -> None:
"""Optionally publish the dashboard port through an atomic ready file.
Windows Desktop can launch dashboard backends with ``pythonw.exe`` to avoid
console flashes. That path cannot rely on stdout for the port announcement,
so Electron passes ``HERMES_DESKTOP_READY_FILE`` and waits for this JSON.
Normal CLI/dashboard launches still use the stdout READY line below.
"""
target = os.environ.get("HERMES_DESKTOP_READY_FILE")
if not target:
return
tmp_name = ""
try:
path = Path(target)
path.parent.mkdir(parents=True, exist_ok=True)
payload = json.dumps({"port": int(actual_port)}, separators=(",", ":"))
with tempfile.NamedTemporaryFile(
"w",
encoding="utf-8",
dir=str(path.parent),
prefix=f"{path.name}.",
suffix=".tmp",
delete=False,
) as fh:
fh.write(payload)
fh.flush()
os.fsync(fh.fileno())
tmp_name = fh.name
os.replace(tmp_name, path)
except Exception as exc:
if tmp_name:
try:
Path(tmp_name).unlink(missing_ok=True)
except Exception:
pass
_log.warning("Failed to write dashboard ready file %r: %s", target, exc)
def _maybe_open_browser(
host: str, actual_port: int, open_browser: bool, initial_profile: str
) -> None:
@ -13134,6 +13189,7 @@ def start_server(
actual_port = _read_bound_port(server, fallback=port)
app.state.bound_port = actual_port
_write_dashboard_ready_file(actual_port)
print(f"HERMES_DASHBOARD_READY port={actual_port}", flush=True)
print(f" Hermes Web UI → http://{host}:{actual_port}")
_maybe_open_browser(host, actual_port, open_browser, initial_profile)