feat(lean,infra): SilverSight-first offline PIST trace classifier

Moves all classification authority from Python into Lean:

- Adds `0-Core-Formalism/lean/Semantics/PistClassifyTrace.lean` executable.
  Reads a raw trace JSON and emits spectral radius, RRC shape, and tactic
  family using `Semantics.PIST.Spectral` and `Semantics.PIST.Classify`.
- Registers `pist-classify-trace` in `lakefile.toml`.
- Fixes `Semantics.PIST.Spectral` power iteration to handle directed
  transition matrices:
  - `symmetrize` now preserves half-integer weights as Q16_16 raw values.
  - `matVecMul` uses saturated Q16_16 arithmetic to prevent overflow.
- Rewrites `4-Infrastructure/shim/pist_trace_classify_offline.py` as a pure
  I/O wrapper: reads JSON, calls the Lean classifier, optionally calls
  `rrc-watchdog`, and emits the combined JSON. Removes Python-side spectral
  computation, shape thresholds, tactic heuristic, and KNN.
- Updates `AGENTS.md` and `4-Infrastructure/AGENTS.md` with the new build
  baseline and shim contract.

Verification:
- `lake build` → 8604 jobs, 0 errors
- Canary trace outputs match previous Python outputs to within Q16_16 rounding
  (e.g., apply_chain λ_q16 = 59044 vs 59045).
- `python3 -m py_compile` on the rewritten shim passes.
This commit is contained in:
allaun 2026-06-21 16:33:31 -05:00
parent 19f40d03c4
commit 67e77d8f24
6 changed files with 192 additions and 177 deletions

View file

@ -0,0 +1,89 @@
-- PistClassifyTrace.lean
-- SilverSight-ruleset-first proof-trace classifier executable.
--
-- Authority: this executable is the sole classifier. It reads a raw trace
-- JSON, computes spectral features with Q16_16 arithmetic, and emits the
-- predicted RRC shape and tactic family. Python shims may only format I/O
-- and call this executable.
import Semantics.PIST.Spectral
import Semantics.PIST.Classify
import Lean.Data.Json
open Lean
open Semantics.PIST
namespace PistClassifyTrace
/-- Raw input expected from the Python shim. -/
structure Input where
name : String
transition_matrix : Array (Array Int)
deriving FromJson
/-- Shape string used when no higher-energy regime is detected.
Matches RRC.Emit.shapeStr for HoldForUnlawfulOrUnderspecifiedShape. -/
def holdShape : String := "HoldForUnlawfulOrUnderspecifiedShape"
/-- Map the library tactic family to the Python-facing string. -/
def tacticFamilyString : Spectral.TacticFamily → String
| .rewrite => "rewrite"
| .normalization => "normalization"
| .arithmetic => "arithmetic"
| .induction => "induction"
| .algebraic => "algebraic"
| .case_analysis => "case_analysis"
| .discharge => "discharge"
| .reflexivity => "reflexivity"
| .unknown => "unknown"
/-- Build a JSON object from a Q16_16 spectral profile. -/
def spectralProfileJson (p : Spectral.SpectralProfile) : Json :=
Json.mkObj [
("matrix_size", toJson p.matrix_size.toInt),
("rank", toJson p.rank.toInt),
("spectral_gap", toJson p.spectral_gap.toInt),
("density", toJson p.density.toInt),
("trace_val", toJson p.trace_val.toInt),
("frobenius_norm", toJson p.frobenius_norm.toInt),
("laplacian_zero_count", toJson p.laplacian_zero_count.toInt),
("adjacency_eigenvalue_max", toJson p.adjacency_eigenvalue_max.toInt),
("laplacian_eigenvalue_max", toJson p.laplacian_eigenvalue_max.toInt),
("singular_value_max", toJson p.singular_value_max.toInt)
]
/-- Build the classifier output JSON. -/
def outputJson (name : String) (profile : Spectral.SpectralProfile)
(shape : String) (tactic : String) : Json :=
Json.mkObj [
("theorem_name", toJson name),
("spectral_radius", toJson profile.adjacency_eigenvalue_max.toInt),
("spectral_radius_q16", toJson profile.adjacency_eigenvalue_max.toInt),
("predicted_rrc_shape", toJson shape),
("tactic_family", toJson tactic),
("spectral_profile", spectralProfileJson profile)
]
end PistClassifyTrace
def main (args : List String) : IO Unit := do
let inputStr ←
match args with
| [] =>
let stdin ← IO.getStdin
stdin.readToEnd
| [path] => IO.FS.readFile path
| _ => throw (IO.userError "Usage: pist-classify-trace [trace.json]")
match Json.parse inputStr with
| .error e =>
IO.println (Json.compress (Json.mkObj [("error", toJson s!"JSON parse error: {e}")]))
| .ok j =>
match (fromJson? j : Except String PistClassifyTrace.Input) with
| .error e =>
IO.println (Json.compress (Json.mkObj [("error", toJson e)]))
| .ok input =>
let profile := Spectral.computeSpectral input.transition_matrix
let shape := (Classify.classifyExact input.transition_matrix).getD PistClassifyTrace.holdShape
let tactic := PistClassifyTrace.tacticFamilyString (Spectral.classifyTacticFromName input.name)
IO.println (Json.compress (PistClassifyTrace.outputJson input.name profile shape tactic))

