[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>
[misc]: update reseed-performance-baseline skill for the hf_store move (#1545 follow-up) (#1553 )
2026-07-05 21:19:38 +00:00 · 2026-07-05 14:16:55 -07:00 · 2026-07-05 14:16:20 -07:00 · 2026-07-05 14:07:38 -07:00 · 2026-07-05 14:05:26 -07:00 · 2026-07-05 12:05:25 -07:00
34 changed files with 1661 additions and 219 deletions
@@ -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.
@@ -12,13 +12,17 @@ Defaults:

 - `HF_REPO_ID=FastVideo/performance-tracking`
 - `PERFORMANCE_TRACKING_ROOT=/tmp/fastvideo-perf-dashboard`
- `PERF_MAX_REGRESSION=0.05`

-Records can include source metadata:
+Records can include source metadata and rolling-baseline policy context:

 - `run_source`: `pr`, `local`, `scheduled_main`, or `unknown`
 - `baseline_eligible`: only successful scheduled-main records should be true
 - Buildkite metadata such as branch, PR number, build URL, build ID, and job ID
+- `regression_thresholds`: per-metric rolling-baseline percent and absolute
+  floors used for recomputed status context
+
+Dashboard/API metric payloads expose `threshold_exceeded` for raw threshold
+crossings; `regressed` remains the gated CI-failure signal.

 Set one of `HF_API_KEY`, `HUGGINGFACE_HUB_TOKEN`, or `HF_TOKEN` if the
 configured dataset repo requires authenticated access:
@@ -89,7 +93,8 @@ Trend charts show metric-specific axes and exact point details on hover/focus:
 - PR number, branch, and Buildkite URL when present

 The latest status table uses the stored JSON `success` value. Recomputed
-baseline context is shown separately and does not override stored status.
+baseline context applies each metric's percent and absolute regression floors
+and does not override stored status.

 ## API

@@ -440,6 +440,8 @@ export default function App() {
                  <th>Throughput</th>
                  <th>Memory</th>
                  <th>Worst</th>
+                  <th>Exceeded</th>
+                  <th>Failing</th>
                </tr>
              </thead>
              <tbody>
@@ -465,6 +467,12 @@ export default function App() {
                    <td>{formatNumber(row.metrics.throughput?.current, 3)}</td>
                    <td>{formatNumber(row.metrics.memory?.current, 1)}</td>
                    <td>{formatNumber(row.worst_regression_pct, 1)}%</td>
+                    <td>
+                      {row.threshold_exceeded_metrics.length
+                        ? row.threshold_exceeded_metrics.join(", ")
+                        : "none"}
+                    </td>
+                    <td>{row.failing_metrics.length ? row.failing_metrics.join(", ") : "none"}</td>
                  </tr>
                ))}
              </tbody>
@@ -2,6 +2,12 @@ export type MetricValue = {
  current: number | null;
  baseline: number | null;
  regression_pct: number | null;
+  absolute_delta: number | null;
+  threshold_percent: number;
+  threshold_absolute: number;
+  gated: boolean;
+  threshold_exceeded: boolean;
+  regressed: boolean;
  label: string;
  lower_is_better: boolean;
  precision: number;
@@ -15,7 +21,8 @@ export type SummaryRow = {
  success: boolean;
  baseline_n: number;
  worst_regression_pct: number | null;
-  regression_threshold_pct: number;
+  threshold_exceeded_metrics: string[];
+  failing_metrics: string[];
  computed_regression_status: "pass" | "fail";
  status: "pass" | "fail";
  run_source: RunSource;
@@ -72,7 +72,10 @@ fastvideo/tests/performance/
    │           writes Markdown summary + (optionally) uploads new records
    ├── dashboard.py
    │       └── builds time-series Plotly HTML from HF history
-    └── hf_store.py               # shared HF I/O + DataFrame helpers
+
+fastvideo/performance/
+    ├── hf_store.py               # shared HF I/O + DataFrame helpers
+    └── metric_policy.py          # shared rolling-baseline threshold policy
 ```

 The HF dataset (`FastVideo/performance-tracking` by default) holds one
@@ -92,25 +95,28 @@ and recipe changes instead of treating all records for a model as equivalent.

 ## Metrics

-Each benchmark records six metrics:
+Each benchmark records six metrics. The rolling-baseline comparator also has a
+per-metric policy with direction, percent threshold, absolute threshold, and a
+`gated` flag.

-| Metric | Raw key | Normalized key | Direction |
-|---|---|---|---|
-| End-to-end generation latency | `avg_generation_time_s` | `latency` | Lower is better |
-| Video throughput | `throughput_fps` | `throughput` | Higher is better |
-| Peak GPU memory | `max_peak_memory_mb` | `memory` | Lower is better |
-| Text encoder time | `text_encoder_time_s` | `text_encoder_time_s` | Lower is better |
-| DiT denoising time | `dit_time_s` | `dit_time_s` | Lower is better |
-| VAE decode time | `vae_decode_time_s` | `vae_decode_time_s` | Lower is better |
+| Metric | Raw key | Normalized key | Direction | Default rolling policy |
+|---|---|---|---|---|
+| End-to-end generation latency | `avg_generation_time_s` | `latency` | Lower is better | 8% and 0.5 s |
+| Video throughput | `throughput_fps` | `throughput` | Higher is better | 8% and 0.05 FPS |
+| Peak GPU memory | `max_peak_memory_mb` | `memory` | Lower is better | 5% and 256 MB |
+| Text encoder time | `text_encoder_time_s` | `text_encoder_time_s` | Lower is better | 5% and 0.25 s |
+| DiT denoising time | `dit_time_s` | `dit_time_s` | Lower is better | 5% and 0.25 s |
+| VAE decode time | `vae_decode_time_s` | `vae_decode_time_s` | Lower is better | 5% and 0.25 s |

 `test_inference_performance.py` temporarily sets `FASTVIDEO_STAGE_LOGGING=1`
 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
@@ -156,9 +162,22 @@ headroom and almost never need touching.

 `compare_baseline.py` loads the last 5 successful, baseline-eligible records
 for the same `(model_id, gpu_type)` from the HF dataset, computes the median
-for each available metric, and fails if the current run regresses by more than
-`PERF_MAX_REGRESSION` (default 5%). For latency, memory, and component times,
-higher values are regressions. For throughput, lower values are regressions.
+for each available metric, and evaluates the current run with the metric's
+rolling regression policy. For latency, memory, and component times, higher
+values are regressions. For throughput, lower values are regressions.
+
+A metric exceeds its rolling threshold when both of these are true:
+
+```text
+percent_delta > threshold_percent
+absolute_delta > threshold_absolute
+```
+
+Gated metrics fail CI when that threshold crossing happens. Set `gated: false`
+for metrics that should remain visible in reports and the dashboard without
+failing CI. Dashboard/API payloads expose `threshold_exceeded` separately from
+`regressed`, where `regressed` means a gated CI failure. Missing or `null`
+metrics are skipped.

 This is the **drift detector** — it catches sub-threshold regressions that
 slowly add up. Only scheduled-main successful records are baseline eligible.
@@ -239,6 +258,13 @@ Written by `test_inference_performance.py`. One file per benchmark run.
    "max_dit_time_s": 10.0,
    "max_vae_decode_time_s": 10.0
  },
+  "regression_thresholds": {
+    "latency": {
+      "threshold_percent": 0.10,
+      "threshold_absolute": 1.0,
+      "gated": true
+    }
+  },
  "commit": "<full sha>",
  "pr_number": "1234",
  "timestamp": "2026-05-08T22:00:00+00:00",
@@ -265,6 +291,13 @@ result, used as the rolling-baseline source of truth.
  "text_encoder_time_s": 2.141,
  "dit_time_s": 8.437,
  "vae_decode_time_s": 3.208,
+  "regression_thresholds": {
+    "latency": {
+      "threshold_percent": 0.08,
+      "threshold_absolute": 0.5,
+      "gated": true
+    }
+  },
  "success": true
 }
 ```
@@ -281,18 +314,17 @@ successful main/full-suite uploads and remain eligible for rolling baselines.

 | Variable | Default | Used by | Purpose |
 |---|---|---|---|
-| `PERF_MAX_REGRESSION` | `0.05` | `compare_baseline.py` | Per-metric regression fraction that fails the build. |
 | `PERFORMANCE_TRACKING_ROOT` | `/tmp/perf-tracking` | `compare_baseline.py`, `dashboard.py` | Local directory the HF dataset is synced to. |
 | `PERF_REPORTS_DIR` | `/root/data/perf_reports` | `compare_baseline.py`, `dashboard.py` | Where the Markdown summary and Plotly HTML get written for Buildkite to pick up. |
-| `HF_REPO_ID` | `FastVideo/performance-tracking` | `hf_store.py` | HF dataset repo holding rolling-baseline records. |
-| `HF_API_KEY`, `HUGGINGFACE_HUB_TOKEN`, `HF_TOKEN` | unset | `hf_store.py` | Required for upload or private dataset reads. |
+| `HF_REPO_ID` | `FastVideo/performance-tracking` | `fastvideo/performance/hf_store.py` | HF dataset repo holding rolling-baseline records. |
+| `HF_API_KEY`, `HUGGINGFACE_HUB_TOKEN`, `HF_TOKEN` | unset | `fastvideo/performance/hf_store.py` | Required for upload or private dataset reads. |
 | `PERF_RUN_SOURCE` | inferred | `compare_baseline.py` | Source metadata for uploaded records: `pr`, `local`, `scheduled_main`, or `unknown`. |
 | `PERF_UPLOAD_POLICY` | `never` | `compare_baseline.py` | Upload policy: `never`, `pass`, or `always`. |
 | `PERF_PYTEST_RC` | unset | `compare_baseline.py` | Static-threshold pytest exit code, used so scheduled-main failures can be uploaded with `success=false`. |
 | `TEST_SCOPE` | unset | `compare_baseline.py` | CI context used to infer scheduled-main runs together with `BUILDKITE_BRANCH=main`. |
 | `BUILDKITE_BRANCH`, `BUILDKITE_COMMIT`, `BUILDKITE_PULL_REQUEST` | unset | `compare_baseline.py`, `test_inference_performance.py` | CI metadata stamped into records. |
 | `DASHBOARD_DAYS` | `30` | `dashboard.py` | Lookback window for the Plotly trend pages. |
-| `PERFORMANCE_TRACKING_SYNC_REUSE_TTL_SECONDS` | `3600` | `hf_store.py` | Freshness window for reusing an existing HF sync when requested by dashboard consumers. |
+| `PERFORMANCE_TRACKING_SYNC_REUSE_TTL_SECONDS` | `3600` | `fastvideo/performance/hf_store.py` | Freshness window for reusing an existing HF sync when requested by dashboard consumers. |
 | `FASTVIDEO_STAGE_LOGGING` | set by the pytest test | `test_inference_performance.py` | Enables pipeline stage timing capture for component metrics during benchmark runs. |

 ## CI integration
@@ -304,9 +336,11 @@ Buildkite artifact upload is in
 `.buildkite/scripts/pr_test.sh:upload_performance_artifacts`.

 Each performance build runs pytest first. If that fixed-threshold phase fails,
-`compare_baseline.py` is skipped, so Markdown summaries and normalized JSON
-artifacts are not emitted. The dashboard still runs best-effort for
-observability. When pytest passes, the rolling-baseline phase emits:
+PR/direct runs skip `compare_baseline.py` because they only upload passing
+records. Scheduled-main runs still execute `compare_baseline.py` with
+`PERF_PYTEST_RC` set so the failed canonical attempt is visible in normalized
+JSON and dashboard history. The dashboard runs best-effort for observability.
+When the rolling-baseline phase runs, it emits:

 * **Markdown summary** — appended to `$GITHUB_STEP_SUMMARY` when that variable
  is set, and written as `perf_<sha>_<ts>.md` for Buildkite upload. Contains a
@@ -347,9 +381,13 @@ observability. When pytest passes, the rolling-baseline phase emits:
         "max_vae_decode_time_s": 10.0
       },
       "default": { "max_generation_time_s": 120.0, "max_peak_memory_mb": 30000.0 }
+     },
+     "regression_thresholds": {
+       "latency": { "threshold_percent": 0.10, "threshold_absolute": 1.0, "gated": true }
     }
-   }
-   ```
+  }
+
+  ```

   Legacy v1 configs without `config_schema_version` still load, but should not
   gain v2 identity or metadata fields until they are migrated to
@@ -372,6 +410,13 @@ observability. When pytest passes, the rolling-baseline phase emits:
   a useful fixed gate. The rolling baseline will still track component times
   when static component thresholds are omitted.

+6. Omit `regression_thresholds` to use the default rolling-baseline policy, or
+   include only benchmark-specific deviations. Tune these independently from
+   the fixed thresholds when a metric is noisy or should be informational. The
+   fixed `thresholds` block is an absolute pytest ceiling. The
+   `regression_thresholds` block controls rolling-baseline comparisons against
+   recent scheduled-main records.
+
 ## Troubleshooting

 **"No baseline for ... Initializing"** — first run for this `(model_id,
@@ -388,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`.
@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Performance benchmark and dashboard utilities."""
@@ -57,9 +57,7 @@ def is_baseline_eligible_record(record: dict[str, Any]) -> bool:
    """
    if record.get("baseline_eligible") is True:
        return True
-    if "baseline_eligible" not in record and "run_source" not in record:
-        return True
-    return False
+    return "baseline_eligible" not in record and "run_source" not in record


 def resolve_hf_token() -> str | None:
@@ -312,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",
 )


@@ -0,0 +1,116 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Metric policy for rolling performance baseline comparisons."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class MetricPolicy:
+    key: str
+    label: str
+    precision: int
+    lower_is_better: bool
+    threshold_percent: float
+    threshold_absolute: float
+    gated: bool = True
+
+
+@dataclass(frozen=True)
+class MetricDelta:
+    absolute: float
+    percent: float
+    threshold_exceeded: bool
+    regressed: bool
+
+
+DEFAULT_METRIC_POLICIES: tuple[MetricPolicy, ...] = (
+    MetricPolicy("latency", "Latency", 3, True, 0.08, 0.5),
+    MetricPolicy("throughput", "Throughput", 3, False, 0.08, 0.05),
+    MetricPolicy("memory", "Memory", 1, True, 0.05, 256.0),
+    MetricPolicy("text_encoder_time_s", "Text Enc", 3, True, 0.05, 0.25),
+    MetricPolicy("dit_time_s", "DiT", 3, True, 0.05, 0.25),
+    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
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _optional_bool(value: Any) -> bool | None:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"1", "true", "yes", "on"}:
+            return True
+        if normalized in {"0", "false", "no", "off"}:
+            return False
+    return None
+
+
+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):
+        threshold_overrides = {}
+    policies: list[MetricPolicy] = []
+    for base_policy in DEFAULT_METRIC_POLICIES:
+        raw_override = threshold_overrides.get(base_policy.key, {})
+        if not isinstance(raw_override, Mapping):
+            raw_override = {}
+
+        threshold_percent = _optional_float(raw_override.get("threshold_percent"))
+        threshold_absolute = _optional_float(raw_override.get("threshold_absolute"))
+        gated = _optional_bool(raw_override.get("gated"))
+
+        policies.append(
+            MetricPolicy(
+                key=base_policy.key,
+                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),
+                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]]:
+    return {
+        policy.key: {
+            "threshold_percent": policy.threshold_percent,
+            "threshold_absolute": policy.threshold_absolute,
+            "gated": policy.gated,
+        }
+        for policy in policies
+    }
+
+
+def regression_delta(
+    policy: MetricPolicy,
+    current: float,
+    baseline: float,
+) -> MetricDelta | None:
+    if baseline <= 0:
+        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)
+    return MetricDelta(
+        absolute=absolute_delta,
+        percent=percent_delta,
+        threshold_exceeded=threshold_exceeded,
+        regressed=policy.gated and threshold_exceeded,
+    )
@@ -13,7 +13,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles

-from fastvideo.tests.performance import hf_store
+from fastvideo.performance import hf_store

 from .service import build_latest_summary, build_trends, filter_records

@@ -150,7 +150,6 @@ def create_app(store: PerformanceDataStore | None = None) -> FastAPI:
        filtered = filter_records(loaded, model_id=model_id, gpu_type=gpu_type)
        rows = build_latest_summary(
            filtered,
-            max_regression=float(os.environ.get("PERF_MAX_REGRESSION", "0.05")),
            run_source=run_source,
        )
        return {
@@ -1,26 +1,8 @@
 # SPDX-License-Identifier: Apache-2.0
 """Metric definitions shared by the performance dashboard backend."""

-from __future__ import annotations
+from fastvideo.performance.metric_policy import DEFAULT_METRIC_POLICIES

-from dataclasses import dataclass
-
-
-@dataclass(frozen=True)
-class MetricDefinition:
-    key: str
-    label: str
-    precision: int
-    lower_is_better: bool
-
-
-METRICS: tuple[MetricDefinition, ...] = (
-    MetricDefinition("latency", "Latency", 3, True),
-    MetricDefinition("throughput", "Throughput", 3, False),
-    MetricDefinition("memory", "Memory", 1, True),
-    MetricDefinition("text_encoder_time_s", "Text Encoder", 3, True),
-    MetricDefinition("dit_time_s", "DiT", 3, True),
-    MetricDefinition("vae_decode_time_s", "VAE Decode", 3, True),
-)
+METRICS = DEFAULT_METRIC_POLICIES

 METRIC_BY_KEY = {metric.key: metric for metric in METRICS}
@@ -13,9 +13,8 @@ from collections import defaultdict
 from datetime import datetime, timezone
 from typing import Any

-from fastvideo.tests.performance.hf_store import is_baseline_eligible_record, safe_float
-
-from .metrics import METRICS
+from fastvideo.performance.hf_store import is_baseline_eligible_record, safe_float
+from fastvideo.performance.metric_policy import regression_delta, resolve_metric_policies

 Record = dict[str, Any]

@@ -95,19 +94,9 @@ def baseline_value(records: list[Record], metric_key: str) -> float | None:
    return float(statistics.median(values))


-def regression_percent(metric_key: str, current: float | None, baseline: float | None) -> float | None:
-    if current is None or baseline is None or baseline <= 0:
-        return None
-    metric = next(metric for metric in METRICS if metric.key == metric_key)
-    if metric.lower_is_better:
-        return (current - baseline) / baseline * 100.0
-    return (baseline - current) / baseline * 100.0
-
-
 def build_latest_summary(records: list[Record],
                         *,
                         baseline_window: int = 5,
-                         max_regression: float = 0.05,
                         run_source: str | None = None) -> list[Record]:
    rows: list[Record] = []
    for (model_id, gpu_type), group in group_by_model_gpu(records).items():
@@ -123,52 +112,58 @@ def build_latest_summary(records: list[Record],
            if record is not latest and record.get("success", True) and is_baseline_eligible_record(record)
        ]
        baseline_records = baseline_pool[-baseline_window:]
+        metric_policies = resolve_metric_policies(latest.get("regression_thresholds"))

        metrics: dict[str, Record] = {}
        regressions: list[float] = []
-        for metric in METRICS:
-            current = safe_float(latest.get(metric.key))
-            baseline = baseline_value(baseline_records, metric.key)
-            regression = regression_percent(metric.key, current, baseline)
-            metrics[metric.key] = {
+        failing_metrics: list[str] = []
+        threshold_exceeded_metrics: list[str] = []
+        for policy in metric_policies:
+            current = safe_float(latest.get(policy.key))
+            baseline = baseline_value(baseline_records, policy.key)
+            delta = None
+            if current is not None and baseline is not None:
+                delta = regression_delta(policy, current, baseline)
+            regression = None if delta is None else delta.percent * 100.0
+            metrics[policy.key] = {
                "current": current,
                "baseline": baseline,
                "regression_pct": regression,
-                "label": metric.label,
-                "lower_is_better": metric.lower_is_better,
-                "precision": metric.precision,
+                "absolute_delta": None if delta is None else delta.absolute,
+                "threshold_percent": policy.threshold_percent * 100.0,
+                "threshold_absolute": policy.threshold_absolute,
+                "gated": policy.gated,
+                "threshold_exceeded": False if delta is None else delta.threshold_exceeded,
+                "regressed": False if delta is None else delta.regressed,
+                "label": policy.label,
+                "lower_is_better": policy.lower_is_better,
+                "precision": policy.precision,
            }
            if regression is not None:
                regressions.append(regression)
+            if delta is not None and delta.threshold_exceeded:
+                threshold_exceeded_metrics.append(policy.key)
+            if delta is not None and delta.regressed:
+                failing_metrics.append(policy.key)

        worst_regression = max(regressions) if regressions else None
        success = bool(latest.get("success", True))
        status = "pass" if success else "fail"

        rows.append({
-            "model_id":
-            model_id,
-            "gpu_type":
-            gpu_type,
-            "timestamp":
-            latest.get("timestamp"),
-            "commit_sha":
-            latest.get("commit_sha"),
+            "model_id": model_id,
+            "gpu_type": gpu_type,
+            "timestamp": latest.get("timestamp"),
+            "commit_sha": latest.get("commit_sha"),
            **record_metadata(latest),
-            "success":
-            success,
-            "baseline_n":
-            len(baseline_records),
-            "worst_regression_pct":
-            worst_regression,
-            "regression_threshold_pct":
-            max_regression * 100.0,
-            "computed_regression_status":
-            "fail" if worst_regression is not None and worst_regression > max_regression * 100.0 else "pass",
-            "status":
-            status,
-            "metrics":
-            metrics,
+            "success": success,
+            "baseline_n": len(baseline_records),
+            "worst_regression_pct": worst_regression,
+            "threshold_exceeded_metrics": threshold_exceeded_metrics,
+            "failing_metrics": failing_metrics,
+            "computed_regression_status": "fail" if failing_metrics else "pass",
+            "status": status,
+            "metrics": metrics,
        })

    return sorted(rows, key=lambda row: (row["status"] != "fail", row["model_id"], row["gpu_type"]))
@@ -179,14 +174,15 @@ def build_trends(records: list[Record]) -> list[Record]:
    for (model_id, gpu_type), group in group_by_model_gpu(records).items():
        points = []
        for record in group:
+            metric_policies = resolve_metric_policies(record.get("regression_thresholds"))
            point = {
                "timestamp": record.get("timestamp"),
                "commit_sha": record.get("commit_sha"),
                **record_metadata(record),
                "success": bool(record.get("success", True)),
                "metrics": {
-                    metric.key: safe_float(record.get(metric.key))
-                    for metric in METRICS
+                    policy.key: safe_float(record.get(policy.key))
+                    for policy in metric_policies
                },
            }
            points.append(point)
@@ -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
@@ -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
@@ -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__()
@@ -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__()
@@ -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}")
@@ -276,7 +276,7 @@ def run_self_forcing_tests():
@app.function(gpu="L40S:1", image=image, timeout=900)
 def run_unit_test():
    run_test(
-        "pytest ./fastvideo/tests/api/ ./fastvideo/tests/contract/ ./fastvideo/tests/dataset/ ./fastvideo/tests/workflow/ ./fastvideo/tests/entrypoints/ ./fastvideo/tests/train/ --ignore=./fastvideo/tests/entrypoints/test_openai_api_integration.py --ignore=./fastvideo/tests/train/models --ignore=./fastvideo/tests/train/methods -vs"
+        "pytest ./fastvideo/tests/api/ ./fastvideo/tests/contract/ ./fastvideo/tests/dataset/ ./fastvideo/tests/workflow/ ./fastvideo/tests/entrypoints/ ./fastvideo/tests/train/ ./fastvideo/tests/stages/ --ignore=./fastvideo/tests/entrypoints/test_openai_api_integration.py --ignore=./fastvideo/tests/train/models --ignore=./fastvideo/tests/train/methods -vs"
    )


@@ -8,8 +8,8 @@ This script:
   baseline-eligible successful records (filtered by gpu_type),
 4) writes normalized records back to the HF dataset repo according to
   PERF_UPLOAD_POLICY,
-5) exits non-zero if any metric regresses by more than PERF_MAX_REGRESSION
-   (default 5%).
+5) exits non-zero if any gated metric exceeds both its percent and absolute
+   regression floors.
 """

 import glob
@@ -21,21 +21,36 @@ from datetime import datetime, timezone
 from typing import Any

 try:
-    from .hf_store import (
+    from fastvideo.performance.hf_store import (
        load_records_for_model,
        safe_float,
        sanitize,
        sync_from_hf,
        upload_record,
    )
+    from fastvideo.performance.metric_policy import (
+        MetricPolicy,
+        regression_delta,
+        resolve_metric_policies,
+        serialize_metric_thresholds,
+    )
 except ImportError:
-    from hf_store import (
+    repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
+    if repo_root not in sys.path:
+        sys.path.insert(0, repo_root)
+    from fastvideo.performance.hf_store import (
        load_records_for_model,
        safe_float,
        sanitize,
        sync_from_hf,
        upload_record,
    )
+    from fastvideo.performance.metric_policy import (
+        MetricPolicy,
+        regression_delta,
+        resolve_metric_policies,
+        serialize_metric_thresholds,
+    )

 RESULTS_DIR = os.path.join(
    os.path.dirname(os.path.abspath(__file__)),
@@ -46,25 +61,9 @@ TRACKING_ROOT = os.environ.get(
    "/tmp/perf-tracking",
 )
 PERF_REPORTS_DIR = os.environ.get("PERF_REPORTS_DIR", "/root/data/perf_reports")
-MAX_REGRESSION = float(os.environ.get("PERF_MAX_REGRESSION", "0.05"))
 UPLOAD_POLICY = os.environ.get("PERF_UPLOAD_POLICY", "never").strip().lower()
 VALID_UPLOAD_POLICIES = {"never", "pass", "always"}
 VALID_RUN_SOURCES = {"pr", "local", "scheduled_main", "unknown"}
-METRICS = (
-    ("latency", "Latency", 3),
-    ("throughput", "Throughput", 3),
-    ("memory", "Memory", 1),
-    ("text_encoder_time_s", "Text Enc", 3),
-    ("dit_time_s", "DiT", 3),
-    ("vae_decode_time_s", "VAE Decode", 3),
-)
-LOWER_IS_BETTER_METRICS = {
-    "latency",
-    "memory",
-    "text_encoder_time_s",
-    "dit_time_s",
-    "vae_decode_time_s",
-}


 def _should_persist_tracking() -> bool:
@@ -168,6 +167,7 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
    text_encoder_time = safe_float(result.get("text_encoder_time_s"))
    dit_time = safe_float(result.get("dit_time_s"))
    vae_decode_time = safe_float(result.get("vae_decode_time_s"))
+    metric_policies = resolve_metric_policies(result.get("regression_thresholds"))

    return {
        "model_id": model_id,
@@ -180,6 +180,7 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
        "text_encoder_time_s": text_encoder_time,
        "dit_time_s": dit_time,
        "vae_decode_time_s": vae_decode_time,
+        "regression_thresholds": serialize_metric_thresholds(metric_policies),
        "success": True,
        **_record_metadata(_detect_run_source(), result),
    }
@@ -229,55 +230,41 @@ def _baseline_metric(records: list[dict[str, Any]], key: str) -> float | None:
    return statistics.median(values)


+def _metric_policy_summary(policy: MetricPolicy) -> str:
+    gated = "gated" if policy.gated else "info"
+    return (
+        f"{gated}, >{policy.threshold_percent * 100:.1f}% "
+        f"and >{policy.threshold_absolute:.{policy.precision}f}"
+    )
+
+
 def _check_regressions(
    current: dict[str, Any],
    baseline_records: list[dict[str, Any]],
-    max_regression: float,
+    metric_policies: tuple[MetricPolicy, ...],
 ) -> list[str]:
    failures: list[str] = []

-    for metric, _label, _precision in METRICS:
-        if metric not in LOWER_IS_BETTER_METRICS:
+    for policy in metric_policies:
+        baseline = _baseline_metric(baseline_records, policy.key)
+        curr = safe_float(current.get(policy.key))
+        if baseline is None or curr is None:
            continue
-        baseline = _baseline_metric(baseline_records, metric)
-        curr = safe_float(current.get(metric))
-        if baseline is None or curr is None or baseline <= 0:
+        delta = regression_delta(policy, curr, baseline)
+        if delta is None or not delta.regressed:
            continue
-        regression = (curr - baseline) / baseline
-        if regression > max_regression:
-            failures.append(f"{current['model_id']} {metric} regressed by "
-                            f"{regression * 100:.1f}% "
-                            f"(current={curr:.3f}, baseline_median={baseline:.3f})")
-
-    baseline_tp = _baseline_metric(baseline_records, "throughput")
-    curr_tp = safe_float(current.get("throughput"))
-    if baseline_tp is not None and curr_tp is not None and baseline_tp > 0:
-        regression = (baseline_tp - curr_tp) / baseline_tp
-        if regression > max_regression:
-            failures.append(f"{current['model_id']} throughput regressed by "
-                            f"{regression * 100:.1f}% "
-                            f"(current={curr_tp:.3f}, baseline_median={baseline_tp:.3f})")
+        failures.append(
+            f"{current['model_id']} {policy.key} regressed by "
+            f"{delta.percent * 100:.1f}% and "
+            f"{delta.absolute:.{policy.precision}f} "
+            f"(current={curr:.{policy.precision}f}, "
+            f"baseline_median={baseline:.{policy.precision}f}, "
+            f"threshold={_metric_policy_summary(policy)})"
+        )

    return failures


-def _metric_delta_percent(
-    metric: str,
-    current: dict[str, Any],
-    baseline_records: list[dict[str, Any]],
-) -> float | None:
-    curr = safe_float(current.get(metric))
-    baseline = _baseline_metric(baseline_records, metric)
-    if curr is None or baseline is None or baseline <= 0:
-        return None
-
-    if metric in LOWER_IS_BETTER_METRICS:
-        return (curr - baseline) / baseline * 100.0
-    if metric == "throughput":
-        return (baseline - curr) / baseline * 100.0
-    return None
-
-
 def _compact_value(value: float | None, precision: int = 3) -> str:
    if value is None:
        return "n/a"
@@ -287,23 +274,42 @@ def _compact_value(value: float | None, precision: int = 3) -> str:
 def _build_summary_row(
    record: dict[str, Any],
    baseline_records: list[dict[str, Any]],
+    metric_policies: tuple[MetricPolicy, ...],
    has_failed: bool,
 ) -> dict[str, Any]:
    """Format a single benchmark result as a row for the Markdown table."""

-    metric_values: dict[str, dict[str, float | None]] = {}
+    metric_values: dict[str, dict[str, Any]] = {}
    regressions: list[float] = []
-    for metric, _label, _precision in METRICS:
-        curr = safe_float(record.get(metric))
-        baseline = _baseline_metric(baseline_records, metric)
-        regression = _metric_delta_percent(metric, record, baseline_records)
-        metric_values[metric] = {
+    failing_metrics: list[str] = []
+    threshold_exceeded_metrics: list[str] = []
+    for policy in metric_policies:
+        curr = safe_float(record.get(policy.key))
+        baseline = _baseline_metric(baseline_records, policy.key)
+        delta = (
+            regression_delta(policy, curr, baseline)
+            if curr is not None and baseline is not None
+            else None
+        )
+        regression = None if delta is None else delta.percent * 100.0
+        absolute_delta = None if delta is None else delta.absolute
+        metric_values[policy.key] = {
            "curr": curr,
            "base": baseline,
            "regression_pct": regression,
+            "absolute_delta": absolute_delta,
+            "threshold_percent": policy.threshold_percent * 100.0,
+            "threshold_absolute": policy.threshold_absolute,
+            "gated": policy.gated,
+            "threshold_exceeded": False if delta is None else delta.threshold_exceeded,
+            "regressed": False if delta is None else delta.regressed,
        }
        if regression is not None:
            regressions.append(regression)
+        if delta is not None and delta.threshold_exceeded:
+            threshold_exceeded_metrics.append(policy.key)
+        if delta is not None and delta.regressed:
+            failing_metrics.append(policy.key)

    worst_regression_pct = max(regressions) if regressions else None

@@ -313,40 +319,50 @@ def _build_summary_row(
        "baseline_n": len(baseline_records),
        "metrics": metric_values,
        "worst_regression_pct": worst_regression_pct,
+        "threshold_exceeded_metrics": threshold_exceeded_metrics,
+        "failing_metrics": failing_metrics,
        "failed": has_failed,
    }


 def _build_markdown_summary(
    summary_rows: list[dict[str, Any]],
-    max_regression: float,
+    metric_policies: tuple[MetricPolicy, ...],
 ) -> str:
    lines = [
        "## Performance Baseline Comparison",
        "",
-        f"Threshold: regressions greater than {max_regression * 100:.1f}% fail",
+        "Threshold: gated metrics fail only when both percent and absolute "
+        "regression floors are exceeded.",
        "",
        ("| Model | GPU | Baseline N | Latency (curr/base) | "
         "Throughput (curr/base) | Memory (curr/base) | "
         "Text Enc (curr/base) | DiT (curr/base) | "
-         "VAE Decode (curr/base) | Worst Regression | Status |"),
-        "|---|---|---:|---|---|---|---|---|---|---:|---|",
+         "VAE Decode (curr/base) | Worst Regression | Exceeded Metrics | "
+         "Failing Metrics | Status |"),
+        "|---|---|---:|---|---|---|---|---|---|---:|---|---|---|",
    ]

    for row in summary_rows:
        metric_cells = []
-        for metric, _label, precision in METRICS:
-            values = row["metrics"][metric]
-            metric_cells.append(f"{_compact_value(values['curr'], precision)} / "
-                                f"{_compact_value(values['base'], precision)}")
+        for policy in metric_policies:
+            values = row["metrics"][policy.key]
+            metric_cells.append(f"{_compact_value(values['curr'], policy.precision)} / "
+                                f"{_compact_value(values['base'], policy.precision)}")

        worst_reg = ("n/a" if row["worst_regression_pct"] is None else f"{row['worst_regression_pct']:.1f}%")
+        exceeded_metrics = (
+            ", ".join(row["threshold_exceeded_metrics"])
+            if row["threshold_exceeded_metrics"]
+            else "none"
+        )
+        failing_metrics = ", ".join(row["failing_metrics"]) if row["failing_metrics"] else "none"
        status = "FAIL" if row["failed"] else "PASS"

        lines.append(f"| {row['model_id']} | {row['gpu_type']} | "
                     f"{row['baseline_n']} | "
                     f"{' | '.join(metric_cells)} | "
-                     f"{worst_reg} | {status} |")
+                     f"{worst_reg} | {exceeded_metrics} | {failing_metrics} | {status} |")

    return "\n".join(lines) + "\n"

@@ -400,6 +416,7 @@ def main() -> int:

    for raw in current_results:
        record = _normalize_record(raw)
+        metric_policies = resolve_metric_policies(record.get("regression_thresholds"))

        baseline_records = load_records_for_model(
            TRACKING_ROOT,
@@ -416,7 +433,7 @@ def main() -> int:
            failures: list[str] = []
            record["success"] = True
        else:
-            failures = _check_regressions(record, baseline_records, MAX_REGRESSION)
+            failures = _check_regressions(record, baseline_records, metric_policies)
        if static_threshold_failed:
            failures.append(f"{record['model_id']} fixed-threshold phase failed "
                            f"(PERF_PYTEST_RC={os.environ.get('PERF_PYTEST_RC')})")
@@ -434,10 +451,10 @@ def main() -> int:
            print("Tracking upload skipped for "
                  f"{record['model_id']} ({record['run_source']}, success={record['success']})")

-        summary_rows.append(_build_summary_row(record, baseline_records, bool(failures)))
+        summary_rows.append(_build_summary_row(record, baseline_records, metric_policies, bool(failures)))

    commit_sha = os.environ.get("BUILDKITE_COMMIT", "unknown")[:7]
-    markdown = _build_markdown_summary(summary_rows, MAX_REGRESSION)
+    markdown = _build_markdown_summary(summary_rows, resolve_metric_policies(None))
    _emit_markdown_summary(markdown, commit_sha)

    if all_failures:
@@ -1,12 +1,19 @@
 # SPDX-License-Identifier: Apache-2.0
 import os
+import sys
 from html import escape
 from datetime import datetime

 import plotly.express as px
 import pandas as pd

-from hf_store import sync_from_hf, load_as_dataframe
+try:
+    from fastvideo.performance.hf_store import load_as_dataframe, sync_from_hf
+except ImportError:
+    repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
+    if repo_root not in sys.path:
+        sys.path.insert(0, repo_root)
+    from fastvideo.performance.hf_store import load_as_dataframe, sync_from_hf

 TRACKING_ROOT = os.environ.get(
    "PERFORMANCE_TRACKING_ROOT",
@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0

 from fastvideo.tests.performance import compare_baseline
+from fastvideo.performance.metric_policy import resolve_metric_policies


 def _raw_result():
@@ -72,9 +73,157 @@ def test_normalized_record_includes_source_metadata(monkeypatch):
    assert record["job_id"] == "job-1"


+def test_normalized_record_includes_effective_regression_thresholds(monkeypatch):
+    monkeypatch.setenv("PERF_RUN_SOURCE", "pr")
+    raw = _raw_result()
+    raw["regression_thresholds"] = {
+        "latency": {
+            "threshold_percent": 0.09,
+            "threshold_absolute": 0.75,
+            "gated": True,
+        },
+        "throughput": {
+            "gated": False,
+        },
+    }
+
+    record = compare_baseline.normalize_performance_result(raw)
+
+    assert record["regression_thresholds"]["latency"] == {
+        "threshold_percent": 0.09,
+        "threshold_absolute": 0.75,
+        "gated": True,
+    }
+    assert record["regression_thresholds"]["throughput"]["gated"] is False
+
+
+def test_invalid_regression_threshold_container_uses_defaults():
+    policies = resolve_metric_policies(["not", "a", "mapping"])
+
+    latency = next(policy for policy in policies if policy.key == "latency")
+    assert latency.threshold_percent == 0.08
+    assert latency.threshold_absolute == 0.5
+    assert latency.gated is True
+
+
+def test_boolean_regression_threshold_values_are_ignored():
+    policies = resolve_metric_policies({
+        "latency": {
+            "threshold_percent": True,
+            "threshold_absolute": False,
+            "gated": "false",
+        }
+    })
+
+    latency = next(policy for policy in policies if policy.key == "latency")
+    assert latency.threshold_percent == 0.08
+    assert latency.threshold_absolute == 0.5
+    assert latency.gated is False
+
+
 def test_baseline_eligibility_only_for_successful_scheduled_main():
    assert compare_baseline._is_baseline_eligible("scheduled_main", True) is True
    assert compare_baseline._is_baseline_eligible("scheduled_main", False) is False
    assert compare_baseline._is_baseline_eligible("pr", True) is False
    assert compare_baseline._is_baseline_eligible("local", True) is False

+
+def test_latency_regression_requires_percent_and_absolute_floors():
+    baseline = [{"latency": 10.0}]
+    current = {"model_id": "wan", "latency": 10.6}
+
+    percent_only = resolve_metric_policies({
+        "latency": {
+            "threshold_percent": 0.05,
+            "threshold_absolute": 0.75,
+        }
+    })
+    absolute_only = resolve_metric_policies({
+        "latency": {
+            "threshold_percent": 0.10,
+            "threshold_absolute": 0.5,
+        }
+    })
+    both = resolve_metric_policies({
+        "latency": {
+            "threshold_percent": 0.05,
+            "threshold_absolute": 0.5,
+        }
+    })
+
+    assert compare_baseline._check_regressions(current, baseline, percent_only) == []
+    assert compare_baseline._check_regressions(current, baseline, absolute_only) == []
+
+    failures = compare_baseline._check_regressions(current, baseline, both)
+    assert len(failures) == 1
+    assert "latency regressed by 6.0% and 0.600" in failures[0]
+
+
+def test_throughput_regression_uses_higher_is_better_direction():
+    baseline = [{"throughput": 10.0}]
+    current = {"model_id": "wan", "throughput": 9.0}
+    policies = resolve_metric_policies({
+        "throughput": {
+            "threshold_percent": 0.05,
+            "threshold_absolute": 0.5,
+        }
+    })
+
+    failures = compare_baseline._check_regressions(current, baseline, policies)
+
+    assert len(failures) == 1
+    assert "throughput regressed by 10.0% and 1.000" in failures[0]
+
+
+def test_memory_regression_uses_metric_specific_absolute_floor():
+    baseline = [{"memory": 10000.0}]
+    current = {"model_id": "wan", "memory": 10600.0}
+    policies = resolve_metric_policies({
+        "memory": {
+            "threshold_percent": 0.05,
+            "threshold_absolute": 256.0,
+        }
+    })
+
+    failures = compare_baseline._check_regressions(current, baseline, policies)
+
+    assert len(failures) == 1
+    assert "memory regressed by 6.0% and 600.0" in failures[0]
+
+
+def test_component_metric_can_gate_independently():
+    baseline = [{"dit_time_s": 8.0}]
+    current = {"model_id": "wan", "dit_time_s": 8.6}
+    policies = resolve_metric_policies({
+        "dit_time_s": {
+            "threshold_percent": 0.05,
+            "threshold_absolute": 0.25,
+        }
+    })
+
+    failures = compare_baseline._check_regressions(current, baseline, policies)
+
+    assert len(failures) == 1
+    assert "dit_time_s regressed by 7.5% and 0.600" in failures[0]
+
+
+def test_informational_metric_remains_visible_without_failing():
+    baseline = [{"throughput": 10.0}]
+    current = {"model_id": "wan", "gpu_type": "NVIDIA L40S", "throughput": 8.0}
+    policies = resolve_metric_policies({
+        "throughput": {
+            "threshold_percent": 0.01,
+            "threshold_absolute": 0.01,
+            "gated": False,
+        }
+    })
+
+    row = compare_baseline._build_summary_row(current, baseline, policies, False)
+
+    assert compare_baseline._check_regressions(current, baseline, policies) == []
+    assert row["metrics"]["throughput"]["regression_pct"] == 20.0
+    assert row["metrics"]["throughput"]["gated"] is False
+    assert row["metrics"]["throughput"]["threshold_exceeded"] is True
+    assert row["metrics"]["throughput"]["regressed"] is False
+    assert row["threshold_exceeded_metrics"] == ["throughput"]
+    assert row["failing_metrics"] == []
@@ -68,6 +68,8 @@ def test_summary_endpoint_returns_latest_group_status():
    assert body["count"] == 1
    assert body["status_counts"] == {"pass": 1, "fail": 0}
    assert body["rows"][0]["metrics"]["latency"]["baseline"] == 10.0
+    assert body["rows"][0]["metrics"]["latency"]["threshold_exceeded"] is True
+    assert body["rows"][0]["threshold_exceeded_metrics"] == ["latency", "throughput"]
    assert body["rows"][0]["computed_regression_status"] == "fail"


@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: Apache-2.0
+from fastvideo.performance import hf_store
 from fastvideo.performance_dashboard.service import build_latest_summary, build_trends, filter_records
-from fastvideo.tests.performance import hf_store


 def _record(ts, commit, latency, throughput, success=True, **metadata):
@@ -28,16 +28,23 @@ def test_build_latest_summary_uses_previous_successful_records_for_baseline():
        _record("2026-01-03T00:00:00+00:00", "c" * 40, 11.0, 9.0),
    ]

-    rows = build_latest_summary(records, max_regression=0.05)
+    rows = build_latest_summary(records)

    assert len(rows) == 1
    row = rows[0]
    assert row["baseline_n"] == 1
    assert row["metrics"]["latency"]["baseline"] == 10.0
    assert row["metrics"]["latency"]["regression_pct"] == 10.0
+    assert row["metrics"]["latency"]["absolute_delta"] == 1.0
+    assert row["metrics"]["latency"]["threshold_percent"] == 8.0
+    assert row["metrics"]["latency"]["threshold_absolute"] == 0.5
+    assert row["metrics"]["latency"]["threshold_exceeded"] is True
+    assert row["metrics"]["latency"]["regressed"] is True
    assert row["metrics"]["throughput"]["regression_pct"] == 10.0
    assert row["status"] == "pass"
    assert row["computed_regression_status"] == "fail"
+    assert row["threshold_exceeded_metrics"] == ["latency", "throughput"]
+    assert row["failing_metrics"] == ["latency", "throughput"]


 def test_build_latest_summary_status_uses_latest_record_success_field():
@@ -73,7 +80,7 @@ def test_build_latest_summary_run_source_filter_keeps_canonical_baseline():
        ),
    ]

-    rows = build_latest_summary(records, max_regression=0.05, run_source="pr")
+    rows = build_latest_summary(records, run_source="pr")

    assert len(rows) == 1
    assert rows[0]["run_source"] == "pr"
@@ -83,6 +90,60 @@ def test_build_latest_summary_run_source_filter_keeps_canonical_baseline():
    assert rows[0]["computed_regression_status"] == "fail"


+def test_build_latest_summary_requires_absolute_floor_for_computed_regression():
+    records = [
+        _record("2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
+        _record(
+            "2026-01-02T00:00:00+00:00",
+            "b" * 40,
+            10.6,
+            10.0,
+            regression_thresholds={
+                "latency": {
+                    "threshold_percent": 0.05,
+                    "threshold_absolute": 0.75,
+                    "gated": True,
+                }
+            },
+        ),
+    ]
+
+    rows = build_latest_summary(records)
+
+    assert round(rows[0]["metrics"]["latency"]["regression_pct"], 1) == 6.0
+    assert round(rows[0]["metrics"]["latency"]["absolute_delta"], 3) == 0.6
+    assert rows[0]["metrics"]["latency"]["threshold_exceeded"] is False
+    assert rows[0]["metrics"]["latency"]["regressed"] is False
+    assert rows[0]["computed_regression_status"] == "pass"
+
+
+def test_build_latest_summary_separates_informational_threshold_crossing():
+    records = [
+        _record("2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
+        _record(
+            "2026-01-02T00:00:00+00:00",
+            "b" * 40,
+            10.6,
+            10.0,
+            regression_thresholds={
+                "latency": {
+                    "threshold_percent": 0.05,
+                    "threshold_absolute": 0.5,
+                    "gated": False,
+                }
+            },
+        ),
+    ]
+
+    rows = build_latest_summary(records)
+
+    assert rows[0]["metrics"]["latency"]["threshold_exceeded"] is True
+    assert rows[0]["metrics"]["latency"]["regressed"] is False
+    assert rows[0]["threshold_exceeded_metrics"] == ["latency"]
+    assert rows[0]["failing_metrics"] == []
+    assert rows[0]["computed_regression_status"] == "pass"
+
+
 def test_filter_records_and_trends_preserve_metric_points():
    records = [
        _record("2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
@@ -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,
@@ -309,6 +313,7 @@ def _run_benchmark(cfg):
        "max_peak_memory_mb": round(max_peak_memory, 1),
        "individual_peak_memories_mb": [round(m, 1) for m in peak_memories],
        "thresholds": thresholds,
+        "regression_thresholds": cfg.get("regression_thresholds", {}),
        "commit": os.environ.get("BUILDKITE_COMMIT", ""),
        "pr_number": os.environ.get("BUILDKITE_PULL_REQUEST", ""),
        "timestamp": datetime.now(timezone.utc).isoformat(),
@@ -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.
@@ -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>`

@@ -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())
@@ -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",
@@ -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(
@@ -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,
@@ -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,
@@ -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,
@@ -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",
@@ -31,6 +31,7 @@ device_reference_folder = resolve_device_reference_folder(
        ("L40S", "L40S"),
        ("H100", "H100"),
        ("H200", "H200"),
+        ("B200", "B200"),
        ("RTX 4090", "RTX4090"),
        ("4090", "RTX4090"),
    ),
@@ -21,6 +21,26 @@ class FakeTokenizer:
            "attention_mask": torch.ones(B, seq_len, dtype=torch.long),
        })

+
+class FakeChatTokenizer:
+    def __init__(self):
+        self.last_messages = None
+        self.last_kwargs = None
+
+    def apply_chat_template(self, messages, **kwargs):
+        self.last_messages = messages
+        self.last_kwargs = kwargs
+        assert isinstance(messages[0], list)
+        assert messages[0][0]["role"] == "system"
+        assert messages[0][1]["role"] == "user"
+        B = len(messages)
+        seq_len = int(kwargs.get("max_length", 4))
+        return TensorDict({
+            "input_ids": torch.arange(B * seq_len).view(B, seq_len),
+            "attention_mask": torch.ones(B, seq_len, dtype=torch.long),
+        })
+
+
 class FakeTextEncoder(torch.nn.Module):
    def __init__(self, hidden_size=8):
        super().__init__()
@@ -38,6 +58,14 @@ class FakeTextEncoder(torch.nn.Module):
 def id_preprocess(x: str) -> str:
    return x

+
+def chat_list_preprocess(x: str):
+    return [
+        {"role": "system", "content": "Describe the video."},
+        {"role": "user", "content": x if x else " "},
+    ]
+
+
 def take_mean_postprocess(outputs: BaseEncoderOutput) -> torch.Tensor:
    # [B, T, H] -> [B, H]
    return outputs.last_hidden_state.mean(dim=1)
@@ -156,3 +184,32 @@ def test_encode_text_does_not_force_hidden_states_for_ltx2_prefix():
    stage.encode_text("a", fastvideo_args, encoder_index=[0])

    assert stage.text_encoders[0].last_output_hidden_states is False
+
+
+def test_chat_list_preprocess_output_is_not_stripped():
+    fastvideo_args, hidden = make_args(num_encoders=1, text_len=5, hidden_size=8)
+    encoder_config = fastvideo_args.pipeline_config.text_encoder_configs[0]
+    encoder_config.is_chat_model = True
+    encoder_config.treat_empty_as_dot = True
+    fastvideo_args.pipeline_config.preprocess_text_funcs = (chat_list_preprocess, )
+
+    tokenizer = FakeChatTokenizer()
+    stage = TextEncodingStage(
+        text_encoders=[FakeTextEncoder(hidden_size=hidden)],
+        tokenizers=[tokenizer],
+    )
+
+    embeds, masks = stage.encode_text(
+        "a robotic arm welding a metal structure",
+        fastvideo_args,
+        encoder_index=[0],
+        return_attention_mask=True,
+    )
+
+    assert embeds[0].shape == (1, hidden)
+    assert masks[0].shape == (1, 5)
+    assert tokenizer.last_messages == [[
+        {"role": "system", "content": "Describe the video."},
+        {"role": "user", "content": "a robotic arm welding a metal structure"},
+    ]]
+    assert tokenizer.last_kwargs["return_tensors"] == "pt"
Author	SHA1	Message	Date
Will LinandClaude Fable 5	95dfc040bd	[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>	2026-07-05 21:19:38 +00:00
William Lin	384c1e9493	[misc]: update reseed-performance-baseline skill for the hf_store move (#1545 follow-up) (#1553 )	2026-07-05 14:16:55 -07:00
William Lin	b1dbcc93f6	[misc]: reformat fastvideo/performance to the repo yapf config (#1554 )	2026-07-05 14:16:20 -07:00
William Lin	b93833772e	[ci]: guard against test directories no CI lane collects (#1552 )	2026-07-05 14:07:38 -07:00
Mac Lee	30b523edd6	[ci] Normalize performance stage component metrics (#1475 ) (#1550 )	2026-07-05 14:05:26 -07:00
Mac LeeandSolitaryThinker	6aab7f3832	[ci] cover Hunyuan 1.5 chat-list text preprocessing (#1518 ) Co-authored-by: SolitaryThinker <wlsaidhi@gmail.com>	2026-07-05 12:05:25 -07:00
Mac Lee	9cd53fe5f8	[ci] Add metric-specific performance thresholds (#1545 )	2026-07-05 12:04:33 -07:00