diff --git a/AGENTS.md b/AGENTS.md index 8d99e82cf..c99fc72ca 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._change_clock`, which wakes the triggers when the change kept the clock's generation, answers the held watches and waits for the followers. Its followers, the computers on it (`CLOCK_FOLLOWER_TOKENS`, each child whose card serves `clock/v1` `sync_time`, by its read token, its control token for now), long-poll `/clock/watch` naming the version they show, each watch held up to the card's `hold_s` (`env/gateway/clock_followers.py`), and a write given `wait_followers_s` waits for each, answering 504 naming those behind; a restarted gateway is unarmed. While its clock is armed it serves agents `get_time` (`env/gateway/get_time.py`), which answers in UTC and, given an IANA `time_zone`, also in that zone; executors' working `hours` (`env/gateway/triggers.py`) take a fixed `utc_offset` or an IANA `time_zone`. | | `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Every artifact but an environment universe, whose `metadata` key holds its legacy metadata file ids, carries a `metadata` dict outside its versioned fields, changed by compare-and-swap (`merge_metadata`). Built-ins: `CliArtifact` (`cli`), `ComputerArtifact` (`computer`, a base computer: its architectures, OS, session user, settings and container form, whose image `put` reads from its registry into `metadata.validated_images`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`: its environment's files by path, stored as the one wrapped `file_artifact_ref` every environment artifact had when it has one file at its filename, else as `file_artifact_refs`; `get_file_artifacts()` reads either, and a load hands each file to the env's `data/add` with its path, which a pre-v1 load, taking one file, refuses), `EnvironmentUniverseArtifact` (`environment_universe`), `SkillArtifact` (`skill`); `[artifacts] type_aliases` maps another stored spelling of a type onto one of them. A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or a published image's tag with its `image_digest` and its build context, which sandboxes pull by the digest and build from the context when they can't (the infra images, and a bundle agent built from a shipped agent's sources, `Plan.published`), or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/docker/context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. | | `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, among them `start_computer_recording` and `end_computer_recording`, and more). On an env whose card serves computer-exec/v1 (a computer), `run_code` with `env_id` and `run_container_unit_tests_verifier` with `env_id` run inside it as its session user through the driver: files staged with `data/add` and read with `data/get`, commands run with `run`; `env_outcome_verifier` hands a `verify` that takes an `env_url` keyword the env's address. A computer's framework-only routes (everything but MCP and the card) take the per-deploy control token its provider boots it with (`DeployedEnv.control_token`, `AGENTENV_CONTROL_TOKEN`); framework calls send `deployed.control_headers`, `Task.run` runs every step under `protocol_v1.authorized` with each deployed env's (a `verify` included), and a `run_code` script inside the computer gets `input["env"]["headers"]`. `teardown_run` and the `teardown_sandboxes` step share one core (`task/teardown.py` `tear_down`): the recordings the run left open end with fresh grants, each env still up closes (`Env.close`; a computer env ends any recording still open), what stayed open is registered from its namespace, then every sandbox is terminated by id. `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). | | `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..78dd344b8 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's a clock follower (``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..33a49e3df 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 child that is a clock follower 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..2066a6353 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,33 @@ 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 self.version_of(self._count) + + def version_of(self, count: int) -> str: + """The version this clock had at its ``count``th write.""" + return f"{self._boot}.{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 ("", "", "") + if boot != self._boot or not (count.isascii() and count.isdigit()): + return None + # Counts carry no leading zeros, so one with more digits than the current count is past it, and is refused before + # int() reads it: int() of thousands of digits raises. + if len(count) > len(str(self._count)) or int(count) > self._count: + return None + return int(count) + @property def rate(self) -> float: """Virtual seconds per real second; 1.0 while unarmed.""" @@ -168,6 +194,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 +229,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 +241,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 +273,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..0217488aa --- /dev/null +++ b/src/agent_env/env/gateway/clock_followers.py @@ -0,0 +1,93 @@ +"""The followers of the gateway's clock (the computers on it): 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 + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ClockError(f"wait_followers_s must be a non-negative number, got {value!r}") + # Only a float can be NaN or infinite; math.isfinite on an int past a float's range raises. + if (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. + Its event and condition bind to the loop that first waits on them, so one gateway's watches and waiting writes all + run on the loop that serves it, never one asyncio.run apiece.""" + + 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, int | 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): + self._shown[follower] = count + 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 < 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, clock = time.monotonic(), self._clock() + return {name: {"version": None if shown is None else clock.version_of(shown), + "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..0a8e48ab6 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 followers that don't show it in time (set_rate keeps t0, as advance does); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.", "params": {"endpoint": "/clock/time", "methods": { "get_time": {"method": "GET", "endpoint": "/clock/time"}, - "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}}, "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..bbf0296a9 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 + followers of 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,41 @@ 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) + + @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"))) @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"))) @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, 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 +362,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 +610,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, *, 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.""" + 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, stale = await self._change_clock(lambda: change(body), wait) + 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, 200) + + async def _change_clock(self, change, wait_s: float = 0.0) -> tuple[dict, list[str]]: + """Every change of the clock goes through here. ``change`` applied, then: the time triggers re-evaluated when it + moved time on the current arm (it kept the clock's generation, as an advance or a rate change does; a re-arm or + clear re-anchors them by its new one), every held watch answered, and, given ``wait_s``, a wait for each declared + follower to show the new version. The state ``change`` left, with the triggers it made ``due``, and the + followers still behind.""" + generation = self._clock.generation + state = change() + if self._clock.generation == generation: + state = {**state, "due": self._trigger_engine.clock_advanced()} + self._clock_followers.changed() + return state, await self._clock_followers.stale_after(state["version"], wait_s) if wait_s else [] + 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..4c7a01b7c 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 is one, 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/computer/app/artifacts/check_notes/run.py b/tst/integration/computer/app/artifacts/check_notes/run.py index 50e145ae3..2f2048b99 100644 --- a/tst/integration/computer/app/artifacts/check_notes/run.py +++ b/tst/integration/computer/app/artifacts/check_notes/run.py @@ -1,7 +1,7 @@ """What the notes app's service wrote, which it can only once its settings reached it, the name the computer's card gives and what its readiness says of the app, asked on loopback from inside the computer, the launcher its server added to the session user's menu, the apps launched since the computer booted, and what its install left: its run script's -output of the apt packages it installed, its PPA's among them, its download, and its install step's file.""" +output of the apt packages it installed, its outside source's among them, its download, and its install step's file.""" import json import time import urllib.request @@ -27,6 +27,6 @@ def run(input): "launchers": sorted(p.name for p in (Path.home() / ".local/share/applications").glob("agentenv-app-*")), "launched": sorted(p.name for p in Path("/run/agentenv/apps").iterdir()), "hello": (PREFIX / "hello.txt").read_text().strip(), - "fastfetch": (PREFIX / "fastfetch.txt").read_text().splitlines(), + "compose": (PREFIX / "compose.txt").read_text().splitlines(), "license": (PREFIX / "LICENSE.txt").read_text().startswith("A. HISTORY OF THE SOFTWARE"), "installed": (Path.home() / ".notes-installed").read_text()} diff --git a/tst/integration/computer/app/envs/notes/app.py b/tst/integration/computer/app/envs/notes/app.py index 67ab9e55a..38d9a5862 100644 --- a/tst/integration/computer/app/envs/notes/app.py +++ b/tst/integration/computer/app/envs/notes/app.py @@ -1,7 +1,7 @@ """A notes app for the computer app tests: its service writes the settings a deploy gave it to ~/.notes.json, and its own server launches its window once the desktop session is up; it's ready once both have happened. It requires no Python of its own, so its service and window run in the venv every app gets for its server. Its install takes an apt -package from Ubuntu's archive and one only a Launchpad PPA has, which its run script uses, a download checked by sha256, +package from Ubuntu's archive and one only Docker's apt repository has, which its run script uses, a download checked by sha256, and an install step run as the session user on the build setting its env.toml fixes, which also leaves a welcome note in ~/Notes, where it keeps its data, so its server serves that folder as its data plane. On Linux its settings gain a font, and its install step and ready check are Linux's.""" @@ -13,7 +13,9 @@ Service, computer_app, install_step, linux, ready_check) LICENSE = "https://raw.githubusercontent.com/python/cpython/v3.12.0/LICENSE" -FASTFETCH = "https://ppa.launchpadcontent.net/zhangsongcui3371/fastfetch/ubuntu" +# A source outside Ubuntu, with its own key: Docker's apt repository (download.docker.com, key 9DC8 5822 9FC7 DD38 854A +# E2D8 8D81 803C 0EBF CD88), whose compose plugin Ubuntu doesn't ship and which needs no other package. +DOCKER = "https://download.docker.com/linux/ubuntu" class LinuxSettings(BaseModel): @@ -21,8 +23,8 @@ class LinuxSettings(BaseModel): @computer_app(supports=[{"os": {"type": "linux", "display": ["x11"]}, "arch": ["amd64", "arm64"]}]) -@linux(packages=Packages(apt=["hello", "fastfetch"]), - sources=[AptSource(apt=FASTFETCH, suite="noble", components=["main"], key="fastfetch.asc")], +@linux(packages=Packages(apt=["hello", "docker-compose-plugin"]), + sources=[AptSource(apt=DOCKER, suite="noble", components=["stable"], key="docker.asc")], downloads=[Download(url=LICENSE, to="LICENSE.txt", sha256="3b2f81fe21d181c499c59a256c8e1968455d6689d269aa85373bfb6af41da3bf")], run=[RunScript(script="setup.sh")], diff --git a/tst/integration/computer/app/envs/notes/docker.asc b/tst/integration/computer/app/envs/notes/docker.asc new file mode 100644 index 000000000..ee7872e5d --- /dev/null +++ b/tst/integration/computer/app/envs/notes/docker.asc @@ -0,0 +1,62 @@ +-----BEGIN PGP PUBLIC KEY BLOCK----- + +mQINBFit2ioBEADhWpZ8/wvZ6hUTiXOwQHXMAlaFHcPH9hAtr4F1y2+OYdbtMuth +lqqwp028AqyY+PRfVMtSYMbjuQuu5byyKR01BbqYhuS3jtqQmljZ/bJvXqnmiVXh +38UuLa+z077PxyxQhu5BbqntTPQMfiyqEiU+BKbq2WmANUKQf+1AmZY/IruOXbnq +L4C1+gJ8vfmXQt99npCaxEjaNRVYfOS8QcixNzHUYnb6emjlANyEVlZzeqo7XKl7 +UrwV5inawTSzWNvtjEjj4nJL8NsLwscpLPQUhTQ+7BbQXAwAmeHCUTQIvvWXqw0N +cmhh4HgeQscQHYgOJjjDVfoY5MucvglbIgCqfzAHW9jxmRL4qbMZj+b1XoePEtht +ku4bIQN1X5P07fNWzlgaRL5Z4POXDDZTlIQ/El58j9kp4bnWRCJW0lya+f8ocodo +vZZ+Doi+fy4D5ZGrL4XEcIQP/Lv5uFyf+kQtl/94VFYVJOleAv8W92KdgDkhTcTD +G7c0tIkVEKNUq48b3aQ64NOZQW7fVjfoKwEZdOqPE72Pa45jrZzvUFxSpdiNk2tZ +XYukHjlxxEgBdC/J3cMMNRE1F4NCA3ApfV1Y7/hTeOnmDuDYwr9/obA8t016Yljj +q5rdkywPf4JF8mXUW5eCN1vAFHxeg9ZWemhBtQmGxXnw9M+z6hWwc6ahmwARAQAB +tCtEb2NrZXIgUmVsZWFzZSAoQ0UgZGViKSA8ZG9ja2VyQGRvY2tlci5jb20+iQI3 +BBMBCgAhBQJYrefAAhsvBQsJCAcDBRUKCQgLBRYCAwEAAh4BAheAAAoJEI2BgDwO +v82IsskP/iQZo68flDQmNvn8X5XTd6RRaUH33kXYXquT6NkHJciS7E2gTJmqvMqd +tI4mNYHCSEYxI5qrcYV5YqX9P6+Ko+vozo4nseUQLPH/ATQ4qL0Zok+1jkag3Lgk +jonyUf9bwtWxFp05HC3GMHPhhcUSexCxQLQvnFWXD2sWLKivHp2fT8QbRGeZ+d3m +6fqcd5Fu7pxsqm0EUDK5NL+nPIgYhN+auTrhgzhK1CShfGccM/wfRlei9Utz6p9P +XRKIlWnXtT4qNGZNTN0tR+NLG/6Bqd8OYBaFAUcue/w1VW6JQ2VGYZHnZu9S8LMc +FYBa5Ig9PxwGQOgq6RDKDbV+PqTQT5EFMeR1mrjckk4DQJjbxeMZbiNMG5kGECA8 +g383P3elhn03WGbEEa4MNc3Z4+7c236QI3xWJfNPdUbXRaAwhy/6rTSFbzwKB0Jm +ebwzQfwjQY6f55MiI/RqDCyuPj3r3jyVRkK86pQKBAJwFHyqj9KaKXMZjfVnowLh +9svIGfNbGHpucATqREvUHuQbNnqkCx8VVhtYkhDb9fEP2xBu5VvHbR+3nfVhMut5 +G34Ct5RS7Jt6LIfFdtcn8CaSas/l1HbiGeRgc70X/9aYx/V/CEJv0lIe8gP6uDoW +FPIZ7d6vH+Vro6xuWEGiuMaiznap2KhZmpkgfupyFmplh0s6knymuQINBFit2ioB +EADneL9S9m4vhU3blaRjVUUyJ7b/qTjcSylvCH5XUE6R2k+ckEZjfAMZPLpO+/tF +M2JIJMD4SifKuS3xck9KtZGCufGmcwiLQRzeHF7vJUKrLD5RTkNi23ydvWZgPjtx +Q+DTT1Zcn7BrQFY6FgnRoUVIxwtdw1bMY/89rsFgS5wwuMESd3Q2RYgb7EOFOpnu +w6da7WakWf4IhnF5nsNYGDVaIHzpiqCl+uTbf1epCjrOlIzkZ3Z3Yk5CM/TiFzPk +z2lLz89cpD8U+NtCsfagWWfjd2U3jDapgH+7nQnCEWpROtzaKHG6lA3pXdix5zG8 +eRc6/0IbUSWvfjKxLLPfNeCS2pCL3IeEI5nothEEYdQH6szpLog79xB9dVnJyKJb +VfxXnseoYqVrRz2VVbUI5Blwm6B40E3eGVfUQWiux54DspyVMMk41Mx7QJ3iynIa +1N4ZAqVMAEruyXTRTxc9XW0tYhDMA/1GYvz0EmFpm8LzTHA6sFVtPm/ZlNCX6P1X +zJwrv7DSQKD6GGlBQUX+OeEJ8tTkkf8QTJSPUdh8P8YxDFS5EOGAvhhpMBYD42kQ +pqXjEC+XcycTvGI7impgv9PDY1RCC1zkBjKPa120rNhv/hkVk/YhuGoajoHyy4h7 +ZQopdcMtpN2dgmhEegny9JCSwxfQmQ0zK0g7m6SHiKMwjwARAQABiQQ+BBgBCAAJ +BQJYrdoqAhsCAikJEI2BgDwOv82IwV0gBBkBCAAGBQJYrdoqAAoJEH6gqcPyc/zY +1WAP/2wJ+R0gE6qsce3rjaIz58PJmc8goKrir5hnElWhPgbq7cYIsW5qiFyLhkdp +YcMmhD9mRiPpQn6Ya2w3e3B8zfIVKipbMBnke/ytZ9M7qHmDCcjoiSmwEXN3wKYI +mD9VHONsl/CG1rU9Isw1jtB5g1YxuBA7M/m36XN6x2u+NtNMDB9P56yc4gfsZVES +KA9v+yY2/l45L8d/WUkUi0YXomn6hyBGI7JrBLq0CX37GEYP6O9rrKipfz73XfO7 +JIGzOKZlljb/D9RX/g7nRbCn+3EtH7xnk+TK/50euEKw8SMUg147sJTcpQmv6UzZ +cM4JgL0HbHVCojV4C/plELwMddALOFeYQzTif6sMRPf+3DSj8frbInjChC3yOLy0 +6br92KFom17EIj2CAcoeq7UPhi2oouYBwPxh5ytdehJkoo+sN7RIWua6P2WSmon5 +U888cSylXC0+ADFdgLX9K2zrDVYUG1vo8CX0vzxFBaHwN6Px26fhIT1/hYUHQR1z +VfNDcyQmXqkOnZvvoMfz/Q0s9BhFJ/zU6AgQbIZE/hm1spsfgvtsD1frZfygXJ9f +irP+MSAI80xHSf91qSRZOj4Pl3ZJNbq4yYxv0b1pkMqeGdjdCYhLU+LZ4wbQmpCk +SVe2prlLureigXtmZfkqevRz7FrIZiu9ky8wnCAPwC7/zmS18rgP/17bOtL4/iIz +QhxAAoAMWVrGyJivSkjhSGx1uCojsWfsTAm11P7jsruIL61ZzMUVE2aM3Pmj5G+W +9AcZ58Em+1WsVnAXdUR//bMmhyr8wL/G1YO1V3JEJTRdxsSxdYa4deGBBY/Adpsw +24jxhOJR+lsJpqIUeb999+R8euDhRHG9eFO7DRu6weatUJ6suupoDTRWtr/4yGqe +dKxV3qQhNLSnaAzqW/1nA3iUB4k7kCaKZxhdhDbClf9P37qaRW467BLCVO/coL3y +Vm50dwdrNtKpMBh3ZpbB1uJvgi9mXtyBOMJ3v8RZeDzFiG8HdCtg9RvIt/AIFoHR +H3S+U79NT6i0KPzLImDfs8T7RlpyuMc4Ufs8ggyg9v3Ae6cN3eQyxcK3w0cbBwsh +/nQNfsA6uu+9H7NhbehBMhYnpNZyrHzCmzyXkauwRAqoCbGCNykTRwsur9gS41TQ +M8ssD1jFheOJf3hODnkKU+HKjvMROl1DK7zdmLdNzA1cvtZH/nCC9KPj1z8QC47S +xx+dTZSx4ONAhwbS/LN3PoKtn8LPjY9NP9uDWI+TWYquS2U+KHDrBDlsgozDbs/O +jCxcpDzNmXpWQHEtHU7649OXHP7UeNST1mCUCH5qdank0V1iejF6/CfTFU4MfcrG +YT90qFF93M3v01BbxP+EIY2/9tiIPbrd +=0YYh +-----END PGP PUBLIC KEY BLOCK----- diff --git a/tst/integration/computer/app/envs/notes/fastfetch.asc b/tst/integration/computer/app/envs/notes/fastfetch.asc deleted file mode 100644 index c54a3728a..000000000 --- a/tst/integration/computer/app/envs/notes/fastfetch.asc +++ /dev/null @@ -1,30 +0,0 @@ ------BEGIN PGP PUBLIC KEY BLOCK----- -Comment: Hostname: -Version: Hockeypuck 2.2 - -xsFNBGXC33sBEADqChRifM9aJb4sJnn9jKV4mdt9Rj90OAYfySPuVyVX+dDKWP8X -ATjtDDNJeROme4+05iSt1SAey9W0ugYo8ft2LeMCjWlog6mlM3EPJ4UelEVBOGpF -su/cgTZbu0A6bHIliFeQhDavd897Dr8ZH9+QIPumNyHB+mtd+U/S9qggZtWs6seu -WJo2eRYXC4ubJ52rjpCAT2bSM2Rn7YATAKxYl35+ef4ytb5i+wkgUVD55AuYzNL7 -eMzSOY4iYF9SWUrYxtNKq2AH6IZ6G49Vjt0yDcH8wl2wImvKcI4hwQYbxHXwBj2l -E7msT65gHyFocRrTp1XvrOGms6iqE3GjXy7CwiWKtxK/DLFx8Pf2A27E+95OKsrj -Cs9g+g5H5625QSp7xgocaNtmHz+gTEGkstDIiAZQ2/tp0jN4sPpFKEp03oMxr9+8 -iON68SD0lPRRfPBwQat0CkkUK1YInnw1ZhPQJJk8XW8mxPYsbQ+hNofUo0gN1dX3 -T/1RNi8x2zDiNP5lyoSzX0V2G7kIvL+G0xUnhMauNXqQcLXfKukDL9Vf4q/TWmAS -wQhJ2oCeRrdlOWhBHOoFNUSpt2PxRHqBd5SMCh29HIdN2CLT3XitPj10ozyrXVGZ -FLKL1uZqZar03ofmw/qNYXG0smG12LsjCjWDtX94Cg+/Fl1dVYfC3ziPzQARAQAB -zRtMYXVuY2hwYWQgUFBBIGZvciBDYXJ0ZXIgTGnCwY4EEwEKADgWIQTrZe4Z2ALz -6xoTz+R+Lly01IZfIQUCZcLfewIbAwULCQgHAgYVCgkICwIEFgIDAQIeAQIXgAAK -CRB+Lly01IZfIbRvD/0Zx9paqlDJWwCTd+1b9kZPOWeylBvQ5A8rvJb7RyccX6AR -gXj1zWkc76OV1A3TiwVqsFhX357jG7TJ2HkAcld7EwhbKjne3nmtCLUwSFdJERP2 -dpobBAtmZV7g4dEHlHH3nFRcowrsAFwlHTp72Iak39WVjxR/XQDX9yeTg8H0hJZ9 -veh8slwcPw0ORevqXnOrKYM3uolLoBEUXe2Y4Zc/gBqN6OnK6uv0ISfG/zG3yOJt -Z1HJ6ITamqHv8QDjg2rbxYt2VtvuOe/8T34nI6otdtJUDNDllLNn/5Td1YQsdgUg -Rgf9bMM7X8LTJW/2zc++I1XCJToVFuXdD/IcIB+e10Y8yrhBjer3R4wtoW7ubXG9 -SmURCLfu9JxmDjl5T+Fsg4YDlSIih02P3SitwPWG0AvqYWDsKGfx5tF5/28wjHoM -bsi1xxAxnlYuzcZT/Xb6ZO4wMcV5QiRbfVHAACt69vyABXfFlqJ8ScSt8pGsR9bl -dCRlrLFQh13NDnSJQtnO3mx4eF1ir1iO4hjJFTDhhE8QbImS3w1hEhZuHGVXFo8a -xsChEDlop6GNFFUUnOwS3X+sKVQjL9Emiv1DkXTSnJmDwKQyOr6T3uTgzT5o9Gt8 -SrnkCBlxrqZHzaUJuSGV0YxxWUAexlj4nlYwoZz5uXTNikhD38q+O9ZNG4SQTg== -=yPNc ------END PGP PUBLIC KEY BLOCK----- diff --git a/tst/integration/computer/app/envs/notes/setup.sh b/tst/integration/computer/app/envs/notes/setup.sh index 075b50881..04425db90 100644 --- a/tst/integration/computer/app/envs/notes/setup.sh +++ b/tst/integration/computer/app/envs/notes/setup.sh @@ -1,6 +1,7 @@ #!/bin/sh -# Run as root once the app's packages are in: hello is one of them, and fastfetch the one only its PPA has. +# Run as root once the app's packages are in: hello is one of them, and docker-compose-plugin the one only Docker's apt +# repository has. set -e hello > hello.txt -dpkg-query -W -f '${Version}\n' fastfetch > fastfetch.txt -fastfetch --version >> fastfetch.txt +dpkg-query -W -f '${Version}\n' docker-compose-plugin > compose.txt +/usr/libexec/docker/cli-plugins/docker-compose version >> compose.txt diff --git a/tst/integration/computer/computer_app_test.py b/tst/integration/computer/computer_app_test.py index 60c597d21..9995d539a 100644 --- a/tst/integration/computer/computer_app_test.py +++ b/tst/integration/computer/computer_app_test.py @@ -58,8 +58,8 @@ def test_an_app_deployed_alone_runs_its_service_on_its_settings_on_a_computer_na (deployed,) = context["deployed_envs"] assert deployed["env_id"].endswith("/notes") and deployed["environment_card"]["name"] == "notes-driver" service = context["metadata"]["script_results"]["service"] - version, fastfetch = service.pop("fastfetch") - assert version.endswith("~noble") and fastfetch.startswith("fastfetch ") + version, compose = service.pop("compose") + assert version.endswith("~noble") and compose.startswith("Docker Compose version v") assert service == { "signature": "-- Ada", "token": "t0ken", "edition": "pro", "font": "DejaVu Sans", "card": "notes-driver", "app_ready": [{"name": "app notes", "ok": True, "detail": "ready"}], @@ -152,7 +152,7 @@ def test_the_locks_debs_and_wheels_install_from_mounts_and_stay_in_no_layer_of_t made = docker("history", "--no-trunc", "--format", "{{.CreatedBy}}", image).stdout.decode().splitlines() installed = docker("run", "--rm", "--entrypoint", "sh", image, "-c", - "dpkg-query -W -f '${Package} ${Status}\\n' hello fastfetch && " + "dpkg-query -W -f '${Package} ${Status}\\n' hello docker-compose-plugin && " "/opt/agentenv/apps/clock/venv/bin/python -c 'import websockets; print(\"websockets\")' && " "if [ -e /tmp/agentenv-packages ]; then echo left; else echo clean; fi" ).stdout.decode().splitlines() @@ -160,7 +160,7 @@ def test_the_locks_debs_and_wheels_install_from_mounts_and_stay_in_no_layer_of_t assert [line for line in made if "packages/" in line and line.startswith("COPY")] == [] assert any(line.startswith("RUN /bin/sh -c apt-get install -y --no-install-recommends /tmp/agentenv-packages/apt/") for line in made) - assert sorted(installed[:2]) == ["fastfetch install ok installed", "hello install ok installed"] + assert sorted(installed[:2]) == ["docker-compose-plugin install ok installed", "hello install ok installed"] assert installed[2:] == ["websockets", "clean"] 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..303c1cd8a --- /dev/null +++ b/tst/unit/env/gateway/clock_watch_test.py @@ -0,0 +1,365 @@ +"""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) + + +@pytest.mark.asyncio +async def test_a_change_on_the_arm_wakes_the_triggers_and_a_re_arm_leaves_them_to_its_generation(gateway): + gw = gateway({}) + async with _client(gw) as client: + armed = (await client.put("/clock/set-time", json={"virtual_time": T0})).json() + rated = (await client.put("/clock/set-rate", json={"virtual_seconds_per_real_second": 2})).json() + 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] 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"