#!/usr/bin/env python3
"""Shared safety, planning, ledger, and registry helpers for K3s Phase 2.

Phase 2 is deliberately dry-run only.  These helpers make the complete future
transaction inspectable without granting the Phase 2 launcher a live mutation
surface.  Phase 3 may reuse the same ledger and plan contracts after a reviewed
receipt authorizes a single named candidate.
"""
from __future__ import annotations

import copy
import hashlib
import ipaddress
import json
import os
import re
import stat
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence

from phase1_common import atomic_write_json, read_json, safe_name, utc_now


class Phase2Error(RuntimeError):
    """Raised when a Phase 2 contract cannot be proven safely."""


HOSTNAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
MACHINE_ID_RE = re.compile(r"^[0-9a-f]{32}$")
HEX64_RE = re.compile(r"^[0-9a-f]{64}$")
BOOTSTRAP_TOKEN_RE = re.compile(r"(?i)\b[a-z0-9]{6}\.[a-z0-9]{16}\b")
K3S_SECURE_TOKEN_RE = re.compile(r"(?i)\bK10[0-9a-f]{64}::[a-z0-9_-]+:[^\s]+")
AGE_SECRET_RE = re.compile(r"AGE-SECRET-KEY-[A-Z0-9-]+")
PRIVATE_KEY_RE = re.compile(r"-----BEGIN (?:OPENSSH |RSA |EC |)PRIVATE KEY-----")
SENSITIVE_KEY_TERMS = {
    "password",
    "secret_value",
    "credential_value",
    "private_key",
    "client_key",
    "identity_value",
    "authorization_value",
    "kubeconfig_data",
    "server_token",
    "agent_token",
    "token_value",
}
TAILSCALE_IPV4_NETWORK = ipaddress.ip_network("100.64.0.0/10")

PLAN_STEP_DEFINITIONS: tuple[tuple[str, str, str, tuple[str, ...]], ...] = (
    (
        "identity-discovery",
        "Resolve exactly one online Tailscale peer and pin name, DNS name and IPv4.",
        "read-only",
        (),
    ),
    (
        "uniqueness-gate",
        "Reject candidate name, machine-id or Tailscale-IP collisions across Tailscale, Kubernetes and both registries.",
        "read-only",
        (),
    ),
    (
        "supported-host-preflight",
        "Verify supported Linux/systemd/cgroup architecture, capacity, clock and non-interactive root path.",
        "read-only",
        (),
    ),
    (
        "transaction-ledger-open",
        "Create the candidate-scoped durable transaction ledger and bind it to the deterministic plan digest.",
        "phase3-write",
        ("remove-ledger-after-conclusive-rollback",),
    ),
    (
        "recovery-door-snapshot",
        "Snapshot current 2222, 2223 and Tailscale-SSH state before touching access configuration.",
        "phase3-write",
        ("restore-recovery-door-snapshot",),
    ),
    (
        "recovery-door-converge",
        "Converge primary SSH, independent rescue SSH and Tailscale SSH serially, proving the prior door before each transition.",
        "phase3-write",
        ("restore-recovery-door-snapshot", "prove-at-least-one-door"),
    ),
    (
        "host-profile-converge",
        "Converge the declared buildbox host profile through the existing fleet/buildbox machinery.",
        "phase3-write",
        ("restore-host-profile-snapshot",),
    ),
    (
        "bootstrap-token-create",
        "Create one short-lived agent bootstrap token on the server without recording its value.",
        "phase3-secret",
        ("revoke-bootstrap-token",),
    ),
    (
        "agent-config-stage",
        "Stage pinned K3s agent configuration using the canonical endpoint, node IP and tailscale0 interface.",
        "phase3-write",
        ("restore-agent-config-snapshot",),
    ),
    (
        "agent-install-and-join",
        "Install the pinned launcher and start the K3s agent with the temporary token supplied only over a protected pipe.",
        "phase3-write",
        ("stop-and-uninstall-agent", "restore-agent-config-snapshot"),
    ),
    (
        "node-identity-ready-gate",
        "Require one Kubernetes Node whose name, UID, machine-id evidence and InternalIP match the pinned candidate.",
        "read-only",
        ("delete-candidate-node",),
    ),
    (
        "protected-metadata-quarantine",
        "Apply controller-owned protected labels plus a pending NoSchedule taint; the candidate cannot self-assert trust.",
        "phase3-write",
        ("remove-protected-metadata", "cordon-candidate"),
    ),
    (
        "node-pinned-proof-job",
        "Run a transaction-unique proof Job pinned to the candidate and capture API-backed completion evidence.",
        "phase3-write",
        ("delete-proof-job",),
    ),
    (
        "bootstrap-token-revoke",
        "Revoke the temporary token after proof and verify the agent remains Ready using issued client credentials.",
        "phase3-write",
        (),
    ),
    (
        "tokenless-restart-proof",
        "Restart the agent without bootstrap material and repeat identity/Ready checks.",
        "phase3-write",
        ("stop-and-uninstall-agent",),
    ),
    (
        "registry-pair-preview",
        "Generate validated fleet.json and buildbox-hosts.json replacements with execution=none and no order membership.",
        "dry-run-output",
        ("restore-registry-pair",),
    ),
    (
        "final-audit-and-receipt",
        "Prove all recovery doors and source-of-truth invariants, then emit the redacted enrollment receipt.",
        "read-only",
        (),
    ),
)


