"""ARR-side artifact verification and independent deterministic replay."""

from __future__ import annotations

import copy
import hashlib
import json
import re
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple

from arr_ingestion.artifacts import ArtifactStore
from arr_ingestion.contracts import (
    ARTIFACT_SIZE_LIMITS,
    ArtifactRef,
    DeliveryEnvelope,
    IngestionError,
    LEGACY_RESULT_SCHEMA_VERSION,
    RESULT_SCHEMA_VERSION,
    ROLE_CONTRACTS,
    SHA256_RE,
)
STRUCTURED_TOP_FIELDS = {
    "result_schema_version",
    "status",
    "activation_eligible",
    "ingestion_mode",
    "business_date",
    "processor_version",
    "rule_set_sha256",
    "source_rows",
    "removed_by_rate_code",
    "removed_as_duplicates",
    "output_rows",
    "outcome_counts",
    "channels",
    "artifacts",
    "records",
    "errors",
}
STRUCTURED_ARTIFACT_FIELDS = {
    "source_xml",
    "daily_report",
    "result_json",
    "exception_report",
}
INNER_ARTIFACT_FIELDS = {
    "file_kind",
    "original_filename",
    "sha256",
    "byte_size",
    "mime_type",
}
OUTCOME_FIELDS = {
    "duplicate",
    "excluded_rate_code",
    "price_unmatched",
    "retained",
    "validation_failed",
}
RECORD_FIELDS = {
    "source_sequence",
    "source_location",
    "source_worksheet",
    "source_row_no",
    "outcome",
    "decision_codes",
    "duplicate_of_source_sequence",
    "adults",
    "children",
    "block_code",
    "no_of_rooms",
    "company_name",
    "company_key",
    "confirmation_no",
    "disp_room_no",
    "effective_rate_amount",
    "full_name",
    "res_comment",
    "group_code_key",
    "booking_source_match_status",
    "trace_text",
    "products",
    "rate_code",
    "normalized_rate_code",
    "room_category_label",
    "arrival",
    "departure",
    "nights",
    "real_price",
    "total_price",
    "kb_amount",
    "channel_key",
    "pricing_method",
}
ERROR_FIELDS = {
    "code",
    "stage",
    "source_location",
    "company_name",
    "rate_code",
    "effective_rate_amount",
    "confirmation_no",
    "message",
}
RESULT_FIELDS = {
    "version",
    "status",
    "business_date",
    "message",
    "metrics",
    "outputs",
    "errors",
}


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _reject_constant(_value: str) -> None:
    raise ValueError("non-finite JSON number")


def _unique_object(pairs: Sequence[Tuple[str, Any]]) -> Dict[str, Any]:
    result: Dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON key")
        result[key] = value
    return result


def strict_json_bytes(raw: bytes, label: str) -> Dict[str, Any]:
    try:
        value = json.loads(
            raw.decode("utf-8"),
            object_pairs_hook=_unique_object,
            parse_constant=_reject_constant,
        )
    except (UnicodeDecodeError, ValueError, json.JSONDecodeError):
        raise IngestionError("DELIVERY_JSON_INVALID", f"{label} JSON is invalid") from None
    if not isinstance(value, dict):
        raise IngestionError("DELIVERY_JSON_INVALID", f"{label} JSON must be an object")
    return value


def strict_json_file(path: Path, label: str) -> Dict[str, Any]:
    try:
        return strict_json_bytes(path.read_bytes(), label)
    except OSError:
        raise IngestionError("ARTIFACT_UNREADABLE", f"{label} artifact is unreadable") from None


