diff --git a/src/agent_env/bundle/_fs.py b/src/agent_env/bundle/_fs.py index 37d3e890b..738fbe8b6 100644 --- a/src/agent_env/bundle/_fs.py +++ b/src/agent_env/bundle/_fs.py @@ -1,9 +1,10 @@ """Filesystem plumbing for the bundle parser: a path spelled as the filesystem stores it, the user's -home, a regular file read without surprises, names compared without case, and paths rendered for -messages.""" +home, a regular file read without surprises, a file's content hash, names compared without case, and paths rendered +for messages.""" from __future__ import annotations +import hashlib import os import stat import sys @@ -96,6 +97,14 @@ def read_regular(path: Path) -> bytes: return data +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with open(path, "rb") as file: + for chunk in iter(lambda: file.read(1 << 20), b""): + digest.update(chunk) + return "sha256:" + digest.hexdigest() + + def fold(name: str) -> str: """The canonical caseless form: casefold output isn't always NFC, so normalize both sides.""" return unicodedata.normalize("NFC", unicodedata.normalize("NFC", name).casefold()) diff --git a/src/agent_env/bundle/_import_report.py b/src/agent_env/bundle/_import_report.py index 108c2a79b..747c099db 100644 --- a/src/agent_env/bundle/_import_report.py +++ b/src/agent_env/bundle/_import_report.py @@ -3,13 +3,20 @@ from __future__ import annotations +import hashlib import json import posixpath +import unicodedata from dataclasses import dataclass from datetime import datetime, timezone from importlib.metadata import version as package_version from pathlib import Path +from typing import Any +from urllib.parse import urlsplit +from agent_env.bundle._fs import file_sha256 +from agent_env.config import get_config +from agent_env.config.describe import is_secret_key from agent_env.config.model import ENV_API_KEY, ENV_BASE_URL from agent_env.entity.refs import EntityKind from agent_env.store.image_store.oci_registry_credentials import registry_host_from_ref @@ -17,8 +24,10 @@ from agent_env.utils.docker.lock import IN_IMAGE, Locked from agent_env.utils.docker.prune import Pruned -# The file an import records what it wrote in, and from where, for publishing the folder back. +# The file an import records what it wrote in, and from where, for registering the folder back. PROVENANCE = "provenance.json" +# Bumped when what the file records changes. +PROVENANCE_SCHEMA = 2 @dataclass(frozen=True) @@ -92,6 +101,8 @@ class Pulled: folder: str key: str image: str + artifact: str # the stored image's id and version + version: int why: str | None = None # why it's pulled though it has a build context: the context can't be written @@ -155,22 +166,72 @@ def size_text(size: int) -> str: return f"{size} B" if size < 1_000 else f"{size / 1_000:.1f} KB" if size < 1_000_000 else f"{size / 1_000_000:.1f} MB" -def provenance(source: dict, imported: BundleImport) -> str: - """What the folder was imported from and what it holds, by store id and version: no store's locations.""" +def provenance(source: dict, imported: BundleImport, root: Path) -> str: + """What the folder was imported from and what it holds, by store id and version, and each file written at ``root``, + so an edit can be told from what the store holds: no store's locations, only a fingerprint of the store's config.""" return json.dumps({ + "schema": PROVENANCE_SCHEMA, "agent_env": package_version("agentenv-framework"), "imported_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "from": source, + "store": store_fingerprint(), + "tasks": [{"file": task.path, "id": task.id, "version": task.version} for task in imported.tasks], + "eval": {"file": imported.eval.path, "id": imported.eval.id, "version": imported.eval.version} + if imported.eval else None, "written": [{"folder": entity.folder, "kind": entity.kind.value, "type": entity.type, "id": entity.id, "version": entity.version} for entity in imported.written], "images": [{"folder": image.folder, "key": image.key, "id": image.artifact, "version": image.version, "dockerfile": image.pruned.dockerfile, "dockerfile_found": image.found} for image in imported.images], - "store": [{"where": pin.where, "kind": pin.kind.value, "id": pin.id, "version": pin.version, "why": pin.why} - for pin in imported.pinned], + "pulled": [{"folder": pull.folder, "key": pull.key, "id": pull.artifact, "version": pull.version} + for pull in imported.pulled], + "pinned": [{"where": pin.where, "kind": pin.kind.value, "id": pin.id, "version": pin.version, "why": pin.why} + for pin in imported.pinned], + "files": written_files(root), }, indent=2, ensure_ascii=False) + "\n" +def store_fingerprint() -> str: + """A hash of where the configured document store is: its section's ``impl``, and the strings its ``config`` holds, + ``env:`` and ``secret:`` references resolved, a URL as its scheme, host and path alone, and nothing a secret-named + key holds. Credentials and tuning options don't tell one store from another, so neither changes it, and the hash + names neither where the store is nor how to reach it.""" + config = get_config() + section = config.trace_section("document").value + located = {"impl": str(section.get("impl")), "config": _location(config.resolve_reference(section.get("config", {})))} + text = json.dumps(located, sort_keys=True, separators=(",", ":")) + return "sha256:" + hashlib.sha256(text.encode()).hexdigest() + + +def _location(value: Any) -> Any: + """What in ``value`` says where a store is: each string, a URL as its scheme, host and path, through dicts but the + items of secret-named keys, and lists; None for anything else.""" + if isinstance(value, dict): + return {key: located for key, item in value.items() + if not is_secret_key(str(key)) and (located := _location(item)) is not None} + if isinstance(value, list): + return [located for item in value if (located := _location(item)) is not None] + if isinstance(value, str): + parts = urlsplit(value) + return f"{parts.scheme}://{parts.netloc.rpartition('@')[2]}{parts.path}" if parts.scheme and parts.netloc else value + return None + + +def written_files(root: Path) -> dict[str, str]: + """Each file under ``root`` but its README and provenance, by its path in NFC, as its ``file_digest``.""" + files = {} + for path in sorted(root.rglob("*")): + relative = unicodedata.normalize("NFC", path.relative_to(root).as_posix()) + if path.is_file() and relative not in ("README.md", PROVENANCE): + files[relative] = file_digest(path) + return files + + +def file_digest(path: Path) -> str: + """``path``'s content hash, ``+x`` after it when it's executable, as an import records each file it wrote.""" + return file_sha256(path) + ("+x" if path.stat().st_mode & 0o111 else "") + + def readme(description: str, imported: BundleImport) -> str: """The bundle's README: its description, where it came from, how to run and share it, what it holds, then the import's report.""" diff --git a/src/agent_env/bundle/_imported.py b/src/agent_env/bundle/_imported.py new file mode 100644 index 000000000..8232e6268 --- /dev/null +++ b/src/agent_env/bundle/_imported.py @@ -0,0 +1,401 @@ +"""What a folder ``bundle init`` wrote holds of the store it was imported from, for registering it back: each entity, +image, task and eval's store id and version, whether the folder still holds each as the import wrote it, and the +user-scoped fields the import left out of each task's steps.""" + +from __future__ import annotations + +import copy +import json +import unicodedata +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, replace +from pathlib import Path +from typing import NamedTuple + +from agent_env.entity.paths import field_sites +from agent_env.entity.refs import ref_sites +from agent_env.entity.scopes import FieldScope +from agent_env.env.registry import get_env_registry +from agent_env.store.base import NotFoundError +from agent_env.task.task import Task +from agent_env.task_step.registry import get_task_step_registry + +from ._import_report import PROVENANCE, PROVENANCE_SCHEMA, file_digest, store_fingerprint +from ._secrets import scrub +from ._values import strings, value_at +from .authoring import entry_files +from .parse import CONFIG_FILES, Bundle, BundleEntry, BundleKind +from .plan import Plan, Write, env_folder_writer +from .resolve import BuiltImage, BundleImage + + +class Source(NamedTuple): + """The store id and version a write was imported from.""" + + id: str + version: int + + +@dataclass(frozen=True) +class ImportedImage: + """An image the import wrote into a folder, under a toml key: built from the Dockerfile it names, or pulled.""" + + source: Source + dockerfile: str | None = None + + +@dataclass(frozen=True) +class Imported: + """What an import recorded: the stored id and version of each entity folder, task file and eval file, and of each + image a folder names, by its folder and toml key; and each file as it wrote it.""" + + entities: Mapping[str, Source] + images: Mapping[tuple[str, str], ImportedImage] + files: Mapping[str, str] + + +def read_imported(root: Path) -> tuple[Imported | None, str | None]: + """What the import of ``root`` recorded, when it was imported from the configured store; else None, and why when + it has a record this can't use.""" + path = root / PROVENANCE + if not path.is_file(): + return None, None + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None, f"{PROVENANCE} can't be read, so the folder registers as a new bundle" + if not isinstance(record, dict) or record.get("schema") != PROVENANCE_SCHEMA: + return None, f"{PROVENANCE} is from an older agent-env, so the folder registers as a new bundle" + if record.get("store") != store_fingerprint(): + return None, "it was imported from another store, so its entries take new ids" + entities = {item["folder"]: Source(item["id"], item["version"]) for item in record.get("written", [])} + entities |= {item["file"]: Source(item["id"], item["version"]) for item in record.get("tasks", [])} + if (stored_eval := record.get("eval")) is not None: + entities[stored_eval["file"]] = Source(stored_eval["id"], stored_eval["version"]) + images = {(item["folder"], item["key"]): ImportedImage(Source(item["id"], item["version"]), item.get("dockerfile")) + for item in record.get("images", [])} + images |= {(item["folder"], item["key"]): ImportedImage(Source(item["id"], item["version"])) + for item in record.get("pulled", [])} + return Imported(entities, images, record.get("files", {})), None + + +def where(root: Path, entry: BundleEntry) -> str: + """``entry``'s folder or file in the bundle at ``root``, as the import's record names it.""" + return unicodedata.normalize("NFC", entry.path.relative_to(root).as_posix()) + + +def source_of(imported: Imported, plan: Plan, write: Write) -> Source | None: + """The store id and version ``write`` was imported from, or None when the import didn't write it, as for what an + installed bundle holds.""" + entry = write.source.entry + if entry in plan.origins: + return None + folder = where(plan.bundle.bundle.root, entry) + if isinstance(write.source, BundleImage) and write.source.id != entry.id: + image = imported.images.get((folder, image_key(write.source))) + return image.source if image is not None else None + return imported.entities.get(folder) + + +def image_key(image: BundleImage) -> str: + """The toml key an image an env or agent folder builds is named by: ``image`` for its own, as the import names it.""" + role = image.id[len(image.entry.id) + 2:] + return "image" if role == f"{image.entry.kind.store}_image" else role + + +def as_imported(imported: Imported, plan: Plan, write: Write, now: Mapping[str, str]) -> bool: + """Whether the files ``write`` is made from are those the import wrote: a task's or eval's file; an image a toml + key builds, its folder but the toml and the Dockerfile the key names; an env's or agent's toml, or every file of an + env type written from its folder; every file of an artifact, skill or image folder.""" + entry = write.source.entry + folder = where(plan.bundle.bundle.root, entry) + if write.kind in (BundleKind.TASK, BundleKind.EVAL): + return folder in imported.files and imported.files[folder] == now.get(folder) + toml = f"{folder}/{CONFIG_FILES[entry.kind]}" if entry.kind in CONFIG_FILES else None + then, current = _within(imported.files, folder), _within(now, folder) + if isinstance(write.source, BuiltImage) and write.source.id != entry.id: + image = imported.images.get((folder, image_key(write.source))) + if image is None or image.dockerfile != write.source.dockerfile: + return False + then.pop(toml, None) + current.pop(toml, None) + elif (isinstance(write.source, BundleImage) and write.source.id != entry.id) or ( + entry.kind in (BundleKind.ENV, BundleKind.AGENT) and not ( + entry.kind is BundleKind.ENV and _written_from_folder(entry))): + then, current = {toml: then.get(toml)}, {toml: current.get(toml)} + return then == current + + +class ImportedWrites: + """What ``imported`` records of each write of ``plan``: the stored id and version it was imported from, and whether + the folder holds it as the import wrote it; and what of the record's tasks ``restored`` couldn't give back. Writes + are told apart by store as well as ``@local`` id, since a task and an env of one name share one.""" + + def __init__(self, imported: Imported | None, plan: Plan, restored: Restored): + self.plan = plan + self.given = restored.given + self._writes = {_key(write): write for write in plan.writes} + self._sources = {_key(write): source for write in plan.writes + if imported is not None and (source := source_of(imported, plan, write)) is not None} + self.tasks = {write.source.entry.id for write in plan.writes + if write.kind is BundleKind.TASK and _key(write) in self._sources} + self.held = _held(imported, plan, [self._writes[key] for key in self._sources]) if imported else frozenset() + self.problems = [problem for write in plan.writes if write.kind is BundleKind.TASK + for problem in restored.unrestored.get(write.source.entry.path, ())] + + def source(self, write: Write) -> Source | None: + return self._sources.get(_key(write)) + + def recorded(self, ids: Mapping[str, str]) -> set[str]: + """The registry ids in ``ids`` the record gives rather than the rules: each imported write's, each imported + task's output's, and each derived from either.""" + taken = {local for _, local in self._sources} + taken |= {local for local in ids if any(local.startswith(f"{task}/") for task in self.tasks)} + return {ids[local] for local in ids if local in taken or any(local.startswith(f"{base}__") for base in taken)} + + def copies(self, store: str, locals_: Sequence[str], registry_id: str) -> list[str] | None: + """When ``locals_``, which take one registry id in ``store``, were imported from one stored entity: a problem + naming those the folder changed, since a registration writes each id once, or none; else None.""" + keys = [(store, local) for local in locals_] + if None in (sources := {self._sources.get(key) for key in keys}) or len(sources) > 1: + return None + paths = sorted(self.plan.path(self._writes[key].source.entry) for key in keys) + if not (edited := sorted(self.plan.path(self._writes[key].source.entry) for key in keys + if key not in self.held)): + return [] + return [f"{', '.join(paths)} hold copies of {registry_id!r} as imported, and {', '.join(edited)} changed it; " + "a registration writes each id once, so undo the change"] + + def versions(self, ids: Mapping[str, str]) -> tuple[dict[tuple[str, str], int], set[tuple[str, str]]]: + """By (store, registry id), the version each imported write was imported at, and which of them the folder holds + as the import wrote it.""" + versions, held = {}, set() + for (store, local), source in self._sources.items(): + key = (store, ids.get(local, local)) + versions[key] = source.version + if (store, local) in self.held: + held.add(key) + return versions, held + + +def _key(write: Write) -> tuple[str, str]: + return (write.kind.store, write.id) + + +def _held(imported: Imported, plan: Plan, writes: list[Write]) -> frozenset[str]: + """The (store, ``@local`` id) of each of ``writes`` the folder holds as ``imported`` records the import wrote it.""" + now = {} + for entry in {write.source.entry for write in writes}: + now |= _files_of(plan.bundle.bundle, entry) + return frozenset(_key(write) for write in writes if as_imported(imported, plan, write, now)) + + +def _files_of(bundle: Bundle, entry: BundleEntry) -> dict[str, str]: + """The files a write of ``entry`` reads, as its writer walks them, by their paths in the bundle, as their + ``file_digest``: a task's or eval's file, or every file of its folder and its toml.""" + folder = where(bundle.root, entry) + if entry.kind in (BundleKind.TASK, BundleKind.EVAL): + return {folder: file_digest(entry.path)} + files = entry_files(bundle, entry, empty_ok=True) + if entry.kind in CONFIG_FILES and (toml := entry.path / CONFIG_FILES[entry.kind]).is_file(): + files[toml.name] = toml + return {f"{folder}/{key}": file_digest(path) for key, path in files.items()} + + +@dataclass(frozen=True) +class Restored: + """A bundle given back what its import left out (``restore``): the paths in each task file's steps that took a + stored value, and a problem for each field still holding what the import masked.""" + + bundle: Bundle + given: Mapping[Path, set[str]] + unrestored: Mapping[Path, list[str]] + + +def restore(bundle: Bundle, imported: Imported | None) -> Restored: + """``bundle`` with each imported task's steps given back, from the stored version each was imported from, matched + by step id and type, what the import left out where the folder doesn't set it: each user-scoped field, but a + reference's pin where the folder names something else; each list item's, from the stored item it was written from; + and each env var and credential the import's scrub took, where the folder holds what the scrub left. And, for each + task's file, the paths in its steps that took one, and a problem for each field still holding what the scrub left, + which the folder changed too much to give back. A bundle with no record is as it is.""" + if imported is None: + return Restored(bundle, {}, {}) + registry = get_task_step_registry() + entries, taken, problems = [], {}, {} + for entry in bundle.entries: + source = imported.entities.get(where(bundle.root, entry)) if entry.kind is BundleKind.TASK else None + try: + stored = Task.get(*source).to_dict()["steps"] if source is not None else None + except NotFoundError: + stored = None + if stored is None: + entries.append(entry) + continue + by_id = {step.get("id"): step for step in stored} + steps = copy.deepcopy(entry.config) + for index, step in enumerate(steps): + origin = by_id.get(step.get("id")) + cls = registry.get(step.get("type")) + if origin is None or cls is None or origin.get("type") != step.get("type"): + continue + given, left = _given_back(step, origin, cls) + for path in given: + taken.setdefault(entry.path, set()).add(f"[{index}].{path}") + problems.setdefault(entry.path, []).extend( + f"{where(bundle.root, entry)}: step {step.get('id')!r}: {path} still holds what the import masked, and " + "the folder changed what it was masked from; set its real value" for path in left) + entries.append(replace(entry, config=steps)) + return Restored(replace(bundle, entries=tuple(entries)), taken, problems) + + +def _given_back(step: dict, origin: dict, cls: type) -> tuple[list[str], list[str]]: + """Give ``step`` back what the import left out of ``origin``, in place; the path of each field it took, and of each + still holding what the scrub left.""" + user = [path for path, scope in cls.field_scopes.items() if scope is FieldScope.USER] + repinned = _repinned(step, origin, cls) + given, lists = [], {} + for path in user: + if path in repinned: + continue + if "[]" in path: + at, _, field = path.partition("[].") + if "[]" not in field: + lists.setdefault(at, []).append(field) + continue + for _, holder, key in field_sites(origin, path): + if holder[key] not in ("", [], {}) and _graft(step, path.split("."), holder[key]): + given.append(path) + for at, fields in lists.items(): + given += _given_back_items(step, origin, at, fields) + scrubbed = copy.deepcopy(origin) + scrub(scrubbed) + given += _unscrubbed(step, origin, scrubbed) + masked = {text for _, text in _masks(origin, scrubbed)} + return given, [path for path, text in strings(step) if text in masked] + + +def _repinned(step: dict, origin: dict, cls: type) -> set[str]: + """The path of each pin in ``origin`` beside a reference ``step`` names something else at, such as an entity the + import wrote into the folder, which can't take the stored entity's pin.""" + refs = cls.entity_refs or () + held = {site.path: site.value for site in ref_sites(refs, step)} + return {site.path[:len(site.path) - len(str(site.key))] + site.version_key for site in ref_sites(refs, origin) + if site.version_key is not None and held.get(site.path, site.value) != site.value} + + +def _given_back_items(step: dict, origin: dict, at: str, fields: list[str]) -> list[str]: + """Give each item of the list ``at`` the ``fields`` the import left out of it, from the stored item it was written + from (``_source_item``): all of them, or none when the item sets one, since together they can be one choice, such as + a source under either of its names.""" + items, stored = value_at(step, at.split(".")), value_at(origin, at.split(".")) + if not isinstance(items, list) or not isinstance(stored, list): + return [] + left_out = {field.split(".")[0] for field in fields} + given = [] + for index, item in enumerate(items): + if not isinstance(item, dict) or any(value_at(item, field.split(".")) is not None for field in fields): + continue + if (source := _source_item(item, stored, left_out)) is None: + continue + for field in fields: + for _, holder, key in field_sites(source, field): + if holder[key] not in ("", [], {}) and _graft(item, field.split("."), holder[key]): + given.append(f"{at}[{index}].{field}") + return given + + +def _source_item(item: dict, stored: list, left_out: set[str]) -> dict | None: + """The stored item ``item`` was written from: the one it equals but for the ``left_out`` fields, when only one + does. An edited item matches none, so it takes back nothing.""" + bare = _bare(item, left_out) + same = [one for one in stored if isinstance(one, dict) and _bare(one, left_out) == bare] + return same[0] if len(same) == 1 else None + + +def _bare(item: dict, left_out: set[str]) -> dict: + return {key: value for key, value in item.items() if key not in left_out and value is not None} + + +def _unscrubbed(value: object, stored: object, scrubbed: object, path: str = "") -> list[str]: + """Put back in ``value``, in place, what scrubbing ``stored`` took where ``value`` holds what it left + (``scrubbed``): each env var it dropped that ``value`` doesn't set, and each credential it masked; each one's + path.""" + given = [] + if isinstance(value, dict) and isinstance(stored, dict) and isinstance(scrubbed, dict): + for key, item in stored.items(): + at = f"{path}.{key}" if path else str(key) + if key not in scrubbed: + if key not in value: + value[key] = copy.deepcopy(item) + given.append(at) + elif key in value: + if isinstance(item, str): + if value[key] == scrubbed[key] != item: + value[key] = item + given.append(at) + else: + given += _unscrubbed(value[key], item, scrubbed[key], at) + elif isinstance(value, list) and isinstance(stored, list) and isinstance(scrubbed, list) and ( + len(stored) == len(scrubbed)): + unchanged = value == scrubbed + for index, item in enumerate(value): + if (source := _counterpart(item, index, stored, scrubbed, unchanged)) is None: + continue + at = f"{path}[{index}]" + if isinstance(stored[source], str): + if item == scrubbed[source] != stored[source]: + value[index] = stored[source] + given.append(at) + else: + given += _unscrubbed(item, stored[source], scrubbed[source], at) + return given + + +def _counterpart(item: object, index: int, stored: list, scrubbed: list, unchanged: bool) -> int | None: + """The index of the stored item ``item``, at ``index``, was scrubbed from: its own while the list is as scrubbed; + else the one scrubbed item it equals, or failing that the one whose masked values it holds, when only one does, so + a reordered or edited list restores each item from its own.""" + if unchanged: + return index + same = [other for other, one in enumerate(scrubbed) if one == item] + if not same: + same = [other for other, one in enumerate(scrubbed) if (masks := _masks(stored[other], one)) + and all(value_at(item, keys) == text for keys, text in masks)] + return same[0] if len(same) == 1 else None + + +def _masks(stored: object, scrubbed: object, keys: tuple = ()) -> list[tuple[tuple, str]]: + """Each string scrubbing ``stored`` masked: the keys to it, and the text the scrub left.""" + if isinstance(stored, str): + return [(keys, scrubbed)] if isinstance(scrubbed, str) and scrubbed != stored else [] + if isinstance(stored, dict) and isinstance(scrubbed, dict): + return [mask for key in stored if key in scrubbed for mask in _masks(stored[key], scrubbed[key], (*keys, key))] + if isinstance(stored, list) and isinstance(scrubbed, list) and len(stored) == len(scrubbed): + return [mask for index, (one, other) in enumerate(zip(stored, scrubbed)) + for mask in _masks(one, other, (*keys, index))] + return [] + + +def _graft(step: dict, keys: list[str], value: object) -> bool: + """Set ``keys`` in ``step`` to ``value`` when it's unset there; whether it was.""" + holder = step + for key in keys[:-1]: + child = holder.get(key) + if child is None: + child = holder[key] = {} + if not isinstance(child, dict): + return False + holder = child + if holder.get(keys[-1]) is not None: + return False + holder[keys[-1]] = copy.deepcopy(value) + return True + + +def _within(files: Mapping[str, str], folder: str) -> dict[str, str]: + return {path: digest for path, digest in files.items() if path.startswith(f"{folder}/")} + + +def _written_from_folder(entry: BundleEntry) -> bool: + return env_folder_writer(get_env_registry().get(entry.type)) diff --git a/src/agent_env/bundle/_values.py b/src/agent_env/bundle/_values.py new file mode 100644 index 000000000..e837e3fe6 --- /dev/null +++ b/src/agent_env/bundle/_values.py @@ -0,0 +1,30 @@ +"""Walks over JSON-like values: each string in one with its path, and the value under a path of keys.""" + +from __future__ import annotations + +from collections.abc import Iterable, Iterator +from typing import Any + + +def strings(value: Any, path: str = "") -> Iterator[tuple[str, str]]: + """Each string in ``value``, with its path: ``a.b`` under a dict, ``a[0]`` in a list.""" + if isinstance(value, str): + yield path, value + elif isinstance(value, dict): + for key, item in value.items(): + yield from strings(item, f"{path}.{key}" if path else str(key)) + elif isinstance(value, list): + for index, item in enumerate(value): + yield from strings(item, f"{path}[{index}]") + + +def value_at(value: Any, keys: Iterable[str | int]) -> Any: + """What ``value`` holds under ``keys``: a dict's item by its key, a list's by its index; None where it holds none.""" + for key in keys: + if isinstance(value, dict) and isinstance(key, str): + value = value.get(key) + elif isinstance(value, list) and isinstance(key, int) and 0 <= key < len(value): + value = value[key] + else: + return None + return value diff --git a/src/agent_env/bundle/importer.py b/src/agent_env/bundle/importer.py index c1c6d1f55..30180a89c 100644 --- a/src/agent_env/bundle/importer.py +++ b/src/agent_env/bundle/importer.py @@ -163,6 +163,7 @@ def __init__(self, dest: Path, include_scopes: Collection[FieldScope] = (), self.progress = progress or (lambda message: None) self.files: dict[str, str] = {} self.taken: dict[str, set[str]] = {kind.value: set() for kind in BundleKind} # folded names, per folder + self.named: dict[str, str] = {} # each folded name, and the id it was taken for self.tasks: list[ImportedTask] = [] self.eval: ImportedEval | None = None self.written: list[Written] = [] @@ -265,7 +266,7 @@ def image(self, out: _EntityOut, artifact: DockerImageArtifact, key: str) -> str out.images.append(job) return job.toml if names_registry(artifact.image_name): - out.pulled.append(Pulled(out.folder, key, artifact.image_name)) + out.pulled.append(Pulled(out.folder, key, artifact.image_name, artifact.id, artifact.version)) return {"ref": artifact.image_name} why = (f"it's stored only as a docker save tarball, under {artifact.image_name}, which names no registry" if artifact.tar_gz_object_url else "it records neither a build context nor a registry to pull it from") @@ -295,7 +296,6 @@ def write(self, description: str, source: dict) -> BundleImport: pinned=tuple(self.pinned), floating=tuple(self.floating), dropped=tuple(self.dropped), needs=tuple(self.needs), undeclared=tuple(self.undeclared), duplicates=tuple(self.duplicates), downloaded=downloaded) - self.files[PROVENANCE] = provenance(source, imported) self.files["README.md"] = readme(description, imported) try: for folder, (name, toml) in self.tomls.items(): @@ -304,6 +304,7 @@ def write(self, description: str, source: dict) -> BundleImport: for path, text in self.files.items(): (staging / path).parent.mkdir(parents=True, exist_ok=True) (staging / path).write_text(text, encoding="utf-8") + (staging / PROVENANCE).write_text(provenance(source, imported, staging), 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)), installed=installed_bundles) @@ -345,7 +346,8 @@ def _image_left_in_store(self, job: ImageJob) -> None: if names_registry(job.artifact.image_name): job.toml.clear() job.toml["ref"] = job.artifact.image_name - self.pulled.append(Pulled(job.folder, job.key, job.artifact.image_name, job.problem)) + self.pulled.append(Pulled(job.folder, job.key, job.artifact.image_name, job.artifact.id, + job.artifact.version, job.problem)) return key = (EntityKind.ARTIFACT, job.artifact.id, job.artifact.version) try: @@ -417,6 +419,7 @@ def _folder(self, kind: EntityKind, entity: Any) -> tuple[BundleKind, str]: raise Unwritable(f"a skill's folder takes its skill's name, {name!r}, which another artifact in the " "folder already has") self.taken[BundleKind.ARTIFACT.value].add(fold(name)) + self.named.setdefault(fold(name), entity.id) return BundleKind.SKILL, name bundle_kind = _FOLDERS[kind] return bundle_kind, self._name(bundle_kind.value, entity.id) @@ -500,14 +503,17 @@ def _default_agents(self, step: dict, refs: tuple[EntityRef, ...], where: str) - "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.""" + """A name in the ``kind`` folder: the id's last segment made valid, numbered when taken, caselessly, or when + another kind's folder holds it for another id, since entries of every kind that share a name share a local id, + and registering the folder gives each the id it was imported from.""" names = _names_of(BundleKind(kind)) base = _valid_name(entity_id.rsplit("/", 1)[-1]) or kind.removesuffix("s") name, n = base, 1 - while fold(name) in self.taken[names]: + while fold(name) in self.taken[names] or self.named.get(fold(name), entity_id) != entity_id: n += 1 name = f"{base}-{n}" self.taken[names].add(fold(name)) + self.named[fold(name)] = entity_id return name diff --git a/src/agent_env/bundle/ledger.py b/src/agent_env/bundle/ledger.py index 573b86d8a..5e8338d5e 100644 --- a/src/agent_env/bundle/ledger.py +++ b/src/agent_env/bundle/ledger.py @@ -31,6 +31,7 @@ from agent_env.store.local_state import holding_locks from agent_env.task.store import TASKS_COLLECTION +from ._fs import file_sha256 from .authoring import build_context_files from .parse import BundleKind from .plan import Plan, Write, artifact_writer, env_folder_writer, env_writer, folder_files, unpinned_store_refs @@ -137,7 +138,7 @@ def digest(self, write: Write, needs: Mapping[tuple[str, str], int]) -> Digest | built = isinstance(write.source, BuiltImage) listing = build_context_files if built else folder_files for key, path in listing(self._plan.bundle_of(write.source.entry), write.source.entry).items(): - inputs["files"][key] = _file_sha256(path) + (_mode(path) if built else "") + 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 self.pins_references or write.kind not in (BundleKind.TASK, BundleKind.EVAL): @@ -328,13 +329,5 @@ def _mode(path: Path) -> str: return "" if mode == _USUAL_MODE else f"+{mode:o}" -def _file_sha256(path: Path) -> str: - digest = hashlib.sha256() - with open(path, "rb") as file: - for chunk in iter(lambda: file.read(1 << 20), b""): - digest.update(chunk) - return "sha256:" + digest.hexdigest() - - def _now() -> str: return datetime.now(timezone.utc).isoformat() diff --git a/src/agent_env/bundle/register.py b/src/agent_env/bundle/register.py index cb2477bde..6c1e97c5f 100644 --- a/src/agent_env/bundle/register.py +++ b/src/agent_env/bundle/register.py @@ -37,9 +37,11 @@ from agent_env.store.routing import configured_store, namespace_routing from agent_env.task_step.registry import get_task_step_registry +from ._imported import ImportedWrites, read_imported, restore from ._secrets import credentials_in +from ._values import strings from .installed import installed_bundles -from .ledger import Ledger +from .ledger import Check, Ledger from .materialize import Materialization, Materialized, materialize from .parse import Bundle, BundleEntry, BundleError, BundleKind, parse_bundle from .plan import AliasUse, Plan, Write, plan_bundle, plan_entities @@ -82,22 +84,29 @@ def register_bundle( It turns namespace routing on while it runs. Raises BundleError listing every problem found before anything is written; a write that fails stops it, and every earlier one stays, in the ledger, for the next run to reuse.""" + say = on_progress or (lambda line: None) with namespace_routing(): bundle = parse_bundle(Path(root)) - resolved = resolve_bundle(bundle) - plan = _plan(resolved, tasks, evals) - ids, problems = registry_ids(plan, _prefix(bundle, prefix)) + imported, why = read_imported(bundle.root) + if why is not None: + say(f" {why}") + restored = restore(bundle, imported) + plan = _plan(resolve_bundle(restored.bundle), tasks, evals) + imports = ImportedWrites(imported, plan, restored) + if imports.problems: + raise BundleError(imports.problems) + ids, problems = registry_ids(plan, _prefix(bundle, prefix), imports) if problems: raise BundleError(problems) registered = translated(plan, ids) target = configured_store(get_config().get_document_store()) shared = shared_store(target) identity, commit, dirty = bundle_identity(bundle.root) - ledger = RegistrationLedger(target, registered, identity, commit, dirty) - problems = list(_unregistrable(registered, shared)) + versions, held = imports.versions(ids) + ledger = RegistrationLedger(target, registered, identity, commit, dirty, versions, held) + problems = list(_unregistrable(registered, shared, imports.given)) if problems: raise BundleError(problems) - say = on_progress or (lambda line: None) materialization = materialize( registered, dry_run=dry_run, ledger=ledger, on_wait=lambda: say("waiting for another agent-env run to finish writing this bundle's ids"), @@ -141,18 +150,26 @@ def _prefix(bundle: Bundle, prefix: str | None) -> str: return f"{name}-" -def registry_ids(plan: Plan, prefix: str) -> tuple[dict[str, str], list[str]]: +def registry_ids(plan: Plan, prefix: str, + imported: ImportedWrites | None = None) -> tuple[dict[str, str], list[str]]: """The registry id each ``@local`` id the plan writes takes, and each problem giving them. - The bundle's own entry is ````, its name lowercased with each run of what an id can't hold made - ``-``. An installed bundle's agent pulled in by a reference takes the first alias its bundle gives it, which the - reference named it by; any other entity of an installed bundle has no registry id, and an env's alias can't be - one, since no store env may start on an installed env alias's id. An id derived from another (``__env_image``) keeps its suffix after the - other's registry id, and a task's output is ``-``.""" + What ``imported`` records the folder was imported from takes the id it had there, and an imported task's outputs the + names its file gives them, as the stored task did. Any other entry of the bundle's own is ````, its + name lowercased with each run of what an id can't hold made ``-``. An installed bundle's agent pulled in by a + reference takes the first alias its bundle gives it, which the reference named it by; any other entity of an + installed bundle has no registry id, and an env's alias can't be one, since no store env may start on an installed + env alias's id. An id derived from another (``__env_image``) keeps its suffix after the other's registry id, and + a task's output is ``-``.""" ids: dict[str, str] = {} problems: list[str] = [] for write in plan.writes: entry = write.source.entry + if imported is not None and (source := imported.source(write)) is not None: + if ids.setdefault(write.id, source.id) != source.id: + problems.append(f"{plan.path(entry)}: shares its name with what's registered as {ids[write.id]!r}, " + f"so it can't keep the id it was imported as, {source.id!r}; rename one") + continue if entry.id in ids or (isinstance(write.source, BundleImage) and write.id != entry.id): continue if entry in plan.origins: @@ -172,8 +189,9 @@ def registry_ids(plan: Plan, prefix: str) -> tuple[dict[str, str], list[str]]: source = write.source if isinstance(source, ResolvedEntry) and source.entry.id in ids: for output in source.outputs: - name = _slug(output.id.removeprefix(f"{source.entry.id}/").rsplit("/", 1)[-1]) - ids[output.id] = f"{ids[source.entry.id]}-{name}" + written = output.id.removeprefix(f"{source.entry.id}/") + ids[output.id] = (written if imported is not None and source.entry.id in imported.tasks + else f"{ids[source.entry.id]}-{_slug(written.rsplit('/', 1)[-1])}") stores: dict[str, set[str]] = {} for write in plan.writes: stores.setdefault(write.id, set()).add(write.kind.store) @@ -181,13 +199,18 @@ def registry_ids(plan: Plan, prefix: str) -> tuple[dict[str, str], list[str]]: for output in write.source.outputs: stores.setdefault(output.id, set()).add(BundleKind.ARTIFACT.store) by_target: dict[tuple[str, str], list[str]] = {} + recorded = imported.recorded(ids) if imported is not None else set() for local, registry_id in ids.items(): for store in stores.get(local, ()): by_target.setdefault((store, registry_id), []).append(local) - if not _REGISTRY_ID.fullmatch(registry_id): + if registry_id not in recorded and not _REGISTRY_ID.fullmatch(registry_id): problems.append(f"{registry_id!r}: isn't an id a registry takes; rename what it's made from") - for (_, registry_id), locals_ in by_target.items(): - if len(locals_) > 1: + for (store, registry_id), locals_ in by_target.items(): + if len(locals_) < 2: + continue + if imported is not None and (copies := imported.copies(store, locals_, registry_id)) is not None: + problems += copies + else: problems.append(f"{', '.join(sorted(locals_))} would all be registered as {registry_id!r}; rename one") return ids, problems @@ -293,9 +316,32 @@ class RegistrationLedger(Ledger): strict = True pushes_images = True - def __init__(self, store: DocumentStore, plan: Plan, identity: str, commit: str | None, dirty: bool | None): + def __init__(self, store: DocumentStore, plan: Plan, identity: str, commit: str | None, dirty: bool | None, + sources: Mapping[tuple[str, str], int] | None = None, kept: set[tuple[str, str]] | None = None): super().__init__(store, plan) self.identity, self.commit, self.dirty = identity, commit, dirty + self.sources, self.kept = sources or {}, kept or set() + + def check(self, write: Write, needs: Mapping[tuple[str, str], int]) -> Check: + """``Ledger.check``, but a write the folder holds as its import wrote it, all it needs kept at the versions they + were imported at, keeps the version it was imported at, and the ledger records it.""" + if (kept := self._as_imported(write, needs)) is not None: + return kept + return super().check(write, needs) + + def _as_imported(self, write: Write, needs: Mapping[tuple[str, str], int]) -> Check | None: + key = (write.kind.store, write.id) + if key not in self.kept or any(needs[need] != self.sources.get(need) for need in write.needs): + return None + stored = self._stored_version(write) + if stored is None or stored < self.sources[key]: + return None + digest = self.digest(write, needs) + row = self._latest(*key) + recorded = row is not None and digest is not None and (row["bundle"], row["version"], row.get("digest")) == ( + self._bundle(write), self.sources[key], digest.value) + return Check(write, digest, self.sources[key], ("as it was imported",), stored, needs, + adopted=digest is not None and not recorded) def _bundle(self, write: Write) -> str: """The bundle's identity, or for an entity an installed bundle comes with, that bundle's, which every bundle @@ -339,32 +385,28 @@ def _remote_name(remote: str) -> str: return f"{host}/{path.strip('/').removesuffix('.git')}" -def _unregistrable(plan: Plan, shared: bool) -> Iterator[str]: +def _unregistrable(plan: Plan, shared: bool, grafts: Mapping[Path, set[str]]) -> Iterator[str]: """A problem for each ``@local`` id left in what the plan writes and, for a ``shared`` store, each credential - written out and each URL that reaches only this machine.""" + written out and each URL that reaches only this machine, but in a field given back from the stored task the folder + was imported from (``grafts``), which the store holds already.""" for write in plan.writes: if not isinstance(write.source, ResolvedEntry): continue where = plan.path(write.source.entry) - for path, text in _strings(write.source.config): + taken = grafts.get(write.source.entry.path, set()) + + def stored(path: str) -> bool: + return any(path == field or path.startswith((f"{field}.", f"{field}[")) for field in taken) + + for path, text in strings(write.source.config): if is_local_id(text): yield f"{where}: {path or 'its value'} names {text!r}, which only this machine's stores hold" - elif shared and (url := _local_url(text)) is not None: + elif shared and not stored(path) and (url := _local_url(text)) is not None: yield f"{where}: {path or 'its value'} names {url}, which only this machine reaches" if shared: for path, why in credentials_in(write.source.config): - yield f"{where}: {path}: {why}; leave it out, for whatever runs the task to give it" - - -def _strings(value: Any, path: str = "") -> Iterator[tuple[str, str]]: - if isinstance(value, str): - yield path, value - elif isinstance(value, dict): - for key, item in value.items(): - yield from _strings(item, f"{path}.{key}" if path else str(key)) - elif isinstance(value, list): - for index, item in enumerate(value): - yield from _strings(item, f"{path}[{index}]") + if not stored(path): + yield f"{where}: {path}: {why}; leave it out, for whatever runs the task to give it" def _local_url(text: str) -> str | None: diff --git a/tst/unit/bundle/importer_entities_test.py b/tst/unit/bundle/importer_entities_test.py index 64bc0a58a..dec45446f 100644 --- a/tst/unit/bundle/importer_entities_test.py +++ b/tst/unit/bundle/importer_entities_test.py @@ -2,6 +2,7 @@ stored build context, downloading a context stored as one object once; and what a folder can't hold fails the import, unless --include-scope user leaves it in the store, pinned.""" +import hashlib import json import os import tomllib @@ -19,11 +20,12 @@ from agent_env.artifact.artifacts.skill import SkillArtifact from agent_env.artifact.store import get_artifact_store from agent_env.bundle import BundleError +from agent_env.bundle._import_report import store_fingerprint from agent_env.bundle.importer import import_task from agent_env.bundle.resolve import build_step from agent_env.cli import cli from agent_env.config import get_config -from agent_env.config.runtime import Config +from agent_env.config.runtime import Config, reset_config from agent_env.entity.scopes import FieldScope from agent_env.env.envs.mcp_server import MCPServerEnv from agent_env.env.envs.website import WebsiteEnv @@ -83,7 +85,7 @@ def test_with_the_user_scope_it_stays_in_the_store_pinned_and_the_report_and_rea ("artifact", "crm-image", 1, "envs/crm: image")] readme = (imported.root / "README.md").read_text() assert "## Still needs the store" in readme and "except what *Still needs the store* lists" in readme - assert json.loads((imported.root / "provenance.json").read_text())["store"][0]["id"] == "crm-image" + assert json.loads((imported.root / "provenance.json").read_text())["pinned"][0]["id"] == "crm-image" def test_the_cli_says_up_front_that_the_folder_still_needs_the_store(home): @@ -274,14 +276,77 @@ def test_provenance_names_each_written_entity_and_image_by_store_id_and_version_ text = (imported.root / "provenance.json").read_text() provenance = json.loads(text) + assert provenance["schema"] == 2 assert provenance["from"] == {"task": "t", "version": 1} + assert provenance["tasks"] == [{"file": "tasks/t.json", "id": "t", "version": 1}] and provenance["eval"] is None assert provenance["written"] == [{"folder": "envs/crm", "kind": "env", "type": "mcp_server", "id": "crm", "version": 1}] assert provenance["images"] == [{"folder": "envs/crm", "key": "image", "id": "crm-image", "version": 1, "dockerfile": "Dockerfile", "dockerfile_found": False}] + assert provenance["store"] == store_fingerprint() and provenance["store"].startswith("sha256:") assert "file://" not in text and str(home) not in text +def test_provenance_holds_each_file_the_import_wrote_as_written_so_an_edit_can_be_told(home): + """Every file but the README and provenance itself, by its path in the folder, as its content hash, +x when + executable.""" + MCPServerEnv.put(id="crm", docker_image_artifact=context_image(home, "crm-image", { + "Dockerfile": "FROM python:3.12-slim\nCOPY run.sh /\n", "run.sh": "#!/bin/sh\n"}), environment_name="crm") + _task([DEPLOY]) + + imported = import_task("t", home / "t") + + files = json.loads((imported.root / "provenance.json").read_text())["files"] + on_disk = {path.relative_to(imported.root).as_posix() for path in imported.root.rglob("*") if path.is_file()} + assert set(files) == on_disk - {"README.md", "provenance.json"} + assert {"tasks/t.json", "envs/crm/env.toml", "envs/crm/Dockerfile"} <= set(files) + for path, digest in files.items(): + executable = os.access(imported.root / path, os.X_OK) + expected = "sha256:" + hashlib.sha256((imported.root / path).read_bytes()).hexdigest() + assert digest == expected + ("+x" if executable else "") + + +def test_the_store_fingerprint_tells_one_configured_document_store_from_another(home, tmp_path, monkeypatch): + here = store_fingerprint() + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "elsewhere")) + + assert store_fingerprint() != here and store_fingerprint() == store_fingerprint() + + +def test_the_store_fingerprint_is_where_the_store_is_whatever_its_credentials_and_tuning(tmp_path, monkeypatch): + """Two stages' configs can hold the same text and reach different stores through what their secrets resolve to.""" + config = tmp_path / "fingerprint.toml" + monkeypatch.setenv("AGENT_ENV_CONFIG", str(config)) + monkeypatch.delenv("AGENT_ENV_DOCUMENT_STORE", raising=False) + monkeypatch.delenv("AGENT_ENV_SECRET_STORE", raising=False) + + def fingerprint(database, password, extra=""): + # Not a credential: the password a URL's user info holds in this test, joined in at run time. + uri = "mongodb://reader:" + password + "@db.example.com:27017/?retryWrites=true" + config.write_text(f'''[stores.document] +impl = "pkg:Store" +[stores.document.config] +uri = "secret:mongodb_uri" +database = "secret:mongodb_name" +{extra} +[stores.secret] +impl = "agent_env.store.secret_store:LocalSecretStore" +[stores.secret.config] +use_env = false +values = {{ mongodb_uri = "{uri}", mongodb_name = "{database}" }} +''') + reset_config() + return store_fingerprint() + + prod = fingerprint("agent_env", "pw-one") + dev = fingerprint("agent_env_dev", "pw-one") + other_password = fingerprint("agent_env", "pw-two") + tuned = fingerprint("agent_env", "pw-one", "connect_timeout_ms = 9000\npassword = \"not-a-password\"") + + assert prod != dev + assert prod == other_password == tuned + + def _saved_image(path, url, files): """A one-layer ``docker save`` tarball holding ``files``, stored at ``url``.""" path.write_bytes(saved_image([files], gz=True)) diff --git a/tst/unit/bundle/importer_test.py b/tst/unit/bundle/importer_test.py index e9ce9ba7c..3c7307857 100644 --- a/tst/unit/bundle/importer_test.py +++ b/tst/unit/bundle/importer_test.py @@ -288,6 +288,12 @@ def test_an_eval_imports_with_each_task_it_names_once_under_names_taken_from_the assert imported.eval == ImportedEval("evals/regression.toml", "nightly/regression", 1) plan = plan_bundle(resolve_bundle(parse_bundle(imported.root))) assert [entry.entry.name for entry in plan.evals] == ["regression"] + provenance = json.loads((imported.root / "provenance.json").read_text()) + assert provenance["eval"] == {"file": "evals/regression.toml", "id": "nightly/regression", "version": 1} + assert provenance["tasks"] == [{"file": "tasks/run.json", "id": "team/run", "version": 1}, + {"file": "tasks/run-2.json", "id": "other/run", "version": 1}, + {"file": "tasks/RUN-3.json", "id": "RUN", "version": 1}, + {"file": "tasks/triage.json", "id": "triage", "version": 1}] def test_an_eval_naming_a_task_the_store_doesnt_hold_is_refused_and_writes_nothing(store): diff --git a/tst/unit/bundle/register_imports_test.py b/tst/unit/bundle/register_imports_test.py new file mode 100644 index 000000000..129ad217d --- /dev/null +++ b/tst/unit/bundle/register_imports_test.py @@ -0,0 +1,439 @@ +"""Registering a folder ``bundle init`` wrote back into the store it came from: each entity under the id it was +imported from, nothing written while the folder holds it as imported, an edit re-versioning only what it reaches, and +the user-scoped fields the import left out given back from the stored task.""" + +import json + +import pytest + +from agent_env.a2a_agent import A2AAgent +from agent_env.artifact import Artifact, FileArtifact +from agent_env.artifact.artifacts.docker_image import DockerImageArtifact +from agent_env.artifact.store import get_artifact_store +from agent_env.bundle import BundleError +from agent_env.bundle import _import_report as import_report_module +from agent_env.bundle import _imported as imported_module +from agent_env.bundle import materialize as materialize_module +from agent_env.bundle import register as register_module +from agent_env.bundle._imported import Imported, ImportedImage, Source, source_of +from agent_env.bundle.importer import import_task +from agent_env.bundle.installed import InstalledBundle +from agent_env.bundle.parse import parse_bundle +from agent_env.bundle.plan import plan_bundle +from agent_env.bundle.register import REGISTRATIONS, register_bundle +from agent_env.bundle.resolve import build_step, resolve_bundle +from agent_env.config import get_config +from agent_env.entity.scopes import FieldScope +from agent_env.env.envs.mcp_server import MCPServerEnv +from agent_env.store import Filter +from agent_env.task import Task +from tst.unit.bundle._support import DIGEST, context_image, layout + +STEPS = [ + {"id": "env", "type": "deploy_env", "env_id": "crm-env"}, + {"id": "load", "type": "load_artifact", "env_id": "crm-env", "artifacts": [{"id": "crm-data"}]}, + {"id": "agent", "type": "deploy_agent", "env_ids": ["crm-env"], "a2a_agent_id": "solver", + "env_vars": {"API_KEY": "abc", "MODE": "fast"}}, +] + + +@pytest.fixture(autouse=True) +def outside_git(monkeypatch): + monkeypatch.setattr(register_module, "_git", lambda root, *args: None) + + +@pytest.fixture +def imported(local_stores, tmp_path, monkeypatch): + """The store's task triage, which names crm-env v1, crm-data v1 and the agent solver, imported into a folder.""" + monkeypatch.setenv("HOME", str(tmp_path)) + MCPServerEnv.put(id="crm-env", docker_image_artifact=context_image(tmp_path, "crm-image"), environment_name="crm") + A2AAgent.put(id="solver", docker_image_artifact=context_image(tmp_path, "solver-image")) + (tmp_path / "crm.json").write_text("{}") + FileArtifact.put("crm-data", description="crm seed", file_path=str(tmp_path / "crm.json")) + Task.put(id="triage", steps=[build_step(dict(step)) for step in STEPS]) + return import_task("triage", tmp_path / "triage").root + + +def _counts(): + documents = get_config().get_document_store() + return {name: documents.count(name, Filter()) for name in ("tasks", "envs", "a2a_agents", "artifacts")} + + +def _written(registration): + return {done.write.id: (done.version, done.reused) for done in registration.materialization.writes} + + +def test_a_folder_registered_as_it_was_imported_writes_nothing_and_keeps_each_version(imported): + before = _counts() + + registered = register_bundle(imported) + + assert _written(registered) == {"crm-image": (1, True), "crm-env": (1, True), "crm-data": (1, True), + "triage": (1, True)} + assert _counts() == before + rows = get_config().get_document_store().query(REGISTRATIONS, Filter.of(status="done")) + assert {(row["id"], row["version"]) for row in rows} == {("crm-image", 1), ("crm-env", 1), ("crm-data", 1), + ("triage", 1)} + (imported / "envs/crm-env/.DS_Store").write_bytes(b"left by Finder") + assert _written(register_bundle(imported)) == _written(registered) + assert len(get_config().get_document_store().query(REGISTRATIONS, Filter())) == len(rows) + + +def test_an_edit_rewrites_only_what_it_reaches_and_gives_back_what_the_import_left_out(imported): + (imported / "artifacts/crm-data/crm.json").write_text('{"edited": true}') + + registered = register_bundle(imported) + + assert _written(registered) == {"crm-image": (1, True), "crm-env": (1, True), "crm-data": (2, False), + "triage": (2, False)} + steps = Task.get("triage", 2).to_dict()["steps"] + assert (steps[0]["env_id"], steps[0]["env_version"]) == ("crm-env", 1) + assert steps[1]["artifacts"] == [{"id": "crm-data", "version": 2}] + assert (steps[2]["a2a_agent_id"], steps[2]["a2a_agent_version"]) == ("solver", 1) + assert steps[2]["env_vars"] == {"API_KEY": "abc", "MODE": "fast"} + assert Artifact.get("crm-data", 2).description == "crm seed" + + +def test_a_given_back_field_the_store_holds_is_no_credential_written_out_to_a_shared_store(imported, monkeypatch): + monkeypatch.setattr(register_module, "shared_store", lambda store: True) + (imported / "artifacts/crm-data/crm.json").write_text('{"edited": true}') + + registered = register_bundle(imported) + + assert _written(registered)["triage"] == (2, False) + + +def test_an_untouched_entry_keeps_its_imported_version_and_an_edit_writes_past_what_the_store_has_since(imported, + tmp_path): + FileArtifact.put("crm-data", description="someone else's", file_path=str(tmp_path / "crm.json")) + kept = register_bundle(imported) + (imported / "artifacts/crm-data/crm.json").write_text('{"edited": true}') + + edited = register_bundle(imported) + + assert _written(kept)["crm-data"] == (1, True) + assert _written(edited)["crm-data"] == (3, False) + + +def test_an_image_whose_build_context_is_edited_is_pushed_and_its_env_rewritten(imported, monkeypatch): + monkeypatch.setattr(materialize_module, "push_context_image", + lambda entity_id, context, dockerfile, before=None: (f"registry.example.com/{entity_id}:x", + DIGEST)) + dockerfile = imported / "envs/crm-env/Dockerfile" + dockerfile.write_text(dockerfile.read_text() + "ENV EDITED=1\n") + + registered = register_bundle(imported) + + assert _written(registered) == {"crm-image": (2, False), "crm-env": (2, False), "crm-data": (1, True), + "triage": (2, False)} + + +def test_a_folder_imported_from_another_store_registers_as_a_new_bundle(imported, monkeypatch): + monkeypatch.setattr(imported_module, "store_fingerprint", lambda: "sha256:another-store") + said = [] + + registered = register_bundle(imported, dry_run=True, on_progress=said.append) + + assert " it was imported from another store, so its entries take new ids" in said + assert set(_written(registered)) == {"triage-crm-env__env_image", "triage-crm-env", "triage-crm-data", + "triage-triage"} + + +def test_a_record_from_an_older_agent_env_registers_the_folder_as_a_new_bundle(imported): + record = json.loads((imported / "provenance.json").read_text()) + (imported / "provenance.json").write_text(json.dumps({**record, "schema": 1})) + said = [] + + registered = register_bundle(imported, dry_run=True, on_progress=said.append) + + assert " provenance.json is from an older agent-env, so the folder registers as a new bundle" in said + assert set(_written(registered)) == {"triage-crm-env__env_image", "triage-crm-env", "triage-crm-data", + "triage-triage"} + + +def test_an_imported_task_with_an_output_keeps_the_output_s_name(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + MCPServerEnv.put(id="crm-env", docker_image_artifact=context_image(tmp_path, "crm-image"), environment_name="crm") + Task.put(id="cli", steps=[build_step({"id": "env", "type": "deploy_env", "env_id": "crm-env"}), + build_step({"id": "cli", "type": "build_mcp_cli", "env_id": "crm-env", + "command_name": "crm", "cli_artifact_id": "crm-cli"})]) + folder = import_task("cli", tmp_path / "cli").root + (folder / "tasks/cli.json").write_text((folder / "tasks/cli.json").read_text().replace('"crm"', '"crm2"')) + + registered = register_bundle(folder) + + assert Task.get("cli", 2).to_dict()["steps"][1]["cli_artifact_id"] == "crm-cli" + assert _written(registered)["cli"] == (2, False) + + +def _pushed(monkeypatch): + monkeypatch.setattr(materialize_module, "push_context_image", + lambda entity_id, context, dockerfile, before=None: (f"registry.example.com/{entity_id}:x", + DIGEST)) + + +@pytest.fixture +def imported_with_user(local_stores, tmp_path, monkeypatch): + """The store's task triage imported with its user-scoped fields: the agent solver written into the folder and named + there, and the secret env var dropped.""" + monkeypatch.setenv("HOME", str(tmp_path)) + MCPServerEnv.put(id="crm-env", docker_image_artifact=context_image(tmp_path, "crm-image"), environment_name="crm") + A2AAgent.put(id="solver", docker_image_artifact=context_image(tmp_path, "solver-image")) + (tmp_path / "crm.json").write_text("{}") + FileArtifact.put("crm-data", description="crm seed", file_path=str(tmp_path / "crm.json")) + Task.put(id="triage", steps=[build_step(dict(step)) for step in STEPS]) + return import_task("triage", tmp_path / "triage", include_scopes={FieldScope.USER}).root + + +def test_a_folder_imported_with_its_user_fields_registers_as_imported(imported_with_user): + step = json.loads((imported_with_user / "tasks/triage.json").read_text())[2] + assert (step["a2a_agent_id"], step.get("a2a_agent_version"), step["env_vars"]) == ("solver", None, {"MODE": "fast"}) + + registered = register_bundle(imported_with_user) + + assert all(reused for _, reused in _written(registered).values()) + + +def test_an_edit_keeps_the_env_vars_the_import_dropped_and_they_are_no_credential_written_out(imported_with_user, + monkeypatch): + monkeypatch.setattr(register_module, "shared_store", lambda store: True) + (imported_with_user / "artifacts/crm-data/crm.json").write_text('{"edited": true}') + + registered = register_bundle(imported_with_user) + + assert _written(registered)["triage"] == (2, False) + assert Task.get("triage", 2).to_dict()["steps"][2]["env_vars"] == {"API_KEY": "abc", "MODE": "fast"} + + +def test_an_imported_skill_takes_back_the_source_the_import_left_out_unless_it_was_edited(local_stores, tmp_path, + monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + notes = {"name": "notes", "description": "how notes are kept", "object_url": "s3://their-bucket/skills/notes/"} + Task.put(id="skilled", steps=[build_step({"id": "skills", "type": "add_skills", "agent_name": "default-agent", + "skills": [notes]})]) + folder = import_task("skilled", tmp_path / "skilled").root + task_file = folder / "tasks/skilled.json" + task_file.write_text(task_file.read_text().replace('"default-agent"', '"helper"')) + + registered = register_bundle(folder) + task_file.write_text(task_file.read_text().replace("how notes are kept", "how notes are filed")) + with pytest.raises(BundleError) as refused: + register_bundle(folder) + + assert _written(registered)["skilled"] == (2, False) + skill = Task.get("skilled", 2).to_dict()["steps"][0]["skills"][0] + assert (skill["description"], skill["object_url"]) == ("how notes are kept", notes["object_url"]) + assert "skill 'notes' names no source" in refused.value.problems[0] + + +def test_envs_sharing_one_stored_image_register_as_imported_and_an_edit_to_one_copy_is_refused(local_stores, tmp_path, + monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + image = context_image(tmp_path, "shared-image") + MCPServerEnv.put(id="crm-env", docker_image_artifact=image, environment_name="crm") + MCPServerEnv.put(id="mail-env", docker_image_artifact=image, environment_name="mail") + Task.put(id="pair", steps=[build_step({"id": "crm", "type": "deploy_env", "env_id": "crm-env"}), + build_step({"id": "mail", "type": "deploy_env", "env_id": "mail-env"})]) + folder = import_task("pair", tmp_path / "pair").root + + kept = register_bundle(folder) + dockerfile = folder / "envs/mail-env/Dockerfile" + dockerfile.write_text(dockerfile.read_text() + "ENV EDITED=1\n") + with pytest.raises(BundleError) as refused: + register_bundle(folder) + + assert all(reused for _, reused in _written(kept).values()) + assert refused.value.problems == ("envs/crm-env, envs/mail-env hold copies of 'shared-image' as imported, and " + "envs/mail-env changed it; a registration writes each id once, so undo the change",) + + +def test_an_image_whose_toml_names_another_dockerfile_is_pushed(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + _pushed(monkeypatch) + files = {"Dockerfile": "FROM python:3.12-slim\nCOPY . /app\n", "app.py": "print(1)\n", + "Dockerfile.slim": "FROM python:3.12-alpine\nCOPY . /app\n"} + MCPServerEnv.put(id="crm-env", docker_image_artifact=context_image(tmp_path, "crm-image", files), + environment_name="crm") + Task.put(id="one", steps=[build_step({"id": "env", "type": "deploy_env", "env_id": "crm-env"})]) + folder = import_task("one", tmp_path / "one").root + toml = folder / "envs/crm-env/env.toml" + assert 'dockerfile = "Dockerfile"' in toml.read_text() + toml.write_text(toml.read_text().replace('dockerfile = "Dockerfile"', 'dockerfile = "Dockerfile.slim"')) + + registered = register_bundle(folder) + + assert _written(registered)["crm-image"] == (2, False) + + +def test_an_imported_task_and_an_env_of_one_name_but_two_ids_keep_their_own(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + MCPServerEnv.put(id="crm", docker_image_artifact=context_image(tmp_path, "crm-image"), environment_name="crm") + Task.put(id="ops/crm", steps=[build_step({"id": "env", "type": "deploy_env", "env_id": "crm"})]) + folder = import_task("ops/crm", tmp_path / "ops").root + + registered = register_bundle(folder) + (folder / "envs/crm-2").rename(folder / "envs/crm") + task_file = folder / "tasks/crm.json" + task_file.write_text(task_file.read_text().replace('"crm-2"', '"crm"')) + with pytest.raises(BundleError) as refused: + register_bundle(folder) + + assert _written(registered) == {"crm-image": (1, True), "crm": (1, True), "ops/crm": (1, True)} + assert refused.value.problems == ("tasks/crm.json: shares its name with what's registered as 'ops-crm', so it " + "can't keep the id it was imported as, 'ops/crm'; rename one",) + + +def test_an_imported_task_s_output_under_a_folder_keeps_its_name(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + MCPServerEnv.put(id="crm-env", docker_image_artifact=context_image(tmp_path, "crm-image"), environment_name="crm") + Task.put(id="cli", steps=[build_step({"id": "env", "type": "deploy_env", "env_id": "crm-env"}), + build_step({"id": "cli", "type": "build_mcp_cli", "env_id": "crm-env", + "command_name": "crm", "cli_artifact_id": "tools/crm-cli"})]) + folder = import_task("cli", tmp_path / "cli").root + (folder / "tasks/cli.json").write_text((folder / "tasks/cli.json").read_text().replace('"crm"', '"crm2"')) + + registered = register_bundle(folder) + + assert Task.get("cli", 2).to_dict()["steps"][1]["cli_artifact_id"] == "tools/crm-cli" + assert _written(registered)["cli"] == (2, False) + + +def test_registering_reads_only_the_files_of_what_the_import_wrote(imported, monkeypatch): + read = [] + hash_file = import_report_module.file_sha256 + monkeypatch.setattr(import_report_module, "file_sha256", lambda path: read.append(path) or hash_file(path)) + layout(imported, {".venv/lib/big.py": "x = 1\n", "notes/todo.md": "later\n"}) + + register_bundle(imported) + + assert read and not [path for path in read if ".venv" in path.parts or "notes" in path.parts] + + +def test_an_installed_bundle_s_agent_takes_no_source_from_the_folder_s_record(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + kit_root = layout(tmp_path / "site" / "kit", {"agents/solver/Dockerfile": "FROM scratch\n"}) + kit = InstalledBundle("kit", "demo-dist", "1.0", "demo", kit_root, None, None) + folder = layout(tmp_path / "triage", {"tasks/t.json": json.dumps([ + {"id": "agent", "type": "deploy_agent", "a2a_agent_id": "@local/demo-dist/kit/solver"}])}) + plan = plan_bundle(resolve_bundle(parse_bundle(folder)), installed=lambda: (kit,)) + record = Imported({"agents/solver": Source("solver", 3)}, + {("agents/solver", "image"): ImportedImage(Source("solver-image", 3))}, {}) + + installed = [write for write in plan.writes if write.source.entry in plan.origins] + + assert installed and [source_of(record, plan, write) for write in installed] == [None] * len(installed) + + +def test_an_env_whose_image_is_pulled_from_its_registry_registers_as_imported(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + image = get_artifact_store().put_document(DockerImageArtifact( + id="crm-image", description="crm", image_name="registry.example.com/crm:v1", + tar_gz_s3_url="file:///nowhere/crm-image.tar.gz")) + MCPServerEnv.put(id="crm-env", docker_image_artifact=image, environment_name="crm") + Task.put(id="pulled", steps=[build_step({"id": "env", "type": "deploy_env", "env_id": "crm-env"})]) + folder = import_task("pulled", tmp_path / "pulled").root + + registered = register_bundle(folder) + + assert _written(registered) == {"crm-image": (1, True), "crm-env": (1, True), "pulled": (1, True)} + + +@pytest.mark.parametrize("source", [{"object_url": "s3://their-bucket/skills/notes-v2/"}, + {"body": "Keep notes in notes.md."}]) +def test_a_skill_the_folder_gives_a_source_of_its_own_keeps_it(local_stores, tmp_path, monkeypatch, source): + monkeypatch.setenv("HOME", str(tmp_path)) + notes = {"name": "notes", "description": "how notes are kept", "object_url": "s3://their-bucket/skills/notes/"} + Task.put(id="skilled", steps=[build_step({"id": "skills", "type": "add_skills", "agent_name": "default-agent", + "skills": [notes]})]) + folder = import_task("skilled", tmp_path / "skilled").root + task_file = folder / "tasks/skilled.json" + steps = json.loads(task_file.read_text()) + steps[0]["skills"][0].update(source) + task_file.write_text(json.dumps(steps)) + + register_bundle(folder) + + skill = Task.get("skilled", 2).to_dict()["steps"][0]["skills"][0] + assert {key: skill.get(key) for key in ("object_url", "s3_url", "body")} == { + "object_url": source.get("object_url"), "s3_url": source.get("object_url"), "body": source.get("body")} + + +def test_a_link_into_an_imported_image_s_folder_is_an_edit_to_its_image(local_stores, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + _pushed(monkeypatch) + files = {"Dockerfile": "FROM python:3.12-slim\nCOPY . /app\n", "app.py": "print(1)\n"} + MCPServerEnv.put(id="crm-env", docker_image_artifact=context_image(tmp_path, "crm-image", files), + environment_name="crm") + Task.put(id="one", steps=[build_step({"id": "env", "type": "deploy_env", "env_id": "crm-env"})]) + folder = import_task("one", tmp_path / "one").root + (folder / "envs/crm-env/tasks").symlink_to("../../tasks", target_is_directory=True) + + registered = register_bundle(folder) + + assert _written(registered)["crm-image"] == (2, False) + + +def _downloads(tmp_path, monkeypatch, names): + """The stored task fetch, downloading one file for each of ``names`` from that user's URL, which the import masks + to the same text, imported with its user-scoped fields; the folder, and the urls the task stores.""" + monkeypatch.setenv("HOME", str(tmp_path)) + # Not credentials: users' URLs for one file, joined in at run time. + urls = [{"url": "https://" + name + ":" + "not-a-token@data.example.com/rows.csv", "filename": f"{name}.csv"} + for name in names] + Task.put(id="fetch", steps=[build_step({"id": "box", "type": "deploy_sandbox", "sandbox_name": "box", + "sandbox_mode": "vm"}), + build_step({"id": "fetch", "type": "load_artifact", "sandbox_name": "box", + "urls": urls, "destination_path": "/data"})]) + return import_task("fetch", tmp_path / "fetch", include_scopes={FieldScope.USER}).root, urls + + +def _edit_urls(folder, edit): + task_file = folder / "tasks/fetch.json" + steps = json.loads(task_file.read_text()) + edit(steps[1]["urls"]) + task_file.write_text(json.dumps(steps)) + + +def test_reordered_downloads_each_take_back_their_own_credentials(local_stores, tmp_path, monkeypatch): + folder, urls = _downloads(tmp_path, monkeypatch, ["ann", "bob"]) + _edit_urls(folder, list.reverse) + + register_bundle(folder) + + assert Task.get("fetch", 2).to_dict()["steps"][1]["urls"] == list(reversed(urls)) + + +def test_a_renamed_download_takes_back_its_credentials(local_stores, tmp_path, monkeypatch): + folder, urls = _downloads(tmp_path, monkeypatch, ["ann"]) + _edit_urls(folder, lambda entries: entries[0].update(filename="rows.csv")) + + register_bundle(folder) + + assert Task.get("fetch", 2).to_dict()["steps"][1]["urls"] == [{"url": urls[0]["url"], "filename": "rows.csv"}] + + +def test_a_masked_value_that_cant_be_told_from_another_is_refused_rather_than_written(local_stores, tmp_path, + monkeypatch): + folder, _ = _downloads(tmp_path, monkeypatch, ["ann", "bob"]) + _edit_urls(folder, lambda entries: [entry.update(filename=f"renamed-{index}.csv") + for index, entry in enumerate(entries)]) + + with pytest.raises(BundleError) as refused: + register_bundle(folder) + + assert refused.value.problems == tuple( + f"tasks/fetch.json: step 'fetch': urls[{index}].url still holds what the import masked, and the folder " + "changed what it was masked from; set its real value" for index in (0, 1)) + assert Task.get("fetch").version == 1 + + +def test_a_masked_value_in_a_task_not_registered_refuses_nothing(local_stores, tmp_path, monkeypatch): + folder, _ = _downloads(tmp_path, monkeypatch, ["ann", "bob"]) + _edit_urls(folder, lambda entries: [entry.update(filename=f"renamed-{index}.csv") + for index, entry in enumerate(entries)]) + (folder / "tasks/box.json").write_text(json.dumps([{"id": "box", "type": "deploy_sandbox", "sandbox_name": "box", + "sandbox_mode": "vm"}])) + + registered = register_bundle(folder, tasks=["box"]) + + assert _written(registered) == {"fetch-box": (1, False)}