From 1dd43cb45afdf7db6eaa69d0d2f8994e88379325 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 11 Nov 2026 15:31:51 -0800 Subject: [PATCH] fix(browser): the CDP relay passes each answer on in order with the events around it The pipe's reader handed events to the relay as it read them, but an answer only reached its client once the command's task resumed, so an event the browser sent after an answer, read in the same chunk, could reach the client first. Playwright, given Runtime.executionContextCreated before Page.getFrameTree's answer, drops the context, and page.evaluate then waits for ever. A request's answered hook now hears its answer as the pipe reads it, and the relay sends its reply from there. Co-Authored-By: Claude Opus 5.5 --- .../agentenv_protocol/website_browser/cdp.py | 23 ++++++++++--- .../website_browser/devtools.py | 16 ++++++--- .../tests/test_browser_cdp.py | 20 +++++++++-- .../tests/test_browser_log.py | 2 +- .../tests/test_devtools_pipe.py | 33 +++++++++++++++++++ .../tests/test_website_browser.py | 2 +- 6 files changed, 83 insertions(+), 13 deletions(-) create mode 100644 packages/agentenv-protocol/tests/test_devtools_pipe.py diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py index 2795fcf84..ba3fcbf23 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py +++ b/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.""" diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py index 986097b85..098b56a08 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/devtools.py +++ b/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,18 +63,22 @@ 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") @@ -81,6 +86,7 @@ async def _answer(self, method: str, params: Optional[dict], session_id: Optiona 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: diff --git a/packages/agentenv-protocol/tests/test_browser_cdp.py b/packages/agentenv-protocol/tests/test_browser_cdp.py index 0ea2c4e9e..aeea8355c 100644 --- a/packages/agentenv-protocol/tests/test_browser_cdp.py +++ b/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``, an attach on a session making its child - ``-c`` and saying so on that session, ``Boom`` failing, ``Hang`` never answered, anything else echoed; + ``-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): diff --git a/packages/agentenv-protocol/tests/test_browser_log.py b/packages/agentenv-protocol/tests/test_browser_log.py index 11cff2fb3..00b4b0807 100644 --- a/packages/agentenv-protocol/tests/test_browser_log.py +++ b/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)} diff --git a/packages/agentenv-protocol/tests/test_devtools_pipe.py b/packages/agentenv-protocol/tests/test_devtools_pipe.py new file mode 100644 index 000000000..fe7e630af --- /dev/null +++ b/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": {}} diff --git a/packages/agentenv-protocol/tests/test_website_browser.py b/packages/agentenv-protocol/tests/test_website_browser.py index b049b4051..816337d61 100644 --- a/packages/agentenv-protocol/tests/test_website_browser.py +++ b/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: