Add continuous-time MoT world model and offline data pipeline

docs/cluster-pipeline.md

@@ -131,6 +131,45 @@ status and signal names and never requeues a cancellation or ordinary child
 failure. A failed upstream dependency automatically retires the dependent
 request; it does not permit bypassing its smoke gate.
 
+For a single-process trainer sharing the runner's Linux container, the generic
+`ice_requeue_runner.py` also supports opt-in `--checkpoint-forwarding trainer-pid`.
+Run the runner from the exact source root, and pass its own Python executable
+as the direct child command (no shell or `srun` wrapper). The existing
+`runtime.slurm_requeue_owner=runner` callback installs SIGUSR2 before publishing
+readiness. The runner binds that proof to its nonce, job/restart, source hashes,
+UID, node, boot ID, PID namespace, cgroup, executable, working directory and
+process start time; `pidfd` prevents signaling a reused PID. It signals only its
+known child and never invokes `scancel`, including on failure. This route requires
+one task/node/process and a working `scontrol` client inside the same container;
+verify the site's client/config/Munge-socket bindings with a read-only query and
+a dedicated requeue test before training. Do not expose or copy a Munge key.
+
+SIGUSR2 only latches a request; the callback saves at a completed batch hook,
+publishes an immutable full-state checkpoint and receipt, and the runner strictly
+validates a newer optimizer step before asking Slurm to requeue. Work after that
+checkpoint can be replayed after restart. Keep periodic checkpoint/validation
+cadences unchanged; allocation-boundary recovery saves are additional. Pass
+`--target-global-step` with a completion sentinel to reject an early zero exit or
+an older COMPLETE record. Boundary and validator time must fit the signal lead;
+restart limits must cover the measured run budget. When the runner lives inside
+an `srun` container, a batch-only Slurm signal may not reach it. Opt in to
+`--allocation-boundary-lead-seconds` to arm a relative kernel timer from the
+native `SLURM_JOB_END_TIME` epoch before discovery. Missing/invalid/past end times
+fail closed; already inside the lead window uses the existing early-boundary
+path. The timer only latches the same safe save request and cannot override
+TERM/INT. It is disarmed/restored on every runner exit and is not inherited by
+the trainer. This route requires a fixed allocation end time for the attempt;
+do not shorten its walltime externally without coordinating the boundary.
+Prove one actual deadline-triggered requeue separately from an injected signal.
+When mid-validation continuation is unverified, set
+`runtime.slurm_signal_train_batch_only=true`. Requests then remain latched through
+validation and epoch/fit-end hooks until the next completed optimizer update.
+This avoids conflicting with a scheduled checkpoint at the same global step.
+The allocation lead and signal grace must exceed the measured **complete**
+validation pass, longest training step, save and strict checkpoint verification;
+a small diagnostic validation cap is not a production timing bound.
+Defaults and the existing `slurm-steps` transport remain unchanged.
+
 In this Action Flow branch, `egomimic/eval/artifact_paths.py` supplies the same
 namespace rule to EnergyScore, UNITE diagnostics, and Action Flow diagnostics.
 The Action Flow smoke verifier selects a matching active job/restart namespace

egomimic/benchmarks/libero/replay.py

@@ -223,15 +223,22 @@ def fetch(row):
     return raw / suite
 
 
-def rebase_demo_xml(xml, robosuite_root, assets_root):
+def rebase_demo_xml(xml, robosuite_root, assets_root, asset_remaps=()):
     """Relocate saved asset paths, preserving every model and physics parameter."""
     tree = ET.fromstring(xml)
     for element in tree.iter():
         old = element.get("file")
         if old is None:
             continue
         parts = Path(old).parts
-        if "robosuite" in parts:
+        matches = [(Path(prefix), Path(target).resolve()) for prefix, target in asset_remaps
+                   if Path(old).is_relative_to(Path(prefix))]
+        if len(matches) > 1:
+            raise ValueError(f"Ambiguous demonstration asset remap: {old}")
+        if matches:
+            prefix, target = matches[0]
+            new = target / Path(old).relative_to(prefix)
+        elif "robosuite" in parts:
             index = max(i for i, part in enumerate(parts) if part == "robosuite")
             new = Path(robosuite_root).joinpath(*parts[index + 1 :])
         elif "assets" in parts: