zozo123 commented on code in PR #71725: URL: https://github.com/apache/airflow/pull/71725#discussion_r4211651785
########## providers/common/ai/src/airflow/providers/common/ai/sandbox/boat.py: ########## @@ -0,0 +1,570 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Boat backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import base64 +import json +import logging +import math +import os +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + _FILE_OP_OUTPUT_CAP, + _FILE_OP_TIMEOUT, + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from boat_sdk.api.boat_api import BoatApi + from boat_sdk.models.command_response import CommandResponse + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +log = logging.getLogger(__name__) + +_DEFAULT_BASE_URL = "https://boat.dev/api/v1"; +_MAX_COMMAND_TIMEOUT = 600 +# The deadline is enforced inside the guest by GNU ``timeout``, which sends SIGTERM and +# then SIGKILL this many seconds later. Boat's own deadline sits _SERVER_TIMEOUT_GRACE +# past that SIGKILL so the wrapper can still print what the command wrote: when Boat +# ends the call first, that output is lost and the sandbox has to be destroyed. +_KILL_AFTER = 5 +_SERVER_TIMEOUT_GRACE = 10 +_MAX_ERROR_DETAIL = 300 +_READY_STATES = frozenset({"ready", "idle", "running"}) +_TERMINAL_STATES = frozenset({"archiving", "archived", "error", "cancelled"}) +_READY_POLL_INTERVAL = 2.0 +_MACHINE_TYPES = frozenset({"small", "default", "large"}) +# CreateSandboxRequest's bound on ttlSeconds: 30 days. +_MAX_TTL_SECONDS = 2_592_000 + + +def _api_error_detail(body: Any) -> str: + """ + Summarize the ``code``/``message`` pair Boat returns in a failed call's body. + + The status code alone hides reasons the task owner has to act on, such as an + account that has no Boat subscription yet. Returns an empty string when the + body is missing or is not the documented JSON error envelope. + """ + if isinstance(body, bytes): + with suppress(UnicodeDecodeError): + body = body.decode("utf-8") + if not isinstance(body, str): + return "" + try: + payload = json.loads(body) + except ValueError: + return "" + if not isinstance(payload, dict): + return "" + error = payload.get("error") + if not isinstance(error, dict): + error = payload + parts = [ + value.strip() + for key in ("code", "message") + if isinstance(value := error.get(key), str) and value.strip() + ] + if len(parts) == 2 and parts[0] == parts[1]: + del parts[1] + detail = ": ".join(parts) + return detail[:_MAX_ERROR_DETAIL] + + +@contextmanager +def _translate_boat_errors( + operation: str, *, recoverable_statuses: frozenset[int] = frozenset() +) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from boat_sdk.exceptions import ApiException + except ImportError: + raise SandboxTerminalError( + 'The Boat SDK is not installed. Install "apache-airflow-providers-common-ai[boat]".' + ) from e + if isinstance(e, ApiException): + status_code = e.status if isinstance(e.status, int) else None + status = f" (HTTP {status_code})" if status_code is not None else "" + detail = _api_error_detail(e.body) + message = f"Boat could not {operation}{status}." + if detail: + message = f"{message} {detail}" + if status_code in recoverable_statuses: + raise SandboxError(message) from e + raise SandboxTerminalError(message) from e + raise SandboxTerminalError(f"Boat could not {operation}: {type(e).__name__}.") from e + + +def _bound_text(text: str, max_bytes: int, *, already_truncated: bool = False) -> tuple[str, bool]: + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text, already_truncated + return encoded[-max_bytes:].decode("utf-8", errors="ignore"), True + + +class BoatSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in a `Boat <https://docs.boat.dev/quickstart>`__ sandbox. + + .. note:: + + Experimental: this can change or be removed in a minor release of this provider. + See :ref:`howto/stability`. + + Boat (formerly Ascii Box) is a hosted cloud-computer API: the Airflow worker + needs only network access and an API key, with no local daemon or host + virtualization. + + **Credentials are ambient.** On first use the backend reads ``BOAT_API_KEY`` + (required) and optional ``BOAT_BASE_URL`` from the worker environment. Modal + reads a ``modal`` connection first, but that connection type is owned by the + Modal provider; a ``boat`` connection type belongs in a future Boat provider, + not in this one. + + Boat cannot enforce a deny-all, per-domain, or CIDR egress policy. ``create`` + therefore refuses a :class:`~airflow.providers.common.ai.sandbox.SandboxSpec` + that asks for ``block_network=True``, ``allow_egress_to``, or + ``allow_egress_to_cidrs``, preserving the fail-closed contract. Pass + ``SandboxSpec(block_network=False)`` (and set that on + :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`) when + open egress is acceptable. + + Writes use Boat's native file API, which accepts only paths that resolve + under ``/home/user`` (where relative paths land) or ``/tmp``. The parent + directory is created first with ``mkdir -p`` in the guest, so a write + anywhere else reaches the model as a recoverable error: ``mkdir``'s own + when the guest user cannot create the parent directory, and one carrying + Boat's ``invalid_path`` code when that directory exists or could be + created. Reads use the inherited shell implementation, because the native + read API takes no size parameter and would land a whole file in worker + memory before ``max_bytes`` could reject it. + + :param machine_type: Boat machine size: ``small``, ``default``, or ``large``. + Default ``"default"``. + :param ttl_seconds: Server-side archive TTL in seconds after which the + sandbox is archived even if the worker never destroyed it: a whole number + from 1 to 2592000 (30 days). Default ``3600``. + :param ready_timeout: Seconds to wait for a newly created sandbox to become + ready. Default ``300``. + :param request_timeout: HTTP timeout in seconds for a Boat API call that answers + at once, such as a status check or a delete, and the time added to the + operation's own for a call that waits on one: a command's deadline, a + create's ``ready_timeout``, or 120 seconds for a file write. Default ``30``. + :param no_env: When ``True`` (default), create a no-env sandbox: none of the + account's stored environment variables, secret files or credentials + reach it, and it is confined so it cannot act on the account or its + other sandboxes. ``False`` attaches the account's ``base`` Boat + environment, which boat-sdk names on every create, and lifts that confinement, so secrets ``SandboxSpec.env`` + never named reach model-written code, which can then act on the account. + """ + + name = "boat" + + def __init__( + self, + *, + machine_type: str = "default", + ttl_seconds: int = 3600, + ready_timeout: float = 300.0, + request_timeout: float = 30.0, + no_env: bool = True, + ) -> None: + if machine_type not in _MACHINE_TYPES: + raise ValueError(f"machine_type must be one of {sorted(_MACHINE_TYPES)}, got {machine_type!r}.") + _validate_positive_finite(ttl_seconds, "ttl_seconds") + # Refused rather than floored: int() would quietly shorten the archive backstop, + # and turn a sub-second value into 0, which the SDK only rejects at create time. + if int(ttl_seconds) != ttl_seconds: + raise ValueError(f"ttl_seconds must be a whole number of seconds, got {ttl_seconds!r}.") + if ttl_seconds > _MAX_TTL_SECONDS: + raise ValueError( + f"ttl_seconds must be at most {_MAX_TTL_SECONDS} (30 days), got {ttl_seconds!r}." + ) + _validate_positive_finite(ready_timeout, "ready_timeout") + _validate_positive_finite(request_timeout, "request_timeout") + if not isinstance(no_env, bool): + raise ValueError(f"no_env must be a boolean, got {no_env!r}.") + self._machine_type = machine_type + self._ttl_seconds = int(ttl_seconds) + self._ready_timeout = ready_timeout + self._request_timeout = request_timeout + self._no_env = no_env + self._boat_api: BoatApi | None = None + self._sandbox_env: dict[str, dict[str, str]] = {} + self._deleted_at_deadline: set[str] = set() + + def _get_api(self) -> BoatApi: + if self._boat_api is not None: + return self._boat_api + with _translate_boat_errors("initialize its client"): + from boat_sdk import ApiClient, Configuration + from boat_sdk.api.boat_api import BoatApi + + api_key = (os.environ.get("BOAT_API_KEY") or "").strip() + if not api_key: + raise SandboxTerminalError("BOAT_API_KEY is not set.") + base_url = (os.environ.get("BOAT_BASE_URL") or _DEFAULT_BASE_URL).rstrip("/") + self._boat_api = BoatApi(ApiClient(Configuration(host=base_url, access_token=api_key))) + return self._boat_api + + def _http_timeout(self, seconds: float) -> float: + """HTTP timeout for a call that waits ``seconds`` on an operation before it answers.""" + return seconds + self._request_timeout + + def _get_state(self, sandbox_id: str, *, timeout: float) -> str: + """ + Return the state Boat reports for a sandbox, read from the raw response. + + Boat reports a cancelled sandbox once with only ``id``, ``state`` and + ``error``, and the SDK's ``Sandbox`` model rejects that body for lacking + fields it requires, so ``BoatApi.get`` would fail before the state could + be seen. + """ + from boat_sdk.exceptions import ApiException + + response = self._get_api().get_without_preload_content(sandbox_id, _request_timeout=timeout) + if not 200 <= response.status <= 299: + raise ApiException.from_response(http_resp=response, body=None, data=None) + return json.loads(response.data)["sandbox"]["state"] + + def _wait_until_ready(self, sandbox_id: str) -> None: + # Not boat_sdk.wait_until_ready: it polls with no HTTP timeout, so one stalled + # response would hold create past ready_timeout indefinitely. + deadline = time.monotonic() + self._ready_timeout + while (remaining := deadline - time.monotonic()) > 0: + with _translate_boat_errors("wait for a sandbox to become ready"): + state = self._get_state(sandbox_id, timeout=min(self._request_timeout, remaining)) + if state in _READY_STATES: + return + if state in _TERMINAL_STATES: + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} entered state {state!r} before it was ready." + ) + time.sleep(min(_READY_POLL_INTERVAL, max(0.0, deadline - time.monotonic()))) + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} was not ready within {self._ready_timeout:g} seconds." + ) + + @staticmethod + def _check_spec(spec: SandboxSpec | None) -> None: + """Refuse a spec this backend cannot carry faithfully, before anything is provisioned.""" + if spec is None: + return + if spec.owner is not None: + # An owner exists so that a later task can attach to the sandbox, and the + # ownership rules live in per-sandbox metadata this backend does not read + # back, so recording one would promise an attach that cannot be checked. + raise SandboxTerminalError( + "SandboxSpec names an owner, but this backend keeps no per-sandbox metadata the " + "ownership rules could be read back from, so a sandbox created here cannot be attached " + "to from another task. Drop owner, or provision the sandbox on a backend that supports " + "attaching, such as ModalSandboxBackend." + ) + if spec.allow_egress_to: + raise SandboxTerminalError( + "The Boat backend cannot apply a per-domain egress allowlist. " + "Drop allow_egress_to, or use a backend with per-domain network rules." + ) + if spec.allow_egress_to_cidrs: + raise SandboxTerminalError( + "The Boat backend cannot apply a CIDR egress allowlist. " + "Drop allow_egress_to_cidrs, or use a backend with network rules." + ) + if spec.block_network: + raise SandboxTerminalError( + "The Boat backend cannot deny outbound network access. Pass " + "SandboxSpec(block_network=False) when open egress is acceptable, or " + "use a backend that can enforce a deny-all policy." + ) + + def create(self, *, spec: SandboxSpec | None = None) -> str: + self._check_spec(spec) + env = dict(spec.env) if spec is not None and spec.env else {} + api = self._get_api() + with _translate_boat_errors("create a sandbox"): + from boat_sdk.models.create_sandbox_request import CreateSandboxRequest + + # The generated request models are typed by their wire aliases. + created = api.create( + create_sandbox_request=CreateSandboxRequest( + type=self._machine_type, + ttlSeconds=self._ttl_seconds, + noEnv=self._no_env, + env=env or None, + ), + _request_timeout=self._http_timeout(self._ready_timeout), + ) + sandbox_id = created.sandbox.id + try: + self._name_sandbox(sandbox_id) + self._wait_until_ready(sandbox_id) + self._sandbox_env[sandbox_id] = env + except BaseException: + # The id has not reached the toolset yet, so nothing else can tear + # this sandbox down. The server-side TTL would archive it eventually, + # but that leaves a billed machine idling for an hour by default. + with suppress(Exception): + self.destroy(sandbox_id) + raise + return sandbox_id + + def _name_sandbox(self, sandbox_id: str) -> None: + """ + Best-effort rename to the ``airflow-sandbox-`` prefix used for correlation. + + Failing to name a sandbox costs nothing at run time, so it must not fail + the create -- but it does cost an operator sweeping for orphans later, + which is why it is logged rather than silently dropped. + """ + from boat_sdk.models.update_sandbox_request import UpdateSandboxRequest + + try: + self._get_api().update( + sandbox_id, + UpdateSandboxRequest(name=_new_sandbox_name()), + _request_timeout=self._request_timeout, + ) + except Exception: + log.warning( + "Could not name Boat sandbox %s; it keeps its server-assigned name and will not " + "match an airflow-sandbox-* orphan sweep.", + sandbox_id, + exc_info=True, + ) + + def _destroy_after_timeout(self, sandbox: str) -> None: + try: + self.destroy(sandbox) + except SandboxError as e: + raise SandboxTerminalError( + "The Boat command timed out and deletion of its sandbox could not be confirmed." + ) from e + self._deleted_at_deadline.add(sandbox) + + @contextmanager + def _terminal_if_deleted_at_deadline(self, sandbox: str, what: str) -> Iterator[None]: + """ + Make a failed file operation terminal when Boat's deadline deleted its sandbox. + + The inherited helpers look only at the exit status, so they would tell the + model to retry against a sandbox that no longer exists. + """ + try: + yield + except SandboxTerminalError: + raise + except SandboxError as e: + if sandbox in self._deleted_at_deadline: + raise SandboxTerminalError(f"The sandbox ended while {what}.") from e + raise + + def run_command( + self, sandbox: str, command: str, *, timeout: float, max_output_bytes: int + ) -> SandboxExecResult: + _validate_positive_finite(timeout, "timeout") + _validate_positive_finite(max_output_bytes, "max_output_bytes") + if timeout > _MAX_COMMAND_TIMEOUT: + raise SandboxError( + f"Boat commands are capped at {_MAX_COMMAND_TIMEOUT} seconds; got timeout={timeout}. " + "Ask for a shorter timeout." + ) + # GNU timeout reads 0 as "no timeout", so round up; a command near the API cap is + # shortened so Boat's own deadline still fits above it. + timeout_seconds = min( + max(1, math.ceil(timeout)), _MAX_COMMAND_TIMEOUT - _KILL_AFTER - _SERVER_TIMEOUT_GRACE + ) + server_timeout = timeout_seconds + _KILL_AFTER + _SERVER_TIMEOUT_GRACE + api = self._get_api() + env = self._sandbox_env.get(sandbox, {}) + if env: + exports = "; ".join( + f"export {shlex.quote(key)}={shlex.quote(value)}" for key, value in env.items() + ) + command = f"{exports}; {command}" + # A login shell, like the one Boat runs this wrapper in, so the functions and + # unexported variables the guest's profile sets still reach the command, with + # the exports applied after them. wait's stderr is dropped because bash reports + # a SIGKILLed job there, quoting the whole command line, exports included. + command = ( + "tmp_dir=$(mktemp -d); trap 'rm -rf \"$tmp_dir\"' EXIT; " + f"timeout --kill-after={_KILL_AFTER} {timeout_seconds} bash -lc {shlex.quote(command)} " + '>"$tmp_dir/stdout" 2>"$tmp_dir/stderr" & ' + 'command_pid=$!; wait "$command_pid" 2>/dev/null; command_status=$?; ' + 'cat "$tmp_dir/stdout"; cat "$tmp_dir/stderr" >&2; exit "$command_status"' Review Comment: Both streams now come back through `tail -c` of `max_output_bytes + 1` instead of `cat` (1bdf2bf1ea), with an executed-wrapper test on a large spool. I left out `ulimit -f`: it caps every file the command writes, so a model writing a large dataset would get SIGXFSZ. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/boat.py: ########## @@ -0,0 +1,570 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Boat backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import base64 +import json +import logging +import math +import os +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + _FILE_OP_OUTPUT_CAP, + _FILE_OP_TIMEOUT, + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from boat_sdk.api.boat_api import BoatApi + from boat_sdk.models.command_response import CommandResponse + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +log = logging.getLogger(__name__) + +_DEFAULT_BASE_URL = "https://boat.dev/api/v1"; +_MAX_COMMAND_TIMEOUT = 600 +# The deadline is enforced inside the guest by GNU ``timeout``, which sends SIGTERM and +# then SIGKILL this many seconds later. Boat's own deadline sits _SERVER_TIMEOUT_GRACE +# past that SIGKILL so the wrapper can still print what the command wrote: when Boat +# ends the call first, that output is lost and the sandbox has to be destroyed. +_KILL_AFTER = 5 +_SERVER_TIMEOUT_GRACE = 10 +_MAX_ERROR_DETAIL = 300 +_READY_STATES = frozenset({"ready", "idle", "running"}) +_TERMINAL_STATES = frozenset({"archiving", "archived", "error", "cancelled"}) +_READY_POLL_INTERVAL = 2.0 +_MACHINE_TYPES = frozenset({"small", "default", "large"}) +# CreateSandboxRequest's bound on ttlSeconds: 30 days. +_MAX_TTL_SECONDS = 2_592_000 + + +def _api_error_detail(body: Any) -> str: + """ + Summarize the ``code``/``message`` pair Boat returns in a failed call's body. + + The status code alone hides reasons the task owner has to act on, such as an + account that has no Boat subscription yet. Returns an empty string when the + body is missing or is not the documented JSON error envelope. + """ + if isinstance(body, bytes): + with suppress(UnicodeDecodeError): + body = body.decode("utf-8") + if not isinstance(body, str): + return "" + try: + payload = json.loads(body) + except ValueError: + return "" + if not isinstance(payload, dict): + return "" + error = payload.get("error") + if not isinstance(error, dict): + error = payload + parts = [ + value.strip() + for key in ("code", "message") + if isinstance(value := error.get(key), str) and value.strip() + ] + if len(parts) == 2 and parts[0] == parts[1]: + del parts[1] + detail = ": ".join(parts) + return detail[:_MAX_ERROR_DETAIL] + + +@contextmanager +def _translate_boat_errors( + operation: str, *, recoverable_statuses: frozenset[int] = frozenset() +) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from boat_sdk.exceptions import ApiException + except ImportError: + raise SandboxTerminalError( + 'The Boat SDK is not installed. Install "apache-airflow-providers-common-ai[boat]".' + ) from e + if isinstance(e, ApiException): + status_code = e.status if isinstance(e.status, int) else None + status = f" (HTTP {status_code})" if status_code is not None else "" + detail = _api_error_detail(e.body) + message = f"Boat could not {operation}{status}." + if detail: + message = f"{message} {detail}" + if status_code in recoverable_statuses: + raise SandboxError(message) from e + raise SandboxTerminalError(message) from e + raise SandboxTerminalError(f"Boat could not {operation}: {type(e).__name__}.") from e + + +def _bound_text(text: str, max_bytes: int, *, already_truncated: bool = False) -> tuple[str, bool]: + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text, already_truncated + return encoded[-max_bytes:].decode("utf-8", errors="ignore"), True + + +class BoatSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in a `Boat <https://docs.boat.dev/quickstart>`__ sandbox. + + .. note:: + + Experimental: this can change or be removed in a minor release of this provider. + See :ref:`howto/stability`. + + Boat (formerly Ascii Box) is a hosted cloud-computer API: the Airflow worker + needs only network access and an API key, with no local daemon or host + virtualization. + + **Credentials are ambient.** On first use the backend reads ``BOAT_API_KEY`` + (required) and optional ``BOAT_BASE_URL`` from the worker environment. Modal + reads a ``modal`` connection first, but that connection type is owned by the + Modal provider; a ``boat`` connection type belongs in a future Boat provider, + not in this one. + + Boat cannot enforce a deny-all, per-domain, or CIDR egress policy. ``create`` + therefore refuses a :class:`~airflow.providers.common.ai.sandbox.SandboxSpec` + that asks for ``block_network=True``, ``allow_egress_to``, or + ``allow_egress_to_cidrs``, preserving the fail-closed contract. Pass + ``SandboxSpec(block_network=False)`` (and set that on + :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`) when + open egress is acceptable. + + Writes use Boat's native file API, which accepts only paths that resolve + under ``/home/user`` (where relative paths land) or ``/tmp``. The parent + directory is created first with ``mkdir -p`` in the guest, so a write + anywhere else reaches the model as a recoverable error: ``mkdir``'s own + when the guest user cannot create the parent directory, and one carrying + Boat's ``invalid_path`` code when that directory exists or could be + created. Reads use the inherited shell implementation, because the native + read API takes no size parameter and would land a whole file in worker + memory before ``max_bytes`` could reject it. + + :param machine_type: Boat machine size: ``small``, ``default``, or ``large``. + Default ``"default"``. + :param ttl_seconds: Server-side archive TTL in seconds after which the + sandbox is archived even if the worker never destroyed it: a whole number + from 1 to 2592000 (30 days). Default ``3600``. + :param ready_timeout: Seconds to wait for a newly created sandbox to become + ready. Default ``300``. + :param request_timeout: HTTP timeout in seconds for a Boat API call that answers + at once, such as a status check or a delete, and the time added to the + operation's own for a call that waits on one: a command's deadline, a + create's ``ready_timeout``, or 120 seconds for a file write. Default ``30``. + :param no_env: When ``True`` (default), create a no-env sandbox: none of the + account's stored environment variables, secret files or credentials + reach it, and it is confined so it cannot act on the account or its + other sandboxes. ``False`` attaches the account's ``base`` Boat + environment, which boat-sdk names on every create, and lifts that confinement, so secrets ``SandboxSpec.env`` + never named reach model-written code, which can then act on the account. + """ + + name = "boat" + + def __init__( + self, + *, + machine_type: str = "default", + ttl_seconds: int = 3600, + ready_timeout: float = 300.0, + request_timeout: float = 30.0, + no_env: bool = True, + ) -> None: + if machine_type not in _MACHINE_TYPES: + raise ValueError(f"machine_type must be one of {sorted(_MACHINE_TYPES)}, got {machine_type!r}.") + _validate_positive_finite(ttl_seconds, "ttl_seconds") + # Refused rather than floored: int() would quietly shorten the archive backstop, + # and turn a sub-second value into 0, which the SDK only rejects at create time. + if int(ttl_seconds) != ttl_seconds: + raise ValueError(f"ttl_seconds must be a whole number of seconds, got {ttl_seconds!r}.") + if ttl_seconds > _MAX_TTL_SECONDS: + raise ValueError( + f"ttl_seconds must be at most {_MAX_TTL_SECONDS} (30 days), got {ttl_seconds!r}." + ) + _validate_positive_finite(ready_timeout, "ready_timeout") + _validate_positive_finite(request_timeout, "request_timeout") + if not isinstance(no_env, bool): + raise ValueError(f"no_env must be a boolean, got {no_env!r}.") + self._machine_type = machine_type + self._ttl_seconds = int(ttl_seconds) + self._ready_timeout = ready_timeout + self._request_timeout = request_timeout + self._no_env = no_env + self._boat_api: BoatApi | None = None + self._sandbox_env: dict[str, dict[str, str]] = {} + self._deleted_at_deadline: set[str] = set() + + def _get_api(self) -> BoatApi: + if self._boat_api is not None: + return self._boat_api + with _translate_boat_errors("initialize its client"): + from boat_sdk import ApiClient, Configuration + from boat_sdk.api.boat_api import BoatApi + + api_key = (os.environ.get("BOAT_API_KEY") or "").strip() + if not api_key: + raise SandboxTerminalError("BOAT_API_KEY is not set.") + base_url = (os.environ.get("BOAT_BASE_URL") or _DEFAULT_BASE_URL).rstrip("/") + self._boat_api = BoatApi(ApiClient(Configuration(host=base_url, access_token=api_key))) + return self._boat_api + + def _http_timeout(self, seconds: float) -> float: + """HTTP timeout for a call that waits ``seconds`` on an operation before it answers.""" + return seconds + self._request_timeout + + def _get_state(self, sandbox_id: str, *, timeout: float) -> str: + """ + Return the state Boat reports for a sandbox, read from the raw response. + + Boat reports a cancelled sandbox once with only ``id``, ``state`` and + ``error``, and the SDK's ``Sandbox`` model rejects that body for lacking + fields it requires, so ``BoatApi.get`` would fail before the state could + be seen. + """ + from boat_sdk.exceptions import ApiException + + response = self._get_api().get_without_preload_content(sandbox_id, _request_timeout=timeout) + if not 200 <= response.status <= 299: + raise ApiException.from_response(http_resp=response, body=None, data=None) + return json.loads(response.data)["sandbox"]["state"] + + def _wait_until_ready(self, sandbox_id: str) -> None: + # Not boat_sdk.wait_until_ready: it polls with no HTTP timeout, so one stalled + # response would hold create past ready_timeout indefinitely. + deadline = time.monotonic() + self._ready_timeout + while (remaining := deadline - time.monotonic()) > 0: + with _translate_boat_errors("wait for a sandbox to become ready"): + state = self._get_state(sandbox_id, timeout=min(self._request_timeout, remaining)) + if state in _READY_STATES: + return + if state in _TERMINAL_STATES: + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} entered state {state!r} before it was ready." + ) + time.sleep(min(_READY_POLL_INTERVAL, max(0.0, deadline - time.monotonic()))) + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} was not ready within {self._ready_timeout:g} seconds." + ) + + @staticmethod + def _check_spec(spec: SandboxSpec | None) -> None: + """Refuse a spec this backend cannot carry faithfully, before anything is provisioned.""" + if spec is None: + return + if spec.owner is not None: + # An owner exists so that a later task can attach to the sandbox, and the + # ownership rules live in per-sandbox metadata this backend does not read + # back, so recording one would promise an attach that cannot be checked. + raise SandboxTerminalError( + "SandboxSpec names an owner, but this backend keeps no per-sandbox metadata the " + "ownership rules could be read back from, so a sandbox created here cannot be attached " + "to from another task. Drop owner, or provision the sandbox on a backend that supports " + "attaching, such as ModalSandboxBackend." + ) + if spec.allow_egress_to: + raise SandboxTerminalError( + "The Boat backend cannot apply a per-domain egress allowlist. " + "Drop allow_egress_to, or use a backend with per-domain network rules." + ) + if spec.allow_egress_to_cidrs: + raise SandboxTerminalError( + "The Boat backend cannot apply a CIDR egress allowlist. " + "Drop allow_egress_to_cidrs, or use a backend with network rules." + ) + if spec.block_network: + raise SandboxTerminalError( + "The Boat backend cannot deny outbound network access. Pass " + "SandboxSpec(block_network=False) when open egress is acceptable, or " + "use a backend that can enforce a deny-all policy." + ) + + def create(self, *, spec: SandboxSpec | None = None) -> str: + self._check_spec(spec) + env = dict(spec.env) if spec is not None and spec.env else {} + api = self._get_api() + with _translate_boat_errors("create a sandbox"): + from boat_sdk.models.create_sandbox_request import CreateSandboxRequest + + # The generated request models are typed by their wire aliases. + created = api.create( + create_sandbox_request=CreateSandboxRequest( + type=self._machine_type, + ttlSeconds=self._ttl_seconds, + noEnv=self._no_env, Review Comment: Creates now pass `snapshots=False` (3b755b07e4). Measured live, Boat then reports no snapshot, and the TTL text in `backends.rst` and the `index.rst` row now say the disk is erased. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/boat.py: ########## @@ -0,0 +1,570 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Boat backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import base64 +import json +import logging +import math +import os +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + _FILE_OP_OUTPUT_CAP, + _FILE_OP_TIMEOUT, + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from boat_sdk.api.boat_api import BoatApi + from boat_sdk.models.command_response import CommandResponse + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +log = logging.getLogger(__name__) + +_DEFAULT_BASE_URL = "https://boat.dev/api/v1"; +_MAX_COMMAND_TIMEOUT = 600 +# The deadline is enforced inside the guest by GNU ``timeout``, which sends SIGTERM and +# then SIGKILL this many seconds later. Boat's own deadline sits _SERVER_TIMEOUT_GRACE +# past that SIGKILL so the wrapper can still print what the command wrote: when Boat +# ends the call first, that output is lost and the sandbox has to be destroyed. +_KILL_AFTER = 5 +_SERVER_TIMEOUT_GRACE = 10 +_MAX_ERROR_DETAIL = 300 +_READY_STATES = frozenset({"ready", "idle", "running"}) +_TERMINAL_STATES = frozenset({"archiving", "archived", "error", "cancelled"}) +_READY_POLL_INTERVAL = 2.0 +_MACHINE_TYPES = frozenset({"small", "default", "large"}) +# CreateSandboxRequest's bound on ttlSeconds: 30 days. +_MAX_TTL_SECONDS = 2_592_000 + + +def _api_error_detail(body: Any) -> str: + """ + Summarize the ``code``/``message`` pair Boat returns in a failed call's body. + + The status code alone hides reasons the task owner has to act on, such as an + account that has no Boat subscription yet. Returns an empty string when the + body is missing or is not the documented JSON error envelope. + """ + if isinstance(body, bytes): + with suppress(UnicodeDecodeError): + body = body.decode("utf-8") + if not isinstance(body, str): + return "" + try: + payload = json.loads(body) + except ValueError: + return "" + if not isinstance(payload, dict): + return "" + error = payload.get("error") + if not isinstance(error, dict): + error = payload + parts = [ + value.strip() + for key in ("code", "message") + if isinstance(value := error.get(key), str) and value.strip() + ] + if len(parts) == 2 and parts[0] == parts[1]: + del parts[1] + detail = ": ".join(parts) + return detail[:_MAX_ERROR_DETAIL] + + +@contextmanager +def _translate_boat_errors( + operation: str, *, recoverable_statuses: frozenset[int] = frozenset() +) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from boat_sdk.exceptions import ApiException + except ImportError: + raise SandboxTerminalError( + 'The Boat SDK is not installed. Install "apache-airflow-providers-common-ai[boat]".' + ) from e + if isinstance(e, ApiException): + status_code = e.status if isinstance(e.status, int) else None + status = f" (HTTP {status_code})" if status_code is not None else "" + detail = _api_error_detail(e.body) + message = f"Boat could not {operation}{status}." + if detail: + message = f"{message} {detail}" + if status_code in recoverable_statuses: + raise SandboxError(message) from e + raise SandboxTerminalError(message) from e + raise SandboxTerminalError(f"Boat could not {operation}: {type(e).__name__}.") from e + + +def _bound_text(text: str, max_bytes: int, *, already_truncated: bool = False) -> tuple[str, bool]: + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text, already_truncated + return encoded[-max_bytes:].decode("utf-8", errors="ignore"), True + + +class BoatSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in a `Boat <https://docs.boat.dev/quickstart>`__ sandbox. + + .. note:: + + Experimental: this can change or be removed in a minor release of this provider. + See :ref:`howto/stability`. + + Boat (formerly Ascii Box) is a hosted cloud-computer API: the Airflow worker + needs only network access and an API key, with no local daemon or host + virtualization. + + **Credentials are ambient.** On first use the backend reads ``BOAT_API_KEY`` + (required) and optional ``BOAT_BASE_URL`` from the worker environment. Modal + reads a ``modal`` connection first, but that connection type is owned by the + Modal provider; a ``boat`` connection type belongs in a future Boat provider, + not in this one. + + Boat cannot enforce a deny-all, per-domain, or CIDR egress policy. ``create`` + therefore refuses a :class:`~airflow.providers.common.ai.sandbox.SandboxSpec` + that asks for ``block_network=True``, ``allow_egress_to``, or + ``allow_egress_to_cidrs``, preserving the fail-closed contract. Pass + ``SandboxSpec(block_network=False)`` (and set that on + :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`) when + open egress is acceptable. + + Writes use Boat's native file API, which accepts only paths that resolve + under ``/home/user`` (where relative paths land) or ``/tmp``. The parent + directory is created first with ``mkdir -p`` in the guest, so a write + anywhere else reaches the model as a recoverable error: ``mkdir``'s own + when the guest user cannot create the parent directory, and one carrying + Boat's ``invalid_path`` code when that directory exists or could be + created. Reads use the inherited shell implementation, because the native + read API takes no size parameter and would land a whole file in worker + memory before ``max_bytes`` could reject it. + + :param machine_type: Boat machine size: ``small``, ``default``, or ``large``. + Default ``"default"``. + :param ttl_seconds: Server-side archive TTL in seconds after which the + sandbox is archived even if the worker never destroyed it: a whole number + from 1 to 2592000 (30 days). Default ``3600``. + :param ready_timeout: Seconds to wait for a newly created sandbox to become + ready. Default ``300``. + :param request_timeout: HTTP timeout in seconds for a Boat API call that answers + at once, such as a status check or a delete, and the time added to the + operation's own for a call that waits on one: a command's deadline, a + create's ``ready_timeout``, or 120 seconds for a file write. Default ``30``. + :param no_env: When ``True`` (default), create a no-env sandbox: none of the + account's stored environment variables, secret files or credentials + reach it, and it is confined so it cannot act on the account or its + other sandboxes. ``False`` attaches the account's ``base`` Boat + environment, which boat-sdk names on every create, and lifts that confinement, so secrets ``SandboxSpec.env`` + never named reach model-written code, which can then act on the account. + """ + + name = "boat" + + def __init__( + self, + *, + machine_type: str = "default", + ttl_seconds: int = 3600, + ready_timeout: float = 300.0, + request_timeout: float = 30.0, + no_env: bool = True, + ) -> None: + if machine_type not in _MACHINE_TYPES: + raise ValueError(f"machine_type must be one of {sorted(_MACHINE_TYPES)}, got {machine_type!r}.") + _validate_positive_finite(ttl_seconds, "ttl_seconds") + # Refused rather than floored: int() would quietly shorten the archive backstop, + # and turn a sub-second value into 0, which the SDK only rejects at create time. + if int(ttl_seconds) != ttl_seconds: + raise ValueError(f"ttl_seconds must be a whole number of seconds, got {ttl_seconds!r}.") + if ttl_seconds > _MAX_TTL_SECONDS: + raise ValueError( + f"ttl_seconds must be at most {_MAX_TTL_SECONDS} (30 days), got {ttl_seconds!r}." + ) + _validate_positive_finite(ready_timeout, "ready_timeout") + _validate_positive_finite(request_timeout, "request_timeout") + if not isinstance(no_env, bool): + raise ValueError(f"no_env must be a boolean, got {no_env!r}.") + self._machine_type = machine_type + self._ttl_seconds = int(ttl_seconds) + self._ready_timeout = ready_timeout + self._request_timeout = request_timeout + self._no_env = no_env + self._boat_api: BoatApi | None = None + self._sandbox_env: dict[str, dict[str, str]] = {} + self._deleted_at_deadline: set[str] = set() + + def _get_api(self) -> BoatApi: + if self._boat_api is not None: + return self._boat_api + with _translate_boat_errors("initialize its client"): + from boat_sdk import ApiClient, Configuration + from boat_sdk.api.boat_api import BoatApi + + api_key = (os.environ.get("BOAT_API_KEY") or "").strip() + if not api_key: + raise SandboxTerminalError("BOAT_API_KEY is not set.") + base_url = (os.environ.get("BOAT_BASE_URL") or _DEFAULT_BASE_URL).rstrip("/") + self._boat_api = BoatApi(ApiClient(Configuration(host=base_url, access_token=api_key))) + return self._boat_api + + def _http_timeout(self, seconds: float) -> float: + """HTTP timeout for a call that waits ``seconds`` on an operation before it answers.""" + return seconds + self._request_timeout + + def _get_state(self, sandbox_id: str, *, timeout: float) -> str: + """ + Return the state Boat reports for a sandbox, read from the raw response. + + Boat reports a cancelled sandbox once with only ``id``, ``state`` and + ``error``, and the SDK's ``Sandbox`` model rejects that body for lacking + fields it requires, so ``BoatApi.get`` would fail before the state could + be seen. + """ + from boat_sdk.exceptions import ApiException + + response = self._get_api().get_without_preload_content(sandbox_id, _request_timeout=timeout) + if not 200 <= response.status <= 299: + raise ApiException.from_response(http_resp=response, body=None, data=None) + return json.loads(response.data)["sandbox"]["state"] + + def _wait_until_ready(self, sandbox_id: str) -> None: + # Not boat_sdk.wait_until_ready: it polls with no HTTP timeout, so one stalled + # response would hold create past ready_timeout indefinitely. + deadline = time.monotonic() + self._ready_timeout + while (remaining := deadline - time.monotonic()) > 0: + with _translate_boat_errors("wait for a sandbox to become ready"): + state = self._get_state(sandbox_id, timeout=min(self._request_timeout, remaining)) + if state in _READY_STATES: + return + if state in _TERMINAL_STATES: + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} entered state {state!r} before it was ready." + ) + time.sleep(min(_READY_POLL_INTERVAL, max(0.0, deadline - time.monotonic()))) + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} was not ready within {self._ready_timeout:g} seconds." + ) + + @staticmethod + def _check_spec(spec: SandboxSpec | None) -> None: + """Refuse a spec this backend cannot carry faithfully, before anything is provisioned.""" + if spec is None: + return + if spec.owner is not None: + # An owner exists so that a later task can attach to the sandbox, and the + # ownership rules live in per-sandbox metadata this backend does not read + # back, so recording one would promise an attach that cannot be checked. + raise SandboxTerminalError( + "SandboxSpec names an owner, but this backend keeps no per-sandbox metadata the " + "ownership rules could be read back from, so a sandbox created here cannot be attached " + "to from another task. Drop owner, or provision the sandbox on a backend that supports " + "attaching, such as ModalSandboxBackend." + ) + if spec.allow_egress_to: + raise SandboxTerminalError( + "The Boat backend cannot apply a per-domain egress allowlist. " + "Drop allow_egress_to, or use a backend with per-domain network rules." + ) + if spec.allow_egress_to_cidrs: + raise SandboxTerminalError( + "The Boat backend cannot apply a CIDR egress allowlist. " + "Drop allow_egress_to_cidrs, or use a backend with network rules." + ) + if spec.block_network: + raise SandboxTerminalError( + "The Boat backend cannot deny outbound network access. Pass " + "SandboxSpec(block_network=False) when open egress is acceptable, or " + "use a backend that can enforce a deny-all policy." + ) + + def create(self, *, spec: SandboxSpec | None = None) -> str: + self._check_spec(spec) + env = dict(spec.env) if spec is not None and spec.env else {} + api = self._get_api() + with _translate_boat_errors("create a sandbox"): + from boat_sdk.models.create_sandbox_request import CreateSandboxRequest + + # The generated request models are typed by their wire aliases. + created = api.create( + create_sandbox_request=CreateSandboxRequest( + type=self._machine_type, + ttlSeconds=self._ttl_seconds, + noEnv=self._no_env, + env=env or None, + ), + _request_timeout=self._http_timeout(self._ready_timeout), Review Comment: Provisioning now runs on one monotonic deadline shared by the create call and the readiness poll, and each create carries an idempotency key with one retry on a lost response (a5159106d0, e38c3b1bf0, 909945198f). Live, a retry with the same key returned the same sandbox. urllib3's own resends, including Retry-After sleeps, are off so each call stays within its timeout (5f9bd90ee6, 0288dc04cd). _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/boat.py: ########## @@ -0,0 +1,570 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Boat backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import base64 +import json +import logging +import math +import os +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + _FILE_OP_OUTPUT_CAP, + _FILE_OP_TIMEOUT, + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from boat_sdk.api.boat_api import BoatApi + from boat_sdk.models.command_response import CommandResponse + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +log = logging.getLogger(__name__) + +_DEFAULT_BASE_URL = "https://boat.dev/api/v1"; +_MAX_COMMAND_TIMEOUT = 600 +# The deadline is enforced inside the guest by GNU ``timeout``, which sends SIGTERM and +# then SIGKILL this many seconds later. Boat's own deadline sits _SERVER_TIMEOUT_GRACE +# past that SIGKILL so the wrapper can still print what the command wrote: when Boat +# ends the call first, that output is lost and the sandbox has to be destroyed. +_KILL_AFTER = 5 +_SERVER_TIMEOUT_GRACE = 10 +_MAX_ERROR_DETAIL = 300 +_READY_STATES = frozenset({"ready", "idle", "running"}) +_TERMINAL_STATES = frozenset({"archiving", "archived", "error", "cancelled"}) +_READY_POLL_INTERVAL = 2.0 +_MACHINE_TYPES = frozenset({"small", "default", "large"}) +# CreateSandboxRequest's bound on ttlSeconds: 30 days. +_MAX_TTL_SECONDS = 2_592_000 + + +def _api_error_detail(body: Any) -> str: + """ + Summarize the ``code``/``message`` pair Boat returns in a failed call's body. + + The status code alone hides reasons the task owner has to act on, such as an + account that has no Boat subscription yet. Returns an empty string when the + body is missing or is not the documented JSON error envelope. + """ + if isinstance(body, bytes): + with suppress(UnicodeDecodeError): + body = body.decode("utf-8") + if not isinstance(body, str): + return "" + try: + payload = json.loads(body) + except ValueError: + return "" + if not isinstance(payload, dict): + return "" + error = payload.get("error") + if not isinstance(error, dict): + error = payload + parts = [ + value.strip() + for key in ("code", "message") + if isinstance(value := error.get(key), str) and value.strip() + ] + if len(parts) == 2 and parts[0] == parts[1]: + del parts[1] + detail = ": ".join(parts) + return detail[:_MAX_ERROR_DETAIL] + + +@contextmanager +def _translate_boat_errors( + operation: str, *, recoverable_statuses: frozenset[int] = frozenset() +) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from boat_sdk.exceptions import ApiException + except ImportError: + raise SandboxTerminalError( + 'The Boat SDK is not installed. Install "apache-airflow-providers-common-ai[boat]".' + ) from e + if isinstance(e, ApiException): + status_code = e.status if isinstance(e.status, int) else None + status = f" (HTTP {status_code})" if status_code is not None else "" + detail = _api_error_detail(e.body) + message = f"Boat could not {operation}{status}." + if detail: + message = f"{message} {detail}" + if status_code in recoverable_statuses: + raise SandboxError(message) from e + raise SandboxTerminalError(message) from e + raise SandboxTerminalError(f"Boat could not {operation}: {type(e).__name__}.") from e + + +def _bound_text(text: str, max_bytes: int, *, already_truncated: bool = False) -> tuple[str, bool]: + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text, already_truncated + return encoded[-max_bytes:].decode("utf-8", errors="ignore"), True + + +class BoatSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in a `Boat <https://docs.boat.dev/quickstart>`__ sandbox. + + .. note:: + + Experimental: this can change or be removed in a minor release of this provider. + See :ref:`howto/stability`. + + Boat (formerly Ascii Box) is a hosted cloud-computer API: the Airflow worker + needs only network access and an API key, with no local daemon or host + virtualization. + + **Credentials are ambient.** On first use the backend reads ``BOAT_API_KEY`` + (required) and optional ``BOAT_BASE_URL`` from the worker environment. Modal + reads a ``modal`` connection first, but that connection type is owned by the + Modal provider; a ``boat`` connection type belongs in a future Boat provider, + not in this one. + + Boat cannot enforce a deny-all, per-domain, or CIDR egress policy. ``create`` + therefore refuses a :class:`~airflow.providers.common.ai.sandbox.SandboxSpec` + that asks for ``block_network=True``, ``allow_egress_to``, or + ``allow_egress_to_cidrs``, preserving the fail-closed contract. Pass + ``SandboxSpec(block_network=False)`` (and set that on + :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`) when + open egress is acceptable. + + Writes use Boat's native file API, which accepts only paths that resolve + under ``/home/user`` (where relative paths land) or ``/tmp``. The parent + directory is created first with ``mkdir -p`` in the guest, so a write + anywhere else reaches the model as a recoverable error: ``mkdir``'s own + when the guest user cannot create the parent directory, and one carrying + Boat's ``invalid_path`` code when that directory exists or could be + created. Reads use the inherited shell implementation, because the native + read API takes no size parameter and would land a whole file in worker + memory before ``max_bytes`` could reject it. + + :param machine_type: Boat machine size: ``small``, ``default``, or ``large``. + Default ``"default"``. + :param ttl_seconds: Server-side archive TTL in seconds after which the + sandbox is archived even if the worker never destroyed it: a whole number + from 1 to 2592000 (30 days). Default ``3600``. + :param ready_timeout: Seconds to wait for a newly created sandbox to become + ready. Default ``300``. + :param request_timeout: HTTP timeout in seconds for a Boat API call that answers + at once, such as a status check or a delete, and the time added to the + operation's own for a call that waits on one: a command's deadline, a + create's ``ready_timeout``, or 120 seconds for a file write. Default ``30``. + :param no_env: When ``True`` (default), create a no-env sandbox: none of the + account's stored environment variables, secret files or credentials + reach it, and it is confined so it cannot act on the account or its + other sandboxes. ``False`` attaches the account's ``base`` Boat + environment, which boat-sdk names on every create, and lifts that confinement, so secrets ``SandboxSpec.env`` + never named reach model-written code, which can then act on the account. + """ + + name = "boat" + + def __init__( + self, + *, + machine_type: str = "default", + ttl_seconds: int = 3600, + ready_timeout: float = 300.0, + request_timeout: float = 30.0, + no_env: bool = True, + ) -> None: + if machine_type not in _MACHINE_TYPES: + raise ValueError(f"machine_type must be one of {sorted(_MACHINE_TYPES)}, got {machine_type!r}.") + _validate_positive_finite(ttl_seconds, "ttl_seconds") + # Refused rather than floored: int() would quietly shorten the archive backstop, + # and turn a sub-second value into 0, which the SDK only rejects at create time. + if int(ttl_seconds) != ttl_seconds: + raise ValueError(f"ttl_seconds must be a whole number of seconds, got {ttl_seconds!r}.") + if ttl_seconds > _MAX_TTL_SECONDS: + raise ValueError( + f"ttl_seconds must be at most {_MAX_TTL_SECONDS} (30 days), got {ttl_seconds!r}." + ) + _validate_positive_finite(ready_timeout, "ready_timeout") + _validate_positive_finite(request_timeout, "request_timeout") + if not isinstance(no_env, bool): + raise ValueError(f"no_env must be a boolean, got {no_env!r}.") + self._machine_type = machine_type + self._ttl_seconds = int(ttl_seconds) + self._ready_timeout = ready_timeout + self._request_timeout = request_timeout + self._no_env = no_env + self._boat_api: BoatApi | None = None + self._sandbox_env: dict[str, dict[str, str]] = {} + self._deleted_at_deadline: set[str] = set() + + def _get_api(self) -> BoatApi: + if self._boat_api is not None: + return self._boat_api + with _translate_boat_errors("initialize its client"): + from boat_sdk import ApiClient, Configuration + from boat_sdk.api.boat_api import BoatApi + + api_key = (os.environ.get("BOAT_API_KEY") or "").strip() + if not api_key: + raise SandboxTerminalError("BOAT_API_KEY is not set.") + base_url = (os.environ.get("BOAT_BASE_URL") or _DEFAULT_BASE_URL).rstrip("/") + self._boat_api = BoatApi(ApiClient(Configuration(host=base_url, access_token=api_key))) + return self._boat_api + + def _http_timeout(self, seconds: float) -> float: + """HTTP timeout for a call that waits ``seconds`` on an operation before it answers.""" + return seconds + self._request_timeout + + def _get_state(self, sandbox_id: str, *, timeout: float) -> str: + """ + Return the state Boat reports for a sandbox, read from the raw response. + + Boat reports a cancelled sandbox once with only ``id``, ``state`` and + ``error``, and the SDK's ``Sandbox`` model rejects that body for lacking + fields it requires, so ``BoatApi.get`` would fail before the state could + be seen. + """ + from boat_sdk.exceptions import ApiException + + response = self._get_api().get_without_preload_content(sandbox_id, _request_timeout=timeout) + if not 200 <= response.status <= 299: + raise ApiException.from_response(http_resp=response, body=None, data=None) + return json.loads(response.data)["sandbox"]["state"] + + def _wait_until_ready(self, sandbox_id: str) -> None: + # Not boat_sdk.wait_until_ready: it polls with no HTTP timeout, so one stalled + # response would hold create past ready_timeout indefinitely. + deadline = time.monotonic() + self._ready_timeout + while (remaining := deadline - time.monotonic()) > 0: + with _translate_boat_errors("wait for a sandbox to become ready"): + state = self._get_state(sandbox_id, timeout=min(self._request_timeout, remaining)) + if state in _READY_STATES: + return + if state in _TERMINAL_STATES: + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} entered state {state!r} before it was ready." + ) + time.sleep(min(_READY_POLL_INTERVAL, max(0.0, deadline - time.monotonic()))) + raise SandboxTerminalError( + f"Boat sandbox {sandbox_id!r} was not ready within {self._ready_timeout:g} seconds." + ) + + @staticmethod + def _check_spec(spec: SandboxSpec | None) -> None: + """Refuse a spec this backend cannot carry faithfully, before anything is provisioned.""" + if spec is None: + return + if spec.owner is not None: + # An owner exists so that a later task can attach to the sandbox, and the + # ownership rules live in per-sandbox metadata this backend does not read + # back, so recording one would promise an attach that cannot be checked. + raise SandboxTerminalError( + "SandboxSpec names an owner, but this backend keeps no per-sandbox metadata the " + "ownership rules could be read back from, so a sandbox created here cannot be attached " + "to from another task. Drop owner, or provision the sandbox on a backend that supports " + "attaching, such as ModalSandboxBackend." + ) + if spec.allow_egress_to: + raise SandboxTerminalError( + "The Boat backend cannot apply a per-domain egress allowlist. " + "Drop allow_egress_to, or use a backend with per-domain network rules." + ) + if spec.allow_egress_to_cidrs: + raise SandboxTerminalError( + "The Boat backend cannot apply a CIDR egress allowlist. " + "Drop allow_egress_to_cidrs, or use a backend with network rules." + ) + if spec.block_network: + raise SandboxTerminalError( + "The Boat backend cannot deny outbound network access. Pass " + "SandboxSpec(block_network=False) when open egress is acceptable, or " + "use a backend that can enforce a deny-all policy." + ) + + def create(self, *, spec: SandboxSpec | None = None) -> str: + self._check_spec(spec) + env = dict(spec.env) if spec is not None and spec.env else {} + api = self._get_api() + with _translate_boat_errors("create a sandbox"): + from boat_sdk.models.create_sandbox_request import CreateSandboxRequest + + # The generated request models are typed by their wire aliases. + created = api.create( + create_sandbox_request=CreateSandboxRequest( + type=self._machine_type, + ttlSeconds=self._ttl_seconds, + noEnv=self._no_env, + env=env or None, + ), + _request_timeout=self._http_timeout(self._ready_timeout), + ) + sandbox_id = created.sandbox.id + try: + self._name_sandbox(sandbox_id) + self._wait_until_ready(sandbox_id) + self._sandbox_env[sandbox_id] = env + except BaseException: + # The id has not reached the toolset yet, so nothing else can tear + # this sandbox down. The server-side TTL would archive it eventually, + # but that leaves a billed machine idling for an hour by default. + with suppress(Exception): + self.destroy(sandbox_id) + raise + return sandbox_id + + def _name_sandbox(self, sandbox_id: str) -> None: + """ + Best-effort rename to the ``airflow-sandbox-`` prefix used for correlation. + + Failing to name a sandbox costs nothing at run time, so it must not fail + the create -- but it does cost an operator sweeping for orphans later, + which is why it is logged rather than silently dropped. + """ + from boat_sdk.models.update_sandbox_request import UpdateSandboxRequest + + try: + self._get_api().update( + sandbox_id, + UpdateSandboxRequest(name=_new_sandbox_name()), + _request_timeout=self._request_timeout, + ) + except Exception: + log.warning( + "Could not name Boat sandbox %s; it keeps its server-assigned name and will not " + "match an airflow-sandbox-* orphan sweep.", + sandbox_id, + exc_info=True, + ) + + def _destroy_after_timeout(self, sandbox: str) -> None: + try: + self.destroy(sandbox) + except SandboxError as e: + raise SandboxTerminalError( + "The Boat command timed out and deletion of its sandbox could not be confirmed." + ) from e + self._deleted_at_deadline.add(sandbox) + + @contextmanager + def _terminal_if_deleted_at_deadline(self, sandbox: str, what: str) -> Iterator[None]: + """ + Make a failed file operation terminal when Boat's deadline deleted its sandbox. + + The inherited helpers look only at the exit status, so they would tell the + model to retry against a sandbox that no longer exists. + """ + try: + yield + except SandboxTerminalError: + raise + except SandboxError as e: + if sandbox in self._deleted_at_deadline: + raise SandboxTerminalError(f"The sandbox ended while {what}.") from e + raise + + def run_command( + self, sandbox: str, command: str, *, timeout: float, max_output_bytes: int + ) -> SandboxExecResult: + _validate_positive_finite(timeout, "timeout") + _validate_positive_finite(max_output_bytes, "max_output_bytes") + if timeout > _MAX_COMMAND_TIMEOUT: Review Comment: Every request above the cap now clamps to 585 s and is reported through `applied_timeout`; nothing is refused (7f0b6596da). _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/boat.py: ########## @@ -0,0 +1,570 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Boat backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import base64 +import json +import logging +import math +import os +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + _FILE_OP_OUTPUT_CAP, + _FILE_OP_TIMEOUT, + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from boat_sdk.api.boat_api import BoatApi + from boat_sdk.models.command_response import CommandResponse + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +log = logging.getLogger(__name__) + +_DEFAULT_BASE_URL = "https://boat.dev/api/v1"; +_MAX_COMMAND_TIMEOUT = 600 +# The deadline is enforced inside the guest by GNU ``timeout``, which sends SIGTERM and +# then SIGKILL this many seconds later. Boat's own deadline sits _SERVER_TIMEOUT_GRACE +# past that SIGKILL so the wrapper can still print what the command wrote: when Boat +# ends the call first, that output is lost and the sandbox has to be destroyed. +_KILL_AFTER = 5 +_SERVER_TIMEOUT_GRACE = 10 +_MAX_ERROR_DETAIL = 300 +_READY_STATES = frozenset({"ready", "idle", "running"}) +_TERMINAL_STATES = frozenset({"archiving", "archived", "error", "cancelled"}) +_READY_POLL_INTERVAL = 2.0 +_MACHINE_TYPES = frozenset({"small", "default", "large"}) +# CreateSandboxRequest's bound on ttlSeconds: 30 days. +_MAX_TTL_SECONDS = 2_592_000 + + +def _api_error_detail(body: Any) -> str: + """ + Summarize the ``code``/``message`` pair Boat returns in a failed call's body. + + The status code alone hides reasons the task owner has to act on, such as an + account that has no Boat subscription yet. Returns an empty string when the + body is missing or is not the documented JSON error envelope. + """ + if isinstance(body, bytes): + with suppress(UnicodeDecodeError): + body = body.decode("utf-8") + if not isinstance(body, str): + return "" + try: + payload = json.loads(body) + except ValueError: + return "" + if not isinstance(payload, dict): + return "" + error = payload.get("error") + if not isinstance(error, dict): + error = payload + parts = [ + value.strip() + for key in ("code", "message") + if isinstance(value := error.get(key), str) and value.strip() + ] + if len(parts) == 2 and parts[0] == parts[1]: + del parts[1] + detail = ": ".join(parts) + return detail[:_MAX_ERROR_DETAIL] + + +@contextmanager +def _translate_boat_errors( + operation: str, *, recoverable_statuses: frozenset[int] = frozenset() +) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from boat_sdk.exceptions import ApiException + except ImportError: + raise SandboxTerminalError( + 'The Boat SDK is not installed. Install "apache-airflow-providers-common-ai[boat]".' + ) from e + if isinstance(e, ApiException): + status_code = e.status if isinstance(e.status, int) else None + status = f" (HTTP {status_code})" if status_code is not None else "" + detail = _api_error_detail(e.body) + message = f"Boat could not {operation}{status}." + if detail: + message = f"{message} {detail}" + if status_code in recoverable_statuses: + raise SandboxError(message) from e + raise SandboxTerminalError(message) from e + raise SandboxTerminalError(f"Boat could not {operation}: {type(e).__name__}.") from e + + +def _bound_text(text: str, max_bytes: int, *, already_truncated: bool = False) -> tuple[str, bool]: + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text, already_truncated + return encoded[-max_bytes:].decode("utf-8", errors="ignore"), True + + +class BoatSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in a `Boat <https://docs.boat.dev/quickstart>`__ sandbox. + + .. note:: + + Experimental: this can change or be removed in a minor release of this provider. + See :ref:`howto/stability`. + + Boat (formerly Ascii Box) is a hosted cloud-computer API: the Airflow worker + needs only network access and an API key, with no local daemon or host + virtualization. + + **Credentials are ambient.** On first use the backend reads ``BOAT_API_KEY`` + (required) and optional ``BOAT_BASE_URL`` from the worker environment. Modal + reads a ``modal`` connection first, but that connection type is owned by the + Modal provider; a ``boat`` connection type belongs in a future Boat provider, + not in this one. + + Boat cannot enforce a deny-all, per-domain, or CIDR egress policy. ``create`` + therefore refuses a :class:`~airflow.providers.common.ai.sandbox.SandboxSpec` + that asks for ``block_network=True``, ``allow_egress_to``, or + ``allow_egress_to_cidrs``, preserving the fail-closed contract. Pass + ``SandboxSpec(block_network=False)`` (and set that on + :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`) when + open egress is acceptable. + + Writes use Boat's native file API, which accepts only paths that resolve + under ``/home/user`` (where relative paths land) or ``/tmp``. The parent + directory is created first with ``mkdir -p`` in the guest, so a write + anywhere else reaches the model as a recoverable error: ``mkdir``'s own + when the guest user cannot create the parent directory, and one carrying + Boat's ``invalid_path`` code when that directory exists or could be + created. Reads use the inherited shell implementation, because the native + read API takes no size parameter and would land a whole file in worker + memory before ``max_bytes`` could reject it. + + :param machine_type: Boat machine size: ``small``, ``default``, or ``large``. + Default ``"default"``. + :param ttl_seconds: Server-side archive TTL in seconds after which the + sandbox is archived even if the worker never destroyed it: a whole number + from 1 to 2592000 (30 days). Default ``3600``. + :param ready_timeout: Seconds to wait for a newly created sandbox to become + ready. Default ``300``. + :param request_timeout: HTTP timeout in seconds for a Boat API call that answers + at once, such as a status check or a delete, and the time added to the + operation's own for a call that waits on one: a command's deadline, a + create's ``ready_timeout``, or 120 seconds for a file write. Default ``30``. + :param no_env: When ``True`` (default), create a no-env sandbox: none of the + account's stored environment variables, secret files or credentials + reach it, and it is confined so it cannot act on the account or its + other sandboxes. ``False`` attaches the account's ``base`` Boat + environment, which boat-sdk names on every create, and lifts that confinement, so secrets ``SandboxSpec.env`` + never named reach model-written code, which can then act on the account. + """ + + name = "boat" + + def __init__( + self, + *, + machine_type: str = "default", + ttl_seconds: int = 3600, + ready_timeout: float = 300.0, + request_timeout: float = 30.0, + no_env: bool = True, Review Comment: Removed `no_env`; every sandbox is created with `noEnv=True` (aa97c8a663). _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
