[diffusion][feature] Add LingBot-Video T2I and TI2V generation modes (#5311)

Signed-off-by: wtz2333 <2955110911@qq.com>
Co-authored-by: Claude <noreply@anthropic.com>

tests/diffusion/models/lingbot_video/test_image_condition.py

@@ -0,0 +1,133 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import torch
+from PIL import Image
+from torch import nn
+
+from vllm_omni.diffusion.models.lingbot_video.image_condition import (
+    apply_clean_prefix,
+    geometry_align_image,
+    prepare_ti2v_image_condition,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.diffusion]
+
+
+def _gradient_image(width: int, height: int) -> Image.Image:
+    x = np.linspace(0, 255, width, dtype=np.uint8)[None, :]
+    y = np.linspace(0, 255, height, dtype=np.uint8)[:, None]
+    rgb = np.stack(
+        [
+            np.broadcast_to(x, (height, width)),
+            np.broadcast_to(y, (height, width)),
+            np.full((height, width), 127, dtype=np.uint8),
+        ],
+        axis=-1,
+    )
+    return Image.fromarray(rgb, mode="RGB")
+
+
+@pytest.mark.parametrize("source_size", [(160, 64), (64, 160), (96, 96)])
+def test_geometry_align_image_supports_source_aspect_ratios(source_size):
+    pixel = geometry_align_image(_gradient_image(*source_size), 64, 96)
+
+    assert pixel.shape == (1, 3, 1, 64, 96)
+    assert pixel.dtype == torch.float32
+    assert 0 <= pixel.min() <= pixel.max() <= 1
+
+
+@pytest.mark.parametrize(
+    ("source_size", "target_size"),
+    [((160, 64), (64, 96)), ((64, 160), (96, 64))],
+)
+def test_geometry_align_image_center_crops_without_stretching(source_size, target_size):
+    width, height = source_size
+    target_height, target_width = target_size
+    rgb = np.full((height, width, 3), (255, 0, 0), dtype=np.uint8)
+    if width > target_width:
+        left = (width - target_width) // 2
+        rgb[:, left : left + target_width] = (0, 255, 0)
+    else:
+        top = (height - target_height) // 2
+        rgb[top : top + target_height] = (0, 255, 0)
+
+    pixel = geometry_align_image(
+        Image.fromarray(rgb, mode="RGB"),
+        target_height,
+        target_width,
+    )
+
+    assert pixel.shape == (1, 3, 1, target_height, target_width)
+    expected_green = torch.tensor((0.0, 1.0, 0.0)).view(1, 3, 1, 1, 1)
+    assert torch.equal(pixel, expected_green.expand_as(pixel))
+
+
+class _LatentDistribution:
+    def __init__(self, latent: torch.Tensor):
+        self.latent = latent
+
+    def sample(self, generator=None):
+        del generator
+        return self.latent
+
+
+class _CapturingVAE(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.anchor = nn.Parameter(torch.zeros(()))
+        self.config = SimpleNamespace(
+            latents_mean=[0.0, 0.0, 0.0],
+            latents_std=[1.0, 1.0, 1.0],
+        )
+        self.encoded_pixel = None
+
+    def encode(self, pixel):
+        self.encoded_pixel = pixel.detach().clone()
+        return SimpleNamespace(latent_dist=_LatentDistribution(pixel))
+
+
+def test_ti2v_vlm_and_vae_paths_share_geometry_aligned_image():
+    vae = _CapturingVAE()
+    condition = prepare_ti2v_image_condition(
+        _gradient_image(160, 64),
+        height=64,
+        width=96,
+        vae=vae,
+        vision_patch_size=16,
+        device=torch.device("cpu"),
+    )
+
+    vae_pixel = (vae.encoded_pixel + 1.0) / 2.0
+    vae_image = vae_pixel[0, :, 0].permute(1, 2, 0).mul(255).byte().numpy()
+    assert condition.vlm_image.size == (96, 64)
+    np.testing.assert_allclose(
+        np.asarray(condition.vlm_image),
+        vae_image,
+        rtol=0,
+        atol=1,
+    )
+
+
+def test_apply_clean_prefix_only_overwrites_condition_frames():
+    latents = torch.arange(20, dtype=torch.float32).reshape(1, 1, 5, 2, 2)
+    original_tail = latents[:, :, 2:].clone()
+    clean = torch.full((1, 1, 2, 2, 2), -3.0)
+
+    returned = apply_clean_prefix(latents, clean)
+
+    assert returned is latents
+    assert torch.equal(latents[:, :, :2], clean)
+    assert torch.equal(latents[:, :, 2:], original_tail)
+
+
+def test_apply_clean_prefix_rejects_incompatible_shapes():
+    with pytest.raises(ValueError, match="shape must match"):
+        apply_clean_prefix(
+            torch.zeros(1, 2, 3, 4, 4),
+            torch.zeros(1, 3, 1, 4, 4),
+        )

tests/diffusion/models/lingbot_video/test_pipeline_lingbot_video.py

@@ -6,9 +6,10 @@ from types import SimpleNamespace
 import numpy as np
 import pytest
 import torch
+from PIL import Image
 from torch import nn
 
-from vllm_omni.diffusion.request import OmniDiffusionRequest
+from vllm_omni.diffusion.request import DUMMY_DIFFUSION_REQUEST_ID, OmniDiffusionRequest
 from vllm_omni.diffusion.worker.request_batch import DiffusionRequestBatch
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
 
@@ -22,6 +23,8 @@ def _make_pipeline():
     nn.Module.__init__(pipeline)
     pipeline.device = torch.device("cpu")
     pipeline.default_negative_prompt = "default negative"
+    pipeline.default_image_negative_prompt = "default image negative"
+    pipeline.img_prompt_template = "<image>"
     pipeline.od_config = SimpleNamespace(flow_shift=None)
     return pipeline
 
@@ -37,17 +40,24 @@ def test_lingbot_video_pipeline_import_and_registry():
         LingBotVideoPipeline,
         LingBotVideoTransformer3DModel,
         get_lingbot_video_post_process_func,
+        get_lingbot_video_pre_process_func,
+    )
+    from vllm_omni.diffusion.registry import (
+        _DIFFUSION_MODELS,
+        _DIFFUSION_POST_PROCESS_FUNCS,
+        _DIFFUSION_PRE_PROCESS_FUNCS,
     )
-    from vllm_omni.diffusion.registry import _DIFFUSION_MODELS, _DIFFUSION_POST_PROCESS_FUNCS
 
     assert LingBotVideoPipeline is not None
     assert LingBotVideoTransformer3DModel is not None
+    assert get_lingbot_video_pre_process_func is not None
     assert get_lingbot_video_post_process_func is not None
     assert _DIFFUSION_MODELS["LingBotVideoPipeline"] == (
         "lingbot_video",
         "pipeline_lingbot_video",
         "LingBotVideoPipeline",
     )
+    assert _DIFFUSION_PRE_PROCESS_FUNCS["LingBotVideoPipeline"] == "get_lingbot_video_pre_process_func"
     assert _DIFFUSION_POST_PROCESS_FUNCS["LingBotVideoPipeline"] == "get_lingbot_video_post_process_func"
 
 
@@ -58,6 +68,7 @@ def test_component_discovery_declarations():
     assert LingBotVideoPipeline._encoder_modules == ["text_encoder"]
     assert LingBotVideoPipeline._vae_modules == ["vae"]
     assert LingBotVideoPipeline.supports_step_execution is False
+    assert LingBotVideoPipeline.max_outputs_per_prompt == 1
 
 
 def test_extra_body_params_include_video_flow_shift_alias():
@@ -69,17 +80,47 @@ def test_extra_body_params_include_video_flow_shift_alias():
     assert "batch_cfg" in params
 
 
-def test_extract_prompt_string_and_mapping():
-    from vllm_omni.diffusion.models.lingbot_video.pipeline_lingbot_video import _extract_prompt
+def test_preprocess_preserves_geometry_and_converts_rgb(tmp_path):
+    from vllm_omni.diffusion.models.lingbot_video import (
+        get_lingbot_video_pre_process_func,
+    )
 
-    string_req = SimpleNamespace(prompt="a robot waves")
-    assert _extract_prompt(string_req) == ("a robot waves", None)
+    image_path = tmp_path / "reference.png"
+    Image.new("L", (48, 32), color=127).save(image_path)
+    request = OmniDiffusionRequest(
+        prompt={
+            "prompt": "a robot walks",
+            "modalities": ["video"],
+            "multi_modal_data": {"image": str(image_path)},
+        },
+        sampling_params=OmniDiffusionSamplingParams(),
+        request_id="lingbot-preprocess-test",
+    )
 
-    mapping_req = SimpleNamespace(prompt={"prompt": "a robot walks", "negative_prompt": "low quality"})
-    assert _extract_prompt(mapping_req) == ("a robot walks", "low quality")
+    result = get_lingbot_video_pre_process_func(SimpleNamespace())(request)
 
-    with pytest.raises(TypeError, match="Unsupported LingBot prompt type"):
-        _extract_prompt(SimpleNamespace(prompt=["not", "supported"]))
+    image = result.prompt["multi_modal_data"]["image"]
+    assert image.mode == "RGB"
+    assert image.size == (48, 32)
+
+
+def test_preprocess_marks_internal_image_warmup_as_video():
+    from vllm_omni.diffusion.models.lingbot_video import (
+        get_lingbot_video_pre_process_func,
+    )
+
+    request = OmniDiffusionRequest(
+        prompt={
+            "prompt": "dummy run",
+            "multi_modal_data": {"image": Image.new("RGB", (32, 32))},
+        },
+        sampling_params=OmniDiffusionSamplingParams(),
+        request_id=DUMMY_DIFFUSION_REQUEST_ID,
+    )
+
+    result = get_lingbot_video_pre_process_func(SimpleNamespace())(request)
+
+    assert result.prompt["modalities"] == ["video"]
 
 
 def test_check_inputs_accepts_t2v_shapes_and_rejects_invalid_values():
@@ -104,6 +145,59 @@ def test_postprocess_keeps_torch_by_default_and_converts_np_when_requested():
     array = post(frames, SimpleNamespace(output_type="np"))
     assert isinstance(array, np.ndarray)
     assert array.shape == (2, 4, 4, 3)
+    video = post({"video": frames}, SimpleNamespace(output_type="pt"))
+    assert set(video) == {"video"}
+    assert video["video"] is frames
+    image = post(
+        {"image": torch.ones(4, 4, 3)},
+        SimpleNamespace(output_type="pt"),
+    )
+    assert set(image) == {"image"}
+    assert isinstance(image["image"], np.ndarray)
+    assert image["image"].shape == (4, 4, 3)
+    latent = torch.ones(1, 4, 1, 2, 2)
+    latent_output = post({"image": latent}, SimpleNamespace(output_type="latent"))
+    assert set(latent_output) == {"image"}
+    assert latent_output["image"] is latent
+
+
+def test_postprocess_preserves_image_contract_for_serving():
+    from vllm_omni.diffusion.data import DiffusionOutput
+    from vllm_omni.diffusion.models.lingbot_video import get_lingbot_video_post_process_func
+    from vllm_omni.diffusion.output_formatter import (
+        format_diffusion_outputs,
+        normalize_diffusion_postprocess_output,
+    )
+    from vllm_omni.entrypoints.openai.api_server import _extract_images_from_result
+
+    sampling = OmniDiffusionSamplingParams(
+        num_inference_steps=1,
+        output_type="pt",
+        resolution=64,
+    )
+    request = OmniDiffusionRequest(
+        prompt={"prompt": "a robot", "modalities": ["image"]},
+        sampling_params=sampling,
+        request_id="lingbot-image-contract-test",
+    )
+    raw_output = {"image": torch.ones(8, 12, 3)}
+    processed = get_lingbot_video_post_process_func(SimpleNamespace())(raw_output, sampling)
+    normalized = normalize_diffusion_postprocess_output(processed)
+
+    assert normalized.primary_key == "image"
+    [result] = format_diffusion_outputs(
+        request=request,
+        od_config=SimpleNamespace(model_class_name="LingBotVideoPipeline"),
+        diffusion_output=DiffusionOutput(output=raw_output),
+        output_data=raw_output,
+        postprocess_output=normalized,
+    )
+
+    assert len(result.images) == 1
+    assert isinstance(result.images[0], np.ndarray)
+    endpoint_images = _extract_images_from_result(result)
+    assert len(endpoint_images) == 1
+    assert endpoint_images[0].size == (12, 8)
 
 
 def test_forward_resolves_t2v_sampling_and_flow_shift_alias():
@@ -116,7 +210,11 @@ def test_forward_resolves_t2v_sampling_and_flow_shift_alias():
 
     pipeline._generate = fake_generate
     req = _make_request_batch(
-        {"prompt": "a robot arm", "negative_prompt": "bad hands"},
+        {
+            "prompt": "a robot arm",
+            "negative_prompt": "bad hands",
+            "modalities": ["video"],
+        },
         height=192,
         width=320,
         num_frames=9,
@@ -129,10 +227,12 @@ def test_forward_resolves_t2v_sampling_and_flow_shift_alias():
 
     out = pipeline.forward(req)
 
-    assert torch.equal(out.output, torch.zeros(9, 192, 320, 3))
+    assert torch.equal(out.output["video"], torch.zeros(9, 192, 320, 3))
     assert len(calls) == 1
     call = calls[0]
     assert call["prompt"] == "a robot arm"
+    assert call["mode"].value == "t2v"
+    assert call["input_image"] is None
     assert call["negative_prompt"] == "bad hands"
     assert call["height"] == 192
     assert call["width"] == 320
@@ -194,6 +294,222 @@ def test_forward_prefers_shift_over_flow_shift_and_defaults_negative_prompt():
     assert calls[0]["negative_prompt"] == "default negative"
 
 
+@pytest.mark.parametrize(
+    ("prompt", "num_frames", "expected_mode", "output_key", "default_negative"),
+    [
+        (
+            {"prompt": "a still robot", "modalities": ["image"]},
+            1,
+            "t2i",
+            "image",
+            "default image negative",
+        ),
+        (
+            {
+                "prompt": "a moving robot",
+                "modalities": ["video"],
+                "multi_modal_data": {"image": Image.new("RGB", (48, 32), color="blue")},
+            },
+            9,
+            "ti2v",
+            "video",
+            "default negative",
+        ),
+    ],
+)
+def test_forward_dispatches_generation_mode_and_output(
+    prompt,
+    num_frames,
+    expected_mode,
+    output_key,
+    default_negative,
+):
+    pipeline = _make_pipeline()
+    calls = []
+
+    def fake_generate(**kwargs):
+        calls.append(kwargs)
+        return torch.zeros(192, 320, 3) if output_key == "image" else torch.zeros(num_frames, 192, 320, 3)
+
+    pipeline._generate = fake_generate
+    output = pipeline.forward(
+        _make_request_batch(
+            prompt,
+            height=192,
+            width=320,
+            num_frames=num_frames,
+            num_inference_steps=2,
+            guidance_scale=3.0,
+            seed=42,
+            output_type="pt",
+        )
+    )
+
+    assert set(output.output) == {output_key}
+    assert calls[0]["mode"].value == expected_mode
+    assert calls[0]["negative_prompt"] == default_negative
+    if expected_mode == "ti2v":
+        assert isinstance(calls[0]["input_image"], Image.Image)
+
+
+def test_batched_cfg_pads_image_conditioned_embeddings_and_masks():
+    from vllm_omni.diffusion.models.lingbot_video.pipeline_lingbot_video import (
+        _batch_cfg_prompt_inputs,
+    )
+
+    positive = torch.ones(1, 7, 4)
+    positive_mask = torch.ones(1, 7, dtype=torch.long)
+    negative = torch.full((1, 9, 4), 2.0)
+    negative_mask = torch.ones(1, 9, dtype=torch.long)
+
+    embeds, mask = _batch_cfg_prompt_inputs(
+        positive,
+        positive_mask,
+        negative,
+        negative_mask,
+    )
+
+    assert embeds.shape == (2, 9, 4)
+    assert mask.shape == (2, 9)
+    assert torch.equal(mask[0], torch.tensor([1, 1, 1, 1, 1, 1, 1, 0, 0]))
+    assert torch.equal(mask[1], torch.ones(9, dtype=torch.long))
+
+
+class _RecordingTransformer(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.anchor = nn.Parameter(torch.zeros(()))
+        self.config = SimpleNamespace(in_channels=1)
+        self.latent_inputs = []
+
+    def forward(
+        self,
+        latents,
+        timestep,
+        prompt_embeds,
+        *,
+        encoder_attention_mask,
+        return_dict,
+    ):
+        del timestep, prompt_embeds, encoder_attention_mask, return_dict
+        self.latent_inputs.append(latents.detach().clone())
+        return (torch.zeros_like(latents),)
+
+
+class _CorruptingScheduler:
+    sigma_max = 1.0
+    sigma_min = 0.0
+
+    def set_timesteps(self, num_inference_steps, *, device, shift, **kwargs):
+        del num_inference_steps, shift, kwargs
+        self.timesteps = torch.tensor([1000.0, 500.0], device=device)
+
+    def step(
+        self,
+        noise_pred,
+        timestep,
+        latents,
+        *,
+        return_dict,
+        generator,
+    ):
+        del noise_pred, timestep, return_dict, generator
+        return (latents + 7.0,)
+
+
+@pytest.mark.parametrize(
+    ("guidance_scale", "batch_cfg", "expected_prompts", "expected_transformer_calls"),
+    [
+        pytest.param(2.0, False, ["positive", "negative"], 4, id="sequential-cfg"),
+        pytest.param(2.0, True, ["positive", "negative"], 2, id="batched-cfg"),
+        pytest.param(1.0, False, ["positive"], 2, id="cfg-disabled"),
+    ],
+)
+def test_ti2v_reinjects_clean_prefix_across_cfg_modes(
+    guidance_scale,
+    batch_cfg,
+    expected_prompts,
+    expected_transformer_calls,
+):
+    from vllm_omni.diffusion.models.lingbot_video import (
+        LingBotGenerationMode,
+        LingBotImageCondition,
+    )
+
+    pipeline = _make_pipeline()
+    pipeline.transformer = _RecordingTransformer()
+    pipeline.scheduler = _CorruptingScheduler()
+    pipeline.progress_bar = lambda timesteps: timesteps
+    pipeline.prepare_latents = lambda *args, **kwargs: torch.zeros(1, 1, 3, 1, 1)
+    vlm_image = Image.new("RGB", (32, 32), color="red")
+    condition = LingBotImageCondition(
+        vlm_image=vlm_image,
+        clean_latent=torch.full((1, 1, 1, 1, 1), 5.0),
+    )
+    pipeline.prepare_ti2v_image_condition = lambda *args, **kwargs: condition
+    encode_calls = []
+
+    def fake_encode(prompt, *, images, device):
+        del device
+        encode_calls.append((prompt, images))
+        return torch.ones(1, 2, 4), torch.ones(1, 2, dtype=torch.long)
+
+    pipeline.encode_prompt = fake_encode
+
+    latents = pipeline._generate(
+        prompt="positive",
+        mode=LingBotGenerationMode.TI2V,
+        input_image=vlm_image,
+        negative_prompt="negative",
+        height=16,
+        width=16,
+        num_frames=9,
+        num_inference_steps=2,
+        guidance_scale=guidance_scale,
+        shift=3.0,
+        output_type="latent",
+        batch_cfg=batch_cfg,
+    )
+
+    assert [call[0] for call in encode_calls] == expected_prompts
+    assert all(call[1] == [vlm_image] for call in encode_calls)
+    assert len(pipeline.transformer.latent_inputs) == expected_transformer_calls
+    for item in pipeline.transformer.latent_inputs:
+        expected_prefix = condition.clean_latent.expand(item.shape[0], -1, -1, -1, -1)
+        assert torch.equal(item[:, :, :1], expected_prefix)
+    assert torch.equal(latents[:, :, :1], condition.clean_latent)
+    assert torch.equal(latents[:, :, 1:], torch.full((1, 1, 2, 1, 1), 14.0))
+
+
+def test_t2i_decodes_and_returns_the_unique_frame():
+    from vllm_omni.diffusion.models.lingbot_video import LingBotGenerationMode
+
+    pipeline = _make_pipeline()
+    pipeline.transformer = _RecordingTransformer()
+    pipeline.scheduler = _CorruptingScheduler()
+    pipeline.progress_bar = lambda timesteps: timesteps
+    pipeline.prepare_latents = lambda *args, **kwargs: torch.zeros(1, 1, 1, 2, 2)
+    pipeline.encode_prompt = lambda *args, **kwargs: (
+        torch.ones(1, 2, 4),
+        torch.ones(1, 2, dtype=torch.long),
+    )
+    pipeline._decode_latents = lambda latents: torch.ones(1, 16, 16, 3)
+
+    image = pipeline._generate(
+        prompt="a still robot",
+        mode=LingBotGenerationMode.T2I,
+        height=16,
+        width=16,
+        num_frames=1,
+        num_inference_steps=2,
+        guidance_scale=1.0,
+        shift=3.0,
+        output_type="pt",
+    )
+
+    assert image.shape == (16, 16, 3)
+
+
 def test_forward_rejects_multi_request_batches():
     pipeline = _make_pipeline()
     first = OmniDiffusionRequest(
@@ -211,6 +527,45 @@ def test_forward_rejects_multi_request_batches():
         pipeline.forward(DiffusionRequestBatch([first, second]))
 
 
+def test_forward_marks_invalid_request_contract_as_client_error():
+    from vllm_omni.errors import OmniClientError
+
+    pipeline = _make_pipeline()
+    request = _make_request_batch(
+        {
+            "prompt": "unsupported image edit",
+            "modalities": ["image"],
+            "multi_modal_data": {"image": Image.new("RGB", (32, 32), color="blue")},
+        },
+        height=192,
+        width=320,
+        num_frames=1,
+    )
+
+    with pytest.raises(OmniClientError, match="does not accept") as exc_info:
+        pipeline.forward(request)
+
+    assert exc_info.value.status_code == 400
+
+
+def test_forward_marks_unsupported_output_count_as_client_error():
+    from vllm_omni.errors import OmniClientError
+
+    pipeline = _make_pipeline()
+    request = _make_request_batch(
+        {"prompt": "motion", "modalities": ["video"]},
+        height=192,
+        width=320,
+        num_frames=9,
+        num_outputs_per_prompt=2,
+    )
+
+    with pytest.raises(OmniClientError, match="one output per prompt") as exc_info:
+        pipeline.forward(request)
+
+    assert exc_info.value.status_code == 400
+
+
 def test_load_weights_rejects_external_weight_stream():
     pipeline = _make_pipeline()
 

tests/diffusion/models/lingbot_video/test_request_utils.py

@@ -0,0 +1,350 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from types import SimpleNamespace
+
+import pytest
+from PIL import Image
+
+from vllm_omni.diffusion.models.lingbot_video.request_utils import (
+    LingBotGenerationMode,
+    caption_from_lingbot_prompt,
+    normalize_lingbot_num_frames,
+    normalize_lingbot_request,
+    resolve_lingbot_mode,
+    resolve_lingbot_num_frames,
+    resolve_lingbot_output_dimensions,
+    resolve_lingbot_size,
+)
+from vllm_omni.inputs.data import OmniDiffusionSamplingParams
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.diffusion]
+
+
+def _request(prompt, **sampling_overrides):
+    return SimpleNamespace(
+        prompt=prompt,
+        sampling_params=OmniDiffusionSamplingParams(**sampling_overrides),
+    )
+
+
+def _normalize(prompt, **sampling_overrides):
+    return normalize_lingbot_request(
+        _request(prompt, **sampling_overrides),
+        default_negative_prompt="video default",
+        default_image_negative_prompt="image default",
+    )
+
+
+@pytest.mark.parametrize(
+    ("prompt", "expected"),
+    [
+        ({"prompt": "still", "modalities": ["image"]}, LingBotGenerationMode.T2I),
+        ({"prompt": "motion", "modalities": ["video"]}, LingBotGenerationMode.T2V),
+        (
+            {
+                "prompt": "motion",
+                "modalities": ["video"],
+                "multi_modal_data": {"image": Image.new("RGB", (8, 8))},
+            },
+            LingBotGenerationMode.TI2V,
+        ),
+    ],
+)
+def test_resolve_lingbot_mode(prompt, expected):
+    assert resolve_lingbot_mode(prompt) is expected
+
+
+@pytest.mark.parametrize(
+    ("prompt", "message"),
+    [
+        (
+            {"prompt": "x", "modalities": ["image"], "multi_modal_data": {"image": Image.new("RGB", (8, 8))}},
+            "does not accept",
+        ),
+        (
+            {
+                "prompt": "x",
+                "modalities": ["video"],
+                "multi_modal_data": {"image": [Image.new("RGB", (8, 8)), Image.new("RGB", (8, 8))]},
+            },
+            "exactly one",
+        ),
+        ({"prompt": "x", "modalities": ["video"], "multi_modal_data": {"video": [1]}}, "reference modalities"),
+        ({"prompt": "x", "multi_modal_data": {"image": Image.new("RGB", (8, 8))}}, "explicitly set"),
+        ({"prompt": "x", "modalities": ["audio"]}, "output modality"),
+    ],
+)
+def test_resolve_lingbot_mode_rejects_invalid_media_combinations(prompt, message):
+    with pytest.raises(ValueError, match=message):
+        resolve_lingbot_mode(prompt)
+
+
+def test_caption_from_lingbot_prompt_preserves_text_and_compact_json():
+    assert caption_from_lingbot_prompt("plain text") == "plain text"
+    assert caption_from_lingbot_prompt({"caption": "caption only", "duration": 4}) == "caption only"
+    assert caption_from_lingbot_prompt({"caption": {"中文": True, "nested": {"value": 2}}}) == (
+        '{"中文":true,"nested":{"value":2}}'
+    )
+    assert (
+        caption_from_lingbot_prompt(
+            {"scene": "海边", "camera": {"move": "pan"}, "duration": 5, "fps": 24, "size": "320x192"}
+        )
+        == '{"scene":"海边","camera":{"move":"pan"}}'
+    )
+
+
+@pytest.mark.parametrize("prompt", [{"duration": 4, "fps": 24}, {"caption": ""}, ""])
+def test_caption_from_lingbot_prompt_rejects_empty_caption(prompt):
+    with pytest.raises(ValueError, match="empty|must not be empty"):
+        caption_from_lingbot_prompt(prompt)
+
+
+@pytest.mark.parametrize(
+    ("num_frames", "expected"),
+    [(1, 1), (2, 5), (4, 5), (5, 5), (96, 97), (98, 101), (121, 121)],
+)
+def test_normalize_lingbot_num_frames_rounds_up_to_4n_plus_1(num_frames, expected):
+    assert normalize_lingbot_num_frames(num_frames) == expected
+
+
+@pytest.mark.parametrize("num_frames", [0, -1, 1.5, "invalid"])
+def test_normalize_lingbot_num_frames_rejects_invalid_values(num_frames):
+    with pytest.raises(ValueError, match="num_frames"):
+        normalize_lingbot_num_frames(num_frames)
+
+
+@pytest.mark.parametrize(
+    ("duration", "fps", "expected"),
+    [(4, 24, 97), (5, 24, 121), (8, 24, 193), (4, 12, 49), (4, 30, 121)],
+)
+def test_resolve_lingbot_num_frames_rounds_up(duration, fps, expected):
+    assert resolve_lingbot_num_frames(duration, fps) == expected
+
+
+@pytest.mark.parametrize(("duration", "fps"), [(0, 24), (-1, 24), (0.01, 24), (1, 0), (1, -1)])
+def test_resolve_lingbot_num_frames_rejects_non_positive_or_empty_results(duration, fps):
+    with pytest.raises(ValueError, match="duration|fps|at least one"):
+        resolve_lingbot_num_frames(duration, fps)
+
+
+@pytest.mark.parametrize(
+    ("resolution", "ratio", "expected"),
+    [
+        ("192p", "1:1", (192, 192)),
+        ("192p", "9:16", (192, 320)),
+        ("192p", "16:9", (320, 192)),
+        ("192p", "3:4", (192, 256)),
+        ("192p", "4:3", (256, 192)),
+        ("480p", "1:1", (480, 480)),
+        ("480p", "9:16", (480, 832)),
+        ("480p", "16:9", (832, 480)),
+        ("480p", "3:4", (480, 640)),
+        ("480p", "4:3", (640, 480)),
+        ("720p", "1:1", (736, 736)),
+        ("720p", "9:16", (736, 1280)),
+        ("720p", "16:9", (1280, 736)),
+        ("720p", "3:4", (736, 960)),
+        ("720p", "4:3", (960, 736)),
+        ("1080p", "1:1", (1088, 1088)),
+        ("1080p", "9:16", (1088, 1920)),
+        ("1080p", "16:9", (1920, 1088)),
+        ("1080p", "3:4", (1088, 1440)),
+        ("1080p", "4:3", (1440, 1088)),
+        ("2k", "1:1", (1440, 1440)),
+        ("2k", "9:16", (1440, 2560)),
+        ("2k", "16:9", (2560, 1440)),
+        ("2k", "3:4", (1440, 1920)),
+        ("2k", "4:3", (1920, 1440)),
+        ("4k", "1:1", (2176, 2176)),
+        ("4k", "9:16", (2176, 3840)),
+        ("4k", "16:9", (3840, 2176)),
+        ("4k", "3:4", (2176, 2880)),
+        ("4k", "4:3", (2880, 2176)),
+    ],
+)
+def test_resolve_lingbot_size_supports_all_official_presets(resolution, ratio, expected):
+    assert resolve_lingbot_size(resolution=resolution, ratio=ratio) == expected
+
+
+def test_resolve_lingbot_size_validates_sources_and_alignment():
+    assert resolve_lingbot_size(size="320x192") == (192, 320)
+    assert resolve_lingbot_size(width=320, height=192) == (192, 320)
+    with pytest.raises(ValueError, match="provided together"):
+        resolve_lingbot_size(width=320)
+    with pytest.raises(ValueError, match="provided together"):
+        resolve_lingbot_size(resolution="192p")
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        resolve_lingbot_size(width=320, height=192, resolution="192p", ratio="9:16")
+    with pytest.raises(ValueError, match="multiples of 16"):
+        resolve_lingbot_size(width=321, height=192)
+
+
+def test_resolve_lingbot_output_dimensions_matches_request_priority():
+    assert resolve_lingbot_output_dimensions(
+        sampling_width=480,
+        sampling_height=480,
+        prompt_fields={"resolution": "720p", "ratio": "16:9"},
+    ) == (736, 1280)
+
+    assert resolve_lingbot_output_dimensions(
+        sampling_width=480,
+        sampling_height=480,
+        prompt_fields={"width": 320, "height": 192},
+    ) == (320, 192)
+
+    assert resolve_lingbot_output_dimensions(
+        sampling_width=320,
+        sampling_height=192,
+    ) == (320, 192)
+
+    assert resolve_lingbot_output_dimensions() == (480, 480)
+
+
+def test_normalize_lingbot_request_uses_mode_defaults_and_preserves_empty_negative():
+    t2i = _normalize(
+        {"prompt": "still", "modalities": ["image"], "negative_prompt": ""},
+        width=320,
+        height=192,
+        num_frames=1,
+    )
+    assert t2i.mode is LingBotGenerationMode.T2I
+    assert t2i.num_frames == 1
+    assert t2i.negative_prompt == ""
+
+    t2v = _normalize(
+        {"prompt": "motion", "modalities": ["video"]},
+        width=320,
+        height=192,
+        num_frames=9,
+    )
+    assert t2v.mode is LingBotGenerationMode.T2V
+    assert t2v.negative_prompt == "video default"
+
+
+def test_normalize_lingbot_request_rejects_explicit_zero_flow_shift():
+    with pytest.raises(ValueError, match="shift.*positive"):
+        _normalize(
+            {"prompt": "still", "modalities": ["image"]},
+            width=320,
+            height=192,
+            num_frames=1,
+            extra_args={"flow_shift": 0},
+        )
+
+
+def test_normalize_lingbot_request_rejects_explicit_zero_width():
+    with pytest.raises(ValueError, match="width"):
+        _normalize(
+            {"prompt": "still", "modalities": ["image"]},
+            width=0,
+            height=192,
+            num_frames=1,
+        )
+
+
+def test_normalize_lingbot_request_rounds_all_video_frame_counts():
+    config = _normalize(
+        {"prompt": "motion", "modalities": ["video"]},
+        width=320,
+        height=192,
+        num_frames=96,
+        fps=24,
+        frame_rate=24,
+    )
+    assert config.num_frames == 97
+
+    config = _normalize(
+        {"prompt": "motion", "modalities": ["video"]},
+        width=320,
+        height=192,
+        num_frames=121,
+        fps=24,
+        frame_rate=24,
+    )
+    assert config.num_frames == 121
+
+
+def test_normalize_lingbot_request_prefers_explicit_frames_over_duration():
+    online = _normalize(
+        {"prompt": "motion", "modalities": ["video"]},
+        width=320,
+        height=192,
+        num_frames=9,
+        fps=24,
+        extra_args={"duration": 5},
+    )
+    assert online.num_frames == 9
+
+    offline = _normalize(
+        {
+            "prompt": {"caption": "motion", "duration": 5, "num_frames": 9},
+            "modalities": ["video"],
+        },
+        width=320,
+        height=192,
+        fps=24,
+    )
+    assert offline.num_frames == 9
+
+
+def test_normalize_lingbot_request_uses_duration_then_model_default():
+    duration = _normalize(
+        {"prompt": {"caption": "motion", "duration": 5}, "modalities": ["video"]},
+        width=320,
+        height=192,
+        fps=24,
+    )
+    assert duration.num_frames == 121
+
+    default = _normalize(
+        {"prompt": "motion", "modalities": ["video"]},
+        width=320,
+        height=192,
+    )
+    assert default.num_frames == 81
+
+
+def test_normalize_lingbot_request_rejects_contract_conflicts():
+    with pytest.raises(ValueError, match="text-to-image.*duration"):
+        _normalize(
+            {"prompt": {"caption": "still", "duration": 4}, "modalities": ["image"]},
+            width=320,
+            height=192,
+            num_frames=1,
+        )
+    with pytest.raises(ValueError, match="text-to-image requires"):
+        _normalize(
+            {"prompt": "still", "modalities": ["image"]},
+            width=320,
+            height=192,
+            num_frames=5,
+        )
+    with pytest.raises(ValueError, match="seconds.*duration"):
+        _normalize(
+            {"prompt": "motion", "modalities": ["video"]},
+            width=320,
+            height=192,
+            num_frames=96,
+            extra_args={"seconds": 4, "duration": 4},
+        )
+
+
+def test_normalize_lingbot_request_model_size_overrides_sampling_dimensions():
+    # A prompt-provided size overrides the sampling width/height.
+    config = _normalize(
+        {"prompt": {"caption": "motion", "size": "320x192"}, "modalities": ["video"]},
+        width=480,
+        height=480,
+        num_frames=81,
+    )
+    assert (config.height, config.width) == (192, 320)
+
+    # Without a prompt size, the sampling dimensions are used.
+    config = _normalize(
+        {"prompt": "motion", "modalities": ["video"]},
+        width=480,
+        height=480,
+        num_frames=81,
+    )
+    assert (config.height, config.width) == (480, 480)