010 — CPython .pyc 失效判定 3.14.5：八個位元組的證據，以及它們是關於哪一次事件的

{
  "name": "010-cpython-pyc-invalidation",
  "upstream": "CPython Lib/importlib/_bootstrap_external.py",
  "examined_version": "3.14.5",
  "license": "PSF-2.0",
  "what_it_is": "The rule that decides whether a cached .pyc still describes its source file.",

  "why_this_one": "mssp-d-003 lists one weakness above all others: every instance of the pattern so far is my own code. This is upstream, it is thirty years old, it is on the hot path of every Python import ever run, and CPython replaced the mechanism in PEP 552 for exactly the reason the thread is about.",

  "declared_and_then_remeasured": {
    "_note": "Every number below is re-measured by main.py against the running interpreter. A disagreement is a failure, not a warning. A file of measurements nobody re-reads is a declaration compared to nothing, which is the pattern this whole entry is about.",
    "bootstrap_external_lines": 1562,
    "validate_timestamp_pyc_lines": 26,
    "validate_hash_pyc_lines": 22,
    "header_bytes": 16,
    "evidence_bytes": 8,
    "default_invalidation_mode": "TIMESTAMP"
  },

  "header": {
    "layout": "magic(4) flags(4) field2(4) field3(4)",
    "flags": {
      "0": {"mode": "TIMESTAMP", "field2": "source mtime, low 32 bits", "field3": "source size"},
      "1": {"mode": "UNCHECKED_HASH", "field2_field3": "8-byte source hash, never compared"},
      "3": {"mode": "CHECKED_HASH", "field2_field3": "8-byte source hash, compared on every import"}
    },
    "unsupported_bits": "flags & ~0b11 discards the cache and recompiles from source. It does NOT stop the import, which is what this field claimed until the probe was given a control that could tell the two apart - see the note in SMS/upstream.unsupported_flags."
  },

  "the_finding": "All three modes store eight bytes of evidence in the same two header fields. TIMESTAMP's eight bytes are about the file's metadata. The two hash modes' eight bytes are about the file's bytes. The size of the evidence is identical; what differs is which event it is evidence of.",

  "the_correction_this_forced": "I expected to find a check stuck on one value. TIMESTAMP is not: across four edits that all changed the source, it answers stale twice and fresh twice. It discriminates. It discriminates on whether the file's metadata moved, which is a different event from the one the import cares about. UNCHECKED_HASH is the one that takes a single value - and it is deliberate, flagged, documented, and correct for the deployment it exists for.",

  "sets": {
    "FMS": "this file: the header layout, the flag meanings, and the numbers main.py re-measures",
    "SCL": "which mode this deployment writes, and whether an unconditional validator must be declared",
    "SMS": "the cache: writing a header, resolving flags to a validator, and running the validation",
    "TMS": "one file per validator - timestamp, checked-hash, unchecked-hash - each takes plain data and imports nothing",
    "DMS": "the discrimination table, the comparison against the real interpreter, and what none of it can see"
  },

  "non_goals": [
    "Saying timestamp invalidation is a bug. It is the default, it is fast, it avoids reading the source at all, and PEP 552 kept it as the default when it added the alternative.",
    "Reimplementing importlib. Three comparison functions and a header struct.",
    "Claiming the re-cut is better than upstream's arrangement. It is the same rule with the validator lifted out, which is worth exactly as much as the comparison against the real interpreter says it is."
  ]
}

{
  "mode": "timestamp",
  "model_must_match_upstream": true,
  "unconditional_validator_must_declare_itself": true,
  "_note": "`mode` is what this deployment would write. The report runs all three regardless: a policy able to silence the comparison would make the comparison worthless."
}

"""What this deployment writes, and what it refuses to let pass."""
import json
import pathlib

_C = json.loads((pathlib.Path(__file__).parent / "policy.json").read_text(encoding="utf-8"))

mode = lambda: _C["mode"]                                                       # noqa: E731
model_must_match_upstream = lambda: bool(_C["model_must_match_upstream"])        # noqa: E731
unconditional_must_declare = lambda: bool(_C["unconditional_validator_must_declare_itself"])  # noqa: E731

"""The re-cut: a header, a flags-to-validator resolution, and the validation.

Upstream all three live inside `_bootstrap_external.py` alongside path finders,
loaders, and the zip importer. Lifting the validator out is the only structural
change; the rule itself is copied, and DMS compares the result against the real
interpreter on every run so that "copied" is a measurement rather than a claim.
"""
import importlib
import struct

MAGIC = b"\x00\x00\x0d\x0a"
HEADER = "<4sIII"
HEADER_BYTES = struct.calcsize(HEADER)

VALIDATORS = ["timestamp", "checked_hash", "unchecked_hash"]


def load_validators():
    """Resolve each validator module and check it agrees about its own flags."""
    loaded, problems = {}, []
    for name in VALIDATORS:
        try:
            module = importlib.import_module(f"TMS.validators.{name}")
        except ModuleNotFoundError:
            problems.append(f'validator "{name}" has no module - fail closed')
            continue
        for attribute in ("MODE", "FLAGS", "READS", "ABOUT", "UNCONDITIONAL"):
            if not hasattr(module, attribute):
                problems.append(f"{name} does not declare {attribute}")
        loaded[module.FLAGS] = module
    return loaded, problems


def resolve(flags, loaded):
    """Upstream's rule: bit 0 says hash-based, bit 1 says check it.

    Anything above those two bits stops the import - which main.py checks by
    writing such a header rather than by reading the source and believing it.
    """
    if flags & ~0b11:
        return None, f"flags {flags} set bits outside 0b11 - fail closed"
    module = loaded.get(flags)
    if module is None:
        return None, f"no validator for flags {flags} - fail closed"
    return module, None


def write_header(flags, field2, field3):
    return struct.pack(HEADER, MAGIC, flags, field2, field3)


def read_header(blob):
    magic, flags, field2, field3 = struct.unpack(HEADER, blob[:HEADER_BYTES])
    return {"magic": magic, "flags": flags, "field2": field2, "field3": field3}


def header_for(module, observed):
    """What each mode would store about a source in this state."""
    if module.MODE == "TIMESTAMP":
        return write_header(module.FLAGS, observed["mtime"] & 0xFFFFFFFF, observed["size"])
    return write_header(module.FLAGS, *observed["hash"])


def validate(blob, observed, loaded):
    header = read_header(blob)
    module, problem = resolve(header["flags"], loaded)
    if problem:
        return None, problem
    return module.is_fresh(header, observed), None

"""What the real interpreter does, measured — not what its source code says it does.

Every row here is a subprocess that imports a module whose source changed after
its .pyc was written, and prints the value that actually ran. The re-cut in
cache.py is compared against these results cell by cell, so "the re-cut copies
upstream's rule" is a measurement that can come out false.
"""
import importlib.util
import os
import py_compile
import shutil
import struct
import subprocess
import sys
import tempfile

FIRST = "VALUE = 'AAA'\n"
SAME_SIZE = "VALUE = 'BBB'\n"       # same byte length as FIRST
LONGER = "VALUE = 'BBBB'\n"         # one byte longer

MODES = {
    "timestamp": py_compile.PycInvalidationMode.TIMESTAMP,
    "checked_hash": py_compile.PycInvalidationMode.CHECKED_HASH,
    "unchecked_hash": py_compile.PycInvalidationMode.UNCHECKED_HASH,
}

EDITS = [
    {"id": "natural", "label": "two writes, whatever the clock did", "to": SAME_SIZE, "mtime": "leave"},
    {"id": "restored", "label": "mtime put back, same size", "to": SAME_SIZE, "mtime": "restore"},
    {"id": "resized", "label": "mtime put back, size changed", "to": LONGER, "mtime": "restore"},
    {"id": "advanced", "label": "mtime forced +5s, same size", "to": SAME_SIZE, "mtime": "forward"},
]


def observe(path):
    stat = os.stat(path)
    fields = struct.unpack("<II", importlib.util.source_hash(open(path, "rb").read()))
    return {"mtime": int(stat.st_mtime), "size": stat.st_size, "hash": fields}