View file

@ -90,11 +90,13 @@ private def getEntry (mat : Array (Array Int)) (i j : Nat) : Int :=
private def rowSum (mat : Array (Array Int)) (i n : Nat) : Int := private def rowSum (mat : Array (Array Int)) (i n : Nat) : Int :=
(List.range n).foldl (fun acc j => acc + getEntry mat i j) 0 (List.range n).foldl (fun acc j => acc + getEntry mat i j) 0
/-- Symmetrize a matrix: sym[i][j] = (mat[i][j] + mat[j][i]) / 2. -/ /-- Symmetrize a matrix: sym[i][j] = (mat[i][j] + mat[j][i]) / 2 as a
Q16_16 raw integer. This preserves half-integer weights for directed
transition matrices instead of truncating them to integers. -/
private def symmetrize (mat : Array (Array Int)) (n : Nat) : Array (Array Int) := private def symmetrize (mat : Array (Array Int)) (n : Nat) : Array (Array Int) :=
Array.ofFn (n := n) fun i => Array.ofFn (n := n) fun i =>
Array.ofFn (n := n) fun j => Array.ofFn (n := n) fun j =>
(getEntry mat i.val j.val + getEntry mat j.val i.val) / 2 (getEntry mat i.val j.val + getEntry mat j.val i.val) * q16Scale / 2
/-- Laplacian of a symmetrized matrix: L[i][j] = deg(i) if i=j else sym[i][j]. -/ /-- Laplacian of a symmetrized matrix: L[i][j] = deg(i) if i=j else sym[i][j]. -/
private def buildLaplacian (sym : Array (Array Int)) (n : Nat) : Array (Array Int) := private def buildLaplacian (sym : Array (Array Int)) (n : Nat) : Array (Array Int) :=
@ -120,12 +122,15 @@ private def normSqRaw (v : Array Q16_16) : Int :=
v.foldl (fun acc x => acc + x.toInt * x.toInt) 0 v.foldl (fun acc x => acc + x.toInt * x.toInt) 0
/-- Matrix-vector multiply: (mat × v)[i] = Σ_j mat[i][j] * v[j]. /-- Matrix-vector multiply: (mat × v)[i] = Σ_j mat[i][j] * v[j].
mat entries are raw Int; v components are Q16_16. Result in Q16_16. -/ mat entries are raw Int (treated as Q16_16 raw values); v components are
Q16_16. Uses saturated Q16_16 arithmetic to prevent overflow during
power iteration on directed transition matrices. -/
private def matVecMul (mat : Array (Array Int)) (n : Nat) (v : Array Q16_16) : Array Q16_16 := private def matVecMul (mat : Array (Array Int)) (n : Nat) (v : Array Q16_16) : Array Q16_16 :=
Array.ofFn (n := n) fun i => Array.ofFn (n := n) fun i =>
let s : Int := (List.range n).foldl (fun acc j => (List.range n).foldl (fun acc j =>
acc + getEntry mat i.val j * (v.getD j zero).toInt) 0 let a : Q16_16 := ofRawInt (getEntry mat i.val j)
ofRawInt s let b : Q16_16 := v.getD j zero
add acc (mul a b)) zero
/-- Power iteration: dominant eigenvalue of mat as Q16_16 raw integer. /-- Power iteration: dominant eigenvalue of mat as Q16_16 raw integer.
max_iter=100 in Python; we use Nat fuel. -/ max_iter=100 in Python; we use Nat fuel. -/

View file

@ -104,3 +104,7 @@ root = "SabotagePreventionCli"
[[lean_exe]] [[lean_exe]]
name = "rrc-watchdog" name = "rrc-watchdog"
root = "RrcWatchdog" root = "RrcWatchdog"
[[lean_exe]]
name = "pist-classify-trace"
root = "PistClassifyTrace"

