From 808d8fc11a0ba3357e94201f849e8f099971bd10 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 11 Nov 2026 20:36:46 -0800 Subject: [PATCH 1/5] feat(browser): browser-recording/v1 with frames after each agent action, keyed to the action log (FD-4971) The first slice of browser-recording/v1 (M8.22), with M8.23's frames as its content. The browser takes a frame of the tab an agent last acted in at the start, after each burst of its acting entries settles and its page loads, and at the end, uploading each under the recording's grant and keying it to the log entries it follows; at the end it uploads the log's lines since the start and a manifest. start_browser_recording and end_browser_recording run it and register the frames, log and manifest as one universe; the end fails the run when the log or a frame is missing. Teardown ends what a run left open. The recording grant and namespace helpers are shared with the computer recording. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 2 +- packages/agentenv-protocol/README.md | 11 + .../website_browser/__init__.py | 7 +- .../website_browser/control.py | 37 ++- .../website_browser/environment.py | 24 ++ .../agentenv_protocol/website_browser/log.py | 14 +- .../website_browser/recording.py | 229 ++++++++++++++++ .../tests/test_browser_recording.py | 199 ++++++++++++++ src/agent_env/task/episode/rules.py | 3 +- src/agent_env/task/teardown.py | 8 +- src/agent_env/task_step/registry.py | 6 + .../task_step/task_steps/browser/recording.py | 246 ++++++++++++++++++ .../task_steps/computer_recording.py | 50 ++-- .../computer/browser_computer_test.py | 39 ++- tst/unit/task_step/browser_recording_test.py | 173 ++++++++++++ 15 files changed, 1018 insertions(+), 30 deletions(-) create mode 100644 packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py create mode 100644 packages/agentenv-protocol/tests/test_browser_recording.py create mode 100644 src/agent_env/task_step/task_steps/browser/recording.py create mode 100644 tst/unit/task_step/browser_recording_test.py 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..90c1f1aa4 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,10 @@ #: 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 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" @@ -100,8 +104,9 @@ def __init__(self, devtools: DevToolsPipe, dialogs: str, viewport: Optional[tupl 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 +618,32 @@ 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 PNG it + last took. 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 in self._open_dialogs: + if (shot := self._shots.get(target)) is None: + raise DeviceUnavailable("a dialog keeps the tab from painting and it had no capture before it") + return shot, "png", target, self.url_of(target) + try: + shot = await self.devtools.call("Page.captureScreenshot", { + "format": "jpeg", "quality": FRAME_QUALITY, "optimizeForSpeed": True}, session_id=session, + timeout_s=CAPTURE_TIMEOUT_S) + 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 + return shot["data"], "jpg", target, self.url_of(target) + 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 +758,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"))) @@ -808,12 +841,14 @@ def _on_Target_detachedFromTarget(self, params: dict, session: Optional[str]) -> if (target := self._sessions.pop(params["sessionId"], None)) is not None: self._shots.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..4c1b0034e --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py @@ -0,0 +1,229 @@ +"""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 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 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 + + +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.""" + + index: int + kind: Literal["first", "action", "last"] + at: str + 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.""" + + 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 + + @property + def missing_frames(self) -> int: + return sum(1 for frame in self.frames if not frame.uploaded) + + +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 = f"{FRAMES}/{index:04d}.{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}"[:300] + 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.recordings.log.read_async(manifest.log_after, limit=1 << 30) + text = "".join(json.dumps(line, default=str) + "\n" for line in lines["lines"] + if line.get("seq", 0) <= manifest.log_last) + 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}"[:300] + 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 + + 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..65b0b0261 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_browser_recording.py @@ -0,0 +1,199 @@ +"""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 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_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" 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..426cd4de9 --- /dev/null +++ b/src/agent_env/task_step/task_steps/browser/recording.py @@ -0,0 +1,246 @@ +"""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 + +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.computer_recording import recording_grant, recording_namespace, refuse_no_network + +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: every frame, the log and the manifest, each frame or the log at most this big. +MAX_OBJECT_BYTES = 256 << 20 +MAX_RECORDING_BYTES = 16 << 30 + + +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": f"{type(e).__name__}: {e}"[:500]} + 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 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("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.""" + 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_browser_recording(context, name, "teardown") + + +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.""" + 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("browser 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) + + +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"] = f"{type(e).__name__}: {e}"[:500] + 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 {type(listing).__name__}: {listing}"[:500] + 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) + 2, 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", f"{type(e).__name__}: {e}"[:500] + 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) + 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, with no log.""" + 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))) + paths = sorted(url[len(root):] for url in urls if url[len(root):].startswith("frames/")) + 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=f"{root}{LOG}" in urls, + log=LOG if f"{root}{LOG}" in urls else None, + frames=[RecordedFrame(index=i, kind="action", at="", path=path, uploaded=True) for i, path in enumerate(paths)]) + + +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 or "image/jpeg") + 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..59e8ff499 100644 --- a/src/agent_env/task_step/task_steps/computer_recording.py +++ b/src/agent_env/task_step/task_steps/computer_recording.py @@ -88,8 +88,8 @@ 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) 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, @@ -214,45 +214,53 @@ 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.""" +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's segments") + "recording") -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.""" +def recording_namespace(request_id: str, prefix: str = _KEY_PREFIX) -> str: + """Where a recording's files 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}") + return config.get_object_store().object_url(f"{config.get_artifact_key_prefix()}{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.""" - 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 {} +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") - 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) + 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=duration + GRANT_MARGIN_S) + 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 _grant(deployed, namespace: str, environment_name: Optional[str] = None, method: str = "start") -> dict: + """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) + return recording_grant(deployed, extension.get("params") or {}, namespace, MAX_SEGMENTS + 1, _MAX_SEGMENT_BYTES, + MAX_RECORDING_BYTES) + + def _from_namespace(entry: dict) -> RecordingManifest: """What the recording's namespace holds, for a recording its driver didn't end for the run: the manifest the driver uploaded beside the segments when it ended it, or else one of the segments there.""" diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index 7d5706ab5..8e64ad551 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -70,7 +70,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 +84,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 @@ -1074,6 +1077,40 @@ 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) + + 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..b2c721e98 --- /dev/null +++ b/tst/unit/task_step/browser_recording_test.py @@ -0,0 +1,173 @@ +"""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.website_browser import BROWSER, BROWSER_RECORDING + +from agent_env.artifact import Artifact +from agent_env.config import set_object_store +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) -> 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 = store, misses, forgets + 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": 1, "log_uploaded": True} + 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_recording_the_browser_forgot_is_registered_from_its_namespace_and_fails_for_its_missing_log(store): + context = await _start().execute(_context(_BrowserComputer(store, forgets=True))) + with pytest.raises(RuntimeError, match="its log didn't upload"): + await _end().execute(context) + entry = context.metadata[RECORDINGS_KEY]["run"] + + assert (entry["state"], entry["frames_taken"], entry["log_uploaded"]) == ("ended", 1, False) + assert "no recording rec-1" in entry["error"] + + +@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="") From e39b2453612fca4b12c1508406615ebc71fc6930 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 11 Nov 2026 20:49:15 -0800 Subject: [PATCH 2/5] fix(browser): a recording's log covers every entry, frames survive dialogs and closing tabs, resets end it - An end waits up to 10 s for the entries still in flight to be written, and lists any it still doesn't hold as log_missing, which fails the end step. - A frame taken while a dialog keeps the tab from painting is the last frame the recording took of it, in its own format. - A tab that closes while its page loads is framed as the active tab. - A computer's reset or close ends its browser apps' recordings (end_all), as it ends the driver's, so their log and manifest upload first. - Errors are cut to ERROR_CHARS. Co-Authored-By: Claude Opus 5.5 --- .../website_browser/control.py | 21 ++++++-- .../website_browser/recording.py | 33 +++++++++--- .../tests/test_browser_recording.py | 50 +++++++++++++++++++ .../env_providers/env_computer_provider.py | 18 +++++-- .../task_step/task_steps/browser/recording.py | 9 ++-- tst/unit/task_step/browser_recording_test.py | 37 ++++++++++++-- 6 files changed, 146 insertions(+), 22 deletions(-) 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 90c1f1aa4..76743ac23 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py @@ -101,6 +101,9 @@ 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() @@ -620,9 +623,10 @@ async def capture_page(self) -> dict: 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 PNG it - last took. The image base64, its format and the tab and its URL; DeviceUnavailable with no tab open, or for one - that paints nothing in time.""" + 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() @@ -630,10 +634,15 @@ async def frame(self, target: Optional[str], load_wait_s: float) -> tuple[str, s 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 (shot := self._shots.get(target)) is None: + 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 shot, "png", target, self.url_of(target) + return *earlier, target, self.url_of(target) try: shot = await self.devtools.call("Page.captureScreenshot", { "format": "jpeg", "quality": FRAME_QUALITY, "optimizeForSpeed": True}, session_id=session, @@ -642,6 +651,7 @@ async def frame(self, target: Optional[str], load_wait_s: float) -> tuple[str, s 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 _capture(self, session: str) -> Optional[str]: @@ -840,6 +850,7 @@ 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) diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py index 4c1b0034e..9ead480e7 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py @@ -16,7 +16,7 @@ from ..computer.errors import InvalidArgument from ..transfers import NamespaceUploader, WriteNamespaceGrant -from .log import BrowserLog +from .log import ERROR_CHARS, BrowserLog if TYPE_CHECKING: from .control import BrowserControl @@ -40,6 +40,10 @@ 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): @@ -60,7 +64,8 @@ class RecordedFrame(BaseModel): 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.""" + ``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 @@ -73,6 +78,7 @@ class BrowserRecordingManifest(BaseModel): log_after: int = 0 log_last: int = 0 log_uploaded: bool = False + log_missing: list[int] = [] @property def missing_frames(self) -> int: @@ -181,7 +187,7 @@ async def take(self, kind: str) -> None: 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}"[:300] + 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: @@ -197,15 +203,16 @@ async def end(self, grant: Optional[WriteNamespaceGrant], reason: str) -> dict: await self.take("last") manifest = self.manifest manifest.log_last = self.recordings.log.next_seq - 1 - lines = await self.recordings.log.read_async(manifest.log_after, limit=1 << 30) - text = "".join(json.dumps(line, default=str) + "\n" for line in lines["lines"] - if line.get("seq", 0) <= manifest.log_last) + 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}"[:300] + 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: @@ -214,6 +221,18 @@ async def end(self, grant: Optional[WriteNamespaceGrant], reason: str) -> dict: 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 + while True: + lines = [line for line in (await log.read_async(manifest.log_after, limit=1 << 30))["lines"] + if line.get("seq", 0) <= manifest.log_last] + if len(lines) >= manifest.log_last - manifest.log_after 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}" diff --git a/packages/agentenv-protocol/tests/test_browser_recording.py b/packages/agentenv-protocol/tests/test_browser_recording.py index 65b0b0261..87420dcd3 100644 --- a/packages/agentenv-protocol/tests/test_browser_recording.py +++ b/packages/agentenv-protocol/tests/test_browser_recording.py @@ -109,6 +109,22 @@ async def test_a_burst_still_settling_at_the_end_gets_its_frame_before_the_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 @@ -197,3 +213,37 @@ async def test_a_frame_waits_for_its_tabs_page_to_load_then_captures_it_as_a_jpe 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) + + +@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_step/task_steps/browser/recording.py b/src/agent_env/task_step/task_steps/browser/recording.py index 426cd4de9..b2261dee5 100644 --- a/src/agent_env/task_step/task_steps/browser/recording.py +++ b/src/agent_env/task_step/task_steps/browser/recording.py @@ -120,12 +120,14 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: def incomplete(entry: dict) -> Optional[str]: - """What's missing from the recording ``entry`` records: that it didn't end, its log, or its frames; None for a - complete one.""" + """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 @@ -199,7 +201,8 @@ def _registered(context: TaskStepContext, name: str, entry: dict, manifest: Brow 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) + 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, diff --git a/tst/unit/task_step/browser_recording_test.py b/tst/unit/task_step/browser_recording_test.py index b2c721e98..2ca1dbb44 100644 --- a/tst/unit/task_step/browser_recording_test.py +++ b/tst/unit/task_step/browser_recording_test.py @@ -8,10 +8,13 @@ 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 ( @@ -32,12 +35,13 @@ 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) -> None: + 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 = store, misses, forgets + 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, **_): @@ -59,7 +63,8 @@ async def invoke(self, uri, method, params=None, environment_name=None, **_): 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": 1, "log_uploaded": True} + "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 @@ -114,6 +119,32 @@ async def test_a_missing_frame_fails_the_end_as_an_infrastructure_failure_unless 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): context = await _start().execute(_context(_BrowserComputer(store, forgets=True))) From 1f09a5010ef98a174b4fa9ff3255183b7e0224ed Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 12 Nov 2026 08:34:53 -0800 Subject: [PATCH 3/5] fix(browser-recording): frame a tab behind the active one Chrome answers a capture of a tab behind the active one only once another capture of it arrives, and then answers both, so a CDP agent's own background tab lost every other frame to the capture timeout. A frame's capture still unanswered after 0.5 s is followed by a second, and the first answer is the frame. Co-Authored-By: Claude Opus 5.5 --- .../website_browser/control.py | 33 ++++++++++- .../tests/test_browser_recording.py | 37 +++++++++++- .../computer/browser_computer_test.py | 57 +++++++++++++++++++ 3 files changed, 123 insertions(+), 4 deletions(-) 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 76743ac23..f8371b997 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/control.py @@ -46,6 +46,9 @@ 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. @@ -644,9 +647,7 @@ async def frame(self, target: Optional[str], load_wait_s: float) -> tuple[str, s 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.devtools.call("Page.captureScreenshot", { - "format": "jpeg", "quality": FRAME_QUALITY, "optimizeForSpeed": True}, session_id=session, - timeout_s=CAPTURE_TIMEOUT_S) + 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: @@ -654,6 +655,32 @@ async def frame(self, target: Optional[str], load_wait_s: float) -> tuple[str, s 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] diff --git a/packages/agentenv-protocol/tests/test_browser_recording.py b/packages/agentenv-protocol/tests/test_browser_recording.py index 87420dcd3..4b35d1899 100644 --- a/packages/agentenv-protocol/tests/test_browser_recording.py +++ b/packages/agentenv-protocol/tests/test_browser_recording.py @@ -15,7 +15,7 @@ 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 BrowserControl +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 @@ -238,6 +238,41 @@ async def test_while_a_dialog_keeps_a_tab_from_painting_its_frame_is_the_last_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() diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index 8e64ad551..5c0f4c339 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -1111,6 +1111,63 @@ async def test_a_browser_recording_frames_each_burst_of_an_agents_actions_keyed_ 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 connect(ws_url, max_size=None, open_timeout=60) 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(), 60))).get("id") != message_id: + pass + return reply + + 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, _): From c55133330a5d036e3ee57de88ae7ebab44bdb5bf Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 12 Nov 2026 08:43:17 -0800 Subject: [PATCH 4/5] refactor(browser-recording): read only the log lines a recording covers Its end read the log with an unnamed ceiling for a limit; the count of entries between its start and its end is the limit it needs. Co-Authored-By: Claude Opus 5.5 --- .../src/agentenv_protocol/website_browser/recording.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py index 9ead480e7..7bcd58044 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py @@ -226,10 +226,11 @@ async def _log_lines(self) -> list[dict]: 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=1 << 30))["lines"] + 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) >= manifest.log_last - manifest.log_after or asyncio.get_running_loop().time() >= deadline: + if len(lines) >= wanted or asyncio.get_running_loop().time() >= deadline: return lines await asyncio.sleep(LOG_POLL_S) From fab1ca7edac90985ddac327c37b60b2e411186b4 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 12 Nov 2026 09:25:27 -0800 Subject: [PATCH 5/5] refactor(recordings): share what computer and browser recordings have in common Both recordings' steps now take from one module, task_steps/recordings.py: - where a recording uploads and the grant it uploads under - the error its entry keeps (why, ERROR_CHARS) - teardown's two passes over the recordings a run left open A manifest rebuilt from a browser recording's namespace now names each frame by the index and type its file is named by, with its kind and time unknown (None), rather than as action frames taken at "". The grant's room beside the frames is named, and the browser computer tests share one CDP client. Co-Authored-By: Claude Opus 5.5 --- .../website_browser/recording.py | 22 ++++- .../task_step/task_steps/browser/recording.py | 56 +++++------ .../task_steps/computer_recording.py | 74 ++++----------- .../task_step/task_steps/recordings.py | 93 +++++++++++++++++++ .../computer/browser_computer_test.py | 51 +++++----- tst/unit/task_step/browser_recording_test.py | 10 +- 6 files changed, 186 insertions(+), 120 deletions(-) create mode 100644 src/agent_env/task_step/task_steps/recordings.py diff --git a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py index 7bcd58044..1894927c3 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/website_browser/recording.py @@ -8,6 +8,7 @@ import base64 import json import logging +import re import uuid from datetime import datetime, timezone from typing import TYPE_CHECKING, Literal, Optional @@ -48,11 +49,12 @@ 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.""" + 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: Literal["first", "action", "last"] - at: str + kind: Optional[Literal["first", "action", "last"]] = None + at: Optional[str] = None after: list[int] = [] tab: Optional[str] = None url: Optional[str] = None @@ -85,6 +87,18 @@ 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.""" @@ -183,7 +197,7 @@ async def take(self, kind: str) -> None: 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 = f"{FRAMES}/{index:04d}.{extension}", CONTENT_TYPES[extension] + 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: diff --git a/src/agent_env/task_step/task_steps/browser/recording.py b/src/agent_env/task_step/task_steps/browser/recording.py index b2261dee5..c0eca067e 100644 --- a/src/agent_env/task_step/task_steps/browser/recording.py +++ b/src/agent_env/task_step/task_steps/browser/recording.py @@ -13,7 +13,7 @@ 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 +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 @@ -25,16 +25,26 @@ 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.computer_recording import recording_grant, recording_namespace, refuse_no_network +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: every frame, the log and the manifest, each frame or the log at most this big. +#: 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): @@ -76,7 +86,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: environment_name=browser) except Exception as e: recordings[self.name] = {"env_id": self.env_id, "browser": self.browser, "frames": self.frames, - "state": "failed", "error": f"{type(e).__name__}: {e}"[:500]} + "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 @@ -136,23 +146,13 @@ def incomplete(entry: dict) -> Optional[str]: 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.""" - 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_browser_recording(context, name, "teardown") + 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.""" - 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("browser 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, "browser recording") async def end_browser_recording(context: TaskStepContext, name: str, reason: str) -> None: @@ -169,12 +169,12 @@ async def end_browser_recording(context: TaskStepContext, name: str, reason: str "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"] = f"{type(e).__name__}: {e}"[:500] + 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 {type(listing).__name__}: {listing}"[:500] + entry["state"], entry["error"] = "failed", f"{entry['error']}; then {why(listing)}"[:ERROR_CHARS] return _registered(context, name, entry, manifest) @@ -187,8 +187,8 @@ def _grant(record: DeployedEnv, card: dict, namespace: str, method: str = "start 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) + 2, MAX_OBJECT_BYTES, - MAX_RECORDING_BYTES) + 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: @@ -197,7 +197,7 @@ def _registered(context: TaskStepContext, name: str, entry: dict, manifest: Brow 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", f"{type(e).__name__}: {e}"[:500] + 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), @@ -211,18 +211,20 @@ def _registered(context: TaskStepContext, name: str, entry: dict, manifest: Brow 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, with no log.""" + 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))) - paths = sorted(url[len(root):] for url in urls if url[len(root):].startswith("frames/")) + 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=f"{root}{LOG}" in urls, - log=LOG if f"{root}{LOG}" in urls else None, - frames=[RecordedFrame(index=i, kind="action", at="", path=path, uploaded=True) for i, path in enumerate(paths)]) + 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: @@ -240,7 +242,7 @@ def _register(context: TaskStepContext, name: str, namespace: str, manifest: Bro 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 or "image/jpeg") + 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}")) 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 59e8ff499..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 @@ -89,13 +92,13 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: try: deployed = _deployed(context, self.env_id) refuse_no_network(deployed) - namespace = recording_namespace(request_id) + 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,41 +207,6 @@ 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 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 = _KEY_PREFIX) -> str: - """Where a recording's files 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()}{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 _grant(deployed, namespace: str, environment_name: Optional[str] = None, method: str = "start") -> dict: """A grant for a computer recording's segments and manifest. EnvCapabilityUnsupported when the env's card, or its child's, declares no recording.""" 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 5c0f4c339..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 @@ -610,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) @@ -747,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 = {} @@ -1146,17 +1149,7 @@ async def test_a_browser_recording_frames_a_cdp_agents_own_tab_behind_the_active 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 connect(ws_url, max_size=None, open_timeout=60) 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(), 60))).get("id") != message_id: - pass - return reply - + 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) diff --git a/tst/unit/task_step/browser_recording_test.py b/tst/unit/task_step/browser_recording_test.py index 2ca1dbb44..229f1bf07 100644 --- a/tst/unit/task_step/browser_recording_test.py +++ b/tst/unit/task_step/browser_recording_test.py @@ -147,13 +147,19 @@ async def invoke(self, uri, method, params=None, environment_name=None, timeout= @pytest.mark.asyncio async def test_a_recording_the_browser_forgot_is_registered_from_its_namespace_and_fails_for_its_missing_log(store): - context = await _start().execute(_context(_BrowserComputer(store, forgets=True))) + 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", 1, False) + 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