voice-agents-python: the .NET runtime holds frames during a resume, fails lost tool calls, reports a drained connection's leftovers

sentdm_voice_agents/agent.py

@@ -27,7 +27,8 @@
 DEFAULT_URL = f"wss://{frames.HOST}{frames.CONNECT_PATH}"
 # The longest wait before another attempt to connect.
 MAX_BACKOFF_S = 30.0
-# How long a handshake may take before the attempt counts as failed.
+# How long a handshake may take before the attempt counts as failed, and how long an open socket may wait for
+# connection.ready before it counts as dead.
 HANDSHAKE_TIMEOUT_S = 10.0
 # How many heartbeat intervals without a frame make a connection dead, as Heartbeats in docs/protocol.md says.
 SILENT_INTERVALS = 3
@@ -89,6 +90,8 @@ def __init__(self, *, api_key: str, agent_id: str, url: str = DEFAULT_URL) -> No
         self._connections: set[_Connection] = set()
         self._sessions: dict[str, Session] = {}
         self._tasks: dict[str, asyncio.Task[None]] = {}
+        # Connection-level handlers run off the reader, so a slow one never delays a ping's pong or a session.
+        self._background: set[asyncio.Task[None]] = set()
         self._attempt = 0
         self._stopped = False
         self._done: asyncio.Future[None] | None = None
@@ -174,9 +177,9 @@ async def _close_all(self) -> None:
                 connection.task.cancel()
             if connection.task is not None:
                 tasks.append(connection.task)
-        for task in self._tasks.values():
+        for task in [*self._tasks.values(), *self._background]:
             task.cancel()
-        await asyncio.gather(*tasks, *self._tasks.values(), return_exceptions=True)
+        await asyncio.gather(*tasks, *self._tasks.values(), *self._background, return_exceptions=True)
 
     # ----- connections -----
 
@@ -215,11 +218,12 @@ async def _serve(self, connection: _Connection, delay: float) -> None:
 
     async def _read(self, connection: _Connection, socket: ClientConnection) -> None:
         while True:
-            timeout = SILENT_INTERVALS * connection.heartbeat_s if connection.ready else None
+            # Before connection.ready the socket has the handshake's deadline; after it, the heartbeat's.
+            timeout = SILENT_INTERVALS * connection.heartbeat_s if connection.ready else HANDSHAKE_TIMEOUT_S
             try:
                 message = await asyncio.wait_for(socket.recv(), timeout)
             except TimeoutError:
-                log.warning("no frame for %.0f s, closing", timeout or 0)
+                log.warning("no %s for %.0f s, closing", "frame" if connection.ready else "connection.ready", timeout)
                 return
             except ConnectionClosed as closed:
                 close = closed.rcvd or closed.sent
@@ -240,6 +244,10 @@ async def _handle(self, connection: _Connection, frame: frames.Frame) -> None:
             await self._ready(connection, frame)
         elif isinstance(frame, frames.Ping):
             await connection.send(frames.Pong(ts=frame.ts))
+            # A connection that lived through a heartbeat interval is a good one: the next loss starts the
+            # backoff over. A connection that dies before its first ping keeps it growing.
+            if connection.ready:
+                self._attempt = 0
         elif isinstance(frame, frames.ConnectionDraining):
             self._drain(connection)
         elif isinstance(frame, frames.SessionStart):
@@ -248,7 +256,7 @@ async def _handle(self, connection: _Connection, frame: frames.Frame) -> None:
             await self._resumed(connection, frame)
         elif isinstance(frame, frames.ErrorFrame):
             self._error(connection, frame)
-            await self._safely(self.on_error(frame))
+            self._spawn(self.on_error(frame))
         elif isinstance(frame, frames.UnknownFrame) and frame.session_id is None:
             log.info("ignored a %s frame", frame.type)
         else:
@@ -257,13 +265,12 @@ async def _handle(self, connection: _Connection, frame: frames.Frame) -> None:
     async def _ready(self, connection: _Connection, frame: frames.ConnectionReady) -> None:
         connection.ready = True
         connection.heartbeat_s = frame.heartbeat_interval_ms / 1000
-        self._attempt = 0
         pending = [s for s in connection.pending if self._sessions.get(s.id) is s]
         connection.pending = []
         log.info("ready: protocol %s, up to %s sessions, resuming %s", frame.protocol_version, frame.max_sessions, len(pending))
         for session in pending:
             await self._resume(connection, session)
