feat: a recipe for every archive, and its build

recipes/archives.json records each of the 2,222 archives database.json lists, .variants/ copies included: members in order with size, crc32 and sha1, the container sha1, md5 and size, exact when the TorrentZip build gives its bytes (1,790) and canonical otherwise (432), the profile and DAT recipes agreeing on its members, and the platforms or profiles pinning its bytes.

romset_recipes.py --capture writes it (pipeline step 1c), --build and --build-all rebuild from held members, loose first, then another archive, then its own copy, read back and checked before the move into place. Under zlib-ng nothing is claimed exact. A test fails when an archive has no recipe or a member is held nowhere.

scripts/archive_build.py

@@ -0,0 +1,537 @@
+"""A recipe for every archive the collection holds, and the build that uses it.
+
+The collection must be able to regenerate each archive it serves from the
+ROMs it holds plus a committed recipe, without any copy of that archive
+from elsewhere. ``recipes/archives.json`` is that recipe list: one record
+per archive in ``database.json``, keyed by its path.
+
+A record names the members in archive order, each with its size, crc32 and
+sha1, and the container's sha1, md5 and size. ``rebuild`` says what the
+build reproduces:
+
+``exact``
+    The archive is TorrentZip and ``build_torrentzip()`` over its members
+    gives its bytes. Proven at capture by rebuilding it.
+``canonical``
+    Same members, other container bytes: the archive was written by another
+    tool. The build writes the TorrentZip form and checks members only.
+
+``sources`` lists the recipes that agree on the member list: profile
+``contents:`` blocks and the MAME and FBNeo DAT snapshots under
+``recipes/``, then ``collection``, the archive itself. A DAT is provenance,
+never the only source. ``bytes_pinned`` marks an archive whose container
+hash a platform or a profile pins, with each pin named.
+
+The build reads each member by sha1: a loose file first, then a member of
+another archive, then the archive's own copy, and verifies what it wrote
+before moving it into place.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+import re
+import sys
+import zipfile
+from collections import defaultdict
+from collections.abc import Callable, Iterable
+from concurrent.futures import ProcessPoolExecutor
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import BinaryIO
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from torrentzip import write_torrentzip, zlib_ng_version  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+ARCHIVES_FILE = REPO_ROOT / "recipes" / "archives.json"
+# The writer contract: build_torrentzip's layout over stock zlib. A change to
+# the layout is a new builder name and a new proof of every record.
+BUILDER = "torrentzip-1"
+_ARCHIVE_PATH = re.compile(r"\.zip(\.[0-9a-f]{8})?$", re.IGNORECASE)
+_TORRENTZIP_COMMENT = b"TORRENTZIPPED-"
+
+
+def is_archive_path(path: str) -> bool:
+    """A ZIP by its name: ``x.zip`` or a ``.variants/`` copy ``x.zip.1a2b3c4d``."""
+    return bool(_ARCHIVE_PATH.search(path))
+
+
+def database_archives(db: dict) -> dict[str, dict]:
+    """Path -> database entry, for every archive the database lists."""
+    return {
+        entry["path"]: entry
+        for entry in (db.get("files") or {}).values()
+        if is_archive_path(entry.get("path", ""))
+    }
+
+
+# --- the ledger file --------------------------------------------------------
+
+
+def load_ledger(path: Path = ARCHIVES_FILE) -> dict[str, dict]:
+    """Path -> record, from ``recipes/archives.json``, empty when absent."""
+    if not path.is_file():
+        return {}
+    with path.open(encoding="utf-8") as handle:
+        data = json.load(handle)
+    return {record["path"]: record for record in data.get("archives", [])}
+
+
+def ledger_text(records: dict[str, dict]) -> str:
+    """One record per line, so a diff names the archive that changed."""
+    lines = [
+        json.dumps(records[path], ensure_ascii=False, separators=(",", ":"))
+        for path in sorted(records)
+    ]
+    body = ",\n".join(f"  {line}" for line in lines)
+    return f'{{"builder": "{BUILDER}", "archives": [\n{body}\n]}}\n'
+
+
+def write_ledger(records: dict[str, dict], path: Path = ARCHIVES_FILE) -> bool:
+    from artifacts import write_if_changed  # noqa: PLC0415
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    return write_if_changed(str(path), ledger_text(records))
+
+
+# --- capture ----------------------------------------------------------------
+
+
+class HashSink:
+    """A write target that hashes what it receives, and keeps it if asked."""
+
+    def __init__(self, handle: BinaryIO | None = None) -> None:
+        self.handle = handle
+        self.sha1 = hashlib.sha1()
+        self.md5 = hashlib.md5()
+        self.size = 0
+
+    def write(self, data: bytes) -> int:
+        self.sha1.update(data)
+        self.md5.update(data)
+        self.size += len(data)
+        if self.handle is not None:
+            self.handle.write(data)
+        return len(data)
+
+
+class _Member(io.RawIOBase):
+    """One member of a ZIP, open for reading, closing the archive with it."""
+
+    def __init__(self, archive: Path, name: str) -> None:
+        super().__init__()
+        self._zip = zipfile.ZipFile(archive)
+        self._stream = self._zip.open(name)
+
+    def readable(self) -> bool:
+        return True
+
+    def read(self, size: int = -1) -> bytes:
+        return self._stream.read(size)
+
+    def close(self) -> None:
+        if not self.closed:
+            self._stream.close()
+            self._zip.close()
+        super().close()
+
+
+def _opener(source: Path, member: str | None) -> Callable[[], BinaryIO]:
+    if member is None:
+        return lambda: source.open("rb")
+    return lambda: _Member(source, member)
+
+
+def _file_infos(path: Path) -> tuple[list[zipfile.ZipInfo], list[str], bool]:
+    with zipfile.ZipFile(path) as archive:
+        infos = archive.infolist()
+        torrentzip = archive.comment.startswith(_TORRENTZIP_COMMENT)
+    files = [info for info in infos if not info.is_dir()]
+    directories = [info.filename for info in infos if info.is_dir()]
+    return files, directories, torrentzip
+
+
+def _member_sha1(path: Path, name: str) -> str:
+    digest = hashlib.sha1()
+    with _Member(path, name) as stream:
+        while chunk := stream.read(16 * 1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def read_archive(path: Path, sha1: str) -> dict:
+    """Members, directories and rebuild class of the archive at *path*.
+
+    A TorrentZip archive is rebuilt from its own members: equal bytes make
+    the recipe ``exact``. A mismatch is rebuilt once more before it is
+    believed, because this host sometimes computes wrongly under load. Any
+    other archive is ``canonical`` and only its members are read.
+    """
+    files, directories, torrentzip = _file_infos(path)
+    rebuild = "canonical"
+    by_name: dict[str, str] = {}
+    if torrentzip:
+        for _attempt in range(2):
+            sink = HashSink()
+            written = write_torrentzip(
+                sink, [(info.filename, _opener(path, info.filename)) for info in files]
+            )
+            by_name = {member.name: member.sha1 for member in written}
+            if sink.sha1.hexdigest() == sha1:
+                rebuild = "exact"
+                break
+    else:
+        by_name = {info.filename: _member_sha1(path, info.filename) for info in files}
+    record: dict = {
+        "rebuild": rebuild,
+        "members": [
+            {
+                "name": info.filename,
+                "size": info.file_size,
+                "crc32": f"{info.CRC:08x}",
+                "sha1": by_name[info.filename],
+            }
+            for info in files
+        ],
+    }
+    if directories:
+        record["directories"] = directories
+    return record
+
+
+def _read_job(job: tuple[str, str]) -> tuple[str, dict | str]:
+    path, sha1 = job
+    try:
+        return path, read_archive(Path(path), sha1)
+    except (OSError, zipfile.BadZipFile, UnicodeEncodeError) as exc:
+        return path, f"{type(exc).__name__}: {exc}"
+
+
+def _hash_values(value: object) -> set[str]:
+    raw = value if isinstance(value, list) else str(value or "").split(",")
+    return {str(item).strip().lower() for item in raw if str(item).strip()}
+
+
+def container_pins(platforms_dir: str, emulators_dir: str) -> dict[str, set[str]]:
+    """Container hash -> who pins it: ``platform:<name>`` or ``profile:<name>``.
+
+    A ``zipped_file`` entry pins a ROM inside the archive, and a Recalbox
+    archive md5 is the composite of its members (``zip_digest: composite``,
+    Bios.cpp:64-68): neither pins the container bytes.
+    """
+    from common import (  # noqa: PLC0415
+        list_registered_platforms,
+        load_emulator_profiles,
+        load_platform_config,
+    )
+
+    pins: dict[str, set[str]] = defaultdict(set)
+    for platform in list_registered_platforms(platforms_dir):
+        config = load_platform_config(platform, platforms_dir)
+        for system in config.get("systems", {}).values():
+            for entry in system.get("files", []) or []:
+                if entry.get("zipped_file") or entry.get("zip_digest") == "composite":
+                    continue
+                for key in ("md5", "sha1", "alt_md5"):
+                    for value in _hash_values(entry.get(key)):
+                        pins[value].add(f"platform:{platform}")
+    for name, profile in load_emulator_profiles(emulators_dir, skip_aliases=False).items():
+        for entry in profile.get("files", []) or []:
+            for key in ("md5", "sha1", "sha256"):
+                for value in _hash_values(entry.get(key)):
+                    pins[value].add(f"profile:{name}")
+    return dict(pins)
+
+
+MemberList = tuple[tuple[str, str], ...]
+
+
+def recipe_index(emulators_dir: str, recipes_dir: str) -> dict[MemberList, set[str]]:
+    """Sorted (name, crc32) -> labels of the recipes with exactly those members.
+
+    Matching on the member list, not the archive name, credits a DAT that
+    knows the set under another name (``aga.zip`` is MAME's ``isa_aga``).
+    """
+    from common import load_emulator_profiles  # noqa: PLC0415
+    from romset_recipes import collect_recipes, load_dat_recipes  # noqa: PLC0415
+
+    index: dict[MemberList, set[str]] = defaultdict(set)
+    profiles = load_emulator_profiles(emulators_dir, skip_aliases=False)
+    for per_label in collect_recipes(profiles).values():
+        for label, recipe in per_label.items():
+            index[tuple(sorted(recipe))].add(f"profile:{label}")
+    for per_label in load_dat_recipes(recipes_dir).values():
+        for label, recipe in per_label.items():
+            index[tuple(sorted(recipe))].add(label)
+    return dict(index)
+
+
+def member_key(members: Iterable[dict]) -> MemberList:
+    return tuple(sorted((m["name"], m["crc32"]) for m in members))
+
+
+def capture(
+    db: dict,
+    previous: dict[str, dict],
+    index: dict[MemberList, set[str]],
+    pins: dict[str, set[str]],
+    workers: int = 4,
+    log: Callable[[str], None] = print,
+) -> tuple[dict[str, dict], dict[str, str]]:
+    """A record for every archive in *db*, and the archives that could not be read.
+
+    A previous record whose sha1 still names the file is kept as read: the
+    container bytes fix the members. Sources and pins are recomputed every
+    time, since profiles, snapshots and platform lists move.
+    """
+    if zlib_ng_version():
+        raise ZlibNgError(zlib_ng_version())
+    archives = database_archives(db)
+    reused: dict[str, dict] = {}
+    jobs: list[tuple[str, str]] = []
+    for path, entry in sorted(archives.items()):
+        old = previous.get(path)
+        if old and old.get("sha1") == entry["sha1"] and old.get("members"):
+            reused[path] = old
+        else:
+            jobs.append((path, entry["sha1"]))
+    read: dict[str, dict] = {}
+    failed: dict[str, str] = {}
+    if jobs:
+        log(f"  reading {len(jobs)} archives ({len(reused)} unchanged)")
+        with ProcessPoolExecutor(max_workers=workers) as pool:
+            for path, result in pool.map(_read_job, jobs, chunksize=4):
+                if isinstance(result, str):
+                    failed[path] = result
+                else:
+                    read[path] = result
+    records: dict[str, dict] = {}
+    for path, entry in sorted(archives.items()):
+        body = reused.get(path) or read.get(path)
+        if body is None:
+            continue
+        record = {
+            "path": path,
+            "sha1": entry["sha1"],
+            "md5": entry["md5"],
+            "size": entry["size"],
+            "rebuild": body["rebuild"],
+            "members": body["members"],
+        }
+        if body.get("directories"):
+            record["directories"] = body["directories"]
+        record["sources"] = sorted(index.get(member_key(body["members"]), set())) + [
+            "collection"
+        ]
+        pinned = set()
+        for value in (entry["md5"], entry["sha1"], entry.get("sha256", "")):
+            pinned |= pins.get(str(value).lower(), set())
+        if pinned:
+            record["bytes_pinned"] = True
+            record["pins"] = sorted(pinned)
+        records[path] = record
+    return records, failed
+
+
+class ZlibNgError(RuntimeError):
+    def __init__(self, version: str | None) -> None:
+        super().__init__(
+            f"this Python deflates with zlib-ng {version}: TorrentZip bytes are "
+            "stock zlib's level-9 output, so an exact rebuild cannot be proven "
+            "or claimed here. Run under a Python linked against stock zlib."
+        )
+
+
+# --- build ------------------------------------------------------------------
+
+
+@dataclass
+class Source:
+    path: Path
+    member: str | None  # None for a loose file
+
+
+@dataclass
+class AtomIndex:
+    """Where each member sha1 can be read in the collection."""
+
+    loose: dict[str, Path] = field(default_factory=dict)
+    inside: dict[str, list[tuple[Path, str]]] = field(default_factory=dict)
+
+    @classmethod
+    def from_collection(cls, ledger: dict[str, dict], db: dict, root: Path) -> AtomIndex:
+        index = cls()
+        for entry in (db.get("files") or {}).values():
+            path = root / entry.get("path", "")
+            if not is_archive_path(entry.get("path", "")) and path.is_file():
+                index.loose.setdefault(entry["sha1"], path)
+        for archive, record in sorted(ledger.items()):
+            path = root / archive
+            if not path.is_file():
+                continue
+            for member in record.get("members", []):
+                index.inside.setdefault(member["sha1"], []).append((path, member["name"]))
+        return index
+
+    def find(self, sha1: str, own: Path) -> Source | None:
+        """A loose copy first, then another archive, then the archive itself."""
+        if sha1 in self.loose:
+            return Source(self.loose[sha1], None)
+        holders = self.inside.get(sha1, [])
+        for path, member in holders:
+            if path != own:
+                return Source(path, member)
+        for path, member in holders:
+            return Source(path, member)
+        return None
+
+    def held(self, sha1: str) -> bool:
+        return sha1 in self.loose or sha1 in self.inside
+
+
+@dataclass
+class BuildResult:
+    path: str
+    status: str  # exact, canonical, impossible, failed
+    output: str = ""
+    detail: str = ""
+    missing: list[dict] = field(default_factory=list)
+    own_members: int = 0
+
+
+def _members_match(built: Path, members: list[dict]) -> str:
+    """Empty when *built* holds exactly *members*, else what differs."""
+    expected = sorted((m["name"], m["size"], m["crc32"], m["sha1"]) for m in members)
+    with zipfile.ZipFile(built) as archive:
+        names = [info.filename for info in archive.infolist() if not info.is_dir()]
+        sizes = {info.filename: info.file_size for info in archive.infolist()}
+        crcs = {info.filename: f"{info.CRC:08x}" for info in archive.infolist()}
+    held = sorted(
+        (name, sizes[name], crcs[name], _member_sha1(built, name)) for name in names
+    )
+    if held == expected:
+        return ""
+    missing = sorted(set(expected) - set(held))
+    extra = sorted(set(held) - set(expected))
+    return f"members differ: expected {missing[:3]}, built {extra[:3]}"
+
+
+def _file_sha1(path: Path) -> str:
+    digest = hashlib.sha1()
+    with path.open("rb") as handle:
+        while chunk := handle.read(16 * 1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def destination(record_path: str, output_dir: Path) -> Path:
+    """Where a build lands: the archive's path below ``bios/``, under *output_dir*."""
+    parts = Path(record_path).parts
+    relative = Path(*parts[1:]) if parts and parts[0] == "bios" else Path(record_path)
+    return output_dir / relative
+
+
+def build_archive(
+    record: dict,
+    atoms: AtomIndex,
+    output_dir: Path,
+    root: Path = REPO_ROOT,
+    attempts: int = 2,
+) -> BuildResult:
+    """Build *record* from held members into *output_dir*, verified.
+
+    The bytes are written to a scratch file beside the destination and read
+    back from disk: an exact recipe must give its recorded sha1, a canonical
+    one its members. A mismatch is built once more, then refused: nothing
+    wrong is ever moved to the destination. Under zlib-ng an exact recipe is
+    checked by its members only and never reported exact.
+    """
+    path = record["path"]
+    own = root / path
+    sources: list[tuple[str, Source]] = []
+    missing: list[dict] = []
+    for member in record["members"]:
+        source = atoms.find(member["sha1"], own)
+        if source is None:
+            missing.append(member)
+        else:
+            sources.append((member["name"], source))
+    if missing:
+        return BuildResult(path, "impossible", missing=missing)
+    own_members = sum(1 for _name, source in sources if source.path == own)
+    ng = zlib_ng_version()
+    exact = record["rebuild"] == "exact" and not ng
+    target = destination(path, output_dir)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    scratch = target.with_name(f".{target.name}.{os.getpid()}.part")
+    detail = ""
+    try:
+        for _attempt in range(attempts):
+            with scratch.open("wb") as handle:
+                write_torrentzip(
+                    handle,
+                    [(name, _opener(source.path, source.member)) for name, source in sources],
+                )
+            if exact:
+                written = _file_sha1(scratch)
+                detail = "" if written == record["sha1"] else (
+                    f"sha1 {written} is not the recorded {record['sha1']}"
+                )
+            else:
+                detail = _members_match(scratch, record["members"])
+            if not detail:
+                os.replace(scratch, target)
+                status = "exact" if exact else "canonical"
+                note = (
+                    f"zlib-ng {ng}: exact bytes not claimed, members verified"
+                    if ng and record["rebuild"] == "exact" else ""
+                )
+                return BuildResult(path, status, str(target), note, own_members=own_members)
+    finally:
+        if scratch.exists():
+            scratch.unlink()
+    return BuildResult(path, "failed", detail=detail, own_members=own_members)
+
+
+def select(ledger: dict[str, dict], wanted: str, root: Path = REPO_ROOT) -> list[dict]:
+    """Records named by a path (relative or absolute) or by a file name."""
+    candidate = Path(wanted)
+    if candidate.is_absolute():
+        try:
+            candidate = candidate.resolve().relative_to(root.resolve())
+        except ValueError:
+            return []
+    key = candidate.as_posix()
+    if key in ledger:
+        return [ledger[key]]
+    return [record for path, record in sorted(ledger.items()) if Path(path).name == wanted]
+
+
+def missing_members(ledger: dict[str, dict], db: dict) -> list[dict]:
+    """Members held nowhere: neither as a loose file nor inside any archive.
+
+    Read from the committed data alone, so it holds without the files: a
+    member is held when database.json has its sha1 as a file, or when the
+    recipe of an archive database.json lists carries it.
+    """
+    archives = database_archives(db)
+    loose = {
+        sha1 for sha1, entry in (db.get("files") or {}).items()
+        if not is_archive_path(entry.get("path", ""))
+    }
+    inside = {
+        member["sha1"]
+        for path, record in ledger.items() if path in archives
+        for member in record.get("members", [])
+    }
+    return [
+        {"archive": path, "member": member["name"], "sha1": member["sha1"]}
+        for path, record in sorted(ledger.items())
+        for member in record.get("members", [])
+        if member["sha1"] not in loose and member["sha1"] not in inside
+    ]

scripts/pipeline.py

@@ -5,7 +5,8 @@ Steps, numbered as the run prints them:
   0    validate_schemas.py          profiles and platform YAMLs, as PR CI checks them
   1    generate_db.py --force       rebuild database.json from bios/
   1b   provenance_report.py         dump-catalog coverage, derivable.json
-  1c   romset_recipes.py            archive identification, reconstruction targets
+  1c   romset_recipes.py            archive identification, reconstruction targets,
+                                    and --capture: recipes/archives.json
   2    refresh_data_dirs.py         Dolphin Sys, PPSSPP assets and the other trees
   2a   mame_hash_scraper            MAME BIOS hashes, and device sets on a new release
   2a2  fbneo_hash_scraper           FBNeo BIOS hashes
@@ -1216,6 +1217,14 @@ def main() -> None:
         [sys.executable, "scripts/romset_recipes.py"],
         STEP_LABELS["romset_recipes"],
     )
