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: