feat(bundle): read an environment_universe folder as the inverse of environment-universe get, whatever its names

src/agent_env/artifact/artifacts/environment_universe.py

@@ -130,8 +130,10 @@ def from_toml(cls, data: dict, ctx: AuthoringContext) -> EnvironmentUniverseArti
         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()
         named = [ctx.artifact(ref, EnvironmentArtifact) for ref in fields.get("environment_artifacts", [])]

src/agent_env/artifact/artifacts/environment_universe_folder.py

@@ -0,0 +1,147 @@
+"""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.
+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.artifact.artifacts.environment_universe import EnvironmentUniverseArtifact
+from agent_env.store.ids import validate_local_id
+
+METADATA = EnvironmentUniverseArtifact.metadata_name
+
+
+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:
+            derive = functools.partial(EnvironmentUniverseArtifact.derived_metadata_id, universe_id)
+            environment_id, file_id = None, _child_id(derive, name, environment=False)
+        else:
+            derive = functools.partial(EnvironmentUniverseArtifact.derived_environment_id, universe_id)
+            environment_id = _child_id(derive, name, environment=True)
+            file_id = EnvironmentArtifact.derived_file_id(environment_id)
+        try:
+            for id in (environment_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, environment_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/bundle/authoring.py

@@ -12,17 +12,16 @@
 
 from agent_env.artifact.artifact import Artifact
 from agent_env.artifact.artifacts.environment import EnvironmentArtifact
-from agent_env.artifact.artifacts.environment_universe import EnvironmentUniverseArtifact
+from agent_env.artifact.artifacts.environment_universe_folder import UniverseFolder, UniverseFolderError, read_folder
 from agent_env.entity_refs import EntityKind, parse_toml_ref
 from agent_env.env.env import Env
 from agent_env.env.envs._deployment import provider_refusal
 from agent_env.providers.env_providers.constants import GATEWAY_SERVICE_NAMES
 from agent_env.providers.env_providers.env_gateway_provider import EnvironmentGatewayProvider
 from agent_env.providers.env_providers.env_provider import _env_provider_class
-from agent_env.store.ids import validate_local_id
 from agent_env.utils.card_naming import card_names_in_files
 
-from ._fs import fold, os_reason, relative, show, with_article
+from ._fs import os_reason, relative, show, with_article
 from .parse import CONFIG_FILES, LEAVINGS, NAMED_BY, Bundle, BundleEntry, BundleError
 
 A = TypeVar("A", bound=Artifact)
@@ -72,7 +71,7 @@ def try_files(self) -> dict[str, Path] | None:
     def file(self) -> tuple[str, Path]:
         return entry_file(self.bundle, self.entry)
 
-    def universe(self) -> UniverseLayout:
+    def universe(self) -> UniverseFolder:
         return universe_layout(self.bundle, self.entry)
 
     def renamed(self, data: dict, names: Mapping[str, str]) -> tuple[dict, list[str]]:
@@ -284,112 +283,16 @@ def environment_files(bundle: Bundle, entry: BundleEntry) -> dict[str, Path]:
     return {}
 
 
-# The folder of a universe's metadata files, as `environment-universe get --output-dir` writes them.
-UNIVERSE_METADATA = EnvironmentUniverseArtifact.metadata_name
-
-
-@dataclass(frozen=True)
-class LaidOut:
-    """One file of a universe 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 UniverseLayout:
-    """An environment_universe folder, laid out as ``environment-universe get --output-dir`` writes one: each
-    ``<environment_name>/`` folder holds one environment's file, and each ``metadata/<key>/`` folder one metadata
-    file. ``files`` lists them all, as ``entry_files`` does."""
-
-    environments: tuple[LaidOut, ...]
-    metadata: tuple[LaidOut, ...]
-    files: dict[str, Path]
-
-
-def universe_layout(bundle: Bundle, entry: BundleEntry) -> UniverseLayout:
-    """The environments and metadata files an environment_universe's folder holds, each with the ids it's
-    written under: ``<universe>__<name>`` over ``<universe>__<name>__file`` for an environment, and
-    ``<universe>__metadata__<key>`` for a metadata file. Raises BundleError listing every problem."""
-    files = entry_files(bundle, entry, empty_ok=True)
-    where = relative(bundle.root, entry.path)
-    top = _folders(entry.path)
-    # A metadata/ folder spelled in another case is a typo, as for every reserved name in a bundle.
-    near = sorted(name for name in top if name != UNIVERSE_METADATA and fold(name) == fold(UNIVERSE_METADATA))
-    problems = [f"{where}/{name}: rename to {UNIVERSE_METADATA}; names are case-sensitive, and an environment named "
-                f"{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.
-    folders: dict[tuple[bool, str], list[str]] = {(False, name): [] for name in top
-                                                  if name != UNIVERSE_METADATA and name not in near}
-    folders.update({(True, name): [] for name in _folders(entry.path / UNIVERSE_METADATA)})
-    for key in files:
-        parts = key.split("/")
-        if parts[0] in near:
-            continue
-        metadata = parts[0] == UNIVERSE_METADATA
-        if len(parts) == (3 if metadata else 2):
-            folders.setdefault((metadata, parts[-2]), []).append(key)
-        elif len(parts) == 1:
-            problems.append(f"{where}/{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 metadata and len(parts) == 2:
-            problems.append(f"{where}/{key}: {UNIVERSE_METADATA}/ holds a folder for each key, with that key's one "
-                            f"file in it ({UNIVERSE_METADATA}/<key>/<file>); an environment named "
-                            f"{UNIVERSE_METADATA} goes in an artifact folder of its own, named in "
-                            "environment_artifacts")
-        else:
-            folder = "/".join(parts[:2]) if metadata else parts[0]
-            what = "a metadata key's" if metadata else "an environment's"
-            problems.append(f"{where}/{key}: {what} folder holds its one file directly, {folder}/<file>, with no "
-                            "folders inside")
-    environments, metadata_files = [], []
-    for (metadata, name), keys in sorted(folders.items()):
-        folder = f"{where}/{UNIVERSE_METADATA}/{name}" if metadata else f"{where}/{name}"
-        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
-        # Renaming an environment's folder renames the environment, so say how to keep its name.
-        rename = "rename the folder" if metadata else (
-            f"rename the folder, which renames the environment, or keep the name by giving the environment an "
-            f'artifact folder of its own, with environment_name = "{name}", named in environment_artifacts')
-        if name != name.strip():
-            problems.append(f"{folder}: a name that opens or closes with a space can't name the artifacts written "
-                            "from it; rename the folder")
-            continue
-        if "__" in name:
-            problems.append(f"{folder}: a name holding __ could clash with the ids derived from it; {rename}")
-            continue
-        if metadata:
-            environment_id, file_id = None, EnvironmentUniverseArtifact.derived_metadata_id(entry.id, name)
-        else:
-            environment_id = EnvironmentUniverseArtifact.derived_environment_id(entry.id, name)
-            file_id = EnvironmentArtifact.derived_file_id(environment_id)
-        try:
-            for id in (environment_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}); {rename}")
-            continue
-        laid = LaidOut(name, keys[0].rsplit("/", 1)[-1], files[keys[0]], file_id, environment_id)
-        (metadata_files if metadata else environments).append(laid)
-    if problems:
-        raise BundleError(problems)
-    return UniverseLayout(tuple(environments), tuple(metadata_files), files)
-
-
-def _folders(path: Path) -> list[str]:
-    """The names of the folders directly in ``path``, as the walk spells them, or none when it isn't one."""
+def universe_layout(bundle: Bundle, entry: BundleEntry) -> UniverseFolder:
+    """An environment_universe's folder, read as ``environment-universe get --output-dir`` writes one
+    (``read_folder``). Raises BundleError listing every problem."""
+    walk = _Walk(bundle, entry, empty_ok=True)
+    files = walk.files()
     try:
-        with os.scandir(path) as listing:
-            return [unicodedata.normalize("NFC", child.name) for child in listing
-                    if child.name not in LEAVINGS and child.is_dir()]
-    except OSError:
-        return []  # the walk reports why it can't be listed
+        return read_folder(entry.id, files, walk.folders)
+    except UniverseFolderError as e:
+        where = relative(bundle.root, entry.path)
+        raise BundleError([f"{where}/{problem}" for problem in e.problems]) from None
 
 
 def universe_files(bundle: Bundle, entry: BundleEntry) -> dict[str, Path]:
@@ -398,7 +301,8 @@ def universe_files(bundle: Bundle, entry: BundleEntry) -> dict[str, Path]:
 
 
 class _Walk:
-    """Lists an entry's folder, following a link only when its target is inside the bundle."""
+    """Lists an entry's folder, following a link only when its target is inside the bundle. ``folders`` then holds
+    every folder under it, by every path to it, as the names on that path in NFC."""
 
     def __init__(self, bundle: Bundle, entry: BundleEntry, empty_ok: bool = False):
         self.bundle = bundle
@@ -407,6 +311,7 @@ def __init__(self, bundle: Bundle, entry: BundleEntry, empty_ok: bool = False):
         self.root = _identity(bundle.root)
         self.problems: list[str] = []
         self.found: dict[str, Path] = {}
+        self.folders: set[tuple[str, ...]] = set()
 
     def files(self) -> dict[str, Path]:
         top = self.entry.path
@@ -452,6 +357,7 @@ def _walk(self, top: Path) -> None:
                     if identity in holders:
                         self._problem(path, "links back to a folder that holds it")
                     else:
+                        self.folders.add(tuple(unicodedata.normalize("NFC", part) for part in (*parts, child.name)))
                         pending.append((path, (*parts, child.name), holders | {identity}))
                 elif kind == stat.S_IFREG:
                     if os.access(path, os.R_OK):

src/agent_env/bundle/plan.py

@@ -25,6 +25,7 @@
     FileArtifactUniverse,
     SkillArtifact,
 )
+from agent_env.artifact.artifacts.environment_universe_folder import METADATA
 from agent_env.artifact.registry import canonical_type, get_artifact_registry
 from agent_env.config import get_config
 from agent_env.entity_refs import EntityKind
@@ -52,7 +53,6 @@
 
 from ._fs import fold, relative, with_article
 from .authoring import (
-    UNIVERSE_METADATA,
     AuthoringContext,
     entry_files,
     environment_files,
@@ -213,7 +213,7 @@ def _written_beside(self, source: ResolvedEntry) -> list[tuple[str, str]]:
                 layout = universe_layout(bundle, source.entry)
                 return [*((id, f"{where}/{laid.name}") for laid in layout.environments
                           for id in (laid.environment_id, laid.file_id)),
-                        *((laid.file_id, f"{where}/{UNIVERSE_METADATA}/{laid.name}") for laid in layout.metadata)]
+                        *((laid.file_id, f"{where}/{METADATA}/{laid.name}") for laid in layout.metadata)]
         except BundleError:
             pass
         return []