fix: eliminate export-time episode truncation and camera-ahead-of-action violations

README.md

@@ -43,6 +43,38 @@ cameras. Examples: [`examples/`](examples/). Schema:
 config field, the export formats, camera resizing, and how the trimming modes
 choose episode boundaries (with figures).
 
+## Live sync + parallel export (`tools/pipeline.sh`)
+
+Pulls recordings while collection is still running and exports them in
+parallel as they finish, instead of waiting for collection to end and then
+exporting one at a time. Worker count and memory budget are computed from the
+machine's own RAM/cores at startup. See [`tools/AGENT.md`](tools/AGENT.md) for
+the full design (crash-safety, bisection, why it's shaped this way).
+
+**Two machines** (collector elsewhere, export runs here) — needs passwordless
+SSH to the collector (`ssh-copy-id`):
+
+```bash
+tools/pipeline.sh                    # default: --mode remote
+```
+
+Edit `REMOTE_HOST`/`REMOTE_DIRS`/`WATCH_DIR` at the top of the script, or
+override per-run via `PIPELINE_REMOTE_HOST`, `PIPELINE_REMOTE_DIRS`, etc.
+(every config value is a `PIPELINE_*` env var — see the top of the script).
+
+**One machine** (collection already finished, or writing straight into a
+local dir) — no network involved:
+
+```bash
+PIPELINE_MODE=local PIPELINE_WATCH_DIR=/path/to/recordings tools/pipeline.sh
+```
+
+Both modes end the same way: once nothing new has shown up for
+`PIPELINE_IDLE_MINUTES` (or immediately, for a backlog that was never live),
+it merges every batch into one dataset at `<EXPORT_ROOT>_merged`. Restarting
+after a crash/kill is always safe — already-exported recordings are never
+redone.
+
 ## Tests
 
 ```bash

docs/export-guide.md

@@ -66,10 +66,13 @@ directly and `--recordings-dir` is ignored.
 | `cameras`                | list[object]                | `[]`              | Camera streams → `observation.images.<source>`. Each may set `width`/`height` to resize. See [Cameras](#cameras--resizing).                                                                    |
 | `trimming`               | object                      | `all_present`     | How episode start/end bounds are chosen. See [Trimming](#trimming-the-important-part).                                                                                                         |
 | `max_episode_duration_s` | float \| null               | `null` (no limit) | Reject a segment if its _raw_ recording span exceeds this many seconds — unrelated to trimming. See [Rejecting stuck or left-running recordings](#rejecting-stuck-or-left-running-recordings). |
-| `task_description`       | string                      | `"task"`          | Natural-language task label written to every frame.                                                                                                                                            |
+| `task_description`       | string                      | `"task"`          | Fallback task label written to every frame when `task_field` is unset (or its meta.json field is missing for a given episode).                                                                |
+| `task_field`             | string \| null              | `null`            | meta.json field name (e.g. `"task"`) holding each episode's own natural-language instruction — lets `task` vary per episode instead of being fixed dataset-wide. Falls back to `task_description`. Requires local export (same as `episode_metadata`).            |
 | `dataset_id`             | string                      | `nova/dataset`    | Dataset identifier — the LeRobot `repo_id` (also used for viz and Hugging Face push).                                                                                                          |
 | `version`                | int                         | `1`               | Config schema version. Leave at `1`.                                                                                                                                                           |
 
+`episode_metadata` values may be any JSON scalar type (string, number, boolean) — not float-only. A field missing from a given episode's `meta.json` is filled with `null` in that episode's row.
+
 ## Formats
 
 - **`lerobot_v3`** — a LeRobot v3.0 dataset (Parquet + MP4 + metadata). Use this
@@ -194,6 +197,17 @@ depends on the signal's units and noise floor:
 Start at `0.01` and raise it only if idle time is leaking in; if episodes come out
 suspiciously short, your threshold is above the real motion and should come down.
 
+**`threshold` is per *consecutive sample*, not total displacement**, so the right
+value depends on the source's sample rate as much as on its units. A slow arm
+logged at 64 ms may never move more than ~0.01 rad between two samples, in which
+case `threshold: 0.01` trims almost the entire episode away. The "no change
+exceeds it" fallback above does *not* save you here — a handful of samples still
+cross, so the episode collapses to a second or two instead of falling back. The
+export logs a warning whenever trimming keeps less than half the raw span; treat
+it as a signal to lower `threshold` or switch to `all_present`. When the action
+stream itself only exists while the task is being commanded, `all_present`
+already trims the idle lead-in for free and is the safer choice.
+
 ### Modes compared
 
 ![All trim modes compared](img/all_modes_compared.png)
@@ -225,7 +239,9 @@ check, so this doesn't need to be tight.
 - **Just want everything recorded?** `all_present` (default).
 - **A signal cleanly brackets the task?** `signal_presence` on that source.
 - **Need to cut idle lead-in/out automatically?** `signal_change` on a motion
-  signal (e.g. `joint_positions`), `threshold` ≈ `0.01`, `tail_ms` ≈ `500`.
+  signal (e.g. `joint_positions`), `threshold` ≈ `0.01`, `tail_ms` ≈ `500` — but
+  check the threshold against your source's actual inter-sample motion first
+  (see [Tuning `threshold`](#signal_change)).
 - **Dataset too big / training input smaller?** Set camera `width`/`height`.
 - **A few episodes are way longer than the rest (stuck sensor, forgotten recording)?**
   Set `max_episode_duration_s` to drop them.

docs/investigations/worker-early-exit.md

@@ -0,0 +1,66 @@
+# Symptom: worker exits early, permanently down one worker
+
+## Observed
+
+During a `tools/pipeline.sh` run exporting the `choreo2` dataset
+(`pick_and_place_sim_20260805_191245`, 1001 recordings, `--mode local`,
+started 2026-08-06 ~17:07), worker `w3` logged:
+
+```
+[w3] [17:17:58] [worker] committed batch_20260806_171301_w3_1147218_10206 (8 recording(s))
+[w3] [17:17:59] [worker] no candidates and collection done, exiting
+```
+
+...and never restarted. From then on, only `w0`, `w1`, `w2` were running
+(confirmed via `ps aux | grep nova-data-cli`), each still committing batches
+every ~5 minutes.
+
+## Why it doesn't add up
+
+At 17:17:59:
+- ~160 episodes committed (21 batches × 8, from the `w0-w3` logs)
+- At most ~32 in flight (4 workers × `CHUNK=8`)
+- That leaves **~800 unclaimed candidates** still sitting in
+  `/home/sebi/ws/Data/raw_data/choreo2/pick_and_place_sim_20260805_191245`
+
+`collection_done` was legitimately set at 17:07:56 (the earlier remote-mode
+acquire pass had already finished its final rsync at 17:01:36, so all 1001
+recordings were genuinely present on disk by the time the local-mode watcher
+started). So the "collection is done" half of the exit condition is correct.
+
+The "no candidates" half is not plausible given ~800 unclaimed recordings
+should have been sitting right there for `list_candidates()` to find.
+
+## Effect
+
+- One of four workers permanently disappears mid-run, with no log line
+  indicating an error — just the same message a worker prints on genuine,
+  correct completion.
+- The remaining three workers keep making progress, so nothing looks wrong
+  at a glance (batches keep committing), but overall throughput drops to
+  75% for the rest of the run.
+- Nothing in the pipeline's own signals (logs, exit code, `.pipeline/`
+  state) distinguishes this from the expected end-of-run shutdown.
+
+## Reproduction notes
+
+Not reproduced in isolation — several attempts to trigger the same premature
+`set -e`-style abort or scan failure in `list_candidates`/`is_candidate`
+(including under a synthetic 50-recording watch dir with the exact same
+functions) came back clean, so the underlying trigger is genuinely
+load-dependent, not a deterministic logic bug in the scan itself.
+
+## Fix
+
+`role_worker`'s exit condition trusted a *single* empty scan as proof there
+was no more work, the moment `COLLECTION_DONE` was also true — with no
+tolerance for a transient glitch in that one scan (e.g. a `find` fork failing
+under the exact system load described above). `role_supervisor`'s drain loop
+had the same shape and a worse consequence (a false-empty scan there would
+merge before everything was actually exported, not just drop a worker).
+
+Both now require repeated confirmation before trusting "nothing left" — 3
+consecutive empty scans for a worker to exit, 2 consecutive clean passes for
+the supervisor to proceed to merge — the same pattern `role_acquire`'s
+idle-detection already used and this code didn't. See `tools/pipeline.sh`
+`role_worker`/`role_supervisor` and `tools/AGENT.md`.

docs/investigations/worker-pool-static-during-collection.md

@@ -0,0 +1,83 @@
+# Symptom: many workers running, but most idle for the whole collection phase
+
+## Observed
+
+During a long-running `tools/pipeline.sh` run, `ps aux | grep nova-data-cli`
+showed a large number of workers up, but only a handful actively logging
+`[worker] committed batch_...` lines — the rest sat silent, doing nothing,
+for most of the run.
+
+## Why it happens
+
+`role_supervisor` calls `top_up_workers` exactly once, then blocks:
+
+```bash
+top_up_workers
+wait "$acquire_pid" || true
+```
+
+`top_up_workers` sizes the pool from `compute_workers()`, which is a pure
+function of **system memory/cores at the instant it's called** — it has no
+idea how many recordings are actually claimable yet. At the moment collection
+starts, `list_candidates` may already show a nontrivial backlog (e.g. a
+`--mode remote` run resuming against a host with data already sitting there),
+so this first call can spawn the machine's *entire* memory-supported worker
+count immediately — before there's any evidence that candidates will keep
+arriving at a matching rate.
+
+`wait "$acquire_pid"` then blocks the supervisor's own control flow until the
+**entire acquisition phase finishes** (which can be hours, for a live
+collection). No further `top_up_workers` call happens until then — the
+drain-loop's periodic top-up (every 60s) only starts *after* `COLLECTION_DONE`
+is set. So whatever pool size the very first call happened to produce is what
+runs, unmonitored, for the whole collection window, regardless of how the
+real backlog behaves afterward (e.g. if it trickles in far slower than
+`workers × CHUNK` can consume, or if more memory frees up later as batches
+commit).
+
+The code comment directly above `top_up_workers` claims "re-evaluated on
+every call... so the worker count tracks memory headroom as it opens up" —
+true of the *function*, but that guarantee only holds once collection is
+already done; during collection, the function is simply never called again
+to exercise it.
+
+## Effect
+
+Most workers spawned at that first call end up polling `list_candidates`
+every 30s and finding nothing (no log line on a non-final empty scan — see
+`role_worker`'s `empty_scans` loop — so this is invisible unless you're
+watching `ps aux`), while a handful of workers that won the claim race keep
+grinding through their batches. Throughput looks fine at a glance (batches do
+keep committing), but most of the machine's provisioned worker capacity sits
+unused for the run.
+
+## Fix
+
+Two changes, both in `tools/pipeline.sh` `role_supervisor`/`top_up_workers`:
+
+1. Keep calling `top_up_workers` on the same 60s cadence during acquisition
+   too, not just after `COLLECTION_DONE`:
+
+   ```bash
+   top_up_workers
+   while kill -0 "$acquire_pid" 2>/dev/null; do
+     sleep 60
+     top_up_workers
+   done
+   wait "$acquire_pid" || true
+   ```
+
+2. `top_up_workers` now caps the new-worker count against the actual
+   unclaimed backlog (`list_candidates | wc -l`) on every call, not only once
+   `COLLECTION_DONE` is set. This cap already existed for the post-collection
+   drain phase; it was deliberately *not* applied during collection because a
+   single `list_candidates` read coming back empty (e.g. `is_candidate`'s
+   60s-untouched freshness gate momentarily reading zero) would otherwise
+   permanently block the very first spawn under the old one-shot-call
+   structure. Change (1) removes that risk: a transient zero reading now just
+   costs one 60s poll, not the rest of the run, so the cap can safely apply
+   throughout.
+
+Together these mean the pool actually scales with the live backlog and
+memory headroom as collection proceeds, instead of being frozen at whatever
+the single startup snapshot allowed.