stack/packages/mosaic/framework/tools/wake/validate-973/check-973.py

#!/usr/bin/env python3
"""check-973.py — computation backend for the #973 validation harness.

Everything here derives from exactly two inputs: the frozen denominator
artifact (denominator-089615f.json) and the SOURCE TEXT of the ten converted
suites at the current tree. It never reads the ledger — set arithmetic against
the runtime trace belongs to validate-973.sh, so the two legs of the
comparison come from independent code paths.

Subcommands (all print sorted, stable output; non-zero exit on any failure):

  expected   The expected coordinate set from the ARTIFACT: one
             "<helper> <file>:<line+7>" row per denominator row (+7 = 3
             converter header lines + 4 lines from the #984 source guard,
             uniform across all ten suites).
             Multi-grep lines stay ONE coordinate.

  static     The converted-site inventory from the SOURCE TEXT at the current
             tree: every non-comment line bearing a has_match/count_lines
             token, as "<helper> <file>:<line>". Independent of the artifact
             row list, so `expected == static` is a real check on the
             conversion, not a tautology. (Amendment ONE, leg 1: the ledger is
             an execution trace, not an inventory — the inventory must come
             from the text.)

  arms       The forced-error arm list: the 19 denominator canaries plus one
             E-form arm (store-ack:733→740, a $(count_lines) capture compared
             afterward — the A6 shape) plus one F-form arm (quarantine:560→567,
             the multi-grep pipeline capture), as "<helper> <file>:<line+7>
             <form>". Both extras are asserted to exist in the artifact with
             the expected form — a renumber that moved them fails here, not
             silently downstream.

  sweep      Residual sweep: the denominator's own classifier (ported from the
             frozen derivation) over the ten suites at the current tree must
             find ZERO unconverted verdict-form grep sites; and, IN THE SAME
             RUN, eight specimens (six per-form + two absorb-branch probes, #985)
             planted into a temp copy of a real
             suite must ALL be found with their correct forms — an instrument
             that reports zero must first be seen finding what it claims to
             find (A5).
"""

import json
import re
import shutil
import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
WAKE = HERE.parent
ART = HERE / "denominator-089615f.json"

HEADER_SHIFT = 7  # 3 converter header lines after SCRIPT_DIR + 4 lines from the
# #984 source guard (1-line `. _wake-common.sh && wake_assert_init` became a 5-line
# guarded block) — both uniform across all ten suites, both above every site.

# The two hand-picked extra arms (base coordinates; forms asserted at load).
EXTRA_ARMS = [
    ("test-wake-store-ack.sh", 733, "E-count-capture"),
    ("test-wake-digest-quarantine.sh", 560, "F-extract-capture"),
]

RX_HELPER = re.compile(r"(^|[^A-Za-z0-9_.-])(has_match|count_lines)([^A-Za-z0-9_.-]|$)")

# ---- classifier, ported verbatim in logic from the frozen denominator
# ---- derivation (docs/journal/fleet/drift-derive-089615f__pepper.py)
RX_FAIL_SAME = re.compile(r"(\|\||&&)\s*fail")
RX_COUNT_SUB = re.compile(r"\$\(.*grep\s+[^)]*-c|\$\(\s*grep\s+-c")
RX_ASSIGN_SUB = re.compile(r'=\s*"?\$\(.*grep')
RX_IF = re.compile(r"^\s*(el)?if\s+.*grep")
RX_GREP = re.compile(r"(^|[^A-Za-z0-9_.-])grep([^A-Za-z0-9_.-]|$)")

# grep in COMMAND position: at line start or after a command separator / subshell
# opener / shell keyword / `!`. Quote-unaware by design — a quoted "grep" after a
# separator reads as a command and lands the line in residual, which fails LOUD;
# the absorb direction (note) is the one that must never fire on a real verdict.
RX_GREP_CMD = re.compile(
    r"(?:^|[;|&(`]|\$\(|\bif\b|\belif\b|\bthen\b|\belse\b|\bdo\b|\bwhile\b|\buntil\b|!)"
    r"\s*grep(?:\s|$)"
)


