from __future__ import annotations import asyncio import hashlib import json import sys import threading import time from io import BytesIO from pathlib import Path import pytest import k1link.device_plugins.xgrids_k1.camera as camera_module from k1link.device_plugins.xgrids_k1.camera import ( CAMERA_MEDIA_TYPE, CAMERA_PREVIEW_SEND_TIMEOUT_SECONDS, CAMERA_SOURCE_IO_TIMEOUT_MICROSECONDS, MAX_CAMERA_PREVIEW_CONSUMERS, MAX_CAMERA_PREVIEW_QUEUE_AGE_SECONDS, MAX_CAMERA_PREVIEW_QUEUED_BYTES, MAX_CAMERA_PREVIEW_QUEUED_SEGMENTS, MAX_FMP4_SEGMENT_BYTES, CommittedCameraSegment, XgridsK1CameraGateway, _build_ffmpeg_argv, _CameraPreviewSegmentQueue, _read_mp4_box, build_xgrids_k1_camera_router, classify_camera_recording_health, ) from k1link.device_plugins.xgrids_k1.connection_supervisor import ( EndpointTarget, HostPathProbeResult, VerifiedControlEvidence, ) from k1link.device_plugins.xgrids_k1.facade import ( XGRIDS_K1_COMPATIBILITY_PROFILE_ID, XGRIDS_K1_PLUGIN_ID, CameraPreviewSelectRequest, CameraPreviewStopRequest, XgridsK1CompatibilityService, ) def _fake_ffmpeg(tmp_path: Path) -> Path: executable = tmp_path / "fake-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" "payload = (box(b'ftyp', b'isom') + box(b'moov') + " "box(b'moof') + box(b'mdat', b'frame'))\n" "sys.stdout.buffer.write(payload)\n" "sys.stdout.buffer.flush()\n" "time.sleep(10)\n", encoding="utf-8", ) executable.chmod(0o700) 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 _preview_buffer_ffmpeg( tmp_path: Path, *, fragments: int, payload_bytes: int, interval_seconds: float = 0.002, ) -> Path: executable = tmp_path / f"preview-buffer-{fragments}-{payload_bytes}" 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" f"payload = bytes({payload_bytes})\n" f"for index in range({fragments}):\n" " sys.stdout.buffer.write(box(b'moof') + box(b'mdat', payload))\n" " sys.stdout.buffer.flush()\n" f" time.sleep({interval_seconds!r})\n" "time.sleep(10)\n", encoding="utf-8", ) executable.chmod(0o700) return executable def _single_fragment_ffmpeg(tmp_path: Path, *, complete_fragment_bytes: int) -> Path: assert complete_fragment_bytes >= 16 executable = tmp_path / f"single-fragment-{complete_fragment_bytes}" 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" f"payload_bytes = {complete_fragment_bytes} - 16\n" "payload = b'PRIVATE-FRAME-BYTES' + bytes(payload_bytes - 19)\n" "sys.stdout.buffer.write(box(b'moof') + box(b'mdat', payload))\n" "sys.stdout.buffer.flush()\n" "time.sleep(10)\n", encoding="utf-8", ) executable.chmod(0o700) return executable def _silent_ffmpeg(tmp_path: Path) -> Path: executable = tmp_path / "silent-ffmpeg" executable.write_text( f"#!{sys.executable}\n" "import time\n" "time.sleep(10)\n", encoding="utf-8", ) executable.chmod(0o700) return executable def _clean_source_end_ffmpeg(tmp_path: Path) -> Path: executable = tmp_path / "clean-source-end-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.write(box(b'moof') + box(b'mdat', b'final-frame'))\n" "sys.stdout.buffer.flush()\n" "time.sleep(0.15)\n", encoding="utf-8", ) executable.chmod(0o700) return executable def _recoverable_source_end_ffmpeg(tmp_path: Path) -> tuple[Path, Path]: executable = tmp_path / "recoverable-source-end-ffmpeg" invocation_count = tmp_path / "recoverable-source-end-count" 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" f"counter = pathlib.Path({str(invocation_count)!r})\n" "count = int(counter.read_text()) + 1 if counter.exists() else 1\n" "counter.write_text(str(count))\n" "sys.stdout.buffer.write(box(b'ftyp', b'isom') + box(b'moov'))\n" "sys.stdout.buffer.write(box(b'moof') + box(b'mdat', b'frame'))\n" "sys.stdout.buffer.flush()\n" "time.sleep(0.15 if count == 1 else 10)\n", encoding="utf-8", ) executable.chmod(0o700) return executable, invocation_count def _gated_media_progress_ffmpeg( tmp_path: Path, *, release_second_media: Path, ) -> Path: executable = tmp_path / "gated-media-progress-ffmpeg" first_media_ready = tmp_path / "gated-media-progress-first-ready" 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" "sys.stdout.buffer.write(box(b'moof') + box(b'mdat', b'first'))\n" "sys.stdout.buffer.flush()\n" f"pathlib.Path({str(first_media_ready)!r}).write_text('ready')\n" f"release = pathlib.Path({str(release_second_media)!r})\n" "deadline = time.monotonic() + 5\n" "while not release.exists() and time.monotonic() < deadline:\n" " time.sleep(0.005)\n" "if release.exists():\n" " sys.stdout.buffer.write(box(b'moof') + box(b'mdat', b'second'))\n" " sys.stdout.buffer.flush()\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, ) -> XgridsK1CameraGateway: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path))) 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, ) -> None: gateway = _gateway(tmp_path, monkeypatch) try: left = gateway.select("sensor.camera.left", "192.168.8.52") assert left["active_source_id"] == "sensor.camera.left" assert left["generation"] == 1 assert left["activation"]["max_active"] == 1 assert left["delivery"]["kind"] == "mse-fmp4-websocket" assert left["delivery"]["media_type"] == CAMERA_MEDIA_TYPE serialized = json.dumps(left) assert "192.168.8.52" not in serialized assert "rtsp://" not in serialized assert "chn_left_main" not in serialized right = gateway.select("sensor.camera.right", "192.168.8.52") assert right["active_source_id"] == "sensor.camera.right" assert right["generation"] == 2 assert right["delivery"]["url"].endswith("/camera-preview/2") with pytest.raises(ValueError, match="generation"): gateway.stop(1) assert gateway.snapshot()["active_source_id"] == "sensor.camera.right" stopped = gateway.stop(2) assert stopped["phase"] == "idle" assert stopped["delivery"] is None finally: gateway.close() def test_camera_derived_observer_runs_only_after_durable_archive_commit( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path))) session = tmp_path / "session" session.mkdir() observed: list[CommittedCameraSegment] = [] def observe(segment: CommittedCameraSegment) -> None: archive = session / "media" / segment.source_id / f"epoch-{segment.generation}" committed = ( archive / "init.mp4" if segment.kind == "init" else archive / "segments" / f"{segment.sequence}.m4s" ) assert committed.read_bytes() == segment.payload observed.append(segment) gateway = XgridsK1CameraGateway( tmp_path, XGRIDS_K1_PLUGIN_ID, committed_segment_observer=observe, ) try: gateway.start_recording(session) gateway.select("sensor.camera.right", "192.168.8.52") _wait_until(lambda: len(observed) == 2) assert [segment.kind for segment in observed] == ["init", "media"] assert gateway.snapshot()["derived_observer_errors"] == 0 finally: gateway.close() def test_live_perception_rejects_late_camera_generation_after_replacement( tmp_path: Path, ) -> None: service = XgridsK1CompatibilityService(tmp_path) try: service.live_perception_ingress.begin_session("acquisition-new") service._live_perception_camera_binding = ( # noqa: SLF001 "acquisition-new", "sensor.camera.right", 12, ) stale = CommittedCameraSegment( source_id="sensor.camera.right", generation=11, kind="media", sequence=4, host_epoch_ns=10, host_monotonic_ns=20, payload=b"stale", ) current = CommittedCameraSegment( source_id="sensor.camera.right", generation=12, kind="media", sequence=1, host_epoch_ns=30, host_monotonic_ns=40, payload=b"current", ) service._observe_committed_camera_segment(stale) # noqa: SLF001 after_stale = service.live_perception_ingress.snapshot() assert after_stale["queues"]["camera-frame"]["published"] == 0 assert after_stale["queues"]["camera-frame"]["depth"] == 0 service._observe_committed_camera_segment(current) # noqa: SLF001 after_current = service.live_perception_ingress.snapshot() assert after_current["queues"]["camera-frame"]["published"] == 1 assert after_current["queues"]["camera-frame"]["depth"] == 1 finally: service.close() def test_camera_ffmpeg_command_is_allowlisted_copy_remux() -> None: argv = _build_ffmpeg_argv( Path("/trusted/ffmpeg"), "10.0.0.24", "sensor.camera.left", ) 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("-c:v") + 1] == "copy" assert argv[argv.index("-allowed_media_types") + 1] == "video" assert argv[argv.index("-timeout") + 1] == str(CAMERA_SOURCE_IO_TIMEOUT_MICROSECONDS) assert argv[argv.index("-flush_packets") + 1] == "1" assert "-c:v" in argv assert ";" not in " ".join(argv) with pytest.raises(ValueError, match="private IPv4"): _build_ffmpeg_argv( Path("/trusted/ffmpeg"), "example.com", "sensor.camera.left", ) def test_iso_bmff_reader_preserves_complete_boxes() -> None: def box(kind: bytes, payload: bytes = b"") -> bytes: return (8 + len(payload)).to_bytes(4, "big") + kind + payload stream = BytesIO(box(b"ftyp", b"isom") + box(b"moov") + box(b"moof")) assert _read_mp4_box(stream) == (b"ftyp", box(b"ftyp", b"isom")) assert _read_mp4_box(stream) == (b"moov", box(b"moov")) assert _read_mp4_box(stream) == (b"moof", box(b"moof")) oversized = (9 * 1024 * 1024).to_bytes(4, "big") + b"mdat" with pytest.raises(ValueError, match="unbounded"): _read_mp4_box(BytesIO(oversized)) def test_current_epoch_is_not_media_ready_from_popen_or_epoch_allocation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path))) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) session = tmp_path / "sessions" / "camera-pending-media" session.mkdir(parents=True) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") first = gateway.start_recording(session) assert first["phase"] == "connecting" assert first["recording"]["active_epoch"] == selected["generation"] assert first["recording"]["media_ready"] is False assert first["recording"]["current_epoch"] == { "generation": selected["generation"], "init_committed": False, "init_committed_age_ms": None, "first_media_committed": False, "first_media_committed_age_ms": None, "committed_media_segment_count": 0, "last_media_segment_age_ms": None, } restarted = gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=selected["generation"], expected_recording_session=session.name, expected_active_epoch=selected["generation"], expected_recording_media_segment_count=0, pre_detach_fence=lambda reserve: reserve(), ) replacement = selected["generation"] + 1 assert restarted["generation"] == replacement assert restarted["recording"]["active_epoch"] == replacement assert restarted["recording"]["media_ready"] is False assert restarted["recording"]["current_epoch"]["generation"] == replacement assert restarted["recording"]["current_epoch"]["init_committed"] is False assert restarted["recording"]["current_epoch"]["first_media_committed"] is False finally: gateway.close() def test_camera_health_classifier_splits_no_init_and_no_media_stalls() -> None: base = { "phase": "connecting", "recording": { "active": True, "producer_age_ms": 10_000, "current_epoch": { "generation": 4, "init_committed": False, "init_committed_age_ms": None, "first_media_committed": False, }, }, "error": None, } assert classify_camera_recording_health(base) == "camera-connecting-no-init-stalled" current = base["recording"]["current_epoch"] current.update({"init_committed": True, "init_committed_age_ms": 10_000}) assert classify_camera_recording_health(base) == "camera-connecting-no-media-stalled" current["first_media_committed"] = True assert classify_camera_recording_health(base) is None def test_backend_camera_watchdog_reports_stall_without_browser_polling( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path))) observed: list[tuple[str, int]] = [] wake = threading.Event() def observe(reason: str, generation: int) -> None: observed.append((reason, generation)) wake.set() gateway = XgridsK1CameraGateway( tmp_path, XGRIDS_K1_PLUGIN_ID, producer_stall_observer=observe, producer_stall_milliseconds=30, producer_watchdog_interval_seconds=0.01, ) session = tmp_path / "sessions" / "camera-watchdog" session.mkdir(parents=True) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) assert wake.wait(timeout=1.0) is True assert observed == [ ("camera-connecting-no-init-stalled", selected["generation"]) ] time.sleep(0.05) assert len(observed) == 1 finally: gateway.close() def test_fragment_above_one_mib_is_durably_committed_and_media_ready( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: complete_fragment_bytes = 2 * 1024 * 1024 monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str( _single_fragment_ffmpeg( tmp_path, complete_fragment_bytes=complete_fragment_bytes, ) ), ) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) session = tmp_path / "sessions" / "camera-large-valid-fragment" session.mkdir(parents=True) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) _wait_until(lambda: gateway.snapshot()["recording"]["media_ready"] is True) snapshot = gateway.snapshot() current = snapshot["recording"]["current_epoch"] assert current["generation"] == selected["generation"] assert current["init_committed"] is True assert current["first_media_committed"] is True assert current["committed_media_segment_count"] == 1 assert isinstance(current["init_committed_age_ms"], int) assert isinstance(current["first_media_committed_age_ms"], int) segment = ( session / "media" / "sensor.camera.right" / f"epoch-{selected['generation']}" / "segments" / "1.m4s" ) assert segment.stat().st_size == complete_fragment_bytes finally: gateway.close() def test_fragment_above_eight_mib_fails_closed_with_size_only_diagnostic( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: observed_bytes = MAX_FMP4_SEGMENT_BYTES + 1 monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str( _single_fragment_ffmpeg( tmp_path, complete_fragment_bytes=observed_bytes, ) ), ) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) session = tmp_path / "sessions" / "camera-oversized-fragment" session.mkdir(parents=True) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) _wait_until( lambda: ( gateway.snapshot()["phase"] == "error" and gateway.snapshot()["recording"]["completed_epochs"] == 1 ) ) snapshot = gateway.snapshot() assert snapshot["error"] == { "code": "segment-too-large", "message": "Camera adapter отклонил слишком большой video segment.", "observed_size_bytes": observed_bytes, "maximum_size_bytes": MAX_FMP4_SEGMENT_BYTES, } assert snapshot["recording"]["committed_media_segment_count"] == 0 serialized = json.dumps(snapshot) assert "PRIVATE-FRAME-BYTES" not in serialized epoch = ( session / "media" / "sensor.camera.right" / f"epoch-{selected['generation']}" ) assert list((epoch / "segments").glob("*.m4s")) == [] summary = json.loads((epoch / "summary.json").read_text(encoding="utf-8")) assert summary["failure_code"] == "segment-too-large" finally: gateway.close() def test_gateway_emits_init_and_complete_media_segments_without_transcoding( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) try: state = gateway.select("sensor.camera.right", "192.168.1.20") lease = gateway.open_delivery(state["generation"]) init_segment = lease.segments.get(timeout=3) media_segment = lease.segments.get(timeout=3) assert init_segment is not None and init_segment[0] == "init" assert b"ftyp" in init_segment[1] and b"moov" in init_segment[1] assert media_segment is not None and media_segment[0] == "media" assert b"moof" in media_segment[1] and b"mdat" in media_segment[1] gateway.mark_streaming(lease) assert gateway.snapshot()["phase"] == "streaming" gateway.release_delivery(lease, client_closed=True) assert gateway.snapshot()["phase"] == "selected" finally: 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_acquisition_allows_independent_browser_consumers_without_lease_churn( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """React/window replacement cannot starve the currently visible camera.""" gateway = _gateway(tmp_path, monkeypatch) session = tmp_path / "sessions" / "camera-browser-consumers" session.mkdir(parents=True) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) first = gateway.open_delivery(selected["generation"]) second = gateway.open_delivery(selected["generation"]) assert first.segments.get(timeout=3)[0] == "init" assert second.segments.get(timeout=3)[0] == "init" assert gateway.snapshot()["recording"]["preview_consumer_count"] == 2 gateway.release_delivery(first, client_closed=True) assert gateway.snapshot()["recording"]["preview_consumer_count"] == 1 assert second.process.poll() is None assert second.failure_code is None gateway.release_delivery(second, client_closed=True) state = gateway.snapshot() assert state["recording"]["preview_consumer_count"] == 0 assert state["recording"]["active"] is True assert state["recording"]["producer_alive"] is True finally: gateway.close() def test_release_delivery_wakes_blocked_reader_without_stopping_recording( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) session = tmp_path / "released-browser-reader" session.mkdir(parents=True) released: list[tuple[str, bytes] | None] = [] reader_started = threading.Event() try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) lease = gateway.open_delivery(selected["generation"]) assert lease.segments.get(timeout=3)[0] == "init" assert lease.segments.get(timeout=3)[0] == "media" def wait_for_next_segment() -> None: reader_started.set() released.append(lease.segments.get()) reader = threading.Thread(target=wait_for_next_segment) reader.start() assert reader_started.wait(timeout=1) time.sleep(0.02) assert released == [] gateway.release_delivery(lease, client_closed=True) reader.join(timeout=1) assert reader.is_alive() is False assert released == [None] recording = gateway.snapshot()["recording"] assert recording["active"] is True assert recording["producer_alive"] is True assert recording["preview_consumer_count"] == 0 finally: gateway.close() def test_preview_queue_enforces_exact_fragment_byte_and_age_bounds() -> None: now = [10.0] fragment_bounded = _CameraPreviewSegmentQueue(clock=lambda: now[0]) for index in range(MAX_CAMERA_PREVIEW_QUEUED_SEGMENTS): assert fragment_bounded.offer(("media", bytes([index % 256]))) is True assert fragment_bounded.offer(("media", b"overflow")) is False byte_bounded = _CameraPreviewSegmentQueue(clock=lambda: now[0]) one_mebibyte = bytes(1024 * 1024) for _ in range(MAX_CAMERA_PREVIEW_QUEUED_BYTES // len(one_mebibyte)): assert byte_bounded.offer(("media", one_mebibyte)) is True assert byte_bounded.queued_bytes == MAX_CAMERA_PREVIEW_QUEUED_BYTES assert byte_bounded.offer(("media", b"x")) is False valid_maximum = _CameraPreviewSegmentQueue() assert valid_maximum.offer(("init", bytes(512 * 1024))) is True assert valid_maximum.offer(("media", bytes(MAX_FMP4_SEGMENT_BYTES))) is True assert valid_maximum.queued_bytes == MAX_FMP4_SEGMENT_BYTES + 512 * 1024 age_bounded = _CameraPreviewSegmentQueue(clock=lambda: now[0]) assert age_bounded.offer(("init", b"init")) is True now[0] += MAX_CAMERA_PREVIEW_QUEUE_AGE_SECONDS assert age_bounded.offer(("media", b"at-boundary")) is True now[0] += 0.001 assert age_bounded.offer(("media", b"too-old")) is False def test_preview_queue_survives_measured_ui_pause() -> None: now = [100.0] preview_queue = _CameraPreviewSegmentQueue(clock=lambda: now[0]) assert preview_queue.offer(("media", b"first")) is True now[0] += 3.2 assert preview_queue.offer(("media", b"next")) is True def test_camera_router_closes_lagging_preview_with_explicit_private_code( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) session = tmp_path / "router-send-timeout" session.mkdir() selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) _wait_until(lambda: gateway.snapshot()["recording"]["active_epoch"] is not None) class SlowWebSocket: def __init__(self) -> None: self.accepted = False self.close_code: int | None = None self.close_reason: str | None = None async def accept(self) -> None: self.accepted = True async def send_bytes(self, _payload: bytes) -> None: await asyncio.sleep(CAMERA_PREVIEW_SEND_TIMEOUT_SECONDS * 4) async def close(self, *, code: int = 1000, reason: str = "") -> None: self.close_code = code self.close_reason = reason websocket = SlowWebSocket() route = build_xgrids_k1_camera_router(gateway, XGRIDS_K1_PLUGIN_ID).routes[0] monkeypatch.setattr(camera_module, "CAMERA_PREVIEW_SEND_TIMEOUT_SECONDS", 0.01) try: asyncio.run(route.endpoint(websocket, selected["generation"])) assert websocket.accepted is True assert websocket.close_code == 4008 assert websocket.close_reason == "preview-consumer-lagged" recording = gateway.snapshot()["recording"] assert recording["preview_consumer_count"] == 0 assert recording["producer_alive"] is True finally: gateway.close() def test_preview_consumer_cardinality_is_bounded_and_newest_keeps_streaming( 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 / "sessions" / "bounded-browser-consumers" session.mkdir(parents=True) leases = [] try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) _wait_until(lambda: gateway.snapshot()["recording"]["active_epoch"] is not None) for _ in range(MAX_CAMERA_PREVIEW_CONSUMERS + 3): leases.append(gateway.open_delivery(selected["generation"])) state = gateway.snapshot() assert state["recording"]["preview_consumer_count"] == MAX_CAMERA_PREVIEW_CONSUMERS assert all(lease.failure_code == "consumer-superseded" for lease in leases[:3]) newest = leases[-1] assert newest.segments.get(timeout=3)[0] == "init" assert newest.segments.get(timeout=3)[0] == "media" assert newest.failure_code is None assert newest.process.poll() is None recording = gateway.snapshot()["recording"] assert recording["producer_alive"] is True assert recording["committed_media_segment_count"] >= 1 finally: for lease in leases: gateway.release_delivery(lease, client_closed=True) gateway.close() def test_acquisition_recovery_cas_restarts_stalled_ffmpeg_and_preserves_old_epoch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) session = tmp_path / "sessions" / "camera-recovery" session.mkdir(parents=True) try: gateway.start_recording(session) selected = gateway.select("sensor.camera.right", "192.168.1.20") first_generation = selected["generation"] first_epoch = session / "media" / "sensor.camera.right" / f"epoch-{first_generation}" _wait_until(lambda: gateway.snapshot()["phase"] == "streaming") _wait_until(lambda: (first_epoch / "init.mp4").is_file()) before = gateway.snapshot() assert before["recording"]["producer_alive"] is True assert before["recording"]["active_epoch"] == first_generation restarted = gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=before["generation"], expected_recording_session=session.name, expected_active_epoch=before["recording"]["active_epoch"], expected_recording_media_segment_count=before["recording"][ "committed_media_segment_count" ], pre_detach_fence=lambda reserve: reserve(), ) assert restarted["generation"] == first_generation + 1 assert restarted["recording"]["active_epoch"] == first_generation + 1 assert restarted["recording"]["completed_epochs"] == 1 first_summary = json.loads((first_epoch / "summary.json").read_text(encoding="utf-8")) assert first_summary["status"] == "interrupted" assert first_summary["failure_code"] == "active-stream-connection-recovery" second_epoch = session / "media" / "sensor.camera.right" / f"epoch-{first_generation + 1}" _wait_until(lambda: (second_epoch / "init.mp4").is_file()) with pytest.raises(ValueError, match="устарело|lineage"): gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=before["generation"], expected_recording_session=session.name, expected_active_epoch=before["recording"]["active_epoch"], expected_recording_media_segment_count=before["recording"][ "committed_media_segment_count" ], pre_detach_fence=lambda reserve: reserve(), ) assert gateway.snapshot()["generation"] == first_generation + 1 finally: gateway.close() def test_stop_first_pre_detach_denial_leaves_g7_and_archive_bytes_unchanged( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) session = tmp_path / "sessions" / "camera-stop-first-reservation" session.mkdir(parents=True) try: # Build a nontrivial lineage so the test proves the exact caller CAS, # not an incidental initial-generation special case. for source_id in ( "sensor.camera.right", "sensor.camera.left", "sensor.camera.right", "sensor.camera.left", "sensor.camera.right", "sensor.camera.left", "sensor.camera.right", ): selected = gateway.select(source_id, "192.168.1.20") assert selected["generation"] == 7 gateway.start_recording(session) _wait_until(lambda: gateway.snapshot()["recording"]["media_ready"] is True) before = gateway.snapshot() before_recording = before["recording"] before_files = { path.relative_to(session).as_posix(): path.read_bytes() for path in session.rglob("*") if path.is_file() } with pytest.raises(ValueError, match="устарела|lineage"): gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=7, expected_recording_session=session.name, expected_active_epoch=7, expected_recording_media_segment_count=before_recording[ "committed_media_segment_count" ], pre_detach_fence=lambda _reserve: False, ) after = gateway.snapshot() after_recording = after["recording"] assert after["generation"] == before["generation"] == 7 assert after["revision"] == before["revision"] assert after["phase"] == before["phase"] == "streaming" assert after["error"] == before["error"] is None assert after_recording["active_epoch"] == before_recording["active_epoch"] == 7 assert after_recording["producer_alive"] is True assert after_recording["completed_epochs"] == before_recording["completed_epochs"] assert after_recording["committed_media_segment_count"] == before_recording[ "committed_media_segment_count" ] assert { path.relative_to(session).as_posix(): path.read_bytes() for path in session.rglob("*") if path.is_file() } == before_files finally: gateway.close() def test_fresh_durable_media_supersedes_stale_watchdog_restart_cas( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: release_second_media = tmp_path / "release-second-camera-media" monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str( _gated_media_progress_ffmpeg( tmp_path, release_second_media=release_second_media, ) ), ) session = tmp_path / "sessions" / "camera-progress-cas" session.mkdir(parents=True) watchdog_snapshotted = threading.Event() allow_stale_cas = threading.Event() cas_done = threading.Event() cas_rejections: list[str] = [] observer_errors: list[Exception] = [] gateway: XgridsK1CameraGateway | None = None def backend_watchdog(reason: str, generation: int) -> None: if reason != "camera-stream-stalled": return try: assert gateway is not None stale = gateway.snapshot() stale_recording = stale["recording"] assert stale_recording["committed_media_segment_count"] == 1 watchdog_snapshotted.set() assert allow_stale_cas.wait(timeout=3) try: gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=generation, expected_recording_session=session.name, expected_active_epoch=stale_recording["active_epoch"], expected_recording_media_segment_count=stale_recording[ "committed_media_segment_count" ], pre_detach_fence=lambda reserve: reserve(), ) except ValueError as exc: cas_rejections.append(str(exc)) else: raise AssertionError("fresh durable media must supersede stale camera CAS") except Exception as exc: observer_errors.append(exc) finally: cas_done.set() gateway = XgridsK1CameraGateway( tmp_path, XGRIDS_K1_PLUGIN_ID, producer_stall_observer=backend_watchdog, producer_stall_milliseconds=40, producer_watchdog_interval_seconds=0.01, ) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) _wait_until( lambda: gateway is not None and gateway.snapshot()["recording"]["committed_media_segment_count"] == 1 ) assert watchdog_snapshotted.wait(timeout=3) # This archive append lands after the watchdog's health snapshot but # before its lifecycle-CAS entry. Generation and active epoch remain # unchanged; only the durable progress token can fence the stale restart. release_second_media.write_text("release", encoding="utf-8") _wait_until( lambda: gateway is not None and gateway.snapshot()["recording"]["committed_media_segment_count"] == 2 ) allow_stale_cas.set() assert cas_done.wait(timeout=3) current = gateway.snapshot() assert observer_errors == [] assert len(cas_rejections) == 1 assert "lineage" in cas_rejections[0] assert current["generation"] == selected["generation"] assert current["phase"] == "streaming" assert current["error"] is None assert current["recording"]["producer_alive"] is True assert current["recording"]["active_epoch"] == selected["generation"] assert current["recording"]["committed_media_segment_count"] == 2 assert current["recording"]["completed_epochs"] == 0 epoch = ( session / "media" / "sensor.camera.right" / f"epoch-{selected['generation']}" ) assert len(list((epoch / "segments").glob("*.m4s"))) == 2 finally: allow_stale_cas.set() gateway.close() def test_partial_same_session_activation_can_retry_exact_local_producer( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = _gateway(tmp_path, monkeypatch) session = tmp_path / "sessions" / "camera-partial-activation" session.mkdir(parents=True) real_popen = camera_module.subprocess.Popen spawn_attempts = 0 def fail_first_popen(*args: object, **kwargs: object) -> object: nonlocal spawn_attempts spawn_attempts += 1 if spawn_attempts == 1: raise OSError("synthetic first Popen failure") return real_popen(*args, **kwargs) monkeypatch.setattr(camera_module.subprocess, "Popen", fail_first_popen) try: gateway.select("sensor.camera.right", "192.168.1.20") with pytest.raises(RuntimeError, match="camera adapter"): gateway.start_recording(session) partial = gateway.snapshot() assert partial["phase"] == "error" assert partial["active_source_id"] == "sensor.camera.right" assert partial["recording"]["active"] is True assert partial["recording"]["session"] == session.name assert partial["recording"]["active_epoch"] is None failed_epoch = session / "media" / "sensor.camera.right" / f"epoch-{partial['generation']}" failed_summary = json.loads((failed_epoch / "summary.json").read_text(encoding="utf-8")) assert failed_summary["status"] == "failed" assert failed_summary["failure_code"] == "ffmpeg-start-failed" retried = gateway.retry_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=partial["generation"], expected_recording_session=session.name, pre_retry_fence=lambda reserve: reserve(), ) assert spawn_attempts == 2 assert retried["phase"] in {"connecting", "streaming"} assert retried["generation"] == partial["generation"] + 1 assert retried["recording"]["active_epoch"] == partial["generation"] + 1 assert retried["recording"]["producer_alive"] is True assert json.loads((failed_epoch / "summary.json").read_text(encoding="utf-8")) == ( failed_summary ) with pytest.raises(ValueError, match="lineage"): gateway.retry_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=partial["generation"], expected_recording_session=session.name, pre_retry_fence=lambda reserve: reserve(), ) finally: gateway.close() def test_expected_camera_source_end_during_device_stop_seals_complete_epoch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str(_clean_source_end_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) gateway.expect_source_end_for_device_stop() _wait_until( lambda: gateway.snapshot()["recording"]["completed_epochs"] == 1, ) state = gateway.snapshot() assert state["phase"] == "idle" assert state["error"] is None assert state["delivery"] is None assert state["recording"]["active"] is True assert state["recording"]["source_end_expected"] is False 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["failure_code"] is None assert summary["media_segment_count"] == 1 finally: gateway.close() def test_unexpected_camera_source_end_remains_a_recording_failure( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str(_clean_source_end_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) _wait_until( lambda: gateway.snapshot()["recording"]["completed_epochs"] == 1, ) state = gateway.snapshot() assert state["phase"] == "error" assert state["error"]["code"] == "camera-source-ended" epoch = session / "media" / "sensor.camera.left" / f"epoch-{selected['generation']}" summary = json.loads((epoch / "summary.json").read_text(encoding="utf-8")) assert summary["status"] == "interrupted" assert summary["failure_code"] == "camera-source-ended" finally: gateway.close() def test_browser_reconnect_cannot_respawn_sealed_epoch_before_backend_cas( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: executable, invocation_count = _recoverable_source_end_ffmpeg(tmp_path) monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(executable)) session = tmp_path / "sessions" / "camera-browser-backend-recovery-race" session.mkdir(parents=True) watchdog_entered = threading.Event() allow_backend_cas = threading.Event() backend_cas_done = threading.Event() observer_errors: list[Exception] = [] recovered_snapshots: list[dict[str, object]] = [] gateway: XgridsK1CameraGateway | None = None def backend_watchdog(reason: str, generation: int) -> None: try: assert reason == "camera-source-ended" watchdog_entered.set() assert allow_backend_cas.wait(timeout=3) assert gateway is not None recovered_snapshots.append( gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=generation, expected_recording_session=session.name, expected_active_epoch=None, expected_recording_media_segment_count=1, pre_detach_fence=lambda reserve: reserve(), ) ) except Exception as exc: observer_errors.append(exc) finally: backend_cas_done.set() gateway = XgridsK1CameraGateway( tmp_path, XGRIDS_K1_PLUGIN_ID, producer_stall_observer=backend_watchdog, ) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") first_generation = selected["generation"] gateway.start_recording(session) assert watchdog_entered.wait(timeout=3) sealed = gateway.snapshot() first_epoch = ( session / "media" / "sensor.camera.right" / f"epoch-{first_generation}" ) first_summary = json.loads( (first_epoch / "summary.json").read_text(encoding="utf-8") ) assert invocation_count.read_text(encoding="utf-8") == "1" assert sealed["generation"] == first_generation assert sealed["phase"] == "error" assert sealed["error"]["code"] == "camera-source-ended" assert sealed["recording"]["producer_alive"] is False assert sealed["recording"]["active_epoch"] is None assert sealed["recording"]["completed_epochs"] == 1 # A browser reconnect reaches the sealed generation before the backend # observer is allowed to take its CAS. It must neither spawn FFmpeg nor # reopen the finalized epoch or misclassify the collision as storage. with pytest.raises(RuntimeError, match="backend recovery"): gateway.open_delivery(first_generation) after_browser = gateway.snapshot() assert invocation_count.read_text(encoding="utf-8") == "1" assert after_browser["generation"] == first_generation assert after_browser["recording"]["producer_alive"] is False assert after_browser["recording"]["active_epoch"] is None assert after_browser["recording"]["completed_epochs"] == 1 assert after_browser["error"]["code"] == "camera-source-ended" assert list((first_epoch.parent).glob("epoch-*")) == [first_epoch] assert json.loads( (first_epoch / "summary.json").read_text(encoding="utf-8") ) == first_summary allow_backend_cas.set() assert backend_cas_done.wait(timeout=3) assert observer_errors == [] _wait_until( lambda: ( gateway is not None and gateway.snapshot()["generation"] == first_generation + 1 and gateway.snapshot()["recording"]["media_ready"] is True ) ) recovered = gateway.snapshot() assert len(recovered_snapshots) == 1 assert invocation_count.read_text(encoding="utf-8") == "2" assert recovered["generation"] == first_generation + 1 assert recovered["phase"] == "streaming" assert recovered["error"] is None assert recovered["recording"]["active_epoch"] == first_generation + 1 assert recovered["recording"]["media_ready"] is True assert recovered["recording"]["current_epoch"]["generation"] == ( first_generation + 1 ) assert json.loads( (first_epoch / "summary.json").read_text(encoding="utf-8") ) == first_summary finally: allow_backend_cas.set() gateway.close() def test_source_end_none_epoch_cas_waits_for_canonical_archive_seal( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: executable, _invocation_count = _recoverable_source_end_ffmpeg(tmp_path) monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(executable)) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) session = tmp_path / "sessions" / "camera-source-end-seal-fence" session.mkdir(parents=True) finalize_entered = threading.Event() release_finalize = threading.Event() real_finalize = gateway._finalize_archive # noqa: SLF001 def blocked_finalize(*args: object, **kwargs: object) -> None: finalize_entered.set() assert release_finalize.wait(timeout=3) real_finalize(*args, **kwargs) # type: ignore[arg-type] monkeypatch.setattr(gateway, "_finalize_archive", blocked_finalize) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) assert finalize_entered.wait(timeout=3) unsealed = gateway.snapshot() assert unsealed["phase"] == "error" assert unsealed["error"]["code"] == "camera-source-ended" assert unsealed["recording"]["active_epoch"] is None assert unsealed["recording"]["completed_epochs"] == 0 with pytest.raises(ValueError, match="устарела|lineage"): gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=selected["generation"], expected_recording_session=session.name, expected_active_epoch=None, expected_recording_media_segment_count=1, pre_detach_fence=lambda reserve: reserve(), ) denied = gateway.snapshot() assert denied["generation"] == selected["generation"] assert denied["recording"]["completed_epochs"] == 0 release_finalize.set() _wait_until(lambda: gateway.snapshot()["recording"]["completed_epochs"] == 1) sealed = gateway.snapshot() summary = sealed["recording"]["last_summary"] assert summary["schema_version"] == "missioncore.camera-recording/v1" assert summary["failure_code"] == "camera-source-ended" restarted = gateway.restart_recording_producer( "sensor.camera.right", "192.168.1.20", expected_generation=selected["generation"], expected_recording_session=session.name, expected_active_epoch=None, expected_recording_media_segment_count=1, pre_detach_fence=lambda reserve: reserve(), ) assert restarted["generation"] == selected["generation"] + 1 finally: release_finalize.set() gateway.close() def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path))) session = tmp_path / "sessions" / "camera-bind-before-reader" session.mkdir(parents=True) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) bound_generation: list[int] = [] reader_start_generation: list[int] = [] def bind(committed: dict[str, object]) -> None: recording = committed["recording"] assert isinstance(recording, dict) generation = recording["active_epoch"] assert isinstance(generation, int) bound_generation.append(generation) def start_threads(producer: object) -> None: generation = producer.generation # type: ignore[attr-defined] assert bound_generation == [generation] reader_start_generation.append(generation) monkeypatch.setattr(gateway, "_start_producer_threads", start_threads) try: activated = gateway.activate_recording_producer( "sensor.camera.right", "192.168.1.20", session, commit_fence=lambda commit: commit(), committed_before_start=bind, ) assert bound_generation == [activated["recording"]["active_epoch"]] assert reader_start_generation == bound_generation assert activated["recording"]["media_ready"] is False 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_routine_browser_stall_is_buffered_without_dropping_live_delivery( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: fragment_count = 40 monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str( _preview_buffer_ffmpeg( tmp_path, fragments=fragment_count, payload_bytes=80 * 1024, ) ), ) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) session = tmp_path / "routine-preview-stall" session.mkdir() try: selected = gateway.select("sensor.camera.left", "192.168.1.20") gateway.start_recording(session) lease = gateway.open_delivery(selected["generation"]) # A busy Rerun/browser main thread can defer this WebSocket task for # several seconds. That is ordinary scheduling jitter, not a dead # consumer, and must not terminate the visible camera transport. _wait_until( lambda: ( gateway.snapshot()["recording"]["committed_media_segment_count"] == fragment_count ), ) assert lease.failure_code is None assert gateway.snapshot()["recording"]["preview_consumer_count"] == 1 assert lease.segments.queued_segments == fragment_count + 1 assert lease.segments.queued_bytes < MAX_CAMERA_PREVIEW_QUEUED_BYTES assert lease.segments.get(timeout=1)[0] == "init" for _ in range(fragment_count): assert lease.segments.get(timeout=1)[0] == "media" assert lease.segments.queued_segments == 0 assert lease.segments.queued_bytes == 0 finally: gateway.close() def test_permanently_slow_browser_is_dropped_without_stopping_archive_producer( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: fragment_count = MAX_CAMERA_PREVIEW_QUEUED_SEGMENTS + 20 monkeypatch.setenv( "MISSIONCORE_FFMPEG_BINARY", str( _preview_buffer_ffmpeg( tmp_path, fragments=fragment_count, payload_bytes=160 * 1024, interval_seconds=0.02, ) ), ) 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"]) fast = gateway.open_delivery(selected["generation"]) producer_pid = fast.process.pid active_epoch = gateway.snapshot()["recording"]["active_epoch"] fast_segments: list[str] = [] def drain_fast_reader() -> None: for _ in range(fragment_count + 1): segment = fast.segments.get(timeout=3) assert segment is not None fast_segments.append(segment[0]) fast_reader = threading.Thread(target=drain_fast_reader) fast_reader.start() # 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 replacement = gateway.open_delivery(selected["generation"]) assert replacement.segments.get(timeout=1)[0] == "init" assert replacement.segments.get(timeout=3)[0] == "media" assert replacement.process.pid == producer_pid assert gateway.snapshot()["generation"] == selected["generation"] assert gateway.snapshot()["recording"]["active_epoch"] == active_epoch 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"))) == fragment_count, ) fast_reader.join(timeout=3) assert fast_reader.is_alive() is False assert fast_segments == ["init", *("media" for _ in range(fragment_count))] assert fast.failure_code is None gateway.release_delivery(fast, client_closed=True) gateway.release_delivery(replacement, client_closed=True) 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"] == fragment_count assert summary["media_segment_count"] == fragment_count 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, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path))) service = XgridsK1CompatibilityService(tmp_path) try: with service._lock: service._selected_device_id = "ble-transport-test" service._k1_ip = "192.168.1.20" service._connection_mode = "bridge" service._device_id = "device-k1-test" service._device_session_id = "device-session-test" service._device_session_opened_at = "2026-07-16T20:00:00Z" service._compatibility_attestation = { "firmware_version": "3.0.2", "topology": "direct-lan", "verification": "live-device-info", "basis": "selected-profile-live-device-info-required", "observed_at": "2026-07-16T20:00:00Z", } supervisor = service._connection_supervisor # noqa: SLF001 target = EndpointTarget("192.168.1.20") supervisor.set_intent( intent_id="camera-test-intent", requested_mode="bridge", ) assert supervisor.observe_device_network_applied( intent_id="camera-test-intent", transport_ref="ble-transport-test", connection_mode="bridge", target=target, source="ble-read-only-status", ) host_epoch = supervisor.observe_host_path( HostPathProbeResult( available=True, fingerprint="camera-test-route", interface="en0", source_ipv4="192.168.1.10", route_class="direct", ) ) assert supervisor.observe_endpoint( target=target, intent_id="camera-test-intent", host_path_epoch=host_epoch, reachable=True, ) assert supervisor.observe_control_evidence( VerifiedControlEvidence( intent_id="camera-test-intent", transport_ref="ble-transport-test", host_path_epoch=host_epoch, target=target, connection_mode="bridge", logical_device_id="device-k1-test", compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, control_session_id="camera-test-control-session", ) ) # This bounded gateway test supplies an explicit supervisor proof # instead of opening the real MQTT application dialogue. Prevent the # ordinary state reducer from correctly expiring that synthetic proof # when it observes the deliberately idle test control session. monkeypatch.setattr( service, "_reconcile_connection_supervisor", lambda *_args, **_kwargs: None, ) state = service.select_camera_preview( CameraPreviewSelectRequest( source_id="sensor.camera.left", device_session_id="device-session-test", ) ) cameras = [ stream for stream in state["sensor_catalog"]["streams"] if stream.get("semantic_channel_id") == "camera.preview.live" ] assert len(cameras) == 2 assert sum(bool(stream["activation"]["selected"]) for stream in cameras) == 1 assert sum(stream["delivery"] is not None for stream in cameras) == 1 assert state["compatibility"]["profile_id"] == XGRIDS_K1_COMPATIBILITY_PROFILE_ID camera_contract = json.dumps( {"camera_preview": state["camera_preview"], "streams": cameras} ) assert "rtsp://" not in camera_contract assert "192.168.1.20" not in camera_contract generation = state["camera_preview"]["generation"] stopped = service.stop_camera_preview( CameraPreviewStopRequest( device_session_id="device-session-test", generation=generation, ) ) assert stopped["camera_preview"]["phase"] == "idle" with pytest.raises(ValueError, match="device-сессия"): service.select_camera_preview( CameraPreviewSelectRequest( source_id="sensor.camera.right", device_session_id="stale-session", ) ) finally: service.close()