Organize host and robot streaming releases

host/robot-command-center/backend/monitoring/__init__.py

@@ -0,0 +1 @@
+

host/robot-command-center/backend/monitoring/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class MonitoringConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "monitoring"
+

host/robot-command-center/backend/monitoring/common.py

@@ -0,0 +1,342 @@
+from __future__ import annotations
+
+import os
+import json
+import struct
+from datetime import datetime, timezone
+from pathlib import Path
+import threading
+import time
+from typing import Any
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+WORKSPACE_ROOT = PROJECT_ROOT.parent
+JPEG_FRAME_DIR = WORKSPACE_ROOT / "RobotDataShow" / "jpeg-frames"
+OMNISOCKET_CONFIG_PATH = PROJECT_ROOT / "config" / "omnisocket_demo.yaml"
+
+VIDEO_SOURCE_MODE = os.getenv("VIDEO_SOURCE_MODE", "auto").strip().lower()
+OMNISOCKET_FRAME_FRESH_SECONDS = 2.0
+VIDEO_TIMESTAMP_SAMPLE_SIZE = 10
+VIDEO_TRAILER_ENDIANNESS = "little"
+VIDEO_TRAILER_TIMESTAMP_UNIT = "ms"
+VIDEO_TRAILER_TIMESTAMP_MULTIPLIER_NS = 1_000_000
+VIDEO_TRAILER_TIMESTAMP_MAX_SKEW_NS = 7 * 24 * 60 * 60 * 1_000_000_000
+VIDEO_TRAILER_COORDINATE_FORMAT = (
+    "uint64 timestamp_ms + float64 latitude + float64 longitude + uint32 capture_to_send_ms (little-endian)"
+)
+VIDEO_TRAILER_STRUCT = struct.Struct("<QddI")
+VIDEO_TRAILER_BYTES = VIDEO_TRAILER_STRUCT.size
+
+CONTROL_PACKET = struct.Struct("<6f")
+CONTROL_PACKET_SIZE = CONTROL_PACKET.size
+CONTROL_SOURCE_NATIVE_UDP = "native_udp"
+CONTROL_SOURCE_WEB = "web"
+CONTROL_SOURCE_PRIORITY = (CONTROL_SOURCE_NATIVE_UDP, CONTROL_SOURCE_WEB)
+ZERO_CONTROL_PAYLOAD = CONTROL_PACKET.pack(0.0, 0.0, 0.0, 0.0, 0.0, 0.0)
+BLITZ_RUN_DIR_RAW = os.getenv("BLITZ_RUN_DIR", "").strip()
+BLITZ_RUN_DIR = Path(BLITZ_RUN_DIR_RAW).expanduser() if BLITZ_RUN_DIR_RAW else None
+BLITZ_INSTANCE_ID = os.getenv("BLITZ_INSTANCE_ID", "").strip() or f"backend-{os.getpid()}"
+
+
+def utc_iso_now() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def parse_simple_yaml_scalar(value: str) -> Any:
+    if value in {'""', "''"}:
+        return ""
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}:
+        return value[1:-1]
+    if value.lower() == "true":
+        return True
+    if value.lower() == "false":
+        return False
+    if value and value.lstrip("-").isdigit():
+        return int(value)
+    return value
+
+
+def load_simple_yaml_config(path: Path) -> dict[str, Any]:
+    parsed: dict[str, Any] = {}
+    current_section: str | None = None
+
+    with path.open("r", encoding="utf-8") as file:
+        for raw_line in file:
+            line = raw_line.split("#", 1)[0].rstrip()
+            if not line.strip():
+                continue
+
+            if not line.startswith(" "):
+                if not line.endswith(":"):
+                    raise ValueError(f"invalid top-level yaml line: {raw_line.strip()}")
+                current_section = line[:-1].strip()
+                parsed[current_section] = {}
+                continue
+
+            if current_section is None:
+                raise ValueError(f"yaml key outside section: {raw_line.strip()}")
+
+            stripped = line.strip()
+            if ":" not in stripped:
+                raise ValueError(f"invalid yaml key line: {raw_line.strip()}")
+
+            key, value = stripped.split(":", 1)
+            parsed[current_section][key.strip()] = parse_simple_yaml_scalar(value.strip())
+
+    return parsed
+
+
+def load_omnisocket_config() -> dict[str, Any]:
+    config: dict[str, Any] = {}
+    if OMNISOCKET_CONFIG_PATH.exists():
+        try:
+            try:
+                import yaml  # type: ignore
+
+                with OMNISOCKET_CONFIG_PATH.open("r", encoding="utf-8") as file:
+                    config = yaml.safe_load(file) or {}
+            except ImportError:
+                config = load_simple_yaml_config(OMNISOCKET_CONFIG_PATH)
+        except Exception:
+            config = {}
+
+    transport_cfg = dict(config.get("transport", {}))
+    video_receiver_cfg = dict(config.get("video_receiver", {}))
+    control_sender_cfg = dict(config.get("control_sender", {}))
+    control_ack_receiver_cfg = dict(config.get("control_ack_receiver", {}))
+    control_ingress_cfg = dict(config.get("control_ingress", {}))
+    video_sender_cfg = dict(config.get("video_sender", {}))
+    telemetry_receiver_cfg = dict(config.get("telemetry_receiver", {}))
+
+    transport_cfg["server_addr"] = os.getenv(
+        "OMNISOCKET_SERVER_ADDR",
+        str(transport_cfg.get("server_addr", "127.0.0.1:10909")),
+    )
+    transport_cfg["relay_via"] = os.getenv(
+        "OMNISOCKET_RELAY_VIA",
+        str(transport_cfg.get("relay_via", "")),
+    )
+    transport_cfg["bind_ip"] = os.getenv(
+        "OMNISOCKET_BIND_IP",
+        str(transport_cfg.get("bind_ip", "")),
+    )
+    transport_cfg["bind_device"] = os.getenv(
+        "OMNISOCKET_BIND_DEVICE",
+        str(transport_cfg.get("bind_device", "")),
+    )
+
+    video_receiver_cfg["peer_id"] = os.getenv(
+        "OMNISOCKET_VIDEO_PEER_ID",
+        str(video_receiver_cfg.get("peer_id", "peer-a-video")),
+    )
+    video_receiver_cfg["buffer_bytes"] = int(
+        os.getenv(
+            "OMNISOCKET_BUFFER_BYTES",
+            str(video_receiver_cfg.get("buffer_bytes", 1024 * 1024)),
+        )
+    )
+
+    control_sender_cfg["peer_id"] = os.getenv(
+        "OMNISOCKET_CONTROL_PEER_ID",
+        str(control_sender_cfg.get("peer_id", "peer-a-ctrl")),
+    )
+    control_sender_cfg["target_peer"] = os.getenv(
+        "OMNISOCKET_CONTROL_TARGET_PEER",
+        str(control_sender_cfg.get("target_peer", "peer-b-ctrl")),
+    )
+
+    control_ack_receiver_cfg["peer_id"] = os.getenv(
+        "OMNISOCKET_CONTROL_ACK_RECEIVER_PEER_ID",
+        str(control_ack_receiver_cfg.get("peer_id", "peer-a-ctrl-ack")),
+    )
+    control_ack_receiver_cfg["expected_sender"] = os.getenv(
+        "OMNISOCKET_CONTROL_ACK_EXPECTED_SENDER",
+        str(control_ack_receiver_cfg.get("expected_sender", "peer-b-ctrl-ack")),
+    )
+
+    video_sender_cfg["peer_id"] = os.getenv(
+        "OMNISOCKET_VIDEO_SENDER_PEER_ID",
+        str(video_sender_cfg.get("peer_id", "peer-b-video")),
+    )
+    video_sender_cfg["target_peer"] = os.getenv(
+        "OMNISOCKET_VIDEO_TARGET_PEER_ID",
+        str(video_sender_cfg.get("target_peer", "peer-a-video")),
+    )
+
+    control_ingress_cfg["native_udp_bind"] = os.getenv(
+        "OMNISOCKET_CONTROL_NATIVE_UDP_BIND",
+        str(control_ingress_cfg.get("native_udp_bind", "127.0.0.1:10921")),
+    )
+    control_ingress_cfg["source_lease_ms"] = int(
+        os.getenv(
+            "OMNISOCKET_CONTROL_SOURCE_LEASE_MS",
+            str(control_ingress_cfg.get("source_lease_ms", 300)),
+        )
+    )
+    control_ingress_cfg["send_rate_hz"] = float(
+        os.getenv(
+            "OMNISOCKET_CONTROL_SEND_RATE_HZ",
+            str(control_ingress_cfg.get("send_rate_hz", 20.0)),
+        )
+    )
+    control_ingress_cfg["zero_burst_packets"] = int(
+        os.getenv(
+            "OMNISOCKET_CONTROL_ZERO_BURST_PACKETS",
+            str(control_ingress_cfg.get("zero_burst_packets", 3)),
+        )
+    )
+
+    telemetry_receiver_cfg["peer_id"] = os.getenv(
+        "OMNISOCKET_TELEMETRY_PEER_ID",
+        str(telemetry_receiver_cfg.get("peer_id", "peer-a-telemetry")),
+    )
+    telemetry_receiver_cfg["interval_ms"] = int(
+        os.getenv(
+            "OMNISOCKET_TELEMETRY_INTERVAL_MS",
+            str(telemetry_receiver_cfg.get("interval_ms", 500)),
+        )
+    )
+    telemetry_receiver_cfg["stale_after_ms"] = int(
+        os.getenv(
+            "OMNISOCKET_TELEMETRY_STALE_AFTER_MS",
+            str(telemetry_receiver_cfg.get("stale_after_ms", telemetry_receiver_cfg["interval_ms"] * 3)),
+        )
+    )
+
+    return {
+        "transport": transport_cfg,
+        "video_receiver": video_receiver_cfg,
+        "control_sender": control_sender_cfg,
+        "control_ack_receiver": control_ack_receiver_cfg,
+        "control_ingress": control_ingress_cfg,
+        "video_sender": video_sender_cfg,
+        "telemetry_receiver": telemetry_receiver_cfg,
+    }
+
+
+class JsonlRunLogger:
+    def __init__(self, stem_env: str, default_stem: str) -> None:
+        explicit_path = os.getenv(stem_env, "").strip()
+        self._path = Path(explicit_path) if explicit_path else (
+            BLITZ_RUN_DIR / f"{default_stem}.{BLITZ_INSTANCE_ID}.jsonl" if BLITZ_RUN_DIR is not None else None
+        )
+        self._lock = threading.Lock()
+        self._file = None
+        self._buffered_bytes = 0
+        self._current_bytes = 0
+        self._flush_bytes = self._positive_int_env("BLITZ_JSONL_FLUSH_BYTES", 262144)
+        self._flush_interval_ms = self._positive_int_env("BLITZ_JSONL_FLUSH_INTERVAL_MS", 1000)
+        self._max_bytes = self._positive_int_env("BLITZ_JSONL_ROTATE_BYTES", 134217728)
+        self._max_files = self._positive_int_env("BLITZ_JSONL_ROTATE_FILES", 8)
+        self._last_flush_monotonic_ms = self._now_ms()
+
+        if self._path is not None:
+            try:
+                self._path.parent.mkdir(parents=True, exist_ok=True)
+                self._file = self._path.open("a", encoding="utf-8")
+                self._current_bytes = self._path.stat().st_size if self._path.exists() else 0
+            except OSError:
+                self._file = None
+
+    @property
+    def path(self) -> str | None:
+        return str(self._path) if self._path is not None else None
+
+    @property
+    def enabled(self) -> bool:
+        return self._file is not None
+
+    def write(self, payload: dict[str, Any]) -> None:
+        if self._file is None:
+            return
+        line = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+        line_bytes = len(line.encode("utf-8")) + 1
+        with self._lock:
+            if self._file is None:
+                return
+            try:
+                self._file.write(line)
+                self._file.write("\n")
+                self._buffered_bytes += line_bytes
+                self._current_bytes += line_bytes
+                now_ms = self._now_ms()
+                if (
+                    self._buffered_bytes >= self._flush_bytes
+                    or (self._flush_interval_ms > 0 and now_ms - self._last_flush_monotonic_ms >= self._flush_interval_ms)
+                ):
+                    self._flush_locked(now_ms)
+                if self._max_bytes > 0 and self._max_files > 0 and self._current_bytes >= self._max_bytes:
+                    self._rotate_locked()
+            except OSError:
+                if self._file is not None:
+                    try:
+                        self._file.close()
+                    except OSError:
+                        pass
+                    self._file = None
+
+    def close(self) -> None:
+        with self._lock:
+            if self._file is not None:
+                try:
+                    self._flush_locked(self._now_ms())
+                except OSError:
+                    pass
+                self._file.close()
+                self._file = None
+
+    def _flush_locked(self, now_ms: int) -> None:
+        if self._file is None:
+            return
+        self._file.flush()
+        self._buffered_bytes = 0
+        self._last_flush_monotonic_ms = now_ms
+
+    def _rotate_locked(self) -> None:
+        if self._path is None or self._file is None or self._max_files <= 0:
+            return
+        self._flush_locked(self._now_ms())
+        self._file.close()
+        self._file = None
+
+        oldest = self._path.with_name(f"{self._path.name}.{self._max_files}")
+        if oldest.exists():
+            oldest.unlink()
+
+        for index in range(self._max_files - 1, 0, -1):
+            src = self._path.with_name(f"{self._path.name}.{index}")
+            if src.exists():
+                dst = self._path.with_name(f"{self._path.name}.{index + 1}")
+                src.replace(dst)
+
+        if self._path.exists():
+            rotated = self._path.with_name(f"{self._path.name}.1")
+            self._path.replace(rotated)
+
+        self._file = self._path.open("a", encoding="utf-8")
+        self._buffered_bytes = 0
+        self._current_bytes = self._path.stat().st_size if self._path.exists() else 0
+        self._last_flush_monotonic_ms = self._now_ms()
+
+    @staticmethod
+    def _now_ms() -> int:
+        return int(time.monotonic() * 1000)
+
+    @staticmethod
+    def _positive_int_env(name: str, default: int) -> int:
+        raw = os.getenv(name, "").strip()
+        try:
+            value = int(raw)
+        except ValueError:
+            return default
+        return value if value > 0 else default
+
+
+def parse_host_port(bind_addr: str) -> tuple[str, int]:
+    host, port_text = bind_addr.rsplit(":", 1)
+    host = host.strip() or "127.0.0.1"
+    port = int(port_text)
+    if port <= 0 or port > 65535:
+        raise ValueError(f"invalid port in bind address: {bind_addr}")
+    return host, port

host/robot-command-center/backend/monitoring/consumers.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import json
+
+from channels.generic.websocket import WebsocketConsumer
+
+from .common import CONTROL_PACKET_SIZE, CONTROL_SOURCE_WEB
+from .services import control_arbiter, native_control_ingress
+
+
+class ControlConsumer(WebsocketConsumer):
+    def connect(self) -> None:
+        control_arbiter.ensure_started()
+        native_control_ingress.ensure_started()
+        self.accept()
+        self.send(
+            text_data=json.dumps(
+                {
+                    "type": "ready",
+                    "packet_bytes": CONTROL_PACKET_SIZE,
+                }
+            )
+        )
+
+    def receive(self, text_data: str | None = None, bytes_data: bytes | None = None) -> None:
+        if bytes_data is None:
+            self.send(text_data=json.dumps({"type": "error", "detail": "binary control payload required"}))
+            return
+
+        if len(bytes_data) != CONTROL_PACKET_SIZE:
+            self.send(
+                text_data=json.dumps(
+                    {
+                        "type": "error",
+                        "detail": f"expected {CONTROL_PACKET_SIZE} bytes, got {len(bytes_data)}",
+                    }
+                )
+            )
+            return
+
+        control_arbiter.ingest_command(CONTROL_SOURCE_WEB, bytes_data)
+

host/robot-command-center/backend/monitoring/control.py

@@ -0,0 +1,951 @@
+from __future__ import annotations
+
+import json
+import socket
+import sys
+import threading
+import time
+from typing import Any
+
+from .common import (
+    CONTROL_PACKET_SIZE,
+    CONTROL_SOURCE_NATIVE_UDP,
+    CONTROL_SOURCE_PRIORITY,
+    JsonlRunLogger,
+    ZERO_CONTROL_PAYLOAD,
+    WORKSPACE_ROOT,
+    load_omnisocket_config,
+    parse_host_port,
+)
+from .video import safe_kcp_stats
+
+
+def _payload_preview(payload: bytes, limit: int = 160) -> str:
+    if not payload:
+        return ""
+    preview = payload[:limit].decode("utf-8", errors="replace")
+    if len(payload) > limit:
+        return f"{preview}..."
+    return preview
+
+
+class ControlAckTracker:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._event_logger = JsonlRunLogger("BLITZ_A_CONTROL_EVENTS_LOG_PATH", "a-control-events")
+        self._ack_logger = JsonlRunLogger("BLITZ_A_CONTROL_ACKS_LOG_PATH", "a-control-acks")
+        self._pending: dict[int, dict[str, Any]] = {}
+        self._latest_estimate: dict[str, Any] = {
+            "ack_available": False,
+            "updated_at": None,
+            "received_mono_ns": 0,
+            "control_loop_rtt_ms": None,
+            "b_recv_to_persist_ms": None,
+            "control_oneway_network_est_ms": None,
+            "control_to_persist_est_ms": None,
+            "sample_reason": None,
+        }
+
+    def register_send(
+        self,
+        *,
+        message_id: int,
+        issued_at_unix_ns: int,
+        issued_at_mono_ns: int,
+        source: str,
+        payload: bytes,
+        send_call_latency_us: int,
+    ) -> None:
+        event = {
+            "ts_unix_nano": issued_at_unix_ns,
+            "message_id": message_id,
+            "issued_at_unix_ns": issued_at_unix_ns,
+            "issued_at_mono_ns": issued_at_mono_ns,
+            "source": source,
+            "command_signature": payload.hex(),
+            "payload_size": len(payload),
+            "send_call_latency_us": send_call_latency_us,
+        }
+        with self._lock:
+            self._pending[message_id] = event
+            self._prune_locked(issued_at_mono_ns)
+        self._event_logger.write(event)
+
+    def handle_ack(self, ack_payload: dict[str, Any], received_unix_ns: int, received_mono_ns: int) -> str:
+        try:
+            message_id = int(ack_payload["message_id"])
+        except (KeyError, TypeError, ValueError):
+            return "invalid_message_id"
+
+        with self._lock:
+            event = self._pending.pop(message_id, None)
+            self._prune_locked(received_mono_ns)
+
+        if event is None:
+            return "pending_missing"
+
+        try:
+            control_loop_rtt_ms = round((received_unix_ns - int(event["issued_at_unix_ns"])) / 1_000_000.0, 3)
+            b_recv_to_persist_ms = round(float(ack_payload.get("b_recv_to_persist_us", 0)) / 1000.0, 3)
+        except (TypeError, ValueError):
+            return "invalid_timing"
+
+        control_oneway_network_est_ms = round(max(0.0, (control_loop_rtt_ms - b_recv_to_persist_ms) / 2.0), 3)
+        control_to_persist_est_ms = round(control_oneway_network_est_ms + b_recv_to_persist_ms, 3)
+        ack_record = {
+            "ts_unix_nano": received_unix_ns,
+            "received_unix_ns": received_unix_ns,
+            "received_mono_ns": received_mono_ns,
+            "message_id": message_id,
+            "ack_phase": str(ack_payload.get("ack_phase") or "persist_end"),
+            "sample_reason": str(ack_payload.get("sample_reason") or ""),
+            "b_recv_to_persist_us": ack_payload.get("b_recv_to_persist_us"),
+            "unix_send_ok": bool(ack_payload.get("unix_send_ok", False)),
+            "issued_at_unix_ns": event["issued_at_unix_ns"],
+            "source": event["source"],
+            "control_loop_rtt_ms": control_loop_rtt_ms,
+            "b_recv_to_persist_ms": b_recv_to_persist_ms,
+            "control_oneway_network_est_ms": control_oneway_network_est_ms,
+            "control_to_persist_est_ms": control_to_persist_est_ms,
+        }
+        self._ack_logger.write(ack_record)
+        with self._lock:
+            self._latest_estimate = {
+                "ack_available": True,
+                "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(received_unix_ns / 1_000_000_000)),
+                "received_mono_ns": received_mono_ns,
+                "control_loop_rtt_ms": control_loop_rtt_ms,
+                "b_recv_to_persist_ms": b_recv_to_persist_ms,
+                "control_oneway_network_est_ms": control_oneway_network_est_ms,
+                "control_to_persist_est_ms": control_to_persist_est_ms,
+                "sample_reason": ack_record["sample_reason"],
+            }
+        return "accepted"
+
+    def get_latest_estimate(self) -> dict[str, Any]:
+        with self._lock:
+            estimate = dict(self._latest_estimate)
+        if int(estimate.get("received_mono_ns", 0) or 0) > 0 and time.monotonic_ns() - int(estimate["received_mono_ns"]) > 10_000_000_000:
+            estimate["ack_available"] = False
+            estimate["control_loop_rtt_ms"] = None
+            estimate["b_recv_to_persist_ms"] = None
+            estimate["control_oneway_network_est_ms"] = None
+            estimate["control_to_persist_est_ms"] = None
+            estimate["sample_reason"] = None
+        estimate.pop("received_mono_ns", None)
+        return estimate
+
+    def close(self) -> None:
+        self._event_logger.close()
+        self._ack_logger.close()
+
+    def _prune_locked(self, now_mono_ns: int) -> None:
+        stale_ids = [
+            message_id
+            for message_id, event in self._pending.items()
+            if now_mono_ns - int(event.get("issued_at_mono_ns", 0)) > 60_000_000_000
+        ]
+        for message_id in stale_ids:
+            self._pending.pop(message_id, None)
+
+
+class OmniSocketControlSender:
+    def __init__(self, ack_tracker: ControlAckTracker) -> None:
+        self._lock = threading.Lock()
+        self._camera_command_lock = threading.Lock()
+        self._camera_condition = threading.Condition(self._lock)
+        self._ack_tracker = ack_tracker
+        self._session = None
+        self._session_cls = None
+        self._msg_type_text = None
+        self._msg_type_error = None
+        self._control_defaults: dict[str, Any] = {}
+        self._started = False
+        self._drain_thread: threading.Thread | None = None
+        self._closing = threading.Event()
+        self._target_peer = ""
+        self._send_count = 0
+        self._send_errors = 0
+        self._drain_errors = 0
+        self._last_error = ""
+        self._reconnect_count = 0
+        self._ever_connected = False
+        self._registered = False
+        self._supports_send_with_id = False
+        self._supports_send_text = False
+        self._requested_camera: str | None = None
+        self._active_camera: str | None = None
+        self._camera_ack_revision = 0
+        self._camera_command_count = 0
+        self._camera_ack_count = 0
+        self._camera_updated_at: str | None = None
+        self._camera_last_error = ""
+        self._load_backend()
+
+    def _load_backend(self) -> None:
+        try:
+            self._import_backend()
+        except Exception as error:  # pragma: no cover - optional runtime dependency
+            self._last_error = f"omnisocket import failed: {error}"
+
+    def _import_backend(self) -> None:
+        try:
+            from omnisocket import CONTROL_DEFAULTS, MSG_TYPE_ERROR, MSG_TYPE_TEXT, Session  # type: ignore
+        except ImportError:
+            python_dir = WORKSPACE_ROOT / "OmniSocketGo" / "python"
+            if python_dir.exists():
+                sys.path.insert(0, str(python_dir))
+            from omnisocket import CONTROL_DEFAULTS, MSG_TYPE_ERROR, MSG_TYPE_TEXT, Session  # type: ignore
+
+        self._session_cls = Session
+        self._msg_type_text = MSG_TYPE_TEXT
+        self._msg_type_error = MSG_TYPE_ERROR
+        self._control_defaults = dict(CONTROL_DEFAULTS)
+        self._supports_send_text = hasattr(Session, "send_text")
+
+    def _connect_session(self):
+        assert self._session_cls is not None
+
+        config = load_omnisocket_config()
+        transport_cfg = config.get("transport", {})
+        control_cfg = config.get("control_sender", {})
+
+        session = self._session_cls()
+        session.connect(
+            server_addr=str(transport_cfg.get("server_addr", "127.0.0.1:10909")),
+            peer_id=str(control_cfg.get("peer_id", "peer-a-ctrl")),
+            relay_via=str(transport_cfg.get("relay_via", "")),
+            bind_ip=str(transport_cfg.get("bind_ip", "")),
+            bind_device=str(transport_cfg.get("bind_device", "")),
+            **self._control_defaults,
+        )
+        target_peer = str(control_cfg.get("target_peer", "peer-b-ctrl"))
+        return session, target_peer
+
+    def ensure_started(self) -> None:
+        if self._session_cls is None:
+            return
+
+        with self._lock:
+            if self._closing.is_set():
+                return
+            if self._started and self._session is not None:
+                return
+            session, target_peer = self._connect_session()
+            self._session = session
+            self._target_peer = target_peer
+            self._closing.clear()
+            self._started = True
+            self._last_error = ""
+            self._registered = bool(dict(session.stats()).get("registered", 0))
+            self._supports_send_with_id = hasattr(session, "send_with_id")
+            self._supports_send_text = hasattr(session, "send_text")
+            if self._ever_connected:
+                self._reconnect_count += 1
+            else:
+                self._ever_connected = True
+            self._drain_thread = threading.Thread(
+                target=self._drain_loop,
+                name="omnisocket-control-drain",
+                daemon=True,
+            )
+            self._drain_thread.start()
+
+    def _reset_session(self, session: Any | None) -> None:
+        with self._lock:
+            if session is not None and session is not self._session:
+                return
+            current = self._session
+            self._session = None
+            self._started = False
+            self._registered = False
+            self._supports_send_with_id = False
+        if current is not None:
+            try:
+                current.close()
+            except Exception:
+                pass
+
+    def send_payload(self, payload: bytes, *, source: str) -> None:
+        if len(payload) != CONTROL_PACKET_SIZE:
+            raise ValueError(f"expected {CONTROL_PACKET_SIZE} bytes, got {len(payload)}")
+        self.ensure_started()
+        with self._lock:
+            session = self._session
+            target_peer = self._target_peer
+            supports_send_with_id = self._supports_send_with_id
+
+        if session is None:
+            raise RuntimeError("control session is not available")
+
+        try:
+            issued_at_unix_ns = time.time_ns()
+            issued_at_mono_ns = time.monotonic_ns()
+            send_started_ns = time.perf_counter_ns()
+            message_id: int | None = None
+            if supports_send_with_id:
+                message_id = int(session.send_with_id(to=target_peer, data=payload))
+            else:
+                session.send(to=target_peer, data=payload)
+            send_call_latency_us = max(0, int((time.perf_counter_ns() - send_started_ns) / 1000))
+        except Exception as error:
+            with self._lock:
+                self._send_errors += 1
+                self._last_error = str(error)
+            self._reset_session(session)
+            raise
+
+        if message_id is not None:
+            self._ack_tracker.register_send(
+                message_id=message_id,
+                issued_at_unix_ns=issued_at_unix_ns,
+                issued_at_mono_ns=issued_at_mono_ns,
+                source=source,
+                payload=payload,
+                send_call_latency_us=send_call_latency_us,
+            )
+        with self._lock:
+            self._send_count += 1
+
+    def send_zero_burst(self, count: int) -> None:
+        for _ in range(max(0, count)):
+            try:
+                self.send_payload(ZERO_CONTROL_PAYLOAD, source="zero_burst")
+            except Exception:
+                return
+
+    def select_camera(self, camera: str, *, timeout: float = 2.0) -> dict[str, Any]:
+        normalized = camera.strip().lower()
+        if normalized not in {"head", "waist"}:
+            raise ValueError("camera must be 'head' or 'waist'")
+
+        with self._camera_command_lock:
+            self.ensure_started()
+            with self._lock:
+                session = self._session
+                target_peer = self._target_peer
+                supports_send_text = self._supports_send_text
+                previous_revision = self._camera_ack_revision
+
+            if session is None:
+                raise RuntimeError("control session is not available")
+            if not supports_send_text:
+                raise RuntimeError("omnisocket extension does not support text messages; rebuild it with make python-ext")
+
+            try:
+                session.send_text(to=target_peer, text=f"camera:{normalized}")
+            except Exception as error:
+                with self._lock:
+                    self._send_errors += 1
+                    self._last_error = str(error)
+                    self._camera_last_error = str(error)
+                self._reset_session(session)
+                raise
+
+            deadline = time.monotonic() + max(0.1, timeout)
+            with self._camera_condition:
+                self._send_count += 1
+                self._camera_command_count += 1
+                self._requested_camera = normalized
+                self._camera_last_error = ""
+                while self._camera_ack_revision == previous_revision:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        self._camera_last_error = "robot camera confirmation timed out"
+                        break
+                    self._camera_condition.wait(timeout=remaining)
+                confirmed = self._camera_ack_revision != previous_revision and self._active_camera == normalized
+
+            status = self.get_camera_status()
+            status["confirmed"] = confirmed
+            return status
+
+    def get_camera_status(self) -> dict[str, Any]:
+        session_stats = self.session_stats()
+        with self._lock:
+            return {
+                "available": self._session_cls is not None and self._supports_send_text,
+                "connected": self._session is not None,
+                "registered": bool(session_stats.get("registered", 0)),
+                "requested_camera": self._requested_camera,
+                "active_camera": self._active_camera,
+                "command_count": self._camera_command_count,
+                "ack_count": self._camera_ack_count,
+                "updated_at": self._camera_updated_at,
+                "last_error": self._camera_last_error or self._last_error,
+            }
+
+    def _drain_loop(self) -> None:
+        while not self._closing.is_set():
+            with self._lock:
+                session = self._session
+            if session is None:
+                return
+
+            try:
+                result = session.recv(timeout_ms=100)
+            except Exception as error:
+                last_server_error = ""
+                try:
+                    last_server_error = str(dict(session.stats()).get("last_server_error", "") or "")
+                except Exception:
+                    last_server_error = ""
+                with self._lock:
+                    self._drain_errors += 1
+                    self._registered = False
+                    self._last_error = last_server_error or str(error)
+                if not self._closing.is_set():
+                    self._reset_session(session)
+                return
+
+            if result is None:
+                try:
+                    stats = dict(session.stats())
+                except Exception:
+                    stats = {}
+                with self._lock:
+                    self._registered = bool(stats.get("registered", 0))
+                    if stats.get("last_server_error"):
+                        self._last_error = str(stats.get("last_server_error"))
+                continue
+
+            from_peer, msg_type, payload = result
+            if msg_type == self._msg_type_error:
+                text = payload.decode("utf-8", errors="replace")
+                try:
+                    stats = dict(session.stats())
+                except Exception:
+                    stats = {}
+                with self._lock:
+                    self._last_error = f"server error from {from_peer}: {text}"
+                    self._registered = bool(stats.get("registered", 0))
+                continue
+
+            if msg_type == self._msg_type_text:
+                with self._lock:
+                    expected_peer = self._target_peer
+                if expected_peer and from_peer != expected_peer:
+                    continue
+                try:
+                    message = json.loads(payload.decode("utf-8"))
+                except (UnicodeDecodeError, json.JSONDecodeError):
+                    continue
+                camera = str(message.get("camera") or "").lower() if isinstance(message, dict) else ""
+                if isinstance(message, dict) and message.get("type") == "camera.selected" and camera in {"head", "waist"}:
+                    with self._camera_condition:
+                        self._active_camera = camera
+                        self._camera_ack_revision += 1
+                        self._camera_ack_count += 1
+                        self._camera_updated_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+                        self._camera_last_error = ""
+                        self._camera_condition.notify_all()
+
+    def session_stats(self) -> dict[str, Any]:
+        with self._lock:
+            session = self._session
+        if session is None:
+            return {"connected": 0, "registered": 0, "last_server_error": self._last_error}
+        try:
+            return dict(session.stats())
+        except Exception:
+            return {"connected": 0, "registered": 0, "last_server_error": self._last_error}
+
+    def session_kcp_stats(self) -> dict[str, Any]:
+        with self._lock:
+            session = self._session
+        return safe_kcp_stats(session)
+
+    def get_status(self) -> dict[str, Any]:
+        config = load_omnisocket_config()
+        control_cfg = config.get("control_sender", {})
+        session_stats = self.session_stats()
+        with self._lock:
+            return {
+                "backend_ready": self._session_cls is not None,
+                "started": self._started,
+                "connected": self._session is not None,
+                "registered": bool(session_stats.get("registered", 0)),
+                "peer_id": str(control_cfg.get("peer_id", "")),
+                "target_peer": str(control_cfg.get("target_peer", "")),
+                "send_count": self._send_count,
+                "send_errors": self._send_errors,
+                "drain_errors": self._drain_errors,
+                "reconnect_count": self._reconnect_count,
+                "last_server_error": str(session_stats.get("last_server_error", "") or ""),
+                "last_error": self._last_error,
+            }
+
+    def close(self) -> None:
+        self._closing.set()
+        self.send_zero_burst(1)
+        self._reset_session(None)
+        drain_thread = self._drain_thread
+        if drain_thread is not None and drain_thread.is_alive():
+            drain_thread.join(timeout=0.5)
+
+
+class OmniSocketControlAckReceiver:
+    def __init__(self, ack_tracker: ControlAckTracker) -> None:
+        self._ack_tracker = ack_tracker
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._started = False
+        self._session = None
+        self._session_cls = None
+        self._msg_type_text = None
+        self._msg_type_error = None
+        self._control_defaults: dict[str, Any] = {}
+        self._closing = threading.Event()
+        self._last_error = ""
+        self._last_server_error = ""
+        self._registered = False
+        self._reconnect_count = 0
+        self._ever_connected = False
+        self._received_messages = 0
+        self._received_bytes = 0
+        self._accepted_count = 0
+        self._pending_missing_count = 0
+        self._invalid_message_id_count = 0
+        self._invalid_timing_count = 0
+        self._unexpected_message_type_count = 0
+        self._unexpected_sender_count = 0
+        self._sender_mismatch_accepted_count = 0
+        self._payload_decode_errors = 0
+        self._last_msg_type: int | None = None
+        self._last_from_peer = ""
+        self._last_payload_preview = ""
+        self._last_ack_result = ""
+        self._load_backend()
+
+    def _load_backend(self) -> None:
+        try:
+            self._import_backend()
+        except Exception as error:  # pragma: no cover
+            self._last_error = f"omnisocket import failed: {error}"
+
+    def _import_backend(self) -> None:
+        try:
+            from omnisocket import CONTROL_DEFAULTS, MSG_TYPE_ERROR, MSG_TYPE_TEXT, Session  # type: ignore
+        except ImportError:
+            python_dir = WORKSPACE_ROOT / "OmniSocketGo" / "python"
+            if python_dir.exists():
+                sys.path.insert(0, str(python_dir))
+            from omnisocket import CONTROL_DEFAULTS, MSG_TYPE_ERROR, MSG_TYPE_TEXT, Session  # type: ignore
+
+        self._session_cls = Session
+        self._msg_type_text = MSG_TYPE_TEXT
+        self._msg_type_error = MSG_TYPE_ERROR
+        self._control_defaults = dict(CONTROL_DEFAULTS)
+
+    def _connect_session(self):
+        assert self._session_cls is not None
+
+        config = load_omnisocket_config()
+        transport_cfg = config.get("transport", {})
+        ack_cfg = config.get("control_ack_receiver", {})
+
+        session = self._session_cls()
+        session.connect(
+            server_addr=str(transport_cfg.get("server_addr", "127.0.0.1:10909")),
+            peer_id=str(ack_cfg.get("peer_id", "peer-a-ctrl-ack")),
+            relay_via=str(transport_cfg.get("relay_via", "")),
+            bind_ip=str(transport_cfg.get("bind_ip", "")),
+            bind_device=str(transport_cfg.get("bind_device", "")),
+            **self._control_defaults,
+        )
+        return session, str(ack_cfg.get("expected_sender", "peer-b-ctrl-ack"))
+
+    def ensure_started(self) -> None:
+        if self._session_cls is None:
+            return
+        with self._lock:
+            if self._started or self._closing.is_set():
+                return
+            self._started = True
+            self._thread = threading.Thread(target=self._run, name="omnisocket-control-ack", daemon=True)
+            self._thread.start()
+
+    @staticmethod
+    def _looks_like_control_ack(ack_payload: Any) -> bool:
+        if not isinstance(ack_payload, dict):
+            return False
+        if "message_id" not in ack_payload:
+            return False
+        return any(field in ack_payload for field in ("ack_phase", "b_recv_to_persist_us", "unix_send_ok", "sample_reason"))
+
+    @staticmethod
+    def _sender_matches(expected_sender: str, from_peer: str) -> bool:
+        normalized_expected = expected_sender.strip()
+        normalized_from = from_peer.strip()
+        if not normalized_expected:
+            return True
+        return normalized_from == normalized_expected
+
+    def _run(self) -> None:
+        while not self._closing.is_set():
+            expected_sender = ""
+            try:
+                session, expected_sender = self._connect_session()
+                with self._lock:
+                    self._session = session
+                    self._last_error = ""
+                    if self._ever_connected:
+                        self._reconnect_count += 1
+                    else:
+                        self._ever_connected = True
+
+                while not self._closing.is_set():
+                    result = session.recv(timeout_ms=1000)
+                    if result is None:
+                        try:
+                            session_stats = dict(session.stats())
+                        except Exception:
+                            session_stats = {}
+                        with self._lock:
+                            self._registered = bool(session_stats.get("registered", 0))
+                            self._last_server_error = str(session_stats.get("last_server_error", "") or "")
+                        continue
+                    from_peer, msg_type, payload = result
+                    with self._lock:
+                        self._received_messages += 1
+                        self._received_bytes += len(payload)
+                        self._last_from_peer = str(from_peer or "")
+                        self._last_msg_type = int(msg_type)
+                        self._last_payload_preview = _payload_preview(payload)
+                        try:
+                            session_stats = dict(session.stats())
+                        except Exception:
+                            session_stats = {}
+                        self._registered = bool(session_stats.get("registered", 0))
+                        self._last_server_error = str(session_stats.get("last_server_error", "") or "")
+                    if msg_type == self._msg_type_error:
+                        with self._lock:
+                            self._last_error = f"ack session error from {from_peer}: {payload.decode('utf-8', errors='replace')}"
+                            self._last_ack_result = "server_error"
+                        continue
+                    if msg_type != self._msg_type_text:
+                        with self._lock:
+                            self._unexpected_message_type_count += 1
+                            self._last_ack_result = "unexpected_message_type"
+                        continue
+                    try:
+                        ack_payload = json.loads(payload.decode("utf-8"))
+                    except (UnicodeDecodeError, json.JSONDecodeError):
+                        with self._lock:
+                            self._payload_decode_errors += 1
+                            self._last_ack_result = "payload_decode_error"
+                        continue
+                    sender_matches = self._sender_matches(expected_sender, from_peer)
+                    if not sender_matches:
+                        with self._lock:
+                            self._unexpected_sender_count += 1
+                        if not self._looks_like_control_ack(ack_payload):
+                            with self._lock:
+                                self._last_ack_result = "unexpected_sender"
+                            continue
+                    ack_result = self._ack_tracker.handle_ack(ack_payload, time.time_ns(), time.monotonic_ns())
+                    with self._lock:
+                        self._last_ack_result = ack_result
+                        if ack_result == "accepted":
+                            self._accepted_count += 1
+                            if not sender_matches:
+                                self._sender_mismatch_accepted_count += 1
+                        elif ack_result == "pending_missing":
+                            self._pending_missing_count += 1
+                        elif ack_result == "invalid_message_id":
+                            self._invalid_message_id_count += 1
+                        elif ack_result == "invalid_timing":
+                            self._invalid_timing_count += 1
+            except Exception as error:  # pragma: no cover
+                if not self._closing.is_set():
+                    with self._lock:
+                        self._last_error = str(error)
+                    time.sleep(2)
+            finally:
+                if self._session is not None:
+                    try:
+                        self._session.close()
+                    except Exception:
+                        pass
+                with self._lock:
+                    self._session = None
+                    if self._closing.is_set():
+                        self._started = False
+
+    def get_status(self) -> dict[str, Any]:
+        config = load_omnisocket_config().get("control_ack_receiver", {})
+        with self._lock:
+            session = self._session
+        if session is not None:
+            try:
+                session_stats = dict(session.stats())
+            except Exception:
+                session_stats = {}
+        else:
+            session_stats = {}
+        with self._lock:
+            return {
+                "backend_ready": self._session_cls is not None,
+                "started": self._started,
+                "connected": self._session is not None,
+                "registered": bool(session_stats.get("registered", 0) or self._registered),
+                "peer_id": str(config.get("peer_id", "")),
+                "expected_sender": str(config.get("expected_sender", "")),
+                "recv_calls": int(session_stats.get("recv_calls", 0)),
+                "recv_bytes": int(session_stats.get("recv_bytes", 0)),
+                "recv_timeouts": int(session_stats.get("recv_timeouts", 0)),
+                "recv_errors": int(session_stats.get("recv_errors", 0)),
+                "received_messages": self._received_messages,
+                "received_bytes": self._received_bytes,
+                "accepted_count": self._accepted_count,
+                "pending_missing_count": self._pending_missing_count,
+                "invalid_message_id_count": self._invalid_message_id_count,
+                "invalid_timing_count": self._invalid_timing_count,
+                "unexpected_message_type_count": self._unexpected_message_type_count,
+                "unexpected_sender_count": self._unexpected_sender_count,
+                "sender_mismatch_accepted_count": self._sender_mismatch_accepted_count,
+                "payload_decode_errors": self._payload_decode_errors,
+                "last_msg_type": self._last_msg_type,
+                "last_from_peer": self._last_from_peer,
+                "last_payload_preview": self._last_payload_preview,
+                "last_ack_result": self._last_ack_result,
+                "reconnect_count": self._reconnect_count,
+                "last_server_error": str(session_stats.get("last_server_error", "") or self._last_server_error),
+                "last_error": self._last_error,
+            }
+
+    def close(self) -> None:
+        self._closing.set()
+        with self._lock:
+            session = self._session
+        if session is not None:
+            try:
+                session.close()
+            except Exception:
+                pass
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)
+
+
+class ControlArbiter:
+    def __init__(self, sender: OmniSocketControlSender) -> None:
+        self._sender = sender
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._closing = threading.Event()
+        self._started = False
+        self._source_lease_ms = 300
+        self._send_rate_hz = 20.0
+        self._zero_burst_packets = 3
+        self._latest_by_source: dict[str, tuple[bytes, float]] = {}
+        self._packet_counts = {source: 0 for source in CONTROL_SOURCE_PRIORITY}
+        self._last_payload = ZERO_CONTROL_PAYLOAD
+        self._last_sent_at = 0.0
+        self._active_source: str | None = None
+        self._last_error = ""
+
+    def _load_config(self) -> None:
+        cfg = load_omnisocket_config().get("control_ingress", {})
+        self._source_lease_ms = max(50, int(cfg.get("source_lease_ms", 300)))
+        self._send_rate_hz = max(1.0, float(cfg.get("send_rate_hz", 20.0)))
+        self._zero_burst_packets = max(1, int(cfg.get("zero_burst_packets", 3)))
+
+    def ensure_started(self) -> None:
+        self._load_config()
+        with self._lock:
+            if self._closing.is_set():
+                return
+            if self._started:
+                return
+            self._started = True
+            self._thread = threading.Thread(
+                target=self._send_loop,
+                name="control-arbiter",
+                daemon=True,
+            )
+            self._thread.start()
+
+    def ingest_command(self, source: str, payload: bytes) -> None:
+        if source not in CONTROL_SOURCE_PRIORITY:
+            raise ValueError(f"unsupported control source: {source}")
+        if len(payload) != CONTROL_PACKET_SIZE:
+            raise ValueError(f"expected {CONTROL_PACKET_SIZE} bytes, got {len(payload)}")
+
+        self.ensure_started()
+        now = time.monotonic()
+        with self._lock:
+            self._latest_by_source[source] = (payload, now)
+            self._packet_counts[source] += 1
+
+    def _resolve_active_locked(self, now: float) -> tuple[str | None, bytes, int]:
+        lease_seconds = self._source_lease_ms / 1000.0
+        expired_sources = [
+            source
+            for source, (_, updated_at) in self._latest_by_source.items()
+            if (now - updated_at) > lease_seconds
+        ]
+        for source in expired_sources:
+            self._latest_by_source.pop(source, None)
+
+        for source in CONTROL_SOURCE_PRIORITY:
+            entry = self._latest_by_source.get(source)
+            if entry is None:
+                continue
+            payload, updated_at = entry
+            remaining_ms = max(0, int((lease_seconds - (now - updated_at)) * 1000))
+            return source, payload, remaining_ms
+
+        return None, ZERO_CONTROL_PAYLOAD, 0
+
+    def _send_loop(self) -> None:
+        interval = 1.0 / max(self._send_rate_hz, 1.0)
+        previous_active: str | None = None
+
+        while not self._closing.is_set():
+            now = time.monotonic()
+            with self._lock:
+                active_source, payload, _lease_ms = self._resolve_active_locked(now)
+                self._active_source = active_source
+                self._last_payload = payload
+
+            if previous_active is not None and active_source is None:
+                try:
+                    self._sender.send_zero_burst(self._zero_burst_packets)
+                except Exception as error:
+                    with self._lock:
+                        self._last_error = str(error)
+            elif active_source is not None:
+                try:
+                    self._sender.send_payload(payload, source=active_source)
+                    with self._lock:
+                        self._last_sent_at = time.monotonic()
+                        self._last_error = ""
+                except Exception as error:
+                    with self._lock:
+                        self._last_error = str(error)
+
+            previous_active = active_source
+            self._closing.wait(interval)
+
+        try:
+            self._sender.send_zero_burst(self._zero_burst_packets)
+        except Exception:
+            pass
+
+    def get_status(self) -> dict[str, Any]:
+        self.ensure_started()
+        now = time.monotonic()
+        with self._lock:
+            active_source, _payload, lease_ms = self._resolve_active_locked(now)
+            return {
+                "active_source": active_source,
+                "control_lease_remaining_ms": lease_ms,
+                "packet_counts": dict(self._packet_counts),
+                "send_rate_hz": self._send_rate_hz,
+                "source_lease_ms": self._source_lease_ms,
+                "zero_burst_packets": self._zero_burst_packets,
+                "last_error": self._last_error,
+                "last_sent_at_monotonic": self._last_sent_at,
+            }
+
+    def close(self) -> None:
+        self._closing.set()
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)
+
+
+class NativeUdpControlIngress:
+    def __init__(self, arbiter: ControlArbiter) -> None:
+        self._arbiter = arbiter
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._closing = threading.Event()
+        self._started = False
+        self._bind_addr = "127.0.0.1:10921"
+        self._packets_received = 0
+        self._invalid_packets = 0
+        self._last_sender = ""
+        self._last_error = ""
+
+    def ensure_started(self) -> None:
+        bind_addr = str(load_omnisocket_config().get("control_ingress", {}).get("native_udp_bind", "127.0.0.1:10921"))
+        with self._lock:
+            self._bind_addr = bind_addr
+            if self._closing.is_set():
+                return
+            if self._thread is not None and self._thread.is_alive():
+                return
+            self._started = True
+            self._thread = threading.Thread(
+                target=self._run,
+                name="native-udp-control-ingress",
+                daemon=True,
+            )
+            self._thread.start()
+
+    def _run(self) -> None:
+        try:
+            try:
+                host, port = parse_host_port(self._bind_addr)
+                sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                sock.bind((host, port))
+                sock.settimeout(0.1)
+            except Exception as error:
+                with self._lock:
+                    self._last_error = str(error)
+                return
+
+            with sock:
+                while not self._closing.is_set():
+                    try:
+                        payload, sender_addr = sock.recvfrom(CONTROL_PACKET_SIZE + 64)
+                    except socket.timeout:
+                        continue
+                    except OSError as error:
+                        with self._lock:
+                            if not self._closing.is_set():
+                                self._last_error = str(error)
+                        return
+
+                    with self._lock:
+                        self._last_sender = f"{sender_addr[0]}:{sender_addr[1]}"
+
+                    if len(payload) != CONTROL_PACKET_SIZE:
+                        with self._lock:
+                            self._invalid_packets += 1
+                        continue
+
+                    try:
+                        self._arbiter.ingest_command(CONTROL_SOURCE_NATIVE_UDP, payload)
+                    except Exception as error:
+                        with self._lock:
+                            self._last_error = str(error)
+                        continue
+
+                    with self._lock:
+                        self._packets_received += 1
+        finally:
+            with self._lock:
+                self._started = False
+                self._thread = None
+
+    def get_status(self) -> dict[str, Any]:
+        self.ensure_started()
+        with self._lock:
+            return {
+                "started": self._started,
+                "bind_addr": self._bind_addr,
+                "packets_received": self._packets_received,
+                "invalid_packets": self._invalid_packets,
+                "last_sender": self._last_sender,
+                "last_error": self._last_error,
+            }
+
+    def close(self) -> None:
+        self._closing.set()
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)

