diff --git a/tests/hermes_cli/test_web_server.py b/tests/hermes_cli/test_web_server.py index 75837f635f0..1afe572832f 100644 --- a/tests/hermes_cli/test_web_server.py +++ b/tests/hermes_cli/test_web_server.py @@ -5455,22 +5455,6 @@ class TestPtyWebSocket: assert env["HERMES_TUI_INLINE"] == "1" assert env["HERMES_TUI_DISABLE_MOUSE"] == "1" - def test_resolve_chat_argv_sets_active_session_file_env(self, monkeypatch): - """Dashboard chat gives the TUI a breadcrumb file for reconnect resume.""" - import hermes_cli.main as main_mod - - monkeypatch.setattr( - main_mod, - "_make_tui_argv", - lambda project_root, tui_dev=False: (["node", "dist/entry.js"], "/tmp/ui-tui"), - ) - - _argv, _cwd, env = self.ws_module._resolve_chat_argv( - active_session_file="/tmp/hermes-active-session.json" - ) - - assert env["HERMES_TUI_ACTIVE_SESSION_FILE"] == "/tmp/hermes-active-session.json" - def test_resolve_chat_argv_applies_terminal_backend_config( self, monkeypatch, _isolate_hermes_home ): @@ -5772,75 +5756,6 @@ class TestPtyWebSocket: pass assert captured.get("resume") == "sess-42" - def test_channel_reconnect_resumes_active_session_file(self, monkeypatch): - """A new /api/pty socket on the same channel resumes the last TUI sid.""" - script = ( - "import json, os, sys; " - "resume = os.environ.get('HERMES_TUI_RESUME', ''); " - "active = os.environ.get('HERMES_TUI_ACTIVE_SESSION_FILE', ''); " - "sys.stdout.write(f'resume={resume}\\n'); sys.stdout.flush(); " - "active and not resume and open(active, 'w').write(json.dumps({'session_id': 'sess-live'}))" - ) - - def fake_resolve(resume=None, sidecar_url=None, profile=None, active_session_file=None): - env = {} - if active_session_file: - env["HERMES_TUI_ACTIVE_SESSION_FILE"] = active_session_file - if resume: - env["HERMES_TUI_RESUME"] = resume - return ([sys.executable, "-c", script], None, env) - - monkeypatch.setattr(self.ws_module, "_resolve_chat_argv", fake_resolve) - - def drain_until(conn, needle: bytes) -> bytes: - buf = b"" - import time - - deadline = time.monotonic() + 5.0 - while time.monotonic() < deadline: - try: - frame = conn.receive_bytes() - except Exception: - break - if frame: - buf += frame - if needle in buf: - break - return buf - - with self.client.websocket_connect(self._url(channel="reconnect-chan")) as conn: - assert b"resume=" in drain_until(conn, b"resume=") - - with self.client.websocket_connect(self._url(channel="reconnect-chan")) as conn: - assert b"resume=sess-live" in drain_until(conn, b"resume=sess-live") - - def test_fresh_param_ignores_channel_active_session_file(self, monkeypatch): - """Explicit fresh starts must not resurrect the prior channel session.""" - channel = "fresh-chan" - active_file = self.ws_module._active_session_file_for_channel( - self.ws_module.app, - channel, - ) - active_file.write_text(json.dumps({"session_id": "sess-old"}), encoding="utf-8") - captured: dict = {} - - def fake_resolve(resume=None, sidecar_url=None, profile=None, active_session_file=None): - captured["resume"] = resume - captured["active_session_file"] = active_session_file - return (["/bin/sh", "-c", "printf fresh-ok"], None, None) - - monkeypatch.setattr(self.ws_module, "_resolve_chat_argv", fake_resolve) - - with self.client.websocket_connect(self._url(channel=channel, fresh="1")) as conn: - try: - conn.receive_bytes() - except Exception: - pass - - assert captured["resume"] is None - assert captured["active_session_file"] == str(active_file) - assert not active_file.exists() - def test_channel_param_propagates_sidecar_url(self, monkeypatch): """When /api/pty is opened with ?channel=, the PTY child gets a HERMES_TUI_SIDECAR_URL env var pointing back at /api/pub on the diff --git a/tests/hermes_cli/test_web_server_pty_reconnect.py b/tests/hermes_cli/test_web_server_pty_reconnect.py new file mode 100644 index 00000000000..4f7878f53d0 --- /dev/null +++ b/tests/hermes_cli/test_web_server_pty_reconnect.py @@ -0,0 +1,130 @@ +"""Focused tests for dashboard PTY reconnect breadcrumbs.""" + +import json +import sys +from pathlib import Path +from urllib.parse import urlencode + +import pytest + + +pytestmark = pytest.mark.skipif( + sys.platform.startswith("win"), reason="PTY bridge is POSIX-only" +) + + +class _OneFrameBridge: + def __init__(self): + self._sent = False + + @classmethod + def spawn(cls, *args, **kwargs): + return cls() + + def read(self, timeout): + if not self._sent: + self._sent = True + return b"ready" + return None + + def resize(self, *, cols, rows): + pass + + def write(self, raw): + pass + + def close(self): + pass + + +@pytest.fixture +def pty_client(monkeypatch, _isolate_hermes_home): + from starlette.testclient import TestClient + + import hermes_cli.web_server as ws + + monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True) + monkeypatch.setattr(ws.PtyBridge, "spawn", _OneFrameBridge.spawn) + ws.app.state.pty_active_session_files = {} + + client = TestClient(ws.app) + return ws, client, ws._SESSION_TOKEN + + +def _url(token: str, **params: str) -> str: + return f"/api/pty?{urlencode({'token': token, **params})}" + + +def test_resolve_chat_argv_sets_active_session_file_env(monkeypatch): + """Dashboard chat gives the TUI a breadcrumb file for reconnect resume.""" + import hermes_cli.main as main_mod + import hermes_cli.web_server as ws + + monkeypatch.setattr( + main_mod, + "_make_tui_argv", + lambda project_root, tui_dev=False: (["node", "dist/entry.js"], "/tmp/ui-tui"), + ) + + _argv, _cwd, env = ws._resolve_chat_argv( + active_session_file="/tmp/hermes-active-session.json" + ) + + assert env["HERMES_TUI_ACTIVE_SESSION_FILE"] == "/tmp/hermes-active-session.json" + + +def test_channel_reconnect_resumes_active_session_file(pty_client, monkeypatch): + """A new /api/pty socket on the same channel resumes the last TUI sid.""" + ws, client, token = pty_client + captured = [] + + def fake_resolve(resume=None, sidecar_url=None, profile=None, active_session_file=None): + captured.append( + { + "active_session_file": active_session_file, + "resume": resume, + "sidecar_url": sidecar_url, + } + ) + if active_session_file and not resume: + Path(active_session_file).write_text( + json.dumps({"session_id": "sess-live"}), + encoding="utf-8", + ) + return (["fake-hermes-tui"], None, None) + + monkeypatch.setattr(ws, "_resolve_chat_argv", fake_resolve) + + with client.websocket_connect(_url(token, channel="reconnect-chan")) as conn: + assert conn.receive_bytes() == b"ready" + + with client.websocket_connect(_url(token, channel="reconnect-chan")) as conn: + assert conn.receive_bytes() == b"ready" + + assert captured[0]["resume"] is None + assert captured[0]["active_session_file"] + assert captured[1]["resume"] == "sess-live" + assert captured[1]["active_session_file"] == captured[0]["active_session_file"] + + +def test_fresh_param_ignores_channel_active_session_file(pty_client, monkeypatch): + """Explicit fresh starts must not resurrect the prior channel session.""" + ws, client, token = pty_client + channel = "fresh-chan" + active_file = ws._active_session_file_for_channel(ws.app, channel) + active_file.write_text(json.dumps({"session_id": "sess-old"}), encoding="utf-8") + captured = {} + + def fake_resolve(resume=None, sidecar_url=None, profile=None, active_session_file=None): + captured["active_session_file"] = active_session_file + captured["resume"] = resume + return (["fake-hermes-tui"], None, None) + + monkeypatch.setattr(ws, "_resolve_chat_argv", fake_resolve) + + with client.websocket_connect(_url(token, channel=channel, fresh="1")) as conn: + assert conn.receive_bytes() == b"ready" + + assert captured["resume"] is None + assert captured["active_session_file"] == str(active_file) + assert not active_file.exists()