[bugfix]: address verify-pass findings on the 1546 fix branch

- recipe_fingerprint also excludes attention.resolved_backend (a silent
  runtime fallback must fail the gate in its cohort, not open a fresh
  ungated one); requested_backend stays in the hash; regression test
  extended to both runtime-resolved values.
- Legacy v1 configs no longer crash after the GPU measurement: identity
  fields are emitted only for config_schema_version=2 configs, and the
  comparator falls back to the legacy (model_id, gpu_type) cohort when a
  record carries no identity fields (partial identity still raises).
- 'recipe' removed from V2_OPTIONAL_METADATA_FIELDS: the generated recipe
  owns the key now; declaring it in a config fails validation loudly
  instead of being silently overwritten.
- Docs: stale 'until those land' paragraph reconciled with what this
  branch actually lands; baseline_status documented.

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

@@ -69,7 +69,7 @@ approval, then upload reviewed accepted baseline records.
 | `model_id` | Yes | Benchmark id, e.g. `wan-t2v-1.3b-2gpu`. This maps to the HF subdirectory after `sanitize(model_id)`. |
 | `gpu_type` | Yes | Exact GPU device string from the performance record, e.g. the L40S device name emitted by CI. Baselines are GPU-specific. |
 | `source_results` | Yes | One or more local paths or Buildkite artifact URLs for accepted shifted performance JSONs. Prefer normalized `normalized_perf_*.json` artifacts emitted by `compare_baseline.py`. Accept `source_result` as an alias only for a single JSON. |
-| `max_intra_batch_regression` | No | Maximum allowed regression of any source JSON against the source batch median. Default: `0.05` (5%). |
+| `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%). |
 | `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,7 +148,8 @@ 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 `0.05`. Print a table with per-source values, batch median, and
+Default `max_intra_batch_regression` to `PERF_MAX_REGRESSION` when set,
+otherwise `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
@@ -182,7 +183,7 @@ present, that run is not a valid source for baseline reseeding.
 
 ### 2. Sync and back up existing HF records under /tmp
 
-Use `fastvideo/performance/hf_store.py` helpers directly. Do **not** use
+Use `fastvideo/tests/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.
 
@@ -191,7 +192,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}"
-python -c 'from fastvideo.performance.hf_store import sync_from_hf; import os; sync_from_hf(os.environ["PERFORMANCE_TRACKING_ROOT"], strict=True)'
+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)'
 ```
 
 Then back up only the sanitized model directory under `/tmp`:
@@ -199,8 +200,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=$(python - <<'PY'
-from fastvideo.performance.hf_store import sanitize
+MODEL_SAFE=$(PYTHONPATH=fastvideo/tests/performance python - <<'PY'
+from hf_store import sanitize
 print(sanitize("<model_id>"))
 PY
 )
@@ -234,7 +235,7 @@ first baseline seed. Continue, but report that baseline history was empty.
 Load the last 5 successful records for the target:
 
 ```python
-from fastvideo.performance.hf_store import load_records_for_model
+from hf_store import load_records_for_model
 
 records = load_records_for_model(
     "/tmp/perf-tracking",
@@ -371,7 +372,7 @@ prepared records plus backup on disk.
 Use the shared storage helper so the path and repo type match CI:
 
 ```python
-from fastvideo.performance.hf_store import upload_record
+from hf_store import upload_record
 
 upload_record("<local_record_path>", record, strict=True)
 ```
@@ -459,7 +460,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/performance/hf_store.py` — HF sync, record loading,
+- `fastvideo/tests/performance/hf_store.py` — HF sync, record loading,
   `sanitize()`, and `upload_record()`.
 - `fastvideo/tests/performance/test_inference_performance.py` — source result
   JSON schema.

apps/performance_dashboard/frontend/src/App.tsx

@@ -1,6 +1,7 @@
 import { useEffect, useMemo, useState } from "react";
 
-import { fetchSummary, fetchTrends, refreshData, RunSource, SummaryResponse, TrendGroup, TrendPoint } from "./api";
+import { fetchSummary, fetchTrends, refreshData } from "./api";
+import type { CohortValue, RunSource, SummaryResponse, TrendGroup, TrendPoint } from "./api";
 
 const METRIC_KEYS = ["latency", "throughput", "memory", "text_encoder_time_s", "dit_time_s", "vae_decode_time_s"];
 const RUN_SOURCES: Array<{ value: "" | RunSource; label: string }> = [
@@ -109,6 +110,61 @@ function metricLabel(metricKey: string) {
   return METRIC_DEFINITIONS[metricKey]?.label ?? metricKey;
 }
 
+type CohortFields = {
+  model_id: string;
+  gpu_type: string;
+  workload_id: CohortValue;
+  variant_id: CohortValue;
+  benchmark_version: CohortValue;
+  recipe_fingerprint: CohortValue;
+  hardware_profile_id: CohortValue;
+  software_profile_id: CohortValue;
+};
+
+function cohortValue(value: CohortValue) {
+  if (value === null || value === undefined || value === "") {
+    return "legacy";
+  }
+  return String(value);
+}
+
+function shortCohortValue(value: CohortValue) {
+  const text = cohortValue(value);
+  if (text === "legacy" || text.length <= 14) {
+    return text;
+  }
+  return text.slice(0, 12);
+}
+
+function cohortKey(cohort: CohortFields) {
+  return [
+    cohort.model_id,
+    cohort.gpu_type,
+    cohortValue(cohort.workload_id),
+    cohortValue(cohort.variant_id),
+    cohortValue(cohort.benchmark_version),
+    cohortValue(cohort.recipe_fingerprint),
+    cohortValue(cohort.hardware_profile_id),
+    cohortValue(cohort.software_profile_id)
+  ].join("|");
+}
+
+function cohortTitle(cohort: CohortFields) {
+  const workload = cohortValue(cohort.workload_id);
+  const variant = cohortValue(cohort.variant_id);
+  const version = cohortValue(cohort.benchmark_version);
+  const versionLabel = version === "legacy" ? version : `v${version}`;
+  return `${workload} / ${variant} / ${versionLabel}`;
+}
+
+function cohortDetail(cohort: CohortFields) {
+  return [
+    `recipe ${shortCohortValue(cohort.recipe_fingerprint)}`,
+    shortCohortValue(cohort.hardware_profile_id),
+    shortCohortValue(cohort.software_profile_id)
+  ].join(" | ");
+}
+
 function formatMetricValue(metricKey: string, value: number | null | undefined, tooltip = false) {
   const definition = METRIC_DEFINITIONS[metricKey];
   if (!definition) {
@@ -171,7 +227,9 @@ function TrendChart({ group, metricKey }: { group: TrendGroup; metricKey: string
         top: `${(activePoint.y / height) * 100}%`
       }
     : undefined;
-  const ariaLabel = `${metricLabel(metricKey)} trend for ${group.model_id} on ${group.gpu_type}`;
+  const ariaLabel = `${metricLabel(metricKey)} trend for ${group.model_id} on ${group.gpu_type}, ${cohortTitle(
+    group
+  )}`;
 
   return (
     <div className="chart-shell">
@@ -419,7 +477,7 @@ export default function App() {
       <section className="panel">
         <div className="panel-header">
           <h2>Latest Status</h2>
-          <span>{latestRows.length} model/GPU groups</span>
+          <span>{latestRows.length} comparison cohorts</span>
         </div>
         {latestRows.length === 0 ? (
           <div className="empty">No records match the selected filters.</div>
@@ -432,6 +490,7 @@ export default function App() {
                   <th>Recomputed</th>
                   <th>Model</th>
                   <th>GPU</th>
+                  <th>Cohort</th>
                   <th>Commit</th>
                   <th>Source</th>
                   <th>Baseline</th>
@@ -446,7 +505,7 @@ export default function App() {
               </thead>
               <tbody>
                 {latestRows.map((row) => (
-                  <tr key={`${row.model_id}-${row.gpu_type}`}>
+                  <tr key={cohortKey(row)}>
                     <td>
                       <span className={`badge ${row.status}`}>{row.status}</span>
                     </td>
@@ -457,6 +516,12 @@ export default function App() {
                     </td>
                     <td>{row.model_id}</td>
                     <td>{row.gpu_type}</td>
+                    <td>
+                      <div className="cohort-cell">
+                        <strong>{cohortTitle(row)}</strong>
+                        <span>{cohortDetail(row)}</span>
+                      </div>
+                    </td>
                     <td>{shortSha(row.commit_sha)}</td>
                     <td>
                       <span className={`source-badge source-${row.run_source}`}>{runSourceLabel(row.run_source)}</span>
@@ -495,11 +560,13 @@ export default function App() {
           ) : (
             trends.map((group) =>
               METRIC_KEYS.map((metricKey) => (
-                <article className="trend-card" key={`${group.model_id}-${group.gpu_type}-${metricKey}`}>
+                <article className="trend-card" key={`${cohortKey(group)}-${metricKey}`}>
                   <div>
                     <h3>{metricLabel(metricKey)}</h3>
                     <p>
                       {group.model_id} | {group.gpu_type}
+                      <span>{cohortTitle(group)}</span>
+                      <span>{cohortDetail(group)}</span>
                     </p>
                   </div>
                   <TrendChart group={group} metricKey={metricKey} />

apps/performance_dashboard/frontend/src/api.ts

@@ -13,6 +13,17 @@ export type MetricValue = {
   precision: number;
 };
 
+export type CohortValue = string | number | null;
+
+export type ComparisonCohort = {
+  workload_id: CohortValue;
+  variant_id: CohortValue;
+  benchmark_version: CohortValue;
+  recipe_fingerprint: CohortValue;
+  hardware_profile_id: CohortValue;
+  software_profile_id: CohortValue;
+};
+
 export type SummaryRow = {
   model_id: string;
   gpu_type: string;
@@ -34,7 +45,7 @@ export type SummaryRow = {
   build_id: string;
   job_id: string;
   metrics: Record<string, MetricValue>;
-};
+} & ComparisonCohort;
 
 export type RunSource = "pr" | "local" | "scheduled_main" | "unknown";
 
@@ -68,13 +79,13 @@ export type TrendPoint = {
   build_id: string;
   job_id: string;
   metrics: Record<string, number | null>;
-};
+} & ComparisonCohort;
 
 export type TrendGroup = {
   model_id: string;
   gpu_type: string;
   points: TrendPoint[];
-};
+} & ComparisonCohort;
 
 export type TrendsResponse = {
   groups: TrendGroup[];

apps/performance_dashboard/frontend/src/styles.css

@@ -149,6 +149,11 @@ h3 {
   font-size: 0.82rem;
 }
 
+.trend-card p {
+  display: grid;
+  gap: 2px;
+}
+
 .stat strong {
   display: block;
   margin-top: 8px;
@@ -186,7 +191,7 @@ h3 {
 
 table {
   width: 100%;
-  min-width: 1120px;
+  min-width: 1260px;
   border-collapse: collapse;
 }
 
@@ -209,6 +214,25 @@ td {
   font-size: 0.9rem;
 }
 
+.cohort-cell {
+  display: grid;
+  gap: 2px;
+}
+
+.cohort-cell strong,
+.trend-card p span {
+  color: #1b2836;
+  font-size: 0.78rem;
+  font-weight: 700;
+}
+
+.cohort-cell span,
+.trend-card p span + span {
+  color: #607080;
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", monospace;
+  font-size: 0.72rem;
+}
+
 .badge {
   display: inline-flex;
   align-items: center;

docs/contributing/performance_benchmarks.md

@@ -112,11 +112,10 @@ per-metric policy with direction, percent threshold, absolute threshold, and a
 while it runs so pipeline stage execution times are available in
 `generate_video(...).logging_info`. Stage logs use pipeline-unique keys such as
 `prompt_encoding_stage` so duplicate stage classes do not collide. For
-`PipelineStage` entries, 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
+`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
 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
@@ -222,13 +221,17 @@ If a config declares `config_schema_version: 2`, loading fails clearly when any
 required v2 identity field is missing. If v2 identity or metadata fields are
 added without `config_schema_version: 2`, loading also fails so partial
 migrations do not silently run as v1 configs. Optional v2 metadata fields
-reserved for follow-up work, such as `recipe`, `metric_threshold_policy`, and
-`quality_metadata`, must be JSON objects when present.
+reserved for follow-up work, such as `metric_threshold_policy` and
+`quality_metadata`, must be JSON objects when present. (`recipe` is emitted
+by the harness and is not config-declarable.)
 
 Recipe fingerprinting, hardware/software profile IDs, exact-identity
-comparison, metric-specific threshold policy behavior, promoted baselines, and
-dashboard regrouping are separate follow-up changes. Until those land, rolling
-baseline comparison remains keyed by `(model_id, gpu_type)`.
+comparison, and dashboard cohort grouping land with this change: v2 records
+compare only within their identity cohort, and a record that opens a NEW
+cohort is marked `baseline_status: "initialized_new_cohort"` (regression
+gating starts once that cohort accumulates history). Legacy v1 configs keep
+the `(model_id, gpu_type)` rolling-baseline comparison. Metric-specific
+threshold policies and promoted baselines remain separate follow-ups.
 
 ### Raw record (`results/perf_*.json`)
 
@@ -270,7 +273,43 @@ Written by `test_inference_performance.py`. One file per benchmark run.
   "timestamp": "2026-05-08T22:00:00+00:00",
   "text_encoder_time_s": 2.141,
   "dit_time_s": 8.437,
-  "vae_decode_time_s": 3.208
+  "vae_decode_time_s": 3.208,
+  "recipe": {
+    "recipe_schema_version": 1,
+    "benchmark": {
+      "benchmark_id": "wan-t2v-1.3b-2gpu",
+      "workload_id": "wan-t2v-1.3b",
+      "variant_id": "canonical",
+      "benchmark_version": 1
+    },
+    "model": { "model_path": "Wan-AI/Wan2.1-T2V-1.3B-Diffusers" },
+    "init_kwargs": { "num_gpus": 2, "sp_size": 2, "tp_size": 1 },
+    "generation_kwargs": { "height": 480, "width": 832, "num_frames": 45 },
+    "inputs": { "prompt_count": 1, "prompt_sha256": ["<measured-prompt-sha256>"] },
+    "attention": { "requested_backend": "FLASH_ATTN", "resolved_backend": "FLASH_ATTN" }
+  },
+  "recipe_fingerprint": "<sha256>",
+  "hardware_profile": {
+    "device_type": "cuda",
+    "gpu_count": 2,
+    "gpus": [{ "name": "NVIDIA L40S", "memory_gb": 48, "compute_capability": "8.9" }],
+    "interconnect": "none_or_partial"
+  },
+  "hardware_profile_id": "hw-<sha256-prefix>",
+  "software_profile": {
+    "python": "3.12",
+    "pytorch": "2.12",
+    "cuda": "13.0",
+    "packages": {
+      "fastvideo_kernel": "0.3.2",
+      "flashinfer": "0.2.11",
+      "nvidia_cutlass_dsl": "4.5.0",
+      "triton": "3.4.1"
+    }
+  },
+  "software_profile_id": "sw-<sha256-prefix>",
+  "environment_metadata": { "env": { "IMAGE_VERSION": "py3.12-cuda13.0.0" } },
+  "environment_fingerprint": "env-<sha256-prefix>"
 }
 ```
 
@@ -282,6 +321,9 @@ result, used as the rolling-baseline source of truth.
 ```jsonc
 {
   "model_id": "wan-t2v-1.3b-2gpu",
+  "workload_id": "wan-t2v-1.3b",
+  "variant_id": "canonical",
+  "benchmark_version": 1,
   "timestamp": "2026-05-08T22:00:00+00:00",
   "commit_sha": "<full sha>",
   "gpu_type": "NVIDIA L40S",
@@ -298,6 +340,10 @@ result, used as the rolling-baseline source of truth.
       "gated": true
     }
   },
+  "recipe_fingerprint": "<sha256>",
+  "hardware_profile_id": "hw-<sha256-prefix>",
+  "software_profile_id": "sw-<sha256-prefix>",
+  "environment_fingerprint": "env-<sha256-prefix>",
   "success": true
 }
 ```
@@ -310,6 +356,18 @@ dashboard lists skipped plots for metric series that have no non-null values.
 Records missing both `run_source` and `baseline_eligible` are treated as legacy
 successful main/full-suite uploads and remain eligible for rolling baselines.
 
+New records compare only against the same `model_id`, `gpu_type`,
+`workload_id`, `variant_id`, `benchmark_version`, `recipe_fingerprint`,
+`hardware_profile_id`, and `software_profile_id` cohort.
+`environment_metadata` and `environment_fingerprint` are audit data and are not
+part of the comparison key.
+The recipe prompt digests describe the prompts actually measured by the
+benchmark run; extra configured prompts are ignored unless the benchmark runner
+executes them.
+Software profile package cohorts keep exact versions for relevant
+attention/kernel packages, including FastVideo kernels, FlashAttention,
+FlashInfer, Cutlass DSL, SageAttention, Triton, and xFormers when installed.
+
 ## Environment variable reference
 
 | Variable | Default | Used by | Purpose |
@@ -391,7 +449,9 @@ When the rolling-baseline phase runs, it emits:
 
    Legacy v1 configs without `config_schema_version` still load, but should not
    gain v2 identity or metadata fields until they are migrated to
-   `config_schema_version: 2`.
+   `config_schema_version: 2`. For v2 configs, `workload_id`, `variant_id`,
+   and `benchmark_version` are part of the comparison key; benchmark runs
+   fail if any of these identity fields are missing.
 
 2. The pytest test auto-discovers all configs — no test code needed. CI
    picks it up on the next `/test performance` run.
@@ -433,5 +493,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 emits `component_metric` or is covered by the legacy
-`STAGE_METRIC_MAP` fallback in `test_inference_performance.py`.
+and that the stage name is listed in `STAGE_METRIC_MAP` in
+`test_inference_performance.py`.

fastvideo/performance/hf_store.py

@@ -268,16 +268,31 @@ def load_records_for_model(
     model_id: str,
     gpu_type: str | None = None,
     *,
+    workload_id: str | None = None,
+    variant_id: str | None = None,
+    benchmark_version: str | None = None,
+    recipe_fingerprint: str | None = None,
+    hardware_profile_id: str | None = None,
+    software_profile_id: str | None = None,
     last_n: int | None = None,
     successful_only: bool = True,
     baseline_eligible_only: bool = False,
 ) -> list[dict[str, Any]]:
-    """Return records for a specific *model_id*, optionally filtered by GPU.
+    """Return records for a specific *model_id*, optionally filtered by cohort.
 
     Args:
         local_dir: Root directory previously populated by :func:`sync_from_hf`.
         model_id: Matches the ``model_id`` field inside each JSON record.
         gpu_type: When set, only records whose ``gpu_type`` matches are returned.
+        workload_id: When set, only records from the same workload are returned.
+        variant_id: When set, only records from the same workload variant are returned.
+        benchmark_version: When set, only records from the same benchmark version are returned.
+        recipe_fingerprint: When set, only records from the same benchmark
+            recipe are returned.
+        hardware_profile_id: When set, only records from the same hardware
+            cohort are returned.
+        software_profile_id: When set, only records from the same software
+            cohort are returned.
         last_n: When set, return only the most recent *n* records (after all
             other filters). Useful for sliding-window baseline calculations.
         successful_only: Passed through to :func:`load_records`.
@@ -299,6 +314,18 @@ def load_records_for_model(
     if gpu_type is not None:
         records = [r for r in records if r.get("gpu_type") == gpu_type]
 
+    identity_filters = {
+        "workload_id": workload_id,
+        "variant_id": variant_id,
+        "benchmark_version": benchmark_version,
+        "recipe_fingerprint": recipe_fingerprint,
+        "hardware_profile_id": hardware_profile_id,
+        "software_profile_id": software_profile_id,
+    }
+    for key, expected in identity_filters.items():
+        if expected is not None:
+            records = [r for r in records if str(r.get(key)) == str(expected)]
+
     if last_n is not None:
         records = records[-last_n:]
 
@@ -310,12 +337,8 @@ def load_records_for_model(
 # ---------------------------------------------------------------------------
 
 _NUMERIC_COLS = (
-    "latency",
-    "throughput",
-    "memory",
-    "text_encoder_time_s",
-    "dit_time_s",
-    "vae_decode_time_s",
+    "latency", "throughput", "memory",
+    "text_encoder_time_s", "dit_time_s", "vae_decode_time_s",
 )
 
 

fastvideo/performance/metric_policy.py

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

fastvideo/performance_dashboard/service.py

@@ -17,6 +17,15 @@ from fastvideo.performance.hf_store import is_baseline_eligible_record, safe_flo
 from fastvideo.performance.metric_policy import regression_delta, resolve_metric_policies
 
 Record = dict[str, Any]
+CohortKey = tuple[str, str, str, str, str, str, str, str]
+COMPARISON_COHORT_KEYS = (
+    "workload_id",
+    "variant_id",
+    "benchmark_version",
+    "recipe_fingerprint",
+    "hardware_profile_id",
+    "software_profile_id",
+)
 
 
 def parse_timestamp(value: Any) -> datetime | None:
@@ -77,15 +86,44 @@ def record_metadata(record: Record) -> Record:
     }
 
 
-def group_by_model_gpu(records: list[Record]) -> dict[tuple[str, str], list[Record]]:
-    groups: dict[tuple[str, str], list[Record]] = defaultdict(list)
+def record_comparison_metadata(record: Record) -> Record:
+    return {key: record.get(key) or "" for key in COMPARISON_COHORT_KEYS}
+
+
+def comparison_cohort_key(record: Record) -> CohortKey:
+    return (
+        str(record.get("model_id") or "unknown"),
+        str(record.get("gpu_type") or "unknown"),
+        str(record.get("workload_id") or ""),
+        str(record.get("variant_id") or ""),
+        str(record.get("benchmark_version") or ""),
+        str(record.get("recipe_fingerprint") or ""),
+        str(record.get("hardware_profile_id") or ""),
+        str(record.get("software_profile_id") or ""),
+    )
+
+
+def group_by_comparison_cohort(records: list[Record]) -> dict[CohortKey, list[Record]]:
+    groups: dict[CohortKey, list[Record]] = defaultdict(list)
     for record in records:
-        model_id = str(record.get("model_id") or "unknown")
-        gpu_type = str(record.get("gpu_type") or "unknown")
-        groups[(model_id, gpu_type)].append(record)
+        groups[comparison_cohort_key(record)].append(record)
     return {key: sorted(value, key=record_sort_key) for key, value in groups.items()}
 
 
+def comparison_sort_key(record: Record) -> CohortKey:
+    return comparison_cohort_key(record)
+
+
+def latest_row_sort_key(row: Record) -> tuple[Any, ...]:
+    return (row["status"] != "fail", *comparison_sort_key(row))
+
+
+def group_identity(key: CohortKey) -> tuple[str, str]:
+    model_id = key[0]
+    gpu_type = key[1]
+    return model_id, gpu_type
+
+
 def baseline_value(records: list[Record], metric_key: str) -> float | None:
     values = [safe_float(record.get(metric_key)) for record in records]
     values = [value for value in values if value is not None]
@@ -99,7 +137,8 @@ def build_latest_summary(records: list[Record],
                          baseline_window: int = 5,
                          run_source: str | None = None) -> list[Record]:
     rows: list[Record] = []
-    for (model_id, gpu_type), group in group_by_model_gpu(records).items():
+    for key, group in group_by_comparison_cohort(records).items():
+        model_id, gpu_type = group_identity(key)
         latest_candidates = group
         if run_source:
             latest_candidates = [record for record in group if record_run_source(record) == run_source]
@@ -156,6 +195,7 @@ def build_latest_summary(records: list[Record],
             "timestamp": latest.get("timestamp"),
             "commit_sha": latest.get("commit_sha"),
             **record_metadata(latest),
+            **record_comparison_metadata(latest),
             "success": success,
             "baseline_n": len(baseline_records),
             "worst_regression_pct": worst_regression,
@@ -166,12 +206,13 @@ def build_latest_summary(records: list[Record],
             "metrics": metrics,
         })
 
-    return sorted(rows, key=lambda row: (row["status"] != "fail", row["model_id"], row["gpu_type"]))
+    return sorted(rows, key=latest_row_sort_key)
 
 
 def build_trends(records: list[Record]) -> list[Record]:
     trends: list[Record] = []
-    for (model_id, gpu_type), group in group_by_model_gpu(records).items():
+    for key, group in group_by_comparison_cohort(records).items():
+        model_id, gpu_type = group_identity(key)
         points = []
         for record in group:
             metric_policies = resolve_metric_policies(record.get("regression_thresholds"))
@@ -179,6 +220,7 @@ def build_trends(records: list[Record]) -> list[Record]:
                 "timestamp": record.get("timestamp"),
                 "commit_sha": record.get("commit_sha"),
                 **record_metadata(record),
+                **record_comparison_metadata(record),
                 "success": bool(record.get("success", True)),
                 "metrics": {
                     policy.key: safe_float(record.get(policy.key))
@@ -189,6 +231,7 @@ def build_trends(records: list[Record]) -> list[Record]:
         trends.append({
             "model_id": model_id,
             "gpu_type": gpu_type,
+            **record_comparison_metadata(group[-1]),
             "points": points,
         })
-    return sorted(trends, key=lambda trend: (trend["model_id"], trend["gpu_type"]))
+    return sorted(trends, key=comparison_sort_key)

fastvideo/pipelines/stages/base.py

@@ -34,7 +34,6 @@ 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:
         """
@@ -156,9 +155,6 @@ class PipelineStage(ABC):
                 logger.info("[%s] Execution completed in %s ms", stage_name, execution_time * 1000)
                 batch.logging_info.add_stage_execution_time(stage_key, execution_time)
                 batch.logging_info.add_stage_metric(stage_key, "stage_class", stage_class_name)
-                component_metric = self.performance_component_metric
-                if component_metric is not None:
-                    batch.logging_info.add_stage_metric(stage_key, "component_metric", component_metric)
             except Exception as e:
                 torch.cuda.synchronize()
                 execution_time = time.perf_counter() - start_time

fastvideo/pipelines/stages/decoding.py

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

fastvideo/pipelines/stages/denoising.py

@@ -51,7 +51,6 @@ 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__()
@@ -1188,7 +1187,6 @@ class Cosmos25V2WDenoisingStage(Cosmos25DenoisingStage):
 
 class Cosmos25AutoDenoisingStage(PipelineStage):
     """Route Cosmos 2.5 denoising to T2W vs V2W/I2W."""
-    performance_component_metric = "dit_time_s"
 
     def __init__(self, transformer, scheduler) -> None:
         super().__init__()

fastvideo/pipelines/stages/text_encoding.py

@@ -24,7 +24,6 @@ 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:
         """
@@ -351,7 +350,6 @@ class Cosmos25TextEncodingStage(PipelineStage):
     Cosmos 2.5 uses Reason1 (Qwen2.5-VL) and relies on the encoder's
     `compute_text_embeddings_online()`.
     """
-    performance_component_metric = "text_encoder_time_s"
 
     def __init__(self, text_encoder) -> None:
         super().__init__()

fastvideo/tests/contract/test_ci_test_collection.py

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

fastvideo/tests/performance/compare_baseline.py

@@ -5,7 +5,8 @@ This script:
 1) reads current benchmark results from fastvideo/tests/performance/results,
 2) syncs the canonical baseline from the configured HF dataset repo,
 3) compares each current record against the median of up to 5 prior
-   baseline-eligible successful records (filtered by gpu_type),
+   baseline-eligible successful records in the same workload/variant/version,
+   GPU, recipe, hardware, and software cohort,
 4) writes normalized records back to the HF dataset repo according to
    PERF_UPLOAD_POLICY,
 5) exits non-zero if any gated metric exceeds both its percent and absolute
@@ -64,6 +65,27 @@ PERF_REPORTS_DIR = os.environ.get("PERF_REPORTS_DIR", "/root/data/perf_reports")
 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"}
+IDENTITY_KEYS = (
+    "workload_id",
+    "variant_id",
+    "benchmark_version",
+    "recipe",
+    "recipe_fingerprint",
+    "hardware_profile",
+    "hardware_profile_id",
+    "software_profile",
+    "software_profile_id",
+    "environment_metadata",
+    "environment_fingerprint",
+)
+COMPARISON_IDENTITY_KEYS = (
+    "workload_id",
+    "variant_id",
+    "benchmark_version",
+    "recipe_fingerprint",
+    "hardware_profile_id",
+    "software_profile_id",
+)
 
 
 def _should_persist_tracking() -> bool:
@@ -136,6 +158,46 @@ def _record_metadata(run_source: str, result: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _identity_metadata(result: dict[str, Any]) -> dict[str, Any]:
+    metadata = {
+        key: result[key]
+        for key in IDENTITY_KEYS
+        if key in result and result[key] is not None
+    }
+    recipe = result.get("recipe")
+    benchmark = recipe.get("benchmark") if isinstance(recipe, dict) else None
+    if isinstance(benchmark, dict):
+        for key in ("workload_id", "variant_id", "benchmark_version"):
+            if key not in metadata and benchmark.get(key) is not None:
+                metadata[key] = benchmark[key]
+    return metadata
+
+
+def _comparison_identity_filters(record: dict[str, Any]) -> dict[str, str]:
+    missing = [
+        key for key in COMPARISON_IDENTITY_KEYS
+        if key not in record or record[key] is None or (isinstance(record[key], str) and not record[key].strip())
+    ]
+    if len(missing) == len(COMPARISON_IDENTITY_KEYS):
+        # Legacy v1 record: no identity fields at all. Keep the pre-v2
+        # (model_id, gpu_type) rolling-baseline comparison instead of
+        # crashing — v1 configs are documented as loadable.
+        return {}
+    if missing:
+        raise ValueError("Performance record missing required comparison identity fields: " + ", ".join(missing))
+    return {
+        key: str(record[key])
+        for key in COMPARISON_IDENTITY_KEYS
+    }
+
+
+def _format_identity_filters(filters: dict[str, str]) -> str:
+    if not filters:
+        return ""
+    compact = ", ".join(f"{key}={value}" for key, value in filters.items())
+    return f" ({compact})"
+
+
 def _load_current_results() -> list[dict[str, Any]]:
     pattern = os.path.join(RESULTS_DIR, "perf_*.json")
     records: list[dict[str, Any]] = []
@@ -169,7 +231,7 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
     vae_decode_time = safe_float(result.get("vae_decode_time_s"))
     metric_policies = resolve_metric_policies(result.get("regression_thresholds"))
 
-    return {
+    record = {
         "model_id": model_id,
         "timestamp": timestamp,
         "commit_sha": commit_sha,
@@ -184,6 +246,8 @@ def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]:
         "success": True,
         **_record_metadata(_detect_run_source(), result),
     }
+    record.update(_identity_metadata(result))
+    return record
 
 
 def _normalize_record(result: dict[str, Any]) -> dict[str, Any]:
@@ -422,17 +486,31 @@ def main() -> int:
             TRACKING_ROOT,
             record["model_id"],
             record["gpu_type"],
+            **_comparison_identity_filters(record),
             last_n=5,
             successful_only=True,
             baseline_eligible_only=True,
         )
 
         if not baseline_records:
-            print(f"No baseline for {record['model_id']} on "
-                  f"{record['gpu_type']}. Initializing...")
+            # A brand-new cohort has NO regression gating until history
+            # accumulates — make that loud and machine-readable instead of
+            # an indistinguishable pass, so a cohort shift (intended or
+            # accidental, e.g. an identity-field change) never silently
+            # blinds the comparison.
+            record["baseline_status"] = "initialized_new_cohort"
+            print("=" * 72)
+            print(f"WARNING: NO BASELINE — initializing a NEW cohort for "
+                  f"{record['model_id']} on {record['gpu_type']}"
+                  f"{_format_identity_filters(_comparison_identity_filters(record))}")
+            print("Regression gating is INACTIVE for this cohort until "
+                  "baseline history accumulates. If this cohort shift is "
+                  "unexpected, check the identity fields above.")
+            print("=" * 72)
             failures: list[str] = []
             record["success"] = True
         else:
+            record["baseline_status"] = "compared"
             failures = _check_regressions(record, baseline_records, metric_policies)
         if static_threshold_failed:
             failures.append(f"{record['model_id']} fixed-threshold phase failed "

fastvideo/tests/performance/dashboard.py

@@ -29,15 +29,76 @@ METRICS = (
     "dit_time_s",
     "vae_decode_time_s",
 )
+COMPARISON_COHORT_KEYS = (
+    "workload_id",
+    "variant_id",
+    "benchmark_version",
+    "recipe_fingerprint",
+    "hardware_profile_id",
+    "software_profile_id",
+)
+GROUP_KEYS = ("model_id", "gpu_type", *COMPARISON_COHORT_KEYS)
+
+
+def _cohort_value(value: object) -> str:
+    if value is None:
+        return ""
+    try:
+        if pd.isna(value):
+            return ""
+    except (TypeError, ValueError):
+        pass
+    return str(value)
+
+
+def _display_value(value: object) -> str:
+    return _cohort_value(value) or "legacy"
+
+
+def _short_value(value: object) -> str:
+    text = _display_value(value)
+    if text == "legacy" or len(text) <= 14:
+        return text
+    return text[:12]
+
+
+def _cohort_title(group_key: tuple[object, ...]) -> str:
+    workload = _display_value(group_key[2])
+    variant = _display_value(group_key[3])
+    version = _display_value(group_key[4])
+    version_label = version if version == "legacy" else f"v{version}"
+    return f"{workload} / {variant} / {version_label}"
+
+
+def _cohort_detail(group_key: tuple[object, ...]) -> str:
+    return " | ".join((
+        f"recipe {_short_value(group_key[5])}",
+        _short_value(group_key[6]),
+        _short_value(group_key[7]),
+    ))
+
+
+def _group_record(group_key: tuple[object, ...]) -> dict[str, str]:
+    return {
+        key: _cohort_value(value)
+        for key, value in zip(GROUP_KEYS, group_key)
+    }
+
+
+def _dashboard_frame(df: pd.DataFrame) -> pd.DataFrame:
+    dashboard_df = df.copy()
+    for key in GROUP_KEYS:
+        if key not in dashboard_df.columns:
+            dashboard_df[key] = ""
+        dashboard_df[key] = dashboard_df[key].map(_cohort_value)
+    return dashboard_df
 
 # -----------------------------
 # 1. Grouping
 # -----------------------------
 def group_data(df: pd.DataFrame):
-    # Group only by model+GPU so each group produces a time-series line.
-    # config_id (commit SHA) is carried as a column for hover/color use.
-    keys = ["model_id", "gpu_type"]
-    return df.groupby(keys, dropna=False)
+    # Group by the same comparison cohort used by baseline gating.
+    return _dashboard_frame(df).groupby(list(GROUP_KEYS), dropna=False)
 
 # -----------------------------
 # 2. Plot builder
@@ -46,15 +107,20 @@ def build_plots(df: pd.DataFrame) -> tuple[list, list[dict[str, object]]]:
     figs = []
     skipped_metrics: list[dict[str, object]] = []
 
-    for (model_id, gpu_type), g in group_data(df):
+    for group_key, g in group_data(df):
+        model_id, gpu_type = group_key[:2]
+        group_record = _group_record(group_key)
+        cohort_title = _cohort_title(group_key)
+        cohort_detail = _cohort_detail(group_key)
         g = g.sort_values("timestamp")
 
         # One chart per metric so the y-axes aren't on wildly different scales
         for metric in METRICS:
             if metric not in g.columns:
                 skipped_metrics.append({
-                    "model_id": model_id,
-                    "gpu_type": gpu_type,
+                    **group_record,
+                    "cohort": cohort_title,
+                    "cohort_detail": cohort_detail,
                     "metric": metric,
                     "reason": "column missing from loaded records",
                     "records": len(g),
@@ -65,8 +131,9 @@ def build_plots(df: pd.DataFrame) -> tuple[list, list[dict[str, object]]]:
             non_null = int(g[metric].notna().sum())
             if non_null == 0:
                 skipped_metrics.append({
-                    "model_id": model_id,
-                    "gpu_type": gpu_type,
+                    **group_record,
+                    "cohort": cohort_title,
+                    "cohort_detail": cohort_detail,
                     "metric": metric,
                     "reason": "no non-null values in loaded records",
                     "records": len(g),
@@ -79,8 +146,8 @@ def build_plots(df: pd.DataFrame) -> tuple[list, list[dict[str, object]]]:
                 x="timestamp",
                 y=metric,
                 markers=True,
-                hover_data=["config_id", "commit_sha"],
-                title=f"{model_id} | {gpu_type} | {metric}",
+                hover_data=["config_id", "commit_sha", *COMPARISON_COHORT_KEYS],
+                title=f"{model_id} | {gpu_type} | {cohort_title} | {cohort_detail} | {metric}",
                 labels={"timestamp": "Time", metric: metric},
             )
             figs.append(fig)
@@ -95,7 +162,7 @@ def render_skipped_metrics(skipped_metrics: list[dict[str, object]]) -> str:
     rows = [
         "<h3>Skipped Metric Plots</h3>",
         "<table>",
-        ("<thead><tr><th>Model</th><th>GPU</th><th>Metric</th>"
+        ("<thead><tr><th>Model</th><th>GPU</th><th>Cohort</th><th>Metric</th>"
          "<th>Records</th><th>Non-null</th><th>Reason</th></tr></thead>"),
         "<tbody>",
     ]
@@ -104,6 +171,7 @@ def render_skipped_metrics(skipped_metrics: list[dict[str, object]]) -> str:
             "<tr>"
             f"<td>{escape(str(item['model_id']))}</td>"
             f"<td>{escape(str(item['gpu_type']))}</td>"
+            f"<td>{escape(str(item['cohort']))}<br><code>{escape(str(item['cohort_detail']))}</code></td>"
             f"<td>{escape(str(item['metric']))}</td>"
             f"<td>{item['records']}</td>"
             f"<td>{item['non_null']}</td>"
@@ -165,6 +233,7 @@ def main() -> None:
         print("Skipped metric plots:")
         for item in skipped_metrics:
             print(f"  - {item['model_id']} | {item['gpu_type']} | "
+                  f"{item['cohort']} | {item['cohort_detail']} | "
                   f"{item['metric']}: {item['reason']} "
                   f"({item['non_null']}/{item['records']} non-null)")
     print(f"Generated {len(figs)} metric plot(s)")

fastvideo/tests/performance/identity.py

@@ -0,0 +1,435 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Comparable identity helpers for performance benchmark records."""
+
+from __future__ import annotations
+
+import dataclasses
+import enum
+import hashlib
+import importlib.metadata
+import json
+import os
+import platform
+import re
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Any
+
+import torch
+
+RECIPE_SCHEMA_VERSION = 1
+PROFILE_ID_LENGTH = 16
+_PATH_EXCLUDED_GENERATION_KEYS = {"output_path", "output_video_name"}
+_PROMPT_KEYS = {"prompt", "negative_prompt", "neg_prompt"}
+REQUIRED_BENCHMARK_IDENTITY_KEYS = (
+    "benchmark_id",
+    "workload_id",
+    "variant_id",
+    "benchmark_version",
+)
+_PROFILE_ENV_VARS = (
+    "CUDA_VISIBLE_DEVICES",
+    "FASTVIDEO_ATTENTION_BACKEND",
+    "IMAGE_VERSION",
+    "UV_TORCH_BACKEND",
+)
+_PACKAGE_DISTRIBUTIONS = {
+    "fastvideo_kernel": ("fastvideo-kernel", "fastvideo_kernel"),
+    "flash_attn": ("flash-attn", "flash_attn"),
+    "flash_attn_4": ("flash-attn-4",),
+    "flash_attention_fp4": ("flash-attention-fp4",),
+    "flashinfer": ("flashinfer-python", "flashinfer"),
+    "nvidia_cutlass_dsl": ("nvidia-cutlass-dsl", "cutlass-dsl"),
+    "sageattention": ("sageattention",),
+    "sageattn3": ("sageattn3",),
+    "triton": ("triton",),
+    "xformers": ("xformers",),
+}
+
+
+def canonical_json(payload: Any) -> str:
+    """Return deterministic, compact JSON for a recipe/profile payload."""
+
+    return json.dumps(
+        _canonicalize(payload),
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+    )
+
+
+def sha256_hexdigest(payload: str | bytes) -> str:
+    if isinstance(payload, str):
+        payload = payload.encode("utf-8")
+    return hashlib.sha256(payload).hexdigest()
+
+
+def recipe_fingerprint(recipe: Mapping[str, Any]) -> str:
+    # Runtime-resolved values are audit metadata, not identity: hashing the
+    # resolved HF snapshot revision would churn the cohort (and via the
+    # no-baseline init path, silently reset the rolling baseline) on ANY
+    # upstream repo commit, including model-card edits with unchanged
+    # weights. The fingerprint is pinned to declared inputs only; the
+    # resolved values stay in the stored recipe for auditability.
+    pruned = {
+        key: (dict(value) if isinstance(value, Mapping) else value)
+        for key, value in recipe.items()
+    }
+    model = pruned.get("model")
+    if isinstance(model, dict):
+        model.pop("resolved_revision", None)
+    attention = pruned.get("attention")
+    if isinstance(attention, dict):
+        # Same reasoning as resolved_revision: with requested_backend="auto",
+        # a silent runtime fallback (e.g. a broken flash-attn wheel) must NOT
+        # open a fresh ungated cohort and seed degraded numbers as the new
+        # baseline — it should fail the regression gate in the old cohort.
+        # Intentional backend changes are declared via requested_backend,
+        # which stays in the hash.
+        attention.pop("resolved_backend", None)
+    return sha256_hexdigest(canonical_json(pruned))
+
+
+def profile_id(prefix: str, profile: Mapping[str, Any]) -> str:
+    return f"{prefix}-{sha256_hexdigest(canonical_json(profile))[:PROFILE_ID_LENGTH]}"
+
+
+def hardware_profile_id(profile: Mapping[str, Any]) -> str:
+    return profile_id("hw", profile)
+
+
+def software_profile_id(profile: Mapping[str, Any]) -> str:
+    return profile_id("sw", profile)
+
+
+def environment_fingerprint(metadata: Mapping[str, Any]) -> str:
+    return profile_id("env", metadata)
+
+
+def benchmark_identity_from_config(cfg: Mapping[str, Any]) -> dict[str, Any]:
+    missing = [
+        key for key in REQUIRED_BENCHMARK_IDENTITY_KEYS
+        if _none_if_empty(cfg.get(key)) is None
+    ]
+    if missing:
+        raise ValueError("Benchmark config missing required identity fields: " + ", ".join(missing))
+    return {
+        key: cfg[key]
+        for key in REQUIRED_BENCHMARK_IDENTITY_KEYS
+    }
+
+
+def build_recipe_from_benchmark_config(
+    cfg: Mapping[str, Any],
+    *,
+    attention_backend: str | None = None,
+    resolved_attention_backend: Any | None = None,
+    resolved_model_revision: Any | None = None,
+    measured_prompts: Sequence[Any] | None = None,
+) -> dict[str, Any]:
+    """Build a deterministic recipe document from a benchmark config.
+
+    The recipe captures fields that describe what benchmark workload was run. It
+    intentionally excludes timings, timestamps, commit metadata, and output paths.
+    """
+
+    model = dict(cfg.get("model") or {})
+    init_kwargs = dict(cfg.get("init_kwargs") or {})
+    generation_kwargs = dict(cfg.get("generation_kwargs") or {})
+    benchmark_identity = benchmark_identity_from_config(cfg)
+    prompts = list(measured_prompts) if measured_prompts is not None else list(
+        cfg.get("test_prompts") or ["A cinematic video."])
+    if attention_backend is None:
+        attention_backend = os.environ.get("FASTVIDEO_ATTENTION_BACKEND")
+
+    negative_prompt = generation_kwargs.pop("negative_prompt", generation_kwargs.pop("neg_prompt", None))
+    generation_recipe = {
+        key: value
+        for key, value in generation_kwargs.items()
+        if key not in _PATH_EXCLUDED_GENERATION_KEYS and key not in _PROMPT_KEYS
+    }
+
+    return {
+        "recipe_schema_version": RECIPE_SCHEMA_VERSION,
+        "benchmark": benchmark_identity,
+        "model": {
+            "model_path": normalize_model_path(model.get("model_path")),
+            "revision": _none_if_empty(model.get("revision") or cfg.get("revision")),
+            "resolved_revision": _none_if_empty(resolved_model_revision),
+        },
+        "init_kwargs": init_kwargs,
+        "generation_kwargs": generation_recipe,
+        "inputs": {
+            "prompt_count": len(prompts),
+            "prompt_sha256": [_optional_digest(prompt) for prompt in prompts],
+            "negative_prompt_sha256": _optional_digest(negative_prompt),
+        },
+        "attention": {
+            "requested_backend": _none_if_empty(attention_backend) or "auto",
+            "resolved_backend": _none_if_empty(resolved_attention_backend),
+        },
+    }
+
+
+def normalize_model_path(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = os.fspath(value) if isinstance(value, os.PathLike) else str(value)
+    text = text.strip()
+    if not text:
+        return None
+    if _looks_like_local_path(text):
+        return Path(text).expanduser().as_posix()
+    return re.sub(r"/+", "/", text)
+
+
+def resolved_revision_from_model_path(value: Any) -> str | None:
+    normalized = normalize_model_path(value)
+    if normalized is None:
+        return None
+    parts = Path(normalized).parts
+    for index, part in enumerate(parts[:-1]):
+        if part == "snapshots":
+            revision = parts[index + 1]
+            if re.fullmatch(r"[0-9a-f]{40}", revision):
+                return revision
+    return None
+
+
+def hardware_profile(
+    *,
+    num_gpus: int | None = None,
+    gpu_devices: Sequence[Mapping[str, Any]] | None = None,
+    device_type: str | None = None,
+    interconnect: str | None = None,
+) -> dict[str, Any]:
+    """Return normalized hardware cohort metadata.
+
+    ``gpu_devices`` is an injection point for tests and non-CUDA callers.
+    """
+
+    if gpu_devices is not None:
+        devices = [_normalize_gpu_device(device) for device in gpu_devices]
+        return {
+            "device_type": device_type or "cuda",
+            "gpu_count": num_gpus if num_gpus is not None else len(devices),
+            "gpus": devices,
+            "interconnect": interconnect or "unknown",
+        }
+
+    if not torch.cuda.is_available():
+        return {
+            "device_type": device_type or "cpu",
+            "gpu_count": 0,
+            "gpus": [],
+            "interconnect": interconnect or "none",
+        }
+
+    gpu_count = int(num_gpus if num_gpus is not None else torch.cuda.device_count())
+    devices = []
+    cuda_device_count = torch.cuda.device_count()
+    for device_id in range(gpu_count):
+        if device_id >= cuda_device_count:
+            devices.append(_normalize_gpu_device({}))
+            continue
+        props = torch.cuda.get_device_properties(device_id)
+        capability = torch.cuda.get_device_capability(device_id)
+        devices.append(
+            _normalize_gpu_device({
+                "name": props.name,
+                "memory_bytes": props.total_memory,
+                "compute_capability": f"{capability[0]}.{capability[1]}",
+            }))
+
+    return {
+        "device_type": device_type or "cuda",
+        "gpu_count": gpu_count,
+        "gpus": devices,
+        "interconnect": interconnect or _detect_cuda_interconnect(gpu_count),
+    }
+
+
+def software_profile(
+    *,
+    python_version: str | None = None,
+    torch_version: str | None = None,
+    cuda_version: str | None = None,
+    package_versions: Mapping[str, str | None] | None = None,
+) -> dict[str, Any]:
+    """Return normalized software cohort metadata.
+
+    Python, PyTorch, and CUDA use major/minor cohorts. Attention and kernel
+    packages keep exact versions because patch releases can change performance.
+    """
+
+    versions = dict(package_versions) if package_versions is not None else _installed_package_versions()
+    return {
+        "python": _major_minor(python_version or platform.python_version()),
+        "pytorch": _major_minor(torch_version or torch.__version__),
+        "cuda": _major_minor(cuda_version or torch.version.cuda),
+        "packages": {
+            name: str(version)
+            for name, version in sorted(versions.items())
+            if version
+        },
+    }
+
+
+def environment_metadata(
+    *,
+    env: Mapping[str, str] | None = None,
+    package_versions: Mapping[str, str | None] | None = None,
+    hardware: Mapping[str, Any] | None = None,
+    software: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Return audit metadata kept separate from comparable identity keys."""
+
+    source_env = env if env is not None else os.environ
+    full_package_versions = dict(package_versions) if package_versions is not None else _installed_package_versions()
+    return {
+        "python": {
+            "version": platform.python_version(),
+            "implementation": platform.python_implementation(),
+        },
+        "platform": {
+            "system": platform.system(),
+            "machine": platform.machine(),
+            "release": platform.release(),
+        },
+        "torch": {
+            "version": torch.__version__,
+            "cuda": torch.version.cuda,
+            "cudnn": torch.backends.cudnn.version(),
+        },
+        "env": {
+            key: source_env.get(key)
+            for key in _PROFILE_ENV_VARS
+            if source_env.get(key) is not None
+        },
+        "packages": {
+            key: value
+            for key, value in sorted(full_package_versions.items())
+            if value
+        },
+        "hardware_profile": dict(hardware) if hardware is not None else hardware_profile(),
+        "software_profile": dict(software) if software is not None else software_profile(
+            package_versions=full_package_versions),
+    }
+
+
+def _canonicalize(value: Any) -> Any:
+    if dataclasses.is_dataclass(value) and not isinstance(value, type):
+        return _canonicalize(dataclasses.asdict(value))
+    if isinstance(value, Mapping):
+        return {
+            str(key): _canonicalize(value[key])
+            for key in sorted(value, key=lambda item: str(item))
+        }
+    if isinstance(value, (set, frozenset)):
+        return [_canonicalize(item) for item in sorted(value, key=lambda item: str(item))]
+    if isinstance(value, tuple):
+        return [_canonicalize(item) for item in value]
+    if isinstance(value, list):
+        return [_canonicalize(item) for item in value]
+    if isinstance(value, enum.Enum):
+        return value.name
+    if isinstance(value, Path):
+        return value.expanduser().as_posix()
+    if isinstance(value, os.PathLike):
+        return os.fspath(value)
+    if isinstance(value, type):
+        return f"{value.__module__}.{value.__qualname__}"
+    if isinstance(value, torch.dtype):
+        return str(value).removeprefix("torch.")
+    return value
+
+
+def _none_if_empty(value: Any) -> Any | None:
+    if value is None:
+        return None
+    if isinstance(value, str) and not value.strip():
+        return None
+    return value
+
+
+def _optional_digest(value: Any) -> str | None:
+    if value is None:
+        return None
+    return sha256_hexdigest(str(value))
+
+
+def _looks_like_local_path(value: str) -> bool:
+    if value.startswith(("/", "./", "../", "~")):
+        return True
+    looks_like_hf_repo = re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", value)
+    has_separator = "/" in value or "\\" in value or os.sep in value
+    return has_separator and looks_like_hf_repo is None
+
+
+def _normalize_gpu_device(device: Mapping[str, Any]) -> dict[str, Any]:
+    memory_gb = device.get("memory_gb")
+    if memory_gb is None and device.get("memory_bytes") is not None:
+        memory_gb = round(float(device["memory_bytes"]) / (1024**3))
+    return {
+        "name": _none_if_empty(device.get("name")) or "unknown",
+        "memory_gb": memory_gb,
+        "compute_capability": _none_if_empty(device.get("compute_capability")),
+    }
+
+
+def _detect_cuda_interconnect(gpu_count: int) -> str:
+    if gpu_count <= 1:
+        return "single_gpu"
+    try:
+        from fastvideo.platforms import current_platform
+
+        if current_platform.is_cuda() and current_platform.is_full_nvlink(list(range(gpu_count))):
+            return "full_nvlink"
+        return "none_or_partial"
+    except Exception:
+        return "unknown"
+
+
+def _installed_package_versions() -> dict[str, str | None]:
+    return {
+        name: _distribution_version(*distribution_names)
+        for name, distribution_names in _PACKAGE_DISTRIBUTIONS.items()
+    }
+
+
+def _distribution_version(*names: str) -> str | None:
+    for name in names:
+        try:
+            return importlib.metadata.version(name)
+        except importlib.metadata.PackageNotFoundError:
+            continue
+    return None
+
+
+def _major_minor(version: str | None) -> str | None:
+    if not version:
+        return None
+    match = re.search(r"(\d+)\.(\d+)", version)
+    if match is None:
+        return version
+    return f"{match.group(1)}.{match.group(2)}"
+
+
+__all__ = [
+    "RECIPE_SCHEMA_VERSION",
+    "REQUIRED_BENCHMARK_IDENTITY_KEYS",
+    "benchmark_identity_from_config",
+    "canonical_json",
+    "environment_fingerprint",
+    "environment_metadata",
+    "hardware_profile",
+    "hardware_profile_id",
+    "normalize_model_path",
+    "profile_id",
+    "recipe_fingerprint",
+    "resolved_revision_from_model_path",
+    "build_recipe_from_benchmark_config",
+    "sha256_hexdigest",
+    "software_profile",
+    "software_profile_id",
+]

fastvideo/tests/performance/test_compare_baseline_policy.py

@@ -1,5 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
+import pytest
+
 from fastvideo.tests.performance import compare_baseline
 from fastvideo.performance.metric_policy import resolve_metric_policies
 
@@ -119,6 +121,110 @@ def test_boolean_regression_threshold_values_are_ignored():
     assert latency.threshold_percent == 0.08
     assert latency.threshold_absolute == 0.5
     assert latency.gated is False
+def test_normalized_record_preserves_identity_metadata(monkeypatch):
+    monkeypatch.setenv("PERF_RUN_SOURCE", "pr")
+    raw = _raw_result()
+    raw.update({
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 2,
+        "recipe": {
+            "recipe_schema_version": 1,
+        },
+        "recipe_fingerprint": "recipe-1",
+        "hardware_profile": {
+            "gpu_count": 1,
+        },
+        "hardware_profile_id": "hw-1",
+        "software_profile": {
+            "python": "3.12",
+        },
+        "software_profile_id": "sw-1",
+        "environment_metadata": {
+            "env": {
+                "IMAGE_VERSION": "latest",
+            },
+        },
+        "environment_fingerprint": "env-1",
+    })
+
+    record = compare_baseline.normalize_performance_result(raw)
+
+    assert record["workload_id"] == "wan-t2v"
+    assert record["variant_id"] == "1.3b-sp2"
+    assert record["benchmark_version"] == 2
+    assert record["recipe"] == {"recipe_schema_version": 1}
+    assert record["recipe_fingerprint"] == "recipe-1"
+    assert record["hardware_profile"] == {"gpu_count": 1}
+    assert record["hardware_profile_id"] == "hw-1"
+    assert record["software_profile"] == {"python": "3.12"}
+    assert record["software_profile_id"] == "sw-1"
+    assert record["environment_metadata"] == {"env": {"IMAGE_VERSION": "latest"}}
+    assert record["environment_fingerprint"] == "env-1"
+
+
+def test_normalized_record_reads_identity_labels_from_recipe(monkeypatch):
+    monkeypatch.setenv("PERF_RUN_SOURCE", "pr")
+    raw = _raw_result()
+    raw["recipe"] = {
+        "benchmark": {
+            "workload_id": "wan-t2v",
+            "variant_id": "1.3b-sp2",
+            "benchmark_version": 2,
+        },
+    }
+
+    record = compare_baseline.normalize_performance_result(raw)
+
+    assert record["workload_id"] == "wan-t2v"
+    assert record["variant_id"] == "1.3b-sp2"
+    assert record["benchmark_version"] == 2
+
+
+def test_comparison_identity_filters_use_full_issue_key():
+    record = {
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 2,
+        "recipe_fingerprint": "recipe-1",
+        "hardware_profile_id": "hw-1",
+        "software_profile_id": "sw-1",
+    }
+
+    assert compare_baseline._comparison_identity_filters(record) == {
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": "2",
+        "recipe_fingerprint": "recipe-1",
+        "hardware_profile_id": "hw-1",
+        "software_profile_id": "sw-1",
+    }
+
+
+def test_comparison_identity_filters_require_full_issue_key():
+    record = {
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 0,
+        "recipe_fingerprint": "recipe-1",
+        "hardware_profile_id": "hw-1",
+    }
+
+    with pytest.raises(ValueError, match="software_profile_id"):
+        compare_baseline._comparison_identity_filters(record)
+
+
+def test_comparison_identity_filters_keep_zero_version():
+    record = {
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 0,
+        "recipe_fingerprint": "recipe-1",
+        "hardware_profile_id": "hw-1",
+        "software_profile_id": "sw-1",
+    }
+
+    assert compare_baseline._comparison_identity_filters(record)["benchmark_version"] == "0"
 
 
 def test_baseline_eligibility_only_for_successful_scheduled_main():

fastvideo/tests/performance/test_dashboard_api.py

@@ -56,8 +56,34 @@ def _record(model_id, gpu_type, ts, commit, latency, throughput, success=True, *
 
 def test_summary_endpoint_returns_latest_group_status():
     app = create_app(FakeStore([
-        _record("wan", "NVIDIA L40S", "2026-01-01T00:00:00+00:00", "a" * 40, 10.0, 10.0),
-        _record("wan", "NVIDIA L40S", "2026-01-02T00:00:00+00:00", "b" * 40, 11.0, 9.0),
+        _record(
+            "wan",
+            "NVIDIA L40S",
+            "2026-01-01T00:00:00+00:00",
+            "a" * 40,
+            10.0,
+            10.0,
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            recipe_fingerprint="recipe-a",
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+        ),
+        _record(
+            "wan",
+            "NVIDIA L40S",
+            "2026-01-02T00:00:00+00:00",
+            "b" * 40,
+            11.0,
+            9.0,
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            recipe_fingerprint="recipe-a",
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+        ),
     ]))
     client = TestClient(app)
 
@@ -71,6 +97,10 @@ def test_summary_endpoint_returns_latest_group_status():
     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"
+    assert body["rows"][0]["workload_id"] == "wan-t2v"
+    assert body["rows"][0]["variant_id"] == "1.3b-sp2"
+    assert body["rows"][0]["benchmark_version"] == 2
+    assert body["rows"][0]["recipe_fingerprint"] == "recipe-a"
 
 
 def test_summary_status_is_independent_of_days_window():

fastvideo/tests/performance/test_dashboard_plotly.py

@@ -0,0 +1,98 @@
+# SPDX-License-Identifier: Apache-2.0
+
+import pandas as pd
+
+from fastvideo.tests.performance import dashboard
+
+
+def _record(recipe_fingerprint="recipe-a", software_profile_id="sw-a", **overrides):
+    record = {
+        "model_id": "wan-t2v-1.3b-2gpu",
+        "gpu_type": "NVIDIA L40S",
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 2,
+        "recipe_fingerprint": recipe_fingerprint,
+        "hardware_profile_id": "hw-l40s",
+        "software_profile_id": software_profile_id,
+        "timestamp": "2026-01-01T00:00:00+00:00",
+        "commit_sha": "a" * 40,
+        "config_id": "aaaaaaa",
+        "latency": 10.0,
+        "throughput": 4.5,
+        "memory": 10000.0,
+        "text_encoder_time_s": 2.0,
+        "dit_time_s": 8.0,
+        "vae_decode_time_s": 3.0,
+    }
+    record.update(overrides)
+    return record
+
+
+def test_group_data_uses_full_comparison_cohort():
+    df = pd.DataFrame([
+        _record(recipe_fingerprint="recipe-a", software_profile_id="sw-a"),
+        _record(recipe_fingerprint="recipe-b", software_profile_id="sw-b"),
+    ])
+
+    groups = list(dashboard.group_data(df))
+
+    assert len(groups) == 2
+    assert {key[5] for key, _group in groups} == {"recipe-a", "recipe-b"}
+    assert {key[7] for key, _group in groups} == {"sw-a", "sw-b"}
+
+
+def test_group_data_fills_legacy_cohort_columns():
+    df = pd.DataFrame([{
+        "model_id": "legacy-wan",
+        "gpu_type": "NVIDIA L40S",
+        "timestamp": "2026-01-01T00:00:00+00:00",
+        "commit_sha": "a" * 40,
+        "config_id": "aaaaaaa",
+        "latency": 10.0,
+    }])
+
+    groups = list(dashboard.group_data(df))
+
+    assert len(groups) == 1
+    assert groups[0][0][2:] == ("", "", "", "", "", "")
+
+
+def test_build_plots_labels_distinct_cohorts():
+    df = pd.DataFrame([
+        _record(recipe_fingerprint="recipe-a", software_profile_id="sw-a"),
+        _record(recipe_fingerprint="recipe-b", software_profile_id="sw-b"),
+    ])
+
+    figs, skipped_metrics = dashboard.build_plots(df)
+    titles = [fig.layout.title.text for fig in figs]
+
+    assert len(figs) == len(dashboard.METRICS) * 2
+    assert skipped_metrics == []
+    assert any("recipe recipe-a | hw-l40s | sw-a" in title for title in titles)
+    assert any("recipe recipe-b | hw-l40s | sw-b" in title for title in titles)
+
+
+def test_skipped_metric_table_includes_cohort_identity():
+    df = pd.DataFrame([{
+        "model_id": "wan-t2v-1.3b-2gpu",
+        "gpu_type": "NVIDIA L40S",
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 2,
+        "recipe_fingerprint": "recipe-a",
+        "hardware_profile_id": "hw-l40s",
+        "software_profile_id": "sw-a",
+        "timestamp": "2026-01-01T00:00:00+00:00",
+        "commit_sha": "a" * 40,
+        "config_id": "aaaaaaa",
+        "latency": 10.0,
+    }])
+
+    _figs, skipped_metrics = dashboard.build_plots(df)
+    html = dashboard.render_skipped_metrics(skipped_metrics[:1])
+
+    assert skipped_metrics[0]["cohort"] == "wan-t2v / 1.3b-sp2 / v2"
+    assert skipped_metrics[0]["cohort_detail"] == "recipe recipe-a | hw-l40s | sw-a"
+    assert "wan-t2v / 1.3b-sp2 / v2" in html
+    assert "recipe recipe-a | hw-l40s | sw-a" in html

fastvideo/tests/performance/test_dashboard_service.py

@@ -142,6 +142,111 @@ def test_build_latest_summary_separates_informational_threshold_crossing():
     assert rows[0]["threshold_exceeded_metrics"] == ["latency"]
     assert rows[0]["failing_metrics"] == []
     assert rows[0]["computed_regression_status"] == "pass"
+def test_build_latest_summary_keeps_identity_cohorts_separate():
+    records = [
+        _record(
+            "2026-01-01T00:00:00+00:00",
+            "a" * 40,
+            10.0,
+            10.0,
+            recipe_fingerprint="recipe-a",
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+            run_source="scheduled_main",
+            baseline_eligible=True,
+        ),
+        _record(
+            "2026-01-02T00:00:00+00:00",
+            "b" * 40,
+            20.0,
+            5.0,
+            recipe_fingerprint="recipe-b",
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+            run_source="scheduled_main",
+            baseline_eligible=True,
+        ),
+        _record(
+            "2026-01-03T00:00:00+00:00",
+            "c" * 40,
+            22.0,
+            4.5,
+            recipe_fingerprint="recipe-b",
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+        ),
+    ]
+
+    rows = build_latest_summary(records, max_regression=0.05)
+    recipe_b_row = next(row for row in rows if row["recipe_fingerprint"] == "recipe-b")
+
+    assert len(rows) == 2
+    assert recipe_b_row["baseline_n"] == 1
+    assert recipe_b_row["metrics"]["latency"]["baseline"] == 20.0
+    assert recipe_b_row["workload_id"] == "wan-t2v"
+    assert recipe_b_row["variant_id"] == "1.3b-sp2"
+    assert recipe_b_row["benchmark_version"] == 2
+
+
+def test_build_latest_summary_keeps_variant_versions_separate():
+    records = [
+        _record(
+            "2026-01-01T00:00:00+00:00",
+            "a" * 40,
+            10.0,
+            10.0,
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=1,
+            recipe_fingerprint="recipe-a",
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+            run_source="scheduled_main",
+            baseline_eligible=True,
+        ),
+        _record(
+            "2026-01-02T00:00:00+00:00",
+            "b" * 40,
+            20.0,
+            5.0,
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            recipe_fingerprint="recipe-a",
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+            run_source="scheduled_main",
+            baseline_eligible=True,
+        ),
+        _record(
+            "2026-01-03T00:00:00+00:00",
+            "c" * 40,
+            22.0,
+            4.5,
+            workload_id="wan-t2v",
+            variant_id="1.3b-sp2",
+            benchmark_version=2,
+            recipe_fingerprint="recipe-a",
+            hardware_profile_id="hw-l40s",
+            software_profile_id="sw-cu130",
+        ),
+    ]
+
+    rows = build_latest_summary(records, max_regression=0.05)
+    version_2_row = next(row for row in rows if row["benchmark_version"] == 2)
+
+    assert len(rows) == 2
+    assert version_2_row["baseline_n"] == 1
+    assert version_2_row["metrics"]["latency"]["baseline"] == 20.0
 
 
 def test_filter_records_and_trends_preserve_metric_points():
@@ -218,3 +323,78 @@ def test_load_records_can_filter_baseline_eligible_records(tmp_path):
         "2026-01-02T00:00:00+00:00",
         "2026-01-03T00:00:00+00:00",
     }
+
+
+def test_load_records_for_model_filters_identity_cohort(tmp_path):
+    model_dir = tmp_path / "wan"
+    model_dir.mkdir()
+    (model_dir / "matching.json").write_text(
+        """
+        {
+          "model_id": "wan",
+          "gpu_type": "NVIDIA L40S",
+          "timestamp": "2026-01-01T00:00:00+00:00",
+          "success": true,
+          "baseline_eligible": true,
+          "workload_id": "wan-t2v",
+          "variant_id": "1.3b-sp2",
+          "benchmark_version": 2,
+          "recipe_fingerprint": "recipe-a",
+          "hardware_profile_id": "hw-l40s",
+          "software_profile_id": "sw-cu130"
+        }
+        """,
+        encoding="utf-8",
+    )
+    (model_dir / "other_recipe.json").write_text(
+        """
+        {
+          "model_id": "wan",
+          "gpu_type": "NVIDIA L40S",
+          "timestamp": "2026-01-02T00:00:00+00:00",
+          "success": true,
+          "baseline_eligible": true,
+          "workload_id": "wan-t2v",
+          "variant_id": "1.3b-sp2",
+          "benchmark_version": 2,
+          "recipe_fingerprint": "recipe-b",
+          "hardware_profile_id": "hw-l40s",
+          "software_profile_id": "sw-cu130"
+        }
+        """,
+        encoding="utf-8",
+    )
+    (model_dir / "other_version.json").write_text(
+        """
+        {
+          "model_id": "wan",
+          "gpu_type": "NVIDIA L40S",
+          "timestamp": "2026-01-03T00:00:00+00:00",
+          "success": true,
+          "baseline_eligible": true,
+          "workload_id": "wan-t2v",
+          "variant_id": "1.3b-sp2",
+          "benchmark_version": 3,
+          "recipe_fingerprint": "recipe-a",
+          "hardware_profile_id": "hw-l40s",
+          "software_profile_id": "sw-cu130"
+        }
+        """,
+        encoding="utf-8",
+    )
+
+    records = hf_store.load_records_for_model(
+        str(tmp_path),
+        "wan",
+        "NVIDIA L40S",
+        workload_id="wan-t2v",
+        variant_id="1.3b-sp2",
+        benchmark_version="2",
+        recipe_fingerprint="recipe-a",
+        hardware_profile_id="hw-l40s",
+        software_profile_id="sw-cu130",
+        baseline_eligible_only=True,
+    )
+
+    assert len(records) == 1
+    assert records[0]["recipe_fingerprint"] == "recipe-a"

fastvideo/tests/performance/test_identity.py

@@ -0,0 +1,422 @@
+# SPDX-License-Identifier: Apache-2.0
+
+from copy import deepcopy
+
+import pytest
+import torch
+
+from fastvideo.tests.performance import identity as identity_module
+from fastvideo.tests.performance.identity import (
+    build_recipe_from_benchmark_config,
+    canonical_json,
+    environment_fingerprint,
+    environment_metadata,
+    hardware_profile,
+    hardware_profile_id,
+    recipe_fingerprint,
+    resolved_revision_from_model_path,
+    software_profile,
+    software_profile_id,
+)
+
+
+def _benchmark_config():
+    return {
+        "benchmark_id": "wan-t2v-1.3b-2gpu",
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 2,
+        "model": {
+            "model_path": "Wan-AI/Wan2.1-T2V-1.3B-Diffusers",
+            "model_short_name": "Wan2.1-T2V-1.3B",
+        },
+        "init_kwargs": {
+            "num_gpus": 2,
+            "sp_size": 2,
+            "tp_size": 1,
+            "vae_sp": True,
+            "vae_tiling": True,
+            "text_encoder_precisions": ["fp32"],
+        },
+        "generation_kwargs": {
+            "height": 480,
+            "width": 832,
+            "num_frames": 45,
+            "num_inference_steps": 4,
+            "guidance_scale": 3,
+            "embedded_cfg_scale": 6,
+            "seed": 1024,
+            "fps": 24,
+            "neg_prompt": "low quality",
+        },
+        "test_prompts": ["A cinematic video."],
+    }
+
+
+def _fingerprint(cfg, attention_backend="FLASH_ATTN"):
+    recipe = build_recipe_from_benchmark_config(cfg, attention_backend=attention_backend)
+    return recipe_fingerprint(recipe)
+
+
+def test_canonical_json_is_stable_for_mapping_order():
+    left = {"b": [2, 1], "a": {"z": "last", "m": "middle"}}
+    right = {"a": {"m": "middle", "z": "last"}, "b": [2, 1]}
+
+    assert canonical_json(left) == canonical_json(right)
+
+
+def test_canonical_json_handles_sets_deterministically():
+    assert canonical_json({"values": {"b", "a"}}) == '{"values":["a","b"]}'
+    assert canonical_json({"values": frozenset(("b", "a"))}) == '{"values":["a","b"]}'
+
+
+def test_same_config_produces_same_recipe_fingerprint():
+    cfg = _benchmark_config()
+
+    assert _fingerprint(cfg) == _fingerprint(deepcopy(cfg))
+
+
+def test_recipe_includes_first_class_benchmark_identity():
+    recipe = build_recipe_from_benchmark_config(_benchmark_config())
+
+    assert recipe["benchmark"] == {
+        "benchmark_id": "wan-t2v-1.3b-2gpu",
+        "workload_id": "wan-t2v",
+        "variant_id": "1.3b-sp2",
+        "benchmark_version": 2,
+    }
+
+
+def test_recipe_requires_first_class_benchmark_identity():
+    cfg = _benchmark_config()
+    del cfg["workload_id"]
+
+    with pytest.raises(ValueError, match="workload_id"):
+        build_recipe_from_benchmark_config(cfg)
+
+
+def test_semantically_equivalent_sequence_forms_match():
+    cfg = _benchmark_config()
+    equivalent = deepcopy(cfg)
+    equivalent["init_kwargs"]["text_encoder_precisions"] = tuple(
+        equivalent["init_kwargs"]["text_encoder_precisions"])
+
+    assert _fingerprint(cfg) == _fingerprint(equivalent)
+
+
+def test_output_path_does_not_change_recipe_fingerprint():
+    cfg = _benchmark_config()
+    with_output_path = deepcopy(cfg)
+    with_output_path["generation_kwargs"]["output_path"] = "/tmp/generated"
+
+    assert _fingerprint(cfg) == _fingerprint(with_output_path)
+
+
+def test_runtime_resolved_values_do_not_change_recipe_fingerprint():
+    # Runtime-resolved values are audit metadata: an upstream repo commit
+    # (even a README-only edit) or a silent attention-backend fallback must
+    # not churn the cohort — that would open a fresh ungated cohort and let
+    # degraded numbers seed a new baseline. Declared inputs (including
+    # requested_backend) stay in the hash.
+    cfg = _benchmark_config()
+    recipe = build_recipe_from_benchmark_config(
+        cfg, attention_backend="FLASH_ATTN",
+        resolved_attention_backend="FLASH_ATTN",
+        resolved_model_revision="aaaa1111")
+    changed = build_recipe_from_benchmark_config(
+        cfg, attention_backend="FLASH_ATTN",
+        resolved_attention_backend="TORCH_SDPA",
+        resolved_model_revision="bbbb2222")
+
+    assert recipe["model"]["resolved_revision"] == "aaaa1111"  # kept for audit
+    assert recipe["attention"]["resolved_backend"] == "FLASH_ATTN"  # kept for audit
+    assert recipe_fingerprint(recipe) == recipe_fingerprint(changed)
+
+
+def test_measured_prompt_override_ignores_unused_configured_prompts():
+    cfg = _benchmark_config()
+    changed = deepcopy(cfg)
+    changed["test_prompts"] = [
+        "Unused prompt that should not affect this measured workload.",
+        cfg["test_prompts"][0],
+    ]
+
+    recipe = build_recipe_from_benchmark_config(cfg, measured_prompts=[cfg["test_prompts"][0]])
+    changed_recipe = build_recipe_from_benchmark_config(changed, measured_prompts=[cfg["test_prompts"][0]])
+
+    assert recipe_fingerprint(recipe) == recipe_fingerprint(changed_recipe)
+
+
+def test_num_inference_steps_changes_recipe_fingerprint():
+    cfg = _benchmark_config()
+    changed = deepcopy(cfg)
+    changed["generation_kwargs"]["num_inference_steps"] = 8
+
+    assert _fingerprint(cfg) != _fingerprint(changed)
+
+
+def test_benchmark_version_changes_recipe_fingerprint():
+    cfg = _benchmark_config()
+    changed = deepcopy(cfg)
+    changed["benchmark_version"] = 3
+
+    assert _fingerprint(cfg) != _fingerprint(changed)
+
+
+def test_attention_backend_changes_recipe_fingerprint():
+    cfg = _benchmark_config()
+
+    assert _fingerprint(cfg, attention_backend="FLASH_ATTN") != _fingerprint(
+        cfg, attention_backend="TORCH_SDPA")
+
+
+def test_precision_changes_recipe_fingerprint():
+    cfg = _benchmark_config()
+    changed = deepcopy(cfg)
+    changed["init_kwargs"]["text_encoder_precisions"] = ["bf16"]
+
+    assert _fingerprint(cfg) != _fingerprint(changed)
+
+
+def test_dimensions_and_seed_change_recipe_fingerprint():
+    cfg = _benchmark_config()
+    different_dimensions = deepcopy(cfg)
+    different_dimensions["generation_kwargs"]["width"] = 1024
+    different_seed = deepcopy(cfg)
+    different_seed["generation_kwargs"]["seed"] = 2048
+
+    assert _fingerprint(cfg) != _fingerprint(different_dimensions)
+    assert _fingerprint(cfg) != _fingerprint(different_seed)
+
+
+def test_distributed_layout_changes_recipe_fingerprint():
+    cfg = _benchmark_config()
+    changed = deepcopy(cfg)
+    changed["init_kwargs"]["sp_size"] = 1
+    changed["init_kwargs"]["tp_size"] = 2
+
+    assert _fingerprint(cfg) != _fingerprint(changed)
+
+
+def test_software_profile_uses_exact_attention_kernel_package_versions_for_id():
+    profile_a = software_profile(
+        python_version="3.12.4",
+        torch_version="2.12.0+cu130",
+        cuda_version="13.0",
+        package_versions={
+            "triton": "3.4.1",
+            "fastvideo_kernel": "0.3.2",
+            "flash_attn_4": "4.0.0.dev0",
+            "flashinfer": "0.2.11",
+            "nvidia_cutlass_dsl": "4.5.0",
+        },
+    )
+    profile_b = software_profile(
+        python_version="3.12.5",
+        torch_version="2.12.1+cu130",
+        cuda_version="13.0",
+        package_versions={
+            "triton": "3.4.9",
+            "fastvideo_kernel": "0.3.7",
+            "flash_attn_4": "4.0.0.dev1",
+            "flashinfer": "0.2.12",
+            "nvidia_cutlass_dsl": "4.5.1",
+        },
+    )
+
+    assert profile_a == {
+        "python": "3.12",
+        "pytorch": "2.12",
+        "cuda": "13.0",
+        "packages": {
+            "fastvideo_kernel": "0.3.2",
+            "flash_attn_4": "4.0.0.dev0",
+            "flashinfer": "0.2.11",
+            "nvidia_cutlass_dsl": "4.5.0",
+            "triton": "3.4.1",
+        },
+    }
+    assert profile_b["python"] == "3.12"
+    assert profile_b["pytorch"] == "2.12"
+    assert profile_b["cuda"] == "13.0"
+    assert software_profile_id(profile_a) != software_profile_id(profile_b)
+
+
+def test_installed_package_versions_checks_attention_kernel_distributions(monkeypatch):
+    versions = {
+        "fastvideo-kernel": "0.3.2",
+        "flash-attn": "2.8.1",
+        "flash-attn-4": "4.0.0.dev0",
+        "flash-attention-fp4": "0.1.0",
+        "flashinfer-python": "0.2.11",
+        "nvidia-cutlass-dsl": "4.5.0",
+    }
+
+    def fake_version(name):
+        if name not in versions:
+            raise identity_module.importlib.metadata.PackageNotFoundError(name)
+        return versions[name]
+
+    monkeypatch.setattr(identity_module.importlib.metadata, "version", fake_version)
+
+    installed = identity_module._installed_package_versions()
+
+    assert installed["fastvideo_kernel"] == "0.3.2"
+    assert installed["flash_attn"] == "2.8.1"
+    assert installed["flash_attn_4"] == "4.0.0.dev0"
+    assert installed["flash_attention_fp4"] == "0.1.0"
+    assert installed["flashinfer"] == "0.2.11"
+    assert installed["nvidia_cutlass_dsl"] == "4.5.0"
+
+
+def test_hardware_profile_id_uses_normalized_gpu_cohort():
+    profile = hardware_profile(
+        num_gpus=2,
+        gpu_devices=[
+            {
+                "name": "NVIDIA L40S",
+                "memory_bytes": 48 * 1024**3,
+                "compute_capability": "8.9",
+            },
+            {
+                "name": "NVIDIA L40S",
+                "memory_gb": 48,
+                "compute_capability": "8.9",
+            },
+        ],
+        interconnect="none_or_partial",
+    )
+
+    assert profile == {
+        "device_type": "cuda",
+        "gpu_count": 2,
+        "gpus": [
+            {
+                "name": "NVIDIA L40S",
+                "memory_gb": 48,
+                "compute_capability": "8.9",
+            },
+            {
+                "name": "NVIDIA L40S",
+                "memory_gb": 48,
+                "compute_capability": "8.9",
+            },
+        ],
+        "interconnect": "none_or_partial",
+    }
+    assert hardware_profile_id(profile).startswith("hw-")
+
+
+def test_hardware_profile_pads_missing_requested_cuda_devices(monkeypatch):
+    class Props:
+        name = "NVIDIA L40S"
+        total_memory = 48 * 1024**3
+
+    monkeypatch.setattr(torch.cuda, "is_available", lambda: True)
+    monkeypatch.setattr(torch.cuda, "device_count", lambda: 1)
+    monkeypatch.setattr(torch.cuda, "get_device_properties", lambda device_id: Props())
+    monkeypatch.setattr(torch.cuda, "get_device_capability", lambda device_id: (8, 9))
+
+    profile = hardware_profile(num_gpus=2, interconnect="unknown")
+
+    assert profile["gpu_count"] == 2
+    assert profile["gpus"] == [
+        {
+            "name": "NVIDIA L40S",
+            "memory_gb": 48,
+            "compute_capability": "8.9",
+        },
+        {
+            "name": "unknown",
+            "memory_gb": None,
+            "compute_capability": None,
+        },
+    ]
+
+
+def test_local_path_detection_checks_both_path_separators(monkeypatch):
+    monkeypatch.setattr(identity_module.os, "sep", "\\")
+
+    assert identity_module._looks_like_local_path("models/local/checkpoint") is True
+    assert identity_module._looks_like_local_path(r"C:\models\checkpoint") is True
+    assert identity_module._looks_like_local_path("Wan-AI/Wan2.1-T2V-1.3B-Diffusers") is False
+
+
+def test_environment_metadata_is_separate_audit_fingerprint():
+    cfg = _benchmark_config()
+    recipe_hash = _fingerprint(cfg)
+    audit_a = environment_metadata(
+        env={"IMAGE_VERSION": "py3.12-cuda13.0.0"},
+        package_versions={"triton": "3.4.1"},
+        hardware={"device_type": "cuda", "gpu_count": 2},
+        software={"python": "3.12", "pytorch": "2.12", "cuda": "13.0"},
+    )
+    audit_b = environment_metadata(
+        env={"IMAGE_VERSION": "py3.12-cuda13.0.1"},
+        package_versions={"triton": "3.4.1"},
+        hardware={"device_type": "cuda", "gpu_count": 2},
+        software={"python": "3.12", "pytorch": "2.12", "cuda": "13.0"},
+    )
+
+    assert recipe_hash == _fingerprint(cfg)
+    assert environment_fingerprint(audit_a) != environment_fingerprint(audit_b)
+
+
+def test_resolved_revision_from_hf_snapshot_path():
+    revision = "a" * 40
+
+    assert resolved_revision_from_model_path(
+        f"/root/.cache/huggingface/hub/models--org--repo/snapshots/{revision}") == revision
+    assert resolved_revision_from_model_path("/models/local-repo") is None
+
+
+def test_runtime_identity_from_generator_summarizes_worker_records():
+    from fastvideo.tests.performance.test_inference_performance import _runtime_identity_from_generator
+
+    revision = "b" * 40
+
+    class FakeExecutor:
+        def collective_rpc(self, _method):
+            return [
+                {
+                    "resolved_attention_backends": ["FLASH_ATTN"],
+                    "resolved_model_path": f"/root/.cache/huggingface/hub/models--org--repo/snapshots/{revision}",
+                },
+                {
+                    "resolved_attention_backends": ["FLASH_ATTN"],
+                    "resolved_model_path": f"/root/.cache/huggingface/hub/models--org--repo/snapshots/{revision}",
+                },
+            ]
+
+    class FakeGenerator:
+        executor = FakeExecutor()
+
+    assert _runtime_identity_from_generator(FakeGenerator()) == {
+        "resolved_attention_backend": "FLASH_ATTN",
+        "resolved_model_revision": revision,
+    }
+
+
+def test_collect_worker_identity_handles_torch_module_pipeline():
+    from fastvideo.tests.performance.test_inference_performance import _collect_worker_identity
+
+    class Backend:
+        name = "FLASH_ATTN"
+
+    class Leaf(torch.nn.Module):
+        backend = Backend()
+
+    class Pipeline(torch.nn.Module):
+        model_path = "/models/local"
+
+        def __init__(self):
+            super().__init__()
+            self.leaf = Leaf()
+
+    class Worker:
+        pipeline = Pipeline()
+
+    assert _collect_worker_identity(Worker()) == {
+        "resolved_attention_backends": ["FLASH_ATTN"],
+        "resolved_model_path": "/models/local",
+    }

fastvideo/tests/performance/test_inference_performance.py

@@ -12,12 +12,25 @@ import os
 import time
 from collections.abc import Mapping
 from datetime import datetime, timezone
+from typing import Any
 
 import torch
 import pytest
 
 from fastvideo import VideoGenerator
 from fastvideo.logger import init_logger
+from fastvideo.tests.performance.identity import (
+    benchmark_identity_from_config,
+    build_recipe_from_benchmark_config,
+    environment_fingerprint,
+    environment_metadata,
+    hardware_profile,
+    hardware_profile_id,
+    recipe_fingerprint,
+    resolved_revision_from_model_path,
+    software_profile,
+    software_profile_id,
+)
 from fastvideo.worker.multiproc_executor import MultiprocExecutor
 
 logger = init_logger(__name__)
@@ -34,8 +47,11 @@ V2_REQUIRED_IDENTITY_FIELDS = (
     "variant_id",
     "benchmark_version",
 )
+# "recipe" is no longer config-declarable: the fingerprint follow-up landed,
+# and the generated recipe document owns that key in emitted records. A config
+# declaring it now fails validation loudly instead of being silently
+# overwritten by the generated one.
 V2_OPTIONAL_METADATA_FIELDS = (
-    "recipe",
     "metric_threshold_policy",
     "quality_metadata",
 )
@@ -179,11 +195,7 @@ 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)
-        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)
+        metric_key = STAGE_METRIC_MAP.get(stage_class)
         if metric_key is None:
             logger.debug("Unmapped stage '%s' class '%s' (%.3fs)",
                          stage_name,
@@ -231,6 +243,110 @@ def _write_results(results):
     logger.info("Performance results written to %s", filepath)
 
 
+def _backend_name(value) -> str:
+    if hasattr(value, "name"):
+        return str(value.name)
+    return str(value)
+
+
+def _collect_worker_identity(worker) -> dict[str, Any]:
+    pipeline = getattr(worker, "pipeline", None)
+    model_path = getattr(pipeline, "model_path", None)
+    backends: set[str] = set()
+
+    modules = getattr(pipeline, "modules", None)
+    if isinstance(modules, Mapping):
+        module_iter = modules.values()
+    elif isinstance(pipeline, torch.nn.Module):
+        module_iter = (pipeline,)
+    else:
+        module_iter = ()
+
+    for module in module_iter:
+        if isinstance(module, torch.nn.Module):
+            for submodule in module.modules():
+                backend = getattr(submodule, "backend", None)
+                if backend is not None:
+                    backends.add(_backend_name(backend))
+        else:
+            backend = getattr(module, "backend", None)
+            if backend is not None:
+                backends.add(_backend_name(backend))
+
+    return {
+        "resolved_attention_backends": sorted(backends),
+        "resolved_model_path": model_path,
+    }
+
+
+def _single_or_list(values: set[str]) -> str | list[str] | None:
+    ordered = sorted(values)
+    if not ordered:
+        return None
+    if len(ordered) == 1:
+        return ordered[0]
+    return ordered
+
+
+def _runtime_identity_from_generator(generator) -> dict[str, Any]:
+    worker_records = generator.executor.collective_rpc(_collect_worker_identity)
+    resolved_backends: set[str] = set()
+    resolved_revisions: set[str] = set()
+
+    for record in worker_records:
+        resolved_backends.update(record.get("resolved_attention_backends") or [])
+        revision = resolved_revision_from_model_path(record.get("resolved_model_path"))
+        if revision is not None:
+            resolved_revisions.add(revision)
+
+    return {
+        "resolved_attention_backend": _single_or_list(resolved_backends),
+        "resolved_model_revision": _single_or_list(resolved_revisions),
+    }
+
+
+def _benchmark_identity_fields(cfg):
+    identity = benchmark_identity_from_config(cfg)
+    return {
+        key: identity[key]
+        for key in ("workload_id", "variant_id", "benchmark_version")
+    }
+
+
+def _build_identity_fields(cfg, init_kwargs, prompt, runtime_identity):
+    # Legacy v1 configs (no config_schema_version) stay on the legacy
+    # (model_id, gpu_type) cohort: they lack the required identity fields,
+    # and building a recipe for them would raise AFTER the GPU measurement
+    # completed. The validator already enforces that v2 configs carry the
+    # identity fields, so v2 records always get the full identity block.
+    if cfg.get("config_schema_version") is None:
+        return {}
+    recipe = build_recipe_from_benchmark_config(
+        cfg,
+        resolved_attention_backend=runtime_identity.get("resolved_attention_backend"),
+        resolved_model_revision=runtime_identity.get("resolved_model_revision"),
+        measured_prompts=[prompt],
+    )
+    num_gpus = init_kwargs.get("num_gpus", cfg.get("run_config", {}).get("required_gpus", 1))
+    hw_profile = hardware_profile(num_gpus=num_gpus)
+    sw_profile = software_profile()
+    env_metadata = environment_metadata(
+        hardware=hw_profile,
+        software=sw_profile,
+    )
+    return {
+        "recipe": recipe,
+        "recipe_fingerprint": recipe_fingerprint(recipe),
+        "hardware_profile": hw_profile,
+        "hardware_profile_id": hardware_profile_id(hw_profile),
+        "software_profile": sw_profile,
+        "software_profile_id": software_profile_id(sw_profile),
+        "environment_metadata": env_metadata,
+        "environment_fingerprint": environment_fingerprint(env_metadata),
+        **_benchmark_identity_fields(cfg),
+    }
+
+
 # -- Test -------------------------------------------------------------------
 
 def _run_benchmark(cfg):
@@ -268,6 +384,7 @@ def _run_benchmark(cfg):
             model_path=model_info["model_path"],
             **init_kwargs,
         )
+        runtime_identity = _runtime_identity_from_generator(generator)
 
         for i in range(num_warmup):
             logger.info("Warmup run %d/%d", i + 1, num_warmup)
@@ -322,6 +439,7 @@ def _run_benchmark(cfg):
         "dit_time_s": _avg_component(all_component_times, "dit_time_s"),
         "vae_decode_time_s": _avg_component(all_component_times,
                                             "vae_decode_time_s"),
+        **_build_identity_fields(cfg, init_kwargs, prompt, runtime_identity),
     }
 
     logger.info(

fastvideo/tests/performance/test_inference_performance_component_times.py

@@ -1,20 +1,13 @@
 # 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,
@@ -52,35 +45,6 @@ def test_extract_component_times_uses_stage_class_for_pipeline_stage_keys():
     }
 
 
-def test_denoising_stage_subclasses_inherit_component_metric():
-    assert SubclassStyleDenoisingStage.performance_component_metric == "dit_time_s"
-
-
-def test_cosmos25_direct_pipeline_stages_define_component_metrics():
-    assert Cosmos25TextEncodingStage.performance_component_metric == "text_encoder_time_s"
-    assert Cosmos25AutoDenoisingStage.performance_component_metric == "dit_time_s"
-
-
-def test_extract_component_times_uses_component_metric_for_stage_subclasses():
-    result = {
-        "logging_info": {
-            "stages": {
-                "denoising_stage": {
-                    "execution_time": 4.2,
-                    "stage_class": "CosmosDenoisingStage",
-                    "component_metric": "dit_time_s",
-                },
-            },
-        },
-    }
-
-    assert _extract_component_times(result) == {
-        "text_encoder_time_s": None,
-        "dit_time_s": 4.2,
-        "vae_decode_time_s": None,
-    }
-
-
 def test_extract_component_times_keeps_legacy_class_name_keys():
     # Backward-compatibility check for logs produced before pipeline-unique
     # stage keys carried a separate stage_class field.

fastvideo/tests/ssim/README.md

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

fastvideo/tests/ssim/run_parallel.py

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

fastvideo/tests/ssim/test_gamecraft_similarity.py

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

fastvideo/tests/ssim/test_gen3c_similarity.py

@@ -20,16 +20,10 @@ 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
 
@@ -56,16 +50,16 @@ def _resolve_gen3c_test_image_path() -> str:
 # Device detection
 # ---------------------------------------------------------------------------
 
-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,
-)
+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}")
 
 # ---------------------------------------------------------------------------
 # GEN3C generation parameters
@@ -92,16 +86,9 @@ 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.",
 ]
 
@@ -128,17 +115,12 @@ 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__))
-    output_dir = build_generated_output_dir(
-        script_dir,
-        device_reference_folder,
-        model_id,
-        ATTENTION_BACKEND,
-    )
+    base_output_dir = os.path.join(script_dir, "generated_videos", model_id)
+    output_dir = os.path.join(base_output_dir, ATTENTION_BACKEND)
     output_video_name = CANDIDATE_VIDEO_NAME
     os.makedirs(output_dir, exist_ok=True)
 
-    params_map = select_ssim_params(MODEL_TO_PARAMS, FULL_QUALITY_MODEL_TO_PARAMS)
-    BASE_PARAMS = params_map[model_id]
+    BASE_PARAMS = MODEL_TO_PARAMS[model_id]
     num_inference_steps = BASE_PARAMS["num_inference_steps"]
     model_path = BASE_PARAMS["model_path"]
 
@@ -196,23 +178,18 @@ def test_gen3c_inference_similarity(prompt, ATTENTION_BACKEND, model_id):
     for stale_video in glob.glob(stale_pattern):
         os.remove(stale_video)
 
-    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()
+    generator = VideoGenerator.from_pretrained(
+        model_path=model_path, **init_kwargs
+    )
+    generator.generate_video(prompt, **generation_kwargs)
+
+    if isinstance(generator.executor, MultiprocExecutor):
+        generator.executor.shutdown()
 
     assert os.path.exists(output_dir), f"Output not generated at {output_dir}"
 
-    reference_folder = build_reference_folder_path(
-        script_dir,
-        device_reference_folder,
-        model_id,
-        ATTENTION_BACKEND,
+    reference_folder = os.path.join(
+        script_dir, device_reference_folder, model_id, ATTENTION_BACKEND
     )
     if not os.path.exists(reference_folder):
         raise FileNotFoundError(

fastvideo/tests/ssim/test_lingbot_similarity.py

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

fastvideo/tests/ssim/test_longcat_similarity.py

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

fastvideo/tests/ssim/test_matrixgame2_similarity.py

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

fastvideo/tests/ssim/test_matrixgame3_similarity.py

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

fastvideo/tests/ssim/test_sd35_similarity.py

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