fix(browser): the CDP relay passes each answer on in order with the events around it

packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py

@@ -281,6 +281,13 @@ async def _command(self, connection: _Connection, message: Any) -> None:
         logged = self._begin(connection, session, method, params)
         answer, guarded = {"error": {"code": SERVER_ERROR, "message": ENDED_UNANSWERED}}, None
         sending: Optional[asyncio.Future] = None
+        replied = False
+
+        def reply(answer: dict) -> None:
+            nonlocal replied
+            replied = True
+            self._put(connection, _reply(message, answer))
+
         try:
             async with connection.order:
                 if logged is not None:
@@ -293,7 +300,8 @@ async def _command(self, connection: _Connection, message: Any) -> None:
                 elif verdict.action != "send":
                     answer = {"result": {}}
                 else:
-                    sending = asyncio.ensure_future(self.devtools.request(method, verdict.params, session_id=session))
+                    sending = asyncio.ensure_future(self.devtools.request(method, verdict.params, session_id=session,
+                                                                          answered=reply))
                     await asyncio.sleep(0)
             if sending is not None:
                 try:
@@ -303,10 +311,8 @@ async def _command(self, connection: _Connection, message: Any) -> None:
             if verdict is not None and (verdict.action != "send" or verdict.changed):
                 guarded = GUARDED[verdict.action]
                 self._guarded(connection, method, verdict)
-            reply = {"id": message.get("id"), **{key: answer[key] for key in ("result", "error") if key in answer}}
-            if message.get("sessionId"):
-                reply["sessionId"] = message["sessionId"]
-            self._put(connection, reply)
+            if not replied:
+                reply(answer)
             if verdict is not None and verdict.action == "close":
                 self._close(connection, CLOSED, dropping=False)
         finally:
@@ -456,6 +462,13 @@ def finish(self, answer: dict, guarded: Optional[str]) -> None:
         self.relay.log.finish(self.begun, **fields)
 
 
+def _reply(message: dict, answer: dict) -> dict:
+    """The client's answer to its command ``message``: the browser's ``answer``, or the relay's own, under its id and
+    on its session."""
+    return {"id": message.get("id"), **{key: answer[key] for key in ("result", "error") if key in answer},
+            **({"sessionId": message["sessionId"]} if message.get("sessionId") else {})}
+
+
 def _well_formed(message: Any) -> bool:
     """Whether ``message`` is a command the relay can read: an object with a string ``method``, and object ``params``
     and a string ``sessionId`` if any."""

packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py

@@ -31,6 +31,7 @@ class DevToolsPipe:
     def __init__(self) -> None:
         self._ids = itertools.count(1)
         self._waiting: dict[int, asyncio.Future] = {}
+        self._answered: dict[int, Callable[[dict], None]] = {}
         self._writer: Optional[asyncio.WriteTransport] = None
         self._reading: Optional[asyncio.Task] = None
         self._listeners: list[Callable[[dict], None]] = []
