rune/scripts/research_harness_invariants.py

"""research_harness_invariants.py — #411: the R1 repair classes as
deterministic invariants.

The repeated repair families (#396-#403 era) showed strong error DETECTION
but insufficient error PREVENTION. This module converts each recurring
repair class into a checkable invariant every harness entry point can run
in one call, plus a fixture corpus pinning the classes as regression tests.

Invariant classes (each maps to a repair family):
  I1 SCHEMA-SHAPE      — artifact has the required round sections
  I2 OUTWARD-EVIDENCE  — OUTWARD carries >= 2 live-verified URLs w/ lessons
  I3 INNOVATION-GROUND — INNOVATION candidates reference OUTWARD sources
  I4 SYNTHESIS-QUALIFY — SYNTHESIS has evidence-sufficiency + kill test
  I5 LEDGER-VALIDITY   — evidence-ledger lines are JSON with required keys
  I6 NO-PHANTOM-PATHS  — every referenced artifact path exists on disk

Language verdict (dichotomy law): deterministic validation over research
TOOLING artifacts — batch/reactive tooling, not runtime backend state —
Python placement; a Rust core would parse the same JSON the harness
already holds in-process. Recorded for the spine.

  python3 scripts/research_harness_invariants.py <artifact.json> [--ledger FILE]
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

REQUIRED_SECTIONS = ("round", "outward", "innovation", "synthesis")
REQUIRED_LEDGER_KEYS = ("ts", "kind", "ref")


def check(artifact: dict, ledger_path: Path | None = None) -> list[str]:
    """Run every invariant. Returns findings (empty == clean)."""
    f: list[str] = []
    f += _i1_schema_shape(artifact)
    f += _i2_outward_evidence(artifact)
    f += _i3_innovation_ground(artifact)
    f += _i4_synthesis_qualify(artifact)
    if ledger_path is not None:
        f += _i5_ledger_validity(ledger_path)
    f += _i6_no_phantom_paths(artifact)
    return f


def _i1_schema_shape(a: dict) -> list[str]:
    missing = [s for s in REQUIRED_SECTIONS if s not in a]
    return [f"I1 SCHEMA-SHAPE: missing section(s) {missing}"] if missing else []


def _i2_outward_evidence(a: dict) -> list[str]:
    outward = a.get("outward") or []
    if not isinstance(outward, list):
        return ["I2 OUTWARD-EVIDENCE: outward is not a list"]
    live = [
        s
        for s in outward
        if isinstance(s, dict)
        and str(s.get("url", "")).startswith(("http://", "https://"))
        and str(s.get("lesson", "")).strip()
    ]
    if len(live) < 2:
        return [f"I2 OUTWARD-EVIDENCE: only {len(live)} live-verified source(s) with lesson annotations (need >= 2)"]
    return []


def _i3_innovation_ground(a: dict) -> list[str]:
    innovation = a.get("innovation") or {}
    candidates = innovation.get("candidates") if isinstance(innovation, dict) else None
    if not candidates:
        return []  # empty INNOVATION is legal on skipped rounds
    outward_urls = {str(s.get("url", "")).strip() for s in (a.get("outward") or []) if isinstance(s, dict)}
    ungrounded = [
        c
        for c in candidates
        if isinstance(c, dict)
        and not (str(c.get("basis", "")).strip() and any(u and u in str(c.get("basis", "")) for u in outward_urls))
    ]
    if ungrounded:
        return [f"I3 INNOVATION-GROUND: {len(ungrounded)} candidate(s) without an OUTWARD basis reference"]
    return []


def _i4_synthesis_qualify(a: dict) -> list[str]:
    syn = a.get("synthesis") or {}
    if not isinstance(syn, dict) or not syn:
        return []  # empty SYNTHESIS is legal on skipped rounds
    f = []
    if not str(syn.get("evidence_sufficiency", "")).strip():
        f.append("I4 SYNTHESIS-QUALIFY: no evidence-sufficiency statement")
    if not str(syn.get("kill_test", "")).strip():
        f.append("I4 SYNTHESIS-QUALIFY: no kill test (cheapest disproof)")
    return f


def _i5_ledger_validity(ledger_path: Path) -> list[str]:
    if not ledger_path.exists():
        # a not-yet-created ledger is the NORMAL early-round state — no
        # findings (a ledger passed by path that should exist is the
        # caller's contract, not the artifact's)
        return []
    f = []
    for i, line in enumerate(ledger_path.read_text().splitlines()):
        if not line.strip():
            continue
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            f.append(f"I5 LEDGER-VALIDITY: line {i + 1} is not JSON")
            continue
        missing = [k for k in REQUIRED_LEDGER_KEYS if k not in entry]
        if missing:
            f.append(f"I5 LEDGER-VALIDITY: line {i + 1} missing {missing}")
    return f


def _i6_no_phantom_paths(a: dict) -> list[str]:
    paths = a.get("artifacts") or []
    if not isinstance(paths, list):
        return []
    phantom = [p for p in paths if isinstance(p, str) and not Path(p).exists()]
    if phantom:
        return [f"I6 NO-PHANTOM-PATHS: {len(phantom)} referenced artifact(s) missing on disk: {phantom[:3]}"]
    return []


def main() -> int:
    import argparse

    ap = argparse.ArgumentParser(description="#411: run the harness invariants on a research artifact")
    ap.add_argument("artifact")
    ap.add_argument("--ledger")
    args = ap.parse_args()
    a = json.loads(Path(args.artifact).read_text())
    findings = check(a, Path(args.ledger) if args.ledger else None)
    for x in findings:
        print("FINDING:", x)
    print("CLEAN" if not findings else f"{len(findings)} finding(s)")
    return 0 if not findings else 1


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