From 6496c24f408b3c1d7808ecbc34a2f151d5210f8a Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 12 Nov 2026 08:00:57 -0800 Subject: [PATCH] feat(gateway): a waiting clock write names a computer that has stopped watching at once A follower holds a watch up to 25 s and reconnects at once, so one whose last watch is older than that and 5 s more (ClockFollowers.QUIET_S) has lost its connection. A write given wait_followers_s waited its whole wait for such a computer; it now names it at once, and one that goes quiet during the wait ends it at its quiet deadline. One that hasn't watched yet is still waited for, as it may be starting. The 504 carries each computer's last shown version and when it last watched. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 2 +- src/agent_env/env/gateway/clock_followers.py | 27 +++++++-- src/agent_env/env/gateway/gateway.py | 10 ++-- tst/unit/env/gateway/clock_watch_test.py | 59 +++++++++++++++++++- 4 files changed, 86 insertions(+), 12 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index c27ce0deb..76f35abfd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -122,7 +122,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; 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; 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`). | | `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/gateway/clock_followers.py b/src/agent_env/env/gateway/clock_followers.py index 0217488aa..5b89a351f 100644 --- a/src/agent_env/env/gateway/clock_followers.py +++ b/src/agent_env/env/gateway/clock_followers.py @@ -32,6 +32,9 @@ class ClockFollowers: run on the loop that serves it, never one asyncio.run apiece.""" WATCH_S = CLOCK_WATCH_HOLD_S + #: How long since a follower last watched before it has gone quiet: longer than a watch is held and a follower takes + #: to reconnect, so its connection is gone. A write doesn't wait for a quiet follower; it names it at once. + QUIET_S = CLOCK_WATCH_HOLD_S + 5 def __init__(self, clock: Callable[[], Clock], tokens: Mapping[str, str]): self._clock = clock @@ -51,6 +54,15 @@ def follower_for(self, name: str | None, token: str) -> str | None: expected = self._tokens.get(name or "") return name if expected and token and hmac.compare_digest(token.encode(), expected.encode()) else None + def _quiet_at(self, name: str) -> float | None: + """When ``name`` goes quiet (monotonic), if it has watched.""" + seen = self._seen.get(name) + return None if seen is None else seen + self.QUIET_S + + def _quiet(self, name: str) -> bool: + at = self._quiet_at(name) + return at is not None and time.monotonic() >= at + async def watch(self, after: str, follower: str | None) -> None: """Take ``after`` as ``follower``'s acknowledgement (one ``follower_for`` named) when it's a version this clock has had, later than the one it last showed; then return at once when the clock's version isn't ``after``, else @@ -71,18 +83,23 @@ async def watch(self, after: str, follower: str | None) -> None: pass async def stale_after(self, version: str, wait_s: float) -> list[str]: - """The followers that haven't shown ``version`` (or a later one) once each has or ``wait_s`` has passed.""" + """The followers that haven't shown ``version`` (or a later one) once each has, or has gone quiet (``QUIET_S``), + or ``wait_s`` has passed: woken by each acknowledgement and when the next follower waited for goes quiet. One + that hasn't watched yet is waited for: it may be starting.""" target = self._clock().count_of(version) + deadline = time.monotonic() + wait_s def stale() -> list[str]: return sorted(name for name, shown in self._shown.items() if shown is None or target is None or shown < target) async with self._acks: - try: - await asyncio.wait_for(self._acks.wait_for(lambda: not stale()), wait_s) - except asyncio.TimeoutError: - pass + while (waited_for := [name for name in stale() if not self._quiet(name)]) and time.monotonic() < deadline: + wake = min([deadline, *filter(None, map(self._quiet_at, waited_for))]) + try: + await asyncio.wait_for(self._acks.wait(), max(0.0, wake - time.monotonic())) + except asyncio.TimeoutError: + pass return stale() def state(self) -> dict: diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index 1c3058b41..5b045c315 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -630,7 +630,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.""" + ``_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.""" try: body = await request.json() if not body_optional or await request.body() else {} except Exception: @@ -643,9 +644,10 @@ 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, "error": ( - f"the clock is at version {state['version']}, but {', '.join(stale)} didn't show it within " - f"{wait:g} s: their clocks are behind")}, 504) + 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) return self._json_response(state, 200) async def _change_clock(self, change, wait_s: float = 0.0) -> tuple[dict, list[str]]: diff --git a/tst/unit/env/gateway/clock_watch_test.py b/tst/unit/env/gateway/clock_watch_test.py index 303c1cd8a..018e5bee3 100644 --- a/tst/unit/env/gateway/clock_watch_test.py +++ b/tst/unit/env/gateway/clock_watch_test.py @@ -1,4 +1,5 @@ -"""The /clock/watch long poll computers follow the gateway's clock through, and clock writes that wait for them.""" +"""The /clock/watch long poll computers follow the gateway's clock through, and clock writes that wait for them, but not +for one that has stopped watching.""" from __future__ import annotations import asyncio @@ -165,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 within 0.4 s" in body["error"] + assert "laptop didn't show it (waited up to 0.4 s;" in body["error"] + assert body["followers"]["laptop"] == {"version": None, "seen_s_ago": None} assert read == {"virtual_time": T0} @@ -363,3 +365,56 @@ async def test_a_change_on_the_arm_wakes_the_triggers_and_a_re_arm_leaves_them_t advanced = (await client.put("/clock/advance", json={"virtual_time": "2026-02-01T00:00:00Z"})).json() cleared = (await client.post("/clock/clear")).json() assert ["due" in answer for answer in (armed, rated, advanced, cleared)] == [False, True, True, False] + + +@pytest.mark.asyncio +async def test_a_write_doesnt_wait_for_a_computer_that_has_stopped_watching(gateway, monkeypatch): + """A computer whose last watch is older than a hold and a reconnect (its connection gone) is named at once, with + when it last watched; the others are waited for as before.""" + monkeypatch.setattr(ClockFollowers, "WATCH_S", SHORT_HOLD_S) + monkeypatch.setattr(ClockFollowers, "QUIET_S", SHORT_HOLD_S + 0.5) + gw = gateway() + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + follower = asyncio.create_task(_follow(client, "desktop", TOKENS["desktop"])) + try: + await _watch(client, gw._clock.version, "laptop", TOKENS["laptop"]) # shows it, then stops watching + await _until_shown(gw, "desktop", "laptop") + await asyncio.sleep(ClockFollowers.QUIET_S + 0.2) + started = time.monotonic() + answer = await client.put("/clock/advance", json={"virtual_time": "2026-01-02T00:00:00Z", "wait_followers_s": 10}) + waited = time.monotonic() - started + finally: + follower.cancel() + body = answer.json() + assert answer.status_code == 504 and body["stale"] == ["laptop"] and waited < 3, (waited, body) + assert body["followers"]["laptop"]["seen_s_ago"] > ClockFollowers.QUIET_S + assert body["followers"]["desktop"]["version"] == gw._clock.version + + +@pytest.mark.asyncio +async def test_a_write_waits_for_a_computer_that_hasnt_watched_yet(gateway): + """One that hasn't watched may be starting: the write waits for it, all its wait_followers_s.""" + gw = gateway({"laptop": TOKENS["laptop"]}) + async with _client(gw) as client: + started = time.monotonic() + answer = await client.put("/clock/set-time", json={"virtual_time": T0, "wait_followers_s": 0.5}) + waited = time.monotonic() - started + assert answer.status_code == 504 and answer.json()["stale"] == ["laptop"] and waited >= 0.5 + + +@pytest.mark.asyncio +async def test_a_write_stops_waiting_for_a_computer_when_it_goes_quiet_during_the_wait(gateway, monkeypatch): + """One that watched shortly before the write is waited for until its quiet deadline, not for the whole wait.""" + monkeypatch.setattr(ClockFollowers, "WATCH_S", SHORT_HOLD_S) + monkeypatch.setattr(ClockFollowers, "QUIET_S", 1.0) + gw = gateway({"laptop": TOKENS["laptop"]}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + await _watch(client, "", "laptop", TOKENS["laptop"]) # answered at once: laptop watched, then stops + await asyncio.sleep(0.3) + started = time.monotonic() + answer = await client.put("/clock/advance", json={"virtual_time": "2026-01-02T00:00:00Z", "wait_followers_s": 10}) + waited = time.monotonic() - started + assert answer.status_code == 504 and answer.json()["stale"] == ["laptop"], answer.text + assert 0.5 <= waited < 3, waited