check_generated_artefacts.py

Go to the documentation of this file.

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Brighton Sikarskie
"""Gate: versioned generated gap artefacts must match a fresh regenerate.
 
The MC/DC and Doxygen gap reports under ``docs/`` are GENERATED, yet they are
also COMMITTED, and nothing compared the committed copy against what the
generator produces from the current tree. On a DO-178C Level B target these are
the human-readable record of where the remaining structural-coverage and
documentation gaps are; a committed copy that has silently drifted describes a
tree that no longer exists (issue #380).
 
This is the "regenerate and gate" resolution -- the same shape
``check_ci_parity.py`` uses on the gate registry. For every versioned generated
artefact this gate re-runs its generator and byte-compares the result against
the tracked worktree candidate; any difference fails the gate and names the
command that refreshes it. Reading the candidate rather than ``HEAD`` makes the
gate useful before a refresh is committed, while the tracked-file check still
rejects untracked or deleted evidence. Deleting the versioned copies
(produce-in-CI-only)
was the rejected alternative: the DO-178C qualification set
(``docs/qualification/PSAC.md``, ``SVR.md``, ``SAS.md``, ``SQAP.md`` ...) cites
these files by path as versioned, retained evidence, so the in-tree history is
load-bearing and cannot move to an ephemeral CI artefact.
 
Two artefact families, checked uniformly:
 
* **mcdc-gaps** -- ``docs/MCDC_GAPS.csv``, ``docs/MCDC_GAPS.md`` and
  ``docs/MCDC_DEACTIVATIONS.md``, regenerated by
  ``scripts/fix/regen_mcdc_gaps.py`` from ``build/mcdc-report/mcdc.txt``. That
  report is the (slow) output of the ``mcdc`` gate, so this gate CONSUMES it and
  must be scheduled after ``mcdc`` in the same job; if the report is absent it
  FAILS LOUDLY naming the dependency rather than skipping.
* **doxygen-gaps** -- ``docs/DOXYGEN_GAPS.csv`` and ``docs/DOXYGEN_GAPS.md``,
  regenerated by ``scripts/checks/doxy_audit.py`` from a static source parse
  (no toolchain input required).
* **driver-status** -- ``docs/DRIVER_STATUS.md``, regenerated by
  ``scripts/gen/gen_driver_status.py`` from the HAL source inventory. This one
  used to be hand-maintained, and by the time it was replaced 33 of its 109 rows
  named a deleted file and it still described a library that had been removed
  (issue #721). Deriving it is only half the fix; gating the derivation is the
  other half.
 
``--selftest`` proves the detector still fires in both directions -- a matching
regenerate passes, a mismatching one fails, and a missing required input fails
without running the generator -- so a gate that quietly stopped comparing cannot
pass as clean.
"""
 
from __future__ import annotations
 
import argparse
import contextlib
import io
import subprocess
import sys
import tempfile
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from types import ModuleType
 
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "dev"))
 
from git_environment import isolated_git_environment, trusted_git_executable
 
REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPTS_CHECKS = REPO_ROOT / "scripts" / "checks"
SCRIPTS_FIX = REPO_ROOT / "scripts" / "fix"
SCRIPTS_GEN = REPO_ROOT / "scripts" / "gen"
MCDC_TXT = REPO_ROOT / "build" / "mcdc-report" / "mcdc.txt"
 
 
@dataclass(frozen=True)
class ArtefactGroup:
    """One generator and the versioned artefacts it owns.
 
    Attributes:
        name: Short identifier printed in gate output.
        artefacts: Repo-relative POSIX paths the generator writes and that are
            versioned in the tree.
        regenerate: Runs the generator in place and returns its exit code (0 on
            success); it writes exactly ``artefacts``.
        requires: Input files that must exist before ``regenerate`` can run. An
            absent input fails the gate loudly instead of skipping the family.
        remediation: The command a human runs to refresh the versioned copies.
    """
 
    name: str
    artefacts: tuple[str, ...]
    regenerate: Callable[[], int]
    requires: tuple[Path, ...]
    remediation: str
 
 
def _read_candidate(root: Path, rel: str) -> bytes | None:
    """Return tracked worktree bytes for ``rel``; reject untracked/deleted copies."""
    argv = [
        "git",
        "-C",
        str(root),
        "ls-files",
        "--error-unmatch",
        "--",
        rel,
    ]
    result = subprocess.run(  # noqa: S603 -- fixed git argv, no shell
        argv,
        capture_output=True,
        check=False,
    )
    if result.returncode != 0:
        return None
    path = root / rel
    return path.read_bytes() if path.is_file() else None
 
 
