diff --git a/.agents/skills/teleo-gcp-parity-ops/SKILL.md b/.agents/skills/teleo-gcp-parity-ops/SKILL.md
index 98ae607..490123e 100644
--- a/.agents/skills/teleo-gcp-parity-ops/SKILL.md
+++ b/.agents/skills/teleo-gcp-parity-ops/SKILL.md
@@ -58,6 +58,26 @@ Track each independently:
 4. no-post `DC-01` through `DC-06` replies and score;
 5. cleanup and no-mutation readback.
 
+## Canonical Restore Rule
+
+The canonical Leo database is VPS Docker Postgres `teleo-pg/teleo`. The
+SQLite `pipeline.db` to `teleo_kb.teleo_restore` drill is a separate legacy
+pipeline-memory proof and never satisfies canonical Leo parity.
+
+Use:
+
+- `ops/capture_vps_canonical_postgres_snapshot.py` for one exported read-only
+  snapshot shared by `pg_dump` and the source manifest;
+- `ops/postgres_parity_manifest.sql` for complete row/catalog/role/performance
+  readback;
+- `ops/verify_postgres_parity_manifest.py --scope gcp_staging` for exact parity
+  plus a staging-compute private-connectivity receipt.
+
+The 2026-07-11 local preflight restored all `39` canonical tables and `52,164`
+rows with zero row-hash, catalog-hash, role, or performance mismatches. That is
+local restore proof only. GCP import, private connectivity, and Cory replay are
+still unproven.
+
 ## Cloud SQL Read-Only Query
 
 Only after Studio is already authenticated with an existing principal:
@@ -80,17 +100,23 @@ skill.
 
 ## Exact Current Gate
 
-Database parity is unproven because Studio requires a database credential and
-did not expose an enabled existing IAM principal. The clear CTA is:
+Database parity is unproven because the privileged gcloud/Console account now
+requires password reauthentication, while both planned WIF identities
+`sa-teleo-readiness` and `sa-teleo-restore-drill` are absent. The only proven
+WIF identity is intentionally Artifact Registry-only. The clear CTA is:
 
 ```text
-Authenticate Cloud SQL Studio with an already-existing authorized database
-principal without sharing the credential, then tell Codex: GCP Studio authenticated.
+Run gcloud auth login billy@livingip.xyz --force, enter the Google password
+yourself, and tell Codex: GCP CLI reauthenticated.
 ```
 
-Otherwise route IAM/database-user enablement to a separately authorized write
-window. Once authenticated, run only the read-only transaction above, retain
-sanitized results, and close the agent-created tab.
+Otherwise route the reviewed IAM split to a separately authorized admin window.
+Once authenticated, create only a generated clone database, import the retained
+canonical dump, run full parity/private-connectivity/Cory replay, and clean up.
+
+For the narrower read-only Studio route, an operator may instead authenticate
+an already-existing database principal and report `GCP Studio authenticated`;
+that permits SQL readback but does not authorize the clone/import replay.
 
 ## Claim Ceiling
 
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 993d505..deaba67 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -42,6 +42,7 @@ jobs:
             lib/llm.py \
             lib/post_extract.py \
             ops/apply_gcp_iam_split.py \
+            ops/capture_vps_canonical_postgres_snapshot.py \
             ops/check_gcp_infra_readiness.py \
             ops/run_gcp_infra_execute_canary.py \
             ops/apply_gcp_runtime_baseline.py \
@@ -50,6 +51,7 @@ jobs:
             ops/redact_sqlite_postgres_restore_canary.py \
             ops/sqlite_to_postgres_dump.py \
             ops/verify_gcp_cloudsql_restore_readback.py \
+            ops/verify_postgres_parity_manifest.py \
             telegram/approvals.py \
             scripts/check_crabbox_ci_contract.py \
             scripts/check_llm_refinement_contract.py \
@@ -59,6 +61,7 @@ jobs:
             tests/test_decision_engine_replay.py \
             tests/test_evaluate_agent_routing.py \
             tests/test_gcp_artifact_workflow.py \
+            tests/test_capture_vps_canonical_postgres_snapshot.py \
             tests/test_gcp_infra_execute_canary.py \
             tests/test_gcp_infra_readiness_checker.py \
             tests/test_gcp_runtime_baseline_apply.py \
@@ -68,6 +71,7 @@ jobs:
             tests/test_gcp_iam_split_apply.py \
             tests/test_gcp_iam_split_plan.py \
             tests/test_gcp_readiness_workflow.py \
+            tests/test_verify_postgres_parity_manifest.py \
             tests/test_phase1b_end_to_end.py \
             tests/test_sqlite_to_postgres_dump.py \
             tests/test_sqlite_postgres_restore_canary_capsule.py \
diff --git a/docs/gcp-kb-cloudsql-restore-runbook.md b/docs/gcp-kb-cloudsql-restore-runbook.md
index 11be44c..0a1ad09 100644
--- a/docs/gcp-kb-cloudsql-restore-runbook.md
+++ b/docs/gcp-kb-cloudsql-restore-runbook.md
@@ -2,16 +2,35 @@
 
 This runbook is for proving Living IP KB/database redundancy on GCP.
 
-Current source reality:
+## Two Different Database Surfaces
 
-- canonical runtime DB: `/opt/teleo-eval/pipeline/pipeline.db`
+Do not call the SQLite shadow restore a canonical Leo database copy.
+
+Canonical Leo knowledge is currently:
+
+- host: VPS `77.42.65.182`;
+- container: `teleo-pg`;
+- engine/database: PostgreSQL 16, database `teleo`;
+- canonical schemas: `public` and `kb_stage`;
+- high-signal rows: claims, sources, claim evidence, claim edges, reasoning
+  tools, and review-gated proposals.
+
+The older pipeline/evaluation database is a separate surface:
+
+- pipeline runtime DB: `/opt/teleo-eval/pipeline/pipeline.db`
 - engine: SQLite WAL
-- canonical Leo files:
+- related Leo files:
   - `/opt/teleo-eval/workspaces/main/agents/leo`
   - `/opt/teleo-eval/workspaces/research-leo/agents/leo`
   - `/opt/teleo-eval/agent-state`
 
-The standby target already exists:
+`ops/run_gcp_cloudsql_restore_drill.sh` remains a legacy SQLite-to-Postgres
+shadow-schema drill. It reconstructs `teleo_restore` inside `teleo_kb`; it does
+not preserve the canonical Postgres schema, constraints, indexes, functions,
+roles, or row hashes.
+
+The last authenticated control-plane readback on 2026-07-10 reported this
+candidate GCP target; refresh it before any mutation:
 
 - project: `teleo-501523`
 - instance: `teleo-pgvector-standby`
@@ -23,7 +42,54 @@ The standby target already exists:
 
 Do not call this redundancy complete until source data has been restored or replicated and queried from GCP.
 
-## Source Backup Canary
+## Canonical Postgres Snapshot And Parity
+
+Capture a custom-format dump and a full JSONL manifest from the same exported,
+read-only PostgreSQL snapshot:
+
+```bash
+python3 ops/capture_vps_canonical_postgres_snapshot.py \
+  --execute \
+  --ssh-target root@77.42.65.182 \
+  --ssh-key ~/.ssh/livingip_hetzner_20260604_ed25519 \
+  --run-id canonical-<timestamp> \
+  --output-dir <private-output-dir>
+```
+
+The capture fails closed if the source service changes while it runs. It
+retains a private custom dump, dump SHA-256, object TOC, catalog/data manifest,
+and before/after service state. It never restarts Leo or writes to the source
+database.
+
+Run `ops/postgres_parity_manifest.sql` against the isolated restored target,
+then compare source and target:
+
+```bash
+python3 ops/verify_postgres_parity_manifest.py \
+  --source <private-output-dir>/source-manifest.jsonl \
+  --target <private-output-dir>/target-manifest.jsonl \
+  --scope gcp_staging \
+  --connectivity-proof <private-output-dir>/gcp-private-connectivity.json \
+  --output <private-output-dir>/gcp-parity.json
+```
+
+The verifier checks all table row counts and collation-independent row hashes,
+plus schemas, columns/defaults, constraints, indexes, sequences, views,
+functions, triggers, enum/domain types, policies, required extensions,
+password-free application-role attributes, and bounded query timings. In GCP
+scope it also requires a receipt proving a staging compute source, a private
+server address, TLS, and public-IP-disabled instance metadata.
+
+Use a generated target database such as `teleo_clone_<run_id>`. Never import a
+drill into `teleo`, `teleo_kb`, or `teleo_canonical`. Database isolation does
+not isolate cluster-global roles or extensions, so verify those separately and
+do not run the Docker-only gate bootstrap against the shared Cloud SQL instance.
+
+After the parity verifier passes, run the no-send Cory composition replay from
+staging compute against that generated database. Only then delete the generated
+database and uploaded import object and retain cleanup readback.
+
+## Legacy SQLite Source Backup Canary
 
 Create a consistent source backup without stopping the VPS service:
 
@@ -43,7 +109,7 @@ 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
+## Legacy 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:
@@ -91,7 +157,7 @@ gh workflow run gcp-readiness.yml \
   -f restore_canary_capsule_b64="${CAPSULE_B64}"
 ```
 
-## Cloud SQL Restore Drill Runner
+## Legacy SQLite Cloud SQL Restore Drill Runner
 
 Prepare the exact GCS import and Cloud SQL import operation without mutating GCP:
 
@@ -211,14 +277,28 @@ Retain only redacted connection metadata. Do not commit or paste credentials.
 
 ## Current Blocker
 
-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:
+As of 2026-07-11, the canonical Postgres exported-snapshot capture and isolated
+local restore parity pass. Live GCP restore and staging replay do not.
+
+- GitHub WIF works for `sa-artifact-builder`, but that identity is intentionally
+  limited to Artifact Registry and cannot inspect or mutate Cloud SQL/Compute.
+- The configured `sa-teleo-readiness` and `sa-teleo-restore-drill` identities
+  return IAM 404 and do not exist.
+- The local privileged `billy@livingip.xyz` gcloud session requires password
+  reauthentication. No password was entered or inspected.
+- Direct VM SSH is closed to the current egress `/32`; IAP requires the same
+  privileged GCP authentication.
+
+That is why the readiness checker still reports:
 
 - `kb_source_restore_access = blocked`
 - `kb_restore_or_replication = blocked`
 
-The next real canary is:
+The immediate operator CTA is to complete
+`gcloud auth login billy@livingip.xyz --force` locally without sharing the
+password, or apply the reviewed IAM split with an authorized GCP administrator.
+The next non-user action is:
 
-source SQLite/KB backup -> upload to versioned GCS bucket -> convert/import into `teleo_kb` -> install/verify `vector` if needed -> run representative KB queries on GCP -> retain proof.
+canonical `teleo` snapshot -> generated Cloud SQL database -> full parity and
+private-connectivity verifier -> no-send Cory composition replay from staging
+compute -> delete the generated database/object -> retain cleanup proof.
diff --git a/docs/reports/leo-working-state-20260709/current-truth-index.md b/docs/reports/leo-working-state-20260709/current-truth-index.md
index 62c5629..f195e4f 100644
--- a/docs/reports/leo-working-state-20260709/current-truth-index.md
+++ b/docs/reports/leo-working-state-20260709/current-truth-index.md
@@ -123,6 +123,8 @@ Use this file before making status claims. Prefer fresh VPS/GCP readbacks when c
 - Current GCP parity probe JSON: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.json`
 - Current GCP Cloud SQL live inventory: `docs/reports/leo-working-state-20260709/gcp-cloud-sql-t3-live-readonly-current.md`
 - Current GCP Cloud SQL live inventory JSON: `docs/reports/leo-working-state-20260709/gcp-cloud-sql-t3-live-readonly-current.json`
+- Canonical Postgres exported-snapshot/local-restore preflight: `docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.md`
+- Canonical Postgres exported-snapshot/local-restore preflight JSON: `docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.json`
 - Current GCP Cloud SQL Studio parity gate: `docs/reports/leo-working-state-20260709/gcp-cloudsql-studio-parity-gate-current.md`
 - Current GCP Cloud SQL Studio parity gate JSON: `docs/reports/leo-working-state-20260709/gcp-cloudsql-studio-parity-gate-current.json`
 - Current GCP Leo runtime observability: `docs/reports/leo-working-state-20260709/gcp-leo-runtime-observability-current.md`
