[ci]: tighten performance cohort identity

2026-07-05 12:19:08 -07:00
parent 663dfb2f37
commit cf5d2209c9
6 changed files with 184 additions and 9 deletions
@@ -276,8 +276,8 @@ Written by `test_inference_performance.py`. One file per benchmark run.
    "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": ["<sha256>"] },
-    "attention": { "requested_backend": "FLASH_ATTN" }
+    "inputs": { "prompt_count": 1, "prompt_sha256": ["<measured-prompt-sha256>"] },
+    "attention": { "requested_backend": "FLASH_ATTN", "resolved_backend": "FLASH_ATTN" }
  },
  "recipe_fingerprint": "<sha256>",
  "hardware_profile": {
@@ -307,6 +307,9 @@ result, used as the rolling-baseline source of truth.
 ```jsonc
 {
  "model_id": "wan-t2v-1.3b-2gpu",
+  "workload_id": "wan-t2v",
+  "variant_id": "1.3b-sp2",
+  "benchmark_version": 2,
  "timestamp": "2026-05-08T22:00:00+00:00",
  "commit_sha": "<full sha>",
  "gpu_type": "NVIDIA L40S",
@@ -343,6 +346,9 @@ New records compare only against the same `model_id`, `gpu_type`,
 `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.

 ## Environment variable reference

@@ -66,6 +66,9 @@ 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",
@@ -153,11 +156,18 @@ def _record_metadata(run_source: str, result: dict[str, Any]) -> dict[str, Any]:


 def _identity_metadata(result: dict[str, Any]) -> dict[str, Any]:
-    return {
+    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]:
@@ -78,8 +78,9 @@ def build_recipe_from_benchmark_config(
    cfg: Mapping[str, Any],
    *,
    attention_backend: str | None = None,
-    resolved_attention_backend: str | None = None,
-    resolved_model_revision: 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.

@@ -90,7 +91,8 @@ def build_recipe_from_benchmark_config(
    model = dict(cfg.get("model") or {})
    init_kwargs = dict(cfg.get("init_kwargs") or {})
    generation_kwargs = dict(cfg.get("generation_kwargs") or {})
-    prompts = list(cfg.get("test_prompts") or ["A cinematic video."])
+    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")

@@ -140,6 +142,19 @@ def normalize_model_path(value: Any) -> str | None:
    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,
@@ -358,6 +373,7 @@ __all__ = [
    "normalize_model_path",
    "profile_id",
    "recipe_fingerprint",
+    "resolved_revision_from_model_path",
    "build_recipe_from_benchmark_config",
    "sha256_hexdigest",
    "software_profile",
@@ -123,6 +123,9 @@ 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,
        },
@@ -145,6 +148,9 @@ def test_normalized_record_preserves_identity_metadata(monkeypatch):

    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}
@@ -155,6 +161,24 @@ def test_normalized_record_preserves_identity_metadata(monkeypatch):
    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_baseline_eligibility_only_for_successful_scheduled_main():
    assert compare_baseline._is_baseline_eligible("scheduled_main", True) is True
    assert compare_baseline._is_baseline_eligible("scheduled_main", False) is False
@@ -10,6 +10,7 @@ from fastvideo.tests.performance.identity import (
    hardware_profile,
    hardware_profile_id,
    recipe_fingerprint,
+    resolved_revision_from_model_path,
    software_profile,
    software_profile_id,
 )
@@ -80,6 +81,20 @@ def test_output_path_does_not_change_recipe_fingerprint():
    assert _fingerprint(cfg) == _fingerprint(with_output_path)


+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)
@@ -211,3 +226,38 @@ def test_environment_metadata_is_separate_audit_fingerprint():

    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,
+    }
@@ -12,6 +12,7 @@ import os
 import time
 from collections.abc import Mapping
 from datetime import datetime, timezone
+from typing import Any

 import torch
 import pytest
@@ -25,6 +26,7 @@ from fastvideo.tests.performance.identity import (
    hardware_profile,
    hardware_profile_id,
    recipe_fingerprint,
+    resolved_revision_from_model_path,
    software_profile,
    software_profile_id,
 )
@@ -237,8 +239,73 @@ def _write_results(results):
    logger.info("Performance results written to %s", filepath)


-def _build_identity_fields(cfg, init_kwargs):
-    recipe = build_recipe_from_benchmark_config(cfg)
+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)
+    modules = getattr(pipeline, "modules", {}) or {}
+    backends: set[str] = set()
+
+    for module in modules.values():
+        if not isinstance(module, torch.nn.Module):
+            continue
+        for submodule in module.modules():
+            backend = getattr(submodule, "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):
+    return {
+        key: cfg[key]
+        for key in ("workload_id", "variant_id", "benchmark_version")
+        if cfg.get(key) is not None
+    }
+
+
+def _build_identity_fields(cfg, init_kwargs, prompt, runtime_identity):
+    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()
@@ -255,6 +322,7 @@ def _build_identity_fields(cfg, init_kwargs):
        "software_profile_id": software_profile_id(sw_profile),
        "environment_metadata": env_metadata,
        "environment_fingerprint": environment_fingerprint(env_metadata),
+        **_benchmark_identity_fields(cfg),
    }


@@ -295,6 +363,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)
@@ -349,7 +418,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),
+        **_build_identity_fields(cfg, init_kwargs, prompt, runtime_identity),
    }

    logger.info(