feat(bundle): use an installed bundle's envs, agents and artifacts from any bundle, the default agent included

AGENTS.md

@@ -81,7 +81,8 @@ or one of them only where it is used.
   what steps staged in `/tmp` under that id. Those processes are found through `/proc`, so on macOS
   they keep running; files the install wrote elsewhere (`/opt`, packages) stay.
 - `[agents] default_a2a_agent_id`: the agent a `deploy_agent` step without an id deploys; built-in
-  `a2a-default`, overridable by `configure(default_a2a_agent_id=...)`.
+  `a2a-default`, overridable by `configure(default_a2a_agent_id=...)`. Any step's agent field left unset
+  falls back to it (`entity.refs.default_agent_refs`), so a bundle run plans the default for it.
 - Secrets never live in the file: use `secret:KEY` or `env:NAME` references, resolved through
   `[stores.secret]`.
 - agent-env has no stage concept: "dev" versus "prod" is which file `AGENT_ENV_CONFIG` names. A
@@ -134,7 +135,10 @@ Out-of-tree code plugs in either through `config.toml`, as a `module:Class` poin
 - Bundles install through `agent_env.bundles` entry points (`<name> = "<package>"`, the folder at
   `<package>/<name>/`), with ids rooted at `@local/<dist>/<name>`. A shared name is not left out:
   a bundle runs by its bare name unless another distribution installs that name too, and then by
-  `<dist>/<name>`, except that agent-env's own bundles keep their bare names.
+  `<dist>/<name>`, except that agent-env's own bundles keep their bare names. Another bundle uses an
+  installed bundle's env, agent or artifact by its id, or as the configured default agent: the planner
+  pulls it into the run as one of the bundle's own (`bundle/plan.py`), and its ledger row is keyed by the
+  installed bundle, so every bundle shares one write of it. Such an id takes no version pin.
 - Environment providers have no table: an installed package registers one as an `agent_env.env_providers`
   entry point named its `type`, which records carry as `env_provider_type`.
 - `[plugins.<distribution name>]` is reserved for an installed plugin's own settings, read with

src/agent_env/a2a_agent/a2a_agent.py

@@ -64,6 +64,15 @@
 _DIND_PORT = 2375
 
 
+def unwritten_local_agent(agent_id: str, field: str) -> str:
+    """Why a step can't deploy ``agent_id``, an ``@local`` agent the store doesn't hold, which its ``field`` names or,
+    left unset, the configured default does: an ``agent-env run`` of a bundle writes one, and a step reads only the
+    store."""
+    return (f"there is no agent {agent_id!r} in the store: an @local agent is written by `agent-env run` of the bundle "
+            f"that defines it or uses it, so run the task through that bundle, or point {field} or [agents] "
+            "default_a2a_agent_id at an agent the store holds")
+
+
 @dataclass
 class DeployedA2AAgent:
     agent_id: str

src/agent_env/artifact/artifacts/docker_image.py

@@ -64,6 +64,9 @@ def _git_clone_commands(owner: str, repo: str, ref: str | None, token: str | Non
         f"trap 'rm -f /tmp/git-askpass.sh' EXIT; GIT_ASKPASS=/tmp/git-askpass.sh git clone --depth 1 {ref_flag} {clone_url} /tmp/repo",
     ]
 
+# How much of a build context's source digest its image's tag holds.
+_SOURCES_TAG_CHARS = 16
+
 class DockerImageArtifact(Artifact):
     """A Docker image artifact: a tar.gz of the image in the object store; without one, a registry reference a
     sandbox pulls; or, with neither, only its build context."""
@@ -235,38 +238,42 @@ def put_ref(cls, id: str, *, description: str, image_name: str,
 
     @classmethod
     def put_context(cls, id: str, *, description: str, context_path: str, dockerfile_path: str,
-                    platform: str = DEFAULT_BUILD_PLATFORM) -> "DockerImageArtifact":
-        """Register an image by its build context alone, to be built where it runs: the context is uploaded and
-        nothing is built, pushed or saved, so no Docker is needed here. Its ``image_name``,
-        ``local/<slug>-<hash>:v<version>``, names no registry: it's the tag the build gives the image.
-        ``source_digest`` is equal for equal sources, so one build can serve both."""
+                    platform: str | None = DEFAULT_BUILD_PLATFORM) -> "DockerImageArtifact":
+        """Register an image by its build context alone, to be built where it runs, for ``platform``, or for each
+        sandbox's own when None: the context is uploaded and nothing is built, pushed or saved, so no Docker is needed
+        here. Its ``image_name``, ``local/<slug>-<hash>:<sources>``, names no registry: it's the tag the build gives the
+        image, the same for the same sources, so a build of one version serves another. ``source_digest`` is equal for
+        equal sources."""
         from agent_env.artifact.store import get_artifact_store
         from agent_env.config import get_config
 
