feat(bundle): write any artifact type from its folder through its own hooks, and read a universe's as the inverse of get

src/agent_env/artifact/artifact.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, ClassVar, Optional, Self
 
 from pydantic import BaseModel, Field, field_validator
@@ -21,6 +22,14 @@ def _write_twin(data: dict[str, Any], legacy: str, neutral: str, value: Any) ->
         data[legacy] = data[neutral] = value
 
 
+@dataclass(frozen=True)
+class FolderChild:
+    """An artifact written from a bundle folder besides the one the folder is for."""
+
+    id: str
+    path: str  # where in the folder it's written from
+
+
 class Artifact(BaseModel):
     """Base class for all environment artifacts. Immutable - each modification creates a new version."""
 
@@ -29,6 +38,7 @@ class Artifact(BaseModel):
     type: str = Field(description="The type of Artifact")
 
     toml_refs: ClassVar[tuple[EntityRef, ...]] = ()
+    toml_keys: ClassVar[dict[str, type]] = {}  # what an artifact.toml may set, for a type without an accept_toml
 
     @field_validator("type", mode="before")
     @classmethod
@@ -74,6 +84,13 @@ def from_toml(cls, data: dict[str, Any], ctx: AuthoringContext) -> Self:
         fields = {key: value for key, value in data.items() if key not in ("id", "type", "version")}
         return cls.put(id=ctx.id, **fields)
 
+    @classmethod
+    def written_besides(cls, ctx: AuthoringContext) -> tuple[FolderChild, ...]:
+        """The artifacts ``from_toml`` writes from the entry's folder besides ``ctx.id``, which a bundle checks
+        before anything is written: none, unless a type writing others too overrides this to name them. Raises
+        BundleError when the folder can't be read."""
+        return ()
+
     @classmethod
     def query(cls) -> "ArtifactQuery":
         from agent_env.artifact.store import ArtifactQuery, get_artifact_store

src/agent_env/artifact/artifacts/environment.py

@@ -102,8 +102,23 @@ def from_toml(cls, data: dict, ctx: AuthoringContext) -> EnvironmentArtifact:
 
     @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)
+        """The keys of ``data``, an artifact.toml, an environment takes, and its folder: its one file, or only
+        artifact.toml when ``file`` names the artifact to wrap. Raises BundleError listing every problem."""
+        problems = []
+        if (files := ctx.try_files(empty_ok=True)) is not None:  # None when the plan's walk reports why
+            if any(key == "file" or cls.toml_stored_names.get(key) == "file" for key in data):
+                if files:
+                    problems.append(f"its file is the artifact file names, so the folder holds only artifact.toml, and "
+                                    f"it also has {', '.join(repr(name) for name in files)}; give the file an artifact "
+                                    "folder of its own, or leave out file to wrap the folder's one file")
+            elif not files:
+                problems.append(f"has no file for the environment to wrap ({', '.join(ctx.leavings)} don't count); put "
+                                "its one file here, or name the artifact to wrap with file")
+            elif problem := ctx.not_one_file(files, "name the artifact to wrap with file, or give each other file an "
+                                                    "artifact folder of its own"):
+                problems.append(problem)
+        data, renamed = ctx.renamed(data, cls.toml_stored_names)
+        problems += renamed
         fields, more = ctx.accepted(data, **cls.toml_keys)
         problems += more
         if "environment_name" not in data:

src/agent_env/artifact/artifacts/environment_universe.py

@@ -8,12 +8,20 @@
 
 from pydantic import AliasChoices, ConfigDict, Field, model_serializer
 
+from agent_env.artifact.artifact import FolderChild
 from agent_env.artifact.artifacts.environment import EnvironmentArtifact
+from agent_env.artifact.artifacts.environment_universe_folder import (
+    METADATA,
+    UniverseFolder,
+    UniverseFolderError,
+    environment_id,
+    metadata_id,
+    read_folder,
+)
 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__)
 
@@ -37,7 +45,7 @@ class EnvironmentUniverseArtifact(Universe):
         "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"
+    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.
@@ -117,28 +125,27 @@ def put(
     @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)
+        return environment_id(id, environment_name)
 
-    @classmethod
-    def derived_metadata_id(cls, id: str, key: str) -> str:
+    @staticmethod
+    def derived_metadata_id(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)
+        return metadata_id(id, 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."""
+        folder's file as ``<id>__metadata__<key>``: a name or key holding ``/`` is nested folders, and one an id can't
+        hold is written under a hash of it (``environment_universe_folder``). 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()
+        layout = cls._read_folder(ctx)
         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])
@@ -153,13 +160,51 @@ def from_toml(cls, data: dict, ctx: AuthoringContext) -> EnvironmentUniverseArti
                     for laid in layout.metadata}
         return cls.put(ctx.id, environment_artifacts=environments + named, metadata=metadata or None)
 
+    @classmethod
+    def written_besides(cls, ctx: AuthoringContext) -> tuple[FolderChild, ...]:
+        """What ``from_toml`` writes besides the universe, as ``read_folder`` names it: each environment its folder
+        holds, that environment's file, and each metadata file."""
+        layout = cls._read_folder(ctx)
+        return (*(FolderChild(id, laid.name) for laid in layout.environments for id in (laid.environment_id,
+                                                                                         laid.file_id)),
+                *(FolderChild(laid.file_id, f"{METADATA}/{laid.name}") for laid in layout.metadata))
+
+    @classmethod
+    def seeded_environments(cls, ctx: AuthoringContext) -> tuple[tuple[str, str], ...]:
+        """``(path in the folder, environment_name)`` of each environment the folder holds, which a bundle checks
+        against the servers of the env a task loads the universe into. Raises BundleError when the folder can't be
+        read."""
+        return tuple((laid.name, laid.name) for laid in cls._read_folder(ctx).environments)
+
+    @classmethod
+    def _read_folder(cls, ctx: AuthoringContext) -> UniverseFolder:
+        """The universe's folder, read as ``environment-universe get --output-dir`` writes one (``read_folder``).
+        Raises BundleError listing every problem."""
+        files, folders = ctx.listing()
+        try:
+            return read_folder(ctx.id, files, folders)
+        except UniverseFolderError as e:
+            ctx.refuse([], within=e.problems)
+
     @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."""
+        """The keys of ``data``, an artifact.toml, a universe takes, and its folder, read as ``environment-universe
+        get --output-dir`` writes one, which needs an environment of its own unless ``environment_artifacts`` names
+        one. 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)
+        problems, within = problems + more, []
+        if (listing := ctx.try_listing()) is not None:  # None when the plan's walk reports why
+            try:
+                layout = read_folder(ctx.id, *listing)
+            except UniverseFolderError as e:
+                within = e.problems
+            else:
+                if not layout.environments and not data.get("environment_artifacts"):
+                    problems.append("a universe needs an environment: a folder of its own, named after it and holding "
+                                    "its file, or one environment_artifacts names")
+        if problems or within:
+            ctx.refuse(problems, within=within)
         return fields
 
     def get_environment_artifacts(self) -> list[EnvironmentArtifact]:

src/agent_env/artifact/artifacts/environment_universe_folder.py

@@ -0,0 +1,156 @@
+"""The folder an environment universe is written to and read back from.
+
+``environment-universe get --output-dir`` writes each environment's file to ``<environment_name>/<file>`` and each
+metadata file to ``metadata/<key>/<file>`` (``environment_path``, ``metadata_path``), a name or key holding ``/`` in
+nested folders. ``read_folder`` is its exact inverse, which a bundle's environment_universe folder is written from,
+and ``environment_id`` and ``metadata_id`` name what's written from each folder.
+The format's own limits: an environment whose name starts with ``metadata/`` reads back as a metadata key, names are
+read in NFC, and names that differ only in case are one folder on a case-insensitive disk.
+"""
+
+from __future__ import annotations
+
+import functools
+import hashlib
+import unicodedata
+from collections.abc import Callable, Iterable, Mapping
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_env.artifact.artifacts.environment import EnvironmentArtifact
+from agent_env.store.ids import derive_id, validate_local_id
+
+# What a universe's metadata is called: the folder its files go in, and the segment of the ids they're written under.
+METADATA = "metadata"
+
+
+def environment_id(universe_id: str, environment_name: str) -> str:
+    """The id an environment is written under when it's written with its universe: ``<universe>__<name>``."""
+    return derive_id(universe_id, environment_name)
+
+
+def metadata_id(universe_id: str, key: str) -> str:
+    """The id a metadata file is written under when it's written with its universe: ``<universe>__metadata__<key>``."""
+    return derive_id(derive_id(universe_id, METADATA), key)
+
+
+def environment_path(folder: Path, environment_name: str, filename: str) -> Path:
+    """Where the file of a universe's environment goes in the universe's folder."""
+    return folder / environment_name / filename
+
+
+def metadata_path(folder: Path, key: str, filename: str) -> Path:
+    """Where a universe's metadata file goes in the universe's folder."""
+    return folder / METADATA / key / filename
+
+
+@dataclass(frozen=True)
+class FolderEntry:
+    """One file of a universe's folder, and the ids it's written under."""
+
+    name: str  # its environment's name, or its metadata key
+    filename: str
+    path: Path
+    file_id: str
+    environment_id: str | None  # None for a metadata file
+
+
+@dataclass(frozen=True)
+class UniverseFolder:
+    """A universe's folder, read: its environments, its metadata files, and every file it holds."""
+
+    environments: tuple[FolderEntry, ...]
+    metadata: tuple[FolderEntry, ...]
+    files: Mapping[str, Path]
+
+
+class UniverseFolderError(ValueError):
+    """Why a folder can't be read as a universe: ``problems``, each ``<path in the folder>: <why>``."""
+
+    def __init__(self, problems: list[str]):
+        super().__init__("; ".join(problems))
+        self.problems = problems
+
+
+def read_folder(universe_id: str, files: Mapping[str, Path],
+                folders: Iterable[tuple[str, ...]]) -> UniverseFolder:
+    """The environments and metadata files a universe's folder holds, from its files (``{path in the folder: path}``)
+    and its folders (the names on each path to one): each folder directly holding one file is an environment named by
+    its path from the universe's folder, or, under ``metadata/``, a metadata key named by its path from there. An
+    environment is written as ``<universe>__<name>`` over ``<universe>__<name>__file`` and a metadata file as
+    ``<universe>__metadata__<key>``, with a hash of the name in its place where an id can't hold it. Raises
+    UniverseFolderError listing every problem."""
+    folders = set(folders)
+    # A metadata/ folder spelled in another case is a typo, as for every reserved name in a bundle.
+    near = sorted(parts[0] for parts in folders
+                  if len(parts) == 1 and parts[0] != METADATA and _fold(parts[0]) == _fold(METADATA))
+    problems = [f"{name}: rename to {METADATA}; names are case-sensitive, and an environment named {name} goes in an "
+                "artifact folder of its own, named in environment_artifacts" for name in near]
+    # Every folder counts, so one left without its file is refused rather than dropped.
+    held: dict[tuple[str, ...], list[str]] = {parts: [] for parts in folders
+                                              if parts[0] not in near and parts != (METADATA,)}
+    for key in files:
+        parts = tuple(key.split("/"))
+        if parts[0] in near:
+            continue
+        if len(parts) == 1:
+            problems.append(f"{key}: a universe folder holds only artifact.toml and a folder for each environment; "
+                            "move it into <environment_name>/, or give it an artifact folder of its own")
+        elif parts[:-1] == (METADATA,):
+            problems.append(f"{key}: {METADATA}/ holds a folder for each key, with that key's one file in it "
+                            f"({METADATA}/<key>/<file>); an environment named {METADATA} goes in an artifact folder "
+                            "of its own, named in environment_artifacts")
+        else:
+            held.setdefault(parts[:-1], []).append(key)
+    environments, metadata_files = [], []
+    # A folder holding only folders is part of their names.
+    holders = {parts[:depth] for parts in held for depth in range(1, len(parts))}
+    for parts, keys in sorted(held.items(), key=lambda item: "/".join(item[0])):
+        folder = "/".join(parts)
+        if not keys and parts in holders:
+            continue
+        if len(keys) != 1:
+            listed = f" ({', '.join(repr(key.rsplit('/', 1)[-1]) for key in keys)})" if keys else ""
+            problems.append(f"{folder}: holds one file, and has {len(keys)}{listed}")
+            continue
+        metadata = parts[0] == METADATA
+        name = "/".join(parts[1:] if metadata else parts)
+        if metadata:
+            env_id, file_id = None, _child_id(functools.partial(metadata_id, universe_id), name, environment=False)
+        else:
+            env_id = _child_id(functools.partial(environment_id, universe_id), name, environment=True)
+            file_id = EnvironmentArtifact.derived_file_id(env_id)
+        try:
+            for id in (env_id, file_id):
+                if id is not None:
+                    validate_local_id(id)
+        except ValueError as e:
+            problems.append(f"{folder}: can't name the artifacts written from it ({e}); give the universe a shorter "
+                            "name")
+            continue
+        entry = FolderEntry(name, keys[0].rsplit("/", 1)[-1], files[keys[0]], file_id, env_id)
+        (metadata_files if metadata else environments).append(entry)
+    if problems:
+        raise UniverseFolderError(problems)
+    return UniverseFolder(tuple(environments), tuple(metadata_files), files)
+
+
+def _child_id(derive: Callable[[str], str], name: str, *, environment: bool) -> str:
+    """The id a universe's environment or metadata file named ``name`` is written under, as ``derive``, the
+    universe's own naming, derives it from a name: from ``name``, or, for a name holding ``__``, which could clash
+    with another's ids, or one an id can't hold (with an environment's file id, ``<id>__file``), from
+    ``sha256__<16 hex of its hash>``, which holds ``__``, so no name's own id can be."""
+    if "__" not in name:
+        id = derive(name)
+        try:
+            for derived in (id, EnvironmentArtifact.derived_file_id(id)) if environment else (id,):
+                validate_local_id(derived)
+            return id
+        except ValueError:
+            pass
+    return derive(f"sha256__{hashlib.sha256(name.encode('utf-8')).hexdigest()[:16]}")
+
+
+def _fold(name: str) -> str:
+    """The caseless form names are compared in: casefold output isn't always NFC, so normalize both sides."""
+    return unicodedata.normalize("NFC", unicodedata.normalize("NFC", name).casefold())

src/agent_env/artifact/artifacts/file.py

@@ -131,11 +131,22 @@ def put_bytes(
     @classmethod
     def from_toml(cls, data: dict, ctx: AuthoringContext) -> Self:
         """Write the folder's one file (``put_attempt``); ``description`` defaults to the file's name."""
-        fields = ctx.accept(data, **cls.toml_keys)
+        fields = cls.accept_toml(data, ctx)
         filename, path = ctx.file()
         return cls.put_attempt(ctx.id, description=fields.get("description", filename), file_path=str(path),
                                filename=filename)
 
+    @classmethod
+    def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict:
+        """The keys of ``data``, an artifact.toml, a file takes, and the folder's one file. Raises BundleError listing
+        every problem."""
+        fields, problems = ctx.accepted(data, **cls.toml_keys)
+        if (files := ctx.try_files()) is not None and (problem := ctx.not_one_file(files)) is not None:
+            problems.append(problem)
+        if problems:
+            ctx.refuse(problems)
+        return fields
+
     @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