Research-Stack/4-Infrastructure/shim/candidate_certification_bridge.py

#!/usr/bin/env python3
"""
candidate_certification_bridge.py — Bridges entropy exploration candidates
into the Lean RRC certification pipeline.

Generates a Lean source file containing candidate BraidState fixtures that
can be compiled and certified by the existing crossStep →
eigensolid_convergence → receipt_invertible theorem chain.

Usage:
    # After running geometric_entropy_explorer.py --batch 50
    python3 candidate_certification_bridge.py --candidates-dir <dir> --output lean/Candidates.lean

Architecture (per AGENTS.md Programming Choice Flow):
    This script is pure I/O + code generation (Python-owned).
    No gating, alignment, scorin or promotion decisions.
    Lean owns all certification.
"""

import os
import json
import sys
import argparse


def to_lean_str(s: str) -> str:
    """Python string → Lean escaped string literal."""
    escaped = s.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def to_lean_q16(val: int) -> str:
    """Q16_16 integer → Lean Q16_16 literal."""
    return f"Q16_16.ofRawInt {val}"


def to_lean_bool(b: bool) -> str:
    return "true" if b else "false"


def generate_lean_candidates(candidates: list, output_path: str) -> str:
    """Generate a Lean file with candidate BraidState fixtures."""

    lines = [
        "import Semantics.BraidEigensolid",
        "import Semantics.BraidStrand",
        "import Semantics.BraidBracket",
        "import Semantics.FixedPoint",
        "",
        "open Semantics.BraidEigensolid",
        "open Semantics.BraidStrand",
        "open Semantics.BraidBracket",
        "open Semantics.FixedPoint",
        "",
        "namespace Semantics.RRC.EntropyCandidates",
        "",
        "/--",
        "  Auto-generated candidate BraidState fixtures from entropy exploration.",
        f"  Source: geometric_entropy_explorer.py ({len(candidates)} candidates)",
        "  These are exploration-phase candidates for Lean certification.",
        "  No promotion or alignment decisions are made here.",
        "-/",
        "",
    ]

    # Generate each candidate as a named def
    for i, cand in enumerate(candidates):
        label = cand.get("label", f"entropy_explorer_{i:03d}")
        eq_id = cand.get("equation_id", f"rrc_eq_{label}")
        braid = cand.get("braid_state", cand)  # if wrapped
        strands = braid.get("strands", [])

        if not strands:
            continue

        lines.append(f"/-- Candidate {label}: entropy={cand.get('genesis', {}).get('entropy_final', '?')} -/")

        # Generate strand entries as a Fin 8 → BraidStrand lambda
        strand_cases = []
        for si, s in enumerate(strands):
            phase = s.get("phaseAcc", {})
            bx = s.get("bracket", {})
            strand_cases.append(
                f"    | ⟨{si}, _⟩ => {{ phaseAcc := {{ x := {to_lean_q16(phase.get('x', 0))}, y := {to_lean_q16(phase.get('y', 0))} }}\n"
                f"      , parity := {to_lean_bool(s.get('parity', False))}\n"
                f"      , slot := {s.get('slot', 0)}\n"
                f"      , residue := {to_lean_q16(s.get('residue', 0))}\n"
                f"      , jitter := {to_lean_q16(s.get('jitter', 0))}\n"
                f"      , bracket := {{ lower := {to_lean_q16(bx.get('lower', 0))}\n"
                f"                    , upper := {to_lean_q16(bx.get('upper', 0))}\n"
                f"                    , gap := {to_lean_q16(bx.get('gap', 0))}\n"
                f"                    , kappa := {to_lean_q16(bx.get('kappa', 0))}\n"
                f"                    , phi := {to_lean_q16(bx.get('phi', 0))}\n"
                f"                    , admissible := {to_lean_bool(bx.get('admissible', False))} }} }}"
            )

        # Build the BraidState definition
        lines.append(f"def candidate_{label} : BraidState :=")
        lines.append(f"  {{ strands := λ")
        for sc in strand_cases:
            lines.append(sc)
        lines.append(f"  , step_count := 0")
        lines.append(f"  }}")
        lines.append("")

    # Build the candidate list for batch certification
    lines.append(f"/-- All {len(candidates)} candidates in a list for batch certification -/")
    lines.append(f"def allCandidates : List BraidStateSud :=")
    lines.append("  [")
    for i, cand in enumerate(candidates):
        label = cand.get("label", f"entropy_explorer_{i:03d}")
        lines.append(f"    candidate_{label},")
    lines.append("  ]")
    lines.append("")

    # Generate a verification function that runs crossStep on all candidates
    lines.append("/-- Verify all candidates: run crossStep and check eigensolid convergence -/")
    lines.append("def verifyAllCandidates : List (String × Bool) :=")
    lines.append("  allCandidates.map (fun s =>")
    lines.append("    let s' := crossStep s")
    lines.append("    let converged := IsEigensolid s'")
    lines.append("    (s'.step_count.repr, converged)")
    lines.append("  )")
    lines.append("")
    lines.append("end Semantics.RRC.EntropyCandidates")
    lines.append("")

    # Join
    content = "\n".join(lines)

    with open(output_path, "w") as f:
        f.write(content)
    print(f"Generated {output_path} — {len(candidates)} candidates, {len(lines)} lines")

    return content


def load_candidates_from_dir(dir_path: str) -> list:
    """Load all candidate JSON files from a directory."""
    candidates = []
    if not os.path.isdir(dir_path):
        print(f"Directory not found: {dir_path}", file=sys.stderr)
        return candidates

    for fname in sorted(os.listdir(dir_path)):
        if not fname.endswith(".json") or fname == "manifest.json":
            continue
        path = os.path.join(dir_path, fname)
        try:
            with open(path) as f:
                cand = json.load(f)
            label = os.path.splitext(fname)[0]
            # Strip prefixes that make bad Lean identifiers
            label = label.replace("candidate_", "").replace("-", "_")
            cand["label"] = label
            candidates.append(cand)
        except (json.JSONDecodeError, KeyError) as e:
            print(f"  Skipping {fname}: {e}")

    return candidates


def main():
    parser = argparse.ArgumentParser(
        description="Entropy candidate → Lean certification bridge"
    )
    parser.add_argument(
        "--candidates-dir", type=str, required=True,
        help="Directory with candidate JSON files"
    )
    parser.add_argument(
        "--output", type=str, required=True,
        help="Output Lean file path"
    )
    args = parser.parse_args()

    candidates = load_candidates_from_dir(args.candidates_dir)
    if not candidates:
        print("No candidates found.")
        sys.exit(1)

    print(f"Loaded {len(candidates)} candidates from {args.candidates_dir}")
    generate_lean_candidates(candidates, args.output)


if __name__ == "__main__":
    main()