Add Cloud SQL restore readback verifier (#49)

.github/workflows/ci.yml

@@ -45,6 +45,7 @@ jobs:
             ops/check_gcp_infra_readiness.py \
             ops/plan_gcp_iam_split.py \
             ops/sqlite_to_postgres_dump.py \
+            ops/verify_gcp_cloudsql_restore_readback.py \
             telegram/approvals.py \
             scripts/check_crabbox_ci_contract.py \
             scripts/check_llm_refinement_contract.py \
@@ -54,6 +55,7 @@ jobs:
             tests/test_decision_engine_replay.py \
             tests/test_evaluate_agent_routing.py \
             tests/test_gcp_cloudsql_restore_drill.py \
+            tests/test_gcp_cloudsql_restore_readback.py \
             tests/test_gcp_iam_split_apply.py \
             tests/test_gcp_iam_split_plan.py \
             tests/test_gcp_readiness_workflow.py \

docs/gcp-infra-hardening-20260707.md

@@ -137,6 +137,17 @@ be converted and restored into PostgreSQL with table/row-count parity, but the
 same import still needs to run against the GCP Cloud SQL standby through an
 approved GCP auth and network path.
 
+After Cloud SQL import, use the generated `target-counts.sql` and verify it with:
+
+```bash
+python3 ops/verify_gcp_cloudsql_restore_readback.py \
+  --drill-proof outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-drill-<timestamp>.json \
+  --target-counts-csv outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-target-counts-<timestamp>.csv \
+  --output outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-readback-verification-<timestamp>.json
+```
+
+The verifier must return `status = pass` before claiming row-count parity in GCP.
+
 ## IAM Split Plan
 
 Do not make `sa-artifact-builder@teleo-501523.iam.gserviceaccount.com` the broad

docs/gcp-kb-cloudsql-restore-runbook.md

@@ -96,6 +96,28 @@ The import operation alone is still not the final proof. The final proof needs
 `target-counts.sql` run against `teleo-pgvector-standby` and compared to the
 source counts in the drill proof.
 
+After the import operation is `DONE`, run the generated count query from a
+trusted VPC runtime or Cloud SQL connector path and retain CSV output:
+
+```bash
+psql "$TELEO_CLOUDSQL_DATABASE_URL" \
+  --csv \
+  -f outputs/gcp-infra-hardening-20260707/private-cloudsql-restore-drills/gcp-cloudsql-restore-drill-<timestamp>/target-counts.sql \
+  > outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-target-counts-<timestamp>.csv
+```
+
+Then compare the Cloud SQL readback to the source proof:
+
+```bash
+python3 ops/verify_gcp_cloudsql_restore_readback.py \
+  --drill-proof outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-drill-<timestamp>.json \
+  --target-counts-csv outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-target-counts-<timestamp>.csv \
+  --output outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-readback-verification-<timestamp>.json
+```
+
+Only a `status = pass` verifier output is enough for row-count parity. It still
+does not prove application cutover or continuous replication.
+
 ## Required Proof
 
 A successful restore or replication canary must retain:
@@ -139,6 +161,7 @@ select schemaname, tablename from pg_tables where schemaname not in ('pg_catalog
 ```
 
 8. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample.
+9. Run `ops/verify_gcp_cloudsql_restore_readback.py` and retain a passing parity proof.
 
 ## Logical Replication Path
 

ops/check_gcp_infra_readiness.py

@@ -600,8 +600,9 @@ def check_local_sqlite_postgres_restore_canary() -> Check:
 
 def check_gcp_cloudsql_restore_drill_contract() -> Check:
     script = Path("ops/run_gcp_cloudsql_restore_drill.sh")
+    verifier = Path("ops/verify_gcp_cloudsql_restore_readback.py")
     runbook = Path("docs/gcp-kb-cloudsql-restore-runbook.md")
-    missing = [str(path) for path in (script, runbook) if not path.exists()]
+    missing = [str(path) for path in (script, verifier, runbook) if not path.exists()]
     if missing:
         return Check(
             "gcp_cloudsql_restore_drill_contract",
@@ -618,10 +619,17 @@ def check_gcp_cloudsql_restore_drill_contract() -> Check:
         "gcloud sql operations wait",
         "EXECUTE",
         "target-counts.sql",
+        "source_table_counts",
     ]
     absent = [item for item in required if item not in script_text]
+    verifier_text = verifier.read_text() if verifier.exists() else ""
+    for item in ("target_counts_csv", "mismatched_tables", "source_total_rows", "target_total_rows"):
+        if item not in verifier_text:
+            absent.append(f"verifier:{item}")
     if "run_gcp_cloudsql_restore_drill.sh" not in runbook_text:
         absent.append("runbook_reference")
+    if "verify_gcp_cloudsql_restore_readback.py" not in runbook_text:
+        absent.append("readback_verifier_runbook_reference")
     if absent:
         return Check(
             "gcp_cloudsql_restore_drill_contract",

ops/run_gcp_cloudsql_restore_drill.sh

@@ -91,6 +91,7 @@ proof = {
     "source_integrity_check": summary["source_integrity_check"],
     "source_table_count": summary["table_count"],
     "source_total_rows": summary["source_total_rows"],
+    "source_table_counts": {table["name"]: table["row_count"] for table in summary["tables"]},
     "gcs_import_uri": gcs_import_uri,
     "local_private_artifacts": {
         "postgres_import_sql": str(import_sql),

ops/verify_gcp_cloudsql_restore_readback.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Verify Cloud SQL restore readback against a Teleo restore drill proof.
+
+This script is intentionally post-import and non-mutating. Run the
+`target-counts.sql` generated by `ops/run_gcp_cloudsql_restore_drill.sh` against
+Cloud SQL with CSV output, then pass that CSV here. The verifier compares table
+counts and total rows with the source counts retained in the drill proof.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+def load_json(path: Path) -> dict[str, object]:
+    return json.loads(path.read_text())
+
+
+def read_target_counts(path: Path) -> dict[str, int]:
+    rows: dict[str, int] = {}
+    with path.open(newline="") as handle:
+        reader = csv.DictReader(handle)
+        required = {"table_name", "row_count"}
+        missing = required - set(reader.fieldnames or [])
+        if missing:
+            raise ValueError(f"target counts CSV missing columns: {sorted(missing)}")
+        for row in reader:
+            table = (row.get("table_name") or "").strip()
+            if not table:
+                continue
+            rows[table] = int(row.get("row_count") or "0")
+    return rows
+
+
+def source_counts_from_drill(proof: dict[str, object]) -> dict[str, int] | None:
+    source_counts = proof.get("source_table_counts")
+    if isinstance(source_counts, dict):
+        return {str(table): int(count) for table, count in source_counts.items()}
+    return None
+
+
+def compare_counts(source_counts: dict[str, int] | None, target_counts: dict[str, int]) -> list[dict[str, object]]:
+    if source_counts is None:
+        return []
+    all_tables = sorted(set(source_counts) | set(target_counts))
+    return [
+        {
+            "table": table,
+            "source_rows": source_counts.get(table),
+            "target_rows": target_counts.get(table),
+        }
+        for table in all_tables
+        if source_counts.get(table) != target_counts.get(table)
+    ]
+
+
+def build_payload(drill_proof_path: Path, target_counts_csv: Path) -> dict[str, object]:
+    proof = load_json(drill_proof_path)
+    target_counts = read_target_counts(target_counts_csv)
+    source_counts = source_counts_from_drill(proof)
+    target_table_count = len(target_counts)
+    target_total_rows = sum(target_counts.values())
+    source_table_count = int(proof.get("source_table_count") or 0)
+    source_total_rows = int(proof.get("source_total_rows") or 0)
+    mismatches = compare_counts(source_counts, target_counts)
+
+    problems = []
+    if proof.get("status") not in {"import_operation_done", "readback_verified"}:
+        problems.append(f"drill_status={proof.get('status')}")
+    if source_table_count != target_table_count:
+        problems.append(f"table_count source={source_table_count} target={target_table_count}")
+    if source_total_rows != target_total_rows:
+        problems.append(f"total_rows source={source_total_rows} target={target_total_rows}")
+    if mismatches:
+        problems.append(f"mismatched_tables={len(mismatches)}")
+
+    return {
+        "artifact": "gcp_cloudsql_restore_readback_verification",
+        "generated_at_utc": datetime.now(UTC).isoformat(),
+        "drill_proof": str(drill_proof_path),
+        "target_counts_csv": str(target_counts_csv),
+        "project_id": proof.get("project_id"),
+        "cloudsql_instance": proof.get("cloudsql_instance"),
+        "database": proof.get("database"),
+        "postgres_schema": proof.get("postgres_schema"),
+        "source_sqlite_backup_sha256": proof.get("source_sqlite_backup_sha256"),
+        "source_table_count": source_table_count,
+        "target_table_count": target_table_count,
+        "source_total_rows": source_total_rows,
+        "target_total_rows": target_total_rows,
+        "mismatched_tables": mismatches,
+        "status": "pass" if not problems else "fail",
+        "problems": problems,
+        "not_proven_by_this_artifact": [
+            "application cutover to Cloud SQL",
+            "ongoing replication after the readback timestamp",
+            "semantic query correctness beyond table row-count parity",
+        ],
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--drill-proof", required=True, type=Path)
+    parser.add_argument("--target-counts-csv", required=True, type=Path)
+    parser.add_argument(
+        "--output",
+        default="outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-readback-verification.json",
+        type=Path,
+    )
+    args = parser.parse_args()
+
+    payload = build_payload(args.drill_proof, args.target_counts_csv)
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0 if payload["status"] == "pass" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_gcp_cloudsql_restore_drill.py

@@ -40,5 +40,6 @@ def test_gcp_cloudsql_restore_drill_dry_run(tmp_path):
     proof = json.loads((proof_root / "gcp-cloudsql-restore-drill-20260707Tdryrun.json").read_text())
     assert proof["status"] == "prepared"
     assert proof["gcs_import_uri"].startswith("gs://teleo-501523-prod-backups/")
+    assert proof["source_table_counts"] == {"response_audit": 1}
     assert "gcloud sql import sql" in "\n".join(proof["planned_commands"])
     assert (output_root / "gcp-cloudsql-restore-drill-20260707Tdryrun" / "target-counts.sql").exists()

tests/test_gcp_cloudsql_restore_readback.py

@@ -0,0 +1,101 @@
+import json
+import subprocess
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def write_drill_proof(path: Path, *, status: str = "import_operation_done") -> None:
+    path.write_text(
+        json.dumps(
+            {
+                "status": status,
+                "project_id": "teleo-501523",
+                "cloudsql_instance": "teleo-pgvector-standby",
+                "database": "teleo_kb",
+                "postgres_schema": "teleo_restore",
+                "source_sqlite_backup_sha256": "abc123",
+                "source_table_count": 2,
+                "source_total_rows": 4,
+                "source_table_counts": {
+                    "claims": 3,
+                    "evidence": 1,
+                },
+            }
+        )
+    )
+
+
+def run_verifier(drill_proof: Path, counts_csv: Path, output: Path) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [
+            "python3",
+            "ops/verify_gcp_cloudsql_restore_readback.py",
+            "--drill-proof",
+            str(drill_proof),
+            "--target-counts-csv",
+            str(counts_csv),
+            "--output",
+            str(output),
+        ],
+        cwd=REPO_ROOT,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def test_cloudsql_restore_readback_passes_matching_counts(tmp_path: Path) -> None:
+    proof = tmp_path / "drill.json"
+    counts = tmp_path / "target-counts.csv"
+    output = tmp_path / "verification.json"
+    write_drill_proof(proof)
+    counts.write_text("table_name,row_count\nclaims,3\nevidence,1\n")
+
+    completed = run_verifier(proof, counts, output)
+
+    assert completed.returncode == 0, completed.stderr
+    payload = json.loads(output.read_text())
+    assert payload["status"] == "pass"
+    assert payload["source_table_count"] == 2
+    assert payload["target_table_count"] == 2
+    assert payload["source_total_rows"] == 4
+    assert payload["target_total_rows"] == 4
+    assert payload["mismatched_tables"] == []
+
+
+def test_cloudsql_restore_readback_fails_mismatched_counts(tmp_path: Path) -> None:
+    proof = tmp_path / "drill.json"
+    counts = tmp_path / "target-counts.csv"
+    output = tmp_path / "verification.json"
+    write_drill_proof(proof)
+    counts.write_text("table_name,row_count\nclaims,2\nevidence,1\n")
+
+    completed = run_verifier(proof, counts, output)
+
+    assert completed.returncode == 1
+    payload = json.loads(output.read_text())
+    assert payload["status"] == "fail"
+    assert "total_rows source=4 target=3" in payload["problems"]
+    assert payload["mismatched_tables"] == [
+        {
+            "source_rows": 3,
+            "table": "claims",
+            "target_rows": 2,
+        }
+    ]
+
+
+def test_cloudsql_restore_readback_requires_completed_import(tmp_path: Path) -> None:
+    proof = tmp_path / "drill.json"
+    counts = tmp_path / "target-counts.csv"
+    output = tmp_path / "verification.json"
+    write_drill_proof(proof, status="prepared")
+    counts.write_text("table_name,row_count\nclaims,3\nevidence,1\n")
+
+    completed = run_verifier(proof, counts, output)
+
+    assert completed.returncode == 1
+    payload = json.loads(output.read_text())
+    assert payload["status"] == "fail"
+    assert "drill_status=prepared" in payload["problems"]