diff --git a/docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.json b/docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.json
new file mode 100644
index 0000000..b3e0b60
--- /dev/null
+++ b/docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.json
@@ -0,0 +1,86 @@
+{
+  "schema": "teleo.gcpCanonicalPostgresRestorePreflight.v1",
+  "captured_at_utc": "2026-07-11T07:27:16Z",
+  "status": "local_restore_preflight_pass_gcp_unproven",
+  "source": {
+    "host": "77.42.65.182",
+    "container": "teleo-pg",
+    "database": "teleo",
+    "snapshot_mode": "one exported repeatable-read read-only snapshot shared by pg_dump and manifest",
+    "dump_bytes": 13462116,
+    "dump_sha256": "055e9bcd8922dbbd8a600be627c5a8f2d4a1329b383c3d9f5c28f325f6db5756",
+    "toc_entries": 294,
+    "table_data_objects": 39,
+    "constraint_objects": 113,
+    "index_objects": 75,
+    "table_count": 39,
+    "total_rows": 52164,
+    "application_roles": [
+      {
+        "name": "kb_apply",
+        "can_login": true,
+        "superuser": false,
+        "create_db": false,
+        "create_role": false,
+        "inherit": true,
+        "replication": false,
+        "bypass_rls": false
+      }
+    ]
+  },
+  "local_restore": {
+    "postgres_major": 16,
+    "network_mode": "none",
+    "published_ports": false,
+    "status": "pass",
+    "source_table_count": 39,
+    "target_table_count": 39,
+    "source_total_rows": 52164,
+    "target_total_rows": 52164,
+    "table_content_mismatches": 0,
+    "application_role_mismatches": 0,
+    "performance_problems": 0,
+    "structural_hash_mismatches": 0,
+    "verified_catalog_dimensions": [
+      "schemas",
+      "columns_and_defaults",
+      "constraints",
+      "indexes",
+      "sequences",
+      "views",
+      "functions",
+      "triggers",
+      "types",
+      "policies",
+      "extensions"
+    ]
+  },
+  "production_safety": {
+    "source_transaction_read_only": true,
+    "service_main_pid": 2999690,
+    "service_restarts": 0,
+    "service_state_unchanged": true,
+    "production_db_mutated": false,
+    "telegram_send_attempted": false
+  },
+  "cleanup": {
+    "local_disposable_containers_remaining": 0,
+    "local_anonymous_volumes_removed": true,
+    "remote_run_directory_removed": true,
+    "production_container_temp_files_removed": true
+  },
+  "gcp": {
+    "canonical_import_executed": false,
+    "private_connectivity_verified": false,
+    "staging_compute_replay_executed": false,
+    "cory_composition_replay_executed": false,
+    "artifact_builder_wif": "authenticated_but_artifact_registry_only",
+    "readiness_service_account": "absent_iam_404",
+    "restore_service_account": "absent_iam_404",
+    "privileged_cli": "password_reauthentication_required",
+    "direct_ssh": "current_egress_not_in_existing_firewall_32",
+    "iap": "requires_privileged_gcp_reauthentication"
+  },
+  "cory_standard_claim_ceiling": "Canonical source export and complete isolated restore are proven. Live GCP import, private connectivity, source-composition reasoning, approved apply, restart persistence, and cleanup replay are not proven.",
+  "next_action_after_auth": "Create a generated Cloud SQL clone database, import this exact dump, run the full manifest verifier from staging compute, replay the no-send 34-of-34 Cory composition benchmark, and delete the generated database and object."
+}
diff --git a/docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.md b/docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.md
new file mode 100644
index 0000000..283cf32
--- /dev/null
+++ b/docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.md
@@ -0,0 +1,40 @@
+# GCP Canonical Postgres Restore Preflight
+
+Status: **local restore preflight passes; GCP restore and Cory replay are not proven**.
+
+The canonical source is VPS Docker Postgres `teleo-pg/teleo`, not the legacy
+SQLite pipeline database. A custom dump and full manifest were captured from
+one exported, repeatable-read, read-only snapshot.
+
+| Check | Result |
+|---|---|
+| Dump | 13,462,116 bytes; SHA-256 `055e9bcd8922dbbd8a600be627c5a8f2d4a1329b383c3d9f5c28f325f6db5756` |
+| Dump catalog | 294 objects; 39 table-data objects; 113 constraints; 75 indexes |
+| Full database | 39 tables; 52,164 rows |
+| Row parity | 39/39 tables; zero count or row-hash mismatches |
+| Catalog parity | schemas, columns/defaults, constraints, indexes, sequences, views, functions, triggers, types, policies, and extensions match |
+| Role parity | required password-free `kb_apply` attributes match |
+| Performance | four bounded claim/evidence/edge/proposal queries; zero threshold failures |
+| Production safety | Leo PID `2999690`; `NRestarts=0`; state unchanged; no DB write; no Telegram send |
+| Cleanup | all local disposable containers/volumes and remote temp files removed |
+
+The first hash attempt correctly detected four apparent staging-table
+mismatches that row counts missed. Investigation showed identical bytes ordered
+under different libc collations. The production verifier now forces `COLLATE
+"C"`, after which the exact exported snapshot restores with zero mismatches.
+
+## GCP Gate
+
+- `sa-artifact-builder` authenticates through GitHub WIF but is intentionally
+  Artifact Registry-only.
+- `sa-teleo-readiness` and `sa-teleo-restore-drill` do not exist; token exchange
+  returns IAM 404.
+- The privileged local/Console identity requires the operator's Google
+  password reauthentication.
+- Direct SSH is closed because the current egress IP is not the configured
+  `/32`; IAP requires the same privileged authentication.
+
+After legitimate authentication, the remaining canary is one bounded flow:
+generated Cloud SQL database -> exact dump import -> manifest and private TLS
+readback from staging compute -> no-send 34/34 Cory composition replay ->
+delete generated database and object -> retain cleanup proof.
diff --git a/ops/capture_vps_canonical_postgres_snapshot.py b/ops/capture_vps_canonical_postgres_snapshot.py
new file mode 100755
index 0000000..055d933
--- /dev/null
+++ b/ops/capture_vps_canonical_postgres_snapshot.py
@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""Capture one pg_dump and parity manifest from the same exported VPS snapshot."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import io
+import json
+import os
+import re
+import shlex
+import shutil
+import subprocess
+import tarfile
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+try:
+    from .verify_postgres_parity_manifest import load_manifest
+except ImportError:  # pragma: no cover - direct script execution
+    from verify_postgres_parity_manifest import load_manifest
+
+
+SAFE_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\Z")
+SAFE_RUN_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{7,63}\Z")
+SAFE_SSH_TARGET_RE = re.compile(
+    r"(?:[A-Za-z0-9][A-Za-z0-9._-]{0,63}@)?[A-Za-z0-9][A-Za-z0-9.-]{0,252}\Z"
+)
+EXPECTED_FILES = {
+    "teleo-canonical.dump",
+    "teleo-canonical.dump.sha256",
+    "teleo-canonical.toc",
+    "source-manifest.jsonl",
+    "source-context.txt",
+    "service-before.txt",
+    "service-after.txt",
+}
+
+
+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()
+
+
+def ssh_base(identity_file: Path) -> list[str]:
+    return [
+        "ssh",
+        "-T",
+        "-i",
+        str(identity_file),
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "ConnectTimeout=12",
+    ]
+
+
+def remote_capture_script(*, run_id: str, container: str, database: str, service: str) -> str:
+    remote_root = f"/tmp/teleo-canonical-snapshot-{run_id}"
+    container_manifest = f"/tmp/teleo-postgres-parity-{run_id}.sql"
+    container_dump = f"/tmp/teleo-canonical-{run_id}.dump"
+    quoted = {
+        "remote_root": shlex.quote(remote_root),
+        "container": shlex.quote(container),
+        "database": shlex.quote(database),
+        "service": shlex.quote(service),
+        "container_manifest": shlex.quote(container_manifest),
+        "container_dump": shlex.quote(container_dump),
+    }
+    return f"""set -euo pipefail
+remote_root={quoted['remote_root']}
+container={quoted['container']}
+database={quoted['database']}
+service={quoted['service']}
+container_manifest={quoted['container_manifest']}
+container_dump={quoted['container_dump']}
+
+cleanup() {{
+  docker exec "$container" rm -f "$container_manifest" "$container_dump" >/dev/null 2>&1 || true
+  rm -rf "$remote_root"
+}}
+trap cleanup EXIT
+
+test -d "$remote_root"
+test -f "$remote_root/postgres-parity-manifest.sql"
+docker inspect "$container" >/dev/null
+docker exec "$container" pg_isready -U postgres -d "$database" >/dev/null
+systemctl show "$service" -p ActiveState -p SubState -p MainPID -p NRestarts -p ActiveEnterTimestamp --no-pager > "$remote_root/service-before.txt"
+docker inspect "$container" --format 'container_id={{{{.Id}}}}\nrunning={{{{.State.Running}}}}\nnetwork_mode={{{{.HostConfig.NetworkMode}}}}' > "$remote_root/source-context.txt"
+printf 'database=%s\ncaptured_at_utc=%s\n' "$database" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$remote_root/source-context.txt"
+docker cp "$remote_root/postgres-parity-manifest.sql" "$container:$container_manifest" >/dev/null
+
+coproc EXPORTER {{
+  docker exec -i "$container" psql -X -U postgres -d "$database" -Atq -v ON_ERROR_STOP=1
+}}
+exporter_pid="$EXPORTER_PID"
+exec 3>&"${{EXPORTER[1]}}"
+exec 4<&"${{EXPORTER[0]}}"
+printf '%s\n' 'begin transaction isolation level repeatable read read only;' 'select pg_export_snapshot();' >&3
+IFS= read -r snapshot_id <&4
+if [[ ! "$snapshot_id" =~ ^[0-9A-F]+-[0-9A-F]+-[0-9]+$ ]]; then
+  echo "invalid exported snapshot id" >&2
+  exit 1
+fi
+
+docker exec "$container" pg_dump \
+  -U postgres -d "$database" \
+  --format=custom --compress=9 --no-owner --no-acl \
+  --snapshot="$snapshot_id" \
+  --file="$container_dump"
+docker exec "$container" psql \
+  -X -U postgres -d "$database" -Atq -v ON_ERROR_STOP=1 \
+  -v "snapshot_id=$snapshot_id" \
+  -f "$container_manifest" \
+  > "$remote_root/source-manifest.jsonl"
+
+printf '%s\n' 'commit;' '\\q' >&3
+exec 3>&-
+while IFS= read -r _ <&4; do :; done
+exec 4<&-
+wait "$exporter_pid"
+
+docker exec "$container" pg_restore --list "$container_dump" > "$remote_root/teleo-canonical.toc"
+docker cp "$container:$container_dump" "$remote_root/teleo-canonical.dump" >/dev/null
+sha256sum "$remote_root/teleo-canonical.dump" > "$remote_root/teleo-canonical.dump.sha256"
+printf 'snapshot_exported=true\n' >> "$remote_root/source-context.txt"
+systemctl show "$service" -p ActiveState -p SubState -p MainPID -p NRestarts -p ActiveEnterTimestamp --no-pager > "$remote_root/service-after.txt"
+cmp -s "$remote_root/service-before.txt" "$remote_root/service-after.txt"
+
+tar -C "$remote_root" -cf - \
+  teleo-canonical.dump \
+  teleo-canonical.dump.sha256 \
+  teleo-canonical.toc \
+  source-manifest.jsonl \
+  source-context.txt \
+  service-before.txt \
+  service-after.txt
+"""
+
+
+def upload_manifest(ssh: list[str], target: str, remote_root: str, manifest: Path) -> None:
+    command = (
+        f"umask 077; rm -rf {shlex.quote(remote_root)}; "
+        f"mkdir -m 700 {shlex.quote(remote_root)}; "
+        f"tee {shlex.quote(remote_root + '/postgres-parity-manifest.sql')} >/dev/null"
+    )
+    completed = subprocess.run(
+        [*ssh, target, "--", command],
+        input=manifest.read_bytes(),
+        capture_output=True,
+        check=False,
+    )
+    if completed.returncode != 0:
+        raise RuntimeError("manifest upload failed: " + completed.stderr.decode(errors="replace")[-1000:])
+
+
+def cleanup_remote(ssh: list[str], target: str, remote_root: str) -> None:
+    subprocess.run(
+        [*ssh, target, "--", f"rm -rf {shlex.quote(remote_root)}"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        check=False,
+    )
+
+
+def extract_capture(payload: bytes, output_dir: Path) -> None:
+    seen: set[str] = set()
+    with tarfile.open(fileobj=io.BytesIO(payload), mode="r:") as archive:
+        for member in archive.getmembers():
+            if not member.isfile() or member.name not in EXPECTED_FILES:
+                raise RuntimeError(f"unexpected remote capture member: {member.name}")
+            if member.name in seen:
+                raise RuntimeError(f"duplicate remote capture member: {member.name}")
+            source = archive.extractfile(member)
+            if source is None:
+                raise RuntimeError(f"missing remote capture payload: {member.name}")
+            destination = output_dir / member.name
+            descriptor = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            with os.fdopen(descriptor, "wb") as handle:
+                handle.write(source.read())
+            seen.add(member.name)
+    if seen != EXPECTED_FILES:
+        raise RuntimeError(f"incomplete remote capture: missing={sorted(EXPECTED_FILES - seen)}")
+
+
+def toc_counts(path: Path) -> dict[str, int]:
+    lines = [line for line in path.read_text(errors="replace").splitlines() if line and not line.startswith(";")]
+    return {
+        "entries": len(lines),
+        "table_data": sum(" TABLE DATA " in line for line in lines),
+        "constraints": sum(" CONSTRAINT " in line or " FK CONSTRAINT " in line for line in lines),
+        "indexes": sum(" INDEX " in line for line in lines),
+    }
+
+
+def capture(args: argparse.Namespace) -> dict[str, Any]:
+    output_dir = args.output_dir.resolve()
+    if output_dir.exists():
+        raise ValueError("--output-dir must not already exist")
+    output_dir.parent.mkdir(parents=True, exist_ok=True)
+    output_dir.mkdir(mode=0o700)
+    remote_root = f"/tmp/teleo-canonical-snapshot-{args.run_id}"
+    ssh = ssh_base(args.ssh_key.resolve())
+    try:
+        upload_manifest(ssh, args.ssh_target, remote_root, args.manifest.resolve())
+        script = remote_capture_script(
+            run_id=args.run_id,
+            container=args.container,
+            database=args.database,
+            service=args.service,
+        )
+        completed = subprocess.run(
+            [*ssh, args.ssh_target, "--", "bash", "-s"],
+            input=script.encode(),
+            capture_output=True,
+            check=False,
+            timeout=args.timeout,
+        )
+        if completed.returncode != 0:
+            raise RuntimeError("remote snapshot capture failed: " + completed.stderr.decode(errors="replace")[-2000:])
+        extract_capture(completed.stdout, output_dir)
+    finally:
+        cleanup_remote(ssh, args.ssh_target, remote_root)
+
+    dump_path = output_dir / "teleo-canonical.dump"
+    expected_sha = (output_dir / "teleo-canonical.dump.sha256").read_text().split()[0]
+    actual_sha = sha256(dump_path)
+    if expected_sha != actual_sha:
+        raise RuntimeError("captured dump SHA-256 mismatch")
+    manifest = load_manifest(output_dir / "source-manifest.jsonl")
+    before = (output_dir / "service-before.txt").read_text()
+    after = (output_dir / "service-after.txt").read_text()
+    return {
+        "artifact": "vps_canonical_postgres_exported_snapshot",
+        "generated_at_utc": datetime.now(UTC).isoformat(),
+        "status": "pass",
+        "source": {
+            "ssh_target": args.ssh_target,
+            "container": args.container,
+            "database": args.database,
+            "service": args.service,
+        },
+        "snapshot_exported": True,
+        "dump": {
+            "path": str(dump_path),
+            "bytes": dump_path.stat().st_size,
+            "sha256": actual_sha,
+            "toc": toc_counts(output_dir / "teleo-canonical.toc"),
+        },
+        "manifest": {
+            "path": str(output_dir / "source-manifest.jsonl"),
+            "table_count": len(manifest["tables"]),
+            "total_rows": sum(int(row["row_count"]) for row in manifest["tables"].values()),
+            "application_roles": manifest["singleton"]["application_roles"]["items"],
+        },
+        "service_unchanged": before == after,
+        "production_db_mutated": False,
+        "telegram_send_attempted": False,
+    }
+
+
+def cleanup_failed_output(output_dir: Path, *, preexisting: bool) -> None:
+    if not preexisting and output_dir.exists():
+        shutil.rmtree(output_dir)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--ssh-target", required=True)
+    parser.add_argument("--ssh-key", required=True, type=Path)
+    parser.add_argument("--container", default="teleo-pg")
+    parser.add_argument("--database", default="teleo")
+    parser.add_argument("--service", default="leoclean-gateway.service")
+    parser.add_argument("--manifest", default=Path("ops/postgres_parity_manifest.sql"), type=Path)
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--output-dir", required=True, type=Path)
+    parser.add_argument("--timeout", default=180, type=int)
+    args = parser.parse_args()
+    for value, flag in ((args.container, "--container"), (args.database, "--database"), (args.service, "--service")):
+        if not SAFE_NAME_RE.fullmatch(value):
+            parser.error(f"{flag} contains unsafe characters")
+    if not SAFE_SSH_TARGET_RE.fullmatch(args.ssh_target):
+        parser.error("--ssh-target must be a host or user@host without options")
+    if not SAFE_RUN_RE.fullmatch(args.run_id):
+        parser.error("--run-id must be 8-64 safe characters")
+    if not args.ssh_key.is_file():
+        parser.error("--ssh-key must be an existing file")
+    if not args.manifest.is_file():
+        parser.error("--manifest must be an existing file")
+    if not args.execute:
+        print(
+            json.dumps(
+                {
+                    "artifact": "vps_canonical_postgres_exported_snapshot_plan",
+                    "status": "dry_run",
+                    "source": f"{args.ssh_target}:{args.container}/{args.database}",
+                    "output_dir": str(args.output_dir),
+                    "mutates_production_db": False,
+                },
+                indent=2,
+                sort_keys=True,
+            )
+        )
+        raise SystemExit(0)
+    return args
+
+
+def main() -> int:
+    args = parse_args()
+    output_dir = args.output_dir.resolve()
+    output_preexisted = output_dir.exists()
+    try:
+        payload = capture(args)
+    except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as exc:
+        cleanup_failed_output(output_dir, preexisting=output_preexisted)
+        payload = {
+            "artifact": "vps_canonical_postgres_exported_snapshot",
+            "generated_at_utc": datetime.now(UTC).isoformat(),
+            "status": "fail",
+            "error": str(exc),
+        }
+        print(json.dumps(payload, indent=2, sort_keys=True))
+        return 1
+    receipt = args.output_dir.resolve() / "receipt.json"
+    receipt.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    os.chmod(receipt, 0o600)
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py
index 729703e..b32662b 100755
--- a/ops/check_gcp_infra_readiness.py
+++ b/ops/check_gcp_infra_readiness.py
@@ -56,18 +56,52 @@ class Check:
     current_tier: str = "T3_live_readonly"
 
 
+class GcloudAccessError(RuntimeError):
+    """The command could not establish an authorized live readback."""
+
+
+ACCESS_ERROR_MARKERS = (
+    "permission_denied",
+    "required '",
+    "does not have permission",
+    "not authorized",
+    "unable to acquire impersonated credentials",
+    "gaia id not found",
+    "reauthentication failed",
+    "gcloud auth login",
+)
+
+
+def _run_gcloud(args: list[str]) -> str:
+    completed = subprocess.run(args, text=True, capture_output=True, check=False)
+    combined = "\n".join(part for part in (completed.stdout, completed.stderr) if part).strip()
+    lowered = combined.lower()
+    if any(marker in lowered for marker in ACCESS_ERROR_MARKERS):
+        raise GcloudAccessError(combined[-2000:])
+    if completed.returncode != 0:
+        raise subprocess.CalledProcessError(completed.returncode, args, output=combined)
+    return completed.stdout
+
+
 def run_json(args: list[str]) -> object:
-    output = subprocess.check_output(args, text=True)
+    output = _run_gcloud(args)
     return json.loads(output or "null")
 
 
 def run_text(args: list[str]) -> str:
-    return subprocess.check_output(args, text=True, stderr=subprocess.STDOUT).strip()
+    return _run_gcloud(args).strip()
 
 
 def safe_check(name: str, fn) -> Check:
     try:
         return fn()
+    except GcloudAccessError as exc:
+        return Check(
+            name,
+            "blocked",
+            "gcloud_live_readback_unavailable: " + str(exc).strip()[-500:],
+            current_tier="T0_no_live_readback",
+        )
     except subprocess.CalledProcessError as exc:
         return Check(name, "fail", exc.output.strip()[-500:] if exc.output else str(exc))
     except Exception as exc:  # pragma: no cover - defensive operator readback
@@ -687,12 +721,116 @@ def check_gcp_cloudsql_restore_drill_contract() -> Check:
     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",
+        "legacy SQLite shadow-schema script can prepare GCS import SQL, optionally upload/import via Cloud SQL Admin, and emits target-count readback SQL; this is not canonical Leo Postgres parity",
         required_tier="T2_restore_execution_contract",
         current_tier="T2_restore_execution_contract",
     )
 
 
