"""Authenticated download worker security boundary.

This module is imported only by the worker mode.  It owns yt-dlp, local-only
FFmpeg operations, and the frozen ``accept-browser-file`` bridge invocation.
Signed media URLs and cookies never leave this process memory.
"""

from __future__ import annotations

import ctypes
import hashlib
import importlib.util
import io
import json
import math
import os
import re
import shutil
import stat
import subprocess
import sys
import time
import uuid
from collections.abc import Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable, Sequence
from urllib.parse import urlparse

from .constants import (
    BRIDGE_TIMEOUT_SECONDS,
    EXTENSION_BUILD,
    EXTRACTOR_RETRIES,
    FILE_ACCESS_RETRIES,
    FRAGMENT_RETRIES,
    HTTP_RETRIES,
    RELOAD_GENERATION,
    SOCKET_TIMEOUT_SECONDS,
    YTDLP_MODULE_SHA256,
    YTDLP_VERSION,
    duration_tolerance_ms,
    validate_bvid,
    validate_creator_uid,
)
from .formal_legacy_identity_manifest import (
    FORMAL_LEGACY_CREATOR_UID,
    FORMAL_LEGACY_INTEGER_UID_ROWS,
    FORMAL_LEGACY_ROWS,
    FORMAL_PREFIX_BYTES,
    FORMAL_PREFIX_LINES,
    FORMAL_PREFIX_SHA256,
)
from .protocol import (
    ProtocolError, strict_json_loads, validate_media_complete_identity, validate_start,
)

FROZEN_BRIDGE_SHA256 = "00F11DAF8387160DB863C89F0B33AB8480422233FF42199189222C989C7ED07E"
SUBPROCESS_POLICY_ERROR_CODES = frozenset({
    "E_SUBPROCESS_POLICY_ARGUMENTS",
    "E_SUBPROCESS_POLICY_ENVIRONMENT",
    "E_SUBPROCESS_POLICY_EVENT_SHAPE",
    "E_SUBPROCESS_POLICY_EXECUTABLE",
    "E_SUBPROCESS_POLICY_LOCAL_PATH",
    "E_SUBPROCESS_POLICY_SECRET",
})
_BRIDGE_MAPPING_KEYS = frozenset({
    "schema_version", "bvid", "source", "published_at", "title", "local_file",
    "bytes", "sha256", "duration_seconds", "remote_duration_seconds",
    "local_duration_seconds", "duration_delta_seconds", "duration_tolerance_seconds",
    "format_name", "video_codec", "audio_codec", "completed_at", "acquisition_mode",
    "handoff_source_sha256",
})
_BRIDGE_ITEM_KEYS = _BRIDGE_MAPPING_KEYS | {"status"}
_BRIDGE_ITEM_WARNING_KEYS = _BRIDGE_ITEM_KEYS | {"warning"}
_LOWER_SHA256_RE = re.compile(r"[0-9a-f]{64}")
_BRIDGE_CLEANUP_WARNING_RE = re.compile(
    r"staging cleanup requires attention: [A-Za-z][A-Za-z0-9_]{0,63}"
)
_UTC_ISO_RE = re.compile(
    r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?\+00:00"
)


class WorkerError(Exception):
    def __init__(self, code: str, diagnostic: dict[str, object] | None = None) -> None:
        super().__init__(code)
        self.code = code
        self.diagnostic = diagnostic


class CancelRequested(BaseException):
    pass


class NullLogger:
    def debug(self, _: object) -> None:
        return None

    def info(self, _: object) -> None:
        return None

    def warning(self, _: object) -> None:
        return None

    def error(self, _: object) -> None:
        return None


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


def verify_frozen_ytdlp() -> Path:
    """Verify the source package without importing it."""
    spec = importlib.util.find_spec("yt_dlp")
    if spec is None or spec.origin is None:
        raise WorkerError("E_YTDLP_FROZEN")
    root = Path(spec.origin).resolve().parent
    for relative, expected in YTDLP_MODULE_SHA256.items():
        path = root / Path(relative)
        if not path.is_file() or sha256_file(path) != expected:
            raise WorkerError("E_YTDLP_FROZEN")
    version_path = root / "version.py"
    version_scope: dict[str, Any] = {}
    exec(compile(version_path.read_bytes(), str(version_path), "exec"), version_scope)
    if version_scope.get("__version__") != YTDLP_VERSION:
        raise WorkerError("E_YTDLP_FROZEN")
    return root


def bootstrap_ytdlp() -> tuple[Any, Any]:
    """Disable every plugin source before the first yt-dlp import."""
    os.environ["YTDLP_NO_PLUGINS"] = "1"
    verify_frozen_ytdlp()
    import yt_dlp  # noqa: PLC0415 - deliberately after the environment gate
    import yt_dlp.globals as yt_globals  # noqa: PLC0415
    import yt_dlp.plugins as yt_plugins  # noqa: PLC0415

    yt_globals.plugin_dirs.value = []
    yt_plugins.load_all_plugins()
    plugin_overrides = dict(yt_globals.plugin_ies_overrides.value)
    if not (
        yt_globals.plugin_dirs.value == []
        and yt_plugins.directories() == []
        and yt_globals.plugin_ies.value == {}
        and yt_globals.plugin_pps.value == {}
        and plugin_overrides == {}
    ):
        raise WorkerError("E_PLUGIN_BOUNDARY")
    return yt_dlp, yt_globals


@dataclass(frozen=True)
class HostConfig:
    creator_allowlist: frozenset[str]
    ffmpeg: Path
    ffprobe: Path
    bridge_python: Path
    bridge_script: Path
    yt_dlp_executable: Path
    destination: Path
    queue_lock_path: Path | None = None
    formal_manifest_path: Path | None = None
    processing_handoff_path: Path | None = None
    creator_name: str = ""

    @staticmethod
    def _safe_absolute_file(value: Any, expected_hash: Any) -> Path:
        if not isinstance(value, str) or not isinstance(expected_hash, str):
            raise WorkerError("E_CONFIG")
        path = Path(value)
        if not path.is_absolute() or str(path).startswith("\\\\"):
            raise WorkerError("E_CONFIG")
        resolved = path.resolve(strict=True)
        if not resolved.is_file() or resolved.is_symlink():
            raise WorkerError("E_CONFIG")
        if sha256_file(resolved) != expected_hash.upper():
            raise WorkerError("E_CONFIG_HASH")
        return resolved

    @classmethod
    def load(cls, path: Path) -> "HostConfig":
        try:
            if path.is_symlink():
                raise WorkerError("E_CONFIG")
            raw = strict_json_loads(path.read_bytes())
        except (OSError, UnicodeError, json.JSONDecodeError, ProtocolError) as exc:
            raise WorkerError("E_CONFIG") from exc
        expected = {
            "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",
            "formal_manifest_path",
            "processing_handoff_path",
            "creator_name",
        }
        if not isinstance(raw, dict) or set(raw) != expected:
            raise WorkerError("E_CONFIG")
        if raw["schema"] != 2:
            raise WorkerError("E_CONFIG")
        if raw["required_extension_build"] != EXTENSION_BUILD or raw["reload_generation"] != RELOAD_GENERATION:
            raise WorkerError("E_CONFIG")
        creators = raw["creator_allowlist"]
        try:
            if (
                not isinstance(creators, list) or not creators or len(creators) > 64
                or [validate_creator_uid(item) for item in creators] != sorted(set(creators))
            ):
                raise WorkerError("E_CONFIG")
        except ValueError as exc:
            raise WorkerError("E_CONFIG") from exc
        for name in (
            "queue_path", "queue_state_path", "queue_lock_path", "reload_state_path",
            "formal_manifest_path", "processing_handoff_path",
        ):
            value = raw[name]
            if not isinstance(value, str):
                raise WorkerError("E_CONFIG")
        creator_name = raw["creator_name"]
        if (
            not isinstance(creator_name, str) or not creator_name.strip()
            or len(creator_name.encode("utf-8")) > 240
            or any(ord(ch) < 32 or ord(ch) == 127 for ch in creator_name)
        ):
            raise WorkerError("E_CONFIG")
        queue_lock_path = Path(raw["queue_lock_path"])
        formal_manifest_path = Path(raw["formal_manifest_path"])
        processing_handoff_path = Path(raw["processing_handoff_path"])
        governed_paths: list[Path] = []
        for governed_path, must_exist in (
            (queue_lock_path, False),
            (formal_manifest_path, True),
            (processing_handoff_path, False),
        ):
            if not governed_path.is_absolute() or str(governed_path).startswith("\\\\"):
                raise WorkerError("E_CONFIG")
            try:
                lexical_parent = governed_path.parent
                parent_stat = lexical_parent.lstat()
                resolved_parent = lexical_parent.resolve(strict=True)
                if (
                    not stat.S_ISDIR(parent_stat.st_mode)
                    or lexical_parent.is_symlink()
                    or _is_reparse(lexical_parent)
                    or resolved_parent != lexical_parent
                ):
                    raise WorkerError("E_CONFIG")
                resolved_path = resolved_parent / governed_path.name
                try:
                    path_stat = governed_path.lstat()
                except FileNotFoundError:
                    if must_exist:
                        raise WorkerError("E_CONFIG")
                else:
                    if (
                        not stat.S_ISREG(path_stat.st_mode)
                        or governed_path.is_symlink()
                        or _is_reparse(governed_path)
                        or governed_path.resolve(strict=True) != resolved_path
                    ):
                        raise WorkerError("E_CONFIG")
            except WorkerError:
                raise
            except OSError as exc:
                raise WorkerError("E_CONFIG") from exc
            governed_paths.append(resolved_path)
        if len(set(governed_paths)) != len(governed_paths):
            raise WorkerError("E_CONFIG")
        queue_lock_path, formal_manifest_path, processing_handoff_path = governed_paths
        bridge_script = cls._safe_absolute_file(raw["bridge_script"], raw["bridge_script_sha256"])
        if raw["bridge_script_sha256"].upper() != FROZEN_BRIDGE_SHA256:
            raise WorkerError("E_CONFIG_HASH")
        destination = Path(raw["destination"])
        if not destination.is_absolute() or str(destination).startswith("\\\\"):
            raise WorkerError("E_CONFIG")
        destination = destination.resolve(strict=True)
        if not destination.is_dir() or destination.is_symlink():
            raise WorkerError("E_CONFIG")
        return cls(
            creator_allowlist=frozenset(creators),
            ffmpeg=cls._safe_absolute_file(raw["ffmpeg"], raw["ffmpeg_sha256"]),
            ffprobe=cls._safe_absolute_file(raw["ffprobe"], raw["ffprobe_sha256"]),
            bridge_python=cls._safe_absolute_file(
                raw["bridge_python"], raw["bridge_python_sha256"]
            ),
            bridge_script=bridge_script,
            yt_dlp_executable=cls._safe_absolute_file(
                raw["yt_dlp_executable"], raw["yt_dlp_executable_sha256"]
            ),
            destination=destination,
            queue_lock_path=queue_lock_path,
            formal_manifest_path=formal_manifest_path,
            processing_handoff_path=processing_handoff_path,
            creator_name=creator_name,
        )


def _is_reparse(path: Path) -> bool:
    try:
        attributes = path.lstat().st_file_attributes
    except AttributeError:
        return path.is_symlink()
    return bool(attributes & 0x400)


def _reject_reparse_path(path: Path, stop: Path | None = None) -> None:
    current = path
    stop_value = stop.resolve(strict=False) if stop is not None else None
    while True:
        if current.exists() and _is_reparse(current):
            raise WorkerError("E_STAGE")
        if (stop_value is not None and current.resolve(strict=False) == stop_value) or current.parent == current:
            break
        current = current.parent


def validated_local_app_data() -> Path:
    """Return the one non-secret environment path retained by the worker."""
    value = os.environ.get("LOCALAPPDATA")
    if not value:
        raise WorkerError("E_STAGE")
    candidate = Path(value)
    if not candidate.is_absolute() or str(candidate).startswith("\\\\"):
        raise WorkerError("E_STAGE")
    _reject_reparse_path(candidate)
    try:
        resolved = candidate.resolve(strict=True)
    except OSError as exc:
        raise WorkerError("E_STAGE") from exc
    if not resolved.is_dir() or _is_reparse(resolved):
        raise WorkerError("E_STAGE")
    return resolved


def fixed_stage_root(bvid: str) -> Path:
    local_app_data = validated_local_app_data()
    try:
        bvid = validate_bvid(bvid)
    except ValueError as exc:
        raise WorkerError("E_STAGE") from exc
    logical_root = (
        local_app_data
        / "project-info"
        / "bili-auth-ingress"
        / bvid
    )
    _reject_reparse_path(logical_root, local_app_data)
    resolved = logical_root.resolve(strict=False)
    _ensure_within(resolved, local_app_data)
    return resolved


def _ensure_within(path: Path, root: Path) -> Path:
    resolved = path.resolve(strict=False)
    try:
        resolved.relative_to(root.resolve(strict=False))
    except ValueError as exc:
        raise WorkerError("E_STAGE") from exc
    return resolved


def _reject_reparse_chain(path: Path, stop: Path) -> None:
    _reject_reparse_path(path, stop)


def cleanup_stale_runs(root: Path, *, boundary: Path | None = None) -> None:
    """Remove only uncommitted run-* directories below the fixed stage root."""
    allowed_root = root.resolve() if boundary is None else boundary.resolve()
    _ensure_within(root, allowed_root)
    if not root.exists():
        return
    _reject_reparse_chain(root, allowed_root)
    for child in root.iterdir():
        if not child.name.startswith("run-") or not child.is_dir() or child.is_symlink():
            raise WorkerError("E_STAGE")
        _ensure_within(child, root)
        shutil.rmtree(child)


def create_run_directory(root: Path) -> Path:
    stage_root = root.resolve()
    _ensure_within(stage_root, stage_root)
    stage_root.mkdir(parents=True, exist_ok=True)
    _reject_reparse_chain(stage_root, stage_root)
    for _ in range(8):
        candidate = stage_root / f"run-{uuid.uuid4().hex}"
        try:
            candidate.mkdir(exist_ok=False)
            return candidate
        except FileExistsError:
            continue
    raise WorkerError("E_STAGE")


def prepare_run_directory(root: Path) -> Path:
    """Clean stale runs and create the secret-free task lease."""
    stage_root = root.resolve()
    cleanup_stale_runs(stage_root, boundary=stage_root)
    return create_run_directory(stage_root)


def cleanup_run_directory(run_directory: Path, stage_root: Path) -> None:
    _ensure_within(run_directory, stage_root)
    if run_directory.exists():
        if not run_directory.is_dir() or _is_reparse(run_directory):
            raise WorkerError("E_STAGE")
        shutil.rmtree(run_directory)
    if stage_root.exists() and not any(stage_root.iterdir()):
        stage_root.rmdir()


def build_cookie_stream(start: dict[str, Any]) -> io.StringIO:
    """Convert validated Chrome records to an in-memory Netscape jar."""
    validate_start(start)
    stream = io.StringIO(newline="\n")
    stream.write("# Netscape HTTP Cookie File\n")
    for cookie in start["cookies"]:
        domain = cookie["domain"]
        if cookie["http_only"]:
            domain = f"#HttpOnly_{domain}"
        fields = (
            domain,
            "FALSE" if cookie["host_only"] else "TRUE",
            cookie["path"],
            "TRUE" if cookie["secure"] else "FALSE",
            "0" if cookie["session"] else str(cookie["expiration_unix"]),
            cookie["name"],
            cookie["value"],
        )
        stream.write("\t".join(fields) + "\n")
    stream.seek(0)
    return stream


def close_cookie_stream(stream: io.StringIO | None) -> bool:
    if stream is None:
        return True
    try:
        stream.seek(0)
        stream.truncate(0)
    finally:
        stream.close()
    return stream.closed


def _finite_number(value: Any) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise WorkerError("E_METADATA")
    converted = float(value)
    if not math.isfinite(converted):
        raise WorkerError("E_METADATA")
    return converted


