fix(gateway): improve Windows detach diagnostics

hermes_cli/gateway.py

@@ -68,6 +68,9 @@ from hermes_cli.colors import Colors, color
 
 logger = logging.getLogger(__name__)
 
+# Private launcher-to-child metadata. This is diagnostic state, not user config.
+_WINDOWS_GATEWAY_BREAKAWAY_ENV = "_HERMES_GATEWAY_BREAKAWAY"
+
 # =============================================================================
 # Process Management (for manual gateway runs)
 # =============================================================================
@@ -2051,6 +2054,30 @@ def _windows_gateway_should_absorb_console_controls() -> bool:
         return True
 
 
+def _windows_console_window_attached() -> bool | None:
+    """Return whether Windows assigned this process a console window."""
+    if not is_windows():
+        return None
+    try:
+        import ctypes
+
+        return bool(ctypes.windll.kernel32.GetConsoleWindow())  # type: ignore[attr-defined]
+    except (OSError, AttributeError):
+        return None
+
+
+def _windows_gateway_breakaway_state() -> bool | None:
+    """Consume private spawn metadata without guessing for older launchers."""
+    if not is_windows():
+        return None
+    value = os.environ.pop(_WINDOWS_GATEWAY_BREAKAWAY_ENV, None)
+    if value == "1":
+        return True
+    if value == "0":
+        return False
+    return None
+
+
 # =============================================================================
 # Service Configuration
 # =============================================================================
@@ -5539,6 +5566,12 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo
         _stdin_is_tty = bool(sys.stdin and sys.stdin.isatty())
     except (ValueError, OSError):
         _stdin_is_tty = False
+    _console_window_attached = _windows_console_window_attached()
+    _gateway_detached = (
+        os.getenv("HERMES_GATEWAY_DETACHED", "").strip().lower()
+        in {"1", "true", "yes", "on"}
+    )
+    _breakaway = _windows_gateway_breakaway_state()
     _absorb_windows_console_controls = _windows_gateway_should_absorb_console_controls()
     if _absorb_windows_console_controls:
         try:
@@ -5641,6 +5674,9 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo
         replace=replace,
         argv=sys.argv,
         stdin_is_tty=_stdin_is_tty,
+        console_window_attached=_console_window_attached,
+        detached=_gateway_detached,
+        breakaway=_breakaway,
         absorb_windows_console_controls=_absorb_windows_console_controls,
     )
 

hermes_cli/gateway_windows.py

@@ -29,6 +29,7 @@ from __future__ import annotations
 
 import ctypes
 import locale
+import logging
 import os
 import re
 import shlex
@@ -45,6 +46,11 @@ from hermes_cli._subprocess_compat import (
     windows_hide_flags,
 )
 
+logger = logging.getLogger(__name__)
+
+# Private launcher-to-child metadata. This is diagnostic state, not user config.
+_GATEWAY_BREAKAWAY_ENV = "_HERMES_GATEWAY_BREAKAWAY"
+
 # Short timeouts: schtasks occasionally wedges and we don't want to hang forever.
 _SCHTASKS_TIMEOUT_S = 15
 _SCHTASKS_NO_OUTPUT_TIMEOUT_S = 30
@@ -913,6 +919,7 @@ def _spawn_detached(script_path: Path | None = None) -> int:
 
     # Inherit PATH etc. from the current env, overlay our required vars.
     env = {**os.environ, **env_overlay}
+    primary_env = {**env, _GATEWAY_BREAKAWAY_ENV: "1"}
 
     # CREATE_NEW_PROCESS_GROUP 0x00000200 — child gets its own group, won't
     #                                       receive Ctrl+C from our group
@@ -942,25 +949,34 @@ def _spawn_detached(script_path: Path | None = None) -> int:
             proc = subprocess.Popen(
                 argv,
                 cwd=working_dir,
-                env=env,
+                env=primary_env,
                 creationflags=flags,
                 close_fds=True,
                 stdin=subprocess.DEVNULL,
                 stdout=log_fh,
                 stderr=log_fh,
             )
-    except OSError:
+    except OSError as exc:
         # CREATE_BREAKAWAY_FROM_JOB can fail with "access denied" when the
         # parent's job object doesn't permit breakaway (some Windows
         # Terminal configs). Retry without the breakaway flag — in most
         # setups the hidden-console CREATE_NO_WINDOW spawn is enough on
         # its own.
+        error_code = getattr(exc, "winerror", None)
+        if error_code is None:
+            error_code = exc.errno
+        logger.warning(
+            "Gateway breakaway spawn failed (error=%s); retrying without "
+            "CREATE_BREAKAWAY_FROM_JOB",
+            error_code,
+        )
         flags_no_breakaway = windows_detach_flags_without_breakaway()
