bili: land reviewed half-hour pipeline and EOF isolation

Cai

10 days ago 9de34a52319e9c5432dfb3988e79a7d420304405

 dev/project-dev/bili_authenticated_extension/worker.py

@@ -7,17 +7,22 @@

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
@@ -25,27 +30,63 @@

from .constants import (
    BRIDGE_TIMEOUT_SECONDS,
    CANONICAL_URL,
    DURATION_TOLERANCE_MS,
    EXPECTED_DURATION_MS,
    EXTENSION_BUILD,
    EXTRACTOR_RETRIES,
    FILE_ACCESS_RETRIES,
    FRAGMENT_RETRIES,
    HTTP_RETRIES,
    RELOAD_GENERATION,
    SOCKET_TIMEOUT_SECONDS,
    TARGET_BVID,
    YTDLP_MODULE_SHA256,
    YTDLP_VERSION,
    duration_tolerance_ms,
    validate_bvid,
    validate_creator_uid,
)
from .protocol import ProtocolError, strict_json_loads, validate_start
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 = "749FC486B0F42315BD463F11771FE2A7C71CAB53DD9AC2E411CE82E1175DFF13"
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) -> None:
    def __init__(self, code: str, diagnostic: dict[str, object] | None = None) -> None:
        super().__init__(code)
        self.code = code
        self.diagnostic = diagnostic


class CancelRequested(BaseException):
@@ -116,13 +157,17 @@

@dataclass(frozen=True)
class HostConfig:
    creator_allowlist: frozenset[str]
    ffmpeg: Path
    ffprobe: Path
    bridge_python: Path
    bridge_script: Path
    batch_json: 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:
@@ -148,8 +193,13 @@
            raise WorkerError("E_CONFIG") from exc
        expected = {
            "schema",
            "target",
            "canonical_url",
            "creator_allowlist",
            "queue_path",
            "queue_state_path",
            "queue_lock_path",
            "reload_state_path",
            "reload_generation",
            "required_extension_build",
            "ffmpeg",
            "ffmpeg_sha256",
            "ffprobe",
@@ -158,16 +208,86 @@
            "bridge_python_sha256",
            "bridge_script",
            "bridge_script_sha256",
            "batch_json",
            "batch_json_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"] != 1 or raw["target"] != TARGET_BVID or raw["canonical_url"] != CANONICAL_URL:
        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")
@@ -178,17 +298,21 @@
        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,
            batch_json=cls._safe_absolute_file(raw["batch_json"], raw["batch_json_sha256"]),
            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,
        )


@@ -229,13 +353,17 @@
    return resolved


def fixed_stage_root() -> Path:
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"
        / TARGET_BVID
        / bvid
    )
    _reject_reparse_path(logical_root, local_app_data)
    resolved = logical_root.resolve(strict=False)
@@ -258,7 +386,7 @@

def cleanup_stale_runs(root: Path, *, boundary: Path | None = None) -> None:
    """Remove only uncommitted run-* directories below the fixed stage root."""
    allowed_root = fixed_stage_root() if boundary is None else boundary.resolve()
    allowed_root = root.resolve() if boundary is None else boundary.resolve()
    _ensure_within(root, allowed_root)
    if not root.exists():
        return
@@ -270,8 +398,8 @@
        shutil.rmtree(child)


def create_run_directory(root: Path | None = None) -> Path:
    stage_root = fixed_stage_root() if root is None else root.resolve()
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)
@@ -285,10 +413,10 @@
    raise WorkerError("E_STAGE")


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


@@ -345,8 +473,8 @@
    return converted


def validate_processed_info(info: Any) -> dict[str, Any]:
    if not isinstance(info, dict) or info.get("id") != TARGET_BVID:
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")
@@ -358,8 +486,11 @@
        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 - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
    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:
@@ -385,10 +516,11 @@
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") != TARGET_BVID:
    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:
@@ -426,6 +558,7 @@

def prepare_download_info(
    ydl: Any,
    job: dict[str, Any],
    *,
    downloader_resolver: Callable[..., Any] | None = None,
) -> tuple[dict[str, Any], bool, tuple[str, ...]]:
@@ -440,8 +573,8 @@
        return original_extract(*args, **kwargs)

    ydl.extract_info = one_extract
    processed = ydl.extract_info(CANONICAL_URL, download=False, process=True)
    validate_processed_info(processed)
    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:
@@ -451,6 +584,7 @@
    leaves, single = validate_download_info(
        download_info,
        ydl.params,
        job,
        downloader_resolver=downloader_resolver,
    )
    signed_urls = tuple(str(leaf["url"]) for leaf in leaves)
@@ -467,6 +601,23 @@
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,
@@ -477,46 +628,269 @@
        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
        executable, argv, _cwd, environment = arguments
        if not isinstance(executable, (str, os.PathLike)) or not isinstance(argv, (list, tuple)):
            raise WorkerError("E_SUBPROCESS_POLICY")
        executable_key = os.path.normcase(str(Path(executable).resolve()))
        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:
            raise WorkerError("E_SUBPROCESS_POLICY")
        text_args = [str(item) for item in argv]
            self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
        folded = "\x00".join(text_args).casefold()
        forbidden = ("://", "-headers", "-cookies", "authorization", "cookie:", "referer:", "user-agent:")
        if any(item in folded for item in forbidden) or any(item in folded for item in self.secrets):
            raise WorkerError("E_SUBPROCESS_POLICY")
        if environment is not None:
            encoded_env = "\x00".join(f"{key}={value}" for key, value in environment.items()).casefold()
            if any(item in encoded_env for item in self.secrets):
                raise WorkerError("E_SUBPROCESS_POLICY")
        if executable_key.endswith("ffmpeg.exe") or executable_key.endswith("ffprobe.exe"):
            for index, argument in enumerate(text_args[:-1]):
                if argument == "-i":
                    self._local_path(text_args[index + 1], must_exist=True)
            output = text_args[-1]
            if output not in {"-", "NUL"} and not output.startswith("-"):
                self._local_path(output, must_exist=False)
        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 path.is_absolute() or str(path).startswith("\\\\"):
            raise WorkerError("E_SUBPROCESS_POLICY")
        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:
            resolved = path.resolve(strict=must_exist)
        except OSError as exc:
            raise WorkerError("E_SUBPROCESS_POLICY") from exc
        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)
        except ValueError as exc:
            raise WorkerError("E_SUBPROCESS_POLICY") from exc
        if must_exist and (not resolved.is_file() or resolved.is_symlink()):
            raise WorkerError("E_SUBPROCESS_POLICY")
                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:
@@ -619,7 +993,7 @@
        raise WorkerError("E_MERGE")


def probe_mkv(ffprobe: Path, candidate: Path) -> None:
def probe_mkv(ffprobe: Path, candidate: Path, job: dict[str, Any] | None = None) -> None:
    try:
        result = _run_local(
            [
@@ -627,7 +1001,7 @@
            "-v",
            "error",
            "-show_entries",
            "format=format_name:stream=codec_type",
            "format=format_name,duration:stream=codec_type",
            "-of",
            "json",
            str(candidate),
@@ -645,6 +1019,13 @@
    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:
@@ -658,96 +1039,981 @@
    return candidates[0]


def _bridge_command(config: HostConfig, candidate: Path) -> list[str]:
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(config.batch_json),
        "--yt-dlp",
        str(config.yt_dlp_executable),
        str(batch_json),
        "accept-browser-file",
        "--bvid",
        TARGET_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) -> tuple[str, str]:
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),
            _bridge_command(config, candidate, job, batch_json),
            BRIDGE_TIMEOUT_SECONDS,
            capture_stdout=True,
        )
    except subprocess.TimeoutExpired as exc:
        raise WorkerError("E_BACKHALF") from 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:
        raise WorkerError("E_BACKHALF")
    try:
        payload = json.loads(result.stdout.decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise WorkerError("E_BACKHALF") from exc
    if not isinstance(payload, dict) or payload.get("result") != "PASS":
        raise WorkerError("E_BACKHALF")
            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):
        raise WorkerError("E_BACKHALF")
    item = items[0]
    required = {"bvid", "local_file", "bytes", "sha256", "duration_seconds", "acquisition_mode"}
    if not required.issubset(item) or item["bvid"] != TARGET_BVID:
        raise WorkerError("E_BACKHALF")
    if item["acquisition_mode"] != "authorized_browser_file_handoff":
        raise WorkerError("E_BACKHALF")
    formal = item["local_file"]
    mapping = f"{TARGET_BVID}.download.json"
    if formal != f"{TARGET_BVID}.mkv":
        raise WorkerError("E_BACKHALF")
    formal_path = config.destination / formal
    mapping_path = config.destination / mapping
    if not formal_path.is_file() or not mapping_path.is_file():
        raise WorkerError("E_BACKHALF")
        preserve_verified_media_before_output_failure()
        raise failure(
            "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
        )
    try:
        duration_ms = round(_finite_number(item["duration_seconds"]) * 1000)
        item = _validate_complete_bridge_item(items[0], job, persisted=False)
    except WorkerError as exc:
        raise WorkerError("E_BACKHALF") from exc
    if abs(duration_ms - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
        raise WorkerError("E_BACKHALF")
    formal_sha = sha256_file(formal_path)
    if (
        isinstance(item["bytes"], bool)
        or not isinstance(item["bytes"], int)
        or item["bytes"] != formal_path.stat().st_size
        or not isinstance(item["sha256"], str)
        or item["sha256"] != formal_sha
    ):
        raise WorkerError("E_BACKHALF")
        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:
        persisted = strict_json_loads(mapping_path.read_bytes())
    except (OSError, ProtocolError) as exc:
        raise WorkerError("E_BACKHALF") from exc
    matched_fields = {
        "bvid": TARGET_BVID,
        "source": CANONICAL_URL,
        "local_file": formal,
        "bytes": item["bytes"],
        "sha256": formal_sha,
        "acquisition_mode": "authorized_browser_file_handoff",
    }
    if any(persisted.get(key) != expected for key, expected in matched_fields.items()):
        raise WorkerError("E_BACKHALF")
    try:
        persisted_duration_ms = round(_finite_number(persisted.get("duration_seconds")) * 1000)
    except WorkerError as exc:
        raise WorkerError("E_BACKHALF") from exc
    if persisted_duration_ms != duration_ms:
        raise WorkerError("E_BACKHALF")
    return formal, mapping
            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(
@@ -799,15 +2065,19 @@
    config: HostConfig,
    *,
    cancel_check: Callable[[], bool],
    report: Callable[[str, int], None],
    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)
    root = fixed_stage_root() if stage_root is None else stage_root.resolve()
    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:
@@ -825,6 +2095,38 @@
                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 = [
@@ -852,7 +2154,7 @@
        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)
            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)
@@ -882,7 +2184,7 @@
        checkpoint()
        candidate = validate_unique_candidate(run_directory)
        checkpoint()
        probe_mkv(config.ffprobe, candidate)
        probe_mkv(config.ffprobe, candidate, job)
        checkpoint()
        cookie_closed = close_cookie_stream(cookie_stream)
        cookie_stream = None
@@ -892,7 +2194,29 @@
        if commit_begin is not None:
            commit_begin()
        checkpoint()
        formal, mapping = run_frozen_bridge(config, candidate)
        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: