"""Governed dynamic-catalog to schema-1 authenticated queue producer.

The command is intentionally separate from the Native Messaging host.  It
does not open Chrome, call the host, read credentials, or accept caller-owned
job records.  Its only mutation is an append to the configured queue through
``QueueStore.append_ingress_jobs`` and that method's shared first-byte lock.
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import os
import re
import stat
import sys
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path, PurePosixPath
from typing import Any

from .constants import (
    EXTENSION_BUILD,
    EXPECTED_EXTENSION_ID,
    EXPECTED_ORIGIN,
    HOST_BUILD,
    MESSAGE_ID_RE,
    RELOAD_GENERATION,
    canonical_url,
    stable_job_id,
    validate_bvid,
    validate_creator_uid,
)
from .protocol import ProtocolError, strict_json_loads
from .queue_state import (
    AuthorizedSuccessor,
    QueueStore,
    ReleaseApproval,
    validate_ingress_record,
)


PRODUCER_ID = "project-info-bili-auth-queue-producer/2"
_LEGACY_PRODUCER_ID = "project-info-bili-auth-queue-producer/1"
_TASK_ID = "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001"
_PROJECT_ID = "project-info"
_HEX64 = re.compile(r"[0-9A-Fa-f]{64}\Z")
_CONTROL = re.compile(r"[\x00-\x1f\x7f]")
_FORBIDDEN_KEYS = re.compile(r"(?:cookie|password|passwd|access_token|refresh_token|signed_url|local_?storage|captcha)", re.I)
_HOST_KEYS = {
    "schema", "creator_allowlist", "queue_path", "queue_state_path", "queue_lock_path",
    "reload_state_path", "reload_generation", "required_extension_build", "ffmpeg",
    "ffmpeg_sha256", "ffprobe", "ffprobe_sha256", "bridge_python",
    "bridge_python_sha256", "bridge_script", "bridge_script_sha256",
    "yt_dlp_executable", "yt_dlp_executable_sha256", "destination",
    "creator_name", "formal_manifest_path", "processing_handoff_path",
}
_CONFIG_V1_KEYS = {
    "schema", "producer_id", "project_root", "host_config_path", "host_config_sha256",
    "creator_name", "creator_uid", "dynamic_manifest_path", "registered_catalog_path",
    "registered_catalog_sha256", "bvid_allowlist",
}
_CONFIG_V2_KEYS = _CONFIG_V1_KEYS | {"successor_authorization_message_id"}
_REGISTERED_KEYS = {"schema", "source", "creator_name", "creator_uid", "items"}
_INGRESS_KEYS = {
    "schema", "bvid", "creator_uid", "expected_duration_ms",
    "discovered_at_unix_ms", "published_at", "title",
}
_TERMINAL_VIDEO_STATUSES = frozenset({
    "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT",
    "VIDEO_DOWNLOADED_COMPLETE",
    "VIDEO_COMPLETE",
})
_SOURCE_MANIFEST_KEYS = {
    "schema", "scope", "extension_id", "extension_build", "host_build",
    "archive_metadata_contract", "dependency_artifact_manifest_bytes",
    "dependency_artifact_manifest_sha256", "files",
}
_ARCHIVE_METADATA_KEYS = {
    "schema", "root", "relative_files", "distribution_name", "distribution_version",
    "allowed_type_codes", "source_date_epoch", "tree_hash_algorithm",
    "canonical_tree_sha256",
}
_ARTIFACT_FILE_KEYS = {"path", "bytes", "sha256"}
_BUILD_APPROVAL_KEYS = {
    "schema", "task_id", "approval_scope", "approved_by_role", "status",
    "source_artifact_manifest_bytes", "source_artifact_manifest_sha256",
}
_BUILD_RECEIPT_KEYS = {
    "schema", "scope", "extension_id", "extension_build", "host_build", "packaging",
    "pyinstaller_version", "yt_dlp_version", "builder_python_sha256",
    "pyinstaller_executable_bytes", "pyinstaller_executable_sha256",
    "builder_provision_receipt_bytes", "builder_provision_receipt_sha256",
    "build_script_sha256", "source_artifact_manifest_bytes",
    "source_artifact_manifest_sha256", "dependency_artifact_manifest_bytes",
    "dependency_artifact_manifest_sha256", "yt_dlp_wheel_sha256",
    "archive_verification", "files",
}
_ARCHIVE_VERIFICATION_KEYS = {
    "status", "method", "required_modules", "metadata_entry", "metadata_files",
    "metadata_type_codes", "metadata_tree_sha256",
}
_INSTALL_APPROVAL_KEYS = {
    "schema", "task_id", "approval_scope", "approved_by_role", "status",
    "source_artifact_manifest_sha256", "build_artifact_manifest_bytes",
    "build_artifact_manifest_sha256", "host_executable_bytes", "host_executable_sha256",
}
_INSTALL_RECEIPT_KEYS = {
    "schema", "task_id", "host_build", "required_extension_build", "extension_id",
    "host_name", "installed_root", "installed_files",
}
_FAST_PATH_INSTALL_RECEIPT_KEYS = _INSTALL_RECEIPT_KEYS | {
    "validation_scope", "validated_by_role", "status",
    "continuous_authorization_handoff_id", "owner_ai_id", "owner_thread_id",
    "owner_role_instance_id", "authorization_file_relative_path",
    "authorization_file_bytes", "authorization_file_sha256", "successor_scope_sha256",
    "implementation_review_audit_id", "implementation_review_audit_bytes",
    "implementation_review_audit_sha256", "source_artifact_manifest_bytes",
    "source_artifact_manifest_sha256", "source_receipt_bytes", "source_receipt_sha256",
    "build_artifact_manifest_bytes", "build_artifact_manifest_sha256",
    "build_receipt_bytes", "build_receipt_sha256", "host_executable_bytes",
    "host_executable_sha256", "native_messaging_host_manifest",
    "host_build_source_manifest", "host_source_binding_mode", "producer_only_changed_files",
    "host_archive_excluded_modules",
    "projection_contract_sha256", "projection_tree_sha256", "reload_state_path",
    "reload_state_bytes", "reload_state_lines", "reload_state_sha256", "queue_path",
    "queue_prefix_bytes", "queue_prefix_lines", "queue_prefix_sha256",
    "queue_state_path", "queue_state_prefix_bytes", "queue_state_prefix_lines",
    "queue_state_prefix_sha256", "formal_manifest_path", "formal_manifest_prefix_bytes",
    "formal_manifest_prefix_lines", "formal_manifest_prefix_sha256", "secret_field_count",
}
_CONTINUOUS_FAST_PATH_HANDOFF_ID = (
    "HANDOFF-INFOADMIN-INFODEV2-BILI-AUTH-V010-GENERATION9-PASS0-"
    "CONTINUOUS-TARGET-FAST-PATH-RELEASE-RUNTIME-20260820-001"
)
_FAST_PATH_OWNER = (
    "infodev-2",
    "019fbcbb-bed7-7c90-83ab-f50610f80d3a",
    "dev.developer.project.secondary",
)


def _is_reparse(stat_result: os.stat_result) -> bool:
    attributes = int(getattr(stat_result, "st_file_attributes", 0))
    reparse = int(getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))
    return stat.S_ISLNK(stat_result.st_mode) or bool(attributes & reparse)


def _absolute_local_path(value: Any, code: str) -> Path:
    if not isinstance(value, str) or not value or _CONTROL.search(value):
        raise ProtocolError(code)
    path = Path(value)
    if not path.is_absolute() or str(path).startswith("\\\\"):
        raise ProtocolError(code)
    return Path(os.path.abspath(path))


def _check_existing_chain(path: Path, *, require_file: bool | None) -> None:
    current = Path(path.anchor)
    for part in path.parts[1:]:
        current = current / part
        try:
            info = os.lstat(current)
        except FileNotFoundError:
            if require_file:
                raise ProtocolError("E_CONFIG")
            return
        if _is_reparse(info):
            raise ProtocolError("E_CONFIG")
    if require_file is True and not path.is_file():
        raise ProtocolError("E_CONFIG")
    if require_file is False and path.exists() and not path.is_dir():
        raise ProtocolError("E_CONFIG")


def _within(path: Path, root: Path) -> bool:
    try:
        return os.path.commonpath((str(path), str(root))) == str(root)
    except ValueError:
        return False


def _reject_secret_keys(value: Any) -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            if not isinstance(key, str) or _FORBIDDEN_KEYS.search(key):
                raise ProtocolError("E_SECRET_FIELD")
            _reject_secret_keys(child)
    elif isinstance(value, list):
        for child in value:
            _reject_secret_keys(child)


def _safe_text(value: Any, maximum: int = 600) -> str:
    if not isinstance(value, str) or not value.strip() or _CONTROL.search(value):
        raise ProtocolError("E_CATALOG")
    if len(value.encode("utf-8")) > maximum:
        raise ProtocolError("E_CATALOG")
    return value


def _time_ms(value: Any) -> int:
    if not isinstance(value, str) or _CONTROL.search(value):
        raise ProtocolError("E_CATALOG")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ProtocolError("E_CATALOG") from exc
    if parsed.utcoffset() is None:
        raise ProtocolError("E_CATALOG")
    return int(parsed.timestamp() * 1000)


def _duration_ms(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ProtocolError("E_CATALOG")
    try:
        milliseconds = Decimal(str(value)) * 1000
    except InvalidOperation as exc:
        raise ProtocolError("E_CATALOG") from exc
    if milliseconds != milliseconds.to_integral_value():
        raise ProtocolError("E_CATALOG")
    result = int(milliseconds)
    if not 1_000 <= result <= 86_400_000:
        raise ProtocolError("E_CATALOG")
    return result


def _read_strict_payload(path: Path, maximum: int, code: str) -> bytes:
    try:
        _check_existing_chain(path, require_file=True)
        with path.open("rb") as source:
            payload = source.read(maximum + 1)
    except (OSError, ProtocolError) as exc:
        raise ProtocolError(code) from exc
    if len(payload) > maximum:
        raise ProtocolError(code)
    return payload


def _enumerate_exact_tree(root: Path, expected_files: set[str], code: str) -> None:
    """Require the actual no-follow tree to equal an exact case-sensitive file set."""
    try:
        _check_existing_chain(root, require_file=False)
        root_info = os.lstat(root)
    except (OSError, ProtocolError) as exc:
        raise ProtocolError(code) from exc
    if _is_reparse(root_info) or not stat.S_ISDIR(root_info.st_mode):
        raise ProtocolError(code)
    expected_directories = {
        PurePosixPath(path).parent.as_posix()
        for path in expected_files
        if PurePosixPath(path).parent.as_posix() != "."
    }
    actual_files: set[str] = set()
    actual_directories: set[str] = set()
    pending: list[tuple[Path, PurePosixPath]] = [(root, PurePosixPath("."))]
    while pending:
        directory, relative_directory = pending.pop()
        try:
            entries = list(os.scandir(directory))
        except OSError as exc:
            raise ProtocolError(code) from exc
        for entry in entries:
            try:
                info = entry.stat(follow_symlinks=False)
            except OSError as exc:
                raise ProtocolError(code) from exc
            if _is_reparse(info):
                raise ProtocolError(code)
            relative = PurePosixPath(entry.name) if relative_directory == PurePosixPath(".") else relative_directory / entry.name
            normalized = relative.as_posix()
            if stat.S_ISDIR(info.st_mode):
                actual_directories.add(normalized)
                pending.append((Path(entry.path), relative))
            elif stat.S_ISREG(info.st_mode):
                actual_files.add(normalized)
            else:
                raise ProtocolError(code)
    if actual_files != expected_files or actual_directories != expected_directories:
        raise ProtocolError(code)


def _validate_source_tree(root: Path, source_payload: bytes, deployment: dict[str, Any]) -> dict[str, Any]:
    code = "E_DEPLOYMENT_NOT_READY"
    source = _plain_mapping(strict_json_loads(source_payload), _SOURCE_MANIFEST_KEYS, code)
    if (
        source["schema"] != 1 or source["scope"] != "generic-bilibili-queue"
        or source["extension_id"] != EXPECTED_EXTENSION_ID
        or source["extension_build"] != EXTENSION_BUILD or source["host_build"] != HOST_BUILD
    ):
        raise ProtocolError(code)
    archive = _plain_mapping(source["archive_metadata_contract"], _ARCHIVE_METADATA_KEYS, code)
    if (
        archive["schema"] != 1 or archive["root"] != "yt_dlp-2026.7.4.dist-info"
        or archive["relative_files"] != [
            "INSTALLER", "METADATA", "RECORD", "REQUESTED", "WHEEL",
            "entry_points.txt", "licenses/LICENSE",
        ]
        or archive["distribution_name"] != "yt-dlp" or archive["distribution_version"] != "2026.7.4"
        or archive["allowed_type_codes"] != ["b", "x"]
        or archive["source_date_epoch"] != 1786207924
        or archive["tree_hash_algorithm"] != "sha256(path-utf8,nul,decimal-bytes-ascii,nul,payload-sha256-lower-hex-ascii,lf)-upper-hex-v1"
        or archive["canonical_tree_sha256"] != "32F1DC23F6704966AE4511CB173318CD11FCBA031323028D268D9F2488589E70"
    ):
        raise ProtocolError(code)
    dependency_size = _positive_integer(source["dependency_artifact_manifest_bytes"], code)
    dependency_hash = _upper_sha(source["dependency_artifact_manifest_sha256"], code)
    files = source["files"]
    if not isinstance(files, list) or len(files) != 21:
        raise ProtocolError(code)
    source_root = root / "dev" / "project-dev" / "bili_authenticated_extension"
    expected: dict[str, tuple[int, str]] = {}
    for item in files:
        item = _plain_mapping(item, _ARTIFACT_FILE_KEYS, code)
        relative = _relative_path(item["path"], ("",), code)
        size = _positive_integer(item["bytes"], code)
        digest = _upper_sha(item["sha256"], code)
        if relative == "source-artifact-manifest.json" or relative in expected:
            raise ProtocolError(code)
        expected[relative] = (size, digest)
    if list(expected) != sorted(expected):
        raise ProtocolError(code)
    _enumerate_exact_tree(source_root, set(expected) | {"source-artifact-manifest.json"}, code)
    for relative, (size, digest) in expected.items():
        payload = _read_strict_payload(source_root / Path(relative), size + 1, code)
        if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
            raise ProtocolError(code)
    dependency = expected.get("dependencies/dependency-artifact-manifest.json")
    if dependency != (dependency_size, dependency_hash):
        raise ProtocolError(code)
    return source


def _strict_file(path: Path, maximum: int, code: str = "E_CATALOG") -> Any:
    payload = _read_strict_payload(path, maximum, code)
    return strict_json_loads(payload)


def load_producer_configuration(path: Path) -> dict[str, Any]:
    path = Path(os.path.abspath(path))
    _check_existing_chain(path, require_file=True)
    raw = _strict_file(path, 128 * 1024, "E_CONFIG")
    if not isinstance(raw, dict) or raw.get("schema") not in {1, 2}:
        raise ProtocolError("E_CONFIG")
    expected_keys = _CONFIG_V1_KEYS if raw["schema"] == 1 else _CONFIG_V2_KEYS
    if set(raw) != expected_keys or raw.get("producer_id") not in {_LEGACY_PRODUCER_ID, PRODUCER_ID}:
        raise ProtocolError("E_CONFIG")
    if raw["schema"] == 2:
        authorization_message_id = raw["successor_authorization_message_id"]
        if authorization_message_id is not None and (
            not isinstance(authorization_message_id, str)
            or not MESSAGE_ID_RE.fullmatch(authorization_message_id)
        ):
            raise ProtocolError("E_CONFIG")
    _reject_secret_keys(raw)
    project_root = _absolute_local_path(raw["project_root"], "E_CONFIG")
    _check_existing_chain(project_root, require_file=False)
    if not project_root.is_dir() or not _within(path, project_root):
        raise ProtocolError("E_CONFIG")
    creator_name = _safe_text(raw["creator_name"], 128)
    try:
        creator_uid = validate_creator_uid(raw["creator_uid"])
    except ValueError as exc:
        raise ProtocolError("E_CONFIG") from exc
    host_path = _absolute_local_path(raw["host_config_path"], "E_CONFIG")
    _check_existing_chain(host_path, require_file=True)
    expected_host_hash = raw["host_config_sha256"]
    host_payload = _read_strict_payload(host_path, 128 * 1024, "E_CONFIG")
    actual_host_hash = hashlib.sha256(host_payload).hexdigest().upper()
    if not isinstance(expected_host_hash, str) or not _HEX64.fullmatch(expected_host_hash) or actual_host_hash != expected_host_hash.upper():
        raise ProtocolError("E_CONFIG_HASH")
    host = strict_json_loads(host_payload)
    if not isinstance(host, dict) or set(host) != _HOST_KEYS or host.get("schema") != 2:
        raise ProtocolError("E_CONFIG")
    creators = host.get("creator_allowlist")
    if not isinstance(creators, list) or creators != sorted(set(creators)) or creator_uid not in creators:
        raise ProtocolError("E_ALLOWLIST")
    if host.get("required_extension_build") != EXTENSION_BUILD:
        raise ProtocolError("E_CONFIG")
    if host.get("reload_generation") != RELOAD_GENERATION:
        raise ProtocolError("E_CONFIG")
    queue_paths = {
        name: _absolute_local_path(host[name], "E_CONFIG")
        for name in ("queue_path", "queue_state_path", "queue_lock_path", "reload_state_path")
    }
    if len(set(queue_paths.values())) != 4 or len({value.parent for value in queue_paths.values()}) != 1:
        raise ProtocolError("E_CONFIG")
    for candidate in queue_paths.values():
        _check_existing_chain(candidate, require_file=None)
        if candidate.exists() and not candidate.is_file():
            raise ProtocolError("E_CONFIG")
    dynamic_path = _absolute_local_path(raw["dynamic_manifest_path"], "E_CONFIG")
    if not _within(dynamic_path, project_root):
        raise ProtocolError("E_CONFIG")
    _check_existing_chain(dynamic_path, require_file=True)
    formal_path = _absolute_local_path(host["formal_manifest_path"], "E_CONFIG")
    handoff_path = _absolute_local_path(host["processing_handoff_path"], "E_CONFIG")
    if (
        host["creator_name"] != creator_name or formal_path != dynamic_path
        or not _within(handoff_path, project_root) or handoff_path == formal_path
    ):
        raise ProtocolError("E_CONFIG")
    _check_existing_chain(handoff_path, require_file=None)
    if handoff_path.exists() and not handoff_path.is_file():
        raise ProtocolError("E_CONFIG")
    registered_path = None
    registered_payload = None
    actual_registered_hash = None
    registered_hash = raw["registered_catalog_sha256"]
    if raw["registered_catalog_path"] is None:
        if registered_hash is not None:
            raise ProtocolError("E_CONFIG")
    else:
        registered_path = _absolute_local_path(raw["registered_catalog_path"], "E_CONFIG")
        if not _within(registered_path, project_root):
            raise ProtocolError("E_CONFIG")
        _check_existing_chain(registered_path, require_file=True)
        registered_payload = _read_strict_payload(registered_path, 4 * 1024 * 1024, "E_CATALOG")
        actual_registered_hash = hashlib.sha256(registered_payload).hexdigest().upper()
        if not isinstance(registered_hash, str) or not _HEX64.fullmatch(registered_hash) or actual_registered_hash != registered_hash.upper():
            raise ProtocolError("E_CONFIG_HASH")
    allowlist = raw["bvid_allowlist"]
    if allowlist is not None:
        if not isinstance(allowlist, list) or not allowlist or allowlist != sorted(set(allowlist)):
            raise ProtocolError("E_CONFIG")
        try:
            allowlist = [validate_bvid(value) for value in allowlist]
        except ValueError as exc:
            raise ProtocolError("E_CONFIG") from exc
    return {
        **raw,
        "project_root": project_root,
        "host_config_path": host_path,
        "dynamic_manifest_path": dynamic_path,
        "registered_catalog_path": registered_path,
        "creator_name": creator_name,
        "creator_uid": creator_uid,
        "bvid_allowlist": None if allowlist is None else frozenset(allowlist),
        "queue_paths": queue_paths,
        "_host_config_sha256": actual_host_hash,
        "_registered_catalog_payload": registered_payload,
        "_registered_catalog_sha256": actual_registered_hash,
    }


_RELEASE_KEYS = {
    "schema", "scope", "project_id", "task_id", "approved_by_role",
    "authorization_message_id", "authorization_handoff_id", "authorization_file",
    "implementation_review", "deployment",
}
_FILE_ID_KEYS = {"relative_path", "bytes", "sha256"}
_REVIEW_KEYS = {
    "result_message_id", "audit_id", "audit_bytes", "audit_sha256",
    "verdict", "blocking_findings",
}
_DEPLOYMENT_KEYS = {
    "extension_build", "host_build", "reload_generation", "source_manifest_sha256",
    "build_approval", "build_receipt", "exe", "install_approval", "install_receipt",
    "installed_config_sha256", "installed_manifest_sha256", "installed_exe_sha256",
    "projection_contract_sha256", "projection_tree_sha256", "extension_id",
}
_AUTH_KEYS = {
    "schema", "scope", "task_id", "authorized_by_role", "authorization_message_id",
    "authorization_handoff_id", "repair", "successors",
}
_AUTH_REPAIR_KEYS = {
    "review_result_message_id", "audit_id", "audit_bytes", "audit_sha256",
    "verdict", "blocking_findings",
}
_AUTH_SUCCESSOR_KEYS = {
    "creator_uid", "bvid", "predecessor_job_id", "retry_generation",
    "terminal_error_code",
}


def _project_root_from_source() -> Path:
    root = Path(__file__).resolve().parents[3]
    _check_existing_chain(root, require_file=False)
    project_config = root / "mbx.project.yaml"
    payload = _read_strict_payload(project_config, 4 * 1024 * 1024, "E_AUTH_TRUST")
    text = payload.decode("utf-8", errors="strict")
    if not re.search(r"(?m)^project:\s*\r?\n(?:^[ \t].*\r?\n)*?^  id: project-info\s*$", text) or not re.search(r"(?m)^  project_admin: project\.admin\s*$", text):
        raise ProtocolError("E_AUTH_TRUST")
    if not re.search(r"(?ms)^- id: project\.admin\s*$.*?^  private_dir: ai-infoadmin\s*$", text):
        raise ProtocolError("E_AUTH_TRUST")
    return root


def _relative_path(value: Any, prefixes: tuple[str, ...], code: str) -> str:
    if not isinstance(value, str) or not value or "\\" in value or _CONTROL.search(value):
        raise ProtocolError(code)
    path = PurePosixPath(value)
    if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
        raise ProtocolError(code)
    normalized = path.as_posix()
    if normalized != value or not any(normalized.startswith(prefix) for prefix in prefixes):
        raise ProtocolError(code)
    return normalized


def _file_identity(root: Path, value: Any, prefixes: tuple[str, ...], code: str) -> tuple[Path, bytes]:
    if not isinstance(value, dict) or set(value) != _FILE_ID_KEYS:
        raise ProtocolError(code)
    relative = _relative_path(value.get("relative_path"), prefixes, code)
    size = value.get("bytes")
    digest = value.get("sha256")
    if (
        isinstance(size, bool) or not isinstance(size, int) or size <= 0
        or not isinstance(digest, str) or not _HEX64.fullmatch(digest)
        or digest != digest.upper()
    ):
        raise ProtocolError(code)
    path = root / Path(relative)
    payload = _read_strict_payload(path, max(size, 1) + 1, code)
    if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
        raise ProtocolError(code)
    return path, payload


def _plain_mapping(value: Any, keys: set[str], code: str) -> dict[str, Any]:
    if not isinstance(value, dict) or set(value) != keys:
        raise ProtocolError(code)
    return value


def _upper_sha(value: Any, code: str) -> str:
    if not isinstance(value, str) or not _HEX64.fullmatch(value) or value != value.upper():
        raise ProtocolError(code)
    return value


def _positive_integer(value: Any, code: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ProtocolError(code)
    return value


def _extension_id_from_manifest_key(value: Any) -> str:
    if not isinstance(value, str) or _CONTROL.search(value):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    try:
        public_key = base64.b64decode(value, validate=True)
    except (ValueError, TypeError) as exc:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY") from exc
    alphabet = "abcdefghijklmnop"
    prefix = hashlib.sha256(public_key).digest()[:16]
    return "".join(alphabet[byte >> 4] + alphabet[byte & 15] for byte in prefix)


def _projection_tree(entries: list[dict[str, Any]]) -> str:
    builder = bytearray()
    for item in sorted(entries, key=lambda candidate: candidate["path"]):
        builder.extend(item["path"].encode("utf-8"))
        builder.extend(b"\0")
        builder.extend(str(item["bytes"]).encode("ascii"))
        builder.extend(b"\0")
        builder.extend(item["sha256"].encode("ascii"))
        builder.extend(b"\n")
    return hashlib.sha256(builder).hexdigest().upper()


def _read_native_host_registry_default() -> str:
    try:
        import winreg

        key_path = rf"Software\Google\Chrome\NativeMessagingHosts\com.project_info.bili_auth_ingress"
        with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_READ) as key:
            value, kind = winreg.QueryValueEx(key, None)
        if kind != winreg.REG_SZ or not isinstance(value, str):
            raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        return value
    except (OSError, ImportError) as exc:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY") from exc


def _read_native_host_registry_exact() -> str:
    """Return the sole default REG_SZ value; reject extra values or subkeys."""
    try:
        import winreg

        key_path = rf"Software\Google\Chrome\NativeMessagingHosts\com.project_info.bili_auth_ingress"
        with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_READ) as key:
            subkey_count, value_count, _ = winreg.QueryInfoKey(key)
            if subkey_count != 0 or value_count != 1:
                raise ProtocolError("E_DEPLOYMENT_NOT_READY")
            name, value, kind = winreg.EnumValue(key, 0)
            if name != "" or kind != winreg.REG_SZ or not isinstance(value, str):
                raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        return value
    except (OSError, ImportError) as exc:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY") from exc


def _fast_path_canonical_paths() -> dict[str, Path]:
    return {
        "queue": Path(r"C:\Users\Cai\AppData\Local\project-info\bili-auth-generic-runtime\queue.jsonl"),
        "queue_state": Path(r"C:\Users\Cai\AppData\Local\project-info\bili-auth-generic-runtime\queue-state.jsonl"),
        "reload_state": Path(r"C:\Users\Cai\AppData\Local\project-info\bili-auth-generic-runtime\reload-state.jsonl"),
        "formal_manifest": Path(r"E:\mb-ms-doc\project-info\ana-data\news-青枫浦上Q\manifest.jsonl"),
    }


def _successor_scope_sha256(successors: Any) -> str:
    if not isinstance(successors, list):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    normalized: list[dict[str, Any]] = []
    for value in successors:
        value = _plain_mapping(value, _AUTH_SUCCESSOR_KEYS, "E_DEPLOYMENT_NOT_READY")
        normalized.append({name: value[name] for name in sorted(_AUTH_SUCCESSOR_KEYS)})
    normalized.sort(key=lambda value: (
        value["creator_uid"], value["bvid"], value["predecessor_job_id"],
        value["retry_generation"], value["terminal_error_code"],
    ))
    payload = json.dumps(
        normalized, ensure_ascii=True, sort_keys=True, separators=(",", ":"),
    ).encode("ascii")
    return hashlib.sha256(payload).hexdigest().upper()


def _validate_prefix_identity(
    path: Path, receipt: dict[str, Any], prefix: str, *, exact: bool,
) -> None:
    code = "E_DEPLOYMENT_NOT_READY"
    expected_path = _absolute_local_path(receipt[f"{prefix}_path"], code)
    if path != expected_path:
        raise ProtocolError(code)
    size_name = f"{prefix}_bytes" if exact else f"{prefix}_prefix_bytes"
    lines_name = f"{prefix}_lines" if exact else f"{prefix}_prefix_lines"
    sha_name = f"{prefix}_sha256" if exact else f"{prefix}_prefix_sha256"
    size = _positive_integer(receipt[size_name], code)
    lines = _positive_integer(receipt[lines_name], code)
    digest = _upper_sha(receipt[sha_name], code)
    payload = _read_strict_payload(path, 32 * 1024 * 1024, code)
    if exact:
        candidate = payload
        if len(payload) != size:
            raise ProtocolError(code)
    else:
        if len(payload) < size:
            raise ProtocolError(code)
        candidate = payload[:size]
    if (
        not candidate.endswith(b"\n") or candidate.count(b"\n") != lines
        or hashlib.sha256(candidate).hexdigest().upper() != digest
    ):
        raise ProtocolError(code)


def _validate_review(value: Any, code: str) -> dict[str, Any]:
    review = _plain_mapping(value, _REVIEW_KEYS, code)
    if (
        not isinstance(review["result_message_id"], str)
        or not MESSAGE_ID_RE.fullmatch(review["result_message_id"])
        or not isinstance(review["audit_id"], str) or not review["audit_id"].startswith("DEV-AUDIT-")
        or _positive_integer(review["audit_bytes"], code) != review["audit_bytes"]
        or _upper_sha(review["audit_sha256"], code) != review["audit_sha256"]
        or review["verdict"] != "PASS/0" or review["blocking_findings"] != 0
    ):
        raise ProtocolError(code)
    return review


def _validate_audit_prefix(root: Path, review: dict[str, Any], code: str) -> None:
    audit_path = root / "dev-doc" / "开发审计报告.md"
    payload = _read_strict_payload(audit_path, 16 * 1024 * 1024, code)
    size = review["audit_bytes"]
    if len(payload) < size or hashlib.sha256(payload[:size]).hexdigest().upper() != review["audit_sha256"]:
        raise ProtocolError(code)
    try:
        prefix = payload[:size].decode("utf-8", errors="strict")
    except UnicodeError as exc:
        raise ProtocolError(code) from exc
    if review["audit_id"] not in prefix:
        raise ProtocolError(code)


def _validate_projection(root: Path, deployment: dict[str, Any]) -> None:
    contract_path = root / "dev" / "project-dev" / "bili_authenticated_extension_unpacked_contract.json"
    contract_payload = _read_strict_payload(contract_path, 128 * 1024, "E_DEPLOYMENT_NOT_READY")
    if hashlib.sha256(contract_payload).hexdigest().upper() != deployment["projection_contract_sha256"]:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    contract = strict_json_loads(contract_payload)
    required = {
        "schema", "task_id", "project_id", "source_root", "projection_root",
        "expected_extension_id", "public_key_der_sha256", "tree_hash_algorithm",
        "tree_sha256", "files",
    }
    if not isinstance(contract, dict) or set(contract) != required or contract.get("schema") != 1:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    if (
        contract["task_id"] != _TASK_ID or contract["project_id"] != _PROJECT_ID
        or contract["expected_extension_id"] != EXPECTED_EXTENSION_ID
        or contract["tree_sha256"] != deployment["projection_tree_sha256"]
        or contract["tree_hash_algorithm"] != "path-nul-bytes-nul-sha256-upper-lf-v1"
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    source_root = root / Path(_relative_path(contract["source_root"], ("dev/project-dev/bili_authenticated_extension",), "E_DEPLOYMENT_NOT_READY"))
    projection_root = root / Path(_relative_path(contract["projection_root"], ("dev/project-dev/bili_authenticated_extension_unpacked",), "E_DEPLOYMENT_NOT_READY"))
    _check_existing_chain(source_root, require_file=False)
    _check_existing_chain(projection_root, require_file=False)
    files = contract["files"]
    if not isinstance(files, list) or len(files) != 5:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    expected_paths = ["background.js", "manifest.json", "sidepanel.css", "sidepanel.html", "sidepanel.js"]
    _enumerate_exact_tree(projection_root, set(expected_paths), "E_DEPLOYMENT_NOT_READY")
    actual_paths: list[str] = []
    entries: list[dict[str, Any]] = []
    for spec in files:
        if not isinstance(spec, dict) or set(spec) != {"path", "bytes", "sha256"}:
            raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        name = spec["path"]
        if name not in expected_paths or name in actual_paths:
            raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        size = _positive_integer(spec["bytes"], "E_DEPLOYMENT_NOT_READY")
        digest = _upper_sha(spec["sha256"], "E_DEPLOYMENT_NOT_READY")
        payload = _read_strict_payload(projection_root / name, size + 1, "E_DEPLOYMENT_NOT_READY")
        source_payload = _read_strict_payload(source_root / name, size + 1, "E_DEPLOYMENT_NOT_READY")
        if len(payload) != size or payload != source_payload or hashlib.sha256(payload).hexdigest().upper() != digest:
            raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        actual_paths.append(name)
        entries.append({"path": name, "bytes": size, "sha256": digest})
    if sorted(actual_paths) != expected_paths or _projection_tree(entries) != contract["tree_sha256"]:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    manifest = strict_json_loads((projection_root / "manifest.json").read_bytes())
    if (
        not isinstance(manifest, dict) or manifest.get("manifest_version") != 3
        or manifest.get("version") != "1.2.25"
        or manifest.get("version_name") != "1.2.25+20260829.generic.v027"
        or _extension_id_from_manifest_key(manifest.get("key")) != EXPECTED_EXTENSION_ID
        or EXTENSION_BUILD.encode("ascii") not in (projection_root / "background.js").read_bytes()
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")


def _validate_reload_applied(path: Path) -> None:
    payload = _read_strict_payload(path, 4 * 1024 * 1024, "E_DEPLOYMENT_NOT_READY")
    if not payload.endswith(b"\n"):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    matching: list[dict[str, Any]] = []
    for line in payload.splitlines():
        value = strict_json_loads(line)
        if not isinstance(value, dict) or set(value) != {
            "schema", "generation", "event", "token", "from_build", "to_build", "at_unix_ms"
        }:
            raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        if value.get("generation") == RELOAD_GENERATION:
            matching.append(value)
    events = [value.get("event") for value in matching]
    if events not in (["APPLIED"], ["OFFERED", "BEGIN", "APPLIED"]):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    last = matching[-1]
    if last.get("from_build") != EXTENSION_BUILD or last.get("to_build") != EXTENSION_BUILD:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    if (
        any(value.get("schema") != 1 for value in matching)
        or any(not isinstance(value.get("token"), str) or not re.fullmatch(r"[0-9a-f]{32}", value["token"]) for value in matching)
        or len({value["token"] for value in matching}) != 1
        or any(value.get("to_build") != EXTENSION_BUILD for value in matching)
        or any(isinstance(value.get("at_unix_ms"), bool) or not isinstance(value.get("at_unix_ms"), int) or value["at_unix_ms"] <= 0 for value in matching)
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")


def _deployment_identity_branch(root: Path, deployment: dict[str, Any]) -> str:
    code = "E_DEPLOYMENT_NOT_READY"
    identities = {
        name: _plain_mapping(deployment[name], _FILE_ID_KEYS, code)["relative_path"]
        for name in ("build_approval", "install_approval", "install_receipt")
    }
    for relative in identities.values():
        _relative_path(relative, ("ai-inforev/worklog/", "ai-infoadmin/worklog/"), code)
    build_parent = PurePosixPath(identities["build_approval"]).parent.as_posix()
    install_parent = PurePosixPath(identities["install_approval"]).parent.as_posix()
    receipt_parent = PurePosixPath(identities["install_receipt"]).parent.as_posix()
    if (
        build_parent == "ai-inforev/worklog"
        and install_parent == "ai-inforev/worklog"
        and receipt_parent == "ai-infoadmin/worklog"
    ):
        return "legacy-reviewer"
    if (
        build_parent == "ai-infoadmin/worklog"
        and install_parent == "ai-infoadmin/worklog"
        and receipt_parent == "ai-infoadmin/worklog"
        and PurePosixPath(identities["build_approval"]).name.endswith("-fast-path-source-receipt.json")
        and PurePosixPath(identities["install_approval"]).name.endswith("-fast-path-build-receipt.json")
        and PurePosixPath(identities["install_receipt"]).name.endswith("-fast-path-install-readiness-receipt.json")
    ):
        return "project-admin-fast-path"
    raise ProtocolError(code)


def _validate_fast_path_host_source(
    root: Path,
    receipt: dict[str, Any],
    current_source: dict[str, Any],
    current_source_payload: bytes,
    build_receipt: dict[str, Any],
    install_approval: dict[str, Any],
) -> None:
    code = "E_DEPLOYMENT_NOT_READY"
    path, payload = _file_identity(
        root, receipt["host_build_source_manifest"], ("ai-infoadmin/worklog/",), code,
    )
    if (
        path.parent != root / "ai-infoadmin" / "worklog"
        or not path.name.endswith("-fast-path-host-build-source-manifest.json")
    ):
        raise ProtocolError(code)
    host_source = _plain_mapping(strict_json_loads(payload), _SOURCE_MANIFEST_KEYS, code)
    for name in _SOURCE_MANIFEST_KEYS - {"files"}:
        if host_source[name] != current_source[name]:
            raise ProtocolError(code)
    current_files = current_source["files"]
    host_files = host_source["files"]
    if not isinstance(current_files, list) or not isinstance(host_files, list) or len(host_files) != 21:
        raise ProtocolError(code)
    current_by_path: dict[str, dict[str, Any]] = {}
    host_by_path: dict[str, dict[str, Any]] = {}
    for target, values in ((current_by_path, current_files), (host_by_path, host_files)):
        for raw in values:
            item = _plain_mapping(raw, _ARTIFACT_FILE_KEYS, code)
            relative = _relative_path(item["path"], ("",), code)
            if relative in target:
                raise ProtocolError(code)
            _positive_integer(item["bytes"], code)
            _upper_sha(item["sha256"], code)
            target[relative] = item
    if set(current_by_path) != set(host_by_path):
        raise ProtocolError(code)
    changed = sorted(
        relative for relative in current_by_path
        if current_by_path[relative] != host_by_path[relative]
    )
    mode = receipt["host_source_binding_mode"]
    if mode == "SAME_SOURCE_HOST_BUILD":
        expected_changed: list[str] = []
        if payload != current_source_payload:
            raise ProtocolError(code)
    elif mode == "PRODUCER_ONLY_DELTA":
        expected_changed = ["queue_producer.py"]
        if payload == current_source_payload:
            raise ProtocolError(code)
    else:
        raise ProtocolError(code)
    host_manifest_hash = hashlib.sha256(payload).hexdigest().upper()
    if (
        changed != expected_changed
        or receipt["producer_only_changed_files"] != expected_changed
        or receipt["host_archive_excluded_modules"] != ["bili_authenticated_extension.queue_producer"]
        or "bili_authenticated_extension.queue_producer"
        in build_receipt["archive_verification"]["required_modules"]
        or build_receipt["source_artifact_manifest_bytes"] != len(payload)
        or build_receipt["source_artifact_manifest_sha256"] != host_manifest_hash
        or install_approval["source_artifact_manifest_sha256"] != host_manifest_hash
    ):
        raise ProtocolError(code)


def _validate_fast_path_receipt(
    root: Path,
    config: dict[str, Any],
    approval: dict[str, Any],
    authorization: dict[str, Any],
    deployment: dict[str, Any],
    source: dict[str, Any],
    build_receipt: dict[str, Any],
    install_approval: dict[str, Any],
    source_payload: bytes,
    source_receipt_payload: bytes,
    build_receipt_payload: bytes,
    build_validation_receipt_payload: bytes,
    exe_payload: bytes,
    receipt: dict[str, Any],
    actual: dict[str, tuple[Path, bytes]],
) -> None:
    code = "E_DEPLOYMENT_NOT_READY"
    owner_ai_id, owner_thread_id, owner_role = _FAST_PATH_OWNER
    authorization_spec = _plain_mapping(approval["authorization_file"], _FILE_ID_KEYS, code)
    review = _plain_mapping(approval["implementation_review"], _REVIEW_KEYS, code)
    source_receipt_hash = hashlib.sha256(source_receipt_payload).hexdigest().upper()
    build_validation_receipt_hash = hashlib.sha256(build_validation_receipt_payload).hexdigest().upper()
    build_receipt_hash = hashlib.sha256(build_receipt_payload).hexdigest().upper()
    exe_hash = hashlib.sha256(exe_payload).hexdigest().upper()
    if (
        receipt["schema"] != 2
        or receipt["task_id"] != _TASK_ID
        or receipt["validation_scope"] != "continuous-fast-path-installed-readiness-v1"
        or receipt["validated_by_role"] != "project.admin"
        or receipt["status"] != "VALIDATED"
        or receipt["continuous_authorization_handoff_id"] != _CONTINUOUS_FAST_PATH_HANDOFF_ID
        or receipt["owner_ai_id"] != owner_ai_id
        or receipt["owner_thread_id"] != owner_thread_id
        or receipt["owner_role_instance_id"] != owner_role
        or receipt["authorization_file_relative_path"] != authorization_spec["relative_path"]
        or receipt["authorization_file_bytes"] != authorization_spec["bytes"]
        or receipt["authorization_file_sha256"] != authorization_spec["sha256"]
        or receipt["successor_scope_sha256"] != _successor_scope_sha256(authorization["successors"])
        or receipt["implementation_review_audit_id"] != review["audit_id"]
        or receipt["implementation_review_audit_bytes"] != review["audit_bytes"]
        or receipt["implementation_review_audit_sha256"] != review["audit_sha256"]
        or receipt["source_artifact_manifest_bytes"] != len(source_payload)
        or receipt["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
        or receipt["source_receipt_bytes"] != len(source_receipt_payload)
        or receipt["source_receipt_sha256"] != source_receipt_hash
        or receipt["build_artifact_manifest_bytes"] != len(build_receipt_payload)
        or receipt["build_artifact_manifest_sha256"] != build_receipt_hash
        or receipt["build_receipt_bytes"] != len(build_validation_receipt_payload)
        or receipt["build_receipt_sha256"] != build_validation_receipt_hash
        or receipt["host_executable_bytes"] != len(exe_payload)
        or receipt["host_executable_sha256"] != exe_hash
        or receipt["native_messaging_host_manifest"] != str(actual["native-host-manifest.json"][0])
        or receipt["projection_contract_sha256"] != deployment["projection_contract_sha256"]
        or receipt["projection_tree_sha256"] != deployment["projection_tree_sha256"]
        or receipt["secret_field_count"] != 0
    ):
        raise ProtocolError(code)
    for name in (
        "authorization_file_sha256", "successor_scope_sha256",
        "implementation_review_audit_sha256", "source_artifact_manifest_sha256",
        "source_receipt_sha256", "build_artifact_manifest_sha256", "build_receipt_sha256",
        "host_executable_sha256", "projection_contract_sha256", "projection_tree_sha256",
    ):
        _upper_sha(receipt[name], code)
    for name in (
        "authorization_file_bytes", "implementation_review_audit_bytes",
        "source_artifact_manifest_bytes", "source_receipt_bytes",
        "build_artifact_manifest_bytes", "build_receipt_bytes", "host_executable_bytes",
    ):
        _positive_integer(receipt[name], code)
    _reject_secret_keys(receipt)
    _validate_fast_path_host_source(
        root, receipt, source, source_payload, build_receipt, install_approval,
    )
    canonical = _fast_path_canonical_paths()
    if (
        config["queue_paths"]["queue_path"] != canonical["queue"]
        or config["queue_paths"]["queue_state_path"] != canonical["queue_state"]
        or config["queue_paths"]["reload_state_path"] != canonical["reload_state"]
    ):
        raise ProtocolError(code)
    _validate_prefix_identity(canonical["reload_state"], receipt, "reload_state", exact=True)
    _validate_prefix_identity(canonical["queue"], receipt, "queue", exact=False)
    _validate_prefix_identity(canonical["queue_state"], receipt, "queue_state", exact=False)
    _validate_prefix_identity(canonical["formal_manifest"], receipt, "formal_manifest", exact=False)
    if Path(_read_native_host_registry_exact()) != actual["native-host-manifest.json"][0]:
        raise ProtocolError(code)


def _load_release_approval(config: dict[str, Any]) -> ReleaseApproval:
    if config.get("schema") != 2:
        raise ProtocolError("E_AUTH_TRUST")
    authorization_message_id = config.get("successor_authorization_message_id")
    if not isinstance(authorization_message_id, str) or not MESSAGE_ID_RE.fullmatch(authorization_message_id):
        raise ProtocolError("E_AUTH_TRUST")
    root = _project_root_from_source()
    if config["project_root"] != root:
        raise ProtocolError("E_AUTH_TRUST")
    approval_path = root / "ai-infoadmin" / "worklog" / f"bili-auth-successor-release-approval-{authorization_message_id}.json"
    approval_payload = _read_strict_payload(approval_path, 256 * 1024, "E_AUTH_TRUST")
    approval = strict_json_loads(approval_payload)
    _plain_mapping(approval, _RELEASE_KEYS, "E_AUTH_TRUST")
    if (
        approval["schema"] != 1 or approval["scope"] != "bili-auth-successor-exact-release-v1"
        or approval["project_id"] != _PROJECT_ID or approval["task_id"] != _TASK_ID
        or approval["approved_by_role"] != "project.admin"
        or approval["authorization_message_id"] != authorization_message_id
        or not isinstance(approval["authorization_handoff_id"], str)
        or not approval["authorization_handoff_id"].startswith("HANDOFF-")
    ):
        raise ProtocolError("E_AUTH_TRUST")
    implementation_review = _validate_review(approval["implementation_review"], "E_AUTH_TRUST")
    _validate_audit_prefix(root, implementation_review, "E_AUTH_TRUST")
    authorization_path, authorization_payload = _file_identity(
        root, approval["authorization_file"], ("ai-infoadmin/worklog/",), "E_AUTH_TRUST"
    )
    if authorization_path.parent != approval_path.parent:
        raise ProtocolError("E_AUTH_TRUST")
    authorization = strict_json_loads(authorization_payload)
    _plain_mapping(authorization, _AUTH_KEYS, "E_AUTH_TRUST")
    if (
        authorization["schema"] != 1 or authorization["scope"] != "bili-auth-successor-lineage-v1"
        or authorization["task_id"] != _TASK_ID or authorization["authorized_by_role"] != "project.admin"
        or authorization["authorization_message_id"] != authorization_message_id
        or authorization["authorization_handoff_id"] != approval["authorization_handoff_id"]
    ):
        raise ProtocolError("E_AUTH_TRUST")
    repair = _validate_review({
        "result_message_id": authorization["repair"].get("review_result_message_id") if isinstance(authorization.get("repair"), dict) else None,
        "audit_id": authorization["repair"].get("audit_id") if isinstance(authorization.get("repair"), dict) else None,
        "audit_bytes": authorization["repair"].get("audit_bytes") if isinstance(authorization.get("repair"), dict) else None,
        "audit_sha256": authorization["repair"].get("audit_sha256") if isinstance(authorization.get("repair"), dict) else None,
        "verdict": authorization["repair"].get("verdict") if isinstance(authorization.get("repair"), dict) else None,
        "blocking_findings": authorization["repair"].get("blocking_findings") if isinstance(authorization.get("repair"), dict) else None,
    }, "E_AUTH_TRUST")
    if not isinstance(authorization["repair"], dict) or set(authorization["repair"]) != _AUTH_REPAIR_KEYS:
        raise ProtocolError("E_AUTH_TRUST")
    _validate_audit_prefix(root, repair, "E_AUTH_TRUST")
    raw_successors = authorization["successors"]
    if not isinstance(raw_successors, list) or not 1 <= len(raw_successors) <= 100:
        raise ProtocolError("E_AUTH_TRUST")
    successors: list[AuthorizedSuccessor] = []
    for raw in raw_successors:
        _plain_mapping(raw, _AUTH_SUCCESSOR_KEYS, "E_AUTH_TRUST")
        successors.append(AuthorizedSuccessor(
            creator_uid=raw["creator_uid"], bvid=raw["bvid"],
            predecessor_job_id=raw["predecessor_job_id"],
            retry_generation=raw["retry_generation"], terminal_error_code=raw["terminal_error_code"],
        ))
    release = ReleaseApproval(
        authorization_message_id=authorization_message_id,
        authorization_handoff_id=approval["authorization_handoff_id"],
        authorization_sha256=hashlib.sha256(authorization_payload).hexdigest().upper(),
        repair_review_result_message_id=repair["result_message_id"],
        repair_audit_id=repair["audit_id"], repair_audit_bytes=repair["audit_bytes"],
        repair_audit_sha256=repair["audit_sha256"], successors=tuple(successors),
    )
    QueueStore._validate_release(release)
    _validate_deployment(root, config, approval, authorization)
    return release


def _validate_deployment(
    root: Path, config: dict[str, Any], approval: dict[str, Any], authorization: dict[str, Any]
) -> None:
    deployment = _plain_mapping(approval["deployment"], _DEPLOYMENT_KEYS, "E_DEPLOYMENT_NOT_READY")
    if (
        deployment["extension_build"] != EXTENSION_BUILD
        or deployment["host_build"] != HOST_BUILD
        or deployment["reload_generation"] != RELOAD_GENERATION
        or deployment["extension_id"] != EXPECTED_EXTENSION_ID
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    for name in (
        "source_manifest_sha256", "installed_config_sha256", "installed_manifest_sha256",
        "installed_exe_sha256", "projection_contract_sha256", "projection_tree_sha256",
    ):
        _upper_sha(deployment[name], "E_DEPLOYMENT_NOT_READY")
    source_manifest_path = root / "dev" / "project-dev" / "bili_authenticated_extension" / "source-artifact-manifest.json"
    source_payload = _read_strict_payload(source_manifest_path, 256 * 1024, "E_DEPLOYMENT_NOT_READY")
    if hashlib.sha256(source_payload).hexdigest().upper() != deployment["source_manifest_sha256"]:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    source = _validate_source_tree(root, source_payload, deployment)
    branch = _deployment_identity_branch(root, deployment)
    approval_prefix = ("ai-inforev/worklog/",) if branch == "legacy-reviewer" else ("ai-infoadmin/worklog/",)
    _, build_approval_payload = _file_identity(root, deployment["build_approval"], approval_prefix, "E_DEPLOYMENT_NOT_READY")
    _, build_receipt_payload = _file_identity(root, deployment["build_receipt"], ("dev/tmp/",), "E_DEPLOYMENT_NOT_READY")
    _, exe_payload = _file_identity(root, deployment["exe"], ("dev/tmp/",), "E_DEPLOYMENT_NOT_READY")
    _, install_approval_payload = _file_identity(root, deployment["install_approval"], approval_prefix, "E_DEPLOYMENT_NOT_READY")
    _, install_receipt_payload = _file_identity(root, deployment["install_receipt"], ("ai-infoadmin/worklog/",), "E_DEPLOYMENT_NOT_READY")
    build_approval = strict_json_loads(build_approval_payload)
    build_receipt = strict_json_loads(build_receipt_payload)
    install_approval = strict_json_loads(install_approval_payload)
    install_receipt = strict_json_loads(install_receipt_payload)
    build_approval = _plain_mapping(build_approval, _BUILD_APPROVAL_KEYS, "E_DEPLOYMENT_NOT_READY")
    build_receipt = _plain_mapping(build_receipt, _BUILD_RECEIPT_KEYS, "E_DEPLOYMENT_NOT_READY")
    install_approval = _plain_mapping(install_approval, _INSTALL_APPROVAL_KEYS, "E_DEPLOYMENT_NOT_READY")
    install_receipt = _plain_mapping(
        install_receipt,
        _INSTALL_RECEIPT_KEYS if branch == "legacy-reviewer" else _FAST_PATH_INSTALL_RECEIPT_KEYS,
        "E_DEPLOYMENT_NOT_READY",
    )
    exe_hash = hashlib.sha256(exe_payload).hexdigest().upper()
    build_receipt_hash = hashlib.sha256(build_receipt_payload).hexdigest().upper()
    expected_role = "dev.reviewer.project" if branch == "legacy-reviewer" else "project.admin"
    if (
        build_approval["schema"] != 1 or build_approval["task_id"] != _TASK_ID
        or build_approval["approval_scope"] != "controlled-build-source-manifest"
        or build_approval["approved_by_role"] != expected_role or build_approval["status"] != "APPROVED"
        or build_approval["source_artifact_manifest_bytes"] != len(source_payload)
        or build_approval["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
        or build_receipt["schema"] != 2 or build_receipt["scope"] != "generic-bilibili-queue"
        or build_receipt["extension_id"] != EXPECTED_EXTENSION_ID
        or build_receipt["extension_build"] != EXTENSION_BUILD or build_receipt["host_build"] != HOST_BUILD
        or build_receipt["packaging"] != "pyinstaller-onefile"
        or build_receipt["pyinstaller_version"] != "6.15.0" or build_receipt["yt_dlp_version"] != "2026.7.4"
        or (
            branch == "legacy-reviewer"
            and (
                build_receipt["source_artifact_manifest_bytes"] != len(source_payload)
                or build_receipt["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
            )
        )
        or build_receipt["dependency_artifact_manifest_bytes"] != source["dependency_artifact_manifest_bytes"]
        or build_receipt["dependency_artifact_manifest_sha256"] != source["dependency_artifact_manifest_sha256"]
        or install_approval["schema"] != 1 or install_approval["task_id"] != _TASK_ID
        or install_approval["approval_scope"] != "install-exact-build"
        or install_approval["approved_by_role"] != expected_role or install_approval["status"] != "APPROVED"
        or (
            branch == "legacy-reviewer"
            and install_approval["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
        )
        or install_approval["build_artifact_manifest_bytes"] != len(build_receipt_payload)
        or install_approval["build_artifact_manifest_sha256"] != build_receipt_hash
        or install_approval["host_executable_bytes"] != len(exe_payload)
        or install_approval["host_executable_sha256"] != exe_hash
        or install_receipt["schema"] != (1 if branch == "legacy-reviewer" else 2)
        or install_receipt["task_id"] != _TASK_ID
        or install_receipt["host_build"] != HOST_BUILD or install_receipt["required_extension_build"] != EXTENSION_BUILD
        or install_receipt["extension_id"] != EXPECTED_EXTENSION_ID
        or install_receipt["host_name"] != "com.project_info.bili_auth_ingress"
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    for name in (
        "builder_python_sha256", "pyinstaller_executable_sha256",
        "builder_provision_receipt_sha256", "build_script_sha256",
        "yt_dlp_wheel_sha256",
    ):
        _upper_sha(build_receipt[name], "E_DEPLOYMENT_NOT_READY")
    for name in (
        "pyinstaller_executable_bytes", "builder_provision_receipt_bytes",
    ):
        _positive_integer(build_receipt[name], "E_DEPLOYMENT_NOT_READY")
    archive = _plain_mapping(build_receipt["archive_verification"], _ARCHIVE_VERIFICATION_KEYS, "E_DEPLOYMENT_NOT_READY")
    if (
        archive["status"] != "PASS"
        or not isinstance(archive["method"], str) or not archive["method"] or _CONTROL.search(archive["method"])
        or archive["required_modules"] != [
            "bili_authenticated_extension.worker", "yt_dlp", "yt_dlp.downloader",
            "yt_dlp.globals", "yt_dlp.plugins", "yt_dlp.version",
        ]
        or archive["metadata_entry"] != "yt_dlp-2026.7.4.dist-info/METADATA"
        or archive["metadata_files"] != 7
        or not isinstance(archive["metadata_type_codes"], list)
        or not archive["metadata_type_codes"]
        or any(value not in {"b", "x"} for value in archive["metadata_type_codes"])
        or len(set(archive["metadata_type_codes"])) != len(archive["metadata_type_codes"])
        or archive["metadata_tree_sha256"] != source["archive_metadata_contract"]["canonical_tree_sha256"]
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    files = build_receipt.get("files")
    if not isinstance(files, list) or len(files) != 1:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    build_file = _plain_mapping(files[0], _ARTIFACT_FILE_KEYS, "E_DEPLOYMENT_NOT_READY")
    if (
        build_file["path"] != "project-info-bili-auth-native-host.exe"
        or build_file["bytes"] != len(exe_payload) or build_file["sha256"] != exe_hash
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    installed_root = _absolute_local_path(install_receipt.get("installed_root"), "E_DEPLOYMENT_NOT_READY")
    installed_files = install_receipt.get("installed_files")
    if not isinstance(installed_files, list) or {item.get("path") for item in installed_files if isinstance(item, dict)} != {
        "project-info-bili-auth-native-host.exe", "config.json", "native-host-manifest.json"
    }:
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    actual: dict[str, tuple[Path, bytes]] = {}
    for item in installed_files:
        item = _plain_mapping(item, _ARTIFACT_FILE_KEYS, "E_DEPLOYMENT_NOT_READY")
        size = _positive_integer(item["bytes"], "E_DEPLOYMENT_NOT_READY")
        digest = _upper_sha(item["sha256"], "E_DEPLOYMENT_NOT_READY")
        candidate = installed_root / item["path"]
        payload = _read_strict_payload(candidate, size + 1, "E_DEPLOYMENT_NOT_READY")
        if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
            raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        actual[item["path"]] = (candidate, payload)
    if (
        hashlib.sha256(actual["config.json"][1]).hexdigest().upper() != config.get("_host_config_sha256")
        or hashlib.sha256(actual["config.json"][1]).hexdigest().upper() != deployment["installed_config_sha256"]
        or hashlib.sha256(actual["native-host-manifest.json"][1]).hexdigest().upper() != deployment["installed_manifest_sha256"]
        or hashlib.sha256(actual["project-info-bili-auth-native-host.exe"][1]).hexdigest().upper() != deployment["installed_exe_sha256"]
        or actual["project-info-bili-auth-native-host.exe"][1] != exe_payload
        or deployment["installed_exe_sha256"] != exe_hash
        or actual["config.json"][0] != config["host_config_path"]
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    installed_config = _plain_mapping(
        strict_json_loads(actual["config.json"][1]), _HOST_KEYS, "E_DEPLOYMENT_NOT_READY"
    )
    if (
        installed_config["schema"] != 2
        or installed_config["required_extension_build"] != EXTENSION_BUILD
        or installed_config["reload_generation"] != RELOAD_GENERATION
        or installed_config["creator_allowlist"] != sorted(set(installed_config["creator_allowlist"]))
        or config.get("creator_uid") not in installed_config["creator_allowlist"]
        or any(
            installed_config[name] != str(config["queue_paths"][name])
            for name in ("queue_path", "queue_state_path", "queue_lock_path", "reload_state_path")
        )
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    native_manifest = strict_json_loads(actual["native-host-manifest.json"][1])
    if (
        not isinstance(native_manifest, dict) or native_manifest.get("name") != "com.project_info.bili_auth_ingress"
        or native_manifest.get("type") != "stdio" or native_manifest.get("allowed_origins") != [EXPECTED_ORIGIN]
        or Path(native_manifest.get("path", "")) != actual["project-info-bili-auth-native-host.exe"][0]
        or Path(_read_native_host_registry_default()) != actual["native-host-manifest.json"][0]
    ):
        raise ProtocolError("E_DEPLOYMENT_NOT_READY")
    _validate_projection(root, deployment)
    _validate_reload_applied(config["queue_paths"]["reload_state_path"])
    if branch == "project-admin-fast-path":
        _validate_fast_path_receipt(
            root, config, approval, authorization, deployment, source, build_receipt,
            install_approval, source_payload,
            build_approval_payload, build_receipt_payload, install_approval_payload,
            exe_payload, install_receipt, actual,
        )


def _formal_record(event: dict[str, Any], config: dict[str, Any]) -> tuple[str, dict[str, Any] | None, bool] | None:
    schema = event.get("schema_version")
    if schema == 1:
        if event.get("creator") != config["creator_name"]:
            raise ProtocolError("E_CATALOG")
        uid = event.get("creator_uid")
        if uid is not None and str(uid) != config["creator_uid"]:
            raise ProtocolError("E_ALLOWLIST")
        if event.get("item_type") != "video":
            return None
        bvid = event.get("stable_id")
        status = event.get("status")
        if not isinstance(status, str) or not status.startswith("VIDEO_"):
            raise ProtocolError("E_CATALOG")
        terminal = status in _TERMINAL_VIDEO_STATUSES or event.get("video_path") is not None
        duration = event.get("expected_duration_seconds")
    elif schema == 2:
        if event.get("creator") != config["creator_name"] or str(event.get("creator_uid")) != config["creator_uid"]:
            raise ProtocolError("E_ALLOWLIST")
        if event.get("event_type") != "DYNAMIC_CONTENT_SAVED" or event.get("status") != "SAVED":
            raise ProtocolError("E_CATALOG")
        if event.get("content_type") != "video":
            return None
        bvid = event.get("bvid")
        terminal = False
        duration = event.get("duration_seconds")
    else:
        raise ProtocolError("E_CATALOG")
    try:
        bvid = validate_bvid(bvid)
    except ValueError as exc:
        raise ProtocolError("E_CATALOG") from exc
    allowlist = config["bvid_allowlist"]
    if allowlist is not None and bvid not in allowlist:
        return bvid, None, terminal
    if event.get("source_url") != canonical_url(bvid):
        raise ProtocolError("E_CATALOG")
    if terminal:
        return bvid, None, True
    required = (event.get("title"), event.get("published_at"), event.get("collected_at"), duration)
    if any(value is None for value in required):
        return bvid, None, terminal
    record = {
        "schema": 1,
        "bvid": bvid,
        "creator_uid": config["creator_uid"],
        "expected_duration_ms": _duration_ms(duration),
        "discovered_at_unix_ms": _time_ms(event["collected_at"]),
        "published_at": _safe_text(event["published_at"], 64),
        "title": _safe_text(event["title"]),
    }
    validate_ingress_record(record, frozenset({config["creator_uid"]}))
    return bvid, record, terminal


def load_dynamic_manifest(config: dict[str, Any]) -> tuple[dict[str, dict[str, Any]], set[str], str]:
    path = config["dynamic_manifest_path"]
    payload = path.read_bytes()
    if len(payload) > 16 * 1024 * 1024 or (payload and not payload.endswith(b"\n")):
        raise ProtocolError("E_CATALOG")
    records: dict[str, dict[str, Any]] = {}
    terminal: set[str] = set()
    seen: set[str] = set()
    for line in payload.splitlines():
        if not line or len(line) > 1024 * 1024:
            raise ProtocolError("E_CATALOG")
        event = strict_json_loads(line)
        if not isinstance(event, dict):
            raise ProtocolError("E_CATALOG")
        _reject_secret_keys(event)
        normalized = _formal_record(event, config)
        if normalized is None:
            continue
        bvid, record, is_terminal = normalized
        seen.add(bvid)
        if is_terminal:
            terminal.add(bvid)
        if record is None:
            continue
        previous = records.get(bvid)
        if previous is None:
            records[bvid] = record
            continue
        left = {key: value for key, value in previous.items() if key != "discovered_at_unix_ms"}
        right = {key: value for key, value in record.items() if key != "discovered_at_unix_ms"}
        if left != right:
            raise ProtocolError("E_CATALOG_CONFLICT")
        previous["discovered_at_unix_ms"] = min(previous["discovered_at_unix_ms"], record["discovered_at_unix_ms"])
    return records, terminal, hashlib.sha256(payload).hexdigest().upper()


def load_registered_catalog(config: dict[str, Any]) -> dict[str, dict[str, Any]]:
    payload = config["_registered_catalog_payload"]
    if payload is None:
        return {}
    raw = strict_json_loads(payload)
    if not isinstance(raw, dict) or set(raw) != _REGISTERED_KEYS or raw.get("schema") != 1 or raw.get("source") != "bili-dynamic-collector-registered-v1":
        raise ProtocolError("E_CATALOG")
    _reject_secret_keys(raw)
    if raw.get("creator_name") != config["creator_name"] or raw.get("creator_uid") != config["creator_uid"]:
        raise ProtocolError("E_ALLOWLIST")
    items = raw.get("items")
    if not isinstance(items, list) or not 1 <= len(items) <= 10_000:
        raise ProtocolError("E_CATALOG")
    records: dict[str, dict[str, Any]] = {}
    for value in items:
        if not isinstance(value, dict) or set(value) != _INGRESS_KEYS:
            raise ProtocolError("E_CATALOG")
        job = validate_ingress_record(value, frozenset({config["creator_uid"]}))
        if job["creator_uid"] != config["creator_uid"]:
            raise ProtocolError("E_ALLOWLIST")
        allowlist = config["bvid_allowlist"]
        if allowlist is not None and job["bvid"] not in allowlist:
            continue
        previous = records.get(job["bvid"])
        if previous is not None and previous != value:
            raise ProtocolError("E_CATALOG_CONFLICT")
        records.setdefault(job["bvid"], value)
    return records


def collect_jobs(config: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    records, terminal, dynamic_hash = load_dynamic_manifest(config)
    registered = load_registered_catalog(config)
    for bvid, record in registered.items():
        previous = records.get(bvid)
        if previous is not None and previous != record:
            raise ProtocolError("E_CATALOG_CONFLICT")
        records.setdefault(bvid, record)
    allowlist = config["bvid_allowlist"]
    if allowlist is not None:
        missing = set(allowlist).difference(records, terminal)
        if missing:
            raise ProtocolError("E_CATALOG_MISSING")
    jobs = [records[bvid] for bvid in sorted(records) if bvid not in terminal]
    if not jobs:
        raise ProtocolError("E_NO_JOBS")
    return jobs, {
        "dynamic_manifest_sha256": dynamic_hash,
        "registered_catalog_sha256": config["_registered_catalog_sha256"],
        "catalog_jobs": len(records),
        "terminal_skipped": len(set(records).intersection(terminal)),
    }


def run(config_path: Path, *, append: bool, append_successors: bool = False) -> dict[str, Any]:
    if append and append_successors:
        raise ProtocolError("E_CONFIG")
    config = load_producer_configuration(config_path)
    jobs, evidence = collect_jobs(config)
    paths = config["queue_paths"]
    result: dict[str, int] = {"appended": 0, "unchanged": 0}
    if append or append_successors:
        runtime_root = paths["queue_path"].parent
        if append_successors:
            _check_existing_chain(runtime_root, require_file=False)
            _check_existing_chain(paths["queue_lock_path"], require_file=True)
            if not runtime_root.is_dir():
                raise ProtocolError("E_DEPLOYMENT_NOT_READY")
        else:
            runtime_root.mkdir(parents=True, exist_ok=True)
        _check_existing_chain(runtime_root, require_file=False)
        store = QueueStore(
            paths["queue_path"], paths["queue_state_path"], paths["queue_lock_path"],
            frozenset({config["creator_uid"]}),
        )
        if append_successors:
            result = store.append_authorized_successors(lambda: _load_release_approval(config), jobs)
        else:
            result = store.append_ingress_jobs(jobs)
    return {
        "schema": 2 if append_successors else 1,
        "producer_id": PRODUCER_ID,
        "status": (
            "SUCCESSOR_APPENDED" if append_successors and result["appended"]
            else "NO_CHANGE" if append_successors
            else "APPENDED" if append and result["appended"]
            else "NO_CHANGE" if append
            else "VALIDATION_PASS_ONLY"
        ),
        "job_count": len(jobs),
        **result,
        **evidence,
    }


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Produce governed Bilibili authenticated queue records.")
    parser.add_argument("--config", required=True, type=Path)
    action = parser.add_mutually_exclusive_group()
    action.add_argument("--append", action="store_true", help="Append initial schema-1 jobs after validation")
    action.add_argument("--append-successors", action="store_true", help="Append one admin-approved schema-2 successor block")
    return parser


def main(argv: list[str] | None = None) -> int:
    try:
        arguments = _parser().parse_args(argv)
        result = run(
            arguments.config, append=arguments.append,
            append_successors=arguments.append_successors,
        )
    except ProtocolError as exc:
        print(json.dumps({"schema": 1, "status": "FAILED", "error_code": exc.code}, sort_keys=True, separators=(",", ":")))
        return 3
    except (OSError, KeyError, TypeError, ValueError):
        print(json.dumps({"schema": 1, "status": "FAILED", "error_code": "E_PRODUCER"}, sort_keys=True, separators=(",", ":")))
        return 3
    print(json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
    return 0


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