From 68595189d7c18496c5ee1b227e3eda2dc27cbdf0 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 11 Nov 2026 21:04:20 -0800 Subject: [PATCH 1/2] feat(gateway): clock writes take the control token, a watch its computer's read token, and a waiting write names a computer that stopped watching (FD-5472) - /clock/set-time, /clock/set-rate, /clock/advance and /clock/clear are control routes: anyone who could reach a gateway could move its clock, an agent with a shell included. Core already sends the token (DeployedEnv.invoke); a gateway without one, as a Harbor export's, stays open. - /clock/watch takes the control token or a declared computer's read token, and a computer's token names that computer (401 without a token, 403 for an unknown one or another name). /clock/state and /clock/time stay open. - A write given wait_followers_s doesn't wait for a computer whose last watch is older than a hold and a reconnect (ClockFollowers.QUIET_S): it names it at once, its 504 carrying when each computer last watched. One that hasn't watched yet is still waited for. - multi_env_computers_test checks the gateway's control token is in no file, environment or command line on a computer. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 2 +- src/agent_env/env/gateway/clock_followers.py | 20 +++++- src/agent_env/env/gateway/constants.py | 10 ++- src/agent_env/env/gateway/gateway.py | 35 ++++++++-- .../computer/multi_env_computers_test.py | 41 ++++++++++- tst/unit/env/gateway/clock_watch_test.py | 68 +++++++++++++++++-- tst/unit/env/gateway/control_token_test.py | 20 +++--- 7 files changed, 173 insertions(+), 23 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 1eb15acde..5e8c770b2 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. With its control token (every deploy's), the clock's writes take it, and a watch takes it or a follower's read token under that follower's own name (401 without a token, 403 for an unknown one or another's name); the clock's time and state stay open. 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..4f647ebd7 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,18 @@ 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 holder_of(self, token: str) -> str | None: + """The declared follower whose read token ``token`` is, else None.""" + found = None + for name, expected in self._tokens.items(): + if token and hmac.compare_digest(token.encode(), expected.encode()): + found = name + return found + + def _quiet(self, name: str) -> bool: + seen = self._seen.get(name) + return seen is not None and time.monotonic() - seen > self.QUIET_S + 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,7 +86,8 @@ 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. One that hasn't watched yet is waited for: it may be starting.""" target = self._clock().count_of(version) def stale() -> list[str]: @@ -80,7 +96,7 @@ def stale() -> list[str]: async with self._acks: try: - await asyncio.wait_for(self._acks.wait_for(lambda: not stale()), wait_s) + await asyncio.wait_for(self._acks.wait_for(lambda: all(map(self._quiet, stale()))), wait_s) except asyncio.TimeoutError: pass return stale() diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py index 0a8e48ab6..ad406fa2c 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/src/agent_env/env/gateway/constants.py @@ -168,14 +168,18 @@ def sources(names: list[str]) -> str: WILDCARD = "*" TOOL_DISABLE_ACTION = "disable" TOOL_ENABLE_ACTION = "enable" -# The harness's routes that take the deploy's control token: the ones that change tool access, triggers and -# notifications. Every other route stays open for now, the clock's writes included: callers outside agent-env set the -# clock over plain HTTP and don't send the token yet. +# The harness's routes that take the deploy's control token: the ones that change tool access, triggers, notifications +# and the clock. Its reads stay open (the clock's state and time show only the universe's time, its rate and which +# version each computer shows), but for the watch, CLOCK_READ_ROUTES. CONTROL_ROUTES = frozenset({ f"/tools/{TOOL_DISABLE_ACTION}", f"/tools/{TOOL_ENABLE_ACTION}", "/triggers/register", "/triggers/remove", "/triggers/clear", "/notifications/config", "/notifications/take", "/notifications/clear", + "/clock/set-time", "/clock/set-rate", "/clock/advance", "/clock/clear", }) +# The routes that take the deploy's control token or a declared computer's read token: the watch, which is a +# computer's acknowledgement of the version it names, so a watch carrying a computer's token names that computer. +CLOCK_READ_ROUTES = frozenset({"/clock/watch"}) class GatewayMode(str, Enum): diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index bbf0296a9..ccb551384 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_READ_ROUTES, CONTROL_ROUTES, DATA_PLANE_LOAD_TIMEOUT_MAX_S, DEFAULT_ROLE, @@ -472,6 +473,10 @@ def _asgi_app(self): async def app(scope, receive, send): if self._control_bearer and scope["type"] in ("http", "websocket"): headers = scope.get("headers", ()) + if scope["type"] == "http" and scope["path"] in CLOCK_READ_ROUTES and ( + refusal := self._clock_read_refusal(scope)) is not None: + await refusal(scope, receive, send) + return if scope["type"] == "http" and scope["path"] in CONTROL_ROUTES and not hmac.compare_digest( dict(headers).get(b"authorization", b""), self._control_bearer): response = self._json_response({"ok": False, "error": "this route takes the env's control token"}, 401) @@ -496,6 +501,26 @@ async def app(scope, receive, send): return app + def _clock_read_refusal(self, scope) -> Response | None: + """Why a read of the clock is refused, or None when it may go on. It takes the gateway's control token or a + declared computer's read token, and a watch carrying a computer's token names that computer (a watch is its + acknowledgement of the version it names); 401 without a token, 403 for one that's neither or names another.""" + authorization = dict(scope.get("headers", ())).get(b"authorization", b"") + if hmac.compare_digest(authorization, self._control_bearer): + return None + if not authorization.startswith(b"Bearer "): + return self._json_response({"ok": False, "error": ( + "the clock's reads take the env's control token or a computer's read token")}, 401) + holder = self._clock_followers.holder_of(authorization[len(b"Bearer "):].decode(errors="replace").strip()) + if holder is None: + return self._json_response({"ok": False, "error": "the token is no computer's read token"}, 403) + if scope["path"] == "/clock/watch": + named = dict(parse_qsl(scope.get("query_string", b"").decode(errors="replace"))).get("follower", "") + if named != holder: + return self._json_response({"ok": False, "error": ( + f"a watch with {holder}'s read token names {holder} as its follower, not {named or 'none'}")}, 403) + return None + @asynccontextmanager async def _session_lifespan(self, mcp: FastMCP): """Once per inbound MCP session: the session's key joins the caller, and a connect gets to heal a failed @@ -612,7 +637,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: @@ -625,9 +651,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/integration/computer/multi_env_computers_test.py b/tst/integration/computer/multi_env_computers_test.py index 6524714d6..b404cdfb3 100644 --- a/tst/integration/computer/multi_env_computers_test.py +++ b/tst/integration/computer/multi_env_computers_test.py @@ -5,12 +5,15 @@ and a role rule narrows them as it would a server's; /svc// reaches the computer. A later process reattaches the computers from the multi env's record, and the run's teardown leaves nothing up. Two computers keep serving through a computer's restart and the gateway's, a failed redeploy leaves the earlier deployment serving, and a computer that fails -mid-deploy, or an entry the multi env refuses, leaves nothing up. Needs Docker and the local registry on localhost:5000.""" +mid-deploy, or an entry the multi env refuses, leaves nothing up. The gateway's control token, which moves its clock, +appears nowhere on a computer: in no file, and in no process's environment or command line. Needs Docker and the local +registry on localhost:5000.""" import asyncio import json import re import shutil +import subprocess import tempfile import uuid from pathlib import Path @@ -166,6 +169,40 @@ def _containers(sandbox_id: str) -> list[str]: return _docker("ps", "-aq", "--filter", f"name=agent-{sandbox_id}").stdout.split() +# Run in a computer: the files and the processes' environments and command lines holding the token read from stdin, as +# far as the user it runs as can read them (so the token is never in its own environment or command line). +TRACES = r"""import os, sys +token = sys.stdin.read().strip().encode() +found = [] +for root in ("/etc", "/opt", "/home", "/root", "/tmp", "/run", "/var", "/srv", "/usr/local"): + for folder, _, names in os.walk(root): + for name in names: + path = os.path.join(folder, name) + try: + if os.path.isfile(path) and os.path.getsize(path) < 50_000_000 and token in open(path, "rb").read(): + found.append(path) + except OSError: + pass +for pid in filter(str.isdigit, os.listdir("/proc")): + for part in ("environ", "cmdline"): + try: + if token in open(f"/proc/{pid}/{part}", "rb").read(): + found.append(f"/proc/{pid}/{part}") + except OSError: + pass +print("\n".join(found)) +""" + + +def _traces(sandbox_id: str, token: str) -> list[str]: + """Where in the computer's container ``token`` is, read as root and as the session user (uid 1000), whose processes' + environments root can't read there.""" + (container,) = _containers(sandbox_id) + return [line for user in ("0", "1000") for line in subprocess.run( + ["docker", "exec", "-i", "-u", user, container, "python3", "-c", TRACES], input=token, capture_output=True, + text=True, check=True).stdout.split()] + + @pytest.fixture def made(monkeypatch) -> list[str]: return local_sandboxes.recorded_sandbox_ids(monkeypatch) @@ -226,11 +263,13 @@ async def test_a_computer_in_a_multi_env_is_served_through_its_gateway_under_its narrowed, narrowed_shot = await _tools(deployed) reattached = await Env.from_instance_id(deployed.instance_id) up = _containers(computer["sandbox_id"]) + traces = _traces(computer["sandbox_id"], deployed.control_token) finally: report = await teardown_run(context) assert not report.failed + assert deployed.control_token and traces == [], traces assert deployed.get_child_env_card("desktop") is not None and card["name"] == "desktop" assert "list_items" in listed and "desktop_computer_screenshot" in listed and shot assert {name for name in listed if "computer_" in name} <= {f"desktop_{tool}" for tool in COMPUTER_TOOLS} diff --git a/tst/unit/env/gateway/clock_watch_test.py b/tst/unit/env/gateway/clock_watch_test.py index 303c1cd8a..c1b7b04aa 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, each watch under its computer's read token +on a gateway with a control token, and clock writes that wait for them, but not for one that has stopped watching.""" from __future__ import annotations import asyncio @@ -18,6 +19,7 @@ T0 = "2026-01-01T00:00:00Z" TOKENS = {"desktop": "token-desktop", "laptop": "token-laptop"} +CONTROL = "token-control" HELD_S = 5.0 # long enough that an answer "at once" can't be a held watch timing out, on a loaded runner too SHORT_HOLD_S = 0.3 @@ -32,9 +34,9 @@ def gateway(tmp_path, monkeypatch): monkeypatch.setattr(ClockFollowers, "WATCH_S", HELD_S) made = [] - def make(followers=TOKENS) -> Gateway: + def make(followers=TOKENS, control_token=None) -> Gateway: made.append(Gateway(host="127.0.0.1", port=0, server_name="unit", internal_mcp_servers=[], - clock_followers=followers)) + clock_followers=followers, control_token=control_token)) return made[-1] yield make @@ -165,7 +167,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 +366,60 @@ 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 +@pytest.mark.parametrize("token, follower, status", [ + ("token-laptop", "laptop", 200), ("token-laptop", "desktop", 403), ("token-laptop", None, 403), + ("wrong", "laptop", 403), (None, "laptop", 401), (CONTROL, None, 200), (CONTROL, "laptop", 200)], + ids=["its-own", "naming-another", "naming-none", "unknown", "none", "control", "control-naming-it"]) +async def test_a_guarded_gateways_watch_takes_its_computers_read_token_under_its_name(gateway, monkeypatch, token, + follower, status): + """With a control token, a watch takes it or a computer's read token, and a computer's names that computer; only + the computer's own acknowledges, and the control token reads without acknowledging.""" + monkeypatch.setattr(ClockFollowers, "WATCH_S", SHORT_HOLD_S) + gw = gateway(control_token=CONTROL) + async with _client(gw) as client: + armed = await client.put("/clock/set-time", json={"virtual_time": T0}, headers={"Authorization": f"Bearer {CONTROL}"}) + answered = await _watch(client, gw._clock.version, follower, token) + assert armed.status_code == 200 and answered.status_code == status, answered.text + own = (token, follower) == ("token-laptop", "laptop") + assert gw._clock_followers.state()["laptop"]["version"] == (gw._clock.version if own else None) + if (token, follower) == ("token-laptop", "desktop"): + assert answered.json()["error"] == "a watch with laptop's read token names laptop as its follower, not desktop" + + +@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 diff --git a/tst/unit/env/gateway/control_token_test.py b/tst/unit/env/gateway/control_token_test.py index b561ba1c9..3a78140da 100644 --- a/tst/unit/env/gateway/control_token_test.py +++ b/tst/unit/env/gateway/control_token_test.py @@ -1,4 +1,5 @@ -"""The gateway's control routes take the deploy's control token when it has one; every other route stays open.""" +"""The gateway's control routes, the clock's writes among them, take the deploy's control token when it has one; every +other route but the clock's watch (clock_watch_test) stays open, the clock's time and state included.""" from __future__ import annotations import httpx @@ -13,10 +14,10 @@ "/tools/disable": "POST", "/tools/enable": "POST", "/triggers/register": "POST", "/triggers/remove": "POST", "/triggers/clear": "POST", "/notifications/config": "POST", "/notifications/take": "POST", "/notifications/clear": "POST", + "/clock/set-time": "PUT", "/clock/set-rate": "PUT", "/clock/advance": "PUT", "/clock/clear": "POST", } OPEN = {"/state": "GET", "/step": "POST", "/clock/time": "GET", "/clock/state": "GET", "/triggers/state": "GET", - "/notifications/state": "GET", "/trajectory": "GET", "/changes": "GET", "/.well-known/agent-env.json": "GET", - "/clock/set-time": "PUT", "/clock/advance": "PUT", "/clock/clear": "POST"} + "/notifications/state": "GET", "/trajectory": "GET", "/changes": "GET", "/.well-known/agent-env.json": "GET"} def _client(token: str | None) -> httpx.AsyncClient: @@ -51,12 +52,15 @@ async def test_a_control_route_answers_a_call_with_the_token(path): @pytest.mark.asyncio -async def test_the_clock_is_armed_and_read_without_the_token(): +async def test_the_clock_is_armed_with_the_token_and_read_without_it(): async with _client(TOKEN) as client: - armed = await client.put("/clock/set-time", json={"virtual_time": "2019-06-15T12:00:00Z"}) - read = await client.get("/clock/time") - assert (armed.status_code, read.status_code) == (200, 200) - assert read.json()["virtual_time"].startswith("2019-06-15T12:00") + refused = await client.put("/clock/set-time", json={"virtual_time": "2019-06-15T12:00:00Z"}, + headers={"Authorization": "Bearer a-computers-read-token"}) + armed = await client.put("/clock/set-time", json={"virtual_time": "2019-06-15T12:00:00Z"}, + headers={"Authorization": f"Bearer {TOKEN}"}) + read, state = await client.get("/clock/time"), await client.get("/clock/state") + assert (refused.status_code, armed.status_code, read.status_code, state.status_code) == (401, 200, 200, 200) + assert read.json()["virtual_time"].startswith("2019-06-15T12:00") and state.json()["armed"] is True @pytest.mark.asyncio From 161420844367b7258bcf8467133a97d6f5f59a6c Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 11 Nov 2026 21:12:15 -0800 Subject: [PATCH 2/2] fix(gateway): a waiting write wakes when a computer it waits for goes quiet; name the token scan's size limit (FD-5472) stale_after only woke on an acknowledgement, so a computer that went quiet during the wait (its last watch 29 s before a 30 s wait) still held the write for all of it. It now also wakes at the next quiet deadline of a computer it waits for. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/env/gateway/clock_followers.py | 23 +++++++++++++------ .../computer/multi_env_computers_test.py | 3 ++- tst/unit/env/gateway/clock_watch_test.py | 17 ++++++++++++++ 3 files changed, 35 insertions(+), 8 deletions(-) diff --git a/src/agent_env/env/gateway/clock_followers.py b/src/agent_env/env/gateway/clock_followers.py index 4f647ebd7..9acc587b8 100644 --- a/src/agent_env/env/gateway/clock_followers.py +++ b/src/agent_env/env/gateway/clock_followers.py @@ -62,9 +62,14 @@ def holder_of(self, token: str) -> str | None: found = name return found - def _quiet(self, name: str) -> bool: + def _quiet_at(self, name: str) -> float | None: + """When ``name`` goes quiet (monotonic), if it has watched.""" seen = self._seen.get(name) - return seen is not None and time.monotonic() - seen > self.QUIET_S + 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 @@ -87,18 +92,22 @@ async def watch(self, after: str, follower: str | None) -> None: 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 has gone quiet (``QUIET_S``), - or ``wait_s`` has passed. One that hasn't watched yet is waited for: it may be starting.""" + 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: all(map(self._quiet, 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/tst/integration/computer/multi_env_computers_test.py b/tst/integration/computer/multi_env_computers_test.py index b404cdfb3..687fdd9b2 100644 --- a/tst/integration/computer/multi_env_computers_test.py +++ b/tst/integration/computer/multi_env_computers_test.py @@ -172,6 +172,7 @@ def _containers(sandbox_id: str) -> list[str]: # Run in a computer: the files and the processes' environments and command lines holding the token read from stdin, as # far as the user it runs as can read them (so the token is never in its own environment or command line). TRACES = r"""import os, sys +LARGEST_FILE_BYTES = 50_000_000 # past this a file is media or a database, not a config or a secret token = sys.stdin.read().strip().encode() found = [] for root in ("/etc", "/opt", "/home", "/root", "/tmp", "/run", "/var", "/srv", "/usr/local"): @@ -179,7 +180,7 @@ def _containers(sandbox_id: str) -> list[str]: for name in names: path = os.path.join(folder, name) try: - if os.path.isfile(path) and os.path.getsize(path) < 50_000_000 and token in open(path, "rb").read(): + if os.path.isfile(path) and os.path.getsize(path) < LARGEST_FILE_BYTES and token in open(path, "rb").read(): found.append(path) except OSError: pass diff --git a/tst/unit/env/gateway/clock_watch_test.py b/tst/unit/env/gateway/clock_watch_test.py index c1b7b04aa..d18061f63 100644 --- a/tst/unit/env/gateway/clock_watch_test.py +++ b/tst/unit/env/gateway/clock_watch_test.py @@ -423,3 +423,20 @@ async def test_a_write_waits_for_a_computer_that_hasnt_watched_yet(gateway): 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