From abcf996b1f749a647a1b213653a80d1eee58f6d1 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 8 Jun 2026 11:32:43 -0700 Subject: [PATCH] feat(windows): enable dashboard /chat tab via ConPTY (win_pty_bridge) + tests (#42251) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(windows): enable dashboard chat tab via ConPTY (win_pty_bridge) Add hermes_cli/win_pty_bridge.py — a pywinpty-backed drop-in for PtyBridge with the same spawn/read/write/resize/close surface — and wire it into the web_server PTY import block so Windows picks it up instead of falling back to None. pywinpty is already a declared win32 dependency (pyproject.toml). The ConPTY read path runs inside run_in_executor so the event loop is never blocked. Spawn/read/write/terminate call shapes are taken directly from tools/process_registry.py which already exercises the same pywinpty version. Co-Authored-By: Claude Sonnet 4.6 * docs: remove WSL2-only caveat for dashboard chat tab The chat pane now works on native Windows via the ConPTY bridge added in the previous commit. Co-Authored-By: Claude Sonnet 4.6 * test(windows): cover ConPTY bridge + web_server platform-branched import Companion to the bridge added in the previous commits. Verified live on native Windows 11 (pywinpty 2.0.15) against `hermes dashboard`'s `/api/pty` WebSocket: the spawned `hermes --tui` (node entry.js) renders through ConPTY, resize escapes reach `setwinsize`, and closing the WS reaps both the node child and the pywinpty agent with zero orphans. tests/hermes_cli/test_win_pty_bridge.py Mirrors the layout of the existing POSIX test_pty_bridge.py: spawn/io/resize/close/env coverage against cmd.exe and python -c, plus the cross-platform fallback surface (PtyUnavailableError, the off-Windows `spawn -> raises PtyUnavailableError` guard, and the load-bearing _clamp() helper that protects setwinsize from garbage winsize values out of xterm.js). tests/hermes_cli/test_web_server_pty_import.py Asserts that web_server.PtyBridge resolves to WinPtyBridge on win32 and to the POSIX PtyBridge on POSIX, that PtyUnavailableError is the matching class on each side (so isinstance checks in /api/pty's spawn fallback path work), and a source-text check that pins the platform-branched import shape so a future refactor can't quietly collapse it back to a POSIX-only import. scripts/release.py AUTHOR_MAP entries so CI release-note generation can resolve both authors' plain (non-noreply) emails to their GitHub logins. Co-Authored-By: JoelJJohnson Co-Authored-By: Nea74 --------- Co-authored-by: JoelJJohnson Co-authored-by: Claude Sonnet 4.6 Co-authored-by: Nea74 --- README.md | 2 +- hermes_cli/web_server.py | 38 ++- hermes_cli/win_pty_bridge.py | 179 ++++++++++ scripts/release.py | 2 + .../hermes_cli/test_web_server_pty_import.py | 83 +++++ tests/hermes_cli/test_win_pty_bridge.py | 315 ++++++++++++++++++ 6 files changed, 605 insertions(+), 14 deletions(-) create mode 100644 hermes_cli/win_pty_bridge.py create mode 100644 tests/hermes_cli/test_web_server_pty_import.py create mode 100644 tests/hermes_cli/test_win_pty_bridge.py diff --git a/README.md b/README.md index 2c587b81ac5..a8db8cb2c29 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,7 @@ If you already have Git installed, the installer detects it and uses that instea > **Android / Termux:** The tested manual path is documented in the [Termux guide](https://hermes-agent.nousresearch.com/docs/getting-started/termux). On Termux, Hermes installs a curated `.[termux]` extra because the full `.[all]` extra currently pulls Android-incompatible voice dependencies. > -> **Windows:** Native Windows is fully supported — the PowerShell one-liner above installs everything. If you'd rather use WSL2, the Linux command works there too. Native Windows install lives under `%LOCALAPPDATA%\hermes`; WSL2 installs under `~/.hermes` as on Linux. The only Hermes feature that currently needs WSL2 specifically is the browser-based dashboard chat pane (it uses a POSIX PTY — classic CLI and gateway both run natively). +> **Windows:** Native Windows is fully supported — the PowerShell one-liner above installs everything. If you'd rather use WSL2, the Linux command works there too. Native Windows install lives under `%LOCALAPPDATA%\hermes`; WSL2 installs under `~/.hermes` as on Linux. After installation: diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index c3e1d7ec3e4..2b4034b2ec5 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -8288,20 +8288,32 @@ async def get_models_analytics(days: int = 30): # though uvicorn binds to 127.0.0.1. # --------------------------------------------------------------------------- -# PTY bridge is POSIX-only (depends on fcntl/termios/ptyprocess). On native -# Windows the import raises; catch and leave PtyBridge=None so the rest of -# the dashboard (sessions, jobs, metrics, config editor) still loads and the -# /api/pty endpoint cleanly refuses with a WSL-suggested message. -try: - from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError - _PTY_BRIDGE_AVAILABLE = True -except ImportError as _pty_import_err: # pragma: no cover - Windows-only path - PtyBridge = None # type: ignore[assignment] - _PTY_BRIDGE_AVAILABLE = False +# PTY bridge: POSIX uses pty_bridge (fcntl/termios/ptyprocess); native Windows +# uses win_pty_bridge (pywinpty/ConPTY, already a declared dependency). Both +# expose the same public surface — spawn/read/write/resize/close/is_available — +# so the /api/pty WebSocket handler needs no platform guards. +if sys.platform.startswith("win"): + try: + from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, PtyUnavailableError + _PTY_BRIDGE_AVAILABLE = True + except ImportError: # pragma: no cover - pywinpty missing + PtyBridge = None # type: ignore[assignment] + _PTY_BRIDGE_AVAILABLE = False - class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] - """Stub on platforms where pty_bridge can't be imported.""" - pass + class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] + """Stub when win_pty_bridge cannot be imported.""" + pass +else: + try: + from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError + _PTY_BRIDGE_AVAILABLE = True + except ImportError: # pragma: no cover - dev env without ptyprocess + PtyBridge = None # type: ignore[assignment] + _PTY_BRIDGE_AVAILABLE = False + + class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] + """Stub on platforms where pty_bridge can't be imported.""" + pass _RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]") _PTY_READ_CHUNK_TIMEOUT = 0.2 diff --git a/hermes_cli/win_pty_bridge.py b/hermes_cli/win_pty_bridge.py new file mode 100644 index 00000000000..fe8ca1acb04 --- /dev/null +++ b/hermes_cli/win_pty_bridge.py @@ -0,0 +1,179 @@ +"""Windows ConPTY bridge for the `hermes dashboard` chat tab. + +Drop-in counterpart to ``hermes_cli.pty_bridge.PtyBridge`` for native +Windows. Mirrors the exact public surface the ``/api/pty`` WebSocket +handler in ``hermes_cli.web_server`` consumes: ``spawn``, ``read``, +``write``, ``resize``, ``close``, ``is_available``, plus the +``PtyUnavailableError`` type. + +Backed by ``pywinpty`` (already a declared win32 dependency in +pyproject.toml) instead of ``ptyprocess``/``fcntl``/``termios``, none of +which exist on native Windows. The read/write/terminate calls here match +the working winpty usage already shipping in ``tools/process_registry.py``. +""" + +from __future__ import annotations + +import os +import sys +import time +from typing import Optional, Sequence + +try: + from winpty import PtyProcess # type: ignore + _PTY_AVAILABLE = sys.platform.startswith("win") +except ImportError: # pragma: no cover - non-Windows or pywinpty missing + PtyProcess = None # type: ignore + _PTY_AVAILABLE = False + + +__all__ = ["WinPtyBridge", "PtyUnavailableError"] + + +# Same clamp ceiling as the POSIX bridge: a broken winsize probe must never +# reach the resize call. ConPTY tolerates large values better than ioctl, +# but we keep parity to avoid layout surprises. +_MIN_DIMENSION = 1 +_MAX_COLS = 2000 +_MAX_ROWS = 1000 + + +def _clamp(value: int, maximum: int) -> int: + try: + n = int(value) + except (TypeError, ValueError, OverflowError): + return _MIN_DIMENSION + if n < _MIN_DIMENSION: + return _MIN_DIMENSION + if n > maximum: + return maximum + return n + + +class PtyUnavailableError(RuntimeError): + """Raised when a PTY cannot be created on this platform.""" + + +class WinPtyBridge: + """pywinpty-backed bridge with the same interface as ``PtyBridge``. + + ``web_server`` calls :meth:`read` inside ``run_in_executor``, so a + blocking/polling read here never stalls the event loop. ConPTY exposes + no selectable fd, so we poll with a short sleep instead of ``select``. + """ + + def __init__(self, proc: "PtyProcess") -> None: # type: ignore[name-defined] + self._proc = proc + self._closed = False + + # -- lifecycle -------------------------------------------------------- + + @classmethod + def is_available(cls) -> bool: + return bool(_PTY_AVAILABLE) + + @classmethod + def spawn( + cls, + argv: Sequence[str], + *, + cwd: Optional[str] = None, + env: Optional[dict] = None, + cols: int = 80, + rows: int = 24, + ) -> "WinPtyBridge": + if not _PTY_AVAILABLE: + if PtyProcess is None: + raise PtyUnavailableError( + "pywinpty is not installed. Install with: pip install pywinpty" + ) + raise PtyUnavailableError("ConPTY is unavailable on this platform.") + spawn_env = (os.environ.copy() if env is None else dict(env)) + if not spawn_env.get("TERM"): + spawn_env["TERM"] = "xterm-256color" + # pywinpty mirrors ptyprocess: dimensions=(rows, cols). + # This call shape is the one already used in tools/process_registry.py. + proc = PtyProcess.spawn( # type: ignore[union-attr] + list(argv), + cwd=cwd, + env=spawn_env, + dimensions=(rows, cols), + ) + return cls(proc) + + @property + def pid(self) -> int: + return int(self._proc.pid) + + def is_alive(self) -> bool: + if self._closed: + return False + try: + return bool(self._proc.isalive()) + except Exception: + return False + + # -- I/O -------------------------------------------------------------- + + def read(self, timeout: float = 0.2) -> Optional[bytes]: + """Up to 64 KiB of child output. + + Returns bytes, ``b""`` when nothing is available this tick, or + ``None`` once the child has exited (EOF). + """ + if self._closed: + return None + try: + data = self._proc.read(65536) # pywinpty returns str + except EOFError: + return None + except Exception: + return None + if not data: + # No fd to select on; poll politely so the executor thread + # doesn't pin a core while the TUI is idle. + time.sleep(min(timeout, 0.02)) + return b"" + if isinstance(data, bytes): + return data + # NOTE: pywinpty decodes internally, so a multibyte UTF-8 sequence + # can in theory split across reads. xterm.js tolerates the rare + # replacement char; this is the one fidelity tradeoff vs the POSIX + # raw-fd path. + return data.encode("utf-8", errors="replace") + + def write(self, data: bytes) -> None: + if self._closed or not data: + return + try: + # The dashboard sends raw keystroke bytes; pywinpty.write wants text. + self._proc.write(data.decode("utf-8", errors="replace")) + except Exception: + return + + def resize(self, cols: int, rows: int) -> None: + if self._closed: + return + cols = _clamp(cols, _MAX_COLS) + rows = _clamp(rows, _MAX_ROWS) + try: + self._proc.setwinsize(rows, cols) # pywinpty: (rows, cols) + except Exception: + pass + + # -- teardown --------------------------------------------------------- + + def close(self) -> None: + if self._closed: + return + self._closed = True + try: + self._proc.terminate(force=True) + except Exception: + pass + + def __enter__(self) -> "WinPtyBridge": + return self + + def __exit__(self, *_exc) -> None: + self.close() diff --git a/scripts/release.py b/scripts/release.py index 3bbff845078..12ad6ed0937 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -1490,6 +1490,8 @@ AUTHOR_MAP = { "leonard@sellem.me": "leonardsellem", # PR #37405 (desktop WS origin guard on remote/Tailscale binds) "42903577+ohMyJason@users.noreply.github.com": "ohMyJason", # PR #29810 (discover_models in custom_providers section 4) "singhsanidhya741@gmail.com": "sanidhyasin", # PR #40403 salvage (model.default_headers for custom OpenAI-compatible providers, #40033) + "josephjohnson.joel@gmail.com": "JoelJJohnson", # PR #39913 salvage (Windows ConPTY dashboard chat bridge) + "andreas@schwarz-ketsch.de": "Nea74", # PR #40022 co-author credit (same Windows ConPTY bridge design) } diff --git a/tests/hermes_cli/test_web_server_pty_import.py b/tests/hermes_cli/test_web_server_pty_import.py new file mode 100644 index 00000000000..8a11f77195d --- /dev/null +++ b/tests/hermes_cli/test_web_server_pty_import.py @@ -0,0 +1,83 @@ +"""Test the platform-branched PTY bridge import in hermes_cli.web_server. + +The /api/pty WebSocket handler in web_server.py picks its bridge at import +time via ``sys.platform.startswith("win")`` — Windows gets the ConPTY +backend, POSIX gets the fcntl/termios one. Both branches must: + + 1. Expose ``PtyBridge`` as the bridge class (or None) and + ``PtyUnavailableError`` as an exception class. + 2. Set ``_PTY_BRIDGE_AVAILABLE`` correctly. + 3. Never raise at import time when the platform-native dependency is + missing — the dashboard's non-chat tabs must keep loading. + +This test asserts the live state on whichever platform CI runs on, plus a +source-text check confirming the branch shape is preserved so a future +refactor can't accidentally collapse it back to a POSIX-only import. +""" + +from __future__ import annotations + +import sys + +import pytest + +from hermes_cli import web_server + + +def test_web_server_exposes_pty_bridge_symbols(): + """The two symbols /api/pty consumes must always exist.""" + assert hasattr(web_server, "PtyBridge") + assert hasattr(web_server, "PtyUnavailableError") + assert hasattr(web_server, "_PTY_BRIDGE_AVAILABLE") + # PtyUnavailableError is always an exception class — either the real + # one from the platform bridge, or the local fallback class. + assert isinstance(web_server.PtyUnavailableError, type) + assert issubclass(web_server.PtyUnavailableError, BaseException) + + +@pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows-only") +def test_web_server_uses_win_pty_bridge_on_windows(): + """On native Windows, web_server.PtyBridge must be the ConPTY backend.""" + from hermes_cli.win_pty_bridge import WinPtyBridge + + assert web_server.PtyBridge is WinPtyBridge + assert web_server._PTY_BRIDGE_AVAILABLE is True + # And the error class must be the one from the same module so isinstance + # checks in /api/pty's spawn fallback path actually work. + from hermes_cli.win_pty_bridge import PtyUnavailableError as WinErr + + assert web_server.PtyUnavailableError is WinErr + + +@pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX-only") +def test_web_server_uses_posix_pty_bridge_on_posix(): + """On POSIX, the bridge must be the fcntl/termios PtyBridge.""" + from hermes_cli.pty_bridge import PtyBridge as PosixBridge + from hermes_cli.pty_bridge import PtyUnavailableError as PosixErr + + assert web_server.PtyBridge is PosixBridge + assert web_server._PTY_BRIDGE_AVAILABLE is True + assert web_server.PtyUnavailableError is PosixErr + + +def test_pty_bridge_import_block_is_platform_branched(): + """Source-level guard: a future refactor must not collapse the branch + back to a single POSIX import. Reads web_server.py directly so this + fails the same way on every OS — the runtime symbol checks above can + pass even when the branch shape is wrong on the current platform.""" + src = pytest.importorskip("inspect").getsource(web_server) + # The shape we expect (from PR #39913): + # + # if sys.platform.startswith("win"): + # try: + # from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, ... + # except ImportError: + # PtyBridge = None + # ... + # else: + # try: + # from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError + # ... + assert 'sys.platform.startswith("win")' in src or "sys.platform.startswith('win')" in src + assert "from hermes_cli.win_pty_bridge import" in src + assert "from hermes_cli.pty_bridge import" in src diff --git a/tests/hermes_cli/test_win_pty_bridge.py b/tests/hermes_cli/test_win_pty_bridge.py new file mode 100644 index 00000000000..a7f97b693b1 --- /dev/null +++ b/tests/hermes_cli/test_win_pty_bridge.py @@ -0,0 +1,315 @@ +"""Unit tests for hermes_cli.win_pty_bridge — ConPTY spawning + byte forwarding. + +Windows-only counterpart to tests/hermes_cli/test_pty_bridge.py. Drives +``WinPtyBridge`` with minimal Windows processes (``cmd.exe``, ``python -c …``) +to verify it behaves like a PTY you can read/write/resize/close, then a small +set of platform-fallback assertions (``is_available``, ``PtyUnavailableError``) +that run on every OS so the import surface stays exercised in CI. + +The bridge is the ConPTY backend behind the dashboard ``/chat`` tab — see +``hermes_cli/web_server.py`` ``/api/pty`` handler — so these tests are the +unit-level half of the integration check that the dashboard chat pane is +actually live on native Windows. +""" + +from __future__ import annotations + +import os +import sys +import time + +import pytest + +# WinPtyBridge can be imported on every platform — ``is_available`` just +# returns False when pywinpty isn't usable. Importing the module itself +# must never raise, otherwise the web_server import branch becomes a trap. +from hermes_cli.win_pty_bridge import PtyUnavailableError, WinPtyBridge + +windows_only = pytest.mark.skipif( + not sys.platform.startswith("win"), + reason="ConPTY bridge is Windows-only", +) + + +def _read_until(bridge: WinPtyBridge, needle: bytes, timeout: float = 10.0) -> bytes: + """Accumulate PTY output until we see ``needle`` or time out. + + Mirrors the helper in test_pty_bridge.py so failures look familiar. + """ + deadline = time.monotonic() + timeout + buf = bytearray() + while time.monotonic() < deadline: + chunk = bridge.read(timeout=0.2) + if chunk is None: + break + buf.extend(chunk) + if needle in buf: + return bytes(buf) + return bytes(buf) + + +# --------------------------------------------------------------------------- +# Cross-platform fallback semantics +# --------------------------------------------------------------------------- + + +class TestWinPtyBridgeUnavailable: + """Module-level surface that must stay importable on every OS so the + web_server platform branch doesn't blow up at import time when pywinpty + is missing or the host isn't Windows.""" + + def test_error_is_importable_and_carries_message(self): + err = PtyUnavailableError("conpty missing") + assert "conpty" in str(err) + + def test_bridge_class_is_importable(self): + # The platform-branched import in web_server.py relies on this: + # from hermes_cli.win_pty_bridge import WinPtyBridge, PtyUnavailableError + # Both symbols must always exist; ``is_available()`` is the gate. + assert WinPtyBridge is not None + assert callable(WinPtyBridge.is_available) + + @pytest.mark.skipif(sys.platform.startswith("win"), reason="non-Windows only") + def test_spawn_raises_unavailable_off_windows(self): + with pytest.raises(PtyUnavailableError): + WinPtyBridge.spawn(["true"]) + + +# --------------------------------------------------------------------------- +# Windows-only end-to-end behaviour +# --------------------------------------------------------------------------- + + +@windows_only +class TestWinPtyBridgeSpawn: + def test_is_available_on_windows(self): + assert WinPtyBridge.is_available() is True + + def test_spawn_returns_bridge_with_pid(self): + bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"]) + try: + assert bridge.pid > 0 + finally: + bridge.close() + + def test_spawn_raises_on_missing_argv0(self, tmp_path): + # pywinpty wraps CreateProcessW failures; surface as OSError / RuntimeError. + bogus = str(tmp_path / "definitely-not-a-real-binary.exe") + with pytest.raises((FileNotFoundError, OSError, RuntimeError, PtyUnavailableError)): + WinPtyBridge.spawn([bogus]) + + +@windows_only +class TestWinPtyBridgeIO: + def test_reads_child_stdout(self): + bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "echo hermes-ok"]) + try: + output = _read_until(bridge, b"hermes-ok") + assert b"hermes-ok" in output + finally: + bridge.close() + + def test_write_sends_to_child_stdin(self): + # python -c reads stdin, echoes a marker, exits. More reliable than + # ``cat`` (not on Windows) and doesn't depend on a particular shell. + script = ( + "import sys; " + "line = sys.stdin.readline().strip(); " + "sys.stdout.write('GOT:' + line + '\\n'); " + "sys.stdout.flush()" + ) + bridge = WinPtyBridge.spawn([sys.executable, "-c", script]) + try: + bridge.write(b"hello-pty\r\n") + output = _read_until(bridge, b"GOT:hello-pty") + assert b"GOT:hello-pty" in output + finally: + bridge.close() + + def test_write_after_close_is_silent(self): + bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"]) + bridge.close() + # Must not raise — the dashboard WebSocket reader sometimes writes + # a final keystroke after the user has already closed the tab. + bridge.write(b"ignored") + + def test_read_returns_none_after_child_exits(self): + bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "echo done"]) + try: + _read_until(bridge, b"done") + # Give the child a beat to exit, then drain until EOF. + deadline = time.monotonic() + 5.0 + while bridge.is_alive() and time.monotonic() < deadline: + bridge.read(timeout=0.1) + got_none = False + for _ in range(20): + if bridge.read(timeout=0.1) is None: + got_none = True + break + assert got_none, "WinPtyBridge.read did not return None after child EOF" + finally: + bridge.close() + + +@windows_only +class TestWinPtyBridgeResize: + def test_resize_does_not_raise_on_live_child(self): + # ConPTY exposes no ioctl-equivalent for reading the child's current + # winsize from Python land, so we can't verify the new dimensions + # the way the POSIX test does (which reads TIOCGWINSZ). What we + # CAN guarantee is what the dashboard depends on: ``resize`` never + # raises, the bridge stays alive, and subsequent I/O still works. + bridge = WinPtyBridge.spawn( + [sys.executable, "-c", "import time; time.sleep(1.0)"], + cols=80, + rows=24, + ) + try: + bridge.resize(cols=123, rows=45) + assert bridge.is_alive() + finally: + bridge.close() + + def test_resize_clamps_garbage_dimensions(self): + # Mirror the POSIX clamp test: a broken winsize probe must never + # propagate to the ConPTY API. 131072 > unsigned short max — the + # bridge has to coerce it down without raising. + bridge = WinPtyBridge.spawn( + [sys.executable, "-c", "import time; time.sleep(1.0)"], + cols=80, + rows=24, + ) + try: + bridge.resize(cols=131072, rows=1) # must not raise + bridge.resize(cols=0, rows=-5) # nor this + assert bridge.is_alive() + finally: + bridge.close() + + def test_resize_after_close_is_silent(self): + bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"]) + bridge.close() + # Must not raise — closed bridges still receive late resize escapes + # from xterm.js when the browser tab is closed mid-stream. + bridge.resize(cols=100, rows=40) + + +@windows_only +class TestClampDimension: + """The clamp helper is the load-bearing piece — the dashboard sends + untrusted winsize values straight from xterm.js, and pywinpty's + setwinsize will happily raise on out-of-range u16 values.""" + + def test_clamps_above_max(self): + from hermes_cli.win_pty_bridge import _MAX_COLS, _MAX_ROWS, _clamp + + assert _clamp(131072, _MAX_COLS) == _MAX_COLS + assert _clamp(131072, _MAX_ROWS) == _MAX_ROWS + + def test_floors_at_one(self): + from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp + + assert _clamp(0, _MAX_COLS) == 1 + assert _clamp(-5, _MAX_COLS) == 1 + + def test_passes_through_sane_values(self): + from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp + + assert _clamp(80, _MAX_COLS) == 80 + assert _clamp(2000, _MAX_COLS) == 2000 + + def test_non_numeric_falls_back_to_min(self): + from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp + + assert _clamp(None, _MAX_COLS) == 1 # type: ignore[arg-type] + assert _clamp("not-a-number", _MAX_COLS) == 1 # type: ignore[arg-type] + assert _clamp(float("nan"), _MAX_COLS) == 1 # type: ignore[arg-type] + assert _clamp(float("inf"), _MAX_COLS) == 1 # type: ignore[arg-type] + + +@windows_only +class TestWinPtyBridgeClose: + def test_close_is_idempotent(self): + bridge = WinPtyBridge.spawn( + [sys.executable, "-c", "import time; time.sleep(30)"] + ) + bridge.close() + bridge.close() # must not raise + assert not bridge.is_alive() + + def test_close_terminates_long_running_child(self): + bridge = WinPtyBridge.spawn( + [sys.executable, "-c", "import time; time.sleep(30)"] + ) + pid = bridge.pid + assert bridge.is_alive(), f"child pid {pid} not alive before close" + bridge.close() + # The bridge itself reports liveness via pywinpty.isalive(), which is + # the same probe the dashboard PTY reader uses to decide when to stop + # forwarding bytes — verifying that flips to False is the contract + # that matters for /api/pty. + deadline = time.monotonic() + 5.0 + while bridge.is_alive() and time.monotonic() < deadline: + time.sleep(0.1) + assert not bridge.is_alive(), ( + f"WinPtyBridge.is_alive() still True after close(); pid {pid}" + ) + + +@windows_only +class TestWinPtyBridgeEnv: + def test_cwd_is_respected(self, tmp_path): + bridge = WinPtyBridge.spawn( + [sys.executable, "-c", "import os; print(os.getcwd())"], + cwd=str(tmp_path), + ) + try: + # Path is case-insensitive on Windows; compare lowercased. + needle_resolved = str(tmp_path.resolve()).lower().encode() + deadline = time.monotonic() + 5.0 + buf = bytearray() + while time.monotonic() < deadline: + chunk = bridge.read(timeout=0.2) + if chunk is None: + break + buf.extend(chunk) + if needle_resolved in bytes(buf).lower(): + break + assert needle_resolved in bytes(buf).lower(), ( + f"cwd {tmp_path!s} not echoed by child; got {bytes(buf)!r}" + ) + finally: + bridge.close() + + def test_env_is_forwarded(self): + bridge = WinPtyBridge.spawn( + [ + sys.executable, + "-c", + "import os; print('HERMES_PTY_TEST=' + os.environ.get('HERMES_PTY_TEST',''))", + ], + env={**os.environ, "HERMES_PTY_TEST": "pty-env-works"}, + ) + try: + output = _read_until(bridge, b"pty-env-works") + assert b"pty-env-works" in output + finally: + bridge.close() + + def test_spawn_defaults_term_when_not_set(self): + # The bridge should set TERM=xterm-256color when the caller's env + # doesn't already carry one — xterm.js expects ANSI/SGR sequences. + env = {k: v for k, v in os.environ.items() if k.upper() != "TERM"} + bridge = WinPtyBridge.spawn( + [ + sys.executable, + "-c", + "import os; print('TERM=' + os.environ.get('TERM',''))", + ], + env=env, + ) + try: + output = _read_until(bridge, b"TERM=") + assert b"TERM=xterm-256color" in output + finally: + bridge.close()