-        platform = normalize_platform(platform)
-        root = Path(context_path)
-        dockerfile = _resolve_dockerfile(root, dockerfile_path)
-        if not dockerfile.is_file():
-            raise ValueError(f"no Dockerfile at {dockerfile}")
-        context = BuildContext.of(root, dockerfile)
-        if context.dockerfile is None:
-            raise ValueError(f"{dockerfile} is outside the build context {root}; the image is built from the context "
-                             "alone, so the Dockerfile must be in it")
+        platform = normalize_platform(platform) if platform else None
+        context = _context_of(Path(context_path), dockerfile_path)
         store = get_artifact_store()
         version = store.next_version(id)
         prefix = store.attempt_prefix("docker_image", id)
         build_context_object_url = _upload_context(get_config().get_object_store_to_write(prefix, id), prefix, context)
+        source_digest = context.source_digest(platform)
         return store.put_document(cls(
             id=id,
             version=version,
             description=description,
-            image_name=f"{local_image_repository(id)}:v{version}",
+            image_name=_context_image_name(id, source_digest),
             build_context_object_url=build_context_object_url,
             dockerfile_path=context.dockerfile,
             platform=platform,
-            source_digest=context.source_digest(platform),
+            source_digest=source_digest,
         ))
 
+    @staticmethod
+    def context_identity(id: str, context_path: str, dockerfile_path: str,
+                         platform: str | None = DEFAULT_BUILD_PLATFORM) -> tuple[str, str]:
+        """The ``source_digest`` and ``image_name`` ``put_context`` gives ``id`` for this build context, found without
+        writing anything."""
+        platform = normalize_platform(platform) if platform else None
+        source_digest = _context_of(Path(context_path), dockerfile_path).source_digest(platform)
+        return source_digest, _context_image_name(id, source_digest)
+
     def load(self) -> bytes:
         from agent_env.artifact.store import get_artifact_store
         if not self.tar_gz_object_url:
@@ -599,6 +606,22 @@ def _resolve_dockerfile(context_dir: Path, dockerfile_path: str | None) -> Path
     return dockerfile if dockerfile.is_absolute() or dockerfile.exists() else context_dir / dockerfile
 
 
+def _context_of(root: Path, dockerfile_path: str) -> BuildContext:
+    dockerfile = _resolve_dockerfile(root, dockerfile_path)
+    if not dockerfile.is_file():
+        raise ValueError(f"no Dockerfile at {dockerfile}")
+    context = BuildContext.of(root, dockerfile)
+    if context.dockerfile is None:
+        raise ValueError(f"{dockerfile} is outside the build context {root}; the image is built from the context "
+                         "alone, so the Dockerfile must be in it")
+    return context
+
+
+def _context_image_name(id: str, source_digest: str) -> str:
+    """The tag a build gives a context-only image: the same for the same sources."""
+    return f"{local_image_repository(id)}:{source_digest.removeprefix('sha256:')[:_SOURCES_TAG_CHARS]}"
+
+
 def _upload_context(objects, prefix: str, context: BuildContext) -> str:
     """Write ``context`` under ``prefix`` as a tar.gz and return its object URL."""
     with tempfile.TemporaryDirectory(prefix="agent-env-context-") as tmp:

src/agent_env/bundle/importer.py

@@ -19,11 +19,12 @@
 from agent_env.bundle import _toml
 from agent_env.bundle._fs import fold, os_reason, show
 from agent_env.bundle._secrets import scrub
+from agent_env.bundle.installed import InstalledBundle, installed_bundles
 from agent_env.bundle.parse import BundleError, parse_bundle, root_problems, segment_problem
 from agent_env.bundle.plan import STORE_GETTERS, check_bundle
 from agent_env.bundle.resolve import resolve_bundle
 from agent_env.config import get_config
-from agent_env.entity.refs import EntityKind, RefRole, RefSite, ref_sites
+from agent_env.entity.refs import EntityKind, EntityRef, RefRole, RefSite, default_agent_refs, ref_sites
 from agent_env.env import registry as env_registry
 from agent_env.env.registry import get_env_registry
 from agent_env.eval import Eval
@@ -34,10 +35,7 @@
 from agent_env.task import Task
 from agent_env.task_step import registry as task_step_registry
 from agent_env.task_step.registry import get_task_step_registry
-from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
 from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep
-from agent_env.task_step.task_steps.install_agent import InstallAgentTaskStep
-from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep
 
 # How long a name taken from an id may be, in bytes; ids are capped far above it, and a longer name helps nobody.
 _MAX_NAME_BYTES = 200
@@ -183,6 +181,7 @@ def __init__(self, dest: Path, include_scopes: Collection[FieldScope] = ()):
         self.read: dict[tuple[EntityKind, str, int | None], int | str] = {}
         self.types: dict[tuple[EntityKind, str, int | None], str] = {}  # each entity read, by its read key
         self.origins: dict[tuple[str, str], str] | None = None
+        self.installed: tuple[InstalledBundle, ...] | None = None
         _check_destination(self.dest)
 
     def add_task(self, task: Task) -> str:
@@ -214,7 +213,7 @@ def add_task(self, task: Task) -> str:
                 if _key(site) not in outputs:  # the run writes an output again, so it takes no pin
                     self._pin(site, f"{where}: {site.path}", deployed)
                     self._need_type(site, f"{where}: {site.path}")