View file

@ -360,6 +360,7 @@ python3 4-Infrastructure/storage/storage_agent.py --loop --interval 900
- `4-Infrastructure/shim/rrc_domain_manifold_graph.py` — Build expanding manifold graph across all gathered math domains. Produces 6 edge types (shared_paper, manifold route, regime, shared_topic, domain adapter, dimension chain) plus 4 intrinsic fallback types (kernel_internal, kernel_hub, topic_overlap, unverified shared_paper) that ensure dense output (~0.61 density) even when live arxiv DB access is unavailable. - `4-Infrastructure/shim/rrc_domain_manifold_graph.py` — Build expanding manifold graph across all gathered math domains. Produces 6 edge types (shared_paper, manifold route, regime, shared_topic, domain adapter, dimension chain) plus 4 intrinsic fallback types (kernel_internal, kernel_hub, topic_overlap, unverified shared_paper) that ensure dense output (~0.61 density) even when live arxiv DB access is unavailable.
- `4-Infrastructure/shim/rrc_refactor_oracle.py` — Chaos-game-driven RRC self-refactoring oracle. Reads the domain manifold graph, runs `BurgersChaosGame` quantum walk eigenvector centrality, generates PRUNE/MERGE/PROMOTE/SPLIT commands from centrality deltas. When graph density < 0.005, attempts live rebuild via `rrc_domain_manifold_graph.py`, then falls back to intrinsic edge inference from node metadata. `--no-live` flag forces fallback for containerized deployments. Receipt: `rrc_refactor_oracle_receipt.json`. Canonical configuration from SLO sweep: `--max-merges 10 --threshold-prune 0.005 --threshold-merge 0.01 --max-iterations 5`. - `4-Infrastructure/shim/rrc_refactor_oracle.py` — Chaos-game-driven RRC self-refactoring oracle. Reads the domain manifold graph, runs `BurgersChaosGame` quantum walk eigenvector centrality, generates PRUNE/MERGE/PROMOTE/SPLIT commands from centrality deltas. When graph density < 0.005, attempts live rebuild via `rrc_domain_manifold_graph.py`, then falls back to intrinsic edge inference from node metadata. `--no-live` flag forces fallback for containerized deployments. Receipt: `rrc_refactor_oracle_receipt.json`. Canonical configuration from SLO sweep: `--max-merges 10 --threshold-prune 0.005 --threshold-merge 0.01 --max-iterations 5`.
- `4-Infrastructure/shim/rrc_slo_analyzer.py` — SLO analyzer for the refactoring oracle. Measures structural SLOs (spectral_gap, modularity, conductance, isolation_ratio, centrality_spread, edge_efficiency, community_count) and performance SLOs (adj_build_ms, eigenvector_ms, evolution_ms, total_ms). Compares baseline vs target graph. Receipt: `rrc_slo_receipt.json`. - `4-Infrastructure/shim/rrc_slo_analyzer.py` — SLO analyzer for the refactoring oracle. Measures structural SLOs (spectral_gap, modularity, conductance, isolation_ratio, centrality_spread, edge_efficiency, community_count) and performance SLOs (adj_build_ms, eigenvector_ms, evolution_ms, total_ms). Compares baseline vs target graph. Receipt: `rrc_slo_receipt.json`.
- `4-Infrastructure/shim/pist_trace_classify_offline.py` — Pure-I/O shim for offline proof-trace classification. Reads trace JSON, forwards it to the `pist-classify-trace` Lean executable for spectral feature extraction and shape/tactic decisions, and optionally invokes `rrc-watchdog` for alignment. Contains no classification logic, no spectral arithmetic, and no floating-point compute path.
- `4-Infrastructure/shim/rrc_slo_sweep.py` — Parameter sweep over merge aggressiveness (max_merges ∈ {20,10,5,2,1}). Composite score: `speedup × community_retention × node_retention × (1 + mod_gain) × (1 - cent_spread_loss)`. Found optimal at max_merges=10: 3.1× speedup, 71% community retention, 100% isolation elimination. Receipt: `rrc_slo_sweep_receipt.json`. - `4-Infrastructure/shim/rrc_slo_sweep.py` — Parameter sweep over merge aggressiveness (max_merges ∈ {20,10,5,2,1}). Composite score: `speedup × community_retention × node_retention × (1 + mod_gain) × (1 - cent_spread_loss)`. Found optimal at max_merges=10: 3.1× speedup, 71% community retention, 100% isolation elimination. Receipt: `rrc_slo_sweep_receipt.json`.
- `4-Infrastructure/shim/rrc_bosonic_tensor_gpu.py` — GPU-accelerated Bosonic Tensor Network Centrality: computes exp(-i*A*theta) using adaptive RK4 on GPU with `wgpu` (Vulkan) to scale N to 10000+ without CPU eigendecomposition bottleneck. Receipt: `rrc_bosonic_tensor_gpu_receipt.json`. - `4-Infrastructure/shim/rrc_bosonic_tensor_gpu.py` — GPU-accelerated Bosonic Tensor Network Centrality: computes exp(-i*A*theta) using adaptive RK4 on GPU with `wgpu` (Vulkan) to scale N to 10000+ without CPU eigendecomposition bottleneck. Receipt: `rrc_bosonic_tensor_gpu_receipt.json`.
- `4-Infrastructure/shim/rrc_bosonic_db_buffer.py` — Asynchronous database ingestion buffer: queues, batches, and flushes PostgreSQL inserts for bosonic tensor network receipts and metrics in thread-safe worker pools. - `4-Infrastructure/shim/rrc_bosonic_db_buffer.py` — Asynchronous database ingestion buffer: queues, batches, and flushes PostgreSQL inserts for bosonic tensor network receipts and metrics in thread-safe worker pools.

