feat(bundle): agent-env bundle register writes a bundle into the configured stores

AGENTS.md

@@ -114,7 +114,7 @@ and deserialized through a registry.
 | `a2a_agent/` | The `A2AAgent` entity (`a2a_agent`), its stores and the validator steps. The protocol package provides the agent-side framework. |
 | `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, and this package is the only one that imports it. |
 | `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) so it runs with no store. |
+| `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) 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/artifact/artifacts/docker_image.py

@@ -22,9 +22,9 @@
 from agent_env.store.ids import fs_safe, image_repository, is_local_id, local_image_repository
 from agent_env.store.image_store.local_registry_image_store import LocalRegistryImageStore
 from agent_env.store.image_store.oci_registry_credentials import is_loopback_host, names_registry, registry_host_from_ref
-from agent_env.store.image_store.registry_api import pin_digest, pinned_from
+from agent_env.store.image_store.registry_api import NoSuchImageError, pin_digest, pinned_from
 from agent_env.utils.docker.context import BuildContext
-from agent_env.utils.docker.build import DEFAULT_BUILD_PLATFORM, normalize_platform
+from agent_env.utils.docker.build import DEFAULT_BUILD_PLATFORM, build_image, normalize_platform
 
 logger = logging.getLogger(__name__)
 
@@ -561,6 +561,33 @@ class GitHubBuildResult:
     docker_context_github_url: str | None
 
 
+def push_context_image(id: str, context_path: str, dockerfile_path: str,
+                       before: Callable[[], None] | None = None) -> tuple[str, str]:
+    """The ref and digest of the image of ``id``'s build context in ``id``'s image store, under the tag of its sources:
+    built for linux/amd64 with this machine's Docker and pushed, calling ``before`` first, unless the store serves the
+    tag already. Raises DockerBuildError when the build fails, RuntimeError when the push does, and ValueError when the
+    registry can't be read."""
+    from agent_env.config import get_config
+
+    context = _context_of(Path(context_path), dockerfile_path)
+    tag = "src-" + context.source_digest(DEFAULT_BUILD_PLATFORM).removeprefix("sha256:")
+    store = get_config().get_image_store_for(id)
+    # ECR, unlike the OCI spec, refuses two separators in a row, which a derived id (``<id>__env_image``) holds.
+    repository = image_repository(id).replace("__", "-")
+    ref = store.image_ref(repository, tag)
+    store.ensure_repository(repository)
+    try:
+        return ref, pin_digest(ref, store.auth(ref)).rpartition("@")[2]
+    except NoSuchImageError:
+        pass
+    if before is not None:
+        before()
+    local = f"{local_image_repository(id)}:{tag}"
+    build_image(Path(context_path) / context.dockerfile, Path(context_path), local, platform=DEFAULT_BUILD_PLATFORM)
+    _push_local_image(local, ref, store)
+    return ref, pin_digest(ref, store.auth(ref)).rpartition("@")[2]
+
+
 def _push_local_image(local_tag: str, image_ref: str, image_store) -> None:
     auth = image_store.auth(image_ref)
     if auth is not None:

src/agent_env/artifact/store.py

@@ -24,6 +24,7 @@
     UpdateSpec,
     VersionedEntityStore,
     VersionedEntityStoreCache,
+    expected_version,
 )
 from agent_env.store.ids import key_segment
 from agent_env.store.query import QueryBuilder, to_document_query
@@ -195,6 +196,9 @@ def put_document(self, artifact: _A) -> _A:
         if artifact.version > 0:
             versioned.insert_next(doc, artifact.version)
             return artifact.model_copy()
+        if (expected := expected_version(artifact.id)) is not None:
+            versioned.insert_next(doc, expected)
+            return artifact.model_copy(update={"version": expected})
         version = versioned.next_version(artifact.id)
         last_exc: Optional[DuplicateKeyError] = None
         for attempt in range(_MAX_VERSION_RETRIES):

src/agent_env/bundle/_secrets.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from agent_env.config.describe import is_secret_key
+from agent_env.config.loader import ENV_REF_PREFIX, SECRET_REF_PREFIX
 
 # The fields that map environment variable names to values, where a credential is most often set.
 _ENV_VAR_MAPS = ("env_vars", "default_env_vars")
@@ -58,3 +59,32 @@ def _masked(text: str) -> tuple[str, list[str]]:
         if count:
             found.append(what)
     return text, found
+
+
+def credentials_in(value: Any, path: str = "") -> list[tuple[str, str]]:
+    """Each secret ``value`` holds that a store would keep as written, by its path, and why: an env var whose name looks
+    like a secret's set to anything, any value holding a credential ``scrub`` masks, and an env var set to an ``env:``
+    or ``secret:`` reference, which nothing resolves in a step's env vars, so the run would get the reference itself."""
+    found = []
+    items = value.items() if isinstance(value, dict) else enumerate(value) if isinstance(value, list) else ()
+    for key, item in items:
+        at = f"{path}.{key}" if path and isinstance(key, str) else f"{path}[{key}]" if isinstance(key, int) else key
+        if isinstance(value, dict) and key in _ENV_VAR_MAPS and isinstance(item, dict):
+            for name, setting in item.items():
+                if not isinstance(setting, str) or not setting:
+                    continue
+                if setting.startswith((ENV_REF_PREFIX, SECRET_REF_PREFIX)):
+                    found.append((f"{at}.{name}", "it's a reference, and nothing resolves one in a step's env vars, so "
+                                                  "the run would get the reference itself"))
+                elif is_secret_key(name):
+                    found.append((f"{at}.{name}", "its name looks like a secret's and it's set, which a shared store "
+                                                  "shows to anyone who reads it"))
+                elif kinds := _masked(setting)[1]:
+                    found.append((f"{at}.{name}", f"it holds {kinds[0]}, which a shared store shows to anyone who "
+                                                  "reads it"))
+        elif isinstance(item, str):
+            if kinds := _masked(item)[1]:
+                found.append((at, f"it holds {', '.join(kinds)}, which a shared store shows to anyone who reads it"))
+        else:
+            found += credentials_in(item, at)
+    return found

src/agent_env/bundle/ledger.py

@@ -13,7 +13,7 @@
 import json
 import stat
 from collections.abc import Callable, Iterable, Iterator, Mapping
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from dataclasses import dataclass
 from datetime import date, datetime, time, timezone
 from pathlib import Path
@@ -27,7 +27,7 @@
 from agent_env.env.store import ENVS_COLLECTION
 from agent_env.eval.store import EVALS_COLLECTION
 from agent_env.store import Filter, Sort
-from agent_env.store.document_store import LocalSqliteDocumentStore
+from agent_env.store.document_store import DocumentStore, LocalSqliteDocumentStore, expecting_version
 from agent_env.store.local_state import holding_locks
 from agent_env.task.store import TASKS_COLLECTION
 
@@ -79,7 +79,17 @@ def next_version(self) -> int:
 class Ledger:
     """One plan's rows in the ``@local`` namespace's store."""
 
-    def __init__(self, store: LocalSqliteDocumentStore, plan: Plan):
+    collection = LEDGER_COLLECTION
+    # Whether a task or eval is written naming the versions of what it references, so is hashed with them.
+    pins_references = False
+    # Whether each write lands on exactly the version its check found next, so one that loses a race to another
+    # writer raises ConcurrentModificationError rather than add a version after the other's.
+    strict = False
+    # Whether each image the bundle builds is built here and pushed to the configured image store, for whatever reads the
+    # store, rather than written as its build context for each deploy to build.
+    pushes_images = False
+
+    def __init__(self, store: DocumentStore, plan: Plan):
         self._store = store
         self._plan = plan
 
@@ -130,10 +140,10 @@ def digest(self, write: Write, needs: Mapping[tuple[str, str], int]) -> Digest |
                 inputs["files"][key] = _file_sha256(path) + (_mode(path) if built else "")
         if (published := self._plan.published.get(write.id)) is not None:
             inputs["published"] = published.pinned
-        if write.kind not in (BundleKind.TASK, BundleKind.EVAL):
+        if self.pins_references or write.kind not in (BundleKind.TASK, BundleKind.EVAL):
             # A document records the versions of what it references (an env its images, a universe its
-            # environments), so one written anew means it must be written again. A task or eval names its
-            # references without a version.
+            # environments), so one written anew means it must be written again. A task or eval a run writes names
+            # its references without a version.
             for store, id in write.needs:
                 inputs["needs"][f"{store} {id}"] = str(needs[store, id])
             for (kind, id), version in unpinned_store_refs(self._plan, write).items():
@@ -146,36 +156,46 @@ def record(self, check: Check, write: Callable[[], int]) -> int:
         A pending row marks the attempt until it's done, with its version once ``write`` returns. One an
         interrupted run of this bundle left is kept by ``check`` when its version can be, and dropped here."""
         key = {"store": check.write.kind.store, "id": check.write.id, "bundle": self._bundle(check.write)}
-        self._store.ensure_index(LEDGER_COLLECTION, ["store", "id", "status"])
-        while self._store.delete(LEDGER_COLLECTION, Filter.of(**key, status="pending")):
+        self._store.ensure_index(self.collection, ["store", "id", "status"])
+        while self._store.delete(self.collection, Filter.of(**key, status="pending")):
             pass
         pending = {**key, "status": "pending", "scheme": SCHEME,
                    "digest": check.digest.value if check.digest else None,
-                   "inputs": check.digest.inputs if check.digest else None, "at": _now()}
-        self._store.insert(LEDGER_COLLECTION, pending)
-        version = write()
+                   "inputs": check.digest.inputs if check.digest else None, "at": _now(), **self.row_fields()}
+        self._store.insert(self.collection, pending)
+        try:
+            with expecting_version(check.write.id, check.next_version) if self.strict else nullcontext():
+                version = write()
+        except Exception:
+            # It wrote no version this run knows of, so the row marks nothing for the next run to keep.
+            self._store.delete(self.collection, Filter.of(**key, status="pending"))
+            raise
         # Stamped before the check below, which can take seconds, so a run interrupted during it leaves the version
         # this attempt wrote, and the next run keeps it rather than writing it again.
-        self._store.replace(LEDGER_COLLECTION, Filter.of(**key, status="pending"), {**pending, "version": version})
+        self._store.replace(self.collection, Filter.of(**key, status="pending"), {**pending, "version": version})
         done = {**pending, "status": "done", "version": version, "at": _now()}
         if check.digest is not None and self.digest(check.write, check.needs) != check.digest:
             # A file changed during the write, so what the version holds is unknown: the next run writes it again.
             done.update(digest=None, inputs=None)
-        self._store.replace(LEDGER_COLLECTION, Filter.of(**key, status="pending"), done, upsert=True)
+        self._store.replace(self.collection, Filter.of(**key, status="pending"), done, upsert=True)
         return version
 
     def adopt(self, check: Check) -> None:
         """Record the version an interrupted run wrote, which ``check`` keeps, as that run would have."""
         key = {"store": check.write.kind.store, "id": check.write.id, "bundle": self._bundle(check.write)}
         done = {**key, "status": "done", "scheme": SCHEME, "digest": check.digest.value, "inputs": check.digest.inputs,
-                "version": check.version, "at": _now()}
-        self._store.replace(LEDGER_COLLECTION, Filter.of(**key, status="pending"), done, upsert=True)
+                "version": check.version, "at": _now(), **self.row_fields()}
+        self._store.replace(self.collection, Filter.of(**key, status="pending"), done, upsert=True)
+
+    def row_fields(self) -> dict[str, Any]:
+        """What each row records beyond the write: nothing here."""
+        return {}
 
     def _orphan(self, write: Write, stored: int | None, digest: Digest) -> int | None:
         """The version an interrupted run of this bundle wrote and didn't record, when it's still the store's
         latest and was made from what ``write`` is now made from."""
-        pending = self._find(LEDGER_COLLECTION, Filter.of(store=write.kind.store, id=write.id,
-                                                          bundle=self._bundle(write), status="pending"))
+        pending = self._find(self.collection, Filter.of(store=write.kind.store, id=write.id,
+                                                        bundle=self._bundle(write), status="pending"))
         if pending is None or pending.get("version") is None or pending["version"] != stored:
             return None
         return stored if (pending["scheme"], pending["digest"]) == (SCHEME, digest.value) else None
@@ -185,15 +205,15 @@ def _bundle(self, write: Write) -> str:
         return self._plan.bundle_of(write.source.entry).id_root
 
     def _latest(self, store: str, id: str) -> dict | None:
-        return self._find(LEDGER_COLLECTION, Filter.of(store=store, id=id, status="done"))
+        return self._find(self.collection, Filter.of(store=store, id=id, status="done"))
 
     def _stored_version(self, write: Write) -> int | None:
         doc = self._find(_COLLECTIONS[write.kind.store], Filter.of(id=write.id))
         return doc["version"] if doc else None
 
     def _find(self, collection: str, filter: Filter) -> dict | None:
-        """The highest version matching ``filter``, read without creating the store."""
-        if not self._store.path.exists():
+        """The highest version matching ``filter``, read without creating a local store."""
+        if isinstance(self._store, LocalSqliteDocumentStore) and not self._store.path.exists():
             return None
         return self._store.find_one(collection, filter, sort=Sort.by("version"))