fix: read the per-turn notification queue so Codex turns stop hanging

The openai-codex SDK routes every turn-scoped notification into a dedicated queue and
drops it into a pending buffer when the turn is not registered; only unscoped events
reach the global queue. `next_notification()` -- which in the old SDK was the whole
stream -- now returns just those global leftovers, so our loop blocked forever while
the turn's deltas, items and turn/completed piled up unread. Same method name, a
narrower contract: the rename-only migration sailed straight past it.

Register the turn and drain its queue, as the SDK's own TurnHandle.stream() does.
Registration replays anything buffered between turn/start returning and the register
call, so no early event is lost.

Also route notifications back through loop.call_soon_threadsafe: the worker thread was
calling asyncio.Queue.put_nowait directly, which is not thread-safe.

Verified with a live turn against codex 0.144.3: 6 events, reply "PONG".

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: sorryhyun

backend/providers/codex/app_server_instance.py

@@ -304,9 +304,15 @@ async def start_turn(
         # Queue for passing notifications from sync thread to async
         queue: asyncio.Queue[Optional[Notification]] = asyncio.Queue()
         turn_error: List[Optional[Exception]] = [None]
+        loop = asyncio.get_event_loop()
 
         def _run_turn_sync():
             """Run the turn in a sync thread, collecting notifications."""
+
+            def emit(item: Optional[Notification]) -> None:
+                # asyncio.Queue is not thread-safe; hand items over on the loop thread.
+                loop.call_soon_threadsafe(queue.put_nowait, item)
+
             try:
                 client = self._client
                 if client is None:
@@ -316,39 +322,39 @@ def _run_turn_sync():
                 started = client.turn_start(thread_id, input_items, params=turn_params)
                 turn_id = started.turn.id
 
-                # Put turn started event
-                turn_started_notif = Notification(
-                    method="turn/started",
-                    payload=UnknownNotification(params={"turnId": turn_id, "threadId": thread_id}),
-                )
-                queue.put_nowait(turn_started_notif)
-
-                # Stream notifications until turn completes
-                while True:
-                    try:
-                        notification = client.next_notification()
-                        queue.put_nowait(notification)
+                # Everything this turn emits is routed to a dedicated queue, NOT the
+                # global one `next_notification()` drains — without registering, the
+                # events pile up unread and the turn never completes. Registering also
+                # replays whatever arrived between turn/start returning and this call.
+                client.register_turn_notifications(turn_id)
+                try:
+                    # Put turn started event
+                    emit(
+                        Notification(
+                            method="turn/started",
+                            payload=UnknownNotification(params={"turnId": turn_id, "threadId": thread_id}),
+                        )
+                    )
 
-                        # Check for turn completion
+                    # Stream notifications until turn completes. The queue carries only
+                    # this turn's events, so any turn/completed is ours.
+                    while True:
+                        try:
+                            notification = client.next_turn_notification(turn_id)
+                        except TransportClosedError:
+                            break
+                        emit(notification)
                         if notification.method == "turn/completed":
-                            if isinstance(notification.payload, TurnCompletedNotification):
-                                if notification.payload.turn.id == turn_id:
-                                    break
-                            else:
-                                break
-                    except TransportClosedError:
-                        break
-                    except Exception as e:
-                        turn_error[0] = e
-                        break
+                            break
+                finally:
+                    client.unregister_turn_notifications(turn_id)
             except Exception as e:
                 turn_error[0] = e
             finally:
                 # Signal completion
-                queue.put_nowait(None)
+                emit(None)
 
         # Run the sync turn in a background thread
-        loop = asyncio.get_event_loop()
         turn_future = loop.run_in_executor(None, _run_turn_sync)
 
         try: