From 87fe88e377911048beb67e07be1111e1bc2f1a41 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Wed, 14 Oct 2026 22:46:51 +0000 Subject: [PATCH] fix(cli): redial a hung-up gateway from every gateway-backed surface A hang-up set GatewayConnection.closed, and every reader of it treated the connection as permanently gone: /sessions, /resume, /branch and /handoff reported session history unavailable, /bg and /btw fell back to an in-process agent, the dock and /stop lost the process view, and shared() built a second connection. A new `final` property means only "the viewer closed it or the first connect failed"; those readers now key off it and let the next call redial. end_session redials before detaching, so /new in that window still sends end_reason. A side task's wait() now fails if the connection redialed since admission, instead of waiting forever. start() makes every caller wait out a dial in progress, and a request that never reached a socket gets NOT_SENT instead of the "outcome unknown, do not resend" copy. Refs: #138624 (review by teknium1, findings 2 and 9), #134889 (N2) --- hermes_cli/gateway_connection.py | 71 ++++++--- hermes_cli/gateway_session_store.py | 16 +- hermes_cli/gateway_side.py | 13 +- hermes_cli/gateway_tools.py | 4 +- .../test_gateway_connection_reconnect.py | 147 ++++++++++++++++++ 5 files changed, 222 insertions(+), 29 deletions(-) diff --git a/hermes_cli/gateway_connection.py b/hermes_cli/gateway_connection.py index d61776099f..85e3de7161 100644 --- a/hermes_cli/gateway_connection.py +++ b/hermes_cli/gateway_connection.py @@ -14,6 +14,7 @@ import asyncio import logging import queue import threading +import time from hermes_cli.gateway_client import DEFAULT_RPC_TIMEOUT, GatewayClientError @@ -21,6 +22,8 @@ logger = logging.getLogger(__name__) DISCONNECTED = ("Gateway disconnected; turn outcome is unknown. Resume the printed session ID to check " "whether it completed or was interrupted; do not resend the work.") +# A request refused before it reached any socket: nothing is in doubt, so never the copy above. +NOT_SENT = "Gateway disconnected before the request was sent; nothing was sent." # Resume refusals meaning the session itself is gone for this viewer (gateway/session_controls). _GONE = frozenset({"not_found", "permission_denied", "profile_mismatch"}) @@ -89,6 +92,9 @@ class GatewayConnection: self.events: queue.Queue = queue.Queue() self.states: dict[str, SessionState] = {} self.client = None + # (client, epoch) of the open socket, swapped as one: a caller never pairs a socket with + # another life's epoch (``call_tagged``). + self._live = None self.loop = None self._thread = None self._ready = threading.Event() @@ -108,10 +114,16 @@ class GatewayConnection: @classmethod def shared(cls): with cls._shared_lock: - if cls._shared is None or cls._shared.closed: + if cls._shared is None or cls._shared.final: cls._shared = cls() return cls._shared + @property + def final(self) -> bool: + """No call dials again: the viewer closed it, or its first connect failed. ``closed`` is also + true while the gateway is hung up, which the next ``call()``/``start()`` redials.""" + return self._detached or (self._error is not None and not self.epoch) + def state(self, session_id) -> SessionState: state = self.states.get(session_id) if state is None: @@ -120,23 +132,36 @@ class GatewayConnection: # -- lifecycle --------------------------------------------------------------------------- def start(self, timeout=90): - with self._start_lock: - if self._thread is not None: - if self._thread.is_alive() or self._detached or (self._error is not None and not self.epoch): + deadline = time.monotonic() + timeout + while True: + with self._start_lock: + if self.final: if self._error is not None: raise self._error return - # The gateway hung up, or the last redial failed: dial again (``connect_gateway`` - # starts a gateway when none is running) and resubscribe this viewer's sessions. - self._error = None - self._ready.clear() - self.closed = False - self._thread = threading.Thread(target=self._run, name="hermes-gateway-loop", daemon=True) - self._thread.start() - if not self._ready.wait(timeout): - raise GatewayClientError("Gateway connection timed out; no local fallback") - if self._error is not None: - raise self._error + thread = self._thread + if thread is None or not thread.is_alive(): + from agent.memory_provider import spawn_context_thread + + if thread is not None: + # The gateway hung up, or the last redial failed: dial again (``connect_gateway`` + # starts a gateway when none is running) and resubscribe this viewer's sessions. + self._error = None + self._ready.clear() + self.closed = False + thread = self._thread = spawn_context_thread(self._run, name="hermes-gateway-loop") + thread.start() + # Every caller waits out a dial in progress, its own or another thread's. + if not self._ready.wait(max(deadline - time.monotonic(), 0)): + raise GatewayClientError("Gateway connection timed out; no local fallback") + if self._error is not None: + raise self._error + if self.client is not None or self.final: + return + # Hung up right after that dial: let its loop end, then dial again. + thread.join(max(deadline - time.monotonic(), 0)) + if time.monotonic() >= deadline: + raise GatewayClientError("Gateway connection timed out; no local fallback") def _run(self): try: @@ -162,6 +187,7 @@ class GatewayConnection: await self._resubscribe(client) self.epoch += 1 self.client = client + self._live = (client, self.epoch) self._ready.set() pump = asyncio.create_task(self._pump(client)) stop = asyncio.create_task(self._stop.wait()) @@ -176,6 +202,7 @@ class GatewayConnection: self._put_loss(GatewayClientError(DISCONNECTED)) finally: self.client = None + self._live = None def _put_loss(self, error): error.connection_epoch = self.epoch @@ -260,16 +287,22 @@ class GatewayConnection: return True def call(self, method, *, timeout=None, **params): + return self.call_tagged(method, timeout=timeout, **params)[0] + + def call_tagged(self, method, *, timeout=None, **params): + """``(result, epoch)``: the epoch of the socket that carried the request, so work it admitted + is tied to that gateway life even if another caller redials before this one reads ``epoch``.""" if self._thread is not None and threading.current_thread() is self._thread: raise RuntimeError("GatewayConnection.call() on the gateway loop thread would deadlock") self.start() - client, loop = self.client, self.loop - if client is None or loop is None: - raise GatewayClientError(DISCONNECTED) + live, loop = self._live, self.loop + if live is None or loop is None: + raise GatewayClientError(NOT_SENT) + client, epoch = live budget = DEFAULT_RPC_TIMEOUT if timeout is None else timeout future = asyncio.run_coroutine_threadsafe(client.rpc(method, _timeout=budget, **params), loop) try: - return future.result(budget + 5) + return future.result(budget + 5), epoch except TimeoutError as exc: future.cancel() raise GatewayClientError(f"Gateway did not answer {method}") from exc diff --git a/hermes_cli/gateway_session_store.py b/hermes_cli/gateway_session_store.py index 7e2d8b6893..7990eaf32d 100644 --- a/hermes_cli/gateway_session_store.py +++ b/hermes_cli/gateway_session_store.py @@ -57,7 +57,8 @@ def _zero(*_a, **_k): class GatewaySessionStore: - """Truthy while the gateway connection is usable (main's ``if self._session_db:`` guards).""" + """Truthy until the viewer's gateway connection is final (main's ``if self._session_db:`` guards); + a gateway hang-up stays truthy, and the next call redials.""" def __init__(self, connection=None): self.connection = connection or GatewayConnection.shared() @@ -66,7 +67,7 @@ class GatewaySessionStore: self._reader = None def __bool__(self): - return not self.connection.closed + return not self.connection.final # -- reads ----------------------------------------------------------------------------- def _db(self): @@ -261,8 +262,17 @@ class GatewaySessionStore: """Main's ``end_session(sid, reason)``: detach this viewer with ``end_reason``; the owner ends the idle local row with that reason once no viewer is left (``ended_at`` / ``end_reason``).""" snapshot = self.snapshots.pop(session_id, None) + state = self.connection.states.get(session_id) + reachable = state is not None and state.subscription_id is not None and not self.connection.final + if reachable: + try: + # Hung up: redial first, so the detach names the subscription the reconnect resumed. + self.connection.start() + except GatewayClientError as exc: + logger.debug("redial for session.detach failed: %s", exc) + reachable = False subscription = self._leave(session_id) - if subscription and not self.connection.closed: + if subscription and reachable: params = {"end_reason": reason} if reason else {} try: self.call("session.detach", session_id=session_id, subscription_id=subscription, **params) diff --git a/hermes_cli/gateway_side.py b/hermes_cli/gateway_side.py index 0c5a8098cf..045d2d2eee 100644 --- a/hermes_cli/gateway_side.py +++ b/hermes_cli/gateway_side.py @@ -16,18 +16,20 @@ _POLL_SECONDS = 1.0 def connection_for(cli): - """The CLI's gateway connection when its session is open, else None (not rewired).""" + """The CLI's gateway connection when its session is open, else None (not rewired). A hung-up + connection is returned (its next call redials); only a final one is None.""" from hermes_cli.gateway_tools_notify import gateway_store store = gateway_store(cli) - if store is None or not getattr(cli, "_gateway_session_open", False) or store.connection.closed: + if store is None or not getattr(cli, "_gateway_session_open", False) or store.connection.final: return None return store.connection def start(connection, session_id, verb, text): """Admit the side task on the owner: ``(task_id, wait)``; ``wait()`` -> the answer (raises on failure).""" - result = connection.call(f"prompt.{verb}", session_id=session_id, text=text) or {} - task_id = result.get("task_id") + # The epoch of the socket that admitted it: a redial before the waiter exists must still end it. + result, epoch = connection.call_tagged(f"prompt.{verb}", session_id=session_id, text=text) + task_id = (result or {}).get("task_id") if not task_id: raise RuntimeError(f"prompt.{verb} returned no task id") inbox = connection.side_waiter(task_id) @@ -38,7 +40,8 @@ def start(connection, session_id, verb, text): try: payload = inbox.get(timeout=_POLL_SECONDS) except queue.Empty: - if connection.closed: + # The task lived on the socket that admitted it: hung up, or already redialed. + if connection.closed or connection.epoch != epoch: from hermes_cli.gateway_connection import DISCONNECTED raise RuntimeError(DISCONNECTED) from None continue diff --git a/hermes_cli/gateway_tools.py b/hermes_cli/gateway_tools.py index cb207e99d7..00a1bdcd42 100644 --- a/hermes_cli/gateway_tools.py +++ b/hermes_cli/gateway_tools.py @@ -143,9 +143,9 @@ def mark_dirty(): def process_view(): """The gateway-backed registry, or None when this process runs its own tools (not rewired) or - its gateway connection is gone.""" + its gateway connection is final (a hang-up keeps the view; its next call redials).""" view = _view - return view if view is not None and not view.connection.closed else None + return view if view is not None and not view.connection.final else None def process_registry(): diff --git a/tests/hermes_cli/test_gateway_connection_reconnect.py b/tests/hermes_cli/test_gateway_connection_reconnect.py index c6ecbeae81..b94b73a0cb 100644 --- a/tests/hermes_cli/test_gateway_connection_reconnect.py +++ b/tests/hermes_cli/test_gateway_connection_reconnect.py @@ -5,6 +5,8 @@ GatewayConnection kept its finished loop thread, so after a gateway restart ever gateway. Each prompt then failed with copy that read as a model failure ("Run /model to switch"). """ import asyncio +import threading +import types from contextlib import asynccontextmanager import pytest @@ -49,6 +51,8 @@ class Life: return {"admission_id": "adm-1", "status": "queued", "execution_generation": 7} if method == "prompt.receipt": return {"result": {"final_response": f"reply {self.number}", "completed": True}, "usage": {}} + if method in ("prompt.background", "prompt.btw"): + return {"task_id": f"task-{self.number}"} return {"status": "ok", "life": self.number} @@ -60,9 +64,14 @@ class Gateway: self.refusals = 0 self.drop_next_submit = False self.resume_errors = [] # raised by the next session.resume calls, across lives + self.dialing = threading.Event() + self.hold = None # an Event the next dials wait on before the socket opens @asynccontextmanager async def dial(self): + self.dialing.set() + if self.hold is not None: + await asyncio.to_thread(self.hold.wait, 5) if self.refusals: self.refusals -= 1 raise GatewayClientError("Gateway starting: deadline. Your message was not sent.") @@ -225,6 +234,144 @@ def test_close_after_a_failed_resume_is_still_final(gateway, conn): assert len(gateway.lives) == 2 +def _cli(store): + return types.SimpleNamespace(_session_db=store, _gateway_session_open=True) + + +@pytest.fixture +def tools(conn, monkeypatch): + from hermes_cli import gateway_tools + monkeypatch.setattr(gateway_tools, "_view", None) + gateway_tools.bind(types.SimpleNamespace(connection=conn, session_id=SID)) + return gateway_tools + + +def test_a_hung_up_store_stays_open_and_its_next_call_redials(gateway, conn, store): + """/sessions, /resume, /branch and /handoff gate on the store: a hang-up is not "Hermes can't + open its session history right now"; the next call dials the restarted gateway.""" + gateway.hang_up(conn) + assert store + assert store.call("status") == {"status": "ok", "life": 2} + assert store + + +def test_bg_and_btw_after_a_hang_up_run_on_the_gateway_not_in_this_process(gateway, conn, store): + """None here sends /bg down the in-process AIAgent path (a local ``bg_HHMMSS_xxxxxx`` task).""" + from hermes_cli import gateway_side + gateway.hang_up(conn) + assert gateway_side.connection_for(_cli(store)) is conn + task_id, _wait = gateway_side.start(conn, SID, "background", "later") + assert task_id == "task-2" and gateway.lives[1].calls[-1][0] == "prompt.background" + + +def test_a_side_answer_lost_to_a_restart_fails_instead_of_waiting_forever(gateway, conn): + """The task died with the old gateway; another thread's redial must not make the wait endless.""" + from hermes_cli import gateway_side + _task, wait = gateway_side.start(conn, SID, "btw", "q") + gateway.hang_up(conn) + conn.call("status") + outcome = [] + waiter = threading.Thread(target=lambda: outcome.append(pytest.raises(RuntimeError, wait)), daemon=True) + waiter.start() + waiter.join(10) + assert not waiter.is_alive() and "outcome is unknown" in str(outcome[0].value) + + +def test_a_redial_between_admission_and_the_waiter_still_ends_the_wait(gateway, conn, monkeypatch): + """Another caller redials after the old gateway admitted the task but before its waiter exists: + the wait belongs to the admitting life, not to whatever epoch is current by then.""" + from hermes_cli import gateway_side + register = conn.side_waiter + + def redial_first(task_id): + gateway.hang_up(conn) + conn.call("status") + return register(task_id) + + monkeypatch.setattr(conn, "side_waiter", redial_first) + task, wait = gateway_side.start(conn, SID, "btw", "q") + assert task == "task-1" and conn.epoch == 2 + outcome = [] + waiter = threading.Thread(target=lambda: outcome.append(pytest.raises(RuntimeError, wait)), daemon=True) + waiter.start() + waiter.join(10) + assert not waiter.is_alive() and "outcome is unknown" in str(outcome[0].value) + + +def test_the_dock_and_stop_after_a_hang_up_redial(gateway, conn, tools): + gateway.hang_up(conn) + view = tools.process_view() + assert view is not None and tools.process_registry() is view + view.kill_all() + assert [method for method, _ in gateway.lives[1].calls][:2] == ["session.resume", "process.stop"] + + +def test_shared_keeps_a_hung_up_connection_instead_of_opening_a_second(gateway, conn, monkeypatch): + monkeypatch.setattr(GatewayConnection, "_shared", conn) + gateway.hang_up(conn) + assert GatewayConnection.shared() is conn + + +def test_new_after_a_hang_up_redials_and_detaches_with_the_end_reason(gateway, conn, store): + """Dropping the subscription without ``session.detach`` leaves the owner never ending the row.""" + gateway.hang_up(conn) + store.end_session(SID, "new_session") + assert gateway.lives[1].calls == [("session.resume", {"session_id": SID}), ("session.detach", { + "session_id": SID, "subscription_id": "sub-2", "end_reason": "new_session"})] + gateway.hang_up(conn) + conn.call("status") + assert _resumes(gateway.lives[2]) == [] + + +def test_after_close_every_reader_reports_the_connection_gone(gateway, conn, store, tools, monkeypatch): + from hermes_cli import gateway_side + monkeypatch.setattr(GatewayConnection, "_shared", conn) + gateway.hang_up(conn) + conn.close() + assert not store + assert gateway_side.connection_for(_cli(store)) is None + assert tools.process_view() is None + assert GatewayConnection.shared() is not conn + store.end_session(SID, "new_session") + with pytest.raises(GatewayClientError, match="disconnected") as refused: + conn.call("status") + assert "outcome is unknown" not in str(refused.value) # nothing was sent + assert len(gateway.lives) == 1 + + +class _Waiters(threading.Event): + """``_ready`` that opens ``release`` once ``count`` callers are waiting on it.""" + + def __init__(self, release, count): + super().__init__() + self.release, self.count = release, count + + def wait(self, timeout=None): + self.count -= 1 + if self.count <= 0: + self.release.set() + return super().wait(timeout) + + +def test_a_call_racing_a_redial_waits_for_it_and_is_sent(gateway, conn): + """The second caller saw the dialing thread alive and no client yet: "turn outcome is unknown, do + not resend" for a request that was never sent.""" + gateway.hang_up(conn) + gateway.hold, gateway.dialing = threading.Event(), threading.Event() + conn._ready = _Waiters(gateway.hold, 2) + results = [] + first = threading.Thread(target=lambda: results.append(conn.call("status")), daemon=True) + first.start() + assert gateway.dialing.wait(5) + try: + second = conn.call("status") + finally: + gateway.hold.set() + first.join(10) + assert second == {"status": "ok", "life": 2} and results == [second] + assert len(gateway.lives) == 2 + + def test_lost_before(): error = GatewayClientError("Gateway disconnected") assert not lost_before(error, 2) # untagged: a live loss