feat(bundle): write environment and environment_universe artifacts from their artifact folders

src/agent_env/artifact/artifacts/environment.py

@@ -3,24 +3,34 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Literal, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, Optional
 
 from pydantic import ConfigDict, Field, model_serializer
 
 from agent_env.artifact.artifact import Artifact
+from agent_env.artifact.artifacts.file import FileArtifact
 from agent_env.artifact.ref import ArtifactRef
+from agent_env.entity_refs import EntityRef
+from agent_env.store.ids import derive_id
 
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    from agent_env.artifact.artifacts.file import FileArtifact
+    from agent_env.bundle.authoring import AuthoringContext
 
 
 class EnvironmentArtifact(Artifact):
     """A FileArtifact wrapped with the name of the environment it seeds."""
 
     model_config = ConfigDict(populate_by_name=True)
 
+    # An artifact.toml names the environment and either the artifact to wrap (file) or nothing, to wrap the folder's
+    # one file, whose description it may set. The stored document's own names for these aren't taken.
+    toml_keys: ClassVar[dict[str, type]] = {"environment_name": str, "file": object, "description": str}
+    toml_refs: ClassVar[tuple[EntityRef, ...]] = (EntityRef.artifact("file", artifact_type="file"),)
+    toml_stored_names: ClassVar[dict[str, str]] = {"service_name": "environment_name", "file_artifact_id": "file",
+                                                   "file_artifact_ref": "file"}
+
     type: Literal["environment"] = "environment"
     environment_name: str = Field(alias="service_name", description="Name of the environment this artifact belongs to")
     file_artifact_ref: Optional[ArtifactRef] = Field(
@@ -70,9 +80,43 @@ def put(
         )
         return get_artifact_store().put_document(instance)
 
-    def get_file_artifact(self) -> FileArtifact:
-        from agent_env.artifact.artifacts.file import FileArtifact
+    @staticmethod
+    def derived_file_id(id: str) -> str:
+        """The id an environment's file is written under when it's written with the environment: ``<id>__file``."""
+        return derive_id(id, "file")
+
+    @classmethod
+    def from_toml(cls, data: dict, ctx: AuthoringContext) -> EnvironmentArtifact:
+        """Write the environment authored as ``data`` (its artifact.toml, with ``file`` resolved to a file artifact's
+        id) under ``ctx.id`` and return it: over the artifact ``file`` names, or over the folder's one file, written
+        as ``<id>__file`` with ``description`` defaulting to its name."""
+        fields = cls.accept_toml(data, ctx)
+        if "file" in fields:
+            file = ctx.artifact(fields["file"], FileArtifact)
+        else:
+            filename, path = ctx.file()
+            file = FileArtifact.put_attempt(cls.derived_file_id(ctx.id), file_path=str(path), filename=filename,
+                                            description=fields.get("description", filename))
+        return cls.put(ctx.id, environment_name=fields["environment_name"], file_artifact=file)
 
+    @classmethod
+    def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict:
+        """The keys of ``data``, an artifact.toml, an environment takes. Raises BundleError listing every problem."""
+        data, problems = ctx.renamed(data, cls.toml_stored_names)
+        fields, more = ctx.accepted(data, **cls.toml_keys)
+        problems += more
+        if "environment_name" not in data:
+            problems.append(ctx.config_problem("environment_name is required: the environment the file seeds"))
+        elif fields.get("environment_name") == "":
+            problems.append(ctx.config_problem("environment_name can't be empty"))
+        if "file" in fields and "description" in fields:
+            problems.append(ctx.config_problem("description describes the folder's own file, and file names an "
+                                               "artifact instead; drop one"))
+        if problems:
+            ctx.refuse(problems)
+        return fields
+
+    def get_file_artifact(self) -> FileArtifact:
         if self.file_artifact_ref is not None:
             return FileArtifact.get(
                 self.file_artifact_ref.id, version=self.file_artifact_ref.version

src/agent_env/artifact/artifacts/environment_universe.py

@@ -3,25 +3,42 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Literal, Optional
+from collections import Counter
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, Optional
 
 from pydantic import AliasChoices, ConfigDict, Field, model_serializer
 
+from agent_env.artifact.artifacts.environment import EnvironmentArtifact
+from agent_env.artifact.artifacts.file import FileArtifact
 from agent_env.artifact.ref import ArtifactRef
 from agent_env.artifact.universe import Universe
+from agent_env.entity_refs import EntityRef
+from agent_env.store.ids import derive_id
 
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    from agent_env.artifact.artifacts.file import FileArtifact
-    from agent_env.artifact.artifacts.environment import EnvironmentArtifact
+    from agent_env.bundle.authoring import AuthoringContext
 
 
 class EnvironmentUniverseArtifact(Universe):
     """A universe artifact that bundles EnvironmentArtifacts together."""
 
     model_config = ConfigDict(populate_by_name=True)
 
+    # An artifact.toml may name environment artifacts to add to the ones its folder holds (see from_toml). The stored
+    # document's own names aren't taken, nor is metadata, which a metadata/<key>/ folder holds.
+    toml_keys: ClassVar[dict[str, type]] = {"environment_artifacts": list}
+    toml_refs: ClassVar[tuple[EntityRef, ...]] = (
+        EntityRef.artifact("environment_artifacts[]", artifact_type="environment"),)
+    toml_stored_names: ClassVar[dict[str, str]] = {
+        "service_artifact_refs": "environment_artifacts", "environment_artifact_refs": "environment_artifacts",
+        "service_artifact_ids": "environment_artifacts", "metadata": "a metadata/<key>/ folder holding its file",
+        "metadata_refs": "a metadata/<key>/ folder holding its file"}
+    # What a universe's metadata is called: the folder `environment-universe get --output-dir` writes its files to, and
+    # the segment of the ids they're written under.
+    metadata_name: ClassVar[str] = "metadata"
+
     type: Literal["environment_universe"] = "environment_universe"
     # serialization_alias pins the emitted key: the attribute is renamed, the document is not.
     environment_artifact_refs: Optional[list[ArtifactRef]] = Field(
@@ -96,9 +113,55 @@ def put(
         )
         return get_artifact_store().put_document(instance)
 
-    def get_environment_artifacts(self) -> list[EnvironmentArtifact]:
-        from agent_env.artifact.artifacts.environment import EnvironmentArtifact
+    @staticmethod
+    def derived_environment_id(id: str, environment_name: str) -> str:
+        """The id an environment is written under when it's written with the universe: ``<id>__<environment_name>``."""
+        return derive_id(id, environment_name)
+
+    @classmethod
+    def derived_metadata_id(cls, id: str, key: str) -> str:
+        """The id a metadata file is written under when it's written with the universe: ``<id>__metadata__<key>``."""
+        return derive_id(derive_id(id, cls.metadata_name), key)
+
+    @classmethod
+    def from_toml(cls, data: dict, ctx: AuthoringContext) -> EnvironmentUniverseArtifact:
+        """Write the universe authored as ``data`` (its artifact.toml, with ``environment_artifacts`` resolved to
+        environment artifacts' ids) and its folder under ``ctx.id``, and return it. The folder is laid out as
+        ``environment-universe get --output-dir`` writes one: each ``<environment_name>/`` folder's file is written as
+        an environment of that name, ``<id>__<name>`` over ``<id>__<name>__file``, and each ``metadata/<key>/``
+        folder's file as ``<id>__metadata__<key>``. Its environments are the folders', in name order, then the ones
+        ``environment_artifacts`` names. Nothing is written when it has none, or two of one name."""
+        fields = cls.accept_toml(data, ctx)
+        layout = ctx.universe()
+        named = [ctx.artifact(ref, EnvironmentArtifact) for ref in fields.get("environment_artifacts", [])]
+        names = Counter([laid.name for laid in layout.environments] + [env.environment_name for env in named])
+        if not names:
+            ctx.refuse([ctx.config_problem("a universe needs an environment: a folder of its own, named after it and "
+                                           "holding its file, or one environment_artifacts names")])
+        if repeated := sorted(name for name, count in names.items() if count > 1):
+            ctx.refuse([ctx.config_problem(f"two of its environments are named {name!r}; a universe's environments "
+                                           "need names of their own") for name in repeated])
+        environments = []
+        for laid in layout.environments:
+            file = FileArtifact.put_attempt(laid.file_id, description=laid.filename, file_path=str(laid.path),
+                                            filename=laid.filename)
+            environments.append(EnvironmentArtifact.put(laid.environment_id, environment_name=laid.name,
+                                                        file_artifact=file))
+        metadata = {laid.name: FileArtifact.put_attempt(laid.file_id, description=laid.filename,
+                                                        file_path=str(laid.path), filename=laid.filename)
+                    for laid in layout.metadata}
+        return cls.put(ctx.id, environment_artifacts=environments + named, metadata=metadata or None)
+
+    @classmethod
+    def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict:
+        """The keys of ``data``, an artifact.toml, a universe takes. Raises BundleError listing every problem."""
+        data, problems = ctx.renamed(data, cls.toml_stored_names)
+        fields, more = ctx.accepted(data, **cls.toml_keys)
+        if problems + more:
+            ctx.refuse(problems + more)
+        return fields
 
+    def get_environment_artifacts(self) -> list[EnvironmentArtifact]:
         if self.environment_artifact_refs is not None:
             return [
                 EnvironmentArtifact.get(ref.id, version=ref.version)
@@ -138,8 +201,6 @@ def get_file_artifacts(self) -> dict[str, FileArtifact]:
         return out
 
     def get_metadata(self) -> dict[str, FileArtifact]:
-        from agent_env.artifact.artifacts.file import FileArtifact
-
         if self.metadata_refs:
             return {
                 k: FileArtifact.get(ref.id, version=ref.version)

src/agent_env/artifact/artifacts/file.py

@@ -120,15 +120,21 @@ def put_bytes(
 
     @classmethod
     def from_toml(cls, data: dict, ctx: AuthoringContext) -> Self:
-        """Write the folder's one file under a prefix of its own (``ArtifactStore.attempt_prefix``);
-        ``description`` defaults to the file's name."""
-        from agent_env.artifact.store import get_artifact_store
-
+        """Write the folder's one file (``put_attempt``); ``description`` defaults to the file's name."""
         fields = ctx.accept(data, **cls.toml_keys)
         filename, path = ctx.file()
-        prefix = get_artifact_store().attempt_prefix(cls.model_fields["type"].default, ctx.id)
-        return cls.put_at(ctx.id, description=fields.get("description", filename), file_path=str(path),
-                          object_url=prefix + filename, filename=filename)
+        return cls.put_attempt(ctx.id, description=fields.get("description", filename), file_path=str(path),
+                               filename=filename)
+
+    @classmethod
+    def put_attempt(cls, id: str, *, description: str, file_path: str, filename: str) -> Self:
+        """``put_at`` under a prefix of this attempt's own (``ArtifactStore.attempt_prefix``), so a write that
+        fails partway never blocks the next one."""
+        from agent_env.artifact.store import get_artifact_store
+
+        prefix = get_artifact_store().attempt_prefix(cls.model_fields["type"].default, id)
+        return cls.put_at(id, description=description, file_path=file_path, object_url=prefix + filename,
+                          filename=filename)
 
     def load(self) -> bytes:
         from agent_env.artifact.store import get_artifact_store