SilverSight/scripts/auto/auto_pipeline.py
allaun 85141a4b94 feat(nuvmap,braid): NUVMAP port + Rossby/Kelvin braid correspondence
- Port NUVMAP projection engine from Research Stack to SilverSight
  with Q16_16 fixed-point (zero Float) and CBOR serialization
- Add Rotational Wave — Braid Correspondence formalization at boundary
  (ChiralLabel, RossbyDrift, rossby_convergence_bound stubbed,
   kelvin_wave_eigensolid proven)
- Add auto-pipeline CI workflow, webhook receiver, Forgejo MCP server
- Add SOPS/Age encryption config
- Add stack compose for portable deployment
- Add rotational wave design doc
2026-06-30 16:38:11 -05:00

103 lines
4.2 KiB
Python

#!/usr/bin/env python3
"""
Auto-pipeline: Lean build → extract metadata → populate DB → RRC classify.
Runs after every push to Semantics Lean sources.
Usage:
python3 auto_pipeline.py # full pipeline
python3 auto_pipeline.py --db-only # recreate DB schema only
python3 auto_pipeline.py --ci # CI mode (no build, just extract+predict)
"""
import subprocess, json, os, sys, argparse
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
NEON_PG = "postgres://postgres:postgres@100.92.88.64:5432"
def sh(cmd, **kw):
return subprocess.run(cmd, shell=True, capture_output=True, text=True, **kw)
def pg(sql, db="research_stack"):
r = sh(f"psql {NEON_PG}/{db} -c {shlex.quote(sql)}")
if r.returncode != 0:
print(f" DB error: {r.stderr[:200]}")
# ── 1. Extract Lean metadata ──────────────────────────────────────────────
SCAN_DIRS = [
ROOT / "0-Core-Formalism" / "lean" / "Semantics",
ROOT / "0-Core-Formalism" / "lean" / "SilverSight",
]
def extract_theorems():
"""Scan Lean files for theorems, lemmas, defs, sorries."""
theorems = []
for d in SCAN_DIRS:
for f in sorted(d.rglob("*.lean")):
if ".lake" in str(f):
continue
rel = f.relative_to(ROOT)
text = f.read_text()
for line_no, line in enumerate(text.split("\n"), 1):
for kw in ("theorem ", "lemma ", "def "):
if kw in line:
name = line.split(kw)[1].split()[0].split(":")[0].split(" ")[0]
theorems.append({
"name": name,
"kind": kw.strip(),
"file": str(rel),
"line": line_no,
"has_sorry": "sorry" in text,
"source": f.readlines(),
})
return theorems
# ── 2. Populate ENE DB ────────────────────────────────────────────────────
def populate_ene(theorems):
pg("CREATE SCHEMA IF NOT EXISTS ene", "research_stack")
for t in theorems:
pg(f"""INSERT INTO ene.packages (pkg, package_type, title, source, domain)
VALUES ('lean:{t["file"]}:{t["name"]}', 'lean_theorem', '{t["name"]}', '{t["file"]}', 'lean')
ON CONFLICT (pkg) DO NOTHING""", "research_stack")
# ── 3. Run RRC Classification ─────────────────────────────────────────────
def run_rrc():
"""Run RRC compile pipeline and update predictions."""
r = sh("cd {} && lake build Compiler 2>&1".format(
ROOT / "0-Core-Formalism" / "lean" / "Semantics"), timeout=600)
return r.returncode == 0
# ── Main ──────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--db-only", action="store_true")
parser.add_argument("--ci", action="store_true")
args = parser.parse_args()
if args.db_only:
import shlex
sql_path = Path(__file__).with_name("ene_schema.sql")
r = sh(f"psql {NEON_PG}/research_stack -f {sql_path}")
print(r.stdout[-200:] if r.stdout else r.stderr[-200:])
return
if not args.ci:
print("[pipeline] Building Lean...")
ok = run_rrc()
print(f"[pipeline] Build {'OK' if ok else 'FAILED'}")
print("[pipeline] Extracting theorems...")
theorems = extract_theorems()
print(f" Found {len(theorems)} theorem/lemma/def sites")
print("[pipeline] Populating ENE...")
populate_ene(theorems)
print(" Done")
print("[pipeline] RRC classification...")
# Call the RRC compile skill logic here
print(" RRC: pending integration")
print("[pipeline] Complete")
if __name__ == "__main__":
main()