#!/usr/bin/env python3
"""Durable half-hour coordinator for one configured Bilibili creator.

The coordinator never reads browser credentials and never downloads media.  It
turns already validated collector/formal records into exact-once role outboxes,
tracks non-overlapping half-hour runs, accepts exact downstream receipts, and
performs narrowly allowlisted Git delivery.  Browser capture and native media
work remain in their reviewed components.
"""

from __future__ import annotations

import argparse
import contextlib
import hashlib
import json
import os
import re
import stat
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, Iterator, Mapping, Sequence
from urllib.parse import urlsplit


SCHEMA = 1
INTERVAL_MINUTES = 30
TASK_ID = "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001"
UID = re.compile(r"^[1-9][0-9]{0,19}$")
BVID = re.compile(r"^BV1[1-9A-HJ-NP-Za-km-z]{9}$")
SHA256 = re.compile(r"^[0-9A-F]{64}$")
SHA256_MIXED_ASCII = re.compile(r"^[0-9A-Fa-f]{64}$", re.ASCII)
THREAD_ID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
SAFE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,240}$")
SECRET_KEY = re.compile(r"(?i)(cookie|sessdata|token|credential|authorization|localstorage|profile|signed_url)")
SECRET_VALUE = re.compile(r"(?i)(SYNTHETIC_SECRET|sessdata=|cookie\s*[:=]|authorization\s*[:=]|bearer\s+|token=|signed_url=|localstorage)")
FORBIDDEN_GIT_SUFFIXES = (
    ".mkv", ".mp4", ".mov", ".webm", ".download.json", ".flac", ".partial", ".crdownload"
)
RECEIPT_GIT_KIND_SUFFIX = {
    "transcript": ".txt",
    "transcript_txt": ".txt",
    "transcript_srt": ".srt",
    "transcript_json": ".json",
    "minutes": ".md",
    "minutes_md": ".md",
    "minutes_pdf": ".pdf",
    "relocation_manifest": ".json",
    "documentation": ".md",
}
CONTENT_TYPES = frozenset({"article", "text", "image"})
VIDEO_COMPLETE = "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT"
CANONICAL_TITLE_MAX_LENGTH = 64
WINDOWS_INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
WINDOWS_RESERVED_NAMES = {
    "CON", "PRN", "AUX", "NUL",
    *(f"COM{value}" for value in range(1, 10)),
    *(f"LPT{value}" for value in range(1, 10)),
}
RELOCATION_REPORT_TYPE = "VIDEO_ARTIFACT_RELOCATION_BATCH"
PUBLIC_VIDEO_KINDS = {
    "transcript_txt": ".txt",
    "transcript_srt": ".srt",
    "transcript_json": ".json",
    "minutes_md": ".md",
    "minutes_pdf": ".pdf",
}


class PipelineError(RuntimeError):
    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code


@dataclass(frozen=True)
class Config:
    path: Path
    project_root: Path
    creator_uid: str
    creator_name: str
    dynamic_url: str
    archive_root: Path
    formal_manifest: Path
    processing_handoffs: Path
    state_dir: Path
    video_root: Path
    video_downloader_thread_id: str
    media_thread_id: str
    minutes_thread_id: str
    reply_thread_id: str
    git_remote: str
    git_branch: str
    git_extensions: frozenset[str]
    git_doc_paths: frozenset[str]

    @property
    def state_path(self) -> Path:
        return self.state_dir / "state.json"

    @property
    def runs_path(self) -> Path:
        return self.state_dir / "runs.jsonl"

    @property
    def outbox_path(self) -> Path:
        return self.state_dir / "outbox.jsonl"

    @property
    def terminals_path(self) -> Path:
        return self.state_dir / "terminals.jsonl"

    @property
    def lock_path(self) -> Path:
        return self.state_dir / "coordinator.lock"

    def git_index_guard_path(self, batch_id: str) -> Path:
        return self.state_dir / f"git-shared-index-guard-{batch_id}.json"

    @property
    def relocation_root(self) -> Path:
        return self.archive_root / "artifact-relocations"


def _exact(value: Any, keys: Iterable[str], field: str) -> Mapping[str, Any]:
    expected = set(keys)
    if not isinstance(value, dict) or set(value) != expected:
        raise PipelineError("E_SCHEMA", f"{field} keys differ")
    return value


def _reject_secrets(value: Any, path: str = "$") -> None:
    if isinstance(value, dict):
        for key, item in value.items():
            if not isinstance(key, str) or SECRET_KEY.search(key):
                raise PipelineError("E_SECRET_FIELD", f"secret-like field at {path}")
            _reject_secrets(item, f"{path}.{key}")
    elif isinstance(value, list):
        for index, item in enumerate(value):
            _reject_secrets(item, f"{path}[{index}]")
    elif isinstance(value, str):
        if SECRET_VALUE.search(value):
            raise PipelineError("E_SECRET_FIELD", f"secret-like value at {path}")
        parsed = urlsplit(value)
        if parsed.scheme in {"http", "https"} and (parsed.username is not None or parsed.password is not None):
            raise PipelineError("E_SECRET_FIELD", f"credential-bearing URL at {path}")


def _canonical(value: Mapping[str, Any]) -> bytes:
    _reject_secrets(value)
    return (json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n").encode("ascii")


def _strict_json(path: Path, field: str) -> tuple[Any, bytes]:
    try:
        payload = path.read_bytes()
    except OSError as exc:
        raise PipelineError("E_INPUT", f"{field} is unavailable") from exc
    if not payload or payload.startswith(b"\xef\xbb\xbf") or b"\r" in payload or not payload.endswith(b"\n"):
        raise PipelineError("E_INPUT", f"{field} is not strict UTF-8 JSON")
    try:
        text = payload.decode("utf-8")
        value = json.loads(text)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise PipelineError("E_INPUT", f"{field} is invalid JSON") from exc
    _reject_secrets(value)
    return value, payload


def _absolute(base: Path, value: Any, field: str) -> Path:
    if not isinstance(value, str) or not value:
        raise PipelineError("E_CONFIG", f"{field} must be a path")
    candidate = Path(value)
    if not candidate.is_absolute():
        candidate = base / candidate
    return Path(os.path.abspath(candidate))


def _within(child: Path, parent: Path) -> bool:
    try:
        child.relative_to(parent)
        return True
    except ValueError:
        return False


def _canonical_dynamic_url(value: Any, uid: str) -> str:
    if not isinstance(value, str):
        raise PipelineError("E_CONFIG", "dynamic_url must be a string")
    parsed = urlsplit(value)
    if parsed.scheme != "https" or parsed.hostname != "space.bilibili.com" or parsed.query or parsed.fragment or parsed.path.rstrip("/") != f"/{uid}/dynamic":
        raise PipelineError("E_CONFIG", "dynamic_url is not canonical")
    return f"https://space.bilibili.com/{uid}/dynamic"


def _canonical_video_url(value: Any, bvid: str) -> str:
    if not isinstance(value, str) or not BVID.fullmatch(bvid):
        raise PipelineError("E_SOURCE_BINDING", "video source identity is incomplete")
    _reject_secrets(value)
    parsed = urlsplit(value)
    if (
        parsed.scheme != "https"
        or parsed.hostname != "www.bilibili.com"
        or parsed.username is not None
        or parsed.password is not None
        or parsed.port is not None
        or parsed.query
        or parsed.fragment
        or parsed.path.rstrip("/") != f"/video/{bvid}"
    ):
        raise PipelineError("E_SOURCE_BINDING", "video source URL is not canonical")
    return f"https://www.bilibili.com/video/{bvid}"


def load_config(path: Path) -> Config:
    value, _ = _strict_json(path, "config")
    root = _exact(value, {"schema_version", "task_id", "interval_minutes", "creator", "paths", "downstream", "git"}, "config")
    if (
        type(root["schema_version"]) is not int
        or root["schema_version"] != SCHEMA
        or root["task_id"] != TASK_ID
        or type(root["interval_minutes"]) is not int
        or root["interval_minutes"] != INTERVAL_MINUTES
    ):
        raise PipelineError("E_CONFIG", "config identity differs")
    creator = _exact(root["creator"], {"uid", "name", "dynamic_url"}, "creator")
    uid = creator["uid"]
    if not isinstance(uid, str) or not UID.fullmatch(uid) or not isinstance(creator["name"], str) or not creator["name"].strip():
        raise PipelineError("E_CONFIG", "creator identity differs")
    paths = _exact(root["paths"], {"project_root", "archive_root", "formal_manifest", "processing_handoffs", "state_dir", "video_root"}, "paths")
    base = path.parent
    project_root = _absolute(base, paths["project_root"], "project_root")
    archive_root = _absolute(project_root, paths["archive_root"], "archive_root")
    formal = _absolute(project_root, paths["formal_manifest"], "formal_manifest")
    handoffs = _absolute(project_root, paths["processing_handoffs"], "processing_handoffs")
    state_dir = _absolute(project_root, paths["state_dir"], "state_dir")
    video_root = _absolute(project_root, paths["video_root"], "video_root")
    if not _within(archive_root, project_root) or not _within(formal, archive_root) or not _within(handoffs, archive_root) or not _within(state_dir, project_root):
        raise PipelineError("E_CONFIG", "project output path escaped its governed root")
    downstream = _exact(
        root["downstream"],
        {"video_downloader_thread_id", "media_processor_thread_id", "minutes_thread_id", "reply_thread_id"},
        "downstream",
    )
    for field, thread_id in downstream.items():
        if not isinstance(thread_id, str) or not THREAD_ID.fullmatch(thread_id):
            raise PipelineError("E_CONFIG", f"{field} is not a thread id")
    git = _exact(root["git"], {"remote", "branch", "allowed_extensions", "allowed_docs"}, "git")
    if (
        git["remote"] != "origin"
        or not isinstance(git["branch"], str)
        or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,199}", git["branch"])
        or git["branch"].endswith(("/", "."))
        or any(marker in git["branch"] for marker in ("..", "//", "@{"))
    ):
        raise PipelineError("E_CONFIG", "git destination differs")
    extensions = git["allowed_extensions"]
    if not isinstance(extensions, list) or not extensions or any(not isinstance(item, str) or not item.startswith(".") or item.lower() in FORBIDDEN_GIT_SUFFIXES for item in extensions):
        raise PipelineError("E_CONFIG", "git extension allowlist is invalid")
    docs = git["allowed_docs"]
    if not isinstance(docs, list) or not docs:
        raise PipelineError("E_CONFIG", "git document allowlist is invalid")
    normalized_docs: list[str] = []
    for value in docs:
        if not isinstance(value, str) or not value or Path(value).is_absolute():
            raise PipelineError("E_CONFIG", "git document allowlist is invalid")
        target = Path(os.path.abspath(project_root / Path(value)))
        if not _within(target, project_root) or target.suffix.lower() != ".md":
            raise PipelineError("E_CONFIG", "git document allowlist escaped the project")
        normalized_docs.append(target.relative_to(project_root).as_posix())
    if len(set(normalized_docs)) != len(normalized_docs):
        raise PipelineError("E_CONFIG", "git document allowlist is duplicated")
    return Config(
        path=path, project_root=project_root, creator_uid=uid, creator_name=creator["name"].strip(),
        dynamic_url=_canonical_dynamic_url(creator["dynamic_url"], uid), archive_root=archive_root,
        formal_manifest=formal, processing_handoffs=handoffs, state_dir=state_dir, video_root=video_root,
        video_downloader_thread_id=downstream["video_downloader_thread_id"],
        media_thread_id=downstream["media_processor_thread_id"], minutes_thread_id=downstream["minutes_thread_id"],
        reply_thread_id=downstream["reply_thread_id"], git_remote=git["remote"], git_branch=git["branch"],
        git_extensions=frozenset(item.lower() for item in extensions), git_doc_paths=frozenset(normalized_docs),
    )


def _file_identity(path: Path) -> dict[str, Any]:
    payload = path.read_bytes() if path.exists() else b""
    if payload and not payload.endswith(b"\n"):
        raise PipelineError("E_JOURNAL", f"{path.name} lacks final LF")
    return {"bytes": len(payload), "lines": payload.count(b"\n"), "sha256": hashlib.sha256(payload).hexdigest().upper()}


def _atomic(path: Path, payload: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, raw = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".partial", dir=path.parent)
    partial = Path(raw)
    try:
        with os.fdopen(fd, "wb") as stream:
            stream.write(payload)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(partial, path)
        if path.read_bytes() != payload:
            raise PipelineError("E_DURABILITY", f"{path.name} readback differs")
    finally:
        with contextlib.suppress(FileNotFoundError):
            partial.unlink()


def _append(path: Path, value: Mapping[str, Any]) -> dict[str, Any]:
    payload = _canonical(value)
    pre = path.read_bytes() if path.exists() else b""
    if pre and not pre.endswith(b"\n"):
        raise PipelineError("E_JOURNAL", f"{path.name} is malformed")
    _atomic(path, pre + payload)
    if not path.read_bytes().startswith(pre):
        raise PipelineError("E_DURABILITY", f"{path.name} prefix changed")
    return _file_identity(path)


