From 79252eeb1acb3f3a302861936d8bf7d03d30d930 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 28 Oct 2026 14:36:35 -0700 Subject: [PATCH 1/2] feat(sandbox): add a boat.dev sandbox provider boat.dev sandboxes are Ubuntu machines with Docker and Compose running, as a user with passwordless sudo, so they run envs and agents as the other VM providers' machines do. The provider, `boat`, is the `boat` extra (boat-sdk 1.13). - A sandbox is the smallest boat type covering the request, created with none of the boat account's secrets (noEnv) and no snapshots, and stopped by boat at the run's timeout. Its ports are hosted on public HTTPS URLs and recorded in its labels, so get_sandbox on a fresh provider restores them. - Every command runs detached and is followed to its end: no 600 s limit, and the output stream reconnects with no gap. A cancelled exec stops its command. - A create is idempotent across retries of a lost answer; a refused one takes a new key. A create that failed or was cancelled deletes what it made, found by a create label. - Files past one exec's worth, and objects a store cannot sign, go through signed upload URLs, are moved into place as root and checked by sha256. - boat has no egress controls, so a policy that restricts egress is refused. Co-Authored-By: Claude Opus 5.5 (1M context) --- .agentenv/config.example.toml | 4 + AGENTS.md | 9 +- THIRD_PARTY_NOTICES.md | 9 +- pyproject.toml | 7 +- src/agent_env/cli/a2a_agent/deploy.py | 2 +- src/agent_env/cli/env/deploy.py | 2 +- .../sandbox_providers/boat/__init__.py | 1 + .../sandbox_providers/boat/config.py | 102 +++++++ .../sandbox_providers/boat/lifecycle.py | 184 ++++++++++++ .../sandbox_providers/boat/provider.py | 244 ++++++++++++++++ .../sandbox_providers/boat/sandbox.py | 239 ++++++++++++++++ .../sandbox_providers/sandbox_provider.py | 1 + tst/integration/env/gateway/gateway_test.py | 7 + .../env/task_step/task_steps_test.py | 8 + .../boat_sandbox_smoke_test.py | 128 +++++++++ .../providers/sandbox_providers/boat_fakes.py | 144 ++++++++++ .../boat_sandbox_provider_test.py | 270 ++++++++++++++++++ .../sandbox_providers/boat_sandbox_test.py | 173 +++++++++++ .../sandbox_providers/egress_hosts_test.py | 1 + .../network_policy_stamping_test.py | 8 + .../provider_capabilities_test.py | 9 +- tst/util/capabilities.py | 12 +- uv.lock | 27 +- 23 files changed, 1572 insertions(+), 19 deletions(-) create mode 100644 src/agent_env/providers/sandbox_providers/boat/__init__.py create mode 100644 src/agent_env/providers/sandbox_providers/boat/config.py create mode 100644 src/agent_env/providers/sandbox_providers/boat/lifecycle.py create mode 100644 src/agent_env/providers/sandbox_providers/boat/provider.py create mode 100644 src/agent_env/providers/sandbox_providers/boat/sandbox.py create mode 100644 tst/integration/providers/sandbox_providers/boat_sandbox_smoke_test.py create mode 100644 tst/unit/providers/sandbox_providers/boat_fakes.py create mode 100644 tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py create mode 100644 tst/unit/providers/sandbox_providers/boat_sandbox_test.py diff --git a/.agentenv/config.example.toml b/.agentenv/config.example.toml index 6146dee22..ab60d7968 100644 --- a/.agentenv/config.example.toml +++ b/.agentenv/config.example.toml @@ -35,6 +35,10 @@ agent_default = "local" # agents too # snapshot = "agentenv-docker" # boot every sandbox from this Docker-capable snapshot instead of building # # the host image; a request bigger than the snapshot is refused +# Optional boat.dev sandboxes (install the 'boat' extra): +# [sandbox.providers.boat.config] +# api_key = "secret:boat_api_key" # or "env:BOAT_API_KEY" + # [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 54a9f6081..f07305138 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -62,9 +62,10 @@ or one of them only where it is used. - Defaults are local and carry no external coordinates: SQLite document store, filesystem object 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 and - Daytona sandboxes exist as implementations and are selected by config (the last seven through their extras: - `sail`, `smol`, `tensorlake`, `tenki`, `vercel`, `boxd`, `daytona`; every extra but `smol` is in `dev`). + 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`). - 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 @@ -110,7 +111,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`, whose open recordings `teardown_run` and `teardown_sandboxes` end first, and more). `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); `[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), `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/trl.py` 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, 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. | diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 83a625be6..25fab8219 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`, `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`, `daytona`, `boat`, `openenv` and `trl` extras, at the versions resolved in `uv.lock`. Development-only dependencies are not listed. ### a2a-sdk 0.3.26 @@ -137,6 +137,13 @@ The runtime dependencies of `agentenv-framework` and its `aws`, `gcp`, `sail`, ` - Source: - [License text 76](#license-text-76) +### boat-sdk 1.13.0 + +- License: MIT +- Copyright: Copyright (c) Ascii +- Source: +- [License text 4](#license-text-4) + ### boto3 1.43.38 - License: Apache-2.0 diff --git a/pyproject.toml b/pyproject.toml index d9eb26d1e..cde6e0471 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,8 +97,13 @@ boxd = [ daytona = [ "daytona~=0.224.0", # AsyncDaytona sandboxes, sessions, file API, preview links and egress lists validated against 0.224 ] +# The boat.dev sandbox provider, selected as the built-in `boat` backend. +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 +] dev = [ - "agentenv-framework[aws,boxd,daytona,gcp,openenv,sail,tensorlake,tenki,trl,vercel]", # exercise portable optional providers and stores in CI + "agentenv-framework[aws,boat,boxd,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 "transformers>=5.2", # get_json_schema, which reads agent_env.training's tool methods as TRL shows them to a model "Pillow>=10.4", # the protocol's computer extra, whose tests the unit tier runs diff --git a/src/agent_env/cli/a2a_agent/deploy.py b/src/agent_env/cli/a2a_agent/deploy.py index 9436c0858..d764f4dff 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, local) " + help="Sandbox backend(s): a built-in (modal, modal_vm, e2b, tensorlake, tenki, boxd, 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 d0d17a71a..7d5fc4bec 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, local) " + help="Sandbox backend(s): a built-in (modal, modal_vm, e2b, tensorlake, tenki, boxd, 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/boat/__init__.py b/src/agent_env/providers/sandbox_providers/boat/__init__.py new file mode 100644 index 000000000..396ee3643 --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/boat/__init__.py @@ -0,0 +1 @@ +"""boat sandbox provider: Docker-capable boat.dev sandboxes as agent-env VMs.""" diff --git a/src/agent_env/providers/sandbox_providers/boat/config.py b/src/agent_env/providers/sandbox_providers/boat/config.py new file mode 100644 index 000000000..8ebda8570 --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/boat/config.py @@ -0,0 +1,102 @@ +"""Provider settings, and how an AgentEnv request becomes a boat sandbox's type, name and labels.""" + +from __future__ import annotations + +import logging +import re +import uuid +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field +from typing import Any + +from agent_env.config.errors import ConfigError + +logger = logging.getLogger(__name__) + +SECTION = "[sandbox.providers.boat.config]" +DEFAULT_API_URL = "https://boat.dev/api/v1" +CONFIG_KEYS = ("api_key", "api_url") + +# boat's sandbox types, smallest first: (type, vCPU, GiB of memory, GB of disk). +SIZES = (("small", 2, 4, 51), ("default", 4, 8, 51), ("large", 8, 16, 111), ("xlarge", 16, 32, 251)) + +MANAGED_LABEL = "agentenv.managed" +PORTS_LABEL = "agentenv.ports" +# Names a sandbox by the create that made it, so a create that failed or was cancelled is found and deleted. +CREATE_LABEL = "agentenv.create" +ATTRIBUTION_LABEL_PREFIX = "agentenv.attr." +MAX_LABELS = 20 +LABEL_KEY_MAX = 63 +LABEL_VALUE_MAX_BYTES = 256 +_NAME_MAX = 120 +_UNSAFE_KEY = re.compile(r"[^A-Za-z0-9._/-]+") +_UNSAFE_NAME = re.compile(r"[^A-Za-z0-9-]+") + + +@dataclass(frozen=True) +class Settings: + """``[sandbox.providers.boat.config]``, parsed. The key never appears in ``repr``.""" + + api_key: str = field(repr=False) + api_url: str = DEFAULT_API_URL + + +def parse_settings(config: Mapping[str, Any]) -> Settings: + """Parse the provider's config table, refusing anything it does not understand.""" + if unknown := sorted(set(config) - set(CONFIG_KEYS)): + raise ConfigError(f"{SECTION} has unknown key(s): {unknown}") + api_key = config.get("api_key") + if not isinstance(api_key, str) or not api_key.strip(): + raise ConfigError(f"{SECTION} requires a non-empty 'api_key' (e.g. \"secret:boat_api_key\")") + api_url = config.get("api_url", DEFAULT_API_URL) + if not isinstance(api_url, str) or not api_url.strip(): + raise ConfigError(f"{SECTION} 'api_url' must be a non-empty string") + return Settings(api_key=api_key.strip(), api_url=api_url.strip().rstrip("/")) + + +def sandbox_type(*, cpu: float, memory_mb: int, disk_gb: float) -> str: + """The smallest boat type that covers the request; a request bigger than the largest is refused.""" + for name, vcpu, memory_gib, disk in SIZES: + if cpu <= vcpu and memory_mb <= memory_gib * 1024 and disk_gb <= disk: + return name + name, vcpu, memory_gib, disk = SIZES[-1] + raise ValueError( + f"boat cannot size a sandbox for cpu={cpu}, memory={memory_mb}MB, disk={disk_gb}GB: its largest type, " + f"{name!r}, has {vcpu} vCPU, {memory_gib} GiB memory and {disk} GB disk" + ) + + +def _label_value(value: object) -> str: + """``value`` cut to boat's limit on a label value, at a character boundary.""" + return str(value).encode()[:LABEL_VALUE_MAX_BYTES].decode(errors="ignore") + + +def sandbox_labels(attribution: Mapping[str, Any], ports: Iterable[int], create_token: str) -> dict[str, str]: + """Labels for reconnect (ports), cleanup (the create token, the managed marker) and attribution (cost joins). boat + keeps at most ``MAX_LABELS``; attribution past that is logged and left off.""" + labels = {MANAGED_LABEL: "true", PORTS_LABEL: ",".join(str(port) for port in ports), CREATE_LABEL: create_token} + for key, value in attribution.items(): + slug = _UNSAFE_KEY.sub("-", str(key)).strip("-./") + label = f"{ATTRIBUTION_LABEL_PREFIX}{slug}"[:LABEL_KEY_MAX] + if value is None or not slug: + continue + if len(labels) == MAX_LABELS: + logger.warning("Ignoring attribution %r on a boat sandbox: boat keeps at most %s labels", key, MAX_LABELS) + continue + labels[label] = _label_value(value) + return labels + + +def sandbox_name(attribution: Mapping[str, Any]) -> str: + """``ae-`` plus the attribution values in key order: unique, findable by people.""" + slugs = [ + _UNSAFE_NAME.sub("-", str(attribution[key])).strip("-") + for key in sorted(attribution) + if attribution[key] is not None + ] + return "-".join(["ae", uuid.uuid4().hex[:8], *filter(None, slugs)])[:_NAME_MAX].rstrip("-") + + +def ports_from_labels(labels: Mapping[str, str] | None) -> list[int]: + """The exposed ports a sandbox's labels record.""" + return [int(port) for port in str((labels or {}).get(PORTS_LABEL, "")).split(",") if port.strip().isdigit()] diff --git a/src/agent_env/providers/sandbox_providers/boat/lifecycle.py b/src/agent_env/providers/sandbox_providers/boat/lifecycle.py new file mode 100644 index 000000000..e9de5f841 --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/boat/lifecycle.py @@ -0,0 +1,184 @@ +"""Calling boat's synchronous SDK from asyncio, and creating and deleting sandboxes without leaking any. + +Every SDK call runs on a thread pool of the provider's own, so a long command's output stream never holds a thread of +asyncio's default executor, which the object stores share. A create is idempotent: a retry after a lost answer sends +the same key and gets the same sandbox, while a refusal (rate limit, no capacity) made nothing, so the next attempt +takes a key of its own. A sandbox a create made but nothing holds, because the create failed or was cancelled, is found +by its create label and deleted. +""" + +from __future__ import annotations + +import asyncio +import functools +import json +import logging +from collections.abc import Awaitable, Callable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from typing import Any, TypeVar + +import urllib3 +from boat_sdk.exceptions import ApiException + +from agent_env.providers.sandbox_providers.boat.config import CREATE_LABEL + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + +# Threads for SDK calls, shared by every boat sandbox in the process: an exec holds one for as long as its command runs. +THREADS = 64 +_EXECUTOR = ThreadPoolExecutor(max_workers=THREADS, thread_name_prefix="agent-env-boat") +# Failures of the connection, not answers from boat. +NETWORK_ERRORS = (urllib3.exceptions.HTTPError, OSError) +# Answers that refuse a create before anything is made: a full rate or concurrency limit, or no machine free. +_REFUSED = frozenset({429, 503}) +_reapers: set[asyncio.Future[Any]] = set() + + +@dataclass(frozen=True) +class Retry: + """How creates are retried and how hard cleanup tries.""" + + attempts: int + base_delay: float + max_delay: float # the longest a create waits before retrying + delete_attempts: int + delete_delay: float + cancel_wait: float # how long a cancelled create waits for its sandbox to be deleted before it is raised + + +async def call(fn: Callable[..., T], /, *args: Any, **kwargs: Any) -> T: + """``fn(*args, **kwargs)`` on the provider's thread pool.""" + return await asyncio.get_running_loop().run_in_executor(_EXECUTOR, functools.partial(fn, *args, **kwargs)) + + +def _spawn(cleanup: Awaitable[object]) -> asyncio.Future[Any]: + """Run ``cleanup`` as a task no caller's cancellation reaches, kept referenced until it ends.""" + reaper = asyncio.ensure_future(cleanup) + _reapers.add(reaper) + reaper.add_done_callback(_reapers.discard) + return reaper + + +async def settle(cleanup: Awaitable[object], *, within: float | None = None) -> None: + """Run ``cleanup`` to its end, also while the caller is being cancelled. A second cancellation, or ``within`` + seconds passing, leaves it running on its own.""" + try: + await asyncio.wait_for(asyncio.shield(_spawn(cleanup)), within) + except TimeoutError: + logger.warning("boat sandbox cleanup is still running after %ss; it continues on its own", within) + + +def _body(exc: ApiException) -> dict[str, Any]: + try: + body = json.loads(exc.body or "{}") + except (TypeError, ValueError): + return {} + return body if isinstance(body, dict) else {} + + +def error_code(exc: ApiException) -> str | None: + """boat's error code for ``exc``, e.g. ``rate_limited``.""" + body = _body(exc) + code = body.get("code") or (body.get("error") or {}).get("code") + return str(code) if code else None + + +def retry_after(exc: BaseException) -> float | None: + """How long boat asks a refused call to wait, in seconds, when it says.""" + if not isinstance(exc, ApiException): + return None + body = _body(exc) + details = (body.get("error") or {}).get("details") or {} + for value in (body.get("retryAfterSeconds"), body.get("retryAfter"), details.get("retryAfterSeconds"), + (exc.headers or {}).get("Retry-After")): + try: + if value is not None: + return float(value) + except (TypeError, ValueError): + continue + return None + + +def is_not_found(exc: BaseException) -> bool: + return isinstance(exc, ApiException) and exc.status == 404 + + +async def delete_sandbox(api: Any, sandbox_id: str, retry: Retry) -> bool: + """Delete ``sandbox_id`` with all its data, retrying through brief failures; whether it is gone.""" + for attempt in range(retry.delete_attempts): + try: + await call(api.delete_sandbox, sandbox_id=sandbox_id, x_ascii_confirm_delete=sandbox_id) + return True + except ApiException as exc: + if exc.status == 404: + return True + logger.warning("Deleting boat sandbox %s failed (attempt %s): %s %s", sandbox_id, attempt + 1, exc.status, + error_code(exc)) + except NETWORK_ERRORS as exc: + logger.warning("Deleting boat sandbox %s failed (attempt %s): %s", sandbox_id, attempt + 1, exc) + if attempt + 1 < retry.delete_attempts: + await asyncio.sleep(retry.delete_delay * 2**attempt) + logger.error("boat sandbox %s could not be deleted and runs until its auto-stop; delete it by id", sandbox_id) + return False + + +async def delete_labelled(api: Any, token: str, retry: Retry) -> None: + """Delete every sandbox the create ``token`` made, if it made one.""" + for attempt in range(retry.delete_attempts): + try: + listed = await call(api.sandboxes, label=[f"{CREATE_LABEL}={token}"]) + except (ApiException, *NETWORK_ERRORS) as exc: + logger.warning("Looking up the boat sandbox of create %s failed (attempt %s): %s", token, attempt + 1, exc) + if attempt + 1 < retry.delete_attempts: + await asyncio.sleep(retry.delete_delay * 2**attempt) + continue + for sandbox in listed.sandboxes or []: + if await delete_sandbox(api, sandbox.id, retry): + logger.info("Deleted boat sandbox %s, made by a create that did not complete", sandbox.id) + return + logger.error("A boat sandbox labelled %s=%s may run until its auto-stop; delete it by label", CREATE_LABEL, token) + + +async def _delete_outcome(api: Any, task: asyncio.Future[Any], token: str, retry: Retry) -> None: + """Delete what a create its caller gave up on made, once it answers: the sandbox it returns, or the one it may + have made though it failed, found by its create label.""" + try: + created = await task + except (Exception, asyncio.CancelledError): + await delete_labelled(api, token, retry) + return + await delete_sandbox(api, created.sandbox.id, retry) + + +async def create(api: Any, request: Any, token: str, retry: Retry) -> Any: + """``api.create(request)``: retried while refused or unanswered, leaving no sandbox behind when it fails.""" + refusals = 0 + last: BaseException | None = None + for attempt in range(retry.attempts): + task = asyncio.ensure_future(call(api.create, idempotency_key=f"agentenv-{token}-{refusals}", + create_sandbox_request=request)) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + # Raised once what the create made is deleted, so an interrupted run doesn't leave it running and billed. + await settle(_delete_outcome(api, task, token, retry), within=retry.cancel_wait) + raise + except ApiException as exc: + if exc.status in _REFUSED: + refusals += 1 + elif exc.status is not None and exc.status < 500 and error_code(exc) != "idempotency_in_progress": + raise + last = exc + except NETWORK_ERRORS as exc: + last = exc + if attempt + 1 < retry.attempts: + delay = min(retry_after(last) or retry.base_delay * 2**attempt, retry.max_delay) + logger.warning("boat refused or did not answer a sandbox create (%s); retrying in %ss", last, delay) + await asyncio.sleep(delay) + # Out of attempts: an unanswered create may still have made its sandbox. + await settle(delete_labelled(api, token, retry)) + assert last is not None + raise last diff --git a/src/agent_env/providers/sandbox_providers/boat/provider.py b/src/agent_env/providers/sandbox_providers/boat/provider.py new file mode 100644 index 000000000..c2f01a89b --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/boat/provider.py @@ -0,0 +1,244 @@ +"""boat sandbox provider. + +Every sandbox is a boat.dev sandbox, an Ubuntu machine with Docker, created with none of the boat account's secrets +(``noEnv``) and no snapshots, since agent-env never resumes one; boat stops it at agent-env's ``timeout``. Exposed ports +are hosted on public HTTPS URLs, recorded in the sandbox's labels so a provider built fresh reconnects to them. boat has +no egress controls, so a policy that restricts egress is refused. +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid +from collections.abc import Callable, Mapping, Sequence +from typing import Any, ClassVar, Self + +from boat_sdk import ApiClient, Configuration, run_command +from boat_sdk.api.boat_api import BoatApi +from boat_sdk.exceptions import ApiException +from boat_sdk.models import CreateSandboxRequest, HostPortRequest + +from agent_env.attribution import PIPELINE_STEP_KEY, RUN_ID_KEY, Attribution +from agent_env.providers.sandbox_providers.boat import lifecycle +from agent_env.providers.sandbox_providers.boat.config import ( + Settings, + parse_settings, + ports_from_labels, + sandbox_labels, + sandbox_name, + sandbox_type, +) +from agent_env.providers.sandbox_providers.boat.sandbox import BoatSandbox, RunCommand +from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy +from agent_env.providers.sandbox_providers.sandbox_provider import ( + SandboxNotFoundError, + SandboxProvider, + apply_default_attribution, + refuse_unenforceable_policy, +) + +logger = logging.getLogger(__name__) + +SANDBOX_STARTED_EVENT = "agent_env.boat_sandbox_started" +# States of a sandbox that accepts commands. +_READY_STATES = frozenset({"ready", "idle", "running"}) +# States a starting sandbox does not come back from. +_FAILED_STATES = frozenset({"archiving", "archived", "error"}) +# A stopped sandbox: with snapshots off, its disk went with it. +_GONE_STATES = frozenset({"archiving", "archived"}) + + +def _api(settings: Settings) -> BoatApi: + configuration = Configuration(host=settings.api_url, access_token=settings.api_key) + # A connection for each thread SDK calls run on, so none is opened and dropped per call. + configuration.connection_pool_maxsize = lifecycle.THREADS + return BoatApi(ApiClient(configuration)) + + +class BoatSandboxProvider(SandboxProvider): + """Docker-capable boat sandboxes as agent-env VMs. ``api_key`` comes from the resolved config (an ``env:`` or + ``secret:`` reference), never from the process environment.""" + + # boat serves hosted ports at https://-.on.boat.dev. + EGRESS_HOSTS: ClassVar[tuple[str, ...]] = ("*.on.boat.dev",) + _CREATE_ATTEMPTS = 6 + _RETRY_BASE_DELAY = 2.0 + _RETRY_MAX_DELAY = 60.0 + _DELETE_ATTEMPTS = 4 + _DELETE_DELAY = 1.0 + _CANCEL_WAIT = 120.0 + _READY_TIMEOUT = 300.0 + _READY_POLL_INTERVAL = 1.0 + + def __init__( + self, + settings: Settings, + *, + api_factory: Callable[[Settings], Any] = _api, + run_command: RunCommand = run_command, + ): + self._settings = settings + # Injectable so unit tests never build an SDK client or stream from boat. + self._api = api_factory(settings) + self._run_command = run_command + self._retry = lifecycle.Retry( + attempts=self._CREATE_ATTEMPTS, + base_delay=self._RETRY_BASE_DELAY, + max_delay=self._RETRY_MAX_DELAY, + delete_attempts=self._DELETE_ATTEMPTS, + delete_delay=self._DELETE_DELAY, + cancel_wait=self._CANCEL_WAIT, + ) + + def __repr__(self) -> str: + return f"BoatSandboxProvider(api_url={self._settings.api_url!r})" + + @classmethod + def from_config(cls, **config: Any) -> Self: + return cls(parse_settings(config)) + + async def close(self) -> None: + """Close the SDK client's connections.""" + api_client = getattr(self._api, "api_client", None) + if api_client is not None: + await lifecycle.call(api_client.close) + + def _handle(self, sandbox_id: str, tunnel_urls: Mapping[int, str]) -> BoatSandbox: + # boat applies no egress policy, so every sandbox's is unrestricted. + return BoatSandbox(self._api, sandbox_id, tunnel_urls=tunnel_urls, network_policy=NetworkPolicy(), + retry=self._retry, run_command=self._run_command) + + async def _host(self, sandbox_id: str, port: int) -> str: + """The public URL of ``port``; hosting a hosted port again answers its URL.""" + hosted = await lifecycle.call(self._api.host_port, sandbox_id, HostPortRequest(port=port, public=True)) + return hosted.url + + async def _hosted_urls(self, sandbox_id: str, ports: Sequence[int], hosted: Sequence[Any] | None) -> dict[int, str]: + """The URL of each port: as the create hosted it, or hosted now where the create could not.""" + urls = { + entry.port: entry.url + for entry in hosted or [] + if entry.port in ports and entry.url and getattr(entry, "success", True) is not False + } + for port in ports: + if port not in urls: + urls[port] = await self._host(sandbox_id, port) + return urls + + async def _wait_ready(self, sandbox_id: str) -> None: + """Poll until the sandbox accepts commands, within ``_READY_TIMEOUT``.""" + deadline = asyncio.get_running_loop().time() + self._READY_TIMEOUT + while True: + state = (await lifecycle.call(self._api.get, sandbox_id)).sandbox.state + if state in _READY_STATES: + return + if state in _FAILED_STATES: + raise RuntimeError(f"boat sandbox {sandbox_id} went to state {state!r} while it started") + if asyncio.get_running_loop().time() >= deadline: + raise TimeoutError(f"boat sandbox {sandbox_id} was not ready after {self._READY_TIMEOUT}s ({state!r})") + await asyncio.sleep(self._READY_POLL_INTERVAL) + + 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, + ) -> BoatSandbox: + """Create a sandbox of the smallest type that covers the request; ``timeout`` is when boat stops it.""" + refuse_unenforceable_policy(self, network_policy) + if image is not None: + raise ValueError("the boat provider boots boat's Ubuntu image; image overrides are unsupported") + if boot_mode is not None: + logger.warning("Ignoring boot_mode=%s: boat sandboxes have no boot modes", boot_mode) + size = sandbox_type(cpu=cpu, memory_mb=memory, disk_gb=disk_size_gb) + ports = list(dict.fromkeys(exposed_ports or [])) + defaults = apply_default_attribution(dict(attribution or {})) + resolved = {key: str(value) for key, value in defaults.items() if value is not None} + token = uuid.uuid4().hex + request = CreateSandboxRequest( + type=size, + ttl_seconds=int(timeout), + name=sandbox_name(resolved), + labels=sandbox_labels(resolved, ports, token), + no_env=True, + snapshots=False, + hosted_ports=[{"port": port, "public": True} for port in ports] or None, + ) + created = await lifecycle.create(self._api, request, token, self._retry) + sandbox_id = created.sandbox.id + try: + await self._wait_ready(sandbox_id) + sandbox = self._handle(sandbox_id, await self._hosted_urls(sandbox_id, ports, created.hosted_ports)) + logger.info( + "boat sandbox started: sandbox_id=%s type=%s ports=%s attribution=%s", + sandbox_id, + size, + ports, + resolved, + extra={ + "event": SANDBOX_STARTED_EVENT, + "boat_sandbox_id": sandbox_id, + "boat_attribution": resolved, + PIPELINE_STEP_KEY: resolved.get(PIPELINE_STEP_KEY), + RUN_ID_KEY: resolved.get(RUN_ID_KEY), + }, + ) + if setup_for_gateway: + await sandbox.setup_vm_for_gateway(ports) + return sandbox + except BaseException: + # The sandbox is deleted, with retries, before the create's own error or cancellation is raised. + await lifecycle.settle(lifecycle.delete_sandbox(self._api, sandbox_id, self._retry)) + raise + + 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, + ) -> BoatSandbox: + """A bare VM the caller loads and starts ``image_name`` in, as on the other VM providers.""" + del image_name, env + return await self.create_vm( + cpu=cpu, + memory=memory, + disk_size_gb=disk_size_gb, + timeout=timeout, + exposed_ports=[port], + attribution=attribution, + network_policy=network_policy, + ) + + async def get_sandbox(self, sandbox_id: str) -> BoatSandbox: + """Reconnect, restoring the ports and URLs the sandbox's labels record; it changes nothing.""" + try: + raw = (await lifecycle.call(self._api.get, sandbox_id)).sandbox + except ApiException as exc: + if lifecycle.is_not_found(exc): + raise SandboxNotFoundError(f"boat sandbox {sandbox_id!r} was not found") from exc + raise + if raw.state in _GONE_STATES: + raise SandboxNotFoundError(f"boat sandbox {sandbox_id!r} is stopped, and without snapshots its disk is gone") + ports = ports_from_labels(raw.labels) + return self._handle(sandbox_id, {port: await self._host(sandbox_id, port) for port in ports}) + + async def terminate_sandbox(self, sandbox_id: str) -> None: + """Delete by id, with nothing read first; a sandbox already gone is done.""" + if not await lifecycle.delete_sandbox(self._api, sandbox_id, self._retry): + raise RuntimeError(f"boat sandbox {sandbox_id} could not be deleted") diff --git a/src/agent_env/providers/sandbox_providers/boat/sandbox.py b/src/agent_env/providers/sandbox_providers/boat/sandbox.py new file mode 100644 index 000000000..c021d3dfe --- /dev/null +++ b/src/agent_env/providers/sandbox_providers/boat/sandbox.py @@ -0,0 +1,239 @@ +"""A boat sandbox presented through agent-env's ``VmSandbox`` contract. + +boat sandboxes boot Ubuntu with Docker and Compose running, as the user ``user``, who has passwordless sudo, so the +``sudo`` agent-env prefixes is kept. Every command runs detached and is followed to its end: a detached command has no +time limit and its output stream reconnects without a gap, where a synchronous one stops at 600 s. Hosted ports reach +the sandbox through boat's tunnel, so the sandbox opens no firewall ports. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import logging +import shlex +import threading +import uuid +from collections.abc import AsyncIterator, Callable, Iterator, Mapping +from typing import IO, TYPE_CHECKING, Any + +import httpx +from boat_sdk import run_command +from boat_sdk.exceptions import ApiException +from boat_sdk.models import FileUrlRequest + +from agent_env.providers.sandbox_providers.boat import lifecycle +from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy, VmSandbox +from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_VM + +if TYPE_CHECKING: + from agent_env.store.object_store import ObjectStore + +logger = logging.getLogger(__name__) + +# Bytes past what one exec carries go through a signed upload URL: one PUT of any size, streamed. +_UPLOAD_URL_TTL = 3600 +_UPLOAD_TIMEOUT = httpx.Timeout(600.0, connect=30.0) +_UPLOAD_PART_BYTES = 4 * 1024 * 1024 + +# What exec reports when the transport drops without an exit status; exec_script retries only this. +UNKNOWN_EXIT_CODE = -1 +# How long a cancelled exec waits for its command to be stopped before the cancellation is raised. +_STOP_WAIT = 30.0 +# The image starts dockerd at boot; the first command can be ahead of it. +DOCKER_READY = r""" +for _ in $(seq 90); do + docker info >/dev/null 2>&1 && exec docker compose version + sleep 1 +done +echo "docker daemon did not become ready" >&2 +exit 1 +""" + +RunCommand = Callable[..., Iterator[dict[str, Any]]] + + +class _BytesReader: + """The ``StreamReader`` subset ``Sandbox.exec_with_output`` reads.""" + + def __init__(self, value: str): + self._value = value.encode() + + async def read(self) -> bytes: + return self._value + + +class _CompletedProcess: + """A finished command as the process ``Sandbox.exec`` returns.""" + + def __init__(self, exit_code: int | None, stdout: str, stderr: str): + self.stdout = _BytesReader(stdout) + self.stderr = _BytesReader(stderr) + self._exit_code = UNKNOWN_EXIT_CODE if exit_code is None else int(exit_code) + + async def wait(self) -> int: + return self._exit_code + + +class _Run: + """One command, shared by the event loop and the thread that follows it, so a cancel stops it whether or not it + has started yet.""" + + def __init__(self) -> None: + self.lock = threading.Lock() + self.process_id: int | None = None + self.cancelled = False + + +def _follow(api: Any, sandbox_id: str, script: str, run: _Run, run_command: RunCommand) -> list[dict[str, Any]]: + """Every frame of ``script`` after its start, run detached in the sandbox. A command cancelled before it started is + stopped as soon as it does.""" + frames = [] + for frame in run_command(api, sandbox_id, script): + if frame.get("type") == "started": + with run.lock: + run.process_id = frame.get("processId") + stop = run.cancelled + if stop: + try: + api.stop_command(sandbox_id, run.process_id) + except (ApiException, *lifecycle.NETWORK_ERRORS) as exc: + logger.warning("Stopping command %s in boat sandbox %s failed: %s", run.process_id, sandbox_id, exc) + return frames + continue + frames.append(frame) + return frames + + +def _completed(frames: list[dict[str, Any]]) -> _CompletedProcess: + stdout = "".join(str(frame.get("data", "")) for frame in frames if frame.get("type") == "stdout") + stderr = "".join(str(frame.get("data", "")) for frame in frames if frame.get("type") == "stderr") + for frame in reversed(frames): + if frame.get("type") == "exit": + return _CompletedProcess(frame.get("exitCode"), stdout, stderr) + if frame.get("type") == "error": + message = frame.get("message") or frame.get("error") or "the command's output stream failed" + return _CompletedProcess(UNKNOWN_EXIT_CODE, stdout, f"{stderr}{message}") + return _CompletedProcess(UNKNOWN_EXIT_CODE, stdout, f"{stderr}the command's output stream ended without an exit") + + +class BoatSandbox(VmSandbox): + """A boat sandbox with a Docker daemon. ``api`` is its provider's ``BoatApi``.""" + + type = "boat" + + def __init__( + self, + api: Any, + sandbox_id: str, + *, + tunnel_urls: Mapping[int, str], + network_policy: NetworkPolicy | None, + retry: lifecycle.Retry, + run_command: RunCommand = run_command, + ): + self._api = api + self._retry = retry + # Injectable so unit tests never stream from boat. + self._run_command = run_command + self.sandbox_id = sandbox_id + self.mode = SANDBOX_MODE_VM + self.vnc_url = None + self.tunnel_urls = dict(tunnel_urls) + self.network_policy = network_policy + + async def terminate(self) -> None: + """Delete the sandbox and its data; one already gone is not an error.""" + if not await lifecycle.delete_sandbox(self._api, self.sandbox_id, self._retry): + raise RuntimeError(f"boat sandbox {self.sandbox_id} could not be deleted") + + async def exec(self, *command: str) -> _CompletedProcess: + """Run argv as one shell command, detached and followed to its end. A cancelled exec stops the command.""" + run = _Run() + script = shlex.join(command) + try: + frames = await lifecycle.call(_follow, self._api, self.sandbox_id, script, run, self._run_command) + except asyncio.CancelledError: + with run.lock: + run.cancelled = True + process_id = run.process_id + if process_id is not None: + await lifecycle.settle(self._stop(process_id), within=_STOP_WAIT) + raise + except lifecycle.NETWORK_ERRORS as exc: + return _CompletedProcess(UNKNOWN_EXIT_CODE, "", f"{type(exc).__name__}: {exc}") + except ApiException as exc: + # boat failed, or limited this client: the command may or may not have run. + if exc.status is None or exc.status >= 500 or exc.status == 429: + return _CompletedProcess(UNKNOWN_EXIT_CODE, "", f"boat {exc.status} {lifecycle.error_code(exc)}") + raise + return _completed(frames) + + async def _stop(self, process_id: int) -> None: + try: + await lifecycle.call(self._api.stop_command, self.sandbox_id, process_id) + except (ApiException, *lifecycle.NETWORK_ERRORS) as exc: + logger.warning("Stopping command %s in boat sandbox %s failed: %s", process_id, self.sandbox_id, exc) + + async def wait_for_vm(self) -> None: + """Wait for Docker, which boat starts at boot, and require Compose v2, which ``deploy_env`` uses.""" + code, version, err = await self.exec_with_output("bash", "-c", DOCKER_READY) + if code != 0: + raise RuntimeError( + f"Docker with Compose v2 is not ready in boat sandbox {self.sandbox_id}: {(err or version).strip()}" + ) + logger.info("Docker Compose ready in boat sandbox %s: %s", self.sandbox_id, version.strip()) + + async def setup_vm_for_gateway(self, exposed_ports: list[int] | None = None) -> None: + """Wait for Docker and Compose; hosted ports need no firewall rule.""" + await self.wait_for_vm() + + async def _write_bytes_to_vm_path(self, data: bytes, vm_path: str) -> None: + """Bytes one exec carries go as base64 in it; more go through an upload URL.""" + if len(base64.b64encode(data)) <= self._WFT_CHUNK_BYTES: + await super()._write_bytes_to_vm_path(data, vm_path) + return + await self._upload(data, vm_path) + + async def _write_unsigned_object(self, object_store: ObjectStore, object_url: str, vm_path: str) -> None: + """Stream an object through an upload URL, not as base64 over exec.""" + source = await asyncio.to_thread(object_store.open, object_url) + try: + await self._upload(_read_parts(source), vm_path) + finally: + source.close() + + async def _upload(self, body: bytes | AsyncIterator[bytes], vm_path: str) -> None: + """PUT ``body`` to a file in ``/tmp``, where boat writes uploads, then move it to ``vm_path`` as root, as an + exec would write it, and check its sha256.""" + request = FileUrlRequest(path=f"/tmp/.agentenv-upload-{uuid.uuid4().hex}", ttl_seconds=_UPLOAD_URL_TTL) + grant = await lifecycle.call(self._api.create_upload_url, self.sandbox_id, request) + digest = hashlib.sha256() + + async def parts() -> AsyncIterator[bytes]: + async for part in body: # type: ignore[union-attr] + digest.update(part) + yield part + + if isinstance(body, bytes): + digest.update(body) + staged, target = shlex.quote(grant.path), shlex.quote(vm_path) + try: + async with httpx.AsyncClient(timeout=_UPLOAD_TIMEOUT) as client: + response = await client.request(grant.method, grant.url, content=body if isinstance(body, bytes) else parts()) + response.raise_for_status() + written = await self.exec_script(f"mv -f {staged} {target} && chown root:root {target} && sha256sum {target}") + except BaseException: + await self._remove_vm_temp_file(grant.path) + raise + actual = written.split()[0] if written.split() else "" + if actual != digest.hexdigest(): + raise RuntimeError(f"{vm_path} in boat sandbox {self.sandbox_id} does not match the upload: its sha256 is " + f"{actual or 'missing'}, the upload's {digest.hexdigest()}") + + +async def _read_parts(source: IO[bytes]) -> AsyncIterator[bytes]: + """``source`` in upload-sized parts, read off the event loop.""" + while part := await asyncio.to_thread(source.read, _UPLOAD_PART_BYTES): + yield part diff --git a/src/agent_env/providers/sandbox_providers/sandbox_provider.py b/src/agent_env/providers/sandbox_providers/sandbox_provider.py index d1214d392..bf0575f97 100644 --- a/src/agent_env/providers/sandbox_providers/sandbox_provider.py +++ b/src/agent_env/providers/sandbox_providers/sandbox_provider.py @@ -321,6 +321,7 @@ def effective_network_policy(cls, policy: Optional[NetworkPolicy]) -> NetworkPol "tenki": "agent_env.providers.sandbox_providers.tenki.provider:TenkiSandboxProvider", "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", "local": "agent_env.providers.sandbox_providers.local_sandbox:LocalSandboxProvider", } diff --git a/tst/integration/env/gateway/gateway_test.py b/tst/integration/env/gateway/gateway_test.py index ec075584d..ffbb878ae 100644 --- a/tst/integration/env/gateway/gateway_test.py +++ b/tst/integration/env/gateway/gateway_test.py @@ -478,6 +478,10 @@ def agentenv_website_env() -> WebsiteEnv: "daytona", id="daytona", marks=skip_without_remote_sandbox("daytona"), + ), pytest.param( + "boat", + id="boat", + marks=skip_without_remote_sandbox("boat"), ), ]) def sandbox_provider(request): @@ -512,6 +516,9 @@ def sandbox_provider(request): elif request.param == "daytona": set_sandbox_provider(build_sandbox_provider("daytona")) set_env_sandbox_provider(build_sandbox_provider("daytona")) + elif request.param == "boat": + set_sandbox_provider(build_sandbox_provider("boat")) + set_env_sandbox_provider(build_sandbox_provider("boat")) 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 59d3c68a8..d4a8030b3 100644 --- a/tst/integration/env/task_step/task_steps_test.py +++ b/tst/integration/env/task_step/task_steps_test.py @@ -297,6 +297,10 @@ def echo_agent() -> A2AAgent: "daytona", id="daytona", marks=skip_without_remote_sandbox("daytona"), + ), pytest.param( + "boat", + id="boat", + marks=skip_without_remote_sandbox("boat"), ), ]) def sandbox_provider(request): @@ -340,6 +344,10 @@ def sandbox_provider(request): set_sandbox_provider(build_sandbox_provider("daytona")) set_env_sandbox_provider(build_sandbox_provider("daytona")) set_agent_sandbox_provider(build_sandbox_provider("daytona")) + elif request.param == "boat": + set_sandbox_provider(build_sandbox_provider("boat")) + set_env_sandbox_provider(build_sandbox_provider("boat")) + set_agent_sandbox_provider(build_sandbox_provider("boat")) try: yield request.param finally: diff --git a/tst/integration/providers/sandbox_providers/boat_sandbox_smoke_test.py b/tst/integration/providers/sandbox_providers/boat_sandbox_smoke_test.py new file mode 100644 index 000000000..64e2749ad --- /dev/null +++ b/tst/integration/providers/sandbox_providers/boat_sandbox_smoke_test.py @@ -0,0 +1,128 @@ +"""boat sandbox smoke test. + +Runs when the resolved config can build the ``boat`` sandbox provider (the ``boat`` extra and +``[sandbox.providers.boat.config]`` with ``api_key = "secret:boat_api_key"``), the ``remote_sandbox`` capability, and +skips with the declared reason otherwise. The test never reads or prints 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 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 + +pytestmark = [ + pytest.mark.integration, + pytest.mark.int_test_slow, + pytest.mark.asyncio, + skip_without_remote_sandbox("boat"), +] + + +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_asyncio.fixture(scope="module") +async def boat_provider(): + provider = build_sandbox_provider("boat") + yield provider + await provider.close() + + +@pytest_asyncio.fixture(scope="module") +async def boat_sandbox(boat_provider): + """One boat sandbox serving on ``_PORT`` from a container, always deleted after the module.""" + sandbox = await boat_provider.create_vm(cpu=2, memory=4096, exposed_ports=[_PORT], timeout=1800) + try: + await sandbox.exec_script( + f"docker run -d --name smoke -p {_PORT}:80 public.ecr.aws/docker/library/nginx:alpine" + ) + await _wait_for_200(sandbox.tunnel_urls[_PORT]) + yield sandbox + finally: + await sandbox.terminate() + + +async def test_boat_sandbox_docker_ports_upload_and_reconnect(boat_sandbox): + """Nested Docker, a hosted port end to end, an upload to a root-owned path, and a reconnect from a fresh + provider.""" + assert (boat_sandbox.type, boat_sandbox.mode) == ("boat", "vm") + assert boat_sandbox.network_policy == NetworkPolicy() + assert boat_sandbox.tunnel_urls[_PORT].startswith("https://") + + exit_code, version, stderr = await boat_sandbox.exec_with_output( + "sudo", "docker", "info", "--format", "{{.ServerVersion}}" + ) + assert exit_code == 0, stderr + assert version.strip() + + payload = os.urandom(9 * 1024 * 1024 + 17) + await boat_sandbox.write_host_file(payload, "/opt/smoke/payload.bin") + digest = await boat_sandbox.exec_script("sha256sum /opt/smoke/payload.bin && stat -c %U /opt/smoke/payload.bin") + assert digest.split()[0] == hashlib.sha256(payload).hexdigest() and digest.split()[-1] == "root" + + fresh = build_sandbox_provider("boat") + try: + reconnected = await fresh.get_sandbox(boat_sandbox.sandbox_id) + assert reconnected.tunnel_urls == boat_sandbox.tunnel_urls + assert reconnected.network_policy == boat_sandbox.network_policy + assert (await reconnected.exec_with_output("true"))[0] == 0 + finally: + await fresh.close() + + +async def test_a_cancelled_exec_stops_its_command_in_the_sandbox(boat_sandbox): + marker = f"agentenv-cancel-{os.urandom(4).hex()}" + # Bracketed, so the pattern matches the command but not the shell boat runs pgrep in, whose line holds it. + pattern = f"[{marker[0]}]{marker[1:]}" + task = asyncio.create_task(boat_sandbox.exec("bash", "-c", f"exec -a {marker} sleep 600")) + for _ in range(30): + if (await boat_sandbox.exec_with_output("pgrep", "-f", pattern))[0] == 0: + break + await asyncio.sleep(1) + else: + pytest.fail("the command never started") + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + for _ in range(30): + if (await boat_sandbox.exec_with_output("pgrep", "-f", pattern))[0] != 0: + return + await asyncio.sleep(1) + pytest.fail("a cancelled exec left its command running") + + +async def test_boat_terminate_twice_then_the_sandbox_is_not_found(boat_provider): + """A second terminate, and one by id, of a deleted sandbox succeed, and ``get_sandbox`` then reports it not + found.""" + sandbox = await boat_provider.create_vm(cpu=2, memory=4096, timeout=600, setup_for_gateway=False) + await sandbox.terminate() + await sandbox.terminate() + await boat_provider.terminate_sandbox(sandbox.sandbox_id) + + for _ in range(30): + try: + await boat_provider.get_sandbox(sandbox.sandbox_id) + except SandboxNotFoundError: + return + await asyncio.sleep(2) + pytest.fail(f"boat sandbox {sandbox.sandbox_id} was still found a minute after terminate") diff --git a/tst/unit/providers/sandbox_providers/boat_fakes.py b/tst/unit/providers/sandbox_providers/boat_fakes.py new file mode 100644 index 000000000..6ae5eb37d --- /dev/null +++ b/tst/unit/providers/sandbox_providers/boat_fakes.py @@ -0,0 +1,144 @@ +"""In-memory stand-ins for the parts of boat's synchronous SDK the provider uses: each records what it is sent and +answers from state a test sets, with no network.""" + +from __future__ import annotations + +import itertools +import json +import shlex +import threading +from collections.abc import Callable, Iterator +from dataclasses import dataclass, field +from types import SimpleNamespace +from typing import Any + +from boat_sdk.exceptions import ApiException + +from agent_env.providers.sandbox_providers.boat.config import Settings + +# (exit_code, stdout, stderr) for one command, or an exception its output stream raises. +Outcome = tuple[int, str, str] | BaseException +Handler = Callable[[list[str]], Outcome] + +SETTINGS = Settings(api_key="boat_test") + + +def api_error(status: int, code: str, **body: Any) -> ApiException: + return ApiException(status=status, reason=code, body=json.dumps({"code": code, **body})) + + +def refused(error: BaseException | None) -> bool: + """Whether boat answers ``error`` to a create having made nothing: a 4xx, or no capacity.""" + return isinstance(error, ApiException) and error.status is not None and (error.status < 500 or error.status == 503) + + +@dataclass +class FakeBoatApi: + """Sandboxes, hosted ports and commands kept in memory. ``create_errors`` are raised by the next creates in turn; + with ``create_gate`` set, a create waits for it after setting ``create_entered``.""" + + states: list[str] = field(default_factory=lambda: ["ready"]) + create_errors: list[BaseException] = field(default_factory=list) + hosted_port_errors: set[int] = field(default_factory=set) + create_gate: threading.Event | None = None + create_entered: threading.Event = field(default_factory=threading.Event) + sandboxes_by_id: dict[str, SimpleNamespace] = field(default_factory=dict) + creates: list[dict[str, Any]] = field(default_factory=list) + deletes: list[str] = field(default_factory=list) + hosts: list[tuple[str, int, bool]] = field(default_factory=list) + label_queries: list[list[str]] = field(default_factory=list) + stopped: list[tuple[str, int]] = field(default_factory=list) + uploads: list[str] = field(default_factory=list) + api_client: SimpleNamespace = field(default_factory=lambda: SimpleNamespace(closed=False)) + _ids: Iterator[int] = field(default_factory=itertools.count) + + def __post_init__(self) -> None: + self.api_client.close = lambda: setattr(self.api_client, "closed", True) + + def url(self, sandbox_id: str, port: int) -> str: + return f"https://box-{sandbox_id}-{port}.on.boat.dev" + + def create(self, idempotency_key: str | None = None, create_sandbox_request: Any = None) -> SimpleNamespace: + self.creates.append({"key": idempotency_key, "request": create_sandbox_request}) + self.create_entered.set() + if self.create_gate is not None: + assert self.create_gate.wait(5) + request = create_sandbox_request + error = self.create_errors.pop(0) if self.create_errors else None + if refused(error): + raise error + # Any other failure stands for an answer lost after boat made the sandbox. + sandbox_id = f"bx_{next(self._ids)}" + self.sandboxes_by_id[sandbox_id] = SimpleNamespace(id=sandbox_id, state="provisioned", + labels=dict(request.labels or {}), states=list(self.states)) + if error is not None: + raise error + hosted = [ + SimpleNamespace(port=entry.port, url=None if entry.port in self.hosted_port_errors else self.url(sandbox_id, entry.port), + success=entry.port not in self.hosted_port_errors) + for entry in request.hosted_ports or [] + ] + return SimpleNamespace(sandbox=SimpleNamespace(id=sandbox_id, state="provisioned"), hosted_ports=hosted) + + def get(self, sandbox_id: str) -> SimpleNamespace: + if sandbox_id not in self.sandboxes_by_id: + raise api_error(404, "not_found") + raw = self.sandboxes_by_id[sandbox_id] + if raw.states: + raw.state = raw.states.pop(0) + return SimpleNamespace(sandbox=SimpleNamespace(id=sandbox_id, state=raw.state, labels=dict(raw.labels))) + + def delete_sandbox(self, x_ascii_confirm_delete: str, sandbox_id: str) -> SimpleNamespace: + assert x_ascii_confirm_delete == sandbox_id + self.deletes.append(sandbox_id) + self.sandboxes_by_id.pop(sandbox_id, None) + return SimpleNamespace(ok=True) + + def host_port(self, sandbox_id: str, host_port_request: Any) -> SimpleNamespace: + self.hosts.append((sandbox_id, host_port_request.port, bool(host_port_request.public))) + return SimpleNamespace(url=self.url(sandbox_id, host_port_request.port)) + + def sandboxes(self, label: list[str] | None = None) -> SimpleNamespace: + self.label_queries.append(list(label or [])) + wanted = dict(entry.split("=", 1) for entry in label or []) + found = [raw for raw in self.sandboxes_by_id.values() + if all(raw.labels.get(key) == value for key, value in wanted.items())] + return SimpleNamespace(sandboxes=[SimpleNamespace(id=raw.id) for raw in found]) + + def stop_command(self, sandbox_id: str, process_id: int) -> SimpleNamespace: + self.stopped.append((sandbox_id, process_id)) + return SimpleNamespace(ok=True) + + def create_upload_url(self, sandbox_id: str, file_url_request: Any) -> SimpleNamespace: + self.uploads.append(file_url_request.path) + return SimpleNamespace(url=f"https://box-{sandbox_id}.on.boat.dev/file-transfer?path={file_url_request.path}", + method="PUT", path=file_url_request.path) + + +@dataclass +class FakeCommands: + """``boat_sdk.run_command`` over ``handler``, which gets each command's argv; by default exit 0 with no output. + With ``gate`` set, a command waits for it after its start, having set ``started``.""" + + handler: Handler = lambda argv: (0, "", "") + commands: list[list[str]] = field(default_factory=list) + gate: threading.Event | None = None + started: threading.Event = field(default_factory=threading.Event) + _process_ids: Iterator[int] = field(default_factory=lambda: itertools.count(100)) + + def __call__(self, api: Any, sandbox_id: str, command: str, cwd: str | None = None) -> Iterator[dict[str, Any]]: + argv = shlex.split(command) + self.commands.append(argv) + yield {"type": "started", "processId": next(self._process_ids)} + self.started.set() + if self.gate is not None: + assert self.gate.wait(5) + outcome = self.handler(argv) + if isinstance(outcome, BaseException): + raise outcome + code, stdout, stderr = outcome + if stdout: + yield {"type": "stdout", "data": stdout, "nextOffset": len(stdout)} + if stderr: + yield {"type": "stderr", "data": stderr, "nextOffset": len(stderr)} + yield {"type": "exit", "exitCode": code} diff --git a/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py b/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py new file mode 100644 index 000000000..efdd8bcf2 --- /dev/null +++ b/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py @@ -0,0 +1,270 @@ +"""Offline unit tests for the boat sandbox provider: config, sizing, labels, create, cleanup, reconnect and terminate.""" + +from __future__ import annotations + +import asyncio +import logging +import threading + +import pytest +import urllib3 + +from agent_env.config.errors import ConfigError +from agent_env.providers.sandbox_providers.boat.config import ( + ATTRIBUTION_LABEL_PREFIX, + CREATE_LABEL, + DEFAULT_API_URL, + LABEL_VALUE_MAX_BYTES, + MANAGED_LABEL, + MAX_LABELS, + PORTS_LABEL, + parse_settings, + ports_from_labels, + sandbox_labels, + sandbox_name, + sandbox_type, +) +from agent_env.providers.sandbox_providers.boat.provider import BoatSandboxProvider +from agent_env.providers.sandbox_providers.boat.sandbox import DOCKER_READY, BoatSandbox +from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy, NetworkPolicyUnsupportedError +from agent_env.providers.sandbox_providers.sandbox_provider import ( + _BUILTIN_SANDBOX_PROVIDERS, + SandboxNotFoundError, +) + +from tst.unit.providers.sandbox_providers.boat_fakes import SETTINGS, FakeBoatApi, FakeCommands, api_error + + +@pytest.fixture(autouse=True) +def no_waits(monkeypatch): + for name in ("_RETRY_BASE_DELAY", "_DELETE_DELAY", "_READY_POLL_INTERVAL"): + monkeypatch.setattr(BoatSandboxProvider, name, 0.0) + + +def provider(api: FakeBoatApi, commands: FakeCommands | None = None) -> BoatSandboxProvider: + return BoatSandboxProvider(SETTINGS, api_factory=lambda settings: api, run_command=commands or FakeCommands()) + + +def docker_ready(argv: list[str]) -> tuple[int, str, str]: + return (0, "Docker Compose version v5.3.1\n", "") if argv == ["bash", "-c", DOCKER_READY] else (0, "", "") + + +# --- config --- + + +def test_settings_need_an_api_key_and_take_no_unknown_keys(): + with pytest.raises(ConfigError, match="requires a non-empty 'api_key'"): + parse_settings({}) + with pytest.raises(ConfigError, match="unknown key"): + parse_settings({"api_key": "boat_x", "region": "eu"}) + settings = parse_settings({"api_key": " boat_x ", "api_url": "https://boat.example/api/v1/"}) + assert (settings.api_key, settings.api_url) == ("boat_x", "https://boat.example/api/v1") + assert parse_settings({"api_key": "boat_x"}).api_url == DEFAULT_API_URL + assert "boat_x" not in repr(settings) + + +def test_from_config_builds_a_provider_without_showing_the_key(): + built = BoatSandboxProvider.from_config(api_key="boat_secret") + assert "boat_secret" not in repr(built) + assert _BUILTIN_SANDBOX_PROVIDERS["boat"].endswith("boat.provider:BoatSandboxProvider") + + +@pytest.mark.parametrize("cpu, memory_mb, disk_gb, expected", [ + (1, 2048, 10, "small"), + (2, 4096, 51, "small"), + (1, 8192, 10, "default"), + (4, 4096, 10, "default"), + (2, 4096, 60, "large"), + (8, 16384, 100, "large"), + (16, 32768, 251, "xlarge"), +]) +def test_a_request_gets_the_smallest_type_that_covers_it(cpu, memory_mb, disk_gb, expected): + assert sandbox_type(cpu=cpu, memory_mb=memory_mb, disk_gb=disk_gb) == expected + + +def test_a_request_bigger_than_the_largest_type_is_refused(): + with pytest.raises(ValueError, match="largest type, 'xlarge'"): + sandbox_type(cpu=17, memory_mb=1024, disk_gb=10) + + +def test_labels_carry_ports_the_create_and_attribution_within_boats_limits(): + labels = sandbox_labels({"project_id": "p 1", "a/b": "x" * 300, "skip": None}, [8000, 8080], "tok") + assert labels[MANAGED_LABEL] == "true" and labels[CREATE_LABEL] == "tok" + assert ports_from_labels(labels) == [8000, 8080] + assert labels[f"{ATTRIBUTION_LABEL_PREFIX}project_id"] == "p 1" + assert len(labels[f"{ATTRIBUTION_LABEL_PREFIX}a/b"].encode()) == LABEL_VALUE_MAX_BYTES + assert f"{ATTRIBUTION_LABEL_PREFIX}skip" not in labels + + +def test_attribution_past_boats_label_limit_is_logged_and_left_off(caplog): + attribution = {f"key{n}": str(n) for n in range(MAX_LABELS)} + with caplog.at_level(logging.WARNING): + labels = sandbox_labels(attribution, [], "tok") + assert len(labels) == MAX_LABELS + assert "at most 20 labels" in caplog.text + + +def test_names_are_unique_and_name_the_attribution(): + first, second = sandbox_name({"run_id": "r1"}), sandbox_name({"run_id": "r1"}) + assert first != second and first.startswith("ae-") and first.endswith("-r1") + + +# --- create --- + + +@pytest.mark.asyncio +async def test_create_vm_sends_an_isolated_sandbox_and_returns_its_hosted_urls(): + api = FakeBoatApi() + commands = FakeCommands(handler=docker_ready) + sandbox = await provider(api, commands).create_vm(cpu=1, memory=8192, exposed_ports=[8000, 8000, 9000], + timeout=1800, attribution={"run_id": "r1"}) + request = api.creates[0]["request"] + assert (request.type, request.ttl_seconds, request.no_env, request.snapshots) == ("default", 1800, True, False) + assert [(port.port, port.public) for port in request.hosted_ports] == [(8000, True), (9000, True)] + assert request.labels[PORTS_LABEL] == "8000,9000" and request.labels[f"{ATTRIBUTION_LABEL_PREFIX}run_id"] == "r1" + assert isinstance(sandbox, BoatSandbox) and (sandbox.type, sandbox.mode) == ("boat", "vm") + assert sandbox.tunnel_urls == {8000: api.url(sandbox.sandbox_id, 8000), 9000: api.url(sandbox.sandbox_id, 9000)} + assert sandbox.network_policy == NetworkPolicy() + assert commands.commands == [["bash", "-c", DOCKER_READY]] + assert api.hosts == [] + + +@pytest.mark.asyncio +async def test_a_port_the_create_could_not_host_is_hosted_again(): + api = FakeBoatApi(hosted_port_errors={8000}) + sandbox = await provider(api).create_vm(exposed_ports=[8000], setup_for_gateway=False) + assert api.hosts == [(sandbox.sandbox_id, 8000, True)] + assert sandbox.tunnel_urls == {8000: api.url(sandbox.sandbox_id, 8000)} + + +@pytest.mark.asyncio +async def test_create_sandbox_is_a_vm_with_the_agents_port(): + api = FakeBoatApi() + sandbox = await provider(api, FakeCommands(handler=docker_ready)).create_sandbox(image_name="agent:1", port=8000, + env={"A": "1"}) + assert list(sandbox.tunnel_urls) == [8000] + + +@pytest.mark.asyncio +async def test_an_egress_policy_or_an_image_is_refused_before_anything_is_made(caplog): + api = FakeBoatApi() + with pytest.raises(NetworkPolicyUnsupportedError): + await provider(api).create_vm(network_policy=NetworkPolicy(mode=NetworkMode.ALLOWLIST, allow_hosts=("x.org",))) + with pytest.raises(ValueError, match="image overrides"): + await provider(api).create_vm(image="ubuntu:24.04") + assert api.creates == [] + assert not BoatSandboxProvider.supports_network_policy(NetworkPolicy(mode=NetworkMode.ALLOWLIST)) + with caplog.at_level(logging.WARNING): + await provider(api).create_vm(boot_mode="uefi", setup_for_gateway=False) + assert "Ignoring boot_mode=uefi" in caplog.text + + +@pytest.mark.asyncio +async def test_a_refused_create_retries_with_a_new_key_and_an_unanswered_one_with_the_same(): + api = FakeBoatApi(create_errors=[api_error(429, "rate_limited"), api_error(500, "http_500")]) + sandbox = await provider(api).create_vm(setup_for_gateway=False) + keys = [create["key"] for create in api.creates] + assert keys[0] != keys[1] and keys[1] == keys[2] + # The 500 stands for a lost answer: its sandbox was made, and the retry's answer is another in this fake. + assert sandbox.sandbox_id in api.sandboxes_by_id + + +@pytest.mark.asyncio +async def test_a_request_boat_rejects_raises_at_once(): + api = FakeBoatApi(create_errors=[api_error(403, "trial_machine_class_not_allowed")]) + with pytest.raises(Exception, match="trial_machine_class_not_allowed"): + await provider(api).create_vm(cpu=8, memory=16384, setup_for_gateway=False) + assert len(api.creates) == 1 and api.label_queries == [] + + +@pytest.mark.asyncio +async def test_a_create_that_never_answers_deletes_what_it_may_have_made(monkeypatch): + monkeypatch.setattr(BoatSandboxProvider, "_CREATE_ATTEMPTS", 2) + api = FakeBoatApi(create_errors=[urllib3.exceptions.ProtocolError("reset")] * 2) + with pytest.raises(urllib3.exceptions.ProtocolError): + await provider(api).create_vm(setup_for_gateway=False) + token = api.creates[0]["request"].labels[CREATE_LABEL] + assert api.label_queries == [[f"{CREATE_LABEL}={token}"]] + assert api.sandboxes_by_id == {} and len(api.deletes) == 2 + + +@pytest.mark.asyncio +async def test_a_cancelled_create_deletes_the_sandbox_once_boat_answers(): + api = FakeBoatApi(create_gate=threading.Event()) + task = asyncio.create_task(provider(api).create_vm(setup_for_gateway=False)) + assert await asyncio.to_thread(api.create_entered.wait, 5) + task.cancel() + await asyncio.sleep(0.05) + api.create_gate.set() + with pytest.raises(asyncio.CancelledError): + await task + assert api.sandboxes_by_id == {} and len(api.deletes) == 1 + + +@pytest.mark.asyncio +async def test_a_sandbox_that_fails_setup_is_deleted_before_the_error_is_raised(): + api = FakeBoatApi() + commands = FakeCommands(handler=lambda argv: (1, "", "docker daemon did not become ready\n")) + with pytest.raises(RuntimeError, match="docker daemon did not become ready"): + await provider(api, commands).create_vm(exposed_ports=[8000]) + assert api.sandboxes_by_id == {} and len(api.deletes) == 1 + + +@pytest.mark.asyncio +async def test_a_sandbox_that_fails_to_start_is_deleted(): + api = FakeBoatApi(states=["provisioning", "error"]) + with pytest.raises(RuntimeError, match="state 'error'"): + await provider(api).create_vm(setup_for_gateway=False) + assert api.sandboxes_by_id == {} + + +# --- reconnect and terminate --- + + +@pytest.mark.asyncio +async def test_a_fresh_provider_reconnects_with_the_ports_and_urls_the_labels_record(): + api = FakeBoatApi() + made = await provider(api).create_vm(exposed_ports=[8000, 9000], setup_for_gateway=False) + again = await provider(api).get_sandbox(made.sandbox_id) + assert again.tunnel_urls == made.tunnel_urls and again.network_policy == NetworkPolicy() + assert [host[1:] for host in api.hosts] == [(8000, True), (9000, True)] + + +@pytest.mark.asyncio +async def test_a_gone_or_stopped_sandbox_is_not_found(): + api = FakeBoatApi(states=["ready", "archived"]) + made = await provider(api).create_vm(setup_for_gateway=False) + with pytest.raises(SandboxNotFoundError, match="stopped"): + await provider(api).get_sandbox(made.sandbox_id) + with pytest.raises(SandboxNotFoundError, match="not found"): + await provider(api).get_sandbox("bx_missing") + + +@pytest.mark.asyncio +async def test_terminate_is_idempotent_and_terminate_sandbox_needs_no_handle(): + api = FakeBoatApi() + made = await provider(api).create_vm(setup_for_gateway=False) + await made.terminate() + await made.terminate() + await provider(api).terminate_sandbox(made.sandbox_id) + assert api.deletes == [made.sandbox_id] * 3 + + +@pytest.mark.asyncio +async def test_a_delete_that_keeps_failing_is_raised(monkeypatch): + api = FakeBoatApi() + made = await provider(api).create_vm(setup_for_gateway=False) + + def failing(**kwargs): + raise api_error(500, "http_500") + + monkeypatch.setattr(api, "delete_sandbox", failing) + with pytest.raises(RuntimeError, match="could not be deleted"): + await provider(api).terminate_sandbox(made.sandbox_id) + + +@pytest.mark.asyncio +async def test_close_closes_the_sdk_client(): + api = FakeBoatApi() + await provider(api).close() + assert api.api_client.closed diff --git a/tst/unit/providers/sandbox_providers/boat_sandbox_test.py b/tst/unit/providers/sandbox_providers/boat_sandbox_test.py new file mode 100644 index 000000000..dba96cf13 --- /dev/null +++ b/tst/unit/providers/sandbox_providers/boat_sandbox_test.py @@ -0,0 +1,173 @@ +"""Offline unit tests for a boat sandbox: exec, its failures and cancellation, Docker readiness and file staging.""" + +from __future__ import annotations + +import asyncio +import hashlib +import io +import os +import threading +from types import SimpleNamespace + +import httpx +import pytest +import urllib3 + +from agent_env.providers.sandbox_providers.boat import lifecycle +from agent_env.providers.sandbox_providers.boat import sandbox as sandbox_module +from agent_env.providers.sandbox_providers.boat.sandbox import DOCKER_READY, UNKNOWN_EXIT_CODE, BoatSandbox +from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy + +from tst.unit.providers.sandbox_providers.boat_fakes import FakeBoatApi, FakeCommands, api_error + +RETRY = lifecycle.Retry(attempts=1, base_delay=0, max_delay=0, delete_attempts=1, delete_delay=0, cancel_wait=1) + + +def sandbox(commands: FakeCommands, api: FakeBoatApi | None = None) -> BoatSandbox: + return BoatSandbox(api or FakeBoatApi(), "bx_1", tunnel_urls={}, network_policy=NetworkPolicy(), retry=RETRY, + run_command=commands) + + +@pytest.mark.asyncio +async def test_exec_sends_argv_as_one_quoted_command_and_keeps_the_streams_apart(): + commands = FakeCommands(handler=lambda argv: (3, "out\n", "err\n")) + code, out, err = await sandbox(commands).exec_with_output("echo", "a b", "c'd") + assert commands.commands == [["echo", "a b", "c'd"]] + assert (code, out, err) == (3, "out\n", "err\n") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [urllib3.exceptions.ReadTimeoutError(None, "/", "timed out"), + ConnectionResetError("reset"), api_error(502, "sandbox_direct_failed"), + api_error(429, "IP_RATE_LIMIT_EXCEEDED")]) +async def test_a_dropped_transport_or_a_failing_boat_reads_as_an_unknown_exit(failure): + code, _, err = await sandbox(FakeCommands(handler=lambda argv: failure)).exec_with_output("true") + assert code == UNKNOWN_EXIT_CODE and err + + +@pytest.mark.asyncio +async def test_a_command_boat_refuses_is_raised_not_retried(): + with pytest.raises(Exception, match="machine_not_running"): + await sandbox(FakeCommands(handler=lambda argv: api_error(409, "machine_not_running"))).exec("true") + + +@pytest.mark.asyncio +async def test_an_error_frame_or_a_stream_with_no_exit_reads_as_an_unknown_exit(): + class ErrorFrame(FakeCommands): + def __call__(self, api, sandbox_id, command, cwd=None): + yield {"type": "started", "processId": 7} + yield {"type": "stdout", "data": "partial"} + yield {"type": "error", "message": "process lost"} + + code, out, err = await sandbox(ErrorFrame()).exec_with_output("true") + assert (code, out, err) == (UNKNOWN_EXIT_CODE, "partial", "process lost") + + class NoExit(FakeCommands): + def __call__(self, api, sandbox_id, command, cwd=None): + yield {"type": "started", "processId": 7} + + code, _, err = await sandbox(NoExit()).exec_with_output("true") + assert code == UNKNOWN_EXIT_CODE and "without an exit" in err + + +@pytest.mark.asyncio +async def test_a_cancelled_exec_stops_its_command(): + api = FakeBoatApi() + commands = FakeCommands(gate=threading.Event()) + task = asyncio.create_task(sandbox(commands, api).exec("sleep", "600")) + assert await asyncio.to_thread(commands.started.wait, 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert api.stopped == [("bx_1", 100)] + commands.gate.set() + + +@pytest.mark.asyncio +async def test_a_command_cancelled_before_it_started_is_stopped_when_it_starts(): + api = FakeBoatApi() + release = threading.Event() + + class SlowStart(FakeCommands): + def __call__(self, api, sandbox_id, command, cwd=None): + assert release.wait(5) + yield {"type": "started", "processId": 42} + yield {"type": "exit", "exitCode": 0} + + task = asyncio.create_task(sandbox(SlowStart(), api).exec("true")) + await asyncio.sleep(0.05) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + release.set() + for _ in range(100): + if api.stopped: + break + await asyncio.sleep(0.01) + assert api.stopped == [("bx_1", 42)] + + +@pytest.mark.asyncio +async def test_setup_waits_for_docker_and_needs_compose(): + commands = FakeCommands(handler=lambda argv: (0, "Docker Compose version v5.3.1\n", "")) + await sandbox(commands).setup_vm_for_gateway([8000]) + assert commands.commands == [["bash", "-c", DOCKER_READY]] + failing = FakeCommands(handler=lambda argv: (1, "", "docker daemon did not become ready\n")) + with pytest.raises(RuntimeError, match="not ready in boat sandbox bx_1"): + await sandbox(failing).wait_for_vm() + + +@pytest.fixture +def received(monkeypatch) -> dict[str, bytes]: + """The bodies PUT to upload URLs, by URL, through a transport that answers 200.""" + bodies: dict[str, bytes] = {} + + async def handler(request: httpx.Request) -> httpx.Response: + bodies[str(request.url)] = await request.aread() + return httpx.Response(200, json={"success": True}) + + real = httpx.AsyncClient + monkeypatch.setattr(sandbox_module.httpx, "AsyncClient", + lambda **kwargs: real(transport=httpx.MockTransport(handler), **kwargs)) + return bodies + + +def moved(payload: bytes, digest: str | None = None) -> FakeCommands: + """Commands that answer the move into place with ``digest``, the payload's own by default.""" + sha = digest or hashlib.sha256(payload).hexdigest() + return FakeCommands(handler=lambda argv: (0, f"{sha} {argv[-1].split()[-1]}\n" if "sha256sum" in argv[-1] else "", "")) + + +@pytest.mark.asyncio +async def test_bytes_one_exec_carries_go_in_it(received): + api, commands = FakeBoatApi(), FakeCommands() + await sandbox(commands, api).write_host_file(b"small", "/opt/smoke/small.bin") + assert api.uploads == [] and received == {} + assert any("base64 -d" in argv[-1] for argv in commands.commands) + + +@pytest.mark.asyncio +async def test_more_bytes_go_through_an_upload_url_and_are_moved_into_place_as_root(received): + api, payload = FakeBoatApi(), os.urandom(300_000) + commands = moved(payload) + await sandbox(commands, api).write_host_file(payload, "/opt/smoke/payload.bin") + (staged,) = api.uploads + assert staged.startswith("/tmp/") and list(received.values()) == [payload] + script = commands.commands[-1] + assert script[:3] == ["sudo", "bash", "-c"] and f"mv -f {staged} /opt/smoke/payload.bin" in script[-1] + assert "chown root:root" in script[-1] + + +@pytest.mark.asyncio +async def test_an_upload_that_arrives_changed_is_refused(received): + payload = os.urandom(300_000) + with pytest.raises(RuntimeError, match="does not match the upload"): + await sandbox(moved(payload, digest="0" * 64)).write_host_file(payload, "/opt/smoke/payload.bin") + + +@pytest.mark.asyncio +async def test_an_object_the_store_cannot_sign_streams_through_an_upload_url(received): + payload = os.urandom(9 * 1024 * 1024 + 17) + store = SimpleNamespace(open=lambda url: io.BytesIO(payload)) + await sandbox(moved(payload))._write_unsigned_object(store, "file:///images/env.tar.gz", "/opt/images/env.tar.gz") + assert list(received.values()) == [payload] diff --git a/tst/unit/providers/sandbox_providers/egress_hosts_test.py b/tst/unit/providers/sandbox_providers/egress_hosts_test.py index 44d0f9b18..113e14023 100644 --- a/tst/unit/providers/sandbox_providers/egress_hosts_test.py +++ b/tst/unit/providers/sandbox_providers/egress_hosts_test.py @@ -102,4 +102,5 @@ def test_the_builtin_floor_is_pinned(builtin_registry): assert set(all_sandbox_egress_hosts()) == { "*.modal.host", "*.w.modal.host", "*.e2b.app", "*.sail.box", "*.sandbox.tensorlake.ai", "*.sb.tenki.sh", "*.us.sb.tenki.sh", "*.vercel.run", "*.boxd.sh", "*.daytonaproxy01.net", "*.daytonaproxy01.eu", + "*.on.boat.dev", } 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 e2b7df6f0..ff64b3b1d 100644 --- a/tst/unit/providers/sandbox_providers/network_policy_stamping_test.py +++ b/tst/unit/providers/sandbox_providers/network_policy_stamping_test.py @@ -13,6 +13,7 @@ import pytest +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.daytona.config import Settings as DaytonaSettings from agent_env.providers.sandbox_providers.daytona.provider import DaytonaSandboxProvider @@ -142,6 +143,12 @@ async def _daytona() -> Sandbox: return await provider.create_vm(exposed_ports=[], setup_for_gateway=False) +async def _boat() -> Sandbox: + from tst.unit.providers.sandbox_providers.boat_fakes import SETTINGS, FakeBoatApi + provider = BoatSandboxProvider(SETTINGS, api_factory=lambda settings: FakeBoatApi()) + return await provider.create_vm(exposed_ports=[], setup_for_gateway=False) + + async def _smol_vm() -> Sandbox: pytest.importorskip("smol") from agent_env.providers.sandbox_providers.smol_vm import provider as smol_module @@ -167,6 +174,7 @@ async def _smol_vm() -> Sandbox: "tenki": _tenki, "boxd": _boxd, "daytona": _daytona, + "boat": _boat, "smol_vm": _smol_vm, "local": _local, } diff --git a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py index c1fbad768..86b3742b4 100644 --- a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py +++ b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py @@ -11,6 +11,8 @@ import pytest from agent_env.artifact.artifacts.docker_image import DockerImageArtifact +from agent_env.providers.sandbox_providers.boat.provider import BoatSandboxProvider +from agent_env.providers.sandbox_providers.boat.sandbox import BoatSandbox 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 @@ -42,7 +44,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, ChainedSandboxProvider] + DaytonaSandboxProvider, BoatSandboxProvider, ChainedSandboxProvider] @pytest.mark.parametrize("cls, on_this_machine, creates_vms, sandbox, container, private_network", [ @@ -55,7 +57,8 @@ (TenkiSandboxProvider, False, True, LOADABLE, NAME, False), (BoxdSandboxProvider, False, True, LOADABLE, NAME, False), (DaytonaSandboxProvider, False, True, LOADABLE, NAME, False), -], ids=["local", "modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "boxd", "daytona"]) + (BoatSandboxProvider, False, True, LOADABLE, NAME, False), +], ids=["local", "modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "boxd", "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) @@ -277,7 +280,7 @@ def test_an_image_that_records_no_platform_isnt_checked(): @pytest.mark.parametrize("sandbox", [Sandbox, ModalSandbox, ModalVmSandbox, E2BSandbox, SailVmSandbox, TensorlakeSandbox, - TenkiSandbox, BoxdSandbox, DaytonaSandbox]) + TenkiSandbox, BoxdSandbox, DaytonaSandbox, BoatSandbox]) def test_a_sandbox_runs_linux_amd64_unless_its_provider_sets_another(sandbox): assert sandbox.platform == "linux/amd64" diff --git a/tst/util/capabilities.py b/tst/util/capabilities.py index 4e9116c53..e2b3dc369 100644 --- a/tst/util/capabilities.py +++ b/tst/util/capabilities.py @@ -21,7 +21,7 @@ MODEL_ENDPOINT = "model_endpoint_configured" #: The remote sandbox providers (``modal``, ``modal_vm``, ``e2b``, ``sail_vm``, ``tensorlake``, ``tenki``, -#: ``vercel``, ``boxd``, ``daytona``) need credentials the resolved config may not carry; the local default +#: ``vercel``, ``boxd``, ``daytona``, ``boat``) need credentials the resolved config may not carry; the local default #: never does. REMOTE_SANDBOX = "remote_sandbox" @@ -78,17 +78,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`` sandbox provider, credentials included. + ``tensorlake`` / ``tenki`` / ``vercel`` / ``boxd`` / ``daytona`` / ``boat`` sandbox provider, credentials included. - Modal: any failure to resolve the credentials answers False. E2B, Sail, Tensorlake, Tenki, Vercel, boxd and - Daytona: building the provider resolves ``[sandbox.providers..config]`` and its ``secret:`` references; + 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; 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"): + if provider not in ("modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "daytona", "boat"): raise ValueError(f"unknown remote sandbox provider {provider!r}") - if provider in ("e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "daytona"): + if provider in ("e2b", "sail_vm", "tensorlake", "tenki", "vercel", "boxd", "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 64036be28..8c462bac3 100644 --- a/uv.lock +++ b/uv.lock @@ -72,6 +72,10 @@ aws = [ { name = "boto3" }, { name = "botocore" }, ] +boat = [ + { name = "boat-sdk" }, + { name = "urllib3" }, +] boxd = [ { name = "boxd" }, ] @@ -82,6 +86,7 @@ dev = [ { name = "anyio" }, { name = "av", version = "18.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "av", version = "19.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, + { name = "boat-sdk" }, { name = "boto3" }, { name = "botocore" }, { name = "boxd" }, @@ -111,6 +116,7 @@ dev = [ { name = "tensorlake" }, { name = "tomli-w" }, { name = "transformers" }, + { name = "urllib3" }, { name = "vercel-sandbox" }, ] gcp = [ @@ -148,11 +154,12 @@ vercel = [ [package.metadata] requires-dist = [ { name = "a2a-sdk", specifier = "==0.3.26" }, - { name = "agentenv-framework", extras = ["aws", "boxd", "daytona", "gcp", "openenv", "sail", "tensorlake", "tenki", "trl", "vercel"], marker = "extra == 'dev'" }, + { name = "agentenv-framework", extras = ["aws", "boat", "boxd", "daytona", "gcp", "openenv", "sail", "tensorlake", "tenki", "trl", "vercel"], marker = "extra == 'dev'" }, { name = "agentenv-framework", extras = ["openenv"], marker = "extra == 'trl'" }, { name = "agentenv-framework-protocol", extras = ["computer"], editable = "packages/agentenv-protocol" }, { name = "anyio", marker = "extra == 'vercel'", specifier = ">=4,<5" }, { name = "av", marker = "extra == 'dev'", specifier = ">=17" }, + { name = "boat-sdk", marker = "extra == 'boat'", specifier = "~=1.13.0" }, { name = "boto3", marker = "extra == 'aws'", specifier = ">=1.35.0" }, { name = "botocore", marker = "extra == 'aws'", specifier = ">=1.35.0" }, { name = "boxd", marker = "extra == 'boxd'", specifier = "~=0.2.17" }, @@ -198,10 +205,11 @@ requires-dist = [ { name = "tensorlake", marker = "extra == 'tensorlake'", specifier = "~=0.5.140" }, { name = "tomli-w", marker = "extra == 'dev'", specifier = "==1.2.0" }, { name = "transformers", marker = "extra == 'dev'", specifier = ">=5.2" }, + { name = "urllib3", marker = "extra == 'boat'", specifier = ">=2.1.0,<3" }, { name = "uvicorn", specifier = ">=0.47" }, { name = "vercel-sandbox", marker = "extra == 'vercel'", specifier = ">=0.7.0,<0.8" }, ] -provides-extras = ["openenv", "trl", "aws", "gcp", "sail", "vercel", "tensorlake", "tenki", "boxd", "daytona", "dev", "smol"] +provides-extras = ["openenv", "trl", "aws", "gcp", "sail", "vercel", "tensorlake", "tenki", "boxd", "daytona", "boat", "dev", "smol"] [[package]] name = "agentenv-framework-protocol" @@ -686,6 +694,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/97/53/2a3c7d562271ec6b6e38e7216b3104899f8aa4180c0713cb8aaf69e29cd5/bidict-0.24.1-py3-none-any.whl", hash = "sha256:fd3eaa737917d8a14f4baa391670c433c4e3f6f5fd2cd99d4bf436437f432364", size = 36175, upload-time = "2026-08-25T23:45:51.096Z" }, ] +[[package]] +name = "boat-sdk" +version = "1.13.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pydantic" }, + { name = "python-dateutil" }, + { name = "typing-extensions" }, + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/3c/9b/64dce867b101234a753e18248fda3cae9fd2511034e6b672a406488c14a7/boat_sdk-1.13.0.tar.gz", hash = "sha256:c06a2e3991e8c90f0dca5dfb0408e2e253398008525d1577ca652e1c7f6862e0", size = 228320, upload-time = "2026-10-28T00:43:34.736Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/53/cbe59c0102952fb093f49ffb840f7c29dffe2961e1f54d8239924f0608fb/boat_sdk-1.13.0-py3-none-any.whl", hash = "sha256:259cdd2d4114a165d08f6e4b67616565e25a971bdc90eb76cef8f269cc181ff9", size = 498388, upload-time = "2026-10-28T00:43:33.252Z" }, +] + [[package]] name = "boto3" version = "1.43.38" From 6e99f94fe5076400c5f023774871dcf29f98bdb1 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 28 Oct 2026 15:09:27 -0700 Subject: [PATCH 2/2] fix(sandbox): keep boat's cleanup off busy streams, and close its connections - A create cancelled while it waits to retry deletes what an unanswered attempt made, found by its create label, as a create cancelled mid-call already did. - Commands' output streams run on a thread pool of their own, so streams that fill it can't keep a stop or a delete from running. - The Docker readiness wait is built from named attempts and interval. - close clears the SDK client's urllib3 pool: the SDK's ApiClient has no close, and the end-to-end run's teardown failed on it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../sandbox_providers/boat/lifecycle.py | 31 ++++++++++++----- .../sandbox_providers/boat/provider.py | 8 ++--- .../sandbox_providers/boat/sandbox.py | 28 +++++++++------ .../providers/sandbox_providers/boat_fakes.py | 7 ++-- .../boat_sandbox_provider_test.py | 34 ++++++++++++++++--- .../sandbox_providers/boat_sandbox_test.py | 28 +++++++++++++-- 6 files changed, 104 insertions(+), 32 deletions(-) diff --git a/src/agent_env/providers/sandbox_providers/boat/lifecycle.py b/src/agent_env/providers/sandbox_providers/boat/lifecycle.py index e9de5f841..1cdb550e7 100644 --- a/src/agent_env/providers/sandbox_providers/boat/lifecycle.py +++ b/src/agent_env/providers/sandbox_providers/boat/lifecycle.py @@ -1,7 +1,8 @@ """Calling boat's synchronous SDK from asyncio, and creating and deleting sandboxes without leaking any. -Every SDK call runs on a thread pool of the provider's own, so a long command's output stream never holds a thread of -asyncio's default executor, which the object stores share. A create is idempotent: a retry after a lost answer sends +SDK calls run on two thread pools of the provider's own: one for commands' output streams, which hold a thread for as +long as their command runs, and one for every other call, so neither a busy stream pool nor asyncio's default executor, +which the object stores share, can keep a stop or a delete from running. A create is idempotent: a retry after a lost answer sends the same key and gets the same sandbox, while a refusal (rate limit, no capacity) made nothing, so the next attempt takes a key of its own. A sandbox a create made but nothing holds, because the create failed or was cancelled, is found by its create label and deleted. @@ -27,9 +28,13 @@ T = TypeVar("T") -# Threads for SDK calls, shared by every boat sandbox in the process: an exec holds one for as long as its command runs. -THREADS = 64 -_EXECUTOR = ThreadPoolExecutor(max_workers=THREADS, thread_name_prefix="agent-env-boat") +# Threads shared by every boat sandbox in the process: an exec's output stream holds one of STREAM_THREADS for as long +# as its command runs; creates, deletes, stops and the other short calls run on CALL_THREADS of their own. +STREAM_THREADS = 64 +CALL_THREADS = 16 +THREADS = STREAM_THREADS + CALL_THREADS +_STREAMS = ThreadPoolExecutor(max_workers=STREAM_THREADS, thread_name_prefix="agent-env-boat-stream") +_CALLS = ThreadPoolExecutor(max_workers=CALL_THREADS, thread_name_prefix="agent-env-boat") # Failures of the connection, not answers from boat. NETWORK_ERRORS = (urllib3.exceptions.HTTPError, OSError) # Answers that refuse a create before anything is made: a full rate or concurrency limit, or no machine free. @@ -50,8 +55,13 @@ class Retry: async def call(fn: Callable[..., T], /, *args: Any, **kwargs: Any) -> T: - """``fn(*args, **kwargs)`` on the provider's thread pool.""" - return await asyncio.get_running_loop().run_in_executor(_EXECUTOR, functools.partial(fn, *args, **kwargs)) + """``fn(*args, **kwargs)``, a short SDK call, on the calls' thread pool.""" + return await asyncio.get_running_loop().run_in_executor(_CALLS, functools.partial(fn, *args, **kwargs)) + + +async def stream(fn: Callable[..., T], /, *args: Any, **kwargs: Any) -> T: + """``fn(*args, **kwargs)``, which follows a command's output to its end, on the streams' thread pool.""" + return await asyncio.get_running_loop().run_in_executor(_STREAMS, functools.partial(fn, *args, **kwargs)) def _spawn(cleanup: Awaitable[object]) -> asyncio.Future[Any]: @@ -177,7 +187,12 @@ async def create(api: Any, request: Any, token: str, retry: Retry) -> Any: if attempt + 1 < retry.attempts: delay = min(retry_after(last) or retry.base_delay * 2**attempt, retry.max_delay) logger.warning("boat refused or did not answer a sandbox create (%s); retrying in %ss", last, delay) - await asyncio.sleep(delay) + try: + await asyncio.sleep(delay) + except asyncio.CancelledError: + # An attempt that went unanswered may have made the sandbox, which nothing else knows of. + await settle(delete_labelled(api, token, retry), within=retry.cancel_wait) + raise # Out of attempts: an unanswered create may still have made its sandbox. await settle(delete_labelled(api, token, retry)) assert last is not None diff --git a/src/agent_env/providers/sandbox_providers/boat/provider.py b/src/agent_env/providers/sandbox_providers/boat/provider.py index c2f01a89b..a15d316ce 100644 --- a/src/agent_env/providers/sandbox_providers/boat/provider.py +++ b/src/agent_env/providers/sandbox_providers/boat/provider.py @@ -99,10 +99,10 @@ def from_config(cls, **config: Any) -> Self: return cls(parse_settings(config)) async def close(self) -> None: - """Close the SDK client's connections.""" - api_client = getattr(self._api, "api_client", None) - if api_client is not None: - await lifecycle.call(api_client.close) + """Close the SDK client's connections, which its urllib3 pool keeps; the SDK's own exit leaves them open.""" + pool = getattr(getattr(getattr(self._api, "api_client", None), "rest_client", None), "pool_manager", None) + if pool is not None: + await lifecycle.call(pool.clear) def _handle(self, sandbox_id: str, tunnel_urls: Mapping[int, str]) -> BoatSandbox: # boat applies no egress policy, so every sandbox's is unrestricted. diff --git a/src/agent_env/providers/sandbox_providers/boat/sandbox.py b/src/agent_env/providers/sandbox_providers/boat/sandbox.py index c021d3dfe..d9d99243b 100644 --- a/src/agent_env/providers/sandbox_providers/boat/sandbox.py +++ b/src/agent_env/providers/sandbox_providers/boat/sandbox.py @@ -41,15 +41,6 @@ UNKNOWN_EXIT_CODE = -1 # How long a cancelled exec waits for its command to be stopped before the cancellation is raised. _STOP_WAIT = 30.0 -# The image starts dockerd at boot; the first command can be ahead of it. -DOCKER_READY = r""" -for _ in $(seq 90); do - docker info >/dev/null 2>&1 && exec docker compose version - sleep 1 -done -echo "docker daemon did not become ready" >&2 -exit 1 -""" RunCommand = Callable[..., Iterator[dict[str, Any]]] @@ -122,6 +113,9 @@ class BoatSandbox(VmSandbox): """A boat sandbox with a Docker daemon. ``api`` is its provider's ``BoatApi``.""" type = "boat" + # The image starts dockerd at boot, and the first command can be ahead of it: polled this often, this many times. + _DOCKER_READY_ATTEMPTS = 90 + _DOCKER_READY_INTERVAL_SECONDS = 1 def __init__( self, @@ -153,7 +147,7 @@ async def exec(self, *command: str) -> _CompletedProcess: run = _Run() script = shlex.join(command) try: - frames = await lifecycle.call(_follow, self._api, self.sandbox_id, script, run, self._run_command) + frames = await lifecycle.stream(_follow, self._api, self.sandbox_id, script, run, self._run_command) except asyncio.CancelledError: with run.lock: run.cancelled = True @@ -176,9 +170,21 @@ async def _stop(self, process_id: int) -> None: except (ApiException, *lifecycle.NETWORK_ERRORS) as exc: logger.warning("Stopping command %s in boat sandbox %s failed: %s", process_id, self.sandbox_id, exc) + @classmethod + def docker_ready_script(cls) -> str: + """A script that waits for the Docker daemon, then prints the Compose version, failing when either is missing.""" + return ( + f"for _ in $(seq {cls._DOCKER_READY_ATTEMPTS}); do\n" + " docker info >/dev/null 2>&1 && exec docker compose version\n" + f" sleep {cls._DOCKER_READY_INTERVAL_SECONDS}\n" + "done\n" + 'echo "docker daemon did not become ready" >&2\n' + "exit 1\n" + ) + async def wait_for_vm(self) -> None: """Wait for Docker, which boat starts at boot, and require Compose v2, which ``deploy_env`` uses.""" - code, version, err = await self.exec_with_output("bash", "-c", DOCKER_READY) + code, version, err = await self.exec_with_output("bash", "-c", self.docker_ready_script()) if code != 0: raise RuntimeError( f"Docker with Compose v2 is not ready in boat sandbox {self.sandbox_id}: {(err or version).strip()}" diff --git a/tst/unit/providers/sandbox_providers/boat_fakes.py b/tst/unit/providers/sandbox_providers/boat_fakes.py index 6ae5eb37d..99eab5187 100644 --- a/tst/unit/providers/sandbox_providers/boat_fakes.py +++ b/tst/unit/providers/sandbox_providers/boat_fakes.py @@ -49,11 +49,14 @@ class FakeBoatApi: label_queries: list[list[str]] = field(default_factory=list) stopped: list[tuple[str, int]] = field(default_factory=list) uploads: list[str] = field(default_factory=list) - api_client: SimpleNamespace = field(default_factory=lambda: SimpleNamespace(closed=False)) + # The SDK's ApiClient, as far as close reaches: its REST client's urllib3 pool. + api_client: SimpleNamespace = field( + default_factory=lambda: SimpleNamespace(rest_client=SimpleNamespace(pool_manager=SimpleNamespace(cleared=False)))) _ids: Iterator[int] = field(default_factory=itertools.count) def __post_init__(self) -> None: - self.api_client.close = lambda: setattr(self.api_client, "closed", True) + pool = self.api_client.rest_client.pool_manager + pool.clear = lambda: setattr(pool, "cleared", True) def url(self, sandbox_id: str, port: int) -> str: return f"https://box-{sandbox_id}-{port}.on.boat.dev" diff --git a/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py b/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py index efdd8bcf2..e63d467f5 100644 --- a/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py +++ b/tst/unit/providers/sandbox_providers/boat_sandbox_provider_test.py @@ -25,7 +25,7 @@ sandbox_type, ) from agent_env.providers.sandbox_providers.boat.provider import BoatSandboxProvider -from agent_env.providers.sandbox_providers.boat.sandbox import DOCKER_READY, BoatSandbox +from agent_env.providers.sandbox_providers.boat.sandbox import BoatSandbox from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy, NetworkPolicyUnsupportedError from agent_env.providers.sandbox_providers.sandbox_provider import ( _BUILTIN_SANDBOX_PROVIDERS, @@ -46,7 +46,7 @@ def provider(api: FakeBoatApi, commands: FakeCommands | None = None) -> BoatSand def docker_ready(argv: list[str]) -> tuple[int, str, str]: - return (0, "Docker Compose version v5.3.1\n", "") if argv == ["bash", "-c", DOCKER_READY] else (0, "", "") + return (0, "Docker Compose version v5.3.1\n", "") if argv == ["bash", "-c", BoatSandbox.docker_ready_script()] else (0, "", "") # --- config --- @@ -125,7 +125,7 @@ async def test_create_vm_sends_an_isolated_sandbox_and_returns_its_hosted_urls() assert isinstance(sandbox, BoatSandbox) and (sandbox.type, sandbox.mode) == ("boat", "vm") assert sandbox.tunnel_urls == {8000: api.url(sandbox.sandbox_id, 8000), 9000: api.url(sandbox.sandbox_id, 9000)} assert sandbox.network_policy == NetworkPolicy() - assert commands.commands == [["bash", "-c", DOCKER_READY]] + assert commands.commands == [["bash", "-c", BoatSandbox.docker_ready_script()]] assert api.hosts == [] @@ -188,6 +188,22 @@ async def test_a_create_that_never_answers_deletes_what_it_may_have_made(monkeyp assert api.sandboxes_by_id == {} and len(api.deletes) == 2 +@pytest.mark.asyncio +async def test_a_create_cancelled_while_it_waits_to_retry_deletes_what_the_unanswered_attempt_made(monkeypatch): + monkeypatch.setattr(BoatSandboxProvider, "_RETRY_BASE_DELAY", 30.0) + api = FakeBoatApi(create_errors=[api_error(500, "http_500")]) + task = asyncio.create_task(provider(api).create_vm(setup_for_gateway=False)) + for _ in range(100): + if api.sandboxes_by_id: + break + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + token = api.creates[0]["request"].labels[CREATE_LABEL] + assert api.label_queries == [[f"{CREATE_LABEL}={token}"]] and api.sandboxes_by_id == {} + + @pytest.mark.asyncio async def test_a_cancelled_create_deletes_the_sandbox_once_boat_answers(): api = FakeBoatApi(create_gate=threading.Event()) @@ -264,7 +280,15 @@ def failing(**kwargs): @pytest.mark.asyncio -async def test_close_closes_the_sdk_client(): +async def test_close_closes_the_sdk_clients_connections(): api = FakeBoatApi() await provider(api).close() - assert api.api_client.closed + assert api.api_client.rest_client.pool_manager.cleared + + +@pytest.mark.asyncio +async def test_close_works_on_the_real_sdk_client(): + """The SDK's ApiClient has no close of its own; the provider clears its connection pool instead.""" + built = BoatSandboxProvider(SETTINGS) + await built.close() + await built.close() diff --git a/tst/unit/providers/sandbox_providers/boat_sandbox_test.py b/tst/unit/providers/sandbox_providers/boat_sandbox_test.py index dba96cf13..3bfcd2be9 100644 --- a/tst/unit/providers/sandbox_providers/boat_sandbox_test.py +++ b/tst/unit/providers/sandbox_providers/boat_sandbox_test.py @@ -7,6 +7,7 @@ import io import os import threading +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import httpx @@ -15,7 +16,7 @@ from agent_env.providers.sandbox_providers.boat import lifecycle from agent_env.providers.sandbox_providers.boat import sandbox as sandbox_module -from agent_env.providers.sandbox_providers.boat.sandbox import DOCKER_READY, UNKNOWN_EXIT_CODE, BoatSandbox +from agent_env.providers.sandbox_providers.boat.sandbox import UNKNOWN_EXIT_CODE, BoatSandbox from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy from tst.unit.providers.sandbox_providers.boat_fakes import FakeBoatApi, FakeCommands, api_error @@ -107,11 +108,34 @@ def __call__(self, api, sandbox_id, command, cwd=None): assert api.stopped == [("bx_1", 42)] +@pytest.mark.asyncio +async def test_output_streams_filling_their_pool_leave_stops_and_deletes_a_thread(monkeypatch): + monkeypatch.setattr(lifecycle, "_STREAMS", ThreadPoolExecutor(max_workers=1)) + api, commands = FakeBoatApi(), FakeCommands(gate=threading.Event()) + busy = sandbox(commands, api) + running = asyncio.create_task(busy.exec("sleep", "600")) + assert await asyncio.to_thread(commands.started.wait, 5) + queued = asyncio.create_task(busy.exec("true")) + await asyncio.wait_for(busy.terminate(), 2) + queued.cancel() + running.cancel() + await asyncio.gather(queued, running, return_exceptions=True) + assert api.deletes == ["bx_1"] and ("bx_1", 100) in api.stopped + commands.gate.set() + + +def test_the_readiness_wait_is_built_from_its_named_settings(monkeypatch): + monkeypatch.setattr(BoatSandbox, "_DOCKER_READY_ATTEMPTS", 7) + monkeypatch.setattr(BoatSandbox, "_DOCKER_READY_INTERVAL_SECONDS", 3) + script = BoatSandbox.docker_ready_script() + assert "$(seq 7)" in script and "sleep 3" in script + + @pytest.mark.asyncio async def test_setup_waits_for_docker_and_needs_compose(): commands = FakeCommands(handler=lambda argv: (0, "Docker Compose version v5.3.1\n", "")) await sandbox(commands).setup_vm_for_gateway([8000]) - assert commands.commands == [["bash", "-c", DOCKER_READY]] + assert commands.commands == [["bash", "-c", BoatSandbox.docker_ready_script()]] failing = FakeCommands(handler=lambda argv: (1, "", "docker daemon did not become ready\n")) with pytest.raises(RuntimeError, match="not ready in boat sandbox bx_1"): await sandbox(failing).wait_for_vm()