feat(computer): a run composes only the forms its sandbox boots

src/agent_env/bundle/materialize.py

@@ -315,13 +315,14 @@ def _write_agent(plan: Plan, write: Write) -> int:
 
 
 def _compose_reused(write: Write, version: int, composing: Composing, dry_run: bool) -> bool:
-    """Whether the computer env ``write`` reuses has pairs left to compose on ``composing``'s provider (``pending``:
-    failed, stopped, or built but unprobed though it runs their arch natively) or its computer to write again;
-    composed unless ``dry_run``, which says what it would compose."""
+    """Whether the computer env ``write`` reuses has pairs of ``composing``'s forms left to compose on its provider
+    (``pending``: never started, failed, stopped, or built but unprobed though it runs their arch natively) or its
+    computer to write again; composed unless ``dry_run``, which says what it would compose."""
     env = Env.get(write.id, version)
     if not isinstance(env, ComputerEnv) or env.computer is None:
         return False
-    pairs = composer.pending(env, composing.sandbox_provider)
+    provider = composing.provider()
+    pairs = composer.pending(env, provider, composing.forms(env, provider))
     if not pairs and composer.current(env):
         return False
     if not dry_run:

src/agent_env/bundle/run.py

@@ -31,6 +31,7 @@
 from agent_env.store.routing import namespace_routing, run_scope
 from agent_env.task.store import record_task_cancelled
 from agent_env.task.interrupts import Interrupts
+from agent_env.utils.progress import run_progress
 from agent_env.task.task import Task, recorded_scores
 from agent_env.task.teardown import TeardownReport, teardown_run
 from agent_env.task_step.context import TaskStepContext, sandbox_overrides