+    # A recipe for every archive database.json lists (recipes/archives.json):
+    # only an archive whose sha1 moved is read again, so the step costs the
+    # archives added since the last run.
+    captured, _ = run(
+        [sys.executable, "scripts/romset_recipes.py", "--capture"],
+        f"{STEP_LABELS['romset_recipes']} (archive recipes)",
+    )
+    ok = ok and captured
     results["romset_recipes"] = ok
     all_ok = all_ok and ok
 

scripts/romset_recipes.py

@@ -20,6 +20,24 @@ Two questions follow, and this module answers both:
 Reconstruction is only possible when the pinned archive is itself TorrentZip.
 An archive whose bytes carry metadata unrelated to its contents cannot be
 derived from ROMs by anyone, and is reported as such rather than guessed at.
+
+Every archive the collection holds also has a recipe of its own in
+``recipes/archives.json`` (scripts/archive_build.py):
+
+``--capture``
+    Write a record for every archive in database.json: members in order with
+    size, crc32 and sha1, container sha1, md5 and size, and whether the
+    TorrentZip build reproduces the bytes (``exact``) or only the members
+    (``canonical``). Archives already recorded under the same sha1 are not
+    read again.
+
+``--build NAME --output-dir DIR``
+    Build an archive from the members the collection holds, loose or inside
+    another archive, and verify it before moving it into place.
+
+``--build-all --output-dir DIR``
+    The same for every recorded archive, with one line per archive: exact,
+    canonical, or impossible with the member nothing holds.
 """
 
 from __future__ import annotations
@@ -43,6 +61,8 @@ from common import (
 )
 from torrentzip import build_torrentzip, is_torrentzip
 
+ARCHIVES_FILE = Path(__file__).resolve().parent.parent / "recipes" / "archives.json"
+
 Recipe = list[tuple[str, str]]
 
 
@@ -459,6 +479,95 @@ def print_missing(gap: dict) -> None:
         print(f"  NO RECIPE: {record['archive']} md5={record['md5'][:8]}")
 
 
+def run_capture(args: argparse.Namespace, db: dict) -> int:
+    """Write recipes/archives.json for every archive the database lists."""
+    import archive_build  # noqa: PLC0415
+
+    try:
+        records, failed = archive_build.capture(
+            db,
+            archive_build.load_ledger(Path(args.archives)),
+            archive_build.recipe_index(args.emulators_dir, args.dat_recipes),
+            archive_build.container_pins(args.platforms_dir, args.emulators_dir),
+        )
+    except archive_build.ZlibNgError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 2
+    for path, reason in sorted(failed.items()):
+        print(f"  UNREADABLE: {path}: {reason}", file=sys.stderr)
+    changed = archive_build.write_ledger(records, Path(args.archives))
+    by_class: dict[str, int] = defaultdict(int)
+    for record in records.values():
+        by_class[record["rebuild"]] += 1
+    agreed = sum(1 for record in records.values() if len(record["sources"]) > 1)
+    pinned = sum(1 for record in records.values() if record.get("bytes_pinned"))
+    print(
+        f"{len(records)} archive recipes: {by_class['exact']} exact, "
+        f"{by_class['canonical']} canonical; {agreed} agree with a profile or DAT "
+        f"recipe, {pinned} pinned as bytes"
+    )
+    print(f"{'Wrote' if changed else 'Unchanged'} {args.archives}")
+    return 1 if failed else 0
+
+
+def run_build(args: argparse.Namespace, db: dict) -> int:
+    """Build one archive, or all of them, from the members the collection holds."""
+    import archive_build  # noqa: PLC0415
+
+    if not args.output_dir:
+        print("ERROR: --build and --build-all need --output-dir", file=sys.stderr)
+        return 2
+    ledger = archive_build.load_ledger(Path(args.archives))
+    if args.build_all:
+        records = [ledger[path] for path in sorted(ledger)]
+    else:
+        records = archive_build.select(ledger, args.build)
+        if not records:
+            print(f"ERROR: no archive recipe named {args.build}", file=sys.stderr)
+            return 2
+    ng = archive_build.zlib_ng_version()
+    if ng:
+        print(
+            f"  zlib-ng {ng} in use: TorrentZip bytes are stock zlib's output, so no "
+            "build is claimed exact here; exact recipes are checked by their members"
+        )
+    root = Path(args.bios_dir).resolve().parent
+    atoms = archive_build.AtomIndex.from_collection(ledger, db, root)
+    counts: dict[str, int] = defaultdict(int)
+    own: list[tuple[str, int]] = []
+    for record in records:
+        result = archive_build.build_archive(record, atoms, Path(args.output_dir), root)
+        counts[result.status] += 1
+        if result.own_members:
+            own.append((result.path, result.own_members))
+        if result.status == "impossible":
+            for member in result.missing:
+                print(
+                    f"  IMPOSSIBLE: {result.path}: missing {member['name']} "
+                    f"({member['sha1']})"
+                )
+        elif result.status == "failed":
+            print(f"  FAILED: {result.path}: {result.detail}")
+        else:
+            note = f" ({result.detail})" if result.detail else ""
+            print(f"  {result.status.upper()}: {result.path}{note}")
+            if args.discard:
+                Path(result.output).unlink()
+    print(
+        f"\n{len(records)} archives: {counts['exact']} rebuilt exact, "
+        f"{counts['canonical']} canonical, {counts['impossible']} impossible, "
+        f"{counts['failed']} failed"
+    )
+    print(
+        f"  {len(own)} of them read {sum(n for _p, n in own)} members from their "
+        "own archive, held nowhere else in the collection"
+    )
+    if counts["failed"]:
+        print("ERROR: a build did not match its recipe and was not kept", file=sys.stderr)
+        return 1
+    return 1 if counts["impossible"] else 0
+
+
 def _resolve_modes(args: argparse.Namespace) -> None:
     """Which reports run: both by default, and --missing whenever --write.
 
