llm-ingestion-okf/tools/okf_gate_mutants.py

"""Mutation run over the content-accounting gate and its witness.

The gate judges `okf build`; this asks what judges the gate. Each mutant is ONE
textual edit that makes the instrument weaker in a way a reader would call a
defect, and the question is whether the suite goes red. A mutant that survives
names a check nothing holds.

The twenty-six mutants are an independent review's, ported to the code as it
stands rather than re-invented: at `0b00de4` twelve of them survived the
forty-two tests, among them `main` always returning 0 and six of the seven
witness mutants -- which had no fixture that could exercise the element they
removed.

    python3 tools/okf_gate_mutants.py

Runs on a COPY of the tree in a temporary directory: the working tree is never
edited, so an interrupted run cannot leave a mutant behind. A pattern that does
not match the expected number of times is reported as an ERROR and the run
exits 2 -- a mutant that could not be applied was never measured, and counting
it as killed is the same mistake as reading an empty search as an absence.

A SURVIVOR EXITS 1. Until 2026-09-19 the run ended `2 if errors else 0`, so
`killed 0 of 1` with the survivor printed beside it was an exit 0 and nothing
could fail on the finding this harness exists to produce (H4).
"""

from __future__ import annotations

import shutil
import subprocess
import sys
import tempfile
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
GATE = "tools/okf_accounting_gate.py"
WITNESS = "tools/okf_witness.py"
SUITE = "tests/test_accounting_gate.py"


@dataclass(frozen=True)
class Mutant:
    label: str
    file: str
    old: str
    new: str
    #: How many times `old` must occur. A role map written twice on purpose
    #: (`_sts_role_xml`, `_sts_role_json`) has two, and the mutant edits the
    #: first -- the XML witness -- so the JSON one stays honest and row 5 has
    #: a chance to see the disagreement.
    occurrences: int = 1
    first_only: bool = False


MUTANTS: tuple[Mutant, ...] = (
    Mutant(
        "M01 m=0 counts as green ('not measured' -> GREEN)",
        GATE,
        "GREEN if m > 0 and k == m else RED",
        "GREEN if k == m else RED",
    ),
    Mutant(
        "M02 row 6 never fails the gate",
        GATE,
        "self.status == RED and self.number <= 6",
        "self.status == RED and self.number <= 5",
    ),
    Mutant(
        "M03 a missing corpus is SKIPPED locally too",
        GATE,
        "status = SKIPPED if ci else RED",
        "status = SKIPPED",
    ),
    Mutant(
        "M04 double booking of elements never counted",
        GATE,
        "double += booked - have",
        "double += 0",
    ),
    Mutant(
        "M05 'declared carried, bytes absent' never a false claim",
        GATE,
        'false_claim = fate == "carried" and not carried',
        "false_claim = False",
    ),
    Mutant(
        "M06 an unpointed duplicate counts as carried",
        GATE,
        "carried = bool(pointed_by) and asset_holds(build, corpus / name)",
        "carried = asset_holds(build, corpus / name)",
    ),
    Mutant(
        "M07 row 2 accepts ANY declared inventory",
        GATE,
        "if got != want:",
        "if got is None:",
    ),
    Mutant(
        "M08 row 1 ignores a stale committed fasit",
        GATE,
        'if fresh["documents"].get(name) == entry:',
        "if True:",
    ),
    Mutant(
        "M09 row 4 ignores the Images 'found' count",
        GATE,
        "elif int(found.group(2)) != declared_images:",
        "elif False:",
    ),
    Mutant(
        "M10 row 4 accepts any element total in the log line",
        GATE,
        'rf"{re.escape(doc)}: {total} elements found in the source, 0 carried: "',
        'rf"{re.escape(doc)}: \\d+ elements found in the source, 0 carried: "',
    ),
    Mutant(
        "M11 an unavailable witness is agreement",
        GATE,
        'return ["a witness is unavailable"]',
        "return []",
    ),
    Mutant(
        "M12 row 5 counts every pair as agreeing",
        GATE,
        "good = sum(1 for _, problems in pairs if not problems)",
        "good = len(pairs)",
    ),
    Mutant(
        "M13 row 6: clean in ANY of the two builds is enough",
        GATE,
        "if all(unit.clean for unit in parts)",
        "if any(unit.clean for unit in parts)",
    ),
    Mutant(
        "M14 main always exits 0",
        GATE,
        "    if any(r.fails for r in rows):\n        return 1",
        "    if False:\n        return 1",
    ),
    Mutant(
        "M15 T silently drops .rtf",
        GATE,
        'return sorted(set(re.findall(r"^\\| `(\\.[a-z0-9]+)` \\|", section, re.MULTILINE)))',
        'return sorted(\n        set(re.findall(r"^\\| `(\\.[a-z0-9]+)` \\|", section, re.MULTILINE)) - {".rtf"}\n    )',
    ),
    Mutant(
        "M16 an element with no declared fate is ignored",
        GATE,
        "for element in sorted(set(elements) | set(fates)):",
        "for element in sorted(set(fates)):",
    ),
    Mutant(
        "M17 conservation always 'held'",
        GATE,
        'return build.exit_code == 0 and "K1b FAILED" not in build.log',
        "return True",
    ),
    Mutant(
        "M18 row 3 counts every unit as clean",
        GATE,
        "clean = sum(1 for u in units if u.clean)",
        "clean = len(units)",
    ),
    Mutant(
        "M19 a negative rejection is absorbed into the total",
        GATE,
        "booked = carried + pointer + sum(rejected.values())",
        "booked = carried + pointer + abs(sum(rejected.values()))",
    ),
    # --- the judge's own reading of the bundle, added 2026-09-18 -------------
    Mutant(
        "M20 a booked carry is never looked for in the bundle",
        GATE,
        "found = sum(1 for pieces in sayable if all(finder(piece) for piece in pieces))",
        "found = len(elements_pieces)",
    ),
    Mutant(
        "M21 an unverified booking is clean",
        GATE,
        "            self.unaccounted or self.double or self.unverified "
        "or self.invalid or self.refused",
        "            self.unaccounted or self.double",
    ),
    # H1's own column, mutated on its own: M21 above removes `unverified`,
    # `invalid` AND `refused` at once, so it would be killed by any one of the
    # three. This one takes only the fifth.
    Mutant(
        "X3 a document refused whole is clean again",
        GATE,
        '    if status == "rejected" and not persisted and total > 0:',
        "    if False:",
    ),
    Mutant(
        "M22 any asset file at all proves a carry",
        GATE,
        "        found == digest and name.startswith(digest[:12])"
        " for name, found in build.assets.items()",
        "        True for name, found in build.assets.items()",
    ),
    Mutant(
        "M23 a corpus refused whole is not reported",
        GATE,
        "    if persisted:\n        return None",
        "    if True:\n        return None",
    ),
    Mutant(
        "M24 a rejection code outside the closed list is accepted",
        GATE,
        "        unknown = sorted(c for c in rejected if c not in REJECTION_CODES)",
        "        unknown = []",
    ),
    Mutant(
        "M25 a skipped row leaves the verdict unqualified",
        GATE,
        '        verdict += f" (row {row.number} not run: {row.reason})"',
        '        verdict += ""',
    ),
    Mutant(
        "W01 docx: separator footnotes counted",
        WITNESS,
        'if role != "comment" and int(note.get(f"{_W}id", "0")) <= 0:',
        'if role != "comment" and int(note.get(f"{_W}id", "0")) <= -1:',
    ),
    Mutant(
        "W02 pptx: pictures never counted",
        WITNESS,
        '            for _ in root.iter(f"{_P}pic"):\n                count.add("image")',
        '            for _ in ():\n                count.add("image")',
    ),
    Mutant(
        "W03 sts (xml witness): footnotes have no role",
        WITNESS,
        '    if tag == "fn":\n        return "footnote"',
        '    if tag == "fn":\n        return None',
        occurrences=2,
        first_only=True,
    ),
    Mutant(
        "W04 xlsx: images never counted",
        WITNESS,
        '                for _ in root.iter(f"{_XDR}pic"):\n                    count.add("image")',
        '                for _ in ():\n                    count.add("image")',
    ),
    Mutant(
        "W05 odt: list items never counted",
        WITNESS,
        '    for item in root.iter(f"{_TEXT}list-item"):',
        "    for item in ():",
    ),
    Mutant(
        "W06 docx: tables never counted",
        WITNESS,
        '    for table in root.iter(f"{_W}tbl"):',
        "    for table in ():",
    ),
    Mutant(
        "W07 sts (xml witness): table cells have no role",
        WITNESS,
        '    if tag in ("td", "th"):\n        return "cell"',
        '    if tag in ("td", "th"):\n        return None',
        occurrences=2,
        first_only=True,
    ),
    Mutant(
        "W08 docx: a header or footer is never counted",
        WITNESS,
        '                    count.add("header_footer", *lines)',
        "                    pass",
    ),
    Mutant(
        "W09 sts json: the twin reuses the XML map",
        WITNESS,
        "        role = _sts_role_json(tag, parent, grandparent)",
        "        role = _sts_role_xml(tag, parent, grandparent)",
    ),
    # PM's own mutant, written outside this file 2026-09-18 and SURVIVING 155
    # green tests (H2). B-1 gave the judge six refusals; this was the one
    # nothing held from either side.
    Mutant(
        "X2 a report may declare a document rejected while the bundle holds it",
        GATE,
        '    if status == "rejected" and persisted:\n        invalid += 1',
        "    if False:\n        invalid += 1",
    ),
)