def _create_new(path: Path, payload: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
    try:
        descriptor = os.open(path, flags, 0o600)
    except FileExistsError as exc:
        raise PipelineError("E_DURABILITY", f"{path.name} already exists") from exc
    try:
        with os.fdopen(descriptor, "wb", closefd=False) as stream:
            stream.write(payload)
            stream.flush()
            os.fsync(stream.fileno())
    finally:
        os.close(descriptor)
    if path.read_bytes() != payload:
        raise PipelineError("E_DURABILITY", f"{path.name} readback differs")


@contextlib.contextmanager
def _lock(config: Config) -> Iterator[None]:
    config.state_dir.mkdir(parents=True, exist_ok=True)
    stream = config.lock_path.open("a+b")
    try:
        if os.name == "nt":
            import msvcrt
            stream.seek(0)
            if stream.tell() == stream.seek(0, os.SEEK_END) == 0:
                stream.write(b"0")
                stream.flush()
            stream.seek(0)
            try:
                msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
            except OSError as exc:
                raise PipelineError("E_RUN_ACTIVE", "another coordinator owns the run lock") from exc
        else:
            import fcntl
            try:
                fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except OSError as exc:
                raise PipelineError("E_RUN_ACTIVE", "another coordinator owns the run lock") from exc
        yield
    finally:
        if os.name == "nt":
            with contextlib.suppress(OSError):
                stream.seek(0)
                msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
        else:
            with contextlib.suppress(OSError):
                fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
        stream.close()


def _read_lines(
    path: Path,
    field: str,
    *,
    reject_secrets: bool = True,
) -> tuple[list[dict[str, Any]], bytes]:
    if not path.exists():
        return [], b""
    payload = path.read_bytes()
    if payload and (b"\r" in payload or not payload.endswith(b"\n")):
        raise PipelineError("E_INPUT", f"{field} is not strict JSONL")
    rows: list[dict[str, Any]] = []
    for index, line in enumerate(payload.splitlines(), 1):
        try:
            row = json.loads(line.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise PipelineError("E_INPUT", f"{field} line {index} is invalid") from exc
        if not isinstance(row, dict):
            raise PipelineError("E_INPUT", f"{field} line {index} is not an object")
        if reject_secrets:
            _reject_secrets(row)
        rows.append(row)
    return rows, payload


def _load_state(config: Config) -> dict[str, Any]:
    value, _ = _strict_json(config.state_path, "state")
    state = _exact(value, {"schema_version", "task_id", "creator_uid", "initialized_at", "cursors", "active_run"}, "state")
    if (
        type(state["schema_version"]) is not int
        or state["schema_version"] != SCHEMA
        or state["task_id"] != TASK_ID
        or state["creator_uid"] != config.creator_uid
    ):
        raise PipelineError("E_STATE", "state identity differs")
    cursors = _exact(state["cursors"], {"formal_lines", "formal_sha256", "handoff_lines", "handoff_sha256"}, "cursors")
    if (
        type(cursors["formal_lines"]) is not int
        or cursors["formal_lines"] < 0
        or type(cursors["handoff_lines"]) is not int
        or cursors["handoff_lines"] < 0
        or not isinstance(cursors["formal_sha256"], str)
        or not SHA256.fullmatch(cursors["formal_sha256"])
        or not isinstance(cursors["handoff_sha256"], str)
        or not SHA256.fullmatch(cursors["handoff_sha256"])
    ):
        raise PipelineError("E_STATE", "state cursor identity differs")
    return dict(state)


def _write_state(config: Config, state: Mapping[str, Any]) -> None:
    _atomic(config.state_path, _canonical(state))


def _now(value: str | None) -> datetime:
    if value is None:
        return datetime.now(timezone.utc)
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise PipelineError("E_TIME", "now is invalid") from exc
    if parsed.tzinfo is None:
        raise PipelineError("E_TIME", "now must be offset-aware")
    return parsed.astimezone(timezone.utc)


def _published_at(value: Any, field: str = "published_at") -> datetime:
    if not isinstance(value, str) or not value:
        raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise PipelineError("E_SOURCE_BINDING", f"{field} differs") from exc
    if parsed.tzinfo is None:
        raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
    return parsed


def _canonical_title(value: Any) -> str:
    if not isinstance(value, str) or not value or "\x00" in value:
        raise PipelineError("E_SOURCE_BINDING", "video title differs")
    cleaned = WINDOWS_INVALID_CHARS.sub("_", value)
    cleaned = re.sub(r"\s+", " ", cleaned).strip(" .")
    if not cleaned:
        cleaned = "untitled"
    if cleaned.upper().split(".", 1)[0] in WINDOWS_RESERVED_NAMES:
        cleaned = "_" + cleaned
    cleaned = cleaned[:CANONICAL_TITLE_MAX_LENGTH].rstrip(" .")
    return cleaned or "untitled"


def _canonical_video_base(stable_id: Any, title: Any, published_at: Any) -> str:
    if not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
        raise PipelineError("E_SOURCE_BINDING", "video stable identity differs")
    published = _published_at(published_at)
    return f"{published:%Y%m%d-%H%M%S}_video_{_canonical_title(title)}_{stable_id}"


def _project_relative(config: Config, path: Path, code: str = "E_ARTIFACT") -> str:
    target = Path(os.path.abspath(path))
    if not _within(target, config.project_root):
        raise PipelineError(code, "path escaped the project root")
    return target.relative_to(config.project_root).as_posix()


def initialize(config: Config, now: datetime) -> dict[str, Any]:
    with _lock(config):
        if config.state_path.exists():
            state = _load_state(config)
            return {"status": "ALREADY_INITIALIZED", "state": state, "state_identity": _file_identity(config.state_path)}
        formal = _file_identity(config.formal_manifest)
        handoff = _file_identity(config.processing_handoffs)
        state = {
            "schema_version": SCHEMA, "task_id": TASK_ID, "creator_uid": config.creator_uid,
            "initialized_at": now.isoformat(),
            "cursors": {
                "formal_lines": formal["lines"], "formal_sha256": formal["sha256"],
                "handoff_lines": handoff["lines"], "handoff_sha256": handoff["sha256"],
            },
            "active_run": None,
        }
        _write_state(config, state)
        return {"status": "INITIALIZED", "baseline": {"formal": formal, "handoff": handoff}, "state_identity": _file_identity(config.state_path)}


def _run_id(config: Config, slot: int) -> str:
    return hashlib.sha256(f"bili-half-hour-v1\0{config.creator_uid}\0{slot}".encode("ascii")).hexdigest()


def begin(config: Config, now: datetime) -> dict[str, Any]:
    with _lock(config):
        state = _load_state(config)
        slot = int(now.timestamp()) // (INTERVAL_MINUTES * 60)
        run_id = _run_id(config, slot)
        active = state["active_run"]
        if active is not None:
            if active.get("run_id") == run_id:
                return {"status": "RUN_RESUMED", "run": active}
            raise PipelineError("E_RUN_ACTIVE", "a prior half-hour run is still active")
        event = {
            "schema_version": SCHEMA, "event": "RUN_STARTED", "run_id": run_id, "slot": slot,
            "creator_uid": config.creator_uid, "started_at": now.isoformat(),
        }
        _append(config.runs_path, event)
        state["active_run"] = {"run_id": run_id, "slot": slot, "started_at": now.isoformat()}
        _write_state(config, state)
        return {"status": "RUN_STARTED", "run": state["active_run"], "dynamic_url": config.dynamic_url}


def _journal_source_row(config: Config, source: Mapping[str, Any]) -> tuple[str, Mapping[str, Any]]:
    if set(source) != {"journal", "line", "sha256"}:
        raise PipelineError("E_SOURCE_BINDING", "journal source shape differs")
    journal = source.get("journal")
    line = source.get("line")
    digest = source.get("sha256")
    if journal not in {"formal", "processing_handoff"} or type(line) is not int or line <= 0 or not isinstance(digest, str) or not SHA256.fullmatch(digest):
        raise PipelineError("E_SOURCE_BINDING", "journal source identity differs")
    path = config.formal_manifest if journal == "formal" else config.processing_handoffs
    # Formal and processing journals contain immutable historical audit fields.
    # Do not treat a legacy key name (for example a handoff identity containing
    # "authorization") as a credential.  Every consumable row is rebound to a
    # narrow source-controlled projection below, and the projected outbox is
    # still subject to the normal key/value secret rejection.
    rows, raw = _read_lines(path, journal, reject_secrets=False)
    raw_lines = raw.splitlines()
    if line > len(rows) or _row_digest(raw_lines[line - 1]) != digest:
        raise PipelineError("E_SOURCE_BINDING", "journal source bytes differ")
    row = rows[line - 1]
    if _creator_uid(row) != config.creator_uid:
        raise PipelineError("E_SOURCE_BINDING", "journal creator differs")
    return journal, row


def _safe_text(value: Any, field: str, *, pattern: re.Pattern[str] | None = None) -> str:
    if not isinstance(value, str) or not value or "\x00" in value or SECRET_KEY.search(value):
        raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
    if pattern is not None and not pattern.fullmatch(value):
        raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
    return value


def _project_formal_payload(config: Config, kind: str, row: Mapping[str, Any]) -> dict[str, Any]:
    item_type = row.get("item_type")
    status_value = row.get("status")
    stable_id = row.get("stable_id")
    if kind == "GIT_DELIVERY_READY":
        if item_type not in CONTENT_TYPES or status_value != "SAVED" or not isinstance(stable_id, str) or not stable_id:
            raise PipelineError("E_SOURCE_BINDING", "formal content source differs")
        return {"stable_id": stable_id, "files": _content_artifacts(row, config), "reason": "CONTENT_ARCHIVED"}
    if kind == "VIDEO_DOWNLOAD_READY":
        if item_type != "video" or status_value == VIDEO_COMPLETE or not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
            raise PipelineError("E_SOURCE_BINDING", "formal video source differs")
        duration = row.get("expected_duration_seconds")
        if isinstance(duration, bool) or not isinstance(duration, (int, float)) or not (0 < float(duration) < 86400):
            raise PipelineError("E_SOURCE_BINDING", "video duration differs")
        return {
            "bvid": stable_id,
            "source_url": _canonical_video_url(row.get("source_url"), stable_id),
            "title": _safe_text(row.get("title"), "video title"),
            "published_at": _safe_text(row.get("published_at"), "video publication"),
            "expected_duration_seconds": duration,
        }
    raise PipelineError("E_SOURCE_BINDING", "formal source cannot produce this kind")


def _project_handoff_payload(config: Config, row: Mapping[str, Any]) -> dict[str, Any]:
    common_keys = {
        "type", "status", "handoff_id", "queue_job_id", "creator_uid", "bvid",
        "source_url", "media_path", "mapping_path", "bytes", "sha256", "duration_seconds",
        "video_codec", "audio_codec", "created_at",
    }
    schema_keys = set(row) - common_keys
    if (
        schema_keys not in ({"schema"}, {"schema_version"})
        or set(row) != common_keys | schema_keys
        or row.get("type") != "media-processing-handoff"
        or row.get("status") != "READY"
    ):
        raise PipelineError("E_SOURCE_BINDING", "processing handoff shape differs")
    schema_value = row[next(iter(schema_keys))]
    if type(schema_value) is not int or schema_value != SCHEMA or row.get("creator_uid") != config.creator_uid:
        raise PipelineError("E_SOURCE_BINDING", "processing handoff identity differs")
    bvid = row.get("bvid")
    byte_count = row.get("bytes")
    duration = row.get("duration_seconds")
    if (
        not isinstance(bvid, str) or not BVID.fullmatch(bvid)
        or type(byte_count) is not int or byte_count <= 0
        or isinstance(duration, bool) or not isinstance(duration, (int, float)) or float(duration) <= 0
        or not isinstance(row.get("sha256"), str) or not SHA256_MIXED_ASCII.fullmatch(row["sha256"])
        or not isinstance(row.get("queue_job_id"), str) or not re.fullmatch(r"[0-9a-f]{64}", row["queue_job_id"])
    ):
        raise PipelineError("E_SOURCE_BINDING", "processing handoff media identity differs")
    projected = {
        "type": "media-processing-handoff", "status": "READY",
        "handoff_id": _safe_text(row.get("handoff_id"), "handoff id", pattern=SAFE_ID),
        "queue_job_id": row["queue_job_id"], "creator_uid": config.creator_uid, "bvid": bvid,
        "source_url": _canonical_video_url(row.get("source_url"), bvid),
        "media_path": _safe_text(row.get("media_path"), "media path"),
        "mapping_path": _safe_text(row.get("mapping_path"), "mapping path"),
        "bytes": byte_count, "sha256": row["sha256"].upper(), "duration_seconds": duration,
        "video_codec": _safe_text(row.get("video_codec"), "video codec", pattern=SAFE_ID),
        "audio_codec": _safe_text(row.get("audio_codec"), "audio codec", pattern=SAFE_ID),
        "created_at": _safe_text(row.get("created_at"), "handoff creation time"),
    }
    return projected


def _project_relocation_payload(config: Config, source: Mapping[str, Any]) -> dict[str, Any]:
    if set(source) != {"journal", "path", "bytes", "sha256"} or source.get("journal") != "relocation_report":
        raise PipelineError("E_SOURCE_BINDING", "relocation source shape differs")
    relative = source.get("path")
    size = source.get("bytes")
    digest = source.get("sha256")
    if (
        not isinstance(relative, str)
        or Path(relative).is_absolute()
        or type(size) is not int
        or size <= 0
        or not isinstance(digest, str)
        or not SHA256.fullmatch(digest)
    ):
        raise PipelineError("E_SOURCE_BINDING", "relocation source identity differs")
    path = Path(os.path.abspath(config.project_root / Path(relative)))
    if not _within(path, config.relocation_root):
        raise PipelineError("E_SOURCE_BINDING", "relocation source escaped its root")
    payload = _stable_artifact_bytes(path, config.relocation_root)
    if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
        raise PipelineError("E_SOURCE_BINDING", "relocation source bytes differ")
    report = _read_relocation_report(config, path)
    files: list[dict[str, Any]] = []
    for item in report["items"]:
        files.extend(
            {
                "path": alias["new_path"],
                "bytes": alias["bytes"],
                "sha256": alias["sha256"],
                "kind": alias["kind"],
            }
            for alias in item["aliases"]
        )
    files.extend(report["docs"])
    files.append({
        "path": relative,
        "bytes": size,
        "sha256": digest,
        "kind": "relocation_manifest",
    })
    return {
        "stable_id": report["batch_id"],
        "files": files,
        "remove_paths": report["remove_paths"],
        "reason": "CANONICAL_VIDEO_ARTIFACT_MIGRATION",
    }


def _expected_outbox_payload(config: Config, kind: str, source: Mapping[str, Any]) -> dict[str, Any]:
    if source.get("journal") == "relocation_report":
        if kind != "GIT_DELIVERY_READY":
            raise PipelineError("E_SOURCE_BINDING", "relocation source kind differs")
        return _project_relocation_payload(config, source)
    if source.get("journal") in {"formal", "processing_handoff"}:
        journal, row = _journal_source_row(config, source)
        if journal == "formal":
            return _project_formal_payload(config, kind, row)
        if kind != "VIDEO_TRANSCRIPTION_READY":
            raise PipelineError("E_SOURCE_BINDING", "handoff source kind differs")
        return _project_handoff_payload(config, row)
    allowed = {"journal", "receipt_sha256"}
    if source.get("projection") is not None:
        allowed.add("projection")
    if set(source) != allowed or source.get("journal") != "terminals" or not isinstance(source.get("receipt_sha256"), str) or not SHA256.fullmatch(source["receipt_sha256"]):
        raise PipelineError("E_SOURCE_BINDING", "terminal source identity differs")
    terminals = [row for row in _terminal_rows(config) if row.get("receipt_sha256") == source["receipt_sha256"]]
    if len(terminals) != 1:
        raise PipelineError("E_SOURCE_BINDING", "terminal source is absent or ambiguous")
    terminal = terminals[0]
    files = terminal["files"]
    stable_id = terminal["stable_id"]
    projection = source.get("projection")
    if kind == "MINUTES_READY" and terminal["event"] == "TRANSCRIPTION_COMPLETE" and projection is None:
        return {"stable_id": stable_id, "transcript_terminal_id": terminal["terminal_id"], "files": files}
    if kind == "GIT_DELIVERY_READY" and terminal["event"] == "TRANSCRIPTION_COMPLETE" and projection == "transcript":
        return {"stable_id": stable_id, "files": files, "reason": "TRANSCRIPT_COMPLETE"}
    if kind == "GIT_DELIVERY_READY" and terminal["event"] == "MINUTES_COMPLETE" and projection == "minutes":
        return {"stable_id": stable_id, "files": files, "reason": "MINUTES_COMPLETE"}
    raise PipelineError("E_SOURCE_BINDING", "terminal source projection differs")


def _outbox_id(kind: str, source: Mapping[str, Any], payload: Mapping[str, Any]) -> str:
    material = _canonical({"kind": kind, "source": dict(source), "payload": dict(payload)})
    return hashlib.sha256(material).hexdigest()


def _outbox_rows(config: Config) -> list[dict[str, Any]]:
    rows, _ = _read_lines(config.outbox_path, "outbox")
    allowed_kinds = {"GIT_DELIVERY_READY", "VIDEO_DOWNLOAD_READY", "VIDEO_TRANSCRIPTION_READY", "MINUTES_READY"}
    grouped: dict[str, list[dict[str, Any]]] = {}
    for row in rows:
        if type(row.get("schema_version")) is not int or row.get("schema_version") != SCHEMA:
            raise PipelineError("E_OUTBOX", "outbox schema identity differs")
        outbox_id = row.get("outbox_id")
        event = row.get("event")
        if not isinstance(outbox_id, str) or not re.fullmatch(r"[0-9a-f]{64}", outbox_id):
            raise PipelineError("E_OUTBOX", "outbox identity differs")
        if event not in {"CREATED", "DISPATCH_INTENT", "OBSERVED", "GIT_COMMIT_INTENT", "COMMIT_CREATED", "COMPLETE"}:
            raise PipelineError("E_OUTBOX", "outbox event differs")
        grouped.setdefault(outbox_id, []).append(row)
    for outbox_id, events in grouped.items():
        created = [row for row in events if row.get("event") == "CREATED"]
        if len(created) != 1 or events[0].get("event") != "CREATED":
            raise PipelineError("E_OUTBOX", "outbox creation history differs")
        origin = created[0]
        if set(origin) != {"schema_version", "event", "outbox_id", "kind", "creator_uid", "source", "payload", "created_at"}:
            raise PipelineError("E_OUTBOX", "outbox creation shape differs")
        kind = origin.get("kind")
        source = origin.get("source")
        if (
            kind not in allowed_kinds
            or origin.get("creator_uid") != config.creator_uid
            or not isinstance(source, dict)
            or not isinstance(origin.get("payload"), dict)
            or _outbox_id(kind, source, origin["payload"]) != outbox_id
            or not isinstance(origin.get("created_at"), str)
        ):
            raise PipelineError("E_OUTBOX", "outbox creation binding differs")
        expected_payload = _expected_outbox_payload(config, kind, source)
        if origin["payload"] != expected_payload:
            raise PipelineError("E_OUTBOX", "outbox payload differs from its immutable source")
        counts = {name: sum(row.get("event") == name for row in events) for name in {"DISPATCH_INTENT", "OBSERVED", "GIT_COMMIT_INTENT", "COMMIT_CREATED", "COMPLETE"}}
        if any(value > 1 for value in counts.values()):
            raise PipelineError("E_OUTBOX", "outbox event is duplicated")
        intent = next((row for row in events if row.get("event") == "DISPATCH_INTENT"), None)
        observed = next((row for row in events if row.get("event") == "OBSERVED"), None)
        git_intent = next((row for row in events if row.get("event") == "GIT_COMMIT_INTENT"), None)
        commit = next((row for row in events if row.get("event") == "COMMIT_CREATED"), None)
        complete = next((row for row in events if row.get("event") == "COMPLETE"), None)
        if intent is not None:
            if (
                kind == "GIT_DELIVERY_READY"
                or set(intent) != {"schema_version", "event", "outbox_id", "kind", "target_thread_id", "created_at"}
                or intent.get("kind") != kind
                or intent.get("target_thread_id") != _dispatch_target(config, kind)
                or not isinstance(intent.get("created_at"), str)
            ):
                raise PipelineError("E_OUTBOX", "dispatch intent binding differs")
        if observed is not None:
            if (
                intent is None
                or set(observed) != {"schema_version", "event", "outbox_id", "delivery_id", "observed_at"}
                or not isinstance(observed.get("delivery_id"), str)
                or not SAFE_ID.fullmatch(observed["delivery_id"])
                or not isinstance(observed.get("observed_at"), str)
            ):
                raise PipelineError("E_OUTBOX", "dispatch observation binding differs")
        if git_intent is not None:
            if (
                kind != "GIT_DELIVERY_READY"
                or set(git_intent) != {
                    "schema_version", "event", "outbox_id", "parent_sha", "tree_sha", "files",
                    "message", "author_name", "author_email", "authored_at", "created_at",
                }
                or not isinstance(git_intent.get("files"), list)
                or not isinstance(git_intent.get("message"), str)
                or not isinstance(git_intent.get("author_name"), str)
                or not isinstance(git_intent.get("author_email"), str)
                or not isinstance(git_intent.get("authored_at"), str)
                or not isinstance(git_intent.get("created_at"), str)
                or git_intent.get("files") != _git_expected_paths(origin.get("payload", {}))
            ):
                raise PipelineError("E_OUTBOX", "Git commit intent binding differs")
            for field in ("parent_sha", "tree_sha"):
                if not isinstance(git_intent.get(field), str) or not re.fullmatch(r"[0-9a-f]{40,64}", git_intent[field]):
                    raise PipelineError("E_OUTBOX", "Git commit intent identity differs")
        if commit is not None:
            if (
                kind != "GIT_DELIVERY_READY"
                or git_intent is None
                or set(commit) != {"schema_version", "event", "outbox_id", "parent_sha", "tree_sha", "commit_sha", "intent_sha256", "files", "created_at"}
                or not isinstance(commit.get("files"), list)
                or not isinstance(commit.get("created_at"), str)
                or commit.get("parent_sha") != git_intent.get("parent_sha")
                or commit.get("tree_sha") != git_intent.get("tree_sha")
                or commit.get("files") != git_intent.get("files")
                or commit.get("intent_sha256") != hashlib.sha256(_canonical(git_intent)).hexdigest().upper()
            ):
                raise PipelineError("E_OUTBOX", "Git commit binding differs")
            for field in ("parent_sha", "tree_sha", "commit_sha"):
                if not isinstance(commit.get(field), str) or not re.fullmatch(r"[0-9a-f]{40,64}", commit[field]):
                    raise PipelineError("E_OUTBOX", "Git commit identity differs")
        if complete is not None:
            result = complete.get("result")
            if not isinstance(complete.get("completed_at"), str):
                raise PipelineError("E_OUTBOX", "outbox completion time differs")
            if kind == "GIT_DELIVERY_READY":
                expected = {"schema_version", "event", "outbox_id", "result", "completed_at"}
                if result == "PUSHED":
                    expected.add("commit_sha")
                if set(complete) != expected or result not in {"PUSHED", "NO_CHANGES"}:
                    raise PipelineError("E_OUTBOX", "Git completion binding differs")
                if result == "PUSHED" and (commit is None or complete.get("commit_sha") != commit.get("commit_sha")):
                    raise PipelineError("E_OUTBOX", "Git completion commit differs")
            else:
                if (
                    intent is None or observed is None
                    or set(complete) != {"schema_version", "event", "outbox_id", "result", "terminal_id", "receipt_sha256", "completed_at"}
                    or result != ({"VIDEO_TRANSCRIPTION_READY": "TRANSCRIPTION_COMPLETE", "MINUTES_READY": "MINUTES_COMPLETE"}.get(kind))
                    or not isinstance(complete.get("terminal_id"), str)
                    or not SAFE_ID.fullmatch(complete["terminal_id"])
                    or not isinstance(complete.get("receipt_sha256"), str)
                    or not SHA256.fullmatch(complete["receipt_sha256"])
                ):
                    raise PipelineError("E_OUTBOX", "role completion binding differs")
        sequence = [row["event"] for row in events]
        if kind == "GIT_DELIVERY_READY":
            if intent is not None or observed is not None:
                raise PipelineError("E_OUTBOX", "Git outbox contains role events")
            if git_intent is not None and sequence.index("GIT_COMMIT_INTENT") <= 0:
                raise PipelineError("E_OUTBOX", "Git intent order differs")
            if commit is not None and sequence.index("COMMIT_CREATED") <= sequence.index("GIT_COMMIT_INTENT"):
                raise PipelineError("E_OUTBOX", "Git commit order differs")
            if complete is not None and sequence.index("COMPLETE") != len(sequence) - 1:
                raise PipelineError("E_OUTBOX", "Git completion order differs")
        else:
            if git_intent is not None or commit is not None:
                raise PipelineError("E_OUTBOX", "role outbox contains Git events")
            if intent is not None and sequence.index("DISPATCH_INTENT") <= 0:
                raise PipelineError("E_OUTBOX", "dispatch intent order differs")
            if observed is not None and (intent is None or sequence.index("OBSERVED") <= sequence.index("DISPATCH_INTENT")):
                raise PipelineError("E_OUTBOX", "dispatch observation order differs")
            if complete is not None and (observed is None or sequence.index("COMPLETE") <= sequence.index("OBSERVED") or sequence.index("COMPLETE") != len(sequence) - 1):
                raise PipelineError("E_OUTBOX", "role completion order differs")
    return rows


def _append_outbox(config: Config, kind: str, source: Mapping[str, Any], payload: Mapping[str, Any], created_at: str) -> str:
    expected_payload = _expected_outbox_payload(config, kind, source)
    if dict(payload) != expected_payload:
        raise PipelineError("E_OUTBOX", "new outbox payload differs from its immutable source")
    outbox_id = _outbox_id(kind, source, payload)
    rows = _outbox_rows(config)
    same_source = [
        row for row in rows
        if row.get("event") == "CREATED" and row.get("kind") == kind and row.get("source") == dict(source)
    ]
    if same_source and not (
        len(same_source) == 1
        and same_source[0].get("outbox_id") == outbox_id
        and same_source[0].get("payload") == dict(payload)
    ):
        raise PipelineError("E_OUTBOX", "source identity is already bound to another payload")
    prior = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id), None)
    if prior is not None:
        if prior.get("kind") != kind or prior.get("source") != dict(source) or prior.get("payload") != dict(payload) or prior.get("creator_uid") != config.creator_uid:
            raise PipelineError("E_OUTBOX", "existing outbox identity differs")
        return outbox_id
    event = {
        "schema_version": SCHEMA, "event": "CREATED", "outbox_id": outbox_id,
        "kind": kind, "creator_uid": config.creator_uid, "source": dict(source),
        "payload": dict(payload), "created_at": created_at,
    }
    _append(config.outbox_path, event)
    return outbox_id