@@ -508,11 +617,35 @@ def main() -> int:
         default="recipes",
         help="recipe snapshot file, or a directory of them",
     )
+    parser.add_argument(
+        "--archives", default=str(ARCHIVES_FILE),
+        help="archive recipe file (default: recipes/archives.json)",
+    )
+    parser.add_argument(
+        "--capture", action="store_true",
+        help="write a recipe for every archive in the database",
+    )
+    build = parser.add_mutually_exclusive_group()
+    build.add_argument("--build", metavar="ARCHIVE", help="build one archive, by path or name")
+    build.add_argument("--build-all", action="store_true", help="build every recorded archive")
+    parser.add_argument("--output-dir", help="where --build and --build-all write")
+    parser.add_argument(
+        "--discard", action="store_true",
+        help="delete each build once verified: a full proof without the disk space",
+    )
     args = parser.parse_args()
     require_directory(args.bios_dir)
-    _resolve_modes(args)
+    if (args.output_dir or args.discard) and not (args.build or args.build_all):
+        parser.error("--output-dir and --discard need --build or --build-all")
+    if args.capture and (args.build or args.build_all):
+        parser.error("--capture runs alone")
 
     db = load_database(args.db)
+    if args.capture:
+        return run_capture(args, db)
+    if args.build or args.build_all:
+        return run_build(args, db)
+    _resolve_modes(args)
     profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False)
     profile_recipes = collect_recipes(profiles)
     dat_recipes = load_dat_recipes(args.dat_recipes)

