fix(connection): preserve notification response ordering

[hallerite]

we had a downstream issue in verifiers where an ACP request could complete before handlers for preceding notifications had finished. we patched it in our codebase by adding a wait, which is not optimal. this is my attempt of upstreaming a better fix.

Summary

• keep an outgoing request pending until notification handlers for messages received before its response have completed
• scope the barrier to notifications received after that request began, so notification handlers can still make nested requests without deadlocking
• record responses immediately before waiting on the barrier, preserving a valid response if the peer closes the transport right afterward

Root cause

Connection publishes notifications to the asynchronous dispatcher, but handles responses directly in the receive loop. A notification followed immediately by its request response could therefore resolve send_request() before the notification handler had updated client state. ACP consumers could observe a completed prompt with its final streamed content still missing.

The response is now stored as soon as it arrives, while send_request() waits on a per-request notification barrier before returning or raising. This keeps wire-order semantics without turning notification dispatch synchronous. Responses to nested requests exclude the notification that initiated them, avoiding a circular wait.

The immediate response storage is important for transport shutdown: an EOF after a valid response can no longer replace that response with ConnectionError while a preceding notification handler is still finishing.

Downstream reproduction: PrimeIntellect-ai/verifiers#2262

Validation

• make check
• make test (199 passed, 1 skipped)

[frostming]

Taking this opportunity, I reread the implementation of connection & transport. It has existed since the initial version and was actually translated from TypeScript. I feel it's a bit too complicated. I'm currently discussing with my agent whether it can be simplified, and there should be a refactoring later.

[frostming]

Please see #132 and check if the issue still exists or you need to update the PR. @hallerite

[hallerite]

hey @frostming,
rebased my PR onto main. The issue unfortunately still exists.

[frostming]

This solution has a problem: you are associating all notifications received after a request is sent with that request. In reality, this is not the case. For example, the session/update notification is only relevant when the session id is the same.

However, to achieve this kind of precise association, the Connection—as a low-level communication abstraction—would need to know too much, which violates the principles of software design.

I feel this is still better suited for the Agent/Client to implement and ensure that, and what we need to provide is a way to tell the caller that the notification has been processed, rather than being completely asynchronous.

What do you think?

[hallerite]

makes a lot of sense. I have refactored it following your recommendations and it's now also closer to the typescript implementation

[frostming]

Maybe a bit nitpicky, but this except branch can be deleted, because CancelledError won't be caught by the following except Exception.

[frostming]

Suggested change:

class _SessionUpdateTracker:
    def __init__(self, client):
        self._client = client
        self._pending = {}  # session_id -> set of unresolved futures
        # no need to track the sequence numbers, all prompt requests of the same session
        # can wait for the same set of futures.

    async def session_update(self, session_id, update):
        # tracking logic
        return await self._client.session_update(session_id, update)

    # other useful methods

    def __getattr__(self, name):
        """delegate to the internal client methods"""
        return getattr(self._client, name)

And pass the tracker to the router builder:

handler = build_client_router(self._session_updates, ...)