DataOps-platform/scripts/validate_phase3_wp01_acceptance.py

#!/usr/bin/env python3
"""Fail-closed validator for the P3-WP01 enterprise acceptance package."""

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import math
import re
from datetime import datetime
from pathlib import Path
from typing import Any

BLOCKED = "ENGINEERING_READY_BLOCKED_EXTERNAL"
COMPLETE = "ENTERPRISE_ACCEPTANCE_COMPLETE"
EXPECTED_VERSION = "v0.3.0-phase2"
EXPECTED_COMMIT = "95023e59bfb9f00be4ed61f7a8307fac56fbcee9"
PLACEHOLDERS = {"", "TBD", "TBD_EXTERNAL", "UNKNOWN", "N/A", "PLACEHOLDER"}
PURE_EVIDENCE_PLACEHOLDERS = PLACEHOLDERS | {"BLOCKED_EXTERNAL", "PASS_LOCAL"}
UNFINISHED_STATE_VALUES = {
    "TBD_EXTERNAL",
    "BLOCKED_EXTERNAL",
    "PASS_LOCAL",
    "ENGINEERING_READY_BLOCKED_EXTERNAL",
}
INCOMPLETE_EVIDENCE_VALUES = PURE_EVIDENCE_PLACEHOLDERS | UNFINISHED_STATE_VALUES
CASE_IDS = {f"P2-WP13-UAT-{number:03d}" for number in range(21, 25)}
METRIC_TARGETS = {
    "core_object_onboarding_rate": 95,
    "incremental_change_accuracy": 99,
    "responsibility_coverage": 95,
    "quality_issue_closure_rate": 85,
    "incident_evidence_coverage": 100,
    "data_product_evidence_coverage": 100,
}
REHEARSAL_STEPS = {"install", "upgrade", "backup", "candidate_rollback", "rto_rpo"}
TRAINING_IDS = {"user_training", "operations_training"}
SIGNOFF_ROLES = {
    "product_owner",
    "business_owner",
    "technical_owner",
    "security_owner",
    "operations_owner",
}
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
P1_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$")
CURRENT_STATUS_RE = re.compile(
    r"^\s*(?:[-*]\s*)?(?:\*\*)?(?:status|state|状态|当前状态)(?:\*\*)?\s*[:：=]\s*"
    r"[`'\"*]*\s*(?:TBD_EXTERNAL|BLOCKED_EXTERNAL|PASS_LOCAL|ENGINEERING_READY_BLOCKED_EXTERNAL)"
    r"\s*[`'\"*]*(?:\s|[。；;,.]|$)",
    re.IGNORECASE | re.MULTILINE,
)
HISTORICAL_ENGLISH_TOKENS = {
    "history",
    "historical",
    "previous",
    "prior",
    "superseded",
}
HISTORICAL_CHINESE_KEYS = {"原状态", "先前状态", "前一状态"}


def _missing(value: Any) -> bool:
    return value is None or (isinstance(value, str) and value.strip().upper() in PLACEHOLDERS)


def _reject_json_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is prohibited: {value}")


def _is_finite_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)


def _parse_iso_time(value: Any) -> datetime | None:
    if not isinstance(value, str) or _missing(value):
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo is not None else None


def _normalized_cell(value: str) -> str:
    return value.strip().strip("`'\"* ").upper()


def _is_historical_key(value: Any) -> bool:
    if not isinstance(value, str):
        return False
    key = value.strip().lower()
    if key in HISTORICAL_ENGLISH_TOKENS:
        return True
    if any(
        key.startswith(f"{token}{separator}")
        for token in HISTORICAL_ENGLISH_TOKENS
        for separator in ("_", "-", " ")
    ):
        return True
    if any(key.endswith(f"{separator}history") for separator in ("_", "-", " ")):
        return True
    if key == "历史" or key.startswith("历史"):
        return True
    return key in HISTORICAL_CHINESE_KEYS or any(
        key.startswith(f"{token}{separator}")
        for token in HISTORICAL_CHINESE_KEYS
        for separator in ("_", "-", " ")
    )


