Route Python SDK turn notifications by ID

[aibrahim-oai]

Why

The Python SDK previously protected the stdio transport with a single active turn-consumer guard. That avoided competing reads from stdout, but it also meant one Codex/AsyncCodex client could not stream multiple active turns at the same time. Notifications could also arrive before the caller received a TurnHandle and registered for streaming, so the SDK needed an explicit routing layer instead of letting individual API calls read directly from the shared transport.

What Changed

• Added a private MessageRouter that owns per-request response queues, per-turn notification queues, pending turn-notification replay, and global notification delivery behind a single stdout reader thread.
• Generated typed notification routing metadata so turn IDs come from known payload shapes instead of router-side attribute guessing, with explicit fallback handling for unknown notification payloads.
• Updated sync and async turn streaming so TurnHandle.stream()/run() and stream_text() consume only notifications for their own turn ID, while AsyncAppServerClient no longer serializes all transport calls behind one async lock.
• Cleared pending turn-notification buffers when unregistered turns complete so never-consumed turn handles do not leave stale queues behind.
• Removed the internal stream-until helper now that turn completion waiting can register directly with routed turn notifications.
• Updated Python SDK docs and focused tests for concurrent transport calls, interleaved turn routing, buffered early notifications, unknown notification routing, async delegation, and routed turn completion behavior.

Validation

• uv run --extra dev ruff format scripts/update_sdk_artifacts.py src/codex_app_server/_message_router.py src/codex_app_server/client.py src/codex_app_server/generated/notification_registry.py tests/test_client_rpc_methods.py tests/test_public_api_runtime_behavior.py tests/test_async_client_behavior.py
• uv run --extra dev ruff check scripts/update_sdk_artifacts.py src/codex_app_server/_message_router.py src/codex_app_server/client.py src/codex_app_server/generated/notification_registry.py tests/test_client_rpc_methods.py tests/test_public_api_runtime_behavior.py tests/test_async_client_behavior.py
• uv run --extra dev pytest tests/test_client_rpc_methods.py tests/test_public_api_runtime_behavior.py tests/test_async_client_behavior.py
• git diff --check

[aibrahim-oai]

@codex review this

[chatgpt-codex-connector]

💡 Codex Review

codex/sdk/python/src/codex_app_server/client.py

Line 426 in 48a780e

notification = self.next_notification()

Make stream_until_methods consume turn queues

turn_start() now registers the returned turn ID, so all turn/* and item notifications for that turn are routed away from the global queue. stream_until_methods("turn/completed") still waits on next_notification(), so the completion event is never observed and callers can block indefinitely. Route turn-scoped targets through next_turn_notification() or stop exposing this helper for turn methods.

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[aibrahim-oai]

@codex review this

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: b22c46a66b

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Preserve transport failures for pending turns

fail_all only wakes queues already in _turn_notifications. Early turn events can be buffered in _pending_turn_notifications before turn_start registers the turn; if stdout closes after the response is delivered but before registration, the handle drains those pending events and then blocks forever instead of seeing the transport error.

Useful? React with 👍 / 👎.

[owenlin0]

One thing I’d consider changing in _notification_turn_id:

def _notification_turn_id(self, notification: Notification) -> str | None:
    payload = notification.payload
    turn_id = getattr(payload, "turn_id", None)
    if isinstance(turn_id, str):
        return turn_id
    turn = getattr(payload, "turn", None)
    nested_turn_id = getattr(turn, "id", None)
    if isinstance(nested_turn_id, str):
        return nested_turn_id
    return None

The getattr approach feels hacky since the router is guessing at generated notification shapes. Since we already have typed generated notification payloads, I think this would be cleaner as generated notification-routing metadata/helper code instead.

For example, the SDK generator could emit a helper next to notification_registry.py that knows which notification payloads have turn_id directly and which have turn.id nested (by introspecting the generated json schema types):

def notification_turn_id(notification: Notification) -> str | None:
    payload = notification.payload

    if isinstance(payload, DIRECT_TURN_ID_NOTIFICATION_TYPES):
        return payload.turn_id

    if isinstance(payload, NESTED_TURN_NOTIFICATION_TYPES):
        return payload.turn.id

    if isinstance(payload, UnknownNotification):
        raw_turn_id = payload.params.get("turnId")
        if isinstance(raw_turn_id, str):
            return raw_turn_id
        raw_turn = payload.params.get("turn")
        if isinstance(raw_turn, dict) and isinstance(raw_turn.get("id"), str):
            return raw_turn["id"]

    return None

Then MessageRouter can then just call notification_turn_id(notification).

[owenlin0]

btw it seems like a TurnHandle that is created but never consumed can leave a live queue forever?

[owenlin0]

pre-approving in case you want to address the above as followups