[bugfix] PAN2: pre-commit and unit-test fixes

- mypy types for the guardrail broadcast and the negative-prompt fill; yapf kept off the negative prompt.
- Classify PAN2PipelineConfig's kernel and guardrail fields in the schema parity inventory.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/design/inference_schema_parity_inventory.yaml

@@ -307,6 +307,21 @@ surfaces:
       i2v_skip_first_latent_cfg:
         sources:
           - fastvideo.configs.pipelines.pan2.PAN2PipelineConfig
+      vae_decode_kernel:
+        sources:
+          - fastvideo.configs.pipelines.pan2.PAN2PipelineConfig
+      vae_decode_kernel_parallel:
+        sources:
+          - fastvideo.configs.pipelines.pan2.PAN2PipelineConfig
+      dit_fused_kernels:
+        sources:
+          - fastvideo.configs.pipelines.pan2.PAN2PipelineConfig
+      enable_guardrails:
+        sources:
+          - fastvideo.configs.pipelines.pan2.PAN2PipelineConfig
+      offload_guardrail_models:
+        sources:
+          - fastvideo.configs.pipelines.pan2.PAN2PipelineConfig
       text_encoder_max_lengths:
         sources:
           - fastvideo.configs.pipelines.hunyuan15.Hunyuan15I2V480PStepDistilledConfig

fastvideo/pipelines/basic/pan2/guardrails.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 import importlib.util
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
-from typing import TypeVar
+from typing import Any, TypeVar, cast
 
 import numpy as np
 import torch
@@ -48,7 +48,7 @@ def _on_first_rank(fn: Callable[[], T], what: str) -> T:
     next collective until it times out.
     """
     rank, world_size = _sp_rank_and_size()
-    outcome: tuple[bool, object] | None = None
+    outcome: tuple[bool, Any] | None = None
     if rank == 0:
         try:
             outcome = (True, fn())
@@ -58,10 +58,11 @@ def _on_first_rank(fn: Callable[[], T], what: str) -> T:
             raise
     if world_size > 1:
         outcome = get_sp_group().broadcast_object(outcome, src=0)
+    assert outcome is not None
     ok, value = outcome
     if not ok:
         raise RuntimeError(f"The first sequence-parallel rank failed to {what}: {value}")
-    return value
+    return cast(T, value)
 
 
 class PAN2Guardrails:
@@ -105,7 +106,7 @@ class PAN2Guardrails:
                 torch.cuda.empty_cache()
 
     def _passes(self, check: Callable[[], bool]) -> bool:
-        return bool(_on_first_rank(lambda: bool(check()), "run the guardrail check"))
+        return bool(_on_first_rank(check, "run the guardrail check"))
 
     def check_prompt(self, prompt: str) -> None:
 
@@ -156,5 +157,9 @@ class PAN2VideoGuardrailStage(PipelineStage):
     def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
         # `batch.output` is `[B, 3, T, H, W]` in [0, 1]; `pixels_to_uint8` quantizes as the generator does.
         for video in batch.output:
-            self.guardrails.check_video(lambda video=video: pixels_to_uint8(video).permute(1, 2, 3, 0).cpu().numpy())
+
+            def frames(video: torch.Tensor = video) -> np.ndarray:
+                return pixels_to_uint8(video).permute(1, 2, 3, 0).cpu().numpy()
+
+            self.guardrails.check_video(frames)
         return batch

fastvideo/pipelines/basic/pan2/negative_prompt.py

@@ -6,7 +6,9 @@ the clip, as the reference sampler does for records without a negative prompt of
 """
 
 import json
+from typing import Any
 
+# yapf: disable
 PAN2_DEFAULT_NEGATIVE_PROMPT = {
     "video_style": "low quality, blur, softness, reduced detail, noise, compression, clipped highlights, "
     "blurred sidebar, black border, letterboxing, watermark, subtitle, AIGC text, oily waxy "
@@ -149,13 +151,14 @@ PAN2_DEFAULT_NEGATIVE_PROMPT = {
         ],
     },
 }
+# yapf: enable
 
 
 def render_pan2_default_negative_prompt(num_frames: int, fps: float) -> str:
     """The default negative prompt as compact one-line JSON, with every time range spanning the clip."""
     span = f"0 to {(num_frames - 1) / fps:g}"
 
-    def fill(node):
+    def fill(node: Any) -> Any:
         if isinstance(node, dict):
             return {key: span if key == "time_range_sec" else fill(value) for key, value in node.items()}
         if isinstance(node, list):