#!/usr/bin/env python3
"""Run the six no-send Cory direct-claim prompts against a generated GCP DB."""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import ipaddress
import json
import os
import shlex
import shutil
import subprocess
import sys
import uuid
from pathlib import Path
from types import SimpleNamespace
from typing import Any

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

import run_leo_clone_bound_handler_checkpoint as bound  # noqa: E402
import working_leo_open_ended_benchmark as benchmark  # noqa: E402

SCHEMA = "livingip.gcpGeneratedDbDirectClaimSuite.v1"
SOURCE_COMPUTE = "teleo-prod-1"
DEFAULT_SERVICE = "leoclean-gcp-prod-parallel.service"
DEFAULT_HOST = "10.61.0.3"
DEFAULT_PROJECT = "teleo-501523"
DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql"
REVIEWED_CLOUDSQL_TOOL_SHA256 = "f54040141551a22a9a39ab020abfd897963bc2a53205f00cd49d7212cbaffff1"
REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a"


def run(command: list[str], *, env: dict[str, str] | None = None, input_text: str | None = None) -> str:
    completed = subprocess.run(
        command,
        env=env,
        input=input_text,
        text=True,
        capture_output=True,
        check=False,
    )
    if completed.returncode != 0:
        detail = bound.redact_text((completed.stderr or completed.stdout).strip())
        raise bound.CheckpointError(f"command failed ({completed.returncode}): {detail}")
    return completed.stdout


def service_state(service: str) -> dict[str, Any]:
    completed = subprocess.run(
        [
            "systemctl",
            "show",
            service,
            "-p",
            "ActiveState",
            "-p",
            "SubState",
            "-p",
            "MainPID",
            "-p",
            "NRestarts",
            "-p",
            "ExecMainStartTimestamp",
            "--no-pager",
        ],
        text=True,
        capture_output=True,
        check=False,
    )
    result: dict[str, Any] = {"returncode": completed.returncode}
    for line in completed.stdout.splitlines():
        key, separator, value = line.partition("=")
        if separator:
            result[key] = value
    if completed.returncode != 0:
        result["error"] = bound.redact_text(completed.stderr or completed.stdout)
    return result


def profile_hashes(profile: Path) -> dict[str, str | None]:
    paths = {
        "wrapper": profile / "bin" / "teleo-kb",
        "cloudsql_tool": profile / "bin" / "cloudsql_memory_tool.py",
        "bridge_skill": profile / "skills" / "teleo-kb-bridge" / "SKILL.md",
    }
    result = {}
    for name, path in paths.items():
        try:
            result[name] = bound.sha256_bytes(path.read_bytes())
        except OSError:
            result[name] = None
    return result


def database_password(args: argparse.Namespace) -> str:
    value = run(
        [
            "gcloud",
            "secrets",
            "versions",
            "access",
            "latest",
            f"--secret={args.password_secret}",
            f"--project={args.project}",
        ]
    ).strip()
    if not value:
        raise bound.CheckpointError("Cloud SQL password secret resolved to an empty value")
    return value


def validate_reviewed_file(path: Path, expected_sha256: str, label: str) -> str:
    actual = bound.sha256_bytes(path.read_bytes())
    if actual != expected_sha256:
        raise bound.CheckpointError(f"{label} does not match the reviewed SHA-256")
    return actual


