diff --git a/AGENTS.md b/AGENTS.md index 8d99e82cf..eaa4834f3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -121,7 +121,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. 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._clock_changed`, which wakes the triggers, answers the held watches and waits for the followers. The computers that follow it (`CLOCK_FOLLOWER_TOKENS`, each child whose card serves `clock/v1` `sync_time`, by its read token, its control token for now) long-poll `/clock/watch` naming the version they show, each watch held up to the card's `hold_s` (`env/gateway/clock_followers.py`), and a write given `wait_followers_s` waits for each, answering 504 naming those behind; a restarted gateway is unarmed. While its clock is armed it serves agents `get_time` (`env/gateway/get_time.py`), which answers in UTC and, given an IANA `time_zone`, also in that zone; executors' working `hours` (`env/gateway/triggers.py`) take a fixed `utc_offset` or an IANA `time_zone`. | | `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Every artifact but an environment universe, whose `metadata` key holds its legacy metadata file ids, carries a `metadata` dict outside its versioned fields, changed by compare-and-swap (`merge_metadata`). Built-ins: `CliArtifact` (`cli`), `ComputerArtifact` (`computer`, a base computer: its architectures, OS, session user, settings and container form, whose image `put` reads from its registry into `metadata.validated_images`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`: its environment's files by path, stored as the one wrapped `file_artifact_ref` every environment artifact had when it has one file at its filename, else as `file_artifact_refs`; `get_file_artifacts()` reads either, and a load hands each file to the env's `data/add` with its path, which a pre-v1 load, taking one file, refuses), `EnvironmentUniverseArtifact` (`environment_universe`), `SkillArtifact` (`skill`); `[artifacts] type_aliases` maps another stored spelling of a type onto one of them. A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or a published image's tag with its `image_digest` and its build context, which sandboxes pull by the digest and build from the context when they can't (the infra images, and a bundle agent built from a shipped agent's sources, `Plan.published`), or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/docker/context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. | | `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, among them `start_computer_recording` and `end_computer_recording`, and more). On an env whose card serves computer-exec/v1 (a computer), `run_code` with `env_id` and `run_container_unit_tests_verifier` with `env_id` run inside it as its session user through the driver: files staged with `data/add` and read with `data/get`, commands run with `run`; `env_outcome_verifier` hands a `verify` that takes an `env_url` keyword the env's address. A computer's framework-only routes (everything but MCP and the card) take the per-deploy control token its provider boots it with (`DeployedEnv.control_token`, `AGENTENV_CONTROL_TOKEN`); framework calls send `deployed.control_headers`, `Task.run` runs every step under `protocol_v1.authorized` with each deployed env's (a `verify` included), and a `run_code` script inside the computer gets `input["env"]["headers"]`. `teardown_run` and the `teardown_sandboxes` step share one core (`task/teardown.py` `tear_down`): the recordings the run left open end with fresh grants, each env still up closes (`Env.close`; a computer env ends any recording still open), what stayed open is registered from its namespace, then every sandbox is terminated by id. `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). | | `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/envs/_multi_computers.py b/src/agent_env/env/envs/_multi_computers.py index 552b87ced..cf9e1d5fd 100644 --- a/src/agent_env/env/envs/_multi_computers.py +++ b/src/agent_env/env/envs/_multi_computers.py @@ -16,6 +16,7 @@ from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env from agent_env.env.envs.computer.utils import sites from agent_env.env.gateway import ExternalChild +from agent_env.env.gateway.constants import EXT_CLOCK_URI from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_env_sandbox_provider if TYPE_CHECKING: @@ -93,6 +94,12 @@ def routes_later(entry: Optional[dict]) -> bool: return isinstance(params, dict) and bool(params.get("site_routing")) and not params.get("sites") +def _clock_read_token(record: DeployedEnv) -> str | None: + """The read token a computer's clock watch carries when its card says it follows a clock (``clock/v1``'s + ``sync_time``): for now its control token, which a token of its own can replace without the gateway knowing.""" + return record.control_token if record.supports(EXT_CLOCK_URI, "sync_time") else None + + def refusal(computer: ComputerEnv, entry: dict, sandbox_type: str | None, websites: Mapping[str, str]) -> str | None: """Why ``computer`` wouldn't deploy in a multi env with ``entry``, its child_envs entry: why it wouldn't alone, ``site_routing`` naming none of the multi env's ``websites`` (their environment names by env id), or a sandbox @@ -213,7 +220,7 @@ async def deploy(reach: Optional[str]) -> list[ExternalChild]: for error in failed[1:]: logger.warning(f"multi env '{self._owner}': another computer failed to deploy too: {error!r}") raise failed[0] - return [ExternalChild(env.environment_name, record.mcp_url, record.environment_url) + return [ExternalChild(env.environment_name, record.mcp_url, record.environment_url, _clock_read_token(record)) for env, record in zip(self.every, results)] return deploy diff --git a/src/agent_env/env/gateway/__init__.py b/src/agent_env/env/gateway/__init__.py index 7a4123f58..9bee37181 100644 --- a/src/agent_env/env/gateway/__init__.py +++ b/src/agent_env/env/gateway/__init__.py @@ -30,11 +30,13 @@ class InternalMCPServer: @dataclass(frozen=True) class ExternalChild: """A child env on a sandbox of its own, which the gateway fronts as it does an MCP server: its name, and its MCP - endpoint and base URL as the process holding it reaches them.""" + endpoint and base URL as the process holding it reaches them. A computer that follows the gateway's clock has the + read token its clock watch carries, and the gateway's clock writes wait for it.""" name: str mcp_url: str base_url: str + clock_read_token: str | None = None __all__ = [ diff --git a/src/agent_env/env/gateway/clock.py b/src/agent_env/env/gateway/clock.py index cceb14364..bb046e9d9 100644 --- a/src/agent_env/env/gateway/clock.py +++ b/src/agent_env/env/gateway/clock.py @@ -7,6 +7,7 @@ import math import re +import secrets import threading import time from datetime import datetime, timedelta, timezone @@ -28,13 +29,14 @@ class ClockError(ValueError): def _parse_rate(rate) -> float: if isinstance(rate, bool) or not isinstance(rate, (int, float)): raise ClockError(f"rate must be a number, got {type(rate).__name__}") - # NaN passes both ordered bounds below, then poisons every later read with an uncatchable 500. - if not math.isfinite(rate): + # NaN passes both ordered bounds below, then poisons every later read with an uncatchable 500; only a float can be + # NaN or infinite, and math.isfinite on an int past a float's range raises. + if isinstance(rate, float) and not math.isfinite(rate): raise ClockError(f"rate must be a finite number, got {rate}") if rate < 0: raise ClockError("rate must be non-negative") if rate > MAX_RATE: - raise ClockError(f"rate {rate:g} exceeds max {MAX_RATE:g} (1 real s = 1 virtual day)") + raise ClockError(f"rate exceeds max {MAX_RATE:g} (1 real s = 1 virtual day)") return float(rate) @@ -125,6 +127,8 @@ def _advance(t0: datetime, seconds: float) -> datetime: class Clock: """Gateway-owned virtual clock — wall-clock-driven, monotonic, threadsafe; `now_fn` injectable for tests.""" + BOOT_ID_BYTES = 6 # a version's per-boot id, random, so a restarted gateway's versions never repeat its last boot's + def __init__(self, now_fn=time.monotonic): self._now = now_fn self._lock = threading.Lock() @@ -135,11 +139,26 @@ def __init__(self, now_fn=time.monotonic): self._generation = 0 self._lived = 0.0 # virtual seconds run before the current arm/clear self._ahead = 0.0 # virtual seconds the current arm was advanced by + self._boot = secrets.token_hex(self.BOOT_ID_BYTES) + self._count = 0 @property def armed(self) -> bool: return self._armed + @property + def version(self) -> str: + """Opaque, changed by every write: this clock's own id and a count, so it never repeats across restarts.""" + return f"{self._boot}.{self._count}" + + def count_of(self, version) -> int | None: + """The count of ``version`` when this clock has had it; None for another clock's, a later one, or junk.""" + boot, _, count = version.partition(".") if isinstance(version, str) else ("", "", "") + current = str(self._count) + if boot != self._boot or not (count.isascii() and count.isdigit()) or len(count) > len(current): + return None + return int(count) if int(count) <= self._count else None + @property def rate(self) -> float: """Virtual seconds per real second; 1.0 while unarmed.""" @@ -168,6 +187,25 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict: self._ahead = 0.0 self._armed = True self._generation += 1 + self._count += 1 + return self.state() + + def set_rate(self, rate) -> dict: + """Change an armed clock's rate from its time at the call, keeping t0 and generation as an advance does, so time + triggers keep their marks and a rate change can't step the clock back by a time read a moment earlier.""" + r = _parse_rate(rate) + with self._lock: + if not self._armed or self._t0 is None: + raise ClockError("clock is not armed") + now = self._now() + ran = (now - self._anchor) * self._rate + if _advance(self._t0, ran + self._ahead) > _MAX_T0 - timedelta(seconds=_OVERFLOW_HORIZON_S * r): + raise ClockError(f"rate {r:g} would overflow datetime near the max") + self._lived += ran + self._ahead += ran + self._anchor = now + self._rate = r + self._count += 1 return self.state() def advance(self, virtual_time: str) -> dict: @@ -184,6 +222,7 @@ def advance(self, virtual_time: str) -> dict: raise ClockError(f"advance only moves forward: {virtual_time!r} is before the current {_iso(current)}") self._ahead += jump self._lived += jump + self._count += 1 return self.state() def clear(self) -> dict: @@ -195,7 +234,8 @@ def clear(self) -> dict: self._rate = 1.0 self._ahead = 0.0 self._generation += 1 - return {"ok": True, "armed": False} + self._count += 1 + return {"ok": True, "armed": False, "version": self.version} def elapsed(self) -> float: """Virtual seconds run so far: real time at each arm's rate (1 unarmed), plus each forward re-arm or advance.""" @@ -226,10 +266,11 @@ def read(self) -> dict: def state(self) -> dict: with self._lock: if not self._armed or self._t0 is None: - return {"armed": False} + return {"armed": False, "version": self.version} elapsed = self._now() - self._anchor return {"armed": True, "t0": _iso(self._t0), "virtual_seconds_per_real_second": self._rate, - "virtual_time": _iso(_advance(self._t0, elapsed * self._rate + self._ahead))} + "virtual_time": _iso(_advance(self._t0, elapsed * self._rate + self._ahead)), + "version": self.version} def _bank(self, t0: datetime | None = None) -> None: """Fold the stretch since the last arm/clear into `_lived` at its rate, and start the next one now.""" diff --git a/src/agent_env/env/gateway/clock_followers.py b/src/agent_env/env/gateway/clock_followers.py new file mode 100644 index 000000000..139497c2a --- /dev/null +++ b/src/agent_env/env/gateway/clock_followers.py @@ -0,0 +1,91 @@ +"""The computers that follow the gateway's clock: each long-polls ``/clock/watch`` with the version it shows, which is +its acknowledgement, and a clock write can wait until every one shows the write's version.""" +from __future__ import annotations + +import asyncio +import hmac +import math +import time +from collections.abc import Callable, Mapping + +from .clock import Clock, ClockError +from .constants import CLOCK_MAX_WAIT_FOLLOWERS_S, CLOCK_WATCH_HOLD_S + + +def parse_wait(value) -> float: + """A write's ``wait_followers_s``: how long it waits for the followers, 0 when absent.""" + if value is None: + return 0.0 + # Only a float can be NaN or infinite; math.isfinite on an int past a float's range raises. + if isinstance(value, bool) or not isinstance(value, (int, float)) or (isinstance(value, float) and not math.isfinite(value)) \ + or value < 0: + raise ClockError(f"wait_followers_s must be a non-negative number, got {value!r}") + if value > CLOCK_MAX_WAIT_FOLLOWERS_S: + raise ClockError(f"wait_followers_s exceeds max {CLOCK_MAX_WAIT_FOLLOWERS_S:g}") + return float(value) + + +class ClockFollowers: + """The declared followers, each by name with the read token its watch carries, and the version each last showed; + bound to the gateway's event loop.""" + + WATCH_S = CLOCK_WATCH_HOLD_S + + def __init__(self, clock: Callable[[], Clock], tokens: Mapping[str, str]): + self._clock = clock + self._tokens = dict(tokens) + self._shown: dict[str, tuple[int, str] | None] = dict.fromkeys(self._tokens) + self._seen: dict[str, float] = {} + self._changed = asyncio.Event() + self._acks = asyncio.Condition() + + def changed(self) -> None: + """Answer every held watch: the clock has a new version.""" + self._changed.set() + self._changed = asyncio.Event() + + def follower_for(self, name: str | None, token: str) -> str | None: + """``name`` when it's a declared follower and ``token`` its read token, else None.""" + expected = self._tokens.get(name or "") + return name if expected and token and hmac.compare_digest(token.encode(), expected.encode()) else None + + 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 + at its next change or after ``WATCH_S``.""" + changed = self._changed + if follower is not None: + count = self._clock().count_of(after) + async with self._acks: + shown = self._shown[follower] + if count is not None and (shown is None or count > shown[0]): + self._shown[follower] = (count, after) + self._seen[follower] = time.monotonic() + self._acks.notify_all() + if after == self._clock().version: + try: + await asyncio.wait_for(changed.wait(), self.WATCH_S) + except asyncio.TimeoutError: + 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.""" + target = self._clock().count_of(version) + + def stale() -> list[str]: + return sorted(name for name, shown in self._shown.items() + if shown is None or target is None or shown[0] < target) + + async with self._acks: + try: + await asyncio.wait_for(self._acks.wait_for(lambda: not stale()), wait_s) + except asyncio.TimeoutError: + pass + return stale() + + def state(self) -> dict: + """Each follower's last shown version and how long ago it last watched, for ``/clock/state``.""" + now = time.monotonic() + return {name: {"version": shown[1] if shown else None, + "seen_s_ago": round(now - self._seen[name], 3) if name in self._seen else None} + for name, shown in self._shown.items()} diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py index 37af9456c..27fc4ad4e 100644 --- a/src/agent_env/env/gateway/constants.py +++ b/src/agent_env/env/gateway/constants.py @@ -69,6 +69,14 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int: METHOD_GET = "data/get" MCP_TRANSPORT = "mcp" CONTROL_TOKEN_ENV = "AGENTENV_CONTROL_TOKEN" +# The computers that follow the gateway's clock, each as `name=read token`, comma-separated; TOKEN in the name keeps +# it out of logged compose files. +CLOCK_FOLLOWERS_ENV = "CLOCK_FOLLOWER_TOKENS" +# How long /clock/watch holds a watch naming the current version before answering it unchanged; a follower's request +# timeout must be longer. The card's watch method advertises it as hold_s. +CLOCK_WATCH_HOLD_S = 25.0 +# The longest a clock write's wait_followers_s may ask it to wait for the followers. +CLOCK_MAX_WAIT_FOLLOWERS_S = 30.0 # Env/gateway-level extensions the gateway itself owns (tool gating + triggers). EXT_DISABLE_TOOL_URI = "urn:agentenv:disable-tool/v1" @@ -201,6 +209,8 @@ def random_mcp_server_name() -> str: "if": {"properties": {"action": {"const": "call_tool"}}}, "then": {"required": ["tool_name"]}, } +_WAIT_FOLLOWERS = {"wait_followers_s": {"type": "number", "minimum": 0, "maximum": CLOCK_MAX_WAIT_FOLLOWERS_S}} + GATEWAY_EXTENSIONS = [ { "uri": EXT_DISABLE_TOOL_URI, @@ -226,13 +236,15 @@ def random_mcp_server_name() -> str: }, { "uri": EXT_CLOCK_URI, - "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set-time/advance/clear are the harness control path.", + "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set_time/set_rate/advance/clear are the harness control path, each changing the clock's opaque version and, given wait_followers_s, answering 504 naming the computers following the clock that don't show it in time (set_rate keeps t0, as advance does); a computer follows it through watch, a long poll naming the version it shows.", "params": {"endpoint": "/clock/time", "methods": { "get_time": {"method": "GET", "endpoint": "/clock/time"}, - "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}}, "required": ["virtual_time"]}}, - "advance": {"method": "PUT", "endpoint": "/clock/advance", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}}, "required": ["virtual_time"]}}, - "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {}}}, - "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}}}}, + "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}}, + "set_rate": {"method": "PUT", "endpoint": "/clock/set-rate", "request": {"type": "object", "properties": {"virtual_seconds_per_real_second": {"type": "number"}, **_WAIT_FOLLOWERS}, "required": ["virtual_seconds_per_real_second"]}}, + "advance": {"method": "PUT", "endpoint": "/clock/advance", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}}, + "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {**_WAIT_FOLLOWERS}}}, + "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}}, + "watch": {"method": "GET", "endpoint": "/clock/watch", "hold_s": CLOCK_WATCH_HOLD_S, "request": {"type": "object", "properties": {"after": {"type": "string"}, "follower": {"type": "string"}}}}}}, "required": False, }, { diff --git a/src/agent_env/env/gateway/entrypoint.py b/src/agent_env/env/gateway/entrypoint.py index 1437372ff..0843b75b1 100644 --- a/src/agent_env/env/gateway/entrypoint.py +++ b/src/agent_env/env/gateway/entrypoint.py @@ -8,7 +8,7 @@ import httpx from . import AGENT_ENV_GATEWAY_MCP_PORT, GatewayMode, InternalMCPServer -from .constants import (CONTROL_TOKEN_ENV, DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV, MCP_TRANSPORT, +from .constants import (CLOCK_FOLLOWERS_ENV, CONTROL_TOKEN_ENV, DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV, MCP_TRANSPORT, WELL_KNOWN_PATH) from .gateway import Gateway, card_url @@ -188,6 +188,19 @@ def parse_name_url_map(env_str: str) -> dict[str, str]: return urls +def parse_clock_followers(spec: str) -> dict[str, str]: + """``CLOCK_FOLLOWER_TOKENS``: each computer that follows the clock as ``name=read token``, comma-separated.""" + followers: dict[str, str] = {} + for entry in filter(None, (part.strip() for part in spec.split(","))): + name, _, token = (part.strip() for part in entry.partition("=")) + if not name or not token: + raise ValueError(f"invalid {CLOCK_FOLLOWERS_ENV} entry for {name or '?'!r}: expected 'name=read token'") + if name in followers: + raise ValueError(f"{CLOCK_FOLLOWERS_ENV} names {name!r} twice") + followers[name] = token + return followers + + if __name__ == "__main__": logging.basicConfig( level=logging.INFO, @@ -218,5 +231,6 @@ def parse_name_url_map(env_str: str) -> dict[str, str]: service_db_url=service_db_url, control_token=os.environ.get(CONTROL_TOKEN_ENV), provider_headers=os.environ.get(GATEWAY_PROVIDER_HEADERS_ENV, "").split(","), + clock_followers=parse_clock_followers(os.environ.get(CLOCK_FOLLOWERS_ENV, "")), ) gateway.run() diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py index 3b257b056..3130bdb04 100644 --- a/src/agent_env/env/gateway/gateway.py +++ b/src/agent_env/env/gateway/gateway.py @@ -10,7 +10,8 @@ import posixpath import socket import subprocess -from collections.abc import Iterable +import time +from collections.abc import Iterable, Mapping from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path @@ -75,6 +76,7 @@ from .changes import ChangeAttribution from .child_sessions import ChildSessions, is_dead_session from .clock import Clock, ClockError, _iso +from .clock_followers import ClockFollowers, parse_wait from .get_time import GET_TIME_TOOL_NAME, build_get_time_tool from .rest_proxy import ( ROLE_HEADER_BYTES, @@ -142,9 +144,11 @@ def __init__( service_db_url: str | None = None, control_token: str | None = None, provider_headers: Iterable[str] = (), + clock_followers: Mapping[str, str] | None = None, ): """`provider_headers` names the headers the sandbox provider in front of the gateway is sent, its credentials, - which the gateway takes off every request it forwards to a service behind it.""" + which the gateway takes off every request it forwards to a service behind it. `clock_followers` names the + computers that follow its clock, each with the read token its watch carries.""" self.host = host self.port = port self.server_name = server_name @@ -204,6 +208,7 @@ async def list_website_urls() -> list[str]: ) ) self._clock = Clock() + self._clock_followers = ClockFollowers(lambda: self._clock, clock_followers or {}) self._notifications = Notifications(lambda: self._clock) self._env_get_time_url_cache: str | None = None self._install_role_filters() @@ -309,39 +314,42 @@ async def triggers_state(request: Request) -> Response: @self._mcp.custom_route("/clock/set-time", methods=["PUT"]) async def clock_set_time(request: Request) -> Response: - try: - body = await request.json() - except Exception: - return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400) if GET_TIME_TOOL_NAME in self._tool_server_urls: return self._json_response( {"ok": False, "error": f"'{GET_TIME_TOOL_NAME}' is already registered by a backing server"}, 409) - try: + + def arm(body: dict) -> dict: state = self._clock.set_time(body.get("virtual_time"), body.get("virtual_seconds_per_real_second")) - except ClockError as e: - return self._json_response({"ok": False, "error": str(e)}, 400) - self._register_get_time_tool() - return self._json_response(state, 200) + self._register_get_time_tool() + return state + + return await self._clock_write(request, arm, time_moved=False) + + @self._mcp.custom_route("/clock/set-rate", methods=["PUT"]) + async def clock_set_rate(request: Request) -> Response: + return await self._clock_write( + request, lambda body: self._clock.set_rate(body.get("virtual_seconds_per_real_second")), time_moved=True) @self._mcp.custom_route("/clock/advance", methods=["PUT"]) async def clock_advance(request: Request) -> Response: - try: - body = await request.json() - except Exception: - return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400) - if not isinstance(body, dict): - return self._json_response({"ok": False, "error": "body must be a JSON object"}, 400) - try: - state = self._clock.advance(body.get("virtual_time")) - except ClockError as e: - return self._json_response({"ok": False, "error": str(e)}, 400) - return self._json_response({**state, "due": self._trigger_engine.clock_advanced()}, 200) + return await self._clock_write(request, lambda body: self._clock.advance(body.get("virtual_time")), + time_moved=True) @self._mcp.custom_route("/clock/clear", methods=["POST"]) async def clock_clear(request: Request) -> Response: - self._unregister_get_time_tool() - state = self._clock.clear() - return self._json_response(state, 200) + def disarm(body: dict) -> dict: + self._unregister_get_time_tool() + return self._clock.clear() + + return await self._clock_write(request, disarm, time_moved=False, body_optional=True) + + @self._mcp.custom_route("/clock/watch", methods=["GET"]) + async def clock_watch(request: Request) -> Response: + received = time.time() + token = request.headers.get("authorization", "").removeprefix("Bearer ").strip() + follower = self._clock_followers.follower_for(request.query_params.get("follower"), token) + await self._clock_followers.watch(request.query_params.get("after", ""), follower) + return self._json_response({**self._clock.state(), "received": received, "sent": time.time()}, 200) @self._mcp.custom_route("/clock/time", methods=["GET"]) async def clock_time(request: Request) -> Response: @@ -355,7 +363,7 @@ async def clock_state(request: Request) -> Response: st = self._clock.state() if st.get("armed"): st["env_get_time_url"] = await self._env_get_time_url() - return self._json_response(st, 200) + return self._json_response({**st, "followers": self._clock_followers.state()}, 200) async def _notifications_write(request: Request, method) -> Response: try: @@ -603,6 +611,39 @@ async def _await_barriers(self, pending: list[asyncio.Task], timeout_s: float | except Exception: logger.exception("Awaiting trigger barriers failed (fail-open)") + async def _clock_write(self, request: Request, change, *, time_moved: bool, body_optional: bool = False) -> Response: + """A clock write route: its JSON object body (none at all when ``body_optional``) applied by ``change``, then + ``_clock_changed``; 400 for a body or value the clock refuses, else 200, or 504 naming the computers behind.""" + try: + body = await request.json() if not body_optional or await request.body() else {} + except Exception: + return self._json_response({"ok": False, "error": "Invalid JSON body"}, 400) + if not isinstance(body, dict): + return self._json_response({"ok": False, "error": "body must be a JSON object"}, 400) + try: + wait = parse_wait(body.get("wait_followers_s")) + state = change(body) + except ClockError as e: + return self._json_response({"ok": False, "error": str(e)}, 400) + state = await self._clock_changed(state, wait, time_moved=time_moved) + if stale := state.get("stale"): + return self._json_response({**state, "ok": False, "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, 200) + + async def _clock_changed(self, state: dict, wait_s: float = 0.0, *, time_moved: bool) -> dict: + """What every change of the clock owes the rest of the gateway, so each one makes it here: the time triggers + re-evaluated when its time moved on its arm (an advance or a rate change; a re-arm or clear re-anchors them by + its generation), every held watch answered, and, given ``wait_s``, a wait for each declared computer to show its + version. ``state`` as the change left it, with the triggers it made ``due`` and the computers still ``stale``.""" + if time_moved: + state = {**state, "due": self._trigger_engine.clock_advanced()} + self._clock_followers.changed() + if wait_s and (stale := await self._clock_followers.stale_after(state["version"], wait_s)): + return {**state, "stale": stale} + return state + async def _env_get_time_url(self) -> str: """The gateway's own server-reachable /clock/time URL (cached, resolved off the event loop): modal i6pn if resolvable, else the compose service name.""" if self._env_get_time_url_cache is None: diff --git a/src/agent_env/providers/env_providers/env_gateway_provider.py b/src/agent_env/providers/env_providers/env_gateway_provider.py index 6537a4a99..4cf431a0f 100644 --- a/src/agent_env/providers/env_providers/env_gateway_provider.py +++ b/src/agent_env/providers/env_providers/env_gateway_provider.py @@ -12,7 +12,8 @@ import shlex import time from collections import Counter -from dataclasses import dataclass, field +from collections.abc import Iterable +from dataclasses import dataclass, field, replace from datetime import datetime, timezone from typing import TYPE_CHECKING, Optional @@ -33,6 +34,7 @@ from agent_env.env.env import DeployedGatewayEnv, _mcp_url from agent_env.env.gateway import AGENT_ENV_GATEWAY_MCP_PORT, ExternalChild, GatewayMode from agent_env.env.gateway.constants import ( + CLOCK_FOLLOWERS_ENV, DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV, STATE_TOOL_NAME_CLASHES_KEY, @@ -533,6 +535,7 @@ def create_docker_compose( gateway_env.append(f" - GATEWAY_MODE={gateway_mode.value}") gateway_env.append(f" - SERVICE_DB_URL={state_provider.base_url(state_instance)}") gateway_env.extend(f" - {key}={value}" for key, value in control_environment(control_token).items()) + gateway_env.extend(f" - {key}={value}" for key, value in clock_followers_environment(external_children).items()) if provider_headers := sandbox_request_header_names(): gateway_env.append(f" - {GATEWAY_PROVIDER_HEADERS_ENV}={','.join(provider_headers)}") publish = host_port or (lambda port: port) @@ -912,8 +915,8 @@ async def _stand_up(_spec): for result in (up, started): if isinstance(result, BaseException): raise result - external = tuple(ExternalChild(child.name, sandbox.url_from_sandbox(child.mcp_url), - sandbox.url_from_sandbox(child.base_url)) for child in started) + external = tuple(replace(child, mcp_url=sandbox.url_from_sandbox(child.mcp_url), + base_url=sandbox.url_from_sandbox(child.base_url)) for child in started) await self._write_file(sandbox, DOCKER_COMPOSE_PATH, compose(external_children=external)) await self._compose_up(sandbox) @@ -1109,6 +1112,7 @@ async def _deploy_via_containers( "SERVICE_DB_URL": self._state_provider.base_url(state_instance), "MCP_SERVER_NAME": mcp_server_name or DEFAULT_MCP_SERVER_NAME, **control_environment(control_token), + **clock_followers_environment(external), **({GATEWAY_PROVIDER_HEADERS_ENV: ",".join(names)} if (names := sandbox_request_header_names()) else {}), } gateway_sb = await sandbox_provider.create_container( @@ -1722,6 +1726,12 @@ def _redact_compose_secrets(compose_content: str) -> str: ) +def clock_followers_environment(children: Iterable[ExternalChild]) -> dict[str, str]: + """The gateway's declared clock followers: each child that follows its clock, by the read token its watch carries.""" + followers = ",".join(f"{child.name}={child.clock_read_token}" for child in children if child.clock_read_token) + return {CLOCK_FOLLOWERS_ENV: followers} if followers else {} + + def _frontend_urls(gateway_url: str, website_configs: Optional[list[WebsiteConfig]]) -> Optional[dict[str, str]]: """Each synthetic site's page route through the gateway, by its environment name; None with none, as when every site is live.""" diff --git a/tst/env/gateway/clock_test.py b/tst/env/gateway/clock_test.py index a431238f4..b5969d898 100644 --- a/tst/env/gateway/clock_test.py +++ b/tst/env/gateway/clock_test.py @@ -119,7 +119,7 @@ def test_monotonic_non_decreasing(): def test_state_reports_fields(): now = FakeNow() c = Clock(now_fn=now) - assert c.state() == {"armed": False} + assert c.state() == {"armed": False, "version": c.version} st = c.set_time(T0, rate=3600) assert st["armed"] is True and st["virtual_seconds_per_real_second"] == 3600.0 and st["t0"] == T0 and st["virtual_time"] == T0 now.tick(1) diff --git a/tst/integration/env/gateway/gateway_test.py b/tst/integration/env/gateway/gateway_test.py index 58790c935..f381bba7c 100644 --- a/tst/integration/env/gateway/gateway_test.py +++ b/tst/integration/env/gateway/gateway_test.py @@ -3358,7 +3358,8 @@ async def _rate_check(client, gw, rate, sleep_s=5.0): cl = await client.post(f"{gw}/clock/clear", timeout=30) assert cl.status_code == 200 and cl.json()["armed"] is False, cl.text assert (await _read(client, gw)).status_code == 404, "cleared clock should 404" - assert (await client.get(f"{gw}/clock/state", timeout=30)).json() == {"armed": False}, "state after clear" + cleared = (await client.get(f"{gw}/clock/state", timeout=30)).json() + assert (cleared["armed"], cleared["followers"], "t0" in cleared) == (False, {}, False), f"state after clear: {cleared}" # Clear round-trips the tool surface byte-for-byte back to the unarmed baseline. cleared = await _get_time_surfaces(client, gw, result.mcp_url) diff --git a/tst/unit/env/envs/test_multi_env_computers.py b/tst/unit/env/envs/test_multi_env_computers.py index b165bc025..135732f28 100644 --- a/tst/unit/env/envs/test_multi_env_computers.py +++ b/tst/unit/env/envs/test_multi_env_computers.py @@ -311,6 +311,14 @@ async def test_deploy_hands_the_gateway_each_computer_deployed_under_its_name_an ("gw", "modal_vm"), ("gw", "modal_vm"), ("sb-desktop", "modal_vm"), ("sb-laptop", "modal_vm")] +def test_a_computer_whose_card_follows_a_clock_hands_the_gateway_its_control_token_as_its_read_token(): + from agent_env.env.envs._multi_computers import _clock_read_token + + following = dataclasses.replace(_record("desktop"), environment_card={"name": "desktop", "capabilities": {"extensions": [ + {"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": {"method": "POST", "endpoint": "/clock/sync"}}}}]}}) + assert (_clock_read_token(following), _clock_read_token(_record("laptop"))) == ("token-desktop", None) + + @pytest.mark.asyncio async def test_a_multi_env_without_computers_gives_the_gateway_no_children(gateway, providers): await _multi(mcp=["slack"]).deploy() diff --git a/tst/unit/env/gateway/clock_advance_test.py b/tst/unit/env/gateway/clock_advance_test.py index 0a76031f9..5f1eaab3e 100644 --- a/tst/unit/env/gateway/clock_advance_test.py +++ b/tst/unit/env/gateway/clock_advance_test.py @@ -120,7 +120,8 @@ def test_the_card_offers_advance(): clock_ext = next(e for e in GATEWAY_EXTENSIONS if e["uri"] == EXT_CLOCK_URI) assert clock_ext["params"]["methods"]["advance"] == { "method": "PUT", "endpoint": "/clock/advance", - "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}}, "required": ["virtual_time"]}} + "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "wait_followers_s": { + "type": "number", "minimum": 0, "maximum": 30.0}}, "required": ["virtual_time"]}} @pytest.fixture diff --git a/tst/unit/env/gateway/clock_version_test.py b/tst/unit/env/gateway/clock_version_test.py new file mode 100644 index 000000000..bb470850d --- /dev/null +++ b/tst/unit/env/gateway/clock_version_test.py @@ -0,0 +1,142 @@ +"""The clock's opaque version, which every write changes, and set_rate, which changes its rate the way advance moves it: +from its time at the write, keeping t0 and generation.""" +from __future__ import annotations + +from datetime import datetime, timedelta + +import pytest + +from agent_env.env.gateway.clock import Clock, ClockError +from agent_env.env.gateway.triggers import TriggerEngine +from tst.unit.env.gateway.triggers_time_test import FakeNow, _FakeGateway, _perm, _settle, _tick_call + +T0 = "2026-01-01T00:00:00Z" + + +def _utc(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +def test_every_write_changes_the_version_and_reads_leave_it(): + clock = Clock(now_fn=FakeNow()) + seen = [clock.version] + clock.set_time(T0, rate=1.0) + seen.append(clock.version) + clock.read(), clock.state(), clock.now() + assert clock.version == seen[-1] + clock.advance("2026-01-02T00:00:00Z") + seen.append(clock.version) + clock.set_rate(60) + seen.append(clock.version) + clock.clear() + seen.append(clock.version) + clock.clear() + seen.append(clock.version) + assert len(set(seen)) == len(seen) + assert [clock.count_of(version) for version in seen] == [0, 1, 2, 3, 4, 5] + + +def test_a_version_never_repeats_across_clocks_and_names_only_what_its_clock_had(): + first, second = Clock(now_fn=FakeNow()), Clock(now_fn=FakeNow()) + assert first.version != second.version + first.set_time(T0) + boot = first.version.split(".")[0] + assert first.count_of(first.version) == 1 + assert first.count_of(second.version) is None + assert [first.count_of(v) for v in (f"{boot}.2", f"{boot}.-1", f"{boot}.²", f"{boot}", "", None, 3)] == [None] * 7 + + +def test_a_rate_change_continues_from_the_clock_as_of_the_write_keeping_its_t0_and_generation(): + """Taking the new time from a state read a moment earlier stepped the universe back by the gap, which a computer, + never stepping back, then ran ahead of; and a re-arm moved every time trigger's relative marks to the new t0.""" + now = FakeNow() + clock = Clock(now_fn=now) + clock.set_time(T0, rate=1.0) + generation = clock.generation + now.tick(10) + read = clock.state() + now.tick(0.07) + state = clock.set_rate(60) + assert _utc(state["virtual_time"]) - _utc(read["virtual_time"]) == timedelta(milliseconds=70) + assert (state["virtual_seconds_per_real_second"], state["t0"], clock.generation) == (60, T0, generation) + now.tick(1) + assert clock.read()["virtual_time"] == "2026-01-01T00:01:10.070000Z" + assert clock.elapsed() == pytest.approx(70.07) + assert clock.at_elapsed(130.07) == _utc("2026-01-01T00:02:10.070000Z") + clock.advance("2026-01-01T01:00:00Z") + now.tick(1) + assert clock.read()["virtual_time"] == "2026-01-01T01:01:00Z" + + +def test_a_rate_change_hands_followers_a_clock_continuous_with_the_one_before(): + """A follower extrapolating the snapshot before the change to the moment of the write is where the one after starts.""" + now = FakeNow() + clock = Clock(now_fn=now) + clock.set_time(T0, rate=1.0) + before, at = clock.state(), now() + now.tick(3.25) + after = clock.set_rate(2) + extrapolated = _utc(before["virtual_time"]) + timedelta(seconds=(now() - at) * before["virtual_seconds_per_real_second"]) + assert abs((_utc(after["virtual_time"]) - extrapolated).total_seconds()) < 0.001 + + +class SteppingNow: + """A real clock's reading moves between any two calls: each call is 100 µs after the one before it.""" + + def __init__(self): + self.t, self.calls = 1000.0, [] + + def __call__(self) -> float: + self.t += 0.0001 + self.calls.append(self.t) + return self.t + + +@pytest.mark.parametrize("before,after", [(1.0, 86400.0), (86400.0, 1.0)]) +def test_a_rate_change_starts_its_new_rate_at_the_instant_it_read_the_time(before, after): + now = SteppingNow() + clock = Clock(now_fn=now) + mark = len(now.calls) + clock.set_time(T0, rate=before) + armed_at = now.calls[mark] + mark = len(now.calls) + clock.set_rate(after) + changed_at = now.calls[mark] + read = clock.now() + expected = _utc(T0) + timedelta(seconds=(changed_at - armed_at) * before + (now.calls[-1] - changed_at) * after) + assert abs((read - expected).total_seconds()) < 0.001 + + +def test_a_rate_change_needs_an_armed_clock_and_a_rate(): + with pytest.raises(ClockError, match="not armed"): + Clock(now_fn=FakeNow()).set_rate(2) + clock = Clock(now_fn=FakeNow()) + clock.set_time(T0) + for rate in (None, -1, "2"): + with pytest.raises(ClockError, match="rate must be"): + clock.set_rate(rate) + + +@pytest.mark.asyncio +async def test_a_relative_time_trigger_keeps_its_mark_across_a_rate_change(): + now = FakeNow() + clock = Clock(now_fn=now) + engine = TriggerEngine(_FakeGateway(clock)) + clock.set_time(T0, rate=1) + engine.register({"watch_roles": ["default"], "triggers": [ + {"id": "rel", "when": {"type": "time", "at": "PT2H"}, "actions": [_perm()]}]}) + _tick_call(engine, now, 3600) + clock.set_rate(3600) + assert engine.clock_advanced() == [] + _tick_call(engine, now, 1.5) + await _settle(engine) + trig = engine.state()["triggers"][0] + assert (trig["status"], trig["fire_count"]) == ("fired", 1) + assert "reanchored" not in {e["kind"] for e in engine.state()["events"]} + + +def test_a_count_too_long_to_be_one_the_clock_had_names_nothing(): + clock = Clock(now_fn=FakeNow()) + clock.set_time(T0) + boot = clock.version.split(".")[0] + assert [clock.count_of(f"{boot}.{'9' * 5000}"), clock.count_of(f"{boot}.01"), clock.count_of(f"{boot}.1")] == [None, None, 1] diff --git a/tst/unit/env/gateway/clock_watch_test.py b/tst/unit/env/gateway/clock_watch_test.py new file mode 100644 index 000000000..16e8a8532 --- /dev/null +++ b/tst/unit/env/gateway/clock_watch_test.py @@ -0,0 +1,354 @@ +"""The /clock/watch long poll computers follow the gateway's clock through, and clock writes that wait for them.""" +from __future__ import annotations + +import asyncio +import time +from datetime import datetime + +import httpx +import pytest + +from agent_env.env.gateway import gateway as gw_mod +from agent_env.env.gateway.clock import Clock +from agent_env.env.gateway.clock_followers import ClockFollowers +from agent_env.env.gateway.constants import (CLOCK_MAX_WAIT_FOLLOWERS_S, CLOCK_WATCH_HOLD_S, EXT_CLOCK_URI, + GATEWAY_EXTENSIONS) +from agent_env.env.gateway.entrypoint import parse_clock_followers +from agent_env.env.gateway.gateway import Gateway + +T0 = "2026-01-01T00:00:00Z" +TOKENS = {"desktop": "token-desktop", "laptop": "token-laptop"} +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 + + +def _utc(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +@pytest.fixture +def gateway(tmp_path, monkeypatch): + monkeypatch.setattr(gw_mod, "GATEWAY_TRAJECTORY_FILE", str(tmp_path / "trajectory.jsonl")) + monkeypatch.setattr(ClockFollowers, "WATCH_S", HELD_S) + made = [] + + def make(followers=TOKENS) -> Gateway: + made.append(Gateway(host="127.0.0.1", port=0, server_name="unit", internal_mcp_servers=[], + clock_followers=followers)) + return made[-1] + + yield make + for gw in made: + gw._close_trajectory_file() + + +def _client(gw: Gateway) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.ASGITransport(app=gw._asgi_app()), base_url="http://gateway") + + +async def _watch(client, after: str, follower: str | None = None, token: str | None = None) -> httpx.Response: + params = {"after": after, **({"follower": follower} if follower else {})} + headers = {"Authorization": f"Bearer {token}"} if token else {} + return await client.get("/clock/watch", params=params, headers=headers) + + +async def _follow(client, name: str, token: str, *, applies_after_s: float = 0.0): + """A computer's follower: each answer shows a version, applied after ``applies_after_s``, and the next watch names it.""" + version = "" + while True: + answer = (await _watch(client, version, name, token)).json() + if answer["version"] != version: + await asyncio.sleep(applies_after_s) + version = answer["version"] + + +async def _until_shown(gw: Gateway, *names: str) -> None: + """Until each of ``names`` has shown the clock's current version: its follower is up and holding a watch.""" + for _ in range(500): + state = gw._clock_followers.state() + if all(state[name]["version"] == gw._clock.version for name in names): + return + await asyncio.sleep(0.01) + raise AssertionError(f"{names} never showed {gw._clock.version}: {gw._clock_followers.state()}") + + +def test_the_card_offers_every_clock_method_with_the_waits_bound_and_the_watchs_hold(): + methods = next(e for e in GATEWAY_EXTENSIONS if e["uri"] == EXT_CLOCK_URI)["params"]["methods"] + wait = {"type": "number", "minimum": 0, "maximum": CLOCK_MAX_WAIT_FOLLOWERS_S} + assert {name: (m["method"], m["endpoint"]) for name, m in methods.items()} == { + "get_time": ("GET", "/clock/time"), "set_time": ("PUT", "/clock/set-time"), "set_rate": ("PUT", "/clock/set-rate"), + "advance": ("PUT", "/clock/advance"), "clear": ("POST", "/clock/clear"), "state": ("GET", "/clock/state"), + "watch": ("GET", "/clock/watch")} + assert all(methods[name]["request"]["properties"]["wait_followers_s"] == wait + for name in ("set_time", "set_rate", "advance", "clear")) + assert methods["set_rate"]["request"]["required"] == ["virtual_seconds_per_real_second"] + assert methods["set_time"]["request"]["required"] == ["virtual_time"] + assert set(methods["watch"]["request"]["properties"]) == {"after", "follower"} + assert methods["watch"]["hold_s"] == CLOCK_WATCH_HOLD_S == ClockFollowers.WATCH_S == 25.0 + + +@pytest.mark.asyncio +async def test_a_watch_naming_another_version_answers_at_once_with_the_clock(gateway): + gw = gateway() + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0, "virtual_seconds_per_real_second": 0}) + started = time.monotonic() + answer = await _watch(client, "some-earlier-version") + body = answer.json() + assert answer.status_code == 200 and time.monotonic() - started < HELD_S / 2 + assert body["version"] == gw._clock.version + assert (body["armed"], body["t0"], body["virtual_time"], body["virtual_seconds_per_real_second"]) == (True, T0, T0, 0) + assert body["received"] <= body["sent"] <= time.time() + + +@pytest.mark.asyncio +async def test_a_watch_naming_the_current_version_answers_at_the_next_change(gateway): + gw = gateway() + async with _client(gw) as client: + held = asyncio.create_task(_watch(client, gw._clock.version)) + await asyncio.sleep(0.1) + assert not held.done() + await client.put("/clock/set-time", json={"virtual_time": T0}) + answer = (await held).json() + assert answer["armed"] and answer["version"] == gw._clock.version + assert answer["received"] < answer["sent"] - 0.09 + + +@pytest.mark.asyncio +async def test_a_watch_with_no_change_answers_after_its_hold_with_the_same_version(gateway, monkeypatch): + monkeypatch.setattr(ClockFollowers, "WATCH_S", SHORT_HOLD_S) + gw = gateway() + async with _client(gw) as client: + started = time.monotonic() + answer = (await _watch(client, gw._clock.version)).json() + assert time.monotonic() - started >= SHORT_HOLD_S + assert answer == {**answer, "armed": False, "version": gw._clock.version} + + +@pytest.mark.asyncio +async def test_a_write_waits_until_every_declared_computer_shows_its_version(gateway): + gw = gateway() + async with _client(gw) as client: + followers = [asyncio.create_task(_follow(client, name, token, applies_after_s=0.05)) for name, token in TOKENS.items()] + try: + await _until_shown(gw, *TOKENS) + started = time.monotonic() + written = await client.put("/clock/set-time", json={"virtual_time": T0, "wait_followers_s": 2}) + waited = time.monotonic() - started + rated = await client.put("/clock/set-rate", json={"virtual_seconds_per_real_second": 60, "wait_followers_s": 2}) + advanced = await client.put("/clock/advance", json={"virtual_time": "2026-01-02T00:00:00Z", "wait_followers_s": 2}) + cleared = await client.post("/clock/clear", json={"wait_followers_s": 2}) + state = (await client.get("/clock/state")).json() + finally: + for follower in followers: + follower.cancel() + assert [r.status_code for r in (written, rated, advanced, cleared)] == [200] * 4 + assert 0.05 <= waited < 2 + assert {name: f["version"] for name, f in state["followers"].items()} == {"desktop": gw._clock.version, + "laptop": gw._clock.version} + + +@pytest.mark.asyncio +async def test_a_write_answers_504_naming_the_computers_still_behind_and_the_change_stands(gateway): + gw = gateway() + async with _client(gw) as client: + follower = asyncio.create_task(_follow(client, "desktop", TOKENS["desktop"])) + try: + await _until_shown(gw, "desktop") + started = time.monotonic() + answer = await client.put("/clock/set-time", json={"virtual_time": T0, "virtual_seconds_per_real_second": 0, + "wait_followers_s": 0.4}) + waited = time.monotonic() - started + read = (await client.get("/clock/time")).json() + finally: + follower.cancel() + 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 read == {"virtual_time": T0} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("token", [None, "token-desktop", "wrong"], ids=["none", "another-computers", "unknown"]) +async def test_a_watch_acknowledges_only_with_its_computers_token(gateway, monkeypatch, token): + monkeypatch.setattr(ClockFollowers, "WATCH_S", SHORT_HOLD_S) + gw = gateway({"laptop": TOKENS["laptop"]}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + answered = await _watch(client, gw._clock.version, "laptop", token) + state = (await client.get("/clock/state")).json() + assert answered.status_code == 200 + assert state["followers"] == {"laptop": {"version": None, "seen_s_ago": None}} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("claim", ["later", "another-clocks"]) +async def test_an_acknowledgement_cant_claim_a_version_the_clock_hasnt_had(gateway, claim): + gw = gateway({"laptop": TOKENS["laptop"]}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + boot, count = gw._clock.version.split(".") + claimed = f"{boot}.{int(count) + 5}" if claim == "later" else Clock().version + await _watch(client, claimed, "laptop", TOKENS["laptop"]) + answer = await client.put("/clock/advance", json={"virtual_time": "2026-02-01T00:00:00Z", "wait_followers_s": 0.1}) + assert answer.status_code == 504 and answer.json()["stale"] == ["laptop"] + + +@pytest.mark.asyncio +async def test_a_late_watch_naming_an_earlier_version_doesnt_move_its_computer_back(gateway): + gw = gateway({"laptop": TOKENS["laptop"]}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + earlier = gw._clock.version + await client.put("/clock/advance", json={"virtual_time": "2026-01-02T00:00:00Z"}) + latest = gw._clock.version + held = asyncio.create_task(_watch(client, latest, "laptop", TOKENS["laptop"])) + await _until_shown(gw, "laptop") + await _watch(client, earlier, "laptop", TOKENS["laptop"]) + long = await _watch(client, f"{latest.split('.')[0]}.{'9' * 5000}", "laptop", TOKENS["laptop"]) + state = (await client.get("/clock/state")).json() + waited = await client.put("/clock/set-time", json={"virtual_time": T0, "wait_followers_s": 0.5}) + held.cancel() + assert state["followers"]["laptop"]["version"] == latest + assert long.status_code == 200 and long.json()["version"] == latest + assert waited.status_code == 504 + + +@pytest.mark.asyncio +async def test_a_write_without_a_wait_or_without_computers_answers_at_once(gateway): + with_computers, without = gateway(), gateway({}) + for gw, body in ((with_computers, {"virtual_time": T0}), (without, {"virtual_time": T0, "wait_followers_s": 5})): + async with _client(gw) as client: + started = time.monotonic() + answer = await client.put("/clock/set-time", json=body) + assert answer.status_code == 200 and time.monotonic() - started < 2.5 + assert "stale" not in answer.json() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wait", [-1, "2", True, 31, 10**309, None], + ids=["negative", "string", "bool", "over-30", "past-a-floats-range", "null"]) +async def test_a_wait_is_a_number_of_seconds_up_to_30(gateway, wait): + gw = gateway({}) + async with _client(gw) as client: + answers = [await client.put("/clock/set-time", json={"virtual_time": T0, "wait_followers_s": wait}), + await client.put("/clock/set-rate", json={"virtual_seconds_per_real_second": 2, "wait_followers_s": wait}), + await client.put("/clock/advance", json={"virtual_time": "2026-02-01T00:00:00Z", "wait_followers_s": wait}), + await client.post("/clock/clear", json={"wait_followers_s": wait})] + if wait is None: + assert [a.status_code for a in answers] == [200] * 4 + else: + assert [a.status_code for a in answers] == [400] * 4 + assert all("wait_followers_s" in a.json()["error"] for a in answers) + assert gw._clock.count_of(gw._clock.version) == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("path,method", [("/clock/set-time", "PUT"), ("/clock/set-rate", "PUT"), + ("/clock/advance", "PUT"), ("/clock/clear", "POST")]) +@pytest.mark.parametrize("body,error", [("{not json", "Invalid JSON body"), ("[]", "must be a JSON object"), + ("null", "must be a JSON object")]) +async def test_every_clock_write_refuses_a_body_that_isnt_a_json_object(gateway, path, method, body, error): + gw = gateway({}) + async with _client(gw) as client: + answer = await client.request(method, path, content=body) + assert answer.status_code == 400 and error in answer.json()["error"] + assert gw._clock.count_of(gw._clock.version) == 0 + + +@pytest.mark.asyncio +async def test_clear_still_takes_no_body(gateway): + gw = gateway({}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + answer = await client.post("/clock/clear") + assert answer.status_code == 200 and answer.json() == {"ok": True, "armed": False, "version": gw._clock.version} + + +@pytest.mark.asyncio +async def test_a_restarted_gateway_answers_a_watch_from_before_it_at_once_and_unarmed(gateway): + before = gateway() + async with _client(before) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + version = before._clock.version + restarted = gateway() + async with _client(restarted) as client: + started = time.monotonic() + answer = (await _watch(client, version, "desktop", TOKENS["desktop"])).json() + state = (await client.get("/clock/state")).json() + assert time.monotonic() - started < HELD_S / 2 + assert answer["armed"] is False and answer["version"] != version + assert state["followers"]["desktop"]["version"] is None + + +@pytest.mark.asyncio +async def test_set_rate_changes_the_rate_at_the_time_of_the_write_and_keeps_t0(gateway): + gw = gateway({}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0}) + read = (await client.get("/clock/state")).json() + await asyncio.sleep(0.07) + changed = await client.put("/clock/set-rate", json={"virtual_seconds_per_real_second": 2}) + body = changed.json() + gap = (_utc(body["virtual_time"]) - _utc(read["virtual_time"])).total_seconds() + assert changed.status_code == 200 and (body["virtual_seconds_per_real_second"], body["t0"], body["due"]) == (2, T0, []) + assert gap >= 0.07 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("path,body,error", [ + ("/clock/set-time", {}, "virtual_time must be an RFC3339 string"), + ("/clock/set-time", {"virtual_seconds_per_real_second": 2}, "virtual_time must be an RFC3339 string"), + ("/clock/set-rate", {}, "rate must be a number"), + ("/clock/set-rate", {"virtual_time": T0}, "rate must be a number"), + ("/clock/set-rate", {"virtual_seconds_per_real_second": 10**309}, "exceeds max"), + ("/clock/set-time", {"virtual_time": T0, "virtual_seconds_per_real_second": 10**309}, "exceeds max"), +]) +async def test_set_time_needs_a_time_and_set_rate_a_rate_it_can_run_at(gateway, path, body, error): + gw = gateway({}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0, "virtual_seconds_per_real_second": 0}) + answer = await client.put(path, json=body) + state = (await client.get("/clock/state")).json() + assert answer.status_code == 400 and error in answer.json()["error"] + assert (state["virtual_seconds_per_real_second"], state["virtual_time"]) == (0, T0) + + +@pytest.mark.asyncio +async def test_set_rate_refuses_an_unarmed_clock(gateway): + gw = gateway({}) + async with _client(gw) as client: + answer = await client.put("/clock/set-rate", json={"virtual_seconds_per_real_second": 2}) + assert answer.status_code == 400 and "not armed" in answer.json()["error"] + + +@pytest.mark.asyncio +async def test_a_hundred_advances_are_each_acknowledged_in_order(gateway): + gw = gateway({"desktop": TOKENS["desktop"]}) + async with _client(gw) as client: + await client.put("/clock/set-time", json={"virtual_time": T0, "virtual_seconds_per_real_second": 0}) + follower = asyncio.create_task(_follow(client, "desktop", TOKENS["desktop"])) + try: + await _until_shown(gw, "desktop") + answers, waits = [], [] + for hour in range(1, 101): + started = time.monotonic() + answers.append(await client.put("/clock/advance", json={ + "virtual_time": f"2026-01-{1 + hour // 24:02d}T{hour % 24:02d}:00:00Z", "wait_followers_s": 2})) + waits.append(time.monotonic() - started) + finally: + follower.cancel() + assert [a.status_code for a in answers] == [200] * 100 + counts = [gw._clock.count_of(a.json()["version"]) for a in answers] + assert counts == sorted(counts) and len(set(counts)) == 100 + assert max(waits) < 2 + + +def test_the_gateway_reads_its_declared_computers_from_its_environment(): + assert parse_clock_followers("") == {} + assert parse_clock_followers(" desktop=read-d ,laptop=read-l,") == {"desktop": "read-d", "laptop": "read-l"} + for spec, fragment in (("desktop", "expected 'name=read token'"), ("=tok", "expected"), ("desktop=", "expected"), + ("desktop=a,desktop=b", "twice")): + with pytest.raises(ValueError, match=fragment): + parse_clock_followers(spec) diff --git a/tst/unit/providers/env_providers/env_gateway_provider_test.py b/tst/unit/providers/env_providers/env_gateway_provider_test.py index 0de98b99d..a7014ee02 100644 --- a/tst/unit/providers/env_providers/env_gateway_provider_test.py +++ b/tst/unit/providers/env_providers/env_gateway_provider_test.py @@ -2857,6 +2857,28 @@ def test_compose_fronts_each_external_child_as_an_mcp_server_and_a_rest_route_by env = dict(line.split("=", 1) for line in compose["services"][GATEWAY_SERVICE_NAME]["environment"]) assert env["INTERNAL_MCP_SERVERS"].split(",") == ["slack", "desktop=https://desk.example/mcp"] assert "desktop=https://desk.example" in env["REST_PROXY_URLS"].split(",") + assert "CLOCK_FOLLOWER_TOKENS" not in env + + +def test_compose_declares_the_children_that_follow_the_clock_by_their_read_tokens_and_logs_none_of_them(): + from agent_env.env.envs.service_db import ServiceDBConfig + from agent_env.env.gateway import ExternalChild + + rendered = EnvironmentGatewayProvider().create_docker_compose( + mcp_servers=[], gateway_image="agent-gateway", + state_provider=LocalPostgresStateProvider(service_db_config=ServiceDBConfig()), + state_instance=LocalPostgresStateProvider.default_instance(), + external_children=(ExternalChild("desktop", "https://desk.example/mcp", "https://desk.example", "read-desk"), + ExternalChild("phone", "https://phone.example/mcp", "https://phone.example"), + ExternalChild("laptop", "https://lap.example/mcp", "https://lap.example", "read-lap"))) + + services = yaml.safe_load(rendered)["services"] + holders = {name: [item for item in service.get("environment", []) if item.startswith("CLOCK_FOLLOWER_TOKENS=")] + for name, service in services.items()} + assert {name: items for name, items in holders.items() if items} == { + GATEWAY_SERVICE_NAME: ["CLOCK_FOLLOWER_TOKENS=desktop=read-desk,laptop=read-lap"]} + redacted = _redact_compose_secrets(rendered) + assert "read-desk" not in redacted and " - CLOCK_FOLLOWER_TOKENS=***" in redacted def test_no_service_depends_on_the_gateway_so_a_vm_deploy_can_start_them_all_before_it(): @@ -2890,7 +2912,7 @@ async def test_a_vm_deploy_starts_its_children_beside_its_services_and_the_gatew async def children(reach): order.append(("children", reach)) - return [ExternalChild("desktop", "http://localhost:20001/mcp", "http://localhost:20001")] + return [ExternalChild("desktop", "http://localhost:20001/mcp", "http://localhost:20001", "read-desk")] def compose(**kwargs): composes.append(kwargs.get("external_children", ())) @@ -2917,7 +2939,8 @@ async def up(sandbox, args=""): assert sorted(order[:2]) == [("children", "https://gw.example"), ("up", f"--scale {GATEWAY_SERVICE_NAME}=0")] assert order[2:] == [("up", "")] - assert composes == [(), (ExternalChild("desktop", "http://host.docker.internal:20001/mcp", "http://host.docker.internal:20001"),)] + assert composes == [(), (ExternalChild("desktop", "http://host.docker.internal:20001/mcp", "http://host.docker.internal:20001", + "read-desk"),)] written = [shlex.split(call.args[0]) for call in sandbox.exec_script.await_args_list] assert ["printf", "%s\\n", "compose 2", ">", f"{GATEWAY_APP_DIR}/docker-compose.yml"] in written @@ -2979,7 +3002,7 @@ async def fake_create_container(**kwargs): async def children(reach): reaches.append(reach) - return [ExternalChild("desktop", "https://desk.example/mcp", "https://desk.example")] + return [ExternalChild("desktop", "https://desk.example/mcp", "https://desk.example", "read-desk")] gateway_env = MagicMock(docker_image_artifact=_image("agent-gateway")) db_env = MagicMock() @@ -2999,3 +3022,4 @@ async def children(reach): assert reaches == [None] assert env["INTERNAL_MCP_SERVERS"].split(",")[-1] == "desktop=https://desk.example/mcp" assert env["REST_PROXY_URLS"].split(",")[-1] == "desktop=https://desk.example" + assert env["CLOCK_FOLLOWER_TOKENS"] == "desktop=read-desk"