host/robot-command-center/backend/monitoring/routing.py

@@ -0,0 +1,9 @@
+from django.urls import re_path
+
+from .consumers import ControlConsumer
+
+
+websocket_urlpatterns = [
+    re_path(r"^ws/control/$", ControlConsumer.as_asgi()),
+]
+

host/robot-command-center/backend/monitoring/services.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import atexit
+
+from .control import ControlAckTracker, ControlArbiter, NativeUdpControlIngress, OmniSocketControlAckReceiver, OmniSocketControlSender
+from .telemetry import GpsDataService, HubTelemetryReceiver, NetworkTelemetryService
+from .video import OmniSocketVideoReceiver, VideoDisplayProbeStore, VideoFrameService
+
+
+_video_receiver = OmniSocketVideoReceiver()
+_control_ack_tracker = ControlAckTracker()
+_control_sender = OmniSocketControlSender(_control_ack_tracker)
+_control_ack_receiver = OmniSocketControlAckReceiver(_control_ack_tracker)
+_hub_telemetry_receiver = HubTelemetryReceiver()
+_video_display_probe_store = VideoDisplayProbeStore()
+
+control_arbiter = ControlArbiter(_control_sender)
+camera_control_service = _control_sender
+native_control_ingress = NativeUdpControlIngress(control_arbiter)
+
+video_service = VideoFrameService(_video_receiver, _video_display_probe_store)
+gps_service = GpsDataService(_video_receiver)
+network_service = NetworkTelemetryService(
+    _video_receiver,
+    _control_sender,
+    _control_ack_tracker,
+    _control_ack_receiver,
+    control_arbiter,
+    native_control_ingress,
+    _hub_telemetry_receiver,
+    _video_display_probe_store,
+)
+
+
+def shutdown_monitoring_services() -> None:
+    for closer in (
+        network_service.close,
+        native_control_ingress.close,
+        control_arbiter.close,
+        _control_ack_receiver.close,
+        _control_ack_tracker.close,
+        _hub_telemetry_receiver.close,
+        _video_display_probe_store.close,
+        _video_receiver.close,
+        _control_sender.close,
+    ):
+        try:
+            closer()
+        except Exception:
+            pass
+
+
+atexit.register(shutdown_monitoring_services)

