[diffusion][benchmark] Add reproducible request-manifest offline benchmark (#32917)

python/sglang/multimodal_gen/benchmarks/bench_offline_throughput.py

@@ -22,19 +22,35 @@ python -m sglang.multimodal_gen.benchmarks.bench_offline_throughput \\
     --batch-size 1 \\
     --num-inference-steps 20 \\
     --output-file results.json
+
+## Reproducible JSONL request manifest with durable output evidence
+python -m sglang.multimodal_gen.benchmarks.bench_offline_throughput \\
+    --model-path Wan-AI/Wan2.1-T2V-1.3B-Diffusers \\
+    --request-manifest workload.jsonl \\
+    --save-output-dir artifacts \\
+    --output-file results.jsonl
 """
 
 import argparse
 import dataclasses
 import json
+import os
+import re
+import statistics
 import time
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
 import torch
 from tqdm import tqdm
 
 from sglang.multimodal_gen.benchmarks.datasets import RandomDataset, VBenchDataset
+from sglang.multimodal_gen.benchmarks.request_manifest import (
+    LoadedRequestManifest,
+    file_sha256,
+    load_request_manifest,
+)
 from sglang.multimodal_gen.runtime.entrypoints.diffusion_generator import DiffGenerator
 from sglang.multimodal_gen.runtime.server_args import ServerArgs, set_global_server_args
 from sglang.multimodal_gen.runtime.utils.logging_utils import (
@@ -57,6 +73,21 @@ class BatchOutput:
     peak_memory_mb: float = 0.0
     success: bool = False
     error: str = ""
+    requests: List["RequestOutput"] = dataclasses.field(default_factory=list)
+
+
+@dataclass
+class RequestOutput:
+    """Evidence recorded for one benchmark request."""
+
+    request_id: str
+    prompt: str
+    sampling_params: Dict[str, Any]
+    success: bool
+    latency_seconds: float
+    error: str = ""
+    output_file_paths: List[str] = dataclasses.field(default_factory=list)
+    output_sha256: List[str] = dataclasses.field(default_factory=list)
 
 
 @dataclass
@@ -84,10 +115,12 @@ class BenchArgs:
     batch_size: int = 1
     random_request_config: str = None
     random_request_seed: int = 42
+    request_manifest: str = ""
 
     # Benchmark Execution
     skip_warmup: bool = False
     output_file: str = ""
+    save_output_dir: str = ""
     disable_tqdm: bool = False
 
     # Profiling
@@ -181,6 +214,15 @@ class BenchArgs:
             default=42,
             help="Random seed for sampling request profiles (default: 42).",
         )
+        parser.add_argument(
+            "--request-manifest",
+            type=str,
+            default="",
+            help=(
+                "JSONL request manifest. When set, every non-empty line is run once "
+                "and --dataset/--dataset-path/--num-prompts are ignored."
+            ),
+        )
 
         # Benchmark Execution
         parser.add_argument(
@@ -192,6 +234,15 @@ class BenchArgs:
             default="",
             help="Output JSON file for results (append mode)",
         )
+        parser.add_argument(
+            "--save-output-dir",
+            type=str,
+            default="",
+            help=(
+                "Persist generated media in this directory and record a SHA256 "
+                "digest for each output."
+            ),
+        )
         parser.add_argument(
             "--disable-tqdm",
             action="store_true",
@@ -240,33 +291,77 @@ def generate_batch(
     bench_args: BenchArgs,
     prompts: List[str],
     user_sampling_params: List[Dict[str, Any]],
+    request_ids: Optional[List[str]] = None,
 ) -> BatchOutput:
     """Generate batch of images/videos synchronously."""
     assert len(user_sampling_params) == len(prompts), (
         f"user_sampling_params length ({len(user_sampling_params)}) must match "
         f"prompts length ({len(prompts)})"
     )
+    if request_ids is None:
+        request_ids = [f"request-{idx:05d}" for idx in range(len(prompts))]
+    assert len(request_ids) == len(prompts), (
+        f"request_ids length ({len(request_ids)}) must match "
+        f"prompts length ({len(prompts)})"
+    )
 
     output = BatchOutput()
     start_time = time.perf_counter()
 
     torch.get_device_module().reset_peak_memory_stats()
 
-    for prompt, params in zip(prompts, user_sampling_params):
+    for prompt, params, request_id in zip(prompts, user_sampling_params, request_ids):
+        request_start_time = time.perf_counter()
         try:
             sampling_params_kwargs = dict(params)
             sampling_params_kwargs["prompt"] = prompt
+            sampling_params_kwargs["request_id"] = request_id
             result = engine.generate(sampling_params_kwargs=sampling_params_kwargs)
 
-            if result is not None:
-                if isinstance(result, list):
-                    output.total_frames += len(result)
-                else:
-                    output.total_frames += 1
+            results = result if isinstance(result, list) else [result]
+            results = [item for item in results if item is not None]
+            if not results:
+                raise RuntimeError("Engine returned no generation result")
+
+            output_paths = [
+                str(item.output_file_path)
+                for item in results
+                if item.output_file_path is not None
+            ]
+            output_hashes = []
+            for output_path in output_paths:
+                if not os.path.isfile(output_path):
+                    raise RuntimeError(
+                        f"Generated output does not exist: {output_path}"
+                    )
+                output_hashes.append(file_sha256(output_path))
+
+            output.total_frames += int(params.get("num_frames", 1)) * len(results)
             output.num_samples += 1
+            output.requests.append(
+                RequestOutput(
+                    request_id=request_id,
+                    prompt=prompt,
+                    sampling_params=dict(params),
+                    success=True,
+                    latency_seconds=time.perf_counter() - request_start_time,
+                    output_file_paths=output_paths,
+                    output_sha256=output_hashes,
+                )
+            )
         except Exception as e:
             logger.error(f"Generation failed for prompt '{prompt[:50]}...': {e}")
             output.error = str(e)
+            output.requests.append(
+                RequestOutput(
+                    request_id=request_id,
+                    prompt=prompt,
+                    sampling_params=dict(params),
+                    success=False,
+                    latency_seconds=time.perf_counter() - request_start_time,
+                    error=str(e),
+                )
+            )
 
     output.latency = time.perf_counter() - start_time
     output.latency_per_sample = output.latency / len(prompts) if prompts else 0.0
@@ -290,18 +385,24 @@ def calculate_metrics(
     all_sampling_params: Optional[List[Dict[str, Any]]] = None,
 ) -> Dict[str, Any]:
     """Calculate generation-specific throughput metrics."""
-    successful = [o for o in outputs if o.success]
-    num_success = sum(o.num_samples for o in successful)
-    total_frames = sum(o.total_frames for o in successful)
+    num_success = sum(o.num_samples for o in outputs)
+    total_frames = sum(o.total_frames for o in outputs)
     peak_memory = max((o.peak_memory_mb for o in outputs), default=0)
+    request_outputs = [request for output in outputs for request in output.requests]
+    successful_request_outputs = [
+        request for request in request_outputs if request.success
+    ]
+    request_latencies = [
+        request.latency_seconds for request in successful_request_outputs
+    ]
 
     width, height, frames = resolution
     if all_sampling_params:
         total_pixels = sum(
-            p.get("width", width)
-            * p.get("height", height)
-            * p.get("num_frames", frames)
-            for p in all_sampling_params[:num_success]
+            request.sampling_params.get("width", width)
+            * request.sampling_params.get("height", height)
+            * request.sampling_params.get("num_frames", frames)
+            for request in successful_request_outputs
         )
     else:
         total_pixels = num_success * width * height * frames
@@ -324,6 +425,15 @@ def calculate_metrics(
         "latency_per_request_seconds": (
             total_duration / num_success if num_success > 0 else 0
         ),
+        "request_latency_mean_seconds": (
+            statistics.fmean(request_latencies) if request_latencies else 0
+        ),
+        "request_latency_median_seconds": (
+            statistics.median(request_latencies) if request_latencies else 0
+        ),
+        "request_latency_min_seconds": min(request_latencies, default=0),
+        "request_latency_max_seconds": max(request_latencies, default=0),
+        "request_results": [dataclasses.asdict(request) for request in request_outputs],
         "peak_memory_mb": peak_memory,
     }
 
@@ -339,7 +449,12 @@ def throughput_test(
     logger.info("Starting offline throughput benchmark...")
 
     engine = initialize_engine(server_args)
+    bench_args.task_name = str(engine.server_args.pipeline_config.task_type)
 
+    if bench_args.request_manifest and bench_args.random_request_config:
+        raise ValueError(
+            "--request-manifest and --random-request-config are mutually exclusive"
+        )
     if bench_args.random_request_config and bench_args.dataset != "random":
         raise ValueError(
             "--random-request-config can only be used with --dataset random"
@@ -350,12 +465,27 @@ def throughput_test(
             "bench_offline_throughput currently supports only --num-outputs-per-prompt 1"
         )
 
-    logger.info(f"Loading {bench_args.dataset} dataset...")
-    if bench_args.dataset == "vbench":
-        bench_args.task_name = str(engine.server_args.pipeline_config.task_type)
+    manifest: Optional[LoadedRequestManifest] = None
+    if bench_args.request_manifest:
+        logger.info(f"Loading request manifest {bench_args.request_manifest}...")
+        manifest = load_request_manifest(bench_args.request_manifest)
+        manifest_requests = manifest.requests
+        total_count = len(manifest_requests)
+        all_prompts = [request.prompt for request in manifest_requests]
+        all_request_ids = [request.request_id for request in manifest_requests]
+        all_sampling_params = [
+            dict(request.sampling_params) for request in manifest_requests
+        ]
+    elif bench_args.dataset == "vbench":
+        logger.info(f"Loading {bench_args.dataset} dataset...")
         dataset = VBenchDataset(bench_args)
+        total_count = min(bench_args.num_prompts, len(dataset))
+        dataset_requests = [dataset[i] for i in range(total_count)]
     elif bench_args.dataset == "random":
+        logger.info(f"Loading {bench_args.dataset} dataset...")
         dataset = RandomDataset(bench_args)
+        total_count = min(bench_args.num_prompts, len(dataset))
+        dataset_requests = [dataset[i] for i in range(total_count)]
     else:
         raise ValueError(f"Unknown dataset: {bench_args.dataset}")
 
@@ -365,6 +495,7 @@ def throughput_test(
         "height": bench_args.height,
         "width": bench_args.width,
         "num_frames": bench_args.num_frames,
+        "fps": bench_args.fps,
         "num_outputs_per_prompt": bench_args.num_outputs_per_prompt,
         "seed": bench_args.seed,
         "profile": bench_args.profile,
@@ -374,28 +505,67 @@ def throughput_test(
     if bench_args.disable_safety_checker:
         _sampling_params["safety_checker"] = None
 
-    total_count = min(bench_args.num_prompts, len(dataset))
-    all_prompts = [dataset[i].prompt for i in range(total_count)]
-
-    if bench_args.random_request_config:
+    if manifest is None:
+        all_prompts = [request.prompt for request in dataset_requests]
+        all_request_ids = [request.request_id for request in dataset_requests]
         all_sampling_params = []
-        for i in range(total_count):
+        for i, request in enumerate(dataset_requests):
             params = dict(_sampling_params)
-            params.update(dataset.get_sampling_params(i))
+            if bench_args.random_request_config:
+                params.update(dataset.get_sampling_params(i))
+            if request.image_paths:
+                params["image_path"] = request.image_paths
             all_sampling_params.append(params)
     else:
-        all_sampling_params = [_sampling_params] * total_count
+        for params in all_sampling_params:
+            defaults = dict(_sampling_params)
+            defaults.update(params)
+            params.clear()
+            params.update(defaults)
+
+    if bench_args.save_output_dir:
+        output_dir = Path(bench_args.save_output_dir).expanduser().resolve()
+        output_dir.mkdir(parents=True, exist_ok=True)
+        for index, (request_id, params) in enumerate(
+            zip(all_request_ids, all_sampling_params)
+        ):
+            safe_request_id = re.sub(r"[^A-Za-z0-9._-]+", "_", request_id).strip("._")
+            safe_request_id = safe_request_id or "request"
+            params.update(
+                {
+                    "output_path": str(output_dir),
+                    "output_file_name": f"{index:05d}-{safe_request_id}",
+                    "return_file_paths_only": True,
+                    "save_output": True,
+                }
+            )
 
     if not bench_args.skip_warmup:
         logger.info("Running warmup batch...")
         warmup_count = min(bench_args.batch_size, total_count)
         warmup_prompts = all_prompts[:warmup_count]
         warmup_sampling_params = [
-            {**p, "profile": False} for p in all_sampling_params[:warmup_count]
+            {
+                **p,
+                "profile": False,
+                "save_output": False,
+                "return_file_paths_only": False,
+                "output_path": None,
+                "output_file_name": None,
+            }
+            for p in all_sampling_params[:warmup_count]
         ]
-        generate_batch(engine, bench_args, warmup_prompts, warmup_sampling_params)
+        generate_batch(
+            engine,
+            bench_args,
+            warmup_prompts,
+            warmup_sampling_params,
+            request_ids=[
+                f"warmup-{request_id}" for request_id in all_request_ids[:warmup_count]
+            ],
+        )
 
-    logger.info(f"Running benchmark with {bench_args.num_prompts} prompts...")
+    logger.info(f"Running benchmark with {total_count} prompts...")
     outputs: List[BatchOutput] = []
 
     start_time = time.perf_counter()
@@ -411,9 +581,14 @@ def throughput_test(
         batch_end = min(batch_start + bench_args.batch_size, total_count)
         batch_prompts = all_prompts[batch_start:batch_end]
         batch_sampling_params = all_sampling_params[batch_start:batch_end]
+        batch_request_ids = all_request_ids[batch_start:batch_end]
 
         batch_output = generate_batch(
-            engine, bench_args, batch_prompts, batch_sampling_params
+            engine,
+            bench_args,
+            batch_prompts,
+            batch_sampling_params,
+            request_ids=batch_request_ids,
         )
         outputs.append(batch_output)
 
@@ -438,7 +613,7 @@ def throughput_test(
     )
 
     if bench_args.output_file:
-        save_results(metrics, bench_args, server_args)
+        save_results(metrics, bench_args, server_args, manifest=manifest)
 
     return metrics
 
@@ -480,6 +655,12 @@ def display_results(
     print_value_formatted(
         "Latency Per Request (sec):", metrics["latency_per_request_seconds"]
     )
+    print_value_formatted(
+        "Request Latency Mean (sec):", metrics["request_latency_mean_seconds"]
+    )
+    print_value_formatted(
+        "Request Latency Median (sec):", metrics["request_latency_median_seconds"]
+    )
     print_value_formatted("Peak Memory (MB):", metrics["peak_memory_mb"])
     print_divider(110, "=")
 
@@ -488,6 +669,7 @@ def save_results(
     metrics: Dict[str, Any],
     bench_args: BenchArgs,
     server_args: ServerArgs,
+    manifest: Optional[LoadedRequestManifest] = None,
 ):
     """Save benchmark results to JSON file."""
     result = {
@@ -496,15 +678,22 @@ def save_results(
             "model_path": server_args.model_path,
             "task_type": bench_args.task_name,
             "backend": "engine",
+            "request_manifest": manifest.path if manifest else None,
+            "request_manifest_sha256": manifest.sha256 if manifest else None,
         },
         "configuration": {
             "num_inference_steps": bench_args.num_inference_steps,
             "guidance_scale": bench_args.guidance_scale,
             "seed": bench_args.seed,
             "batch_size": bench_args.batch_size,
-            "num_prompts": bench_args.num_prompts,
+            "num_prompts": metrics["num_requests"],
             "resolution": f"{bench_args.width}x{bench_args.height}x{bench_args.num_frames}",
-            "dataset": bench_args.dataset,
+            "dataset": "request_manifest" if manifest else bench_args.dataset,
+            "save_output_dir": (
+                str(Path(bench_args.save_output_dir).expanduser().resolve())
+                if bench_args.save_output_dir
+                else None
+            ),
         },
         "results": metrics,
     }

python/sglang/multimodal_gen/benchmarks/request_manifest.py

@@ -0,0 +1,183 @@
+"""JSONL request manifests for reproducible offline diffusion benchmarks."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+_TOP_LEVEL_SAMPLING_FIELDS = {
+    "fps",
+    "guidance_scale",
+    "height",
+    "image_paths",
+    "negative_prompt",
+    "num_frames",
+    "num_inference_steps",
+    "num_outputs_per_prompt",
+    "seed",
+    "width",
+}
+_RESERVED_SAMPLING_FIELDS = {
+    "image_path",
+    "output_file_name",
+    "output_path",
+    "prompt",
+    "request_id",
+    "return_file_paths_only",
+    "save_output",
+}
+_ALLOWED_FIELDS = {
+    "prompt",
+    "request_id",
+    "sampling_params",
+    *_TOP_LEVEL_SAMPLING_FIELDS,
+}
+
+
+@dataclass(frozen=True)
+class ManifestRequest:
+    """One fully resolved request from a benchmark manifest."""
+
+    request_id: str
+    prompt: str
+    sampling_params: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class LoadedRequestManifest:
+    """Parsed requests plus the digest of the exact input manifest."""
+
+    path: str
+    sha256: str
+    requests: list[ManifestRequest]
+
+
+def file_sha256(path: str | Path) -> str:
+    """Return the SHA256 digest of a file without loading it all into memory."""
+    digest = hashlib.sha256()
+    with Path(path).open("rb") as file:
+        for chunk in iter(lambda: file.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _is_url(value: str) -> bool:
+    return urlparse(value).scheme in {"http", "https"}
+
+
+def _resolve_image_paths(value: Any, base_dir: Path, line_number: int) -> Any:
+    if isinstance(value, str):
+        image_paths = [value]
+        return_scalar = True
+    elif (
+        isinstance(value, list)
+        and value
+        and all(isinstance(item, str) and item for item in value)
+    ):
+        image_paths = value
+        return_scalar = False
+    else:
+        raise ValueError(
+            f"Manifest line {line_number}: image_paths must be a non-empty "
+            "string or list of strings"
+        )
+
+    resolved = [
+        item if _is_url(item) else str((base_dir / item).resolve())
+        for item in image_paths
+    ]
+    return resolved[0] if return_scalar else resolved
+
+
+def load_request_manifest(path: str | Path) -> LoadedRequestManifest:
+    """Load and validate a JSONL request manifest.
+
+    Relative condition-image paths are resolved against the manifest directory.
+    A manifest is authoritative: every non-empty line becomes exactly one request.
+    """
+    manifest_path = Path(path).expanduser().resolve()
+    if not manifest_path.is_file():
+        raise ValueError(f"Request manifest does not exist: {manifest_path}")
+
+    requests: list[ManifestRequest] = []
+    request_ids: set[str] = set()
+    with manifest_path.open(encoding="utf-8") as file:
+        for line_number, line in enumerate(file, start=1):
+            if not line.strip():
+                continue
+            try:
+                record = json.loads(line)
+            except json.JSONDecodeError as error:
+                raise ValueError(
+                    f"Manifest line {line_number}: invalid JSON: {error.msg}"
+                ) from error
+            if not isinstance(record, dict):
+                raise ValueError(
+                    f"Manifest line {line_number}: each line must be a JSON object"
+                )
+
+            unknown_fields = set(record) - _ALLOWED_FIELDS
+            if unknown_fields:
+                raise ValueError(
+                    f"Manifest line {line_number}: unsupported field(s): "
+                    f"{', '.join(sorted(unknown_fields))}"
+                )
+
+            prompt = record.get("prompt")
+            if not isinstance(prompt, str) or not prompt.strip():
+                raise ValueError(
+                    f"Manifest line {line_number}: prompt must be a non-empty string"
+                )
+
+            request_id = record.get("request_id", f"request-{line_number:05d}")
+            if not isinstance(request_id, str) or not request_id:
+                raise ValueError(
+                    f"Manifest line {line_number}: request_id must be a non-empty string"
+                )
+            if request_id in request_ids:
+                raise ValueError(
+                    f"Manifest line {line_number}: duplicate request_id {request_id!r}"
+                )
+            request_ids.add(request_id)
+
+            sampling_params = record.get("sampling_params", {})
+            if not isinstance(sampling_params, dict):
+                raise ValueError(
+                    f"Manifest line {line_number}: sampling_params must be an object"
+                )
+            sampling_params = dict(sampling_params)
+            reserved_fields = set(sampling_params) & _RESERVED_SAMPLING_FIELDS
+            if reserved_fields:
+                raise ValueError(
+                    f"Manifest line {line_number}: sampling_params cannot set reserved "
+                    f"field(s): {', '.join(sorted(reserved_fields))}"
+                )
+
+            for field in _TOP_LEVEL_SAMPLING_FIELDS - {"image_paths"}:
+                if field in record:
+                    sampling_params[field] = record[field]
+            if "image_paths" in record:
+                sampling_params["image_path"] = _resolve_image_paths(
+                    record["image_paths"], manifest_path.parent, line_number
+                )
+
+            requests.append(
+                ManifestRequest(
+                    request_id=request_id,
+                    prompt=prompt,
+                    sampling_params=sampling_params,
+                )
+            )
+
+    if not requests:
+        raise ValueError(f"Request manifest contains no requests: {manifest_path}")
+
+    return LoadedRequestManifest(
+        path=str(manifest_path),
+        sha256=file_sha256(manifest_path),
+        requests=requests,
+    )

python/sglang/multimodal_gen/test/unit/test_request_manifest.py

@@ -0,0 +1,190 @@
+import hashlib
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from sglang.multimodal_gen.benchmarks.bench_offline_throughput import (
+    BatchOutput,
+    BenchArgs,
+    RequestOutput,
+    calculate_metrics,
+    generate_batch,
+)
+from sglang.multimodal_gen.benchmarks.request_manifest import load_request_manifest
+
+
+def _write_jsonl(path, records):
+    path.write_text(
+        "".join(json.dumps(record) + "\n" for record in records), encoding="utf-8"
+    )
+
+
+def test_load_request_manifest_resolves_inputs_and_preserves_overrides(tmp_path):
+    image_path = tmp_path / "condition.png"
+    image_path.write_bytes(b"condition-image")
+    manifest_path = tmp_path / "workload.jsonl"
+    records = [
+        {
+            "request_id": "i2v-1",
+            "prompt": "Make the water move",
+            "image_paths": "condition.png",
+            "seed": 7,
+            "num_frames": 17,
+            "sampling_params": {"flow_shift": 5.0},
+        },
+        {
+            "prompt": "A lighthouse in a storm",
+            "image_paths": [
+                "https://example.com/reference.png",
+                "condition.png",
+            ],
+            "width": 832,
+            "height": 480,
+        },
+    ]
+    _write_jsonl(manifest_path, records)
+
+    manifest = load_request_manifest(manifest_path)
+
+    assert manifest.path == str(manifest_path.resolve())
+    assert manifest.sha256 == hashlib.sha256(manifest_path.read_bytes()).hexdigest()
+    assert [request.request_id for request in manifest.requests] == [
+        "i2v-1",
+        "request-00002",
+    ]
+    assert manifest.requests[0].sampling_params == {
+        "flow_shift": 5.0,
+        "seed": 7,
+        "num_frames": 17,
+        "image_path": str(image_path.resolve()),
+    }
+    assert manifest.requests[1].sampling_params["image_path"] == [
+        "https://example.com/reference.png",
+        str(image_path.resolve()),
+    ]
+
+
+@pytest.mark.parametrize(
+    ("record", "error"),
+    [
+        ({"prompt": ""}, "prompt must be a non-empty string"),
+        ({"prompt": "test", "typo": 1}, "unsupported field"),
+        (
+            {"prompt": "test", "sampling_params": {"output_path": "/tmp"}},
+            "cannot set reserved",
+        ),
+        (
+            {"prompt": "test", "image_paths": []},
+            "image_paths must be a non-empty",
+        ),
+    ],
+)
+def test_load_request_manifest_rejects_invalid_records(tmp_path, record, error):
+    manifest_path = tmp_path / "invalid.jsonl"
+    _write_jsonl(manifest_path, [record])
+
+    with pytest.raises(ValueError, match=error):
+        load_request_manifest(manifest_path)
+
+
+def test_load_request_manifest_rejects_duplicate_ids(tmp_path):
+    manifest_path = tmp_path / "duplicate.jsonl"
+    _write_jsonl(
+        manifest_path,
+        [
+            {"request_id": "same", "prompt": "first"},
+            {"request_id": "same", "prompt": "second"},
+        ],
+    )
+
+    with pytest.raises(ValueError, match="duplicate request_id"):
+        load_request_manifest(manifest_path)
+
+
+def test_load_request_manifest_rejects_empty_file(tmp_path):
+    manifest_path = tmp_path / "empty.jsonl"
+    manifest_path.write_text("\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="contains no requests"):
+        load_request_manifest(manifest_path)
+
+
+def test_metrics_use_the_successful_request_shapes_after_a_failure():
+    failed = RequestOutput(
+        request_id="failed",
+        prompt="first",
+        sampling_params={"width": 1024, "height": 1024, "num_frames": 81},
+        success=False,
+        latency_seconds=1.0,
+        error="expected failure",
+    )
+    succeeded = RequestOutput(
+        request_id="succeeded",
+        prompt="second",
+        sampling_params={"width": 832, "height": 480, "num_frames": 17},
+        success=True,
+        latency_seconds=2.0,
+    )
+    batch = BatchOutput(
+        num_samples=1,
+        total_frames=17,
+        success=True,
+        requests=[failed, succeeded],
+    )
+
+    metrics = calculate_metrics(
+        [batch],
+        total_duration=3.0,
+        resolution=(32, 32, 1),
+        num_requests=2,
+        all_sampling_params=[failed.sampling_params, succeeded.sampling_params],
+    )
+
+    assert metrics["successful_requests"] == 1
+    assert metrics["failed_requests"] == 1
+    assert metrics["total_pixels_generated"] == 832 * 480 * 17
+
+
+def test_generate_batch_records_request_id_output_path_and_digest(
+    tmp_path, monkeypatch
+):
+    output_path = tmp_path / "generated.mp4"
+    output_path.write_bytes(b"generated-video")
+
+    class FakeDeviceModule:
+        @staticmethod
+        def reset_peak_memory_stats():
+            return None
+
+        @staticmethod
+        def max_memory_allocated():
+            return 0
+
+    class FakeEngine:
+        @staticmethod
+        def generate(sampling_params_kwargs):
+            assert sampling_params_kwargs["request_id"] == "video-1"
+            return SimpleNamespace(output_file_path=output_path)
+
+    monkeypatch.setattr(
+        "sglang.multimodal_gen.benchmarks.bench_offline_throughput.torch.get_device_module",
+        lambda: FakeDeviceModule,
+    )
+
+    output = generate_batch(
+        FakeEngine(),
+        BenchArgs(),
+        prompts=["A moving test pattern"],
+        user_sampling_params=[{"num_frames": 17, "seed": 42}],
+        request_ids=["video-1"],
+    )
+
+    assert output.success
+    assert output.num_samples == 1
+    assert output.total_frames == 17
+    assert output.requests[0].request_id == "video-1"
+    assert output.requests[0].output_file_paths == [str(output_path)]
+    assert output.requests[0].output_sha256 == [
+        hashlib.sha256(output_path.read_bytes()).hexdigest()
+    ]