def validate_processed_info(info: Any, job: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(info, dict) or info.get("id") != job["bvid"]:
        raise WorkerError("E_METADATA")
    if info.get("entries") not in (None, []) or info.get("_type") not in (None, "video"):
        raise WorkerError("E_MULTI_PART")
    if info.get("playlist_count") not in (None, 1) or info.get("playlist_index") not in (None, 1):
        raise WorkerError("E_MULTI_PART")
    if info.get("is_live") is True or info.get("live_status") not in (None, "not_live"):
        raise WorkerError("E_LIVE")
    if info.get("has_drm") is True:
        raise WorkerError("E_DRM")
    if info.get("availability") not in (None, "public", "unlisted"):
        raise WorkerError("E_ENTITLEMENT")
    owner_ids = {str(value) for value in (info.get("uploader_id"), info.get("channel_id")) if value is not None}
    if job["creator_uid"] not in owner_ids:
        raise WorkerError("E_OWNER")
    duration_ms = round(_finite_number(info.get("duration")) * 1000)
    if abs(duration_ms - job["expected_duration_ms"]) > duration_tolerance_ms(job["expected_duration_ms"]):
        raise WorkerError("E_DURATION")
    formats = info.get("formats")
    if not isinstance(formats, list) or not formats:
        raise WorkerError("E_FORMAT")
    return info


def _format_leaves(download_info: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]:
    requested = download_info.get("requested_formats")
    if requested is None:
        leaves = [download_info]
        single = True
    else:
        if not isinstance(requested, list) or len(requested) != 2:
            raise WorkerError("E_FORMAT")
        if not all(isinstance(item, dict) for item in requested):
            raise WorkerError("E_FORMAT")
        leaves = requested
        single = False
    return leaves, single


def validate_download_info(
    download_info: Any,
    params: dict[str, Any],
    job: dict[str, Any],
    *,
    downloader_resolver: Callable[..., Any] | None = None,
) -> tuple[list[dict[str, Any]], bool]:
    if not isinstance(download_info, dict) or download_info.get("id") != job["bvid"]:
        raise WorkerError("E_FORMAT")
    leaves, single = _format_leaves(download_info)
    if single:
        if download_info.get("vcodec") in (None, "none") or download_info.get("acodec") in (None, "none"):
            raise WorkerError("E_FORMAT")
    else:
        video_only = sum(
            leaf.get("vcodec") not in (None, "none") and leaf.get("acodec") == "none"
            for leaf in leaves
        )
        audio_only = sum(
            leaf.get("vcodec") == "none" and leaf.get("acodec") not in (None, "none")
            for leaf in leaves
        )
        if video_only != 1 or audio_only != 1:
            raise WorkerError("E_FORMAT")
    if downloader_resolver is None:
        from yt_dlp.downloader import get_suitable_downloader  # noqa: PLC0415

        downloader_resolver = get_suitable_downloader
    for leaf in leaves:
        if leaf.get("has_drm") is True:
            raise WorkerError("E_DRM")
        url = leaf.get("url")
        protocol = leaf.get("protocol")
        if not isinstance(url, str) or urlparse(url).scheme != "https":
            raise WorkerError("E_FORMAT")
        if protocol not in {"https", "http_dash_segments"}:
            raise WorkerError("E_DOWNLOADER")
        downloader = downloader_resolver(leaf, params)
        if getattr(downloader, "__name__", "") not in {"HttpFD", "DashSegmentsFD"}:
            raise WorkerError("E_DOWNLOADER")
    return leaves, single


def prepare_download_info(
    ydl: Any,
    job: dict[str, Any],
    *,
    downloader_resolver: Callable[..., Any] | None = None,
) -> tuple[dict[str, Any], bool, tuple[str, ...]]:
    extract_count = 0
    original_extract = ydl.extract_info

    def one_extract(*args: Any, **kwargs: Any) -> Any:
        nonlocal extract_count
        extract_count += 1
        if extract_count != 1:
            raise WorkerError("E_SECOND_EXTRACT")
        return original_extract(*args, **kwargs)

    ydl.extract_info = one_extract
    processed = ydl.extract_info(job["canonical_url"], download=False, process=True)
    validate_processed_info(processed, job)
    selector = ydl.build_format_selector("bestvideo+bestaudio/best")
    selected = list(ydl._select_formats(ydl._get_formats(processed), selector))
    if len(selected) != 1:
        raise WorkerError("E_FORMAT")
    download_info = ydl._copy_infodict(processed)
    download_info.update(selected[0])
    leaves, single = validate_download_info(
        download_info,
        ydl.params,
        job,
        downloader_resolver=downloader_resolver,
    )
    signed_urls = tuple(str(leaf["url"]) for leaf in leaves)

    def reject_second_extract(*_: Any, **__: Any) -> Any:
        raise WorkerError("E_SECOND_EXTRACT")

    ydl.extract_info = reject_second_extract
    if extract_count != 1:
        raise WorkerError("E_SECOND_EXTRACT")
    return download_info, single, signed_urls


class SubprocessPolicy:
    """Pre-CreateProcess audit for local-only child command lines."""

    _FORBIDDEN = ("://", "-headers", "-cookies", "authorization", "cookie:", "referer:", "user-agent:")
    _FFMPEG_FLAGS = frozenset({"-nostdin", "-y"})
    _FFMPEG_SCALAR_OPTIONS = frozenset(
        {"-v", "-loglevel", "-map", "-c", "-map_metadata", "-f", "-movflags"}
    )
    _FFMPEG_REPEATABLE_OPTIONS = frozenset({"-map"})
    _FFPROBE_FLAGS = frozenset({"-hide_banner", "-show_format", "-show_streams"})
    _FFPROBE_SCALAR_OPTIONS = frozenset({"-v", "-show_entries", "-of", "-print_format"})
    _WINDOWS_RESERVED_NAMES = frozenset(
        {"CON", "PRN", "AUX", "NUL", *(f"COM{index}" for index in range(1, 10)),
         *(f"LPT{index}" for index in range(1, 10))}
    )
    _MAX_ARGUMENTS = 256
    _MAX_COMMAND_LINE = 32_767
    _MAX_ENVIRONMENT_ITEMS = 256
    _MAX_LOCAL_FILE_OPERANDS = 16

    def __init__(
        self,
        run_root: Path,
        executables: Iterable[Path],
        secrets: Iterable[str] = (),
    ) -> None:
        self.run_root = run_root.resolve()
        self.executables = {os.path.normcase(str(item.resolve())) for item in executables}
        self.secrets = {item.casefold() for item in secrets if item}

    @staticmethod
    def _fail(code: str) -> None:
        if code not in SUBPROCESS_POLICY_ERROR_CODES:
            code = "E_SUBPROCESS_POLICY_EVENT_SHAPE"
        raise WorkerError(code)

    @classmethod
    def _windows_arguments(cls, command_line: Any) -> list[str]:
        if (
            not isinstance(command_line, str)
            or not command_line
            or len(command_line) > cls._MAX_COMMAND_LINE
            or "\x00" in command_line
        ):
            cls._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
        argc = ctypes.c_int()
        command_line_to_argv = ctypes.windll.shell32.CommandLineToArgvW
        command_line_to_argv.argtypes = (ctypes.c_wchar_p, ctypes.POINTER(ctypes.c_int))
        command_line_to_argv.restype = ctypes.POINTER(ctypes.c_wchar_p)
        pointer = command_line_to_argv(command_line, ctypes.byref(argc))
        if not pointer:
            cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
        try:
            result = [pointer[index] for index in range(argc.value)]
        finally:
            local_free = ctypes.windll.kernel32.LocalFree
            local_free.argtypes = (ctypes.c_void_p,)
            local_free.restype = ctypes.c_void_p
            local_free(ctypes.cast(pointer, ctypes.c_void_p))
        if (
            not result
            or len(result) > cls._MAX_ARGUMENTS
            or any(not isinstance(item, str) or "\x00" in item for item in result)
            or subprocess.list2cmdline(result) != command_line
        ):
            cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
        return result

    @classmethod
    def _portable_arguments(cls, raw: Any) -> list[str]:
        if not isinstance(raw, (list, tuple)) or not raw or len(raw) > cls._MAX_ARGUMENTS:
            cls._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
        if any(type(item) is not str or "\x00" in item for item in raw):
            cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
        return list(raw)

    @staticmethod
    def _absolute_executable(value: Any) -> tuple[str, Path]:
        if not isinstance(value, (str, os.PathLike)):
            SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
        filesystem_value = os.fspath(value)
        if not isinstance(filesystem_value, str) or not filesystem_value or "\x00" in filesystem_value:
            SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
        path = Path(filesystem_value)
        if not path.is_absolute() or str(path).startswith("\\\\"):
            SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
        try:
            resolved = path.resolve(strict=True)
        except OSError as exc:
            raise WorkerError("E_SUBPROCESS_POLICY_EXECUTABLE") from exc
        if not resolved.is_file() or resolved.is_symlink() or _is_reparse(resolved):
            SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
        return os.path.normcase(str(resolved)), resolved

    def _validate_environment(self, environment: Any) -> None:
        if environment is None:
            return
        if not isinstance(environment, Mapping) or len(environment) > self._MAX_ENVIRONMENT_ITEMS:
            self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
        encoded: list[str] = []
        for key, value in environment.items():
            if (
                type(key) is not str
                or type(value) is not str
                or not key
                or "\x00" in key
                or "\x00" in value
            ):
                self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
            encoded.append(f"{key}={value}")
        folded = "\x00".join(encoded).casefold()
        if any(item in folded for item in self._FORBIDDEN) or any(item in folded for item in self.secrets):
            self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")

    def check(self, event: str, arguments: tuple[Any, ...]) -> None:
        if event != "subprocess.Popen":
            return
        if not isinstance(arguments, tuple) or len(arguments) != 4:
            self._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
        executable, raw_arguments, cwd, environment = arguments
        if cwd is not None:
            self._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
        text_args = (
            self._windows_arguments(raw_arguments)
            if os.name == "nt"
            else self._portable_arguments(raw_arguments)
        )
        argv_executable_key, _ = self._absolute_executable(text_args[0])
        if executable is None:
            executable_key = argv_executable_key
        else:
            executable_key, _ = self._absolute_executable(executable)
            if executable_key != argv_executable_key:
                self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
        if executable_key not in self.executables:
            self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
        folded = "\x00".join(text_args).casefold()
        if any(item in folded for item in self._FORBIDDEN) or any(item in folded for item in self.secrets):
            self._fail("E_SUBPROCESS_POLICY_SECRET")
        self._validate_environment(environment)
        executable_name = Path(executable_key).name.casefold()
        if executable_name == "ffmpeg.exe":
            self._validate_ffmpeg_arguments(text_args)
        elif executable_name == "ffprobe.exe":
            self._validate_ffprobe_arguments(text_args)

    @staticmethod
    def _valid_ffmpeg_scalar(option: str, value: str) -> bool:
        if option == "-v":
            return value == "error"
        if option == "-loglevel":
            return value == "repeat+info"
        if option == "-map":
            return re.fullmatch(r"\d+(?::[av](?::\d+)?)?", value) is not None
        if option == "-c":
            return value == "copy"
        if option == "-map_metadata":
            return value == "-1"
        if option == "-f":
            return value == "matroska"
        if option == "-movflags":
            return value == "+faststart"
        return False

    @staticmethod
    def _valid_ffprobe_scalar(option: str, value: str) -> bool:
        if option == "-v":
            return value == "error"
        if option == "-show_entries":
            return value == "format=format_name,duration:stream=codec_type"
        if option in {"-of", "-print_format"}:
            return value == "json"
        return False

    def _validate_ffmpeg_arguments(self, arguments: Sequence[str]) -> None:
        if tuple(arguments[1:]) == ("-bsfs",):
            return
        index = 1
        input_count = 0
        file_operand_count = 0
        output_seen = False
        seen_options: set[str] = set()
        while index < len(arguments):
            option = arguments[index]
            if option in self._FFMPEG_FLAGS:
                if option in seen_options:
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                seen_options.add(option)
                index += 1
                continue
            if option in {"-i", "-attach"}:
                if index + 1 >= len(arguments) or arguments[index + 1].startswith("-"):
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                self._local_path(arguments[index + 1], must_exist=True)
                file_operand_count += 1
                input_count += option == "-i"
                if file_operand_count > self._MAX_LOCAL_FILE_OPERANDS:
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                index += 2
                continue
            if option in self._FFMPEG_SCALAR_OPTIONS:
                if index + 1 >= len(arguments):
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                value = arguments[index + 1]
                if not self._valid_ffmpeg_scalar(option, value):
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                if option not in self._FFMPEG_REPEATABLE_OPTIONS:
                    if option in seen_options:
                        self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                    seen_options.add(option)
                index += 2
                continue
            if re.fullmatch(r"-bsf:a:\d+", option):
                if index + 1 >= len(arguments) or arguments[index + 1] != "aac_adtstoasc":
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                if option in seen_options:
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                seen_options.add(option)
                index += 2
                continue
            if option.startswith("-") or output_seen or index != len(arguments) - 1:
                self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
            self._local_path(option, must_exist=False)
            output_seen = True
            file_operand_count += 1
            index += 1
        if input_count < 1 or not output_seen or file_operand_count > self._MAX_LOCAL_FILE_OPERANDS:
            self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")

    def _validate_ffprobe_arguments(self, arguments: Sequence[str]) -> None:
        if tuple(arguments[1:]) == ("-bsfs",):
            return
        index = 1
        input_seen = False
        seen_options: set[str] = set()
        while index < len(arguments):
            option = arguments[index]
            if option in self._FFPROBE_FLAGS:
                if option in seen_options:
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                seen_options.add(option)
                index += 1
                continue
            if option in self._FFPROBE_SCALAR_OPTIONS:
                if index + 1 >= len(arguments) or not self._valid_ffprobe_scalar(
                    option, arguments[index + 1]
                ):
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                if option in seen_options:
                    self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
                seen_options.add(option)
                index += 2
                continue
            if option.startswith("-") or input_seen or index != len(arguments) - 1:
                self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
            self._local_path(option, must_exist=True)
            input_seen = True
            index += 1
        if not input_seen:
            self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")

    def _local_path(self, value: str, *, must_exist: bool) -> Path:
        if value.startswith("file:"):
            value = value[5:]
        path = Path(value)
        if (
            not value
            or not path.is_absolute()
            or str(path).startswith("\\\\")
            or ":" in value[2:]
            or any(part == ".." or part.endswith((" ", ".")) for part in path.parts)
            or path.name.split(".", 1)[0].upper() in self._WINDOWS_RESERVED_NAMES
        ):
            self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
        try:
            path.relative_to(self.run_root)
            _reject_reparse_chain(path, self.run_root)
            if must_exist:
                resolved = path.resolve(strict=True)
                resolved.relative_to(self.run_root)
                if not resolved.is_file() or resolved.is_symlink() or _is_reparse(resolved):
                    self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
            else:
                if path.exists() or path.is_symlink():
                    self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
                parent = path.parent.resolve(strict=True)
                parent.relative_to(self.run_root)
                _reject_reparse_chain(path.parent, self.run_root)
                if not parent.is_dir() or parent.is_symlink() or _is_reparse(parent):
                    self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
                resolved = parent / path.name
        except (OSError, ValueError, WorkerError) as exc:
            raise WorkerError("E_SUBPROCESS_POLICY_LOCAL_PATH") from exc
        return resolved

    def install(self) -> None:
        sys.addaudithook(self.check)


def sanitized_environment() -> dict[str, str]:
    allowed = {"PATH", "PATHEXT", "SYSTEMROOT", "WINDIR", "TEMP", "TMP", "COMSPEC"}
    result = {key: value for key, value in os.environ.items() if key.upper() in allowed}
    result["LOCALAPPDATA"] = str(validated_local_app_data())
    result["YTDLP_NO_PLUGINS"] = "1"
    return result


def _run_local(
    command: Sequence[str],
    timeout: int,
    *,
    capture_stdout: bool = False,
) -> subprocess.CompletedProcess[bytes]:
    return subprocess.run(
        list(command),
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE if capture_stdout else subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
        timeout=timeout,
        env=sanitized_environment(),
        creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
    )


def remux_single_to_mkv(ffmpeg: Path, source: Path, destination: Path) -> None:
    if destination.exists():
        raise WorkerError("E_COLLISION")
    try:
        result = _run_local(
            [
            str(ffmpeg),
            "-nostdin",
            "-v",
            "error",
            "-i",
            str(source),
            "-map",
            "0:v:0",
            "-map",
            "0:a:0",
            "-c",
            "copy",
            "-map_metadata",
            "-1",
            "-f",
            "matroska",
            str(destination),
            ],
            timeout=BRIDGE_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired as exc:
        destination.unlink(missing_ok=True)
        raise WorkerError("E_MERGE") from exc
    if result.returncode != 0 or not destination.is_file() or destination.stat().st_size <= 0:
        destination.unlink(missing_ok=True)
        raise WorkerError("E_MERGE")


def merge_local_streams(ffmpeg: Path, video: Path, audio: Path, destination: Path) -> None:
    if destination.exists():
        raise WorkerError("E_COLLISION")
    try:
        result = _run_local(
            [
            str(ffmpeg),
            "-nostdin",
            "-v",
            "error",
            "-i",
            str(video),
            "-i",
            str(audio),
            "-map",
            "0:v:0",
            "-map",
            "1:a:0",
            "-c",
            "copy",
            "-map_metadata",
            "-1",
            "-f",
            "matroska",
            str(destination),
            ],
            timeout=BRIDGE_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired as exc:
        destination.unlink(missing_ok=True)
        raise WorkerError("E_MERGE") from exc
    if result.returncode != 0 or not destination.is_file() or destination.stat().st_size <= 0:
        destination.unlink(missing_ok=True)
        raise WorkerError("E_MERGE")


def probe_mkv(ffprobe: Path, candidate: Path, job: dict[str, Any] | None = None) -> None:
    try:
        result = _run_local(
            [
            str(ffprobe),
            "-v",
            "error",
            "-show_entries",
            "format=format_name,duration:stream=codec_type",
            "-of",
            "json",
            str(candidate),
            ],
            timeout=120,
            capture_stdout=True,
        )
    except subprocess.TimeoutExpired as exc:
        raise WorkerError("E_MEDIA_VALIDATION") from exc
    try:
        payload = json.loads(result.stdout.decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise WorkerError("E_MEDIA_VALIDATION") from exc
    stream_types = {item.get("codec_type") for item in payload.get("streams", []) if isinstance(item, dict)}
    format_name = payload.get("format", {}).get("format_name", "")
    if result.returncode != 0 or {"video", "audio"} - stream_types or "matroska" not in format_name:
        raise WorkerError("E_MEDIA_VALIDATION")
    if job is not None:
        raw_duration = payload.get("format", {}).get("duration")
        if isinstance(raw_duration, str) and re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", raw_duration):
            raw_duration = float(raw_duration)
        duration_ms = round(_finite_number(raw_duration) * 1000)
        if abs(duration_ms - job["expected_duration_ms"]) > duration_tolerance_ms(job["expected_duration_ms"]):
            raise WorkerError("E_DURATION")


def validate_unique_candidate(run_directory: Path) -> Path:
    forbidden_suffixes = {".part", ".tmp", ".crdownload", ".txt", ".json", ".url"}
    files = [item for item in run_directory.iterdir() if item.is_file()]
    if any(item.suffix.casefold() in forbidden_suffixes for item in files):
        raise WorkerError("E_STAGE")
    candidates = [item for item in files if item.suffix.casefold() == ".mkv"]
    if len(candidates) != 1 or len(files) != 1:
        raise WorkerError("E_STAGE")
    return candidates[0]


def _ordinary_exact_file(path: Path, parent: Path) -> os.stat_result:
    try:
        if path.parent.resolve(strict=True) != parent.resolve(strict=True):
            raise WorkerError("E_COLLISION")
        _reject_reparse_path(path, parent)
        value = path.lstat()
    except OSError as exc:
        raise WorkerError("E_COLLISION") from exc
    if not stat.S_ISREG(value.st_mode) or _is_reparse(path):
        raise WorkerError("E_COLLISION")
    return value


def _stable_file_identity(value: os.stat_result) -> tuple[int, ...]:
    """Return cross-API file identity fields; content is bound separately by SHA."""
    return (
        value.st_dev,
        value.st_ino,
        value.st_mode,
        value.st_nlink,
        value.st_size,
    )


def _stable_pair_test_seam(_: str) -> None:
    """Named no-op seams used only by production-shaped race regressions."""
    return None


def _completion_test_seam(_: str) -> None:
    """Named no-op seams for durable completion transaction regressions."""
    return None


def _canonical_json_line(value: Mapping[str, Any]) -> bytes:
    return json.dumps(
        dict(value), ensure_ascii=False, allow_nan=False, separators=(",", ":")
    ).encode("utf-8") + b"\n"


_FORMAL_PRIOR_REQUIRED_KEYS = frozenset({
    "stable_id", "creator_uid", "source_url", "published_at", "item_type", "status",
})
_FORMAL_PRIOR_STATUS_RE = re.compile(r"VIDEO_[A-Z0-9_]{1,127}\Z")
_FORMAL_LEGACY_BY_LINE = {row[0]: row for row in FORMAL_LEGACY_ROWS}
_FORMAL_LEGACY_INTEGER_UID_BY_LINE = {
    row[0]: row for row in FORMAL_LEGACY_INTEGER_UID_ROWS
}


def _formal_raw_lines(payload: bytes) -> list[bytes]:
    complete = payload if not payload or payload.endswith(b"\n") else payload[:payload.rfind(b"\n") + 1]
    return complete.splitlines()


def _validate_formal_legacy_manifest() -> None:
    if (
        FORMAL_PREFIX_BYTES != 103_766
        or FORMAL_PREFIX_LINES != 119
        or not re.fullmatch(r"[A-F0-9]{64}", FORMAL_PREFIX_SHA256)
        or not re.fullmatch(r"[1-9][0-9]{1,19}", FORMAL_LEGACY_CREATOR_UID)
        or len(FORMAL_LEGACY_ROWS) != 26
        or len(_FORMAL_LEGACY_BY_LINE) != len(FORMAL_LEGACY_ROWS)
        or len(FORMAL_LEGACY_INTEGER_UID_ROWS) != 1
        or len(_FORMAL_LEGACY_INTEGER_UID_BY_LINE)
        != len(FORMAL_LEGACY_INTEGER_UID_ROWS)
        or set(_FORMAL_LEGACY_BY_LINE) & set(_FORMAL_LEGACY_INTEGER_UID_BY_LINE)
    ):
        raise WorkerError("E_COMPLETION_FORMAL")
    for row in FORMAL_LEGACY_ROWS:
        if (
            not isinstance(row, tuple) or len(row) != 8
            or not isinstance(row[0], int) or not 1 <= row[0] <= FORMAL_PREFIX_LINES
            or not isinstance(row[1], int) or row[1] < 2
            or not isinstance(row[2], str) or not re.fullmatch(r"[A-F0-9]{64}", row[2])
            or not all(isinstance(value, str) and value for value in row[3:])
            or row[6] != "video" or _FORMAL_PRIOR_STATUS_RE.fullmatch(row[7]) is None
        ):
            raise WorkerError("E_COMPLETION_FORMAL")
    for row in FORMAL_LEGACY_INTEGER_UID_ROWS:
        if (
            not isinstance(row, tuple) or len(row) != 9
            or type(row[0]) is not int or not 1 <= row[0] <= FORMAL_PREFIX_LINES
            or type(row[1]) is not int or row[1] < 2
            or not isinstance(row[2], str) or not re.fullmatch(r"[A-F0-9]{64}", row[2])
            or type(row[3]) is not int or row[3] <= 0
            or str(row[3]) != FORMAL_LEGACY_CREATOR_UID
            or not all(isinstance(value, str) and value for value in row[4:])
            or row[7] != "video" or _FORMAL_PRIOR_STATUS_RE.fullmatch(row[8]) is None
        ):
            raise WorkerError("E_COMPLETION_FORMAL")


def _validate_legacy_formal_prior(
    payload: bytes,
    raw_lines: Sequence[bytes],
    line_ordinal: int,
    value: dict[str, Any],
    config: HostConfig,
    job: dict[str, Any],
) -> str:
    _validate_formal_legacy_manifest()
    if (
        len(payload) < FORMAL_PREFIX_BYTES
        or len(raw_lines) < FORMAL_PREFIX_LINES
        or hashlib.sha256(payload[:FORMAL_PREFIX_BYTES]).hexdigest().upper() != FORMAL_PREFIX_SHA256
        or payload[:FORMAL_PREFIX_BYTES].count(b"\n") != FORMAL_PREFIX_LINES
        or not payload[:FORMAL_PREFIX_BYTES].endswith(b"\n")
        or job["creator_uid"] != FORMAL_LEGACY_CREATOR_UID
    ):
        raise WorkerError("E_COMPLETION_FORMAL")
    expected = _FORMAL_LEGACY_BY_LINE.get(line_ordinal)
    if expected is None:
        raise WorkerError("E_COMPLETION_FORMAL")
    raw_line = raw_lines[line_ordinal - 1]
    if (
        len(raw_line) != expected[1]
        or hashlib.sha256(raw_line).hexdigest().upper() != expected[2]
        or set(_FORMAL_PRIOR_REQUIRED_KEYS) - set(value) != {"creator_uid"}
        or value.get("stable_id") != expected[3]
        or value.get("source_url") != expected[4]
        or value.get("published_at") != expected[5]
        or value.get("item_type") != expected[6]
        or value.get("status") != expected[7]
        or value.get("schema_version") != 1
        or value.get("creator") != config.creator_name
        or expected[3] != job["bvid"]
        or expected[4] != job["canonical_url"]
        or expected[5] != job["published_at"]
    ):
        raise WorkerError("E_COMPLETION_FORMAL")
    return expected[7]


def _validate_current_formal_prior(
    value: dict[str, Any], config: HostConfig, job: dict[str, Any]
) -> str:
    if (
        not _FORMAL_PRIOR_REQUIRED_KEYS.issubset(value)
        or type(value.get("schema_version")) is not int
        or value["schema_version"] != 1
        or value.get("creator") != config.creator_name
        or not isinstance(value.get("stable_id"), str)
        or not isinstance(value.get("creator_uid"), str)
        or not isinstance(value.get("source_url"), str)
        or not isinstance(value.get("published_at"), str)
        or not isinstance(value.get("item_type"), str)
        or not isinstance(value.get("status"), str)
        or value["stable_id"] != job["bvid"]
        or value["creator_uid"] != job["creator_uid"]
        or value["source_url"] != job["canonical_url"]
        or value["published_at"] != job["published_at"]
        or value["item_type"] != "video"
        or _FORMAL_PRIOR_STATUS_RE.fullmatch(value["status"]) is None
    ):
        raise WorkerError("E_COMPLETION_FORMAL")
    return value["status"]


def _validate_legacy_integer_uid_formal_prior(
    payload: bytes,
    raw_lines: Sequence[bytes],
    line_ordinal: int,
    value: dict[str, Any],
    config: HostConfig,
    job: dict[str, Any],
) -> str:
    _validate_formal_legacy_manifest()
    if (
        len(payload) < FORMAL_PREFIX_BYTES
        or len(raw_lines) < FORMAL_PREFIX_LINES
        or hashlib.sha256(payload[:FORMAL_PREFIX_BYTES]).hexdigest().upper()
        != FORMAL_PREFIX_SHA256
        or payload[:FORMAL_PREFIX_BYTES].count(b"\n") != FORMAL_PREFIX_LINES
        or not payload[:FORMAL_PREFIX_BYTES].endswith(b"\n")
    ):
        raise WorkerError("E_COMPLETION_FORMAL")
    expected = _FORMAL_LEGACY_INTEGER_UID_BY_LINE.get(line_ordinal)
    if expected is None:
        raise WorkerError("E_COMPLETION_FORMAL")
    raw_line = raw_lines[line_ordinal - 1]
    if (
        len(raw_line) != expected[1]
        or hashlib.sha256(raw_line).hexdigest().upper() != expected[2]
        or not _FORMAL_PRIOR_REQUIRED_KEYS.issubset(value)
        or type(value.get("schema_version")) is not int
        or value["schema_version"] != 1
        or value.get("creator") != config.creator_name
        or type(value.get("creator_uid")) is not int
        or value["creator_uid"] != expected[3]
        or value.get("stable_id") != expected[4]
        or value.get("source_url") != expected[5]
        or value.get("published_at") != expected[6]
        or value.get("item_type") != expected[7]
        or value.get("status") != expected[8]
        or str(expected[3]) != job["creator_uid"]
        or expected[4] != job["bvid"]
        or expected[5] != job["canonical_url"]
        or expected[6] != job["published_at"]
    ):
        raise WorkerError("E_COMPLETION_FORMAL")
    return expected[8]


def _read_jsonl_objects(path: Path, *, limit: int, error_code: str) -> tuple[bytes, list[dict[str, Any]]]:
    try:
        if path.exists():
            parent = path.parent.resolve(strict=True)
            _ordinary_exact_file(path, parent)
            payload = path.read_bytes()
        else:
            payload = b""
        if len(payload) > limit:
            raise WorkerError(error_code)
        complete = payload if not payload or payload.endswith(b"\n") else payload[:payload.rfind(b"\n") + 1]
        records: list[dict[str, Any]] = []
        for raw_line in complete.splitlines():
            if not raw_line:
                raise WorkerError(error_code)
            value = strict_json_loads(raw_line)
            records.append(value)
        return payload, records
    except WorkerError:
        raise
    except (OSError, ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
        raise WorkerError(error_code) from exc


def _append_jsonl_idempotent(
    path: Path,
    record: dict[str, Any],
    *,
    identity_key: str,
    identity_value: str,
    limit: int,
    error_code: str,
) -> None:
    expected = _canonical_json_line(record)
    payload, records = _read_jsonl_objects(path, limit=limit, error_code=error_code)
    matches = [value for value in records if value.get(identity_key) == identity_value]
    if len(matches) > 1 or (matches and matches[0] != record):
        raise WorkerError("E_COMPLETION_REPLAY")
    if matches:
        return
    suffix = b"" if not payload or payload.endswith(b"\n") else payload[payload.rfind(b"\n") + 1:]
    if suffix and (len(suffix) >= len(expected) or expected[:len(suffix)] != suffix):
        raise WorkerError(error_code)
    parent = path.parent.resolve(strict=True)
    try:
        if not path.exists():
            with path.open("xb") as created:
                created.flush()
                os.fsync(created.fileno())
        _ordinary_exact_file(path, parent)
        with path.open("r+b", buffering=0) as stream:
            current = stream.read()
            if current != payload:
                raise WorkerError(error_code)
            stream.seek(0, os.SEEK_END)
            stream.write(expected[len(suffix):])
            stream.flush()
            os.fsync(stream.fileno())
        _completion_test_seam(f"AFTER_{identity_key.upper()}_APPEND")
        final_payload, final_records = _read_jsonl_objects(path, limit=limit, error_code=error_code)
        if not final_payload.endswith(b"\n") or sum(
            value.get(identity_key) == identity_value and value == record for value in final_records
        ) != 1:
            raise WorkerError(error_code)
    except WorkerError:
        raise
    except OSError as exc:
        raise WorkerError(error_code) from exc


@contextmanager
def _completion_lock(path: Path) -> Iterable[None]:
    try:
        path.parent.resolve(strict=True)
        with path.open("a+b") as stream:
            if stream.seek(0, os.SEEK_END) == 0:
                stream.write(b"\0")
                stream.flush()
                os.fsync(stream.fileno())
            _ordinary_exact_file(path, path.parent)
            stream.seek(0)
            if os.name == "nt":
                import msvcrt  # noqa: PLC0415

                msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
                try:
                    yield
                finally:
                    stream.seek(0)
                    msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
            else:
                import fcntl  # noqa: PLC0415

                fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                try:
                    yield
                finally:
                    fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
    except WorkerError:
        raise
    except OSError as exc:
        raise WorkerError("E_COMPLETION_BUSY") from exc


def _commit_formal_and_handoff(
    config: HostConfig,
    job: dict[str, Any],
    formal_name: str,
    mapping_name: str,
    persisted: dict[str, Any],
    *,
    media_complete_acknowledged: bool = False,
) -> None:
    """Idempotently close formal publication and processing handoff before COMPLETE."""
    if (
        not isinstance(config.queue_lock_path, Path)
        or not isinstance(config.formal_manifest_path, Path)
        or not isinstance(config.processing_handoff_path, Path)
        or not isinstance(config.creator_name, str) or not config.creator_name
    ):
        raise WorkerError("E_COMPLETION_CONFIG")
    handoff_id = f"HANDOFF-BILI-MEDIA-{job['job_id'][:32].upper()}"
    with _completion_lock(config.queue_lock_path):
        formal_payload, formal_records = _read_jsonl_objects(
            config.formal_manifest_path, limit=32 * 1024 * 1024, error_code="E_COMPLETION_FORMAL"
        )
        raw_lines = _formal_raw_lines(formal_payload)
        if len(raw_lines) != len(formal_records):
            raise WorkerError("E_COMPLETION_FORMAL")
        prior_status = None
        seen_prior_identities: set[tuple[str, str, str, str, str, str]] = set()
        for line_ordinal, value in enumerate(formal_records, 1):
            if value.get("queue_job_id") == job["job_id"]:
                continue
            stable_matches = value.get("stable_id") == job["bvid"]
            source_matches = value.get("source_url") == job["canonical_url"]
            if not stable_matches and not source_matches:
                continue
            item_type = value.get("item_type")
            if item_type == "video_transcript" and stable_matches and source_matches:
                continue
            if not stable_matches or not source_matches:
                raise WorkerError("E_COMPLETION_FORMAL")
            missing = _FORMAL_PRIOR_REQUIRED_KEYS - set(value)
            if line_ordinal in _FORMAL_LEGACY_INTEGER_UID_BY_LINE:
                prior_status = _validate_legacy_integer_uid_formal_prior(
                    formal_payload, raw_lines, line_ordinal, value, config, job
                )
                creator_uid = job["creator_uid"]
            elif missing == {"creator_uid"}:
                prior_status = _validate_legacy_formal_prior(
                    formal_payload, raw_lines, line_ordinal, value, config, job
                )
                creator_uid = job["creator_uid"]
            else:
                prior_status = _validate_current_formal_prior(value, config, job)
                creator_uid = value["creator_uid"]
            prior_identity = (
                value["stable_id"], creator_uid, value["source_url"],
                value["published_at"], value["item_type"], prior_status,
            )
            if prior_identity in seen_prior_identities:
                raise WorkerError("E_COMPLETION_FORMAL")
            seen_prior_identities.add(prior_identity)
        if prior_status is None and media_complete_acknowledged is not True:
            # A first formal row has no historical status to supersede.  It is
            # permitted only after the typed Host ACK proves the governed queue
            # job's exact MEDIA_COMPLETE identity is already durable.
            raise WorkerError("E_COMPLETION_FORMAL")
        formal_record = {
            "schema_version": 1,
            "creator": config.creator_name,
            "creator_uid": job["creator_uid"],
            "item_type": "video",
            "stable_id": job["bvid"],
            "title": job["title"],
            "source_url": job["canonical_url"],
            "published_at": job["published_at"],
            "collected_at": persisted["completed_at"],
            "status": "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT",
            "video_path": str(config.destination / formal_name),
            "mapping_path": str(config.destination / mapping_name),
            "bytes": persisted["bytes"],
            "sha256": persisted["sha256"],
            "duration_seconds": persisted["duration_seconds"],
            "video_codec": persisted["video_codec"],
            "audio_codec": persisted["audio_codec"],
            "processing_handoff_id": handoff_id,
            "queue_job_id": job["job_id"],
        }
        if prior_status is not None:
            formal_record["supersedes_status"] = prior_status
        handoff_record = {
            "schema": 1,
            "type": "media-processing-handoff",
            "status": "READY",
            "handoff_id": handoff_id,
            "queue_job_id": job["job_id"],
            "creator_uid": job["creator_uid"],
            "bvid": job["bvid"],
            "source_url": job["canonical_url"],
            "media_path": str(config.destination / formal_name),
            "mapping_path": str(config.destination / mapping_name),
            "bytes": persisted["bytes"],
            "sha256": persisted["sha256"],
            "duration_seconds": persisted["duration_seconds"],
            "video_codec": persisted["video_codec"],
            "audio_codec": persisted["audio_codec"],
            "created_at": persisted["completed_at"],
        }
        _append_jsonl_idempotent(
            config.formal_manifest_path, formal_record,
            identity_key="queue_job_id", identity_value=job["job_id"],
            limit=32 * 1024 * 1024, error_code="E_COMPLETION_FORMAL",
        )
        _completion_test_seam("BETWEEN_FORMAL_AND_HANDOFF")
        _append_jsonl_idempotent(
            config.processing_handoff_path, handoff_record,
            identity_key="queue_job_id", identity_value=job["job_id"],
            limit=8 * 1024 * 1024, error_code="E_COMPLETION_HANDOFF",
        )
        _completion_test_seam("BEFORE_COMPLETION_RETURN")


@dataclass
class _LockedPublishedFile:
    """Read-only handle whose sharing mode denies writers, deletion and replacement."""

    path: Path
    parent: Path
    error_code: str
    stream: Any
    identity: tuple[int, ...]

    @classmethod
    def open(cls, path: Path, parent: Path, error_code: str) -> "_LockedPublishedFile":
        stream: Any | None = None
        try:
            before = _ordinary_exact_file(path, parent)
            if os.name != "nt":
                # The deployed Host is Windows-only. Keep non-Windows imports
                # fail-closed while retaining a no-follow advisory read lock.
                import fcntl  # noqa: PLC0415

                flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
                descriptor = os.open(path, flags)
                try:
                    fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
                    stream = os.fdopen(descriptor, "rb", closefd=True)
                except BaseException:
                    os.close(descriptor)
                    raise
            else:
                import msvcrt  # noqa: PLC0415

                kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
                create_file = kernel32.CreateFileW
                create_file.argtypes = (
                    ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
                    ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
                )
                create_file.restype = ctypes.c_void_p
                close_handle = kernel32.CloseHandle
                close_handle.argtypes = (ctypes.c_void_p,)
                close_handle.restype = ctypes.c_int
                handle = create_file(
                    str(path),
                    0x80000000,  # GENERIC_READ
                    0x00000001,  # FILE_SHARE_READ: deny write/delete/path replacement
                    None,
                    3,  # OPEN_EXISTING
                    0x00000080 | 0x00200000 | 0x08000000,
                    # FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OPEN_REPARSE_POINT |
                    # FILE_FLAG_SEQUENTIAL_SCAN
                    None,
                )
                if handle in (None, ctypes.c_void_p(-1).value):
                    code = ctypes.get_last_error()
                    raise OSError(code, ctypes.FormatError(code), str(path))
                descriptor: int | None = None
                try:
                    descriptor = msvcrt.open_osfhandle(
                        int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0)
                    )
                    handle = None
                    stream = os.fdopen(descriptor, "rb", closefd=True)
                    descriptor = None
                finally:
                    if descriptor is not None:
                        os.close(descriptor)
                    if handle is not None:
                        close_handle(handle)
            after = _ordinary_exact_file(path, parent)
            identity = _stable_file_identity(os.fstat(stream.fileno()))
            if identity != _stable_file_identity(before) or identity != _stable_file_identity(after):
                raise WorkerError(error_code)
            return cls(path=path, parent=parent, error_code=error_code, stream=stream, identity=identity)
        except BaseException as exc:
            if stream is not None:
                stream.close()
            if isinstance(exc, WorkerError) and exc.code == error_code:
                raise
            if isinstance(exc, (KeyboardInterrupt, SystemExit)):
                raise
            raise WorkerError(error_code) from exc

    def close(self) -> None:
        self.stream.close()

    def assert_path_identity(self) -> os.stat_result:
        try:
            path_stat = _ordinary_exact_file(self.path, self.parent)
            handle_stat = os.fstat(self.stream.fileno())
        except (OSError, WorkerError) as exc:
            raise WorkerError(self.error_code) from exc
        if (
            _stable_file_identity(path_stat) != self.identity
            or _stable_file_identity(handle_stat) != self.identity
        ):
            raise WorkerError(self.error_code)
        return handle_stat

    def read_all(self, maximum: int) -> bytes:
        try:
            self.stream.seek(0)
            payload = self.stream.read(maximum + 1)
            if len(payload) > maximum or self.stream.read(1) != b"":
                raise WorkerError(self.error_code)
            self.assert_path_identity()
            return payload
        except WorkerError:
            raise
        except OSError as exc:
            raise WorkerError(self.error_code) from exc

    def sha256(self) -> str:
        digest = hashlib.sha256()
        try:
            self.stream.seek(0)
            for chunk in iter(lambda: self.stream.read(1024 * 1024), b""):
                digest.update(chunk)
            self.assert_path_identity()
            return digest.hexdigest()
        except WorkerError:
            raise
        except OSError as exc:
            raise WorkerError(self.error_code) from exc


def _bridge_number(value: Any) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
    number = float(value)
    if not math.isfinite(number) or number < 0:
        raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
    return number


def _bridge_remote_matches_expected(expected_duration_ms: int, remote: float) -> bool:
    expected = expected_duration_ms / 1000
    if abs(remote - expected) <= 0.001:
        return True
    return (
        expected_duration_ms % 1000 == 0
        and 0 < expected - remote < 1
        and math.ceil(remote) == int(expected)
    )


def _validate_complete_bridge_item(
    item: Any,
    job: dict[str, Any],
    *,
    persisted: bool,
) -> dict[str, Any]:
    if not isinstance(item, dict):
        raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
    expected_keys = _BRIDGE_MAPPING_KEYS if persisted else _BRIDGE_ITEM_KEYS
    if set(item) != expected_keys:
        if persisted or set(item) != _BRIDGE_ITEM_WARNING_KEYS:
            raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
        warning = item["warning"]
        if (
            not isinstance(warning, str)
            or _BRIDGE_CLEANUP_WARNING_RE.fullmatch(warning) is None
        ):
            raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
    if not persisted and item["status"] != "COMPLETE":
        raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
    if (
        item["schema_version"] != "1.0"
        or item["bvid"] != job["bvid"]
        or item["source"] != job["canonical_url"]
        or item["published_at"] != job["published_at"]
        or item["local_file"] != f"{job['bvid']}.mkv"
        or item["acquisition_mode"] != "authorized_browser_file_handoff"
        or not isinstance(item["title"], str)
        or not 1 <= len(item["title"]) <= 1024
        or any(ord(character) < 0x20 for character in item["title"])
        or not isinstance(item["bytes"], int)
        or isinstance(item["bytes"], bool)
        or item["bytes"] <= 0
        or not isinstance(item["sha256"], str)
        or _LOWER_SHA256_RE.fullmatch(item["sha256"]) is None
        or item["handoff_source_sha256"] != item["sha256"]
        or not isinstance(item["format_name"], str)
        or "matroska" not in item["format_name"].split(",")
        or not isinstance(item["video_codec"], str)
        or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", item["video_codec"])
        or not isinstance(item["audio_codec"], str)
        or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", item["audio_codec"])
        or not isinstance(item["completed_at"], str)
        or _UTC_ISO_RE.fullmatch(item["completed_at"]) is None
    ):
        raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
    duration = _bridge_number(item["duration_seconds"])
    remote = _bridge_number(item["remote_duration_seconds"])
    local = _bridge_number(item["local_duration_seconds"])
    delta = _bridge_number(item["duration_delta_seconds"])
    tolerance = _bridge_number(item["duration_tolerance_seconds"])
    expected_tolerance = max(3.0, remote * 0.001)
    if (
        not _bridge_remote_matches_expected(job["expected_duration_ms"], remote)
        or abs(duration - local) > 1e-9
        or abs(delta - abs(local - remote)) > 1e-9
        or abs(tolerance - expected_tolerance) > 1e-9
        or delta > tolerance
    ):
        raise WorkerError("E_BRIDGE_DURATION_SHA")
    return item


def _read_exact_published_bridge_result(
    config: HostConfig,
    job: dict[str, Any],
    *,
    expected_item: dict[str, Any] | None = None,
    on_media_verified: Callable[[str, str, dict[str, Any], dict[str, Any]], None] | None = None,
    on_verified: Callable[[str, str, dict[str, Any]], None] | None = None,
) -> tuple[str, str] | None:
    destination = config.destination.resolve(strict=True)
    formal_name = f"{job['bvid']}.mkv"
    mapping_name = f"{job['bvid']}.download.json"
    formal_path = destination / formal_name
    mapping_path = destination / mapping_name
    present = (formal_path.exists(), mapping_path.exists())
    if present == (False, False):
        return None
    if present != (True, True):
        raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
    formal_lock = _LockedPublishedFile.open(formal_path, destination, "E_BRIDGE_DURATION_SHA")
    try:
        mapping_lock = _LockedPublishedFile.open(
            mapping_path, destination, "E_BRIDGE_MAPPING_READBACK"
        )
        try:
            _stable_pair_test_seam("LOCKS_ACQUIRED")
            mapping_stat = mapping_lock.assert_path_identity()
            if mapping_stat.st_size <= 0 or mapping_stat.st_size > 64 * 1024:
                raise WorkerError("E_BRIDGE_MAPPING_READBACK")
            mapping_bytes = mapping_lock.read_all(64 * 1024)
            try:
                persisted = _validate_complete_bridge_item(
                    strict_json_loads(mapping_bytes), job, persisted=True
                )
            except WorkerError as exc:
                if exc.code == "E_BRIDGE_DURATION_SHA":
                    raise
                raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
            except (ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
                raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
            if expected_item is not None:
                if {key: expected_item[key] for key in _BRIDGE_MAPPING_KEYS} != persisted:
                    raise WorkerError("E_BRIDGE_MAPPING_READBACK")
            formal_stat = formal_lock.assert_path_identity()
            formal_sha = formal_lock.sha256().casefold()
            if persisted["bytes"] != formal_stat.st_size or persisted["sha256"] != formal_sha:
                raise WorkerError("E_BRIDGE_DURATION_SHA")
            _stable_pair_test_seam("AFTER_INITIAL_PAIR")

            # The two Windows handles deny write/delete/path replacement while
            # FFprobe opens its read-only view. Hashing and JSON parsing use these
            # same handles, and the consumer commit executes before handle release.
            probe_mkv(config.ffprobe, formal_path, job)
            formal_lock.assert_path_identity()
            mapping_lock.assert_path_identity()
            final_formal_sha = formal_lock.sha256().casefold()
            _stable_pair_test_seam("AFTER_FINAL_MEDIA_HASH")
            final_mapping_bytes = mapping_lock.read_all(64 * 1024)
            _stable_pair_test_seam("AFTER_FINAL_MAPPING_READ")
            if final_formal_sha != persisted["sha256"]:
                raise WorkerError("E_BRIDGE_DURATION_SHA")
            if final_mapping_bytes != mapping_bytes:
                raise WorkerError("E_BRIDGE_MAPPING_READBACK")
            try:
                final_persisted = _validate_complete_bridge_item(
                    strict_json_loads(final_mapping_bytes), job, persisted=True
                )
            except (WorkerError, ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
                raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
            if final_persisted != persisted:
                raise WorkerError("E_BRIDGE_MAPPING_READBACK")
            formal_lock.assert_path_identity()
            mapping_lock.assert_path_identity()
            _stable_pair_test_seam("BEFORE_COMMIT")
            media_identity = validate_media_complete_identity(
                {
                    "formal_filename": formal_name,
                    "mapping_filename": mapping_name,
                    "media_bytes": persisted["bytes"],
                    "media_sha256": persisted["sha256"].upper(),
                    "mapping_bytes": len(final_mapping_bytes),
                    "mapping_sha256": hashlib.sha256(final_mapping_bytes).hexdigest().upper(),
                    "duration_milliseconds": int(round(
                        float(persisted["local_duration_seconds"]) * 1_000
                    )),
                    "video_codec": persisted["video_codec"],
                    "audio_codec": persisted["audio_codec"],
                },
                job,
            )
            if on_media_verified is not None:
                on_media_verified(formal_name, mapping_name, persisted, media_identity)
            if on_verified is not None:
                on_verified(formal_name, mapping_name, persisted)
            return formal_name, mapping_name
        finally:
            mapping_lock.close()
    finally:
        formal_lock.close()


def recover_published_task(
    config: HostConfig,
    job: dict[str, Any],
    *,
    cancel_check: Callable[[], bool],
    report: Callable[..., None],
    commit_begin: Callable[[], None],
) -> tuple[str, str]:
    """Verify and consume one exact published pair before secret transfer."""
    if job["creator_uid"] not in config.creator_allowlist:
        raise WorkerError("E_ALLOWLIST")
    committed: tuple[str, str] | None = None

    def media_verified(
        _formal_name: str, _mapping_name: str, _persisted: dict[str, Any],
        media_identity: dict[str, Any],
    ) -> None:
        report("MEDIA_COMPLETE", 100, media_identity)

    def consume(formal_name: str, mapping_name: str, persisted: dict[str, Any]) -> None:
        nonlocal committed
        if cancel_check():
            raise CancelRequested()
        report("POSTPROCESS_PENDING", 100)
        commit_begin()
        _commit_formal_and_handoff(
            config, job, formal_name, mapping_name, persisted,
            media_complete_acknowledged=True,
        )
        committed = (formal_name, mapping_name)

    recovered = _read_exact_published_bridge_result(
        config, job, on_media_verified=media_verified, on_verified=consume
    )
    if recovered is None:
        raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
    if committed != recovered:
        raise WorkerError("E_BRIDGE_MAPPING_READBACK")
    return recovered


def _bridge_command(
    config: HostConfig, candidate: Path, job: dict[str, Any], batch_json: Path
) -> list[str]:
    return [
        str(config.bridge_python),
        str(config.bridge_script),
        "--input",
        str(batch_json),
        "accept-browser-file",
        "--bvid",
        job["bvid"],
        "--media-file",
        str(candidate),
        "--destination",
        str(config.destination),
        "--ffprobe",
        str(config.ffprobe),
        "--expected-duration-ms",
        str(job["expected_duration_ms"]),
    ]


def run_frozen_bridge(
    config: HostConfig,
    candidate: Path,
    job: dict[str, Any],
    *,
    on_media_verified: Callable[[str, str, dict[str, Any], dict[str, Any]], None] | None = None,
    on_verified: Callable[[str, str, dict[str, Any]], None] | None = None,
) -> tuple[str, str]:
    started = time.monotonic()

    def failure(code: str, state: str, reason: str) -> WorkerError:
        return WorkerError(code, {
            "attempts": 1,
            "elapsed_ms": max(0, min(7_200_000, int((time.monotonic() - started) * 1000))),
            "state": state,
            "reason": reason,
        })

    batch_json = candidate.parent / "bridge-input.json"
    batch_payload = {
        "schema_version": "1.0",
        "batch_id": f"generic-{job['job_id'][:16]}",
        "items": [{
            "bvid": job["bvid"],
            "source_url": job["canonical_url"],
            "published_at": job["published_at"],
            "title": job["title"],
            "expected_duration_ms": job["expected_duration_ms"],
        }],
    }
    try:
        with batch_json.open("xb") as stream:
            stream.write((json.dumps(batch_payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8"))
            stream.flush()
            os.fsync(stream.fileno())
    except BaseException as exc:
        batch_json.unlink(missing_ok=True)
        raise failure(
            "E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "BATCH_RECEIPT_CREATE_FAILED"
        ) from exc
    try:
        result = _run_local(
            _bridge_command(config, candidate, job, batch_json),
            BRIDGE_TIMEOUT_SECONDS,
            capture_stdout=True,
        )
    except subprocess.TimeoutExpired as exc:
        raise failure(
            "E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "SUBPROCESS_TIMEOUT"
        ) from exc
    except OSError as exc:
        raise failure(
            "E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "SUBPROCESS_INVOCATION_FAILED"
        ) from exc
    finally:
        batch_json.unlink(missing_ok=True)
    if result.returncode != 0:
        try:
            stopped = strict_json_loads(result.stdout)
        except (ProtocolError, UnicodeError, json.JSONDecodeError):
            stopped = None
        bridge_stops = {
            "E_BRIDGE_SOURCE_STABILITY": (
                "BRIDGE_SOURCE_STABILITY", "SOURCE_FILE_INVALID"
            ),
            "E_BRIDGE_METADATA_BINDING": (
                "BRIDGE_METADATA_BINDING", "EXPECTED_METADATA_MISMATCH"
            ),
            "E_BRIDGE_FFPROBE": (
                "BRIDGE_FFPROBE", "LOCAL_MEDIA_PROBE_FAILED"
            ),
            "E_BRIDGE_DURATION_SHA": (
                "DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH"
            ),
            "E_BRIDGE_PUBLISH": (
                "CREATE_NEW_PUBLISH", "BRIDGE_REPORTED_PUBLISH_FAILURE"
            ),
        }
        if (
            isinstance(stopped, dict)
            and set(stopped) == {"result", "error_code"}
            and stopped.get("result") == "SAFETY_STOP"
            and stopped.get("error_code") in bridge_stops
        ):
            code = stopped["error_code"]
            state, reason = bridge_stops[code]
            raise failure(code, state, reason)
        raise failure("E_BRIDGE_EXIT", "BRIDGE_EXIT", "NONZERO_EXIT")

    def read_published(expected_item: dict[str, Any] | None) -> tuple[str, str]:
        try:
            published_value = _read_exact_published_bridge_result(
                config, job, expected_item=expected_item,
                on_media_verified=on_media_verified, on_verified=on_verified,
            )
        except WorkerError as exc:
            if exc.code.startswith("E_COMPLETION_"):
                raise
            code = exc.code if exc.code.startswith("E_BRIDGE_") else "E_BRIDGE_MAPPING_READBACK"
            state, reason = {
                "E_BRIDGE_MEDIA_MAPPING": ("MEDIA_MAPPING_PRESENCE", "MEDIA_OR_MAPPING_MISSING"),
                "E_BRIDGE_DURATION_SHA": ("DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH"),
            }.get(code, ("MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"))
            raise failure(code, state, reason) from exc
        if published_value is None:
            raise failure(
                "E_BRIDGE_MEDIA_MAPPING", "MEDIA_MAPPING_PRESENCE", "MEDIA_OR_MAPPING_MISSING"
            )
        return published_value

    def preserve_verified_media_before_output_failure(_cause: BaseException | None = None) -> bool:
        """Persist media truth, but do not commit postprocess through an invalid wire result."""
        try:
            if os.path.lexists(config.destination / ".bili-download-staging"):
                raise failure(
                    "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
                )
            published_value = _read_exact_published_bridge_result(
                config, job, on_media_verified=on_media_verified
            )
            if published_value is None:
                raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
            return True
        except WorkerError as exc:
            if exc.code.startswith("E_COMPLETION_"):
                raise
            return False

    try:
        payload = strict_json_loads(result.stdout)
    except (ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
        preserve_verified_media_before_output_failure(exc)
        raise failure(
            "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
        ) from exc
    if (
        not isinstance(payload, dict)
        or set(payload) != {"batch_id", "command", "result", "success_count", "failure_count", "items"}
        or payload.get("batch_id") != f"generic-{job['job_id'][:16]}"
        or payload.get("result") != "PASS"
        or payload.get("command") != "accept-browser-file"
        or payload.get("success_count") != 1
        or payload.get("failure_count") != 0
    ):
        preserve_verified_media_before_output_failure()
        raise failure(
            "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
        )
    items = payload.get("items")
    if not isinstance(items, list) or len(items) != 1 or not isinstance(items[0], dict):
        preserve_verified_media_before_output_failure()
        raise failure(
            "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
        )
    try:
        item = _validate_complete_bridge_item(items[0], job, persisted=False)
    except WorkerError as exc:
        preserve_verified_media_before_output_failure(exc)
        code = exc.code if exc.code == "E_BRIDGE_DURATION_SHA" else "E_BRIDGE_OUTPUT_SCHEMA"
        state, reason = (
            ("DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH")
            if code == "E_BRIDGE_DURATION_SHA"
            else ("BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID")
        )
        raise failure(code, state, reason) from exc
    if "warning" in item:
        try:
            if os.path.lexists(config.destination / ".bili-download-staging"):
                raise failure(
                    "E_BRIDGE_MAPPING_READBACK", "MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"
                )
        except OSError as exc:
            raise failure(
                "E_BRIDGE_MAPPING_READBACK", "MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"
            ) from exc
    return read_published(item)


def ytdlp_options(
    config: HostConfig,
    run_directory: Path,
    cookie_stream: io.StringIO,
    progress_hook: Callable[[dict[str, Any]], None],
) -> dict[str, Any]:
    return {
        "cookiefile": cookie_stream,
        "format": "bestvideo+bestaudio/best",
        "merge_output_format": "mkv",
        "outtmpl": str(run_directory / "%(id)s.%(format_id)s.%(ext)s"),
        "noplaylist": True,
        "continuedl": False,
        "overwrites": False,
        "cachedir": False,
        "quiet": True,
        "no_warnings": True,
        "logger": NullLogger(),
        "progress_hooks": [progress_hook],
        "postprocessor_hooks": [progress_hook],
        "socket_timeout": SOCKET_TIMEOUT_SECONDS,
        "extractor_retries": EXTRACTOR_RETRIES,
        "retries": HTTP_RETRIES,
        "fragment_retries": FRAGMENT_RETRIES,
        "file_access_retries": FILE_ACCESS_RETRIES,
        "retry_sleep_functions": {
            "http": lambda _attempt: 1,
            "fragment": lambda _attempt: 1,
            "file_access": lambda _attempt: 1,
            "extractor": lambda _attempt: 1,
        },
        "ffmpeg_location": str(config.ffmpeg),
        "writethumbnail": False,
        "writesubtitles": False,
        "writeautomaticsub": False,
        "writeinfojson": False,
        "writedescription": False,
        "writecomments": False,
        "getcomments": False,
        "allow_playlist_files": False,
        "external_downloader": {},
    }


def run_authenticated_task(
    start: dict[str, Any],
    config: HostConfig,
    *,
    cancel_check: Callable[[], bool],
    report: Callable[..., None],
    stage_root: Path | None = None,
    prepared_run_directory: Path | None = None,
    commit_begin: Callable[[], None] | None = None,
    closure_report: Callable[[bool], None] | None = None,
    recovery_required: bool = False,
) -> tuple[str, str, bool]:
    """Run the exact task.  The caller must already own this worker in a job."""
    validate_start(start)
    job = start["job"]
    if job["creator_uid"] not in config.creator_allowlist:
        raise WorkerError("E_ALLOWLIST")
    root = fixed_stage_root(job["bvid"]) if stage_root is None else stage_root.resolve()
    if prepared_run_directory is None:
        run_directory = prepare_run_directory(root)
    else:
        run_directory = prepared_run_directory.resolve(strict=True)
        _ensure_within(run_directory, root)
        if not run_directory.is_dir() or _is_reparse(run_directory):
            raise WorkerError("E_STAGE")
    cookie_stream: io.StringIO | None = None
    cookie_closed = False
    committed = False
    outcome: tuple[str, str, bool] | None = None
    try:
        def checkpoint() -> None:
            if cancel_check():
                raise CancelRequested()

        checkpoint()
        def consume_recovered(
            formal_name: str, mapping_name: str, persisted: dict[str, Any]
        ) -> None:
            nonlocal cookie_closed, committed, outcome
            checkpoint()
            report("POSTPROCESS_PENDING", 100)
            if commit_begin is not None:
                commit_begin()
            _commit_formal_and_handoff(
                config, job, formal_name, mapping_name, persisted,
                media_complete_acknowledged=True,
            )
            cookie_closed = True
            committed = True
            outcome = (formal_name, mapping_name, True)

        def recovered_media_verified(
            _formal_name: str, _mapping_name: str, _persisted: dict[str, Any],
            media_identity: dict[str, Any],
        ) -> None:
            report("MEDIA_COMPLETE", 100, media_identity)

        recovered = _read_exact_published_bridge_result(
            config, job, on_media_verified=recovered_media_verified,
            on_verified=consume_recovered,
        )
        if recovered is not None:
            if outcome != (recovered[0], recovered[1], True):
                raise WorkerError("E_BRIDGE_MAPPING_READBACK")
            return outcome
        if recovery_required:
            raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
        cookie_stream = build_cookie_stream(start)
        yt_dlp, _ = bootstrap_ytdlp()
        secret_values = [
            value
            for cookie in start["cookies"]
            for value in (cookie["name"], cookie["value"])
        ]
        policy = SubprocessPolicy(
            run_directory,
            {config.ffmpeg, config.ffprobe, config.bridge_python},
            secret_values,
        )
        policy.install()

        def progress_hook(status: dict[str, Any]) -> None:
            if cancel_check():
                raise CancelRequested()
            downloaded = status.get("downloaded_bytes")
            total = status.get("total_bytes") or status.get("total_bytes_estimate")
            progress = 0
            if isinstance(downloaded, (int, float)) and isinstance(total, (int, float)) and total > 0:
                progress = max(0, min(95, int(float(downloaded) * 95 / float(total))))
            report("DOWNLOADING", progress)

        opts = ytdlp_options(config, run_directory, cookie_stream, progress_hook)
        with yt_dlp.YoutubeDL(opts) as ydl:
            report("CHECKING", 0)
            download_info, single, signed_urls = prepare_download_info(ydl, job)
            policy.secrets.update(url.casefold() for url in signed_urls)
            checkpoint()
            ydl.process_info(download_info)
        checkpoint()
        report("MERGING", 96)
        files = [item for item in run_directory.iterdir() if item.is_file()]
        if single:
            originals = [item for item in files if item.suffix.casefold() not in {".part", ".tmp"}]
            if len(originals) != 1:
                raise WorkerError("E_STAGE")
            final = run_directory / "complete.mkv"
            checkpoint()
            remux_single_to_mkv(config.ffmpeg, originals[0], final)
            checkpoint()
            originals[0].unlink()
        else:
            mkv_candidates = [item for item in files if item.suffix.casefold() == ".mkv"]
            if len(mkv_candidates) != 1:
                raise WorkerError("E_STAGE")
            final = run_directory / "complete.mkv"
            if mkv_candidates[0] != final:
                if final.exists():
                    raise WorkerError("E_COLLISION")
                checkpoint()
                mkv_candidates[0].rename(final)
                checkpoint()
        checkpoint()
        candidate = validate_unique_candidate(run_directory)
        checkpoint()
        probe_mkv(config.ffprobe, candidate, job)
        checkpoint()
        cookie_closed = close_cookie_stream(cookie_stream)
        cookie_stream = None
        report("VALIDATING", 98)
        checkpoint()
        report("PUBLISHING", 99)
        if commit_begin is not None:
            commit_begin()
        checkpoint()
        completion_committed = False

        def consume_published(
            formal_name: str, mapping_name: str, persisted: dict[str, Any]
        ) -> None:
            nonlocal completion_committed
            checkpoint()
            report("POSTPROCESS_PENDING", 100)
            _commit_formal_and_handoff(
                config, job, formal_name, mapping_name, persisted,
                media_complete_acknowledged=True,
            )
            completion_committed = True

        formal, mapping = run_frozen_bridge(
            config, candidate, job,
            on_media_verified=lambda _formal, _mapping, _persisted, media: report(
                "MEDIA_COMPLETE", 100, media
            ),
            on_verified=consume_published,
        )
        if not completion_committed:
            raise WorkerError("E_COMPLETION_HANDOFF")
        committed = True
        outcome = (formal, mapping, cookie_closed)
    finally:
        active_error = sys.exc_info()[1]
        if cookie_stream is not None:
            cookie_closed = close_cookie_stream(cookie_stream)
        if closure_report is not None:
            closure_report(cookie_closed)
        try:
            cleanup_run_directory(run_directory, root)
        except OSError as exc:
            if not committed:
                if active_error is not None:
                    active_error.add_note("stage cleanup failed")
                else:
                    raise WorkerError("E_STAGE") from exc
    if outcome is None:
        raise WorkerError("E_WORKER")
    return outcome