[feat]: B200 SSIM full-quality support and resource-aware local runner

Add B200 to the device-reference mappings across the SSIM suite and migrate
test_gen3c_similarity.py onto the shared reference_utils helpers (device
resolution, tiered output/reference layout, full-quality param selection,
executor shutdown in try/finally). Add run_parallel.py, a local scheduler
that mirrors the Modal static discovery and packs one-, two-, and four-GPU
SSIM jobs onto disjoint CUDA_VISIBLE_DEVICES sets, with per-task logs, JUnit
XML, and a summary.json under generated_videos/<tier>/_runs/. Verified: all
files compile and --dry-run discovers all 19 tasks with correct GPU sizing
against current main. Full-quality B200 references generated on GB200 are
uploaded to the HF reference repo separately.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.agents/skills/reseed-performance-baseline/SKILL.md

@@ -69,7 +69,7 @@ approval, then upload reviewed accepted baseline records.
 | `model_id` | Yes | Benchmark id, e.g. `wan-t2v-1.3b-2gpu`. This maps to the HF subdirectory after `sanitize(model_id)`. |
 | `gpu_type` | Yes | Exact GPU device string from the performance record, e.g. the L40S device name emitted by CI. Baselines are GPU-specific. |
 | `source_results` | Yes | One or more local paths or Buildkite artifact URLs for accepted shifted performance JSONs. Prefer normalized `normalized_perf_*.json` artifacts emitted by `compare_baseline.py`. Accept `source_result` as an alias only for a single JSON. |
-| `max_intra_batch_regression` | No | Maximum allowed regression of any source JSON against the source batch median. Default: `PERF_MAX_REGRESSION` if set, otherwise `0.05` (5%). |
+| `max_intra_batch_regression` | No | Maximum allowed regression of any source JSON against the source batch median. Default: `0.05` (5%). |
 | `intent_rationale` | Yes | One-line explanation for why the baseline shift is legitimate. This is written into provenance and should be reused in the PR. |
 
 Hardcoded defaults:
@@ -148,8 +148,7 @@ For each metric with at least two non-null source values:
 4. Stop if any source record regresses against the batch median by more than
    `max_intra_batch_regression`.
 
-Default `max_intra_batch_regression` to `PERF_MAX_REGRESSION` when set,
-otherwise `0.05`. Print a table with per-source values, batch median, and
+Default `max_intra_batch_regression` to `0.05`. Print a table with per-source values, batch median, and
 worst intra-batch regression.
 
 This check prevents uploading a mixed batch where one JSON is materially
@@ -183,7 +182,7 @@ present, that run is not a valid source for baseline reseeding.
 
 ### 2. Sync and back up existing HF records under /tmp
 
-Use `fastvideo/tests/performance/hf_store.py` helpers directly. Do **not** use
+Use `fastvideo/performance/hf_store.py` helpers directly. Do **not** use
 `compare_baseline.py` as a sync shortcut; on full main runs it can persist
 records, while this step must only fetch and back up existing history.
 
@@ -192,7 +191,7 @@ The sync command pattern is:
 ```bash
 export PERFORMANCE_TRACKING_ROOT="${PERFORMANCE_TRACKING_ROOT:-/tmp/perf-tracking}"
 export HF_REPO_ID="${HF_REPO_ID:-FastVideo/performance-tracking}"
-PYTHONPATH=fastvideo/tests/performance python -c 'from hf_store import sync_from_hf; import os; sync_from_hf(os.environ["PERFORMANCE_TRACKING_ROOT"], strict=True)'
+python -c 'from fastvideo.performance.hf_store import sync_from_hf; import os; sync_from_hf(os.environ["PERFORMANCE_TRACKING_ROOT"], strict=True)'
 ```
 
 Then back up only the sanitized model directory under `/tmp`:
@@ -200,8 +199,8 @@ Then back up only the sanitized model directory under `/tmp`:
 ```bash
 SHORT_COMMIT=$(git rev-parse --short=12 HEAD)
 TIMESTAMP=$(date -u +%Y%m%d_%H%M%S)
-MODEL_SAFE=$(PYTHONPATH=fastvideo/tests/performance python - <<'PY'
-from hf_store import sanitize
+MODEL_SAFE=$(python - <<'PY'
+from fastvideo.performance.hf_store import sanitize
 print(sanitize("<model_id>"))
 PY
 )
@@ -235,7 +234,7 @@ first baseline seed. Continue, but report that baseline history was empty.
 Load the last 5 successful records for the target:
 
 ```python
-from hf_store import load_records_for_model
+from fastvideo.performance.hf_store import load_records_for_model
 
 records = load_records_for_model(
     "/tmp/perf-tracking",
@@ -372,7 +371,7 @@ prepared records plus backup on disk.
 Use the shared storage helper so the path and repo type match CI:
 
 ```python
-from hf_store import upload_record
+from fastvideo.performance.hf_store import upload_record
 
 upload_record("<local_record_path>", record, strict=True)
 ```
@@ -460,7 +459,7 @@ directories created for this reseed. Never remove unrelated `/tmp` contents.
   intentional baseline replacement.
 - `fastvideo/tests/performance/compare_baseline.py` — normalization, rolling
   median comparison, and persistence rules.
-- `fastvideo/tests/performance/hf_store.py` — HF sync, record loading,
+- `fastvideo/performance/hf_store.py` — HF sync, record loading,
   `sanitize()`, and `upload_record()`.
 - `fastvideo/tests/performance/test_inference_performance.py` — source result
   JSON schema.

docs/contributing/performance_benchmarks.md

@@ -112,10 +112,11 @@ per-metric policy with direction, percent threshold, absolute threshold, and a
 while it runs so pipeline stage execution times are available in
 `generate_video(...).logging_info`. Stage logs use pipeline-unique keys such as
 `prompt_encoding_stage` so duplicate stage classes do not collide. For
-`PipelineStage` entries, the extractor maps the `stage_class` field:
-`TextEncodingStage` maps to `text_encoder_time_s`, `DenoisingStage` and
-`DmdDenoisingStage` map to `dit_time_s`, and `DecodingStage` maps to
-`vae_decode_time_s`, with a fallback for older logs that used the class name as
+`PipelineStage` entries, shared component stage bases emit a stable
+`component_metric`: text encoding stages map to `text_encoder_time_s`,
+denoising stages and subclasses map to `dit_time_s`, and decoding stages map to
+`vae_decode_time_s`. The extractor falls back to known `stage_class` names for
+older logs that do not include `component_metric` or that used the class name as
 the stage key. Generator-side timings such as `PostDecodeFrameProcessStage`,
 `VideoSaveStage`, and `AudioMuxStage` are intentionally ignored. If a pipeline
 does not report one of the mapped stages, that component metric is stored as
@@ -432,5 +433,5 @@ pipelines that did not report a mapped component stage.
 
 **Component timing is `null`** — the generated result did not include a mapped
 stage in `logging_info.stages`. Check that the pipeline emits stage logging
-and that the stage name is listed in `STAGE_METRIC_MAP` in
-`test_inference_performance.py`.
+and that the stage emits `component_metric` or is covered by the legacy
+`STAGE_METRIC_MAP` fallback in `test_inference_performance.py`.

fastvideo/performance/hf_store.py

@@ -310,8 +310,12 @@ def load_records_for_model(
 # ---------------------------------------------------------------------------
 
 _NUMERIC_COLS = (
-    "latency", "throughput", "memory",
-    "text_encoder_time_s", "dit_time_s", "vae_decode_time_s",
+    "latency",
+    "throughput",
+    "memory",
+    "text_encoder_time_s",
+    "dit_time_s",
+    "vae_decode_time_s",
 )
 
 

fastvideo/performance/metric_policy.py

@@ -36,6 +36,7 @@ DEFAULT_METRIC_POLICIES: tuple[MetricPolicy, ...] = (
     MetricPolicy("vae_decode_time_s", "VAE Decode", 3, True, 0.05, 0.25),
 )
 
+
 def _optional_float(value: Any) -> float | None:
     if value is None or isinstance(value, bool):
         return None
@@ -57,9 +58,7 @@ def _optional_bool(value: Any) -> bool | None:
     return None
 
 
-def resolve_metric_policies(
-    threshold_overrides: Mapping[str, Any] | None,
-) -> tuple[MetricPolicy, ...]:
+def resolve_metric_policies(threshold_overrides: Mapping[str, Any] | None, ) -> tuple[MetricPolicy, ...]:
     """Return default metric policies with optional per-metric overrides."""
 
     if not isinstance(threshold_overrides, Mapping):
@@ -80,25 +79,15 @@ def resolve_metric_policies(
                 label=base_policy.label,
                 precision=base_policy.precision,
                 lower_is_better=base_policy.lower_is_better,
-                threshold_percent=(
-                    base_policy.threshold_percent
-                    if threshold_percent is None
-                    else threshold_percent
-                ),
-                threshold_absolute=(
-                    base_policy.threshold_absolute
-                    if threshold_absolute is None
-                    else threshold_absolute
-                ),
+                threshold_percent=(base_policy.threshold_percent if threshold_percent is None else threshold_percent),
+                threshold_absolute=(base_policy.threshold_absolute
+                                    if threshold_absolute is None else threshold_absolute),
                 gated=base_policy.gated if gated is None else gated,
-            )
-        )
+            ))
     return tuple(policies)
 
 
