diff --git a/sentdm_voice_agents/agent.py b/sentdm_voice_agents/agent.py index 94152e9..fab8535 100644 --- a/sentdm_voice_agents/agent.py +++ b/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: diff --git a/sentdm_voice_agents/frames.py b/sentdm_voice_agents/frames.py index 3f2681e..12af10e 100644 --- a/sentdm_voice_agents/frames.py +++ b/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",) diff --git a/sentdm_voice_agents/py.typed b/sentdm_voice_agents/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/sentdm_voice_agents/session.py b/sentdm_voice_agents/session.py index ad63956..21179ea 100644 --- a/sentdm_voice_agents/session.py +++ b/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)) diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 0000000..815d0a7 --- /dev/null +++ b/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 == [] diff --git a/tests/test_session.py b/tests/test_session.py index bc8e73c..776ef4e 100644 --- a/tests/test_session.py +++ b/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