diff --git a/.agentenv/config.example.toml b/.agentenv/config.example.toml index c2e04609b..deb3ec35b 100644 --- a/.agentenv/config.example.toml +++ b/.agentenv/config.example.toml @@ -39,6 +39,13 @@ agent_default = "local" # agents too # [sandbox.providers.boat.config] # api_key = "secret:boat_api_key" # or "env:BOAT_API_KEY" +# Optional CoreWeave Sandboxes (install the 'coreweave' extra): +# [sandbox.providers.coreweave.config] +# api_key = "secret:cwsandbox_api_key" # or wandb_api_key, with optional wandb_entity and wandb_project +# image = "public.ecr.aws/docker/library/docker:dind" # the VM's Docker host; a prepared host image avoids installs at boot +# docker_storage = "loop" # or "memory": Docker's storage in RAM, faster and smaller +# runner_ids = ["your-cks-runner"] # with placement_mode = "cks" + # [agents] # default_a2a_agent_id = "my-agent" # what a deploy_agent step that names no agent deploys; by default the # # agents bundle's Claude Code, @local/agentenv-framework/agents/claude-code diff --git a/AGENTS.md b/AGENTS.md index 5b813d6b6..2cd413c47 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -65,9 +65,9 @@ or one of them only where it is used. store, an OCI registry at `localhost:5000`, env-var secret store, `LocalRunner`, `local` Docker sandboxes for envs and agents. MongoDB, DynamoDB, Firestore, S3, Cloud Storage, ECR, AWS Secrets Manager, Google Cloud Secret Manager, and the Modal, E2B, Sail, SmolVM, Tensorlake, Tenki, Vercel, boxd, - Daytona and boat sandboxes exist as implementations and are selected by config (the last eight through their - extras: `sail`, `smol`, `tensorlake`, `tenki`, `vercel`, `boxd`, `daytona`, `boat`; every extra but `smol` is in - `dev`). + CoreWeave, Daytona and boat sandboxes exist as implementations and are selected by config (the last nine + through their extras: `sail`, `smol`, `tensorlake`, `tenki`, `vercel`, `boxd`, `coreweave`, `daytona`, `boat`; + every extra but `smol` is in `dev`). - The model endpoint is unset until `[model] base_url` / `api_key` (or `LITELLM_BASE_URL` / `LITELLM_API_KEY`) is configured. A loopback URL handed into a `local` container (the model endpoint, an agent, peer or gateway on this machine) becomes `host.docker.internal`. On Linux that @@ -118,7 +118,7 @@ and deserialized through a registry. | `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/`, `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. `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` namespace: while the CLI or a `build_app` server runs, an `@local/…` id and what an `@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/`. | | `config/` | The `Config` singleton (`get_config`, `configure`, `reset_config`) in `config/runtime.py`, file discovery in `config/loader.py`, and `load_impl`, which resolves `module:Class` pointers. `agent_env.store` re-exports the config names for compatibility. | -| `providers/` | `providers/sandbox_providers/` holds the sandbox providers `local`, `modal`, `modal_vm`, `e2b`, `sail_vm` (the `sail` extra), `smol_vm` (the `smol` extra), `tensorlake` (the `tensorlake` extra), `tenki` (the `tenki` extra), `vercel` (the `vercel` extra), `boxd` (the `boxd` extra), `daytona` (the `daytona` extra), `boat` (the `boat` extra); `[sandbox] default` and `agent_default` accept a comma-separated fallback chain, whose deploys skip a provider that can't run them. A sandbox provider declares what it runs (`ON_THIS_MACHINE`, and for VMs on this machine the address their requests to it arrive at, `REACHES_HOST_AT`; `CONTAINER_ACCEPTS`, `PRIVATE_NETWORK`; `SUPPORTED_SANDBOX_MODES`, derived from whether it implements `create_sandbox` and `create_vm`; and `SANDBOX_ACCEPTS`, the images an agent's sandbox runs, which defaults to those a VM loads when it implements `create_vm` and to `CONTAINER_ACCEPTS` otherwise, so only a provider that implements `create_vm` yet gives an agent a container, like the local one, declares it), each sandbox carries the `platform` of the images it runs (linux/amd64 unless its provider sets another when it creates it; None, any, unchecked, for a local sandbox), and code outside `providers/sandbox_providers/` reads those, never which provider it has. `providers/env_providers/` holds the environment providers: `EnvironmentProvider` (an env's containers and state store), `EnvironmentGatewayProvider` (`gateway`, an env's default), which renders a docker-compose for the gateway and its MCP servers inside the sandbox, and `EnvironmentServerProvider` (`server`), one MCP server in its own container with no gateway in front and no state store; `providers/env_state/` holds env-state providers (`local_postgres` built in). | +| `providers/` | `providers/sandbox_providers/` holds the sandbox providers `local`, `modal`, `modal_vm`, `e2b`, `sail_vm` (the `sail` extra), `smol_vm` (the `smol` extra), `tensorlake` (the `tensorlake` extra), `tenki` (the `tenki` extra), `vercel` (the `vercel` extra), `boxd` (the `boxd` extra), `coreweave` (the `coreweave` extra), `daytona` (the `daytona` extra), `boat` (the `boat` extra); `[sandbox] default` and `agent_default` accept a comma-separated fallback chain, whose deploys skip a provider that can't run them. A sandbox provider declares what it runs (`ON_THIS_MACHINE`, and for VMs on this machine the address their requests to it arrive at, `REACHES_HOST_AT`; `CONTAINER_ACCEPTS`, `PRIVATE_NETWORK`; `SUPPORTED_SANDBOX_MODES`, derived from whether it implements `create_sandbox` and `create_vm`; and `SANDBOX_ACCEPTS`, the images an agent's sandbox runs, which defaults to those a VM loads when it implements `create_vm` and to `CONTAINER_ACCEPTS` otherwise, so only a provider that implements `create_vm` yet gives an agent a container, like the local one, declares it), each sandbox carries the `platform` of the images it runs (linux/amd64 unless its provider sets another when it creates it; None, any, unchecked, for a local sandbox), and code outside `providers/sandbox_providers/` reads those, never which provider it has. `providers/env_providers/` holds the environment providers: `EnvironmentProvider` (an env's containers and state store), `EnvironmentGatewayProvider` (`gateway`, an env's default), which renders a docker-compose for the gateway and its MCP servers inside the sandbox, and `EnvironmentServerProvider` (`server`), one MCP server in its own container with no gateway in front and no state store; `providers/env_state/` holds env-state providers (`local_postgres` built in). | | `a2a_agent/` | The `A2AAgent` entity (`a2a_agent`), its stores and the validator steps. The protocol package provides the agent-side framework. | | `openenv/` | `build_app(bundle)`: the [OpenEnv](https://github.com/openenv/OpenEnv) server for a bundle's tasks, each session playing one episode through the Episode API, with the envs' MCP tools, an added `end_turn` tool and the task's verifier score as the reward. It prepares bundles as `agent-env run` does (`prepare_bundle(played=True)`, without the agent the client plays) and keeps namespace routing on for the app's life. `openenv` comes with the `openenv` extra, a floor with no cap; this package and `training/` are the only ones that import it. Each task's description in the Task API names its `toolset` (the played agent's envs, role, and whether it gets `bash`). | | `training/` | One module per outside trainer, each behind its own extra; `_openenv.py`, the session on a server `openenv/` makes that both play an episode through (`Session`: a background start, tool calls, the turns left open ended until scored), with the Task API reads, the per-toolset probe and an in-process `Server`; and `_tools.py`, which turns an env's tools into typed, documented Python methods from their JSON Schema, as transformers' `get_json_schema` reads them. `trl.py` (`trl` extra): `AgentEnv(url)` over a server `openenv/` makes, with `dataset` (a row per task) and `factory` (TRL's `environment_factory`: a generated class per toolset whose methods are the tools, a background `reset` and an owned `get_reward`); `AgentEnv.local(bundle)` serves the bundle in process. `nemo_gym/` (`nemo-gym` extra): `python -m agent_env.training.nemo_gym SOURCE ROOT` writes a bundle's NeMo Gym resources server into a Gym root, `NEMO_GYM_EXTRA_ROOTS`'s or a Gym checkout's (`app.py`; a `pyproject.toml`, since uv applies the nearest one's `[tool.uv]` and a checkout's `exclude-dependencies` would drop `cryptography`; a config with `simple_agent` and the `legacy_agent` environment server, and `data/.jsonl` rows with each task's tools); `server.AgentEnvResourcesServer` is that server, run in the venv Gym builds: a row's seed starts its episode, a catch-all route calls its tools, and `verify` ends the turns left open with the model's last reply and returns the reward, masking a rollout the environment failed; with `reward_components` it also returns each score by step id, their sum the reward, for NeMo RL's GDPO. | diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 25fab8219..34f918d06 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -14,7 +14,7 @@ marks, or contributor names to endorse or promote Scale AI, or related products. ## Python dependencies -The runtime dependencies of `agentenv-framework` and its `aws`, `gcp`, `sail`, `tensorlake`, `tenki`, `boxd`, `daytona`, `boat`, `openenv` and `trl` extras, at the versions resolved in `uv.lock`. Development-only dependencies are not listed. +The runtime dependencies of `agentenv-framework` and its `aws`, `gcp`, `sail`, `tensorlake`, `tenki`, `boxd`, `coreweave`, `daytona`, `boat`, `openenv` and `trl` extras, at the versions resolved in `uv.lock`. Development-only dependencies are not listed. ### a2a-sdk 0.3.26 @@ -257,6 +257,14 @@ The runtime dependencies of `agentenv-framework` and its `aws`, `gcp`, `sail`, ` - Source: - [License text 14](#license-text-14), [License text 15](#license-text-15), [License text 16](#license-text-16) +### cwsandbox 1.19.6 + +- License: Apache-2.0 +- Copyright: Copyright 2025 CoreWeave, Inc. +- Author: CoreWeave, Inc. +- Source: +- [License text 1](#license-text-1), [License text 12](#license-text-12) + ### cyclopts 5.2.0 - License: Apache-2.0 diff --git a/pyproject.toml b/pyproject.toml index 95c443553..803bfe245 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,8 +110,12 @@ boat = [ "boat-sdk~=1.13.0", # sandboxes, labels, hosted ports, detached commands and their output stream validated against 1.13 "urllib3>=2.1.0,<3", # the SDK's transport, whose errors are what a dropped connection raises ] +# The CoreWeave Sandbox provider, selected as the built-in `coreweave` backend. +coreweave = [ + "cwsandbox>=1.19.6,<2", # Reject exec streams that close without an exit status. +] dev = [ - "agentenv-framework[aws,boat,boxd,daytona,gcp,openenv,sail,tensorlake,tenki,trl,vercel]", # exercise portable optional providers and stores in CI + "agentenv-framework[aws,boat,boxd,coreweave,daytona,gcp,openenv,sail,tensorlake,tenki,trl,vercel]", # exercise portable optional providers and stores in CI "fastmcp>=3.4,<4", # the standalone-fastmcp mount path of the protocol; <4 keeps mcp 1.x "Pillow>=10.4", # the protocol's computer extra, whose tests the unit tier runs "av>=17", # the protocol's computer extra, whose tests the unit tier runs and which the recording tests decode with diff --git a/src/agent_env/cli/a2a_agent/deploy.py b/src/agent_env/cli/a2a_agent/deploy.py index d764f4dff..17a8f89df 100644 --- a/src/agent_env/cli/a2a_agent/deploy.py +++ b/src/agent_env/cli/a2a_agent/deploy.py @@ -18,7 +18,7 @@ @click.option("--ttl-seconds", type=click.IntRange(min=MIN_TTL_SECONDS, max=MAX_TTL_SECONDS), default=DEFAULT_TTL_SECONDS, help=f"VM lifetime in seconds (default {DEFAULT_TTL_SECONDS})") @click.option("--sandbox", default=None, - help="Sandbox backend(s): a built-in (modal, modal_vm, e2b, tensorlake, tenki, boxd, daytona, boat, local) " + help="Sandbox backend(s): a built-in (modal, modal_vm, e2b, tensorlake, tenki, boxd, coreweave, daytona, boat, local) " "or a name from [sandbox.providers] in .agentenv/config.toml; comma-separated for a " "fallback chain. Defaults to [sandbox].agent_default (else local) when omitted.") def deploy(agent_id: str, agent_version: int | None, env_var_pairs: tuple[str, ...], ttl_seconds: int, sandbox: str): diff --git a/src/agent_env/cli/env/deploy.py b/src/agent_env/cli/env/deploy.py index ee4f75bc5..fe1984c28 100644 --- a/src/agent_env/cli/env/deploy.py +++ b/src/agent_env/cli/env/deploy.py @@ -19,7 +19,7 @@ @click.option("--gateway-mode", type=click.Choice([m.value for m in GatewayMode], case_sensitive=False), default=GatewayMode.PERFORMANCE.value, help="Gateway mode (performance or consistent)") @click.option("--sandbox", default=None, - help="Sandbox backend(s): a built-in (modal, modal_vm, e2b, tensorlake, tenki, boxd, daytona, boat, local) " + help="Sandbox backend(s): a built-in (modal, modal_vm, e2b, tensorlake, tenki, boxd, coreweave, daytona, boat, local) " "or a name from [sandbox.providers] in .agentenv/config.toml; comma-separated for a " "fallback chain. Defaults to [sandbox].default (else local) when omitted.") @click.option("--service-db", "service_db_env_id", default=None, diff --git a/src/agent_env/providers/sandbox_providers/coreweave/__init__.py b/src/agent_env/providers/sandbox_providers/coreweave/__init__.py new file mode 100644 index 000000000..3a9f654a5 --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/coreweave/__init__.py @@ -0,0 +1 @@ +"""CoreWeave Sandbox provider.""" diff --git a/src/agent_env/providers/sandbox_providers/coreweave/host_image.Dockerfile b/src/agent_env/providers/sandbox_providers/coreweave/host_image.Dockerfile new file mode 100644 index 000000000..f4857ec82 --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/coreweave/host_image.Dockerfile @@ -0,0 +1,6 @@ +# A prepared Docker host image for CoreWeave sandboxes: docker:dind with the tools the provider's host setup otherwise +# installs at boot, so a sandbox whose network policy blocks the Alpine package mirror still starts. Push it to a +# registry the sandbox can pull from and set `image` in [sandbox.providers.coreweave.config]. +FROM public.ecr.aws/docker/library/docker:dind + +RUN apk add --no-cache bash curl e2fsprogs diff --git a/src/agent_env/providers/sandbox_providers/coreweave/provider.py b/src/agent_env/providers/sandbox_providers/coreweave/provider.py new file mode 100644 index 000000000..df13857a1 --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/coreweave/provider.py @@ -0,0 +1,508 @@ +"""CoreWeave Sandbox provider. + +Each sandbox is one CoreWeave Sandbox. A container (``create_container``, ``create_sandbox``) runs the workload's image +itself. A VM (``create_vm``) is a privileged container in a Kata VM, booted from a Docker host image, where the gateway's +compose, and the containers steps start, run under Docker through the inherited ``VmSandbox`` helpers. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import math +import re +import uuid +from collections.abc import Awaitable, Callable +from typing import Any, ClassVar, Self +from urllib.parse import urlparse + +import cwsandbox +from cwsandbox import ( + AuthHeaders, + Container, + Endpoint, + EndpointAuth, + EndpointKind, + ResourceOptions, + SandboxStatus, + SecurityContext, + Service, + ServiceVisibility, +) + +from agent_env.artifact.artifacts.docker_image import DockerImageArtifact +from agent_env.attribution import Attribution +from agent_env.config import get_config +from agent_env.config.errors import ConfigError +from agent_env.providers.sandbox_providers.coreweave.sandbox import ( + DOCKER_STORAGE_LOOP, + DOCKER_STORAGE_MEMORY, + Client, + CoreWeaveContainerSandbox, + CoreWeaveSandbox, + host_setup_script, + network_options, + policy_from_egress, + runs_docker_host, +) +from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy, NetworkPolicyUnsupportedError +from agent_env.providers.sandbox_providers.sandbox_provider import ( + Accepts, + SandboxNotFoundError, + SandboxProvider, + apply_default_attribution, +) + +logger = logging.getLogger(__name__) + +DEFAULT_IMAGE = "public.ecr.aws/docker/library/docker:dind" +DEFAULT_RUNTIME_CLASS = "kata-qemu" +SANDBOX_TAG = "agentenv" +_SECTION = "[sandbox.providers.coreweave.config]" +_STRING_KEYS = ("api_key", "wandb_api_key", "wandb_entity", "wandb_project", "base_url", "image", "runtime_class", + "placement_mode") +_CONFIG_KEYS = frozenset({*_STRING_KEYS, "docker_storage", "runner_ids"}) + +# Docker keeps both the compressed and the extracted layers, so 10 GB runs out once a gateway and a few MCP server +# images are on the sandbox. A loop-mounted disk is a sparse file, so only what Docker writes is used. +MIN_DOCKER_DISK_GB = 20 +# The platform closes an HTTPS request to an exposed port after this many seconds, its maximum; the default, 15, is +# shorter than an agent's turn. +ENDPOINT_REQUEST_TIMEOUT_SEC = 900 +# A sandbox reconnected to has no lifetime agent-env can read, so its commands get this budget. +_RECONNECT_COMMAND_TIMEOUT_SEC = 2 * 3600 +_MAX_TAG_LENGTH = 59 +_REAP_ATTEMPTS = 3 +# How long a failed or cancelled create waits for its start request to answer, so it learns the sandbox's id, before it +# looks for the sandbox by its tag instead. +_START_SETTLE_SEC = 30 +# The Sandbox API keeps a deleted sandbox's record, so one draining or ended is as gone as one it no longer knows. +_GONE_STATUSES = frozenset({SandboxStatus.TERMINATING, SandboxStatus.COMPLETED, SandboxStatus.FAILED, + SandboxStatus.TERMINATED}) + +_MAX_ENDPOINT_PARENTS = 16 + + +def resources(cpu: float, memory_mb: int) -> ResourceOptions: + """Requests and equal limits, so the sandbox gets exactly ``cpu`` (in millicores, rounded up) and ``memory_mb``.""" + if isinstance(cpu, bool) or not isinstance(cpu, (int, float)) or not math.isfinite(cpu) or cpu <= 0: + raise ValueError("cpu must be a finite number greater than zero") + if isinstance(memory_mb, bool) or not isinstance(memory_mb, int) or memory_mb <= 0: + raise ValueError("memory must be a positive integer number of MB") + shape = {"cpu": f"{math.ceil(cpu * 1000)}m", "memory": f"{memory_mb}Mi"} + return ResourceOptions(requests=shape, limits=dict(shape)) + + +def docker_disk_gb(disk_size_gb: float) -> int: + if (isinstance(disk_size_gb, bool) or not isinstance(disk_size_gb, (int, float)) + or not math.isfinite(disk_size_gb) or disk_size_gb <= 0): + raise ValueError("disk_size_gb must be a finite number greater than zero") + return max(MIN_DOCKER_DISK_GB, math.ceil(disk_size_gb)) + + +def _tag(text: str) -> str: + """``text`` as a CoreWeave tag: letters, digits, ``.``, ``_`` and ``-``, at most 59, ending with a letter or digit.""" + return re.sub(r"[^A-Za-z0-9._-]+", "-", text)[:_MAX_TAG_LENGTH].rstrip("._-") + + +def sandbox_tags(attribution: Attribution, create_tag: str) -> list[str]: + """``agentenv``, the create's unique tag, and one ``key.value`` tag per attribution entry.""" + tags = [SANDBOX_TAG, create_tag] + tags += [tag for key in sorted(attribution) if (tag := _tag(f"{key}.{attribution[key]}"))] + return list(dict.fromkeys(tags)) + + +def services(ports: list[int]) -> list[Service]: + endpoint = Endpoint(kind=EndpointKind.HTTPS, auth=EndpointAuth.OPEN, + request_timeout_seconds=ENDPOINT_REQUEST_TIMEOUT_SEC) + return [Service(port=port, visibility=ServiceVisibility.PUBLIC, endpoint=endpoint) for port in ports] + + +def tunnel_urls(raw: Any) -> dict[int, str]: + urls = {endpoint.port: endpoint.url for endpoint in raw.service_endpoints if endpoint.url} + for url in urls.values(): + if parent := (urlparse(url).hostname or "").partition(".")[2]: + _remember_egress_host(f"*.{parent}") + return urls + + +def auth_headers(config: dict[str, Any]) -> AuthHeaders: + """The headers the Sandbox API authenticates: a CoreWeave API key, or a W&B API key with an optional entity and + project. Built here rather than by the SDK, which would read the process environment or W&B's settings files.""" + if config.get("api_key"): + return AuthHeaders(headers={"Authorization": f"Bearer {config['api_key']}"}, strategy="api_key") + headers = {"x-wandb-api-key": config["wandb_api_key"]} + if config.get("wandb_entity"): + headers["x-entity-id"] = config["wandb_entity"] + if config.get("wandb_project"): + headers["x-project-name"] = config["wandb_project"] + return AuthHeaders(headers=headers, strategy="wandb_api_key") + + +async def _finish(cleanup: Any) -> None: + """Run ``cleanup`` to its end in a task of its own though this one is cancelled meanwhile, and then honor the + cancellation. ``asyncio.run`` cancels every task left when it ends and closes the loop once they are done, so + cleanup left to a later callback would not run before the loop closed.""" + task = asyncio.ensure_future(cleanup) + cancelled = False + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + cancelled = True + if cancelled: + raise asyncio.CancelledError + + +async def _reap(client: Client, sandbox_id: str) -> None: + """Delete ``sandbox_id``, retrying.""" + for attempt in range(_REAP_ATTEMPTS): + try: + await client.delete(sandbox_id) + logger.info("Deleted CoreWeave sandbox %s after its create failed or was cancelled", sandbox_id) + return + except Exception as exc: # noqa: BLE001 - retried, then logged + logger.warning("Deleting CoreWeave sandbox %s failed (attempt %s): %s", sandbox_id, attempt + 1, exc) + if attempt + 1 < _REAP_ATTEMPTS: + await asyncio.sleep(2**attempt) + logger.error("CoreWeave sandbox %s from a failed create may still be running; it stops at its max lifetime", + sandbox_id) + + +async def _reclaim(client: Client, raw: Any, create_tag: str, starting: asyncio.Future) -> None: + """Delete the sandbox a failed or cancelled create made, or may have made with its response lost: the one with the + id its start request answered, or else the one tagged with the create's unique tag.""" + if not starting.done(): + await asyncio.wait({starting}, timeout=_START_SETTLE_SEC) + if not starting.done(): + logger.warning("A cancelled CoreWeave create's start request is still unanswered after %ss; looking for its " + "sandbox by its tag %s", _START_SETTLE_SEC, create_tag) + starting.add_done_callback(lambda done: _reap_late_start(client, raw, create_tag, done)) + elif not starting.cancelled(): + starting.exception() + await _reap_created(client, raw, create_tag) + + +class CoreWeaveSandboxProvider(SandboxProvider): + """Provision CoreWeave Sandboxes: containers that run a workload's image, and Docker-capable VMs. + + Credentials come from resolved provider config, never from the process environment: ``api_key`` (a CoreWeave + Sandbox API key) or ``wandb_api_key``, each an ``env:`` or ``secret:`` reference in + ``[sandbox.providers.coreweave.config]``. Workloads never see them. ``image``, ``docker_storage`` and + ``runtime_class`` configure the VM's Docker host only. + """ + + # The platform pulls a container's image by its name; only a VM loads a tarball or builds a context. + CONTAINER_ACCEPTS = Accepts.NAME + SANDBOX_ACCEPTS = Accepts.NAME + # An exposed port's URL is -.., and a CoreWeave wildcard covers exactly one + # label, so no one entry covers every runner. Each parent seen on an exposed port is added here, and every + # provider's allowlists include it, so a restricted agent on any provider reaches environments served there. + EGRESS_HOSTS: ClassVar[tuple[str, ...]] = () + + def __init__( + self, + *, + api_key: str | None = None, + wandb_api_key: str | None = None, + wandb_entity: str | None = None, + wandb_project: str | None = None, + base_url: str | None = None, + image: str = DEFAULT_IMAGE, + docker_storage: str = DOCKER_STORAGE_LOOP, + runtime_class: str = DEFAULT_RUNTIME_CLASS, + placement_mode: str | None = None, + runner_ids: list[str] | None = None, + sandbox_cls: Any = cwsandbox.Sandbox, + ): + if bool(api_key) == bool(wandb_api_key): + raise ValueError("Pass exactly one of api_key and wandb_api_key") + self._image = image + self._docker_storage = docker_storage + self._runtime_class = runtime_class + self._placement_mode = placement_mode + self._runner_ids = runner_ids + self._pinned_images: dict[str, str] = {} + # sandbox_cls is injectable so unit tests never reach the Sandbox API. + self._client = Client( + sandbox_cls=sandbox_cls, + auth=auth_headers({"api_key": api_key, "wandb_api_key": wandb_api_key, "wandb_entity": wandb_entity, + "wandb_project": wandb_project}), + base_url=base_url, + ) + # The SDK's SIGINT/SIGTERM handlers would stop sandboxes a later run reconnects to; agent-env tears down its own. + # Raises only when another user of the SDK in this process installed them first. + with contextlib.suppress(RuntimeError): + cwsandbox.disable_signal_handlers() + + @classmethod + def from_config(cls, **config: Any) -> Self: + unknown = set(config) - _CONFIG_KEYS + if unknown: + raise ConfigError(f"{_SECTION} has unknown key(s): {sorted(unknown)}") + for key in _STRING_KEYS: + value = config.get(key) + if value is not None and (not isinstance(value, str) or not value.strip()): + raise ConfigError(f"{_SECTION} '{key}' must be a non-empty string") + if bool(config.get("api_key")) == bool(config.get("wandb_api_key")): + raise ConfigError(f"{_SECTION} requires exactly one of 'api_key' and 'wandb_api_key'") + if config.get("api_key") and (config.get("wandb_entity") or config.get("wandb_project")): + raise ConfigError(f"{_SECTION} takes 'wandb_entity' and 'wandb_project' only with 'wandb_api_key'") + if config.get("docker_storage", DOCKER_STORAGE_LOOP) not in (DOCKER_STORAGE_LOOP, DOCKER_STORAGE_MEMORY): + raise ConfigError(f"{_SECTION} 'docker_storage' must be {DOCKER_STORAGE_LOOP!r} or " + f"{DOCKER_STORAGE_MEMORY!r}") + runner_ids = config.get("runner_ids") + if runner_ids is not None and (not isinstance(runner_ids, list) or not runner_ids or not all( + isinstance(runner, str) and runner.strip() for runner in runner_ids)): + raise ConfigError(f"{_SECTION} 'runner_ids' must be a non-empty list of non-empty strings") + return cls(**config) + + async def prepare_image(self, image: DockerImageArtifact, *, attribution: Attribution | None = None) -> None: + """Remember the artifact's digest so native containers pull that version rather than its mutable tag.""" + if image.image_digest: + self._pinned_images[image.image_name] = image.pinned_name + else: + self._pinned_images.pop(image.image_name, None) + + @classmethod + def supports_network_policy(cls, policy: NetworkPolicy) -> bool: + """Allow-all, or an allowlist of hostnames, IP addresses and CIDRs. A hostname entry allows HTTPS on port 443 + only. The runner's policy may still refuse hostname rules, or ranges outside the public internet, when the + sandbox is created.""" + try: + network_options(policy) + except ValueError: + return False + return True + + async def create_vm( + self, + *, + image: str | None = None, + boot_mode: str | None = None, + cpu: float = 1.0, + memory: int = 8192, + disk_size_gb: float = 10, + timeout: int = 3600 * 2, + exposed_ports: list[int] | None = None, + setup_for_gateway: bool = True, + attribution: Attribution | None = None, + network_policy: NetworkPolicy | None = None, + ) -> CoreWeaveSandbox: + """Create a CoreWeave Sandbox that is a Docker host and expose ``exposed_ports`` as public HTTPS endpoints. + ``timeout`` is the sandbox's hard maximum lifetime. It boots the configured image, so a per-call ``image`` is + refused rather than ignored; ``boot_mode`` has no CoreWeave equivalent and is ignored.""" + if image is not None: + raise ValueError( + "CoreWeave sandboxes boot the image in [sandbox.providers.coreweave.config]; a per-call image is " + "unsupported" + ) + if boot_mode is not None: + logger.warning("Ignoring boot_mode=%s: CoreWeave sandboxes have no boot modes", boot_mode) + ports = list(dict.fromkeys(exposed_ports or [])) + return await self._launch( + CoreWeaveSandbox, + ports=ports, + timeout=timeout, + attribution=attribution, + network_policy=network_policy, + ready=(lambda sandbox: sandbox.setup_vm_for_gateway(ports)) if setup_for_gateway else None, + command="/bin/sh", + args=["-c", host_setup_script(self._docker_storage, docker_disk_gb(disk_size_gb))], + container_image=self._image, + resources=resources(cpu, memory), + runtime_class=self._runtime_class, + security_context=SecurityContext(privileged=True), + ) + + async def create_container( + self, + *, + image_name: str, + port: int, + env: dict[str, str], + cpu: float = 1.0, + memory: int = 8192, + disk_size_gb: float = 10, + timeout: int = 3600 * 2, + attribution: Attribution | None = None, + network_policy: NetworkPolicy | None = None, + private_network: bool = False, + ) -> CoreWeaveContainerSandbox: + """Create a CoreWeave Sandbox that runs ``image_name`` itself, unprivileged, with its own entrypoint and + command and ``env``, and expose ``port`` as a public HTTPS endpoint. ``timeout`` is its hard maximum lifetime; + ``disk_size_gb`` has no CoreWeave equivalent and is logged as ignored. The platform pulls the image without the + image store's credentials, so an image in a registry they are for is refused.""" + if private_network: + raise ValueError("CoreWeave sandboxes share no private network, so containers can't be put on one") + # Every caller passes a size, so this is info rather than a warning on each deploy. + logger.info("Ignoring disk_size_gb=%s: a CoreWeave container sandbox has no disk size to set", disk_size_gb) + # Chosen before any await, so another deploy preparing this tag at another digest meanwhile can't swap it. + image = self._pinned_images.get(image_name, image_name) + store = get_config().get_image_store_at(image_name) + if (auth := await asyncio.to_thread(store.auth, image_name)) is not None: + raise ValueError( + f"CoreWeave pulls {image_name} itself, without the credentials agent-env's image store holds for " + f"{auth.registry}; run it from a registry CoreWeave can pull from without them" + ) + return await self._launch( + CoreWeaveContainerSandbox, + ports=[port], + timeout=timeout, + attribution=attribution, + network_policy=network_policy, + # A single container's spec, not the SDK's single-container arguments, which would replace an unset command + # with its own default: the image's entrypoint and command must run. + containers=[Container(image=image, environment_variables=dict(env), resources=resources(cpu, memory))], + ) + + async def create_sandbox( + self, + *, + image_name: str, + port: int, + env: dict[str, str], + cpu: float = 1.0, + memory: int = 8192, + disk_size_gb: float = 10, + timeout: int = 3600 * 2, + attribution: Attribution | None = None, + network_policy: NetworkPolicy | None = None, + ) -> CoreWeaveContainerSandbox: + return await self.create_container( + image_name=image_name, port=port, env=env, cpu=cpu, memory=memory, disk_size_gb=disk_size_gb, + timeout=timeout, attribution=attribution, network_policy=network_policy, + ) + + async def _launch( + self, + adapter: Any, + *, + ports: list[int], + timeout: int, + attribution: Attribution | None, + network_policy: NetworkPolicy | None, + ready: Callable[[Any], Awaitable[None]] | None = None, + **spec: Any, + ) -> Any: + """Create a sandbox from ``spec``, wait for it to run and for ``ready``, and wrap it in ``adapter``; one whose + create fails or is cancelled is deleted before this ends.""" + if isinstance(timeout, bool) or not isinstance(timeout, int) or timeout <= 0: + raise ValueError("timeout must be a positive integer number of seconds") + effective_policy = self.effective_network_policy(network_policy) + if not self.supports_network_policy(effective_policy): + raise NetworkPolicyUnsupportedError( + f"CoreWeave enforces allow-all or an allowlist of hostnames, IP addresses and CIDRs, not " + f"{effective_policy.to_dict()}" + ) + metadata = { + key: str(value) + for key, value in apply_default_attribution(dict(attribution or {})).items() + if value is not None + } + create_tag = f"{SANDBOX_TAG}-{uuid.uuid4().hex[:12]}" + raw = self._client.sandbox_cls( + auth=self._client.auth, + base_url=self._client.base_url, + max_lifetime_seconds=timeout, + tags=sandbox_tags(metadata, create_tag), + runner_ids=self._runner_ids, + placement_mode=self._placement_mode, + network=network_options(effective_policy), + services=services(ports), + **spec, + ) + await self._start(raw, create_tag) + applied = policy_from_egress(raw.effective_egress) + if applied is None: + logger.warning("CoreWeave sandbox %s reports egress rules agent-env can't represent; its network policy " + "is recorded as unknown", raw.sandbox_id) + sandbox = adapter(raw, client=self._client, tunnel_urls=tunnel_urls(raw), network_policy=applied, + command_timeout=timeout) + try: + if missing := [port for port in ports if port not in sandbox.tunnel_urls]: + raise RuntimeError(f"CoreWeave sandbox {sandbox.sandbox_id} has no endpoint for port(s) {missing}") + logger.info("CoreWeave sandbox started: sandbox_id=%s mode=%s runner=%s ports=%s attribution=%s", + sandbox.sandbox_id, sandbox.mode, raw.runner_id, ports, metadata) + if ready is not None: + await ready(sandbox) + return sandbox + except BaseException: + await _finish(_reap(self._client, sandbox.sandbox_id)) + raise + + async def _start(self, raw: Any, create_tag: str) -> None: + """Start ``raw`` and wait for it to run. The SDK runs the create on its own event loop, so cancelling this task + doesn't stop a create the server has taken; a sandbox that would otherwise run with no handle until its max + lifetime is deleted before this ends: the one a failed or cancelled create makes, and the one it may have made + with its response lost.""" + starting = asyncio.ensure_future(raw.start()) + try: + await asyncio.shield(starting) + await raw + except BaseException: + await _finish(_reclaim(self._client, raw, create_tag, starting)) + raise + + async def get_sandbox(self, sandbox_id: str) -> CoreWeaveContainerSandbox | CoreWeaveSandbox: + """Reconnect, restoring the mode, the exposed ports and the applied egress policy. The mode is read from the + sandbox's own spec: a VM is the one started by the Docker host's setup script.""" + try: + raw = await self._client.get(sandbox_id) + except cwsandbox.SandboxNotFoundError as exc: + raise SandboxNotFoundError(f"CoreWeave sandbox {sandbox_id!r} was not found") from exc + if raw.status in _GONE_STATUSES: + raise SandboxNotFoundError(f"CoreWeave sandbox {sandbox_id!r} has ended ({raw.status})") + docker_host = runs_docker_host(raw) + if docker_host is None: + raise RuntimeError(f"CoreWeave sandbox {sandbox_id!r} reports no container spec, so whether it is a " + "container or a VM is unknown") + adapter = CoreWeaveSandbox if docker_host else CoreWeaveContainerSandbox + return adapter( + raw, + client=self._client, + tunnel_urls=tunnel_urls(raw), + network_policy=policy_from_egress(raw.effective_egress), + command_timeout=_RECONNECT_COMMAND_TIMEOUT_SEC, + ) + + async def terminate_sandbox(self, sandbox_id: str) -> None: + await self._client.delete(sandbox_id) + + +def _remember_egress_host(host: str) -> None: + known = CoreWeaveSandboxProvider.EGRESS_HOSTS + if host not in known and len(known) < _MAX_ENDPOINT_PARENTS: + CoreWeaveSandboxProvider.EGRESS_HOSTS = (*known, host) + + +# Deletes of sandboxes a create made after its cleanup stopped waiting, held so they aren't collected mid-run. +_late_reaps: set[asyncio.Task] = set() + + +def _reap_late_start(client: Client, raw: Any, create_tag: str, starting: asyncio.Future) -> None: + """Delete the sandbox a start request made after its create's cleanup stopped waiting for it, whether the request + answered with its id or failed with its response lost. This runs only while the event loop does; a sandbox made + after it closes stops at its max lifetime.""" + if starting.cancelled(): + return + starting.exception() + task = asyncio.ensure_future(_reap_created(client, raw, create_tag)) + _late_reaps.add(task) + task.add_done_callback(_late_reaps.discard) + + +async def _reap_created(client: Client, raw: Any, create_tag: str) -> None: + """Delete the sandbox a create made: the one with the id its start request answered, or else the one tagged with + the create's unique tag.""" + if raw.sandbox_id: + await _reap(client, raw.sandbox_id) + return + try: + found = await client.list([create_tag]) + except Exception as exc: # noqa: BLE001 - best effort; the sandbox still stops at its max lifetime + logger.warning("Could not look for a CoreWeave sandbox tagged %s left by a failed create: %s", create_tag, exc) + return + for sandbox in found: + await _reap(client, sandbox.sandbox_id) diff --git a/src/agent_env/providers/sandbox_providers/coreweave/sandbox.py b/src/agent_env/providers/sandbox_providers/coreweave/sandbox.py new file mode 100644 index 000000000..d7c82eb8d --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/coreweave/sandbox.py @@ -0,0 +1,405 @@ +"""CoreWeave Sandboxes presented through the :class:`Sandbox` contracts, in two modes. + +A container sandbox runs a workload's image itself, with its own entrypoint and command; exec and file writes reach +that container. A VM sandbox is one privileged container in a Kata VM, booted from a Docker host image: its start +command readies the host (installs the tools it lacks, mounts Docker's storage, starts dockerd), and the gateway's +docker-compose, an agent's ``docker run`` and image loading run under that Docker. Every command on a VM runs as root, +so a leading ``sudo`` is dropped. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import shlex +from collections.abc import AsyncIterator, Callable, Iterable +from dataclasses import dataclass +from ipaddress import ip_address, ip_network +from typing import TYPE_CHECKING, Any +from urllib.parse import urlparse + +import httpx +from cwsandbox import ( + EgressRule, + NetworkOptions, + SandboxError, + SandboxExecutionError, + SandboxRequestTimeoutError, + SandboxStreamTruncatedError, + SandboxTimeoutError, + SandboxUnavailableError, +) + +from agent_env.config import get_config +from agent_env.providers.sandbox_providers.sandbox import ( + CURL_RETRY_FLAGS, + NetworkMode, + NetworkPolicy, + Sandbox, + VmSandbox, + push_object_over_stdin, +) +from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER, SANDBOX_MODE_VM + +if TYPE_CHECKING: + from agent_env.store.object_store import ObjectStore + +logger = logging.getLogger(__name__) + +DOCKER_STORAGE_LOOP = "loop" +DOCKER_STORAGE_MEMORY = "memory" +_HOST_DIR = "/run/agentenv" +_HOST_LOG = "/var/log/agentenv-host.log" +HOST_SETUP_TIMEOUT_SEC = 600 +_DOCKERD_READY_TIMEOUT_SEC = 120 +# How long the host-wait exec outlives the wait it runs, so the wait's own timeout message comes back. +_HOST_WAIT_GRACE_SEC = 60 +_URL_DOWNLOAD_TIMEOUT_SEC = 120 +# What exec returns when the stream drops without an exit status; exec_script retries only this. +_UNKNOWN_EXIT_CODE = -1 +_TIMED_OUT_EXIT_CODE = 124 +_WAIT_ATTEMPTS = 3 + + +def host_setup_script(docker_storage: str, disk_gb: int) -> str: + """The sandbox's start command: install bash, curl and (for a loop-mounted disk) e2fsprogs when the image lacks + them, mount Docker's storage, start dockerd, and mark the host ready or failed. overlayfs can't sit on the + sandbox's virtiofs root, so ``/var/lib/docker`` is an ext4 file loop-mounted there or, for ``memory``, a tmpfs.""" + if docker_storage == DOCKER_STORAGE_LOOP: + tools, packages = "bash curl mkfs.ext4", "bash curl e2fsprogs" + mount = (f"truncate -s {disk_gb}G /agentenv-docker.img && mkfs.ext4 -q -F /agentenv-docker.img && " + "mount -o loop /agentenv-docker.img /var/lib/docker") + else: + tools, packages = "bash curl", "bash curl" + mount = f"mount -t tmpfs -o size={disk_gb}g tmpfs /var/lib/docker" + return f""" +# The image's entrypoint, when it runs, points the client at a docker:2375 service when it starts before dockerd. +export DOCKER_HOST=unix:///var/run/docker.sock +setup() {{ + for tool in {tools}; do + command -v "$tool" >/dev/null 2>&1 && continue + if ! command -v apk >/dev/null 2>&1 || ! apk add --no-cache {packages}; then + echo "the image lacks {packages} and they could not be installed; set image to a prepared host image" \\ + "(a network policy restricting egress blocks the package mirror)" + return 1 + fi + break + done + mkdir -p /var/lib/docker && {mount} || return 1 + dockerd --host=unix:///var/run/docker.sock & + for _ in $(seq {_DOCKERD_READY_TIMEOUT_SEC}); do + docker info >/dev/null 2>&1 && return 0 + sleep 1 + done + echo "dockerd did not become ready" + return 1 +}} +mkdir -p {_HOST_DIR} +if setup >{_HOST_LOG} 2>&1; then : >{_HOST_DIR}/ready; else : >{_HOST_DIR}/failed; fi +trap "exit 0" TERM INT +sleep infinity & +wait +""" + + +_WAIT_FOR_HOST = f""" +for _ in $(seq {HOST_SETUP_TIMEOUT_SEC}); do + [ -e {_HOST_DIR}/ready ] && exit 0 + [ -e {_HOST_DIR}/failed ] && break + sleep 1 +done +[ -e {_HOST_DIR}/failed ] || echo "host setup did not finish within {HOST_SETUP_TIMEOUT_SEC}s" >&2 +tail -n 40 {_HOST_LOG} >&2 +exit 1 +""" + + +def runs_docker_host(raw: Any) -> bool | None: + """Whether the sandbox ``raw`` describes was started as a Docker host, by ``host_setup_script``; None when the + Sandbox API echoes no container spec to tell by, as while it prepares the image.""" + containers = list(raw.containers or ()) + if not containers: + return None + primary = next((container for container in containers if container.primary), containers[0]) + return any(_HOST_LOG in arg for arg in primary.args or ()) + + +def _cidr_or_none(entry: str) -> str | None: + try: + return str(ip_network(entry, strict=False)) + except ValueError: + return None + + +def network_options(policy: NetworkPolicy) -> NetworkOptions | None: + """``Sandbox.run`` network options for ``policy``: none for allow-all, which leaves the runner's default egress (the + public internet on CoreWeave's runners), and deny-all for an allowlist with no entries. An IP address listed as a + host becomes a CIDR rule. A malformed entry raises ValueError.""" + if not policy.restricts_egress: + return None + rules = [EgressRule(cidr=cidr) if (cidr := _cidr_or_none(host)) else EgressRule(dns_name=host) + for host in policy.allow_hosts] + for entry in policy.allow_cidrs: + if (cidr := _cidr_or_none(entry)) is None: + raise ValueError(f"{entry!r} is not a CIDR") + rules.append(EgressRule(cidr=cidr)) + return NetworkOptions(egress=rules) if rules else NetworkOptions(deny_egress=True) + + +def policy_from_egress(rules: Iterable[Any]) -> NetworkPolicy | None: + """The policy a sandbox's effective egress rules apply, or None for rules agent-env can't represent. A rule for + every destination, or for all of IPv4 with private ranges carved out (a runner's default), is allow-all: no + sandbox off this machine reaches those ranges anyway.""" + hosts: list[str] = [] + cidrs: list[str] = [] + for rule in rules: + if rule.any: + return NetworkPolicy() + if rule.dns_name: + hosts.append(rule.dns_name) + continue + if rule.cidr is None or rule.ports: + return None + if rule.cidr.cidr == "0.0.0.0/0": + return NetworkPolicy() + if rule.cidr.except_cidrs: + return None + cidrs.append(rule.cidr.cidr) + return NetworkPolicy(mode=NetworkMode.ALLOWLIST, allow_hosts=tuple(hosts), allow_cidrs=tuple(cidrs)) + + +def allows_download(policy: NetworkPolicy | None, url: str | None) -> bool: + """Whether a sandbox under ``policy`` can fetch ``url`` itself; an unknown policy counts as no. CoreWeave grants a + hostname for HTTPS on port 443 only, and a ``*.domain`` entry covers exactly one label under ``domain``.""" + if url is None or policy is None: + return False + if not policy.restricts_egress: + return True + parsed = urlparse(url) + host = (parsed.hostname or "").lower().rstrip(".") + try: + address = ip_address(host) + except ValueError: + if parsed.scheme != "https" or parsed.port not in (None, 443): + return False + for entry in policy.allow_hosts: + entry = entry.lower().rstrip(".") + if entry == host: + return True + if entry.startswith("*.") and host.endswith(entry[1:]): + label = host.removesuffix(entry[1:]) + if label and "." not in label: + return True + return False + networks = [cidr for entry in (*policy.allow_hosts, *policy.allow_cidrs) if (cidr := _cidr_or_none(entry))] + return any(address in ip_network(cidr) for cidr in networks) + + +@dataclass(frozen=True) +class Client: + """The SDK's ``Sandbox`` class, with the credentials and API endpoint every call a provider makes uses.""" + + sandbox_cls: Any + auth: Any + base_url: str | None + + async def get(self, sandbox_id: str) -> Any: + return await self.sandbox_cls.from_id(sandbox_id, auth=self.auth, base_url=self.base_url) + + async def list(self, tags: list[str]) -> list[Any]: + return await self.sandbox_cls.list(tags=tags, auth=self.auth, base_url=self.base_url) + + async def delete(self, sandbox_id: str) -> None: + await self.sandbox_cls.delete(sandbox_id, auth=self.auth, base_url=self.base_url, missing_ok=True) + + +async def _settle(process: Any, timeout: float) -> tuple[int, bytes, bytes]: + """``process``'s exit code, stdout and stderr. A stream that dropped or was cut short is exit -1 with no stdout, so + a caller that skips the exit code can't mistake partial output for the whole; a timed-out command is exit 124.""" + try: + result = await process + except SandboxError as exc: + # A bare SandboxExecutionError without a result is a stream that closed with no exit status. + if isinstance(exc, (SandboxUnavailableError, SandboxStreamTruncatedError, SandboxRequestTimeoutError)) or ( + type(exc) is SandboxExecutionError and exc.exec_result is None): + return _UNKNOWN_EXIT_CODE, b"", f"{type(exc).__name__}: {exc}".encode() + if isinstance(exc, SandboxTimeoutError): + return _TIMED_OUT_EXIT_CODE, b"", f"command timed out after {timeout}s".encode() + raise + return result.returncode, result.stdout_bytes, result.stderr_bytes + + +async def _sent(send: Callable[[], Any], finished: asyncio.Future) -> bool: + """Send one stdin write or close; True when the command ended first, so nothing more is worth sending. The SDK + queues stdin with a bound, and a command that has ended stops reading it, so a send could otherwise wait forever.""" + try: + sending = asyncio.ensure_future(send()) + except SandboxExecutionError: + return True + await asyncio.wait({sending, finished}, return_when=asyncio.FIRST_COMPLETED) + if not sending.done(): + sending.cancel() + return True + if isinstance(sending.exception(), SandboxExecutionError): + return True + sending.result() + return False + + +class _BytesReader: + """The small ``StreamReader`` subset consumed by ``Sandbox.exec_with_output``.""" + + def __init__(self, value: bytes): + self._value = value + + async def read(self) -> bytes: + return self._value + + +class _CompletedProcess: + """A finished CoreWeave exec in the common process protocol.""" + + def __init__(self, exit_code: int, stdout: bytes, stderr: bytes): + self.stdout = _BytesReader(stdout) + self.stderr = _BytesReader(stderr) + self._exit_code = exit_code + + async def wait(self) -> int: + return self._exit_code + + +class _CoreWeaveSandbox(Sandbox): + """A CoreWeave Sandbox. ``command_timeout`` is every command's execution budget, normally the sandbox's lifetime: + the SDK otherwise bounds a command to five minutes.""" + + type = "coreweave" + + def __init__( + self, + sandbox: Any, + *, + client: Client, + tunnel_urls: dict[int, str], + network_policy: NetworkPolicy | None, + command_timeout: int, + ): + self._sandbox = sandbox + self._client = client + self.command_timeout = command_timeout + self.sandbox_id = sandbox.sandbox_id + self.vnc_url = None + self.tunnel_urls = tunnel_urls + self.network_policy = network_policy + + async def terminate(self) -> None: + await self._client.delete(self.sandbox_id) + + async def _run(self, *command: str, timeout: float) -> tuple[int, bytes, bytes]: + return await _settle(self._sandbox.exec(list(command), timeout_seconds=timeout), timeout) + + async def exec(self, *command: str) -> _CompletedProcess: + """Run argv to completion. A lost stream is exit -1, which a VM's ``exec_script`` retries.""" + return _CompletedProcess(*await self._run(*command, timeout=self.command_timeout)) + + async def exec_with_output(self, *args: str) -> tuple[int, str, str]: + exit_code, stdout, stderr = await self._run(*args, timeout=self.command_timeout) + return exit_code, stdout.decode(errors="replace"), stderr.decode(errors="replace") + + +class CoreWeaveContainerSandbox(_CoreWeaveSandbox): + """A CoreWeave Sandbox running a workload's image itself, with no Docker. Commands run as the image's user. Files + go through the Sandbox API's file write, which needs ``sh``, ``mkdir``, ``head`` and ``tar`` in the image and takes + at most about 256 MiB.""" + + mode = SANDBOX_MODE_CONTAINER + + async def write_file_from_text(self, content: str, destination_path: str) -> None: + await self._sandbox.write_file(destination_path, content.encode()) + + async def write_file_from_object(self, object_url: str, destination_path: str) -> None: + store = get_config().get_object_store_at(object_url) + await self._sandbox.write_file(destination_path, await asyncio.to_thread(_read_object, store, object_url)) + + async def write_file_from_url(self, url: str, destination_path: str) -> None: + """Fetched here and written over the Sandbox API, since the image need not have curl.""" + async with httpx.AsyncClient(follow_redirects=True, timeout=_URL_DOWNLOAD_TIMEOUT_SEC) as http: + response = await http.get(url) + response.raise_for_status() + await self._sandbox.write_file(destination_path, response.content) + + +def _read_object(store: ObjectStore, object_url: str) -> bytes: + with contextlib.closing(store.open(object_url)) as body: + return body.read() + + +class CoreWeaveSandbox(_CoreWeaveSandbox, VmSandbox): + """A CoreWeave Sandbox that is a Docker host.""" + + mode = SANDBOX_MODE_VM + + async def _run(self, *command: str, timeout: float) -> tuple[int, bytes, bytes]: + return await super()._run(*(command[1:] if command[:1] == ("sudo",) else command), timeout=timeout) + + async def _exec_with_stdin(self, script: str, stdin: AsyncIterator[bytes]) -> tuple[int, str, str]: + """Run ``script`` with bash as root, its standard input the bytes ``stdin`` yields.""" + process = self._sandbox.exec(["bash", "-c", script], stdin=True, timeout_seconds=self.command_timeout) + finished = asyncio.ensure_future(_settle(process, self.command_timeout)) + try: + async for piece in stdin: + if await _sent(lambda: process.stdin.write(piece), finished): + break + else: + await _sent(process.stdin.close, finished) + exit_code, stdout, stderr = await finished + except BaseException: + # The command reads to the end of its input, so closing it ends the command rather than its lifetime. + process.stdin.close() + finished.cancel() + raise + return exit_code, stdout.decode(errors="replace"), stderr.decode(errors="replace") + + async def wait_for_vm(self) -> None: + """Wait for the start command to ready Docker, and raise what it reported if it couldn't.""" + for _ in range(_WAIT_ATTEMPTS): + exit_code, stdout, stderr = await self._run("sh", "-c", _WAIT_FOR_HOST, + timeout=HOST_SETUP_TIMEOUT_SEC + _HOST_WAIT_GRACE_SEC) + if exit_code != _UNKNOWN_EXIT_CODE: + break + if exit_code != 0: + output = (stderr or stdout).decode(errors="replace").strip() + raise RuntimeError(f"Docker not ready in CoreWeave sandbox {self.sandbox_id}: {output[-1500:]}") + logger.info("Docker ready in CoreWeave sandbox %s", self.sandbox_id) + + async def setup_vm_for_gateway(self, exposed_ports: list[int] | None = None) -> None: + """Ready Docker; CoreWeave routes exposed ports to the sandbox, so no firewall.""" + await self.wait_for_vm() + + # Bytes go over exec stdin as root, which reaches any path and needs no egress, so they arrive whatever the + # sandbox's egress policy, which is set when the sandbox is created. + + async def _write_bytes_to_vm_path(self, data: bytes, vm_path: str) -> None: + async def pieces() -> AsyncIterator[bytes]: + yield data + + exit_code, stdout, stderr = await self._exec_with_stdin(f"cat > {shlex.quote(vm_path)}", pieces()) + if exit_code != 0: + raise RuntimeError(f"Writing {vm_path} failed (exit {exit_code}): {(stderr or stdout)[-1500:]}") + + async def _write_unsigned_object(self, object_store: ObjectStore, object_url: str, vm_path: str) -> None: + await push_object_over_stdin(self, object_store, object_url, vm_path) + + async def _download_object_to_vm(self, object_url: str, vm_path: str) -> None: + object_store = get_config().get_object_store_at(object_url) + signed = await asyncio.to_thread(object_store.signed_get_url, object_url) + if allows_download(self.network_policy, signed): + await self.exec_script(f"curl -fsSL {CURL_RETRY_FLAGS} {shlex.quote(signed)} -o {shlex.quote(vm_path)}") + return + await push_object_over_stdin(self, object_store, object_url, vm_path) + + async def _load_tarballs(self, artifacts: list) -> None: + """Load image tarballs, pushing over stdin those the sandbox's egress policy can't download.""" + signed_urls = await self._signed_image_urls(artifacts) + reachable = [url if allows_download(self.network_policy, url) else None for url in signed_urls] + await self._load_docker_images(artifacts, reachable) diff --git a/src/agent_env/providers/sandbox_providers/sandbox_provider.py b/src/agent_env/providers/sandbox_providers/sandbox_provider.py index ac3483577..2ee4dfaef 100644 --- a/src/agent_env/providers/sandbox_providers/sandbox_provider.py +++ b/src/agent_env/providers/sandbox_providers/sandbox_provider.py @@ -341,6 +341,7 @@ def effective_network_policy(cls, policy: Optional[NetworkPolicy]) -> NetworkPol "boxd": "agent_env.providers.sandbox_providers.boxd.provider:BoxdSandboxProvider", "daytona": "agent_env.providers.sandbox_providers.daytona.provider:DaytonaSandboxProvider", "boat": "agent_env.providers.sandbox_providers.boat.provider:BoatSandboxProvider", + "coreweave": "agent_env.providers.sandbox_providers.coreweave.provider:CoreWeaveSandboxProvider", "local": "agent_env.providers.sandbox_providers.local_docker.provider:LocalSandboxProvider", } diff --git a/tst/integration/env/gateway/gateway_test.py b/tst/integration/env/gateway/gateway_test.py index 5d0cbf6b6..ab4057715 100644 --- a/tst/integration/env/gateway/gateway_test.py +++ b/tst/integration/env/gateway/gateway_test.py @@ -483,6 +483,11 @@ def agentenv_website_env() -> WebsiteEnv: id="boat", marks=skip_without_remote_sandbox("boat"), ), + pytest.param( + "coreweave", + id="coreweave", + marks=skip_without_remote_sandbox("coreweave"), + ), ]) def sandbox_provider(request): from agent_env.providers import ( @@ -519,6 +524,9 @@ def sandbox_provider(request): elif request.param == "boat": set_sandbox_provider(build_sandbox_provider("boat")) set_env_sandbox_provider(build_sandbox_provider("boat")) + elif request.param == "coreweave": + set_sandbox_provider(build_sandbox_provider("coreweave")) + set_env_sandbox_provider(build_sandbox_provider("coreweave")) try: yield request.param finally: diff --git a/tst/integration/env/task_step/task_steps_test.py b/tst/integration/env/task_step/task_steps_test.py index d4a8030b3..7745747a4 100644 --- a/tst/integration/env/task_step/task_steps_test.py +++ b/tst/integration/env/task_step/task_steps_test.py @@ -302,6 +302,11 @@ def echo_agent() -> A2AAgent: id="boat", marks=skip_without_remote_sandbox("boat"), ), + pytest.param( + "coreweave", + id="coreweave", + marks=skip_without_remote_sandbox("coreweave"), + ), ]) def sandbox_provider(request): from agent_env.providers import ( @@ -348,6 +353,10 @@ def sandbox_provider(request): set_sandbox_provider(build_sandbox_provider("boat")) set_env_sandbox_provider(build_sandbox_provider("boat")) set_agent_sandbox_provider(build_sandbox_provider("boat")) + elif request.param == "coreweave": + set_sandbox_provider(build_sandbox_provider("coreweave")) + set_env_sandbox_provider(build_sandbox_provider("coreweave")) + set_agent_sandbox_provider(build_sandbox_provider("coreweave")) try: yield request.param finally: diff --git a/tst/integration/providers/sandbox_providers/coreweave_sandbox_smoke_test.py b/tst/integration/providers/sandbox_providers/coreweave_sandbox_smoke_test.py new file mode 100644 index 000000000..d66f22611 --- /dev/null +++ b/tst/integration/providers/sandbox_providers/coreweave_sandbox_smoke_test.py @@ -0,0 +1,217 @@ +"""CoreWeave Sandbox smoke test. + +Runs when the resolved config can build the ``coreweave`` sandbox provider (the ``coreweave`` extra and +``[sandbox.providers.coreweave.config]`` with ``api_key = "secret:cwsandbox_api_key"`` or a ``wandb_api_key``), the +``remote_sandbox`` capability, and skips with the declared reason otherwise. The allowlist test needs a runner whose +policy allows hostname egress rules; on one that allows CIDR rules only, its create fails with the runner's refusal. +The container tests run public nginx images as the sandbox itself, so their runner must be able to pull them. The tests +never read or print the API key. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import os + +import httpx +import pytest +import pytest_asyncio + +from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy +from agent_env.providers.sandbox_providers.sandbox_provider import SandboxNotFoundError, build_sandbox_provider +from tst.util.capabilities import skip_without_remote_sandbox + +_PORT = 8080 +# Serves on port 80 from its own entrypoint, and has the sh, mkdir, head and tar the Sandbox API's file write needs. +_SERVING_IMAGE = "public.ecr.aws/nginx/nginx:stable-alpine" +# The same, run as uid 101 and serving on 8080. +_NON_ROOT_IMAGE = "public.ecr.aws/nginx/nginx-unprivileged:stable-alpine" +_MACHINES = {"linux/amd64": "x86_64", "linux/arm64": "aarch64"} + +pytestmark = [ + pytest.mark.integration, + pytest.mark.int_test_slow, + pytest.mark.asyncio, + skip_without_remote_sandbox("coreweave"), +] + + +async def _wait_for_200(url: str) -> None: + async with httpx.AsyncClient(timeout=10) as client: + for _ in range(60): + try: + if (await client.get(url)).status_code == 200: + return + except httpx.HTTPError: + pass + await asyncio.sleep(2) + pytest.fail(f"{url} never answered 200") + + +@pytest.fixture(scope="module") +def coreweave_provider(): + return build_sandbox_provider("coreweave") + + +@pytest_asyncio.fixture(scope="module") +async def coreweave_sandbox(coreweave_provider): + """One sandbox serving on ``_PORT`` from a container, always deleted after the module.""" + sandbox = await coreweave_provider.create_vm(cpu=1, memory=2048, exposed_ports=[_PORT], timeout=900) + try: + await sandbox.exec_script( + f"docker run -d --name smoke -p {_PORT}:{_PORT} public.ecr.aws/docker/library/python:3.12-slim " + f"python -m http.server {_PORT}" + ) + await _wait_for_200(sandbox.tunnel_urls[_PORT]) + yield sandbox + finally: + await sandbox.terminate() + + +async def test_coreweave_docker_ports_upload_and_reconnect(coreweave_provider, coreweave_sandbox): + """Nested Docker, a public port end to end, an upload over stdin to a root-owned path, and reconnect.""" + assert (coreweave_sandbox.type, coreweave_sandbox.mode) == ("coreweave", "vm") + assert coreweave_sandbox.network_policy == NetworkPolicy() + + exit_code, version, stderr = await coreweave_sandbox.exec_with_output( + "sudo", "docker", "info", "--format", "{{.ServerVersion}}" + ) + assert exit_code == 0, stderr + assert version.strip() + + payload = os.urandom(1024 * 1024 + 17) + await coreweave_sandbox.write_host_file(payload, "/opt/smoke/payload.bin") + digest = await coreweave_sandbox.exec_script("sha256sum /opt/smoke/payload.bin") + assert digest.split()[0] == hashlib.sha256(payload).hexdigest() + + reconnected = await build_sandbox_provider("coreweave").get_sandbox(coreweave_sandbox.sandbox_id) + assert reconnected.tunnel_urls == coreweave_sandbox.tunnel_urls + assert reconnected.network_policy == NetworkPolicy() + assert (await reconnected.exec_with_output("true"))[0] == 0 + + +async def test_coreweave_allowlisted_sandbox_reaches_its_environment_and_nothing_else( + coreweave_provider, coreweave_sandbox, +): + """An allowlisted sandbox reaches an allowed host and another sandbox's public port, and nothing else. Docker + isn't readied: the policy blocks the package mirror the stock image installs curl from, so busybox fetches.""" + policy = NetworkPolicy(mode=NetworkMode.ALLOWLIST, allow_hosts=("example.com",)) + agent = await coreweave_provider.create_vm(cpu=1, memory=2048, timeout=600, setup_for_gateway=False, + network_policy=policy) + try: + async def fetch(url: str) -> int: + return (await agent.exec_with_output("wget", "-q", "-O", "/dev/null", "-T", "15", url))[0] + + assert await fetch("https://example.com") == 0 + assert await fetch(coreweave_sandbox.tunnel_urls[_PORT]) == 0 + assert await fetch("https://www.wikipedia.org") != 0 + + reconnected = await build_sandbox_provider("coreweave").get_sandbox(agent.sandbox_id) + assert reconnected.network_policy.restricts_egress + assert set(reconnected.network_policy.allow_hosts) == set(agent.network_policy.allow_hosts) + finally: + await agent.terminate() + + +async def test_coreweave_container_runs_its_image_as_built_and_reconnects_as_a_container(coreweave_provider): + """The image's own entrypoint serves the port, the env reaches it, files land in it, and a fresh provider + reconnects it as a container; terminating twice is fine.""" + sandbox = await coreweave_provider.create_container(image_name=_SERVING_IMAGE, port=80, env={"SMOKE_MARKER": "on"}, + cpu=1, memory=1024, timeout=600) + try: + assert (sandbox.type, sandbox.mode) == ("coreweave", "container") + assert sandbox.network_policy == NetworkPolicy() + await _wait_for_200(sandbox.tunnel_urls[80]) + + exit_code, cmdline, stderr = await sandbox.exec_with_output("cat", "/proc/1/cmdline") + assert exit_code == 0, stderr + assert "nginx" in cmdline and "sleep" not in cmdline + assert await sandbox.exec_with_output("printenv", "SMOKE_MARKER") == (0, "on\n", "") + exit_code, _, stderr = await sandbox.exec_with_output("ls", "/var/lib/docker") + assert exit_code != 0 # no Docker host was set up + + await sandbox.write_file_from_text("smoke text", "/tmp/smoke/dir/file.txt") + assert await sandbox.exec_with_output("cat", "/tmp/smoke/dir/file.txt") == (0, "smoke text", "") + + reconnected = await build_sandbox_provider("coreweave").get_sandbox(sandbox.sandbox_id) + assert (reconnected.mode, reconnected.tunnel_urls) == ("container", sandbox.tunnel_urls) + assert reconnected.network_policy == NetworkPolicy() + assert (await reconnected.exec_with_output("true"))[0] == 0 + finally: + await sandbox.terminate() + await sandbox.terminate() + with pytest.raises(SandboxNotFoundError): + await build_sandbox_provider("coreweave").get_sandbox(sandbox.sandbox_id) + + +async def test_coreweave_vm_reconnects_as_a_vm(coreweave_sandbox): + reconnected = await build_sandbox_provider("coreweave").get_sandbox(coreweave_sandbox.sandbox_id) + assert reconnected.mode == "vm" + + +async def test_coreweave_allowlisted_container_reaches_what_it_is_allowed_and_nothing_else(coreweave_provider): + """A container's egress is the allowlist it was created with, its own port still serves, and a fresh provider reads + the policy back.""" + policy = NetworkPolicy(mode=NetworkMode.ALLOWLIST, allow_hosts=("example.com",)) + sandbox = await coreweave_provider.create_container(image_name=_SERVING_IMAGE, port=80, env={}, cpu=1, memory=1024, + timeout=600, network_policy=policy) + try: + assert sandbox.network_policy.restricts_egress + await _wait_for_200(sandbox.tunnel_urls[80]) + + async def fetch(url: str) -> int: + return (await sandbox.exec_with_output("wget", "-q", "-O", "/dev/null", "-T", "15", url))[0] + + assert await fetch("https://example.com") == 0 + assert await fetch("https://www.wikipedia.org") != 0 + + reconnected = await build_sandbox_provider("coreweave").get_sandbox(sandbox.sandbox_id) + assert reconnected.mode == "container" + assert set(reconnected.network_policy.allow_hosts) == set(sandbox.network_policy.allow_hosts) + finally: + await sandbox.terminate() + + +async def test_coreweave_container_of_a_non_root_image_takes_files_its_user_can_change(coreweave_provider): + """Commands run as the image's own user, files written over the Sandbox API are that user's to change, and a write + where that user can't write fails rather than appearing to succeed.""" + sandbox = await coreweave_provider.create_container(image_name=_NON_ROOT_IMAGE, port=8080, env={}, cpu=1, + memory=1024, timeout=600) + try: + await _wait_for_200(sandbox.tunnel_urls[8080]) + exit_code, uid, stderr = await sandbox.exec_with_output("id", "-u") + assert exit_code == 0, stderr + assert uid.strip() != "0" + + await sandbox.write_file_from_text("smoke text", "/tmp/smoke/dir/file.txt") + assert await sandbox.exec_with_output("cat", "/tmp/smoke/dir/file.txt") == (0, "smoke text", "") + exit_code, _, stderr = await sandbox.exec_with_output("sh", "-c", "echo more >> /tmp/smoke/dir/file.txt") + assert exit_code == 0, stderr + + with pytest.raises(Exception): + await sandbox.write_file_from_text("smoke text", "/etc/smoke/file.txt") + assert (await sandbox.exec_with_output("test", "-e", "/etc/smoke/file.txt"))[0] != 0 + finally: + await sandbox.terminate() + + +async def test_coreweave_sandboxes_run_the_platform_and_runtime_they_report(coreweave_provider, coreweave_sandbox): + """A container and a VM, and the VM's Docker, run the architecture ``platform`` names, and the VM runs under the + configured runtime class.""" + container = await coreweave_provider.create_container(image_name=_SERVING_IMAGE, port=80, env={}, cpu=1, + memory=1024, timeout=600) + try: + assert container.platform in _MACHINES + assert await container.exec_with_output("uname", "-m") == (0, _MACHINES[container.platform] + "\n", "") + finally: + await container.terminate() + + assert coreweave_sandbox.platform in _MACHINES + machine = _MACHINES[coreweave_sandbox.platform] + assert (await coreweave_sandbox.exec_script("uname -m")).strip() == machine + exit_code, architecture, stderr = await coreweave_sandbox.exec_with_output( + "sudo", "docker", "info", "--format", "{{.Architecture}}" + ) + assert (exit_code, architecture.strip()) == (0, machine), stderr + assert coreweave_sandbox._sandbox.effective_runtime_class == coreweave_provider._runtime_class diff --git a/tst/unit/providers/sandbox_providers/coreweave_deploy_paths_test.py b/tst/unit/providers/sandbox_providers/coreweave_deploy_paths_test.py new file mode 100644 index 000000000..fb8ad2e74 --- /dev/null +++ b/tst/unit/providers/sandbox_providers/coreweave_deploy_paths_test.py @@ -0,0 +1,126 @@ +"""Which CoreWeave sandbox each deploy gets: an agent and a server without a gateway run their image as a container of +their own, and a gateway runs on a Docker host. The Sandbox API is faked; the deploys run for real.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from agent_env.a2a_agent.a2a_agent import A2AAgent +from agent_env.artifact.artifacts.docker_image import DockerImageArtifact +from agent_env.env.gateway import AGENT_ENV_GATEWAY_MCP_PORT +from agent_env.providers.env_providers import env_server_provider +from agent_env.providers.env_providers.env_gateway_provider import EnvironmentGatewayProvider +from agent_env.providers.env_providers.env_server_provider import EnvironmentServerProvider +from agent_env.providers.sandbox_providers import sandbox_provider +from agent_env.providers.sandbox_providers.coreweave.provider import CoreWeaveSandboxProvider +from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER, Accepts +from tst.unit.providers.sandbox_providers.coreweave_fakes import fake_sandbox_cls + +AGENT_IMAGE = DockerImageArtifact(id="solver-image", version=2, description="d", image_name="registry.example/solver:v2") +PUBLISHED_IMAGE = AGENT_IMAGE.model_copy(update={ + "image_digest": "sha256:" + "a" * 64, + "build_context_object_url": "s3://bucket/solver-context.tar.gz", + "dockerfile_path": "Dockerfile", +}) +TARBALL_ONLY = DockerImageArtifact(id="solver-image", version=3, description="d", image_name="local/solver:v3", + tar_gz_object_url="s3://bucket/solver.tar.gz") + + +@pytest.fixture(autouse=True) +def _no_endpoint_parents(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(CoreWeaveSandboxProvider, "EGRESS_HOSTS", ()) + + +@pytest.fixture +def litellm(monkeypatch): + monkeypatch.setenv("LITELLM_API_KEY", "k") + monkeypatch.setenv("LITELLM_BASE_URL", "http://litellm.example/v1") + + +@pytest.fixture +def coreweave(monkeypatch): + sandbox_cls = fake_sandbox_cls() + provider = CoreWeaveSandboxProvider(api_key="cw_test", sandbox_cls=sandbox_cls) + monkeypatch.setattr(sandbox_provider, "build_sandbox_provider", lambda name: provider) + return SimpleNamespace(provider=provider, server=sandbox_cls.server) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("image", [AGENT_IMAGE, PUBLISHED_IMAGE], ids=["tag", "published-with-context"]) +async def test_an_agent_runs_its_image_as_a_container_with_no_docker(monkeypatch, local_stores, litellm, coreweave, + image): + monkeypatch.setattr(A2AAgent, "_wait_for_agent_card", AsyncMock(return_value={"name": "solver"})) + + deployed = await A2AAgent(id="solver", version=1, docker_image_artifact=image).deploy(sandbox_type="coreweave") + + [raw] = coreweave.server.created + [container] = raw.kwargs["containers"] + assert (container.image, container.command, container.args) == (image.pinned_name, None, None) + assert container.environment_variables["A2A_PORT"] == "8000" + assert container.environment_variables["LITELLM_BASE_URL"] == "http://litellm.example/v1" + assert [service.port for service in raw.kwargs["services"]] == [8000] + assert "security_context" not in raw.kwargs + assert raw.exec_calls == [] # nothing loaded or started under Docker + assert (deployed.sandbox_type, deployed.sandbox_id, deployed.a2a_url) == ("coreweave", "sb-1", raw.service_endpoints[0].url) + assert deployed.network_policy == {"mode": "allow_all", "allow_hosts": [], "allow_cidrs": []} + + +@pytest.mark.asyncio +async def test_an_agent_whose_image_is_only_a_tarball_is_refused_before_any_sandbox(local_stores, litellm, coreweave): + with pytest.raises(ValueError, match="which runs an agent's image by name"): + await A2AAgent(id="solver", version=1, docker_image_artifact=TARBALL_ONLY).deploy(sandbox_type="coreweave") + assert coreweave.server.created == [] + + +@pytest.mark.asyncio +async def test_an_agent_that_wants_docker_on_its_own_sandbox_is_refused_and_its_sandbox_deleted( + local_stores, litellm, coreweave): + """Docker for an agent needs a VM: deploy one with a deploy_sandbox step and place the agent on it.""" + with pytest.raises(ValueError, match="enable_docker=True requires a VM sandbox"): + await A2AAgent(id="solver", version=1, docker_image_artifact=AGENT_IMAGE).deploy( + sandbox_type="coreweave", enable_docker=True) + assert coreweave.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("digest", [None, "sha256:" + "b" * 64], ids=["tag", "digest"]) +async def test_a_server_without_a_gateway_runs_its_image_as_a_container(coreweave, digest): + env = SimpleNamespace(id="slack-bare", version=2, environment_name="slack", + docker_image_artifact=DockerImageArtifact(id="slack-image", description="d", + image_name="registry.example/mcp-slack:1", + image_digest=digest)) + server = EnvironmentServerProvider() + server._wait_for_tunnel = AsyncMock(return_value={"name": "slack"}) + with patch.object(env_server_provider, "_tool_names", new=AsyncMock(return_value=["slack_list"])): + record = await server.deploy(env, coreweave.provider, ttl_seconds=600) + + [raw] = coreweave.server.created + [container] = raw.kwargs["containers"] + assert (container.image, container.command) == (env.docker_image_artifact.pinned_name, None) + assert container.environment_variables["ENVIRONMENT_NAME"] == "slack" + assert [service.port for service in raw.kwargs["services"]] == [AGENT_ENV_GATEWAY_MCP_PORT] + assert raw.kwargs["max_lifetime_seconds"] == 600 and raw.exec_calls == [] + assert (record.sandbox_type, record.sandbox_id) == ("coreweave", "sb-1") + assert server.environment_sandbox("slack").mode == SANDBOX_MODE_CONTAINER + + +@pytest.mark.asyncio +async def test_preparing_another_version_of_a_tag_replaces_or_removes_its_digest(coreweave): + for digest in ("sha256:" + "a" * 64, "sha256:" + "b" * 64, None): + image = AGENT_IMAGE.model_copy(update={"image_digest": digest}) + await coreweave.provider.prepare_image(image) + await coreweave.provider.create_container(image_name=image.image_name, port=8000, env={}) + assert coreweave.server.created[-1].kwargs["containers"][0].image == image.pinned_name + + +def test_a_gateway_runs_on_a_docker_host_and_loads_any_image_it_can(): + provider = CoreWeaveSandboxProvider(api_key="cw_test", sandbox_cls=fake_sandbox_cls()) + assert EnvironmentGatewayProvider.in_vm(provider) + assert EnvironmentGatewayProvider.accepts(provider) is Accepts.LOADABLE + assert EnvironmentGatewayProvider.sandbox_problem(provider, websites=True) is None + assert EnvironmentServerProvider.accepts(provider) is Accepts.NAME + assert A2AAgent.accepts(provider) is Accepts.NAME + assert A2AAgent.accepts(provider, sandbox_mode="vm") is Accepts.LOADABLE diff --git a/tst/unit/providers/sandbox_providers/coreweave_fakes.py b/tst/unit/providers/sandbox_providers/coreweave_fakes.py new file mode 100644 index 000000000..f3181485d --- /dev/null +++ b/tst/unit/providers/sandbox_providers/coreweave_fakes.py @@ -0,0 +1,172 @@ +"""In-memory stand-ins for the cwsandbox SDK's Sandbox, Process and stdin writer.""" + +from __future__ import annotations + +import asyncio +import itertools +from dataclasses import dataclass, field +from types import SimpleNamespace +from typing import Any + +from cwsandbox import CidrBlock, Container, EgressRule, HttpsEndpointStatus, SandboxExecutionError, SandboxNotFoundError +from cwsandbox import EndpointAuth, EndpointKind, SandboxStatus + +# A runner's default egress: the public IPv4 internet. +DEFAULT_EGRESS = (EgressRule(cidr=CidrBlock(cidr="0.0.0.0/0", except_cidrs=("10.0.0.0/8", "192.168.0.0/16"))),) + + +@dataclass +class Result: + returncode: int = 0 + stdout_bytes: bytes = b"" + stderr_bytes: bytes = b"" + + +def _done(value: Any = None) -> asyncio.Future: + """An operation that has completed, as an SDK ``OperationRef`` the caller may or may not await.""" + future = asyncio.get_running_loop().create_future() + future.set_result(value) + return future + + +class FakeStdin: + def __init__(self, process: FakeProcess): + self._process = process + self.pieces: list[bytes] = [] + self.closed = False + + def write(self, data: bytes) -> Any: + if self._process.finished.done(): + raise SandboxExecutionError("Cannot write to stdin: stream has failed") + if self._process.read_limit is not None and len(self.pieces) >= self._process.read_limit: + return asyncio.get_running_loop().create_future() # the command stopped reading: the queue stays full + self.pieces.append(data) + return _done() + + def close(self) -> Any: + self.closed = True + if not self._process.finished.done() and self._process.outcome_on_close: + self._process.finish() + return _done() + + +class FakeProcess: + """Resolves to ``outcome`` (a Result, or an exception raised) when awaited, or, with stdin, once stdin closes or + the command ends by itself after reading ``read_limit`` pieces.""" + + def __init__(self, outcome: Result | BaseException, *, stdin: bool, read_limit: int | None = None): + self.outcome = outcome + self.finished: asyncio.Future = asyncio.get_running_loop().create_future() + self.read_limit = read_limit + self.outcome_on_close = stdin + self.stdin = FakeStdin(self) if stdin else None + if not stdin: + self.finish() + + def finish(self) -> None: + if isinstance(self.outcome, BaseException): + self.finished.set_exception(self.outcome) + else: + self.finished.set_result(self.outcome) + + def __await__(self): + return asyncio.shield(self.finished).__await__() + + +@dataclass +class FakeServer: + """What the Sandbox API holds and was asked for, shared by one fake ``Sandbox`` class and its instances.""" + + sandboxes: dict[str, FakeSandbox] = field(default_factory=dict) + created: list[FakeSandbox] = field(default_factory=list) + deleted: list[str] = field(default_factory=list) + listed: list[list[str]] = field(default_factory=list) + calls: list[tuple[str, dict[str, Any]]] = field(default_factory=list) + start_error: BaseException | None = None + # The server made the sandbox but the start's response was lost: the sandbox exists, then start_error is raised. + start_commits: bool = False + start_delay: float = 0.0 + ids: Any = field(default_factory=lambda: (f"sb-{n}" for n in itertools.count(1))) + + +class FakeSandbox: + """Records the create kwargs; ``results`` answers exec in order, ``exec_calls`` records each call, and ``files`` + holds what ``write_file`` wrote. ``containers`` echoes the container spec, as the Sandbox API does.""" + + server: FakeServer + + def __init__(self, **kwargs: Any): + self.kwargs = kwargs + self.sandbox_id: str | None = None + self.runner_id = "runner-a" + self.status = SandboxStatus.RUNNING + self.effective_egress = DEFAULT_EGRESS if kwargs.get("network") is None else self._egress(kwargs["network"]) + self.service_endpoints = tuple( + HttpsEndpointStatus(port=service.port, name="", kind=EndpointKind.HTTPS, auth=EndpointAuth.OPEN, + url=f"https://{service.port}-x.r1abc.ep1.cwsandbox.test", request_timeout_seconds=900) + for service in kwargs.get("services") or () + ) + self.containers = tuple(kwargs.get("containers") or ( + (Container(image=kwargs["container_image"], command=kwargs.get("command"), args=kwargs.get("args"), + primary=True),) if kwargs.get("container_image") else ())) + self.results: list[Result | BaseException | FakeProcess] = [] + self.exec_calls: list[dict[str, Any]] = [] + self.processes: list[FakeProcess] = [] + self.files: dict[str, bytes] = {} + self.server.created.append(self) + + @staticmethod + def _egress(network: Any) -> tuple[EgressRule, ...]: + return () if network.deny_egress else tuple(network.egress) + + async def start(self) -> None: + server = self.server + if server.start_delay: + await asyncio.sleep(server.start_delay) + if server.start_error is not None and not server.start_commits: + raise server.start_error + self.sandbox_id = next(server.ids) + server.sandboxes[self.sandbox_id] = self + if server.start_error is not None: + self.sandbox_id = None + raise server.start_error + + def __await__(self): + return _done(self).__await__() + + def exec(self, command: list[str], *, timeout_seconds: float | None = None, stdin: bool = False) -> FakeProcess: + self.exec_calls.append({"argv": list(command), "timeout": timeout_seconds, "stdin": stdin}) + outcome = self.results.pop(0) if self.results else Result() + if isinstance(outcome, FakeProcess): + process = outcome + else: + process = FakeProcess(outcome, stdin=stdin) + self.processes.append(process) + return process + + async def write_file(self, filepath: str, contents: bytes) -> None: + self.files[filepath] = contents + + @classmethod + async def from_id(cls, sandbox_id: str, **kwargs: Any) -> FakeSandbox: + cls.server.calls.append(("from_id", kwargs)) + if sandbox_id not in cls.server.sandboxes: + raise SandboxNotFoundError(f"sandbox {sandbox_id} not found") + return cls.server.sandboxes[sandbox_id] + + @classmethod + async def list(cls, *, tags: list[str], **kwargs: Any) -> list[SimpleNamespace]: + cls.server.listed.append(tags) + return [SimpleNamespace(sandbox_id=sandbox_id) for sandbox_id, sb in cls.server.sandboxes.items() + if set(tags) <= set(sb.kwargs.get("tags") or ())] + + @classmethod + async def delete(cls, sandbox_id: str, *, missing_ok: bool = False, **kwargs: Any) -> None: + cls.server.calls.append(("delete", {"missing_ok": missing_ok, **kwargs})) + cls.server.deleted.append(sandbox_id) + cls.server.sandboxes.pop(sandbox_id, None) + + +def fake_sandbox_cls() -> type[FakeSandbox]: + """A fresh ``FakeSandbox`` class with a server of its own, so no test sees another's sandboxes.""" + return type("FakeSandbox", (FakeSandbox,), {"server": FakeServer()}) diff --git a/tst/unit/providers/sandbox_providers/coreweave_sandbox_provider_test.py b/tst/unit/providers/sandbox_providers/coreweave_sandbox_provider_test.py new file mode 100644 index 000000000..d1b2db3d6 --- /dev/null +++ b/tst/unit/providers/sandbox_providers/coreweave_sandbox_provider_test.py @@ -0,0 +1,551 @@ +"""Offline unit tests for the CoreWeave sandbox provider.""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import cwsandbox +import pytest +from cwsandbox import CidrBlock, EgressRule, EndpointAuth, EndpointKind, SandboxFailedError, SandboxStatus +from cwsandbox import ServiceVisibility + +from agent_env.config.errors import ConfigError +from agent_env.providers.sandbox_providers.coreweave import provider as provider_module +from agent_env.providers.sandbox_providers.coreweave.provider import ( + DEFAULT_IMAGE, + DEFAULT_RUNTIME_CLASS, + ENDPOINT_REQUEST_TIMEOUT_SEC, + MIN_DOCKER_DISK_GB, + SANDBOX_TAG, + CoreWeaveSandboxProvider, + auth_headers, + docker_disk_gb, + resources, + sandbox_tags, +) +from agent_env.providers.sandbox_providers.coreweave.sandbox import CoreWeaveContainerSandbox, CoreWeaveSandbox +from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy, NetworkPolicyUnsupportedError +from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER, SANDBOX_MODE_VM, Accepts +from agent_env.providers.sandbox_providers.sandbox_provider import SandboxNotFoundError, SandboxProvider, _unimplemented +from agent_env.providers.sandbox_providers.sandbox_provider import all_sandbox_egress_hosts +from agent_env.store.image_store.oci_registry_credentials import RegistryAuth +from tst.unit.providers.sandbox_providers.coreweave_fakes import Result, fake_sandbox_cls + +ALLOWLIST = NetworkMode.ALLOWLIST + + +@pytest.fixture(autouse=True) +def _no_endpoint_parents(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(CoreWeaveSandboxProvider, "EGRESS_HOSTS", ()) + + +@pytest.fixture +def sandbox_cls(): + return fake_sandbox_cls() + + +def _provider(sandbox_cls, **kwargs) -> CoreWeaveSandboxProvider: + return CoreWeaveSandboxProvider(api_key="cw_test", sandbox_cls=sandbox_cls, **kwargs) + + +async def _no_sleep(_seconds): + return None + + +def test_the_provider_creates_vms_and_containers_and_implements_what_it_must(): + assert _unimplemented(CoreWeaveSandboxProvider) is None + assert CoreWeaveSandboxProvider.SUPPORTED_SANDBOX_MODES == {SANDBOX_MODE_VM, SANDBOX_MODE_CONTAINER} + + +@pytest.mark.parametrize(("config", "message"), [ + ({}, "exactly one of 'api_key' and 'wandb_api_key'"), + ({"api_key": "a", "wandb_api_key": "w"}, "exactly one of 'api_key' and 'wandb_api_key'"), + ({"api_key": "a", "wandb_entity": "team"}, "only with 'wandb_api_key'"), + ({"api_key": "a", "region": "us"}, "unknown key"), + ({"api_key": "a", "image": ""}, "'image' must be a non-empty string"), + ({"api_key": "a", "docker_storage": "vfs"}, "'docker_storage' must be 'loop' or 'memory'"), + ({"api_key": "a", "runner_ids": "runner-a"}, "'runner_ids' must be a non-empty list"), + ({"api_key": "a", "runner_ids": []}, "'runner_ids' must be a non-empty list"), +]) +def test_from_config_refuses_a_bad_section(config, message): + with pytest.raises(ConfigError, match=message): + CoreWeaveSandboxProvider.from_config(**config) + + +def test_from_config_takes_every_documented_key(sandbox_cls, monkeypatch): + monkeypatch.setattr(provider_module.cwsandbox, "Sandbox", sandbox_cls) + provider = CoreWeaveSandboxProvider.from_config( + wandb_api_key="wb", wandb_entity="team", wandb_project="evals", base_url="https://api.test", + image="registry.test/host:1", docker_storage="memory", runtime_class="kata-clh", placement_mode="cks", + runner_ids=["runner-a"], + ) + assert provider._client.base_url == "https://api.test" + assert (provider._image, provider._docker_storage, provider._runtime_class) == ( + "registry.test/host:1", "memory", "kata-clh") + + +def test_auth_headers_for_a_coreweave_key(): + headers = auth_headers({"api_key": "cw_secret"}) + assert headers.headers == {"Authorization": "Bearer cw_secret"} + + +def test_auth_headers_for_a_wandb_key_carry_its_entity_and_project(): + headers = auth_headers({"wandb_api_key": "wb_secret", "wandb_entity": "team", "wandb_project": "evals"}) + assert headers.headers == {"x-wandb-api-key": "wb_secret", "x-entity-id": "team", "x-project-name": "evals"} + + +def test_resources_request_and_limit_exactly_what_is_asked(): + shape = resources(1.5, 8192) + assert shape.requests == shape.limits == {"cpu": "1500m", "memory": "8192Mi"} + assert resources(0.0001, 1).requests["cpu"] == "1m" + + +@pytest.mark.parametrize(("cpu", "memory"), [(0, 1024), (float("nan"), 1024), (True, 1024), (1, 0), (1, 1.5)]) +def test_resources_refuse_a_nonsense_shape(cpu, memory): + with pytest.raises(ValueError): + resources(cpu, memory) + + +def test_dockers_disk_has_a_floor(): + assert docker_disk_gb(10) == MIN_DOCKER_DISK_GB + assert docker_disk_gb(40.2) == 41 + + +def test_tags_are_slugged_into_coreweaves_alphabet_and_length(): + tags = sandbox_tags({"run_id": "run 1/2", "pipeline_step": "deploy:" + "x" * 80}, "agentenv-abc") + assert tags[:3] == [SANDBOX_TAG, "agentenv-abc", "pipeline_step.deploy-" + "x" * 38] + assert tags[3] == "run_id.run-1-2" + assert all(len(tag) <= 59 for tag in tags) + + +def test_supports_allow_all_and_allowlists_of_valid_entries(): + assert CoreWeaveSandboxProvider.supports_network_policy(NetworkPolicy()) + assert CoreWeaveSandboxProvider.supports_network_policy( + NetworkPolicy(mode=ALLOWLIST, allow_hosts=("a.com", "*.b.com", "10.0.0.1"), allow_cidrs=("1.0.0.0/8",))) + assert not CoreWeaveSandboxProvider.supports_network_policy(NetworkPolicy(mode=ALLOWLIST, allow_hosts=("a b",))) + + +@pytest.mark.asyncio +async def test_create_vm_makes_a_privileged_docker_host_with_https_endpoints(sandbox_cls): + sandbox = await _provider(sandbox_cls).create_vm( + cpu=2, memory=4096, disk_size_gb=30, timeout=900, exposed_ports=[8080, 8080, 9000], + attribution={"run_id": "r1"}, + ) + raw = sandbox_cls.server.created[0] + kwargs = raw.kwargs + assert (kwargs["command"], kwargs["args"][0]) == ("/bin/sh", "-c") + assert "truncate -s 30G" in kwargs["args"][1] + assert kwargs["container_image"] == DEFAULT_IMAGE + assert kwargs["runtime_class"] == DEFAULT_RUNTIME_CLASS + assert kwargs["security_context"].privileged is True + assert kwargs["max_lifetime_seconds"] == 900 + assert kwargs["resources"].limits == {"cpu": "2000m", "memory": "4096Mi"} + assert kwargs["network"] is None + assert kwargs["auth"].headers == {"Authorization": "Bearer cw_test"} + assert kwargs["tags"][0] == SANDBOX_TAG and kwargs["tags"][1].startswith(f"{SANDBOX_TAG}-") + assert "run_id.r1" in kwargs["tags"] + services = kwargs["services"] + assert [service.port for service in services] == [8080, 9000] + assert {(s.visibility, s.endpoint.kind, s.endpoint.auth, s.endpoint.request_timeout_seconds) for s in services} == { + (ServiceVisibility.PUBLIC, EndpointKind.HTTPS, EndpointAuth.OPEN, ENDPOINT_REQUEST_TIMEOUT_SEC)} + assert (sandbox.type, sandbox.mode, sandbox.sandbox_id, sandbox.command_timeout) == ("coreweave", "vm", "sb-1", 900) + assert set(sandbox.tunnel_urls) == {8080, 9000} + assert sandbox.network_policy == NetworkPolicy() + assert raw.exec_calls[0]["argv"][:2] == ["sh", "-c"] # the gateway setup waited for Docker + + +@pytest.mark.asyncio +async def test_create_vm_sends_and_stamps_an_allowlist(sandbox_cls): + policy = NetworkPolicy(mode=ALLOWLIST, allow_hosts=("pypi.org",), allow_cidrs=("1.1.1.0/24",)) + sandbox = await _provider(sandbox_cls).create_vm(network_policy=policy, setup_for_gateway=False) + rules = sandbox_cls.server.created[0].kwargs["network"].egress + assert "pypi.org" in [rule.dns_name for rule in rules] + assert sandbox.network_policy.restricts_egress + assert "pypi.org" in sandbox.network_policy.allow_hosts + assert "1.1.1.0/24" in sandbox.network_policy.allow_cidrs + + +@pytest.mark.asyncio +async def test_create_vm_refuses_a_per_call_image_and_an_unenforceable_policy(sandbox_cls): + provider = _provider(sandbox_cls) + with pytest.raises(ValueError, match="per-call image is unsupported"): + await provider.create_vm(image="other") + with pytest.raises(NetworkPolicyUnsupportedError): + await provider.create_vm(network_policy=NetworkPolicy(mode=ALLOWLIST, allow_hosts=("not valid",))) + assert sandbox_cls.server.created == [] + + +@pytest.mark.asyncio +async def test_create_container_runs_the_image_itself_with_its_env_and_port(sandbox_cls): + sandbox = await _provider(sandbox_cls).create_container( + image_name="registry.example/agent:1", port=8000, env={"A2A_PORT": "8000"}, cpu=2, memory=4096, timeout=900, + attribution={"run_id": "r1"}, + ) + raw = sandbox_cls.server.created[0] + kwargs = raw.kwargs + [container] = kwargs["containers"] + assert (container.image, container.command, container.args) == ("registry.example/agent:1", None, None) + assert container.environment_variables == {"A2A_PORT": "8000"} + assert container.resources.limits == {"cpu": "2000m", "memory": "4096Mi"} + assert not {"command", "args", "container_image", "resources", "security_context", "runtime_class"} & set(kwargs) + assert kwargs["max_lifetime_seconds"] == 900 + assert "run_id.r1" in kwargs["tags"] and kwargs["network"] is None + assert [service.port for service in kwargs["services"]] == [8000] + assert type(sandbox) is CoreWeaveContainerSandbox + assert (sandbox.type, sandbox.mode, sandbox.command_timeout) == ("coreweave", SANDBOX_MODE_CONTAINER, 900) + assert list(sandbox.tunnel_urls) == [8000] and sandbox.network_policy == NetworkPolicy() + assert raw.exec_calls == [] # no Docker host to ready + + +@pytest.mark.asyncio +async def test_the_sdk_sends_a_container_with_no_command_so_the_image_starts_as_it_was_built(sandbox_cls): + """The SDK's single-container arguments fill an unset command with ``/bin/sh -c 'sleep infinity'``.""" + await _provider(sandbox_cls).create_container(image_name="registry.example/agent:1", port=8000, env={"A": "1"}) + real = cwsandbox.Sandbox(**sandbox_cls.server.created[0].kwargs) + request = real._build_create_request(request_id="r", start_kwargs=dict(real._start_kwargs)) + [container] = request.sandbox.spec.containers + assert (container.image, container.command, list(container.args)) == ("registry.example/agent:1", "", []) + assert dict(container.environment_variables) == {"A": "1"} + assert not container.HasField("security_context") + + +@pytest.mark.asyncio +async def test_create_sandbox_is_a_container_of_the_agents_image(sandbox_cls): + sandbox = await _provider(sandbox_cls).create_sandbox(image_name="registry.example/agent:1", port=8000, + env={"A": "1"}) + assert sandbox.mode == SANDBOX_MODE_CONTAINER and list(sandbox.tunnel_urls) == [8000] + assert sandbox_cls.server.created[0].kwargs["containers"][0].image == "registry.example/agent:1" + + +def test_containers_and_agents_run_images_by_name_and_vms_create_gateways(): + """Declared, since a provider implementing create_vm otherwise gives agents what its VMs load.""" + assert CoreWeaveSandboxProvider.SANDBOX_ACCEPTS is CoreWeaveSandboxProvider.CONTAINER_ACCEPTS is Accepts.NAME + assert not CoreWeaveSandboxProvider.PRIVATE_NETWORK + + +@pytest.mark.asyncio +async def test_a_container_logs_its_disk_size_as_ignored(sandbox_cls, caplog): + caplog.set_level("INFO", logger=provider_module.__name__) + await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}, + disk_size_gb=40) + assert "Ignoring disk_size_gb=40" in caplog.text + + +@pytest.mark.asyncio +async def test_a_container_on_a_private_network_is_refused(sandbox_cls): + with pytest.raises(ValueError, match="share no private network"): + await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=1, env={}, + private_network=True) + assert sandbox_cls.server.created == [] + + +@pytest.mark.asyncio +async def test_an_image_the_image_store_holds_credentials_for_is_refused(sandbox_cls, monkeypatch): + store = SimpleNamespace(auth=lambda ref: RegistryAuth(registry="123.dkr.ecr.test", username="AWS", password="t")) + monkeypatch.setattr(provider_module, "get_config", lambda: SimpleNamespace(get_image_store_at=lambda ref: store)) + with pytest.raises(ValueError, match="without the credentials agent-env's image store holds for 123.dkr.ecr.test"): + await _provider(sandbox_cls).create_container(image_name="123.dkr.ecr.test/agent:1", port=1, env={}) + assert sandbox_cls.server.created == [] + + +@pytest.mark.asyncio +async def test_a_container_runs_the_digest_prepared_before_its_create_began(sandbox_cls, monkeypatch): + """Another deploy preparing the same tag at another digest while this create awaits the image store doesn't + change the image this one runs.""" + provider = _provider(sandbox_cls) + name, first, second = "registry.example/a:1", "sha256:" + "a" * 64, "sha256:" + "b" * 64 + await provider.prepare_image(SimpleNamespace(image_name=name, image_digest=first, pinned_name=f"{name}@{first}")) + + def auth(ref): + provider._pinned_images[name] = f"{name}@{second}" + + monkeypatch.setattr(provider_module, "get_config", + lambda: SimpleNamespace(get_image_store_at=lambda ref: SimpleNamespace(auth=auth))) + await provider.create_container(image_name=name, port=8000, env={}) + assert sandbox_cls.server.created[0].kwargs["containers"][0].image == f"{name}@{first}" + + +@pytest.mark.asyncio +async def test_a_container_whose_image_fails_to_start_is_deleted(sandbox_cls, monkeypatch): + async def failing_wait(self): + raise SandboxFailedError("ImagePullBackOff") + + monkeypatch.setattr(sandbox_cls, "__await__", lambda self: failing_wait(self).__await__()) + with pytest.raises(SandboxFailedError, match="ImagePullBackOff"): + await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}) + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_container_without_its_endpoint_is_deleted(sandbox_cls): + original_init = sandbox_cls.__init__ + + def init(self, **kwargs): + original_init(self, **kwargs) + self.service_endpoints = () + + sandbox_cls.__init__ = init + with pytest.raises(RuntimeError, match=r"no endpoint for port\(s\) \[8000\]"): + await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}) + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_failed_start_deletes_the_sandbox_it_made(sandbox_cls, monkeypatch): + monkeypatch.setattr("asyncio.sleep", _no_sleep) + + async def failing_wait(self): + raise SandboxFailedError("image pull failed") + + monkeypatch.setattr(sandbox_cls, "__await__", lambda self: failing_wait(self).__await__()) + with pytest.raises(SandboxFailedError, match="image pull failed"): + await _provider(sandbox_cls).create_vm() + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_start_whose_response_was_lost_is_found_by_its_unique_tag_and_deleted(sandbox_cls): + sandbox_cls.server.start_error = ConnectionError("connection reset") + sandbox_cls.server.start_commits = True + with pytest.raises(ConnectionError): + await _provider(sandbox_cls).create_vm() + create_tag = sandbox_cls.server.created[0].kwargs["tags"][1] + assert sandbox_cls.server.listed == [[create_tag]] + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_cancelled_create_deletes_its_sandbox_once_it_exists(sandbox_cls): + sandbox_cls.server.start_delay = 0.05 + create = asyncio.ensure_future(_provider(sandbox_cls).create_vm()) + await asyncio.sleep(0.01) + create.cancel() + with pytest.raises(asyncio.CancelledError): + await create + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_start_that_answers_after_cleanup_stopped_waiting_has_its_sandbox_deleted_then(sandbox_cls, monkeypatch): + monkeypatch.setattr(provider_module, "_START_SETTLE_SEC", 0.01) + sandbox_cls.server.start_delay = 0.1 + create = asyncio.ensure_future(_provider(sandbox_cls).create_vm()) + await asyncio.sleep(0.02) + create.cancel() + with pytest.raises(asyncio.CancelledError): + await create + assert sandbox_cls.server.deleted == [] + await asyncio.sleep(0.2) + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_start_that_fails_after_cleanup_stopped_waiting_has_the_sandbox_it_made_found_by_its_tag_then( + sandbox_cls, monkeypatch): + monkeypatch.setattr(provider_module, "_START_SETTLE_SEC", 0.01) + sandbox_cls.server.start_delay = 0.1 + sandbox_cls.server.start_error = ConnectionError("connection reset") + sandbox_cls.server.start_commits = True + create = asyncio.ensure_future(_provider(sandbox_cls).create_vm()) + await asyncio.sleep(0.02) + create.cancel() + with pytest.raises(asyncio.CancelledError): + await create + create_tag = sandbox_cls.server.created[0].kwargs["tags"][1] + assert sandbox_cls.server.listed == [[create_tag]] + assert sandbox_cls.server.deleted == [] + await asyncio.sleep(0.2) + assert sandbox_cls.server.listed == [[create_tag], [create_tag]] + assert sandbox_cls.server.deleted == ["sb-1"] + + +def _slow_round_trips(sandbox_cls) -> None: + """List and delete take a network round trip, as the SDK's do.""" + listed, deleted = sandbox_cls.list.__func__, sandbox_cls.delete.__func__ + + async def slow_list(cls, **kwargs): + await asyncio.sleep(0.05) + return await listed(cls, **kwargs) + + async def slow_delete(cls, sandbox_id, **kwargs): + await asyncio.sleep(0.05) + await deleted(cls, sandbox_id, **kwargs) + + sandbox_cls.list, sandbox_cls.delete = classmethod(slow_list), classmethod(slow_delete) + + +@pytest.mark.parametrize("create", ["create_vm", "create_container"]) +def test_a_create_cancelled_by_the_loop_ending_deletes_the_sandbox_its_lost_response_made(sandbox_cls, create): + """``asyncio.run`` cancels what is left when it ends, the SDK's start among it, and closes the loop once those are + done: cleanup must finish inside the cancelled create.""" + + async def start(self): + sandbox_id = next(self.server.ids) + self.server.sandboxes[sandbox_id] = self # the server took the create; its response is still on its way + await asyncio.sleep(1) + self.sandbox_id = sandbox_id + + sandbox_cls.start = start + _slow_round_trips(sandbox_cls) + kwargs = {"setup_for_gateway": False} if create == "create_vm" else { + "image_name": "registry.example/a:1", "port": 8000, "env": {}} + + async def main(): + asyncio.ensure_future(getattr(_provider(sandbox_cls), create)(**kwargs)) + await asyncio.sleep(0.01) + + asyncio.run(main()) + assert sandbox_cls.server.listed == [[sandbox_cls.server.created[0].kwargs["tags"][1]]] + assert sandbox_cls.server.deleted == ["sb-1"] and sandbox_cls.server.sandboxes == {} + + +def test_a_create_cancelled_by_the_loop_ending_while_it_waits_to_run_deletes_its_sandbox(sandbox_cls): + async def slow_running(self): + await asyncio.sleep(1) + return self + + sandbox_cls.__await__ = lambda self: slow_running(self).__await__() + _slow_round_trips(sandbox_cls) + + async def main(): + asyncio.ensure_future(_provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, + env={})) + await asyncio.sleep(0.01) + + asyncio.run(main()) + assert sandbox_cls.server.deleted == ["sb-1"] and sandbox_cls.server.listed == [] + + +@pytest.mark.asyncio +async def test_a_failed_docker_setup_deletes_the_sandbox_and_says_why(sandbox_cls): + original_init = sandbox_cls.__init__ + + def init(self, **kwargs): + original_init(self, **kwargs) + self.results = [Result(returncode=1, stderr_bytes=b"the image lacks bash curl e2fsprogs")] + + sandbox_cls.__init__ = init + with pytest.raises(RuntimeError, match="the image lacks bash curl e2fsprogs"): + await _provider(sandbox_cls).create_vm() + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_a_port_without_an_endpoint_fails_the_create_and_deletes_the_sandbox(sandbox_cls): + original_init = sandbox_cls.__init__ + + def init(self, **kwargs): + original_init(self, **kwargs) + self.service_endpoints = () + + sandbox_cls.__init__ = init + with pytest.raises(RuntimeError, match=r"no endpoint for port\(s\) \[8080\]"): + await _provider(sandbox_cls).create_vm(exposed_ports=[8080], setup_for_gateway=False) + assert sandbox_cls.server.deleted == ["sb-1"] + + +@pytest.mark.asyncio +async def test_exposed_port_domains_join_later_allowlists(sandbox_cls): + """A restricted agent must reach an environment another CoreWeave sandbox serves.""" + provider = _provider(sandbox_cls) + await provider.create_vm(exposed_ports=[8080], setup_for_gateway=False) + agent = await provider.create_vm(network_policy=NetworkPolicy(mode=ALLOWLIST), setup_for_gateway=False) + assert "*.r1abc.ep1.cwsandbox.test" in agent.network_policy.allow_hosts + + +@pytest.mark.asyncio +async def test_exposed_port_domains_join_every_providers_allowlists(sandbox_cls): + """A restricted agent on another provider must reach an environment a CoreWeave sandbox serves.""" + await _provider(sandbox_cls).create_vm(exposed_ports=[8080], setup_for_gateway=False) + assert "*.r1abc.ep1.cwsandbox.test" in all_sandbox_egress_hosts() + elsewhere = SandboxProvider.effective_network_policy(NetworkPolicy(mode=ALLOWLIST)) + assert "*.r1abc.ep1.cwsandbox.test" in elsewhere.allow_hosts + + +@pytest.mark.asyncio +async def test_get_sandbox_restores_endpoints_and_the_applied_policy(sandbox_cls): + provider = _provider(sandbox_cls) + policy = NetworkPolicy(mode=ALLOWLIST, allow_cidrs=("1.1.1.0/24",)) + created = await provider.create_vm(exposed_ports=[8080], network_policy=policy, setup_for_gateway=False) + reconnected = await provider.get_sandbox(created.sandbox_id) + assert reconnected.tunnel_urls == created.tunnel_urls + assert reconnected.network_policy == created.network_policy + assert reconnected.command_timeout > 0 + assert sandbox_cls.server.calls[-1] == ("from_id", {"auth": provider._client.auth, + "base_url": provider._client.base_url}) + + +@pytest.mark.asyncio +async def test_unrepresentable_applied_egress_is_an_unknown_policy_at_create_and_reconnect(sandbox_cls): + """Never the requested policy: callers decide from it whether the sandbox can download for itself.""" + original_init = sandbox_cls.__init__ + + def init(self, **kwargs): + original_init(self, **kwargs) + self.effective_egress = (EgressRule(cidr=CidrBlock(cidr="1.1.1.0/24"), ports=[443]),) + + sandbox_cls.__init__ = init + policy = NetworkPolicy(mode=ALLOWLIST, allow_cidrs=("1.1.1.0/24",)) + vm = await _provider(sandbox_cls).create_vm(network_policy=policy, setup_for_gateway=False) + container = await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}, + network_policy=policy) + fresh = _provider(sandbox_cls) + assert [vm.network_policy, container.network_policy] == [None, None] + assert [(await fresh.get_sandbox(s.sandbox_id)).network_policy for s in (vm, container)] == [None, None] + + +@pytest.mark.asyncio +async def test_a_fresh_provider_reconnects_each_sandbox_in_the_mode_it_was_created_in(sandbox_cls): + policy = NetworkPolicy(mode=ALLOWLIST, allow_cidrs=("1.1.1.0/24",)) + vm = await _provider(sandbox_cls).create_vm(exposed_ports=[8080], network_policy=policy, setup_for_gateway=False) + container = await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}, + network_policy=policy) + fresh = _provider(sandbox_cls) + vm_again, container_again = await fresh.get_sandbox(vm.sandbox_id), await fresh.get_sandbox(container.sandbox_id) + assert (type(vm_again), vm_again.mode) == (CoreWeaveSandbox, SANDBOX_MODE_VM) + assert (type(container_again), container_again.mode) == (CoreWeaveContainerSandbox, SANDBOX_MODE_CONTAINER) + for created, again in ((vm, vm_again), (container, container_again)): + assert (again.tunnel_urls, again.network_policy) == (created.tunnel_urls, created.network_policy) + + +@pytest.mark.asyncio +async def test_a_sandbox_whose_spec_is_not_echoed_is_not_guessed_at(sandbox_cls): + created = await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}) + sandbox_cls.server.sandboxes[created.sandbox_id].containers = () + with pytest.raises(RuntimeError, match="whether it is a container or a VM is unknown"): + await _provider(sandbox_cls).get_sandbox(created.sandbox_id) + + +@pytest.mark.asyncio +async def test_get_sandbox_of_a_gone_sandbox_raises_not_found(sandbox_cls): + with pytest.raises(SandboxNotFoundError, match="sb-gone"): + await _provider(sandbox_cls).get_sandbox("sb-gone") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [SandboxStatus.TERMINATING, SandboxStatus.COMPLETED, SandboxStatus.FAILED, + SandboxStatus.TERMINATED]) +async def test_get_sandbox_of_a_sandbox_that_has_ended_raises_not_found(sandbox_cls, status): + created = await _provider(sandbox_cls).create_container(image_name="registry.example/a:1", port=8000, env={}) + sandbox_cls.server.sandboxes[created.sandbox_id].status = status + with pytest.raises(SandboxNotFoundError, match="has ended"): + await _provider(sandbox_cls).get_sandbox(created.sandbox_id) + + +@pytest.mark.asyncio +async def test_terminate_sandbox_deletes_by_id_without_reconnecting(sandbox_cls): + await _provider(sandbox_cls).terminate_sandbox("sb-gone") + assert sandbox_cls.server.deleted == ["sb-gone"] + assert [name for name, _ in sandbox_cls.server.calls] == ["delete"] + + +@pytest.mark.asyncio +async def test_terminating_a_container_twice_tolerates_it_being_gone(sandbox_cls): + provider = _provider(sandbox_cls) + container = await provider.create_container(image_name="registry.example/a:1", port=8000, env={}) + await container.terminate() + await container.terminate() + await provider.terminate_sandbox(container.sandbox_id) + assert sandbox_cls.server.deleted == ["sb-1"] * 3 + assert all(kwargs["missing_ok"] for name, kwargs in sandbox_cls.server.calls if name == "delete") diff --git a/tst/unit/providers/sandbox_providers/coreweave_sandbox_test.py b/tst/unit/providers/sandbox_providers/coreweave_sandbox_test.py new file mode 100644 index 000000000..743615612 --- /dev/null +++ b/tst/unit/providers/sandbox_providers/coreweave_sandbox_test.py @@ -0,0 +1,364 @@ +"""Offline unit tests for the CoreWeave sandboxes: exec, file transfer, stdin transfer, host setup and egress.""" + +from __future__ import annotations + +import asyncio +import io +from types import SimpleNamespace + +import httpx +import pytest +from cwsandbox import ( + CidrBlock, + Container, + EgressRule, + SandboxError, + SandboxExecutionError, + SandboxNotRunningError, + SandboxRequestTimeoutError, + SandboxStreamTruncatedError, + SandboxTimeoutError, + SandboxUnavailableError, + TenantScope, +) + +from agent_env.providers.sandbox_providers.coreweave import sandbox as sandbox_module +from agent_env.providers.sandbox_providers.coreweave.sandbox import ( + Client, + CoreWeaveContainerSandbox, + CoreWeaveSandbox, + allows_download, + host_setup_script, + network_options, + policy_from_egress, + runs_docker_host, +) +from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy, stage_files_into_container +from tst.unit.providers.sandbox_providers.coreweave_fakes import FakeProcess, Result, fake_sandbox_cls + +ALLOWLIST = NetworkMode.ALLOWLIST + + +def _sandbox(*, policy=NetworkPolicy()) -> tuple[CoreWeaveSandbox, object]: + cls = fake_sandbox_cls() + raw = cls() + raw.sandbox_id = "sb-1" + cls.server.sandboxes["sb-1"] = raw + client = Client(sandbox_cls=cls, auth="auth", base_url="https://api.test") + return CoreWeaveSandbox(raw, client=client, tunnel_urls={}, network_policy=policy, command_timeout=600), raw + + +def _container() -> tuple[CoreWeaveContainerSandbox, object]: + cls = fake_sandbox_cls() + raw = cls(containers=[Container(image="registry.example/agent:1")]) + raw.sandbox_id = "sb-1" + cls.server.sandboxes["sb-1"] = raw + client = Client(sandbox_cls=cls, auth="auth", base_url="https://api.test") + return CoreWeaveContainerSandbox(raw, client=client, tunnel_urls={}, network_policy=None, command_timeout=600), raw + + +async def _no_sleep(_seconds): + return None + + +@pytest.mark.asyncio +async def test_exec_runs_with_the_lifetime_budget_and_drops_sudo(): + sandbox, raw = _sandbox() + raw.results = [Result(stdout_bytes=b"hi\n")] + assert await sandbox.exec_with_output("sudo", "echo", "hi") == (0, "hi\n", "") + assert raw.exec_calls == [{"argv": ["echo", "hi"], "timeout": 600, "stdin": False}] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [ + SandboxUnavailableError("unavailable"), + SandboxStreamTruncatedError("output lost in transit"), + SandboxRequestTimeoutError("request timed out"), + SandboxExecutionError("Command cat /big ended without an exit status from the sandbox; it may not have finished"), +]) +async def test_a_lost_or_cut_stream_is_exit_minus_one_with_no_stdout(error): + sandbox, raw = _sandbox() + raw.results = [error] + exit_code, stdout, stderr = await sandbox.exec_with_output("cat", "/big") + assert (exit_code, stdout) == (-1, "") + assert type(error).__name__ in stderr + + +@pytest.mark.asyncio +async def test_a_timed_out_command_is_exit_124(): + sandbox, raw = _sandbox() + raw.results = [SandboxTimeoutError("deadline exceeded")] + exit_code, _, stderr = await sandbox.exec_with_output("sleep", "9999") + assert exit_code == 124 and "timed out after 600s" in stderr + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [ + SandboxNotRunningError("gone"), + SandboxError("rejected"), + SandboxExecutionError("Command true failed with exit code 1", exec_result=object()), +]) +async def test_other_sdk_errors_propagate(error): + sandbox, raw = _sandbox() + raw.results = [error] + with pytest.raises(type(error)): + await sandbox.exec_with_output("true") + + +@pytest.mark.asyncio +async def test_exec_script_retries_a_lost_stream(monkeypatch): + monkeypatch.setattr("asyncio.sleep", _no_sleep) + sandbox, raw = _sandbox() + raw.results = [SandboxUnavailableError("unavailable"), Result(stdout_bytes=b"ok")] + assert await sandbox.exec_script("echo ok", max_retries=1) == "ok" + assert [call["argv"] for call in raw.exec_calls] == [["bash", "-c", "echo ok"]] * 2 + + +@pytest.mark.asyncio +async def test_exec_with_stdin_streams_every_piece_then_closes(): + sandbox, raw = _sandbox() + + async def pieces(): + yield b"a" + yield b"b" + + raw.results = [Result(stdout_bytes=b"digest -\n")] + assert await sandbox._exec_with_stdin("sha256sum", pieces()) == (0, "digest -\n", "") + process = raw.processes[0] + assert raw.exec_calls[0] == {"argv": ["bash", "-c", "sha256sum"], "timeout": 600, "stdin": True} + assert (process.stdin.pieces, process.stdin.closed) == ([b"a", b"b"], True) + + +@pytest.mark.asyncio +async def test_stdin_stops_once_the_command_ended_rather_than_wait_on_a_full_queue(): + """A command that exits before reading all its input leaves the SDK's bounded queue full; the next write would + wait forever.""" + sandbox, raw = _sandbox() + raw.results = [FakeProcess(Result(returncode=1, stderr_bytes=b"disk full"), stdin=True, read_limit=1)] + read = [] + + async def pieces(): + for piece in (b"a", b"b", b"c"): + read.append(piece) + yield piece + + process = raw.results[0] + asyncio.get_running_loop().call_later(0.05, process.finish) + assert await asyncio.wait_for(sandbox._exec_with_stdin("cat > /x", pieces()), 5) == (1, "", "disk full") + assert read == [b"a", b"b"] + + +@pytest.mark.asyncio +async def test_a_failing_reader_closes_stdin_and_raises_its_error(): + sandbox, raw = _sandbox() + + async def pieces(): + yield b"a" + raise RuntimeError("the object changed during the push") + + with pytest.raises(RuntimeError, match="changed during the push"): + await sandbox._exec_with_stdin("cat > /x", pieces()) + assert raw.processes[0].stdin.closed + + +@pytest.mark.asyncio +async def test_write_host_file_goes_over_stdin_to_any_path(): + sandbox, raw = _sandbox() + await sandbox.write_host_file(b"payload", "/opt/x/file.bin") + assert raw.exec_calls[-1]["argv"] == ["bash", "-c", "cat > /opt/x/file.bin"] + assert raw.processes[-1].stdin.pieces == [b"payload"] + + +@pytest.mark.asyncio +async def test_terminate_deletes_by_id_and_tolerates_a_gone_sandbox(): + sandbox, raw = _sandbox() + await sandbox.terminate() + await sandbox.terminate() + assert raw.server.deleted == ["sb-1", "sb-1"] + assert {call[1]["missing_ok"] for call in raw.server.calls if call[0] == "delete"} == {True} + + +@pytest.mark.asyncio +async def test_wait_for_vm_waits_with_sh_since_bash_may_not_be_installed_yet(): + sandbox, raw = _sandbox() + await sandbox.wait_for_vm() + assert raw.exec_calls[0]["argv"][:2] == ["sh", "-c"] + + +@pytest.mark.asyncio +async def test_wait_for_vm_reports_what_the_host_setup_logged(): + sandbox, raw = _sandbox() + raw.results = [Result(returncode=1, stderr_bytes=b"the image lacks bash curl e2fsprogs")] + with pytest.raises(RuntimeError, match="Docker not ready in CoreWeave sandbox sb-1: the image lacks bash"): + await sandbox.wait_for_vm() + + +@pytest.mark.asyncio +async def test_wait_for_vm_rides_out_a_lost_stream(): + sandbox, raw = _sandbox() + raw.results = [SandboxUnavailableError("unavailable"), Result()] + await sandbox.wait_for_vm() + assert len(raw.exec_calls) == 2 + + +@pytest.mark.parametrize(("storage", "present", "absent"), [ + ("loop", ["apk add --no-cache bash curl e2fsprogs", "mount -o loop", "truncate -s 30G"], ["tmpfs"]), + ("memory", ["apk add --no-cache bash curl", "mount -t tmpfs -o size=30g"], ["e2fsprogs", "mount -o loop"]), +]) +def test_host_setup_mounts_dockers_storage_off_the_virtiofs_root(storage, present, absent): + script = host_setup_script(storage, 30) + assert all(text in script for text in present) + assert not any(text in script for text in absent) + assert "export DOCKER_HOST=unix:///var/run/docker.sock" in script + + +def test_network_options_for_allow_all_leaves_the_runner_default(): + assert network_options(NetworkPolicy()) is None + + +def test_network_options_for_an_empty_allowlist_denies_all_egress(): + assert network_options(NetworkPolicy(mode=ALLOWLIST)).deny_egress is True + + +def test_network_options_split_hostnames_from_addresses_and_ranges(): + options = network_options(NetworkPolicy(mode=ALLOWLIST, allow_hosts=("pypi.org", "*.vercel.run", "10.1.2.3"), + allow_cidrs=("1.1.1.0/24",))) + assert [(rule.dns_name, rule.cidr.cidr if rule.cidr else None) for rule in options.egress] == [ + ("pypi.org", None), ("*.vercel.run", None), (None, "10.1.2.3/32"), (None, "1.1.1.0/24"), + ] + + +@pytest.mark.parametrize("policy", [ + NetworkPolicy(mode=ALLOWLIST, allow_hosts=("not a host",)), + NetworkPolicy(mode=ALLOWLIST, allow_hosts=("a.*.example.com",)), + NetworkPolicy(mode=ALLOWLIST, allow_cidrs=("not-a-cidr",)), +]) +def test_network_options_refuse_an_entry_coreweave_cannot_take(policy): + with pytest.raises(ValueError): + network_options(policy) + + +@pytest.mark.parametrize(("rules", "policy"), [ + ((EgressRule(any=True),), NetworkPolicy()), + ((EgressRule(cidr=CidrBlock(cidr="0.0.0.0/0", except_cidrs=("10.0.0.0/8",))),), NetworkPolicy()), + ((), NetworkPolicy(mode=ALLOWLIST)), + ((EgressRule(dns_name="pypi.org"), EgressRule(cidr="1.1.1.0/24")), + NetworkPolicy(mode=ALLOWLIST, allow_hosts=("pypi.org",), allow_cidrs=("1.1.1.0/24",))), + ((EgressRule(tenant=TenantScope.SAME_ORG),), None), + ((EgressRule(cidr="1.1.1.0/24", ports=[443]),), None), + ((EgressRule(cidr=CidrBlock(cidr="1.0.0.0/8", except_cidrs=("1.1.0.0/16",))),), None), +]) +def test_policy_from_egress(rules, policy): + assert policy_from_egress(rules) == policy + + +def test_policy_from_egress_reads_back_what_network_options_sends(): + policy = NetworkPolicy(mode=ALLOWLIST, allow_hosts=("pypi.org",), allow_cidrs=("1.1.1.0/24",)) + assert policy_from_egress(network_options(policy).egress) == policy + + +@pytest.mark.parametrize(("policy", "url", "allowed"), [ + (NetworkPolicy(), "http://anything:8080/x", True), + (None, "https://bucket.s3.amazonaws.com/o", False), + (NetworkPolicy(), None, False), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("bucket.s3.amazonaws.com",)), "https://bucket.s3.amazonaws.com/o", + True), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("*.s3.amazonaws.com",)), "https://bucket.s3.amazonaws.com/o", True), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("*.amazonaws.com",)), "https://bucket.s3.amazonaws.com/o", False), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("store.test",)), "http://store.test/o", False), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("store.test",)), "https://store.test:8443/o", False), + (NetworkPolicy(mode=ALLOWLIST, allow_cidrs=("203.0.113.0/24",)), "http://203.0.113.7:9000/o", True), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("203.0.113.7",)), "http://203.0.113.7:9000/o", True), + (NetworkPolicy(mode=ALLOWLIST, allow_cidrs=("203.0.113.0/24",)), "http://198.51.100.1/o", False), +]) +def test_allows_download_follows_coreweaves_rules(policy, url, allowed): + """A hostname grant is HTTPS on 443 and a wildcard covers one label; a CIDR grant covers every port.""" + assert allows_download(policy, url) is allowed + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("policy", "signed", "curled"), [ + (NetworkPolicy(), "https://bucket.s3.amazonaws.com/obj?sig", True), + (NetworkPolicy(mode=ALLOWLIST, allow_hosts=("pypi.org",)), "https://bucket.s3.amazonaws.com/obj?sig", False), + (None, "https://bucket.s3.amazonaws.com/obj?sig", False), + (NetworkPolicy(), None, False), +]) +async def test_a_download_the_sandbox_cannot_fetch_goes_over_stdin(monkeypatch, policy, signed, curled): + store = SimpleNamespace(signed_get_url=lambda _url: signed) + monkeypatch.setattr(sandbox_module, "get_config", lambda: SimpleNamespace(get_object_store_at=lambda _url: store)) + pushed = [] + + async def push(sandbox, object_store, object_url, vm_path): + pushed.append(vm_path) + + monkeypatch.setattr(sandbox_module, "push_object_over_stdin", push) + sandbox, raw = _sandbox(policy=policy) + await sandbox._download_object_to_vm("s3://b/obj", "/tmp/obj") + assert any("curl" in " ".join(call["argv"]) for call in raw.exec_calls) is curled + assert bool(pushed) is not curled + + +@pytest.mark.asyncio +async def test_image_tarballs_the_policy_blocks_are_pushed_rather_than_downloaded(monkeypatch): + sandbox, _ = _sandbox(policy=NetworkPolicy(mode=ALLOWLIST, allow_hosts=("ok.test",))) + signed = ["https://ok.test/a.tar.gz", "https://blocked.test/b.tar.gz"] + loaded = [] + + async def signed_urls(artifacts): + return signed + + async def load(artifacts, urls): + loaded.append(urls) + + monkeypatch.setattr(sandbox, "_signed_image_urls", signed_urls) + monkeypatch.setattr(sandbox, "_load_docker_images", load) + await sandbox._load_tarballs([object(), object()]) + assert loaded == [["https://ok.test/a.tar.gz", None]] + + +@pytest.mark.asyncio +async def test_a_container_runs_commands_as_given_in_its_own_container(): + sandbox, raw = _container() + raw.results = [Result(stdout_bytes=b"hi\n"), SandboxUnavailableError("unavailable")] + assert await sandbox.exec_with_output("sudo", "echo", "hi") == (0, "hi\n", "") + process = await sandbox.exec("cat", "/big") + assert (await process.wait(), await process.stdout.read()) == (-1, b"") + assert [call["argv"] for call in raw.exec_calls] == [["sudo", "echo", "hi"], ["cat", "/big"]] + assert {call["timeout"] for call in raw.exec_calls} == {600} + + +@pytest.mark.asyncio +async def test_a_container_takes_files_over_the_sandbox_api_rather_than_tools_its_image_may_lack(monkeypatch): + store = SimpleNamespace(open=lambda url: io.BytesIO(b"object bytes")) + monkeypatch.setattr(sandbox_module, "get_config", lambda: SimpleNamespace(get_object_store_at=lambda url: store)) + transport = httpx.MockTransport(lambda request: httpx.Response(200, content=b"url bytes")) + real_client = httpx.AsyncClient + monkeypatch.setattr(sandbox_module.httpx, "AsyncClient", lambda **kwargs: real_client(transport=transport, **kwargs)) + sandbox, raw = _container() + await sandbox.write_file_from_text("text", "/app/a.txt") + await sandbox.write_file_from_object("s3://b/obj", "/app/b.bin") + await sandbox.write_file_from_url("https://files.example/c", "/app/c.bin") + assert raw.files == {"/app/a.txt": b"text", "/app/b.bin": b"object bytes", "/app/c.bin": b"url bytes"} + assert raw.exec_calls == [] + + +@pytest.mark.asyncio +async def test_files_staged_into_a_container_are_written_into_it_directly(monkeypatch): + store = SimpleNamespace(open=lambda url: io.BytesIO(url.encode())) + monkeypatch.setattr(sandbox_module, "get_config", lambda: SimpleNamespace(get_object_store_at=lambda url: store)) + sandbox, raw = _container() + loaded = await stage_files_into_container(sandbox, {"d/x.txt": SimpleNamespace(object_url="s3://b/x")}, "/work") + assert loaded == {"d/x.txt": "/work/d/x.txt"} + assert raw.files == {"/work/d/x.txt": b"s3://b/x"} + assert all(call["argv"][:2] == ["mkdir", "-p"] for call in raw.exec_calls) + + +@pytest.mark.parametrize(("containers", "docker_host"), [ + ((Container(image="docker:dind", command="/bin/sh", args=("-c", host_setup_script("loop", 20)), primary=True),), + True), + ((Container(image="registry.example/agent:1", primary=True),), False), + ((Container(image="registry.example/agent:1", command="/bin/sh", args=("-c", "serve")),), False), + ((), None), +]) +def test_a_docker_host_is_told_by_the_setup_script_it_was_started_with(containers, docker_host): + assert runs_docker_host(SimpleNamespace(containers=containers)) is docker_host diff --git a/tst/unit/providers/sandbox_providers/local_sandbox_host_ports_test.py b/tst/unit/providers/sandbox_providers/local_sandbox_host_ports_test.py index 240aa2941..9016cc6d0 100644 --- a/tst/unit/providers/sandbox_providers/local_sandbox_host_ports_test.py +++ b/tst/unit/providers/sandbox_providers/local_sandbox_host_ports_test.py @@ -32,6 +32,7 @@ from agent_env.providers.sandbox_providers.modal_vm_sandbox import ModalVmSandbox from agent_env.providers.sandbox_providers.sandbox import port_bindings from agent_env.providers.sandbox_providers.tenki.sandbox import TenkiSandbox +from agent_env.providers.sandbox_providers.coreweave.sandbox import CoreWeaveSandbox from agent_env.providers.sandbox_providers.tensorlake.sandbox import TensorlakeSandbox RESERVED = [AGENT_ENV_GATEWAY_MCP_PORT, DB_WEB_PORT, DB_MCP_PORT] @@ -56,7 +57,7 @@ def test_base_sandbox_host_port_is_identity(): @pytest.mark.parametrize("sandbox_class", [ModalVmSandbox, ModalSandbox, E2BSandbox, TensorlakeSandbox, TenkiSandbox, - BoxdSandbox, DaytonaSandbox]) + BoxdSandbox, CoreWeaveSandbox, DaytonaSandbox]) def test_remote_sandboxes_publish_on_every_interface(sandbox_class): """Their tunnels reach a published port from outside the VM, so it stays on every interface.""" assert port_bindings(sandbox_class.host_ips, 18765, 18765) == ["18765:18765"] diff --git a/tst/unit/providers/sandbox_providers/network_policy_stamping_test.py b/tst/unit/providers/sandbox_providers/network_policy_stamping_test.py index f609552e2..616ea88f3 100644 --- a/tst/unit/providers/sandbox_providers/network_policy_stamping_test.py +++ b/tst/unit/providers/sandbox_providers/network_policy_stamping_test.py @@ -15,6 +15,7 @@ from agent_env.providers.sandbox_providers.boat.provider import BoatSandboxProvider from agent_env.providers.sandbox_providers.boxd.provider import BoxdSandboxProvider +from agent_env.providers.sandbox_providers.coreweave.provider import CoreWeaveSandboxProvider from agent_env.providers.sandbox_providers.daytona.config import Settings as DaytonaSettings from agent_env.providers.sandbox_providers.daytona.provider import DaytonaSandboxProvider from agent_env.providers.sandbox_providers.e2b.provider import E2BSandboxProvider @@ -28,6 +29,7 @@ from agent_env.providers.sandbox_providers.tenki.provider import TenkiSandboxProvider from agent_env.providers.sandbox_providers.tensorlake.provider import TensorlakeSandboxProvider from agent_env.providers.sandbox_providers.vercel.provider import VercelSandboxProvider +from tst.unit.providers.sandbox_providers.coreweave_fakes import fake_sandbox_cls def _modal_provider(cls): @@ -136,6 +138,11 @@ async def _boxd() -> Sandbox: return await provider.create_vm(exposed_ports=[], setup_for_gateway=False) +async def _coreweave() -> Sandbox: + provider = CoreWeaveSandboxProvider(api_key="test-key", sandbox_cls=fake_sandbox_cls()) + return await provider.create_vm(exposed_ports=[], setup_for_gateway=False) + + async def _daytona() -> Sandbox: client = MagicMock() client.create = AsyncMock(return_value=MagicMock(id="daytona-test")) @@ -173,6 +180,7 @@ async def _smol_vm() -> Sandbox: "vercel": _vercel, "tenki": _tenki, "boxd": _boxd, + "coreweave": _coreweave, "daytona": _daytona, "boat": _boat, "smol_vm": _smol_vm, diff --git a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py index 0c81f3195..b18058004 100644 --- a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py +++ b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py @@ -16,6 +16,8 @@ from agent_env.providers.sandbox_providers.boxd.provider import BoxdSandboxProvider from agent_env.providers.sandbox_providers.boxd.sandbox import BoxdSandbox from agent_env.providers.sandbox_providers.chained_sandbox_provider import ChainedSandboxProvider +from agent_env.providers.sandbox_providers.coreweave.provider import CoreWeaveSandboxProvider +from agent_env.providers.sandbox_providers.coreweave.sandbox import CoreWeaveSandbox from agent_env.providers.sandbox_providers.daytona.provider import DaytonaSandboxProvider from agent_env.providers.sandbox_providers.daytona.sandbox import DaytonaSandbox from agent_env.providers.sandbox_providers.e2b.provider import E2BSandboxProvider @@ -44,7 +46,7 @@ LOADABLE, NAME, NAME_OR_CONTEXT = Accepts.LOADABLE, Accepts.NAME, Accepts.NAME_OR_CONTEXT PROVIDER_CLASSES = [LocalSandboxProvider, ModalSandboxProvider, ModalVmSandboxProvider, E2BSandboxProvider, SailVmSandboxProvider, TensorlakeSandboxProvider, TenkiSandboxProvider, BoxdSandboxProvider, - DaytonaSandboxProvider, BoatSandboxProvider, ChainedSandboxProvider] + CoreWeaveSandboxProvider, DaytonaSandboxProvider, BoatSandboxProvider, ChainedSandboxProvider] @pytest.mark.parametrize("cls, on_this_machine, creates_vms, sandbox, container, private_network", [ @@ -56,9 +58,10 @@ (TensorlakeSandboxProvider, False, True, LOADABLE, NAME, False), (TenkiSandboxProvider, False, True, LOADABLE, NAME, False), (BoxdSandboxProvider, False, True, LOADABLE, NAME, False), + (CoreWeaveSandboxProvider, False, True, NAME, NAME, False), (DaytonaSandboxProvider, False, True, LOADABLE, NAME, False), (BoatSandboxProvider, False, True, LOADABLE, NAME, False), -], ids=["local", "modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "boxd", "daytona", "boat"]) +], ids=["local", "modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "boxd", "coreweave", "daytona", "boat"]) def test_each_provider_declares_where_it_runs_and_the_images_it_runs(cls, on_this_machine, creates_vms, sandbox, container, private_network): assert ((cls.ON_THIS_MACHINE, SANDBOX_MODE_VM in cls.SUPPORTED_SANDBOX_MODES, cls.PRIVATE_NETWORK) @@ -280,7 +283,7 @@ def test_an_image_that_records_no_platform_isnt_checked(): @pytest.mark.parametrize("sandbox", [Sandbox, ModalSandbox, ModalVmSandbox, E2BSandbox, SailVmSandbox, TensorlakeSandbox, - TenkiSandbox, BoxdSandbox, DaytonaSandbox, BoatSandbox]) + TenkiSandbox, BoxdSandbox, CoreWeaveSandbox, DaytonaSandbox, BoatSandbox]) def test_a_sandbox_runs_linux_amd64_unless_its_provider_sets_another(sandbox): assert sandbox.platform == "linux/amd64" diff --git a/tst/unit/providers/sandbox_providers/test_config_sandbox_providers.py b/tst/unit/providers/sandbox_providers/test_config_sandbox_providers.py index 84848e8bb..5431ccac2 100644 --- a/tst/unit/providers/sandbox_providers/test_config_sandbox_providers.py +++ b/tst/unit/providers/sandbox_providers/test_config_sandbox_providers.py @@ -10,6 +10,7 @@ from agent_env.providers.sandbox_providers import sandbox_provider from agent_env.config import reset_config from agent_env.providers.sandbox_providers.boxd.provider import BoxdSandboxProvider +from agent_env.providers.sandbox_providers.coreweave.provider import CoreWeaveSandboxProvider from agent_env.providers.sandbox_providers.daytona.config import Settings as DaytonaSettings from agent_env.providers.sandbox_providers.daytona.provider import DaytonaSandboxProvider from agent_env.providers.sandbox_providers.e2b.provider import E2BSandboxProvider @@ -345,6 +346,34 @@ def test_boxd_missing_api_key_is_a_config_error(monkeypatch, tmp_path): build_sandbox_provider("boxd") +def test_coreweave_builtin_receives_an_interpolated_key(monkeypatch, tmp_path): + cfg = _write_config(tmp_path, """ + [sandbox.providers.coreweave.config] + wandb_api_key = "env:CW_TEST_WANDB_API_KEY" + wandb_entity = "team" + docker_storage = "memory" + """) + monkeypatch.setenv("AGENT_ENV_CONFIG", str(cfg)) + monkeypatch.setenv("CW_TEST_WANDB_API_KEY", "resolved-test-key") + + provider = build_sandbox_provider("coreweave") + + assert isinstance(provider, CoreWeaveSandboxProvider) + assert provider._client.auth.headers == {"x-wandb-api-key": "resolved-test-key", "x-entity-id": "team"} + assert provider._docker_storage == "memory" + + +def test_coreweave_without_a_key_is_a_config_error(monkeypatch, tmp_path): + cfg = _write_config(tmp_path, """ + [sandbox.providers.coreweave.config] + image = "registry.test/host:1" + """) + monkeypatch.setenv("AGENT_ENV_CONFIG", str(cfg)) + + with pytest.raises(ConfigError, match="requires exactly one of 'api_key' and 'wandb_api_key'"): + build_sandbox_provider("coreweave") + + def test_daytona_builtin_receives_interpolated_key_and_settings(monkeypatch, tmp_path): cfg = _write_config(tmp_path, """ [sandbox.providers.daytona.config] diff --git a/tst/unit/test_capabilities.py b/tst/unit/test_capabilities.py index 7fb529a3e..44ad7b3d9 100644 --- a/tst/unit/test_capabilities.py +++ b/tst/unit/test_capabilities.py @@ -151,7 +151,7 @@ def build(spec): monkeypatch.setattr("agent_env.providers.sandbox_providers.sandbox_provider.build_sandbox_provider", build) -_CONFIGURED_PROVIDERS = pytest.mark.parametrize("provider", ["e2b", "sail_vm", "tensorlake", "tenki", "boxd", "daytona"]) +_CONFIGURED_PROVIDERS = pytest.mark.parametrize("provider", ["e2b", "sail_vm", "tensorlake", "tenki", "boxd", "coreweave", "daytona"]) @_CONFIGURED_PROVIDERS diff --git a/tst/unit/test_optional_extras_stay_optional.py b/tst/unit/test_optional_extras_stay_optional.py index 17c8a6726..b981a4935 100644 --- a/tst/unit/test_optional_extras_stay_optional.py +++ b/tst/unit/test_optional_extras_stay_optional.py @@ -217,6 +217,11 @@ def test_core_and_every_store_module_import_without_any_extra(): "agent_env.providers.sandbox_providers.boxd.provider:BoxdSandboxProvider", "agent_env.providers.sandbox_providers.sandbox_provider:SandboxProvider", ), + ( + "coreweave", + "agent_env.providers.sandbox_providers.coreweave.provider:CoreWeaveSandboxProvider", + "agent_env.providers.sandbox_providers.sandbox_provider:SandboxProvider", + ), ( "daytona", "agent_env.providers.sandbox_providers.daytona.provider:DaytonaSandboxProvider", @@ -228,7 +233,7 @@ def test_core_and_every_store_module_import_without_any_extra(): ), ], ids=["gcs", "secret-manager", "s3", "s3-reexport", "secrets-manager-reexport", "dynamodb", "dynamodb-reexport", "tensorlake", - "tenki", "boxd", "daytona", "smol"], + "tenki", "boxd", "coreweave", "daytona", "smol"], ) def test_an_impl_without_its_extra_names_the_extra_to_install(extra, impl, base): message = _run(_LOAD_IMPL, _only_in(extra), impl, base) diff --git a/tst/util/capabilities.py b/tst/util/capabilities.py index 9a663a86c..cff41070e 100644 --- a/tst/util/capabilities.py +++ b/tst/util/capabilities.py @@ -21,8 +21,8 @@ MODEL_ENDPOINT = "model_endpoint_configured" #: The remote sandbox providers (``modal``, ``modal_vm``, ``e2b``, ``sail_vm``, ``tensorlake``, ``tenki``, -#: ``vercel``, ``boxd``, ``daytona``, ``boat``) need credentials the resolved config may not carry; the local default -#: never does. +#: ``vercel``, ``boxd``, ``coreweave``, ``daytona``, ``boat``) need credentials the resolved config may not carry; the +#: local default never does. REMOTE_SANDBOX = "remote_sandbox" #: The A2A agent a ``deploy_agent`` step without an id resolves to (``default_a2a_agent_id``); the @@ -86,17 +86,17 @@ def skip_without_model_endpoint() -> pytest.MarkDecorator: def remote_sandbox_is_available(provider: str) -> bool: """Whether the resolved config can build the ``modal`` / ``modal_vm`` / ``e2b`` / ``sail_vm`` / - ``tensorlake`` / ``tenki`` / ``vercel`` / ``boxd`` / ``daytona`` / ``boat`` sandbox provider, credentials included. + ``tensorlake`` / ``tenki`` / ``vercel`` / ``boxd`` / ``coreweave`` / ``daytona`` / ``boat`` sandbox provider, credentials included. Modal: any failure to resolve the credentials answers False. E2B, Sail, Tensorlake, Tenki, Vercel, boxd, - Daytona and boat: building the provider resolves ``[sandbox.providers..config]`` and its ``secret:`` references; + CoreWeave, Daytona and boat: building the provider resolves ``[sandbox.providers..config]`` and its ``secret:`` references; only a ``ConfigError`` (absent or incomplete config, or the provider's extra not installed) answers False, any other failure answers True and lets the test fail on the real problem, as ``model_endpoint_is_configured`` does. Vercel, which builds without credentials, also needs them in its config or the environment.""" - if provider not in ("modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "daytona", "boat"): + if provider not in ("modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "coreweave", "daytona", "boat"): raise ValueError(f"unknown remote sandbox provider {provider!r}") - if provider in ("e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "daytona", "boat"): + if provider in ("e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "coreweave", "daytona", "boat"): from agent_env.config.errors import ConfigError from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider diff --git a/uv.lock b/uv.lock index 6bf2e58c4..6d0805ede 100644 --- a/uv.lock +++ b/uv.lock @@ -83,6 +83,9 @@ boat = [ boxd = [ { name = "boxd" }, ] +coreweave = [ + { name = "cwsandbox" }, +] daytona = [ { name = "daytona" }, ] @@ -95,6 +98,7 @@ dev = [ { name = "boto3" }, { name = "botocore" }, { name = "boxd" }, + { name = "cwsandbox" }, { name = "datasets" }, { name = "daytona" }, { name = "fastmcp" }, @@ -165,7 +169,7 @@ vercel = [ [package.metadata] requires-dist = [ { name = "a2a-sdk", specifier = "==0.3.26" }, - { name = "agentenv-framework", extras = ["aws", "boat", "boxd", "daytona", "gcp", "openenv", "sail", "tensorlake", "tenki", "trl", "vercel"], marker = "extra == 'dev'" }, + { name = "agentenv-framework", extras = ["aws", "boat", "boxd", "coreweave", "daytona", "gcp", "openenv", "sail", "tensorlake", "tenki", "trl", "vercel"], marker = "extra == 'dev'" }, { name = "agentenv-framework", extras = ["openenv"], marker = "extra == 'nemo-gym'" }, { name = "agentenv-framework", extras = ["openenv"], marker = "extra == 'trl'" }, { name = "agentenv-framework-protocol", extras = ["computer"], editable = "packages/agentenv-protocol" }, @@ -178,6 +182,7 @@ requires-dist = [ { name = "certifi", specifier = ">=2024.2.2" }, { name = "click", specifier = ">=8.0.0" }, { name = "cryptography", specifier = ">=42" }, + { name = "cwsandbox", marker = "extra == 'coreweave'", specifier = ">=1.19.6,<2" }, { name = "datasets", marker = "extra == 'trl'", specifier = ">=3" }, { name = "daytona", marker = "extra == 'daytona'", specifier = "~=0.224.0" }, { name = "e2b", specifier = "~=2.46.4" }, @@ -223,7 +228,7 @@ requires-dist = [ { name = "uvicorn", specifier = ">=0.47" }, { name = "vercel-sandbox", marker = "extra == 'vercel'", specifier = ">=0.7.0,<0.8" }, ] -provides-extras = ["openenv", "trl", "nemo-gym", "aws", "gcp", "sail", "vercel", "tensorlake", "tenki", "boxd", "daytona", "boat", "dev", "smol"] +provides-extras = ["openenv", "trl", "nemo-gym", "aws", "gcp", "sail", "vercel", "tensorlake", "tenki", "boxd", "daytona", "boat", "coreweave", "dev", "smol"] [[package]] name = "agentenv-framework-protocol" @@ -1441,6 +1446,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b5/96/b3b348954f88915e60e2f1fbe6d09250584e53a20d949e55786f31dbaeb0/cwevals-0.0.1b1-py3-none-any.whl", hash = "sha256:6936baf0ccdc5d9092abd5c2ab628cedcdae9bb72edfdf730f7aaac19118855a", size = 176544, upload-time = "2026-09-28T18:21:21.957Z" }, ] +[[package]] +name = "cwsandbox" +version = "1.19.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography" }, + { name = "googleapis-common-protos" }, + { name = "grpcio" }, + { name = "protobuf" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/74/64/e771f3767a3812b1f28809e9c96ba22062fb569932e562e6d266b13ecb5c/cwsandbox-1.19.6.tar.gz", hash = "sha256:cab64a199672c28217ddcf1be26703020d50cd4fc1eb339d8ef7fa2e30625160", size = 671647, upload-time = "2026-10-23T19:03:14.869Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/08/ed/1d985250537b72c3d14da94ffaaf866ac78ae694fb5e65014f4a38ee659e/cwsandbox-1.19.6-py3-none-any.whl", hash = "sha256:b308fafbfa35cb95177be1a60c9bc7185ab2afad531c68957b1b5b2b4a42135e", size = 261122, upload-time = "2026-10-23T19:03:13.108Z" }, +] + [[package]] name = "cycler" version = "0.12.1"