-        await self._safely(self.on_connected(frame))
+        self._spawn(self.on_connected(frame))
 
     async def _resume(self, connection: _Connection, session: Session) -> None:
         for other in self._connections:
@@ -319,10 +326,14 @@ async def _resumed(self, connection: _Connection, frame: frames.SessionResumed)
             await connection.send(frame_again)
 
     def _error(self, connection: _Connection, frame: frames.ErrorFrame) -> None:
+        """An error is uncounted, as Counting frames in docs/protocol.md says. Only one that refuses a
+        session.resume changes the session's fate; any other is the handler's to read."""
         log.warning("error %s: %s", frame.error.code, frame.error.message)
         session = connection.sessions.get(frame.session_id or "")
         if session is None or not session._resuming:
             return
+        if frame.frame_type != "session.resume" and frame.error.code not in ("resume_failed", "connection_full"):
+            return
         session._resuming = False
         connection.sessions.pop(session.id, None)
         if frame.error.code == "connection_full":
@@ -338,6 +349,7 @@ def _forget(self, session: Session) -> None:
 
     def _lost(self, session: Session) -> None:
         self._forget(session)
+        session._abandon()
         session._inbox.put_nowait(None)
 
     async def _closed(self, connection: _Connection) -> None:
@@ -391,6 +403,12 @@ async def _run_session(self, session: Session) -> None:
                 await self._safely(self.on_frame(session, item))
         self._tasks.pop(session.id, None)
 
+    def _spawn(self, handler: object) -> None:
+        """Runs a connection-level handler on its own task, off the reader."""
+        task = asyncio.create_task(self._safely(handler))
+        self._background.add(task)
+        task.add_done_callback(self._background.discard)
+
     async def _safely(self, handler: object) -> None:
         """Runs a handler; an exception in it is logged and does not take the session or the agent down."""
         try:

sentdm_voice_agents/frames.py

@@ -244,6 +244,10 @@ class UnknownFrame:
     session_id: str | None
     data: dict[str, Any]
 
+    @property
+    def TYPE(self) -> str:  # noqa: N802 - the wire type, as the known frames spell it
+        return self.type
+
 
 Frame = Union[ConnectionReady, ConnectionDraining, Ping, ErrorFrame, SessionStart, UserSaid, ToolResult, SessionEnd, SessionResumed, AgentSay, ToolCall, AgentClear, SessionResume, Pong, UnknownFrame]
 
@@ -259,7 +263,7 @@ class UnknownFrame:
 SESSION_TYPES: dict[str, frozenset[str]] = {"session.start": frozenset({"text", "audio"}), "user.said": frozenset({"text"}), "tool.result": frozenset({"text", "audio"}), "session.end": frozenset({"text", "audio"}), "session.resumed": frozenset({"text", "audio"}), "agent.say": frozenset({"text"}), "tool.call": frozenset({"text", "audio"}), "agent.clear": frozenset({"text", "audio"}), "session.resume": frozenset({"text", "audio"})}
 # Session frames that count towards `received` on each side; every other frame with a session_id,
 # also one of an unknown type, counts too. The ones not counted are listed so a runtime can tell.
-UNCOUNTED = frozenset({"session.start", "session.resumed", "session.resume", "audio"})
+UNCOUNTED = frozenset({"error", "session.start", "session.resumed", "session.resume", "audio"})
 
 # The values Sent lists today; it may add more, so a runtime treats others as it would the fallback.
 PARTY_KINDS = ("user", "number", "conference", "agent", "anonymous",)

sentdm_voice_agents/session.py

@@ -50,6 +50,8 @@ def __init__(self, start: frames.SessionStart, connection: _Connection) -> None:
         self.received = 0
         # session.end, once it arrived.
         self.end: frames.SessionEnd | None = None
+        # The connection was lost and the session could not be resumed: Sent ends the call without a session.end.
+        self.lost = False
         self._connection = connection
         # A session.resume was sent on the current connection and Sent has not answered it yet.
         self._resuming = False
@@ -64,7 +66,8 @@ def __init__(self, start: frames.SessionStart, connection: _Connection) -> None:
 
     @property
     def ended(self) -> bool:
-        return self.end is not None
+        """Whether the call has left the agent: session.end arrived, or the session was lost."""
+        return self.end is not None or self.lost
 
     @property
     def resumable(self) -> bool:
@@ -87,9 +90,15 @@ async def clear(self) -> None:
 
     async def send_audio(self, samples: bytes) -> None:
         """Audio sessions: play `samples`, 16-bit little-endian mono PCM at the session's sample rate. Send
-        20 ms per call, and no more than 200 ms."""
+        20 ms per call, and no more than 200 ms. Raises SessionEnded once the call has left the agent."""
+        if self.ended:
+            raise SessionEnded(self.id)
         if self.stream is None:
             raise RuntimeError(f"session {self.id} is a text session")
+        if self._resuming:
+            # The stream number belongs to the old connection; on the new one it may be another caller's until
+            # session.resumed assigns this session's. Audio is not counted, so there is nothing to send again.
+            return
         await self._connection.send_bytes(audio_frame(self.stream, samples))
 
     async def transfer(self, to: frames.Party | str) -> frames.ToolResult:
@@ -125,10 +134,15 @@ async def tool(self, name: str, arguments: dict[str, Any] | None = None) -> fram
 
     async def _send(self, frame: frames.Frame) -> None:
         """Sends a frame for the session on its connection, keeping it if it counts. A frame that could not be
-        sent still counts: Sent's `received` on the next resume says whether to send it again."""
+        sent still counts: Sent's `received` on the next resume says whether to send it again. A frame that
+        breaks the protocol is refused before it counts. While a resume is unanswered the frame is kept and not
+        sent: the replay after session.resumed sends it, once, on the connection the session is on."""
         if self.ended:
             raise SessionEnded(self.id)
+        frames.encode(frame)
         self._keep(frame)
+        if self._resuming:
+            return
         await self._connection.send(frame)
 
     def _keep(self, frame: frames.Frame) -> None:
@@ -180,6 +194,16 @@ def _answer_tool(self, result: frames.ToolResult) -> None:
 
     def _finish(self, end: frames.SessionEnd) -> None:
         self.end = end
+        self._fail_pending()
+
+    def _abandon(self) -> None:
+        """The session is lost: no tool.result or session.end will come, so every waiting tool call fails, and
+        anything sent from now on is refused."""
+        self.lost = True
+        self._resuming = False
+        self._fail_pending()
+
+    def _fail_pending(self) -> None:
         for future in self._pending.values():
             if not future.done():
                 future.set_exception(SessionEnded(self.id))

tests/test_agent.py

@@ -0,0 +1,84 @@
+"""The agent's bookkeeping around a connection, without a socket: which errors refuse a resume, when the
+backoff starts over, and what a lost session's task sees."""
+
+from __future__ import annotations
+
+import asyncio
+
+from sentdm_voice_agents import frames
+from sentdm_voice_agents.agent import Agent, _Connection
+from sentdm_voice_agents.session import Session, SessionEnded
+
+from test_session import Recorder, start
+
+
+def error(**fields) -> frames.ErrorFrame:
+    return frames.decode({"type": "error", "session_id": "vs_1", "error": {"code": "frame_not_allowed", "message": "no"}, **fields})
+
+
+def resuming_session(agent: Agent, connection: _Connection) -> Session:
+    session = Session(start(), Recorder())
+    session._resuming = True
+    agent._sessions[session.id] = session
+    connection.sessions[session.id] = session
+    agent._connections.add(connection)
+    return session
+
+
+async def test_only_an_error_about_the_resume_refuses_it():
+    agent = Agent(api_key="k", agent_id="agent_1")
+    connection = _Connection([])
+    session = resuming_session(agent, connection)
+    agent._error(connection, error(frame_type="agent.say"))
+    assert session._resuming and session.id in connection.sessions and not session.lost
+    agent._error(connection, error(frame_type="session.resume", error={"code": "resume_failed", "message": "gone"}))
+    assert session.lost and session.id not in connection.sessions and session.id not in agent._sessions
+    assert await session._inbox.get() is None
+
+
+async def test_a_full_connection_moves_the_resume_elsewhere():
+    agent = Agent(api_key="k", agent_id="agent_1")
+    connection = _Connection([])
+    session = resuming_session(agent, connection)
+    agent._error(connection, error(error={"code": "connection_full", "message": "full"}))
+    assert not session.lost and session.id not in connection.sessions
+    opened = [c for c in agent._connections if c is not connection]
+    assert len(opened) == 1 and opened[0].pending == [session]
+    for c in opened:
+        c.task.cancel()
+    await asyncio.gather(*(c.task for c in opened), return_exceptions=True)
+
+
+async def test_the_backoff_starts_over_after_a_heartbeat_not_on_ready():
+    agent = Agent(api_key="k", agent_id="agent_1")
+    connection = _Connection([])
+    agent._attempt = 3
+    await agent._ready(connection, frames.decode({"type": "connection.ready", "protocol_version": "0.2.0", "heartbeat_interval_ms": 15000, "max_sessions": 10}))
+    assert agent._attempt == 3
+    await agent._handle(connection, frames.decode({"type": "ping", "ts": 1}))
+    assert agent._attempt == 0
+    await asyncio.gather(*agent._background, return_exceptions=True)
+
+
+async def test_a_lost_session_reaches_its_handler_after_its_tool_call_fails():
+    seen: list[str] = []
+
+    class Noticing(Agent):
+        async def on_session_start(self, session: Session) -> None:
+            try:
+                await session.hangup()
+            except SessionEnded:
+                seen.append("tool failed")
+
+        async def on_session_lost(self, session: Session) -> None:
+            seen.append("lost")
+
+    agent = Noticing(api_key="k", agent_id="agent_1")
+    connection = _Connection([])
+    agent._connections.add(connection)
+    agent._start(connection, start())
+    await asyncio.sleep(0)
+    session = agent.sessions[0]
+    agent._lost(session)
+    await asyncio.wait_for(agent._tasks.get(session.id) or asyncio.sleep(0), 1)
+    assert seen == ["tool failed", "lost"] and agent.sessions == []