def validate_parity_receipt(path: Path, target_db: str) -> dict[str, Any]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise bound.CheckpointError("canonical parity receipt is unreadable") from exc
    details = payload.get("details") or {}
    structural = details.get("structural_hashes") or {}
    checks = {
        "artifact": payload.get("artifact") == "canonical_postgres_parity_verification",
        "status": payload.get("status") == "pass",
        "problems_empty": payload.get("problems") == [],
        "target_database": details.get("target_database") == target_db,
        "nonempty_table_set": int(details.get("target_table_count") or 0) > 0,
        "nonempty_row_set": int(details.get("target_total_rows") or 0) > 0,
        "table_counts_equal": details.get("source_table_count") == details.get("target_table_count"),
        "row_counts_equal": details.get("source_total_rows") == details.get("target_total_rows"),
        "table_mismatches_empty": details.get("table_mismatches") == [],
        "role_mismatches_empty": details.get("application_role_mismatches") == {},
        "extension_mismatches_empty": details.get("required_extension_mismatches") == {},
        "performance_problems_empty": details.get("performance_problems") == [],
        "structural_hashes_equal": bool(structural)
        and all(item.get("source") == item.get("target") for item in structural.values()),
    }
    failed = sorted(name for name, ok in checks.items() if not ok)
    if failed:
        raise bound.CheckpointError(f"canonical parity receipt failed validation: {', '.join(failed)}")
    return {
        "sha256": bound.sha256_bytes(path.read_bytes()),
        "scope": payload.get("scope"),
        "target_database": details.get("target_database"),
        "target_table_count": details.get("target_table_count"),
        "target_total_rows": details.get("target_total_rows"),
        "checks": checks,
    }


def run_target_psql(
    args: argparse.Namespace,
    *,
    input_text: str | None = None,
    sql_file: Path | None = None,
) -> str:
    password = database_password(args)
    env = {
        **os.environ,
        "PGPASSWORD": password,
        "PGOPTIONS": "-c default_transaction_read_only=on",
    }
    command = [
        "psql",
        f"host={args.host} port=5432 dbname={args.target_db} user=postgres sslmode=require connect_timeout=8",
        "-X",
        "-Atq",
        "-v",
        "ON_ERROR_STOP=1",
    ]
    if sql_file is not None:
        command.extend(["-f", str(sql_file)])
    try:
        return run(command, env=env, input_text=input_text)
    finally:
        env.pop("PGPASSWORD", None)
        env.pop("PGOPTIONS", None)
        password = ""


def database_identity(args: argparse.Namespace) -> dict[str, Any]:
    sql = """
begin transaction read only;
select jsonb_build_object(
  'current_database', current_database(),
  'system_identifier', (select system_identifier::text from pg_control_system()),
  'server_address', host(inet_server_addr()),
  'server_port', inet_server_port(),
  'ssl', coalesce((select ssl from pg_stat_ssl where pid=pg_backend_pid()), false),
  'transaction_read_only', current_setting('transaction_read_only'),
  'default_transaction_read_only', current_setting('default_transaction_read_only')
)::text;
rollback;
"""
    output = run_target_psql(args, input_text=sql)
    rows = [line for line in output.splitlines() if line.strip().startswith("{")]
    if not rows:
        raise bound.CheckpointError("Cloud SQL identity query returned no JSON")
    return json.loads(rows[-1])


def validate_database_identity(identity: dict[str, Any], target_db: str) -> None:
    if identity.get("current_database") != target_db:
        raise bound.CheckpointError("Cloud SQL identity does not match the generated target database")
    if identity.get("ssl") is not True:
        raise bound.CheckpointError("Cloud SQL target session is not protected by TLS")
    if identity.get("transaction_read_only") != "on":
        raise bound.CheckpointError("Cloud SQL identity receipt was not captured in a read-only transaction")
    if identity.get("default_transaction_read_only") != "on":
        raise bound.CheckpointError("Cloud SQL target session does not default every transaction to read-only")
    try:
        address = ipaddress.ip_address(str(identity.get("server_address") or ""))
    except ValueError as exc:
        raise bound.CheckpointError("Cloud SQL target did not expose a valid server address") from exc
    private_networks = (
        ipaddress.ip_network("10.0.0.0/8"),
        ipaddress.ip_network("172.16.0.0/12"),
        ipaddress.ip_network("192.168.0.0/16"),
    )
    if not any(address in network for network in private_networks):
        raise bound.CheckpointError("Cloud SQL target address is not RFC1918-private")
    if not identity.get("system_identifier"):
        raise bound.CheckpointError("Cloud SQL target system identifier is missing")


