From 007ff9fff2732ad22c5064f3ac92c8afeb517544 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 17 Jul 2026 17:51:05 +0300 Subject: [PATCH] feat(k1): integrate durable evidence with acquisition lifecycle --- src/k1link/viewer/runtime.py | 4 +- src/k1link/web/app.py | 131 +++- src/k1link/web/xgrids_k1_camera.py | 847 +++++++++++++++++------- src/k1link/web/xgrids_k1_facade.py | 194 +++++- tests/test_session_catalog_lifecycle.py | 357 ++++++++++ tests/test_xgrids_camera_gateway.py | 347 +++++++++- 6 files changed, 1638 insertions(+), 242 deletions(-) create mode 100644 tests/test_session_catalog_lifecycle.py diff --git a/src/k1link/viewer/runtime.py b/src/k1link/viewer/runtime.py index 9ac3da1..a2291e7 100644 --- a/src/k1link/viewer/runtime.py +++ b/src/k1link/viewer/runtime.py @@ -514,9 +514,9 @@ class VisualizationRuntime: callback() -def new_live_session_dir(repository_root: Path) -> Path: +def new_live_session_dir(sessions_root: Path) -> Path: stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") - base = repository_root / "sessions" / f"{stamp}_viewer_live" + base = sessions_root / f"{stamp}_viewer_live" candidate = base suffix = 1 while candidate.exists(): diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index d0ea93a..4c9f566 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -2,7 +2,7 @@ from __future__ import annotations import asyncio from collections.abc import AsyncIterator -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, suppress from pathlib import Path from typing import Any @@ -13,6 +13,19 @@ from fastapi.staticfiles import StaticFiles from pydantic import ValidationError from k1link import __version__ +from k1link.sessions import ( + MaterializedRecording, + RecordedMediaInspector, + RecordedMediaManifest, + RecordingPreparationQueueFull, + ReplayCommand, + SessionRecordingMaterializer, + SessionRecordingPreparationManager, + SessionStore, + recover_stale_active_session_marker, + resolve_missioncore_evidence_dir, +) +from k1link.web.camera_archive import recover_incomplete_camera_archives from k1link.web.device_plugin_composition import load_installed_device_plugins from k1link.web.plugin_catalog import DevicePluginCatalog, PluginCatalogError from k1link.web.plugin_runtime import ( @@ -23,21 +36,125 @@ from k1link.web.plugin_runtime import ( PluginExecutionError, PluginNotFoundError, ) +from k1link.web.session_api import build_session_router REPOSITORY_ROOT = Path(__file__).resolve().parents[3] INVALID_REQUEST_DETAIL = "Некорректные параметры запроса." +legacy_observation_sessions_root = REPOSITORY_ROOT / "sessions" +observation_sessions_root = resolve_missioncore_evidence_dir(REPOSITORY_ROOT) plugin_environment = load_installed_device_plugins(REPOSITORY_ROOT) plugin_catalog: DevicePluginCatalog = plugin_environment.catalog plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher +session_store = SessionStore(REPOSITORY_ROOT) +session_recording_materializer = SessionRecordingMaterializer(session_store.data_dir) +session_recorded_media_inspector = RecordedMediaInspector( + session_store.data_dir / "recorded-media-preparations" +) + + +def _prepare_recorded_media_for_launch( + command: ReplayCommand, + _: MaterializedRecording, +) -> tuple[RecordedMediaManifest, ...]: + """Prepare all camera descriptors inside the background job.""" + + return tuple( + session_recorded_media_inspector.inspect(artifact, command) + for artifact in session_store.list_recorded_media(command.session_id) + ) + + +session_recording_preparation_manager = SessionRecordingPreparationManager( + session_recording_materializer, + ready_preparer=_prepare_recorded_media_for_launch, +) + + +def refresh_observation_catalog() -> tuple[str, ...]: + """Discover completed or recoverable local evidence without copying payloads.""" + + imported = [ + *session_store.import_legacy_viewer_live(legacy_observation_sessions_root), + *session_store.import_legacy_viewer_live(observation_sessions_root), + ] + return tuple(dict.fromkeys(imported)) + + +def enqueue_replayable_recordings() -> tuple[str, ...]: + """Reconcile finalized catalog sessions into the durable RRD work queue.""" + + enqueued: list[str] = [] + cursor: str | None = None + while True: + page = session_store.list_recent(limit=100, cursor=cursor) + for summary in page.items: + if not summary.replayable or summary.status not in { + "ready", + "interrupted", + "failed", + }: + continue + try: + command = session_store.prepare_replay(summary.session_id) + session_recording_preparation_manager.enqueue( + command, + retry_interrupted=True, + ) + except RecordingPreparationQueueFull: + return tuple(enqueued) + except Exception: + # One stale/corrupt catalog row must not starve every valid + # finalized session that follows it in the reconciliation + # page. The row remains visible as failed/interrupted evidence + # and can be repaired independently. + continue + enqueued.append(summary.session_id) + cursor = page.next_cursor + if cursor is None: + return tuple(enqueued) + + +async def _recording_preparation_reconciler() -> None: + """Discover newly completed/recovered captures without blocking requests.""" + + while True: + try: + await asyncio.to_thread(refresh_observation_catalog) + await asyncio.to_thread(enqueue_replayable_recordings) + except Exception: + # A transient filesystem/catalog failure must not permanently + # disable preparation of sessions completed later in the run. + pass + await asyncio.sleep(2.0) @asynccontextmanager async def app_lifespan(_: FastAPI) -> AsyncIterator[None]: + reconciler: asyncio.Task[None] | None = None try: + session_recording_preparation_manager.start() + # Recovery is intentionally a one-shot startup phase. The archive + # helper owns a cross-process lease, while ordinary catalog requests + # only perform discovery and therefore never touch a live writer. + await asyncio.to_thread(recover_stale_active_session_marker, observation_sessions_root) + for sessions_root in ( + legacy_observation_sessions_root, + observation_sessions_root, + ): + await asyncio.to_thread(recover_incomplete_camera_archives, sessions_root) + # Full evidence discovery, hashing and RRD queue reconciliation can be + # expensive on field captures. Start it immediately in the background + # instead of holding the ASGI startup gate. + reconciler = asyncio.create_task(_recording_preparation_reconciler()) yield finally: + if reconciler is not None: + reconciler.cancel() + with suppress(asyncio.CancelledError): + await reconciler + await asyncio.to_thread(session_recording_preparation_manager.close) plugin_environment.close() @@ -127,6 +244,18 @@ async def device_plugin_events(websocket: WebSocket, plugin_id: str) -> None: for legacy_router in plugin_environment.legacy_routers: app.include_router(legacy_router) +app.include_router( + build_session_router( + session_store, + # Production discovery belongs to the startup/background reconciler. + # HTTP list/replay paths must never rescan evidence roots inline. + catalog_refresher=None, + recording_materializer=session_recording_materializer, + recording_preparation_manager=session_recording_preparation_manager, + media_inspector=session_recorded_media_inspector, + ) +) + frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist" if frontend_dist.is_dir(): diff --git a/src/k1link/web/xgrids_k1_camera.py b/src/k1link/web/xgrids_k1_camera.py index fb9d095..97f3878 100644 --- a/src/k1link/web/xgrids_k1_camera.py +++ b/src/k1link/web/xgrids_k1_camera.py @@ -15,6 +15,12 @@ from typing import IO, Any, Final, Literal from fastapi import APIRouter, WebSocket, WebSocketDisconnect from k1link.mqtt import validate_private_ipv4 +from k1link.web.camera_archive import ( + CameraArchiveError, + CameraArchiveKind, + CameraArchiveStatus, + CameraArchiveWriter, +) CameraSourceId = Literal["sensor.camera.left", "sensor.camera.right"] @@ -31,6 +37,7 @@ CAMERA_EXCLUSIVE_GROUP: Final = "camera.preview.decoder" MAX_FMP4_BOX_BYTES: Final = 8 * 1024 * 1024 MAX_FMP4_SEGMENT_BYTES: Final = 1024 * 1024 MAX_QUEUED_SEGMENTS: Final = 4 +CAMERA_DRAIN_TIMEOUT_SECONDS: Final = 5.0 @dataclass @@ -38,34 +45,52 @@ class CameraProcessLease: generation: int source_id: CameraSourceId process: subprocess.Popen[bytes] - stderr_tail: deque[str] = field(default_factory=lambda: deque(maxlen=12)) + stderr_tail: deque[str] segments: queue.Queue[tuple[str, bytes] | None] = field( default_factory=lambda: queue.Queue(maxsize=MAX_QUEUED_SEGMENTS) ) failure_code: str | None = None -class XgridsK1CameraGateway: - """One fail-closed K1 RTSP owner with a browser-safe fMP4 delivery plane. +@dataclass +class _CameraProducer: + generation: int + source_id: CameraSourceId + process: subprocess.Popen[bytes] + archive: CameraArchiveWriter | None + stderr_tail: deque[str] = field(default_factory=lambda: deque(maxlen=12)) + delivery: CameraProcessLease | None = None + init_segment: bytes | None = None + failure_code: str | None = None + stop_requested: bool = False + drain_requested: bool = False + reader_started: bool = False + reader_done: threading.Event = field(default_factory=threading.Event) - Selection and delivery are deliberately separate. A plugin action selects - one allowlisted producer and publishes an opaque generation. Only a matching - WebSocket may then start FFmpeg. This keeps the device address and RTSP path - out of the generic UI and prevents two browser windows from opening two K1 - sessions. + +class XgridsK1CameraGateway: + """One fail-closed K1 producer with independent archive and preview planes. + + During an acquisition the gateway, rather than a browser WebSocket, owns + FFmpeg. Every complete fMP4 segment is durably appended before it can enter + the bounded preview queue. Attaching, dropping, or disconnecting a browser + therefore cannot stop or back-pressure the source-of-record camera stream. + Outside an acquisition the legacy lazy-preview lifecycle remains available. """ def __init__(self, repository_root: Path, plugin_id: str) -> None: self._repository_root = repository_root.resolve() self._plugin_id = plugin_id self._lock = threading.RLock() + self._lifecycle_lock = threading.Lock() self._revision = 0 self._generation = 0 self._phase = "idle" self._source_id: CameraSourceId | None = None self._target_host: str | None = None - self._process: subprocess.Popen[bytes] | None = None - self._process_generation: int | None = None + self._producer: _CameraProducer | None = None + self._recording_root: Path | None = None + self._archive_summaries: list[dict[str, Any]] = [] self._error: dict[str, str] | None = None self._closed = False self._ffmpeg_path, self._ffmpeg_source = _resolve_ffmpeg(self._repository_root) @@ -97,6 +122,21 @@ class XgridsK1CameraGateway: "exclusive_group": CAMERA_EXCLUSIVE_GROUP, "max_active": 1, }, + "recording": { + "active": self._recording_root is not None, + "session": ( + self._recording_root.name if self._recording_root is not None else None + ), + "active_epoch": ( + self._producer.generation + if self._producer is not None and self._producer.archive is not None + else None + ), + "completed_epochs": len(self._archive_summaries), + "last_summary": ( + dict(self._archive_summaries[-1]) if self._archive_summaries else None + ), + }, "delivery": delivery, "runtime_dependency": { "kind": "ffmpeg", @@ -110,192 +150,537 @@ class XgridsK1CameraGateway: if source_id not in CAMERA_SOURCE_PATHS: raise ValueError("неизвестный camera source") target = validate_private_ipv4(target_host) + with self._lifecycle_lock: + with self._lock: + self._require_open_locked() + if self._ffmpeg_path is None: + self._generation += 1 + self._revision += 1 + self._source_id = source_id + self._target_host = target + self._set_error_locked( + "ffmpeg-unavailable", + "Локальный camera adapter FFmpeg не найден.", + ) + raise RuntimeError("локальный camera adapter FFmpeg не найден") + if ( + self._source_id == source_id + and self._target_host == target + and self._phase in {"selected", "connecting", "streaming"} + ): + return self.snapshot() - old_process: subprocess.Popen[bytes] | None = None - with self._lock: - if self._closed: - raise RuntimeError("camera gateway уже закрыт") - if self._ffmpeg_path is None: + old_producer, old_delivery = self._detach_producer_locked() self._generation += 1 self._revision += 1 self._source_id = source_id self._target_host = target - self._phase = "error" - self._error = { - "code": "ffmpeg-unavailable", - "message": "Локальный camera adapter FFmpeg не найден.", - } - raise RuntimeError("локальный camera adapter FFmpeg не найден") - if ( - self._source_id == source_id - and self._target_host == target - and self._phase in {"selected", "connecting", "streaming"} - ): - return self.snapshot() + self._phase = "selected" + self._error = None + recording_active = self._recording_root is not None - old_process = self._detach_process_locked() - self._generation += 1 - self._revision += 1 - self._source_id = source_id - self._target_host = target - self._phase = "selected" - self._error = None - - _terminate_process(old_process) - return self.snapshot() + self._shutdown_producer( + old_producer, + old_delivery, + status="complete", + failure_code="source-switch", + ) + if recording_active: + self._spawn_selected_producer() + return self.snapshot() def stop(self, generation: int) -> dict[str, Any]: - old_process: subprocess.Popen[bytes] | None - with self._lock: - if self._source_id is None: - return self.snapshot() - if generation != self._generation: - raise ValueError("camera preview generation устарело") - old_process = self._detach_process_locked() - self._revision += 1 - self._phase = "idle" - self._source_id = None - self._target_host = None - self._error = None - _terminate_process(old_process) - return self.snapshot() + with self._lifecycle_lock: + with self._lock: + if self._source_id is None: + return self.snapshot() + if generation != self._generation: + raise ValueError("camera preview generation устарело") + producer, delivery = self._detach_producer_locked() + self._revision += 1 + self._phase = "idle" + self._source_id = None + self._target_host = None + self._error = None + self._shutdown_producer( + producer, + delivery, + status="complete", + failure_code="source-stopped", + ) + return self.snapshot() def stop_current(self) -> dict[str, Any]: - old_process: subprocess.Popen[bytes] | None - with self._lock: - old_process = self._detach_process_locked() - changed = self._source_id is not None or self._phase != "idle" - if changed: + with self._lifecycle_lock: + with self._lock: + producer, delivery = self._detach_producer_locked() + changed = self._source_id is not None or self._phase != "idle" + if changed: + self._revision += 1 + self._phase = "idle" + self._source_id = None + self._target_host = None + self._recording_root = None + self._error = None + self._shutdown_producer( + producer, + delivery, + status="interrupted", + failure_code="gateway-stop", + ) + return self.snapshot() + + def start_recording(self, session_dir: Path) -> dict[str, Any]: + """Make an existing observation session the camera recording root.""" + + root = session_dir.expanduser().resolve() + if not root.is_dir(): + raise ValueError("observation session directory does not exist") + if not root.is_relative_to(self._repository_root): + raise ValueError("camera recording root must stay inside the repository") + + with self._lifecycle_lock: + with self._lock: + self._require_open_locked() + if self._recording_root is not None: + if self._recording_root == root: + return self.snapshot() + raise RuntimeError("для camera gateway уже активна другая acquisition-сессия") + producer, delivery = self._detach_producer_locked() + self._recording_root = root + self._archive_summaries = [] + selected = self._source_id is not None + if selected: + self._phase = "selected" self._revision += 1 - self._phase = "idle" - self._source_id = None - self._target_host = None - self._error = None - _terminate_process(old_process) - return self.snapshot() + + # A pre-acquisition browser-owned process cannot become evidence + # retrospectively. Restart it at a clean codec epoch instead. + self._shutdown_producer( + producer, + delivery, + status="interrupted", + failure_code="recording-start-restart", + ) + if selected: + self._spawn_selected_producer() + return self.snapshot() + + def stop_recording( + self, + *, + status: CameraArchiveStatus = "complete", + failure_code: str | None = None, + ) -> dict[str, Any]: + """Stop the acquisition-owned producer and seal its active epoch.""" + + with self._lifecycle_lock: + with self._lock: + producer, delivery = self._detach_producer_locked() + recording_was_active = self._recording_root is not None + self._recording_root = None + if self._source_id is not None and self._phase != "error": + self._phase = "selected" + if recording_was_active or producer is not None: + self._revision += 1 + self._shutdown_producer( + producer, + delivery, + status=status, + failure_code=failure_code, + ) + return self.snapshot() def open_delivery(self, generation: int) -> CameraProcessLease: - with self._lock: - if self._closed: - raise RuntimeError("camera gateway закрыт") - if generation != self._generation or self._source_id is None: - raise ValueError("camera preview generation не активно") - if self._process is not None: - raise RuntimeError("для camera preview уже открыт browser consumer") - if self._ffmpeg_path is None or self._target_host is None: - raise RuntimeError("camera preview runtime не готов") - - argv = _build_ffmpeg_argv( - self._ffmpeg_path, - self._target_host, - self._source_id, - ) - try: - process = subprocess.Popen( - argv, - stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - shell=False, - start_new_session=(os.name == "posix"), + with self._lifecycle_lock: + with self._lock: + self._require_open_locked() + if generation != self._generation or self._source_id is None: + raise ValueError("camera preview generation не активно") + producer = self._producer + if producer is None: + producer = self._spawn_selected_producer() + with self._lock: + if self._producer is not producer or producer.generation != generation: + raise ValueError("camera preview generation не активно") + if producer.delivery is not None: + raise RuntimeError("для camera preview уже открыт browser consumer") + lease = CameraProcessLease( + generation=producer.generation, + source_id=producer.source_id, + process=producer.process, + stderr_tail=producer.stderr_tail, ) - except OSError as exc: + producer.delivery = lease + if producer.init_segment is not None: + lease.segments.put_nowait(("init", producer.init_segment)) self._revision += 1 - self._phase = "error" - self._error = { - "code": "ffmpeg-start-failed", - "message": "Не удалось запустить локальный camera adapter.", - } - raise RuntimeError("не удалось запустить camera adapter") from exc - - if process.stdout is None or process.stderr is None: - _terminate_process(process) - raise RuntimeError("camera adapter не открыл media pipes") - lease = CameraProcessLease( - generation=generation, - source_id=self._source_id, - process=process, - ) - self._process = process - self._process_generation = generation - self._revision += 1 - self._phase = "connecting" - self._error = None - - threading.Thread( - target=_drain_stderr, - args=(lease,), - name=f"k1-camera-stderr-{generation}", - daemon=True, - ).start() - threading.Thread( - target=_read_fmp4_stdout, - args=(lease,), - name=f"k1-camera-fmp4-{generation}", - daemon=True, - ).start() - return lease + return lease def mark_streaming(self, lease: CameraProcessLease) -> None: with self._lock: - if ( - lease.generation != self._generation - or self._process is not lease.process - or self._phase == "streaming" - ): + producer = self._producer + if producer is None or producer.delivery is not lease: return + self._mark_streaming_locked(producer) + + def release_delivery(self, lease: CameraProcessLease, *, client_closed: bool) -> None: + producer_to_stop: _CameraProducer | None = None + with self._lifecycle_lock: + with self._lock: + producer = self._producer + if producer is None or producer.delivery is not lease: + return + producer.delivery = None + self._revision += 1 + # During acquisition the browser is a disposable observer. In + # legacy preview-only mode retain the old lazy-owner behavior. + if self._recording_root is None: + producer.stop_requested = True + self._producer = None + producer_to_stop = producer + self._phase = "selected" + self._error = None + if producer_to_stop is not None: + self._shutdown_producer( + producer_to_stop, + None, + status="interrupted", + failure_code=(lease.failure_code or "browser-disconnected"), + ) + + def close(self) -> None: + with self._lifecycle_lock: + with self._lock: + if self._closed: + return + self._closed = True + producer, delivery = self._detach_producer_locked() + self._revision += 1 + self._phase = "idle" + self._source_id = None + self._target_host = None + self._recording_root = None + self._error = None + self._shutdown_producer( + producer, + delivery, + status="interrupted", + failure_code="gateway-closed", + ) + + def _spawn_selected_producer(self) -> _CameraProducer: + with self._lock: + self._require_open_locked() + source_id = self._source_id + target_host = self._target_host + ffmpeg_path = self._ffmpeg_path + generation = self._generation + recording_root = self._recording_root + if source_id is None or target_host is None or ffmpeg_path is None: + raise RuntimeError("camera preview runtime не готов") + if self._producer is not None: + return self._producer + + archive: CameraArchiveWriter | None = None + if recording_root is not None: + try: + archive = CameraArchiveWriter(recording_root, source_id, generation) + except (OSError, ValueError, CameraArchiveError) as exc: + with self._lock: + self._set_error_locked( + "camera-storage-failed", + "Не удалось открыть долговременное хранилище camera stream.", + ) + raise RuntimeError( + "Не удалось открыть долговременное хранилище camera stream." + ) from exc + try: + process = subprocess.Popen( + _build_ffmpeg_argv(ffmpeg_path, target_host, source_id), + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + shell=False, + start_new_session=(os.name == "posix"), + ) + except (OSError, ValueError) as exc: + if archive is not None: + with suppress(CameraArchiveError): + self._record_archive_summary( + archive.close(status="failed", failure_code="ffmpeg-start-failed") + ) + with self._lock: + self._set_error_locked( + "ffmpeg-start-failed", + "Не удалось запустить локальный camera adapter.", + ) + raise RuntimeError("Не удалось запустить локальный camera adapter.") from exc + + if process.stdout is None or process.stderr is None: + _terminate_process(process) + if archive is not None: + self._record_archive_summary( + archive.close(status="failed", failure_code="ffmpeg-pipes-unavailable") + ) + with self._lock: + self._set_error_locked( + "ffmpeg-pipes-unavailable", + "Camera adapter не открыл media pipes.", + ) + raise RuntimeError("camera adapter не открыл media pipes") + + producer = _CameraProducer( + generation=generation, + source_id=source_id, + process=process, + archive=archive, + ) + with self._lock: + if ( + self._closed + or generation != self._generation + or source_id != self._source_id + or target_host != self._target_host + ): + producer.stop_requested = True + stale = True + else: + self._producer = producer + self._revision += 1 + self._phase = "connecting" + self._error = None + stale = False + if stale: + self._shutdown_producer( + producer, + None, + status="interrupted", + failure_code="stale-generation", + ) + raise RuntimeError("camera generation изменилась во время запуска adapter") + + threading.Thread( + target=_drain_stderr, + args=(producer,), + name=f"k1-camera-stderr-{generation}", + daemon=True, + ).start() + producer.reader_started = True + threading.Thread( + target=_read_fmp4_stdout, + args=(self, producer), + name=f"k1-camera-fmp4-{generation}", + daemon=True, + ).start() + return producer + + def _publish_segment( + self, + producer: _CameraProducer, + kind: CameraArchiveKind, + payload: bytes, + ) -> bool: + if len(payload) > MAX_FMP4_SEGMENT_BYTES: + self._mark_producer_failure( + producer, + "segment-too-large", + "Camera adapter отклонил слишком большой video segment.", + ) + return False + + with self._lock: + producer_owned = self._producer is producer or ( + producer.drain_requested and producer.archive is not None + ) + if not producer_owned or producer.stop_requested: + return False + archive = producer.archive + if archive is not None: + try: + # Source of record first; preview is always expendable. + archive.append(kind, payload) + except (CameraArchiveError, OSError, ValueError): + self._mark_producer_failure( + producer, + "camera-storage-failed", + "Долговременная запись camera stream завершилась ошибкой.", + ) + return False + + with self._lock: + producer_owned = self._producer is producer or ( + producer.drain_requested and producer.archive is not None + ) + if not producer_owned or producer.stop_requested: + return False + if kind == "init": + producer.init_segment = payload + else: + self._mark_streaming_locked(producer) + delivery = producer.delivery + if delivery is None: + return True + try: + delivery.segments.put_nowait((kind, payload)) + except queue.Full: + self._drop_slow_delivery(producer, delivery) + return True + + def _drop_slow_delivery( + self, + producer: _CameraProducer, + delivery: CameraProcessLease, + ) -> None: + with self._lock: + if self._producer is producer and producer.delivery is delivery: + producer.delivery = None + delivery.failure_code = "consumer-too-slow" + self._revision += 1 + _close_segment_queue(delivery) + + def _mark_producer_failure( + self, + producer: _CameraProducer, + code: str, + message: str, + ) -> None: + with self._lock: + producer_owned = self._producer is producer or producer.drain_requested + if not producer_owned or producer.stop_requested: + return + producer.failure_code = code + self._set_error_locked(code, message) + with suppress(OSError): + producer.process.terminate() + + def _producer_ended(self, producer: _CameraProducer) -> None: + with self._lock: + delivery = producer.delivery + producer.delivery = None + owns_shutdown = self._producer is producer and not producer.stop_requested + if owns_shutdown: + self._producer = None + self._revision += 1 + if producer.failure_code is None: + producer.failure_code = "camera-source-ended" + self._set_error_locked( + "camera-source-ended", + _safe_ffmpeg_message(producer.stderr_tail), + ) + if delivery is not None: + delivery.failure_code = producer.failure_code or "camera-source-ended" + _close_segment_queue(delivery) + if not owns_shutdown: + return + _terminate_process(producer.process) + status: CameraArchiveStatus = ( + "interrupted" + if producer.failure_code + in {"camera-source-ended", "incomplete-fmp4-fragment"} + else "failed" + ) + try: + self._finalize_archive(producer, status, producer.failure_code) + except CameraArchiveError: + with self._lock: + self._set_error_locked( + "camera-storage-finalize-failed", + "Не удалось завершить долговременную запись camera stream.", + ) + + def _shutdown_producer( + self, + producer: _CameraProducer | None, + delivery: CameraProcessLease | None, + *, + status: CameraArchiveStatus, + failure_code: str | None, + ) -> None: + if delivery is not None: + delivery.failure_code = producer.failure_code if producer is not None else failure_code + _close_segment_queue(delivery) + if producer is None: + return + _terminate_process(producer.process, close_streams=False) + drained = ( + not producer.reader_started + or producer.reader_done.wait(timeout=CAMERA_DRAIN_TIMEOUT_SECONDS) + ) + _close_process_streams(producer.process) + if not drained: + producer.failure_code = "camera-drain-timeout" + with self._lock: + self._set_error_locked( + "camera-drain-timeout", + "Camera adapter не завершил долговременную запись вовремя.", + ) + effective_failure_code = producer.failure_code or failure_code + effective_status: CameraArchiveStatus = ( + "interrupted" + if producer.failure_code == "incomplete-fmp4-fragment" + else "failed" + if producer.failure_code is not None + else status + ) + if delivery is not None: + delivery.failure_code = effective_failure_code + try: + self._finalize_archive( + producer, + effective_status, + effective_failure_code, + ) + except CameraArchiveError: + with self._lock: + self._set_error_locked( + "camera-storage-finalize-failed", + "Не удалось завершить долговременную запись camera stream.", + ) + raise + + def _finalize_archive( + self, + producer: _CameraProducer, + status: CameraArchiveStatus, + failure_code: str | None, + ) -> None: + archive = producer.archive + producer.archive = None + if archive is None: + return + self._record_archive_summary(archive.close(status=status, failure_code=failure_code)) + + def _record_archive_summary(self, summary: dict[str, Any]) -> None: + with self._lock: + self._archive_summaries.append(dict(summary)) + + def _detach_producer_locked( + self, + ) -> tuple[_CameraProducer | None, CameraProcessLease | None]: + producer = self._producer + self._producer = None + if producer is None: + return None, None + producer.drain_requested = producer.archive is not None + producer.stop_requested = not producer.drain_requested + delivery = producer.delivery + producer.delivery = None + return producer, delivery + + def _mark_streaming_locked(self, producer: _CameraProducer) -> None: + if self._producer is producer and self._phase != "streaming": self._revision += 1 self._phase = "streaming" - def release_delivery(self, lease: CameraProcessLease, *, client_closed: bool) -> None: - _terminate_process(lease.process) - with self._lock: - if self._process is lease.process: - self._process = None - self._process_generation = None - if lease.generation != self._generation or self._source_id is None: - return - self._revision += 1 - if client_closed: - self._phase = "selected" - self._error = None - else: - self._phase = "error" - if lease.failure_code == "consumer-too-slow": - message = ( - "Browser не успевает принимать camera stream; " - "канал остановлен без накопления задержки." - ) - elif lease.failure_code == "invalid-fmp4": - message = "Camera adapter вернул некорректный fMP4 stream." - elif lease.failure_code == "segment-too-large": - message = "Camera adapter отклонил слишком большой video segment." - else: - message = _safe_ffmpeg_message(lease.stderr_tail) - self._error = { - "code": lease.failure_code or "camera-source-ended", - "message": message, - } + def _set_error_locked(self, code: str, message: str) -> None: + self._revision += 1 + self._phase = "error" + self._error = {"code": code, "message": message} - def close(self) -> None: - old_process: subprocess.Popen[bytes] | None - with self._lock: - if self._closed: - return - self._closed = True - old_process = self._detach_process_locked() - self._revision += 1 - self._phase = "idle" - self._source_id = None - self._target_host = None - self._error = None - _terminate_process(old_process) - - def _detach_process_locked(self) -> subprocess.Popen[bytes] | None: - process = self._process - self._process = None - self._process_generation = None - return process + def _require_open_locked(self) -> None: + if self._closed: + raise RuntimeError("camera gateway уже закрыт") def _resolve_ffmpeg(repository_root: Path) -> tuple[Path | None, str]: @@ -394,36 +779,10 @@ def _read_mp4_box(stream: IO[bytes]) -> tuple[bytes, bytes]: return box_type, header + _read_exact(stream, size - len(header)) -def _enqueue_segment(lease: CameraProcessLease, kind: str, payload: bytes) -> bool: - if len(payload) > MAX_FMP4_SEGMENT_BYTES: - lease.failure_code = "segment-too-large" - _finish_segment_queue(lease) - with suppress(OSError): - lease.process.terminate() - return False - try: - lease.segments.put_nowait((kind, payload)) - return True - except queue.Full: - lease.failure_code = "consumer-too-slow" - while True: - try: - lease.segments.get_nowait() - except queue.Empty: - break - lease.segments.put_nowait(None) - with suppress(OSError): - lease.process.terminate() - return False - - -def _finish_segment_queue(lease: CameraProcessLease) -> None: +def _close_segment_queue(lease: CameraProcessLease) -> None: try: lease.segments.put_nowait(None) except queue.Full: - # A full queue is itself a bounded-latency failure. Never leave FFmpeg - # back-pressured while the browser accumulates tens of seconds. - lease.failure_code = lease.failure_code or "consumer-too-slow" while True: try: lease.segments.get_nowait() @@ -432,11 +791,21 @@ def _finish_segment_queue(lease: CameraProcessLease) -> None: lease.segments.put_nowait(None) -def _read_fmp4_stdout(lease: CameraProcessLease) -> None: - stream = lease.process.stdout +def _read_fmp4_stdout( + gateway: XgridsK1CameraGateway, + producer: _CameraProducer, +) -> None: + stream = producer.process.stdout if stream is None: - lease.failure_code = "invalid-fmp4" - _finish_segment_queue(lease) + try: + gateway._mark_producer_failure( + producer, + "invalid-fmp4", + "Camera adapter вернул некорректный fMP4 stream.", + ) + gateway._producer_ended(producer) + finally: + producer.reader_done.set() return init_parts: list[bytes] = [] @@ -448,7 +817,11 @@ def _read_fmp4_stdout(lease: CameraProcessLease) -> None: if not init_sent: init_parts.append(box) if box_type == b"moov": - if not _enqueue_segment(lease, "init", b"".join(init_parts)): + if not gateway._publish_segment( + producer, + "init", + b"".join(init_parts), + ): return init_sent = True continue @@ -459,7 +832,11 @@ def _read_fmp4_stdout(lease: CameraProcessLease) -> None: if fragment_parts: fragment_parts.append(box) if box_type == b"mdat": - if not _enqueue_segment(lease, "media", b"".join(fragment_parts)): + if not gateway._publish_segment( + producer, + "media", + b"".join(fragment_parts), + ): return fragment_parts = [] continue @@ -469,15 +846,32 @@ def _read_fmp4_stdout(lease: CameraProcessLease) -> None: fragment_parts.append(box) except EOFError: if not init_sent: - lease.failure_code = "invalid-fmp4" + gateway._mark_producer_failure( + producer, + "invalid-fmp4", + "Camera adapter вернул некорректный fMP4 stream.", + ) + elif fragment_parts: + gateway._mark_producer_failure( + producer, + "incomplete-fmp4-fragment", + "Camera stream завершился на неполном video fragment.", + ) except (OSError, ValueError): - lease.failure_code = "invalid-fmp4" + gateway._mark_producer_failure( + producer, + "invalid-fmp4", + "Camera adapter вернул некорректный fMP4 stream.", + ) finally: - _finish_segment_queue(lease) + try: + gateway._producer_ended(producer) + finally: + producer.reader_done.set() -def _drain_stderr(lease: CameraProcessLease) -> None: - stream = lease.process.stderr +def _drain_stderr(producer: _CameraProducer) -> None: + stream = producer.process.stderr if stream is None: return try: @@ -487,7 +881,7 @@ def _drain_stderr(lease: CameraProcessLease) -> None: return text = line.decode("utf-8", errors="replace").strip() if text: - lease.stderr_tail.append(text[-240:]) + producer.stderr_tail.append(text[-240:]) except (OSError, ValueError): return @@ -504,13 +898,21 @@ def _safe_ffmpeg_message(stderr_tail: deque[str]) -> str: return "Camera stream завершился до первого пригодного видеофрагмента." -def _terminate_process(process: subprocess.Popen[bytes] | None) -> None: +def _terminate_process( + process: subprocess.Popen[bytes] | None, + *, + close_streams: bool = True, +) -> None: if process is None: return try: if process.poll() is None: if os.name == "posix": - with suppress(ProcessLookupError): + # The process can have exited or lost its dedicated process + # group between poll() and killpg(). Either outcome means + # there is no group left for us to signal; still reap the + # child and finish the archive instead of aborting teardown. + with suppress(OSError): os.killpg(process.pid, signal.SIGINT) else: process.terminate() @@ -524,10 +926,15 @@ def _terminate_process(process: subprocess.Popen[bytes] | None) -> None: process.kill() process.wait(timeout=1.0) finally: - for stream in (process.stdout, process.stderr): - if stream is not None: - with suppress(OSError): - stream.close() + if close_streams: + _close_process_streams(process) + + +def _close_process_streams(process: subprocess.Popen[bytes]) -> None: + for stream in (process.stdout, process.stderr): + if stream is not None: + with suppress(OSError): + stream.close() def build_xgrids_k1_camera_router( @@ -536,9 +943,7 @@ def build_xgrids_k1_camera_router( ) -> APIRouter: router = APIRouter(include_in_schema=False) - @router.websocket( - f"/api/v1/device-plugins/{plugin_id}/camera-preview/{{generation}}" - ) + @router.websocket(f"/api/v1/device-plugins/{plugin_id}/camera-preview/{{generation}}") async def camera_preview(websocket: WebSocket, generation: int) -> None: await websocket.accept() try: diff --git a/src/k1link/web/xgrids_k1_facade.py b/src/k1link/web/xgrids_k1_facade.py index 8548900..75f3bb4 100644 --- a/src/k1link/web/xgrids_k1_facade.py +++ b/src/k1link/web/xgrids_k1_facade.py @@ -7,7 +7,9 @@ import importlib.util import json import secrets import threading +import time from collections.abc import Mapping +from contextlib import suppress from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal, Protocol, cast @@ -20,6 +22,7 @@ from k1link.ble.scanner import scan from k1link.ble.wifi_provisioning import AP_FALLBACK_IPV4, provision_wifi_once from k1link.mqtt import validate_private_ipv4 from k1link.protocol.normalizer import normalize_k1_message +from k1link.sessions import ActiveSessionLease, resolve_missioncore_evidence_dir from k1link.viewer.rerun_bridge import RerunSceneSettings from k1link.viewer.runtime import VisualizationRuntime, new_live_session_dir from k1link.web.device_lifecycle import ( @@ -181,6 +184,7 @@ class XgridsK1CompatibilityService: def __init__(self, repository_root: Path) -> None: self.repository_root = repository_root.resolve() + self.evidence_root = resolve_missioncore_evidence_dir(self.repository_root) self._lock = threading.Lock() self._provisioning_gate = threading.Lock() self._provisioning_active = False @@ -206,6 +210,7 @@ class XgridsK1CompatibilityService: self._acquisition_out_dir: Path | None = None self._acquisition_start_operation_id: str | None = None self._acquisition_stop_operation_id: str | None = None + self._acquisition_session_lease: ActiveSessionLease | None = None # The host-owned visual runtime receives the vendor normalizer # explicitly. There is no implicit K1 decoder in the visual layer. self.runtime = VisualizationRuntime(normalizer=normalize_k1_message) @@ -216,7 +221,8 @@ class XgridsK1CompatibilityService: def state(self) -> dict[str, Any]: runtime = self.runtime.snapshot() - self._reconcile_acquisition(runtime) + camera_preview = self.camera_preview.snapshot() + self._reconcile_acquisition(runtime, camera_preview) camera_preview = self.camera_preview.snapshot() metrics = runtime["metrics"] with self._lock: @@ -401,9 +407,6 @@ class XgridsK1CompatibilityService: known_ids = {str(item["device_id"]) for item in self.state()["devices"]} if request.device_id not in known_ids: raise ValueError("сначала найдите и выберите устройство через Bluetooth") - # A reprovision can replace both the device session and its address. - # Revoke the previous browser lease before any new device-side write. - self.camera_preview.stop_current() # Unwrap once at the provisioning service boundary. The plain value is # kept only in this stack frame, included in a keyed request digest, and # passed to the reviewed BLE write boundary; it is never journaled. @@ -463,12 +466,17 @@ class XgridsK1CompatibilityService: ) self._provisioning_active = True + # A reprovision can replace both the device session and its address. + # Revoke preview/producer state only after the active-acquisition + # guard; a rejected network write must never stop evidence capture. + self.camera_preview.stop_current() + self._set_operation( "provisioning", "Передаём устройству настройки Wi-Fi одним подтверждённым запросом.", ) session_dir = _new_operation_session_dir( - self.repository_root, + self.evidence_root, "viewer_wifi_provisioning", ) session_dir.mkdir(parents=True, exist_ok=False) @@ -654,7 +662,7 @@ class XgridsK1CompatibilityService: }, ) self._acquisition = acquisition - self._acquisition_out_dir = new_live_session_dir(self.repository_root) + self._acquisition_out_dir = new_live_session_dir(self.evidence_root) self._acquisition_start_operation_id = None self._acquisition_stop_operation_id = None self._compatibility_attestation = _attestation_snapshot( @@ -725,12 +733,27 @@ class XgridsK1CompatibilityService: message_code="acquisition.start.waiting_receiver_ready", ) self._acquisition_start_operation_id = operation.operation_id + lease = ActiveSessionLease.acquire(out_dir.parent, out_dir) + with self._lock: + if self._acquisition_session_lease is not None: + lease.release() + raise RuntimeError("evidence-сессия уже удерживается активным acquisition") + self._acquisition_session_lease = lease self.runtime.start_live( acquisition.target_host, out_dir, duration_seconds=acquisition.duration_seconds, ) + self._arm_camera_recording(out_dir) except Exception as exc: + with suppress(Exception): + self.camera_preview.stop_recording( + status="failed", + failure_code="acquisition-start-failed", + ) + with suppress(Exception): + self.runtime.stop() + self._release_acquisition_session_lease() with self._lock: if acquisition.state not in TERMINAL_ACQUISITION_STATES: acquisition.transition("failed", message_code="acquisition.start.failed") @@ -845,8 +868,10 @@ class XgridsK1CompatibilityService: try: with self._lock: acquisition.transition("stopping", message_code="acquisition.stopping") - self.camera_preview.stop_current() - self.runtime.stop() + self._stop_acquisition_sources( + camera_status="complete", + camera_failure_code=None, + ) self._cancel_pending_acquisition_operations( exclude_operation_id=operation.operation_id, reason_code="superseded-by-stop", @@ -915,8 +940,10 @@ class XgridsK1CompatibilityService: return self.state() try: if acquisition.state not in TERMINAL_ACQUISITION_STATES: - self.camera_preview.stop_current() - self.runtime.stop() + self._stop_acquisition_sources( + camera_status="interrupted", + camera_failure_code="acquisition-aborted", + ) self._cancel_pending_acquisition_operations( exclude_operation_id=operation.operation_id, reason_code="superseded-by-abort", @@ -997,10 +1024,10 @@ class XgridsK1CompatibilityService: def stop(self) -> dict[str, Any]: """Deprecated capture-only shim; it never claims that K1 stopped scanning.""" - self.camera_preview.stop_current() with self._lock: acquisition = self._acquisition if acquisition is None or acquisition.state in TERMINAL_ACQUISITION_STATES: + self.camera_preview.stop_current() self.runtime.stop() return self.state() return self.stop_acquisition( @@ -1012,6 +1039,16 @@ class XgridsK1CompatibilityService: def select_camera_preview(self, request: CameraPreviewSelectRequest) -> dict[str, Any]: target = self._camera_target_for_session(request.device_session_id) + with self._lock: + acquisition = self._acquisition + out_dir = self._acquisition_out_dir + acquisition_active = ( + acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES + ) + if acquisition_active: + if out_dir is None: + raise RuntimeError("для acquisition не выделена evidence-сессия") + self._arm_camera_recording(out_dir, require_session=True) self.camera_preview.select(request.source_id, target) return self.state() @@ -1035,6 +1072,8 @@ class XgridsK1CompatibilityService: f"{type(camera_error).__name__}: {camera_error}" ) raise + finally: + self._release_acquisition_session_lease() if camera_error is not None: raise camera_error @@ -1053,6 +1092,71 @@ class XgridsK1CompatibilityService: ) return self.state() + def _arm_camera_recording( + self, + out_dir: Path, + *, + require_session: bool = False, + ) -> None: + camera = self.camera_preview.snapshot() + if camera["recording"]["active"]: + self.camera_preview.start_recording(out_dir) + return + if not out_dir.is_dir(): + if not require_session and camera["active_source_id"] is None: + # The real runtime creates the session in its source thread. If + # no camera has been selected yet, defer arming until selection + # instead of racing that mkdir. No media can be lost yet. + return + deadline = time.monotonic() + 5.0 + while not out_dir.is_dir(): + runtime = self.runtime.snapshot() + if runtime.get("phase") == "error": + raise RuntimeError( + "evidence-сессия не создана: live runtime завершился ошибкой" + ) + if time.monotonic() >= deadline: + raise RuntimeError( + "evidence-сессия не появилась вовремя для записи camera stream" + ) + time.sleep(0.01) + self.camera_preview.start_recording(out_dir) + + def _stop_acquisition_sources( + self, + *, + camera_status: Literal["complete", "interrupted", "failed"], + camera_failure_code: str | None, + ) -> None: + camera_error: Exception | None = None + try: + self.camera_preview.stop_recording( + status=camera_status, + failure_code=camera_failure_code, + ) + except Exception as exc: + camera_error = exc + try: + self.runtime.stop() + except Exception as exc: + if camera_error is not None: + exc.add_note( + "camera archive cleanup also failed: " + f"{type(camera_error).__name__}: {camera_error}" + ) + raise + finally: + self._release_acquisition_session_lease() + if camera_error is not None: + raise camera_error + + def _release_acquisition_session_lease(self) -> None: + with self._lock: + lease = self._acquisition_session_lease + self._acquisition_session_lease = None + if lease is not None: + lease.release() + def _set_operation(self, phase: str, message: str) -> None: with self._lock: self._operation_phase = phase @@ -1127,13 +1231,20 @@ class XgridsK1CompatibilityService: raise ValueError("указана неизвестная acquisition-сессия") return acquisition - def _reconcile_acquisition(self, runtime: Mapping[str, Any]) -> None: + def _reconcile_acquisition( + self, + runtime: Mapping[str, Any], + camera: Mapping[str, Any], + ) -> None: completed_operation_id: str | None = None failed_operation_id: str | None = None failed_stop_operation_id: str | None = None unconfirmed_stop_operation_id: str | None = None receiver_ready_operation_id: str | None = None acquisition_id: str | None = None + camera_terminal_status: Literal["complete", "failed"] | None = None + camera_failure_code: str | None = None + stop_runtime_for_camera_failure = False with self._lock: acquisition = self._acquisition if acquisition is None or acquisition.state in TERMINAL_ACQUISITION_STATES: @@ -1143,7 +1254,42 @@ class XgridsK1CompatibilityService: phase = runtime.get("phase") source_mode = runtime.get("source_mode") source_ready = runtime.get("source_ready") is True - if acquisition.state in {"starting", "awaiting_external_start"} and point_frames > 0: + camera_recording = camera.get("recording") + camera_recording_active = ( + isinstance(camera_recording, dict) and camera_recording.get("active") is True + ) + camera_error = camera.get("error") + if camera.get("phase") == "error" and camera_recording_active: + waiting_for_start = acquisition.state in { + "starting", + "awaiting_external_start", + } + waiting_for_stop = acquisition.state in { + "awaiting_external_stop", + "stopping", + "finalizing", + } + camera_failure_code = ( + str(camera_error.get("code")) + if isinstance(camera_error, dict) and camera_error.get("code") + else "camera-producer-failed" + ) + acquisition.transition( + "failed", + message_code="acquisition.camera_failed", + result={ + "receiver_stopped": False, + "device_state": "unknown", + "camera_failure_code": camera_failure_code, + }, + ) + if waiting_for_start: + failed_operation_id = self._acquisition_start_operation_id + if waiting_for_stop: + failed_stop_operation_id = self._acquisition_stop_operation_id + camera_terminal_status = "failed" + stop_runtime_for_camera_failure = True + elif acquisition.state in {"starting", "awaiting_external_start"} and point_frames > 0: acquisition.transition("acquiring", message_code="acquisition.acquiring") completed_operation_id = self._acquisition_start_operation_id acquisition_id = acquisition.acquisition_id @@ -1168,6 +1314,8 @@ class XgridsK1CompatibilityService: "finalizing", } acquisition.transition("failed", message_code="acquisition.runtime_failed") + camera_terminal_status = "failed" + camera_failure_code = "runtime-failed" if waiting_for_start: failed_operation_id = self._acquisition_start_operation_id if waiting_for_stop: @@ -1181,6 +1329,8 @@ class XgridsK1CompatibilityService: message_code="acquisition.receiver_completed_without_point_data", result={"receiver_stopped": True, "device_state": "unknown"}, ) + camera_terminal_status = "failed" + camera_failure_code = "receiver-completed-without-point-data" failed_operation_id = self._acquisition_start_operation_id elif acquisition.state == "acquiring" and source_mode == "idle": acquisition.transition( @@ -1188,14 +1338,28 @@ class XgridsK1CompatibilityService: message_code="acquisition.receiver_completed", result={"receiver_stopped": True, "device_state": "unknown"}, ) + camera_terminal_status = "complete" elif acquisition.state == "awaiting_external_stop" and source_mode == "idle": acquisition.transition( "failed", message_code="acquisition.receiver_completed_before_stop_confirmation", result={"receiver_stopped": True, "device_state": "unknown"}, ) + camera_terminal_status = "failed" + camera_failure_code = "receiver-completed-before-stop-confirmation" unconfirmed_stop_operation_id = self._acquisition_stop_operation_id + if camera_terminal_status is not None: + try: + self.camera_preview.stop_recording( + status=camera_terminal_status, + failure_code=camera_failure_code, + ) + finally: + self._release_acquisition_session_lease() + if stop_runtime_for_camera_failure: + self.runtime.stop() + if receiver_ready_operation_id is not None: self._operations.transition_if_pending( receiver_ready_operation_id, @@ -1581,9 +1745,9 @@ def _device_calibration_snapshot(active_profile_id: str | None) -> dict[str, Any } -def _new_operation_session_dir(repository_root: Path, suffix: str) -> Path: +def _new_operation_session_dir(sessions_root: Path, suffix: str) -> Path: stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") - base = repository_root / "sessions" / f"{stamp}_{suffix}" + base = sessions_root / f"{stamp}_{suffix}" candidate = base serial = 1 while candidate.exists(): diff --git a/tests/test_session_catalog_lifecycle.py b/tests/test_session_catalog_lifecycle.py new file mode 100644 index 0000000..f1c3709 --- /dev/null +++ b/tests/test_session_catalog_lifecycle.py @@ -0,0 +1,357 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +import threading +import time +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import pytest +from fastapi import APIRouter, HTTPException +from fastapi.routing import APIRoute + +from k1link.mqtt.capture import FRAME_HEADER, RAW_MAGIC +from k1link.sessions import ( + SessionRecordingMaterializer, + SessionRecordingPreparationManager, + SessionStore, +) +from k1link.web import app as app_module +from k1link.web.session_api import build_session_router + + +def _make_completed_session(root: Path, session_id: str) -> Path: + capture = root / session_id / "captures" / "mqtt_live" + capture.mkdir(parents=True) + topics = ( + "lixel/application/report/lio_pcl", + "lixel/application/report/lio_pose", + ) + raw = bytearray(RAW_MAGIC) + metadata: list[dict[str, object]] = [] + for sequence, topic in enumerate(topics, start=1): + encoded_topic = topic.encode("utf-8") + payload = b"fixture" + frame_offset = len(raw) + raw.extend(FRAME_HEADER.pack(len(encoded_topic), len(payload))) + raw.extend(encoded_topic) + raw.extend(payload) + metadata.append( + { + "record_type": "message", + "sequence": sequence, + "received_at_epoch_ns": 1_000_000_000 + sequence, + "received_monotonic_ns": 2_000_000_000 + sequence, + "topic": topic, + "payload_bytes": len(payload), + "raw_frame_offset": frame_offset, + "raw_payload_offset": frame_offset + FRAME_HEADER.size + len(encoded_topic), + "raw_frame_bytes": FRAME_HEADER.size + len(encoded_topic) + len(payload), + } + ) + (capture / "mqtt.raw.k1mqtt").write_bytes(raw) + metadata_path = capture / "mqtt.metadata.jsonl" + metadata_path.write_text( + "".join(json.dumps(record) + "\n" for record in metadata), + encoding="utf-8", + ) + (capture / "mqtt.summary.json").write_text( + json.dumps( + { + "created_at_utc": "2026-07-16T20:56:32.699Z", + "completed_at_utc": "2026-07-16T20:57:02.699Z", + "capture_elapsed_seconds": 30.0, + "stop_reason": "external_stop", + "error": None, + "message_count": 2, + "raw_bytes": len(raw), + "topic_counts": {topic: 1 for topic in topics}, + "artifact_hashes": { + "raw_sha256": hashlib.sha256(raw).hexdigest(), + "metadata_jsonl_sha256": hashlib.sha256(metadata_path.read_bytes()).hexdigest(), + }, + } + ), + encoding="utf-8", + ) + return capture.parents[1] + + +def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]: + return next( + route.endpoint + for route in router.routes + if isinstance(route, APIRoute) and route.path == path and method in route.methods + ) + + +def test_list_refresh_discovers_new_completed_session_without_server_restart( + tmp_path: Path, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + store = SessionStore(repository, data_dir=tmp_path / "data") + refresh_calls = 0 + + def refresh() -> tuple[str, ...]: + nonlocal refresh_calls + refresh_calls += 1 + return store.import_legacy_viewer_live(sessions) + + router = build_session_router(store, catalog_refresher=refresh) + list_route = _endpoint(router, "/api/v1/observation-sessions", "GET") + + assert list_route(limit=20, cursor=None) == {"items": []} + session = _make_completed_session(sessions, "20260716T205632Z_viewer_live") + listing = list_route(limit=20, cursor=None) + + assert refresh_calls == 2 + assert [item["id"] for item in listing["items"]] == [session.name] + + +def test_detail_and_replay_refresh_catalog_before_lookup(tmp_path: Path) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = _make_completed_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + refresh_calls = 0 + + def refresh() -> tuple[str, ...]: + nonlocal refresh_calls + refresh_calls += 1 + return store.import_legacy_viewer_live(sessions) + + router = build_session_router(store, catalog_refresher=refresh) + detail_route = _endpoint( + router, + "/api/v1/observation-sessions/{session_id}", + "GET", + ) + replay_route = _endpoint( + router, + "/api/v1/observation-sessions/{session_id}/replay", + "POST", + ) + + detail = detail_route(session_id=session.name) + replay = asyncio.run(replay_route(session_id=session.name, request=None)) + + assert detail["session_id"] == session.name + assert replay["launch"]["session_id"] == session.name + assert refresh_calls == 2 + + +def test_catalog_refresh_failure_is_a_stable_service_error(tmp_path: Path) -> None: + store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data") + + def fail_refresh() -> None: + raise OSError("private local path must not escape") + + router = build_session_router(store, catalog_refresher=fail_refresh) + list_route = _endpoint(router, "/api/v1/observation-sessions", "GET") + + with pytest.raises(HTTPException) as error: + list_route(limit=20, cursor=None) + assert error.value.status_code == 503 + assert "private local path" not in str(error.value.detail) + + +def test_startup_warmup_enqueues_every_finalized_replayable_session( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = _make_completed_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.import_legacy_viewer_live(sessions) + + def exporter(source: Path, destination: Path) -> dict[str, object]: + payload = b"startup-prepared-recording" + destination.write_bytes(payload) + return { + "source_sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + "rrd_sha256": hashlib.sha256(payload).hexdigest(), + "rrd_bytes": len(payload), + "timeline": "session_time", + "timeline_start_ns": 0, + "timeline_end_ns": 1, + } + + manager = SessionRecordingPreparationManager( + SessionRecordingMaterializer(store.data_dir, exporter=exporter) + ) + monkeypatch.setattr(app_module, "session_store", store) + monkeypatch.setattr(app_module, "session_recording_preparation_manager", manager) + try: + enqueued = app_module.enqueue_replayable_recordings() + deadline = time.monotonic() + 2 + snapshot = manager.status(session.name) + while snapshot is not None and snapshot.state != "ready" and time.monotonic() < deadline: + time.sleep(0.005) + snapshot = manager.status(session.name) + + assert enqueued == (session.name,) + assert snapshot is not None + assert snapshot.state == "ready" + assert snapshot.recording is not None + finally: + manager.close() + + +def test_reconciliation_skips_one_stale_session_and_prepares_later_valid_session( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + good = _make_completed_session(sessions, "20260716T205632Z_viewer_live") + stale = _make_completed_session(sessions, "20260716T205633Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.import_legacy_viewer_live(sessions) + (stale / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt").unlink() + + def exporter(source: Path, destination: Path) -> dict[str, object]: + payload = b"prepared-after-stale-row" + destination.write_bytes(payload) + return { + "source_sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + "rrd_sha256": hashlib.sha256(payload).hexdigest(), + "rrd_bytes": len(payload), + "timeline": "session_time", + "timeline_start_ns": 0, + "timeline_end_ns": 1, + } + + manager = SessionRecordingPreparationManager( + SessionRecordingMaterializer(store.data_dir, exporter=exporter) + ) + monkeypatch.setattr(app_module, "session_store", store) + monkeypatch.setattr(app_module, "session_recording_preparation_manager", manager) + try: + enqueued = app_module.enqueue_replayable_recordings() + deadline = time.monotonic() + 2 + snapshot = manager.status(good.name) + while snapshot is not None and snapshot.state != "ready" and time.monotonic() < deadline: + time.sleep(0.005) + snapshot = manager.status(good.name) + + assert enqueued == (good.name,) + assert manager.status(stale.name) is None + assert snapshot is not None and snapshot.state == "ready" + finally: + manager.close() + + +def test_reconciler_requeues_only_a_restart_interrupted_job( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = _make_completed_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.import_legacy_viewer_live(sessions) + started = threading.Event() + release = threading.Event() + calls = 0 + + def exporter(source: Path, destination: Path) -> dict[str, object]: + nonlocal calls + calls += 1 + if calls == 1: + started.set() + assert release.wait(timeout=2) + payload = f"restart-{calls}".encode() + destination.write_bytes(payload) + return { + "source_sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + "rrd_sha256": hashlib.sha256(payload).hexdigest(), + "rrd_bytes": len(payload), + "timeline": "session_time", + "timeline_start_ns": 0, + "timeline_end_ns": 1, + } + + manager = SessionRecordingPreparationManager( + SessionRecordingMaterializer(store.data_dir, exporter=exporter) + ) + monkeypatch.setattr(app_module, "session_store", store) + monkeypatch.setattr(app_module, "session_recording_preparation_manager", manager) + try: + app_module.enqueue_replayable_recordings() + assert started.wait(timeout=1) + manager.close(timeout=0.001) + manager.start() + release.set() + + deadline = time.monotonic() + 2 + snapshot = manager.status(session.name) + while ( + snapshot is None or snapshot.state != "cancelled" + ) and time.monotonic() < deadline: + time.sleep(0.005) + snapshot = manager.status(session.name) + assert snapshot is not None and snapshot.state == "cancelled" + interrupted_id = snapshot.preparation_id + + app_module.enqueue_replayable_recordings() + deadline = time.monotonic() + 2 + while time.monotonic() < deadline: + snapshot = manager.status(session.name) + if snapshot is not None and snapshot.state == "ready": + break + time.sleep(0.005) + assert snapshot is not None and snapshot.state == "ready" + assert snapshot.preparation_id != interrupted_id + assert calls == 2 + finally: + release.set() + manager.close() + + +def test_reconciler_does_not_loop_retry_a_genuine_failed_job( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = _make_completed_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.import_legacy_viewer_live(sessions) + calls = 0 + + def exporter(_source: Path, _destination: Path) -> dict[str, object]: + nonlocal calls + calls += 1 + raise OSError("genuine export failure") + + manager = SessionRecordingPreparationManager( + SessionRecordingMaterializer(store.data_dir, exporter=exporter) + ) + monkeypatch.setattr(app_module, "session_store", store) + monkeypatch.setattr(app_module, "session_recording_preparation_manager", manager) + try: + app_module.enqueue_replayable_recordings() + deadline = time.monotonic() + 2 + snapshot = manager.status(session.name) + while ( + snapshot is None or snapshot.state != "failed" + ) and time.monotonic() < deadline: + time.sleep(0.005) + snapshot = manager.status(session.name) + assert snapshot is not None and snapshot.state == "failed" + failed_id = snapshot.preparation_id + + app_module.enqueue_replayable_recordings() + time.sleep(0.05) + unchanged = manager.status(session.name) + assert unchanged is not None + assert unchanged.preparation_id == failed_id + assert unchanged.state == "failed" + assert calls == 1 + finally: + manager.close() diff --git a/tests/test_xgrids_camera_gateway.py b/tests/test_xgrids_camera_gateway.py index d9579cb..720572e 100644 --- a/tests/test_xgrids_camera_gateway.py +++ b/tests/test_xgrids_camera_gateway.py @@ -1,12 +1,15 @@ from __future__ import annotations +import hashlib import json import sys +import time from io import BytesIO from pathlib import Path import pytest +import k1link.web.xgrids_k1_camera as camera_module from k1link.web.xgrids_k1_camera import ( CAMERA_MEDIA_TYPE, XgridsK1CameraGateway, @@ -40,6 +43,63 @@ def _fake_ffmpeg(tmp_path: Path) -> Path: return executable +def _burst_ffmpeg(tmp_path: Path) -> Path: + executable = tmp_path / "burst-ffmpeg" + executable.write_text( + f"#!{sys.executable}\n" + "import sys, time\n" + "def box(kind, payload=b''):\n" + " return (8 + len(payload)).to_bytes(4, 'big') + kind + payload\n" + "sys.stdout.buffer.write(box(b'ftyp', b'isom') + box(b'moov'))\n" + "sys.stdout.buffer.flush()\n" + "time.sleep(0.2)\n" + "for index in range(10):\n" + " sys.stdout.buffer.write(box(b'moof') + box(b'mdat', bytes([index])))\n" + " sys.stdout.buffer.flush()\n" + " time.sleep(0.01)\n" + "time.sleep(10)\n", + encoding="utf-8", + ) + executable.chmod(0o700) + return executable + + +def _buffered_tail_ffmpeg(tmp_path: Path, sentinel: Path, *, fragments: int) -> Path: + executable = tmp_path / "buffered-tail-ffmpeg" + executable.write_text( + f"#!{sys.executable}\n" + "import pathlib, sys, time\n" + "def box(kind, payload=b''):\n" + " return (8 + len(payload)).to_bytes(4, 'big') + kind + payload\n" + "sys.stdout.buffer.write(box(b'ftyp', b'isom') + box(b'moov'))\n" + f"for index in range({fragments}):\n" + " sys.stdout.buffer.write(box(b'moof') + box(b'mdat', index.to_bytes(2, 'big')))\n" + "sys.stdout.buffer.flush()\n" + f"pathlib.Path({str(sentinel)!r}).write_text('ready')\n" + "time.sleep(10)\n", + encoding="utf-8", + ) + executable.chmod(0o700) + return executable + + +def _incomplete_tail_ffmpeg(tmp_path: Path, sentinel: Path) -> Path: + executable = tmp_path / "incomplete-tail-ffmpeg" + executable.write_text( + f"#!{sys.executable}\n" + "import pathlib, sys, time\n" + "def box(kind, payload=b''):\n" + " return (8 + len(payload)).to_bytes(4, 'big') + kind + payload\n" + "sys.stdout.buffer.write(box(b'ftyp', b'isom') + box(b'moov') + box(b'moof'))\n" + "sys.stdout.buffer.flush()\n" + f"pathlib.Path({str(sentinel)!r}).write_text('ready')\n" + "time.sleep(10)\n", + encoding="utf-8", + ) + executable.chmod(0o700) + return executable + + def _gateway( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -48,6 +108,16 @@ def _gateway( return XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) +def _wait_until(predicate: object, *, timeout: float = 3.0) -> None: + assert callable(predicate) + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.01) + raise AssertionError("condition was not reached before timeout") + + def test_camera_selection_is_exclusive_and_hides_device_transport( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -90,9 +160,7 @@ def test_camera_ffmpeg_command_is_allowlisted_copy_remux() -> None: ) assert argv[0] == "/trusted/ffmpeg" - assert argv[argv.index("-i") + 1] == ( - "rtsp://10.0.0.24:8554/live/chn_left_main" - ) + assert argv[argv.index("-i") + 1] == ("rtsp://10.0.0.24:8554/live/chn_left_main") assert argv[argv.index("-c:v") + 1] == "copy" assert argv[argv.index("-allowed_media_types") + 1] == "video" assert argv[argv.index("-flush_packets") + 1] == "1" @@ -145,6 +213,279 @@ def test_gateway_emits_init_and_complete_media_segments_without_transcoding( gateway.close() +def test_acquisition_records_without_browser_and_source_switch_seals_epochs( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway = _gateway(tmp_path, monkeypatch) + session = tmp_path / "sessions" / "camera-acquisition" + try: + left = gateway.select("sensor.camera.left", "192.168.1.20") + with pytest.raises(ValueError, match="does not exist"): + gateway.start_recording(session) + assert session.exists() is False + session.mkdir(parents=True) + gateway.start_recording(session) + left_epoch = session / "media" / "sensor.camera.left" / f"epoch-{left['generation']}" + + # No open_delivery/WebSocket exists: acquisition ownership alone starts + # FFmpeg and commits the init segment before any preview consumer. + _wait_until( + lambda: ( + (left_epoch / "init.mp4").is_file() + and len(list((left_epoch / "segments").glob("*.m4s"))) == 1 + ) + ) + _wait_until(lambda: gateway.snapshot()["phase"] == "streaming") + + right = gateway.select("sensor.camera.right", "192.168.1.20") + right_epoch = session / "media" / "sensor.camera.right" / f"epoch-{right['generation']}" + _wait_until(lambda: gateway.snapshot()["phase"] == "streaming") + _wait_until(lambda: len(list((right_epoch / "segments").glob("*.m4s"))) == 1) + + lease = gateway.open_delivery(right["generation"]) + init_segment = lease.segments.get(timeout=1) + assert init_segment is not None and init_segment[0] == "init" + gateway.release_delivery(lease, client_closed=True) + + # Browser disposal is not producer disposal while recording is active. + assert lease.process.poll() is None + assert gateway.snapshot()["recording"]["active"] is True + assert gateway.snapshot()["phase"] == "streaming" + + stopped = gateway.stop_recording(status="complete") + assert stopped["recording"]["active"] is False + assert stopped["recording"]["completed_epochs"] == 2 + + summaries = [] + for epoch in (left_epoch, right_epoch): + init = (epoch / "init.mp4").read_bytes() + entries = [ + json.loads(line) + for line in (epoch / "index.jsonl").read_text(encoding="utf-8").splitlines() + ] + summary = json.loads((epoch / "summary.json").read_text(encoding="utf-8")) + summaries.append(summary) + assert [entry["kind"] for entry in entries] == ["media"] + assert all( + entry["length"] == len((epoch / entry["path"]).read_bytes()) + and hashlib.sha256((epoch / entry["path"]).read_bytes()).hexdigest() + == entry["sha256"] + for entry in entries + ) + assert summary["status"] == "complete" + assert summary["segment_count"] == 1 + assert summary["entry_count"] == 1 + assert summary["valid_bytes"] == len(init) + sum( + entry["length"] for entry in entries + ) + + assert summaries[0]["failure_code"] == "source-switch" + assert summaries[1]["failure_code"] is None + serialized = json.dumps(stopped) + assert "192.168.1.20" not in serialized + assert "rtsp://" not in serialized + finally: + gateway.close() + + +def test_camera_storage_open_failure_is_loud_and_never_starts_ffmpeg( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway = _gateway(tmp_path, monkeypatch) + session = tmp_path / "session" + session.mkdir() + popen_called = False + + class FailingArchive: + def __init__(self, *_: object, **__: object) -> None: + raise camera_module.CameraArchiveError("synthetic storage failure") + + def forbidden_popen(*_: object, **__: object) -> object: + nonlocal popen_called + popen_called = True + raise AssertionError("FFmpeg must not start without durable storage") + + monkeypatch.setattr(camera_module, "CameraArchiveWriter", FailingArchive) + monkeypatch.setattr(camera_module.subprocess, "Popen", forbidden_popen) + try: + gateway.select("sensor.camera.left", "192.168.1.20") + with pytest.raises(RuntimeError, match="хранилище"): + gateway.start_recording(session) + + state = gateway.snapshot() + assert popen_called is False + assert state["phase"] == "error" + assert state["error"]["code"] == "camera-storage-failed" + assert "192.168.1.20" not in json.dumps(state) + finally: + gateway.close() + + +def test_camera_storage_append_failure_fails_producer_and_epoch_loudly( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + original_append = camera_module.CameraArchiveWriter.append + + def failing_append( + writer: object, + kind: object, + payload: bytes, + **timestamps: object, + ) -> dict[str, object]: + if kind == "media": + raise camera_module.CameraArchiveError("synthetic media commit failure") + return original_append(writer, kind, payload, **timestamps) # type: ignore[arg-type] + + monkeypatch.setattr(camera_module.CameraArchiveWriter, "append", failing_append) + gateway = _gateway(tmp_path, monkeypatch) + session = tmp_path / "session" + session.mkdir() + try: + selected = gateway.select("sensor.camera.left", "192.168.1.20") + gateway.start_recording(session) + + _wait_until(lambda: gateway.snapshot()["phase"] == "error") + state = gateway.snapshot() + assert state["error"]["code"] == "camera-storage-failed" + summary_path = ( + session + / "media" + / "sensor.camera.left" + / f"epoch-{selected['generation']}" + / "summary.json" + ) + _wait_until( + lambda: ( + summary_path.is_file() + and json.loads(summary_path.read_text(encoding="utf-8"))["status"] + == "failed" + ) + ) + summary = json.loads(summary_path.read_text(encoding="utf-8")) + assert summary["status"] == "failed" + assert summary["failure_code"] == "camera-storage-failed" + assert "192.168.1.20" not in json.dumps(state) + finally: + gateway.close() + + +def test_slow_browser_is_dropped_without_stopping_archive_producer( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_burst_ffmpeg(tmp_path))) + gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) + session = tmp_path / "session" + session.mkdir() + try: + selected = gateway.select("sensor.camera.left", "192.168.1.20") + gateway.start_recording(session) + lease = gateway.open_delivery(selected["generation"]) + + # Deliberately never drain the bounded queue. + _wait_until(lambda: lease.failure_code == "consumer-too-slow") + _wait_until(lambda: gateway.snapshot()["phase"] == "streaming") + assert lease.process.poll() is None + assert gateway.snapshot()["recording"]["active"] is True + + gateway.release_delivery(lease, client_closed=False) + epoch = ( + session + / "media" + / "sensor.camera.left" + / f"epoch-{selected['generation']}" + ) + _wait_until( + lambda: len(list((epoch / "segments").glob("*.m4s"))) == 10, + ) + gateway.stop_recording(status="complete") + + summary_path = epoch / "summary.json" + summary = json.loads(summary_path.read_text(encoding="utf-8")) + assert summary["status"] == "complete" + assert summary["segment_count"] == 10 + assert summary["media_segment_count"] == 10 + finally: + gateway.close() + + +def test_clean_stop_drains_ffmpeg_stdout_before_sealing_archive( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fragment_count = 100 + sentinel = tmp_path / "ffmpeg-wrote-buffered-tail" + monkeypatch.setenv( + "MISSIONCORE_FFMPEG_BINARY", + str(_buffered_tail_ffmpeg(tmp_path, sentinel, fragments=fragment_count)), + ) + gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) + session = tmp_path / "session" + session.mkdir() + try: + selected = gateway.select("sensor.camera.left", "192.168.1.20") + gateway.start_recording(session) + _wait_until(sentinel.is_file) + + gateway.stop_recording(status="complete") + + epoch = ( + session + / "media" + / "sensor.camera.left" + / f"epoch-{selected['generation']}" + ) + summary = json.loads((epoch / "summary.json").read_text(encoding="utf-8")) + assert summary["status"] == "complete" + assert summary["segment_count"] == fragment_count + assert len(list((epoch / "segments").glob("*.m4s"))) == fragment_count + assert len((epoch / "index.jsonl").read_text(encoding="utf-8").splitlines()) == ( + fragment_count + ) + finally: + gateway.close() + + +def test_clean_stop_marks_incomplete_fragment_tail_interrupted( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + sentinel = tmp_path / "ffmpeg-wrote-incomplete-tail" + monkeypatch.setenv( + "MISSIONCORE_FFMPEG_BINARY", + str(_incomplete_tail_ffmpeg(tmp_path, sentinel)), + ) + gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) + session = tmp_path / "session" + session.mkdir() + try: + selected = gateway.select("sensor.camera.left", "192.168.1.20") + gateway.start_recording(session) + _wait_until(sentinel.is_file) + + gateway.stop_recording(status="complete") + + summary_path = ( + session + / "media" + / "sensor.camera.left" + / f"epoch-{selected['generation']}" + / "summary.json" + ) + summary = json.loads(summary_path.read_text(encoding="utf-8")) + assert summary["status"] == "interrupted" + assert summary["failure_code"] == "incomplete-fmp4-fragment" + assert summary["segment_count"] == 0 + state = gateway.snapshot() + assert state["phase"] == "error" + assert state["error"]["code"] == "incomplete-fmp4-fragment" + finally: + gateway.close() + + def test_service_publishes_two_dynamic_camera_rows_and_stale_stop_is_safe( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,