+def check_canonical_postgres_restore_contract() -> Check:
+    capture = Path("ops/capture_vps_canonical_postgres_snapshot.py")
+    manifest = Path("ops/postgres_parity_manifest.sql")
+    verifier = Path("ops/verify_postgres_parity_manifest.py")
+    runbook = Path("docs/gcp-kb-cloudsql-restore-runbook.md")
+    missing = [str(path) for path in (capture, manifest, verifier, runbook) if not path.exists()]
+    if missing:
+        return Check(
+            "canonical_postgres_restore_contract",
+            "blocked",
+            f"missing={missing}",
+            required_tier="T2_canonical_restore_contract",
+            current_tier="T1_foundation",
+        )
+    combined = "\n".join(path.read_text() for path in (capture, manifest, verifier, runbook))
+    required = [
+        "pg_export_snapshot",
+        "--snapshot=",
+        'collate "C"',
+        "rowset_md5",
+        "constraints",
+        "indexes",
+        "sequences",
+        "functions",
+        "application_roles",
+        "gcp_staging",
+        "private_connectivity",
+        "teleo_clone_",
+    ]
+    absent = [item for item in required if item not in combined]
+    if absent:
+        return Check(
+            "canonical_postgres_restore_contract",
+            "fail",
+            f"missing_contract={absent}",
+            required_tier="T2_canonical_restore_contract",
+            current_tier="T1_foundation",
+        )
+    return Check(
+        "canonical_postgres_restore_contract",
+        "pass",
+        "exported-snapshot capture and verifier cover full canonical Postgres row/catalog/role/performance parity plus required GCP private-connectivity proof",
+        required_tier="T2_canonical_restore_contract",
+        current_tier="T2_canonical_restore_contract",
+    )
+
+
+def check_local_canonical_postgres_restore_preflight() -> Check:
+    proof_path = Path(
+        "docs/reports/leo-working-state-20260709/gcp-canonical-postgres-restore-preflight-current.json"
+    )
+    if not proof_path.exists():
+        return Check(
+            "local_canonical_postgres_restore_preflight",
+            "blocked",
+            f"missing={proof_path}",
+            required_tier="T2_local_canonical_restore",
+            current_tier="T1_foundation",
+        )
+    proof = json.loads(proof_path.read_text())
+    source = proof.get("source", {})
+    local = proof.get("local_restore", {})
+    safety = proof.get("production_safety", {})
+    cleanup = proof.get("cleanup", {})
+    problems = []
+    if proof.get("status") != "local_restore_preflight_pass_gcp_unproven":
+        problems.append(f"status={proof.get('status')}")
+    if source.get("table_count") != local.get("source_table_count") or local.get("source_table_count") != local.get(
+        "target_table_count"
+    ):
+        problems.append("table_count_mismatch")
+    if source.get("total_rows") != local.get("source_total_rows") or local.get("source_total_rows") != local.get(
+        "target_total_rows"
+    ):
+        problems.append("total_row_mismatch")
+    for key in (
+        "table_content_mismatches",
+        "application_role_mismatches",
+        "performance_problems",
+        "structural_hash_mismatches",
+    ):
+        if local.get(key) != 0:
+            problems.append(f"{key}={local.get(key)}")
+    if safety.get("production_db_mutated") is not False or safety.get("service_state_unchanged") is not True:
+        problems.append("production_safety_not_proven")
+    if cleanup.get("local_disposable_containers_remaining") != 0 or cleanup.get("remote_run_directory_removed") is not True:
+        problems.append("cleanup_not_proven")
+    if problems:
+        return Check(
+            "local_canonical_postgres_restore_preflight",
+            "fail",
+            " ".join(problems),
+            required_tier="T2_local_canonical_restore",
+            current_tier="T1_foundation",
+        )
+    return Check(
+        "local_canonical_postgres_restore_preflight",
+        "pass",
+        f"exported_snapshot_tables={source.get('table_count')} rows={source.get('total_rows')} dump_sha256={source.get('dump_sha256')}",
+        required_tier="T2_local_canonical_restore",
+        current_tier="T2_local_canonical_restore",
+    )
+
+
 def check_gcp_runtime_baseline_contract() -> Check:
     script = Path("ops/apply_gcp_runtime_baseline.py")
     runbook = Path("docs/gcp-infra-hardening-20260707.md")
