tools/predeploy.py

#!/usr/bin/env python3
"""Pre-deploy gate for the CanLex MCP Space (roadmap T0.1).

    python tools/predeploy.py [--tree REV ...] [--skip-tests] [--allow-held]
    python tools/predeploy.py --print-review-state COMMIT

Checks, each printed PASS / WARN / FAIL (exit status 1 on any FAIL):

 1. tests     the offline suite (`unittest discover -s tests`) in this working tree
 2. curated   refresh.check_curated on this working tree (dataset age, regen drift)
 3. blobs     no NON-LFS blob over 10 MiB in the deploy tree (Hugging Face rejects the
              push; RUNBOOK "Deploying to the Hugging Face Space")
 4. commentary  the deploy tree's commentary.json is exactly what canlex.commentary
              renders from that tree's own data/curated/*.json ...
 5. embeddings  ... and every commentary chunk has a current vector in its
              embeddings.npz (the index silently degrades a chunk without one to BM25)
 6. review    the deploy tree's curated legal JSON against the approved baseline in
              data/curated/review_state.json: prints `git diff --stat` and a per-record
              summary of what differs, and FAILS when held content would ship.
              --allow-held (or CANLEX_PREDEPLOY_ALLOW_HELD=1) downgrades that to a
              warning -- for a rollback to an older approved tree, never for shipping
              held content.
 7. review-code  the deploy tree's officer-facing legal text in code (the definitions
              canlex/reviewed_code.py registers, each with every canlex definition it
              reaches) against the 'approved_code' commit in review_state.json; FAILS
              when held code would ship (--allow-held downgrades it too).

--tree defaults to HEAD, which on main FAILS the review check by design: main carries
held content. Deploy the composed release (tools/compose_release.py) and check it with
`--tree release-composed`; the pre-push hook (tools/hooks/pre-push) passes every commit
being pushed to the 'space' remote.

--print-review-state COMMIT prints the "approved" and "curated_files" blocks for
data/curated/review_state.json when the maintainer approves a new baseline.
"""
import argparse
import ast
import contextlib
import copy
import datetime
import hashlib
import io
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

HF_LIMIT = 10 * 1024 * 1024          # Hugging Face's non-LFS file limit (10 MiB)
CURATED_DIR = "data/curated"
COMMENTARY = "data/processed/commentary.json"
EMBEDDINGS = "data/processed/embeddings.npz"
REVIEW_STATE = "data/curated/review_state.json"
CODE_REGISTRY = "canlex/reviewed_code.py"   # officer-facing legal text in code
LEGAL_FILES = ("us_dispositions.json", "us_equivalency.json")   # what the renderer reads
# ... plus the authority registry the renderer's pointer lines are read from
# (canlex.authorities.PATH): a render is only reproducible with the tree's own
# copy (2026-09-21 audit: the check rendered with the WORKING tree's).
RENDER_INPUTS = LEGAL_FILES + ("authorities.json",)
# Review memos and the held batches' source drafts. Not curated JSON, never
# read at runtime, but held with their batch: a composed release ships the
# approved commit's subtree, and the review check holds any difference
# (2026-09-21 audit: the next compose would have shipped the held batch-1
# memo and its 138 KB of source drafts to the public Space).
DRAFTS_DIR = "data/curated/drafts"
# Written by tools/compose_release.py into a composed tree only: the ids and
# matching words of held content (canlex/server.py reads it). Generated from the
# curated files, no content of its own -- so it is not a curated file here.
HELD_MANIFEST = "held_manifest.json"

PASS, WARN, FAIL = "PASS", "WARN", "FAIL"


# ---- git helpers --------------------------------------------------------------

def git(*args, cwd=ROOT, check=True, stdin=None):
    out = subprocess.run(["git", *args], cwd=str(cwd), capture_output=True, input=stdin)
    if check and out.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)}: "
                           f"{out.stderr.decode('utf-8', 'replace').strip()}")
    return out.stdout


def rev(ref, cwd=ROOT):
    return git("rev-parse", "--verify", f"{ref}^{{commit}}", cwd=cwd).decode().strip()


def show_bytes(commit, path, cwd=ROOT):
    out = subprocess.run(["git", "show", f"{commit}:{path}"], cwd=str(cwd),
                         capture_output=True)
    return out.stdout if out.returncode == 0 else None


def show_json(commit, path, cwd=ROOT):
    blob = show_bytes(commit, path, cwd)
    return None if blob is None else json.loads(blob.decode("utf-8"))


def subtree(commit, path, cwd=ROOT):
    """The tree hash of `path` at a commit, or None when the commit has no
    such directory."""
    out = subprocess.run(["git", "rev-parse", "--verify", "--quiet", f"{commit}:{path}"],
                         cwd=str(cwd), capture_output=True)
    return out.stdout.decode().strip() if out.returncode == 0 else None


def curated_names(commit, cwd=ROOT):
    """The *.json files directly in data/curated at a commit -- the curated
    content, so never the held manifest a composed tree carries (HELD_MANIFEST)."""
    listing = git("ls-tree", "--name-only", commit, CURATED_DIR + "/", cwd=cwd)
    return sorted(Path(p).name for p in listing.decode("utf-8").splitlines()
                  if p.endswith(".json") and p.count("/") == 2
                  and Path(p).name != HELD_MANIFEST)


def canonical_sha256(obj):
    blob = json.dumps(obj, sort_keys=True, ensure_ascii=False,
                      separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(blob).hexdigest()


def review_state(cwd=ROOT):
    return json.loads((Path(cwd) / REVIEW_STATE).read_text(encoding="utf-8"))


def lfs_object(pointer, cwd=ROOT):
    """The content an LFS pointer names, read from the LOCAL LFS store (never
    fetched), or None when it is not there. A non-pointer blob is returned
    as is."""
    if not pointer.startswith(b"version https://git-lfs"):
        return pointer
    oid = next((l.split(b":", 1)[1].strip().decode() for l in pointer.splitlines()
                if l.startswith(b"oid sha256:")), None)
    if not oid:
        return None
    common = Path(git("rev-parse", "--git-common-dir", cwd=cwd).decode().strip())
    if not common.is_absolute():
        common = Path(cwd) / common
    path = common / "lfs" / "objects" / oid[:2] / oid[2:4] / oid
    return path.read_bytes() if path.exists() else None


# ---- checks -------------------------------------------------------------------

def _python_env():
    env = dict(os.environ, PYTHONIOENCODING="utf-8")
    return env


def run_unittest(args, cwd=ROOT):
    """(status, lines) for `python -m unittest <args>` in cwd."""
    out = subprocess.run([sys.executable, "-m", "unittest", *args], cwd=str(cwd),
                         capture_output=True, env=_python_env())
    text = (out.stderr + out.stdout).decode("utf-8", "replace")
    lines = [l for l in text.splitlines() if l.startswith(("FAIL:", "ERROR:",
                                                           "UNEXPECTED SUCCESS:"))]
    tail = [l for l in text.splitlines() if l.startswith(("Ran ", "OK", "FAILED"))]
    return (PASS if out.returncode == 0 else FAIL), lines[:40] + tail


def check_curated_working():
    from canlex import refresh
    rows = refresh.check_curated(datetime.date.today().isoformat())
    lines = [f"{r['file']}: {r['status']} {r.get('reviewed', '')} {r.get('error', '')}".rstrip()
             for r in rows]
    if any(r["status"] in ("drift", "error") for r in rows):
        return FAIL, lines
    if any(r["status"] == "aging" for r in rows):
        return WARN, lines
    return PASS, lines


def check_blobs(commit, cwd=ROOT):
    listing = git("ls-tree", "-r", "-l", commit, cwd=cwd).decode("utf-8", "replace")
    big, near = [], []
    for line in listing.splitlines():
        meta, _, path = line.partition("\t")
        parts = meta.split()
        if len(parts) < 4 or parts[1] != "blob" or not parts[3].isdigit():
            continue
        size = int(parts[3])
        if size > HF_LIMIT:
            big.append(f"{size:,} bytes  {path}  -- track it in LFS (.gitattributes, exact path)")
        elif size > 0.9 * HF_LIMIT:
            near.append(f"{size:,} bytes  {path}  -- within 10% of the limit")
    if big:
        return FAIL, big + near
    return (WARN if near else PASS), near or ["no non-LFS blob over 10 MiB"]


def _clear_authority_caches(au):
    """Every lru_cache in canlex.authorities (the registry, the corpus files
    it reads, the act sections, the statute sentences): a render with another
    registry must not see the working tree's through any of them."""
    for value in vars(au).values():
        if hasattr(value, "cache_clear"):
            value.cache_clear()


def render_commentary(curated_blobs):
    """The chunks canlex.commentary renders from {file name: bytes}, written
    to a temporary directory (the repo's own renderer, pointed elsewhere).

    The authority registry is pointed at the tree's authorities.json too --
    absent from the blobs, at a path that does not exist, so the render
    prints no pointers, as that tree would. Until 2026-09-21 the renderer
    read canlex.authorities.PATH, the WORKING tree's file, so the check
    passed or failed on a registry the deploy tree does not carry."""
    from canlex import authorities as au, commentary as cm
    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        for name, blob in curated_blobs.items():
            (tmp / name).write_bytes(blob)
        saved = cm.CURATED, cm.EQUIV, cm.OUT, au.PATH
        cm.CURATED = tmp / "us_dispositions.json"
        cm.EQUIV = tmp / "us_equivalency.json"
        cm.OUT = tmp / "commentary.json"
        au.PATH = tmp / "authorities.json"
        _clear_authority_caches(au)
        try:
            with contextlib.redirect_stdout(io.StringIO()):
                ok = cm.build(allow_shrink=True)
        finally:
            cm.CURATED, cm.EQUIV, cm.OUT, au.PATH = saved
            _clear_authority_caches(au)
        if not ok:
            raise RuntimeError("canlex.commentary refused to write")
        return json.loads((tmp / "commentary.json").read_text(encoding="utf-8"))


def check_commentary(commit, cwd=ROOT):
    """(status, lines, stored chunks) -- the tree's commentary.json against a
    fresh render of the tree's own curated files (the authority registry
    included)."""
    blobs = {n: show_bytes(commit, f"{CURATED_DIR}/{n}", cwd) for n in RENDER_INPUTS}
    blobs = {n: b for n, b in blobs.items() if b is not None}
    stored = show_json(commit, COMMENTARY, cwd)
    if stored is None:
        return FAIL, [f"{COMMENTARY} missing from the tree"], []
    if "us_dispositions.json" not in blobs:
        return WARN, ["no us_dispositions.json in the tree; render not checked"], stored
    rendered = render_commentary(blobs)
    if rendered == stored:
        return PASS, [f"{len(stored)} chunks, identical to a fresh render"], stored
    s_ids, r_ids = [c["id"] for c in stored], [c["id"] for c in rendered]
    lines = [f"stored {len(stored)} chunks, fresh render {len(rendered)}"]
    only_s, only_r = sorted(set(s_ids) - set(r_ids)), sorted(set(r_ids) - set(s_ids))
    if only_s:
        lines.append(f"only in stored ({len(only_s)}): {', '.join(only_s[:8])}")
    if only_r:
        lines.append(f"only in render ({len(only_r)}): {', '.join(only_r[:8])}")
    by_id = {c["id"]: c for c in rendered}
    diff = [c["id"] for c in stored if c["id"] in by_id and by_id[c["id"]] != c]
    if diff:
        lines.append(f"text differs ({len(diff)}): {', '.join(diff[:8])}")
    lines.append("rerun canlex.commentary (then canlex.embed) on this tree")
    return FAIL, lines, stored


def embeddings_gap(chunks, npz_path):
    """Chunk ids with no current vector in the npz (canlex.embed's own reuse
    rule: same id, same content hash, same window count)."""
    from canlex.embed import load_stored, plan
    _, to_embed = plan(chunks, load_stored(Path(npz_path)))
    return [cid for cid, _, _ in to_embed]


def tree_chunks(commit, cwd=ROOT):
    """Every chunk of the tree's data/processed/*.json (LFS objects read
    from the local store), or None when one of them is not there. A file
    that is there but is not a JSON list raises ValueError naming it."""
    listing = git("ls-tree", "--name-only", commit, "data/processed/", cwd=cwd)
    chunks = []
    for path in sorted(p for p in listing.decode("utf-8").splitlines()
                       if p.endswith(".json")):
        blob = show_bytes(commit, path, cwd)
        data = None if blob is None else lfs_object(blob, cwd)
        if data is None:
            return None
        try:
            parsed = json.loads(data.decode("utf-8"))
        except ValueError as exc:            # JSONDecodeError, UnicodeDecodeError
            raise ValueError(f"{path}: {exc}") from exc
        if not isinstance(parsed, list):
            raise ValueError(f"{path}: a JSON {type(parsed).__name__}, not a list of chunks")
        chunks.extend(parsed)
    return chunks


def embeddings_currency(commit, npz_path, cwd=ROOT):
    """(missing, stale) over the tree's WHOLE corpus (canlex.embed.currency),
    or None when the corpus cannot be read here (audit follow-up 2026-09-21,
    Missing 15: only the commentary chunks were checked). A corpus file that
    does not parse raises ValueError: until the 2026-09-22 verification
    (A-EMB-3) it was swallowed as 'not readable here' and the check PASSed,
    though the Space's index json.loads every processed file at start."""
    try:
        chunks = tree_chunks(commit, cwd)
    except (RuntimeError, OSError):
        return None
    if chunks is None:
        return None
    from canlex.embed import currency, load_stored
    return currency(chunks, load_stored(Path(npz_path)))


def check_embeddings(commit, chunks, cwd=ROOT):
    pointer = show_bytes(commit, EMBEDDINGS, cwd)
    if pointer is None:
        return FAIL, [f"{EMBEDDINGS} missing from the tree"]
    data = lfs_object(pointer, cwd)
    if data is None:
        return WARN, ["embeddings.npz is not in the local LFS store; coverage not checked"]
    try:
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "embeddings.npz"
            path.write_bytes(data)
            gap = embeddings_gap(chunks, path)
            try:
                current = embeddings_currency(commit, path, cwd)
            except ValueError as exc:        # a corpus file that does not parse
                return FAIL, [f"the tree's corpus does not parse: {exc}",
                              "the Space's index reads every data/processed file at "
                              "start: regenerate or restore that file"]
    except ImportError as exc:
        return WARN, [f"canlex.embed not importable here ({exc}); coverage not checked"]
    if gap:
        return FAIL, [f"{len(gap)} commentary chunk(s) have no current vector: "
                      + ", ".join(gap[:8]), "run canlex.embed on this tree"]
    lines = [f"all {len(chunks)} commentary chunks have current vectors"]
    if current is None:
        # unchecked is not a pass (verification 2026-09-22, A-EMB-3 / C-GATE-3):
        # the same WARN as an npz absent from the local LFS store
        return WARN, lines + ["whole corpus not readable here; its currency not checked"]
    missing, stale = current
    if missing or stale:
        return FAIL, [f"{missing} chunk(s) have no vector and {stale} a stale one "
                      "(canlex.embed's plan)", "run canlex.embed on this tree"]
    return PASS, lines + ["whole corpus: 0 missing, 0 stale vectors"]


def _by_id(items):
    return {x.get("id"): x for x in items or [] if isinstance(x, dict)}


def _records(label, old, new):
    o, n = _by_id(old), _by_id(new)
    lines = []
    added, removed = sorted(set(n) - set(o)), sorted(set(o) - set(n))
    if added:
        lines.append(f"  {label} added ({len(added)}): {', '.join(map(str, added))}")
    if removed:
        lines.append(f"  {label} removed ({len(removed)}): {', '.join(map(str, removed))}")
    for k in sorted(set(o) & set(n), key=str):
        if o[k] == n[k]:
            continue
        fields = sorted(f for f in set(o[k]) | set(n[k]) if o[k].get(f) != n[k].get(f))
        detail = []
        for f in fields:
            if f == "state_variations":
                ov = {json.dumps(r, sort_keys=True) for r in o[k].get(f) or []}
                nv = {json.dumps(r, sort_keys=True) for r in n[k].get(f) or []}
                detail.append(f"state_variations (+{len(nv - ov)} -{len(ov - nv)} rows)")
            elif f in ("names", "aliases", "us_terms", "negative_terms", "caveats"):
                ov, nv = set(o[k].get(f) or []), set(n[k].get(f) or [])
                if ov == nv:     # e.g. absent -> [] or null -> []: a formatting change
                    detail.append(f"{f} {o[k].get(f)!r} -> {n[k].get(f)!r}")
                else:
                    detail.append(f"{f} (+{len(nv - ov)} -{len(ov - nv)})")
            elif isinstance(o[k].get(f), (int, float, bool, type(None))) and \
                    isinstance(n[k].get(f), (int, float, bool, type(None))):
                detail.append(f"{f} {o[k].get(f)!r} -> {n[k].get(f)!r}")
            else:
                detail.append(f)
        lines.append(f"  {label} changed: {k}: {'; '.join(detail)}")
    return lines


def record_summary(old, new):
    """Per-record lines describing how a curated dataset differs."""
    lines = []
    for key in sorted(set(old) | set(new)):
        if key in ("pairings", "dispositions", "methodology", "aliases"):
            continue
        if old.get(key) != new.get(key):
            a, b = old.get(key), new.get(key)
            short = all(isinstance(x, (str, int, type(None))) and len(str(x)) <= 40
                        for x in (a, b))
            lines.append(f"  top-level {key}: " + (f"{a!r} -> {b!r}" if short else "changed"))
    for key, label in (("pairings", "pairing"), ("dispositions", "entry"),
                       ("methodology", "methodology")):
        if key in old or key in new:
            lines += _records(label, old.get(key), new.get(key))
    oa, na = old.get("aliases") or {}, new.get("aliases") or {}
    if oa != na:
        added = sorted(set(na) - set(oa))
        removed = sorted(set(oa) - set(na))
        changed = sorted(k for k in set(oa) & set(na) if oa[k] != na[k])
        lines.append(f"  aliases: +{len(added)} -{len(removed)} ~{len(changed)}"
                     + (f" (e.g. {', '.join(added[:10])})" if added else ""))
    return lines


def check_review(commit, approved, state, allow_held, cwd=ROOT):
    lines, held = [], []
    meta = set(state.get("metadata_files", []))
    recorded = state.get("curated_files", {})
    tree_names, appr_names = curated_names(commit, cwd), curated_names(approved, cwd)
    for name in sorted(set(tree_names) | set(appr_names)):
        if name in meta:
            continue
        path = f"{CURATED_DIR}/{name}"
        new = show_json(commit, path, cwd) if name in tree_names else None
        old = show_json(approved, path, cwd) if name in appr_names else None
        if old is not None and name in recorded and \
                canonical_sha256(old) != recorded[name].get("canonical_sha256"):
            lines.append(f"WARNING: {name} at the approved commit does not match the "
                         "hash recorded in review_state.json")
        if old is None:
            held.append(f"{name}: new curated file, not in the approved baseline")
        elif new is None:
            held.append(f"{name}: approved file missing from the deploy tree")
        elif canonical_sha256(old) != canonical_sha256(new):
            held.append(f"{name}: differs from the approved baseline")
            lines += [f"{name}:"] + record_summary(old, new)
    d_new, d_old = subtree(commit, DRAFTS_DIR, cwd), subtree(approved, DRAFTS_DIR, cwd)
    if d_new != d_old:
        held.append(f"{DRAFTS_DIR}/: "
                    + ("absent from the deploy tree" if d_new is None else
                       "not in the approved commit" if d_old is None else
                       "differs from the approved commit")
                    + " (review memos and source drafts ship with their approved batch; "
                      "tools/compose_release.py puts the approved commit's copy in the tree)")
    stat = git("diff", "--stat", approved, commit, "--", CURATED_DIR, COMMENTARY,
               cwd=cwd).decode("utf-8", "replace").rstrip()
    lines = [f"approved baseline {approved[:7]} ({state.get('approved', {}).get('date', '?')}); "
             f"deploy tree {commit[:7]}",
             f"git diff --stat {approved[:7]} {commit[:7]} -- {CURATED_DIR} {COMMENTARY}:"] \
        + ["  " + l for l in (stat.splitlines() or ["(no differences)"])] + lines
    if not held:
        return PASS, lines + ["curated legal content equals the approved baseline"]
    lines += [f"HELD: {h}" for h in held]
    if allow_held:
        return WARN, lines + ["--allow-held: shipping content that differs from the "
                              "approved baseline"]
    return FAIL, lines + ["held-for-review content would ship: deploy the composed release "
                          "(tools/compose_release.py), or record the approval in "
                          "data/curated/review_state.json"]


# ---- officer-facing legal text in code (verifier, 2026-09-12) -----------------

def _body_without_docstring(node):
    body = list(node.body)
    if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) \
            and isinstance(body[0].value.value, str):
        body = body[1:]
    return body or [ast.Pass()]


