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

#!/usr/bin/env python3
"""
pist_trace_classify_offline.py
==============================

SilverSight-ruleset-first offline proof-trace classifier.

This shim owns **only** I/O and subprocess orchestration. All classification
authority lives in Lean:

  - Spectral feature extraction   →  `pist-classify-trace` (Semantics.PIST.Spectral)
  - Shape threshold / color gate  →  `pist-classify-trace` (Semantics.PIST.Classify)
  - Tactic-family inference       →  `pist-classify-trace` (Semantics.PIST.Spectral)
  - RRC alignment verification    →  `rrc-watchdog`

The Python script never computes a spectral radius, never chooses a shape, and
never applies a heuristic. It reads the trace JSON, forwards it to the Lean
classifier, optionally invokes the alignment watchdog, and emits the combined
result JSON.

Usage:
    python3 pist_trace_classify_offline.py trace.json --rrc-shape signalShapedRouteCompiler
"""

import argparse
import json
import os
import subprocess
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]

PIST_CLASSIFY_HOST = (
    REPO_ROOT / "0-Core-Formalism" / "lean" / "Semantics" /
    ".lake" / "build" / "bin" / "pist-classify-trace"
)
PIST_CLASSIFY_CONTAINER = (
    "/home/researcher/stack/0-Core-Formalism/lean/Semantics/"
    ".lake/build/bin/pist-classify-trace"
)
WATCHDOG_CONTAINER = (
    "/home/researcher/stack/0-Core-Formalism/lean/Semantics/"
    ".lake/build/bin/rrc-watchdog"
)


def run_pist_classify(trace: dict) -> dict:
    """Forward the raw trace JSON to the Lean classifier via stdin.

    Uses the host binary when available; otherwise falls back to the
    research-stack container (with stdin attached).
    """
    payload = json.dumps(trace)
    if PIST_CLASSIFY_HOST.exists():
        cmd = [str(PIST_CLASSIFY_HOST)]
        res = subprocess.run(
            cmd, input=payload, capture_output=True, text=True, timeout=30
        )
    else:
        cmd = [
            "podman", "exec", "-i", "research-stack",
            PIST_CLASSIFY_CONTAINER,
        ]
        res = subprocess.run(
            cmd, input=payload, capture_output=True, text=True, timeout=30
        )

    if res.returncode != 0:
        return {
            "error": f"pist-classify-trace exited {res.returncode}",
            "stderr": res.stderr,
        }
    try:
        return json.loads(res.stdout)
    except json.JSONDecodeError as e:
        return {
            "error": f"invalid JSON from pist-classify-trace: {e}",
            "stdout": res.stdout,
            "stderr": res.stderr,
        }


def run_lean_watchdog(predicted_shape: str, expected_shape: str) -> dict:
    """Invoke the Lean RRC alignment watchdog."""
    cmd = [
        "podman", "exec", "research-stack",
        WATCHDOG_CONTAINER,
        "--pist-label", predicted_shape,
        "--exact-label", predicted_shape,
        "--rrc-shape", expected_shape,
    ]
    try:
        res = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        if res.returncode not in (0, 1):
            return {"error": f"watchdog returned exit code {res.returncode}", "stderr": res.stderr}
        return json.loads(res.stdout)
    except Exception as e:
        return {"error": f"failed to run rrc-watchdog: {e}"}


def main():
    parser = argparse.ArgumentParser(
        description="SilverSight-first offline RRC trace classifier"
    )
    parser.add_argument("trace_path", help="Path to ProofTraceReceipt v2 JSON file")
    parser.add_argument(
        "--rrc-shape",
        default="signalShapedRouteCompiler",
        help="Expected shape for Lean alignment gate check",
    )
    args = parser.parse_args()

    with open(args.trace_path) as f:
        trace = json.load(f)

    if not trace.get("transition_matrix"):
        print(json.dumps({"error": "Empty transition matrix"}))
        sys.exit(1)

    # 1. Lean classification (spectral + shape + tactic)
    classify_res = run_pist_classify(trace)
    if "error" in classify_res:
        print(json.dumps(classify_res))
        sys.exit(1)

    predicted_shape = classify_res.get("predicted_rrc_shape", "HoldForUnlawfulOrUnderspecifiedShape")

    # 2. Lean alignment verification
    watchdog_res = run_lean_watchdog(predicted_shape, args.rrc_shape)

    # 3. Combine into the legacy output shape for downstream consumers.
    #    knn_predictions is retired: KNN is decision logic and has no Lean
    #    authority yet. It is kept as a placeholder to avoid breaking parsers.
    output = {
        "theorem_name": classify_res.get("theorem_name", trace.get("name", "unnamed")),
        "spectral_radius": classify_res.get("spectral_radius"),
        "spectral_radius_q16": classify_res.get("spectral_radius_q16"),
        "predicted_rrc_shape": predicted_shape,
        "tactic_family_heuristic": classify_res.get("tactic_family", "unknown"),
        "knn_predictions": {
            "tactic_family": "unknown",
            "status": "unknown",
            "support": 0,
        },
        "lean_alignment": watchdog_res,
    }

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()