def canonical_status(args: argparse.Namespace) -> dict[str, Any]:
    sql = """
begin transaction read only;
select jsonb_build_object(
  'db_identity', current_database() || '|' || current_user,
  'schema_tables', (
    select coalesce(jsonb_object_agg(table_schema, table_count order by table_schema), '{}'::jsonb)
      from (
        select table_schema, count(*)::integer as table_count
          from information_schema.tables
         where table_schema in ('public', 'kb_stage')
         group by table_schema
      ) counts
  ),
  'high_signal_rows', jsonb_build_object(
    'agents', (select count(*) from public.agents),
    'personas', (select count(*) from public.personas),
    'strategies', (select count(*) from public.strategies),
    'beliefs', (select count(*) from public.beliefs),
    'claims', (select count(*) from public.claims),
    'claim_edges', (select count(*) from public.claim_edges),
    'claim_evidence', (select count(*) from public.claim_evidence),
    'sources', (select count(*) from public.sources),
    'kb_proposals', (select count(*) from kb_stage.kb_proposals)
  )
)::text;
rollback;
"""
    output = run_target_psql(args, input_text=sql)
    rows = [line for line in output.splitlines() if line.strip().startswith("{")]
    if not rows:
        raise bound.CheckpointError("canonical status query returned no JSON")
    return json.loads(rows[-1])