tests/test_session.py

@@ -94,3 +94,60 @@ async def test_audio_goes_out_on_the_session_stream():
     session = Session(start(session_type="audio", audio={"stream": 7}), recorder)
     await session.send_audio(b"\x01\x00" * 320)
     assert recorder.audio[0][:2] == b"\x00\x07" and len(recorder.audio[0]) == 642
+
+
+async def test_a_frame_that_breaks_the_protocol_is_refused_before_it_counts():
+    recorder = Recorder()
+    session = Session(start(), recorder)
+    with pytest.raises(frames.FrameError):
+        await session.say("")
+    with pytest.raises(frames.FrameError):
+        await session.send_dtmf("xyz")
+    assert session.sent == 0 and recorder.frames == []
+
+
+async def test_frames_sent_during_a_resume_wait_for_the_replay():
+    recorder = Recorder()
+    session = Session(start(), recorder)
+    await session.say("before")
+    session._resuming = True
+    await session.say("during")
+    task = asyncio.create_task(session.hangup())
+    await asyncio.sleep(0)
+    assert [f.text for f in recorder.frames] == ["before"] and session.sent == 3
+    again = session._resumed(frames.decode({"type": "session.resumed", "session_id": "vs_1", "received": 1, "resume": {"token": "vsr_" + "d" * 30, "window_ms": 15000}}))
+    assert [getattr(f, "text", getattr(f, "name", None)) for f in again] == ["during", "hangup"]
+    session._answer_tool(frames.decode({"type": "tool.result", "session_id": "vs_1", "call_ref": again[1].call_ref, "ok": True}))
+    assert (await task).ok
+
+
+async def test_audio_during_a_resume_is_dropped_and_after_the_end_refused():
+    recorder = Recorder()
+    session = Session(start(session_type="audio", audio={"stream": 7}), recorder)
+    session._resuming = True
+    await session.send_audio(b"\x01\x00" * 320)
+    assert recorder.audio == []
+    session._resuming = False
+    session._finish(frames.decode({"type": "session.end", "session_id": "vs_1", "reason": "caller_hung_up", "duration_ms": 10}))
+    with pytest.raises(SessionEnded):
+        await session.send_audio(b"\x01\x00" * 320)
+
+
+async def test_a_lost_session_fails_its_tool_calls_and_refuses_more():
+    session = Session(start(), Recorder())
+    task = asyncio.create_task(session.hangup())
+    await asyncio.sleep(0)
+    session._abandon()
+    with pytest.raises(SessionEnded):
+        await task
+    assert session.ended and session.lost
+    with pytest.raises(SessionEnded):
+        await session.say("too late")
+
+
+async def test_unknown_session_frames_count_and_errors_do_not():
+    session = Session(start(), Recorder())
+    session._read(frames.decode({"type": "session.note", "session_id": "vs_1", "x": 1}))
+    assert session.received == 1
+    session._read(frames.decode({"type": "error", "session_id": "vs_1", "error": {"code": "frame_not_allowed", "message": "no"}}))
+    assert session.received == 1