#!/usr/bin/env python3
"""Green-before / red-after, with the four silent controls made loud.

This repo does not accept "the test is load-bearing" as a claim. It mutates
what the test rests on, runs the test, and requires it to go RED -- then
restores the tree and proves the restore was byte-identical. That procedure was
hand-built ten times before it became this file, and the reason it became a
file is not the typing. It is that four of its controls fail SILENTLY:

  1. AN ANCHOR THAT IS NOT UNIQUE mutates more than the seam, so the redness is
     not attributable to the seam.
  2. A MISTYPED NODE ID never runs the test. pytest exits 4, which is non-zero,
     so a harness that asks "rc != 0?" reads a typo as a proof. (MEASURED
     2026-08-25: green rc=0, real failure rc=1, unknown id or broken collection
     rc=4, nothing collected rc=5. And `--collect-only -q` exits 0 on an id
     that does not exist, so it is not an existence check either.)
  3. AN UNVERIFIED RESTORE leaves the tree mutated and says "restored". Under
     `.venv/` nothing is tracked, so `git status` will not catch it.
  4. RED ON THE WRONG LINE -- a NameError or a broken import -- is achievable by
     any garbage edit and proves nothing about the seam.

Each one turns a broken query into a positive-looking fact, which is the defect
class the operating model names. Prose cannot enforce them; this can.

THREE ROLES, AND THE ORDER FOLLOWS FROM WHAT EACH ONE MEASURES. The rule
"positive controls before negatives" was never in tension with this procedure;
it only looked that way because "the control" was read as one thing when it is
three:

  * BEFORE, mandatory -- the measuring apparatus. Every node id is collected and
    GREEN right now. Without this, "red after" may be red because the id was
    always wrong. This IS the positive control, and it runs first.
  * AFTER, by definition -- the measurement. "Can this test go red at all?" is
    not answerable until the mutation exists. This is not a control that ran
    late; it is the thing being measured.
  * BOTH, must stay green -- the population/collateral control. If it reds, the
    mutation landed wider than the seam and the target's redness is void.

SCOPE. This is for mutating the REAL tree and running pytest as a subprocess.
The other mutation class -- mutate a copy in memory, call the guard directly --
already lives in the suite as ``test_guard_red_when_*`` and needs no tool. This
one deliberately does NOT live in the suite: the suite runs every session, and a
fixture that writes to disk turns every interrupted run into a mutated tree.

Usage:

    python scripts/mutation_harness.py \
        --target .venv/lib/python3.x/site-packages/some/module.py \
        --anchor 'the exact text, unique in the file' \
        --replacement 'what detaches the seam' \
        --red  tests/test_x.py::TestY::test_the_seam \
        --green tests/test_x.py::TestY::test_the_population
"""

from __future__ import annotations

import argparse
import hashlib
import re
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, Mapping, Sequence

Verdict = Literal["green", "red", "not-collected"]

# MEASURED 2026-08-25, pytest 8, this repo's .venv. The point of the table is
# that 4 and 5 are NOT red -- they are the apparatus reporting that it never
# measured anything.
_GREEN = 0
_FAILED = 1
_USAGE_OR_COLLECTION_ERROR = 4
_NOTHING_COLLECTED = 5

# Reads the exception NAME off pytest's `E   <dotted.Name>: <msg>` lines. Kept
# deliberately dumb: it reports what pytest emitted, and nothing downstream is
# allowed to gate on the value (see CONTROL 4). `E   assert 1 == 2` has no
# colon-terminated name and correctly yields nothing.
_ERROR_LINE = re.compile(r"^E\s+([A-Za-z_][\w.]*)\s*:", re.MULTILINE)


class HarnessError(Exception):
    """Base for every refusal. A refusal is a result, never a silent pass."""


class AnchorNotUnique(HarnessError):
    """The anchor does not occur exactly once, so the mutation is not targeted."""


class NodeNotCollected(HarnessError):
    """pytest never ran the id. Non-zero here mimics red and must not be read as it."""


class NotGreenBefore(HarnessError):
    """A target was already red, so redness after is not caused by the mutation."""


class ControlWentRed(HarnessError):
    """A control reddened: the mutation landed wider than the seam."""


class NotAValueProof(HarnessError):
    """The mutation ran, and the seam did not care. That is a finding, not an error."""


class RestoreFailed(HarnessError):
    """The bytes on disk are not the bytes we started with. Loudest possible failure."""


@dataclass(frozen=True)
class Outcome:
    node_id: str
    returncode: int
    verdict: Verdict
    error_types: tuple[str, ...]
    output: str = ""

    @property
    def summary(self) -> str:
        kinds = ", ".join(self.error_types) if self.error_types else "no named exception"
        return f"{self.verdict} (rc={self.returncode}, {kinds})"


@dataclass(frozen=True)
class Mutation:
    target: Path
    anchor: str
    replacement: str


@dataclass(frozen=True)
class Report:
    mutation: Mutation
    before: tuple[Outcome, ...]
    after: tuple[Outcome, ...]
    sha256_before: str
    sha256_after: str
    targets: tuple[str, ...] = field(default_factory=tuple)
    controls: tuple[str, ...] = field(default_factory=tuple)
    pins: dict[str, str] = field(default_factory=dict)

    @property
    def after_by_id(self) -> dict[str, Outcome]:
        return {o.node_id: o for o in self.after}

    @property
    def before_by_id(self) -> dict[str, Outcome]:
        return {o.node_id: o for o in self.before}

    @property
    def value_proved(self) -> bool:
        after = self.after_by_id
        return bool(self.targets) and all(after[n].verdict == "red" for n in self.targets)


def classify(returncode: int, output: str, node_id: str = "") -> Outcome:
    """Map a pytest exit code to a verdict, keeping "never ran" out of "red".

    The whole tool turns on this function. Everything else is bookkeeping.
    """
    if returncode == _GREEN:
        verdict: Verdict = "green"
    elif returncode == _FAILED:
        verdict = "red"
    elif returncode in (_USAGE_OR_COLLECTION_ERROR, _NOTHING_COLLECTED):
        verdict = "not-collected"
    else:
        # Unknown code: refuse to guess which side of the line it falls on.
        verdict = "not-collected"
    return Outcome(
        node_id=node_id,
        returncode=returncode,
        verdict=verdict,
        error_types=tuple(dict.fromkeys(m.rsplit(".", 1)[-1] for m in _ERROR_LINE.findall(output))),
        output=output,
    )


def run_node(node_id: str, cwd: Path) -> Outcome:
    completed = subprocess.run(
        [sys.executable, "-m", "pytest", node_id, "-q", "--no-header", "-p", "no:cacheprovider"],
        cwd=cwd,
        capture_output=True,
        text=True,
    )
    return classify(completed.returncode, completed.stdout + completed.stderr, node_id)


def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _verify_restore(target: Path, original: bytes, expected_sha: str) -> str:
    """Write the original bytes back and prove from DISK that they are there.

    Reads the file again rather than trusting the write, because the failure
    this exists to catch is precisely a write that did not fully land.
    """
    target.write_bytes(original)
    actual = _sha256(target.read_bytes())
    if actual != expected_sha:
        raise RestoreFailed(
            f"restore of {target} is NOT byte-identical: expected {expected_sha}, got {actual}. "
            "The tree is dirty -- fix it by hand before trusting anything else."
        )
    return actual


