mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
fix(cli): a send on a dead gateway socket no longer kills the spinner thread
GatewayClient.rpc let websockets' ConnectionClosedError escape from send(): after a keepalive ping timeout the dock's subagent.list poll raised it past callers that catch GatewayClientError, and the bare spinner loop died with it, leaving the spinner and live-work dock frozen for the rest of the session. rpc now reports a failed send as the same unknown-outcome GatewayClientError a lost answer already gets, and the monitor tick skips a beat on a failure instead of ending the thread that runs it. (cherry picked from commit ea136305ac597dcd62bc569426321d9d601034aa) (cherry picked from commit cedb79f9a691f9ef9b8f3d9c04fdfefb63d9fb94) (cherry picked from commit f5c1b54199dc02e9654a6d11ba6ee5e0e2a84cc4)
This commit is contained in:
1 parent
ef75bc7088
commit
7e7d9a043d
4 files changed
+60
-6
No files matched your search
@@ -4,6 +4,7 @@ prompts) and scoped controls; no agent-loop state is changed. Process rows come
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
|
||||
from prompt_toolkit.utils import get_cwidth
|
||||
@@ -12,6 +13,8 @@ from agent.i18n import t
|
||||
from hermes_cli import cli_process_dock as procs
|
||||
from hermes_cli import cli_session_dock as session_rows
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _clip(value, width):
|
||||
text = ' '.join(str(value or '').split())
|
||||
@@ -124,8 +127,13 @@ class SubagentMonitor:
|
||||
if now - self._last_poll < 1:
|
||||
return
|
||||
self._last_poll = now
|
||||
changed = self.refresh()
|
||||
if self._poll_tail() or changed:
|
||||
try:
|
||||
changed = self.refresh()
|
||||
changed = self._poll_tail() or changed
|
||||
except Exception: # health: allow BLE001 -- the spinner thread runs this tick; one raise would end that thread for the session
|
||||
logger.debug("subagent monitor tick failed", exc_info=True)
|
||||
return
|
||||
if changed:
|
||||
if self.app is not None:
|
||||
self.invalidate()
|
||||
else:
|
||||
|
||||
@@ -49,6 +49,10 @@ class GatewayClientError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
_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.")
|
||||
|
||||
|
||||
class GatewayRPCError(GatewayClientError):
|
||||
"""A received owner refusal, distinct from an ambiguous transport failure.
|
||||
|
||||
@@ -196,8 +200,7 @@ class GatewayClient:
|
||||
except (ConnectionClosed, OSError, ValueError, TimeoutError):
|
||||
pass
|
||||
finally:
|
||||
error = GatewayClientError("Gateway disconnected; turn outcome is unknown. "
|
||||
"Resume the printed session ID to check whether it completed or was interrupted; do not resend the work.")
|
||||
error = GatewayClientError(_DISCONNECTED)
|
||||
for future in self.pending.values():
|
||||
if not future.done():
|
||||
future.set_exception(error)
|
||||
@@ -216,8 +219,13 @@ class GatewayClient:
|
||||
future = asyncio.get_running_loop().create_future()
|
||||
self.pending[rid] = future
|
||||
self.events.moved.set() # a reader waiting on the consumer reads on toward this answer
|
||||
from websockets.exceptions import ConnectionClosed
|
||||
try:
|
||||
await self.websocket.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}))
|
||||
try:
|
||||
await self.websocket.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}))
|
||||
except (ConnectionClosed, OSError) as exc:
|
||||
# The frame may have left before the socket died: the same unknown outcome as a lost answer.
|
||||
raise GatewayClientError(_DISCONNECTED) from exc
|
||||
budget = DEFAULT_RPC_TIMEOUT if _timeout is None else _timeout
|
||||
return await asyncio.wait_for(future, None if budget == float("inf") else budget)
|
||||
finally:
|
||||
|
||||
@@ -4,7 +4,7 @@ import json
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.gateway_client import GatewayClient, GatewayRPCError
|
||||
from hermes_cli.gateway_client import GatewayClient, GatewayClientError, GatewayRPCError
|
||||
|
||||
|
||||
class _Socket:
|
||||
@@ -73,3 +73,19 @@ async def test_compress_waits_past_the_default_rpc_budget(monkeypatch):
|
||||
assert result == {'status': 'applied'}
|
||||
# The budget is a client-side knob, never a wire parameter the closed contract would refuse.
|
||||
assert all('_timeout' not in req['params'] for req in sent)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_on_a_dead_socket_is_a_client_error_not_a_raw_websockets_one():
|
||||
"""A keepalive timeout closes the socket under a poller (the dock's ``subagent.list``): the send
|
||||
must surface as the transport's own error, which every caller catches, never as websockets'."""
|
||||
from websockets.exceptions import ConnectionClosedError
|
||||
|
||||
class _Dead(_Socket):
|
||||
async def send(self, raw):
|
||||
raise ConnectionClosedError(None, None)
|
||||
|
||||
async with GatewayClient(_Dead(lambda req: None)) as client:
|
||||
with pytest.raises(GatewayClientError, match='turn outcome is unknown'):
|
||||
await client.rpc('subagent.list', session_id='s')
|
||||
assert client.pending == {}
|
||||
@@ -172,3 +172,25 @@ def test_detail_view_paints_the_cached_tail_and_only_the_tick_asks_the_gateway(g
|
||||
assert view.subagent_tail("sa-0-1", fetch=True) is None
|
||||
assert view.subagent_tail("sa-0-1", fetch=True) is None # the failure is cached for the poll interval
|
||||
assert tails() == 2
|
||||
|
||||
|
||||
def test_a_failing_tick_keeps_the_spinner_thread_alive(gateway, monkeypatch):
|
||||
"""The spinner thread runs ``tick`` in a bare loop: one raise would end it, and with it the
|
||||
spinner and the dock, for the rest of the session. A failed poll skips a beat instead."""
|
||||
from hermes_cli.cli_subagent_monitor import SubagentMonitor
|
||||
agent, _client = gateway
|
||||
repaints = []
|
||||
monitor = SubagentMonitor(SimpleNamespace(agent=agent, _agent_running=True,
|
||||
_invalidate=lambda: repaints.append(1)))
|
||||
view = gateway_tools.process_view()
|
||||
real_rows = view.subagent_rows
|
||||
|
||||
def dead(**_kw):
|
||||
raise OSError("socket gone")
|
||||
monkeypatch.setattr(view, "subagent_rows", dead)
|
||||
monitor.tick() # must not raise
|
||||
assert repaints == []
|
||||
monkeypatch.setattr(view, "subagent_rows", real_rows)
|
||||
monitor._last_poll = 0
|
||||
monitor.tick()
|
||||
assert [r["subagent_id"] for r in monitor.entries] == ["sa-0-1"] and repaints == [1]
|
||||
Reference in new issue
Block a user