def classify(lines):
    """Return (sites, dispo). Every line containing the word grep gets a row."""
    sites, dispo = [], []
    n = len(lines)
    for i, raw in enumerate(lines):
        line = raw
        ln = i + 1
        if not RX_GREP.search(line):
            continue
        stripped = line.strip()
        if stripped.startswith("#"):
            dispo.append((ln, "comment", stripped))
            continue
        nxt = ""
        for j in range(i + 1, min(i + 3, n)):
            if lines[j].strip():
                nxt = lines[j].strip()
                break
        if "$(" in line and RX_COUNT_SUB.search(line):
            sites.append((ln, "E-count-capture", stripped))
            continue
        if RX_ASSIGN_SUB.search(line) and "grep -c" not in line:
            sites.append((ln, "F-extract-capture", stripped))
            continue
        if RX_FAIL_SAME.search(line):
            sites.append((ln, "A-same-line", stripped))
            continue
        if stripped.endswith("\\"):
            k = i + 1
            joined = stripped[:-1]
            while k < n:
                cont = lines[k].strip()
                joined += " " + (cont[:-1] if cont.endswith("\\") else cont)
                if not cont.endswith("\\"):
                    break
                k += 1
            if re.search(r"(\|\||&&)\s*fail", joined) or (
                joined.rstrip().endswith(("||", "&&"))
                and k + 1 < n
                and lines[k + 1].strip().startswith("fail")
            ):
                sites.append((ln, "C-cont-backslash", stripped))
                continue
            dispo.append((ln, "backslash-no-fail-continuation", stripped))
            continue
        if stripped.endswith(("||", "&&")) and nxt.startswith("fail"):
            sites.append((ln, "B-cont-operator", stripped))
            continue
        if RX_IF.search(line):
            window = " ".join(lines[j] for j in range(i, min(i + 5, n)))
            if "fail" in window:
                sites.append((ln, "D-if-form", stripped))
                continue
            dispo.append((ln, "if-grep-no-fail-window", stripped))
            continue
        win = " ".join(lines[j] for j in range(max(0, i - 2), min(i + 3, n)))
        if re.search(r"fail", win, re.I):
            dispo.append((ln, "BACKSTOP-HAND-REVIEW", stripped))
        else:
            dispo.append((ln, "no-verdict-context", stripped))
    return sites, dispo


def residual_sites(lines):
    """classify() plus the absorb decision — the ONE path both sweep legs share.

    A classified site is absorbed as a note only when its line carries a wake
    helper token AND the line shows no grep in command position: a converted
    line whose PATTERN argument merely contains the word grep. A helper line
    that also runs a real grep verdict (has_match ... && grep -q SECRET ... &&
    fail) stays residual (#985). Multi-line forms anchor the site at the line
    containing grep, so a command-position grep on a continuation line never
    shares its line with the helper token and stays residual by construction.
    """
    sites, dispo = classify(lines)
    residual, notes = [], []
    for ln, form, text in sites:
        line = lines[ln - 1]
        if RX_HELPER.search(line) and not RX_GREP_CMD.search(line):
            notes.append((ln, form, text))
        else:
            residual.append((ln, form, text))
    return residual, notes, dispo


def load_art():
    art = json.loads(ART.read_text())
    assert art["total"] == 261 == len(art["rows"]), "artifact self-consistency"
    return art


def helper_for(row):
    return "count_lines" if row["form"].startswith("E") else "has_match"


def suite_files(art):
    return sorted({r["file"] for r in art["rows"]})


def cmd_expected():
    art = load_art()
    out = sorted(
        f"{helper_for(r)} {r['file']}:{r['line'] + HEADER_SHIFT}" for r in art["rows"]
    )
    assert len(out) == len(set(out)) == 261, "expected set must be 261 distinct rows"
    print("\n".join(out))
    return 0


def cmd_static():
    art = load_art()
    rows = []
    for f in suite_files(art):
        for i, line in enumerate((WAKE / f).read_text().split("\n"), start=1):
            if line.strip().startswith("#"):
                continue
            m = RX_HELPER.search(line)
            if not m:
                continue
            helper = (
                "count_lines"
                if RX_HELPER.search(line).group(2) == "count_lines"
                else "has_match"
            )
            rows.append(f"{helper} {f}:{i}")
    print("\n".join(sorted(rows)))
    return 0


def cmd_arms():
    art = load_art()
    by_key = {(r["file"], r["line"]): r for r in art["rows"]}
    rows = []
    canaries = [r for r in art["rows"] if r.get("canary")]
    assert len(canaries) == 19, f"expected 19 canaries, artifact has {len(canaries)}"
    for r in canaries:
        rows.append(f"{helper_for(r)} {r['file']}:{r['line'] + HEADER_SHIFT} {r['form']}")
    for f, ln, want_form in EXTRA_ARMS:
        r = by_key.get((f, ln))
        assert r is not None, f"extra arm {f}:{ln} not in artifact — renumbered?"
        assert r["form"] == want_form, f"extra arm {f}:{ln} form {r['form']} != {want_form}"
        rows.append(f"{helper_for(r)} {f}:{ln + HEADER_SHIFT} {r['form']}")
    assert len(rows) == 21
    print("\n".join(rows))
    return 0


