allaun/Research-Stack
1
0
Fork 0
mirror of https://github.com/allaunthefox/Research-Stack.git synced 2026-07-31 03:05:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(pist): Tier 2 trace bridge — tactic-level goal transitions

Browse source 
- 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
This commit is contained in:
Brandon Schneider 2026-05-26 02:28:16 -05:00
parent e7525fb6f4
commit 29bef16216
2 changed files with 765 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

489
4-Infrastructure/shim/lean_trace_bridge.py Normal file
View file

@ -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 <canary_receipt.jsonl [index]> [--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()

276
4-Infrastructure/shim/pist_trace_decompose.py Normal file
View file

@ -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()