diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index 967a1bbc5..89fd05516 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -1000,7 +1000,35 @@ are the profile's template. The base class adds: and, completed, its size and sha256; and `dialogs`, each with its type, message, default text, answer and who gave it. `hidden_page()` gives agentenv a page of its own for reading state, in no window and never in the record. -- **`browser/v1`** on its card: `chromium` (its `version`), `clock: "real"` and, alone, `window`; and +- **Its log**, `self.log`, a `BrowserLog` (`agentenv_protocol.website_browser.log`): one entry for each action agents take on it + and an event each time a tab opens, navigates or closes. Its server writes it, the one writer, to + `/var/log/agentenv/browser-.jsonl` (its own `.agentenv/` folder when that can't be written), one JSON object a + line with its `seq` and `at` and one of `action`, `tab`, `outcome` or `reset`. A line's number is taken when what it + records begins and the line written when it ends, so `log {after}` returns lines in order up to the last whose every + earlier line is written (`last`), and `last` never moves back. An action still going after 60 s is written by the + next read as `unfinished`, so it holds back nothing after it, and once it ends an `outcome` line names it (`entry`) + with what it ended with. A relayed connection's commands reach the browser in the order they came. The file lasts the + computer's boot, a restarted server carrying on the sequence and the + browser's own reset (`data/reset`) being a line. A computer's `reset_env` boots it afresh, with a new log, so a task + that needs the one before takes it first, with `snapshot_env`. + - **An action** is a CDP command through the relay (`via: "cdp"`, `who` the connection's user, its `connection` + number, `method`, and the tab its session is on), an op a driver makes for an agent's MCP tool call (`via: "mcp"`, + `who` the caller's role, `tool`, `call`, `op`), or an acting op agentenv makes itself (`via: "agentenv"`). Each + carries `tab`, `url`, `ok` and, failed, `error`, and is marked `acts` or not by its method or op + (`log.acts`, `ACTING_OPS`): input, navigation, tab commands, `Runtime.evaluate` and `callFunctionOn`, DOM + writes, faked responses, and viewport and cookie changes act; the rest read. An acting entry keeps its `params` + (and a tool call's `args`) with what could carry a secret taken out: text typed while a password field has focus, + as its length; cookie and header values; the values of URL query keys that name a credential; a request's body; + and each string past 2 KB cut, with its length and sha256. A reading entry keeps none. + - **A tab event** (`opened`, `navigated`, `closed`) says what caused it: `by: "command"` with the `entry`, `via` and + `who` of the command that asked for it, matched exactly by the target a new tab's command answers with and the + loader a navigation's does; `by: "agentenv"`; `exit` or `stop`; or `by: "page"`, with the `after` of the latest + acting entry on its tab (its opener's, for a popup), for a click that follows a link, a redirect or a page script. + - A driver passes an agent's tool call to the browser in the `AgentEnv-Tool-Call` header (`tool_call_header`), with + the caller's `AgentEnv-Role`; the page driver sends it on every call it makes for a tool. `data/get` writes the log + beside `browser-state.json` as `browser-log.jsonl`, which `data/add` passes over. +- **`browser/v1`** on its card: `chromium` (its `version`), `clock: "real"` and, alone, `window`; `log {after}` + (GET), framework-only, its log's lines after `after`; and `record` (GET), framework-only, which returns that record with `launches`, the app server's log of each launch (with its cause: `boot`, `server_restart`, `launch`, `data/add` or `data/reset`), each stop it made and each exit it didn't, with its exit `code` or the `signal` that ended it, since the computer booted. diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py index f13b2f865..e594e33f9 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py @@ -539,6 +539,10 @@ async def _computer_app_launch(self) -> dict: async def _computer_app_ready(self) -> dict: return (await self._computer_app_runtime().readiness()).model_dump() + def served_asgi(self, asgi: Any) -> Any: + """The application its server serves, ``asgi`` as built from its card: a subclass may wrap it.""" + return asgi + def _computer_app_runtime(self) -> AppRuntime: if self._computer_app is None: raise TypeError(f"{type(self).__qualname__}.__init__ must call super().__init__()") diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py index 02d6a900d..cb605edcd 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py @@ -61,12 +61,13 @@ async def lifespan(scope): started.cancel() asgi.router.lifespan_context = lifespan + served = app.served_asgi(asgi) if LOOPBACK: listening, token = bind(Path.cwd()) - server = uvicorn.Server(uvicorn.Config(guarded(asgi, token), log_level="warning")) + server = uvicorn.Server(uvicorn.Config(guarded(served, token), log_level="warning")) server.run(sockets=[listening]) else: - server = uvicorn.Server(uvicorn.Config(asgi, uds=str(SOCKET), log_level="warning")) + server = uvicorn.Server(uvicorn.Config(served, uds=str(SOCKET), log_level="warning")) SOCKET.unlink(missing_ok=True) server.run() if failed: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/tool_context.py b/packages/agentenv-protocol/src/agentenv_protocol/tool_context.py index 97fcbe26b..757c35c6d 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/tool_context.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/tool_context.py @@ -97,6 +97,8 @@ class ToolContext: tool: str = "" arguments: Mapping[str, Any] = dataclasses.field(default_factory=lambda: types.MappingProxyType({}), hash=False) call_id: str = "" + #: Whether the call is an MCP tool call, rather than a REST or extension call. + mcp: bool = False @property def role(self) -> Optional[str]: @@ -189,7 +191,7 @@ def context_from_mcp(context: Any, tool: str, arguments: Mapping[str, Any]) -> T request = getattr(request_context, "request", None) headers = getattr(request, "headers", None) if request is not None else None return ToolContext(caller=caller_from_request(extra, headers), tool=tool, arguments=_frozen(arguments), - call_id=uuid.uuid4().hex) + call_id=uuid.uuid4().hex, mcp=True) def _meta_entries(meta: Any) -> Optional[Mapping[str, Any]]: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/browser_cdp.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py similarity index 66% rename from packages/agentenv-protocol/src/agentenv_protocol/browser_cdp.py rename to packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py index 0ca208cb7..2795fcf84 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/browser_cdp.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/cdp.py @@ -18,8 +18,11 @@ from starlette.requests import Request from starlette.websockets import WebSocket, WebSocketDisconnect +from .log import ERROR_CHARS, KEY_FIELDS, TYPING_METHODS, BrowserLog, acts, redact_text, redacted_params + if TYPE_CHECKING: - from .website_browser import DevToolsPipe + from .control import BrowserControl + from .devtools import DevToolsPipe logger = logging.getLogger(__name__) @@ -55,6 +58,10 @@ GUARDED_KEPT = 100 #: What the record says the guard did, by its action. GUARDED = {"refuse": "refused", "send": "changed", "answer": "answered", "close": "closed"} +#: The commands that navigate a tab without naming the loader they navigate with, whose next navigation is theirs. +NAVIGATES_NEXT = frozenset({"Page.reload", "Page.navigateToHistoryEntry"}) +#: What a command's entry says when its connection ended before the browser answered it. +ENDED_UNANSWERED = "the connection ended before the browser answered" @dataclass @@ -192,9 +199,14 @@ class _Connection: """One client's WebSocket, its sessions over the pipe (its browser-level one and those it attached), its unanswered commands, what waits to be sent to it, and its entry in the record.""" - def __init__(self, websocket: WebSocket, user: str, session: str, record: dict) -> None: - self.websocket, self.user, self.session, self.record = websocket, user, session, record + def __init__(self, websocket: WebSocket, user: str, session: str, record: dict, number: int) -> None: + self.websocket, self.user, self.session, self.record, self.number = websocket, user, session, record, number self.sessions = {session} + #: The tab each session it attached is on. + self.targets: dict[str, str] = {} + #: Held by each of its commands from when it arrives until it's sent, so they're sent in the order they came, + #: whatever one waits on before it's sent. + self.order = asyncio.Lock() self.commands: set[asyncio.Future] = set() self.outbox: asyncio.Queue[str | tuple[int, str]] = asyncio.Queue() self.waiting_bytes = 0 @@ -205,10 +217,13 @@ class Relay: """Each WebSocket connection given its own browser-level session over ``devtools``, its commands sent on it past the guard and the answers and events of its sessions sent back, so several clients share one browser; and a record of who connected when, and what the guard did to their commands. ``dialogs_answered`` says agentenv answers the - browser's dialogs itself.""" + browser's dialogs itself. With a ``log``, each command is an entry in it, on the tab its session is on, and + ``control``, agentenv's own connection, tells its tab's URL, whether a password field has focus there, and what + causes the tab events a command brings.""" - def __init__(self, devtools: DevToolsPipe, *, dialogs_answered: bool = False) -> None: - self.devtools, self.dialogs_answered = devtools, dialogs_answered + def __init__(self, devtools: DevToolsPipe, *, dialogs_answered: bool = False, log: Optional[BrowserLog] = None, + control: Optional[BrowserControl] = None) -> None: + self.devtools, self.dialogs_answered, self.log, self.control = devtools, dialogs_answered, log, control self.connections: list[dict] = [] self._by_session: dict[str, _Connection] = {} self._open: set[_Connection] = set() @@ -222,7 +237,7 @@ async def serve(self, websocket: WebSocket, user: str) -> None: await websocket.close(code=1013, reason=f"the browser isn't answering: {type(e).__name__}") return record = {"user": user, "opened_at": _now(), "guarded": []} - connection = _Connection(websocket, user, session, record) + connection = _Connection(websocket, user, session, record, len(self.connections) + 1) self._by_session[session] = connection self._open.add(connection) self.connections.append(record) @@ -263,25 +278,56 @@ async def _command(self, connection: _Connection, message: Any) -> None: method, params = message["method"], message.get("params") or {} session = message.get("sessionId") or connection.session verdict = guard(method, params, dialogs_answered=self.dialogs_answered) - if session not in connection.sessions: - verdict, answer = None, {"error": NO_SESSION} - elif verdict.action == "refuse": - answer = {"error": {"code": SERVER_ERROR, "message": f"agentenv refuses {method}: {verdict.reason}"}} - elif verdict.action != "send": - answer = {"result": {}} - else: - try: - answer = await self.devtools.request(method, verdict.params, session_id=session) - except ConnectionError as e: - answer = {"error": {"code": SERVER_ERROR, "message": str(e)}} - if verdict is not None and (verdict.action != "send" or verdict.changed): - 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 verdict is not None and verdict.action == "close": - self._close(connection, CLOSED, dropping=False) + logged = self._begin(connection, session, method, params) + answer, guarded = {"error": {"code": SERVER_ERROR, "message": ENDED_UNANSWERED}}, None + sending: Optional[asyncio.Future] = None + try: + async with connection.order: + if logged is not None: + await logged.check_password() + if session not in connection.sessions: + verdict, answer = None, {"error": NO_SESSION} + elif verdict.action == "refuse": + answer = {"error": {"code": SERVER_ERROR, + "message": f"agentenv refuses {method}: {verdict.reason}"}} + elif verdict.action != "send": + answer = {"result": {}} + else: + sending = asyncio.ensure_future(self.devtools.request(method, verdict.params, session_id=session)) + await asyncio.sleep(0) + if sending is not None: + try: + answer = await sending + except ConnectionError as e: + answer = {"error": {"code": SERVER_ERROR, "message": str(e)}} + 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 verdict is not None and verdict.action == "close": + self._close(connection, CLOSED, dropping=False) + finally: + if sending is not None and not sending.done(): + sending.cancel() + if logged is not None: + logged.finish(answer, guarded) + + def _begin(self, connection: _Connection, session: str, method: str, params: dict) -> Optional[_Logged]: + """``connection``'s command, begun in the log, if there is one, on the tab ``session`` is on, or for a browser + command the tab it names; what it causes made known to the control.""" + if self.log is None: + return None + target = connection.targets.get(session) if session != connection.session else None + if target is None and isinstance(params.get("targetId"), str): + target = params["targetId"] + acting = acts(method) + url = self.control.url_of(target) if self.control is not None and target else None + begun = self.log.begin({"via": "cdp", "who": connection.user, "connection": connection.number, + "method": method, "acts": acting, "tab": target, "url": url}) + return _Logged(self, begun, method, params, target, acting) @staticmethod def _guarded(connection: _Connection, method: str, verdict: Verdict) -> None: @@ -297,8 +343,11 @@ def _event(self, message: dict) -> None: if message.get("method") == "Target.attachedToTarget" and params.get("sessionId"): connection.sessions.add(params["sessionId"]) self._by_session[params["sessionId"]] = connection + if isinstance(target := (params.get("targetInfo") or {}).get("targetId"), str): + connection.targets[params["sessionId"]] = target elif message.get("method") == "Target.detachedFromTarget" and params.get("sessionId"): connection.sessions.discard(params["sessionId"]) + connection.targets.pop(params["sessionId"], None) self._by_session.pop(params["sessionId"], None) if message["sessionId"] == connection.session: message = {key: value for key, value in message.items() if key != "sessionId"} @@ -343,6 +392,70 @@ async def _send(self, connection: _Connection) -> None: await connection.websocket.send_text(message) +class _Logged: + """A relayed command's entry in the log, from when it arrives until it's answered or its connection ends: what it + causes made known to the control as it's sent, whether a password field has focus on its tab before it types, + and its answer.""" + + def __init__(self, relay: Relay, begun: dict, method: str, params: dict, target: Optional[str], + acting: bool) -> None: + self.relay, self.begun, self.method, self.params, self.target, self.acting = \ + relay, begun, method, params, target, acting + control = relay.control + self._cause = cause = {"seq": begun["seq"], "via": "cdp", "who": begun["action"]["who"]} + self._landed: Optional[asyncio.Future] = None + self._checks_password = self.password = False + if control is None or not acting: + return + if method == "Target.createTarget": + self._landed = asyncio.get_running_loop().create_future() + control.causes.creating(self._landed, cause) + elif method == "Target.closeTarget" and isinstance(params.get("targetId"), str): + control.causes.closing(params["targetId"], cause) + elif method == "Page.navigate" and target: + self._landed = asyncio.get_running_loop().create_future() + control.causes.navigating(target, self._landed, cause) + elif method in NAVIGATES_NEXT and target: + control.causes.navigates_next(target, cause) + self._checks_password = method in TYPING_METHODS and bool(target) and any(field in params + for field in KEY_FIELDS) + #: What a command that types types is withheld until its check says no password field has focus, so one + #: that never gets checked, its connection gone first, keeps it withheld. + self.password = self._checks_password + + async def check_password(self) -> None: + """Before a command that types is sent: whether a password field has focus on its tab, true when that can't + be told.""" + if self._checks_password: + self.password = await self.relay.control.password_focused(self.target) + + def finish(self, answer: dict, guarded: Optional[str]) -> None: + """The entry written with its answer, or with why there is none: whether it worked, why not, what the guard + did, and, acting, its parameters as the log keeps them; what it caused made known.""" + result = answer.get("result") or {} + if self._landed is not None and not self._landed.done(): + if "error" in answer or result.get("errorText"): + self._landed.cancel() + else: + self._landed.set_result(result.get("targetId") if self.method == "Target.createTarget" + else result.get("loaderId")) + fields: dict[str, Any] = {"ok": "error" not in answer} + if "error" in answer: + fields["error"] = redact_text(str((answer["error"] or {}).get("message", answer["error"])))[:ERROR_CHARS] + if self.method in NAVIGATES_NEXT and self.target and self.relay.control is not None: + self.relay.control.causes.navigated_nowhere(self.target, self._cause) + if guarded is not None: + fields["guard"] = guarded + if self.acting: + fields["params"] = redacted_params(self.method, self.params, password=self.password) + opened = result.get("targetId") if self.method == "Target.createTarget" else None + if isinstance(opened, str): + fields["tab"] = self.target = opened + if self.target and self.relay.control is not None and self.begun["action"].get("url") is None: + fields["url"] = self.relay.control.url_of(self.target) + self.relay.log.finish(self.begun, **fields) + + 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/control.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py index 65545482b..03ac3e25f 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py @@ -14,6 +14,7 @@ from typing import Any, Awaitable, Callable, Optional from urllib.parse import urlsplit +from .log import ACTOR, KEY_FIELDS, TYPING_METHODS, BrowserLog, Causes, caused, log_time, redact_url from ..computer.errors import DeviceUnavailable, InvalidArgument from .devtools import DevToolsError, DevToolsPipe from .shapes import (ClosedTab, Cookie, CookieToSet, Dialog, Download, OpenTab, OriginStorage, StartTab, StorageItem, @@ -53,6 +54,20 @@ LEAVING, ARRIVED = "window.__agentenvLeft = true", "window.__agentenvLeft ? '' : location.origin" #: Hostnames a secure cookie is sent to over http, as browsers treat them. LOCAL_HOSTS = ("localhost", "127.0.0.1", "[::1]") +#: Whether the focused element is a password field, through shadow roots and frames of the page's own origin; a frame +#: of another origin can't be looked into. +PASSWORD_FOCUSED = """(() => { + let element = document.activeElement; + for (;;) { + if (element && element.shadowRoot && element.shadowRoot.activeElement) element = element.shadowRoot.activeElement; + else if (element && (element.tagName === 'IFRAME' || element.tagName === 'FRAME')) { + try { element = element.contentDocument.activeElement; } catch (e) { return false; } + } else break; + } + return !!element && element.tagName === 'INPUT' && (element.type || '').toLowerCase() === 'password'; +})()""" +#: How long the password-field check waits on a page. +PASSWORD_CHECK_TIMEOUT_S = 2.0 class BrowserControl: @@ -91,6 +106,10 @@ def __init__(self, devtools: DevToolsPipe, dialogs: str, viewport: Optional[tupl #: of agentenv's own pages whose requests it answers itself. self.origins: set[str] = set() self._serving: set[str] = set() + #: The browser's log, where its tab events go, and the commands in flight that cause them. + self.log: Optional[BrowserLog] = None + self.causes = Causes() + self._launching = False devtools.listen(self._event) async def attach(self) -> None: @@ -98,7 +117,11 @@ async def attach(self) -> None: this one has or opens attached, and its downloads reported, which go where its own policies say, as Chrome downloads unattended: into ``~/Downloads``, a name downloaded again kept beside the first.""" self.ended(STOP) - await self.devtools.call("Target.setDiscoverTargets", {"discover": True, "filter": PAGES}) + self._launching = True + try: + await self.devtools.call("Target.setDiscoverTargets", {"discover": True, "filter": PAGES}) + finally: + self._launching = False await self.devtools.call("Target.setAutoAttach", {"autoAttach": True, "waitForDebuggerOnStart": self.viewport is not None, "flatten": True, "filter": PAGES}) @@ -108,6 +131,9 @@ def ended(self, how: str) -> None: """The browser is gone, ``how`` it ended: its open tabs are closed by that, and its sessions with them.""" now = _now() self.closed_tabs += [{**tab, "closed_at": now, "closed_by": how} for tab in self.tabs.values()] + if self.log is not None: + for tab in self.tabs.values(): + self.log.note("tab", {"event": "closed", "id": tab["id"], "url": redact_url(tab["url"]), "by": how}) self.tabs.clear() self._sessions.clear() self._open_dialogs.clear() @@ -120,10 +146,66 @@ def ended(self, how: str) -> None: load.settled.set() async def close_tab(self, target: str) -> None: - """Close the tab ``target``, which the record says agentenv closed.""" + """Close the tab ``target``, which the record says agentenv closed, and the log says who asked.""" self._closing.add(target) + self.causes.closing(target, _cause()) await self.devtools.call("Target.closeTarget", {"targetId": target}) + async def create_target(self, params: dict) -> str: + """A new tab, ``Target.createTarget`` with ``params``, which the log says who asked for; its id.""" + landed = asyncio.get_running_loop().create_future() + self.causes.creating(landed, _cause()) + try: + target = (await self.devtools.call("Target.createTarget", params))["targetId"] + except BaseException: + landed.cancel() + raise + landed.set_result(target) + return target + + async def _navigate(self, target: str, method: str, params: dict, session: str) -> dict: + """``Page.navigate`` of ``target`` on ``session``, which the log says who asked for; its answer.""" + landed = asyncio.get_running_loop().create_future() + self.causes.navigating(target, landed, _cause()) + try: + answer = await self.devtools.call(method, params, session_id=session) + except BaseException: + landed.cancel() + raise + if answer.get("errorText"): + landed.cancel() + else: + landed.set_result(answer.get("loaderId")) + return answer + + def url_of(self, target: Optional[str]) -> Optional[str]: + """The URL of the open tab ``target`` as the log keeps it; None for one it doesn't know.""" + tab = self.tabs.get(target) if target else None + return redact_url(tab["url"]) if tab is not None else None + + async def password_focused(self, target: str) -> bool: + """Whether a password field has focus in the tab ``target``, so what is typed there isn't logged; True when it + can't be told, as for a tab it has no session on.""" + session = next((s for s, t in self._sessions.items() if t == target), None) + if session is None: + return True + try: + answer = await self.devtools.call("Runtime.evaluate", {"expression": PASSWORD_FOCUSED, + "returnByValue": True, "silent": True}, + session_id=session, timeout_s=PASSWORD_CHECK_TIMEOUT_S) + return answer["result"]["value"] is not False + except (ConnectionError, DevToolsError, asyncio.TimeoutError, KeyError, TypeError): + return True + + def _acted_on(self, target: Optional[str]) -> None: + """The tab the op under way acts on or reads, for its entry; an acting op's becomes that tab's latest.""" + actor = ACTOR.get() + if actor is None or target is None: + return + actor["tab"] = target + if actor.get("acts") and self.log is not None and self.log.last_acting.get(target, 0) < actor["seq"]: + self.log.last_acting[target] = actor["seq"] + async def tab_list(self) -> list[dict]: """The open tabs in opened order, each with its URL and title now and whether it's active.""" await self._refresh() @@ -137,11 +219,12 @@ async def open_tab(self, url: Optional[str] = None, background: bool = False) -> active = await self.active_tab() if background else None held = self.viewport is not None and url start = "about:blank" if held else url or "about:blank" - target = (await self.devtools.call("Target.createTarget", {"url": start, "background": background}))["targetId"] + target = await self.create_target({"url": start, "background": background}) + self._acted_on(target) if held: await asyncio.wait_for(self._pinned.setdefault(target, asyncio.Event()).wait(), PIN_TIMEOUT_S) session = next(s for s, t in self._sessions.items() if t == target) - await self.devtools.call("Page.navigate", {"url": url}, session_id=session) + await self._navigate(target, "Page.navigate", {"url": url}, session) if not background: await self._bring_to_front(target) elif active is not None and active in self.tabs and await self.active_tab() == target: @@ -161,8 +244,7 @@ async def load_start_tabs(self, urls: list[str], site_of: Callable[[str], Option return [] targets = list(self.tabs)[:len(urls)] or [await self._first_tab()] while len(targets) < len(urls): - targets.append((await self.devtools.call("Target.createTarget", - {"url": "about:blank", "background": True}))["targetId"]) + targets.append(await self.create_target({"url": "about:blank", "background": True})) for target in targets: await asyncio.wait_for(self._pinned.setdefault(target, asyncio.Event()).wait(), PIN_TIMEOUT_S) loading = [] @@ -198,9 +280,12 @@ async def _start_load(self, target: str, url: str, site: Optional[str]) -> _Star load.error = f"{type(e).__name__}: {e}" load.check() return load + landed = asyncio.get_running_loop().create_future() + self.causes.navigating(target, landed, _cause()) load.navigating = asyncio.ensure_future(self.devtools.request("Page.navigate", {"url": url}, session_id=session)) load.navigating.add_done_callback(load.answered) + load.navigating.add_done_callback(lambda done: _landed_at(landed, done)) return load async def _settled(self, target: str, load: _StartLoad, timeout_s: float) -> dict: @@ -248,24 +333,29 @@ def forget(self) -> None: async def activate(self, target: str) -> None: """The open tab ``target`` brought to the front, the active one.""" self._open(target) + self._acted_on(target) await self._bring_to_front(target) async def close(self, target: str) -> None: """The open tab ``target`` closed, which the record says agentenv closed.""" self._open(target) + self._acted_on(target) await self.close_tab(target) async def navigate(self, url: str) -> None: """The active tab opened at ``url``; InvalidArgument, with the browser's reason, when it can't.""" - answer = await self.devtools.call("Page.navigate", {"url": url}, session_id=await self.active_session()) + session = await self.active_session() + self._acted_on(target := self._sessions[session]) + answer = await self._navigate(target, "Page.navigate", {"url": url}, session) if error := answer.get("errorText"): raise InvalidArgument(f"the browser couldn't open {url}: {error}") async def history(self, direction: str) -> None: """The active tab's page reloaded, or the one before or after it in its history opened.""" session = await self.active_session() + self._acted_on(target := self._sessions[session]) if direction == "reload": - await self.devtools.call("Page.reload", {}, session_id=session) + await self._moves(target, "Page.reload", {}, session) return if direction not in ("back", "forward"): raise InvalidArgument(f"direction is back, forward or reload, not {direction!r}") @@ -273,12 +363,24 @@ async def history(self, direction: str) -> None: index = history["currentIndex"] + (-1 if direction == "back" else 1) if not 0 <= index < len(history["entries"]): raise InvalidArgument(f"the active tab has no page to go {direction} to") - await self.devtools.call("Page.navigateToHistoryEntry", {"entryId": history["entries"][index]["id"]}, - session_id=session) + await self._moves(target, "Page.navigateToHistoryEntry", {"entryId": history["entries"][index]["id"]}, + session) + + async def _moves(self, target: str, method: str, params: dict, session: str) -> None: + """A reload or move in history of ``target``, whose next navigation the log says the op under way caused, + unless it fails.""" + cause = _cause() + self.causes.navigates_next(target, cause) + try: + await self.devtools.call(method, params, session_id=session) + except BaseException: + self.causes.navigated_nowhere(target, cause) + raise async def answer_dialog(self, accept: bool, text: Optional[str] = None) -> None: """The dialog open on the active tab answered, as the agent's answer; InvalidArgument with none open.""" active = await self.active_tab() + self._acted_on(active) session = next((s for s, dialog in self._open_dialogs.items() if dialog["tab"] == active), None) if session is None: raise InvalidArgument("no dialog is open on the page") @@ -289,6 +391,7 @@ async def choose_files(self, paths: list[str]) -> None: """The file chooser open on the active tab given ``paths``, files in the session user's ``~/Uploads``; InvalidArgument for one outside it or missing, for several where it takes one, or with no chooser open.""" active = await self.active_tab() + self._acted_on(active) session = next((s for s in self._choosers if self._sessions.get(s) == active), None) if session is None: raise InvalidArgument("no file chooser is open on the page") @@ -314,6 +417,7 @@ async def state(self) -> dict: active = await self.active_tab() if active is None: raise DeviceUnavailable("the browser has no tab open") + self._acted_on(active) return self._state_of(active) def _state_of(self, target: str) -> dict: @@ -504,6 +608,7 @@ async def capture_page(self) -> dict: active meanwhile.""" session = await self.active_session() target = self._sessions[session] + self._acted_on(target) png = await self._capture(session) await self._refresh(target) return {**({"png": png} if png is not None else {"blank": True}), "page": self._state_of(target)} @@ -531,8 +636,12 @@ async def dispatch(self, events: list[dict]) -> None: if refused := [event.get("method") for event in events if event.get("method") not in PAGE_INPUT]: raise InvalidArgument(f"page input takes {', '.join(sorted(PAGE_INPUT))}, not {refused}") session = await self.active_session() + self._acted_on(target := self._sessions.get(session)) + actor = ACTOR.get() opened = self._dialog_opened.setdefault(session, asyncio.Event()) for event in events: + if actor is not None and target is not None and actor["password"] is not True and _types(event): + actor["password"] = await self.password_focused(target) if self.dialogs == AGENT and session in self._open_dialogs: return opened.clear() @@ -579,6 +688,34 @@ def _on_Target_targetCreated(self, params: dict, session: Optional[str]) -> None "title": info.get("title", ""), "opened_at": _now(), "opener": info.get("openerId")} self._saw(info.get("url", "")) + if self.log is not None: + self._spawn(self._log_opened(self.log.reserve(), log_time(), info["targetId"], info.get("url", ""), + info.get("openerId"), self._launching)) + + async def _log_opened(self, seq: int, at: str, target: str, url: str, opener: Optional[str], + launching: bool) -> None: + """The tab ``target`` opened, in the log as ``seq``, once the commands in flight that may have opened it have + answered: by the command that did, by agentenv as the browser launched, else by its page, its opener's.""" + cause = None + try: + cause = {"via": "agentenv"} if launching else await self.causes.opened(target) + finally: + self.log.write(seq, {"tab": {"event": "opened", "id": target, "url": redact_url(url), + **({"opener": opener} if opener else {}), + **caused(cause, self.log, opener or target)}}, at=at) + + async def _log_navigated(self, seq: int, at: str, target: str, url: str, loader: Optional[str] | bool) -> None: + """The tab ``target`` navigated to ``url``, in the log as ``seq``, once its navigations in flight have + answered: by the command whose loader it is, else by its page; with ``loader`` False, within its + document.""" + cause = None + try: + cause = await (self.causes.navigated(target, loader) if loader is not False else + self.causes.navigated_within(target)) + finally: + self.log.write(seq, {"tab": {"event": "navigated", "id": target, "url": redact_url(url), + **({"within_document": True} if loader is False else {}), + **caused(cause, self.log, target)}}, at=at) def _on_Target_targetInfoChanged(self, params: dict, session: Optional[str]) -> None: info = params["targetInfo"] @@ -590,10 +727,20 @@ def _on_Page_frameNavigated(self, params: dict, session: Optional[str]) -> None: if session in self._sessions: self._saw(params.get("frame", {}).get("url", "")) frame = params.get("frame", {}) + if self.log is not None and not frame.get("parentId") and (target := self._sessions.get(session)) in self.tabs: + self._spawn(self._log_navigated(self.log.reserve(), log_time(), target, frame.get("url", ""), + frame.get("loaderId"))) if (load := self._loading.get(session)) is not None and not frame.get("parentId") and frame.get("loaderId"): load.commits.append(frame["loaderId"]) load.check() + def _on_Page_navigatedWithinDocument(self, params: dict, session: Optional[str]) -> None: + """A tab's page moving to another URL of its document, a fragment or its history API: a command's, when one + asked for it, else its page's.""" + target = self._sessions.get(session) + if self.log is not None and target in self.tabs and params.get("frameId") == target: + self._spawn(self._log_navigated(self.log.reserve(), log_time(), target, params.get("url", ""), False)) + def _on_Network_requestWillBeSent(self, params: dict, session: Optional[str]) -> None: if (load := self._loading.get(session)) is not None and params.get("type") == "Document": load.requests[params["requestId"]] = params.get("loaderId", "") @@ -626,6 +773,11 @@ def _on_Target_targetDestroyed(self, params: dict, session: Optional[str]) -> No if (tab := self.tabs.pop(target, None)) is not None: self.closed_tabs.append({**tab, "closed_at": _now(), "closed_by": AGENTENV if target in self._closing else UNKNOWN}) + if self.log is not None: + self.log.note("tab", {"event": "closed", "id": target, "url": redact_url(tab["url"]), + **caused(self.causes.closed(target), self.log, target)}) + else: + self.causes.closed(target) self._closing.discard(target) self._hidden.discard(target) @@ -782,6 +934,30 @@ def check(self) -> None: self.settled.set() +def _landed_at(landed: asyncio.Future, navigating: asyncio.Future) -> None: + """``landed`` given the loader a start tab's navigation answered with; cancelled when it failed or named none.""" + if landed.done(): + return + answer = None if navigating.cancelled() or navigating.exception() is not None else navigating.result() + loader = ((answer or {}).get("result") or {}).get("loaderId") + if isinstance(loader, str): + landed.set_result(loader) + else: + landed.cancel() + + +def _types(event: dict) -> bool: + """Whether a page input event types something.""" + return event.get("method") in TYPING_METHODS and any(field in (event.get("params") or {}) for field in KEY_FIELDS) + + +def _cause() -> dict: + """Who the control's command is for: the op under way, else agentenv itself.""" + actor = ACTOR.get() + return {"seq": actor["seq"], "via": actor["via"], "who": actor.get("who")} if actor is not None else \ + {"via": "agentenv"} + + def _size_and_sha256(path: Path) -> tuple[int, str]: digest = hashlib.sha256() with path.open("rb") as file: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/data.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/data.py index bc418e53a..f9fc1ee81 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/data.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/data.py @@ -16,6 +16,9 @@ #: The browser's typed state, as ``data/get`` returns it beside its files and ``data/add`` reads it back. STATE_FILE = "browser-state.json" +#: The browser's log, as ``data/get`` returns it beside its state; ``data/add`` passes over it, since a log isn't +#: loaded back. +LOG_FILE = "browser-log.jsonl" #: What a copy of the profile leaves out: caches, which the browser rebuilds, and the locks a running browser holds, #: which would keep another from opening the copy. CACHES = frozenset({"Cache", "Code Cache", "GPUCache", "ShaderCache", "GrShaderCache", "GraphiteDawnCache", @@ -51,6 +54,8 @@ async def sorted_load(parts: list, profile: str) -> Load: under = PurePosixPath(path) if under in (PurePosixPath(profile), PurePosixPath(DOWNLOADS), PurePosixPath(UPLOADS)): raise ValueError(f"{path!r} is a folder of the browser's, not a file in it") + if path == LOG_FILE: + continue if path == STATE_FILE: if load.state is not None: raise ValueError(f"data/add takes the browser's state once, as a data part or as {STATE_FILE}") diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py index 0a507c32a..40e852d82 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py @@ -9,7 +9,7 @@ import tempfile from pathlib import Path from typing import Any, Awaitable, Callable, ClassVar, Literal, Optional -from urllib.parse import urlsplit +from urllib.parse import parse_qsl, urlsplit from pydantic import BaseModel, Field, ValidationError, field_validator from starlette.requests import Request @@ -18,15 +18,18 @@ from ..agent_env_environment import (OP_ADD_DATA, add_data, add_websocket_route, agent_skill, extension, get_data, reset_data) -from ..browser_cdp import ROUTE, Relay, Tokens, reached, under, websocket_form +from .cdp import ROUTE, Relay, Tokens, reached, under, websocket_form +from .log import (ACTOR, ERROR_CHARS, TOOL_CALL_HEADER, UNLOGGED_OPS, BrowserLog, op_acts, read_tool_call, + redact_text, redacted_args, redacted_op) from ..computer import AppDeclaration, ComputerAppEnvironment, Launch, install_step, ready_check from ..computer import app as computer_app from ..computer import app_data from ..computer.app import SOCKET, agent_acted, forget_agent_acted from ..computer.dataplane import file_parts from ..computer.errors import DeviceUnavailable, InvalidArgument +from ..tool_context import DEFAULT_ROLE, ROLE_HEADER, normalize_role from ..transfers import WriteNamespaceGrant -from ..types import METHOD_ADD, EnvironmentExtension, IntakeDeclaration, IntakeFormat, intake_extension +from ..types import METHOD_ADD, RPC_PATH, EnvironmentExtension, IntakeDeclaration, IntakeFormat, intake_extension from . import data from .control import AGENT, DOWNLOADS, EXIT, LIVE_START_TAB_TIMEOUT_S, LOADED, STOP, UPLOADS, BrowserControl, _now from .devtools import PIPE_IN, PIPE_OUT, DevToolsError, DevToolsPipe @@ -36,6 +39,10 @@ SITE_PROXY_ENV = "AGENTENV_SITE_PROXY" BROWSER = "urn:agentenv:browser/v1" +#: Where ``browser/v1``'s ops are served on its server. +OPS_PATH = f"{RPC_PATH}/ext/browser/" +#: The largest op body whose parameters its entry keeps; a longer one keeps only its length. +LOGGED_BODY_BYTES = 1 << 20 #: Where its server keeps the env's sites ``sites/set`` hands it, which outlive a relaunch or the server's restart. SITES = SOCKET.parent / "sites.json" #: The flags agentenv sets, by prefix, which a subclass's command may not carry: among them every way to open a debugging @@ -140,9 +147,14 @@ def __init__(self) -> None: self.control = BrowserControl(self.devtools, self.settings.dialogs, (self.settings.window.width, self.settings.window.height) if _alone() else None) + #: Its log: an entry for each action agents take on it, and its tabs' events. + self.log = BrowserLog.for_app(os.environ.get("ENVIRONMENT_NAME") or type(self).__name__.lower(), + SOCKET.parent) + self.control.log = self.log #: The CDP addresses it has handed out, and the relay agents' connections to them go through. self.cdp_tokens = Tokens(SOCKET.parent / "cdp-tokens.json") - self.relay = Relay(self.devtools, dialogs_answered=self.settings.dialogs != AGENT) + self.relay = Relay(self.devtools, dialogs_answered=self.settings.dialogs != AGENT, log=self.log, + control=self.control) self.devtools.on_close(self._pipe_closed) @classmethod @@ -269,7 +281,7 @@ def _launch_running(self) -> Callable[[], Awaitable[None]]: async def _open_window(self) -> None: """Another window, which a launch of the browser while it runs opens over its pipe: a second copy would only hand its window to the first and exit, taking its new pipe with it.""" - await self.devtools.call("Target.createTarget", {"url": "about:blank", "newWindow": True}) + await self.control.create_target({"url": "about:blank", "newWindow": True}) @install_step(scope="system") def write_policies(self) -> None: @@ -316,12 +328,19 @@ async def start_tabs_loaded(self) -> bool: "sites/set hands it the env's own sites, by name and domains: a start tab on one must load, its document's " "status 2xx, or its launch counts it failed, while one on any other site is waited on 30 s at most. record " "keeps each launch's start_tabs: the URL asked for and ended on, the status, any network error, the load time, " - "the env's site and the outcome, loaded, failed or timed_out.")) + "the env's site and the outcome, loaded, failed or timed_out. log returns its log's lines after the sequence " + "number after, in order: an action entry for each CDP command, MCP tool call and pixel action an agent took, " + "marked acting or reading, with who, the tab and its URL and, acting, its parameters with secrets taken out; " + "and a tab event each time a tab opened, navigated or closed, with what caused it.")) async def browser_record(self) -> dict: return BrowserRecord(**await self.control.record(), launches=self._computer_app.launches(), cdp_connections=list(self.relay.connections), start_tabs=self.start_tab_launches).model_dump() + @extension(BROWSER, op="log", method="GET") + async def browser_log(self, after: int = 0) -> dict: + return await self.log.read_async(after) + @extension(BROWSER, op="sites/set") async def browser_sites_set(self, sites: list[dict]) -> dict: """The env's own sites, each by its name and domains, which it keeps for each launch after: a start tab on one @@ -344,6 +363,9 @@ async def browser_cdp(self, user: str, request: Request, origin: Optional[str] = address = self.cdp_tokens.mint(user, base, named=origin is not None) return {"http_url": address.http_url, "ws_url": address.ws_url} + def served_asgi(self, asgi: Any) -> Any: + return _LoggedOps(super().served_asgi(asgi), self) + def create_app(self) -> Any: app = super().create_app() for path in ("", "/", "/version", "/version/", "/list", "/list/"): @@ -495,14 +517,15 @@ def _resets_failed_add(self) -> bool: return not agent_acted() def declared_extensions(self) -> list[EnvironmentExtension]: - """Its data plane moves files by their path, a file tree: its state as ``browser-state.json``, its profile, its - downloads and files for agents to upload. A load takes its state as a data part too.""" + """Its data plane moves files by their path, a file tree: its state as ``browser-state.json``, its log, its + profile, its downloads and files for agents to upload. A load takes its state as a data part too.""" tree = IntakeDeclaration(get=[IntakeFormat(part="file", format="file-tree")], add=[IntakeFormat(part="file", format="file-tree", load="replace"), IntakeFormat(part="data", format="json", load="replace")]) return [*super().declared_extensions(), intake_extension(tree, ( f"data/get returns the browser's state as {data.STATE_FILE}: its cookies and local storage as Playwright's " - f"storageState has them, its tabs open and closed, its downloads and its clock; beside it, its profile " + f"storageState has them, its tabs open and closed, its downloads and its clock; beside it, its log as " + f"{data.LOG_FILE}, which data/add passes over, its profile " f"without caches, and its finished downloads. data/add applies such a state, while the browser runs, and " f"writes files by their path: its profile, restored with the browser stopped, ~/{DOWNLOADS} and " f"~/{UPLOADS}. data/reset puts the profile back as installed and relaunches it at its start tabs."))] @@ -523,9 +546,10 @@ async def browser_get_data(self, write_namespace: Optional[WriteNamespaceGrant] with tempfile.TemporaryDirectory(dir=home / ".cache") as staging: root = Path(staging) (root / data.STATE_FILE).write_text(json.dumps(written, indent=1), encoding="utf-8") + (root / data.LOG_FILE).write_text(await self.log.complete_text(), encoding="utf-8") with self._computer_app.paused(): await app_data.in_thread(data.stage_profile, home, self.profile, root) - parts = await file_parts(root, [data.STATE_FILE, self.profile], write_namespace) + parts = await file_parts(root, [data.STATE_FILE, data.LOG_FILE, self.profile], write_namespace) return parts + await file_parts(home, data.finished_downloads(home), write_namespace) @add_data @@ -580,6 +604,7 @@ async def browser_reset_data(self) -> None: a record that starts now. Its data plane launches it again after, at its start tabs.""" self.relay.ended() self.cdp_tokens.revoke_all() + self.log.note("reset", {}) await app_data.reset(self.paths) await app_data.in_thread(data.empty, Path.home(), DOWNLOADS, UPLOADS) self.control.forget() @@ -616,6 +641,97 @@ def browser_skill(self) -> tuple[str, str]: return "How to use this website browser.", SKILL +class _LoggedOps: + """Its server's application, with each ``browser/v1`` op an entry in its log as it runs: every op a driver makes + for an agent's tool call, which carries that call, and every acting op agentenv makes itself. While it runs, the op + is the action the browser's control credits the tab events it causes to.""" + + def __init__(self, app: Any, browser: WebsiteBrowserEnvironment) -> None: + self.app, self.browser = app, browser + + async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None: + op = scope["path"][len(OPS_PATH):] if scope["type"] == "http" and scope["path"].startswith(OPS_PATH) else None + if op is None or op in UNLOGGED_OPS: + return await self.app(scope, receive, send) + headers = {name.decode("latin-1").lower(): value.decode("latin-1") for name, value in scope.get("headers", [])} + call, acting = read_tool_call(headers.get(TOOL_CALL_HEADER.lower())), op_acts(op) + if call is None and not acting: + return await self.app(scope, receive, send) + chunks, more = [], True + while more: + message = await receive() + chunks.append(message.get("body", b"")) + more = message.get("more_body", False) and message["type"] == "http.request" + body = b"".join(chunks) + action: dict[str, Any] = {"via": "mcp" if call else "agentenv", "op": op, "acts": acting} + if call is not None: + action.update(who=normalize_role(headers.get(ROLE_HEADER.lower())) or DEFAULT_ROLE, tool=call["tool"], + call=call["call"]) + log = self.browser.log + begun = log.begin(action) + actor = {"seq": begun["seq"], "via": action["via"], "who": action.get("who"), "acts": acting, "tab": None, + "password": None} + answered: dict[str, Any] = {"status": 500, "error": b""} + replayed = False + + async def receiving() -> dict: + nonlocal replayed + if not replayed: + replayed = True + return {"type": "http.request", "body": body, "more_body": False} + return await receive() + + async def sending(message: dict) -> None: + if message["type"] == "http.response.start": + answered["status"] = message["status"] + elif message["type"] == "http.response.body" and answered["status"] >= 400 and \ + len(answered["error"]) < ERROR_CHARS * 4: + answered["error"] += message.get("body", b"") + await send(message) + + token = ACTOR.set(actor) + try: + await self.app(scope, receiving, sending) + finally: + ACTOR.reset(token) + fields: dict[str, Any] = {"tab": actor["tab"], "url": self.browser.control.url_of(actor["tab"]), + "ok": answered["status"] < 400} + if answered["status"] >= 400: + fields["error"] = _error_text(answered["error"]) + # What it typed is kept only once every check it made found no password field focused; one that failed, + # or never got to check, keeps it withheld. + withheld_typing = actor["password"] is not False or answered["status"] >= 400 + if acting and len(body) > LOGGED_BODY_BYTES: + fields["params"] = {"too_long": True, "length": len(body)} + elif acting: + fields["params"] = redacted_op(op, _op_params(scope, body), password=withheld_typing) + if call is not None: + fields["args"] = redacted_args(call["tool"], call["args"], password=withheld_typing) + log.finish(begun, **fields) + + +def _op_params(scope: dict, body: bytes) -> dict: + """An op's parameters: its query's for a GET, else its JSON body's.""" + if scope.get("method") == "GET": + return dict(parse_qsl(scope.get("query_string", b"").decode("latin-1"))) + try: + params = json.loads(body) if body else {} + except ValueError: + return {} + return params if isinstance(params, dict) else {} + + +def _error_text(body: bytes) -> str: + """What an op's error answer says, its message when it's agentenv's error shape.""" + text = body.decode(errors="replace") + try: + error = json.loads(text).get("error") + text = str(error.get("message", error)) if isinstance(error, dict) else str(error or text) + except (ValueError, AttributeError): + pass + return redact_text(text)[:ERROR_CHARS] + + def _read_sites() -> list[SiteNames]: """The env's sites ``sites/set`` handed it, none before it did or for a file it can't read.""" try: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py new file mode 100644 index 000000000..34e88e778 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py @@ -0,0 +1,549 @@ +"""A website browser's log: one entry for each action agents take on it, through CDP, MCP tools or pixels, and an +event each time a tab opens, navigates or closes, with what caused it. Its browser's server writes it, as JSON lines +with a sequence number each, in a file that lasts the computer's boot: a server that restarts carries on the sequence, +and the browser's own reset is a line, not a new file. A computer's reset boots it afresh, with a new log; a task +that needs the one before takes it first, with ``snapshot_env``. + +An entry is acting or reading by its CDP method or ``browser/v1`` op. An acting entry keeps its parameters with what +could carry a secret taken out: text typed while a password field has focus, cookie and header values, the values of +URL query keys that name a credential, and a request's body; a long value is cut, with its length and sha256. A +reading entry keeps none.""" +from __future__ import annotations + +import asyncio +import base64 +import bisect +import contextvars +import hashlib +import json +import logging +import os +import re +import time +from collections.abc import Mapping +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional +from urllib.parse import parse_qsl, quote, urlsplit, urlunsplit + +logger = logging.getLogger(__name__) + +#: Where a browser's server writes its log: the computer's log folder, else its own folder when that can't be written. +LOG_DIR = Path("/var/log/agentenv") +#: The header a driver forwards an MCP tool call to the browser with: the tool, its call's id and its arguments. +TOOL_CALL_HEADER = "AgentEnv-Tool-Call" +#: The longest a string in an entry may be before it's cut, with its length and sha256 kept. +CUT_CHARS = 2048 +#: The longest an error's message in an entry is kept. +ERROR_CHARS = 300 +#: The longest the tool-call header may be; past it, its arguments are dropped. +HEADER_CHARS = 12_000 +#: How many lines a read of the log returns at most. +READ_LIMIT = 5000 +#: How long an action not yet finished holds back a read; past it, a read writes it as ``unfinished`` in its place, and +#: its outcome, once it ends, is a line of its own. +STALLED_S = 60.0 +#: How long a tab event waits on the commands in flight that may have caused it. +CAUSE_WAIT_S = 5.0 +#: How long after a reload or a move in history its tab's next navigation is taken as its. +NEXT_NAVIGATION_S = 10.0 +REDACTED = "" + +#: The domain whose every command acts. +ACTING_DOMAINS = frozenset({"Input"}) +#: Per domain, the method names that act on the browser or a page, by their start; any other method reads. +ACTING_METHODS: dict[str, tuple[str, ...]] = { + "Page": ("navigate", "reload", "stopLoading", "close", "bringToFront", "handleJavaScriptDialog", + "setDocumentContent", "addScriptToEvaluateOnNewDocument", "removeScriptToEvaluateOnNewDocument", + "setBypassCSP", "setGeolocationOverride", "setWebLifecycleState", "crash", "resetNavigationHistory"), + "Runtime": ("evaluate", "callFunctionOn", "runScript", "addBinding"), + "Target": ("createTarget", "closeTarget", "activateTarget", "createBrowserContext", "disposeBrowserContext"), + "Browser": ("setWindowBounds", "setDownloadBehavior", "grantPermissions", "resetPermissions", "setPermission", + "cancelDownload", "close", "crash"), + "DOM": ("set", "remove", "move", "copy", "undo", "redo", "focus", "scrollIntoView"), + "CSS": ("set", "add", "create"), + "Emulation": ("set", "clear", "reset"), + "Fetch": ("fulfill", "continue", "fail"), + "Network": ("set", "clear", "delete", "emulate", "replay"), + "Storage": ("set", "clear", "delete", "reset", "override"), + "DOMStorage": ("set", "remove", "clear"), + "IndexedDB": ("clear", "delete"), + "CacheStorage": ("delete",), + "ServiceWorker": ("unregister", "skipWaiting", "stopWorker", "stopAllWorkers", "dispatch", "deliverPushMessage"), +} +#: The ``browser/v1`` ops that act; any other reads. +ACTING_OPS = frozenset({"page/input", "navigate", "history", "tabs/open", "tabs/activate", "tabs/close", + "dialog/answer", "files/choose", "cookies/set", "cookies/delete", "local_storage/set", + "local_storage/clear", "sites/set"}) +#: The ``browser/v1`` ops an agent's calls are never logged for: the log itself, and agentenv's own bookkeeping. +UNLOGGED_OPS = frozenset({"log", "record"}) +#: The CDP methods and ops whose text is typed into the page. +TYPING_METHODS = frozenset({"Input.insertText", "Input.dispatchKeyEvent", "Input.imeSetComposition"}) +#: What a key event says of the key it types, all of it dropped when a password field has focus. +KEY_FIELDS = ("text", "unmodifiedText", "key", "code", "keyIdentifier", "windowsVirtualKeyCode", + "nativeVirtualKeyCode", "commands") +#: The words of a URL query key that name a credential, so its value is taken out. +SECRET_WORDS = frozenset({"token", "key", "apikey", "secret", "password", "passwd", "pwd", "auth", "authorization", + "code", "session", "sessionid", "sid", "sig", "signature", "credential", "credentials", "jwt", + "ticket", "assertion", "otp", "saml", "samlresponse", "nonce"}) +#: The query keys whose every value is taken out, by their start: signed-URL parameters. +SECRET_PREFIXES = ("x-amz-", "x-goog-", "x-ms-") +#: The fields, at any depth, that hold a URL. +URL_FIELDS = frozenset({"url", "urls", "href", "referrer", "documentURL"}) +#: The fields a page's content or a response's body is in, which an entry withholds, keeping their length and sha256. +BODY_FIELDS = frozenset({"body", "postData", "html", "rawResponse"}) +#: A tool call's arguments that say what is typed, withheld while a password field has focus. +TYPED_ARGS = frozenset({"text", "keys"}) +_WORDS = re.compile(r"[A-Z]?[a-z0-9]+|[A-Z]+(?![a-z])") +#: A URL inside free text, such as an error message. +_URL_IN_TEXT = re.compile(r"[A-Za-z][A-Za-z0-9+.-]*://[^\s\"'<>]+") + +#: The action under way in this task: the entry an op of the browser's server or a relayed command is, which the +#: browser's control credits the tab events it causes to, and the tab it acted on. +ACTOR: contextvars.ContextVar[Optional[dict]] = contextvars.ContextVar("agentenv_browser_actor", default=None) + + +def acts(method: str) -> bool: + """Whether the CDP ``method`` acts on the browser or a page, rather than reading it.""" + domain, _, name = method.partition(".") + return domain in ACTING_DOMAINS or name.startswith(ACTING_METHODS.get(domain, ("\0",))) + + +def op_acts(op: str) -> bool: + """Whether the ``browser/v1`` ``op`` acts on the browser or a page, rather than reading it.""" + return op in ACTING_OPS + + +def cut(value: Any) -> Any: + """``value`` with each string longer than ``CUT_CHARS`` cut to that, its length and sha256 beside it.""" + if isinstance(value, str): + if len(value) <= CUT_CHARS: + return value + return {"cut": value[:CUT_CHARS], "length": len(value), "sha256": hashlib.sha256(value.encode()).hexdigest()} + if isinstance(value, Mapping): + return {key: cut(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [cut(item) for item in value] + return value + + +def withheld(value: Any) -> dict: + """What an entry keeps of a value it can't hold: its length and sha256.""" + text = value if isinstance(value, str) else json.dumps(value, sort_keys=True) + return {"redacted": True, "length": len(text), "sha256": hashlib.sha256(text.encode()).hexdigest()} + + +def redact_url(url: Any) -> Any: + """``url`` with its password and the values of its query keys, and of a query-like fragment, that name a credential + taken out; its scheme, host, path and other query values stay.""" + if not isinstance(url, str) or "://" not in url: + return url + try: + parts = urlsplit(url) + netloc = parts.netloc + if "@" in netloc: + userinfo, _, host = netloc.rpartition("@") + netloc = f"{userinfo.partition(':')[0]}:{REDACTED}@{host}" if ":" in userinfo else netloc + return urlunsplit((parts.scheme, netloc, parts.path, _redact_query(parts.query), + _redact_query(parts.fragment) if "=" in parts.fragment else parts.fragment)) + except ValueError: + return url + + +def _redact_query(query: str) -> str: + if not query: + return query + pairs = parse_qsl(query, keep_blank_values=True) + if not pairs: + return query + return "&".join(f"{quote(key, safe='')}={REDACTED if _secret_key(key) else quote(value, safe='')}" + for key, value in pairs) + + +def redact_text(text: str) -> str: + """``text`` with each URL in it as ``redact_url`` keeps it, the punctuation after it left as it was.""" + def redacted(found: re.Match) -> str: + url = found.group(0) + bare = url.rstrip(".,:;!?)]") + return redact_url(bare) + url[len(bare):] + return _URL_IN_TEXT.sub(redacted, text) + + +def _secret_key(key: str) -> bool: + lowered = key.lower() + return lowered.startswith(SECRET_PREFIXES) or bool(SECRET_WORDS & {word.lower() for word in _WORDS.findall(key)}) + + +def _urls_and_bodies(value: Any) -> Any: + """``value`` with every URL field, at any depth, as ``redact_url`` keeps it, and every body or page content + withheld.""" + if isinstance(value, Mapping): + return {key: withheld(item) if key in BODY_FIELDS and item not in (None, "") else + ([_url(url) for url in item] if isinstance(item, list) else _url(item)) + if key in URL_FIELDS else _urls_and_bodies(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_urls_and_bodies(item) for item in value] + return value + + +def _url(value: Any) -> Any: + """A URL field's value as ``redact_url`` keeps it, a URL already cut by the part it kept.""" + if isinstance(value, Mapping) and isinstance(value.get("cut"), str): + return {**value, "cut": redact_url(value["cut"])[:CUT_CHARS]} + return redact_url(value) + + +def _cookie_values(cookies: Any) -> Any: + return [{**cookie, "value": REDACTED} if isinstance(cookie, Mapping) else cookie for cookie in cookies] \ + if isinstance(cookies, list) else cookies + + +def redacted_params(method: str, params: Mapping[str, Any], *, password: bool = False) -> dict: + """The parameters of an acting CDP ``method`` as its entry keeps them: what could carry a secret taken out, and + with a password field focused, what it types; then cut.""" + kept = dict(params) + if method == "Network.setCookie" and "value" in kept: + kept["value"] = REDACTED + if method in ("Network.setCookies", "Storage.setCookies"): + kept["cookies"] = _cookie_values(kept.get("cookies")) + if method == "Network.setExtraHTTPHeaders" and isinstance(kept.get("headers"), Mapping): + kept["headers"] = {name: REDACTED for name in kept["headers"]} + for field in ("headers", "responseHeaders"): + if method.startswith("Fetch.") and isinstance(kept.get(field), list): + kept[field] = [{**header, "value": REDACTED} if isinstance(header, Mapping) else header + for header in kept[field]] + if method.startswith("Fetch.") and "binaryResponseHeaders" in kept: + kept["binaryResponseHeaders"] = REDACTED + if method == "Fetch.continueWithAuth" and isinstance(kept.get("authChallengeResponse"), Mapping): + kept["authChallengeResponse"] = {key: (REDACTED if key in ("username", "password") else value) + for key, value in kept["authChallengeResponse"].items()} + kept = _urls_and_bodies(kept) + if password and method in TYPING_METHODS: + kept = {key: value for key, value in kept.items() if key not in KEY_FIELDS} + if "text" in params: + kept["text"] = {"length": len(params["text"]) if isinstance(params["text"], str) else 0} + return cut(kept) + + +def redacted_op(op: str, params: dict, *, password: bool = False) -> dict: + """The parameters of an acting ``browser/v1`` ``op`` as its entry keeps them, as ``redacted_params`` keeps a CDP + method's: page input event by event.""" + kept = dict(params) + if op == "page/input" and isinstance(kept.get("events"), list): + kept["events"] = [{**event, "params": redacted_params(str(event.get("method")), event.get("params") or {}, + password=password)} + if isinstance(event, Mapping) else event for event in kept["events"]] + if op == "cookies/set": + kept["cookies"] = _cookie_values(kept.get("cookies")) + return cut(_urls_and_bodies(kept)) + + +def redacted_args(tool: str, args: Mapping[str, Any], *, password: bool = False) -> dict: + """An MCP tool call's arguments as its entry keeps them: URLs as ``redact_url`` keeps them, and what it types, its + text or its keys in whatever shape the header carried them, as its length while a password field has focus.""" + kept = _urls_and_bodies(args) + if password: + kept = {key: {"length": _length(value)} if key in TYPED_ARGS else value for key, value in kept.items()} + return cut(kept) + + +def _length(value: Any) -> int: + """How long typed text or keys were, a cut or withheld value by the length it kept.""" + if isinstance(value, Mapping) and isinstance(value.get("length"), int): + return value["length"] + return len(value) if isinstance(value, (str, list, tuple)) else 0 + + +def tool_call_header(tool: str, call: str, args: Mapping[str, Any]) -> str: + """The header a driver forwards a tool call with: the tool, its call's id and its arguments, cut; without them when + even cut they'd make it too long.""" + body = {"tool": tool, "call": call, "args": cut(_urls_and_bodies(args))} + encoded = base64.urlsafe_b64encode(json.dumps(body).encode()).decode() + if len(encoded) > HEADER_CHARS: + body["args"] = {key: withheld(cut(value)) for key, value in args.items()} + encoded = base64.urlsafe_b64encode(json.dumps(body).encode()).decode() + if len(encoded) > HEADER_CHARS: + encoded = base64.urlsafe_b64encode(json.dumps({"tool": tool, "call": call, "args": {}}).encode()).decode() + return encoded + + +def read_tool_call(value: Optional[str]) -> Optional[dict]: + """The tool call a driver forwarded, from its header; None without one or for one it can't read.""" + if not value: + return None + try: + body = json.loads(base64.urlsafe_b64decode(value.encode())) + except (ValueError, TypeError): + return None + if not isinstance(body, dict) or not isinstance(body.get("tool"), str): + return None + return {"tool": body["tool"], "call": str(body.get("call") or ""), + "args": body["args"] if isinstance(body.get("args"), dict) else {}} + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +class BrowserLog: + """The log in ``path``: each line a JSON object with its ``seq`` and its time, ``at``, and one of ``action``, + ``tab``, ``outcome`` or ``reset``. A line's sequence number is taken when what it records begins, and the line is + written once it ends, so lines may be written out of order; a read returns them in order, up to the last whose + every earlier line is written. An action still going after ``STALLED_S`` is written by the next read as + ``unfinished``, so it holds back nothing after it, and its ``outcome`` line names it once it ends. The file is + opened at the first line, after any a server before this one wrote, and where each line is kept, so a read reads + only the lines it returns.""" + + def __init__(self, path: Path) -> None: + self.path = path + self._seq, self._partial, self._where, self._size = _scan(path) + self._seqs = sorted(self._where) + self._pending: dict[int, float] = {} + #: The actions begun and not yet finished, and those already written as unfinished. + self._begun: dict[int, dict] = {} + self._stood_in: set[int] = set() + self._file: Any = None + self._unwritable = False + #: Per tab, the sequence number of the latest acting entry on it. + self.last_acting: dict[str, int] = {} + + @classmethod + def for_app(cls, name: str, fallback: Path) -> BrowserLog: + """The log of the browser app ``name``: in the computer's log folder, else in ``fallback``, its own folder.""" + folder = LOG_DIR if os.access(LOG_DIR, os.W_OK) else fallback + return cls(folder / f"browser-{name}.jsonl") + + def reserve(self) -> int: + """The next sequence number, held until its line is written.""" + self._seq += 1 + self._pending[self._seq] = time.monotonic() + return self._seq + + def write(self, seq: int, line: dict, at: Optional[str] = None) -> None: + """The line held as ``seq`` written, with its time.""" + self._pending.pop(seq, None) + if self._file is None and not self._open(): + return + data = (json.dumps({"seq": seq, "at": at or _now(), **line}, default=str) + "\n").encode() + try: + self._file.write(data) + self._file.flush() + except (OSError, ValueError) as e: + logger.warning(f"the browser's log couldn't take line {seq}: {type(e).__name__}: {e}") + return + self._where[seq] = (self._size, len(data)) + self._size += len(data) + bisect.insort(self._seqs, seq) + + def _open(self) -> bool: + """The file open for appending, on a new line; False, once warned, when it can't be.""" + if self._unwritable: + return False + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + self._file = self.path.open("ab") + if self._partial: + self._file.write(b"\n") + self._size += 1 + return True + except OSError as e: + logger.warning(f"the browser's log can't be written at {self.path}: {type(e).__name__}: {e}") + self._unwritable = True + return False + + def begin(self, action: dict) -> dict: + """An action that begins now, held until ``finish``: its sequence number, time and fields so far. An acting one + on a tab becomes that tab's latest.""" + seq = self.reserve() + if action.get("acts") and action.get("tab"): + self.last_acting[action["tab"]] = seq + self._begun[seq] = begun = {"seq": seq, "at": _now(), "action": action} + return begun + + def finish(self, begun: dict, **fields: Any) -> None: + """The action ``begun`` written, with what it ended with; one a read already wrote as unfinished gets an + ``outcome`` line of its own, naming it.""" + seq = begun["seq"] + self._begun.pop(seq, None) + action = {**begun["action"], **fields} + if action.get("acts") and action.get("tab") and self.last_acting.get(action["tab"], 0) < seq: + self.last_acting[action["tab"]] = seq + if seq in self._stood_in: + self._stood_in.discard(seq) + self.note("outcome", {"entry": seq, **fields}) + else: + self.write(seq, {"action": action}, at=begun["at"]) + + def note(self, kind: str, body: dict) -> int: + """A line of ``kind`` written now: a ``tab`` event, or a ``reset``.""" + seq = self.reserve() + self.write(seq, {kind: body}) + return seq + + def read(self, after: int = 0, limit: int = READ_LIMIT) -> dict: + """The lines after sequence number ``after``, in order, up to ``limit`` of them and to the last whose every + earlier line is written or stalled: ``lines``, ``last``, the sequence number of that last one, and ``more`` + when the limit left some.""" + complete, picked, more = self._picked(after, limit) + return self._loaded(after, complete, picked, more) + + async def read_async(self, after: int = 0, limit: int = READ_LIMIT) -> dict: + """``read``, its file read in a thread.""" + complete, picked, more = self._picked(after, limit) + return await asyncio.to_thread(self._loaded, after, complete, picked, more) + + def _picked(self, after: int, limit: int) -> tuple[int, list[tuple[int, int, int]], bool]: + """Where the lines a read returns are: the last sequence number it may return, and each line's number, offset + and length, in order.""" + now = time.monotonic() + for seq, began in list(self._pending.items()): + if now - began >= STALLED_S and (begun := self._begun.get(seq)) is not None: + self._stood_in.add(seq) + self.write(seq, {"action": {**begun["action"], "unfinished": True}}, at=begun["at"]) + complete = min(self._pending) - 1 if self._pending else self._seq + start = bisect.bisect_right(self._seqs, after) + end = bisect.bisect_right(self._seqs, complete) + picked = [(seq, *self._where[seq]) for seq in self._seqs[start:min(end, start + limit)]] + return complete, picked, end - start > limit + + def _loaded(self, after: int, complete: int, picked: list[tuple[int, int, int]], more: bool) -> dict: + lines = [] + if picked: + try: + with self.path.open("rb") as file: + for _, offset, length in picked: + file.seek(offset) + try: + lines.append(json.loads(file.read(length))) + except ValueError: + continue + except OSError: + pass + last = (picked[-1][0] if picked else after) if more else max(after, complete) + return {"lines": lines, "last": last, "more": more} + + async def complete_text(self) -> str: + """Every line ``read`` returns, in order, as the file's text.""" + read = await self.read_async(0, limit=1 << 30) + return "".join(json.dumps(line, default=str) + "\n" for line in read["lines"]) + + +def _scan(path: Path) -> tuple[int, bool, dict[int, tuple[int, int]], int]: + """The log at ``path`` as a server before this one left it: its highest sequence number, whether its last line + was left partial, where each line is, and its size.""" + try: + data = path.read_bytes() + except OSError: + return 0, False, {}, 0 + where: dict[int, tuple[int, int]] = {} + offset = 0 + for raw in data.splitlines(keepends=True): + try: + line = json.loads(raw) + except ValueError: + line = None + if isinstance(line, dict) and isinstance(line.get("seq"), int) and raw.endswith(b"\n"): + where[line["seq"]] = (offset, len(raw)) + offset += len(raw) + return max(where, default=0), bool(data) and not data.endswith(b"\n"), where, len(data) + + +def caused(cause: Optional[dict], log: BrowserLog, tab: Optional[str]) -> dict: + """What a tab event says caused it: the command it was, by its entry, ``via`` and ``who``; agentenv itself; or, + with no ``cause``, the page, after the latest acting entry on ``tab`` if there is one.""" + if cause is None: + after = log.last_acting.get(tab) if tab else None + return {"by": "page", **({"after": after} if after else {})} + if cause.get("via") == "agentenv": + return {"by": "agentenv"} + return {"by": "command", "entry": cause["seq"], "via": cause["via"], "who": cause.get("who")} + + +def log_time() -> str: + """Now, as the log writes times.""" + return _now() + + +class Causes: + """The commands in flight that may open, close or navigate a tab, so the tab event each causes can name it: a new + tab's command answers after the browser says the tab opened, and a navigation's may too.""" + + def __init__(self) -> None: + self._creating: set[asyncio.Future] = set() + self._created: dict[str, dict] = {} + self._closing: dict[str, dict] = {} + self._navigating: dict[str, set[asyncio.Future]] = {} + self._loaders: dict[str, dict] = {} + self._same_document: dict[str, dict] = {} + self._next: dict[str, tuple[dict, float]] = {} + + def creating(self, answer: asyncio.Future, cause: dict) -> None: + """A ``Target.createTarget`` in flight, ``answer`` its answer's target id, sent by ``cause``: once it lands, the + new tab's first navigation, to the URL it was opened at, is taken as ``cause``'s too.""" + self._creating.add(answer) + answer.add_done_callback(lambda done: self._landed(done, self._created, cause, self._creating)) + answer.add_done_callback(lambda done: self._first_navigation(done, cause)) + + def _first_navigation(self, done: asyncio.Future, cause: dict) -> None: + if not done.cancelled() and done.exception() is None and isinstance(done.result(), str): + self._next.setdefault(done.result(), (cause, asyncio.get_running_loop().time() + NEXT_NAVIGATION_S)) + + def closing(self, target: str, cause: dict) -> None: + self._closing[target] = cause + + def navigating(self, target: str, answer: asyncio.Future, cause: dict) -> None: + """A ``Page.navigate`` of ``target`` in flight, ``answer`` its answer's loader id, sent by ``cause``; one + answered with none moved within its document.""" + waiting = self._navigating.setdefault(target, set()) + waiting.add(answer) + answer.add_done_callback(lambda done: self._landed(done, self._loaders, cause, waiting)) + answer.add_done_callback(lambda done: done.cancelled() or done.exception() is not None or done.result() + or self._same_document.__setitem__(target, cause)) + + def navigates_next(self, target: str, cause: dict) -> None: + """A reload or a move in history of ``target``, whose answer names no loader: its next navigation is taken as + its, within ``NEXT_NAVIGATION_S``.""" + self._next[target] = (cause, asyncio.get_running_loop().time() + NEXT_NAVIGATION_S) + + def navigated_nowhere(self, target: str, cause: dict) -> None: + """A reload or move in history of ``target`` that failed: its next navigation isn't its.""" + if (pending := self._next.get(target)) is not None and pending[0] == cause: + self._next.pop(target, None) + + @staticmethod + def _landed(done: asyncio.Future, into: dict, cause: dict, waiting: set) -> None: + waiting.discard(done) + if not done.cancelled() and done.exception() is None and isinstance(done.result(), str): + into[done.result()] = cause + + async def opened(self, target: str) -> Optional[dict]: + """What opened ``target``, once the creates in flight have answered.""" + if target not in self._created and self._creating: + await asyncio.wait(set(self._creating), timeout=CAUSE_WAIT_S) + return self._created.pop(target, None) + + def closed(self, target: str) -> Optional[dict]: + return self._closing.pop(target, None) + + async def navigated(self, target: str, loader: Optional[str]) -> Optional[dict]: + """What navigated ``target`` to ``loader``, once its navigations in flight have answered.""" + if loader and loader not in self._loaders and self._navigating.get(target): + await asyncio.wait(set(self._navigating[target]), timeout=CAUSE_WAIT_S) + pending = self._next.pop(target, None) + if loader and (cause := self._loaders.pop(loader, None)) is not None: + return cause + if pending is not None and pending[1] >= asyncio.get_running_loop().time(): + return pending[0] + return None + + async def navigated_within(self, target: str) -> Optional[dict]: + """What moved ``target`` within its document, once its navigations in flight have answered: a navigation + answered with no loader, else a reload or move in history, which it uses up.""" + if target not in self._same_document and self._navigating.get(target): + await asyncio.wait(set(self._navigating[target]), timeout=CAUSE_WAIT_S) + pending = self._next.pop(target, None) + if (cause := self._same_document.pop(target, None)) is not None: + return cause + if pending is not None and pending[1] >= asyncio.get_running_loop().time(): + return pending[0] + return None diff --git a/packages/agentenv-protocol/tests/test_browser_cdp.py b/packages/agentenv-protocol/tests/test_browser_cdp.py index 41493acf4..0ea2c4e9e 100644 --- a/packages/agentenv-protocol/tests/test_browser_cdp.py +++ b/packages/agentenv-protocol/tests/test_browser_cdp.py @@ -21,12 +21,14 @@ from websockets.asyncio.client import connect # noqa: E402 from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK # noqa: E402 -from agentenv_protocol import browser_cdp # noqa: E402 -from agentenv_protocol.browser_cdp import NO_SESSION, Relay, Tokens, guard # noqa: E402 +from agentenv_protocol.website_browser import cdp as browser_cdp # noqa: E402 +from agentenv_protocol.website_browser import log as browser_log # noqa: E402 +from agentenv_protocol.website_browser.cdp import NO_SESSION, Relay, Tokens, guard # noqa: E402 +from agentenv_protocol.website_browser.log import REDACTED, TOOL_CALL_HEADER, BrowserLog, Causes, tool_call_header # noqa: E402 from agentenv_protocol.computer import Launch, Platform, Ready, app, computer_app, linux # noqa: E402 from agentenv_protocol.computer.app import APP_SETTINGS_ENV # noqa: E402 from agentenv_protocol.computer.app_runtime import AppReadiness # noqa: E402 -from agentenv_protocol.computer.errors import DeviceUnavailable # noqa: E402 +from agentenv_protocol.computer.errors import DeviceUnavailable, InvalidArgument # noqa: E402 from agentenv_protocol.website_browser import SITE_PROXY_ENV, WebsiteBrowserEnvironment # noqa: E402 FORWARDED = {"x-forwarded-host": "computer.test:443", "x-forwarded-proto": "https", @@ -106,11 +108,11 @@ async def close(self, code: int = 1000, reason: str = "") -> None: @contextlib.asynccontextmanager -async def _relay(dialogs_answered: bool = False): +async def _relay(dialogs_answered: bool = False, **logging): """The relay over a fake pipe, served on loopback with each connection's user its path's: the pipe, the relay and the server's ws:// base.""" pipe = Pipe() - relay = Relay(pipe, dialogs_answered=dialogs_answered) + relay = Relay(pipe, dialogs_answered=dialogs_answered, **logging) async def endpoint(websocket): await relay.serve(websocket, websocket.path_params["user"]) @@ -279,6 +281,82 @@ async def test_a_client_that_falls_too_far_behind_is_closed_and_what_waited_for_ assert "closed_at" in relay.connections[0] and ("Target.detachFromTarget", None) in pipe.sent +class Control: + """agentenv's control as the relay asks it: the URL of tab T1, a password field focused wherever it's asked, and + the causes of tab events; each check made with how many commands ``pipe`` had been sent by then.""" + + def __init__(self, pipe=None) -> None: + self.causes, self.asked, self.pipe = Causes(), [], pipe + + def url_of(self, target): + return "https://a.test/page" if target == "T1" else None + + async def password_focused(self, target) -> bool: + self.asked.append((target, len(self.pipe.sent) if self.pipe is not None else None)) + return True + + +@pytest.mark.asyncio +async def test_every_relayed_command_is_an_entry_on_its_sessions_tab_an_acting_one_with_its_parameters_kept(tmp_path): + log, control = BrowserLog(tmp_path / "browser-chromium.jsonl"), Control() + async with _relay(log=log, control=control) as (pipe, relay, base): + async with connect(f"{base}/ada") as ada: + attached = await _ask(ada, {"id": 1, "method": "Target.attachToTarget", + "params": {"targetId": "T1", "flatten": True}}) + child = attached["result"]["sessionId"] + for n, (method, params) in enumerate([ + ("DOM.getDocument", {"depth": 1}), + ("Network.setCookie", {"name": "sid", "value": "s3cret", "url": "https://a.test/?token=t"}), + ("Input.insertText", {"text": "hunter2"}), ("Boom", {})], start=2): + await _ask(ada, {"id": n, "method": method, "params": params, "sessionId": child}) + await _ask(ada, {"id": 6, "method": "Target.exposeDevToolsProtocol", "params": {"targetId": "T1"}}) + await _ask(ada, {"id": 7, "method": "Page.navigate", "params": {"url": "https://a.test/"}, + "sessionId": "theirs"}) + await asyncio.sleep(0.05) + + common = {"via": "cdp", "who": "ada", "connection": 1} + page = {**common, "tab": "T1", "url": "https://a.test/page"} + assert [line["action"] for line in log.read()["lines"]] == [ + {**page, "method": "Target.attachToTarget", "acts": False, "ok": True}, + {**page, "method": "DOM.getDocument", "acts": False, "ok": True}, + {**page, "method": "Network.setCookie", "acts": True, "ok": True, + "params": {"name": "sid", "value": REDACTED, "url": f"https://a.test/?token={REDACTED}"}}, + {**page, "method": "Input.insertText", "acts": True, "ok": True, "params": {"text": {"length": 7}}}, + {**page, "method": "Boom", "acts": False, "ok": False, "error": "boom"}, + {**page, "method": "Target.exposeDevToolsProtocol", "acts": False, "ok": False, "guard": "refused", + "error": "agentenv refuses Target.exposeDevToolsProtocol: it would give a page's script the browser's own " + "CDP channel"}, + {**common, "method": "Page.navigate", "acts": True, "tab": None, "url": None, "ok": False, + "error": NO_SESSION["message"], "params": {"url": "https://a.test/"}}, + ] + assert [target for target, _ in control.asked] == ["T1"] + + +@pytest.mark.asyncio +async def test_a_command_is_checked_for_a_password_before_it_is_sent_and_logged_though_its_client_leaves(tmp_path): + log = BrowserLog(tmp_path / "browser-chromium.jsonl") + async with _relay(log=log) as (pipe, relay, base): + control = Control(pipe) + relay.control = control + async with connect(f"{base}/ada") as ada: + child = (await _ask(ada, {"id": 1, "method": "Target.attachToTarget", + "params": {"targetId": "T1", "flatten": True}}))["result"]["sessionId"] + await _ask(ada, {"id": 2, "method": "Input.insertText", "params": {"text": "hunter2"}, "sessionId": child}) + sent_before = next(n for n, (method, _) in enumerate(pipe.sent) if method == "Input.insertText") + await ada.send(json.dumps({"id": 3, "method": "Hang", "params": {}, "sessionId": child})) + await asyncio.sleep(0.1) + await asyncio.sleep(0.2) + after_leaving = log.read() + assert pipe.given_up == ["Hang"] + + assert control.asked == [("T1", sent_before)] + entries = [line["action"] for line in after_leaving["lines"]] + assert [entry["method"] for entry in entries] == ["Target.attachToTarget", "Input.insertText", "Hang"] + assert entries[1]["params"] == {"text": {"length": 7}} + assert entries[2]["ok"] is False and entries[2]["error"] == browser_cdp.ENDED_UNANSWERED + assert after_leaving["last"] == 3 + + def _browser(monkeypatch, tmp_path) -> Browser: """The browser alone, in ``tmp_path``.""" monkeypatch.chdir(tmp_path) @@ -287,9 +365,157 @@ def _browser(monkeypatch, tmp_path) -> Browser: monkeypatch.setenv("ENVIRONMENT_NAME", "chromium") monkeypatch.setenv(SITE_PROXY_ENV, "http://127.0.0.1:3128") monkeypatch.delenv(APP_SETTINGS_ENV, raising=False) + monkeypatch.setattr(browser_log, "LOG_DIR", tmp_path / "logs") return Browser() +@pytest.mark.asyncio +async def test_an_op_for_an_agents_tool_call_and_agentenvs_own_acting_ops_are_entries_and_its_own_reads_are_not( + monkeypatch, tmp_path): + (tmp_path / "logs").mkdir() + browser = _browser(monkeypatch, tmp_path) + + async def navigate(url): + browser.control._acted_on("T1") + + async def tabs(): + return [] + + monkeypatch.setattr(browser.control, "navigate", navigate) + monkeypatch.setattr(browser.control, "tab_list", tabs) + monkeypatch.setattr(browser.control, "url_of", lambda target: "https://a.test/" if target else None) + asgi = browser.served_asgi(browser.create_app().streamable_http_app()) + + def tool(name: str, call: str, args: dict) -> dict: + return {TOOL_CALL_HEADER: tool_call_header(name, call, args)} + + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=asgi), base_url="http://app") as http: + url = "https://a.test/?token=t" + await http.post("/agentenv/ext/browser/navigate", json={"url": url}, + headers={**tool("computer_navigate", "c1", {"url": url}), "AgentEnv-Role": "ada"}) + await http.get("/agentenv/ext/browser/tabs/list") + await http.get("/agentenv/ext/browser/tabs/list", headers=tool("computer_tabs", "c2", {"action": "list"})) + await http.post("/agentenv/ext/browser/sites/set", json={"sites": []}) + closed = await http.post("/agentenv/ext/browser/tabs/close", json={"tab": "nope"}, + headers=tool("computer_tabs", "c3", {"action": "close", "tab": "nope"})) + read = (await http.get("/agentenv/ext/browser/log", params={"after": 0})).json() + + redacted = f"https://a.test/?token={REDACTED}" + assert closed.status_code >= 400 + assert (tmp_path / "logs" / "browser-chromium.jsonl").is_file() + assert [line["action"] for line in read["lines"]] == [ + {"via": "mcp", "op": "navigate", "acts": True, "who": "ada", "tool": "computer_navigate", "call": "c1", + "tab": "T1", "url": "https://a.test/", "ok": True, "params": {"url": redacted}, "args": {"url": redacted}}, + {"via": "mcp", "op": "tabs/list", "acts": False, "who": "default", "tool": "computer_tabs", "call": "c2", + "tab": None, "url": None, "ok": True}, + {"via": "agentenv", "op": "sites/set", "acts": True, "tab": None, "url": None, "ok": True, + "params": {"sites": []}}, + {"via": "mcp", "op": "tabs/close", "acts": True, "who": "default", "tool": "computer_tabs", "call": "c3", + "tab": None, "url": None, "ok": False, "error": "no tab nope is open; list the tabs for their ids", + "params": {"tab": "nope"}, "args": {"action": "close", "tab": "nope"}}, + ] + assert read["last"] == 4 and not read["more"] + + +@pytest.mark.asyncio +async def test_a_commands_wait_for_its_password_check_lets_none_after_it_be_sent_first(tmp_path): + log = BrowserLog(tmp_path / "browser-chromium.jsonl") + async with _relay(log=log) as (pipe, relay, base): + control = Control(pipe) + + async def slowly(target) -> bool: + await asyncio.sleep(0.2) + return False + + control.password_focused = slowly + relay.control = control + async with connect(f"{base}/ada") as ada: + child = (await _ask(ada, {"id": 1, "method": "Target.attachToTarget", + "params": {"targetId": "T1", "flatten": True}}))["result"]["sessionId"] + await ada.send(json.dumps({"id": 2, "method": "Input.insertText", "params": {"text": "desk"}, + "sessionId": child})) + await ada.send(json.dumps({"id": 3, "method": "Input.dispatchMouseEvent", + "params": {"type": "mousePressed", "x": 1, "y": 1}, "sessionId": child})) + answered = [json.loads(await ada.recv())["id"] for _ in range(2)] + + sent = [method for method, _ in pipe.sent if method.startswith("Input.")] + assert sent == ["Input.insertText", "Input.dispatchMouseEvent"] and sorted(answered) == [2, 3] + + +@pytest.mark.asyncio +async def test_typing_still_waiting_for_its_check_when_its_client_leaves_is_logged_withheld(tmp_path): + log = BrowserLog(tmp_path / "browser-chromium.jsonl") + async with _relay(log=log) as (pipe, relay, base): + control = Control(pipe) + + async def never(target) -> bool: + await asyncio.Event().wait() + + control.password_focused = never + relay.control = control + async with connect(f"{base}/ada") as ada: + child = (await _ask(ada, {"id": 1, "method": "Target.attachToTarget", + "params": {"targetId": "T1", "flatten": True}}))["result"]["sessionId"] + for n, text in ((2, "hunter2"), (3, "hunter3")): + await ada.send(json.dumps({"id": n, "method": "Input.insertText", "params": {"text": text}, + "sessionId": child})) + await asyncio.sleep(0.1) + await asyncio.sleep(0.2) + + typed = [line["action"] for line in log.read()["lines"] if line["action"]["method"] == "Input.insertText"] + assert [entry["params"] for entry in typed] == [{"text": {"length": 7}}, {"text": {"length": 7}}] + assert all(entry["ok"] is False for entry in typed) and "hunter" not in json.dumps(log.read()) + + +@pytest.mark.asyncio +async def test_typing_that_fails_before_its_check_and_a_failed_navigation_keep_no_secrets(monkeypatch, tmp_path): + browser = _browser(monkeypatch, tmp_path) + + async def no_tab(events): + raise DeviceUnavailable("the browser has no tab open") + + async def unreachable(url): + raise InvalidArgument(f"the browser couldn't open {url}: net::ERR_NAME_NOT_RESOLVED") + + monkeypatch.setattr(browser.control, "dispatch", no_tab) + monkeypatch.setattr(browser.control, "navigate", unreachable) + asgi = browser.served_asgi(browser.create_app().streamable_http_app()) + typed = [{"method": "Input.insertText", "params": {"text": "hunter2"}}] + url = "https://a.test/?token=secret" + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=asgi), base_url="http://app") as http: + await http.post("/agentenv/ext/browser/page/input", json={"events": typed}, + headers={TOOL_CALL_HEADER: tool_call_header("computer_type", "c1", {"text": "hunter2"})}) + await http.post("/agentenv/ext/browser/navigate", json={"url": url}, + headers={TOOL_CALL_HEADER: tool_call_header("computer_navigate", "c2", {"url": url})}) + read = (await http.get("/agentenv/ext/browser/log")).json() + + typing, navigating = (line["action"] for line in read["lines"]) + assert typing["ok"] is False and typing["args"] == {"text": {"length": 7}} + assert typing["params"] == {"events": [{"method": "Input.insertText", "params": {"text": {"length": 7}}}]} + assert navigating["ok"] is False and "secret" not in navigating["error"] and REDACTED in navigating["error"] + assert "hunter2" not in json.dumps(read) and "secret" not in json.dumps(read) + + +@pytest.mark.asyncio +async def test_an_op_body_too_long_to_keep_is_logged_by_its_length(monkeypatch, tmp_path): + browser = _browser(monkeypatch, tmp_path) + monkeypatch.setattr("agentenv_protocol.website_browser.environment.LOGGED_BODY_BYTES", 100) + + async def dispatch(events): + return None + + monkeypatch.setattr(browser.control, "dispatch", dispatch) + asgi = browser.served_asgi(browser.create_app().streamable_http_app()) + events = [{"method": "Input.insertText", "params": {"text": "x" * 200}}] + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=asgi), base_url="http://app") as http: + answered = await http.post("/agentenv/ext/browser/page/input", json={"events": events}, + headers={TOOL_CALL_HEADER: tool_call_header("computer_type", "c1", {"text": "x"})}) + read = (await http.get("/agentenv/ext/browser/log")).json() + + assert answered.status_code == 200 + assert read["lines"][0]["action"]["params"] == {"too_long": True, "length": len(answered.request.content)} + + @pytest.mark.asyncio async def test_cdp_waits_for_what_computer_app_ready_reports_up_to_its_declared_ready_timeout(monkeypatch, tmp_path): browser = _browser(monkeypatch, tmp_path) diff --git a/packages/agentenv-protocol/tests/test_browser_data.py b/packages/agentenv-protocol/tests/test_browser_data.py index aa1f40ae4..aacf4f444 100644 --- a/packages/agentenv-protocol/tests/test_browser_data.py +++ b/packages/agentenv-protocol/tests/test_browser_data.py @@ -16,6 +16,7 @@ import pytest from agentenv_protocol import website_browser +from agentenv_protocol.website_browser import log as browser_log from agentenv_protocol.computer import Launch, Ready, app, linux from agentenv_protocol.computer.app import AGENT_ACTED from agentenv_protocol.computer.errors import DeviceUnavailable, InvalidArgument @@ -90,6 +91,7 @@ def browser(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path / "folder") monkeypatch.setenv("ENVIRONMENT_NAME", "chromium") monkeypatch.setenv("AGENTENV_APP_SETTINGS", json.dumps({"chromium": {"start_urls": ["https://shop.test/"]}})) + monkeypatch.setattr(browser_log, "LOG_DIR", tmp_path / "logs") made = Browser() made.control = FakeControl() launches = [] @@ -149,10 +151,25 @@ async def test_data_get_is_its_state_its_profile_without_caches_or_locks_and_its assert state["cookies"] == FakeControl().cookie_list and "partitionKey" not in state["cookies"][0] assert state["clock"]["mode"] == "real" and state["tabs"][0]["id"] == "t1" and state["closed_tabs"][0]["id"] == "t0" assert files == {f"{PROFILE}/Default/Cookies": b"db", f"{PROFILE}/Default/Cookies-journal": b"j", - f"{PROFILE}/Default/Local Storage/leveldb/CURRENT": b"m", "Downloads/report.csv": b"a,b"} + f"{PROFILE}/Default/Local Storage/leveldb/CURRENT": b"m", "Downloads/report.csv": b"a,b", + "browser-log.jsonl": b""} assert not list((home / ".cache").iterdir()) +@pytest.mark.asyncio +async def test_data_get_exports_the_log_written_so_far_and_data_add_passes_over_it(browser): + browser.log.note("reset", {}) + pending = browser.log.reserve() + files = _files(await browser.browser_get_data()) + browser.log.write(pending, {"reset": {}}) + + logged = [json.loads(line) for line in files["browser-log.jsonl"].decode().splitlines()] + assert [line["seq"] for line in logged] == [1] and logged[0]["reset"] == {} + await browser.browser_add_data(_parts(_file("browser-state.json", _snapshot_state()), + _file("browser-log.jsonl", files["browser-log.jsonl"]))) + assert ("cookies", ["sid"]) in browser.control.did + + @pytest.mark.asyncio async def test_data_get_reads_the_state_then_copies_the_profile_with_the_browser_paused(browser, monkeypatch): stage = website_browser.data.stage_profile @@ -183,7 +200,8 @@ def removed_meanwhile(source, target): monkeypatch.setattr(website_browser.data.shutil, "copy2", removed_meanwhile) files = _files(await browser.browser_get_data()) - assert set(files) == {"browser-state.json", f"{PROFILE}/Default/Cookies", f"{PROFILE}/Default/Session Storage/LOG"} + assert set(files) == {"browser-state.json", "browser-log.jsonl", f"{PROFILE}/Default/Cookies", + f"{PROFILE}/Default/Session Storage/LOG"} def _snapshot_state(**changes) -> bytes: diff --git a/packages/agentenv-protocol/tests/test_browser_log.py b/packages/agentenv-protocol/tests/test_browser_log.py new file mode 100644 index 000000000..11cff2fb3 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_browser_log.py @@ -0,0 +1,329 @@ +"""A website browser's log: CDP methods and ops marked acting or reading; what could carry a secret taken out of an +entry, and a long value cut; a driver's tool call carried in a header; lines numbered in order, read in order up to +the last whose every earlier line is written, the sequence carried on by a server that restarts; and each tab event +credited to the command that caused it, to agentenv, or to the page after the latest action on its tab.""" +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +import types + +import pytest + +from agentenv_protocol.website_browser import log as browser_log +from agentenv_protocol.website_browser.log import (ACTOR, CUT_CHARS, REDACTED, BrowserLog, acts, cut, op_acts, read_tool_call, redact_text, + redact_url, redacted_args, redacted_op, redacted_params, tool_call_header) +from agentenv_protocol.computer.errors import InvalidArgument +from agentenv_protocol.website_browser.control import BrowserControl + + +@pytest.mark.parametrize("method, acting", [ + ("Input.dispatchMouseEvent", True), ("Input.insertText", True), ("Page.navigate", True), ("Page.reload", True), + ("Runtime.evaluate", True), ("Runtime.callFunctionOn", True), ("DOM.setAttributeValue", True), + ("Fetch.fulfillRequest", True), ("Network.setCookie", True), ("Storage.setCookies", True), + ("Emulation.setDeviceMetricsOverride", True), ("Target.createTarget", True), ("Target.closeTarget", True), + ("DOM.getDocument", False), ("DOMSnapshot.captureSnapshot", False), ("Accessibility.getFullAXTree", False), + ("Page.captureScreenshot", False), ("Page.enable", False), ("Page.setLifecycleEventsEnabled", False), + ("Network.getCookies", False), ("Target.getTargets", False), ("Target.attachToTarget", False), + ("Runtime.getProperties", False), ("Browser.getVersion", False), +]) +def test_a_cdp_method_acts_when_it_can_change_the_browser_or_a_page_and_reads_otherwise(method, acting): + assert acts(method) is acting + + +def test_an_op_acts_when_it_can_change_the_browser_or_a_page(): + assert all(op_acts(op) for op in ("page/input", "navigate", "history", "tabs/open", "tabs/close", "cookies/set")) + assert not any(op_acts(op) for op in ("page/capture", "page/state", "tabs/list", "cookies/get", "downloads/list")) + + +def test_a_url_keeps_its_search_terms_and_loses_its_credentials(): + assert redact_url("https://shop.test/search?q=standing+desk&keyword=oak&page=2") == \ + "https://shop.test/search?q=standing%20desk&keyword=oak&page=2" + assert redact_url("https://app.test/cb?code=abc&state=s1&access_token=t#id_token=x&scope=email") == \ + f"https://app.test/cb?code={REDACTED}&state=s1&access_token={REDACTED}#id_token={REDACTED}&scope=email" + assert redact_url("https://bucket.test/k?X-Amz-Signature=sig&X-Amz-Date=1&apiKey=k&sessionId=s") == \ + f"https://bucket.test/k?X-Amz-Signature={REDACTED}&X-Amz-Date={REDACTED}&apiKey={REDACTED}&sessionId={REDACTED}" + assert redact_url("https://ada:hunter2@site.test/x#top") == f"https://ada:{REDACTED}@site.test/x#top" + assert redact_url("about:blank") == "about:blank" and redact_url(None) is None + + +def test_cookie_header_and_body_values_are_taken_out_of_an_acting_entry(): + assert redacted_params("Network.setCookie", {"name": "sid", "value": "s3cret", "url": "https://a.test/?token=t"}) \ + == {"name": "sid", "value": REDACTED, "url": f"https://a.test/?token={REDACTED}"} + assert redacted_params("Storage.setCookies", {"cookies": [{"name": "a", "value": "1"}]}) == \ + {"cookies": [{"name": "a", "value": REDACTED}]} + assert redacted_params("Network.setCookies", {"cookies": [{"name": "a", "value": "1", "url": "https://a.test/?token=t"}]}) \ + == {"cookies": [{"name": "a", "value": REDACTED, "url": f"https://a.test/?token={REDACTED}"}]} + fulfilled = redacted_params("Fetch.fulfillRequest", {"requestId": "r", "responseCode": 200, "body": "PGh0bWw+"}) + assert fulfilled["body"] == {"redacted": True, "length": 8, "sha256": hashlib.sha256(b"PGh0bWw+").hexdigest()} + assert redacted_params("Page.setDocumentContent", {"frameId": "f", "html": "

faked

"})["html"]["redacted"] + assert redacted_params("Network.setExtraHTTPHeaders", {"headers": {"Authorization": "Bearer x"}}) == \ + {"headers": {"Authorization": REDACTED}} + continued = redacted_params("Fetch.continueRequest", {"requestId": "r", "postData": "cGFzcz0x", + "headers": [{"name": "Cookie", "value": "sid=1"}]}) + assert continued["headers"] == [{"name": "Cookie", "value": REDACTED}] + assert continued["postData"] == {"redacted": True, "length": 8, + "sha256": hashlib.sha256(b"cGFzcz0x").hexdigest()} + assert redacted_params("Fetch.continueWithAuth", {"authChallengeResponse": { + "response": "ProvideCredentials", "username": "ada", "password": "pw"}})["authChallengeResponse"] == { + "response": "ProvideCredentials", "username": REDACTED, "password": REDACTED} + + +def test_typing_while_a_password_field_has_focus_keeps_only_how_much_was_typed(): + key = {"type": "keyDown", "key": "h", "code": "KeyH", "text": "h", "unmodifiedText": "h", + "windowsVirtualKeyCode": 72, "modifiers": 0} + assert redacted_params("Input.dispatchKeyEvent", key, password=True) == \ + {"type": "keyDown", "modifiers": 0, "text": {"length": 1}} + assert redacted_params("Input.dispatchKeyEvent", key) == key + assert redacted_params("Input.insertText", {"text": "hunter2"}, password=True) == {"text": {"length": 7}} + events = [{"method": "Input.insertText", "params": {"text": "hunter2"}}] + assert redacted_op("page/input", {"events": events}, password=True) == \ + {"events": [{"method": "Input.insertText", "params": {"text": {"length": 7}}}]} + assert redacted_args("computer_type", {"text": "hunter2"}, password=True) == {"text": {"length": 7}} + assert redacted_args("computer_type", {"text": "desk"}) == {"text": "desk"} + assert redacted_args("computer_key", {"keys": ["h"]}, password=True) == {"keys": {"length": 1}} + long_text = read_tool_call(tool_call_header("computer_type", "c", {"text": "p" * (CUT_CHARS + 5)}))["args"] + assert redacted_args("computer_type", long_text, password=True) == {"text": {"length": CUT_CHARS + 5}} + assert redacted_op("cookies/set", {"cookies": [{"name": "a", "value": "1"}]}) == \ + {"cookies": [{"name": "a", "value": REDACTED}]} + + +def test_a_long_value_is_cut_with_its_length_and_sha256(): + script = "x" * (CUT_CHARS + 10) + assert redacted_params("Runtime.evaluate", {"expression": script})["expression"] == { + "cut": "x" * CUT_CHARS, "length": CUT_CHARS + 10, "sha256": hashlib.sha256(script.encode()).hexdigest()} + assert cut({"a": ("b", types.MappingProxyType({"c": 1}))}) == {"a": ["b", {"c": 1}]} + + +def test_a_tool_call_header_carries_the_tool_its_call_and_its_arguments_cut(): + args = types.MappingProxyType({"text": "y" * (CUT_CHARS + 1), "x": 3}) + read = read_tool_call(tool_call_header("computer_type", "c1", args)) + assert read == {"tool": "computer_type", "call": "c1", "args": {"text": cut("y" * (CUT_CHARS + 1)), "x": 3}} + assert read_tool_call(None) is None and read_tool_call("not base64 json") is None + assert read_tool_call(base64.urlsafe_b64encode(b'{"call": 1}').decode()) is None + + +def test_a_long_url_and_a_url_in_an_error_keep_no_credentials(): + long_url = "https://a.test/?token=secret&pad=" + "x" * CUT_CHARS + sent = read_tool_call(tool_call_header("computer_navigate", "c", {"url": long_url}))["args"]["url"] + assert sent["cut"].startswith(f"https://a.test/?token={REDACTED}&pad=") and "secret" not in json.dumps(sent) + kept = redacted_args("computer_navigate", {"url": cut(long_url)})["url"] + assert kept["cut"].startswith(f"https://a.test/?token={REDACTED}") and "secret" not in json.dumps(kept) + assert redact_text("the browser couldn't open https://a.test/?token=t: net::ERR_FAILED") == \ + f"the browser couldn't open https://a.test/?token={REDACTED}: net::ERR_FAILED" + + +def test_a_tool_call_header_too_long_even_cut_drops_its_arguments(monkeypatch): + monkeypatch.setattr(browser_log, "HEADER_CHARS", 200) + read = read_tool_call(tool_call_header("computer_type", "c1", {"text": "z" * 500})) + assert read["tool"] == "computer_type" and read["args"] == {} + + +@pytest.mark.asyncio +async def test_lines_are_numbered_as_they_begin_and_read_in_order_up_to_the_first_unwritten(tmp_path): + log = BrowserLog(tmp_path / "logs" / "browser-chromium.jsonl") + first = log.begin({"via": "cdp", "method": "Page.navigate", "acts": True, "tab": "T1"}) + second = log.note("tab", {"event": "navigated", "id": "T1"}) + assert log.read() == {"lines": [], "last": 0, "more": False} + log.finish(first, ok=True) + read = log.read() + assert [line["seq"] for line in read["lines"]] == [1, 2] and read["last"] == second + assert read["lines"][0]["action"] == {"via": "cdp", "method": "Page.navigate", "acts": True, "tab": "T1", + "ok": True} + assert log.read(after=1)["lines"][0]["tab"] == {"event": "navigated", "id": "T1"} + assert log.last_acting == {"T1": 1} + assert log.read(after=1, limit=0) == {"lines": [], "last": 1, "more": True} + assert (await log.complete_text()).count("\n") == 2 + assert await log.read_async(after=1) == log.read(after=1) + + +def test_an_action_still_going_after_the_limit_is_written_unfinished_and_its_outcome_after(tmp_path, monkeypatch): + log = BrowserLog(tmp_path / "browser-chromium.jsonl") + slow = log.begin({"via": "cdp", "method": "Runtime.evaluate", "acts": True, "tab": "T1"}) + log.note("reset", {}) + assert log.read()["lines"] == [] + monkeypatch.setattr(browser_log, "STALLED_S", 0) + first = log.read() + assert [line["seq"] for line in first["lines"]] == [slow["seq"], slow["seq"] + 1] and first["last"] == 2 + assert first["lines"][0]["action"] == {"via": "cdp", "method": "Runtime.evaluate", "acts": True, "tab": "T1", + "unfinished": True} + log.finish(slow, ok=True) + later = log.read(after=first["last"]) + assert later["lines"] == [{"seq": 3, "at": later["lines"][0]["at"], "outcome": {"entry": slow["seq"], "ok": True}}] + assert [line["seq"] for line in log.read()["lines"]] == [1, 2, 3] + + +def test_a_restarted_server_carries_on_the_sequence_on_a_new_line_after_a_partial_one(tmp_path): + path = tmp_path / "browser-chromium.jsonl" + path.write_text('{"seq": 1, "at": "t", "reset": {}}\n{"seq": 2, "at": "t", "tab": {}}\n{"seq": 3, "at"') + log = BrowserLog(path) + log.note("reset", {}) + assert [line["seq"] for line in log.read()["lines"]] == [1, 2, 3] + assert path.read_text().endswith('\n{"seq": 3, "at"\n' + json.dumps(log.read()["lines"][-1]) + "\n") + + +def test_a_log_that_cant_be_written_warns_and_keeps_its_numbers(tmp_path, caplog): + blocked = tmp_path / "file" + blocked.write_text("") + log = BrowserLog(blocked / "browser-chromium.jsonl") + log.note("reset", {}) + log.note("reset", {}) + assert log.read() == {"lines": [], "last": 2, "more": False} + assert len([r for r in caplog.records if "can't be written" in r.getMessage()]) == 1 + + +class Pipe: + """A browser's pipe for agentenv's control: a new tab's target said to open before it's answered, a navigation + said to commit before its loader is answered, a close said to destroy its target.""" + + def __init__(self) -> None: + self.listeners, self.tabs, self.focus, self.checked = [], 0, "text", [] + + def listen(self, listener) -> None: + self.listeners.append(listener) + + def emit(self, method: str, params: dict, session=None) -> None: + for listener in self.listeners: + listener({"method": method, "params": params, **({"sessionId": session} if session else {})}) + + def open(self, target: str, opener=None) -> None: + self.emit("Target.targetCreated", {"targetInfo": {"targetId": target, "type": "page", "url": "about:blank", + **({"openerId": opener} if opener else {})}}) + self.emit("Target.attachedToTarget", {"sessionId": f"s-{target}", "targetInfo": {"targetId": target, + "type": "page"}}) + + async def call(self, method, params=None, *, session_id=None, timeout_s=None): + if method == "Target.createTarget": + self.tabs += 1 + self.open(target := f"N{self.tabs}") + return {"targetId": target} + if method == "Page.navigate": + target = session_id.removeprefix("s-") + self.emit("Page.frameNavigated", {"frame": {"id": target, "url": params["url"], "loaderId": "L-nav"}}, + session_id) + return {"frameId": target, "loaderId": "L-nav"} + if method == "Target.closeTarget": + self.emit("Target.targetDestroyed", {"targetId": params["targetId"]}) + return {"success": True} + if method == "Runtime.evaluate" and "visibilityState" in params["expression"]: + return {"result": {"value": ["visible", True]}} + if method == "Runtime.evaluate" and "password" in params["expression"]: + self.checked.append(self.focus == "password") + return {"result": {"value": self.focus == "password"}} + if method == "Input.dispatchKeyEvent" and params.get("key") == "Tab": + self.focus = "password" + if method == "Page.getNavigationHistory": + return {"currentIndex": 0, "entries": [{"id": 1, "url": "about:blank"}]} + return {} + + async def request(self, method, params=None, *, session_id=None): + return {"result": await self.call(method, params, session_id=session_id)} + + +async def _settled() -> None: + for _ in range(5): + await asyncio.sleep(0) + + +@pytest.mark.asyncio +async def test_a_move_within_a_document_is_the_commands_that_asked_for_it_and_uses_up_a_history_move(): + causes = browser_log.Causes() + asked, back = {"seq": 3, "via": "cdp", "who": "ada"}, {"seq": 4, "via": "mcp", "who": "bo"} + answer = asyncio.get_running_loop().create_future() + causes.navigating("T1", answer, asked) + answer.set_result(None) + assert await causes.navigated_within("T1") == asked + causes.navigates_next("T1", back) + assert await causes.navigated_within("T1") == back + assert await causes.navigated("T1", "L-page") is None + failed = asyncio.get_running_loop().create_future() + causes.navigating("T1", failed, asked) + failed.cancel() + await asyncio.sleep(0) + assert await causes.navigated_within("T1") is None + + +@pytest.mark.asyncio +async def test_a_move_back_with_nowhere_to_go_leaves_no_cause_behind(tmp_path): + pipe = Pipe() + control = BrowserControl(pipe, "agent", (1280, 800)) + control.log = log = BrowserLog(tmp_path / "browser-chromium.jsonl") + pipe.open("T1") + await _settled() + token = ACTOR.set({"seq": 99, "via": "mcp", "who": "bo", "acts": True, "tab": None, "password": None}) + try: + with pytest.raises(InvalidArgument): + await control.history("back") + finally: + ACTOR.reset(token) + pipe.emit("Page.frameNavigated", {"frame": {"id": "T1", "url": "https://a.test/", "loaderId": "L-page"}}, "s-T1") + await _settled() + assert [line["tab"] for line in log.read()["lines"] if line["tab"]["event"] == "navigated"] == [ + {"event": "navigated", "id": "T1", "url": "https://a.test/", "by": "page", "after": 99}] + + +@pytest.mark.asyncio +async def test_typing_that_tabs_into_a_password_field_is_withheld_from_then_on(tmp_path): + pipe = Pipe() + control = BrowserControl(pipe, "agent", (1280, 800)) + control.log = BrowserLog(tmp_path / "browser-chromium.jsonl") + pipe.open("T1") + await _settled() + actor = {"seq": 1, "via": "mcp", "who": "bo", "acts": True, "tab": None, "password": None} + token = ACTOR.set(actor) + try: + await control.dispatch([{"method": "Input.dispatchKeyEvent", "params": {"type": "keyDown", "key": "a", "text": "a"}}, + {"method": "Input.dispatchKeyEvent", "params": {"type": "keyDown", "key": "Tab"}}, + {"method": "Input.dispatchKeyEvent", "params": {"type": "keyDown", "key": "s", "text": "s"}}]) + finally: + ACTOR.reset(token) + assert actor["password"] is True and actor["tab"] == "T1" + assert pipe.checked == [False, False, True] + + +@pytest.mark.asyncio +async def test_each_tab_event_names_what_caused_it(tmp_path): + pipe = Pipe() + control = BrowserControl(pipe, "agent", (1280, 800)) + control.log = log = BrowserLog(tmp_path / "browser-chromium.jsonl") + control._launching = True + pipe.open("T0") + control._launching = False + await _settled() + + begun = log.begin({"via": "mcp", "who": "ada", "op": "tabs/open", "acts": True}) + token = ACTOR.set({"seq": begun["seq"], "via": "mcp", "who": "ada", "acts": True, "tab": None}) + try: + opened = await control.open_tab("https://shop.test/?q=desk&token=t") + finally: + ACTOR.reset(token) + log.finish(begun, ok=True) + await _settled() + clicked = log.begin({"via": "cdp", "who": "bo", "method": "Input.dispatchMouseEvent", "acts": True, "tab": opened}) + log.finish(clicked, ok=True) + pipe.open("P1", opener=opened) + pipe.emit("Page.frameNavigated", {"frame": {"id": opened, "url": "https://shop.test/item", "loaderId": "L-page"}}, + f"s-{opened}") + pipe.emit("Page.navigatedWithinDocument", {"frameId": opened, "url": "https://shop.test/item#reviews"}, + f"s-{opened}") + await control.close_tab("P1") + control.ended("stop") + await _settled() + + events = [line["tab"] for line in log.read()["lines"] if "tab" in line] + assert events == [ + {"event": "opened", "id": "T0", "url": "about:blank", "by": "agentenv"}, + {"event": "opened", "id": opened, "url": "about:blank", "by": "command", "entry": begun["seq"], "via": "mcp", + "who": "ada"}, + {"event": "navigated", "id": opened, "url": f"https://shop.test/?q=desk&token={REDACTED}", "by": "command", + "entry": begun["seq"], "via": "mcp", "who": "ada"}, + {"event": "opened", "id": "P1", "url": "about:blank", "opener": opened, "by": "page", "after": clicked["seq"]}, + {"event": "navigated", "id": opened, "url": "https://shop.test/item", "by": "page", "after": clicked["seq"]}, + {"event": "navigated", "id": opened, "url": "https://shop.test/item#reviews", "within_document": True, + "by": "page", "after": clicked["seq"]}, + {"event": "closed", "id": "P1", "url": "about:blank", "by": "agentenv"}, + {"event": "closed", "id": "T0", "url": "about:blank", "by": "stop"}, + {"event": "closed", "id": opened, "url": "about:blank", "by": "stop"}, + ] diff --git a/packages/agentenv-protocol/tests/test_tool_context_injection.py b/packages/agentenv-protocol/tests/test_tool_context_injection.py index 4d42b6fd4..4da22c5ea 100644 --- a/packages/agentenv-protocol/tests/test_tool_context_injection.py +++ b/packages/agentenv-protocol/tests/test_tool_context_injection.py @@ -38,7 +38,7 @@ tool, ) from agentenv_protocol.agent_env_environment import _is_fastmcp_context_annotation, _schema_from_signature # noqa: E402 -from agentenv_protocol.tool_context import EMPTY, bound, context_from_mcp # noqa: E402 +from agentenv_protocol.tool_context import EMPTY, bound, context_from_http, context_from_mcp # noqa: E402 def _text(result: Any) -> str: @@ -182,6 +182,11 @@ def test_a_dict_meta_is_read_as_the_custom_entries(): assert context_from_mcp(SimpleNamespace(request_context=SimpleNamespace(meta=42, request=None)), "t", {}).caller == Caller() +def test_only_an_mcp_tool_calls_context_says_it_is_one(): + assert context_from_mcp(None, "computer_click", {}).mcp is True + assert context_from_http(None, "_computer_ready_status", {}).mcp is False and EMPTY.mcp is False + + @pytest.mark.asyncio async def test_header_fallback_and_default_blank_roles(): env = _Env() diff --git a/src/agent_env/browsers/README.md b/src/agent_env/browsers/README.md index 5ee314f81..04204faf8 100644 --- a/src/agent_env/browsers/README.md +++ b/src/agent_env/browsers/README.md @@ -24,6 +24,11 @@ its website browser with `browser_envs = ["chromium"]`, in place of the run's `p sandbox of its own beside the multi env's computers, sized by the deploy step's `child_envs.chromium`, its tools served under `chromium_`, routing the multi env's synthetic sites through the gateway, its network its own env's, as any computer child's. +Its server keeps its log in `/var/log/agentenv/browser-chromium.jsonl`: an entry for each CDP command, MCP tool call +and pixel action an agent takes, and an event for each tab opened, navigated or closed with what caused it, secrets +taken out (`agentenv_protocol.website_browser.log`); `browser/v1` `log` reads it and `data/get` exports it as +`browser-log.jsonl`. It lasts the computer's boot: `reset_env` boots the computer afresh with a new one, so a task that +needs the log before a reset takes it first with `snapshot_env`. A task changes a browser's state with the steps in `agent_env.task_step.task_steps.browser`, each a call to one `browser/v1` operation on the browser in the env `env_id`, the one `browser` names when the env has several: diff --git a/src/agent_env/drivers/README.md b/src/agent_env/drivers/README.md index bae467356..ab74b5948 100644 --- a/src/agent_env/drivers/README.md +++ b/src/agent_env/drivers/README.md @@ -19,7 +19,7 @@ and input reaches it as CDP input events, both through `browser/v1`'s `page/*` o a tab that paints nothing shows white. It serves the page group too: `computer_navigate`, `computer_history`, `computer_tabs`, `computer_dialog` and `computer_upload`, and every frame names the page, its tab and what is open on it. The browser ending on its own fails the computer for good: its ready check is -`final`, and its tools answer with why. Name it as `page`, or by its id, `@local/agentenv-framework/drivers/page`. +`final`, and its tools answer with why. Each call it makes to the browser for an agent's tool carries that call, its tool, id, arguments and the caller's role, in the `AgentEnv-Tool-Call` and `AgentEnv-Role` headers, so the browser's log credits the action to it. Name it as `page`, or by its id, `@local/agentenv-framework/drivers/page`. A run that names it writes it: its files, what its class declares for each platform it supports, read without `mss`, `python-xlib` or PyAV installed, and how a computer installs it. Its `env.toml`'s `[install.linux]` has a diff --git a/src/agent_env/drivers/envs/page/driver.py b/src/agent_env/drivers/envs/page/driver.py index fdbc8ae68..4a75e4144 100644 --- a/src/agent_env/drivers/envs/page/driver.py +++ b/src/agent_env/drivers/envs/page/driver.py @@ -15,6 +15,7 @@ import httpx from agentenv_protocol import client +from agentenv_protocol.website_browser.log import TOOL_CALL_HEADER, tool_call_header from agentenv_protocol.computer import ( ComputerDriverEnvironment, DeviceUnavailable, @@ -51,6 +52,7 @@ ) from agentenv_protocol.computer import apps from agentenv_protocol.computer.served import served +from agentenv_protocol.tool_context import ROLE_HEADER, ToolContext from agentenv_protocol.website_browser import BROWSER BUTTONS = {"left": 1, "right": 2, "middle": 4, "back": 8, "forward": 16} @@ -264,7 +266,7 @@ async def _call(self, op: str, params: Optional[dict] = None) -> Any: raise Unsupported(f"the browser in {folder.name} serves no {op}") try: async with _http(folder) as http: - answer = await http.request(method["method"], method["endpoint"], + answer = await http.request(method["method"], method["endpoint"], headers=_tool_call(), **({"json": params} if method["method"] != "GET" else {})) except (httpx.HTTPError, OSError) as e: raise DeviceUnavailable(_ended(folder) or f"the browser's server isn't answering: {type(e).__name__}: {e}" @@ -313,6 +315,16 @@ def _http(folder: Path) -> httpx.AsyncClient: return served(folder).client(CALL_TIMEOUT_S) +def _tool_call() -> dict[str, str]: + """The agent's tool call a call to the browser is made for, which the browser's log credits it to: its tool, its + call's id, its arguments and its caller's role; nothing outside one.""" + context = ToolContext.current() + if not context.mcp or not context.tool: + return {} + return {TOOL_CALL_HEADER: tool_call_header(context.tool, context.call_id, context.arguments), + **({ROLE_HEADER: context.caller.role} if context.caller.role else {})} + + def _mouse(kind: str, x: float, y: float, button: str = "none", buttons: int = 0, count: int = 0, modifiers: int = 0) -> dict: return {"method": "Input.dispatchMouseEvent", "params": { diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index 5e3656cca..54527640d 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -29,7 +29,9 @@ reset_env brings it back after its container is killed and after the gateway restarts; a multi env naming none takes the run's browser, chromium too, the same way, and ``browser_envs = []`` deploys none; and browser_connect_cdp hands an agent an address to it at the multi env's gateway, whose relay takes the agent's page to the site by name, -with the multi env's tools hidden from the agent's role. +with the multi env's tools hidden from the agent's role. Its log has an entry for every action a CDP agent and an MCP agent take, over CDP, page +tools and pixels, with who, the tab and its URL and no secret either typed or set, and each tab event names the +command that caused it. Needs Docker and the local registry on localhost:5000, and Node 20 or later and npm for the CDP clients.""" @@ -59,6 +61,7 @@ from agentenv_protocol.client import find_extension_method, invoke_extension from agentenv_protocol.computer import COMPUTER_PAGE, COMPUTER_READY from agentenv_protocol.computer.conformance.client import image +from agentenv_protocol.tool_context import ROLE_HEADER from agentenv_protocol.website_browser import BROWSER from agent_env.a2a_agent import conversation_store @@ -514,6 +517,115 @@ async def test_its_data_calls_take_turns_a_server_restart_takes_a_load_and_with_ assert not report.failed +#: A form whose password field and text field fill the page's top, for typing into either by a click. +FORM = "data:text/html," + urllib.parse.quote( + "form") +#: The secrets the agents send, which the log must never hold, and the text they type that it must. +SECRETS, TYPED = ("hunter2", "hunter3", "s3cret"), ("standing desk", "oak desk") + + +@pytest.mark.asyncio +async def test_its_log_has_an_entry_for_every_action_over_cdp_page_tools_and_pixels_with_no_secrets(browser_computer): + """M8.19: one task driven over CDP, page tools and pixels has an entry for every action, with its time, who, tab, + URL and method or tool, and no secrets; each tab event names the command that caused it.""" + with namespace_routing(): + deployed = await Env.get(browser_computer.id).deploy() + context = TaskStepContext(deployed_envs=[deployed]) + try: + sent = await _cdp_agent_types((await _cdp(deployed, "ada"))["ws_url"]) + async with streamablehttp_client(deployed.mcp_url, headers={ROLE_HEADER: "bo"}) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + await _tool(session, "computer_navigate", {"url": FORM}) + await _shown(session, lambda page: page.get("title") == "form") + await _tool(session, "computer_click", {"x": 300, "y": 40, "space": "frame"}) + await _tool(session, "computer_type", {"text": "hunter3"}) + await _tool(session, "computer_click", {"x": 300, "y": 160, "space": "frame"}) + await _tool(session, "computer_type", {"text": "oak desk"}) + await _tool(session, "computer_tabs", {"action": "list"}) + shot = await session.call_tool("computer_screenshot", {"space": "frame"}) + await asyncio.sleep(1) + logged = await _browser_log(deployed) + finally: + report = await teardown_run(context) + + lines = logged["lines"] + actions = [line["action"] | {"seq": line["seq"], "at": line["at"]} for line in lines if "action" in line] + tab_events = [line["tab"] for line in lines if "tab" in line] + by_ada = [action for action in actions if action["via"] == "cdp" and action["who"] == "ada"] + by_bo = [action for action in actions if action["via"] == "mcp"] + text = json.dumps(logged) + assert not shot.isError and [line["seq"] for line in lines] == sorted(line["seq"] for line in lines) + assert [action["method"] for action in by_ada] == [method for method, _ in sent], by_ada + assert all(action["at"] and action["tab"] == sent[0][1] and isinstance(action["url"], str) for action in by_ada) + assert {action["method"]: action["acts"] for action in by_ada} == { + "Target.createTarget": True, "Target.attachToTarget": False, "Runtime.evaluate": True, "Input.insertText": True, + "Network.setCookie": True, "DOM.getDocument": False, "Page.navigate": True, "Target.closeTarget": True} + typed = [action["params"]["text"] for action in by_ada if action["method"] == "Input.insertText"] + assert typed == [{"length": 7}, "standing desk"] + tools = {action["tool"] for action in by_bo} + assert tools == {"computer_navigate", "computer_click", "computer_type", "computer_tabs", "computer_screenshot"} + assert all(action["who"] == "bo" and action["at"] and action["op"] and action["call"] for action in by_bo) + acting = [action for action in by_bo if action["acts"]] + assert all(action["tab"] and isinstance(action["url"], str) for action in acting), acting + assert [action["args"]["text"] for action in acting if action["tool"] == "computer_type"] == \ + [{"length": 7}, "oak desk"] + assert any(action["tool"] == "computer_click" and action["op"] == "page/input" and action["params"]["events"] + for action in acting) + assert not any(secret in text for secret in SECRETS) and all(words in text for words in TYPED) + created = next(action for action in by_ada if action["method"] == "Target.createTarget") + navigated = next(action for action in by_ada if action["method"] == "Page.navigate") + closed = next(action for action in by_ada if action["method"] == "Target.closeTarget") + opened_by_bo = next(action for action in by_bo if action["op"] == "navigate") + ada_tab = [event for event in tab_events if event["id"] == created["tab"]] + assert {"event": "opened", "by": "command", "entry": created["seq"], "via": "cdp", "who": "ada"}.items() <= \ + ada_tab[0].items(), ada_tab + assert any(event["event"] == "navigated" and event.get("entry") == navigated["seq"] for event in ada_tab), ada_tab + assert {"event": "closed", "by": "command", "entry": closed["seq"]}.items() <= ada_tab[-1].items(), ada_tab + assert any(event["event"] == "navigated" and event.get("entry") == opened_by_bo["seq"] and event["who"] == "bo" + for event in tab_events), tab_events + assert not report.failed + + +async def _cdp_agent_types(ws_url: str) -> list[tuple[str, str]]: + """A CDP agent, ada, that opens the form in a tab of its own, types a password and a search into it, sets a cookie, + reads the page, navigates and closes the tab: each command it sent, in order, and the tab, the same for all.""" + async with connect(ws_url, max_size=None, open_timeout=60) as ws: + ids = itertools.count(1) + sent: list[str] = [] + + async def call(method: str, params: dict, session: str | None = None) -> dict: + message_id = next(ids) + sent.append(method) + await ws.send(json.dumps({"id": message_id, "method": method, "params": params, + **({"sessionId": session} if session else {})})) + while (reply := json.loads(await asyncio.wait_for(ws.recv(), 60))).get("id") != message_id: + pass + return reply + + target = (await call("Target.createTarget", {"url": FORM}))["result"]["targetId"] + session = (await call("Target.attachToTarget", {"targetId": target, "flatten": True}))["result"]["sessionId"] + await asyncio.sleep(1) + await call("Runtime.evaluate", {"expression": "document.getElementById('p').focus()"}, session) + await call("Input.insertText", {"text": "hunter2"}, session) + await call("Runtime.evaluate", {"expression": "document.getElementById('t').focus()"}, session) + await call("Input.insertText", {"text": "standing desk"}, session) + await call("Network.setCookie", {"name": "sid", "value": "s3cret", "url": "https://shop.test/"}, session) + await call("DOM.getDocument", {"depth": 1}, session) + await call("Page.navigate", {"url": FORM + "%3C!--again--%3E"}, session) + await asyncio.sleep(1) + await call("Target.closeTarget", {"targetId": target}) + return [(method, target) for method in sent] + + +async def _browser_log(deployed, after: int = 0) -> dict: + card = (await _get(deployed, "/.well-known/agent-env.json")).json() + app = next(child for child in card["children_environments"] if child["name"] == "chromium") + path = protocol_v1.find_extension_method(app, BROWSER, "log")["endpoint"] + return (await _get(deployed, path, headers=deployed.control_headers, params={"after": after})).json() + + async def _shown(session: ClientSession, holds) -> dict: """A screenshot's result once the page it shows is what ``holds`` waits for, within ``WAIT_S``.""" last, deadline = None, time.monotonic() + WAIT_S diff --git a/tst/integration/computer/website_browser_test.py b/tst/integration/computer/website_browser_test.py index a3fc07c38..f2b0f7a58 100644 --- a/tst/integration/computer/website_browser_test.py +++ b/tst/integration/computer/website_browser_test.py @@ -157,7 +157,8 @@ async def test_chromium_runs_with_agentenvs_flags_on_its_pipe_and_a_reset_relaun assert (params["chromium"], params["clock"], sorted(params["methods"])) == ( "156.0.8078.4", "real", ["cdp", "cookies/delete", "cookies/get", "cookies/set", "dialog/answer", "downloads/get", "downloads/list", "files/choose", "history", "local_storage/clear", - "local_storage/get", "local_storage/set", "navigate", "page/capture", "page/input", + "local_storage/get", "local_storage/set", "log", "navigate", "page/capture", + "page/input", "page/state", "page/viewport", "record", "sites/set", "tabs/activate", "tabs/close", "tabs/list", "tabs/open"]) assert readiness["ready"] is True, readiness diff --git a/tst/unit/env/gateway/browser_route_test.py b/tst/unit/env/gateway/browser_route_test.py index e0d0ad723..4ca80866a 100644 --- a/tst/unit/env/gateway/browser_route_test.py +++ b/tst/unit/env/gateway/browser_route_test.py @@ -15,7 +15,7 @@ import pytest import uvicorn import websockets -from agentenv_protocol.browser_cdp import ROUTE +from agentenv_protocol.website_browser.cdp import ROUTE from agentenv_protocol.computer.app import COMPUTER_APP from agentenv_protocol.website_browser import BROWSER from pytest_socket import enable_socket