from __future__ import annotations

import hashlib
import inspect
import json
import os
import re
import shutil
import stat
import threading
from collections.abc import Callable, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Any, cast
from uuid import uuid4

from k1link.artifact_gateway import (
    ArtifactGateway,
    ArtifactGatewayError,
    ArtifactNotFound,
    ArtifactStoreUnavailable,
)

from .models import ReplayArtifact, ReplayCommand
from .plugin_contract import (
    PluginRecordingExportCancelled,
    PluginRecordingExportError,
    RecordingExporter,
)

# v12 publishes a 2 Hz point-cloud operator projection while retaining complete
# normal K1 point batches. v11 also spatially thinned every ~2.4k-point frame;
# the latest-at AI view therefore looked visibly bald even though the raw
# capture was complete. Frames above the explicit emergency threshold remain
# bounded, and the native capture stays the source of record and AI input.
CACHE_SCHEMA = "missioncore.derived-rerun-recording-cache/v12"
COMPATIBLE_CACHE_SCHEMAS = frozenset({CACHE_SCHEMA})
RECORDING_CACHE_FILENAME = "scene.operator-v12.rrd"
RECORDING_CACHE_SIDECAR_FILENAME = f"{RECORDING_CACHE_FILENAME}.cache.json"
RERUN_RECORDING_MEDIA_TYPE = "application/vnd.rerun.rrd"
RERUN_SESSION_TIMELINE = "session_time"
SESSION_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
DEFAULT_FREE_SPACE_RESERVE_BYTES = 2 * 1024 * 1024 * 1024
DEFAULT_CACHE_MAX_BYTES = 8 * 1024 * 1024 * 1024

RrdExporter = Callable[..., Mapping[str, object]]
RecordingProgressCallback = Callable[[str, float], None]


class RecordingMaterializationError(RuntimeError):
    """A seekable browser recording could not be prepared safely."""


class RecordingMaterializationCancelled(RecordingMaterializationError):
    """A background conversion was cooperatively cancelled before publication."""


@dataclass(frozen=True, slots=True)
class MaterializedRecording:
    """Internal handle for a private, derived Rerun recording."""

    session_id: str
    path: Path
    media_type: str
    byte_length: int
    sha256: str
    source_sha256: str
    timeline: str
    timeline_start_ns: int
    timeline_end_ns: int


@dataclass(frozen=True, slots=True)
class RecordingCacheStatus:
    entry_count: int
    total_bytes: int
    cache_max_bytes: int
    free_space_reserve_bytes: int
    free_bytes: int


@dataclass(frozen=True, slots=True)
class _ValidatedMemoryEntry:
    source_identity: tuple[object, ...]
    recording_identity: tuple[int, int, int, int, int]
    recording: MaterializedRecording


@dataclass(frozen=True, slots=True)
class _ValidatedArtifact:
    artifact_id: str
    path: Path
    media_type: str
    file_stat: os.stat_result
    replay_byte_length: int
    expected_sha256: str | None

    @property
    def identity(self) -> tuple[object, ...]:
        return (
            self.artifact_id,
            self.media_type,
            *_stat_identity(self.file_stat),
            self.replay_byte_length,
            self.expected_sha256,
        )


@dataclass(frozen=True, slots=True)
class _ValidatedSource:
    plugin_id: str
    primary_artifact_id: str
    artifacts: tuple[_ValidatedArtifact, ...]

    @property
    def primary(self) -> _ValidatedArtifact:
        matches = tuple(
            artifact
            for artifact in self.artifacts
            if artifact.artifact_id == self.primary_artifact_id
        )
        if len(matches) != 1:
            raise RecordingMaterializationError("recording source has no primary artifact")
        return matches[0]

    @property
    def identity(self) -> tuple[object, ...]:
        return (
            self.plugin_id,
            self.primary_artifact_id,
            *(item for artifact in self.artifacts for item in artifact.identity),
        )

    @property
    def replay_byte_length(self) -> int:
        return sum(artifact.replay_byte_length for artifact in self.artifacts)


class SessionRecordingMaterializer:
    """Build and validate a per-session seekable RRD under the private data root.

    Native plugin evidence remains the source of record. Derived RRDs
    live below ``data_dir/recordings`` and are reused only when both their
    source identity and output digest still match an atomically written cache
    sidecar. Calls for one session are serialized, so concurrent browser
    requests cannot start duplicate exports or observe a half-published file.
    """

    def __init__(
        self,
        data_dir: Path,
        *,
        exporter: RrdExporter | None = None,
        exporters: Mapping[str, RecordingExporter] | None = None,
        cache_max_bytes: int | None = None,
        free_space_reserve_bytes: int | None = None,
        artifact_gateway: ArtifactGateway | None = None,
    ) -> None:
        private_root = data_dir.expanduser().resolve()
        private_root.mkdir(mode=0o700, parents=True, exist_ok=True)
        _chmod_best_effort(private_root, 0o700)
        recordings_root = private_root / "recordings"
        recordings_root.mkdir(mode=0o700, parents=True, exist_ok=True)
        if recordings_root.is_symlink():
            raise RecordingMaterializationError("recording cache root must not be a symlink")
        _chmod_best_effort(recordings_root, 0o700)
        self.recordings_root = recordings_root.resolve()
        if not self.recordings_root.is_relative_to(private_root):
            raise RecordingMaterializationError("recording cache escapes the private data root")
        if exporter is not None and exporters is not None:
            raise ValueError("configure either one test exporter or plugin exporters")
        self._fallback_exporter = exporter
        self._exporters = dict(exporters or {})
        if len(self._exporters) != len(set(self._exporters)):
            raise ValueError("recording exporter plugin ids must be unique")
        configured_cache_max_bytes = _optional_positive_configuration(
            cache_max_bytes,
            environment_name="MISSIONCORE_RRD_CACHE_MAX_BYTES",
        )
        self.cache_max_bytes = (
            configured_cache_max_bytes
            if configured_cache_max_bytes is not None
            else DEFAULT_CACHE_MAX_BYTES
        )
        self.free_space_reserve_bytes = _non_negative_configuration(
            free_space_reserve_bytes,
            environment_name="MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES",
            default=DEFAULT_FREE_SPACE_RESERVE_BYTES,
        )
        self.artifact_gateway = artifact_gateway
        self._locks_guard = threading.Lock()
        self._session_locks: dict[str, threading.Lock] = {}
        self._memory_guard = threading.Lock()
        self._validated_memory: dict[str, _ValidatedMemoryEntry] = {}
        # Expensive RRD exports stay single-flight, while this much shorter
        # guard protects cache publication, eviction and response leases. A
        # cache hit must never wait behind a multi-minute export of another
        # session.
        self._export_lock = threading.Lock()
        self._cache_guard = threading.RLock()
        self._pinned_sessions: dict[str, int] = {}
        self._interprocess_lock_path = self.recordings_root / ".export.lock"

    @property
    def supports_cooperative_cancellation(self) -> bool:
        """Whether the configured exporter observes a cancellation event."""

        exporters: tuple[Callable[..., object], ...] = tuple(self._exporters.values())
        if self._fallback_exporter is not None:
            exporters = (*exporters, self._fallback_exporter)
        return bool(exporters) and all(
            _callable_accepts_keyword(exporter, "cancel_event") for exporter in exporters
        )

    def cache_status(self) -> RecordingCacheStatus:
        """Return bounded cache capacity and occupancy for runtime readiness."""

        with self._cache_guard:
            total_bytes, entries = _cache_entries(self.recordings_root)
            free_bytes = shutil.disk_usage(self.recordings_root).free
        return RecordingCacheStatus(
            entry_count=len(entries),
            total_bytes=total_bytes,
            cache_max_bytes=self.cache_max_bytes,
            free_space_reserve_bytes=self.free_space_reserve_bytes,
            free_bytes=free_bytes,
        )

    def is_recording_available(self, recording: MaterializedRecording) -> bool:
        """Cheap no-follow check for a previously validated ready handle."""

        path = recording.path
        try:
            path_stat = path.lstat()
        except OSError:
            return False
        return (
            path.parent.parent == self.recordings_root
            and path.name == RECORDING_CACHE_FILENAME
            and stat.S_ISREG(path_stat.st_mode)
            and not stat.S_ISLNK(path_stat.st_mode)
            and path_stat.st_size == recording.byte_length
        )

    def pin_recording(
        self,
        recording: MaterializedRecording,
    ) -> Callable[[], None] | None:
        """Lease an in-memory validated handle using only cheap file metadata."""

        if SESSION_ID_PATTERN.fullmatch(recording.session_id) is None:
            return None
        with self._cache_guard:
            if not self.is_recording_available(recording):
                return None
            self._increment_pin_locked(recording.session_id)
        return self._release_callback(recording.session_id)

    def delete_cached(self, session_id: str) -> bool:
        """Delete one exact derived RRD cache unless a response still leases it."""

        if SESSION_ID_PATTERN.fullmatch(session_id) is None:
            raise ValueError("recording cache session id is invalid")
        with self._lock_for(session_id), self._cache_guard:
            if self._pinned_sessions.get(session_id, 0) > 0:
                return False
            session_root = self.recordings_root / session_id
            if session_root.exists() or session_root.is_symlink():
                metadata = session_root.lstat()
                if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
                    raise RecordingMaterializationError(
                        "session recording cache is not a real directory"
                    )
                if session_root.parent.resolve(strict=True) != self.recordings_root:
                    raise RecordingMaterializationError(
                        "session recording cache escapes the private data root"
                    )
                shutil.rmtree(session_root)
            with self._memory_guard:
                self._validated_memory.pop(session_id, None)
            return True

    def __call__(self, command: ReplayCommand) -> MaterializedRecording:
        """Alias for :meth:`materialize`, suitable for the web API protocol."""

        return self.materialize(command)

    def get_cached(self, command: ReplayCommand) -> MaterializedRecording | None:
        """Return a validated cache hit without starting an export."""

        session_id = _validate_command_shape(command)
        if not self._has_compatible_cache_candidate(session_id):
            return None
        with self._lock_for(session_id):
            return self._load_cached_recording(session_id, _validate_source(command))

    def restore_published(self, command: ReplayCommand) -> MaterializedRecording | None:
        """Restore an immutable published cache using bounded metadata checks.

        Full source/output digests are proved before the cache is published.
        On a later process start the private cache is restored only when its
        schema, source stat identities and derived-file stat identity still
        match the durable sidecar. This keeps a prepared recording durable
        across restarts without rereading gigabytes or starting an exporter.
        """

        session_id = _validate_command_shape(command)
        if not self._has_compatible_cache_candidate(session_id):
            return None
        with self._lock_for(session_id):
            return self._load_cached_recording(
                session_id,
                _validate_source(command),
                verify_digests=False,
            )

    def get_cached_pinned(
        self,
        command: ReplayCommand,
    ) -> tuple[MaterializedRecording, Callable[[], None]] | None:
        """Lease a validated cache hit without starting an export."""

        session_id = _validate_command_shape(command)
        if not self._has_compatible_cache_candidate(session_id):
            return None
        with self._lock_for(session_id):
            recording = self._load_cached_recording(
                session_id,
                _validate_source(command),
                pin=True,
            )
        if recording is None:
            return None
        return recording, self._release_callback(session_id)

    def _has_compatible_cache_candidate(self, session_id: str) -> bool:
        _root, recording_path, sidecar_path = self._cache_paths(session_id)
        if (
            recording_path.is_symlink()
            or sidecar_path.is_symlink()
            or not recording_path.is_file()
            or not sidecar_path.is_file()
        ):
            return False
        try:
            value = json.loads(sidecar_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return False
        return isinstance(value, dict) and value.get("schema_version") in COMPATIBLE_CACHE_SCHEMAS

    def source_identity(self, command: ReplayCommand) -> tuple[object, ...]:
        """Return the validated source identity used for job deduplication."""

        session_id = _validate_command_shape(command)
        return (session_id, *_validate_source(command).identity)

    def materialize(
        self,
        command: ReplayCommand,
        *,
        progress_callback: RecordingProgressCallback | None = None,
        cancel_event: threading.Event | None = None,
    ) -> MaterializedRecording:
        session_id = _validate_command_shape(command)
        session_lock = self._lock_for(session_id)
        # Per-session locks deduplicate callers for one recording. Validate a
        # ready cache before entering the global export gate so another large
        # conversion cannot block playback of this already prepared session.
        with session_lock:
            _report_progress(progress_callback, "validating", 0.05)
            _raise_if_cancelled(cancel_event)
            source = _validate_source(command)
            cached = self._load_cached_recording(session_id, source)
            if cached is not None:
                _report_progress(progress_callback, "ready", 1.0)
                return cached
            # One global export gate bounds peak CPU, RAM and temporary disk
            # use. Revalidate after waiting because the native capture may
            # have changed while another session was being converted.
            with self._export_lock, _exclusive_file_lock(self._interprocess_lock_path):
                self._scavenge_export_artifacts_locked()
                _raise_if_cancelled(cancel_event)
                source = _validate_source(command)
                cached = self._load_cached_recording(session_id, source)
                if cached is not None:
                    _report_progress(progress_callback, "ready", 1.0)
                    return cached
                _report_progress(progress_callback, "restoring", 0.15)
                restored = self._restore_gateway_recording_locked(session_id, source)
                if restored is not None:
                    _report_progress(progress_callback, "ready", 1.0)
                    return restored
                _report_progress(progress_callback, "exporting", 0.2)
                return self._export_recording_locked(
                    session_id,
                    source,
                    progress_callback=progress_callback,
                    cancel_event=cancel_event,
                )

    def materialize_pinned(
        self,
        command: ReplayCommand,
    ) -> tuple[MaterializedRecording, Callable[[], None]]:
        """Materialize and lease one cache entry for the response lifetime."""

        session_id = _validate_command_shape(command)
        session_lock = self._lock_for(session_id)
        with session_lock:
            source = _validate_source(command)
            recording = self._load_cached_recording(session_id, source, pin=True)
            if recording is None:
                with self._export_lock, _exclusive_file_lock(self._interprocess_lock_path):
                    self._scavenge_export_artifacts_locked()
                    source = _validate_source(command)
                    recording = self._load_cached_recording(session_id, source, pin=True)
                    if recording is None:
                        recording = self._restore_gateway_recording_locked(session_id, source)
                    if recording is None:
                        recording = self._export_recording_locked(session_id, source)
                    with self._cache_guard:
                        self._increment_pin_locked(session_id)

        return recording, self._release_callback(session_id)

    def _restore_gateway_recording_locked(
        self,
        session_id: str,
        source: _ValidatedSource,
    ) -> MaterializedRecording | None:
        if self.artifact_gateway is None:
            return None
        try:
            resolved = self.artifact_gateway.resolve_role(
                "sessions",
                session_id,
                "base-rrd",
            )
        except ArtifactNotFound:
            return None
        except ArtifactStoreUnavailable as exc:
            raise RecordingMaterializationError(
                "central recording is unavailable and is not present "
                "in the local artifact cache"
            ) from exc
        except ArtifactGatewayError as exc:
            raise RecordingMaterializationError(
                "central recording artifact failed validation"
            ) from exc
        metadata = resolved.manifest.metadata
        if (
            resolved.manifest.artifact_type != "recorded-session"
            or resolved.manifest.subject_id != session_id
            or resolved.member.media_type != RERUN_RECORDING_MEDIA_TYPE
            or metadata.get("base-timeline") != RERUN_SESSION_TIMELINE
        ):
            raise RecordingMaterializationError("central recording manifest is invalid")
        source_sha256 = metadata.get("base-source-sha256")
        try:
            timeline_start_ns = int(metadata["base-timeline-start-ns"])
            timeline_end_ns = int(metadata["base-timeline-end-ns"])
        except (KeyError, TypeError, ValueError) as exc:
            raise RecordingMaterializationError(
                "central recording timeline metadata is invalid"
            ) from exc
        artifact_digests = _validated_artifact_digests(source)
        if (
            not isinstance(source_sha256, str)
            or source_sha256 != artifact_digests[source.primary_artifact_id]
            or timeline_start_ns < 0
            or timeline_end_ns < timeline_start_ns
        ):
            raise RecordingMaterializationError(
                "central recording does not match the local source evidence"
            )
        with self._cache_guard:
            session_root, recording_path, sidecar_path = self._cache_paths(session_id)
            self._ensure_cache_capacity(
                required_bytes=resolved.member.byte_length + 4 * 1024,
                protected_session_id=session_id,
            )
            candidate = session_root / f".scene.{uuid4().hex}.candidate.rrd"
            try:
                digest = hashlib.sha256()
                byte_length = 0
                with resolved.path.open("rb") as source_stream, candidate.open("xb") as output:
                    while chunk := source_stream.read(1024 * 1024):
                        output.write(chunk)
                        digest.update(chunk)
                        byte_length += len(chunk)
                    output.flush()
                    os.fsync(output.fileno())
                if (
                    digest.hexdigest() != resolved.member.sha256
                    or byte_length != resolved.member.byte_length
                ):
                    raise RecordingMaterializationError(
                        "central recording changed during local restoration"
                    )
                _chmod_best_effort(candidate, 0o600)
                os.replace(candidate, recording_path)
                _fsync_directory(session_root)
                recording_stat = _regular_file_stat(recording_path, "derived recording")
                recording = MaterializedRecording(
                    session_id=session_id,
                    path=recording_path,
                    media_type=RERUN_RECORDING_MEDIA_TYPE,
                    byte_length=byte_length,
                    sha256=resolved.member.sha256,
                    source_sha256=source_sha256,
                    timeline=RERUN_SESSION_TIMELINE,
                    timeline_start_ns=timeline_start_ns,
                    timeline_end_ns=timeline_end_ns,
                )
                _write_json_atomic(
                    sidecar_path,
                    _cache_document(recording, source, recording_stat),
                )
                _chmod_best_effort(sidecar_path, 0o600)
                self._remember(recording, source, recording_stat)
                _touch_lru(session_root)
                return recording
            except OSError as exc:
                raise RecordingMaterializationError(
                    "central recording could not be restored locally"
                ) from exc
            finally:
                candidate.unlink(missing_ok=True)

    def _release_callback(self, session_id: str) -> Callable[[], None]:
        released = False
        release_guard = threading.Lock()

        def release() -> None:
            nonlocal released
            with release_guard:
                if released:
                    return
                released = True
            with self._cache_guard:
                count = self._pinned_sessions.get(session_id, 0)
                if count <= 1:
                    self._pinned_sessions.pop(session_id, None)
                else:
                    self._pinned_sessions[session_id] = count - 1

        return release

    def _scavenge_export_artifacts_locked(self) -> None:
        """Remove crash leftovers while holding the cross-process writer lock."""

        try:
            session_roots = tuple(self.recordings_root.iterdir())
        except OSError as exc:
            raise RecordingMaterializationError("recording cache could not be scavenged") from exc
        for session_root in session_roots:
            try:
                root_stat = session_root.lstat()
            except OSError:
                continue
            if stat.S_ISLNK(root_stat.st_mode) or not stat.S_ISDIR(root_stat.st_mode):
                continue
            try:
                children = tuple(session_root.iterdir())
            except OSError:
                continue
            for child in children:
                name = child.name
                stale = (
                    (name.startswith(".scene.") and name.endswith(".candidate.rrd"))
                    or (
                        name.startswith("..scene.")
                        and ".candidate.rrd." in name
                        and name.endswith(".tmp")
                    )
                    or (name.startswith(".source.") and name.endswith(".tmp"))
                    or (
                        name.startswith(f".{RECORDING_CACHE_SIDECAR_FILENAME}.")
                        and name.endswith(".tmp")
                    )
                )
                if not stale:
                    continue
                try:
                    child_stat = child.lstat()
                    if stat.S_ISDIR(child_stat.st_mode) and not stat.S_ISLNK(child_stat.st_mode):
                        shutil.rmtree(child)
                    else:
                        child.unlink()
                except FileNotFoundError:
                    continue
                except OSError as exc:
                    raise RecordingMaterializationError(
                        "stale recording export artifact could not be removed"
                    ) from exc

    def _lock_for(self, session_id: str) -> threading.Lock:
        with self._locks_guard:
            lock = self._session_locks.get(session_id)
            if lock is None:
                lock = threading.Lock()
                self._session_locks[session_id] = lock
            return lock

    def _cache_paths(
        self,
        session_id: str,
    ) -> tuple[Path, Path, Path]:
        session_root = self.recordings_root / session_id
        session_root.mkdir(mode=0o700, parents=True, exist_ok=True)
        if session_root.is_symlink():
            raise RecordingMaterializationError("session recording cache must not be a symlink")
        resolved_session_root = session_root.resolve()
        if not resolved_session_root.is_relative_to(self.recordings_root):
            raise RecordingMaterializationError("recording cache escapes the private data root")
        _chmod_best_effort(resolved_session_root, 0o700)
        return (
            resolved_session_root,
            resolved_session_root / RECORDING_CACHE_FILENAME,
            resolved_session_root / RECORDING_CACHE_SIDECAR_FILENAME,
        )

    def _load_cached_recording(
        self,
        session_id: str,
        source: _ValidatedSource,
        *,
        pin: bool = False,
        verify_digests: bool = True,
    ) -> MaterializedRecording | None:
        # Cache validation and the optional lease are one transaction with
        # eviction. This makes it safe for FileResponse to open the path after
        # this method returns even while another session is being exported.
        with self._cache_guard:
            resolved_session_root, recording_path, sidecar_path = self._cache_paths(session_id)
            memory_cached = self._load_memory_cache(
                session_id=session_id,
                source=source,
                recording_path=recording_path,
            )
            if memory_cached is not None:
                if pin:
                    self._increment_pin_locked(session_id)
                _touch_lru(resolved_session_root)
                return memory_cached
            cached = self._load_valid_cache(
                session_id=session_id,
                source=source,
                recording_path=recording_path,
                sidecar_path=sidecar_path,
                verify_digests=verify_digests,
            )
            if cached is None:
                return None
            self._remember(cached, source, _regular_file_stat(recording_path, "recording"))
            if pin:
                self._increment_pin_locked(session_id)
            _touch_lru(resolved_session_root)
            return cached

    def _increment_pin_locked(self, session_id: str) -> None:
        self._pinned_sessions[session_id] = self._pinned_sessions.get(session_id, 0) + 1

    def _export_recording_locked(
        self,
        session_id: str,
        source: _ValidatedSource,
        *,
        progress_callback: RecordingProgressCallback | None = None,
        cancel_event: threading.Event | None = None,
    ) -> MaterializedRecording:
        with self._cache_guard:
            resolved_session_root, recording_path, sidecar_path = self._cache_paths(session_id)
            self._ensure_cache_capacity(
                required_bytes=max(source.replay_byte_length * 2, 1) + 4 * 1024,
                protected_session_id=session_id,
            )

        # Keep a valid old artifact and sidecar published while rebuilding.
        # The unique confined candidate is atomically swapped in only after
        # export, source-stability and digest validation have all succeeded.
        candidate_path = resolved_session_root / f".scene.{uuid4().hex}.candidate.rrd"
        try:
            candidate_path.touch(mode=0o600, exist_ok=False)
        except OSError as exc:
            raise RecordingMaterializationError(
                "derived recording candidate could not be reserved"
            ) from exc

        staged_root: Path | None = None
        export_source = source.primary.path
        export_artifacts = {artifact.artifact_id: artifact.path for artifact in source.artifacts}
        try:
            # Catalog digests bind every replay input, not only the primary
            # raw stream. In particular the capture-clock artifact must be
            # proven before its bounds influence an RRD export.
            _validated_artifact_digests(source)
            if any(
                artifact.replay_byte_length != artifact.file_stat.st_size
                for artifact in source.artifacts
            ):
                staged_root, export_source, export_artifacts = _stage_replay_prefix(
                    resolved_session_root,
                    source,
                    cancel_event=cancel_event,
                    activity_callback=lambda: _report_progress(
                        progress_callback,
                        "exporting",
                        0.3,
                    ),
                )
            summary = self._invoke_exporter(
                source.plugin_id,
                export_source,
                candidate_path,
                artifacts=export_artifacts,
                cancel_event=cancel_event,
                activity_callback=lambda: _report_progress(
                    progress_callback,
                    "exporting",
                    0.5,
                ),
            )
            _raise_if_cancelled(cancel_event)
            _report_progress(progress_callback, "finalizing", 0.9)
        except PluginRecordingExportCancelled as exc:
            candidate_path.unlink(missing_ok=True)
            raise RecordingMaterializationCancelled("recording preparation was cancelled") from exc
        except PluginRecordingExportError as exc:
            candidate_path.unlink(missing_ok=True)
            raise RecordingMaterializationError("native capture could not be exported") from exc
        except OSError as exc:
            candidate_path.unlink(missing_ok=True)
            raise RecordingMaterializationError("derived recording could not be written") from exc
        except BaseException:
            candidate_path.unlink(missing_ok=True)
            raise
        finally:
            if staged_root is not None:
                shutil.rmtree(staged_root, ignore_errors=True)

        try:
            source_after = _validate_source_state(source)
            if source_after.identity != source.identity:
                raise RecordingMaterializationError("native capture changed during RRD export")
            _chmod_best_effort(candidate_path, 0o600)
            artifact_digests = _validated_artifact_digests(source)
            source_sha256 = artifact_digests[source.primary_artifact_id]
            candidate = _materialized_from_export(
                session_id=session_id,
                source_sha256=source_sha256,
                recording_path=candidate_path,
                summary=summary,
            )
        except BaseException:
            candidate_path.unlink(missing_ok=True)
            raise
        try:
            with self._cache_guard:
                candidate_stat = _regular_file_stat(
                    candidate_path,
                    "derived recording candidate",
                )
                if (
                    self.cache_max_bytes is not None
                    and candidate_stat.st_size > self.cache_max_bytes
                ):
                    raise RecordingMaterializationError("derived recording exceeds the cache quota")
                self._ensure_cache_capacity(
                    required_bytes=4 * 1024,
                    protected_session_id=session_id,
                )
                _raise_if_cancelled(cancel_event)
                os.replace(candidate_path, recording_path)
                _fsync_directory(resolved_session_root)
                recording_stat = _regular_file_stat(recording_path, "derived recording")
                materialized = MaterializedRecording(
                    session_id=candidate.session_id,
                    path=recording_path,
                    media_type=candidate.media_type,
                    byte_length=candidate.byte_length,
                    sha256=candidate.sha256,
                    source_sha256=candidate.source_sha256,
                    timeline=candidate.timeline,
                    timeline_start_ns=candidate.timeline_start_ns,
                    timeline_end_ns=candidate.timeline_end_ns,
                )
                document = _cache_document(materialized, source, recording_stat)
                _write_json_atomic(sidecar_path, document)
                _chmod_best_effort(sidecar_path, 0o600)
                self._remember(materialized, source, recording_stat)
                _touch_lru(resolved_session_root)
        finally:
            candidate_path.unlink(missing_ok=True)
        _report_progress(progress_callback, "ready", 1.0)
        return materialized

    def _invoke_exporter(
        self,
        plugin_id: str,
        source: Path,
        destination: Path,
        *,
        artifacts: Mapping[str, Path],
        cancel_event: threading.Event | None,
        activity_callback: Callable[[], None],
    ) -> Mapping[str, object]:
        selected = self._fallback_exporter or self._exporters.get(plugin_id)
        if selected is None:
            raise PluginRecordingExportError(
                f"device plugin has no recording exporter: {plugin_id}"
            )
        exporter = cast(RrdExporter, selected)
        kwargs: dict[str, object] = {}
        if _callable_accepts_keyword(exporter, "artifacts"):
            kwargs["artifacts"] = dict(artifacts)
        if _callable_accepts_keyword(exporter, "cancel_event"):
            kwargs["cancel_event"] = cancel_event
        if _callable_accepts_keyword(exporter, "activity_callback"):
            kwargs["activity_callback"] = activity_callback
        return exporter(source, destination, **kwargs)

    def _load_memory_cache(
        self,
        *,
        session_id: str,
        source: _ValidatedSource,
        recording_path: Path,
    ) -> MaterializedRecording | None:
        with self._memory_guard:
            entry = self._validated_memory.get(session_id)
        if entry is None or entry.source_identity != source.identity:
            return None
        if recording_path.is_symlink():
            return None
        try:
            recording_stat = _regular_file_stat(recording_path, "derived recording")
        except RecordingMaterializationError:
            return None
        if entry.recording_identity != _stat_identity(recording_stat):
            return None
        return entry.recording

    def _remember(
        self,
        recording: MaterializedRecording,
        source: _ValidatedSource,
        recording_stat: os.stat_result,
    ) -> None:
        entry = _ValidatedMemoryEntry(
            source_identity=source.identity,
            recording_identity=_stat_identity(recording_stat),
            recording=recording,
        )
        with self._memory_guard:
            self._validated_memory[recording.session_id] = entry

    def _load_valid_cache(
        self,
        *,
        session_id: str,
        source: _ValidatedSource,
        recording_path: Path,
        sidecar_path: Path,
        verify_digests: bool = True,
    ) -> MaterializedRecording | None:
        if recording_path.is_symlink() or sidecar_path.is_symlink():
            return None
        try:
            raw_document = json.loads(sidecar_path.read_text(encoding="utf-8"))
            document = _validate_cache_document(raw_document, session_id)
            recording_stat = _regular_file_stat(recording_path, "derived recording")
        except (OSError, ValueError, json.JSONDecodeError, RecordingMaterializationError):
            return None

        if document["plugin_id"] != source.plugin_id:
            return None
        if document["primary_artifact_id"] != source.primary_artifact_id:
            return None
        cached_artifacts = document["source_artifacts"]
        if len(cached_artifacts) != len(source.artifacts):
            return None
        for cached, artifact in zip(cached_artifacts, source.artifacts, strict=True):
            if cached["artifact_id"] != artifact.artifact_id:
                return None
            if cached["media_type"] != artifact.media_type:
                return None
            if cached["file_byte_length"] != artifact.file_stat.st_size:
                return None
            if cached["mtime_ns"] != artifact.file_stat.st_mtime_ns:
                return None
            if cached["ctime_ns"] != artifact.file_stat.st_ctime_ns:
                return None
            if cached["replay_byte_length"] != artifact.replay_byte_length:
                return None
        if document["recording_byte_length"] != recording_stat.st_size:
            return None
        if document["recording_mtime_ns"] != recording_stat.st_mtime_ns:
            return None

        source_sha256: str | None = None
        for cached, artifact in zip(cached_artifacts, source.artifacts, strict=True):
            digest = (
                _sha256_prefix_stable(
                    artifact.path,
                    artifact.file_stat,
                    artifact.replay_byte_length,
                )
                if verify_digests
                else cached["sha256"]
            )
            if (
                artifact.expected_sha256 is not None
                and cached["sha256"] != artifact.expected_sha256
            ):
                raise RecordingMaterializationError(
                    "recording artifact digest no longer matches catalog"
                )
            if digest != cached["sha256"]:
                return None
            if artifact.artifact_id == source.primary_artifact_id:
                source_sha256 = digest
        if source_sha256 is None or source_sha256 != document["source_sha256"]:
            return None
        recording_sha256 = document["recording_sha256"]
        if verify_digests:
            recording_sha256 = _sha256_stable(recording_path, recording_stat)
            if recording_sha256 != document["recording_sha256"]:
                return None
        _chmod_best_effort(recording_path, 0o600)
        _chmod_best_effort(sidecar_path, 0o600)
        if document["schema_version"] != CACHE_SCHEMA:
            migrated = dict(document)
            migrated["schema_version"] = CACHE_SCHEMA
            _write_json_atomic(sidecar_path, migrated)
            _chmod_best_effort(sidecar_path, 0o600)
        return MaterializedRecording(
            session_id=session_id,
            path=recording_path,
            media_type=RERUN_RECORDING_MEDIA_TYPE,
            byte_length=recording_stat.st_size,
            sha256=recording_sha256,
            source_sha256=source_sha256,
            timeline=RERUN_SESSION_TIMELINE,
            timeline_start_ns=document["timeline_start_ns"],
            timeline_end_ns=document["timeline_end_ns"],
        )

    def _ensure_cache_capacity(
        self,
        *,
        required_bytes: int,
        protected_session_id: str,
    ) -> None:
        if required_bytes < 0:
            raise RecordingMaterializationError("cache reservation is invalid")
        while True:
            total_bytes, entries = _cache_entries(self.recordings_root)
            free_bytes = shutil.disk_usage(self.recordings_root).free
            quota_ok = (
                self.cache_max_bytes is None
                or total_bytes + required_bytes <= self.cache_max_bytes
            )
            reserve_ok = free_bytes >= self.free_space_reserve_bytes + required_bytes
            if quota_ok and reserve_ok:
                return
            victim = next(
                (
                    entry
                    for entry in entries
                    if entry[0].name != protected_session_id
                    and not self._session_is_pinned(entry[0].name)
                ),
                None,
            )
            if victim is None:
                reason = "cache quota" if not quota_ok else "free-space reserve"
                raise RecordingMaterializationError(
                    f"derived recording cannot satisfy the {reason}"
                )
            victim_path, _victim_bytes, _victim_mtime = victim
            _remove_cache_session(self.recordings_root, victim_path)
            with self._memory_guard:
                self._validated_memory.pop(victim_path.name, None)

    def _session_is_pinned(self, session_id: str) -> bool:
        with self._cache_guard:
            return self._pinned_sessions.get(session_id, 0) > 0


def _validate_command_shape(command: ReplayCommand) -> str:
    session_id = getattr(command, "session_id", None)
    if not isinstance(session_id, str) or SESSION_ID_PATTERN.fullmatch(session_id) is None:
        raise RecordingMaterializationError("session id has an invalid shape")
    return session_id


def _report_progress(
    callback: RecordingProgressCallback | None,
    state: str,
    progress: float,
) -> None:
    if callback is not None:
        callback(state, progress)


def _raise_if_cancelled(cancel_event: threading.Event | None) -> None:
    if cancel_event is not None and cancel_event.is_set():
        raise RecordingMaterializationCancelled("recording preparation was cancelled")


def _validate_source(command: ReplayCommand) -> _ValidatedSource:
    plugin_id = getattr(command, "plugin_id", None)
    allowed_root = getattr(command, "allowed_root", None)
    session_root = getattr(command, "session_root", None)
    primary_artifact_id = getattr(command, "primary_artifact_id", None)
    artifacts = getattr(command, "artifacts", None)
    if not isinstance(plugin_id, str) or SESSION_ID_PATTERN.fullmatch(plugin_id) is None:
        raise RecordingMaterializationError("replay command has an invalid plugin id")
    if (
        not isinstance(primary_artifact_id, str)
        or SESSION_ID_PATTERN.fullmatch(primary_artifact_id) is None
    ):
        raise RecordingMaterializationError("replay command has an invalid primary artifact")
    if not isinstance(artifacts, tuple) or not artifacts:
        raise RecordingMaterializationError("replay command has no source artifacts")
    try:
        if not isinstance(allowed_root, Path) or not isinstance(session_root, Path):
            raise RecordingMaterializationError("replay command has no confinement roots")
        allowed = allowed_root.expanduser().resolve(strict=True)
        session = session_root.expanduser().resolve(strict=True)
    except OSError as exc:
        raise RecordingMaterializationError("recording confinement root is missing") from exc
    if not allowed.is_dir() or not session.is_dir() or not session.is_relative_to(allowed):
        raise RecordingMaterializationError("recording source escapes its allowed session root")

    validated: list[_ValidatedArtifact] = []
    seen_ids: set[str] = set()
    seen_names: set[str] = set()
    for artifact in artifacts:
        if not isinstance(artifact, ReplayArtifact):
            raise RecordingMaterializationError("replay command contains an invalid artifact")
        if (
            SESSION_ID_PATTERN.fullmatch(artifact.artifact_id) is None
            or artifact.artifact_id in seen_ids
        ):
            raise RecordingMaterializationError("replay artifact id is invalid or duplicated")
        path = artifact.path.expanduser().absolute()
        try:
            parent = path.parent.resolve(strict=True)
        except OSError as exc:
            raise RecordingMaterializationError("recording source artifact is missing") from exc
        if not parent.is_relative_to(session):
            raise RecordingMaterializationError("recording source artifact escapes its session")
        file_stat = _regular_file_stat_nofollow(path, "recording source artifact")
        if artifact.file_byte_length > file_stat.st_size:
            raise RecordingMaterializationError("recording artifact was truncated after cataloging")
        if (
            isinstance(artifact.replay_byte_length, bool)
            or not 1 <= artifact.replay_byte_length <= file_stat.st_size
        ):
            raise RecordingMaterializationError("recording artifact replay boundary is invalid")
        if artifact.expected_sha256 is not None:
            if not _is_sha256(artifact.expected_sha256):
                raise RecordingMaterializationError("recording artifact digest is invalid")
            if artifact.replay_byte_length != file_stat.st_size:
                raise RecordingMaterializationError(
                    "a full-artifact digest cannot describe a replay prefix"
                )
        if path.name in seen_names:
            raise RecordingMaterializationError("recording artifact filenames are duplicated")
        seen_ids.add(artifact.artifact_id)
        seen_names.add(path.name)
        validated.append(
            _ValidatedArtifact(
                artifact_id=artifact.artifact_id,
                path=path,
                media_type=artifact.media_type,
                file_stat=file_stat,
                replay_byte_length=artifact.replay_byte_length,
                expected_sha256=artifact.expected_sha256,
            )
        )
    if sum(artifact.artifact_id == primary_artifact_id for artifact in validated) != 1:
        raise RecordingMaterializationError("recording primary artifact is unavailable")
    return _ValidatedSource(
        plugin_id=plugin_id,
        primary_artifact_id=primary_artifact_id,
        artifacts=tuple(validated),
    )


def _validate_source_state(source: _ValidatedSource) -> _ValidatedSource:
    return _ValidatedSource(
        plugin_id=source.plugin_id,
        primary_artifact_id=source.primary_artifact_id,
        artifacts=tuple(
            _ValidatedArtifact(
                artifact_id=artifact.artifact_id,
                path=artifact.path,
                media_type=artifact.media_type,
                file_stat=_regular_file_stat_nofollow(
                    artifact.path,
                    "recording source artifact",
                ),
                replay_byte_length=artifact.replay_byte_length,
                expected_sha256=artifact.expected_sha256,
            )
            for artifact in source.artifacts
        ),
    )


def _validated_artifact_digests(source: _ValidatedSource) -> dict[str, str]:
    digests: dict[str, str] = {}
    for artifact in source.artifacts:
        digest = _sha256_prefix_stable(
            artifact.path,
            artifact.file_stat,
            artifact.replay_byte_length,
        )
        if artifact.expected_sha256 is not None and digest != artifact.expected_sha256:
            raise RecordingMaterializationError(
                "recording artifact digest no longer matches catalog"
            )
        digests[artifact.artifact_id] = digest
    return digests


def _materialized_from_export(
    *,
    session_id: str,
    source_sha256: str,
    recording_path: Path,
    summary: Mapping[str, object],
) -> MaterializedRecording:
    recording_stat = _regular_file_stat(recording_path, "derived recording")
    exported_source_sha256 = _require_sha256(summary, "source_sha256")
    recording_sha256 = _require_sha256(summary, "rrd_sha256")
    if exported_source_sha256 != source_sha256:
        raise RecordingMaterializationError("export source digest does not match native capture")
    if recording_sha256 != _sha256_stable(recording_path, recording_stat):
        raise RecordingMaterializationError("export digest does not match derived recording")
    if _require_int(summary, "rrd_bytes") != recording_stat.st_size:
        raise RecordingMaterializationError("export byte length does not match derived recording")
    if summary.get("timeline") != RERUN_SESSION_TIMELINE:
        raise RecordingMaterializationError("export uses an unsupported playback timeline")
    timeline_start_ns = _require_int(summary, "timeline_start_ns")
    timeline_end_ns = _require_int(summary, "timeline_end_ns")
    if timeline_start_ns < 0 or timeline_end_ns < timeline_start_ns:
        raise RecordingMaterializationError("export timeline bounds are invalid")
    return MaterializedRecording(
        session_id=session_id,
        path=recording_path,
        media_type=RERUN_RECORDING_MEDIA_TYPE,
        byte_length=recording_stat.st_size,
        sha256=recording_sha256,
        source_sha256=exported_source_sha256,
        timeline=RERUN_SESSION_TIMELINE,
        timeline_start_ns=timeline_start_ns,
        timeline_end_ns=timeline_end_ns,
    )


def _cache_document(
    recording: MaterializedRecording,
    source: _ValidatedSource,
    recording_stat: os.stat_result,
) -> dict[str, object]:
    return {
        "schema_version": CACHE_SCHEMA,
        "session_id": recording.session_id,
        "plugin_id": source.plugin_id,
        "primary_artifact_id": source.primary_artifact_id,
        "source_sha256": recording.source_sha256,
        "source_artifacts": [
            {
                "artifact_id": artifact.artifact_id,
                "media_type": artifact.media_type,
                "file_byte_length": artifact.file_stat.st_size,
                "replay_byte_length": artifact.replay_byte_length,
                "mtime_ns": artifact.file_stat.st_mtime_ns,
                "ctime_ns": artifact.file_stat.st_ctime_ns,
                "sha256": _sha256_prefix_stable(
                    artifact.path,
                    artifact.file_stat,
                    artifact.replay_byte_length,
                ),
            }
            for artifact in source.artifacts
        ],
        "recording_byte_length": recording.byte_length,
        "recording_mtime_ns": recording_stat.st_mtime_ns,
        "recording_sha256": recording.sha256,
        "timeline": recording.timeline,
        "timeline_start_ns": recording.timeline_start_ns,
        "timeline_end_ns": recording.timeline_end_ns,
    }


def _validate_cache_document(value: object, session_id: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError("recording cache sidecar is not an object")
    expected_keys = {
        "schema_version",
        "session_id",
        "plugin_id",
        "primary_artifact_id",
        "source_sha256",
        "source_artifacts",
        "recording_byte_length",
        "recording_mtime_ns",
        "recording_sha256",
        "timeline",
        "timeline_start_ns",
        "timeline_end_ns",
    }
    if set(value) != expected_keys:
        raise ValueError("recording cache sidecar has an unsupported shape")
    if value["schema_version"] not in COMPATIBLE_CACHE_SCHEMAS or value["session_id"] != session_id:
        raise ValueError("recording cache sidecar identity does not match")
    if value["timeline"] != RERUN_SESSION_TIMELINE:
        raise ValueError("recording cache timeline is unsupported")
    for key in ("plugin_id", "primary_artifact_id"):
        if not isinstance(value[key], str) or SESSION_ID_PATTERN.fullmatch(value[key]) is None:
            raise ValueError("recording cache contains an invalid identifier")
    for key in (
        "recording_byte_length",
        "recording_mtime_ns",
        "timeline_start_ns",
        "timeline_end_ns",
    ):
        if not isinstance(value[key], int) or isinstance(value[key], bool) or value[key] < 0:
            raise ValueError("recording cache contains an invalid integer")
    for key in ("source_sha256", "recording_sha256"):
        digest = value[key]
        if not isinstance(digest, str) or not _is_sha256(digest):
            raise ValueError("recording cache contains an invalid digest")
    source_artifacts = value["source_artifacts"]
    if not isinstance(source_artifacts, list) or not source_artifacts:
        raise ValueError("recording cache source artifacts are invalid")
    artifact_ids: set[str] = set()
    artifact_keys = {
        "artifact_id",
        "media_type",
        "file_byte_length",
        "replay_byte_length",
        "mtime_ns",
        "ctime_ns",
        "sha256",
    }
    for artifact in source_artifacts:
        if not isinstance(artifact, dict) or set(artifact) != artifact_keys:
            raise ValueError("recording cache source artifact is invalid")
        artifact_id = artifact["artifact_id"]
        if (
            not isinstance(artifact_id, str)
            or SESSION_ID_PATTERN.fullmatch(artifact_id) is None
            or artifact_id in artifact_ids
        ):
            raise ValueError("recording cache source artifact id is invalid")
        artifact_ids.add(artifact_id)
        if not isinstance(artifact["media_type"], str) or not artifact["media_type"]:
            raise ValueError("recording cache source media type is invalid")
        for key in (
            "file_byte_length",
            "replay_byte_length",
            "mtime_ns",
            "ctime_ns",
        ):
            item = artifact[key]
            if not isinstance(item, int) or isinstance(item, bool) or item < 0:
                raise ValueError("recording cache source artifact boundary is invalid")
        if not 1 <= artifact["replay_byte_length"] <= artifact["file_byte_length"]:
            raise ValueError("recording cache source replay boundary is invalid")
        if not isinstance(artifact["sha256"], str) or not _is_sha256(artifact["sha256"]):
            raise ValueError("recording cache source digest is invalid")
    if value["primary_artifact_id"] not in artifact_ids:
        raise ValueError("recording cache primary artifact is unavailable")
    if value["timeline_end_ns"] < value["timeline_start_ns"]:
        raise ValueError("recording cache timeline bounds are invalid")
    return cast(dict[str, Any], value)


def _regular_file_stat(path: Path, label: str) -> os.stat_result:
    try:
        stat_result = path.stat()
    except OSError as exc:
        raise RecordingMaterializationError(f"{label} is missing") from exc
    if not path.is_file():
        raise RecordingMaterializationError(f"{label} is not a regular file")
    return stat_result


def _regular_file_stat_nofollow(path: Path, label: str) -> os.stat_result:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags)
    except OSError as exc:
        raise RecordingMaterializationError(f"{label} is missing or unsafe") from exc
    try:
        result = os.fstat(descriptor)
        if not stat.S_ISREG(result.st_mode):
            raise RecordingMaterializationError(f"{label} is not a regular file")
        current = os.lstat(path)
        if stat.S_ISLNK(current.st_mode) or (current.st_dev, current.st_ino) != (
            result.st_dev,
            result.st_ino,
        ):
            raise RecordingMaterializationError(f"{label} changed during no-follow open")
        return result
    finally:
        os.close(descriptor)


def _stat_identity(value: os.stat_result) -> tuple[int, int, int, int, int]:
    return (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns)


def _sha256_stable(path: Path, expected_stat: os.stat_result) -> str:
    return _sha256_prefix_stable(path, expected_stat, expected_stat.st_size)


def _sha256_prefix_stable(
    path: Path,
    expected_stat: os.stat_result,
    byte_length: int,
) -> str:
    if not 0 <= byte_length <= expected_stat.st_size:
        raise RecordingMaterializationError("recording digest boundary is invalid")
    return _sha256_prefix_cached(
        str(path),
        _stat_identity(expected_stat),
        byte_length,
    )


@lru_cache(maxsize=128)
def _sha256_prefix_cached(
    path_text: str,
    expected_identity: tuple[int, int, int, int, int],
    byte_length: int,
) -> str:
    path = Path(path_text)
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags)
    except OSError as exc:
        raise RecordingMaterializationError(
            "recording artifact could not be opened without following links"
        ) from exc
    try:
        before = os.fstat(descriptor)
        if not stat.S_ISREG(before.st_mode) or _stat_identity(before) != expected_identity:
            raise RecordingMaterializationError("recording artifact changed before validation")
        digest = hashlib.sha256()
        remaining = byte_length
        while remaining:
            chunk = os.read(descriptor, min(1024 * 1024, remaining))
            if not chunk:
                raise RecordingMaterializationError("recording artifact is truncated")
            digest.update(chunk)
            remaining -= len(chunk)
        after = os.fstat(descriptor)
        current = os.lstat(path)
        if (
            _stat_identity(after) != expected_identity
            or stat.S_ISLNK(current.st_mode)
            or (current.st_dev, current.st_ino) != (after.st_dev, after.st_ino)
        ):
            raise RecordingMaterializationError("recording artifact changed during validation")
        return digest.hexdigest()
    except OSError as exc:
        raise RecordingMaterializationError("recording artifact could not be validated") from exc
    finally:
        os.close(descriptor)


def _stage_replay_prefix(
    session_cache_root: Path,
    source: _ValidatedSource,
    *,
    cancel_event: threading.Event | None = None,
    activity_callback: Callable[[], None] | None = None,
) -> tuple[Path, Path, dict[str, Path]]:
    staged_root = session_cache_root / f".source.{uuid4().hex}.tmp"
    try:
        staged_root.mkdir(mode=0o700)
        staged_primary: Path | None = None
        staged_artifacts: dict[str, Path] = {}
        for artifact in source.artifacts:
            staged = staged_root / artifact.path.name
            _copy_prefix_nofollow(
                artifact.path,
                artifact.file_stat,
                staged,
                artifact.replay_byte_length,
                cancel_event=cancel_event,
                activity_callback=activity_callback,
            )
            staged_artifacts[artifact.artifact_id] = staged
            if artifact.artifact_id == source.primary_artifact_id:
                staged_primary = staged
        if staged_primary is None:
            raise RecordingMaterializationError("staged recording has no primary artifact")
        _fsync_directory(staged_root)
        return staged_root, staged_primary, staged_artifacts
    except BaseException:
        shutil.rmtree(staged_root, ignore_errors=True)
        raise


def _copy_prefix_nofollow(
    source: Path,
    expected_stat: os.stat_result,
    destination: Path,
    byte_length: int,
    *,
    cancel_event: threading.Event | None = None,
    activity_callback: Callable[[], None] | None = None,
) -> None:
    source_flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    source_descriptor = os.open(source, source_flags)
    destination_descriptor = os.open(
        destination,
        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
        0o600,
    )
    try:
        before = os.fstat(source_descriptor)
        if not stat.S_ISREG(before.st_mode) or _stat_identity(before) != _stat_identity(
            expected_stat
        ):
            raise RecordingMaterializationError("native capture changed before prefix copy")
        remaining = byte_length
        while remaining:
            _raise_if_cancelled(cancel_event)
            if activity_callback is not None:
                activity_callback()
            chunk = os.read(source_descriptor, min(1024 * 1024, remaining))
            if not chunk:
                raise RecordingMaterializationError("native capture prefix is truncated")
            view = memoryview(chunk)
            while view:
                written = os.write(destination_descriptor, view)
                view = view[written:]
            remaining -= len(chunk)
        os.fsync(destination_descriptor)
        after = os.fstat(source_descriptor)
        if _stat_identity(after) != _stat_identity(expected_stat):
            raise RecordingMaterializationError("native capture changed during prefix copy")
    finally:
        os.close(source_descriptor)
        os.close(destination_descriptor)


def _cache_entries(root: Path) -> tuple[int, list[tuple[Path, int, int]]]:
    total = 0
    entries: list[tuple[Path, int, int]] = []
    try:
        children = tuple(root.iterdir())
    except OSError as exc:
        raise RecordingMaterializationError("recording cache could not be inspected") from exc
    for child in children:
        if child.name == ".export.lock":
            continue
        try:
            child_stat = child.lstat()
        except OSError:
            continue
        byte_length = _cache_path_bytes(child)
        total += byte_length
        entries.append((child, byte_length, child_stat.st_mtime_ns))
    entries.sort(key=lambda item: (item[2], item[0].name))
    return total, entries


def _cache_path_bytes(path: Path) -> int:
    try:
        path_stat = path.lstat()
    except OSError:
        return 0
    if stat.S_ISLNK(path_stat.st_mode) or stat.S_ISREG(path_stat.st_mode):
        return path_stat.st_size
    if not stat.S_ISDIR(path_stat.st_mode):
        return 0
    total = path_stat.st_size
    try:
        children = tuple(path.iterdir())
    except OSError:
        return total
    return total + sum(_cache_path_bytes(child) for child in children)


def _remove_cache_file(path: Path) -> None:
    try:
        path_stat = path.lstat()
    except FileNotFoundError:
        return
    except OSError as exc:
        raise RecordingMaterializationError("recording cache entry could not be inspected") from exc
    if stat.S_ISDIR(path_stat.st_mode):
        raise RecordingMaterializationError("recording cache file was replaced by a directory")
    try:
        path.unlink()
    except OSError as exc:
        raise RecordingMaterializationError("recording cache entry could not be removed") from exc


def _remove_cache_session(root: Path, path: Path) -> None:
    if path.parent != root or path.name in {"", ".", ".."}:
        raise RecordingMaterializationError("cache eviction target escapes recording root")
    try:
        path_stat = path.lstat()
        if stat.S_ISLNK(path_stat.st_mode) or not stat.S_ISDIR(path_stat.st_mode):
            path.unlink()
        else:
            shutil.rmtree(path)
        _fsync_directory(root)
    except FileNotFoundError:
        return
    except OSError as exc:
        raise RecordingMaterializationError("derived cache eviction failed") from exc


def _touch_lru(path: Path) -> None:
    try:
        os.utime(path, None, follow_symlinks=False)
    except OSError:
        return


def _optional_positive_configuration(
    configured: int | None,
    *,
    environment_name: str,
) -> int | None:
    value = configured
    if value is None:
        raw = os.environ.get(environment_name, "").strip()
        value = int(raw) if raw else None
    if value is None:
        return None
    if isinstance(value, bool) or value <= 0:
        raise ValueError(f"{environment_name} must be a positive integer")
    return value


def _non_negative_configuration(
    configured: int | None,
    *,
    environment_name: str,
    default: int,
) -> int:
    value = configured
    if value is None:
        raw = os.environ.get(environment_name, "").strip()
        value = int(raw) if raw else default
    if isinstance(value, bool) or value < 0:
        raise ValueError(f"{environment_name} must be a non-negative integer")
    return value


def _callable_accepts_keyword(callback: Callable[..., object], keyword: str) -> bool:
    try:
        parameters = inspect.signature(callback).parameters
    except (TypeError, ValueError):
        return False
    return keyword in parameters or any(
        parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters.values()
    )


@contextmanager
def _exclusive_file_lock(path: Path) -> Any:
    """Serialize exporters across Mission Core processes on POSIX hosts."""

    try:
        import fcntl
    except ImportError as exc:  # pragma: no cover - production targets are POSIX
        raise RecordingMaterializationError("cross-process recording lock is unavailable") from exc

    flags = (
        os.O_RDWR
        | os.O_CREAT
        | getattr(os, "O_CLOEXEC", 0)
        | getattr(
            os,
            "O_NOFOLLOW",
            0,
        )
    )
    try:
        descriptor = os.open(path, flags, 0o600)
    except OSError as exc:
        raise RecordingMaterializationError(
            "cross-process recording lock could not be opened"
        ) from exc
    try:
        lock_stat = os.fstat(descriptor)
        if not stat.S_ISREG(lock_stat.st_mode):
            raise RecordingMaterializationError(
                "cross-process recording lock is not a regular file"
            )
        fcntl.flock(descriptor, fcntl.LOCK_EX)
        yield
    except OSError as exc:
        raise RecordingMaterializationError("cross-process recording lock failed") from exc
    finally:
        try:
            fcntl.flock(descriptor, fcntl.LOCK_UN)
        finally:
            os.close(descriptor)


