diff --git a/hermes_cli/gateway_client.py b/hermes_cli/gateway_client.py index 1eaf0d48a0..4e88b54993 100644 --- a/hermes_cli/gateway_client.py +++ b/hermes_cli/gateway_client.py @@ -51,6 +51,8 @@ class GatewayClientError(ValueError): 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 DISCONNECTED. +NOT_SENT = "Gateway disconnected before the request was sent; nothing was sent." class GatewayRPCError(GatewayClientError): @@ -214,6 +216,10 @@ class GatewayClient: """Send one request and await its answer; ``_timeout`` (seconds, never sent on the wire) overrides ``DEFAULT_RPC_TIMEOUT`` for a verb that legitimately takes longer (``math.inf``: no deadline; the answer, a hang-up or the caller's cancel ends the wait).""" + if self.reader is not None and self.reader.done(): + # The reader ended (the socket closed, e.g. a keepalive timeout, or it gave up): no answer + # could arrive, so nothing is sent and the caller may safely send again. + raise GatewayClientError(NOT_SENT) self.sequence += 1 rid = self.sequence future = asyncio.get_running_loop().create_future() diff --git a/hermes_cli/gateway_connection.py b/hermes_cli/gateway_connection.py index 525ef5ac6f..fe3a5c1cf1 100644 --- a/hermes_cli/gateway_connection.py +++ b/hermes_cli/gateway_connection.py @@ -16,13 +16,11 @@ import queue import threading import time -from hermes_cli.gateway_client import (DEFAULT_RPC_TIMEOUT, DISCONNECTED, RESUME_KEY_MISMATCH, GatewayClientError, - GatewayRPCError) +from hermes_cli.gateway_client import (DEFAULT_RPC_TIMEOUT, DISCONNECTED, NOT_SENT, RESUME_KEY_MISMATCH, + GatewayClientError, GatewayRPCError) logger = logging.getLogger(__name__) -# A request refused before it reached any socket: nothing is in doubt, so never DISCONNECTED. -NOT_SENT = "Gateway disconnected before the request was sent; nothing was sent." # ``session.resume`` refusals that mean "this viewer may not open that session": the caller keeps the # session it is on instead of adopting the target (startup resume, bare -c fallback). RESUME_REFUSALS = frozenset({"not_found", "permission_denied"}) diff --git a/tests/hermes_cli/test_gateway_client_rpc.py b/tests/hermes_cli/test_gateway_client_rpc.py index 83e50e8036..3ebdb7f19f 100644 --- a/tests/hermes_cli/test_gateway_client_rpc.py +++ b/tests/hermes_cli/test_gateway_client_rpc.py @@ -4,7 +4,7 @@ import json import pytest -from hermes_cli.gateway_client import GatewayClient, GatewayClientError, GatewayRPCError +from hermes_cli.gateway_client import NOT_SENT, GatewayClient, GatewayClientError, GatewayRPCError class _Socket: @@ -89,3 +89,26 @@ async def test_send_on_a_dead_socket_is_a_client_error_not_a_raw_websockets_one( with pytest.raises(GatewayClientError, match='turn outcome is unknown'): await client.rpc('subagent.list', session_id='s') assert client.pending == {} + + +@pytest.mark.asyncio +async def test_a_request_after_the_reader_ended_is_not_sent(): + """The keepalive timeout closes the socket and ends the reader before the dock's next poll: that + send is never written (websockets refuses a send on a closed socket), so the caller is told + nothing was sent, not that a turn's outcome is unknown.""" + sent = [] + + class _Closed(_Socket): + async def send(self, raw): + sent.append(raw) + + async def __anext__(self): + raise StopAsyncIteration # the socket closed: iteration ends + + async with GatewayClient(_Closed(lambda req: None)) as client: + await asyncio.sleep(0) # let the reader see the close + assert client.reader.done() + with pytest.raises(GatewayClientError) as refused: + await client.rpc('prompt.submit', session_id='s') + assert str(refused.value) == NOT_SENT + assert sent == [] and client.pending == {}