This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 96da58058ef Add Islo backend for sandbox tools (#71672)
96da58058ef is described below
commit 96da58058efd9648401722458d90e8b8a8aa4a95
Author: Yossi Eliaz <[email protected]>
AuthorDate: Wed Oct 14 10:23:08 2026 +0300
Add Islo backend for sandbox tools (#71672)
---
docs/spelling_wordlist.txt | 2 +
providers/common/ai/docs/index.rst | 1 +
providers/common/ai/docs/installation.rst | 7 +-
providers/common/ai/docs/sandbox/backends.rst | 119 +-
providers/common/ai/docs/sandbox/index.rst | 37 +-
providers/common/ai/provider.yaml | 3 +
providers/common/ai/pyproject.toml | 4 +
.../providers/common/ai/get_provider_info.py | 1 +
.../providers/common/ai/sandbox/__init__.py | 6 +
.../airflow/providers/common/ai/sandbox/islo.py | 757 ++++++++++++
.../common/ai/tests/system/common/ai/README.md | 12 +
.../common/ai/example_sandbox_toolset_islo.py | 134 ++
.../ai/tests/unit/common/ai/sandbox/test_islo.py | 1276 ++++++++++++++++++++
uv.lock | 22 +-
14 files changed, 2348 insertions(+), 33 deletions(-)
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index ca29e970906..00f4f3081b5 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -216,6 +216,7 @@ bugfix
bugfixes
buildType
burstable
+busybox
bytestring
cacert
Cadwyn
@@ -910,6 +911,7 @@ irreproducible
IRSA
isfile
ish
+islo
isn
isort
istio
diff --git a/providers/common/ai/docs/index.rst
b/providers/common/ai/docs/index.rst
index 33f18f69880..3962b0bd468 100644
--- a/providers/common/ai/docs/index.rst
+++ b/providers/common/ai/docs/index.rst
@@ -232,6 +232,7 @@ Extra Dependencies
``code-mode`` ``pydantic-ai-harness[codemode]>=0.24.0``
``shields`` ``pydantic-ai-shields>=0.3.4``
``skills`` ``apache-airflow-providers-git>=0.4.0``,
``pydantic-ai-skills>=1.2.0``
+``islo`` ``islo>=0.3.19``
``avro`` ``fastavro>=1.10.0; python_version < "3.14"``,
``fastavro>=1.12.1; python_version >= "3.14"``
``parquet`` ``pyarrow>=18.0.0; python_version < '3.14'``,
``pyarrow>=22.0.0; python_version >= '3.14'``
``sql`` ``apache-airflow-providers-common-sql>=2.2.0``,
``sqlglot>=30.0.0``
diff --git a/providers/common/ai/docs/installation.rst
b/providers/common/ai/docs/installation.rst
index a5451a09e16..d7ab8dd7417 100644
--- a/providers/common/ai/docs/installation.rst
+++ b/providers/common/ai/docs/installation.rst
@@ -48,9 +48,10 @@ The provider's extras split into a few groups:
than these, each under its own extra name, so check the
`pydantic-ai install docs <https://ai.pydantic.dev/install/#slim-install>`__
for the full list.
* **Agent tooling** (``mcp``, ``skills``, ``code-mode``, ``shields``,
``modal``,
- ``opensandbox``, ``boat``): MCP servers, Agent Skills, code-mode tool
execution, shield
- capabilities (input/output guards, tool guards, cost tracking), and the
hosted Modal and
- Boat and self-hosted OpenSandbox backends for :doc:`sandboxed execution
<sandbox/index>`.
+ ``opensandbox``, ``boat``, ``islo``): MCP servers, Agent Skills, code-mode
tool
+ execution, shield capabilities (input/output guards, tool guards, cost
tracking), and
+ the hosted Modal, Boat and Islo backends and the self-hosted OpenSandbox
backend for
+ :doc:`sandboxed execution <sandbox/index>`.
* **Document loading** (``pdf``, ``docx``, ``avro``, ``parquet``): file
formats for
document pipelines.
* **Retrieval / SQL** (``sql``, ``common.sql``, ``langchain``,
``llamaindex``): RAG and
diff --git a/providers/common/ai/docs/sandbox/backends.rst
b/providers/common/ai/docs/sandbox/backends.rst
index 63928eef8f4..2ecc67a142e 100644
--- a/providers/common/ai/docs/sandbox/backends.rst
+++ b/providers/common/ai/docs/sandbox/backends.rst
@@ -31,12 +31,12 @@ Modal (hosted)
:class:`~airflow.providers.common.ai.sandbox.modal.ModalSandboxBackend` runs
each
sandbox in Modal, provisioned over the API. Of the backends that ship with the
provider, **this is the managed one to use in production**, and with
-:ref:`OpenSandbox <sandbox-backend-opensandbox>` one of the two that run on
-Kubernetes: nothing has to be installed on the worker, model-written code never
-executes on the worker host, and Modal reclaims a sandbox at its own lifetime
-whether or not the worker survives. It needs the ``modal`` extra and Modal
-credentials, from a ``modal`` connection or the worker environment, as under
-:ref:`Quick start <sandbox-quick-start>`.
+:ref:`OpenSandbox <sandbox-backend-opensandbox>`, :ref:`Boat
<sandbox-backend-boat>` and
+:ref:`Islo <sandbox-backend-islo>` one of the four that run on Kubernetes:
nothing has to
+be installed on the worker, model-written code never executes on the worker
host, and Modal reclaims
+a sandbox at its own lifetime whether or not the worker survives. It needs the
+``modal`` extra and Modal credentials, from a ``modal`` connection or the
worker
+environment, as under :ref:`Quick start <sandbox-quick-start>`.
Constructor parameters:
@@ -185,6 +185,92 @@ them.
that is a symlink replaces the link with a regular file and leaves the original
target untouched, where a shell redirect would follow the link.
+.. _sandbox-backend-islo:
+
+Islo (hosted microVM)
+---------------------
+
+:class:`~airflow.providers.common.ai.sandbox.islo.IsloSandboxBackend` runs each
+sandbox in an `islo.dev <https://islo.dev>`__ microVM, provisioned over the
API.
+Like Modal and unlike ``sbx``, the worker only speaks HTTP: it needs no local
+daemon and no host virtualization, so it runs from a containerized worker and
on
+Kubernetes.
+
+Install the SDK extra:
+
+.. code-block:: bash
+
+ pip install "apache-airflow-providers-common-ai[islo]"
+
+.. code-block:: python
+
+ from airflow.providers.common.ai.sandbox import IsloSandboxBackend
+ from airflow.providers.common.ai.toolsets import SandboxToolset
+
+ SandboxToolset(IsloSandboxBackend())
+
+Credentials are ambient. On first use the SDK reads ``ISLO_API_KEY``, and
optionally
+``ISLO_BASE_URL`` and ``ISLO_COMPUTE_URL``, from the worker environment.
Modal's
+``modal`` connection comes from the Modal provider; an ``islo`` connection
type waits
+for an Islo provider.
+
+Constructor parameters:
+
+- ``image``, ``vcpus``, ``memory_mb``: image and sizing. ``None`` (default)
uses
+ the server default for each. The server default image is Debian based, and
any
+ Debian or Ubuntu based image, including ``python:*-slim``, has the GNU
``find``,
+ ``stat`` and ``tail`` the file operations need; Alpine and other busybox
images
+ do not.
+- ``pause_after_idle``: Seconds without a command or file operation after which
+ the server pauses the microVM and stops charging for its compute. Default
+ ``600``; ``None`` disables it.
+- ``auto_resume``: ``"on_activity"`` (default) resumes a paused sandbox on the
+ next tool call, so a long think between calls costs a resume rather than the
+ run; ``"never"`` leaves it paused, and the backend then treats a paused
sandbox
+ as unusable.
+- ``delete_after``: Seconds after **creation** at which the server deletes the
+ sandbox, whether or not it is in use. Default ``86400``; ``None`` disables
it.
+ This is the backstop for a worker killed mid-run; it is not renewed while the
+ sandbox is busy, so keep it longer than the longest run you expect.
+
+``SandboxSpec.env`` becomes the process environment of every command, and
+``block_network`` maps to the API's ``internet_enabled``. Four things Islo
cannot
+do are refused at ``create`` rather than silently dropped: ``allow_egress_to``
and
+``allow_egress_to_cidrs`` (the API turns outbound access on or off, with no
+per-host or per-address rule), a ``PATH`` entry in ``env`` (the runner sets
+``PATH`` for every command itself), and ``owner`` (this backend keeps no
+per-sandbox metadata, so it cannot be attached to from another task).
+
+File reads, writes and exports move file contents through Islo's native
+streaming APIs. Everything else runs through a shell wrapper in the sandbox, so
+the image needs ``sh``, ``mkdir`` and ``rm`` for the wrapper, ``tail`` to bound
+command output, ``dirname`` to create a written file's parent directory,
+``stat`` to size an over-budget read and to check a file before it is exported,
+``realpath`` to resolve a relative export path (the download API takes absolute
+paths only), and GNU ``find`` for directory listings. The compute API caps
each output stream
+at 1 MiB and keeps the tail. The backend captures each stream to a scratch
file in
+the sandbox and returns only its last ``max_output_bytes`` -- where a traceback
+and the exit status live -- so the worker sees a window sized to the caller's
+budget, a budget above the server cap is clamped to it, and total output is
+bounded by the sandbox's own ephemeral disk rather than by worker memory.
+
+The backend enforces the command deadline itself, because the API's
+``timeout_secs`` is accepted but not enforced, and times it from the API's
answer
+to the start request, so a slow start does not use it up. Polling rides out
+transient API errors until the deadline. If the command is still running when
+the deadline passes, the backend deletes the microVM and reports the command as
+timed out with ``sandbox_terminated`` set, so the toolset provisions a fresh
+sandbox for the next call. A deletion the API refused is logged rather than
+failing the task, and ``delete_after`` reclaims the microVM later; with
+``delete_after=None`` nothing does, and the warning says so. If the API was
still
+failing at the deadline, nothing is known about the command, so the task fails
+instead and Airflow's retry takes over.
+
+A missing file, a directory passed as a file, a relative path, and a write
onto a
+directory or under a pseudo-filesystem such as ``/proc`` or ``/sys`` come back
to
+the model as a recoverable error it can correct, once the backend has confirmed
+the sandbox itself is still usable.
+
.. _sandbox-backend-opensandbox:
OpenSandbox (self-hosted remote)
@@ -441,17 +527,18 @@ behaves identically everywhere:
- **CPU.** ``sbx`` gives a sandbox every host CPU; Modal defaults to a request
of
0.125 of one, so set ``cpu``; OpenSandbox takes ``cpu`` as a limit the
server enforces;
- Boat sizes by ``machine_type``.
+ Boat sizes by ``machine_type``; Islo uses the server default unless you set
``vcpus``.
- **Egress allowlists.** ``sbx`` enforces ``allow_egress_to`` at the host
policy
layer; Modal matches TLS handshake names, which is weaker and has to be opted
into; OpenSandbox enforces it in an egress sidecar, and the backend reads the
enforced policy and the sidecar's mode back rather than trusting the create
request, refusing a sidecar that filters only name resolution (``dns``)
rather
than connections (``dns+nft``). ``allow_egress_to_cidrs`` is enforced at the
- address layer on Modal and refused on ``sbx`` and OpenSandbox. Boat has no
form of
- egress control at all and refuses every network restriction rather than
- provisioning something weaker.
-- **Command timeouts.** A timeout destroys an ``sbx`` sandbox and its files;
+ address layer on Modal and refused on ``sbx``, OpenSandbox and Islo. Islo
has no
+ per-host form and refuses the spec; Boat has no form of egress control at
all and
+ refuses every network restriction. Both refuse rather than provision
something
+ weaker.
+- **Command timeouts.** A timeout destroys an ``sbx`` or Islo sandbox and its
files;
Modal, Boat and a server-enforced OpenSandbox timeout preserve the sandbox
and
files. OpenSandbox destroys it only if the command event stream itself stalls
past the client-side grace period, and Boat only if its own deadline, set
past
@@ -461,9 +548,9 @@ behaves identically everywhere:
with a regular file on Modal and OpenSandbox.
- **Attaching.** A Modal sandbox can be provisioned by one task and used by an
agent in another (:ref:`sandbox-attach`). An ``sbx`` microVM lives on the
worker
- that created it and cannot be reached from another task, and OpenSandbox and
Boat
- have no per-sandbox metadata the ownership rules could be kept in, so all
three refuse
- ``SandboxSpec.owner`` and the toolset refuses ``attach_to`` for them.
+ that created it and cannot be reached from another task, and OpenSandbox,
Boat and
+ Islo have no per-sandbox metadata the ownership rules could be kept in, so
all four
+ refuse ``SandboxSpec.owner`` and the toolset refuses ``attach_to`` for them.
.. _sandbox-byo:
@@ -481,8 +568,8 @@ commands. The default ``export_file``, behind
``SandboxToolset(exports=...)``,
copies a file in 4 MiB slices, one command each, and needs ``stat``, ``tail``,
``head`` and ``base64`` in the guest. It relies on ``run_command`` returning
each
slice's output intact, or setting ``stdout_truncated`` when it could not.
Override
-it when the vendor can stream a download, as the ``sbx`` and OpenSandbox
backends do. Override the others
-only when the vendor has a native file API:
+it when the vendor can stream a download, as the ``sbx``, OpenSandbox and Islo
+backends do. Override the others only when the vendor has a native file API:
.. code-block:: python
diff --git a/providers/common/ai/docs/sandbox/index.rst
b/providers/common/ai/docs/sandbox/index.rst
index 7ec165073c7..9e58c25f0c3 100644
--- a/providers/common/ai/docs/sandbox/index.rst
+++ b/providers/common/ai/docs/sandbox/index.rst
@@ -73,7 +73,7 @@ stays where the task runs:
end
sbx["sbx microVM<br/>local development only"]
end
- remote["Modal, OpenSandbox or Boat sandbox<br/>run_command,
read_file,<br/>write_file, list_directory"]
+ remote["Modal, OpenSandbox, Boat or Islo sandbox<br/>run_command,
read_file,<br/>write_file, list_directory"]
model <--> loop
db <--> other
loop <--> other
@@ -89,11 +89,13 @@ agent loop, the model calls and every other toolset keep
the task's authority, s
sandbox contains model-written code, not the agent (:ref:`sandbox-security`).
Under
``KubernetesExecutor`` the outer box is the task's pod: the pod protects the
cluster
from the task, and the sandbox protects the task from its model. An ``sbx``
microVM
-cannot run inside an unprivileged pod, so that combination needs Modal,
OpenSandbox or Boat.
+cannot run inside an unprivileged pod, so that combination needs Modal,
OpenSandbox, Boat
+or Islo.
The credential that provisions the sandbox depends on the backend: a ``modal``
connection (``modal_conn_id``; see the :ref:`Modal connection page
<howto/connection:modal>`)
for Modal; ``boat_conn_id``, or ``BOAT_API_KEY`` and optional
``BOAT_BASE_URL``, for Boat;
+``ISLO_API_KEY`` (and optional ``ISLO_BASE_URL`` and ``ISLO_COMPUTE_URL``) for
Islo;
``opensandbox_conn_id``, or ``OPEN_SANDBOX_DOMAIN`` and
``OPEN_SANDBOX_API_KEY``,
for OpenSandbox; and the host's ``sbx login`` for ``sbx``.
@@ -118,9 +120,9 @@ The four tools:
- Lists a directory. Directories are shown with a trailing ``/``.
A :class:`~airflow.providers.common.ai.sandbox.SandboxBackend` provisions the
sandbox
-on the model's first tool call and tears it down when the agent run ends. Four
-backends ship: `Modal <https://modal.com/docs/guide/sandbox>`__ and
-`Boat <https://docs.boat.dev/quickstart>`__ (hosted) and
+on the model's first tool call and tears it down when the agent run ends. Five
+backends ship: `Modal <https://modal.com/docs/guide/sandbox>`__,
+`Boat <https://docs.boat.dev/quickstart>`__ and `Islo <https://islo.dev>`__
(hosted) and
`OpenSandbox <https://open-sandbox.ai/>`__ (self-hosted) for production and
Kubernetes, and `Docker Sandboxes <https://docs.docker.com/ai/sandboxes/>`__
(``sbx``,
a microVM on the worker host) for local development. The tool names are the
ones
@@ -354,9 +356,15 @@ leaves its contents alone: the same four tools run in the
sandbox.
- Boat's infrastructure, off the worker.
- Stopped by Boat at ``ttl_seconds``. Sandboxes are created without
snapshots
or monitoring, so stopping one erases its disk and Boat keeps no
command history.
+ * - ``IsloSandboxBackend``
+ - A microVM that Islo provisions per sandbox.
+ - Islo's infrastructure, off the worker.
+ - Deleted by Islo at ``delete_after``, counted from creation. ``None``
+ disables that backstop.
When a run ends normally, the task calls the backend's ``destroy``. ``sbx``
runs its
-removal command and waits up to two minutes for it; Modal, OpenSandbox and
Boat each send a
+removal command and waits up to two minutes for it; Modal, OpenSandbox, Boat
and Islo each
+send a
termination request and return without waiting for the sandbox to stop. Any of
them
can return with the sandbox still present, and none of those cases fails the
task. A
SIGKILL, an out-of-memory kill or a lost node skips that teardown entirely,
and then
@@ -405,7 +413,7 @@ whatever it wants inside the pod that also holds your
credentials.
- Only what ``SandboxSpec.env`` names, which is nothing by default
Running an agent in a pod *and* giving it a sandbox is the setup a Kubernetes
-deployment usually wants, with Modal, OpenSandbox or Boat as the backend.
+deployment usually wants, with Modal, OpenSandbox, Boat or Islo as the backend.
When to choose it
-----------------
@@ -423,7 +431,7 @@ task, a script the model writes, runs, and fixes from its
own traceback.
:ref:`sandbox-security`.
- **Nothing survives the run** in a sandbox the toolset provisions itself,
including across task retries. On Modal, a sandbox a task provisions and the
agent
- attaches to does; ``sbx``, OpenSandbox and Boat cannot be attached to.
+ attaches to does; ``sbx``, OpenSandbox, Boat and Islo cannot be attached to.
:ref:`Lifecycle <sandbox-lifecycle>`,
:ref:`A sandbox another task owns <sandbox-attach>`.
- **A file the agent built leaves through** ``exports`` **or a task**, never
through
@@ -446,10 +454,12 @@ task, a script the model writes, runs, and fixes from its
own traceback.
its address allowlist is enforced properly but cannot serve a package
registry whose
addresses rotate. Boat cannot restrict egress at all, so it refuses every
network
restriction, including the default ``block_network=True``: pass
- ``SandboxSpec(block_network=False)`` to accept open egress there.
+ ``SandboxSpec(block_network=False)`` to accept open egress there. Islo turns
outbound
+ access on or off and refuses hostname and CIDR allowlists.
:ref:`Configuring a sandbox <sandbox-configuring>`,
:ref:`Modal <sandbox-backend-modal>`,
- :ref:`Boat <sandbox-backend-boat>`.
+ :ref:`Boat <sandbox-backend-boat>`,
+ :ref:`Islo <sandbox-backend-islo>`.
- **On Modal, commands run as root and** ``workdir`` **is not a jail.**
:ref:`Modal <sandbox-backend-modal>`.
- **On Boat, commands can become root through** ``sudo``, **the sandbox has a
public
@@ -458,7 +468,7 @@ task, a script the model writes, runs, and fixes from its
own traceback.
- **The** ``sbx`` **backend is for local development.** It needs the ``sbx``
binary,
a Docker login and, on Linux, KVM or nested virtualization, so it does not
run on
unprivileged Kubernetes, and a worker killed outright leaves its microVM
behind.
- Production and Kubernetes use Modal, OpenSandbox or Boat instead.
+ Production and Kubernetes use Modal, OpenSandbox, Boat or Islo instead.
:ref:`sbx <sandbox-backend-sbx>`.
- **A failed teardown is logged, not raised**, so a teardown blip cannot fail a
finished run; reclaiming the sandbox is then the backend's lifetime or an
@@ -469,13 +479,14 @@ has the two agents in the quick start, a ``@task``
producing a file through a
sandbox, an agent attaching to a task-owned sandbox, and an agent exporting a
file,
shown on this page and in :doc:`configuration`. The system tests
``example_sandbox_toolset_sbx.py``, ``example_sandbox_toolset_modal.py``,
-``example_sandbox_toolset_opensandbox.py`` and
``example_sandbox_toolset_boat.py``
+``example_sandbox_toolset_opensandbox.py``,
``example_sandbox_toolset_boat.py`` and
+``example_sandbox_toolset_islo.py``
run against a real backend and are
reachable from the System Tests entry in the sidebar.
**Credentials and where it runs.** Only what ``SandboxSpec.env`` names enters
the
sandbox; the provisioning credential for each backend is listed under
-:ref:`sandbox-placement`. Work runs in Modal's or Boat's infrastructure, on the
+:ref:`sandbox-placement`. Work runs in Modal's, Boat's or Islo's
infrastructure, on the
OpenSandbox server's runtime (off the worker unless you run the server there),
or in a per-run
microVM on the worker host with ``sbx``. Its tool calls act as barriers, as
they do
for the other routes that build their own tools; see
:ref:`toolset-call-barriers`.
diff --git a/providers/common/ai/provider.yaml
b/providers/common/ai/provider.yaml
index cfb9e9c5643..4aa3d8112d6 100644
--- a/providers/common/ai/provider.yaml
+++ b/providers/common/ai/provider.yaml
@@ -74,6 +74,9 @@ integrations:
- integration-name: Modal Sandboxes
external-doc-url: https://modal.com/docs/guide/sandbox
tags: [service]
+ - integration-name: Islo
+ external-doc-url: https://docs.islo.dev/
+ tags: [service]
- integration-name: OpenSandbox
external-doc-url: https://open-sandbox.ai/
tags: [software]
diff --git a/providers/common/ai/pyproject.toml
b/providers/common/ai/pyproject.toml
index 4f7ab1faa68..7078737fbf7 100644
--- a/providers/common/ai/pyproject.toml
+++ b/providers/common/ai/pyproject.toml
@@ -125,6 +125,9 @@ dependencies = [
"apache-airflow-providers-git>=0.4.0",
"pydantic-ai-skills>=1.2.0",
]
+# Islo-hosted microVM sandboxes for SandboxToolset: like modal, needs no
host-level
+# installation, only ISLO_API_KEY in the worker environment.
+"islo" = ["islo>=0.3.19"]
"avro" = [
'fastavro>=1.10.0; python_version < "3.14"',
'fastavro>=1.12.1; python_version >= "3.14"',
@@ -182,6 +185,7 @@ dev = [
"openai>=2.47.0",
"anthropic>=1.0.0",
"boat-sdk>=1.9.3",
+ "islo>=0.3.19",
"opensandbox>=1.1.0",
]
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
b/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
index 0e518bd7e1d..100f4f067ef 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py
@@ -76,6 +76,7 @@ def get_provider_info():
"external-doc-url": "https://modal.com/docs/guide/sandbox";,
"tags": ["service"],
},
+ {"integration-name": "Islo", "external-doc-url":
"https://docs.islo.dev/";, "tags": ["service"]},
{
"integration-name": "OpenSandbox",
"external-doc-url": "https://open-sandbox.ai/";,
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
b/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
index 3f588c62681..3e770ed26b7 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/sandbox/__init__.py
@@ -29,12 +29,18 @@ from airflow.providers.common.ai.sandbox.base import (
dag_run_owner,
)
from airflow.providers.common.ai.sandbox.boat import BoatSandboxBackend
+
+# The Islo backend imports its SDK lazily, on first use, so the package is
+# importable without the optional ``islo`` extra; a missing SDK surfaces as an
+# actionable error from the backend instead.
+from airflow.providers.common.ai.sandbox.islo import IsloSandboxBackend
from airflow.providers.common.ai.sandbox.opensandbox import OpenSandboxBackend
from airflow.providers.common.ai.sandbox.sbx import SbxSandboxBackend
__all__ = [
"AttachableSandboxBackend",
"BoatSandboxBackend",
+ "IsloSandboxBackend",
"OpenSandboxBackend",
"ModalSandboxBackend",
"SandboxBackend",
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py
b/providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py
new file mode 100644
index 00000000000..b9c44b60054
--- /dev/null
+++ b/providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py
@@ -0,0 +1,757 @@
+# 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.
+"""islo.dev microVM backend for
:class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`."""
+
+from __future__ import annotations
+
+import logging
+import math
+import os
+import shlex
+import time
+from contextlib import closing, contextmanager, suppress
+from typing import TYPE_CHECKING, Any, Literal, NoReturn
+
+import httpx
+
+from airflow.providers.common.ai.sandbox.base import (
+ SandboxBackend,
+ SandboxError,
+ SandboxExecResult,
+ SandboxFileTooLargeError,
+ SandboxTerminalError,
+ _check_export_deadline,
+ _export_deadline,
+ _new_sandbox_name,
+ _validate_positive_finite,
+)
+
+# ``islo`` is the optional ``[islo]`` extra and the package exports this
backend
+# eagerly, so the SDK is imported where it is first used, never at module
level.
+if TYPE_CHECKING:
+ from collections.abc import Generator, Iterator
+ from typing import BinaryIO
+
+ from islo import Islo
+ from islo.core.request_options import RequestOptions
+ from islo.types import ExecResponse, ExecResultResponse, SandboxResponse
+
+ from airflow.providers.common.ai.sandbox.base import SandboxSpec
+
+log = logging.getLogger(__name__)
+
+# The vendor types ``status`` as a plain string on a model that allows extra
+# fields, so the vocabulary is open-ended. Track the statuses that mean "not
+# finished yet" instead of the ones that mean "finished": an unrecognised
status
+# then reads as terminal, which surfaces a failure, rather than as
still-running,
+# which would poll to the deadline and cost the agent its sandbox.
+_RUNNING_EXEC_STATUSES = frozenset({"pending", "queued", "starting",
"running"})
+# Sandbox statuses that cannot serve a request. Same open vocabulary, opposite
+# bias: after a missing-file response the sandbox was reachable a moment ago,
so
+# an unknown status reads as usable and only a known-dead one fails the task.
+_UNUSABLE_SANDBOX_STATUSES = frozenset({"stopping", "stopped", "deleting",
"deleted", "error", "failed"})
+_AUTO_RESUME_POLICIES = frozenset({"never", "on_activity"})
+_POLL_INITIAL = 0.2
+_POLL_MAX = 2.0
+_POLL_BACKOFF = 1.5
+# The HTTP timeout of the requests that start and poll a command never drops
+# below this: a small command budget, or the last poll before the deadline,
must
+# not become a one-second request budget that a slow API -- the first call
after
+# an auto-resume, say -- turns into a task failure instead of a command result.
+_POLL_HTTP_TIMEOUT_MIN = 5.0
+# Starting a command is sent without the SDK's retries (see
``_request_options``),
+# so the backend resends it itself, and only after a failure that proves the
+# command never ran.
+_START_RETRIES = 2
+_START_RETRY_DELAY = 1.0
+_FILE_OP_TIMEOUT = 120.0
+# File-transfer responses that say nothing the model can act on: bad
credentials,
+# a rate limit, or an overloaded API. Every other status is checked against the
+# sandbox before it is handed to the model; see ``_raise_file_op_error``.
+_TERMINAL_FILE_OP_STATUSES = frozenset({401, 403, 429})
+_API_MESSAGE_MAX_CHARS = 200
+# Measured against the compute API: each stream is capped at exactly this many
+# bytes with the tail kept, and one ``truncated`` flag covers both streams. The
+# wrapper never asks for more than this per stream, so that flag is not read: a
+# stream long enough to reach the cap is over the budget and marked as cut
anyway.
+_SERVER_STREAM_CAP = 1024 * 1024
+_HELPER_OUTPUT_CAP = _SERVER_STREAM_CAP - 1
+# Runs the agent's command with each stream captured to a scratch file, then
+# emits only the last ``$2`` bytes of each. The tail is what the model needs (a
+# traceback and the exit status live at the end), and bounding inside the guest
+# keeps the transfer and the worker's copy at the caller's budget rather than
+# the server's 1 MiB.
+#
+# ``sh -c`` rather than a login shell: the spec's variables are the process
+# environment of every exec, and ``/etc/profile`` would run after them and win
+# for anything it also exports.
+#
+# Deliberately free of fifos, background jobs and ``wait``: a command that
+# backgrounds a process hands it the capture descriptor, so anything waiting
for
+# end-of-input would block until that process exits -- ``sleep 20 & echo
+# started`` took 20s in a real microVM before this. Redirecting to files means
+# only the foreground command is waited on. The cost is that the scratch file
+# grows with total output, on the sandbox's own ephemeral disk.
+_COMMAND_WRAPPER = """\
+dir="${TMPDIR:-/tmp}/airflow-sandbox-$$"
+mkdir -m 700 "$dir" || exit 70
+trap 'rm -rf "$dir"' EXIT
+trap 'rm -rf "$dir"; exit 143' HUP INT TERM
+sh -c "$1" >"$dir/out" 2>"$dir/err"
+status=$?
+tail -c "$2" <"$dir/out"
+tail -c "$2" <"$dir/err" >&2
+exit "$status"
+"""
+
+
+@contextmanager
+def _translate_islo_errors(operation: str) -> Iterator[None]:
+ try:
+ yield
+ except SandboxError:
+ raise
+ except Exception as e:
+ try:
+ from islo.core.api_error import ApiError
+ except ImportError:
+ raise SandboxTerminalError(
+ 'The Islo SDK is not installed. Install
"apache-airflow-providers-common-ai[islo]".'
+ ) from e
+ if isinstance(e, ApiError):
+ status = f" (HTTP {e.status_code})" if e.status_code is not None
else ""
+ raise SandboxTerminalError(f"Islo could not {operation}{status}.")
from e
+ if isinstance(e, httpx.HTTPStatusError):
+ # The SDK exchanges ISLO_API_KEY for a token with a bare ``httpx``
call
+ # whose failures arrive here rather than as an ``ApiError``.
+ code = e.response.status_code
+ if code in (401, 403):
+ raise SandboxTerminalError(
+ f"Islo rejected ISLO_API_KEY (HTTP {code}) when asked to
{operation}."
+ ) from e
+ raise SandboxTerminalError(f"Islo could not {operation} (HTTP
{code}).") from e
+ raise SandboxTerminalError(f"Islo could not {operation}:
{type(e).__name__}.") from e
+
+
+def _raise_translated(error: Exception, operation: str) -> NoReturn:
+ """Raise ``error`` the way :func:`_translate_islo_errors` would have, for
an error caught elsewhere."""
+ with _translate_islo_errors(operation):
+ raise error
+
+
+def _api_error_message(error: Exception) -> str:
+ """Return the ``message`` of an Islo error body, trimmed for a prompt, or
``""``."""
+ body = getattr(error, "body", None)
+ message = body.get("message") if isinstance(body, dict) else None
+ if not isinstance(message, str):
+ return ""
+ message = message.strip().rstrip(".")
+ if len(message) > _API_MESSAGE_MAX_CHARS:
+ message = message[:_API_MESSAGE_MAX_CHARS] + "..."
+ return message
+
+
+def _is_transient_error(error: Exception) -> bool:
+ """Whether a failed call says nothing about the command: a 5xx, a 429, or
no response at all."""
+ from islo.core.api_error import ApiError
+
+ if isinstance(error, ApiError):
+ return error.status_code is None or error.status_code == 429 or
error.status_code >= 500
+ if isinstance(error, httpx.HTTPStatusError):
+ # A token refresh that failed the same way, before the call itself was
sent.
+ code = error.response.status_code
+ return code == 429 or code >= 500
+ return isinstance(error, httpx.TransportError)
+
+
+def _is_undelivered_start(error: Exception) -> bool:
+ """
+ Whether a failed start proves the command never ran.
+
+ That is: no connection was made, a 429 refused it, or the token exchange
that
+ runs before it failed with a 429 or 5xx.
+ """
+ from islo.core.api_error import ApiError
+
+ if isinstance(error, ApiError):
+ return error.status_code == 429
+ if isinstance(error, httpx.HTTPStatusError):
+ # The token refresh runs before the start is sent, so its 429 or 5xx
means
+ # the command was never asked for. A 401 or 403 stays terminal.
+ code = error.response.status_code
+ return code == 429 or code >= 500
+ return isinstance(error, httpx.ConnectError)
+
+
+def _bound_result_stream(text: str, max_bytes: int) -> tuple[str, bool]:
+ """
+ Trim one stream to ``max_bytes``, keeping the tail, and report whether
bytes were dropped.
+
+ The sandbox is asked for one byte more than the budget, so a stream that
+ comes back over budget is the signal that the guest had more to give.
+ """
+ encoded = text.encode("utf-8", errors="surrogatepass")
+ truncated = False
+ if len(encoded) > max_bytes:
+ encoded = encoded[-max_bytes:]
+ truncated = True
+ # A byte-aligned cut usually lands mid-record, so drop the leading
partial
+ # line -- but only while half the window survives, as Modal's
``_drain``
+ # does. One line longer than the budget ends in its only newline, and
+ # dropping through it would leave nothing, which the model reads as
"(no
+ # output)"; a partial line kept instead is still marked as cut.
+ newline = encoded.find(b"\n")
+ if newline != -1 and len(encoded) - (newline + 1) >= max_bytes // 2:
+ encoded = encoded[newline + 1 :]
+ return encoded.decode("utf-8", errors="replace"), truncated
+
+
+class IsloSandboxBackend(SandboxBackend):
+ """
+ Sandbox backend that runs agent commands in an `islo.dev
<https://islo.dev>`__ microVM.
+
+ .. note::
+
+ Experimental: this can change or be removed in a minor release of this
provider.
+ See :ref:`howto/stability`.
+
+ Islo is a hosted API with no local daemon or host-virtualization
requirement,
+ so this backend works from an Airflow worker running in a container.
+
+ **Credentials are ambient.** The SDK reads ``ISLO_API_KEY``, and optionally
+ ``ISLO_BASE_URL`` and ``ISLO_COMPUTE_URL``, from the worker environment on
first
+ use; an unset ``ISLO_API_KEY`` fails the task at that point. Modal reads a
+ ``modal`` connection first, but that connection type is owned by the Modal
+ provider; an ``islo`` connection type belongs in a future Islo provider,
not in
+ this one.
+
+ File reads, writes and exports move file contents through Islo's native
+ streaming APIs. Everything else runs through a shell wrapper in the
sandbox,
+ so the image needs ``sh``, ``mkdir`` and ``rm`` for the wrapper, ``tail``
to
+ bound command output, ``dirname`` to create a written file's parent
+ directory, ``stat`` to size an over-budget read and to check a file before
it
+ is exported, ``realpath`` to resolve a relative export path, and GNU
``find``
+ for directory listings. The server default
+ image and any Debian or Ubuntu based image provide them. Each command's
+ output is captured to a scratch file in the sandbox and only its last
+ ``max_output_bytes`` are returned, so the worker sees a bounded tail while
+ the sandbox's own ephemeral disk absorbs the rest.
+
+ Islo sets ``PATH`` for every command itself and drops a ``PATH`` given at
+ creation, so a spec that names it is refused rather than silently ignored.
+ ``SandboxSpec.owner`` is refused for the same reason: this backend keeps no
+ per-sandbox metadata, so a sandbox created here cannot be attached to
later.
+
+ :param image: Sandbox image. ``None`` (default) uses the server default.
+ :param vcpus: Number of virtual CPUs. ``None`` uses the server default.
+ :param memory_mb: Memory in MB. ``None`` uses the server default.
+ :param pause_after_idle: Seconds without a command or file operation after
+ which the server pauses the microVM and releases its compute. ``None``
+ disables it. Default ``600``.
+ :param auto_resume: ``"on_activity"`` (default) resumes a paused sandbox on
+ the next command or file operation; ``"never"`` leaves it paused, and
the
+ backend then treats a paused sandbox as unusable.
+ :param delete_after: Seconds after *creation* at which the server deletes
the
+ sandbox whether or not it is in use. ``None`` disables it. Default
+ ``86400``.
+ """
+
+ name = "islo"
+
+ def __init__(
+ self,
+ *,
+ image: str | None = None,
+ vcpus: int | None = None,
+ memory_mb: int | None = None,
+ pause_after_idle: int | None = 600,
+ auto_resume: Literal["never", "on_activity"] = "on_activity",
+ delete_after: int | None = 86400,
+ ) -> None:
+ if pause_after_idle is not None:
+ _validate_positive_finite(pause_after_idle, "pause_after_idle")
+ if delete_after is not None:
+ _validate_positive_finite(delete_after, "delete_after")
+ if auto_resume not in _AUTO_RESUME_POLICIES:
+ raise ValueError(
+ f"auto_resume must be one of {sorted(_AUTO_RESUME_POLICIES)},
got {auto_resume!r}."
+ )
+ if vcpus is not None:
+ _validate_positive_finite(vcpus, "vcpus")
+ if memory_mb is not None:
+ _validate_positive_finite(memory_mb, "memory_mb")
+ if image == "":
+ raise ValueError("image must not be empty.")
+ self._image = image
+ self._vcpus = vcpus
+ self._memory_mb = memory_mb
+ self._pause_after_idle = pause_after_idle
+ self._auto_resume = auto_resume
+ self._delete_after = delete_after
+ self._client: Islo | None = None
+
+ def _get_client(self) -> Islo:
+ if self._client is not None:
+ return self._client
+ with _translate_islo_errors("initialize its client"):
+ from islo import Islo
+
+ # Without a key the SDK builds a client that sends no credentials,
and
+ # the first sign of it would be a bare HTTP error from create.
+ if not os.environ.get("ISLO_API_KEY"):
+ raise SandboxTerminalError(
+ "ISLO_API_KEY is not set in the worker environment; the
Islo backend reads its "
+ "credentials only from there."
+ )
+ self._client = Islo()
+ return self._client
+
+ @staticmethod
+ def _request_options(
+ *, timeout: float, chunk_size: int | None = None, max_retries: int |
None = None
+ ) -> RequestOptions:
+ # ``max_retries`` is left to the SDK's default unless asked for:
passing
+ # 0 switches off its retries of dropped connections and of 5xx, 408,
409
+ # and 429 answers. Starting a command is the exception: that POST
carries no
+ # idempotency key, so a resend after a lost response or a 5xx would
run the
+ # agent's command a second time.
+ options: RequestOptions = {"timeout_in_seconds": max(1,
math.ceil(timeout))}
+ if chunk_size is not None:
+ options["chunk_size"] = chunk_size
+ if max_retries is not None:
+ options["max_retries"] = max_retries
+ return options
+
+ def _ensure_sandbox_usable(self, info: SandboxResponse) -> None:
+ status = info.status
+ unusable = info.deleted_at is not None or status in
_UNUSABLE_SANDBOX_STATUSES
+ if status == "paused" and self._auto_resume != "on_activity":
+ unusable = True
+ if unusable:
+ raise SandboxTerminalError(
+ f"Islo sandbox {info.name!r} cannot serve requests
(status={status!r})."
+ )
+
+ @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 keeps
none of,
+ # 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 Islo backend cannot apply a per-domain egress allowlist;
it can only turn "
+ "outbound access on or off. Drop allow_egress_to, or use a
backend with "
+ "per-domain network rules."
+ )
+ if spec.allow_egress_to_cidrs:
+ raise SandboxTerminalError(
+ "SandboxSpec names allow_egress_to_cidrs, which the Islo
backend cannot enforce: "
+ "it can only turn outbound access on or off. Drop
allow_egress_to_cidrs, or use "
+ "a backend with an address-layer allowlist."
+ )
+ if spec.env and "PATH" in spec.env:
+ raise SandboxTerminalError(
+ "Islo sets PATH for every command itself and drops a PATH
given at creation; "
+ "remove PATH from SandboxSpec.env."
+ )
+
+ def create(self, *, spec: SandboxSpec | None = None) -> str:
+ self._check_spec(spec)
+ with _translate_islo_errors("create a sandbox"):
+ from islo.types import AutoResumePolicy, LifecyclePolicy
+
+ kwargs: dict[str, Any] = {
+ "internet_enabled": False if spec is None else not
spec.block_network,
+ "lifecycle": LifecyclePolicy(
+ pause_after_idle=self._pause_after_idle,
+ auto_resume=AutoResumePolicy(self._auto_resume),
+ delete_after=self._delete_after,
+ ),
+ }
+ if self._image is not None:
+ kwargs["image"] = self._image
+ if self._vcpus is not None:
+ kwargs["vcpus"] = self._vcpus
+ if self._memory_mb is not None:
+ kwargs["memory_mb"] = self._memory_mb
+ if spec is not None and spec.env:
+ # Verified against a live microVM: variables set here are the
+ # process environment of every later exec.
+ kwargs["env"] = dict(spec.env)
+ # Bind the name before the call. If creation fails after the server
+ # provisioned the microVM -- a response timeout, a reset, a 5xx --
+ # this is the only handle that can still delete it, and without it
+ # the leak is neither cleanable nor traceable to a run.
+ name = _new_sandbox_name()
+ try:
+ sandbox: SandboxResponse =
self._get_client().sandboxes.create_sandbox(
+ name=name,
+
request_options=self._request_options(timeout=_FILE_OP_TIMEOUT),
+ **kwargs,
+ )
+ except BaseException:
+ with suppress(Exception):
+ self.destroy(name)
+ raise
+ try:
+ self._ensure_sandbox_usable(sandbox)
+ except SandboxTerminalError:
+ with suppress(Exception):
+ self.destroy(name)
+ raise
+ return sandbox.name
+
+ def _await_exec(self, sandbox: str, exec_id: str, *, deadline: float) ->
ExecResultResponse | None:
+ client = self._get_client()
+ interval = _POLL_INITIAL
+ last_error: Exception | None = None
+ while time.monotonic() < deadline:
+ remaining = deadline - time.monotonic()
+ try:
+ result: ExecResultResponse = client.sandboxes.get_exec_result(
+ sandbox,
+ exec_id,
+
request_options=self._request_options(timeout=max(_POLL_HTTP_TIMEOUT_MIN,
remaining)),
+ )
+ except Exception as e:
+ if not _is_transient_error(e):
+ _raise_translated(e, "poll a sandbox command")
+ # One failed poll says nothing about the command; the deadline
decides.
+ last_error = e
+ else:
+ last_error = None
+ if result.status not in _RUNNING_EXEC_STATUSES:
+ return result
+ time.sleep(min(interval, max(0.0, deadline - time.monotonic())))
+ interval = min(interval * _POLL_BACKOFF, _POLL_MAX)
+ if last_error is not None:
+ # Nothing was heard after the last failure, so "timed out" would
be a guess.
+ _raise_translated(last_error, "poll a sandbox command")
+ return None
+
+ def _destroy_after_timeout(self, sandbox: str) -> None:
+ try:
+ self.destroy(sandbox)
+ except SandboxError:
+ # Warn rather than fail the task: the command merely ran long, and
+ # failing here would turn a timeout the model can react to into a
+ # task failure over a transient error. The warning says whether the
+ # lifecycle policy will reclaim the microVM, since with
+ # ``delete_after=None`` nothing will.
+ if self._delete_after is not None:
+ reclaim = f"its delete_after policy removes it
{self._delete_after}s after creation"
+ else:
+ reclaim = "delete_after is disabled, so it persists until
deleted by hand"
+ log.warning(
+ "Timed out running a command in Islo sandbox %s and could not
confirm its deletion; %s.",
+ sandbox,
+ reclaim,
+ exc_info=True,
+ )
+
+ 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")
+ client = self._get_client()
+ # The server returns at most _SERVER_STREAM_CAP bytes per stream, so a
+ # larger budget cannot be honoured and is clamped below it.
+ budget = min(max_output_bytes, _SERVER_STREAM_CAP - 1)
+ attempt = 0
+ while True:
+ try:
+ response: ExecResponse = client.sandboxes.exec_in_sandbox(
+ sandbox,
+ # One byte over the budget, so a stream that comes back
over it
+ # is proof the guest had more to give.
+ command=[
+ "sh",
+ "-c",
+ _COMMAND_WRAPPER,
+ "airflow-sandbox",
+ command,
+ str(budget + 1),
+ ],
+ timeout_secs=max(1, math.ceil(timeout)),
+ request_options=self._request_options(
+ timeout=max(_POLL_HTTP_TIMEOUT_MIN, timeout),
max_retries=0
+ ),
+ )
+ break
+ except Exception as e:
+ if attempt >= _START_RETRIES or not _is_undelivered_start(e):
+ _raise_translated(e, "start a sandbox command")
+ time.sleep(_START_RETRY_DELAY * 2**attempt)
+ attempt += 1
+ # Timed from the start's answer, so a slow start cannot spend the
+ # command's budget and leave it destroyed without a single poll.
+ deadline = time.monotonic() + timeout
+ result = self._await_exec(sandbox, response.exec_id, deadline=deadline)
+ if result is None:
+ self._destroy_after_timeout(sandbox)
+ return SandboxExecResult(
+ exit_code=-1, stdout="", stderr="", timed_out=True,
sandbox_terminated=True
+ )
+
+ stdout, out_truncated = _bound_result_stream(result.stdout, budget)
+ stderr, err_truncated = _bound_result_stream(result.stderr, budget)
+ if result.status == "timeout":
+ self._destroy_after_timeout(sandbox)
+ return SandboxExecResult(
+ exit_code=-1,
+ stdout=stdout,
+ stderr=stderr,
+ timed_out=True,
+ stdout_truncated=out_truncated,
+ stderr_truncated=err_truncated,
+ sandbox_terminated=True,
+ )
+ return SandboxExecResult(
+ exit_code=result.exit_code if result.exit_code is not None else -1,
+ stdout=stdout,
+ stderr=stderr,
+ stdout_truncated=out_truncated,
+ stderr_truncated=err_truncated,
+ )
+
+ def _run_helper(self, sandbox: str, script: str, *, operation: str) ->
SandboxExecResult:
+ result = self.run_command(
+ sandbox, script, timeout=_FILE_OP_TIMEOUT,
max_output_bytes=_HELPER_OUTPUT_CAP
+ )
+ if result.timed_out or result.sandbox_terminated:
+ raise SandboxTerminalError(f"The sandbox was destroyed after it
timed out while {operation}.")
+ if result.exit_code:
+ raise SandboxError(result.stderr.strip() or f"Could not
{operation}.")
+ return result
+
+ def _raise_file_op_error(
+ self, sandbox: str, path: str, error: Exception, *, operation: str,
hint: str = ""
+ ) -> NoReturn:
+ """
+ Raise a failed file transfer as a prompt for the model when the
sandbox explains it.
+
+ Measured against the file APIs: a missing file and a directory are
both a
+ 404, a relative path is a 400 whose message names the problem, and
writing
+ onto a directory or under a pseudo-filesystem such as ``/proc`` or
``/sys``
+ is a bare 500. Each of those is the model's to fix. A gone or stopped
+ sandbox fails file calls too, and no tool call can fix that, so the
sandbox
+ is checked before anything reaches the model. No response at all, bad
+ credentials, a rate limit and a 5xx above 500 say nothing about the
path and
+ fail the task.
+ """
+ from islo.core.api_error import ApiError
+
+ status = error.status_code if isinstance(error, ApiError) else None
+ if status is None or status in _TERMINAL_FILE_OP_STATUSES or status >
500:
+ _raise_translated(error, f"{operation} a sandbox file")
+ with _translate_islo_errors(f"check a sandbox after a failed file
{operation}"):
+ info: SandboxResponse = self._get_client().sandboxes.get_sandbox(
+ sandbox,
request_options=self._request_options(timeout=_FILE_OP_TIMEOUT)
+ )
+ self._ensure_sandbox_usable(info)
+ if status == 404:
+ raise SandboxError(
+ f"{path!r} does not exist in the sandbox, or is not a regular
file."
+ ) from error
+ message = f"Could not {operation} {path!r} (HTTP {status})"
+ detail = _api_error_message(error) if status < 500 else ""
+ if detail:
+ message += f": {detail}"
+ message += "."
+ if hint and status >= 500:
+ message += f" {hint}"
+ raise SandboxError(message) from error
+
+ def _get_file_size(self, sandbox: str, path: str) -> int | None:
+ """Ask the guest for the file's size; ``None`` when it cannot say."""
+ try:
+ result = self._run_helper(
+ sandbox, f"stat -Lc %s -- {shlex.quote(path)}",
operation=f"size {path!r}"
+ )
+ return int(result.stdout.strip())
+ except SandboxTerminalError:
+ raise
+ except (SandboxError, ValueError):
+ return None
+
+ def _download(
+ self, sandbox: str, path: str, *, max_bytes: int, operation: str
+ ) -> Generator[bytes, None, None]:
+ """
+ Yield a sandbox file in chunks, raising a failed download through
:meth:`_raise_file_op_error`.
+
+ A generator, so a failure in whatever the caller does with a chunk is
raised in
+ the caller and never mistaken for a failed download.
+ """
+ client = self._get_client()
+ chunks = None
+ try:
+ chunks = client.sandboxes.download_file(
+ sandbox,
+ path=path,
+ request_options=self._request_options(
+ timeout=_FILE_OP_TIMEOUT, chunk_size=min(65536, max_bytes
+ 1)
+ ),
+ )
+ yield from chunks
+ except Exception as e:
+ self._raise_file_op_error(sandbox, path, e, operation=operation)
+ finally:
+ close = getattr(chunks, "close", None)
+ if close is not None:
+ with suppress(Exception):
+ close()
+
+ def read_file(self, sandbox: str, path: str, *, max_bytes: int) -> bytes:
+ _validate_positive_finite(max_bytes, "max_bytes")
+ data = bytearray()
+ with closing(self._download(sandbox, path, max_bytes=max_bytes,
operation="read")) as chunks:
+ for chunk in chunks:
+ data.extend(chunk[: max_bytes + 1 - len(data)])
+ if len(data) > max_bytes:
+ break
+ if len(data) > max_bytes:
+ # The download API has no size endpoint, so the guest's own stat
+ # supplies the number the model plans around. Without it, say only
+ # what is known rather than report the budget back as the size.
+ size = self._get_file_size(sandbox, path)
+ if size is None:
+ raise SandboxError(
+ f"{path!r} is larger than the {max_bytes} byte read limit.
Read just the part you "
+ "need with a shell command instead (e.g. head, tail, sed
-n, or grep)."
+ )
+ raise SandboxFileTooLargeError(path, max(size, len(data)),
max_bytes)
+ return bytes(data)
+
+ def export_file(self, sandbox: str, path: str, dest: BinaryIO, *,
max_bytes: int) -> int:
+ """
+ Override: stream the file through Islo's download API into ``dest``.
+
+ The default reads 4 MiB slices through ``run_command``, which here
returns at
+ most 1 MiB per stream. The guest still runs the default's checks
first, so a
+ refusal means what it does on every backend and the size it reports
catches
+ a file that is still being written.
+ """
+ _validate_positive_finite(max_bytes, "max_bytes")
+ deadline = _export_deadline(max_bytes)
+ quoted = shlex.quote(path)
+ check = self.run_command(
+ sandbox,
+ f"{self._export_checks(quoted, max_bytes)}
{self._print_export_size()}",
+ timeout=_FILE_OP_TIMEOUT,
+ max_output_bytes=_HELPER_OUTPUT_CAP,
+ )
+ self._raise_for_export_status(path, check, max_bytes)
+ size = self._parse_export_size(check.stdout)
+ source = path
+ if not path.startswith("/"):
+ # An export path may be relative to the sandbox's working
directory, as it
+ # is for the checks above, but the download API answers a relative
path
+ # with a 400. The export runs after the agent has finished, so
that 400
+ # would fail the task rather than reach the model: resolve the
path in
+ # the guest instead, so the checks and the download see one file.
+ # Only realpath's own newline is dropped: whitespace at the end of
a
+ # file name is part of the name.
+ source = self._run_helper(
+ sandbox, f"realpath -e -- {quoted}", operation=f"resolve
{path!r}"
+ ).stdout.removesuffix("\n")
+ written = 0
+ with closing(self._download(sandbox, source, max_bytes=max_bytes,
operation="export")) as chunks:
+ for chunk in chunks:
+ _check_export_deadline(path, deadline, max_bytes)
+ written += len(chunk)
+ if written > max_bytes:
+ raise SandboxFileTooLargeError(path, written, max_bytes)
+ dest.write(chunk)
+ self._check_export_size(path, expected=size, written=written)
+ return written
+
+ def write_file(self, sandbox: str, path: str, content: bytes) -> None:
+ quoted = shlex.quote(path)
+ self._run_helper(
+ sandbox,
+ f'mkdir -p -- "$(dirname -- {quoted})"',
+ operation=f"create the parent directory for {path!r}",
+ )
+ client = self._get_client()
+ try:
+ client.sandboxes.upload_file(
+ sandbox,
+ path=path,
+ file=("upload", content, "application/octet-stream"),
+
request_options=self._request_options(timeout=_FILE_OP_TIMEOUT),
+ )
+ except Exception as e:
+ self._raise_file_op_error(
+ sandbox,
+ path,
+ e,
+ operation="write",
+ hint=(
+ "Islo answers this way when the path is a directory or
under a"
+ " pseudo-filesystem such as /proc or /sys."
+ ),
+ )
+
+ def list_directory(self, sandbox: str, path: str) -> list[tuple[str,
bool]]:
+ quoted = shlex.quote(path)
+ result = self._run_helper(
+ sandbox,
+ f"find -- {quoted} -maxdepth 1 -mindepth 1 -printf '%y %f\\0'",
+ operation=f"list {path!r}",
+ )
+ records = result.stdout.split("\0")
+ if result.stdout_truncated and records:
+ # Both the wrapper and the server keep the tail, so the cut is at
+ # the head: drop the leading record rather than report a mangled
+ # entry name the model cannot open.
+ records = records[1:]
+ entries: list[tuple[str, bool]] = []
+ for record in records:
+ if not record:
+ continue
+ kind, _, name = record.partition(" ")
+ if name:
+ entries.append((name, kind == "d"))
+ return entries
+
+ def destroy(self, sandbox: str) -> None:
+ client = self._get_client()
+ from islo.errors import NotFoundError
+
+ try:
+ client.sandboxes.delete_sandbox(
+ sandbox_name=sandbox,
+ # Deletion is idempotent, and it is the one call whose failure
+ # strands a microVM, so make the SDK's retries explicit here.
+
request_options=self._request_options(timeout=_FILE_OP_TIMEOUT, max_retries=2),
+ )
+ except NotFoundError:
+ return
+ except Exception as e:
+ _raise_translated(e, "delete a sandbox")
diff --git a/providers/common/ai/tests/system/common/ai/README.md
b/providers/common/ai/tests/system/common/ai/README.md
index 8d6484ea758..bf3a225a59b 100644
--- a/providers/common/ai/tests/system/common/ai/README.md
+++ b/providers/common/ai/tests/system/common/ai/README.md
@@ -52,6 +52,18 @@ modal token new
pytest --system
providers/common/ai/tests/system/common/ai/example_sandbox_toolset_modal.py
```
+## Islo
+
+Install the Islo extra and export a short-lived API key into the task process:
+
+```console
+pip install "apache-airflow-providers-common-ai[islo]"
+export ISLO_API_KEY="..."
+pytest --system
providers/common/ai/tests/system/common/ai/example_sandbox_toolset_islo.py
+```
+
+The test requests the default no-egress policy and a 15-minute server-side TTL.
+
## OpenSandbox
Install the OpenSandbox extra, point the SDK at a running server, and run:
diff --git
a/providers/common/ai/tests/system/common/ai/example_sandbox_toolset_islo.py
b/providers/common/ai/tests/system/common/ai/example_sandbox_toolset_islo.py
new file mode 100644
index 00000000000..30f2bb89cad
--- /dev/null
+++ b/providers/common/ai/tests/system/common/ai/example_sandbox_toolset_islo.py
@@ -0,0 +1,134 @@
+# 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.
+"""End-to-end system test for SandboxToolset with Islo."""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+
+from airflow.providers.common.compat.sdk import dag as airflow_dag, task
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = f"common_ai_sandbox_toolset_islo_{ENV_ID}" if ENV_ID else
"common_ai_sandbox_toolset_islo"
+
+MARKER = "boundary-ok"
+STATE_PATH = "/tmp/airflow_sandbox_e2e"
+
+
+@airflow_dag(
+ dag_id=DAG_ID,
+ schedule="@once",
+ start_date=datetime(2024, 1, 1, tzinfo=UTC),
+ catchup=False,
+ tags=["common.ai", "sandbox", "islo", "system_test"],
+)
+def example_sandbox_toolset_islo():
+ @task
+ def run_sandbox_agent() -> str:
+ from pydantic_ai import Agent
+ from pydantic_ai.messages import ModelMessage, ModelResponse,
TextPart, ToolCallPart
+ from pydantic_ai.models.function import AgentInfo, FunctionModel
+
+ from airflow.providers.common.ai.sandbox import IsloSandboxBackend
+ from airflow.providers.common.ai.toolsets import SandboxToolset
+
+ def model_function(messages: list[ModelMessage], _info: AgentInfo) ->
ModelResponse:
+ returns = [
+ part.content
+ for message in messages
+ for part in message.parts
+ if part.part_kind == "tool-return"
+ ]
+
+ if not returns:
+ return ModelResponse(
+ parts=[
+ ToolCallPart(
+ tool_name="write_file",
+ args={"path": STATE_PATH, "content": MARKER},
+ tool_call_id="write",
+ )
+ ]
+ )
+ if "Wrote" not in str(returns[0]):
+ raise RuntimeError(f"Unexpected write_file result:
{returns[0]!r}")
+
+ if len(returns) == 1:
+ return ModelResponse(
+ parts=[
+ ToolCallPart(
+ tool_name="run_command",
+ args={"command": f"cat {STATE_PATH} && echo $((6 *
7))"},
+ tool_call_id="shell",
+ )
+ ]
+ )
+ shell_out = str(returns[1])
+ if MARKER not in shell_out or "42" not in shell_out:
+ raise RuntimeError(f"Unexpected run_command result:
{shell_out!r}")
+
+ if len(returns) == 2:
+ return ModelResponse(
+ parts=[
+ ToolCallPart(tool_name="read_file", args={"path":
STATE_PATH}, tool_call_id="read")
+ ]
+ )
+ if MARKER not in str(returns[2]):
+ raise RuntimeError(f"Unexpected read_file result:
{returns[2]!r}")
+
+ if len(returns) == 3:
+ return ModelResponse(
+ parts=[
+ ToolCallPart(
+ tool_name="run_command",
+ args={"command": "echo to-stderr >&2; exit 3"},
+ tool_call_id="fail",
+ )
+ ]
+ )
+ failed = str(returns[3])
+ if "[exit code: 3]" not in failed or "to-stderr" not in failed:
+ raise RuntimeError(f"Unexpected failure result: {failed!r}")
+
+ if len(returns) == 4:
+ return ModelResponse(
+ parts=[ToolCallPart(tool_name="list_directory",
args={"path": "/tmp"}, tool_call_id="ls")]
+ )
+ if "airflow_sandbox_e2e" not in str(returns[4]):
+ raise RuntimeError(f"Unexpected listing: {returns[4]!r}")
+
+ return ModelResponse(parts=[TextPart(content="sandbox boundary e2e
passed")])
+
+ agent = Agent(
+ FunctionModel(model_function),
+ instructions="Use the sandbox tools as requested.",
+ toolsets=[SandboxToolset(IsloSandboxBackend(delete_after=900))],
+ )
+ result = agent.run_sync("Run the sandbox boundary system test.")
+ if result.output != "sandbox boundary e2e passed":
+ raise RuntimeError(f"Unexpected agent output: {result.output!r}")
+ return result.output
+
+ run_sandbox_agent()
+
+
+dag = example_sandbox_toolset_islo()
+
+from tests_common.test_utils.system_tests import get_test_run # noqa: E402
+
+test_run = get_test_run(dag)
diff --git a/providers/common/ai/tests/unit/common/ai/sandbox/test_islo.py
b/providers/common/ai/tests/unit/common/ai/sandbox/test_islo.py
new file mode 100644
index 00000000000..285681a6c3b
--- /dev/null
+++ b/providers/common/ai/tests/unit/common/ai/sandbox/test_islo.py
@@ -0,0 +1,1276 @@
+# 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.
+from __future__ import annotations
+
+import builtins
+import io
+import os
+import signal
+import subprocess
+import time
+from contextlib import suppress
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+
+pytest.importorskip("islo")
+
+import httpx
+from islo import Islo
+from islo.core.api_error import ApiError
+from islo.errors import NotFoundError
+from islo.sandboxes.client import SandboxesClient
+
+from airflow.providers.common.ai.sandbox.base import (
+ SandboxError,
+ SandboxExecResult,
+ SandboxFileTooLargeError,
+ SandboxSpec,
+ SandboxTerminalError,
+)
+from airflow.providers.common.ai.sandbox.islo import (
+ _COMMAND_WRAPPER,
+ _SERVER_STREAM_CAP,
+ IsloSandboxBackend,
+ _bound_result_stream,
+)
+
+_MODULE = "airflow.providers.common.ai.sandbox.islo"
+_ISLO_PATH = "islo.Islo"
+
+
+def _exec_result(status="completed", exit_code=0, stdout="", stderr="",
truncated=False):
+ return SimpleNamespace(
+ status=status, exit_code=exit_code, stdout=stdout, stderr=stderr,
truncated=truncated
+ )
+
+
+def _sandbox_info(name="box-1", status="running", deleted_at=None):
+ return SimpleNamespace(name=name, status=status, deleted_at=deleted_at)
+
+
+def _token_error(status_code):
+ """How the SDK's ``ISLO_API_KEY`` token exchange fails: a bare ``httpx``
error, not an ``ApiError``."""
+ request = httpx.Request("POST", "https://api.islo.dev/auth/token";)
+ return httpx.HTTPStatusError(
+ "token exchange failed", request=request,
response=httpx.Response(status_code, request=request)
+ )
+
+
+def _export_check(size):
+ """What the guest's export check prints for a regular file of ``size``
bytes."""
+ return _exec_result(stdout=f"\nairflow-export-size:{size}\n")
+
+
+def _backend_with_client(**kwargs) -> tuple[IsloSandboxBackend,
mock.MagicMock]:
+ backend = IsloSandboxBackend(**kwargs)
+ client = mock.MagicMock(spec=["sandboxes"])
+ client.sandboxes = mock.create_autospec(SandboxesClient, instance=True)
+ client.sandboxes.exec_in_sandbox.return_value =
SimpleNamespace(exec_id="exec-1")
+ client.sandboxes.create_sandbox.return_value = _sandbox_info()
+ client.sandboxes.get_sandbox.return_value = _sandbox_info()
+ client.sandboxes.get_exec_result.return_value = _exec_result()
+ backend._client = client
+ return backend, client
+
+
+class TestCredentials:
+ @mock.patch(_ISLO_PATH, autospec=True)
+ def test_client_comes_from_the_sdk_environment(self, islo, monkeypatch):
+ monkeypatch.setenv("ISLO_API_KEY", "ak_test")
+
+ client = IsloSandboxBackend()._get_client()
+
+ islo.assert_called_once_with()
+ assert client is islo.return_value
+
+ @mock.patch(_ISLO_PATH, autospec=True)
+ def test_client_is_resolved_once_and_cached(self, islo, monkeypatch):
+ monkeypatch.setenv("ISLO_API_KEY", "ak_test")
+ backend = IsloSandboxBackend()
+
+ backend._get_client()
+ backend._get_client()
+
+ islo.assert_called_once_with()
+
+ @pytest.mark.parametrize("value", [None, ""], ids=["unset", "empty"])
+ @mock.patch(_ISLO_PATH, autospec=True)
+ def test_a_missing_api_key_is_terminal_and_names_the_variable(self, islo,
monkeypatch, value):
+ if value is None:
+ monkeypatch.delenv("ISLO_API_KEY", raising=False)
+ else:
+ monkeypatch.setenv("ISLO_API_KEY", value)
+
+ with pytest.raises(SandboxTerminalError, match="ISLO_API_KEY"):
+ IsloSandboxBackend()._get_client()
+
+ islo.assert_not_called()
+
+ def test_a_rejected_api_key_names_the_variable_and_status(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.create_sandbox.side_effect = _token_error(401)
+
+ with pytest.raises(SandboxTerminalError, match=r"Islo rejected
ISLO_API_KEY \(HTTP 401\)"):
+ backend.create()
+
+ def test_missing_sdk_error_is_actionable(self):
+ real_import = builtins.__import__
+
+ def blocked_import(name, *args, **kwargs):
+ if name.startswith("islo"):
+ raise ImportError("blocked for test")
+ return real_import(name, *args, **kwargs)
+
+ backend = IsloSandboxBackend()
+ with mock.patch("builtins.__import__", side_effect=blocked_import):
+ with pytest.raises(SandboxTerminalError, match=r"\[islo\]"):
+ backend.create()
+
+
[email protected](
+ ("kwargs", "message"),
+ [
+ ({"image": ""}, "image"),
+ ({"vcpus": 0}, "vcpus"),
+ ({"memory_mb": 0}, "memory_mb"),
+ ({"delete_after": 0}, "delete_after"),
+ ({"pause_after_idle": 0}, "pause_after_idle"),
+ ({"auto_resume": "sometimes"}, "auto_resume"),
+ ],
+)
+def test_constructor_rejects_invalid_values(kwargs, message):
+ with pytest.raises(ValueError, match=message):
+ IsloSandboxBackend(**kwargs)
+
+
+class TestCreate:
+ @pytest.mark.parametrize(
+ ("spec", "match"),
+ [
+ (SandboxSpec(owner="example_dag/manual__1"), "owner"),
+ (SandboxSpec(allow_egress_to=["example.com"]), "per-domain egress
allowlist"),
+ # internet_enabled is all-or-nothing, so honouring
block_network=True alone
+ # would silently drop the ranges the Dag author asked to reach.
+ (SandboxSpec(allow_egress_to_cidrs=["203.0.113.0/24"]),
"allow_egress_to_cidrs"),
+ (SandboxSpec(env={"PATH": "/opt/tool/bin"}), "PATH"),
+ ],
+ ids=["owner", "domain-allowlist", "cidr-allowlist", "path"],
+ )
+ def test_refuses_a_spec_it_cannot_carry_before_provisioning(self, spec,
match):
+ backend, client = _backend_with_client()
+
+ with pytest.raises(SandboxTerminalError, match=match):
+ backend.create(spec=spec)
+
+ client.sandboxes.create_sandbox.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("spec", "expected"),
+ [
+ (None, False),
+ (SandboxSpec(), False),
+ (SandboxSpec(block_network=True), False),
+ (SandboxSpec(block_network=False), True),
+ ],
+ )
+ def test_block_network_maps_to_internet_enabled(self, spec, expected):
+ backend, client = _backend_with_client()
+
+ backend.create(spec=spec)
+
+ assert
client.sandboxes.create_sandbox.call_args.kwargs["internet_enabled"] is expected
+
+ def test_spec_and_sizing_are_passed_at_creation(self):
+ backend, client = _backend_with_client(
+ image="python",
+ vcpus=2,
+ memory_mb=1024,
+ pause_after_idle=300,
+ auto_resume="never",
+ delete_after=120,
+ )
+
+ name = backend.create(spec=SandboxSpec(env={"TOKEN": "value"}))
+
+ assert name == "box-1"
+ kwargs = client.sandboxes.create_sandbox.call_args.kwargs
+ assert kwargs["image"] == "python"
+ assert kwargs["vcpus"] == 2
+ assert kwargs["memory_mb"] == 1024
+ assert kwargs["env"] == {"TOKEN": "value"}
+ assert kwargs["lifecycle"].pause_after_idle == 300
+ assert kwargs["lifecycle"].auto_resume == "never"
+ assert kwargs["lifecycle"].delete_after == 120
+ assert kwargs["request_options"] == {"timeout_in_seconds": 120}
+
+ def
test_default_lifecycle_pauses_idle_sandboxes_and_deletes_after_a_day(self):
+ backend, client = _backend_with_client()
+
+ backend.create()
+
+ lifecycle =
client.sandboxes.create_sandbox.call_args.kwargs["lifecycle"]
+ assert lifecycle.pause_after_idle == 600
+ assert lifecycle.auto_resume == "on_activity"
+ assert lifecycle.delete_after == 86400
+
+ def test_disabled_lifecycle_timers_are_sent_as_unset(self):
+ backend, client = _backend_with_client(pause_after_idle=None,
delete_after=None)
+
+ backend.create()
+
+ lifecycle =
client.sandboxes.create_sandbox.call_args.kwargs["lifecycle"]
+ assert lifecycle.pause_after_idle is None
+ assert lifecycle.delete_after is None
+
+ def test_omitted_sizing_is_left_to_the_server(self):
+ backend, client = _backend_with_client()
+
+ backend.create()
+
+ assert not {"image", "vcpus", "memory_mb"} &
client.sandboxes.create_sandbox.call_args.kwargs.keys()
+
+ def test_api_failure_is_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.create_sandbox.side_effect = ApiError(status_code=503)
+
+ with pytest.raises(SandboxTerminalError, match="HTTP 503"):
+ backend.create()
+
+ def test_a_failed_create_deletes_the_name_it_had_already_bound(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.create_sandbox.side_effect = ApiError(status_code=503)
+
+ with pytest.raises(SandboxTerminalError):
+ backend.create()
+
+ # The server may have provisioned the microVM before failing to answer,
+ # and this name is the only handle that can still reclaim it.
+ requested = client.sandboxes.create_sandbox.call_args.kwargs["name"]
+ client.sandboxes.delete_sandbox.assert_called_once()
+ assert
client.sandboxes.delete_sandbox.call_args.kwargs["sandbox_name"] == requested
+
+ def test_a_cleanup_failure_does_not_mask_the_original_create_error(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.create_sandbox.side_effect = ApiError(status_code=503)
+ client.sandboxes.delete_sandbox.side_effect = ApiError(status_code=500)
+
+ with pytest.raises(SandboxTerminalError, match="HTTP 503"):
+ backend.create()
+
+ def
test_a_sandbox_that_cannot_serve_after_creation_is_destroyed_and_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.create_sandbox.return_value =
_sandbox_info(status="stopped")
+
+ with pytest.raises(SandboxTerminalError, match="cannot serve
requests"):
+ backend.create()
+
+ client.sandboxes.delete_sandbox.assert_called_once()
+
+ def test_an_empty_spec_env_sends_no_env(self):
+ backend, client = _backend_with_client()
+
+ backend.create(spec=SandboxSpec(env={}))
+
+ assert "env" not in client.sandboxes.create_sandbox.call_args.kwargs
+
+
+class TestRunCommand:
+ def test_polls_with_backoff(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.side_effect = [
+ _exec_result(status="running"),
+ _exec_result(status="running"),
+ _exec_result(status="running"),
+ _exec_result(stdout="done"),
+ ]
+
+ with mock.patch("time.sleep", autospec=True) as sleep:
+ result = backend.run_command("box", "x", timeout=60,
max_output_bytes=1024)
+
+ intervals = [call.args[0] for call in sleep.call_args_list]
+ assert result.stdout == "done"
+ assert intervals == sorted(intervals)
+ assert intervals[-1] > intervals[0]
+
+ def test_user_command_is_an_argument_to_the_bounding_wrapper(self):
+ backend, client = _backend_with_client()
+ user_command = "echo '$HOME'; rm -f /tmp/nope"
+
+ backend.run_command("box", user_command, timeout=5,
max_output_bytes=1024)
+
+ command = client.sandboxes.exec_in_sandbox.call_args.kwargs["command"]
+ assert command[:2] == ["sh", "-c"]
+ assert user_command not in command[2]
+ # One byte over the budget, so an over-budget stream proves truncation.
+ assert command[4:] == [user_command, "1025"]
+
+ def test_budget_above_the_server_cap_is_clamped_to_it(self):
+ backend, client = _backend_with_client()
+
+ backend.run_command("box", "x", timeout=5, max_output_bytes=5 *
_SERVER_STREAM_CAP)
+
+ command = client.sandboxes.exec_in_sandbox.call_args.kwargs["command"]
+ assert command[-1] == str(_SERVER_STREAM_CAP)
+
+ def test_polls_keep_the_sdk_retries_but_the_start_is_sent_once(self):
+ backend, client = _backend_with_client()
+
+ backend.run_command("box", "x", timeout=5, max_output_bytes=1024)
+
+ assert
client.sandboxes.exec_in_sandbox.call_args.kwargs["request_options"]["max_retries"]
== 0
+ assert "max_retries" not in
client.sandboxes.get_exec_result.call_args.kwargs["request_options"]
+
+ @staticmethod
+ def _real_client(monkeypatch, first_start):
+ """
+ A real SDK client whose first command start fails with ``first_start``.
+
+ The SDK's own retries are what the backend has to keep off this POST,
so
+ they must run for real rather than through a mocked
``SandboxesClient``.
+ """
+ # With a key the SDK exchanges it for a token over the network,
outside this
+ # transport, before the start is ever sent.
+ monkeypatch.delenv("ISLO_API_KEY", raising=False)
+ posts = []
+
+ def handler(request: httpx.Request) -> httpx.Response:
+ if request.method == "GET":
+ return httpx.Response(
+ 200,
+ json={
+ "exec_id": "exec-2",
+ "status": "completed",
+ "exit_code": 0,
+ "stdout": "ok\n",
+ "stderr": "",
+ "truncated": False,
+ },
+ )
+ posts.append(request.url.path)
+ if len(posts) == 1:
+ if isinstance(first_start, Exception):
+ raise first_start
+ return first_start
+ return httpx.Response(200, json={"exec_id": "exec-2",
"sandbox_id": "sb-1", "status": "running"})
+
+ client = Islo(
+ base_url="https://islo.test";,
+ compute_url="https://compute.islo.test";,
+ httpx_client=httpx.Client(transport=httpx.MockTransport(handler)),
+ )
+ return client, posts
+
+ @pytest.mark.parametrize(
+ "first_start",
+ [httpx.RemoteProtocolError("connection dropped"), httpx.Response(503)],
+ ids=["lost-response", "5xx"],
+ )
+ @mock.patch(f"{_MODULE}.time.sleep", autospec=True)
+ def test_a_start_that_may_have_run_is_never_resent(self, sleep,
first_start, monkeypatch):
+ # Islo may have started the command before the response was lost or
the 5xx
+ # was sent, so a resend could run a "git push" or an ">> file" twice.
+ backend = IsloSandboxBackend()
+ backend._client, posts = self._real_client(monkeypatch, first_start)
+
+ with pytest.raises(SandboxTerminalError, match="start a sandbox
command"):
+ backend.run_command("box", "x", timeout=5, max_output_bytes=1024)
+
+ assert len(posts) == 1
+
+ @pytest.mark.parametrize(
+ "first_start",
+ [httpx.ConnectError("connection refused"), httpx.Response(429)],
+ ids=["no-connection", "rate-limited"],
+ )
+ @mock.patch(f"{_MODULE}.time.sleep", autospec=True)
+ def test_a_start_that_never_ran_is_resent(self, sleep, first_start,
monkeypatch):
+ backend = IsloSandboxBackend()
+ backend._client, posts = self._real_client(monkeypatch, first_start)
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert len(posts) == 2
+ assert result.stdout == "ok\n"
+ sleep.assert_called_once_with(1.0)
+
+ @pytest.mark.parametrize("status_code", [429, 503])
+ def test_a_start_whose_token_refresh_failed_transiently_is_resent(self,
status_code):
+ # The token refresh runs before the start is sent, so the command
never ran.
+ backend, client = _backend_with_client()
+ client.sandboxes.exec_in_sandbox.side_effect = [
+ _token_error(status_code),
+ SimpleNamespace(exec_id="exec-1"),
+ ]
+
+ with mock.patch(f"{_MODULE}.time.sleep", autospec=True):
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert client.sandboxes.exec_in_sandbox.call_count == 2
+ assert result.exit_code == 0
+
+ @pytest.mark.parametrize("status_code", [401, 403])
+ def test_a_start_with_a_rejected_key_is_not_resent(self, status_code):
+ backend, client = _backend_with_client()
+ client.sandboxes.exec_in_sandbox.side_effect =
_token_error(status_code)
+
+ with mock.patch(f"{_MODULE}.time.sleep", autospec=True):
+ with pytest.raises(SandboxTerminalError, match=rf"rejected
ISLO_API_KEY \(HTTP {status_code}\)"):
+ backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert client.sandboxes.exec_in_sandbox.call_count == 1
+
+ def test_a_start_that_keeps_failing_to_connect_gives_up(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.exec_in_sandbox.side_effect =
httpx.ConnectError("connection refused")
+
+ with mock.patch(f"{_MODULE}.time.sleep", autospec=True):
+ with pytest.raises(SandboxTerminalError, match="start a sandbox
command"):
+ backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert client.sandboxes.exec_in_sandbox.call_count == 3
+
+ def test_a_stream_within_budget_is_passed_through_untouched(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(stdout="a\nb\n", stderr="err\n")
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert result.stdout == "a\nb\n"
+ assert result.stderr == "err\n"
+ assert not result.stdout_truncated
+ assert not result.stderr_truncated
+
+ def test_an_over_budget_stream_keeps_the_tail_and_reports_truncation(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _exec_result(
+ stdout="line1\nline2\nline3\n", stderr="e1\ne2\ne3\n"
+ )
+
+ result = backend.run_command("box", "x", timeout=5, max_output_bytes=8)
+
+ assert result.stdout == "line3\n"
+ assert result.stderr == "e2\ne3\n"
+ assert result.stdout_truncated
+ assert result.stderr_truncated
+
+ def test_truncation_drops_a_short_partial_leading_line(self):
+ backend, client = _backend_with_client()
+ # A byte-aligned cut of the last 12 bytes would land inside "line988".
+ client.sandboxes.get_exec_result.return_value =
_exec_result(stdout="line988\nline989\n")
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=12)
+
+ assert result.stdout == "line989\n"
+ assert result.stdout_truncated
+
+ @pytest.mark.parametrize(
+ ("stdout", "expected"),
+ [
+ ("abcdefghij", "cdefghij"),
+ # The line's only newline is its last byte, so dropping through it
+ # would leave nothing and the model would read "(no output)".
+ ("x" * 20 + "\n", "xxxxxxx\n"),
+ ("z" * 20 + "\nok\n", "zzzz\nok\n"),
+ ],
+ ids=["no-newline", "trailing-newline", "short-line-after"],
+ )
+ def test_a_single_line_over_budget_is_cut_rather_than_dropped(self,
stdout, expected):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(stdout=stdout)
+
+ result = backend.run_command("box", "x", timeout=5, max_output_bytes=8)
+
+ assert result.stdout == expected
+ assert result.stdout_truncated
+
+ @pytest.mark.parametrize(
+ ("max_bytes", "expected"),
+ [(4, "éé"), (3, "\ufffdé")],
+ ids=["whole-characters", "cut-inside-a-character"],
+ )
+ def test_applies_the_byte_cap_to_utf8_and_replaces_a_split_character(self,
max_bytes, expected):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(stdout="ééé")
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=max_bytes)
+
+ assert result.stdout == expected
+ assert result.stdout_truncated
+
+ def test_the_server_flag_alone_does_not_mark_streams_that_fit(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _exec_result(
+ stdout="a\n", stderr="b\n", truncated=True
+ )
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert not result.stdout_truncated
+ assert not result.stderr_truncated
+
+ @mock.patch.object(IsloSandboxBackend, "_await_exec", autospec=True,
return_value=None)
+ def test_poll_deadline_destroys_the_sandbox(self, _await_exec):
+ backend, client = _backend_with_client()
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert result.timed_out
+ assert result.sandbox_terminated
+ client.sandboxes.delete_sandbox.assert_called_once()
+
+ @pytest.mark.parametrize(
+ ("delete_after", "reclaim"),
+ [
+ (3600, "its delete_after policy removes it 3600s after creation"),
+ (None, "delete_after is disabled, so it persists until deleted by
hand"),
+ ],
+ )
+ @mock.patch(f"{_MODULE}.log", autospec=True)
+ @mock.patch.object(IsloSandboxBackend, "_await_exec", autospec=True,
return_value=None)
+ def
test_timeout_cleanup_failure_warns_and_says_whether_anything_reclaims_it(
+ self, _await_exec, logger, delete_after, reclaim
+ ):
+ backend, client = _backend_with_client(delete_after=delete_after)
+ client.sandboxes.delete_sandbox.side_effect = ApiError(status_code=503)
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ # A command that merely ran long must not fail the task because one
+ # cleanup call was refused, but the operator must learn whether the
+ # microVM will be reclaimed at all.
+ assert result.timed_out
+ assert result.sandbox_terminated
+ assert "could not confirm its deletion" in
logger.warning.call_args.args[0]
+ assert logger.warning.call_args.args[2] == reclaim
+
+ @pytest.mark.parametrize("status", ["cancelled", "dead", "something-new"])
+ def
test_an_unrecognised_status_is_terminal_rather_than_still_running(self, status):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(status=status, exit_code=None)
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ # The vendor types status as a plain string, so the vocabulary can
grow.
+ # Reading an unknown value as still-running would poll to the deadline
+ # and then destroy the sandbox, costing the agent its files.
+ assert result.exit_code == -1
+ assert not result.timed_out
+ assert not result.sandbox_terminated
+ client.sandboxes.delete_sandbox.assert_not_called()
+
+ @pytest.mark.parametrize("status", ["pending", "queued", "starting",
"running"])
+ def test_in_flight_statuses_keep_polling(self, status):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.side_effect = [
+ _exec_result(status=status),
+ _exec_result(stdout="eventually\n"),
+ ]
+
+ with mock.patch("time.sleep", autospec=True):
+ result = backend.run_command("box", "x", timeout=60,
max_output_bytes=1024)
+
+ assert result.stdout == "eventually\n"
+
+ def test_server_timeout_also_destroys_the_sandbox(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(status="timeout", exit_code=None)
+
+ result = backend.run_command("box", "x", timeout=5,
max_output_bytes=1024)
+
+ assert result.timed_out
+ assert result.exit_code == -1
+ assert result.sandbox_terminated
+ client.sandboxes.delete_sandbox.assert_called_once()
+
+ def test_start_failure_is_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.exec_in_sandbox.side_effect =
ApiError(status_code=401)
+
+ with pytest.raises(SandboxTerminalError, match="HTTP 401"):
+ backend.run_command("box", "x", timeout=5, max_output_bytes=1024)
+
+ def test_transient_poll_errors_are_ridden_out(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.side_effect = [
+ ApiError(status_code=502),
+ httpx.ReadError("connection reset"),
+ ApiError(status_code=None),
+ _token_error(503),
+ _exec_result(stdout="done\n"),
+ ]
+
+ with mock.patch("time.sleep", autospec=True):
+ result = backend.run_command("box", "x", timeout=60,
max_output_bytes=1024)
+
+ assert result.stdout == "done\n"
+ assert not result.timed_out
+ client.sandboxes.delete_sandbox.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("error", "message"),
+ [
+ (ApiError(status_code=404), "poll a sandbox command"),
+ (RuntimeError("bad client state"), "poll a sandbox command"),
+ (_token_error(401), r"rejected ISLO_API_KEY \(HTTP 401\)"),
+ (_token_error(403), r"rejected ISLO_API_KEY \(HTTP 403\)"),
+ (_token_error(400), r"poll a sandbox command \(HTTP 400\)"),
+ ],
+ ids=["404", "client-error", "rejected-key", "forbidden-key",
"token-4xx"],
+ )
+ def test_a_non_transient_poll_error_is_terminal_at_once(self, error,
message):
+ backend, client = _backend_with_client()
+ # A misclassified error would be ridden out to this result, so the
test fails
+ # fast instead of spinning to the deadline.
+ client.sandboxes.get_exec_result.side_effect = [error, _exec_result()]
+
+ with mock.patch("time.sleep", autospec=True):
+ with pytest.raises(SandboxTerminalError, match=message):
+ backend.run_command("box", "x", timeout=60,
max_output_bytes=1024)
+
+ # Raised on the first poll, not ridden out to the deadline.
+ assert client.sandboxes.get_exec_result.call_count == 1
+
+ def
test_transient_errors_lasting_past_the_deadline_are_terminal_not_a_timeout(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.side_effect =
ApiError(status_code=503)
+
+ with pytest.raises(SandboxTerminalError, match="poll a sandbox
command"):
+ backend.run_command("box", "x", timeout=0.05,
max_output_bytes=1024)
+
+ # Nothing was heard from the command, so claiming it timed out would
be a guess.
+ client.sandboxes.delete_sandbox.assert_not_called()
+
+ @pytest.mark.parametrize("request_name", ["exec_in_sandbox",
"get_exec_result"])
+ def test_the_start_and_the_last_poll_keep_a_minimum_http_budget(self,
request_name):
+ backend, client = _backend_with_client()
+
+ backend.run_command("box", "x", timeout=1, max_output_bytes=1024)
+
+ options = getattr(client.sandboxes,
request_name).call_args.kwargs["request_options"]
+ assert options["timeout_in_seconds"] == 5
+
+ def test_a_slow_start_leaves_the_command_its_whole_budget(self,
monkeypatch):
+ backend, client = _backend_with_client()
+ clock = [1000.0]
+ monkeypatch.setattr(f"{_MODULE}.time.monotonic", lambda: clock[0])
+
+ def slow_start(*args, **kwargs):
+ # Longer than the command's whole budget, as the first call after
an
+ # auto-resume can be.
+ clock[0] += 10
+ return SimpleNamespace(exec_id="exec-1")
+
+ client.sandboxes.exec_in_sandbox.side_effect = slow_start
+ client.sandboxes.get_exec_result.return_value =
_exec_result(stdout="hi\n")
+
+ result = backend.run_command("box", "echo hi", timeout=1,
max_output_bytes=1024)
+
+ assert result.stdout == "hi\n"
+ assert not result.timed_out
+ client.sandboxes.delete_sandbox.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("timeout", "max_bytes", "message"), [(0, 1, "timeout"), (1, 0,
"max_output_bytes")]
+ )
+ def test_rejects_invalid_budgets(self, timeout, max_bytes, message):
+ backend, _ = _backend_with_client()
+
+ with pytest.raises(ValueError, match=message):
+ backend.run_command("box", "x", timeout=timeout,
max_output_bytes=max_bytes)
+
+
+class TestFileOperations:
+ def test_read_file_uses_the_native_streaming_api(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.return_value = iter([b"he", b"llo"])
+
+ data = backend.read_file("box", "/w/a", max_bytes=100)
+
+ assert data == b"hello"
+ client.sandboxes.download_file.assert_called_once_with(
+ "box",
+ path="/w/a",
+ request_options={"timeout_in_seconds": 120, "chunk_size": 101},
+ )
+ client.sandboxes.exec_in_sandbox.assert_not_called()
+
+ def
test_oversized_read_stops_closes_the_stream_and_reports_the_real_size(self):
+ backend, client = _backend_with_client()
+ closed: list[bool] = []
+
+ def chunks():
+ try:
+ yield b"x" * 11
+ raise AssertionError("the backend must stop after the sentinel
byte")
+ finally:
+ closed.append(True)
+
+ client.sandboxes.download_file.return_value = chunks()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(stdout="123456\n")
+
+ with pytest.raises(SandboxFileTooLargeError) as error:
+ backend.read_file("box", "/w/a", max_bytes=10)
+
+ assert closed == [True]
+ assert error.value.size_bytes == 123456
+ assert
client.sandboxes.exec_in_sandbox.call_args.kwargs["command"][4].startswith("stat
-Lc %s -- ")
+
+ def
test_oversized_read_without_a_size_says_so_rather_than_inventing_one(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.return_value = iter([b"x" * 11])
+ client.sandboxes.get_exec_result.return_value =
_exec_result(exit_code=1, stderr="stat: no such file")
+
+ with pytest.raises(SandboxError, match="larger than the 10 byte read
limit") as error:
+ backend.read_file("box", "/w/a", max_bytes=10)
+
+ assert not isinstance(error.value, (SandboxTerminalError,
SandboxFileTooLargeError))
+
+ def test_missing_file_is_recoverable_when_the_sandbox_exists(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.side_effect = NotFoundError({})
+
+ with pytest.raises(SandboxError, match="does not exist") as error:
+ backend.read_file("box", "/w/missing", max_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ client.sandboxes.get_sandbox.assert_called_once()
+
+ @pytest.mark.parametrize(
+ "info",
+ [_sandbox_info(status="stopped"),
_sandbox_info(deleted_at="2026-09-19T00:00:00Z")],
+ ids=["stopped", "deleted"],
+ )
+ def test_missing_file_on_a_sandbox_that_cannot_serve_is_terminal(self,
info):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.side_effect = NotFoundError({})
+ client.sandboxes.get_sandbox.return_value = info
+
+ with pytest.raises(SandboxTerminalError, match="cannot serve
requests"):
+ backend.read_file("box", "/w/a", max_bytes=100)
+
+ @pytest.mark.parametrize(("auto_resume", "terminal"), [("on_activity",
False), ("never", True)])
+ def test_a_paused_sandbox_is_usable_only_when_it_resumes_on_activity(self,
auto_resume, terminal):
+ backend, client = _backend_with_client(auto_resume=auto_resume)
+ client.sandboxes.download_file.side_effect = NotFoundError({})
+ client.sandboxes.get_sandbox.return_value =
_sandbox_info(status="paused")
+
+ with pytest.raises(SandboxError) as error:
+ backend.read_file("box", "/w/a", max_bytes=100)
+
+ assert isinstance(error.value, SandboxTerminalError) is terminal
+
+ def test_missing_sandbox_is_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.side_effect = NotFoundError({})
+ client.sandboxes.get_sandbox.side_effect = NotFoundError({})
+
+ with pytest.raises(SandboxTerminalError, match="check a sandbox"):
+ backend.read_file("box", "/w/a", max_bytes=100)
+
+ @pytest.mark.parametrize(
+ "error",
+ [
+ ApiError(status_code=503),
+ ApiError(status_code=401),
+ ApiError(status_code=429),
+ httpx.ConnectError("reset"),
+ ],
+ ids=["503", "401", "429", "no-response"],
+ )
+ def test_a_failure_that_says_nothing_about_the_path_is_terminal(self,
error):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.side_effect = error
+
+ with pytest.raises(SandboxTerminalError, match="read a sandbox file"):
+ backend.read_file("box", "/w/a", max_bytes=100)
+
+ client.sandboxes.get_sandbox.assert_not_called()
+
+ def test_a_rejected_path_is_recoverable_and_carries_the_api_message(self):
+ # Measured: a relative path is a 400 whose message names the problem.
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.side_effect = ApiError(
+ status_code=400,
+ body={"code": "INVALID_REQUEST", "message": "path must be
absolute: rel.txt"},
+ )
+
+ with pytest.raises(SandboxError) as error:
+ backend.read_file("box", "rel.txt", max_bytes=100)
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ assert str(error.value) == "Could not read 'rel.txt' (HTTP 400): path
must be absolute: rel.txt."
+
+ def test_a_long_api_message_is_trimmed(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.download_file.side_effect = ApiError(status_code=400,
body={"message": "x" * 5000})
+
+ with pytest.raises(SandboxError) as error:
+ backend.read_file("box", "/w/a", max_bytes=100)
+
+ assert len(str(error.value)) < 300
+
+ def test_write_file_creates_parents_then_uses_native_upload(self):
+ backend, client = _backend_with_client()
+
+ backend.write_file("box", "/w/sub/a", b"data")
+
+ command = client.sandboxes.exec_in_sandbox.call_args.kwargs["command"]
+ assert "mkdir -p" in command[4]
+ client.sandboxes.upload_file.assert_called_once_with(
+ "box",
+ path="/w/sub/a",
+ file=("upload", b"data", "application/octet-stream"),
+ request_options={"timeout_in_seconds": 120},
+ )
+
+ def test_write_stops_when_parent_creation_fails(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(exit_code=1, stderr="read-only")
+
+ with pytest.raises(SandboxError, match="read-only"):
+ backend.write_file("box", "/w/a", b"data")
+
+ client.sandboxes.upload_file.assert_not_called()
+
+ def test_upload_outage_is_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.upload_file.side_effect = ApiError(status_code=503)
+
+ with pytest.raises(SandboxTerminalError, match="write a sandbox file"):
+ backend.write_file("box", "/w/a", b"data")
+
+ def test_an_unwritable_target_is_recoverable_with_a_hint(self):
+ # Measured: uploading onto a directory, or into /proc or /sys, is a
bare
+ # 500 whose body says only "An internal error occurred".
+ backend, client = _backend_with_client()
+ client.sandboxes.upload_file.side_effect = ApiError(
+ status_code=500, body={"code": "INTERNAL_ERROR", "message": "An
internal error occurred"}
+ )
+
+ with pytest.raises(SandboxError) as error:
+ backend.write_file("box", "/tmp", b"data")
+
+ assert not isinstance(error.value, SandboxTerminalError)
+ assert "(HTTP 500)." in str(error.value)
+ assert "directory or under a pseudo-filesystem such as /proc or /sys"
in str(error.value)
+ assert "internal error" not in str(error.value)
+ client.sandboxes.get_sandbox.assert_called_once()
+
+ def test_an_upload_error_on_a_sandbox_that_cannot_serve_is_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.upload_file.side_effect = ApiError(status_code=500)
+ client.sandboxes.get_sandbox.return_value =
_sandbox_info(status="stopped")
+
+ with pytest.raises(SandboxTerminalError, match="cannot serve
requests"):
+ backend.write_file("box", "/w/a", b"data")
+
+ def test_list_directory_marks_directories_and_preserves_newlines(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _exec_result(stdout="f
a.txt\0d new\nline\0")
+
+ assert backend.list_directory("box", "/w") == [("a.txt", False),
("new\nline", True)]
+
+ @mock.patch(f"{_MODULE}._HELPER_OUTPUT_CAP", 20)
+ def test_a_listing_cut_at_the_head_drops_only_the_leading_record(self):
+ backend, client = _backend_with_client()
+ # The 20-byte tail starts at " name\0", a fragment that would parse as
an
+ # entry called "name" if the leading record were kept.
+ client.sandboxes.get_exec_result.return_value = _exec_result(
+ stdout="f old file name\0d bbbb\0d cccc\0"
+ )
+
+ assert backend.list_directory("box", "/w") == [("bbbb", True),
("cccc", True)]
+
+ def test_the_server_flag_alone_leaves_a_short_listing_intact(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _exec_result(
+ stdout="f a.txt\0d sub\0", truncated=True
+ )
+
+ assert backend.list_directory("box", "/w") == [("a.txt", False),
("sub", True)]
+
+ def test_list_failure_is_recoverable(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(exit_code=1, stderr="missing")
+
+ with pytest.raises(SandboxError, match="missing"):
+ backend.list_directory("box", "/w")
+
+ def test_helper_timeout_is_terminal(self):
+ backend, _ = _backend_with_client()
+ backend.run_command = mock.create_autospec(
+ backend.run_command,
+ return_value=SandboxExecResult(
+ exit_code=-1,
+ stdout="",
+ stderr="",
+ timed_out=True,
+ sandbox_terminated=True,
+ ),
+ )
+
+ with pytest.raises(SandboxTerminalError, match="destroyed"):
+ backend.list_directory("box", "/w")
+
+ def test_read_rejects_an_invalid_budget(self):
+ backend, _ = _backend_with_client()
+
+ with pytest.raises(ValueError, match="max_bytes"):
+ backend.read_file("box", "/w/a", max_bytes=0)
+
+
+class TestExportFile:
+ def test_streams_a_file_larger_than_a_command_can_return(self):
+ backend, client = _backend_with_client()
+ chunks = [bytes([n]) * (512 * 1024) for n in range(3)]
+ data = b"".join(chunks)
+ client.sandboxes.get_exec_result.return_value =
_export_check(len(data))
+ client.sandboxes.download_file.return_value = iter(chunks)
+ dest = io.BytesIO()
+
+ written = backend.export_file("box", "/w/out.bin", dest, max_bytes=10
* 1024 * 1024)
+
+ # Over the 1 MiB the server returns per stream, which the default's
+ # base64 slices through run_command could never carry.
+ assert len(data) > _SERVER_STREAM_CAP
+ assert written == len(data)
+ assert dest.getvalue() == data
+ client.sandboxes.download_file.assert_called_once_with(
+ "box",
+ path="/w/out.bin",
+ request_options={"timeout_in_seconds": 120, "chunk_size": 65536},
+ )
+ client.sandboxes.exec_in_sandbox.assert_called_once()
+
+ @pytest.mark.parametrize(
+ ("relative", "quoted"),
+ [("report.csv", "report.csv"), ("report.csv ", "'report.csv '")],
+ ids=["plain", "trailing-space"],
+ )
+ def
test_a_relative_path_is_resolved_in_the_guest_before_it_is_downloaded(self,
relative, quoted):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.side_effect = [
+ _export_check(4),
+ _exec_result(stdout=f"/workspace/{relative}\n"),
+ ]
+ client.sandboxes.download_file.return_value = iter([b"abcd"])
+ dest = io.BytesIO()
+
+ written = backend.export_file("box", relative, dest, max_bytes=100)
+
+ # The download API answers a relative path with a 400, so only the
resolved
+ # absolute path may reach it, and whitespace that is part of the name
stays.
+ assert written == 4
+ assert dest.getvalue() == b"abcd"
+ resolve =
client.sandboxes.exec_in_sandbox.call_args_list[1].kwargs["command"][4]
+ assert resolve == f"realpath -e -- {quoted}"
+ assert client.sandboxes.download_file.call_args.kwargs["path"] ==
f"/workspace/{relative}"
+
+ @pytest.mark.parametrize(
+ ("exit_code", "stderr", "error", "message"),
+ [
+ (IsloSandboxBackend._MISSING_PATH_STATUS, "", SandboxError, "does
not exist"),
+ (IsloSandboxBackend._IS_DIRECTORY_STATUS, "", SandboxError, "is a
directory"),
+ (IsloSandboxBackend._NOT_REGULAR_FILE_STATUS, "", SandboxError,
"is not a regular file"),
+ (
+ IsloSandboxBackend._TOO_LARGE_STATUS,
+ "\nairflow-export-size:5000\n",
+ SandboxFileTooLargeError,
+ "is 5000 bytes",
+ ),
+ ],
+ ids=["missing", "directory", "not-regular", "too-large"],
+ )
+ def test_the_guest_check_refuses_before_anything_is_downloaded(self,
exit_code, stderr, error, message):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value =
_exec_result(exit_code=exit_code, stderr=stderr)
+
+ with pytest.raises(error, match=message) as raised:
+ backend.export_file("box", "/w/a", io.BytesIO(), max_bytes=100)
+
+ assert not isinstance(raised.value, SandboxTerminalError)
+ client.sandboxes.download_file.assert_not_called()
+
+ def test_a_stream_longer_than_the_budget_is_stopped_and_closed(self):
+ backend, client = _backend_with_client()
+ closed: list[bool] = []
+
+ def chunks():
+ try:
+ yield b"x" * 6
+ yield b"x" * 6
+ raise AssertionError("the backend must stop once the budget is
passed")
+ finally:
+ closed.append(True)
+
+ # Within budget when the guest checked it, then the file grew.
+ client.sandboxes.get_exec_result.return_value = _export_check(10)
+ client.sandboxes.download_file.return_value = chunks()
+ dest = io.BytesIO()
+
+ with pytest.raises(SandboxFileTooLargeError) as error:
+ backend.export_file("box", "/w/a", dest, max_bytes=10)
+
+ assert error.value.size_bytes == 12
+ assert dest.getvalue() == b"x" * 6
+ assert closed == [True]
+
+ def test_a_file_that_changed_size_while_exported_is_an_error(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _export_check(10)
+ client.sandboxes.download_file.return_value = iter([b"abcd"])
+
+ with pytest.raises(SandboxError, match="changed while it was
exported"):
+ backend.export_file("box", "/w/a", io.BytesIO(), max_bytes=100)
+
+ @pytest.mark.parametrize(
+ ("error", "expected", "message"),
+ [
+ (NotFoundError({}), SandboxError, "does not exist"),
+ (ApiError(status_code=503), SandboxTerminalError, "export a
sandbox file"),
+ ],
+ ids=["404", "503"],
+ )
+ def test_a_failed_download_is_classified_like_any_file_transfer(self,
error, expected, message):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _export_check(10)
+ client.sandboxes.download_file.side_effect = error
+
+ with pytest.raises(SandboxError, match=message) as raised:
+ backend.export_file("box", "/w/a", io.BytesIO(), max_bytes=100)
+
+ assert type(raised.value) is expected
+
+ def test_a_failing_destination_is_not_reported_as_a_sandbox_error(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _export_check(4)
+ client.sandboxes.download_file.return_value = iter([b"abcd"])
+ dest = mock.create_autospec(io.BytesIO, instance=True)
+ dest.write.side_effect = OSError("object store unavailable")
+
+ with pytest.raises(OSError, match="object store unavailable"):
+ backend.export_file("box", "/w/a", dest, max_bytes=100)
+
+ client.sandboxes.get_sandbox.assert_not_called()
+
+ @mock.patch(f"{_MODULE}._export_deadline", autospec=True, return_value=0.0)
+ def test_an_export_past_its_deadline_is_ended(self, _export_deadline):
+ backend, client = _backend_with_client()
+ client.sandboxes.get_exec_result.return_value = _export_check(4)
+ client.sandboxes.download_file.return_value = iter([b"abcd"])
+ dest = io.BytesIO()
+
+ with pytest.raises(SandboxError, match="took longer than"):
+ backend.export_file("box", "/w/a", dest, max_bytes=100)
+
+ assert dest.getvalue() == b""
+
+ def test_rejects_an_invalid_budget(self):
+ backend, client = _backend_with_client()
+
+ with pytest.raises(ValueError, match="max_bytes"):
+ backend.export_file("box", "/w/a", io.BytesIO(), max_bytes=0)
+
+ client.sandboxes.exec_in_sandbox.assert_not_called()
+
+
+class TestDestroy:
+ def test_is_idempotent_when_the_sandbox_is_already_gone(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.delete_sandbox.side_effect = NotFoundError({})
+
+ backend.destroy("box")
+
+ def test_delete_failure_is_terminal(self):
+ backend, client = _backend_with_client()
+ client.sandboxes.delete_sandbox.side_effect = ApiError(status_code=503)
+
+ with pytest.raises(SandboxTerminalError, match="delete a sandbox"):
+ backend.destroy("box")
+
+ def test_delete_keeps_the_sdk_retries(self):
+ backend, client = _backend_with_client()
+
+ backend.destroy("box")
+
+ # Deletion is idempotent and is the only call whose failure strands a
+ # microVM, so it must not be the one call that never retries.
+ options =
client.sandboxes.delete_sandbox.call_args.kwargs["request_options"]
+ assert options["max_retries"] > 0
+
+
+def _run_wrapper(command: str, max_output_bytes: int, *, timeout: float =
30.0, env: dict | None = None):
+ """Run the real wrapper through a local ``sh``, exactly as the backend
invokes it."""
+ return subprocess.run(
+ [
+ "sh",
+ "-c",
+ _COMMAND_WRAPPER,
+ "airflow-sandbox",
+ command,
+ str(max_output_bytes + 1),
+ ],
+ capture_output=True,
+ text=True,
+ timeout=timeout,
+ check=False,
+ env=env,
+ )
+
+
+class TestCommandWrapper:
+ """
+ Execute ``_COMMAND_WRAPPER`` for real, rather than asserting on its text.
+
+ Every other test in this module mocks ``exec_in_sandbox``, so without these
+ the wrapper only ever runs in the system test, which needs a live API key
and
+ so does not run in ordinary CI. A local ``sh`` needs no Islo access at all.
+ """
+
+ def test_reports_stdout_stderr_and_the_exit_status_separately(self):
+ result = _run_wrapper("echo out; echo err >&2; exit 3", 1024)
+
+ assert result.stdout == "out\n"
+ assert result.stderr == "err\n"
+ assert result.returncode == 3
+
+ def test_the_process_environment_reaches_the_command_unchanged(self):
+ # A login shell would run /etc/profile after the spec's variables were
set,
+ # and Debian's and macOS's both rewrite PATH.
+ env = {**os.environ, "PATH": "/spec/wins:" + os.environ["PATH"]}
+ result = _run_wrapper('echo "${PATH%%:*}"', 1024, env=env)
+
+ assert result.stdout == "/spec/wins\n"
+
+ def test_a_backgrounded_process_does_not_hold_the_command_open(self):
+ # The command's foreground part finishes at once. Waiting for the
capture
+ # to reach end-of-input would block until the backgrounded child exits,
+ # and past the deadline the backend destroys the sandbox -- so the
agent
+ # would lose its files over a command that already finished.
+ start = time.monotonic()
+ result = _run_wrapper("sleep 20 & echo $! >&2; echo started", 1024)
+ elapsed = time.monotonic() - start
+ try:
+ assert result.stdout == "started\n"
+ assert result.returncode == 0
+ assert elapsed < 5.0
+ finally:
+ with suppress(ProcessLookupError):
+ os.kill(int(result.stderr.strip()), signal.SIGTERM)
+
+ def test_a_long_lived_daemon_does_not_hold_the_command_open(self):
+ start = time.monotonic()
+ result = _run_wrapper("nohup sleep 300 >/dev/null 2>&1 & echo $! >&2;
echo server-started", 1024)
+ elapsed = time.monotonic() - start
+ try:
+ assert result.stdout == "server-started\n"
+ assert elapsed < 5.0
+ finally:
+ with suppress(ProcessLookupError):
+ os.kill(int(result.stderr.strip()), signal.SIGTERM)
+
+ def test_does_not_change_the_permissions_of_what_the_agent_creates(self,
tmp_path):
+ script = "touch a_file && mkdir a_dir && stat -c '%a %n' a_dir a_file
2>/dev/null || stat -f '%Lp %N' a_dir a_file"
+ (tmp_path / "wrapped").mkdir()
+ (tmp_path / "bare").mkdir()
+
+ wrapped = _run_wrapper(f"cd {tmp_path / 'wrapped'} && {script}", 4096)
+ bare = subprocess.run(
+ ["sh", "-c", f"cd {tmp_path / 'bare'} && {script}"],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+
+ # A umask left in force for the agent's command would tighten these;
the
+ # baseline is the host's own umask, so the test holds whatever that is.
+ assert wrapped.returncode == 0
+ assert wrapped.stdout == bare.stdout
+
+ def
test_the_scratch_directory_is_private_while_in_use_and_gone_after(self,
tmp_path):
+ scratch_root = tmp_path
+ pattern = "airflow-sandbox-[0-9]*"
+
+ process = subprocess.Popen(
+ ["sh", "-c", _COMMAND_WRAPPER, "airflow-sandbox", "sleep 2",
"1024"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ env={**os.environ, "TMPDIR": str(tmp_path)},
+ )
+ try:
+ deadline = time.monotonic() + 5.0
+ scratch: list[Path] = []
+ while not scratch and time.monotonic() < deadline:
+ scratch = list(scratch_root.glob(pattern))
+ time.sleep(0.05)
+ assert scratch, "the wrapper never created its scratch directory"
+ # The capture files sit here, so the agent's command must not be
able
+ # to hand them to another user in a shared image.
+ assert scratch[0].stat().st_mode & 0o777 == 0o700
+ finally:
+ process.communicate(timeout=30)
+
+ assert not list(scratch_root.glob(pattern))
+
+ def test_keeps_the_tail_and_drops_a_short_partial_leading_line(self):
+ cap = 100
+ result = _run_wrapper("i=1; while [ $i -le 1000 ]; do echo line$i;
i=$((i+1)); done", cap)
+
+ # The wrapper is asked for cap+1 bytes, so the backend can tell the
+ # stream was over budget. These records are a few bytes each, so the
+ # cut one is dropped and every record left is whole.
+ assert len(result.stdout.encode()) == cap + 1
+ payload, truncated = _bound_result_stream(result.stdout, cap)
+ assert truncated
+ assert payload.endswith("line1000\n")
+ assert all(line.startswith("line") for line in payload.splitlines())
+
+ def test_a_terminated_command_exits_143_and_emits_nothing(self):
+ process = subprocess.Popen(
+ ["sh", "-c", _COMMAND_WRAPPER, "airflow-sandbox", "sleep 30",
"1024"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ start_new_session=True,
+ )
+ pgid = os.getpgid(process.pid)
+ # The traps are in place once the agent's command itself is running,
+ # which on a loaded host can take longer than any fixed pause.
+ deadline = time.monotonic() + 10.0
+ while (
+ subprocess.run(
+ ["pgrep", "-g", str(pgid), "-x", "sleep"],
capture_output=True, check=False
+ ).returncode
+ and time.monotonic() < deadline
+ ):
+ time.sleep(0.05)
+ # What stopping the microVM looks like from inside it.
+ os.killpg(pgid, signal.SIGTERM)
+ stdout, stderr = process.communicate(timeout=30)
+
+ # Without the trap the wrapper dies of the signal itself (a negative
+ # return code); a trap that cleans up and then falls through reaches
+ # ``tail`` with its scratch files already deleted and reports that on
+ # stderr. bash may still announce the child's death ("Terminated: 15"),
+ # which is the shell's notice, not wrapper output.
+ assert process.returncode == 143
+ assert stdout == ""
+ assert "No such file" not in stderr
diff --git a/uv.lock b/uv.lock
index 5adc4dda962..3a470996ebd 100644
--- a/uv.lock
+++ b/uv.lock
@@ -4595,6 +4595,9 @@ git = [
google = [
{ name = "pydantic-ai-slim", extra = ["google"] },
]
+islo = [
+ { name = "islo" },
+]
langchain = [
{ name = "langchain" },
]
@@ -4651,6 +4654,7 @@ dev = [
{ name = "apache-airflow-providers-standard" },
{ name = "apache-airflow-task-sdk" },
{ name = "boat-sdk" },
+ { name = "islo" },
{ name = "langchain" },
{ name = "llama-index-core" },
{ name = "llama-index-embeddings-openai" },
@@ -4680,6 +4684,7 @@ requires-dist = [
{ name = "dataclasses-json", marker = "extra == 'llamaindex'", specifier =
">=0.6.7" },
{ name = "fastavro", marker = "python_full_version >= '3.14' and extra ==
'avro'", specifier = ">=1.12.1" },
{ name = "fastavro", marker = "python_full_version < '3.14' and extra ==
'avro'", specifier = ">=1.10.0" },
+ { name = "islo", marker = "extra == 'islo'", specifier = ">=0.3.19" },
{ name = "langchain", marker = "extra == 'langchain'", specifier =
">=1.0.0" },
{ name = "llama-index-core", marker = "extra == 'llamaindex'", specifier =
">=0.14.5" },
{ name = "llama-index-embeddings-openai", marker = "extra ==
'llamaindex'", specifier = ">=0.6.0" },
@@ -4704,7 +4709,7 @@ requires-dist = [
{ name = "structlog", specifier = ">=24.2.0" },
{ name = "typesafe-sdk", marker = "extra == 'typesafe'", specifier =
">=0.6.0" },
]
-provides-extras = ["anthropic", "bedrock", "google", "openai", "typesafe",
"mcp", "modal", "opensandbox", "boat", "code-mode", "shields", "skills",
"avro", "parquet", "sql", "common-sql", "langchain", "llamaindex", "pdf",
"docx", "git"]
+provides-extras = ["anthropic", "bedrock", "google", "openai", "typesafe",
"mcp", "modal", "opensandbox", "boat", "code-mode", "shields", "skills",
"islo", "avro", "parquet", "sql", "common-sql", "langchain", "llamaindex",
"pdf", "docx", "git"]
[package.metadata.requires-dev]
dev = [
@@ -4719,6 +4724,7 @@ dev = [
{ name = "apache-airflow-providers-standard", editable =
"providers/standard" },
{ name = "apache-airflow-task-sdk", editable = "task-sdk" },
{ name = "boat-sdk", specifier = ">=1.9.3" },
+ { name = "islo", specifier = ">=0.3.19" },
{ name = "langchain", specifier = ">=1.0.0" },
{ name = "llama-index-core", specifier = ">=0.14.5" },
{ name = "llama-index-embeddings-openai", specifier = ">=0.6.0" },
@@ -15314,6 +15320,20 @@ wheels = [
{ url =
"https://files.pythonhosted.org/packages/d9/33/1f075bf72b0b747cb3288d011319aaf64083cf2efef8354174e3ed4540e2/ipython_pygments_lexers-1.1.1-py3-none-any.whl";,
hash =
"sha256:a9462224a505ade19a605f71f8fa63c2048833ce50abc86768a0d81d876dc81c", size
= 8074, upload-time = "2025-01-17T11:24:33.271Z" },
]
+[[package]]
+name = "islo"
+version = "0.3.19"
+source = { registry = "https://pypi.org/simple"; }
+dependencies = [
+ { name = "httpx" },
+ { name = "pydantic" },
+ { name = "typing-extensions" },
+]
+sdist = { url =
"https://files.pythonhosted.org/packages/45/87/7678bf4559e775c58e80df263faff46b9f1cef626540bc852dbb8157ff8f/islo-0.3.19.tar.gz";,
hash =
"sha256:b3f9fa7b996551eb6e77b6c0202f3af0c1d21935eefd4933b0232b271a464612", size
= 155588, upload-time = "2026-08-27T12:48:20.149Z" }
+wheels = [
+ { url =
"https://files.pythonhosted.org/packages/c0/b5/a8ecdea027ccc20f5ab44042c8629cd22588ca7c33cc199c5f103f92ead4/islo-0.3.19-py3-none-any.whl";,
hash =
"sha256:ab7e9bab5678e486e0bdfa8ed25e40fe2eef95c5748c3d666b5c6b1226354126", size
= 372082, upload-time = "2026-08-27T12:48:18.582Z" },
+]
+
[[package]]
name = "isodate"
version = "0.7.2"