def _row_digest(raw_line: bytes) -> str:
    return hashlib.sha256(raw_line).hexdigest().upper()


def _prefix_digest(payload: bytes, lines: int, field: str) -> str:
    if type(lines) is not int or lines < 0:
        raise PipelineError("E_STATE", f"{field} cursor line count differs")
    chunks = payload.splitlines(keepends=True)
    if len(chunks) < lines:
        raise PipelineError("E_HISTORY_REWRITE", f"{field} lost rows")
    return hashlib.sha256(b"".join(chunks[:lines])).hexdigest().upper()


def _creator_uid(row: Mapping[str, Any]) -> str | None:
    value = row.get("creator_uid")
    if type(value) is int:
        return str(value)
    if isinstance(value, str):
        return value
    creator = row.get("creator")
    if isinstance(creator, dict) and isinstance(creator.get("uid"), str):
        return creator["uid"]
    return None


def _is_reparse(info: os.stat_result) -> bool:
    return bool(getattr(info, "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))


def _stat_identity(info: os.stat_result) -> tuple[int, ...]:
    return (
        int(info.st_dev), int(info.st_ino), int(info.st_mode), int(info.st_nlink), int(info.st_size),
        int(info.st_mtime_ns), int(info.st_ctime_ns), int(getattr(info, "st_file_attributes", 0)),
    )


def _path_handle_identity(info: os.stat_result) -> tuple[int, ...]:
    identity = _stat_identity(info)
    return identity[:6] + identity[7:]


def _chain_identity(info: os.stat_result, *, final_file: bool) -> tuple[int, ...]:
    if final_file:
        return _stat_identity(info)
    return (int(info.st_dev), int(info.st_ino), int(info.st_mode), int(getattr(info, "st_file_attributes", 0)))


def _strict_chain(root: Path, target: Path, *, final_file: bool) -> tuple[tuple[str, tuple[int, ...]], ...]:
    root = Path(os.path.abspath(root))
    target = Path(os.path.abspath(target))
    if not _within(target, root):
        raise PipelineError("E_ARTIFACT", "path escaped its governed root")
    root_real = Path(os.path.realpath(root))
    target_real = Path(os.path.realpath(target))
    if not _within(target_real, root_real) or os.path.normcase(str(target_real)) != os.path.normcase(str(target)):
        raise PipelineError("E_ARTIFACT", "path resolution escaped or drifted")
    anchor = Path(target.anchor)
    paths: list[Path] = []
    current = anchor
    if str(anchor):
        paths.append(anchor)
    for part in target.parts[1:] if str(anchor) else target.parts:
        current = current / part
        paths.append(current)
    snapshots: list[tuple[str, tuple[int, ...]]] = []
    for index, item in enumerate(paths):
        try:
            info = os.lstat(item)
        except OSError as exc:
            raise PipelineError("E_ARTIFACT", "governed path is unavailable") from exc
        if stat.S_ISLNK(info.st_mode) or _is_reparse(info):
            raise PipelineError("E_ARTIFACT", "governed path contains a reparse object")
        is_final = index == len(paths) - 1
        if (is_final and final_file and not stat.S_ISREG(info.st_mode)) or ((not is_final or not final_file) and not stat.S_ISDIR(info.st_mode)):
            raise PipelineError("E_ARTIFACT", "governed path object type differs")
        snapshots.append((os.path.normcase(str(item)), _chain_identity(info, final_file=is_final and final_file)))
    return tuple(snapshots)


def _stable_artifact_bytes(target: Path, root: Path) -> bytes:
    before = _strict_chain(root, target, final_file=True)
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(target, flags)
    except OSError as exc:
        raise PipelineError("E_ARTIFACT", "artifact cannot be opened safely") from exc
    try:
        opened_before = os.fstat(descriptor)
        if _path_handle_identity(opened_before) != before[-1][1][:6] + before[-1][1][7:] or not stat.S_ISREG(opened_before.st_mode) or _is_reparse(opened_before):
            raise PipelineError("E_ARTIFACT", "artifact path and handle differ")
        with os.fdopen(descriptor, "rb", closefd=False) as stream:
            payload = stream.read()
        opened_after = os.fstat(descriptor)
        after = _strict_chain(root, target, final_file=True)
        if _stat_identity(opened_before) != _stat_identity(opened_after) or before != after:
            raise PipelineError("E_ARTIFACT", "artifact identity drifted during read")
        return payload
    finally:
        os.close(descriptor)


def _relocation_batch_id(report: Mapping[str, Any]) -> str:
    material = {
        "schema_version": report.get("schema_version"),
        "type": report.get("type"),
        "task_id": report.get("task_id"),
        "creator_uid": report.get("creator_uid"),
        "baseline_head": report.get("baseline_head"),
        "items": report.get("items"),
        "docs": report.get("docs"),
        "remove_paths": report.get("remove_paths"),
    }
    return hashlib.sha256(_canonical(material)).hexdigest().upper()


def _read_relocation_report(config: Config, path: Path) -> dict[str, Any]:
    _strict_chain(config.relocation_root, path, final_file=True)
    value, _ = _strict_json(path, "artifact relocation report")
    report = _exact(
        value,
        {
            "schema_version", "type", "task_id", "creator_uid", "batch_id", "created_at",
            "baseline_head", "items", "docs", "remove_paths",
        },
        "artifact relocation report",
    )
    if (
        type(report["schema_version"]) is not int
        or report["schema_version"] != SCHEMA
        or report["type"] != RELOCATION_REPORT_TYPE
        or report["task_id"] != TASK_ID
        or report["creator_uid"] != config.creator_uid
        or not isinstance(report["baseline_head"], str)
        or not re.fullmatch(r"[0-9a-f]{40,64}", report["baseline_head"])
        or not isinstance(report["batch_id"], str)
        or not SHA256.fullmatch(report["batch_id"])
        or report["batch_id"] != _relocation_batch_id(report)
        or path.name != f"{report['batch_id']}.json"
        or not isinstance(report["created_at"], str)
        or not isinstance(report["items"], list)
        or not report["items"]
        or not isinstance(report["docs"], list)
        or not isinstance(report["remove_paths"], list)
    ):
        raise PipelineError("E_RELOCATION", "artifact relocation report identity differs")
    aliases: list[dict[str, Any]] = []
    stable_ids: set[str] = set()
    item_ids: list[str] = []
    for item_value in report["items"]:
        item = _exact(
            item_value,
            {"stable_id", "title", "published_at", "canonical_base", "aliases", "intermediates"},
            "artifact relocation item",
        )
        stable_id = item["stable_id"]
        if (
            not isinstance(stable_id, str)
            or not BVID.fullmatch(stable_id)
            or stable_id in stable_ids
            or item["canonical_base"] != _canonical_video_base(stable_id, item["title"], item["published_at"])
            or not isinstance(item["aliases"], list)
            or not item["aliases"]
            or not isinstance(item["intermediates"], list)
        ):
            raise PipelineError("E_RELOCATION", "artifact relocation item identity differs")
        stable_ids.add(stable_id)
        item_ids.append(stable_id)
        canonical_base = item["canonical_base"]
        expected_alias_paths = {
            "transcript_txt": (
                config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.txt",
                config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.txt",
            ),
            "transcript_srt": (
                config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.srt",
                config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.srt",
            ),
            "transcript_json": (
                config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.json",
                config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.json",
            ),
            "minutes_md": (
                config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.md",
                config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.md",
            ),
            "minutes_pdf": (
                config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.pdf",
                config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.pdf",
            ),
        }
        item_kinds: set[str] = set()
        for alias_value in item["aliases"]:
            alias = _exact(alias_value, {"kind", "old_path", "new_path", "bytes", "sha256"}, "artifact relocation alias")
            kind = alias["kind"]
            if (
                kind not in PUBLIC_VIDEO_KINDS
                or not isinstance(alias["old_path"], str)
                or not isinstance(alias["new_path"], str)
                or alias["old_path"] == alias["new_path"]
                or type(alias["bytes"]) is not int
                or alias["bytes"] < 0
                or not isinstance(alias["sha256"], str)
                or not SHA256.fullmatch(alias["sha256"])
                or Path(alias["new_path"]).suffix.lower() != PUBLIC_VIDEO_KINDS[kind]
            ):
                raise PipelineError("E_RELOCATION", "artifact relocation alias differs")
            if kind in item_kinds:
                raise PipelineError("E_RELOCATION", "artifact relocation kind is duplicated")
            item_kinds.add(kind)
            expected_old, expected_new = expected_alias_paths[kind]
            if (
                alias["old_path"] != _project_relative(config, expected_old)
                or alias["new_path"] != _project_relative(config, expected_new)
            ):
                raise PipelineError("E_RELOCATION", "artifact relocation path grammar differs")
            for field in ("old_path", "new_path"):
                target = Path(os.path.abspath(config.project_root / Path(alias[field])))
                if Path(alias[field]).is_absolute() or not _within(target, config.archive_root):
                    raise PipelineError("E_RELOCATION", "artifact relocation path escaped the archive")
            aliases.append(dict(alias))
        transcript_kinds = {kind for kind in item_kinds if kind.startswith("transcript_")}
        minutes_kinds = {kind for kind in item_kinds if kind.startswith("minutes_")}
        if transcript_kinds != {"transcript_txt", "transcript_srt", "transcript_json"} or minutes_kinds not in (
            set(), {"minutes_md", "minutes_pdf"},
        ):
            raise PipelineError("E_RELOCATION", "artifact relocation kind set differs")
        if len(item["intermediates"]) > 1:
            raise PipelineError("E_RELOCATION", "artifact relocation intermediate set differs")
        for intermediate_value in item["intermediates"]:
            intermediate = _exact(
                intermediate_value,
                {"kind", "old_path", "new_path", "bytes", "sha256"},
                "artifact relocation intermediate",
            )
            if (
                intermediate["kind"] != "audio_flac"
                or not isinstance(intermediate["old_path"], str)
                or not isinstance(intermediate["new_path"], str)
                or type(intermediate["bytes"]) is not int
                or intermediate["bytes"] <= 0
                or not isinstance(intermediate["sha256"], str)
                or not SHA256.fullmatch(intermediate["sha256"])
                or not str(intermediate["new_path"]).lower().endswith(".audio.flac")
            ):
                raise PipelineError("E_RELOCATION", "artifact relocation intermediate differs")
            expected_old = config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.audio.flac"
            expected_new = config.video_root / "intermediate" / "transcription" / f"{stable_id}.audio.flac"
            if (
                intermediate["old_path"] != _project_relative(config, expected_old)
                or os.path.normcase(intermediate["new_path"]) != os.path.normcase(str(Path(os.path.abspath(expected_new))))
            ):
                raise PipelineError("E_RELOCATION", "artifact relocation intermediate grammar differs")
            old_target = Path(os.path.abspath(config.project_root / Path(intermediate["old_path"])))
            new_target = Path(os.path.abspath(Path(intermediate["new_path"])))
            if Path(intermediate["old_path"]).is_absolute() or not _within(old_target, config.archive_root) or not _within(new_target, config.video_root):
                raise PipelineError("E_RELOCATION", "artifact relocation intermediate escaped its boundary")
    if item_ids != sorted(item_ids):
        raise PipelineError("E_RELOCATION", "artifact relocation item order differs")
    old_paths = [alias["old_path"] for alias in aliases]
    new_paths = [alias["new_path"] for alias in aliases]
    if len(set(old_paths)) != len(old_paths) or len(set(new_paths)) != len(new_paths):
        raise PipelineError("E_RELOCATION", "artifact relocation aliases are duplicated")
    if (
        report["remove_paths"] != sorted(set(report["remove_paths"]))
        or any(value not in old_paths for value in report["remove_paths"])
    ):
        raise PipelineError("E_RELOCATION", "artifact relocation removal scope differs")
    docs: list[dict[str, Any]] = []
    for value in report["docs"]:
        doc = _exact(value, {"path", "bytes", "sha256", "kind"}, "artifact relocation document")
        if (
            doc["kind"] != "documentation"
            or doc["path"] not in config.git_doc_paths
            or type(doc["bytes"]) is not int
            or doc["bytes"] <= 0
            or not isinstance(doc["sha256"], str)
            or not SHA256.fullmatch(doc["sha256"])
        ):
            raise PipelineError("E_RELOCATION", "artifact relocation document differs")
        docs.append(dict(doc))
    if (
        len({value["path"] for value in docs}) != len(docs)
        or [value["path"] for value in docs] != sorted(config.git_doc_paths)
    ):
        raise PipelineError("E_RELOCATION", "artifact relocation document set differs")
    return dict(report)


def _relocation_reports(config: Config) -> list[tuple[Path, dict[str, Any]]]:
    if not config.relocation_root.exists():
        return []
    _strict_chain(config.archive_root, config.relocation_root, final_file=False)
    values: list[tuple[Path, dict[str, Any]]] = []
    for path in sorted(config.relocation_root.glob("*.json"), key=lambda value: value.name):
        values.append((path, _read_relocation_report(config, path)))
    return values


def _relocated_artifact_target(
    config: Config,
    old_relative: str,
    expected_bytes: int,
    expected_sha256: str,
) -> Path | None:
    matches: list[dict[str, Any]] = []
    for _, report in _relocation_reports(config):
        for item in report["items"]:
            matches.extend(
                alias for alias in item["aliases"]
                if alias["old_path"] == old_relative
                and alias["bytes"] == expected_bytes
                and alias["sha256"] == expected_sha256.upper()
            )
    if not matches:
        return None
    if len(matches) != 1:
        raise PipelineError("E_RELOCATION", "artifact relocation alias is ambiguous")
    return Path(os.path.abspath(config.project_root / Path(matches[0]["new_path"])))


def _artifact(path_value: Any, bytes_value: Any, sha_value: Any, config: Config) -> dict[str, Any]:
    if not isinstance(path_value, str) or not path_value or type(bytes_value) is not int or bytes_value < 0 or not isinstance(sha_value, str) or not SHA256.fullmatch(sha_value.upper()):
        raise PipelineError("E_ARTIFACT", "artifact identity is incomplete")
    path = Path(path_value)
    if path.is_absolute():
        requested = Path(os.path.abspath(path))
        targets = [requested] if requested.exists() or requested.is_symlink() else []
        if not targets and _within(requested, config.archive_root):
            relocated = _relocated_artifact_target(
                config,
                requested.relative_to(config.project_root).as_posix(),
                bytes_value,
                sha_value.upper(),
            )
            if relocated is not None:
                targets.append(relocated)
    else:
        candidates = [Path(os.path.abspath(config.archive_root / path)), Path(os.path.abspath(config.project_root / path))]
        requested_candidates = [candidate for candidate in candidates if _within(candidate, config.archive_root)]
        targets = [candidate for candidate in requested_candidates if candidate.exists() or candidate.is_symlink()]
        if not targets:
            for candidate in requested_candidates:
                relocated = _relocated_artifact_target(
                    config,
                    candidate.relative_to(config.project_root).as_posix(),
                    bytes_value,
                    sha_value.upper(),
                )
                if relocated is not None:
                    targets.append(relocated)
        requested = requested_candidates[0] if len(requested_candidates) == 1 else None
    distinct = {os.path.normcase(str(candidate)): candidate for candidate in targets}
    if len(distinct) != 1:
        raise PipelineError("E_ARTIFACT", "artifact path is absent or ambiguous")
    target = next(iter(distinct.values()))
    if not _within(target, config.archive_root) or any(str(target).lower().endswith(suffix) for suffix in FORBIDDEN_GIT_SUFFIXES):
        raise PipelineError("E_ARTIFACT", "artifact path is outside the Git boundary")
    payload = _stable_artifact_bytes(target, config.archive_root)
    digest = hashlib.sha256(payload).hexdigest().upper()
    if len(payload) != bytes_value or digest != sha_value.upper():
        raise PipelineError("E_ARTIFACT", "artifact readback differs")
    if path.is_absolute():
        relative = requested.relative_to(config.project_root).as_posix()
    else:
        logical_candidates = [
            candidate.relative_to(config.project_root).as_posix()
            for candidate in requested_candidates
            if _relocated_artifact_target(config, candidate.relative_to(config.project_root).as_posix(), bytes_value, sha_value.upper()) == target
            or candidate == target
        ]
        if len(set(logical_candidates)) != 1:
            raise PipelineError("E_ARTIFACT", "artifact logical path is ambiguous")
        relative = logical_candidates[0]
    if target.suffix.lower() not in config.git_extensions:
        raise PipelineError("E_ARTIFACT", "artifact extension is not allowlisted")
    return {"path": relative, "bytes": len(payload), "sha256": digest}


def _content_artifacts(row: Mapping[str, Any], config: Config) -> list[dict[str, Any]]:
    artifacts = [_artifact(row.get("path"), row.get("bytes"), row.get("sha256"), config)]
    if row.get("image_path") is not None:
        artifacts.append(_artifact(row.get("image_path"), row.get("image_bytes"), row.get("image_sha256"), config))
    images = row.get("images")
    if images is not None:
        if not isinstance(images, list):
            raise PipelineError("E_ARTIFACT", "images is not a list")
        for image in images:
            if not isinstance(image, dict):
                raise PipelineError("E_ARTIFACT", "image identity is invalid")
            candidate = _artifact(image.get("path"), image.get("bytes"), image.get("sha256"), config)
            if candidate not in artifacts:
                artifacts.append(candidate)
    return artifacts


def reconcile(config: Config, now: datetime) -> dict[str, Any]:
    with _lock(config):
        state = _load_state(config)
        active = state["active_run"]
        if active is None:
            raise PipelineError("E_NO_ACTIVE_RUN", "begin is required before reconcile")
        formal_rows, formal_payload = _read_lines(
            config.formal_manifest, "formal manifest", reject_secrets=False
        )
        handoff_rows, handoff_payload = _read_lines(
            config.processing_handoffs, "processing handoff", reject_secrets=False
        )
        cursors = state["cursors"]
        if (
            _prefix_digest(formal_payload, cursors["formal_lines"], "formal manifest") != cursors["formal_sha256"]
            or _prefix_digest(handoff_payload, cursors["handoff_lines"], "processing handoff") != cursors["handoff_sha256"]
        ):
            raise PipelineError("E_HISTORY_REWRITE", "append-only input prefix changed")
        created: list[str] = []
        formal_raw = formal_payload.splitlines()
        for index in range(cursors["formal_lines"], len(formal_rows)):
            row = formal_rows[index]
            if _creator_uid(row) != config.creator_uid:
                raise PipelineError("E_CREATOR", "new formal row creator differs")
            source = {"journal": "formal", "line": index + 1, "sha256": _row_digest(formal_raw[index])}
            item_type = row.get("item_type")
            if item_type in CONTENT_TYPES and row.get("status") == "SAVED":
                payload = _expected_outbox_payload(config, "GIT_DELIVERY_READY", source)
                created.append(_append_outbox(config, "GIT_DELIVERY_READY", source, payload, now.isoformat()))
            elif item_type == "video" and row.get("status") != VIDEO_COMPLETE:
                payload = _expected_outbox_payload(config, "VIDEO_DOWNLOAD_READY", source)
                created.append(_append_outbox(config, "VIDEO_DOWNLOAD_READY", source, payload, now.isoformat()))
        handoff_raw = handoff_payload.splitlines()
        for index in range(cursors["handoff_lines"], len(handoff_rows)):
            row = handoff_rows[index]
            if _creator_uid(row) != config.creator_uid or row.get("status") != "READY":
                raise PipelineError("E_HANDOFF", "processing handoff identity differs")
            stable_id = row.get("bvid")
            if not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
                raise PipelineError("E_HANDOFF", "processing handoff BVID differs")
            source = {"journal": "processing_handoff", "line": index + 1, "sha256": _row_digest(handoff_raw[index])}
            payload = _expected_outbox_payload(config, "VIDEO_TRANSCRIPTION_READY", source)
            created.append(_append_outbox(config, "VIDEO_TRANSCRIPTION_READY", source, payload, now.isoformat()))
        state["cursors"] = {
            "formal_lines": len(formal_rows), "formal_sha256": hashlib.sha256(formal_payload).hexdigest().upper(),
            "handoff_lines": len(handoff_rows), "handoff_sha256": hashlib.sha256(handoff_payload).hexdigest().upper(),
        }
        _write_state(config, state)
        return {"status": "RECONCILED", "run_id": active["run_id"], "created_outbox_ids": sorted(set(created)), "cursors": state["cursors"]}


def pending(config: Config) -> dict[str, Any]:
    with _lock(config):
        rows = _outbox_rows(config)
        latest: dict[str, str] = {}
        created: dict[str, dict[str, Any]] = {}
        for row in rows:
            outbox_id = row.get("outbox_id")
            if isinstance(outbox_id, str):
                latest[outbox_id] = str(row.get("event"))
                if row.get("event") == "CREATED":
                    created[outbox_id] = row
        values = []
        for key in sorted(created):
            state = latest.get(key)
            if state not in {"CREATED", "DISPATCH_INTENT"}:
                continue
            values.append({**created[key], "delivery_state": state})
        return {"status": "PENDING", "count": len(values), "items": values}


def _dispatch_target(config: Config, kind: str) -> str:
    targets = {
        "VIDEO_DOWNLOAD_READY": config.video_downloader_thread_id,
        "VIDEO_TRANSCRIPTION_READY": config.media_thread_id,
        "MINUTES_READY": config.minutes_thread_id,
    }
    target = targets.get(kind)
    if target is None:
        raise PipelineError("E_DISPATCH_KIND", "outbox item is not a role handoff")
    return target


def _dispatch_envelope(config: Config, created: Mapping[str, Any]) -> dict[str, Any]:
    return {
        "schema_version": SCHEMA,
        "type": created["kind"],
        "outbox_id": created["outbox_id"],
        "creator_uid": config.creator_uid,
        "payload": created["payload"],
    }


def dispatch_intent(config: Config, outbox_id: str, now: datetime) -> dict[str, Any]:
    with _lock(config):
        rows = _outbox_rows(config)
        matches = [row for row in rows if row.get("outbox_id") == outbox_id]
        created = next((row for row in matches if row.get("event") == "CREATED"), None)
        if created is None:
            raise PipelineError("E_OUTBOX", "outbox item is unknown")
        kind = created["kind"]
        target = _dispatch_target(config, kind)
        envelope = _dispatch_envelope(config, created)
        observed = next((row for row in matches if row.get("event") == "OBSERVED"), None)
        completed = next((row for row in matches if row.get("event") == "COMPLETE"), None)
        if completed is not None:
            return {
                "status": "DISPATCH_ALREADY_COMPLETE", "outbox_id": outbox_id,
                "target_thread_id": target, "envelope": envelope,
            }
        if observed is not None:
            return {
                "status": "DISPATCH_ALREADY_OBSERVED", "outbox_id": outbox_id,
                "target_thread_id": target, "delivery_id": observed["delivery_id"], "envelope": envelope,
            }
        prior = next((row for row in matches if row.get("event") == "DISPATCH_INTENT"), None)
        if prior is not None:
            if prior.get("kind") != kind or prior.get("target_thread_id") != target:
                raise PipelineError("E_OUTBOX", "dispatch intent identity drifted")
            return {
                "status": "DISPATCH_INTENT_RESUMED", "outbox_id": outbox_id,
                "target_thread_id": target, "envelope": envelope,
            }
        event = {"schema_version": SCHEMA, "event": "DISPATCH_INTENT", "outbox_id": outbox_id, "kind": kind, "target_thread_id": target, "created_at": now.isoformat()}
        _append(config.outbox_path, event)
        return {
            "status": "DISPATCH_INTENT_DURABLE", "outbox_id": outbox_id,
            "target_thread_id": target, "envelope": envelope,
        }


def observe_dispatch(config: Config, outbox_id: str, delivery_id: str, now: datetime) -> dict[str, Any]:
    if not isinstance(delivery_id, str) or not SAFE_ID.fullmatch(delivery_id):
        raise PipelineError("E_DISPATCH_RECEIPT", "delivery identity differs")
    with _lock(config):
        rows = _outbox_rows(config)
        matches = [row for row in rows if row.get("outbox_id") == outbox_id]
        if not any(row.get("event") == "DISPATCH_INTENT" for row in matches):
            raise PipelineError("E_DISPATCH_RECEIPT", "dispatch intent is absent")
        if any(row.get("event") == "COMPLETE" for row in matches):
            raise PipelineError("E_DISPATCH_RECEIPT", "completed dispatch cannot accept a late observation")
        observed = [row for row in matches if row.get("event") == "OBSERVED"]
        if observed:
            if len(observed) == 1 and observed[0].get("delivery_id") == delivery_id:
                return {"status": "DISPATCH_ALREADY_OBSERVED", "outbox_id": outbox_id, "delivery_id": delivery_id}
            raise PipelineError("E_DISPATCH_RECEIPT", "dispatch receipt conflicts")
        event = {
            "schema_version": SCHEMA, "event": "OBSERVED", "outbox_id": outbox_id,
            "delivery_id": delivery_id, "observed_at": now.isoformat(),
        }
        _append(config.outbox_path, event)
        return {"status": "DISPATCH_OBSERVED", "outbox_id": outbox_id, "delivery_id": delivery_id}


def _receipt_files(value: Any, config: Config) -> list[dict[str, Any]]:
    if not isinstance(value, list) or not value:
        raise PipelineError("E_RECEIPT", "receipt files are empty")
    files: list[dict[str, Any]] = []
    for item in value:
        item = _exact(item, {"path", "bytes", "sha256", "kind"}, "receipt file")
        if not isinstance(item["kind"], str) or not SAFE_ID.fullmatch(item["kind"]):
            raise PipelineError("E_RECEIPT", "receipt file kind differs")
        files.append({**_artifact(item["path"], item["bytes"], item["sha256"], config), "kind": item["kind"]})
    if len({item["path"] for item in files}) != len(files):
        raise PipelineError("E_RECEIPT", "receipt files are duplicated")
    return files


def _terminal_rows(config: Config) -> list[dict[str, Any]]:
    rows, _ = _read_lines(config.terminals_path, "terminals")
    seen_sources: set[str] = set()
    seen_terminals: set[str] = set()
    for row in rows:
        if set(row) != {
            "schema_version", "event", "source_outbox_id", "stable_id", "terminal_id",
            "receipt_bytes", "receipt_sha256", "files", "committed_at",
        }:
            raise PipelineError("E_RECEIPT_CONFLICT", "terminal shape differs")
        source = row.get("source_outbox_id")
        terminal = row.get("terminal_id")
        if (
            type(row.get("schema_version")) is not int
            or row.get("schema_version") != SCHEMA
            or row.get("event") not in {"TRANSCRIPTION_COMPLETE", "MINUTES_COMPLETE"}
            or not isinstance(source, str)
            or not re.fullmatch(r"[0-9a-f]{64}", source)
            or not isinstance(row.get("stable_id"), str)
            or not isinstance(terminal, str)
            or not SAFE_ID.fullmatch(terminal)
            or type(row.get("receipt_bytes")) is not int
            or row["receipt_bytes"] <= 0
            or not isinstance(row.get("receipt_sha256"), str)
            or not SHA256.fullmatch(row["receipt_sha256"])
            or not isinstance(row.get("files"), list)
            or not row["files"]
            or not isinstance(row.get("committed_at"), str)
        ):
            raise PipelineError("E_RECEIPT_CONFLICT", "terminal identity differs")
        if not BVID.fullmatch(row["stable_id"]):
            raise PipelineError("E_RECEIPT_CONFLICT", "terminal stable identity differs")
        rebound: list[dict[str, Any]] = []
        for item in row["files"]:
            item = _exact(item, {"path", "bytes", "sha256", "kind"}, "terminal file")
            if not isinstance(item["kind"], str) or not SAFE_ID.fullmatch(item["kind"]):
                raise PipelineError("E_RECEIPT_CONFLICT", "terminal file kind differs")
            rebound.append({**_artifact(item["path"], item["bytes"], item["sha256"], config), "kind": item["kind"]})
        if rebound != row["files"]:
            raise PipelineError("E_RECEIPT_CONFLICT", "terminal files drifted")
        if source in seen_sources or terminal in seen_terminals:
            raise PipelineError("E_RECEIPT_CONFLICT", "terminal identity is duplicated")
        seen_sources.add(source)
        seen_terminals.add(terminal)
    return rows


def ingest_receipt(config: Config, receipt_path: Path, now: datetime) -> dict[str, Any]:
    with _lock(config):
        value, raw = _strict_json(receipt_path, "downstream receipt")
        receipt = _exact(value, {"schema_version", "type", "outbox_id", "stable_id", "terminal_id", "status", "files", "created_at"}, "receipt")
        if (
            type(receipt["schema_version"]) is not int
            or receipt["schema_version"] != SCHEMA
            or receipt["status"] != "COMPLETE"
            or not isinstance(receipt["outbox_id"], str)
            or not isinstance(receipt["stable_id"], str)
            or not isinstance(receipt["terminal_id"], str)
            or not SAFE_ID.fullmatch(receipt["terminal_id"])
        ):
            raise PipelineError("E_RECEIPT", "receipt identity differs")
        kind = receipt["type"]
        if kind not in {"TRANSCRIPTION_COMPLETE", "MINUTES_COMPLETE"}:
            raise PipelineError("E_RECEIPT", "receipt type differs")
        rows = _outbox_rows(config)
        source = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == receipt["outbox_id"]), None)
        expected_kind = "VIDEO_TRANSCRIPTION_READY" if kind == "TRANSCRIPTION_COMPLETE" else "MINUTES_READY"
        if source is None or source.get("kind") != expected_kind:
            raise PipelineError("E_RECEIPT", "receipt source outbox differs")
        source_events = [row for row in rows if row.get("outbox_id") == receipt["outbox_id"]]
        if not any(row.get("event") == "DISPATCH_INTENT" for row in source_events):
            raise PipelineError("E_RECEIPT", "receipt has no durable dispatch intent")
        if not any(row.get("event") == "OBSERVED" for row in source_events):
            raise PipelineError("E_RECEIPT", "receipt arrived before durable dispatch observation")
        expected_stable_id = source.get("payload", {}).get("bvid" if kind == "TRANSCRIPTION_COMPLETE" else "stable_id")
        if receipt["stable_id"] != expected_stable_id:
            raise PipelineError("E_RECEIPT", "receipt stable identity differs")
        files = _receipt_files(receipt["files"], config)
        receipt_sha = hashlib.sha256(raw).hexdigest().upper()
        terminals = _terminal_rows(config)
        by_source = [item for item in terminals if item.get("source_outbox_id") == receipt["outbox_id"]]
        if by_source and not (len(by_source) == 1 and by_source[0].get("receipt_sha256") == receipt_sha):
            raise PipelineError("E_RECEIPT_CONFLICT", "source outbox already has a different terminal")
        if any(item.get("terminal_id") == receipt["terminal_id"] and item.get("source_outbox_id") != receipt["outbox_id"] for item in terminals):
            raise PipelineError("E_RECEIPT_CONFLICT", "terminal identity is already bound elsewhere")
        already_committed = bool(by_source)
        terminal = {
            "schema_version": SCHEMA, "event": kind, "source_outbox_id": receipt["outbox_id"],
            "stable_id": receipt["stable_id"], "terminal_id": receipt["terminal_id"],
            "receipt_bytes": len(raw), "receipt_sha256": receipt_sha, "files": files,
            "committed_at": now.isoformat(),
        }
        if not already_committed:
            _append(config.terminals_path, terminal)
        source_identity = {"journal": "terminals", "receipt_sha256": receipt_sha}
        created: list[str] = []
        if kind == "TRANSCRIPTION_COMPLETE":
            created.append(_append_outbox(config, "MINUTES_READY", source_identity, {
                "stable_id": receipt["stable_id"], "transcript_terminal_id": receipt["terminal_id"], "files": files
            }, now.isoformat()))
            created.append(_append_outbox(config, "GIT_DELIVERY_READY", {**source_identity, "projection": "transcript"}, {
                "stable_id": receipt["stable_id"], "files": files, "reason": "TRANSCRIPT_COMPLETE"
            }, now.isoformat()))
        else:
            created.append(_append_outbox(config, "GIT_DELIVERY_READY", {**source_identity, "projection": "minutes"}, {
                "stable_id": receipt["stable_id"], "files": files, "reason": "MINUTES_COMPLETE"
            }, now.isoformat()))
        if not any(row.get("event") == "COMPLETE" and row.get("outbox_id") == receipt["outbox_id"] for row in rows):
            _append(config.outbox_path, {
                "schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": receipt["outbox_id"],
                "result": kind, "terminal_id": receipt["terminal_id"],
                "receipt_sha256": receipt_sha, "completed_at": now.isoformat(),
            })
        return {
            "status": "RECEIPT_ALREADY_COMMITTED" if already_committed else "RECEIPT_COMMITTED",
            "receipt_sha256": receipt_sha,
            "created_outbox_ids": [] if already_committed else created,
        }


