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

@@ -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

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)

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

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():

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