Add SQLite to Postgres restore canary (#43)

docs/gcp-infra-hardening-20260707.md

@@ -39,6 +39,10 @@ This records the current GCP hardening contract for `teleo-501523`.
   - source DB: `/opt/teleo-eval/pipeline/pipeline.db`
   - source Leo/KB files: `workspaces/*/agents/leo` plus `agent-state`
   - excludes secrets and logs
+- Local SQLite-to-Postgres restore canary exists:
+  - scripts: `ops/sqlite_to_postgres_dump.py` and `ops/run_sqlite_postgres_restore_canary.sh`
+  - target: disposable local `postgres:16-alpine` shadow schema
+  - verifies source SQLite integrity and per-table source/target row-count parity
 
 ## How To Build
 
@@ -106,6 +110,11 @@ Database redundancy is not complete. The current project now has a GCP Cloud SQL
 
 Do not call the empty `teleo-pgvector-standby` instance redundancy by itself. It only counts after source data, restore/replication, access controls, and query readback are proven.
 
+The local restore canary narrows the remaining gap: the source SQLite backup can
+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.
+
 ## Communication Posture
 
 Current GCP communication hardening is limited to the surfaces already read back:

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

@@ -43,6 +43,29 @@ The script:
 
 This proves source exportability and local restore integrity. It does not prove GCP DB redundancy until a GCP restore/import/query canary also passes.
 
+## Local SQLite-To-Postgres Restore Canary
+
+Before importing into Cloud SQL, prove that the current SQLite backup can be
+converted and restored into PostgreSQL without row loss:
+
+```bash
+SQLITE_BACKUP=./outputs/gcp-infra-hardening-20260707/private-backups/teleo-pipeline-sqlite-<timestamp>.db.gz \
+  ops/run_sqlite_postgres_restore_canary.sh
+```
+
+The canary:
+
+- generates a PostgreSQL import script with `ops/sqlite_to_postgres_dump.py`;
+- recreates a shadow schema in a disposable `postgres:16-alpine` container;
+- imports all user tables from the SQLite backup;
+- compares source and target row counts for every table;
+- writes a proof JSON under `outputs/gcp-infra-hardening-20260707/proofs/`;
+- removes only its temporary canary container.
+
+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.
+
 ## Required Proof
 
 A successful restore or replication canary must retain:
@@ -68,14 +91,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. Convert or replay the SQLite data into Cloud SQL with an explicit migration script. Do not run blind string rewrites against the SQLite dump.
+3. Run the local SQLite-to-Postgres restore canary above and retain its proof.
-4. Install required extensions on Cloud SQL:
+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.
+5. Install required extensions on Cloud SQL:
 
 ```sql
 create extension if not exists vector;
 ```
 
-5. From a trusted VPC runtime or Cloud SQL connector path, run readbacks:
+6. From a trusted VPC runtime or Cloud SQL connector path, run readbacks:
 
 ```sql
 select current_database();
@@ -83,7 +107,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;
 ```
 
-6. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample.
+7. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample.
 
 ## Logical Replication Path
 
@@ -108,7 +132,10 @@ Retain only redacted connection metadata. Do not commit or paste credentials.
 
 ## Current Blocker
 
-As of this run, GCP has the standby target and a repeatable SQLite/KB export script, but it does not have an approved GCP upload/restore credential in the current local session, nor a retained Cloud SQL import/conversion/query proof. That is why the readiness checker still reports:
+As of this run, GCP has the standby target, a repeatable SQLite/KB export script,
+and a local SQLite-to-Postgres restore canary. It does not have an approved GCP
+upload/restore credential in the current local session, nor a retained Cloud SQL
+import/query proof. That is why the readiness checker still reports:
 
 - `kb_source_restore_access = blocked`
 - `kb_restore_or_replication = blocked`

ops/check_gcp_infra_readiness.py

@@ -9,11 +9,11 @@ accidentally round up to "ready" without live readbacks.
 from __future__ import annotations
 
 import json
+import os
 import subprocess
 from dataclasses import dataclass
 from pathlib import Path
 
-
 PROJECT = "teleo-501523"
 REGION = "europe-west6"
 ZONE = "europe-west6-a"
@@ -43,6 +43,7 @@ GITHUB_REPOSITORY = "living-ip/teleo-infrastructure"
 CLOUDSQL_STANDBY_INSTANCE = "teleo-pgvector-standby"
 CLOUDSQL_STANDBY_DATABASE = "teleo_kb"
 CLOUDSQL_STANDBY_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
+PROOF_ROOT = Path(os.environ.get("TELEO_GCP_PROOF_ROOT", "outputs/gcp-infra-hardening-20260707/proofs"))
 
 
 @dataclass
@@ -516,6 +517,47 @@ def check_kb_source_restore_access() -> Check:
     return Check("kb_source_restore_access", "pass", "candidate_secret_names=" + ",".join(source_candidates))
 
 
+def check_local_sqlite_postgres_restore_canary() -> Check:
+    candidates = sorted(PROOF_ROOT.glob("sqlite-postgres-restore-canary-*.json"))
+    if not candidates:
+        return Check(
+            "local_sqlite_postgres_restore_canary",
+            "blocked",
+            f"no sqlite-postgres restore canary proof found under {PROOF_ROOT}",
+            required_tier="T2_local_restore_canary",
+            current_tier="T1_foundation",
+        )
+
+    latest = candidates[-1]
+    proof = json.loads(latest.read_text())
+    mismatches = proof.get("mismatched_tables") or []
+    source_tables = proof.get("source_table_count")
+    target_tables = proof.get("target_table_count")
+    source_rows = proof.get("source_total_rows")
+    target_rows = proof.get("target_total_rows")
+    if (
+        proof.get("status") == "pass"
+        and proof.get("source_integrity_check") == "ok"
+        and not mismatches
+        and source_tables == target_tables
+        and source_rows == target_rows
+    ):
+        return Check(
+            "local_sqlite_postgres_restore_canary",
+            "pass",
+            f"latest={latest} tables={source_tables} rows={source_rows} postgres_image={proof.get('postgres_image')}",
+            required_tier="T2_local_restore_canary",
+            current_tier="T2_local_restore_canary",
+        )
+    return Check(
+        "local_sqlite_postgres_restore_canary",
+        "fail",
+        f"latest={latest} status={proof.get('status')} mismatches={len(mismatches)} source_tables={source_tables} target_tables={target_tables} source_rows={source_rows} target_rows={target_rows}",
+        required_tier="T2_local_restore_canary",
+        current_tier="T1_foundation",
+    )
+
+
 def check_kb_restore_or_replication() -> Check:
     return Check(
         "kb_restore_or_replication",
@@ -538,6 +580,7 @@ def main() -> int:
         safe_check("backup_buckets", check_backup_buckets),
         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("kb_restore_or_replication", check_kb_restore_or_replication),
     ]
     payload = {

ops/run_sqlite_postgres_restore_canary.sh

@@ -0,0 +1,139 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Restores a Teleo SQLite backup into a disposable local Postgres container and
+# compares source/target table row counts. It does not touch GCP or production.
+
+SQLITE_BACKUP="${SQLITE_BACKUP:-}"
+POSTGRES_IMAGE="${POSTGRES_IMAGE:-postgres:16-alpine}"
+OUTPUT_ROOT="${OUTPUT_ROOT:-./outputs/gcp-infra-hardening-20260707/private-restore-canaries}"
+PROOF_ROOT="${PROOF_ROOT:-./outputs/gcp-infra-hardening-20260707/proofs}"
+TS="${TS:-$(date -u +%Y%m%dT%H%M%SZ)}"
+SCHEMA="${SCHEMA:-teleo_restore}"
+CONTAINER_NAME="${CONTAINER_NAME:-teleo-pg-restore-canary-${TS}}"
+
+if [[ -z "$SQLITE_BACKUP" ]]; then
+  echo "SQLITE_BACKUP is required, for example SQLITE_BACKUP=outputs/.../teleo-pipeline-sqlite-...db.gz" >&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/sqlite-postgres-restore-canary-${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"
+TARGET_COUNTS_CSV="$WORK_DIR/target-counts.csv"
+IMPORT_LOG="$WORK_DIR/postgres-import.log"
+PROOF_JSON="$PROOF_ROOT/sqlite-postgres-restore-canary-${TS}.json"
+
+cleanup() {
+  if [[ "${KEEP_CONTAINER:-0}" != "1" ]]; then
+    docker rm -f "$CONTAINER_NAME" >/dev/null 2>&1 || true
+  fi
+}
+trap cleanup EXIT
+
+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"
+
+docker run --rm -d \
+  --name "$CONTAINER_NAME" \
+  -e POSTGRES_PASSWORD=postgres \
+  -e POSTGRES_DB=teleo_restore_canary \
+  "$POSTGRES_IMAGE" >/dev/null
+
+for _ in $(seq 1 60); do
+  if docker exec "$CONTAINER_NAME" pg_isready -U postgres -d teleo_restore_canary >/dev/null 2>&1; then
+    break
+  fi
+  sleep 1
+done
+
+docker exec "$CONTAINER_NAME" pg_isready -U postgres -d teleo_restore_canary >/dev/null
+
+docker exec -i "$CONTAINER_NAME" \
+  psql -v ON_ERROR_STOP=1 -U postgres -d teleo_restore_canary \
+  < "$IMPORT_SQL" > "$IMPORT_LOG"
+
+docker exec -i "$CONTAINER_NAME" \
+  psql -v ON_ERROR_STOP=1 -U postgres -d teleo_restore_canary \
+  < "$TARGET_COUNTS_SQL" > "$TARGET_COUNTS_CSV"
+
+python3 - "$SOURCE_SUMMARY" "$TARGET_COUNTS_CSV" "$PROOF_JSON" "$SQLITE_BACKUP" "$IMPORT_SQL" "$POSTGRES_IMAGE" <<'PY'
+import csv
+import hashlib
+import json
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+source_summary_path = Path(sys.argv[1])
+target_counts_path = Path(sys.argv[2])
+proof_path = Path(sys.argv[3])
+sqlite_backup = Path(sys.argv[4])
+import_sql = Path(sys.argv[5])
+postgres_image = sys.argv[6]
+
+source_summary = json.loads(source_summary_path.read_text())
+source_counts = {table["name"]: int(table["row_count"]) for table in source_summary["tables"]}
+with target_counts_path.open(newline="") as handle:
+    target_counts = {row["table_name"]: int(row["row_count"]) for row in csv.DictReader(handle)}
+
+all_tables = sorted(set(source_counts) | set(target_counts))
+mismatches = [
+    {
+        "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 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": "sqlite_postgres_restore_canary",
+    "generated_at_utc": datetime.now(UTC).isoformat(),
+    "status": "pass" if not mismatches and source_summary["source_integrity_check"] == "ok" else "fail",
+    "source_sqlite_backup": str(sqlite_backup),
+    "source_sqlite_backup_sha256": sha256(sqlite_backup),
+    "postgres_image": postgres_image,
+    "source_integrity_check": source_summary["source_integrity_check"],
+    "postgres_schema": source_summary["schema"],
+    "source_table_count": len(source_counts),
+    "target_table_count": len(target_counts),
+    "source_total_rows": sum(source_counts.values()),
+    "target_total_rows": sum(target_counts.values()),
+    "mismatched_tables": mismatches,
+    "conversion_notes": source_summary["conversion_notes"],
+    "conversion_stats": source_summary["conversion_stats"],
+    "private_artifacts": {
+        "source_summary_json": str(source_summary_path),
+        "postgres_import_sql": str(import_sql),
+        "target_counts_csv": str(target_counts_path),
+    },
+}
+proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
+print(json.dumps(proof, indent=2, sort_keys=True))
+sys.exit(0 if proof["status"] == "pass" else 1)
+PY

ops/sqlite_to_postgres_dump.py

@@ -0,0 +1,272 @@
+#!/usr/bin/env python3
+"""Generate a PostgreSQL restore script from a Teleo SQLite backup.
+
+This is a shadow-restore converter for restore drills while the canonical
+runtime database is still SQLite. It preserves tables and rows, infers safe
+Postgres column types from actual SQLite storage classes, and avoids blind
+string rewrites of `.dump` output.
+"""
+
+from __future__ import annotations
+
+import argparse
+import gzip
+import json
+import shutil
+import sqlite3
+import tempfile
+from contextlib import contextmanager
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+PG_NUMERIC_TYPES = {"bigint", "double precision"}
+
+
+def pg_quote_identifier(value: str) -> str:
+    return '"' + value.replace('"', '""') + '"'
+
+
+def sqlite_quote_identifier(value: str) -> str:
+    return '"' + value.replace('"', '""') + '"'
+
+
+def copy_escape(value: Any, pg_type: str, stats: dict[str, int]) -> str:
+    if value is None:
+        return r"\N"
+    if isinstance(value, bytes):
+        text = value.hex()
+        stats["blob_values_hex_encoded"] += 1
+    elif pg_type in PG_NUMERIC_TYPES:
+        text = str(value)
+    else:
+        text = str(value)
+
+    if "\x00" in text:
+        stats["nul_chars_replaced"] += text.count("\x00")
+        text = text.replace("\x00", r"\u0000")
+
+    return (
+        text.replace("\\", "\\\\")
+        .replace("\t", r"\t")
+        .replace("\n", r"\n")
+        .replace("\r", r"\r")
+    )
+
+
+def infer_pg_type(declared_type: str | None, storage_counts: dict[str, int]) -> tuple[str, str]:
+    storage_types = {name for name, count in storage_counts.items() if count and name != "null"}
+    declared = (declared_type or "").upper()
+
+    if not storage_types:
+        if "INT" in declared:
+            return "bigint", "all_null_declared_integer"
+        if any(token in declared for token in ("REAL", "FLOA", "DOUB")):
+            return "double precision", "all_null_declared_real"
+        return "text", "all_null_text_default"
+
+    if storage_types == {"integer"}:
+        return "bigint", "sqlite_integer_storage"
+    if storage_types <= {"integer", "real"}:
+        return "double precision", "sqlite_numeric_storage"
+    return "text", "sqlite_text_or_mixed_storage"
+
+
+def read_storage_counts(conn: sqlite3.Connection, table: str, column: str) -> dict[str, int]:
+    sql = (
+        f"select typeof({sqlite_quote_identifier(column)}) as storage_type, count(*) "
+        f"from {sqlite_quote_identifier(table)} group by storage_type"
+    )
+    return {str(row[0]): int(row[1]) for row in conn.execute(sql)}
+
+
+def user_tables(conn: sqlite3.Connection) -> list[dict[str, str]]:
+    rows = conn.execute(
+        """
+        select name, sql
+        from sqlite_master
+        where type = 'table'
+          and name not like 'sqlite_%'
+        order by name
+        """
+    ).fetchall()
+    return [{"name": str(row[0]), "sqlite_sql": row[1] or ""} for row in rows]
+
+
+@contextmanager
+def sqlite_database_path(path: Path):
+    if path.suffix != ".gz":
+        yield path
+        return
+
+    tmpdir = Path(tempfile.mkdtemp(prefix="teleo-sqlite-restore-"))
+    try:
+        db_path = tmpdir / "pipeline.db"
+        with gzip.open(path, "rb") as source, db_path.open("wb") as target:
+            shutil.copyfileobj(source, target)
+        yield db_path
+    finally:
+        shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+def build_table_metadata(conn: sqlite3.Connection, table: str) -> dict[str, Any]:
+    columns: list[dict[str, Any]] = []
+    for cid, name, declared_type, notnull, default_value, primary_key in conn.execute(
+        f"pragma table_info({sqlite_quote_identifier(table)})"
+    ):
+        storage_counts = read_storage_counts(conn, table, name)
+        pg_type, type_reason = infer_pg_type(declared_type, storage_counts)
+        columns.append(
+            {
+                "cid": cid,
+                "name": name,
+                "sqlite_declared_type": declared_type or "",
+                "sqlite_notnull": bool(notnull),
+                "sqlite_default": default_value,
+                "sqlite_primary_key_position": primary_key,
+                "postgres_type": pg_type,
+                "postgres_type_reason": type_reason,
+                "sqlite_storage_counts": storage_counts,
+            }
+        )
+    row_count = conn.execute(f"select count(*) from {sqlite_quote_identifier(table)}").fetchone()[0]
+    return {"name": table, "row_count": int(row_count), "columns": columns}
+
+
+def write_target_counts_sql(path: Path, schema: str, tables: list[dict[str, Any]]) -> None:
+    selects = []
+    for table in tables:
+        table_name = table["name"]
+        table_literal = table_name.replace("'", "''")
+        table_ref = f"{pg_quote_identifier(schema)}.{pg_quote_identifier(table_name)}"
+        selects.append(f"select '{table_literal}' as table_name, count(*)::bigint as row_count from {table_ref}")
+    body = "\nunion all\n".join(selects) or "select '' as table_name, 0::bigint as row_count where false"
+    path.write_text(
+        "copy (\n"
+        f"{body}\n"
+        "order by table_name\n"
+        ") to stdout with (format csv, header true);\n"
+    )
+
+
+def convert_sqlite_to_postgres(
+    sqlite_path: Path,
+    output_sql: Path,
+    summary_json: Path,
+    schema: str,
+    target_counts_sql: Path | None = None,
+) -> dict[str, Any]:
+    output_sql.parent.mkdir(parents=True, exist_ok=True)
+    summary_json.parent.mkdir(parents=True, exist_ok=True)
+    if target_counts_sql:
+        target_counts_sql.parent.mkdir(parents=True, exist_ok=True)
+
+    with sqlite_database_path(sqlite_path) as db_path:
+        conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+        conn.row_factory = sqlite3.Row
+        integrity = conn.execute("pragma integrity_check").fetchone()[0]
+        tables = []
+        table_names = [table["name"] for table in user_tables(conn)]
+        for table_name in table_names:
+            tables.append(build_table_metadata(conn, table_name))
+
+        stats = {"blob_values_hex_encoded": 0, "nul_chars_replaced": 0}
+        with output_sql.open("w", encoding="utf-8", newline="\n") as out:
+            out.write("\\set ON_ERROR_STOP on\n")
+            out.write("begin;\n")
+            out.write(f"drop schema if exists {pg_quote_identifier(schema)} cascade;\n")
+            out.write(f"create schema {pg_quote_identifier(schema)};\n")
+            out.write(f"set search_path to {pg_quote_identifier(schema)};\n")
+
+            for table in tables:
+                table_ref = f"{pg_quote_identifier(schema)}.{pg_quote_identifier(table['name'])}"
+                column_defs = [
+                    f"{pg_quote_identifier(column['name'])} {column['postgres_type']}"
+                    for column in table["columns"]
+                ]
+                out.write(f"create table {table_ref} (\n  " + ",\n  ".join(column_defs) + "\n);\n")
+
+                if not table["columns"]:
+                    continue
+
+                column_list = ", ".join(pg_quote_identifier(column["name"]) for column in table["columns"])
+                out.write(f"copy {table_ref} ({column_list}) from stdin with (format text, delimiter E'\\t', null '\\N');\n")
+
+                select_columns = ", ".join(sqlite_quote_identifier(column["name"]) for column in table["columns"])
+                for row in conn.execute(f"select {select_columns} from {sqlite_quote_identifier(table['name'])}"):
+                    fields = [
+                        copy_escape(row[column["name"]], column["postgres_type"], stats)
+                        for column in table["columns"]
+                    ]
+                    out.write("\t".join(fields))
+                    out.write("\n")
+                out.write("\\.\n")
+
+            out.write("commit;\n")
+            out.write("analyze;\n")
+
+        if target_counts_sql:
+            write_target_counts_sql(target_counts_sql, schema, tables)
+
+        summary = {
+            "artifact": "teleo_sqlite_to_postgres_dump",
+            "generated_at_utc": datetime.now(UTC).isoformat(),
+            "source_sqlite_path": str(sqlite_path),
+            "source_integrity_check": integrity,
+            "schema": schema,
+            "table_count": len(tables),
+            "source_total_rows": sum(table["row_count"] for table in tables),
+            "tables": tables,
+            "conversion_notes": {
+                "mode": "shadow_restore",
+                "constraints_and_indexes": "not_recreated",
+                "blob_storage": "hex_encoded_text_when_needed",
+                "postgres_types": "inferred_from_sqlite_storage_classes",
+            },
+            "conversion_stats": stats,
+            "output_sql": str(output_sql),
+            "target_counts_sql": str(target_counts_sql) if target_counts_sql else None,
+        }
+        summary_json.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n")
+        return summary
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--sqlite-backup", required=True, type=Path, help="SQLite .db or .db.gz backup")
+    parser.add_argument("--output-sql", required=True, type=Path, help="Postgres import SQL path")
+    parser.add_argument("--summary-json", required=True, type=Path, help="Non-secret conversion summary JSON")
+    parser.add_argument("--schema", default="teleo_restore", help="Postgres schema to recreate")
+    parser.add_argument("--target-counts-sql", type=Path, help="Optional psql query file for target row counts")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    summary = convert_sqlite_to_postgres(
+        sqlite_path=args.sqlite_backup,
+        output_sql=args.output_sql,
+        summary_json=args.summary_json,
+        schema=args.schema,
+        target_counts_sql=args.target_counts_sql,
+    )
+    print(
+        json.dumps(
+            {
+                "artifact": summary["artifact"],
+                "source_integrity_check": summary["source_integrity_check"],
+                "schema": summary["schema"],
+                "table_count": summary["table_count"],
+                "source_total_rows": summary["source_total_rows"],
+                "output_sql": summary["output_sql"],
+                "target_counts_sql": summary["target_counts_sql"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0 if summary["source_integrity_check"] == "ok" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_sqlite_to_postgres_dump.py

@@ -0,0 +1,46 @@
+import json
+import sqlite3
+
+from ops.sqlite_to_postgres_dump import convert_sqlite_to_postgres
+
+
+def test_sqlite_to_postgres_dump_escapes_and_counts(tmp_path):
+    db_path = tmp_path / "source.db"
+    conn = sqlite3.connect(db_path)
+    conn.execute("create table audit_log (id integer primary key, detail text, score real, payload blob)")
+    conn.execute(
+        "insert into audit_log (id, detail, score, payload) values (?, ?, ?, ?)",
+        (1, "line one\nline two\tbackslash\\literal \\N", 1.25, b"\x00\x01abc"),
+    )
+    conn.execute("insert into audit_log (id, detail, score, payload) values (?, ?, ?, ?)", (2, None, None, None))
+    conn.execute("create table mixed (value integer)")
+    conn.execute("insert into mixed (value) values (?)", ("not actually integer",))
+    conn.commit()
+    conn.close()
+
+    sql_path = tmp_path / "restore.sql"
+    summary_path = tmp_path / "summary.json"
+    target_counts_path = tmp_path / "counts.sql"
+
+    summary = convert_sqlite_to_postgres(
+        sqlite_path=db_path,
+        output_sql=sql_path,
+        summary_json=summary_path,
+        schema="teleo_restore",
+        target_counts_sql=target_counts_path,
+    )
+
+    sql = sql_path.read_text()
+    assert 'create schema "teleo_restore";' in sql
+    assert 'copy "teleo_restore"."audit_log"' in sql
+    assert r"line one\nline two\tbackslash\\literal \\N" in sql
+    assert "0001616263" in sql
+    assert summary["source_integrity_check"] == "ok"
+    assert summary["table_count"] == 2
+    assert summary["source_total_rows"] == 3
+
+    stored_summary = json.loads(summary_path.read_text())
+    mixed_column = stored_summary["tables"][1]["columns"][0]
+    assert mixed_column["name"] == "value"
+    assert mixed_column["postgres_type"] == "text"
+    assert target_counts_path.read_text().startswith("copy (")