"""Tests for tui_gateway JSON-RPC protocol plumbing.""" import io import json import os import subprocess import sys import threading import time import types from unittest.mock import MagicMock, patch from pathlib import Path import pytest _original_stdout = sys.stdout @pytest.fixture(autouse=True) def _restore_stdout(): yield sys.stdout = _original_stdout @pytest.fixture() def server(): # The sys.modules mocks only need to cover the *initial* import — once # tui_gateway.server is cached, they are inert. Keeping them active for # the whole test poisons any module first imported inside a test body: # e.g. hermes_cli.active_sessions would bind the mocked get_hermes_home # (a fixed shared path) forever, leaking active-session registry entries # across every later test in the process. Scope the patch to the import. # # Import server_requests (pure stdlib) and transport BEFORE the window: the patch drops every module first # imported inside it, so otherwise the module server.py binds its sinks on (write/emit/answerable) would vanish # from sys.modules and a test's own ``from tui_gateway import server_requests`` would get a fresh, # unbound copy whose default sinks drop frames and treat every client as answerable. Likewise a test's # ``from tui_gateway.transport import bind_transport`` would bind a fresh module's ContextVar that the # server's ``current_transport()`` never reads (first-in-process test sees ``_stdio_transport`` as caller). import tui_gateway.server_requests # noqa: F401 import tui_gateway.transport # noqa: F401 with patch.dict("sys.modules", { "hermes_constants": MagicMock(get_hermes_home=MagicMock(return_value="/tmp/hermes_test")), "hermes_cli.env_loader": MagicMock(), "hermes_cli.banner": MagicMock(), "hermes_state": MagicMock(), }): import importlib mod = importlib.import_module("tui_gateway.server") # Snapshot the RPC registry: several tests below stub handlers # ("slash.exec", "fast.ping", ...) directly in the module-level dict, # which is shared with every other test file in the process. methods = dict(mod._methods) real_stdout = mod._real_stdout yield mod # Reset module-level state without re-importing. importlib.reload # would re-register the module's atexit hooks (ThreadPoolExecutor # shutdown, _shutdown_sessions); the duplicates race the stderr # buffer at interpreter shutdown and surface as Fatal Python error: # _enter_buffered_busy. Restoring the dicts in place gives the next # test a clean slate. mod._methods.clear() mod._methods.update(methods) mod._real_stdout = real_stdout for sid in list(mod._sessions): mod._close_session_by_id(sid, end_reason="test_cleanup") from tui_gateway import server_requests server_requests.reset_for_tests() mod._live_transports.clear() @pytest.fixture() def capture(server): """Redirect server's real stdout to a StringIO and return (server, buf).""" buf = io.StringIO() server._real_stdout = buf return server, buf # ── JSON-RPC envelope ──────────────────────────────────────────────── def test_unknown_method(server): resp = server.handle_request({"id": "1", "method": "bogus"}) assert resp["error"]["code"] == -32601 @pytest.mark.parametrize("kind", ["legacy", "hard-only", "dynamic-getattr"]) def test_session_interrupt_uses_explicit_stop_compatibility(server, monkeypatch, kind): calls = [] class _Legacy: def interrupt(self): calls.append("legacy") class _HardOnly: def hard_interrupt(self): calls.append("hard") class _Dynamic: def interrupt(self): calls.append("legacy") def __getattr__(self, name): if name == "hard_interrupt": return lambda: calls.append("fabricated-hard") raise AttributeError(name) agent = { "legacy": _Legacy(), "hard-only": _HardOnly(), "dynamic-getattr": _Dynamic(), }[kind] session = { "agent": agent, "history_lock": threading.Lock(), "running": True, "queued_prompt": "later", "session_key": "session-key", "_run_thread": None, } monkeypatch.setattr(server, "_tts_stream_stop", lambda: None) monkeypatch.setattr(server, "_sess_nowait", lambda _params, _rid: (session, None)) monkeypatch.setattr(server, "_sess", lambda _params, _rid: (session, None)) monkeypatch.setattr(server, "_session_uses_compute_host", lambda _session: False) monkeypatch.setattr(server, "_clear_pending", lambda _sid: None) response = server._methods["session.interrupt"]( "stop", {"session_id": "ui-session"} ) assert response["result"]["status"] == "interrupted" assert calls == ["hard" if kind == "hard-only" else "legacy"] # ── write_json ──────────────────────────────────────────────── def test_live_session_payload_replays_pending_approval(server, monkeypatch): """A reattached client receives the approval that was emitted while detached.""" from tools import approval from tools import approval_gateway_wait session = { "agent": types.SimpleNamespace(), "cols": 80, "created_at": 1.0, "history": [], "history_lock": threading.Lock(), "running": True, "session_key": "stored-session", } first = { "choices": ["once", "deny"], "command": "rm -rf /tmp/example", "description": "recursive delete", } second = {"command": "rm -rf /tmp/later", "description": "later"} saved_queue = approval._gateway_queues.pop("stored-session", None) approval._gateway_queues["stored-session"] = [ approval_gateway_wait._ApprovalEntry(first), approval_gateway_wait._ApprovalEntry(second), ] monkeypatch.setattr(server, "_approval_request_payload", lambda data: dict(data or {})) try: payload = server._live_session_payload("runtime-session", session) finally: approval._gateway_queues.pop("stored-session", None) if saved_queue is not None: approval._gateway_queues["stored-session"] = saved_queue assert payload["pending_approval"] is not first replayed = payload["pending_approval"] # request_id is injected by _ApprovalEntry so reconnecting clients can # correlate their approval.respond with the exact queued request. assert replayed.pop("request_id") assert replayed == first def test_live_session_payload_replays_open_requests(server): """A reattached client receives the server→client request still blocking the session (a clarify sent while detached), scoped to the owning runtime session only, as a snapshot not a live reference.""" from tui_gateway import server_requests session = { "agent": types.SimpleNamespace(), "cols": 80, "created_at": 1.0, "history": [], "history_lock": threading.Lock(), "running": True, "session_key": "stored-session", } req = server_requests.ServerRequest("runtime-session", "clarify", {"question": "Which?", "choices": ["a", "b"]}) server_requests._open[req.id] = req try: payload = server._live_session_payload("runtime-session", session) other = server._live_session_payload("other-session", session) finally: server_requests._open.pop(req.id, None) assert payload["open_requests"] == [{"id": req.id, "method": "clarify", "params": {"session_id": "runtime-session", "question": "Which?", "choices": ["a", "b"]}}] assert payload["open_requests"][0]["params"] is not req.params assert "open_requests" not in other # ── _emit ──────────────────────────────────────────────────────────── def test_emit_with_payload(capture): server, buf = capture server._emit("test.event", "s1", {"key": "val"}) msg = json.loads(buf.getvalue()) assert msg["method"] == "event" assert msg["params"]["type"] == "test.event" assert msg["params"]["session_id"] == "s1" assert msg["params"]["payload"]["key"] == "val" # ── Server→client requests (tui_gateway/server_requests.py) ───────── def _frames(buf): return [json.loads(line) for line in buf.getvalue().splitlines()] def _wait_open(server_requests, buf=None, timeout=10.0): """The open request once its frame has been written (registration precedes the write). Generous deadline: the first send() in a process lazily imports the ws/contracts/replay modules, which can take seconds on a loaded CI box when this is the first server request of the run.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: with server_requests._lock: req = next(iter(server_requests._open.values()), None) if req is not None and (buf is None or req.id in buf.getvalue()): return req time.sleep(0.01) raise AssertionError("server request never registered") def test_server_request_round_trip_uses_response_frame(capture): """The backend sends a real JSON-RPC request (string id, method, params.session_id) and the client's response frame — not a method call — resolves it; the response frame itself gets no reply.""" from tui_gateway import server_requests server, buf = capture box = {} thread = threading.Thread(target=lambda: box.__setitem__("r", server._ask("sudo", "s1", {}, timeout=5)), daemon=True) thread.start() req = _wait_open(server_requests, buf) frame = _frames(buf)[-1] assert frame == {"jsonrpc": "2.0", "id": req.id, "method": "sudo", "params": {"session_id": "s1"}} assert req.id.startswith("srq-") assert server.dispatch({"jsonrpc": "2.0", "id": req.id, "result": {"value": "hunter2"}}) is None thread.join(timeout=5) assert box["r"] == "hunter2" with server_requests._lock: assert not server_requests._open @pytest.mark.parametrize("method, qids, settle, expected", [ ("sudo", None, lambda sr, req: sr.resolve_response({"id": req.id, "result": {"value": "yes"}}) is True, {"value": "yes"}), # Batch clarify's lock-based resolution follows the same first-settlement rule. ("clarify", ["q1"], lambda sr, req: sr.lock_answer(req.id, "q1", "yes") == [], {"answers": {"q1": "yes"}, "outcome": "submitted"}), ]) def test_settlement_wins_over_a_later_cancel(capture, method, qids, settle, expected): """A response and cancellation may race; the first settlement owns the result.""" from tui_gateway import server_requests req = server_requests.ServerRequest("s1", method, {}, qids=qids) with server_requests._lock: server_requests._open[req.id] = req assert settle(server_requests, req) assert server_requests.cancel("s1") == 0 assert req.answered is True assert req.result == expected assert req.event.is_set() def test_send_returns_an_answer_committed_after_the_deadline_expired(capture, monkeypatch): """An answer accepted by resolve_response is never reported as a timeout (#112548): the response frame can land after event.wait() gave up and before send() withdraws the request, and the renderer must not get a bogus request.cancel for a card the user just answered.""" from tui_gateway import server_requests cancels: list[dict] = [] monkeypatch.setattr(server_requests, "_emit", lambda event, sid, payload: cancels.append(payload)) real_wait = server_requests.threading.Event.wait def answered_during_the_gap(event, timeout=None): # Deadline expires, then the response frame lands before send() re-enters the lock. expired = real_wait(event, timeout) rid = next(iter(server_requests._open)) assert server_requests.resolve_response({"id": rid, "result": {"value": "yes"}}) return expired monkeypatch.setattr(server_requests.threading.Event, "wait", answered_during_the_gap) assert server_requests.send("sudo", "s1", {}, timeout=0.001) == {"value": "yes"} assert cancels == [] assert not server_requests._open def _silent_ws(): """A WebSocket client build that predates server→client requests: receives the frame, never answers.""" from tui_gateway.ws import WSTransport class _SilentWS(WSTransport): def __init__(self): self._ws, self._loop, self._peer, self._auth_identity = object(), None, "test", None self.frames: list[dict] = [] def write(self, obj): self.frames.append(obj) return True def close(self): pass return _SilentWS() def _ws_session(server, sid, peer): server._sessions[sid] = {"session_key": sid, "transport": peer, "history": [], "history_lock": threading.Lock(), "agent_ready": None} def test_server_request_fails_fast_for_a_ws_client_that_never_advertised(server): """A WebSocket client that never sent ``client.capabilities`` cannot answer, so send() returns the error-response shape (None) at once instead of stalling the agent for the deadline (#112548). The frame is never written; nothing is left open for a reconnect replay.""" from tui_gateway import server_requests peer = _silent_ws() _ws_session(server, "ws-old", peer) t0 = time.monotonic() assert server_requests.send("clarify", "ws-old", {"question": "q?", "choices": None, "multi_select": False}, timeout=5) is None assert time.monotonic() - t0 < 1 assert peer.frames == [] assert server_requests.open_requests("ws-old") == [] settled = [] server_requests.send_async("approval", "ws-old", {"request_id": "r1", "command": "rm", "description": "", "pattern_key": "", "pattern_keys": []}, settled.append) assert settled == [None] def test_server_request_waits_for_a_ws_client_that_advertised(server): """``client.capabilities {server_requests: true}`` on the connection marks it answerable: the frame is written and the wait is a real one (here: answered by the response frame).""" from tui_gateway import server_requests from tui_gateway.transport import bind_transport, reset_transport peer = _silent_ws() _ws_session(server, "ws-new", peer) token = bind_transport(peer) try: response = server.handle_request({"id": 1, "method": "client.capabilities", "params": {"server_requests": True}}) finally: reset_transport(token) assert "clarify" in response["result"]["server_requests"] box = {} thread = threading.Thread(target=lambda: box.setdefault("r", server_requests.send("sudo", "ws-new", {}, timeout=5)), daemon=True) thread.start() req = _wait_open(server_requests) assert peer.frames[-1]["id"] == req.id assert server.dispatch({"jsonrpc": "2.0", "id": req.id, "result": {"value": "yes"}}) is None thread.join(timeout=5) assert box["r"] == {"value": "yes"} # Disconnect forgets the advertisement; the next connection must advertise again. server.unregister_live_transport(peer) assert server_requests.answers_requests(peer) is False def test_server_request_error_response_fails_fast(server): """A client that advertised but has no handler for the method answers -32601: that error frame settles send() to None at once instead of the agent waiting for the deadline.""" from tui_gateway import server_requests box = {} thread = threading.Thread(target=lambda: box.setdefault("result", server_requests.send("sudo", "s1", {}, timeout=5)), daemon=True) thread.start() req = _wait_open(server_requests) t0 = time.monotonic() assert server_requests.resolve_response({"id": req.id, "error": {"code": -32601}}) thread.join(timeout=1) assert not thread.is_alive() and time.monotonic() - t0 < 1 assert box["result"] is None def _two_window_session(server, sid): """A session two advertised Desktop windows are attached to (a FanoutTransport of WS peers).""" import asyncio from tui_gateway import server_requests from tui_gateway.transport import FanoutTransport # The fanout awaits a WS peer's write_async on its loop; a peer without one is pruned on first write. loop = asyncio.new_event_loop() threading.Thread(target=loop.run_forever, daemon=True).start() windows = (_silent_ws(), _silent_ws()) for window in windows: async def _write_async(obj, window=window): return window.write(obj) window._loop, window.write_async = loop, _write_async server_requests.advertise(window, True) _ws_session(server, sid, FanoutTransport(*windows)) return windows def _not_shown(req): from tui_gateway import server_requests return {"jsonrpc": "2.0", "id": req.id, "error": {"code": server_requests.NOT_SHOWN_CODE, "message": "not here"}} def test_window_owned_request_refuses_at_once_when_no_window_shows_the_session(server): """#119333: every attached window says it does not show the session → preview.read resolves with the distinct not-shown refusal now, instead of the agent waiting out the 45s bridge deadline.""" from tui_gateway import server_requests windows = _two_window_session(server, "s-none") box = {} thread = threading.Thread(target=lambda: box.setdefault("r", server._ask("preview.read", "s-none", {}, timeout=30)), daemon=True) thread.start() req = _wait_open(server_requests) t0 = time.monotonic() assert server.dispatch(_not_shown(req), windows[0]) is None # A replayed decline from the same window is still one window's vote. server.dispatch(_not_shown(req), windows[0]) time.sleep(0.05) assert thread.is_alive() server.dispatch(_not_shown(req), windows[1]) thread.join(timeout=2) assert not thread.is_alive() and time.monotonic() - t0 < 2 assert json.loads(box["r"]) == {"success": False, "error": server_requests.NOT_SHOWN_MESSAGE} assert server_requests.open_requests("s-none") == [] def test_window_owned_request_decline_does_not_beat_the_owner_window(server): """#113348 stays fixed: one bystander window declining leaves the request open for the window that shows the session, and its answer is the result.""" from tui_gateway import server_requests bystander, owner = _two_window_session(server, "s-own") box = {} thread = threading.Thread(target=lambda: box.setdefault("r", server._ask("preview.read", "s-own", {}, timeout=5)), daemon=True) thread.start() req = _wait_open(server_requests) server.dispatch(_not_shown(req), bystander) # An unattributed decline (proxied / relayed frame) is not a window's vote either. assert server_requests.resolve_response(_not_shown(req)) is True time.sleep(0.05) assert thread.is_alive() and [r["id"] for r in server_requests.open_requests("s-own")] == [req.id] server.dispatch({"jsonrpc": "2.0", "id": req.id, "result": {"value": '{"text": "page"}'}}, owner) thread.join(timeout=2) assert box["r"] == '{"text": "page"}' def test_client_capabilities_advertises_counting_not_shown_declines(server): from tui_gateway.transport import bind_transport, reset_transport peer = _silent_ws() token = bind_transport(peer) try: response = server.handle_request({"id": 1, "method": "client.capabilities", "params": {"server_requests": True}}) finally: reset_transport(token) assert response["result"]["declines_not_shown"] is True @pytest.mark.parametrize("method", ["secret", "sudo", "terminal.read", "tour"]) def test_server_request_timeout_emits_one_request_cancel(capture, method): from tui_gateway import server_requests server, buf = capture assert server_requests.send(method, "s1", {}, timeout=0) is None request, cancel = _frames(buf) assert request["method"] == method assert cancel["params"]["type"] == "request.cancel" assert cancel["params"]["session_id"] == "s1" assert cancel["params"]["payload"] == {"id": request["id"], "method": method, "reason": "timeout"} def test_late_response_and_lock_are_dropped_quietly(server): """A response for an id that already ended is ignored (no reply, no error); a late clarify.lock answers `expired` instead of a protocol error a card would surface verbatim.""" assert server.dispatch({"jsonrpc": "2.0", "id": "srq-gone", "result": {"value": ""}}) is None response = server.handle_request({"id": "late", "method": "clarify.lock", "params": {"request_id": "srq-gone", "question_id": "q0", "answer": ""}}) assert response["result"] == {"status": "expired"} def _start_batch_clarify(server, buf, qids, timeout=None): from tui_gateway import server_requests box = {} normalized = [{"qid": q, "id": "", "question": q, "choices": None, "choices_offered": [], "multi_select": False} for q in qids] if timeout is not None: server._clarify_timeout_seconds = lambda: timeout thread = threading.Thread( target=lambda: box.__setitem__("answer", server._clarify_block("s1", normalized)), daemon=True) thread.start() return thread, box, _wait_open(server_requests, buf) def test_clarify_batch_locks_resolve_in_order_and_keep_partial_on_timeout(capture): """Per-question locks (clarify.lock) are editable until every qid is locked; the last lock resolves the request with the full answer set. Only wire fields reach the renderer.""" server, buf = capture thread, box, req = _start_batch_clarify(server, buf, ["q0", "q1"]) sent = _frames(buf)[-1]["params"]["questions"][0] assert set(sent) == {"qid", "question", "choices", "multi_select"} first = server.handle_request({"id": "a1", "method": "clarify.lock", "params": {"request_id": req.id, "question_id": "q0", "answer": "x"}}) assert first["result"] == {"status": "ok", "remaining": ["q1"]} redo = server.handle_request({"id": "a1b", "method": "clarify.lock", "params": {"request_id": req.id, "question_id": "q0", "answer": "y"}}) assert redo["result"] == {"status": "ok", "remaining": ["q1"]} bad = server.handle_request({"id": "bad", "method": "clarify.lock", "params": {"request_id": req.id, "question_id": "nope", "answer": ""}}) assert bad["error"]["code"] == 4002 assert server._open_requests("s1")[0]["params"]["answers"] == {"q0": "y"} last = server.handle_request({"id": "a2", "method": "clarify.lock", "params": {"request_id": req.id, "question_id": "q1", "answer": ""}}) assert last["result"] == {"status": "ok", "remaining": []} thread.join(timeout=5) assert box["answer"] == {"answers": {"q0": "y", "q1": ""}, "outcome": "submitted"} # Deadline: locked answers survive, outcome timed_out, one request.cancel. original_timeout = server._clarify_timeout_seconds try: thread, box, req = _start_batch_clarify(server, buf, ["q0", "q1"], timeout=1.5) locked = server.handle_request({"id": "b1", "method": "clarify.lock", "params": {"request_id": req.id, "question_id": "q0", "answer": "kept"}}) assert locked["result"]["status"] == "ok" thread.join(timeout=5) finally: server._clarify_timeout_seconds = original_timeout assert box["answer"] == {"answers": {"q0": "kept"}, "outcome": "timed_out"} cancels = [f for f in _frames(buf) if f.get("method") == "event" and f["params"]["type"] == "request.cancel"] assert [c["params"]["payload"]["id"] for c in cancels] == [req.id] def test_clarify_batch_cancel_all_is_a_response_without_answers(capture): server, buf = capture thread, box, req = _start_batch_clarify(server, buf, ["q0", "q1"]) server.dispatch({"jsonrpc": "2.0", "id": req.id, "result": {}}) thread.join(timeout=5) assert box["answer"] == {"answers": {}, "outcome": "cancelled"} def test_clear_pending_cancels_only_that_session(capture): from tui_gateway import server_requests server, buf = capture a = threading.Thread(target=lambda: server_requests.send("sudo", "sid-a", {}, timeout=None), daemon=True) b = threading.Thread(target=lambda: server_requests.send("sudo", "sid-b", {}, timeout=None), daemon=True) a.start(); b.start() deadline = time.monotonic() + 2 while time.monotonic() < deadline and len(server_requests._open) < 2: time.sleep(0.01) assert server._session_pending_kind("sid-a") == "sudo" server._clear_pending("sid-a") a.join(timeout=2) assert not a.is_alive() and b.is_alive() assert server._session_pending_kind("sid-a") == "" and server._session_pending_kind("sid-b") == "sudo" server._clear_pending() b.join(timeout=2) assert not b.is_alive() reasons = [f["params"]["payload"]["reason"] for f in _frames(buf) if f.get("method") == "event"] assert reasons == ["interrupted", "shutdown"] def test_approval_pending_replays_unresolved_requests(server, monkeypatch): from tools import approval server._sessions["ui-1"] = {"session_key": "agent-1", "history": []} pending = [{"request_id": "req-1", "command": "danger"}] monkeypatch.setattr(approval, "list_gateway_approvals", lambda key: pending if key == "agent-1" else []) response = server.handle_request( {"id": "r1", "method": "approval.pending", "params": {"session_id": "ui-1"}} ) assert response["result"] == {"approvals": pending} def test_approval_received_acknowledges_exact_request(server, monkeypatch): from tools import approval server._sessions["ui-1"] = {"session_key": "agent-1", "history": []} calls = [] monkeypatch.setattr( approval, "ack_gateway_approval", lambda key, request_id: calls.append((key, request_id)) or True, ) response = server.handle_request( { "id": "r2", "method": "approval.received", "params": {"session_id": "ui-1", "request_id": "req-1"}, } ) assert response["result"] == {"acknowledged": True} assert calls == [("agent-1", "req-1")] def test_approval_response_correlates_request_id(server, monkeypatch): from tools import approval server._sessions["ui-1"] = {"session_key": "agent-1", "history": []} calls = [] monkeypatch.setattr( approval, "resolve_gateway_approval", lambda key, choice, **kwargs: calls.append((key, choice, kwargs)) or 1, ) response = server.handle_request( { "id": "r3", "method": "approval.respond", "params": {"session_id": "ui-1", "request_id": "req-1", "choice": "once"}, } ) assert response["result"] == {"resolved": 1} assert calls == [("agent-1", "once", {"resolve_all": False, "request_id": "req-1"})] def test_approval_respond_falls_back_to_request_id_lookup(server, monkeypatch): """A stale live sid must not 4001 an approval answer when the request_id resolves to a live session (durable-identity fallback, #91684).""" from tools import approval live = {"session_key": "agent-live", "history": []} server._sessions["ui-live"] = live calls = [] monkeypatch.setattr( approval, "list_gateway_approvals", lambda key: [{"request_id": "req-91684"}] if key == "agent-live" else [], ) monkeypatch.setattr( approval, "resolve_gateway_approval", lambda key, choice, **kwargs: calls.append((key, choice, kwargs)) or 1, ) response = server.handle_request( { "id": "r-fallback", "method": "approval.respond", "params": { "session_id": "gone-sid", "request_id": "req-91684", "choice": "once", }, } ) assert response["result"] == {"resolved": 1} assert calls == [ ("agent-live", "once", {"resolve_all": False, "request_id": "req-91684"}) ] def test_approval_respond_falls_back_to_stored_session_id(server, monkeypatch): """session_id holding a STORED id maps to the live runtime record.""" from tools import approval live = {"session_key": "stored-91684", "history": []} server._sessions["ui-stored"] = live calls = [] monkeypatch.setattr(approval, "list_gateway_approvals", lambda key: []) monkeypatch.setattr( approval, "resolve_gateway_approval", lambda key, choice, **kwargs: calls.append((key, choice, kwargs)) or 1, ) response = server.handle_request( { "id": "r-stored", "method": "approval.respond", "params": {"session_id": "stored-91684", "choice": "deny"}, } ) assert response["result"] == {"resolved": 1} assert calls == [ ("stored-91684", "deny", {"resolve_all": False, "request_id": None}) ] def test_approval_respond_4001_when_nothing_resolves(server, monkeypatch): from tools import approval monkeypatch.setattr(approval, "list_gateway_approvals", lambda key: []) response = server.handle_request( { "id": "r-nope", "method": "approval.respond", "params": {"session_id": "nope", "request_id": "req-x", "choice": "once"}, } ) assert response["error"]["code"] == 4001 # ── Session lookup ─────────────────────────────────────────────────── # ── session.resume payload ──────────────────────────────────────────── def test_session_resume_returns_hydrated_messages(server, monkeypatch): class _DB: def get_session(self, _sid): return {"id": "20260409_010101_abc123"} def get_session_by_title(self, _title): return None def reopen_session(self, _sid): return None def get_resume_conversations(self, session_id): return ( self.get_messages_as_conversation(session_id, repair_alternation=True), self.get_messages_as_conversation(session_id, include_ancestors=True), ) def get_ancestor_display_prefix(self, _sid): return [] def get_messages_as_conversation(self, _sid, include_ancestors=False, repair_alternation=False): return [ {"role": "user", "content": "hello"}, {"role": "assistant", "content": "yo", "reasoning": "thoughts"}, {"role": "tool", "content": "searched"}, {"role": "assistant", "content": " "}, {"role": "assistant", "content": None}, {"role": "narrator", "content": "skip"}, ] monkeypatch.setattr(server, "_get_db", lambda: _DB()) monkeypatch.setattr(server, "_make_agent", lambda sid, key, session_id=None, session_db=None, **_kwargs: object()) monkeypatch.setattr(server, "_init_session", lambda sid, key, agent, history, cols=80, **_kwargs: None) monkeypatch.setattr(server, "_session_info", lambda _agent, _session=None: {"model": "test/model"}) resp = server.handle_request( { "id": "r1", "method": "session.resume", # eager_build: exercise the synchronous build path (this test # monkeypatches _make_agent/_init_session/_session_info). "params": {"session_id": "20260409_010101_abc123", "cols": 100, "eager_build": True}, } ) assert "error" not in resp assert resp["result"]["message_count"] == 3 assert resp["result"]["messages"] == [ {"role": "user", "text": "hello"}, {"role": "assistant", "text": "yo", "reasoning": "thoughts"}, {"role": "tool", "name": "tool", "context": ""}, ] def test_session_resume_rejects_runaway_transcript_before_history_load( server, monkeypatch ): class _DB: def get_session(self, sid): return {"id": sid, "message_count": 20_001} def get_session_by_title(self, _title): return None def resolve_resume_session_id(self, sid): return sid def reopen_session(self, _sid): raise AssertionError("oversized session must be rejected before reopen") monkeypatch.setattr(server, "_get_db", lambda: _DB()) response = server.handle_request( { "id": "r1", "method": "session.resume", "params": { "session_id": "runaway-session", "omit_messages": True, }, } ) assert response["error"]["code"] == 4130 def test_session_resume_deferred_and_omitted_paths_guard_the_tip_only(server, monkeypatch): """A deep compression lineage behind a small tip must open on Desktop. Desktop's cold resume sends ``defer_history`` + ``omit_messages`` and pages the transcript over REST, so the process only ever holds the tip segment. Counting the whole lineage there returned 4130 for the healthiest sessions (85 compaction segments / ~29k rows / ~700-row tip: Bot Chat stuck on "Waking up…"). The guard must count what each path loads. """ calls = [] class _DB: def get_session(self, sid): return {"id": sid, "message_count": 28_730} def get_session_by_title(self, _title): return None def resolve_resume_session_id(self, sid): return sid def assert_resume_safe(self, sid, max_messages=None, *, tip_only=False): calls.append(tip_only) if not tip_only: from hermes_state import SessionResumeTooLargeError raise SessionResumeTooLargeError(20_001, 20_000) return 666 def reopen_session(self, _sid): raise RuntimeError("stop before history load") monkeypatch.setattr(server, "_get_db", lambda: _DB()) monkeypatch.setattr(server, "_enable_gateway_prompts", lambda: None) for params in ( {"defer_history": True, "omit_messages": True, "source": "desktop"}, {"omit_messages": True}, {"lazy": True}, ): calls.clear() response = server.handle_request( { "id": "r-tip", "method": "session.resume", "params": {"session_id": "deep-lineage", **params}, } ) err = response.get("error") or {} assert err.get("code") != 4130, params assert calls == [True], params # The non-deferred, non-omitted resume materializes the full lineage in # memory, so it keeps the lineage-wide bound. calls.clear() response = server.handle_request( {"id": "r-full", "method": "session.resume", "params": {"session_id": "deep-lineage"}} ) assert response["error"]["code"] == 4130 assert calls == [False] def test_deferred_hydration_falls_back_to_tip_when_lineage_exceeds_limit(server, monkeypatch): """The hydration worker never loads a lineage the guard would refuse.""" import threading from hermes_state import SessionResumeTooLargeError tip = [{"role": "user", "content": "tip"}] reads = [] class _DB: def reopen_session(self, _sid): return True def assert_resume_safe(self, sid, max_messages=None, *, tip_only=False): if not tip_only: raise SessionResumeTooLargeError(20_001, 20_000) return 1 def get_resume_conversations(self, _sid): reads.append("lineage") raise AssertionError("must not materialize the runaway lineage") def get_ancestor_display_prefix(self, _sid): reads.append("prefix") raise AssertionError("must not materialize the runaway lineage") def get_messages_as_conversation(self, sid, **kwargs): reads.append(("tip", kwargs.get("repair_alternation"))) return list(tip) built = threading.Event() monkeypatch.setattr(server, "_start_agent_build", lambda _sid, _session: built.set()) monkeypatch.setattr(server, "_maybe_schedule_auto_continue", lambda *_a, **_k: None) session = server._deferred_session_record( "deep-lineage", cols=80, cwd="/tmp", history=[], lease=None ) session["resume_history_ready"] = threading.Event() session["resume_hydrating"] = True session["resume_message_count"] = 28_730 server._sessions["hyd"] = session try: server._schedule_resume_hydration("hyd", "deep-lineage", _DB()) assert session["resume_history_ready"].wait(timeout=5) assert built.wait(timeout=5) assert session.get("resume_history_error") is None assert session["history"] == tip assert session["display_history_prefix"] == [] assert session["resume_message_count"] == 1 assert reads == [("tip", True)] finally: server._sessions.pop("hyd", None) def test_session_resume_guard_failure_fails_open(server, monkeypatch): """A transient guard error must not block resume (fail open, log only).""" reopened = [] class _DB: def get_session(self, sid): return {"id": sid} def get_session_by_title(self, _title): return None def resolve_resume_session_id(self, sid): return sid def assert_resume_safe(self, _sid): raise RuntimeError("database is locked") def reopen_session(self, sid): reopened.append(sid) return True monkeypatch.setattr(server, "_get_db", lambda: _DB()) response = server.handle_request( { "id": "r-open", "method": "session.resume", "params": { "session_id": "transient-guard-session", "omit_messages": True, }, } ) # The guard must not block: no 4130. Reopen being attempted proves # execution moved past the guard. err = response.get("error") or {} assert err.get("code") != 4130 assert reopened == ["transient-guard-session"] def test_session_resume_active_turn_payload_matches_desktop_fixture(server, monkeypatch): """A live resume serializes the exact timer payload consumed by Desktop.""" fixture = json.loads( (Path(__file__).parents[1] / "fixtures" / "session-resume-active-turn.json").read_text( encoding="utf-8" ) ) class _DB: def get_session(self, session_id): return {"id": session_id} def get_session_by_title(self, _title): return None def resolve_resume_session_id(self, session_id): return session_id active_turn = { "assistant": "partial answer", "started_at": fixture["turn_started_at"], "streaming": True, "user": "current prompt", } server._sessions[fixture["session_id"]] = { "agent": types.SimpleNamespace(session_id=fixture["session_key"]), "created_at": fixture["started_at"], "history": [{"content": "earlier prompt", "role": "user"}], "history_lock": threading.Lock(), "inflight_turn": active_turn, "running": True, "session_key": fixture["session_key"], } monkeypatch.setattr(server, "_get_db", lambda: _DB()) monkeypatch.setattr(server, "_session_info", lambda _agent, _session=None: fixture["info"]) # JSON round-trip the real RPC envelope: the desktop fixture must stay # faithful to what the gateway actually serializes, not a copied shape. response = json.loads( json.dumps( server.handle_request( { "id": "resume-running", "method": "session.resume", "params": {"session_id": fixture["session_key"]}, } ) ) ) result = response["result"] assert result["running"] is True assert result["turn_started_at"] == active_turn["started_at"] assert result == fixture def test_enforce_session_cap_evicts_oldest_detached_only(server, monkeypatch): """The LRU cap frees the least-recently-active DETACHED sessions when over the limit, and never a live-transport / running / mid-build one.""" monkeypatch.setattr(server, "_load_cfg", lambda: {"max_live_sessions": 2}) evicted: list[str] = [] monkeypatch.setattr( server, "_close_session_by_id", lambda sid, end_reason=None, predicate=None: evicted.append(sid), ) def _ready() -> threading.Event: ev = threading.Event() ev.set() return ev detached = server._detached_ws_transport live = object() # no _closed attr -> live transport, never evictable server._sessions.clear() server._sessions.update( { "old_detached": {"transport": detached, "last_active": 100.0, "agent_ready": _ready()}, "new_detached": {"transport": detached, "last_active": 300.0, "agent_ready": _ready()}, "running_detached": { "transport": detached, "last_active": 50.0, "running": True, "agent_ready": _ready(), }, "focused_live": {"transport": live, "last_active": 200.0, "agent_ready": _ready()}, } ) server._enforce_session_cap() # 4 sessions, cap 2 -> evict 2. Only detached+idle+built are eligible, oldest # first; the running one and the live-transport one are exempt. assert evicted == ["old_detached", "new_detached"] @pytest.mark.parametrize("closed_transport", [False, True]) def test_idle_reaper_rearms_missing_ws_orphan_timer(server, monkeypatch, tmp_path, closed_transport): """A detached lane cannot keep its lease forever if initial timer setup was lost.""" from hermes_cli.active_sessions import ( active_session_registry_snapshot, try_acquire_active_session, ) home = tmp_path / ".hermes" monkeypatch.setenv("HERMES_HOME", str(home)) sid = "detached-without-reaper" sibling_sid = "live-sibling" orphan_lease, message = try_acquire_active_session( session_id=sid, surface="desktop", config={}, registry_home=home, track_liveness=True, ) assert orphan_lease is not None and message is None sibling_lease, message = try_acquire_active_session( session_id=sibling_sid, surface="desktop", config={}, registry_home=home, track_liveness=True, ) assert sibling_lease is not None and message is None def _session(session_key, lease, transport): return { "active_session_lease": lease, "created_at": time.time(), "history": [], "history_lock": threading.Lock(), "last_active": time.time(), "session_key": session_key, "source": "tui", "transport": transport, } server._sessions.clear() class ClosedTransport: _closed = True dead_transport = ClosedTransport() if closed_transport else server._detached_ws_transport server._sessions.update({ sid: _session(sid, orphan_lease, dead_transport), sibling_sid: _session(sibling_sid, sibling_lease, object()), }) server._pending_ws_reaps.clear() monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 0.05) monkeypatch.setattr(server, "_SESSION_TTL_S", 3600.0) monkeypatch.setattr(server, "_flush_dirty_sessions", lambda: 0) monkeypatch.setattr(server, "_enforce_session_cap", lambda: None) server._reap_idle_sessions() deadline = time.monotonic() + 2.0 while (sid in server._sessions or not orphan_lease.released) and time.monotonic() < deadline: time.sleep(0.01) assert sid not in server._sessions assert orphan_lease.released is True assert sibling_sid in server._sessions assert [entry["session_id"] for entry in active_session_registry_snapshot(home)] == [sibling_sid] repo_root = Path(__file__).resolve().parents[2] env = os.environ.copy() env["HERMES_HOME"] = str(home) env["PYTHONPATH"] = os.pathsep.join( part for part in (str(repo_root), env.get("PYTHONPATH", "")) if part ) successor = subprocess.run( [ sys.executable, "-c", ( "from hermes_cli.active_sessions import try_acquire_active_session; " f"lease, refusal = try_acquire_active_session(session_id={sid!r}, surface='desktop', " "config={}, track_liveness=True); " "assert lease is not None and refusal is None, refusal; lease.release()" ), ], cwd=repo_root, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30, ) assert successor.returncode == 0, successor.stderr assert [entry["session_id"] for entry in active_session_registry_snapshot(home)] == [sibling_sid] def test_sync_session_key_after_compress_reanchors_active_session_lease( server, monkeypatch, tmp_path ): home = tmp_path / ".hermes" monkeypatch.setenv("HERMES_HOME", str(home)) from hermes_cli.active_sessions import ( active_session_registry_snapshot, try_acquire_active_session, ) lease, message = try_acquire_active_session( session_id="session-old", surface="tui", config={"max_concurrent_sessions": 1}, metadata={"live_session_id": "ui-1"}, ) assert message is None assert lease is not None session = { "active_session_lease": lease, "agent": types.SimpleNamespace(session_id="session-new"), "session_key": "session-old", } fake_approval = types.SimpleNamespace( disable_session_yolo=lambda *_args, **_kwargs: None, enable_session_yolo=lambda *_args, **_kwargs: None, is_session_yolo_enabled=lambda *_args, **_kwargs: False, register_gateway_notify=lambda *_args, **_kwargs: None, unregister_gateway_notify=lambda *_args, **_kwargs: None, ) monkeypatch.setattr(server, "_restart_slash_worker", lambda *_args, **_kwargs: None) with patch.dict(sys.modules, {"tools.approval": fake_approval}): server._sync_session_key_after_compress("ui-1", session) snapshot = active_session_registry_snapshot() assert session["session_key"] == "session-new" assert lease.session_id == "session-new" assert [entry["session_id"] for entry in snapshot] == ["session-new"] lease.release() def test_make_agent_accepts_list_system_prompt(server, monkeypatch): captured = {} class _Agent: def __init__(self, **kwargs): captured.update(kwargs) self.model = kwargs.get("model", "") monkeypatch.setitem(sys.modules, "run_agent", types.SimpleNamespace(AIAgent=_Agent)) monkeypatch.setitem( sys.modules, "hermes_cli.runtime_provider", types.SimpleNamespace( resolve_runtime_provider=lambda **_kwargs: { "provider": "test", "base_url": None, "api_key": None, "api_mode": None, } ), ) monkeypatch.setattr(server, "_load_cfg", lambda: {"agent": {"system_prompt": ["one", "two"]}}) monkeypatch.setattr(server, "_resolve_startup_runtime", lambda: ("test/model", "test")) monkeypatch.setattr(server, "_get_db", lambda: None) server._make_agent("sid", "session-key", session_id="session-key") assert captured["ephemeral_system_prompt"] == "one\ntwo" # ── Config I/O ─────────────────────────────────────────────────────── def test_config_roundtrip(server, tmp_path, monkeypatch): # monkeypatch, not assignment: a bare ``server._hermes_home = tmp_path`` outlives this test and every # later ``_load_cfg()`` in the process reads this file's ``model: test/model`` shorthand. monkeypatch.setattr(server, "_hermes_home", tmp_path) server._save_cfg({"model": "test/model"}) assert server._load_cfg()["model"] == "test/model" # ── _cli_exec_blocked ──────────────────────────────────────────────── @pytest.mark.parametrize("argv", [ [], ["setup"], ["gateway"], ["sessions", "browse"], ["config", "edit"], ]) def test_cli_exec_blocked(server, argv): assert server._cli_exec_blocked(argv) is not None # ── slash.exec skill command interception ──────────────────────────── def test_slash_exec_rejects_skill_commands(server): """slash.exec must reject skill commands so the TUI falls through to command.dispatch.""" # Register a mock session sid = "test-session" server._sessions[sid] = {"session_key": sid, "agent": None} # Mock scan_skill_commands to return a known skill fake_skills = {"/hermes-agent-dev": {"name": "hermes-agent-dev", "description": "Dev workflow"}} with patch("agent.skill_commands.get_skill_commands", return_value=fake_skills): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "hermes-agent-dev", "session_id": sid}, }) # Should return an error so the TUI's .catch() fires command.dispatch assert "error" in resp assert resp["error"]["code"] == 4018 def test_slash_exec_scopes_skill_lookup_to_session_profile(server, tmp_path): """slash.exec must resolve get_skill_commands() against the session's own profile_home rather than the gateway process's ambient HERMES_HOME (#88023). A Desktop session that switches profiles mid-session shares the same gateway process, so a skill declared only under the new profile's skills.external_dirs must still be recognized here — else the command falls through to the slash-worker dead path instead of routing to command.dispatch. """ import agent.skill_commands as sc_mod empty_local_dir = tmp_path / "no-local-skills" empty_local_dir.mkdir() profile_b = tmp_path / "profile_b" external_b = tmp_path / "external_b" profile_b.mkdir() skill_dir = external_b / "b-only" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\nname: b-only\ndescription: Only in profile b.\n---\n\n# b-only\n\nDo the thing.\n" ) (profile_b / "config.yaml").write_text( f"skills:\n external_dirs:\n - {external_b}\n" ) sid = "test-session-profile-b" server._sessions[sid] = { "session_key": sid, "agent": None, "profile_home": str(profile_b), } with ( patch("tools.skills_tool.SKILLS_DIR", empty_local_dir), patch.object(sc_mod, "_skill_commands_by_key", {}), ): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "b-only", "session_id": sid}, }) # The gateway's own HERMES_HOME (the test-isolation tempdir, no # skills.external_dirs) has no "b-only" skill — the only way this # resolves is by scoping the lookup to the session's profile_home. assert "error" in resp assert resp["error"]["code"] == 4018 def test_command_dispatch_expands_stacked_skills_from_temp_home(server, tmp_path, monkeypatch): """#74705: Desktop/TUI command.dispatch must expand every real leading /skill token (CLI cli.py and the messaging gateway already do), so '/nature-figure /academic-plotting Plot the results' loads BOTH skills over the remaining instruction instead of leaving the second token in the prompt as plain text.""" import agent.skill_commands as skill_commands import tools.skills_tool as skills_tool home = tmp_path / ".hermes" skills_dir = home / "skills" for name, instructions in ( ("nature-figure", "Render figures with natural colors."), ("academic-plotting", "Label every axis and include units."), ): skill_dir = skills_dir / name skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( f"---\nname: {name}\ndescription: Test {name}.\n---\n\n{instructions}\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setattr(skills_tool, "SKILLS_DIR", skills_dir) monkeypatch.setattr(skill_commands, "_skill_commands_by_key", {}) sid = "test-session" server._sessions[sid] = {"session_key": sid, "agent": None} resp = server.handle_request({ "id": "stacked-skills", "method": "command.dispatch", "params": { "name": "nature-figure", "arg": "/academic-plotting Plot the results", "session_id": sid, }, }) assert "error" not in resp result = resp["result"] assert result["type"] == "skill" assert result["notice"] == ( "⚡ Loading 2 stacked skills: nature-figure, academic-plotting" ) assert "Render figures with natural colors." in result["message"] assert "Label every axis and include units." in result["message"] assert "Plot the results" in result["message"] # UIs render `display`: the projection shows the invocation the user typed. assert result["display"] == ( "/nature-figure /academic-plotting Plot the results" ) def test_command_dispatch_stacked_split_keeps_unknown_tokens_as_instruction(server, tmp_path, monkeypatch): """A non-skill or repeated token stops the stack and stays instruction text — the split must never eat content the user meant as the prompt.""" import agent.skill_commands as skill_commands import tools.skills_tool as skills_tool home = tmp_path / ".hermes" skills_dir = home / "skills" skill_dir = skills_dir / "nature-figure" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\nname: nature-figure\ndescription: Test.\n---\n\nRender figures.\n", encoding="utf-8", ) monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setattr(skills_tool, "SKILLS_DIR", skills_dir) monkeypatch.setattr(skill_commands, "_skill_commands_by_key", {}) sid = "test-session-unknown" server._sessions[sid] = {"session_key": sid, "agent": None} resp = server.handle_request({ "id": "stacked-unknown", "method": "command.dispatch", "params": { "name": "nature-figure", "arg": "/not-a-skill-command but /model is a registry command", "session_id": sid, }, }) assert "error" not in resp result = resp["result"] assert result["type"] == "skill" # No stacked notice — the single-skill path ran with the post-split text intact. assert "notice" not in result assert "/not-a-skill-command but /model is a registry command" in result["message"] def test_sessionless_slash_palette_follows_profile_param(server, tmp_path, monkeypatch): """A Desktop draft has no session yet: ``commands.catalog`` / ``complete.slash`` must scan the named ``profile``'s home, not the launch profile's — A→B→A under multiplexing (#124651). The palette's quick_commands are that profile's too, and an unknown profile is 4064, not a launch-profile palette.""" import agent.skill_commands as sc_mod from agent.secret_scope import is_multiplex_active, set_multiplex_active root = tmp_path / "hermes_home" for name in ("s6probe-a", "s6probe-b"): skill_dir = tmp_path / f"external_{name}" / f"{name}-only" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text(f"---\nname: {name}-only\ndescription: Only in {name}.\n---\n\n# x\n") (root / "profiles" / name).mkdir(parents=True) (root / "profiles" / name / "config.yaml").write_text( f"skills:\n external_dirs:\n - {skill_dir.parent}\n" f"quick_commands:\n {name}-qc:\n type: exec\n command: echo {name}\n") monkeypatch.setenv("HERMES_HOME", str(root)) monkeypatch.setattr(server, "_hermes_home", str(root)) def palette(profile): catalog = server.handle_request({"id": "r1", "method": "commands.catalog", "params": {"profile": profile}}) typed = server.handle_request({ "id": "r2", "method": "complete.slash", "params": {"text": "/s6probe", "profile": profile}}) assert "result" in catalog and "result" in typed, (catalog, typed) quick = {key for key, _ in catalog["result"]["pairs"] if key.endswith("-qc")} return set(catalog["result"]["skills"]), {item["text"].strip("/") for item in typed["result"]["items"]}, quick previous = is_multiplex_active() set_multiplex_active(True) try: with ( patch("tools.skills_tool.SKILLS_DIR", tmp_path / "no-local-skills"), patch.object(sc_mod, "_skill_commands_by_key", {}), ): assert palette("s6probe-a") == ({"/s6probe-a-only"}, {"s6probe-a-only"}, {"/s6probe-a-qc"}) assert palette("s6probe-b") == ({"/s6probe-b-only"}, {"s6probe-b-only"}, {"/s6probe-b-qc"}) assert palette("s6probe-a") == ({"/s6probe-a-only"}, {"s6probe-a-only"}, {"/s6probe-a-qc"}) for method in ("commands.catalog", "skills.reload"): bad = server.handle_request({"id": "r3", "method": method, "params": {"profile": "../x"}}) assert bad.get("error", {}).get("code") == 4064, (method, bad) finally: set_multiplex_active(previous) class _BannerWorker: """Stand-in for the slash worker's current skill path: ok-reply the banner.""" def __init__(self): self.calls = [] self.closed = False def run(self, command): self.calls.append(command) return "⚡ Loading skill: grilling" def close(self): self.closed = True def _grilling_profile(tmp_path): """A session whose profile has only the grilling skill, plus a banner worker.""" empty_local_dir = tmp_path / "no-local-skills" empty_local_dir.mkdir() profile = tmp_path / "profile" profile.mkdir() external = tmp_path / "external" skill_dir = external / "grilling" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\nname: grilling\ndescription: Grill the plan.\n---\n\n# grilling\n\nAsk hard questions.\n" ) (profile / "config.yaml").write_text(f"skills:\n external_dirs:\n - {external}\n") sid = "skill-failopen-session" worker = _BannerWorker() return empty_local_dir, { "session_key": sid, "agent": None, "profile_home": str(profile), "slash_worker": worker, }, worker def _assert_not_ok_banner(resp, worker): blob = json.dumps(resp) assert "Loading skill" not in blob assert worker.calls == [] result = resp.get("result") or {} if result.get("type") == "skill": assert result.get("message") assert result.get("name") == "grilling" return assert "error" in resp assert resp["error"]["code"] != 0 def test_slash_exec_skill_scan_raise_returns_dispatch_payload_not_banner(server, tmp_path): """A skill-scan exception must not fail open into an ok loading banner. The client gets command.dispatch's skill payload (the expanded prompt), never a silent success that drops it. """ import agent.skill_commands as sc_mod empty_local_dir, session, worker = _grilling_profile(tmp_path) sid = session["session_key"] server._sessions[sid] = session real_get = sc_mod.get_skill_commands calls = {"n": 0} def flaky(): calls["n"] += 1 if calls["n"] == 1: raise OSError("external_dirs hiccup") return real_get() with ( patch("tools.skills_tool.SKILLS_DIR", empty_local_dir), patch.object(sc_mod, "get_skill_commands", flaky), patch.object(sc_mod, "_skill_commands_by_key", {}), ): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "grilling tighten this", "session_id": sid}, }) _assert_not_ok_banner(resp, worker) assert resp["result"]["type"] == "skill" assert "tighten this" in resp["result"]["message"] def test_slash_exec_skill_scan_raise_is_hard_error_not_banner_when_dispatch_misses(server, tmp_path): """If the scan keeps failing, slash.exec still must not ok-reply the banner.""" import agent.skill_commands as sc_mod empty_local_dir, session, worker = _grilling_profile(tmp_path) sid = session["session_key"] server._sessions[sid] = session def always_raise(): raise OSError("external_dirs hiccup") with ( patch("tools.skills_tool.SKILLS_DIR", empty_local_dir), patch.object(sc_mod, "get_skill_commands", always_raise), patch.object(sc_mod, "_skill_commands_by_key", {}), ): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "/grilling", "session_id": sid}, }) _assert_not_ok_banner(resp, worker) assert "error" in resp def test_slash_exec_skill_scan_raise_still_runs_registry_commands(server): """A skill-scan exception must not block built-ins the worker owns.""" import agent.skill_commands as sc_mod class _StatusWorker: def __init__(self): self.calls = [] def run(self, command): self.calls.append(command) return "verbose ok" def close(self): pass sid = "registry-during-skill-scan-failure" worker = _StatusWorker() server._sessions[sid] = {"session_key": sid, "agent": None, "slash_worker": worker} with patch.object(sc_mod, "get_skill_commands", side_effect=OSError("external_dirs hiccup")): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "/verbose", "session_id": sid}, }) assert worker.calls == ["/verbose"] assert resp.get("result", {}).get("output") == "verbose ok" assert "error" not in resp def test_slash_exec_worker_skill_refuse_returns_dispatch_payload(server, tmp_path): """A worker that refuses a skill before process_command must not become a 5030 drop. The gate can miss (stale empty scan) while dispatch still resolves the skill. The client gets that payload, and the worker stays up. """ import agent.skill_commands as sc_mod empty_local_dir, session, worker = _grilling_profile(tmp_path) sid = session["session_key"] class _RefuseWorker(_BannerWorker): def run(self, command): self.calls.append(command) raise RuntimeError("skill command refused before process: /grilling") worker = _RefuseWorker() session["slash_worker"] = worker server._sessions[sid] = session real_get = sc_mod.get_skill_commands calls = {"n": 0} def stale_then_real(): calls["n"] += 1 if calls["n"] == 1: return {} return real_get() with ( patch("tools.skills_tool.SKILLS_DIR", empty_local_dir), patch.object(sc_mod, "get_skill_commands", stale_then_real), patch.object(sc_mod, "_skill_commands_by_key", {}), ): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "/grilling", "session_id": sid}, }) assert worker.closed is False assert resp.get("result", {}).get("type") == "skill" assert resp["result"].get("message") assert "Loading skill" not in json.dumps(resp) def test_command_dispatch_scopes_skill_lookup_to_session_profile(server, tmp_path): """command.dispatch must load a skill that exists only in the session profile.""" import agent.skill_commands as sc_mod empty_local_dir = tmp_path / "no-local-skills" empty_local_dir.mkdir() profile_b = tmp_path / "profile_b" external_b = tmp_path / "external_b" profile_b.mkdir() skill_dir = external_b / "b-only" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\nname: b-only\ndescription: Only in profile b.\n---\n\n# b-only\n\nDo the thing.\n" ) (profile_b / "config.yaml").write_text( f"skills:\n external_dirs:\n - {external_b}\n" ) sid = "test-session-profile-b-dispatch" server._sessions[sid] = { "session_key": sid, "agent": None, "profile_home": str(profile_b), } with ( patch("tools.skills_tool.SKILLS_DIR", empty_local_dir), patch.object(sc_mod, "_skill_commands_by_key", {}), ): resp = server.handle_request({ "id": "r1", "method": "command.dispatch", "params": {"name": "b-only", "arg": "with an argument", "session_id": sid}, }) assert "error" not in resp assert resp["result"]["type"] == "skill" assert resp["result"]["name"] == "b-only" def test_slash_exec_routes_a_secondary_only_bundle_to_dispatch(server, tmp_path, monkeypatch): """A skill bundle that exists only under the session profile's ``skill-bundles/`` must be resolved (and routed to command.dispatch) against that profile, not the launch home (#110695).""" import agent.skill_bundles as sb_mod import agent.skill_commands as sc_mod monkeypatch.delenv("HERMES_BUNDLES_DIR", raising=False) profile_b = tmp_path / "profile_b" external_b = tmp_path / "external_b" for name in ("one", "two"): d = external_b / name d.mkdir(parents=True) (d / "SKILL.md").write_text(f"---\nname: {name}\ndescription: {name}.\n---\n\n# {name}\n") (profile_b / "skill-bundles").mkdir(parents=True) (profile_b / "skill-bundles" / "b-pack.yaml").write_text("name: b-pack\nskills: [one, two]\n") (profile_b / "config.yaml").write_text(f"skills:\n external_dirs:\n - {external_b}\n") sid = "test-session-profile-b-bundle" server._sessions[sid] = {"session_key": sid, "agent": None, "profile_home": str(profile_b)} with ( patch("tools.skills_tool.SKILLS_DIR", tmp_path / "no-local-skills"), patch.object(sb_mod, "_bundles_cache", {}), patch.object(sb_mod, "_bundles_cache_mtime", None), patch.object(sc_mod, "_skill_commands_by_key", {}), ): resp = server.handle_request({ "id": "r1", "method": "slash.exec", "params": {"command": "/b-pack go", "session_id": sid}}) assert "error" not in resp, resp assert resp["result"]["type"] == "send" and "b-pack" in resp["result"]["notice"] def test_command_dispatch_queue_sends_message(server): """command.dispatch /queue returns {type: 'send', message: ...} for the TUI.""" sid = "test-session" server._sessions[sid] = {"session_key": sid} resp = server.handle_request({ "id": "r1", "method": "command.dispatch", "params": {"name": "queue", "arg": "tell me about quantum computing", "session_id": sid}, }) assert "error" not in resp result = resp["result"] assert result["type"] == "send" assert result["message"] == "tell me about quantum computing" # ── dispatch(): pool routing for long handlers (#12546) ────────────── def test_dispatch_runs_short_handlers_inline(server): """Non-long handlers return their response synchronously from dispatch().""" server._methods["fast.ping"] = lambda rid, params: server._ok(rid, {"pong": True}) resp = server.dispatch({"id": "r1", "method": "fast.ping", "params": {}}) assert resp == {"jsonrpc": "2.0", "id": "r1", "result": {"pong": True}} @pytest.mark.parametrize( "slow_method", ["complete.path", "complete.slash", "voice.toggle", "voice.record", "voice.tts", "wake.start", "wake.status"], ) def test_slow_handlers_run_off_the_reader_thread(slow_method, server, monkeypatch): """dispatch() must hand these RPCs to the pool and return at once, so a stalled handler never blocks the stdin/WS reader behind it. Completion (#21123: git ls-files / skill scan froze prompt.submit for the 120s RPC timeout) and voice/wake (synchronous faster-whisper lazy install, up to 300s: sent messages never reached the agent) are the same bug class as #50005.""" release = threading.Event() written = [] class _Transport: def write(self, obj): written.append(obj) return True def close(self): pass ran_on = [] def _stalled(rid, params): ran_on.append(threading.get_ident()) release.wait(timeout=10) return server._ok(rid, {}) monkeypatch.setitem(server._methods, slow_method, _stalled) t0 = time.monotonic() resp = server.dispatch({"id": "slow", "method": slow_method, "params": {}}, _Transport()) elapsed = time.monotonic() - t0 release.set() assert resp is None, f"{slow_method} ran inline on the reader thread" assert elapsed < 5 deadline = time.monotonic() + 10 while not written and time.monotonic() < deadline: time.sleep(0.01) # The pool worker, not the reader, ran the handler and wrote its response frame. assert written and written[0]["id"] == "slow", written assert ran_on and ran_on[0] != threading.get_ident() def test_skin_live_switch_end_to_end(server, tmp_path, monkeypatch): """Real config + skin files: activating a skin (as `hermes config set` does) makes the per-tool reconcile broadcast skin.changed with the resolved palette. Exercises _load_cfg → _skin_sig → resolve_skin → _emit with no mocks in between.""" import hermes_cli.skin_engine as skin_engine (tmp_path / "skins").mkdir() (tmp_path / "skins" / "midnight.yaml").write_text( "name: midnight\ndescription: t\ncolors:\n banner_title: '#00ffcc'\n background: '#001010'\ncustomCSS: |\n .chat-input { font-size: 16px; }\n" ) monkeypatch.setattr(skin_engine, "get_hermes_home", lambda: tmp_path) monkeypatch.setattr(server, "_hermes_home", tmp_path) monkeypatch.setattr(server, "_last_skin_sig", None, raising=False) server._cfg_cache = server._cfg_sig = server._cfg_path = None emitted = [] monkeypatch.setattr(server, "_stdio_is_rpc_channel", True) # stdio TUI: no WS peers, stdout is the client monkeypatch.setattr(server, "_emit", lambda ev, sid, payload=None: emitted.append((ev, payload))) # Baseline (default) — seeds the signature. (tmp_path / "config.yaml").write_text("display:\n skin: default\n", encoding="utf-8") server._broadcast_skin_if_changed() emitted.clear() # Activate midnight, as `hermes config set display.skin midnight` would. time.sleep(0.01) # ensure the config mtime moves (tmp_path / "config.yaml").write_text("display:\n skin: midnight\n", encoding="utf-8") server._broadcast_skin_if_changed() assert [ev for ev, _ in emitted] == ["skin.changed"] assert emitted[0][1]["name"] == "midnight" assert emitted[0][1]["colors"]["banner_title"] == "#00ffcc" # customCSS rides the same payload end-to-end: parsed from the YAML, # stripped, and emitted by resolve_skin(). assert emitted[0][1]["customCSS"] == ".chat-input { font-size: 16px; }" def test_broadcast_skin_if_changed_on_any_signature_move(server, monkeypatch): """A skin the agent changes mid-turn goes live once per real move: a name switch (incl. switch-then-revert) OR an in-place color edit to the active skin (same name, new file mtime). An unchanged signature never re-broadcasts.""" emitted = [] # switch, no-op, switch, then a color edit (same name, bumped mtime). sigs = iter([("neon", 1.0), ("neon", 1.0), ("forest", 1.0), ("forest", 2.0)]) monkeypatch.setattr(server, "_stdio_is_rpc_channel", True) # stdio TUI: no WS peers, stdout is the client monkeypatch.setattr(server, "_emit", lambda ev, sid, payload=None: emitted.append((ev, payload))) monkeypatch.setattr(server, "_last_skin_sig", None, raising=False) monkeypatch.setattr(server, "_skin_sig", lambda: next(sigs)) monkeypatch.setattr(server, "resolve_skin", lambda: {"name": "x", "colors": {}}) for _ in range(4): server._broadcast_skin_if_changed() assert [ev for ev, _ in emitted] == ["skin.changed"] * 3 # ── global-event broadcast (session-less events reach every WS client) ── class _RecordingTransport: """Minimal Transport stand-in that records the frames written to it.""" def __init__(self) -> None: self.frames: list[dict] = [] def write(self, obj: dict) -> bool: self.frames.append(obj) return True def close(self) -> None: pass def test_unregister_live_transport_stops_delivery(capture, monkeypatch): """A disconnected peer (unregistered in the ws finally block) receives nothing — and a stale write is never attempted against its closed socket.""" server, buf = capture monkeypatch.setattr(server, "_stdio_is_rpc_channel", True) # stdio TUI process a = _RecordingTransport() server.register_live_transport(a) server.unregister_live_transport(a) server._broadcast_global_event("skin.changed", {"name": "x"}) assert a.frames == [] # No live transports left → fell back to stdio (the stdio TUI's JSON-RPC channel). assert json.loads(buf.getvalue())["params"]["type"] == "skin.changed" def test_approval_for_a_ws_client_that_never_advertised_settles_the_queue_entry(server, monkeypatch): """The approval wait is owned by ``tools.approval``'s queue, not by ``server_requests``. When the request cannot be sent (the only client predates server→client requests) the queue entry must be withdrawn too, otherwise ``_await_gateway_decision`` idles for the whole approvals.timeout with no prompt anywhere (#112548). The decision is a withdrawal (``cancelled`` cause), never a user deny.""" from tools import approval as approval_mod from tools import approval_gateway_wait as wait_mod peer = _silent_ws() _ws_session(server, "ws-old-approval", peer) monkeypatch.setattr(wait_mod._ctx, "_get_approval_timeout", lambda: 3) monkeypatch.setattr(wait_mod._ctx, "_fire_approval_hook", lambda name, **kw: None) approval_mod.register_gateway_notify("ws-old-approval", lambda data: server._emit_approval_request("ws-old-approval", data)) try: t0 = time.monotonic() decision = wait_mod._await_gateway_decision( "ws-old-approval", approval_mod._gateway_notify_cbs["ws-old-approval"], {"command": "rm -rf build", "description": "", "pattern_key": "dangerous", "pattern_keys": ["dangerous"]}) waited = time.monotonic() - t0 finally: approval_mod.unregister_gateway_notify("ws-old-approval") assert waited < 1, decision assert decision["choice"] is None and decision["cancelled"] assert peer.frames == [] assert "ws-old-approval" not in approval_mod._gateway_queues def test_approval_that_ends_before_its_settle_hook_attaches_is_still_withdrawn(server): """The client can answer (``approval.respond`` RPC, or another surface) between the frame going out and the settle hook attaching; ``register_gateway_settle`` then reports the entry gone. The sent request must still be withdrawn, or every later ``session.resume`` replays a prompt nobody is waiting on.""" from tui_gateway import server_requests from tui_gateway.transport import bind_transport, reset_transport peer = _silent_ws() _ws_session(server, "ws-raced", peer) token = bind_transport(peer) try: server.handle_request({"id": 1, "method": "client.capabilities", "params": {"server_requests": True}}) finally: reset_transport(token) try: # No queue entry carries this request id: the wait already ended when the hook tries to attach. server._emit_approval_request("ws-raced", {"command": "rm -rf build", "description": "", "request_id": "appr-already-resolved"}) assert len(peer.frames) == 2, f"the sent approval was never withdrawn: {peer.frames}" sent, cancel = peer.frames assert sent["method"] == "approval" assert cancel["params"]["type"] == "request.cancel" assert cancel["params"]["payload"]["id"] == sent["id"] assert server_requests.open_requests("ws-raced") == [] finally: server.unregister_live_transport(peer) def test_peerless_global_broadcast_never_reaches_stdout_in_ws_backend(capture, monkeypatch): """`hermes serve` / dashboard speak JSON-RPC over WS only; Desktop captures their stdout into desktop.log. After the last WS client leaves, the change watcher keeps ticking — its sessions.changed / setup.ready / session.reclaimed frames must be dropped, not printed (~1100 `[hermes] {"jsonrpc": ...}` lines in desktop.log).""" server, buf = capture monkeypatch.setattr(server, "_stdio_is_rpc_channel", False, raising=False) a = _RecordingTransport() server.register_live_transport(a) server._broadcast_global_event("sessions.changed", {}) server.unregister_live_transport(a) server._broadcast_global_event("sessions.changed", {}) assert len(a.frames) == 1 assert buf.getvalue() == "" # ── session.create idempotency (#65410 / PR #65411) ─────────────────── def _stub_session_create_dependencies(server, monkeypatch): """Stub out the heavy deps ``session.create`` touches so it can run without a real agent/DB. ``session.create`` lives in the split ``methods_session`` module but its handlers read server globals (``bind_module`` rebinds them), so patching the server module covers both.""" # session.create's handler reads SERVER globals (bind_module rebinds the # split module's functions onto server vars), so patching the server module # covers everything the create path touches. monkeypatch.setattr(server, "_register_session_cwd", lambda *a, **kw: None) monkeypatch.setattr(server, "_schedule_agent_build", lambda *a, **kw: None) monkeypatch.setattr(server, "_schedule_session_cap_enforcement", lambda *a, **kw: None) monkeypatch.setattr(server, "_enable_gateway_prompts", lambda: None) monkeypatch.setattr(server, "_load_show_reasoning", lambda: False) monkeypatch.setattr(server, "_load_tool_progress_mode", lambda: None) monkeypatch.setattr(server, "_profile_home", lambda p: None) monkeypatch.setattr(server, "_profile_build_scope", _null_scope) monkeypatch.setattr(server, "_seed_row", lambda *a, **kw: None) monkeypatch.setattr(server, "_seed_branch_row", lambda *a, **kw: None) monkeypatch.setattr(server, "_ensure_session_db_row", lambda *a, **kw: None) # Fresh registry per test: the server and the rebound handler share the # same dict object only if we swap it in place. server._idempotency_keys.clear() class _NullScope: def __enter__(self): return self def __exit__(self, *exc): return False def _null_scope(profile_home): return _NullScope() def test_session_create_idempotency_key_dedupes_retry(server, monkeypatch): """A retried session.create with the same idempotency_key returns the SAME sid instead of spawning a duplicate child (#65410): a create whose first response was lost must not leave two children behind.""" _stub_session_create_dependencies(server, monkeypatch) params = { "cols": 96, "source": "desktop", "messages": [{"role": "user", "content": "branch me"}], "parent_session_id": "parent-1", "idempotency_key": "branch-retry-abc", } first = server.handle_request({"id": "c1", "method": "session.create", "params": dict(params)}) assert "error" not in first, first.get("error") first_sid = first["result"]["session_id"] assert first_sid assert len(server._sessions) == 1 # Client retries after a lost response: same key, same params. second = server.handle_request({"id": "c2", "method": "session.create", "params": dict(params)}) assert "error" not in second, second.get("error") assert second["result"]["session_id"] == first_sid assert len(server._sessions) == 1 def test_session_create_no_idempotency_key_creates_distinct_sessions(server, monkeypatch): """Without an idempotency_key, repeated creates keep the historic behavior.""" _stub_session_create_dependencies(server, monkeypatch) first = server.handle_request({"id": "c1", "method": "session.create", "params": {"cols": 96, "source": "desktop"}}) second = server.handle_request({"id": "c2", "method": "session.create", "params": {"cols": 96, "source": "desktop"}}) assert "error" not in first and "error" not in second assert first["result"]["session_id"] != second["result"]["session_id"] assert len(server._sessions) == 2 def test_session_create_idempotency_key_expires_with_session(server, monkeypatch): """If the original session closed between create and retry, the same key falls through and creates a fresh session (the key does not pin a dead sid).""" _stub_session_create_dependencies(server, monkeypatch) first = server.handle_request( {"id": "c1", "method": "session.create", "params": {"cols": 96, "source": "desktop", "idempotency_key": "branch-retry-xyz"}} ) first_sid = first["result"]["session_id"] server._sessions.pop(first_sid, None) second = server.handle_request( {"id": "c2", "method": "session.create", "params": {"cols": 96, "source": "desktop", "idempotency_key": "branch-retry-xyz"}} ) assert "error" not in second assert second["result"]["session_id"] != first_sid assert len(server._sessions) == 1 def test_session_branch_stored_accepts_idempotency_key(server, monkeypatch): """The desktop's whole-session branch (session.branch_stored) rides the same create plumbing and sends idempotency_key on EVERY branch (#65410): the contract must accept it, the create must succeed (the lineage-sidebar e2e failed with a 4000 because SessionBranchStoredParams forbade the key), and a retried branch_stored with the SAME key must return the SAME child instead of a duplicate.""" _stub_session_create_dependencies(server, monkeypatch) class _Scope: def __init__(self, db): self.db = db def __enter__(self): return self.db def __exit__(self, *_args): return False class _FakeDB: def get_resume_conversations(self, key): assert key == "parent" return [], [ {"role": "user", "content": "first question", "timestamp": 1}, {"role": "assistant", "content": "first answer", "timestamp": 2}, ] monkeypatch.setattr(server, "_profile_db", lambda _params: _Scope(_FakeDB())) params = { "cols": 96, "parent_session_id": "parent", "source": "desktop", "idempotency_key": "branch-stored-retry-abc", } first = server.handle_request({"id": "b1", "method": "session.branch_stored", "params": dict(params)}) assert "error" not in first, first.get("error") first_sid = first["result"]["session_id"] assert len(server._sessions) == 1 # A lost-response retry: same key, same params, same child. second = server.handle_request({"id": "b2", "method": "session.branch_stored", "params": dict(params)}) assert "error" not in second, second.get("error") assert second["result"]["session_id"] == first_sid assert len(server._sessions) == 1