def _diff_artefact(candidate: bytes | None, fresh: bytes) -> str | None:
    """Describe how a tracked candidate differs from a fresh regenerate.
 
    Args:
        candidate: Tracked worktree bytes, or None when absent or untracked.
        fresh: Bytes the generator just produced.
 
    Returns:
        A one-line drift description, or None when the two are byte-identical.
    """
    if candidate is None:
        return "candidate copy is absent or untracked; cannot verify retained evidence"
    if candidate == fresh:
        return None
    return f"stale: candidate {len(candidate)} bytes differ from the {len(fresh)}-byte regenerate"
 
 
def _restore(root: Path, saved: dict[str, bytes | None]) -> None:
    """Put every artefact back to the bytes captured before regeneration."""
    for rel, data in saved.items():
        path = root / rel
        if data is None:
            path.unlink(missing_ok=True)
        else:
            path.write_bytes(data)
 
 
def _check_group(
    group: ArtefactGroup,
    root: Path,
    read_candidate: Callable[[Path, str], bytes | None],
) -> list[str]:
    """Regenerate one group and return a drift message per stale artefact.
 
    Restores every artefact to its pre-run bytes afterwards, so the check never
    leaves the working tree dirtier than it found it. A missing required input
    is a hard failure -- the generator is never run -- rather than a skip.
    """
    missing = [str(p) for p in group.requires if not p.exists()]
    if missing:
        joined = ", ".join(missing)
        return [
            f"{group.name}: required input(s) absent: {joined}. This gate consumes them "
            f"and must run after the gate that produces them. {group.remediation}"
        ]
 
    candidate = {rel: read_candidate(root, rel) for rel in group.artefacts}
    saved: dict[str, bytes | None] = {}
    for rel in group.artefacts:
        path = root / rel
        saved[rel] = path.read_bytes() if path.exists() else None
 
    errors: list[str] = []
    try:
        rc = group.regenerate()
        if rc != 0:
            errors.append(f"{group.name}: regenerator exited {rc} (see its output above)")
        else:
            for rel in group.artefacts:
                drift = _diff_artefact(candidate[rel], (root / rel).read_bytes())
                if drift is not None:
                    errors.append(f"{rel}: {drift}. {group.remediation}")
    finally:
        _restore(root, saved)
    return errors
 
 
def _regenerate_mcdc() -> int:
    """Rewrite docs/MCDC_GAPS.* from build/mcdc-report/mcdc.txt; return its rc."""
    if str(SCRIPTS_FIX) not in sys.path:
        sys.path.insert(0, str(SCRIPTS_FIX))
    import regen_mcdc_gaps  # noqa: PLC0415 -- lazy: adds scripts/fix to sys.path first
 
    with contextlib.redirect_stdout(io.StringIO()):
        return regen_mcdc_gaps.main()
 
 
def _regenerate_doxygen() -> int:
    """Rewrite docs/DOXYGEN_GAPS.* from a static source parse; return its rc."""
    if str(SCRIPTS_CHECKS) not in sys.path:
        sys.path.insert(0, str(SCRIPTS_CHECKS))
    import doxy_report  # noqa: PLC0415 -- lazy: adds scripts/checks to sys.path first
 
    with contextlib.redirect_stdout(io.StringIO()):
        return doxy_report.run_report()
 
 
def _load_driver_status() -> ModuleType:
    """Import gen_driver_status, putting scripts/gen on the path first."""
    if str(SCRIPTS_GEN) not in sys.path:
        sys.path.insert(0, str(SCRIPTS_GEN))
    import gen_driver_status  # noqa: PLC0415 -- lazy: adds scripts/gen to sys.path first
 
    return gen_driver_status
 
 
def _regenerate_driver_status() -> int:
    """Rewrite docs/DRIVER_STATUS.md from the HAL inventory; return its rc."""
    module = _load_driver_status()
    with contextlib.redirect_stdout(io.StringIO()):
        return module.write(REPO_ROOT)
 
 
