feat(bundle): task_images lists the images a task's import writes, downloading nothing

AGENTS.md

@@ -122,7 +122,7 @@ and deserialized through a registry.
 | `openenv/` | `build_app(bundle)`: the [OpenEnv](https://github.com/openenv/OpenEnv) server for a bundle's tasks, each session playing one episode through the Episode API, with the envs' MCP tools, an added `end_turn` tool and the task's verifier score as the reward. It prepares bundles as `agent-env run` does (`prepare_bundle(played=True)`, without the agent the client plays) and keeps namespace routing on for the app's life. `openenv` comes with the `openenv` extra, a floor with no cap; this package and `training/trl.py` are the only ones that import it. Each task's description in the Task API names its `toolset` (the played agent's envs, role, and whether it gets `bash`). |
 | `training/` | One module per outside trainer, each behind its own extra, and `_tools.py`, which turns an env's tools into typed, documented Python methods from their JSON Schema, as transformers' `get_json_schema` reads them. `trl.py` (`trl` extra): `AgentEnv(url)` over a server `openenv/` makes, with `dataset` (a row per task) and `factory` (TRL's `environment_factory`: a generated class per toolset whose methods are the tools, a background `reset` and an owned `get_reward`); `AgentEnv.local(bundle)` serves the bundle in process. |
 | `runner/` | The `[runner]` seam: `Runner.submit()` returns a `RunHandle` (`run_id`, `instance_id`) before the run starts; `LocalRunner` is built in. `runner/store.py` is the run index every runner writes and the explorer reads. |
-| `bundle/` | A bundle is a folder of entities (`envs/`, `agents/`, `artifacts/`, `skills/`, `tasks/<name>.json`, `evals/<name>.toml`) that runs without a registry: `agent-env run` plans it, writes it into the `@local` namespace (reusing each version whose inputs haven't changed) and runs its evals or tasks; `agent-env bundle init` writes a stored task or eval as one, with every env, agent and artifact it uses in its own folder (each image as its build context, pruned to what its Dockerfile reads and pinned to the Python and npm versions of the stored image, or, given `--image-ref ID=REF`, as that registry image pinned by digest with its context beside it, `image = { ref, dockerfile }`, which a run pulls and builds only when it can't) so it runs with no store; `agent-env bundle register` writes a folder into the configured stores under registry ids (`<prefix><name>`), each task and eval naming the versions it reads and each image it builds pushed, through a ledger in the target store so registering again writes only what changed. |
+| `bundle/` | A bundle is a folder of entities (`envs/`, `agents/`, `artifacts/`, `skills/`, `tasks/<name>.json`, `evals/<name>.toml`) that runs without a registry: `agent-env run` plans it, writes it into the `@local` namespace (reusing each version whose inputs haven't changed) and runs its evals or tasks; `agent-env bundle init` writes a stored task or eval as one, with every env, agent and artifact it uses in its own folder (each image as its build context, pruned to what its Dockerfile reads and pinned to the Python and npm versions of the stored image, or, given `--image-ref ID=REF`, as that registry image pinned by digest with its context beside it, `image = { ref, dockerfile }`, which a run pulls and builds only when it can't; `task_images` lists the images an import writes, downloading nothing, so an export can push them first) so it runs with no store; `agent-env bundle register` writes a folder into the configured stores under registry ids (`<prefix><name>`), each task and eval naming the versions it reads and each image it builds pushed, through a ledger in the target store so registering again writes only what changed. |
 | `eval/` | `Eval` (`eval`), a versioned list of task references, and its store. |
 | `plugins/` | Entry-point plugins (see Extension points): the group names, `inventory()`, which `agent-env plugin list/show/check` report, `load_failures` and `settings`. The top level is public; the underscored modules are internal. |
 | `explorer/` | Optional local web UI: `agent-env up`, binds loopback `:8234`. |

src/agent_env/bundle/importer.py

@@ -89,6 +89,18 @@ def import_task(task_id: str, dest: Path | str, *, version: int | None = None,
                                                                                          "version": task.version})
 
 
+def task_images(task_id: str, *, version: int | None = None,
+                include_scopes: Collection[FieldScope] = ()) -> tuple[DockerImageArtifact, ...]:
+    """The images ``import_task`` writes for the stored task ``task_id``, at ``version`` or its latest, each once in the
+    order it reaches them, found without downloading anything: what an export pushes to a registry before importing the
+    task with ``image_refs`` naming them there. Raises BundleError when the import would fail on what it reads."""
+    importer = _Importer(None, include_scopes)
+    importer.add_task(Task.get(task_id, version))
+    if importer.problems:
+        raise BundleError(importer.problems)
+    return tuple(importer.images_named.values())
+
+
 def import_eval(eval_id: str, dest: Path | str, *, version: int | None = None,
                 include_scopes: Collection[FieldScope] = (), image_refs: Mapping[str, str] | None = None,
                 progress: Callable[[str], None] | None = None) -> BundleImport:
@@ -158,13 +170,15 @@ def left_out(self, field: str, why: str) -> None:
 
 
 class _Importer:
-    def __init__(self, dest: Path, include_scopes: Collection[FieldScope] = (),
+    def __init__(self, dest: Path | None, include_scopes: Collection[FieldScope] = (),
                  progress: Callable[[str], None] | None = None, image_refs: Mapping[str, str] | None = None):
-        self.dest = Path(os.path.abspath(dest.expanduser()))  # so ``.`` has a name, and a sibling to stage in
+        # So ``.`` has a name, and a sibling to stage in; none when the import is only read, never written.
+        self.dest = Path(os.path.abspath(dest.expanduser())) if dest is not None else None
         self.include_scopes = frozenset({FieldScope.TASK, *include_scopes})
         self.image_refs = dict(image_refs or {})  # the registry image to write for an image, by its store id
         self.refs_used: set[str] = set()
         self.refs_pinned: dict[str, str] = {}  # each given ref as pinned to its digest, by image store id
+        self.images_named: dict[tuple[str, int], DockerImageArtifact] = {}  # each image a folder names, in order
         self.progress = progress or (lambda message: None)
         self.files: dict[str, str] = {}
         self.taken: dict[str, set[str]] = {kind.value: set() for kind in BundleKind}  # folded names, per folder
@@ -188,7 +202,8 @@ def __init__(self, dest: Path, include_scopes: Collection[FieldScope] = (),
         self.jobs: list[ImageJob] = []
         self.providers = Providers()
         self.installed: tuple[InstalledBundle, ...] | None = None
-        _check_destination(self.dest)
+        if self.dest is not None:
+            _check_destination(self.dest)
 
     def add_task(self, task: Task) -> str:
         """Write ``task`` under a name taken from its id, and what it references in folders of their own; return its
@@ -268,6 +283,7 @@ def image(self, out: _EntityOut, artifact: DockerImageArtifact, key: str) -> str
         """The toml value of an image a folder is built from: its stored context, pruned into the folder, else its
         registry ref, else a pinned store image. Given a registry reference for it, that reference, its tag pinned to
         the digest the registry serves now when it gives none, with the context kept beside it when it has one."""
+        self.images_named.setdefault((artifact.id, artifact.version), artifact)
         ref = self.image_refs.get(artifact.id)
         if ref is not None:
             self.refs_used.add(artifact.id)

tst/unit/bundle/importer_image_refs_test.py

@@ -12,13 +12,14 @@
 from agent_env.artifact.store import get_artifact_store
 from agent_env.bundle import BundleError
 from agent_env.bundle import importer as importer_module
-from agent_env.bundle.importer import import_task
+from agent_env.bundle.importer import import_task, task_images
 from agent_env.bundle.materialize import materialize
 from agent_env.bundle.resolve import build_step
 from agent_env.cli import cli
 from agent_env.config import get_config
 from agent_env.env import Env
 from agent_env.env.envs.mcp_server import MCPServerEnv
+from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 from agent_env.task import Task
 from tst.unit.bundle._support import context_image, local_builds, plan_of
 
@@ -187,3 +188,35 @@ def test_the_cli_takes_each_ref_as_id_equals_ref(home):
     assert written.exit_code == 0, written.output
     assert _toml(home / "t", "envs/crm/env.toml")["image"]["ref"] == REF
     assert refused.exit_code == 2 and "--image-ref takes ID=REF" in refused.output
+
+
+def test_the_images_a_task_writes_are_listed_once_in_order_without_downloading_and_take_refs(home, monkeypatch):
+    shared, registry_image = context_image(home, "crm-image"), get_artifact_store().put_document(DockerImageArtifact(
+        id="mail-image", description="d", image_name="ghcr.io/acme/mail:1"))
+    MCPServerEnv.put(id="crm", docker_image_artifact=shared, environment_name="crm")
+    MCPServerEnv.put(id="crm-eu", docker_image_artifact=shared, environment_name="crm")
+    MCPServerEnv.put(id="inbox", docker_image_artifact=registry_image, environment_name="inbox")
+    Task.put(id="t", steps=[build_step({**DEPLOY, "id": f"env-{env}", "env_id": env})
+                            for env in ("crm", "crm-eu", "inbox")])
+    downloads = []
+    real_download = LocalFilesystemObjectStore.download_to_file
+    monkeypatch.setattr(LocalFilesystemObjectStore, "download_to_file",
+                        lambda self, url, dest: downloads.append(url) or real_download(self, url, dest))
+
+    images = task_images("t")
+
+    assert ([(image.id, image.version) for image in images], downloads) == ([("crm-image", 1), ("mail-image", 1)], [])
+    refs = {image.id: f"ghcr.io/acme/delivery/{image.id}:1@sha256:" + "e" * 64 for image in images}
+    imported = import_task("t", home / "t", image_refs=refs)
+    assert [_toml(imported.root, f"envs/{env}/env.toml")["image"]["ref"] for env in ("crm", "crm-eu", "inbox")] == [
+        refs["crm-image"], refs["crm-image"], refs["mail-image"]]
+
+
+def test_listing_a_task_the_import_cant_write_fails_before_any_push_would_start(home):
+    _stored(get_artifact_store().put_document(DockerImageArtifact(
+        id="crm-image", description="d", image_name="crm-image:v1", tar_gz_s3_url="file:///nowhere/crm.tar.gz")))
+
+    with pytest.raises(BundleError) as caught:
+        task_images("t")
+
+    assert any("stored only as a docker save tarball" in problem for problem in caught.value.problems)