From 313b24fe0449c62dfdd49c846947e5b29264734f Mon Sep 17 00:00:00 2001 From: Ryuichi Leo Takashige Date: Fri, 16 Jan 2026 18:26:44 +0000 Subject: [PATCH] Handle broken resource error gracefully # Conflicts: # src/exo/master/api.py --- src/exo/master/api.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/src/exo/master/api.py b/src/exo/master/api.py index 3f36c3c8..56643d52 100644 --- a/src/exo/master/api.py +++ b/src/exo/master/api.py @@ -654,14 +654,17 @@ class API: for idx, event in self.event_buffer.drain_indexed(): self._event_log.append(event) self.state = apply(self.state, IndexedEvent(event=event, idx=idx)) - if isinstance(event, ChunkGenerated): + if ( + isinstance(event, ChunkGenerated) + and event.command_id in self._chat_completion_queues + ): assert isinstance(event.chunk, TokenChunk) - queue = self._chat_completion_queues.get(event.command_id) - if queue is not None: - try: - await queue.send(event.chunk) - except BrokenResourceError: - self._chat_completion_queues.pop(event.command_id, None) + try: + await self._chat_completion_queues[event.command_id].send( + event.chunk + ) + except BrokenResourceError: + del self._chat_completion_queues[event.command_id] async def _pause_on_new_election(self): with self.election_receiver as ems: