feat(meta_quest): propagate clock_domain + connection health events

src/syncfield/adapters/meta_quest.py

@@ -119,6 +119,12 @@ class MetaQuestHandStream(StreamBase):
     BUFFER_SIZE = 65536
     CONNECTION_TIMEOUT_S = 2.0
 
+    # Samples arrive via WiFi UDP from a remote Quest device, so they do
+    # not share the host monotonic clock domain. Tag them separately so
+    # downstream sync tooling can account for the wireless jitter.
+    CLOCK_DOMAIN = "remote_quest3"
+    UNCERTAINTY_NS = 10_000_000  # 10 ms — typical WiFi jitter budget
+
     _discovery_kind = "sensor"
     _discovery_adapter_type = "meta_quest"
 
@@ -152,7 +158,11 @@ def __init__(
         self._first_at: Optional[int] = None
         self._last_at: Optional[int] = None
         self._consecutive_errors = 0
-        self._last_packet_mono = 0.0
+        self._last_packet_mono: float = 0.0
+        # "waiting"  = socket is up, no packet has ever been received yet
+        # "connected" = a packet arrived within CONNECTION_TIMEOUT_S
+        # "lost"     = had packets, none for > CONNECTION_TIMEOUT_S (DROP emitted)
+        self._connection_state: str = "waiting"
 
     # ------------------------------------------------------------------
     # 4-phase lifecycle
@@ -168,6 +178,8 @@ def connect(self) -> None:
         self._first_at = None
         self._last_at = None
         self._consecutive_errors = 0
+        self._last_packet_mono = 0.0
+        self._connection_state = "waiting"
 
         self._create_socket()
         self._receive_thread = threading.Thread(
@@ -248,6 +260,22 @@ def _create_socket(self) -> None:
         self._socket.bind((self._host, self._port))
         self._socket.settimeout(1.0)
 
+    @property
+    def is_connected(self) -> bool:
+        """True when a packet has arrived within ``CONNECTION_TIMEOUT_S``.
+
+        Mirrors the recorder's ``is_connected`` semantics: the socket
+        being open is not enough — the Quest must actually be streaming.
+        Callers can poll this from any thread; it reads a float that
+        CPython updates atomically.
+        """
+        if self._last_packet_mono == 0.0:
+            return False
+        return (
+            time.monotonic() - self._last_packet_mono
+            <= self.CONNECTION_TIMEOUT_S
+        )
+
     def _receive_loop(self) -> None:
         while not self._stop_event.is_set():
             if self._socket is None:
@@ -256,11 +284,29 @@ def _receive_loop(self) -> None:
                 data, _ = self._socket.recvfrom(self.BUFFER_SIZE)
                 self._process_packet(data)
             except TimeoutError:
+                # The 1-second recv timeout doubles as a watchdog tick —
+                # use it to notice when the Quest has stopped streaming.
+                self._check_connection_timeout()
                 continue
             except Exception as exc:
                 if not self._stop_event.is_set():
                     self._handle_socket_error(exc)
 
+    def _check_connection_timeout(self) -> None:
+        """Emit DROP if we had packets and now haven't seen one for > timeout."""
+        if self._connection_state != "connected":
+            return
+        if self._last_packet_mono == 0.0:
+            return
+        silence_s = time.monotonic() - self._last_packet_mono
+        if silence_s > self.CONNECTION_TIMEOUT_S:
+            self._connection_state = "lost"
+            self._emit_health(HealthEvent(
+                self.id, HealthEventKind.DROP,
+                time.monotonic_ns(),
+                f"No packet for {silence_s:.1f}s (timeout {self.CONNECTION_TIMEOUT_S}s)",
+            ))
+
     def _handle_socket_error(self, error: Exception) -> None:
         self._consecutive_errors += 1
         if self._consecutive_errors >= self.MAX_CONSECUTIVE_ERRORS:
@@ -292,6 +338,7 @@ def _process_packet(self, data: bytes) -> None:
 
         self._last_packet_mono = time.monotonic()
         self._consecutive_errors = 0
+        self._update_connection_state_on_packet(capture_ns)
 
         # Build channels from packet
         channels = self._parse_channels(packet)
@@ -307,6 +354,25 @@ def _process_packet(self, data: bytes) -> None:
                 frame_number=frame_number,
                 capture_ns=capture_ns,
                 channels=channels,
+                uncertainty_ns=self.UNCERTAINTY_NS,
+                clock_domain=self.CLOCK_DOMAIN,
+            ))
+
+    def _update_connection_state_on_packet(self, at_ns: int) -> None:
+        """Emit HEARTBEAT on first packet, RECONNECT after a drop."""
+        prev = self._connection_state
+        if prev == "connected":
+            return
+        self._connection_state = "connected"
+        if prev == "waiting":
+            self._emit_health(HealthEvent(
+                self.id, HealthEventKind.HEARTBEAT,
+                at_ns, "First Quest 3 packet received",
+            ))
+        elif prev == "lost":
+            self._emit_health(HealthEvent(
+                self.id, HealthEventKind.RECONNECT,
+                at_ns, "Quest 3 packet stream resumed",
             ))
 
     def _parse_channels(self, packet: Dict[str, Any]) -> Dict[str, Any]:

src/syncfield/orchestrator.py

@@ -2094,14 +2094,15 @@ def _handle(event: SampleEvent) -> None:
             if not active[0]:
                 return
             try:
+                clock_domain = event.clock_domain or host_id
                 if isinstance(writer, SensorWriter):
                     writer.write(
                         SensorSample(
                             frame_number=event.frame_number,
                             capture_ns=event.capture_ns,
                             channels=event.channels or {},
                             clock_source="host_monotonic",
-                            clock_domain=host_id,
+                            clock_domain=clock_domain,
                             uncertainty_ns=event.uncertainty_ns,
                         )
                     )
