#!/usr/bin/env python3
"""Converge four Quote approval metadata blobs to singular action references.

This narrow production reconcile treats four retained XML references as authority
only after one explicit sanitization: user names inside ``assignedApprover``
blocks are replaced with the manifest-bound placeholder.  The sanitized XML
must be semantically identical to one of the manifest's exact before/after
states; formatting-only inter-element whitespace is not authoritative.  Live
identities are never copied from the seed and never emitted as evidence.

The database is used in two sessions.  The first performs a payload-private
HEX read which is parsed and validated in memory.  Desired live bytes are then
derived by changing only the seven plural wrapper tag pairs in those live
bytes.  The second session locks every broad name match and CAS-updates the
snapshot only when all four original rows are byte-for-byte unchanged.
"""

from __future__ import annotations

import argparse
import base64
from dataclasses import dataclass
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
from typing import Any, Callable, Sequence
import xml.etree.ElementTree as ET


FORMAT = "quote-approval-action-wrapper-reconcile-v2"
EVIDENCE_FORMAT = "quote-approval-action-wrapper-reconcile-evidence-v2"
SOURCE_RELPATH = "deployments/offline/runtime/reconcile/quote_approval_action_wrapper_reference.json"
DEFAULT_MANIFEST = Path(__file__).with_name("quote_approval_action_wrappers.v2.json")
TARGET_DEVELOPER_NAMES = (
    "NA_OP_Service_250502",
    "NA_OP_Service_Partner_250502",
    "NA_DBaaS_250502",
    "NA_DBaaS_Partner_250502",
)
EXPECTED_USER_APPROVER_COUNTS = (6, 8, 6, 8)
IDENTITY_REDACTION = {
    "kind": "assigned-approver-user-name-v1",
    "placeholder": "admin@example.invalid",
}
IDENTITY_PLACEHOLDER = IDENTITY_REDACTION["placeholder"].encode("utf-8")
EXPECTED_REFERENCES = (
    ("Approved", "FieldUpdate"),
    ("statusapproved", "FieldUpdate"),
    ("rejected", "FieldUpdate"),
    ("statusrejected", "FieldUpdate"),
    ("Inprocess", "FieldUpdate"),
    ("Inreview", "FieldUpdate"),
    ("pending", "FieldUpdate"),
)
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
GIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
RESET_ATTEMPT_RE = re.compile(r"^[0-9a-f]{32}$")
SENTINEL = "QUOTE_APPROVAL_ACTION_WRAPPER_RECONCILE_RESULT:"
ASSERTION_TABLE = "_quote_approval_action_wrapper_reconcile_assert"
SINGULAR_REFERENCE_RE = re.compile(
    rb"<action>(?P<body>\s*<name>(?P<name>[^<]+)</name>"
    rb"\s*<type>(?P<type>[^<]+)</type>\s*)</action>"
)
PLURAL_REFERENCE_RE = re.compile(
    rb"<actions>(?P<body>\s*<name>(?P<name>[^<]+)</name>"
    rb"\s*<type>(?P<type>[^<]+)</type>\s*)</actions>"
)
ASSIGNED_APPROVER_BLOCK_RE = re.compile(
    rb"(?P<open><assignedApprover>)(?P<body>.*?)(?P<close></assignedApprover>)",
    re.DOTALL,
)
EXPLICIT_USER_APPROVER_RE = re.compile(
    rb"(?P<prefix><approver>\s*<name>)(?P<name>[^<]+)"
    rb"(?P<suffix></name>\s*<type>user</type>.*?</approver>)",
    re.DOTALL,
)


class ReconcileError(RuntimeError):
    """Raised when the reconcile cannot prove its exact mutation contract."""


@dataclass(frozen=True)
class TargetAuthority:
    developer_name: str
    canonical_before_metadata: bytes
    canonical_after_metadata: bytes
    canonical_before_sha256: str
    canonical_after_sha256: str
    explicit_user_approver_count: int


@dataclass(frozen=True)
class ReconcileAuthority:
    manifest_path: Path
    manifest_sha256: str
    source_path: Path
    source_sha256: str
    targets: tuple[TargetAuthority, ...]


@dataclass(frozen=True)
class DeploymentBinding:
    backup_reset_receipt_sha256: str
    backup_reset_identity_sha256: str
    backup_reset_attempt_id: str
    resolved_downstream_ref: str
    resolved_upstream_ref: str
    resolved_official_app_packs_ref: str
    release_fingerprint_sha256: str


@dataclass(frozen=True)
class LiveTarget:
    id_bytes: bytes
    developer_name: str
    developer_name_bytes: bytes
    table_enum_or_id_bytes: bytes
    owner_pack_bytes: bytes
    live_before_metadata: bytes
    live_after_metadata: bytes
    live_before_sha256: str
    live_after_sha256: str
    explicit_user_approver_count: int
    pre_state: str


@dataclass(frozen=True)
class ReconcileResult:
    targets: tuple[LiveTarget, ...]
    updated_rows: int


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _load_json_bytes(path: Path, *, label: str) -> tuple[dict[str, Any], bytes]:
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise ReconcileError(f"unable to read {label}: {path}") from exc
    try:
        payload = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ReconcileError(f"invalid JSON {label}: {path}") from exc
    if not isinstance(payload, dict):
        raise ReconcileError(f"{label} must be a JSON object: {path}")
    return payload, raw


def _require_exact_keys(value: dict[str, Any], expected: set[str], *, label: str) -> None:
    actual = set(value)
    if actual != expected:
        raise ReconcileError(
            f"{label} keys do not match the contract: "
            f"missing={sorted(expected - actual)} unexpected={sorted(actual - expected)}"
        )


def load_manifest(path: Path) -> tuple[dict[str, Any], bytes]:
    payload, raw = _load_json_bytes(path, label="reconcile manifest")
    _require_exact_keys(
        payload,
        {
            "format",
            "table_enum_or_id",
            "owner_pack_state",
            "source_path",
            "identity_redaction",
            "expected_references",
            "targets",
        },
        label="reconcile manifest",
    )
    if payload["format"] != FORMAT:
        raise ReconcileError(f"unsupported reconcile manifest format: {payload['format']!r}")
    if payload["table_enum_or_id"] != "Quote":
        raise ReconcileError("reconcile manifest may target only TableEnumOrId='Quote'")
    if payload["owner_pack_state"] != "empty_string":
        raise ReconcileError("reconcile manifest OwnerPack state must be the reviewed empty string")
    if payload["source_path"] != SOURCE_RELPATH:
        raise ReconcileError("reconcile manifest does not name the reviewed Quote wrapper reference")
    if payload["identity_redaction"] != IDENTITY_REDACTION:
        raise ReconcileError("reconcile manifest identity redaction is not the reviewed contract")

    raw_references = payload["expected_references"]
    if not isinstance(raw_references, list):
        raise ReconcileError("reconcile manifest expected_references must be a list")
    references: list[tuple[str, str]] = []
    for index, raw_reference in enumerate(raw_references):
        if not isinstance(raw_reference, dict):
            raise ReconcileError(f"expected reference {index} must be an object")
        _require_exact_keys(raw_reference, {"name", "type"}, label=f"expected reference {index}")
        references.append((str(raw_reference["name"]), str(raw_reference["type"])))
    if tuple(references) != EXPECTED_REFERENCES:
        raise ReconcileError("reconcile manifest reference sequence is not the reviewed seven-item sequence")

    raw_targets = payload["targets"]
    if not isinstance(raw_targets, list) or len(raw_targets) != len(TARGET_DEVELOPER_NAMES):
        raise ReconcileError("reconcile manifest must contain exactly four targets")
    target_names: list[str] = []
    for index, raw_target in enumerate(raw_targets):
        if not isinstance(raw_target, dict):
            raise ReconcileError(f"reconcile target {index} must be an object")
        _require_exact_keys(
            raw_target,
            {
                "developer_name",
                "canonical_before_metadata_sha256",
                "canonical_after_metadata_sha256",
                "explicit_user_approver_count",
            },
            label=f"reconcile target {index}",
        )
        developer_name = str(raw_target["developer_name"])
        target_names.append(developer_name)
        for key in (
            "canonical_before_metadata_sha256",
            "canonical_after_metadata_sha256",
        ):
            digest = str(raw_target[key])
            if not SHA256_RE.fullmatch(digest):
                raise ReconcileError(f"reconcile target {developer_name} has invalid {key}")
        if (
            raw_target["canonical_before_metadata_sha256"]
            == raw_target["canonical_after_metadata_sha256"]
        ):
            raise ReconcileError(f"reconcile target {developer_name} has identical before/after hashes")
        explicit_count = raw_target["explicit_user_approver_count"]
        if type(explicit_count) is not int or explicit_count != EXPECTED_USER_APPROVER_COUNTS[index]:
            raise ReconcileError(
                f"reconcile target {developer_name} has an unreviewed explicit approver count"
            )
    if tuple(target_names) != TARGET_DEVELOPER_NAMES or len(set(target_names)) != len(target_names):
        raise ReconcileError("reconcile manifest target order/identity is not the exact reviewed set")
    return payload, raw


def _resolve_source(source_root: Path, relpath: str) -> Path:
    root = source_root.expanduser().resolve(strict=True)
    candidate = (root / relpath).resolve(strict=True)
    try:
        candidate.relative_to(root)
    except ValueError as exc:
        raise ReconcileError(f"reconcile source escapes source root: {relpath}") from exc
    if not candidate.is_file():
        raise ReconcileError(f"reconcile source is not a regular file: {relpath}")
    return candidate


def _reference_sequence(pattern: re.Pattern[bytes], payload: bytes) -> tuple[tuple[str, str], ...]:
    references: list[tuple[str, str]] = []
    for match in pattern.finditer(payload):
        try:
            references.append(
                (match.group("name").decode("utf-8"), match.group("type").decode("utf-8"))
            )
        except UnicodeDecodeError as exc:
            raise ReconcileError("approval metadata reference wrapper is not valid UTF-8") from exc
    return tuple(references)


def _reference_wrapper_state(payload: bytes) -> str:
    singular_references = _reference_sequence(SINGULAR_REFERENCE_RE, payload)
    plural_references = _reference_sequence(PLURAL_REFERENCE_RE, payload)
    if singular_references == EXPECTED_REFERENCES and not plural_references:
        return "singular"
    if plural_references == EXPECTED_REFERENCES and not singular_references:
        return "plural"
    raise ReconcileError(
        "approval metadata must contain either all seven exact singular references "
        "or all seven exact plural references, without a mixed state"
    )


def derive_before_metadata(after_metadata: bytes) -> bytes:
    """Change only the seven reviewed singular wrapper tag pairs to plural."""

    if _reference_wrapper_state(after_metadata) != "singular":
        raise ReconcileError("normalized Quote metadata is not in the reviewed singular state")
    replacements = 0

    def replace(match: re.Match[bytes]) -> bytes:
        nonlocal replacements
        replacements += 1
        return b"<actions>" + match.group("body") + b"</actions>"

    before_metadata = SINGULAR_REFERENCE_RE.sub(replace, after_metadata)
    if replacements != len(EXPECTED_REFERENCES):
        raise ReconcileError("normalized Quote metadata did not yield exactly seven wrapper replacements")
    if _reference_wrapper_state(before_metadata) != "plural":
        raise ReconcileError("derived before metadata is not in the reviewed plural state")
    return before_metadata


def derive_after_metadata(before_metadata: bytes) -> bytes:
    """Change only the seven reviewed plural wrapper tag pairs to singular."""

    if _reference_wrapper_state(before_metadata) != "plural":
        raise ReconcileError("live Quote metadata is not in the reviewed plural state")
    replacements = 0

    def replace(match: re.Match[bytes]) -> bytes:
        nonlocal replacements
        replacements += 1
        return b"<action>" + match.group("body") + b"</action>"

    after_metadata = PLURAL_REFERENCE_RE.sub(replace, before_metadata)
    if replacements != len(EXPECTED_REFERENCES):
        raise ReconcileError("live Quote metadata did not yield exactly seven wrapper replacements")
    if _reference_wrapper_state(after_metadata) != "singular":
        raise ReconcileError("derived live metadata is not in the reviewed singular state")
    return after_metadata