def finish(config: Config, status: str, now: datetime) -> dict[str, Any]:
    if status not in {"COMPLETE", "FAILED"}:
        raise PipelineError("E_SCHEMA", "finish status differs")
    with _lock(config):
        state = _load_state(config)
        active = state["active_run"]
        if active is None:
            raise PipelineError("E_NO_ACTIVE_RUN", "there is no active run")
        event = {
            "schema_version": SCHEMA, "event": f"RUN_{status}", "run_id": active["run_id"],
            "slot": active["slot"], "creator_uid": config.creator_uid, "finished_at": now.isoformat(),
        }
        _append(config.runs_path, event)
        state["active_run"] = None
        _write_state(config, state)
        return {"status": f"RUN_{status}", "run_id": active["run_id"]}


def _run(command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[str]]) -> subprocess.CompletedProcess[str]:
    return runner(list(command), cwd=cwd, text=True, capture_output=True, check=False)


def _run_env(
    command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[str]], env: Mapping[str, str],
) -> subprocess.CompletedProcess[str]:
    return runner(list(command), cwd=cwd, text=True, capture_output=True, check=False, env=dict(env))


def _run_env_bytes(
    command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[Any]], env: Mapping[str, str],
) -> subprocess.CompletedProcess[bytes]:
    return runner(list(command), cwd=cwd, text=False, capture_output=True, check=False, env=dict(env))


