[bugfix]: enable perf stage timing in spawned workers

[ci] Reuse performance tracking HF sync for dashboard

  Add a local sync marker after successfully downloading the performance
  tracking snapshot from Hugging Face. The dashboard now reuses that existing
  snapshot when it runs after compare_baseline.py in the same CI job, avoiding
  a second snapshot_download call.

  If no prior sync marker exists, dashboard still falls back to syncing from HF,
  so local and standalone dashboard runs keep working.

.buildkite/performance-benchmarks/tests/wan-t2v-1.3b.json

@@ -37,16 +37,16 @@
     "L40S": {
       "max_generation_time_s": 34.0,
       "max_peak_memory_mb": 11000.0,
-      "max_text_encoder_time_s": 2.0,
-      "max_dit_time_s": 25.0,
-      "max_vae_decode_time_s": 5.0
+      "max_text_encoder_time_s": 5.0,
+      "max_dit_time_s": 10.0,
+      "max_vae_decode_time_s": 10.0
     },
     "default": {
       "max_generation_time_s": 120.0,
       "max_peak_memory_mb": 30000.0,
-      "max_text_encoder_time_s": 2.0,
-      "max_dit_time_s": 25.0,
-      "max_vae_decode_time_s": 5.0
+      "max_text_encoder_time_s": 5.0,
+      "max_dit_time_s": 10.0,
+      "max_vae_decode_time_s": 10.0
     }
   }
 }

fastvideo/tests/performance/compare_baseline.py

@@ -55,11 +55,9 @@ LOWER_IS_BETTER_METRICS = {
 
 
 def _should_persist_tracking() -> bool:
-    # test_scope = os.environ.get("TEST_SCOPE", "")
-    # branch = os.environ.get("BUILDKITE_BRANCH", "")
-    # return test_scope == "full" and branch == "main"
-    return True
-
+    test_scope = os.environ.get("TEST_SCOPE", "")
+    branch = os.environ.get("BUILDKITE_BRANCH", "")
+    return test_scope == "full" and branch == "main"
 
 def _load_current_results() -> list[dict[str, Any]]:
     pattern = os.path.join(RESULTS_DIR, "perf_*.json")

fastvideo/tests/performance/dashboard.py

@@ -135,7 +135,7 @@ def render_html(figs: list, skipped_metrics: list[dict[str, object]],
 def main() -> None:
     days = int(os.environ.get("DASHBOARD_DAYS", "30"))
 
-    local_dir = sync_from_hf("/tmp/perf-tracking")
+    local_dir = sync_from_hf("/tmp/perf-tracking", reuse_existing=True)
     df = load_as_dataframe(local_dir, days=days)
 
     if df.empty:

fastvideo/tests/performance/hf_store.py

@@ -25,6 +25,7 @@ from huggingface_hub import HfApi, snapshot_download
 
 HF_REPO_ID: str = os.environ.get("HF_REPO_ID", "FastVideo/performance-tracking")
 HF_TOKEN: str | None = os.environ.get("HF_API_KEY")
+SYNC_MARKER = ".hf_sync_complete"
 
 # ---------------------------------------------------------------------------
 # Low-level helpers
@@ -51,7 +52,16 @@ def safe_float(value: Any) -> float | None:
 # ---------------------------------------------------------------------------
 
 
-def sync_from_hf(local_dir: str, *, strict: bool = False) -> str:
+def _sync_marker_path(local_dir: str) -> str:
+    return os.path.join(local_dir, SYNC_MARKER)
+
+
+def sync_from_hf(
+    local_dir: str,
+    *,
+    strict: bool = False,
+    reuse_existing: bool = False,
+) -> str:
     """Download the HF dataset repo snapshot to *local_dir*.
 
     Returns *local_dir* so callers can chain: ``load_records(sync_from_hf(...))``.
@@ -61,7 +71,19 @@ def sync_from_hf(local_dir: str, *, strict: bool = False) -> str:
     unavailable. Callers that depend on the sync for correctness (e.g. the
     main-branch baseline writer) must pass ``strict=True`` so that misconfig
     or transient HF errors fail loud rather than silently reset the baseline.
+
+    When ``reuse_existing=True``, a previous successful sync in ``local_dir``
+    is reused. This avoids duplicate HF snapshot checks when compare and
+    dashboard scripts run sequentially in the same CI job.
     """
+    marker_path = _sync_marker_path(local_dir)
+    if reuse_existing and os.path.exists(marker_path):
+        print(f"hf_store: reusing existing sync at {local_dir}")
+        return local_dir
+
+    if not reuse_existing and os.path.exists(marker_path):
+        os.remove(marker_path)
+
     if not HF_REPO_ID:
         msg = "hf_store: HF_REPO_ID not set"
         if strict:
@@ -78,6 +100,12 @@ def sync_from_hf(local_dir: str, *, strict: bool = False) -> str:
             token=HF_TOKEN,
             allow_patterns="*.json",
         )
+        os.makedirs(local_dir, exist_ok=True)
+        with open(marker_path, "w", encoding="utf-8") as marker:
+            json.dump({
+                "repo_id": HF_REPO_ID,
+                "synced_at": datetime.now(timezone.utc).isoformat(),
+            }, marker)
     except Exception as exc:
         if strict:
             raise

fastvideo/tests/performance/test_inference_performance.py

@@ -10,6 +10,7 @@ import glob
 import json
 import os
 import time
+from collections.abc import Mapping
 from datetime import datetime, timezone
 
 import torch
@@ -24,13 +25,16 @@ logger = init_logger(__name__)
 
 STAGE_METRIC_MAP: dict[str, str] = {
     "TextEncoderStage": "text_encoder_time_s",
+    "TextEncodingStage": "text_encoder_time_s",
     "CLIPTextEncoderStage": "text_encoder_time_s",
     "T5TextEncoderStage": "text_encoder_time_s",
     "DiTStage": "dit_time_s",
     "DenoisingStage": "dit_time_s",
+    "DmdDenoisingStage": "dit_time_s",
     "TransformerStage": "dit_time_s",
     "VAEDecodeStage": "vae_decode_time_s",
     "VAEDecoderStage": "vae_decode_time_s",
+    "DecodingStage": "vae_decode_time_s",
     "DecodeStage": "vae_decode_time_s",
 }
 
@@ -95,7 +99,10 @@ def _extract_component_times(result: dict) -> dict[str, float | None]:
     logging_info = result.get("logging_info")
     if logging_info is None:
         return component_times
-    stages: dict = getattr(logging_info, "stages", {}) or {}
+    if isinstance(logging_info, Mapping):
+        stages: dict = logging_info.get("stages", {}) or {}
+    else:
+        stages: dict = getattr(logging_info, "stages", {}) or {}
     if not stages:
         return component_times
     logger.info("Discovered pipeline stages: %s", list(stages.keys()))
@@ -279,9 +286,15 @@ def test_inference_performance(cfg):
     (text encoder, DiT, VAE decode). Assert each against device-aware thresholds.
     """
 
-    original = envs.environment_variables["FASTVIDEO_STAGE_LOGGING"]
+    original_env = os.environ.get("FASTVIDEO_STAGE_LOGGING")
+    original_getter = envs.environment_variables["FASTVIDEO_STAGE_LOGGING"]
+    os.environ["FASTVIDEO_STAGE_LOGGING"] = "1"
     envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] = lambda: True
     try:
         _run_benchmark(cfg)
     finally:
-        envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] = original
+        if original_env is None:
+            os.environ.pop("FASTVIDEO_STAGE_LOGGING", None)
+        else:
+            os.environ["FASTVIDEO_STAGE_LOGGING"] = original_env
+        envs.environment_variables["FASTVIDEO_STAGE_LOGGING"] = original_getter