# FastMCP publishes a tool, prompt or resource function's docstring verbatim as
# its `description` -- the text the client model reads (all 14 tools; 16,903
# characters; canlex_screen_record's states law: a discharge is not a
# conviction, a conditional sentence is not imprisonment...). The docstring
# and the decorator's own keywords (description=, name=) of such a function
# are officer-facing text and count in its fingerprint; every other
# definition's docstring and decorators are documentation and do not.
_MCP_DECORATORS = ("tool", "prompt", "resource")


def _publishes(node):
    """True when a decorator is mcp.tool / mcp.prompt / mcp.resource."""
    for d in getattr(node, "decorator_list", []):
        f = d.func if isinstance(d, ast.Call) else d
        if isinstance(f, ast.Attribute) and f.attr in _MCP_DECORATORS \
                and isinstance(f.value, ast.Name) and f.value.id == "mcp":
            return True
    return False


def _fingerprint(node):
    """sha256 of a definition's AST: a function or class without its
    docstring and decorators (kept for an mcp.tool/prompt/resource function,
    whose docstring IS its published description), or an assignment's value.
    Formatting, comments and string-literal line wrapping do not change it;
    words and logic do."""
    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
        node = copy.deepcopy(node)
        if not _publishes(node):
            node.body = _body_without_docstring(node)
            node.decorator_list = []
    return hashlib.sha256(ast.dump(node, include_attributes=False).encode("utf-8")).hexdigest()