def _run_env_input(
    command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[str]], env: Mapping[str, str], payload: bytes,
) -> subprocess.CompletedProcess[str]:
    return runner(list(command), cwd=cwd, input=payload, capture_output=True, check=False, env=dict(env))


def _git_artifacts(config: Config, files: Any) -> tuple[list[str], list[dict[str, Any]], list[bytes]]:
    if not isinstance(files, list) or not files:
        raise PipelineError("E_GIT_SCOPE", "Git file list is empty")
    allowed: list[str] = []
    rebound: list[dict[str, Any]] = []
    blobs: list[bytes] = []
    for artifact in files:
        identity_keys = {"path", "bytes", "sha256"}
        if not isinstance(artifact, dict) or frozenset(artifact) not in {frozenset(identity_keys), frozenset(identity_keys | {"kind"})}:
            raise PipelineError("E_GIT_SCOPE", "Git artifact shape differs")
        path = artifact.get("path")
        size = artifact.get("bytes")
        digest = artifact.get("sha256")
        if not isinstance(path, str) or type(size) is not int or size < 0 or not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise PipelineError("E_GIT_SCOPE", "Git artifact identity differs")
        if Path(path).is_absolute():
            raise PipelineError("E_GIT_SCOPE", "Git path must be project-relative")
        requested = Path(os.path.abspath(config.project_root / Path(path)))
        target = requested
        if not target.exists() and not target.is_symlink():
            relocated = _relocated_artifact_target(config, Path(path).as_posix(), size, digest.upper())
            if relocated is not None:
                target = relocated
        relative = target.relative_to(config.project_root).as_posix() if _within(target, config.project_root) else ""
        if (
            not (_within(target, config.archive_root) or relative in config.git_doc_paths)
            or target.suffix.lower() not in config.git_extensions
            or any(str(target).lower().endswith(suffix) for suffix in FORBIDDEN_GIT_SUFFIXES)
        ):
            raise PipelineError("E_GIT_SCOPE", "Git path escaped its allowlist")
        if "kind" in artifact and RECEIPT_GIT_KIND_SUFFIX.get(artifact["kind"]) != target.suffix.lower():
            raise PipelineError("E_GIT_SCOPE", "Git receipt artifact kind differs")
        payload = _stable_artifact_bytes(target, config.archive_root if _within(target, config.archive_root) else config.project_root)
        current = {"path": relative, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper()}
        expected_identity = {"bytes": artifact["bytes"], "sha256": artifact["sha256"]}
        if {"bytes": current["bytes"], "sha256": current["sha256"]} != expected_identity:
            raise PipelineError("E_GIT_SCOPE", "Git artifact identity drifted")
        allowed.append(current["path"])
        rebound.append(current)
        blobs.append(payload)
    if len(set(allowed)) != len(allowed):
        raise PipelineError("E_GIT_SCOPE", "Git paths are duplicated")
    return allowed, rebound, blobs


def _git_remove_paths(config: Config, value: Any) -> list[str]:
    if value is None:
        return []
    if not isinstance(value, list) or not value or any(not isinstance(item, str) or not item for item in value):
        raise PipelineError("E_GIT_SCOPE", "Git removal scope differs")
    normalized: list[str] = []
    for item in value:
        if Path(item).is_absolute():
            raise PipelineError("E_GIT_SCOPE", "Git removal path must be project-relative")
        target = Path(os.path.abspath(config.project_root / Path(item)))
        if not _within(target, config.archive_root) or any(str(target).lower().endswith(suffix) for suffix in FORBIDDEN_GIT_SUFFIXES):
            raise PipelineError("E_GIT_SCOPE", "Git removal path escaped the archive")
        if target.exists() or target.is_symlink():
            raise PipelineError("E_GIT_SCOPE", "relocated Git source still exists in the worktree")
        ancestor = target.parent
        while not ancestor.exists() and ancestor != config.archive_root:
            ancestor = ancestor.parent
        _strict_chain(config.archive_root, ancestor, final_file=False)
        normalized.append(target.relative_to(config.project_root).as_posix())
    if normalized != sorted(set(normalized)):
        raise PipelineError("E_GIT_SCOPE", "Git removal paths are not canonical")
    return normalized


def _git_expected_paths(payload: Mapping[str, Any]) -> list[str]:
    files = payload.get("files")
    add_paths = [item.get("path") for item in files] if isinstance(files, list) else []
    removals = payload.get("remove_paths")
    if removals is None:
        return add_paths
    if not isinstance(removals, list):
        return []
    return sorted([*add_paths, *removals])


def _remove_task_index(path: Path, state_dir: Path) -> None:
    if not path.exists() and not path.is_symlink():
        return
    _strict_chain(state_dir, path, final_file=True)
    path.unlink()


def _git_batch(rows: Sequence[Mapping[str, Any]], item: Mapping[str, Any]) -> tuple[str, list[str]]:
    created_at = item.get("created_at")
    if not isinstance(created_at, str):
        raise PipelineError("E_GIT_INDEX_DIRTY", "Git batch creation identity differs")
    outbox_ids = sorted(
        row["outbox_id"] for row in rows
        if row.get("event") == "CREATED"
        and row.get("kind") == "GIT_DELIVERY_READY"
        and row.get("created_at") == created_at
    )
    if not outbox_ids or item.get("outbox_id") not in outbox_ids:
        raise PipelineError("E_GIT_INDEX_DIRTY", "Git batch membership differs")
    binding = {"created_at": created_at, "outbox_ids": outbox_ids}
    batch_id = hashlib.sha256(_canonical(binding)).hexdigest().upper()
    return batch_id, outbox_ids


def _decode_git_paths(result: subprocess.CompletedProcess[Any], error_code: str) -> list[str]:
    if result.returncode != 0 or not isinstance(result.stdout, bytes):
        raise PipelineError(error_code, "Git staged paths are unavailable")
    try:
        return [part.decode("utf-8").replace("\\", "/") for part in result.stdout.split(b"\0") if part]
    except UnicodeDecodeError as exc:
        raise PipelineError(error_code, "Git staged path encoding differs") from exc


def _shared_index_snapshot(
    config: Config,
    baseline_head: str,
    runner: Callable[..., subprocess.CompletedProcess[Any]],
) -> dict[str, Any]:
    if not isinstance(baseline_head, str) or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head):
        raise PipelineError("E_GIT_INDEX_DIRTY", "Git baseline HEAD identity differs")
    index_path = Path(os.path.abspath(config.project_root / ".git" / "index"))
    payload = _stable_artifact_bytes(index_path, config.project_root)
    chain = _strict_chain(config.project_root, index_path, final_file=True)
    staged = _run_env_bytes(
        ["git", "diff", "--cached", "--name-only", "-z", baseline_head, "--"],
        config.project_root,
        runner,
        os.environ,
    )
    return {
        "index_bytes": len(payload),
        "index_sha256": hashlib.sha256(payload).hexdigest().upper(),
        "index_identity": list(chain[-1][1]),
        "staged_paths": _decode_git_paths(staged, "E_GIT_INDEX_DIRTY"),
    }


def _git_guard_value(
    config: Config,
    batch_id: str,
    outbox_ids: Sequence[str],
    baseline_head: str,
    snapshot: Mapping[str, Any],
) -> dict[str, Any]:
    return {
        "schema_version": SCHEMA,
        "task_id": TASK_ID,
        "creator_uid": config.creator_uid,
        "batch_id": batch_id,
        "outbox_ids": list(outbox_ids),
        "baseline_head": baseline_head,
        "index_bytes": snapshot["index_bytes"],
        "index_sha256": snapshot["index_sha256"],
        "index_identity": snapshot["index_identity"],
        "staged_paths": snapshot["staged_paths"],
    }


@contextlib.contextmanager
def _open_no_write_handle(path: Path, error_code: str) -> Iterator[Any]:
    descriptor: int | None = None
    stream = None
    try:
        if os.name == "nt":
            import ctypes  # noqa: PLC0415
            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, 0x00000001, None, 3,
                0x00000080 | 0x00200000 | 0x08000000, None,
            )
            if handle in (None, ctypes.c_void_p(-1).value):
                raise OSError(ctypes.get_last_error(), "CreateFileW failed", str(path))
            try:
                descriptor = msvcrt.open_osfhandle(int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0))
                handle = None
            finally:
                if handle is not None:
                    close_handle(handle)
        else:
            descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
            import fcntl  # noqa: PLC0415
            fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
        stream = os.fdopen(descriptor, "rb", closefd=True)
        descriptor = None
        yield stream
    except PipelineError:
        raise
    except OSError as exc:
        raise PipelineError(error_code, "stable file handle is unavailable") from exc
    finally:
        if stream is not None:
            stream.close()
        elif descriptor is not None:
            os.close(descriptor)


def _read_exact_held_file(
    stream: Any,
    path: Path,
    root: Path,
    expected_chain: tuple[tuple[str, tuple[int, ...]], ...],
    error_code: str,
) -> bytes:
    opened_before = os.fstat(stream.fileno())
    stream.seek(0)
    payload = stream.read()
    opened_after = os.fstat(stream.fileno())
    chain = _strict_chain(root, path, final_file=True)
    if (
        not stat.S_ISREG(opened_before.st_mode)
        or _is_reparse(opened_before)
        or _path_handle_identity(opened_before) != expected_chain[-1][1][:6] + expected_chain[-1][1][7:]
        or _stat_identity(opened_before) != _stat_identity(opened_after)
        or chain != expected_chain
    ):
        raise PipelineError(error_code, "held file identity drifted")
    return payload


@contextlib.contextmanager
def _open_no_write_delete_handle(path: Path, error_code: str) -> Iterator[Any]:
    descriptor: int | None = None
    stream = None
    try:
        if os.name == "nt":
            import ctypes  # noqa: PLC0415
            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 | 0x00010000, 0x00000001, None, 3,
                0x00000080 | 0x00200000 | 0x08000000, None,
            )
            if handle in (None, ctypes.c_void_p(-1).value):
                raise OSError(ctypes.get_last_error(), "CreateFileW failed", str(path))
            try:
                descriptor = msvcrt.open_osfhandle(int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0))
                handle = None
            finally:
                if handle is not None:
                    close_handle(handle)
        else:
            descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
            import fcntl  # noqa: PLC0415
            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        stream = os.fdopen(descriptor, "rb", closefd=True)
        descriptor = None
        yield stream
    except PipelineError:
        raise
    except OSError as exc:
        raise PipelineError(error_code, "stable deletable file handle is unavailable") from exc
    finally:
        if stream is not None:
            stream.close()
        elif descriptor is not None:
            os.close(descriptor)


def _mark_held_file_for_delete(stream: Any, path: Path) -> None:
    if os.name == "nt":
        import ctypes  # noqa: PLC0415
        import msvcrt  # noqa: PLC0415

        class FileDispositionInfo(ctypes.Structure):
            _fields_ = [("DeleteFile", ctypes.c_int)]

        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
        setter = kernel32.SetFileInformationByHandle
        setter.argtypes = (ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32)
        setter.restype = ctypes.c_int
        info = FileDispositionInfo(1)
        handle = ctypes.c_void_p(msvcrt.get_osfhandle(stream.fileno()))
        if not setter(handle, 4, ctypes.byref(info), ctypes.sizeof(info)):
            raise OSError(ctypes.get_last_error(), "SetFileInformationByHandle failed", str(path))
        return
    os.unlink(path)


@contextlib.contextmanager
def _held_relocation_source(
    source: Path,
    source_root: Path,
    size: int,
    digest: str,
) -> Iterator[tuple[Any, bytes, tuple[tuple[str, tuple[int, ...]], ...]]]:
    before = _strict_chain(source_root, source, final_file=True)
    with _open_no_write_delete_handle(source, "E_RELOCATION") as stream:
        opened_before = os.fstat(stream.fileno())
        if (
            _path_handle_identity(opened_before) != before[-1][1][:6] + before[-1][1][7:]
            or not stat.S_ISREG(opened_before.st_mode)
            or _is_reparse(opened_before)
        ):
            raise PipelineError("E_RELOCATION", "migration source path and handle differ")
        stream.seek(0)
        payload = stream.read()
        opened_after = os.fstat(stream.fileno())
        after = _strict_chain(source_root, source, final_file=True)
        if (
            len(payload) != size
            or hashlib.sha256(payload).hexdigest().upper() != digest
            or _stat_identity(opened_before) != _stat_identity(opened_after)
            or before != after
        ):
            raise PipelineError("E_RELOCATION", "migration source identity differs")
        yield stream, payload, before


def _delete_held_relocation_source(
    stream: Any,
    source: Path,
    source_root: Path,
    payload: bytes,
    before: tuple[tuple[str, tuple[int, ...]], ...],
) -> None:
    stream.seek(0)
    rebound = stream.read()
    opened = os.fstat(stream.fileno())
    after = _strict_chain(source_root, source, final_file=True)
    if (
        rebound != payload
        or before != after
        or _path_handle_identity(opened) != after[-1][1][:6] + after[-1][1][7:]
    ):
        raise PipelineError("E_RELOCATION", "migration source drifted before handle-bound delete")
    try:
        _mark_held_file_for_delete(stream, source)
    except OSError as exc:
        raise PipelineError("E_RELOCATION", "migration source handle delete failed") from exc


@contextlib.contextmanager
def _held_exact_git_artifacts(config: Config, allowed: Sequence[str], blobs: Sequence[bytes]) -> Iterator[None]:
    with contextlib.ExitStack() as stack:
        for relative, expected in zip(allowed, blobs, strict=True):
            target = Path(os.path.abspath(config.project_root / relative))
            governed_root = config.archive_root if _within(target, config.archive_root) else config.project_root
            before = _strict_chain(governed_root, target, final_file=True)
            stream = stack.enter_context(_open_no_write_handle(target, "E_GIT_SCOPE"))
            opened_before = os.fstat(stream.fileno())
            if (
                _path_handle_identity(opened_before) != before[-1][1][:6] + before[-1][1][7:]
                or not stat.S_ISREG(opened_before.st_mode)
                or _is_reparse(opened_before)
            ):
                raise PipelineError("E_GIT_SCOPE", "published artifact path and handle differ")
            stream.seek(0)
            payload = stream.read()
            opened_after = os.fstat(stream.fileno())
            after = _strict_chain(governed_root, target, final_file=True)
            if payload != expected or _stat_identity(opened_before) != _stat_identity(opened_after) or before != after:
                raise PipelineError("E_GIT_SCOPE", "published artifact changed before terminal append")
        yield


@contextlib.contextmanager
def _held_git_ref_locks(
    config: Config,
    runner: Callable[..., subprocess.CompletedProcess[Any]],
) -> Iterator[None]:
    symbolic = _run(["git", "symbolic-ref", "-q", "HEAD"], config.project_root, runner)
    expected_local = f"refs/heads/{config.git_branch}"
    local_ref = symbolic.stdout.strip() if isinstance(symbolic.stdout, str) else ""
    git_dir_result = _run(["git", "rev-parse", "--absolute-git-dir"], config.project_root, runner)
    git_dir_text = git_dir_result.stdout.strip() if isinstance(git_dir_result.stdout, str) else ""
    if symbolic.returncode != 0 or local_ref != expected_local or git_dir_result.returncode != 0 or not git_dir_text:
        raise PipelineError("E_GIT_PUSH", "published branch ref identity differs")
    git_dir = Path(os.path.abspath(git_dir_text))
    _strict_chain(config.project_root, git_dir, final_file=False)
    lock_paths = sorted({
        git_dir / "HEAD.lock",
        git_dir / f"{expected_local}.lock",
        git_dir / f"refs/remotes/{config.git_remote}/{config.git_branch}.lock",
    }, key=lambda value: os.path.normcase(str(value)))
    with contextlib.ExitStack() as stack:
        for lock_path in lock_paths:
            _strict_chain(git_dir, lock_path.parent, final_file=False)
            try:
                _create_new(lock_path, b"mbx-published-no-changes-lock\n")
            except PipelineError as exc:
                raise PipelineError("E_GIT_PUSH", "published Git ref lock is unavailable") from exc

            def cleanup(path: Path = lock_path) -> None:
                with contextlib.suppress(OSError, PipelineError):
                    _strict_chain(git_dir, path, final_file=True)
                    path.unlink()

            stack.callback(cleanup)
            stream = stack.enter_context(_open_no_write_handle(lock_path, "E_GIT_PUSH"))
            if stream.read() != b"mbx-published-no-changes-lock\n":
                raise PipelineError("E_GIT_PUSH", "published Git ref lock identity differs")
        yield


@contextlib.contextmanager
def _published_head_with_exact_artifacts(
    config: Config,
    allowed: Sequence[str],
    blobs: Sequence[bytes],
    remove_paths: Sequence[str],
    runner: Callable[..., subprocess.CompletedProcess[Any]],
) -> Iterator[str | None]:
    head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
    head_sha = head.stdout.strip() if isinstance(head.stdout, str) else ""
    if head.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", head_sha):
        yield None
        return
    for relative, payload in zip(allowed, blobs, strict=True):
        committed = _run_env_bytes(
            ["git", "show", f"{head_sha}:{relative}"],
            config.project_root,
            runner,
            os.environ,
        )
        if committed.returncode != 0 or not isinstance(committed.stdout, bytes) or committed.stdout != payload:
            yield None
            return
    for relative in remove_paths:
        committed = _run_env_bytes(
            ["git", "show", f"{head_sha}:{relative}"],
            config.project_root,
            runner,
            os.environ,
        )
        if committed.returncode == 0:
            yield None
            return
    preliminary_remote = _run(
        ["git", "rev-parse", f"refs/remotes/{config.git_remote}/{config.git_branch}"],
        config.project_root,
        runner,
    )
    preliminary_remote_sha = preliminary_remote.stdout.strip() if isinstance(preliminary_remote.stdout, str) else ""
    if preliminary_remote.returncode != 0 or preliminary_remote_sha != head_sha:
        raise PipelineError("E_GIT_PUSH", "exact artifacts are not bound to the published branch")
    with _held_exact_git_artifacts(config, allowed, blobs), _held_git_ref_locks(config, runner):
        rebound_head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
        remote = _run(
            ["git", "rev-parse", f"refs/remotes/{config.git_remote}/{config.git_branch}"],
            config.project_root,
            runner,
        )
        rebound_sha = rebound_head.stdout.strip() if isinstance(rebound_head.stdout, str) else ""
        remote_sha = remote.stdout.strip() if isinstance(remote.stdout, str) else ""
        if rebound_head.returncode != 0 or rebound_sha != head_sha or remote.returncode != 0 or remote_sha != head_sha:
            raise PipelineError("E_GIT_PUSH", "exact artifacts are not bound to the stable published branch")
        for relative, payload in zip(allowed, blobs, strict=True):
            committed = _run_env_bytes(
                ["git", "show", f"{head_sha}:{relative}"],
                config.project_root,
                runner,
                os.environ,
            )
            if committed.returncode != 0 or not isinstance(committed.stdout, bytes) or committed.stdout != payload:
                raise PipelineError("E_GIT_SCOPE", "published artifact commit binding drifted")
        for relative in remove_paths:
            committed = _run_env_bytes(
                ["git", "show", f"{head_sha}:{relative}"],
                config.project_root,
                runner,
                os.environ,
            )
            target = Path(os.path.abspath(config.project_root / relative))
            if committed.returncode == 0 or target.exists() or target.is_symlink():
                raise PipelineError("E_GIT_SCOPE", "published relocation removal binding drifted")
        yield head_sha


def _read_git_guard(config: Config, path: Path) -> dict[str, Any]:
    value, _ = _strict_json(path, "Git shared-index guard")
    guard = _exact(value, {
        "schema_version", "task_id", "creator_uid", "batch_id", "outbox_ids", "baseline_head",
        "index_bytes", "index_sha256", "index_identity", "staged_paths",
    }, "Git shared-index guard")
    if (
        type(guard["schema_version"]) is not int or guard["schema_version"] != SCHEMA
        or guard["task_id"] != TASK_ID or guard["creator_uid"] != config.creator_uid
        or not isinstance(guard["batch_id"], str) or not SHA256.fullmatch(guard["batch_id"])
        or not isinstance(guard["outbox_ids"], list) or not guard["outbox_ids"]
        or any(not isinstance(value, str) or not SHA256.fullmatch(value.upper()) for value in guard["outbox_ids"])
        or not isinstance(guard["baseline_head"], str) or not re.fullmatch(r"[0-9a-f]{40,64}", guard["baseline_head"])
        or type(guard["index_bytes"]) is not int or guard["index_bytes"] < 0
        or not isinstance(guard["index_sha256"], str) or not SHA256.fullmatch(guard["index_sha256"])
        or not isinstance(guard["index_identity"], list) or not guard["index_identity"]
        or any(type(value) is not int for value in guard["index_identity"])
        or not isinstance(guard["staged_paths"], list)
        or any(not isinstance(value, str) for value in guard["staged_paths"])
    ):
        raise PipelineError("E_GIT_INDEX_DIRTY", "Git shared-index guard identity differs")
    return dict(guard)


def _ensure_git_index_guard(
    config: Config,
    rows: Sequence[Mapping[str, Any]],
    item: Mapping[str, Any],
    runner: Callable[..., subprocess.CompletedProcess[Any]],
) -> dict[str, Any]:
    batch_id, outbox_ids = _git_batch(rows, item)
    guard_path = config.git_index_guard_path(batch_id)
    if guard_path.exists() or guard_path.is_symlink():
        _strict_chain(config.state_dir, guard_path, final_file=True)
        guard = _read_git_guard(config, guard_path)
    else:
        batch_events = [
            row for row in rows
            if row.get("outbox_id") in outbox_ids and row.get("event") in {"GIT_COMMIT_INTENT", "COMMIT_CREATED", "COMPLETE"}
        ]
        if batch_events:
            raise PipelineError("E_GIT_INDEX_DIRTY", "Git baseline guard is absent after batch progress")
        head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
        baseline_head = head.stdout.strip() if isinstance(head.stdout, str) else ""
        if head.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head):
            raise PipelineError("E_GIT_INDEX_DIRTY", "Git baseline HEAD is unavailable")
        snapshot = _shared_index_snapshot(config, baseline_head, runner)
        guard = _git_guard_value(config, batch_id, outbox_ids, baseline_head, snapshot)
        _create_new(guard_path, _canonical(guard))
        _strict_chain(config.state_dir, guard_path, final_file=True)
        guard = _read_git_guard(config, guard_path)
    expected = _git_guard_value(
        config,
        batch_id,
        outbox_ids,
        guard["baseline_head"],
        _shared_index_snapshot(config, guard["baseline_head"], runner),
    )
    if guard != expected:
        raise PipelineError("E_GIT_INDEX_DIRTY", "shared Git index changed after the batch baseline")
    return guard


def recover_git_index_guard(
    config: Config,
    outbox_id: str,
    baseline_head: str,
    expected_bytes: int,
    expected_sha256: str,
    *,
    runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
) -> dict[str, Any]:
    with _lock(config):
        rows = _outbox_rows(config)
        item = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id), None)
        if item is None or item.get("kind") != "GIT_DELIVERY_READY":
            raise PipelineError("E_GIT_SCOPE", "Git recovery outbox differs")
        batch_id, outbox_ids = _git_batch(rows, item)
        guard_path = config.git_index_guard_path(batch_id)
        if (
            not isinstance(baseline_head, str) or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head)
            or type(expected_bytes) is not int or expected_bytes < 0
            or not isinstance(expected_sha256, str) or not SHA256.fullmatch(expected_sha256.upper())
        ):
            raise PipelineError("E_GIT_INDEX_DIRTY", "expected shared-index identity differs")
        expected_sha256 = expected_sha256.upper()
        if guard_path.exists() or guard_path.is_symlink():
            try:
                _strict_chain(config.state_dir, guard_path, final_file=True)
                guard = _read_git_guard(config, guard_path)
                if (
                    guard["batch_id"] != batch_id
                    or guard["outbox_ids"] != outbox_ids
                    or guard["baseline_head"] != baseline_head
                    or guard["index_bytes"] != expected_bytes
                    or guard["index_sha256"] != expected_sha256
                ):
                    raise PipelineError("E_GIT_INDEX_DIRTY", "durable Git guard recovery binding differs")
                fresh = _git_guard_value(
                    config,
                    batch_id,
                    outbox_ids,
                    guard["baseline_head"],
                    _shared_index_snapshot(config, guard["baseline_head"], runner),
                )
                if guard != fresh:
                    raise PipelineError("E_GIT_INDEX_DIRTY", "shared Git index changed after the durable guard")
            except PipelineError as exc:
                if exc.code == "E_GIT_INDEX_DIRTY":
                    raise
                raise PipelineError("E_GIT_INDEX_DIRTY", "durable Git guard cannot be revalidated") from exc
            return {"status": "GIT_INDEX_GUARD_ALREADY_DURABLE", "batch_id": batch_id}
        batch_rows = [row for row in rows if row.get("outbox_id") in outbox_ids]
        intents = [row for row in batch_rows if row.get("event") == "GIT_COMMIT_INTENT"]
        commits = [row for row in batch_rows if row.get("event") == "COMMIT_CREATED"]
        if not intents or len(intents) != len(commits) or intents[0].get("parent_sha") != baseline_head:
            raise PipelineError("E_GIT_INDEX_DIRTY", "Git recovery history does not bind the baseline")
        expected_parent = baseline_head
        for intent, commit in zip(intents, commits, strict=True):
            if intent.get("parent_sha") != expected_parent or commit.get("parent_sha") != expected_parent:
                raise PipelineError("E_GIT_INDEX_DIRTY", "Git recovery commit chain differs")
            expected_parent = commit.get("commit_sha")
        head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
        if head.returncode != 0 or head.stdout.strip() != expected_parent:
            raise PipelineError("E_GIT_INDEX_DIRTY", "Git recovery HEAD differs")
        snapshot = _shared_index_snapshot(config, baseline_head, runner)
        if snapshot["index_bytes"] != expected_bytes or snapshot["index_sha256"] != expected_sha256:
            raise PipelineError("E_GIT_INDEX_DIRTY", "shared Git index differs from its frozen first-run bytes")
        guard = _git_guard_value(config, batch_id, outbox_ids, baseline_head, snapshot)
        _create_new(guard_path, _canonical(guard))
        if _read_git_guard(config, guard_path) != guard:
            raise PipelineError("E_DURABILITY", "Git shared-index guard readback differs")
        return {"status": "GIT_INDEX_GUARD_RECOVERED", "batch_id": batch_id, "outbox_count": len(outbox_ids)}