def normalize_database_manifest_rows(output: str) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for line_number, raw_line in enumerate(output.splitlines(), 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 bound.CheckpointError(f"database manifest line {line_number} is not JSON") from exc
        kind = row.get("kind")
        if kind == "performance":
            continue
        if kind == "identity":
            row.pop("captured_at", None)
        rows.append(row)
    if not rows or not any(row.get("kind") == "identity" for row in rows):
        raise bound.CheckpointError("database manifest is missing its identity row")
    table_rows = [row for row in rows if row.get("kind") == "table"]
    if not table_rows:
        raise bound.CheckpointError("database manifest contains no table rows")
    return sorted(
        rows,
        key=lambda row: (
            str(row.get("kind") or ""),
            str(row.get("schema") or ""),
            str(row.get("table") or ""),
        ),
    )


def database_fingerprint(args: argparse.Namespace) -> dict[str, Any]:
    output = run_target_psql(args, sql_file=args.manifest_sql)
    rows = normalize_database_manifest_rows(output)
    payload = json.dumps(rows, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    table_rows = [row for row in rows if row.get("kind") == "table"]
    identity = next(row for row in rows if row.get("kind") == "identity")
    return {
        "sha256": hashlib.sha256(payload.encode()).hexdigest(),
        "table_count": len(table_rows),
        "total_rows": sum(int(row.get("row_count") or 0) for row in table_rows),
        "database": identity.get("database"),
        "server_address": identity.get("server_address"),
        "ssl": identity.get("ssl"),
        "transaction_read_only": identity.get("transaction_read_only"),
    }


def build_cloudsql_wrapper(
    *,
    cloudsql_tool: Path,
    tool_log: Path,
    target_db: str,
    host: str,
    project: str,
    password_secret: str,
    run_nonce: str,
    system_exec_path: str = bound.SYSTEM_EXEC_PATH,
) -> str:
    python = str(Path(sys.executable).resolve())
    return f"""#!/bin/bash
set -uo pipefail

CLOUDSQL_TOOL={shlex.quote(str(cloudsql_tool))}
TOOL_LOG={shlex.quote(str(tool_log))}
TARGET_DATABASE={shlex.quote(target_db)}
TARGET_HOST={shlex.quote(host)}
SOURCE_COMPUTE={shlex.quote(SOURCE_COMPUTE)}
PROJECT={shlex.quote(project)}
PASSWORD_SECRET={shlex.quote(password_secret)}
RUN_NONCE={shlex.quote(run_nonce)}
PYTHON={shlex.quote(python)}
export PATH={shlex.quote(system_exec_path)}
export CLOUDSDK_CONFIG=/home/teleo/.config/gcloud
export PGOPTIONS="-c default_transaction_read_only=on"
INVOCATION_ID="$($PYTHON -c 'import uuid; print(uuid.uuid4())')"
if ! PGPASSWORD="$(gcloud secrets versions access latest --secret="$PASSWORD_SECRET" --project="$PROJECT")"; then
  exit 124
fi
export PGPASSWORD
trap 'unset PGPASSWORD PGOPTIONS' EXIT
DB_RECEIPT="$(psql "host=$TARGET_HOST port=5432 dbname=$TARGET_DATABASE user=postgres sslmode=require connect_timeout=8" -X -Atq -v ON_ERROR_STOP=1 <<'SQL'
begin transaction read only;
select jsonb_build_object(
  'current_database', current_database(),
  'system_identifier', (select system_identifier::text from pg_control_system()),
  'server_address', host(inet_server_addr()),
  'server_port', inet_server_port(),
  'ssl', coalesce((select ssl from pg_stat_ssl where pid=pg_backend_pid()), false),
  'transaction_read_only', current_setting('transaction_read_only'),
  'default_transaction_read_only', current_setting('default_transaction_read_only')
)::text;
rollback;
SQL
)"
if [[ -z "$DB_RECEIPT" ]]; then
  unset PGPASSWORD
  exit 125
fi

$PYTHON - "$TOOL_LOG" "$RUN_NONCE" "$INVOCATION_ID" "$SOURCE_COMPUTE" "$TARGET_DATABASE" "$DB_RECEIPT" "$@" <<'PY'
import json
import sys
from datetime import datetime, timezone

path, run_nonce, invocation_id, compute, database, db_receipt, *argv = sys.argv[1:]
event = {{
    "phase": "start",
    "at_utc": datetime.now(timezone.utc).isoformat(),
    "invocation_id": invocation_id,
    "run_nonce": run_nonce,
    "container": compute,
    "database": database,
    "database_identity": json.loads(db_receipt),
    "argv": argv,
}}
with open(path, "a", encoding="utf-8") as handle:
    handle.write(json.dumps(event, sort_keys=True) + "\\n")
PY

$PYTHON "$CLOUDSQL_TOOL" \
  --host "$TARGET_HOST" \
  --db "$TARGET_DATABASE" \
  --canonical-db "$TARGET_DATABASE" \
  --project "$PROJECT" \
  --password-secret "$PASSWORD_SECRET" \
  "$@"
STATUS=$?

$PYTHON - "$TOOL_LOG" "$RUN_NONCE" "$INVOCATION_ID" "$SOURCE_COMPUTE" "$TARGET_DATABASE" "$STATUS" <<'PY'
import json
import sys
from datetime import datetime, timezone

path, run_nonce, invocation_id, compute, database, status = sys.argv[1:]
event = {{
    "phase": "end",
    "at_utc": datetime.now(timezone.utc).isoformat(),
    "invocation_id": invocation_id,
    "run_nonce": run_nonce,
    "container": compute,
    "database": database,
    "returncode": int(status),
}}
with open(path, "a", encoding="utf-8") as handle:
    handle.write(json.dumps(event, sort_keys=True) + "\\n")
PY

exit "$STATUS"
"""


def patch_temp_bridge(args: argparse.Namespace, temp_profile: Path) -> dict[str, Any]:
    wrapper = temp_profile / "bin" / "teleo-kb"
    target_tool = temp_profile / "bin" / "cloudsql_memory_tool.py"
    skill = temp_profile / "skills" / "teleo-kb-bridge" / "SKILL.md"
    tool_log = temp_profile / "gcp-checkpoint-tool-calls.jsonl"
    if not skill.is_file():
        raise bound.CheckpointError("temporary teleo-kb bridge skill is missing")
    shutil.copyfile(args.cloudsql_tool, target_tool)
    target_tool.chmod(0o700)
    run_nonce = uuid.uuid4().hex
    wrapper_text = build_cloudsql_wrapper(
        cloudsql_tool=target_tool,
        tool_log=tool_log,
        target_db=args.target_db,
        host=args.host,
        project=args.project,
        password_secret=args.password_secret,
        run_nonce=run_nonce,
    )
    binding = f"""
## Generated GCP Database Checkpoint

This temporary no-send profile is bound only to private-TLS database
`{args.target_db}` on `{args.host}` from `{SOURCE_COMPUTE}`. Every terminal call
must start directly with the bare command `teleo-kb` and contain exactly one
read command. Never use an absolute path, Python, shell variables, `which`,
`env`, pipes, redirects, heredocs, cache files, or a second shell command. Use
only search, context, show, evidence, edges, list-proposals,
search-proposals, show-proposal, and decision-matrix-status. Keep list/search
limits at 30 or below. For aggregate proposal status counts, call
`teleo-kb decision-matrix-status`; do not parse cached output. Do not stage,
approve, apply, write, send, or use live-profile paths. Distinguish pending,
approved, applied, and canonical rows from database readback; never infer
decision-matrix approval when its tables are absent. `status` and the restored
audit fallback are unavailable in this clone-only profile. Every helper session
is server-enforced with `default_transaction_read_only=on`.

For every direct-claim answer, use this sequence: direct yes/no/partly answer;
exact proposal/canonical-row readback; what is not proven; then a final line
beginning exactly `Next Cory-style follow-up:` with one concrete query, review,
apply-authorization, renderer, or demo-tier action. When a proposal is approved
but has no applied timestamp, say that approval is not application and that
explicit operator authorization is required before apply. When discussing the
matrix, say the conclusion comes from the fresh schema readback. When discussing
identity, name the DB rows/readback needed in addition to the rendered file.
"""
    skill_text = skill.read_text(encoding="utf-8")
    for live_wrapper in {
        "/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb",
        str(args.live_profile / "bin" / "teleo-kb"),
    }:
        skill_text = skill_text.replace(live_wrapper, "teleo-kb")
    skill_text = bound.inject_skill_binding(skill_text, binding.lstrip())
    descriptor = os.open(tool_log, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    os.close(descriptor)
    bound._detach_and_write(wrapper, wrapper_text, mode=0o700)
    bound._detach_and_write(skill, skill_text, mode=0o600)
    return {
        "wrapper_path": str(wrapper),
        "tool_log_path": str(tool_log),
        "run_nonce": run_nonce,
        "wrapper_sha256": bound.sha256_bytes(wrapper_text.encode()),
        "cloudsql_tool_sha256": bound.sha256_bytes(target_tool.read_bytes()),
        "reviewed_cloudsql_tool_sha256": REVIEWED_CLOUDSQL_TOOL_SHA256,
        "bridge_skill_sha256": bound.sha256_bytes(skill_text.encode()),
        "source_compute": SOURCE_COMPUTE,
        "target_database": args.target_db,
    }


def turn_args(args: argparse.Namespace, prompt: dict[str, Any]) -> argparse.Namespace:
    return SimpleNamespace(
        chat_id=args.chat_id,
        user_id=args.user_id,
        user_name=args.user_name,
        prompt=prompt["message"],
        prompt_id=prompt["id"],
        run_id=args.run_id,
        turn_timeout=args.turn_timeout,
    )


def direct_prompts() -> list[dict[str, Any]]:
    return [
        {"id": row["id"], "dimension": row["dimension"], "message": row["message"]}
        for row in benchmark.CORY_DIRECT_CLAIM_FOLLOWUP_SCENARIOS
    ]


async def run_suite(args: argparse.Namespace) -> dict[str, Any]:
    report: dict[str, Any] = {
        "schema": SCHEMA,
        "generated_at_utc": bound.utc_now(),
        "mode": "gcp_generated_db_gatewayrunner_no_send",
        "source_compute": SOURCE_COMPUTE,
        "target_database": args.target_db,
        "private_tls": False,
        "posted_to_telegram": False,
        "production_service_restart_attempted": False,
        "database_write_attempted": False,
        "results": [],
        "errors": [],
    }
    before_service: dict[str, Any] = {}
    before_hashes: dict[str, str | None] = {}
    before_identity: dict[str, Any] = {}
    before_fingerprint: dict[str, Any] = {}
    before_status: dict[str, Any] = {}
    temp_profile: Path | None = None
    provider_environment: dict[str, str] = {}
    provider_secrets: tuple[str, ...] = ()
    bridge: dict[str, Any] = {}
    termination: bound.TerminationRequested | None = None
    try:
        tool_sha = validate_reviewed_file(
            args.cloudsql_tool,
            REVIEWED_CLOUDSQL_TOOL_SHA256,
            "Cloud SQL bridge helper",
        )
        manifest_sha = validate_reviewed_file(
            args.manifest_sql,
            REVIEWED_MANIFEST_SQL_SHA256,
            "Postgres parity manifest",
        )
        report["reviewed_inputs"] = {
            "cloudsql_tool_sha256": tool_sha,
            "manifest_sql_sha256": manifest_sha,
        }
        report["parity_receipt"] = validate_parity_receipt(args.parity_receipt, args.target_db)
        before_service = service_state(args.service)
        before_hashes = profile_hashes(args.live_profile)
        before_identity = database_identity(args)
        validate_database_identity(before_identity, args.target_db)
        report["private_tls"] = before_identity.get("ssl") is True
        before_fingerprint = database_fingerprint(args)
        before_status = canonical_status(args)
        report["service_before"] = before_service
        report["live_profile_hashes_before"] = before_hashes
        report["database_identity_before"] = before_identity
        report["database_fingerprint_before"] = before_fingerprint
        report["canonical_status_before"] = before_status
        temp_profile, copy_audit = bound.copy_profile(
            run_id=args.run_id,
            prompt_id="gcp-direct-claim",
            live_profile=args.live_profile,
        )
        report["temporary_profile_copy_audit"] = copy_audit
        report["model_auth_binding"] = bound.copy_ephemeral_model_auth(
            temp_profile,
            live_profile=args.live_profile,
        )
        provider_environment, environment_audit = bound.load_ephemeral_provider_environment(temp_profile)
        provider_secrets = tuple(provider_environment.values())
        report["model_auth_binding"]["environment_binding"] = environment_audit
        bridge = patch_temp_bridge(args, temp_profile)
        report["temporary_bridge"] = bridge
        for index, prompt in enumerate(direct_prompts(), 1):
            gateway = await bound.invoke_gateway_subprocess(
                turn_args(args, prompt),
                temp_profile,
                provider_environment=provider_environment,
            )
            report["results"].append(
                {
                    "turn": index,
                    "prompt_id": prompt["id"],
                    "dimension": prompt["dimension"],
                    "prompt": prompt["message"],
                    "reply": gateway.get("reply") or "",
                    "ok": bool(str(gateway.get("reply") or "").strip()),
                    "mutates_kb": False,
                    "evidence_tier": "gcp_generated_db_gatewayrunner_no_send",
                    "gateway": gateway,
                }
            )
        expected_ids = [prompt["id"] for prompt in direct_prompts()]
        report["score"] = benchmark.score_result_subset(
            report["results"],
            catalog=benchmark.CORY_DIRECT_CLAIM_FOLLOWUP_SCENARIOS,
            expected_prompt_ids=expected_ids,
        )
        report["tool_proof"] = bound.read_tool_proof(
            Path(bridge["tool_log_path"]),
            SOURCE_COMPUTE,
            args.target_db,
            run_nonce=bridge["run_nonce"],
            database_identity=before_identity,
            require_database_read_only=True,
            require_default_read_only=True,
        )
    except bound.TerminationRequested as exc:
        termination = exc
        report["errors"].append(
            {"phase": "execution", "type": type(exc).__name__, "message": f"signal={exc.signum}"}
        )
    except Exception as exc:
        report["errors"].append(
            {"phase": "execution", "type": type(exc).__name__, "message": bound.redact_text(str(exc))}
        )
    finally:
        provider_environment.clear()
        report["gateway_child_cleanup"] = bound.cleanup_active_gateway_children()
        temp_root = temp_profile.parent if temp_profile else None
        cleanup_clear = not bound._ACTIVE_GATEWAY_CHILDREN
        report["temp_profile_removed"] = bound.remove_tree(temp_root) if cleanup_clear else False
        report["temp_profile_absent"] = bound.temp_path_absent(temp_root)
        report["temp_root_registry_clear"] = temp_root not in bound._ACTIVE_TEMP_ROOTS if temp_root else True
        postflight_steps = {
            "service_after": lambda: service_state(args.service),
            "live_profile_hashes_after": lambda: profile_hashes(args.live_profile),
        }
        if before_identity:
            postflight_steps["database_identity_after"] = lambda: database_identity(args)
            postflight_steps["canonical_status_after"] = lambda: canonical_status(args)
        if before_fingerprint and report.get("reviewed_inputs"):
            postflight_steps["database_fingerprint_after"] = lambda: database_fingerprint(args)
        for key, capture in postflight_steps.items():
            try:
                report[key] = capture()
                if key == "database_identity_after":
                    validate_database_identity(report[key], args.target_db)
            except BaseException as exc:
                if isinstance(exc, bound.TerminationRequested) and termination is None:
                    termination = exc
                report["errors"].append(
                    {"phase": key, "type": type(exc).__name__, "message": bound.redact_text(str(exc))}
                )
                report[key] = None

        after_service = report.get("service_after") or {}
        after_hashes = report.get("live_profile_hashes_after") or {}
        after_identity = report.get("database_identity_after") or {}
        after_fingerprint = report.get("database_fingerprint_after") or {}
        gateways = [row.get("gateway") or {} for row in report["results"]]
        allowed_tools = {"skills_list", "skill_view", "terminal"}
        tool_proof = report.get("tool_proof") or {}
        report["checks"] = {
            "reviewed_inputs_pinned": bool(report.get("reviewed_inputs")),
            "parity_receipt_validated": bool(report.get("parity_receipt")),
            "six_replies_returned": len(report["results"]) == 6 and all(row.get("ok") for row in report["results"]),
            "strict_score_passed": report.get("score", {}).get("pass") is True,
            "tool_calls_bound_read_only_and_successful": tool_proof.get("all_bound_to_supplied_target") is True
            and tool_proof.get("all_completed_successfully") is True
            and tool_proof.get("default_read_only_required") is True,
            "gateway_adapters_absent": bool(gateways)
            and all((gateway.get("tool_surface") or {}).get("gateway_adapter_count") == 0 for gateway in gateways),
            "send_message_tool_absent": bool(gateways)
            and all((gateway.get("tool_surface") or {}).get("send_message_tool_enabled") is False for gateway in gateways),
            "tool_registry_allowlisted": bool(gateways)
            and all(
                set((gateway.get("tool_surface") or {}).get("actual_registry_tools") or []) == allowed_tools
                for gateway in gateways
            ),
            "all_children_exited": bool(gateways)
            and all(
                (gateway.get("child_process") or {}).get("alive_after_readback") is False
                and (gateway.get("child_process") or {}).get("process_group_alive_after_readback") is False
                for gateway in gateways
            ),
            "generated_database_unchanged": bool(before_fingerprint)
            and before_fingerprint == after_fingerprint,
            "database_identity_unchanged": bool(before_identity) and before_identity == after_identity,
            "live_service_unchanged": bool(before_service) and before_service == after_service,
            "live_profile_unchanged": any(before_hashes.values()) and before_hashes == after_hashes,
            "temporary_profile_removed": report.get("temp_profile_removed") is True
            and report.get("temp_profile_absent") is True
            and report.get("temp_root_registry_clear") is True,
            "no_telegram_send": report.get("posted_to_telegram") is False,
        }
        report["status"] = "pass" if not report["errors"] and all(report["checks"].values()) else "fail"
        report["claim_ceiling"] = (
            "A pass proves six no-send direct-claim prompts through an adapter-free GatewayRunner temporary profile "
            "bound to one generated Cloud SQL database over private TLS. Actual helper sessions default to read-only, "
            "and the public/kb_stage catalog plus table-row fingerprints remain unchanged. It validates a supplied "
            "point-in-time parity receipt; it does not independently prove database creation provenance, Telegram "
            "delivery, separate-VM staging isolation, source extraction/apply composition, ongoing replication, or "
            "production cutover."
        )
        report["completed_at_utc"] = bound.utc_now()
        bound.write_report(args.output, report, secret_values=provider_secrets)
    if termination is not None:
        raise termination
    return report


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--execute", action="store_true")
    parser.add_argument("--target-db", required=True)
    parser.add_argument("--cloudsql-tool", required=True, type=Path)
    parser.add_argument("--manifest-sql", default=DEFAULT_MANIFEST_SQL, type=Path)
    parser.add_argument("--parity-receipt", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--host", default=DEFAULT_HOST)
    parser.add_argument("--project", default=DEFAULT_PROJECT)
    parser.add_argument("--password-secret", default=DEFAULT_SECRET)
    parser.add_argument("--service", default=DEFAULT_SERVICE)
    parser.add_argument("--live-profile", default=bound.LIVE_PROFILE, type=Path)
    parser.add_argument("--chat-id", default=bound.DEFAULT_CHAT_ID)
    parser.add_argument("--user-id", default=bound.DEFAULT_USER_ID)
    parser.add_argument("--user-name", default="codex GCP generated DB checkpoint")
    parser.add_argument("--run-id", default="gcp-direct-claim-" + uuid.uuid4().hex[:12])
    parser.add_argument("--turn-timeout", default=300, type=int)
    args = parser.parse_args(argv)
    if not args.execute:
        parser.error("--execute is required because this runs paid model calls")
    if not bound.SAFE_DB_NAME_RE.fullmatch(args.target_db):
        parser.error("--target-db must be a safe Postgres database name")
    if not args.target_db.startswith("teleo_clone_"):
        parser.error("--target-db must start with teleo_clone_ and cannot name a live database")
    if not args.cloudsql_tool.is_file():
        parser.error("--cloudsql-tool must exist")
    if not args.manifest_sql.is_file():
        parser.error("--manifest-sql must exist")
    if not args.parity_receipt.is_file():
        parser.error("--parity-receipt must exist")
    if not args.live_profile.is_dir():
        parser.error("--live-profile must exist")
    if args.turn_timeout <= 0:
        parser.error("--turn-timeout must be positive")
    try:
        args.output = bound.validate_private_output_path(args.output)
    except bound.CheckpointError as exc:
        parser.error(str(exc))
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    with bound.termination_cleanup_handlers():
        report = asyncio.run(run_suite(args))
    print(
        json.dumps(
            {
                "status": report.get("status"),
                "score": (report.get("score") or {}).get("coverage"),
                "checks_passed": sum(bool(value) for value in report.get("checks", {}).values()),
                "checks_total": len(report.get("checks", {})),
                "output": str(args.output),
            },
            sort_keys=True,
        )
    )
    return 0 if report.get("status") == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main())