[skycap] Mirror records to remote storage, and say where each is at finish

[kouroshHakha]

What does this PR do?

TLDR: a capture server can copy each record it writes to remote storage (any fsspec URL: s3://, gs://, ...) in the background, and finish says where each record is. The local record_dir stays the source of truth; the copy fails open. Splits the storage half out of #2333; the W&B half is #2351.

How it works

• Options: --record-mirror URL and --record-mirror-config JSON, or record_mirror and record_mirror_config on CaptureServer and CaptureService. Both need record_dir.

  skycap serve ... --record-dir ./record --record-mirror s3://bucket/run-7 \
    --record-mirror-config '{"exclude": ["experts", "sampling_mask"], "timeout": 120}'

  The config keys are exclude, workers, queue_size, timeout, attempts, backoff, shutdown_timeout and storage_options. fsspec comes from the new optional remote extra and is imported only when a mirror is made.

• RecordMirror (skycap/mirror.py): worker threads copy a record's sidecars, then its document. It fails open:

  • the queue is bounded, and a record that finds it full is dropped;
  • a copy that overruns its timeout is abandoned, not retried, since it may still land;
  • transient errors get bounded retries with backoff;
  • shutdown waits until a deadline, then drops what's left.

  Every loss is logged and counted on /healthz under record_mirror.

• exclude skips sidecar kinds when copying; the document is copied unchanged. A copy may therefore lack sidecar files its manifest lists, and readers treat those as absent: without tokens they show message text only, and they need nothing else, since only training uses experts and sampling_mask.

• finish returns record: {host, path, mirror, files}, also available as FinishResult.record (a RecordLocation).

  • host is the machine path is on, so another node can reach a record that exists only there. RecordLocation.local gives host:path, the scp form.
  • host comes from --record-host if set, otherwise from a service's advertise_host unless that's loopback, otherwise from the machine's primary IP.
  • files lists the record's files as the mirror holds them, or as the record directory does without a mirror.

• format.md:

  • every sidecar is optional, and a copy (unlike the record directory) may lack sidecar files its manifest lists, which readers treat as absent;
  • the mirror;
  • finish's record.

Test plan

• skycap/tests/test_mirror.py: copies are byte-for-byte, sidecars first; a failing or slow store fails open; transient errors are retried and permanent ones aren't; a sidecar failure keeps the document out; a timeout is abandoned without a duplicate; a full queue drops; shutdown respects its deadline; exclude (an unchanged document, and missing sidecars reading as absent); record.host; the config over the CLI and the API; finish's record.files.
• The skycap suite passes with --extra tokens --extra remote (230) and on a clean install (224 passed, 1 skipped). test_imports.py checks that fsspec isn't loaded without a mirror.
• uv lock --check and pre-commit are clean.

🤖 Generated with Claude Code

---

Note

Medium Risk
New async upload path and changed finish API/record read semantics; failures are isolated from trajectories but remote copies may be incomplete or missing sidecars by design.

Overview
Adds background mirroring of finished trajectory records to any fsspec URL (s3://, gs://, …) via --record-mirror / CaptureService(record_mirror=…) and the new optional remote extra (fsspec). Local --record-dir stays authoritative; RecordMirror copies sidecars then the document on worker threads with a bounded queue, per-file timeouts, transient retries, and fail-open behavior (drops/timeouts never fail finish). /healthz exposes record_mirror counters; shutdown drains the queue up to a deadline.

finish now returns record (RecordLocation: host, path, optional mirror, files) so trainers can fetch from the node or remote URI; --record-host overrides the reported address. Mirror exclude skips sidecar kinds in the upload while leaving the JSON manifest unchanged; record.load and format.md treat missing listed sidecars as absent (e.g. text-only when tokens is excluded).

CLI validation ties mirror to record_dir; extensive coverage in test_mirror.py.

^{Reviewed by Cursor Bugbot for commit 1a29349. Bugbot is set up for automated code reviews on this repo. Configure here.}

[gemini-code-assist]

Code Review

This pull request introduces a background record mirroring feature using fsspec to copy written trajectory records to remote storage. A critical issue was identified in the shutdown logic of the RecordMirror where worker threads can leak and block indefinitely on self._queue.get() if they do not join before the shutdown deadline. It is recommended to implement a timeout on the queue fetch and check the stopping event to ensure clean thread termination.

[greptile-apps]

Confidence Score: 4/5

[Medium risk] Adds background mirroring of trajectory records to remote storage.

The PR should not merge until a concurrent submission can no longer be accepted after mirror shutdown.

Findings

1.  Submission after shutdown ▶
2.  Pending count briefly underreports ▶

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  F[Finish or shutdown] --> W[Write local record]
  W --> Q[Submit to mirror queue]
  Q --> C[Worker copies sidecars]
  C --> D[Worker copies document]
  Q --> H[Health counters]
  W --> R[Finish returns record locations]

Loading

_{Reviews (1) · Last reviewed commit: "[skycap] Mirror each record to remote st..."}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit dc3a553. Configure here.}

[kouroshHakha]

leaving comments