def git_preflight(
    config: Config,
    *,
    runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
) -> dict[str, Any]:
    """Read-only shared-index diagnostics; never repairs or rewrites the index."""
    index_path = Path(os.path.abspath(config.project_root / ".git" / "index"))
    before_chain = _strict_chain(config.project_root, index_path, final_file=True)
    with _open_no_write_handle(index_path, "E_GIT_INDEX_DIRTY") as index_stream:
        before = _read_exact_held_file(
            index_stream,
            index_path,
            config.project_root,
            before_chain,
            "E_GIT_INDEX_DIRTY",
        )
        index_view = index_path
        if os.name != "nt":
            proc_view = Path(f"/proc/self/fd/{index_stream.fileno()}")
            if proc_view.exists():
                index_view = proc_view
        git_env = {
            **os.environ,
            "GIT_OPTIONAL_LOCKS": "0",
            "GIT_INDEX_FILE": str(index_view),
        }
        try:
            head_result = _run_env(["git", "rev-parse", "HEAD"], config.project_root, runner, git_env)
            head = head_result.stdout.strip() if isinstance(head_result.stdout, str) else ""
            if head_result.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", head):
                raise PipelineError("E_GIT_INDEX_DIRTY", "Git preflight HEAD is unavailable")
            staged = _run_env_bytes(
                ["git", "diff", "--cached", "--name-only", "-z", head, "--"],
                config.project_root,
                runner,
                git_env,
            )
            staged_paths = _decode_git_paths(staged, "E_GIT_INDEX_DIRTY")
            _read_exact_held_file(index_stream, index_path, config.project_root, before_chain, "E_GIT_INDEX_DIRTY")
            archive_relative = config.archive_root.relative_to(config.project_root).as_posix()
            deleted = _run_env_bytes(
                ["git", "diff", "--cached", "--diff-filter=D", "--name-only", "-z", head, "--", archive_relative],
                config.project_root,
                runner,
                git_env,
            )
            deleted_paths = _decode_git_paths(deleted, "E_GIT_INDEX_DIRTY")
            _read_exact_held_file(index_stream, index_path, config.project_root, before_chain, "E_GIT_INDEX_DIRTY")
            present_deletions: list[str] = []
            for relative in deleted_paths:
                target = Path(os.path.abspath(config.project_root / relative))
                if target.exists() or target.is_symlink():
                    _strict_chain(config.archive_root, target, final_file=True)
                    present_deletions.append(relative)
            lock_path = Path(os.path.abspath(config.project_root / ".git" / "index.lock"))
            stale_lock: dict[str, Any] | None = None
            if lock_path.exists() or lock_path.is_symlink():
                lock_payload = _stable_artifact_bytes(lock_path, config.project_root)
                lock_info = os.lstat(lock_path)
                stale_lock = {
                    "path": ".git/index.lock",
                    "bytes": len(lock_payload),
                    "sha256": hashlib.sha256(lock_payload).hexdigest().upper(),
                    "mtime_ns": lock_info.st_mtime_ns,
                }
            return {
                "status": "GIT_PREFLIGHT",
                "head": head,
                "index_bytes": len(before),
                "index_sha256": hashlib.sha256(before).hexdigest().upper(),
                "index_identity": list(before_chain[-1][1]),
                "index_matches_head": not staged_paths,
                "staged_paths": staged_paths,
                "archive_staged_delete_present": present_deletions,
                "stale_index_lock": stale_lock,
            }
        finally:
            after = _read_exact_held_file(
                index_stream,
                index_path,
                config.project_root,
                before_chain,
                "E_GIT_INDEX_MUTATION",
            )
            if before != after:
                raise PipelineError("E_GIT_INDEX_MUTATION", "Git preflight changed the shared index")


def _migration_file_identity(config: Config, path: Path, kind: str, root: Path) -> dict[str, Any]:
    payload = _stable_artifact_bytes(path, root)
    return {
        "kind": kind,
        "bytes": len(payload),
        "sha256": hashlib.sha256(payload).hexdigest().upper(),
    }


def _video_metadata(config: Config) -> dict[str, dict[str, Any]]:
    rows, _ = _read_lines(config.formal_manifest, "formal manifest", reject_secrets=False)
    selected: dict[str, dict[str, Any]] = {}
    for row in rows:
        stable_id = row.get("stable_id")
        row_creator_uid = _creator_uid(row)
        creator_matches = row_creator_uid == config.creator_uid or (
            row_creator_uid is None and row.get("creator") == config.creator_name
        )
        if (
            row.get("item_type") != "video"
            or not creator_matches
            or not isinstance(stable_id, str)
            or not BVID.fullmatch(stable_id)
        ):
            continue
        title = row.get("title")
        published_at = row.get("published_at")
        if not isinstance(title, str) or not isinstance(published_at, str):
            continue
        _canonical_video_base(stable_id, title, published_at)
        candidate = {
            "stable_id": stable_id,
            "title": title,
            "published_at": published_at,
            "complete": row.get("status") == VIDEO_COMPLETE,
        }
        existing = selected.get(stable_id)
        if existing is not None and existing["complete"] and candidate["complete"] and (
            existing["title"] != title or existing["published_at"] != published_at
        ):
            raise PipelineError("E_RELOCATION", "completed video metadata is ambiguous")
        if candidate["complete"] or existing is None:
            selected[stable_id] = candidate
    return selected


def _video_artifact_ids(config: Config) -> list[str]:
    values: set[str] = set()
    for suffix in (".transcript", ".minutes"):
        for path in config.archive_root.glob(f"*{suffix}"):
            name = path.name.removesuffix(suffix)
            stable_id = name if BVID.fullmatch(name) else name.rsplit("_", 1)[-1]
            if path.is_dir() and BVID.fullmatch(stable_id):
                _strict_chain(config.archive_root, path, final_file=False)
                values.add(stable_id)
    return sorted(values)


def _git_tracked(config: Config, relative: str, runner: Callable[..., subprocess.CompletedProcess[Any]]) -> bool:
    result = _run(["git", "ls-files", "--error-unmatch", "--", relative], config.project_root, runner)
    return result.returncode == 0


def _git_head(config: Config, runner: Callable[..., subprocess.CompletedProcess[Any]]) -> str:
    result = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
    head = result.stdout.strip() if isinstance(result.stdout, str) else ""
    if result.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", head):
        raise PipelineError("E_RELOCATION", "migration Git baseline is unavailable")
    exists = _run(["git", "cat-file", "-e", f"{head}^{{commit}}"], config.project_root, runner)
    if exists.returncode != 0:
        raise PipelineError("E_RELOCATION", "migration Git baseline differs")
    return head


def _git_blob_at(
    config: Config,
    baseline_head: str,
    relative: str,
    runner: Callable[..., subprocess.CompletedProcess[Any]],
) -> bytes | None:
    result = _run_env_bytes(
        ["git", "show", f"{baseline_head}:{relative}"],
        config.project_root,
        runner,
        {**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
    )
    if result.returncode != 0:
        return None
    if not isinstance(result.stdout, bytes):
        raise PipelineError("E_RELOCATION", "migration Git blob encoding differs")
    return result.stdout


def _migration_current_identity(
    config: Config,
    old_path: Path,
    new_path: Path,
    kind: str,
    old_root: Path,
    new_root: Path,
    *,
    allow_recovery_pair: bool,
) -> dict[str, Any]:
    old_exists = old_path.exists() or old_path.is_symlink()
    new_exists = new_path.exists() or new_path.is_symlink()
    if not old_exists and not new_exists:
        raise PipelineError("E_RELOCATION", "migration artifact is absent")
    if old_exists and new_exists and not allow_recovery_pair:
        raise PipelineError("E_RELOCATION", "legacy and canonical artifact both exist")
    identities: list[dict[str, Any]] = []
    if old_exists:
        identities.append(_migration_file_identity(config, old_path, kind, old_root))
    if new_exists:
        identities.append(_migration_file_identity(config, new_path, kind, new_root))
    if len(identities) == 2 and identities[0] != identities[1]:
        raise PipelineError("E_RELOCATION", "migration recovery pair differs")
    return identities[0]


def _build_video_artifact_migration_report(
    config: Config,
    created_at: str,
    baseline_head: str,
    stable_ids: Sequence[str],
    runner: Callable[..., subprocess.CompletedProcess[Any]],
) -> dict[str, Any]:
    _published_at(created_at, "created_at")
    if (
        not isinstance(baseline_head, str)
        or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head)
        or list(stable_ids) != sorted(set(stable_ids))
        or not stable_ids
    ):
        raise PipelineError("E_RELOCATION", "migration plan identity differs")
    exists = _run(["git", "cat-file", "-e", f"{baseline_head}^{{commit}}"], config.project_root, runner)
    if exists.returncode != 0:
        raise PipelineError("E_RELOCATION", "migration baseline commit is absent")
    metadata = _video_metadata(config)
    items: list[dict[str, Any]] = []
    remove_paths: list[str] = []
    kind_order = tuple(PUBLIC_VIDEO_KINDS)
    for stable_id in stable_ids:
        if not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
            raise PipelineError("E_RELOCATION", "migration stable identity differs")
        source = metadata.get(stable_id)
        if source is None or not source["complete"]:
            raise PipelineError("E_RELOCATION", "legacy video lacks a completed formal identity")
        canonical_base = _canonical_video_base(stable_id, source["title"], source["published_at"])
        aliases: list[dict[str, Any]] = []
        intermediates: list[dict[str, Any]] = []
        locations = {
            "transcript_txt": (
                config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.txt",
                config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.txt",
            ),
            "transcript_srt": (
                config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.srt",
                config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.srt",
            ),
            "transcript_json": (
                config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.json",
                config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.json",
            ),
            "minutes_md": (
                config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.md",
                config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.md",
            ),
            "minutes_pdf": (
                config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.pdf",
                config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.pdf",
            ),
        }
        for kind in kind_order:
            old_path, new_path = locations[kind]
            old_exists = old_path.exists() or old_path.is_symlink()
            new_exists = new_path.exists() or new_path.is_symlink()
            if not old_exists and not new_exists:
                continue
            identity = _migration_current_identity(
                config,
                old_path,
                new_path,
                kind,
                config.archive_root,
                config.archive_root,
                allow_recovery_pair=True,
            )
            old_relative = _project_relative(config, old_path)
            new_relative = _project_relative(config, new_path)
            aliases.append({
                "kind": kind,
                "old_path": old_relative,
                "new_path": new_relative,
                "bytes": identity["bytes"],
                "sha256": identity["sha256"],
            })
            committed = _git_blob_at(config, baseline_head, old_relative, runner)
            if committed is not None:
                remove_paths.append(old_relative)
        transcript_kinds = {alias["kind"] for alias in aliases if alias["kind"].startswith("transcript_")}
        minutes_kinds = {alias["kind"] for alias in aliases if alias["kind"].startswith("minutes_")}
        if transcript_kinds != {"transcript_txt", "transcript_srt", "transcript_json"}:
            raise PipelineError("E_RELOCATION", "legacy video transcript set is incomplete")
        if minutes_kinds not in (set(), {"minutes_md", "minutes_pdf"}):
            raise PipelineError("E_RELOCATION", "legacy video minutes set is incomplete")
        old_flac = config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.audio.flac"
        new_flac = config.video_root / "intermediate" / "transcription" / f"{stable_id}.audio.flac"
        if old_flac.exists() or old_flac.is_symlink() or new_flac.exists() or new_flac.is_symlink():
            identity = _migration_current_identity(
                config,
                old_flac,
                new_flac,
                "audio_flac",
                config.archive_root,
                config.video_root,
                allow_recovery_pair=True,
            )
            intermediates.append({
                "kind": "audio_flac",
                "old_path": _project_relative(config, old_flac),
                "new_path": str(Path(os.path.abspath(new_flac))),
                "bytes": identity["bytes"],
                "sha256": identity["sha256"],
            })
        items.append({
            "stable_id": stable_id,
            "title": source["title"],
            "published_at": source["published_at"],
            "canonical_base": canonical_base,
            "aliases": aliases,
            "intermediates": intermediates,
        })
    docs: list[dict[str, Any]] = []
    for relative in sorted(config.git_doc_paths):
        path = Path(os.path.abspath(config.project_root / relative))
        identity = _migration_file_identity(config, path, "documentation", config.project_root)
        docs.append({"path": relative, **identity})
    report: dict[str, Any] = {
        "schema_version": SCHEMA,
        "type": RELOCATION_REPORT_TYPE,
        "task_id": TASK_ID,
        "creator_uid": config.creator_uid,
        "batch_id": "",
        "created_at": created_at,
        "baseline_head": baseline_head,
        "items": items,
        "docs": docs,
        "remove_paths": sorted(set(remove_paths)),
    }
    report["batch_id"] = _relocation_batch_id(report)
    return report


def _relocation_outbox_exists(config: Config, report_path: Path, report: Mapping[str, Any]) -> bool:
    if not config.outbox_path.exists():
        return False
    payload = _stable_artifact_bytes(report_path, config.relocation_root)
    source = {
        "journal": "relocation_report",
        "path": _project_relative(config, report_path),
        "bytes": len(payload),
        "sha256": hashlib.sha256(payload).hexdigest().upper(),
    }
    expected_payload = _project_relocation_payload(config, source)
    expected_id = _outbox_id("GIT_DELIVERY_READY", source, expected_payload)
    rows = _outbox_rows(config)
    created = [
        row for row in rows
        if row.get("event") == "CREATED" and row.get("outbox_id") == expected_id
    ]
    if len(created) > 1:
        raise PipelineError("E_RELOCATION", "relocation outbox is duplicated")
    return len(created) == 1


def plan_video_artifact_migration(
    config: Config,
    now: datetime,
    *,
    runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
) -> dict[str, Any]:
    reports = _relocation_reports(config)
    stable_ids = _video_artifact_ids(config)
    if reports:
        report_path, existing = reports[-1]
        report_ids = [item["stable_id"] for item in existing["items"]]
        outbox_exists = _relocation_outbox_exists(config, report_path, existing)
        if not outbox_exists and not set(stable_ids).issubset(set(report_ids)):
            raise PipelineError("E_RELOCATION", "durable relocation report omitted a legacy identity")
        rebound = _build_video_artifact_migration_report(
            config,
            existing["created_at"],
            existing["baseline_head"],
            report_ids,
            runner,
        )
        if rebound != existing:
            raise PipelineError("E_RELOCATION", "durable relocation report differs from the external plan")
        if not outbox_exists:
            return existing
        new_ids = sorted(set(stable_ids) - set(report_ids))
        if not new_ids:
            return existing
        historical_ids = {
            item["stable_id"]
            for _, report in reports
            for item in report["items"]
        }
        if any(stable_id in historical_ids for stable_id in new_ids):
            raise PipelineError("E_RELOCATION", "legacy video identity was already migrated")
        stable_ids = new_ids
    if not stable_ids:
        raise PipelineError("E_RELOCATION", "no legacy video artifacts were found")
    return _build_video_artifact_migration_report(
        config,
        now.isoformat(),
        _git_head(config, runner),
        stable_ids,
        runner,
    )


def _ensure_directory_chain(root: Path, target: Path) -> None:
    if not _within(target, root):
        raise PipelineError("E_RELOCATION", "migration directory escaped its root")
    missing: list[Path] = []
    current = target
    while not current.exists() and current != root:
        missing.append(current)
        current = current.parent
    _strict_chain(root, current, final_file=False)
    for path in reversed(missing):
        path.mkdir()
        _strict_chain(root, path, final_file=False)


def _move_relocation_file(
    source_root: Path,
    target_root: Path,
    source: Path,
    target: Path,
    size: int,
    digest: str,
) -> None:
    source_exists = source.exists() or source.is_symlink()
    target_exists = target.exists() or target.is_symlink()
    if not source_exists:
        if not target_exists:
            raise PipelineError("E_RELOCATION", "migration source and target are absent")
        payload = _stable_artifact_bytes(target, target_root)
        if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
            raise PipelineError("E_RELOCATION", "migration target identity differs")
        return
    with _held_relocation_source(source, source_root, size, digest) as (stream, payload, source_chain):
        if target_exists:
            rebound = _stable_artifact_bytes(target, target_root)
            if rebound != payload:
                raise PipelineError("E_RELOCATION", "migration recovery target differs")
        else:
            _ensure_directory_chain(target_root, target.parent)
            _create_new(target, payload)
            rebound = _stable_artifact_bytes(target, target_root)
            if rebound != payload:
                raise PipelineError("E_RELOCATION", "migration target readback differs")
        _delete_held_relocation_source(stream, source, source_root, payload, source_chain)
    if source.exists() or source.is_symlink():
        raise PipelineError("E_RELOCATION", "migration source remained after handle-bound delete")


