#!/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())