"""Crash-safe export of completed dual-OAK capture datasets.

The capture directory remains the authoritative dataset.  This module only
reconciles immutable, completed manifests into the existing ``camera.image``
DataHub outbox; it never performs network I/O and never changes capture status.
"""

from __future__ import annotations

import hashlib
import json
import math
import os
import re
import stat
import threading
import time
import uuid
from collections import OrderedDict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple

from .capture_runs import (
    CAPTURE_IMAGE_COUNT,
    CAPTURE_INTERVAL_SECONDS,
    CAPTURE_SCHEMA,
    CAPTURE_TIMEPOINTS,
    DEFAULT_MAX_PAIR_SKEW_SECONDS,
    MAX_JPEG_BYTES,
    MAX_MANIFEST_BYTES,
    REQUIRED_ROLES,
    RUN_DURATION_SECONDS,
)
from .integrations.datahub import DataHubPublisher

EXPORT_STATUS_SCHEMA_VERSION = 1
FRAME_EVENT_NAMESPACE = uuid.UUID("f48335bb-e8c5-54cb-93db-cea7e274d9e1")
PAIR_NAMESPACE = uuid.UUID("1a65dedf-d70a-5a44-b289-377d9be91a17")
MIN_CAPTURE_SPACING_SECONDS = 0.5
MAX_CAPTURE_SPACING_SECONDS = 1.5

_RUN_ID = re.compile(r"^[0-9a-f]{32}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_SOURCE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,255}$")
_SAFE_CODE = re.compile(r"[^a-z0-9_]+")


class CaptureExportError(RuntimeError):
    """A bounded, machine-readable export failure."""

    def __init__(self, code: str, message: str = "") -> None:
        normalized = _SAFE_CODE.sub("_", str(code).lower()).strip("_")
        self.code = (normalized or "capture_export_failed")[:64]
        self.safe_message = str(message).replace("\n", " ")[:160]
        super().__init__(self.safe_message or self.code)


class _CaptureExportPaused(RuntimeError):
    """Internal control flow used to yield CPU and storage during motion."""


@dataclass(frozen=True)
class CaptureExportResult:
    """Result of reconciling one completed capture manifest."""

    run_id: str
    frames: int
    enqueued: int
    pending: int
    acknowledged: int
    dead_lettered: int
    event_ids: Tuple[str, ...]


@dataclass(frozen=True)
class CaptureReconcileResult:
    """Bounded result of one dataset-root reconciliation pass."""

    scanned: int
    completed: int
    skipped: int
    failed: int
    enqueued: int
    pending: int
    acknowledged: int
    dead_lettered: int
    failures: Tuple[Tuple[str, str], ...]


@dataclass(frozen=True)
class _FrameSpec:
    index: int
    scheduled_offset_seconds: int
    role: str
    source_id: str
    sequence: int
    captured_at_s: float
    filename: str
    byte_length: int
    sha256: str


class CaptureDatasetExporter:
    """Reconcile completed manifests into deterministic image events."""

    def __init__(
        self,
        *,
        dataset_root: Path,
        publisher: DataHubPublisher,
        max_manifest_bytes: int = MAX_MANIFEST_BYTES,
        max_jpeg_bytes: Optional[int] = None,
    ) -> None:
        self.dataset_root = Path(dataset_root)
        self.publisher = publisher
        self.max_manifest_bytes = int(max_manifest_bytes)
        self.max_jpeg_bytes = int(
            publisher.max_image_bytes
            if max_jpeg_bytes is None
            else max_jpeg_bytes
        )
        if not self.dataset_root.is_absolute():
            raise ValueError("capture dataset root must be absolute")
        if not 1024 <= self.max_manifest_bytes <= MAX_MANIFEST_BYTES:
            raise ValueError("capture manifest size limit is invalid")
        if (
            not 1024
            <= self.max_jpeg_bytes
            <= min(MAX_JPEG_BYTES, publisher.max_image_bytes)
        ):
            raise ValueError("capture JPEG size limit is invalid")
        self._status_lock = threading.Lock()
        self._worker_active = False
        self._last_scan_at = 0.0
        self._last_result: Optional[CaptureReconcileResult] = None
        self._last_error_code = ""
        self._run_results: "OrderedDict[str, CaptureExportResult]" = (
            OrderedDict()
        )
        descriptor = self._open_root()
        os.close(descriptor)

    def export_completed_run(self, run_id: str) -> CaptureExportResult:
        """Verify and queue every unrecorded frame from one completed run.

        Delivery is intentionally left to :class:`DataHubPublisher`'s worker.
        An offline server therefore cannot change a successful local capture.
        """

        run_id = self._validate_run_id(run_id)
        manifest, fingerprint = self._read_manifest(run_id)
        result = self._export_manifest(run_id, manifest, fingerprint)
        self._remember_run_result(result)
        return result

    def reconcile_completed_runs(
        self,
        *,
        limit: int = 1000,
        pause_predicate: Optional[Callable[[], bool]] = None,
    ) -> CaptureReconcileResult:
        """Scan the dataset root and queue only completed manifests.

        A malformed run is reported and isolated so another completed run can
        still be reconciled during the same pass.
        """

        if (
            isinstance(limit, bool)
            or not isinstance(limit, int)
            or not 1 <= limit <= 10000
        ):
            raise ValueError(
                "capture reconcile limit must be between 1 and 10000"
            )
        if pause_predicate is not None and not callable(pause_predicate):
            raise ValueError("pause_predicate must be callable")
        if _pause_requested(pause_predicate):
            result = CaptureReconcileResult(
                scanned=0,
                completed=0,
                skipped=0,
                failed=0,
                enqueued=0,
                pending=0,
                acknowledged=0,
                dead_lettered=0,
                failures=(),
            )
            with self._status_lock:
                self._last_scan_at = time.time()
                self._last_result = result
            return result
        run_ids = self._scan_run_ids(limit=limit)
        scanned = 0
        completed = 0
        skipped = 0
        failures: List[Tuple[str, str]] = []
        enqueued = 0
        pending = 0
        acknowledged = 0
        dead_lettered = 0
        for run_id in run_ids:
            if _pause_requested(pause_predicate):
                break
            scanned += 1
            try:
                manifest, fingerprint = self._read_manifest(run_id)
                if manifest.get("status") != "completed":
                    skipped += 1
                    continue
                result = self._export_manifest(
                    run_id,
                    manifest,
                    fingerprint,
                    pause_predicate=pause_predicate,
                )
                self._remember_run_result(result)
                completed += 1
                enqueued += result.enqueued
                pending += result.pending
                acknowledged += result.acknowledged
                dead_lettered += result.dead_lettered
            except _CaptureExportPaused:
                break
            except CaptureExportError as exc:
                failures.append((run_id, exc.code))
            except Exception:
                failures.append((run_id, "capture_export_failed"))
        result = CaptureReconcileResult(
            scanned=scanned,
            completed=completed,
            skipped=skipped,
            failed=len(failures),
            enqueued=enqueued,
            pending=pending,
            acknowledged=acknowledged,
            dead_lettered=dead_lettered,
            failures=tuple(failures),
        )
        with self._status_lock:
            self._last_scan_at = time.time()
            self._last_result = result
            self._last_error_code = failures[0][1] if failures else ""
        return result

    def run_worker(
        self,
        stop_event: threading.Event,
        *,
        interval_seconds: float = 5.0,
        scan_limit: int = 1000,
        pause_predicate: Optional[Callable[[], bool]] = None,
        operation_lock: Any = None,
    ) -> None:
        """Reconcile immediately and periodically until stopped."""

        interval_seconds = float(interval_seconds)
        if not math.isfinite(interval_seconds) or interval_seconds < 0.1:
            raise ValueError(
                "capture export interval must be at least 0.1 seconds"
            )
        if pause_predicate is not None and not callable(pause_predicate):
            raise ValueError("pause_predicate must be callable")
        if operation_lock is not None and not all(
            callable(getattr(operation_lock, name, None))
            for name in ("acquire", "release")
        ):
            raise ValueError("operation_lock must provide acquire/release")
        with self._status_lock:
            self._worker_active = True
        try:
            while not stop_event.is_set():
                if pause_predicate is not None:
                    try:
                        paused = pause_predicate()
                    except Exception:
                        self._set_worker_error("pause_predicate_error")
                        stop_event.wait(0.25)
                        continue
                    if not isinstance(paused, bool):
                        self._set_worker_error("pause_predicate_invalid")
                        stop_event.wait(0.25)
                        continue
                    if paused:
                        stop_event.wait(0.25)
                        continue
                leased = False
                if operation_lock is not None:
                    try:
                        leased = operation_lock.acquire(blocking=False)
                    except Exception:
                        self._set_worker_error("operation_lock_error")
                        stop_event.wait(0.25)
                        continue
                    if type(leased) is not bool:
                        self._set_worker_error("operation_lock_invalid")
                        stop_event.wait(0.25)
                        continue
                    if not leased:
                        stop_event.wait(0.25)
                        continue
                try:
                    if operation_lock is not None or pause_predicate is None:
                        self.reconcile_completed_runs(limit=scan_limit)
                    else:
                        self.reconcile_completed_runs(
                            limit=scan_limit,
                            pause_predicate=pause_predicate,
                        )
                except Exception as exc:
                    code = getattr(exc, "code", "capture_export_failed")
                    normalized = _SAFE_CODE.sub("_", str(code).lower()).strip(
                        "_"
                    )
                    self._set_worker_error(
                        normalized or "capture_export_failed"
                    )
                finally:
                    if leased:
                        operation_lock.release()
                stop_event.wait(interval_seconds)
        finally:
            with self._status_lock:
                self._worker_active = False

    def status(self, run_id: str = "") -> Mapping[str, Any]:
        """Return a bounded, path- and credential-free worker status."""

        with self._status_lock:
            result = self._last_result
            requested_run_id = str(run_id or "")
            if requested_run_id:
                run_result = self._run_results.get(requested_run_id)
            else:
                run_result = (
                    next(reversed(self._run_results.values()))
                    if self._run_results
                    else None
                )
            snapshot = {
                "schema_version": EXPORT_STATUS_SCHEMA_VERSION,
                "configured": True,
                "worker_active": self._worker_active,
                "last_scan_at": self._last_scan_at,
                "initial_scan_complete": result is not None,
                "scanned_runs": 0 if result is None else result.scanned,
                "completed_runs": 0 if result is None else result.completed,
                "skipped_runs": 0 if result is None else result.skipped,
                "failed_runs": 0 if result is None else result.failed,
                "aggregate_pending_frames": (
                    0 if result is None else result.pending
                ),
                "aggregate_acknowledged_frames": (
                    0 if result is None else result.acknowledged
                ),
                "aggregate_dead_lettered_frames": (
                    0 if result is None else result.dead_lettered
                ),
                "error_code": self._last_error_code,
            }
        lookup_failed = False
        try:
            states = (
                {}
                if run_result is None
                else self.publisher.outbox.event_states(run_result.event_ids)
            )
        except Exception:
            # `/healthz` must remain available even if its receipt lookup is
            # temporarily unavailable.  Fail closed without exposing SQLite
            # paths or exception text.
            states = {}
            lookup_failed = True
        pending = sum(value == "pending" for value in states.values())
        acknowledged = sum(
            value == "acknowledged" for value in states.values()
        )
        dead_lettered = sum(
            value == "dead_lettered" for value in states.values()
        )
        known = len(states)
        complete = bool(
            run_result is not None
            and not lookup_failed
            and known == CAPTURE_IMAGE_COUNT
            and acknowledged == CAPTURE_IMAGE_COUNT
            and pending == 0
            and dead_lettered == 0
        )
        snapshot.update(
            {
                "run_known": run_result is not None,
                "run_id": "" if run_result is None else run_result.run_id,
                "total_frames": CAPTURE_IMAGE_COUNT,
                "known_frames": known,
                "unqueued_frames": max(0, CAPTURE_IMAGE_COUNT - known),
                "enqueued_frames": (
                    0 if run_result is None else run_result.enqueued
                ),
                "pending_frames": pending,
                "acknowledged_frames": acknowledged,
                "dead_lettered_frames": dead_lettered,
                "complete": complete,
                "failed": lookup_failed or dead_lettered > 0,
            }
        )
        if lookup_failed:
            snapshot["error_code"] = "outbox_status_unavailable"
        return snapshot

    def _remember_run_result(self, result: CaptureExportResult) -> None:
        with self._status_lock:
            self._run_results.pop(result.run_id, None)
            self._run_results[result.run_id] = result
            while len(self._run_results) > 128:
                self._run_results.popitem(last=False)

    def _set_worker_error(self, code: str) -> None:
        with self._status_lock:
            self._last_scan_at = time.time()
            self._last_error_code = str(code)[:64]

    def _export_manifest(
        self,
        run_id: str,
        manifest: Mapping[str, Any],
        fingerprint: str,
        *,
        pause_predicate: Optional[Callable[[], bool]] = None,
    ) -> CaptureExportResult:
        if _pause_requested(pause_predicate):
            raise _CaptureExportPaused()
        specs, profile_sha256 = self._validate_completed_manifest(
            run_id, manifest
        )
        frames_fd = self._open_frames(run_id)
        try:
            actual_names = set(os.listdir(frames_fd))
            expected_names = {spec.filename for spec in specs}
            if actual_names != expected_names:
                raise CaptureExportError(
                    "capture_frames_mismatch",
                    "Capture frame directory does not match its manifest",
                )

            enqueued = 0
            pending = 0
            acknowledged = 0
            dead_lettered = 0
            event_ids = []
            for spec in specs:
                if _pause_requested(pause_predicate):
                    raise _CaptureExportPaused()
                event_id = frame_event_id(
                    device_uid=self.publisher.uid_device,
                    run_id=run_id,
                    index=spec.index,
                    role=spec.role,
                    jpeg_sha256=spec.sha256,
                )
                event_ids.append(event_id)
                state = self.publisher.outbox.event_state(event_id)
                if state == "acknowledged":
                    acknowledged += 1
                    continue
                if state == "pending":
                    pending += 1
                    continue
                if state == "dead_lettered":
                    dead_lettered += 1
                    continue

                jpeg = self._read_frame(
                    frames_fd,
                    spec,
                    pause_predicate=pause_predicate,
                )
                if _pause_requested(pause_predicate):
                    raise _CaptureExportPaused()
                observed_at = _datetime_from_timestamp(spec.captured_at_s)
                pair_id = capture_pair_id(
                    device_uid=self.publisher.uid_device,
                    run_id=run_id,
                    index=spec.index,
                )
                self.publisher.enqueue_image(
                    jpeg,
                    camera_id=spec.source_id,
                    content_type="image/jpeg",
                    observed_at=observed_at,
                    event_id=event_id,
                    attributes={
                        "capture_schema": CAPTURE_SCHEMA,
                        "run_id": run_id,
                        "timepoint_index": spec.index,
                        "pair_id": pair_id,
                        "role": spec.role,
                        "source_mxid": spec.source_id,
                        "sequence": spec.sequence,
                        "scheduled_offset_seconds": (
                            spec.scheduled_offset_seconds
                        ),
                        "motion_profile_sha256": profile_sha256,
                    },
                )
                enqueued += 1
                pending += 1
        finally:
            os.close(frames_fd)

        if _pause_requested(pause_predicate):
            raise _CaptureExportPaused()
        _manifest, final_fingerprint = self._read_manifest(run_id)
        if final_fingerprint != fingerprint:
            raise CaptureExportError(
                "capture_manifest_changed",
                "Capture manifest changed during export",
            )
        return CaptureExportResult(
            run_id=run_id,
            frames=CAPTURE_IMAGE_COUNT,
            enqueued=enqueued,
            pending=pending,
            acknowledged=acknowledged,
            dead_lettered=dead_lettered,
            event_ids=tuple(event_ids),
        )

    def _validate_completed_manifest(
        self,
        run_id: str,
        manifest: Mapping[str, Any],
    ) -> Tuple[Tuple[_FrameSpec, ...], str]:
        if manifest.get("schema") != CAPTURE_SCHEMA:
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Capture manifest schema is invalid",
            )
        if manifest.get("run_id") != run_id:
            raise CaptureExportError(
                "capture_manifest_identity",
                "Capture manifest identity is invalid",
            )
        if manifest.get("status") != "completed":
            raise CaptureExportError(
                "capture_not_completed",
                "Only completed captures may be exported",
            )
        error = manifest.get("error")
        if not isinstance(error, Mapping) or error.get("code") not in (
            "",
            None,
        ):
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Completed capture contains an error",
            )
        requested = manifest.get("requested")
        if not isinstance(requested, Mapping):
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Capture request metadata is invalid",
            )
        if (
            not _same_number(
                requested.get("duration_seconds"), RUN_DURATION_SECONDS
            )
            or not _same_number(
                requested.get("interval_seconds"), CAPTURE_INTERVAL_SECONDS
            )
            or requested.get("timepoints") != CAPTURE_TIMEPOINTS
            or requested.get("images") != CAPTURE_IMAGE_COUNT
            or requested.get("roles") != list(REQUIRED_ROLES)
        ):
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Capture request invariants are invalid",
            )
        motion = manifest.get("motion")
        if not isinstance(motion, Mapping):
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Capture motion metadata is invalid",
            )
        profile_sha256 = motion.get("profile_sha256")
        if (
            motion.get("required") is not True
            or motion.get("executed") is not True
            or not isinstance(profile_sha256, str)
            or _SHA256.fullmatch(profile_sha256) is None
            or isinstance(motion.get("command_count"), bool)
            or not isinstance(motion.get("command_count"), int)
            or motion.get("command_count") < 1
            or not _same_number(
                motion.get("duration_seconds"),
                RUN_DURATION_SECONDS,
                tolerance=0.1,
            )
        ):
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Capture motion evidence is invalid",
            )

        samples = manifest.get("samples")
        if not isinstance(samples, list) or len(samples) != CAPTURE_TIMEPOINTS:
            raise CaptureExportError(
                "capture_manifest_invalid", "Capture sample count is invalid"
            )
        specs: List[_FrameSpec] = []
        last_sequences: Dict[str, int] = {}
        last_captured_times: Dict[str, float] = {}
        last_received_times: Dict[str, float] = {}
        stable_sources: Dict[str, str] = {}
        for expected_index, sample in enumerate(samples):
            if not isinstance(sample, Mapping):
                raise CaptureExportError(
                    "capture_manifest_invalid", "Capture sample is invalid"
                )
            if (
                sample.get("index") != expected_index
                or sample.get("scheduled_offset_seconds") != expected_index
            ):
                raise CaptureExportError(
                    "capture_manifest_invalid",
                    "Capture sample schedule is invalid",
                )
            pair_skew = _finite_number(sample.get("pair_skew_seconds"))
            if not 0.0 <= pair_skew <= DEFAULT_MAX_PAIR_SKEW_SECONDS:
                raise CaptureExportError(
                    "capture_manifest_invalid",
                    "Capture pair skew exceeds its commissioned limit",
                )
            frames = sample.get("frames")
            if not isinstance(frames, list) or len(frames) != len(
                REQUIRED_ROLES
            ):
                raise CaptureExportError(
                    "capture_manifest_invalid", "Capture pair is incomplete"
                )
            by_role: Dict[str, Mapping[str, Any]] = {}
            for frame in frames:
                if not isinstance(frame, Mapping):
                    raise CaptureExportError(
                        "capture_manifest_invalid", "Capture frame is invalid"
                    )
                role = frame.get("role")
                if role not in REQUIRED_ROLES or role in by_role:
                    raise CaptureExportError(
                        "capture_manifest_invalid", "Capture roles are invalid"
                    )
                by_role[str(role)] = frame
            if set(by_role) != set(REQUIRED_ROLES):
                raise CaptureExportError(
                    "capture_manifest_invalid", "Capture pair is incomplete"
                )

            captured_times: Dict[str, float] = {}
            for role in REQUIRED_ROLES:
                frame = by_role[role]
                source_id = frame.get("source_id")
                sequence = frame.get("sequence")
                captured_at_s = _finite_number(frame.get("captured_at_s"))
                received_at_s = _finite_number(
                    frame.get("received_monotonic_s")
                )
                byte_length = frame.get("byte_length")
                jpeg_sha256 = frame.get("sha256")
                expected_path = "frames/{:03d}-{}.jpg".format(
                    expected_index, role
                )
                if (
                    not isinstance(source_id, str)
                    or _SOURCE_ID.fullmatch(source_id) is None
                    or isinstance(sequence, bool)
                    or not isinstance(sequence, int)
                    or sequence < 1
                    or captured_at_s <= 0.0
                    or received_at_s < 0.0
                    or isinstance(byte_length, bool)
                    or not isinstance(byte_length, int)
                    or not 4 <= byte_length <= self.max_jpeg_bytes
                    or not isinstance(jpeg_sha256, str)
                    or _SHA256.fullmatch(jpeg_sha256) is None
                    or frame.get("path") != expected_path
                    or frame.get("content_type") != "image/jpeg"
                ):
                    raise CaptureExportError(
                        "capture_manifest_invalid",
                        "Capture frame metadata is invalid",
                    )
                if (
                    role in stable_sources
                    and stable_sources[role] != source_id
                ):
                    raise CaptureExportError(
                        "capture_camera_identity_changed",
                        "Capture camera identity changed during the run",
                    )
                if sequence <= last_sequences.get(role, 0):
                    raise CaptureExportError(
                        "capture_sequence_invalid",
                        "Capture frame sequence is not strictly increasing",
                    )
                prior_capture = last_captured_times.get(role)
                if (
                    prior_capture is not None
                    and captured_at_s <= prior_capture
                ):
                    raise CaptureExportError(
                        "capture_timestamp_invalid",
                        "Capture timestamps are not strictly increasing",
                    )
                prior_received = last_received_times.get(role)
                if prior_received is not None:
                    spacing = received_at_s - prior_received
                    if not (
                        MIN_CAPTURE_SPACING_SECONDS
                        <= spacing
                        <= MAX_CAPTURE_SPACING_SECONDS
                    ):
                        raise CaptureExportError(
                            "capture_timestamp_invalid",
                            "Capture timestamps do not follow the "
                            "one-second schedule",
                        )
                stable_sources.setdefault(role, source_id)
                last_sequences[role] = sequence
                last_captured_times[role] = captured_at_s
                last_received_times[role] = received_at_s
                captured_times[role] = captured_at_s
                specs.append(
                    _FrameSpec(
                        index=expected_index,
                        scheduled_offset_seconds=expected_index,
                        role=role,
                        source_id=source_id,
                        sequence=sequence,
                        captured_at_s=captured_at_s,
                        filename="{:03d}-{}.jpg".format(expected_index, role),
                        byte_length=byte_length,
                        sha256=jpeg_sha256,
                    )
                )
            actual_skew = abs(
                captured_times["tool"] - captured_times["workspace"]
            )
            if abs(actual_skew - pair_skew) > 1e-6:
                raise CaptureExportError(
                    "capture_manifest_invalid",
                    "Capture pair skew is inconsistent",
                )
        if len(specs) != CAPTURE_IMAGE_COUNT:
            raise CaptureExportError(
                "capture_manifest_invalid", "Capture frame count is invalid"
            )
        return tuple(specs), profile_sha256

    def _read_manifest(self, run_id: str) -> Tuple[Mapping[str, Any], str]:
        run_fd = self._open_run(run_id)
        descriptor: Optional[int] = None
        try:
            descriptor = os.open(
                "manifest.json", _file_read_flags(), dir_fd=run_fd
            )
            info = os.fstat(descriptor)
            if (
                not stat.S_ISREG(info.st_mode)
                or info.st_nlink != 1
                or info.st_size < 2
                or info.st_size > self.max_manifest_bytes
            ):
                raise CaptureExportError(
                    "capture_manifest_unsafe",
                    "Capture manifest is not a safe file",
                )
            payload = _read_bounded(
                descriptor,
                expected_size=info.st_size,
                maximum=self.max_manifest_bytes,
            )
            final = os.fstat(descriptor)
            if not _same_file_version(info, final):
                raise CaptureExportError(
                    "capture_manifest_changed",
                    "Capture manifest changed while read",
                )
        except CaptureExportError:
            raise
        except OSError as exc:
            raise CaptureExportError(
                "capture_manifest_io", "Capture manifest could not be opened"
            ) from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            os.close(run_fd)
        try:
            value = json.loads(payload.decode("utf-8"))
        except (UnicodeError, ValueError, json.JSONDecodeError) as exc:
            raise CaptureExportError(
                "capture_manifest_invalid",
                "Capture manifest is not valid JSON",
            ) from exc
        if not isinstance(value, Mapping):
            raise CaptureExportError(
                "capture_manifest_invalid", "Capture manifest is not an object"
            )
        if (
            value.get("run_id") != run_id
            or value.get("schema") != CAPTURE_SCHEMA
        ):
            raise CaptureExportError(
                "capture_manifest_identity",
                "Capture manifest identity is invalid",
            )
        return dict(value), hashlib.sha256(payload).hexdigest()

    def _read_frame(
        self,
        frames_fd: int,
        spec: _FrameSpec,
        *,
        pause_predicate: Optional[Callable[[], bool]] = None,
    ) -> bytes:
        descriptor: Optional[int] = None
        try:
            descriptor = os.open(
                spec.filename, _file_read_flags(), dir_fd=frames_fd
            )
            info = os.fstat(descriptor)
            if (
                not stat.S_ISREG(info.st_mode)
                or info.st_nlink != 1
                or info.st_size != spec.byte_length
                or info.st_size > self.max_jpeg_bytes
            ):
                raise CaptureExportError(
                    "capture_frame_unsafe", "Capture frame is not a safe file"
                )
            payload = _read_bounded(
                descriptor,
                expected_size=spec.byte_length,
                maximum=self.max_jpeg_bytes,
                pause_predicate=pause_predicate,
            )
            final = os.fstat(descriptor)
            if not _same_file_version(info, final):
                raise CaptureExportError(
                    "capture_frame_changed", "Capture frame changed while read"
                )
        except CaptureExportError:
            raise
        except OSError as exc:
            raise CaptureExportError(
                "capture_frame_io", "Capture frame could not be opened"
            ) from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
        if _pause_requested(pause_predicate):
            raise _CaptureExportPaused()
        if (
            not payload.startswith(b"\xff\xd8")
            or not payload.endswith(b"\xff\xd9")
            or hashlib.sha256(payload).hexdigest() != spec.sha256
        ):
            raise CaptureExportError(
                "capture_frame_integrity",
                "Capture frame failed integrity checks",
            )
        return payload

    def _scan_run_ids(self, *, limit: int) -> Tuple[str, ...]:
        root_fd = self._open_root()
        try:
            names = sorted(os.listdir(root_fd))
            run_ids = []
            for name in names:
                if _RUN_ID.fullmatch(name) is None:
                    continue
                try:
                    info = os.stat(name, dir_fd=root_fd, follow_symlinks=False)
                except OSError as exc:
                    raise CaptureExportError(
                        "capture_dataset_io",
                        "Capture run could not be inspected",
                    ) from exc
                if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(
                    info.st_mode
                ):
                    raise CaptureExportError(
                        "capture_run_unsafe",
                        "Capture run is not a safe directory",
                    )
                run_ids.append(name)
                if len(run_ids) >= limit:
                    break
            return tuple(run_ids)
        finally:
            os.close(root_fd)

    def _open_root(self) -> int:
        try:
            descriptor = os.open(str(self.dataset_root), _directory_flags())
            info = os.fstat(descriptor)
            if not stat.S_ISDIR(info.st_mode):
                raise OSError("dataset root is not a directory")
            return descriptor
        except OSError as exc:
            raise CaptureExportError(
                "capture_dataset_unsafe", "Capture dataset root is unavailable"
            ) from exc

    def _open_run(self, run_id: str) -> int:
        run_id = self._validate_run_id(run_id)
        root_fd = self._open_root()
        try:
            return os.open(run_id, _directory_flags(), dir_fd=root_fd)
        except OSError as exc:
            raise CaptureExportError(
                "capture_run_unsafe", "Capture run directory is unavailable"
            ) from exc
        finally:
            os.close(root_fd)

    def _open_frames(self, run_id: str) -> int:
        run_fd = self._open_run(run_id)
        try:
            return os.open("frames", _directory_flags(), dir_fd=run_fd)
        except OSError as exc:
            raise CaptureExportError(
                "capture_frames_unsafe",
                "Capture frames directory is unavailable",
            ) from exc
        finally:
            os.close(run_fd)

    @staticmethod
    def _validate_run_id(run_id: str) -> str:
        run_id = str(run_id)
        if _RUN_ID.fullmatch(run_id) is None:
            raise CaptureExportError(
                "capture_run_identity", "Capture run identifier is invalid"
            )
        return run_id