def migrate_video_artifacts(
    config: Config,
    now: datetime,
    *,
    runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
) -> dict[str, Any]:
    with _lock(config):
        state = _load_state(config)
        if state["active_run"] is not None:
            raise PipelineError("E_RUN_ACTIVE", "artifact migration requires no active half-hour run")
        report = plan_video_artifact_migration(config, now, runner=runner)
        report_path = config.relocation_root / f"{report['batch_id']}.json"
        if not report_path.exists() and not report_path.is_symlink():
            _ensure_directory_chain(config.archive_root, config.relocation_root)
            _create_new(report_path, _canonical(report))
        rebound = _read_relocation_report(config, report_path)
        if rebound != report:
            raise PipelineError("E_RELOCATION", "durable relocation report differs")
        for item in report["items"]:
            for alias in item["aliases"]:
                source = Path(os.path.abspath(config.project_root / alias["old_path"]))
                target = Path(os.path.abspath(config.project_root / alias["new_path"]))
                _move_relocation_file(
                    config.archive_root,
                    config.archive_root,
                    source,
                    target,
                    alias["bytes"],
                    alias["sha256"],
                )
            for intermediate in item["intermediates"]:
                source = Path(os.path.abspath(config.project_root / intermediate["old_path"]))
                target = Path(os.path.abspath(intermediate["new_path"]))
                _move_relocation_file(
                    config.archive_root,
                    config.video_root,
                    source,
                    target,
                    intermediate["bytes"],
                    intermediate["sha256"],
                )
            for suffix in (".transcript", ".minutes"):
                legacy = config.archive_root / f"{item['stable_id']}{suffix}"
                if legacy.exists():
                    _strict_chain(config.archive_root, legacy, final_file=False)
                    try:
                        legacy.rmdir()
                    except OSError as exc:
                        raise PipelineError("E_RELOCATION", "legacy artifact directory is not empty") from exc
        completed = _build_video_artifact_migration_report(
            config,
            report["created_at"],
            report["baseline_head"],
            [item["stable_id"] for item in report["items"]],
            runner,
        )
        if completed != report:
            raise PipelineError("E_RELOCATION", "completed migration differs from its durable external plan")
        report_payload = _stable_artifact_bytes(report_path, config.relocation_root)
        source = {
            "journal": "relocation_report",
            "path": _project_relative(config, report_path),
            "bytes": len(report_payload),
            "sha256": hashlib.sha256(report_payload).hexdigest().upper(),
        }
        payload = _project_relocation_payload(config, source)
        outbox_id = _append_outbox(config, "GIT_DELIVERY_READY", source, payload, report["created_at"])
        return {
            "status": "VIDEO_ARTIFACTS_MIGRATED",
            "batch_id": report["batch_id"],
            "report": source,
            "item_count": len(report["items"]),
            "public_file_count": sum(len(item["aliases"]) for item in report["items"]),
            "intermediate_count": sum(len(item["intermediates"]) for item in report["items"]),
            "git_outbox_id": outbox_id,
        }


def git_deliver(config: Config, outbox_id: str, now: datetime, *, runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run) -> dict[str, Any]:
    with _lock(config):
        rows = _outbox_rows(config)
        item = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id), None)
        if item is None or item.get("kind") != "GIT_DELIVERY_READY":
            raise PipelineError("E_GIT_SCOPE", "Git outbox differs")
        if any(row.get("event") == "COMPLETE" and row.get("outbox_id") == outbox_id for row in rows):
            return {"status": "GIT_ALREADY_COMPLETE", "outbox_id": outbox_id}
        _ensure_git_index_guard(config, rows, item, runner)
        payload = item.get("payload", {})
        files = payload.get("files")
        allowed, _, blobs = _git_artifacts(config, files)
        remove_paths = _git_remove_paths(config, payload.get("remove_paths"))
        expected_paths = _git_expected_paths(payload)
        if expected_paths != (allowed if not remove_paths else sorted([*allowed, *remove_paths])):
            raise PipelineError("E_GIT_SCOPE", "Git staged scope differs from its payload")
        with _published_head_with_exact_artifacts(config, allowed, blobs, remove_paths, runner) as published_head:
            if published_head is not None:
                _append(config.outbox_path, {
                    "schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id,
                    "result": "NO_CHANGES", "completed_at": now.isoformat(),
                })
                return {
                    "status": "GIT_NO_CHANGES", "outbox_id": outbox_id,
                    "commit_sha": published_head, "files": allowed,
                }
        git_intents = [row for row in rows if row.get("event") == "GIT_COMMIT_INTENT" and row.get("outbox_id") == outbox_id]
        commit_events = [row for row in rows if row.get("event") == "COMMIT_CREATED" and row.get("outbox_id") == outbox_id]
        if len(git_intents) > 1 or len(commit_events) > 1:
            raise PipelineError("E_GIT_COMMIT", "Git commit journal is ambiguous")
        if commit_events:
            event = commit_events[0]
            commit_sha = event.get("commit_sha")
            parent_sha = event.get("parent_sha")
            if (
                not isinstance(commit_sha, str) or not re.fullmatch(r"[0-9a-f]{40,64}", commit_sha)
                or not isinstance(parent_sha, str) or not re.fullmatch(r"[0-9a-f]{40,64}", parent_sha)
                or event.get("files") != expected_paths
            ):
                raise PipelineError("E_GIT_COMMIT", "Git commit journal identity differs")
            exists = _run(["git", "cat-file", "-e", f"{commit_sha}^{{commit}}"], config.project_root, runner)
            head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
            current_head = head.stdout.strip()
            if exists.returncode != 0 or head.returncode != 0 or current_head not in {parent_sha, commit_sha}:
                raise PipelineError("E_GIT_COMMIT", "Git recovery identity differs")
            if current_head == parent_sha:
                _git_artifacts(config, files)
                _git_remove_paths(config, payload.get("remove_paths"))
                updated = _run(["git", "update-ref", "HEAD", commit_sha, parent_sha], config.project_root, runner)
                if updated.returncode != 0:
                    raise PipelineError("E_GIT_COMMIT", "Git recovery ref update failed")
            pushed = _run(["git", "push", config.git_remote, f"{commit_sha}:refs/heads/{config.git_branch}"], config.project_root, runner)
            if pushed.returncode != 0:
                raise PipelineError("E_GIT_PUSH", "non-force push failed")
            _append(config.outbox_path, {
                "schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id,
                "result": "PUSHED", "commit_sha": commit_sha, "completed_at": now.isoformat(),
            })
            return {"status": "GIT_PUSHED", "outbox_id": outbox_id, "commit_sha": commit_sha, "files": allowed}
        task_index = config.state_dir / f"git-index-{outbox_id}"
        task_lock = Path(str(task_index) + ".lock")
        task_env = dict(os.environ)
        task_env["GIT_INDEX_FILE"] = str(task_index)
        try:
            git_intent = git_intents[0] if git_intents else None
            if git_intent is None:
                _remove_task_index(task_index, config.state_dir)
                _remove_task_index(task_lock, config.state_dir)
                read_tree = _run_env(["git", "read-tree", "HEAD"], config.project_root, runner, task_env)
                added_ok = True
                for relative, artifact_payload in zip(allowed, blobs, strict=True):
                    blob = _run_env_input(["git", "hash-object", "-w", "--stdin"], config.project_root, runner, task_env, artifact_payload)
                    blob_sha = blob.stdout.decode("ascii").strip() if isinstance(blob.stdout, bytes) else blob.stdout.strip()
                    if blob.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", blob_sha):
                        added_ok = False
                        break
                    indexed = _run_env(["git", "update-index", "--add", "--cacheinfo", f"100644,{blob_sha},{relative}"], config.project_root, runner, task_env)
                    if indexed.returncode != 0:
                        added_ok = False
                        break
                for relative in remove_paths:
                    removed = _run_env(
                        ["git", "update-index", "--force-remove", "--", relative],
                        config.project_root,
                        runner,
                        task_env,
                    )
                    if removed.returncode != 0:
                        added_ok = False
                        break
                # `-z` returns repository path bytes without C quoting or console
                # code-page conversion.  Decode Git's UTF-8 path contract directly
                # before enforcing the exact task-index allowlist.
                staged_command = ["git", "diff", "--cached", "--name-only", "-z"]
                if remove_paths:
                    staged_command.append("--no-renames")
                staged_command.append("--")
                staged = _run_env_bytes(
                    staged_command,
                    config.project_root,
                    runner,
                    task_env,
                )
                try:
                    staged_paths = [part.decode("utf-8").replace("\\", "/") for part in staged.stdout.split(b"\0") if part]
                except (AttributeError, UnicodeDecodeError) as exc:
                    raise PipelineError("E_GIT_SCOPE", "task index path encoding differs") from exc
                if read_tree.returncode != 0 or not added_ok:
                    raise PipelineError("E_GIT_ADD", "task-index preparation failed")
                if staged.returncode != 0 or set(staged_paths) != set(expected_paths):
                    raise PipelineError(
                        "E_GIT_SCOPE",
                        f"task index escaped the exact allowlist ({len(staged_paths)}/{len(expected_paths)})",
                    )
                if not staged_paths:
                    _append(config.outbox_path, {"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id, "result": "NO_CHANGES", "completed_at": now.isoformat()})
                    return {"status": "GIT_NO_CHANGES", "outbox_id": outbox_id}
                parent = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
                tree = _run_env(["git", "write-tree"], config.project_root, runner, task_env)
                parent_sha = parent.stdout.strip()
                tree_sha = tree.stdout.strip()
                if parent.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", parent_sha) or tree.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", tree_sha):
                    raise PipelineError("E_GIT_COMMIT", "Git parent or tree identity is unavailable")
                git_intent = {
                    "schema_version": SCHEMA, "event": "GIT_COMMIT_INTENT", "outbox_id": outbox_id,
                    "parent_sha": parent_sha, "tree_sha": tree_sha, "files": expected_paths,
                    "message": (
                        f"chore(project-info): migrate Bilibili artifacts {outbox_id[:12]}"
                        if remove_paths else f"chore(project-info): archive Bilibili dynamic {outbox_id[:12]}"
                    ),
                    "author_name": "MB-X Bilibili Pipeline", "author_email": "mbx-bili-pipeline@localhost",
                    "authored_at": now.isoformat(), "created_at": now.isoformat(),
                }
                _append(config.outbox_path, git_intent)
            parent_sha = git_intent["parent_sha"]
            tree_sha = git_intent["tree_sha"]
            head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
            tree_exists = _run(["git", "cat-file", "-e", f"{tree_sha}^{{tree}}"], config.project_root, runner)
            if head.returncode != 0 or head.stdout.strip() != parent_sha or tree_exists.returncode != 0 or git_intent["files"] != expected_paths:
                raise PipelineError("E_GIT_COMMIT", "Git commit intent recovery identity differs")
            _git_artifacts(config, files)
            _git_remove_paths(config, payload.get("remove_paths"))
            commit_env = dict(os.environ)
            commit_env.update({
                "GIT_AUTHOR_NAME": git_intent["author_name"], "GIT_COMMITTER_NAME": git_intent["author_name"],
                "GIT_AUTHOR_EMAIL": git_intent["author_email"], "GIT_COMMITTER_EMAIL": git_intent["author_email"],
                "GIT_AUTHOR_DATE": git_intent["authored_at"], "GIT_COMMITTER_DATE": git_intent["authored_at"],
            })
            committed = _run_env(["git", "commit-tree", tree_sha, "-p", parent_sha, "-m", git_intent["message"]], config.project_root, runner, commit_env)
        finally:
            _remove_task_index(task_index, config.state_dir)
            _remove_task_index(task_lock, config.state_dir)
        commit_sha = committed.stdout.strip()
        if committed.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", commit_sha):
            raise PipelineError("E_GIT_COMMIT", "commit identity is unavailable")
        _append(config.outbox_path, {
            "schema_version": SCHEMA, "event": "COMMIT_CREATED", "outbox_id": outbox_id,
            "parent_sha": parent_sha, "tree_sha": tree_sha, "commit_sha": commit_sha,
            "intent_sha256": hashlib.sha256(_canonical(git_intent)).hexdigest().upper(),
            "files": expected_paths, "created_at": now.isoformat(),
        })
        _git_artifacts(config, files)
        _git_remove_paths(config, payload.get("remove_paths"))
        updated = _run(["git", "update-ref", "HEAD", commit_sha, parent_sha], config.project_root, runner)
        if updated.returncode != 0:
            raise PipelineError("E_GIT_COMMIT", "task-scoped ref update failed")
        pushed = _run(["git", "push", config.git_remote, f"{commit_sha}:refs/heads/{config.git_branch}"], config.project_root, runner)
        if pushed.returncode != 0:
            raise PipelineError("E_GIT_PUSH", "non-force push failed")
        event = {"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id, "result": "PUSHED", "commit_sha": commit_sha, "completed_at": now.isoformat()}
        _append(config.outbox_path, event)
        return {"status": "GIT_PUSHED", "outbox_id": outbox_id, "commit_sha": commit_sha, "files": allowed}


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Durable Bilibili half-hour pipeline coordinator")
    parser.add_argument("--config", type=Path, required=True)
    sub = parser.add_subparsers(dest="command", required=True)
    for name in ("init", "begin", "reconcile", "pending", "git-preflight", "plan-video-artifact-migration", "migrate-video-artifacts"):
        child = sub.add_parser(name)
        if name not in {"pending", "git-preflight"}:
            child.add_argument("--now")
    dispatch = sub.add_parser("dispatch-intent")
    dispatch.add_argument("--outbox-id", required=True)
    dispatch.add_argument("--now")
    observed = sub.add_parser("observe-dispatch")
    observed.add_argument("--outbox-id", required=True)
    observed.add_argument("--delivery-id", required=True)
    observed.add_argument("--now")
    receipt = sub.add_parser("ingest-receipt")
    receipt.add_argument("--receipt", type=Path, required=True)
    receipt.add_argument("--now")
    finish_parser = sub.add_parser("finish")
    finish_parser.add_argument("--status", choices=["COMPLETE", "FAILED"], required=True)
    finish_parser.add_argument("--now")
    git_parser = sub.add_parser("git-deliver")
    git_parser.add_argument("--outbox-id", required=True)
    git_parser.add_argument("--now")
    guard_parser = sub.add_parser("recover-git-index-guard")
    guard_parser.add_argument("--outbox-id", required=True)
    guard_parser.add_argument("--baseline-head", required=True)
    guard_parser.add_argument("--expected-bytes", required=True, type=int)
    guard_parser.add_argument("--expected-sha256", required=True)
    return parser


def run(argv: Sequence[str] | None = None) -> tuple[int, dict[str, Any]]:
    args = build_parser().parse_args(argv)
    try:
        config = load_config(Path(os.path.abspath(args.config)))
        if args.command == "init":
            result = initialize(config, _now(args.now))
        elif args.command == "begin":
            result = begin(config, _now(args.now))
        elif args.command == "reconcile":
            result = reconcile(config, _now(args.now))
        elif args.command == "pending":
            result = pending(config)
        elif args.command == "git-preflight":
            result = git_preflight(config)
        elif args.command == "plan-video-artifact-migration":
            result = plan_video_artifact_migration(config, _now(args.now))
        elif args.command == "migrate-video-artifacts":
            result = migrate_video_artifacts(config, _now(args.now))
        elif args.command == "dispatch-intent":
            result = dispatch_intent(config, args.outbox_id, _now(args.now))
        elif args.command == "observe-dispatch":
            result = observe_dispatch(config, args.outbox_id, args.delivery_id, _now(args.now))
        elif args.command == "ingest-receipt":
            result = ingest_receipt(config, Path(os.path.abspath(args.receipt)), _now(args.now))
        elif args.command == "finish":
            result = finish(config, args.status, _now(args.now))
        elif args.command == "recover-git-index-guard":
            result = recover_git_index_guard(
                config,
                args.outbox_id,
                args.baseline_head,
                args.expected_bytes,
                args.expected_sha256,
            )
        else:
            result = git_deliver(config, args.outbox_id, _now(args.now))
        return 0, result
    except PipelineError as exc:
        return 2, {"status": "FAILED", "error_code": exc.code}


def main(argv: Sequence[str] | None = None) -> int:
    code, result = run(argv)
    sys.stdout.write(json.dumps(result, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n")
    return code


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