def _base_name(node):
    """The Name at the root of a Subscript / Attribute chain (`X[k]`,
    `X.update`), or None."""
    while isinstance(node, (ast.Subscript, ast.Attribute)):
        node = node.value
    return node.id if isinstance(node, ast.Name) else None


def _definitions(source):
    """{top-level name: node} of a Python source: a function or class node,
    or an assignment's value. A tuple or list target (`A, B = "a", "b"`)
    yields the whole assignment under each name -- screening.py's
    STRUCTURED_LIMB / STATED_LIMB, the limb keys the registered
    _resolve_pairing selects penalty_history rows by, were invisible to the
    gate until 2026-09-21. A later top-level mutation of a defined name
    (`X[k] = v`, `X += ...`, `X.update(...)`) is folded into that name's node
    so it counts too (none in canlex/ today)."""
    out, extra = {}, {}
    for node in ast.parse(source).body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            out[node.name] = node
        elif isinstance(node, ast.Assign):
            for t in node.targets:
                if isinstance(t, ast.Name):
                    out[t.id] = node.value
                elif isinstance(t, (ast.Tuple, ast.List)):
                    for elt in t.elts:
                        if isinstance(elt, ast.Name):
                            out[elt.id] = node
                elif _base_name(t):
                    extra.setdefault(_base_name(t), []).append(node)
        elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) \
                and node.value is not None:
            out[node.target.id] = node.value
        elif isinstance(node, ast.AugAssign) and _base_name(node.target):
            extra.setdefault(_base_name(node.target), []).append(node)
        elif isinstance(node, ast.Expr) and isinstance(node.value, ast.Call) \
                and _base_name(node.value.func):
            extra.setdefault(_base_name(node.value.func), []).append(node)
    for name, nodes in extra.items():
        if name in out:
            out[name] = ast.Module(body=[out[name]] + nodes, type_ignores=[])
    return out


