From 53cd3ac7c8443b7f953979950dd7bd1cfb3bec16 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 7 Oct 2026 20:47:03 -0700
Subject: [PATCH 1/6] feat(images): read image documents with no tarball, and
 pull them by name

A DockerImageArtifact's tar_gz_object_url is now optional. One with no tar.gz
names a registry reference in image_name, which a sandbox pulls instead.

- load_problem(): an image loads from its tar.gz, or, with none, by pulling an
  image_name that spells out its registry (names_registry). Anything else is
  refused up front, by the VM loader and by preflight on every provider.
- VmSandbox.load_docker_images loads tarballs exactly as before (same script,
  same check) and pulls the rest through pull_images, the docker login + pull
  that create_container already ran, now shared.
- The E2B and Sail overrides keep widening a restrictive policy for signed
  tarball downloads only; a pull goes through the policy as it is.
- put_tar refuses an empty tar_gz_object_url; load() refuses an image with none.
- Preflight's file:// check no longer misreads a missing tar.gz, and an agent
  the store can't read is left to the default-agent report instead of raising.

Nothing writes a tarball-less document yet: readers ship first.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../artifact/artifacts/docker_image.py        | 23 ++++-
 src/agent_env/bundle/preflight.py             | 38 +++++---
 .../sandbox_providers/e2b/sandbox.py          |  8 +-
 .../sandbox_providers/sail_vm/sandbox.py      |  6 +-
 .../providers/sandbox_providers/sandbox.py    | 43 ++++++++-
 .../sandbox_providers/sandbox_provider.py     | 24 +----
 src/agent_env/store/image_store/__init__.py   |  2 +
 .../image_store/oci_registry_credentials.py   | 22 +++--
 tst/unit/artifact/test_docker_image_pulled.py | 52 +++++++++++
 tst/unit/bundle/preflight_test.py             | 39 ++++++++
 .../sandbox_providers/e2b_sandbox_test.py     | 34 ++++++-
 .../sandbox_providers/local_sandbox_test.py   |  7 +-
 .../sandbox_providers/sail_vm_sandbox_test.py | 32 +++++--
 .../sandbox_provider_test.py                  |  4 +-
 .../sandbox_providers/vm_sandbox_test.py      | 92 +++++++++++++++++--
 .../store/oci_registry_image_store_test.py    | 18 +++-
 16 files changed, 362 insertions(+), 82 deletions(-)
 create mode 100644 tst/unit/artifact/test_docker_image_pulled.py

diff --git a/src/agent_env/artifact/artifacts/docker_image.py b/src/agent_env/artifact/artifacts/docker_image.py
index 8a77cec9..6b13687d 100644
--- a/src/agent_env/artifact/artifacts/docker_image.py
+++ b/src/agent_env/artifact/artifacts/docker_image.py
@@ -23,7 +23,7 @@
 
 from agent_env.artifact.artifact import Artifact, _write_twin
 from agent_env.store.ids import fs_safe, image_repository, is_local_id
-from agent_env.store.image_store.oci_registry_credentials import is_loopback_host, registry_host_from_ref
+from agent_env.store.image_store.oci_registry_credentials import is_loopback_host, names_registry, registry_host_from_ref
 from agent_env.utils.deprecation import OMITTED, renamed_keyword
 
 logger = logging.getLogger(__name__)
@@ -63,7 +63,8 @@ def _git_clone_commands(owner: str, repo: str, ref: str | None, token: str | Non
     ]
 
 class DockerImageArtifact(Artifact):
-    """A Docker image artifact stored as tar.gz in the object store."""
+    """A Docker image artifact: a tar.gz of the image in the object store, or, without one, a registry reference a
+    sandbox pulls."""
 
     model_config = ConfigDict(populate_by_name=True)
 
@@ -71,8 +72,10 @@ class DockerImageArtifact(Artifact):
 
     type: Literal["docker_image"] = "docker_image"
     description: str = Field(description="Description of the Docker image")
-    image_name: str = Field(description="Docker image name/tag")
-    tar_gz_object_url: str = Field(alias="tar_gz_s3_url", description="Object-store locator of the tar.gz file")
+    image_name: str = Field(description="Docker image name/tag; with no tar.gz, the registry reference pulled")
+    tar_gz_object_url: str | None = Field(
+        default=None, alias="tar_gz_s3_url", description="Object-store locator of the tar.gz file, if it has one"
+    )
     build_context_object_url: str | None = Field(default=None, alias="build_context_s3_url", description="Object-store locator of the build context tar.gz")
 
     # No return annotation: pydantic builds the serialization schema from one, and a dict drops the fields.
@@ -188,6 +191,8 @@ def put_tar(
         )
         if tar_gz_object_url is None:
             raise TypeError(f"{owner}() missing required keyword argument: 'tar_gz_object_url'")
+        if not tar_gz_object_url:
+            raise ValueError(f"{owner}(): tar_gz_object_url is empty")
         for url in (tar_gz_object_url, build_context_object_url):
             if url:
                 get_config().check_object_url(id, url)
