#!/usr/bin/env python3
"""Value-safe preflight for RAILIANCE-WP-0024 recovery exercises.

This helper is deliberately read-only.  It never reads Kubernetes Secret data,
database usernames/passwords, OpenBao lease payloads, or unseal material.  Live
lease revocation and a host reboot remain separate, attended actions.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shlex
import subprocess
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any


EXPECTED_IMAGE = (
    "forgejo.coulomb.social/coulomb/audit-core@"
    "sha256:ec15f63d49226bfe507af2bc38ffbd5e83f549ba2c6d1ae7a338e7e053f34615"
)
EXPECTED_DB_KEYS = {"username", "password", "host", "port", "dbname"}


class PreflightError(RuntimeError):
    pass


def serving_pods(pod_list: dict[str, Any]) -> list[dict[str, Any]]:
    """Deployment pods only: audit-core attest-chain Job pods share the app label."""
    return [
        pod
        for pod in pod_list.get("items", [])
        if not any(
            owner.get("kind") == "Job"
            for owner in pod.get("metadata", {}).get("ownerReferences", [])
        )
    ]


KUBECONFIG_ENV = "AUDIT_KUBECONFIG"


def attended_kubeconfig(environ: dict[str, str] | None = None) -> str | None:
    """Absolute path of the attended admin kubeconfig, or None.

    Since RCLUSTER-WP-0019 (2026-10-20) the node account can no longer read the
    k3s admin kubeconfig. Attended cluster access uses the operator's own file
    (ops-warden route ``kubernetes-railiance01-admin``), named explicitly for one
    command. The file is passed to kubectl and never read here. Unset keeps
    the earlier ``ssh <host> kubectl`` path.
    """
    value = (os.environ if environ is None else environ).get(KUBECONFIG_ENV)
    if not value:
        return None
    path = Path(value)
    if not path.is_absolute() or not path.is_file():
        raise PreflightError(f"{KUBECONFIG_ENV} must be an absolute path to an existing file")
    return str(path)


class Remote:
    def __init__(self, host: str, kubeconfig: str | None = None) -> None:
        self.host = host
        self.kubeconfig = kubeconfig if kubeconfig is not None else attended_kubeconfig()

    def run(self, command: list[str], *, label: str) -> str:
        remote_command = " ".join(shlex.quote(part) for part in command)
        completed = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", self.host, remote_command],
            text=True,
            capture_output=True,
            check=False,
        )
        if completed.returncode != 0:
            # Remote output can include provider or application detail that is
            # inappropriate for durable evidence.  Report only the failed step.
            raise PreflightError(f"{label} failed (exit {completed.returncode})")
        return completed.stdout.strip()

    def kubectl(self, args: list[str], *, label: str) -> str:
        if self.kubeconfig is None:
            return self.run(["kubectl", *args], label=label)
        completed = subprocess.run(
            ["kubectl", "--kubeconfig", self.kubeconfig, *args],
            text=True,
            capture_output=True,
            check=False,
        )
        if completed.returncode != 0:
            # Same rule as the ssh path: report only the failed step.
            raise PreflightError(f"{label} failed (exit {completed.returncode})")
        return completed.stdout.strip()

    def kubectl_json(self, args: list[str], *, label: str) -> dict[str, Any]:
        raw = self.kubectl([*args, "-o", "json"], label=label)
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise PreflightError(f"{label} returned invalid JSON") from exc
        if not isinstance(value, dict):
            raise PreflightError(f"{label} did not return an object")
        return value


def condition(resource: dict[str, Any], kind: str) -> str | None:
    for item in resource.get("status", {}).get("conditions", []):
        if item.get("type") == kind:
            return item.get("status")
    return None


def rfc3339(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(UTC)


def latest_completed_backup(backups: dict[str, Any], now: datetime) -> dict[str, Any]:
    completed = [
        item
        for item in backups.get("items", [])
        if item.get("status", {}).get("phase") == "completed"
        and item.get("status", {}).get("stoppedAt")
    ]
    if not completed:
        raise PreflightError("no completed platform-pg backup is visible")
    latest = max(completed, key=lambda item: item["status"]["stoppedAt"])
    stopped = rfc3339(latest["status"]["stoppedAt"])
    return {
        "name": latest["metadata"]["name"],
        "stopped_at": stopped.isoformat().replace("+00:00", "Z"),
        "age_hours": round((now - stopped).total_seconds() / 3600, 2),
        "method": latest.get("spec", {}).get("method"),
    }


def running_pod_image(pod: dict[str, Any]) -> dict[str, Any]:
    """Image actually running in the serving pod, not only the Deployment spec.

    ``status.containerStatuses[].imageID`` carries the digest the kubelet
    resolved; a spec that matches while an older ReplicaSet pod still runs must
    fail closed.
    """
    containers = pod.get("spec", {}).get("containers", [])
    statuses = pod.get("status", {}).get("containerStatuses", [])
    name = containers[0].get("name") if containers else None
    spec_image = containers[0].get("image") if containers else None
    status = next((item for item in statuses if item.get("name") == name), None)
    image_id = (status or {}).get("imageID") or ""
    match = re.search(r"@(sha256:[0-9a-f]{64})$", image_id)
    return {"image": spec_image, "digest": match.group(1) if match else None}


def resource_condition(resource: dict[str, Any]) -> dict[str, Any]:
    conditions = resource.get("status", {}).get("conditions", [])
    current = conditions[0] if conditions else {}
    return {
        "ready": current.get("status") == "True",
        "reason": current.get("reason"),
    }


def secret_key_names(description: str) -> set[str]:
    """Extract names from ``kubectl describe secret`` without reading data."""
    keys: set[str] = set()
    for line in description.splitlines():
        match = re.match(r"^([^\s:]+):\s+\d+ bytes$", line.strip())
        if match:
            keys.add(match.group(1))
    return keys


def validate_snapshot_receipt(
    path: Path,
    *,
    live_openbao: dict[str, Any],
    now: datetime,
    max_age_hours: float,
) -> dict[str, Any]:
    try:
        receipt = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise PreflightError("OpenBao snapshot receipt is unavailable or invalid JSON") from exc
    if not isinstance(receipt, dict) or receipt.get("receipt_version") != 1:
        raise PreflightError("OpenBao snapshot receipt has an unsupported shape/version")
    required_strings = (
        "receipt_id", "created_at", "operator", "source_cluster",
        "source_namespace", "source_pod", "cluster_id", "snapshot_sha256",
        "encrypted_snapshot_sha256", "encrypted_location_ref",
    )
    for key in required_strings:
        if not isinstance(receipt.get(key), str) or not receipt[key].strip():
            raise PreflightError(f"OpenBao snapshot receipt is missing {key}")
    required_true = (
        "snapshot_created", "source_initialized", "source_unsealed",
        "snapshot_encrypted", "encrypted_copy_off_host", "encryption_verified",
        "hash_verified", "no_secret_material_recorded",
    )
    for key in required_true:
        if receipt.get(key) is not True:
            raise PreflightError(f"OpenBao snapshot receipt requires {key}=true")
    # A snapshot whose only decryption key lives inside OpenBao cannot restore OpenBao.
    escrow = receipt.get("custodian_escrow_ref")
    if not isinstance(escrow, str) or not escrow.strip() or "replace" in escrow.lower():
        raise PreflightError("OpenBao snapshot receipt names no decryption key held outside OpenBao")
    if (
        receipt["source_cluster"] != "railiance01"
        or receipt["source_namespace"] != "openbao"
        or receipt["source_pod"] != "openbao-0"
        or receipt["cluster_id"] != live_openbao.get("cluster_id")
    ):
        raise PreflightError("OpenBao snapshot receipt does not match the live source")
    try:
        created = rfc3339(receipt["created_at"])
        snapshot_index = int(receipt["raft_applied_index"])
        live_index = int(live_openbao["raft_applied_index"])
    except (KeyError, TypeError, ValueError) as exc:
        raise PreflightError("OpenBao snapshot receipt has invalid time/index metadata") from exc
    age_hours = (now - created).total_seconds() / 3600
    if age_hours < -(5 / 60) or age_hours > max_age_hours:
        raise PreflightError("OpenBao snapshot receipt is outside the permitted age")
    if snapshot_index <= 0 or snapshot_index > live_index:
        raise PreflightError("OpenBao snapshot receipt has an impossible Raft index")
    digest_pattern = re.compile(r"^sha256:[0-9a-f]{64}$")
    digests = (receipt["snapshot_sha256"], receipt["encrypted_snapshot_sha256"])
    if any(not digest_pattern.fullmatch(value) for value in digests):
        raise PreflightError("OpenBao snapshot receipt has an invalid digest")
    if any(len(set(value.removeprefix("sha256:"))) <= 1 for value in digests):
        raise PreflightError("OpenBao snapshot receipt contains a placeholder digest")
    if digests[0] == digests[1]:
        raise PreflightError("plain and encrypted snapshot digests must differ")
    encoded = json.dumps(receipt, sort_keys=True)
    for marker in (
        "BEGIN PRIVATE KEY", "BEGIN OPENSSH PRIVATE KEY", "AGE-SECRET-KEY-1",
        "OPENBAO_ROOT_TOKEN", "VAULT_TOKEN", "hvs.", "<", "YYYY-MM-DD",
    ):
        if marker in encoded:
            raise PreflightError("OpenBao snapshot receipt contains forbidden material/placeholder")
    return {
        "receipt_id": receipt["receipt_id"],
        "created_at": created.isoformat().replace("+00:00", "Z"),
        "age_hours": round(age_hours, 2),
        "cluster_id_matches": True,
        "raft_applied_index": snapshot_index,
        "encrypted_copy_off_host": True,
        "verified": True,
        "secret_values_observed": False,
    }


def common_state(remote: Remote, now: datetime) -> dict[str, Any]:
    node_list = remote.kubectl_json(["get", "nodes"], label="read node state")
    nodes = node_list.get("items", [])
    if len(nodes) != 1:
        raise PreflightError(f"expected one railiance01 node, observed {len(nodes)}")
    node = nodes[0]

    cluster = remote.kubectl_json(
        ["-n", "databases", "get", "cluster", "platform-pg"],
        label="read platform-pg state",
    )
    deployment = remote.kubectl_json(
        ["-n", "audit-core", "get", "deployment", "audit-core"],
        label="read audit-core deployment",
    )
    pod_list = remote.kubectl_json(
        ["-n", "audit-core", "get", "pods", "-l", "app.kubernetes.io/name=audit-core"],
        label="read audit-core pod metadata",
    )
    pods = serving_pods(pod_list)
    if len(pods) != 1:
        raise PreflightError(f"expected one audit-core pod, observed {len(pods)}")
    pod = pods[0]
    container_statuses = pod.get("status", {}).get("containerStatuses", [])
    restart_count = sum(int(item.get("restartCount", 0)) for item in container_statuses)

    stores: dict[str, Any] = {}
    for name in ("openbao-audit-core", "openbao-audit-core-database"):
        resource = remote.kubectl_json(
            ["get", "clustersecretstore", name], label=f"read {name} state"
        )
        stores[name] = resource_condition(resource)

    external_secrets: dict[str, Any] = {}
    for name in ("audit-core-database", "audit-core-senders"):
        resource = remote.kubectl_json(
            ["-n", "audit-core", "get", "externalsecret", name],
            label=f"read {name} state",
        )
        external_secrets[name] = resource_condition(resource)

    bao_raw = remote.kubectl(
        ["-n", "openbao", "exec", "openbao-0", "--", "bao", "status", "-format=json"],
        label="read OpenBao seal state",
    )
    try:
        bao_status = json.loads(bao_raw)
    except json.JSONDecodeError as exc:
        raise PreflightError("OpenBao seal state returned invalid JSON") from exc
    image = deployment["spec"]["template"]["spec"]["containers"][0]["image"]
    running = running_pod_image(pod)
    state = {
        "captured_at": now.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "node": {
            "name": node["metadata"]["name"],
            "ready": condition(node, "Ready") == "True",
            "k3s_active": remote.run(
                ["systemctl", "is-active", "k3s"], label="read k3s service state"
            ) == "active",
            "uptime_seconds": float(
                remote.run(["cat", "/proc/uptime"], label="read host uptime").split()[0]
            ),
        },
        "platform_pg": {
            "ready_instances": cluster.get("status", {}).get("readyInstances", 0),
            "instances": cluster.get("spec", {}).get("instances", 0),
            "phase": cluster.get("status", {}).get("phase"),
            "continuous_archiving": condition(cluster, "ContinuousArchiving") == "True",
            "last_backup_succeeded": condition(cluster, "LastBackupSucceeded") == "True",
        },
        "openbao": {
            "initialized": bool(bao_status.get("initialized")),
            "sealed": bool(bao_status.get("sealed")),
            "seal_type": bao_status.get("type"),
            "threshold": bao_status.get("t"),
            "shares": bao_status.get("n"),
            "cluster_id": bao_status.get("cluster_id"),
            "raft_applied_index": bao_status.get("raft_applied_index"),
        },
        "external_secret_stores": stores,
        "external_secrets": external_secrets,
        "audit_core": {
            "image": image,
            "image_matches_reviewed_digest": image == EXPECTED_IMAGE,
            "running_pod_image": running["image"],
            "running_pod_image_digest": running["digest"],
            "running_pod_image_matches_reviewed_digest": (
                running["image"] == EXPECTED_IMAGE
                and running["digest"] == EXPECTED_IMAGE.rsplit("@", 1)[1]
            ),
            "ready_replicas": deployment.get("status", {}).get("readyReplicas", 0),
            "replicas": deployment.get("spec", {}).get("replicas", 0),
            "pod_uid": pod["metadata"]["uid"],
            "restart_count": restart_count,
        },
        "secret_values_observed": False,
    }
    return state


def automated_common_pass(state: dict[str, Any]) -> bool:
    return all(
        (
            state["node"]["ready"],
            state["node"]["k3s_active"],
            state["platform_pg"]["ready_instances"] == state["platform_pg"]["instances"] == 1,
            state["platform_pg"]["continuous_archiving"],
            state["platform_pg"]["last_backup_succeeded"],
            state["openbao"]["initialized"],
            not state["openbao"]["sealed"],
            all(item["ready"] for item in state["external_secret_stores"].values()),
            all(item["ready"] for item in state["external_secrets"].values()),
            state["audit_core"]["image_matches_reviewed_digest"],
            state["audit_core"]["running_pod_image_matches_reviewed_digest"],
            state["audit_core"]["ready_replicas"] == state["audit_core"]["replicas"] == 1,
        )
    )


def endpoint_status(remote: Remote, path: str) -> int:
    code = remote.kubectl(
        [
            "-n", "audit-core", "exec", "deploy/audit-core", "--", "python", "-c",
            "import urllib.request,urllib.error; "
            f"u='http://127.0.0.1:8080{path}'; "
            "\ntry:\n r=urllib.request.urlopen(u,timeout=3); print(r.status)"
            "\nexcept urllib.error.HTTPError as e:\n print(e.code)",
        ],
        label=f"probe audit-core {path}",
    )
    try:
        return int(code)
    except ValueError as exc:
        raise PreflightError(f"audit-core {path} returned an invalid status") from exc


def database_lease_preflight(remote: Remote, args: argparse.Namespace) -> dict[str, Any]:
    now = datetime.now(UTC)
    state = common_state(remote, now)
    secret_rv = remote.kubectl(
        ["-n", "audit-core", "get", "secret", "audit-core-database", "-o", "jsonpath={.metadata.resourceVersion}"],
        label="read database Secret metadata",
    )
    description = remote.kubectl(
        ["-n", "audit-core", "describe", "secret", "audit-core-database"],
        label="read database Secret key names",
    )
    keys = secret_key_names(description)
    health_code = endpoint_status(remote, "/healthz")
    ready_code = endpoint_status(remote, "/readyz")
    gates = {
        "approved_window_id_recorded": bool(args.approved_window_id),
        "audit_core_owner_acknowledged": args.audit_core_owner_ack,
        "rapp_postgres_owner_acknowledged": args.rapp_postgres_owner_ack,
        "approved_synthetic_load_contract": bool(args.synthetic_load_id),
        "attended_abort_operator_named": bool(args.abort_operator),
    }
    automated = automated_common_pass(state) and keys == EXPECTED_DB_KEYS and health_code == 200 and ready_code == 200
    return {
        "procedure": "audit-core-database-lease-recovery",
        "preflight_only": True,
        "automated_checks_passed": automated,
        "ready_for_live_execution": automated and all(gates.values()),
        "operator_gates": gates,
        "baseline": {
            "database_secret_resource_version": secret_rv,
            "database_secret_keys": sorted(keys),
            "health_status": health_code,
            "readiness_status": ready_code,
            "pod_uid": state["audit_core"]["pod_uid"],
            "restart_count": state["audit_core"]["restart_count"],
        },
        "state": state,
        "secret_values_observed": False,
    }


# --- node-reboot specific checks -------------------------------------------

# Named core platform workloads (namespace, kind, name) that must be fully
# Ready.  Every other Deployment/StatefulSet in these namespaces must be Ready
# as well, so a renamed or newly added operator component cannot hide.
CORE_PLATFORM_WORKLOADS = (
    ("kube-system", "Deployment", "coredns"),
    ("kube-system", "Deployment", "local-path-provisioner"),
    ("kube-system", "Deployment", "traefik"),
    ("kube-system", "Deployment", "metrics-server"),
    ("cert-manager", "Deployment", "cert-manager"),
    ("cnpg-system", "Deployment", "cnpg-controller-manager"),
    ("external-secrets", "Deployment", "external-secrets"),
    ("knative-serving", "Deployment", "activator"),
    ("knative-serving", "Deployment", "controller"),
    ("knative-serving", "Deployment", "net-kourier-controller"),
    ("kourier-system", "Deployment", "3scale-kourier-gateway"),
    ("argocd", "Deployment", "argocd-repo-server"),
    ("argocd", "StatefulSet", "argocd-application-controller"),
    ("openbao", "StatefulSet", "openbao"),
)
CORE_PLATFORM_NAMESPACES = frozenset(ns for ns, _, _ in CORE_PLATFORM_WORKLOADS)

# CNPG clusters observed on 2026-10-14.  The live list is discovered; this set
# only makes a vanished cluster fail closed instead of silently shrinking scope.
EXPECTED_CNPG_CLUSTERS = frozenset({
    "databases/apps-pg", "databases/forgejo-db", "databases/net-kingdom-pg",
    "databases/platform-pg", "databases/platform-pg-2", "databases/state-hub-db",
    "target-revenue/target-revenue-pg", "user-engine/user-engine-pg",
})
FRESH_BACKUP_REQUIRED = frozenset({"databases/platform-pg"})
FRESH_BACKUP_RECOMMENDED = frozenset({"databases/state-hub-db", "user-engine/user-engine-pg"})

# Declared host firewall (railiance-infra RAIL-HO-WP-0009 / Goss).
UFW_ALLOWED = frozenset({"OpenSSH", "2224/tcp"})
UFW_FORBIDDEN_PORTS = ("6443", "8472")
K3S_API_TUNNEL = "k3s-api-railiance01"

OPS_WARDEN_RECEIPT = "reviews/RPF-WP-0017-ops-warden-reacceptance-2026-10-14.json"
CONSUMED_DECISION_PREFIXES = ("449a697a", "85724c0c")
CONSUMED_SCENARIO_IDS = frozenset({"WARDEN-WP-0027-T02-DRILL-20260822-01"})
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{5,127}$")
FORBIDDEN_RECEIPT_MARKERS = (
    "BEGIN PRIVATE KEY", "BEGIN OPENSSH PRIVATE KEY", "AGE-SECRET-KEY-1",
    "OPENBAO_ROOT_TOKEN", "VAULT_TOKEN", "hvs.", "password=", "://",
)


def workload_ready(resource: dict[str, Any]) -> bool:
    wanted = resource.get("spec", {}).get("replicas", 1)
    return wanted >= 1 and resource.get("status", {}).get("readyReplicas", 0) == wanted


def core_platform_workloads(workloads: dict[str, Any]) -> dict[str, Any]:
    """Readiness of named core workloads plus every workload in core namespaces."""
    observed: dict[tuple[str, str, str], bool] = {}
    for item in workloads.get("items", []):
        meta = item.get("metadata", {})
        if meta.get("namespace") in CORE_PLATFORM_NAMESPACES:
            key = (meta["namespace"], item.get("kind"), meta.get("name"))
            observed[key] = workload_ready(item)
    missing = sorted("/".join(key) for key in CORE_PLATFORM_WORKLOADS if key not in observed)
    unready = sorted("/".join(key) for key, ready in observed.items() if not ready)
    return {
        "required": ["/".join(key) for key in CORE_PLATFORM_WORKLOADS],
        "missing": missing,
        "unready": unready,
        "ready": not missing and not unready,
    }


def failing_pods(pod_list: dict[str, Any]) -> list[str]:
    """``namespace/name`` of every pod that is neither Succeeded nor Running+Ready."""
    failing = []
    for pod in pod_list.get("items", []):
        status = pod.get("status", {})
        phase = status.get("phase")
        if phase == "Succeeded":
            continue
        statuses = status.get("containerStatuses", [])
        if phase == "Running" and statuses and all(item.get("ready") for item in statuses):
            continue
        meta = pod.get("metadata", {})
        failing.append(f"{meta.get('namespace')}/{meta.get('name')}")
    return sorted(failing)


def failing_pod_diff(baseline: list[str], current: list[str]) -> dict[str, Any]:
    new = sorted(set(current) - set(baseline))
    return {
        "baseline_count": len(baseline),
        "current_count": len(current),
        "new_failing": new,
        "resolved": sorted(set(baseline) - set(current)),
        "no_new_failures": not new,
    }


def system_identifier(controldata: str) -> str | None:
    match = re.search(r"^Database system identifier:\s+(\d+)\s*$", controldata, re.MULTILINE)
    return match.group(1) if match else None


def cnpg_identity(cluster: dict[str, Any], pvc_uids: dict[str, str]) -> dict[str, Any]:
    """Value-safe identity/health of one CNPG cluster (no credentials, no DSNs)."""
    meta = cluster.get("metadata", {})
    status = cluster.get("status", {})
    primary = status.get("currentPrimary")
    backup_configured = bool(cluster.get("spec", {}).get("backup"))
    archiving = condition(cluster, "ContinuousArchiving")
    return {
        "uid": meta.get("uid"),
        "instances": cluster.get("spec", {}).get("instances", 0),
        "ready_instances": status.get("readyInstances", 0),
        "phase": status.get("phase"),
        "current_primary": primary,
        "primary_pvc_uid": pvc_uids.get(primary) if primary else None,
        "healthy_pvc": sorted(status.get("healthyPVC", [])),
        "backup_configured": backup_configured,
        "continuous_archiving": archiving == "True" if backup_configured else None,
        "system_id": None,
    }


def cnpg_healthy(identity: dict[str, Any]) -> bool:
    return (
        identity["ready_instances"] == identity["instances"] == 1
        and bool(identity["uid"])
        and bool(identity["primary_pvc_uid"])
        and bool(identity["system_id"])
        and identity["current_primary"] in identity["healthy_pvc"]
        and identity["continuous_archiving"] is not False
    )


def latest_backups_by_cluster(backups: dict[str, Any], now: datetime) -> dict[str, Any]:
    grouped: dict[str, list[dict[str, Any]]] = {}
    for item in backups.get("items", []):
        meta = item.get("metadata", {})
        cluster = item.get("spec", {}).get("cluster", {}).get("name")
        if cluster:
            grouped.setdefault(f"{meta.get('namespace')}/{cluster}", []).append(item)
    result = {}
    for key, items in grouped.items():
        try:
            result[key] = latest_completed_backup({"items": items}, now)
        except PreflightError:
            result[key] = None
    return result


def eso_inventory(stores: dict[str, Any], external_secrets: dict[str, Any]) -> dict[str, Any]:
    """Every ClusterSecretStore Valid and every ExternalSecret SecretSynced."""
    def bad(items: list[dict[str, Any]], reason: str, namespaced: bool) -> list[str]:
        out = []
        for item in items:
            ready = next(
                (c for c in item.get("status", {}).get("conditions", []) if c.get("type") == "Ready"),
                {},
            )
            if ready.get("status") != "True" or ready.get("reason") != reason:
                meta = item.get("metadata", {})
                out.append(f"{meta.get('namespace')}/{meta.get('name')}" if namespaced else meta.get("name"))
        return sorted(out)

    store_items = stores.get("items", [])
    es_items = external_secrets.get("items", [])
    invalid_stores = bad(store_items, "Valid", False)
    unsynced = bad(es_items, "SecretSynced", True)
    return {
        "cluster_store_count": len(store_items),
        "invalid_cluster_stores": invalid_stores,
        "external_secret_count": len(es_items),
        "unsynced_external_secrets": unsynced,
        "ready": bool(store_items) and bool(es_items) and not invalid_stores and not unsynced,
    }


def ufw_state(status_text: str) -> dict[str, Any]:
    """Compare ``ufw status`` with the declared OpenSSH + Nydus 2224 allow set."""
    active = bool(re.search(r"^Status:\s+active\s*$", status_text, re.MULTILINE))
    targets: set[str] = set()
    in_rules = False
    for line in status_text.splitlines():
        if line.startswith("--"):
            in_rules = True
            continue
        if in_rules and line.strip():
            target = re.split(r"\s{2,}", line.strip())[0].replace(" (v6)", "")
            targets.add(target)
    forbidden = sorted(
        target for target in targets
        if any(re.match(rf"^{port}(/|:|$)", target) for port in UFW_FORBIDDEN_PORTS)
    )
    return {
        "active": active,
        "allowed_targets": sorted(targets),
        "forbidden_ports_open": forbidden,
        "matches_declaration": active and targets == UFW_ALLOWED and not forbidden,
    }


def boots_since(list_boots_json: str, since: datetime) -> int:
    """Count boots whose first entry is after ``since`` (journalctl --list-boots -o json)."""
    try:
        boots = json.loads(list_boots_json)
    except json.JSONDecodeError as exc:
        raise PreflightError("boot list returned invalid JSON") from exc
    count = 0
    for boot in boots:
        first = boot.get("first_entry")
        if isinstance(first, int) and datetime.fromtimestamp(first / 1_000_000, UTC) > since:
            count += 1
    return count


def validate_ops_warden_receipt(path: Path) -> dict[str, Any]:
    """RPF-WP-0017 re-acceptance of the ops-warden revision for the attended login."""
    try:
        raw = path.read_text(encoding="utf-8")
        receipt = json.loads(raw)
    except (OSError, json.JSONDecodeError) as exc:
        raise PreflightError("ops-warden acceptance receipt is unavailable or invalid JSON") from exc
    if not isinstance(receipt, dict) or receipt.get("workplan_id", "RPF-WP-0017") != "RPF-WP-0017" or "RPF-WP-0017" not in raw:
        raise PreflightError("ops-warden acceptance receipt is not an RPF-WP-0017 receipt")
    if any(marker in raw for marker in FORBIDDEN_RECEIPT_MARKERS[:-1]):
        raise PreflightError("ops-warden acceptance receipt contains forbidden material")
    if receipt.get("secret_values_observed") is True or receipt.get("sensitive_material_recorded") is True:
        raise PreflightError("ops-warden acceptance receipt records sensitive material")
    disposition = next(
        (str(receipt[key]).lower() for key in ("current_decision", "disposition", "decision", "status", "result") if key in receipt),
        "",
    )
    if disposition not in {"accept", "accepted", "approve", "approved"}:
        raise PreflightError("ops-warden acceptance receipt does not record acceptance")
    revision = next(
        (
            receipt[key] for key in ("current_reviewed_revision", "reviewed_revision", "accepted_revision", "ops_warden_revision", "revision")
            if isinstance(receipt.get(key), str) and re.fullmatch(r"[0-9a-f]{40}", receipt[key])
        ),
        None,
    )
    if not revision:
        raise PreflightError("ops-warden acceptance receipt names no exact 40-hex revision")
    return {"path": str(path), "accepted_revision": revision, "verified": True}


OPS_WARDEN_REPO = Path("/home/worsch/ops-warden")


def installed_warden_package(tool_python: Path) -> Path:
    """Package directory of the warden actually on PATH (uv tool environment)."""
    completed = subprocess.run(
        [str(tool_python), "-I", "-c", "import os, warden; print(os.path.dirname(warden.__file__))"],
        text=True, capture_output=True, check=False,
    )
    if completed.returncode != 0 or not completed.stdout.strip():
        raise PreflightError("installed warden package could not be located")
    return Path(completed.stdout.strip())


def verify_installed_warden(package: Path, repo: Path, revision: str) -> dict[str, Any]:
    """Prove the installed warden equals the accepted, published ops-warden revision.

    The uv tool is a non-editable directory install that records no revision, so
    compare every installed module byte-for-byte with that revision's tree.
    """
    def git(*args: str) -> subprocess.CompletedProcess[bytes]:
        return subprocess.run(["git", "-C", str(repo), *args], capture_output=True, check=False)

    if git("merge-base", "--is-ancestor", revision, "origin/main").returncode != 0:
        raise PreflightError("accepted ops-warden revision is not published on origin/main")
    listed = git("ls-tree", "-r", "--name-only", revision, "src/warden/")
    if listed.returncode != 0:
        raise PreflightError("accepted ops-warden revision is not available locally")
    expected = {
        name.removeprefix("src/warden/")
        for name in listed.stdout.decode().splitlines()
        if name.endswith(".py")
    }
    installed = {path.relative_to(package).as_posix() for path in package.rglob("*.py")}
    if installed != expected:
        raise PreflightError("installed warden module set differs from the accepted revision")
    for name in sorted(expected):
        blob = git("show", f"{revision}:src/warden/{name}")
        if blob.returncode != 0 or blob.stdout != (package / name).read_bytes():
            raise PreflightError("installed warden differs from the accepted revision")
    return {"installed_revision_matches": True, "modules_compared": len(expected)}


CONSOLE_RECEIPT_ID = re.compile(r"^RAIL01-CONSOLE-(\d{8})T\d{6}Z$")


def validate_console_receipt(path: Path, *, window_id: str | None, now: datetime) -> dict[str, Any]:
    """railiance-infra console attestation (docs/railiance01-console-access.md).

    Accept only a same-day receipt for this window that proves the console
    shows the host and is independent of the guest network.  Role-only fields;
    any credential-looking or placeholder content is denied.
    """
    try:
        import yaml  # type: ignore[import-untyped]

        raw = path.read_text(encoding="utf-8")
        receipt = yaml.safe_load(raw)
    except Exception as exc:  # noqa: BLE001 - any parse/import failure is fail-closed
        raise PreflightError("provider-console receipt is unavailable or invalid YAML") from exc
    if not isinstance(receipt, dict):
        raise PreflightError("provider-console receipt has an unsupported shape")
    if any(marker in raw for marker in FORBIDDEN_RECEIPT_MARKERS) or "<" in raw:
        raise PreflightError("provider-console receipt contains forbidden material/placeholder")
    match = CONSOLE_RECEIPT_ID.fullmatch(str(receipt.get("receipt_id", "")))
    if not match or path.stem != receipt["receipt_id"]:
        raise PreflightError("provider-console receipt id is not RAIL01-CONSOLE-<YYYYMMDDTHHMMSSZ>")
    try:
        attested = rfc3339(str(receipt["attested_at"]))
    except (KeyError, ValueError) as exc:
        raise PreflightError("provider-console receipt has invalid attested_at") from exc
    today = now.astimezone(UTC).strftime("%Y%m%d")
    if match.group(1) != today or attested.strftime("%Y%m%d") != today or attested > now:
        raise PreflightError("provider-console receipt is not from the window day")
    if receipt.get("host") != "railiance01" or receipt.get("panel_login") != "ok":
        raise PreflightError("provider-console receipt does not prove panel access to railiance01")
    if receipt.get("console_shows_host") is not True or receipt.get("independent_of_guest_network") is not True:
        raise PreflightError("provider-console receipt does not prove an independent host console")
    if not window_id or str(receipt.get("reboot_window_id")) != window_id:
        raise PreflightError("provider-console receipt does not name the approved window id")
    return {
        "receipt_id": receipt["receipt_id"],
        "attested_at": attested.isoformat().replace("+00:00", "Z"),
        "attested_by_role": receipt.get("attested_by_role"),
        "console_shows_host": True,
        "independent_of_guest_network": True,
        "window_id_matches": True,
        "verified": True,
    }


def fresh_decision_gates(scenario_id: str | None, decision_id: str | None) -> dict[str, bool]:
    """Consumed drill scenario/decisions can never authorize this window."""
    return {
        "new_scenario_id_recorded": bool(scenario_id)
        and bool(SAFE_ID.fullmatch(scenario_id))
        and scenario_id not in CONSUMED_SCENARIO_IDS,
        "fresh_human_decision_recorded": bool(decision_id)
        and bool(SAFE_ID.fullmatch(decision_id))
        and not decision_id.lower().startswith(CONSUMED_DECISION_PREFIXES),
    }


def reboot_comparison(pre: dict[str, Any], post: dict[str, Any], boots: int) -> dict[str, Any]:
    """Post-reboot proof: exactly one reboot, same database identities, no new failures."""
    pre_host, post_host = pre.get("host", {}), post["host"]
    identity_changes = []
    for key, before in pre.get("cnpg_clusters", {}).items():
        after = post["cnpg_clusters"].get(key)
        for field in ("uid", "primary_pvc_uid", "system_id"):
            if after is None or after.get(field) != before.get(field):
                identity_changes.append(f"{key}:{field}")
    pod_diff = failing_pod_diff(pre.get("failing_pods", []), post["failing_pods"])
    return {
        "boot_id_changed": bool(pre_host.get("boot_id"))
        and pre_host.get("boot_id") != post_host["boot_id"],
        "boots_since_pre_capture": boots,
        "exactly_one_reboot": boots == 1 and pre_host.get("boot_id") != post_host["boot_id"],
        "database_identity_changes": sorted(identity_changes),
        "failing_pod_diff": pod_diff,
        "passed": boots == 1
        and pre_host.get("boot_id") != post_host["boot_id"]
        and not identity_changes
        and pod_diff["no_new_failures"],
    }


def tunnel_answers(name: str) -> bool:
    try:
        completed = subprocess.run(
            ["bridge", "check", name], text=True, capture_output=True, check=False, timeout=30
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    rows = [line.split() for line in completed.stdout.splitlines() if line.startswith(name)]
    return completed.returncode == 0 and bool(rows) and rows[0][-1] == "yes"


def reboot_state(remote: Remote, now: datetime) -> dict[str, Any]:
    workloads = remote.kubectl_json(
        ["get", "deployments,statefulsets", "-A"], label="read core workload state"
    )
    pods = remote.kubectl_json(["get", "pods", "-A"], label="read all-namespace pod phases")
    clusters = remote.kubectl_json(
        ["get", "clusters.postgresql.cnpg.io", "-A"], label="discover CNPG clusters"
    )
    pvcs = remote.kubectl_json(["get", "pvc", "-A"], label="read PVC identities")
    pvc_uids = {
        f"{item['metadata']['namespace']}/{item['metadata']['name']}": item["metadata"]["uid"]
        for item in pvcs.get("items", [])
    }
    terminating_pvcs = sorted(
        f"{item['metadata']['namespace']}/{item['metadata']['name']}"
        for item in pvcs.get("items", [])
        if item["metadata"].get("deletionTimestamp")
    )
    cnpg: dict[str, Any] = {}
    for cluster in clusters.get("items", []):
        ns, name = cluster["metadata"]["namespace"], cluster["metadata"]["name"]
        ns_pvcs = {
            key.split("/", 1)[1]: uid for key, uid in pvc_uids.items() if key.startswith(f"{ns}/")
        }
        identity = cnpg_identity(cluster, ns_pvcs)
        if identity["current_primary"]:
            controldata = remote.kubectl(
                [
                    "-n", ns, "exec", identity["current_primary"], "-c", "postgres", "--",
                    "pg_controldata", "/var/lib/postgresql/data/pgdata",
                ],
                label=f"read {ns}/{name} system identifier",
            )
            identity["system_id"] = system_identifier(controldata)
        cnpg[f"{ns}/{name}"] = identity
    backups = remote.kubectl_json(["get", "backups.postgresql.cnpg.io", "-A"], label="read CNPG backup metadata")
    stores = remote.kubectl_json(["get", "clustersecretstores"], label="read ESO store conditions")
    projections = remote.kubectl_json(["get", "externalsecrets", "-A"], label="read ESO projection conditions")
    ufw = remote.run(["sudo", "-n", "ufw", "status"], label="read host firewall state")
    ntp = remote.run(
        ["timedatectl", "show", "-p", "NTPSynchronized", "--value"], label="read time sync state"
    )
    return {
        "host": {
            "boot_id": remote.run(["cat", "/proc/sys/kernel/random/boot_id"], label="read boot id"),
            "ntp_synchronized": ntp.strip() == "yes",
            "firewall": ufw_state(ufw),
        },
        "k3s_api_tunnel_answers": tunnel_answers(K3S_API_TUNNEL),
        "core_platform_workloads": core_platform_workloads(workloads),
        "failing_pods": failing_pods(pods),
        "terminating_pvcs": terminating_pvcs,
        "cnpg_clusters": cnpg,
        "missing_expected_cnpg_clusters": sorted(EXPECTED_CNPG_CLUSTERS - set(cnpg)),
        "cnpg_backups": latest_backups_by_cluster(backups, now),
        "eso": eso_inventory(stores, projections),
    }


def reboot_automated_pass(extra: dict[str, Any], max_backup_age_hours: float) -> dict[str, bool]:
    backups = extra["cnpg_backups"]
    return {
        "ntp_synchronized": extra["host"]["ntp_synchronized"],
        "firewall_matches_declaration": extra["host"]["firewall"]["matches_declaration"],
        "k3s_api_tunnel_answers": extra["k3s_api_tunnel_answers"],
        "core_platform_workloads_ready": extra["core_platform_workloads"]["ready"],
        "all_cnpg_clusters_healthy": not extra["missing_expected_cnpg_clusters"]
        and all(cnpg_healthy(item) for item in extra["cnpg_clusters"].values()),
        "required_backups_fresh": all(
            backups.get(key) is not None and backups[key]["age_hours"] <= max_backup_age_hours
            for key in FRESH_BACKUP_REQUIRED
        ),
        "all_eso_stores_valid_and_synced": extra["eso"]["ready"],
    }


def reboot_preflight(remote: Remote, args: argparse.Namespace) -> dict[str, Any]:
    now = datetime.now(UTC)
    state = common_state(remote, now)
    extra = reboot_state(remote, now)
    latest = extra["cnpg_backups"].get("databases/platform-pg")
    if latest is None:
        raise PreflightError("no completed platform-pg backup is visible")
    snapshot_evidence = Path(args.openbao_snapshot_evidence).resolve() if args.openbao_snapshot_evidence else None
    snapshot_receipt = None
    if snapshot_evidence:
        snapshot_receipt = validate_snapshot_receipt(
            snapshot_evidence,
            live_openbao=state["openbao"],
            now=now,
            max_age_hours=args.max_snapshot_age_hours,
        )
    # Gate failures stay value-safe gate results; they never abort evidence capture.
    receipt_errors: dict[str, str] = {}
    warden_receipt = None
    try:
        warden_receipt = validate_ops_warden_receipt(Path(args.ops_warden_acceptance_receipt))
        warden_receipt.update(
            verify_installed_warden(
                installed_warden_package(Path(args.warden_tool_python)),
                OPS_WARDEN_REPO,
                warden_receipt["accepted_revision"],
            )
        )
    except PreflightError as exc:
        receipt_errors["ops_warden_acceptance_receipt"] = str(exc)
    console_receipt = None
    if args.provider_console_receipt:
        try:
            console_receipt = validate_console_receipt(
                Path(args.provider_console_receipt), window_id=args.approved_window_id, now=now
            )
        except PreflightError as exc:
            receipt_errors["provider_console_receipt"] = str(exc)
    else:
        receipt_errors["provider_console_receipt"] = "not supplied (RAIL-HO-WP-0011-T09 pending)"
    checks = reboot_automated_pass(extra, args.max_backup_age_hours)
    comparison = None
    if args.pre_reboot_evidence:
        try:
            pre = json.loads(Path(args.pre_reboot_evidence).read_text(encoding="utf-8"))["reboot_evidence"]
        except (OSError, json.JSONDecodeError, KeyError, TypeError) as exc:
            raise PreflightError("pre-reboot evidence is unavailable or has no reboot_evidence") from exc
        boot_list = remote.run(
            ["sudo", "-n", "journalctl", "--list-boots", "-o", "json", "--no-pager"],
            label="read boot list",
        )
        comparison = reboot_comparison(pre, extra, boots_since(boot_list, rfc3339(pre["captured_at"])))
        checks["post_reboot_comparison_passed"] = comparison["passed"]
    else:
        # Pre-reboot: stuck Terminating PVCs must be resolved or explicitly listed.
        listed = set(args.accept_terminating_pvc or [])
        checks["terminating_pvcs_resolved_or_listed"] = set(extra["terminating_pvcs"]) <= listed
    decision = fresh_decision_gates(args.scenario_id, args.human_decision_id)
    gates = {
        "approved_window_id_recorded": bool(args.approved_window_id),
        **decision,
        "ops_warden_revision_accepted_for_attended_login": bool(warden_receipt),
        "openbao_snapshot_evidence_present": bool(snapshot_receipt),
        "unseal_quorum_attested": args.unseal_quorum_attested,
        "provider_console_receipt_verified": bool(console_receipt),
        "host_cluster_platform_database_audit_owners_acknowledged": args.all_owners_ack,
        "railiance_infra_independent_abort_owner_acknowledged": args.infra_abort_owner_ack,
        "attended_abort_operator_named": bool(args.abort_operator),
    }
    automated = (
        automated_common_pass(state)
        and latest["age_hours"] <= args.max_backup_age_hours
        and all(checks.values())
    )
    return {
        "procedure": "railiance01-coordinated-reboot",
        "phase": "post-reboot" if comparison else "pre-reboot",
        "preflight_only": True,
        "automated_checks_passed": automated,
        "automated_checks": checks,
        "ready_for_live_execution": automated and all(gates.values()) and not comparison,
        "operator_gates": gates,
        "latest_platform_pg_backup": latest,
        "max_backup_age_hours": args.max_backup_age_hours,
        "fresh_backup_recommended_for": sorted(FRESH_BACKUP_RECOMMENDED),
        "openbao_snapshot_receipt": snapshot_receipt,
        "ops_warden_acceptance_receipt": warden_receipt,
        "provider_console_receipt": console_receipt,
        "receipt_gate_failures": receipt_errors,
        "max_snapshot_age_hours": args.max_snapshot_age_hours,
        "reboot_evidence": {"captured_at": state["captured_at"], **extra},
        "reboot_comparison": comparison,
        "state": state,
        "secret_values_observed": False,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("procedure", choices=["database-lease", "node-reboot"])
    parser.add_argument("--remote", default="railiance01")
    parser.add_argument("--approved-window-id")
    parser.add_argument("--abort-operator")
    parser.add_argument("--audit-core-owner-ack", action="store_true")
    parser.add_argument("--rapp-postgres-owner-ack", action="store_true")
    parser.add_argument("--synthetic-load-id")
    parser.add_argument("--openbao-snapshot-evidence")
    parser.add_argument("--unseal-quorum-attested", action="store_true")
    parser.add_argument(
        "--provider-console-receipt",
        help="railiance-infra RAIL01-CONSOLE-<ts>.yaml receipt for this window",
    )
    parser.add_argument("--infra-abort-owner-ack", action="store_true")
    parser.add_argument("--scenario-id", help="new scenario id; consumed drill ids are denied")
    parser.add_argument("--human-decision-id", help="fresh human GO decision id")
    parser.add_argument("--ops-warden-acceptance-receipt", default=OPS_WARDEN_RECEIPT)
    parser.add_argument(
        "--warden-tool-python",
        default=str(Path.home() / ".local/share/uv/tools/ops-warden/bin/python"),
        help="Python of the uv tool environment that provides `warden` on PATH",
    )
    parser.add_argument(
        "--accept-terminating-pvc", action="append",
        help="namespace/name of a pre-existing Terminating PVC accepted for the window",
    )
    parser.add_argument(
        "--pre-reboot-evidence",
        help="pre-reboot preflight JSON; switches node-reboot to post-reboot comparison",
    )
    parser.add_argument("--all-owners-ack", action="store_true")
    parser.add_argument("--max-backup-age-hours", type=float, default=26.0)
    parser.add_argument("--max-snapshot-age-hours", type=float, default=24.0)
    args = parser.parse_args()
    try:
        remote = Remote(args.remote)
        result = (
            database_lease_preflight(remote, args)
            if args.procedure == "database-lease"
            else reboot_preflight(remote, args)
        )
    except (OSError, ValueError, KeyError, PreflightError) as exc:
        print(f"recovery preflight failed: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


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