def _local_xml_name(tag: object) -> str:
    raw = str(tag)
    return raw.rsplit("}", 1)[-1]


def _explicit_user_approver_count(root: ET.Element) -> int:
    count = 0
    for assigned_approver in root.iter():
        if _local_xml_name(assigned_approver.tag) != "assignedApprover":
            continue
        for approver in assigned_approver.iter():
            if approver is assigned_approver or _local_xml_name(approver.tag) != "approver":
                continue
            names = [
                child
                for child in list(approver)
                if _local_xml_name(child.tag) == "name"
            ]
            types = [
                child
                for child in list(approver)
                if _local_xml_name(child.tag) == "type"
            ]
            if any((node.text or "").strip() == "user" for node in types):
                if len(names) != 1 or len(types) != 1 or (types[0].text or "").strip() != "user":
                    raise ReconcileError(
                        "assignedApprover user entry does not have one exact name/type pair"
                    )
                if not (names[0].text or ""):
                    raise ReconcileError("assignedApprover user name is empty")
                count += 1
    return count


def canonicalize_metadata(
    metadata: bytes,
    *,
    expected_user_approver_count: int,
    require_placeholder: bool,
) -> bytes:
    """Redact only explicit assignedApprover user-name values."""

    try:
        metadata.decode("utf-8")
        root = ET.fromstring(metadata)
    except (UnicodeDecodeError, ET.ParseError) as exc:
        raise ReconcileError("Quote approval metadata is not valid UTF-8 XML") from exc
    xml_count = _explicit_user_approver_count(root)
    if xml_count != expected_user_approver_count:
        raise ReconcileError(
            "Quote approval metadata explicit user-approver count does not match the manifest"
        )

    replacements = 0

    def redact_block(block_match: re.Match[bytes]) -> bytes:
        nonlocal replacements

        def redact_approver(approver_match: re.Match[bytes]) -> bytes:
            nonlocal replacements
            identity = approver_match.group("name")
            try:
                identity.decode("utf-8")
            except UnicodeDecodeError as exc:
                raise ReconcileError("assignedApprover user name is not valid UTF-8") from exc
            if require_placeholder and identity != IDENTITY_PLACEHOLDER:
                raise ReconcileError(
                    "reviewed Quote wrapper reference contains an unsanitized assignedApprover user name"
                )
            replacements += 1
            return (
                approver_match.group("prefix")
                + IDENTITY_PLACEHOLDER
                + approver_match.group("suffix")
            )

        redacted_body = EXPLICIT_USER_APPROVER_RE.sub(
            redact_approver,
            block_match.group("body"),
        )
        return block_match.group("open") + redacted_body + block_match.group("close")

    canonical = ASSIGNED_APPROVER_BLOCK_RE.sub(redact_block, metadata)
    if replacements != expected_user_approver_count or replacements != xml_count:
        raise ReconcileError(
            "assignedApprover user-name redaction did not cover the exact reviewed entries"
        )
    try:
        ET.fromstring(canonical)
    except ET.ParseError as exc:
        raise ReconcileError("identity-redacted Quote metadata is invalid XML") from exc
    return canonical


def _semantic_xml_tag(tag: object) -> str:
    if tag is ET.Comment:
        return "#comment"
    if tag is ET.ProcessingInstruction:
        return "#processing-instruction"
    return str(tag)


def _semantic_xml_signature(element: ET.Element) -> tuple[object, ...]:
    """Return the exact ordered XML structure, excluding indentation whitespace."""

    children = list(element)
    text = element.text or ""
    if children and not text.strip():
        text = ""
    return (
        _semantic_xml_tag(element.tag),
        tuple(sorted((str(name), value) for name, value in element.attrib.items())),
        text,
        tuple(
            (
                _semantic_xml_signature(child),
                "" if child.tail is None or not child.tail.strip() else child.tail,
            )
            for child in children
        ),
    )


def metadata_semantically_equal(left: bytes, right: bytes) -> bool:
    """Compare XML meaning while preserving byte-level mutation authority."""

    try:
        left_root = ET.fromstring(
            left,
            parser=ET.XMLParser(
                target=ET.TreeBuilder(insert_comments=True, insert_pis=True),
            ),
        )
        right_root = ET.fromstring(
            right,
            parser=ET.XMLParser(
                target=ET.TreeBuilder(insert_comments=True, insert_pis=True),
            ),
        )
    except ET.ParseError as exc:
        raise ReconcileError("canonical Quote approval metadata is not valid XML") from exc
    return _semantic_xml_signature(left_root) == _semantic_xml_signature(right_root)


def load_authority(source_root: Path, manifest_path: Path = DEFAULT_MANIFEST) -> ReconcileAuthority:
    manifest_path = manifest_path.expanduser().resolve(strict=True)
    manifest, manifest_raw = load_manifest(manifest_path)
    source_path = _resolve_source(source_root, str(manifest["source_path"]))
    seed, source_raw = _load_json_bytes(source_path, label="Quote wrapper reference")
    definitions = seed.get("processDefinitions")
    if not isinstance(definitions, list):
        raise ReconcileError("Quote wrapper reference has no processDefinitions list")

    by_name: dict[str, list[dict[str, Any]]] = {name: [] for name in TARGET_DEVELOPER_NAMES}
    for raw_definition in definitions:
        if not isinstance(raw_definition, dict):
            continue
        developer_name = str(raw_definition.get("developerName") or "")
        if developer_name in by_name:
            by_name[developer_name].append(raw_definition)

    manifest_targets = {str(item["developer_name"]): item for item in manifest["targets"]}
    authorities: list[TargetAuthority] = []
    for developer_name in TARGET_DEVELOPER_NAMES:
        matches = by_name[developer_name]
        if len(matches) != 1:
            raise ReconcileError(
                f"Quote wrapper reference must contain exactly one {developer_name} definition; got {len(matches)}"
            )
        encoded = matches[0].get("metadataBase64")
        if not isinstance(encoded, str) or not encoded:
            raise ReconcileError(f"Quote wrapper reference {developer_name} has no metadataBase64")
        try:
            source_after_metadata = base64.b64decode(encoded, validate=True)
        except ValueError as exc:
            raise ReconcileError(f"normalized Quote metadata is invalid for {developer_name}") from exc
        manifest_target = manifest_targets[developer_name]
        explicit_count = manifest_target["explicit_user_approver_count"]
        canonical_after_metadata = canonicalize_metadata(
            source_after_metadata,
            expected_user_approver_count=explicit_count,
            require_placeholder=True,
        )
        canonical_before_metadata = derive_before_metadata(canonical_after_metadata)
        try:
            ET.fromstring(canonical_before_metadata)
        except ET.ParseError as exc:
            raise ReconcileError(f"derived before metadata is invalid XML for {developer_name}") from exc

        canonical_before_sha256 = sha256_bytes(canonical_before_metadata)
        canonical_after_sha256 = sha256_bytes(canonical_after_metadata)
        if (
            canonical_before_sha256
            != manifest_target["canonical_before_metadata_sha256"]
        ):
            raise ReconcileError(f"derived before hash drift for {developer_name}")
        if (
            canonical_after_sha256
            != manifest_target["canonical_after_metadata_sha256"]
        ):
            raise ReconcileError(f"normalized after hash drift for {developer_name}")
        authorities.append(
            TargetAuthority(
                developer_name=developer_name,
                canonical_before_metadata=canonical_before_metadata,
                canonical_after_metadata=canonical_after_metadata,
                canonical_before_sha256=canonical_before_sha256,
                canonical_after_sha256=canonical_after_sha256,
                explicit_user_approver_count=explicit_count,
            )
        )

    return ReconcileAuthority(
        manifest_path=manifest_path,
        manifest_sha256=sha256_bytes(manifest_raw),
        source_path=source_path,
        source_sha256=sha256_bytes(source_raw),
        targets=tuple(authorities),
    )


def deployment_binding_from_args(args: argparse.Namespace) -> DeploymentBinding:
    digest_values = {
        "backup reset receipt": str(args.backup_reset_receipt_sha256),
        "backup reset identity": str(args.backup_reset_identity_sha256),
        "release fingerprint": str(args.release_fingerprint_sha256),
    }
    for label, digest in digest_values.items():
        if not SHA256_RE.fullmatch(digest):
            raise ReconcileError(f"{label} must be a lowercase SHA-256")
    attempt_id = str(args.backup_reset_attempt_id)
    if not RESET_ATTEMPT_RE.fullmatch(attempt_id):
        raise ReconcileError("backup reset attempt id must be 32 lowercase hexadecimal characters")
    refs = {
        "resolved downstream ref": str(args.resolved_downstream_ref),
        "resolved upstream ref": str(args.resolved_upstream_ref),
        "resolved official app-packs ref": str(args.resolved_official_app_packs_ref),
    }
    for label, ref in refs.items():
        if not GIT_SHA_RE.fullmatch(ref):
            raise ReconcileError(f"{label} must be a 40-character lowercase commit SHA")
    fingerprint_input = {
        "backup_reset_attempt_id": attempt_id,
        "backup_reset_identity_sha256": digest_values["backup reset identity"],
        "backup_reset_receipt_sha256": digest_values["backup reset receipt"],
        "resolved_downstream_ref": refs["resolved downstream ref"],
        "resolved_official_app_packs_ref": refs["resolved official app-packs ref"],
        "resolved_upstream_ref": refs["resolved upstream ref"],
    }
    actual_fingerprint = sha256_bytes(
        json.dumps(
            fingerprint_input,
            ensure_ascii=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    )
    if actual_fingerprint != digest_values["release fingerprint"]:
        raise ReconcileError("release fingerprint does not match the reset receipt and resolved refs")
    return DeploymentBinding(
        backup_reset_receipt_sha256=digest_values["backup reset receipt"],
        backup_reset_identity_sha256=digest_values["backup reset identity"],
        backup_reset_attempt_id=attempt_id,
        resolved_downstream_ref=refs["resolved downstream ref"],
        resolved_upstream_ref=refs["resolved upstream ref"],
        resolved_official_app_packs_ref=refs["resolved official app-packs ref"],
        release_fingerprint_sha256=actual_fingerprint,
    )


def _sql_string(value: str) -> str:
    return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'"


def _hex_literal(value: bytes) -> str:
    return "0x" + value.hex()


def render_live_read_sql(authority: ReconcileAuthority) -> str:
    """Render the payload-private first-session read."""

    names = ", ".join(_sql_string(target.developer_name) for target in authority.targets)
    return (
        "SET NAMES utf8mb4;\n"
        "SET @quote_wrapper_tidb_target = TIDB_VERSION();\n"
        "SELECT HEX(Id), HEX(DeveloperName), HEX(TableEnumOrId), "
        "HEX(OwnerPack), LOWER(SHA2(Metadata, 256)), HEX(Metadata)\n"
        "  FROM ProcessDefinition\n"
        f" WHERE DeveloperName IN ({names})\n"
        " ORDER BY BINARY DeveloperName, BINARY Id;\n"
    )


def _decode_hex_field(
    raw: str,
    *,
    label: str,
    allow_empty: bool,
    allow_null: bool = False,
) -> bytes | None:
    if raw == "NULL":
        if allow_null:
            return None
        raise ReconcileError(f"live read {label} is NULL")
    if len(raw) % 2 != 0 or not re.fullmatch(r"[0-9A-F]*", raw):
        raise ReconcileError(f"live read {label} is not exact uppercase HEX")
    value = bytes.fromhex(raw)
    if not allow_empty and not value:
        raise ReconcileError(f"live read {label} is empty")
    return value


def parse_live_read(
    authority: ReconcileAuthority,
    stdout: str,
) -> tuple[LiveTarget, ...]:
    """Validate exactly four physical rows and derive identity-preserving bytes."""

    lines = stdout.splitlines()
    if len(lines) != len(authority.targets) or any(not line for line in lines):
        raise ReconcileError("live read did not return exactly four physical rows")
    rows: dict[str, LiveTarget] = {}
    seen_ids: set[bytes] = set()
    authorities = {target.developer_name: target for target in authority.targets}
    for row_index, line in enumerate(lines):
        fields = line.split("\t")
        if len(fields) != 6:
            raise ReconcileError(
                f"live read row {row_index} does not contain the exact hash/HEX fields"
            )
        id_bytes = _decode_hex_field(
            fields[0],
            label=f"row {row_index} Id",
            allow_empty=False,
        )
        developer_name_bytes = _decode_hex_field(
            fields[1],
            label=f"row {row_index} DeveloperName",
            allow_empty=False,
        )
        table_enum_or_id_bytes = _decode_hex_field(
            fields[2],
            label=f"row {row_index} TableEnumOrId",
            allow_empty=False,
        )
        owner_pack_bytes = _decode_hex_field(
            fields[3],
            label=f"row {row_index} OwnerPack",
            allow_empty=True,
        )
        database_metadata_sha256 = fields[4]
        if not SHA256_RE.fullmatch(database_metadata_sha256):
            raise ReconcileError(f"live read row {row_index} Metadata hash is invalid")
        metadata = _decode_hex_field(
            fields[5],
            label=f"row {row_index} Metadata",
            allow_empty=False,
        )
        assert isinstance(id_bytes, bytes)
        assert isinstance(developer_name_bytes, bytes)
        assert isinstance(table_enum_or_id_bytes, bytes)
        assert isinstance(owner_pack_bytes, bytes)
        assert isinstance(metadata, bytes)
        if sha256_bytes(metadata) != database_metadata_sha256:
            raise ReconcileError(f"live read row {row_index} Metadata hash does not match its bytes")
        try:
            developer_name = developer_name_bytes.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise ReconcileError("live read DeveloperName is not valid UTF-8") from exc
        target_authority = authorities.get(developer_name)
        if (
            target_authority is None
            or developer_name_bytes != developer_name.encode("utf-8")
        ):
            raise ReconcileError("live read contains a non-exact target DeveloperName")
        if developer_name in rows:
            raise ReconcileError("live read contains a duplicate target DeveloperName")
        if id_bytes in seen_ids:
            raise ReconcileError("live read contains a duplicate physical Id")
        seen_ids.add(id_bytes)
        if table_enum_or_id_bytes != b"Quote":
            raise ReconcileError(f"live read {developer_name} is not an exact Quote row")
        if owner_pack_bytes != b"":
            raise ReconcileError(f"live read {developer_name} OwnerPack is not the exact empty value")

        canonical_metadata = canonicalize_metadata(
            metadata,
            expected_user_approver_count=target_authority.explicit_user_approver_count,
            require_placeholder=False,
        )
        wrapper_state = _reference_wrapper_state(metadata)
        matches_before = metadata_semantically_equal(
            canonical_metadata,
            target_authority.canonical_before_metadata,
        )
        matches_after = metadata_semantically_equal(
            canonical_metadata,
            target_authority.canonical_after_metadata,
        )
        if matches_before:
            if wrapper_state != "plural":
                raise ReconcileError(
                    f"live read {developer_name} semantic-before XML has the wrong wrapper state"
                )
            pre_state = "known_before"
            live_after_metadata = derive_after_metadata(metadata)
        elif matches_after:
            if wrapper_state != "singular":
                raise ReconcileError(
                    f"live read {developer_name} semantic-after XML has the wrong wrapper state"
                )
            pre_state = "already_target"
            live_after_metadata = metadata
        else:
            raise ReconcileError(
                f"live read {developer_name} is neither manifest-bound semantic before nor after"
            )

        desired_canonical = canonicalize_metadata(
            live_after_metadata,
            expected_user_approver_count=target_authority.explicit_user_approver_count,
            require_placeholder=False,
        )
        if not metadata_semantically_equal(
            desired_canonical,
            target_authority.canonical_after_metadata,
        ):
            raise ReconcileError(
                f"derived live desired XML does not match semantic after for {developer_name}"
            )
        rows[developer_name] = LiveTarget(
            id_bytes=id_bytes,
            developer_name=developer_name,
            developer_name_bytes=developer_name_bytes,
            table_enum_or_id_bytes=table_enum_or_id_bytes,
            owner_pack_bytes=owner_pack_bytes,
            live_before_metadata=metadata,
            live_after_metadata=live_after_metadata,
            live_before_sha256=sha256_bytes(metadata),
            live_after_sha256=sha256_bytes(live_after_metadata),
            explicit_user_approver_count=target_authority.explicit_user_approver_count,
            pre_state=pre_state,
        )

    if set(rows) != set(TARGET_DEVELOPER_NAMES):
        raise ReconcileError("live read target set does not match the manifest")
    return tuple(rows[name] for name in TARGET_DEVELOPER_NAMES)


def _snapshot_condition(index: int) -> str:
    return (
        f"(BINARY Id = BINARY @quote_wrapper_id_{index} "
        f"AND BINARY DeveloperName = BINARY @quote_wrapper_name_{index} "
        f"AND BINARY TableEnumOrId = BINARY @quote_wrapper_object_{index} "
        f"AND BINARY OwnerPack = BINARY @quote_wrapper_owner_{index} "
        f"AND BINARY Metadata = BINARY @quote_wrapper_live_before_{index})"
    )


def _final_condition(index: int) -> str:
    return (
        f"(BINARY Id = BINARY @quote_wrapper_id_{index} "
        f"AND BINARY DeveloperName = BINARY @quote_wrapper_name_{index} "
        f"AND BINARY TableEnumOrId = BINARY @quote_wrapper_object_{index} "
        f"AND BINARY OwnerPack = BINARY @quote_wrapper_owner_{index} "
        f"AND BINARY Metadata = BINARY @quote_wrapper_live_after_{index})"
    )


def render_reconcile_sql(
    authority: ReconcileAuthority,
    live_targets: Sequence[LiveTarget],
    *,
    execution_mode: str,
) -> str:
    """Render the second-session lock, CAS update, assertion, and outcome."""

    if execution_mode not in {"apply", "dry-run"}:
        raise ReconcileError(f"unsupported execution mode: {execution_mode!r}")
    if len(live_targets) != len(authority.targets):
        raise ReconcileError("live target count does not match reconcile authority")
    if tuple(target.developer_name for target in live_targets) != TARGET_DEVELOPER_NAMES:
        raise ReconcileError("live target order does not match reconcile authority")
    names = ", ".join(_sql_string(target.developer_name) for target in authority.targets)
    variable_lines: list[str] = []
    snapshot_conditions: list[str] = []
    update_conditions: list[str] = []
    update_cases: list[str] = []
    final_conditions: list[str] = []
    expected_updated_rows = 0
    for index, target in enumerate(live_targets):
        variable_lines.extend(
            [
                f"SET @quote_wrapper_id_{index} = {_hex_literal(target.id_bytes)};",
                f"SET @quote_wrapper_name_{index} = {_hex_literal(target.developer_name_bytes)};",
                f"SET @quote_wrapper_object_{index} = {_hex_literal(target.table_enum_or_id_bytes)};",
                f"SET @quote_wrapper_owner_{index} = '';",
                f"SET @quote_wrapper_live_before_{index} = {_hex_literal(target.live_before_metadata)};",
                f"SET @quote_wrapper_live_after_{index} = {_hex_literal(target.live_after_metadata)};",
            ]
        )
        snapshot_conditions.append(_snapshot_condition(index))
        final_conditions.append(_final_condition(index))
        if target.pre_state == "known_before":
            expected_updated_rows += 1
            update_conditions.append(_snapshot_condition(index))
            update_cases.append(
                f"WHEN {_snapshot_condition(index)} THEN @quote_wrapper_live_after_{index}"
            )
        elif target.pre_state != "already_target":
            raise ReconcileError(f"unsupported live pre-state: {target.pre_state!r}")

    snapshot_sql = " OR\n                    ".join(snapshot_conditions)
    update_sql = (
        " OR\n       ".join(update_conditions)
        if update_conditions
        else "0 = 1"
    )
    cases_sql = (
        "\n        ".join(update_cases)
        if update_cases
        else "WHEN 0 = 1 THEN Metadata"
    )
    final_sql = " OR\n                    ".join(final_conditions)
    transaction_end = "COMMIT;" if execution_mode == "apply" else "ROLLBACK;"
    return (
        "SET NAMES utf8mb4;\n"
        "SET @quote_wrapper_tidb_target = TIDB_VERSION();\n"
        f"DROP TEMPORARY TABLE IF EXISTS `{ASSERTION_TABLE}`;\n"
        f"CREATE TEMPORARY TABLE `{ASSERTION_TABLE}` "
        "(`id` TINYINT NOT NULL PRIMARY KEY) ENGINE=MEMORY;\n"
        f"INSERT INTO `{ASSERTION_TABLE}` (`id`) VALUES (1);\n"
        + "\n".join(variable_lines)
        + "\nSTART TRANSACTION;\n"
        "SELECT HEX(Id)\n"
        "  FROM ProcessDefinition\n"
        f" WHERE DeveloperName IN ({names})\n"
        " ORDER BY BINARY DeveloperName, BINARY Id\n"
        " FOR UPDATE;\n"
        "SET @quote_wrapper_guard_ok = (\n"
        "    SELECT IF(\n"
        f"        COUNT(*) = {len(live_targets)}\n"
        "        AND SUM(CASE WHEN\n"
        f"                    {snapshot_sql}\n"
        f"                THEN 1 ELSE 0 END) = {len(live_targets)},\n"
        "        1, 0\n"
        "    )\n"
        "      FROM ProcessDefinition\n"
        f"     WHERE DeveloperName IN ({names})\n"
        ");\n"
        "UPDATE ProcessDefinition\n"
        "   SET Metadata = CASE\n"
        f"        {cases_sql}\n"
        "        ELSE Metadata\n"
        "       END,\n"
        "       LastModifiedDate = CURRENT_TIMESTAMP\n"
        " WHERE @quote_wrapper_guard_ok = 1\n"
        "   AND (\n"
        f"       {update_sql}\n"
        "   );\n"
        "SET @quote_wrapper_updated_rows = ROW_COUNT();\n"
        "SET @quote_wrapper_final_ok = (\n"
        "    SELECT IF(\n"
        f"        COUNT(*) = {len(live_targets)}\n"
        "        AND SUM(CASE WHEN\n"
        f"                    {final_sql}\n"
        f"                THEN 1 ELSE 0 END) = {len(live_targets)},\n"
        "        1, 0\n"
        "    )\n"
        "      FROM ProcessDefinition\n"
        f"     WHERE DeveloperName IN ({names})\n"
        ");\n"
        f"INSERT INTO `{ASSERTION_TABLE}` (`id`)\n"
        "SELECT IF(\n"
        "    @quote_wrapper_guard_ok = 1\n"
        "    AND @quote_wrapper_final_ok = 1\n"
        f"    AND @quote_wrapper_updated_rows = {expected_updated_rows},\n"
        "    2, 1\n"
        ");\n"
        "SELECT CONCAT(\n"
        f"    '{SENTINEL}',\n"
        "    JSON_OBJECT(\n"
        "        'guard_ok', @quote_wrapper_guard_ok,\n"
        "        'final_ok', @quote_wrapper_final_ok,\n"
        "        'updated_rows', @quote_wrapper_updated_rows,\n"
        f"        'expected_updated_rows', {expected_updated_rows},\n"
        f"        'target_count', {len(live_targets)}\n"
        "    )\n"
        ");\n"
        f"{transaction_end}\n"
        f"DROP TEMPORARY TABLE `{ASSERTION_TABLE}`;\n"
    )


TIDB_TLS_TRUE_TOKENS = frozenset({"1", "true", "yes", "on"})


def _tidb_tls_enabled(raw: str) -> bool:
    return str(raw).strip().lower() in TIDB_TLS_TRUE_TOKENS


def _mysql_client_args(tidb_env: dict[str, str]) -> list[str]:
    command = shutil.which("mysql") or shutil.which("mariadb")
    if not command:
        raise ReconcileError("missing required TiDB SQL client command: mysql or mariadb")
    required_env = (
        "TIDB_HOST",
        "TIDB_PORT",
        "TIDB_USER",
        "TIDB_PASSWORD",
        "TIDB_DATABASE",
        "TIDB_TLS",
    )
    missing = [key for key in required_env if not str(tidb_env.get(key, "")).strip()]
    if missing:
        raise ReconcileError(f"missing required TiDB environment values: {', '.join(missing)}")
    try:
        port = int(str(tidb_env["TIDB_PORT"]))
    except ValueError as exc:
        raise ReconcileError("TIDB_PORT must be an integer") from exc
    if not 1 <= port <= 65535:
        raise ReconcileError("TIDB_PORT must be between 1 and 65535")
    if not _tidb_tls_enabled(tidb_env["TIDB_TLS"]):
        raise ReconcileError(
            "TiDB approval reconcile requires TIDB_TLS to be one of: 1, true, yes, on"
        )

    probe_env = {"PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin")}
    version_result = subprocess.run(
        [command, "--no-defaults", "--version"],
        text=True,
        capture_output=True,
        check=False,
        env=probe_env,
    )
    version_text = f"{version_result.stdout}\n{version_result.stderr}"
    if version_result.returncode != 0:
        raise ReconcileError("TiDB SQL client version probe failed")
    if re.search(r"\bmariadb\b", version_text, re.IGNORECASE):
        client_family = "mariadb"
    elif re.search(r"\bmysql\b", version_text, re.IGNORECASE):
        client_family = "mysql"
    else:
        raise ReconcileError("TiDB approval reconcile could not verify SQL client identity")
    # Option files may set `force`, reconnect, or other session-changing client
    # behavior. The duplicate-key assertion is our pre-COMMIT abort gate, so no
    # ambient client configuration is allowed to weaken it. MariaDB has no
    # login-path mechanism, so --no-defaults is its complete ambient-config
    # exclusion while Oracle MySQL also needs --no-login-paths.
    client_args = [command, "--no-defaults"]
    if client_family == "mariadb":
        client_args.append("--ssl")
    else:
        client_args.extend(["--no-login-paths", "--ssl-mode", "REQUIRED"])
    client_args.extend(
        [
            "--default-character-set=utf8mb4",
            "--skip-reconnect",
            "-N",
            "-B",
        ]
    )
    # --version and --help may exit successfully without rejecting unknown
    # options. A deliberate final unknown option proves that every preceding
    # safety option was parsed and accepted without opening a connection.
    option_probe_stop = "--nexuscrm-client-option-probe-stop"
    capability_result = subprocess.run(
        [*client_args, option_probe_stop],
        text=True,
        capture_output=True,
        check=False,
        env=probe_env,
    )
    capability_text = f"{capability_result.stdout}\n{capability_result.stderr}"
    capability_diagnostics = [
        line.strip() for line in capability_text.splitlines() if line.strip()
    ]
    if (
        capability_result.returncode == 0
        or len(capability_diagnostics) != 1
        or option_probe_stop not in capability_diagnostics[0]
    ):
        raise ReconcileError("TiDB SQL transport client lacks required safety options")
    client_args.extend(
        [
            f"-h{tidb_env['TIDB_HOST']}",
            f"-P{port}",
            f"-u{tidb_env['TIDB_USER']}",
            tidb_env["TIDB_DATABASE"],
        ]
    )
    return client_args


def _parse_sentinel(stdout: str, *, expected_updated_rows: int) -> dict[str, int]:
    matches = [line[len(SENTINEL) :] for line in stdout.splitlines() if line.startswith(SENTINEL)]
    if len(matches) != 1:
        raise ReconcileError("database reconcile did not emit exactly one verification result")
    try:
        raw = json.loads(matches[0])
    except json.JSONDecodeError as exc:
        raise ReconcileError("database reconcile emitted invalid verification JSON") from exc
    expected_keys = {
        "guard_ok",
        "final_ok",
        "updated_rows",
        "expected_updated_rows",
        "target_count",
    }
    if not isinstance(raw, dict) or set(raw) != expected_keys:
        raise ReconcileError("database reconcile verification result has an invalid shape")
    numeric_keys = (
        "guard_ok",
        "final_ok",
        "updated_rows",
        "expected_updated_rows",
        "target_count",
    )
    if any(type(raw[key]) is not int for key in numeric_keys):
        raise ReconcileError("database reconcile verification values are not exact integers")
    parsed = {key: raw[key] for key in numeric_keys}
    if parsed["guard_ok"] != 1 or parsed["final_ok"] != 1:
        raise ReconcileError("database reconcile verification did not pass")
    if parsed["target_count"] != len(TARGET_DEVELOPER_NAMES):
        raise ReconcileError("database reconcile verification target count drifted")
    if (
        parsed["expected_updated_rows"] != expected_updated_rows
        or parsed["updated_rows"] != expected_updated_rows
    ):
        raise ReconcileError("database reconcile updated-row count does not match the live snapshot")
    return parsed