-            self._default_agents(step, where)
+            self._default_agents(step, cls.entity_refs or (), where)
         self.files[path] = json.dumps(steps, indent=2, ensure_ascii=False, allow_nan=False) + "\n"
         self.tasks.append(ImportedTask(path, task.id, task.version, len(steps)))
         return name
@@ -242,7 +241,7 @@ def write(self, description: str) -> BundleImport:
                     (staging / path).write_text(text, encoding="utf-8")
             except OSError as e:
                 raise BundleError([f"{show(self.dest.parent)}: can't write the import there: {os_reason(e)}"]) from None
-            check_bundle(resolve_bundle(parse_bundle(staging)))
+            check_bundle(resolve_bundle(parse_bundle(staging)), installed=installed_bundles)
             self._move(staging)
         finally:
             shutil.rmtree(staging, ignore_errors=True)
@@ -267,6 +266,9 @@ def _move(self, staging: Path) -> None:
 
     def _pin(self, site: RefSite, where: str, deployed: set[str]) -> None:
         kind, entity_id = site.ref.kind, site.value
+        if (bundle := self._installed_for(kind, entity_id)) is not None:
+            self._from_installed(site, where, bundle)
+            return
         if site.version is not None or site.version_key is not None:
             read = self._read(kind, entity_id, site.version)
             if isinstance(read, str):
@@ -282,6 +284,23 @@ def _pin(self, site: RefSite, where: str, deployed: set[str]) -> None:
             self.floating.append(Floating(where, kind, entity_id, "the field takes no version and no step of the task "
                                                                   "deploys it, so a run reads the latest"))
 
+    def _installed_for(self, kind: EntityKind, entity_id: str) -> InstalledBundle | None:
+        """The installed bundle whose ids ``entity_id``, an env, agent or artifact, is under: a run writes it from
+        there."""
+        if kind is EntityKind.TASK or not entity_id.startswith(LOCAL_PREFIX):
+            return None
+        if self.installed is None:
+            self.installed = installed_bundles()
+        return next((bundle for bundle in self.installed if entity_id.startswith(f"{bundle.id_root}/")), None)
+
+    def _from_installed(self, site: RefSite, where: str, bundle: InstalledBundle) -> None:
+        """An installed bundle's entity has no fixed version, so it isn't pinned, and a pin the store held is dropped."""
+        why = f"it comes with the installed bundle {bundle.qualified}, which a run writes it from"
+        if site.version is not None and site.version_key is not None:
+            site.owner[site.version_key] = None
+            why += f"; the stored pin to v{site.version} is dropped, as such an entity has no fixed version"
+        self.floating.append(Floating(where, site.ref.kind, site.value, why))
+
     def _read(self, kind: EntityKind, entity_id: str, version: int | None = None) -> int | str:
         """The version of ``entity_id`` the store holds, at ``version`` or its latest, or why it can't be read."""
         if (kind, entity_id, version) not in self.read:
@@ -325,17 +344,13 @@ def _need(self, where: str, what: str, group: str, name: str, cls: type | None)
         by = distribution if registered and _names(value, cls) else "configuration"
         self.needs.append(Need(where, what, name, by))
 
-    def _default_agents(self, step: dict, where: str) -> None:
-        """A step that names no agent deploys the configured default, which no field pins."""
-        if step["type"] in (DeployAgentTaskStep.type, InstallAgentTaskStep.type) and step.get("a2a_agent_id") is None:
-            field, what = "a2a_agent_id", "names no agent"
-        elif (step["type"] == RubricsVerifierTaskStep.type and step.get("use_agent_judge")
-              and step.get("agent_name") is None and step.get("judge_a2a_agent_id") is None):
-            field, what = "judge_a2a_agent_id", "names no judge agent"
-        else:
-            return
-        self.floating.append(Floating(f"{where}: {field}", EntityKind.AGENT, get_config().get_default_a2a_agent_id(),
-                                      f"the step {what}, so a run deploys the configured default"))
+    def _default_agents(self, step: dict, refs: tuple[EntityRef, ...], where: str) -> None:
+        """A step that leaves an agent field unset falls back to the configured default, which no field pins."""
+        for ref in default_agent_refs(refs, step):
+            self.floating.append(Floating(f"{where}: {ref.path}", EntityKind.AGENT,
+                                          get_config().get_default_a2a_agent_id(),
+                                          f"the step leaves {ref.path} unset, so it falls back to the configured "
+                                          "default"))
 
     def _name(self, kind: str, entity_id: str) -> str:
         """A name in the ``kind`` folder: the id's last segment made valid, numbered when taken, caselessly."""

src/agent_env/bundle/installed.py

@@ -102,7 +102,7 @@ def checked(bundle: InstalledBundle, *, load: bool = True) -> Bundle:
     store. Raises BundleError."""
     parsed = parse_bundle(bundle.root, id_root=bundle.id_root)
     if load:
-        check_bundle(resolve_bundle(parsed))
+        check_bundle(resolve_bundle(parsed), installed=installed_bundles)
     return parsed