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

581 lines
20 KiB
Python

#!/usr/bin/env python3
"""PIST -> RRC receipt-density backfill injector.
NOTE (ontology migration):
This file is a **legacy shim**. It exists to keep historical backfill workflows
running while the AVM rewrite is underway.
**Target architecture:** Lean-only AVM ISA + backend shims.
- Lean defines all semantics.
- Shims do JSON/RDS I/O only.
This script still contains scoring math in Python (float-based) and therefore
MUST be treated as a non-authoritative conversion surface.
Rules until ported:
- Output is always `promotion = not_promoted`.
- Output must carry an explicit `strip_receipt` section explaining:
- which constructs were computed in shim space
- what must be ported to Lean/AVM
TODO(lean-port): Replace all scoring and warning decisions with Lean/AVM.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import re
import sys
from collections import Counter
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
REPO_ROOT = Path(__file__).resolve().parents[2]
SHIM_DIR = Path(__file__).resolve().parent
if str(SHIM_DIR) not in sys.path:
sys.path.insert(0, str(SHIM_DIR))
DEFAULT_RRC_FILE = REPO_ROOT / "6-Documentation/docs/rrc_equation_classification.md"
DEFAULT_PIST_REPORT = REPO_ROOT / "shared-data/rrc_pist_exact_validation.json"
DEFAULT_OUT = REPO_ROOT / "shared-data/rrc_receipt_density_backfill.json"
DEFAULT_RDS_TABLE = "ene.rrc_receipt_density"
ONTOLOGY_VERSION = "shim-ontology-migration-v1"
TARGET_AXES = {
"projection_declared",
"negative_control_strength",
"witness_declared",
"scale_band_declared",
"shape_closure",
}
STATUS_BASE = {
"BLOCKED": 0.0,
"HOLD": 0.12,
"CANDIDATE": 0.45,
"REVIEWED": 0.78,
"VERIFIED": 0.84,
}
SHAPE_DOMAIN = {
"CognitiveLoadField": "analysis",
"SignalShapedRouteCompiler": "topology",
"ProjectableGeometryTopology": "geometry",
"CadForceProbeReceipt": "physics",
"LogogramProjection": "symbolic",
"HoldForUnlawfulOrUnderspecifiedShape": "unknown",
}
@dataclass(frozen=True)
class RRCEquationRow:
equation_id: str
rrc_shape: str
status: str
top_axes: list[str]
@dataclass(frozen=True)
class ReceiptDensityRecord:
receipt_version: str
equation_id: str
rrc_shape: str
domain: str
source_status: str
receipt_density: float
confidence: float
density_components: dict[str, float]
shape_prediction: dict[str, Any]
top_axes: list[str]
status: str
promotion: str
source: str
receipt_hash: str
warnings: list[str]
def stable_hash(payload: Any) -> str:
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def clamp01(value: float) -> float:
if math.isnan(value) or math.isinf(value):
return 0.0
return max(0.0, min(1.0, value))
def parse_axis_list(raw: str) -> list[str]:
return [part.strip().strip("`") for part in raw.split(",") if part.strip()]
def is_table_noise(equation: str, shape: str, status: str) -> bool:
bad = {"", "---", "Equation", "RRC shape", "Status"}
if equation in bad or shape in bad or status in bad:
return True
return bool(re.fullmatch(r"-+", equation)) or bool(re.fullmatch(r"-+", shape))
def parse_rrc_table(path: Path) -> list[RRCEquationRow]:
if not path.exists():
raise FileNotFoundError(f"RRC classification file not found: {path}")
text = path.read_text(encoding="utf-8")
if "## Sample Projections" not in text:
raise ValueError(f"No '## Sample Projections' section found in {path}")
sample_section = text.split("## Sample Projections", 1)[1]
sample_section = sample_section.split("\n## ", 1)[0]
rows: list[RRCEquationRow] = []
for line in sample_section.splitlines():
line = line.strip()
if not line.startswith("|"):
continue
parts = [p.strip().strip("`") for p in line.strip("|").split("|")]
if len(parts) < 4:
continue
equation, shape, status, axes = parts[:4]
if is_table_noise(equation, shape, status):
continue
rows.append(
RRCEquationRow(
equation_id=equation,
rrc_shape=shape,
status=status,
top_axes=parse_axis_list(axes),
)
)
return rows
def load_pist_predictions(path: Path) -> dict[str, dict[str, Any]]:
if not path.exists():
return {}
data = json.loads(path.read_text(encoding="utf-8"))
predictions = data.get("predictions", [])
out: dict[str, dict[str, Any]] = {}
for pred in predictions:
equation = str(pred.get("equation", ""))
ground_truth = str(pred.get("ground_truth", ""))
if is_table_noise(equation, ground_truth, pred.get("proxy_pred", "")):
continue
out[equation] = pred
return out
def spectral_quality(pred: dict[str, Any] | None) -> float:
if not pred:
return 0.0
rank = float(pred.get("rank_estimate") or 0.0)
gap = float(pred.get("spectral_gap") or 0.0)
crossing_density = float(pred.get("crossing_density") or 0.0)
entropy = float(pred.get("strand_entropy") or 0.0)
lap_zero = float(pred.get("laplacian_zero_count") or 0.0)
rank_score = clamp01(rank / 8.0)
gap_score = clamp01(gap)
entropy_score = clamp01(entropy / 3.0)
density_score = clamp01(crossing_density / 0.5)
lap_score = 1.0 if lap_zero >= 1.0 else 0.45
hash_score = 1.0 if pred.get("canonical_hash") and pred.get("matrix_hash") else 0.0
return round(
clamp01(
0.24 * rank_score
+ 0.18 * gap_score
+ 0.18 * entropy_score
+ 0.12 * density_score
+ 0.12 * lap_score
+ 0.16 * hash_score
),
6,
)
def shape_agreement(row: RRCEquationRow, pred: dict[str, Any] | None) -> float:
if not pred:
return 0.0
exact = pred.get("exact_pred")
proxy = pred.get("proxy_pred")
if exact == row.rrc_shape:
return 1.0
if proxy == row.rrc_shape:
return 0.82
if exact or proxy:
return 0.35
return 0.0
def axis_score(row: RRCEquationRow) -> float:
if not row.top_axes:
return 0.0
hits = len(TARGET_AXES.intersection(row.top_axes))
return round(clamp01(hits / 4.0), 6)
def status_score(row: RRCEquationRow) -> float:
return STATUS_BASE.get(row.status.upper(), 0.2)
def compute_density(row: RRCEquationRow, pred: dict[str, Any] | None) -> tuple[float, float, dict[str, float], list[str]]:
warnings: list[str] = []
s_status = status_score(row)
s_axes = axis_score(row)
s_spectral = spectral_quality(pred)
s_shape = shape_agreement(row, pred)
if pred is None:
warnings.append("missing_pist_prediction")
elif s_shape < 0.5:
warnings.append("pist_shape_disagreement")
density = clamp01(
0.26 * s_status
+ 0.24 * s_axes
+ 0.26 * s_spectral
+ 0.24 * s_shape
)
confidence = clamp01(
0.20 * s_status
+ 0.20 * s_axes
+ 0.28 * s_spectral
+ 0.32 * s_shape
)
components = {
"status_score": round(s_status, 6),
"axis_score": round(s_axes, 6),
"spectral_quality": round(s_spectral, 6),
"shape_agreement": round(s_shape, 6),
}
return round(density, 6), round(confidence, 6), components, warnings
def build_record(row: RRCEquationRow, pred: dict[str, Any] | None) -> ReceiptDensityRecord:
density, confidence, components, warnings = compute_density(row, pred)
shape_prediction = {
"ground_truth_hint": row.rrc_shape,
"proxy_pred": pred.get("proxy_pred") if pred else None,
"exact_pred": pred.get("exact_pred") if pred else None,
"matrix_hash": pred.get("matrix_hash") if pred else None,
"canonical_hash": pred.get("canonical_hash") if pred else None,
"spectral_gap": pred.get("spectral_gap") if pred else None,
"rank_estimate": pred.get("rank_estimate") if pred else None,
"laplacian_zero_count": pred.get("laplacian_zero_count") if pred else None,
}
unsigned_payload = {
"equation_id": row.equation_id,
"rrc_shape": row.rrc_shape,
"source_status": row.status,
"receipt_density": density,
"confidence": confidence,
"shape_prediction": shape_prediction,
"top_axes": row.top_axes,
"promotion": "not_promoted",
"source": "pist_receipt_density_injector_v1",
"ontology_version": ONTOLOGY_VERSION,
}
receipt_hash = stable_hash(unsigned_payload)
return ReceiptDensityRecord(
receipt_version="pist-receipt-density-v1",
equation_id=row.equation_id,
rrc_shape=row.rrc_shape,
domain=SHAPE_DOMAIN.get(row.rrc_shape, "unknown"),
source_status=row.status,
receipt_density=density,
confidence=confidence,
density_components=components,
shape_prediction=shape_prediction,
top_axes=row.top_axes,
status="CANDIDATE" if density > 0.0 else "HOLD",
promotion="not_promoted",
source="pist_receipt_density_injector_v1",
receipt_hash=receipt_hash,
warnings=warnings,
)
def summarize(records: list[ReceiptDensityRecord], total_rows: int, prediction_count: int) -> dict[str, Any]:
by_shape = Counter(r.rrc_shape for r in records)
by_status = Counter(r.status for r in records)
warning_counts: Counter[str] = Counter()
for r in records:
warning_counts.update(r.warnings)
densities = [r.receipt_density for r in records]
confidences = [r.confidence for r in records]
populated = sum(1 for r in records if r.receipt_density > 0.0)
return {
"receipt_version": "pist-receipt-density-v1",
"ontology_version": ONTOLOGY_VERSION,
"shim_role": "legacy_scoring_surface_pending_avm",
"input_rows": total_rows,
"records": len(records),
"pist_predictions_loaded": prediction_count,
"receipt_density_populated": populated,
"receipt_density_missing": len(records) - populated,
"coverage": round(populated / len(records), 6) if records else 0.0,
"mean_receipt_density": round(sum(densities) / len(densities), 6) if densities else 0.0,
"min_receipt_density": round(min(densities), 6) if densities else 0.0,
"max_receipt_density": round(max(densities), 6) if densities else 0.0,
"mean_confidence": round(sum(confidences) / len(confidences), 6) if confidences else 0.0,
"by_shape": dict(sorted(by_shape.items())),
"by_status": dict(sorted(by_status.items())),
"warning_counts": dict(sorted(warning_counts.items())),
"promotion_policy": "no automatic promotion; density populates routing evidence only",
"float_policy": {
"status": "legacy_float_math_present",
"reason": "shim computes density components using Python float; must be ported to Lean/AVM",
},
}
def emit_jsonl(records: Iterable[ReceiptDensityRecord], path: Path) -> None:
with path.open("w", encoding="utf-8") as f:
for record in records:
f.write(json.dumps(asdict(record), sort_keys=True) + "\n")
def split_qualified_table(table: str) -> tuple[str, str]:
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)?", table):
raise ValueError(f"Unsafe SQL table identifier: {table!r}")
if "." in table:
schema, name = table.split(".", 1)
else:
schema, name = "public", table
return schema, name
def quote_ident(identifier: str) -> str:
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", identifier):
raise ValueError(f"Unsafe SQL identifier: {identifier!r}")
return '"' + identifier.replace('"', '""') + '"'
def create_sidecar_table(cur: Any, table: str) -> None:
schema, name = split_qualified_table(table)
q_schema = quote_ident(schema)
q_name = quote_ident(name)
full = f"{q_schema}.{q_name}"
cur.execute(f"CREATE SCHEMA IF NOT EXISTS {q_schema}")
cur.execute(
f"""
CREATE TABLE IF NOT EXISTS {full} (
equation_id TEXT PRIMARY KEY,
rrc_shape TEXT NOT NULL,
domain TEXT NOT NULL,
source_status TEXT NOT NULL,
receipt_density DOUBLE PRECISION NOT NULL,
receipt_density_source TEXT NOT NULL,
receipt_density_hash TEXT NOT NULL,
receipt_density_status TEXT NOT NULL,
receipt_density_warnings JSONB NOT NULL,
confidence DOUBLE PRECISION NOT NULL,
top_axes JSONB NOT NULL,
shape_prediction JSONB NOT NULL,
density_components JSONB NOT NULL,
promotion TEXT NOT NULL CHECK (promotion = 'not_promoted'),
payload JSONB NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
"""
)
def upsert_sidecar_records(conn: Any, records: list[ReceiptDensityRecord], table: str) -> dict[str, Any]:
schema, name = split_qualified_table(table)
full = f"{quote_ident(schema)}.{quote_ident(name)}"
now = datetime.now(timezone.utc).isoformat()
rows = [asdict(r) for r in records]
with conn.cursor() as cur:
create_sidecar_table(cur, table)
for row in rows:
cur.execute(
f"""
INSERT INTO {full} (
equation_id,
rrc_shape,
domain,
source_status,
receipt_density,
receipt_density_source,
receipt_density_hash,
receipt_density_status,
receipt_density_warnings,
confidence,
top_axes,
shape_prediction,
density_components,
promotion,
payload,
updated_at
) VALUES (
%(equation_id)s,
%(rrc_shape)s,
%(domain)s,
%(source_status)s,
%(receipt_density)s,
%(source)s,
%(receipt_hash)s,
%(status)s,
%(warnings_json)s::jsonb,
%(confidence)s,
%(top_axes_json)s::jsonb,
%(shape_prediction_json)s::jsonb,
%(density_components_json)s::jsonb,
%(promotion)s,
%(payload_json)s::jsonb,
%(updated_at)s
)
ON CONFLICT (equation_id) DO UPDATE SET
rrc_shape = EXCLUDED.rrc_shape,
domain = EXCLUDED.domain,
source_status = EXCLUDED.source_status,
receipt_density = EXCLUDED.receipt_density,
receipt_density_source = EXCLUDED.receipt_density_source,
receipt_density_hash = EXCLUDED.receipt_density_hash,
receipt_density_status = EXCLUDED.receipt_density_status,
receipt_density_warnings = EXCLUDED.receipt_density_warnings,
confidence = EXCLUDED.confidence,
top_axes = EXCLUDED.top_axes,
shape_prediction = EXCLUDED.shape_prediction,
density_components = EXCLUDED.density_components,
promotion = EXCLUDED.promotion,
payload = EXCLUDED.payload,
updated_at = EXCLUDED.updated_at
""",
{
**row,
"warnings_json": json.dumps(row["warnings"], sort_keys=True),
"top_axes_json": json.dumps(row["top_axes"], sort_keys=True),
"shape_prediction_json": json.dumps(row["shape_prediction"], sort_keys=True),
"density_components_json": json.dumps(row["density_components"], sort_keys=True),
"payload_json": json.dumps(row, sort_keys=True),
"updated_at": now,
},
)
conn.commit()
return {
"enabled": True,
"mode": "sidecar",
"table": table,
"records_upserted": len(records),
"promotion_policy": "not_promoted only",
}
def write_rds(records: list[ReceiptDensityRecord], table: str, connect_timeout: int | None) -> dict[str, Any]:
try:
from rds_connect import connect_rds
except ImportError as exc:
raise RuntimeError(
"--write-rds requires 4-Infrastructure/shim/rds_connect.py to be importable"
) from exc
overrides: dict[str, Any] = {}
if connect_timeout is not None:
overrides["connect_timeout"] = connect_timeout
conn = connect_rds(**overrides)
try:
return upsert_sidecar_records(conn, records, table)
except Exception:
conn.rollback()
raise
finally:
conn.close()
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Generate RRC receipt-density records from PIST outputs.")
parser.add_argument("--rrc-file", type=Path, default=DEFAULT_RRC_FILE)
parser.add_argument("--pist-report", type=Path, default=DEFAULT_PIST_REPORT)
parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
parser.add_argument("--jsonl-out", type=Path, default=None, help="Optional JSONL output path for DB import.")
parser.add_argument("--fail-on-missing-pist", action="store_true", help="Exit nonzero if any row lacks a PIST prediction.")
parser.add_argument("--write-rds", action="store_true", help="Opt-in RDS write. Defaults to false / audit JSON only.")
parser.add_argument("--rds-table", default=DEFAULT_RDS_TABLE, help=f"Qualified sidecar table. Default: {DEFAULT_RDS_TABLE}")
parser.add_argument("--connect-timeout", type=int, default=10, help="RDS connection timeout override passed to rds_connect.connect_rds.")
args = parser.parse_args(argv)
rows = parse_rrc_table(args.rrc_file)
predictions = load_pist_predictions(args.pist_report)
records = [build_record(row, predictions.get(row.equation_id)) for row in rows]
summary = summarize(records, total_rows=len(rows), prediction_count=len(predictions))
rds_result = None
if args.write_rds:
rds_result = write_rds(records, table=args.rds_table, connect_timeout=args.connect_timeout)
summary["rds_write"] = rds_result
else:
summary["rds_write"] = {"enabled": False, "reason": "--write-rds not set"}
payload = {
"summary": summary,
"strip_receipt": {
"ontology_version": ONTOLOGY_VERSION,
"shim_role": "legacy_scoring_surface_pending_avm",
"computed_in_shim": [
"receipt_density",
"confidence",
"density_components",
"warnings",
],
"must_port_to_lean_avm": [
"compute_density",
"spectral_quality",
"shape_agreement",
"axis_score",
"status_score",
"warning assignment",
],
"float_policy": "legacy_float_math_present; reject once AVM port is active",
},
"inputs": {
"rrc_file": str(args.rrc_file),
"pist_report": str(args.pist_report),
},
"claim_boundary": {
"receipt_density_means": "routing evidence is populated (legacy shim surface)",
"receipt_density_does_not_mean": "mathematical proof or promotion",
"promotion_policy": "not_promoted for every generated record",
"rds_policy": "--write-rds upserts sidecar receipt-density metadata only via rds_connect.connect_rds",
},
"records": [asdict(r) for r in records],
}
args.out.parent.mkdir(parents=True, exist_ok=True)
args.out.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
if args.jsonl_out is not None:
args.jsonl_out.parent.mkdir(parents=True, exist_ok=True)
emit_jsonl(records, args.jsonl_out)
print(json.dumps(summary, indent=2, sort_keys=True))
print(f"Wrote audit JSON: {args.out}", file=sys.stderr)
if args.jsonl_out is not None:
print(f"Wrote JSONL import file: {args.jsonl_out}", file=sys.stderr)
if rds_result is not None:
print(f"RDS write complete: {rds_result}", file=sys.stderr)
if args.fail_on_missing_pist and summary["warning_counts"].get("missing_pist_prediction", 0) > 0:
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())