def top_level_fingerprints(source):
    """{name: fingerprint} for the top-level assignments and definitions of a
    Python source."""
    return {name: _fingerprint(node) for name, node in _definitions(source).items()}


class CodeTree:
    """The canlex modules of one tree -- a commit (reader: git show) or the
    working tree (reader: the file) -- for the review gate's reach (wave 2
    verifier): a registered definition is fingerprinted TOGETHER WITH every
    top-level canlex definition it reaches (the names its body uses, in its
    own module or imported from another canlex module, and theirs in turn),
    so a change to a helper (a unit table, a negation window, a history
    reader) holds the registered name that relies on it."""

    def __init__(self, reader):
        self._read, self._src = reader, {}
        self._mods = {}

    def reader(self, path):
        if path not in self._src:
            self._src[path] = self._read(path)
        return self._src[path]

    @classmethod
    def at(cls, commit, cwd=ROOT):
        def read(path):
            blob = show_bytes(commit, path, cwd)
            return None if blob is None else blob.decode("utf-8")
        return cls(read)

    @classmethod
    def working(cls, root=ROOT):
        def read(path):
            p = Path(root) / path
            return p.read_text(encoding="utf-8") if p.exists() else None
        return cls(read)

    def module(self, path):
        """({name: node}, {local name: (path, name or None for a module)})."""
        if path not in self._mods:
            src = self.reader(path)
            if src is None:
                self._mods[path] = ({}, {})
            else:
                tree = ast.parse(src)
                self._mods[path] = (_definitions(src), self._imports(tree.body, path))
        return self._mods[path]

    def _module_path(self, mod, level, here):
        if level:
            base = Path(here).parent
            for _ in range(level - 1):
                base = base.parent
            return (base / mod.replace(".", "/")).as_posix() if mod else base.as_posix()
        return mod.replace(".", "/") if mod and mod.split(".")[0] == "canlex" else None

    def _imports(self, nodes, here):
        out = {}
        for top in nodes:
            for node in ([top] if isinstance(top, (ast.Import, ast.ImportFrom))
                         else ast.walk(top)):
                if isinstance(node, ast.ImportFrom):
                    mp = self._module_path(node.module, node.level, here)
                    if not mp:
                        continue
                    is_module = self.reader(mp + ".py") is not None
                    for a in node.names:
                        sub = f"{mp}/{a.name}.py"
                        if not is_module and self.reader(sub) is not None:
                            out[a.asname or a.name] = (sub, None)    # a submodule
                        else:
                            out[a.asname or a.name] = (mp + ".py", a.name)
                elif isinstance(node, ast.Import):
                    for a in node.names:
                        if a.name.split(".")[0] == "canlex" and a.asname:
                            out[a.asname] = (a.name.replace(".", "/") + ".py", None)
        return out

    def refs(self, path, name):
        """The (path, name) top-level canlex definitions `name`'s body uses."""
        defs, imports = self.module(path)
        node = defs.get(name)
        if node is None:
            return set()
        local = dict(imports)
        local.update(self._imports([node], path))
        out = set()
        for n in ast.walk(node):
            if isinstance(n, ast.Name) and n.id != name:
                if n.id in defs:
                    out.add((path, n.id))
                elif n.id in local and local[n.id][1] is not None:
                    out.add(local[n.id])
            elif isinstance(n, ast.Attribute) and isinstance(n.value, ast.Name):
                tgt = local.get(n.value.id)
                if tgt and tgt[1] is None:
                    out.add((tgt[0], n.attr))
        return {r for r in out if r[1] in self.module(r[0])[0]}

    def reach(self, path, name):
        """Every (path, name) the definition reaches, itself included."""
        seen, todo = set(), [(path, name)]
        while todo:
            key = todo.pop()
            if key in seen:
                continue
            seen.add(key)
            todo += [r for r in self.refs(*key) if r not in seen]
        return seen

    def own(self, path, name):
        node = self.module(path)[0].get(name)
        return None if node is None else _fingerprint(node)

    def fingerprint(self, path, name):
        """sha256 over the definition and everything it reaches, or None when
        the tree has no such definition."""
        if self.own(path, name) is None:
            return None
        parts = sorted(f"{p}:{n}:{self.own(p, n)}" for p, n in self.reach(path, name))
        return hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()