host/robot-command-center/backend/monitoring/telemetry.py

@@ -0,0 +1,903 @@
+from __future__ import annotations
+
+from collections import deque
+import json
+import os
+import sys
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from .common import (
+    VIDEO_TRAILER_COORDINATE_FORMAT,
+    WORKSPACE_ROOT,
+    load_omnisocket_config,
+    utc_iso_now,
+)
+from .control import ControlAckTracker, ControlArbiter, NativeUdpControlIngress, OmniSocketControlAckReceiver, OmniSocketControlSender
+from .video import FrameTrailerMetadata, OmniSocketVideoReceiver, VideoDisplayProbeStore
+
+
+LOCAL_SAMPLE_INTERVAL_MS = 500
+TREND_HISTORY_SIZE = 10
+TREND_WINDOW_SIZE = 5
+BLITZ_RUNTIME_DIR = Path(os.getenv("BLITZ_RUNTIME_DIR", "/run/blitz-robot"))
+WATCHDOG_STATUS_PATH = BLITZ_RUNTIME_DIR / "watchdog.status.json"
+WATCHDOG_STATUS_STALE_MS = max(int(os.getenv("BLITZ_HEALTH_STALE_SEC", "15")), 1) * 1000
+WATCHDOG_FAULT_REASON_MAP: dict[str, tuple[str, str | None]] = {
+    "": ("none", None),
+    "none": ("none", None),
+    "camera_missing": ("video_pipeline_stalled", "degraded"),
+    "camera_recovered": ("video_session_recovering", "recovering"),
+    "camera-reappeared-escalated": ("video_session_recovering", "recovering"),
+    "bside_status_stale": ("video_session_recovering", "recovering"),
+    "bside-unhealthy-escalated": ("video_session_recovering", "recovering"),
+    "ros_receiver_unhealthy": ("control_session_recovering", "recovering"),
+    "ros-unhealthy": ("control_session_recovering", "recovering"),
+    "network_or_robot_unreachable": ("network_or_robot_unreachable", "recovering"),
+    "network-recovered-ros-unhealthy": ("control_session_recovering", "recovering"),
+    "network-recovered-escalated": ("network_or_robot_unreachable", "recovering"),
+}
+
+
+def _utc_from_epoch(epoch_seconds: float | None) -> str | None:
+    if epoch_seconds is None or epoch_seconds <= 0.0:
+        return None
+    return datetime.fromtimestamp(epoch_seconds, timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def _coerce_int(value: Any, default: int = 0) -> int:
+    try:
+        if value is None:
+            return default
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_float(value: Any, default: float = 0.0) -> float:
+    try:
+        if value is None:
+            return default
+        return float(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _load_optional_json(path: Path) -> dict[str, Any] | None:
+    try:
+        if not path.exists():
+            return None
+        with path.open("r", encoding="utf-8") as file:
+            payload = json.load(file)
+        if isinstance(payload, dict):
+            return payload
+    except Exception:
+        return None
+    return None
+
+
+class GpsDataService:
+    def __init__(self, receiver: OmniSocketVideoReceiver) -> None:
+        self._receiver = receiver
+
+    def get_latest(self) -> dict[str, Any]:
+        metadata = self._receiver.get_latest_frame_metadata()
+        if metadata is None:
+            return self._build_waiting_payload()
+        return self._build_payload_from_metadata(metadata)
+
+    def _build_waiting_payload(self) -> dict[str, Any]:
+        return {
+            "has_fix": False,
+            "utc_time": "--:--:--",
+            "latitude": None,
+            "longitude": None,
+            "satellites": None,
+            "altitude_m": None,
+            "coordinate_system": "WGS84",
+            "source_sentence": "VIDEO_TRAILER",
+            "raw_coordinate_format": VIDEO_TRAILER_COORDINATE_FORMAT,
+            "source_mode": "video-frame-trailer-waiting",
+            "updated_at": "",
+        }
+
+    def _build_payload_from_metadata(self, metadata: FrameTrailerMetadata) -> dict[str, Any]:
+        updated_at = _utc_from_epoch(metadata.received_at) or ""
+        if not metadata.has_gps_fix:
+            return {
+                "has_fix": False,
+                "utc_time": "--:--:--",
+                "latitude": None,
+                "longitude": None,
+                "raw_latitude_hex": f"0x{metadata.raw_latitude_hex}",
+                "raw_longitude_hex": f"0x{metadata.raw_longitude_hex}",
+                "satellites": None,
+                "altitude_m": None,
+                "coordinate_system": "WGS84",
+                "source_sentence": "VIDEO_TRAILER",
+                "raw_coordinate_format": VIDEO_TRAILER_COORDINATE_FORMAT,
+                "source_mode": "video-frame-trailer-no-fix",
+                "updated_at": updated_at,
+            }
+
+        timestamp_seconds = metadata.timestamp_ns / 1_000_000_000
+        return {
+            "has_fix": True,
+            "utc_time": datetime.fromtimestamp(timestamp_seconds, timezone.utc).strftime("%H:%M:%S"),
+            "latitude": round(metadata.latitude, 6) if metadata.latitude is not None else None,
+            "longitude": round(metadata.longitude, 6) if metadata.longitude is not None else None,
+            "raw_latitude_hex": f"0x{metadata.raw_latitude_hex}",
+            "raw_longitude_hex": f"0x{metadata.raw_longitude_hex}",
+            "satellites": None,
+            "altitude_m": None,
+            "coordinate_system": "WGS84",
+            "source_sentence": "VIDEO_TRAILER",
+            "raw_coordinate_format": VIDEO_TRAILER_COORDINATE_FORMAT,
+            "source_mode": "video-frame-trailer",
+            "updated_at": updated_at,
+        }
+
+
+class KcpTrendTracker:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._samples: dict[str, deque[dict[str, Any]]] = {}
+
+    def _normalize(self, stats: dict[str, Any] | None) -> dict[str, Any]:
+        raw = dict(stats or {})
+        snd_wnd = _coerce_int(raw.get("snd_wnd"))
+        rmt_wnd = _coerce_int(raw.get("rmt_wnd"))
+        inflight = _coerce_int(raw.get("inflight"))
+        window_limit = _coerce_int(raw.get("window_limit"), min(snd_wnd, rmt_wnd) if snd_wnd and rmt_wnd else 0)
+        return {
+            "connected": _coerce_int(raw.get("connected")),
+            "conv": _coerce_int(raw.get("conv")),
+            "rto_ms": _coerce_int(raw.get("rto_ms")),
+            "srtt_ms": _coerce_int(raw.get("srtt_ms")),
+            "min_srtt_ms": _coerce_int(raw.get("min_srtt_ms")),
+            "srttvar_ms": _coerce_int(raw.get("srttvar_ms")),
+            "last_feedback_age_ms": _coerce_int(raw.get("last_feedback_age_ms")),
+            "snd_wnd": snd_wnd,
+            "rmt_wnd": rmt_wnd,
+            "inflight": inflight,
+            "window_limit": window_limit,
+            "window_pressure_pct": round(_coerce_float(raw.get("window_pressure_pct")), 3),
+            "snd_queue": _coerce_int(raw.get("snd_queue")),
+            "rcv_queue": _coerce_int(raw.get("rcv_queue")),
+            "snd_buffer": _coerce_int(raw.get("snd_buffer")),
+            "out_segs_total": _coerce_int(raw.get("out_segs_total")),
+            "retrans_total": _coerce_int(raw.get("retrans_total")),
+            "fast_retrans_total": _coerce_int(raw.get("fast_retrans_total")),
+            "lost_total": _coerce_int(raw.get("lost_total")),
+            "repeat_total": _coerce_int(raw.get("repeat_total")),
+            "xmit_total": _coerce_int(raw.get("xmit_total")),
+        }
+
+    def add_sample(self, key: str, stats: dict[str, Any] | None) -> None:
+        sample = {
+            "ts_monotonic": time.monotonic(),
+            "updated_at": utc_iso_now(),
+            "stats": self._normalize(stats),
+        }
+        with self._lock:
+            history = self._samples.setdefault(key, deque(maxlen=TREND_HISTORY_SIZE))
+            history.append(sample)
+
+    def latest_updated_at(self, key: str) -> str | None:
+        with self._lock:
+            history = self._samples.get(key)
+            if not history:
+                return None
+            return str(history[-1].get("updated_at") or "")
+
+    def describe(self, key: str, current_stats: dict[str, Any] | None) -> dict[str, Any]:
+        current = self._normalize(current_stats)
+        with self._lock:
+            history = list(self._samples.get(key, ()))
+
+        timeline = history + [{"stats": current, "updated_at": utc_iso_now()}]
+        previous = timeline[-2]["stats"] if len(timeline) >= 2 else None
+        trend_window = [entry["stats"] for entry in timeline[-TREND_WINDOW_SIZE:]]
+        deadband = max(2.0, 0.05 * float(max(current.get("window_limit", 0), 1)))
+
+        snd_queue_delta = 0
+        snd_buffer_delta = 0
+        retrans_delta = 0
+        fast_retrans_delta = 0
+        lost_delta = 0
+        repeat_delta = 0
+        out_segs_delta = 0
+        if previous is not None:
+            snd_queue_delta = max(0, current["snd_queue"] - _coerce_int(previous.get("snd_queue")))
+            snd_buffer_delta = max(0, current["snd_buffer"] - _coerce_int(previous.get("snd_buffer")))
+            retrans_delta = max(0, current["retrans_total"] - _coerce_int(previous.get("retrans_total")))
+            fast_retrans_delta = max(0, current["fast_retrans_total"] - _coerce_int(previous.get("fast_retrans_total")))
+            lost_delta = max(0, current["lost_total"] - _coerce_int(previous.get("lost_total")))
+            repeat_delta = max(0, current["repeat_total"] - _coerce_int(previous.get("repeat_total")))
+            out_segs_delta = max(0, current["out_segs_total"] - _coerce_int(previous.get("out_segs_total")))
+
+        def classify(field: str) -> str:
+            if len(trend_window) < 2:
+                return "stable"
+            oldest = float(_coerce_int(trend_window[0].get(field)))
+            newest = float(_coerce_int(trend_window[-1].get(field)))
+            delta = newest - oldest
+            if abs(delta) < deadband:
+                return "stable"
+            return "rising" if delta > 0 else "falling"
+
+        repair_rate_pct = 0.0
+        if out_segs_delta > 0:
+            repair_rate_pct = round((retrans_delta / out_segs_delta) * 100.0, 3)
+
+        return {
+            "kcp": current,
+            "trend": {
+                "snd_queue_delta": snd_queue_delta,
+                "snd_buffer_delta": snd_buffer_delta,
+                "snd_queue_trend": classify("snd_queue"),
+                "snd_buffer_trend": classify("snd_buffer"),
+                "retrans_delta": retrans_delta,
+                "fast_retrans_delta": fast_retrans_delta,
+                "lost_delta": lost_delta,
+                "repeat_delta": repeat_delta,
+                "out_segs_delta": out_segs_delta,
+                "repair_rate_pct": repair_rate_pct,
+            },
+        }
+
+
+class HubTelemetryReceiver:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._started = False
+        self._session = None
+        self._session_cls = None
+        self._msg_type_text = None
+        self._msg_type_error = None
+        self._telemetry_defaults: dict[str, Any] = {}
+        self._latest_snapshot: dict[str, Any] | None = None
+        self._last_error = ""
+        self._last_received_wall = 0.0
+        self._last_received_monotonic = 0.0
+        self._reconnect_count = 0
+        self._ever_connected = False
+        self._closing = threading.Event()
+        self._load_backend()
+
+    def _load_backend(self) -> None:
+        try:
+            self._import_backend()
+        except Exception as error:  # pragma: no cover - optional runtime dependency
+            self._last_error = f"omnisocket import failed: {error}"
+
+    def _import_backend(self) -> None:
+        try:
+            from omnisocket import MSG_TYPE_ERROR, MSG_TYPE_TEXT, Session, TELEMETRY_DEFAULTS  # type: ignore
+        except ImportError:
+            python_dir = WORKSPACE_ROOT / "OmniSocketGo" / "python"
+            if python_dir.exists():
+                sys.path.insert(0, str(python_dir))
+            from omnisocket import MSG_TYPE_ERROR, MSG_TYPE_TEXT, Session, TELEMETRY_DEFAULTS  # type: ignore
+
+        self._msg_type_error = MSG_TYPE_ERROR
+        self._msg_type_text = MSG_TYPE_TEXT
+        self._session_cls = Session
+        self._telemetry_defaults = dict(TELEMETRY_DEFAULTS)
+
+    def _connect_session(self):
+        assert self._session_cls is not None
+
+        config = load_omnisocket_config()
+        transport_cfg = config.get("transport", {})
+        telemetry_cfg = config.get("telemetry_receiver", {})
+
+        session = self._session_cls()
+        session.connect(
+            server_addr=str(transport_cfg.get("server_addr", "127.0.0.1:10909")),
+            peer_id=str(telemetry_cfg.get("peer_id", "peer-a-telemetry")),
+            relay_via=str(transport_cfg.get("relay_via", "")),
+            bind_ip=str(transport_cfg.get("bind_ip", "")),
+            bind_device=str(transport_cfg.get("bind_device", "")),
+            **self._telemetry_defaults,
+        )
+        return session
+
+    def ensure_started(self) -> None:
+        if self._session_cls is None:
+            return
+
+        with self._lock:
+            if self._started or self._closing.is_set():
+                return
+            self._started = True
+            self._thread = threading.Thread(
+                target=self._run,
+                name="hub-telemetry-receiver",
+                daemon=True,
+            )
+            self._thread.start()
+
+    def _run(self) -> None:
+        while not self._closing.is_set():
+            try:
+                session = self._connect_session()
+                with self._lock:
+                    self._session = session
+                    self._last_error = ""
+                    if self._ever_connected:
+                        self._reconnect_count += 1
+                    else:
+                        self._ever_connected = True
+
+                while not self._closing.is_set():
+                    result = session.recv(timeout_ms=1000)
+                    if result is None:
+                        continue
+
+                    from_peer, msg_type, payload = result
+                    if msg_type == self._msg_type_error:
+                        with self._lock:
+                            self._last_error = f"hub error from {from_peer}: {payload.decode('utf-8', errors='replace')}"
+                        continue
+                    if msg_type != self._msg_type_text:
+                        continue
+
+                    snapshot = json.loads(payload.decode("utf-8"))
+                    if snapshot.get("type") != "hub_kcp_snapshot":
+                        continue
+
+                    now_wall = time.time()
+                    now_mono = time.monotonic()
+                    with self._lock:
+                        self._latest_snapshot = snapshot
+                        self._last_received_wall = now_wall
+                        self._last_received_monotonic = now_mono
+                        self._last_error = ""
+            except Exception as error:  # pragma: no cover - runtime integration path
+                if not self._closing.is_set():
+                    session_error = ""
+                    if self._session is not None:
+                        try:
+                            session_error = str(dict(self._session.stats()).get("last_server_error", "") or "")
+                        except Exception:
+                            session_error = ""
+                    with self._lock:
+                        self._last_error = session_error or str(error)
+            finally:
+                with self._lock:
+                    session = self._session
+                    self._session = None
+                    if self._closing.is_set():
+                        self._started = False
+                if session is not None:
+                    try:
+                        session.close()
+                    except Exception:
+                        pass
+                if not self._closing.is_set():
+                    time.sleep(2)
+
+    def get_snapshot(self) -> dict[str, Any]:
+        self.ensure_started()
+        cfg = load_omnisocket_config().get("telemetry_receiver", {})
+        stale_after_ms = max(500, int(cfg.get("stale_after_ms", 1500)))
+
+        with self._lock:
+            received_monotonic = self._last_received_monotonic
+            received_wall = self._last_received_wall
+            snapshot = self._latest_snapshot
+            connected = self._session is not None
+            last_error = self._last_error
+            reconnect_count = self._reconnect_count
+            if self._session is not None:
+                try:
+                    session_stats = dict(self._session.stats())
+                except Exception:
+                    session_stats = {}
+            else:
+                session_stats = {}
+
+        stale = True
+        if received_monotonic > 0.0:
+            stale = (time.monotonic() - received_monotonic) * 1000.0 > stale_after_ms
+
+        return {
+            "connected": connected,
+            "updated_at": _utc_from_epoch(received_wall),
+            "received_at_monotonic": received_monotonic,
+            "stale": stale,
+            "peer_id": str(cfg.get("peer_id", "peer-a-telemetry")),
+            "snapshot": snapshot or {"sessions": []},
+            "last_error": last_error,
+            "registered": bool(session_stats.get("registered", 0)),
+            "last_server_error": str(session_stats.get("last_server_error", "") or ""),
+            "reconnect_count": reconnect_count,
+        }
+
+    def close(self) -> None:
+        self._closing.set()
+        with self._lock:
+            session = self._session
+        if session is not None:
+            try:
+                session.close()
+            except Exception:
+                pass
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)
+
+
+class NetworkTelemetryService:
+    def __init__(
+        self,
+        video_receiver: OmniSocketVideoReceiver,
+        control_sender: OmniSocketControlSender,
+        control_ack_tracker: ControlAckTracker,
+        control_ack_receiver: OmniSocketControlAckReceiver,
+        control_arbiter: ControlArbiter,
+        native_ingress: NativeUdpControlIngress,
+        hub_receiver: HubTelemetryReceiver,
+        video_display_probe_store: VideoDisplayProbeStore,
+    ) -> None:
+        self._video_receiver = video_receiver
+        self._control_sender = control_sender
+        self._control_ack_tracker = control_ack_tracker
+        self._control_ack_receiver = control_ack_receiver
+        self._control_arbiter = control_arbiter
+        self._native_ingress = native_ingress
+        self._hub_receiver = hub_receiver
+        self._video_display_probe_store = video_display_probe_store
+        self._trend_tracker = KcpTrendTracker()
+        self._rate_lock = threading.Lock()
+        self._last_rate_sample: tuple[float, int, int] | None = None
+        self._sample_thread: threading.Thread | None = None
+        self._sample_started = False
+        self._last_remote_snapshot_at = 0.0
+        self._closing = threading.Event()
+
+    def _ensure_started(self) -> None:
+        self._video_receiver.ensure_started()
+        self._control_arbiter.ensure_started()
+        self._control_ack_receiver.ensure_started()
+        self._native_ingress.ensure_started()
+        self._hub_receiver.ensure_started()
+        with self._rate_lock:
+            if self._sample_started or self._closing.is_set():
+                return
+            self._sample_started = True
+            self._sample_thread = threading.Thread(
+                target=self._sample_loop,
+                name="network-telemetry-sampler",
+                daemon=True,
+            )
+            self._sample_thread.start()
+
+    def _sample_loop(self) -> None:
+        interval_seconds = LOCAL_SAMPLE_INTERVAL_MS / 1000.0
+        while not self._closing.is_set():
+            try:
+                self._trend_tracker.add_sample("a_to_d.video", self._video_receiver.session_kcp_stats())
+                self._trend_tracker.add_sample("a_to_d.control", self._control_sender.session_kcp_stats())
+            except Exception:
+                pass
+            time.sleep(interval_seconds)
+
+    def _compute_rates(self, send_bytes: int, recv_bytes: int) -> tuple[float, float]:
+        now = time.monotonic()
+        with self._rate_lock:
+            previous = self._last_rate_sample
+            self._last_rate_sample = (now, send_bytes, recv_bytes)
+
+        if previous is None:
+            return 0.0, 0.0
+
+        prev_time, prev_send, prev_recv = previous
+        elapsed = now - prev_time
+        if elapsed <= 0.0:
+            return 0.0, 0.0
+
+        tx_kbps = max(0.0, ((send_bytes - prev_send) * 8.0) / elapsed / 1000.0)
+        rx_kbps = max(0.0, ((recv_bytes - prev_recv) * 8.0) / elapsed / 1000.0)
+        return tx_kbps, rx_kbps
+
+    def _ingest_remote_snapshot(self, telemetry_state: dict[str, Any]) -> None:
+        received_at = float(telemetry_state.get("received_at_monotonic") or 0.0)
+        if received_at <= 0.0 or received_at <= self._last_remote_snapshot_at:
+            return
+
+        snapshot = telemetry_state.get("snapshot") or {}
+        sessions = snapshot.get("sessions") or []
+        for session in sessions:
+            peer_id = str(session.get("peer_id", "")).strip()
+            if not peer_id:
+                continue
+            self._trend_tracker.add_sample(f"hub::{peer_id}", session)
+        self._last_remote_snapshot_at = received_at
+
+    def _build_session_payload(
+        self,
+        trend_key: str,
+        peer_id: str,
+        app_stats: dict[str, Any] | None,
+        current_kcp: dict[str, Any] | None,
+        updated_at: str | None,
+        stale: bool,
+    ) -> dict[str, Any]:
+        described = self._trend_tracker.describe(trend_key, current_kcp)
+        connected = bool(described["kcp"].get("connected"))
+        if app_stats is not None and "registered" in app_stats:
+            connected = bool(app_stats.get("registered"))
+        return {
+            "peer_id": peer_id,
+            "connected": connected,
+            "updated_at": updated_at,
+            "stale": stale,
+            "app": app_stats,
+            "kcp": described["kcp"],
+            "trend": described["trend"],
+        }
+
+    def _build_link(self, source: str, updated_at: str | None, stale: bool, sessions: dict[str, dict[str, Any]]) -> dict[str, Any]:
+        session_items = list(sessions.values())
+        active_sessions = [session for session in session_items if session.get("connected") and not session.get("stale")]
+        retrans_sum = sum(_coerce_int(session.get("trend", {}).get("retrans_delta")) for session in active_sessions)
+        out_segs_sum = sum(_coerce_int(session.get("trend", {}).get("out_segs_delta")) for session in active_sessions)
+        repair_rate_pct = round((retrans_sum / out_segs_sum) * 100.0, 3) if out_segs_sum > 0 else 0.0
+
+        return {
+            "source": source,
+            "updated_at": updated_at,
+            "stale": stale,
+            "aggregate": {
+                "online_sessions": len(active_sessions),
+                "max_window_pressure_pct": max(
+                    (_coerce_float(session.get("kcp", {}).get("window_pressure_pct")) for session in active_sessions),
+                    default=0.0,
+                ),
+                "sum_snd_queue": sum(_coerce_int(session.get("kcp", {}).get("snd_queue")) for session in active_sessions),
+                "sum_snd_buffer": sum(_coerce_int(session.get("kcp", {}).get("snd_buffer")) for session in active_sessions),
+                "sum_retrans_delta": retrans_sum,
+                "sum_out_segs_delta": out_segs_sum,
+                "repair_rate_pct": repair_rate_pct,
+            },
+            "sessions": sessions,
+        }
+
+    def _pick_primary_session(self, links: dict[str, dict[str, Any]]) -> dict[str, Any] | None:
+        candidates = (
+            links["a_to_d"]["sessions"]["control"],
+            links["a_to_d"]["sessions"]["video"],
+            links["d_to_b"]["sessions"]["control"],
+            links["d_to_b"]["sessions"]["video"],
+        )
+        for session in candidates:
+            if session.get("connected") and not session.get("stale"):
+                return session
+        return None
+
+    def _derive_robot_health(
+        self,
+        *,
+        video_receiver_status: dict[str, Any],
+        local_control_registered: bool,
+        remote_control_fresh: bool,
+        remote_video_fresh: bool,
+        telemetry_state: dict[str, Any],
+        watchdog_status: dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        if watchdog_status is not None:
+            explicit_health = self._derive_robot_health_from_watchdog(watchdog_status)
+            if explicit_health is not None:
+                return explicit_health
+
+        has_recent_frame = bool(video_receiver_status.get("has_recent_frame"))
+        telemetry_connected = bool(telemetry_state.get("connected"))
+        telemetry_stale = bool(telemetry_state.get("stale", True))
+
+        if has_recent_frame and remote_control_fresh and remote_video_fresh:
+            fault_reason = "none"
+            recovery_state = "ok"
+        elif not remote_control_fresh and not remote_video_fresh and not has_recent_frame:
+            fault_reason = "network_or_robot_unreachable"
+            recovery_state = "recovering" if telemetry_connected and not telemetry_stale else "degraded"
+        elif remote_control_fresh and not remote_video_fresh:
+            fault_reason = "video_session_recovering"
+            recovery_state = "recovering"
+        elif not remote_control_fresh and local_control_registered:
+            fault_reason = "control_session_recovering"
+            recovery_state = "recovering"
+        elif remote_control_fresh and not has_recent_frame:
+            fault_reason = "video_pipeline_stalled"
+            recovery_state = "degraded"
+        else:
+            fault_reason = "unknown"
+            recovery_state = "degraded"
+
+        return {
+            "fault_reason": fault_reason,
+            "recovery_state": recovery_state,
+            "confidence": "derived",
+            "updated_at": utc_iso_now(),
+        }
+
+    def _derive_robot_health_from_watchdog(self, watchdog_status: dict[str, Any]) -> dict[str, Any] | None:
+        updated_at_epoch_ms = _coerce_int(watchdog_status.get("updated_at_epoch_ms"))
+        if updated_at_epoch_ms <= 0:
+            return None
+
+        now_epoch_ms = int(time.time() * 1000)
+        if now_epoch_ms - updated_at_epoch_ms > WATCHDOG_STATUS_STALE_MS:
+            return None
+
+        raw_fault_reason = str(watchdog_status.get("fault_reason", "") or "")
+        raw_recovery_state = str(watchdog_status.get("recovery_state", "") or "")
+        normalized_fault_reason = "unknown"
+        normalized_recovery_state = raw_recovery_state or "degraded"
+        mapped_health = WATCHDOG_FAULT_REASON_MAP.get(raw_fault_reason)
+
+        if mapped_health is not None:
+            normalized_fault_reason, recovery_override = mapped_health
+            if recovery_override is not None:
+                normalized_recovery_state = recovery_override
+        if raw_recovery_state == "backoff":
+            normalized_recovery_state = "backoff"
+
+        return {
+            "fault_reason": normalized_fault_reason,
+            "recovery_state": normalized_recovery_state,
+            "confidence": "derived",
+            "updated_at": _utc_from_epoch(updated_at_epoch_ms / 1000.0) or utc_iso_now(),
+        }
+
+    def _derive_latency_estimate(
+        self,
+        *,
+        links: dict[str, dict[str, Any]],
+        video_receiver_status: dict[str, Any],
+        display_probe_status: dict[str, Any],
+    ) -> dict[str, Any]:
+        a_to_d_control_raw = links["a_to_d"]["sessions"]["control"]["kcp"].get("srtt_ms")
+        d_to_b_control_raw = links["d_to_b"]["sessions"]["control"]["kcp"].get("srtt_ms")
+        a_to_d_control_min_raw = links["a_to_d"]["sessions"]["control"]["kcp"].get("min_srtt_ms")
+        d_to_b_control_min_raw = links["d_to_b"]["sessions"]["control"]["kcp"].get("min_srtt_ms")
+        a_to_d_video_raw = links["a_to_d"]["sessions"]["video"]["kcp"].get("srtt_ms")
+        d_to_b_video_raw = links["d_to_b"]["sessions"]["video"]["kcp"].get("srtt_ms")
+
+        a_to_d_control = _coerce_float(a_to_d_control_raw) if a_to_d_control_raw is not None else None
+        d_to_b_control = _coerce_float(d_to_b_control_raw) if d_to_b_control_raw is not None else None
+        a_to_d_control_min = _coerce_float(a_to_d_control_min_raw) if a_to_d_control_min_raw is not None else None
+        d_to_b_control_min = _coerce_float(d_to_b_control_min_raw) if d_to_b_control_min_raw is not None else None
+        a_to_d_video = _coerce_float(a_to_d_video_raw) if a_to_d_video_raw is not None else None
+        d_to_b_video = _coerce_float(d_to_b_video_raw) if d_to_b_video_raw is not None else None
+        ack_estimate = self._control_ack_tracker.get_latest_estimate()
+        capture_to_send_raw = video_receiver_status.get("latest_capture_to_send_ms")
+        request_to_paint_raw = display_probe_status.get("request_to_paint_ms")
+        capture_to_send_ms = _coerce_float(capture_to_send_raw) if capture_to_send_raw is not None else None
+        request_to_paint_ms = _coerce_float(request_to_paint_raw) if request_to_paint_raw is not None else None
+        video_network_oneway_est_ms = (
+            round((a_to_d_video + d_to_b_video) / 2.0, 3)
+            if a_to_d_video is not None and d_to_b_video is not None
+            else None
+        )
+        video_partial_est_ms = None
+        if capture_to_send_ms is not None and video_network_oneway_est_ms is not None:
+            video_partial_est_ms = round(capture_to_send_ms + video_network_oneway_est_ms, 3)
+        video_e2e_est_ms = None
+        if video_partial_est_ms is not None and request_to_paint_ms is not None:
+            video_e2e_est_ms = round(video_partial_est_ms + request_to_paint_ms, 3)
+
+        return {
+            "control_loop_rtt_ms": ack_estimate.get("control_loop_rtt_ms"),
+            "control_to_persist_est_ms": ack_estimate.get("control_to_persist_est_ms"),
+            "control_oneway_srtt_est_ms": (
+                round((a_to_d_control + d_to_b_control) / 2.0, 3)
+                if a_to_d_control is not None and d_to_b_control is not None
+                else None
+            ),
+            "control_oneway_bestcase_est_ms": (
+                round((a_to_d_control_min + d_to_b_control_min) / 2.0, 3)
+                if a_to_d_control_min is not None and d_to_b_control_min is not None
+                else None
+            ),
+            "video_network_oneway_est_ms": video_network_oneway_est_ms,
+            "video_partial_est_ms": video_partial_est_ms,
+            "video_e2e_est_ms": video_e2e_est_ms,
+            "estimate_method": {
+                "control": "ack_loop" if ack_estimate.get("ack_available") else "srtt_fallback",
+                "video": "capture_to_send+srtt/2+request_to_paint" if video_e2e_est_ms is not None else "capture_to_send+srtt/2",
+            },
+            "clock_sync_required": False,
+            "assumptions": [
+                "control one-way estimate uses ACK loop when available",
+                "video one-way estimate uses per-leg SRTT and local paint timing",
+            ],
+            "confidence": {
+                "control": "derived_ack" if ack_estimate.get("ack_available") else "fallback_srtt",
+                "video": "derived_local_probe" if video_e2e_est_ms is not None else "partial_without_probe",
+            },
+        }
+
+    def get_latest(self) -> dict[str, Any]:
+        self._ensure_started()
+
+        config = load_omnisocket_config()
+        video_receiver_cfg = config.get("video_receiver", {})
+        control_sender_cfg = config.get("control_sender", {})
+        video_sender_cfg = config.get("video_sender", {})
+
+        video_app = self._video_receiver.session_stats()
+        control_app = self._control_sender.session_stats()
+        video_kcp = self._video_receiver.session_kcp_stats()
+        control_kcp = self._control_sender.session_kcp_stats()
+        video_receiver_status = self._video_receiver.get_status()
+        arbiter_status = self._control_arbiter.get_status()
+        ingress_status = self._native_ingress.get_status()
+        sender_status = self._control_sender.get_status()
+        ack_receiver_status = self._control_ack_receiver.get_status()
+        ack_status = self._control_ack_tracker.get_latest_estimate()
+        telemetry_state = self._hub_receiver.get_snapshot()
+        display_probe_status = self._video_display_probe_store.get_status()
+
+        total_send_bytes = int(video_app.get("send_bytes", 0)) + int(control_app.get("send_bytes", 0))
+        total_recv_bytes = int(video_app.get("recv_bytes", 0)) + int(control_app.get("recv_bytes", 0))
+        tx_kbps, rx_kbps = self._compute_rates(total_send_bytes, total_recv_bytes)
+
+        local_updated_at = utc_iso_now()
+        local_sessions = {
+            "video": self._build_session_payload(
+                "a_to_d.video",
+                str(video_receiver_cfg.get("peer_id", "peer-a-video")),
+                video_app,
+                video_kcp,
+                local_updated_at,
+                False,
+            ),
+            "control": self._build_session_payload(
+                "a_to_d.control",
+                str(control_sender_cfg.get("peer_id", "peer-a-ctrl")),
+                control_app,
+                control_kcp,
+                local_updated_at,
+                False,
+            ),
+        }
+
+        remote_snapshot = telemetry_state.get("snapshot") or {}
+        remote_sessions_by_peer = {
+            str(session.get("peer_id", "")).strip(): session
+            for session in remote_snapshot.get("sessions", []) or []
+            if str(session.get("peer_id", "")).strip()
+        }
+        remote_updated_at = telemetry_state.get("updated_at")
+        remote_stale = bool(telemetry_state.get("stale", True))
+        remote_sessions = {
+            "video": self._build_session_payload(
+                f"hub::{str(video_sender_cfg.get('peer_id', 'peer-b-video'))}",
+                str(video_sender_cfg.get("peer_id", "peer-b-video")),
+                None,
+                remote_sessions_by_peer.get(str(video_sender_cfg.get("peer_id", "peer-b-video")), {}),
+                remote_updated_at,
+                remote_stale,
+            ),
+            "control": self._build_session_payload(
+                f"hub::{str(control_sender_cfg.get('target_peer', 'peer-b-ctrl'))}",
+                str(control_sender_cfg.get("target_peer", "peer-b-ctrl")),
+                None,
+                remote_sessions_by_peer.get(str(control_sender_cfg.get("target_peer", "peer-b-ctrl")), {}),
+                remote_updated_at,
+                remote_stale,
+            ),
+        }
+        self._video_receiver.update_remote_video_srtt(
+            _coerce_int(remote_sessions["video"]["kcp"].get("srtt_ms")) if remote_sessions["video"]["kcp"].get("srtt_ms") is not None else None
+        )
+
+        links = {
+            "a_to_d": self._build_link("local-a-side", local_updated_at, False, local_sessions),
+            "d_to_b": self._build_link("hub-telemetry", remote_updated_at, remote_stale, remote_sessions),
+        }
+
+        primary_session = self._pick_primary_session(links)
+        primary_kcp = dict(primary_session.get("kcp", {})) if primary_session is not None else {}
+        self._ingest_remote_snapshot(telemetry_state)
+
+        fresh_connected_sessions = (
+            links["a_to_d"]["aggregate"]["online_sessions"] + links["d_to_b"]["aggregate"]["online_sessions"]
+        )
+        latency_ms = primary_kcp.get("srtt_ms") if primary_session is not None else None
+        jitter_ms = primary_kcp.get("srttvar_ms") if primary_session is not None else None
+        local_control_registered = bool(control_app.get("registered", 0))
+        remote_control_fresh = bool(remote_sessions["control"].get("connected")) and not bool(remote_sessions["control"].get("stale"))
+        remote_video_fresh = bool(remote_sessions["video"].get("connected")) and not bool(remote_sessions["video"].get("stale"))
+        watchdog_status = _load_optional_json(WATCHDOG_STATUS_PATH)
+        robot_health = self._derive_robot_health(
+            video_receiver_status=video_receiver_status,
+            local_control_registered=local_control_registered,
+            remote_control_fresh=remote_control_fresh,
+            remote_video_fresh=remote_video_fresh,
+            telemetry_state=telemetry_state,
+            watchdog_status=watchdog_status,
+        )
+        latency_estimate = self._derive_latency_estimate(
+            links=links,
+            video_receiver_status=video_receiver_status,
+            display_probe_status=display_probe_status,
+        )
+
+        if local_control_registered and remote_control_fresh:
+            peer_status = "online"
+        elif local_control_registered or bool(local_sessions["video"].get("connected")):
+            peer_status = "degraded"
+        elif sender_status.get("backend_ready"):
+            peer_status = "idle"
+        else:
+            peer_status = "backend-unavailable"
+
+        return {
+            "peer_status": peer_status,
+            "latency_ms": latency_ms,
+            "jitter_ms": jitter_ms,
+            "packet_loss_pct": None,
+            "tx_kbps": round(tx_kbps, 3),
+            "rx_kbps": round(rx_kbps, 3),
+            "transport": "OmniSocket / kcp",
+            "source_mode": "omnisocket-live" if fresh_connected_sessions > 0 else "omnisocket-idle",
+            "updated_at": utc_iso_now(),
+            "active_control_source": arbiter_status["active_source"],
+            "control_lease_remaining_ms": arbiter_status["control_lease_remaining_ms"],
+            "combined": {
+                "connected_sessions": fresh_connected_sessions,
+                "send_bytes": total_send_bytes,
+                "recv_bytes": total_recv_bytes,
+                "tx_kbps": round(tx_kbps, 3),
+                "rx_kbps": round(rx_kbps, 3),
+            },
+            "sessions": {
+                "video": {
+                    "app": video_app,
+                    "kcp": local_sessions["video"]["kcp"],
+                },
+                "control": {
+                    "app": control_app,
+                    "kcp": local_sessions["control"]["kcp"],
+                },
+            },
+            "links": links,
+            "latency_estimate": latency_estimate,
+            "video_freshness": video_receiver_status.get("freshness", {}),
+            "control_ack_status": {
+                **ack_status,
+                "receiver": ack_receiver_status,
+            },
+            "telemetry_receiver": {
+                "hub_connected": bool(telemetry_state.get("connected")),
+                "hub_updated_at": telemetry_state.get("updated_at"),
+                "hub_stale": remote_stale,
+                "last_error": telemetry_state.get("last_error", ""),
+                "peer_id": telemetry_state.get("peer_id", ""),
+                "registered": bool(telemetry_state.get("registered", False)),
+                "last_server_error": str(telemetry_state.get("last_server_error", "") or ""),
+                "reconnect_count": int(telemetry_state.get("reconnect_count", 0)),
+            },
+            "robot_health": robot_health,
+            "ingress": {
+                "native_udp": ingress_status,
+            },
+            "control": {
+                "arbiter": arbiter_status,
+                "sender": sender_status,
+                "ack_receiver": ack_receiver_status,
+            },
+        }
+
+    def close(self) -> None:
+        self._closing.set()
+        thread = self._sample_thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)

host/robot-command-center/backend/monitoring/urls.py

@@ -0,0 +1,16 @@
+from django.urls import path
+
+from . import views
+
+
+urlpatterns = [
+    path("dashboard/", views.dashboard_snapshot, name="dashboard-snapshot"),
+    path("gps/latest/", views.gps_latest, name="gps-latest"),
+    path("network/latest/", views.network_latest, name="network-latest"),
+    path("clock/calibrate/", views.clock_calibration, name="clock-calibration"),
+    path("video/status/", views.video_status, name="video-status"),
+    path("video/camera/", views.video_camera, name="video-camera"),
+    path("video/frame/", views.video_frame, name="video-frame"),
+    path("video/display-probe/", views.video_display_probe, name="video-display-probe"),
+    path("video/stream/", views.video_stream, name="video-stream"),
+]

host/robot-command-center/backend/monitoring/video.py

@@ -0,0 +1,751 @@
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+import hashlib
+import math
+import struct
+import sys
+import threading
+import time
+from typing import Any, Iterator
+
+from .common import (
+    JPEG_FRAME_DIR,
+    OMNISOCKET_CONFIG_PATH,
+    OMNISOCKET_FRAME_FRESH_SECONDS,
+    JsonlRunLogger,
+    VIDEO_SOURCE_MODE,
+    VIDEO_TIMESTAMP_SAMPLE_SIZE,
+    VIDEO_TRAILER_BYTES,
+    VIDEO_TRAILER_ENDIANNESS,
+    VIDEO_TRAILER_STRUCT,
+    VIDEO_TRAILER_TIMESTAMP_MAX_SKEW_NS,
+    VIDEO_TRAILER_TIMESTAMP_MULTIPLIER_NS,
+    VIDEO_TRAILER_TIMESTAMP_UNIT,
+    WORKSPACE_ROOT,
+    load_omnisocket_config,
+)
+
+
+def safe_kcp_stats(session: Any) -> dict[str, Any]:
+    if session is None or not hasattr(session, "kcp_stats"):
+        return {}
+    try:
+        return dict(session.kcp_stats())
+    except Exception:
+        return {}
+
+
+class VideoDisplayProbeStore:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._logger = JsonlRunLogger("BLITZ_A_VIDEO_DISPLAY_PROBE_LOG_PATH", "a-video-display-probe")
+        self._latest: VideoDisplayProbeStatus = VideoDisplayProbeStatus(
+            updated_at=None,
+            frame_seq=None,
+            frame_hash="",
+            input_to_next_fresh_frame_ms=None,
+            input_to_next_changed_frame_ms=None,
+            input_to_next_paint_ms=None,
+            request_to_paint_ms=None,
+            response_to_paint_ms=None,
+            backend_to_request_ms=None,
+            backend_to_request_ms_raw=None,
+            backend_to_paint_ms=None,
+            backend_to_paint_ms_raw=None,
+            browser_backend_clock_offset_ms=None,
+            browser_backend_clock_rtt_ms=None,
+            browser_backend_clock_sample_count=0,
+            browser_backend_clock_calibrated_at=None,
+        )
+
+    def record_event(self, payload: dict[str, Any]) -> None:
+        backend_received_unix_ns = payload.get("backend_received_unix_ns")
+        request_started_unix_ms = payload.get("request_started_unix_ms")
+        response_received_unix_ms = payload.get("response_received_unix_ms")
+        paint_unix_ms = payload.get("paint_unix_ms")
+        browser_backend_clock_offset_ms = self._coerce_float(payload.get("browser_backend_clock_offset_ms"))
+        browser_backend_clock_rtt_ms = self._coerce_float(payload.get("browser_backend_clock_rtt_ms"))
+        browser_backend_clock_sample_count = self._coerce_int(payload.get("browser_backend_clock_sample_count"))
+        browser_backend_clock_calibrated_at = self._coerce_text(payload.get("browser_backend_clock_calibrated_at"))
+        request_to_paint_ms = self._duration_ms(paint_unix_ms, request_started_unix_ms, clamp_floor_zero=True)
+        response_to_paint_ms = self._duration_ms(paint_unix_ms, response_received_unix_ms, clamp_floor_zero=True)
+        backend_received_unix_ms = None
+        try:
+            if backend_received_unix_ns is not None:
+                backend_received_unix_ms = int(backend_received_unix_ns) / 1_000_000.0
+        except (TypeError, ValueError):
+            backend_received_unix_ms = None
+        backend_received_browser_unix_ms = None
+        if backend_received_unix_ms is not None and browser_backend_clock_offset_ms is not None:
+            backend_received_browser_unix_ms = round(backend_received_unix_ms + browser_backend_clock_offset_ms, 3)
+        backend_to_request_ms_raw = self._duration_ms(request_started_unix_ms, backend_received_unix_ms, clamp_floor_zero=False)
+        backend_to_paint_ms_raw = self._duration_ms(paint_unix_ms, backend_received_unix_ms, clamp_floor_zero=False)
+        backend_to_request_ms = self._duration_ms(
+            request_started_unix_ms,
+            backend_received_browser_unix_ms,
+            clamp_floor_zero=True,
+        )
+        backend_to_paint_ms = self._duration_ms(
+            paint_unix_ms,
+            backend_received_browser_unix_ms,
+            clamp_floor_zero=True,
+        )
+
+        status = VideoDisplayProbeStatus(
+            updated_at=self._coerce_text(payload.get("updated_at")),
+            frame_seq=int(payload["frame_seq"]) if payload.get("frame_seq") is not None else None,
+            frame_hash=str(payload.get("frame_hash") or ""),
+            input_to_next_fresh_frame_ms=self._coerce_float(payload.get("input_to_next_fresh_frame_ms")),
+            input_to_next_changed_frame_ms=self._coerce_float(payload.get("input_to_next_changed_frame_ms")),
+            input_to_next_paint_ms=self._coerce_float(payload.get("input_to_next_paint_ms")),
+            request_to_paint_ms=request_to_paint_ms,
+            response_to_paint_ms=response_to_paint_ms,
+            backend_to_request_ms=backend_to_request_ms,
+            backend_to_request_ms_raw=backend_to_request_ms_raw,
+            backend_to_paint_ms=backend_to_paint_ms,
+            backend_to_paint_ms_raw=backend_to_paint_ms_raw,
+            browser_backend_clock_offset_ms=browser_backend_clock_offset_ms,
+            browser_backend_clock_rtt_ms=browser_backend_clock_rtt_ms,
+            browser_backend_clock_sample_count=browser_backend_clock_sample_count,
+            browser_backend_clock_calibrated_at=browser_backend_clock_calibrated_at,
+        )
+        logged_payload = dict(payload)
+        logged_payload.update(
+            {
+                "request_to_paint_ms": request_to_paint_ms,
+                "response_to_paint_ms": response_to_paint_ms,
+                "backend_received_browser_unix_ms": backend_received_browser_unix_ms,
+                "backend_to_request_ms": backend_to_request_ms,
+                "backend_to_request_ms_raw": backend_to_request_ms_raw,
+                "backend_to_paint_ms": backend_to_paint_ms,
+                "backend_to_paint_ms_raw": backend_to_paint_ms_raw,
+            }
+        )
+        with self._lock:
+            self._latest = status
+        self._logger.write(logged_payload)
+
+    def get_status(self) -> dict[str, Any]:
+        with self._lock:
+            latest = self._latest
+        return {
+            "updated_at": latest.updated_at,
+            "frame_seq": latest.frame_seq,
+            "frame_hash": latest.frame_hash,
+            "input_to_next_fresh_frame_ms": latest.input_to_next_fresh_frame_ms,
+            "input_to_next_changed_frame_ms": latest.input_to_next_changed_frame_ms,
+            "input_to_next_paint_ms": latest.input_to_next_paint_ms,
+            "request_to_paint_ms": latest.request_to_paint_ms,
+            "response_to_paint_ms": latest.response_to_paint_ms,
+            "backend_to_request_ms": latest.backend_to_request_ms,
+            "backend_to_request_ms_raw": latest.backend_to_request_ms_raw,
+            "backend_to_paint_ms": latest.backend_to_paint_ms,
+            "backend_to_paint_ms_raw": latest.backend_to_paint_ms_raw,
+            "browser_backend_clock_offset_ms": latest.browser_backend_clock_offset_ms,
+            "browser_backend_clock_rtt_ms": latest.browser_backend_clock_rtt_ms,
+            "browser_backend_clock_sample_count": latest.browser_backend_clock_sample_count,
+            "browser_backend_clock_calibrated_at": latest.browser_backend_clock_calibrated_at,
+        }
+
+    def close(self) -> None:
+        self._logger.close()
+
+    @staticmethod
+    def _coerce_float(value: Any) -> float | None:
+        try:
+            if value is None:
+                return None
+            return round(float(value), 3)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _coerce_int(value: Any) -> int:
+        try:
+            if value is None:
+                return 0
+            return int(value)
+        except (TypeError, ValueError):
+            return 0
+
+    @staticmethod
+    def _coerce_text(value: Any) -> str | None:
+        text = str(value or "").strip()
+        return text or None
+
+    @classmethod
+    def _duration_ms(cls, end_ms: Any, start_ms: Any, *, clamp_floor_zero: bool) -> float | None:
+        end_value = cls._coerce_float(end_ms)
+        start_value = cls._coerce_float(start_ms)
+        if end_value is None or start_value is None:
+            return None
+        delta = round(end_value - start_value, 3)
+        if clamp_floor_zero:
+            delta = max(0.0, delta)
+        return delta
+
+
+@dataclass(frozen=True)
+class FrameTrailerMetadata:
+    timestamp_ns: int
+    latitude: float | None
+    longitude: float | None
+    capture_to_send_ms: int
+    raw_latitude_hex: str
+    raw_longitude_hex: str
+    received_at: float
+
+    @property
+    def has_gps_fix(self) -> bool:
+        return self.latitude is not None and self.longitude is not None
+
+
+@dataclass(frozen=True)
+class VideoDisplayProbeStatus:
+    updated_at: str | None
+    frame_seq: int | None
+    frame_hash: str
+    input_to_next_fresh_frame_ms: float | None
+    input_to_next_changed_frame_ms: float | None
+    input_to_next_paint_ms: float | None
+    request_to_paint_ms: float | None
+    response_to_paint_ms: float | None
+    backend_to_request_ms: float | None
+    backend_to_request_ms_raw: float | None
+    backend_to_paint_ms: float | None
+    backend_to_paint_ms_raw: float | None
+    browser_backend_clock_offset_ms: float | None
+    browser_backend_clock_rtt_ms: float | None
+    browser_backend_clock_sample_count: int
+    browser_backend_clock_calibrated_at: str | None
+
+
+class OmniSocketVideoReceiver:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._started = False
+        self._session = None
+        self._session_cls = None
+        self._binary_msg_type = None
+        self._video_defaults: dict[str, Any] = {}
+        self._latest_frame: bytes | None = None
+        self._latest_received_at = 0.0
+        self._latest_backend_received_unix_ns = 0
+        self._latest_backend_received_mono_ns = 0
+        self._latest_sequence: int | None = None
+        self._latest_metadata: FrameTrailerMetadata | None = None
+        self._latest_sender_clock_delta_ms_raw: float | None = None
+        self._latest_timestamp_unit: str | None = None
+        self._latest_timestamp_endianness: str | None = None
+        self._sender_clock_delta_samples_ms_raw: deque[float] = deque(maxlen=VIDEO_TIMESTAMP_SAMPLE_SIZE)
+        self._latest_frame_hash = ""
+        self._latest_frame_bytes = 0
+        self._last_frame_hash = ""
+        self._last_sequence: int | None = None
+        self._last_backend_received_mono_ns = 0
+        self._interarrival_ms_samples: deque[float] = deque(maxlen=120)
+        self._repeat_samples: deque[int] = deque(maxlen=120)
+        self._skip_samples: deque[int] = deque(maxlen=120)
+        self._freeze_samples_ms: deque[float] = deque(maxlen=120)
+        self._current_stale_frame_run_length = 0
+        self._latest_remote_video_srtt_ms: int | None = None
+        self._frames_received = 0
+        self._last_error = ""
+        self._reconnect_count = 0
+        self._ever_connected = False
+        self._closing = threading.Event()
+        self._frame_recv_logger = JsonlRunLogger("BLITZ_A_VIDEO_FRAME_RECV_LOG_PATH", "a-video-frame-recv")
+        self._load_backend()
+
+    def _load_backend(self) -> None:
+        try:
+            self._import_backend()
+        except Exception as error:  # pragma: no cover - optional runtime dependency
+            self._last_error = f"omnisocket import failed: {error}"
+
+    def _import_backend(self) -> None:
+        try:
+            from omnisocket import MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS  # type: ignore
+        except ImportError:
+            python_dir = WORKSPACE_ROOT / "OmniSocketGo" / "python"
+            if python_dir.exists():
+                sys.path.insert(0, str(python_dir))
+            from omnisocket import MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS  # type: ignore
+
+        self._binary_msg_type = MSG_TYPE_BINARY
+        self._session_cls = Session
+        self._video_defaults = dict(VIDEO_DEFAULTS)
+
+    def ensure_started(self) -> None:
+        if self._session_cls is None or self._binary_msg_type is None:
+            return
+
+        with self._lock:
+            if self._started or self._closing.is_set():
+                return
+            self._started = True
+            self._thread = threading.Thread(
+                target=self._run,
+                name="omnisocket-video-receiver",
+                daemon=True,
+            )
+            self._thread.start()
+
+    def _connect_session(self):
+        assert self._session_cls is not None
+
+        config = load_omnisocket_config()
+        transport_cfg = config.get("transport", {})
+        video_cfg = config.get("video_receiver", {})
+
+        session = self._session_cls()
+        session.connect(
+            server_addr=str(transport_cfg.get("server_addr", "127.0.0.1:10909")),
+            peer_id=str(video_cfg.get("peer_id", "peer-a-video")),
+            relay_via=str(transport_cfg.get("relay_via", "")),
+            bind_ip=str(transport_cfg.get("bind_ip", "")),
+            bind_device=str(transport_cfg.get("bind_device", "")),
+            **self._video_defaults,
+        )
+        return session, int(video_cfg.get("buffer_bytes", 1024 * 1024))
+
+    def _extract_jpeg_payload(self, frame: bytes) -> bytes | None:
+        if frame.startswith(b"\xff\xd8"):
+            return frame
+        if len(frame) > 8 and frame[8:10] == b"\xff\xd8":
+            return frame[8:]
+        return None
+
+    def _split_jpeg_frame_and_trailer(self, frame: bytes) -> tuple[bytes, bytes] | None:
+        jpeg_payload = self._extract_jpeg_payload(frame)
+        if jpeg_payload is None:
+            return None
+
+        if jpeg_payload.endswith(b"\xff\xd9"):
+            return jpeg_payload, b""
+
+        if (
+            len(jpeg_payload) >= VIDEO_TRAILER_BYTES + 2
+            and jpeg_payload[-(VIDEO_TRAILER_BYTES + 2) : -VIDEO_TRAILER_BYTES] == b"\xff\xd9"
+        ):
+            return jpeg_payload[:-VIDEO_TRAILER_BYTES], jpeg_payload[-VIDEO_TRAILER_BYTES:]
+
+        eoi_index = jpeg_payload.rfind(b"\xff\xd9")
+        if eoi_index < 0:
+            return jpeg_payload, b""
+
+        trailer_start = eoi_index + 2
+        return jpeg_payload[:trailer_start], jpeg_payload[trailer_start:]
+
+    def _extract_jpeg_frame(self, frame: bytes) -> bytes | None:
+        split_payload = self._split_jpeg_frame_and_trailer(frame)
+        if split_payload is None:
+            return None
+        jpeg_frame, _ = split_payload
+        return jpeg_frame
+
+    def _extract_sequence(self, frame: bytes) -> int | None:
+        if len(frame) >= 8 and not frame.startswith(b"\xff\xd8"):
+            return int.from_bytes(frame[:8], "big")
+        return None
+
+    def _extract_frame_tail(self, frame: bytes) -> bytes:
+        split_payload = self._split_jpeg_frame_and_trailer(frame)
+        if split_payload is None:
+            return b""
+        _, trailer = split_payload
+        return trailer
+
+    def _extract_frame_metadata(self, frame: bytes, received_at: float | None = None) -> FrameTrailerMetadata | None:
+        trailer = self._extract_frame_tail(frame)
+        if len(trailer) != VIDEO_TRAILER_BYTES:
+            return None
+
+        try:
+            timestamp_ms, latitude, longitude, capture_to_send_ms = VIDEO_TRAILER_STRUCT.unpack(trailer)
+        except struct.error:
+            return None
+
+        if timestamp_ms <= 0:
+            return None
+
+        timestamp_ns = timestamp_ms * VIDEO_TRAILER_TIMESTAMP_MULTIPLIER_NS
+        if abs(time.time_ns() - timestamp_ns) > VIDEO_TRAILER_TIMESTAMP_MAX_SKEW_NS:
+            return None
+
+        gps_fix_available = (
+            math.isfinite(latitude)
+            and math.isfinite(longitude)
+            and (-90.0 <= latitude <= 90.0)
+            and (-180.0 <= longitude <= 180.0)
+            and not (abs(latitude) < 1e-9 and abs(longitude) < 1e-9)
+        )
+
+        return FrameTrailerMetadata(
+            timestamp_ns=timestamp_ns,
+            latitude=latitude if gps_fix_available else None,
+            longitude=longitude if gps_fix_available else None,
+            capture_to_send_ms=int(capture_to_send_ms),
+            raw_latitude_hex=trailer[8:16].hex(),
+            raw_longitude_hex=trailer[16:24].hex(),
+            received_at=received_at if received_at is not None else time.time(),
+        )
+
+    def _has_fresh_frame_locked(self) -> bool:
+        return self._latest_frame is not None and (
+            time.time() - self._latest_received_at <= OMNISOCKET_FRAME_FRESH_SECONDS
+        )
+
+    def update_remote_video_srtt(self, srtt_ms: int | None) -> None:
+        with self._lock:
+            self._latest_remote_video_srtt_ms = srtt_ms
+
+    def _freshness_payload_locked(self) -> dict[str, Any]:
+        interarrival_samples = list(self._interarrival_ms_samples)
+        repeat_samples = list(self._repeat_samples)
+        skip_samples = list(self._skip_samples)
+        freeze_samples_ms = list(self._freeze_samples_ms)
+
+        inter_frame_avg_ms = round(sum(interarrival_samples) / len(interarrival_samples), 3) if interarrival_samples else None
+        if interarrival_samples:
+            ordered = sorted(interarrival_samples)
+            p95_index = min(len(ordered) - 1, max(0, math.ceil(len(ordered) * 0.95) - 1))
+            inter_frame_p95_ms = round(ordered[p95_index], 3)
+        else:
+            inter_frame_p95_ms = None
+
+        repeated_frame_ratio = round(sum(repeat_samples) / len(repeat_samples), 4) if repeat_samples else 0.0
+        total_skip = sum(skip_samples)
+        expected_frames = len(skip_samples) + total_skip
+        skip_ratio = round(total_skip / expected_frames, 4) if expected_frames > 0 else 0.0
+        longest_freeze_ms = round(max(freeze_samples_ms), 3) if freeze_samples_ms else 0.0
+        return {
+            "inter_frame_avg_ms": inter_frame_avg_ms,
+            "inter_frame_p95_ms": inter_frame_p95_ms,
+            "repeated_frame_ratio": repeated_frame_ratio,
+            "skip_ratio": skip_ratio,
+            "longest_freeze_ms": longest_freeze_ms,
+            "stale_frame_run_length": self._current_stale_frame_run_length,
+            "relative_freshness_lag_frames": self._current_stale_frame_run_length + (skip_samples[-1] if skip_samples else 0),
+        }
+
+    def _frame_headers_locked(self) -> dict[str, str]:
+        capture_to_send_ms = self._latest_metadata.capture_to_send_ms if self._latest_metadata is not None else None
+        headers: dict[str, str] = {}
+        if self._latest_sequence is not None:
+            headers["X-Blitz-Frame-Seq"] = str(self._latest_sequence)
+        if self._latest_backend_received_unix_ns > 0:
+            headers["X-Blitz-Backend-Received-Unix-Ns"] = str(self._latest_backend_received_unix_ns)
+        if self._latest_frame_hash:
+            headers["X-Blitz-Frame-Hash"] = self._latest_frame_hash
+        if capture_to_send_ms is not None:
+            headers["X-Blitz-BSide-Capture-To-Send-Ms"] = str(capture_to_send_ms)
+        return headers
+
+    def get_latest_frame_headers(self) -> dict[str, str]:
+        snapshot = self.get_latest_frame_snapshot()
+        return snapshot[1] if snapshot is not None else {}
+
+    def _run(self) -> None:
+        while not self._closing.is_set():
+            try:
+                session, buffer_bytes = self._connect_session()
+                with self._lock:
+                    self._session = session
+                    self._last_error = ""
+                    if self._ever_connected:
+                        self._reconnect_count += 1
+                    else:
+                        self._ever_connected = True
+                buffer = bytearray(buffer_bytes)
+
+                while not self._closing.is_set():
+                    meta = session.recv_into(buffer, timeout_ms=1000)
+                    if meta is None:
+                        continue
+                    if meta.get("msg_type") != self._binary_msg_type:
+                        continue
+
+                    frame = bytes(buffer[: meta["body_len"]])
+                    jpeg_frame = self._extract_jpeg_frame(frame)
+                    if jpeg_frame is None:
+                        self._last_error = "received non-JPEG binary frame"
+                        continue
+
+                    received_at = time.time()
+                    received_unix_ns = time.time_ns()
+                    received_mono_ns = time.monotonic_ns()
+                    frame_metadata = self._extract_frame_metadata(frame, received_at=received_at)
+                    sender_clock_delta_ms_raw = None
+                    if frame_metadata is not None:
+                        sender_clock_delta_ms_raw = round((received_unix_ns - frame_metadata.timestamp_ns) / 1_000_000, 3)
+                        unit = VIDEO_TRAILER_TIMESTAMP_UNIT
+                        endianness = VIDEO_TRAILER_ENDIANNESS
+                    else:
+                        unit = None
+                        endianness = None
+                    frame_sequence = self._extract_sequence(frame)
+                    frame_hash = hashlib.blake2s(jpeg_frame, digest_size=8).hexdigest()
+                    local_kcp = safe_kcp_stats(session) if self._frame_recv_logger.enabled else {}
+                    frame_log_record: dict[str, Any] | None = None
+
+                    with self._lock:
+                        interarrival_ms = None
+                        if self._last_backend_received_mono_ns > 0:
+                            interarrival_ms = round((received_mono_ns - self._last_backend_received_mono_ns) / 1_000_000, 3)
+                            self._interarrival_ms_samples.append(interarrival_ms)
+
+                        sequence_gap = 0
+                        if frame_sequence is not None and self._last_sequence is not None and frame_sequence > self._last_sequence:
+                            sequence_gap = max(0, frame_sequence - self._last_sequence - 1)
+
+                        repeat_flag = bool(self._last_frame_hash) and frame_hash == self._last_frame_hash
+                        self._repeat_samples.append(1 if repeat_flag else 0)
+                        self._skip_samples.append(sequence_gap)
+                        if repeat_flag:
+                            self._current_stale_frame_run_length += 1
+                        else:
+                            self._current_stale_frame_run_length = 0
+                        if interarrival_ms is not None and repeat_flag:
+                            self._freeze_samples_ms.append(interarrival_ms)
+                        elif interarrival_ms is not None:
+                            self._freeze_samples_ms.append(0.0)
+
+                        self._latest_frame = jpeg_frame
+                        self._latest_received_at = received_at
+                        self._latest_backend_received_unix_ns = received_unix_ns
+                        self._latest_backend_received_mono_ns = received_mono_ns
+                        self._latest_sequence = frame_sequence
+                        self._last_sequence = frame_sequence
+                        self._latest_metadata = frame_metadata
+                        self._latest_sender_clock_delta_ms_raw = sender_clock_delta_ms_raw
+                        self._latest_timestamp_unit = unit
+                        self._latest_timestamp_endianness = endianness
+                        self._latest_frame_hash = frame_hash
+                        self._latest_frame_bytes = len(jpeg_frame)
+                        self._last_frame_hash = frame_hash
+                        self._last_backend_received_mono_ns = received_mono_ns
+                        if sender_clock_delta_ms_raw is not None:
+                            self._sender_clock_delta_samples_ms_raw.append(sender_clock_delta_ms_raw)
+                        self._frames_received += 1
+
+                        if self._frame_recv_logger.enabled:
+                            frame_log_record = {
+                                "ts_unix_nano": received_unix_ns,
+                                "frame_seq": frame_sequence,
+                                "backend_received_unix_ns": received_unix_ns,
+                                "backend_received_mono_ns": received_mono_ns,
+                                "jpeg_bytes": len(jpeg_frame),
+                                "interarrival_ms": interarrival_ms,
+                                "sequence_gap": sequence_gap,
+                                "repeat_flag": repeat_flag,
+                                "skip_count": sequence_gap,
+                                "frame_hash": frame_hash,
+                                "a_to_d_video_srtt_ms": local_kcp.get("srtt_ms"),
+                                "d_to_b_video_srtt_ms": self._latest_remote_video_srtt_ms,
+                                "b_side_capture_to_send_ms": frame_metadata.capture_to_send_ms if frame_metadata is not None else None,
+                                "sender_clock_delta_ms_raw": sender_clock_delta_ms_raw,
+                            }
+                    if frame_log_record is not None:
+                        self._frame_recv_logger.write(frame_log_record)
+            except Exception as error:  # pragma: no cover - runtime integration path
+                if not self._closing.is_set():
+                    session_error = ""
+                    if self._session is not None:
+                        try:
+                            session_error = str(dict(self._session.stats()).get("last_server_error", "") or "")
+                        except Exception:
+                            session_error = ""
+                    self._last_error = session_error or str(error)
+                    time.sleep(2)
+            finally:
+                if self._session is not None:
+                    try:
+                        self._session.close()
+                    except Exception:
+                        pass
+                with self._lock:
+                    self._session = None
+                    if self._closing.is_set():
+                        self._started = False
+
+    def get_latest_frame(self) -> bytes | None:
+        snapshot = self.get_latest_frame_snapshot()
+        return snapshot[0] if snapshot is not None else None
+
+    def get_latest_frame_snapshot(self) -> tuple[bytes, dict[str, str]] | None:
+        self.ensure_started()
+        with self._lock:
+            if not self._has_fresh_frame_locked():
+                return None
+            if self._latest_frame is None:
+                return None
+            return self._latest_frame, self._frame_headers_locked()
+
+    def get_latest_frame_metadata(self) -> FrameTrailerMetadata | None:
+        self.ensure_started()
+        with self._lock:
+            if not self._has_fresh_frame_locked():
+                return None
+            return self._latest_metadata
+
+    def session_stats(self) -> dict[str, Any]:
+        self.ensure_started()
+        with self._lock:
+            session = self._session
+        if session is None:
+            return {"connected": 0, "registered": 0, "last_server_error": self._last_error}
+        try:
+            return dict(session.stats())
+        except Exception:
+            return {"connected": 0, "registered": 0, "last_server_error": self._last_error}
+
+    def session_kcp_stats(self) -> dict[str, Any]:
+        self.ensure_started()
+        with self._lock:
+            session = self._session
+        return safe_kcp_stats(session)
+
+    def get_status(self) -> dict[str, Any]:
+        self.ensure_started()
+        config = load_omnisocket_config()
+        transport_cfg = config.get("transport", {})
+        video_cfg = config.get("video_receiver", {})
+        session_stats = self.session_stats()
+        with self._lock:
+            has_recent_frame = self._has_fresh_frame_locked()
+            freshness_status = self._freshness_payload_locked()
+            if has_recent_frame and self._latest_sender_clock_delta_ms_raw is not None:
+                timing_status = {
+                    "available": True,
+                    "sender_clock_delta_ms_raw": self._latest_sender_clock_delta_ms_raw,
+                    "sender_clock_delta_samples_ms_raw": list(reversed(self._sender_clock_delta_samples_ms_raw)),
+                    "sample_count": len(self._sender_clock_delta_samples_ms_raw),
+                    "sample_window_size": VIDEO_TIMESTAMP_SAMPLE_SIZE,
+                    "timestamp_unit": self._latest_timestamp_unit,
+                    "timestamp_endianness": self._latest_timestamp_endianness,
+                    "unsynced_clock": True,
+                }
+            else:
+                timing_status = {
+                    "available": False,
+                    "sender_clock_delta_ms_raw": None,
+                    "sender_clock_delta_samples_ms_raw": [],
+                    "sample_count": 0,
+                    "sample_window_size": VIDEO_TIMESTAMP_SAMPLE_SIZE,
+                    "timestamp_unit": None,
+                    "timestamp_endianness": None,
+                    "unsynced_clock": True,
+                }
+            return {
+                "backend_ready": self._session_cls is not None,
+                "mode": VIDEO_SOURCE_MODE,
+                "connected": self._session is not None,
+                "registered": bool(session_stats.get("registered", 0)),
+                "has_recent_frame": has_recent_frame,
+                "frames_received": self._frames_received,
+                "latest_sequence": self._latest_sequence,
+                "latest_frame_hash": self._latest_frame_hash,
+                "latest_backend_received_unix_ns": self._latest_backend_received_unix_ns or None,
+                "latest_backend_received_mono_ns": self._latest_backend_received_mono_ns or None,
+                "latest_frame_bytes": self._latest_frame_bytes,
+                "latest_capture_to_send_ms": self._latest_metadata.capture_to_send_ms if self._latest_metadata is not None else None,
+                "reconnect_count": self._reconnect_count,
+                "last_server_error": str(session_stats.get("last_server_error", "") or ""),
+                "last_error": self._last_error,
+                "config_path": str(OMNISOCKET_CONFIG_PATH),
+                "server_addr": str(transport_cfg.get("server_addr", "")),
+                "relay_via": str(transport_cfg.get("relay_via", "")),
+                "peer_id": str(video_cfg.get("peer_id", "")),
+                "buffer_bytes": int(video_cfg.get("buffer_bytes", 0)),
+                "timing": timing_status,
+                "freshness": freshness_status,
+            }
+
+    def close(self) -> None:
+        self._closing.set()
+        with self._lock:
+            session = self._session
+        if session is not None:
+            try:
+                session.close()
+            except Exception:
+                pass
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)
+        self._frame_recv_logger.close()
+
+
+class VideoFrameService:
+    def __init__(self, receiver: OmniSocketVideoReceiver, display_probe_store: VideoDisplayProbeStore) -> None:
+        self._receiver = receiver
+        self._display_probe_store = display_probe_store
+
+    def get_status(self) -> dict[str, Any]:
+        receiver_status = self._receiver.get_status()
+        receiver_frame = self._receiver.get_latest_frame()
+        display_probe_status = self._display_probe_store.get_status()
+
+        if receiver_frame is not None:
+            return {
+                "available": True,
+                "source_mode": "omnisocket-jpeg-live",
+                "frame_count": receiver_status["frames_received"],
+                "fps": 30,
+                "frame_dir": str(JPEG_FRAME_DIR),
+                "source_detail": f"peer stream active, frames={receiver_status['frames_received']}",
+                "receiver": receiver_status,
+                "timing": receiver_status["timing"],
+                "freshness": receiver_status.get("freshness", {}),
+                "display_probe": display_probe_status,
+            }
+
+        wait_detail = receiver_status["last_error"] or (
+            "waiting for live OmniSocket JPEG frames; check the hub, sender, and receiver configuration"
+        )
+        return {
+            "available": False,
+            "source_mode": "omnisocket-waiting",
+            "frame_count": receiver_status["frames_received"],
+            "fps": 30,
+            "frame_dir": str(JPEG_FRAME_DIR),
+            "source_detail": wait_detail,
+            "receiver": receiver_status,
+            "timing": receiver_status["timing"],
+            "freshness": receiver_status.get("freshness", {}),
+            "display_probe": display_probe_status,
+        }
+
+    def get_next_frame(self) -> bytes:
+        receiver_frame = self._receiver.get_latest_frame()
+        if receiver_frame is not None:
+            return receiver_frame
+        raise RuntimeError("no live OmniSocket JPEG frame is currently available")
+
+    def get_next_frame_with_headers(self) -> tuple[bytes, dict[str, str]]:
+        snapshot = self._receiver.get_latest_frame_snapshot()
+        if snapshot is not None:
+            return snapshot
+        raise RuntimeError("no live OmniSocket JPEG frame is currently available")
+
+    def get_latest_frame_headers(self) -> dict[str, str]:
+        return self._receiver.get_latest_frame_headers()
+
+    def record_display_probe(self, payload: dict[str, Any]) -> None:
+        self._display_probe_store.record_event(payload)
+
+    def iter_mjpeg(self, fps: float = 6.0) -> Iterator[bytes]:
+        frame_interval = 1.0 / max(1.0, min(fps, 30.0))
+        while True:
+            frame = self.get_next_frame()
+            header = (
+                b"--frame\r\n"
+                b"Content-Type: image/jpeg\r\n"
+                + f"Content-Length: {len(frame)}\r\n\r\n".encode("ascii")
+            )
+            yield header + frame + b"\r\n"
+            time.sleep(frame_interval)

host/robot-command-center/backend/monitoring/views.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import json
+import time
+
+from django.views.decorators.csrf import csrf_exempt
+from django.http import HttpResponse, StreamingHttpResponse
+from rest_framework.decorators import api_view
+from rest_framework.response import Response
+
+from .services import camera_control_service, gps_service, network_service, video_service
+
+
+@api_view(["GET"])
+def dashboard_snapshot(request):
+    return Response(
+        {
+            "gps": gps_service.get_latest(),
+            "network": network_service.get_latest(),
+            "video": video_service.get_status(),
+        }
+    )
+
+
+@api_view(["GET"])
+def gps_latest(request):
+    return Response(gps_service.get_latest())
+
+
+@api_view(["GET"])
+def network_latest(request):
+    return Response(network_service.get_latest())
+
+
+@api_view(["GET"])
+def clock_calibration(request):
+    server_received_unix_ns = time.time_ns()
+    server_sent_unix_ns = time.time_ns()
+    response = Response(
+        {
+            "server_received_unix_ms": round(server_received_unix_ns / 1_000_000.0, 3),
+            "server_sent_unix_ms": round(server_sent_unix_ns / 1_000_000.0, 3),
+        }
+    )
+    response["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
+    return response
+
+
+@api_view(["GET"])
+def video_status(request):
+    return Response(video_service.get_status())
+
+
+@csrf_exempt
+@api_view(["GET", "POST"])
+def video_camera(request):
+    if request.method == "GET":
+        return Response(camera_control_service.get_camera_status())
+
+    camera = str(request.data.get("camera") or "").strip().lower()
+    if camera not in {"head", "waist"}:
+        return Response({"detail": "camera must be 'head' or 'waist'"}, status=400)
+
+    try:
+        result = camera_control_service.select_camera(camera)
+    except (OSError, RuntimeError) as error:
+        return Response({"detail": str(error)}, status=503)
+    return Response(result, status=200 if result.get("confirmed") else 504)
+
+
+def video_frame(request):
+    try:
+        frame, headers = video_service.get_next_frame_with_headers()
+    except (FileNotFoundError, RuntimeError) as error:
+        status = video_service.get_status()
+        return HttpResponse(
+            status.get("source_detail") or str(error),
+            status=503,
+            content_type="text/plain; charset=utf-8",
+        )
+
+    response = HttpResponse(frame, content_type="image/jpeg")
+    response["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
+    for key, value in headers.items():
+        response[key] = value
+    return response
+
+
+def video_stream(request):
+    status = video_service.get_status()
+    if not status["available"]:
+        return HttpResponse(
+            status.get("source_detail") or f"JPEG frame directory not found: {status['frame_dir']}",
+            status=503,
+            content_type="text/plain; charset=utf-8",
+        )
+
+    try:
+        fps = float(request.GET.get("fps", status["fps"]))
+    except (TypeError, ValueError):
+        fps = float(status["fps"])
+
+    response = StreamingHttpResponse(
+        video_service.iter_mjpeg(fps=fps),
+        content_type="multipart/x-mixed-replace; boundary=frame",
+    )
+    response["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
+    return response
+
+
+@csrf_exempt
+@api_view(["POST"])
+def video_display_probe(request):
+    try:
+        payload = json.loads(request.body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return Response({"detail": "invalid json"}, status=400)
+
+    if not isinstance(payload, dict):
+        return Response({"detail": "expected json object"}, status=400)
+
+    video_service.record_display_probe(payload)
+    return Response({"ok": True})