def _apply(text: str, mutant: Mutant) -> str:
    return text.replace(mutant.old, mutant.new, 1 if mutant.first_only else -1)


def verdict(survived: Sequence[str], errors: Sequence[str]) -> int:
    """The run's exit code: 0 clean, 1 a mutant survived, 2 one was not measured.

    Until 2026-09-19 this was `2 if errors else 0`, so a run that printed
    `killed 0 of 1` and named its survivor exited 0 and no caller could fail
    on it (H4). A mutant that survived names a check nothing holds, which is
    the finding this harness exists to produce; a mutant that could not be
    applied was never measured at all, and that outranks it.
    """
    if errors:
        return 2
    return 1 if survived else 0


def main(argv: list[str] | None = None) -> int:
    del argv
    survived: list[str] = []
    killed = 0
    errors: list[str] = []
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp) / "tree"
        shutil.copytree(
            REPO,
            root,
            ignore=shutil.ignore_patterns(".git", ".venv", "__pycache__", "*.egg-info"),
        )
        for mutant in MUTANTS:
            path = root / mutant.file
            original = path.read_text(encoding="utf-8")
            seen = original.count(mutant.old)
            if seen != mutant.occurrences:
                errors.append(
                    f"{mutant.label}: pattern occurs {seen} times, expected "
                    f"{mutant.occurrences} -- NOT MEASURED"
                )
                continue
            path.write_text(_apply(original, mutant), encoding="utf-8")
            try:
                run = subprocess.run(
                    [sys.executable, "-m", "pytest", SUITE, "-q", "-x", "-p", "no:cacheprovider"],
                    cwd=root,
                    capture_output=True,
                    text=True,
                )
            finally:
                path.write_text(original, encoding="utf-8")
            tail = (run.stdout.strip().splitlines() or [run.stderr.strip()[-160:]])[-1]
            if run.returncode != 0:
                killed += 1
                print(f"killed   {mutant.label}  [{tail}]")
            else:
                survived.append(mutant.label)
                print(f"SURVIVED {mutant.label}  [{tail}]")
    print(f"\nkilled {killed} of {len(MUTANTS)}")
    for label in survived:
        print(f"  survived: {label}")
    for problem in errors:
        print(f"  ERROR: {problem}")
    return verdict(survived, errors)


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