from __future__ import annotations import asyncio import hashlib import json import sys import threading import time from collections.abc import Callable 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() @pytest.mark.parametrize("entrypoint", ["acquisition", "selected-preview"]) def test_camera_records_in_configured_evidence_outside_source_checkout( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, entrypoint: str, ) -> None: checkout = tmp_path / "checkout" checkout.mkdir() evidence = tmp_path / "durable" / "sessions" session = evidence / "synthetic-camera-session" session.mkdir(parents=True) monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_fake_ffmpeg(tmp_path))) gateway = XgridsK1CameraGateway(checkout, XGRIDS_K1_PLUGIN_ID, evidence_root=evidence) try: if entrypoint == "acquisition": gateway.activate_recording_producer( "sensor.camera.right", "192.168.1.20", session, pre_prepare_fence=lambda reserve: reserve(), commit_fence=lambda commit: commit(), ) else: gateway.select("sensor.camera.right", "192.168.1.20") gateway.start_recording(session) _wait_until(lambda: gateway.snapshot()["recording"]["media_ready"] is True) assert gateway.snapshot()["recording"]["session"] == session.name assert not list(checkout.rglob("*.m4s")) finally: gateway.close() assert any(session.rglob("*.m4s")) @pytest.mark.parametrize("entrypoint", ["acquisition", "selected-preview"]) @pytest.mark.parametrize("escape", ["sibling", "symlink"]) def test_camera_rejects_paths_outside_configured_evidence_before_admission( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, entrypoint: str, escape: str, ) -> None: checkout = tmp_path / "checkout" checkout.mkdir() evidence = tmp_path / "evidence" evidence.mkdir() outside = checkout / "not-evidence" outside.mkdir() session = outside if escape == "symlink": session = evidence / "escaped-session" session.symlink_to(outside, target_is_directory=True) gateway = XgridsK1CameraGateway(checkout, XGRIDS_K1_PLUGIN_ID, evidence_root=evidence) def forbidden(*_args: object, **_kwargs: object) -> bool: raise AssertionError("An escaped path must not reach authority or FFmpeg") monkeypatch.setattr(camera_module.subprocess, "Popen", forbidden) try: with pytest.raises(ValueError, match="configured evidence root"): if entrypoint == "acquisition": gateway.activate_recording_producer( "sensor.camera.right", "192.168.1.20", session, pre_prepare_fence=forbidden, commit_fence=forbidden, ) else: gateway.start_recording(session) assert gateway.snapshot()["recording"]["active"] is False finally: gateway.close() def test_service_camera_uses_the_same_external_evidence_root_as_acquisition( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: checkout = tmp_path / "checkout" checkout.mkdir() evidence = tmp_path / "configured-evidence" session = evidence / "synthetic-session" session.mkdir(parents=True) monkeypatch.setenv("MISSIONCORE_EVIDENCE_DIR", str(evidence)) service = XgridsK1CompatibilityService(checkout) authority_entered = False def deny_authority(_reserve: Callable[[], bool]) -> bool: nonlocal authority_entered authority_entered = True return False try: with pytest.raises(ValueError, match="authority"): service.camera_preview.activate_recording_producer( "sensor.camera.right", "192.168.1.20", session, pre_prepare_fence=deny_authority, commit_fence=lambda _commit: False, ) assert authority_entered is True assert service.camera_preview.snapshot()["active_source_id"] is None finally: service.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_never_spawns_before_authority_reservation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("MISSIONCORE_FFMPEG_BINARY", str(_silent_ffmpeg(tmp_path))) session = tmp_path / "sessions" / "camera-pre-popen-denied" session.mkdir(parents=True) gateway = XgridsK1CameraGateway(tmp_path, XGRIDS_K1_PLUGIN_ID) popen_called = False def forbidden_popen(*_: object, **__: object) -> object: nonlocal popen_called popen_called = True raise AssertionError("FFmpeg must not start before authority reservation") monkeypatch.setattr(camera_module.subprocess, "Popen", forbidden_popen) try: with pytest.raises(ValueError, match="authority"): gateway.activate_recording_producer( "sensor.camera.right", "192.168.1.20", session, pre_prepare_fence=lambda _reserve: False, commit_fence=lambda commit: commit(), ) snapshot = gateway.snapshot() assert popen_called is False assert snapshot["active_source_id"] is None assert snapshot["recording"]["active"] is False finally: gateway.close() def test_acquisition_candidate_binds_epoch_before_any_reader_thread_starts( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> 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) caplog.set_level("INFO", logger="k1link.device_plugins.xgrids_k1.camera") bound_generation: list[int] = [] reader_start_generation: list[int] = [] authority_reserved = False def reserve_authority(reserve: Callable[[], bool]) -> bool: nonlocal authority_reserved assert authority_reserved is False assert reserve() is True authority_reserved = True return True 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 authority_reserved is True 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, pre_prepare_fence=reserve_authority, 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 timing = [ record for record in caplog.records if getattr(record, "event_code", None) == "k1_camera_activation_timing" ] assert len(timing) == 1 assert timing[0].camera_generation == bound_generation[0] assert timing[0].camera_authority_wait_ms >= 0 assert timing[0].camera_ffmpeg_prepare_ms >= 0 assert timing[0].camera_post_spawn_commit_ms >= 0 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() def test_onboard_preview_never_spawns_a_camera_producer(tmp_path, monkeypatch): gateway = _gateway(tmp_path, monkeypatch) try: selected = gateway.select("sensor.camera.right", "192.168.1.20") monkeypatch.setattr( gateway, "_spawn_selected_producer", lambda: pytest.fail("an onboard preview must not own camera activation"), ) with pytest.raises(RuntimeError, match="producer is not ready"): gateway.open_delivery(selected["generation"], require_recording=True) assert gateway.snapshot()["recording"]["producer_alive"] is False finally: gateway.close()