from __future__ import annotations

import inspect
import re
from collections.abc import Awaitable, Callable, Iterator, Mapping
from itertools import chain
from pathlib import Path
from threading import Lock
from typing import Annotated, Any, Literal, Protocol
from urllib.parse import quote

from fastapi import APIRouter, Body, Header, HTTPException, Query, Request, Response
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from pydantic import BaseModel, ConfigDict, Field, StrictBool, field_validator, model_validator
from starlette.concurrency import run_in_threadpool
from starlette.types import Receive, Scope, Send

from k1link.compute import (
    RecordedPerceptionOverlayArtifact,
    RecordedPerceptionOverlayError,
    RecordedPerceptionVideo,
)
from k1link.sessions import (
    LayoutConflictError,
    MaterializedRecording,
    RecordedMediaFile,
    RecordedMediaInspector,
    RecordedMediaManifest,
    RecordingMaterializationError,
    RecordingPreparationQueueFull,
    RecordingPreparationSnapshot,
    ReplayCommand,
    SessionIntegrityError,
    SessionNotFoundError,
    SessionNotReplayableError,
    SessionRecordingPreparationManager,
    SessionStore,
    validate_recorded_media_timeline,
)
from k1link.sessions.canonical_lab_spatial import (
    CANONICAL_LAB_SPATIAL_PROFILE,
    canonical_lab_spatial_frame,
)
from k1link.sessions.plugin_contract import RecordedPointColorRenderer
from k1link.viewer.recorded import (
    APPLICATION_ID as RECORDED_APPLICATION_ID,
)
from k1link.viewer.recorded import (
    RecordedBlueprintError,
    recorded_blueprint_rrd,
)
from k1link.viewer.rerun_bridge import RerunSceneSettings

DEFAULT_REPLAY_ACTION_ID = "stream.start-replay"
RECORDED_MEDIA_STREAM_MANIFEST_SCHEMA = "missioncore.observation-recorded-media/v4"
RECORDED_PERCEPTION_STREAM_MANIFEST_SCHEMA = "missioncore.observation-recorded-media/v3"
SAFE_SOURCE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,255}$")
SAFE_SHA256 = re.compile(r"^[a-f0-9]{64}$")
MAX_SAFE_INTEGER = 9_007_199_254_740_991
REVALIDATED_RECORDING_CACHE_CONTROL = "private, no-cache, no-transform"
IMMUTABLE_RECORDING_CACHE_CONTROL = "private, max-age=31536000, immutable, no-transform"


class _ReleasingFileResponse(FileResponse):
    """Release a cache pin exactly once after every ASGI completion path."""

    def __init__(
        self,
        *args: Any,
        release: Callable[[], None],
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self._release = release
        self._release_guard = Lock()
        self._released = False

    def _release_once(self) -> None:
        with self._release_guard:
            if self._released:
                return
            self._released = True
        self._release()

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        try:
            await super().__call__(scope, receive, send)
        finally:
            # Starlette background tasks do not run when `send` is cancelled
            # or raises.  The finally block makes the response lifetime the
            # authoritative pin lifetime for success, disconnect and failure.
            self._release_once()


class StrictApiModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class ReplayRequest(StrictApiModel):
    speed: float = Field(default=1.0, ge=0.0, le=100.0)
    loop: bool = False


class RecordedBlueprintRequest(StrictApiModel):
    application_id: Literal["nodedc_mission_core_recorded"]
    recording_id: str = Field(
        min_length=1,
        max_length=128,
        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
    )
    blueprint_session_id: str = Field(
        min_length=32,
        max_length=32,
        pattern=r"^[a-f0-9]{32}$",
    )
    accumulation_seconds: float = Field(strict=True, ge=0.0, le=3600.0)
    show_points: StrictBool
    show_trajectory: StrictBool
    show_grid: StrictBool
    point_size: float = Field(default=2.5, strict=True, ge=0.1, le=32.0)
    color_mode: Literal["intensity", "height", "distance", "rgb", "class"] = "intensity"
    palette: Literal["turbo", "viridis", "plasma", "grayscale", "custom"] = "turbo"
    custom_color: str = Field(default="#f7f8f4", pattern=r"^#[0-9A-Fa-f]{6}$")
    active_view: Literal["spatial", "perception", "perception3d", "metrics"] = "spatial"
    view_reset_generation: Literal[0, 1] = 0
    unified_perception: StrictBool = False
    show_detections_2d: StrictBool = False
    show_segmentation: StrictBool = False
    show_cuboids_3d: StrictBool = False
    follow_trajectory: StrictBool = False


class RecordedPerceptionRequest(StrictApiModel):
    application_id: Literal["nodedc_mission_core_recorded"]
    recording_id: str = Field(
        min_length=1,
        max_length=128,
        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
    )


class RecordedPerceptionPreparationStatus(StrictApiModel):
    state: Literal["idle", "preparing", "ready", "unavailable", "error"]
    phase: Literal[
        "idle",
        "cache-lookup",
        "queued",
        "artifact-validation",
        "rendering",
        "cache-write",
        "ready",
        "unavailable",
        "error",
    ]
    elapsed_seconds: float = Field(ge=0.0)
    byte_length: int | None = Field(default=None, ge=0)


class RecordedPointColorsRequest(StrictApiModel):
    application_id: Literal["nodedc_mission_core_recorded"]
    recording_id: str = Field(
        min_length=1,
        max_length=128,
        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
    )
    color_mode: Literal["intensity", "height", "distance", "rgb", "class"]
    palette: Literal["turbo", "viridis", "plasma", "grayscale", "custom"]
    custom_color: str = Field(pattern=r"^#[0-9A-Fa-f]{6}$")


class SceneSettingsDocument(StrictApiModel):
    projection: Literal["3d", "2d", "map"]
    point_size: float = Field(ge=0.1, le=32.0)
    color_mode: Literal["intensity", "height", "distance", "rgb", "class"]
    palette: Literal["turbo", "viridis", "plasma", "grayscale", "custom"]
    custom_color: str = Field(pattern=r"^#[0-9A-Fa-f]{6}$")
    accumulation_seconds: float = Field(ge=0.0, le=3600.0)
    show_points: bool
    show_trajectory: bool
    show_grid: bool
    show_labels: bool
    show_camera_frustums: bool


class LegacyToolWindowsDocument(StrictApiModel):
    sources_open: bool
    display_open: bool
    layers_open: bool
    order: list[Literal["sources", "display", "layers"]]

    @field_validator("order")
    @classmethod
    def validate_order(cls, value: list[str]) -> list[str]:
        if len(value) != 3 or len(value) != len(set(value)):
            raise ValueError("tool window order must be a permutation of all three tools")
        return value


class NormalizedWindowRect(StrictApiModel):
    x: float = Field(ge=0.0, le=1.0)
    y: float = Field(ge=0.0, le=1.0)
    width: float = Field(gt=0.0, le=1.0)
    height: float = Field(gt=0.0, le=1.0)

    @model_validator(mode="after")
    def validate_bounds(self) -> NormalizedWindowRect:
        epsilon = 1e-9
        if self.x + self.width > 1.0 + epsilon or self.y + self.height > 1.0 + epsilon:
            raise ValueError("normalized window rectangle escapes the viewport")
        return self


class ViewportSize(StrictApiModel):
    width: float = Field(gt=0.0, le=100_000.0)
    height: float = Field(gt=0.0, le=100_000.0)


class LegacyLayoutV1Document(StrictApiModel):
    version: Literal[1]
    revision: int = Field(ge=0, le=MAX_SAFE_INTEGER)
    workspace_id: Literal["observation.spatial"]
    scene_settings: SceneSettingsDocument
    tool_windows: LegacyToolWindowsDocument
    visible_source_ids: list[str]
    active_floating_source_id: str | None
    window_rects: dict[str, NormalizedWindowRect]
    viewport_size: ViewportSize


class LayoutPutRequest(StrictApiModel):
    version: Literal[2]
    revision: int = Field(ge=0, le=MAX_SAFE_INTEGER)
    workspace_id: Literal["observation.spatial"]
    scene_settings: SceneSettingsDocument
    visible_source_ids: list[str]
    active_floating_source_id: str | None
    window_rects: dict[str, NormalizedWindowRect]
    viewport_size: ViewportSize

    @field_validator("visible_source_ids")
    @classmethod
    def validate_visible_source_ids(cls, value: list[str]) -> list[str]:
        if len(value) > 256 or len(value) != len(set(value)):
            raise ValueError("visible source identifiers must be unique and bounded")
        for source_id in value:
            _validate_source_id(source_id)
        return value

    @field_validator("active_floating_source_id")
    @classmethod
    def validate_active_source_id(cls, value: str | None) -> str | None:
        if value is not None:
            _validate_source_id(value)
        return value

    @field_validator("window_rects")
    @classmethod
    def validate_window_rect_ids(
        cls,
        value: dict[str, NormalizedWindowRect],
    ) -> dict[str, NormalizedWindowRect]:
        if len(value) > 256:
            raise ValueError("window rectangle catalog exceeds its boundary")
        for source_id in value:
            _validate_source_id(source_id)
        return value

    @model_validator(mode="after")
    def validate_active_source_visibility(self) -> LayoutPutRequest:
        if (
            self.active_floating_source_id is not None
            and self.active_floating_source_id not in self.visible_source_ids
        ):
            raise ValueError("active floating source must be visible")
        return self


class ReplayLauncher(Protocol):
    def __call__(
        self,
        command: ReplayCommand,
    ) -> Mapping[str, Any] | Awaitable[Mapping[str, Any]]: ...


class RecordingMaterializer(Protocol):
    def __call__(self, command: ReplayCommand) -> MaterializedRecording: ...


class CatalogRefresher(Protocol):
    """Synchronous app-owned recovery + discovery transaction hook.

    The host composes hardware-specific recovery before generic session
    discovery, then passes the resulting callable here. Return values stay
    internal and are intentionally ignored by the API.
    """

    def __call__(self) -> object: ...


class RecordedPerceptionOverlayProvider(Protocol):
    def render(
        self,
        session_id: str,
        *,
        application_id: str,
        recording_id: str,
    ) -> bytes | None: ...


class RecordedPerceptionMediaProvider(Protocol):
    def video(
        self,
        session_id: str,
        result_id: str | None = None,
    ) -> RecordedPerceptionVideo | None: ...


def build_session_router(
    store: SessionStore,
    *,
    catalog_refresher: CatalogRefresher | None = None,
    replay_launcher: ReplayLauncher | None = None,
    recording_materializer: RecordingMaterializer | None = None,
    recording_preparation_manager: SessionRecordingPreparationManager | None = None,
    media_inspector: RecordedMediaInspector | None = None,
    perception_overlay_provider: RecordedPerceptionOverlayProvider | None = None,
    perception_media_provider: RecordedPerceptionMediaProvider | None = None,
    point_color_renderers: Mapping[str, RecordedPointColorRenderer] | None = None,
    allow_synchronous_recording_fallback: bool = False,
    replay_action_id: str = DEFAULT_REPLAY_ACTION_ID,
) -> APIRouter:
    """Build generic host APIs without exposing filesystem locators to clients."""

    router = APIRouter(tags=["observation-sessions"])
    recorded_media_inspector = media_inspector or RecordedMediaInspector()

    @router.get("/api/v1/observation-sessions")
    def list_observation_sessions(
        limit: int = Query(default=20, ge=1, le=100),
        cursor: str | None = Query(default=None, max_length=128),
        scope: Literal["all", "source", "laboratory"] = "all",
    ) -> dict[str, Any]:
        try:
            _refresh_catalog(catalog_refresher)
            page = store.list_recent(limit=limit, cursor=cursor, scope=scope)
            return {
                "items": [
                    {
                        "id": item.session_id,
                        "label": item.display_name,
                        "started_at_utc": item.started_at_utc,
                        "completed_at_utc": item.completed_at_utc,
                        "status": item.status,
                        "modalities": list(item.modalities),
                        "duration_seconds": item.duration_seconds or 0.0,
                        "replayable": item.replayable,
                        **({"lab": item.lab.as_dict()} if item.lab is not None else {}),
                        **(
                            {
                                "preparation": _catalog_preparation_document(
                                    recording_preparation_manager,
                                    item.session_id,
                                    item.replayable,
                                )
                            }
                            if recording_preparation_manager is not None
                            else {}
                        ),
                    }
                    for item in page.items
                    if item.started_at_utc is not None
                ]
            }
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except ValueError as exc:
            raise HTTPException(status_code=422, detail="Некорректные параметры запроса.") from exc

    @router.get("/api/v1/observation-sessions/{session_id}")
    def get_observation_session(session_id: str) -> dict[str, Any]:
        try:
            _refresh_catalog(catalog_refresher)
            return store.get_session(session_id).as_dict()
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail="Некорректный идентификатор сессии.",
            ) from exc

    @router.delete(
        "/api/v1/observation-sessions/{session_id}",
        status_code=204,
    )
    async def delete_observation_session(session_id: str) -> Response:
        try:
            store.get_session(session_id)
            recorded_artifacts = store.list_recorded_media(session_id)
        except SessionNotFoundError:
            # DELETE is idempotent. Multiple mounted catalogs or another
            # operator tab may have already removed the exact session.
            return Response(status_code=204)
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail="Некорректный идентификатор сессии.",
            ) from exc

        if recording_preparation_manager is not None:
            snapshot = recording_preparation_manager.status(session_id)
            if snapshot is not None and snapshot.state in {
                "queued",
                "validating",
                "exporting",
                "finalizing",
            }:
                raise HTTPException(
                    status_code=409,
                    detail="Дождитесь завершения подготовки записи перед удалением.",
                )
            await run_in_threadpool(
                recording_preparation_manager.release_launch_reservation,
                session_id,
            )

        cache_deleter = getattr(recording_materializer, "delete_cached", None)
        if callable(cache_deleter):
            try:
                cache_deleted = await run_in_threadpool(cache_deleter, session_id)
            except (OSError, RecordingMaterializationError) as exc:
                raise HTTPException(
                    status_code=409,
                    detail="Подготовленную запись пока нельзя удалить.",
                ) from exc
            if cache_deleted is not True:
                raise HTTPException(
                    status_code=409,
                    detail="Запись сейчас открыта в интерфейсе. Закройте её и повторите удаление.",
                )

        if recording_preparation_manager is not None and not recording_preparation_manager.discard(
            session_id
        ):
            raise HTTPException(
                status_code=409,
                detail="Подготовка записи ещё выполняется.",
            )

        try:
            await run_in_threadpool(
                recorded_media_inspector.delete_prepared,
                session_id,
                (artifact.artifact_id for artifact in recorded_artifacts),
            )
            await run_in_threadpool(store.delete_session, session_id)
        except SessionNotFoundError:
            # A concurrent idempotent delete won the race after both callers
            # resolved the catalog row.
            return Response(status_code=204)
        except (OSError, SessionIntegrityError) as exc:
            raise HTTPException(
                status_code=409,
                detail="Сервер не смог безопасно удалить файлы этой сессии.",
            ) from exc
        return Response(status_code=204)

    @router.post("/api/v1/observation-sessions/{session_id}/replay")
    async def replay_observation_session(
        session_id: str,
        request: Annotated[ReplayRequest | None, Body()] = None,
    ) -> Any:
        replay_request = request or ReplayRequest()
        try:
            command = await run_in_threadpool(
                _prepare_replay,
                store,
                catalog_refresher,
                session_id,
                replay_request.speed,
                replay_request.loop,
                True,
            )
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except SessionNotReplayableError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except SessionIntegrityError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except ValueError as exc:
            raise HTTPException(status_code=422, detail="Некорректные параметры повтора.") from exc

        if recording_preparation_manager is not None:
            try:
                snapshot = recording_preparation_manager.status(command.session_id)
                if snapshot is None:
                    snapshot = await run_in_threadpool(
                        recording_preparation_manager.restore_published,
                        command,
                    )
                if snapshot is None or snapshot.state in {"failed", "cancelled"}:
                    # A terminal preparation describes one attempt, not the
                    # durable session. A fresh POST /replay is explicit retry
                    # intent and must replace that attempt. This also recovers
                    # the camera-finalization race where the RRD is already
                    # published but the media archive seals moments later.
                    snapshot = recording_preparation_manager.enqueue(
                        command,
                        retry_failed=True,
                    )
            except RecordingPreparationQueueFull as exc:
                raise HTTPException(
                    status_code=503,
                    detail="Очередь подготовки записей переполнена.",
                ) from exc
            if snapshot.state != "ready" or snapshot.recording is None:
                return _preparation_response(snapshot)
            reserved = recording_preparation_manager.reserve_ready(
                command.session_id,
                preparation_id=snapshot.preparation_id,
            )
            if reserved is None or reserved.recording is None:
                try:
                    replacement = recording_preparation_manager.enqueue(
                        command,
                        retry_failed=True,
                    )
                except RecordingPreparationQueueFull as exc:
                    raise HTTPException(
                        status_code=503,
                        detail="Очередь подготовки записей переполнена.",
                    ) from exc
                return _preparation_response(replacement)
            return _recording_launch_document(
                reserved.recording,
                command,
                reserved.recorded_media or (),
                perception_media_provider=perception_media_provider,
            )

        if recording_materializer is not None and allow_synchronous_recording_fallback:
            recording = await _materialize(recording_materializer, command)
            try:
                recorded_media = await run_in_threadpool(
                    _inspect_recorded_media,
                    store,
                    recorded_media_inspector,
                    command,
                    recording,
                )
            except SessionIntegrityError as exc:
                raise HTTPException(status_code=409, detail=str(exc)) from exc
            return _recording_launch_document(
                recording,
                command,
                recorded_media,
                perception_media_provider=perception_media_provider,
            )
        if recording_materializer is not None:
            raise HTTPException(
                status_code=503,
                detail="Сервис фоновой подготовки записей не настроен.",
            )

        runtime_state: dict[str, Any] | None = None
        dispatched = replay_launcher is not None
        if replay_launcher is not None:
            launched = replay_launcher(command)
            result = await launched if inspect.isawaitable(launched) else launched
            runtime_state = dict(result)
        return {
            "schema_version": "missioncore.observation-session-replay/v1",
            "launch": {
                "kind": "plugin-action",
                "plugin_id": command.plugin_id,
                "action_id": replay_action_id,
                "session_id": command.session_id,
                "speed": command.speed,
                "loop": command.loop,
                "dispatched": dispatched,
            },
            "runtime_state": runtime_state,
        }

    @router.get("/api/v1/observation-sessions/{session_id}/recording-preparation")
    async def get_recording_preparation(
        session_id: str,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
    ) -> Any:
        if recording_preparation_manager is None:
            raise HTTPException(
                status_code=409,
                detail="Сервис фоновой подготовки записей не настроен.",
            )
        snapshot = recording_preparation_manager.status(session_id)
        if snapshot is None:
            if if_match is None:
                raise HTTPException(
                    status_code=428,
                    detail="Укажите ETag подготовки в заголовке If-Match.",
                )
            # A polling URL is generation-bound. It must never silently create
            # a replacement and bind an old browser poller to the new job.
            raise HTTPException(
                status_code=412,
                detail="Подготовка была заменена более новой операцией.",
            )
        _require_matching_preparation(snapshot, if_match)
        if snapshot.state == "ready" and snapshot.recording is not None:
            reserved = recording_preparation_manager.reserve_ready(
                session_id,
                preparation_id=snapshot.preparation_id,
            )
            if reserved is None or reserved.recording is None:
                raise HTTPException(
                    status_code=412,
                    detail="Подготовка была заменена более новой операцией.",
                )
            snapshot = reserved
            return JSONResponse(
                content=_recording_launch_document(
                    reserved.recording,
                    snapshot.command,
                    snapshot.recorded_media or (),
                    perception_media_provider=perception_media_provider,
                ),
                headers={
                    "Cache-Control": "no-store",
                    "ETag": _preparation_etag(snapshot.preparation_id),
                    "X-Content-Type-Options": "nosniff",
                },
            )
        return _preparation_response(snapshot)

    @router.delete(
        "/api/v1/observation-sessions/{session_id}/recording-preparation",
        status_code=204,
    )
    def delete_recording_preparation(
        session_id: str,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
    ) -> Response:
        if recording_preparation_manager is None:
            raise HTTPException(
                status_code=409,
                detail="Сервис фоновой подготовки записей не настроен.",
            )
        snapshot = recording_preparation_manager.status(session_id)
        if snapshot is None:
            raise HTTPException(status_code=404, detail="Подготовка записи не найдена.")
        _require_matching_preparation(snapshot, if_match)
        recording_preparation_manager.cancel(
            session_id,
            preparation_id=snapshot.preparation_id,
        )
        return Response(status_code=204)

    @router.get("/api/v1/observation-sessions/{session_id}/recording.rrd")
    async def get_observation_session_recording(
        session_id: str,
        generation: Annotated[str | None, Query(max_length=64)] = None,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
        if_none_match: Annotated[str | None, Header()] = None,
    ) -> Response:
        if generation is not None and SAFE_SHA256.fullmatch(generation) is None:
            raise HTTPException(
                status_code=412,
                detail="Поколение подготовленной записи не совпадает.",
            )
        if recording_materializer is None:
            raise HTTPException(
                status_code=409,
                detail="Сервис seekable-записей не настроен.",
            )
        release_recording: Callable[[], None] | None
        if recording_preparation_manager is not None:
            if if_match is None and generation is None:
                raise HTTPException(
                    status_code=428,
                    detail="Укажите digest записи через generation или If-Match.",
                )
            snapshot = recording_preparation_manager.status(session_id)
            if snapshot is None:
                try:
                    command = await run_in_threadpool(
                        _prepare_replay,
                        store,
                        catalog_refresher,
                        session_id,
                        1.0,
                        False,
                        False,
                    )
                    snapshot = await run_in_threadpool(
                        recording_preparation_manager.restore_published,
                        command,
                    )
                    if snapshot is None:
                        snapshot = recording_preparation_manager.enqueue(command)
                except SessionNotFoundError as exc:
                    raise HTTPException(status_code=404, detail=str(exc)) from exc
                except (SessionNotReplayableError, SessionIntegrityError) as exc:
                    raise HTTPException(status_code=409, detail=str(exc)) from exc
                except RecordingPreparationQueueFull as exc:
                    raise HTTPException(
                        status_code=503,
                        detail="Очередь подготовки записей переполнена.",
                    ) from exc
                except ValueError as exc:
                    raise HTTPException(
                        status_code=422,
                        detail="Некорректный идентификатор сессии.",
                    ) from exc
            if snapshot.state in {
                "queued",
                "validating",
                "exporting",
                "finalizing",
            }:
                return _preparation_response(snapshot)
            if snapshot.state != "ready" or snapshot.recording is None:
                return _preparation_response(snapshot)
            recording = snapshot.recording
            etag = f'"sha256:{recording.sha256}"'
            if generation is not None:
                _require_matching_recording_generation(recording.sha256, generation)
            if if_match is not None:
                _require_matching_digest(etag, if_match)
            pinned = recording_preparation_manager.pin_ready(
                session_id,
                preparation_id=snapshot.preparation_id,
            )
            if pinned is None:
                raise HTTPException(
                    status_code=412,
                    detail="Подготовленная запись была заменена.",
                )
            snapshot, pinned_release = pinned
            release_recording = pinned_release
            pinned_recording = snapshot.recording
            if pinned_recording is None:
                release_recording()
                raise HTTPException(
                    status_code=500,
                    detail="Подготовленная запись сессии недоступна.",
                )
            recording = pinned_recording
        elif allow_synchronous_recording_fallback:
            try:
                command = await run_in_threadpool(
                    _prepare_replay,
                    store,
                    catalog_refresher,
                    session_id,
                    1.0,
                    False,
                    False,
                )
            except SessionNotFoundError as exc:
                raise HTTPException(status_code=404, detail=str(exc)) from exc
            except (SessionNotReplayableError, SessionIntegrityError) as exc:
                raise HTTPException(status_code=409, detail=str(exc)) from exc
            except ValueError as exc:
                raise HTTPException(
                    status_code=422,
                    detail="Некорректный идентификатор сессии.",
                ) from exc
            recording, release_recording = await _materialize_pinned(
                recording_materializer,
                command,
            )
            try:
                if generation is not None:
                    _require_matching_recording_generation(recording.sha256, generation)
                if if_match is not None:
                    _require_matching_digest(f'"sha256:{recording.sha256}"', if_match)
            except HTTPException:
                if release_recording is not None:
                    release_recording()
                raise
        else:
            raise HTTPException(
                status_code=503,
                detail="Сервис фоновой подготовки записей не настроен.",
            )
        etag = f'"sha256:{recording.sha256}"'
        cache_control = (
            IMMUTABLE_RECORDING_CACHE_CONTROL
            if generation is not None
            else REVALIDATED_RECORDING_CACHE_CONTROL
        )
        if _if_none_match_matches(if_none_match, etag):
            if release_recording is not None:
                release_recording()
            return Response(
                status_code=304,
                headers={
                    "Cache-Control": cache_control,
                    "ETag": etag,
                    "X-Content-Type-Options": "nosniff",
                },
            )
        try:
            stat_result = recording.path.stat()
        except OSError as exc:
            if release_recording is not None:
                release_recording()
            raise HTTPException(
                status_code=500,
                detail="Подготовленная запись сессии недоступна.",
            ) from exc
        response_type = _ReleasingFileResponse if release_recording is not None else FileResponse
        response_kwargs: dict[str, Any] = {}
        if release_recording is not None:
            response_kwargs["release"] = release_recording
        return response_type(
            recording.path,
            media_type=recording.media_type,
            filename=f"{recording.session_id}.rrd",
            content_disposition_type="inline",
            stat_result=stat_result,
            headers={
                "Cache-Control": cache_control,
                "ETag": etag,
                "Content-Length": str(recording.byte_length),
                "X-Content-Type-Options": "nosniff",
            },
            **response_kwargs,
        )

    @router.get(
        "/api/v1/observation-sessions/{session_id}/canonical-lab/spatial-frame"
    )
    async def get_observation_session_canonical_lab_spatial_frame(
        session_id: str,
        generation: Annotated[str, Query(min_length=64, max_length=64)],
        time_ns: Annotated[int, Query(ge=0, le=MAX_SAFE_INTEGER)],
        profile: Literal["source-paced-ground-v3"],
    ) -> JSONResponse:
        """Serve one body-frame sample for the canonical recorded-LAB clock.

        The camera media clock owns playback. Spatial evidence is sampled from
        the same immutable recording instead of starting a second Rerun clock.
        """

        if SAFE_SHA256.fullmatch(generation) is None:
            raise HTTPException(
                status_code=412,
                detail="Поколение spatial-записи не совпадает.",
            )
        if recording_preparation_manager is None:
            raise HTTPException(
                status_code=503,
                detail="Сервис canonical LAB spatial playback не настроен.",
            )
        snapshot = recording_preparation_manager.status(session_id)
        if snapshot is None or snapshot.state != "ready" or snapshot.recording is None:
            raise HTTPException(
                status_code=409,
                detail="Запись canonical LAB ещё не подготовлена.",
            )
        _require_matching_recording_generation(snapshot.recording.sha256, generation)
        pinned = recording_preparation_manager.pin_ready(
            session_id,
            preparation_id=snapshot.preparation_id,
        )
        if pinned is None:
            raise HTTPException(
                status_code=412,
                detail="Подготовленная spatial-запись была заменена.",
            )
        pinned_snapshot, release_recording = pinned
        try:
            recording = pinned_snapshot.recording
            if recording is None:
                raise HTTPException(
                    status_code=500,
                    detail="Подготовленная spatial-запись недоступна.",
                )
            payload = await run_in_threadpool(
                canonical_lab_spatial_frame,
                recording.path,
                generation,
                time_ns,
            )
        except (OSError, ValueError) as exc:
            raise HTTPException(
                status_code=503,
                detail="Canonical LAB spatial frame не прошёл проверку.",
            ) from exc
        finally:
            release_recording()
        return JSONResponse(
            payload,
            headers={
                "Cache-Control": "private, max-age=31536000, immutable",
                "ETag": (
                    f'"{generation}:{CANONICAL_LAB_SPATIAL_PROFILE}:'
                    f'{payload["source_time_ns"]}"'
                ),
                "X-Content-Type-Options": "nosniff",
            },
        )

    @router.post("/api/v1/observation-sessions/{session_id}/blueprint.rrd")
    async def get_observation_session_blueprint(
        session_id: str,
        request: RecordedBlueprintRequest,
    ) -> Response:
        try:
            await run_in_threadpool(
                _prepare_replay,
                store,
                catalog_refresher,
                session_id,
                1.0,
                False,
                False,
            )
            payload = await run_in_threadpool(
                recorded_blueprint_rrd,
                RerunSceneSettings(
                    point_size=request.point_size,
                    color_mode=request.color_mode,
                    palette=request.palette,
                    custom_color=request.custom_color,
                    accumulation_seconds=request.accumulation_seconds,
                    show_points=request.show_points,
                    show_trajectory=request.show_trajectory,
                    show_grid=request.show_grid,
                ),
                application_id=RECORDED_APPLICATION_ID,
                recording_id=request.recording_id,
                blueprint_session_id=request.blueprint_session_id,
                active_view=request.active_view,
                view_reset_generation=request.view_reset_generation,
                unified_perception=request.unified_perception,
                show_detections_2d=request.show_detections_2d,
                show_segmentation=request.show_segmentation,
                show_cuboids_3d=request.show_cuboids_3d,
                follow_trajectory=request.follow_trajectory,
            )
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except (SessionNotReplayableError, SessionIntegrityError) as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except RecordedBlueprintError as exc:
            raise HTTPException(
                status_code=500,
                detail="Не удалось подготовить настройки визуализатора.",
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail="Некорректный идентификатор сессии.",
            ) from exc
        return Response(
            content=payload,
            media_type="application/vnd.rerun.rrd",
            headers={
                "Cache-Control": "no-store",
                "X-Content-Type-Options": "nosniff",
                "Content-Disposition": 'inline; filename="blueprint.rrd"',
            },
        )

    @router.post("/api/v1/observation-sessions/{session_id}/perception.rrd")
    async def get_observation_session_perception(
        session_id: str,
        request: RecordedPerceptionRequest,
    ) -> Response:
        if perception_overlay_provider is None:
            return Response(status_code=204, headers={"Cache-Control": "no-store"})
        try:
            await run_in_threadpool(
                _prepare_replay,
                store,
                catalog_refresher,
                session_id,
                1.0,
                False,
                False,
            )
            materializer = getattr(perception_overlay_provider, "materialize", None)
            payload = await run_in_threadpool(
                materializer if callable(materializer) else perception_overlay_provider.render,
                session_id,
                application_id=request.application_id,
                recording_id=request.recording_id,
            )
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except (SessionNotReplayableError, SessionIntegrityError) as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except RecordedPerceptionOverlayError as exc:
            raise HTTPException(
                status_code=500,
                detail="Не удалось подготовить слой распознавания.",
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail="Некорректный идентификатор сессии.",
            ) from exc
        if payload is None:
            return Response(status_code=204, headers={"Cache-Control": "no-store"})
        if isinstance(payload, RecordedPerceptionOverlayArtifact):
            try:
                metadata = payload.path.stat()
            except OSError as exc:
                raise HTTPException(
                    status_code=500,
                    detail="Не удалось подготовить слой распознавания.",
                ) from exc
            if not metadata.st_size == payload.byte_length:
                raise HTTPException(
                    status_code=500,
                    detail="Не удалось подготовить слой распознавания.",
                )
            return FileResponse(
                path=payload.path,
                media_type="application/vnd.rerun.rrd",
                filename="perception.rrd",
                content_disposition_type="inline",
                stat_result=metadata,
                headers={
                    "Cache-Control": "private, no-cache, no-transform",
                    "Content-Length": str(payload.byte_length),
                    "ETag": f'"sha256:{payload.sha256}"',
                    "X-Content-Type-Options": "nosniff",
                },
            )
        return Response(
            content=payload,
            media_type="application/vnd.rerun.rrd",
            headers={
                "Cache-Control": "private, no-cache, no-transform",
                "Content-Length": str(len(payload)),
                "X-Content-Type-Options": "nosniff",
                "Content-Disposition": 'inline; filename="perception.rrd"',
            },
        )

    @router.get("/api/v1/observation-sessions/{session_id}/perception/status")
    async def get_observation_session_perception_status(
        session_id: str,
        response: Response,
        recording_id: Annotated[
            str,
            Query(
                min_length=1,
                max_length=128,
                pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
            ),
        ],
    ) -> dict[str, Any]:
        try:
            await run_in_threadpool(
                _prepare_replay,
                store,
                catalog_refresher,
                session_id,
                1.0,
                False,
                False,
            )
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except (SessionNotReplayableError, SessionIntegrityError) as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        status_provider = (
            getattr(perception_overlay_provider, "status", None)
            if perception_overlay_provider is not None
            else None
        )
        value: object = (
            await run_in_threadpool(
                status_provider,
                session_id,
                recording_id=recording_id,
            )
            if callable(status_provider)
            else {
                "state": "idle",
                "phase": "idle",
                "elapsed_seconds": 0.0,
                "byte_length": None,
            }
        )
        response.headers["Cache-Control"] = "no-store"
        try:
            return RecordedPerceptionPreparationStatus.model_validate(value).model_dump()
        except ValueError as exc:
            raise HTTPException(
                status_code=500,
                detail="Сервис слоя распознавания вернул некорректный статус.",
            ) from exc

    @router.post("/api/v1/observation-sessions/{session_id}/point-colors.rrd")
    async def get_observation_session_point_colors(
        session_id: str,
        request: RecordedPointColorsRequest,
    ) -> Response:
        try:
            command = await run_in_threadpool(
                _prepare_replay,
                store,
                catalog_refresher,
                session_id,
                1.0,
                False,
                False,
            )
            point_color_renderer = (point_color_renderers or {}).get(command.plugin_id)
            if point_color_renderer is None:
                raise HTTPException(
                    status_code=409,
                    detail="Сервис архивной окраски точек не настроен.",
                )
            payload = await run_in_threadpool(
                point_color_renderer,
                command,
                application_id=request.application_id,
                recording_id=request.recording_id,
                color_mode=request.color_mode,
                palette=request.palette,
                custom_color=request.custom_color,
            )
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except (SessionNotReplayableError, SessionIntegrityError) as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail="Некорректные параметры окраски точек.",
            ) from exc
        except RuntimeError as exc:
            raise HTTPException(
                status_code=500,
                detail="Не удалось подготовить окраску облака точек.",
            ) from exc
        return Response(
            content=payload,
            media_type="application/vnd.rerun.rrd",
            headers={
                "Cache-Control": "private, no-cache, no-transform",
                "Content-Length": str(len(payload)),
                "X-Content-Type-Options": "nosniff",
                "Content-Disposition": 'inline; filename="point-colors.rrd"',
            },
        )

    @router.get("/api/v1/observation-sessions/{session_id}/perception-media/{result_id}/manifest")
    def get_recorded_perception_media_manifest(
        session_id: str,
        result_id: str,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
    ) -> JSONResponse:
        video = _resolve_recorded_perception_video(
            store,
            catalog_refresher,
            perception_media_provider,
            session_id,
            result_id,
        )
        etag = f'"sha256:{video.sha256}"'
        _require_matching_digest(etag, if_match)
        return JSONResponse(
            content=_recorded_perception_manifest_document(video),
            headers={
                "Cache-Control": "private, no-cache",
                "ETag": etag,
                "X-Content-Type-Options": "nosniff",
            },
        )

    @router.api_route(
        "/api/v1/observation-sessions/{session_id}/perception-media/{result_id}/recording.mp4",
        methods=["GET", "HEAD"],
    )
    def stream_recorded_perception_media(
        request: Request,
        session_id: str,
        result_id: str,
        generation: Annotated[str, Query(min_length=64, max_length=64)],
        range_header: Annotated[str | None, Header(alias="Range")] = None,
    ) -> Response:
        video = _resolve_recorded_perception_video(
            store,
            catalog_refresher,
            perception_media_provider,
            session_id,
            result_id,
        )
        if SAFE_SHA256.fullmatch(generation) is None or generation != video.sha256:
            raise HTTPException(
                status_code=412,
                detail="Поколение видео сегментации было заменено.",
            )
        return _recorded_perception_video_response(
            video,
            range_header,
            head_only=request.method == "HEAD",
        )

    @router.get("/api/v1/observation-sessions/{session_id}/media/{artifact_id}/manifest")
    def get_recorded_media_manifest(
        session_id: str,
        artifact_id: str,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
    ) -> JSONResponse:
        manifest = _resolve_recorded_media_manifest(
            store,
            recorded_media_inspector,
            recording_preparation_manager,
            allow_synchronous_recording_fallback,
            catalog_refresher,
            session_id,
            artifact_id,
        )
        manifest_etag = f'"sha256:{manifest.generation_sha256}"'
        _require_matching_digest(manifest_etag, if_match)
        return JSONResponse(
            content=_recorded_media_manifest_document(manifest),
            headers={
                "Cache-Control": "private, no-cache",
                "ETag": manifest_etag,
                "X-Content-Type-Options": "nosniff",
            },
        )

    @router.get(
        "/api/v1/observation-sessions/{session_id}/media/{artifact_id}/"
        "epochs/{epoch_ordinal}/init.mp4"
    )
    def get_recorded_media_init(
        session_id: str,
        artifact_id: str,
        epoch_ordinal: int,
        generation: Annotated[str | None, Query(min_length=64, max_length=64)] = None,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
        range_header: Annotated[str | None, Header(alias="Range")] = None,
    ) -> Response:
        manifest = _resolve_recorded_media_manifest(
            store,
            recorded_media_inspector,
            recording_preparation_manager,
            allow_synchronous_recording_fallback,
            catalog_refresher,
            session_id,
            artifact_id,
        )
        expected_sha256 = _recorded_media_init_sha256(manifest, epoch_ordinal)
        if generation is None:
            _require_matching_digest(f'"sha256:{expected_sha256}"', if_match)
        elif SAFE_SHA256.fullmatch(generation) is None or generation != manifest.generation_sha256:
            raise HTTPException(
                status_code=412,
                detail="Поколение записанного видео было заменено.",
            )
        try:
            media_file = recorded_media_inspector.get_init(manifest, epoch_ordinal)
        except SessionIntegrityError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        return _recorded_media_file_response(media_file, range_header)

    @router.get(
        "/api/v1/observation-sessions/{session_id}/media/{artifact_id}/"
        "epochs/{epoch_ordinal}/segments/{segment_sequence}.m4s"
    )
    def get_recorded_media_segment(
        session_id: str,
        artifact_id: str,
        epoch_ordinal: int,
        segment_sequence: int,
        generation: Annotated[str | None, Query(min_length=64, max_length=64)] = None,
        if_match: Annotated[str | None, Header(alias="If-Match")] = None,
        range_header: Annotated[str | None, Header(alias="Range")] = None,
    ) -> Response:
        manifest = _resolve_recorded_media_manifest(
            store,
            recorded_media_inspector,
            recording_preparation_manager,
            allow_synchronous_recording_fallback,
            catalog_refresher,
            session_id,
            artifact_id,
        )
        expected_sha256 = _recorded_media_segment_sha256(
            manifest,
            epoch_ordinal,
            segment_sequence,
        )
        if generation is None:
            _require_matching_digest(f'"sha256:{expected_sha256}"', if_match)
        elif SAFE_SHA256.fullmatch(generation) is None or generation != manifest.generation_sha256:
            raise HTTPException(
                status_code=412,
                detail="Поколение записанного видео было заменено.",
            )
        try:
            media_file = recorded_media_inspector.get_segment(
                manifest,
                epoch_ordinal,
                segment_sequence,
            )
        except SessionIntegrityError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        return _recorded_media_file_response(media_file, range_header)

    @router.api_route(
        "/api/v1/observation-sessions/{session_id}/media/{artifact_id}/"
        "epochs/{epoch_ordinal}/recording.mp4",
        methods=["GET", "HEAD"],
    )
    def stream_recorded_media_epoch(
        request: Request,
        session_id: str,
        artifact_id: str,
        epoch_ordinal: int,
        generation: Annotated[str, Query(min_length=64, max_length=64)],
        range_header: Annotated[str | None, Header(alias="Range")] = None,
    ) -> Response:
        manifest = _resolve_recorded_media_manifest(
            store,
            recorded_media_inspector,
            recording_preparation_manager,
            allow_synchronous_recording_fallback,
            catalog_refresher,
            session_id,
            artifact_id,
        )
        if SAFE_SHA256.fullmatch(generation) is None or generation != manifest.generation_sha256:
            raise HTTPException(
                status_code=412,
                detail="Поколение записанного видео было заменено.",
            )
        return _recorded_media_epoch_response(
            recorded_media_inspector,
            manifest,
            epoch_ordinal,
            range_header,
            head_only=request.method == "HEAD",
        )

    @router.get("/api/v1/workspace-layouts/{workspace_id}")
    def get_workspace_layout(workspace_id: str, response: Response) -> dict[str, Any]:
        try:
            stored = store.get_layout(workspace_id)
            document = _layout_document(
                stored.workspace_id,
                stored.schema_version,
                stored.revision,
                stored.layout,
            )
            response.headers["ETag"] = f'"{stored.revision}"'
            return document
        except SessionNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        except ValueError as exc:
            raise HTTPException(status_code=422, detail="Некорректный workspace id.") from exc

    @router.put("/api/v1/workspace-layouts/{workspace_id}")
    def put_workspace_layout(
        workspace_id: str,
        request: LayoutPutRequest,
        response: Response,
        if_match: Annotated[str, Header(alias="If-Match")],
    ) -> dict[str, Any]:
        header_revision = _parse_if_match(if_match)
        if workspace_id != request.workspace_id:
            raise HTTPException(status_code=422, detail="Workspace id не совпадает с документом.")
        if header_revision != request.revision:
            raise HTTPException(status_code=412, detail="Ревизия документа устарела.")
        payload = request.model_dump(
            mode="json",
            exclude={"version", "revision", "workspace_id"},
        )
        try:
            stored = store.save_layout(
                workspace_id,
                schema_version=request.version,
                expected_revision=request.revision,
                name="Пространственная сцена",
                layout=payload,
            )
            response.headers["ETag"] = f'"{stored.revision}"'
            return _layout_document(
                stored.workspace_id,
                stored.schema_version,
                stored.revision,
                stored.layout,
            )
        except LayoutConflictError as exc:
            raise HTTPException(status_code=412, detail=str(exc)) from exc
        except ValueError as exc:
            raise HTTPException(status_code=422, detail="Некорректная компоновка.") from exc

    return router


def _refresh_catalog(refresher: CatalogRefresher | None) -> None:
    if refresher is None:
        return
    try:
        result = refresher()
        if inspect.isawaitable(result):
            if inspect.iscoroutine(result):
                result.close()
            raise TypeError("catalog refresher must be synchronous")
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail="Не удалось обновить каталог сохранённых сессий.",
        ) from exc


def _prepare_replay(
    store: SessionStore,
    catalog_refresher: CatalogRefresher | None,
    session_id: str,
    speed: float,
    loop: bool,
    refresh_first: bool,
) -> ReplayCommand:
    if refresh_first:
        _refresh_catalog(catalog_refresher)
    try:
        return store.prepare_replay(session_id, speed=speed, loop=loop)
    except SessionNotFoundError:
        if refresh_first or catalog_refresher is None:
            raise
        _refresh_catalog(catalog_refresher)
        return store.prepare_replay(session_id, speed=speed, loop=loop)


def _validate_source_id(source_id: str) -> None:
    if SAFE_SOURCE_ID.fullmatch(source_id) is None:
        raise ValueError("source identifier has an invalid shape")


def _parse_if_match(value: str) -> int:
    normalized = value.strip()
    if len(normalized) < 3 or normalized[0] != '"' or normalized[-1] != '"':
        raise HTTPException(status_code=422, detail="If-Match должен содержать quoted revision.")
    revision = normalized[1:-1]
    if not revision.isascii() or not revision.isdecimal():
        raise HTTPException(status_code=422, detail="If-Match содержит некорректную ревизию.")
    return int(revision)


def _layout_document(
    workspace_id: str,
    schema_version: int,
    revision: int,
    layout: dict[str, Any],
) -> dict[str, Any]:
    if schema_version == 1:
        try:
            legacy = LegacyLayoutV1Document.model_validate(
                {
                    "version": 1,
                    "revision": revision,
                    "workspace_id": workspace_id,
                    **layout,
                }
            )
        except ValueError as exc:
            raise SessionIntegrityError(
                "stored workspace layout violates schema version 1"
            ) from exc
        layout = legacy.model_dump(
            mode="json",
            exclude={"version", "revision", "workspace_id", "tool_windows"},
        )
        schema_version = 2
    if schema_version != 2:
        raise SessionIntegrityError("stored workspace layout schema is unsupported")
    try:
        document = LayoutPutRequest.model_validate(
            {
                "version": 2,
                "revision": revision,
                "workspace_id": workspace_id,
                **layout,
            }
        )
    except ValueError as exc:
        raise SessionIntegrityError("stored workspace layout violates schema version 2") from exc
    return document.model_dump(mode="json")


ReplayLauncherCallable = Callable[
    [ReplayCommand],
    Mapping[str, Any] | Awaitable[Mapping[str, Any]],
]


async def _materialize(
    materializer: RecordingMaterializer,
    command: ReplayCommand,
) -> MaterializedRecording:
    try:
        recording = await run_in_threadpool(materializer, command)
    except RecordingMaterializationError as exc:
        raise HTTPException(
            status_code=500,
            detail="Не удалось подготовить запись сессии.",
        ) from exc
    if not isinstance(recording, MaterializedRecording):
        raise HTTPException(
            status_code=500,
            detail="Сервис записи вернул некорректный результат.",
        )
    return recording


async def _materialize_pinned(
    materializer: RecordingMaterializer,
    command: ReplayCommand,
) -> tuple[MaterializedRecording, Callable[[], None] | None]:
    pin = getattr(materializer, "materialize_pinned", None)
    if not callable(pin):
        return await _materialize(materializer, command), None
    try:
        result = await run_in_threadpool(pin, command)
    except RecordingMaterializationError as exc:
        raise HTTPException(
            status_code=500,
            detail="Не удалось подготовить запись сессии.",
        ) from exc
    if (
        not isinstance(result, tuple)
        or len(result) != 2
        or not isinstance(result[0], MaterializedRecording)
        or not callable(result[1])
    ):
        raise HTTPException(
            status_code=500,
            detail="Сервис записи вернул некорректный pinned-результат.",
        )
    return result[0], result[1]


def _preparation_response(snapshot: RecordingPreparationSnapshot) -> JSONResponse:
    encoded_session_id = quote(snapshot.session_id, safe="")
    status_url = f"/api/v1/observation-sessions/{encoded_session_id}/recording-preparation"
    preparation: dict[str, Any] = {
        "preparation_id": snapshot.preparation_id,
        "session_id": snapshot.session_id,
        "state": snapshot.state,
        "progress": snapshot.progress,
        "updated_at_utc": snapshot.updated_at_utc,
        "status_url": status_url,
        "cancellable": snapshot.cancellable,
    }
    terminal_failure = snapshot.state in {"failed", "cancelled"}
    if terminal_failure:
        preparation["retryable"] = snapshot.retryable
        preparation["error"] = snapshot.error or "Подготовка записи отменена."
    headers = {
        "Location": status_url,
        "Cache-Control": "no-store",
        "ETag": _preparation_etag(snapshot.preparation_id),
        "X-Content-Type-Options": "nosniff",
    }
    if not terminal_failure:
        headers["Retry-After"] = "1"
    return JSONResponse(
        status_code=409 if terminal_failure else 202,
        content={
            "schema_version": "missioncore.observation-session-preparation/v1",
            "preparation": preparation,
        },
        headers=headers,
    )


def _preparation_etag(preparation_id: str) -> str:
    return f'"{preparation_id}"'


def _require_matching_preparation(
    snapshot: RecordingPreparationSnapshot,
    if_match: str | None,
) -> None:
    if if_match is None:
        raise HTTPException(
            status_code=428,
            detail="Укажите ETag подготовки в заголовке If-Match.",
        )
    if if_match.strip() != _preparation_etag(snapshot.preparation_id):
        raise HTTPException(
            status_code=412,
            detail="Подготовка была заменена более новой операцией.",
        )


def _require_matching_digest(expected_etag: str, if_match: str | None) -> None:
    if if_match is None:
        raise HTTPException(
            status_code=428,
            detail="Укажите digest поколения в заголовке If-Match.",
        )
    if if_match.strip() != expected_etag:
        raise HTTPException(
            status_code=412,
            detail="Запрошенное поколение данных устарело.",
        )


def _require_matching_recording_generation(
    expected_sha256: str,
    generation: str,
) -> None:
    if SAFE_SHA256.fullmatch(generation) is None or generation != expected_sha256:
        raise HTTPException(
            status_code=412,
            detail="Поколение подготовленной записи не совпадает.",
        )


def _catalog_preparation_document(
    manager: SessionRecordingPreparationManager | None,
    session_id: str,
    replayable: bool,
) -> dict[str, Any] | None:
    if manager is None or not replayable:
        return None
    snapshot = manager.status(session_id)
    if snapshot is None:
        return None
    document: dict[str, Any] = {
        "preparation_id": snapshot.preparation_id,
        "state": snapshot.state,
        "progress": snapshot.progress,
        "updated_at_utc": snapshot.updated_at_utc,
        "cancellable": snapshot.cancellable,
        "retryable": snapshot.retryable,
    }
    if snapshot.error is not None:
        document["error"] = snapshot.error
    return document


def _recording_launch_document(
    recording: MaterializedRecording,
    command: ReplayCommand,
    recorded_media: tuple[RecordedMediaManifest, ...] = (),
    *,
    perception_media_provider: RecordedPerceptionMediaProvider | None = None,
) -> dict[str, Any]:
    del perception_media_provider
    encoded_session_id = quote(recording.session_id, safe="")
    source_url = f"/api/v1/observation-sessions/{encoded_session_id}/recording.rrd"
    media_sources = [
        _recorded_media_launch_source(manifest, index=index)
        for index, manifest in enumerate(recorded_media, start=1)
    ]
    return {
        "schema_version": "missioncore.observation-session-replay/v2",
        "launch": {
            "kind": "rerun-recording",
            "session_id": recording.session_id,
            "source_url": source_url,
            "viewer_source_url": f"{source_url}?generation={recording.sha256}",
            "media_type": recording.media_type,
            "timeline": recording.timeline,
            "timeline_start_seconds": recording.timeline_start_ns / 1_000_000_000,
            "timeline_end_seconds": recording.timeline_end_ns / 1_000_000_000,
            "seekable": True,
            "byte_length": recording.byte_length,
            "sha256": recording.sha256,
            "playback": {
                "speed": command.speed,
                "loop": command.loop,
            },
            "media_sources": media_sources,
        },
    }


def _inspect_recorded_media(
    store: SessionStore,
    inspector: RecordedMediaInspector,
    command: ReplayCommand,
    recording: MaterializedRecording,
) -> tuple[RecordedMediaManifest, ...]:
    manifests = tuple(
        inspector.inspect(artifact, command)
        for artifact in store.list_recorded_media(command.session_id)
    )
    validate_recorded_media_timeline(
        manifests,
        recording_start_seconds=recording.timeline_start_ns / 1_000_000_000,
        recording_end_seconds=recording.timeline_end_ns / 1_000_000_000,
    )
    return manifests


def _recorded_media_launch_source(
    manifest: RecordedMediaManifest,
    *,
    index: int,
) -> dict[str, Any]:
    encoded_session_id = quote(manifest.session_id, safe="")
    encoded_artifact_id = quote(manifest.artifact_id, safe="")
    return {
        "id": manifest.public_source_id,
        "label": f"Записанная камера {index}",
        "modality": "video",
        "manifest_url": (
            f"/api/v1/observation-sessions/{encoded_session_id}/media/"
            f"{encoded_artifact_id}/manifest"
        ),
        "media_type": "video/mp4",
        "manifest_generation_sha256": manifest.generation_sha256,
        "byte_length": manifest.byte_length,
        "timeline_start_seconds": manifest.timeline_start_seconds,
        "timeline_end_seconds": manifest.timeline_end_seconds,
        "seekable": True,
        "synchronization": manifest.synchronization,
    }


def _recorded_perception_launch_source(video: RecordedPerceptionVideo) -> dict[str, Any]:
    encoded_session_id = quote(video.session_id, safe="")
    encoded_result_id = quote(video.result_id, safe="")
    return {
        "id": video.public_source_id,
        "label": video.label,
        "modality": "video",
        "manifest_url": (
            f"/api/v1/observation-sessions/{encoded_session_id}/perception-media/"
            f"{encoded_result_id}/manifest"
        ),
        "media_type": "video/mp4",
        "manifest_generation_sha256": video.sha256,
        "byte_length": video.byte_length,
        "timeline_start_seconds": video.timeline_start_seconds,
        "timeline_end_seconds": video.timeline_end_seconds,
        "seekable": True,
        "synchronization": "host-arrival-best-effort",
    }


def _resolve_recorded_perception_video(
    store: SessionStore,
    catalog_refresher: CatalogRefresher | None,
    provider: RecordedPerceptionMediaProvider | None,
    session_id: str,
    result_id: str,
) -> RecordedPerceptionVideo:
    if provider is None:
        raise HTTPException(status_code=404, detail="Видео сегментации не найдено.")
    try:
        _prepare_replay(
            store,
            catalog_refresher,
            session_id,
            1.0,
            False,
            False,
        )
        video = provider.video(session_id, result_id)
    except SessionNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except (SessionNotReplayableError, SessionIntegrityError) as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(
            status_code=422,
            detail="Некорректный идентификатор результата сегментации.",
        ) from exc
    if video is None:
        raise HTTPException(status_code=404, detail="Видео сегментации не найдено.")
    return video


def _recorded_perception_manifest_document(
    video: RecordedPerceptionVideo,
) -> dict[str, Any]:
    encoded_session_id = quote(video.session_id, safe="")
    encoded_result_id = quote(video.result_id, safe="")
    base = f"/api/v1/observation-sessions/{encoded_session_id}/perception-media/{encoded_result_id}"
    return {
        "schema_version": RECORDED_PERCEPTION_STREAM_MANIFEST_SCHEMA,
        "source_id": video.public_source_id,
        "generation_sha256": video.sha256,
        "byte_length": video.byte_length,
        "timeline_start_seconds": video.timeline_start_seconds,
        "timeline_end_seconds": video.timeline_end_seconds,
        "synchronization": "host-arrival-best-effort",
        "epochs": [
            {
                "ordinal": 1,
                "timeline_start_seconds": video.timeline_start_seconds,
                "timeline_end_seconds": video.timeline_end_seconds,
                "media_type": video.media_type,
                "byte_length": video.byte_length,
                "stream_url": f"{base}/recording.mp4?generation={video.sha256}",
            }
        ],
    }


def _recorded_perception_video_response(
    video: RecordedPerceptionVideo,
    range_header: str | None,
    *,
    head_only: bool,
) -> Response:
    start, end = 0, video.byte_length - 1
    status_code = 200
    if range_header is not None:
        start, end = _parse_byte_range(range_header, video.byte_length)
        status_code = 206
    headers = {
        "Accept-Ranges": "bytes",
        "Cache-Control": "private, max-age=31536000, immutable, no-transform",
        "ETag": f'"sha256:{video.sha256}"',
        "Content-Length": str(end - start + 1),
        "X-Content-Type-Options": "nosniff",
        "Content-Disposition": 'inline; filename="recorded-perception.mp4"',
    }
    if status_code == 206:
        headers["Content-Range"] = f"bytes {start}-{end}/{video.byte_length}"
    if head_only:
        return Response(
            status_code=status_code,
            media_type=video.media_type,
            headers=headers,
        )
    return StreamingResponse(
        _iter_regular_file_range(video.path, start, end),
        status_code=status_code,
        media_type=video.media_type,
        headers=headers,
    )


def _iter_regular_file_range(path: Path, start: int, end: int) -> Iterator[bytes]:
    remaining = end - start + 1
    with path.open("rb") as stream:
        stream.seek(start)
        while remaining:
            chunk = stream.read(min(1024 * 1024, remaining))
            if not chunk:
                raise SessionIntegrityError("recorded perception video ended unexpectedly")
            remaining -= len(chunk)
            yield chunk


def _recorded_media_manifest_document(
    manifest: RecordedMediaManifest,
) -> dict[str, Any]:
    encoded_session_id = quote(manifest.session_id, safe="")
    encoded_artifact_id = quote(manifest.artifact_id, safe="")
    base = f"/api/v1/observation-sessions/{encoded_session_id}/media/{encoded_artifact_id}"
    return {
        "schema_version": RECORDED_MEDIA_STREAM_MANIFEST_SCHEMA,
        "source_id": manifest.public_source_id,
        "generation_sha256": manifest.generation_sha256,
        "byte_length": manifest.byte_length,
        "timeline_start_seconds": manifest.timeline_start_seconds,
        "timeline_end_seconds": manifest.timeline_end_seconds,
        "synchronization": manifest.synchronization,
        "epochs": [
            {
                "ordinal": epoch.ordinal,
                "timeline_start_seconds": epoch.timeline_start_seconds,
                "timeline_end_seconds": epoch.timeline_end_seconds,
                "media_type": epoch.media_type,
                "byte_length": epoch.init_byte_length
                + sum(segment.byte_length for segment in epoch.segments),
                "segment_count": len(epoch.segments),
                "random_access_sequences": sorted(
                    segment.sequence for segment in epoch.segments if segment.random_access
                ),
                "segment_end_times_seconds": [
                    segment.end_time_seconds for segment in epoch.segments
                ],
                "stream_url": (
                    f"{base}/epochs/{epoch.ordinal}/recording.mp4"
                    f"?generation={manifest.generation_sha256}"
                ),
            }
            for epoch in manifest.epochs
        ],
    }


def _resolve_recorded_media_manifest(
    store: SessionStore,
    inspector: RecordedMediaInspector,
    manager: SessionRecordingPreparationManager | None,
    allow_synchronous_fallback: bool,
    catalog_refresher: CatalogRefresher | None,
    session_id: str,
    artifact_id: str,
) -> RecordedMediaManifest:
    if manager is not None:
        try:
            _validate_source_id(artifact_id)
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail="Некорректный идентификатор записанного медиаканала.",
            ) from exc
        snapshot = manager.status(session_id)
        if snapshot is None or snapshot.state != "ready" or snapshot.recorded_media is None:
            raise HTTPException(
                status_code=409,
                detail="Медиаканал ещё не подготовлен.",
            )
        matches = tuple(
            manifest for manifest in snapshot.recorded_media if manifest.artifact_id == artifact_id
        )
        if len(matches) != 1:
            raise HTTPException(status_code=404, detail="Записанный медиаканал не найден.")
        return matches[0]
    if not allow_synchronous_fallback:
        raise HTTPException(
            status_code=503,
            detail="Сервис фоновой подготовки медиаканалов не настроен.",
        )
    try:
        command = _prepare_replay(
            store,
            catalog_refresher,
            session_id,
            1.0,
            False,
            False,
        )
        artifact = store.get_recorded_media(session_id, artifact_id)
        return inspector.inspect(artifact, command)
    except SessionNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except (SessionNotReplayableError, SessionIntegrityError) as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(
            status_code=422,
            detail="Некорректный идентификатор записанного медиаканала.",
        ) from exc


def _recorded_media_init_sha256(
    manifest: RecordedMediaManifest,
    epoch_ordinal: int,
) -> str:
    matches = tuple(epoch for epoch in manifest.epochs if epoch.ordinal == epoch_ordinal)
    if len(matches) != 1:
        raise HTTPException(status_code=404, detail="Эпоха записанного медиаканала не найдена.")
    return matches[0].init_sha256


def _recorded_media_segment_sha256(
    manifest: RecordedMediaManifest,
    epoch_ordinal: int,
    segment_sequence: int,
) -> str:
    matches = tuple(epoch for epoch in manifest.epochs if epoch.ordinal == epoch_ordinal)
    if len(matches) != 1:
        raise HTTPException(status_code=404, detail="Эпоха записанного медиаканала не найдена.")
    segments = tuple(
        segment for segment in matches[0].segments if segment.sequence == segment_sequence
    )
    if len(segments) != 1:
        raise HTTPException(status_code=404, detail="Сегмент записанного медиаканала не найден.")
    return segments[0].sha256


def _if_none_match_matches(value: str | None, etag: str) -> bool:
    if value is None:
        return False
    # RFC 9110 requires weak comparison for If-None-Match. The recording ETag
    # is content-addressed, so accepting its weak spelling is safe and lets a
    # browser reuse an already-downloaded multi-hundred-megabyte RRD.
    expected = etag.removeprefix("W/")
    for candidate in value.split(","):
        normalized = candidate.strip()
        if normalized == "*" or normalized.removeprefix("W/") == expected:
            return True
    return False


def _recorded_media_file_response(
    media_file: RecordedMediaFile,
    range_header: str | None,
) -> Response:
    payload = media_file.payload
    status_code = 200
    content_range: str | None = None
    if range_header is not None:
        start, end = _parse_byte_range(range_header, len(payload))
        payload = payload[start : end + 1]
        status_code = 206
        content_range = f"bytes {start}-{end}/{media_file.byte_length}"
    headers = {
        "Accept-Ranges": "bytes",
        "Cache-Control": "private, max-age=31536000, immutable, no-transform",
        "ETag": f'"sha256:{media_file.sha256}"',
        "Content-Length": str(len(payload)),
        "X-Content-Type-Options": "nosniff",
        "Content-Disposition": f'inline; filename="{media_file.filename}"',
    }
    if content_range is not None:
        headers["Content-Range"] = content_range
    return Response(
        content=payload,
        status_code=status_code,
        media_type=media_file.media_type,
        headers=headers,
    )


def _recorded_media_epoch_response(
    inspector: RecordedMediaInspector,
    manifest: RecordedMediaManifest,
    epoch_ordinal: int,
    range_header: str | None,
    *,
    head_only: bool,
) -> Response:
    matches = tuple(epoch for epoch in manifest.epochs if epoch.ordinal == epoch_ordinal)
    if len(matches) != 1:
        raise HTTPException(status_code=404, detail="Эпоха записанного медиаканала не найдена.")
    epoch = matches[0]
    byte_length = epoch.init_byte_length + sum(segment.byte_length for segment in epoch.segments)
    start, end = (0, byte_length - 1)
    status_code = 200
    if range_header is not None:
        start, end = _parse_byte_range(range_header, byte_length)
        status_code = 206

    etag = f'"generation:{manifest.generation_sha256}:epoch:{epoch.ordinal}"'
    headers = {
        "Accept-Ranges": "bytes",
        "Cache-Control": "private, max-age=31536000, immutable, no-transform",
        "ETag": etag,
        "Content-Length": str(end - start + 1),
        "X-Content-Type-Options": "nosniff",
        "Content-Disposition": (f'inline; filename="recorded-camera-{epoch.ordinal}.mp4"'),
    }
    if status_code == 206:
        headers["Content-Range"] = f"bytes {start}-{end}/{byte_length}"
    if head_only:
        return Response(
            status_code=status_code,
            media_type=epoch.media_type,
            headers=headers,
        )
    return StreamingResponse(
        _iter_recorded_media_epoch_range(
            inspector,
            manifest,
            epoch.ordinal,
            start,
            end,
        ),
        status_code=status_code,
        media_type=epoch.media_type,
        headers=headers,
    )


def _iter_recorded_media_epoch_range(
    inspector: RecordedMediaInspector,
    manifest: RecordedMediaManifest,
    epoch_ordinal: int,
    start: int,
    end: int,
) -> Iterator[bytes]:
    matches = tuple(epoch for epoch in manifest.epochs if epoch.ordinal == epoch_ordinal)
    if len(matches) != 1:
        raise SessionIntegrityError("recorded media epoch is unavailable")
    epoch = matches[0]
    offset = 0
    parts = chain(
        ((epoch.init_byte_length, None),),
        ((segment.byte_length, segment.sequence) for segment in epoch.segments),
    )
    for part_length, sequence in parts:
        part_end = offset + part_length - 1
        if part_end < start:
            offset += part_length
            continue
        if offset > end:
            break
        media_file = (
            inspector.get_init(manifest, epoch_ordinal)
            if sequence is None
            else inspector.get_segment(manifest, epoch_ordinal, sequence)
        )
        local_start = max(0, start - offset)
        local_end = min(part_length - 1, end - offset)
        yield media_file.payload[local_start : local_end + 1]
        offset += part_length


def _parse_byte_range(value: str, byte_length: int) -> tuple[int, int]:
    match = re.fullmatch(r"bytes=(\d*)-(\d*)", value.strip())
    if match is None or byte_length < 1:
        raise HTTPException(
            status_code=416,
            detail="Некорректный Range для записанного медиасегмента.",
            headers={"Content-Range": f"bytes */{byte_length}"},
        )
    raw_start, raw_end = match.groups()
    if not raw_start and not raw_end:
        raise HTTPException(
            status_code=416,
            detail="Некорректный Range для записанного медиасегмента.",
            headers={"Content-Range": f"bytes */{byte_length}"},
        )
    if not raw_start:
        suffix = int(raw_end)
        if suffix < 1:
            raise HTTPException(
                status_code=416,
                detail="Некорректный Range для записанного медиасегмента.",
                headers={"Content-Range": f"bytes */{byte_length}"},
            )
        start = max(0, byte_length - suffix)
        return start, byte_length - 1
    start = int(raw_start)
    end = byte_length - 1 if not raw_end else int(raw_end)
    if start >= byte_length or end < start:
        raise HTTPException(
            status_code=416,
            detail="Range выходит за границы записанного медиасегмента.",
            headers={"Content-Range": f"bytes */{byte_length}"},
        )
    return start, min(end, byte_length - 1)