fix(eval,cookbooks,ci): harden evaluation streaming, multi-node checks, and preview generation

.github/workflows/validate-notebooks.yml

@@ -97,5 +97,6 @@ jobs:
         shell: bash
         run: |
           mapfile -d '' notebooks < <(git ls-files -z -- '*.ipynb')
+          (( ${#notebooks[@]} )) || exit 0
           # Catch syntax and runtime-invalid constructs without enforcing style on legacy notebooks.
           ruff check --output-format=github --select=E9,F63,F7,F82 -- "${notebooks[@]}"

cookbooks/cosmos3/generator/action/finetune/launch_sft_action_fd_droid_posttrain.sh

@@ -101,6 +101,13 @@ fi
 TRAILING_ARGS=(-- "${TAIL_OVERRIDES[@]}")
 
 TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}" --master_port="${MASTER_PORT:-50012}")
+if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then
+    # Fail fast with an actionable message instead of torchrun's opaque
+    # rendezvous error later.
+    : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+fi
 [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES")
 [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK")
 [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR")

cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_droid_nano.sh

@@ -71,6 +71,13 @@ fi
 
 TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}")
 TORCHRUN_ARGS+=(--master_port="${MASTER_PORT:-50012}")
+if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then
+    # Fail fast with an actionable message instead of torchrun's opaque
+    # rendezvous error later.
+    : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+fi
 [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES")
 [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK")
 [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR")

cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_10_nano.sh

@@ -61,6 +61,13 @@ fi
 
 TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}")
 TORCHRUN_ARGS+=(--master_port="${MASTER_PORT:-50012}")
+if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then
+    # Fail fast with an actionable message instead of torchrun's opaque
+    # rendezvous error later.
+    : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+fi
 [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES")
 [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK")
 [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR")

cookbooks/cosmos3/generator/action/finetune/launch_sft_action_policy_libero_all_nano.sh

@@ -68,6 +68,13 @@ fi
 
 TORCHRUN_ARGS=(--nproc_per_node="${NPROC_PER_NODE:-8}")
 TORCHRUN_ARGS+=(--master_port="${MASTER_PORT:-50012}")
+if [[ -n "${NNODES:-}" || -n "${NODE_RANK:-}" || -n "${MASTER_ADDR:-}" ]]; then
+    # Fail fast with an actionable message instead of torchrun's opaque
+    # rendezvous error later.
+    : "${NNODES:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${NODE_RANK:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+    : "${MASTER_ADDR:?set NNODES, NODE_RANK and MASTER_ADDR together for multi-node}"
+fi
 [[ -n "${NNODES:-}" ]] && TORCHRUN_ARGS+=(--nnodes="$NNODES")
 [[ -n "${NODE_RANK:-}" ]] && TORCHRUN_ARGS+=(--node_rank="$NODE_RANK")
 [[ -n "${MASTER_ADDR:-}" ]] && TORCHRUN_ARGS+=(--master_addr="$MASTER_ADDR")

cookbooks/cosmos3/generator/transfer/preview_helpers.py

@@ -71,27 +71,36 @@ def load_transfer_spec(control: str) -> dict:
 def make_preview(src: Path, crf: int = 28) -> Path:
     preview = src.with_name(f"{src.stem}_preview.mp4")
     if not preview.exists() or preview.stat().st_mtime < src.stat().st_mtime:
-        subprocess.run(
-            [
-                _ffmpeg_exe(),
-                "-y",
-                "-loglevel",
-                "error",
-                "-i",
-                str(src),
-                "-c:v",
-                "libx264",
-                "-crf",
-                str(crf),
-                "-preset",
-                "veryfast",
-                "-an",
-                "-pix_fmt",
-                "yuv420p",
-                str(preview),
-            ],
-            check=True,
-        )
+        # Encode beside the target (keeping the .mp4 suffix so ffmpeg infers
+        # the container), then swap atomically so an interrupted cell never
+        # leaves a half-written video behind under the final name.
+        partial = src.with_name(f"{src.stem}_preview.partial.mp4")
+        try:
+            subprocess.run(
+                [
+                    _ffmpeg_exe(),
+                    "-y",
+                    "-loglevel",
+                    "error",
+                    "-i",
+                    str(src),
+                    "-c:v",
+                    "libx264",
+                    "-crf",
+                    str(crf),
+                    "-preset",
+                    "veryfast",
+                    "-an",
+                    "-pix_fmt",
+                    "yuv420p",
+                    str(partial),
+                ],
+                check=True,
+            )
+        except BaseException:
+            partial.unlink(missing_ok=True)
+            raise
+        os.replace(partial, preview)
     return preview
 
 

evaluation/cosmos3/generator/paibench_c/run_paibench_c.sh

@@ -10,11 +10,11 @@
 # automatically on first run.
 #
 # --- Quick start --------------------------------------------------------------
-#   # Smoke-test: 1 task, edge, Cosmos3-Nano
+#   # Default demo: 4 tasks, edge control, Cosmos3-Nano
 #   bash run_paibench_c.sh
 #
-#   # Demo with 4 tasks (default)
-#   bash run_paibench_c.sh
+#   # Minimal smoke: 1 task, edge control
+#   PAIBENCH_C_NUM_SAMPLES=1 bash run_paibench_c.sh
 #
 #   # Full 600-task run, all modalities, Cosmos3-Super
 #   PAIBENCH_C_NUM_SAMPLES=600 \
@@ -704,6 +704,16 @@ unset MPLBACKEND  # prevent Jupyter's inline backend from leaking into subproces
 
 _ckpt_slug="${PAIBENCH_C_CHECKPOINT//\//-}"
 
+# Evaluation shards one torchrun worker per visible GPU. Recompute the real CUDA
+# device list here instead of trusting COSMOS3_NUM_GPUS, so bumping that knob
+# past what CUDA_VISIBLE_DEVICES exports cannot oversubscribe workers below.
+IFS=',' read -r -a _eval_gpu_ids <<< "${CUDA_VISIBLE_DEVICES}"
+_max_eval_gpus=${#_eval_gpu_ids[@]}
+if (( _max_eval_gpus == 0 )); then
+    log "  WARNING: CUDA_VISIBLE_DEVICES exposes no GPUs; capping evaluation at 1 GPU"
+    _max_eval_gpus=1
+fi
+
 # Use a robust python for printing results - prefer cosmos3 venv if functional,
 # fall back to system python3 (only stdlib used).
 _py_for_summary() {
@@ -720,6 +730,10 @@ for _mod in $PAIBENCH_C_MODALITIES; do
     _videos_dir="$_videos_parent/videos"
     _metrics_out="$_videos_parent/metrics.json"
     _eval_ngpu="$(int_min "$PAIBENCH_C_NUM_SAMPLES" "$COSMOS3_NUM_GPUS")"
+    if (( _eval_ngpu > _max_eval_gpus )); then
+        log "  WARNING: clamping evaluation GPUs $_eval_ngpu -> $_max_eval_gpus (visible: $CUDA_VISIBLE_DEVICES)"
+        _eval_ngpu=$_max_eval_gpus
+    fi
 
     # Verify videos exist before launching torchrun to surface errors early.
     if [[ ! -d "$_videos_dir" ]]; then

evaluation/cosmos3/generator/paibench_g/merge_result_selfcheck.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""Self-check for run_motion_smoothness_sharded.write_merged_result (CPU-only).
+
+Covers the merge-step contract, including the partial-results behaviour added so
+permanently failed videos no longer discard cleanly computed scores:
+
+1. complete run       -> overall mean, no missing list in the payload;
+2. failed video       -> partial mean plus result-file ``missing_videos`` key;
+3. strict mode        -> default call still raises when gaps exist;
+4. foreign records    -> records outside the requested set are always rejected.
+
+Run: python merge_result_selfcheck.py   (exit 0 == pass)
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from run_motion_smoothness_sharded import write_merged_result
+
+
+def main() -> int:
+    with tempfile.TemporaryDirectory() as tmp_str:
+        tmp = Path(tmp_str)
+
+        # --- 1. complete run --------------------------------------------------
+        d1 = tmp / "complete"
+        d1.mkdir()
+        r1 = d1 / "result.json"
+        # Absolute canonical temp paths: Path(...).absolute() leaves them intact,
+        # matching how collect_saved_records keys records on every platform.
+        vids = [str(tmp / f"v{i}.mp4") for i in range(3)]
+        (d1 / "worker_00.json").write_text(json.dumps([
+            {"video_path": vids[0], "video_results": 0.5},
+            {"video_path": vids[1], "video_results": 0.25},
+        ]))
+        (d1 / "worker_01.json").write_text(json.dumps([
+            {"video_path": vids[2], "video_results": 0.25},
+        ]))
+        score, missing = write_merged_result(output_dir=d1, result_file=r1, videos=vids)
+        assert math.isclose(score, 1 / 3, rel_tol=1e-12), score
+        assert missing == [], missing
+        assert "missing_videos" not in json.loads(r1.read_text())
+
+        # --- 2. permanently failed video -> partial payload --------------------
+        d2 = tmp / "partial"
+        d2.mkdir()
+        r2 = d2 / "result.json"
+        vids2 = [str(tmp / f"w{i}.mp4") for i in range(2)]
+        (d2 / "worker_00.json").write_text(json.dumps([
+            {"video_path": vids2[0], "video_results": 0.75},
+        ]))
+        score_b, missing_b = write_merged_result(
+            output_dir=d2, result_file=r2, videos=vids2, require_complete=False
+        )
+        assert score_b == 0.75 and missing_b == [vids2[1]], (score_b, missing_b)
+        payload = json.loads(r2.read_text())
+        assert payload["missing_videos"] == [vids2[1]]
+        assert len(payload["motion_smoothness"][1]) == 1
+
+        # --- 3. strict mode (default) rejects gaps -----------------------------
+        try:
+            write_merged_result(output_dir=d2, result_file=r2, videos=vids2)
+        except RuntimeError:
+            pass
+        else:
+            raise AssertionError("strict merge accepted missing results")
+
+        # --- 4. records outside the requested set are rejected ------------------
+        d4 = tmp / "unexpected"
+        d4.mkdir()
+        (d4 / "worker_00.json").write_text(json.dumps([
+            {"video_path": str(tmp / "stray.mp4"), "video_results": 0.0},
+        ]))
+        try:
+            write_merged_result(
+                output_dir=d4, result_file=d4 / "result.json",
+                videos=[str(tmp / "kept.mp4")],
+            )
+        except RuntimeError as exc:
+            assert "outside the input manifest" in str(exc), exc
+        else:
+            raise AssertionError("unexpected record accepted")
+
+        print("merge_result_selfcheck: PASS (complete/partial/strict/unexpected)")
+        return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

evaluation/cosmos3/generator/paibench_g/run_motion_smoothness_sharded.py

@@ -15,6 +15,9 @@
 
 
 RUNNER_VERSION = 2
+# Quick second attempt before recording a video as permanently errored; AMT
+# scoring failures (decoder hiccups, transient CUDA errors) are often transient.
+VIDEO_RETRIES = 1
 
 
 def parse_args() -> argparse.Namespace:
@@ -156,32 +159,50 @@ def worker_main(
                 continue
             if video_path is None:
                 break
-            try:
-                score = float(motion.motion_score(video_path))
+            final_error = ""
+            for _attempt in range(1 + VIDEO_RETRIES):
+                try:
+                    score = float(motion.motion_score(video_path))
+                except Exception:
+                    final_error = traceback.format_exc()
+                    continue
                 records.append({"video_path": video_path, "video_results": score})
                 write_json_atomic(output_path, records)
                 status_queue.put(("done", worker_id, video_path, score))
-            except Exception:
-                status_queue.put(("error", worker_id, video_path, traceback.format_exc()))
+                break
+            else:
+                status_queue.put(("error", worker_id, video_path, final_error))
         status_queue.put(("exit", worker_id, physical_gpu, ""))
     except Exception:
         status_queue.put(("fatal", worker_id, physical_gpu, traceback.format_exc()))
         raise
 
 
-def write_merged_result(*, output_dir: Path, result_file: Path, videos: list[str]) -> float:
+def write_merged_result(
+    *,
+    output_dir: Path,
+    result_file: Path,
+    videos: list[str],
+    require_complete: bool = True,
+) -> tuple[float, list[str]]:
     saved = collect_saved_records(output_dir)
     target = set(videos)
     unexpected = set(saved) - target
     if unexpected:
         raise RuntimeError(f"found {len(unexpected)} results outside the input manifest")
     missing = [video_path for video_path in videos if video_path not in saved]
-    if missing:
+    if missing and require_complete:
         raise RuntimeError(f"missing {len(missing)} motion-smoothness results")
-    ordered = [saved[video_path] for video_path in videos]
+    scored = [video_path for video_path in videos if video_path in saved]
+    if not scored:
+        raise RuntimeError("no motion-smoothness results were computed")
+    ordered = [saved[video_path] for video_path in scored]
     mean_score = fmean(record["video_results"] for record in ordered)
-    write_json_atomic(result_file, {"motion_smoothness": [mean_score, ordered]})
-    return mean_score
+    payload: dict[str, object] = {"motion_smoothness": [mean_score, ordered]}
+    if missing:
+        payload["missing_videos"] = missing
+    write_json_atomic(result_file, payload)
+    return mean_score, missing
 
 
 def main() -> None:
@@ -248,6 +269,7 @@ def main() -> None:
         flush=True,
     )
 
+    run_failed = False
     if pending:
         context = mp.get_context("spawn")
         task_queue = context.Queue()
@@ -314,8 +336,13 @@ def main() -> None:
             for process in workers:
                 process.join()
             bad_exit_codes = [process.exitcode for process in workers if process.exitcode != 0]
-            if bad_exit_codes or failures:
-                raise RuntimeError(f"worker failures={len(failures)} exit_codes={bad_exit_codes}")
+            run_failed = bool(bad_exit_codes or failures)
+            if run_failed:
+                print(
+                    f"workers finished with failures={len(failures)} "
+                    f"exit_codes={bad_exit_codes}; merging partial results",
+                    flush=True,
+                )
         except BaseException:
             for process in workers:
                 if process.is_alive():
@@ -329,11 +356,21 @@ def main() -> None:
             task_queue.close()
             status_queue.close()
 
-    mean_score = write_merged_result(
+    # Persist whatever scored cleanly even when workers failed mid-run; the
+    # missing entries stay discoverable via the result file's "missing_videos".
+    mean_score, missing = write_merged_result(
         output_dir=output_dir,
         result_file=result_file,
         videos=videos,
+        require_complete=not run_failed,
     )
+    if run_failed:
+        print(
+            f"INCOMPLETE run: scored={len(videos) - len(missing)}/{len(videos)}; "
+            f"partial results: {result_file} -- re-invoke the runner to rescore missing videos",
+            flush=True,
+        )
+        sys.exit(1)
     print(f"motion_smoothness {mean_score:.12f}", flush=True)
     print(f"saved {result_file}", flush=True)