egocap: time-align kept episodes and estimate tracking time

examples/egocap/AGENTS.md

@@ -72,3 +72,8 @@ Run all three test suites per "Run the tests yourself" above — check each pack
 - Bound mutation request streams before JSON parsing, even when Content-Length is absent or false; return only deliberate public error messages.
 - Dialogs must move, contain and restore focus as well as suppress recording shortcuts.
 - Concurrent coverage runs must use separate `COVERAGE_FILE` paths so one suite cannot overwrite another suite’s measurements.
+## Derived output never touches the raw recording
+
+An aligned or otherwise derived copy of a recording must live in its own tree and read the raw session
+read-only. Check that the output path is not the raw session, inside it or containing it (resolving links)
+before any file is written. An existing derived episode is refused, never replaced.

examples/egocap/LIVEKIT_PREVIEW.md

@@ -37,6 +37,9 @@ livekit:
   [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.
+- Automatic time alignment of a kept episode needs the stream's time to be the sensor's capture event, so use
+  `capture_monotonic_us`. With `monotonic_us`, which is when the frame was ready, the take is recorded but not
+  aligned.
 
 Start CaptureServer with the file (add `--mock` for synthetic tracking):
 

examples/egocap/README.md

@@ -56,7 +56,11 @@ pytest fixture; the rest of `tests/backend` needs no Node/npm.
 Three processes, each in its own terminal, from `egocap/`:
 
 ```bash
-# 1. A standalone mock CaptureServer -- synthetic tracking and camera frames, no hardware
+# 1. A standalone mock CaptureServer -- synthetic tracking and camera frames, no hardware.
+#    This starts with no synthetic delay (--controller-delay-ms/--glove-delay-ms/--delay-jitter-ms add
+#    one), and its cameras are plain CaptureServer JPEG streams, which carry no camera clock evidence:
+#    a recorded take is kept but automatic time alignment refuses it. Aligned output needs the cameras
+#    published through LiveKit (see LIVEKIT_PREVIEW.md).
 uv run capture-server --mock
 
 # 2. The backend, connecting to CaptureServer and serving both frontends -- note the

examples/egocap/backend/egocap_web/app.py

@@ -8,12 +8,14 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
 import secrets
 import sys
 import time
 import uuid
+from collections.abc import Callable
 from contextlib import asynccontextmanager
 from dataclasses import replace
 from pathlib import Path
@@ -32,13 +34,19 @@
     VideoSourceSpec,
 )
 
+from .auto_align import AutoAligner
 from .capture_preview import MAX_CAMERA_JPEG_BYTES
 from .capture_status import CaptureStatusMonitor
 from .config import VisualizationConfig, config_from_env, visualization_from_env
 from .livekit_preview import API_KEY_ENV, API_SECRET_ENV, subscriber_token
 from .livekit_preview import router as livekit_preview_router
 from .livekit_receiver import RECEIVER_FRAMES_PER_STREAM
-from .livekit_recording import LiveKitRecordingSource, recordable_livekit_streams, unrecordable_reason
+from .livekit_recording import (
+    LiveKitRecordingSource,
+    recordable_livekit_streams,
+    sensor_capture_names,
+    unrecordable_reason,
+)
 from .operator_settings import OperatorSettingsStore, operator_settings_path
 from .operator_settings import router as operator_settings_router
 from .preview_routes import router as preview_router
@@ -91,6 +99,8 @@
 # LiveKit stream's decoded frame when that is larger.
 _INGRESS_QUEUE_BYTES = 64 * 1024 * 1024
 _INGRESS_QUEUE_CAPACITY = 4096
+#: How long shutdown waits for queued time alignments before leaving them for the next start.
+_ALIGNER_SHUTDOWN_SECONDS = 30.0
 # LiveKit frames are decoded RGB (megabytes each, ~15 MB at 2560x1984), so
 # the JPEG allowance above holds only a few and a short consumer stall would
 # fail the take. Each recordable stream therefore adds this many of its own
@@ -159,7 +169,12 @@ def _recorder_livekit_token(room: str) -> str:
     return subscriber_token(api_key=api_key, signing_key=signing_key, room=room, now=int(time.time()))
 
 
-def _recording_config_factory(monitor: CaptureStatusMonitor, station_label: str | None, output_root: Path):
+def _recording_config_factory(
+    monitor: CaptureStatusMonitor,
+    station_label: str | None,
+    output_root: Path,
+    on_video_sources: Callable[[frozenset[str], frozenset[str]], None] | None = None,
+):
     """Builds the real ``ServiceConfig`` lazily, read fresh each time the
     coordinator actually opens/reopens a session (not once at import time)
     -- so it reflects whichever cameras CaptureServer has ACTUALLY accepted
@@ -181,6 +196,13 @@ def factory() -> ServiceConfig:
             VideoSourceSpec(s.name, s.width, s.height, s.fps, intrinsics=False)
             for s in recordable_livekit_streams(monitor.native.started)
         )
+        if on_video_sources is not None:
+            # Tells the aligner, once per session open, which cameras this session records and which of them
+            # have a sensor capture time (only LiveKit streams stamped that way do).
+            on_video_sources(
+                frozenset(v.name for v in video_sources),
+                sensor_capture_names(recordable_livekit_streams(monitor.native.started)),
+            )
         # A supported profile must record actual tracking AND the
         # selected cameras -- pico_controller and every configured camera
         # are required for a KEPT result. manus_glove is NOT required:
@@ -393,15 +415,28 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     app.state.imported_task_file_root = operator_settings_file.parent / "taskfiles"
     monitor = CaptureStatusMonitor(config, backend_instance_id=str(uuid.uuid4()))
     livekit_source = LiveKitRecordingSource(
-        lambda: recordable_livekit_streams(monitor.native.started), _recorder_livekit_token
+        lambda: recordable_livekit_streams(monitor.native.started),
+        _recorder_livekit_token,
+    )
+    # Kept episodes are time-aligned in the background when the station configures an index camera.
+    aligner = (
+        AutoAligner(recording_output_root, resolved_station.alignment_index)
+        if resolved_station is not None and resolved_station.alignment_index is not None
+        else None
     )
     coordinator = RecordingSessionCoordinator(
-        _recording_config_factory(monitor, config.station_label, recording_output_root),
+        _recording_config_factory(
+            monitor,
+            config.station_label,
+            recording_output_root,
+            aligner.note_video_sources if aligner is not None else None,
+        ),
         queue_capacity=_INGRESS_QUEUE_CAPACITY,
         queue_bytes=lambda: _ingress_queue_bytes(monitor.native.started),
         readiness_check=_recording_readiness_check(monitor, station_config_error, config.capture_server_url),
         initial_context=initial_context,
         livekit_source=livekit_source,
+        tracking_latency=resolved_station.latency if resolved_station is not None else None,
     )
     monitor.set_recording(coordinator)
     await monitor.start()
@@ -435,6 +470,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         visualization_error=visualization_error,
         task_file_root=task_file_root,
     )
+    # Kept episodes are time-aligned in the background when the station configures an index camera.
+    aligner: AutoAligner | None = None
+    aligner_task: asyncio.Task[None] | None = None
+    if aligner is not None:
+        coordinator.on_episode_kept = aligner.submit
+        aligner_task = asyncio.create_task(aligner.run())
     visualization: VisualizationScene | None = None
     try:
         visualization = _optional_visualization(monitor, visualization_config, visualization_error)
@@ -451,6 +492,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
             # retire any in-flight take cleanly. An unreleased close is
             # surfaced, not silently discarded.
             close_result = await coordinator.close(final=True)
+            if aligner is not None and aligner_task is not None:
+                # After the recorder closed, so its last kept episode is queued; bounded so shutdown is too.
+                await aligner.close(aligner_task, _ALIGNER_SHUTDOWN_SECONDS)
             if not close_result.released:
                 logger.error("recording coordinator did not fully release on shutdown: %s", close_result.issues)
             await monitor.stop()

examples/egocap/backend/egocap_web/auto_align.py

@@ -0,0 +1,212 @@
+# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Time-aligns each kept episode in the background, right after its take is finalized.
+
+One worker handles episodes in the order they were kept. Each alignment runs as a low-priority child process (the
+same ``recording/tools/align_episode.py`` an operator can run by hand), so its CPU and memory never compete with a
+live take inside the recorder's own process, and a shutdown can stop it. An episode that cannot be aligned is
+logged with the reason and left as recorded; the raw episode is never modified. Output is published by one rename,
+so an alignment that is killed leaves no partial result. At startup the worker re-queues the kept episodes that
+still have no aligned copy.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import shutil
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+from egocap.recording.aligned_writer import episode_qualification
+from egocap.recording.episode_layout import (
+    FINALIZATION_KEEP,
+    MANIFEST_DIR,
+    destination_path,
+    read_manifest,
+)
+from recording_service import Disposition, EpisodeResult
+
+logger = logging.getLogger("isaaccapture.egocap_web.auto_align")
+
+#: The command-line aligner, run once per episode.
+_ALIGN_TOOL = Path(__file__).resolve().parents[2] / "recording" / "tools" / "align_episode.py"
+#: A staging directory untouched this long belongs to a run that is no longer going.
+_STALE_STAGING_SECONDS = 24 * 3600
+
+
+@dataclass(frozen=True)
+class _Job:
+    """One kept episode to align: its recording session, its index there, and where it was kept."""
+
+    session_id: str
+    episode_index: int
+    final: Path
+
+    @property
+    def aligned(self) -> Path:
+        """Where this episode's aligned copy goes, beside its group."""
+        return self.final.parent.with_name(f"{self.final.parent.name}_aligned") / self.final.name
+
+    @property
+    def refused_marker(self) -> Path:
+        """A hidden file in the aligned session recording that this episode was refused; deleting it retries."""
+        return self.aligned.parent / f".{self.final.name}.refused"
+
+
+class AutoAligner:
+    """Aligns kept episodes one at a time on its own worker task."""
+
+    def __init__(self, output_root: Path, index_source: str) -> None:
+        """``output_root`` holds the recording sessions; ``index_source`` is the camera that sets the timeline."""
+        self._output_root = output_root
+        self._index_source = index_source
+        # Unbounded: a job is three small fields, and the number waiting is the number of kept episodes.
+        self._queue: asyncio.Queue[_Job] = asyncio.Queue()
+        self._process: asyncio.subprocess.Process | None = None
+        # False while the open session cannot be aligned; set once per session open.
+        self._session_alignable = True
+
+    def note_video_sources(self, recorded: frozenset[str], qualified: frozenset[str]) -> None:
+        """Note whether the session being opened can be aligned; if not, its episodes are not queued.
+
+        It can when the index camera is among the cameras recorded with a sensor capture time and every recorded
+        camera has one, since an episode with any other camera is refused whole.
+        """
+        alignable = self._index_source in qualified and recorded <= qualified
+        if not alignable and self._session_alignable:
+            logger.warning(
+                "automatic time alignment is off: index source %r and every recorded camera need a sensor "
+                "capture time (recorded: %s; with one: %s)",
+                self._index_source,
+                sorted(recorded),
+                sorted(qualified),
+            )
+        self._session_alignable = alignable
+
+    def submit(self, result: EpisodeResult) -> None:
+        """Queue a finished take for alignment; anything other than a kept episode is ignored."""
+        if result.disposition is Disposition.KEPT and result.final_path is not None and self._session_alignable:
+            self._queue.put_nowait(_Job(result.session_id, result.episode_index, result.final_path))
+
+    def scan(self) -> list[_Job]:
+        """Every kept episode, in any session, that has no aligned copy and whose evidence allows one.
+
+        Skips episodes already refused (see ``_Job.refused_marker``), and an episode recorded without qualified
+        timing evidence (an older recording, or a station with no tracking delay set), so history is not retried at
+        every start. Also removes staging directories left by a run that was killed long ago. Blocking file work.
+        """
+        jobs: list[_Job] = []
+        groups: set[Path] = set()
+        for path in sorted(self._output_root.glob(f"*/{MANIFEST_DIR}/episode_*.json")):
+            try:
+                manifest = read_manifest(path)
+                session = path.parent.parent
+                final = (
+                    self._output_root / destination_path(manifest.assignment)
+                    if manifest.assignment
+                    else session / manifest.final_path
+                )
+                job = _Job(manifest.session_id, manifest.episode_index, final)
+                groups.add(final.parent)
+                eligible = (
+                    manifest.storage_phase == "FINALIZED"
+                    and manifest.finalization == FINALIZATION_KEEP
+                    and not job.aligned.exists()
+                    and not job.refused_marker.exists()
+                    and all(
+                        episode_qualification(
+                            final.parent, int(final.name.removeprefix("episode_")), self._index_source
+                        ).values()
+                    )
+                )
+            except Exception:  # noqa: BLE001, S112 - an unreadable record is simply not eligible
+                continue
+            if eligible:
+                jobs.append(job)
+        for group in groups:
+            for stale in group.parent.glob(f".{group.name}_aligned.staging-*"):
+                with contextlib.suppress(OSError):
+                    if time.time() - stale.stat().st_mtime > _STALE_STAGING_SECONDS:
+                        shutil.rmtree(stale, ignore_errors=True)
+        return jobs
+
+    async def run(self) -> None:
+        """Re-queue unfinished work, then align queued episodes until cancelled; failures never stop the worker."""
+        try:
+            for job in await asyncio.to_thread(self.scan):
+                self._queue.put_nowait(job)
+        except Exception as exc:  # noqa: BLE001 - recovery is best effort; live episodes are still aligned
+            logger.warning("time-alignment recovery failed: %s", exc)
+        while True:
+            job = await self._queue.get()
+            try:
+                await self._align(job)
+            except Exception as exc:  # noqa: BLE001 - one episode's failure must not end the worker
+                logger.warning("%s episode %d was not time-aligned: %s", job.session_id, job.episode_index, exc)
+            finally:
+                self._queue.task_done()
+
+    async def close(self, task: asyncio.Task[None], timeout: float) -> None:
+        """Let the queued episodes finish for up to ``timeout`` seconds, then stop the worker ``task``.
+
+        Call after the recorder has closed, so the last kept episode is already queued. An alignment still running
+        at the deadline is killed; it publishes only by one rename, so nothing partial appears, and the episode is
+        picked up again at the next start.
+        """
+        try:
+            await asyncio.wait_for(self._queue.join(), timeout)
+        except asyncio.TimeoutError:
+            logger.warning("time alignment still has queued episodes at shutdown; they are retried at next start")
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+    async def _align(self, job: _Job) -> None:
+        """Align one kept episode in a child process; raises with the reason when it is refused or fails."""
+        if job.aligned.exists():
+            return
+        # The manifest stays in the recording session; a routed episode sits elsewhere, under its slot.
+        command = [
+            "nice",
+            "-n",
+            "10",
+            sys.executable,
+            str(_ALIGN_TOOL),
+            str(job.final.parent),
+            "--episode",
+            str(int(job.final.name.removeprefix("episode_"))),
+            "--index-source",
+            self._index_source,
+            "--manifest-session",
+            str(self._output_root / job.session_id),
+            "--manifest-episode",
+            str(job.episode_index),
+        ]
+        process = await asyncio.create_subprocess_exec(
+            *command, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE
+        )
+        self._process = process
+        try:
+            _, stderr = await process.communicate()
+        except asyncio.CancelledError:
+            # Cancelled at shutdown: stop the child rather than leave it running past the recorder.
+            process.kill()
+            await process.wait()
+            raise
+        finally:
+            self._process = None
+        reason = stderr.decode(errors="replace").strip()
+        if process.returncode == 1:
+            # Exit 1 is a refusal, which is the same every time, so remember it instead of retrying at every start;
+            # any other failure (an unreadable manifest, a killed child) stays retryable.
+            job.refused_marker.parent.mkdir(parents=True, exist_ok=True)
+            job.refused_marker.write_text(reason + "\n", encoding="utf-8")
+        if process.returncode != 0:
+            raise RuntimeError(reason or f"aligner exited with status {process.returncode}")
+        logger.info("%s episode %d was time-aligned", job.session_id, job.episode_index)