def _production_groups() -> list[ArtefactGroup]:
    """The versioned generated gap artefacts this gate enforces freshness on."""
    return [
        ArtefactGroup(
            name="mcdc-gaps",
            artefacts=(
                "docs/MCDC_GAPS.csv",
                "docs/MCDC_GAPS.md",
                "docs/MCDC_DEACTIVATIONS.md",
            ),
            regenerate=_regenerate_mcdc,
            requires=(MCDC_TXT,),
            remediation=(
                "Run `just quality::local::mcdc` (or `just quality::gate::run mcdc`) and "
                "commit the regenerated docs/MCDC_GAPS.csv, docs/MCDC_GAPS.md and "
                "docs/MCDC_DEACTIVATIONS.md."
            ),
        ),
        ArtefactGroup(
            name="doxygen-gaps",
            artefacts=("docs/DOXYGEN_GAPS.csv", "docs/DOXYGEN_GAPS.md"),
            regenerate=_regenerate_doxygen,
            requires=(),
            remediation=(
                "Run `python3 scripts/checks/doxy_audit.py` and commit the regenerated "
                "docs/DOXYGEN_GAPS.csv and docs/DOXYGEN_GAPS.md."
            ),
        ),
        ArtefactGroup(
            name="driver-status",
            artefacts=("docs/DRIVER_STATUS.md",),
            regenerate=_regenerate_driver_status,
            requires=(),
            remediation=(
                "Run `python3 scripts/gen/gen_driver_status.py` and commit the "
                "regenerated docs/DRIVER_STATUS.md."
            ),
        ),
    ]
 
 
def check() -> int:
    """Fail when any tracked candidate artefact differs from a fresh regenerate."""
    errors: list[str] = []
    for group in _production_groups():
        errors.extend(_check_group(group, REPO_ROOT, _read_candidate))
    if errors:
        sys.stderr.write("check_generated_artefacts.py: candidate gap artefacts are stale.\n\n")
        for error in errors:
            sys.stderr.write(f"  {error}\n\n")
        sys.stderr.write(f"{len(errors)} stale artefact(s).\n")
        return 1
    print(
        "check_generated_artefacts.py: clean -- every tracked candidate gap artefact "
        "matches a fresh regenerate."
    )
    return 0
 
 
def _synthetic_generator(path: Path, content: bytes) -> Callable[[], int]:
    """Build a regenerate() that writes fixed ``content`` to ``path`` and succeeds."""
 
    def _regenerate() -> int:
        path.write_bytes(content)
        return 0
 
    return _regenerate
 
 
def _counting_generator(sink: list[int], path: Path) -> Callable[[], int]:
    """Build a regenerate() that records each call in ``sink`` before writing."""
 
    def _regenerate() -> int:
        sink.append(1)
        path.write_bytes(b"regenerated\n")
        return 0
 
    return _regenerate
 
 
def _run_git(root: Path, *args: str) -> None:
    """Run a git subcommand in ``root``, raising on non-zero exit."""
    subprocess.run(  # noqa: S603 -- fixed argv, no shell
        [trusted_git_executable(), "-C", str(root), *args],
        check=True,
        capture_output=True,
    )
 
 
def _seed_driver_fixture(root: Path) -> None:
    """Generate the driver-status fixture so git commits it with the seed."""
    real = _load_driver_status()
    hal = root / "hal-fixture"
    with contextlib.redirect_stdout(io.StringIO()):
        real._seed_tree(hal)  # noqa: SLF001 -- the generator's own fixture builder
        real.write(hal)
    target = root / real.ARTEFACT
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes((hal / real.ARTEFACT).read_bytes())
 
 
def _driver_status_cases(root: Path) -> list[tuple[str, bool]]:
    """Drive the REAL driver-status generator through the freshness comparator.
 
    The synthetic groups prove the comparator; they say nothing about whether a
    real generator is wired to it. A freshness check whose selftest only ever
    exercised a stand-in would keep passing after the real one stopped producing
    what it used to. So this regenerates against the throwaway versioned HAL at
    seed time and asserts clean, then adds a driver and asserts drift.
 
    Args:
        root: The throwaway git repository seeded by ``_selftest_cases``.
 
    Returns:
        The two ``(label, passed)`` tuples, in both directions.
    """
    real = _load_driver_status()
    hal = root / "hal-fixture"
    driver_rel = real.ARTEFACT
 
    def _regen_driver() -> int:
        (root / driver_rel).write_bytes((hal / driver_rel).read_bytes())
        return 0
 
    group = ArtefactGroup("driver-real-match", (driver_rel,), _regen_driver, (), "n/a")
    clean = _check_group(group, root, _read_candidate) == []
 
    (hal / "libs" / "ra8_hal" / "inc" / "ra8_extra.h").write_text(
        "#pragma once\nra8_err_t ra8_extra_init(void);\n", encoding="ascii"
    )
    with contextlib.redirect_stdout(io.StringIO()):
        real.write(hal)
    drift = bool(_check_group(group, root, _read_candidate))
 
    return [
        ("driver-status: fresh regenerate matches candidate -> clean", clean),
        ("driver-status: a new driver changes the page -> drift", drift),
    ]
 
 