@@ -795,7 +933,7 @@ def check_kb_restore_or_replication() -> Check:
     return Check(
         "kb_restore_or_replication",
         "blocked",
-        "GCP standby target exists, but no source-data restore, pg_dump import, logical replication, or query readback has been retained",
+        "no canonical source-data restore, pg_dump import, logical replication, or query readback has been retained; target existence must be established separately by cloud_sql_standby_target",
         required_tier="T4_restore_or_replication_readback",
         current_tier="T1_foundation",
     )
@@ -816,6 +954,8 @@ def main() -> int:
         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("canonical_postgres_restore_contract", check_canonical_postgres_restore_contract),
+        safe_check("local_canonical_postgres_restore_preflight", check_local_canonical_postgres_restore_preflight),
         safe_check("gcp_runtime_baseline_contract", check_gcp_runtime_baseline_contract),
         safe_check("gcp_service_communications_contract", check_gcp_service_communications_contract),
         safe_check("kb_restore_or_replication", check_kb_restore_or_replication),
@@ -830,7 +970,7 @@ def main() -> int:
         "fail_count": sum(check.status == "fail" for check in checks),
     }
     print(json.dumps(payload, indent=2))
-    return 1 if payload["fail_count"] else 0
+    return 1 if payload["fail_count"] or payload["blocked_count"] else 0
 
 
 if __name__ == "__main__":
