diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index 828fe8299..93779fc3a 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -757,8 +757,11 @@ are the profile's template. The base class adds: - **Its settings**, `BrowserSettings`, which a subclass's `Settings` extends: `start_urls`, opened in tabs at launch, the first active (`["about:blank"]` by default; none may start with `-`), headless the first from its command line and the rest over its pipe, since headless Chromium takes one; `window`, - 1280 × 800, alone only; `locale`; `time_zone`, set as `TZ` in its launch's `env`; and `geolocation` and - `notifications`, `deny` (the default), which answers a page's prompt for it, or `ask`. + 1280 × 800, alone only; `locale`; `time_zone`, set as `TZ` in its launch's `env`; `geolocation` and + `notifications`, `deny` (the default), which answers a page's prompt for it, or `ask`; and `dialogs`, who + answers a page's JavaScript dialogs: `agent` (the default), or agentenv, accepting (`accept`, a `prompt()` + with its default text) or dismissing (`dismiss`) each. Either way agentenv leaves a page that asks before + it's left (`beforeunload`), so none of its answers blocks a navigation. - **The flags agentenv owns**, after the subclass's command and before the start tabs: its profile, its HTTP cache under `~/.cache` (`--disk-cache-dir`), `--remote-debugging-pipe`, `--no-first-run`, `--no-default-browser-check`, `--lang` and, when `/dev/shm` holds under 1 GiB (Docker gives 64 MiB), @@ -786,8 +789,19 @@ are the profile's template. The base class adds: launches it again, and a launch while it runs, from its menu launcher say, opens another window over the pipe rather than a second copy, which would hand its window to the first and exit. The ready check `devtools_answers` asks `Browser.getVersion` over the pipe. -- **`browser/v1`** on its card, params only: `chromium` (its `version`), `clock: "real"` and, alone, - `window`. +- **Its control connection**, `self.control`, a `BrowserControl` over the pipe. At each launch it attaches + to every page with flattened sessions and keeps a record, `since` its server started: open tabs in + opened order, the one whose page has focus `active`; `closed_tabs`, each with `closed_at` and + `closed_by` (`agentenv`, over this connection; `unknown`, anyone else; `exit` or `stop`, when the browser + ended on its own or its server stopped it); `downloads`, into `~/Downloads`, each with its state + and, completed, its size and sha256; and `dialogs`, each with its type, message, 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 + `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. + Nothing relaunches a browser that exits on its own: in a computer the agent can, and alone it's an infra + failure, its ready check false. - **Its skill**: sites by their real names, files to upload in `~/Uploads` and downloads in `~/Downloads`, folders it makes. diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py index 3505055ef..c1ab84f2d 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py @@ -15,7 +15,7 @@ from ..agent_env_environment import OP_ADD_DATA, OP_GET_DATA, OP_RESET_DATA from ..transfers import WriteNamespaceGrant -from ..types import FilePart, uploaded_object_path +from ..types import METHOD_ADD, METHOD_RESET, FilePart, uploaded_object_path from .dataplane import file_parts, part_bytes if TYPE_CHECKING: @@ -23,6 +23,8 @@ logger = logging.getLogger(__name__) +#: The cause a launch after each data call logs. +_CAUSES = {OP_ADD_DATA: METHOD_ADD, OP_RESET_DATA: METHOD_RESET} #: Where an app's server keeps its paths as installed, in the app's folder. INITIAL = Path(".agentenv/initial") @@ -80,7 +82,7 @@ async def _change(self, op: str, served: Callable, args: tuple, answer: asyncio. raise finally: if running: - await self.runtime.launch_app() + await self.runtime.launch_app(_CAUSES[op]) except Exception as e: if not answer.done(): answer.set_exception(e) diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py index fc30ec1bf..a5f0c88d2 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py @@ -5,10 +5,13 @@ import asyncio import inspect +import json import os import shlex +import signal import time from collections.abc import Awaitable, Callable +from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any, Optional @@ -26,6 +29,11 @@ #: Where an app's server keeps the pid of the app it launched last, in the app's folder, so a server that restarts can #: still stop it. PID = Path(".agentenv/app.pid") +#: Where an app's server logs each launch, stop and exit of the app, beside its pid; a boot starts the app's folder as +#: its image has it, so the log is the boot's. +LAUNCHES = Path(".agentenv/launches") +#: How long an app that ended on its own may take to be reaped, for its exit code. +EXIT_WAIT_S = 5.0 #: How long a stopped app may take to exit before it's killed. STOP_TIMEOUT_S = 10.0 @@ -59,6 +67,7 @@ def __init__(self, launch: Optional[Launch], ready_checks: dict[str, Callable[[] self.launch_fds = launch_fds self.relaunch_on_start = relaunch_on_start self.launch_running = launch_running + self.exits: set[asyncio.Future] = set() self.started = time.time() self.launch_failed: Optional[str] = None self.checking: dict[str, asyncio.Future] = {} @@ -80,14 +89,14 @@ async def start(self) -> None: try: if relaunch: await self.stop_app() - await self.launch_app() + await self.launch_app("server_restart" if relaunch else "boot") except Exception as e: self.launch_failed = f"{type(e).__name__}: {e}" - async def launch_app(self) -> int: + async def launch_app(self, cause: str = "launch") -> int: """Start the app from its folder, marked as launched, with the fds the app gives it, whose copies here are - closed once it has them; its pid. With ``launch_running``, an app this server launched that still runs does - that instead of starting a second copy, as a browser opens another window.""" + closed once it has them, and logged with ``cause``; its pid. With ``launch_running``, an app this server + launched that still runs does that instead of starting a second copy, as a browser opens another window.""" if self.launch_running is not None and (running := _running()) is not None: await self.launch_running() return running @@ -104,6 +113,7 @@ async def launch_app(self) -> int: PID.write_text(str(pid)) _mark().parent.mkdir(parents=True, exist_ok=True) _mark().touch() + self.log("launch", pid, cause=cause) return pid async def stop_app(self) -> bool: @@ -115,9 +125,39 @@ async def stop_app(self) -> bool: PID.unlink() if not session.alive(pid): return False + self.log("stop", pid) await session.stop(pid, STOP_TIMEOUT_S) return True + def exited(self) -> bool: + """Log that the app this server launched last ended on its own, with its exit code or the signal that ended + it, unless the server stopped it; whether it did.""" + if (pid := _recorded_pid()) is None: + return False + noting = asyncio.ensure_future(self._log_exit(pid)) + self.exits.add(noting) + noting.add_done_callback(self.exits.discard) + return True + + async def _log_exit(self, pid: int) -> None: + code = await session.returncode(pid, EXIT_WAIT_S) + self.log("exit", pid, **({} if code is None else + {"signal": signal.Signals(-code).name} if code < 0 else {"code": code})) + + def log(self, event: str, pid: int, **fields: Any) -> None: + """Append ``event`` for the app's ``pid`` to its launch log, which outlives its server for the boot.""" + entry = {"at": datetime.now(timezone.utc).isoformat(), "event": event, "pid": pid, **fields} + _log().parent.mkdir(parents=True, exist_ok=True) + with _log().open("a", encoding="utf-8") as log: + log.write(json.dumps(entry) + "\n") + + def launches(self) -> list[dict]: + """Every launch, stop and exit of the app since the computer booted, in order.""" + try: + return [json.loads(line) for line in _log().read_text(encoding="utf-8").splitlines() if line] + except FileNotFoundError: + return [] + async def readiness(self) -> AppReadiness: ready = self.launch.ready if self.launch is not None else None checks: list[tuple[str, Awaitable[Optional[str]]]] = [] @@ -154,6 +194,17 @@ def _mark() -> Path: return LAUNCHED / Path.cwd().name +def _log() -> Path: + return LAUNCHES + + +def _recorded_pid() -> Optional[int]: + try: + return int(PID.read_text()) + except (FileNotFoundError, ValueError): + return None + + def _launched_at() -> Optional[float]: """When the app was last launched since the computer booted, by its mark.""" try: @@ -164,11 +215,8 @@ def _launched_at() -> Optional[float]: def _running() -> Optional[int]: """The pid of the app this server launched last, while it runs.""" - try: - pid = int(PID.read_text()) - except (FileNotFoundError, ValueError): - return None - return pid if session.alive(pid) else None + pid = _recorded_pid() + return pid if pid is not None and session.alive(pid) else None async def _bounded(name: str, check: Awaitable[Optional[str]]) -> Optional[str]: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py index 01f704429..c1c98b42c 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py @@ -15,7 +15,7 @@ from pydantic import TypeAdapter, ValidationError from ..agent_env_environment import add_data, get_data -from ..transfers import NamespaceUploader, ReadObject, RelativePath, WriteNamespaceGrant, download +from ..transfers import NamespaceUploader, ReadObject, RelativePath, TransferError, WriteNamespaceGrant, download from ..types import DATA_OBJECTS_EXTENSION_URI, FilePart, FileWithBytes, FileWithUri, uploaded_file_part, uploaded_object_path if TYPE_CHECKING: @@ -64,22 +64,31 @@ async def _computer_data_get(self, write_namespace: Optional[WriteNamespaceGrant async def file_parts(home: Path, paths: list[str], write_namespace: Optional[WriteNamespaceGrant], leaving_out: list[str] = ()) -> list: """A part for each file under ``paths`` of ``home`` but under none of ``leaving_out``, naming its path relative to - ``home``: uploaded under ``write_namespace`` when there is one, else inline.""" + ``home``: uploaded under ``write_namespace`` when there is one, else inline. A file a running app removes before it's + read is left out.""" files = [(relative, source) for relative, source in _data_files(home, paths) if not any(PurePosixPath(relative).is_relative_to(owned) for owned in leaving_out)] + parts = [] if write_namespace is not None: uploader = NamespaceUploader(write_namespace) - parts = [] for relative, source in files: - await uploader.upload(relative, source) + try: + await uploader.upload(relative, source) + except (TransferError, OSError): + if source.exists(): + raise + continue parts.append(uploaded_file_part(relative, name=source.name, mime_type=_mime(source))) return parts - return [ - FilePart(file=FileWithBytes(bytes=base64.b64encode(source.read_bytes()).decode("ascii"), - name=source.name, mimeType=_mime(source)), - metadata={DATA_OBJECTS_EXTENSION_URI: {"path": relative}}) - for relative, source in files - ] + for relative, source in files: + try: + data = source.read_bytes() + except FileNotFoundError: + continue + parts.append(FilePart(file=FileWithBytes(bytes=base64.b64encode(data).decode("ascii"), name=source.name, + mimeType=_mime(source)), + metadata={DATA_OBJECTS_EXTENSION_URI: {"path": relative}})) + return parts async def part_bytes(part: FilePart) -> bytes: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py index f9ca0d261..b56b8eb0b 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py @@ -17,6 +17,10 @@ WITH_FDS = Path(__file__).with_name("_with_fds.py") #: The processes launched into the session, kept until they exit so each is reaped. _running: set[asyncio.Task] = set() +#: Each launched process's reaping, by pid, until its exit is asked for: its return code, negative for a signal. +_exits: dict[int, asyncio.Future] = {} +#: How many exits no one has asked for are kept, the latest. +KEPT_EXITS = 16 async def wait_until_up() -> None: @@ -38,12 +42,35 @@ async def launch(argv: list[str], cwd: Path, env: dict[str, str], fds: Optional[ reaped = asyncio.ensure_future(process.wait()) _running.add(reaped) reaped.add_done_callback(_running.discard) + _exits[process.pid] = reaped + reaped.add_done_callback(lambda _: _forget_old_exits()) return process.pid +def _forget_old_exits() -> None: + """Drop all but the latest ``KEPT_EXITS`` exits no one has asked for.""" + for pid in [pid for pid, reaped in _exits.items() if reaped.done()][:-KEPT_EXITS]: + del _exits[pid] + + +async def returncode(pid: int, timeout_s: float) -> Optional[int]: + """The return code the process ``launch`` started as ``pid`` exited with, negative for the signal that ended it, + once it's reaped within ``timeout_s``; None for one ``launch`` didn't start or that hasn't exited.""" + reaped = _exits.get(pid) + if reaped is None: + return None + try: + code = await asyncio.wait_for(asyncio.shield(reaped), timeout_s) + except TimeoutError: + return None + _exits.pop(pid, None) + return code + + async def stop(pid: int, timeout_s: float) -> None: """End the session ``launch`` started as ``pid``, every process in it: asked to exit, then killed after - ``timeout_s``.""" + ``timeout_s``. No one asks how it exited once it's stopped.""" + _exits.pop(pid, None) try: os.killpg(pid, signal.SIGTERM) except ProcessLookupError: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py index 8d7c354ee..083cfb251 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser.py @@ -7,19 +7,22 @@ from __future__ import annotations import asyncio +import contextlib +import hashlib import itertools import json import logging import os import shutil +from collections.abc import AsyncIterator +from datetime import datetime, timezone from pathlib import Path from typing import Any, Awaitable, Callable, ClassVar, Literal, Optional from pydantic import BaseModel, Field, field_validator -from .agent_env_environment import agent_skill +from .agent_env_environment import agent_skill, extension from .computer import AppDeclaration, ComputerAppEnvironment, Launch, install_step, ready_check -from .types import EnvironmentExtension logger = logging.getLogger(__name__) @@ -36,6 +39,18 @@ SHM_MIN_BYTES = 1 << 30 #: Chromium's content setting that denies a permission without a prompt. BLOCK = 2 +#: Who closed a tab: agentenv over its own connection, or someone it can't tell; or how its browser ended, on its own +#: or stopped by its server. +AGENTENV, UNKNOWN, EXIT, STOP = "agentenv", "unknown", "exit", "stop" +#: How long the active tab's check waits on each page. +ACTIVE_CHECK_TIMEOUT_S = 2.0 +#: How much of a download each read hashes. +HASH_CHUNK_BYTES = 1 << 20 + +#: Who answered a dialog when agentenv didn't. +AGENT = "agent" +#: The targets agentenv's connection follows. +PAGES = [{"type": "page"}] #: Alone, loopback goes through the proxy too. In a computer, the apps on loopback beside it stay reachable; ``<-loopback>`` #: removes Chromium's own rule, so ``*.localhost`` names still don't. ALONE_BYPASS = "<-loopback>" @@ -77,7 +92,8 @@ class DevToolsError(RuntimeError): class DevToolsPipe: """A CDP connection over a Chromium's debugging pipe, each message JSON ended by a NUL. It connects anew at every - launch; a call with no browser connected, or whose browser exits before answering, raises ConnectionError.""" + launch; a call with no browser connected, or whose browser exits before answering, raises ConnectionError. Each + event, a message with no ``id``, goes to every listener, and each closer hears when the browser closes the pipe.""" #: How long a call waits for the browser's answer when it names no wait of its own. call_timeout_s: ClassVar[float] = 30.0 @@ -87,6 +103,16 @@ def __init__(self) -> None: self._waiting: dict[int, asyncio.Future] = {} self._writer: Optional[asyncio.WriteTransport] = None self._reading: Optional[asyncio.Task] = None + self._listeners: list[Callable[[dict], None]] = [] + self._closers: list[Callable[[], None]] = [] + + def listen(self, listener: Callable[[dict], None]) -> None: + """Hand ``listener`` each event: its ``method``, ``params`` and, on a page's session, ``sessionId``.""" + self._listeners.append(listener) + + def on_close(self, closer: Callable[[], None]) -> None: + """Call ``closer`` when the browser closes its end of the pipe, not when this end closes it.""" + self._closers.append(closer) async def connect(self, to_browser: int, from_browser: int) -> None: """Take over this end of a new pipe, ending any connection to a browser launched before.""" @@ -128,6 +154,10 @@ async def _read(self, reader: asyncio.StreamReader) -> None: try: while True: message = json.loads((await reader.readuntil(b"\0"))[:-1]) + if "id" not in message: + for listener in self._listeners: + _heard(listener, message) + continue answer = self._waiting.get(message.get("id")) if answer is not None and not answer.done(): answer.set_result(message) @@ -135,6 +165,8 @@ async def _read(self, reader: asyncio.StreamReader) -> None: if self._writer is not None: self._writer.close() self._fail_waiting() + for closer in self._closers: + _heard(closer) def _fail_waiting(self) -> None: for answer in self._waiting.values(): @@ -142,6 +174,195 @@ def _fail_waiting(self) -> None: answer.set_exception(ConnectionError("the browser closed its debugging pipe")) +def _heard(callback: Callable, *args: Any) -> None: + try: + callback(*args) + except Exception: + logger.exception("a DevToolsPipe listener failed") + +class BrowserControl: + """agentenv's control of a browser over its pipe: at each launch it attaches to every page with flattened sessions + and keeps a record of the browser since this server started, its open tabs in opened order, those closed and who + closed them, its downloads with their size and sha256, and its JavaScript dialogs, which it answers when + ``dialogs`` says to: always leaving a page that asks before it's left, so no answer of agentenv's blocks a + navigation.""" + + def __init__(self, devtools: DevToolsPipe, dialogs: str) -> None: + self.devtools, self.dialogs = devtools, dialogs + #: When this record starts: a server that restarts starts another. + self.since = _now() + self.tabs: dict[str, dict] = {} + self.closed_tabs: list[dict] = [] + self.downloads: dict[str, dict] = {} + self.dialog_log: list[dict] = [] + self._sessions: dict[str, str] = {} + self._hidden: set[str] = set() + self._closing: set[str] = set() + self._open_dialogs: dict[str, dict] = {} + self._answering: set[str] = set() + self._tasks: set[asyncio.Future] = set() + devtools.listen(self._event) + + async def attach(self) -> None: + """Follow a browser just launched: any tabs the one before left open closed by its server's stop, every page + 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}) + await self.devtools.call("Target.setAutoAttach", {"autoAttach": True, "waitForDebuggerOnStart": False, + "flatten": True, "filter": PAGES}) + await self.devtools.call("Browser.setDownloadBehavior", {"behavior": "default", "eventsEnabled": True}) + + 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()] + self.tabs.clear() + self._sessions.clear() + self._open_dialogs.clear() + + async def close_tab(self, target: str) -> None: + """Close the tab ``target``, which the record says agentenv closed.""" + self._closing.add(target) + await self.devtools.call("Target.closeTarget", {"targetId": target}) + + @contextlib.asynccontextmanager + async def hidden_page(self, url: str = "about:blank") -> AsyncIterator[str]: + """A page of agentenv's own for reading state, in no window and never in the record; its session's id. It's + closed after.""" + target = (await self.devtools.call("Target.createTarget", + {"url": url, "hidden": True, "background": True}))["targetId"] + self._hidden.add(target) + self.tabs.pop(target, None) + try: + yield (await self.devtools.call("Target.attachToTarget", {"targetId": target, "flatten": True}))["sessionId"] + finally: + with contextlib.suppress(ConnectionError, DevToolsError): + await self.devtools.call("Target.closeTarget", {"targetId": target}) + + async def active_tab(self) -> Optional[str]: + """The tab whose page has focus, else the first visible one; CDP has no event for a tab made active.""" + visible = None + for session, target in list(self._sessions.items()): + if target not in self.tabs: + continue + try: + state = (await self.devtools.call( + "Runtime.evaluate", {"expression": "[document.visibilityState, document.hasFocus()]", + "returnByValue": True}, session_id=session, + timeout_s=ACTIVE_CHECK_TIMEOUT_S))["result"]["value"] + except (ConnectionError, DevToolsError, asyncio.TimeoutError, KeyError): + continue + if state[1]: + return target + if state[0] == "visible" and visible is None: + visible = target + return visible + + async def record(self) -> dict: + """The record ``since`` it started: open ``tabs``, each with whether it's ``active``, ``closed_tabs``, + ``downloads`` and ``dialogs``.""" + active = await self.active_tab() + return {"since": self.since, "tabs": [{**tab, "active": tab["id"] == active} for tab in self.tabs.values()], + "closed_tabs": list(self.closed_tabs), "downloads": list(self.downloads.values()), + "dialogs": list(self.dialog_log)} + + def _event(self, message: dict) -> None: + handler = getattr(self, "_on_" + message.get("method", "").replace(".", "_"), None) + if handler is not None: + handler(message.get("params", {}), message.get("sessionId")) + + def _on_Target_targetCreated(self, params: dict, session: Optional[str]) -> None: + info = params["targetInfo"] + if info.get("type") == "page" and info["targetId"] not in self._hidden and info["targetId"] not in self.tabs: + self.tabs[info["targetId"]] = {"id": info["targetId"], "url": info.get("url", ""), + "title": info.get("title", ""), "opened_at": _now()} + + def _on_Target_targetInfoChanged(self, params: dict, session: Optional[str]) -> None: + info = params["targetInfo"] + if (tab := self.tabs.get(info["targetId"])) is not None: + tab.update(url=info.get("url", tab["url"]), title=info.get("title", tab["title"])) + + def _on_Target_targetDestroyed(self, params: dict, session: Optional[str]) -> None: + target = params["targetId"] + 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}) + self._closing.discard(target) + self._hidden.discard(target) + + def _on_Target_attachedToTarget(self, params: dict, session: Optional[str]) -> None: + if params["targetInfo"].get("type") == "page": + self._sessions[params["sessionId"]] = params["targetInfo"]["targetId"] + self._spawn(self.devtools.call("Page.enable", session_id=params["sessionId"])) + + def _on_Target_detachedFromTarget(self, params: dict, session: Optional[str]) -> None: + self._sessions.pop(params["sessionId"], None) + + def _on_Page_javascriptDialogOpening(self, params: dict, session: Optional[str]) -> None: + dialog = {"tab": self._sessions.get(session), "type": params.get("type"), "message": params.get("message", ""), + "url": params.get("url", ""), "opened_at": _now(), "answer": None, "by": None} + self.dialog_log.append(dialog) + self._open_dialogs[session] = dialog + if self.dialogs != AGENT: + self._answering.add(session) + leave = params.get("type") == "beforeunload" + self._spawn(self.devtools.call("Page.handleJavaScriptDialog", { + "accept": leave or self.dialogs == "accept", "promptText": params.get("defaultPrompt", "")}, + session_id=session)) + + def _on_Page_javascriptDialogClosed(self, params: dict, session: Optional[str]) -> None: + if (dialog := self._open_dialogs.pop(session, None)) is not None: + dialog.update(answer="accepted" if params.get("result") else "dismissed", answered_at=_now(), + by=AGENTENV if session in self._answering else AGENT, + **({"input": params["userInput"]} if params.get("userInput") else {})) + self._answering.discard(session) + + def _on_Browser_downloadWillBegin(self, params: dict, session: Optional[str]) -> None: + self.downloads[params["guid"]] = {"guid": params["guid"], "url": params.get("url", ""), + "filename": params.get("suggestedFilename", ""), "state": "inProgress", + "started_at": _now()} + + def _on_Browser_downloadProgress(self, params: dict, session: Optional[str]) -> None: + if (download := self.downloads.get(params["guid"])) is None: + return + download["state"] = params["state"] + if params["state"] == "completed": + if not params.get("filePath"): + download["ended_at"] = _now() + return + download["path"] = params["filePath"] + self._spawn(self._hashed(download, Path(params["filePath"]))) + elif params["state"] == "canceled": + download["ended_at"] = _now() + + async def _hashed(self, download: dict, path: Path) -> None: + size, sha256 = await asyncio.to_thread(_size_and_sha256, path) + download.update(size=size, sha256=sha256, ended_at=_now()) + + def _spawn(self, work: Awaitable) -> None: + task = asyncio.ensure_future(work) + self._tasks.add(task) + task.add_done_callback(self._finished) + + def _finished(self, task: asyncio.Future) -> None: + self._tasks.discard(task) + if not task.cancelled() and (error := task.exception()) is not None: + logger.warning(f"the browser's control connection: {type(error).__name__}: {error}") + + +def _size_and_sha256(path: Path) -> tuple[int, str]: + digest = hashlib.sha256() + with path.open("rb") as file: + while chunk := file.read(HASH_CHUNK_BYTES): + digest.update(chunk) + return path.stat().st_size, digest.hexdigest() + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + class Window(BaseModel): width: int = Field(1280, gt=0) height: int = Field(800, gt=0) @@ -160,6 +381,8 @@ class BrowserSettings(BaseModel): #: Whether a page asking for the location, or to show notifications, is denied without a prompt or asks. geolocation: Literal["deny", "ask"] = "deny" notifications: Literal["deny", "ask"] = "deny" + #: Who answers a page's JavaScript dialogs: the agent, or agentenv accepting or dismissing each. + dialogs: Literal["agent", "accept", "dismiss"] = "agent" @field_validator("start_urls") @classmethod @@ -186,10 +409,13 @@ class WebsiteBrowserEnvironment(ComputerAppEnvironment): def __init__(self) -> None: #: Its CDP connection, over the pipe its server launches it with. self.devtools = DevToolsPipe() - self._start_tabs: Optional[asyncio.Future] = None + self._following: Optional[asyncio.Future] = None super().__init__() for folder in (UPLOADS, DOWNLOADS): (Path.home() / folder).mkdir(exist_ok=True) + #: agentenv's control of it over that connection, and the record it keeps. + self.control = BrowserControl(self.devtools, self.settings.dialogs) + self.devtools.on_close(self._pipe_closed) @classmethod def declaration(cls) -> AppDeclaration: @@ -250,19 +476,28 @@ def _command_line_tabs(self) -> list[str]: async def _launch_fds(self) -> dict[int, int]: """Before each launch, an empty cache and the deploy's prompt answers in its profile; and a new debugging pipe: - the browser's ends as its fds 3 and 4, this server's ends its CDP connection, over which the start tabs its - command line can't open are opened.""" + the browser's ends as its fds 3 and 4, this server's ends its CDP connection, which follows the browser and + then opens the start tabs its command line can't.""" shutil.rmtree(self.cache_dir(), ignore_errors=True) self._write_prompt_answers() to_browser_read, to_browser_write = os.pipe() from_browser_read, from_browser_write = os.pipe() await self.devtools.connect(to_browser_write, from_browser_read) - rest = self.settings.start_urls[len(self._command_line_tabs()):] - self._start_tabs = asyncio.ensure_future(self._open_tabs(rest)) if rest else None - if self._start_tabs is not None: - self._start_tabs.add_done_callback(lambda done: done.cancelled() or done.exception()) + self._following = asyncio.ensure_future(self._follow(self.settings.start_urls[len(self._command_line_tabs()):])) + self._following.add_done_callback(lambda done: done.cancelled() or done.exception()) return {PIPE_IN: to_browser_read, PIPE_OUT: from_browser_write} + def _pipe_closed(self) -> None: + """The browser closed its pipe: its tabs closed by its exit when it ended on its own, else by its server's + stop.""" + self.control.ended(EXIT if self._computer_app.exited() else STOP) + + async def _follow(self, urls: list[str]) -> None: + """Its control attached, then each of ``urls`` in a tab behind the active one, in order.""" + await self.control.attach() + for url in urls: + await self.devtools.call("Target.createTarget", {"url": url, "background": True}) + def _write_prompt_answers(self) -> None: """Its profile's preferences, which it reads as it starts, denying location and notification prompts or leaving them to ask, as the deploy's settings say.""" @@ -281,11 +516,6 @@ def _write_prompt_answers(self) -> None: preferences.parent.mkdir(parents=True, exist_ok=True) preferences.write_text(json.dumps(current), encoding="utf-8") - async def _open_tabs(self, urls: list[str]) -> None: - """Each of ``urls`` in a tab behind the active one, in order.""" - for url in urls: - await self.devtools.call("Target.createTarget", {"url": url, "background": True}) - def _relaunch_on_start(self) -> bool: return True @@ -306,22 +536,30 @@ def write_policies(self) -> None: @ready_check async def devtools_answers(self) -> bool: - """It answers over its pipe and has opened its start tabs.""" + """It answers over its pipe, its control follows it and it has opened its start tabs.""" try: await self.devtools.call("Browser.getVersion", timeout_s=self.devtools_timeout_s) except (ConnectionError, DevToolsError, asyncio.TimeoutError): return False - tabs = self._start_tabs - return tabs is None or (tabs.done() and not tabs.cancelled() and tabs.exception() is None) - - def declared_extensions(self) -> list[EnvironmentExtension]: - params: dict[str, Any] = {"chromium": self.version, "clock": "real"} - if _alone(): - params["window"] = self.settings.window.model_dump() - browser = EnvironmentExtension( - uri=BROWSER, params=params, - description="A website browser agentenv runs: its build, the clock its pages see and, alone, its window.") - return [*super().declared_extensions(), browser] + following = self._following + return following is not None and following.done() and not following.cancelled() and \ + following.exception() is None + + @extension(BROWSER, op="record", method="GET", description=( + "A website browser agentenv runs: its build, the clock its pages see and, alone, its window. record returns " + "what agentenv's own connection to it saw: its open tabs, each with whether it's active, the tabs closed and " + "who closed them, its downloads with their size and sha256, its JavaScript dialogs and who answered each, and " + "each launch, stop and exit of it since the computer booted, with its cause.")) + async def browser_record(self) -> dict: + return {**await self.control.record(), "launches": self._computer_app.launches()} + + def extension_params(self, uri: str) -> dict: + if uri == BROWSER: + params: dict[str, Any] = {"chromium": self.version, "clock": "real"} + if _alone(): + params["window"] = self.settings.window.model_dump() + return params + return super().extension_params(uri) @agent_skill def browser_skill(self) -> tuple[str, str]: diff --git a/packages/agentenv-protocol/tests/test_computer_dataplane.py b/packages/agentenv-protocol/tests/test_computer_dataplane.py index f6a565fc8..b48992d2c 100644 --- a/packages/agentenv-protocol/tests/test_computer_dataplane.py +++ b/packages/agentenv-protocol/tests/test_computer_dataplane.py @@ -16,7 +16,9 @@ from computer_fake import FakeDriver, http_client, rpc # noqa: E402 from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI # noqa: E402 +from agentenv_protocol.computer import dataplane # noqa: E402 from agentenv_protocol.computer.dataplane import DEFAULT_DATA_PATHS # noqa: E402 +from agentenv_protocol.transfers import TransferError # noqa: E402 @pytest.fixture @@ -168,6 +170,47 @@ async def test_data_get_uploads_under_a_write_namespace_grant(home): "metadata": {DATA_OBJECTS_EXTENSION_URI: {"path": "notes.md"}}}] +@pytest.mark.asyncio +async def test_data_get_leaves_out_a_file_removed_after_its_listed(home): + (home / "app").mkdir() + (home / "app" / "kept.txt").write_text("kept") + gone = home / "app" / "gone.pma" + gone.write_text("metrics") + listed = dataplane._data_files + + def listed_then_removed(*args): + found = listed(*args) + gone.unlink() + return found + + with patch("agentenv_protocol.computer.dataplane._data_files", side_effect=listed_then_removed): + async with http_client(FakeDriver()) as http: + parts = (await http.post("/agentenv", json=rpc("data/get", {"paths": ["app"]}))).json()["result"]["parts"] + assert [p["metadata"][DATA_OBJECTS_EXTENSION_URI]["path"] for p in parts] == ["app/kept.txt"] + + +@pytest.mark.asyncio +async def test_data_get_under_a_grant_leaves_out_a_file_removed_before_its_uploaded(home): + (home / "kept.txt").write_text("kept") + grant = {"root_path": "runs/1", "expires_at": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(), + "max_objects": 5, "max_object_bytes": 100, "max_total_bytes": 500, + "write": {"kind": "http-post-policy", "url": "https://store.test/", "fields": {}, + "path_field": "key", "file_field": "file"}} + + async def upload(relative, source): + if relative == "gone.txt": + raise TransferError("invalid_transfer", "The transfer source is unavailable.") + + with patch("agentenv_protocol.computer.dataplane._data_files", + return_value=[("gone.txt", home / "gone.txt"), ("kept.txt", home / "kept.txt")]), \ + patch("agentenv_protocol.computer.dataplane.NamespaceUploader") as uploader: + uploader.return_value.upload = upload + async with http_client(FakeDriver()) as http: + parts = (await http.post("/agentenv", json=rpc("data/get", {"write_namespace": grant, "paths": ["kept.txt"]})) + ).json()["result"]["parts"] + assert [p["metadata"][DATA_OBJECTS_EXTENSION_URI]["path"] for p in parts] == ["kept.txt"] + + @pytest.mark.asyncio async def test_a_failed_write_leaves_no_temporary_file(home): with patch("agentenv_protocol.computer._linux.os.replace", side_effect=OSError("disk full")): diff --git a/packages/agentenv-protocol/tests/test_website_browser.py b/packages/agentenv-protocol/tests/test_website_browser.py index a4470a2ad..d21dd495c 100644 --- a/packages/agentenv-protocol/tests/test_website_browser.py +++ b/packages/agentenv-protocol/tests/test_website_browser.py @@ -7,8 +7,10 @@ from __future__ import annotations import asyncio +import hashlib import json import os +import signal import sys import pytest @@ -18,7 +20,7 @@ from agentenv_protocol.computer import app_runtime from agentenv_protocol.computer.platform import Platform from agentenv_protocol.website_browser import (ALONE_BYPASS, ALONE_ENV, BROWSER, COMPUTER_BYPASS, OPEN_WEB_ENV, - POLICIES, SITE_PROXY_ENV, SKILL, DevToolsError, + POLICIES, SITE_PROXY_ENV, SKILL, BrowserControl, DevToolsError, WebsiteBrowserEnvironment) PROXY = "http://127.0.0.1:3128" @@ -36,6 +38,11 @@ message = json.loads(raw) if message["method"] == "Browser.getVersion": answer = {"result": {"product": "Fake/1"}} + elif message["method"] in ("Target.setDiscoverTargets", "Target.setAutoAttach", "Browser.setDownloadBehavior"): + answer = {"result": {}} + if message["method"] == "Target.setDiscoverTargets": + created = {"targetInfo": {"targetId": "start", "type": "page", "url": "about:blank", "title": ""}} + os.write(4, json.dumps({"method": "Target.targetCreated", "params": created}).encode() + b"\\0") elif message["method"] == "Target.createTarget": with open("tabs.txt", "a") as tabs: tabs.write(("window " if message["params"].get("newWindow") else "") + message["params"]["url"] + "\\n") @@ -269,11 +276,11 @@ async def test_each_launch_gets_a_new_pipe_its_server_talks_cdp_over_and_its_rea version = await browser.devtools.call("Browser.getVersion") with pytest.raises(DevToolsError, match="unknown method"): await browser.devtools.call("Page.navigate", {"url": "about:blank"}) - launched = await browser.devtools_answers() + launched = await _answering(browser) await runtime.stop_app() stopped = await browser.devtools_answers() await runtime.launch_app() - relaunched = await browser.devtools_answers() + relaunched = await _answering(browser) await runtime.stop_app() browser.devtools.close() @@ -294,7 +301,7 @@ async def test_a_launch_while_it_runs_opens_a_window_over_its_pipe_and_starts_no (browser_home / ".cache/chromium").mkdir(parents=True, exist_ok=True) (browser_home / ".cache/chromium/kept").touch() second = await runtime.launch_app() - answers = await browser.devtools_answers() + answers = await _answering(browser) await runtime.stop_app() browser.devtools.close() @@ -391,13 +398,13 @@ def test_its_policies_are_agentenvs_with_its_own_over_them_where_its_build_reads def test_its_card_declares_browser_v1_with_its_build_and_clock_and_alone_its_window(monkeypatch): - in_computer = next(e for e in Browser().declared_extensions() if e.uri == BROWSER) + in_computer = Browser().extension_params(BROWSER) monkeypatch.setenv(ALONE_ENV, "1") monkeypatch.setenv(SITE_PROXY_ENV, PROXY) - alone = next(e for e in Browser().declared_extensions() if e.uri == BROWSER) + alone = Browser().extension_params(BROWSER) - assert in_computer.params == {"chromium": "156.0.8078.4", "clock": "real"} - assert alone.params == {"chromium": "156.0.8078.4", "clock": "real", "window": {"width": 1280, "height": 800}} + assert in_computer == {"chromium": "156.0.8078.4", "clock": "real"} + assert alone == {"chromium": "156.0.8078.4", "clock": "real", "window": {"width": 1280, "height": 800}} @pytest.mark.asyncio @@ -420,3 +427,163 @@ def test_an_app_that_isnt_a_browser_launches_and_declares_as_before(): assert stored["os"]["linux"]["launch"] == {"at": "boot", "command": ["mail"], "window": {"class": "Mail"}, "ready": {"window": True, "timeout_s": 60.0}} + +#: How long a test waits on the fake browser, and how often it looks. +WAIT_S, POLL_S = 5.0, 0.05 + + +async def _answering(browser) -> bool: + """Whether ``browser``'s ready check says it answers, within ``WAIT_S``.""" + for _ in range(int(WAIT_S / POLL_S)): + if await browser.devtools_answers(): + return True + await asyncio.sleep(POLL_S) + return False + + +class StubDevTools: + """A browser connection that answers each call from ``answers`` by method, records it, and hands the control the + events a test sends.""" + + def __init__(self, answers=None): + self.calls, self.answers, self.listener = [], answers or {}, None + + def listen(self, listener): + self.listener = listener + + async def call(self, method, params=None, *, session_id=None, timeout_s=None): + self.calls.append((method, params or {}, session_id)) + answer = self.answers.get(method, {}) + return answer(params, session_id) if callable(answer) else answer + + def send(self, method, params, session=None): + self.listener({"method": method, "params": params, **({"sessionId": session} if session else {})}) + + +def _page(target, url="about:blank", title=""): + return {"targetInfo": {"targetId": target, "type": "page", "url": url, "title": title}} + + +@pytest.mark.asyncio +async def test_its_control_records_tabs_in_opened_order_and_who_closed_each_and_leaves_out_its_hidden_pages(tmp_path): + devtools = StubDevTools({"Target.createTarget": {"targetId": "hidden"}, + "Target.attachToTarget": {"sessionId": "s-hidden"}}) + control = BrowserControl(devtools, "agent") + await control.attach() + + for target in ("a", "b", "c"): + devtools.send("Target.targetCreated", _page(target)) + devtools.send("Target.targetInfoChanged", _page("a", "https://slack.com/", "Slack")) + devtools.send("Target.targetDestroyed", {"targetId": "b"}) + await control.close_tab("c") + devtools.send("Target.targetDestroyed", {"targetId": "c"}) + async with control.hidden_page() as session: + devtools.send("Target.targetCreated", _page("hidden")) + record = await control.record() + await control.attach() + + assert [tab["id"] for tab in record["tabs"]] == ["a"] and record["tabs"][0]["title"] == "Slack" + assert [(tab["id"], tab["closed_by"]) for tab in record["closed_tabs"]] == [("b", "unknown"), ("c", "agentenv")] + assert record["since"] == control.since + assert session == "s-hidden" and ("Target.closeTarget", {"targetId": "hidden"}, None) in devtools.calls + assert [(tab["id"], tab["closed_by"]) for tab in control.closed_tabs][-1] == ("a", "stop") + assert [call[0] for call in devtools.calls[:3]] == ["Target.setDiscoverTargets", "Target.setAutoAttach", + "Browser.setDownloadBehavior"] + assert devtools.calls[2][1] == {"behavior": "default", "eventsEnabled": True} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("dialogs, kind, sent, answer, by", [ + ("accept", "prompt", {"accept": True, "promptText": "yes"}, "accepted", "agentenv"), + ("dismiss", "prompt", {"accept": False, "promptText": "yes"}, "dismissed", "agentenv"), + ("dismiss", "beforeunload", {"accept": True, "promptText": "yes"}, "accepted", "agentenv"), + ("agent", "prompt", None, "dismissed", "agent"), +], ids=["accept", "dismiss", "dismiss-leaves", "agent"]) +async def test_its_control_answers_a_dialog_on_its_pages_session_as_its_setting_says_and_records_who_did( + tmp_path, dialogs, kind, sent, answer, by): + devtools = StubDevTools() + control = BrowserControl(devtools, dialogs) + devtools.send("Target.targetCreated", _page("a")) + devtools.send("Target.attachedToTarget", {"sessionId": "s-a", **_page("a")}) + + devtools.send("Page.javascriptDialogOpening", {"type": kind, "message": "Sure?", "url": "https://a.test/", + "defaultPrompt": "yes"}, "s-a") + await asyncio.sleep(0) + devtools.send("Page.javascriptDialogClosed", {"result": answer == "accepted"}, "s-a") + + assert ("Page.enable", {}, "s-a") in devtools.calls + handled = [(params, session) for method, params, session in devtools.calls + if method == "Page.handleJavaScriptDialog"] + assert handled == ([(sent, "s-a")] if sent else []) + (dialog,) = control.dialog_log + assert (dialog["tab"], dialog["type"], dialog["message"], dialog["answer"], dialog["by"]) == ( + "a", kind, "Sure?", answer, by) + + +@pytest.mark.asyncio +async def test_its_control_records_a_completed_download_with_its_size_and_sha256(tmp_path): + devtools = StubDevTools() + control = BrowserControl(devtools, "agent") + (tmp_path / "report.pdf").write_bytes(b"%PDF-1.7 report") + + devtools.send("Browser.downloadWillBegin", {"guid": "g", "url": "https://a.test/report.pdf", + "suggestedFilename": "report.pdf"}) + devtools.send("Browser.downloadProgress", {"guid": "g", "state": "inProgress", "receivedBytes": 3}) + in_progress = dict(control.downloads["g"]) + devtools.send("Browser.downloadProgress", {"guid": "g", "state": "completed", + "filePath": str(tmp_path / "report.pdf")}) + for _ in range(int(WAIT_S / POLL_S)): + if "sha256" in control.downloads["g"]: + break + await asyncio.sleep(POLL_S) + + done = control.downloads["g"] + assert in_progress["state"] == "inProgress" and "sha256" not in in_progress + assert (done["state"], done["size"], done["path"]) == ("completed", 15, str(tmp_path / "report.pdf")) + assert done["sha256"] == hashlib.sha256(b"%PDF-1.7 report").hexdigest() + + +@pytest.mark.asyncio +async def test_its_control_says_the_tab_whose_page_has_focus_is_active_else_the_first_visible_one(tmp_path): + states = {"s-a": ["hidden", False], "s-b": ["visible", False], "s-c": ["visible", True]} + devtools = StubDevTools({"Runtime.evaluate": lambda params, session: {"result": {"value": states[session]}}}) + control = BrowserControl(devtools, "agent") + for target in ("a", "b", "c"): + devtools.send("Target.targetCreated", _page(target)) + devtools.send("Target.attachedToTarget", {"sessionId": f"s-{target}", **_page(target)}) + + focused = await control.active_tab() + states["s-c"] = ["hidden", False] + visible = await control.active_tab() + + assert (focused, visible) == ("c", "b") + + +@pytest.mark.asyncio +async def test_its_connection_hears_events_its_server_logs_each_launch_and_an_exit_it_didnt_ask_for(monkeypatch, + tmp_path): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(app_runtime, "LAUNCHED", tmp_path / "launched") + browser = FakeBrowser() + runtime = browser._computer_app + + first = await runtime.launch_app("boot") + await _answering(browser) + record = await browser.browser_record() + await runtime.stop_app() + second = await runtime.launch_app("data/reset") + await _answering(browser) + os.killpg(second, signal.SIGKILL) + for _ in range(int(WAIT_S / POLL_S)): + if any(entry["event"] == "exit" for entry in runtime.launches()): + break + await asyncio.sleep(POLL_S) + after_exit = await browser.browser_record() + browser.devtools.close() + + assert [tab["id"] for tab in record["tabs"]] == ["start"] + assert after_exit["tabs"] == [] and [(tab["id"], tab["closed_by"]) for tab in after_exit["closed_tabs"]] == [ + ("start", "stop"), ("start", "exit")] + assert [(entry["event"], entry.get("cause"), entry["pid"]) for entry in runtime.launches()] == [ + ("launch", "boot", first), ("stop", None, first), ("launch", "data/reset", second), ("exit", None, second)] + assert runtime.launches()[-1]["signal"] == "SIGKILL" diff --git a/src/agent_env/browsers/README.md b/src/agent_env/browsers/README.md index 18e3e0901..43bb3783d 100644 --- a/src/agent_env/browsers/README.md +++ b/src/agent_env/browsers/README.md @@ -8,7 +8,7 @@ build Playwright 1.64.1 pins, fetched by sha256 when a computer env that lists i flags, and writes managed policies to `/etc/opt/chrome_for_testing/policies/managed/` for unattended use. It runs without Chromium's sandbox, which a container can't give it. Its env.toml spells out its settings and their defaults under `[settings]`: `start_urls`, opened at boot, the first tab active; `window`, alone only; `locale`; `time_zone`; and -`geolocation` and `notifications`, `deny` or `ask`. A computer env sets its own beside it +`geolocation` and `notifications`, `deny` or `ask`; and `dialogs`, `agent`, `accept` or `dismiss`. A computer env sets its own beside it (`apps = [{ env = "chromium", settings = { start_urls = ["https://example.test/"] } }]`), and a deploy overrides any of them (`env_params.app_settings = {chromium = {start_urls = [...]}}`). Its data is its profile, `~/.config/chromium`, which the image has none of, so each deploy and each reset starts with a clean one. Name it from another bundle as `chromium`, diff --git a/src/agent_env/browsers/envs/chromium/env.toml b/src/agent_env/browsers/envs/chromium/env.toml index 520321950..8a3a20ae1 100644 --- a/src/agent_env/browsers/envs/chromium/env.toml +++ b/src/agent_env/browsers/envs/chromium/env.toml @@ -10,3 +10,4 @@ locale = "en-US" time_zone = "UTC" geolocation = "deny" notifications = "deny" +dialogs = "agent" diff --git a/tst/integration/computer/chromium_app_test.py b/tst/integration/computer/chromium_app_test.py index f0e121f60..4ca5b9701 100644 --- a/tst/integration/computer/chromium_app_test.py +++ b/tst/integration/computer/chromium_app_test.py @@ -5,22 +5,25 @@ website browser it runs Chrome itself with agentenv's flags on its debugging pipe, keeps its shared memory in files on Docker's 64 MiB /dev/shm, answers location prompts and leaves notification ones to ask as the deploy says, and its menu launcher, run while it runs, opens another window of it with no second copy, its cache kept and it still ready; and the -desktop that gives it French defaults deploys with them, a deploy's time zone over them, each recorded. Needs Docker and -the local registry on localhost:5000.""" +desktop that gives it French defaults deploys with them, a deploy's time zone over them, each recorded; and the record +its control connection keeps lists a tab an agent opened and closed, a download with its size and sha256, a confirm() +accepted for it, and each launch, exit and stop with its cause, across a kill and a restart of its server. Needs Docker +and the local registry on localhost:5000.""" import asyncio +import hashlib import json import platform import re +import httpx import pytest -from mcp import ClientSession -from mcp.client.streamable_http import streamablehttp_client from agentenv_protocol import client as protocol_v1 from agent_env.env import Env from agentenv_protocol.types import WELL_KNOWN_PATH +from agentenv_protocol.website_browser import BROWSER from tst.integration.computer.website_browser_test import _app, _browser_process, _keys, _ready from agent_env.env.envs.computer.utils import inside from agent_env.store.routing import namespace_routing @@ -29,6 +32,8 @@ pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] +#: A page server's time to listen, the record's wait and poll, and the Control+Tab presses for one tab change. +SERVE_START_S, RECORD_WAIT_S, RECORD_POLL_S, TAB_PRESSES = 3, 60, 1, 3 ARCH = {"x86_64": "amd64", "amd64": "amd64", "aarch64": "arm64", "arm64": "arm64"}[platform.machine().lower()] ALPHA = "data:text/html,Alpha page

Alpha

" BRAVO = "data:text/html,Bravo page

Bravo

" @@ -54,6 +59,36 @@ def log_message(self, *args): threading.Thread(target=server.serve_forever, daemon=True).start() time.sleep(float(sys.argv[1])) """ +RECORDED = "http://127.0.0.1:8767/" +# Serves, on loopback for the seconds it's given, a page that asks to confirm and then downloads report.bin, whose +# content differs at each request: the nth is _report(n). +SERVE_RECORDED = """import http.server, sys, threading, time +served = [] +class Page(http.server.BaseHTTPRequestHandler): + def do_GET(self): + if self.path == "/report.bin": + served.append(1) + body, kind = b"agentenv report %d\\n" % len(served) * 64, "application/octet-stream" + else: + body = (b"asking") + kind = "text/html" + self.send_response(200) + self.send_header("Content-Type", kind) + self.send_header("Content-Length", str(len(body))) + if self.path == "/report.bin": + self.send_header("Content-Disposition", "attachment; filename=report.bin") + self.end_headers() + self.wfile.write(body) + def log_message(self, *args): + pass +server = http.server.ThreadingHTTPServer(("127.0.0.1", 8767), Page) +threading.Thread(target=server.serve_forever, daemon=True).start() +time.sleep(float(sys.argv[1])) +""" +# A page that asks to confirm as soon as it loads, and says in its title what it was told. +AGENT_DIALOG = ("data:text/html,asking") # Each window the window manager manages, its class and its title. WINDOWS = ('for w in $(xprop -root _NET_CLIENT_LIST | sed "s/.*# //; s/,//g"); do ' 'xprop -id "$w" WM_CLASS _NET_WM_NAME; done') @@ -74,7 +109,7 @@ async def test_a_deploy_opens_its_start_urls_unprompted_and_a_reset_clears_its_p context = TaskStepContext(deployed_envs=[deployed]) try: opened = await _chromium_windows(deployed, "Alpha page") - keyed, tabbed = await _tab_to(deployed, "Bravo page") + tabbed = await _tab_to(deployed, "Bravo page") policies = await inside.run( deployed, ["cat", "/etc/opt/chrome_for_testing/policies/managed/agentenv.json"], timeout_s=30) await inside.run(deployed, ["sh", "-c", "touch ~/.config/chromium/marker"], timeout_s=30) @@ -85,7 +120,7 @@ async def test_a_deploy_opens_its_start_urls_unprompted_and_a_reset_clears_its_p report = await teardown_run(context) assert opened == ["Alpha page - Google Chrome for Testing"] - assert not keyed.isError and tabbed == ["Bravo page - Google Chrome for Testing"] + assert tabbed == ["Bravo page - Google Chrome for Testing"] assert {"BrowserSignin": 0, "CommandLineFlagSecurityWarningsEnabled": False}.items() <= json.loads( policies["stdout"]).items() assert marker["exit_code"] != 0 and reopened == ["Alpha page - Google Chrome for Testing"] @@ -103,7 +138,7 @@ async def test_it_runs_chrome_itself_on_its_pipe_answers_prompts_as_the_deploy_s browser = await _browser_process(deployed) serving = asyncio.ensure_future(inside.run(deployed, ["python3", "-c", SERVE_PERMISSIONS, "60"], timeout_s=90)) - await asyncio.sleep(3) + await asyncio.sleep(SERVE_START_S) await _keys(deployed, ["F5"]) permissions = await _chromium_windows(deployed, "denied") await serving @@ -149,20 +184,112 @@ async def test_a_desktop_that_gives_it_defaults_deploys_with_them_and_a_deploys_ assert not report.failed -async def _tab_to(deployed, title: str): - """Control+Tab pressed as an agent presses it, again while the window still shows the tab before, which a press - can reach before the next tab opens; the last press's result, and the window titles once one starts with - ``title``.""" - async with streamablehttp_client(f"{deployed.environment_url}/mcp") as (read, write, _): - async with ClientSession(read, write) as session: - await session.initialize() - for attempt in range(3): - keyed = await session.call_tool("computer_key", {"keys": ["Control", "Tab"]}) - try: - return keyed, await _chromium_windows(deployed, title, seconds=20) - except TimeoutError: - if attempt == 2: - raise +@pytest.mark.asyncio +async def test_its_control_records_tabs_a_download_a_dialog_and_each_launch_with_its_cause(chromium_desktop): + settings = {"start_urls": [RECORDED], "dialogs": "accept"} + with namespace_routing(): + deployed = await Env.get(chromium_desktop.id).deploy(env_params={"app_settings": {"chromium": settings}}) + context = TaskStepContext(deployed_envs=[deployed]) + try: + card = await _app(deployed, WELL_KNOWN_PATH, "chromium") + serving = asyncio.ensure_future(inside.run(deployed, ["python3", "-c", SERVE_RECORDED, "90"], + timeout_s=120)) + await asyncio.sleep(SERVE_START_S) + await _keys(deployed, ["F5"]) + accepted = await _chromium_windows(deployed, "accepted") + await _keys(deployed, ["Control", "t"], ["Control", "w"]) + first = await _record(deployed, card, lambda r: _hashed(r) and r["closed_tabs"]) + await _keys(deployed, ["F5"]) + recorded = await _record(deployed, card, lambda r: len(_hashed(r)) > len(_hashed(first))) + await serving + browser = await _browser_process(deployed) + await inside.run(deployed, ["kill", "-9", str(browser["pid"])], timeout_s=30) + exited = await _record(deployed, card, lambda r: r["launches"][-1]["event"] == "exit") + await inside.run(deployed, ["sh", "-c", LAUNCHER], timeout_s=60) + relaunched = await _ready(deployed, card, "chromium") + await inside.run(deployed, ["pkill", "-f", "computer.serve_app chromium.py"], timeout_s=30) + restarted = await _record(deployed, card, lambda r: r["launches"][-1].get("cause") == "server_restart") + finally: + report = await teardown_run(context) + + assert accepted[0].startswith("accepted") + # Chrome reloads its error page for the start tab once the page's server is up, and F5 reloads it too, so the page + # may run more than once; each run's dialog and download must be recorded alike. + assert recorded["dialogs"] and {(d["type"], d["message"], d["answer"], d["by"]) for d in recorded["dialogs"]} == { + ("confirm", "Proceed?", "accepted", "agentenv")}, recorded + # Every download is named report.bin, so Chrome saves the later ones as report (1).bin and on: each must be hashed + # at its own path, as the content served for it. + downloads = recorded["downloads"] + assert len(downloads) >= 2 and {(d["filename"], d["state"]) for d in downloads} == {("report.bin", "completed")} + assert len({d["path"] for d in downloads}) == len(downloads), recorded + assert sorted(d["sha256"] for d in downloads) == sorted( + hashlib.sha256(_report(n)).hexdigest() for n in range(1, len(downloads) + 1)), recorded + assert [tab["closed_by"] for tab in recorded["closed_tabs"]] == ["unknown"], recorded + assert [(tab["url"], tab["active"]) for tab in recorded["tabs"]] == [(RECORDED, True)], recorded + assert [entry["event"] for entry in exited["launches"]] == ["launch", "exit"], exited + assert exited["launches"][1]["signal"] == "SIGKILL" and recorded["since"], exited + assert exited["launches"][0]["cause"] == "boot" and relaunched["ready"] is True, relaunched + events = [(entry["event"], entry.get("cause")) for entry in restarted["launches"]] + assert events[:3] == [("launch", "boot"), ("exit", None), ("launch", "launch")] + assert events[3:] in ([("stop", None), ("launch", "server_restart")], [("launch", "server_restart")]), events + assert not report.failed + + +@pytest.mark.asyncio +async def test_with_dialogs_left_to_the_agent_chromium_shows_one_and_records_the_agents_answer(chromium_desktop): + with namespace_routing(): + deployed = await Env.get(chromium_desktop.id).deploy( + env_params={"app_settings": {"chromium": {"start_urls": [AGENT_DIALOG]}}}) + context = TaskStepContext(deployed_envs=[deployed]) + try: + card = await _app(deployed, WELL_KNOWN_PATH, "chromium") + pending = await _record(deployed, card, lambda r: r["dialogs"]) + await _keys(deployed, ["Enter"]) + answered_title = await _chromium_windows(deployed, "accepted") + answered = await _record(deployed, card, lambda r: r["dialogs"] and r["dialogs"][0]["answer"]) + finally: + report = await teardown_run(context) + + assert [(d["type"], d["answer"], d["by"]) for d in pending["dialogs"]] == [("confirm", None, None)], pending + assert answered_title[0].startswith("accepted") + assert [(d["type"], d["message"], d["answer"], d["by"]) for d in answered["dialogs"]] == [ + ("confirm", "Proceed?", "accepted", "agent")], answered + assert not report.failed + + +def _report(n: int) -> bytes: + """The content SERVE_RECORDED serves at its nth request for report.bin.""" + return b"agentenv report %d\n" % n * 64 + + +def _hashed(record: dict) -> list[dict]: + return [download for download in record["downloads"] if "sha256" in download] + + +async def _record(deployed, card: dict, until) -> dict: + """The browser's record, once ``until`` says it has what's waited for, or as it last was within ``RECORD_WAIT_S``.""" + path = protocol_v1.find_extension_method(card, BROWSER, "record")["endpoint"] + for _ in range(RECORD_WAIT_S // RECORD_POLL_S): + try: + record = await _app(deployed, path, "chromium") + if until(record): + return record + except httpx.HTTPError: + pass + await asyncio.sleep(RECORD_POLL_S) + return record + + +async def _tab_to(deployed, title: str) -> list[str]: + """Control+Tab pressed as an agent presses it, and again while the window still shows the tab before, which a + press can reach before the next tab opens; the window titles once one starts with ``title``.""" + for attempt in range(TAB_PRESSES): + await _keys(deployed, ["Control", "Tab"]) + try: + return await _chromium_windows(deployed, title, seconds=20) + except TimeoutError: + if attempt == TAB_PRESSES - 1: + raise async def _chromium_windows(deployed, title: str, seconds: int = 60, count: int = 1) -> list[str]: diff --git a/tst/integration/computer/website_browser_test.py b/tst/integration/computer/website_browser_test.py index 9f22a7c58..2155b86d7 100644 --- a/tst/integration/computer/website_browser_test.py +++ b/tst/integration/computer/website_browser_test.py @@ -74,6 +74,8 @@ def log_message(self, *args): argv = open(f"/proc/{pid}/cmdline", "rb").read().replace(b"\\0", b" ").decode().split() print(json.dumps({"pid": pid, "argv": argv, "listening": len(listening & sockets)})) """ +#: How long apart the key chords a test presses into Chromium are. +KEY_GAP_S = 2.0 # Each window the window manager manages, its class and its title. WINDOWS = ('for w in $(xprop -root _NET_CLIENT_LIST | sed "s/.*# //; s/,//g"); do ' 'xprop -id "$w" WM_CLASS _NET_WM_NAME; done') @@ -127,7 +129,8 @@ async def test_chromium_runs_with_agentenvs_flags_on_its_pipe_and_a_reset_relaun assert zone.startswith("Europe/Paris") and loopback.startswith("Loopback page") assert {"BrowserSignin": 0, "DownloadDirectory": "${user_home}/Downloads"}.items() <= json.loads( policies["stdout"]).items() - assert protocol_v1.extension_params(card, BROWSER) == {"chromium": "156.0.8078.4", "clock": "real"} + params = protocol_v1.extension_params(card, BROWSER) + assert (params["chromium"], params["clock"], list(params["methods"])) == ("156.0.8078.4", "real", ["record"]) assert readiness["ready"] is True, readiness assert relaunched["pid"] != browser["pid"] and relaunched["listening"] == 0 assert relaunched_readiness["ready"] is True, relaunched_readiness @@ -147,13 +150,29 @@ async def _window_titled(deployed, title: str) -> str: async def _keys(deployed, *chords: list[str]) -> None: - """Each chord pressed through the computer's own tools, as an agent presses it.""" + """Each chord pressed through the computer's own tools, as an agent presses it, ``KEY_GAP_S`` apart, after a click + into the middle of Chromium's window, which a window opened with no input may need before it has the keyboard.""" + x, y = await _window_center(deployed) async with streamablehttp_client(f"{deployed.environment_url}/mcp") as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() + await session.call_tool("computer_screenshot", {}) + clicked = await session.call_tool("computer_click", {"x": x, "y": y}) + assert not clicked.isError, clicked for keys in chords: pressed = await session.call_tool("computer_key", {"keys": keys}) assert not pressed.isError, pressed + await asyncio.sleep(KEY_GAP_S) + + +async def _window_center(deployed) -> tuple[int, int]: + """The middle of Chromium's largest window, in screen pixels.""" + tree = (await inside.run(deployed, ["xwininfo", "-root", "-tree"], timeout_s=30))["stdout"] + windows = [(int(w) * int(h), int(x) + int(w) // 2, int(y) + int(h) // 2) for w, h, x, y in re.findall( + r'\("[^"]*" "Chromium"\)\s+(\d+)x(\d+)\+-?\d+\+-?\d+\s+\+(-?\d+)\+(-?\d+)', tree)] + assert windows, tree + _, x, y = max(windows) + return x, y async def _browser_process(deployed, other_than: int | None = None) -> dict: