egocap: align camera, glove and controller time; show latency and fps

.gitignore

@@ -95,3 +95,7 @@ tmp/
 !/examples/mujoco_xr/python/isaacteleop_examples/mujoco_xr/assets/follower/follower_arm.xml
 # Wheel SBOM and license evidence written by the collector in scripts/sbom/
 /sbom/
+examples/egocap/recordings/
+
+# Agent-only plans and status notes; never checked in.
+/agentPlans/

examples/egocap/LIVEKIT_PREVIEW.md

@@ -26,15 +26,17 @@ livekit:
   url: wss://LIVEKIT_HOST
   room: sensing-preview
   publisher: orin-sensing
-  timestamp_basis: capture_unix_us
+  timestamp_basis: monotonic_us
   streams:
     - {name: Head left, track: sensing-left, width: 1280, height: 720, fps: 30}
 ```
 
 - `track` must equal the name the publisher gives the track (the emulator's `livekit=NAME`).
 - `publisher` must equal the identity in the publisher's join token.
-- Preview works with any `timestamp_basis`. Recording needs `capture_unix_us` and `width`, `height` and
-  `fps`; recording is refused while any offered stream lacks them.
+- `timestamp_basis` (default `monotonic_us`) must equal what the publisher stamps; the values are listed in the
+  [streams file reference](capture_server/README.md#the---livekit-streams-file). Preview works with any of them.
+  Recording needs a monotonic basis, `monotonic_us` or `capture_monotonic_us`, and `width`, `height` and `fps`;
+  recording is refused while any offered stream lacks them.
 
 Start CaptureServer with the file (add `--mock` for synthetic tracking):
 
@@ -67,8 +69,10 @@ uv run python tools/livekit_orin_emulator.py \
   --width 1280 --height 720 --fps 30
 ```
 
-`--add-stream sensor=N,livekit=NAME` is repeatable. `--timestamp-basis` takes `capture_unix_us` (default),
-`processing_unix_us` or `monotonic_us`; set the same value in the streams file. Frames go through a software
+`--add-stream sensor=N,livekit=NAME` is repeatable. `--timestamp-basis` takes `monotonic_us` (default),
+`capture_monotonic_us`, `capture_unix_us` or `processing_unix_us`; set the same value in the streams file. The Orin's
+sender (`--timestamp-basis=` on the sensing app) takes `monotonic_us` (default), `capture_monotonic_us` or
+`processing_unix_us`. Both publishers answer the PC's clock-sync requests, which recording relies on. Frames go through a software
 encoder, so this checks the PC side and the metadata, not the Jetson encoder.
 
 ## 4. Preview and record on the PC
@@ -80,9 +84,20 @@ export LIVEKIT_API_KEY=... LIVEKIT_API_SECRET=...
 uv run uvicorn egocap_web.app:app --app-dir backend --port 8000
 ```
 
-- The browser asks `GET /api/livekit-token` for a subscribe-only token, only for a room the current session
-  offers, and shows the stream in the camera grid. Without the key pair the backend answers 503 and the page
+- The browser asks `GET /api/livekit-token` for a viewer token, only for a room the current session
+  offers, and shows the stream in the camera grid. The token can subscribe and send data messages (the clock-sync
+  exchange needs them) but cannot publish media. The route is unauthenticated, so anyone who can reach the
+  backend can send data messages into the room, so the publisher must act only on well-formed
+  `isaaccapture.clock-sync` requests. Both publishers do: a packet is answered only if its topic is
+  `isaaccapture.clock-sync` and it is exactly a 13-byte request starting with `Q`; everything else is ignored, and a
+  reply goes only to the requester. The native sender keeps at most 16 pending requests and drops the oldest when
+  more arrive. Neither has a per-second rate limit, and the emulator has no bound on pending replies, so keep the
+  LiveKit room and the backend on a trusted network. Without the key pair the backend answers 503 and the page
   reports the preview as unavailable.
-- Recording joins the room with its own subscribe-only token signed with the same key pair. It starts only
+- Recording joins the room with its own viewer-capability token signed with the same key pair. It starts only
   when every offered stream is recordable (see step 1) and stops with a reason if a frame cannot be recorded
-  honestly. The backend must run on the machine whose clock CaptureServer reports.
+  honestly. The backend must run on the machine whose clock CaptureServer reports: recording is refused unless
+  CaptureServer's URL is a loopback address.
+- Recording maps the publisher's monotonic timestamps onto the PC's clock by sending the publisher clock-sync
+  requests (data messages on topic `isaaccapture.clock-sync`) and reading its replies. The publisher must stay
+  in the room and answer them; without replies the mapping goes stale.

examples/egocap/backend/README.md

@@ -33,7 +33,7 @@ or restarts it.
 | `GET`/`PUT /api/operator-settings` | The operator's station setup: the camera type (`sensing_gmsl` or `oak_d`), camera roles (`ego`, `wrist_left`, `wrist_right`), the task catalog the recording panel's picker offers (`task_file`) and the task it preselects (`preselected_task`). Kept in `.egocap/operator_settings.json` under the recording output (hidden from the recordings browser) (`EGOCAP_OPERATOR_SETTINGS_FILE` overrides the path) so every browser sees the same values. `PUT` names only the fields it changes and uses `X-Control-Token`; legacy JSON `control_token` remains accepted. A wrong token is 403 and bodies above 16 KiB are rejected with 413 before parsing. |
 | `POST /api/task-files/import`, `DELETE /api/task-files/{name}` | Import a task catalog the operator supplies — by `name` + `content` (text uploaded from the browser) — into the imported-catalog directory beside the operator settings, once it parses; and remove an imported one. `POST` and `DELETE` use the `X-Control-Token` header (legacy JSON tokens remain accepted for `POST`); `DELETE` refuses removal of the selected catalog (409); imports also refuse replacement of that catalog until it is deselected. Imports appear in `GET /api/task-files` (as `imported`) and shadow a shipped catalog of the same name. |
 | `GET /api/tools` | The operator tools the Settings drawer offers (headset calibration, glove calibration, fisheye rectification for the SENSING cameras), each with whether this build can launch it. None can yet. |
-| `GET /api/livekit-token?room=ROOM` | Subscribe-only viewer token for a LiveKit room the session offers (404 otherwise, 503 without signing credentials). Which streams exist comes from CaptureServer's session message. |
+| `GET /api/livekit-token?room=ROOM` | Viewer token (subscribe and data messages, no media publishing) for a LiveKit room the session offers (404 otherwise, 503 without signing credentials). Which streams exist comes from CaptureServer's session message. |
 | `GET /api/task-files`, `GET /api/task-files/{name}` | Read-only task-catalog API; see [`../data/taskfiles/README.md`](../data/taskfiles/README.md). |
 | `GET /recordings/...` | The optional recordings browser's API and file routes, off unless `EGOCAP_RECORDINGS_ENABLED=1`; see [`../recordings_frontend/README.md`](../recordings_frontend/README.md). |
 
@@ -55,7 +55,7 @@ Everything else is an environment variable:
 | --- | --- | --- |
 | `EGOCAP_STATION_CONFIG` | unset (development default) | Path to the deployment station YAML described above. |
 | `EGOCAP_CAPTURE_SERVER_URL` | `ws://127.0.0.1:8082` | Where to connect for tracking/camera data. |
-| `LIVEKIT_API_KEY`, `LIVEKIT_API_SECRET` | unset | The LiveKit server's key pair. The backend signs two subscribe-only tokens with it: viewer tokens for `GET /api/livekit-token`, and the recorder's own token. LiveKit recording will not start without both variables; see [`LIVEKIT_PREVIEW.md`](../LIVEKIT_PREVIEW.md). |
+| `LIVEKIT_API_KEY`, `LIVEKIT_API_SECRET` | unset | The LiveKit server's key pair. The backend signs two viewer tokens with it (subscribe and data messages, no media publishing): viewer tokens for `GET /api/livekit-token`, and the recorder's own token. LiveKit recording will not start without both variables; see [`LIVEKIT_PREVIEW.md`](../LIVEKIT_PREVIEW.md). |
 | `EGOCAP_TASK_FILES_DIR` | the shipped `data/taskfiles/` | Which task-catalog directory the task-file API serves. |
 | `EGOCAP_RECORDINGS_ENABLED` | off | Turns on the recordings browser's routes. |
 | `EGOCAP_ALLOWED_ORIGINS` | none (same-origin only) | Comma-separated extra origins allowed to open `WS /api/session` — needed when the frontend is hosted separately or behind a reverse proxy. |

examples/egocap/backend/egocap_web/app.py

@@ -152,7 +152,7 @@ def _station_calibration_assets() -> tuple[CalibrationAsset, ...]:
 
 
 def _recorder_livekit_token(room: str) -> str:
-    """A subscribe-only token for the recorder's own LiveKit connection, signed with the backend's key pair."""
+    """A viewer-capability token (subscribe and data messages, no media) for the recorder's LiveKit connection."""
     api_key, signing_key = os.environ.get(API_KEY_ENV, ""), os.environ.get(API_SECRET_ENV, "")
     if not api_key or not signing_key:
         raise RuntimeError(f"{API_KEY_ENV} and {API_SECRET_ENV} must be set to record LiveKit video")

examples/egocap/backend/egocap_web/clock_exchange.py

@@ -0,0 +1,173 @@
+# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Maps a remote machine's monotonic clock onto this one's from four-timestamp exchanges.
+
+The pattern is the one CloudXR uses for its clock sync: this side stamps ``t1`` when it sends a request, the remote
+side stamps ``t2`` on receipt and ``t3`` when it replies, and this side stamps ``t4`` on the reply. The round trip
+without the remote's processing time is ``(t4 - t1) - (t3 - t2)``, and the remote-minus-local offset is
+``((t2 - t1) + (t3 - t4)) / 2``. That point estimate assumes the two directions took equally long; the true offset
+can lie anywhere within half the round trip of it, so that half is reported as the mapping's uncertainty and no
+smaller bound is ever claimed. All times are integer nanoseconds and the offset is signed.
+
+This module only turns exchanges into a mapping. Sending the requests, and the transport under them, belong to
+the caller, and so does the policy for what to do when no mapping is available: a mapping that is missing, stale
+or discontinuous means the remote's timestamps cannot be placed on this clock.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass, replace
+
+#: Exchanges whose round trip exceeds this are not used: a slow or queued exchange says little about the clocks.
+DEFAULT_MAX_RTT_NS = 250_000_000
+#: A mapping older than this (by this side's clock) is stale and is withheld.
+DEFAULT_STALE_AFTER_NS = 5_000_000_000
+#: The most recent accepted exchanges the best one is chosen from.
+DEFAULT_WINDOW = 8
+#: Relative clock rate difference tolerated between two mappings before their offsets are judged discontinuous.
+DEFAULT_DRIFT_ALLOWANCE = 200e-6
+#: Adopted mappings kept as evidence; the newest are the ones a take can be recorded under.
+_MAX_HISTORY = 256
+
+
+@dataclass(frozen=True)
+class Exchange:
+    """One completed four-timestamp exchange: ``t1``/``t4`` on the local clock, ``t2``/``t3`` on the remote's."""
+
+    t1: int
+    t2: int
+    t3: int
+    t4: int
+
+    @property
+    def rtt_ns(self) -> int:
+        """Round trip with the remote's own processing time removed."""
+        return (self.t4 - self.t1) - (self.t3 - self.t2)
+
+    @property
+    def offset_ns(self) -> int:
+        """Remote minus local, assuming symmetric transit."""
+        return ((self.t2 - self.t1) + (self.t3 - self.t4)) // 2
+
+    @property
+    def well_formed(self) -> bool:
+        """Time moved forward on both clocks and the round trip is not negative."""
+        return self.t4 >= self.t1 and self.t3 >= self.t2 and self.rtt_ns >= 0
+
+
+@dataclass(frozen=True)
+class ClockMapping:
+    """An immutable mapping of the remote clock onto the local one; frame conversion only reads this."""
+
+    #: Starts at 1 and changes whenever the estimator starts over or finds a discontinuity.
+    epoch: int
+    offset_ns: int
+    rtt_ns: int
+    #: Half the round trip, plus the drift allowed over the exchange's age: the error unknown path asymmetry and
+    #: clock rate difference can cause.
+    uncertainty_ns: int
+    #: Local time at which the exchange behind this mapping completed.
+    accepted_at_ns: int
+
+    def to_local_ns(self, remote_ns: int) -> int:
+        """The local-clock time of a remote-clock reading."""
+        return remote_ns - self.offset_ns
+
+
+class ClockEstimator:
+    """Keeps the best recent exchange as the current mapping and notices when the clocks jump apart."""
+
+    def __init__(
+        self,
+        *,
+        max_rtt_ns: int = DEFAULT_MAX_RTT_NS,
+        stale_after_ns: int = DEFAULT_STALE_AFTER_NS,
+        window: int = DEFAULT_WINDOW,
+        drift_allowance: float = DEFAULT_DRIFT_ALLOWANCE,
+    ) -> None:
+        self._max_rtt_ns = max_rtt_ns
+        self._stale_after_ns = stale_after_ns
+        self._drift_allowance = drift_allowance
+        self._recent: deque[Exchange] = deque(maxlen=window)
+        self._epoch = 1
+        self._current: ClockMapping | None = None
+        #: Local time of the newest accepted exchange: staleness is about whether replies are still arriving, not
+        #: about how old the best exchange in the window is.
+        self._newest_ns: int | None = None
+        #: Every mapping adopted, oldest first, so a take can record the epochs and quality it was made under.
+        self.history: list[ClockMapping] = []  # trimmed to the newest _MAX_HISTORY
+        #: Set, and left set, when a new best mapping disagrees with the last by more than their uncertainties and
+        #: the allowed drift explain: mappings before it stay as they were, and a take spanning it is not aligned.
+        self.discontinuity = False
+
+    def reset(self) -> None:
+        """Start over, for a new source session or a restarted remote: the old mapping no longer applies."""
+        self._recent.clear()
+        self._current = None
+        self._newest_ns = None
+        self._epoch += 1
+
+    def add(self, exchange: Exchange) -> str | None:
+        """Take one exchange into account; returns why it was rejected, or None when it was accepted."""
+        if not exchange.well_formed:
+            return "malformed exchange"
+        if exchange.rtt_ns > self._max_rtt_ns:
+            return "round trip too long"
+        previous = self._current
+        self._newest_ns = exchange.t4
+        newest = self._mapping_of(exchange)
+        if previous is not None and self._disagrees(previous, newest):
+            # Not a refinement: the clocks moved. Judged on the newest exchange, not on the best of the window, so
+            # a jump is seen at once even when old exchanges have an equally good round trip. Keep the evidence and
+            # start a new epoch from this exchange alone.
+            self.discontinuity = True
+            self._epoch += 1
+            self._recent.clear()
+            self._recent.append(exchange)
+            candidate = self._mapping_of(exchange)
+        else:
+            self._recent.append(exchange)
+            # The exchange with the smallest uncertainty now: half its round trip plus the drift its age could have
+            # added. A low round trip beats a fresher one only while its age stays cheaper; the newest wins ties.
+            best = min(self._recent, key=lambda e: (self._uncertainty_ns(e, exchange.t4), -e.t4))
+            candidate = self._mapping_of(best, exchange.t4)
+        if candidate != previous:
+            self._current = candidate
+            self.history.append(candidate)
+            del self.history[:-_MAX_HISTORY]
+        return None
+
+    def mapping(self, now_ns: int) -> ClockMapping | None:
+        """The current mapping, or None when there is none yet or no exchange arrived within the stale limit.
+
+        The uncertainty is aged to ``now_ns``: while replies are late the offset's drift allowance keeps growing,
+        so a budget check on a returned mapping never sees a figure older than the call. History is not appended.
+        """
+        current = self._current
+        if current is None or self._newest_ns is None or now_ns - self._newest_ns > self._stale_after_ns:
+            return None
+        aged = current.rtt_ns // 2 + int(self._drift_allowance * max(0, now_ns - current.accepted_at_ns))
+        if aged <= current.uncertainty_ns:
+            return current
+        return replace(current, uncertainty_ns=aged)
+
+    def _uncertainty_ns(self, exchange: Exchange, now_ns: int) -> int:
+        """Half the round trip plus the offset drift the exchange's age allows at the allowed relative rate."""
+        return exchange.rtt_ns // 2 + int(self._drift_allowance * max(0, now_ns - exchange.t4))
+
+    def _mapping_of(self, exchange: Exchange, now_ns: int | None = None) -> ClockMapping:
+        return ClockMapping(
+            epoch=self._epoch,
+            offset_ns=exchange.offset_ns,
+            rtt_ns=exchange.rtt_ns,
+            uncertainty_ns=self._uncertainty_ns(exchange, exchange.t4 if now_ns is None else now_ns),
+            accepted_at_ns=exchange.t4,
+        )
+
+    def _disagrees(self, previous: ClockMapping, candidate: ClockMapping) -> bool:
+        elapsed_ns = max(0, candidate.accepted_at_ns - previous.accepted_at_ns)
+        allowed = previous.uncertainty_ns + candidate.uncertainty_ns + self._drift_allowance * elapsed_ns
+        return abs(candidate.offset_ns - previous.offset_ns) > allowed