@dataclass(frozen=True)
class ProcessorPolicy:
    processor_version: str
    rule_set_sha256: str
    skill_root: Path
    python_binary: str = sys.executable
    # The retired direct-MCP/callback route keeps accepting only the frozen v3
    # success contract.  Its replay is deliberately opt-in so the active
    # artifact-callback processor can never silently emit a v3 result.
    legacy_direct_v3: bool = False

    def __post_init__(self) -> None:
        if (
            not self.processor_version
            or not SHA256_RE.fullmatch(self.rule_set_sha256)
            or not isinstance(self.legacy_direct_v3, bool)
        ):
            raise ValueError("processor policy identity is invalid")
        resolved = self.skill_root.resolve()
        object.__setattr__(self, "skill_root", resolved)
        if not (resolved / "scripts" / "process_daily.py").is_file():
            raise ValueError("processor executable is unavailable")
        if not (resolved / "scripts" / "validate_daily.py").is_file():
            raise ValueError("processor validator is unavailable")
        if not (resolved / "references" / "价格对照.xlsx").is_file():
            raise ValueError("processor price reference is unavailable")


@dataclass(frozen=True)
class VerifiedDelivery:
    envelope: DeliveryEnvelope
    envelope_sha256: str
    result_payload: Mapping[str, Any]
    structured_payload: Mapping[str, Any]


def _require_exact_mapping(value: Any, fields: set[str], label: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping) or set(value) != fields:
        raise IngestionError("RESULT_CONTRACT_INVALID", f"{label} contract is invalid")
    return value


def _require_nonnegative_integer(value: Any, label: str) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value < 0:
        raise IngestionError("RESULT_CONTRACT_INVALID", f"{label} must be a non-negative integer")
    return value


def _inner_artifact(
    role: str,
    value: Any,
    reference: Optional[ArtifactRef],
) -> Optional[Mapping[str, Any]]:
    if value is None:
        if reference is not None:
            raise IngestionError(
                "RESULT_ARTIFACT_MISMATCH", f"{role} processor artifact is missing"
            )
        return None
    fields = _require_exact_mapping(value, INNER_ARTIFACT_FIELDS, f"{role} artifact")
    if reference is None:
        raise IngestionError(
            "RESULT_ARTIFACT_MISMATCH", f"{role} processor artifact was not delivered"
        )
    expected_kind, _extension, expected_mime = ROLE_CONTRACTS[role]
    if (
        fields.get("file_kind") != expected_kind
        or fields.get("original_filename") != reference.original_filename
        or fields.get("sha256") != reference.sha256
        or fields.get("byte_size") != reference.byte_size
        or fields.get("mime_type") != expected_mime
    ):
        raise IngestionError(
            "RESULT_ARTIFACT_MISMATCH", f"{role} processor artifact metadata does not match delivery"
        )
    return fields


