From 27efd54f6d6f95507ba63014d579a17eea6d3189 Mon Sep 17 00:00:00 2001 From: allaun Date: Mon, 22 Jun 2026 02:20:17 -0500 Subject: [PATCH] feat(infra): add lake build ingestion shim for ENE database MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds 4-Infrastructure/shim/lake_build_ingest.py, a pure-I/O shim that: - runs `lake build ` in the Semantics directory, - parses job count, error count, warnings, and status, - inserts a session, package, receipt, and ingest_event into the ENE database on neon-64gb (arxiv-pg container). The script contains no admissibility logic and no Float arithmetic. Tested with `lake build Semantics.NKHodgeFAMM` — 8316 jobs, 0 errors, session d1ea53db2185df87dd41497da12101f5. --- 4-Infrastructure/shim/lake_build_ingest.py | 301 +++++++++++++++++++++ 1 file changed, 301 insertions(+) create mode 100644 4-Infrastructure/shim/lake_build_ingest.py diff --git a/4-Infrastructure/shim/lake_build_ingest.py b/4-Infrastructure/shim/lake_build_ingest.py new file mode 100644 index 00000000..a04ce87a --- /dev/null +++ b/4-Infrastructure/shim/lake_build_ingest.py @@ -0,0 +1,301 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.10" +# dependencies = [] +# /// +""" +lake_build_ingest.py — Run `lake build` and ingest the result into ENE on neon-64gb. + +This is a pure-I/O shim: it runs the build, parses the output, and writes rows to +Postgres. It contains no admissibility logic and no Float arithmetic. + +Target database: arxiv-pg container on neon-64gb (Tailscale 100.92.88.64). +Tables used: + ene.sessions — one row per build invocation + ene.packages — one row per build target/receipt + ene.receipts — verification receipt bound to the package + ene.ingest_events — provenance for the ingestion itself + +Usage: + python3 4-Infrastructure/shim/lake_build_ingest.py [target] [--dir LEAN_DIR] [--actor NAME] + +Examples: + python3 4-Infrastructure/shim/lake_build_ingest.py + python3 4-Infrastructure/shim/lake_build_ingest.py Compiler + python3 4-Infrastructure/shim/lake_build_ingest.py Semantics.NKHodgeFAMM --actor opencode +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +ROOT = Path("/home/allaun/Research Stack") +LEAN_DIR = ROOT / "0-Core-Formalism/lean/Semantics" +NEON_HOST = "100.92.88.64" +CONTAINER = "arxiv-pg" +DB = "ene" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _run(cmd: list[str], cwd: Path, timeout: float = 3600.0) -> subprocess.CompletedProcess: + return subprocess.run( + cmd, + cwd=cwd, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def _ssh_psql(sql: str, timeout: int = 60) -> str: + """Run a psql command on neon-64gb's arxiv-pg container.""" + escaped = sql.replace('"', '\\"') + result = subprocess.run( + [ + "ssh", + f"allaun@{NEON_HOST}", + f'podman exec {CONTAINER} psql -U postgres -d {DB} -c "{escaped}"', + ], + capture_output=True, + text=True, + timeout=timeout, + ) + if result.returncode != 0: + raise RuntimeError(f"psql failed: {result.stderr[:500]}") + return result.stdout + + +def _ssh_psql_rows(sql: str, timeout: int = 60) -> list[list[str]]: + """Run a psql query on neon-64gb and return rows.""" + escaped = sql.replace('"', '\\"') + result = subprocess.run( + [ + "ssh", + f"allaun@{NEON_HOST}", + f'podman exec {CONTAINER} psql -U postgres -d {DB} -t -A -F "|" -c "{escaped}"', + ], + capture_output=True, + text=True, + timeout=timeout, + ) + if result.returncode != 0: + raise RuntimeError(f"psql failed: {result.stderr[:500]}") + rows = [line.split("|") for line in result.stdout.strip().split("\n") if line] + return rows + + +def _git_info() -> dict[str, str]: + result = _run(["git", "rev-parse", "HEAD"], ROOT) + commit = result.stdout.strip() if result.returncode == 0 else "" + result = _run(["git", "rev-parse", "--abbrev-ref", "HEAD"], ROOT) + branch = result.stdout.strip() if result.returncode == 0 else "" + result = _run(["git", "status", "--short"], ROOT) + dirty = bool(result.stdout.strip()) + return { + "commit": commit, + "branch": branch, + "dirty": str(dirty), + } + + +def _parse_build_output(stdout: str, stderr: str) -> dict[str, Any]: + """Extract job count, error count, and warnings from lake build output.""" + combined = stdout + "\n" + stderr + total: int | None = None + errors: int | None = None + + # Match "[N/M] Built Module" and final "Built X (Ys)" or "Build completed" + built_re = re.compile(r"\[(\d+)/(\d+)\] Built") + for m in built_re.finditer(combined): + total = int(m.group(2)) + + # Lake final summary: "Build completed successfully (N jobs, M errors)" + summary_re = re.compile(r"Build completed.*?\((\d+) jobs?.*?\)", re.IGNORECASE) + m = summary_re.search(combined) + if m: + total = int(m.group(1)) + + # Error count from "X errors" or error markers + error_re = re.compile(r"(\d+) error", re.IGNORECASE) + m = error_re.search(combined) + if m: + errors = int(m.group(1)) + + # Warnings + warnings = len(re.findall(r"warning:", combined, re.IGNORECASE)) + + # Determine status + if "Build completed successfully" in combined or "Built" in combined: + status = "success" if not errors else "warnings" + else: + status = "failed" + + return { + "total": total, + "errors": errors if errors is not None else (0 if status != "failed" else None), + "warnings": warnings, + "status": status, + "raw_output": combined[-20000:], # cap size + } + + +def _insert_session(target: str, git: dict[str, str], build: dict[str, Any], elapsed_ms: int, actor: str) -> str: + """Insert a row into ene.sessions and return the session id.""" + session_id = hashlib.sha256(f"{git['commit']}:{target}:{_now()}".encode()).hexdigest()[:32] + title = f"lake build {target}" if target else "lake build" + summary = json.dumps( + { + "target": target, + "jobs_total": build["total"], + "jobs_failed": build["errors"], + "warnings": build["warnings"], + "status": build["status"], + }, + separators=(",", ":"), + ) + sql = f""" + INSERT INTO ene.sessions + (id, project_id, title, started_at, ended_at, actor, toolchain, + summary, memory_hash, promotion_state, build_target, jobs_total, + jobs_failed, elapsed_ms, git_commit, git_branch) + VALUES + ('{session_id}', 'research-stack', '{title}', '{_now()}', '{_now()}', + '{actor}', 'lake build', + '{summary}'::jsonb, '', 'held', '{target}', + {build['total'] if build['total'] is not None else 'NULL'}, + {build['errors'] if build['errors'] is not None else 'NULL'}, + {elapsed_ms}, '{git['commit']}', '{git['branch']}') + ON CONFLICT (id) DO NOTHING; + """ + _ssh_psql(sql) + return session_id + + +def _insert_package(session_id: str, target: str, git: dict[str, str], build: dict[str, Any]) -> str: + """Insert a package row representing the build artifact and return its id.""" + pkg_id = hashlib.sha256(f"lake-build:{git['commit']}:{target}".encode()).hexdigest()[:32] + content = build["raw_output"] + content_hash = hashlib.sha256(content.encode()).hexdigest() + title = f"lake build {target}" if target else "lake build" + sql = f""" + INSERT INTO ene.packages + (pkg, package_type, title, content, content_hash, element, + tags, source, created_at, updated_at, session_id, provenance, + verification_status, promotion_state, domain, archetype, notes) + VALUES + ('{pkg_id}', 'build_log', '{title}', '{_escape_sql(content)}', + '{content_hash}', 'METAL', + '["lake", "build", "lean"]'::jsonb, + '4-Infrastructure/shim/lake_build_ingest.py', + '{_now()}', '{_now()}', '{session_id}', + '{{"git_commit": "{git['commit']}", "git_branch": "{git['branch']}", "dirty": {str(git['dirty']).lower()}}}'::jsonb, + '{build['status']}', 'held', 'core-formalism', 'build_artifact', + 'Build output captured from lake build {target}') + ON CONFLICT (pkg) DO UPDATE SET + content = EXCLUDED.content, + content_hash = EXCLUDED.content_hash, + updated_at = EXCLUDED.updated_at, + session_id = EXCLUDED.session_id, + verification_status = EXCLUDED.verification_status, + notes = EXCLUDED.notes; + """ + _ssh_psql(sql) + return pkg_id + + +def _insert_receipt(pkg_id: str, session_id: str, build: dict[str, Any]) -> None: + """Insert a verification receipt for the build package.""" + receipt_hash = hashlib.sha256(build["raw_output"].encode()).hexdigest() + input_hash = hashlib.sha256(build["raw_output"].encode()).hexdigest() + residual = json.dumps( + {"warnings": build["warnings"], "errors": build["errors"]}, + separators=(",", ":"), + ) + sql = f""" + INSERT INTO ene.receipts + (package_id, receipt_type, receipt_hash, input_hash, output_hash, + toolchain, verifier, status, residual) + VALUES + ('{pkg_id}', 'lake_build', '{receipt_hash}', '{input_hash}', + '{receipt_hash}', 'lake build', 'opencode', '{build['status']}', + '{residual}'::jsonb); + """ + _ssh_psql(sql) + + +def _insert_ingest_event(pkg_id: str, session_id: str) -> None: + """Record provenance for the ingestion itself.""" + sql = f""" + INSERT INTO ene.ingest_events + (package_id, source_uri, source_type, source_hash, ingest_profile, + parser_version, extracted_entities, extracted_equations, + extracted_links, extracted_bibtex, warnings) + VALUES + ('{pkg_id}', 'file://4-Infrastructure/shim/lake_build_ingest.py', + 'build_log_ingest', '', 'lake_build_ingest_v1', '1.0.0', + '{{"session_id": "{session_id}"}}'::jsonb, '[]'::jsonb, + '[]'::jsonb, '[]'::jsonb, '[]'::jsonb); + """ + _ssh_psql(sql) + + +def _escape_sql(s: str) -> str: + return s.replace("'", "''").replace("\\", "\\\\") + + +def main() -> int: + parser = argparse.ArgumentParser(description="Ingest lake build output into ENE on neon-64gb") + parser.add_argument("target", nargs="?", default="", help="lake build target (default: full build)") + parser.add_argument("--dir", type=Path, default=LEAN_DIR, help="directory to run lake build in") + parser.add_argument("--actor", default="opencode", help="actor name for session") + parser.add_argument("--dry-run", action="store_true", help="print parsed build info without writing") + args = parser.parse_args() + + actor = args.actor + target = args.target + cmd = ["lake", "build"] + if target: + cmd.append(target) + + print(f"Running: {' '.join(cmd)} in {args.dir}", file=sys.stderr) + start = time.time() + proc = _run(cmd, args.dir) + elapsed_ms = int((time.time() - start) * 1000) + + git = _git_info() + build = _parse_build_output(proc.stdout, proc.stderr) + + print( + f"Build status: {build['status']} | jobs: {build['total']} | " + f"errors: {build['errors']} | warnings: {build['warnings']} | " + f"elapsed: {elapsed_ms}ms", + file=sys.stderr, + ) + + if args.dry_run: + print(json.dumps({"git": git, "build": build, "elapsed_ms": elapsed_ms}, indent=2)) + return 0 + + session_id = _insert_session(target, git, build, elapsed_ms, actor) + pkg_id = _insert_package(session_id, target, git, build) + _insert_receipt(pkg_id, session_id, build) + _insert_ingest_event(pkg_id, session_id) + + print(f"Ingested: session={session_id} package={pkg_id}", file=sys.stderr) + return 0 if build["status"] != "failed" else 1 + + +if __name__ == "__main__": + sys.exit(main())