def reviewed_registry(commit, cwd=ROOT):
    """[(path, name)] from the tree's canlex/reviewed_code.py REVIEWED, read
    with ast (never imported); None when the tree has no registry."""
    blob = show_bytes(commit, CODE_REGISTRY, cwd)
    if blob is None:
        return None
    for node in ast.parse(blob.decode("utf-8")).body:
        if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == "REVIEWED"
                                                for t in node.targets):
            return [tuple(e[:2]) for e in ast.literal_eval(node.value)]
    raise RuntimeError(f"{CODE_REGISTRY} at {commit[:7]} has no REVIEWED tuple")


def code_fingerprints(commit, entries, cwd=ROOT, tree=None):
    """{(path, name): fingerprint or None} at a commit: each registered
    definition together with every canlex definition it reaches
    (CodeTree.fingerprint)."""
    tree = tree or CodeTree.at(commit, cwd)
    return {(path, name): tree.fingerprint(path, name) for path, name in entries}


def _changed_members(old_tree, new_tree, key):
    """The definitions `key` reaches in the new tree that are new or differ
    from the old tree's (itself excluded): why a registered name is held."""
    out = []
    for p, n in sorted(new_tree.reach(*key) - {key}):
        before = old_tree.own(p, n)
        if before is None:
            out.append(f"{p}:{n} (new)")
        elif before != new_tree.own(p, n):
            out.append(f"{p}:{n}")
    gone = sorted(old_tree.reach(*key) - new_tree.reach(*key))
    return out + [f"{p}:{n} (no longer reached)" for p, n in gone]