def prove(
    mutation: Mutation,
    expect_red: Sequence[str],
    expect_green: Sequence[str] = (),
    *,
    red_at: Mapping[str, str] | None = None,
    cwd: Path | None = None,
) -> Report:
    """Run the full procedure, refusing loudly at every control it fails."""
    targets = tuple(expect_red)
    controls = tuple(expect_green)
    pins = dict(red_at or {})
    if not targets:
        raise ValueError(
            "a proof with no --red target reds nothing and proves nothing; name at least one"
        )

    work_dir = Path(cwd) if cwd is not None else Path.cwd()
    target_file = mutation.target

    # CONTROL 1 -- the anchor, checked before a single byte is written.
    original = target_file.read_bytes()
    text = original.decode("utf-8")
    occurrences = text.count(mutation.anchor)
    if occurrences != 1:
        raise AnchorNotUnique(
            f"anchor must occur exactly once in {target_file}; measured {occurrences} "
            f"occurrence(s). A wider anchor mutates more than the seam."
        )
    sha_before = _sha256(original)

    node_ids = list(dict.fromkeys([*targets, *controls]))

    # CONTROL 2 -- the measuring apparatus, and it runs FIRST.
    before = tuple(run_node(n, work_dir) for n in node_ids)
    for outcome in before:
        if outcome.verdict == "not-collected":
            raise NodeNotCollected(
                f"pytest never ran {outcome.node_id} (rc={outcome.returncode}). "
                "That exit is non-zero but it is not red -- the id is wrong, or "
                "collection is broken. Nothing measured."
            )
        if outcome.verdict == "red":
            raise NotGreenBefore(
                f"{outcome.node_id} was ALREADY red before the mutation, so red after "
                "is not caused by it. Fix the tree, then measure."
            )

    mutated = text.replace(mutation.anchor, mutation.replacement, 1)
    try:
        target_file.write_text(mutated, encoding="utf-8")
        after = tuple(run_node(n, work_dir) for n in node_ids)
    finally:
        # CONTROL 3 -- the restore, verified from disk, in `finally` or not at all.
        sha_after = _verify_restore(target_file, original, sha_before)

    report = Report(
        mutation=mutation,
        before=before,
        after=after,
        sha256_before=sha_before,
        sha256_after=sha_after,
        targets=targets,
        controls=controls,
        pins=pins,
    )
    results = report.after_by_id

    for node_id in controls:
        if results[node_id].verdict != "green":
            raise ControlWentRed(
                f"control {node_id} went {results[node_id].verdict} under the mutation. "
                "The mutation landed wider than the seam, so the target's redness "
                "attributes to nothing."
            )

    for node_id in targets:
        outcome = results[node_id]
        if outcome.verdict == "not-collected":
            raise NodeNotCollected(
                f"after the mutation, pytest could not collect {node_id} "
                f"(rc={outcome.returncode}). Collection broke -- which any garbage edit "
                "achieves. That is the loudest way to prove nothing."
            )
        if outcome.verdict == "green":
            raise NotAValueProof(
                f"{node_id} stayed GREEN under the mutation. The seam it claims to guard "
                "is detached and the test did not notice -- green-but-dead, measured."
            )
        # CONTROL 4 -- red, but WHERE? Not derivable from the exception type: a
        # legitimately red test may die as AssertionError, as `Failed: DID NOT
        # RAISE`, or as a custom exception. So the caller PINS the line the
        # proof is about and this checks the pin against pytest's real output.
        # Unpinned, the type is reported instead of silently blessed.
        pin = pins.get(node_id)
        if pin is not None and pin not in outcome.output:
            raise NotAValueProof(
                f"{node_id} went red ({outcome.summary}), but the failure output does "
                f"not contain the pinned location {pin!r}. It died somewhere else than "
                "the line this proof is about."
            )

    return report


def _format(report: Report) -> str:
    lines = [
        f"VALUE-PROVED  {report.mutation.target}",
        f"  anchor      {report.mutation.anchor!r} -> {report.mutation.replacement!r}",
        f"  restored    sha256 {report.sha256_after} (byte-identical, verified from disk)",
    ]
    for node_id in report.targets:
        outcome = report.after_by_id[node_id]
        pinned = f", pinned at {report.pins[node_id]!r}" if node_id in report.pins else ""
        lines.append(f"  TARGET      green before / red after   {node_id}")
        lines.append(f"              died of: {outcome.summary}{pinned}")
    for node_id in report.controls:
        lines.append(f"  CONTROL     green before / green after {node_id}")
    return "\n".join(lines)


def main(argv: Sequence[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="mutation_harness.py",
        description="Prove a test is load-bearing: green before, red after, restore verified.",
    )
    parser.add_argument("--target", required=True, type=Path, help="file to mutate in place")
    parser.add_argument("--anchor", required=True, help="exact text, must occur exactly once")
    parser.add_argument("--replacement", required=True, help="what the anchor becomes")
    parser.add_argument(
        "--red",
        action="append",
        default=[],
        metavar="NODE_ID",
        help="test that must be green before and red after (repeatable)",
    )
    parser.add_argument(
        "--green",
        action="append",
        default=[],
        metavar="NODE_ID",
        help="control that must stay green in both runs (repeatable)",
    )
    parser.add_argument(
        "--red-at",
        action="append",
        default=[],
        metavar="NODE_ID=TEXT",
        help="require this text in that target's failure output (repeatable)",
    )
    parser.add_argument("--cwd", type=Path, default=None, help="directory to run pytest from")
    args = parser.parse_args(argv)

    try:
        report = prove(
            Mutation(target=args.target, anchor=args.anchor, replacement=args.replacement),
            expect_red=args.red,
            expect_green=args.green,
            red_at=dict(pin.split("=", 1) for pin in args.red_at),
            cwd=args.cwd,
        )
    except HarnessError as exc:
        print(f"{type(exc).__name__}: {exc}")
        if isinstance(exc, NotAValueProof):
            print("NOT a value proof.")
        return 1
    except ValueError as exc:
        print(f"ValueError: {exc}")
        return 2

    print(_format(report))
    return 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())