def _require_int(summary: Mapping[str, object], key: str) -> int:
    value = summary.get(key)
    if not isinstance(value, int) or isinstance(value, bool):
        raise RecordingMaterializationError(f"RRD export summary has no valid {key}")
    return value


def _require_sha256(summary: Mapping[str, object], key: str) -> str:
    value = summary.get(key)
    if not isinstance(value, str) or not _is_sha256(value):
        raise RecordingMaterializationError(f"RRD export summary has no valid {key}")
    return value


def _is_sha256(value: str) -> bool:
    return len(value) == 64 and all(character in "0123456789abcdef" for character in value)


def _write_json_atomic(path: Path, document: Mapping[str, object]) -> None:
    temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
    payload = (json.dumps(document, sort_keys=True, separators=(",", ":")) + "\n").encode()
    try:
        descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        try:
            with os.fdopen(descriptor, "wb") as stream:
                stream.write(payload)
                stream.flush()
                os.fsync(stream.fileno())
        except BaseException:
            # fdopen owns the descriptor after successful construction.
            raise
        os.replace(temporary, path)
        _fsync_directory(path.parent)
    except OSError as exc:
        raise RecordingMaterializationError(
            "recording cache sidecar could not be published"
        ) from exc
    finally:
        temporary.unlink(missing_ok=True)


def _fsync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _chmod_best_effort(path: Path, mode: int) -> None:
    try:
        if path.stat().st_mode & 0o777 != mode:
            path.chmod(mode)
    except OSError:
        return