def check_review_code(commit, state, allow_held, cwd=ROOT):
    """The registered officer-facing definitions in the deploy tree against
    the approved-code commit in review_state.json ('approved_code').

    The registry is read at BOTH commits and the union is checked: an entry
    may be added, never dropped, without a HELD line naming the drop, and a
    tree with no registry while the approved code has one FAILS (2026-09-21
    audit: the check read the deploy tree's registry alone, so deleting an
    entry that no other registered definition reaches -- canlex_screen_record
    was one of 13 such roots -- removed it from the gate, and a tree with no
    canlex/reviewed_code.py passed)."""
    entries_new = reviewed_registry(commit, cwd)
    base = (state.get("approved_code") or {}).get("commit")
    base_full = entries_old = None
    if base:
        try:
            base_full = rev(base, cwd)
        except RuntimeError:
            return FAIL, [f"approved_code commit {base} is not in this repository"]
        entries_old = reviewed_registry(base_full, cwd)
    if entries_new is None:
        if entries_old:
            lines = [f"no {CODE_REGISTRY} in this tree while the approved code "
                     f"{base_full[:7]} registers {len(entries_old)} definitions: the review "
                     "gate cannot be dropped by removing the registry"]
            if allow_held:
                return WARN, lines + ["--allow-held: a rollback to a pre-registry release"]
            return FAIL, lines + ["a rollback to a pre-registry release needs "
                                  "CANLEX_PREDEPLOY_ALLOW_HELD=1 (RUNBOOK 'Rollback')"]
        return PASS, [f"no {CODE_REGISTRY} in this tree (an older release): not checked"]
    dropped = [e for e in (entries_old or []) if e not in set(entries_new)]
    entries = list(entries_new) + dropped
    new_tree = CodeTree.at(commit, cwd)
    new = code_fingerprints(commit, entries, cwd, tree=new_tree)
    missing = [f"{p}:{n}" for (p, n), h in new.items() if h is None and (p, n) in entries_new]
    if missing:
        return FAIL, ["the registry names definitions this tree does not have: "
                      + ", ".join(missing), f"fix {CODE_REGISTRY}"]
    held, lines = [], []
    if not base:
        lines.append("no 'approved_code' commit in review_state.json")
        held = [f"{p}:{n}: no approved code baseline" for p, n in entries]
    else:
        old_tree = CodeTree.at(base_full, cwd)
        old = code_fingerprints(base_full, entries, cwd, tree=old_tree)
        lines.append(f"approved code baseline {base_full[:7]} "
                     f"({state['approved_code'].get('date', '?')}); deploy tree {commit[:7]}; "
                     f"{len(entries_new)} registered definitions, each with the canlex "
                     "definitions it reaches"
                     + (f"; {len(dropped)} more registered at the approved code only"
                        if dropped else ""))
        for key in entries:
            what = []
            if key in dropped:
                what.append("dropped from the registry since the approved code")
            if old[key] is None:
                what.append("new since the approved code")
            elif new[key] is None:
                what.append("removed from the tree since the approved code")
            elif old[key] != new[key]:
                via = _changed_members(old_tree, new_tree, key)
                what.append("differs from the approved code"
                            + (f" (via {', '.join(via)})" if via else ""))
            if what:
                held.append(f"{key[0]}:{key[1]}: {'; '.join(what)}")
    if not held:
        return PASS, lines + ["officer-facing legal text in code equals the approved code"]
    files = sorted({p for p, _ in entries})
    lines += [f"HELD: {h}" for h in held]
    lines.append(f"review with: git diff {(base or '<approved>')[:7]} {commit[:7]} -- "
                 + " ".join(files))
    if allow_held:
        return WARN, lines + ["--allow-held: shipping officer-facing legal text in code "
                              "that the maintainer has not approved"]
    return FAIL, lines + ["held officer-facing legal text in code would ship: record the "
                          "reviewed commit as 'approved_code' in data/curated/review_state.json "
                          "(python tools/predeploy.py --print-review-state <commit>), or pass "
                          "--allow-held"]


# ---- refresh drift of the approved curated table (wave 2 verifier) -------------

def _corpus_json(blob, cwd=ROOT):
    """A processed act file's chunks from its blob (an LFS pointer is read
    from the local store); [] when unreadable."""
    if blob is None:
        return []
    blob = lfs_object(blob, cwd)
    if blob is None:
        return []
    try:
        return json.loads(blob.decode("utf-8"))
    except ValueError:
        return []


def corpus_reader(commit=None, cwd=ROOT):
    """section_text(act, section) over a commit's data/processed (None: the
    working tree), whitespace squashed -- the text a curated verify phrase is
    checked against."""
    import re
    cache = {}

    def text(act, section):
        if act not in cache:
            path = f"data/processed/{act}.json"
            if commit is None:
                p = Path(cwd) / path
                cache[act] = _corpus_json(p.read_bytes() if p.exists() else None, cwd)
            else:
                cache[act] = _corpus_json(show_bytes(commit, path, cwd), cwd)
        return re.sub(r"\s+", " ", " ".join(c.get("text", "") for c in cache[act]
                                            if str(c.get("section")) == section))
    return text