@dataclass(frozen=True)
class CandidateIdentity:
    name: str
    dns_name: str
    tailscale_ipv4: str
    machine_id: str
    os_id: str
    os_version_id: str
    architecture: str
    ssh_user: str
    rustdesk: str | None = None

    @classmethod
    def from_mapping(cls, value: Mapping[str, Any]) -> "CandidateIdentity":
        required = {
            "name",
            "dns_name",
            "tailscale_ipv4",
            "machine_id",
            "os_id",
            "os_version_id",
            "architecture",
            "ssh_user",
        }
        missing = sorted(required - set(value))
        if missing:
            raise Phase2Error(f"candidate identity is missing: {', '.join(missing)}")
        candidate = cls(
            name=str(value["name"]),
            dns_name=str(value["dns_name"]),
            tailscale_ipv4=str(value["tailscale_ipv4"]),
            machine_id=str(value["machine_id"]),
            os_id=str(value["os_id"]),
            os_version_id=str(value["os_version_id"]),
            architecture=str(value["architecture"]),
            ssh_user=str(value["ssh_user"]),
            rustdesk=None if value.get("rustdesk") in {None, ""} else str(value.get("rustdesk")),
        )
        candidate.validate()
        return candidate

    def validate(self) -> None:
        if not HOSTNAME_RE.fullmatch(self.name):
            raise Phase2Error(f"invalid candidate name: {self.name!r}")
        if not self.dns_name or any(ch.isspace() for ch in self.dns_name):
            raise Phase2Error("candidate DNS name is invalid")
        try:
            ip = ipaddress.ip_address(self.tailscale_ipv4)
        except ValueError as exc:
            raise Phase2Error(f"candidate Tailscale address is invalid: {self.tailscale_ipv4}") from exc
        if ip.version != 4 or ip not in TAILSCALE_IPV4_NETWORK:
            raise Phase2Error("candidate must use a Tailscale IPv4 address in 100.64.0.0/10")
        if not MACHINE_ID_RE.fullmatch(self.machine_id):
            raise Phase2Error("candidate machine-id must be 32 lowercase hex characters")
        if self.os_id not in {"debian", "ubuntu"}:
            raise Phase2Error(f"unsupported candidate OS: {self.os_id!r}")
        if self.architecture not in {"x86_64", "amd64"}:
            raise Phase2Error(f"unsupported candidate architecture: {self.architecture!r}")
        if not HOSTNAME_RE.fullmatch(self.ssh_user):
            raise Phase2Error(f"invalid candidate SSH user: {self.ssh_user!r}")
        if self.rustdesk is not None and not re.fullmatch(r"[0-9]{6,16}", self.rustdesk):
            raise Phase2Error("RustDesk peer id must be 6-16 digits or null")

    def as_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "dns_name": self.dns_name,
            "tailscale_ipv4": self.tailscale_ipv4,
            "machine_id": self.machine_id,
            "os_id": self.os_id,
            "os_version_id": self.os_version_id,
            "architecture": self.architecture,
            "ssh_user": self.ssh_user,
            "rustdesk": self.rustdesk,
        }