def probe(edit, mode_name):
    """Compile, edit the source, import in a fresh interpreter, report what ran."""
    tmp = tempfile.mkdtemp()
    try:
        source = os.path.join(tmp, "probe_mod.py")
        with open(source, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(FIRST)
        cached = importlib.util.cache_from_source(source)
        py_compile.compile(source, cfile=cached, invalidation_mode=MODES[mode_name], doraise=True)

        before = observe(source)
        header = open(cached, "rb").read(16)
        stat = os.stat(source)
        with open(source, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(edit["to"])
        if edit["mtime"] == "restore":
            os.utime(source, (stat.st_atime, stat.st_mtime))
        elif edit["mtime"] == "forward":
            os.utime(source, (stat.st_atime, stat.st_mtime + 5))
        after = observe(source)

        result = subprocess.run([sys.executable, "-c", "import probe_mod; print(probe_mod.VALUE)"],
                                cwd=tmp, capture_output=True, text=True)
        ran = result.stdout.strip()
        wanted = edit["to"].split("'")[1]
        return {
            "ran": ran,
            "ran_stale": ran != wanted,
            "before": before,
            "after": after,
            "header": header,
            "metadata_moved": before["mtime"] != after["mtime"] or before["size"] != after["size"],
            "bytes_moved": before["hash"] != after["hash"],
        }
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def measured_facts():
    """The numbers FMS declares, taken from the interpreter running right now."""
    import inspect

    import importlib._bootstrap_external as bootstrap

    return {
        "bootstrap_external_lines": sum(1 for _ in open(bootstrap.__file__, encoding="utf-8")),
        "validate_timestamp_pyc_lines": len(inspect.getsource(bootstrap._validate_timestamp_pyc).splitlines()),
        "validate_hash_pyc_lines": len(inspect.getsource(bootstrap._validate_hash_pyc).splitlines()),
        "header_bytes": 16,
        "evidence_bytes": 8,
        "default_invalidation_mode": py_compile._get_default_invalidation_mode().name,
    }


def unsupported_flags(flag_value):
    """What an unreadable flags field does to the cache.

    The first version of this probe left the source matching the cache, so
    "cache used" and "cache rejected" printed the same string and the
    measurement could not tell them apart. It reported exit 0 and I nearly
    filed that as confirmation of a claim in FMS. It is the same defect this
    whole entry is about, committed while writing about it.

    So: the source is edited to disagree with the cache first, and the mtime is
    put back. Reading AAA now means the cache was used; reading BBB means it was
    rejected and the source recompiled. Pass flag_value=0 for the control.
    """
    tmp = tempfile.mkdtemp()
    try:
        source = os.path.join(tmp, "probe_mod.py")
        with open(source, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(FIRST)
        cached = importlib.util.cache_from_source(source)
        py_compile.compile(source, cfile=cached, doraise=True)

        stat = os.stat(source)
        with open(source, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(SAME_SIZE)
        os.utime(source, (stat.st_atime, stat.st_mtime))

        blob = bytearray(open(cached, "rb").read())
        blob[4:8] = struct.pack("<I", flag_value)
        with open(cached, "wb") as handle:
            handle.write(blob)

        result = subprocess.run([sys.executable, "-c", "import probe_mod; print(probe_mod.VALUE)"],
                                cwd=tmp, capture_output=True, text=True)
        ran = result.stdout.strip()
        return {"flags": flag_value, "exit": result.returncode, "ran": ran,
                "cache_used": ran == "AAA", "recompiled": ran == "BBB",
                "stderr_tail": result.stderr.strip().splitlines()[-1] if result.stderr.strip() else ""}
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

"""PEP 552. The two header fields hold an eight-byte hash of the source bytes.

Upstream this is `_validate_hash_pyc`, 22 lines, and it is four lines shorter
than the timestamp validator while answering a strictly harder question. The
cost is not in the comparison - it is that somebody had to read the source file
to compute the hash to compare against.
"""
MODE = "CHECKED_HASH"
FLAGS = 3
READS = ["an 8-byte hash of the source bytes"]
ABOUT = "the file's bytes"
UNCONDITIONAL = False


def is_fresh(header, observed):
    return (header["field2"], header["field3"]) == observed["hash"]

"""The default. Compares the header's two fields against the source file's metadata.

Upstream this is `_validate_timestamp_pyc`, 26 lines, and it reads the source
file's stat - never its contents. That is the whole point of it: it is fast
because it does not open the source.
"""
MODE = "TIMESTAMP"
FLAGS = 0
READS = ["source mtime, low 32 bits", "source size"]
ABOUT = "the file's metadata"
UNCONDITIONAL = False


def is_fresh(header, observed):
    if header["field2"] != (observed["mtime"] & 0xFFFFFFFF):
        return False
    return header["field3"] == observed["size"]

"""The header carries a hash and nothing ever compares it.

This is not an oversight. It is for deployments where a build system already
guarantees the .pyc matches, and paying to read every source file at import
time buys nothing. The hash is still recorded so a tool outside the import
system can check it later.

A validator that can never refuse is allowed. What makes this one legitimate
rather than broken is that the refusal to check is declared in the header's own
flag bits, where anything can read it.
"""
MODE = "UNCHECKED_HASH"
FLAGS = 1
READS = []
ABOUT = "the file's bytes, as of whenever the cache was written"
UNCONDITIONAL = True


def is_fresh(header, observed):
    return True

"""The discrimination table, the comparison against the real interpreter, and the gaps."""


def table(rows, out):
    out(f"\n  {'the edit':<34} {'metadata':<10} {'bytes':<7} {'timestamp':<11} {'checked':<9} unchecked")
    for row in rows:
        cells = [f"{row['verdict'][mode]:<11}" for mode in ("timestamp", "checked_hash")]
        out(f"  {row['label']:<34} {row['metadata']:<10} {row['bytes']:<7} "
            f"{cells[0]}{cells[1][:9]:<9} {row['verdict']['unchecked_hash']}")


def distinct_values(rows, out):
    out("\n  how many answers each validator gave across four edits that ALL changed the source:")
    for mode in ("timestamp", "checked_hash", "unchecked_hash"):
        values = sorted({row["verdict"][mode] for row in rows})
        out(f"    {mode:<15} {len(values)}  {', '.join(values)}")


def evidence_about(validators, out):
    out("\n  every mode stores eight bytes in the same two header fields:")
    for module in validators:
        reads = ", ".join(module.READS) or "nothing at import time"
        out(f"    {module.MODE:<15} flags={module.FLAGS}  about {module.ABOUT}")
        out(f"    {'':<15} reads {reads}")


def gaps(out):
    out("\n  measurable, not measured here:")
    out("    - how often a real edit lands inside one mtime second in ordinary work")
    out("    - what checked-hash costs on a large import graph, in wall clock")
    out("\n  not measurable by this entry at all:")
    out("    - whether upstream's default is wrong. It is the default because reading")
    out("      every source file at import time is a real cost, and PEP 552 left it")
    out("      the default when it added the alternative. Nothing here weighs that.")
    out("    - whether anyone has ever been bitten by it in production. Reproducing a")
    out("      failure is not evidence of its frequency.")

"""The island test, and the drill that shows the comparison against upstream can fail.

    python src/island_test.py

Section 3 compares the re-cut validator against the interpreter that is running
right now, twelve cells. Section 3b then breaks the re-cut on purpose and
requires the comparison to notice - because a comparison that has never been
seen to fail is a claim, not a check (改良點 6).
"""
import pathlib
import re
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from SCL import policy  # noqa: E402
from SMS import cache, upstream  # noqa: E402

FAILURES = []


def check(label, ok, detail=""):
    print(f"  {'PASS' if ok else 'FAIL'}  {label}{' - ' + detail if detail else ''}")
    if not ok:
        FAILURES.append(label)


LOADED, PROBLEMS = cache.load_validators()
BY_NAME = {module.MODE.lower(): module for module in LOADED.values()}

print("\n== 1. every validator is an island and declares what it reads")
check("all three validators loaded with no problems", not PROBLEMS, "; ".join(PROBLEMS))
validator_dir = HERE / "TMS" / "validators"
files = sorted(f for f in validator_dir.iterdir() if f.suffix == ".py" and f.name != "__init__.py")
check("there are three validator files", len(files) == 3, ", ".join(f.stem for f in files))
for f in files:
    reaches = re.findall(r"^\s*(?:from|import)\s+(\S+)", f.read_text(encoding="utf-8"), re.M)
    check(f"{f.stem} imports nothing", not reaches, ", ".join(reaches) or "no imports at all")
for module in LOADED.values():
    check(f"{module.MODE} says what its evidence is about", bool(module.ABOUT), module.ABOUT)
unconditional = [m for m in LOADED.values() if m.UNCONDITIONAL]
check("exactly one validator declares itself unconditional", len(unconditional) == 1,
      ", ".join(m.MODE for m in unconditional))
check("and it reads nothing at import time, which is why", unconditional[0].READS == [],
      repr(unconditional[0].READS))

print("\n== 2. flags select a validator, and unknown bits do not fall through")
observed = {"mtime": 1000, "size": 10, "hash": (1, 2)}
header = cache.read_header(cache.write_header(0, 1000 & 0xFFFFFFFF, 10))
answers = {}
for flags in (0, 1, 3):
    module, problem = cache.resolve(flags, LOADED)
    answers[flags] = module.is_fresh(header, observed)
check("the same header and state give different answers under different flags",
      len(set(answers.values())) > 1, ", ".join(f"flags {f}={v}" for f, v in answers.items()))
_, problem = cache.resolve(0b100, LOADED)
check("flags outside 0b11 resolve to nothing", problem is not None, problem or "resolved anyway")
_, problem = cache.resolve(2, LOADED)
check("flags=2 (check-source without hash-based) resolves to nothing",
      problem is not None, problem or "resolved anyway")

print("\n== 3. the re-cut against the real interpreter, twelve cells")
CELLS = []
for edit in upstream.EDITS:
    for mode_name in upstream.MODES:
        measured = upstream.probe(edit, mode_name)
        module = BY_NAME[mode_name]
        head = cache.read_header(cache.header_for(module, measured["before"]))
        CELLS.append({"edit": edit["id"], "mode": mode_name, "module": module,
                      "header": head, "before": measured["before"], "after": measured["after"],
                      "upstream_used_cache": measured["ran_stale"],
                      "model_fresh": module.is_fresh(head, measured["after"])})
agree = [c for c in CELLS if c["model_fresh"] == c["upstream_used_cache"]]
check(f"all {len(CELLS)} cells agree", len(agree) == len(CELLS),
      f"{len(agree)}/{len(CELLS)} agree")

print("\n== 3b. the drill: can that comparison come out false?")


def broken_is_fresh(header, observed):
    """timestamp, but comparing the source hash instead of the metadata."""
    return (header["field2"], header["field3"]) == observed["hash"]


broken = [c for c in CELLS if c["mode"] == "timestamp"
          and broken_is_fresh(c["header"], c["after"]) != c["upstream_used_cache"]]
check("swapping timestamp's comparison for the hash one breaks cells", bool(broken),
      f"{len(broken)} of 4 timestamp cells disagree once the rule is wrong")

print("\n== 4. the header the re-cut writes is the header py_compile wrote")
# Two independent producers of the same eight bytes. If these agree it is
# because both read the same source, not because one copied the other.
for mode_name in upstream.MODES:
    measured = upstream.probe(upstream.EDITS[1], mode_name)
    module = BY_NAME[mode_name]
    mine = cache.header_for(module, measured["before"])
    theirs = measured["header"]
    check(f"{mode_name}: bytes 4:16 match py_compile's", mine[4:16] == theirs[4:16],
          f"{mine[4:16].hex()} vs {theirs[4:16].hex()}")

print("\n== 5. how many answers each validator gives, and about what")
for mode_name in upstream.MODES:
    values = sorted({("used" if c["model_fresh"] else "rejected")
                     for c in CELLS if c["mode"] == mode_name})
    module = BY_NAME[mode_name]
    print(f"        {mode_name:<15} {len(values)} answer(s): {', '.join(values):<16} about {module.ABOUT}")
timestamp_values = {c["model_fresh"] for c in CELLS if c["mode"] == "timestamp"}
check("timestamp is not stuck on one answer", len(timestamp_values) == 2,
      "it discriminates - on metadata, which is not the event the import cares about")
unchecked_values = {c["model_fresh"] for c in CELLS if c["mode"] == "unchecked_hash"}
check("unchecked-hash IS stuck on one answer, and declares it",
      len(unchecked_values) == 1 and BY_NAME["unchecked_hash"].UNCONDITIONAL,
      "flags=1 is that declaration, readable by anything")

print("\n== 6. the probe that could not tell two things apart, and its control")
control = upstream.unsupported_flags(0)
unreadable = upstream.unsupported_flags(0b100)
check("with flags=0 the stale cache is used", control["cache_used"], f"ran {control['ran']}")
check("with flags=0b100 the cache is rejected and the source recompiled",
      unreadable["recompiled"], f"ran {unreadable['ran']}")
check("the two runs differ, which is the only reason either means anything",
      control["ran"] != unreadable["ran"], f"{control['ran']} vs {unreadable['ran']}")
check("and neither stops the import - the first version of this probe claimed it did",
      control["exit"] == 0 and unreadable["exit"] == 0,
      f"exits {control['exit']} and {unreadable['exit']}")

print("\n== 7. what this entry cannot see")
print("        MEASURABLE, NOT MEASURED")
print("          - how often a real edit lands inside one mtime second in ordinary work")
print("          - what checked-hash costs on a large import graph")
print("        NOT MEASURABLE HERE")
print("          - whether the default is wrong. Reading every source file at import")
print("            time is a real cost and PEP 552 left timestamp the default. Nothing")
print("            here weighs the two against each other.")
print("          - frequency in the wild. Reproducing a failure says nothing about")
print("            how often anyone meets it.")

print(f"\n{len(FAILURES)} failure(s)" if FAILURES else "\nall checks passed")
for f in FAILURES:
    print(f"  - {f}")
sys.exit(1 if FAILURES else 0)

"""CPython's .pyc invalidation, re-cut into MSSP sets and checked against the real thing.

    python src/main.py            the discrimination table and the comparison
    python src/main.py --strict   exit 1 if the re-cut disagrees with the interpreter

Twelve subprocesses run: four edits that all change the source, under three
invalidation modes. Takes a couple of seconds.
"""
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from DMS import report  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import cache, upstream  # noqa: E402

ARCH = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))


def collect():
    """Measure upstream, run the re-cut over the same states, and pair them up."""
    loaded, problems = cache.load_validators()
    by_name = {module.MODE.lower(): module for module in loaded.values()}
    rows, disagreements = [], []

    for edit in upstream.EDITS:
        verdicts, model_verdicts = {}, {}
        metadata = bytes_moved = None
        for mode_name in upstream.MODES:
            measured = upstream.probe(edit, mode_name)
            metadata = "moved" if measured["metadata_moved"] else "unchanged"
            bytes_moved = "changed" if measured["bytes_moved"] else "same"

            module = by_name[mode_name]
            header = cache.read_header(cache.header_for(module, measured["before"]))
            model_fresh = module.is_fresh(header, measured["after"])

            verdicts[mode_name] = "STALE RAN" if measured["ran_stale"] else "recompiled"
            model_verdicts[mode_name] = model_fresh
            if model_fresh != measured["ran_stale"]:
                disagreements.append(
                    f"{edit['id']}/{mode_name}: re-cut says fresh={model_fresh}, "
                    f"interpreter ran {'stale' if measured['ran_stale'] else 'fresh'} code")

        rows.append({"id": edit["id"], "label": edit["label"], "metadata": metadata,
                     "bytes": bytes_moved, "verdict": verdicts, "model": model_verdicts})
    return rows, [module for _, module in sorted(loaded.items())], disagreements, problems


def main(argv):
    out = lambda line="": sys.stdout.write(line + "\n")  # noqa: E731
    rows, validators, disagreements, problems = collect()
    if problems:
        for problem in problems:
            out(f"  !! {problem}")
        return 1

    out(f"\n== four edits, every one of them a change to the source  "
        f"[{ARCH['upstream'].split('/')[-1]} {ARCH['examined_version']}]")
    report.table(rows, out)
    report.distinct_values(rows, out)
    report.evidence_about(validators, out)

    out("\n== the re-cut against the interpreter that is running right now")
    if disagreements:
        for line in disagreements:
            out(f"    DISAGREES  {line}")
    else:
        out(f"    {len(rows) * 3} of {len(rows) * 3} cells agree - the lifted validator decides "
            f"what the import system decided")

    out("\n== the numbers FMS declares, re-measured")
    declared = ARCH["declared_and_then_remeasured"]
    measured = upstream.measured_facts()
    wrong = []
    for key, value in measured.items():
        agrees = declared.get(key) == value
        if not agrees:
            wrong.append(f"{key}: declared {declared.get(key)!r}, measured {value!r}")
        out(f"    {'ok ' if agrees else 'NO '} {key:<32} {value}")

    out("\n== what an unreadable flags field does, against a control")
    # The control is the point. Without a run where the cache IS used, "it
    # printed BBB" means nothing - see the note in upstream.unsupported_flags.
    control = upstream.unsupported_flags(0)
    unreadable = upstream.unsupported_flags(0b100)
    for probe, label in ((control, "flags=0 (timestamp, stale cache)"),
                         (unreadable, "flags=0b100 (bits nothing defines)")):
        outcome = "cache used, stale code ran" if probe["cache_used"] else (
            "cache rejected, source recompiled" if probe["recompiled"] else "neither: " + probe["ran"])
        out(f"    {label:<38} exit {probe['exit']}  ran {probe['ran']:<5} {outcome}")
    out("    so unsupported flags do not stop the import - they discard the cache")

    report.gaps(out)

    if "--strict" in argv and (wrong or (disagreements and policy.model_must_match_upstream())):
        for line in wrong:
            out(f"\n  !! {line}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))