Research-Stack/4-Infrastructure/shim/sidon_generation_kernel.py

#!/usr/bin/env python3
"""
sidon_generation_kernel.py — Build complete Sidon generation kernel from all sources.

Combines:
  - Arxiv DB: all Sidon-related papers (math.NT, math.CO)
  - RRC equations classified as Sidon-adjacent
  - OpenWebMath Sidon samples (41 extracted entries)
  - AlphaProof Nexus proofs (13 Lean files)
  - Existing SidonSets.lean theorems

Output: shared-data/data/sidon_generation_kernel_v1.json
"""
from __future__ import annotations

import json
import re
import subprocess
import sys
from collections import Counter, defaultdict
from pathlib import Path

NEON_HOST = "neon-64gb"
CONTAINER = "arxiv-pg"
DB = "arxiv"
ROOT = Path(__file__).resolve().parents[2]
OUT_PATH = ROOT / "shared-data/data/sidon_generation_kernel_v1.json"

# Sidon-related arxiv search queries
SIDON_QUERIES = {
    "sidon_sets": "title ILIKE '%Sidon%'",
    "sumset_additive": "(title ILIKE '%sumset%' OR title ILIKE '%sum set%') AND categories LIKE '%math.CO%'",
    "freiman": "title ILIKE '%Freiman%' AND (categories LIKE '%math.CO%' OR categories LIKE '%math.NT%')",
    "additive_energy": "title ILIKE '%additive energy%' OR title ILIKE '%additive combinatoric%'",
    "difference_sets": "title ILIKE '%difference set%' AND categories LIKE '%math.CO%'",
    "szemeredi": "title ILIKE '%Szemerédi%' AND categories LIKE '%math.CO%'",
    "cauchy_davenport": "title ILIKE '%Cauchy-Davenport%' OR title ILIKE '%Cauchy Davenport%'",
    "cap_set": "title ILIKE '%cap set%' AND categories LIKE '%math.CO%'",
    "projective_plane": "title ILIKE '%finite projective plane%' AND categories LIKE '%math.CO%'",
    "singer_difference": "title ILIKE '%Singer%' AND (title ILIKE '%difference%' OR title ILIKE '%Sidon%')",
}


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 main():
    print("=" * 60, file=sys.stderr)
    print("Sidon Generation Kernel Builder", file=sys.stderr)
    print("=" * 60, file=sys.stderr)

    sidon_papers: dict[str, dict] = {}
    sidon_types = Counter()
    source_counts = Counter()
    sidon_eqs: list[dict] = []

    # ── 1. Harvest Sidon papers from arxiv DB ──
    print("\n[1/4] Harvesting Sidon papers from arxiv DB...", file=sys.stderr)
    for topic, condition in SIDON_QUERIES.items():
        rows = ssh_query(f"SELECT paper_id, title, categories, substring(abstract, 1, 300) FROM arxiv_papers WHERE {condition} LIMIT 50")
        for r in rows:
            if len(r) >= 2:
                pid = r[0]
                if pid not in sidon_papers:
                    sidon_papers[pid] = {
                        "paper_id": pid,
                        "title": r[1],
                        "categories": r[2] if len(r) > 2 else "",
                        "abstract_snippet": r[3] if len(r) > 3 else "",
                        "topic": topic,
                    }
                    sidon_types[topic] += 1

    print(f"  Found {len(sidon_papers)} unique Sidon papers from arxiv DB", file=sys.stderr)
    for t, c in sorted(sidon_types.items(), key=lambda x: -x[1]):
        print(f"    {t:25s} {c}", file=sys.stderr)

    # ── 2. Incorporate RRC Sidon-adjacent equations ──
    print("\n[2/4] Incorporating RRC equation data...", file=sys.stderr)
    receipt_path = ROOT / "archive/experimental-shim-probes/rrc_equation_classifier_receipt.json"
    if receipt_path.exists():
        receipt = json.loads(receipt_path.read_text())
        sidon_eqs = []
        for e in receipt["compiled_equations"]:
            rec = e["equation_record"]
            route = rec.get("manifold_route", "")
            regime = rec.get("manifold_regime", "")
            name = rec.get("name", "")
            if route in ("number_theory", "combinatorics", "geometry_topology"):
                sidon_eqs.append({
                    "name": name,
                    "route": route,
                    "regime": regime,
                    "paper_id": rec.get("arxiv_paper_id", ""),
                    "sidon_label": rec.get("manifold_sidon_label", 0),
                    "strand": rec.get("manifold_strand", 0),
                    "slack": rec.get("manifold_slack", 0),
                })
        print(f"  {len(sidon_eqs)} Sidon-adjacent RRC equations", file=sys.stderr)

    # ── 3. Incorporate APN proofs ──
    print("\n[3/4] Incorporating AlphaProof Nexus proofs...", file=sys.stderr)
    apn_dir = ROOT / "0-Core-Formalism/lean/Semantics/Semantics/Adapters/AlphaProofNexus"
    apn_files = sorted([f for f in apn_dir.iterdir() if f.suffix == ".lean" and f.name != "Bridge.lean"])
    for f in apn_files:
        content = f.read_text()
        has_sidon = "Sidon" in content or "sidon" in content or "IsSidon" in content
        size = f.stat().st_size
        print(f"    {f.name:45s} {size/1024:6.1f} KB {'[Sidon]' if has_sidon else ''}", file=sys.stderr)
        source_counts["apn_proof"] += 1

    # ── 4. Incorporate OpenWebMath Sidon samples ──
    print("\n[4/4] Incorporating OpenWebMath Sidon samples...", file=sys.stderr)
    owm_path = ROOT / "shared-data/data/sidon_samples.jsonl"
    if owm_path.exists():
        with open(owm_path) as f:
            samples = [json.loads(line) for line in f if line.strip()]
        print(f"  {len(samples)} OpenWebMath Sidon samples", file=sys.stderr)
        source_counts["openwebmath"] = len(samples)

    # ── Build the kernel ──
    kernel = {
        "schema": "sidon_generation_kernel_v1",
        "generated_at": __import__("time").strftime("%Y-%m-%dT%H:%M:%SZ"),
        "description": "Complete Sidon generation kernel — all known Sidon-related content",
        "sources": {
            "arxiv_papers": len(sidon_papers),
            "rrc_equations": len(sidon_eqs) if 'sidon_eqs' in dir() else 0,
            "apn_proofs": source_counts.get("apn_proof", 0),
            "openwebmath_samples": source_counts.get("openwebmath", 0),
        },
        "sidon_types": {t: c for t, c in sorted(sidon_types.items(), key=lambda x: -x[1])},
        "papers": sorted(sidon_papers.values(), key=lambda x: x["paper_id"]),
        "rrc_equations": sidon_eqs if 'sidon_eqs' in dir() else [],
        "lean_proofs": [{"file": f.name, "size": f.stat().st_size} for f in apn_files],
    }

    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    with open(OUT_PATH, "w") as f:
        json.dump(kernel, f, indent=2, ensure_ascii=False)

    total = kernel["sources"]["arxiv_papers"] + kernel["sources"]["rrc_equations"] + kernel["sources"]["apn_proofs"]
    print(f"\n{'='*60}", file=sys.stderr)
    print(f"Sidon generation kernel complete", file=sys.stderr)
    print(f"  Arxiv Sidon papers: {kernel['sources']['arxiv_papers']}", file=sys.stderr)
    print(f"  RRC Sidon equations: {kernel['sources']['rrc_equations']}", file=sys.stderr)
    print(f"  APN Lean proofs: {kernel['sources']['apn_proofs']}", file=sys.stderr)
    print(f"  OpenWebMath samples: {kernel['sources']['openwebmath_samples']}", file=sys.stderr)
    print(f"  Total: {sum(kernel['sources'].values())} entries", file=sys.stderr)
    print(f"  Saved to {OUT_PATH}", file=sys.stderr)

    return 0


if __name__ == "__main__":
    sys.exit(main())