def _json_contains_incomplete_value(value: Any, *, historical_context: bool = False) -> bool:
    if isinstance(value, dict):
        for key, child in value.items():
            child_is_historical = historical_context or _is_historical_key(key)
            if _json_contains_incomplete_value(child, historical_context=child_is_historical):
                return True
        return False
    if isinstance(value, list):
        return any(
            _json_contains_incomplete_value(child, historical_context=historical_context)
            for child in value
        )
    return (
        not historical_context
        and isinstance(value, str)
        and value.strip().upper() in INCOMPLETE_EVIDENCE_VALUES
    )


def _is_separator_row(cells: list[str]) -> bool:
    return bool(cells) and all(not cell or re.fullmatch(r":?-{3,}:?", cell.strip()) for cell in cells)


def _table_contains_current_placeholder(rows: list[list[str]]) -> bool:
    rows = [[cell.strip() for cell in row] for row in rows if any(cell.strip() for cell in row)]
    if not rows:
        return False
    header = rows[0]
    historical_columns = {
        index for index, cell in enumerate(header) if _is_historical_key(_normalized_cell(cell))
    }
    for cells in rows:
        if _is_separator_row(cells):
            continue
        if cells and _is_historical_key(_normalized_cell(cells[0])):
            continue
        for index, cell in enumerate(cells):
            if (
                index not in historical_columns
                and _normalized_cell(cell) in INCOMPLETE_EVIDENCE_VALUES
            ):
                return True
    return False


def _is_placeholder_evidence(evidence_bytes: bytes) -> bool:
    """Reject structured current-state placeholders without guessing about binary evidence."""
    try:
        evidence_text = evidence_bytes.decode("utf-8")
    except UnicodeDecodeError:
        return False
    if evidence_text.strip().upper() in INCOMPLETE_EVIDENCE_VALUES:
        return True
    try:
        parsed_json = json.loads(evidence_text)
    except json.JSONDecodeError:
        parsed_json = None
    if parsed_json is not None and _json_contains_incomplete_value(parsed_json):
        return True
    if CURRENT_STATUS_RE.search(evidence_text):
        return True
    markdown_rows = [line.split("|")[1:-1] if line.strip().startswith("|") and line.strip().endswith("|") else line.split("|") for line in evidence_text.splitlines() if "|" in line]
    if _table_contains_current_placeholder(markdown_rows):
        return True
    try:
        csv_rows = list(csv.reader(io.StringIO(evidence_text)))
    except csv.Error:
        csv_rows = []
    return any(len(row) > 1 for row in csv_rows) and _table_contains_current_placeholder(csv_rows)


def _records_by_id(records: Any, key: str, expected: set[str], failures: list[str], label: str) -> dict[str, dict]:
    if not isinstance(records, list):
        failures.append(f"{label} must be a list")
        return {}
    mapped: dict[str, dict] = {}
    for record in records:
        if (
            not isinstance(record, dict)
            or not isinstance(record.get(key), str)
            or _missing(record.get(key))
        ):
            failures.append(f"{label} contains an invalid record")
            continue
        record_id = record[key]
        if record_id in mapped:
            failures.append(f"{label} contains duplicate {record_id}")
        mapped[record_id] = record
    if set(mapped) != expected:
        failures.append(f"{label} ids must be exactly {sorted(expected)}")
    return mapped


