[do not merge] diffusion: pr ci consistency

python/sglang/multimodal_gen/test/server/consistency_threshold.json

@@ -0,0 +1,6 @@
+{
+    "diffusers_version": "0.36.0",
+    "cases": {},
+    "default_clip_threshold_image": 0.92,
+    "default_clip_threshold_video": 0.90
+}

python/sglang/multimodal_gen/test/server/test_server_common.py

@@ -35,8 +35,15 @@
 )
 from sglang.multimodal_gen.test.test_utils import (
     _consistency_gt_filenames,
+    check_diffusers_version_match,
+    compare_with_gt,
     extract_key_frames_from_video,
+    get_clip_threshold,
+    get_consistency_gt_candidates,
     get_dynamic_server_port,
+    gt_exists,
+    image_bytes_to_numpy,
+    load_gt_embeddings,
     wait_for_req_perf_record,
 )
 
@@ -417,6 +424,111 @@ def _dump_baseline_for_testcase(
 """
         logger.error(output)
 
+    def _validate_consistency(
+        self,
+        case: DiffusionTestCase,
+        content: bytes,
+    ) -> None:
+        """Validate output consistency against ground truth using CLIP similarity.
+
+        Args:
+            case: Test case configuration
+            content: Generated content bytes (image or video)
+
+        Raises:
+            pytest.fail: If consistency check fails (GT exists but doesn't match)
+
+        Note:
+            If GT doesn't exist, the test is skipped (not failed) so that new
+            cases don't block CI. Run the GT generation workflow after merging.
+
+        Environment Variables:
+            SGLANG_SKIP_CONSISTENCY: Set to "1" to skip consistency checks
+        """
+        # Allow skipping consistency checks for debugging/development
+        if os.environ.get("SGLANG_SKIP_CONSISTENCY", "0") == "1":
+            logger.info(
+                f"[Consistency] Skipping consistency check for {case.id} (SGLANG_SKIP_CONSISTENCY=1)"
+            )
+            return
+
+        # Skip if content is empty (e.g., video generation timed out)
+        if not content:
+            logger.warning(
+                f"[Consistency] Skipping consistency check for {case.id}: "
+                "content is empty (generation may have timed out)"
+            )
+            return
+
+        # Warn if diffusers version doesn't match GT generation version
+        check_diffusers_version_match()
+
+        num_gpus = case.server_args.num_gpus
+        is_video = case.server_args.modality == "video"
+
+        # Check GT exists - if not, skip with instructions to generate GT
+        output_format = case.sampling_params.output_format
+        if not gt_exists(
+            case.id, num_gpus, is_video=is_video, output_format=output_format
+        ):
+            names = ", ".join(
+                get_consistency_gt_candidates(
+                    case.id, num_gpus, is_video, output_format
+                )
+            )
+            logger.warning(
+                f"[Consistency] GT not found for '{case.id}'. "
+                f"Expected file(s): {names}. "
+                f"Run the 'Diffusion CI Ground Truth Generation' workflow after merging."
+            )
+            pytest.skip(
+                f"GT not found for {case.id}. "
+                f"Run GT generation workflow to enable consistency check."
+            )
+
+        # Load GT embeddings (format matches case; PIL converts to RGB for CLIP)
+        try:
+            gt_embeddings = load_gt_embeddings(
+                case.id, num_gpus, is_video=is_video, output_format=output_format
+            )
+        except Exception as e:
+            logger.warning(f"[Consistency] Failed to load GT for {case.id}: {e}")
+            pytest.skip(f"Failed to load GT for {case.id}: {e}")
+
+        # Convert output to frames
+        if is_video:
+            output_frames = extract_key_frames_from_video(content)
+        else:
+            output_frames = [image_bytes_to_numpy(content)]
+
+        threshold = get_clip_threshold(case)
+
+        # Compare frames with GT embeddings using CLIP similarity
+        result = compare_with_gt(
+            output_frames=output_frames,
+            gt_embeddings=gt_embeddings,
+            threshold=threshold,
+            case_id=case.id,
+        )
+
+        if not result.passed:
+            # Build detailed failure message
+            failed_frames = [
+                f"  Frame {d['frame_index']}: similarity={d['similarity']:.4f}"
+                for d in result.frame_details
+                if not d["passed"]
+            ]
+            pytest.fail(
+                f"Consistency check failed for {case.id}:\n"
+                f"  Min similarity: {result.min_similarity:.4f}\n"
+                f"  Threshold: {result.threshold}\n"
+                f"  Failed frames:\n" + "\n".join(failed_frames)
+            )
+
+        logger.info(
+            f"[Consistency] {case.id}: PASSED (min_similarity={result.min_similarity:.4f})"
+        )
+
     def _save_gt_output(
         self,
         case: DiffusionTestCase,
@@ -874,6 +986,9 @@ def test_diffusion_generation(
         self._test_v1_models_endpoint(diffusion_server, case)
         self._test_t2v_rejects_input_reference(diffusion_server, case)
 
+        # Validation 2: Consistency (reuse the same content)
+        self._validate_consistency(case, content)
+
         # LoRA API functionality test with E2E validation (only for LoRA-enabled cases)
         if case.server_args.lora_path or case.server_args.dynamic_lora_path:
             self._test_lora_api_functionality(diffusion_server, case, generate_fn)