@@ -205,8 +210,18 @@ def put_tar(
 
     def load(self) -> bytes:
         from agent_env.artifact.store import get_artifact_store
+        if not self.tar_gz_object_url:
+            raise ValueError(f"{self.id!r} v{self.version} has no tar.gz; its image is pulled from {self.image_name}")
         return get_artifact_store().get_object(self.tar_gz_object_url)
 
+    def load_problem(self) -> str | None:
+        """Why no sandbox can get this image, or None when one can: a tar.gz is loaded, and with none, ``image_name``
+        is pulled, so it must name its registry."""
+        if self.tar_gz_object_url or names_registry(self.image_name):
+            return None
+        return (f"{self.id!r} v{self.version} has no tar.gz, and its image name {self.image_name!r} doesn't name a "
+                "registry to pull it from")
+
     @classmethod
     async def put_from_github(
         cls,
diff --git a/src/agent_env/bundle/preflight.py b/src/agent_env/bundle/preflight.py
index 4a1e5547..e22212bd 100644
--- a/src/agent_env/bundle/preflight.py
+++ b/src/agent_env/bundle/preflight.py
@@ -95,6 +95,7 @@ class _Image:
 
     what: str
     local_only: str | None  # why only this machine has it, or None
+    unloadable: str | None = None  # why no sandbox can get it, or None
 
 
 @dataclass(frozen=True)
@@ -179,6 +180,7 @@ def _env(self, where: str, step: DeployEnvTaskStep) -> None:
         else:  # a plugin's provider, or one deploy_env's own preflight refuses for this env
             return
         env_id, images = step.env_id, deployment.images
+        self._loadable(where, images)
         if _local_link(provider):
             self.infra |= kinds
             self.docker_users.append(where)
@@ -256,8 +258,7 @@ def _planned_image(self, env_id: str, ref: Reference) -> _Image | None:
         if ref.id in self.written:
             return None  # another of the bundle's writes, which materialize refuses or writes first
         try:
-            return _Image(what, _local_only(DockerImageArtifact.get(ref.id, self._planned_version(ref.kind, ref.id,
-                                                                                                    ref.version))))
+            return _image(what, DockerImageArtifact.get(ref.id, self._planned_version(ref.kind, ref.id, ref.version)))
         except NotFoundError:
             return None  # the plan reports a store image that isn't there
 
@@ -288,10 +289,11 @@ def _agent_deploy(self, where: str, provider: SandboxProvider, agent_id: str | N
         if agent_id is None:
             self.default_agent_users.append((where, unnamed))
             agent_id = get_config().get_default_a2a_agent_id()
+        if (image := self._agent_image(agent_id, version)) is None:
+            return
+        self._loadable(where, [image])
         if remote := _remote_links(provider):
-            image = self._agent_image(agent_id, version)
-            if image is not None:
-                self._reachable(where, remote, [image])
+            self._reachable(where, remote, [image])
 
     def _sandbox(self, where: str, step: DeploySandboxTaskStep) -> None:
         provider = _provider(self.sandbox or step.sandbox_type, get_sandbox_provider)
@@ -308,8 +310,8 @@ def _sandbox(self, where: str, step: DeploySandboxTaskStep) -> None:
     # What a deploy runs
 
     def _agent_image(self, agent_id: str, version: int | None) -> _Image | None:
-        """The image the agent ``agent_id`` runs, or None when the store doesn't hold the agent, which the plan or
-        ``_default_agent`` reports."""
+        """The image the agent ``agent_id`` runs, or None when the store doesn't hold the agent or can't read it, which
+        the plan or ``_default_agent`` reports."""
         what = f"agent {agent_id!r}'s image"
         if (agent := self.agents.get(agent_id)) is not None:
             image_id, image_version = parse_toml_ref(EntityKind.ARTIFACT, agent.config.get("image"))
@@ -319,12 +321,17 @@ def _agent_image(self, agent_id: str, version: int | None) -> _Image | None:
             if image_id in self.written:
                 return None  # another of the bundle's writes, which materialize refuses or writes first
             planned = self._planned_version(EntityKind.ARTIFACT, image_id, image_version)
-            return _Image(what, _local_only(DockerImageArtifact.get(image_id, planned)))
+            return _image(what, DockerImageArtifact.get(image_id, planned))
         try:
-            return _Image(what, _local_only(A2AAgent.get(agent_id, version).docker_image_artifact))
-        except NotFoundError:
+            return _image(what, A2AAgent.get(agent_id, version).docker_image_artifact)
+        except (NotFoundError, ValueError, KeyError, TypeError):
             return None
 
+    def _loadable(self, where: str, images: list[_Image]) -> None:
+        for image in images:
+            if image.unloadable:
+                self._problem(where, f"deploys {image.what}, which no sandbox can get: {image.unloadable}")
+
     def _reachable(self, where: str, remote: list[SandboxProvider], images: list[_Image]) -> None:
         for image in images:
             if image.local_only:
@@ -343,7 +350,7 @@ def _remote_infra(self, where: str, provider: SandboxProvider, kinds: set[str])
                 continue
             images = ([env.db_docker_image_artifact, env.db_web_docker_image_artifact, env.db_mcp_docker_image_artifact]
                       if isinstance(env, ServiceDBEnv) else [env.docker_image_artifact])
-            self._reachable(where, [provider], [_Image(f"the {kind} env {env_id!r}'s image {image.id!r}", _local_only(image))
+            self._reachable(where, [provider], [_image(f"the {kind} env {env_id!r}'s image {image.id!r}", image)
                                                 for image in images])
 
     def _default_agent(self) -> None:
@@ -373,8 +380,7 @@ def _stored_images(env: Env) -> tuple[list[_Image], bool]:
     """The images a store env runs through a gateway, and whether it has websites."""
     topology = _gateway_topology(env)
     images = [*topology.mcp_server_images, *(topology.website_images or [])]
-    return [_Image(f"env {env.id!r}'s image {image.id!r}", _local_only(image)) for image in images], bool(
-        topology.website_configs)
+    return [_image(f"env {env.id!r}'s image {image.id!r}", image) for image in images], bool(topology.website_configs)
 
 
 def _state_type(step: DeployEnvTaskStep) -> str | None:
@@ -422,12 +428,16 @@ def _name(provider: SandboxProvider) -> str:
     return cls.__name__
 
 
+def _image(what: str, image: DockerImageArtifact) -> _Image:
+    return _Image(what, _local_only(image), image.load_problem())
+
+
 def _local_only(image: DockerImageArtifact | str) -> str | None:
     """Why only this machine has ``image``: its reference names a registry on this machine, or it's saved in this
     machine's object store. None when neither."""
     ref = image if isinstance(image, str) else image.image_name
     if is_loopback_host(registry_host_from_ref(ref)):
         return f"{ref} is in a registry on this machine"
-    if not isinstance(image, str) and urlparse(image.tar_gz_object_url).scheme == "file":
+    if not isinstance(image, str) and image.tar_gz_object_url and urlparse(image.tar_gz_object_url).scheme == "file":
         return f"{image.id!r} is saved in this machine's object store"
     return None
diff --git a/src/agent_env/providers/sandbox_providers/e2b/sandbox.py b/src/agent_env/providers/sandbox_providers/e2b/sandbox.py
index 896f7593..14f7342a 100644
--- a/src/agent_env/providers/sandbox_providers/e2b/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/e2b/sandbox.py
@@ -291,8 +291,8 @@ async def apply_network_policy(self, policy: NetworkPolicy) -> None:
         await self._sandbox.update_network(network)
         self.network_policy = policy
 
-    async def load_docker_images(self, artifacts: list) -> None:
-        """Load images after adding their signed-download hosts to the policy.
+    async def _load_tarballs(self, artifacts: list) -> None:
+        """Load image tarballs after adding their signed-download hosts to the policy.
 
         With a restrictive policy, the object store that supplies image tarballs
         is infrastructure rather than a workload destination.  E2B replaces an
@@ -301,8 +301,6 @@ async def load_docker_images(self, artifacts: list) -> None:
         :meth:`NetworkPolicy.with_hosts`.
         """
         policy = self.network_policy
-        if not artifacts:
-            return
         if policy is None:
             raise RuntimeError(
                 f"Cannot load Docker images in reconnected E2B sandbox {self.sandbox_id}: "
@@ -310,7 +308,7 @@ async def load_docker_images(self, artifacts: list) -> None:
                 "be added safely"
             )
         if not policy.restricts_egress:
-            await super().load_docker_images(artifacts)
+            await super()._load_tarballs(artifacts)
             return
 
         signed_urls = await self._signed_image_urls(artifacts)
diff --git a/src/agent_env/providers/sandbox_providers/sail_vm/sandbox.py b/src/agent_env/providers/sandbox_providers/sail_vm/sandbox.py
index d709981e..8644a777 100644
--- a/src/agent_env/providers/sandbox_providers/sail_vm/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sail_vm/sandbox.py
@@ -464,10 +464,8 @@ async def _allow_download_hosts(self, urls: list[str | None], purpose: str) -> N
                 )
             await self._apply_policy(policy.with_hosts(missing))
 
-    async def load_docker_images(self, artifacts: list) -> None:
-        """Load images, first adding their signed-download hosts to a restrictive policy."""
-        if not artifacts:
-            return
+    async def _load_tarballs(self, artifacts: list) -> None:
+        """Load image tarballs, first adding their signed-download hosts to a restrictive policy."""
         self._known_policy("load Docker images")
         signed_urls = await self._signed_image_urls(artifacts)
         await self._allow_download_hosts(signed_urls, "load Docker images")
diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index e5d2bc12..f55a22cb 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox.py
@@ -284,11 +284,39 @@ async def wait_for_vm(self) -> None:
         )
 
     async def load_docker_images(self, artifacts: list) -> None:
-        """Load Docker images from DockerImageArtifacts into the sandbox in parallel."""
+        """Put each DockerImageArtifact's image on the VM: those with a tar.gz are loaded from it, in parallel, and the
+        rest are pulled by image name. An image no sandbox can get is refused before anything is loaded."""
         if not artifacts:
             return
+        if problems := [problem for artifact in artifacts if (problem := artifact.load_problem())]:
+            raise RuntimeError(f"Can't load images: {'; '.join(problems)}")
+        if tarballs := [artifact for artifact in artifacts if artifact.tar_gz_object_url]:
+            await self._load_tarballs(tarballs)
+        if pulled := [artifact.image_name for artifact in artifacts if not artifact.tar_gz_object_url]:
+            await self.pull_images(pulled)
+
+    async def _load_tarballs(self, artifacts: list) -> None:
+        """Load the images of ``artifacts``, each with a tar.gz, in parallel."""
         await self._load_docker_images(artifacts, await self._signed_image_urls(artifacts))
 
+    async def pull_images(self, image_names: list[str]) -> None:
+        """``docker pull`` each of ``image_names``, concurrently, after logging in to each registry the image store
+        holds credentials for. A network policy restricting egress isn't widened for them: it must allow the
+        registries itself."""
+        image_names = list(dict.fromkeys(image_names))
+        config = get_config()
+        logins: dict[str, Any] = {}
+        for image_name in image_names:
+            auth = await asyncio.to_thread(config.get_image_store().auth, image_name)
+            if auth is not None:
+                logins.setdefault(auth.registry, auth)
+        for auth in logins.values():
+            await self.exec_script(
+                f"echo {shlex.quote(auth.password)} | docker login "
+                f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}"
+            )
+        await asyncio.gather(*(_pull(self, image_name) for image_name in image_names))
+
     @staticmethod
     async def _signed_image_urls(artifacts: list) -> list[str | None]:
         """Each image tarball's signed URL, or None where the store cannot sign one. Signed
@@ -455,6 +483,19 @@ def port_bindings(host_ips: Iterable[str], host_port: int, container_port: int)
     return [f"{ip}:{host_port}:{container_port}" for ip in host_ips] or [f"{host_port}:{container_port}"]
 
 
+async def _pull(sandbox: VmSandbox, image_name: str) -> None:
+    """``docker pull image_name``. An image built for linux/amd64 only has nothing for an arm64 host (an
+    Apple Silicon Mac running the local provider), so that pull falls back to the amd64 image, which the
+    host's Docker runs emulated."""
+    try:
+        await sandbox.exec_script(f"docker pull {shlex.quote(image_name)}")
+    except RuntimeError as e:
+        if "no matching manifest" not in str(e):
+            raise
+        logger.warning("%s has no image for this host's platform; pulling linux/amd64, which runs emulated", image_name)
+        await sandbox.exec_script(f"docker pull --platform linux/amd64 {shlex.quote(image_name)}")
+
+
 async def stage_files_into_container(sandbox: Sandbox, file_artifacts: dict[str, Any], destination: str) -> dict[str, str]:
     """Write each object-store file into the sandbox's container under ``destination``; returns ``{name: path}``.
     A VM-backed sandbox stages through its host into ``container_name``; a container sandbox is written directly."""
diff --git a/src/agent_env/providers/sandbox_providers/sandbox_provider.py b/src/agent_env/providers/sandbox_providers/sandbox_provider.py
index 68009580..b0a8fdd0 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox_provider.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox_provider.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import logging
 import os
 import shlex
@@ -36,19 +35,6 @@ class SandboxProviderTypeError(ConfigError):
     """A config-registered provider produced a Sandbox whose ``.type`` != its ``[sandbox.providers.<name>]`` key."""
 
 
-async def _pull(sandbox: VmSandbox, image_name: str) -> None:
-    """``docker pull image_name``. An image built for linux/amd64 only has nothing for an arm64 host (an
-    Apple Silicon Mac running the local provider), so that pull falls back to the amd64 image, which the
-    host's Docker runs emulated."""
-    try:
-        await sandbox.exec_script(f"docker pull {shlex.quote(image_name)}")
-    except RuntimeError as e:
-        if "no matching manifest" not in str(e):
-            raise
-        logger.warning("%s has no image for this host's platform; pulling linux/amd64, which runs emulated", image_name)
-        await sandbox.exec_script(f"docker pull --platform linux/amd64 {shlex.quote(image_name)}")
-
-
 class SandboxProvider(ABC):
     """Compute backend that provisions sandboxes. Selected via [sandbox] default / build_sandbox_provider(); swap via set_sandbox_provider()."""
 
@@ -111,15 +97,7 @@ async def create_container(
             network_policy=network_policy,
         )
         try:
-            from agent_env.config import get_config
-
-            auth = await asyncio.to_thread(get_config().get_image_store().auth, image_name)
-            if auth is not None:
-                await sandbox.exec_script(
-                    f"echo {shlex.quote(auth.password)} | docker login "
-                    f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}"
-                )
-            await _pull(sandbox, image_name)
+            await sandbox.pull_images([image_name])
             await self._start_container(sandbox, image_name=image_name, port=port, env=env)
             sandbox.mode = SANDBOX_MODE_CONTAINER
             return sandbox
diff --git a/src/agent_env/store/image_store/__init__.py b/src/agent_env/store/image_store/__init__.py
index eb99d041..9bb7e406 100644
--- a/src/agent_env/store/image_store/__init__.py
+++ b/src/agent_env/store/image_store/__init__.py
@@ -16,6 +16,7 @@
     OciRegistryCredentials,
     RegistryAuth,
     SecretStoreCredentials,
+    names_registry,
     normalize_registry_host,
     registry_host_from_ref,
 )
@@ -25,6 +26,7 @@
     "OciRegistryImageStore",
     "RegistryAuth",
     "OciRegistryCredentials",
+    "names_registry",
     "normalize_registry_host",
     "registry_host_from_ref",
     "EcrCredentials",
diff --git a/src/agent_env/store/image_store/oci_registry_credentials.py b/src/agent_env/store/image_store/oci_registry_credentials.py
index be46a2a3..6d3e4d97 100644
--- a/src/agent_env/store/image_store/oci_registry_credentials.py
+++ b/src/agent_env/store/image_store/oci_registry_credentials.py
@@ -60,21 +60,29 @@ def normalize_registry_host(host: str) -> str:
     return "docker.io" if normalized in _DOCKER_HUB_HOSTS else normalized
 
 
-def registry_host_from_ref(ref: str) -> str | None:
-    """Return the normalized registry authority from a Docker/OCI image reference.
+def names_registry(ref: str) -> bool:
+    """Whether the image reference ``ref`` spells out its registry, as ``ghcr.io/x/y`` and ``localhost:5000/x`` do.
 
     Docker treats a first path component as a registry only when it is ``localhost``
-    or contains a dot or port. Unqualified references therefore resolve to Docker Hub.
+    or contains a dot or port, and resolves any other reference on Docker Hub.
     Schemed values are not valid image references and deliberately do not match.
     """
     if not ref or "://" in ref:
-        return None
+        return False
     first, separator, _ = ref.partition("/")
+    return bool(separator) and (first.lower() == "localhost" or "." in first or ":" in first)
+
+
+def registry_host_from_ref(ref: str) -> str | None:
+    """Return the normalized registry authority from a Docker/OCI image reference: the one it names
+    (``names_registry``), else Docker Hub's. None for a value that isn't a reference.
+    """
+    if not ref or "://" in ref:
+        return None
+    first = ref.partition("/")[0]
     if not first:
         return None
-    if not separator or not (
-        first.lower() == "localhost" or "." in first or ":" in first
-    ):
+    if not names_registry(ref):
         return "docker.io"
     try:
         return normalize_registry_host(first)
diff --git a/tst/unit/artifact/test_docker_image_pulled.py b/tst/unit/artifact/test_docker_image_pulled.py
new file mode 100644
index 00000000..8c37eee9
--- /dev/null
+++ b/tst/unit/artifact/test_docker_image_pulled.py
@@ -0,0 +1,52 @@
+"""An image with no tar.gz: read, written back, and loadable only when its name names the registry to pull it from."""
+
+import pytest
+
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
+
+REF = "ghcr.io/example/image@sha256:" + "0" * 64
+
+
+@pytest.mark.parametrize("document", [
+    {"id": "img", "version": 1, "type": "docker_image", "description": "d", "image_name": REF},
+    {"id": "img", "version": 1, "type": "docker_image", "description": "d", "image_name": REF,
+     "tar_gz_s3_url": None, "tar_gz_object_url": None},
+])
+def test_a_document_with_no_tar_gz_is_read(document):
+    image = DockerImageArtifact.model_validate(document)
+
+    assert image.tar_gz_object_url is None
+    assert image.load_problem() is None
+
+
+def test_one_is_written_with_both_spellings_of_its_tar_gz_empty_and_reads_back_the_same():
+    image = DockerImageArtifact(id="img", version=1, description="d", image_name=REF)
+
+    dumped = image.model_dump()
+
+    assert dumped["tar_gz_s3_url"] is None and dumped["tar_gz_object_url"] is None
+    assert DockerImageArtifact.model_validate(dumped) == image
+
+
+@pytest.mark.parametrize("image_name, tar_gz, problem", [
+    ("img:v1", "s3://bucket/img.tar.gz", None),
+    (REF, None, None),
+    ("localhost:5000/img:v1", None, None),
+    ("img:v1", None, "'img' v1 has no tar.gz, and its image name 'img:v1' doesn't name a registry to pull it from"),
+    ("team/img:v1", None, "'img' v1 has no tar.gz, and its image name 'team/img:v1' doesn't name a registry to pull it "
+                          "from"),
+])
+def test_an_image_loads_from_its_tar_gz_or_by_pulling_a_name_that_names_its_registry(image_name, tar_gz, problem):
+    image = DockerImageArtifact(id="img", version=1, description="d", image_name=image_name, tar_gz_object_url=tar_gz)
+
+    assert image.load_problem() == problem
+
+
+def test_one_with_no_tar_gz_has_no_bytes_to_load():
+    with pytest.raises(ValueError, match="has no tar.gz; its image is pulled from ghcr.io/example/image@sha256:"):
+        DockerImageArtifact(id="img", version=1, description="d", image_name=REF).load()
+
+
+def test_put_tar_refuses_an_empty_tar_gz():
+    with pytest.raises(ValueError, match="tar_gz_object_url is empty"):
+        DockerImageArtifact.put_tar("img", description="d", image_name="img:1", tar_gz_object_url="")
diff --git a/tst/unit/bundle/preflight_test.py b/tst/unit/bundle/preflight_test.py
index 2372c98a..e8204580 100644
--- a/tst/unit/bundle/preflight_test.py
+++ b/tst/unit/bundle/preflight_test.py
@@ -17,6 +17,7 @@
 from agent_env.artifact.store import get_artifact_store
 from agent_env.bundle import BundleError, dry_run_bundle, run_bundle
 from agent_env.cli import cli
+from agent_env.config import get_config
 from agent_env.config.runtime import Config
 from agent_env.env import Env, GatewayEnv, MCPServerEnv, MultiEnv
 from agent_env.env import bootstrap
@@ -152,6 +153,33 @@ def test_a_sandbox_image_in_this_machines_registry_is_refused_on_another_provide
     ]
 
 
+# Images with no tarball, which a sandbox pulls by name
+
+PULLED = ("ghcr.io/team/img@sha256:" + "0" * 64, None)
+
+
+@pytest.mark.parametrize("sandbox", ["local", "modal", "modal_vm"])
+def test_an_image_with_no_tarball_is_pulled_so_it_runs_on_any_provider(bundle_dir, sandbox):
+    _agent("solver", PULLED)
+    _task(bundle_dir, [AGENT])
+
+    assert [entry.name for entry in dry_run_bundle(bundle_dir, sandbox=sandbox).runs] == ["t"]
+
+
+@pytest.mark.parametrize("sandbox", ["local", "modal"])
+def test_an_image_with_no_tarball_and_no_registry_to_pull_it_from_is_refused_on_every_provider(bundle_dir, sandbox):
+    _agent("solver", ("img:v1", None))
+    _env("crm", ("crm:v1", None))
+    _task(bundle_dir, [AGENT, {"id": "env", "type": "deploy_env", "env_id": "crm"}])
+
+    problems = _problems(lambda: dry_run_bundle(bundle_dir, sandbox=sandbox))
+
+    assert ("tasks/t.json: step 'agent': deploys agent 'solver''s image, which no sandbox can get: 'solver-image' v1 has "
+            "no tar.gz, and its image name 'img:v1' doesn't name a registry to pull it from") in problems
+    assert ("tasks/t.json: step 'env': deploys env 'crm''s image 'crm-image', which no sandbox can get: 'crm-image' v1 "
+            "has no tar.gz, and its image name 'crm:v1' doesn't name a registry to pull it from") in problems
+
+
 # What a provider can create
 
 
@@ -281,6 +309,17 @@ def test_a_step_that_names_no_agent_needs_the_default_in_the_store(bundle_dir, m
     dry_run_bundle(bundle_dir)
 
 
+def test_a_default_agent_the_store_cant_read_is_reported_not_raised(bundle_dir, monkeypatch):
+    monkeypatch.setattr("agent_env.config.runtime.Config.get_default_a2a_agent_id", lambda self: "house-agent")
+    get_config().get_document_store().insert("a2a_agents", {"id": "house-agent", "version": 1, "type": "a2a_agent"})
+    _task(bundle_dir, [{"id": "agent", "type": "deploy_agent", "env_ids": []}])
+
+    (problem,) = _problems(lambda: dry_run_bundle(bundle_dir))
+
+    assert problem.startswith("tasks/t.json: step 'agent': names no agent, so it deploys the default, 'house-agent', "
+                              "and agent 'house-agent' can't be read (KeyError:")
+
+
 def test_a_judge_the_task_deploys_itself_or_the_direct_llm_judge_needs_no_default_agent(bundle_dir, monkeypatch):
     monkeypatch.setattr("agent_env.config.runtime.Config.get_default_a2a_agent_id", lambda self: "house-agent")
     judge = {"type": "rubrics_verifier", "prompt_id": "p", "verifier_id": "v", "criteria": [{"id": "c", "description": "done"}]}
diff --git a/tst/unit/providers/sandbox_providers/e2b_sandbox_test.py b/tst/unit/providers/sandbox_providers/e2b_sandbox_test.py
index 92a12f04..608a95ad 100644
--- a/tst/unit/providers/sandbox_providers/e2b_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/e2b_sandbox_test.py
@@ -7,12 +7,17 @@
 
 import pytest
 
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from agent_env.providers.sandbox_providers.e2b.sandbox import E2BSandbox
 from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy, VmSandbox
 from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_VM
 from agent_env.store import set_object_store
 
 
+def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
+    return DockerImageArtifact(id="img", description="", image_name=image_name, tar_gz_object_url=tar_gz_object_url)
+
+
 def _inner() -> MagicMock:
     sandbox = MagicMock()
     sandbox.sandbox_id = "e2b-test"
@@ -219,7 +224,7 @@ async def test_restricted_image_loading_adds_signed_host_to_applied_policy(
         allow_cidrs=("10.0.0.0/8",),
     )
     sandbox = E2BSandbox(inner, network_policy=policy)
-    artifact = SimpleNamespace(tar_gz_object_url="s3://bucket/image.tar.gz")
+    artifact = _image("s3://bucket/image.tar.gz")
     store = MagicMock()
     store.signed_get_url.return_value = "https://downloads.example/path?signature=x"
     set_object_store(store)
@@ -258,7 +263,7 @@ async def fail_load(_self, _artifacts, _signed_urls):
     monkeypatch.setattr(VmSandbox, "_load_docker_images", fail_load)
 
     with pytest.raises(RuntimeError, match="download failed"):
-        await sandbox.load_docker_images([SimpleNamespace(tar_gz_object_url="s3://bucket/img")])
+        await sandbox.load_docker_images([_image("s3://bucket/img")])
 
     inner.update_network.assert_awaited_once_with(
         {
@@ -273,12 +278,35 @@ async def fail_load(_self, _artifacts, _signed_urls):
 @pytest.mark.asyncio
 async def test_reconnected_sandbox_with_unknown_policy_refuses_image_load():
     sandbox = E2BSandbox(_inner(), network_policy=None)
-    artifact = SimpleNamespace(tar_gz_object_url="s3://bucket/image.tar.gz")
+    artifact = _image("s3://bucket/image.tar.gz")
 
     with pytest.raises(RuntimeError, match="applied network policy is unknown"):
         await sandbox.load_docker_images([artifact])
 
 
+@pytest.mark.asyncio
+async def test_restricted_image_loading_widens_the_policy_for_tarballs_only_and_pulls_the_rest_through_it(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    inner = _inner()
+    inner.update_network = AsyncMock()
+    policy = NetworkPolicy(mode=NetworkMode.ALLOWLIST, allow_hosts=("workload.example",))
+    sandbox = E2BSandbox(inner, network_policy=policy)
+    tarball, pulled = _image("s3://bucket/image.tar.gz"), _image(None, "ghcr.io/team/tool:v1")
+    store = MagicMock()
+    store.signed_get_url.return_value = "https://downloads.example/image.tar.gz"
+    set_object_store(store)
+    base_load, pull = AsyncMock(), AsyncMock()
+    monkeypatch.setattr(VmSandbox, "_load_docker_images", base_load)
+    monkeypatch.setattr(VmSandbox, "pull_images", pull)
+
+    await sandbox.load_docker_images([tarball, pulled])
+
+    assert sandbox.network_policy == policy.with_hosts(["downloads.example"])
+    base_load.assert_awaited_once_with([tarball], ["https://downloads.example/image.tar.gz"])
+    pull.assert_awaited_once_with(["ghcr.io/team/tool:v1"])
+
+
 @pytest.mark.asyncio
 async def test_allow_all_network_policy_explicitly_enables_internet_without_ingress_field():
     inner = _inner()
diff --git a/tst/unit/providers/sandbox_providers/local_sandbox_test.py b/tst/unit/providers/sandbox_providers/local_sandbox_test.py
index 9591b018..a9b0d204 100644
--- a/tst/unit/providers/sandbox_providers/local_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/local_sandbox_test.py
@@ -9,6 +9,7 @@
 
 import agent_env.providers.sandbox_providers.local_sandbox as ls
 from agent_env.a2a_agent.a2a_agent import A2AAgent
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from agent_env.config import reset_config, set_object_store
 from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox, LocalSandboxProvider
 from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER, SANDBOX_MODE_VM
@@ -19,6 +20,10 @@
 from agent_env.a2a_agent.a2a_agent import A2AAgent
 from tst.unit.store.fakes import FakeObjectStore
 
+
+def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
+    return DockerImageArtifact(id="img", description="", image_name=image_name, tar_gz_object_url=tar_gz_object_url)
+
 _real_copy_into_container = ls._copy_into_container  # before the fixtures below replace them
 _real_local_grant_trust = ls.local_grant_trust
 
@@ -214,7 +219,7 @@ def signed_get_url(self, url, expires_in=3600):
         def download_to_file(self, url, dest_path):
             self.staged.append(dest_path)
 
-    artifact = SimpleNamespace(tar_gz_object_url="file:///store/svc.tar.gz", image_name="svc:latest")
+    artifact = _image("file:///store/svc.tar.gz", "svc:latest")
     store = _Local()
     set_object_store(store)
     try:
diff --git a/tst/unit/providers/sandbox_providers/sail_vm_sandbox_test.py b/tst/unit/providers/sandbox_providers/sail_vm_sandbox_test.py
index e60027a7..fbae7578 100644
--- a/tst/unit/providers/sandbox_providers/sail_vm_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/sail_vm_sandbox_test.py
@@ -7,10 +7,15 @@
 
 import pytest
 
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from agent_env.providers.sandbox_providers.sail_vm.sandbox import SailVmSandbox, egress_document, policy_from_document
 from agent_env.providers.sandbox_providers.sandbox import NetworkMode, NetworkPolicy
 
 
+def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
+    return DockerImageArtifact(id="img", description="", image_name=image_name, tar_gz_object_url=tar_gz_object_url)
+
+
 class _HostLost(Exception):
     pass
 
@@ -225,7 +230,19 @@ async def test_terminate_tolerates_an_already_deleted_sailbox():
 async def test_image_loading_fails_closed_when_the_policy_is_unknown():
     sandbox, _ = _sandbox(policy=None)
     with pytest.raises(RuntimeError, match="applied egress policy is unknown"):
-        await sandbox.load_docker_images([object()])
+        await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz")])
+
+
+@pytest.mark.asyncio
+async def test_an_image_pulled_by_name_leaves_the_policy_alone_even_when_it_is_unknown(monkeypatch):
+    sandbox, sailbox = _sandbox(policy=None)
+    pull = AsyncMock()
+    monkeypatch.setattr(SailVmSandbox, "pull_images", pull)
+
+    await sandbox.load_docker_images([_image(None, "ghcr.io/team/tool:v1")])
+
+    pull.assert_awaited_once_with(["ghcr.io/team/tool:v1"])
+    sailbox.set_egress_policy.aio.assert_not_awaited()
 
 
 @pytest.mark.asyncio
@@ -237,14 +254,15 @@ async def test_image_loading_widens_an_allowlist_with_the_signed_download_hosts(
     )
     load = AsyncMock()
     monkeypatch.setattr(SailVmSandbox, "_load_docker_images", load)
+    images = [_image("s3://bucket/a.tar.gz"), _image("file:///store/b.tar.gz")]
 
-    await sandbox.load_docker_images(["a", "b"])
+    await sandbox.load_docker_images(images)
 
     sailbox.set_egress_policy.aio.assert_awaited_once_with(
         {"allowlist": ["pypi.org", "bucket.s3.amazonaws.com", "10.0.0.0/8"]}
     )
     assert sandbox.network_policy.allow_hosts == ("pypi.org", "bucket.s3.amazonaws.com")
-    load.assert_awaited_once_with(["a", "b"], ["https://bucket.s3.amazonaws.com/a?sig=1", None])
+    load.assert_awaited_once_with(images, ["https://bucket.s3.amazonaws.com/a?sig=1", None])
 
 
 @pytest.mark.asyncio
@@ -252,7 +270,7 @@ async def test_image_loading_leaves_an_allow_all_policy_alone(monkeypatch):
     sandbox, sailbox = _sandbox()
     monkeypatch.setattr(SailVmSandbox, "_signed_image_urls", AsyncMock(return_value=["https://x.example/a"]))
     monkeypatch.setattr(SailVmSandbox, "_load_docker_images", AsyncMock())
-    await sandbox.load_docker_images(["a"])
+    await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz")])
     sailbox.set_egress_policy.aio.assert_not_awaited()
     sandbox._sdk.Sailbox.get.aio.assert_not_awaited()
 
@@ -263,7 +281,7 @@ async def test_widening_past_sails_allowlist_limit_is_refused(monkeypatch):
     sandbox, sailbox = _sandbox(policy=full)
     monkeypatch.setattr(SailVmSandbox, "_signed_image_urls", AsyncMock(return_value=["https://bucket.example/a"]))
     with pytest.raises(RuntimeError, match="exceed Sail's 128-entry egress allowlist"):
-        await sandbox.load_docker_images(["a"])
+        await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz")])
     sailbox.set_egress_policy.aio.assert_not_awaited()
 
 
@@ -275,7 +293,7 @@ async def test_a_host_under_an_allowed_wildcard_needs_no_new_entry(monkeypatch):
     sandbox, sailbox = _sandbox(policy=full)
     monkeypatch.setattr(SailVmSandbox, "_signed_image_urls", AsyncMock(return_value=["https://bucket.s3.amazonaws.com/a"]))
     monkeypatch.setattr(SailVmSandbox, "_load_docker_images", AsyncMock())
-    await sandbox.load_docker_images(["a"])
+    await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz")])
     sailbox.set_egress_policy.aio.assert_not_awaited()
     sandbox._sdk.Sailbox.get.aio.assert_not_awaited()
 
@@ -298,7 +316,7 @@ async def test_a_host_already_allowed_is_not_reapplied(monkeypatch):
     sandbox, sailbox = _sandbox(policy=_ALLOWLIST)
     monkeypatch.setattr(SailVmSandbox, "_signed_image_urls", AsyncMock(return_value=["https://pypi.org/a"]))
     monkeypatch.setattr(SailVmSandbox, "_load_docker_images", AsyncMock())
-    await sandbox.load_docker_images(["a"])
+    await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz")])
     sailbox.set_egress_policy.aio.assert_not_awaited()
 
 
diff --git a/tst/unit/providers/sandbox_providers/sandbox_provider_test.py b/tst/unit/providers/sandbox_providers/sandbox_provider_test.py
index 70385c42..f674bc7d 100644
--- a/tst/unit/providers/sandbox_providers/sandbox_provider_test.py
+++ b/tst/unit/providers/sandbox_providers/sandbox_provider_test.py
@@ -12,8 +12,8 @@
 from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox, LocalSandboxProvider
 from agent_env.providers.sandbox_providers.modal_sandbox import ModalSandbox, ModalSandboxProvider
 from agent_env.providers.sandbox_providers.modal_vm_sandbox import ModalVmSandbox, ModalVmSandboxProvider
-from agent_env.providers.sandbox_providers.sandbox import Sandbox, VmSandbox
-from agent_env.providers.sandbox_providers.sandbox_provider import _BUILTIN_SANDBOX_PROVIDERS, SandboxProvider, _pull, build_sandbox_provider
+from agent_env.providers.sandbox_providers.sandbox import Sandbox, VmSandbox, _pull
+from agent_env.providers.sandbox_providers.sandbox_provider import _BUILTIN_SANDBOX_PROVIDERS, SandboxProvider, build_sandbox_provider
 from agent_env.store import ImageStore, RegistryAuth
 from tst.util.exec_scripts import script_run
 
diff --git a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
index 511ea945..361f3d5f 100644
--- a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
@@ -5,18 +5,22 @@
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
-from types import SimpleNamespace
 
 import pytest
 
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from agent_env.providers.sandbox_providers import sandbox as sandbox_module
 from agent_env.providers.sandbox_providers.sandbox import VmSandbox
-from agent_env.store import set_object_store
+from agent_env.store import ImageStore, RegistryAuth, set_object_store
 from agent_env.config import get_config, reset_config
 from tst.unit.store.fakes import ConfiguredObjectStore
 from tst.util.exec_scripts import script_run
 
 
+def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
+    return DockerImageArtifact(id="img", description="", image_name=image_name, tar_gz_object_url=tar_gz_object_url)
+
+
 class _SigningStore:
     """Fake object store that 'signs' any url into a fake https URL (the VM-curl path)."""
 
@@ -84,7 +88,7 @@ async def test_load_docker_images_downloads_to_file_before_load(signing_store):
     must only apply to a `-o file` download.
     """
     sandbox = _RecordingVmSandbox(images_stdout="myimage\n")
-    artifact = SimpleNamespace(tar_gz_object_url="s3://bucket/img.tar.gz", image_name="myimage:latest")
+    artifact = _image("s3://bucket/img.tar.gz", "myimage:latest")
 
     await sandbox.load_docker_images([artifact])
 
@@ -103,8 +107,8 @@ async def test_load_docker_images_downloads_to_file_before_load(signing_store):
 async def test_load_docker_images_uses_unique_tmp_per_artifact(signing_store):
     sandbox = _RecordingVmSandbox(images_stdout="a\nb\n")
     artifacts = [
-        SimpleNamespace(tar_gz_object_url="s3://bucket/a.tar.gz", image_name="a:1"),
-        SimpleNamespace(tar_gz_object_url="s3://bucket/b.tar.gz", image_name="b:1"),
+        _image("s3://bucket/a.tar.gz", "a:1"),
+        _image("s3://bucket/b.tar.gz", "b:1"),
     ]
 
     await sandbox.load_docker_images(artifacts)
@@ -114,6 +118,74 @@ async def test_load_docker_images_uses_unique_tmp_per_artifact(signing_store):
     assert "/tmp/_docker_image_vm-test_1.tar.gz" in load_script
 
 
+@pytest.mark.asyncio
+async def test_a_load_of_tarballs_runs_the_script_it_always_has(signing_store):
+    sandbox = _RecordingVmSandbox(images_stdout="a\nb\n")
+
+    await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz", "a:1"), _image("s3://bucket/b.tar.gz", "b:1")])
+
+    flags = "--retry 5 --retry-all-errors --retry-delay 1"
+    assert sandbox.scripts == [
+        f'(curl -fsSL {flags} "https://signed/a.tar.gz" -o /tmp/_docker_image_vm-test_0.tar.gz '
+        "&& gunzip -c /tmp/_docker_image_vm-test_0.tar.gz | docker load && rm -f /tmp/_docker_image_vm-test_0.tar.gz) & "
+        f'(curl -fsSL {flags} "https://signed/b.tar.gz" -o /tmp/_docker_image_vm-test_1.tar.gz '
+        "&& gunzip -c /tmp/_docker_image_vm-test_1.tar.gz | docker load && rm -f /tmp/_docker_image_vm-test_1.tar.gz) & wait"
+    ]
+
+
+class _Registry(ImageStore):
+    """Holds credentials for registry.example only."""
+
+    def image_ref(self, repository, tag):
+        return f"registry.example/{repository}:{tag}"
+
+    def auth(self, ref):
+        return RegistryAuth("registry.example", "user", "token") if ref.startswith("registry.example/") else None
+
+
+@pytest.fixture
+def registry():
+    get_config().set_image_store(_Registry())
+    yield
+    reset_config()
+
+
+PRIVATE = "registry.example/team/app@sha256:" + "0" * 64
+PUBLIC = "ghcr.io/team/tool:v1"
+
+
+@pytest.mark.asyncio
+async def test_an_image_with_no_tarball_is_pulled_after_logging_in_to_the_registry_the_store_holds(registry):
+    sandbox = _RecordingVmSandbox()
+
+    await sandbox.load_docker_images([_image(None, PRIVATE), _image(None, PUBLIC), _image(None, PRIVATE)])
+
+    login, *pulls = sandbox.scripts
+    assert login == "echo token | docker login --username user --password-stdin registry.example"
+    assert sorted(pulls) == sorted([f"docker pull {PRIVATE}", f"docker pull {PUBLIC}"])
+
+
+@pytest.mark.asyncio
+async def test_tarballs_are_loaded_and_the_rest_pulled(signing_store, registry):
+    sandbox = _RecordingVmSandbox(images_stdout="a\n")
+
+    await sandbox.load_docker_images([_image(None, PUBLIC), _image("s3://bucket/a.tar.gz", "a:1")])
+
+    load, pull = sandbox.scripts
+    assert '"https://signed/a.tar.gz"' in load and PUBLIC not in load
+    assert pull == f"docker pull {PUBLIC}"
+
+
+@pytest.mark.asyncio
+async def test_an_image_no_sandbox_can_get_is_refused_before_anything_is_loaded(signing_store):
+    sandbox = _RecordingVmSandbox(images_stdout="a\n")
+
+    with pytest.raises(RuntimeError, match="Can't load images: 'img' v0 has no tar.gz, and its image name 'img:v1' doesn't name"):
+        await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz", "a:1"), _image(None, "img:v1")])
+
+    assert sandbox.scripts == []
+
+
 class _ScriptRecorder(VmSandbox):
     """Records exec_script invocations, running the real write_file_from_text."""
 
@@ -257,7 +329,7 @@ def open(self, object_url):
     set_object_store(_LocalStore())
     try:
         sandbox = _PushTargetVmSandbox(b"IMGBYTES", images_stdout="myimage\n")
-        artifact = SimpleNamespace(tar_gz_object_url="file:///store/img.tar.gz", image_name="myimage:latest")
+        artifact = _image("file:///store/img.tar.gz", "myimage:latest")
         await sandbox.load_docker_images([artifact])
     finally:
         reset_config()
@@ -298,7 +370,7 @@ async def test_the_store_is_called_off_the_event_loop(signs):
     set_object_store(store)
     try:
         sandbox = _PushTargetVmSandbox(b"IMG", images_stdout="a\nb\n")
-        artifacts = [SimpleNamespace(tar_gz_object_url=f"s3://bucket/{n}.tar.gz", image_name=f"{n}:1") for n in "ab"]
+        artifacts = [_image(f"s3://bucket/{n}.tar.gz", f"{n}:1") for n in "ab"]
         await sandbox.load_docker_images(artifacts)
         await sandbox.load_object_file("s3://bucket/data.json", "/tmp/data.json")
     finally:
@@ -349,7 +421,7 @@ async def test_a_failed_sign_stops_the_signs_still_waiting():
     store = _FailingSigner()
     set_object_store(store)
     try:
-        artifacts = [SimpleNamespace(tar_gz_object_url=f"s3://bucket/{n}.tar.gz", image_name=f"{n}:1") for n in range(20)]
+        artifacts = [_image(f"s3://bucket/{n}.tar.gz", f"{n}:1") for n in range(20)]
         with pytest.raises(ConnectionError):
             await _RecordingVmSandbox().load_docker_images(artifacts)
         await asyncio.sleep(0.2)
@@ -385,7 +457,7 @@ async def test_a_load_bounds_how_many_signs_run_at_once():
         names = [f"i{n}" for n in range(20)]
         sandbox = _RecordingVmSandbox(images_stdout="\n".join(names))
         await sandbox.load_docker_images(
-            [SimpleNamespace(tar_gz_object_url=f"s3://bucket/{n}.tar.gz", image_name=f"{n}:1") for n in names]
+            [_image(f"s3://bucket/{n}.tar.gz", f"{n}:1") for n in names]
         )
     finally:
         reset_config()
@@ -473,6 +545,6 @@ async def push(sandbox, store, object_url, vm_path):
     sandbox = _RecordingVmSandbox(images_stdout="myimage\n")
 
     await sandbox.load_object_file(data, "/tmp/data.json")
-    await sandbox.load_docker_images([SimpleNamespace(tar_gz_object_url=image, image_name="myimage:latest")])
+    await sandbox.load_docker_images([_image(image, "myimage:latest")])
 
     assert pushed == [(local, data), (local, image)]
diff --git a/tst/unit/store/oci_registry_image_store_test.py b/tst/unit/store/oci_registry_image_store_test.py
index 6c914bbf..acd41259 100644
--- a/tst/unit/store/oci_registry_image_store_test.py
+++ b/tst/unit/store/oci_registry_image_store_test.py
@@ -3,7 +3,7 @@
 import pytest
 
 from agent_env.store import ImageStore, OciRegistryCredentials, OciRegistryImageStore, RegistryAuth
-from agent_env.store.image_store import registry_host_from_ref
+from agent_env.store.image_store import names_registry, registry_host_from_ref
 
 
 class _Credentials(OciRegistryCredentials):
@@ -54,6 +54,22 @@ def test_unqualified_references_resolve_to_docker_hub():
     assert registry_host_from_ref("/invalid:v1") is None
 
 
+@pytest.mark.parametrize("ref, named", [
+    ("ghcr.io/example/image:v1", True),
+    ("registry.example:5000/image@sha256:" + "0" * 64, True),
+    ("localhost/image:v1", True),
+    ("LOCALHOST:5000/image", True),
+    ("docker.io/library/ubuntu:24.04", True),
+    ("ubuntu:24.04", False),
+    ("library/ubuntu:24.04", False),
+    ("team/image:v1", False),
+    ("", False),
+    ("https://ghcr.io/example/image:v1", False),
+])
+def test_a_reference_names_its_registry_only_as_docker_reads_one(ref, named):
+    assert names_registry(ref) is named
+
+
 @pytest.mark.parametrize("host", ["", "   ", "https://ghcr.io", "ghcr.io/team"])
 def test_rejects_invalid_registry_host(host):
     with pytest.raises(ValueError, match="registry_host"):

From 11ada1c92ccf0e7807d03d271e76998449ab0576 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 7 Oct 2026 20:54:07 -0700
Subject: [PATCH 2/6] fix(preflight): check that a remote deploy's stored infra
 images are loadable too

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/agent_env/bundle/preflight.py |  9 +++++----
 tst/unit/bundle/preflight_test.py | 13 +++++++++++++
 2 files changed, 18 insertions(+), 4 deletions(-)

diff --git a/src/agent_env/bundle/preflight.py b/src/agent_env/bundle/preflight.py
index e22212bd..2624e469 100644
--- a/src/agent_env/bundle/preflight.py
+++ b/src/agent_env/bundle/preflight.py
@@ -348,10 +348,11 @@ def _remote_infra(self, where: str, provider: SandboxProvider, kinds: set[str])
                                      f"{env_id!r}, and the store doesn't hold it; agent-env builds it only for the local "
                                      f"sandbox provider, so put it in a store that provider can reach (`{put_command(kind)}`)")
                 continue
-            images = ([env.db_docker_image_artifact, env.db_web_docker_image_artifact, env.db_mcp_docker_image_artifact]
-                      if isinstance(env, ServiceDBEnv) else [env.docker_image_artifact])
-            self._reachable(where, [provider], [_image(f"the {kind} env {env_id!r}'s image {image.id!r}", image)
-                                                for image in images])
+            artifacts = ([env.db_docker_image_artifact, env.db_web_docker_image_artifact,
+                          env.db_mcp_docker_image_artifact] if isinstance(env, ServiceDBEnv) else [env.docker_image_artifact])
+            images = [_image(f"the {kind} env {env_id!r}'s image {image.id!r}", image) for image in artifacts]
+            self._loadable(where, images)
+            self._reachable(where, [provider], images)
 
     def _default_agent(self) -> None:
         if not self.default_agent_users:
diff --git a/tst/unit/bundle/preflight_test.py b/tst/unit/bundle/preflight_test.py
index e8204580..5c5ceb2e 100644
--- a/tst/unit/bundle/preflight_test.py
+++ b/tst/unit/bundle/preflight_test.py
@@ -401,6 +401,19 @@ def test_a_gateway_deploy_on_another_provider_needs_infra_it_can_reach(bundle_di
     assert len(problems) == 4  # the gateway's image, and the service-db's three
 
 
+def test_infra_on_another_provider_with_no_tarball_and_no_registry_to_pull_it_from_is_refused(bundle_dir):
+    _env("crm", REMOTE)
+    _infra(("img:v1", None))
+    _task(bundle_dir, [{"id": "env", "type": "deploy_env", "env_id": "crm"}])
+
+    problems = _problems(lambda: dry_run_bundle(bundle_dir, sandbox="modal_vm"))
+
+    assert problems[0] == ("tasks/t.json: step 'env': deploys the gateway env 'default''s image 'gateway-default', which "
+                           "no sandbox can get: 'gateway-default' v1 has no tar.gz, and its image name 'img:v1' doesn't "
+                           "name a registry to pull it from")
+    assert len(problems) == 4  # the gateway's image, and the service-db's three
+
+
 def test_a_problem_several_deploys_share_is_reported_once_naming_the_first(bundle_dir):
     _env("crm")
     with namespace_routing():

From b6e664f0c238e9c99edda9ad633903b73996639c Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 7 Oct 2026 21:12:57 -0700
Subject: [PATCH 3/6] fix(preflight): check that an image is loadable only
 where a VM loads it

A container path (Modal's gateway and agents, the local provider's agents) runs an
image by name whatever it holds, and Modal's container gateway swaps out service-db
images the image store doesn't hold, so an image with no tar.gz is refused only on
a gateway VM, an agent VM, or a VM provider's infra.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/agent_env/bundle/preflight.py | 23 +++++++++++-----
 tst/unit/bundle/preflight_test.py | 44 ++++++++++++++++++++++---------
 2 files changed, 47 insertions(+), 20 deletions(-)

diff --git a/src/agent_env/bundle/preflight.py b/src/agent_env/bundle/preflight.py
index 2624e469..3c30a9b6 100644
--- a/src/agent_env/bundle/preflight.py
+++ b/src/agent_env/bundle/preflight.py
@@ -180,7 +180,9 @@ def _env(self, where: str, step: DeployEnvTaskStep) -> None:
         else:  # a plugin's provider, or one deploy_env's own preflight refuses for this env
             return
         env_id, images = step.env_id, deployment.images
-        self._loadable(where, images)
+        if GATEWAY in kinds and (loaders := [link for link in _links(provider)
+                                             if not isinstance(link, ModalSandboxProvider)]):
+            self._loadable(where, loaders, images)  # a gateway VM loads its images; Modal's runs each by name
         if _local_link(provider):
             self.infra |= kinds
             self.docker_users.append(where)
@@ -289,10 +291,13 @@ def _agent_deploy(self, where: str, provider: SandboxProvider, agent_id: str | N
         if agent_id is None:
             self.default_agent_users.append((where, unnamed))
             agent_id = get_config().get_default_a2a_agent_id()
-        if (image := self._agent_image(agent_id, version)) is None:
+        loaders = [link for link in _links(provider) if not isinstance(link, (LocalSandboxProvider, ModalSandboxProvider))]
+        remote = _remote_links(provider)
+        if not (loaders or remote) or (image := self._agent_image(agent_id, version)) is None:
             return
-        self._loadable(where, [image])
-        if remote := _remote_links(provider):
+        if loaders:
+            self._loadable(where, loaders, [image])  # those two run an agent in a container, by image name
+        if remote:
             self._reachable(where, remote, [image])
 
     def _sandbox(self, where: str, step: DeploySandboxTaskStep) -> None:
@@ -327,10 +332,13 @@ def _agent_image(self, agent_id: str, version: int | None) -> _Image | None:
         except (NotFoundError, ValueError, KeyError, TypeError):
             return None
 
-    def _loadable(self, where: str, images: list[_Image]) -> None:
+    def _loadable(self, where: str, loaders: list[SandboxProvider], images: list[_Image]) -> None:
+        """Refuse each of ``images`` a VM on ``loaders`` can't get: it loads an image's tar.gz, or pulls an image with
+        none by name."""
         for image in images:
             if image.unloadable:
-                self._problem(where, f"deploys {image.what}, which no sandbox can get: {image.unloadable}")
+                self._problem(where, f"deploys {image.what} on the {_shown(loaders[0])} sandbox provider, which can't "
+                                     f"load it: {image.unloadable}")
 
     def _reachable(self, where: str, remote: list[SandboxProvider], images: list[_Image]) -> None:
         for image in images:
@@ -351,7 +359,8 @@ def _remote_infra(self, where: str, provider: SandboxProvider, kinds: set[str])
             artifacts = ([env.db_docker_image_artifact, env.db_web_docker_image_artifact,
                           env.db_mcp_docker_image_artifact] if isinstance(env, ServiceDBEnv) else [env.docker_image_artifact])
             images = [_image(f"the {kind} env {env_id!r}'s image {image.id!r}", image) for image in artifacts]
-            self._loadable(where, images)
+            if not isinstance(provider, ModalSandboxProvider):  # whose containers run by name, or are swapped out
+                self._loadable(where, [provider], images)
             self._reachable(where, [provider], images)
 
     def _default_agent(self) -> None:
diff --git a/tst/unit/bundle/preflight_test.py b/tst/unit/bundle/preflight_test.py
index 5c5ceb2e..f8d39e96 100644
--- a/tst/unit/bundle/preflight_test.py
+++ b/tst/unit/bundle/preflight_test.py
@@ -166,18 +166,35 @@ def test_an_image_with_no_tarball_is_pulled_so_it_runs_on_any_provider(bundle_di
     assert [entry.name for entry in dry_run_bundle(bundle_dir, sandbox=sandbox).runs] == ["t"]
 
 
-@pytest.mark.parametrize("sandbox", ["local", "modal"])
-def test_an_image_with_no_tarball_and_no_registry_to_pull_it_from_is_refused_on_every_provider(bundle_dir, sandbox):
-    _agent("solver", ("img:v1", None))
-    _env("crm", ("crm:v1", None))
-    _task(bundle_dir, [AGENT, {"id": "env", "type": "deploy_env", "env_id": "crm"}])
+NO_REGISTRY = ("img:v1", None)
+CANT_LOAD = ("on the {sandbox!r} sandbox provider, which can't load it: '{id}' v1 has no tar.gz, and its image name "
+             "'img:v1' doesn't name a registry to pull it from")
 
-    problems = _problems(lambda: dry_run_bundle(bundle_dir, sandbox=sandbox))
 
-    assert ("tasks/t.json: step 'agent': deploys agent 'solver''s image, which no sandbox can get: 'solver-image' v1 has "
-            "no tar.gz, and its image name 'img:v1' doesn't name a registry to pull it from") in problems
-    assert ("tasks/t.json: step 'env': deploys env 'crm''s image 'crm-image', which no sandbox can get: 'crm-image' v1 "
-            "has no tar.gz, and its image name 'crm:v1' doesn't name a registry to pull it from") in problems
+@pytest.mark.parametrize("sandbox, refused", [("modal_vm", True), ("local", False), ("modal", False)])
+def test_an_agent_image_with_no_tarball_and_no_registry_is_refused_where_a_vm_loads_it(bundle_dir, sandbox, refused):
+    """The local and Modal providers run an agent in a container, by image name, whether or not it has a tar.gz."""
+    _agent("solver", NO_REGISTRY)
+    _task(bundle_dir, [AGENT])
+
+    if refused:
+        assert _problems(lambda: dry_run_bundle(bundle_dir, sandbox=sandbox)) == [
+            "tasks/t.json: step 'agent': deploys agent 'solver''s image " + CANT_LOAD.format(sandbox=sandbox,
+                                                                                             id="solver-image")]
+    else:
+        assert [entry.name for entry in dry_run_bundle(bundle_dir, sandbox=sandbox).runs] == ["t"]
+
+
+def test_an_env_image_with_no_tarball_and_no_registry_is_refused_where_a_gateway_vm_loads_it(bundle_dir):
+    """The local provider's gateway is a VM, which loads the env's images; Modal's runs each server by image name."""
+    _env("crm", NO_REGISTRY)
+    _task(bundle_dir, [{"id": "env", "type": "deploy_env", "env_id": "crm"}])
+
+    assert _problems(lambda: dry_run_bundle(bundle_dir, sandbox="local")) == [
+        "tasks/t.json: step 'env': deploys env 'crm''s image 'crm-image' " + CANT_LOAD.format(sandbox="local",
+                                                                                              id="crm-image")]
+    _infra(REMOTE)
+    assert [entry.name for entry in dry_run_bundle(bundle_dir, sandbox="modal").runs] == ["t"]
 
 
 # What a provider can create
@@ -408,10 +425,11 @@ def test_infra_on_another_provider_with_no_tarball_and_no_registry_to_pull_it_fr
 
     problems = _problems(lambda: dry_run_bundle(bundle_dir, sandbox="modal_vm"))
 
-    assert problems[0] == ("tasks/t.json: step 'env': deploys the gateway env 'default''s image 'gateway-default', which "
-                           "no sandbox can get: 'gateway-default' v1 has no tar.gz, and its image name 'img:v1' doesn't "
-                           "name a registry to pull it from")
+    assert problems[0] == ("tasks/t.json: step 'env': deploys the gateway env 'default''s image 'gateway-default' "
+                           + CANT_LOAD.format(sandbox="modal_vm", id="gateway-default"))
     assert len(problems) == 4  # the gateway's image, and the service-db's three
+    # Modal's container gateway runs the gateway by image name and swaps out service-db images the store doesn't hold
+    assert [entry.name for entry in dry_run_bundle(bundle_dir, sandbox="modal").runs] == ["t"]
 
 
 def test_a_problem_several_deploys_share_is_reported_once_naming_the_first(bundle_dir):

From 7ab1f250b4b7d080b1d7dab9f24e4eb9dbd5de22 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 7 Oct 2026 21:36:45 -0700
Subject: [PATCH 4/6] fix(images): mint one registry login per registry, and
 cancel the other pulls when one fails

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../providers/sandbox_providers/sandbox.py    | 31 ++++++-----
 .../sandbox_providers/vm_sandbox_test.py      | 53 +++++++++++++++++++
 2 files changed, 72 insertions(+), 12 deletions(-)

diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index f55a22cb..94127919 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox.py
@@ -22,6 +22,7 @@
 from typing import IO, TYPE_CHECKING, Any, AsyncIterator, Callable, Iterable, Optional
 
 from agent_env.config import get_config
+from agent_env.store.image_store.oci_registry_credentials import registry_host_from_ref
 from agent_env.utils.deprecation import warn_deprecated
 from agent_env.utils.paths import validate_relative_filename
 
@@ -301,21 +302,27 @@ async def _load_tarballs(self, artifacts: list) -> None:
 
     async def pull_images(self, image_names: list[str]) -> None:
         """``docker pull`` each of ``image_names``, concurrently, after logging in to each registry the image store
-        holds credentials for. A network policy restricting egress isn't widened for them: it must allow the
+        holds credentials for: once per registry, since minting a login can be a network round trip. The first pull
+        to fail cancels the rest. A network policy restricting egress isn't widened for them: it must allow the
         registries itself."""
         image_names = list(dict.fromkeys(image_names))
-        config = get_config()
-        logins: dict[str, Any] = {}
-        for image_name in image_names:
-            auth = await asyncio.to_thread(config.get_image_store().auth, image_name)
+        store = get_config().get_image_store()
+        one_per_registry = {registry_host_from_ref(image_name): image_name for image_name in reversed(image_names)}
+        for image_name in one_per_registry.values():
+            auth = await asyncio.to_thread(store.auth, image_name)
             if auth is not None:
-                logins.setdefault(auth.registry, auth)
-        for auth in logins.values():
-            await self.exec_script(
-                f"echo {shlex.quote(auth.password)} | docker login "
-                f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}"
-            )
-        await asyncio.gather(*(_pull(self, image_name) for image_name in image_names))
+                await self.exec_script(
+                    f"echo {shlex.quote(auth.password)} | docker login "
+                    f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}"
+                )
+        pulls = [asyncio.ensure_future(_pull(self, image_name)) for image_name in image_names]
+        try:
+            await asyncio.gather(*pulls)
+        except BaseException:
+            for pull in pulls:
+                pull.cancel()
+            await asyncio.gather(*pulls, return_exceptions=True)
+            raise
 
     @staticmethod
     async def _signed_image_urls(artifacts: list) -> list[str | None]:
diff --git a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
index 361f3d5f..2c8d2c46 100644
--- a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
@@ -186,6 +186,59 @@ async def test_an_image_no_sandbox_can_get_is_refused_before_anything_is_loaded(
     assert sandbox.scripts == []
 
 
+@pytest.mark.asyncio
+async def test_a_login_is_minted_once_per_registry():
+    minted = []
+
+    class _Counting(_Registry):
+        def auth(self, ref):
+            minted.append(ref)
+            return super().auth(ref)
+
+    get_config().set_image_store(_Counting())
+    try:
+        sandbox = _RecordingVmSandbox()
+        await sandbox.pull_images([PRIVATE, "registry.example/team/other:v1", PUBLIC, "ghcr.io/team/more:v2"])
+    finally:
+        reset_config()
+
+    assert len(minted) == 2  # registry.example's and ghcr.io's
+    assert sum("docker login" in script for script in sandbox.scripts) == 1
+    assert sum(script.startswith("docker pull") for script in sandbox.scripts) == 4
+
+
+class _OnePullFailsVm(_RecordingVmSandbox):
+    """``docker pull`` of BAD fails; every other pull waits until it's cancelled."""
+
+    BAD = "ghcr.io/team/bad:v1"
+
+    def __init__(self):
+        super().__init__()
+        self.cancelled: list[str] = []
+
+    async def exec_script(self, script, *, max_retries=0):
+        if script == f"docker pull {self.BAD}":
+            await asyncio.sleep(0)
+            raise RuntimeError("Script failed (exit 1):\nstderr: manifest unknown")
+        if script.startswith("docker pull"):
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                self.cancelled.append(script)
+                raise
+        return ""
+
+
+@pytest.mark.asyncio
+async def test_the_first_pull_to_fail_cancels_the_rest(registry):
+    sandbox = _OnePullFailsVm()
+
+    with pytest.raises(RuntimeError, match="manifest unknown"):
+        await sandbox.pull_images([PUBLIC, _OnePullFailsVm.BAD, PRIVATE])
+
+    assert sorted(sandbox.cancelled) == sorted([f"docker pull {PUBLIC}", f"docker pull {PRIVATE}"])
+
+
 class _ScriptRecorder(VmSandbox):
     """Records exec_script invocations, running the real write_file_from_text."""
 

From c30b98497c48301c2b3e2c99404ae15c28c711f8 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 7 Oct 2026 21:51:25 -0700
Subject: [PATCH 5/6] fix(images): let every pull finish before raising the
 first failure

Cancelling an exec doesn't stop its command on the VM, and the Scale and Sail
sandboxes leave a cancelled exec's connection or readers open, so the other
pulls run to their end instead of being cancelled.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../providers/sandbox_providers/sandbox.py     | 18 +++++++-----------
 .../sandbox_providers/vm_sandbox_test.py       | 17 +++++++----------
 2 files changed, 14 insertions(+), 21 deletions(-)

diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index 94127919..a5efc6f8 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox.py
@@ -302,9 +302,10 @@ async def _load_tarballs(self, artifacts: list) -> None:
 
     async def pull_images(self, image_names: list[str]) -> None:
         """``docker pull`` each of ``image_names``, concurrently, after logging in to each registry the image store
-        holds credentials for: once per registry, since minting a login can be a network round trip. The first pull
-        to fail cancels the rest. A network policy restricting egress isn't widened for them: it must allow the
-        registries itself."""
+        holds credentials for: once per registry, since minting a login can be a network round trip. Every pull runs
+        to its end and the first failure is raised after: cancelling an exec doesn't stop its command on the VM, and
+        some providers leave its connection open. A network policy restricting egress isn't widened for them: it must
+        allow the registries itself."""
         image_names = list(dict.fromkeys(image_names))
         store = get_config().get_image_store()
         one_per_registry = {registry_host_from_ref(image_name): image_name for image_name in reversed(image_names)}
@@ -315,14 +316,9 @@ async def pull_images(self, image_names: list[str]) -> None:
                     f"echo {shlex.quote(auth.password)} | docker login "
                     f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}"
                 )
-        pulls = [asyncio.ensure_future(_pull(self, image_name)) for image_name in image_names]
-        try:
-            await asyncio.gather(*pulls)
-        except BaseException:
-            for pull in pulls:
-                pull.cancel()
-            await asyncio.gather(*pulls, return_exceptions=True)
-            raise
+        pulls = await asyncio.gather(*(_pull(self, image_name) for image_name in image_names), return_exceptions=True)
+        if failures := [pull for pull in pulls if isinstance(pull, BaseException)]:
+            raise failures[0]
 
     @staticmethod
     async def _signed_image_urls(artifacts: list) -> list[str | None]:
diff --git a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
index 2c8d2c46..3ed953f7 100644
--- a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
@@ -208,35 +208,32 @@ def auth(self, ref):
 
 
 class _OnePullFailsVm(_RecordingVmSandbox):
-    """``docker pull`` of BAD fails; every other pull waits until it's cancelled."""
+    """``docker pull`` of BAD fails at once; every other pull finishes a moment later."""
 
     BAD = "ghcr.io/team/bad:v1"
 
     def __init__(self):
         super().__init__()
-        self.cancelled: list[str] = []
+        self.finished: list[str] = []
 
     async def exec_script(self, script, *, max_retries=0):
         if script == f"docker pull {self.BAD}":
-            await asyncio.sleep(0)
             raise RuntimeError("Script failed (exit 1):\nstderr: manifest unknown")
         if script.startswith("docker pull"):
-            try:
-                await asyncio.Event().wait()
-            except asyncio.CancelledError:
-                self.cancelled.append(script)
-                raise
+            await asyncio.sleep(0.01)
+            self.finished.append(script)
         return ""
 
 
 @pytest.mark.asyncio
-async def test_the_first_pull_to_fail_cancels_the_rest(registry):
+async def test_a_failed_pull_is_raised_once_the_others_have_finished(registry):
+    """Cancelling an exec wouldn't stop its command on the VM, so the other pulls run to their end."""
     sandbox = _OnePullFailsVm()
 
     with pytest.raises(RuntimeError, match="manifest unknown"):
         await sandbox.pull_images([PUBLIC, _OnePullFailsVm.BAD, PRIVATE])
 
-    assert sorted(sandbox.cancelled) == sorted([f"docker pull {PUBLIC}", f"docker pull {PRIVATE}"])
+    assert sorted(sandbox.finished) == sorted([f"docker pull {PUBLIC}", f"docker pull {PRIVATE}"])
 
 
 class _ScriptRecorder(VmSandbox):

From 5fea87dd4d3c32754a69eb5707fcf0801b8db980 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 8 Oct 2026 07:34:07 -0700
Subject: [PATCH 6/6] fix(images): cancel the other pulls when one fails, so a
 stalled one can't hold the failure back

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../providers/sandbox_providers/sandbox.py    | 19 ++++++++++++-------
 .../sandbox_providers/vm_sandbox_test.py      | 17 ++++++++++-------
 2 files changed, 22 insertions(+), 14 deletions(-)

diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index a5efc6f8..fc4b210d 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox.py
@@ -302,10 +302,10 @@ async def _load_tarballs(self, artifacts: list) -> None:
 
     async def pull_images(self, image_names: list[str]) -> None:
         """``docker pull`` each of ``image_names``, concurrently, after logging in to each registry the image store
-        holds credentials for: once per registry, since minting a login can be a network round trip. Every pull runs
-        to its end and the first failure is raised after: cancelling an exec doesn't stop its command on the VM, and
-        some providers leave its connection open. A network policy restricting egress isn't widened for them: it must
-        allow the registries itself."""
+        holds credentials for: once per registry, since minting a login can be a network round trip. The first pull
+        to fail cancels the rest, since one that stalls would otherwise hold the failure back; the command a cancelled
+        exec started keeps running on the VM until it ends or the VM does. A network policy restricting egress isn't
+        widened for them: it must allow the registries itself."""
         image_names = list(dict.fromkeys(image_names))
         store = get_config().get_image_store()
         one_per_registry = {registry_host_from_ref(image_name): image_name for image_name in reversed(image_names)}
@@ -316,9 +316,14 @@ async def pull_images(self, image_names: list[str]) -> None:
                     f"echo {shlex.quote(auth.password)} | docker login "
                     f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}"
                 )
-        pulls = await asyncio.gather(*(_pull(self, image_name) for image_name in image_names), return_exceptions=True)
-        if failures := [pull for pull in pulls if isinstance(pull, BaseException)]:
-            raise failures[0]
+        pulls = [asyncio.ensure_future(_pull(self, image_name)) for image_name in image_names]
+        try:
+            await asyncio.gather(*pulls)
+        except BaseException:
+            for pull in pulls:
+                pull.cancel()
+            await asyncio.gather(*pulls, return_exceptions=True)
+            raise
 
     @staticmethod
     async def _signed_image_urls(artifacts: list) -> list[str | None]:
diff --git a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
index 3ed953f7..0d89bfe4 100644
--- a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
@@ -208,32 +208,35 @@ def auth(self, ref):
 
 
 class _OnePullFailsVm(_RecordingVmSandbox):
-    """``docker pull`` of BAD fails at once; every other pull finishes a moment later."""
+    """``docker pull`` of BAD fails; every other pull stalls until it's cancelled."""
 
     BAD = "ghcr.io/team/bad:v1"
 
     def __init__(self):
         super().__init__()
-        self.finished: list[str] = []
+        self.cancelled: list[str] = []
 
     async def exec_script(self, script, *, max_retries=0):
         if script == f"docker pull {self.BAD}":
+            await asyncio.sleep(0)
             raise RuntimeError("Script failed (exit 1):\nstderr: manifest unknown")
         if script.startswith("docker pull"):
-            await asyncio.sleep(0.01)
-            self.finished.append(script)
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                self.cancelled.append(script)
+                raise
         return ""
 
 
 @pytest.mark.asyncio
-async def test_a_failed_pull_is_raised_once_the_others_have_finished(registry):
-    """Cancelling an exec wouldn't stop its command on the VM, so the other pulls run to their end."""
+async def test_the_first_pull_to_fail_cancels_the_rest_so_a_stalled_one_cant_hold_it_back(registry):
     sandbox = _OnePullFailsVm()
 
     with pytest.raises(RuntimeError, match="manifest unknown"):
         await sandbox.pull_images([PUBLIC, _OnePullFailsVm.BAD, PRIVATE])
 
-    assert sorted(sandbox.finished) == sorted([f"docker pull {PUBLIC}", f"docker pull {PRIVATE}"])
+    assert sorted(sandbox.cancelled) == sorted([f"docker pull {PUBLIC}", f"docker pull {PRIVATE}"])
 
 
 class _ScriptRecorder(VmSandbox):