Add GCP Cloud SQL restore drill (#44)

.github/workflows/ci.yml

@@ -41,6 +41,8 @@ jobs:
             lib/evaluate.py \
             lib/llm.py \
             lib/post_extract.py \
+            ops/check_gcp_infra_readiness.py \
+            ops/sqlite_to_postgres_dump.py \
             telegram/approvals.py \
             scripts/check_crabbox_ci_contract.py \
             scripts/check_llm_refinement_contract.py \
@@ -49,10 +51,18 @@ jobs:
             tests/test_agent_routing.py \
             tests/test_decision_engine_replay.py \
             tests/test_evaluate_agent_routing.py \
+            tests/test_gcp_cloudsql_restore_drill.py \
             tests/test_phase1b_end_to_end.py \
+            tests/test_sqlite_to_postgres_dump.py \
             tests/test_eval_parse.py \
             tests/test_contributor.py \
             tests/test_search.py
+      - name: Shell syntax
+        run: |
+          bash -n \
+            ops/backup_vps_sqlite_kb.sh \
+            ops/run_gcp_cloudsql_restore_drill.sh \
+            ops/run_sqlite_postgres_restore_canary.sh
 
   test:
     name: Unit tests

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

@@ -66,6 +66,36 @@ This is a local restore/parity proof, not GCP redundancy by itself. It is the
 preflight that should pass before the same generated import is applied through
 the approved Cloud SQL connector/VPC path.
 
+## Cloud SQL Restore Drill Runner
+
+Prepare the exact GCS import and Cloud SQL import operation without mutating GCP:
+
+```bash
+SQLITE_BACKUP=./outputs/gcp-infra-hardening-20260707/private-backups/teleo-pipeline-sqlite-<timestamp>.db.gz \
+  ops/run_gcp_cloudsql_restore_drill.sh
+```
+
+Execute it only from an authenticated operator environment that can write the
+versioned backup bucket and administer the standby Cloud SQL instance:
+
+```bash
+EXECUTE=1 \
+SQLITE_BACKUP=./outputs/gcp-infra-hardening-20260707/private-backups/teleo-pipeline-sqlite-<timestamp>.db.gz \
+  ops/run_gcp_cloudsql_restore_drill.sh
+```
+
+The runner:
+
+- regenerates the explicit PostgreSQL import script;
+- targets the shadow schema `teleo_restore` inside `teleo_kb`;
+- uploads the import script to `gs://teleo-501523-prod-backups/kb-dumps/cloudsql-restore-drills/...` when `EXECUTE=1`;
+- starts and waits for `gcloud sql import sql`;
+- writes `target-counts.sql` for the required trusted VPC/Cloud SQL connector query readback.
+
+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.
+
 ## Required Proof
 
 A successful restore or replication canary must retain:
@@ -92,14 +122,15 @@ Use this while the canonical DB remains SQLite on the VPS and we need a GCP rest
 1. Run `ops/backup_vps_sqlite_kb.sh`.
 2. Upload the resulting SQLite backup and Leo/KB tarball to a versioned GCS bucket such as `gs://teleo-501523-prod-backups/kb-dumps/`.
 3. Run the local SQLite-to-Postgres restore canary above and retain its proof.
-4. Convert or replay the SQLite data into Cloud SQL with the explicit generated migration/import script. Do not run blind string rewrites against the SQLite dump.
+4. Run `ops/run_gcp_cloudsql_restore_drill.sh` in dry-run mode to generate the GCS import plan.
-5. Install required extensions on Cloud SQL:
+5. Run `EXECUTE=1 ops/run_gcp_cloudsql_restore_drill.sh` from an authenticated operator environment to upload and import the generated SQL. Do not run blind string rewrites against the SQLite dump.
+6. Install required extensions on Cloud SQL:
 
 ```sql
 create extension if not exists vector;
 ```
 
-6. From a trusted VPC runtime or Cloud SQL connector path, run readbacks:
+7. From a trusted VPC runtime or Cloud SQL connector path, run readbacks:
 
 ```sql
 select current_database();
@@ -107,7 +138,7 @@ select extname, extversion from pg_extension where extname = 'vector';
 select schemaname, tablename from pg_tables where schemaname not in ('pg_catalog', 'information_schema') order by 1, 2 limit 50;
 ```
 
-7. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample.
+8. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample.
 
 ## Logical Replication Path
 

ops/check_gcp_infra_readiness.py

@@ -558,6 +558,47 @@ 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")
+    runbook = Path("docs/gcp-kb-cloudsql-restore-runbook.md")
+    missing = [str(path) for path in (script, runbook) if not path.exists()]
+    if missing:
+        return Check(
+            "gcp_cloudsql_restore_drill_contract",
+            "blocked",
+            f"missing={missing}",
+            required_tier="T2_restore_execution_contract",
+            current_tier="T1_foundation",
+        )
+    script_text = script.read_text()
+    runbook_text = runbook.read_text()
+    required = [
+        "gcloud storage cp",
+        "gcloud sql import sql",
+        "gcloud sql operations wait",
+        "EXECUTE",
+        "target-counts.sql",
+    ]
+    absent = [item for item in required if item not in script_text]
+    if "run_gcp_cloudsql_restore_drill.sh" not in runbook_text:
+        absent.append("runbook_reference")
+    if absent:
+        return Check(
+            "gcp_cloudsql_restore_drill_contract",
+            "fail",
+            f"missing_contract={absent}",
+            required_tier="T2_restore_execution_contract",
+            current_tier="T1_foundation",
+        )
+    return Check(
+        "gcp_cloudsql_restore_drill_contract",
+        "pass",
+        "script can prepare GCS import SQL, optionally upload/import via Cloud SQL Admin, and emits target-count readback SQL",
+        required_tier="T2_restore_execution_contract",
+        current_tier="T2_restore_execution_contract",
+    )
+
+
 def check_kb_restore_or_replication() -> Check:
     return Check(
         "kb_restore_or_replication",
@@ -581,6 +622,7 @@ def main() -> int:
         safe_check("cloud_sql_standby_target", check_cloud_sql_standby),
         safe_check("kb_source_restore_access", check_kb_source_restore_access),
         safe_check("local_sqlite_postgres_restore_canary", check_local_sqlite_postgres_restore_canary),
+        safe_check("gcp_cloudsql_restore_drill_contract", check_gcp_cloudsql_restore_drill_contract),
         safe_check("kb_restore_or_replication", check_kb_restore_or_replication),
     ]
     payload = {

ops/run_gcp_cloudsql_restore_drill.sh

@@ -0,0 +1,235 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Prepare, and optionally execute, a Cloud SQL shadow-restore drill from a Teleo
+# SQLite backup. Default mode is dry-run: it generates the Postgres import SQL
+# and proof manifest, but does not upload to GCS or mutate Cloud SQL.
+
+PROJECT_ID="${PROJECT_ID:-teleo-501523}"
+INSTANCE="${INSTANCE:-teleo-pgvector-standby}"
+DATABASE="${DATABASE:-teleo_kb}"
+REGION="${REGION:-europe-west6}"
+BUCKET="${BUCKET:-teleo-501523-prod-backups}"
+GCS_PREFIX="${GCS_PREFIX:-kb-dumps/cloudsql-restore-drills}"
+SCHEMA="${SCHEMA:-teleo_restore}"
+SQLITE_BACKUP="${SQLITE_BACKUP:-}"
+EXECUTE="${EXECUTE:-0}"
+OUTPUT_ROOT="${OUTPUT_ROOT:-./outputs/gcp-infra-hardening-20260707/private-cloudsql-restore-drills}"
+PROOF_ROOT="${PROOF_ROOT:-./outputs/gcp-infra-hardening-20260707/proofs}"
+TS="${TS:-$(date -u +%Y%m%dT%H%M%SZ)}"
+
+if [[ -z "$SQLITE_BACKUP" ]]; then
+  echo "SQLITE_BACKUP is required" >&2
+  exit 2
+fi
+
+if [[ ! -f "$SQLITE_BACKUP" ]]; then
+  echo "SQLite backup not found: $SQLITE_BACKUP" >&2
+  exit 2
+fi
+
+mkdir -p "$OUTPUT_ROOT" "$PROOF_ROOT"
+chmod 700 "$OUTPUT_ROOT"
+
+WORK_DIR="$OUTPUT_ROOT/gcp-cloudsql-restore-drill-${TS}"
+mkdir -p "$WORK_DIR"
+chmod 700 "$WORK_DIR"
+
+IMPORT_SQL="$WORK_DIR/teleo-pipeline-postgres-import.sql"
+SOURCE_SUMMARY="$WORK_DIR/source-summary.json"
+TARGET_COUNTS_SQL="$WORK_DIR/target-counts.sql"
+CONVERT_LOG="$WORK_DIR/sqlite-to-postgres-dump.stdout"
+OBJECT_NAME="${GCS_PREFIX%/}/${TS}/teleo-pipeline-postgres-import.sql"
+GCS_IMPORT_URI="gs://${BUCKET}/${OBJECT_NAME}"
+PROOF_JSON="$PROOF_ROOT/gcp-cloudsql-restore-drill-${TS}.json"
+IMPORT_START_JSON="$WORK_DIR/cloudsql-import-start.json"
+IMPORT_WAIT_JSON="$WORK_DIR/cloudsql-import-wait.json"
+GCS_OBJECT_JSON="$WORK_DIR/gcs-import-object.json"
+
+python3 ops/sqlite_to_postgres_dump.py \
+  --sqlite-backup "$SQLITE_BACKUP" \
+  --output-sql "$IMPORT_SQL" \
+  --summary-json "$SOURCE_SUMMARY" \
+  --schema "$SCHEMA" \
+  --target-counts-sql "$TARGET_COUNTS_SQL" \
+  > "$CONVERT_LOG"
+
+python3 - "$SOURCE_SUMMARY" "$SQLITE_BACKUP" "$IMPORT_SQL" "$TARGET_COUNTS_SQL" "$PROOF_JSON" "$PROJECT_ID" "$INSTANCE" "$DATABASE" "$REGION" "$GCS_IMPORT_URI" "$EXECUTE" <<'PY'
+import hashlib
+import json
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+summary_path = Path(sys.argv[1])
+sqlite_backup = Path(sys.argv[2])
+import_sql = Path(sys.argv[3])
+target_counts_sql = Path(sys.argv[4])
+proof_path = Path(sys.argv[5])
+project_id, instance, database, region, gcs_import_uri, execute = sys.argv[6:12]
+summary = json.loads(summary_path.read_text())
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+proof = {
+    "artifact": "gcp_cloudsql_restore_drill",
+    "generated_at_utc": datetime.now(UTC).isoformat(),
+    "mode": "execute" if execute == "1" else "dry_run",
+    "status": "prepared" if execute != "1" else "prepared_for_execute",
+    "project_id": project_id,
+    "region": region,
+    "cloudsql_instance": instance,
+    "database": database,
+    "postgres_schema": summary["schema"],
+    "source_sqlite_backup": str(sqlite_backup),
+    "source_sqlite_backup_sha256": sha256(sqlite_backup),
+    "source_integrity_check": summary["source_integrity_check"],
+    "source_table_count": summary["table_count"],
+    "source_total_rows": summary["source_total_rows"],
+    "gcs_import_uri": gcs_import_uri,
+    "local_private_artifacts": {
+        "postgres_import_sql": str(import_sql),
+        "source_summary_json": str(summary_path),
+        "target_counts_sql": str(target_counts_sql),
+    },
+    "planned_commands": [
+        f"gcloud storage cp {import_sql} {gcs_import_uri}",
+        f"gcloud sql import sql {instance} {gcs_import_uri} --project={project_id} --database={database} --quiet --format=json",
+        "run target-counts.sql from a trusted VPC/Cloud SQL connector path and compare source_total_rows/source_table_count",
+    ],
+    "not_proven_by_this_artifact": [
+        "GCS object generation until EXECUTE=1 succeeds",
+        "Cloud SQL import operation until EXECUTE=1 succeeds",
+        "Cloud SQL query readback until target-counts.sql is run from a trusted VPC/connector path",
+    ],
+}
+proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
+print(json.dumps({
+    "artifact": proof["artifact"],
+    "mode": proof["mode"],
+    "status": proof["status"],
+    "source_integrity_check": proof["source_integrity_check"],
+    "source_table_count": proof["source_table_count"],
+    "source_total_rows": proof["source_total_rows"],
+    "gcs_import_uri": proof["gcs_import_uri"],
+    "proof": str(proof_path),
+}, indent=2, sort_keys=True))
+PY
+
+if [[ "$EXECUTE" != "1" ]]; then
+  exit 0
+fi
+
+mark_execute_failure() {
+  local rc="$1"
+  local line="$2"
+  python3 - "$PROOF_JSON" "$rc" "$line" <<'PY'
+import json
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+proof_path = Path(sys.argv[1])
+rc = int(sys.argv[2])
+line = sys.argv[3]
+proof = json.loads(proof_path.read_text())
+proof["status"] = "execute_failed"
+proof["execute_failure"] = {
+    "exitcode": rc,
+    "line": line,
+    "recorded_at_utc": datetime.now(UTC).isoformat(),
+    "stderr_files": {
+        "gcs_upload": str(Path(proof["local_private_artifacts"]["postgres_import_sql"]).with_name("gcs-upload.stderr")),
+    },
+}
+proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
+PY
+}
+
+trap 'rc=$?; mark_execute_failure "$rc" "$LINENO"; exit "$rc"' ERR
+
+gcloud storage cp "$IMPORT_SQL" "$GCS_IMPORT_URI" \
+  --project="$PROJECT_ID" \
+  > "$WORK_DIR/gcs-upload.stdout" \
+  2> "$WORK_DIR/gcs-upload.stderr"
+
+gcloud storage objects describe "$GCS_IMPORT_URI" \
+  --project="$PROJECT_ID" \
+  --format=json \
+  > "$GCS_OBJECT_JSON"
+
+gcloud sql import sql "$INSTANCE" "$GCS_IMPORT_URI" \
+  --project="$PROJECT_ID" \
+  --database="$DATABASE" \
+  --quiet \
+  --format=json \
+  > "$IMPORT_START_JSON"
+
+python3 - "$PROOF_JSON" "$GCS_OBJECT_JSON" "$IMPORT_START_JSON" <<'PY'
+import json
+import sys
+from pathlib import Path
+
+proof_path = Path(sys.argv[1])
+object_path = Path(sys.argv[2])
+operation_path = Path(sys.argv[3])
+proof = json.loads(proof_path.read_text())
+gcs_object = json.loads(object_path.read_text())
+operation = json.loads(operation_path.read_text() or "{}")
+proof["status"] = "import_started"
+proof["gcs_object"] = {
+    "name": gcs_object.get("name"),
+    "bucket": gcs_object.get("bucket"),
+    "generation": gcs_object.get("generation"),
+    "size": gcs_object.get("size"),
+    "crc32c": gcs_object.get("crc32c"),
+}
+proof["cloudsql_import_operation"] = {
+    "name": operation.get("name"),
+    "status": operation.get("status"),
+    "operationType": operation.get("operationType"),
+    "targetId": operation.get("targetId"),
+}
+proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
+print(json.dumps(proof["cloudsql_import_operation"], indent=2, sort_keys=True))
+PY
+
+operation_name="$(python3 - "$IMPORT_START_JSON" <<'PY'
+import json, sys
+print(json.loads(open(sys.argv[1]).read()).get("name", ""))
+PY
+)"
+
+if [[ -n "$operation_name" ]]; then
+  gcloud sql operations wait "$operation_name" \
+    --project="$PROJECT_ID" \
+    --timeout=3600 \
+    --format=json \
+    > "$IMPORT_WAIT_JSON"
+
+  python3 - "$PROOF_JSON" "$IMPORT_WAIT_JSON" <<'PY'
+import json
+import sys
+from pathlib import Path
+
+proof_path = Path(sys.argv[1])
+wait_path = Path(sys.argv[2])
+proof = json.loads(proof_path.read_text())
+operation = json.loads(wait_path.read_text() or "{}")
+proof["cloudsql_import_wait"] = {
+    "name": operation.get("name"),
+    "status": operation.get("status"),
+    "operationType": operation.get("operationType"),
+    "targetId": operation.get("targetId"),
+    "error": operation.get("error"),
+}
+proof["status"] = "import_operation_done" if operation.get("status") == "DONE" and not operation.get("error") else "import_operation_failed"
+proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
+print(json.dumps({"status": proof["status"], "proof": str(proof_path)}, indent=2, sort_keys=True))
+PY
+fi

tests/test_gcp_cloudsql_restore_drill.py

@@ -0,0 +1,44 @@
+import json
+import os
+import sqlite3
+import subprocess
+
+
+def test_gcp_cloudsql_restore_drill_dry_run(tmp_path):
+    db_path = tmp_path / "source.db"
+    conn = sqlite3.connect(db_path)
+    conn.execute("create table response_audit (id integer primary key, query text)")
+    conn.execute("insert into response_audit (query) values (?)", ("hello",))
+    conn.commit()
+    conn.close()
+
+    proof_root = tmp_path / "proofs"
+    output_root = tmp_path / "private"
+    env = {
+        **os.environ,
+        "SQLITE_BACKUP": str(db_path),
+        "OUTPUT_ROOT": str(output_root),
+        "PROOF_ROOT": str(proof_root),
+        "TS": "20260707Tdryrun",
+        "PROJECT_ID": "teleo-501523",
+        "INSTANCE": "teleo-pgvector-standby",
+        "DATABASE": "teleo_kb",
+    }
+    result = subprocess.run(
+        ["bash", "ops/run_gcp_cloudsql_restore_drill.sh"],
+        check=True,
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+
+    payload = json.loads(result.stdout)
+    assert payload["mode"] == "dry_run"
+    assert payload["source_table_count"] == 1
+    assert payload["source_total_rows"] == 1
+
+    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 "gcloud sql import sql" in "\n".join(proof["planned_commands"])
+    assert (output_root / "gcp-cloudsql-restore-drill-20260707Tdryrun" / "target-counts.sql").exists()