#!/usr/bin/env python3 """ rrc_manifold_refine.py — Refine RRC classification using manifold + slack regimes. Uses the Anti-Diophantine slack to refine arxiv matches: - Diophantine regime (slack < 8, tight constraints): need better paper matches → search arxiv DB for more specific category-matched papers - Anti-Diophantine regime (slack ≥ 128, roomy): matches are fine → verify category alignment - Transition regime: check for upgrades Usage: python3 4-Infrastructure/shim/rrc_manifold_refine.py """ from __future__ import annotations import json import re import subprocess import sys from collections import defaultdict from pathlib import Path NEON_HOST = "neon-64gb" CONTAINER = "arxiv-pg" DB = "arxiv" RECEIPT_PATH = Path("archive/experimental-shim-probes/rrc_equation_classifier_receipt.json") ROUTE_SEARCH_TERMS = { "thermodynamic_energy": ["thermodynamic", "energy", "entropy", "heat", "temperature"], "geometry_topology": ["geometry", "topology", "manifold", "curvature", "riemannian"], "cognitive_load": ["cognitive", "neural", "brain", "attention", "cognition"], "compression_route": ["compression", "coding", "entropy", "rate-distortion", "source coding"], "magnetic_signal": ["magnetic", "plasma", "magnetohydrodynamic", "alfven"], "control_signal": ["control", "feedback", "optimal control", "stability"], "chaotic_couch": ["chaos", "chaotic", "strange attractor", "turbulence"], "number_theory": ["number theory", "diophantine", "prime", "arithmetic"], } def ssh_query(sql: str, timeout: int = 30) -> list[list[str]]: result = subprocess.run([ "ssh", NEON_HOST, f"podman exec {CONTAINER} psql -U postgres -d {DB} -t -A -F '|' -c \"{sql}\"" ], capture_output=True, text=True, timeout=timeout) return [line.split("|") for line in result.stdout.strip().split("\n") if line] def get_arxiv_category(paper_id: str) -> tuple[str, str]: """Get arxiv category for a paper. Returns (category, title).""" rows = ssh_query(f"SELECT categories, title FROM arxiv_papers WHERE paper_id = '{paper_id}'") if rows and len(rows[0]) >= 1: cats = rows[0][0] title = rows[0][1] if len(rows[0]) > 1 else "" primary = cats.split()[0] if cats else "unknown" return primary, title return "unknown", "" def find_better_match(name: str, route: str, terms: list[str]) -> dict | None: """Search arxiv DB for a better paper match using route-specific keywords. Scores results by keyword density in title for best match.""" if not terms: return None # Require at least one title match title_where = " OR ".join(f"title ILIKE '%{t}%'" for t in terms[:5]) sql = f""" SELECT paper_id, title, categories, substring(abstract, 1, 200) FROM arxiv_papers WHERE ({title_where}) AND (categories LIKE '%math%' OR categories LIKE '%nlin%' OR categories LIKE '%cs%') ORDER BY paper_id LIMIT 100 """ rows = ssh_query(sql, timeout=15) if not rows or len(rows[0]) < 2: return None # Score by how many terms appear in the title best = None best_score = 0 for r in rows: if len(r) < 2: continue title_lower = r[1].lower() score = sum(3 for t in terms if t in title_lower) # Bonus for matching name components for part in name.replace("_", " ").lower().split(): if part in title_lower and len(part) > 3: score += 2 if score > best_score: best_score = score best = { "paper_id": r[0], "title": r[1], "categories": r[2] if len(r) > 2 else "", "snippet": r[3] if len(r) > 3 else "", } return best def main(): print("=" * 60, file=sys.stderr) print("RRC Manifold Refinement", file=sys.stderr) print("=" * 60, file=sys.stderr) d = json.loads(RECEIPT_PATH.read_text()) eqs = d["compiled_equations"] refined = 0 regime_changes = 0 new_matches = 0 route_alignments = defaultdict(lambda: {"total": 0, "known": 0, "math_nt": 0}) for e in eqs: rec = e["equation_record"] name = rec.get("name", "?") route = rec.get("manifold_route", "unclassified") regime = rec.get("manifold_regime", "diophantine") slack = rec.get("manifold_slack", 0) match_count = rec.get("arxiv_match_count") or 0 paper_id = rec.get("arxiv_paper_id", "") stage = rec.get("arxiv_match_stage", "none") route_alignments[route]["total"] += 1 if not paper_id or paper_id in ["", "?"]: continue category, title = get_arxiv_category(paper_id) if category != "unknown": route_alignments[route]["known"] += 1 if category.startswith("math.NT"): route_alignments[route]["math_nt"] += 1 # ---- Refinement 1: Regime adjustment based on actual match count ---- if isinstance(match_count, (int, float)) and route != "unclassified": old_regime = regime if match_count >= 100 and regime != "anti_diophantine": rec["manifold_regime"] = "anti_diophantine" rec["manifold_slack"] = 128 regime_changes += 1 elif match_count < 5 and regime != "diophantine" and regime != "transition_tight": rec["manifold_regime"] = "diophantine" rec["manifold_slack"] = 4 regime_changes += 1 # ---- Refinement 2: Category-based paper upgrade for Diophantine eqs ---- if rec.get("manifold_regime", regime) in ("diophantine", "transition_tight"): if category == "unknown" and route != "unclassified": terms = ROUTE_SEARCH_TERMS.get(route, []) + [name.replace("_", " ")] better = find_better_match(name, route, terms) if better: rec["arxiv_paper_id_previous"] = rec["arxiv_paper_id"] rec["arxiv_paper_id"] = better["paper_id"] rec["arxiv_match_title"] = better["title"] rec["arxiv_match_abstract"] = better["snippet"] rec["arxiv_match_category"] = better["categories"] rec["arxiv_match_count"] = 1 rec["arxiv_match_stage"] = "manifold_refine_v1" new_matches += 1 print(f" UPGRADE: {name:35s} {route:20s} → {better['paper_id']} [{better['categories'][:20]}]", file=sys.stderr) refined += 1 RECEIPT_PATH.write_text(json.dumps(d, indent=2, ensure_ascii=False)) # Summary print(f"\n{'='*60}", file=sys.stderr) print(f"Refinement Summary", file=sys.stderr) print(f" Equations refined: {refined}/250", file=sys.stderr) print(f" Regime changes: {regime_changes}", file=sys.stderr) print(f" New paper matches: {new_matches}", file=sys.stderr) print(f"\nRoute arxiv coverage:", file=sys.stderr) for route, stats in sorted(route_alignments.items(), key=lambda x: -x[1]["total"]): pct = stats["known"] / stats["total"] * 100 if stats["total"] > 0 else 0 nt = stats["math_nt"] print(f" {route:30s} {stats['known']:3d}/{stats['total']:3d} known ({pct:5.1f}%) NT={nt}", file=sys.stderr) # Show final regime distribution regimes = defaultdict(int) for e in d["compiled_equations"]: regimes[e["equation_record"].get("manifold_regime", "?")] += 1 print(f"\nRegimes after refinement: {dict(regimes)}", file=sys.stderr) if __name__ == "__main__": main()