@@ -62,25 +63,30 @@ async def call(self, method: str, params: Optional[dict] = None, *, session_id:
             raise DevToolsError(f"{method}: {answered['error'].get('message', answered['error'])}")
         return answered.get("result", {})
 
-    async def request(self, method: str, params: Optional[dict] = None, *, session_id: Optional[str] = None) -> dict:
+    async def request(self, method: str, params: Optional[dict] = None, *, session_id: Optional[str] = None,
+                      answered: Optional[Callable[[dict], None]] = None) -> dict:
         """The browser's whole answer to ``method``, its ``result`` or its ``error``, however long it takes: what a
-        relay passes on as it is."""
-        return await self._answer(method, params, session_id, None)
+        relay passes on as it is. ``answered`` hears it as the pipe reads it, in order with the events before and after
+        it, which the caller, resumed only later, may not be."""
+        return await self._answer(method, params, session_id, None, answered)
 
     async def _answer(self, method: str, params: Optional[dict], session_id: Optional[str],
-                      timeout_s: Optional[float]) -> dict:
+                      timeout_s: Optional[float], answered: Optional[Callable[[dict], None]] = None) -> dict:
         if self._writer is None or self._writer.is_closing():
             raise ConnectionError("no browser is connected over its debugging pipe")
         message_id = next(self._ids)
         answer = asyncio.get_running_loop().create_future()
         self._waiting[message_id] = answer
+        if answered is not None:
+            self._answered[message_id] = answered
         message = {"id": message_id, "method": method, "params": params or {},
                    **({"sessionId": session_id} if session_id else {})}
         self._writer.write(json.dumps(message).encode() + b"\0")
         try:
             return await asyncio.wait_for(answer, timeout_s)
         finally:
             self._waiting.pop(message_id, None)
+            self._answered.pop(message_id, None)
 
     def close(self) -> None:
         if self._reading is not None:
@@ -100,6 +106,8 @@ async def _read(self, reader: asyncio.StreamReader) -> None:
                     continue
                 answer = self._waiting.get(message.get("id"))
                 if answer is not None and not answer.done():
+                    if (answered := self._answered.pop(message["id"], None)) is not None:
+                        _heard(answered, message)
                     answer.set_result(message)
         except (asyncio.IncompleteReadError, ConnectionError):
             if self._writer is not None:

packages/agentenv-protocol/tests/test_browser_cdp.py

@@ -45,7 +45,8 @@ class Browser(WebsiteBrowserEnvironment):
 
 class Pipe:
     """A browser's pipe: each connection's browser-level session ``b<n>``, an attach on a session making its child
-    ``<session>-c`` and saying so on that session, ``Boom`` failing, ``Hang`` never answered, anything else echoed;
+    ``<session>-c`` and saying so on that session, ``Boom`` failing, ``Hang`` never answered, ``Ordered`` answered and
+    then followed at once by an event, as the real pipe reads them when both came in one read, anything else echoed;
     every request recorded with the last params of each method, and each ``Hang`` given up."""
 
     def __init__(self) -> None:
@@ -65,9 +66,14 @@ async def call(self, method, params=None, *, session_id=None, timeout_s=None):
             return {"sessionId": f"b{self.sessions}"}
         return {}
 
-    async def request(self, method, params=None, *, session_id=None):
+    async def request(self, method, params=None, *, session_id=None, answered=None):
         self.sent.append((method, session_id))
         self.params[method] = params
+        if method == "Ordered":
+            answer = {"result": {"method": method}}
+            answered(answer)
+            self.emit({"method": "Page.after", "sessionId": session_id, "params": {}})
+            return answer
         if method == "Target.attachToTarget":
             child = f"{session_id}-c"
             self.emit({"method": "Target.attachedToTarget", "sessionId": session_id,
@@ -154,6 +160,16 @@ def test_a_users_token_is_minted_once_kept_by_hash_and_still_known_after_a_resta
     assert not ada.named and restarted.address(cys) == cy and cy.named
 
 
+@pytest.mark.asyncio
+async def test_an_answer_reaches_its_client_before_the_events_the_browser_sent_after_it():
+    async with _relay() as (pipe, relay, base):
+        async with connect(f"{base}/ada") as ada:
+            await ada.send(json.dumps({"id": 1, "method": "Ordered"}))
+            first, second = json.loads(await ada.recv()), json.loads(await ada.recv())
+
+    assert first == {"id": 1, "result": {"method": "Ordered"}} and second == {"method": "Page.after", "params": {}}
+
+
 @pytest.mark.asyncio
 async def test_each_connection_gets_its_own_browser_session_its_own_answers_and_events_and_its_commands_end_with_it():
     async with _relay() as (pipe, relay, base):

packages/agentenv-protocol/tests/test_browser_log.py

@@ -218,7 +218,7 @@ async def call(self, method, params=None, *, session_id=None, timeout_s=None):
             return {"currentIndex": 0, "entries": [{"id": 1, "url": "about:blank"}]}
         return {}
 
-    async def request(self, method, params=None, *, session_id=None):
+    async def request(self, method, params=None, *, session_id=None, answered=None):
         return {"result": await self.call(method, params, session_id=session_id)}
 
 

packages/agentenv-protocol/tests/test_devtools_pipe.py

@@ -0,0 +1,33 @@
+"""A browser's debugging pipe hands a request's ``answered`` its answer as it reads it, before the events the browser
+wrote after it, even when both come in one read and the caller hasn't resumed yet."""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+
+import pytest
+
+from agentenv_protocol.website_browser.devtools import DevToolsPipe
+
+
+@pytest.mark.asyncio
+async def test_an_answer_is_heard_before_the_events_written_after_it_in_the_same_read():
+    from_browser, to_us = os.pipe()
+    from_us, to_browser = os.pipe()
+    pipe, heard = DevToolsPipe(), []
+    pipe.listen(lambda event: heard.append(event["method"]))
+    await pipe.connect(to_browser, from_browser)
+    try:
+        requested = asyncio.ensure_future(pipe.request("Page.getFrameTree", answered=lambda answer: heard.append("answer")))
+        await asyncio.sleep(0)
+        command = json.loads(os.read(from_us, 4096).rstrip(b"\0"))
+        os.write(to_us, json.dumps({"id": command["id"], "result": {}}).encode() + b"\0"
+                 + json.dumps({"method": "Runtime.executionContextCreated", "params": {}}).encode() + b"\0")
+        answer = await asyncio.wait_for(requested, 5)
+    finally:
+        pipe.close()
+        for fd in (to_us, from_us):
+            os.close(fd)
+
+    assert heard == ["answer", "Runtime.executionContextCreated"] and answer == {"id": command["id"], "result": {}}

packages/agentenv-protocol/tests/test_website_browser.py

@@ -517,7 +517,7 @@ async def call(self, method, params=None, *, session_id=None, timeout_s=None):
         answer = self.answers.get(method, {})
         return answer(params, session_id) if callable(answer) else answer
 
-    async def request(self, method, params=None, *, session_id=None):
+    async def request(self, method, params=None, *, session_id=None, answered=None):
         try:
             return {"result": await self.call(method, params, session_id=session_id)}
         except DevToolsError as e: