[Python SDK] Turn and global reads block forever after the transport fails
What happened
MessageRouter.fail_all states its own contract in a comment:
Put the same transport failure into every queue so no SDK call blocks forever waiting for a response that cannot arrive.
That holds for callers already parked inside queue.get() — the failure is appended to every queue it knows about, which wakes them.
It does not hold for callers that arrive after that item has been drained. The reader thread has exited, so nothing will ever be enqueued again, yet next_turn_notification and next_global_notification still call a blocking get(). They wait forever.
Reproduction
Against MessageRouter directly, no app-server needed:
from openai_codex._message_router import MessageRouter
from openai_codex.errors import TransportClosedError
router = MessageRouter()
router.register_turn("turn-1")
router.fail_all(TransportClosedError("transport closed"))
router.next_turn_notification("turn-1") # raises TransportClosedError
router.next_turn_notification("turn-1") # never returnsA turn registered after the failure is worse: its queue is empty by definition, so the very first read blocks.
next_login_notification is not affected today, but only incidentally — fail_all clears _login_notifications, so the lookup fails fast with RuntimeError("login ... is not registered for waiting") rather than reporting the transport failure.
Suggested fix
Record the terminal failure in fail_all and have the read paths consult it. A read that finds its queue empty raises the transport failure instead of waiting.
Queues are deliberately not cleared. A consumer must still be able to drain events that really did arrive before the transport died — fail_all appends the failure rather than replacing the contents, and that ordering is load-bearing: clearing _turn_notifications there discards buffered notifications and breaks test_client_reader_routes_interleaved_turn_notifications_by_turn_id. (I tried that first; the existing suite caught it.)
diff --git a/sdk/python/src/openai_codex/_message_router.py b/sdk/python/src/openai_codex/_message_router.py
index c979c8c..2b10c19 100644
--- a/sdk/python/src/openai_codex/_message_router.py
+++ b/sdk/python/src/openai_codex/_message_router.py
@@ -33,6 +33,10 @@ class MessageRouter:
self._pending_turn_notifications: dict[str, deque[Notification]] = {}
self._goal_operations: dict[str, _GoalOperationState] = {}
self._global_notifications: queue.Queue[NotificationQueueItem] = queue.Queue()
+ # Set once the reader thread has exited. Queues stay readable so callers
+ # can drain what already arrived, but nothing new can ever be enqueued,
+ # so a read that finds one empty has to raise instead of waiting.
+ self._transport_failure: BaseException | None = None
def create_response_waiter(self, request_id: str) -> queue.Queue[ResponseQueueItem]:
"""Register a one-shot queue for a JSON-RPC response id."""
@@ -48,14 +52,33 @@ class MessageRouter:
with self._lock:
self._response_waiters.pop(request_id, None)
- def next_global_notification(self) -> Notification:
- """Block until the next notification that is not scoped to a turn."""
+ def _next_notification(self, notifications: queue.Queue[NotificationQueueItem]) -> Notification:
+ """Return the next queued notification, raising once nothing can arrive.
- item = self._global_notifications.get()
+ ``fail_all`` appends the transport failure to every queue it knows about,
+ so a caller blocked here is always woken. A caller that arrives *after*
+ that failure has already been drained would otherwise wait on a queue the
+ dead reader thread can never fill again.
+ """
+
+ with self._lock:
+ failure = self._transport_failure
+ if failure is not None:
+ try:
+ item = notifications.get_nowait()
+ except queue.Empty:
+ raise failure from None
+ else:
+ item = notifications.get()
if isinstance(item, BaseException):
raise item
return item
+ def next_global_notification(self) -> Notification:
+ """Block until the next notification that is not scoped to a turn."""
+
+ return self._next_notification(self._global_notifications)
+
def register_login(self, login_id: str) -> None:
"""Register a queue for one interactive login attempt."""
@@ -81,10 +104,7 @@ class MessageRouter:
login_queue = self._login_notifications.get(login_id)
if login_queue is None:
raise RuntimeError(f"login {login_id!r} is not registered for waiting")
- item = login_queue.get()
- if isinstance(item, BaseException):
- raise item
- return item
+ return self._next_notification(login_queue)
def register_turn(self, turn_id: str) -> None:
"""Register a queue for a turn stream and replay early events."""
@@ -113,10 +133,7 @@ class MessageRouter:
turn_queue = self._turn_notifications.get(turn_id)
if turn_queue is None:
raise RuntimeError(f"turn {turn_id!r} is not registered for streaming")
- item = turn_queue.get()
- if isinstance(item, BaseException):
- raise item
- return item
+ return self._next_notification(turn_queue)
def register_goal(self, thread_id: str) -> _GoalOperationState:
"""Register one thread-scoped logical goal operation before it starts."""
@@ -218,6 +235,7 @@ class MessageRouter:
"""Wake every blocked waiter when the reader thread exits."""
with self._lock:
+ self._transport_failure = exc
response_waiters = list(self._response_waiters.values())
self._response_waiters.clear()
login_queues = list(self._login_notifications.values())Tests
Branch with the change and tests: https://github.com/wenxuanzhang1209-cyber/codex/tree/fix/python-sdk-reads-block-after-transport-failure
MessageRouter currently has no test coverage. The branch adds sdk/python/tests/test_message_router_failure.py with five tests; two fail without the change, the rest pin down behavior that has to survive it (notably that buffered notifications stay readable). Each read runs on a worker thread with a deadline rather than being called directly, so a regression fails the suite instead of hanging it.
Verification
sdk/python suite before and after: the same 43 pre-existing failures (they need an app-server binary that was not available in my environment), passing count 88 → 93.
I would have opened a pull request, but POST /repos/openai/codex/pulls returns 404 for outside accounts, so I am filing the patch here instead. Happy to open one if there is a path for external contributions.