diff --git a/AGENTS.md b/AGENTS.md index c27ce0deb..e486c332c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -124,7 +124,7 @@ and deserialized through a registry. |---|---| | `env/` | `Env` base class: a subclass declares `type`, implements `from_dict` and `async deploy(...) -> DeployedEnv`, and may override `reset`, `close` and `load_file_artifact_universe`; teardown closes each env a run deployed before terminating its sandboxes by id. Built-ins in `env/envs/`: `MCPServerEnv` (`mcp_server`), `MultiEnv` (`multi`, MCP servers plus websites and computers; its website browser is the one its `browser_envs` names, none for `[]`, else the run's when it has websites: an MCP server env such as `playwright-mcp` runs among the gateway's MCP servers, and a browser computer such as `chromium` on a sandbox of its own as one more computer child, sized by its `child_envs` entry, its tools under its name and `_`, routing the multi env's synthetic sites unless that entry's `site_routing` says, its network its own env's, as any computer child's; a website env deployed alone refuses a browser computer as the run's browser), `ServiceDBEnv` (`service_db`, PostgreSQL), `WebsiteEnv` (`website`, one site: a server image, `backend_image` in its env.toml, serving the pages and API, ready when its card's `website/v1` `ready` answers a `WebsiteReadiness` that says so, as `agentenv_protocol.website` (`website_readiness`, `env_readiness`) defines a website env's health for the deploy and `DeployedEnv.readiness()` alike, or, for a card without `ready`, when its `health` path answers 200, which is deprecated; its env.toml's `ready_timeout_s`, how long it may take from its server's container starting to ready, is that container's health check `start_period` and what the deploy waits, or its card's `ready` timeout if longer, and its `shm_mb` is the container's `shm_size`, each left to the gateway and Docker without them; the gateway container's `env/gateway/constants.py` mirrors the URI and op, a test keeping them equal; a separate frontend image is deprecated, warned of at put and optional in a bundle, and a stored `backend_docker_image_artifact` reads as the server image; a live one, `live = true`, is a real site whose server serves none of its pages, its `live`, `domains` and `allow_hosts` read at put from its server's `@website(...)` (`utils/card_naming.py` `WebsiteSource`) unless given, its `settings`' secrets only as references the gateway resolves as it writes the site's container (`WEBSITE_SETTINGS_ENV`); the gateway gives it no site route, a computer's site table leaves it out, a computer whose network is an allowlist is let reach its names (`MultiEnv.live_hosts`), a browser alone's forward port takes them straight to the real site (`site_proxy.py` `direct`), as the browser base's stand-in proxy takes any host, refusing nothing, each proxy logging each such connection as a JSON line, and a run that deploys one is marked live, `reproducible: false`), `GatewayEnv` (`gateway_server`), `ComputerDriverEnv` (`computer_driver`, `env/envs/computer/driver_env.py`: a computer driver's files and what its class declares for each platform it supports, read at put in a process of its own, `agentenv_protocol.computer._declare`, where the driver's own dependencies needn't be installed, with the `port` it serves on and its install by OS, `[install.]`, or for one distribution `[install."/"]`, which replaces its OS type's there, as its `known_failures` do (`env/envs/computer/utils/install.py`): files, apt `packages` with extra `sources`, a `python` requirements file for a venv, `downloads` by URL and sha256, `run` scripts, the services the base's init runs, and `ports`; it deploys only as a computer env's driver), `ComputerEnv` (`computer`, in `env/envs/computer/`: a base computer and a driver, matched at put by gate 1 before anything is stored, then composed by `composer.py` into an image for each form and arch on a builder of the put's sandbox provider, by default the env's (`utils/builder.py`; `--sandbox` on `env build`, `bundle register` and `run`): this machine's Docker for `local`, else a VM of the provider, which fetches the put's stored files by signed URL, takes the rest inline and pushes to the image store itself, so it builds only with stores a remote VM reaches; each arch goes to the first link that runs it natively, else one that emulates it (a VM registers QEMU; `--no-emulate` leaves it `skipped` instead, for a later put), and its pair records the provider that built it (`builder`); a run or register that reuses a computer env composes what it has left (`pending`), a run only the forms its sandbox's deploys boot (`booted_forms`: on each link, the first form it boots natively, by the deploy's own rule, `boot.first_native`; a compose of a disk adds its container), and a deploy composes first what it boots that no compose started, and a container it runs natively that was built but not probed, as one whose probe a setting that didn't resolve stopped, booting it only once ready (`ComputerEnv.compose_unstarted`); tagged `agentenv-computers:` in the image store by a key of the image's Dockerfile, build context and lock so any env with the same inputs reuses it; every layer's packages, requirements and downloads are resolved at put in a container of the base for each arch (`utils/packages.py`, `utils/resolve_in_base.py`), stored in the object store by sha256 under `agentenv-packages/`, and listed in a lock stored by a digest of its inputs, which any env with the same inputs reuses, and builds install only the lock's files, every install step `RUN --network=none`; each pair's status goes into its metadata, `composed`, the arches built are written as its computer, a `computer` artifact over one index, and `agent-env env build` resumes what a put left; gate 3 (`utils/probing.py`) then boots each native build on the put's sandbox provider and probes it at `full`, making it `ready` or `failed`, and a build no provider runs natively stays `unverified`. Its deploy, through the `computer` env provider (`providers/env_providers/env_computer_provider.py`), boots its first `ready` build on a sandbox provider that runs its arch natively (`native_platforms()`), with no gateway in front and the driver's card as the env card, its `/dev/shm` the most its apps' Linux installs ask for (`shm_mb`, in MiB, which its composed computer and the deploy record keep; `--shm-size` on a container, `AGENTENV_SHM_MB` read by a Linux disk's init, passed only to a provider whose `SHARED_MEMORY` says it gives that much, else the sandbox's default, which the deploy logs), then checks its `display`; `deploy_env` `env_params` take a `form`, `computer_settings` (deploy-phase values of the base's settings; a secret one, here or on the env, only as a bare `secret:`/`env:` reference) and `driver_settings` (the driver's `ComputerSettings`, such as its default coordinate space); a `network` that allows nothing runs it on local Docker's internal network behind a relay (`providers/sandbox_providers/local_docker/relay/`); `reset` and `close` have the driver end its recordings first (`computer-recording/v1` `end_all`, each uploading what's left and its manifest under the grant it started with), and `reset` then boots a fresh container in its place (`reset_container`), keeping the computer's address but not an agent's MCP session to the old driver; its `load_file_artifact_universe` writes files under the session user's home through the driver's `data/add`, `Downloads` by default, each file's bytes through a read grant when the store's grants reach it and its network allows everything, else inline (`env/envs/computer/utils/inside.py`)), `ComputerAppEnv` (`computer_app`, `env/envs/computer/app_env.py`: an app on a computer, a `ComputerAppEnvironment` class its env.toml's `module` names, read at put by the same `_declare`, which declares each OS with its own decorator (`@linux(...)`, or `@install(os_type, spec)` for any OS type; `@linux(distribution=...)` or `@install("/", spec)` for one distribution, which replaces its OS type's there) holding its install, its launch and, optionally, a skill and settings and build settings that add to the class's there, and install steps and ready checks for every OS or those their `os=` names; everything that reads its settings, launch or steps asks for the computer's OS type; deployed alone it is a one-app computer env, a `ComputerEnv` of the computer and driver its env.toml names, the composer installing it under `/opt/agentenv/apps//` with its services as supervisord programs beside its own server (`agentenv_protocol.computer.serve_app`, in a venv with the protocol's `app` extra, on `.agentenv/server.sock` in its folder), which adds its launcher, launches a `boot` app once per boot and says through `computer-app/v1` whether it's ready, as the driver's `computer-ready` asks, its card named after it, and `env_params` `app_settings` handing it values of its `Settings` in `AGENTENV_APP_SETTINGS`, a secret one only as a bare reference, over its env.toml's `[settings]` and a computer env's `settings` beside it in `apps`, each by top-level field (`params.app_settings`; checked at put, a required one left to the deploy; the deploy record keeps the merged values as `app_settings`, secrets withheld); its env.toml's `build_settings`, values of its `BuildSettings` checked at put, are written to `.agentenv/build_settings.json` in its folder, so they're part of the image's key, its install steps read them, and a deploy can't change them). An env type whose services take environment data defines `load_environment_artifact` and `load_environment_universe_artifact`; `load_artifact` refuses an environment artifact aimed at a type that doesn't, at save and before it loads. An env built on the protocol SDK can serve agents a skill (`@agent_skill`, `skill/v1`), which `add_skills` `env_ids` installs under its environment name (behind a gateway, each server's under that server's name); a skill with files beside its SKILL.md goes as a bundle stored once per content under `agentenv-env-skills/` in the object store. `env/gateway/gateway.py` is the env-side gateway server, container-only code. Its site route, `/website//`, never forwards a site's data plane or card (`/agentenv/…`, `/.well-known/agent-env.json`), which agentenv reaches through `/svc//`; a site with no frontend image is served by its server image there. Its CDP route, `/browser//cdp//` (`env/gateway/browser_proxy.py`), reaches the website browser behind the `/svc` child ``, found from that child's card (it, or an app among its children, offering `browser/v1` `cdp` with `/cdp/` in its `computer-app/v1` `agent_routes`): it relays the browser's discovery and WebSocket both ways, each leg pinged every 20 s, and every CDP address the browser hands out, in `json/version`, `json/list` and its `cdp` answer through `/svc//`, comes back at the gateway, everything before `/cdp//` becoming the address the caller reached the gateway at and `/browser/`, keeping the https or wss of an address the browser minted under an origin its caller named, as `browser_connect_cdp` names the agent's, since a tunnel that ends TLS, as Modal's, may send no `X-Forwarded-Proto`; each connection is a `browser_connection` trajectory event, without its token. Its clock (`env/gateway/clock.py`) has one opaque version, a per-boot id and a count, that every `set_time`, `advance` and `clear` changes; `set_rate` (`/clock/set-rate`) changes an armed clock's rate from its time at the write, keeping its `t0` and generation as `advance` does, so time triggers keep their marks. Every clock change goes through `Gateway._change_clock`, which wakes the triggers when the change kept the clock's generation, answers the held watches and waits for the followers. Its followers, the computers on it (`CLOCK_FOLLOWER_TOKENS`, each child whose card serves `clock/v1` `sync_time`, by its read token, its control token for now), long-poll `/clock/watch` naming the version they show, each watch held up to the card's `hold_s` (`env/gateway/clock_followers.py`), and a write given `wait_followers_s` waits for each, answering 504 naming those behind; a restarted gateway is unarmed. While its clock is armed it serves agents `get_time` (`env/gateway/get_time.py`), which answers in UTC and, given an IANA `time_zone`, also in that zone; executors' working `hours` (`env/gateway/triggers.py`) take a fixed `utc_offset` or an IANA `time_zone`. | | `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Every artifact but an environment universe, whose `metadata` key holds its legacy metadata file ids, carries a `metadata` dict outside its versioned fields, changed by compare-and-swap (`merge_metadata`). Built-ins: `CliArtifact` (`cli`), `ComputerArtifact` (`computer`, a base computer: its architectures, OS, session user, settings and container form, whose image `put` reads from its registry into `metadata.validated_images`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`: its environment's files by path, stored as the one wrapped `file_artifact_ref` every environment artifact had when it has one file at its filename, else as `file_artifact_refs`; `get_file_artifacts()` reads either, and a load hands each file to the env's `data/add` with its path, which a pre-v1 load, taking one file, refuses), `EnvironmentUniverseArtifact` (`environment_universe`), `SkillArtifact` (`skill`); `[artifacts] type_aliases` maps another stored spelling of a type onto one of them. A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or a published image's tag with its `image_digest` and its build context, which sandboxes pull by the digest and build from the context when they can't (the infra images, and a bundle agent built from a shipped agent's sources, `Plan.published`), or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/docker/context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. | -| `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, among them `start_computer_recording` and `end_computer_recording`, and more). On an env whose card serves computer-exec/v1 (a computer), `run_code` with `env_id` and `run_container_unit_tests_verifier` with `env_id` run inside it as its session user through the driver: files staged with `data/add` and read with `data/get`, commands run with `run`; `env_outcome_verifier` hands a `verify` that takes an `env_url` keyword the env's address. A computer's framework-only routes (everything but MCP and the card) take the per-deploy control token its provider boots it with (`DeployedEnv.control_token`, `AGENTENV_CONTROL_TOKEN`); framework calls send `deployed.control_headers`, `Task.run` runs every step under `protocol_v1.authorized` with each deployed env's (a `verify` included), and a `run_code` script inside the computer gets `input["env"]["headers"]`. `teardown_run` and the `teardown_sandboxes` step share one core (`task/teardown.py` `tear_down`): the recordings the run left open end with fresh grants, each env still up closes (`Env.close`; a computer env ends any recording still open), what stayed open is registered from its namespace, then every sandbox is terminated by id. `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). | +| `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, among them `start_computer_recording` and `end_computer_recording`, `start_browser_recording` and `end_browser_recording` (`browser-recording/v1`: frames of the tab an agent last acted in after each burst of its actions, keyed to the browser's action log, with the log, one universe per recording, a missing frame failing the run), and more). On an env whose card serves computer-exec/v1 (a computer), `run_code` with `env_id` and `run_container_unit_tests_verifier` with `env_id` run inside it as its session user through the driver: files staged with `data/add` and read with `data/get`, commands run with `run`; `env_outcome_verifier` hands a `verify` that takes an `env_url` keyword the env's address. A computer's framework-only routes (everything but MCP and the card) take the per-deploy control token its provider boots it with (`DeployedEnv.control_token`, `AGENTENV_CONTROL_TOKEN`); framework calls send `deployed.control_headers`, `Task.run` runs every step under `protocol_v1.authorized` with each deployed env's (a `verify` included), and a `run_code` script inside the computer gets `input["env"]["headers"]`. `teardown_run` and the `teardown_sandboxes` step share one core (`task/teardown.py` `tear_down`): the recordings the run left open end with fresh grants, each env still up closes (`Env.close`; a computer env ends any recording still open), what stayed open is registered from its namespace, then every sandbox is terminated by id. `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). | | `store/` | Four store ABCs with local and cloud implementations: `DocumentStore` (SQLite, MongoDB, DynamoDB, Firestore with MongoDB compatibility), `ObjectStore` (filesystem, S3, Cloud Storage), `ImageStore` (local OCI registry, ECR), `SecretStore` (env vars or file, AWS Secrets Manager, Google Cloud Secret Manager). `VersionedEntityStore` implements the shared versioned get/put logic, `QueryBuilder` is the immutable chained query API, `store/base.py` holds the error types. `store/routing.py` is the local namespaces, `@local` and `@hf` (`is_local_id`), each with a documents file of its own: while the CLI or a `build_app` server runs, a local id and what a local task's run records go to local stores beside the configured ones, and that run's writes to the configured stores are refused. A new backend must pass the conformance kits in `tst/store/`. | | `config/` | The `Config` singleton (`get_config`, `configure`, `reset_config`) in `config/runtime.py`, file discovery in `config/loader.py`, and `load_impl`, which resolves `module:Class` pointers. `agent_env.store` re-exports the config names for compatibility. | | `providers/` | `providers/sandbox_providers/` holds the sandbox providers `local`, `modal`, `modal_vm`, `e2b`, `sail_vm` (the `sail` extra), `smol_vm` (the `smol` extra), `tensorlake` (the `tensorlake` extra), `tenki` (the `tenki` extra), `vercel` (the `vercel` extra), `boxd` (the `boxd` extra), `coreweave` (the `coreweave` extra), `daytona` (the `daytona` extra), `boat` (the `boat` extra), `agent37` (no extra); `[sandbox] default` and `agent_default` accept a comma-separated fallback chain, whose deploys skip a provider that can't run them. A sandbox provider declares what it runs (`ON_THIS_MACHINE`, and for VMs on this machine the address their requests to it arrive at, `REACHES_HOST_AT`; `CONTAINER_ACCEPTS`, `PRIVATE_NETWORK`, `WEBSOCKETS`; `SUPPORTED_SANDBOX_MODES`, derived from whether it implements `create_sandbox` and `create_vm`; and `SANDBOX_ACCEPTS`, the images an agent's sandbox runs, which defaults to those a VM loads when it implements `create_vm` and to `CONTAINER_ACCEPTS` otherwise, so only a provider that implements `create_vm` yet gives an agent a container, like the local one, declares it), each sandbox carries the `platform` of the images it runs (linux/amd64 unless its provider sets another when it creates it; None, any, unchecked, for a local sandbox), and code outside `providers/sandbox_providers/` reads those, never which provider it has. `providers/env_providers/` holds the environment providers: `EnvironmentProvider` (an env's containers and state store), `EnvironmentGatewayProvider` (`gateway`, an env's default), which renders a docker-compose for the gateway and its MCP servers inside the sandbox, and `EnvironmentServerProvider` (`server`), one MCP server in its own container with no gateway in front and no state store; `providers/env_state/` holds env-state providers (`local_postgres` built in). | diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index 5cb46e842..fec7ad0dc 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -1128,6 +1128,17 @@ are the profile's template. The base class adds: The server keeps each token in memory and only its SHA-256, with its user and base address, in `.agentenv/cdp-tokens.json`, which code in the computer's session can read but not use; an address outlives a relaunch of the browser and of the server, and the computer's reset ends it. +- **`browser-recording/v1`** on its card, framework-only, with `max_duration_s`, `max_frames` and `frames`: + `start {request_id, upload, frames?}` records under the write namespace grant `upload`, a start again with the + same `request_id` answering the recording it began; `end {recording_id, upload?, reason?}` ends it, under a fresh + grant when one is given, and answers its manifest; `end_all {reason?}` ends every open one. With `frames`, a frame + is a JPEG of the tab an agent last acted in, else of the active tab: one at the start, one after each burst of an + agent's acting entries (CDP commands and MCP calls, not agentenv's own) has gone quiet for 0.5 s and its tab's page + has loaded (up to 5 s), and one at the end, each uploaded as it's taken under `/frames/` and keyed in + the manifest to the entries it follows (`after`); a frame it couldn't take is in the manifest with its `error`. At + the end it uploads the log's lines since its start, actions and tab events, as `log.jsonl`, and `manifest.json` + (`BrowserRecordingManifest`). `start_browser_recording` and `end_browser_recording` run it, registering the frames, + log and manifest as one file artifact universe, and the end fails the run when its log or a frame is missing. - **Its CDP relay**, at those addresses: `json/version` and `json/list` (also with a trailing slash, and `json`) answered from the browser, since a browser on a pipe has none, with its WebSocket at the address the client reached it at; and `devtools/browser`, where each WebSocket connection gets a diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py index d40a684be..7c92d8b01 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/__init__.py @@ -12,14 +12,17 @@ from .environment import (ALONE_BYPASS, BLOCK, BROWSER, COMPUTER_BYPASS, OWNED_FLAGS, POLICIES, READY_POLL_S, SHM_MIN_BYTES, SITE_PROXY_ENV, SITES, SKILL, BrowserSettings, WebsiteBrowserEnvironment, Window, start_tab_why) +from .recording import BROWSER_RECORDING, BrowserRecordingManifest, BrowserRecordings, RecordedFrame from .shapes import (BrowserRecord, ClosedTab, Cookie, CookieToSet, Dialog, Download, LaunchEntry, OpenTab, OriginStorage, PartitionKey, SiteNames, StartTab, StartTabsLaunch, StorageItem, origin_of) -__all__ = ["ACTIVE_CHECK_TIMEOUT_S", "AGENT", "AGENTENV", "ALONE_BYPASS", "BLOCK", "BROWSER", "CAPTURE_TIMEOUT_S", +__all__ = ["ACTIVE_CHECK_TIMEOUT_S", "AGENT", "AGENTENV", "ALONE_BYPASS", "BLOCK", "BROWSER", "BROWSER_RECORDING", + "CAPTURE_TIMEOUT_S", "COMPUTER_BYPASS", "DOWNLOADS", "EXIT", "FAILED", "HASH_CHUNK_BYTES", "LIVE_START_TAB_TIMEOUT_S", "LOADED", "MAX_MESSAGE_BYTES", "OWNED_FLAGS", "PAGE_INPUT", "PAGES", "PIN_TIMEOUT_S", "PIPE_IN", "PIPE_OUT", "POLICIES", "READY_POLL_S", "SHM_MIN_BYTES", "SITE_PROXY_ENV", "SITES", "SKILL", "STOP", "TIMED_OUT", "UNKNOWN", "UNNAVIGATED_CAPTURE_TIMEOUT_S", "UPLOADS", "BrowserControl", "BrowserSettings", "DevToolsError", "DevToolsPipe", "WebsiteBrowserEnvironment", "Window", "BrowserRecord", "ClosedTab", "Cookie", "CookieToSet", "Dialog", "Download", "LaunchEntry", "OpenTab", "OriginStorage", "PartitionKey", "SiteNames", "StartTab", - "StartTabsLaunch", "StorageItem", "origin_of", "start_tab_why"] + "StartTabsLaunch", "StorageItem", "origin_of", "start_tab_why", "BrowserRecordingManifest", "BrowserRecordings", + "RecordedFrame"] 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 03ac3e25f..f8371b997 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py @@ -44,6 +44,13 @@ #: The CDP input a computer's page driver may send to the active tab, and how long its screenshot may take. PAGE_INPUT = frozenset({"Input.dispatchMouseEvent", "Input.dispatchKeyEvent", "Input.insertText"}) CAPTURE_TIMEOUT_S, UNNAVIGATED_CAPTURE_TIMEOUT_S = 3.0, 1.0 +#: A frame's quality as a JPEG, which is a fifth to a tenth of the PNG's size. +FRAME_QUALITY = 80 +#: How long a frame's capture may go unanswered before a second capture is sent: Chrome answers a capture of a tab +#: behind the active one only once another capture of it arrives, and then answers both. +NUDGE_S = 0.5 +#: How often a wait for a tab to load looks again. +LOAD_POLL_S = 0.1 #: Where an agent's files to upload are, and where downloads land, under the session user's home. UPLOADS = "Uploads" DOWNLOADS = "Downloads" @@ -97,11 +104,15 @@ def __init__(self, devtools: DevToolsPipe, dialogs: str, viewport: Optional[tupl #: Each tab's latest capture, which shows while a dialog keeps its page from painting; and, by session, what #: input sent there waits on to learn a dialog opened. self._shots: dict[str, str] = {} + #: Each tab's latest frame, the image and its format, which a frame shows again while a dialog keeps it from + #: painting. + self._frames: dict[str, tuple[str, str]] = {} self._dialog_opened: dict[str, asyncio.Event] = {} self._pinned: dict[str, asyncio.Event] = {} self._tasks: set[asyncio.Future] = set() - #: Each start tab loading, by its session. + #: Each start tab loading, by its session; and each tab whose page has committed and not yet loaded. self._loading: dict[str, _StartLoad] = {} + self._page_loading: set[str] = set() #: Every http and https origin a page or frame it follows loaded since this server started, and the sessions #: of agentenv's own pages whose requests it answers itself. self.origins: set[str] = set() @@ -613,6 +624,63 @@ async def capture_page(self) -> dict: await self._refresh(target) return {**({"png": png} if png is not None else {"blank": True}), "page": self._state_of(target)} + async def frame(self, target: Optional[str], load_wait_s: float) -> tuple[str, str, Optional[str], Optional[str]]: + """A frame of the tab ``target``, else, for none or one that's gone, of the active tab: once its page has loaded, + waiting up to ``load_wait_s``, a JPEG of it as it shows, or, with a dialog keeping it from painting, the frame it + last took, else its last capture, a PNG; a tab that closes while its page loads is the active tab's. The image + base64, its format and the tab and its URL; DeviceUnavailable with no tab open, or for one that paints nothing + in time.""" + session = next((s for s, t in self._sessions.items() if t == target), None) if target else None + if session is None: + session = await self.active_session() + target = self._sessions[session] + deadline = asyncio.get_running_loop().time() + load_wait_s + while target in self._page_loading and asyncio.get_running_loop().time() < deadline: + await asyncio.sleep(LOAD_POLL_S) + if session not in self._sessions: + session = await self.active_session() + target = self._sessions[session] + if session in self._open_dialogs: + if (earlier := self._frames.get(target)) is None and (shot := self._shots.get(target)) is not None: + earlier = shot, "png" + if earlier is None: + raise DeviceUnavailable("a dialog keeps the tab from painting and it had no capture before it") + return *earlier, target, self.url_of(target) + try: + shot = await self._nudged_capture(session) + except asyncio.TimeoutError: + raise DeviceUnavailable(f"the tab painted nothing within {CAPTURE_TIMEOUT_S:g} s") from None + except (ConnectionError, DevToolsError) as e: + raise DeviceUnavailable(f"the tab couldn't be captured: {e}") from None + self._frames[target] = shot["data"], "jpg" + return shot["data"], "jpg", target, self.url_of(target) + + async def _nudged_capture(self, session: str) -> dict: + """The tab's JPEG from the first of its captures to answer: a second follows one unanswered after ``NUDGE_S``, + since a tab behind the active one answers only then, all within ``CAPTURE_TIMEOUT_S``.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + CAPTURE_TIMEOUT_S + + def capture() -> asyncio.Future: + return asyncio.ensure_future(self.devtools.call("Page.captureScreenshot", { + "format": "jpeg", "quality": FRAME_QUALITY, "optimizeForSpeed": True}, session_id=session, + timeout_s=max(0.0, deadline - loop.time()))) + + captures = {capture()} + try: + done, _ = await asyncio.wait(captures, timeout=NUDGE_S) + if not done: + captures.add(capture()) + while True: + done, captures = await asyncio.wait(captures, return_when=asyncio.FIRST_COMPLETED) + if answers := [c.result() for c in done if c.exception() is None]: + return answers[0] + if not captures: + raise next(iter(done)).exception() + finally: + for pending in captures: + pending.cancel() + async def _capture(self, session: str) -> Optional[str]: """The tab's page as it shows now; with a dialog open on it, which keeps it from painting, the last it took.""" target = self._sessions[session] @@ -727,6 +795,8 @@ 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 not frame.get("parentId") and (target := self._sessions.get(session)) is not None: + self._page_loading.add(target) 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"))) @@ -807,13 +877,16 @@ async def _let_run(self, session: str) -> None: def _on_Target_detachedFromTarget(self, params: dict, session: Optional[str]) -> None: if (target := self._sessions.pop(params["sessionId"], None)) is not None: self._shots.pop(target, None) + self._frames.pop(target, None) self._pinned.pop(target, None) + self._page_loading.discard(target) self._choosers.pop(params["sessionId"], None) self._dialog_opened.pop(params["sessionId"], None) def _on_Page_loadEventFired(self, params: dict, session: Optional[str]) -> None: self._choosers.pop(session, None) if (target := self._sessions.get(session)) is not None: + self._page_loading.discard(target) self._spawn(self._refresh(target)) if (load := self._loading.get(session)) is not None and load.commits: load.loaded.add(load.commits[-1]) 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 b301c50fc..0ca7e4345 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/environment.py @@ -33,6 +33,7 @@ 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 +from .recording import BROWSER_RECORDING, MAX_DURATION_S, MAX_FRAMES, BrowserRecordings from .shapes import BrowserRecord, BrowserState, CookieToSet, OriginStorage, SiteNames, StateToLoad logger = logging.getLogger(__name__) @@ -155,6 +156,8 @@ def __init__(self) -> None: self.cdp_tokens = Tokens(SOCKET.parent / "cdp-tokens.json") self.relay = Relay(self.devtools, dialogs_answered=self.settings.dialogs != AGENT, log=self.log, control=self.control) + #: Its recordings: frames of the tab an agent last acted in, and its log, uploaded under each one's grant. + self.recordings = BrowserRecordings(self.log, self.control) self.devtools.on_close(self._pipe_closed) @classmethod @@ -372,6 +375,25 @@ async def browser_calls_end(self, entry: int, ok: bool, error: Optional[str] = N raise InvalidArgument(f"no tool call is open as entry {entry}") return {} + @extension(BROWSER_RECORDING, op="start") + async def browser_recording_start(self, request_id: str, upload: dict, frames: bool = True) -> dict: + """A recording of this browser, its frames and its log uploaded under the write namespace grant ``upload``: + with ``frames``, a frame of the tab an agent last acted in at the start, after each burst of its actions + settles and at the end. A start again with the same ``request_id`` answers the recording it began.""" + return await self.recordings.start(request_id, upload, frames) + + @extension(BROWSER_RECORDING, op="end") + async def browser_recording_end(self, recording_id: str, upload: Optional[dict] = None, + reason: str = "end") -> dict: + """The recording ``recording_id`` ended, under a fresh grant ``upload`` when one is given: its last frames, + its log's lines since its start and its manifest uploaded, and the manifest answered.""" + return await self.recordings.end(recording_id, upload, reason) + + @extension(BROWSER_RECORDING, op="end_all") + async def browser_recording_end_all(self, reason: str = "teardown") -> dict: + """Every open recording ended, under its start's grant, as the env closes or resets: their manifests.""" + return {"recordings": await self.recordings.end_all(reason)} + @extension(BROWSER, op="cdp") async def browser_cdp(self, user: str, request: Request, origin: Optional[str] = None) -> dict: """``user``'s CDP address once the browser's ready, the same one each time it's asked for, as the server that @@ -649,6 +671,8 @@ async def _start_tabs_settled(self) -> list[dict]: return self.start_tab_launches[-1]["tabs"] if self.start_tab_launches else [] def extension_params(self, uri: str) -> dict: + if uri == BROWSER_RECORDING: + return {"max_duration_s": MAX_DURATION_S, "max_frames": MAX_FRAMES, "frames": True} if uri == BROWSER: params: dict[str, Any] = {"chromium": self.version, "clock": "real", "mode": _mode()} if _alone(): diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py index f86f62f95..c9dd0b8b2 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/log.py @@ -23,7 +23,7 @@ from collections.abc import Mapping from datetime import datetime, timezone from pathlib import Path -from typing import Any, Optional +from typing import Any, Callable, Optional from urllib.parse import parse_qsl, quote, urlsplit, urlunsplit logger = logging.getLogger(__name__) @@ -314,6 +314,8 @@ def __init__(self, path: Path) -> None: self._unwritable = False #: Per tab, the sequence number of the latest acting entry on it. self.last_acting: dict[str, int] = {} + #: Each told every action as it's written, its sequence number and its fields, as a recording follows them. + self.watchers: list[Callable[[int, dict], None]] = [] @classmethod def for_app(cls, name: str, fallback: Path) -> BrowserLog: @@ -381,6 +383,16 @@ def finish(self, begun: dict, **fields: Any) -> None: self.note("outcome", {"entry": seq, **fields}) else: self.write(seq, {"action": action}, at=begun["at"]) + for watcher in self.watchers: + try: + watcher(seq, action) + except Exception: + logger.exception("a watcher of the browser's log failed") + + @property + def next_seq(self) -> int: + """The sequence number the next line will take.""" + return self._seq + 1 def note(self, kind: str, body: dict) -> int: """A line of ``kind`` written now: a ``tab`` event, or a ``reset``.""" diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py new file mode 100644 index 000000000..1894927c3 --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py @@ -0,0 +1,263 @@ +"""Recording a website browser, browser-recording/v1: frames of the tab an agent last acted in, one at the start, one +after each burst of its actions settles and one at the end, each keyed to the log entries it follows and uploaded under +the recording's grant as it's taken; and, at the end, the log's lines since the start, its actions and tab events, and +a manifest of it all. A frame the browser couldn't take is in the manifest with why, and counts as missing.""" +from __future__ import annotations + +import asyncio +import base64 +import json +import logging +import re +import uuid +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Literal, Optional + +from pydantic import BaseModel, ValidationError + +from ..computer.errors import InvalidArgument +from ..transfers import NamespaceUploader, WriteNamespaceGrant +from .log import ERROR_CHARS, BrowserLog + +if TYPE_CHECKING: + from .control import BrowserControl + +logger = logging.getLogger(__name__) + +BROWSER_RECORDING = "urn:agentenv:browser-recording/v1" +MANIFEST = "manifest.json" +LOG = "log.jsonl" +FRAMES = "frames" +REASONS = ("end", "reset", "teardown") +#: How long a burst of an agent's actions must go quiet before its frame is taken, so typing a word is one frame. +SETTLE_S = 0.5 +#: How long a frame waits for its tab's page to load after a navigation. +LOAD_WAIT_S = 5.0 +#: How long a recording may run, which its grant lasts, and how many frames it takes at most. +MAX_DURATION_S = 4 * 3600 +MAX_FRAMES = 5000 +#: The ways in whose actions are an agent's: CDP commands and MCP tool calls, not agentenv's own. +AGENTS = ("cdp", "mcp") +CONTENT_TYPES = {"jpg": "image/jpeg", "png": "image/png"} +#: How long the recordings a server ended are kept for an end that comes again. +KEEP_ENDED = 64 +#: How long an end waits for the entries still in flight to be written, so its log covers them, and how often it +#: looks again. +LOG_WAIT_S = 10.0 +LOG_POLL_S = 0.2 + + +class RecordedFrame(BaseModel): + """One frame: the first, one after a burst of actions, whose entries it names in ``after``, or the last; its tab + and URL, and its file under the recording's folder, or why there is none. Its ``kind`` and ``at`` are None where + they aren't known, in a manifest rebuilt from the files a recording uploaded.""" + + index: int + kind: Optional[Literal["first", "action", "last"]] = None + at: Optional[str] = None + after: list[int] = [] + tab: Optional[str] = None + url: Optional[str] = None + path: Optional[str] = None + content_type: Optional[str] = None + uploaded: bool = False + error: Optional[str] = None + + +class BrowserRecordingManifest(BaseModel): + """A recording as it ended: its frames, and its log's lines, entries ``log_after`` + 1 to ``log_last``, as + ``log``, the actions and tab events between its start and its end, with any of those entries it doesn't hold, still + unwritten or unreadable at its end, in ``log_missing``.""" + + recording_id: str + started_at: str + ended_at: Optional[str] = None + ended_by: Optional[Literal["end", "reset", "teardown", "error"]] = None + error: Optional[str] = None + frames_on: bool = True + frames: list[RecordedFrame] = [] + log: Optional[str] = None + log_after: int = 0 + log_last: int = 0 + log_uploaded: bool = False + log_missing: list[int] = [] + + @property + def missing_frames(self) -> int: + return sum(1 for frame in self.frames if not frame.uploaded) + + +def frame_path(index: int, extension: str) -> str: + """Where frame ``index`` uploads, under its recording's folder.""" + return f"{FRAMES}/{index:04d}.{extension}" + + +def frame_of(path: str) -> Optional[tuple[int, str]]: + """The index and content type of the frame ``frame_path`` put at ``path``, or None for another file.""" + if (found := re.fullmatch(rf"{FRAMES}/(\d+)\.(\w+)", path)) and found[2] in CONTENT_TYPES: + return int(found[1]), CONTENT_TYPES[found[2]] + return None + + +class BrowserRecordings: + """The recordings of the browser that ``log`` and ``control`` are, followed from the log as its actions are + written.""" + + def __init__(self, log: BrowserLog, control: BrowserControl) -> None: + self.log, self.control = log, control + self._open: dict[str, _Recording] = {} + self._by_request: dict[str, str] = {} + self._ended: dict[str, dict] = {} + log.watchers.append(self.acted) + + async def start(self, request_id: str, upload: dict, frames: bool = True) -> dict: + """A recording under the grant ``upload``, or the one a start with the same ``request_id`` began, which a + retried start joins: its ``recording_id`` and ``started_at``.""" + if (known := self._by_request.get(request_id)) in self._open: + recording = self._open[known] + return {"recording_id": recording.manifest.recording_id, "started_at": recording.manifest.started_at} + recording = _Recording(self, _grant(upload), frames) + self._open[recording.manifest.recording_id] = recording + self._by_request[request_id] = recording.manifest.recording_id + if frames: + await recording.take("first") + return {"recording_id": recording.manifest.recording_id, "started_at": recording.manifest.started_at} + + async def end(self, recording_id: str, upload: Optional[dict] = None, reason: str = "end") -> dict: + """The recording ended, its last frames taken and its log and manifest uploaded, under ``upload`` when one is + given, else under its start's grant: its manifest. One already ended answers its manifest again.""" + if reason not in REASONS: + raise InvalidArgument(f"reason is one of {', '.join(REASONS)}, not {reason!r}") + if recording_id in self._ended: + return self._ended[recording_id] + recording = self._open.get(recording_id) + if recording is None: + raise InvalidArgument(f"no recording {recording_id} is open here") + manifest = await recording.end(_grant(upload) if upload is not None else None, reason) + self._open.pop(recording_id, None) + self._ended[recording_id] = manifest + while len(self._ended) > KEEP_ENDED: + self._ended.pop(next(iter(self._ended))) + return manifest + + async def end_all(self, reason: str = "teardown") -> list[dict]: + """Every open recording ended, as its env closes or resets.""" + return [await self.end(recording_id, None, reason) for recording_id in list(self._open)] + + def acted(self, seq: int, action: dict) -> None: + """An action written to the log: an agent's acting one starts, or holds back, the frame of its burst.""" + if action.get("acts") and action.get("via") in AGENTS: + for recording in self._open.values(): + recording.acted(seq, action.get("tab")) + + +class _Recording: + def __init__(self, recordings: BrowserRecordings, grant: WriteNamespaceGrant, frames: bool) -> None: + self.recordings, self.uploader, self.frames_on = recordings, NamespaceUploader(grant), frames + recording_id = uuid.uuid4().hex + self.manifest = BrowserRecordingManifest(recording_id=recording_id, started_at=_iso(), frames_on=frames, + log_after=recordings.log.next_seq - 1) + #: The acting entries since the last frame, and the tab the latest of them was on. + self.pending: list[int] = [] + self.tab: Optional[str] = None + self._timer: Optional[asyncio.TimerHandle] = None + self._taking: set[asyncio.Future] = set() + self._lock = asyncio.Lock() + self._ending = False + + def acted(self, seq: int, tab: Optional[str]) -> None: + if not self.frames_on or self._ending: + return + self.pending.append(seq) + if tab: + self.tab = tab + if self._timer is not None: + self._timer.cancel() + self._timer = asyncio.get_running_loop().call_later(SETTLE_S, self._settled) + + def _settled(self) -> None: + self._timer = None + taking = asyncio.ensure_future(self.take("action")) + self._taking.add(taking) + taking.add_done_callback(self._taking.discard) + + async def take(self, kind: str) -> None: + """The frame of ``kind``, of the tab the agent last acted in, uploaded as it's taken; one an action burst + brings only once that burst has entries no frame followed yet.""" + async with self._lock: + after, self.pending = self.pending, [] + if kind == "action" and not after: + return + index = len(self.manifest.frames) + frame = RecordedFrame(index=index, kind=kind, at=_iso(), after=after, tab=self.tab) + self.manifest.frames.append(frame) + if index >= MAX_FRAMES: + frame.error = f"the recording reached its {MAX_FRAMES} frames" + return + try: + data, extension, frame.tab, frame.url = await self.recordings.control.frame(self.tab, LOAD_WAIT_S) + frame.at = _iso() + frame.path, frame.content_type = frame_path(index, extension), CONTENT_TYPES[extension] + await self.uploader.upload(self._path(frame.path), base64.b64decode(data)) + frame.uploaded = True + except Exception as e: + frame.error = f"{type(e).__name__}: {e}"[:ERROR_CHARS] + logger.warning("browser recording %s: frame %d: %s", self.manifest.recording_id, index, frame.error) + + async def end(self, grant: Optional[WriteNamespaceGrant], reason: str) -> dict: + self._ending = True + if self._timer is not None: + self._timer.cancel() + self._timer = None + await asyncio.gather(*self._taking, return_exceptions=True) + if grant is not None: + self.uploader = NamespaceUploader(grant) + if self.frames_on: + await self.take("action") + await self.take("last") + manifest = self.manifest + manifest.log_last = self.recordings.log.next_seq - 1 + lines = await self._log_lines() + manifest.log_missing = sorted(set(range(manifest.log_after + 1, manifest.log_last + 1)) + - {line.get("seq") for line in lines}) + text = "".join(json.dumps(line, default=str) + "\n" for line in lines) + manifest.log = LOG + try: + await self.uploader.upload(self._path(LOG), text.encode()) + manifest.log_uploaded = True + except Exception as e: + manifest.error = f"the log didn't upload: {type(e).__name__}: {e}"[:ERROR_CHARS] + manifest.ended_at, manifest.ended_by = _iso(), reason + wire = manifest.model_dump(mode="json") + try: + await self.uploader.upload(self._path(MANIFEST), json.dumps(wire, indent=1).encode()) + except Exception as e: + logger.warning("browser recording %s: its manifest didn't upload: %s", manifest.recording_id, e) + return wire + + async def _log_lines(self) -> list[dict]: + """The log's lines from the recording's start to its end, once every one of them is written, or what's + written of them after ``LOG_WAIT_S``.""" + manifest, log = self.manifest, self.recordings.log + deadline = asyncio.get_running_loop().time() + LOG_WAIT_S + wanted = manifest.log_last - manifest.log_after + while True: + lines = [line for line in (await log.read_async(manifest.log_after, limit=wanted))["lines"] + if line.get("seq", 0) <= manifest.log_last] + if len(lines) >= wanted or asyncio.get_running_loop().time() >= deadline: + return lines + await asyncio.sleep(LOG_POLL_S) + + def _path(self, name: str) -> str: + return f"{self.manifest.recording_id}/{name}" + + +def _grant(upload: object) -> WriteNamespaceGrant: + try: + return WriteNamespaceGrant.model_validate(upload) + except ValidationError as e: + raise InvalidArgument(f"upload must be a write namespace grant: {e.errors()[0]['msg']}") from None + + +def _iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") diff --git a/packages/agentenv-protocol/tests/test_browser_recording.py b/packages/agentenv-protocol/tests/test_browser_recording.py new file mode 100644 index 000000000..4b35d1899 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_browser_recording.py @@ -0,0 +1,284 @@ +"""A website browser's recording, browser-recording/v1: a first frame, one after each burst of an agent's acting +actions once it settles, of the tab the burst last acted in, keyed to its entries, and a last frame, each uploaded as +it's taken; at the end its log's lines since the start and its manifest; a frame it couldn't take kept with why; a +start again joining its recording and an end again answering the same; and a frame waiting for its tab's page to +load, of the active tab once the agent's has closed.""" +from __future__ import annotations + +import asyncio +import base64 +import json +from datetime import datetime, timedelta, timezone + +import pytest + +from agentenv_protocol.computer.errors import DeviceUnavailable, InvalidArgument +from agentenv_protocol.transfers import HttpPostPolicyGrant, WriteNamespaceGrant +from agentenv_protocol.website_browser import recording as browser_recording +from agentenv_protocol.website_browser.control import NUDGE_S, BrowserControl +from agentenv_protocol.website_browser.log import BrowserLog +from agentenv_protocol.website_browser.recording import BrowserRecordingManifest, BrowserRecordings + + +def _grant() -> dict: + return WriteNamespaceGrant( + root_path="agentenv-browser-recordings/r1", expires_at=datetime.now(timezone.utc) + timedelta(minutes=5), + max_objects=100, max_object_bytes=1 << 20, max_total_bytes=1 << 24, + write=HttpPostPolicyGrant(kind="http-post-policy", url="https://objects.example.test/upload", + fields={"policy": "p"}, path_field="key", file_field="file"), + ).model_dump(mode="json") + + +class Uploads: + """Every upload, by grant root and path, in place of the object store.""" + + def __init__(self) -> None: + self.by_path: dict[str, bytes] = {} + + def uploader(self, grant): + uploads = self + + class Uploader: + async def upload(self, path, source): + uploads.by_path[f"{grant.root_path}/{path}"] = source + return Uploader() + + +class Control: + """The browser's control as a recording asks it: each frame the tab asked for, else T0, or why there's none.""" + + def __init__(self) -> None: + self.asked, self.fail = [], None + + async def frame(self, target, load_wait_s): + self.asked.append(target) + if self.fail: + raise DeviceUnavailable(self.fail) + tab = target or "T0" + return base64.b64encode(f"jpeg of {tab}".encode()).decode(), "jpg", tab, f"https://a.test/{tab}" + + +@pytest.fixture +def recorded(monkeypatch, tmp_path): + uploads, control = Uploads(), Control() + monkeypatch.setattr(browser_recording, "NamespaceUploader", uploads.uploader) + monkeypatch.setattr(browser_recording, "SETTLE_S", 0.05) + log = BrowserLog(tmp_path / "browser-chromium.jsonl") + return log, control, uploads, BrowserRecordings(log, control) + + +def _acted(log: BrowserLog, via: str, method: str, tab: str, acts: bool = True) -> int: + begun = log.begin({"via": via, "who": "ada", "method": method, "acts": acts, "tab": tab}) + log.finish(begun, ok=True) + return begun["seq"] + + +@pytest.mark.asyncio +async def test_a_frame_follows_each_settled_burst_of_an_agents_actions_on_its_tab_keyed_to_its_entries(recorded): + log, control, uploads, recordings = recorded + started = await recordings.start("run-1:record", _grant()) + typed = [_acted(log, "cdp", "Input.insertText", "T1") for _ in range(3)] + _acted(log, "cdp", "DOM.getDocument", "T1", acts=False) + _acted(log, "agentenv", "navigate", "T1") + await asyncio.sleep(0.2) + clicked = _acted(log, "mcp", "browser_click", "T2") + await asyncio.sleep(0.2) + manifest = BrowserRecordingManifest.model_validate(await recordings.end(started["recording_id"])) + + frames = [(f.kind, f.after, f.tab, f.path) for f in manifest.frames] + assert frames == [("first", [], "T0", "frames/0000.jpg"), ("action", typed, "T1", "frames/0001.jpg"), + ("action", [clicked], "T2", "frames/0002.jpg"), ("last", [], "T2", "frames/0003.jpg")] + assert all(f.uploaded and f.content_type == "image/jpeg" and f.url == f"https://a.test/{f.tab}" + for f in manifest.frames) + root = f"agentenv-browser-recordings/r1/{started['recording_id']}" + assert uploads.by_path[f"{root}/frames/0001.jpg"] == b"jpeg of T1" + logged = [json.loads(line)["seq"] for line in uploads.by_path[f"{root}/log.jsonl"].decode().splitlines()] + assert logged == list(range(manifest.log_after + 1, manifest.log_last + 1)) and len(logged) == 6 + assert json.loads(uploads.by_path[f"{root}/manifest.json"]) == manifest.model_dump(mode="json") + assert (manifest.ended_by, manifest.log_uploaded, manifest.missing_frames) == ("end", True, 0) + + +@pytest.mark.asyncio +async def test_a_burst_still_settling_at_the_end_gets_its_frame_before_the_last(recorded): + log, control, uploads, recordings = recorded + started = await recordings.start("run-1:record", _grant()) + seq = _acted(log, "cdp", "Input.dispatchMouseEvent", "T1") + manifest = await recordings.end(started["recording_id"], _grant(), "teardown") + + assert [(f["kind"], f["after"]) for f in manifest["frames"]] == [("first", []), ("action", [seq]), ("last", [])] + assert manifest["ended_by"] == "teardown" + + +@pytest.mark.asyncio +async def test_an_end_waits_for_entries_in_flight_and_lists_any_still_unwritten_as_missing(recorded, monkeypatch): + log, control, uploads, recordings = recorded + monkeypatch.setattr(browser_recording, "LOG_WAIT_S", 0.3) + started = await recordings.start("run-1:record", _grant(), frames=False) + finishing = log.begin({"via": "cdp", "who": "ada", "method": "Page.navigate", "acts": True, "tab": "T1"}) + asyncio.get_running_loop().call_later(0.1, lambda: log.finish(finishing, ok=True)) + complete = await recordings.end(started["recording_id"]) + started = await recordings.start("run-2:record", _grant(), frames=False) + stuck = log.begin({"via": "cdp", "who": "ada", "method": "Page.navigate", "acts": True, "tab": "T1"}) + incomplete = await recordings.end(started["recording_id"]) + + assert complete["log_missing"] == [] and complete["log_last"] == finishing["seq"] + assert incomplete["log_missing"] == [stuck["seq"]] + + +@pytest.mark.asyncio +async def test_a_frame_the_browser_cant_take_is_kept_with_why_and_counts_as_missing(recorded): + log, control, uploads, recordings = recorded + started = await recordings.start("run-1:record", _grant()) + control.fail = "the tab painted nothing within 3 s" + _acted(log, "cdp", "Input.dispatchMouseEvent", "T1") + await asyncio.sleep(0.2) + manifest = BrowserRecordingManifest.model_validate(await recordings.end(started["recording_id"])) + + assert [f.uploaded for f in manifest.frames] == [True, False, False] and manifest.missing_frames == 2 + assert manifest.frames[1].error == "DeviceUnavailable: the tab painted nothing within 3 s" + + +@pytest.mark.asyncio +async def test_without_frames_a_recording_takes_none_and_still_uploads_its_log(recorded): + log, control, uploads, recordings = recorded + started = await recordings.start("run-1:record", _grant(), frames=False) + _acted(log, "cdp", "Input.dispatchMouseEvent", "T1") + await asyncio.sleep(0.2) + manifest = await recordings.end(started["recording_id"]) + + assert manifest["frames"] == [] and manifest["log_uploaded"] and control.asked == [] + + +@pytest.mark.asyncio +async def test_a_start_again_joins_its_recording_an_end_again_answers_the_same_and_an_unknown_one_is_refused(recorded): + log, control, uploads, recordings = recorded + first = await recordings.start("run-1:record", _grant()) + again = await recordings.start("run-1:record", _grant()) + ended = await recordings.end(first["recording_id"]) + + assert again == first and await recordings.end(first["recording_id"]) == ended + with pytest.raises(InvalidArgument, match="no recording nope"): + await recordings.end("nope") + with pytest.raises(InvalidArgument, match="reason is one of"): + await recordings.end(first["recording_id"], reason="later") + with pytest.raises(InvalidArgument, match="write namespace grant"): + await recordings.start("run-2:record", {"root_path": "x"}) + + +@pytest.mark.asyncio +async def test_every_open_recording_ends_as_its_env_closes(recorded): + log, control, uploads, recordings = recorded + await recordings.start("run-1:a", _grant()) + await recordings.start("run-1:b", _grant(), frames=False) + + ended = await recordings.end_all("reset") + + assert [m["ended_by"] for m in ended] == ["reset", "reset"] and await recordings.end_all() == [] + + +class Pipe: + """The browser's pipe for a frame: each capture recorded with its params, its answer a JPEG's bytes.""" + + def __init__(self) -> None: + self.listeners, self.captured = [], [] + + 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 {})}) + + async def call(self, method, params=None, *, session_id=None, timeout_s=None): + if method == "Page.captureScreenshot": + self.captured.append((session_id, params, asyncio.get_running_loop().time())) + return {"data": base64.b64encode(b"jpeg").decode()} + return {} + + +@pytest.mark.asyncio +async def test_a_frame_waits_for_its_tabs_page_to_load_then_captures_it_as_a_jpeg_and_a_closed_tabs_is_the_active(): + pipe = Pipe() + control = BrowserControl(pipe, "agent", (1280, 800)) + pipe.emit("Target.targetCreated", {"targetInfo": {"targetId": "T1", "type": "page", "url": "about:blank"}}) + pipe.emit("Target.attachedToTarget", {"sessionId": "s-T1", "targetInfo": {"targetId": "T1", "type": "page"}}) + pipe.emit("Page.frameNavigated", {"frame": {"id": "T1", "url": "https://a.test/", "loaderId": "L1"}}, "s-T1") + loop = asyncio.get_running_loop() + loaded_at = loop.time() + 0.2 + loop.call_later(0.2, pipe.emit, "Page.loadEventFired", {}, "s-T1") + + data, extension, tab, _ = await control.frame("T1", load_wait_s=5) + + (session, params, captured_at), = pipe.captured + assert (session, extension, tab, base64.b64decode(data)) == ("s-T1", "jpg", "T1", b"jpeg") + assert params["format"] == "jpeg" and params["quality"] == 80 and captured_at >= loaded_at + assert (await control.frame("closed", load_wait_s=0))[2] == "T1" + + +def _two_tabs() -> tuple[Pipe, BrowserControl]: + pipe = Pipe() + control = BrowserControl(pipe, "agent", (1280, 800)) + for tab in ("T1", "T2"): + pipe.emit("Target.targetCreated", {"targetInfo": {"targetId": tab, "type": "page", "url": "about:blank"}}) + pipe.emit("Target.attachedToTarget", {"sessionId": f"s-{tab}", "targetInfo": {"targetId": tab, "type": "page"}}) + return pipe, control + + +@pytest.mark.asyncio +async def test_while_a_dialog_keeps_a_tab_from_painting_its_frame_is_the_last_it_took(): + pipe, control = _two_tabs() + first = await control.frame("T1", load_wait_s=0) + pipe.emit("Page.javascriptDialogOpening", {"type": "confirm", "message": "Delete?", "url": "https://a.test/"}, "s-T1") + + during = await control.frame("T1", load_wait_s=0) + + assert during[:3] == first[:3] == (base64.b64encode(b"jpeg").decode(), "jpg", "T1") and len(pipe.captured) == 1 + pipe.emit("Page.javascriptDialogOpening", {"type": "alert", "message": "Hi", "url": "https://a.test/"}, "s-T2") + with pytest.raises(DeviceUnavailable, match="no capture before it"): + await control.frame("T2", load_wait_s=0) + + +class BehindPipe(Pipe): + """A pipe whose tab is behind the active one, as Chrome answers its captures: only once another arrives, then + both.""" + + def __init__(self) -> None: + super().__init__() + self.waiting: list[asyncio.Future] = [] + + async def call(self, method, params=None, *, session_id=None, timeout_s=None): + if method != "Page.captureScreenshot": + return {} + self.captured.append((session_id, params, asyncio.get_running_loop().time())) + answer = asyncio.get_running_loop().create_future() + self.waiting.append(answer) + if len(self.waiting) > 1: + for waiting in self.waiting: + if not waiting.done(): + waiting.set_result({"data": base64.b64encode(b"behind").decode()}) + self.waiting = [] + return await asyncio.wait_for(answer, timeout_s) + + +@pytest.mark.asyncio +async def test_a_tab_behind_the_active_one_is_framed_once_a_second_capture_nudges_it(): + pipe = BehindPipe() + control = BrowserControl(pipe, "agent", (1280, 800)) + pipe.emit("Target.targetCreated", {"targetInfo": {"targetId": "T1", "type": "page", "url": "about:blank"}}) + pipe.emit("Target.attachedToTarget", {"sessionId": "s-T1", "targetInfo": {"targetId": "T1", "type": "page"}}) + + data, _, tab, _ = await control.frame("T1", load_wait_s=0) + + (_, _, first), (_, _, second) = pipe.captured + assert (base64.b64decode(data), tab) == (b"behind", "T1") and second - first >= NUDGE_S + + +@pytest.mark.asyncio +async def test_a_tab_that_closes_while_its_page_loads_is_framed_as_the_active_tab(): + pipe, control = _two_tabs() + pipe.emit("Page.frameNavigated", {"frame": {"id": "T2", "url": "https://a.test/pop", "loaderId": "L2"}}, "s-T2") + asyncio.get_running_loop().call_later(0.1, pipe.emit, "Target.detachedFromTarget", {"sessionId": "s-T2"}) + + _, _, tab, _ = await control.frame("T2", load_wait_s=5) + + assert tab != "T2" and [session for session, _, _ in pipe.captured] == [f"s-{tab}"] diff --git a/src/agent_env/providers/env_providers/env_computer_provider.py b/src/agent_env/providers/env_providers/env_computer_provider.py index 1322c06ba..d6fac8032 100644 --- a/src/agent_env/providers/env_providers/env_computer_provider.py +++ b/src/agent_env/providers/env_providers/env_computer_provider.py @@ -14,7 +14,7 @@ from agentenv_protocol.computer import COMPUTER_DISPLAY, COMPUTER_RECORDING, tool_prefix_problem from agentenv_protocol.computer.app import COMPUTER_APP from agentenv_protocol.computer.app_runtime import HOLD_BOOT_LAUNCH_ENV -from agentenv_protocol.website_browser import BROWSER, LOADED, start_tab_why +from agentenv_protocol.website_browser import BROWSER, BROWSER_RECORDING, LOADED, start_tab_why from agent_env.artifact.artifacts.computer import ComputerArtifact from agent_env.attribution import Attribution @@ -336,14 +336,24 @@ def _refuse_failed_start_tabs(record: DeployedSandboxEnv, start_tabs: dict[str, async def end_recordings(record: DeployedSandboxEnv, reason: str) -> None: - """Have the computer's driver end every recording not yet ended, each uploading what's left and its manifest under - the grant it was started with, before the computer goes; a driver that doesn't record has none.""" + """Have the computer's driver, and each browser app on it that records, end every recording not yet ended, each + uploading what's left and its manifest under the grant it was started with, before the computer goes; a driver or + app that doesn't record has none.""" try: await record.invoke(COMPUTER_RECORDING, "end_all", {"reason": reason}, timeout=_END_RECORDINGS_TIMEOUT_S) except EnvCapabilityUnsupported: - return + pass except Exception as e: logger.warning(f"'{record.env_id}': its recordings weren't ended before its computer went: {e}") + for child in (record.environment_card or {}).get("children_environments") or []: + if protocol_v1.find_extension(child, BROWSER_RECORDING) is None: + continue + try: + await record.invoke(BROWSER_RECORDING, "end_all", {"reason": reason}, environment_name=child.get("name"), + timeout=_END_RECORDINGS_TIMEOUT_S) + except Exception as e: + logger.warning(f"'{record.env_id}': {child.get('name')}'s browser recordings weren't ended before its " + f"computer went: {e}") def _reaching(policy: Optional[NetworkPolicy], sandbox_provider: SandboxProvider, url: str, diff --git a/src/agent_env/task/episode/rules.py b/src/agent_env/task/episode/rules.py index 0bfb1d51d..ab3791635 100644 --- a/src/agent_env/task/episode/rules.py +++ b/src/agent_env/task/episode/rules.py @@ -152,7 +152,8 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s "aggregate_verifiers", "apply_server_config", "browser_activate_tab", "browser_clear_local_storage", "browser_close_tab", "browser_delete_cookies", "browser_open_tab", "browser_set_cookies", "browser_sign_in", "browser_set_local_storage", "build_mcp_cli", "combine_universe_verdicts", "configure_scheduled_checks", - "deploy_env", "deploy_sandbox", "end_computer_recording", "start_computer_recording", "env_outcome_verifier", + "deploy_env", "deploy_sandbox", "end_browser_recording", "end_computer_recording", "start_browser_recording", + "start_computer_recording", "env_outcome_verifier", "modify_env_tool_access", "reset_env", "review", "run_container_unit_tests_verifier", "run_docker_container", "snapshot_env", "snapshot_computer_env", "sync_env_clock", "validation_gate_aggregator", "verify_env_card", "verify_env_core_protocol", "verify_mcp_tool_schema", diff --git a/src/agent_env/task/teardown.py b/src/agent_env/task/teardown.py index a2bc14ffd..be07a461d 100644 --- a/src/agent_env/task/teardown.py +++ b/src/agent_env/task/teardown.py @@ -31,6 +31,8 @@ from agent_env.task.interrupts import tearing_down from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps.computer_recording import end_open_recordings, register_open_recordings +from agent_env.task_step.task_steps.browser.recording import (end_open_browser_recordings, + register_open_browser_recordings) logger = logging.getLogger(__name__) @@ -116,8 +118,8 @@ async def teardown_run(context: TaskStepContext, *, timeout: float = TERMINATE_T async def tear_down(context: TaskStepContext, sandboxes: Sequence[RecordedSandbox], envs: Sequence[DeployedEnv], *, timeout: float) -> TeardownReport: - """End the recordings the run left open on ``envs`` with fresh grants, close each env still up, register any - recording still open from its namespace, then terminate each of ``sandboxes`` by id, all at once and each within + """End the recordings the run left open on ``envs``, its computers' and its browsers', with fresh grants, close each + env still up, register any recording still open from its namespace, then terminate each of ``sandboxes`` by id, all at once and each within ``timeout`` seconds, and remove a local one's work folder once it's down. A close that fails is logged: its env's sandboxes are terminated by id regardless, which leaves one the close took down as it is. Cancelling it returns what it reached, with the rest in ``left``.""" pending = _pending(context, sandboxes) @@ -127,11 +129,13 @@ async def tear_down(context: TaskStepContext, sandboxes: Sequence[RecordedSandbo try: async with asyncio.timeout(timeout): await end_open_recordings(context, env_ids) + await end_open_browser_recordings(context, env_ids) except TimeoutError: logger.warning("teardown: the run's open recordings weren't all ended within %gs", timeout) await asyncio.gather(*(_close(env, timeout) for env in envs if any(sandbox_id in up for sandbox_id, _ in env_sandbox_ids(env)))) await asyncio.to_thread(register_open_recordings, context, env_ids) + await asyncio.to_thread(register_open_browser_recordings, context, env_ids) except asyncio.CancelledError: return TeardownReport(left=pending) terminated, failed = [], [] diff --git a/src/agent_env/task_step/registry.py b/src/agent_env/task_step/registry.py index d428aeb41..814779a47 100644 --- a/src/agent_env/task_step/registry.py +++ b/src/agent_env/task_step/registry.py @@ -44,6 +44,10 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: EndComputerRecordingTaskStep, StartComputerRecordingTaskStep, ) + from agent_env.task_step.task_steps.browser.recording import ( + EndBrowserRecordingTaskStep, + StartBrowserRecordingTaskStep, + ) from agent_env.task_step.task_steps.mcp_cli_builder.build_mcp_cli import BuildMcpCliTaskStep from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep @@ -111,6 +115,8 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: _get_type(CollectArtifactsTaskStep): CollectArtifactsTaskStep, _get_type(StartComputerRecordingTaskStep): StartComputerRecordingTaskStep, _get_type(EndComputerRecordingTaskStep): EndComputerRecordingTaskStep, + _get_type(StartBrowserRecordingTaskStep): StartBrowserRecordingTaskStep, + _get_type(EndBrowserRecordingTaskStep): EndBrowserRecordingTaskStep, _get_type(DeployAgentTaskStep): DeployAgentTaskStep, _get_type(DeployEnvTaskStep): DeployEnvTaskStep, _get_type(DeployHumanAgentTaskStep): DeployHumanAgentTaskStep, diff --git a/src/agent_env/task_step/task_steps/browser/recording.py b/src/agent_env/task_step/task_steps/browser/recording.py new file mode 100644 index 000000000..c0eca067e --- /dev/null +++ b/src/agent_env/task_step/task_steps/browser/recording.py @@ -0,0 +1,251 @@ +"""Record a website browser. ``start_browser_recording`` starts browser-recording/v1 on the browser in a deployed env, +with frames by default, under a grant to upload its frames, log and manifest in a namespace of its own; +``end_browser_recording`` ends it and registers them as one FileArtifactUniverse, and fails when its log didn't upload +or, with frames, when a frame is missing: an infrastructure failure. Teardown ends what a run left open, with fresh +grants, before any env closes (``end_open_browser_recordings``), and registers from its namespace one it couldn't end +(``register_open_browser_recordings``).""" +from __future__ import annotations + +import json +import logging +import uuid +from typing import ClassVar, Optional + +from agentenv_protocol.client import find_extension +from agentenv_protocol.website_browser import BROWSER_RECORDING, BrowserRecordingManifest, RecordedFrame +from agentenv_protocol.website_browser.recording import LOG, MANIFEST, MAX_FRAMES, frame_of + +from agent_env.artifact import FileArtifact +from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse +from agent_env.artifact.store import get_artifact_store +from agent_env.config import get_config +from agent_env.entity.refs import EntityRef +from agent_env.env.env import DeployedEnv, EnvCapabilityUnsupported +from agent_env.store.ids import derive_id, is_local_id, validate_local_id +from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.task_step import TaskStep, TaskStepDependency +from agent_env.task_step.task_steps.browser.base import BrowserStep, deployed_env, find_browser +from agent_env.task_step.task_steps.recordings import ( + ERROR_CHARS, + end_open, + recording_grant, + recording_namespace, + refuse_no_network, + register_open, + why, +) + +logger = logging.getLogger(__name__) + +#: ``context.metadata[RECORDINGS_KEY][name]``: each browser recording a run started, open or ended. +RECORDINGS_KEY = "browser_recordings" +_KEY_PREFIX = "agentenv-browser-recordings" +#: A grant's limits: each frame or the log at most this big, and the whole recording at most this. +MAX_OBJECT_BYTES = 256 << 20 +MAX_RECORDING_BYTES = 16 << 30 +#: What a recording uploads besides its frames, which its grant has room for too. +BESIDE_FRAMES = (LOG, MANIFEST) + + +class StartBrowserRecordingTaskStep(BrowserStep): + """Start recording the website browser in the env ``env_id`` (the one ``browser`` names, else its only one) under + ``name``: with ``frames``, a frame of the tab an agent last acted in at the start, after each burst of its actions + and at the end. A start that fails fails the task unless ``fail_task_on_error`` is off.""" + + type: ClassVar[str] = "start_browser_recording" + entity_refs = (EntityRef.env("env_id"),) + + def __init__(self, id: str, version: Optional[int], env_id: str, name: str, frames: bool = True, + browser: Optional[str] = None, depends_on: Optional[list[TaskStepDependency]] = None, + fail_task_on_error: bool = True): + super().__init__(id, version, env_id, browser, depends_on=depends_on, fail_task_on_error=fail_task_on_error) + if not name: + raise ValueError("start_browser_recording needs a name for the recording") + self.name, self.frames = name, frames + + def to_dict(self) -> dict: + return {**super().to_dict(), "name": self.name, "frames": self.frames} + + @classmethod + def from_dict(cls, data: dict) -> StartBrowserRecordingTaskStep: + return cls(**cls._browser_from_dict(data), name=data["name"], frames=data.get("frames", True)) + + async def execute(self, context: TaskStepContext) -> TaskStepContext: + recordings = context.metadata.setdefault(RECORDINGS_KEY, {}) + if (recordings.get(self.name) or {}).get("state") == "open": + raise RuntimeError(f"a browser recording named {self.name!r} is already open in this run; give this one " + "another name") + # A retried step joins its first attempt's recording, so it keeps that attempt's namespace too. + request_id = f"{context.instance_id}:{self.id}" if context.instance_id else uuid.uuid4().hex + try: + record, browser, card = self._find(context) + refuse_no_network(record) + namespace = recording_namespace(request_id, _KEY_PREFIX) + started = await record.invoke(BROWSER_RECORDING, "start", { + "request_id": request_id, "frames": self.frames, "upload": _grant(record, card, namespace)}, + environment_name=browser) + except Exception as e: + recordings[self.name] = {"env_id": self.env_id, "browser": self.browser, "frames": self.frames, + "state": "failed", "error": why(e)} + logger.warning("start_browser_recording %s on %s: %s", self.name, self.env_id, e) + if self.fail_task_on_error: + raise + return context + recordings[self.name] = { + "env_id": self.env_id, "browser": browser, "frames": self.frames, "recording_id": started["recording_id"], + "started_at": started["started_at"], "namespace": namespace, "state": "open"} + logger.info("start_browser_recording %s on %s: %s", self.name, self.env_id, started["recording_id"]) + return context + + +class EndBrowserRecordingTaskStep(TaskStep): + """End the browser recording ``name`` and register it; fail, unless ``fail_task_on_error`` is off, when it didn't + end, its log didn't upload, or, with frames, a frame is missing.""" + + type: ClassVar[str] = "end_browser_recording" + entity_refs = () + + def __init__(self, id: str, version: Optional[int], name: str, + depends_on: Optional[list[TaskStepDependency]] = None, fail_task_on_error: bool = True): + super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) + if not name: + raise ValueError("end_browser_recording needs the name its start_browser_recording gave") + self.name = name + + def to_dict(self) -> dict: + return {**super().to_dict(), "name": self.name} + + @classmethod + def from_dict(cls, data: dict) -> EndBrowserRecordingTaskStep: + return cls(**cls._base_from_dict(data), name=data["name"]) + + async def execute(self, context: TaskStepContext) -> TaskStepContext: + entry = (context.metadata.get(RECORDINGS_KEY) or {}).get(self.name) + if entry is None: + raise RuntimeError(f"no browser recording named {self.name!r} was started in this run") + await end_browser_recording(context, self.name, "end") + if self.fail_task_on_error and (problem := incomplete(entry)): + raise RuntimeError(f"browser recording {self.name!r} is incomplete, an infrastructure failure: {problem}") + return context + + +def incomplete(entry: dict) -> Optional[str]: + """What's missing from the recording ``entry`` records: that it didn't end, its log or entries of it, or its + frames; None for a complete one.""" + if entry.get("state") != "ended": + return f"it didn't end: {entry.get('error')}" + if not entry.get("log_uploaded"): + return f"its log didn't upload: {entry.get('error')}" + if entry.get("log_missing"): + return f"its log is missing {entry['log_missing']} of the entries between its start and its end" + if entry.get("frames") and entry.get("missing_frames"): + return f"{entry['missing_frames']} of its {entry['frames_taken'] + entry['missing_frames']} frames are missing" + return None + + +async def end_open_browser_recordings(context: TaskStepContext, env_ids: Optional[list[str]] = None) -> None: + """End and register every browser recording still open in ``context`` (of ``env_ids`` only, when given), each with + a fresh grant, before its env is torn down; ``end_browser_recording`` says what happened to each.""" + await end_open(context, RECORDINGS_KEY, env_ids, end_browser_recording) + + +def register_open_browser_recordings(context: TaskStepContext, env_ids: Optional[list[str]] = None) -> None: + """Register every browser recording still open in ``context`` (of ``env_ids`` only, when given) from what its + namespace holds, once its env has closed. Never raises: each entry says what happened.""" + register_open(context, RECORDINGS_KEY, env_ids, _from_namespace, _registered, "browser recording") + + +async def end_browser_recording(context: TaskStepContext, name: str, reason: str) -> None: + """End the open browser recording ``name`` with a fresh grant and register it; when the browser no longer knows + it, as after its server restarted, what its namespace holds is registered instead. Never raises: the entry says + what happened.""" + entry = context.metadata[RECORDINGS_KEY][name] + if entry.get("state") != "open": + return + try: + record, browser, card = find_browser(deployed_env(context, entry["env_id"]), entry["browser"]) + manifest = BrowserRecordingManifest.model_validate(await record.invoke(BROWSER_RECORDING, "end", { + "recording_id": entry["recording_id"], "upload": _grant(record, card, entry["namespace"], "end"), + "reason": reason}, environment_name=browser)) + except Exception as e: + logger.warning("end_browser_recording %s: %s; registering what its namespace holds", name, e) + entry["error"] = why(e) + try: + manifest = _from_namespace(entry) + except Exception as listing: + logger.warning("end_browser_recording %s: listing its namespace failed: %s", name, listing) + entry["state"], entry["error"] = "failed", f"{entry['error']}; then {why(listing)}"[:ERROR_CHARS] + return + _registered(context, name, entry, manifest) + + +def _grant(record: DeployedEnv, card: dict, namespace: str, method: str = "start") -> dict: + """A grant for a browser recording's frames, log and manifest; EnvCapabilityUnsupported for a browser whose card + declares no recording.""" + extension = find_extension(card or {}, BROWSER_RECORDING) + if extension is None: + raise EnvCapabilityUnsupported(BROWSER_RECORDING, method, record.env_id, getattr(record, "instance_id", None), + card.get("name")) + params = extension.get("params") or {} + return recording_grant(record, params, namespace, params.get("max_frames", MAX_FRAMES) + len(BESIDE_FRAMES), + MAX_OBJECT_BYTES, MAX_RECORDING_BYTES) + + +def _registered(context: TaskStepContext, name: str, entry: dict, manifest: BrowserRecordingManifest) -> None: + """Register the recording ``name`` as ``manifest`` says it went, and its entry as ended, or as failed.""" + try: + entry["artifact"] = _register(context, name, entry["namespace"], manifest) + except Exception as e: + logger.warning("browser recording %s: registering it failed: %s", name, e) + entry["state"], entry["error"] = "failed", why(e) + return + taken = [frame for frame in manifest.frames if frame.uploaded] + entry.update(state="ended", ended_by=manifest.ended_by, frames_taken=len(taken), + missing_frames=manifest.missing_frames, log_uploaded=manifest.log_uploaded, + log_missing=len(manifest.log_missing)) + if manifest.error: + entry["error"] = manifest.error + logger.info("browser recording %s: %d frames, %d missing, ended by %s", name, len(taken), manifest.missing_frames, + manifest.ended_by) + + +def _from_namespace(entry: dict) -> BrowserRecordingManifest: + """What the recording's namespace holds, for one its browser didn't end for the run: the manifest the browser + uploaded when it ended it, or else its frames there, each by the index its file is named by, its kind and when it + was taken unknown, and its log if that uploaded.""" + store = get_config().get_object_store() + root = f"{entry['namespace'].rstrip('/')}/{entry.get('recording_id')}/" + urls = store.list_at(root) + if (uploaded := f"{root}{MANIFEST}") in urls: + return BrowserRecordingManifest.model_validate(json.loads(store.get(uploaded))) + frames = sorted((found, url[len(root):]) for url in urls if (found := frame_of(url[len(root):]))) + logged = f"{root}{LOG}" in urls + return BrowserRecordingManifest( + recording_id=entry.get("recording_id") or "", started_at=entry.get("started_at") or "", ended_by="error", + error=entry.get("error"), frames_on=bool(entry.get("frames")), log_uploaded=logged, log=LOG if logged else None, + frames=[RecordedFrame(index=index, path=path, content_type=content_type, uploaded=True) + for (index, content_type), path in frames]) + + +def _register(context: TaskStepContext, name: str, namespace: str, manifest: BrowserRecordingManifest) -> dict: + """The manifest, the log and the uploaded frames as FileArtifacts, pointing at the objects, in one universe.""" + universe_id = derive_id(context.instance_id or "run", f"browser-recording-{name}") + if is_local_id(universe_id): + validate_local_id(universe_id) + objects, documents = get_config().get_object_store(), get_artifact_store() + root = f"{namespace.rstrip('/')}/{manifest.recording_id}" + manifest_url = objects.put(objects.get_object_key(f"{root}/{MANIFEST}"), + json.dumps(manifest.model_dump(mode="json"), indent=1).encode(), + content_type="application/json", allow_overwrite=True) + files = {MANIFEST: (manifest_url, "application/json")} + if manifest.log_uploaded and manifest.log: + files[manifest.log] = (f"{root}/{manifest.log}", "application/x-ndjson") + for frame in manifest.frames: + if frame.uploaded and frame.path: + files[frame.path.replace("/", "-")] = (f"{root}/{frame.path}", frame.content_type) + stored = {filename: documents.put_document(FileArtifact( + id=derive_id(universe_id, filename), filename=filename, content_type=content_type, object_url=url, + description=f"Browser recording {name!r} of task instance {context.instance_id}")) + for filename, (url, content_type) in files.items()} + universe = FileArtifactUniverse.put(id=universe_id, file_artifacts=stored) + return {"id": universe.id, "version": universe.version} diff --git a/src/agent_env/task_step/task_steps/computer_recording.py b/src/agent_env/task_step/task_steps/computer_recording.py index 7044fdda8..591a04aaf 100644 --- a/src/agent_env/task_step/task_steps/computer_recording.py +++ b/src/agent_env/task_step/task_steps/computer_recording.py @@ -17,27 +17,30 @@ from agentenv_protocol.computer.recording import MANIFEST from agentenv_protocol.client import find_extension -from agent_env.a2a_agent.object_transfer import ObjectLimits, namespace_grant from agent_env.artifact import FileArtifact from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse from agent_env.artifact.store import get_artifact_store from agent_env.config import get_config from agent_env.entity.refs import EntityRef from agent_env.env.env import EnvCapabilityUnsupported -from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy -from agent_env.store.base import GrantUnavailableError from agent_env.store.ids import derive_id, is_local_id, validate_local_id -from agent_env.store.object_store.object_store import issues_grants_to from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_step import TaskStep, TaskStepDependency +from agent_env.task_step.task_steps.recordings import ( + ERROR_CHARS, + end_open, + recording_grant, + recording_namespace, + refuse_no_network, + register_open, + why, +) logger = logging.getLogger(__name__) #: ``context.metadata[RECORDINGS_KEY][name]``: each recording a run started, open or ended. RECORDINGS_KEY = "computer_recordings" _KEY_PREFIX = "agentenv-recordings" -#: How much longer than the env's ``max_duration_s`` a recording's grant lasts. -GRANT_MARGIN_S = 600 _MAX_SEGMENT_BYTES = 5 << 30 #: The segments a grant takes: a recording also rolls over early, on a size change or after a gap. MAX_SEGMENTS = 10_000 @@ -88,14 +91,14 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: request_id = f"{context.instance_id}:{self.id}" if context.instance_id else uuid.uuid4().hex try: deployed = _deployed(context, self.env_id) - _refuse_no_network(deployed) - namespace = _namespace(request_id) + refuse_no_network(deployed) + namespace = recording_namespace(request_id, _KEY_PREFIX) grant = _grant(deployed, namespace, self.environment_name) started = await deployed.invoke(COMPUTER_RECORDING, "start", { "request_id": request_id, "fps": self.fps, "max_edge": self.max_edge, "cursor": self.cursor, "upload": grant}, environment_name=self.environment_name) except Exception as e: - recordings[self.name] = {"env_id": self.env_id, "state": "failed", "error": f"{type(e).__name__}: {e}"[:500]} + recordings[self.name] = {"env_id": self.env_id, "state": "failed", "error": why(e)} logger.warning("start_computer_recording %s on %s: %s", self.name, self.env_id, e) if self.fail_task_on_error: raise @@ -148,9 +151,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: async def end_open_recordings(context: TaskStepContext, env_ids: Optional[list[str]] = None) -> None: """End and register every recording still open in ``context`` (of ``env_ids`` only, when given), each with a fresh grant, before its computer is torn down, whatever env serves it; ``end_recording`` says what happened to each.""" - for name, entry in list((context.metadata.get(RECORDINGS_KEY) or {}).items()): - if entry.get("state") == "open" and (env_ids is None or entry["env_id"] in env_ids): - await end_recording(context, name, "teardown") + await end_open(context, RECORDINGS_KEY, env_ids, end_recording) def register_open_recordings(context: TaskStepContext, env_ids: Optional[list[str]] = None) -> None: @@ -158,15 +159,7 @@ def register_open_recordings(context: TaskStepContext, env_ids: Optional[list[st holds: the manifest its driver uploaded when its env ended it, or else the segments alone. Teardown calls it once the envs have closed, for one ``end_open_recordings`` didn't end in time. Never raises: each entry says what happened.""" - for name, entry in list((context.metadata.get(RECORDINGS_KEY) or {}).items()): - if entry.get("state") == "open" and (env_ids is None or entry["env_id"] in env_ids): - try: - manifest = _from_namespace(entry) - except Exception as e: - logger.warning("recording %s: listing its namespace failed: %s", name, e) - entry["state"], entry["error"] = "failed", f"{type(e).__name__}: {e}"[:500] - continue - _registered(context, name, entry, manifest) + register_open(context, RECORDINGS_KEY, env_ids, _from_namespace, _registered, "recording") async def end_recording(context: TaskStepContext, name: str, reason: str) -> None: @@ -183,12 +176,12 @@ async def end_recording(context: TaskStepContext, name: str, reason: str) -> Non "reason": reason}, environment_name=environment_name)) except Exception as e: logger.warning("end_computer_recording %s: %s; registering what its namespace holds", name, e) - entry["error"] = f"{type(e).__name__}: {e}"[:500] + entry["error"] = why(e) try: manifest = _from_namespace(entry) except Exception as listing: logger.warning("end_computer_recording %s: listing its namespace failed: %s", name, listing) - entry["state"], entry["error"] = "failed", f"{entry['error']}; then {type(listing).__name__}: {listing}"[:500] + entry["state"], entry["error"] = "failed", f"{entry['error']}; then {why(listing)}"[:ERROR_CHARS] return _registered(context, name, entry, manifest) @@ -199,7 +192,7 @@ def _registered(context: TaskStepContext, name: str, entry: dict, manifest: Reco entry["artifact"] = _register(context, name, entry["namespace"], manifest) except Exception as e: logger.warning("recording %s: registering it failed: %s", name, e) - entry["state"], entry["error"] = "failed", f"{type(e).__name__}: {e}"[:500] + entry["state"], entry["error"] = "failed", why(e) return uploaded = [s for s in manifest.segments if s.uploaded] entry.update(state="ended", ended_by=manifest.ended_by, segments=len(uploaded), @@ -214,43 +207,16 @@ def _deployed(context: TaskStepContext, env_id: str): return deployed -def _refuse_no_network(deployed) -> None: - """A computer deployed with no network can't reach the store its recording's segments upload to.""" - network = ((deployed.metadata or {}).get("computer") or {}).get("network") - if network and NetworkPolicy.from_dict(network).closed: - raise RuntimeError(f"env {deployed.env_id!r} was deployed with no network, so its computer can't upload a " - "recording's segments") - - -def _namespace(request_id: str) -> str: - """Where a recording's segments go: named by its start's request id, so a retried start uploads alongside.""" - config = get_config() - key = uuid.uuid5(uuid.NAMESPACE_URL, request_id).hex - return config.get_object_store().object_url(f"{config.get_artifact_key_prefix()}{_KEY_PREFIX}/{key}") - - def _grant(deployed, namespace: str, environment_name: Optional[str] = None, method: str = "start") -> dict: - """A grant that uploads under ``namespace``, lasting the recorded env's ``max_duration_s`` and a margin, or the - longest the store issues. EnvCapabilityUnsupported when the env's card, or its child's, declares no recording.""" + """A grant for a computer recording's segments and manifest. EnvCapabilityUnsupported when the env's card, or its + child's, declares no recording.""" card = deployed.environment_card if environment_name is None else deployed.get_child_env_card(environment_name) extension = find_extension(card or {}, COMPUTER_RECORDING) if extension is None: raise EnvCapabilityUnsupported(COMPUTER_RECORDING, method, deployed.env_id, getattr(deployed, "instance_id", None), environment_name) - params = extension.get("params") or {} - store = get_config().get_object_store() - sandbox_type = getattr(deployed, "sandbox_type", None) - if not issues_grants_to(store, sandbox_type): - raise RuntimeError(f"the object store issues no upload grant that reaches env {deployed.env_id}'s sandbox " - f"({sandbox_type or 'unknown'}), so its recording couldn't upload") - duration = params["max_duration_s"] - cap = min(store.max_single_upload_bytes or _MAX_SEGMENT_BYTES, _MAX_SEGMENT_BYTES) - limits = ObjectLimits(max_objects=MAX_SEGMENTS + 1, max_object_bytes=cap, max_total_bytes=MAX_RECORDING_BYTES) - try: - grant = namespace_grant(store, namespace, limits=limits, expires_in=duration + GRANT_MARGIN_S) - except GrantUnavailableError: - grant = namespace_grant(store, namespace, limits=limits, expires_in=store.grant_lifetime_seconds) - return grant.model_dump(mode="json") + return recording_grant(deployed, extension.get("params") or {}, namespace, MAX_SEGMENTS + 1, _MAX_SEGMENT_BYTES, + MAX_RECORDING_BYTES) def _from_namespace(entry: dict) -> RecordingManifest: diff --git a/src/agent_env/task_step/task_steps/recordings.py b/src/agent_env/task_step/task_steps/recordings.py new file mode 100644 index 000000000..f6edaa68a --- /dev/null +++ b/src/agent_env/task_step/task_steps/recordings.py @@ -0,0 +1,93 @@ +"""What a run's recordings share, a computer's (``computer_recording``) and a website browser's +(``browser.recording``): where each uploads and the grant it uploads under, the error its entry keeps, and teardown's +two passes over the ones a run left open, the first ending them, the second registering from its namespace any still +open once the envs have closed.""" +from __future__ import annotations + +import logging +import uuid +from collections.abc import Awaitable, Callable +from typing import Any, Optional + +from agent_env.a2a_agent.object_transfer import ObjectLimits, namespace_grant +from agent_env.config import get_config +from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy +from agent_env.store.base import GrantUnavailableError +from agent_env.store.object_store.object_store import issues_grants_to +from agent_env.task_step.context import TaskStepContext + +logger = logging.getLogger(__name__) + +#: How much longer than the env's ``max_duration_s`` a recording's grant lasts. +GRANT_MARGIN_S = 600 +#: How much of an error a recording's entry keeps. +ERROR_CHARS = 500 + + +def why(error: BaseException) -> str: + """``error`` as a recording's entry keeps it: its type and message, cut to ``ERROR_CHARS``.""" + return f"{type(error).__name__}: {error}"[:ERROR_CHARS] + + +def refuse_no_network(deployed) -> None: + """A computer deployed with no network can't reach the store its recordings upload to.""" + network = ((deployed.metadata or {}).get("computer") or {}).get("network") + if network and NetworkPolicy.from_dict(network).closed: + raise RuntimeError(f"env {deployed.env_id!r} was deployed with no network, so its computer can't upload a " + "recording") + + +def recording_namespace(request_id: str, prefix: str) -> str: + """Where a recording's files go, under ``prefix``: named by its start's request id, so a retried start uploads + alongside.""" + config = get_config() + key = uuid.uuid5(uuid.NAMESPACE_URL, request_id).hex + return config.get_object_store().object_url(f"{config.get_artifact_key_prefix()}{prefix}/{key}") + + +def recording_grant(deployed, params: dict, namespace: str, max_objects: int, max_object_bytes: int, + max_total_bytes: int) -> dict: + """A grant that uploads under ``namespace``, within the limits given and the store's own, lasting the recording + extension's ``max_duration_s`` (``params``) and a margin, or the longest the store issues; refused when the store + issues no grant that reaches the env's sandbox.""" + store = get_config().get_object_store() + sandbox_type = getattr(deployed, "sandbox_type", None) + if not issues_grants_to(store, sandbox_type): + raise RuntimeError(f"the object store issues no upload grant that reaches env {deployed.env_id}'s sandbox " + f"({sandbox_type or 'unknown'}), so its recording couldn't upload") + limits = ObjectLimits(max_objects=max_objects, max_object_bytes=min(store.max_single_upload_bytes or max_object_bytes, + max_object_bytes), + max_total_bytes=max_total_bytes) + try: + grant = namespace_grant(store, namespace, limits=limits, expires_in=params["max_duration_s"] + GRANT_MARGIN_S) + except GrantUnavailableError: + grant = namespace_grant(store, namespace, limits=limits, expires_in=store.grant_lifetime_seconds) + return grant.model_dump(mode="json") + + +def open_entries(context: TaskStepContext, key: str, env_ids: Optional[list[str]]) -> list[tuple[str, dict]]: + """The recordings ``context.metadata[key]`` holds that are still open, by name, of ``env_ids`` only when given.""" + return [(name, entry) for name, entry in (context.metadata.get(key) or {}).items() + if entry.get("state") == "open" and (env_ids is None or entry["env_id"] in env_ids)] + + +async def end_open(context: TaskStepContext, key: str, env_ids: Optional[list[str]], + end: Callable[[TaskStepContext, str, str], Awaitable[None]]) -> None: + """``end`` each recording still open, for teardown.""" + for name, _ in open_entries(context, key, env_ids): + await end(context, name, "teardown") + + +def register_open(context: TaskStepContext, key: str, env_ids: Optional[list[str]], + from_namespace: Callable[[dict], Any], registered: Callable[[TaskStepContext, str, dict, Any], None], + label: str) -> None: + """Register each recording still open from what its namespace holds, once its env has closed: ``from_namespace`` + reads its manifest there and ``registered`` registers it. Never raises: each entry says what happened.""" + for name, entry in open_entries(context, key, env_ids): + try: + manifest = from_namespace(entry) + except Exception as e: + logger.warning("%s %s: listing its namespace failed: %s", label, name, e) + entry["state"], entry["error"] = "failed", why(e) + continue + registered(context, name, entry, manifest) diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index 7d5706ab5..a6e9f246b 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -40,6 +40,7 @@ import asyncio import calendar +import contextlib import itertools import json import os @@ -70,7 +71,7 @@ import agent_env from agent_env.a2a_agent import conversation_store -from agent_env.artifact import EnvironmentUniverseArtifact +from agent_env.artifact import Artifact, EnvironmentUniverseArtifact from agent_env.artifact.artifacts.environment import EnvironmentArtifact from agent_env.artifact.artifacts.file import FileArtifact from agent_env.config import get_config @@ -84,12 +85,15 @@ from agent_env.env.store import get_env_store from agent_env.providers.env_providers.constants import AGENT_ENV_WEBSITE_BACKEND_PORT, AGENT_ENV_WEBSITE_BACKEND_SUFFIX from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandboxProvider +from agent_env.store.object_store.local.grant_server import grant_server from agent_env.store.object_store.local.tls import LocalCA, local_ca from agent_env.store.routing import namespace_routing from agent_env.task import Task from agent_env.task.teardown import teardown_run from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps.browser.connect_cdp import BrowserConnectCdpTaskStep +from agent_env.task_step.task_steps.browser.recording import (RECORDINGS_KEY as BROWSER_RECORDINGS, + EndBrowserRecordingTaskStep, StartBrowserRecordingTaskStep) from agent_env.task_step.task_steps.browser.sign_in import BrowserSignInTaskStep from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep @@ -607,22 +611,34 @@ async def test_its_log_has_an_entry_for_every_action_over_cdp_page_tools_and_pix 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: +@contextlib.asynccontextmanager +async def _cdp_calls(ws_url: str, tls: ssl.SSLContext | None = None, reply_s: float = 60): + """A CDP connection to ``ws_url``, trusting ``tls`` for a wss address, as its ``call``: a command's whole reply, + within ``reply_s``, the events before it skipped.""" + async with connect(ws_url, max_size=None, open_timeout=60, **({"ssl": tls} if tls else {})) 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: + while (reply := json.loads(await asyncio.wait_for(ws.recv(), reply_s))).get("id") != message_id: pass return reply + yield call + + +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.""" + sent: list[str] = [] + async with _cdp_calls(ws_url) as cdp: + + async def call(method: str, params: dict, session: str | None = None) -> dict: + sent.append(method) + return await cdp(method, params, session) + target = (await call("Target.createTarget", {"url": FORM}))["result"]["targetId"] session = (await call("Target.attachToTarget", {"targetId": target, "flatten": True}))["result"]["sessionId"] await asyncio.sleep(1) @@ -744,17 +760,7 @@ async def _through_port(deployed, url: str, proxy: str = "http://127.0.0.1:3129" async def _cdp_agent_loads(ws_url: str, urls: list[str], tls: ssl.SSLContext | None = None) -> dict: """What a CDP agent at ``ws_url`` sees navigating a tab of its own to each of ``urls``, one after another: ``Page.navigate``'s error, if any, and the page once it has loaded; trusting ``tls`` for a wss address.""" - async with connect(ws_url, max_size=None, open_timeout=60, **({"ssl": tls} if tls else {})) as ws: - ids = itertools.count(1) - - async def call(method: str, params: dict, session: str | None = None) -> dict: - message_id = next(ids) - 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(), 90))).get("id") != message_id: - pass - return reply - + async with _cdp_calls(ws_url, tls=tls, reply_s=90) as call: target = (await call("Target.createTarget", {"url": "about:blank"}))["result"]["targetId"] session = (await call("Target.attachToTarget", {"targetId": target, "flatten": True}))["result"]["sessionId"] seen = {} @@ -1074,6 +1080,87 @@ async def test_playwright_mcps_tools_drive_its_page_hide_the_code_tools_and_come assert not report.failed +@pytest.mark.asyncio +async def test_a_browser_recording_frames_each_burst_of_an_agents_actions_keyed_to_its_log_and_ends_on_its_last_page( + browser_computer): + with namespace_routing(): + deployed = await Env.get(browser_computer.id).deploy( + env_params={"app_settings": {"chromium": {"start_urls": [PAGE_URL]}}}) + context = TaskStepContext(instance_id=f"run-{uuid.uuid4().hex[:8]}", deployed_envs=[deployed]) + try: + context = await StartBrowserRecordingTaskStep(id="start", version=None, env_id=browser_computer.id, + name="run").execute(context) + await _mcp(deployed, _type_hello) + await _mcp(deployed, lambda session: _called(session, "browser_navigate", {"url": DATA_PAGE})) + context = await EndBrowserRecordingTaskStep(id="end", version=None, name="run").execute(context) + finally: + report = await teardown_run(context) + grant_server(None, None).close() + entry = context.metadata[BROWSER_RECORDINGS]["run"] + universe = Artifact.get(entry["artifact"]["id"], version=entry["artifact"]["version"]) + objects = get_config().get_object_store() + files = {name: objects.get(Artifact.get(artifact_id).object_url) + for name, artifact_id in universe.file_artifact_ids.items()} + + manifest = json.loads(files["manifest.json"]) + log = [json.loads(line) for line in files["log.jsonl"].decode().splitlines()] + calls = {line["action"]["tool"]: line["seq"] for line in log if line.get("action", {}).get("via") == "mcp"} + frames = manifest["frames"] + followed = {seq for frame in frames for seq in frame["after"]} + assert not report.failed and (entry["state"], entry["missing_frames"]) == ("ended", 0), entry + assert frames[0]["kind"] == "first" and frames[-1]["kind"] == "last" and "action" in {f["kind"] for f in frames} + assert {calls["browser_type"], calls["browser_navigate"]} <= followed, (calls, frames) + assert frames[-1]["url"].startswith("data:text/html,") and "Home" in urllib.parse.unquote(frames[-1]["url"]) + assert all(files[frame["path"].replace("/", "-")][:2] == b"\xff\xd8" for frame in frames) + + +@pytest.mark.asyncio +async def test_a_browser_recording_frames_a_cdp_agents_own_tab_behind_the_active_one(browser_computer): + with namespace_routing(): + deployed = await Env.get(browser_computer.id).deploy( + env_params={"app_settings": {"chromium": {"start_urls": [PAGE_URL]}}}) + context = TaskStepContext(instance_id=f"run-{uuid.uuid4().hex[:8]}", deployed_envs=[deployed]) + try: + context = await StartBrowserRecordingTaskStep(id="start", version=None, env_id=browser_computer.id, + name="behind").execute(context) + tab = await _cdp_agent_acts_behind((await _cdp(deployed, "pi"))["ws_url"]) + active = [t for t in (await _browser_op(deployed, "tabs/list", {}))["tabs"] if t.get("active")] + context = await EndBrowserRecordingTaskStep(id="end", version=None, name="behind", + fail_task_on_error=False).execute(context) + finally: + report = await teardown_run(context) + grant_server(None, None).close() + entry = context.metadata[BROWSER_RECORDINGS]["behind"] + universe = Artifact.get(entry["artifact"]["id"], version=entry["artifact"]["version"]) + manifest = json.loads(get_config().get_object_store().get( + Artifact.get(universe.file_artifact_ids["manifest.json"]).object_url)) + + frames = manifest["frames"] + assert active and active[0]["id"] != tab, active + assert not report.failed and (entry["state"], entry["missing_frames"]) == ("ended", 0), frames + assert [f["tab"] for f in frames if f["kind"] == "action"] and frames[-1]["tab"] == tab, frames + assert "Own two" in urllib.parse.unquote(frames[-1]["url"]), frames[-1] + + +#: The pages a CDP agent moves between in a tab of its own behind the active one. +BEHIND = ("data:text/html,Own one

Own one

", "data:text/html,Own two

Own two

") + + +async def _cdp_agent_acts_behind(ws_url: str) -> str: + """A CDP agent, pi, that opens a tab of its own behind the active one and never brings it to the front: it + navigates it, changes its page and navigates it again. The tab.""" + async with _cdp_calls(ws_url) as call: + target = (await call("Target.createTarget", {"url": "about:blank", "background": True}))["result"]["targetId"] + session = (await call("Target.attachToTarget", {"targetId": target, "flatten": True}))["result"]["sessionId"] + await call("Page.navigate", {"url": BEHIND[0]}, session) + await asyncio.sleep(2) + await call("Runtime.evaluate", {"expression": "document.body.append('typed')"}, session) + await asyncio.sleep(2) + await call("Page.navigate", {"url": BEHIND[1]}, session) + await asyncio.sleep(2) + return target + + async def _mcp(deployed, work): """What ``work`` makes of one MCP session on the computer, as a role-less agent opens one.""" async with streamablehttp_client(deployed.mcp_url) as (read, write, _): diff --git a/tst/unit/task_step/browser_recording_test.py b/tst/unit/task_step/browser_recording_test.py new file mode 100644 index 000000000..229f1bf07 --- /dev/null +++ b/tst/unit/task_step/browser_recording_test.py @@ -0,0 +1,210 @@ +"""start_browser_recording and end_browser_recording against a browser computer standing in for its browser: the grant +it's handed through the browser's own card, the artifact its frames, log and manifest become, a recording whose frames +or log are missing failing as an infrastructure failure, what's registered when the browser forgot the recording, and +what a run left open ended at teardown.""" + +from __future__ import annotations + +import json + +import pytest +from agentenv_protocol.computer import COMPUTER_RECORDING +from agentenv_protocol.website_browser import BROWSER, BROWSER_RECORDING + +from agent_env.artifact import Artifact +from agent_env.config import set_object_store +from agent_env.env.env import EnvCapabilityUnsupported +from agent_env.providers.env_providers.env_computer_provider import end_recordings +from agent_env.task import teardown +from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.task_steps.browser.recording import ( + RECORDINGS_KEY, + EndBrowserRecordingTaskStep, + StartBrowserRecordingTaskStep, + end_open_browser_recordings, +) +from tst.util.granting_object_store import GrantingObjectStore + +_BROWSER = {"name": "chromium", "capabilities": {"extensions": [ + {"uri": BROWSER, "params": {"chromium": "156.0.8078.4", "methods": {}}}, + {"uri": BROWSER_RECORDING, "params": {"max_duration_s": 3600, "max_frames": 100, "frames": True, + "methods": {"start": {"method": "POST"}, "end": {"method": "POST"}}}}]}} + + +class _BrowserComputer: + """A browser computer whose chromium records: its end uploads a frame, the log and the manifest under its grant, + with a frame it couldn't take when ``misses``; it ``forgets`` the recording or lists no browser recording.""" + + def __init__(self, store, misses: bool = False, forgets: bool = False, recording: bool = True, + gaps: bool = False) -> None: + children = [_BROWSER if recording else {**_BROWSER, "capabilities": {"extensions": [ + _BROWSER["capabilities"]["extensions"][0]]}}] + self.env_id, self.sandbox_type, self.metadata = "browser", "local", None + self.environment_card = {"name": "browser", "children_environments": children} + self.store, self.misses, self.forgets, self.gaps = store, misses, forgets, gaps + self.calls: list[tuple[str, dict, str]] = [] + + async def invoke(self, uri, method, params=None, environment_name=None, **_): + assert uri == BROWSER_RECORDING + self.calls.append((method, params, environment_name)) + if method == "start": + self.root = params["upload"]["root_path"] + self.store.put(f"{self.root}/rec-1/frames/0000.jpg", b"jpeg-0", content_type="image/jpeg") + return {"recording_id": "rec-1", "started_at": "2026-11-12T10:00:00.000Z"} + if self.forgets: + raise RuntimeError("400 invalid_argument: no recording rec-1 is open here") + root = params["upload"]["root_path"] + frames = [{"index": 0, "kind": "first", "at": "t0", "path": "frames/0000.jpg", "content_type": "image/jpeg", + "uploaded": True}, + {"index": 1, "kind": "last", "at": "t1", "uploaded": not self.misses, + **({"error": "DeviceUnavailable: the tab painted nothing within 3 s"} if self.misses + else {"path": "frames/0001.jpg", "content_type": "image/jpeg"})}] + if not self.misses: + self.store.put(f"{root}/rec-1/frames/0001.jpg", b"jpeg-1", content_type="image/jpeg") + self.store.put(f"{root}/rec-1/log.jsonl", b'{"seq": 1}\n', content_type="application/x-ndjson") + manifest = {"recording_id": "rec-1", "started_at": "2026-11-12T10:00:00.000Z", "ended_by": params["reason"], + "frames": frames, "log": "log.jsonl", "log_after": 0, "log_last": 2, "log_uploaded": True, + "log_missing": [2] if self.gaps else []} + self.store.put(f"{root}/rec-1/manifest.json", json.dumps(manifest).encode(), content_type="application/json") + return manifest + + +@pytest.fixture +def store(local_stores, tmp_path): + granting = GrantingObjectStore(str(tmp_path / "objects")) + set_object_store(granting) + return granting + + +def _context(*envs, instance_id: str = "run-1") -> TaskStepContext: + return TaskStepContext(instance_id=instance_id, deployed_envs=list(envs), metadata={}) + + +def _start(name: str = "run", **kwargs) -> StartBrowserRecordingTaskStep: + return StartBrowserRecordingTaskStep(id=f"start-{name}", version=None, env_id="browser", name=name, **kwargs) + + +def _end(name: str = "run", **kwargs) -> EndBrowserRecordingTaskStep: + return EndBrowserRecordingTaskStep(id=f"end-{name}", version=None, name=name, **kwargs) + + +@pytest.mark.asyncio +async def test_a_browser_recording_is_started_through_its_browsers_card_and_registered_with_its_frames_and_log(store): + computer = _BrowserComputer(store) + context = await _start().execute(_context(computer)) + started = dict(context.metadata[RECORDINGS_KEY]["run"]) + context = await _end().execute(context) + entry = context.metadata[RECORDINGS_KEY]["run"] + + (_, start, browser), (_, end, _) = computer.calls + assert (start["request_id"], start["frames"], browser) == ("run-1:start-run", True, "chromium") + assert start["upload"]["max_objects"] == 102 and "/agentenv-browser-recordings/" in started["namespace"] + assert end["upload"]["root_path"] == start["upload"]["root_path"] and end["reason"] == "end" + assert (entry["state"], entry["frames_taken"], entry["missing_frames"], entry["log_uploaded"]) == \ + ("ended", 2, 0, True) + universe = Artifact.get(entry["artifact"]["id"], version=entry["artifact"]["version"]) + assert sorted(universe.file_artifact_ids) == ["frames-0000.jpg", "frames-0001.jpg", "log.jsonl", "manifest.json"] + frame = Artifact.get(universe.file_artifact_ids["frames-0001.jpg"]) + assert frame.object_url.endswith("/rec-1/frames/0001.jpg") and store.get(frame.object_url) == b"jpeg-1" + + +@pytest.mark.asyncio +async def test_a_missing_frame_fails_the_end_as_an_infrastructure_failure_unless_told_not_to(store): + context = await _start().execute(_context(_BrowserComputer(store, misses=True))) + with pytest.raises(RuntimeError, match="infrastructure failure: 1 of its 2 frames are missing"): + await _end().execute(context) + + tolerated = await _start().execute(_context(_BrowserComputer(store, misses=True), instance_id="run-2")) + tolerated = await _end(fail_task_on_error=False).execute(tolerated) + assert tolerated.metadata[RECORDINGS_KEY]["run"]["missing_frames"] == 1 + + +@pytest.mark.asyncio +async def test_a_log_missing_entries_fails_the_end(store): + context = await _start().execute(_context(_BrowserComputer(store, gaps=True))) + with pytest.raises(RuntimeError, match="its log is missing 1 of the entries"): + await _end().execute(context) + + +@pytest.mark.asyncio +async def test_a_computer_ends_its_browsers_recordings_before_it_goes(): + calls = [] + + class Record: + env_id = "browser" + environment_card = {"name": "browser", "children_environments": [{"name": "page"}, _BROWSER]} + + async def invoke(self, uri, method, params=None, environment_name=None, timeout=None): + calls.append((uri, method, params, environment_name)) + if uri == COMPUTER_RECORDING: + raise EnvCapabilityUnsupported(uri, method, "browser", None, None) + return {"recordings": []} + + await end_recordings(Record(), "reset") + + assert calls[1:] == [(BROWSER_RECORDING, "end_all", {"reason": "reset"}, "chromium")] + + +@pytest.mark.asyncio +async def test_a_recording_the_browser_forgot_is_registered_from_its_namespace_and_fails_for_its_missing_log(store): + computer = _BrowserComputer(store, forgets=True) + context = await _start().execute(_context(computer)) + store.put(f"{computer.root}/rec-1/frames/0003.png", b"png-3", content_type="image/png") + with pytest.raises(RuntimeError, match="its log didn't upload"): + await _end().execute(context) + entry = context.metadata[RECORDINGS_KEY]["run"] + universe = Artifact.get(entry["artifact"]["id"], version=entry["artifact"]["version"]) + manifest = json.loads(store.get(Artifact.get(universe.file_artifact_ids["manifest.json"]).object_url)) + + assert (entry["state"], entry["frames_taken"], entry["log_uploaded"]) == ("ended", 2, False) + assert "no recording rec-1" in entry["error"] + assert [(f["index"], f["kind"], f["at"], f["content_type"]) for f in manifest["frames"]] == [ + (0, None, None, "image/jpeg"), (3, None, None, "image/png")] + + +@pytest.mark.asyncio +async def test_a_browser_without_recording_fails_the_start_saying_so(store): + computer = _BrowserComputer(store, recording=False) + with pytest.raises(Exception, match="browser-recording/v1"): + await _start().execute(_context(computer)) + + context = await _start(fail_task_on_error=False).execute(_context(computer)) + assert context.metadata[RECORDINGS_KEY]["run"]["state"] == "failed" and computer.calls == [] + + +@pytest.mark.asyncio +async def test_what_a_run_left_open_is_ended_with_a_fresh_grant_and_teardown_ends_it_before_envs_close(store, + monkeypatch): + computer = _BrowserComputer(store) + context = await _start().execute(_context(computer)) + await end_open_browser_recordings(context, ["browser"]) + ended = [] + + async def ending(context, env_ids): + ended.append(env_ids) + + monkeypatch.setattr(teardown, "end_open_browser_recordings", ending) + await teardown.tear_down(context, [], [], timeout=30) + + (_, start, _), (_, end, _) = computer.calls + entry = context.metadata[RECORDINGS_KEY]["run"] + assert end["reason"] == "teardown" and end["upload"] != start["upload"] and entry["state"] == "ended" + assert ended == [[]] + + +@pytest.mark.asyncio +async def test_a_start_reusing_an_open_recordings_name_is_refused(): + context = TaskStepContext(instance_id="run-1", metadata={RECORDINGS_KEY: {"run": {"state": "open"}}}) + with pytest.raises(RuntimeError, match="already open"): + await _start().execute(context) + + +def test_the_steps_round_trip(): + start = _start(frames=False, browser="chromium") + end = _end() + + assert StartBrowserRecordingTaskStep.from_dict(start.to_dict()).to_dict() == start.to_dict() + assert EndBrowserRecordingTaskStep.from_dict(end.to_dict()).to_dict() == end.to_dict() + assert start.to_dict()["frames"] is False and start.fail_task_on_error and end.fail_task_on_error + with pytest.raises(ValueError): + StartBrowserRecordingTaskStep(id="s", version=None, env_id="browser", name="")