From b3d4ef4206ccb37adfdaf54acee821b7f82b2402 Mon Sep 17 00:00:00 2001 From: Brandon Schneider Date: Tue, 26 May 2026 02:28:16 -0500 Subject: [PATCH] =?UTF-8?q?feat(pist):=20Tier=202=20trace=20bridge=20?= =?UTF-8?q?=E2=80=94=20tactic-level=20goal=20transitions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - lean_trace_bridge.py: captures Goal_i → tactic → Goal_{i+1} transitions - Builds ProofTraceReceipt v1 with step deltas, transition matrix, flexure joints - Handles single-line by-blocks, semicolon-separated, and indented multi-line - pist_trace_decompose.py: spectral analysis of transition matrix - Power iteration for eigenvalue estimation - Spectral gap, rank, density, Laplacian zero count - Tactic family distribution, delta statistics - Full pipeline: Lean theorem → trace → transition graph → spectral features --- 4-Infrastructure/shim/lean_trace_bridge.py | 489 ++++++++++++++++++ 4-Infrastructure/shim/pist_trace_decompose.py | 276 ++++++++++ 2 files changed, 765 insertions(+) create mode 100644 4-Infrastructure/shim/lean_trace_bridge.py create mode 100644 4-Infrastructure/shim/pist_trace_decompose.py diff --git a/4-Infrastructure/shim/lean_trace_bridge.py b/4-Infrastructure/shim/lean_trace_bridge.py new file mode 100644 index 00000000..965de974 --- /dev/null +++ b/4-Infrastructure/shim/lean_trace_bridge.py @@ -0,0 +1,489 @@ +#!/usr/bin/env python3 +"""Lean Trace Bridge — captures tactic-level goal state transitions. + +Tier 2 MVP: splits a Lean proof into tactic steps, replays each prefix +through the proof worker, and builds a structured trace with goal-state +snapshots captured from Lean's error output. + +Usage: + python3 lean_trace_bridge.py [--out trace.json] + python3 lean_trace_bridge.py --code 'theorem t : 1+1=2 := by omega' --name t [--out trace.json] +""" + +import hashlib +import json +import os +import re +import subprocess +import sys +import time +import uuid +from pathlib import Path + +PROOF_SERVER_TOKEN = os.environ.get("PROOF_SERVER_TOKEN", "") +if not PROOF_SERVER_TOKEN: + tf = os.environ.get("PROOF_SERVER_TOKEN_FILE", + os.path.expanduser("~/.config/ene/language-proof-server.token")) + try: + PROOF_SERVER_TOKEN = Path(tf).read_text().strip() + except (FileNotFoundError, OSError): + PROOF_SERVER_TOKEN = "" + +WORKER_URL = os.environ.get("CANARY_WORKER_URL", "http://100.110.163.82:8787") + + +def split_tactic_lines(code: str) -> list[str]: + """Split a Lean proof into individual tactic steps. + + Handles single-line proofs (by simp), semicolon-separated (by simp; omega), + and indented multi-line blocks (induction with case branches). + """ + lines = code.strip().split("\n") + + # Find the `by` block + by_content_lines = [] + found_by = False + + for line in lines: + stripped = line.strip() + + # Detect `:= by` on the same line + if not found_by and ":= by " in stripped: + parts = stripped.split(":= by ", 1) + by_content_lines.append(parts[1].strip()) + found_by = True + continue + if not found_by and ":= by" in stripped: + found_by = True + continue + + if not found_by: + # Check for standalone `by` on this line + if stripped == "by": + found_by = True + continue + continue # still in header + + # We're in the by-block + if stripped and not stripped.startswith("--") and not stripped.startswith("/-"): + by_content_lines.append(stripped) + + by_content = "\n".join(by_content_lines).strip() + if not by_content: + return [code] + + # Merge continuation lines into their parent tactic: + # - Lines starting with | are induction case branches + # - Lines at deeper indent continue the previous tactic + # - Then split on semicolons for finer granularity + merged = [] + current = "" + for line in by_content_lines: + stripped = line.strip() + if not stripped: + continue + if current and not stripped.startswith("|"): + # Check if this is a continuation (indented) or new tactic + if line[0].isspace(): + current += " " + stripped + else: + merged.append(current.strip()) + current = stripped + else: + current += (" " if current else "") + stripped + if current.strip(): + merged.append(current.strip()) + + # Split on semicolons for truly independent steps + tactics = [] + for m in merged: + parts = re.split(r';', m) + tactics.extend(p.strip() for p in parts if p.strip()) + + return tactics if tactics else [code] + + # Fallback: split on semicolons if we got nothing + if not tactics: + for part in re.split(r'[;\n]', by_content): + t = part.strip() + if t: + tactics.append(t) + + return tactics if tactics else [code] + + +def sha256(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def prove_step(code: str, timeout_s: int = 120) -> dict: + """Send Lean code to the proof worker.""" + result = subprocess.run( + ["curl", "-s", "--connect-timeout", "10", "-X", "POST", + f"{WORKER_URL}/lean/check", + "-H", "Content-Type: application/json", + "-H", f"Authorization: Bearer {PROOF_SERVER_TOKEN}", + "-d", json.dumps({"code": code, "name": "trace_step"})], + capture_output=True, text=True, timeout=timeout_s, + ) + if result.returncode != 0: + return {"ok": False, "error": f"curl: {result.stderr[:200]}", "stdout": "", "stderr": ""} + try: + resp = json.loads(result.stdout) + receipt = resp.get("receipt", resp) + return { + "ok": resp.get("ok", False), + "stdout": receipt.get("stdout", ""), + "stderr": receipt.get("stderr", ""), + "returncode": receipt.get("returncode", -1), + "elapsed_ms": receipt.get("elapsed_ms", 0), + "error": receipt.get("error", ""), + } + except json.JSONDecodeError as e: + return {"ok": False, "error": f"json: {e}", "stdout": result.stdout[:500], "stderr": ""} + + +def extract_goal_text(output: str, is_error: bool = False) -> str: + """Extract the goal state from Lean's output. + + Lean prints the goal state when a tactic block is incomplete. + The goal appears after 'unsolved goals' or as the error context. + """ + # Try to find the goal statement + patterns = [ + r"(?<=unsolved goals\n).*?(?=\n\n)", + r"(?<=⊢\n).*?(?=\n|$)", + r"(?<=⊢ ).*$", + r"(?<=expected\n).*?(?=\n|$)", + ] + for p in patterns: + m = re.search(p, output, re.DOTALL) + if m and m.group().strip(): + return m.group().strip()[:500] + + # Fallback: last significant line + lines = [l.strip() for l in output.split("\n") if l.strip() and "error" not in l.lower()] + return lines[-1][:500] if lines else output[:500] + + +def extract_hypotheses(output: str) -> list[str]: + """Extract hypothesis names from the goal context.""" + hyps = [] + for m in re.finditer(r"(?:^|\n)(\w+)\s*:", output): + hyps.append(m.group(1)) + return hyps[:20] + + +def count_symbols(text: str) -> dict: + """Count operator symbols in a goal text.""" + ops = { + "+": 0, "*": 0, "-": 0, "/": 0, "^": 0, + "≤": 0, "≥": 0, "<": 0, ">": 0, "=": 0, + "∧": 0, "∨": 0, "→": 0, "¬": 0, "∀": 0, "∃": 0, + "∑": 0, "∫": 0, "∪": 0, "∩": 0, "⊆": 0, + } + for char in text: + if char in ops: + ops[char] += 1 + return {k: v for k, v in ops.items() if v > 0} + + +def build_trace(code: str, name: str = "unnamed") -> dict: + """Build a ProofTraceReceipt by replaying tactic steps.""" + tactics = split_tactic_lines(code) + + if not tactics or tactics == [code]: + result = prove_step(code) + ok = result.get("ok", False) + return { + "trace_version": "proof-trace-v1", + "receipt_hash": sha256(code), + "theorem_name": name, + "status": "verified" if ok else "failed", + "tactic_count": 1, + "total_elapsed_ms": result.get("elapsed_ms", 0), + "steps": [{"step": 0, "tactic": code, "result": "success" if ok else "failure"}], + "goal_transition_matrix": [[1] if ok else [0]], + "flexure_joints": [], + "warning": "full_code_fallback" + } + + # Reconstruct the theorem header (everything before `:= by` or `by`) + by_pos = code.rfind(":= by") + if by_pos < 0: + by_pos = code.rfind("\nby") + if by_pos < 0: + by_pos = code.find("by ") + header = code[:by_pos] + ":= by" if by_pos > 0 else code.split("by")[0] + + steps = [] + prev_stdout = "" + prev_stderr = "" + + for i, tactic in enumerate(tactics): + # Build incremental proof + prefix = header + "\n" + for j in range(i + 1): + prefix += " " + tactics[j] + "\n" + + t0 = time.time() + result = prove_step(prefix) + dt = time.time() - t0 + + stdout = result.get("stdout", "") + stderr = result.get("stderr", "") + ok = result.get("ok", False) + + goal_before = extract_goal_text(prev_stdout + "\n" + prev_stderr) + goal_after = extract_goal_text(stdout + "\n" + stderr) + hyps_before = extract_hypotheses(prev_stdout + "\n" + prev_stderr) + hyps_after = extract_hypotheses(stdout + "\n" + stderr) + + delta = { + "symbol_delta": len(count_symbols(goal_after)) - len(count_symbols(goal_before)), + "hypothesis_delta": len(hyps_after) - len(hyps_before), + "goal_count_delta": (len(hyps_after) + 1 if goal_after else 0) - (len(hyps_before) + 1 if goal_before else 0), + } + + steps.append({ + "step": i, + "tactic": tactic, + "before_goal_hash": sha256(goal_before) if goal_before else "", + "after_goal_hash": sha256(goal_after) if goal_after else "", + "before_goal_text": goal_before[:300], + "after_goal_text": goal_after[:300], + "goal_count_before": len(hyps_before) + 1 if goal_before else 0, + "goal_count_after": len(hyps_after) + 1 if goal_after else 0, + "hypothesis_count_before": len(hyps_before), + "hypothesis_count_after": len(hyps_after), + "operator_count_before": len(count_symbols(goal_before)), + "operator_count_after": len(count_symbols(goal_after)), + "delta": delta, + "elapsed_ms": result.get("elapsed_ms", int(dt * 1000)), + "result": "success" if ok else "failure", + "stdout_preview": stdout[:200], + "stderr_preview": stderr[:200], + }) + + prev_stdout = stdout + prev_stderr = stderr + + # Final verification + final_result = prove_step(code) + + # Build transition matrix + hashes = [] + for s in steps: + if s.get("before_goal_hash"): + hashes.append(s["before_goal_hash"]) + if steps: + hashes.append(steps[-1].get("after_goal_hash", "")) + unique = list(dict.fromkeys(hashes)) + h2i = {h: i for i, h in enumerate(unique)} + n = len(unique) + matrix = [[0] * n for _ in range(n)] + for s in steps: + bh = s.get("before_goal_hash", "") + ah = s.get("after_goal_hash", "") + if bh in h2i and ah in h2i: + matrix[h2i[bh]][h2i[ah]] += 1 + + # Extract flexure joints + joints = [] + for s in steps: + d = s.get("delta", {}) + score = abs(d.get("goal_count_delta", 0)) * 3 + abs(d.get("hypothesis_delta", 0)) + abs(d.get("symbol_delta", 0)) * 2 + joints.append({ + "step": s["step"], + "tactic": s["tactic"], + "tactic_family": classify_tactic(s["tactic"]), + "delta_score": score, + "delta": d, + "result": s.get("result", "unknown"), + }) + + return { + "trace_version": "proof-trace-v1", + "receipt_hash": sha256(code), + "theorem_name": name, + "status": "verified" if final_result.get("ok") else "failed", + "tactic_count": len(steps), + "total_elapsed_ms": sum(s.get("elapsed_ms", 0) for s in steps), + "steps": steps, + "goal_transition_matrix": matrix, + "flexure_joints": joints, + } + + +def build_transition_matrix(steps: list[dict]) -> list[list[int]]: + """Build an adjacency matrix from goal-state hash transitions.""" + n = len(steps) + 1 # +1 for the final state + matrix = [[0] * n for _ in range(n)] + + # Collect unique goal hashes + hashes = [] + for s in steps: + if s.get("before_goal_hash"): + hashes.append(s["before_goal_hash"]) + if steps: + hashes.append(steps[-1].get("after_goal_hash", "")) + + # Assign indices + unique = list(dict.fromkeys(hashes)) + hash_to_idx = {h: i for i, h in enumerate(unique)} + + for s in steps: + bh = s.get("before_goal_hash", "") + ah = s.get("after_goal_hash", "") + if bh in hash_to_idx and ah in hash_to_idx: + i = hash_to_idx[bh] + j = hash_to_idx[ah] + matrix[i][j] += 1 + + return matrix + + +def extract_flexure_joints(steps: list[dict]) -> list[dict]: + """Extract flexure joints from tactic transitions. + + A flexure is a transition that significantly changes the goal state. + """ + joints = [] + for s in steps: + d = s.get("delta", {}) + score = ( + abs(d.get("goal_count_delta", 0)) * 3 + + abs(d.get("hypothesis_delta", 0)) + + abs(d.get("symbol_delta", 0)) * 2 + ) + joint = { + "step": s["step"], + "tactic": s["tactic"], + "tactic_family": classify_tactic(s["tactic"]), + "delta_score": score, + "delta": d, + "result": s.get("result", "unknown"), + } + joints.append(joint) + return joints + + +def classify_tactic(tactic: str) -> str: + """Classify a tactic into a family.""" + tactic_lower = tactic.lower() + if "simp" in tactic_lower: + return "normalization" + if "omega" in tactic_lower: + return "arithmetic" + if "ring" in tactic_lower or "nlinarith" in tactic_lower: + return "algebraic" + if "induction" in tactic_lower: + return "induction" + if "cases" in tactic_lower: + return "case_analysis" + if "rw" in tactic_lower or "rewrite" in tactic_lower: + return "rewrite" + if "apply" in tactic_lower or "exact" in tactic_lower: + return "discharge" + if "intro" in tactic_lower or "refine" in tactic_lower: + return "introduction" + if "calc" in tactic_lower: + return "calculation" + if "rfl" in tactic_lower: + return "reflexivity" + if "constructor" in tactic_lower: + return "constructor" + if "have" in tactic_lower or "let" in tactic_lower: + return "lemma_introduction" + return "unknown" + + +def main(): + if len(sys.argv) < 2: + print("Usage:", file=sys.stderr) + print(" python3 lean_trace_bridge.py --code 'theorem t ... := by ...' --name t", file=sys.stderr) + print(" python3 lean_trace_bridge.py shared-data/pist_canary_receipts.jsonl [index]", file=sys.stderr) + return 1 + + out_path = None + code = None + name = "unnamed" + + # Parse arguments + args = sys.argv[1:] + for i, arg in enumerate(args): + if arg == "--code" and i + 1 < len(args): + code = args[i + 1] + elif arg == "--name" and i + 1 < len(args): + name = args[i + 1] + elif arg == "--out" and i + 1 < len(args): + out_path = args[i + 1] + + # If no --code, check for receipts file + if code is None: + for arg in args: + if arg.endswith(".jsonl") and not arg.startswith("--"): + idx = 0 + for j, a2 in enumerate(args): + if a2 == arg and j + 1 < len(args) and args[j + 1].isdigit(): + idx = int(args[j + 1]) + with open(arg) as f: + for line_idx, line in enumerate(f): + if line_idx == idx: + receipt = json.loads(line) + code = receipt.get("theorem_statement", "") + name = receipt.get("theorem_name", f"receipt_{idx}") + break + break + + if code is None: + print("ERROR: No code provided", file=sys.stderr) + return 1 + + print(f"Building trace for: {name}", flush=True) + print(f"Code length: {len(code)} chars", flush=True) + + trace = build_trace(code, name) + + steps = trace.get("steps", []) + print(f"\nTrace complete:") + print(f" Steps: {len(steps)}") + print(f" Status: {trace.get('status')}") + print(f" Total time: {trace.get('total_elapsed_ms')}ms") + + families = {} + for s in steps: + j = s.get("flexure_joints", []) if isinstance(s, dict) else [] + for j in trace.get("flexure_joints", []): + fam = j.get("tactic_family", "?") + families[fam] = families.get(fam, 0) + 1 + + print(f"\nTactic families:") + for fam, count in sorted(families.items(), key=lambda x: -x[1]): + print(f" {fam:20s}: {count:3d}") + + print(f"\nTransition matrix: {len(trace.get('goal_transition_matrix', []))}x" + f"{len(trace.get('goal_transition_matrix', [[]]))}") + + # Save + if out_path: + with open(out_path, "w") as f: + json.dump(trace, f, indent=2) + print(f"\nTrace saved: {out_path}", flush=True) + else: + # Print summary + print(f"\nStep details:") + for s in steps[:5]: + jd = s.get("delta", {}) + print(f" [{s['step']}] {s['tactic']:30s} → {s['result']:8s} " + f"|{s.get('goal_count_before',0)}→{s.get('goal_count_after',0)}| " + f"d(g)={jd.get('goal_count_delta',0):+d} " + f"d(h)={jd.get('hypothesis_delta',0):+d}") + if len(steps) > 5: + print(f" ... ({len(steps) - 5} more steps)") + + return 0 + + +if __name__ == "__main__": + main() diff --git a/4-Infrastructure/shim/pist_trace_decompose.py b/4-Infrastructure/shim/pist_trace_decompose.py new file mode 100644 index 00000000..f3bf7f15 --- /dev/null +++ b/4-Infrastructure/shim/pist_trace_decompose.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +"""pist-trace-decompose — PIST spectral decomposition of a Lean proof trace. + +Analyzes the transition matrix from ProofTraceReceipt v1, +builds spectral features (eigenvalues, gap, rank, density), +and classifies the trace by tactic family distribution. + +Usage: + python3 pist_trace_decompose.py trace.json [--out report.json] +""" + +import json +import math +import os +import sys +from collections import Counter, defaultdict + +FEATURE_NAMES = [ + "n_steps", + "n_unique_goal_states", + "transition_density", + "spectral_gap", + "laplacian_zero_count", + "rank_estimate", + "avg_delta_goal_count", + "avg_delta_hypothesis", + "avg_delta_symbol", + "tactic_family_entropy", + "verified_ratio", + "total_elapsed", +] + + +def build_transition_matrix(steps: list[dict]) -> tuple[list[list[int]], list[str]]: + """Build adjacency matrix from goal-state hash transitions.""" + hashes = [] + for s in steps: + bh = s.get("before_goal_hash", "") + if bh: + hashes.append(bh) + if steps: + ah = steps[-1].get("after_goal_hash", "") + if ah: + hashes.append(ah) + + unique = list(dict.fromkeys(hashes)) + h2i = {h: i for i, h in enumerate(unique)} + n = len(unique) + if n == 0: + return [[0]], ["_empty"] + + matrix = [[0] * n for _ in range(n)] + for s in steps: + bh = s.get("before_goal_hash", "") + ah = s.get("after_goal_hash", "") + if bh in h2i and ah in h2i: + matrix[h2i[bh]][h2i[ah]] += 1 + + return matrix, unique + + +def symmetrize(matrix: list[list[int]]) -> list[list[float]]: + n = len(matrix) + if n == 0: + return [] + sym = [[0.0] * n for _ in range(n)] + for i in range(n): + for j in range(n): + sym[i][j] = (matrix[i][j] + matrix[j][i]) / 2.0 + return sym + + +def laplacian(matrix: list[list[float]]) -> list[list[float]]: + n = len(matrix) + if n == 0: + return [] + lap = [[0.0] * n for _ in range(n)] + for i in range(n): + deg = sum(matrix[i]) + for j in range(n): + if i == j: + lap[i][j] = deg + else: + lap[i][j] = -matrix[i][j] + return lap + + +def power_iteration(matrix: list[list[float]], max_iter: int = 100) -> tuple[float, list[float]]: + """Estimate the largest eigenvalue using power iteration.""" + n = len(matrix) + if n == 0: + return 0.0, [0.0] * n + v = [1.0 / math.sqrt(n)] * n + for _ in range(max_iter): + v_new = [sum(matrix[i][j] * v[j] for j in range(n)) for i in range(n)] + norm = math.sqrt(sum(x * x for x in v_new)) + if norm < 1e-12: + return 0.0, v + v = [x / norm for x in v_new] + # Check convergence + if all(abs(v_new[i] - v[i]) < 1e-6 for i in range(n)): + break + # Rayleigh quotient + 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, v + + +def spectral_analysis(matrix: list[list[int]]) -> dict: + """Compute spectral features from the transition matrix.""" + n = len(matrix) + if n == 0: + return {"error": "empty_matrix"} + + sym = symmetrize(matrix) + lap = laplacian(sym) + + # Largest eigenvalue via power iteration (symmetric matrix) + eig_max, _ = power_iteration(sym) + + # Spectral gap: difference between largest and 2nd largest + # (approximate by shifting the matrix and re-running) + shift = [[sym[i][j] for j in range(n)] for i in range(n)] + for i in range(n): + shift[i][i] -= 0.9 * eig_max # shift by 90% of largest + eig_shift, _ = power_iteration(shift) + eig_second_approx = 0.9 * eig_max + eig_shift + + gap = eig_max - eig_second_approx if eig_second_approx < eig_max else eig_max + + # Laplacian eigenvalues (power iteration on -lap for smallest) + lap_max, _ = power_iteration(lap) + + # Rank estimate: count rows with non-zero sum + row_sums = [sum(row) for row in matrix] + rank_est = sum(1 for s in row_sums if s > 0) + + # Zero-mode proxy: count near-zero rows + zero_rows = sum(1 for s in row_sums if s == 0) + + # Frobenius norm + frob_norm = math.sqrt(sum(sum(cell * cell for cell in row) for row in matrix)) + + # Trace + trace = sum(matrix[i][i] for i in range(n)) + + return { + "n_states": n, + "eigenvalue_max": round(eig_max, 6), + "spectral_gap": round(gap, 6), + "laplacian_eigenvalue_max": round(lap_max, 6), + "rank_estimate": rank_est, + "zero_rows": zero_rows, + "frobenius_norm": round(frob_norm, 6), + "trace": trace, + "density": sum(row_sums) / max(n * n, 1) if n > 0 else 0, + } + + +def analyze_trace(trace: dict) -> dict: + """Full spectral analysis of a proof trace.""" + steps = trace.get("steps", []) + flexure_joints = trace.get("flexure_joints", []) + + n = len(steps) + matrix, unique_hashes = build_transition_matrix(steps) + spectral = spectral_analysis(matrix) + + # Tactic family distribution + families = Counter(j.get("tactic_family", "?") for j in flexure_joints) + n_families = len(families) + if n_families > 0 and n > 0: + probs = [f / n for f in families.values()] + entropy = -sum(p * math.log2(p) for p in probs if p > 0) + else: + entropy = 0.0 + + # Step delta stats + goal_deltas = [abs(s.get("delta", {}).get("goal_count_delta", 0)) for s in steps if "delta" in s] + hyp_deltas = [abs(s.get("delta", {}).get("hypothesis_delta", 0)) for s in steps if "delta" in s] + sym_deltas = [abs(s.get("delta", {}).get("symbol_delta", 0)) for s in steps if "delta" in s] + + avg_gd = sum(goal_deltas) / max(len(goal_deltas), 1) + avg_hd = sum(hyp_deltas) / max(len(hyp_deltas), 1) + avg_sd = sum(sym_deltas) / max(len(sym_deltas), 1) + + # Result ratio + successes = sum(1 for s in steps if s.get("result") == "success") + + # Flexure delta scores + delta_scores = [j.get("delta_score", 0) for j in flexure_joints] + + feature_vector = [ + n, + len(unique_hashes), + spectral.get("density", 0), + spectral.get("spectral_gap", 0), + spectral.get("zero_rows", 0), + spectral.get("rank_estimate", 0), + avg_gd, + avg_hd, + avg_sd, + round(entropy, 4), + successes / max(n, 1), + trace.get("total_elapsed_ms", 0), + ] + + return { + "trace_summary": { + "theorem_name": trace.get("theorem_name", "?"), + "status": trace.get("status", "?"), + "n_steps": n, + "n_unique_goal_states": len(unique_hashes), + "n_flexure_joints": len(flexure_joints), + }, + "spectral": spectral, + "feature_vector": feature_vector, + "feature_names": FEATURE_NAMES, + "tactic_family_distribution": dict(families), + "delta_stats": { + "avg_goal_delta": round(avg_gd, 4), + "avg_hypothesis_delta": round(avg_hd, 4), + "avg_symbol_delta": round(avg_sd, 4), + "max_delta_score": max(delta_scores) if delta_scores else 0, + }, + "flexure_joints": flexure_joints, + } + + +def main(): + if len(sys.argv) < 2: + print("Usage: python3 pist_trace_decompose.py trace.json [--out report.json]", file=sys.stderr) + return 1 + + trace_path = sys.argv[1] + out_path = None + for i, arg in enumerate(sys.argv): + if arg == "--out" and i + 1 < len(sys.argv): + out_path = sys.argv[i + 1] + + with open(trace_path) as f: + trace = json.load(f) + + print(f"Analyzing trace: {trace.get('theorem_name', '?')}", flush=True) + print(f" Steps: {len(trace.get('steps', []))}", flush=True) + print(f" Status: {trace.get('status', '?')}", flush=True) + + result = analyze_trace(trace) + + ts = result["trace_summary"] + sp = result["spectral"] + print(f"\n Unique goal states: {ts['n_unique_goal_states']}", flush=True) + print(f" Spectral gap: {sp.get('spectral_gap', 0):.4f}", flush=True) + print(f" Rank estimate: {sp.get('rank_estimate', 0)}", flush=True) + print(f" Density: {sp.get('density', 0):.4f}", flush=True) + print(f" Flexure joints: {ts['n_flexure_joints']}", flush=True) + + print(f"\n Tactic families:", flush=True) + for fam, count in sorted(result["tactic_family_distribution"].items(), key=lambda x: -x[1]): + print(f" {fam:20s}: {count:3d}", flush=True) + + print(f"\n Delta stats:", flush=True) + ds = result["delta_stats"] + print(f" Avg goal delta: {ds['avg_goal_delta']:.2f}", flush=True) + print(f" Max delta score: {ds['max_delta_score']}", flush=True) + + if out_path: + with open(out_path, "w") as f: + json.dump(result, f, indent=2) + print(f"\nDecomposition: {out_path}", flush=True) + + return 0 + + +if __name__ == "__main__": + main()