@@ -405,9 +406,9 @@ def _planned(root: Path | str, tasks: Sequence[str], evals: Sequence[str], id_ro
 
 
 def _composing(sandbox: str | None, emulate: bool = True) -> Composing:
-    """How a run composes its computer envs: on the provider ``sandbox`` names, and emulating unless ``emulate`` is
-    False. Raises ValueError when ``sandbox`` names no provider."""
-    return Composing(build_sandbox_provider(sandbox) if sandbox else None, emulate)
+    """How a run composes its computer envs: on the provider ``sandbox`` names, emulating unless ``emulate`` is False,
+    only the forms its deploys there boot. Raises ValueError when ``sandbox`` names no provider."""
+    return Composing.for_run(build_sandbox_provider(sandbox) if sandbox else None, emulate)
 
 
 def planned_tasks(root: Path | str, *, tasks: Sequence[str] = (), evals: Sequence[str] = (),
@@ -522,13 +523,14 @@ async def run_one(entry: BundleEntry, task: Task) -> TaskRun:
         error, cancelled = None, False
         try:
             try:
-                await task.run(
-                    on_step_start=lambda i, total, step, _: say(f"{tag} step {i + 1}/{total} {step.id} ({step.type})"),
-                    on_step_complete=lambda i, total, step, _, seconds: say(
-                        f"{tag} step {i + 1}/{total} {step.id} done in {seconds:.1f}s"),
-                    agent_model=model,
-                    context=context,
-                )
+                with run_progress(lambda line: say(f"{tag} {line}")):
+                    await task.run(
+                        on_step_start=lambda i, total, step, _: say(f"{tag} step {i + 1}/{total} {step.id} ({step.type})"),
+                        on_step_complete=lambda i, total, step, _, seconds: say(
+                            f"{tag} step {i + 1}/{total} {step.id} done in {seconds:.1f}s"),
+                        agent_model=model,
+                        context=context,
+                    )
             except Exception as e:
                 error = e
             except asyncio.CancelledError:

src/agent_env/computers/README.md

@@ -20,6 +20,16 @@ partition kept, so nothing is emulated or privileged. A disk boots where `/dev/k
 (`disk/linux/launcher/`): gate 3 probes it there, and elsewhere it stays `unverified` until a deploy's first boot with
 KVM probes it. A deploy boots it with `env_params = { form = "disk" }`.
 
+A put, `env build` and the release compose every form; a run composes only the forms its sandbox's deploys boot: on each
+link of a chain, the first form it boots natively, as a deploy picks, so local Docker composes containers, and a sandbox
+that boots only disks composes disks. A compose of a disk also composes the container it's made from, unless that's
+built. A deploy that boots a form no compose has started, a disk named by `env_params` included, composes it first,
+reusing what's built, as every build is reused by its key; deploys of one env at once compose it once, the others
+waiting for it, a deploy that finds it building included. In `agent-env run` it says why and each stage as it goes, and
+a deploy waiting for another's compose says what it waits for and since when. A local run doesn't compose a registered
+env's form at deploy where a configured store isn't the per-user one, since the run's fence refuses its record: the
+deploy refuses for want of a ready build and names `agent-env env build`.
+
 Its settings reach the computer as environment variables, `AGENTENV_SETTING_<NAME>`, read as it starts:
 
 | Setting | Default | What it does |

src/agent_env/config/runtime.py

@@ -724,16 +724,21 @@ def check_object_url(self, entity_id: str, object_url: str) -> None:
             routing.refuse_in_local_run(f"a write to {object_url!r}")
 
     def check_local_run_write(self, entity_id: str) -> None:
-        """Refuse writing bare-id ``entity_id`` while an ``@local`` task runs where a configured store
-        isn't the per-user one: the write would land in a shared store."""
+        """Refuse writing ``entity_id`` where ``local_run_refuses_write`` says so."""
         from agent_env.store import routing
-        from agent_env.store.ids import is_local_id
 
-        if not routing.in_local_run() or is_local_id(entity_id):
-            return
-        if any(self._namespace_stores(kind)[1] is not None for kind in ("document", "object", "image")):
+        if self.local_run_refuses_write(entity_id):
             routing.refuse_in_local_run(f"writing {entity_id!r}")
 
+    def local_run_refuses_write(self, entity_id: str) -> bool:
+        """Whether writing bare-id ``entity_id`` now is refused: an ``@local`` task runs where a configured store isn't
+        the per-user one, so the write would land in a shared store."""
+        from agent_env.store import routing
+        from agent_env.store.ids import is_local_id
+
+        return routing.in_local_run() and not is_local_id(entity_id) and any(
+            self._namespace_stores(kind)[1] is not None for kind in ("document", "object", "image"))
+
     def _configured_object_store(self) -> ObjectStore:
         if self._object_store is None:
             with self._object_store_lock:

src/agent_env/env/envs/computer/env.py

@@ -4,15 +4,19 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import posixpath
+from collections.abc import Callable
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, ClassVar, Optional, Self
 
 from agentenv_protocol import METHOD_ADD, RPC_PATH
 from agentenv_protocol import client as protocol_v1
 
 from agent_env.artifact.artifacts.computer import ComputerArtifact, SettingValue
 from agent_env.attribution import Attribution
+from agent_env.config import get_config
 from agent_env.entity.refs import EntityRef
 from agent_env.env.env import Env, LoadFileArtifactUniverseResult, gateway_url_of
 from agent_env.env.envs._deployment import (
@@ -29,6 +33,8 @@
 from agent_env.env.envs.computer.driver_env import ComputerDriverEnv
 from agent_env.env.gateway import GatewayMode
 from agent_env.env.store import get_env_store
+from agent_env.store.local_state import awaiting_lock
+from agent_env.utils.progress import current_progress
 from agent_env.providers.env_providers.env_computer_provider import end_recordings
 from agent_env.utils.paths import validate_relative_filename
 
@@ -38,6 +44,7 @@
     from agent_env.env.env import DeployedEnv
     from agent_env.env.envs.computer.app_env import ComputerAppEnv
     from agent_env.providers.env_providers.env_provider import EnvironmentProvider
+    from agent_env.providers.sandbox_providers.sandbox_provider import SandboxProvider
 
 logger = logging.getLogger(__name__)
 
@@ -167,10 +174,50 @@ def computer_values(self) -> dict[str, Any]:
         display = self.display
         return {**self.computer_settings, **({"resolution": f"{display['width']}x{display['height']}"} if display else {})}
 
+    def composed_arches(self) -> list[str]:
+        """The arches the env composes: those its base, driver and apps all run on."""
+        return gate.arches(self.computer, self.driver, self.composed_apps())
+
     def computer_artifact(self) -> ComputerArtifact:
         """The env's computer: its base composed with its driver, over the images built."""
         return ComputerArtifact.get(self.metadata["computer"]["id"], self.metadata["computer"]["version"])
 
+    async def compose_unstarted(self, sandbox_provider: SandboxProvider, form: Optional[str] = None) -> None:
+        """Compose first what a deploy on ``sandbox_provider`` boots that no compose started, once for deploys at once,
+        unless the run's fence would refuse the record (rules in ``computers/README.md``). A cancelled deploy holds the
+        lock until its compose's thread, which no cancel stops, ends, so the next deploy waits for what it builds."""
+        if get_config().local_run_refuses_write(self.stored_id):
+            return
+        if not composer.unstarted_or_building(self, sandbox_provider, form):
+            return
+        say, progress = _telling(self.id, current_progress())
+        awaited: dict[tuple[str, str], str | None] = {}
+
+        async def waiting(seconds: float) -> None:
+            self.metadata = (await asyncio.to_thread(Env.get, self.stored_id, self.version)).metadata
+            composed = self.metadata.get("composed", {})
+            awaited.update({(name, arch): composed[name][arch].get("started_at") for name, arch in
+                            composer.unstarted_or_building(self, sandbox_provider, form) if arch in composed.get(name, {})})
+            say(_waiting(awaited, seconds))
+
+        async with awaiting_lock(f"computer-compose:{self.stored_id}:{self.version}", waiting, composer.HEARTBEAT_S):
+            self.metadata = (await asyncio.to_thread(Env.get, self.stored_id, self.version)).metadata
+            if pairs := composer.unstarted(self, sandbox_provider, form):
+                them = "it" if len(pairs) == 1 else "them"
+                say(f"composing {_pairs(pairs)} before this deploy (no compose built {them} for "
+                    f"{' or '.join(type(link).__name__ for link in sandbox_provider.links)} yet); later deploys reuse {them}")
+                composing = asyncio.ensure_future(asyncio.to_thread(
+                    composer.compose, self, pairs, Composing(sandbox_provider, on_progress=progress)))
+                try:
+                    await asyncio.shield(composing)
+                except asyncio.CancelledError:
+                    await asyncio.wait({composing})
+                    raise
+            elif awaited:
+                composed = self.metadata["composed"]
+                say("reusing what another deploy composed: " + ", ".join(
+                    f"{name}/{arch} {composed[name][arch]['status']}" for name, arch in awaited))
+
     def deployable_builds(self, form: Optional[str] = None) -> list[tuple[str, str, dict]]:
         """Each build a deploy may boot, as (form, arch, record), in the base's order of forms and of arches; only
         ``form``'s when it's given: a ready one, or a disk no compose could probe (unverified), which its first boot on
@@ -213,7 +260,7 @@ def deploy_refusal(self, *, sandbox_type: str | None = None, env_params: Optiona
             return refusal
         params, _, problems = computer_params.checked(self, env_params)
         sandbox_provider = build_sandbox_provider(sandbox_type) if sandbox_type else get_env_sandbox_provider()
-        links, arches = sandbox_provider.links, gate.arches(self.computer, self.driver, self.composed_apps())
+        links, arches = sandbox_provider.links, self.composed_arches()
         forms = [params.form] if params.form else gate.forms(self.computer)
         if not any(boot.native_links(sandbox_provider, arch, form) for form in forms for arch in arches):
             problems.append(f"{', '.join(type(link).__name__ for link in links)} runs none of its arches "
@@ -362,3 +409,29 @@ def _listed_apps(refs: list, ctx: AuthoringContext) -> tuple[list[ComputerAppEnv
 
 def _not_an_app(env: Env) -> str | None:
     return f"{env.id} is a {env.type} env, not a {APP_TYPE} env" if env.type != APP_TYPE else None
+
+
+def _telling(env_id: str, progress: Optional[Callable[[str], None]]) -> tuple[
+        Callable[[str], None], Optional[Callable[[str], None]]]:
+    """What a deploy that composes says, to the log and the run's ``progress`` when there's one, after the env's id;
+    and the progress alone, for the compose, which logs its own lines."""
+    told = None if progress is None else (lambda line: progress(f"{env_id}: {line}"))
+
+    def say(line: str) -> None:
+        logger.info(f"{env_id}: {line}")
+        if told is not None:
+            told(line)
+
+    return say, told
+
+
+def _pairs(pairs: list[tuple[str, str]]) -> str:
+    return ", ".join(f"{name}/{arch}" for name, arch in pairs)
+
+
+def _waiting(awaited: dict[tuple[str, str], str | None], seconds: float) -> str:
+    """What a deploy waiting for another's compose of ``awaited``, by when each started, says after ``seconds``."""
+    since = min((at for at in awaited.values() if at), default=None)
+    building = f"building since {datetime.fromisoformat(since).strftime('%H:%M')} UTC" if since else "building"
+    so_far = f", {composer.took(seconds)} so far" if seconds >= 1 else ""
+    return f"waiting for another deploy composing {_pairs(list(awaited)) or 'it'} ({building}{so_far})"

src/agent_env/env/envs/computer/utils/boot.py

@@ -8,6 +8,7 @@
 import contextlib
 import json
 import secrets
+from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -56,6 +57,12 @@ def native_links(sandbox_provider: SandboxProvider, arch: str, form: str = "cont
         link.disk_image_platforms() if form == "disk" else link.native_platforms())]
 
 
+def first_native(link: SandboxProvider, pairs: Iterable[tuple[str, str]]) -> tuple[str, str] | None:
+    """The first of ``pairs``, forms and arches in the base's order, that ``link`` boots natively: the one a deploy on it
+    boots, and so the one a run composes for it."""
+    return next(((form, arch) for form, arch in pairs if native_links(link, arch, form)), None)
+
+
 def settings_environment(computer: ComputerArtifact, values: dict[str, Any], *, environment_name: str,
                          driver_settings: Optional[dict] = None) -> dict[str, str]:
     """The environment a boot of ``computer`` gives it: each deploy-phase setting as ``AGENTENV_SETTING_<NAME>``, from