+        fallback_env = {**env, _GATEWAY_BREAKAWAY_ENV: "0"}
         with open(stray_log, "ab", buffering=0) as log_fh:
             proc = subprocess.Popen(
                 argv,
                 cwd=working_dir,
-                env=env,
+                env=fallback_env,
                 creationflags=flags_no_breakaway,
                 close_fds=True,
                 stdin=subprocess.DEVNULL,

tests/hermes_cli/test_gateway.py

@@ -1,6 +1,7 @@
 """Tests for hermes_cli.gateway."""
 
 import argparse
+import json
 import os
 import signal
 import subprocess
@@ -13,6 +14,9 @@ import pytest
 import hermes_cli.gateway as gateway
 
 
+_BREAKAWAY_MARKER = "_HERMES_GATEWAY_BREAKAWAY"
+
+
 def _install_fake_gateway_run(monkeypatch, start_gateway):
     module = ModuleType("gateway.run")
     module.start_gateway = start_gateway
@@ -49,6 +53,102 @@ def _install_fake_gateway_run(monkeypatch, start_gateway):
     )
 
 
+def _run_native_windows_gateway_start_diag(
+    tmp_path, breakaway_marker: str | None
+):
+    script = textwrap.dedent(
+        """
+        import ctypes
+        import json
+        import os
+        import pathlib
+        import sys
+        import types
+
+        import hermes_cli.gateway as gateway_cli
+
+        async def start_gateway(*, replace, verbosity):
+            assert "_HERMES_GATEWAY_BREAKAWAY" not in os.environ
+            return True
+
+        fake_run = types.ModuleType("gateway.run")
+        fake_run.start_gateway = start_gateway
+        fake_run._exit_after_graceful_shutdown = lambda code: None
+        sys.modules["gateway.run"] = fake_run
+
+        gateway_cli._guard_official_docker_root_gateway = lambda: None
+        gateway_cli._guard_named_profile_under_multiplexer = lambda force=False: None
+        gateway_cli._guard_supervised_gateway_conflict = lambda force=False: None
+        gateway_cli._guard_existing_gateway_process_conflict = lambda replace=False: None
+        gateway_cli.supports_systemd_services = lambda: False
+        gateway_cli.run_gateway(quiet=True)
+
+        diag_path = pathlib.Path(os.environ["HERMES_HOME"]) / "logs" / "gateway-exit-diag.log"
+        rows = [json.loads(line) for line in diag_path.read_text(encoding="utf-8").splitlines()]
+        start = next(row for row in rows if row["tag"] == "gateway.start")
+        payload = {
+            "diag": start,
+            "get_console_window": bool(ctypes.windll.kernel32.GetConsoleWindow()),
+        }
+        print("DIAG_JSON=" + json.dumps(payload))
+        """
+    )
+    env: dict[str, str] = dict(os.environ)
+    env.update(
+        {
+            "HERMES_HOME": str(tmp_path),
+            "HERMES_GATEWAY_DETACHED": "1",
+            "HERMES_GATEWAY_EXIT_DIAG": "1",
+            "HERMES_GATEWAY_MAX_STARTS": "0",
+            "PYTHONIOENCODING": "utf-8",
+        }
+    )
+    if breakaway_marker is None:
+        env.pop(_BREAKAWAY_MARKER, None)
+    else:
+        env[_BREAKAWAY_MARKER] = breakaway_marker
+
+    from hermes_cli._subprocess_compat import windows_detach_flags_without_breakaway
+
+    completed = subprocess.run(
+        [sys.executable, "-c", script],
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        creationflags=windows_detach_flags_without_breakaway(),
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        env=env,
+        timeout=30,
+        check=False,
+    )
+    assert completed.returncode == 0, completed.stderr
+    line = next(
+        line for line in completed.stdout.splitlines() if line.startswith("DIAG_JSON=")
+    )
+    return json.loads(line.removeprefix("DIAG_JSON="))
+
+
+@pytest.mark.windows_only
+@pytest.mark.parametrize(
+    ("marker", "expected_breakaway"),
+    [("1", True), ("0", False), (None, None)],
+)
+def test_windows_gateway_start_diag_reports_detach_state(
+    tmp_path, marker, expected_breakaway
+):
+    """DEVNULL is a Windows TTY but must not masquerade as a console window."""
+    payload = _run_native_windows_gateway_start_diag(tmp_path, marker)
+    diag = payload["diag"]
+
+    assert payload["get_console_window"] is False
+    assert diag["stdin_is_tty"] is True
+    assert diag["console_window_attached"] is False
+    assert diag["detached"] is True
+    assert diag["breakaway"] is expected_breakaway
+
+
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX PTY coverage")

tests/hermes_cli/test_gateway_windows.py

@@ -1,6 +1,9 @@
 """Tests for hermes_cli.gateway_windows."""
 
+import logging
+import subprocess
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -9,6 +12,9 @@ import hermes_cli.gateway_windows as gateway_windows
 import hermes_cli.setup as setup
 
 
+_BREAKAWAY_MARKER = "_HERMES_GATEWAY_BREAKAWAY"
+
+
 
 
 def test_schtasks_encoding_falls_back_to_utf8(monkeypatch):
