mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
allaun
b038778361
- SDPVerify.lean: certificate verification engine (714 lines, compiles cleanly) - GoormaghtighCert.lean: Goormaghtigh conjecture SDP certificate - HachimojiManifoldAxiom/HachimojiSubstitution: Hachimoji DNA encoding - GeneticBraidBridge.lean: genetic algorithm braid bridge - load_dependency_graph/load_module_graph: Gremlin Cosmos DB graph loaders - test_graph_queries/rrc_math_xref: graph verification queries - Gremlin mathblob DB provisioned and accessible - Branch cleanup: deleted 11 stale remote branches Build: 3297 jobs, 0 errors (lake build Semantics.SDPVerify)
399 lines
13 KiB
Python
399 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
rrc_slo_analyzer.py — Service Level Objectives for the RRC refactoring oracle.
|
||
|
||
Measures structural and performance SLOs on one or two graph states
|
||
and logs whether refactoring improved each metric.
|
||
|
||
SLO dimensions:
|
||
Structural: spectral_gap, modularity, conductance, isolation_ratio,
|
||
centrality_spread, edge_efficiency, community_count
|
||
Performance: adj_build_ms, eigenvector_ms, command_gen_ms,
|
||
iteration_ms, total_ms
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import sys
|
||
import time
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
import numpy as np
|
||
|
||
SHIM = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(SHIM))
|
||
from burgers_chaos_game import BurgersChaosGame
|
||
|
||
|
||
SLO_THRESHOLDS = {
|
||
"spectral_gap": {"good": 0.15, "acceptable": 0.05},
|
||
"modularity": {"good": 0.4, "acceptable": 0.2},
|
||
"conductance": {"good": 0.3, "acceptable": 0.1},
|
||
"isolation_ratio": {"good": 0.0, "acceptable": 0.05},
|
||
"centrality_spread": {"good": 0.02, "acceptable": 0.005},
|
||
"edge_efficiency": {"good": 0.3, "acceptable": 0.1},
|
||
"community_count": {"good": 8, "acceptable": 4},
|
||
}
|
||
|
||
|
||
def b_kappa(v: float, kappa: float) -> float:
|
||
"""Softplus retraction map from "A Differentiable IPM in Single Precision".
|
||
|
||
b_κ(v) = (v + √(v² + 4κ)) / 2
|
||
|
||
Key properties:
|
||
· b_κ(v) · b_κ(−v) = κ (complementarity by construction)
|
||
· 0 < ∂b_κ/∂v ≤ 1 (bounded KKT block, prevents 10¹⁶ conditioning)
|
||
|
||
In the RRC/Q16_16 context: kappa maps to BraidBracket.kappa (≤ 0.25 at
|
||
eigensolid, i.e. IsTopologicallyTrivial), keeping the spectral gap SLO
|
||
well-conditioned in fixed-point arithmetic.
|
||
"""
|
||
return (v + math.sqrt(v * v + 4.0 * kappa)) / 2.0
|
||
|
||
|
||
def jsrr_spectral_loss(residues: list[float]) -> float:
|
||
"""STARS JSRR loss: mean squared residual over strands.
|
||
|
||
L_JSRR^(t) = (1/N) Σᵢ ‖j^(i)‖₂²
|
||
|
||
In the RRC context: residues are the per-strand Q16_16 residue values
|
||
(BraidEigensolid.strandResidue), normalized to [0, 1]. At eigensolid
|
||
(IsEigensolid), this value stabilizes — the spectral radius proxy ρ²(J)
|
||
has reached its fixed point, matching BraidEigensolid.jsrr_profile_fixed.
|
||
|
||
Returns 0.0 for an empty residue list.
|
||
"""
|
||
if not residues:
|
||
return 0.0
|
||
return sum(r * r for r in residues) / len(residues)
|
||
|
||
|
||
def load_graph(path: Path) -> dict:
|
||
return json.loads(path.read_text())
|
||
|
||
|
||
def graph_to_adj(graph: dict) -> tuple[np.ndarray, list[str]]:
|
||
nodes = graph.get("nodes", [])
|
||
edges = graph.get("edges", [])
|
||
node_ids = [n["id"] for n in nodes]
|
||
id_to_idx = {nid: i for i, nid in enumerate(node_ids)}
|
||
N = len(node_ids)
|
||
A = np.zeros((N, N), dtype=np.float64)
|
||
for e in edges:
|
||
src, tgt = e.get("source", ""), e.get("target", "")
|
||
if src in id_to_idx and tgt in id_to_idx:
|
||
i, j = id_to_idx[src], id_to_idx[tgt]
|
||
A[i, j] = A[j, i] = 1.0
|
||
return A, node_ids
|
||
|
||
|
||
def laplacian(A: np.ndarray) -> np.ndarray:
|
||
D = np.diag(A.sum(axis=1))
|
||
return D - A
|
||
|
||
|
||
def measure_structural_slos(graph: dict) -> dict:
|
||
A, node_ids = graph_to_adj(graph)
|
||
N = A.shape[0]
|
||
nodes = graph.get("nodes", [])
|
||
edges = graph.get("edges", [])
|
||
D = A.sum(axis=1)
|
||
|
||
isolation_ratio = float(np.sum(D == 0)) / max(N, 1)
|
||
|
||
if N < 2:
|
||
return {
|
||
"spectral_gap": 0.0,
|
||
"modularity": 0.0,
|
||
"conductance": 0.0,
|
||
"isolation_ratio": isolation_ratio,
|
||
"centrality_spread": 0.0,
|
||
"edge_efficiency": 0.0,
|
||
"community_count": 0,
|
||
"num_nodes": N,
|
||
"num_edges": len(edges),
|
||
"density": 0.0,
|
||
}
|
||
|
||
density = (2 * len(edges)) / max(N * (N - 1), 1)
|
||
|
||
# Spectral gap of Laplacian
|
||
if N >= 3:
|
||
eigvals = np.sort(np.linalg.eigvalsh(laplacian(A)))
|
||
spectral_gap = float(eigvals[-1] - eigvals[-2]) if len(eigvals) >= 2 else 0.0
|
||
else:
|
||
spectral_gap = 0.0
|
||
|
||
# Modularity (Newman)
|
||
m = A.sum() / 2
|
||
if m > 0:
|
||
mod = 0.0
|
||
for i in range(N):
|
||
for j in range(N):
|
||
if A[i, j] > 0:
|
||
mod += A[i, j] - (D[i] * D[j]) / (2 * m)
|
||
modularity = float(mod / (2 * m))
|
||
else:
|
||
modularity = 0.0
|
||
|
||
# Conductance (average over non-isolated nodes)
|
||
total_edges = A.sum() / 2
|
||
conductances = []
|
||
for i in range(N):
|
||
if D[i] > 0:
|
||
vol_i = D[i]
|
||
cut_i = 0
|
||
for j in range(N):
|
||
if A[i, j] > 0 and D[j] <= D[i]:
|
||
cut_i += 1
|
||
conductances.append(cut_i / min(vol_i, total_edges - vol_i + 1))
|
||
conductance = float(np.mean(conductances)) if conductances else 0.0
|
||
|
||
# Centrality spread
|
||
game = BurgersChaosGame(A, node_ids)
|
||
centrality = game.eigenvector_centrality()
|
||
centrality_spread = float(np.std(centrality))
|
||
|
||
# Edge efficiency (fraction of pairs with edges that have strong centrality product)
|
||
centrality_product_thresh = 0.001
|
||
paired = 0
|
||
efficient = 0
|
||
for i in range(N):
|
||
for j in range(i + 1, N):
|
||
if A[i, j] > 0:
|
||
paired += 1
|
||
if centrality[i] * centrality[j] > centrality_product_thresh:
|
||
efficient += 1
|
||
edge_efficiency = float(efficient) / max(paired, 1)
|
||
|
||
# Community count (connected components of thresholded centrality graph)
|
||
threshold = np.percentile(centrality, 50)
|
||
adj_strong = (A > 0) & (np.outer(centrality, centrality) > threshold**2)
|
||
visited = set()
|
||
community_count = 0
|
||
for i in range(N):
|
||
if i not in visited:
|
||
community_count += 1
|
||
stack = [i]
|
||
while stack:
|
||
v = stack.pop()
|
||
if v not in visited:
|
||
visited.add(v)
|
||
for u in range(N):
|
||
if adj_strong[v, u] and u not in visited:
|
||
stack.append(u)
|
||
|
||
return {
|
||
"spectral_gap": round(spectral_gap, 6),
|
||
"modularity": round(modularity, 6),
|
||
"conductance": round(conductance, 6),
|
||
"isolation_ratio": round(isolation_ratio, 6),
|
||
"centrality_spread": round(centrality_spread, 6),
|
||
"edge_efficiency": round(edge_efficiency, 6),
|
||
"community_count": community_count,
|
||
"num_nodes": N,
|
||
"num_edges": len(edges),
|
||
"density": round(density, 6),
|
||
}
|
||
|
||
|
||
def measure_performance_slos(graph: dict) -> dict:
|
||
A, node_ids = graph_to_adj(graph)
|
||
N = A.shape[0]
|
||
|
||
t0 = time.perf_counter()
|
||
_ = graph_to_adj(graph)
|
||
adj_time = (time.perf_counter() - t0) * 1000
|
||
|
||
game = BurgersChaosGame(A, node_ids)
|
||
|
||
t0 = time.perf_counter()
|
||
_ = game.eigenvector_centrality()
|
||
cent_time = (time.perf_counter() - t0) * 1000
|
||
|
||
t0 = time.perf_counter()
|
||
_ = game.evolve(1.0)
|
||
evolve_time = (time.perf_counter() - t0) * 1000
|
||
|
||
return {
|
||
"adj_build_ms": round(adj_time, 2),
|
||
"eigenvector_ms": round(cent_time, 2),
|
||
"evolution_ms": round(evolve_time, 2),
|
||
"total_ms": round(adj_time + cent_time + evolve_time, 2),
|
||
"num_nodes": N,
|
||
}
|
||
|
||
|
||
def grade_slo(name: str, value: float, thresholds: dict) -> str:
|
||
if value >= thresholds["good"]:
|
||
return "GOOD"
|
||
elif value >= thresholds["acceptable"]:
|
||
return "ACCEPTABLE"
|
||
else:
|
||
return "FAIL"
|
||
|
||
|
||
def compare_slos(
|
||
baseline: dict,
|
||
target: dict,
|
||
label: str = "refactored",
|
||
) -> list[dict]:
|
||
results = []
|
||
for key in SLO_THRESHOLDS:
|
||
b = baseline.get(key, 0)
|
||
t = target.get(key, 0)
|
||
thresholds = SLO_THRESHOLDS[key]
|
||
|
||
higher_better = key not in ("isolation_ratio",)
|
||
improved = (t > b) if higher_better else (t < b)
|
||
regressed = (t < b) if higher_better else (t > b)
|
||
|
||
grade_b = grade_slo(key, b, thresholds)
|
||
grade_t = grade_slo(key, t, thresholds)
|
||
|
||
results.append({
|
||
"slo": key,
|
||
"baseline": b,
|
||
"target": t,
|
||
"delta": round(t - b, 6),
|
||
"improved": improved,
|
||
"regressed": regressed,
|
||
"baseline_grade": grade_b,
|
||
"target_grade": grade_t,
|
||
})
|
||
return results
|
||
|
||
|
||
def main() -> int:
|
||
import argparse
|
||
parser = argparse.ArgumentParser(
|
||
description="RRC SLO Analyzer — compare structural & perf objectives"
|
||
)
|
||
parser.add_argument("--baseline", "-b", required=True,
|
||
help="Baseline graph JSON (e.g. original)")
|
||
parser.add_argument("--target", "-t",
|
||
help="Target graph JSON (e.g. refactored sacrificial)")
|
||
parser.add_argument("--output", "-o", default=None,
|
||
help="Output receipt path")
|
||
args = parser.parse_args()
|
||
|
||
print("=" * 60)
|
||
print("RRC SLO Analyzer")
|
||
print("=" * 60)
|
||
print()
|
||
|
||
baseline_graph = load_graph(Path(args.baseline))
|
||
print(f"Baseline: {Path(args.baseline).name}")
|
||
print(f" {len(baseline_graph.get('nodes', []))} nodes, "
|
||
f"{len(baseline_graph.get('edges', []))} edges")
|
||
|
||
target_graph = None
|
||
if args.target:
|
||
target_graph = load_graph(Path(args.target))
|
||
print(f"Target: {Path(args.target).name}")
|
||
print(f" {len(target_graph.get('nodes', []))} nodes, "
|
||
f"{len(target_graph.get('edges', []))} edges")
|
||
print()
|
||
|
||
# ── Structural SLOs ──
|
||
print("--- Structural SLOs ---")
|
||
baseline_s = measure_structural_slos(baseline_graph)
|
||
print(f" Baseline: N={baseline_s['num_nodes']} "
|
||
f"E={baseline_s['num_edges']} "
|
||
f"ρ={baseline_s['density']} "
|
||
f"λ_gap={baseline_s['spectral_gap']} "
|
||
f"Q={baseline_s['modularity']} "
|
||
f"c_std={baseline_s['centrality_spread']}")
|
||
|
||
comparison = []
|
||
if target_graph:
|
||
target_s = measure_structural_slos(target_graph)
|
||
print(f" Target: N={target_s['num_nodes']} "
|
||
f"E={target_s['num_edges']} "
|
||
f"ρ={target_s['density']} "
|
||
f"λ_gap={target_s['spectral_gap']} "
|
||
f"Q={target_s['modularity']} "
|
||
f"c_std={target_s['centrality_spread']}")
|
||
comparison = compare_slos(baseline_s, target_s)
|
||
|
||
for row in comparison:
|
||
arrow = "↑" if row["improved"] else ("↓" if row["regressed"] else "→")
|
||
print(f" {row['slo']:20s} "
|
||
f"{row['baseline']:10.6f} → {row['target']:10.6f} "
|
||
f"({row['delta']:+9.6f}) {arrow} "
|
||
f"[{row['baseline_grade']}→{row['target_grade']}]")
|
||
|
||
improved = sum(1 for r in comparison if r["improved"])
|
||
regressed = sum(1 for r in comparison if r["regressed"])
|
||
total = len(comparison)
|
||
print(f"\n SLO verdict: {improved}/{total} improved, "
|
||
f"{regressed}/{total} regressed")
|
||
else:
|
||
print(f" (no target — structural SLO report only)")
|
||
|
||
print()
|
||
|
||
# ── Performance SLOs ──
|
||
print("--- Performance SLOs ---")
|
||
baseline_p = measure_performance_slos(baseline_graph)
|
||
print(f" Baseline: A={baseline_p['adj_build_ms']}ms "
|
||
f"EV={baseline_p['eigenvector_ms']}ms "
|
||
f"ev={baseline_p['evolution_ms']}ms "
|
||
f"total={baseline_p['total_ms']}ms")
|
||
|
||
if target_graph:
|
||
target_p = measure_performance_slos(target_graph)
|
||
print(f" Target: A={target_p['adj_build_ms']}ms "
|
||
f"EV={target_p['eigenvector_ms']}ms "
|
||
f"ev={target_p['evolution_ms']}ms "
|
||
f"total={target_p['total_ms']}ms")
|
||
|
||
for key in ("adj_build_ms", "eigenvector_ms", "evolution_ms", "total_ms"):
|
||
b = baseline_p[key]
|
||
t = target_p[key]
|
||
delta = t - b
|
||
arrow = "↓" if t < b else ("↑" if t > b else "→")
|
||
print(f" {key:20s} {b:8.2f} → {t:8.2f} ms ({delta:+8.2f}) {arrow}")
|
||
else:
|
||
print(f" (no target — performance SLO report only)")
|
||
|
||
# ── Build receipt ──
|
||
result = {
|
||
"schema": "rrc_slo_analysis_v1",
|
||
"claim_boundary": (
|
||
"structural-and-performance-slo-analysis;"
|
||
"no-decision-logic;measurement-only"
|
||
),
|
||
"baseline": str(Path(args.baseline).resolve()),
|
||
"target": str(Path(args.target).resolve()) if args.target else None,
|
||
"baseline_structural": baseline_s,
|
||
"target_structural": target_s if target_graph else None,
|
||
"baseline_performance": baseline_p,
|
||
"target_performance": target_p if target_graph else None,
|
||
"comparison": comparison if comparison else None,
|
||
"slo_thresholds": SLO_THRESHOLDS,
|
||
}
|
||
|
||
import hashlib
|
||
from datetime import datetime, timezone
|
||
canonical = json.dumps(result, sort_keys=True, separators=(",", ":"))
|
||
result["receipt_sha256"] = hashlib.sha256(canonical.encode()).hexdigest()
|
||
result["computed_at"] = datetime.now(timezone.utc).isoformat()
|
||
|
||
if args.output:
|
||
output_path = Path(args.output)
|
||
else:
|
||
output_path = SHIM / "rrc_slo_receipt.json"
|
||
output_path.write_text(json.dumps(result, indent=2, default=str))
|
||
print(f"\nReceipt: {output_path}")
|
||
print(f"SHA256: {result['receipt_sha256']}")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|