View file

@ -3,11 +3,20 @@
pist_trace_classify_offline.py pist_trace_classify_offline.py
============================== ==============================
Offline, token-free proof trace classifier. SilverSight-ruleset-first offline proof-trace classifier.
Computes transition matrix spectra, finds nearest neighbors from the 57-theorem
flexure library (offline vectors), determines the RRC shape proxy using This shim owns **only** I/O and subprocess orchestration. All classification
color-space thresholds matching Lean's `Semantics.PIST.Classify`, and invokes authority lives in Lean:
the local Lean `rrc-watchdog` binary in the podman container to verify alignment.
- 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: Usage:
python3 pist_trace_classify_offline.py trace.json --rrc-shape signalShapedRouteCompiler python3 pist_trace_classify_offline.py trace.json --rrc-shape signalShapedRouteCompiler
@ -15,151 +24,71 @@ Usage:
import argparse import argparse
import json import json
import math
import os import os
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
from collections import Counter
REPO_ROOT = Path(__file__).resolve().parents[2] REPO_ROOT = Path(__file__).resolve().parents[2]
VECTORS_PATH = REPO_ROOT / "shared-data" / "pist_trace_scaled_vectors.jsonl"
WATCHDOG_PATH = "/home/researcher/stack/0-Core-Formalism/lean/Semantics/.lake/build/bin/rrc-watchdog"
FEATURE_KEYS = [ PIST_CLASSIFY_HOST = (
"matrix_size", REPO_ROOT / "0-Core-Formalism" / "lean" / "Semantics" /
"rank", ".lake" / "build" / "bin" / "pist-classify-trace"
"spectral_gap", )
"laplacian_zero_count", PIST_CLASSIFY_CONTAINER = (
"density", "/home/researcher/stack/0-Core-Formalism/lean/Semantics/"
"adjacency_eigenvalue_max", ".lake/build/bin/pist-classify-trace"
] )
WATCHDOG_CONTAINER = (
"/home/researcher/stack/0-Core-Formalism/lean/Semantics/"
".lake/build/bin/rrc-watchdog"
)
def power_iteration(matrix, max_iter=100): def run_pist_classify(trace: dict) -> dict:
n = len(matrix) """Forward the raw trace JSON to the Lean classifier via stdin.
if n == 0:
return 0.0
v = [1.0 / math.sqrt(n)] * n
for _ in range(max_iter):
vn = [sum(matrix[i][j] * v[j] for j in range(n)) for i in range(n)]
nm = math.sqrt(sum(x * x for x in vn))
if nm < 1e-12:
return 0.0
v = [x / nm for x in vn]
num = sum(v[i] * sum(matrix[i][j] * v[j] for j in range(n)) for i in range(n))
den = sum(v[i] * v[i] for i in range(n))
return num / den if den > 0 else 0.0
Uses the host binary when available; otherwise falls back to the
def compute_spectral(matrix): research-stack container (with stdin attached).
n = len(matrix) """
if n == 0: payload = json.dumps(trace)
return {} if PIST_CLASSIFY_HOST.exists():
sym = [[(matrix[i][j] + matrix[j][i]) / 2.0 for j in range(n)] for i in range(n)] cmd = [str(PIST_CLASSIFY_HOST)]
lap = [[sum(sym[i]) if i == j else -sym[i][j] for j in range(n)] for i in range(n)] res = subprocess.run(
ev_max = power_iteration(sym) cmd, input=payload, capture_output=True, text=True, timeout=30
shifted = [[sym[i][j] - 0.9 * ev_max * (1 if i == j else 0) for j in range(n)] for i in range(n)] )
ev_shift = power_iteration(shifted)
ev_second = max(0, ev_max - ev_shift) if ev_shift < ev_max else ev_max
gap = ev_max - ev_second
lap_max = power_iteration(lap)
neg_lap = [[-lap[i][j] for j in range(n)] for i in range(n)]
lap_min = -power_iteration(neg_lap)
ata = [[sum(matrix[k][i] * matrix[k][j] for k in range(n)) for j in range(n)] for i in range(n)]
sv_max = math.sqrt(max(0, power_iteration(ata)))
rank = sum(1 for row in matrix if sum(row) > 0)
total = sum(sum(row) for row in matrix)
frob = math.sqrt(sum(cell * cell for row in matrix for cell in row))
lap_zero = sum(1 for i in range(n) if abs(sum(matrix[i]) - matrix[i][i]) < 1e-9)
return {
"matrix_size": n,
"rank": rank,
"spectral_gap": round(gap, 6),
"density": round(total / max(n * n, 1), 6),
"trace": sum(matrix[i][i] for i in range(n)),
"frobenius_norm": round(frob, 6),
"laplacian_zero_count": lap_zero,
"adjacency_eigenvalue_max": round(ev_max, 6),
"laplacian_eigenvalue_max": round(lap_max, 6),
"singular_value_max": round(sv_max, 6),
}
def classify_tactic_family(name: str) -> str:
n = name.lower()
if "rw" in n:
return "rewrite"
if "simp" in n:
return "normalization"
if "omega" in n:
return "arithmetic"
if "induct" in n:
return "induction"
if "ring" in n or "calc" in n:
return "algebraic"
if "cases" in n or "constructor" in n:
return "case_analysis"
if any(k in n for k in ["apply", "intro", "have", "logic"]):
return "discharge"
if "rfl" in n:
return "reflexivity"
return "unknown"
def get_rrc_shape(ev_max: float) -> str:
# Q16.16 conversion
lam = int(ev_max * 65536)
if lam >= 262144: # 4.0
return "cognitiveLoadField"
elif lam >= 131072: # 2.0
return "signalShapedRouteCompiler"
else: else:
return "holdForUnlawfulOrUnderspecifiedShape" 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:
def load_library(): return {
library = [] "error": f"pist-classify-trace exited {res.returncode}",
if not VECTORS_PATH.exists(): "stderr": res.stderr,
print(f"Warning: local vectors file not found at {VECTORS_PATH}", file=sys.stderr) }
return library try:
with open(VECTORS_PATH) as f: return json.loads(res.stdout)
for line in f: except json.JSONDecodeError as e:
if line.strip(): return {
r = json.loads(line) "error": f"invalid JSON from pist-classify-trace: {e}",
library.append(r) "stdout": res.stdout,
return library "stderr": res.stderr,
}
def find_nearest_neighbors(features, library, top_k=3):
if not library:
return []
scored = []
for r in library:
# Distance over the FEATURE_KEYS
dist = 0.0
for k in FEATURE_KEYS:
val_a = features.get(k, 0.0)
val_b = r.get(k, 0.0)
dist += (val_a - val_b) ** 2
dist = math.sqrt(dist)
scored.append({"dist": dist, "record": r})
scored.sort(key=lambda x: x["dist"])
return scored[:top_k]
def run_lean_watchdog(predicted_shape: str, expected_shape: str) -> dict: def run_lean_watchdog(predicted_shape: str, expected_shape: str) -> dict:
"""Invoke the Lean RRC alignment watchdog."""
cmd = [ cmd = [
"podman", "podman", "exec", "research-stack",
"exec", WATCHDOG_CONTAINER,
"research-stack", "--pist-label", predicted_shape,
WATCHDOG_PATH, "--exact-label", predicted_shape,
"--pist-label", "--rrc-shape", expected_shape,
predicted_shape,
"--exact-label",
predicted_shape,
"--rrc-shape",
expected_shape,
] ]
try: try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=15) res = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
@ -171,7 +100,9 @@ def run_lean_watchdog(predicted_shape: str, expected_shape: str) -> dict:
def main(): def main():
parser = argparse.ArgumentParser(description="Offline RRC Trace Classifier") 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("trace_path", help="Path to ProofTraceReceipt v2 JSON file")
parser.add_argument( parser.add_argument(
"--rrc-shape", "--rrc-shape",
@ -180,52 +111,37 @@ def main():
) )
args = parser.parse_args() args = parser.parse_args()
# Load trace
with open(args.trace_path) as f: with open(args.trace_path) as f:
trace = json.load(f) trace = json.load(f)
name = trace.get("name", "unnamed") if not trace.get("transition_matrix"):
matrix = trace.get("transition_matrix", [])
if not matrix:
print(json.dumps({"error": "Empty transition matrix"})) print(json.dumps({"error": "Empty transition matrix"}))
sys.exit(1) sys.exit(1)
# 1. Compute spectral profile # 1. Lean classification (spectral + shape + tactic)
spectral = compute_spectral(matrix) classify_res = run_pist_classify(trace)
ev_max = spectral["adjacency_eigenvalue_max"] if "error" in classify_res:
print(json.dumps(classify_res))
sys.exit(1)
# 2. Map color domain (Lean-anchored logic) predicted_shape = classify_res.get("predicted_rrc_shape", "HoldForUnlawfulOrUnderspecifiedShape")
rrc_shape = get_rrc_shape(ev_max)
# 3. K-NN tactic family matching (free offline) # 2. Lean alignment verification
library = load_library() watchdog_res = run_lean_watchdog(predicted_shape, args.rrc_shape)
neighbors = find_nearest_neighbors(spectral, library)
tactic_family = classify_tactic_family(name)
predicted_tactic_family = "unknown"
predicted_status = "failed"
knn_support = 0
if neighbors:
families = [classify_tactic_family(n["record"].get("name", "")) for n in neighbors]
statuses = [n["record"].get("status", "failed") for n in neighbors]
predicted_tactic_family = Counter(families).most_common(1)[0][0]
predicted_status = Counter(statuses).most_common(1)[0][0]
knn_support = len(neighbors)
# 4. Lean RRC alignment verification
watchdog_res = run_lean_watchdog(rrc_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 = { output = {
"theorem_name": name, "theorem_name": classify_res.get("theorem_name", trace.get("name", "unnamed")),
"spectral_radius": ev_max, "spectral_radius": classify_res.get("spectral_radius"),
"spectral_radius_q16": int(ev_max * 65536), "spectral_radius_q16": classify_res.get("spectral_radius_q16"),
"predicted_rrc_shape": rrc_shape, "predicted_rrc_shape": predicted_shape,
"tactic_family_heuristic": tactic_family, "tactic_family_heuristic": classify_res.get("tactic_family", "unknown"),
"knn_predictions": { "knn_predictions": {
"tactic_family": predicted_tactic_family, "tactic_family": "unknown",
"status": predicted_status, "status": "unknown",
"support": knn_support, "support": 0,
}, },
"lean_alignment": watchdog_res, "lean_alignment": watchdog_res,
} }

View file

@ -153,7 +153,7 @@ See respective repositories for components. Shared utilities have been duplicate
## Core Surfaces ## Core Surfaces
- Lean/Semantics: `0-Core-Formalism/lean/Semantics/` — Compiler surface includes `Semantics.SieveLemmas` and `Semantics.InteractionGraphSidon` (commit `e61bb627`, 3314 jobs, 0 errors). New exploration module `Semantics.CompleteInteractionGraph` (complete directed graph / every-point-touches-every-point) builds standalone with one bounded `walkMatrix_off_diag` sorry. - Lean/Semantics: `0-Core-Formalism/lean/Semantics/` — Compiler surface includes `Semantics.SieveLemmas`, `Semantics.InteractionGraphSidon`, `Semantics.PIST.Spectral`, and `Semantics.PIST.Classify` (8604 jobs, 0 errors). The `pist-classify-trace` executable is the sole authority for offline proof-trace shape/tactic classification; Python shims are pure I/O.
- Infrastructure shims and probes: `4-Infrastructure/shim/` - Infrastructure shims and probes: `4-Infrastructure/shim/`
- Hardware bring-up: `4-Infrastructure/hardware/` - Hardware bring-up: `4-Infrastructure/hardware/`
- Documentation and wiki surfaces: `6-Documentation/` - Documentation and wiki surfaces: `6-Documentation/`