clay-borg/tools/vendor-editions.py

#!/usr/bin/env python3
"""Re-vendor the edition mirror from ground-game and record its digests.

`edition-check` asks two questions: does every vendored file match its
recorded digest, and is every vendored copy current with the sibling
checkout. Answering "no" is routine — ground-game is edited by its own
maintainer and the mirror goes stale several times a day.

**This does the sync; it does not decide anything.** It copies what
upstream has, records what it copied, and refuses to invent either.

## Why this is a tool and not a habit

The mirror went stale three times in one session and was refreshed by
hand each time: copy the changed files, recompute a digest, edit
`PROVENANCE.md`. A hand-repeated sync is exactly the thing that drifts —
the second time, one file gets missed and its digest keeps certifying a
version nobody has.

The digest block is **generated by walking `editions/`**, never typed.
CB-REV-0002 #8 and CB-REV-0003 #8 both found hand-written file lists that
made their own controls vacuous, and a mirror that grows a *directory* is
precisely the case a maintained list loses.

## What it will not do

- It does not delete files upstream no longer has, and it does not add
  files upstream does not have. `edition-check` reports both, and a sync
  tool that silently resolved them would remove the only signal that the
  mirror and upstream disagree about what exists.
- It does not touch `editions/ground-darvo-r0/` except to write the
  digest block. That directory is the baseline package (ADR-0011) and its
  own provenance is a separate record.
"""

import hashlib
import os
import shutil
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
EDITIONS = os.path.join(ROOT, "editions")
EDITION_DIR = os.path.join(EDITIONS, "ground-darvo-r0")
PROVENANCE = os.path.join(EDITION_DIR, "PROVENANCE.md")
UPSTREAM = os.path.join(ROOT, "..", "ground-game", "editions")

SKIP = {".DS_Store"}


def digest(path):
    return hashlib.sha256(open(path, "rb").read()).hexdigest()


def siblings():
    """Every file beside the edition directory, as `../`-relative paths.

    Mirrors `edition-check`'s own walk. Two walks of the same tree is a
    duplication worth naming: if they ever disagree the check fails
    loudly, which is the direction the duplication should fail in."""
    out = []
    for dirpath, _dirs, files in os.walk(EDITIONS):
        if os.path.abspath(dirpath).startswith(os.path.abspath(EDITION_DIR)):
            continue
        for name in sorted(files):
            if name in SKIP:
                continue
            full = os.path.join(dirpath, name)
            rel = os.path.relpath(full, EDITION_DIR).replace(os.sep, "/")
            out.append((rel, full))
    return sorted(out)


def refresh():
    if not os.path.isdir(UPSTREAM):
        print(f"  upstream not checked out at {UPSTREAM}")
        print("  NOT a failure: the sync could not be asked for.")
        return 1

    copied, missing, extra = [], [], []
    for rel, full in siblings():
        up = os.path.normpath(os.path.join(UPSTREAM, rel.replace("../", "", 1)))
        if not os.path.exists(up):
            # Ours, or gone upstream. Either way it is a question for a
            # human -- see the module docstring.
            extra.append(rel)
            continue
        if digest(up) != digest(full):
            shutil.copy2(up, full)
            copied.append(rel)

    # Files upstream has that we do not. Reported, never copied blind:
    # vendoring a new package is a decision (ADR-0011), not a sync.
    for dirpath, _dirs, files in os.walk(UPSTREAM):
        for name in files:
            if name in SKIP:
                continue
            up = os.path.join(dirpath, name)
            rel = os.path.relpath(up, UPSTREAM).replace(os.sep, "/")
            here = os.path.join(EDITIONS, rel)
            if rel.startswith("ground-darvo-r0/"):
                continue
            if not os.path.exists(here):
                missing.append(rel)

    rows = [(digest(full), rel) for rel, full in siblings()]
    rows.sort(key=lambda r: r[1])
    block = "\n".join(f"sha256  {h}  {r}" for h, r in rows)

    text = open(PROVENANCE, encoding="utf-8").read()
    a = text.index("```\nsha256")
    b = text.index("```", a + 3) + 3
    open(PROVENANCE, "w", encoding="utf-8").write(
        text[:a] + "```\n" + block + "\n```" + text[b:]
    )

    print("vendor-editions — the mirror, and what it now records")
    print(f"  refreshed from upstream   {len(copied)}")
    for r in copied:
        print(f"    {r}")
    print(f"  digests recorded          {len(rows)}")
    if extra:
        print(f"  here and NOT upstream     {len(extra)}   (not deleted — a human decides)")
        for r in extra:
            print(f"    {r}")
    if missing:
        print(f"  upstream and NOT here     {len(missing)}   (not copied — vendoring is a decision)")
        for r in missing:
            print(f"    {r}")
    return 0


