revert(windows): roll back terminal-popup PRs #53791 #53810 #53829 (#53853)

* Revert "fix(windows): capture is not a no-window boundary; route flashing spawns through chokepoint (#53829)"

This reverts commit 2ecca1e7d3.

* Revert "fix(windows): stop terminal-window popups from background spawns (#53810)"

This reverts commit 5db1430af9.

* Revert "fix(windows): stop subprocess console-window popups + add CI guard (#53791)"

This reverts commit ef17cd204d.
This commit is contained in:
Teknium 2026-06-27 15:59:00 -07:00 committed by GitHub
parent 1d32e5d98c
commit d3d621f7c3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
65 changed files with 172 additions and 1284 deletions

View file

@ -819,37 +819,6 @@ that touches the OS, assume *any* platform can hit your code path.
_quote_cmd_script_arg` and `_quote_schtasks_arg` for the reference
pair.
17. **Spawning a console program from a background/GUI parent needs a
no-window flag on Windows — and CI enforces it.** A `subprocess.run(["git",
...])` / `Popen(...)` of a cross-platform console exe (git, gh, npm, node,
python, uv, ffmpeg, docker, …) allocates and flashes a cmd/conhost window
on Windows when the parent has no console of its own (Desktop/Electron,
`pythonw.exe`, a detached gateway/cron). **Capturing or redirecting stdio
does NOT prevent this** — `capture_output=`/`stdout=` controls where the
child's *output* goes, not whether a console is *allocated*. Only
`CREATE_NO_WINDOW` suppresses the window. This was the single biggest
source of "terminal popups" bug reports. Prefer the chokepoint wrapper —
it always injects the flag on Windows and is a no-op on POSIX:
```python
from hermes_cli import _subprocess_compat
_subprocess_compat.run(cmd, capture_output=True, text=True) # never flashes
_subprocess_compat.popen(cmd) # never flashes
# detached background daemon:
subprocess.Popen(cmd, **windows_detach_popen_kwargs())
# or, at a site you can't route through the wrapper:
subprocess.run(cmd, creationflags=windows_hide_flags())
```
`scripts/check-windows-footguns.py` (AST-based) flags raw `subprocess.*`
calls that can create a new console. It exempts calls that pass
`creationflags=`, use `**windows_*_kwargs` spread, or run a provably
POSIX-only program (`launchctl`, `systemctl`, `brew`, …). It does **not**
treat `capture_output`/`stdout=`/`check_output` as safe for the known
Windows-flashing programs above. Calls routed through
`_subprocess_compat.run/popen` are inherently safe (the wrapper carries the
flag). If a visible window is genuinely intended (interactive editor/terminal
launch, foreground re-exec, `cmd /c start`), add `# windows-footgun: ok` on
the call line.
### Testing cross-platform
Tests that use POSIX-only syscalls need a skip marker. Common ones:

View file

@ -1274,7 +1274,7 @@ def run_oauth_setup_token() -> Optional[str]:
# concern does not apply to an interactive login the user explicitly
# invokes. noqa: subprocess-stdin
try:
subprocess.run([claude_path, "setup-token"]) # windows-footgun: ok — claude setup-token is interactive OAuth
subprocess.run([claude_path, "setup-token"])
except (KeyboardInterrupt, EOFError):
return None

View file

