From 0686b39428ab2a50186b10dbf426aa408e23a180 Mon Sep 17 00:00:00 2001 From: Allaun Silverfox <28494262+allaunthefox@users.noreply.github.com> Date: Tue, 26 May 2026 15:13:35 -0500 Subject: [PATCH] feat(pist): use shared rds_connect for receipt-density writer --- .../shim/pist_receipt_density_injector.py | 176 +++++++++++++++++- 1 file changed, 172 insertions(+), 4 deletions(-) diff --git a/4-Infrastructure/shim/pist_receipt_density_injector.py b/4-Infrastructure/shim/pist_receipt_density_injector.py index 7b2a8e52..1112128e 100644 --- a/4-Infrastructure/shim/pist_receipt_density_injector.py +++ b/4-Infrastructure/shim/pist_receipt_density_injector.py @@ -8,7 +8,9 @@ It is intentionally conservative: * It DOES NOT promote equations. * It DOES NOT mutate RDS/ENE by default. -* It writes an audit JSON that can later be consumed by a database writer. +* RDS writes require the explicit --write-rds flag. +* RDS writes default to a sidecar table: ene.rrc_receipt_density. +* Database connectivity is delegated to the shared rds_connect.connect_rds helper. * It filters Markdown table header/separator rows that older validation scripts accidentally treated as equations. @@ -32,17 +34,24 @@ import argparse import hashlib import json import math +import os 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" TARGET_AXES = { "projection_declared", @@ -235,8 +244,6 @@ def compute_density(row: RRCEquationRow, pred: dict[str, Any] | None) -> tuple[f elif s_shape < 0.5: warnings.append("pist_shape_disagreement") - # Receipt density is deliberately not a proof score. It measures routing - # evidence density from declared axes + status + structural PIST signal. density = clamp01( 0.26 * s_status + 0.24 * s_axes @@ -244,7 +251,6 @@ def compute_density(row: RRCEquationRow, pred: dict[str, Any] | None) -> tuple[f + 0.24 * s_shape ) - # Confidence is slightly more classifier-weighted than density. confidence = clamp01( 0.20 * s_status + 0.20 * s_axes @@ -342,6 +348,155 @@ def emit_jsonl(records: Iterable[ReceiptDensityRecord], path: Path) -> None: 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) @@ -349,6 +504,9 @@ def main(argv: list[str] | None = None) -> int: 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) @@ -357,6 +515,13 @@ def main(argv: list[str] | None = None) -> int: 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, "inputs": { @@ -367,6 +532,7 @@ def main(argv: list[str] | None = None) -> int: "receipt_density_means": "routing evidence is populated", "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], } @@ -382,6 +548,8 @@ def main(argv: list[str] | None = None) -> int: 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