@@ -73,6 +79,101 @@ def test_build_gateway_argv_keeps_venv_console_python_for_uv_venv(monkeypatch, t
     assert str(project) in env_overlay["PYTHONPATH"].split(gateway_windows.os.pathsep)
 
 
+@pytest.mark.windows_only
+def test_spawn_detached_marks_primary_breakaway_success(monkeypatch, tmp_path, caplog):
+    """A successful breakaway spawn reports true without a warning."""
+    argv = ["python.exe", "-m", "hermes_cli.main", "gateway", "run"]
+    cwd = str(tmp_path)
+    calls = []
+
+    def fake_popen(call_argv, **kwargs):
+        calls.append((call_argv, kwargs))
+        return SimpleNamespace(pid=12345)
+
+    monkeypatch.setattr(
+        gateway_windows,
+        "_build_gateway_argv",
+        lambda: (argv, cwd, {"HERMES_GATEWAY_DETACHED": "1"}),
+    )
+    monkeypatch.setattr("hermes_cli.config.get_hermes_home", lambda: tmp_path)
+    monkeypatch.setattr(gateway_windows.subprocess, "Popen", fake_popen)
+    caplog.set_level(logging.WARNING, logger=gateway_windows.__name__)
+
+    assert gateway_windows._spawn_detached() == 12345
+    assert len(calls) == 1
+    actual_argv, kwargs = calls[0]
+    assert actual_argv == argv
+    assert kwargs["cwd"] == cwd
+    assert kwargs["creationflags"] == gateway_windows.windows_detach_flags()
+    assert kwargs["env"][_BREAKAWAY_MARKER] == "1"
+    assert kwargs["stdin"] is subprocess.DEVNULL
+    assert kwargs["stdout"] is kwargs["stderr"]
+    assert not caplog.records
+
+
+@pytest.mark.windows_only
+def test_spawn_detached_warns_and_marks_no_breakaway_fallback(
+    monkeypatch, tmp_path, caplog
+):
+    """A denied breakaway retries once with private false metadata."""
+    argv = ["python.exe", "-m", "hermes_cli.main", "gateway", "run"]
+    cwd = str(tmp_path)
+    calls = []
+
+    def fake_popen(call_argv, **kwargs):
+        calls.append((call_argv, kwargs))
+        if len(calls) == 1:
+            error = OSError(13, "Access is denied")
+            error.winerror = 5
+            raise error
+        return SimpleNamespace(pid=23456)
+
+    monkeypatch.setattr(
+        gateway_windows,
+        "_build_gateway_argv",
+        lambda: (
+            argv,
+            cwd,
+            {"HERMES_GATEWAY_DETACHED": "1", "SECRET_SENTINEL": "do-not-log"},
+        ),
+    )
+    monkeypatch.setattr("hermes_cli.config.get_hermes_home", lambda: tmp_path)
+    monkeypatch.setattr(gateway_windows.subprocess, "Popen", fake_popen)
+    caplog.set_level(logging.WARNING, logger=gateway_windows.__name__)
+
+    assert gateway_windows._spawn_detached() == 23456
+    assert len(calls) == 2
+    (argv_primary, primary), (argv_fallback, fallback) = calls
+    assert argv_primary == argv_fallback == argv
+    assert primary["cwd"] == fallback["cwd"] == cwd
+    assert primary["creationflags"] == gateway_windows.windows_detach_flags()
+    assert (
+        fallback["creationflags"]
+        == gateway_windows.windows_detach_flags_without_breakaway()
+    )
+    assert primary["stdin"] is fallback["stdin"] is subprocess.DEVNULL
+    assert primary["stdout"] is primary["stderr"]
+    assert fallback["stdout"] is fallback["stderr"]
+    assert Path(primary["stdout"].name) == Path(fallback["stdout"].name)
+    assert primary["close_fds"] is fallback["close_fds"] is True
+    assert primary["env"] is not fallback["env"]
+    assert primary["env"][_BREAKAWAY_MARKER] == "1"
+    assert fallback["env"][_BREAKAWAY_MARKER] == "0"
+    assert {
+        key: value for key, value in primary["env"].items() if key != _BREAKAWAY_MARKER
+    } == {
+        key: value for key, value in fallback["env"].items() if key != _BREAKAWAY_MARKER
+    }
+
+    warnings = [
+        record for record in caplog.records if record.levelno == logging.WARNING
+    ]
+    assert len(warnings) == 1
+    assert "5" in warnings[0].getMessage()
+    assert "do-not-log" not in warnings[0].getMessage()
+    assert str(tmp_path) not in warnings[0].getMessage()
+
+
 class TestStableWindowsGatewayWorkingDir:
     def test_stable_gateway_working_dir_uses_hermes_home(self, tmp_path, monkeypatch):
         home = tmp_path / ".hermes"
@@ -269,5 +370,3 @@ def test_gateway_vbs_script_is_console_less(monkeypatch):
 
 
 
-
-