def _seed_selftest_repo(root: Path) -> str:
    """Create and commit the synthetic freshness candidate."""
    tracked = "docs/SYNTH_GAPS.md"
    (root / "docs").mkdir(parents=True, exist_ok=True)
    (root / tracked).write_bytes(b"committed-AAA\n")
    _seed_driver_fixture(root)
    _run_git(root, "init", "-q")
    _run_git(root, "add", "-A")
    _run_git(
        root, "-c", "user.email=ci@localhost", "-c", "user.name=ci", "commit", "-q", "-m", "seed"
    )
    return tracked
 
 
def _synthetic_selftest_groups(
    root: Path, tracked: str
) -> tuple[dict[str, ArtefactGroup], list[int]]:
    """Build the synthetic group variants and the missing-input run probe."""
    ran: list[int] = []
    untracked = "docs/NOT_COMMITTED.md"
    groups = {
        "match": ArtefactGroup(
            "synth-match",
            (tracked,),
            _synthetic_generator(root / tracked, b"committed-AAA\n"),
            (),
            "n/a",
        ),
        "drift": ArtefactGroup(
            "synth-drift",
            (tracked,),
            _synthetic_generator(root / tracked, b"regenerated-BBB\n"),
            (),
            "n/a",
        ),
        "candidate": ArtefactGroup(
            "synth-candidate",
            (tracked,),
            _synthetic_generator(root / tracked, b"candidate-CCC\n"),
            (),
            "n/a",
        ),
        "missing": ArtefactGroup(
            "synth-missing",
            (tracked,),
            _counting_generator(ran, root / tracked),
            (root / "absent-input.txt",),
            "n/a",
        ),
        "untracked": ArtefactGroup(
            "synth-untracked",
            (untracked,),
            _synthetic_generator(root / untracked, b"y\n"),
            (),
            "n/a",
        ),
    }
    return groups, ran
 
 
def _selftest_cases(root: Path) -> list[tuple[str, bool]]:
    """Exercise every freshness direction against a throwaway git repo."""
    tracked = _seed_selftest_repo(root)
    groups, ran = _synthetic_selftest_groups(root, tracked)
 
    match_clean = _check_group(groups["match"], root, _read_candidate) == []
    drift_seen = bool(_check_group(groups["drift"], root, _read_candidate))
 
    (root / tracked).write_bytes(b"candidate-CCC\n")
    candidate_clean = _check_group(groups["candidate"], root, _read_candidate) == []
    (root / tracked).unlink()
    deleted_drift = bool(_check_group(groups["match"], root, _read_candidate))
    (root / tracked).write_bytes(b"committed-AAA\n")
 
    missing_errors = _check_group(groups["missing"], root, _read_candidate)
    untracked_drift = bool(_check_group(groups["untracked"], root, _read_candidate))
 
    return [
        ("fresh matches tracked candidate -> clean", match_clean),
        ("fresh differs from tracked candidate -> drift", drift_seen),
        ("unstaged generated refresh matching fresh output -> clean", candidate_clean),
        ("deleted tracked candidate -> drift", deleted_drift),
        ("missing required input -> fail, generator not run", bool(missing_errors) and not ran),
        ("untracked candidate copy -> drift", untracked_drift),
        *_driver_status_cases(root),
    ]
 
 
def _selftest_body() -> int:
    """Prove the detector fires in both directions and never silently skips."""
    with tempfile.TemporaryDirectory() as tmp:
        cases = _selftest_cases(Path(tmp))
    failures = 0
    for label, passed in cases:
        status = "ok" if passed else "FAIL"
        if not passed:
            failures += 1
        print(f"  [{status}] {label}")
    if failures:
        sys.stderr.write(f"check_generated_artefacts.py --selftest: {failures} case(s) failed.\n")
        return 1
    print("check_generated_artefacts.py --selftest: all cases pass.")
    return 0
 
 
def selftest() -> int:
    """Run generated-artifact fixtures without inheriting the caller's repo."""
    with isolated_git_environment():
        return _selftest_body()
 
 
def main() -> int:
    """Parse arguments and dispatch to the freshness check or its selftest."""
    parser = argparse.ArgumentParser(
        description="Verify tracked candidate gap artefacts match a fresh regenerate."
    )
    parser.add_argument(
        "--selftest",
        action="store_true",
        help="run the detector's self-test (both directions) and exit",
    )
    args = parser.parse_args()
    if args.selftest:
        return selftest()
    return check()
 
 
if __name__ == "__main__":
    raise SystemExit(main())