scripts.gates

Reference › scripts › gates

scripts/gates.py is the single entry point for the RUL-008 review battery: one command that runs every repository gate script in parallel and prints one line per gate. A human or a pre-push step runs it; nothing imports it. The GATES table maps a gate name to the argv of a gate script (for example architecture_checker) and the source roots whose content decides whether its last pass is still valid.

Options are --all (ignore the cache), --no-fix-ast (do not refresh a stale AST snapshot), --no-fix (skip the ruff --fix and black pass done by autoformat) and --only with one or more gate names. main first formats, then fingerprints the src, tests and scripts trees with _fingerprint, which hashes content through git's own object hashing. If that fingerprint differs from the cached one, refresh_ast runs ast-sync and then build_runtime_callgraph.py reaugment, always in that order, because the first step drops every runtime_calls entry and the second restores them; it fails if the runtime call graph is empty afterwards. See build_runtime_callgraph.

Each selected gate whose roots are unchanged since its last pass is reported as SKIP. Gates marked always-run (filesystem or system-state gates, which no content hash covers) are never skipped. The rest run through a ThreadPoolExecutor, one worker per gate, via _run_gate, which records the elapsed time and the last lines of output. Results are saved in the cache by _save_cache. The exit code is 0 when all gates pass and 1 when the AST refresh or any gate fails; the final line names the failing gates.

At a glance

Source scripts/gates.py
Lines of code 578
Classes 0
Functions 9
Imports (internal) none
Imported by none
Tests that import it none

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Constants