@ -59,7 +59,6 @@ import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
from hermes_cli import _subprocess_compat
logger = logging.getLogger("hermes.coding_context")
@ -649,7 +648,7 @@ def _enabled_mcp_servers(config: Optional[dict[str, Any]]) -> list[str]:
def _git(cwd: Path, *args: str) -> str:
try:
out = _subprocess_compat.run(
out = subprocess.run(
["git", "-C", str(cwd), *args],
capture_output=True,
text=True,

View file

@ -12,7 +12,6 @@ from pathlib import Path
from typing import Awaitable, Callable
from agent.model_metadata import estimate_tokens_rough
from hermes_cli import _subprocess_compat
_QUOTED_REFERENCE_VALUE = r'(?:`[^`\n]+`|"[^"\n]+"|\'[^\'\n]+\')'
REFERENCE_PATTERN = re.compile(
@ -292,7 +291,7 @@ def _expand_git_reference(
label: str,
) -> tuple[str | None, str | None]:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", *args],
cwd=cwd,
capture_output=True,

View file

@ -1565,17 +1565,18 @@ function readVenvHome(venvRoot) {
function getNoConsoleVenvPython(venvRoot) {
if (!IS_WINDOWS) return getVenvPython(venvRoot)
// uv venv launchers can re-exec console python.exe, which allocates conhost /
// Windows Terminal. Use base pythonw directly and provide imports via env.
// Prefer the venv's own pythonw shim — it carries pyvenv.cfg / site-packages
// wiring. Falling back to the base uv/python.org pythonw.exe skips the venv
// and breaks imports (yaml, hermes_cli, …) even when PYTHONPATH is patched.
const venvPythonw = path.join(venvRoot, 'Scripts', 'pythonw.exe')
if (fileExists(venvPythonw)) return venvPythonw
const baseHome = readVenvHome(venvRoot)
if (baseHome) {
const basePythonw = path.join(baseHome, 'pythonw.exe')
if (fileExists(basePythonw)) return basePythonw
}
const venvPythonw = path.join(venvRoot, 'Scripts', 'pythonw.exe')
if (fileExists(venvPythonw)) return venvPythonw
return venvPythonw
}
@ -2796,7 +2797,7 @@ function createPythonBackend(root, label, dashboardArgs, options = {}) {
args: ['-m', 'hermes_cli.main', ...dashboardArgs],
env: buildDesktopBackendEnv({
hermesHome: HERMES_HOME,
pythonPathEntries: [root, ...getVenvSitePackagesEntries(venvRoot)],
pythonPathEntries: [root],
venvRoot
}),
root,
@ -2820,7 +2821,7 @@ function createActiveBackend(dashboardArgs) {
args: ['-m', 'hermes_cli.main', ...dashboardArgs],
env: buildDesktopBackendEnv({
hermesHome: HERMES_HOME,
pythonPathEntries: [ACTIVE_HERMES_ROOT, ...getVenvSitePackagesEntries(VENV_ROOT)],
pythonPathEntries: [ACTIVE_HERMES_ROOT],
venvRoot: VENV_ROOT
}),
root: ACTIVE_HERMES_ROOT,

View file

@ -3046,11 +3046,4 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
if __name__ == "__main__":
# Standalone background scheduler: drop any console a uv pythonw→python
# re-exec auto-allocated. No-op on POSIX / when run in-gateway.
try:
import hermes_bootstrap
hermes_bootstrap.detach_orphan_console()
except Exception:
pass
tick(verbose=True)

View file

@ -54,7 +54,6 @@ from gateway.platforms.base import (
MessageType,
SendResult,
)
from hermes_cli import _subprocess_compat
logger = logging.getLogger(__name__)
@ -940,7 +939,7 @@ class WebhookAdapter(BasePlatformAdapter):
)
try:
result = _subprocess_compat.run(
result = subprocess.run(
[
"gh",
"pr",

View file

@ -18606,13 +18606,6 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
def main():
"""CLI entry point for the gateway."""
# Background daemon: drop any console auto-allocated by a uv pythonw→python
# re-exec so no terminal lingers. No-op on POSIX / when already detached.
try:
hermes_bootstrap.detach_orphan_console()
except Exception:
pass
# Force UTF-8 stdio on Windows — gateway logs and startup banner would
# otherwise UnicodeEncodeError on cp1252 consoles. No-op on POSIX.
try:

View file

@ -81,10 +81,8 @@ def terminate_pid(pid: int, *, force: bool = False) -> None:
because os.kill(..., SIGTERM) is not equivalent to a tree-killing hard stop.
"""
if force and _IS_WINDOWS:
from hermes_cli import _subprocess_compat
try:
result = _subprocess_compat.run(
result = subprocess.run(
["taskkill", "/PID", str(pid), "/T", "/F"],
capture_output=True,
text=True,

View file

@ -80,26 +80,6 @@ def apply_windows_utf8_bootstrap() -> bool:
os.environ.setdefault("PYTHONUTF8", "1")
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
# Python's platform.win32_ver()/platform.platform() can shell out to
# ``cmd.exe /c ver`` on Windows. In pythonw-launched background processes
# that still creates a visible terminal handoff on machines where Windows
# Terminal is the default console host. Disable that subprocess path early.
try:
import platform
def _no_subprocess_syscmd_ver(
system: str = "",
release: str = "",
version: str = "",
*_args,
**_kwargs,
) -> tuple[str, str, str]:
return system or "Windows", release, version
platform._syscmd_ver = _no_subprocess_syscmd_ver # type: ignore[attr-defined]
except Exception:
pass
# 2. Reconfigure the current process's stdio to UTF-8. Needed
# because os.environ changes don't retroactively rebind sys.stdout
# — those were bound at interpreter startup based on the console
@ -142,44 +122,6 @@ def apply_windows_utf8_bootstrap() -> bool:
return True
def detach_orphan_console() -> bool:
"""Free a console window that was auto-allocated for this process alone.
Background-only entry points (gateway daemon, dashboard backend, cron
runner, TUI/desktop stdio backends) call this explicitly. uv-created venvs
ship a ``Scripts\\pythonw.exe`` redirector that re-execs the *base* console
``python.exe``; that re-exec allocates its own conhost/Windows Terminal
window even though the launcher wanted no console. We drop it so nothing
lingers.
This is NOT wired into the import-time bootstrap on purpose: the discriminator
(``GetConsoleProcessList() == 1``) cannot tell a phantom console apart from a
user who deliberately opened the *interactive* CLI/TUI in its own fresh
console (double-click, Start-menu shortcut, a ConPTY), since both report a
single attached process with a tty. Intent is only knowable from the entry
point so only known-background mains call this, never the interactive CLI.
A properly detached daemon (``DETACHED_PROCESS``) has no console at all, so
``GetConsoleWindow()`` is NULL and this is a no-op. Returns True iff a console
was actually freed. No-op (returns False) on non-Windows.
"""
if not _IS_WINDOWS:
return False
try:
import ctypes
kernel32 = ctypes.windll.kernel32
if not kernel32.GetConsoleWindow():
return False
buf = (ctypes.c_uint * 4)()
if kernel32.GetConsoleProcessList(buf, 4) == 1:
kernel32.FreeConsole()
return True
except Exception:
pass
return False
def harden_import_path(src_root: str | None = None) -> None:
"""Stop a package in the current directory from shadowing Hermes modules.

View file

@ -28,15 +28,12 @@ guarantee.
from __future__ import annotations
import shutil
import subprocess
import sys
from typing import Sequence
__all__ = [
"IS_WINDOWS",
"resolve_node_command",
"run",
"popen",
"windows_detach_flags",
"windows_detach_flags_without_breakaway",
"windows_hide_flags",
@ -204,44 +201,6 @@ def windows_hide_flags() -> int:
return _CREATE_NO_WINDOW
# -----------------------------------------------------------------------------
# The single chokepoint for spawning a process without a console window.
# -----------------------------------------------------------------------------
def _no_window(kwargs: dict) -> dict:
"""OR ``CREATE_NO_WINDOW`` into ``creationflags`` on Windows (no-op on POSIX).
Merges rather than overwrites, so a caller that needs detach semantics can
pass ``creationflags=windows_detach_flags()`` and still go through here
``CREATE_NO_WINDOW`` is already part of that bundle, so the OR is idempotent.
"""
if IS_WINDOWS:
kwargs["creationflags"] = kwargs.get("creationflags", 0) | _CREATE_NO_WINDOW
return kwargs
def run(cmd, **kwargs):
"""``subprocess.run`` that never flashes a console window on Windows.
This is the primitive every Hermes spawn of a *console-subsystem* program
(``taskkill``, ``schtasks``, ``agent-browser``, ``git-bash``, version
probes, ) must use. Routing through one function makes "no visible
terminal" structural instead of a per-call-site rule that gets forgotten —
which is exactly how cron-driven and future spawns leaked windows before.
Python child processes are additionally covered by the ``FreeConsole``
catch-all in :mod:`hermes_bootstrap`, but native exes can't run that, so the
spawn-time flag here is the only thing that helps them.
"""
return subprocess.run(cmd, **_no_window(kwargs))
def popen(cmd, **kwargs):
"""``subprocess.Popen`` counterpart of :func:`run` — see its docstring."""
return subprocess.Popen(cmd, **_no_window(kwargs))
def windows_detach_popen_kwargs() -> dict:
"""Return a dict of Popen kwargs that detach a child on Windows and
fall back to the POSIX equivalent (``start_new_session=True``) on

View file

@ -60,7 +60,6 @@ def _skin_color(key: str, fallback: str) -> str:
# =========================================================================
from hermes_cli import __version__ as VERSION, __release_date__ as RELEASE_DATE
from hermes_cli import _subprocess_compat
HERMES_AGENT_LOGO = """[bold #FFD700]██╗ ██╗███████╗██████╗ ███╗ ███╗███████╗███████╗ █████╗ ██████╗ ███████╗███╗ ██╗████████╗[/]
[bold #FFD700]██║ ██║██╔════╝██╔══██╗████╗ ████║██╔════╝██╔════╝ ██╔══██╗██╔════╝ ██╔════╝████╗ ██║╚══██╔══╝[/]
@ -158,7 +157,7 @@ def _is_official_ssh_remote(url: str | None) -> bool:
def _git_stdout(args: list[str], *, cwd: Path, timeout: int = 5) -> Optional[str]:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", *args],
capture_output=True,
text=True,
@ -179,7 +178,7 @@ def _check_via_rev(local_rev: str) -> Optional[int]:
or ``None`` on failure.
"""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "ls-remote", _UPSTREAM_REPO_URL, "refs/heads/main"],
capture_output=True, text=True, timeout=10,
)
@ -241,7 +240,7 @@ def _check_via_local_git(repo_dir: Path) -> Optional[int]:
return 0 if head_rev == target_rev else UPDATE_AVAILABLE_NO_COUNT
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "rev-list", "--count", "HEAD..origin/main"],
capture_output=True, text=True, timeout=5,
cwd=str(repo_dir),
@ -388,7 +387,7 @@ def _resolve_repo_dir() -> Optional[Path]:
def _git_short_hash(repo_dir: Path, rev: str) -> Optional[str]:
"""Resolve a git revision to an 8-character short hash."""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "rev-parse", "--short=8", rev],
capture_output=True,
text=True,
@ -444,7 +443,7 @@ def get_git_banner_state(repo_dir: Optional[Path] = None) -> Optional[dict]:
ahead = 0
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "rev-list", "--count", "origin/main..HEAD"],
capture_output=True,
text=True,
@ -480,7 +479,7 @@ def get_latest_release_tag(repo_dir: Optional[Path] = None) -> Optional[tuple]:
return None
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "describe", "--tags", "--abbrev=0"],
capture_output=True,
text=True,

View file

@ -77,11 +77,9 @@ def _detect_openclaw_processes() -> list[str]:
# -- process scan ------------------------------------------------------
if sys.platform == "win32":
from hermes_cli import _subprocess_compat
try:
for exe in ("openclaw.exe", "clawd.exe"):
result = _subprocess_compat.run(
result = subprocess.run(
["tasklist", "/FI", f"IMAGENAME eq {exe}"],
capture_output=True, text=True, timeout=5,
)
@ -95,7 +93,7 @@ def _detect_openclaw_processes() -> list[str]:
'Where-Object { $_.CommandLine -match "openclaw|clawd" } | '
'Select-Object -First 1 ProcessId'
)
result = _subprocess_compat.run(
result = subprocess.run(
["powershell", "-NoProfile", "-Command", ps_cmd],
capture_output=True, text=True, timeout=5,
)

View file

@ -2260,11 +2260,11 @@ class CLICommandsMixin:
if initial_text:
fh.write(initial_text)
try:
subprocess.call([*shlex.split(editor), path]) # windows-footgun: ok — $EDITOR launch is interactive/foreground
subprocess.call([*shlex.split(editor), path])
except Exception:
# Fall back to a bare invocation (editor value may not be a
# simple argv-splittable string on some platforms).
subprocess.call(f"{editor} {shlex.quote(path)}", shell=True) # windows-footgun: ok — $EDITOR launch is interactive/foreground
subprocess.call(f"{editor} {shlex.quote(path)}", shell=True)
with open(path, "r", encoding="utf-8") as fh:
raw = fh.read()
finally:

View file

@ -198,9 +198,7 @@ _POWERSHELL_EXTRACT_IMAGE_SCRIPTS = (
def _run_powershell(exe: str, script: str, timeout: int) -> subprocess.CompletedProcess:
from hermes_cli import _subprocess_compat
return _subprocess_compat.run(
return subprocess.run(
[exe, "-NoProfile", "-NonInteractive", "-Command", script],
capture_output=True, text=True, timeout=timeout,
)
@ -256,11 +254,9 @@ def _powershell_save_image(exe: str, dest: Path, *, timeout: int, label: str) ->
def _find_powershell() -> str | None:
"""Return the first available PowerShell executable, or None."""
from hermes_cli import _subprocess_compat
for name in ("powershell", "pwsh"):
try:
r = _subprocess_compat.run(
r = subprocess.run(
[name, "-NoProfile", "-NonInteractive", "-Command", "echo ok"],
capture_output=True, text=True, timeout=5,
)

View file

@ -7019,7 +7019,7 @@ def edit_config():
return
print(f"Opening {config_path} in {editor}...")
subprocess.run([editor, str(config_path)]) # windows-footgun: ok — $EDITOR launch is interactive/foreground
subprocess.run([editor, str(config_path)])
def set_config_value(key: str, value: str):

View file

@ -23,7 +23,6 @@ import sys
from pathlib import Path
from hermes_constants import agent_browser_runnable
from hermes_cli._subprocess_compat import windows_hide_flags
_IS_WINDOWS = platform.system() == "Windows"
@ -153,7 +152,6 @@ def ensure_dependency(
result = subprocess.run(
cmd,
env=run_env,
creationflags=windows_hide_flags(),
)
if result.returncode != 0:
return False

View file

@ -56,7 +56,6 @@ _PROVIDER_ENV_HINTS = (
from hermes_constants import is_termux as _is_termux
from hermes_cli import _subprocess_compat
def _python_install_cmd() -> str:
@ -1437,7 +1436,7 @@ def run_doctor(args):
if _safe_which("docker"):
# Check if docker daemon is running
try:
result = _subprocess_compat.run(["docker", "info"], capture_output=True, timeout=10)
result = subprocess.run(["docker", "info"], capture_output=True, timeout=10)
except subprocess.TimeoutExpired:
result = None
if result is not None and result.returncode == 0:
@ -2194,7 +2193,7 @@ def run_doctor(args):
def _gh_authenticated() -> bool:
"""Check if gh CLI is authenticated via token file or device flow."""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["gh", "auth", "status", "--json", "authenticated"],
capture_output=True, timeout=10,
)

View file

@ -17,7 +17,6 @@ from hermes_cli.config import get_hermes_home, get_env_path, get_project_root, l
from hermes_cli.env_loader import load_hermes_dotenv
from hermes_constants import display_hermes_home
from agent.skill_utils import is_excluded_skill_path
from hermes_cli import _subprocess_compat
def _get_git_commit(project_root: Path) -> str:
@ -31,7 +30,7 @@ def _get_git_commit(project_root: Path) -> str:
The output format is identical regardless of source.
"""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "rev-parse", "--short=8", "HEAD"],
capture_output=True, text=True, timeout=5,
cwd=str(project_root),
@ -66,7 +65,7 @@ def _get_git_commit_date(project_root: Path) -> str:
build).
"""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "log", "-1", "--format=%cd", "--date=short", "HEAD"],
capture_output=True, text=True, timeout=5,
cwd=str(project_root),

View file

@ -12,7 +12,6 @@ import shlex
import shutil
import signal
import subprocess
from hermes_cli._subprocess_compat import windows_hide_flags
import sys
import textwrap
import time
@ -380,24 +379,6 @@ def _scan_gateway_pids(
try:
if is_windows():
try:
import psutil # type: ignore
for proc in psutil.process_iter(["pid", "cmdline"]):
pid = int(proc.info.get("pid") or 0)
if pid == os.getpid() or pid in exclude_pids:
continue
command = " ".join(proc.info.get("cmdline") or [])
if _matches_gateway_runtime(command) and (
all_profiles or _matches_current_profile(command)
):
_append_unique_pid(pids, pid, exclude_pids)
return _filter_venv_launcher_stubs(pids) if len(pids) > 1 else pids
except Exception:
pass
from hermes_cli import _subprocess_compat
# Prefer wmic when present (fast, stable output format). On
# modern Windows 11 / Win 10 late builds, wmic has been
# removed as part of the WMIC deprecation — fall back to
@ -408,7 +389,7 @@ def _scan_gateway_pids(
result = None
if wmic_path is not None:
try:
result = _subprocess_compat.run(
result = subprocess.run(
[
wmic_path,
"process",
@ -439,7 +420,7 @@ def _scan_gateway_pids(
"}"
)
try:
result = _subprocess_compat.run(
result = subprocess.run(
[powershell, "-NoProfile", "-Command", ps_cmd],
capture_output=True,
text=True,
@ -3404,7 +3385,7 @@ def systemd_status(deep: bool = False, system: bool = False, full: bool = False)
]
if full:
log_cmd.append("-l")
subprocess.run(log_cmd, timeout=10, creationflags=windows_hide_flags())
subprocess.run(log_cmd, timeout=10)
# =============================================================================
@ -6650,6 +6631,7 @@ def _gateway_command_inner(args):
# path that can be reaped with the old gateway process. If the
# Windows backend raises, intentionally preserve the existing
# generic failure fallback below.
service_configured = gateway_windows.is_installed()
try:
gateway_windows.restart()
return

View file

@ -39,10 +39,10 @@ import time
from pathlib import Path
from xml.sax.saxutils import escape
from hermes_cli import _subprocess_compat
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.
@ -157,7 +157,7 @@ def _exec_schtasks(args: list[str]) -> tuple[int, str, str]:
if schtasks is None:
return (1, "", "schtasks.exe not found on PATH")
try:
proc = _subprocess_compat.run(
proc = subprocess.run(
[schtasks, *args],
capture_output=True,
text=True,
@ -168,6 +168,10 @@ def _exec_schtasks(args: list[str]) -> tuple[int, str, str]:
encoding=_schtasks_encoding(),
errors="replace",
timeout=_SCHTASKS_TIMEOUT_S,
# 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=windows_hide_flags(),
)
return (proc.returncode, proc.stdout or "", proc.stderr or "")
except subprocess.TimeoutExpired:
@ -1601,17 +1605,7 @@ def stop() -> None:
drained = _drain_gateway_pid(pid, _windows_stop_drain_timeout())
stopped_any = drained
has_service_artifact = (
get_task_script_path().exists()
or get_task_script_path().with_suffix(".vbs").exists()
or get_startup_entry_path().exists()
or _legacy_startup_entry_path().exists()
)
if (
has_service_artifact
and os.getenv("HERMES_NONINTERACTIVE") != "1"
and is_task_registered()
):
if is_task_registered():
code, _out, err = _exec_schtasks(["/End", "/TN", get_task_name()])
# schtasks returns nonzero when the task isn't currently running — don't treat that as an error.
if code == 0:
@ -1679,8 +1673,7 @@ def restart() -> None:
# Give Windows a moment to release the listening port.
time.sleep(1.0)
pid = _spawn_detached()
_report_gateway_start(f"direct spawn (PID {pid})")
start()
if not _wait_for_gateway_ready(timeout_s=15.0):
raise RuntimeError(

View file

@ -90,7 +90,6 @@ from typing import Any, Iterable, Optional
from hermes_cli.sqlite_util import add_column_if_missing as _add_column_if_missing
from toolsets import get_toolset_names
from hermes_cli import _subprocess_compat
_log = logging.getLogger(__name__)
@ -5208,7 +5207,7 @@ def delete_task(conn: sqlite3.Connection, task_id: str) -> bool:
def _git_toplevel(path: Path) -> Optional[Path]:
"""Return the git toplevel containing ``path``, or ``None`` if not in a repo."""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "-C", str(path), "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
@ -5230,7 +5229,7 @@ def _git_toplevel(path: Path) -> Optional[Path]:
def _git_branch_exists(repo_root: Path, branch_name: str) -> bool:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "-C", str(repo_root), "show-ref", "--verify", f"refs/heads/{branch_name}"],
capture_output=True,
text=True,
@ -5244,7 +5243,7 @@ def _git_branch_exists(repo_root: Path, branch_name: str) -> bool:
def _git_common_dir(path: Path) -> Optional[Path]:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "-C", str(path), "rev-parse", "--path-format=absolute", "--git-common-dir"],
capture_output=True,
text=True,
@ -5263,7 +5262,7 @@ def _git_common_dir(path: Path) -> Optional[Path]:
def _git_dir(path: Path) -> Optional[Path]:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "-C", str(path), "rev-parse", "--path-format=absolute", "--git-dir"],
capture_output=True,
text=True,
@ -5282,7 +5281,7 @@ def _git_dir(path: Path) -> Optional[Path]:
def _git_current_branch(path: Path) -> Optional[str]:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "-C", str(path), "branch", "--show-current"],
capture_output=True,
text=True,

View file

@ -258,10 +258,6 @@ import json
import shutil
import stat
import subprocess
from hermes_cli._subprocess_compat import (
windows_detach_popen_kwargs,
windows_hide_flags,
)
from pathlib import Path
from typing import Optional
@ -2109,7 +2105,7 @@ def _launch_tui(
code: Optional[int] = None
try:
try:
code = subprocess.call(argv, cwd=str(cwd), env=env) # windows-footgun: ok — foreground TUI hand-off, console is intentional
code = subprocess.call(argv, cwd=str(cwd), env=env)
except KeyboardInterrupt:
code = 130
@ -2622,7 +2618,6 @@ def cmd_whatsapp(args):
],
cwd=str(bridge_dir),
env=with_hermes_node_path(),
creationflags=windows_hide_flags(),
)
except KeyboardInterrupt:
pass
@ -5294,7 +5289,7 @@ def _redownload_electron_dist(
if mirror:
dl_env["ELECTRON_MIRROR"] = mirror
try:
subprocess.run([node, str(installer)], cwd=str(electron_dir), env=dl_env, check=False, creationflags=windows_hide_flags())
subprocess.run([node, str(installer)], cwd=str(electron_dir), env=dl_env, check=False)
except OSError:
return False
return _electron_dist_ok(project_root)
@ -5414,7 +5409,7 @@ def _desktop_macos_relaunchable_fixup(desktop_dir: Path) -> None:
return
try:
subprocess.run(["xattr", "-cr", str(app)], check=False)
subprocess.run([codesign, "--force", "--deep", "--sign", "-", str(app)], check=False, creationflags=windows_hide_flags())
subprocess.run([codesign, "--force", "--deep", "--sign", "-", str(app)], check=False)
except Exception as exc:
print(f" (warning: macOS relaunch fixup skipped: {exc})")
@ -5479,7 +5474,7 @@ def _desktop_linux_sandbox_fixup(packaged_executable: Path) -> bool:
print("→ Configuring Electron Linux sandbox helper (sudo required)...")
for command in ([sudo, "chown", "root:root", str(sandbox)], [sudo, "chmod", "4755", str(sandbox)]):
if subprocess.run(command, check=False, creationflags=windows_hide_flags()).returncode != 0:
if subprocess.run(command, check=False).returncode != 0:
print(f"✗ Failed to configure Electron's Linux sandbox helper: {sandbox}")
return False
return True
@ -5590,7 +5585,7 @@ def cmd_gui(args: argparse.Namespace):
stopped = _stop_desktop_processes_locking_build(desktop_dir)
if stopped:
print(f" ⚠ Stopped running desktop app to free the build output (pid {', '.join(map(str, stopped))})")
build_result = subprocess.run([npm, "run", build_script], cwd=desktop_dir, env=env, check=False, creationflags=windows_hide_flags())
build_result = subprocess.run([npm, "run", build_script], cwd=desktop_dir, env=env, check=False)
if (
build_result.returncode != 0
and not source_mode
@ -5617,7 +5612,7 @@ def cmd_gui(args: argparse.Namespace):
# The purge can't remove a win-unpacked tree whose Hermes.exe
# is still locked by a running instance; stop it before retry.
_stop_desktop_processes_locking_build(desktop_dir)
build_result = subprocess.run([npm, "run", build_script], cwd=desktop_dir, env=env, check=False, creationflags=windows_hide_flags())
build_result = subprocess.run([npm, "run", build_script], cwd=desktop_dir, env=env, check=False)
if (
build_result.returncode != 0
and not source_mode
@ -5633,7 +5628,7 @@ def cmd_gui(args: argparse.Namespace):
if not _electron_dist_ok(PROJECT_ROOT):
_redownload_electron_dist(PROJECT_ROOT, env, mirror=mirror)
_stop_desktop_processes_locking_build(desktop_dir)
build_result = subprocess.run([npm, "run", build_script], cwd=desktop_dir, env=mirror_env, check=False, creationflags=windows_hide_flags())
build_result = subprocess.run([npm, "run", build_script], cwd=desktop_dir, env=mirror_env, check=False)
if build_result.returncode != 0:
print("✗ Desktop GUI build failed")
print(f" Run manually: cd apps/desktop && npm run {build_script}")
@ -5675,7 +5670,7 @@ def cmd_gui(args: argparse.Namespace):
if source_mode:
print("→ Launching Hermes Desktop from source build...")
launch_result = subprocess.run([npm, "exec", "--", "electron", "."], cwd=desktop_dir, env=env, check=False, creationflags=windows_hide_flags())
launch_result = subprocess.run([npm, "exec", "--", "electron", "."], cwd=desktop_dir, env=env, check=False)
sys.exit(launch_result.returncode)
if packaged_executable is None:
@ -5687,7 +5682,7 @@ def cmd_gui(args: argparse.Namespace):
sys.exit(1)
print(f"→ Launching packaged Hermes Desktop: {packaged_executable}")
launch_result = subprocess.run([str(packaged_executable)], cwd=desktop_dir, env=env, check=False, creationflags=windows_hide_flags())
launch_result = subprocess.run([str(packaged_executable)], cwd=desktop_dir, env=env, check=False)
sys.exit(launch_result.returncode)
@ -5730,8 +5725,6 @@ def _find_stale_dashboard_pids(
try:
if sys.platform == "win32":
from hermes_cli import _subprocess_compat
# wmic may emit text in the system code page (for example cp936
# on zh-CN systems), not UTF-8. In text mode, subprocess output
# decoding depends on Python's configuration (locale-dependent
@ -5739,7 +5732,7 @@ def _find_stale_dashboard_pids(
# here is errors="ignore": it prevents a reader-thread
# UnicodeDecodeError from leaving result.stdout=None and turning
# the later .split() into an AttributeError (#17049).
result = _subprocess_compat.run(
result = subprocess.run(
["wmic", "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"],
capture_output=True,
text=True,
@ -5979,11 +5972,9 @@ def _kill_stale_dashboard_processes(
failed: list[tuple[int, str]] = []
if sys.platform == "win32":
from hermes_cli import _subprocess_compat
for pid in pids:
try:
result = _subprocess_compat.run(
result = subprocess.run(
["taskkill", "/PID", str(pid), "/F"],
capture_output=True,
text=True,
@ -6235,7 +6226,6 @@ def _update_via_zip(args):
[sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"],
cwd=PROJECT_ROOT,
check=True,
creationflags=windows_hide_flags(),
)
_install_python_dependencies_with_optional_fallback(pip_cmd)
@ -6325,7 +6315,6 @@ def _stash_local_changes_if_needed(git_cmd: list[str], cwd: Path) -> Optional[st
git_cmd + ["stash", "push", "--include-untracked", "-m", stash_name],
cwd=cwd,
check=True,
creationflags=windows_hide_flags(),
)
stash_ref = subprocess.run(
git_cmd + ["rev-parse", "--verify", "refs/stash"],
@ -6746,7 +6735,6 @@ def _sync_with_upstream_if_needed(git_cmd: list[str], cwd: Path) -> None:
git_cmd + ["pull", "--ff-only", "upstream", "main"],
cwd=cwd,
check=True,
creationflags=windows_hide_flags(),
)
except subprocess.CalledProcessError:
print(
@ -7018,7 +7006,6 @@ def _run_install_with_heartbeat(
cwd=PROJECT_ROOT,
check=True,
env=env,
creationflags=windows_hide_flags(),
)
finally:
done.set()
@ -7816,7 +7803,6 @@ def _ensure_uv_for_termux(pip_cmd: list[str]) -> str | None:
pip_cmd + ["install", "uv", "--only-binary", ":all:"],
cwd=PROJECT_ROOT,
check=False,
creationflags=windows_hide_flags(),
)
if result.returncode != 0:
return None
@ -8918,7 +8904,7 @@ def _cmd_update_pip(args):
cmd = [sys.executable, "-m", "pip", "install", "--upgrade", "hermes-agent"]
print(f"→ Running: {' '.join(cmd)}")
run_kwargs = {"creationflags": windows_hide_flags()}
run_kwargs = {}
if export_virtualenv:
run_kwargs["env"] = {**os.environ, "VIRTUAL_ENV": sys.prefix}
result = subprocess.run(cmd, **run_kwargs)
@ -9015,7 +9001,7 @@ def _cmd_update_impl(args, gateway_mode: bool):
# On Windows, git can fail with "unable to write loose object file: Invalid argument"
# due to filesystem atomicity issues. Set the recommended workaround.
if sys.platform == "win32" and git_dir.exists():
_subprocess_compat.run(
subprocess.run(
[
"git",
"-c",
@ -9391,7 +9377,6 @@ def _cmd_update_impl(args, gateway_mode: bool):
[sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"],
cwd=PROJECT_ROOT,
check=True,
creationflags=windows_hide_flags(),
)
if _is_termux_env():
install_group = "termux-all"
@ -11483,7 +11468,7 @@ def cmd_dashboard(args):
# re-executing the dashboard for a non-default profile. Use
# subprocess.Popen + sys.exit() on Windows to avoid the crash.
if sys.platform == "win32":
proc = subprocess.Popen(reexec_argv, env=env) # windows-footgun: ok — foreground re-exec, child owns the console
proc = subprocess.Popen(reexec_argv, env=env)
sys.exit(proc.wait())
else:
os.execvpe(sys.executable, reexec_argv, env)

View file

@ -20,7 +20,6 @@ from pathlib import Path
from typing import Optional
from hermes_constants import get_hermes_home
from hermes_cli import _subprocess_compat
logger = logging.getLogger(__name__)
@ -244,7 +243,7 @@ def _install_uv_windows(env: dict[str, str]) -> None:
cmd = (
'irm https://astral.sh/uv/install.ps1 | iex'
)
_subprocess_compat.run(
subprocess.run(
["powershell", "-ExecutionPolicy", "Bypass", "-c", cmd],
env=env,
check=True,

View file

@ -41,7 +41,6 @@ from hermes_cli.config import (
save_env_value,
)
from hermes_cli.cli_output import prompt as _prompt_input
from hermes_cli._subprocess_compat import windows_hide_flags
_MANIFEST_VERSION = 1
@ -365,7 +364,7 @@ def _run_bootstrap(cwd: Path, commands: List[str]) -> None:
"""
for cmd in commands:
print(color(f" $ {cmd}", Colors.DIM))
proc = subprocess.run(cmd, cwd=str(cwd), shell=True, creationflags=windows_hide_flags())
proc = subprocess.run(cmd, cwd=str(cwd), shell=True)
if proc.returncode != 0:
raise CatalogError(
f"bootstrap step failed (exit {proc.returncode}): {cmd}"
@ -400,7 +399,6 @@ def _do_git_install(entry: CatalogEntry) -> Path:
if not is_sha_ref:
proc = subprocess.run(
[git, "clone", "--depth", "1", "--branch", install.ref, install.url, str(dest)],
creationflags=windows_hide_flags(),
)
if proc.returncode == 0:
pass
@ -412,10 +410,10 @@ def _do_git_install(entry: CatalogEntry) -> Path:
is_sha_ref = True # treat the same as a SHA ref from here
if is_sha_ref:
proc = subprocess.run([git, "clone", install.url, str(dest)], creationflags=windows_hide_flags())
proc = subprocess.run([git, "clone", install.url, str(dest)])
if proc.returncode != 0:
raise CatalogError(f"git clone failed for {install.url}")
proc = subprocess.run([git, "-C", str(dest), "checkout", install.ref], creationflags=windows_hide_flags())
proc = subprocess.run([git, "-C", str(dest), "checkout", install.ref])
if proc.returncode != 0:
raise CatalogError(f"git checkout {install.ref} failed")

View file

@ -71,7 +71,6 @@ from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from agent.skill_utils import is_excluded_skill_path
from hermes_cli import _subprocess_compat
# ---------------------------------------------------------------------------
@ -378,7 +377,7 @@ def _git_clone(url: str, dest: Path) -> None:
if re.match(r"^github\.com/[\w.-]+/[\w.-]+/?$", url):
url = f"https://{url.rstrip('/')}"
try:
_subprocess_compat.run(
subprocess.run(
["git", "clone", "--depth", "1", url, str(dest)],
check=True,
capture_output=True,

View file

@ -185,7 +185,7 @@ def relaunch(
# Windows: subprocess + exit, because execvp can't swap to .cmd/.exe shims.
import subprocess
try:
result = subprocess.run(new_argv) # windows-footgun: ok — re-exec replaces the foreground process
result = subprocess.run(new_argv)
sys.exit(result.returncode)
except KeyboardInterrupt:
sys.exit(130)

View file

@ -160,7 +160,6 @@ from hermes_cli.cli_output import ( # noqa: E402
print_warning,
)
from hermes_cli.secret_prompt import masked_secret_prompt # noqa: E402
from hermes_cli._subprocess_compat import windows_hide_flags
def is_interactive_stdin() -> bool:
@ -806,11 +805,11 @@ def _install_neutts_deps() -> bool:
if prompt_yes_no("Install espeak-ng now?", True):
try:
if sys.platform == "darwin":
subprocess.run(["brew", "install", "espeak-ng"], check=True, creationflags=windows_hide_flags())
subprocess.run(["brew", "install", "espeak-ng"], check=True)
elif sys.platform == "win32":
subprocess.run(["choco", "install", "espeak-ng", "-y"], check=True, creationflags=windows_hide_flags())
subprocess.run(["choco", "install", "espeak-ng", "-y"], check=True)
else:
subprocess.run(["sudo", "apt", "install", "-y", "espeak-ng"], check=True, creationflags=windows_hide_flags())
subprocess.run(["sudo", "apt", "install", "-y", "espeak-ng"], check=True)
print_success("espeak-ng installed")
except (subprocess.CalledProcessError, FileNotFoundError) as e:
print_warning(f"Could not install espeak-ng automatically: {e}")
@ -828,7 +827,6 @@ def _install_neutts_deps() -> bool:
subprocess.run(
[sys.executable, "-m", "pip", "install", "-U", "neutts[all]", "--quiet"],
check=True, timeout=300,
creationflags=windows_hide_flags(),
)
print_success("neutts installed successfully")
return True
@ -854,7 +852,6 @@ def _install_kittentts_deps() -> bool:
subprocess.run(
[sys.executable, "-m", "pip", "install", "-U", wheel_url, "soundfile", "--quiet"],
check=True, timeout=300,
creationflags=windows_hide_flags(),
)
print_success("kittentts installed successfully")
return True

View file

@ -228,7 +228,6 @@ def _checklist_toolset_keys(platform: str) -> Set[str]:
# module shares the same data. Kept as dict-of-dicts for backward
# compatibility with existing ``PLATFORMS[key]["label"]`` access patterns.
from hermes_cli.platforms import PLATFORMS as _PLATFORMS_REGISTRY
from hermes_cli._subprocess_compat import windows_hide_flags
PLATFORMS = {
k: {"label": info.label, "default_toolset": info.default_toolset}
@ -887,7 +886,7 @@ def _run_cua_driver_installer(label: str = "Installing", verbose: bool = True) -
# debuggable. Verbose installs (interactive `computer-use install`)
# keep streaming live.
if verbose:
result = subprocess.run(install_cmd, shell=use_shell, timeout=300, env=_cua_driver_env(), creationflags=windows_hide_flags())
result = subprocess.run(install_cmd, shell=use_shell, timeout=300, env=_cua_driver_env())
else:
result = subprocess.run(
install_cmd, shell=use_shell, timeout=300, env=_cua_driver_env(),

View file

@ -2481,52 +2481,16 @@ 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_details() -> Tuple[str, Dict[str, str]]:
"""Return (executable, env overlay) for detached dashboard actions.
On Windows this mirrors the gateway's uv-safe detached launcher logic so
action spawns do not regress to console python.exe (which creates a visible
terminal window). Non-Windows callsites get the current interpreter and no
env overlay.
"""
def _dashboard_spawn_executable() -> str:
"""Prefer pythonw.exe for detached dashboard actions on Windows."""
if sys.platform != "win32":
return sys.executable, {}
return sys.executable
exe = sys.executable
try:
from hermes_cli.gateway_windows import _resolve_detached_python
venv_root = os.environ.get("VIRTUAL_ENV", "").strip()
if not venv_root:
for candidate in (PROJECT_ROOT / "venv", PROJECT_ROOT / ".venv"):
if (candidate / "Scripts" / "python.exe").exists():
venv_root = str(candidate)
break
probe_exe = (
os.path.join(venv_root, "Scripts", "python.exe")
if venv_root
else exe
)
windowless_exe, venv_dir, extra_pythonpath = _resolve_detached_python(probe_exe)
env_overlay: Dict[str, str] = {}
if venv_dir:
env_overlay["VIRTUAL_ENV"] = str(venv_dir)
site_packages = Path(venv_dir) / "Lib" / "site-packages"
if site_packages.exists() and str(site_packages) not in extra_pythonpath:
extra_pythonpath = [*extra_pythonpath, str(site_packages)]
if extra_pythonpath:
existing = os.environ.get("PYTHONPATH", "")
env_overlay["PYTHONPATH"] = os.pathsep.join(
[*extra_pythonpath, existing] if existing else list(extra_pythonpath)
)
return windowless_exe, env_overlay
except Exception:
# Best-effort fallback: sibling pythonw keeps the legacy no-console path.
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, {}
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:
@ -2543,20 +2507,15 @@ 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()
)
spawn_executable, spawn_env_overlay = _dashboard_spawn_details()
cmd = [spawn_executable, "-m", "hermes_cli.main", *subcommand]
cmd = [_dashboard_spawn_executable(), "-m", "hermes_cli.main", *subcommand]
popen_kwargs: Dict[str, Any] = {
"cwd": str(PROJECT_ROOT),
"stdin": subprocess.DEVNULL,
"stdout": log_file,
"stderr": subprocess.STDOUT,
"env": {**os.environ, "HERMES_NONINTERACTIVE": "1", **spawn_env_overlay},
"env": {**os.environ, "HERMES_NONINTERACTIVE": "1"},
}
log_file.write(f"spawn executable: {spawn_executable}\n".encode())
if spawn_env_overlay:
keys = ",".join(sorted(spawn_env_overlay.keys()))
log_file.write(f"spawn env overlay keys: {keys}\n".encode())
if sys.platform == "win32":
popen_kwargs["creationflags"] = windows_detach_flags()
else:
@ -2814,7 +2773,7 @@ def _recent_upstream_commits(n: int = 20) -> List[Dict[str, Any]]:
or git is unavailable. Never raises into the request path.
"""
try:
out = _subprocess_compat.run(
out = subprocess.run(
[
"git",
"-C",
@ -10574,7 +10533,7 @@ async def open_profile_terminal_endpoint(name: str):
command = _profile_setup_command(name)
if sys.platform.startswith("win"):
subprocess.Popen(["cmd.exe", "/c", "start", "", command]) # windows-footgun: ok — open terminal for user (Windows branch)
subprocess.Popen(["cmd.exe", "/c", "start", "", command])
elif sys.platform == "darwin":
escaped = command.replace("\\", "\\\\").replace('"', '\\"')
applescript = (
@ -10583,7 +10542,7 @@ async def open_profile_terminal_endpoint(name: str):
f'do script "{escaped}"\n'
"end tell"
)
subprocess.Popen(["osascript", "-e", applescript]) # windows-footgun: ok — open Terminal.app (macOS, visible by design)
subprocess.Popen(["osascript", "-e", applescript])
else:
terminal_commands = [
("x-terminal-emulator", ["x-terminal-emulator", "-e", "sh", "-lc", command]),
@ -10603,7 +10562,7 @@ async def open_profile_terminal_endpoint(name: str):
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
) == 0:
subprocess.Popen(popen_args) # windows-footgun: ok — open OS terminal for user
subprocess.Popen(popen_args)
break
else:
raise HTTPException(
@ -13474,7 +13433,6 @@ _mount_plugin_api_routes()
# always mounted — the gate middleware decides whether to enforce auth,
# not whether the routes exist.
from hermes_cli.dashboard_auth.routes import router as _dashboard_auth_router # noqa: E402
from hermes_cli import _subprocess_compat
app.include_router(_dashboard_auth_router)
mount_spa(app)
@ -13589,15 +13547,6 @@ def start_server(
used when a profile alias (``<profile> dashboard``) routes to the
machine dashboard.
"""
# Desktop spawns this backend via a no-console venv python; a uv
# pythonw→python re-exec can still auto-allocate a console. Drop it.
# No-op on POSIX and when launched from an interactive shell.
try:
import hermes_bootstrap
hermes_bootstrap.detach_orphan_console()
except Exception:
pass
import uvicorn
try:

View file

@ -561,15 +561,12 @@ def agent_browser_runnable(path: str | None) -> bool:
return False
import subprocess
from hermes_cli import _subprocess_compat
try:
result = _subprocess_compat.run(
result = subprocess.run(
[path, "--version"],
capture_output=True,
timeout=10,
env=with_hermes_node_path(),
stdin=subprocess.DEVNULL,
)
except (OSError, subprocess.TimeoutExpired, ValueError):
return False

View file

@ -24,7 +24,6 @@ from hermes_constants import get_hermes_home
from hermes_cli.profiles import _get_default_hermes_home
from plugins.plugin_utils import SingletonSlot
from typing import Any, TYPE_CHECKING
from hermes_cli import _subprocess_compat
if TYPE_CHECKING:
from honcho import Honcho
@ -626,7 +625,7 @@ class HonchoClientConfig:
import subprocess
try:
root = _subprocess_compat.run(
root = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
capture_output=True, text=True, cwd=cwd, timeout=5,
stdin=subprocess.DEVNULL,

View file

@ -22,7 +22,6 @@ from ._oss_providers import (
KNOWN_DIMS,
validate_oss_config,
)
from hermes_cli import _subprocess_compat
def _curses_select(title: str, items: list[tuple[str, str]], default: int = 0) -> int:
@ -406,13 +405,13 @@ def _ensure_pgvector(host: str = "localhost", port: int = 5432) -> dict | None:
# Check if our container already exists but is stopped
if shutil.which("docker"):
try:
result = _subprocess_compat.run(
result = subprocess.run(
["docker", "inspect", _PGVECTOR_CONTAINER, "--format", "{{.State.Status}}"],
capture_output=True, text=True, timeout=10, stdin=subprocess.DEVNULL,
)
if result.returncode == 0 and "exited" in result.stdout:
print(f" Found stopped container '{_PGVECTOR_CONTAINER}', restarting...")
_subprocess_compat.run(["docker", "start", _PGVECTOR_CONTAINER],
subprocess.run(["docker", "start", _PGVECTOR_CONTAINER],
capture_output=True, timeout=15,
stdin=subprocess.DEVNULL)
_wait_for_port(host, port, timeout=15)
@ -439,17 +438,17 @@ def _start_pgvector_docker(host: str, port: int) -> dict | None:
"""Pull and start pgvector Docker container."""
try:
print(f" Pulling {_PGVECTOR_IMAGE}...")
_subprocess_compat.run(["docker", "pull", _PGVECTOR_IMAGE],
subprocess.run(["docker", "pull", _PGVECTOR_IMAGE],
capture_output=True, timeout=120,
stdin=subprocess.DEVNULL)
# Remove existing container if present
_subprocess_compat.run(["docker", "rm", "-f", _PGVECTOR_CONTAINER],
subprocess.run(["docker", "rm", "-f", _PGVECTOR_CONTAINER],
capture_output=True, timeout=10,
stdin=subprocess.DEVNULL)
print(f" Starting container '{_PGVECTOR_CONTAINER}' on port {port}...")
_subprocess_compat.run([
subprocess.run([
"docker", "run", "-d",
"--name", _PGVECTOR_CONTAINER,
"-e", f"POSTGRES_PASSWORD={_PGVECTOR_PASSWORD}",
@ -523,8 +522,7 @@ def _ensure_ollama(models: list[str]) -> bool:
print(f" Pulling '{model}'... (this may take a few minutes)")
try:
subprocess.run([ollama_bin or "ollama", "pull", model], timeout=600,
stdin=subprocess.DEVNULL,
creationflags=windows_hide_flags())
stdin=subprocess.DEVNULL)
print(f" ✓ Model '{model}' pulled")
except Exception as e:
print(f" Warning: Could not pull '{model}': {e}")
@ -735,7 +733,7 @@ def _install_provider_deps(llm_id: str, embedder_id: str, vector_id: str) -> Non
for dep in sorted(deps):
try:
print(f" Installing {dep}...")
_subprocess_compat.run(
subprocess.run(
["uv", "pip", "install", "--python", sys.executable, dep],
capture_output=True, timeout=60,
)

View file

@ -665,9 +665,7 @@ class VoiceReceiver:
f.write(pcm_data)
pcm_path = f.name
try:
from hermes_cli import _subprocess_compat
_subprocess_compat.run(
subprocess.run(
[
"ffmpeg", "-y", "-loglevel", "error",
"-f", "s16le",

View file

@ -45,7 +45,6 @@ the mixer's output cannot echo back into transcription.
import logging
import threading
from typing import TYPE_CHECKING, List, Optional
from hermes_cli import _subprocess_compat
if TYPE_CHECKING: # numpy is an optional ("voice" extra) dep — never import at runtime top-level
import numpy as np
@ -310,7 +309,7 @@ def decode_to_pcm(path: str, *, timeout: float = 30.0) -> Optional[bytes]:
import subprocess
try:
proc = _subprocess_compat.run(
proc = subprocess.run(
[
"ffmpeg", "-y", "-loglevel", "error",
"-i", path,

View file

@ -387,7 +387,6 @@ def _install_sidecar() -> int:
[npm, "ci"],
cwd=str(_SIDECAR_DIR),
check=False,
creationflags=windows_hide_flags(),
)
if proc.returncode != 0:
print(f" npm ci failed — falling back to: {npm} install")
@ -395,7 +394,6 @@ def _install_sidecar() -> int:
[npm, "install"],
cwd=str(_SIDECAR_DIR),
check=False,
creationflags=windows_hide_flags(),
)
if proc.returncode != 0:
print("npm install failed", file=sys.stderr)

View file

@ -20,7 +20,6 @@ import re
import secrets
import shutil
import subprocess
from hermes_cli._subprocess_compat import windows_detach_popen_kwargs
import threading
import time
import uuid
@ -543,7 +542,7 @@ class RaftAdapter(BasePlatformAdapter):
env = {**os.environ, "RAFT_CHANNEL_TOKEN": self._bridge_token}
try:
self._bridge_process = subprocess.Popen(
cmd, env=env, stdin=subprocess.DEVNULL, **windows_detach_popen_kwargs()
cmd, env=env, stdin=subprocess.DEVNULL
)
logger.info("[raft] Spawned bridge pid=%d profile=%s endpoint=%s", self._bridge_process.pid, profile, endpoint)
except Exception:

View file

@ -78,10 +78,8 @@ def _kill_port_process(port: int) -> None:
"""Kill any process *listening* on the given TCP port (a stale bridge)."""
try:
if _IS_WINDOWS:
from hermes_cli import _subprocess_compat
# Use netstat to find the PID bound to this port, then taskkill
result = _subprocess_compat.run(
result = subprocess.run(
["netstat", "-ano", "-p", "TCP"],
capture_output=True, text=True, timeout=5,
)
@ -91,7 +89,7 @@ def _kill_port_process(port: int) -> None:
local_addr = parts[1]
if local_addr.endswith(f":{port}"):
try:
_subprocess_compat.run(
subprocess.run(
["taskkill", "/PID", parts[4], "/F"],
capture_output=True, timeout=5,
)
@ -209,13 +207,11 @@ def _write_bridge_pidfile(session_path: Path, pid: int) -> None:
def _terminate_bridge_process(proc, *, force: bool = False) -> None:
"""Terminate the bridge process using process-tree semantics where possible."""
if _IS_WINDOWS:
from hermes_cli import _subprocess_compat
cmd = ["taskkill", "/PID", str(proc.pid), "/T"]
if force:
cmd.append("/F")
try:
result = _subprocess_compat.run(
result = subprocess.run(
cmd,
capture_output=True,
text=True,

View file

@ -29,7 +29,6 @@ Suppress an intentional use (e.g. tests or platform-gated code) with:
from __future__ import annotations
import argparse
import ast
import os
import re
import subprocess
@ -328,260 +327,6 @@ FOOTGUNS: list[Footgun] = [
]
# -----------------------------------------------------------------------------
# AST-based rule: subprocess calls that flash a console window on Windows
# -----------------------------------------------------------------------------
#
# This is the high-volume Windows complaint: every `subprocess.run(...)` /
# `subprocess.Popen(...)` of a console program on Windows briefly flashes a
# cmd window unless the child either (a) inherits the parent's stdio handles
# via output redirection, or (b) is spawned with a no-window creationflag
# (CREATE_NO_WINDOW / DETACHED_PROCESS). The fix landscape already exists in
# `hermes_cli/_subprocess_compat.py` (windows_hide_flags / windows_detach_*),
# but nothing stopped new bare calls from re-introducing the popup — so the
# bug kept coming back PR after PR. This rule is the chokepoint.
#
# It is AST-based (not regex) because the deciding factor — whether the call
# redirects stdout/stderr — frequently lives several lines below the
# `subprocess.run(` opener, which a line-oriented regex cannot see.
#
# Comprehensive, not restrictive: a call is only flagged when it can ACTUALLY
# create a new console. Calls that capture or redirect output (capture_output=,
# stdout=, stderr=), or use check_output (which always captures), cannot pop a
# window and are silently ignored — no suppression comment needed. The intent
# is that the overwhelming majority of subprocess calls require no change at
# all; only the genuine window-spawners do.
# The subprocess functions that can spawn a child process.
_SUBPROCESS_FUNCS = frozenset({"run", "Popen", "call", "check_call", "check_output"})
# Module aliases we recognise as the stdlib subprocess module.
_SUBPROCESS_ALIASES = frozenset({"subprocess", "sp"})
# Executables that simply do not exist on Windows. A subprocess call whose
# program is one of these can never create a Windows console window, so the
# no-window flag is irrelevant — flagging them would force pointless
# suppression comments on macOS/Linux-only service-management and packaging
# code (launchctl, systemctl, brew, codesign …). Matched against the FIRST
# element of a list/tuple argv literal only; anything dynamic still gets
# flagged (we can't prove it's POSIX-only).
_POSIX_ONLY_PROGRAMS = frozenset(
{
"launchctl",
"systemctl",
"journalctl",
"loginctl",
"osascript",
"codesign",
"xattr",
"defaults",
"brew",
"apt",
"apt-get",
"dpkg",
"pacman",
"dnf",
"yum",
"sudo",
"open", # macOS `open`
"tail",
"sw_vers",
"scutil",
"diskutil",
"hdiutil",
"dscl",
}
)
# Cross-platform console programs that DO exist on Windows and allocate a
# console window when spawned from a console-less parent (Desktop/Electron,
# pythonw.exe, a detached gateway/cron). For these, capturing or redirecting
# stdio is NOT a safety boundary — stream redirection controls where the
# child's output goes, it does NOT suppress console *allocation*. Only
# CREATE_NO_WINDOW (or routing through hermes_cli._subprocess_compat.run/popen,
# which injects it) prevents the flash. So a call to one of these is flagged
# even with capture_output=/stdout=/stderr= set. Matched against the first
# element of a literal argv (bare name or .exe, path-stripped).
_WINDOWS_FLASHING_PROGRAMS = frozenset(
{
"git",
"gh",
"node",
"npm",
"npx",
"yarn",
"pnpm",
"python",
"python3",
"pythonw",
"pip",
"uv",
"uvx",
"ffmpeg",
"ffprobe",
"ollama",
"docker",
"cmd",
"cmd.exe",
"powershell",
"powershell.exe",
"pwsh",
"where",
"taskkill",
"schtasks",
"wmic",
"tasklist",
"netstat",
}
)
SUBPROCESS_FOOTGUN_NAME = "subprocess without Windows no-window flag"
SUBPROCESS_FOOTGUN_MESSAGE = (
"subprocess.run/Popen/call on Windows flashes a console (cmd) window "
"unless the child inherits stdio (output is captured/redirected) or is "
"spawned with a no-window creationflag. This is the #1 source of Windows "
"'terminal popup' bug reports."
)
SUBPROCESS_FOOTGUN_FIX = (
"Pass creationflags=windows_hide_flags() (for short-lived/captured spawns) "
"or **windows_detach_popen_kwargs() (for detached daemons) from "
"hermes_cli._subprocess_compat (both no-op on POSIX). If a visible window "
"is intended (interactive launch, shell hand-off), add "
"'# windows-footgun: ok' on the call line."
)
def _call_attr_name(node: ast.Call) -> str | None:
"""Return 'run'/'Popen'/... when node is subprocess.<func>(...), else None."""
f = node.func
if not isinstance(f, ast.Attribute):
return None
if f.attr not in _SUBPROCESS_FUNCS:
return None
mod = getattr(f.value, "id", None)
if mod not in _SUBPROCESS_ALIASES:
return None
return f.attr
def _suppresses_window(node: ast.Call, func_name: str) -> bool:
"""True if this subprocess call cannot create a new console window.
The honest invariant (corrected after review of PR #53791): capturing or
redirecting stdio is NOT the same as suppressing console allocation. From a
console-less parent (Desktop/Electron, pythonw.exe, a detached gateway/cron)
a console-subsystem child still allocates and flashes a window even with
capture_output=True. Only CREATE_NO_WINDOW (or routing through
hermes_cli._subprocess_compat.run/popen, which injects it) prevents it.
So capture/stdout/stderr/check_output is treated as window-safe ONLY when the
program is not a known cross-platform console exe that flashes on Windows
(see _WINDOWS_FLASHING_PROGRAMS git/gh/npm/node/python/uv/ffmpeg/docker/).
For those, even a fully-captured call is flagged.
Always window-safe regardless of program:
* creationflags=... author is already managing the console
* **<spread> kwargs may carry a _subprocess_compat helper;
flag-via-spread is the recommended fix, so we
must not penalise it.
* POSIX-only program can't run on Windows, can't flash.
Conditionally safe (only when NOT a known flashing program):
* check_output / capture_output= / stdout= / stderr=
"""
explicit = {kw.arg for kw in node.keywords if kw.arg}
if "creationflags" in explicit:
return True
if any(kw.arg is None for kw in node.keywords): # **kwargs spread
return True
if _is_posix_only_program(node):
return True
# Capture/redirect is only a safety boundary for programs that don't
# allocate a Windows console — NOT for git/npm/node/python/ffmpeg/etc.
if not _is_windows_flashing_program(node):
if func_name == "check_output":
return True
if explicit & {"stdout", "stderr", "capture_output"}:
return True
return False
def _argv_head(node: ast.Call) -> str | None:
"""Return the path-stripped first argv element if it's a string literal."""
if not node.args:
return None
first = node.args[0]
if isinstance(first, (ast.List, ast.Tuple)) and first.elts:
head = first.elts[0]
if isinstance(head, ast.Constant) and isinstance(head.value, str):
return head.value.rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
return None
def _is_windows_flashing_program(node: ast.Call) -> bool:
"""True if the call's program is a known cross-platform console exe that
allocates a Windows console window (so capture is NOT a safe boundary)."""
prog = _argv_head(node)
return prog is not None and prog in _WINDOWS_FLASHING_PROGRAMS
def _is_posix_only_program(node: ast.Call) -> bool:
"""True if the call's program is a statically-known POSIX-only executable.
Only inspects a literal list/tuple first arg whose first element is a
string constant (e.g. ``["launchctl", "bootout", target]``). Dynamic
argv (variables, f-strings) is treated as unknown and still flagged.
"""
if not node.args:
return False
first = node.args[0]
if isinstance(first, (ast.List, ast.Tuple)) and first.elts:
head = first.elts[0]
if isinstance(head, ast.Constant) and isinstance(head.value, str):
prog = head.value.rsplit("/", 1)[-1]
return prog in _POSIX_ONLY_PROGRAMS
return False
def scan_subprocess_window_footguns(
path: Path, text: str
) -> list[tuple[int, str, Footgun]]:
"""AST pass: flag subprocess calls that can flash a Windows console.
Honours the same `# windows-footgun: ok` line suppression as the regex
rules. Returns the same (lineno, line, Footgun) shape so results merge
cleanly into scan_file's output.
"""
try:
tree = ast.parse(text)
except SyntaxError:
return []
lines = text.splitlines()
rule = Footgun(
name=SUBPROCESS_FOOTGUN_NAME,
pattern=re.compile(r"^$"), # unused; AST-driven
message=SUBPROCESS_FOOTGUN_MESSAGE,
fix=SUBPROCESS_FOOTGUN_FIX,
)
out: list[tuple[int, str, Footgun]] = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func_name = _call_attr_name(node)
if func_name is None:
continue
if _suppresses_window(node, func_name):
continue
lineno = node.lineno
line = lines[lineno - 1] if 0 <= lineno - 1 < len(lines) else ""
# Inline suppression — check the opener line AND, for multi-line calls,
# any line in the call's span (a developer may mark the closing paren).
end = getattr(node, "end_lineno", lineno) or lineno
span = lines[lineno - 1 : end]
if any(SUPPRESS_MARKER.search(l) for l in span):
continue
out.append((lineno, line.rstrip(), rule))
return out
def should_scan_file(path: Path) -> bool:
"""Return True if this file is in scope for the checker."""
# Skip the excluded dirs
@ -671,11 +416,6 @@ def scan_file(path: Path, footguns: list[Footgun]) -> list[tuple[int, str, Footg
return []
matches: list[tuple[int, str, Footgun]] = []
# AST-based rule (subprocess console-window footgun). Runs only on Python
# source; merges into the same result list as the regex rules below.
if path.suffix in {".py", ".pyw", ".pyi"}:
matches.extend(scan_subprocess_window_footguns(path, text))
# Track whether we're inside a triple-quoted string (docstring/raw block).
# Simple state machine — handles both ''' and """, toggled by the FIRST
# triple-quote we see; we don't try to handle nested or f-string cases.
@ -748,7 +488,7 @@ def scan_file(path: Path, footguns: list[Footgun]) -> list[tuple[int, str, Footg
def get_staged_files() -> list[Path]:
"""Return paths staged in the current git index. Empty on non-git trees."""
try:
out = subprocess.check_output( # windows-footgun: ok — dev-only checker, runs on Linux CI
out = subprocess.check_output(
["git", "diff", "--cached", "--name-only", "--diff-filter=ACMR"],
cwd=REPO_ROOT,
stderr=subprocess.DEVNULL,
@ -762,7 +502,7 @@ def get_staged_files() -> list[Path]:
def get_diff_files(ref: str) -> list[Path]:
"""Return paths modified vs. the given git ref."""
try:
out = subprocess.check_output( # windows-footgun: ok — dev-only checker, runs on Linux CI
out = subprocess.check_output(
["git", "diff", f"{ref}...HEAD", "--name-only", "--diff-filter=ACMR"],
cwd=REPO_ROOT,
stderr=subprocess.DEVNULL,
@ -808,12 +548,6 @@ def print_rules() -> None:
print(f" {fg.message}")
print(f" Fix: {fg.fix}")
print()
# AST-based rule (not in the regex FOOTGUNS list).
n = len(FOOTGUNS) + 1
print(f"{n:2}. {SUBPROCESS_FOOTGUN_NAME} (AST-based)")
print(f" {SUBPROCESS_FOOTGUN_MESSAGE}")
print(f" Fix: {SUBPROCESS_FOOTGUN_FIX}")
print()
def main(argv: list[str]) -> int:

View file

@ -97,7 +97,7 @@ def gh_pr_list():
Returns an empty list if gh is not available or the call fails.
"""
try:
result = subprocess.run( # windows-footgun: ok — dev-only contributor-audit script
result = subprocess.run(
[
"gh", "pr", "list",
"--repo", "NousResearch/hermes-agent",

View file

@ -42,7 +42,6 @@ from hermes_cli.psutil_android import (
PsutilAndroidInstallError,
prepare_patched_psutil_sdist,
)
from hermes_cli._subprocess_compat import windows_hide_flags
@ -91,7 +90,7 @@ def main() -> int:
cmd = install_cmd_prefix + ["install", "--no-build-isolation", str(src_root)]
print(f" $ {' '.join(cmd)}")
result = subprocess.run(cmd, creationflags=windows_hide_flags())
result = subprocess.run(cmd)
if result.returncode != 0:
return result.returncode

View file

@ -568,7 +568,7 @@ def loop_mode(args: argparse.Namespace) -> int:
if iteration > 1:
print("• rebuilding…")
result = subprocess.run( # windows-footgun: ok — dev-only TUI build script
result = subprocess.run(
["npm", "run", "build"],
cwd=tui_dir,
capture_output=True,

View file

@ -618,25 +618,20 @@ class TestGetProcessStartTime:
class TestTerminatePid:
def test_force_uses_taskkill_on_windows(self, monkeypatch):
from hermes_cli import _subprocess_compat
calls = []
monkeypatch.setattr(status, "_IS_WINDOWS", True)
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", True)
def fake_run(cmd, capture_output=False, text=False, timeout=None, **kwargs):
calls.append((cmd, capture_output, text, timeout, kwargs))
def fake_run(cmd, capture_output=False, text=False, timeout=None):
calls.append((cmd, capture_output, text, timeout))
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(status.subprocess, "run", fake_run)
status.terminate_pid(123, force=True)
assert len(calls) == 1
cmd, capture_output, text, timeout, kwargs = calls[0]
assert cmd == ["taskkill", "/PID", "123", "/T", "/F"]
assert (capture_output, text, timeout) == (True, True, 10)
assert kwargs["creationflags"] & 0x08000000
assert calls == [
(["taskkill", "/PID", "123", "/T", "/F"], True, True, 10)
]
def test_force_falls_back_to_sigterm_when_taskkill_missing(self, monkeypatch):
calls = []

View file

@ -725,9 +725,8 @@ class TestDetectOpenclawProcesses:
def test_returns_match_on_windows_when_openclaw_exe_running(self):
with patch.object(claw_mod, "sys") as mock_sys:
mock_sys.platform = "win32"
# Windows scans go through the hidden-spawn primitive (no console flash).
with patch("hermes_cli._subprocess_compat.run") as mock_run:
mock_run.side_effect = [
with patch.object(claw_mod, "subprocess") as mock_subprocess:
mock_subprocess.run.side_effect = [
MagicMock(returncode=0, stdout="openclaw.exe 1234 Console 1 45,056 K\n"),
]
result = claw_mod._detect_openclaw_processes()
@ -737,8 +736,8 @@ class TestDetectOpenclawProcesses:
def test_returns_match_on_windows_when_node_exe_has_openclaw_in_cmdline(self):
with patch.object(claw_mod, "sys") as mock_sys:
mock_sys.platform = "win32"
with patch("hermes_cli._subprocess_compat.run") as mock_run:
mock_run.side_effect = [
with patch.object(claw_mod, "subprocess") as mock_subprocess:
mock_subprocess.run.side_effect = [
MagicMock(returncode=0, stdout=""), # tasklist openclaw.exe
MagicMock(returncode=0, stdout=""), # tasklist clawd.exe
MagicMock(returncode=0, stdout="1234\n"), # PowerShell
@ -750,8 +749,8 @@ class TestDetectOpenclawProcesses:
def test_returns_empty_on_windows_when_nothing_found(self):
with patch.object(claw_mod, "sys") as mock_sys:
mock_sys.platform = "win32"
with patch("hermes_cli._subprocess_compat.run") as mock_run:
mock_run.side_effect = [
with patch.object(claw_mod, "subprocess") as mock_subprocess:
mock_subprocess.run.side_effect = [
MagicMock(returncode=0, stdout=""),
MagicMock(returncode=0, stdout=""),
MagicMock(returncode=0, stdout=""),

View file

@ -887,20 +887,23 @@ def test_reap_unsupervised_orphans_returns_false_when_none_found(monkeypatch):
def test_scan_gateway_pids_detects_windows_hermes_exe_case_variants(monkeypatch):
# Windows scan now goes through psutil first (no console spawn). A
# uppercase ``Hermes.EXE gateway run`` must still match case-insensitively.
import psutil
monkeypatch.setattr(gateway, "is_windows", lambda: True)
monkeypatch.setattr(gateway, "_get_ancestor_pids", lambda: set())
monkeypatch.setattr(gateway.shutil, "which", lambda name: "wmic.exe" if name == "wmic" else None)
proc = SimpleNamespace(
info={
"pid": 2468,
"cmdline": ["C:\\Program Files\\Hermes\\Hermes.EXE", "gateway", "run", "--replace"],
}
)
monkeypatch.setattr(psutil, "process_iter", lambda attrs=None: [proc])
def fake_run(cmd, **kwargs):
if cmd[:4] == ["wmic.exe", "process", "get", "ProcessId,CommandLine"]:
return SimpleNamespace(
returncode=0,
stdout=(
"CommandLine=C:\\Program Files\\Hermes\\Hermes.EXE gateway run --replace\n"
"ProcessId=2468\n\n"
),
stderr="",
)
raise AssertionError(f"Unexpected command: {cmd}")
monkeypatch.setattr(gateway.subprocess, "run", fake_run)
assert gateway._scan_gateway_pids(set(), all_profiles=True) == [2468]

View file

@ -29,31 +29,6 @@ def test_schtasks_fallback_does_not_hide_unknown_errors():
assert gateway_windows._should_fall_back(1, "ERROR: The system cannot find the file specified.") is False
def test_noninteractive_stop_skips_schtasks_query(monkeypatch, tmp_path):
"""Desktop-triggered restarts must not invoke schtasks.exe.
schtasks is a console-subsystem binary; on Windows Terminal default hosts it
can visibly pop a terminal even for `/Query`. Noninteractive desktop actions
already stop the known gateway PID directly, so service-manager probing is
unnecessary.
"""
script = tmp_path / "Hermes_Gateway.cmd"
script.write_text("", encoding="utf-8")
monkeypatch.setenv("HERMES_NONINTERACTIVE", "1")
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "get_task_script_path", lambda: script)
monkeypatch.setattr(gateway_windows, "get_startup_entry_path", lambda: tmp_path / "Hermes_Gateway.vbs")
monkeypatch.setattr(gateway_windows, "_legacy_startup_entry_path", lambda: tmp_path / "Hermes_Gateway.cmd")
monkeypatch.setattr(gateway_windows, "is_task_registered", lambda: pytest.fail("must not call schtasks"))
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
monkeypatch.setattr(gateway_windows, "_collect_gateway_stop_pids", lambda pid=None: [])
monkeypatch.setattr(gateway_windows, "_force_terminate_known_gateway_pids", lambda pids: 0)
gateway_windows.stop()
def test_schtasks_encoding_falls_back_to_utf8(monkeypatch):
"""A broken/empty locale must not leave us without a decoder (issue #38172)."""
@ -137,23 +112,6 @@ def test_build_gateway_argv_uses_base_pythonw_for_uv_venv_launcher(monkeypatch,
assert str(site_packages) in env_overlay["PYTHONPATH"].split(gateway_windows.os.pathsep)
def test_restart_relaunches_directly_without_start_service_probe(monkeypatch):
calls = []
monkeypatch.setattr(gateway_windows, "_assert_windows", lambda: None)
monkeypatch.setattr(gateway_windows, "stop", lambda: calls.append("stop"))
monkeypatch.setattr(gateway_windows, "_wait_for_gateway_absent", lambda *a, **k: True)
monkeypatch.setattr(gateway_windows.time, "sleep", lambda *_a, **_k: None)
monkeypatch.setattr(gateway_windows, "_spawn_detached", lambda: 4321)
monkeypatch.setattr(gateway_windows, "_report_gateway_start", lambda via: calls.append(("report", via)))
monkeypatch.setattr(gateway_windows, "_wait_for_gateway_ready", lambda *a, **k: [4321])
monkeypatch.setattr(gateway_windows, "start", lambda: pytest.fail("restart must not call start()"))
gateway_windows.restart()
assert calls == ["stop", ("report", "direct spawn (PID 4321)")]
class TestStableWindowsGatewayWorkingDir:
def test_stable_gateway_working_dir_uses_hermes_home(self, tmp_path, monkeypatch):
home = tmp_path / ".hermes"

View file

@ -109,7 +109,7 @@ def test_install_psutil_android_script_uses_patched_tree(tmp_path, monkeypatch,
shutil.copyfile(archive, dest)
return str(dest), None
def fake_subprocess_run(cmd: list[str], **kwargs):
def fake_subprocess_run(cmd: list[str]):
src_root = Path(cmd[-1])
patched = (src_root / "psutil" / "_common.py").read_text(encoding="utf-8")
assert REPLACEMENT in patched

View file

@ -215,75 +215,6 @@ class TestSessionTokenInjection:
assert ws._SESSION_TOKEN and len(ws._SESSION_TOKEN) >= 32
class TestDashboardActionSpawnDetails:
def test_windows_uses_uv_safe_pythonw_with_env_overlay(self, monkeypatch):
import hermes_cli.web_server as ws
monkeypatch.setattr(ws.sys, "platform", "win32")
monkeypatch.setattr(ws.sys, "executable", r"C:\venv\Scripts\python.exe")
monkeypatch.setenv("PYTHONPATH", r"C:\existing")
monkeypatch.setattr(
"hermes_cli.gateway_windows._resolve_detached_python",
lambda _exe: (
r"C:\base\pythonw.exe",
Path(r"C:\venv"),
[r"C:\venv\Lib\site-packages"],
),
)
executable, env_overlay = ws._dashboard_spawn_details()
assert executable == r"C:\base\pythonw.exe"
assert env_overlay["VIRTUAL_ENV"] == str(Path(r"C:\venv"))
assert env_overlay["PYTHONPATH"] == os.pathsep.join(
[r"C:\venv\Lib\site-packages", r"C:\existing"]
)
def test_windows_falls_back_to_sibling_pythonw_when_resolver_fails(self, monkeypatch):
import hermes_cli.web_server as ws
exe = "C:/venv/Scripts/python.exe"
expected = "C:/venv/Scripts/pythonw.exe"
monkeypatch.setattr(ws.sys, "platform", "win32")
monkeypatch.setattr(ws.sys, "executable", exe)
monkeypatch.setattr(
"hermes_cli.gateway_windows._resolve_detached_python",
lambda _exe: (_ for _ in ()).throw(RuntimeError("boom")),
)
monkeypatch.setattr(ws.os.path, "isfile", lambda candidate: candidate == expected)
executable, env_overlay = ws._dashboard_spawn_details()
assert executable == expected
assert env_overlay == {}
def test_windows_resolves_venv_from_project_when_virtual_env_missing(self, monkeypatch, tmp_path):
import hermes_cli.web_server as ws
project = tmp_path / "project"
scripts = project / "venv" / "Scripts"
site_packages = project / "venv" / "Lib" / "site-packages"
scripts.mkdir(parents=True)
site_packages.mkdir(parents=True)
(scripts / "python.exe").write_text("", encoding="utf-8")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(ws.sys, "platform", "win32")
monkeypatch.setattr(ws.sys, "executable", r"C:\base\pythonw.exe")
monkeypatch.setattr(ws, "PROJECT_ROOT", project)
monkeypatch.setattr(
"hermes_cli.gateway_windows._resolve_detached_python",
lambda exe: (r"C:\base\pythonw.exe", project / "venv", []),
)
executable, env_overlay = ws._dashboard_spawn_details()
assert executable == r"C:\base\pythonw.exe"
assert env_overlay["VIRTUAL_ENV"] == str(project / "venv")
assert str(site_packages) in env_overlay["PYTHONPATH"].split(os.pathsep)
# ---------------------------------------------------------------------------
# web_server tests (FastAPI endpoints)
# ---------------------------------------------------------------------------

View file

@ -1,185 +0,0 @@
"""Tests for the subprocess console-window rule in check-windows-footguns.py.
These assert behavior contracts of the AST rule which call shapes get
flagged and which are correctly exempt NOT a snapshot of how many sites
the repo currently has. The rule's job: flag subprocess calls that can spawn
a NEW Windows console window, ignore the ones that physically cannot.
"""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
import pytest
# The checker lives at scripts/check-windows-footguns.py (hyphenated, not a
# normal importable module name) — load it by path.
_REPO_ROOT = Path(__file__).resolve().parents[2]
_CHECKER_PATH = _REPO_ROOT / "scripts" / "check-windows-footguns.py"
@pytest.fixture(scope="module")
def checker():
spec = importlib.util.spec_from_file_location("_wf_checker", _CHECKER_PATH)
mod = importlib.util.module_from_spec(spec)
# Register before exec so the module's dataclasses can resolve their
# __module__ via sys.modules (dataclasses._is_type looks it up there).
sys.modules["_wf_checker"] = mod
spec.loader.exec_module(mod)
return mod
def _flag(checker, src: str) -> list[int]:
"""Return the line numbers the subprocess rule flags for a source string."""
hits = checker.scan_subprocess_window_footguns(Path("x.py"), src)
return [lineno for (lineno, _line, _fg) in hits]
# --- Calls that SHOULD be flagged (can pop a Windows console) --------------
@pytest.mark.parametrize(
"src",
[
'subprocess.run(["git", "status"])',
'subprocess.Popen(["node", "x.js"])',
'subprocess.call(["npm", "run", "build"])',
'subprocess.check_call(["python", "setup.py"])',
"subprocess.run(cmd)", # dynamic argv, no redirection
'sp.run(["foo"])', # `sp` alias
],
)
def test_flags_bare_window_spawning_calls(checker, src):
assert _flag(checker, src) == [1], src
def test_flags_multiline_call_without_redirection(checker):
src = (
"subprocess.run(\n"
" [npm, 'run', 'build'],\n"
" cwd=desktop_dir,\n"
" check=False,\n"
")\n"
)
assert _flag(checker, src) == [1]
# --- Calls that should NOT be flagged (no new console possible) ------------
@pytest.mark.parametrize(
"src",
[
# captured/redirected AND not a known Windows-flashing program -> safe.
# (espeak-ng / a non-console-exe; capture inherits the parent console.)
'subprocess.run(["espeak-ng", "hi"], capture_output=True)',
'subprocess.run(["mytool", "x"], stdout=subprocess.PIPE)',
'subprocess.check_output(["mytool", "rev-parse"])',
# already managing the console
'subprocess.run(["git", "x"], creationflags=windows_hide_flags())',
# ** spread may carry a helper -> not penalised
"subprocess.Popen(argv, **windows_detach_popen_kwargs())",
"subprocess.run(cmd, **run_kwargs)",
# routed through the chokepoint wrapper -> different prefix, never flagged
"_subprocess_compat.run(['git', 'status'])",
],
)
def test_exempts_window_safe_calls(checker, src):
assert _flag(checker, src) == [], src
@pytest.mark.parametrize(
"src",
[
# Cross-platform console exes that allocate a Windows console even when
# captured — capture is NOT a safety boundary for these (Gille review,
# PR #53791 follow-up). They must be flagged despite capture/redirect.
'subprocess.run(["git", "status"], capture_output=True)',
'subprocess.run(["git", "x"], stdout=subprocess.PIPE)',
'subprocess.run(["gh", "pr", "list"], stderr=subprocess.DEVNULL)',
'subprocess.check_output(["git", "rev-parse", "HEAD"])',
'subprocess.run(["npm", "ci"], capture_output=True)',
'subprocess.run(["ffmpeg", "-i", "x"], capture_output=True)',
'subprocess.run(["docker", "info"], capture_output=True, timeout=10)',
'subprocess.run(["uv", "pip", "install"], capture_output=True)',
],
)
def test_flags_flashing_programs_even_when_captured(checker, src):
assert _flag(checker, src) == [1], src
@pytest.mark.parametrize(
"src",
[
'subprocess.run(["launchctl", "bootout", target])',
'subprocess.run(["systemctl", "status", svc])',
'subprocess.run(["brew", "install", "espeak-ng"])',
'subprocess.run(["codesign", "--sign", "-", app])',
'subprocess.run(["/usr/bin/sudo", "chmod", "4755", p])', # path-qualified
],
)
def test_exempts_posix_only_programs(checker, src):
"""launchctl/systemctl/brew/etc. don't exist on Windows -> can't pop a
Windows console, so they must not require a creationflag or suppression."""
assert _flag(checker, src) == [], src
def test_inline_suppression_marker(checker):
src = 'subprocess.run(["git", "x"]) # windows-footgun: ok\n'
assert _flag(checker, src) == []
def test_inline_suppression_on_multiline_closing_paren(checker):
src = (
"subprocess.run(\n"
" [npm, 'run', 'build'],\n"
" cwd=d,\n"
") # windows-footgun: ok\n"
)
assert _flag(checker, src) == []
def test_non_subprocess_calls_ignored(checker):
# A .run() on something that isn't the subprocess module is not our concern.
src = "loop.run(coro)\nclient.run()\n"
assert _flag(checker, src) == []
def test_syntax_error_returns_empty(checker):
assert _flag(checker, "def (:\n") == []
def test_repo_is_clean_of_window_footguns(checker):
"""Full-repo invariant: no unsuppressed window-spawning subprocess calls
remain in shippable Python packages. This is the chokepoint the rule
exists to hold."""
roots = [
_REPO_ROOT / d
for d in (
"hermes_cli",
"gateway",
"tools",
"cron",
"agent",
"plugins",
"scripts",
"acp_adapter",
"acp_registry",
)
]
roots = [r for r in roots if r.exists()]
offenders: list[str] = []
for path in checker.iter_files(roots):
if path.suffix not in {".py", ".pyw", ".pyi"}:
continue
try:
text = path.read_text(encoding="utf-8", errors="replace")
except OSError:
continue
for lineno, _line, _fg in checker.scan_subprocess_window_footguns(path, text):
offenders.append(f"{path.relative_to(_REPO_ROOT)}:{lineno}")
assert not offenders, "Unsuppressed Windows console footguns:\n" + "\n".join(
offenders
)

View file

@ -21,7 +21,6 @@ from __future__ import annotations
import io
import os
import platform
import subprocess
import sys
import textwrap
@ -232,82 +231,6 @@ class TestStdioReconfigureErrorHandling:
hb.apply_windows_utf8_bootstrap()
class TestWindowsPlatformProbeGuard:
def test_windows_bootstrap_disables_platform_syscmd_subprocess(self):
hb = _fresh_import()
hb._IS_WINDOWS = True
hb._bootstrap_applied = False
original = getattr(platform, "_syscmd_ver", None)
try:
hb.apply_windows_utf8_bootstrap()
assert platform._syscmd_ver("Windows", "", "") == ("Windows", "", "")
finally:
if original is not None:
platform._syscmd_ver = original
class TestDetachOrphanConsole:
"""detach_orphan_console() frees a solo-owned console (the uv pythonw→python
phantom) but leaves a shared interactive console attached, and is a pure
no-op on POSIX. It is intentionally NOT run at import time."""
def test_noop_on_posix(self):
hb = _fresh_import()
hb._IS_WINDOWS = False
assert hb.detach_orphan_console() is False
def test_not_called_at_import_time(self):
# The FreeConsole catch-all must be opt-in per background entry point,
# never an import side effect (would detach the interactive CLI/TUI).
import pathlib
src = pathlib.Path(_fresh_import().__file__).read_text(encoding="utf-8")
body = src.split("def detach_orphan_console")[0]
assert "FreeConsole" not in body, (
"FreeConsole must live only inside detach_orphan_console(), not in "
"apply_windows_utf8_bootstrap() / module import path"
)
def _fake_ctypes(self, monkeypatch, window, nproc):
import ctypes
class _K:
def __init__(self):
self.freed = False
def GetConsoleWindow(self):
return window
def GetConsoleProcessList(self, buf, n):
return nproc
def FreeConsole(self):
self.freed = True
k = _K()
monkeypatch.setattr(ctypes, "windll", type("_W", (), {"kernel32": k})(), raising=False)
return k
def test_frees_when_solo_owner(self, monkeypatch):
hb = _fresh_import()
hb._IS_WINDOWS = True
k = self._fake_ctypes(monkeypatch, window=1, nproc=1)
assert hb.detach_orphan_console() is True
assert k.freed is True
def test_leaves_shared_console_attached(self, monkeypatch):
hb = _fresh_import()
hb._IS_WINDOWS = True
k = self._fake_ctypes(monkeypatch, window=1, nproc=2)
assert hb.detach_orphan_console() is False
assert k.freed is False
def test_noop_without_console(self, monkeypatch):
hb = _fresh_import()
hb._IS_WINDOWS = True
k = self._fake_ctypes(monkeypatch, window=0, nproc=1)
assert hb.detach_orphan_console() is False
assert k.freed is False
class TestEntryPointsImportBootstrap:
"""Every Hermes entry point must import hermes_bootstrap as its
first non-docstring import. We check this by scanning source files

View file

@ -1,8 +1,6 @@
"""Tests for hermes_constants module."""
import os
import subprocess
import sys
from pathlib import Path
import pytest
@ -614,30 +612,6 @@ class TestAgentBrowserRunnable:
assert agent_browser_runnable("/usr/local/bin/npx agent-browser") is True
class TestAgentBrowserRunnableWindows:
def test_windows_validation_hides_console_subprocess(self, tmp_path, monkeypatch):
from hermes_cli import _subprocess_compat
exe = tmp_path / "agent-browser-win32-x64.exe"
exe.write_text("", encoding="utf-8")
exe.chmod(0o755)
captured = {}
def fake_run(args, **kwargs):
captured["args"] = args
captured.update(kwargs)
return type("Result", (), {"returncode": 0})()
monkeypatch.setattr(sys, "platform", "win32")
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", True)
monkeypatch.setattr(subprocess, "run", fake_run)
assert agent_browser_runnable(str(exe)) is True
assert captured["args"] == [str(exe), "--version"]
assert "creationflags" in captured
assert captured["stdin"] is subprocess.DEVNULL
class TestGetHermesDir:
"""Tests for ``get_hermes_dir(new_subpath, old_name)``.

View file

@ -1,121 +0,0 @@
"""Enforcement for the "no visible terminal on Windows" invariant.
Windows console-subsystem programs (``taskkill``, ``schtasks``, ``agent-browser``,
``git-bash`` ) pop a console window unless spawned with ``CREATE_NO_WINDOW``.
Relying on each call site to remember the flag is how cron-driven and future
spawns leaked terminal windows. The durable fix is a single chokepoint
``hermes_cli._subprocess_compat.run`` / ``.popen`` that always injects the
flag on Windows, plus the ``FreeConsole`` catch-all in ``hermes_bootstrap`` for
Python children.
These tests pin both halves of that contract:
1. The primitive actually injects ``CREATE_NO_WINDOW`` (and merges, so detach
callers still work).
2. No source file spawns a known console exe with a *raw* ``subprocess`` call,
which would bypass the primitive and reintroduce the window.
"""
from __future__ import annotations
import re
from pathlib import Path
import pytest
from hermes_cli import _subprocess_compat
REPO_ROOT = Path(__file__).resolve().parent.parent
_CREATE_NO_WINDOW = 0x08000000
class TestPrimitiveInjectsNoWindow:
def test_run_injects_create_no_window_on_windows(self, monkeypatch):
captured: dict = {}
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", True)
monkeypatch.setattr(
_subprocess_compat.subprocess, "run", lambda cmd, **kw: captured.update(kw) or "ok"
)
_subprocess_compat.run(["taskkill"], timeout=5)
assert captured["creationflags"] & _CREATE_NO_WINDOW
def test_popen_injects_create_no_window_on_windows(self, monkeypatch):
captured: dict = {}
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", True)
monkeypatch.setattr(
_subprocess_compat.subprocess, "Popen", lambda cmd, **kw: captured.update(kw) or "ok"
)
_subprocess_compat.popen(["agent-browser"])
assert captured["creationflags"] & _CREATE_NO_WINDOW
def test_merges_with_existing_detach_flags(self, monkeypatch):
captured: dict = {}
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", True)
monkeypatch.setattr(
_subprocess_compat.subprocess, "run", lambda cmd, **kw: captured.update(kw) or "ok"
)
detach = _subprocess_compat.windows_detach_flags()
_subprocess_compat.run(["x"], creationflags=detach)
assert captured["creationflags"] & _CREATE_NO_WINDOW
assert captured["creationflags"] & detach == detach
def test_no_op_on_posix(self, monkeypatch):
captured: dict = {}
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", False)
monkeypatch.setattr(
_subprocess_compat.subprocess, "run", lambda cmd, **kw: captured.update(kw) or "ok"
)
_subprocess_compat.run(["x"])
assert "creationflags" not in captured
# Windows-only console tools — they have no POSIX use, so a raw ``subprocess``
# spawn is unambiguously a Windows path that flashes a terminal. Banning them
# repo-wide is a pure win (cross-platform tools like git/ffmpeg/node are NOT
# listed: they have legitimate foreground/POSIX uses a blanket ban would break;
# their Windows-background call sites are routed through the primitive instead).
# ``_subprocess_compat.run/.popen`` calls never match these (different prefix).
_WINDOWS_ONLY_CONSOLE_EXES = ("taskkill", "schtasks", "wmic", "netstat", "tasklist")
_RAW_CONSOLE_SPAWNS = [
re.compile(rf"""subprocess\.(?:run|Popen|call)\(\s*\[\s*["']{exe}["']""")
for exe in _WINDOWS_ONLY_CONSOLE_EXES
]
# The primitive itself is allowed to call raw subprocess — it IS the chokepoint.
_ALLOWED = {REPO_ROOT / "hermes_cli" / "_subprocess_compat.py"}
# Dev/CI tooling that never ships to a user's Windows desktop, where a flashing
# console is irrelevant and importing hermes_cli would be inappropriate.
_SKIP_DIRS = {"tests", "node_modules", ".venv", "venv", "scripts"}
def _python_sources():
for path in REPO_ROOT.rglob("*.py"):
if _SKIP_DIRS & set(path.parts):
continue
if path in _ALLOWED:
continue
yield path
@pytest.mark.parametrize("pattern", _RAW_CONSOLE_SPAWNS, ids=_WINDOWS_ONLY_CONSOLE_EXES)
def test_no_raw_console_exe_spawns(pattern):
offenders = [
str(path.relative_to(REPO_ROOT))
for path in _python_sources()
if pattern.search(path.read_text(encoding="utf-8", errors="ignore"))
]
assert not offenders, (
"Console-subsystem exe spawned via raw subprocess (flashes a terminal on "
f"Windows). Route through hermes_cli._subprocess_compat.run/.popen instead: {offenders}"
)

View file

@ -89,28 +89,6 @@ class TestFindAgentBrowserCache:
with pytest.raises(FileNotFoundError, match="cached"):
bt._find_agent_browser()
def test_windows_prefers_native_agent_browser_exe_over_cmd_shim(self, tmp_path, monkeypatch):
import tools.browser_tool as bt
repo = tmp_path / "repo"
native = repo / "node_modules" / "agent-browser" / "bin" / "agent-browser-win32-x64.exe"
cmd = repo / "node_modules" / ".bin" / "agent-browser.cmd"
native.parent.mkdir(parents=True)
cmd.parent.mkdir(parents=True)
native.write_text("", encoding="utf-8")
cmd.write_text("", encoding="utf-8")
def fake_which(command, path=None):
return str(cmd) if path == str(cmd.parent) else None
monkeypatch.setattr(bt.sys, "platform", "win32")
monkeypatch.setattr(bt.shutil, "which", fake_which)
monkeypatch.setattr(bt, "agent_browser_runnable", lambda path: True)
monkeypatch.setattr(bt, "_merge_browser_path", lambda path: "")
monkeypatch.setattr(bt, "__file__", str(repo / "tools" / "browser_tool.py"))
assert bt._find_agent_browser() == str(native)
# ---------------------------------------------------------------------------
# Caching: _get_command_timeout

View file

@ -188,13 +188,11 @@ class TestTerminatePidRoutingOnWindows:
def test_force_uses_taskkill_on_windows(self, monkeypatch):
from gateway import status
from hermes_cli import _subprocess_compat
captured = {}
def fake_run(args, **kwargs):
captured["args"] = args
captured.update(kwargs)
result = MagicMock()
result.returncode = 0
result.stderr = ""
@ -202,7 +200,6 @@ class TestTerminatePidRoutingOnWindows:
return result
monkeypatch.setattr(status, "_IS_WINDOWS", True)
monkeypatch.setattr(_subprocess_compat, "IS_WINDOWS", True)
monkeypatch.setattr(status.subprocess, "run", fake_run)
status.terminate_pid(12345, force=True)
@ -211,7 +208,6 @@ class TestTerminatePidRoutingOnWindows:
assert "12345" in captured["args"]
assert "/T" in captured["args"]
assert "/F" in captured["args"]
assert captured["creationflags"] & 0x08000000
def test_force_taskkill_failure_raises_oserror(self, monkeypatch):
from gateway import status

View file

@ -68,7 +68,7 @@ from agent.auxiliary_client import call_llm
from hermes_constants import agent_browser_runnable, get_hermes_home
from utils import env_int, is_truthy_value
from hermes_cli.config import DEFAULT_CONFIG, cfg_get
from hermes_cli import _subprocess_compat
from hermes_cli._subprocess_compat import windows_hide_flags
try:
from tools.website_policy import check_website_access
@ -905,19 +905,23 @@ def _run_chrome_fallback_command(
# fileno=1 (stderr dup'd onto stdout at the OS level).
# * close_fds=True → block inheritance of every other handle.
# (Default on POSIX; must be explicit on Windows for stdio.)
# CREATE_NO_WINDOW is applied by _subprocess_compat.popen. We do NOT
# add CREATE_NEW_PROCESS_GROUP: on Python 3.11 Windows it interacts
# with asyncio's ProactorEventLoop such that the subprocess creation
# cancels the running loop task, surfacing as KeyboardInterrupt in
# app.run() and tearing down the CLI mid-turn (diag:
# "asyncio.CancelledError → KeyboardInterrupt").
_popen_extra: dict = {}
if os.name == "nt":
# CREATE_NO_WINDOW → don't attach a console (cmd.exe would
# otherwise briefly allocate one for the .cmd shim).
# Do NOT add CREATE_NEW_PROCESS_GROUP: on Python 3.11 Windows
# it interacts with asyncio's ProactorEventLoop such that the
# subprocess creation cancels the running loop task, which
# surfaces as KeyboardInterrupt in app.run() and tears down
# the CLI mid-turn. The agent thread's subprocess spawn
# unwound MainThread's prompt_toolkit loop that way — see
# diag log: "asyncio.CancelledError → KeyboardInterrupt".
_popen_extra["creationflags"] = windows_hide_flags()
_popen_extra["close_fds"] = True
_si = subprocess.STARTUPINFO()
_si.dwFlags |= subprocess.STARTF_USESTDHANDLES
_popen_extra["startupinfo"] = _si
proc = _subprocess_compat.popen(
proc = subprocess.Popen(
full, stdout=stdout_fd, stderr=stderr_fd,
stdin=subprocess.DEVNULL, env=browser_env,
**_popen_extra,
@ -1938,12 +1942,6 @@ def _find_agent_browser() -> str:
repo_root = Path(__file__).parent.parent
local_bin_dir = repo_root / "node_modules" / ".bin"
if local_bin_dir.is_dir():
if sys.platform == "win32":
native = repo_root / "node_modules" / "agent-browser" / "bin" / "agent-browser-win32-x64.exe"
if native.exists() and agent_browser_runnable(str(native)):
_cached_agent_browser = str(native)
_agent_browser_resolved = True
return _cached_agent_browser
local_which = shutil.which("agent-browser", path=str(local_bin_dir))
if local_which and agent_browser_runnable(local_which):
_cached_agent_browser = local_which
@ -2193,16 +2191,17 @@ def _run_browser_command(
# three explicit handles (no leaked parent-console handles to
# confuse the Rust binary's daemon-spawn), and close_fds=True to
# block inheritance of everything else.
# CREATE_NO_WINDOW via _subprocess_compat.popen; NO
# CREATE_NEW_PROCESS_GROUP (cancels asyncio loop task on Python 3.11
# Windows → KeyboardInterrupt in CLI MainThread).
_popen_extra: dict = {}
if os.name == "nt":
# See matching block at the other Popen site — CREATE_NO_WINDOW
# only, NO CREATE_NEW_PROCESS_GROUP (cancels asyncio loop task
# on Python 3.11 Windows → KeyboardInterrupt in CLI MainThread).
_popen_extra["creationflags"] = windows_hide_flags()
_popen_extra["close_fds"] = True
_si = subprocess.STARTUPINFO()
_si.dwFlags |= subprocess.STARTF_USESTDHANDLES
_popen_extra["startupinfo"] = _si
proc = _subprocess_compat.popen(
proc = subprocess.Popen(
cmd_parts,
stdout=stdout_fd,
stderr=stderr_fd,

View file

@ -61,7 +61,6 @@ from hermes_constants import get_hermes_home
from typing import Dict, List, Optional, Set, Tuple
from utils import env_int
from hermes_cli import _subprocess_compat
logger = logging.getLogger(__name__)
@ -446,7 +445,7 @@ def _init_store(store: Path, working_dir: str) -> Optional[str]:
"GIT_ALTERNATE_OBJECT_DIRECTORIES"):
init_env.pop(k, None)
try:
result = _subprocess_compat.run(
result = subprocess.run(
["git", "init", "--bare", str(store)],
capture_output=True, text=True,
env=init_env, timeout=_GIT_TIMEOUT,

View file

@ -29,7 +29,6 @@ import shutil
import subprocess
import sys
from typing import Any, Dict, List, Optional
from hermes_cli._subprocess_compat import windows_hide_flags
# Platforms with a cua-driver runtime backend (mirrors the toolset platform_gate).
_RUNTIME_PLATFORMS = frozenset({"darwin", "win32", "linux"})
@ -181,7 +180,6 @@ def request_permissions_grant(driver_cmd: Optional[str] = None) -> int:
[binary, "permissions", "grant"],
env=_child_env(),
stdin=subprocess.DEVNULL,
creationflags=windows_hide_flags(),
).returncode
)
except KeyboardInterrupt: # pragma: no cover - interactive

View file

@ -13,7 +13,7 @@ import time
from pathlib import Path
from tools.environments.base import BaseEnvironment, _pipe_stdin
from hermes_cli import _subprocess_compat
from hermes_cli._subprocess_compat import windows_hide_flags
_IS_WINDOWS = platform.system() == "Windows"
@ -738,7 +738,9 @@ class LocalEnvironment(BaseEnvironment):
_popen_cwd = self.cwd
proc = _subprocess_compat.popen(
_popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {}
proc = subprocess.Popen(
args,
text=True,
env=run_env,
@ -749,6 +751,7 @@ class LocalEnvironment(BaseEnvironment):
stdin=subprocess.PIPE if stdin_data is not None else subprocess.DEVNULL,
start_new_session=True,
cwd=_popen_cwd,
**_popen_kwargs,
)
if not _IS_WINDOWS:
try:

View file

@ -42,7 +42,6 @@ import uuid
_IS_WINDOWS = platform.system() == "Windows"
from tools.environments.local import _find_shell, _resolve_safe_cwd, _sanitize_subprocess_env
from hermes_cli import _subprocess_compat
from hermes_cli._subprocess_compat import windows_hide_flags
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
@ -570,11 +569,12 @@ class ProcessRegistry:
return
if _IS_WINDOWS:
try:
_subprocess_compat.run(
subprocess.run(
["taskkill", "/PID", str(pid), "/T", "/F"],
capture_output=True,
text=True,
timeout=10,
creationflags=windows_hide_flags(),
stdin=subprocess.DEVNULL,
)
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):

View file

@ -38,7 +38,6 @@ from tools.skills_guard import (
)
from tools.url_safety import is_safe_url
from tools.website_policy import check_website_access
from hermes_cli import _subprocess_compat
logger = logging.getLogger(__name__)
@ -299,7 +298,7 @@ class GitHubAuth:
def _try_gh_cli(self) -> Optional[str]:
"""Try to get a token from the gh CLI."""
try:
result = _subprocess_compat.run(
result = subprocess.run(
["gh", "auth", "token"],
capture_output=True, text=True, timeout=5,
stdin=subprocess.DEVNULL,

View file

@ -37,7 +37,6 @@ from pathlib import Path
from typing import Optional, Dict, Any
from urllib.parse import urljoin
from hermes_cli import _subprocess_compat
from utils import is_truthy_value
from tools.managed_tool_gateway import resolve_managed_tool_gateway
from tools.tool_backend_helpers import (
@ -486,7 +485,7 @@ def _terminate_command_stt_process_tree(proc: subprocess.Popen) -> None:
if os.name == "nt":
try:
_subprocess_compat.run(
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(proc.pid)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
@ -557,7 +556,7 @@ def _run_command_stt(command: str, timeout: float) -> subprocess.CompletedProces
else:
popen_kwargs["start_new_session"] = True
proc = _subprocess_compat.popen(command, **popen_kwargs, stdin=subprocess.DEVNULL)
proc = subprocess.Popen(command, **popen_kwargs, stdin=subprocess.DEVNULL)
try:
stdout, stderr = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired as exc:
@ -1188,7 +1187,7 @@ def _prepare_local_audio(file_path: str, work_dir: str) -> tuple[Optional[str],
command = [ffmpeg, "-y", "-i", file_path, converted_path]
try:
_subprocess_compat.run(command, check=True, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
subprocess.run(command, check=True, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
return converted_path, None
except subprocess.TimeoutExpired:
logger.error("ffmpeg conversion timed out for %s", file_path)
@ -1234,9 +1233,9 @@ def _transcribe_local_command(file_path: str, model_name: str) -> Dict[str, Any]
# User-provided templates (env var) may contain shell syntax; auto-detected commands are safe for list mode.
use_shell = bool(os.getenv(LOCAL_STT_COMMAND_ENV, "").strip())
if use_shell:
_subprocess_compat.run(command, shell=True, check=True, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
subprocess.run(command, shell=True, check=True, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
else:
_subprocess_compat.run(shlex.split(command), check=True, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
subprocess.run(shlex.split(command), check=True, capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
txt_files = sorted(Path(output_dir).glob("*.txt"))

View file

@ -52,7 +52,6 @@ from pathlib import Path
from typing import Callable, Dict, Any, Optional
from urllib.parse import urljoin
from hermes_cli import _subprocess_compat
from hermes_constants import display_hermes_home
logger = logging.getLogger(__name__)
@ -715,7 +714,7 @@ def _terminate_command_tts_process_tree(proc: subprocess.Popen) -> None:
if os.name == "nt":
try:
_subprocess_compat.run(
subprocess.run(
["taskkill", "/F", "/T", "/PID", str(proc.pid)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
@ -773,7 +772,7 @@ def _run_command_tts(command: str, timeout: float) -> subprocess.CompletedProces
else:
popen_kwargs["start_new_session"] = True
proc = _subprocess_compat.popen(command, **popen_kwargs, stdin=subprocess.DEVNULL)
proc = subprocess.Popen(command, **popen_kwargs, stdin=subprocess.DEVNULL)
try:
stdout, stderr = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired as exc:
@ -906,7 +905,7 @@ def _convert_to_opus(mp3_path: str) -> Optional[str]:
ogg_path = mp3_path.rsplit(".", 1)[0] + ".ogg"
try:
result = _subprocess_compat.run(
result = subprocess.run(
["ffmpeg", "-i", mp3_path, "-acodec", "libopus",
"-ac", "1", "-b:a", "64k", "-vbr", "off", ogg_path, "-y"],
capture_output=True, timeout=30,
@ -1777,7 +1776,7 @@ def _generate_gemini_tts(text: str, output_path: str, tts_config: Dict[str, Any]
]
else:
cmd = [ffmpeg, "-i", wav_path, "-y", "-loglevel", "error", output_path]
result = _subprocess_compat.run(cmd, capture_output=True, timeout=30, stdin=subprocess.DEVNULL)
result = subprocess.run(cmd, capture_output=True, timeout=30, stdin=subprocess.DEVNULL)
if result.returncode != 0:
stderr = result.stderr.decode("utf-8", errors="ignore")[:300]
raise RuntimeError(f"ffmpeg conversion failed: {stderr}")
@ -1860,7 +1859,7 @@ def _generate_neutts(text: str, output_path: str, tts_config: Dict[str, Any]) ->
"--device", device,
]
result = _subprocess_compat.run(cmd, capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL)
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL)
if result.returncode != 0:
stderr = result.stderr.strip()
# Filter out the "OK:" line from stderr
@ -1872,7 +1871,7 @@ def _generate_neutts(text: str, output_path: str, tts_config: Dict[str, Any]) ->
ffmpeg = shutil.which("ffmpeg")
if ffmpeg:
conv_cmd = [ffmpeg, "-i", wav_path, "-y", "-loglevel", "error", output_path]
_subprocess_compat.run(conv_cmd, check=True, timeout=30, stdin=subprocess.DEVNULL)
subprocess.run(conv_cmd, check=True, timeout=30, stdin=subprocess.DEVNULL)
os.remove(wav_path)
else:
# No ffmpeg — just rename the WAV to the expected path
@ -1939,7 +1938,7 @@ def _resolve_piper_voice_path(voice: str, download_dir: Path) -> str:
import sys as _sys
logger.info("[Piper] Downloading voice '%s' to %s (first use)", voice, download_dir)
try:
result = _subprocess_compat.run(
result = subprocess.run(
[_sys.executable, "-m", "piper.download_voices", voice,
"--download-dir", str(download_dir)],
capture_output=True, text=True, timeout=300,
@ -2051,7 +2050,7 @@ def _generate_piper_tts(text: str, output_path: str, tts_config: Dict[str, Any])
ffmpeg = shutil.which("ffmpeg")
if ffmpeg:
conv_cmd = [ffmpeg, "-i", wav_path, "-y", "-loglevel", "error", output_path]
_subprocess_compat.run(conv_cmd, check=True, timeout=30, stdin=subprocess.DEVNULL)
subprocess.run(conv_cmd, check=True, timeout=30, stdin=subprocess.DEVNULL)
try:
os.remove(wav_path)
except OSError:
@ -2117,7 +2116,7 @@ def _generate_kittentts(text: str, output_path: str, tts_config: Dict[str, Any])
ffmpeg = shutil.which("ffmpeg")
if ffmpeg:
conv_cmd = [ffmpeg, "-i", wav_path, "-y", "-loglevel", "error", output_path]
_subprocess_compat.run(conv_cmd, check=True, timeout=30, stdin=subprocess.DEVNULL)
subprocess.run(conv_cmd, check=True, timeout=30, stdin=subprocess.DEVNULL)
os.remove(wav_path)
else:
# No ffmpeg — rename the WAV to the expected path

View file

@ -260,13 +260,6 @@ def join_mcp_discovery(timeout: float | None = None) -> bool:
def main():
# Stdio backend spawned by Node/Electron: drop any console a uv
# pythonw→python re-exec auto-allocated. No-op on POSIX.
try:
hermes_bootstrap.detach_orphan_console()
except Exception:
pass
_install_sidecar_publisher()
# MCP tool discovery — runs in a background daemon thread so a slow or

View file

@ -46,9 +46,7 @@ def run_git(cwd: str, *args: str) -> str:
if not cwd:
return ""
try:
from hermes_cli import _subprocess_compat
result = _subprocess_compat.run(
result = subprocess.run(
["git", "-C", cwd, *args],
capture_output=True,
text=True,

View file

@ -93,14 +93,6 @@ def _run(cli: HermesCLI, command: str) -> str:
def main():
# Stdio worker spawned by the gateway: drop any console a uv pythonw→python
# re-exec auto-allocated. No-op on POSIX.
try:
import hermes_bootstrap
hermes_bootstrap.detach_orphan_console()
except Exception:
pass
p = argparse.ArgumentParser(add_help=False)
p.add_argument("--session-key", required=True)
p.add_argument("--model", default="")