tests/test_archive_build.py

@@ -0,0 +1,349 @@
+"""Every archive has a recipe, every member is held, and the build reproduces it.
+
+The collection must regenerate each archive it serves from the ROMs it holds
+plus recipes/archives.json, without a copy of that archive from elsewhere.
+"""
+
+from __future__ import annotations
+
+import tests  # noqa: F401  offline, scratch under tmp/, the suite's own cache
+
+import hashlib
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+import zipfile
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parent.parent
+TMP_ROOT = ROOT / "tmp" / "tests"
+TMP_ROOT.mkdir(parents=True, exist_ok=True)
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import archive_build  # noqa: E402
+from torrentzip import build_torrentzip  # noqa: E402
+
+ROM_A = b"first rom payload" * 50
+ROM_B = b"second rom payload" * 40
+ROM_C = b"third rom, held only inside its own archive" * 30
+
+
+def _sha1(data: bytes) -> str:
+    return hashlib.sha1(data).hexdigest()
+
+
+class TheLedgerCoversTheCollection(unittest.TestCase):
+    """Read from the committed files: database.json and recipes/archives.json."""
+
+    @classmethod
+    def setUpClass(cls):
+        with (ROOT / "database.json").open(encoding="utf-8") as handle:
+            cls.db = json.load(handle)
+        cls.ledger = archive_build.load_ledger()
+        cls.archives = archive_build.database_archives(cls.db)
+
+    def test_every_database_archive_has_a_recipe(self):
+        absent = sorted(set(self.archives) - set(self.ledger))
+        self.assertEqual(
+            absent, [],
+            f"{len(absent)} archives without a recipe; run "
+            "`python scripts/romset_recipes.py --capture`",
+        )
+
+    def test_no_recipe_outlives_its_archive(self):
+        self.assertEqual(sorted(set(self.ledger) - set(self.archives)), [])
+
+    def test_each_recipe_names_the_container_the_database_holds(self):
+        wrong = [
+            path for path, record in self.ledger.items()
+            if path in self.archives and any(
+                record[key] != self.archives[path][key] for key in ("sha1", "md5", "size")
+            )
+        ]
+        self.assertEqual(wrong, [], "recaptured archives keep a stale recipe")
+
+    def test_each_recipe_is_complete(self):
+        for path, record in self.ledger.items():
+            self.assertIn(record["rebuild"], ("exact", "canonical"), path)
+            self.assertEqual(record["sources"][-1], "collection", path)
+            self.assertTrue(record["members"], path)
+            for member in record["members"]:
+                self.assertEqual(set(member), {"name", "size", "crc32", "sha1"}, path)
+            if record.get("bytes_pinned"):
+                self.assertTrue(record["pins"], path)
+
+    def test_every_member_is_held_in_the_collection(self):
+        """The list printed here is the acquisition list: a member neither held
+        as a loose file nor inside any archive the database lists."""
+        missing = archive_build.missing_members(self.ledger, self.db)
+        listing = "\n".join(
+            f"  {m['archive']}: {m['member']} ({m['sha1']})" for m in missing
+        )
+        self.assertEqual(missing, [], f"members held nowhere:\n{listing}")
+
+    def test_the_file_is_written_the_one_way(self):
+        text = archive_build.ARCHIVES_FILE.read_text(encoding="utf-8")
+        self.assertEqual(text, archive_build.ledger_text(self.ledger))
+
+    def test_recipes_describe_the_archives_on_disk(self):
+        """Names in order, sizes and crc32 against each central directory."""
+        checked = 0
+        for path, record in sorted(self.ledger.items()):
+            archive = ROOT / path
+            if not archive.is_file():
+                continue  # a release asset not restored in this checkout
+            with zipfile.ZipFile(archive) as handle:
+                held = [
+                    (info.filename, info.file_size, f"{info.CRC:08x}")
+                    for info in handle.infolist() if not info.is_dir()
+                ]
+            recorded = [(m["name"], m["size"], m["crc32"]) for m in record["members"]]
+            self.assertEqual(held, recorded, path)
+            checked += 1
+        if not checked:
+            self.skipTest("bios/ is not present")
+
+
+class Fixture(unittest.TestCase):
+    """A small collection: a loose ROM, a TorrentZip set, a plain ZIP."""
+
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(dir=TMP_ROOT)
+        self.root = Path(self.temp.name)
+        bios = self.root / "bios"
+        bios.mkdir()
+        (bios / "a.rom").write_bytes(ROM_A)
+        self.exact_path = bios / "set.zip"
+        self.exact_path.write_bytes(
+            build_torrentzip([("a.rom", ROM_A), ("b.rom", ROM_B), ("c.rom", ROM_C)])
+        )
+        self.other_path = bios / "other.zip"
+        with zipfile.ZipFile(self.other_path, "w", zipfile.ZIP_DEFLATED) as archive:
+            archive.writestr("dir/", b"")
+            archive.writestr("second.rom", ROM_B)
+            archive.writestr("first.rom", ROM_A)
+        self.db = {"files": {}}
+        for path in (bios / "a.rom", self.exact_path, self.other_path):
+            data = path.read_bytes()
+            self.db["files"][_sha1(data)] = {
+                "path": path.relative_to(self.root).as_posix(),
+                "name": path.name,
+                "sha1": _sha1(data),
+                "md5": hashlib.md5(data).hexdigest(),
+                "size": len(data),
+            }
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def capture(self, **kwargs) -> dict[str, dict]:
+        records, failed = archive_build.capture(
+            self.db, kwargs.get("previous", {}), kwargs.get("index", {}),
+            kwargs.get("pins", {}), workers=1, log=lambda _line: None,
+        )
+        self.assertEqual(failed, {})
+        return records
+
+
+class Capture(Fixture):
+    def setUp(self):
+        super().setUp()
+        self._cwd = Path.cwd()
+        import os
+
+        os.chdir(self.root)
+
+    def tearDown(self):
+        import os
+
+        os.chdir(self._cwd)
+        super().tearDown()
+
+    def test_a_torrentzip_archive_is_exact_and_another_canonical(self):
+        records = self.capture()
+        exact, other = records["bios/set.zip"], records["bios/other.zip"]
+        self.assertEqual(exact["rebuild"], "exact")
+        self.assertEqual(other["rebuild"], "canonical")
+        self.assertEqual(
+            [m["name"] for m in other["members"]], ["second.rom", "first.rom"],
+            "members keep the archive's own order",
+        )
+        self.assertEqual(other["directories"], ["dir/"])
+        self.assertEqual(other["members"][0]["sha1"], _sha1(ROM_B))
+        self.assertEqual(exact["sources"], ["collection"])
+
+    def test_dat_agreement_and_pins_are_recorded(self):
+        key = archive_build.member_key(
+            [{"name": n, "crc32": c} for n, c in (
+                ("a.rom", self._crc(ROM_A)), ("b.rom", self._crc(ROM_B)),
+                ("c.rom", self._crc(ROM_C)),
+            )]
+        )
+        md5 = hashlib.md5(self.exact_path.read_bytes()).hexdigest()
+        records = self.capture(
+            index={key: {"mame:MAME mame0289"}}, pins={md5: {"platform:demo"}}
+        )
+        record = records["bios/set.zip"]
+        self.assertEqual(record["sources"], ["mame:MAME mame0289", "collection"])
+        self.assertEqual((record["bytes_pinned"], record["pins"]), (True, ["platform:demo"]))
+        self.assertNotIn("bytes_pinned", records["bios/other.zip"])
+
+    def test_an_unchanged_archive_is_not_read_again(self):
+        previous = self.capture()
+        previous["bios/set.zip"]["members"][0]["sha1"] = "kept"
+        records = self.capture(previous=previous)
+        self.assertEqual(records["bios/set.zip"]["members"][0]["sha1"], "kept")
+
+    def test_zlib_ng_refuses_to_capture(self):
+        with mock.patch.object(archive_build, "zlib_ng_version", lambda: "2.2.4"):
+            with self.assertRaises(archive_build.ZlibNgError):
+                self.capture()
+
+    def test_composite_and_zipped_pins_name_no_container(self):
+        platforms = self.root / "platforms"
+        platforms.mkdir()
+        (platforms / "_registry.yml").write_text(
+            "platforms:\n"
+            "  plain:\n    config: plain.yml\n"
+            "  composite:\n    config: composite.yml\n    zip_digest: composite\n",
+            encoding="utf-8",
+        )
+        entry = "      - name: set.zip\n        destination: set.zip\n"
+        (platforms / "plain.yml").write_text(
+            "platform: Plain\nsystems:\n  arcade:\n    files:\n"
+            f"{entry}        md5: {'a' * 32}\n"
+            f"{entry}        md5: {'b' * 32}\n        zipped_file: a.rom\n",
+            encoding="utf-8",
+        )
+        (platforms / "composite.yml").write_text(
+            "platform: Composite\nsystems:\n  arcade:\n    files:\n"
+            f"{entry}        md5: {'c' * 32}\n",
+            encoding="utf-8",
+        )
+        emulators = self.root / "emulators"
+        emulators.mkdir()
+        (emulators / "core.yml").write_text(
+            "emulator: Core\ntype: libretro\nfiles:\n"
+            f"  - name: set.zip\n    sha1: {'d' * 40}\n",
+            encoding="utf-8",
+        )
+        pins = archive_build.container_pins(str(platforms), str(emulators))
+        self.assertEqual(pins, {"a" * 32: {"platform:plain"}, "d" * 40: {"profile:core"}})
+
+    @staticmethod
+    def _crc(data: bytes) -> str:
+        import zlib
+
+        return f"{zlib.crc32(data) & 0xFFFFFFFF:08x}"
+
+
+class Build(Fixture):
+    def setUp(self):
+        super().setUp()
+        import os
+
+        self._cwd = Path.cwd()
+        os.chdir(self.root)
+        self.ledger = self.capture()
+        os.chdir(self._cwd)
+        self.atoms = archive_build.AtomIndex.from_collection(self.ledger, self.db, self.root)
+        self.out = self.root / "out"
+
+    def build(self, path: str):
+        return archive_build.build_archive(self.ledger[path], self.atoms, self.out, self.root)
+
+    def test_an_exact_recipe_rebuilds_the_bytes(self):
+        result = self.build("bios/set.zip")
+        self.assertEqual(result.status, "exact")
+        built = self.out / "set.zip"
+        self.assertEqual(built.read_bytes(), self.exact_path.read_bytes())
+        self.assertEqual(result.own_members, 1, "c.rom is held only in set.zip")
+
+    def test_members_come_from_a_loose_file_then_another_archive(self):
+        own = self.root / "bios" / "set.zip"
+        self.assertEqual(self.atoms.find(_sha1(ROM_A), own).member, None)
+        self.assertEqual(self.atoms.find(_sha1(ROM_B), own).path.name, "other.zip")
+        self.assertEqual(self.atoms.find(_sha1(ROM_C), own).path, own)
+
+    def test_a_canonical_recipe_is_checked_by_its_members(self):
+        result = self.build("bios/other.zip")
+        self.assertEqual(result.status, "canonical")
+        with zipfile.ZipFile(self.out / "other.zip") as archive:
+            self.assertEqual(sorted(archive.namelist()), ["first.rom", "second.rom"])
+            self.assertTrue(archive.comment.startswith(b"TORRENTZIPPED-"))
+
+    def test_a_member_held_nowhere_makes_it_impossible(self):
+        record = dict(self.ledger["bios/set.zip"])
+        record["members"] = record["members"] + [
+            {"name": "gone.rom", "size": 1, "crc32": "00000000", "sha1": "f" * 40}
+        ]
+        result = archive_build.build_archive(record, self.atoms, self.out, self.root)
+        self.assertEqual(result.status, "impossible")
+        self.assertEqual([m["name"] for m in result.missing], ["gone.rom"])
+        self.assertFalse(self.out.exists())
+
+    def test_a_mismatch_is_never_promoted(self):
+        record = dict(self.ledger["bios/set.zip"], sha1="0" * 40)
+        result = archive_build.build_archive(record, self.atoms, self.out, self.root)
+        self.assertEqual(result.status, "failed")
+        self.assertIn("is not the recorded", result.detail)
+        self.assertEqual(list(self.out.rglob("*")), [], "no file, no scratch left")
+
+    def test_a_canonical_mismatch_is_never_promoted(self):
+        record = dict(self.ledger["bios/other.zip"])
+        record["members"] = [dict(m) for m in record["members"]]
+        record["members"][0]["crc32"] = "00000000"
+        result = archive_build.build_archive(record, self.atoms, self.out, self.root)
+        self.assertEqual(result.status, "failed")
+        self.assertEqual(list(self.out.rglob("*")), [])
+
+    def test_zlib_ng_never_claims_exact(self):
+        with mock.patch.object(archive_build, "zlib_ng_version", lambda: "2.2.4"):
+            result = self.build("bios/set.zip")
+        self.assertEqual(result.status, "canonical")
+        self.assertIn("zlib-ng 2.2.4", result.detail)
+
+    def test_an_archive_is_named_by_path_or_by_name(self):
+        self.assertEqual(
+            [r["path"] for r in archive_build.select(self.ledger, "set.zip", self.root)],
+            ["bios/set.zip"],
+        )
+        self.assertEqual(
+            len(archive_build.select(self.ledger, str(self.exact_path), self.root)), 1
+        )
+        self.assertEqual(archive_build.select(self.ledger, "none.zip", self.root), [])
+
+    def test_missing_members_reads_the_committed_data(self):
+        self.assertEqual(archive_build.missing_members(self.ledger, self.db), [])
+        del self.db["files"][_sha1(self.exact_path.read_bytes())]
+        missing = archive_build.missing_members(self.ledger, self.db)
+        self.assertEqual(
+            [(m["archive"], m["member"]) for m in missing], [("bios/set.zip", "c.rom")]
+        )
+
+    def test_the_command_reports_each_archive(self):
+        archives = self.root / "archives.json"
+        archives.write_text(archive_build.ledger_text(self.ledger), encoding="utf-8")
+        db = self.root / "database.json"
+        db.write_text(json.dumps(self.db), encoding="utf-8")
+        run = subprocess.run(
+            [
+                sys.executable, str(ROOT / "scripts" / "romset_recipes.py"),
+                "--build-all", "--output-dir", str(self.out), "--discard",
+                "--bios-dir", str(self.root / "bios"), "--db", str(db),
+                "--archives", str(archives),
+            ],
+            capture_output=True, text=True, check=False, timeout=120,
+        )
+        self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+        self.assertIn("EXACT: bios/set.zip", run.stdout)
+        self.assertIn("CANONICAL: bios/other.zip", run.stdout)
+        self.assertIn("2 archives: 1 rebuilt exact, 1 canonical, 0 impossible", run.stdout)
+        self.assertEqual(list(self.out.rglob("*.zip")), [], "--discard keeps nothing")
+
+
+if __name__ == "__main__":
+    unittest.main()