"""P0-005 credential escrow evidence intake readiness.

This read-only projection turns the Backup / DR escrow blocker into a direct
API contract. It only reads committed, redacted metadata snapshots; it never
reads credential values, writes escrow markers, triggers backups/restores, or
touches hosts.
"""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

from src.services.backup_dr_readiness_matrix import (
    load_latest_backup_dr_readiness_matrix,
)
from src.services.snapshot_paths import default_operations_dir, default_security_dir

_DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__))
_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
_OWNER_REQUEST_FILE = "credential-escrow-evidence-owner-request.snapshot.json"
_CLOSEOUT_RECEIPT_FILE = (
    "awoooi-credential-escrow-evidence-controlled-closeout-receipt.snapshot.json"
)
_SCHEMA_VERSION = "credential_escrow_evidence_intake_readiness_v1"
_CLOSEOUT_RECEIPT_SCHEMA_VERSION = (
    "credential_escrow_evidence_controlled_closeout_receipt_v1"
)
_VALIDATION_SCHEMA_VERSION = "credential_escrow_evidence_owner_response_validation_v1"
_EVIDENCE_REFS_VALIDATION_SCHEMA_VERSION = (
    "credential_escrow_evidence_refs_validation_v1"
)
_SINGLE_PREFLIGHT_INTAKE_SCHEMA_VERSION = "credential_escrow_single_preflight_intake_v1"
_OWNER_PACKET_SCHEMA = "awoooi_post_reboot_next_gate_owner_packets_v1"
_OWNER_REQUEST_SCOPE = "credential_escrow_evidence_owner_request"
_OWNER_RESPONSE_SCHEMA = "awoooi_post_reboot_next_gate_owner_response_v1"
_GATE_ID = "credential_escrow_evidence"
_REQUIRED_ITEM_ORDER = (
    "restic_repository_password",
    "offsite_provider_credentials",
    "break_glass_admin_credentials",
    "dns_registrar_recovery",
    "oauth_ai_provider_recovery",
)
_REQUIRED_ITEM_IDS = set(_REQUIRED_ITEM_ORDER)
_FORBIDDEN_BOOLEAN_FIELDS = {
    "runtime_action_requested",
    "runtime_action_authorized",
    "host_write_requested",
    "host_write_authorized",
    "secret_value_included",
    "secret_value_collection_allowed",
    "credential_marker_write_requested",
    "credential_marker_write_authorized",
}
_PLACEHOLDER_VALUES = {
    "",
    "pending",
    "todo",
    "tbd",
    "n/a",
    "na",
    "owner_role_here",
    "owner_team_here",
    "decision_reason_here",
    "redacted_evidence_ref_here",
    "non_secret_evidence_ref_here",
    "followup_owner_here",
    "reviewer_here",
}
_SECRET_VALUE_PATTERNS = {
    "authorization_header": re.compile(r"Authorization\s*:", re.IGNORECASE),
    "bearer_token": re.compile(r"Bearer\s+[A-Za-z0-9._~+/=-]{12,}", re.IGNORECASE),
    "client_keys": re.compile(r"\bclient\.keys\b", re.IGNORECASE),
    "password_assignment": re.compile(r"\bpassword\s*[:=]\s*[^,\s]+", re.IGNORECASE),
    "private_key": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
    "token_assignment": re.compile(r"\btoken\s*[:=]\s*[^,\s]+", re.IGNORECASE),
}


def load_latest_credential_escrow_evidence_intake_readiness(
    security_dir: Path | None = None,
    backup_dr_readiness_matrix: dict[str, Any] | None = None,
    operations_dir: Path | None = None,
) -> dict[str, Any]:
    """Load P0-005 credential escrow evidence intake readiness."""
    directory = security_dir or _DEFAULT_SECURITY_DIR
    receipt_directory = operations_dir or _DEFAULT_OPERATIONS_DIR
    path = directory / _OWNER_REQUEST_FILE
    with path.open(encoding="utf-8") as handle:
        owner_request = json.load(handle)

    if not isinstance(owner_request, dict):
        raise ValueError(f"{path}: expected JSON object")
    matrix = backup_dr_readiness_matrix or load_latest_backup_dr_readiness_matrix()
    if not isinstance(matrix, dict):
        raise ValueError("backup_dr_readiness_matrix: expected JSON object")

    closeout_receipt_path = receipt_directory / _CLOSEOUT_RECEIPT_FILE
    closeout_receipt = _load_closeout_receipt(closeout_receipt_path)
    _require_owner_request(owner_request, str(path))
    _require_backup_rollups(matrix, "backup_dr_readiness_matrix")
    payload = _build_payload(
        owner_request,
        matrix,
        path,
        closeout_receipt=closeout_receipt,
        closeout_receipt_path=closeout_receipt_path if closeout_receipt else None,
    )
    _require_operation_boundaries(payload, str(path))
    _require_rollup_consistency(payload, str(path))
    _require_single_preflight_intake(payload, str(path))
    _require_payload_closeout_consistency(payload, str(path))
    return payload


def validate_credential_escrow_evidence_owner_response(
    owner_response: dict[str, Any],
    readiness: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Validate one redacted P0-005 owner response without persisting it."""
    current = readiness or load_latest_credential_escrow_evidence_intake_readiness()
    blockers: list[str] = []
    sensitive_hits = _find_sensitive_strings(owner_response)
    forbidden_boolean_hits = _find_forbidden_booleans(owner_response)
    if owner_response.get("schema_version") != _OWNER_RESPONSE_SCHEMA:
        blockers.append(f"schema_version={owner_response.get('schema_version')!r}")

    responses = [
        item
        for item in _list(owner_response.get("responses"))
        if isinstance(item, dict) and str(item.get("gate_id") or "") == _GATE_ID
    ]
    response = responses[0] if responses else {}
    if not response:
        blockers.append(f"{_GATE_ID}.response_missing")

    for key in (
        "owner_role",
        "owner_team",
        "decision",
        "decision_reason",
        "affected_scope",
        "followup_owner",
    ):
        if _is_placeholder(response.get(key)):
            blockers.append(f"{_GATE_ID}.{key}_missing")
    decision = str(response.get("decision") or "").strip().lower()
    if decision not in {"accepted", "needs_supplement", "rejected"}:
        blockers.append(f"{_GATE_ID}.decision_invalid={decision!r}")
    elif decision != "accepted":
        blockers.append(f"{_GATE_ID}.decision_not_accepted={decision!r}")

    evidence_refs = [
        ref
        for ref in _list(response.get("redacted_evidence_refs"))
        if not _is_placeholder(ref)
    ]
    if not evidence_refs:
        blockers.append(f"{_GATE_ID}.redacted_evidence_refs_missing")

    escrow_items = [
        item for item in _list(response.get("escrow_items")) if isinstance(item, dict)
    ]
    seen_item_ids = {
        str(item.get("item_id") or "").strip()
        for item in escrow_items
        if str(item.get("item_id") or "").strip()
    }
    missing_item_ids = sorted(_REQUIRED_ITEM_IDS - seen_item_ids)
    unknown_item_ids = sorted(seen_item_ids - _REQUIRED_ITEM_IDS)
    if missing_item_ids:
        blockers.append(f"{_GATE_ID}.missing_items={missing_item_ids}")
    for item_id in unknown_item_ids:
        blockers.append(f"{_GATE_ID}.unknown_item={item_id!r}")
    accepted_item_ids: list[str] = []
    for item in escrow_items:
        item_id = str(item.get("item_id") or "").strip()
        if item_id not in _REQUIRED_ITEM_IDS:
            continue
        item_blocked = False
        for key in (
            "non_secret_evidence_ref",
            "recovery_owner",
            "reviewer",
            "last_reviewed_at",
        ):
            if _is_placeholder(item.get(key)):
                blockers.append(f"{_GATE_ID}.{item_id}.{key}_missing")
                item_blocked = True
        if item.get("contains_secret_value") is not False:
            blockers.append(f"{_GATE_ID}.{item_id}.contains_secret_value_not_false")
            item_blocked = True
        if not item_blocked:
            accepted_item_ids.append(item_id)

    if forbidden_boolean_hits:
        status = "rejected_runtime_or_marker_action"
    elif sensitive_hits:
        status = "quarantined_sensitive_payload"
    elif blockers:
        status = "needs_supplement"
    else:
        status = "accepted_for_independent_reviewer_readiness_only"

    owner_response_received_count = 1 if status in {
        "accepted_for_independent_reviewer_readiness_only",
        "needs_supplement",
    } and response else 0
    owner_response_accepted_count = (
        1 if status == "accepted_for_independent_reviewer_readiness_only" else 0
    )
    current_rollups = _dict(current.get("rollups"))
    accepted_item_count = len(set(accepted_item_ids))
    projected_effective_missing = (
        0
        if status == "accepted_for_independent_reviewer_readiness_only"
        and accepted_item_count == len(_REQUIRED_ITEM_IDS)
        else _int(current_rollups.get("effective_escrow_missing_count"))
    )
    return {
        "schema_version": _VALIDATION_SCHEMA_VERSION,
        "status": status,
        "priority": "P0-005",
        "scope": "credential_escrow_evidence_intake",
        "source_readiness_status": current.get("status"),
        "result": {
            "owner_response_received_count": owner_response_received_count,
            "owner_response_accepted_count": owner_response_accepted_count,
            "required_item_count": len(_REQUIRED_ITEM_IDS),
            "provided_item_count": len(seen_item_ids & _REQUIRED_ITEM_IDS),
            "accepted_item_count": accepted_item_count,
            "current_effective_escrow_missing_count": _int(
                current_rollups.get("effective_escrow_missing_count")
            ),
            "projected_effective_escrow_missing_count": projected_effective_missing,
            "missing_item_ids": missing_item_ids,
            "unknown_item_ids": unknown_item_ids,
            "redacted_evidence_ref_count": len(evidence_refs),
            "blocker_count": len(blockers),
            "sensitive_payload_hit_count": len(sensitive_hits),
            "forbidden_true_field_count": len(forbidden_boolean_hits),
            "runtime_gate_count": 0,
            "credential_marker_write_authorized_count": 0,
            "secret_value_collection_allowed": False,
        },
        "blockers": blockers,
        "sensitive_payload_hits": sensitive_hits,
        "forbidden_boolean_hits": forbidden_boolean_hits,
        "operation_boundaries": {
            "payload_persisted": False,
            "backup_execution_performed": False,
            "restore_execution_performed": False,
            "offsite_sync_execution_performed": False,
            "credential_marker_write_performed": False,
            "credential_read_performed": False,
            "secret_plaintext_read": False,
            "secret_value_collection_allowed": False,
            "workflow_trigger_performed": False,
            "host_or_k8s_write_performed": False,
            "raw_session_or_sqlite_read_performed": False,
            "runtime_action_performed": False,
        },
        "reviewer_readiness": {
            "schema_version": "credential_escrow_reviewer_readiness_v1",
            "status": (
                "ready_for_reviewer_acceptance_writeback"
                if status == "accepted_for_independent_reviewer_readiness_only"
                else "not_ready_for_reviewer_acceptance_writeback"
            ),
            "accepted_required_item_ids": sorted(set(accepted_item_ids)),
            "missing_required_item_ids": missing_item_ids,
            "current_effective_escrow_missing_count": _int(
                current_rollups.get("effective_escrow_missing_count")
            ),
            "projected_effective_escrow_missing_count": projected_effective_missing,
            "redacted_receipt_writeback_ready_count": (
                1 if status == "accepted_for_independent_reviewer_readiness_only" else 0
            ),
            "credential_marker_write_authorized_count": 0,
            "runtime_gate_count": 0,
            "secret_value_collection_allowed": False,
            "payload_persisted": False,
            "safe_next_step": (
                "write_redacted_reviewer_receipt_snapshot_without_secret_values"
                if status == "accepted_for_independent_reviewer_readiness_only"
                else "supplement_redacted_non_secret_evidence_refs_without_secret_values"
            ),
            "blocked_operations": [
                "credential_marker_write",
                "credential_read",
                "secret_plaintext_export",
                "backup_execution",
                "restore_execution",
                "offsite_sync_execution",
                "workflow_dispatch",
                "host_or_k8s_write",
            ],
        },
        "safe_next_step": (
            "independent_reviewer_acceptance_then_marker_dry_run"
            if status == "accepted_for_independent_reviewer_readiness_only"
            else "supplement_redacted_non_secret_evidence_refs_without_secret_values"
        ),
    }


def validate_credential_escrow_evidence_refs(
    evidence_refs_payload: dict[str, Any],
    readiness: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Validate simplified redacted evidence refs without persistence or marker writes."""
    current = readiness or load_latest_credential_escrow_evidence_intake_readiness()
    source_sensitive_hits = _find_sensitive_strings(evidence_refs_payload)
    source_forbidden_boolean_hits = _find_forbidden_booleans(evidence_refs_payload)
    owner_response = _owner_response_from_evidence_refs(evidence_refs_payload)
    validation = validate_credential_escrow_evidence_owner_response(
        owner_response,
        readiness=current,
    )
    rollups = _dict(current.get("rollups"))
    submissions = _normalized_evidence_ref_submissions(evidence_refs_payload)
    provided_item_ids = sorted(
        {
            str(item.get("item_id") or "").strip()
            for item in submissions
            if str(item.get("item_id") or "").strip() in _REQUIRED_ITEM_IDS
        }
    )

    status = str(validation.get("status") or "")
    if source_forbidden_boolean_hits:
        status = "rejected_runtime_or_marker_action"
    elif source_sensitive_hits:
        status = "quarantined_sensitive_payload"

    result = dict(_dict(validation.get("result")))
    result.update(
        {
            "simplified_evidence_ref_submission_count": len(submissions),
            "provided_required_item_ids": provided_item_ids,
            "missing_required_item_ids": sorted(
                _REQUIRED_ITEM_IDS - set(provided_item_ids)
            ),
            "source_sensitive_payload_hit_count": len(source_sensitive_hits),
            "source_forbidden_true_field_count": len(source_forbidden_boolean_hits),
            "payload_persisted": False,
            "runtime_gate_count": 0,
            "credential_marker_write_authorized_count": 0,
            "secret_value_collection_allowed": False,
        }
    )
    if status != validation.get("status"):
        result["owner_response_accepted_count"] = 0
        result["projected_effective_escrow_missing_count"] = _int(
            rollups.get("effective_escrow_missing_count")
        )

    reviewer_readiness = dict(_dict(validation.get("reviewer_readiness")))
    if status != validation.get("status"):
        reviewer_readiness.update(
            {
                "status": "not_ready_for_reviewer_acceptance_writeback",
                "redacted_receipt_writeback_ready_count": 0,
                "projected_effective_escrow_missing_count": _int(
                    rollups.get("effective_escrow_missing_count")
                ),
            }
        )

    return {
        "schema_version": _EVIDENCE_REFS_VALIDATION_SCHEMA_VERSION,
        "status": status,
        "priority": "P0-005",
        "scope": "credential_escrow_evidence_refs_intake",
        "source_readiness_status": current.get("status"),
        "result": result,
        "blockers": validation.get("blockers", []),
        "sensitive_payload_hits": sorted(
            set(_list(validation.get("sensitive_payload_hits")) + source_sensitive_hits)
        ),
        "forbidden_boolean_hits": sorted(
            set(
                _list(validation.get("forbidden_boolean_hits"))
                + source_forbidden_boolean_hits
            )
        ),
        "operation_boundaries": _dict(validation.get("operation_boundaries")),
        "reviewer_readiness": reviewer_readiness,
        "source_owner_response_validation": {
            "schema_version": validation.get("schema_version"),
            "status": validation.get("status"),
            "blocker_count": len(_list(validation.get("blockers"))),
            "accepted_item_count": _dict(validation.get("result")).get(
                "accepted_item_count"
            ),
            "projected_effective_escrow_missing_count": _dict(
                validation.get("result")
            ).get("projected_effective_escrow_missing_count"),
        },
        "safe_next_step": (
            "independent_reviewer_acceptance_then_marker_dry_run"
            if status == "accepted_for_independent_reviewer_readiness_only"
            else "supplement_redacted_non_secret_evidence_refs_without_secret_values"
        ),
    }


def _build_payload(
    owner_request: dict[str, Any],
    matrix: dict[str, Any],
    owner_request_path: Path,
    *,
    closeout_receipt: dict[str, Any] | None = None,
    closeout_receipt_path: Path | None = None,
) -> dict[str, Any]:
    rollups = _dict(matrix.get("rollups"))
    receipt = _dict(closeout_receipt)
    receipt_result = _dict(receipt.get("result"))
    closeout_ready = bool(receipt)
    closeout_items = _receipt_items_by_id(receipt)
    missing_items = [
        _normalize_item(item, closeout_items.get(str(_dict(item).get("item") or _dict(item).get("item_id") or "")))
        for item in _list(owner_request.get("missing_items"))
    ]
    blocked_items = [
        item["item_id"]
        for item in missing_items
        if item.get("status") != "verified"
    ]
    effective_missing = (
        0
        if closeout_ready
        else _int(rollups.get("credential_escrow_effective_missing_count"))
    )
    forbidden_values = _strings(owner_request.get("forbidden_values"))
    status = str(
        rollups.get("credential_escrow_intake_status")
        or "blocked_waiting_non_secret_credential_escrow_evidence"
    )
    if closeout_ready:
        status = "closed_credential_escrow_evidence_refs_controlled_closeout"
    elif effective_missing == 0 and not blocked_items:
        status = "ready_for_escrow_marker_review"
    safe_next_step = (
        str(receipt.get("safe_next_step") or "")
        if closeout_ready
        else "collect_redacted_non_secret_evidence_refs_then_rerun_preflight"
    )
    if not safe_next_step:
        safe_next_step = "continue_to_p0_006_source_to_runtime_drift_cleanup"
    rollup_readback = {
        "required_item_count": len(missing_items),
        "missing_item_count": len(blocked_items),
        "blocked_item_ids": blocked_items,
        "effective_escrow_missing_count": effective_missing,
        "accepted_item_count": _int(
            receipt_result.get("accepted_item_count")
        ),
        "owner_response_received_count": (
            _int(receipt_result.get("owner_response_received_count"))
            if closeout_ready
            else _int(rollups.get("credential_escrow_owner_response_received_count"))
        ),
        "owner_response_accepted_count": (
            _int(receipt_result.get("owner_response_accepted_count"))
            if closeout_ready
            else _int(rollups.get("credential_escrow_owner_response_accepted_count"))
        ),
        "runtime_gate_count": _int(rollups.get("credential_escrow_runtime_gate_count")),
        "credential_marker_write_authorized_count": _int(
            rollups.get("credential_marker_write_authorized_count")
        ),
        "secret_value_collection_allowed": bool(
            rollups.get("credential_escrow_secret_value_collection_allowed")
        ),
        "forbidden_true_field_count": _int(
            rollups.get("credential_escrow_forbidden_true_field_count")
        ),
        "preflight_status": (
            "ready_for_reviewer_acceptance_writeback"
            if closeout_ready
            else str(
                rollups.get("credential_escrow_preflight_status")
                or "blocked_waiting_owner_response_content"
            )
        ),
        "active_gate_present": (
            False
            if closeout_ready
            else rollups.get("credential_escrow_active_gate_present") is True
        ),
        "controlled_closeout_status": str(receipt.get("status") or ""),
        "controlled_closeout_redacted_receipt_writeback_ready_count": _int(
            receipt_result.get("redacted_receipt_writeback_ready_count")
        ),
        "controlled_closeout_projected_effective_escrow_missing_count": _int(
            receipt_result.get("projected_effective_escrow_missing_count")
        ),
        "forbidden_value_count": len(forbidden_values),
        "single_preflight_intake_ready_count": 1,
    }

    payload = {
        "schema_version": _SCHEMA_VERSION,
        "generated_at": owner_request.get("generated_at"),
        "priority": "P0-005",
        "scope": "credential_escrow_evidence_intake",
        "status": status,
        "safe_next_step": safe_next_step,
        **rollup_readback,
        "readback": {
            "workplan_id": "P0-005",
            "workplan_title": "產品資料與備份 contract",
            "source_owner_request_ref": (
                f"docs/security/{owner_request_path.name}"
            ),
            "source_backup_dr_readiness_matrix_ref": (
                "docs/evaluations/backup_dr_readiness_matrix_2026-06-04.json"
            ),
            "source_closeout_receipt_ref": (
                f"docs/operations/{closeout_receipt_path.name}"
                if closeout_receipt_path
                else ""
            ),
            "single_preflight_intake_schema_version": (
                _SINGLE_PREFLIGHT_INTAKE_SCHEMA_VERSION
            ),
            "safe_next_step": safe_next_step,
        },
        "required_evidence_items": missing_items,
        "controlled_closeout_receipt": receipt,
        "reviewer_readiness": _build_reviewer_readiness(
            closeout_ready=closeout_ready,
            accepted_item_ids=[item["item_id"] for item in missing_items if item.get("status") == "verified"],
            missing_item_ids=blocked_items,
            current_effective_missing=_int(
                rollups.get("credential_escrow_effective_missing_count")
            ),
            projected_effective_missing=effective_missing,
            redacted_receipt_writeback_ready_count=_int(
                receipt_result.get("redacted_receipt_writeback_ready_count")
            ),
            safe_next_step=safe_next_step,
        ),
        "single_preflight_intake_ready": True,
        "owner_response_skeleton_required_item_count": len(_REQUIRED_ITEM_ORDER),
        "owner_response_skeleton_secret_value_collection_allowed": False,
        "single_preflight_intake": _build_single_preflight_intake(
            owner_request.get("generated_at"),
            closeout_ready=closeout_ready,
        ),
        "forbidden_values": forbidden_values,
        "next_actions": (
            [
                "keep_credential_marker_write_closed_until_marker_dry_run_gate",
                "continue_to_p0_006_source_to_runtime_drift_cleanup",
            ]
            if closeout_ready
            else [
                "collect_redacted_non_secret_evidence_refs_for_all_missing_items",
                "rerun_owner_response_preflight_without_secret_values",
                "keep_credential_marker_write_closed_until_preflight_accepts_all_items",
            ]
        ),
        "blocked_operations": [
            "credential_marker_write",
            "credential_read",
            "secret_plaintext_export",
            "restore_execution",
            "offsite_sync_execution",
            "backup_execution",
            "workflow_dispatch",
            "host_or_k8s_write",
        ],
        "rollups": rollup_readback,
        "operation_boundaries": {
            "read_only_api_allowed": True,
            "backup_execution_allowed": False,
            "restore_execution_allowed": False,
            "offsite_sync_execution_allowed": False,
            "credential_marker_write_allowed": False,
            "credential_read_allowed": False,
            "secret_plaintext_allowed": False,
            "secret_value_collection_allowed": False,
            "workflow_trigger_allowed": False,
            "host_or_k8s_write_allowed": False,
            "raw_session_or_sqlite_read_allowed": False,
        },
    }
    return payload


def _build_single_preflight_intake(
    generated_at: Any,
    *,
    closeout_ready: bool = False,
) -> dict[str, Any]:
    return {
        "schema_version": _SINGLE_PREFLIGHT_INTAKE_SCHEMA_VERSION,
        "generated_at": generated_at,
        "workplan_id": "P0-005",
        "status": (
            "completed_redacted_non_secret_evidence_refs_receipt_ready"
            if closeout_ready
            else "waiting_for_five_redacted_non_secret_evidence_refs"
        ),
        "active_gate": _GATE_ID,
        "required_item_count": len(_REQUIRED_ITEM_ORDER),
        "required_item_ids": list(_REQUIRED_ITEM_ORDER),
        "required_items": [
            _build_single_preflight_evidence_item(item_id)
            for item_id in _REQUIRED_ITEM_ORDER
        ],
        "owner_packet": _build_single_preflight_owner_packet(),
        "owner_response_skeleton": _build_single_preflight_owner_response_skeleton(
            generated_at,
        ),
        "evidence_refs_validation_endpoint": (
            "/api/v1/agents/credential-escrow-evidence-intake-readiness/"
            "validate-evidence-refs"
        ),
        "evidence_refs_validation_mode": (
            "validate_redacted_non_secret_refs_only_no_persist_no_marker_write"
        ),
        "single_preflight_command": (
            "python3 scripts/reboot-recovery/post-reboot-owner-response-preflight.py "
            "--owner-packet-file <owner-packet.json> "
            "--response-file <filled-owner-response.json> --json --no-color"
        ),
        "scorecard_command": (
            "python3 scripts/reboot-recovery/post-reboot-credential-escrow-intake-scorecard.py "
            "--summary-file <summary.txt> --owner-packet-file <owner-packet.json> "
            "--response-file <filled-owner-response.json> "
            "--offsite-report-file <offsite-report.txt> "
            "--escrow-status-file <escrow-status.txt> --json --no-color"
        ),
        "exit_criteria": [
            "preflight_status=ready_for_independent_reviewer_acceptance",
            "owner_response_received_count=1",
            "owner_response_accepted_count=1",
            "forbidden_true_field_count=0",
            "projected_effective_escrow_missing_count=0",
        ],
        "operation_boundaries": {
            "payload_persisted": False,
            "backup_execution_performed": False,
            "restore_execution_performed": False,
            "offsite_sync_execution_performed": False,
            "credential_marker_write_performed": False,
            "credential_read_performed": False,
            "secret_plaintext_read": False,
            "secret_value_collection_allowed": False,
            "workflow_trigger_performed": False,
            "host_or_k8s_write_performed": False,
            "raw_session_or_sqlite_read_performed": False,
            "runtime_action_performed": False,
        },
        "execution_rules": [
            "Use this as the only P0-005 intake packet.",
            "Do not create per-item owner packets.",
            "Do not include secret values in evidence refs.",
            "Do not reopen cold-start or CI/CD while this checklist is pending.",
        ],
    }


def _build_single_preflight_evidence_item(item_id: str) -> dict[str, Any]:
    return {
        "item_id": item_id,
        "required_fields": [
            "non_secret_evidence_ref",
            "recovery_owner",
            "reviewer",
            "last_reviewed_at",
            "contains_secret_value=false",
        ],
        "accepted_ref_classes": [
            "password_manager_item_id",
            "sealed_envelope_id",
            "recovery_checklist_id",
            "review_ticket_id",
        ],
        "rejected_values": [
            "passwords",
            "tokens",
            "private keys",
            "recovery codes",
            "secret URLs",
            "session cookies",
        ],
        "marker_dry_run_command": (
            "/backup/scripts/mark-credential-escrow-verified.sh "
            f"--item {item_id} --evidence-id <NON_SECRET_REF_FOR_{item_id}> --dry-run"
        ),
    }


def _build_single_preflight_owner_packet() -> dict[str, Any]:
    return {
        "schema_version": _OWNER_PACKET_SCHEMA,
        "source": {"next_required_gates": [_GATE_ID]},
        "owner_packets": [
            {
                "packet_id": _GATE_ID,
                "title": "P0-005 DR credential escrow evidence",
                "priority": "P0",
                "required_items": list(_REQUIRED_ITEM_ORDER),
            }
        ],
    }


def _build_single_preflight_owner_response_skeleton(
    generated_at: Any,
) -> dict[str, Any]:
    return {
        "schema_version": _OWNER_RESPONSE_SCHEMA,
        "generated_at": generated_at,
        "responses": [
            {
                "gate_id": _GATE_ID,
                "owner_role": "owner_role_here",
                "owner_team": "owner_team_here",
                "decision": "pending",
                "decision_reason": "decision_reason_here",
                "affected_scope": "P0-005 DR credential escrow evidence",
                "redacted_evidence_refs": ["redacted_evidence_ref_here"],
                "followup_owner": "followup_owner_here",
                "runtime_action_requested": False,
                "runtime_action_authorized": False,
                "host_write_requested": False,
                "host_write_authorized": False,
                "secret_value_included": False,
                "secret_value_collection_allowed": False,
                "credential_marker_write_requested": False,
                "credential_marker_write_authorized": False,
                "escrow_items": [
                    {
                        "item_id": item_id,
                        "non_secret_evidence_ref": "non_secret_evidence_ref_here",
                        "recovery_owner": "owner_role_here",
                        "reviewer": "reviewer_here",
                        "last_reviewed_at": "pending",
                        "contains_secret_value": False,
                    }
                    for item_id in _REQUIRED_ITEM_ORDER
                ],
            }
        ],
    }


def _owner_response_from_evidence_refs(payload: dict[str, Any]) -> dict[str, Any]:
    submissions = _normalized_evidence_ref_submissions(payload)
    defaults = _dict(payload.get("defaults"))
    owner_role = _field(payload, defaults, "owner_role", "owner_role_here")
    owner_team = _field(payload, defaults, "owner_team", "owner_team_here")
    reviewer = _field(payload, defaults, "reviewer", "reviewer_here")
    last_reviewed_at = _field(payload, defaults, "last_reviewed_at", "pending")
    followup_owner = _field(payload, defaults, "followup_owner", owner_role)
    recovery_owner = _field(payload, defaults, "recovery_owner", owner_role)
    evidence_refs = _strings(payload.get("redacted_evidence_refs"))
    if not evidence_refs:
        evidence_refs = [
            str(item.get("non_secret_evidence_ref") or "").strip()
            for item in submissions
            if not _is_placeholder(item.get("non_secret_evidence_ref"))
        ]
    item_by_id = {
        str(item.get("item_id") or "").strip(): item
        for item in submissions
        if str(item.get("item_id") or "").strip()
    }
    return {
        "schema_version": _OWNER_RESPONSE_SCHEMA,
        "generated_at": payload.get("generated_at"),
        "responses": [
            {
                "gate_id": _GATE_ID,
                "owner_role": owner_role,
                "owner_team": owner_team,
                "decision": str(payload.get("decision") or "accepted"),
                "decision_reason": str(
                    payload.get("decision_reason")
                    or "validated redacted non-secret evidence refs only"
                ),
                "affected_scope": str(
                    payload.get("affected_scope")
                    or "P0-005 DR credential escrow evidence"
                ),
                "redacted_evidence_refs": evidence_refs,
                "followup_owner": followup_owner,
                "runtime_action_requested": False,
                "runtime_action_authorized": False,
                "host_write_requested": False,
                "host_write_authorized": False,
                "secret_value_included": False,
                "secret_value_collection_allowed": False,
                "credential_marker_write_requested": False,
                "credential_marker_write_authorized": False,
                "escrow_items": [
                    _owner_response_escrow_item(
                        item_id,
                        _dict(item_by_id.get(item_id)),
                        recovery_owner,
                        reviewer,
                        last_reviewed_at,
                    )
                    for item_id in _REQUIRED_ITEM_ORDER
                ],
            }
        ],
    }


def _owner_response_escrow_item(
    item_id: str,
    submission: dict[str, Any],
    recovery_owner: str,
    reviewer: str,
    last_reviewed_at: str,
) -> dict[str, Any]:
    return {
        "item_id": item_id,
        "non_secret_evidence_ref": str(
            submission.get("non_secret_evidence_ref") or "non_secret_evidence_ref_here"
        ),
        "recovery_owner": str(submission.get("recovery_owner") or recovery_owner),
        "reviewer": str(submission.get("reviewer") or reviewer),
        "last_reviewed_at": str(
            submission.get("last_reviewed_at") or last_reviewed_at
        ),
        "contains_secret_value": submission.get("contains_secret_value") is True,
    }


def _normalized_evidence_ref_submissions(payload: dict[str, Any]) -> list[dict[str, Any]]:
    raw = payload.get("evidence_refs")
    if raw is None:
        raw = payload.get("escrow_items")
    if isinstance(raw, dict):
        return [
            {"item_id": str(item_id), "non_secret_evidence_ref": str(ref)}
            for item_id, ref in raw.items()
        ]
    return [
        _dict(item)
        for item in _list(raw)
        if isinstance(item, dict)
    ]


def _field(
    payload: dict[str, Any],
    defaults: dict[str, Any],
    key: str,
    fallback: str,
) -> str:
    return str(payload.get(key) or defaults.get(key) or fallback)


def _load_closeout_receipt(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    with path.open(encoding="utf-8") as handle:
        receipt = json.load(handle)
    if not isinstance(receipt, dict):
        raise ValueError(f"{path}: expected JSON object")
    _require_closeout_receipt(receipt, str(path))
    return receipt


def _receipt_items_by_id(receipt: dict[str, Any]) -> dict[str, dict[str, Any]]:
    return {
        str(item.get("item_id") or ""): _dict(item)
        for item in _list(receipt.get("evidence_refs"))
        if isinstance(item, dict)
    }


def _build_reviewer_readiness(
    *,
    closeout_ready: bool,
    accepted_item_ids: list[str],
    missing_item_ids: list[str],
    current_effective_missing: int,
    projected_effective_missing: int,
    redacted_receipt_writeback_ready_count: int,
    safe_next_step: str,
) -> dict[str, Any]:
    return {
        "schema_version": "credential_escrow_reviewer_readiness_v1",
        "status": (
            "ready_for_reviewer_acceptance_writeback"
            if closeout_ready
            else "not_ready_for_reviewer_acceptance_writeback"
        ),
        "accepted_required_item_ids": sorted(set(accepted_item_ids)),
        "missing_required_item_ids": missing_item_ids,
        "current_effective_escrow_missing_count": current_effective_missing,
        "projected_effective_escrow_missing_count": projected_effective_missing,
        "redacted_receipt_writeback_ready_count": redacted_receipt_writeback_ready_count,
        "credential_marker_write_authorized_count": 0,
        "runtime_gate_count": 0,
        "secret_value_collection_allowed": False,
        "payload_persisted": False,
        "safe_next_step": safe_next_step,
        "blocked_operations": [
            "credential_marker_write",
            "credential_read",
            "secret_plaintext_export",
            "backup_execution",
            "restore_execution",
            "offsite_sync_execution",
            "workflow_dispatch",
            "host_or_k8s_write",
        ],
    }


def _normalize_item(
    item: Any,
    closeout_item: dict[str, Any] | None = None,
) -> dict[str, Any]:
    data = _dict(item)
    receipt_item = _dict(closeout_item)
    verified = bool(receipt_item)
    return {
        "item_id": str(data.get("item") or data.get("item_id") or ""),
        "status": "verified" if verified else str(data.get("status") or "missing"),
        "allowed_evidence_id_types": _strings(data.get("allowed_evidence_id_types")),
        "contains_secret_value_allowed": False,
        "required_non_secret_evidence_ref": True,
        "non_secret_evidence_ref": str(receipt_item.get("non_secret_evidence_ref") or ""),
        "recovery_owner": str(receipt_item.get("recovery_owner") or ""),
        "reviewer": str(receipt_item.get("reviewer") or ""),
        "last_reviewed_at": str(receipt_item.get("last_reviewed_at") or ""),
        "controlled_closeout_receipt_accepted": verified,
    }


def _require_owner_request(payload: dict[str, Any], label: str) -> None:
    if payload.get("scope") != _OWNER_REQUEST_SCOPE:
        raise ValueError(f"{label}: scope must be {_OWNER_REQUEST_SCOPE}")
    if payload.get("gates", {}).get("secret_value_collection_authorized") is not False:
        raise ValueError(f"{label}: secret value collection must remain false")
    if payload.get("gates", {}).get("marker_write_completed") is not False:
        raise ValueError(f"{label}: marker_write_completed must remain false")
    if not _list(payload.get("missing_items")):
        raise ValueError(f"{label}: missing_items must be present")


def _require_backup_rollups(payload: dict[str, Any], label: str) -> None:
    rollups = _dict(payload.get("rollups"))
    if rollups.get("credential_escrow_secret_value_collection_allowed") is not False:
        raise ValueError(f"{label}: secret value collection must remain false")
    if _int(rollups.get("credential_marker_write_authorized_count")) != 0:
        raise ValueError(f"{label}: credential marker write must remain closed")


def _require_operation_boundaries(payload: dict[str, Any], label: str) -> None:
    boundaries = _dict(payload.get("operation_boundaries"))
    if boundaries.get("read_only_api_allowed") is not True:
        raise ValueError(f"{label}: read_only_api_allowed must be true")
    blocked_flags = {
        "backup_execution_allowed",
        "restore_execution_allowed",
        "offsite_sync_execution_allowed",
        "credential_marker_write_allowed",
        "credential_read_allowed",
        "secret_plaintext_allowed",
        "secret_value_collection_allowed",
        "workflow_trigger_allowed",
        "host_or_k8s_write_allowed",
        "raw_session_or_sqlite_read_allowed",
    }
    allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
    if allowed:
        raise ValueError(f"{label}: operation boundaries must remain false: {allowed}")


def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
    items = _list(payload.get("required_evidence_items"))
    blocked = [
        str(_dict(item).get("item_id") or "")
        for item in items
        if _dict(item).get("status") != "verified"
    ]
    rollups = _dict(payload.get("rollups"))
    if rollups.get("required_item_count") != len(items):
        raise ValueError(f"{label}: required_item_count mismatch")
    if rollups.get("missing_item_count") != len(blocked):
        raise ValueError(f"{label}: missing_item_count mismatch")
    if rollups.get("blocked_item_ids") != blocked:
        raise ValueError(f"{label}: blocked_item_ids mismatch")
    if rollups.get("secret_value_collection_allowed") is not False:
        raise ValueError(f"{label}: secret_value_collection_allowed must be false")
    if rollups.get("credential_marker_write_authorized_count") != 0:
        raise ValueError(f"{label}: marker write must remain closed")


def _require_single_preflight_intake(payload: dict[str, Any], label: str) -> None:
    intake = _dict(payload.get("single_preflight_intake"))
    if intake.get("schema_version") != _SINGLE_PREFLIGHT_INTAKE_SCHEMA_VERSION:
        raise ValueError(f"{label}: single preflight intake schema mismatch")
    if intake.get("required_item_count") != len(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: single preflight intake item count mismatch")
    if intake.get("required_item_ids") != list(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: single preflight intake item order mismatch")

    skeleton = _dict(intake.get("owner_response_skeleton"))
    responses = _list(skeleton.get("responses"))
    response = _dict(responses[0]) if responses else {}
    escrow_items = [_dict(item) for item in _list(response.get("escrow_items"))]
    if len(escrow_items) != len(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: owner response skeleton item count mismatch")
    if [item.get("item_id") for item in escrow_items] != list(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: owner response skeleton item order mismatch")
    if response.get("secret_value_collection_allowed") is not False:
        raise ValueError(f"{label}: skeleton secret collection must remain false")
    if response.get("credential_marker_write_authorized") is not False:
        raise ValueError(f"{label}: skeleton marker write must remain false")
    for item in escrow_items:
        if item.get("contains_secret_value") is not False:
            raise ValueError(f"{label}: skeleton escrow item secret flag must be false")

    boundaries = _dict(intake.get("operation_boundaries"))
    blocked_flags = {
        "payload_persisted",
        "backup_execution_performed",
        "restore_execution_performed",
        "offsite_sync_execution_performed",
        "credential_marker_write_performed",
        "credential_read_performed",
        "secret_plaintext_read",
        "secret_value_collection_allowed",
        "workflow_trigger_performed",
        "host_or_k8s_write_performed",
        "raw_session_or_sqlite_read_performed",
        "runtime_action_performed",
    }
    open_flags = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
    if open_flags:
        raise ValueError(
            f"{label}: single preflight intake boundaries opened: {open_flags}"
        )


def _require_payload_closeout_consistency(payload: dict[str, Any], label: str) -> None:
    receipt = _dict(payload.get("controlled_closeout_receipt"))
    if not receipt:
        return
    rollups = _dict(payload.get("rollups"))
    reviewer = _dict(payload.get("reviewer_readiness"))
    if payload.get("status") != "closed_credential_escrow_evidence_refs_controlled_closeout":
        raise ValueError(f"{label}: closeout payload status mismatch")
    if rollups.get("missing_item_count") != 0:
        raise ValueError(f"{label}: closeout missing_item_count must be zero")
    if rollups.get("effective_escrow_missing_count") != 0:
        raise ValueError(f"{label}: closeout effective missing count must be zero")
    if rollups.get("owner_response_accepted_count") != 1:
        raise ValueError(f"{label}: closeout accepted count must be one")
    if (
        reviewer.get("status")
        != "ready_for_reviewer_acceptance_writeback"
    ):
        raise ValueError(f"{label}: closeout reviewer readiness mismatch")
    if reviewer.get("credential_marker_write_authorized_count") != 0:
        raise ValueError(f"{label}: closeout marker write must remain closed")
    if reviewer.get("secret_value_collection_allowed") is not False:
        raise ValueError(f"{label}: closeout secret collection must remain false")


def _require_closeout_receipt(receipt: dict[str, Any], label: str) -> None:
    if receipt.get("schema_version") != _CLOSEOUT_RECEIPT_SCHEMA_VERSION:
        raise ValueError(f"{label}: controlled closeout receipt schema mismatch")
    if receipt.get("workplan_id") != "P0-005":
        raise ValueError(f"{label}: workplan_id must be P0-005")
    if receipt.get("status") != "ready_for_p0_005_controlled_closeout":
        raise ValueError(f"{label}: controlled closeout receipt status mismatch")
    sensitive_hits = _find_sensitive_strings(receipt)
    if sensitive_hits:
        raise ValueError(f"{label}: sensitive strings present: {sensitive_hits}")
    forbidden_hits = _find_forbidden_booleans(receipt)
    if forbidden_hits:
        raise ValueError(f"{label}: forbidden boolean fields true: {forbidden_hits}")

    result = _dict(receipt.get("result"))
    if _int(result.get("required_item_count")) != len(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: required item count mismatch")
    if _int(result.get("accepted_item_count")) != len(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: accepted item count mismatch")
    if _int(result.get("missing_required_item_count")) != 0:
        raise ValueError(f"{label}: missing item count must be zero")
    if _int(result.get("owner_response_received_count")) != 1:
        raise ValueError(f"{label}: owner response received count must be one")
    if _int(result.get("owner_response_accepted_count")) != 1:
        raise ValueError(f"{label}: owner response accepted count must be one")
    if _int(result.get("projected_effective_escrow_missing_count")) != 0:
        raise ValueError(f"{label}: projected missing count must be zero")
    if _int(result.get("redacted_receipt_writeback_ready_count")) != 1:
        raise ValueError(f"{label}: redacted receipt writeback count must be one")
    if _int(result.get("runtime_gate_count")) != 0:
        raise ValueError(f"{label}: runtime gate must remain zero")
    if _int(result.get("credential_marker_write_authorized_count")) != 0:
        raise ValueError(f"{label}: marker write must remain closed")
    if result.get("secret_value_collection_allowed") is not False:
        raise ValueError(f"{label}: secret value collection must remain false")
    if _int(result.get("forbidden_true_field_count")) != 0:
        raise ValueError(f"{label}: forbidden true field count must be zero")

    items = [_dict(item) for item in _list(receipt.get("evidence_refs"))]
    item_ids = [str(item.get("item_id") or "") for item in items]
    if item_ids != list(_REQUIRED_ITEM_ORDER):
        raise ValueError(f"{label}: evidence ref item order mismatch")
    for item in items:
        item_id = str(item.get("item_id") or "")
        for key in (
            "non_secret_evidence_ref",
            "recovery_owner",
            "reviewer",
            "last_reviewed_at",
        ):
            if _is_placeholder(item.get(key)):
                raise ValueError(f"{label}: {item_id}.{key} missing")
        if item.get("contains_secret_value") is not False:
            raise ValueError(f"{label}: {item_id}.contains_secret_value must be false")

    writeback = _dict(receipt.get("writeback"))
    if writeback.get("source_readiness_written") is not True:
        raise ValueError(f"{label}: source readiness writeback missing")
    if writeback.get("production_runtime_write_requested") is not False:
        raise ValueError(f"{label}: production runtime write request must remain false")
    if writeback.get("production_runtime_write_performed") is not False:
        raise ValueError(f"{label}: production runtime write must remain false")

    boundaries = _dict(receipt.get("operation_boundaries"))
    if boundaries.get("source_readiness_written") is not True:
        raise ValueError(f"{label}: source readiness boundary missing")
    blocked_flags = {
        "payload_persisted",
        "backup_execution_performed",
        "restore_execution_performed",
        "offsite_sync_execution_performed",
        "credential_marker_write_performed",
        "credential_read_performed",
        "secret_plaintext_read",
        "secret_value_collection_allowed",
        "workflow_trigger_performed",
        "host_or_k8s_write_performed",
        "raw_session_or_sqlite_read_performed",
        "runtime_action_performed",
    }
    open_flags = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
    if open_flags:
        raise ValueError(f"{label}: closeout boundaries opened: {open_flags}")


def _dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def _list(value: Any) -> list[Any]:
    return value if isinstance(value, list) else []


def _strings(value: Any) -> list[str]:
    return [str(item) for item in _list(value) if str(item)]


def _int(value: Any) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _normalized(value: Any) -> str:
    return "" if value is None else str(value).strip()


def _is_placeholder(value: Any) -> bool:
    text = _normalized(value)
    lower = text.lower()
    if lower in _PLACEHOLDER_VALUES:
        return True
    if re.fullmatch(r"<[^<>]+>", text):
        return True
    if lower in {"yyyy-mm-dd", "yyyy/mm/dd"}:
        return True
    return lower.startswith(
        (
            "vault-item-id-for-",
            "sealed-envelope-id-for-",
            "recovery-checklist-id-for-",
            "ticket-id-for-",
        )
    )


def _collect_strings(value: Any, path: str = "$") -> list[tuple[str, str]]:
    strings: list[tuple[str, str]] = []
    if isinstance(value, str):
        strings.append((path, value))
    elif isinstance(value, dict):
        for key, child in value.items():
            strings.extend(_collect_strings(child, f"{path}.{key}"))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            strings.extend(_collect_strings(child, f"{path}[{index}]"))
    return strings


def _find_sensitive_strings(payload: dict[str, Any]) -> list[str]:
    hits: list[str] = []
    for path, value in _collect_strings(payload):
        if path.endswith(".item_id") and value in _REQUIRED_ITEM_IDS:
            continue
        if path.endswith(".gate_id") and value == _GATE_ID:
            continue
        for label, pattern in _SECRET_VALUE_PATTERNS.items():
            if pattern.search(value):
                hits.append(f"{label}_at={path}")
                break
    return hits


def _find_forbidden_booleans(value: Any, path: str = "$") -> list[str]:
    hits: list[str] = []
    if isinstance(value, dict):
        for key, child in value.items():
            child_path = f"{path}.{key}"
            if key in _FORBIDDEN_BOOLEAN_FIELDS and child is not False:
                hits.append(f"{child_path}={child!r}")
            hits.extend(_find_forbidden_booleans(child, child_path))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            hits.extend(_find_forbidden_booleans(child, f"{path}[{index}]"))
    return hits