# (expected classify form, expected disposition through residual_sites, snippet)
PLANTS = [
    ("A-same-line", "residual", ['grep -q needle haystack || fail "plant-A"']),
    ("B-cont-operator", "residual", ["grep -q needle haystack ||", '  fail "plant-B"']),
    ("C-cont-backslash", "residual", ["grep -q needle \\", '  haystack || fail "plant-C"']),
    ("D-if-form", "residual", ["if ! grep -q needle haystack; then", '  fail "plant-D"', "fi"]),
    ("E-count-capture", "residual", ['[ "$(grep -c needle haystack)" = "1" ] || fail "plant-E"']),
    ("F-extract-capture", "residual", ['val="$(grep needle haystack)"']),
    # G: a converted line that ALSO runs a raw grep verdict — the helper token
    # must not absorb it (#985)
    ("A-same-line", "residual", ['has_match -q needle "$F" && grep -q SECRET "$F" && fail "plant-G"']),
    # H: negative control — helper whose PATTERN argument is the word grep;
    # must be absorbed as a note, never residual
    ("A-same-line", "note", ['has_match -q "grep" haystack || fail "plant-H"']),
]


def cmd_sweep():
    art = load_art()
    bad = 0

    # leg 1: real suites at the current tree must be residual-free
    for f in suite_files(art):
        residual, notes, _dispo = residual_sites((WAKE / f).read_text().split("\n"))
        for ln, form, text in notes:
            # converted line whose PATTERN argument contains the word grep:
            # not an unconverted site, but never silently absorbed either
            print(f"SWEEP-NOTE {f}:{ln} converted line matches grep-token ({form}): {text[:80]}")
        for ln, form, text in residual:
            print(f"SWEEP-RESIDUAL {f}:{ln} {form}: {text[:100]}")
            bad += 1
        print(f"SWEEP {f}: {len(residual)} residual verdict site(s)")

    # leg 2, SAME RUN, SAME PATH as leg 1: the instrument must find every plant
    # with the right form AND the right absorb disposition — plants G/H exercise
    # the absorb branch itself, so this leg must go through residual_sites(),
    # not raw classify()
    donor = suite_files(art)[0]
    with tempfile.TemporaryDirectory() as td:
        planted = Path(td) / donor
        shutil.copy(WAKE / donor, planted)
        base_lines = planted.read_text().split("\n")
        offset = len(base_lines)
        expect = {}
        for form, dispo, snippet in PLANTS:
            expect[offset + 1] = (form, dispo)  # first physical line of each plant
            base_lines.extend(snippet)
            offset = len(base_lines)
        planted.write_text("\n".join(base_lines))
        residual, notes, _ = residual_sites(planted.read_text().split("\n"))
        found = {ln: (form, "residual") for ln, form, _t in residual}
        found.update({ln: (form, "note") for ln, form, _t in notes})
        unexpected = [(ln, form) for ln, form, _t in residual if ln not in expect]
        hits = sum(1 for ln, want in expect.items() if found.get(ln) == want)
        n_plants = len(PLANTS)
        print(f"SWEEP-PLANTS found={hits}/{n_plants} in planted copy of {donor}")
        if hits != n_plants:
            for ln, want in sorted(expect.items()):
                got = found.get(ln, ("<missed>", "<missed>"))
                if got != want:
                    print(f"SWEEP-PLANT-MISS line {ln}: expected {want}, got {got}")
            bad += 1
        if unexpected:
            # the donor is a converted suite: any non-plant RESIDUAL site in the
            # copy contradicts the zero leg 1 just reported on the original
            # (non-plant notes mirror leg 1's treatment: printed there, not bad)
            for ln, form in unexpected:
                print(f"SWEEP-PLANT-UNEXPECTED {donor}(copy):{ln} {form}")
            bad += 1

    return 1 if bad else 0


def main():
    cmds = {
        "expected": cmd_expected,
        "static": cmd_static,
        "arms": cmd_arms,
        "sweep": cmd_sweep,
    }
    if len(sys.argv) != 2 or sys.argv[1] not in cmds:
        sys.exit(f"usage: check-973.py {{{'|'.join(cmds)}}}")
    sys.exit(cmds[sys.argv[1]]())


if __name__ == "__main__":
    main()