diff --git a/ops/postgres_parity_manifest.sql b/ops/postgres_parity_manifest.sql
new file mode 100644
index 0000000..2f6a74e
--- /dev/null
+++ b/ops/postgres_parity_manifest.sql
@@ -0,0 +1,359 @@
+\set ON_ERROR_STOP on
+begin transaction isolation level repeatable read read only;
+\if :{?snapshot_id}
+set transaction snapshot :'snapshot_id';
+\endif
+set local statement_timeout = '30s';
+
+select jsonb_build_object(
+  'kind', 'identity',
+  'database', current_database(),
+  'current_user', current_user,
+  'server_version_num', current_setting('server_version_num')::integer,
+  'server_encoding', current_setting('server_encoding'),
+  'database_collation', (select datcollate from pg_database where datname = current_database()),
+  'database_ctype', (select datctype from pg_database where datname = current_database()),
+  'transaction_read_only', current_setting('transaction_read_only'),
+  'server_address', inet_server_addr()::text,
+  'server_port', inet_server_port(),
+  'ssl', coalesce((select ssl from pg_stat_ssl where pid = pg_backend_pid()), false),
+  'ssl_version', (select version from pg_stat_ssl where pid = pg_backend_pid()),
+  'captured_at', clock_timestamp()
+)::text;
+
+select jsonb_build_object(
+  'kind', 'schemas',
+  'items', coalesce(
+    (select jsonb_agg(nspname order by nspname)
+       from pg_namespace
+      where nspname in ('public', 'kb_stage')),
+    '[]'::jsonb
+  )
+)::text;
+
+select jsonb_build_object(
+  'kind', 'extensions',
+  'items', coalesce(
+    (select jsonb_agg(jsonb_build_object('name', extname, 'version', extversion) order by extname)
+       from pg_extension),
+    '[]'::jsonb
+  )
+)::text;
+
+select jsonb_build_object(
+  'kind', 'application_roles',
+  'items', coalesce(
+    (select jsonb_agg(
+       jsonb_build_object(
+         'name', rolname,
+         'can_login', rolcanlogin,
+         'superuser', rolsuper,
+         'create_db', rolcreatedb,
+         'create_role', rolcreaterole,
+         'inherit', rolinherit,
+         'replication', rolreplication,
+         'bypass_rls', rolbypassrls
+       ) order by rolname)
+       from pg_roles
+      where rolname in ('kb_apply', 'kb_review')),
+    '[]'::jsonb
+  )
+)::text;
+
+select jsonb_build_object(
+  'kind', 'columns',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', table_schema,
+        'table', table_name,
+        'position', ordinal_position,
+        'name', column_name,
+        'data_type', data_type,
+        'udt_name', udt_name,
+        'nullable', is_nullable,
+        'default', column_default
+      ) order by table_schema, table_name, ordinal_position
+    ),
+    '[]'::jsonb
+  )
+)::text
+from information_schema.columns
+where table_schema in ('public', 'kb_stage');
+
+select jsonb_build_object(
+  'kind', 'constraints',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', namespace.nspname,
+        'table', relation.relname,
+        'name', constraint_row.conname,
+        'type', constraint_row.contype,
+        'definition', pg_get_constraintdef(constraint_row.oid, true),
+        'validated', constraint_row.convalidated,
+        'deferrable', constraint_row.condeferrable,
+        'initially_deferred', constraint_row.condeferred
+      ) order by namespace.nspname, relation.relname, constraint_row.conname
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_constraint constraint_row
+join pg_class relation on relation.oid = constraint_row.conrelid
+join pg_namespace namespace on namespace.oid = relation.relnamespace
+where namespace.nspname in ('public', 'kb_stage');
+
+select jsonb_build_object(
+  'kind', 'indexes',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', schemaname,
+        'table', tablename,
+        'name', indexname,
+        'definition', indexdef
+      ) order by schemaname, tablename, indexname
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_indexes
+where schemaname in ('public', 'kb_stage');
+
+select jsonb_build_object(
+  'kind', 'sequences',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', schemaname,
+        'name', sequencename,
+        'data_type', data_type,
+        'start_value', start_value,
+        'min_value', min_value,
+        'max_value', max_value,
+        'increment_by', increment_by,
+        'cycle', cycle,
+        'cache_size', cache_size,
+        'last_value', last_value
+      ) order by schemaname, sequencename
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_sequences
+where schemaname in ('public', 'kb_stage');
+
+select jsonb_build_object(
+  'kind', 'views',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', schemaname,
+        'name', viewname,
+        'definition', definition
+      ) order by schemaname, viewname
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_views
+where schemaname in ('public', 'kb_stage');
+
+select jsonb_build_object(
+  'kind', 'functions',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', namespace.nspname,
+        'name', procedure.proname,
+        'identity_arguments', pg_get_function_identity_arguments(procedure.oid),
+        'result', pg_get_function_result(procedure.oid),
+        'language', language.lanname,
+        'volatility', procedure.provolatile,
+        'security_definer', procedure.prosecdef,
+        'definition', pg_get_functiondef(procedure.oid)
+      ) order by namespace.nspname, procedure.proname, pg_get_function_identity_arguments(procedure.oid)
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_proc procedure
+join pg_namespace namespace on namespace.oid = procedure.pronamespace
+join pg_language language on language.oid = procedure.prolang
+where namespace.nspname in ('public', 'kb_stage');
+
+select jsonb_build_object(
+  'kind', 'triggers',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', namespace.nspname,
+        'table', relation.relname,
+        'name', trigger_row.tgname,
+        'enabled', trigger_row.tgenabled,
+        'definition', pg_get_triggerdef(trigger_row.oid, true)
+      ) order by namespace.nspname, relation.relname, trigger_row.tgname
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_trigger trigger_row
+join pg_class relation on relation.oid = trigger_row.tgrelid
+join pg_namespace namespace on namespace.oid = relation.relnamespace
+where namespace.nspname in ('public', 'kb_stage')
+  and not trigger_row.tgisinternal;
+
+select jsonb_build_object(
+  'kind', 'types',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', namespace.nspname,
+        'name', type_row.typname,
+        'kind', type_row.typtype,
+        'category', type_row.typcategory,
+        'base_type', case when type_row.typbasetype = 0 then null else format_type(type_row.typbasetype, null) end,
+        'not_null', type_row.typnotnull,
+        'default', type_row.typdefault,
+        'enum_values', (
+          select coalesce(jsonb_agg(enum_row.enumlabel order by enum_row.enumsortorder), '[]'::jsonb)
+          from pg_enum enum_row
+          where enum_row.enumtypid = type_row.oid
+        )
+      ) order by namespace.nspname, type_row.typname
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_type type_row
+join pg_namespace namespace on namespace.oid = type_row.typnamespace
+where namespace.nspname in ('public', 'kb_stage')
+  and type_row.typtype in ('d', 'e');
+
+select jsonb_build_object(
+  'kind', 'policies',
+  'items', coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'schema', schemaname,
+        'table', tablename,
+        'name', policyname,
+        'permissive', permissive,
+        'roles', roles,
+        'command', cmd,
+        'using', qual,
+        'check', with_check
+      ) order by schemaname, tablename, policyname
+    ),
+    '[]'::jsonb
+  )
+)::text
+from pg_policies
+where schemaname in ('public', 'kb_stage');
+
+select format(
+  $manifest$
+  select jsonb_build_object(
+    'kind', 'table',
+    'schema', %L,
+    'table', %L,
+    'row_count', count(*)::bigint,
+    'rowset_md5', md5(coalesce(string_agg(to_jsonb(row_value)::text, E'\n'
+      order by (to_jsonb(row_value)::text) collate "C"), ''))
+  )::text
+  from %I.%I row_value;
+  $manifest$,
+  schemaname,
+  tablename,
+  schemaname,
+  tablename
+)
+from pg_tables
+where schemaname in ('public', 'kb_stage')
+order by schemaname, tablename
+\gexec
+
+with started as materialized (
+  select clock_timestamp() as at
+), observed as materialized (
+  select count(*)::bigint as row_count, min(started.at) as started_at
+  from public.claims cross join started
+), finished as materialized (
+  select clock_timestamp() as finished_at, max(row_count) as row_count, min(started_at) as started_at
+  from observed
+)
+select jsonb_build_object(
+  'kind', 'performance',
+  'query', 'count_claims',
+  'elapsed_ms', extract(epoch from (finished_at - started_at)) * 1000,
+  'observed_rows', row_count
+)::text
+from finished;
+
+with started as materialized (
+  select clock_timestamp() as at
+), candidate as materialized (
+  select id from public.claims order by id limit 1
+), observed as materialized (
+  select count(*)::bigint as row_count, min(started.at) as started_at
+  from public.claim_evidence evidence
+  join candidate on candidate.id = evidence.claim_id
+  cross join started
+), finished as materialized (
+  select clock_timestamp() as finished_at, max(row_count) as row_count, min(started_at) as started_at
+  from observed
+)
+select jsonb_build_object(
+  'kind', 'performance',
+  'query', 'claim_evidence_by_claim_id',
+  'elapsed_ms', extract(epoch from (finished_at - started_at)) * 1000,
+  'observed_rows', row_count
+)::text
+from finished;
+
+with started as materialized (
+  select clock_timestamp() as at
+), candidate as materialized (
+  select id from public.claims order by id limit 1
+), observed as materialized (
+  select count(*)::bigint as row_count, min(started.at) as started_at
+  from public.claim_edges edge_row
+  join candidate on candidate.id in (edge_row.from_claim, edge_row.to_claim)
+  cross join started
+), finished as materialized (
+  select clock_timestamp() as finished_at, max(row_count) as row_count, min(started_at) as started_at
+  from observed
+)
+select jsonb_build_object(
+  'kind', 'performance',
+  'query', 'claim_edges_by_claim_id',
+  'elapsed_ms', extract(epoch from (finished_at - started_at)) * 1000,
+  'observed_rows', row_count
+)::text
+from finished;
+
+with started as materialized (
+  select clock_timestamp() as at
+), observed as materialized (
+  select count(*)::bigint as row_count, min(started.at) as started_at
+  from (
+    select id from kb_stage.kb_proposals
+    where status = 'pending_review'
+    order by created_at desc
+    limit 25
+  ) proposal_row
+  cross join started
+), finished as materialized (
+  select clock_timestamp() as finished_at, max(row_count) as row_count, min(started_at) as started_at
+  from observed
+)
+select jsonb_build_object(
+  'kind', 'performance',
+  'query', 'pending_proposals_latest',
+  'elapsed_ms', extract(epoch from (finished_at - started_at)) * 1000,
+  'observed_rows', row_count
+)::text
+from finished;
+
+rollback;
diff --git a/ops/verify_postgres_parity_manifest.py b/ops/verify_postgres_parity_manifest.py
new file mode 100755
index 0000000..f8f5abe
--- /dev/null
+++ b/ops/verify_postgres_parity_manifest.py
@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+"""Compare canonical Postgres source and restored-target JSONL manifests."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import ipaddress
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+SINGLETON_KINDS = {
+    "identity",
+    "schemas",
+    "extensions",
+    "application_roles",
+    "columns",
+    "constraints",
+    "indexes",
+    "sequences",
+    "views",
+    "functions",
+    "triggers",
+    "types",
+    "policies",
+}
+
+
+def canonical_hash(value: Any) -> str:
+    payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+    return hashlib.sha256(payload.encode()).hexdigest()
+
+
+def load_manifest(path: Path) -> dict[str, Any]:
+    singleton: dict[str, dict[str, Any]] = {}
+    tables: dict[str, dict[str, Any]] = {}
+    performance: dict[str, dict[str, Any]] = {}
+    with path.open(encoding="utf-8") as handle:
+        for line_number, raw_line in enumerate(handle, 1):
+            line = raw_line.strip()
+            if not line or line in {"BEGIN", "SET", "ROLLBACK"}:
+                continue
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError as exc:
+                raise ValueError(f"{path}:{line_number}: invalid JSONL manifest row") from exc
+            kind = row.get("kind")
+            if kind in SINGLETON_KINDS:
+                if kind in singleton:
+                    raise ValueError(f"{path}: duplicate {kind} row")
+                singleton[str(kind)] = row
+            elif kind == "table":
+                key = f"{row.get('schema')}.{row.get('table')}"
+                if key in tables:
+                    raise ValueError(f"{path}: duplicate table row {key}")
+                tables[key] = row
+            elif kind == "performance":
+                key = str(row.get("query"))
+                if key in performance:
+                    raise ValueError(f"{path}: duplicate performance row {key}")
+                performance[key] = row
+            else:
+                raise ValueError(f"{path}:{line_number}: unknown manifest kind {kind!r}")
+    missing = SINGLETON_KINDS - set(singleton)
+    if missing:
+        raise ValueError(f"{path}: missing singleton rows {sorted(missing)}")
+    if not tables:
+        raise ValueError(f"{path}: no table rows")
+    return {"singleton": singleton, "tables": tables, "performance": performance}
+
+
+def item_map(row: dict[str, Any], key: str) -> dict[str, dict[str, Any]]:
+    return {str(item[key]): item for item in row.get("items", [])}
+
+
+def compare_manifests(
+    source: dict[str, Any],
+    target: dict[str, Any],
+    *,
+    max_target_query_ms: float,
+    max_target_source_ratio: float,
+) -> tuple[list[str], dict[str, Any]]:
+    problems: list[str] = []
+    source_singleton = source["singleton"]
+    target_singleton = target["singleton"]
+
+    source_identity = source_singleton["identity"]
+    target_identity = target_singleton["identity"]
+    source_major = int(source_identity["server_version_num"]) // 10000
+    target_major = int(target_identity["server_version_num"]) // 10000
+    if source_major != target_major:
+        problems.append(f"postgres_major source={source_major} target={target_major}")
+    if source_identity.get("transaction_read_only") != "on":
+        problems.append("source_manifest_not_captured_read_only")
+    if target_identity.get("transaction_read_only") != "on":
+        problems.append("target_manifest_not_captured_read_only")
+
+    source_tables = source["tables"]
+    target_tables = target["tables"]
+    if set(source_tables) != set(target_tables):
+        problems.append(
+            "table_set mismatch source_only="
+            + json.dumps(sorted(set(source_tables) - set(target_tables)))
+            + " target_only="
+            + json.dumps(sorted(set(target_tables) - set(source_tables)))
+        )
+    table_mismatches = []
+    for table in sorted(set(source_tables) & set(target_tables)):
+        source_row = source_tables[table]
+        target_row = target_tables[table]
+        mismatch = {
+            field: {"source": source_row.get(field), "target": target_row.get(field)}
+            for field in ("row_count", "rowset_md5")
+            if source_row.get(field) != target_row.get(field)
+        }
+        if mismatch:
+            table_mismatches.append({"table": table, "mismatch": mismatch})
+    if table_mismatches:
+        problems.append(f"table_content_mismatches={len(table_mismatches)}")
+
+    structural_hashes: dict[str, dict[str, str]] = {}
+    for kind in (
+        "schemas",
+        "columns",
+        "constraints",
+        "indexes",
+        "sequences",
+        "views",
+        "functions",
+        "triggers",
+        "types",
+        "policies",
+    ):
+        source_hash = canonical_hash(source_singleton[kind].get("items", []))
+        target_hash = canonical_hash(target_singleton[kind].get("items", []))
+        structural_hashes[kind] = {"source": source_hash, "target": target_hash}
+        if source_hash != target_hash:
+            problems.append(f"{kind}_hash_mismatch")
+
+    source_extensions = item_map(source_singleton["extensions"], "name")
+    target_extensions = item_map(target_singleton["extensions"], "name")
+    extension_mismatches = {
+        name: {"source": item, "target": target_extensions.get(name)}
+        for name, item in source_extensions.items()
+        if target_extensions.get(name) != item
+    }
+    if extension_mismatches:
+        problems.append(f"required_extension_mismatches={len(extension_mismatches)}")
+
+    source_roles = item_map(source_singleton["application_roles"], "name")
+    target_roles = item_map(target_singleton["application_roles"], "name")
+    role_mismatches = {
+        name: {"source": item, "target": target_roles.get(name)}
+        for name, item in source_roles.items()
+        if target_roles.get(name) != item
+    }
+    if role_mismatches:
+        problems.append(f"application_role_mismatches={len(role_mismatches)}")
+
+    performance_problems = []
+    performance_rows = []
+    for query, target_row in sorted(target["performance"].items()):
+        target_ms = float(target_row.get("elapsed_ms") or 0)
+        source_row = source["performance"].get(query)
+        source_ms = float((source_row or {}).get("elapsed_ms") or 0)
+        ratio = target_ms / max(source_ms, 5.0)
+        row = {"query": query, "source_ms": source_ms, "target_ms": target_ms, "source_ratio": ratio}
+        performance_rows.append(row)
+        if source_row is None:
+            performance_problems.append(f"{query}:missing_source_benchmark")
+        if target_ms > max_target_query_ms:
+            performance_problems.append(f"{query}:target_ms={target_ms:.3f}>{max_target_query_ms:.3f}")
+        if ratio > max_target_source_ratio:
+            performance_problems.append(f"{query}:source_ratio={ratio:.3f}>{max_target_source_ratio:.3f}")
+    missing_target_benchmarks = sorted(set(source["performance"]) - set(target["performance"]))
+    if missing_target_benchmarks:
+        performance_problems.append("missing_target_benchmarks=" + ",".join(missing_target_benchmarks))
+    if performance_problems:
+        problems.append(f"performance_problems={len(performance_problems)}")
+
+    details = {
+        "source_database": source_identity.get("database"),
+        "target_database": target_identity.get("database"),
+        "source_connection": {
+            "server_address": source_identity.get("server_address"),
+            "server_port": source_identity.get("server_port"),
+            "ssl": source_identity.get("ssl"),
+            "ssl_version": source_identity.get("ssl_version"),
+            "database_collation": source_identity.get("database_collation"),
+            "database_ctype": source_identity.get("database_ctype"),
+        },
+        "target_connection": {
+            "server_address": target_identity.get("server_address"),
+            "server_port": target_identity.get("server_port"),
+            "ssl": target_identity.get("ssl"),
+            "ssl_version": target_identity.get("ssl_version"),
+            "database_collation": target_identity.get("database_collation"),
+            "database_ctype": target_identity.get("database_ctype"),
+        },
+        "source_table_count": len(source_tables),
+        "target_table_count": len(target_tables),
+        "source_total_rows": sum(int(row["row_count"]) for row in source_tables.values()),
+        "target_total_rows": sum(int(row["row_count"]) for row in target_tables.values()),
+        "table_mismatches": table_mismatches,
+        "structural_hashes": structural_hashes,
+        "required_extension_mismatches": extension_mismatches,
+        "application_role_mismatches": role_mismatches,
+        "target_extra_application_roles": sorted(set(target_roles) - set(source_roles)),
+        "performance": performance_rows,
+        "performance_problems": performance_problems,
+    }
+    return problems, details
+
+
+def is_rfc1918(address: str) -> bool:
+    parsed = ipaddress.ip_address(address)
+    return any(
+        parsed in network
+        for network in (
+            ipaddress.ip_network("10.0.0.0/8"),
+            ipaddress.ip_network("172.16.0.0/12"),
+            ipaddress.ip_network("192.168.0.0/16"),
+        )
+    )
+
+
+def load_connectivity(
+    path: Path | None,
+    scope: str,
+    target_identity: dict[str, Any],
+) -> tuple[list[str], dict[str, Any]]:
+    if scope == "local_restore":
+        return [], {"required": False, "status": "not_applicable_local_restore"}
+    if path is None:
+        return ["gcp_private_connectivity_proof_missing"], {"required": True, "status": "missing"}
+    proof = json.loads(path.read_text())
+    problems = []
+    if proof.get("status") != "pass":
+        problems.append("gcp_private_connectivity_status_not_pass")
+    if proof.get("private_connectivity") is not True:
+        problems.append("gcp_private_connectivity_not_proven")
+    if proof.get("public_ip_disabled") is not True:
+        problems.append("gcp_public_ip_disabled_not_proven")
+    if not proof.get("source_compute"):
+        problems.append("gcp_source_compute_missing")
+    target_database = target_identity.get("database")
+    target_address = target_identity.get("server_address")
+    target_ssl = target_identity.get("ssl")
+    if not target_database:
+        problems.append("gcp_target_manifest_database_missing")
+    if proof.get("target_database") != target_database:
+        problems.append("gcp_connectivity_target_database_mismatch")
+    if not target_address:
+        problems.append("gcp_target_manifest_server_address_missing")
+    else:
+        try:
+            if not is_rfc1918(str(target_address)):
+                problems.append("gcp_target_manifest_server_address_not_rfc1918")
+        except ValueError:
+            problems.append("gcp_target_manifest_server_address_invalid")
+    if proof.get("server_address") != target_address:
+        problems.append("gcp_connectivity_server_address_mismatch")
+    if target_ssl is not True:
+        problems.append("gcp_target_manifest_tls_not_proven")
+    if proof.get("ssl") is not True:
+        problems.append("gcp_connectivity_tls_not_proven")
+    return problems, {"required": True, "path": str(path), "proof": proof}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source", required=True, type=Path)
+    parser.add_argument("--target", required=True, type=Path)
+    parser.add_argument("--scope", choices=("local_restore", "gcp_staging"), default="local_restore")
+    parser.add_argument("--connectivity-proof", type=Path)
+    parser.add_argument("--max-target-query-ms", type=float, default=2000.0)
+    parser.add_argument("--max-target-source-ratio", type=float, default=20.0)
+    parser.add_argument("--output", required=True, type=Path)
+    args = parser.parse_args()
+
+    try:
+        source = load_manifest(args.source)
+        target = load_manifest(args.target)
+        problems, details = compare_manifests(
+            source,
+            target,
+            max_target_query_ms=args.max_target_query_ms,
+            max_target_source_ratio=args.max_target_source_ratio,
+        )
+        connectivity_problems, connectivity = load_connectivity(
+            args.connectivity_proof,
+            args.scope,
+            target["singleton"]["identity"],
+        )
+        problems.extend(connectivity_problems)
+        status = "pass" if not problems else "fail"
+        error = None
+    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
+        status = "fail"
+        problems = [f"manifest_error={exc}"]
+        details = {}
+        connectivity = {"required": args.scope == "gcp_staging", "status": "not_checked"}
+        error = str(exc)
+
+    payload = {
+        "artifact": "canonical_postgres_parity_verification",
+        "generated_at_utc": datetime.now(UTC).isoformat(),
+        "scope": args.scope,
+        "source_manifest": str(args.source),
+        "target_manifest": str(args.target),
+        "status": status,
+        "problems": problems,
+        "details": details,
+        "private_connectivity": connectivity,
+        "error": error,
+        "not_proven_by_this_artifact": [
+            "GCP staging Leo composition replay",
+            "ongoing replication after the manifest timestamps",
+            "production cutover",
+        ],
+    }
+    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 status == "pass" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_capture_vps_canonical_postgres_snapshot.py b/tests/test_capture_vps_canonical_postgres_snapshot.py
new file mode 100644
index 0000000..b8f07a4
--- /dev/null
+++ b/tests/test_capture_vps_canonical_postgres_snapshot.py
@@ -0,0 +1,68 @@
+from pathlib import Path
+
+from ops.capture_vps_canonical_postgres_snapshot import (
+    SAFE_SSH_TARGET_RE,
+    cleanup_failed_output,
+    remote_capture_script,
+    toc_counts,
+)
+
+
+def test_remote_capture_uses_one_exported_read_only_snapshot() -> None:
+    script = remote_capture_script(
+        run_id="canonical-20260711",
+        container="teleo-pg",
+        database="teleo",
+        service="leoclean-gateway.service",
+    )
+
+    assert "begin transaction isolation level repeatable read read only" in script
+    assert "select pg_export_snapshot()" in script
+    assert "[0-9A-F]+-[0-9A-F]+-[0-9]+" in script
+    assert '--snapshot="$snapshot_id"' in script
+    assert '-v "snapshot_id=$snapshot_id"' in script
+    assert "--no-owner --no-acl" in script
+    assert "cmp -s" in script
+    assert "systemctl restart" not in script
+    assert "docker restart" not in script
+
+
+def test_manifest_hash_order_is_collation_independent() -> None:
+    manifest = Path("ops/postgres_parity_manifest.sql").read_text()
+
+    assert 'collate "C"' in manifest
+
+
+def test_toc_counts_database_objects(tmp_path: Path) -> None:
+    toc = tmp_path / "dump.toc"
+    toc.write_text(
+        "; header\n"
+        "1; 0 0 TABLE DATA public claims postgres\n"
+        "2; 0 0 CONSTRAINT public claims claims_pkey postgres\n"
+        "3; 0 0 FK CONSTRAINT public evidence evidence_claim_fkey postgres\n"
+        "4; 0 0 INDEX public claims claims_text_idx postgres\n"
+    )
+
+    assert toc_counts(toc) == {"entries": 4, "table_data": 1, "constraints": 2, "indexes": 1}
+
+
+def test_failure_cleanup_never_removes_preexisting_output(tmp_path: Path) -> None:
+    existing = tmp_path / "existing"
+    existing.mkdir()
+    marker = existing / "user-file"
+    marker.write_text("keep")
+    cleanup_failed_output(existing, preexisting=True)
+
+    created_by_run = tmp_path / "created-by-run"
+    created_by_run.mkdir()
+    cleanup_failed_output(created_by_run, preexisting=False)
+
+    assert marker.read_text() == "keep"
+    assert not created_by_run.exists()
+
+
+def test_ssh_target_rejects_open_ssh_option_injection() -> None:
+    assert SAFE_SSH_TARGET_RE.fullmatch("root@77.42.65.182")
+    assert SAFE_SSH_TARGET_RE.fullmatch("teleo-staging-1.internal")
+    assert not SAFE_SSH_TARGET_RE.fullmatch("-oProxyCommand=malicious")
+    assert not SAFE_SSH_TARGET_RE.fullmatch("root@host command")
diff --git a/tests/test_gcp_infra_readiness_checker.py b/tests/test_gcp_infra_readiness_checker.py
index ca96e36..8a78acd 100644
--- a/tests/test_gcp_infra_readiness_checker.py
+++ b/tests/test_gcp_infra_readiness_checker.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import subprocess
 from pathlib import Path
 
 import ops.check_gcp_infra_readiness as checker
@@ -81,7 +82,61 @@ def test_backup_bucket_check_rejects_weak_bucket_settings(monkeypatch) -> None:
 
 
 def test_readiness_checker_does_not_use_gsutil() -> None:
-    assert "gsutil" not in Path("ops/check_gcp_infra_readiness.py").read_text()
+    source = Path("ops/check_gcp_infra_readiness.py").read_text()
+
+    assert "gsutil" not in source
+    assert 'payload["fail_count"] or payload["blocked_count"]' in source
+
+
+def test_run_json_rejects_permission_warning_even_when_gcloud_exits_zero(monkeypatch) -> None:
+    completed = subprocess.CompletedProcess(
+        ["gcloud", "compute", "instances", "list"],
+        0,
+        stdout="[]\n",
+        stderr="WARNING: Required 'compute.instances.list' permission for 'projects/teleo-501523'\n",
+    )
+    monkeypatch.setattr(checker.subprocess, "run", lambda *args, **kwargs: completed)
+
+    try:
+        checker.run_json(completed.args)
+    except checker.GcloudAccessError as exc:
+        assert "compute.instances.list" in str(exc)
+    else:  # pragma: no cover - assertion branch
+        raise AssertionError("permission warning was treated as an empty live inventory")
+
+
+def test_safe_check_classifies_access_error_as_blocked() -> None:
+    def denied() -> checker.Check:
+        raise checker.GcloudAccessError("PERMISSION_DENIED")
+
+    result = checker.safe_check("compute_runtime_service_accounts", denied)
+
+    assert result.status == "blocked"
+    assert result.current_tier == "T0_no_live_readback"
+    assert "PERMISSION_DENIED" in result.detail
+
+
+def test_restore_status_does_not_assert_unread_target_exists() -> None:
+    result = checker.check_kb_restore_or_replication()
+
+    assert result.status == "blocked"
+    assert "target existence must be established separately" in result.detail
+    assert "standby target exists" not in result.detail
+
+
+def test_canonical_postgres_restore_contract_is_complete() -> None:
+    result = checker.check_canonical_postgres_restore_contract()
+
+    assert result.status == "pass"
+    assert result.current_tier == "T2_canonical_restore_contract"
+
+
+def test_local_canonical_postgres_restore_preflight_is_retained() -> None:
+    result = checker.check_local_canonical_postgres_restore_preflight()
+
+    assert result.status == "pass"
+    assert "exported_snapshot_tables=39" in result.detail
+    assert "rows=52164" in result.detail
 
 
 def test_restore_canary_check_uses_newest_generated_at_not_filename(tmp_path, monkeypatch) -> None:
diff --git a/tests/test_verify_postgres_parity_manifest.py b/tests/test_verify_postgres_parity_manifest.py
new file mode 100644
index 0000000..dcf1e7b
--- /dev/null
+++ b/tests/test_verify_postgres_parity_manifest.py
@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+
+def manifest_rows(
+    *,
+    database: str,
+    row_count: int = 2,
+    rowset_md5: str = "abc",
+    elapsed_ms: float = 4.0,
+    server_address: str | None = None,
+    ssl: bool = False,
+):
+    return [
+        {
+            "kind": "identity",
+            "database": database,
+            "current_user": "postgres",
+            "server_version_num": 160014,
+            "transaction_read_only": "on",
+            "server_address": server_address,
+            "ssl": ssl,
+        },
+        {"kind": "schemas", "items": ["kb_stage", "public"]},
+        {"kind": "extensions", "items": [{"name": "plpgsql", "version": "1.0"}]},
+        {
+            "kind": "application_roles",
+            "items": [{"name": "kb_apply", "can_login": True, "superuser": False}],
+        },
+        {"kind": "columns", "items": [{"schema": "public", "table": "claims", "name": "id"}]},
+        {"kind": "constraints", "items": [{"schema": "public", "table": "claims", "name": "claims_pkey"}]},
+        {"kind": "indexes", "items": [{"schema": "public", "table": "claims", "name": "claims_pkey"}]},
+        {"kind": "sequences", "items": []},
+        {"kind": "views", "items": []},
+        {"kind": "functions", "items": []},
+        {"kind": "triggers", "items": []},
+        {"kind": "types", "items": []},
+        {"kind": "policies", "items": []},
+        {
+            "kind": "table",
+            "schema": "public",
+            "table": "claims",
+            "row_count": row_count,
+            "rowset_md5": rowset_md5,
+        },
+        {"kind": "performance", "query": "count_claims", "elapsed_ms": elapsed_ms, "observed_rows": row_count},
+    ]
+
+
+def write_jsonl(path: Path, rows) -> None:
+    path.write_text("\n".join(json.dumps(row, sort_keys=True) for row in rows) + "\n")
+
+
+def run_verifier(tmp_path: Path, source_rows, target_rows, *, scope: str = "local_restore", connectivity=None):
+    source = tmp_path / "source.jsonl"
+    target = tmp_path / "target.jsonl"
+    output = tmp_path / "result.json"
+    write_jsonl(source, source_rows)
+    write_jsonl(target, target_rows)
+    command = [
+        "python3",
+        "ops/verify_postgres_parity_manifest.py",
+        "--source",
+        str(source),
+        "--target",
+        str(target),
+        "--scope",
+        scope,
+        "--output",
+        str(output),
+    ]
+    if connectivity is not None:
+        connectivity_path = tmp_path / "connectivity.json"
+        connectivity_path.write_text(json.dumps(connectivity))
+        command.extend(["--connectivity-proof", str(connectivity_path)])
+    completed = subprocess.run(command, text=True, capture_output=True, check=False)
+    return completed, json.loads(output.read_text())
+
+
+def test_matching_local_manifests_pass(tmp_path: Path) -> None:
+    completed, payload = run_verifier(
+        tmp_path,
+        manifest_rows(database="teleo"),
+        manifest_rows(database="teleo_copy"),
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    assert payload["status"] == "pass"
+    assert payload["details"]["source_total_rows"] == 2
+    assert payload["private_connectivity"]["status"] == "not_applicable_local_restore"
+
+
+def test_row_hash_and_role_mismatches_fail(tmp_path: Path) -> None:
+    target = manifest_rows(database="teleo_copy", row_count=1, rowset_md5="different")
+    target[3]["items"] = []
+    completed, payload = run_verifier(tmp_path, manifest_rows(database="teleo"), target)
+
+    assert completed.returncode == 1
+    assert "table_content_mismatches=1" in payload["problems"]
+    assert "application_role_mismatches=1" in payload["problems"]
+
+
+def test_gcp_scope_requires_private_connectivity_receipt(tmp_path: Path) -> None:
+    completed, payload = run_verifier(
+        tmp_path,
+        manifest_rows(database="teleo"),
+        manifest_rows(database="teleo_gcp_copy"),
+        scope="gcp_staging",
+    )
+
+    assert completed.returncode == 1
+    assert "gcp_private_connectivity_proof_missing" in payload["problems"]
+
+
+def test_gcp_scope_accepts_matching_private_connectivity_receipt(tmp_path: Path) -> None:
+    completed, payload = run_verifier(
+        tmp_path,
+        manifest_rows(database="teleo"),
+        manifest_rows(database="teleo_gcp_copy", server_address="10.61.0.3", ssl=True),
+        scope="gcp_staging",
+        connectivity={
+            "status": "pass",
+            "private_connectivity": True,
+            "public_ip_disabled": True,
+            "source_compute": "teleo-staging-1",
+            "target_database": "teleo_gcp_copy",
+            "server_address": "10.61.0.3",
+            "ssl": True,
+        },
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    assert payload["status"] == "pass"
+    assert payload["private_connectivity"]["proof"]["source_compute"] == "teleo-staging-1"
+
+
+def test_gcp_scope_rejects_receipt_for_a_different_server(tmp_path: Path) -> None:
+    completed, payload = run_verifier(
+        tmp_path,
+        manifest_rows(database="teleo"),
+        manifest_rows(database="teleo_gcp_copy", server_address="10.61.0.3", ssl=True),
+        scope="gcp_staging",
+        connectivity={
+            "status": "pass",
+            "private_connectivity": True,
+            "public_ip_disabled": True,
+            "source_compute": "teleo-staging-1",
+            "target_database": "teleo_gcp_copy",
+            "server_address": "10.61.0.4",
+            "ssl": True,
+        },
+    )
+
+    assert completed.returncode == 1
+    assert "gcp_connectivity_server_address_mismatch" in payload["problems"]
+
+
+def test_gcp_scope_rejects_public_or_non_tls_target_manifest(tmp_path: Path) -> None:
+    completed, payload = run_verifier(
+        tmp_path,
+        manifest_rows(database="teleo"),
+        manifest_rows(database="teleo_gcp_copy", server_address="34.1.2.3", ssl=False),
+        scope="gcp_staging",
+        connectivity={
+            "status": "pass",
+            "private_connectivity": True,
+            "public_ip_disabled": True,
+            "source_compute": "teleo-staging-1",
+            "target_database": "teleo_gcp_copy",
+            "server_address": "34.1.2.3",
+            "ssl": False,
+        },
+    )
+
+    assert completed.returncode == 1
+    assert "gcp_target_manifest_server_address_not_rfc1918" in payload["problems"]
+    assert "gcp_target_manifest_tls_not_proven" in payload["problems"]
+    assert "gcp_connectivity_tls_not_proven" in payload["problems"]