def _validate_acceptance_bundle(package_path: Path, evidence_root: Path | None = None) -> dict[str, Any]:
    package_path = Path(package_path).resolve()
    failures: list[str] = []
    blockers: list[str] = []
    try:
        payload = json.loads(
            package_path.read_text(encoding="utf-8"), parse_constant=_reject_json_constant
        )
    except (OSError, json.JSONDecodeError, ValueError) as exc:
        return {"status": "FAILED", "declared_status": None, "blockers": [], "failures": [f"cannot read package: {exc}"]}
    if not isinstance(payload, dict):
        return {"status": "FAILED", "declared_status": None, "blockers": [], "failures": ["package root must be an object"]}

    declared = payload.get("declared_status")
    if not isinstance(declared, str) or declared not in {BLOCKED, COMPLETE}:
        failures.append(f"invalid declared_status: {declared!r}")
    enterprise_claim = declared == COMPLETE
    root = Path(evidence_root).resolve() if evidence_root else (
        package_path.parent.parent.resolve() if package_path.parent.name == "acceptance" else package_path.parent.resolve()
    )

    def pending_or_fail(label: str) -> None:
        message = f"{label} is missing or placeholder"
        (failures if enterprise_claim else blockers).append(message)

    def require(value: Any, label: str) -> bool:
        if _missing(value):
            pending_or_fail(label)
            return False
        return True

    def require_string(value: Any, label: str) -> bool:
        if value is None or (
            isinstance(value, str)
            and value.strip().upper() in INCOMPLETE_EVIDENCE_VALUES
        ):
            pending_or_fail(label)
            return False
        if not isinstance(value, str) or not value.strip():
            failures.append(f"{label} must be a non-empty, non-placeholder string")
            return False
        return True

    def object_field(container: dict[str, Any], key: str, label: str) -> dict[str, Any]:
        value = container.get(key)
        if not isinstance(value, dict):
            failures.append(f"{label} must be an object")
            return {}
        return value

    def require_time(value: Any, label: str) -> datetime | None:
        if not require_string(value, label):
            return None
        parsed = _parse_iso_time(value)
        if parsed is None:
            failures.append(f"{label} must be an ISO-8601 timestamp with timezone")
        return parsed

    def validate_refs(refs: Any, label: str) -> None:
        if not isinstance(refs, list):
            failures.append(f"{label}.evidence_refs must be a list")
            return
        if not refs:
            pending_or_fail(f"{label}.evidence_refs")
            return
        for index, ref in enumerate(refs):
            ref_label = f"{label}.evidence_refs[{index}]"
            if not isinstance(ref, dict):
                failures.append(f"{ref_label} must be an object")
                continue
            raw_path = ref.get("path")
            if not require_string(raw_path, f"{ref_label}.path"):
                continue
            relative = Path(raw_path)
            if relative.is_absolute() or ".." in relative.parts:
                failures.append(f"{ref_label} evidence path escapes evidence root")
                continue
            resolved = (root / relative).resolve()
            try:
                resolved.relative_to(root)
            except ValueError:
                failures.append(f"{ref_label} evidence path escapes evidence root")
                continue
            digest = ref.get("sha256")
            level = ref.get("evidence_level")
            if not resolved.is_file():
                if enterprise_claim or not _missing(digest):
                    failures.append(f"{ref_label} missing evidence file: {raw_path}")
                else:
                    blockers.append(f"{ref_label} external evidence file is not yet captured")
            if require_string(digest, f"{ref_label}.sha256") and not SHA256_RE.fullmatch(digest):
                failures.append(f"{ref_label}.sha256 must be a lowercase SHA-256 digest")
            elif isinstance(digest, str) and SHA256_RE.fullmatch(digest) and resolved.is_file():
                evidence_bytes = resolved.read_bytes()
                actual = hashlib.sha256(evidence_bytes).hexdigest()
                if actual != digest:
                    failures.append(f"{ref_label} digest mismatch")
                if _is_placeholder_evidence(evidence_bytes):
                    failures.append(f"{ref_label} placeholder evidence cannot satisfy enterprise acceptance")
            if require_string(level, f"{ref_label}.evidence_level") and level != "ENTERPRISE_FORMAL":
                failures.append(f"{ref_label} evidence level {level!r} is not ENTERPRISE_FORMAL")

    if payload.get("schema_version") != "1.0" or payload.get("work_package") != "P3-WP01":
        failures.append("schema_version/work_package contract mismatch")
    source_cases = payload.get("source_cases")
    if (
        not isinstance(source_cases, list)
        or any(not isinstance(item, str) or _missing(item) for item in source_cases)
        or set(source_cases) != CASE_IDS
    ):
        failures.append("source_cases must contain UAT-021 through UAT-024 exactly")

    context = object_field(payload, "execution_context", "execution_context")
    for field in ("enterprise_name", "environment_id"):
        require_string(context.get(field), f"execution_context.{field}")
    if require_string(context.get("environment_type"), "execution_context.environment_type") and context.get("environment_type") != "PREPRODUCTION":
        failures.append("execution_context.environment_type must be PREPRODUCTION")
    commit = context.get("executed_commit")
    if require_string(commit, "execution_context.executed_commit") and commit != EXPECTED_COMMIT:
        failures.append(f"execution_context.executed_commit must equal {EXPECTED_COMMIT}")
    context_started = require_time(context.get("started_at"), "execution_context.started_at")
    context_completed = require_time(context.get("completed_at"), "execution_context.completed_at")
    if context_started and context_completed and context_started > context_completed:
        failures.append("execution_context.started_at must not be after completed_at")

    completion_events: list[tuple[str, datetime]] = []

    def bind_time(value: Any, label: str, *, completion_event: bool = False) -> datetime | None:
        parsed = require_time(value, label)
        if parsed and context_started and parsed < context_started:
            failures.append(f"{label} must be within execution_context time range")
        if parsed and context_completed and parsed > context_completed:
            failures.append(f"{label} must be within execution_context time range")
        if parsed and completion_event:
            completion_events.append((label, parsed))
        return parsed

    source = object_field(payload, "source", "source")
    for field in ("source_id", "source_type", "network_zone"):
        require_string(source.get(field), f"source.{field}")
    if source.get("readonly") is None:
        pending_or_fail("source.readonly")
    elif source.get("readonly") is not True:
        failures.append("source.readonly must be true")
    if source.get("credentials_in_package") is not False:
        failures.append("source.credentials_in_package must be false")

    sample = object_field(payload, "sample_summary", "sample_summary")
    if require_string(sample.get("classification"), "sample_summary.classification") and sample.get("classification") != "DESENSITIZED_ENTERPRISE_SAMPLE":
        failures.append("sample_summary.classification must be DESENSITIZED_ENTERPRISE_SAMPLE")
    for field in ("object_count", "row_count"):
        value = sample.get(field)
        if not require(value, f"sample_summary.{field}"):
            continue
        if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
            failures.append(f"sample_summary.{field} must be a positive integer")
    sample_digest = sample.get("sha256")
    if require_string(sample_digest, "sample_summary.sha256") and not SHA256_RE.fullmatch(sample_digest):
        failures.append("sample_summary.sha256 must be a lowercase SHA-256 digest")
    if sample.get("raw_data_in_evidence") is not False:
        failures.append("sample_summary.raw_data_in_evidence must be false")

    artifact = object_field(payload, "release_artifact", "release_artifact")
    version = artifact.get("version")
    if require_string(version, "release_artifact.version") and version != EXPECTED_VERSION:
        failures.append(f"release_artifact.version must equal {EXPECTED_VERSION}")
    require_string(artifact.get("artifact_name"), "release_artifact.artifact_name")
    artifact_digest = artifact.get("sha256")
    if require_string(artifact_digest, "release_artifact.sha256") and not SHA256_RE.fullmatch(artifact_digest):
        failures.append("release_artifact.sha256 must be a lowercase SHA-256 digest")
    validate_refs(artifact.get("evidence_refs"), "release_artifact")

    cases = _records_by_id(payload.get("uat_cases"), "id", CASE_IDS, failures, "uat_cases")
    for case_id, case in cases.items():
        result = case.get("result")
        if result == "PASS_LOCAL":
            failures.append(f"{case_id} PASS_LOCAL cannot satisfy enterprise acceptance")
            result_valid = False
        else:
            result_valid = require_string(result, f"{case_id}.result")
        if result_valid and result != "ENTERPRISE_PASS":
            if enterprise_claim:
                failures.append(f"{case_id} must be ENTERPRISE_PASS")
            else:
                blockers.append(f"{case_id} awaits enterprise execution")
        for field in ("environment_id", "executed_by_person_id", "approved_by_person_id"):
            require_string(case.get(field), f"{case_id}.{field}")
        if isinstance(case.get("environment_id"), str) and isinstance(context.get("environment_id"), str) and case.get("environment_id") != context.get("environment_id"):
            failures.append(f"{case_id}.environment_id must match execution_context.environment_id")
        bind_time(case.get("executed_at"), f"{case_id}.executed_at", completion_event=True)
        if case.get("executed_by_person_id") and case.get("executed_by_person_id") == case.get("approved_by_person_id"):
            failures.append(f"{case_id} self-approval is prohibited")
        validate_refs(case.get("evidence_refs"), case_id)

    metrics = _records_by_id(payload.get("metrics"), "id", set(METRIC_TARGETS), failures, "metrics")
    for metric_id, metric in metrics.items():
        target = METRIC_TARGETS[metric_id]
        comparator_valid = require_string(metric.get("comparator"), f"{metric_id}.comparator")
        unit_valid = require_string(metric.get("unit"), f"{metric_id}.unit")
        target_value = metric.get("target")
        target_ready = require(target_value, f"{metric_id}.target")
        target_valid = target_ready and _is_finite_number(target_value)
        if target_ready and not target_valid:
            failures.append(f"{metric_id}.target must be a finite number")
        if not (comparator_valid and unit_valid and target_valid) or metric.get("comparator") != "GTE" or target_value != target or metric.get("unit") != "percent":
            failures.append(f"{metric_id} threshold contract must be GTE {target} percent")
        metric_status_valid = require_string(metric.get("status"), f"{metric_id}.status")
        if metric_status_valid and metric.get("status") != "ENTERPRISE_PASS":
            (failures if enterprise_claim else blockers).append(f"{metric_id} awaits enterprise measurement")
        numerator, denominator, value = metric.get("numerator"), metric.get("denominator"), metric.get("value")
        values_ready = True
        for item, field in ((numerator, "numerator"), (denominator, "denominator"), (value, "value")):
            if not require(item, f"{metric_id}.{field}"):
                values_ready = False
            elif not _is_finite_number(item):
                failures.append(f"{metric_id}.{field} must be a finite number")
                values_ready = False
        if values_ready:
            if numerator < 0 or denominator <= 0 or value < 0:
                failures.append(f"{metric_id} metric values are invalid")
            else:
                calculated = numerator / denominator * 100
                if abs(calculated - value) > 1e-6:
                    failures.append(f"{metric_id} value does not match numerator/denominator")
                if value < target:
                    failures.append(f"{metric_id} threshold {target}% is not met")
        window_start = bind_time(metric.get("window_start"), f"{metric_id}.window_start")
        window_end = bind_time(metric.get("window_end"), f"{metric_id}.window_end", completion_event=True)
        if window_start and window_end and window_start > window_end:
            failures.append(f"{metric_id}.window_start must not be after window_end")
        require_string(metric.get("approved_by_person_id"), f"{metric_id}.approved_by_person_id")
        validate_refs(metric.get("evidence_refs"), metric_id)

    rehearsal = object_field(payload, "preproduction_rehearsal", "preproduction_rehearsal")
    rehearsal_status_valid = require_string(rehearsal.get("status"), "preproduction_rehearsal.status")
    if rehearsal_status_valid and rehearsal.get("status") != "ENTERPRISE_PASS":
        (failures if enterprise_claim else blockers).append("preproduction rehearsal is incomplete")
    for field in ("environment_id", "executed_by_person_id", "approved_by_person_id"):
        require_string(rehearsal.get(field), f"preproduction_rehearsal.{field}")
    if isinstance(rehearsal.get("environment_id"), str) and isinstance(context.get("environment_id"), str) and rehearsal.get("environment_id") != context.get("environment_id"):
        failures.append("preproduction_rehearsal.environment_id must match execution_context.environment_id")
    bind_time(rehearsal.get("executed_at"), "preproduction_rehearsal.executed_at", completion_event=True)
    if rehearsal.get("executed_by_person_id") and rehearsal.get("executed_by_person_id") == rehearsal.get("approved_by_person_id"):
        failures.append("preproduction rehearsal self-approval is prohibited")
    steps = _records_by_id(rehearsal.get("steps"), "id", REHEARSAL_STEPS, failures, "preproduction rehearsal steps")
    for step_id, step in steps.items():
        step_status_valid = require_string(step.get("status"), f"preproduction rehearsal step {step_id}.status")
        if step_status_valid and step.get("status") != "ENTERPRISE_PASS":
            (failures if enterprise_claim else blockers).append(f"preproduction rehearsal step {step_id} is incomplete")
    for observed, target_field in (("rto_minutes", "rto_target_minutes"), ("rpo_minutes", "rpo_target_minutes")):
        target_value = rehearsal.get(target_field)
        target_valid = require(target_value, f"preproduction_rehearsal.{target_field}")
        if target_valid and (not _is_finite_number(target_value) or target_value <= 0):
            failures.append(f"preproduction_rehearsal.{target_field} must be a positive finite number")
            target_valid = False
        value = rehearsal.get(observed)
        value_valid = require(value, f"preproduction_rehearsal.{observed}")
        if value_valid and (not _is_finite_number(value) or value < 0):
            failures.append(f"preproduction_rehearsal.{observed} must be a non-negative finite number")
            value_valid = False
        if target_valid and value_valid and value > target_value:
            failures.append(f"preproduction rehearsal {observed} exceeds its target")
    validate_refs(rehearsal.get("evidence_refs"), "preproduction_rehearsal")

    trainings = _records_by_id(payload.get("training_records"), "id", TRAINING_IDS, failures, "training_records")
    for training_id, record in trainings.items():
        training_status_valid = require_string(record.get("status"), f"training {training_id}.status")
        if training_status_valid and record.get("status") != "ENTERPRISE_PASS":
            (failures if enterprise_claim else blockers).append(f"training {training_id} is incomplete")
        participants = record.get("participants")
        if not isinstance(participants, list):
            failures.append(f"training {training_id}.participants must be a list")
        elif not participants:
            pending_or_fail(f"training {training_id}.participants")
        else:
            for index, participant in enumerate(participants):
                require_string(participant, f"training {training_id}.participants[{index}]")
        require_string(record.get("trainer_person_id"), f"training {training_id}.trainer_person_id")
        bind_time(record.get("completed_at"), f"training {training_id}.completed_at", completion_event=True)
        if record.get("exercise_passed") is not True:
            (failures if enterprise_claim else blockers).append(f"training {training_id} exercise is incomplete")
        if record.get("retraining_required") is None:
            pending_or_fail(f"training {training_id}.retraining_required")
        elif not isinstance(record.get("retraining_required"), bool):
            failures.append(f"training {training_id}.retraining_required must be boolean")
        if record.get("retraining_completed") is None:
            pending_or_fail(f"training {training_id}.retraining_completed")
        elif not isinstance(record.get("retraining_completed"), bool):
            failures.append(f"training {training_id}.retraining_completed must be boolean")
        if record.get("retraining_required") is True and record.get("retraining_completed") is not True:
            failures.append(f"training {training_id} required retraining is incomplete")
        validate_refs(record.get("evidence_refs"), f"training {training_id}")

    defects = object_field(payload, "defect_gate", "defect_gate")
    defect_status_valid = require_string(defects.get("status"), "defect_gate.status")
    if defect_status_valid and defects.get("status") != "ENTERPRISE_PASS":
        (failures if enterprise_claim else blockers).append("defect gate awaits enterprise review")
    p0_open = defects.get("p0_open")
    if require(p0_open, "defect_gate.p0_open") and (
        not isinstance(p0_open, int) or isinstance(p0_open, bool) or p0_open != 0
    ):
        failures.append("defect_gate.p0_open must be 0")
    p1_open = defects.get("p1_open")
    p1_ready = require(p1_open, "defect_gate.p1_open")
    if p1_ready and (not isinstance(p1_open, int) or isinstance(p1_open, bool) or p1_open < 0):
        failures.append("defect_gate.p1_open must be a non-negative integer")
        p1_ready = False
    open_p1_ids = defects.get("open_p1_ids")

    def valid_nonnegative_count(value: Any, label: str) -> bool:
        if not isinstance(value, int) or isinstance(value, bool) or value < 0:
            failures.append(f"{label} must be a non-negative integer")
            return False
        return True

    def valid_p1_ids(value: Any, label: str) -> bool:
        if not isinstance(value, list):
            failures.append(f"{label} must be a list")
            return False
        if any(not isinstance(item, str) or _missing(item) or not P1_ID_RE.fullmatch(item) for item in value):
            failures.append(f"{label} contains an invalid or placeholder P1 id")
            return False
        if len(value) != len(set(value)):
            failures.append(f"{label} must contain unique P1 ids")
            return False
        return True

    disposition = object_field(defects, "p1_disposition", "defect_gate.p1_disposition")
    disposition_status = disposition.get("status")
    if not require_string(disposition_status, "defect_gate.p1_disposition.status"):
        pass
    elif p1_ready and p1_open == 0:
        if open_p1_ids != []:
            failures.append("defect_gate.open_p1_ids must be empty when p1_open is 0")
        if disposition_status != "NOT_REQUIRED":
            failures.append("defect_gate.p1_disposition.status must be NOT_REQUIRED when p1_open is 0")
        covered_count = disposition.get("covered_p1_count")
        if not valid_nonnegative_count(covered_count, "defect_gate.p1_disposition.covered_p1_count") or covered_count != 0:
            failures.append("defect_gate.p1_disposition.covered_p1_count must be 0 when p1_open is 0")
        if disposition.get("covered_p1_ids") != []:
            failures.append("defect_gate.p1_disposition.covered_p1_ids must be empty when p1_open is 0")
    elif p1_ready and p1_open > 0:
        open_ids_valid = valid_p1_ids(open_p1_ids, "defect_gate.open_p1_ids")
        if open_ids_valid and len(open_p1_ids) != p1_open:
            failures.append("defect_gate.open_p1_ids count must equal p1_open")
        if disposition_status != "WRITTEN_DECISION_APPROVED":
            failures.append("open P1 defects require WRITTEN_DECISION_APPROVED disposition")
        covered_count = disposition.get("covered_p1_count")
        if not valid_nonnegative_count(covered_count, "defect_gate.p1_disposition.covered_p1_count") or covered_count != p1_open:
            failures.append("P1 written decision must cover every open P1 defect")
        covered_p1_ids = disposition.get("covered_p1_ids")
        covered_ids_valid = valid_p1_ids(covered_p1_ids, "defect_gate.p1_disposition.covered_p1_ids")
        if open_ids_valid and covered_ids_valid and set(covered_p1_ids) != set(open_p1_ids):
            failures.append("P1 written decision covered_p1_ids must exactly match open_p1_ids")
        disposition_business = disposition.get("business_person_id")
        disposition_technical = disposition.get("technical_person_id")
        require_string(disposition_business, "defect_gate.p1_disposition.business_person_id")
        require_string(disposition_technical, "defect_gate.p1_disposition.technical_person_id")
        if disposition_business and disposition_business == disposition_technical:
            failures.append("P1 written decision business and technical approvers must be distinct")
        bind_time(disposition.get("decided_at"), "defect_gate.p1_disposition.decided_at", completion_event=True)
        validate_refs(disposition.get("evidence_refs"), "defect_gate.p1_disposition")
    business_reviewer = defects.get("reviewed_by_business_person_id")
    technical_reviewer = defects.get("reviewed_by_technical_person_id")
    require_string(business_reviewer, "defect_gate.reviewed_by_business_person_id")
    require_string(technical_reviewer, "defect_gate.reviewed_by_technical_person_id")
    if business_reviewer and business_reviewer == technical_reviewer:
        failures.append("defect gate business and technical reviewers must be distinct")
    bind_time(defects.get("reviewed_at"), "defect_gate.reviewed_at", completion_event=True)
    validate_refs(defects.get("evidence_refs"), "defect_gate")

    signoffs = _records_by_id(payload.get("signoffs"), "role", SIGNOFF_ROLES, failures, "signoffs")
    signer_ids: list[str] = []
    signoff_times: list[tuple[str, datetime]] = []
    for role, signoff in signoffs.items():
        signoff_status_valid = require_string(signoff.get("status"), f"{role}.status")
        if signoff_status_valid and signoff.get("status") != "SIGNED":
            (failures if enterprise_claim else blockers).append(f"{role} signoff is not signed")
        if require_string(signoff.get("person_id"), f"{role}.person_id"):
            signer_ids.append(signoff["person_id"])
        require_string(signoff.get("person_name"), f"{role}.person_name")
        signed_at = bind_time(signoff.get("signed_at"), f"{role}.signed_at")
        if signed_at:
            signoff_times.append((role, signed_at))
        validate_refs(signoff.get("evidence_refs"), role)
    if len(signer_ids) != len(set(signer_ids)):
        failures.append("five-party proxy signing is prohibited; each role needs a distinct person_id")
    if completion_events:
        latest_label, latest_completion = max(completion_events, key=lambda item: item[1])
        for role, signed_at in signoff_times:
            if signed_at < latest_completion:
                failures.append(f"{role}.signed_at must not precede completion event {latest_label}")

    failures = list(dict.fromkeys(failures))
    blockers = list(dict.fromkeys(blockers))
    if failures:
        status = "FAILED"
    elif blockers:
        status = "FAILED" if enterprise_claim else "BLOCKED"
        if enterprise_claim:
            failures = [f"acceptance claim is incomplete: {blocker}" for blocker in blockers]
            blockers = []
    elif enterprise_claim:
        status = "ACCEPTED"
    else:
        status = "FAILED"
        failures = ["blocked package contains no external blockers; declared status is inconsistent"]
    return {"status": status, "declared_status": declared, "blockers": blockers, "failures": failures}


def validate_acceptance_bundle(package_path: Path, evidence_root: Path | None = None) -> dict[str, Any]:
    """Validate a bundle and convert expected malformed-input errors into FAILED."""
    try:
        return _validate_acceptance_bundle(package_path, evidence_root)
    except (KeyError, OverflowError, TypeError, ValueError) as exc:
        return {
            "status": "FAILED",
            "declared_status": None,
            "blockers": [],
            "failures": [f"malformed acceptance package: {type(exc).__name__}: {exc}"],
        }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("package", type=Path, help="enterprise acceptance JSON package")
    parser.add_argument("--evidence-root", type=Path)
    args = parser.parse_args()
    result = validate_acceptance_bundle(args.package, args.evidence_root)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return {"ACCEPTED": 0, "FAILED": 2, "BLOCKED": 3}[result["status"]]


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