def canonical_json_bytes(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def sha256_json(value: Any) -> str:
    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()


def deep_copy_json(value: Any) -> Any:
    return json.loads(json.dumps(value))


def validate_sha256(value: str, label: str) -> None:
    if not HEX64_RE.fullmatch(value):
        raise Phase2Error(f"{label} must be a lowercase SHA-256 digest")


def find_secret_strings(value: Any, *, path: str = "$", findings: list[str] | None = None) -> list[str]:
    out = findings if findings is not None else []
    if isinstance(value, Mapping):
        for key, child in value.items():
            normalized = str(key).lower().replace("-", "_")
            if any(term in normalized for term in SENSITIVE_KEY_TERMS):
                permitted = child is None or child is False or (
                    isinstance(child, str) and child in {"", "<redacted>", "not-recorded"}
                )
                if not permitted:
                    out.append(f"{path}.{key}: sensitive key carries a value")
            find_secret_strings(child, path=f"{path}.{key}", findings=out)
    elif isinstance(value, list):
        for index, child in enumerate(value):
            find_secret_strings(child, path=f"{path}[{index}]", findings=out)
    elif isinstance(value, str):
        for label, pattern in (
            ("bootstrap token", BOOTSTRAP_TOKEN_RE),
            ("K3s secure token", K3S_SECURE_TOKEN_RE),
            ("age secret key", AGE_SECRET_RE),
            ("private key", PRIVATE_KEY_RE),
        ):
            if pattern.search(value):
                out.append(f"{path}: contains {label}")
    return out


def assert_secret_free(value: Any, label: str = "document") -> None:
    findings = find_secret_strings(value)
    if findings:
        raise Phase2Error(f"{label} contains secret-like material: {'; '.join(findings[:8])}")


def validate_candidate_uniqueness(
    identity: CandidateIdentity,
    fleet: Mapping[str, Any],
    hosts: Mapping[str, Any],
    cluster_nodes: Sequence[Mapping[str, Any]],
    tailscale_peers: Sequence[Mapping[str, Any]],
) -> dict[str, Any]:
    collisions: list[str] = []
    nodes = fleet.get("nodes")
    if isinstance(nodes, Mapping) and identity.name in nodes:
        collisions.append(f"fleet node name {identity.name!r} already exists")
    host_items = hosts.get("hosts")
    if not isinstance(host_items, list):
        raise Phase2Error("buildbox registry has no hosts array")
    for host in host_items:
        if not isinstance(host, Mapping):
            continue
        if host.get("name") == identity.name or host.get("ssh_alias") == identity.name:
            collisions.append(f"buildbox host name/alias {identity.name!r} already exists")
        if host.get("machine_id") == identity.machine_id:
            collisions.append(f"machine-id already belongs to buildbox host {host.get('name')!r}")
        access = host.get("access")
        if isinstance(access, Mapping):
            for door_name, door in access.items():
                if isinstance(door, Mapping) and door.get("host") == identity.tailscale_ipv4:
                    collisions.append(f"Tailscale IP already belongs to {host.get('name')!r}.{door_name}")
    for node in cluster_nodes:
        if not isinstance(node, Mapping):
            continue
        if node.get("name") == identity.name or node.get("hostname") == identity.name:
            collisions.append(f"Kubernetes node name {identity.name!r} already exists")
        if node.get("internal_ip") == identity.tailscale_ipv4:
            collisions.append(f"Kubernetes InternalIP already belongs to {node.get('name')!r}")
    peer_matches = []
    for peer in tailscale_peers:
        if not isinstance(peer, Mapping):
            continue
        names = {str(peer.get("name") or ""), str(peer.get("dns_name") or "").split(".")[0]}
        ips = {str(item) for item in peer.get("ips", []) if isinstance(item, str)}
        if identity.name in names or identity.dns_name == str(peer.get("dns_name") or "") or identity.tailscale_ipv4 in ips:
            peer_matches.append(peer)
    exact = [
        peer
        for peer in peer_matches
        if identity.tailscale_ipv4 in {str(item) for item in peer.get("ips", []) if isinstance(item, str)}
        and identity.name
        in {str(peer.get("name") or ""), str(peer.get("dns_name") or "").split(".")[0]}
    ]
    if len(exact) != 1:
        collisions.append(f"expected one exact online Tailscale identity; found {len(exact)}")
    elif exact[0].get("online") is not True:
        collisions.append("the exact Tailscale peer is not online")
    if collisions:
        raise Phase2Error("candidate uniqueness gate failed: " + "; ".join(collisions))
    return {
        "status": "passed",
        "candidate": identity.name,
        "exact_tailscale_matches": 1,
        "fleet_name_unique": True,
        "registry_identity_unique": True,
        "cluster_identity_unique": True,
    }


def validate_host_preflight(preflight: Mapping[str, Any]) -> dict[str, Any]:
    required_true = (
        "systemd",
        "cgroup_v2",
        "tailscale_online",
        "tailscale_interface",
        "sudo_noninteractive",
        "clock_synchronized",
    )
    failures = [name for name in required_true if preflight.get(name) is not True]
    memory = int(preflight.get("memory_bytes") or 0)
    disk = int(preflight.get("disk_free_bytes") or 0)
    if memory < 4 * 1024**3:
        failures.append("memory_bytes<4GiB")
    if disk < 20 * 1024**3:
        failures.append("disk_free_bytes<20GiB")
    existing = str(preflight.get("k3s_agent_state") or "absent")
    if existing not in {"absent", "inactive"}:
        failures.append(f"k3s_agent_state={existing}")
    if failures:
        raise Phase2Error("candidate preflight failed: " + ", ".join(failures))
    return {
        "status": "passed",
        "memory_bytes": memory,
        "disk_free_bytes": disk,
        "k3s_agent_state": existing,
        "required_checks": list(required_true),
    }


def _validate_fleet_shape(fleet: Mapping[str, Any]) -> None:
    if fleet.get("schema_version") != 1 or not isinstance(fleet.get("nodes"), Mapping):
        raise Phase2Error("unsupported fleet.json structure")
    fallback = fleet.get("fallback")
    if not isinstance(fallback, Mapping):
        raise Phase2Error("fleet.json fallback block is missing")


def _validate_host_registry_shape(hosts: Mapping[str, Any]) -> None:
    if hosts.get("schema_version") != 1 or not isinstance(hosts.get("hosts"), list):
        raise Phase2Error("unsupported buildbox-hosts.json structure")
    if not isinstance(hosts.get("orders"), Mapping):
        raise Phase2Error("buildbox registry orders block is missing")


def build_registry_previews(
    fleet: Mapping[str, Any],
    hosts: Mapping[str, Any],
    identity: CandidateIdentity,
    *,
    identity_file: str = "~/.ssh/id_ed25519_buildbox",
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
    _validate_fleet_shape(fleet)
    _validate_host_registry_shape(hosts)
    fleet_out = deep_copy_json(fleet)
    hosts_out = deep_copy_json(hosts)
    if identity.name in fleet_out["nodes"]:
        raise Phase2Error(f"fleet preview would replace existing node {identity.name!r}")
    if any(item.get("name") == identity.name for item in hosts_out["hosts"] if isinstance(item, dict)):
        raise Phase2Error(f"host preview would replace existing host {identity.name!r}")
    fleet_out["nodes"][identity.name] = {
        "transport": "ssh",
        "host_ref": identity.name,
        "roles": ["builder", "agent-runtime"],
        "profiles": ["shared-agent-tools", "buildbox", "buildbox-root"],
        "execution": "none",
    }
    host_entry = {
        "name": identity.name,
        "ssh_alias": identity.name,
        "state": "reachable",
        "machine_id": identity.machine_id,
        "roles": ["builder", "agent-seat", "e2e", "agent-sandbox"],
        "access": {
            "lan": None,
            "tailscale_ip": {
                "host": identity.tailscale_ipv4,
                "port": 2222,
                "user": identity.ssh_user,
                "identity_file": identity_file,
            },
            "tailscale_ssh": {
                "host": identity.dns_name,
                "port": 22,
                "user": identity.ssh_user,
                "identity_file": None,
            },
        },
        "rustdesk": identity.rustdesk,
        "notes": "Planned Phase 3 enrollment; execution remains disabled until live proof and a separate dispatch-authorization change.",
    }
    hosts_out["hosts"].append(host_entry)
    for order_name, order in hosts_out["orders"].items():
        if identity.name in order:
            raise Phase2Error(f"candidate must not enter registry order {order_name!r} during enrollment")
    for name in fleet_out.get("fallback", {}).get("requires_all_unavailable", []):
        if name == identity.name:
            raise Phase2Error("candidate must not enter fallback dependencies during enrollment")
    proof = {
        "schema_version": 1,
        "candidate": identity.name,
        "fleet_sha256": sha256_json(fleet_out),
        "hosts_sha256": sha256_json(hosts_out),
        "execution": "none",
        "in_build_order": False,
        "in_e2e_order": False,
        "in_fallback_dependencies": False,
        "pair_commit_required": True,
    }
    assert_secret_free(fleet_out, "fleet preview")
    assert_secret_free(hosts_out, "buildbox registry preview")
    return fleet_out, hosts_out, proof


def build_enrollment_plan(
    identity: CandidateIdentity,
    *,
    control_plane: Mapping[str, Any],
    version_lock: Mapping[str, Any],
    source_digests: Mapping[str, str],
    recovery_doors: Sequence[Mapping[str, Any]],
) -> dict[str, Any]:
    endpoint = str(control_plane.get("api", {}).get("endpoint") or "")
    ca_sha = str(control_plane.get("api", {}).get("cacerts_sha256") or "")
    version = str(version_lock.get("version") or "")
    launcher = version_lock.get("launcher") if isinstance(version_lock.get("launcher"), Mapping) else {}
    launcher_sha = str(launcher.get("sha256") or "")
    if not endpoint.startswith("https://"):
        raise Phase2Error("control-plane contract has no canonical HTTPS endpoint")
    validate_sha256(ca_sha, "control-plane CA hash")
    validate_sha256(launcher_sha, "K3s launcher hash")
    if not version.startswith("v"):
        raise Phase2Error("version lock has no pinned K3s version")
    for label, digest in source_digests.items():
        validate_sha256(str(digest), f"source digest {label}")
    if len(recovery_doors) != 3:
        raise Phase2Error("exactly three recovery-door definitions are required")
    base = {
        "schema_version": 1,
        "phase": 2,
        "mode": "dry-run",
        "live_mutation_allowed": False,
        "candidate": identity.as_dict(),
        "cluster": {
            "endpoint": endpoint,
            "cacerts_sha256": ca_sha,
            "k3s_version": version,
            "launcher_sha256": launcher_sha,
            "new_agent_node_ip": identity.tailscale_ipv4,
            "new_agent_flannel_interface": "tailscale0",
        },
        "source_digests": dict(sorted((str(k), str(v)) for k, v in source_digests.items())),
        "recovery_doors": [deep_copy_json(item) for item in recovery_doors],
        "bootstrap_token": {
            "kind": "temporary-kubeadm-style-agent-token",
            "ttl_seconds": 600,
            "value_recorded": False,
            "revoke_before_finalization": True,
        },
        "scheduling": {
            "protected_label_prefix": "node-restriction.kubernetes.io/",
            "initial_taint": "overdeck.io/enrollment=pending:NoSchedule",
            "proof_job_node_pinned": True,
            "registry_execution": "none",
        },
        "steps": [],
    }
    seed = sha256_json(base)
    transaction_id = f"od-enroll-{safe_name(identity.name)}-{seed[:16]}"
    steps: list[dict[str, Any]] = []
    for index, (name, description, boundary, rollback) in enumerate(PLAN_STEP_DEFINITIONS, start=1):
        steps.append(
            {
                "ordinal": index,
                "name": name,
                "description": description,
                "boundary": boundary,
                "resume_policy": "verify-then-continue",
                "rollback": list(rollback),
            }
        )
    base["transaction_id"] = transaction_id
    base["steps"] = steps
    base["rollback_order"] = [
        item
        for step in reversed(steps)
        for item in step["rollback"]
        if item
    ]
    base["plan_sha256"] = sha256_json(base)
    assert_secret_free(base, "enrollment plan")
    return base


class EnrollmentLedger:
    """Atomic, append-only logical ledger for one enrollment transaction.

    The file itself is rewritten atomically so a torn append cannot make a later
    run guess.  Event sequencing is validated against the deterministic plan.
    """

    def __init__(self, path: Path, plan: Mapping[str, Any], *, create: bool = False) -> None:
        self.path = path
        self.plan = deep_copy_json(plan)
        if create:
            if path.exists() or path.is_symlink():
                raise Phase2Error(f"refusing to replace enrollment ledger: {path}")
            document = {
                "schema_version": 1,
                "transaction_id": plan.get("transaction_id"),
                "candidate": plan.get("candidate", {}).get("name"),
                "plan_sha256": plan.get("plan_sha256"),
                "status": "planned",
                "events": [],
                "created_utc": utc_now(),
                "updated_utc": utc_now(),
            }
            atomic_write_json(path, document, mode=0o600)
        self.document = self._load()
        self._validate_binding()

    def _load(self) -> dict[str, Any]:
        document = read_json(self.path)
        if not isinstance(document, dict) or document.get("schema_version") != 1:
            raise Phase2Error(f"unsupported enrollment ledger: {self.path}")
        if not isinstance(document.get("events"), list):
            raise Phase2Error("enrollment ledger events must be an array")
        return document

    def _validate_binding(self) -> None:
        for key in ("transaction_id", "plan_sha256"):
            if self.document.get(key) != self.plan.get(key):
                raise Phase2Error(f"enrollment ledger {key} does not match the plan")
        if self.document.get("candidate") != self.plan.get("candidate", {}).get("name"):
            raise Phase2Error("enrollment ledger candidate does not match the plan")

    def _write(self) -> None:
        self.document["updated_utc"] = utc_now()
        assert_secret_free(self.document, "enrollment ledger")
        atomic_write_json(self.path, self.document, mode=0o600)

    @property
    def completed_steps(self) -> list[str]:
        return [
            str(event["step"])
            for event in self.document["events"]
            if event.get("event") == "completed" and isinstance(event.get("step"), str)
        ]

    def next_step(self) -> str | None:
        completed = set(self.completed_steps)
        for step in self.plan.get("steps", []):
            name = step.get("name")
            if isinstance(name, str) and name not in completed:
                return name
        return None

    def record(self, step: str, event: str, **detail: Any) -> None:
        valid_steps = [str(item.get("name")) for item in self.plan.get("steps", [])]
        if step not in valid_steps:
            raise Phase2Error(f"ledger event names unknown step: {step}")
        if event not in {"started", "completed", "failed", "rolled-back", "verified"}:
            raise Phase2Error(f"unsupported ledger event: {event}")
        if event in {"started", "completed"}:
            expected = self.next_step()
            if step != expected:
                raise Phase2Error(f"ledger expected step {expected!r}, not {step!r}")
        item = {"sequence": len(self.document["events"]) + 1, "at_utc": utc_now(), "step": step, "event": event}
        item.update(detail)
        assert_secret_free(item, "ledger event")
        self.document["events"].append(item)
        if event == "failed":
            self.document["status"] = "failed"
        elif event == "rolled-back":
            self.document["status"] = "rolled-back"
        elif event == "completed" and self.next_step() is None:
            self.document["status"] = "completed"
        elif event in {"started", "completed", "verified"}:
            self.document["status"] = "running"
        self._write()

    def pending_rollback(self) -> list[str]:
        completed = set(self.completed_steps)
        actions: list[str] = []
        for step in reversed(self.plan.get("steps", [])):
            if step.get("name") not in completed:
                continue
            for action in step.get("rollback", []):
                if action not in actions:
                    actions.append(str(action))
        return actions


def secure_regular_file(path: Path, *, owner_uid: int | None = None, max_mode: int = 0o600) -> os.stat_result:
    if path.is_symlink():
        raise Phase2Error(f"refusing symlink: {path}")
    try:
        info = path.stat()
    except OSError as exc:
        raise Phase2Error(f"cannot stat {path}: {exc}") from exc
    if not stat.S_ISREG(info.st_mode):
        raise Phase2Error(f"required regular file: {path}")
    if owner_uid is not None and info.st_uid != owner_uid:
        raise Phase2Error(f"unexpected owner for {path}: uid {info.st_uid}")
    if stat.S_IMODE(info.st_mode) & ~max_mode:
        raise Phase2Error(f"permissions too broad for {path}: {stat.filemode(info.st_mode)}")
    return info


def write_json_pair_atomic(
    first_path: Path,
    first_value: Any,
    second_path: Path,
    second_value: Any,
    *,
    mode: int = 0o600,
) -> None:
    """Publish two files with rollback if the second rename fails.

    This helper is used only in tests and future Phase 3 registry publication.
    Phase 2 writes previews into a receipt directory, never into the repository.
    """

    if first_path.parent != second_path.parent:
        raise Phase2Error("atomic pair publication requires one directory")
    directory = first_path.parent
    directory.mkdir(parents=True, exist_ok=True)
    originals: dict[Path, bytes | None] = {}
    modes: dict[Path, int] = {}
    for path in (first_path, second_path):
        if path.is_symlink():
            raise Phase2Error(f"refusing symlink pair target: {path}")
        if path.exists():
            originals[path] = path.read_bytes()
            modes[path] = stat.S_IMODE(path.stat().st_mode)
        else:
            originals[path] = None
    temp_paths: list[Path] = []
    try:
        for path, value in ((first_path, first_value), (second_path, second_value)):
            fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=directory)
            tmp = Path(tmp_name)
            temp_paths.append(tmp)
            with os.fdopen(fd, "wb") as handle:
                handle.write(json.dumps(value, indent=2, sort_keys=False).encode("utf-8") + b"\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.chmod(tmp, mode)
        os.replace(temp_paths[0], first_path)
        os.replace(temp_paths[1], second_path)
        dir_fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
        try:
            os.fsync(dir_fd)
        finally:
            os.close(dir_fd)
    except BaseException:
        for path in (first_path, second_path):
            original = originals[path]
            if original is None:
                path.unlink(missing_ok=True)
            else:
                fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.rollback.", dir=directory)
                tmp = Path(tmp_name)
                with os.fdopen(fd, "wb") as handle:
                    handle.write(original)
                    handle.flush()
                    os.fsync(handle.fileno())
                os.chmod(tmp, modes[path])
                os.replace(tmp, path)
        raise
    finally:
        for tmp in temp_paths:
            tmp.unlink(missing_ok=True)


def recovery_door_contract(identity: CandidateIdentity) -> list[dict[str, Any]]:
    return [
        {
            "name": "primary-sshd",
            "host": identity.tailscale_ipv4,
            "port": 2222,
            "owner": "ssh.service + buildbox-sshd-access",
            "independence": "wildcard-bound OpenSSH; does not depend on tailscaled",
        },
        {
            "name": "rescue-sshd",
            "host": identity.tailscale_ipv4,
            "port": 2223,
            "owner": "buildbox-rescue-sshd.socket",
            "independence": "separate config and host key outside /etc/ssh",
        },
        {
            "name": "tailscale-ssh",
            "host": identity.dns_name,
            "port": 22,
            "owner": "tailscaled + buildbox-tailscale-ssh.timer",
            "independence": "not served by system OpenSSH configuration",
        },
    ]


def summarize_plan(plan: Mapping[str, Any]) -> str:
    lines = [
        f"transaction: {plan.get('transaction_id')}",
        f"candidate: {plan.get('candidate', {}).get('name')}",
        f"mode: {plan.get('mode')}",
        f"plan_sha256: {plan.get('plan_sha256')}",
        f"steps: {len(plan.get('steps', []))}",
        "live_mutation_allowed: false",
    ]
    for step in plan.get("steps", []):
        lines.append(f"{int(step['ordinal']):02d}. {step['name']} [{step['boundary']}]")
    return "\n".join(lines) + "\n"


def ensure_exact_keys(mapping: Mapping[str, Any], expected: Iterable[str], label: str) -> None:
    actual = set(mapping)
    wanted = set(expected)
    if actual != wanted:
        raise Phase2Error(f"{label} keys differ: missing={sorted(wanted-actual)} unknown={sorted(actual-wanted)}")