def _client_failure_detail(stderr: object, *, password: str) -> str:
    detail_lines = [line.strip() for line in str(stderr or "").splitlines() if line.strip()]
    detail = detail_lines[-1][:300] if detail_lines else "mysql client returned no diagnostic"
    if password:
        detail = detail.replace(password, "[REDACTED]")
    detail = re.sub(r"(?i)0x[0-9a-f]+", "[REDACTED_HEX]", detail)
    detail = re.sub(
        r"(?i)(?<![0-9a-f])[0-9a-f]{32,}(?![0-9a-f])",
        "[REDACTED_DIGEST]",
        detail,
    )
    return detail


def execute_reconcile(
    authority: ReconcileAuthority,
    *,
    execution_mode: str,
    client_args: Sequence[str],
    password: str,
    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> ReconcileResult:
    if execution_mode not in {"apply", "dry-run"}:
        raise ReconcileError(f"unsupported execution mode: {execution_mode!r}")
    client_env = {
        "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
    }
    if password:
        client_env["MYSQL_PWD"] = password
    else:
        client_env.pop("MYSQL_PWD", None)
    read_result = runner(
        list(client_args),
        input=render_live_read_sql(authority),
        text=True,
        capture_output=True,
        check=False,
        env=client_env,
    )
    if read_result.returncode != 0:
        detail = _client_failure_detail(read_result.stderr, password=password)
        raise ReconcileError(f"database reconcile live read failed: {detail}")
    live_targets = parse_live_read(authority, str(read_result.stdout or ""))
    expected_updated_rows = sum(
        target.pre_state == "known_before" for target in live_targets
    )

    mutation_result = runner(
        list(client_args),
        input=render_reconcile_sql(
            authority,
            live_targets,
            execution_mode=execution_mode,
        ),
        text=True,
        capture_output=True,
        check=False,
        env=client_env,
    )
    if mutation_result.returncode != 0:
        detail = _client_failure_detail(mutation_result.stderr, password=password)
        raise ReconcileError(f"database reconcile failed before commit: {detail}")
    parsed_result = _parse_sentinel(
        str(mutation_result.stdout or ""),
        expected_updated_rows=expected_updated_rows,
    )
    return ReconcileResult(
        targets=live_targets,
        updated_rows=parsed_result["updated_rows"],
    )


def build_evidence(
    authority: ReconcileAuthority,
    *,
    execution_mode: str,
    result: ReconcileResult,
    deployment_binding: DeploymentBinding,
) -> dict[str, Any]:
    updated_rows = result.updated_rows
    authorities = {target.developer_name: target for target in authority.targets}
    return {
        "format": EVIDENCE_FORMAT,
        "status": "passed",
        "execution_mode": execution_mode,
        "transaction_outcome": "committed" if execution_mode == "apply" else "rolled_back",
        "manifest_sha256": authority.manifest_sha256,
        "source_sha256": authority.source_sha256,
        "table_enum_or_id": "Quote",
        "owner_pack_state": "empty_string",
        "target_count": len(authority.targets),
        "reference_wrappers_per_target": len(EXPECTED_REFERENCES),
        "updated_rows": updated_rows,
        "already_converged_rows": len(authority.targets) - updated_rows,
        "deployment_binding": {
            "backup_reset_receipt_sha256": deployment_binding.backup_reset_receipt_sha256,
            "backup_reset_identity_sha256": deployment_binding.backup_reset_identity_sha256,
            "backup_reset_attempt_id": deployment_binding.backup_reset_attempt_id,
            "resolved_downstream_ref": deployment_binding.resolved_downstream_ref,
            "resolved_upstream_ref": deployment_binding.resolved_upstream_ref,
            "resolved_official_app_packs_ref": deployment_binding.resolved_official_app_packs_ref,
            "release_fingerprint_sha256": deployment_binding.release_fingerprint_sha256,
        },
        "targets": [
            {
                "developer_name": target.developer_name,
                "canonical_before_metadata_sha256": authorities[
                    target.developer_name
                ].canonical_before_sha256,
                "canonical_after_metadata_sha256": authorities[
                    target.developer_name
                ].canonical_after_sha256,
                "live_before_metadata_sha256": target.live_before_sha256,
                "live_after_metadata_sha256": target.live_after_sha256,
                "explicit_user_approver_count": target.explicit_user_approver_count,
                "pre_state": target.pre_state,
                "outcome": (
                    "already_target"
                    if target.pre_state == "already_target"
                    else "updated" if execution_mode == "apply" else "would_update"
                ),
            }
            for target in result.targets
        ],
    }


def write_evidence(path: Path, evidence: dict[str, Any]) -> None:
    path = path.expanduser()
    if path.is_symlink():
        raise ReconcileError(f"evidence path is a symlink: {path}")
    path = path.resolve()
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists() and (path.is_symlink() or not path.is_file()):
        raise ReconcileError(f"evidence path is not a regular file: {path}")
    rendered = json.dumps(evidence, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=path.parent,
            prefix=f".{path.name}.",
            delete=False,
        ) as handle:
            handle.write(rendered)
            handle.flush()
            os.fsync(handle.fileno())
            temp_path = Path(handle.name)
        os.replace(temp_path, path)
    except OSError as exc:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)
        raise ReconcileError(f"unable to write reconcile evidence: {path}") from exc


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source-root", required=True, type=Path)
    parser.add_argument("--manifest-file", type=Path, default=DEFAULT_MANIFEST)
    parser.add_argument("--evidence-file", required=True, type=Path)
    parser.add_argument("--execution-mode", required=True, choices=("dry-run", "apply"))
    parser.add_argument("--backup-reset-receipt-sha256", required=True)
    parser.add_argument("--backup-reset-identity-sha256", required=True)
    parser.add_argument("--backup-reset-attempt-id", required=True)
    parser.add_argument("--resolved-downstream-ref", required=True)
    parser.add_argument("--resolved-upstream-ref", required=True)
    parser.add_argument("--resolved-official-app-packs-ref", required=True)
    parser.add_argument("--release-fingerprint-sha256", required=True)
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        tidb_env = {
            key: os.environ.get(key, "")
            for key in (
                "TIDB_HOST",
                "TIDB_PORT",
                "TIDB_USER",
                "TIDB_PASSWORD",
                "TIDB_DATABASE",
                "TIDB_TLS",
            )
        }
        authority = load_authority(args.source_root, args.manifest_file)
        deployment_binding = deployment_binding_from_args(args)
        result = execute_reconcile(
            authority,
            execution_mode=args.execution_mode,
            client_args=_mysql_client_args(tidb_env),
            password=tidb_env["TIDB_PASSWORD"],
        )
        evidence = build_evidence(
            authority,
            execution_mode=args.execution_mode,
            result=result,
            deployment_binding=deployment_binding,
        )
        write_evidence(args.evidence_file, evidence)
        print(json.dumps(evidence, ensure_ascii=False, sort_keys=True))
        return 0
    except ReconcileError as exc:
        print(f"Quote approval action-wrapper reconcile failed: {exc}", file=sys.stderr)
        return 2


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