feat(delphi): storage v2 P6a — per-run input snapshots (capture side) + purge tool

delphi/delphi_storage/inputs.py

@@ -0,0 +1,191 @@
+"""Per-run input snapshots — the capture side of P6 (design §4.2 entity 2,
+§4.4).
+
+Captures, once at job start, EXACTLY what the pipeline stages read from the
+source PostgreSQL — most critically the votes as the RAW stream stage 1
+consumes: ``ORDER BY created``, raw PG signs (the production math path never
+flips; ``fetch_votes``'s flip is dead code), superseded votes included, order
+preserved (the math KMeans init deliberately depends on vote-encounter order).
+The SQL constants here are imported by ``polismath/run_math_pipeline.py`` so
+capture and stage 1 can never diverge.
+
+Snapshots are stored as ``run_inputs`` items (pk = job_id, sk = kind) using
+the codec envelopes; each returns a fingerprint (sha256 of the canonical
+uncompressed payload + row counts + max vote ``created``) for the P7 manifest.
+"""
+
+import base64
+import hashlib
+import os
+from decimal import Decimal
+from typing import Any, Optional
+
+from delphi_storage.codec import canonical_json_dumps, encode_payload, decode_payload
+from delphi_storage.interface import DelphiStore, Invalid, NotFound
+from delphi_storage.models import StoreItem
+
+SNAPSHOT_KINDS = (
+    "votes",
+    "comments",
+    "participants",
+    "conversation_meta",
+    "report_comment_selections",
+    "clojure_math_main",
+)
+
+#: Stage 1's exact vote reads (polismath/run_math_pipeline.py imports these).
+VOTES_COUNT_SQL = "SELECT COUNT(*) FROM votes WHERE zid = %s"
+VOTES_BATCH_SQL = (
+    "SELECT v.created, v.tid, v.pid, v.vote FROM votes v "
+    "WHERE v.zid = %s ORDER BY v.created LIMIT %s OFFSET %s"
+)
+
+_TABLE_QUERIES = {
+    "comments": "SELECT * FROM comments WHERE zid = %s ORDER BY tid",
+    "participants": "SELECT * FROM participants WHERE zid = %s ORDER BY pid",
+    "conversation_meta": "SELECT * FROM conversations WHERE zid = %s",
+    "report_comment_selections": (
+        "SELECT * FROM report_comment_selections WHERE zid = %s ORDER BY rid, tid"
+    ),
+    "clojure_math_main": "SELECT * FROM math_main WHERE zid = %s ORDER BY math_env",
+}
+
+
+def _connect(pg_url: Optional[str]):
+    """Same source-PG resolution as stage 1 (run_math_pipeline.connect_to_db)."""
+    import psycopg2
+
+    timeout = int(os.environ.get("POSTGRES_CONNECT_TIMEOUT", "30"))
+    url = pg_url or os.environ.get("DATABASE_URL")
+    if url:
+        return psycopg2.connect(url, connect_timeout=timeout)
+    return psycopg2.connect(
+        host=os.environ.get("DATABASE_HOST", "localhost"),
+        port=os.environ.get("DATABASE_PORT", "5432"),
+        dbname=os.environ.get("DATABASE_NAME", "polis-dev"),
+        user=os.environ.get("DATABASE_USER", "postgres"),
+        password=os.environ.get("DATABASE_PASSWORD", ""),
+        connect_timeout=timeout,
+    )
+
+
+def _json_safe(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        as_int = int(value)
+        return as_int if value == as_int else float(value)
+    if isinstance(value, (bytes, memoryview)):
+        return base64.b64encode(bytes(value)).decode("ascii")
+    if hasattr(value, "isoformat"):  # date/datetime columns
+        return value.isoformat()
+    return value
+
+
+def _snapshot_payload(columns: list, rows: list) -> dict:
+    return {
+        "columns": columns,
+        "rows": [[_json_safe(value) for value in row] for row in rows],
+    }
+
+
+def _write_snapshot(
+    store: DelphiStore,
+    job_id: str,
+    kind: str,
+    payload: dict,
+    force: Optional[str] = None,
+    extra: Optional[dict] = None,
+    context: Optional[dict] = None,
+) -> dict:
+    sha256 = hashlib.sha256(canonical_json_dumps(payload).encode("utf-8")).hexdigest()
+    encoded = encode_payload(payload, force=force)
+    fingerprint = {"sha256": sha256, "row_count": len(payload["rows"])}
+    if extra:
+        fingerprint.update(extra)
+    attributes = {**encoded.meta, "kind": kind, **fingerprint}
+    if context:
+        attributes.update(context)
+    store.put(
+        "run_inputs",
+        StoreItem(pk=job_id, sk=kind, attributes=attributes, blob=encoded.blob),
+    )
+    return fingerprint
+
+
+def capture_run_inputs(
+    store: DelphiStore,
+    job_id: str,
+    zid: int,
+    rid: Optional[int] = None,
+    *,
+    pg_url: Optional[str] = None,
+    batch_size: int = 50000,
+) -> dict:
+    """Snapshot all input kinds for a run; returns per-kind fingerprints.
+
+    ``zid`` and ``rid`` are stamped on every snapshot item's attributes for
+    provenance; selections are captured for the whole conversation (what
+    stage 2 reads), never filtered by ``rid``.
+    """
+    context = {"zid": zid, "rid": rid}
+    fingerprints: dict = {}
+    conn = _connect(pg_url)
+    try:
+        cursor = conn.cursor()
+        cursor.execute(VOTES_COUNT_SQL, (zid,))
+        total_votes = cursor.fetchone()[0]
+        cursor.close()
+
+        vote_rows: list = []
+        for offset in range(0, max(total_votes, 1), batch_size):
+            cursor = conn.cursor()
+            cursor.execute(VOTES_BATCH_SQL, (zid, batch_size, offset))
+            vote_rows.extend(cursor.fetchall())
+            cursor.close()
+        payload = _snapshot_payload(["created", "tid", "pid", "vote"], vote_rows)
+        max_created = max((row[0] for row in payload["rows"]), default=None)
+        fingerprints["votes"] = _write_snapshot(
+            store, job_id, "votes", payload,
+            force="json+zstd", extra={"max_created": max_created}, context=context,
+        )
+
+        for kind, query in _TABLE_QUERIES.items():
+            cursor = conn.cursor()
+            cursor.execute(query, (zid,))
+            columns = [description[0] for description in cursor.description]
+            rows = cursor.fetchall()
+            cursor.close()
+            fingerprints[kind] = _write_snapshot(
+                store, job_id, kind, _snapshot_payload(columns, rows), context=context
+            )
+    finally:
+        conn.close()
+    return fingerprints
+
+
+def load_run_input(store: DelphiStore, job_id: str, kind: str) -> Any:
+    """Decode a snapshot payload; the read side of the P6b --input-source seam."""
+    if kind not in SNAPSHOT_KINDS:
+        raise Invalid(f"unknown snapshot kind {kind!r}")
+    item = store.get("run_inputs", job_id, kind)
+    if item is None:
+        raise NotFound(f"run {job_id!r} has no {kind!r} snapshot")
+    return decode_payload(item.attributes, item.blob)
+
+
+def derive_votes_latest_unique(vote_rows: list) -> dict:
+    """Derive the latest vote per (pid, tid) from the snapshot stream: the
+    last occurrence in stream order wins.
+
+    Faithfulness caveat (documented, not hidden): PG's RULE-maintained
+    votes_latest_unique picks winners by INSERTION order, while the snapshot
+    stream is ``ORDER BY created`` with no tiebreaker (the votes table has no
+    serial/PK column, so none is possible without changing stage 1's read —
+    which golden invariance forbids). For two votes on the same (pid, tid)
+    with EQUAL ``created`` (same-millisecond double-submit), the RULE's
+    winner and this derivation's winner may differ. The derivation is always
+    deterministic with respect to the snapshot itself, which is what replay
+    requires; the tie divergence risk is pinned by a test."""
+    latest: dict = {}
+    for _created, tid, pid, vote in vote_rows:
+        latest[(pid, tid)] = vote
+    return latest

delphi/delphi_storage/purge.py

@@ -0,0 +1,19 @@
+"""Per-job purge (design §9): input snapshots copy votes/comments outside the
+source PG, so GDPR deletion needs a path that removes them. Deletes every
+generic-entity partition for a job (run_inputs, artifacts, topic_moderation,
+collective_statements).
+
+SCOPE (P6a): per-JOB only — the caller must know the job ids. The per-zid
+purge tool §9 calls for (enumerate all runs of a conversation, purge each)
+requires the run manifests, which land in P7; it follows there via
+list_runs(zid=...) + purge_job. Run-manifest rows themselves are removed by
+the retention machinery (P7+)."""
+
+from delphi_storage.interface import DelphiStore
+from delphi_storage.keys import GENERIC_ENTITIES
+
+
+def purge_job(store: DelphiStore, job_id: str) -> int:
+    """Delete every stored item for a job across all generic entities;
+    returns the number of logical items removed."""
+    return sum(store.delete_partition(entity, job_id) for entity in GENERIC_ENTITIES)

delphi/polismath/run_math_pipeline.py

@@ -18,6 +18,10 @@
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+# Stage 1's vote reads are shared with the input-snapshot capture
+# (delphi_storage/inputs.py) so the two can never diverge (design §4.4/P6).
+from delphi_storage.inputs import VOTES_BATCH_SQL, VOTES_COUNT_SQL
+
 
 def prepare_for_json(obj):
     import numpy as np
@@ -295,7 +299,7 @@ def main():
         logger.info(f"[{time.time() - start_time:.2f}s] Moderation applied")
 
         cursor = conn.cursor()
-        cursor.execute("SELECT COUNT(*) FROM votes WHERE zid = %s", (zid,))
+        cursor.execute(VOTES_COUNT_SQL, (zid,))
         total_votes = cursor.fetchone()[0]
         cursor.close()
         logger.info(f"[{time.time() - start_time:.2f}s] {total_votes} total votes")
@@ -317,10 +321,7 @@ def main():
             logger.info(f"[{time.time() - start_time:.2f}s] Processing votes {offset+1} to {end_idx} of {total_votes}")
 
             cursor = conn.cursor()
-            batch_query = """
-            SELECT v.created, v.tid, v.pid, v.vote FROM votes v WHERE v.zid = %s ORDER BY v.created LIMIT %s OFFSET %s
-            """
-            cursor.execute(batch_query, (zid, batch_size, offset))
+            cursor.execute(VOTES_BATCH_SQL, (zid, batch_size, offset))
             vote_batch = cursor.fetchall()
             cursor.close()
 

delphi/run_delphi.py

@@ -25,6 +25,16 @@ def show_usage():
     print("  --validate                Run extra validation checks")
     print("  --help                    Show this help message")
 
+def _capture_run_inputs(job_id, zid, rid):
+    """Snapshot inputs into the V2 store (module-level so tests can stub it)."""
+    from delphi_storage import get_store
+    from delphi_storage.inputs import capture_run_inputs
+
+    fingerprints = capture_run_inputs(get_store(), job_id, zid, rid=rid)
+    for kind, fingerprint in fingerprints.items():
+        print(f"{YELLOW}Snapshotted {kind}: {fingerprint}{NC}")
+
+
 def main():
     parser = argparse.ArgumentParser(description="Process a Polis conversation with the Delphi analytics pipeline.", add_help=False)
     parser.add_argument("--zid", required=True, help="The Polis conversation ID to process")
@@ -36,6 +46,9 @@ def main():
     parser.add_argument('--include_moderation', type=bool, default=False, help='Whether or not to include moderated comments in reports. If false, moderated comments will appear.')
     parser.add_argument('--exclude_comment_selections', type=bool, default=True, help='Whether to exclude comments with selection=-1 in report_comment_selections table.')
     parser.add_argument('--region', type=str, default='us-east-1', help='AWS region')
+    parser.add_argument('--snapshot-inputs', dest='snapshot_inputs', action='store_true',
+                        help='Snapshot all pipeline inputs into Delphi Storage V2 at job start '
+                             '(also enabled by DELPHI_SNAPSHOT_INPUTS=1; design §4.4/P6)')
     parser.add_argument('--job-id', dest='job_id', default=None,
                         help='Pipeline job id (auto local-<uuid4> when omitted; '
                              'threaded to every stage, see docs/STORAGE_V2_DESIGN.md §4.4)')
@@ -59,6 +72,19 @@ def main():
     # validate_arg is not used in the python script execution steps, but kept for parity with bash
     # validate_arg = "--validate" if args.validate else ""
 
+    snapshot_enabled = args.snapshot_inputs or os.environ.get(
+        "DELPHI_SNAPSHOT_INPUTS", ""
+    ).lower() in ("1", "true", "yes")
+    if snapshot_enabled:
+        print(f"{YELLOW}Snapshotting pipeline inputs for job {job_id}...{NC}")
+        try:
+            _capture_run_inputs(job_id, zid, rid)
+        except Exception as e:
+            # A run without recorded inputs defeats the point when enabled.
+            print(f"{RED}Input snapshot failed: {e}. Aborting pipeline.{NC}")
+            sys.exit(1)
+        print(f"{GREEN}Input snapshot complete.{NC}")
+
     # --- Reset all data before processing ---
     print(f"{YELLOW}Resetting all existing data for conversation {zid} before processing...{NC}")
     reset_command = [