-def serialize_metric_thresholds(
-    policies: tuple[MetricPolicy, ...],
-) -> dict[str, dict[str, float | bool]]:
+def serialize_metric_thresholds(policies: tuple[MetricPolicy, ...], ) -> dict[str, dict[str, float | bool]]:
     return {
         policy.key: {
             "threshold_percent": policy.threshold_percent,
@@ -118,10 +107,7 @@ def regression_delta(
         return None
     absolute_delta = current - baseline if policy.lower_is_better else baseline - current
     percent_delta = absolute_delta / baseline
-    threshold_exceeded = (
-        percent_delta > policy.threshold_percent
-        and absolute_delta > policy.threshold_absolute
-    )
+    threshold_exceeded = (percent_delta > policy.threshold_percent and absolute_delta > policy.threshold_absolute)
     return MetricDelta(
         absolute=absolute_delta,
         percent=percent_delta,

fastvideo/pipelines/stages/base.py

@@ -34,6 +34,7 @@ class PipelineStage(ABC):
     composed with other stages to create a complete pipeline. Each stage is responsible
     for a specific part of the process, such as prompt encoding, latent preparation, etc.
     """
+    performance_component_metric: str | None = None
 
     def verify_input(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult:
         """
@@ -155,6 +156,9 @@ class PipelineStage(ABC):
                 logger.info("[%s] Execution completed in %s ms", stage_name, execution_time * 1000)
                 batch.logging_info.add_stage_execution_time(stage_key, execution_time)
                 batch.logging_info.add_stage_metric(stage_key, "stage_class", stage_class_name)
+                component_metric = self.performance_component_metric
+                if component_metric is not None:
+                    batch.logging_info.add_stage_metric(stage_key, "component_metric", component_metric)
             except Exception as e:
                 torch.cuda.synchronize()
                 execution_time = time.perf_counter() - start_time

fastvideo/pipelines/stages/decoding.py

@@ -28,6 +28,7 @@ class DecodingStage(PipelineStage):
     This stage handles the decoding of latent representations into the final
     output format (e.g., pixel values).
     """
+    performance_component_metric = "vae_decode_time_s"
 
     def __init__(self, vae, pipeline=None) -> None:
         self.vae: ParallelTiledVAE = vae

fastvideo/pipelines/stages/denoising.py

@@ -51,6 +51,7 @@ class DenoisingStage(PipelineStage):
     This stage handles the iterative denoising process that transforms
     the initial noise into the final output.
     """
+    performance_component_metric = "dit_time_s"
 
     def __init__(self, transformer, scheduler, pipeline=None, transformer_2=None, vae=None) -> None:
         super().__init__()
@@ -1187,6 +1188,7 @@ class Cosmos25V2WDenoisingStage(Cosmos25DenoisingStage):
 
 class Cosmos25AutoDenoisingStage(PipelineStage):
     """Route Cosmos 2.5 denoising to T2W vs V2W/I2W."""
+    performance_component_metric = "dit_time_s"
 
     def __init__(self, transformer, scheduler) -> None:
         super().__init__()

fastvideo/pipelines/stages/text_encoding.py

@@ -24,6 +24,7 @@ class TextEncodingStage(PipelineStage):
     This stage handles the encoding of text prompts into the embedding space
     expected by the diffusion model.
     """
+    performance_component_metric = "text_encoder_time_s"
 
     def __init__(self, text_encoders, tokenizers) -> None:
         """
@@ -350,6 +351,7 @@ class Cosmos25TextEncodingStage(PipelineStage):
     Cosmos 2.5 uses Reason1 (Qwen2.5-VL) and relies on the encoder's
     `compute_text_embeddings_online()`.
     """
+    performance_component_metric = "text_encoder_time_s"
 
     def __init__(self, text_encoder) -> None:
         super().__init__()

fastvideo/tests/contract/test_ci_test_collection.py

@@ -0,0 +1,85 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Guard: every test directory must be collected by some CI lane or be on
+the explicit allowlist below.
+
+Three separate incidents on 2026-07-05 found test files that no CI lane
+ever collects (fastvideo/tests/stages/, tests/local_tests/ additions in
+PR #1509, and this sweep found seven dark directories in total): the tests
+pass review, merge, and then silently never run. This test makes going
+dark an explicit, reviewed decision instead of an accident: adding a new
+test directory fails CI until it is either wired into a lane or
+allowlisted here with a reason.
+
+Pure text analysis — no fastvideo imports, no GPU, no torch.
+"""
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+TESTS_ROOT = REPO_ROOT / "fastvideo" / "tests"
+
+# Files whose text constitutes "a CI lane references this directory".
+CI_SOURCES = [
+    TESTS_ROOT / "modal" / "pr_test.py",
+    TESTS_ROOT / "modal" / "ssim_test.py",
+    *sorted((REPO_ROOT / ".buildkite").rglob("*.yml")),
+    *sorted((REPO_ROOT / ".buildkite").rglob("*.sh")),
+]
+
+# Directories that intentionally have no CI lane today. Every entry needs a
+# reason; remove the entry when the directory gets wired into a lane.
+# State as found on 2026-07-05 — these SHOULD shrink over time, not grow.
+ALLOWLIST = {
+    "attention": "no lane yet — GPU attention-backend tests, run manually",
+    "audio": "no lane yet — audio encoder tests, run manually",
+    "distributed": "no lane yet — multi-GPU torchrun tests, run manually",
+    "hooks": "no lane yet — run manually",
+    "layers": "no lane yet — torchrun FSDP dispatch tests, run manually",
+    "nightly": "by design: nightly cadence, not per-PR",
+    "ops": "no lane yet — GPU op tests, run manually",
+    "modal": "CI infrastructure itself, not a test suite",
+}
+
+
+def _dirs_with_tests() -> list[str]:
+    dirs = []
+    for child in sorted(TESTS_ROOT.iterdir()):
+        if child.is_dir() and any(child.rglob("test_*.py")):
+            dirs.append(child.name)
+    return dirs
+
+
+def _ci_text() -> str:
+    return "\n".join(
+        src.read_text(errors="replace") for src in CI_SOURCES if src.exists())
+
+
+def test_every_test_directory_is_collected_or_allowlisted():
+    ci_text = _ci_text()
+    dark = [
+        name for name in _dirs_with_tests()
+        if f"tests/{name}" not in ci_text and name not in ALLOWLIST
+    ]
+    assert not dark, (
+        f"Test directories not referenced by any CI lane and not "
+        f"allowlisted: {dark}. Wire them into a lane in "
+        f"fastvideo/tests/modal/pr_test.py (or a Buildkite step), or add an "
+        f"allowlist entry with a reason in {__file__}.")
+
+
+def test_local_tests_stays_out_of_ci():
+    # tests/local_tests/ (repo root) is developer-local by design (author
+    # decision, 2026-07-05): parity scaffolds and machine-specific checks
+    # that must never gate CI. Fail if any CI source starts collecting it.
+    assert "tests/local_tests" not in _ci_text(), (
+        "tests/local_tests/ is local-only by design; remove the CI "
+        "reference or move the tests into a fastvideo/tests/ lane.")
+
+
+def test_allowlist_entries_are_still_real_directories():
+    # A stale allowlist hides regressions; entries must track reality.
+    missing = [
+        name for name in ALLOWLIST
+        if name != "modal" and not (TESTS_ROOT / name).is_dir()
+    ]
+    assert not missing, (
+        f"Allowlisted directories no longer exist — remove them: {missing}")

fastvideo/tests/performance/test_inference_performance.py

@@ -179,7 +179,11 @@ def _extract_component_times(result: dict) -> dict[str, float | None]:
             logger.debug("Skipping malformed stage '%s' data: %r", stage_name, stage_data)
             continue
         stage_class = stage_data.get("stage_class", stage_name)
-        metric_key = STAGE_METRIC_MAP.get(stage_class)
+        component_metric = stage_data.get("component_metric")
+        if isinstance(component_metric, str) and component_metric in component_times:
+            metric_key = component_metric
+        else:
+            metric_key = STAGE_METRIC_MAP.get(stage_class)
         if metric_key is None:
             logger.debug("Unmapped stage '%s' class '%s' (%.3fs)",
                          stage_name,

fastvideo/tests/performance/test_inference_performance_component_times.py

@@ -1,13 +1,20 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from fastvideo.pipelines.pipeline_batch_info import PipelineLoggingInfo
+from fastvideo.pipelines.stages.denoising import Cosmos25AutoDenoisingStage, DenoisingStage
+from fastvideo.pipelines.stages.text_encoding import Cosmos25TextEncodingStage
 from fastvideo.tests.performance.test_inference_performance import _extract_component_times
 
 
+class SubclassStyleDenoisingStage(DenoisingStage):
+    pass
+
+
 def test_extract_component_times_handles_pipeline_logging_info_object():
     logging_info = PipelineLoggingInfo()
     logging_info.add_stage_execution_time("prompt_encoding_stage", 1.25)
     logging_info.add_stage_metric("prompt_encoding_stage", "stage_class", "TextEncodingStage")
+    logging_info.add_stage_metric("prompt_encoding_stage", "component_metric", "text_encoder_time_s")
 
     assert _extract_component_times({"logging_info": logging_info}) == {
         "text_encoder_time_s": 1.25,
@@ -45,6 +52,35 @@ def test_extract_component_times_uses_stage_class_for_pipeline_stage_keys():
     }
 
 
+def test_denoising_stage_subclasses_inherit_component_metric():
+    assert SubclassStyleDenoisingStage.performance_component_metric == "dit_time_s"
+
+
+def test_cosmos25_direct_pipeline_stages_define_component_metrics():
+    assert Cosmos25TextEncodingStage.performance_component_metric == "text_encoder_time_s"
+    assert Cosmos25AutoDenoisingStage.performance_component_metric == "dit_time_s"
+
+
+def test_extract_component_times_uses_component_metric_for_stage_subclasses():
+    result = {
+        "logging_info": {
+            "stages": {
+                "denoising_stage": {
+                    "execution_time": 4.2,
+                    "stage_class": "CosmosDenoisingStage",
+                    "component_metric": "dit_time_s",
+                },
+            },
+        },
+    }
+
+    assert _extract_component_times(result) == {
+        "text_encoder_time_s": None,
+        "dit_time_s": 4.2,
+        "vae_decode_time_s": None,
+    }
+
+
 def test_extract_component_times_keeps_legacy_class_name_keys():
     # Backward-compatibility check for logs produced before pipeline-unique
     # stage keys carried a separate stage_class field.

fastvideo/tests/ssim/README.md

@@ -34,6 +34,28 @@ run `pytest fastvideo/tests/ssim/ -vs --ssim-full-quality` to use the
 `*_FULL_QUALITY_PARAMS` configs (default run keeps the original shortened test
 configs).
 
+For a local machine with mixed one-, two-, and four-GPU SSIM jobs, use the
+resource-aware runner instead of plain pytest:
+
+```bash
+.venv/bin/python fastvideo/tests/ssim/run_parallel.py \
+  --ssim-full-quality \
+  --hf-home /mnt/.cache \
+  --gpu-ids 0,1,2,3
+```
+
+The runner assigns disjoint `CUDA_VISIBLE_DEVICES` sets, keeps scheduling tasks
+that fit the currently free GPUs, and continues after individual failures by
+default. Per-task logs, JUnit XML, and `summary.json` are written under
+`generated_videos/<quality-tier>/_runs/<timestamp>/`. Use `--dry-run` to inspect
+the task/GPU plan, `--fail-fast` to stop after the first failure, or repeat
+`--test-file` / `--model-id` to select a subset. Worker core dumps are disabled
+by default; pass `--allow-core-dumps` only when a native crash needs debugging.
+
+Full-quality references are device-specific. A missing reference can therefore
+make pytest fail after generation even though the generated video or latent was
+successfully preserved under `generated_videos/full_quality/`.
+
 to override the HF reference repo at test time:
 `pytest fastvideo/tests/ssim/ -vs --ssim-reference-repo <org/repo>`
 

fastvideo/tests/ssim/run_parallel.py

@@ -0,0 +1,802 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Run SSIM tests locally with exclusive, mixed-size GPU allocations.
+
+The SSIM suite contains one-, two-, and four-GPU tests. Plain pytest runs
+them serially, while pytest-xdist cannot safely infer how many GPUs each test
+needs. This runner mirrors the Modal scheduler's static discovery and assigns
+disjoint ``CUDA_VISIBLE_DEVICES`` sets to concurrent pytest subprocesses.
+
+Example:
+
+.. code-block:: bash
+
+    .venv/bin/python fastvideo/tests/ssim/run_parallel.py \
+        --ssim-full-quality \
+        --hf-home /mnt/.cache \
+        --gpu-ids 0,1,2,3
+
+Non-zero pytest exits do not stop other tasks by default. This is important
+for visual-generation audits where a missing device-specific reference may
+fail after the generated artifact has already been saved.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import json
+import os
+import re
+import resource
+import signal
+import shutil
+import subprocess
+import sys
+import time
+import xml.etree.ElementTree as ET
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+SSIM_DIR = Path(__file__).resolve().parent
+DEFAULT_HF_HOME = Path("/mnt/.cache")
+DEFAULT_ALLOC_CONF = "expandable_segments:False"
+HF_TOKEN_ENV_KEYS = ("HF_API_KEY", "HUGGINGFACE_HUB_TOKEN", "HF_TOKEN")
+HANDLED_SIGNALS = (signal.SIGHUP, signal.SIGINT, signal.SIGTERM)
+
+
+def _disable_core_dumps() -> None:
+    """Prevent routine worker failures from writing multi-gigabyte core files."""
+    resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
+
+
+def _raise_keyboard_interrupt(_signum: int, _frame: Any) -> None:
+    """Route termination signals through the normal worker cleanup path."""
+    raise KeyboardInterrupt
+
+
+def _unblock_task_signals() -> None:
+    """Reset the launcher's short-lived blocked mask before child exec."""
+    signal.pthread_sigmask(signal.SIG_UNBLOCK, HANDLED_SIGNALS)
+
+
+@dataclass(frozen=True)
+class SSIMTask:
+    task_id: int
+    test_file: Path
+    required_gpus: int
+    model_id: str | None = None
+    test_function: str | None = None
+
+    @property
+    def node_id(self) -> str:
+        relative_file = self.test_file.relative_to(REPO_ROOT).as_posix()
+        if self.test_function is None:
+            return relative_file
+        return f"{relative_file}::{self.test_function}"
+
+    @property
+    def display_name(self) -> str:
+        parts = [self.test_file.name]
+        if self.test_function is not None:
+            parts.append(self.test_function)
+        if self.model_id is not None:
+            parts.append(self.model_id)
+        return "::".join(parts)
+
+    @property
+    def sort_key(self) -> tuple[str, str, str]:
+        return (
+            self.test_file.name,
+            self.test_function or "",
+            self.model_id or "",
+        )
+
+
+@dataclass
+class RunningTask:
+    task: SSIMTask
+    process: subprocess.Popen[str]
+    gpu_ids: list[str]
+    log_path: Path
+    junit_path: Path
+    log_handle: Any
+    started_at: float
+    command: list[str]
+
+
+@dataclass
+class TaskResult:
+    task_id: int
+    test_name: str
+    node_id: str
+    model_id: str | None
+    required_gpus: int
+    gpu_ids: list[str]
+    status: str
+    returncode: int | None
+    duration_seconds: float
+    tests: int
+    passed: int
+    failures: int
+    errors: int
+    skipped: int
+    log_path: str
+    junit_path: str
+    command: list[str]
+
+
+def _assignment_names(node: ast.Assign | ast.AnnAssign) -> list[str]:
+    if isinstance(node, ast.AnnAssign):
+        if isinstance(node.target, ast.Name):
+            return [node.target.id]
+        return []
+    return [target.id for target in node.targets if isinstance(target, ast.Name)]
+
+
+def _extract_required_gpus(module_ast: ast.Module, filepath: Path) -> int:
+    for node in module_ast.body:
+        if not isinstance(node, (ast.Assign, ast.AnnAssign)):
+            continue
+        if "REQUIRED_GPUS" not in _assignment_names(node):
+            continue
+        if isinstance(node.value, ast.Constant) and isinstance(node.value.value, int):
+            required_gpus = node.value.value
+            if required_gpus < 1:
+                raise ValueError(f"{filepath}: REQUIRED_GPUS must be positive")
+            return required_gpus
+        raise ValueError(f"{filepath}: REQUIRED_GPUS must be an integer literal")
+    return 1
+
+
+def _extract_string_constants(module_ast: ast.Module) -> dict[str, str]:
+    constants: dict[str, str] = {}
+    for node in module_ast.body:
+        if not isinstance(node, (ast.Assign, ast.AnnAssign)):
+            continue
+        if not isinstance(node.value, ast.Constant) or not isinstance(node.value.value, str):
+            continue
+        for name in _assignment_names(node):
+            constants[name] = node.value.value
+    return constants
+
+
+def _extract_model_ids(module_ast: ast.Module) -> list[str]:
+    """Extract literal model IDs from all ``*_MODEL_TO_PARAMS`` maps.
+
+    Both default and full-quality maps are inspected and de-duplicated. Named
+    string constants are supported as dictionary keys (for example SD3.5's
+    ``MODEL_ID``), fixing a limitation in the current Modal discovery helper.
+    """
+    string_constants = _extract_string_constants(module_ast)
+    model_ids: list[str] = []
+    seen: set[str] = set()
+    for node in module_ast.body:
+        if not isinstance(node, (ast.Assign, ast.AnnAssign)):
+            continue
+        if not any(name.endswith("MODEL_TO_PARAMS") for name in _assignment_names(node)):
+            continue
+        if not isinstance(node.value, ast.Dict):
+            continue
+        for key in node.value.keys:
+            model_id: str | None = None
+            if isinstance(key, ast.Constant) and isinstance(key.value, str):
+                model_id = key.value
+            elif isinstance(key, ast.Name):
+                model_id = string_constants.get(key.id)
+            if model_id is not None and model_id not in seen:
+                seen.add(model_id)
+                model_ids.append(model_id)
+    return model_ids
+
+
+def _extract_test_functions(module_ast: ast.Module) -> list[str]:
+    return [
+        node.name for node in module_ast.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test_")
+    ]
+
+
+def _normalize_selected_files(values: list[str]) -> set[str]:
+    selected: set[str] = set()
+    for value in values:
+        for part in value.split(","):
+            cleaned = part.strip()
+            if cleaned:
+                selected.add(Path(cleaned).name)
+    return selected
+
+
+def _normalize_selected_models(values: list[str]) -> set[str]:
+    selected: set[str] = set()
+    for value in values:
+        selected.update(part.strip() for part in value.split(",") if part.strip())
+    return selected
+
+
+def discover_tasks(
+    *,
+    selected_test_files: set[str] | None = None,
+    selected_model_ids: set[str] | None = None,
+) -> list[SSIMTask]:
+    selected_test_files = selected_test_files or set()
+    selected_model_ids = selected_model_ids or set()
+    matched_files: set[str] = set()
+    matched_models: set[str] = set()
+    task_specs: list[tuple[Path, int, str | None, str | None]] = []
+
+    for filepath in sorted(SSIM_DIR.glob("test_*_similarity.py")):
+        if selected_test_files and filepath.name not in selected_test_files:
+            continue
+        matched_files.add(filepath.name)
+        module_ast = ast.parse(filepath.read_text(encoding="utf-8"), filename=str(filepath))
+        required_gpus = _extract_required_gpus(module_ast, filepath)
+        model_ids = _extract_model_ids(module_ast)
+
+        if model_ids:
+            for model_id in model_ids:
+                if selected_model_ids and model_id not in selected_model_ids:
+                    continue
+                matched_models.add(model_id)
+                task_specs.append((filepath, required_gpus, model_id, None))
+            continue
+
+        if selected_model_ids:
+            continue
+        test_functions = _extract_test_functions(module_ast)
+        if not test_functions:
+            task_specs.append((filepath, required_gpus, None, None))
+        else:
+            # Function-level splitting gives independent LongCat T2V/I2V/VC
+            # tasks, keeping three one-GPU workers busy instead of serializing
+            # the whole file on one GPU.
+            for test_function in test_functions:
+                task_specs.append((filepath, required_gpus, None, test_function))
+
+    missing_files = sorted(selected_test_files - matched_files)
+    if missing_files:
+        raise ValueError(f"Requested SSIM test file(s) not found: {', '.join(missing_files)}")
+    missing_models = sorted(selected_model_ids - matched_models)
+    if missing_models:
+        raise ValueError(f"Requested SSIM model ID(s) not found: {', '.join(missing_models)}")
+
+    task_specs.sort(key=lambda spec: (spec[0].name, spec[3] or "", spec[2] or ""))
+    return [
+        SSIMTask(
+            task_id=index,
+            test_file=filepath,
+            required_gpus=required_gpus,
+            model_id=model_id,
+            test_function=test_function,
+        )
+        for index, (filepath, required_gpus, model_id, test_function) in enumerate(task_specs)
+    ]
+
+
+def _safe_name(value: str) -> str:
+    return re.sub(r"[^A-Za-z0-9_.-]+", "_", value).strip("_")
+
+
+def _parse_gpu_ids(value: str) -> list[str]:
+    gpu_ids: list[str] = []
+    seen: set[str] = set()
+    for item in value.split(","):
+        gpu_id = item.strip()
+        if not gpu_id or gpu_id in seen:
+            continue
+        seen.add(gpu_id)
+        gpu_ids.append(gpu_id)
+    if not gpu_ids:
+        raise ValueError("--gpu-ids must contain at least one GPU ID")
+    return gpu_ids
+
+
+def _preflight_gpus(gpu_ids: list[str]) -> list[str]:
+    probe = (
+        "import json, torch\n"
+        "names=[]\n"
+        "for i in range(torch.cuda.device_count()):\n"
+        "    names.append(torch.cuda.get_device_name(i))\n"
+        "    torch.empty(1, device=f'cuda:{i}')\n"
+        "torch.cuda.synchronize() if names else None\n"
+        "print(json.dumps({'count': torch.cuda.device_count(), 'names': names}))\n"
+    )
+    env = os.environ.copy()
+    env["CUDA_VISIBLE_DEVICES"] = ",".join(gpu_ids)
+    result = subprocess.run(
+        [sys.executable, "-c", probe],
+        cwd=REPO_ROOT,
+        env=env,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        raise RuntimeError(f"CUDA preflight failed: {detail or f'exit code {result.returncode}'}")
+    try:
+        payload = json.loads(result.stdout.strip().splitlines()[-1])
+    except (IndexError, json.JSONDecodeError) as error:
+        raise RuntimeError(f"CUDA preflight returned invalid output: {result.stdout!r}") from error
+    if payload.get("count") != len(gpu_ids):
+        recovery_hint = ""
+        if payload.get("count", 0) == 0:
+            recovery_hint = (
+                " No pytest task was launched. Verify `nvidia-smi` and recreate "
+                "the container/workspace with NVIDIA GPU device grants if it "
+                "reports `Failed to initialize NVML`."
+            )
+        raise RuntimeError(
+            f"Requested GPU IDs {gpu_ids}, but PyTorch can initialize only "
+            f"{payload.get('count', 0)} device(s). stderr: {result.stderr.strip()}"
+            f"{recovery_hint}"
+        )
+    return [str(name) for name in payload.get("names", [])]
+
+
+def _read_junit_counts(junit_path: Path) -> dict[str, int]:
+    counts = {"tests": 0, "failures": 0, "errors": 0, "skipped": 0, "passed": 0}
+    if not junit_path.exists():
+        return counts
+    try:
+        root = ET.parse(junit_path).getroot()
+    except (ET.ParseError, OSError):
+        return counts
+
+    suites = [root] if root.tag == "testsuite" else list(root.findall("testsuite"))
+    for suite in suites:
+        for key in ("tests", "failures", "errors", "skipped"):
+            counts[key] += int(suite.attrib.get(key, 0))
+    counts["passed"] = max(0, counts["tests"] - counts["failures"] - counts["errors"] - counts["skipped"])
+    return counts
+
+
+def _classify_status(returncode: int, counts: dict[str, int]) -> str:
+    if returncode != 0:
+        return "failed"
+    if counts["tests"] > 0 and counts["skipped"] == counts["tests"]:
+        return "skipped"
+    return "passed"
+
+
+def _build_command(
+    task: SSIMTask,
+    *,
+    junit_path: Path,
+    full_quality: bool,
+    skip_reference_download: bool,
+    pytest_args: list[str],
+) -> list[str]:
+    command = [
+        sys.executable,
+        "-m",
+        "pytest",
+        task.node_id,
+        "-vs",
+        f"--junitxml={junit_path}",
+        "-o",
+        "junit_family=xunit2",
+    ]
+    if full_quality:
+        command.append("--ssim-full-quality")
+    if skip_reference_download:
+        command.append("--skip-ssim-reference-download")
+    command.extend(pytest_args)
+    return command
+
+
+def _launch_task(
+    task: SSIMTask,
+    *,
+    gpu_ids: list[str],
+    run_dir: Path,
+    hf_home: Path,
+    full_quality: bool,
+    skip_reference_download: bool,
+    pytest_args: list[str],
+) -> RunningTask:
+    safe_name = _safe_name(f"{task.task_id:03d}_{task.display_name}")
+    log_path = run_dir / "logs" / f"{safe_name}.log"
+    junit_path = run_dir / "junit" / f"{safe_name}.xml"
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    junit_path.parent.mkdir(parents=True, exist_ok=True)
+    command = _build_command(
+        task,
+        junit_path=junit_path,
+        full_quality=full_quality,
+        skip_reference_download=skip_reference_download,
+        pytest_args=pytest_args,
+    )
+    env = os.environ.copy()
+    env.update({
+        "CUDA_VISIBLE_DEVICES": ",".join(gpu_ids),
+        "HF_HOME": str(hf_home),
+        "PYTHONUNBUFFERED": "1",
+        "PYTORCH_CUDA_ALLOC_CONF": env.get("PYTORCH_CUDA_ALLOC_CONF", DEFAULT_ALLOC_CONF),
+        "TOKENIZERS_PARALLELISM": "false",
+    })
+    if full_quality:
+        env["FASTVIDEO_SSIM_FULL_QUALITY"] = "1"
+    else:
+        env.pop("FASTVIDEO_SSIM_FULL_QUALITY", None)
+    if task.model_id is None:
+        env.pop("FASTVIDEO_SSIM_MODEL_ID", None)
+    else:
+        env["FASTVIDEO_SSIM_MODEL_ID"] = task.model_id
+
+    log_handle = log_path.open("w", encoding="utf-8")
+    try:
+        process = subprocess.Popen(
+            command,
+            cwd=REPO_ROOT,
+            env=env,
+            stdout=log_handle,
+            stderr=subprocess.STDOUT,
+            text=True,
+            preexec_fn=_unblock_task_signals,
+            start_new_session=True,
+        )
+    except BaseException:
+        log_handle.close()
+        raise
+    return RunningTask(
+        task=task,
+        process=process,
+        gpu_ids=gpu_ids,
+        log_path=log_path,
+        junit_path=junit_path,
+        log_handle=log_handle,
+        started_at=time.monotonic(),
+        command=command,
+    )
+
+
+def _terminate_running(running: list[RunningTask], timeout_seconds: float = 30.0) -> None:
+    for item in running:
+        if item.process.poll() is None:
+            try:
+                os.killpg(item.process.pid, signal.SIGTERM)
+            except ProcessLookupError:
+                pass
+    deadline = time.monotonic() + timeout_seconds
+    while time.monotonic() < deadline and any(item.process.poll() is None for item in running):
+        time.sleep(0.25)
+    for item in running:
+        if item.process.poll() is None:
+            try:
+                os.killpg(item.process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+
+def _finalize_task(item: RunningTask, returncode: int) -> TaskResult:
+    item.log_handle.close()
+    counts = _read_junit_counts(item.junit_path)
+    return TaskResult(
+        task_id=item.task.task_id,
+        test_name=item.task.display_name,
+        node_id=item.task.node_id,
+        model_id=item.task.model_id,
+        required_gpus=item.task.required_gpus,
+        gpu_ids=item.gpu_ids,
+        status=_classify_status(returncode, counts),
+        returncode=returncode,
+        duration_seconds=round(time.monotonic() - item.started_at, 3),
+        tests=counts["tests"],
+        passed=counts["passed"],
+        failures=counts["failures"],
+        errors=counts["errors"],
+        skipped=counts["skipped"],
+        log_path=str(item.log_path),
+        junit_path=str(item.junit_path),
+        command=item.command,
+    )
+
+
+def run_tasks(
+    tasks: list[SSIMTask],
+    *,
+    gpu_ids: list[str],
+    run_dir: Path,
+    hf_home: Path,
+    full_quality: bool,
+    skip_reference_download: bool,
+    pytest_args: list[str],
+    fail_fast: bool,
+) -> tuple[list[TaskResult], bool]:
+    pending = list(tasks)
+    running: list[RunningTask] = []
+    results: list[TaskResult] = []
+    free_gpu_ids = list(gpu_ids)
+    gpu_order = {gpu_id: index for index, gpu_id in enumerate(gpu_ids)}
+    interrupted = False
+    previous_handlers = {
+        signum: signal.getsignal(signum) for signum in HANDLED_SIGNALS
+    }
+    for signum in HANDLED_SIGNALS:
+        signal.signal(signum, _raise_keyboard_interrupt)
+
+    try:
+        while pending or running:
+            while pending:
+                next_index = next(
+                    (index for index, task in enumerate(pending) if task.required_gpus <= len(free_gpu_ids)),
+                    None,
+                )
+                if next_index is None:
+                    break
+                task = pending.pop(next_index)
+                assigned = free_gpu_ids[:task.required_gpus]
+                del free_gpu_ids[:task.required_gpus]
+                previous_mask = signal.pthread_sigmask(
+                    signal.SIG_BLOCK,
+                    HANDLED_SIGNALS,
+                )
+                try:
+                    item = _launch_task(
+                        task,
+                        gpu_ids=assigned,
+                        run_dir=run_dir,
+                        hf_home=hf_home,
+                        full_quality=full_quality,
+                        skip_reference_download=skip_reference_download,
+                        pytest_args=pytest_args,
+                    )
+                    running.append(item)
+                finally:
+                    signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+                print(
+                    f"START [{task.task_id:03d}] {task.display_name} "
+                    f"on GPU(s) {','.join(assigned)} -> {item.log_path}",
+                    flush=True,
+                )
+
+            completed = [item for item in running if item.process.poll() is not None]
+            if not completed:
+                time.sleep(0.5)
+                continue
+
+            stop_after_failure = False
+            for item in completed:
+                returncode = item.process.wait()
+                result = _finalize_task(item, returncode)
+                results.append(result)
+                running.remove(item)
+                free_gpu_ids.extend(item.gpu_ids)
+                free_gpu_ids.sort(key=lambda gpu_id: gpu_order[gpu_id])
+                print(
+                    f"DONE  [{item.task.task_id:03d}] {item.task.display_name}: "
+                    f"{result.status} rc={returncode} in {result.duration_seconds:.1f}s",
+                    flush=True,
+                )
+                stop_after_failure = stop_after_failure or result.status == "failed"
+
+            if stop_after_failure and fail_fast:
+                print("Fail-fast enabled; terminating active tasks.", flush=True)
+                _terminate_running(running)
+                break
+    except KeyboardInterrupt:
+        interrupted = True
+        print("Interrupted; terminating active SSIM tasks.", flush=True)
+        for signum in HANDLED_SIGNALS:
+            signal.signal(signum, signal.SIG_IGN)
+        _terminate_running(running)
+    except BaseException:
+        print("Runner error; terminating active SSIM tasks.", flush=True)
+        for signum in HANDLED_SIGNALS:
+            signal.signal(signum, signal.SIG_IGN)
+        _terminate_running(running)
+        raise
+    finally:
+        for signum, previous_handler in previous_handlers.items():
+            signal.signal(signum, previous_handler)
+        for item in list(running):
+            returncode = item.process.wait()
+            results.append(_finalize_task(item, returncode))
+        for task in pending:
+            results.append(
+                TaskResult(
+                    task_id=task.task_id,
+                    test_name=task.display_name,
+                    node_id=task.node_id,
+                    model_id=task.model_id,
+                    required_gpus=task.required_gpus,
+                    gpu_ids=[],
+                    status="not_run",
+                    returncode=None,
+                    duration_seconds=0.0,
+                    tests=0,
+                    passed=0,
+                    failures=0,
+                    errors=0,
+                    skipped=0,
+                    log_path="",
+                    junit_path="",
+                    command=[],
+                )
+            )
+    results.sort(key=lambda result: result.task_id)
+    return results, interrupted
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--gpu-ids",
+        default=os.environ.get("CUDA_VISIBLE_DEVICES", "0,1,2,3"),
+        help="Comma-separated physical GPU IDs or UUIDs (default: 0,1,2,3).",
+    )
+    parser.add_argument(
+        "--hf-home",
+        type=Path,
+        default=DEFAULT_HF_HOME,
+        help=f"Shared Hugging Face cache root (default: {DEFAULT_HF_HOME}).",
+    )
+    parser.add_argument("--ssim-full-quality", action="store_true", help="Select full-quality SSIM parameters.")
+    parser.add_argument(
+        "--skip-reference-download",
+        action="store_true",
+        help="Pass --skip-ssim-reference-download to every pytest task.",
+    )
+    parser.add_argument("--fail-fast", action="store_true", help="Stop after the first failed task.")
+    parser.add_argument(
+        "--allow-core-dumps",
+        action="store_true",
+        help="Keep the shell's core-dump limit instead of disabling worker core dumps.",
+    )
+    parser.add_argument("--test-file", action="append", default=[], help="Run only this test file (repeatable/CSV).")
+    parser.add_argument("--model-id", action="append", default=[], help="Run only this model ID (repeatable/CSV).")
+    parser.add_argument("--pytest-arg", action="append", default=[], help="Additional argument passed to pytest.")
+    parser.add_argument("--run-dir", type=Path, help="Directory for task logs, JUnit XML, and summary.json.")
+    parser.add_argument(
+        "--keep-existing-generated",
+        action="store_true",
+        help="Do not archive existing generated artifacts before the run.",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Print discovery/allocation requirements without probing GPUs or launching pytest.",
+    )
+    return parser
+
+
+def _default_run_dir(full_quality: bool) -> Path:
+    tier = "full_quality" if full_quality else "default"
+    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    return SSIM_DIR / "generated_videos" / tier / "_runs" / timestamp
+
+
+def _archive_existing_generated(full_quality: bool, run_dir: Path) -> list[Path]:
+    """Move stale generated artifacts aside before starting a new run.
+
+    Video output helpers add numeric suffixes when a target already exists,
+    while several tests later inspect the unsuffixed path. Archiving avoids a
+    stale artifact being mistaken for the result of the current run.
+    """
+    tier = "full_quality" if full_quality else "default"
+    tier_root = SSIM_DIR / "generated_videos" / tier
+    if not tier_root.exists():
+        return []
+
+    archive_root = run_dir / "previous_generated"
+    archived: list[Path] = []
+    for path in sorted(tier_root.iterdir()):
+        if path.name == "_runs":
+            continue
+        destination = archive_root / path.name
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        shutil.move(str(path), str(destination))
+        archived.append(destination)
+    return archived
+
+
+def _write_summary(
+    path: Path,
+    *,
+    started_at: str,
+    finished_at: str,
+    hf_home: Path,
+    gpu_ids: list[str],
+    gpu_names: list[str],
+    full_quality: bool,
+    interrupted: bool,
+    results: list[TaskResult],
+) -> None:
+    payload = {
+        "started_at": started_at,
+        "finished_at": finished_at,
+        "hf_home": str(hf_home),
+        "gpu_ids": gpu_ids,
+        "gpu_names": gpu_names,
+        "full_quality": full_quality,
+        "interrupted": interrupted,
+        "results": [asdict(result) for result in results],
+    }
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _build_parser().parse_args(argv)
+    if not args.allow_core_dumps:
+        _disable_core_dumps()
+    gpu_ids = _parse_gpu_ids(args.gpu_ids)
+    selected_files = _normalize_selected_files(args.test_file)
+    selected_models = _normalize_selected_models(args.model_id)
+    tasks = discover_tasks(
+        selected_test_files=selected_files,
+        selected_model_ids=selected_models,
+    )
+    if not tasks:
+        raise RuntimeError("No SSIM tasks matched the requested filters")
+    max_required = max(task.required_gpus for task in tasks)
+    if max_required > len(gpu_ids):
+        raise RuntimeError(
+            f"A selected task requires {max_required} GPUs, but --gpu-ids provides only {len(gpu_ids)}"
+        )
+
+    print(f"Discovered {len(tasks)} SSIM task(s) across {len(gpu_ids)} GPU slot(s):", flush=True)
+    for task in tasks:
+        print(f"  [{task.task_id:03d}] {task.required_gpus} GPU(s)  {task.display_name}", flush=True)
+    if args.dry_run:
+        return 0
+
+    hf_home = args.hf_home.expanduser().resolve()
+    hf_home.mkdir(parents=True, exist_ok=True)
+    if not any(os.environ.get(key, "").strip() for key in HF_TOKEN_ENV_KEYS):
+        print(
+            "WARNING: no Hugging Face token is set; gated model tasks may fail.",
+            flush=True,
+        )
+    run_dir = (args.run_dir or _default_run_dir(args.ssim_full_quality)).expanduser().resolve()
+    gpu_names = _preflight_gpus(gpu_ids)
+    if not args.keep_existing_generated:
+        archived = _archive_existing_generated(args.ssim_full_quality, run_dir)
+        if archived:
+            print(f"Archived {len(archived)} existing generated path(s) under {run_dir}", flush=True)
+    run_dir.mkdir(parents=True, exist_ok=True)
+    print(f"CUDA preflight passed: {dict(zip(gpu_ids, gpu_names, strict=True))}", flush=True)
+    print(f"HF_HOME={hf_home}", flush=True)
+    print(f"Run artifacts: {run_dir}", flush=True)
+
+    started_at = datetime.now(timezone.utc).isoformat()
+    results, interrupted = run_tasks(
+        tasks,
+        gpu_ids=gpu_ids,
+        run_dir=run_dir,
+        hf_home=hf_home,
+        full_quality=args.ssim_full_quality,
+        skip_reference_download=args.skip_reference_download,
+        pytest_args=args.pytest_arg,
+        fail_fast=args.fail_fast,
+    )
+    finished_at = datetime.now(timezone.utc).isoformat()
+    summary_path = run_dir / "summary.json"
+    _write_summary(
+        summary_path,
+        started_at=started_at,
+        finished_at=finished_at,
+        hf_home=hf_home,
+        gpu_ids=gpu_ids,
+        gpu_names=gpu_names,
+        full_quality=args.ssim_full_quality,
+        interrupted=interrupted,
+        results=results,
+    )
+
+    status_counts: dict[str, int] = {}
+    for result in results:
+        status_counts[result.status] = status_counts.get(result.status, 0) + 1
+    print(f"Summary: {status_counts}", flush=True)
+    print(f"Detailed results: {summary_path}", flush=True)
+    if interrupted:
+        return 130
+    return 1 if any(result.status in {"failed", "not_run"} for result in results) else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

fastvideo/tests/ssim/test_gamecraft_similarity.py

@@ -47,6 +47,7 @@ device_reference_folder = resolve_device_reference_folder(
         ("A100", "A100"),
         ("H100", "H100"),
         ("H200", "H200"),
+        ("B200", "B200"),
     ),
     device_name=get_cuda_device_name(),
     unknown_device_prefix="Unknown",

fastvideo/tests/ssim/test_gen3c_similarity.py

@@ -20,10 +20,16 @@ import glob
 from pathlib import Path
 
 import pytest
-import torch
 
 from fastvideo import VideoGenerator
 from fastvideo.logger import init_logger
+from fastvideo.tests.ssim.reference_utils import (
+    build_generated_output_dir,
+    build_reference_folder_path,
+    get_cuda_device_name,
+    resolve_device_reference_folder,
+    select_ssim_params,
+)
 from fastvideo.tests.utils import compute_video_ssim_torchvision, write_ssim_results
 from fastvideo.worker.multiproc_executor import MultiprocExecutor
 
@@ -50,16 +56,16 @@ def _resolve_gen3c_test_image_path() -> str:
 # Device detection
 # ---------------------------------------------------------------------------
 
-device_name = torch.cuda.get_device_name() if torch.cuda.is_available() else "cpu"
-device_reference_folder_suffix = "_reference_videos"
-
-if "A40" in device_name:
-    device_reference_folder = "A40" + device_reference_folder_suffix
-elif "L40S" in device_name:
-    device_reference_folder = "L40S" + device_reference_folder_suffix
-else:
-    device_reference_folder = None
-    logger.warning(f"Unsupported device for GEN3C SSIM tests: {device_name}")
+device_name = get_cuda_device_name()
+device_reference_folder = resolve_device_reference_folder(
+    (
+        ("A40", "A40"),
+        ("L40S", "L40S"),
+        ("B200", "B200"),
+    ),
+    device_name=device_name,
+    logger=logger,
+)
 
 # ---------------------------------------------------------------------------
 # GEN3C generation parameters
@@ -86,9 +92,16 @@ GEN3C_T2V_PARAMS = {
 MODEL_TO_PARAMS = {
     "GEN3C-Cosmos-7B": GEN3C_T2V_PARAMS,
 }
+FULL_QUALITY_MODEL_TO_PARAMS = {
+    # GEN3C has one production recipe today. Keeping an explicit map makes the
+    # full-quality tier selection and reference/output layout consistent with
+    # the rest of the SSIM suite without claiming a different recipe.
+    "GEN3C-Cosmos-7B": dict(GEN3C_T2V_PARAMS),
+}
 
 TEST_PROMPTS = [
-    "A camera slowly orbits around a young woman sitting at a table with a coffee mug with coffee in it in front of her, "
+    "A camera slowly orbits around a young woman sitting at a table with a "
+    "coffee mug with coffee in it in front of her, "
     "looking away naturally. Soft indoor lighting, cinematic framing, shallow depth of field, smooth camera motion.",
 ]
 
@@ -115,12 +128,17 @@ def test_gen3c_inference_similarity(prompt, ATTENTION_BACKEND, model_id):
     os.environ["FASTVIDEO_ATTENTION_BACKEND"] = ATTENTION_BACKEND
 
     script_dir = os.path.dirname(os.path.abspath(__file__))
-    base_output_dir = os.path.join(script_dir, "generated_videos", model_id)
-    output_dir = os.path.join(base_output_dir, ATTENTION_BACKEND)
+    output_dir = build_generated_output_dir(
+        script_dir,
+        device_reference_folder,
+        model_id,
+        ATTENTION_BACKEND,
+    )
     output_video_name = CANDIDATE_VIDEO_NAME
     os.makedirs(output_dir, exist_ok=True)
 
-    BASE_PARAMS = MODEL_TO_PARAMS[model_id]
+    params_map = select_ssim_params(MODEL_TO_PARAMS, FULL_QUALITY_MODEL_TO_PARAMS)
+    BASE_PARAMS = params_map[model_id]
     num_inference_steps = BASE_PARAMS["num_inference_steps"]
     model_path = BASE_PARAMS["model_path"]
 
@@ -178,18 +196,23 @@ def test_gen3c_inference_similarity(prompt, ATTENTION_BACKEND, model_id):
     for stale_video in glob.glob(stale_pattern):
         os.remove(stale_video)
 
-    generator = VideoGenerator.from_pretrained(
-        model_path=model_path, **init_kwargs
-    )
-    generator.generate_video(prompt, **generation_kwargs)
-
-    if isinstance(generator.executor, MultiprocExecutor):
-        generator.executor.shutdown()
+    generator = None
+    try:
+        generator = VideoGenerator.from_pretrained(
+            model_path=model_path, **init_kwargs
+        )
+        generator.generate_video(prompt, **generation_kwargs)
+    finally:
+        if generator is not None and isinstance(generator.executor, MultiprocExecutor):
+            generator.executor.shutdown()
 
     assert os.path.exists(output_dir), f"Output not generated at {output_dir}"
 
-    reference_folder = os.path.join(
-        script_dir, device_reference_folder, model_id, ATTENTION_BACKEND
+    reference_folder = build_reference_folder_path(
+        script_dir,
+        device_reference_folder,
+        model_id,
+        ATTENTION_BACKEND,
     )
     if not os.path.exists(reference_folder):
         raise FileNotFoundError(

fastvideo/tests/ssim/test_lingbot_similarity.py

@@ -55,6 +55,7 @@ device_reference_folder = resolve_device_reference_folder(
         ("L40S", "L40S"),
         ("H100", "H100"),
         ("H200", "H200"),
+        ("B200", "B200"),
     ),
     device_name=device_name,
     logger=logger,

fastvideo/tests/ssim/test_longcat_similarity.py

@@ -42,6 +42,7 @@ device_reference_folder = resolve_device_reference_folder(
         ("L40S", "L40S"),
         ("H100", "H100"),
         ("H200", "H200"),
+        ("B200", "B200"),
     ),
     device_name=get_cuda_device_name(),
     logger=logger,

fastvideo/tests/ssim/test_matrixgame2_similarity.py

@@ -30,6 +30,7 @@ device_reference_folder = resolve_device_reference_folder(
         ("A40", "A40"),
         ("L40S", "L40S"),
         ("H200", "H200"),
+        ("B200", "B200"),
     ),
     device_name=get_cuda_device_name(),
     logger=logger,

fastvideo/tests/ssim/test_matrixgame3_similarity.py

@@ -28,6 +28,7 @@ device_reference_folder = resolve_device_reference_folder(
         ("A40", "A40"),
         ("L40S", "L40S"),
         ("H200", "H200"),
+        ("B200", "B200"),
     ),
     device_name=get_cuda_device_name(),
     fallback_device_prefix="L40S",

fastvideo/tests/ssim/test_sd35_similarity.py

@@ -31,6 +31,7 @@ device_reference_folder = resolve_device_reference_folder(
         ("L40S", "L40S"),
         ("H100", "H100"),
         ("H200", "H200"),
+        ("B200", "B200"),
         ("RTX 4090", "RTX4090"),
         ("4090", "RTX4090"),
     ),