def _validate_result_payload_v3(
    payload: Mapping[str, Any],
    envelope: DeliveryEnvelope,
    structured: Mapping[str, Any],
) -> None:
    _require_exact_mapping(payload, RESULT_FIELDS, "result")
    expected_date = envelope.business_date.isoformat() if envelope.business_date else None
    if (
        payload.get("version") != LEGACY_RESULT_SCHEMA_VERSION
        or payload.get("status") != envelope.status
        or payload.get("business_date") != expected_date
        or not isinstance(payload.get("message"), str)
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result identity is invalid")
    metrics = _require_exact_mapping(
        payload.get("metrics"),
        {
            "source_rows",
            "removed_by_rate_code",
            "removed_as_duplicates",
            "output_rows",
            "channels",
        },
        "result metrics",
    )
    for field in (
        "source_rows",
        "removed_by_rate_code",
        "removed_as_duplicates",
        "output_rows",
    ):
        _require_nonnegative_integer(metrics.get(field), f"metrics.{field}")
        if metrics.get(field) != structured.get(field):
            raise IngestionError("RESULT_CONTRACT_INVALID", "result counters do not reconcile")
    if metrics.get("channels") != structured.get("channels"):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result channels do not reconcile")
    outputs = _require_exact_mapping(
        payload.get("outputs"),
        {"daily_report", "structured_result", "exception_report"},
        "result outputs",
    )
    daily_ref = envelope.artifacts["daily_report"]
    structured_ref = envelope.artifacts["structured_result_json"]
    exception_ref = envelope.artifacts["exception_report"]
    if (
        outputs.get("daily_report")
        != (daily_ref.original_filename if daily_ref is not None else None)
        or outputs.get("structured_result")
        != (
            structured_ref.original_filename
            if structured_ref is not None
            else None
        )
        or outputs.get("exception_report")
        != (exception_ref.original_filename if exception_ref is not None else None)
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result output names are invalid")
    errors = payload.get("errors")
    if not isinstance(errors, list) or errors != structured.get("errors"):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result errors do not reconcile")


def _validate_structured_payload_v3(
    payload: Mapping[str, Any], envelope: DeliveryEnvelope
) -> None:
    _require_exact_mapping(payload, STRUCTURED_TOP_FIELDS, "structured result")
    expected_date = envelope.business_date.isoformat() if envelope.business_date else None
    if (
        payload.get("result_schema_version") != LEGACY_RESULT_SCHEMA_VERSION
        or payload.get("status") != envelope.status
        or payload.get("activation_eligible") != (envelope.status == "success")
        or payload.get("ingestion_mode") != "opera_xml"
        or payload.get("business_date") != expected_date
        or payload.get("processor_version") != envelope.processor_version
        or payload.get("rule_set_sha256") != envelope.rule_set_sha256
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured result identity is invalid")
    counts: Dict[str, int] = {}
    for field in (
        "source_rows",
        "removed_by_rate_code",
        "removed_as_duplicates",
        "output_rows",
    ):
        counts[field] = _require_nonnegative_integer(payload.get(field), field)
    outcomes = _require_exact_mapping(payload.get("outcome_counts"), OUTCOME_FIELDS, "outcome counts")
    for outcome in OUTCOME_FIELDS:
        _require_nonnegative_integer(outcomes.get(outcome), f"outcome_counts.{outcome}")
    records = payload.get("records")
    if not isinstance(records, list) or len(records) != counts["source_rows"]:
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured record count is invalid")
    actual_outcomes = {outcome: 0 for outcome in OUTCOME_FIELDS}
    for index, record in enumerate(records, 1):
        values = _require_exact_mapping(record, RECORD_FIELDS, "structured record")
        if values.get("source_sequence") != index or values.get("source_location") != f"reservation[{index}]":
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured source order is invalid")
        if values.get("source_worksheet") is not None or values.get("source_row_no") is not None:
            raise IngestionError(
                "RESULT_CONTRACT_INVALID", "direct XML source coordinates are invalid"
            )
        outcome = values.get("outcome")
        if outcome not in OUTCOME_FIELDS:
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured outcome is invalid")
        actual_outcomes[str(outcome)] += 1
        duplicate_of = values.get("duplicate_of_source_sequence")
        if (
            outcome == "duplicate"
            and (
                not isinstance(duplicate_of, int)
                or isinstance(duplicate_of, bool)
                or duplicate_of < 1
                or duplicate_of >= index
            )
        ) or (outcome != "duplicate" and duplicate_of is not None):
            raise IngestionError(
                "RESULT_CONTRACT_INVALID", "structured duplicate lineage is invalid"
            )
        decisions = values.get("decision_codes")
        if (
            not isinstance(decisions, list)
            or not decisions
            or any(not isinstance(code, str) or not code for code in decisions)
            or len(decisions) != len(set(decisions))
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured decisions are invalid")
        group_code = values.get("group_code_key")
        booking_status = values.get("booking_source_match_status")
        if group_code is None:
            if booking_status != "missing_group_code":
                raise IngestionError(
                    "RESULT_CONTRACT_INVALID", "booking source status is invalid"
                )
        elif (
            not isinstance(group_code, str)
            or not group_code
            or group_code != group_code.strip().upper()
            or booking_status != "not_checked"
        ):
            raise IngestionError(
                "RESULT_CONTRACT_INVALID", "structured Group Code is invalid"
            )
    if dict(outcomes) != actual_outcomes:
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured outcomes do not reconcile")
    if (
        counts["removed_by_rate_code"] != outcomes.get("excluded_rate_code")
        or counts["removed_as_duplicates"] != outcomes.get("duplicate")
        or counts["output_rows"] != outcomes.get("retained")
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured counters do not reconcile")
    channels = payload.get("channels")
    errors = payload.get("errors")
    if not isinstance(channels, list) or not isinstance(errors, list):
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured arrays are invalid")
    channel_names: set[str] = set()
    channel_rows = 0
    for channel in channels:
        values = _require_exact_mapping(
            channel, {"worksheet", "rows"}, "structured channel"
        )
        worksheet = values.get("worksheet")
        rows = _require_nonnegative_integer(
            values.get("rows"), "structured channel rows"
        )
        if (
            not isinstance(worksheet, str)
            or not worksheet.strip()
            or worksheet in channel_names
        ):
            raise IngestionError(
                "RESULT_CONTRACT_INVALID", "structured channel is invalid"
            )
        channel_names.add(worksheet)
        channel_rows += rows
    for error in errors:
        _require_exact_mapping(error, ERROR_FIELDS, "structured error")
    if envelope.status == "success":
        if (
            errors
            or outcomes.get("validation_failed") != 0
            or outcomes.get("price_unmatched") != 0
            or counts["source_rows"]
            != counts["removed_by_rate_code"]
            + counts["removed_as_duplicates"]
            + counts["output_rows"]
            or channel_rows != counts["output_rows"]
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "successful result is not balanced")
    elif counts["output_rows"] != 0 or channels or not errors:
        raise IngestionError("RESULT_CONTRACT_INVALID", "failed result shape is invalid")

    artifacts = _require_exact_mapping(
        payload.get("artifacts"), STRUCTURED_ARTIFACT_FIELDS, "structured artifacts"
    )
    _inner_artifact("source_xml", artifacts.get("source_xml"), envelope.artifacts["source_xml"])
    _inner_artifact("daily_report", artifacts.get("daily_report"), envelope.artifacts["daily_report"])
    _inner_artifact("result_json", artifacts.get("result_json"), envelope.artifacts["result_json"])
    _inner_artifact(
        "exception_report",
        artifacts.get("exception_report"),
        envelope.artifacts["exception_report"],
    )


V4_STRUCTURED_TOP_FIELDS = {
    "result_schema_version",
    "status",
    "activation_eligible",
    "ingestion_mode",
    "business_date",
    "processor_version",
    "rule_set_sha256",
    "source_rows",
    "removed_by_rate_code",
    "removed_as_duplicates",
    "output_rows",
    "outcome_counts",
    "candidate_rows",
    "review_required_rows",
    "review_issue_count",
    "review_issues",
    "review_case_id",
    "manual_override_sha256",
    "manually_priced_rows",
    "channels",
    "artifacts",
    "records",
    "errors",
}
V4_STRUCTURED_ARTIFACT_FIELDS = {
    "source_xml",
    "daily_report",
    "result_json",
    "exception_report",
    "manual_override_json",
}
V4_OUTCOME_FIELDS = {
    "candidate",
    "duplicate",
    "excluded_rate_code",
    "price_unmatched",
    "retained",
    "validation_failed",
}
V4_RESULT_METRIC_FIELDS = {
    "source_rows",
    "removed_by_rate_code",
    "removed_as_duplicates",
    "output_rows",
    "candidate_rows",
    "review_required_rows",
    "review_issue_count",
    "channels",
}
REVIEW_ISSUE_FIELDS = {
    "company_key",
    "rate_code",
    "effective_rate_amount",
    "candidate_prices",
    "affected_records",
    "affected_rooms",
    "affected_room_nights",
}
REVIEW_CANDIDATE_FIELDS = {"effective_rate_amount", "real_price"}


def _require_nonnegative_number(value: Any, label: str) -> None:
    if (
        not isinstance(value, (int, float))
        or isinstance(value, bool)
        or value < 0
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", f"{label} must be a non-negative number")


def _validate_v4_records(payload: Mapping[str, Any]) -> Tuple[Dict[str, int], int]:
    source_rows = _require_nonnegative_integer(payload.get("source_rows"), "source_rows")
    records = payload.get("records")
    if not isinstance(records, list) or len(records) != source_rows:
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured record count is invalid")
    actual_outcomes = {outcome: 0 for outcome in V4_OUTCOME_FIELDS}
    manual_rows = 0
    for index, record in enumerate(records, 1):
        values = _require_exact_mapping(record, RECORD_FIELDS, "structured record")
        if values.get("source_sequence") != index or values.get("source_location") != f"reservation[{index}]":
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured source order is invalid")
        if values.get("source_worksheet") is not None or values.get("source_row_no") is not None:
            raise IngestionError("RESULT_CONTRACT_INVALID", "direct XML source coordinates are invalid")
        outcome = values.get("outcome")
        if outcome not in V4_OUTCOME_FIELDS:
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured outcome is invalid")
        actual_outcomes[str(outcome)] += 1
        duplicate_of = values.get("duplicate_of_source_sequence")
        if (
            outcome == "duplicate"
            and (
                not isinstance(duplicate_of, int)
                or isinstance(duplicate_of, bool)
                or duplicate_of < 1
                or duplicate_of >= index
            )
        ) or (outcome != "duplicate" and duplicate_of is not None):
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured duplicate lineage is invalid")
        decisions = values.get("decision_codes")
        if (
            not isinstance(decisions, list)
            or not decisions
            or any(not isinstance(code, str) or not code for code in decisions)
            or len(decisions) != len(set(decisions))
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured decisions are invalid")
        group_code = values.get("group_code_key")
        booking_status = values.get("booking_source_match_status")
        if group_code is None:
            if booking_status != "missing_group_code":
                raise IngestionError("RESULT_CONTRACT_INVALID", "booking source status is invalid")
        elif (
            not isinstance(group_code, str)
            or not group_code
            or group_code != group_code.strip().upper()
            or booking_status != "not_checked"
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured Group Code is invalid")
        if values.get("pricing_method") == "manual_review":
            if outcome not in {"retained", "candidate"} or "MANUAL_PRICE_APPLIED" not in decisions:
                raise IngestionError("RESULT_CONTRACT_INVALID", "manual pricing record is invalid")
            manual_rows += 1
    return actual_outcomes, manual_rows


def _validate_review_issues(value: Any) -> List[Tuple[str, str, float]]:
    if not isinstance(value, list):
        raise IngestionError("RESULT_CONTRACT_INVALID", "review issues are invalid")
    keys: List[Tuple[str, str, float]] = []
    for raw in value:
        issue = _require_exact_mapping(raw, REVIEW_ISSUE_FIELDS, "review issue")
        company = issue.get("company_key")
        rate_code = issue.get("rate_code")
        amount = issue.get("effective_rate_amount")
        if (
            not isinstance(company, str)
            or not company
            or not isinstance(rate_code, str)
            or not rate_code
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "review issue key is invalid")
        _require_nonnegative_number(amount, "review issue Opera amount")
        candidates = issue.get("candidate_prices")
        if not isinstance(candidates, list):
            raise IngestionError("RESULT_CONTRACT_INVALID", "review issue candidates are invalid")
        for raw_candidate in candidates:
            candidate = _require_exact_mapping(
                raw_candidate, REVIEW_CANDIDATE_FIELDS, "review price candidate"
            )
            _require_nonnegative_number(
                candidate.get("effective_rate_amount"), "review candidate Opera amount"
            )
            _require_nonnegative_number(candidate.get("real_price"), "review candidate price")
        for field in ("affected_records", "affected_rooms", "affected_room_nights"):
            _require_nonnegative_integer(issue.get(field), f"review issue {field}")
        if issue["affected_records"] < 1 or issue["affected_rooms"] < 1:
            raise IngestionError("RESULT_CONTRACT_INVALID", "review issue impact is invalid")
        keys.append((company, rate_code, float(amount)))
    if keys != sorted(keys) or len(keys) != len(set(keys)):
        raise IngestionError("RESULT_CONTRACT_INVALID", "review issues must be unique and canonical")
    return keys


def _validate_structured_payload_v4(
    payload: Mapping[str, Any], envelope: DeliveryEnvelope
) -> None:
    _require_exact_mapping(payload, V4_STRUCTURED_TOP_FIELDS, "structured result")
    expected_date = envelope.business_date.isoformat() if envelope.business_date else None
    if (
        payload.get("result_schema_version") != RESULT_SCHEMA_VERSION
        or payload.get("status") != envelope.status
        or payload.get("activation_eligible") != (envelope.status == "success")
        or payload.get("ingestion_mode") != "opera_xml"
        or payload.get("business_date") != expected_date
        or payload.get("processor_version") != envelope.processor_version
        or payload.get("rule_set_sha256") != envelope.rule_set_sha256
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured result identity is invalid")
    counts = {
        field: _require_nonnegative_integer(payload.get(field), field)
        for field in (
            "source_rows",
            "removed_by_rate_code",
            "removed_as_duplicates",
            "output_rows",
            "candidate_rows",
            "review_required_rows",
            "review_issue_count",
            "manually_priced_rows",
        )
    }
    outcomes = _require_exact_mapping(payload.get("outcome_counts"), V4_OUTCOME_FIELDS, "outcome counts")
    for outcome in V4_OUTCOME_FIELDS:
        _require_nonnegative_integer(outcomes.get(outcome), f"outcome_counts.{outcome}")
    actual_outcomes, manual_rows = _validate_v4_records(payload)
    if dict(outcomes) != actual_outcomes:
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured outcomes do not reconcile")
    if (
        counts["removed_by_rate_code"] != outcomes.get("excluded_rate_code")
        or counts["removed_as_duplicates"] != outcomes.get("duplicate")
        or counts["output_rows"] != outcomes.get("retained")
        or counts["candidate_rows"] != outcomes.get("candidate")
        or counts["review_required_rows"] != outcomes.get("price_unmatched")
        or counts["manually_priced_rows"] != manual_rows
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured counters do not reconcile")
    channels = payload.get("channels")
    errors = payload.get("errors")
    if not isinstance(channels, list) or not isinstance(errors, list):
        raise IngestionError("RESULT_CONTRACT_INVALID", "structured arrays are invalid")
    channel_names: set[str] = set()
    channel_rows = 0
    for channel in channels:
        values = _require_exact_mapping(channel, {"worksheet", "rows"}, "structured channel")
        worksheet = values.get("worksheet")
        rows = _require_nonnegative_integer(values.get("rows"), "structured channel rows")
        if not isinstance(worksheet, str) or not worksheet.strip() or worksheet in channel_names:
            raise IngestionError("RESULT_CONTRACT_INVALID", "structured channel is invalid")
        channel_names.add(worksheet)
        channel_rows += rows
    for error in errors:
        _require_exact_mapping(error, ERROR_FIELDS, "structured error")
    review_keys = _validate_review_issues(payload.get("review_issues"))
    if len(review_keys) != counts["review_issue_count"]:
        raise IngestionError("RESULT_CONTRACT_INVALID", "review issue count does not reconcile")

    artifacts = _require_exact_mapping(
        payload.get("artifacts"), V4_STRUCTURED_ARTIFACT_FIELDS, "structured artifacts"
    )
    _inner_artifact("source_xml", artifacts.get("source_xml"), envelope.artifacts["source_xml"])
    _inner_artifact("daily_report", artifacts.get("daily_report"), envelope.artifacts["daily_report"])
    _inner_artifact("result_json", artifacts.get("result_json"), envelope.artifacts["result_json"])
    _inner_artifact("exception_report", artifacts.get("exception_report"), envelope.artifacts["exception_report"])
    _inner_artifact(
        "manual_override_json",
        artifacts.get("manual_override_json"),
        envelope.artifacts.get("manual_override_json"),
    )
    manual_ref = envelope.artifacts.get("manual_override_json")
    review_case_id = payload.get("review_case_id")
    override_sha256 = payload.get("manual_override_sha256")

    if envelope.status == "success":
        if (
            errors
            or outcomes.get("validation_failed") != 0
            or outcomes.get("price_unmatched") != 0
            or outcomes.get("candidate") != 0
            or counts["source_rows"]
            != counts["removed_by_rate_code"] + counts["removed_as_duplicates"] + counts["output_rows"]
            or channel_rows != counts["output_rows"]
            or review_keys
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "successful result is not balanced")
        if manual_ref is None:
            if review_case_id is not None or override_sha256 is not None or manual_rows != 0:
                raise IngestionError("RESULT_CONTRACT_INVALID", "automatic success has manual-review lineage")
        elif (
            not isinstance(review_case_id, str)
            or re.fullmatch(r"dailyreview-[0-9a-f]{32}", review_case_id) is None
            or override_sha256 != manual_ref.sha256
            or manual_rows <= 0
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "manual success lineage is invalid")
    elif envelope.status == "review_required":
        if (
            counts["output_rows"] != 0
            or not errors
            or any(error.get("code") != "PRICE_UNMATCHED" for error in errors)
            or outcomes.get("validation_failed") != 0
            or outcomes.get("retained") != 0
            or outcomes.get("price_unmatched", 0) <= 0
            or counts["source_rows"]
            != counts["removed_by_rate_code"]
            + counts["removed_as_duplicates"]
            + counts["candidate_rows"]
            + counts["review_required_rows"]
            or channels
            or not review_keys
            or manual_ref is not None
            or review_case_id is not None
            or override_sha256 is not None
            or manual_rows != 0
        ):
            raise IngestionError("RESULT_CONTRACT_INVALID", "review result shape is invalid")
    elif (
        counts["output_rows"] != 0
        or counts["candidate_rows"] != 0
        or channels
        or not errors
        or manual_ref is not None
        or review_case_id is not None
        or override_sha256 is not None
        or manual_rows != 0
        or review_keys
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "failed result shape is invalid")


def _validate_result_payload_v4(
    payload: Mapping[str, Any], envelope: DeliveryEnvelope, structured: Mapping[str, Any]
) -> None:
    _require_exact_mapping(payload, RESULT_FIELDS, "result")
    expected_date = envelope.business_date.isoformat() if envelope.business_date else None
    if (
        payload.get("version") != RESULT_SCHEMA_VERSION
        or payload.get("status") != envelope.status
        or payload.get("business_date") != expected_date
        or not isinstance(payload.get("message"), str)
    ):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result identity is invalid")
    metrics = _require_exact_mapping(payload.get("metrics"), V4_RESULT_METRIC_FIELDS, "result metrics")
    for field in V4_RESULT_METRIC_FIELDS - {"channels"}:
        _require_nonnegative_integer(metrics.get(field), f"metrics.{field}")
        if metrics.get(field) != structured.get(field):
            raise IngestionError("RESULT_CONTRACT_INVALID", "result counters do not reconcile")
    if metrics.get("channels") != structured.get("channels"):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result channels do not reconcile")
    outputs = _require_exact_mapping(
        payload.get("outputs"), {"daily_report", "structured_result", "exception_report"}, "result outputs"
    )
    for output, role in (
        ("daily_report", "daily_report"),
        ("structured_result", "structured_result_json"),
        ("exception_report", "exception_report"),
    ):
        reference = envelope.artifacts[role]
        if outputs.get(output) != (reference.original_filename if reference is not None else None):
            raise IngestionError("RESULT_CONTRACT_INVALID", "result output names are invalid")
    errors = payload.get("errors")
    if not isinstance(errors, list) or errors != structured.get("errors"):
        raise IngestionError("RESULT_CONTRACT_INVALID", "result errors do not reconcile")


def _validate_structured_payload(payload: Mapping[str, Any], envelope: DeliveryEnvelope) -> None:
    if envelope.result_schema_version == LEGACY_RESULT_SCHEMA_VERSION:
        _validate_structured_payload_v3(payload, envelope)
        return
    if envelope.result_schema_version == RESULT_SCHEMA_VERSION:
        _validate_structured_payload_v4(payload, envelope)
        return
    raise IngestionError("RESULT_CONTRACT_INVALID", "structured result schema is unsupported")


def _validate_result_payload(
    payload: Mapping[str, Any], envelope: DeliveryEnvelope, structured: Mapping[str, Any]
) -> None:
    if envelope.result_schema_version == LEGACY_RESULT_SCHEMA_VERSION:
        _validate_result_payload_v3(payload, envelope, structured)
        return
    if envelope.result_schema_version == RESULT_SCHEMA_VERSION:
        _validate_result_payload_v4(payload, envelope, structured)
        return
    raise IngestionError("RESULT_CONTRACT_INVALID", "result schema is unsupported")


class DeliveryValidator:
    def __init__(self, store: ArtifactStore, policy: ProcessorPolicy) -> None:
        self._store = store
        self._policy = policy

    def validate(self, raw_envelope: bytes) -> VerifiedDelivery:
        envelope_payload = strict_json_bytes(raw_envelope, "delivery envelope")
        envelope = DeliveryEnvelope.from_dict(envelope_payload)
        if (
            envelope.processor_version != self._policy.processor_version
            or envelope.rule_set_sha256 != self._policy.rule_set_sha256
        ):
            raise IngestionError(
                "PROCESSOR_NOT_ALLOWED", "delivery processor identity is not approved"
            )
        with tempfile.TemporaryDirectory(prefix="arr-ingestion-") as temporary:
            root = Path(temporary)
            paths: Dict[str, Path] = {}
            for role, reference in envelope.artifacts.items():
                if reference is None:
                    continue
                limit = ARTIFACT_SIZE_LIMITS[role]
                if reference.byte_size > limit:
                    raise IngestionError(
                        "ARTIFACT_TOO_LARGE", "delivery artifact exceeds its size limit"
                    )
                destination = root / role / reference.original_filename
                self._store.materialize(reference.object_key, destination, limit)
                if (
                    not destination.is_file()
                    or destination.stat().st_size != reference.byte_size
                    or sha256_file(destination) != reference.sha256
                ):
                    raise IngestionError(
                        "ARTIFACT_HASH_MISMATCH", "delivery artifact identity does not match"
                    )
                paths[role] = destination

            result_payload = strict_json_file(paths["result_json"], "result")
            structured_payload = strict_json_file(
                paths["structured_result_json"], "structured result"
            )
            _validate_structured_payload(structured_payload, envelope)
            _validate_result_payload(result_payload, envelope, structured_payload)
            if envelope.status in {"success", "review_required"}:
                self._run_independent_validator(paths, envelope, structured_payload)
        return VerifiedDelivery(
            envelope=envelope,
            envelope_sha256=hashlib.sha256(raw_envelope).hexdigest(),
            result_payload=copy.deepcopy(result_payload),
            structured_payload=copy.deepcopy(structured_payload),
        )

    def _run_independent_validator(
        self,
        paths: Mapping[str, Path],
        envelope: DeliveryEnvelope,
        structured_payload: Mapping[str, Any],
    ) -> None:
        validator = self._policy.skill_root / "scripts" / "validate_daily.py"
        price_reference = self._policy.skill_root / "references" / "价格对照.xlsx"
        try:
            command = [
                self._policy.python_binary,
                str(validator),
                "--xml",
                str(paths["source_xml"].resolve()),
                "--result-json",
                str(paths["result_json"].resolve()),
                "--structured-result-json",
                str(paths["structured_result_json"].resolve()),
                "--price-reference",
                str(price_reference.resolve()),
            ]
            if envelope.status == "review_required":
                command.append("--review-only")
            else:
                command.extend(("--daily", str(paths["daily_report"].resolve())))
                manual_path = paths.get("manual_override_json")
                if manual_path is not None:
                    command.extend(
                        (
                            "--manual-override-json",
                            str(manual_path.resolve()),
                            "--review-job-id",
                            envelope.job_id,
                            "--review-case-id",
                            str(structured_payload.get("review_case_id")),
                            "--manual-override-sha256",
                            str(structured_payload.get("manual_override_sha256")),
                        )
                    )
            if self._policy.legacy_direct_v3:
                command.append("--legacy-v3-output")
            completed = subprocess.run(
                command,
                cwd=self._policy.skill_root,
                capture_output=True,
                text=True,
                timeout=120,
                check=False,
            )
        except (OSError, subprocess.TimeoutExpired):
            raise IngestionError(
                "OUTPUT_VALIDATION_UNAVAILABLE",
                "independent result validation did not complete",
                retryable=True,
            ) from None
        if completed.returncode != 0:
            raise IngestionError(
                "OUTPUT_VALIDATION_FAILED", "independent result validation rejected the delivery"
            )