def frame_event_id(
    *,
    device_uid: str,
    run_id: str,
    index: int,
    role: str,
    jpeg_sha256: str,
) -> str:
    """Return the stable UUIDv5 id for one capture frame event."""

    name = json.dumps(
        [
            "labcobot.capture-frame.v1",
            device_uid,
            run_id,
            index,
            role,
            jpeg_sha256,
        ],
        ensure_ascii=True,
        separators=(",", ":"),
    )
    return str(uuid.uuid5(FRAME_EVENT_NAMESPACE, name))


def capture_pair_id(*, device_uid: str, run_id: str, index: int) -> str:
    """Return the stable UUIDv5 shared by both images at one timepoint."""

    name = json.dumps(
        ["labcobot.capture-pair.v1", device_uid, run_id, index],
        ensure_ascii=True,
        separators=(",", ":"),
    )
    return str(uuid.uuid5(PAIR_NAMESPACE, name))


def _directory_flags() -> int:
    return (
        os.O_RDONLY
        | getattr(os, "O_CLOEXEC", 0)
        | getattr(os, "O_DIRECTORY", 0)
        | getattr(os, "O_NOFOLLOW", 0)
    )


def _file_read_flags() -> int:
    return (
        os.O_RDONLY
        | getattr(os, "O_CLOEXEC", 0)
        | getattr(os, "O_NOFOLLOW", 0)
    )


def _read_bounded(
    descriptor: int,
    *,
    expected_size: int,
    maximum: int,
    pause_predicate: Optional[Callable[[], bool]] = None,
) -> bytes:
    payload = bytearray()
    while len(payload) <= maximum:
        if _pause_requested(pause_predicate):
            raise _CaptureExportPaused()
        chunk = os.read(descriptor, min(65536, maximum + 1 - len(payload)))
        if not chunk:
            break
        payload.extend(chunk)
    if len(payload) != expected_size or len(payload) > maximum:
        raise CaptureExportError(
            "capture_file_size",
            "Capture file size changed or exceeded its limit",
        )
    return bytes(payload)


def _same_file_version(before: os.stat_result, after: os.stat_result) -> bool:
    fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
    return all(
        getattr(before, field) == getattr(after, field) for field in fields
    )


def _finite_number(value: Any) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise CaptureExportError(
            "capture_manifest_invalid", "Capture manifest number is invalid"
        )
    result = float(value)
    if not math.isfinite(result):
        raise CaptureExportError(
            "capture_manifest_invalid", "Capture manifest number is invalid"
        )
    return result


def _same_number(
    value: Any, expected: float, *, tolerance: float = 1e-9
) -> bool:
    try:
        return abs(_finite_number(value) - expected) <= tolerance
    except CaptureExportError:
        return False


def _datetime_from_timestamp(value: float) -> datetime:
    if value <= 0:
        raise CaptureExportError(
            "capture_timestamp_invalid", "Capture timestamp is invalid"
        )
    try:
        return datetime.fromtimestamp(value, timezone.utc)
    except (OverflowError, OSError, ValueError) as exc:
        raise CaptureExportError(
            "capture_timestamp_invalid", "Capture timestamp is invalid"
        ) from exc


def _pause_requested(
    predicate: Optional[Callable[[], bool]],
) -> bool:
    if predicate is None:
        return False
    try:
        value = predicate()
    except Exception as exc:
        raise CaptureExportError(
            "pause_predicate_error", "Capture export pause check failed"
        ) from exc
    if not isinstance(value, bool):
        raise CaptureExportError(
            "pause_predicate_invalid",
            "Capture export pause check did not return a boolean",
        )
    return value