def recorded_paths():
    """The SIBLING paths the digest block names.

    **`../`-prefixed only.** `PROVENANCE.md` carries two digest blocks —
    one for the edition's own CSVs and one for everything beside it — and
    this tool owns the second. Reading both reported ten edition CSVs as
    "recorded but absent" the first time the self-test ran, because they
    are absent *from the sibling walk*, which is a different question
    from absent.

    The tool's own rewrite is likewise scoped: it replaces the first
    `sha256` block, which is the sibling one, and never touches the
    edition's.
    """
    text = open(PROVENANCE, encoding="utf-8").read()
    out = []
    for line in text.splitlines():
        parts = line.split()
        if len(parts) == 3 and parts[0] == "sha256" and parts[2].startswith("../"):
            out.append(parts[2])
    return sorted(out)


def self_test():
    """A sync tool that cannot detect an unrecorded file is decoration.

    **These are read-only.** A self-test that ran `refresh` would write
    to `PROVENANCE.md` to prove that it can, which is a control that
    changes the thing it measures.
    """
    ok = True

    def check(name, cond, detail=""):
        nonlocal ok
        ok = ok and bool(cond)
        print(f"  [{'ok ' if cond else 'FAIL'}] {name}" + (f"  — {detail}" if detail else ""))

    import tempfile

    with tempfile.NamedTemporaryFile("wb", delete=False) as f:
        f.write(b"clay-borg")
        tmp = f.name
    check("digest is sha256 of the bytes",
          digest(tmp) == hashlib.sha256(b"clay-borg").hexdigest())
    os.unlink(tmp)

    found = [rel for rel, _ in siblings()]
    check("the walk finds files, not just the top level", bool(found))
    # CB-REV-0003 #8: the first edition-check listed three glob patterns
    # and missed everything a directory deeper. The mirror is now three
    # deep, so this asserts the walk actually descends.
    deep = [r for r in found if r.count("/") >= 3]
    check("the walk descends into module directories", bool(deep),
          f"deepest: {max(found, key=lambda r: r.count('/')) if found else 'none'}")

    # THE control. The recorded block must name exactly what is on disk:
    # a file added to the mirror without re-running this tool is the
    # failure it exists to prevent, and `edition-check` would then be
    # certifying a set nobody chose.
    rec = set(recorded_paths())
    disk = set(found)
    check("every vendored file has a recorded digest", not (disk - rec),
          f"unrecorded: {sorted(disk - rec)}" if disk - rec else "")
    check("no digest is recorded for a file that is not here", not (rec - disk),
          f"recorded but absent: {sorted(rec - disk)}" if rec - disk else "")

    # And the digests are the CURRENT bytes, not a stale record that
    # happens to name the right files.
    stale = [rel for rel, full in siblings()
             if digest(full) not in open(PROVENANCE, encoding="utf-8").read()]
    check("every recorded digest is the file's current bytes", not stale,
          f"stale: {stale}" if stale else "")

    print("vendor-editions self-test (positive control)")
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(self_test() if "--self-test" in sys.argv else refresh())