mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
feat(infra): add lake build ingestion shim for ENE database
Adds 4-Infrastructure/shim/lake_build_ingest.py, a pure-I/O shim that: - runs `lake build <target>` 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.
This commit is contained in:
parent
69988453bc
commit
27efd54f6d
1 changed files with 301 additions and 0 deletions
301
4-Infrastructure/shim/lake_build_ingest.py
Normal file
301
4-Infrastructure/shim/lake_build_ingest.py
Normal file
|
|
@ -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())
|
||||
Loading…
Add table
Reference in a new issue