@@ -2111,7 +2112,7 @@ def _handle(event: SampleEvent) -> None:
                             frame_number=event.frame_number,
                             capture_ns=event.capture_ns,
                             clock_source="host_monotonic",
-                            clock_domain=host_id,
+                            clock_domain=clock_domain,
                             uncertainty_ns=event.uncertainty_ns,
                         )
                     )

src/syncfield/types.py

@@ -257,13 +257,21 @@ def to_dict(self) -> dict[str, Any]:
 
 @dataclass(frozen=True)
 class SampleEvent:
-    """A stream reports a sample (timestamp + optional channels) to the orchestrator."""
+    """A stream reports a sample (timestamp + optional channels) to the orchestrator.
+
+    ``clock_domain`` lets an adapter override the default host clock domain
+    when the timestamp's *origin* is a remote device rather than the local
+    monotonic clock (e.g. a Meta Quest streaming poses over WiFi). Leaving
+    it ``None`` — the common case for on-host captures — makes the
+    orchestrator stamp the host's id so all local streams share one domain.
+    """
 
     stream_id: str
     frame_number: int
     capture_ns: int
     channels: dict[str, "ChannelValue"] | None = None
     uncertainty_ns: int = 5_000_000
+    clock_domain: str | None = None
 
 
 @dataclass

tests/unit/adapters/test_meta_quest.py

@@ -300,6 +300,116 @@ def test_samples_emitted_during_recording(self):
         assert len(event.channels["hand_joints"]) == 156
 
 
+# ---------------------------------------------------------------------------
+# Clock metadata (clock_domain + uncertainty_ns) emitted on SampleEvent
+# ---------------------------------------------------------------------------
+
+
+class TestClockMetadata:
+    def test_sample_event_has_remote_quest_clock_domain(self):
+        from syncfield.clock import SessionClock, SyncPoint
+
+        port = _find_free_port()
+        stream = MetaQuestHandStream("test", port=port)
+        received = []
+        stream.on_sample(lambda e: received.append(e))
+
+        stream.connect()
+        stream.start_recording(SessionClock(sync_point=SyncPoint.create_now("h")))
+        _send_packet(port, _make_quest3_packet())
+        time.sleep(0.2)
+        stream.stop_recording()
+        stream.disconnect()
+
+        assert received, "expected at least one sample"
+        assert received[0].clock_domain == "remote_quest3"
+        assert received[0].uncertainty_ns == 10_000_000
+
+
+# ---------------------------------------------------------------------------
+# Connection health — is_connected property + DROP/HEARTBEAT/RECONNECT
+# ---------------------------------------------------------------------------
+
+
+class TestConnectionHealth:
+    def test_is_connected_false_before_any_packet(self):
+        stream = MetaQuestHandStream("test", port=0)
+        assert stream.is_connected is False
+
+    def test_is_connected_true_after_recent_packet(self):
+        port = _find_free_port()
+        stream = MetaQuestHandStream("test", port=port)
+        stream.connect()
+        try:
+            _send_packet(port, _make_quest3_packet())
+            time.sleep(0.2)
+            assert stream.is_connected is True
+        finally:
+            stream.disconnect()
+
+    def test_heartbeat_emitted_on_first_packet(self):
+        port = _find_free_port()
+        stream = MetaQuestHandStream("test", port=port)
+        events = []
+        stream.on_health(lambda e: events.append(e))
+
+        stream.connect()
+        try:
+            _send_packet(port, _make_quest3_packet())
+            time.sleep(0.2)
+        finally:
+            stream.disconnect()
+
+        kinds = [e.kind.value for e in events]
+        assert "heartbeat" in kinds
+
+    def test_drop_emitted_after_silence(self):
+        # Shorten timeout so the watchdog fires quickly in tests.
+        port = _find_free_port()
+        stream = MetaQuestHandStream("test", port=port)
+        stream.CONNECTION_TIMEOUT_S = 0.3  # override at instance level
+        events = []
+        stream.on_health(lambda e: events.append(e))
+
+        stream.connect()
+        try:
+            # Mark the stream connected, then let the receive loop's
+            # 1-second socket timeout fire once with no further packets.
+            _send_packet(port, _make_quest3_packet())
+            time.sleep(0.2)
+            assert stream.is_connected is True
+            # Wait long enough for silence to exceed CONNECTION_TIMEOUT_S
+            # and for at least one recv timeout tick (socket timeout = 1.0s).
+            time.sleep(1.5)
+        finally:
+            stream.disconnect()
+
+        kinds = [e.kind.value for e in events]
+        assert "drop" in kinds, f"expected drop in {kinds}"
+
+    def test_reconnect_emitted_after_drop_resumes(self):
+        port = _find_free_port()
+        stream = MetaQuestHandStream("test", port=port)
+        stream.CONNECTION_TIMEOUT_S = 0.3
+        events = []
+        stream.on_health(lambda e: events.append(e))
+
+        stream.connect()
+        try:
+            _send_packet(port, _make_quest3_packet())
+            time.sleep(0.2)
+            time.sleep(1.5)  # allow drop to fire
+            _send_packet(port, _make_quest3_packet())
+            time.sleep(0.2)
+        finally:
+            stream.disconnect()
+
+        kinds = [e.kind.value for e in events]
+        assert "drop" in kinds
+        assert "reconnect" in kinds
+        assert kinds.index("reconnect") > kinds.index("drop")
+
+
 def _find_free_port() -> int:
     with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
         s.bind(("127.0.0.1", 0))