Name Type Value
GATES dict[str, tuple[list[str], tuple[str, ...]]] {'architecture': ([sys.executable, 'scripts/architecture_checker.py'], ('src',)…

Functions

Function Signature Summary
_fingerprint _fingerprint(roots: tuple[str, ...]) -> str Content hash of roots, from git's own object hashing.
_interesting _interesting(path: Path) -> bool
_load_cache _load_cache() -> dict[str, dict]
_save_cache _save_cache(cache: dict[str, dict]) -> None
autoformat autoformat() -> None Apply the mechanical fixes instead of failing the run over them.
refresh_ast refresh_ast() -> tuple[bool, str] Re-derive the AST snapshot, then RE-AUGMENT it.
_runtime_call_count _runtime_call_count() -> int
_run_gate _run_gate(name: str, argv: list[str]) -> tuple[str, bool, str]
main main() -> int

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/gates.py

#!/usr/bin/env python3
"""THE entry point for the RUL-008 review battery.

One command, run in parallel, with three behaviours that exist because the
manual alternative kept costing whole minutes and whole sessions:

1. **Parallel.** The gates are independent read-only analyses over a tree
   nobody is mutating, so wall-clock is the slowest gate, not their sum.
   Measured 2026-08-07: ~50s serial -> ~25s parallel, and the serial version
   was what a human (or an agent) typed by hand, one gate per shell command.

2. **Deduplicated.** Each gate's verdict is cached against a fingerprint of
   the inputs it actually reads. Re-running with nothing changed is a no-op
   that prints the cached PASS instead of re-deriving it. This is the fix for
   duplicate runs: the second and third invocation of an unchanged gate cost
   nothing, so there is no incentive to skip the battery to save time.

3. **Self-healing.** Two failure modes here are mechanical and were being
   fixed by hand every time, so they are fixed automatically instead:
     - lint: ``ruff --fix`` + ``black`` run BEFORE the gates. The user-level
       autoformat hook only fires on Write/Edit tool calls, so any file
       written by a script (heredoc, codemod, generator) skipped it and
       failed the lint gate later.
     - AST staleness: ``plans/ast-ultimate.json`` goes stale on every source
       change, and ``uv run ast-sync`` ZEROES ``runtime_calls`` unless
       ``build_runtime_callgraph.py reaugment`` runs after it. That two-step
       ordering was documented and still forgotten repeatedly; ``--fix-ast``
       does both, in order, and verifies the augmentation landed.

Usage::

    python scripts/gates.py                # changed gates only (the default)
    python scripts/gates.py --all          # ignore the cache, run everything
    python scripts/gates.py --fix-ast      # also refresh a stale AST snapshot
    python scripts/gates.py --no-fix       # skip the auto-format step

Exit code is 0 only if every gate passed.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
CACHE_PATH = REPO_ROOT / ".cache" / "gates.json"

# Trees whose CONTENT decides whether a gate's verdict can be reused.
_SRC = ("src", "scripts", "tests", "pyproject.toml")

# name -> (argv, input roots). Input roots are hashed to build the
# fingerprint; a gate is re-run only when its own inputs changed.
# The sentinel root "!always" opts a gate OUT of fingerprint caching
# entirely: its subject is filesystem/system state no content hash can see
# (worktrees appear, .venv symlinks get swapped) — a cached PASS would be a
# statement about a world that no longer exists.
ALWAYS_RUN = "!always"

GATES: dict[str, tuple[list[str], tuple[str, ...]]] = {
    "architecture": ([sys.executable, "scripts/architecture_checker.py"], ("src",)),
    # worktree-envs guards RUL-021's poison class ONLY (venv-symlink: a
    # worktree sharing another checkout's venv re-points the shared editable
    # install — ship bonr4xuj5). The detector existed but was wired into
    # NOTHING, which is how the incident escaped: a gate nobody runs is not
    # a gate. Restricted to the deterministic check because venv-missing /
    # env-missing are normal transient states for an agent worktree between
    # `git worktree add` and its `uv sync`, and a battery run racing that
    # window must not flake. Never cached: worktree state is invisible to
    # the content fingerprint.
    "worktree-envs": (
        [
            sys.executable,
            "scripts/check_worktree_envs.py",
            "--checks",
            "venv-symlink",
        ],
        (ALWAYS_RUN,),
    ),
    "check-deps": (
        [
            sys.executable,
            "scripts/serialize_codebase_ast.py",
            "check-deps",
            "--include-tests",
        ],
        _SRC,
    ),
    # external-tools is check-deps' counterpart for NON-PyPI executables
    # (RUL-014): every tool `[tool.orchestrata.external-tools]` declares must
    # be present & new enough, and every literal subprocess argv[0] under
    # src/+scripts/ must be declared or allowlisted. pyproject.toml is an
    # input root in its own right — the gate reads its declared-tools table,
    # its [project.scripts] console-script allowlist, and its [tool.ruff]
    # extend-exclude vendored-path exclusion straight out of it, so editing
    # any of those three tables must invalidate the cached verdict the same
    # way editing src/scripts does.
    "external-tools": (
        [sys.executable, "scripts/check_external_tools.py"],
        ("src", "scripts", "pyproject.toml"),
    ),
    # The Prolog gate layer (plans/prolog-gates-implementation-plan.md). It
    # REPLACES the `standards` entry that used to sit here — same three rules
    # plus infra_key_with_default, over an inferred fact base rather than a
    # per-file walk, with three outcomes instead of two: a stale fact base or a
    # run with no rules loaded exits 2 INVALID, never 0.
    #
    # Its input set is wider than `standards`' was because its facts are wider.
    # The rules and the fact ontology are .pl files under src; the ban registry
    # is .claude/banned-libraries.txt; the ruff select/ignore lists that decide
    # tool_rule_status come from pyproject.toml. Fingerprinting only `src`
    # would let a package be un-banned, or a ruff rule turned off, with the
    # gate cached as "unchanged since last PASS" — the false-green this cache
    # has already produced once for validate-ci.
    "prolog-gate": (
        [sys.executable, "scripts/gate.py", "run"],
        (
            "src",
            "tests",
            "scripts",
            ".claude/banned-libraries.txt",
            "pyproject.toml",
        ),
    ),
    # uncovered-policies is REQ-GATE-011's own drift detector
    # (plans/prolog-gates-implementation-plan.md), wired ADVISORY per the
    # user's "severity tier in ONE channel" ruling (2026-08-19,
    # gate_rules.pl policy_requirement/1): uncovered(Policy) names a policy
    # requirement with no enabled mechanism behind it, and that is DRIFT —
    # no file:line, nothing broken, must stay visible without ever blocking
    # until someone promotes it.
    #
    # This needs NO second severity mechanism here. `gate uncovered`
    # (scripts/gate.py cmd_uncovered) already returns EXIT_OK on every
    # successful query regardless of what it finds, by design — the SAME
    # shape `gate facts --stats` already has (a report, not a verdict). The
    # ordinary ok/fail wiring below (`done.returncode == 0`) therefore
    # already reports this gate OK while the tail-captured stdout still
    # prints the count and every policy name, so the "1" becomes visible on
    # every battery run instead of needing to be asked for — a genuinely
    # RED battery entry would mean `gate uncovered` itself broke (a bad
    # snapshot, a Prolog consult failure), not that a policy is unmechanized.
    # Same input set as prolog-gate, because it queries the identical
    # extracted fact base (policy_mechanism/3, tool_rule_status/3) plus
    # gate_rules.pl's own violation/4 heads.
    "uncovered-policies": (
        [sys.executable, "scripts/gate.py", "uncovered"],
        (
            "src",
            "tests",
            "scripts",
            ".claude/banned-libraries.txt",
            "pyproject.toml",
        ),
    ),
    # test-purity's positional `paths` scope is tests/ ONLY — mock/stub/
    # monkeypatch/skip are RUL-002 test-purity categories (real infra, no
    # doubles) that do not generalize to production code:
    # src/orchestrata_tools/test_generator/{combined_generator,
    # stub_generator}.py legitimately name a `stub_gen` attribute because
    # they GENERATE test-stub files (a first-class AST-toolkit capability),
    # not because they fake a dependency; running the full category set over
    # src/+scripts/ (verified 2026-08-12) flags exactly those two files and
    # nothing else — a false positive, not a defect. banned_import is
    # different in kind: an import of a banned library (fakeredis, pyswip,
    # docker, unittest.mock, grpcio, xdist, ...) is an architecture violation
    # wherever it appears, so --banned-import-only sweeps src/ and scripts/
    # for that ONE category, closing the gap where such an import outside
    # tests/ was invisible to this gate entirely. The banned-libraries
    # registry itself is an input root for the same reason prolog-gate and
    # validate-ci list their own config files: a package added there must
    # invalidate a cached PASS even when no file under tests/src/scripts
    # changed, or the cache reproduces the exact false-green class those
    # comments already document.
    "test-purity": (
        [
            sys.executable,
            "scripts/audit_test_purity.py",
            "tests/",
            "--banned-import-only",
            "src/",
            "scripts/",
        ],
        ("tests", "src", "scripts", ".claude/banned-libraries.txt"),
    ),
    # validate-ci VALIDATES THE CI CONFIG, so the CI config files are part of
    # its input set. Fingerprinting only _SRC meant editing .gitlab-ci.yml or
    # Dockerfile.ci left the gate cached as "unchanged since last PASS" and
    # skipped -- the exact false-green the cache exists to prevent. Caught
    # 2026-08-07 while changing both files.
    "validate-ci": (
        [sys.executable, "scripts/validate_ci.py"],
        (*_SRC, ".gitlab-ci.yml", "docker/Dockerfile.ci"),
    ),
    "env-contract": ([sys.executable, "scripts/check_env_contract.py"], ("src",)),
    # env-subnet-scheme (backlog-413 ruling) fingerprints on the env profiles
    # it validates AND on its own script, because the migration ratchet
    # (PENDING_MIGRATION) lives in the script: flipping an env from pending
    # to enforced is a script edit that must re-run the gate.
    "env-subnet-scheme": (
        [sys.executable, "scripts/check_env_subnet_scheme.py"],
        ("env", "scripts/check_env_subnet_scheme.py"),
    ),
    # prod-separation reads the INSTALLED systemd user units, which are outside
    # the repo, so no input-set fingerprint can detect drift in them. It is
    # fingerprinted on the repo's unit TEMPLATES and always cheap to re-run;
    # caching it on _SRC would let prod be re-pointed at the dev tree without
    # the battery ever noticing.
    "prod-separation": (
        [sys.executable, "scripts/check_prod_tree_separation.py"],
        ("deploy/systemd",),
    ),
    # generated-artifacts (RUL-030 + bidirectional generator provenance) runs
    # ALWAYS, never cached. Its subject is now three things a single content
    # fingerprint cannot jointly cover: git's INDEX state (tracked/ignored,
    # same reasoning as prod-separation above), git's FULL HISTORY (the
    # LOSS-vs-NEVER-EXISTED check walks `git log --all`, which can change
    # when a new ref is fetched with no working-tree diff at all), and the
    # CONTENT of every header under tests/src/docs. This gate is explicitly
    # NON-NEGOTIABLE (module docstring's "AUTHORITY" section): it must never
    # serve a stale cached PASS for a generator that was deleted since the
    # cache was written, so it is exempted from caching entirely rather than
    # trusted to a fingerprint that cannot see everything it depends on.
    "generated-artifacts": (
        [sys.executable, "scripts/check_generated_artifacts.py"],
        (ALWAYS_RUN,),
    ),
    # requirement-corpus fingerprints on BOTH sides of the comparison it makes:
    # the spec corpus that defines requirement ids, and the src/ @traces that
    # cite them. Hashing only one side would let a narrowed loader or a new bad
    # decorator land while the battery served a cached pass -- and a stale pass
    # here is exactly the drift that hid 21 dangling traces for months.
    "requirement-corpus": (
        [sys.executable, "scripts/check_requirement_corpus.py"],
        ("specs", "src", "scripts/generate_rtm_evidence.py"),
    ),
    # feature-loss (user ruling 2026-08-19: "add checks at the gates so we
    # never face the lost code and features again"). Unlike
    # requirement-corpus above, which grades only the CURRENT tree, this
    # gate walks git DELETION history for a REQ-* that lost its only
    # implementation, a src/ symbol whose @traces citation is now
    # unreachable anywhere, and a specs/**|docs/adr/** document deleted
    # while a citation still survives it -- see the module docstring for the
    # full split against requirement-corpus/check_spec_authority/
    # trace_intel.untraced-reqs. Fingerprinted on the same specs/src trees
    # requirement-corpus uses (they define the "current" half of every
    # comparison) plus tests/ (check 4 treats a surviving TEST citation as
    # evidence too), the evidence file behind the RECLASSIFICATION carve-out
    # (plans/feature-loss-invalid-citations.json -- a record added or
    # removed there changes the verdict with no other tree touched, so it
    # must invalidate a cached PASS same as the code does), and its own
    # script, so editing the detector invalidates a cached PASS too.
    "feature-loss": (
        [sys.executable, "scripts/check_feature_loss.py"],
        (
            "specs",
            "src",
            "tests",
            "plans/feature-loss-invalid-citations.json",
            "scripts/check_feature_loss.py",
        ),
    ),
    "wiring": ([sys.executable, "scripts/check_wiring.py"], _SRC),
    "parallel-safety": (
        [sys.executable, "scripts/gate_parallel_safety.py"],
        ("tests",),
    ),
    "prolog-boundary": (
        [sys.executable, "scripts/check_prolog_boundary.py"],
        ("src",),
    ),
    # lexicon-collision compares the intent lexicon against the package/module
    # names in the same tree, so BOTH sides are inputs: `src` covers the
    # vocabulary and the tree layout, `scripts` covers the gate's own exemption
    # list. Renaming a module is enough to create a collision that no
    # vocabulary edit would show.
    "lexicon-collision": (
        [sys.executable, "scripts/check_lexicon_collisions.py"],
        ("src", "scripts"),
    ),
    # lexicon-drift was a CI-ONLY gate until now — the same hole this file
    # already documents for mypy. It matters more than it looks: the parser
    # matches against the GENERATED intent_lexicon.gen.pl, while
    # lexicon-collision above reads IntentVocabulary. If those two drift, the
    # collision gate is checking a vocabulary the parser does not use, and
    # passes while the real lexicon holds a colliding word.
    "lexicon-drift": (
        [sys.executable, "scripts/gen_intent_lexicon.py", "--check"],
        ("src", "scripts"),
    ),
    # The plugin under plugins/orchestrata is GENERATED from claude/ by
    # claude/scripts/gen_plugin.py — never hand-placed. This gate is what
    # makes that generation ENFORCED rather than available: a hook, command,
    # skill, agent, or tool added to claude/ without regenerating fails the
    # battery instead of silently shipping a stale plugin. Same failure class
    # as settings.hooks.json (declared, read by nothing, drifted for months).
    "plugin-drift": (
        [sys.executable, "claude/scripts/gen_plugin.py", "--check"],
        ("claude", "plugins", ".claude-plugin", "pyproject.toml"),
    ),
    # coverage-claude is the RATCHET half of the same "claude/ is production,
    # not scaffolding" ruling plugin-drift enforces above: claude/ (the
    # deployed hooks + tools — task_context_hook.py rewrites every dispatched
    # agent's prompt, route_pytest_to_threads.py decides whether a test
    # command runs at all) is in [tool.coverage.run] source but was NEVER
    # gated on a floor (user ruling 2026-08-19). scripts/coverage_ratchet.py
    # mirrors pyright_ratchet.py's shape (now in codetools:
    # src/codetools/refactor/pyright/ratchet.py): a real
    # ``pytest --cov=claude --cov-report=json`` run, exit 1 ONLY if the
    # percentage DROPS below --min. --min tightened to 19.5 on 2026-08-19
    # (test-writer session): tests/claude/test_instruction_enforcer_hook.py
    # (NEW, ~99% of instruction_enforcer_hook.py) and
    # tests/claude/test_module_context.py (NEW, ~98% of module_context.py)
    # landed real coverage on claude/'s two highest-blast-radius previously-
    # untested modules, moving the measured claude/ scope from 12.96% to
    # 19.80% (stable across two consecutive runs) — 19.5 keeps ~0.3 points of
    # float-jitter headroom, the same proportion the original 12.5 baseline
    # used against its 12.96% measurement. A hard `fail_under = 90` would
    # still be red on arrival for the ~85 files in claude/ this session did
    # not touch — this project has watched a red-on-arrival gate get waived,
    # then ignored, then defanged. Tighten --min by hand as real tests land;
    # never loosen it to paper over a drop.
    "coverage-claude": (
        [
            sys.executable,
            "scripts/coverage_ratchet.py",
            "--scope",
            "claude",
            "--paths",
            "claude",
            "--tests",
            "tests/claude",
            "tests/scripts/test_task_context_hook.py",
            "--min",
            "19.5",
        ],
        ("claude", "tests/claude", "tests/scripts/test_task_context_hook.py"),
    ),
    "ruff": (["uv", "run", "ruff", "check", "src/", "tests/", "scripts/"], _SRC),
    # THE SAME COMMAND CI's `lint` job runs. Until 2026-08-07 mypy was a
    # CI-ONLY gate: this battery could report 10/10 while `uv run mypy src/`
    # was red, so the first sight of a type error was a failed pipeline —
    # which is exactly what happened (16 errors, two files, after a local
    # run reported 0 from a stale incremental cache).
    #
    # --cache-dir is pinned INSIDE the repo on purpose. The default
    # `.mypy_cache` resolves per working directory, and sibling worktrees
    # sharing this checkout's caches have already produced results attributed
    # to the wrong tree. Pinning it makes a gate PASS mean this tree.
    "mypy": (
        ["uv", "run", "mypy", "--cache-dir", ".cache/mypy", "src/"],
        _SRC,
    ),
}


def _fingerprint(roots: tuple[str, ...]) -> str:
    """Content hash of *roots*, from git's own object hashing.

    ``git hash-object`` is content-addressed, so this is insensitive to mtime
    churn (a file rewritten with identical bytes does not invalidate a gate)
    and covers untracked files, which a plain ``git diff`` would miss.
    """
    digest = hashlib.sha256()
    for root in sorted(roots):
        target = REPO_ROOT / root
        if not target.exists():
            continue
        files = (
            sorted(p for p in target.rglob("*") if p.is_file() and _interesting(p))
            if target.is_dir()
            else [target]
        )
        for path in files:
            digest.update(str(path.relative_to(REPO_ROOT)).encode())
            digest.update(hashlib.sha256(path.read_bytes()).digest())
    return digest.hexdigest()


def _interesting(path: Path) -> bool:
    if any(
        part in {"__pycache__", ".pytest_cache", ".ruff_cache"} for part in path.parts
    ):
        return False
    # .json/.md/.sh joined for the plugin-drift gate: the plugin tree is
    # mostly markdown commands/skills, JSON manifests, and shell hooks — a
    # suffix filter that skipped them left that gate's fingerprint blind to
    # the very drift it checks, so a stale plugin could ride a cached PASS.
    # Over-inclusion only costs a re-run; under-inclusion costs a false green.
    return path.suffix in {".py", ".toml", ".pl", ".json", ".md", ".sh"}


def _load_cache() -> dict[str, dict]:
    if not CACHE_PATH.exists():
        return {}
    try:
        return json.loads(CACHE_PATH.read_text())
    except ValueError, OSError:
        return {}  # a corrupt cache must never block a run


def _save_cache(cache: dict[str, dict]) -> None:
    CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
    tmp = CACHE_PATH.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(cache, indent=2, sort_keys=True))
    tmp.replace(CACHE_PATH)  # atomic: a killed run never leaves a partial cache


def autoformat() -> None:
    """Apply the mechanical fixes instead of failing the run over them."""
    subprocess.run(
        ["uv", "run", "ruff", "check", "--fix", "-q", "src/", "tests/", "scripts/"],
        cwd=REPO_ROOT,
        capture_output=True,
    )
    subprocess.run(
        ["uv", "run", "black", "-q", "src/", "tests/", "scripts/"],
        cwd=REPO_ROOT,
        capture_output=True,
    )


def refresh_ast() -> tuple[bool, str]:
    """Re-derive the AST snapshot, then RE-AUGMENT it.

    Both steps, in this order, always. ``ast-sync`` rebuilds
    ``plans/ast-ultimate.json`` from source and drops every ``runtime_calls``
    entry; ``build_runtime_callgraph.py reaugment`` puts them back from the
    recorded runtime capture. Running the first without the second silently
    destroys the runtime call graph the RTM depends on -- measured twice, and
    the reason this is one function rather than two commands to remember.
    """
    sync = subprocess.run(
        ["uv", "run", "ast-sync"], cwd=REPO_ROOT, capture_output=True, text=True
    )
    if sync.returncode != 0:
        return False, f"ast-sync failed:\n{sync.stdout[-2000:]}{sync.stderr[-2000:]}"
    aug = subprocess.run(
        [sys.executable, "scripts/build_runtime_callgraph.py", "reaugment"],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
    )
    if aug.returncode != 0:
        return False, f"reaugment failed:\n{aug.stdout[-2000:]}{aug.stderr[-2000:]}"
    touched = _runtime_call_count()
    if touched == 0:
        return False, (
            "ast-sync ran but runtime_calls is still empty after reaugment — "
            "the runtime call graph was destroyed, not refreshed"
        )
    return True, f"AST refreshed; {touched} functions carry runtime_calls"


def _runtime_call_count() -> int:
    snapshot = REPO_ROOT / "plans" / "ast-ultimate.json"
    if not snapshot.exists():
        return 0
    data = json.loads(snapshot.read_text())
    count = 0
    stack = [data]
    while stack:
        node = stack.pop()
        if isinstance(node, dict):
            if "runtime_calls" in node:
                count += 1
            stack.extend(node.values())
        elif isinstance(node, list):
            stack.extend(node)
    return count


def _run_gate(name: str, argv: list[str]) -> tuple[str, bool, str]:
    started = time.monotonic()
    done = subprocess.run(argv, cwd=REPO_ROOT, capture_output=True, text=True)
    elapsed = time.monotonic() - started
    tail = (done.stdout + done.stderr).strip().splitlines()[-6:]
    return name, done.returncode == 0, f"{elapsed:5.1f}s | " + " ".join(tail)[-300:]


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--all", action="store_true", help="ignore the cache")
    ap.add_argument(
        "--no-fix-ast",
        action="store_true",
        help="do NOT auto-refresh a stale AST snapshot",
    )
    ap.add_argument("--no-fix", action="store_true", help="skip ruff --fix / black")
    ap.add_argument("--only", nargs="+", help="run only these gates")
    args = ap.parse_args()

    if not args.no_fix:
        autoformat()

    cache = {} if args.all else _load_cache()

    # Auto-refresh the AST snapshot when source moved under it. Staleness is
    # the single most frequent gate failure and its fix is mechanical and
    # always identical (ast-sync then reaugment, in that order), so doing it
    # by hand only creates opportunities to do the second step wrong -- or
    # skip it, which silently destroys the runtime call graph. Runs BEFORE
    # the gates so validate-ci sees a fresh snapshot rather than failing and
    # demanding a second invocation.
    if not args.no_fix_ast:
        # Every tree the snapshot covers (ast_pipeline.SOURCE_TREES), not
        # just src: fingerprinting one of three trees means a change under
        # tests/ or scripts/ leaves the cached snapshot looking fresh, and
        # the gate then answers from a fact base that predates the change.
        ast_fp = _fingerprint(("src", "tests", "scripts"))
        # Read through `cache`, not the file: under --all the cache is empty
        # by construction, so the refresh is forced along with everything else.
        if cache.get("__ast__", {}).get("fingerprint") != ast_fp:
            ok, detail = refresh_ast()
            print(
                f"[gates] {'OK  ' if ok else 'FAIL'} ast-refresh      {detail}",
                flush=True,
            )
            if not ok:
                return 1
            cache["__ast__"] = {"fingerprint": ast_fp, "ok": True}
            _save_cache(cache)
    selected = {k: v for k, v in GATES.items() if not args.only or k in args.only}

    to_run: dict[str, list[str]] = {}
    reused: list[str] = []
    prints: dict[str, str] = {}
    for name, (argv, roots) in selected.items():
        if ALWAYS_RUN in roots:
            # Filesystem/system-state gate: no content hash covers its
            # subject, so a cache hit would answer for a vanished world.
            prints[name] = ALWAYS_RUN
            to_run[name] = argv
            continue
        fp = _fingerprint(roots)
        prints[name] = fp
        prev = cache.get(name)
        if prev and prev.get("fingerprint") == fp and prev.get("ok"):
            reused.append(name)
            continue
        to_run[name] = argv

    for name in sorted(reused):
        print(f"[gates] SKIP {name:16s} unchanged since last PASS", flush=True)

    failures: list[str] = []
    if to_run:
        started = time.monotonic()
        with ThreadPoolExecutor(max_workers=len(to_run)) as pool:
            for name, ok, detail in pool.map(
                lambda item: _run_gate(*item), to_run.items()
            ):
                status = "OK  " if ok else "FAIL"
                print(f"[gates] {status} {name:16s} {detail}", flush=True)
                cache[name] = {"fingerprint": prints[name], "ok": ok}
                if not ok:
                    failures.append(name)
        print(
            f"[gates] {len(to_run)} gate(s) in {time.monotonic() - started:.1f}s "
            f"wall ({len(reused)} reused)",
            flush=True,
        )

    _save_cache(cache)
    if failures:
        print(f"[gates] FAILED: {', '.join(sorted(failures))}")
        return 1
    print("[gates] ALL GATES PASS")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())