Add GCP parity readiness probe

.agents/skills/teleo-gcp-parity-ops/SKILL.md

@@ -24,7 +24,12 @@ Handle GCP parity as its own lane, with precise auth/account/project facts and n
 - Expected project-access account: `billy@livingip.xyz`
 - Retained blocker: selected account cannot refresh access token non-interactively.
 - Other locally available accounts refresh but lack access to `teleo-501523`.
-- Current artifact: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-blocker-20260709.json`
+- Current probe artifact: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.json`
+- Prior auth-capable blocker artifact: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-blocker-20260709.json`
+
+The current probe may report `blocked_on_network_dns_before_auth_refresh` when the
+runner cannot resolve `oauth2.googleapis.com`; that is an environment/network
+gate before the prior Google account reauth or project-permission checks.
 
 Always refresh if cheap before making a current claim.
 
@@ -50,6 +55,7 @@ GCP parity needs its own:
 Allowed:
 
 - `gcloud config list` / account list readbacks,
+- `scripts/probe_gcp_db_parity.py`,
 - non-secret auth status checks,
 - project access probes,
 - read-only SQL or service discovery where auth works,

docs/reports/leo-working-state-20260709/current-truth-index.md

@@ -98,7 +98,9 @@ Use this file before making status claims. Prefer fresh VPS/GCP readbacks when c
 
 ## GCP
 
-- Current GCP blocker: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-blocker-20260709.json`
+- Current GCP parity probe: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.md`
+- Current GCP parity probe JSON: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.json`
+- Prior auth-capable GCP blocker: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-blocker-20260709.json`
 
 ## Images
 

docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.json

@@ -0,0 +1,180 @@
+{
+  "accounts_tested": [
+    {
+      "account": "billy@livingip.xyz",
+      "token_refresh": {
+        "command": [
+          "/opt/homebrew/bin/gcloud",
+          "auth",
+          "print-access-token",
+          "--account=billy@livingip.xyz"
+        ],
+        "ok": false,
+        "returncode": 1,
+        "stderr_excerpt": "ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError(\"<urllib3.connection.HTTPSConnection object at 0x107996710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)\"))",
+        "stdout_excerpt": "",
+        "stdout_json": null,
+        "stdout_redacted": true
+      }
+    },
+    {
+      "account": "billyattnmarket@gmail.com",
+      "token_refresh": {
+        "command": [
+          "/opt/homebrew/bin/gcloud",
+          "auth",
+          "print-access-token",
+          "--account=billyattnmarket@gmail.com"
+        ],
+        "ok": false,
+        "returncode": 1,
+        "stderr_excerpt": "ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError(\"<urllib3.connection.HTTPSConnection object at 0x10733e710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)\"))",
+        "stdout_excerpt": "",
+        "stdout_json": null,
+        "stdout_redacted": true
+      }
+    },
+    {
+      "account": "contact@attn.markets",
+      "token_refresh": {
+        "command": [
+          "/opt/homebrew/bin/gcloud",
+          "auth",
+          "print-access-token",
+          "--account=contact@attn.markets"
+        ],
+        "ok": false,
+        "returncode": 1,
+        "stderr_excerpt": "ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError(\"<urllib3.connection.HTTPSConnection object at 0x10789a710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)\"))",
+        "stdout_excerpt": "",
+        "stdout_json": null,
+        "stdout_redacted": true
+      }
+    },
+    {
+      "account": "valentin.meylan68@gmail.com",
+      "token_refresh": {
+        "command": [
+          "/opt/homebrew/bin/gcloud",
+          "auth",
+          "print-access-token",
+          "--account=valentin.meylan68@gmail.com"
+        ],
+        "ok": false,
+        "returncode": 1,
+        "stderr_excerpt": "ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError(\"<urllib3.connection.HTTPSConnection object at 0x10a1d2710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)\"))",
+        "stdout_excerpt": "",
+        "stdout_json": null,
+        "stdout_redacted": true
+      }
+    }
+  ],
+  "artifact": "gcp_db_parity_current_probe",
+  "attempted_routes": [
+    "static active_config/config_default readback",
+    "temporary writable copy of local gcloud config to avoid sandbox writes under ~/.config/gcloud",
+    "gcloud auth list without printing secrets",
+    "gcloud auth print-access-token for configured accounts with stdout redacted",
+    "gcloud projects describe teleo-501523",
+    "if project readback works: gcloud compute instances describe teleo-prod-1",
+    "if project readback works: gcloud sql instances list"
+  ],
+  "claim_ceiling": "This probe proves only current local gcloud/GCP read-only parity readiness. It does not prove VPS behavior, Telegram-visible behavior, production DB mutation, or GCP-hosted Leo correctness.",
+  "clear_CTA": "Run this probe from a network-enabled shell where oauth2.googleapis.com resolves, with `billy@livingip.xyz` authenticated for project `teleo-501523`. Then rerun `python3 scripts/probe_gcp_db_parity.py` and proceed only if it reports `ready_for_gcp_readonly_parity`.",
+  "current_canary": "gcloud account can read teleo-501523 project metadata plus target VM/Cloud SQL metadata",
+  "exact_gate": "blocked_on_network_dns_before_auth_refresh",
+  "expected": {
+    "gce_instance": "teleo-prod-1",
+    "gce_zone": "europe-west6-a",
+    "project": "teleo-501523",
+    "project_access_account": "billy@livingip.xyz"
+  },
+  "generated_at_utc": "2026-07-10T02:20:00.883398+00:00",
+  "local_gcloud": {
+    "auth_list": {
+      "command": [
+        "/opt/homebrew/bin/gcloud",
+        "auth",
+        "list",
+        "--format=json"
+      ],
+      "ok": true,
+      "returncode": 0,
+      "stderr_excerpt": "",
+      "stdout_excerpt": "[\n  {\n    \"account\": \"billy@livingip.xyz\",\n    \"status\": \"ACTIVE\"\n  },\n  {\n    \"account\": \"billyattnmarket@gmail.com\",\n    \"status\": \"\"\n  },\n  {\n    \"account\": \"contact@attn.markets\",\n    \"status\": \"\"\n  },\n  {\n    \"account\": \"valentin.meylan68@gmail.com\",\n    \"status\": \"\"\n  }\n]",
+      "stdout_json": [
+        {
+          "account": "billy@livingip.xyz",
+          "status": "ACTIVE"
+        },
+        {
+          "account": "billyattnmarket@gmail.com",
+          "status": ""
+        },
+        {
+          "account": "contact@attn.markets",
+          "status": ""
+        },
+        {
+          "account": "valentin.meylan68@gmail.com",
+          "status": ""
+        }
+      ],
+      "stdout_redacted": false
+    },
+    "config_list": {
+      "command": [
+        "/opt/homebrew/bin/gcloud",
+        "config",
+        "list",
+        "--format=json"
+      ],
+      "ok": false,
+      "returncode": 1,
+      "stderr_excerpt": "ERROR: (gcloud.config.list) There was a problem refreshing your current auth tokens: HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError(\"<urllib3.connection.HTTPSConnection object at 0x108187ed0>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)\"))\nPlease run:\n\n  $ gcloud auth login\n\nto obtain new credentials.\n\nIf you have already logged in with a different account, run:\n\n  $ gcloud config set account ACCOUNT\n\nto select an already authenticated account to use.",
+      "stdout_excerpt": "",
+      "stdout_json": null,
+      "stdout_redacted": false
+    },
+    "gcloud_path": "/opt/homebrew/bin/gcloud",
+    "source_config": "/Users/user/.config/gcloud",
+    "static_active_config": {
+      "account": "billy@livingip.xyz",
+      "active_config": "default",
+      "config_file": "/private/tmp/teleo-gcp-parity-3jp00f00/configurations/config_default",
+      "project": "teleo-501523"
+    },
+    "temporary_config_removed_after_probe": true,
+    "used_temporary_writable_config_copy": true
+  },
+  "mode": "read_only_gcloud_parity_probe",
+  "mutates_db": false,
+  "mutates_gcp": false,
+  "next_non_user_action": [
+    "Run this same probe again and require ready_for_gcp_readonly_parity=true.",
+    "SSH into teleo-prod-1 or use Cloud SQL read-only access for DB/runtime parity proof.",
+    "Run the Working Leo benchmark against GCP only after read-only GCP metadata is proven current."
+  ],
+  "project_readback": {
+    "command": [
+      "/opt/homebrew/bin/gcloud",
+      "projects",
+      "describe",
+      "teleo-501523",
+      "--format=json"
+    ],
+    "ok": false,
+    "returncode": 1,
+    "stderr_excerpt": "ERROR: (gcloud.projects.describe) There was a problem refreshing your current auth tokens: HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError(\"<urllib3.connection.HTTPSConnection object at 0x107ac0550>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)\"))\nPlease run:\n\n  $ gcloud auth login\n\nto obtain new credentials.\n\nIf you have already logged in with a different account, run:\n\n  $ gcloud config set account ACCOUNT\n\nto select an already authenticated account to use.",
+    "stdout_excerpt": "",
+    "stdout_json": null,
+    "stdout_redacted": false,
+    "summary": null
+  },
+  "ready_for_gcp_readonly_parity": false,
+  "runtime_readbacks": {
+    "cloudsql_instances": null,
+    "gce_instance": null
+  },
+  "status": "blocked_on_network_dns_before_auth_refresh"
+}

docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.md

@@ -0,0 +1,57 @@
+# GCP DB Parity Current Probe
+
+Generated UTC: `2026-07-10T02:20:00.883398+00:00`
+Status: `blocked_on_network_dns_before_auth_refresh`
+Ready for GCP read-only parity: `False`
+Mutates GCP: `False`
+Mutates DB: `False`
+Current canary: gcloud account can read teleo-501523 project metadata plus target VM/Cloud SQL metadata
+
+## Active Config
+
+- Account: `billy@livingip.xyz`
+- Project: `teleo-501523`
+- Temporary writable config copy: `True`
+- Temporary config removed after probe: `True`
+
+## Account Checks
+
+- `billy@livingip.xyz`: `ok=False`, `returncode=1`, `stdout_redacted=True`
+  - stderr: `ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError("<urllib3.connection.HTTPSConnection object at 0x107996710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)"))`
+- `billyattnmarket@gmail.com`: `ok=False`, `returncode=1`, `stdout_redacted=True`
+  - stderr: `ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError("<urllib3.connection.HTTPSConnection object at 0x10733e710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)"))`
+- `contact@attn.markets`: `ok=False`, `returncode=1`, `stdout_redacted=True`
+  - stderr: `ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError("<urllib3.connection.HTTPSConnection object at 0x10789a710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)"))`
+- `valentin.meylan68@gmail.com`: `ok=False`, `returncode=1`, `stdout_redacted=True`
+  - stderr: `ERROR: (gcloud.auth.print-access-token) HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError("<urllib3.connection.HTTPSConnection object at 0x10a1d2710>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)"))`
+
+## Project Readback
+
+- `ok=False`, `returncode=1`
+- stderr: `ERROR: (gcloud.projects.describe) There was a problem refreshing your current auth tokens: HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NameResolutionError("<urllib3.connection.HTTPSConnection object at 0x107ac0550>: Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided, or not known)"))
+Please run:
+
+  $ gcloud auth login
+
+to obtain new credentials.
+
+If you have already logged in with a different account, run:
+
+  $ gcloud config set account ACCOUNT
+
+to select an already authenticated account to use.`
+
+## Gate
+
+- Exact gate: `blocked_on_network_dns_before_auth_refresh`
+- Clear CTA: Run this probe from a network-enabled shell where oauth2.googleapis.com resolves, with `billy@livingip.xyz` authenticated for project `teleo-501523`. Then rerun `python3 scripts/probe_gcp_db_parity.py` and proceed only if it reports `ready_for_gcp_readonly_parity`.
+
+## Next Non-User Action
+
+- Run this same probe again and require ready_for_gcp_readonly_parity=true.
+- SSH into teleo-prod-1 or use Cloud SQL read-only access for DB/runtime parity proof.
+- Run the Working Leo benchmark against GCP only after read-only GCP metadata is proven current.
+
+## Claim Ceiling
+
+This probe proves only current local gcloud/GCP read-only parity readiness. It does not prove VPS behavior, Telegram-visible behavior, production DB mutation, or GCP-hosted Leo correctness.

docs/reports/leo-working-state-20260709/operator-surface-map.md

@@ -33,8 +33,9 @@ Always refresh before claiming current state.
 
 - Project: `teleo-501523`
 - Expected active account for project access: `billy@livingip.xyz`
-- Current retained blocker: `billy@livingip.xyz` selected but token refresh fails non-interactively; other locally available accounts refresh but lack project access.
+- Current retained probe: local gcloud config still selects `billy@livingip.xyz` and `teleo-501523`, but this runner cannot resolve `oauth2.googleapis.com`, so parity is blocked before token refresh/project access can be rechecked.
-- Artifact: `outputs/gcp-db-parity-current-blocker-20260709.json`
+- Current artifact: `docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.json`
+- Prior auth-capable artifact: `outputs/gcp-db-parity-current-blocker-20260709.json`
 
 ## Telegram
 

docs/reports/leo-working-state-20260709/working-leo-current-state-20260709.md

@@ -188,11 +188,11 @@ A working Leo, for Cory/m3taversal's current expectation, is a Telegram-facing a
   - `teleo-kb-apply-worker.service` exists but is disabled/inactive and ships inert/report-only unless `KB_APPLY_WORKER_ENABLED=1`; no active `KB_APPLY_RENDER_CMD` hook or scheduled 24-hour SOUL renderer was proven
   - current VPS schema does not contain the designed decision-matrix tables `kb_stage.matrix_voters`, `kb_stage.proposal_votes`, or `kb_stage.proposal_decisions`; current approval evidence lives in `kb_stage.kb_proposals`
   - live proposal inspection showed pending/approved rows have `source_ref` pointers, but inspected source refs did not directly match `public.sources`; document proposals are real staging rows but still need guarded mapping into canonical sources/evidence/edges
-- GCP parity was rechecked and remains blocked on account authentication:
+- GCP parity was rechecked with a repeatable current probe and remains not proven:
-  - `billy@livingip.xyz` is selected for project `teleo-501523` but token refresh fails non-interactively
+  - current local config still selects `billy@livingip.xyz` for project `teleo-501523`
-  - `billyattnmarket@gmail.com` and `valentin.meylan68@gmail.com` can refresh tokens but do not have access to `teleo-501523`
+  - the current Codex runner cannot resolve `oauth2.googleapis.com`, so all token refresh/project readbacks stop at `blocked_on_network_dns_before_auth_refresh`
-  - `contact@attn.markets` also fails token refresh
+  - the prior auth-capable July 9 artifact remains useful context: `billy@livingip.xyz` then failed non-interactive reauth, two other locally cached accounts refreshed but lacked `teleo-501523`, and `contact@attn.markets` failed token refresh
-  - active gcloud account was restored to `billy@livingip.xyz`
+  - no GCP parity claim is valid until `scripts/probe_gcp_db_parity.py` reports `ready_for_gcp_readonly_parity=true`
 
 ## Current Evidence
 
@@ -256,7 +256,8 @@ A working Leo, for Cory/m3taversal's current expectation, is a Telegram-facing a
 - Governance/concept-map schema proof image: `outputs/leo-db-state-24-governance-concept-schema.svg`
 - Integrated apply rehearsal proof image: `outputs/leo-db-state-25-integrated-apply-rehearsal.svg`
 - Direct-claim handler pass image: `work/teleo-infra-main/docs/reports/leo-working-state-20260709/leo-db-state-27-direct-claim-handler-pass.svg`
-- GCP parity blocker: `outputs/gcp-db-parity-current-blocker-20260709.json`
+- Current GCP parity probe: `work/teleo-infra-main/docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.md`
+- Prior GCP parity blocker: `outputs/gcp-db-parity-current-blocker-20260709.json`
 
 ## Next Actions
 

docs/reports/leo-working-state-20260709/working-leo-execution-plan-current.md

@@ -130,10 +130,11 @@ Claim ceiling: VPS Telegram memory/KB audit, full live-safe open-ended Telegram
   - `/Users/user/Documents/Codex/2026-07-09/019f34eb-d297-72d0-b7e2-b222d5515ab9-load/outputs/rich-proposal-creation-plan-20260709/staging-db-mapped-commit-rehearsal-current.log`
 
 ## WL-EXEC-09 - GCP parity
-- Status: `blocked_on_gcloud_reauth`
+- Status: `blocked_on_network_dns_before_auth_refresh`
-- Result: Rechecked local `gcloud`: `billy@livingip.xyz` remains selected for `teleo-501523` but cannot refresh tokens non-interactively; two other locally cached accounts can refresh but lack project access; active account restored to `billy@livingip.xyz`.
+- Result: Fresh current probe uses a temporary writable gcloud config copy and keeps token stdout redacted. Local config still selects `billy@livingip.xyz` for `teleo-501523`, but every token/project readback fails at DNS resolution for `oauth2.googleapis.com`; this runner cannot reach the prior auth/project-access gate.
-- Not done condition: No GCP parity claim until an account with `teleo-501523` access can refresh and run read-only GCP VM/SQL/service readbacks.
+- Not done condition: No GCP parity claim until `scripts/probe_gcp_db_parity.py` reports `ready_for_gcp_readonly_parity=true` and then read-only VM/Cloud SQL/runtime parity is captured.
 - Evidence:
+  - `/Users/user/Documents/Codex/2026-07-09/019f34eb-d297-72d0-b7e2-b222d5515ab9-load/work/teleo-infra-main/docs/reports/leo-working-state-20260709/gcp-db-parity-current-probe-current.md`
   - `/Users/user/Documents/Codex/2026-07-09/019f34eb-d297-72d0-b7e2-b222d5515ab9-load/outputs/gcp-db-parity-current-blocker-20260709.json`
 
 ## WL-EXEC-10 - Cleanup and retained evidence

scripts/probe_gcp_db_parity.py

@@ -0,0 +1,532 @@
+#!/usr/bin/env python3
+"""Probe current GCP DB parity readiness without mutating GCP.
+
+The probe is intentionally read-only. It copies the local gcloud config into a
+temporary writable directory so sandboxed agents can run gcloud without writing
+under ~/.config/gcloud, redirects access-token output out of artifacts, and
+keeps GCP parity separate from VPS proof.
+"""
+
+from __future__ import annotations
+
+import argparse
+import configparser
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+
+REPORT_DIR = Path("docs/reports/leo-working-state-20260709")
+DEFAULT_OUT = REPORT_DIR / "gcp-db-parity-current-probe-current.json"
+DEFAULT_MARKDOWN_OUT = REPORT_DIR / "gcp-db-parity-current-probe-current.md"
+DEFAULT_GCLOUD = Path("/opt/homebrew/bin/gcloud")
+DEFAULT_SOURCE_CONFIG = Path.home() / ".config" / "gcloud"
+DEFAULT_PROJECT = "teleo-501523"
+DEFAULT_INSTANCE = "teleo-prod-1"
+DEFAULT_ZONE = "europe-west6-a"
+EXPECTED_ACCESS_ACCOUNT = "billy@livingip.xyz"
+
+
+@dataclass(frozen=True)
+class CommandResult:
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+Runner = Callable[[list[str], dict[str, str], int], CommandResult]
+
+
+def subprocess_runner(command: list[str], env: dict[str, str], timeout: int) -> CommandResult:
+    proc = subprocess.run(command, text=True, capture_output=True, env=env, timeout=timeout, check=False)
+    return CommandResult(proc.returncode, proc.stdout, proc.stderr)
+
+
+def _json_or_none(text: str) -> Any:
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return None
+
+
+def _excerpt(text: str, limit: int = 1200) -> str:
+    text = text.strip()
+    if len(text) <= limit:
+        return text
+    return text[:limit] + "...[truncated]"
+
+
+def _run(
+    command: list[str],
+    *,
+    env: dict[str, str],
+    runner: Runner,
+    timeout: int,
+    redact_stdout: bool = False,
+) -> dict[str, Any]:
+    try:
+        result = runner(command, env, timeout)
+        stdout = "" if redact_stdout else result.stdout
+        return {
+            "command": redact_command(command),
+            "returncode": result.returncode,
+            "ok": result.returncode == 0,
+            "stdout_redacted": redact_stdout,
+            "stdout_excerpt": _excerpt(stdout) if stdout else "",
+            "stderr_excerpt": _excerpt(result.stderr),
+            "stdout_json": None if redact_stdout else _json_or_none(result.stdout),
+        }
+    except FileNotFoundError as exc:
+        return {
+            "command": redact_command(command),
+            "returncode": 127,
+            "ok": False,
+            "stdout_redacted": redact_stdout,
+            "stdout_excerpt": "",
+            "stderr_excerpt": str(exc),
+            "stdout_json": None,
+        }
+    except subprocess.TimeoutExpired as exc:
+        return {
+            "command": redact_command(command),
+            "returncode": 124,
+            "ok": False,
+            "stdout_redacted": redact_stdout,
+            "stdout_excerpt": "",
+            "stderr_excerpt": f"timed out after {exc.timeout}s",
+            "stdout_json": None,
+        }
+
+
+def redact_command(command: list[str]) -> list[str]:
+    redacted: list[str] = []
+    for part in command:
+        if part.startswith("--account="):
+            redacted.append(part)
+        else:
+            redacted.append(part)
+    return redacted
+
+
+def copy_gcloud_config(source: Path) -> tuple[tempfile.TemporaryDirectory[str] | None, Path, bool]:
+    if not source.exists():
+        return None, source, False
+    tmp = tempfile.TemporaryDirectory(prefix="teleo-gcp-parity-", dir="/private/tmp")
+    target = Path(tmp.name)
+
+    def ignore(_: str, names: list[str]) -> set[str]:
+        return {"logs"} & set(names)
+
+    shutil.copytree(source, target, dirs_exist_ok=True, ignore=ignore)
+    return tmp, target, True
+
+
+def read_static_config(config_dir: Path) -> dict[str, Any]:
+    active_name = "default"
+    active_file = config_dir / "active_config"
+    if active_file.exists():
+        active_name = active_file.read_text(encoding="utf-8").strip() or "default"
+    config_file = config_dir / "configurations" / f"config_{active_name}"
+    parser = configparser.ConfigParser()
+    parser.read(config_file)
+    core = parser["core"] if parser.has_section("core") else {}
+    return {
+        "active_config": active_name,
+        "config_file": str(config_file),
+        "account": core.get("account"),
+        "project": core.get("project"),
+    }
+
+
+def accounts_from_auth_list(auth_list: dict[str, Any], fallback_account: str | None) -> list[str]:
+    parsed = auth_list.get("stdout_json")
+    accounts: list[str] = []
+    if isinstance(parsed, list):
+        for item in parsed:
+            if isinstance(item, dict) and item.get("account"):
+                accounts.append(str(item["account"]))
+    if fallback_account and fallback_account not in accounts:
+        accounts.insert(0, fallback_account)
+    return accounts
+
+
+def has_network_dns_error(results: list[dict[str, Any]]) -> bool:
+    needle_set = ("Failed to resolve", "NameResolutionError", "nodename nor servname", "oauth2.googleapis.com")
+    return any(any(needle in item.get("stderr_excerpt", "") for needle in needle_set) for item in results)
+
+
+def has_reauth_error(results: list[dict[str, Any]]) -> bool:
+    needle_set = ("Reauthentication failed", "cannot prompt during non-interactive execution", "gcloud auth login")
+    return any(any(needle in item.get("stderr_excerpt", "") for needle in needle_set) for item in results)
+
+
+def summarize_project(project_result: dict[str, Any]) -> dict[str, Any] | None:
+    parsed = project_result.get("stdout_json")
+    if not isinstance(parsed, dict):
+        return None
+    return {
+        "projectId": parsed.get("projectId"),
+        "projectNumber": parsed.get("projectNumber"),
+        "lifecycleState": parsed.get("lifecycleState"),
+    }
+
+
+def summarize_instance(instance_result: dict[str, Any]) -> dict[str, Any] | None:
+    parsed = instance_result.get("stdout_json")
+    if not isinstance(parsed, dict):
+        return None
+    return {
+        "name": parsed.get("name"),
+        "status": parsed.get("status"),
+        "zone": parsed.get("zone"),
+        "machineType": parsed.get("machineType"),
+        "serviceAccounts": [
+            {"email": item.get("email"), "scopes": item.get("scopes", [])}
+            for item in parsed.get("serviceAccounts", [])
+            if isinstance(item, dict)
+        ],
+    }
+
+
+def summarize_sql_instances(sql_result: dict[str, Any]) -> list[dict[str, Any]]:
+    parsed = sql_result.get("stdout_json")
+    if not isinstance(parsed, list):
+        return []
+    summary: list[dict[str, Any]] = []
+    for item in parsed:
+        if isinstance(item, dict):
+            summary.append(
+                {
+                    "name": item.get("name"),
+                    "state": item.get("state"),
+                    "databaseVersion": item.get("databaseVersion"),
+                    "region": item.get("region"),
+                }
+            )
+    return summary
+
+
+def status_from_results(
+    *,
+    selected_account: str | None,
+    expected_account: str,
+    project: str | None,
+    expected_project: str,
+    token_checks: list[dict[str, Any]],
+    project_readback: dict[str, Any],
+    instance_readback: dict[str, Any] | None,
+    sql_readback: dict[str, Any] | None,
+) -> str:
+    command_results = [*token_checks, project_readback]
+    if instance_readback is not None:
+        command_results.append(instance_readback)
+    if sql_readback is not None:
+        command_results.append(sql_readback)
+    if has_network_dns_error(command_results):
+        return "blocked_on_network_dns_before_auth_refresh"
+    if selected_account != expected_account:
+        return "blocked_on_wrong_active_account"
+    if project != expected_project:
+        return "blocked_on_wrong_active_project"
+    selected_token = next((item for item in token_checks if item["account"] == selected_account), None)
+    if selected_token and not selected_token["token_refresh"]["ok"]:
+        return "blocked_on_gcloud_reauth" if has_reauth_error([selected_token["token_refresh"]]) else "blocked_on_token_refresh"
+    if not project_readback["ok"]:
+        return "blocked_on_project_access"
+    if instance_readback is not None and not instance_readback["ok"]:
+        return "blocked_on_gce_instance_readback"
+    if sql_readback is not None and not sql_readback["ok"]:
+        return "blocked_on_cloudsql_readback"
+    return "ready_for_gcp_readonly_parity"
+
+
+def probe_gcp_parity(
+    *,
+    gcloud_path: Path = DEFAULT_GCLOUD,
+    source_config: Path = DEFAULT_SOURCE_CONFIG,
+    expected_project: str = DEFAULT_PROJECT,
+    expected_account: str = EXPECTED_ACCESS_ACCOUNT,
+    instance: str = DEFAULT_INSTANCE,
+    zone: str = DEFAULT_ZONE,
+    runner: Runner = subprocess_runner,
+    use_temp_config: bool = True,
+    timeout: int = 30,
+) -> dict[str, Any]:
+    temp_dir: tempfile.TemporaryDirectory[str] | None = None
+    temp_config_created = False
+    config_dir = source_config
+    try:
+        if use_temp_config:
+            temp_dir, config_dir, temp_config_created = copy_gcloud_config(source_config)
+        static_config = read_static_config(config_dir)
+        env = os.environ.copy()
+        env["CLOUDSDK_CONFIG"] = str(config_dir)
+        env["CLOUDSDK_CORE_DISABLE_PROMPTS"] = "1"
+
+        config_list = _run(
+            [str(gcloud_path), "config", "list", "--format=json"],
+            env=env,
+            runner=runner,
+            timeout=timeout,
+        )
+        auth_list = _run(
+            [str(gcloud_path), "auth", "list", "--format=json"],
+            env=env,
+            runner=runner,
+            timeout=timeout,
+        )
+        accounts = accounts_from_auth_list(auth_list, static_config.get("account"))
+        token_checks = []
+        for account in accounts:
+            token_checks.append(
+                {
+                    "account": account,
+                    "token_refresh": _run(
+                        [str(gcloud_path), "auth", "print-access-token", f"--account={account}"],
+                        env=env,
+                        runner=runner,
+                        timeout=timeout,
+                        redact_stdout=True,
+                    ),
+                }
+            )
+
+        project_readback = _run(
+            [str(gcloud_path), "projects", "describe", expected_project, "--format=json"],
+            env=env,
+            runner=runner,
+            timeout=timeout,
+        )
+
+        instance_readback: dict[str, Any] | None = None
+        sql_readback: dict[str, Any] | None = None
+        if project_readback["ok"]:
+            instance_readback = _run(
+                [
+                    str(gcloud_path),
+                    "compute",
+                    "instances",
+                    "describe",
+                    instance,
+                    "--zone",
+                    zone,
+                    "--project",
+                    expected_project,
+                    "--format=json",
+                ],
+                env=env,
+                runner=runner,
+                timeout=timeout,
+            )
+            sql_readback = _run(
+                [str(gcloud_path), "sql", "instances", "list", "--project", expected_project, "--format=json"],
+                env=env,
+                runner=runner,
+                timeout=timeout,
+            )
+
+        status = status_from_results(
+            selected_account=static_config.get("account"),
+            expected_account=expected_account,
+            project=static_config.get("project"),
+            expected_project=expected_project,
+            token_checks=token_checks,
+            project_readback=project_readback,
+            instance_readback=instance_readback,
+            sql_readback=sql_readback,
+        )
+        ready = status == "ready_for_gcp_readonly_parity"
+        clear_cta = (
+            "Run this probe from a network-enabled shell where oauth2.googleapis.com resolves, with "
+            f"`{expected_account}` authenticated for project `{expected_project}`. Then rerun "
+            "`python3 scripts/probe_gcp_db_parity.py` and proceed only if it reports "
+            "`ready_for_gcp_readonly_parity`."
+        )
+        if status == "blocked_on_gcloud_reauth":
+            clear_cta = (
+                f"In a local terminal, run `gcloud auth login {expected_account}` and complete Google "
+                f"reauth for project `{expected_project}`. Then rerun "
+                "`python3 scripts/probe_gcp_db_parity.py`."
+            )
+        elif status == "blocked_on_wrong_active_account":
+            clear_cta = (
+                f"Run `gcloud config set account {expected_account}` in the active gcloud config, then rerun "
+                "`python3 scripts/probe_gcp_db_parity.py`."
+            )
+        elif status == "blocked_on_project_access":
+            clear_cta = (
+                f"Grant `{static_config.get('account')}` read access to project `{expected_project}` or switch "
+                "to an account that already has it, then rerun `python3 scripts/probe_gcp_db_parity.py`."
+            )
+
+        return {
+            "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+            "artifact": "gcp_db_parity_current_probe",
+            "mode": "read_only_gcloud_parity_probe",
+            "status": status,
+            "ready_for_gcp_readonly_parity": ready,
+            "mutates_gcp": False,
+            "mutates_db": False,
+            "current_canary": "gcloud account can read teleo-501523 project metadata plus target VM/Cloud SQL metadata",
+            "expected": {
+                "project": expected_project,
+                "project_access_account": expected_account,
+                "gce_instance": instance,
+                "gce_zone": zone,
+            },
+            "local_gcloud": {
+                "gcloud_path": str(gcloud_path),
+                "source_config": str(source_config),
+                "used_temporary_writable_config_copy": temp_config_created,
+                "temporary_config_removed_after_probe": True,
+                "static_active_config": static_config,
+                "config_list": config_list,
+                "auth_list": auth_list,
+            },
+            "accounts_tested": token_checks,
+            "project_readback": {
+                **project_readback,
+                "summary": summarize_project(project_readback),
+            },
+            "runtime_readbacks": {
+                "gce_instance": None
+                if instance_readback is None
+                else {**instance_readback, "summary": summarize_instance(instance_readback)},
+                "cloudsql_instances": None
+                if sql_readback is None
+                else {**sql_readback, "summary": summarize_sql_instances(sql_readback)},
+            },
+            "attempted_routes": [
+                "static active_config/config_default readback",
+                "temporary writable copy of local gcloud config to avoid sandbox writes under ~/.config/gcloud",
+                "gcloud auth list without printing secrets",
+                "gcloud auth print-access-token for configured accounts with stdout redacted",
+                "gcloud projects describe teleo-501523",
+                "if project readback works: gcloud compute instances describe teleo-prod-1",
+                "if project readback works: gcloud sql instances list",
+            ],
+            "exact_gate": status,
+            "clear_CTA": clear_cta,
+            "next_non_user_action": [
+                "Run this same probe again and require ready_for_gcp_readonly_parity=true.",
+                "SSH into teleo-prod-1 or use Cloud SQL read-only access for DB/runtime parity proof.",
+                "Run the Working Leo benchmark against GCP only after read-only GCP metadata is proven current.",
+            ],
+            "claim_ceiling": (
+                "This probe proves only current local gcloud/GCP read-only parity readiness. It does not prove "
+                "VPS behavior, Telegram-visible behavior, production DB mutation, or GCP-hosted Leo correctness."
+            ),
+        }
+    finally:
+        if temp_dir is not None:
+            temp_dir.cleanup()
+
+
+def write_json(path: Path, data: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def write_markdown(path: Path, data: dict[str, Any]) -> None:
+    lines = [
+        "# GCP DB Parity Current Probe",
+        "",
+        f"Generated UTC: `{data['generated_at_utc']}`",
+        f"Status: `{data['status']}`",
+        f"Ready for GCP read-only parity: `{data['ready_for_gcp_readonly_parity']}`",
+        f"Mutates GCP: `{data['mutates_gcp']}`",
+        f"Mutates DB: `{data['mutates_db']}`",
+        f"Current canary: {data['current_canary']}",
+        "",
+        "## Active Config",
+        "",
+        f"- Account: `{data['local_gcloud']['static_active_config'].get('account')}`",
+        f"- Project: `{data['local_gcloud']['static_active_config'].get('project')}`",
+        f"- Temporary writable config copy: `{data['local_gcloud']['used_temporary_writable_config_copy']}`",
+        f"- Temporary config removed after probe: `{data['local_gcloud']['temporary_config_removed_after_probe']}`",
+        "",
+        "## Account Checks",
+        "",
+    ]
+    for item in data["accounts_tested"]:
+        token = item["token_refresh"]
+        lines.append(
+            f"- `{item['account']}`: `ok={token['ok']}`, `returncode={token['returncode']}`, "
+            f"`stdout_redacted={token['stdout_redacted']}`"
+        )
+        if token["stderr_excerpt"]:
+            lines.append(f"  - stderr: `{token['stderr_excerpt']}`")
+    project = data["project_readback"]
+    lines.extend(
+        [
+            "",
+            "## Project Readback",
+            "",
+            f"- `ok={project['ok']}`, `returncode={project['returncode']}`",
+        ]
+    )
+    if project.get("summary"):
+        lines.append(f"- Summary: `{project['summary']}`")
+    if project["stderr_excerpt"]:
+        lines.append(f"- stderr: `{project['stderr_excerpt']}`")
+    lines.extend(["", "## Gate", "", f"- Exact gate: `{data['exact_gate']}`", f"- Clear CTA: {data['clear_CTA']}"])
+    lines.extend(["", "## Next Non-User Action", ""])
+    lines.extend(f"- {item}" for item in data["next_non_user_action"])
+    lines.extend(["", "## Claim Ceiling", "", data["claim_ceiling"], ""])
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--gcloud", type=Path, default=DEFAULT_GCLOUD)
+    parser.add_argument("--source-config", type=Path, default=DEFAULT_SOURCE_CONFIG)
+    parser.add_argument("--project", default=DEFAULT_PROJECT)
+    parser.add_argument("--expected-account", default=EXPECTED_ACCESS_ACCOUNT)
+    parser.add_argument("--instance", default=DEFAULT_INSTANCE)
+    parser.add_argument("--zone", default=DEFAULT_ZONE)
+    parser.add_argument("--timeout", type=int, default=30)
+    parser.add_argument("--no-temp-config-copy", action="store_true")
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
+    parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    data = probe_gcp_parity(
+        gcloud_path=args.gcloud,
+        source_config=args.source_config,
+        expected_project=args.project,
+        expected_account=args.expected_account,
+        instance=args.instance,
+        zone=args.zone,
+        use_temp_config=not args.no_temp_config_copy,
+        timeout=args.timeout,
+    )
+    write_json(args.out, data)
+    write_markdown(args.markdown_out, data)
+    print(
+        json.dumps(
+            {
+                "out": str(args.out),
+                "markdown_out": str(args.markdown_out),
+                "status": data["status"],
+                "ready_for_gcp_readonly_parity": data["ready_for_gcp_readonly_parity"],
+                "mutates_gcp": data["mutates_gcp"],
+                "mutates_db": data["mutates_db"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0 if data["ready_for_gcp_readonly_parity"] else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_probe_gcp_db_parity.py

@@ -0,0 +1,170 @@
+"""Tests for scripts/probe_gcp_db_parity.py."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import probe_gcp_db_parity as probe  # noqa: E402
+
+
+def write_gcloud_config(root: Path, *, account: str = "billy@livingip.xyz", project: str = "teleo-501523") -> None:
+    (root / "configurations").mkdir(parents=True)
+    (root / "active_config").write_text("default\n", encoding="utf-8")
+    (root / "configurations" / "config_default").write_text(
+        f"[core]\naccount = {account}\nproject = {project}\n",
+        encoding="utf-8",
+    )
+
+
+def test_dns_gate_uses_static_config_and_redacts_token_stdout(tmp_path: Path):
+    config_dir = tmp_path / "gcloud"
+    write_gcloud_config(config_dir)
+    accounts = [
+        {"account": "billy@livingip.xyz", "status": "ACTIVE"},
+        {"account": "billyattnmarket@gmail.com", "status": ""},
+    ]
+
+    def fake_runner(command: list[str], _env: dict[str, str], _timeout: int) -> probe.CommandResult:
+        if command[1:3] == ["auth", "list"]:
+            return probe.CommandResult(0, json.dumps(accounts), "")
+        if command[1:3] == ["auth", "print-access-token"]:
+            return probe.CommandResult(
+                1,
+                "secret-token-should-not-leak",
+                "Failed to resolve 'oauth2.googleapis.com' ([Errno 8] nodename nor servname provided)",
+            )
+        return probe.CommandResult(
+            1,
+            "",
+            "HTTPSConnectionPool(host='oauth2.googleapis.com'): Failed to resolve 'oauth2.googleapis.com'",
+        )
+
+    result = probe.probe_gcp_parity(
+        gcloud_path=Path("/fake/gcloud"),
+        source_config=config_dir,
+        runner=fake_runner,
+        use_temp_config=False,
+    )
+
+    assert result["status"] == "blocked_on_network_dns_before_auth_refresh"
+    assert result["ready_for_gcp_readonly_parity"] is False
+    assert result["mutates_gcp"] is False
+    assert result["mutates_db"] is False
+    assert result["local_gcloud"]["static_active_config"]["account"] == "billy@livingip.xyz"
+    assert result["local_gcloud"]["static_active_config"]["project"] == "teleo-501523"
+    assert result["accounts_tested"][0]["token_refresh"]["stdout_redacted"] is True
+    assert "secret-token" not in json.dumps(result)
+    assert "oauth2.googleapis.com" in result["clear_CTA"]
+
+
+def test_ready_status_requires_project_vm_and_cloudsql_readbacks(tmp_path: Path):
+    config_dir = tmp_path / "gcloud"
+    write_gcloud_config(config_dir)
+
+    def fake_runner(command: list[str], _env: dict[str, str], _timeout: int) -> probe.CommandResult:
+        suffix = command[1:]
+        if suffix[:2] == ["auth", "list"]:
+            return probe.CommandResult(0, json.dumps([{"account": "billy@livingip.xyz", "status": "ACTIVE"}]), "")
+        if suffix[:2] == ["auth", "print-access-token"]:
+            return probe.CommandResult(0, "secret-token-should-not-leak", "")
+        if suffix[:2] == ["projects", "describe"]:
+            return probe.CommandResult(
+                0,
+                json.dumps(
+                    {
+                        "projectId": "teleo-501523",
+                        "projectNumber": "123456789",
+                        "lifecycleState": "ACTIVE",
+                    }
+                ),
+                "",
+            )
+        if suffix[:3] == ["compute", "instances", "describe"]:
+            return probe.CommandResult(
+                0,
+                json.dumps(
+                    {
+                        "name": "teleo-prod-1",
+                        "status": "RUNNING",
+                        "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a",
+                        "machineType": "e2-standard-2",
+                        "serviceAccounts": [{"email": "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com"}],
+                    }
+                ),
+                "",
+            )
+        if suffix[:3] == ["sql", "instances", "list"]:
+            return probe.CommandResult(
+                0,
+                json.dumps(
+                    [
+                        {
+                            "name": "teleo-prod-pgvector",
+                            "state": "RUNNABLE",
+                            "databaseVersion": "POSTGRES_16",
+                            "region": "europe-west6",
+                        }
+                    ]
+                ),
+                "",
+            )
+        return probe.CommandResult(0, "{}", "")
+
+    result = probe.probe_gcp_parity(
+        gcloud_path=Path("/fake/gcloud"),
+        source_config=config_dir,
+        runner=fake_runner,
+        use_temp_config=False,
+    )
+
+    assert result["status"] == "ready_for_gcp_readonly_parity"
+    assert result["ready_for_gcp_readonly_parity"] is True
+    assert result["project_readback"]["summary"]["projectId"] == "teleo-501523"
+    assert result["runtime_readbacks"]["gce_instance"]["summary"]["status"] == "RUNNING"
+    assert result["runtime_readbacks"]["cloudsql_instances"]["summary"][0]["name"] == "teleo-prod-pgvector"
+    assert "secret-token" not in json.dumps(result)
+
+
+def test_markdown_names_gate_and_cta(tmp_path: Path):
+    report = {
+        "generated_at_utc": "2026-07-10T00:00:00+00:00",
+        "status": "blocked_on_network_dns_before_auth_refresh",
+        "ready_for_gcp_readonly_parity": False,
+        "mutates_gcp": False,
+        "mutates_db": False,
+        "current_canary": "gcloud account can read teleo-501523",
+        "local_gcloud": {
+            "static_active_config": {"account": "billy@livingip.xyz", "project": "teleo-501523"},
+            "used_temporary_writable_config_copy": True,
+            "temporary_config_removed_after_probe": True,
+        },
+        "accounts_tested": [
+            {
+                "account": "billy@livingip.xyz",
+                "token_refresh": {
+                    "ok": False,
+                    "returncode": 1,
+                    "stdout_redacted": True,
+                    "stderr_excerpt": "Failed to resolve oauth2.googleapis.com",
+                },
+            }
+        ],
+        "project_readback": {"ok": False, "returncode": 1, "summary": None, "stderr_excerpt": "Failed"},
+        "exact_gate": "blocked_on_network_dns_before_auth_refresh",
+        "clear_CTA": "Run from a network-enabled shell.",
+        "next_non_user_action": ["Rerun the probe."],
+        "claim_ceiling": "Read-only only.",
+    }
+    out = tmp_path / "probe.md"
+
+    probe.write_markdown(out, report)
+
+    text = out.read_text(encoding="utf-8")
+    assert "GCP DB Parity Current Probe" in text
+    assert "Exact gate: `blocked_on_network_dns_before_auth_refresh`" in text
+    assert "Run from a network-enabled shell." in text