def verify_sections(pairing):
    """{(act, section): the maxima the pairing's verify rows (and its limbs')
    quote for that section} -- canlex.screening.maxima_stated on each
    phrase."""
    from canlex.screening import maxima_stated
    out = {}
    rows = list(pairing.get("verify") or []) + [
        v for l in pairing.get("limbs") or [] if isinstance(l, dict) for v in l.get("verify") or []]
    for v in rows:
        out.setdefault((v["act"], v["section"]), set()).update(maxima_stated(v.get("contains")))
    return out


def refresh_drift(equivalency, now_text, then_text):
    """The curated table against a corpus refresh: (drift, changed).

    drift: [(pairing id, act, section, sorted new maxima)] -- a maximum a
    verify section states NOW that it did not state in the corpus the table
    was approved against, and that the pairing does not quote (an amendment
    the approved text cannot know: 2026, c. 19 raised s. 173(2)(a) to 10
    years). A composed release shipping such a table ships stale law.
    changed: [(pairing id, act, section)] -- every verify section whose text
    changed at all (re-read before relying on the approved prose)."""
    from canlex.screening import maxima_stated
    drift, changed = [], []
    for p in (equivalency or {}).get("pairings") or []:
        for (act, sec), quoted in sorted(verify_sections(p).items()):
            now, then = now_text(act, sec), then_text(act, sec)
            if not now or not then:
                continue
            if now != then:
                changed.append((p["id"], act, sec))
            new = maxima_stated(now) - maxima_stated(then) - quoted
            if new:
                drift.append((p["id"], act, sec, sorted(new)))
    return drift, changed


def review_state_for(commit, cwd=ROOT):
    full = rev(commit, cwd)
    date = git("show", "-s", "--format=%cs", full, cwd=cwd).decode().strip()
    files = {n: {"canonical_sha256": canonical_sha256(show_json(full, f"{CURATED_DIR}/{n}", cwd))}
             for n in curated_names(full, cwd) if n != "review_state.json"}
    return {"approved": {"commit": full, "date": date,
                         "note": "<who approved it, when, and what the batch was>"},
            "curated_files": files,
            "approved_code": {"commit": full, "date": date,
                              "note": "<who reviewed the officer-facing text in code "
                                      f"({CODE_REGISTRY}), when>"}}


# ---- driver -------------------------------------------------------------------

def report(name, status, lines):
    print(f"== {name}: {status}")
    for l in lines:
        print(f"   {l}")
    sys.stdout.flush()


def tree_checks(commit, approved, state, allow_held, cwd=ROOT, allow_held_code=None):
    """[(name, status, lines)] for one deploy tree -- also used by
    tools/compose_release.py on the release commit. allow_held_code (default:
    allow_held) governs the review-code check alone."""
    short = commit[:7]
    results = [(f"blobs {short}",) + check_blobs(commit, cwd)]
    status, lines, chunks = check_commentary(commit, cwd)
    results.append((f"commentary {short}", status, lines))
    if chunks:
        results.append((f"embeddings {short}",) + check_embeddings(commit, chunks, cwd))
    results.append((f"review {short}",) + check_review(commit, approved, state,
                                                      allow_held, cwd))
    results.append((f"review-code {short}",) + check_review_code(
        commit, state, allow_held if allow_held_code is None else allow_held_code, cwd))
    return results


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--tree", action="append",
                    help="commit to check as the deploy tree (repeatable; default HEAD)")
    ap.add_argument("--approved", help="approved baseline commit (default: review_state.json)")
    ap.add_argument("--skip-tests", action="store_true", help="skip the offline suite")
    ap.add_argument("--allow-held", action="store_true",
                    help="warn instead of failing when curated content differs from the baseline")
    ap.add_argument("--allow-held-code", action="store_true",
                    help="warn instead of failing on the review-code check alone (what "
                         "tools/compose_release.py --allow-held-code passes through)")
    ap.add_argument("--print-review-state", metavar="COMMIT")
    args = ap.parse_args(argv)
    with contextlib.suppress(AttributeError, ValueError):
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    if args.print_review_state:
        print(json.dumps(review_state_for(args.print_review_state), indent=1))
        return 0
    allow_held = args.allow_held or os.environ.get("CANLEX_PREDEPLOY_ALLOW_HELD") == "1"
    results = []
    if not args.skip_tests:
        results.append(("tests",) + run_unittest(["discover", "-s", "tests", "-q"]))
        report(*results[-1])
    results.append(("curated (working tree)",) + check_curated_working())
    report(*results[-1])
    state = review_state()
    approved = rev(args.approved or state["approved"]["commit"])
    for ref in args.tree or ["HEAD"]:
        for r in tree_checks(rev(ref), approved, state, allow_held,
                             allow_held_code=True if args.allow_held_code else None):
            results.append(r)
            report(*r)
    fails = [n for n, s, _ in results if s == FAIL]
    warns = [n for n, s, _ in results if s == WARN]
    print(f"\npredeploy: {'FAIL' if fails else 'PASS'}"
          + (f" -- failed: {', '.join(fails)}" if fails else "")
          + (f"; warnings: {', '.join(warns)}" if warns else ""))
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main())