From 2a449a1a4093190367637692a1f8c463ba15e5df Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Fri, 13 Nov 2026 11:42:33 -0800 Subject: [PATCH 1/2] feat(clock): core's clock writes wait for the computers on the clock and fail their step naming any behind (FD-5473) Every clock write core makes goes through DeployedEnv.write_clock: sync_env_clock's arm, the always-on step's freezes, thaws and advances, and a post-turn hook's. It passes wait_followers_s (CLOCK_WAIT_FOLLOWERS_S, 2 s), so the gateway answers once each computer following its clock shows the change, and a 504 raises ClockFollowersBehind naming the computers still behind; the write itself has applied, so it isn't retried, and the step fails. A clock no computer follows answers at once, as before. - The gateway's 504 carries the protocol's error shape (code clock_followers_behind, applied yes) beside its stale and followers, so the client reads its code and message instead of a body of its own. - The always-on step neither retries such a write nor swallows it where it tolerates other failures of an advance. - A hook that catches the error still fails its step: HookEnv keeps it, and the runner raises it after the hook. - An integration test pauses a multi env's computer: core's advance fails after its wait naming it, and once the computer is back the next is acknowledged within the wait, the session on the new time. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 2 +- src/agent_env/env/env.py | 31 ++++- src/agent_env/env/gateway/constants.py | 7 +- src/agent_env/env/gateway/gateway.py | 13 ++- .../task_step/task_steps/always_on.py | 14 ++- .../task_step/task_steps/post_turn_hooks.py | 20 +++- .../task_step/task_steps/sync_env_clock.py | 12 +- .../computer/computer_clock_test.py | 51 +++++++- tst/unit/env/gateway/clock_watch_test.py | 7 +- tst/unit/env/write_clock_test.py | 110 ++++++++++++++++++ tst/unit/task_step/always_on_harness.py | 9 +- tst/unit/task_step/sync_env_clock_test.py | 31 ++++- .../test_always_on_clock_followers.py | 50 ++++++++ tst/unit/task_step/test_always_on_no_clock.py | 3 + .../test_prompt_agent_post_turn_hooks.py | 43 ++++++- 15 files changed, 366 insertions(+), 37 deletions(-) create mode 100644 tst/unit/env/write_clock_test.py create mode 100644 tst/unit/task_step/test_always_on_clock_followers.py diff --git a/AGENTS.md b/AGENTS.md index 27a4d2fca..1653954cf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -123,7 +123,7 @@ and deserialized through a registry. | Package | Contents | |---|---| -| `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, but not for one that hasn't watched for longer than a hold and a reconnect (`ClockFollowers.QUIET_S`), which it names at once with when each last watched; 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`. | +| `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, but not for one that hasn't watched for longer than a hold and a reconnect (`ClockFollowers.QUIET_S`), which it names at once with when each last watched, the 504's error coded `clock_followers_behind`. Every clock write core makes (`sync_env_clock`'s arm, the always-on step's, a post-turn hook's) goes through `DeployedEnv.write_clock`, which waits `CLOCK_WAIT_FOLLOWERS_S` (2 s) and fails its step with `ClockFollowersBehind` naming the computers behind, not retried, since the write applied; a hook that catches it fails the step all the same. 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`, `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/`. | diff --git a/src/agent_env/env/env.py b/src/agent_env/env/env.py index 76a1676a2..06f1b6e55 100644 --- a/src/agent_env/env/env.py +++ b/src/agent_env/env/env.py @@ -18,13 +18,16 @@ from agent_env.utils.paths import validate_relative_filename from .gateway import GatewayMode -from .gateway.constants import WELL_KNOWN_PATH +from .gateway.constants import CLOCK_FOLLOWERS_BEHIND, CLOCK_WRITES, EXT_CLOCK_URI, WELL_KNOWN_PATH from .store import get_env_store logger = logging.getLogger(__name__) # Default timeout for one invoke() call; above the class because a default argument needs it. _EXTENSION_CALL_TIMEOUT_S = 30 +#: How long each clock write core makes waits for the computers following the clock to show it, so no step hands an +#: agent a computer still showing the old time. +CLOCK_WAIT_FOLLOWERS_S = 2.0 if TYPE_CHECKING: from agent_env.bundle.authoring import AuthoringContext @@ -46,6 +49,17 @@ def __init__(self, capability: str, method: str, env_id: str, instance_id: Optio super().__init__(f"env '{env_id}'{child} does not offer '{method}' on {capability}.") +class ClockFollowersBehind(RuntimeError): + """A clock write the env applied that some of its followers, the computers on its clock, didn't show in time: they + still show the old time. ``stale`` names them.""" + + def __init__(self, env_id: str, method: str, stale: list[str], detail: str) -> None: + self.env_id = env_id + self.method = method + self.stale = stale + super().__init__(f"env '{env_id}': clock {method}: {detail}") + + class EnvNeedsGateway(RuntimeError): """A caller reaches an env through its gateway, and the env was deployed without one.""" @@ -171,6 +185,21 @@ async def invoke(self, uri: str, method: str, params: Optional[dict] = None, *, return await protocol_v1.invoke_extension(self.environment_url, card, uri, params, timeout, method=method, headers=self.control_headers) + async def write_clock(self, method: str, params: Optional[dict] = None, *, + timeout: int = _EXTENSION_CALL_TIMEOUT_S) -> Any: + """A clock/v1 write (``CLOCK_WRITES``) that waits up to ``CLOCK_WAIT_FOLLOWERS_S`` for every computer following + the clock to show it, unless ``params`` sets its own ``wait_followers_s``. Raises ``ClockFollowersBehind`` + naming those that didn't; the write itself has applied. A clock no computer follows answers at once.""" + if method not in CLOCK_WRITES: + raise ValueError(f"{method!r} isn't a clock write; one of {list(CLOCK_WRITES)}") + params = {"wait_followers_s": CLOCK_WAIT_FOLLOWERS_S, **(params or {})} + try: + return await self.invoke(EXT_CLOCK_URI, method, params, timeout=timeout) + except protocol_v1.ExtensionCallError as e: + if e.code != CLOCK_FOLLOWERS_BEHIND: + raise + raise ClockFollowersBehind(self.env_id, method, e.response.json()["stale"], e.detail) from e + async def readiness(self, timeout: float = READY_TIMEOUT_S) -> WebsiteReadiness: """Whether the env is ready, as the protocol determines a website env's health (``env_readiness``): every site its card, the stored one or, for a record without it, the card it serves now, advertises ``website/v1``'s diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py index 06ecbcfd3..f258d9d14 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/src/agent_env/env/gateway/constants.py @@ -77,6 +77,11 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int: CLOCK_WATCH_HOLD_S = 25.0 # The longest a clock write's wait_followers_s may ask it to wait for the followers. CLOCK_MAX_WAIT_FOLLOWERS_S = 30.0 +# The clock/v1 methods that change the clock, each taking wait_followers_s. +CLOCK_WRITES = ("set_time", "set_rate", "advance", "clear") +# The error code a clock write answers 504 with when a declared follower didn't show it within its wait_followers_s: +# the write applied, and the followers named in its stale are behind. +CLOCK_FOLLOWERS_BEHIND = "clock_followers_behind" # Env/gateway-level extensions the gateway itself owns (tool gating + triggers). EXT_DISABLE_TOOL_URI = "urn:agentenv:disable-tool/v1" @@ -244,7 +249,7 @@ def random_mcp_server_name() -> str: }, { "uri": EXT_CLOCK_URI, - "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set_time/set_rate/advance/clear are the harness control path, each changing the clock's opaque version and, given wait_followers_s, answering 504 naming the followers that don't show it in time (set_rate keeps t0, as advance does, so time triggers keep their marks; rate 0 freezes the clock); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.", + "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set_time/set_rate/advance/clear are the harness control path, each changing the clock's opaque version and, given wait_followers_s, answering 504, error code clock_followers_behind, with stale naming the followers that don't show it in time (set_rate keeps t0, as advance does, so time triggers keep their marks; rate 0 freezes the clock); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.", "params": {"endpoint": "/clock/time", "methods": { "get_time": {"method": "GET", "endpoint": "/clock/time"}, "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}}, diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index 20ae28735..50df6cc50 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -48,6 +48,7 @@ CHANGES_PAGE_DEFAULT, CHANGES_PAGE_MAX, CHILD_DISCOVERY_TIMEOUT_S, + CLOCK_FOLLOWERS_BEHIND, CONTROL_ROUTES, DATA_PLANE_LOAD_TIMEOUT_MAX_S, DEFAULT_ROLE, @@ -692,7 +693,8 @@ async def _await_barriers(self, pending: list[asyncio.Task], timeout_s: float | async def _clock_write(self, request: Request, change, *, body_optional: bool = False) -> Response: """A clock write route: its JSON object body (none at all when ``body_optional``) applied by ``change`` through ``_change_clock``; 400 for a body or value the clock refuses, else 200, or 504 naming the followers behind, with - what each follower last showed and when it last watched.""" + what each follower last showed and when it last watched, its error in the protocol's shape so a client reads its + code (``CLOCK_FOLLOWERS_BEHIND``) and message.""" try: body = await request.json() if not body_optional or await request.body() else {} except Exception: @@ -705,10 +707,11 @@ async def _clock_write(self, request: Request, change, *, body_optional: bool = except ClockError as e: return self._json_response({"ok": False, "error": str(e)}, 400) if stale: - return self._json_response({**state, "ok": False, "stale": stale, "followers": self._clock_followers.state(), - "error": (f"the clock is at version {state['version']}, but {', '.join(stale)} " - f"didn't show it (waited up to {wait:g} s; one that has stopped " - "watching isn't waited for): their clocks are behind")}, 504) + message = (f"the clock is at version {state['version']}, but {', '.join(stale)} didn't show it (waited " + f"up to {wait:g} s; one that has stopped watching isn't waited for): their clocks are behind") + error = {"code": CLOCK_FOLLOWERS_BEHIND, "message": message, "retryable": False, "applied": "yes"} + return self._json_response({**state, "ok": False, "error": error, "stale": stale, + "followers": self._clock_followers.state()}, 504) return self._json_response(state, 200) async def _change_clock(self, change, wait_s: float = 0.0) -> tuple[dict, list[str]]: diff --git a/src/agent_env/task_step/task_steps/always_on.py b/src/agent_env/task_step/task_steps/always_on.py index bb60c912a..b85af40da 100644 --- a/src/agent_env/task_step/task_steps/always_on.py +++ b/src/agent_env/task_step/task_steps/always_on.py @@ -95,10 +95,11 @@ from agent_env.a2a_agent import A2AAgent, conversation_store, protocol from agent_env.a2a_agent.object_transfer import send_and_wait from agent_env.artifact import FileArtifact -from agent_env.env.env import EnvCapabilityUnsupported +from agent_env.env.env import ClockFollowersBehind, EnvCapabilityUnsupported from agent_env.env.gateway.notifications import DEFAULT_MAX_ACTIVE from agent_env.env.gateway.constants import ( CHANGES_PAGE_MAX, + CLOCK_WRITES, DEFAULT_ROLE, EXT_CHANGES_URI, EXT_CLOCK_URI, @@ -776,6 +777,8 @@ async def advance(self, target: datetime, reason: str) -> None: for env in armed: try: await _call(env, EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target)}, attempts=2) + except ClockFollowersBehind: + raise except Exception: state = await _call(env, EXT_CLOCK_URI, "state") if not (state.get("armed") and _utc(state["virtual_time"]) >= target): @@ -841,6 +844,8 @@ async def settle(self) -> list[dict]: elif due: # an event already due: an advance to now sets it going try: await _call(env, EXT_CLOCK_URI, "advance", {"virtual_time": _iso(now)}, attempts=1) + except ClockFollowersBehind: + raise except Exception: pass if waiting: @@ -1817,12 +1822,15 @@ async def _call(env, uri: str, method: str, params: Optional[dict] = None, *, at timeout: int = 60) -> Any: """A gateway control call, retried on a transient failure (a busy gateway can take seconds to answer). Every call the step retries is safe to repeat: reads, config, advance (checked against the clock after) and takes - with a take_id.""" + with a take_id. A clock write waits for the computers following the clock, and one they didn't show isn't retried: + it applied, and the step fails naming them.""" last: Optional[Exception] = None for attempt in range(attempts): try: + if uri == EXT_CLOCK_URI and method in CLOCK_WRITES: + return await env.write_clock(method, params, timeout=timeout) return await env.invoke(uri, method, params, timeout=timeout) - except EnvCapabilityUnsupported: + except (EnvCapabilityUnsupported, ClockFollowersBehind): raise except Exception as e: last = e diff --git a/src/agent_env/task_step/task_steps/post_turn_hooks.py b/src/agent_env/task_step/task_steps/post_turn_hooks.py index 0afb54b71..62850256a 100644 --- a/src/agent_env/task_step/task_steps/post_turn_hooks.py +++ b/src/agent_env/task_step/task_steps/post_turn_hooks.py @@ -20,9 +20,10 @@ from agentenv_protocol import client as protocol_v1 from agent_env.artifact import FileArtifact -from agent_env.env.env import DeployedEnv, EnvCapabilityUnsupported +from agent_env.env.env import ClockFollowersBehind, DeployedEnv, EnvCapabilityUnsupported from agent_env.env.gateway.constants import ( AGENT_ENV_ROLE_HEADER, + CLOCK_WRITES, EXT_CLOCK_URI, EXT_NOTIFICATIONS_URI, EXT_STEP_URI, @@ -120,12 +121,14 @@ def close(self) -> None: class HookEnv: """A deployed env as a hook reaches it: clock, triggers and notifications calls, and tool calls recorded as the - hook's.""" + hook's. A clock write waits for the computers following the clock (``DeployedEnv.write_clock``); one they didn't + show is kept in ``behind`` too, so the step fails on it even when the hook catches it.""" def __init__(self, deployed: DeployedEnv, hook_artifact_id: str) -> None: self.env_id = deployed.env_id self.env_step_id = (deployed.metadata or {}).get("deploy_step_id") self.instance_id = deployed.instance_id + self.behind: Optional[ClockFollowersBehind] = None self._deployed = deployed self._hook_artifact_id = hook_artifact_id @@ -135,6 +138,12 @@ def supports(self, extension: str, method: str) -> bool: async def invoke(self, extension: str, method: str, params: Optional[dict] = None) -> Any: if extension not in _INVOKABLE: raise ValueError(f"a post-turn hook invokes {list(_INVOKABLE)} only; call tools with call_tool") + if extension == EXT_CLOCK_URI and method in CLOCK_WRITES: + try: + return await self._deployed.write_clock(method, params) + except ClockFollowersBehind as e: + self.behind = self.behind or e + raise return await self._deployed.invoke(extension, method, params) async def call_tool(self, tool: str, args: Optional[dict], role: str) -> dict: @@ -265,12 +274,17 @@ async def run_post_turn_hooks(hooks: PostTurnHooks, *, context: TaskStepContext, agent_reply: str, agent_parts: list[dict], ask_user: Callable[[], Awaitable[Answer]], user_triggers: Callable[[], Awaitable[Optional[Answer]]]) -> Optional[Answer]: - """Run the hooks in order; the first to return a message or an end decides, and one that asked returns that.""" + """Run the hooks in order; the first to return a message or an end decides, and one that asked returns that. A + clock write a hook made that the computers following the clock didn't show fails the step, caught by the hook or + not, so the agent never acts on a computer still showing the old time.""" answer = _Answer(ask_user, user_triggers) for hook in hooks.hooks: turn = Turn(context=context, step_id=step_id, hook=hook, answer=answer, number=number, last=last, agent_reply=agent_reply, agent_parts=agent_parts) returned = await hook.post_turn(turn, copy.deepcopy(hook.params)) + behind = next((env.behind for env in turn.envs if env.behind), None) + if behind is not None: + raise behind if answer.value is not None: if returned != answer.value: raise RuntimeError(f"post-turn hook '{hook.artifact_id}' asked the user side at turn {number} and " diff --git a/src/agent_env/task_step/task_steps/sync_env_clock.py b/src/agent_env/task_step/task_steps/sync_env_clock.py index b4c92793e..455c08be7 100644 --- a/src/agent_env/task_step/task_steps/sync_env_clock.py +++ b/src/agent_env/task_step/task_steps/sync_env_clock.py @@ -1,8 +1,10 @@ """Arm a deployed env gateway's virtual clock, then sync its backing servers to it. Arms the clock (set_time) and reads the gateway's server-reachable env_get_time_url (state), both as the -stored env card declares clock/v1, then invokes sync_time on each backing server, MCP server or website, the -env card lists (``legacy_protocol.backing_servers``) whose own card advertises it, at the route that card gives. +stored env card declares clock/v1. The arm waits for the computers that follow the clock to show it, and fails the +step naming any that don't (``DeployedEnv.write_clock``). It then invokes sync_time on each backing server, MCP server +or website, the env card lists (``legacy_protocol.backing_servers``) whose own card advertises it, at the route that +card gives. Non-advertisers are skipped when tolerate_missing_sync_time (default True), and websites always. Place right before prompt_agent so the virtual timeline starts at agent-start. """ @@ -126,9 +128,9 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: if deployed is None: raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs") try: - await deployed.invoke(EXT_CLOCK_URI, "set_time", - {"virtual_time": self.armed_virtual_time, "virtual_seconds_per_real_second": self.virtual_seconds_per_real_second}, - timeout=self.timeout_seconds) + arm = {"virtual_time": self.armed_virtual_time, + "virtual_seconds_per_real_second": self.virtual_seconds_per_real_second} + await deployed.write_clock("set_time", arm, timeout=self.timeout_seconds) except httpx.HTTPStatusError as e: raise RuntimeError(f"clock arm failed (HTTP {e.response.status_code}): {e.response.text}") from e state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=self.timeout_seconds) diff --git a/tst/integration/computer/computer_clock_test.py b/tst/integration/computer/computer_clock_test.py index e6fe5c1ee..1f3701535 100644 --- a/tst/integration/computer/computer_clock_test.py +++ b/tst/integration/computer/computer_clock_test.py @@ -2,8 +2,10 @@ the gateway, as the computer's sandbox reaches it, under the computer's name and control token, which the record keeps but the token; sync_env_clock arms the clock as a task does, calling the computer's sync_time; the session reads universe time, and each change is acknowledged in time for a write that waits for it; a follower killed comes -back within 3 s and acknowledges the next advance; and a reset brings the page and the follower back on the clock's -current version. Needs Docker and the local registry on localhost:5000.""" +back within 3 s and acknowledges the next advance; a reset brings the page and the follower back on the clock's +current version; and with the computer paused, a clock write core makes fails after its wait, naming the computer, +while once it's back the next is acknowledged within that wait. Needs Docker and the local registry on +localhost:5000.""" import asyncio import time @@ -13,7 +15,7 @@ import pytest from agent_env.env import Env -from agent_env.env.env import DeployedEnv +from agent_env.env.env import CLOCK_WAIT_FOLLOWERS_S, ClockFollowersBehind, DeployedEnv from agent_env.env.envs.computer.utils import inside from agent_env.env.gateway.constants import EXT_CLOCK_URI from agent_env.store.routing import namespace_routing @@ -22,7 +24,8 @@ from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep from agent_env.task_step.task_steps.reset_env import ResetEnvTaskStep from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep -from tst.integration.computer.conftest import CONTAINER, FOLLOWER_BACK_S, computer_container, kill_follower, multi_env +from tst.integration.computer.conftest import (CONTAINER, FOLLOWER_BACK_S, computer_container, docker, kill_follower, + multi_env) pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] @@ -79,6 +82,46 @@ async def test_a_multi_envs_computer_follows_its_gateways_clock_through_arming_a assert not report.failed +@pytest.mark.asyncio +async def test_a_paused_computer_fails_cores_clock_write_naming_it_and_once_back_follows_the_next(desktop, items): + with namespace_routing(): + office = multi_env(items, Env.get(desktop.id)) + step = DeployEnvTaskStep(id="office", version=None, env_id=office.id, sandbox_type="local", ttl_seconds=900, + child_envs={"desktop": CONTAINER}) + context = TaskStepContext() + try: + [deployed] = (await step.execute(context)).deployed_envs + own = DeployedEnv.from_dict(deployed.child_envs["desktop"]) + container = computer_container(own.sandbox_id) + universe = datetime.now(timezone.utc) - timedelta(days=182) + await SyncEnvClockTaskStep(id="clock", version=None, env_id=office.id, + virtual_time=_iso(universe)).execute(context) + + docker("pause", container) + failed, started = None, time.monotonic() + try: + await deployed.write_clock("advance", {"virtual_time": _iso(universe + timedelta(hours=8))}) + except ClockFollowersBehind as e: + failed = e + finally: + took = time.monotonic() - started + docker("unpause", container) + + day_on = universe + timedelta(days=1) + started = time.monotonic() + await deployed.write_clock("advance", {"virtual_time": _iso(day_on)}) + acknowledged = time.monotonic() - started + offset = await _session_offset(own) - (day_on.timestamp() - time.time()) + finally: + report = await teardown_run(context) + + assert failed is not None and failed.stale == ["desktop"], failed + assert CLOCK_WAIT_FOLLOWERS_S <= took < CLOCK_WAIT_FOLLOWERS_S + 2, took + assert acknowledged < CLOCK_WAIT_FOLLOWERS_S and abs(offset) < 5, (acknowledged, offset) + print(f"paused: failed after {took:.2f} s; back: acknowledged in {acknowledged * 1000:.0f} ms") + assert not report.failed + + async def _advance(deployed, to: datetime) -> None: """The clock moved to ``to``, waiting up to 15 s for every computer to show it, as a 504 naming any that doesn't.""" await deployed.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(to), "wait_followers_s": 15}, timeout=30) diff --git a/tst/unit/env/gateway/clock_watch_test.py b/tst/unit/env/gateway/clock_watch_test.py index 018e5bee3..fd1f2f541 100644 --- a/tst/unit/env/gateway/clock_watch_test.py +++ b/tst/unit/env/gateway/clock_watch_test.py @@ -12,8 +12,8 @@ from agent_env.env.gateway import gateway as gw_mod from agent_env.env.gateway.clock import Clock from agent_env.env.gateway.clock_followers import ClockFollowers -from agent_env.env.gateway.constants import (CLOCK_MAX_WAIT_FOLLOWERS_S, CLOCK_WATCH_HOLD_S, EXT_CLOCK_URI, - GATEWAY_EXTENSIONS) +from agent_env.env.gateway.constants import (CLOCK_FOLLOWERS_BEHIND, CLOCK_MAX_WAIT_FOLLOWERS_S, CLOCK_WATCH_HOLD_S, + EXT_CLOCK_URI, GATEWAY_EXTENSIONS) from agent_env.env.gateway.entrypoint import parse_clock_followers from agent_env.env.gateway.gateway import Gateway @@ -166,7 +166,8 @@ async def test_a_write_answers_504_naming_the_computers_still_behind_and_the_cha body = answer.json() assert answer.status_code == 504 and waited >= 0.4 assert (body["ok"], body["stale"], body["version"]) == (False, ["laptop"], gw._clock.version) - assert "laptop didn't show it (waited up to 0.4 s;" in body["error"] + assert (body["error"]["code"], body["error"]["applied"]) == (CLOCK_FOLLOWERS_BEHIND, "yes") + assert "laptop didn't show it (waited up to 0.4 s;" in body["error"]["message"] assert body["followers"]["laptop"] == {"version": None, "seen_s_ago": None} assert read == {"virtual_time": T0} diff --git a/tst/unit/env/write_clock_test.py b/tst/unit/env/write_clock_test.py new file mode 100644 index 000000000..91eaee3c2 --- /dev/null +++ b/tst/unit/env/write_clock_test.py @@ -0,0 +1,110 @@ +"""``DeployedEnv.write_clock``, against a gateway in this process: each write waits for the computers that follow the +clock, one they didn't show raises ``ClockFollowersBehind`` naming them with the write applied, a clock no computer +follows answers at once, and a caller's own wait stands.""" + +from __future__ import annotations + +import json +import time + +import httpx +import pytest +from agentenv_protocol.client import ExtensionCallError + +from agent_env.env import env as env_mod +from agent_env.env.env import CLOCK_WAIT_FOLLOWERS_S, ClockFollowersBehind, DeployedGatewayEnv +from agent_env.env.gateway import gateway as gw_mod +from agent_env.env.gateway.constants import EXT_CLOCK_URI, GATEWAY_EXTENSIONS, WELL_KNOWN_PATH +from agent_env.env.gateway.gateway import Gateway + +GATEWAY = "http://gateway" +T0 = "2026-01-01T00:00:00Z" +LATER = "2026-01-02T00:00:00Z" +#: Short enough to keep the tests quick, long enough that answering "at once" can't be a wait running out. +WAIT_S = 0.5 + + +@pytest.fixture +def world(tmp_path, monkeypatch): + """A gateway whose clock ``laptop`` follows (``followers``), reached through a deployed env record; every request + body the record sends is kept in ``sent``.""" + monkeypatch.setattr(gw_mod, "GATEWAY_TRAJECTORY_FILE", str(tmp_path / "trajectory.jsonl")) + monkeypatch.setattr(env_mod, "CLOCK_WAIT_FOLLOWERS_S", WAIT_S) + made, sent = [], [] + real = httpx.AsyncClient + + def make(followers: dict | None = None) -> DeployedGatewayEnv: + gw = Gateway(host="127.0.0.1", port=0, server_name="unit", internal_mcp_servers=[], clock_followers=followers) + made.append(gw) + app = httpx.ASGITransport(app=gw._asgi_app()) + + async def handle(request: httpx.Request) -> httpx.Response: + body = await request.aread() + sent.append((request.url.path, json.loads(body) if body else None)) + return await app.handle_async_request(request) + + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(*a, **{**k, "transport": httpx.MockTransport(handle)})) + card = {"capabilities": {"extensions": list(GATEWAY_EXTENSIONS)}} + return DeployedGatewayEnv(env_id="office", env_version=1, gateway_url=GATEWAY, mcp_url="", sandbox_id="sb", + environment_card_url=f"{GATEWAY}{WELL_KNOWN_PATH}", environment_card=card) + + yield make, sent + for gw in made: + gw._close_trajectory_file() + + +@pytest.mark.asyncio +async def test_a_clock_no_computer_follows_answers_at_once_and_each_write_asked_to_wait(world): + make, sent = world + office = make() + + started = time.monotonic() + await office.write_clock("set_time", {"virtual_time": T0, "virtual_seconds_per_real_second": 0}) + await office.write_clock("advance", {"virtual_time": LATER}) + await office.write_clock("set_rate", {"virtual_seconds_per_real_second": 1}) + took = time.monotonic() - started + + assert took < WAIT_S, took + assert [(path, body["wait_followers_s"]) for path, body in sent] == [ + ("/clock/set-time", WAIT_S), ("/clock/advance", WAIT_S), ("/clock/set-rate", WAIT_S)] + assert CLOCK_WAIT_FOLLOWERS_S == 2.0 + + +@pytest.mark.asyncio +async def test_a_write_a_computer_doesnt_show_raises_naming_it_and_the_write_stands(world): + make, _ = world + office = make({"laptop": "token-laptop"}) + + started = time.monotonic() + with pytest.raises(ClockFollowersBehind) as behind: + await office.write_clock("set_time", {"virtual_time": T0, "virtual_seconds_per_real_second": 0}) + waited = time.monotonic() - started + state = await office.invoke(EXT_CLOCK_URI, "state") + + assert (behind.value.env_id, behind.value.method, behind.value.stale) == ("office", "set_time", ["laptop"]) + assert "env 'office': clock set_time: " in str(behind.value) and "laptop didn't show it" in str(behind.value) + assert waited >= WAIT_S + assert state["armed"] and state["virtual_time"] == T0 + + +@pytest.mark.asyncio +async def test_a_callers_own_wait_stands(world): + make, sent = world + office = make({"laptop": "token-laptop"}) + + await office.write_clock("set_time", {"virtual_time": T0, "virtual_seconds_per_real_second": 0, "wait_followers_s": 0}) + + assert sent[0][1]["wait_followers_s"] == 0 + + +@pytest.mark.asyncio +async def test_a_write_the_clock_refuses_is_the_gateways_error_and_a_read_isnt_a_write(world): + make, _ = world + office = make({"laptop": "token-laptop"}) + + with pytest.raises(ExtensionCallError) as refused: + await office.write_clock("advance", {"virtual_time": LATER}) + with pytest.raises(ValueError, match="'state' isn't a clock write"): + await office.write_clock("state") + + assert refused.value.response.status_code == 400 and "not armed" in refused.value.response.text diff --git a/tst/unit/task_step/always_on_harness.py b/tst/unit/task_step/always_on_harness.py index 3da8eab02..ea08e63cf 100644 --- a/tst/unit/task_step/always_on_harness.py +++ b/tst/unit/task_step/always_on_harness.py @@ -54,12 +54,15 @@ class Rig: user sim got in ``reads``. A ``plugin`` replaces the scripted one; what it sends the user sim agent with ``ask`` is in ``user_calls``, answered in turn by ``user_texts`` (then no_reply). ``during`` has, for each of the agent's turns, the gateway's settings and the agent's tools as the agent saw them. With ``takes_config`` the agent's card - takes its config (``CONFIG_CARD``), whose posts land in ``configs``.""" + takes its config (``CONFIG_CARD``), whose posts land in ``configs``. ``followers`` declares computers that follow + the gateway's clock, by name and token, as a multi env's do.""" def __init__(self, monkeypatch, tmp_path, *, agent: Optional[Turn] = None, answers: tuple = (), rate: float = 0, - t0: str = T0, plugin: Any = None, user_texts: tuple = (), takes_config: bool = False): + t0: str = T0, plugin: Any = None, user_texts: tuple = (), takes_config: bool = False, + followers: Optional[dict] = None): monkeypatch.setattr(gw_mod, "GATEWAY_TRAJECTORY_FILE", str(tmp_path / "trajectory.jsonl")) - self.gw = Gateway(host="127.0.0.1", port=0, server_name="AgentEnvGateway", internal_mcp_servers=[]) + self.gw = Gateway(host="127.0.0.1", port=0, server_name="AgentEnvGateway", internal_mcp_servers=[], + clock_followers=followers) self.gw._tools_discovered = True real_client = httpx.AsyncClient transport = httpx.ASGITransport(app=self.gw._mcp.streamable_http_app()) diff --git a/tst/unit/task_step/sync_env_clock_test.py b/tst/unit/task_step/sync_env_clock_test.py index e95acf2b3..c40e70b1d 100644 --- a/tst/unit/task_step/sync_env_clock_test.py +++ b/tst/unit/task_step/sync_env_clock_test.py @@ -4,8 +4,9 @@ import pytest -from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, EnvCapabilityUnsupported -from agent_env.env.gateway.constants import EXT_CLOCK_URI +from agent_env.env.env import (CLOCK_WAIT_FOLLOWERS_S, ClockFollowersBehind, DeployedEnv, DeployedGatewayEnv, + EnvCapabilityUnsupported) +from agent_env.env.gateway.constants import CLOCK_FOLLOWERS_BEHIND, EXT_CLOCK_URI from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps import sync_env_clock as mod from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep, resolve_virtual_time @@ -104,7 +105,8 @@ async def test_execute_syncs_each_child_env_through_the_stored_card(monkeypatch) ("PUT", f"{_CARD}/clock/set-time"), ("GET", f"{_CARD}/clock/state"), ("POST", f"{_CARD}/svc/mcp-email/agentenv/ext/sync_time"), ] - assert json.loads(sent[0].content) == {"virtual_time": T0, "virtual_seconds_per_real_second": 1.0} + assert json.loads(sent[0].content) == {"virtual_time": T0, "virtual_seconds_per_real_second": 1.0, + "wait_followers_s": CLOCK_WAIT_FOLLOWERS_S} assert [s["service"] for s in context.metadata["clock_configurations"][0]["synced"]] == ["email"] @@ -117,6 +119,22 @@ async def test_execute_raises_before_any_request_when_the_card_lacks_set_time(mo assert sent == [] +@pytest.mark.asyncio +async def test_an_arm_a_computer_didnt_show_fails_the_step_naming_it_before_any_server_is_synced(monkeypatch): + message = "the clock is at version b.2, but laptop didn't show it (waited up to 2 s; ...): their clocks are behind" + behind = {"ok": False, "stale": ["laptop"], "error": {"code": CLOCK_FOLLOWERS_BEHIND, "message": message, + "retryable": False, "applied": "yes"}} + sent = _mock_http(monkeypatch, set_time_status=504, set_time_body=behind) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]}))]) + + with pytest.raises(ClockFollowersBehind) as failed: + await _step().execute(context) + + assert failed.value.stale == ["laptop"] and str(failed.value) == f"env 'e1': clock set_time: {message}" + assert [(r.method, str(r.url)) for r in sent] == [("PUT", f"{_CARD}/clock/set-time")] + assert "clock_configurations" not in context.metadata + + @pytest.mark.asyncio async def test_execute_keeps_the_arm_failure_message(monkeypatch): _mock_http(monkeypatch, set_time_status=409) @@ -295,8 +313,9 @@ def _carded(card: dict) -> DeployedEnv: sandbox_id="sb-1", environment_card_url=f"{_CARD}{WELL_KNOWN_PATH}", environment_card=card) -def _mock_http(monkeypatch, state: dict | None = None, set_time_status: int = 200): - """Record every request: `/clock/state` answers `state`, `/clock/set-time` `set_time_status`, anything else an empty object.""" +def _mock_http(monkeypatch, state: dict | None = None, set_time_status: int = 200, set_time_body: dict | None = None): + """Record every request: `/clock/state` answers `state`, `/clock/set-time` `set_time_status` with `set_time_body`, + anything else an empty object.""" import httpx sent, real = [], httpx.AsyncClient @@ -304,7 +323,7 @@ def _mock_http(monkeypatch, state: dict | None = None, set_time_status: int = 20 def handle(request: httpx.Request) -> httpx.Response: sent.append(request) if request.url.path.endswith("/clock/set-time"): - return httpx.Response(set_time_status, json={}) + return httpx.Response(set_time_status, json=set_time_body or {}) return httpx.Response(200, json=state if request.url.path.endswith("/clock/state") else {}) monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle))) diff --git a/tst/unit/task_step/test_always_on_clock_followers.py b/tst/unit/task_step/test_always_on_clock_followers.py new file mode 100644 index 000000000..b022524db --- /dev/null +++ b/tst/unit/task_step/test_always_on_clock_followers.py @@ -0,0 +1,50 @@ +"""The always-on step's clock writes wait for the computers that follow the gateway's clock: one a computer doesn't +show fails the step naming it, without the write being retried, so the agent is never woken on a computer still +showing the old time.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from agent_env.env import env as env_mod +from agent_env.env.env import ClockFollowersBehind +from tst.unit.task_step.always_on_harness import Rig + +pytestmark = pytest.mark.usefixtures("local_stores") + +DAY = {"schedule": {"wake": "07:30", "sleep": "23:30", "work": ["09:00", "18:00"]}, + "response_times": {"work": 150, "free": 30, "morning": 15}} +WAIT_S = 0.3 + + +def _asks_once(n, texts, rig): + if n == 1: + rig.message("blue house or green?") + else: + rig.quiet() + return "" + + +def _changes(rig: Rig) -> int: + """How many times the gateway's clock has changed: the count in its opaque version.""" + return int(rig.gw._clock.version.rsplit(".", 1)[1]) + + +@pytest.mark.asyncio +async def test_a_clock_write_a_computer_doesnt_show_fails_the_step_naming_it_and_isnt_retried(monkeypatch, tmp_path): + monkeypatch.setattr(env_mod, "CLOCK_WAIT_FOLLOWERS_S", WAIT_S) + rig = Rig(monkeypatch, tmp_path, agent=_asks_once, answers=[{"outcome": "reply", "message": "blue"}], + followers={"laptop": "token-laptop"}) + await rig.arm() + armed = _changes(rig) + ctx = rig.context(user=DAY) + + with pytest.raises(ClockFollowersBehind) as behind: + await asyncio.wait_for(rig.run(rig.step(), ctx), 30) + + assert behind.value.stale == ["laptop"] + [error] = rig.events(ctx, "error") + assert "ClockFollowersBehind" in error["error"] and "laptop didn't show it" in error["error"] + assert _changes(rig) - armed == 1, rig.timeline(ctx) diff --git a/tst/unit/task_step/test_always_on_no_clock.py b/tst/unit/task_step/test_always_on_no_clock.py index 757fb7c69..532c0ce71 100644 --- a/tst/unit/task_step/test_always_on_no_clock.py +++ b/tst/unit/task_step/test_always_on_no_clock.py @@ -64,6 +64,9 @@ async def invoke(self, uri, method, params=None, **kw): return {"armed": True, "t0": "2027-03-01T15:00:00Z", "virtual_time": ao._iso(self.at), "virtual_seconds_per_real_second": 0} + async def write_clock(self, method, params=None, **kw): + return await self.invoke(EXT_CLOCK_URI, method, params) + @pytest.mark.asyncio async def test_a_message_the_chat_env_couldn_t_time_is_read_on_the_task_s_clock(monkeypatch, tmp_path): diff --git a/tst/unit/task_step/test_prompt_agent_post_turn_hooks.py b/tst/unit/task_step/test_prompt_agent_post_turn_hooks.py index f54d8a200..0248d2889 100644 --- a/tst/unit/task_step/test_prompt_agent_post_turn_hooks.py +++ b/tst/unit/task_step/test_prompt_agent_post_turn_hooks.py @@ -17,9 +17,9 @@ from agent_env.bundle.plan import plan_bundle from agent_env.bundle.resolve import resolve_bundle from agent_env.entity.refs import ref_sites -from agent_env.env.env import DeployedGatewayEnv +from agent_env.env.env import CLOCK_WAIT_FOLLOWERS_S, ClockFollowersBehind, DeployedGatewayEnv from agent_env.env.gateway.constants import ( - EXT_CLOCK_URI, EXT_STEP_URI, EXT_TRIGGERS_URI, GATEWAY_EXTENSIONS, WELL_KNOWN_PATH, + CLOCK_FOLLOWERS_BEHIND, EXT_CLOCK_URI, EXT_STEP_URI, EXT_TRIGGERS_URI, GATEWAY_EXTENSIONS, WELL_KNOWN_PATH, ) from agent_env.task_step.context import HEARTBEAT_KEY from agent_env.task_step.task_steps.post_turn_hooks import PostTurnHooks @@ -504,6 +504,45 @@ async def test_a_tool_call_on_a_gateway_that_cannot_tag_it_as_the_hooks_is_refus assert peers.steps == [] +class BehindPeers(EnvPeers): + """``EnvPeers`` whose gateway applies each clock advance that a computer on its clock, ``laptop``, doesn't show.""" + + def __init__(self, monkeypatch, replies, **kw): + super().__init__(monkeypatch, replies, **kw) + self.advances: list[dict] = [] + + def _answer(self, request: httpx.Request) -> httpx.Response: + if request.url.path == "/clock/advance": + self.advances.append(json.loads(request.content)) + message = "the clock is at version b.3, but laptop didn't show it (waited up to 2 s): it's behind" + return httpx.Response(504, json={"ok": False, "stale": ["laptop"], "error": { + "code": CLOCK_FOLLOWERS_BEHIND, "message": message, "retryable": False, "applied": "yes"}}) + return super()._answer(request) + + +@pytest.mark.asyncio +async def test_a_clock_jump_a_computer_didnt_show_fails_the_step_though_the_hook_caught_it(monkeypatch): + peers = BehindPeers(monkeypatch, {AGENT: [text("one"), text("two")]}) + ctx = _context() + ctx.deployed_envs.append(_env(EXT_CLOCK_URI)) + hook = _hook("jumper", """ + async def post_turn(turn, params): + [env] = turn.envs + try: + await env.invoke("urn:agentenv:clock/v1", "advance", {"virtual_time": "2026-03-03T10:00:00Z"}) + except Exception: + pass + return {"message": "an hour later"} + """) + + with pytest.raises(ClockFollowersBehind, match="laptop didn't show it") as behind: + await _step(hook, turns=2).execute(ctx) + + assert behind.value.stale == ["laptop"] + assert peers.advances == [{"virtual_time": "2026-03-03T10:00:00Z", "wait_followers_s": CLOCK_WAIT_FOLLOWERS_S}] + assert len(peers.tasks) == 1 # the agent wasn't sent the hook's message + + # ---- the env clock on stored messages ------------------------------------------------------------------------------- @pytest.mark.asyncio From 3f92b0f3d3406cbca41e60f426f9a1b22076b62c Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Fri, 13 Nov 2026 11:54:20 -0800 Subject: [PATCH 2/2] fix(clock): a gateway from before the error's code still has its followers' 504 read as computers behind Such a gateway answers a write its computers didn't show with a 504 whose error is its message, so the client saw no code and write_clock raised the plain error: the always-on step would retry or swallow an applied write, and a hook's catch would hide it (Greptile). Any 504 whose JSON names its stale computers now counts, whatever its error; a proxy's own 504 page stays the error it is. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/env/env.py | 22 +++++++++++++++++++--- tst/unit/env/write_clock_test.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/src/agent_env/env/env.py b/src/agent_env/env/env.py index 06f1b6e55..5998d0304 100644 --- a/src/agent_env/env/env.py +++ b/src/agent_env/env/env.py @@ -8,6 +8,7 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, ClassVar, Optional, Self +import httpx from agentenv_protocol import client as protocol_v1 from agentenv_protocol.website import READY_TIMEOUT_S, WEBSITE_EXTENSION_URI, WEBSITE_READY_OP, WebsiteReadiness, env_readiness @@ -60,6 +61,19 @@ def __init__(self, env_id: str, method: str, stale: list[str], detail: str) -> N super().__init__(f"env '{env_id}': clock {method}: {detail}") +def _followers_behind(response: httpx.Response) -> list[str] | None: + """The computers a clock write's 504 names as behind (``stale``), or None for any other answer, a proxy's own 504 + page among them.""" + if response.status_code != 504: + return None + try: + body = response.json() + except ValueError: + return None + stale = body.get("stale") if isinstance(body, dict) else None + return stale if isinstance(stale, list) else None + + class EnvNeedsGateway(RuntimeError): """A caller reaches an env through its gateway, and the env was deployed without one.""" @@ -189,16 +203,18 @@ async def write_clock(self, method: str, params: Optional[dict] = None, *, timeout: int = _EXTENSION_CALL_TIMEOUT_S) -> Any: """A clock/v1 write (``CLOCK_WRITES``) that waits up to ``CLOCK_WAIT_FOLLOWERS_S`` for every computer following the clock to show it, unless ``params`` sets its own ``wait_followers_s``. Raises ``ClockFollowersBehind`` - naming those that didn't; the write itself has applied. A clock no computer follows answers at once.""" + naming those that didn't; the write itself has applied. A clock no computer follows answers at once. A gateway + from before the error's code answers the same 504 with its message as ``error``, which counts the same.""" if method not in CLOCK_WRITES: raise ValueError(f"{method!r} isn't a clock write; one of {list(CLOCK_WRITES)}") params = {"wait_followers_s": CLOCK_WAIT_FOLLOWERS_S, **(params or {})} try: return await self.invoke(EXT_CLOCK_URI, method, params, timeout=timeout) except protocol_v1.ExtensionCallError as e: - if e.code != CLOCK_FOLLOWERS_BEHIND: + if (stale := _followers_behind(e.response)) is None: raise - raise ClockFollowersBehind(self.env_id, method, e.response.json()["stale"], e.detail) from e + detail = e.detail if e.code == CLOCK_FOLLOWERS_BEHIND else e.response.json()["error"] + raise ClockFollowersBehind(self.env_id, method, stale, detail) from e async def readiness(self, timeout: float = READY_TIMEOUT_S) -> WebsiteReadiness: """Whether the env is ready, as the protocol determines a website env's health (``env_readiness``): every site diff --git a/tst/unit/env/write_clock_test.py b/tst/unit/env/write_clock_test.py index 91eaee3c2..9eb038134 100644 --- a/tst/unit/env/write_clock_test.py +++ b/tst/unit/env/write_clock_test.py @@ -108,3 +108,34 @@ async def test_a_write_the_clock_refuses_is_the_gateways_error_and_a_read_isnt_a await office.write_clock("state") assert refused.value.response.status_code == 400 and "not armed" in refused.value.response.text + + +def _answering(monkeypatch, response: httpx.Response) -> DeployedGatewayEnv: + """A deployed env whose gateway answers every clock write with ``response``.""" + real = httpx.AsyncClient + transport = httpx.MockTransport(lambda request: response) + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(*a, **{**k, "transport": transport})) + card = {"capabilities": {"extensions": list(GATEWAY_EXTENSIONS)}} + return DeployedGatewayEnv(env_id="office", env_version=1, gateway_url=GATEWAY, mcp_url="", sandbox_id="sb", + environment_card_url=f"{GATEWAY}{WELL_KNOWN_PATH}", environment_card=card) + + +@pytest.mark.asyncio +async def test_an_older_gateways_504_naming_computers_behind_counts_the_same(monkeypatch): + message = "the clock is at version b.2, but laptop didn't show it (waited up to 2 s): their clocks are behind" + office = _answering(monkeypatch, httpx.Response(504, json={"ok": False, "stale": ["laptop"], "error": message})) + + with pytest.raises(ClockFollowersBehind) as behind: + await office.write_clock("advance", {"virtual_time": LATER}) + + assert (behind.value.stale, str(behind.value)) == (["laptop"], f"env 'office': clock advance: {message}") + + +@pytest.mark.asyncio +async def test_a_proxys_own_504_stays_the_error_it_is(monkeypatch): + office = _answering(monkeypatch, httpx.Response(504, text="Gateway Timeout")) + + with pytest.raises(ExtensionCallError) as failed: + await office.write_clock("advance", {"virtual_time": LATER}) + + assert failed.value.response.status_code == 504 and not isinstance(failed.value, ClockFollowersBehind)