[Refactor] P0.2: Migrate API server helpers out of api_server (#5453)

Signed-off-by: herotai214 <herotai214@gmail.com>

tests/diffusion/models/lingbot_video/test_pipeline_lingbot_video.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
 from types import SimpleNamespace
 
@@ -168,7 +168,7 @@ def test_postprocess_preserves_image_contract_for_serving():
         format_diffusion_outputs,
         normalize_diffusion_postprocess_output,
     )
-    from vllm_omni.entrypoints.openai.api_server import _extract_images_from_result
+    from vllm_omni.entrypoints.openai.images.helpers import _extract_images_from_result
 
     sampling = OmniDiffusionSamplingParams(
         num_inference_steps=1,

tests/entrypoints/openai_api/test_api_server_guards.py

@@ -55,6 +55,7 @@ from starlette.websockets import WebSocketDisconnect
 from vllm.v1.engine.exceptions import EngineDeadError, EngineGenerateError
 
 from vllm_omni.entrypoints.openai import api_server
+from vllm_omni.entrypoints.serve.utils import errors as serve_errors
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
@@ -656,14 +657,16 @@ def test_images_generation_without_engine_preserves_service_unavailable_error()
     assert exc_info.value.detail == "Multi-stage engine not initialized. Start server with a multi-stage omni model."
 
 
-def test_images_generation_without_multistage_chat_handler_preserves_unavailable_error(monkeypatch) -> None:
+def test_images_generation_without_multistage_chat_handler_preserves_unavailable_error() -> None:
     """Lock images generation when chat serving was not wired.
 
     Fails if multi-stage images stop requiring ``openai_serving_chat``, or the
     503 detail changes after images/chat ownership splits.
     """
     app = FastAPI()
-    app.state.engine_client = SimpleNamespace(stage_configs=[object(), object()])
+    # Prefer real stage_type input over monkeypatching ``get_stage_type``:
+    # ``_get_engine_and_model`` lives in ``app_state`` and binds utils directly.
+    app.state.engine_client = SimpleNamespace(stage_configs=[{"stage_type": "llm"}, {"stage_type": "diffusion"}])
     app.state.stage_configs = app.state.engine_client.stage_configs
     app.state.openai_serving_models = SimpleNamespace(base_model_paths=[SimpleNamespace(name="demo-model")])
     app.state.openai_serving_chat = None
@@ -671,7 +674,6 @@ def test_images_generation_without_multistage_chat_handler_preserves_unavailable
 
     raw_request = _request_for(app, method="POST", path="/v1/images/generations")
     request = api_server.ImageGenerationRequest(prompt="a cat", model="demo-model")
-    monkeypatch.setattr(api_server, "get_stage_type", lambda _stage_cfg: "diffusion")
 
     with pytest.raises(HTTPException) as exc_info:
         asyncio.run(api_server.generate_images(request, raw_request))
@@ -709,9 +711,7 @@ def test_engine_error_json_response_includes_request_and_stage_fields(monkeypatc
     in P0.2 is not a false red.
     """
     app = FastAPI()
-    # TODO(P0.2): retarget this call if ``_register_omni_exception_handlers``
-    # leaves api_server (red here is a rename, not a surface regression).
-    api_server._register_omni_exception_handlers(app)
+    serve_errors._register_omni_exception_handlers(app)
     app.state.engine_client = SimpleNamespace(errored=True, engine=SimpleNamespace(is_alive=lambda: False))
     app.state.server = object()
     app.state.args = SimpleNamespace(log_error_stack=False)
@@ -719,7 +719,7 @@ def test_engine_error_json_response_includes_request_and_stage_fields(monkeypatc
     req = _request_for(app, method="POST", path="/v1/chat/completions")
     req.state.request_metadata = SimpleNamespace(request_id="req-123")
 
-    monkeypatch.setattr(api_server, "terminate_if_errored", lambda **_kwargs: None)
+    monkeypatch.setattr(serve_errors, "terminate_if_errored", lambda **_kwargs: None)
 
     exc = EngineGenerateError("boom")
     exc.error_stage_id = 2  # type: ignore[attr-defined]
@@ -740,14 +740,12 @@ def test_engine_dead_error_handler_registered_returns_json(monkeypatch) -> None:
     JSON with ``request_id``.
     """
     app = FastAPI()
-    # TODO(P0.2): retarget this call if ``_register_omni_exception_handlers``
-    # leaves api_server (red here is a rename, not a surface regression).
-    api_server._register_omni_exception_handlers(app)
+    serve_errors._register_omni_exception_handlers(app)
     app.state.engine_client = SimpleNamespace(errored=True, engine=SimpleNamespace(is_alive=lambda: False))
     app.state.server = object()
     app.state.args = SimpleNamespace(log_error_stack=False)
 
-    monkeypatch.setattr(api_server, "terminate_if_errored", lambda **_kwargs: None)
+    monkeypatch.setattr(serve_errors, "terminate_if_errored", lambda **_kwargs: None)
 
     handler = app.exception_handlers[EngineDeadError]
     req = _request_for(app)
@@ -766,7 +764,7 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None:
     models), wires a handler that must stay None, or leaves a required
     handler as None.
     """
-    stage = SimpleNamespace(engine_args={})
+    stage = SimpleNamespace(stage_type="diffusion", engine_args={})
     engine = _FakeEngineClient(stage_configs=[stage])
 
     def _for_diffusion_factory(label: str):
@@ -776,7 +774,6 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None:
 
         return _factory
 
-    monkeypatch.setattr(api_server, "get_stage_type", lambda _cfg: "diffusion")
     monkeypatch.setattr(api_server.OmniOpenAIServingChat, "for_diffusion", _for_diffusion_factory("chat"))
     monkeypatch.setattr(api_server.OmniOpenAIServingChatBatch, "for_diffusion", _for_diffusion_factory("chat_batch"))
     monkeypatch.setattr(
@@ -811,7 +808,7 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None:
 
 @pytest.mark.asyncio
 async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) -> None:
-    stage = SimpleNamespace(engine_args={})
+    stage = SimpleNamespace(stage_type="diffusion", engine_args={})
     engine = _FakeEngineClient(stage_configs=[stage])
     speech_kwargs = {}
 
@@ -827,7 +824,6 @@ async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) ->
         speech_kwargs.update(kwargs)
         return _marker("speech")
 
-    monkeypatch.setattr(api_server, "get_stage_type", lambda _cfg: "diffusion")
     monkeypatch.setattr(api_server.OmniOpenAIServingChat, "for_diffusion", _for_diffusion_factory("chat"))
     monkeypatch.setattr(api_server.OmniOpenAIServingChatBatch, "for_diffusion", _for_diffusion_factory("chat_batch"))
     monkeypatch.setattr(

tests/entrypoints/openai_api/test_image_server.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """
 Tests for async image generation API endpoints.
 
@@ -24,11 +24,13 @@ from vllm.entrypoints.openai.models.protocol import BaseModelPath
 from vllm.sampling_params import RequestOutputKind
 
 from vllm_omni.entrypoints.async_omni import AsyncOmni
-from vllm_omni.entrypoints.openai.api_server import _check_max_generated_image_size, _DiffusionServingModels, router
+from vllm_omni.entrypoints.openai.api_server import router
 from vllm_omni.entrypoints.openai.image_api_utils import (
     encode_image_base64,
     parse_size,
 )
+from vllm_omni.entrypoints.openai.images.helpers import _check_max_generated_image_size
+from vllm_omni.entrypoints.openai.models.serving import _DiffusionServingModels
 from vllm_omni.entrypoints.openai.serving_chat import OmniOpenAIServingChat
 from vllm_omni.errors import GuardrailViolationError
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
@@ -219,7 +221,7 @@ def test_client(mock_async_diffusion):
     app.state.stage_configs = [SimpleNamespace(stage_type="diffusion")]
     from vllm.entrypoints.openai.models.protocol import BaseModelPath
 
-    from vllm_omni.entrypoints.openai.api_server import _DiffusionServingModels
+    from vllm_omni.entrypoints.openai.models.serving import _DiffusionServingModels
 
     app.state.openai_serving_models = _DiffusionServingModels(
         [BaseModelPath(name="Qwen/Qwen-Image", model_path="Qwen/Qwen-Image")]
@@ -2126,7 +2128,7 @@ def test_normalize_image():
     """Test _normalize_image with various input types"""
     import numpy as np
 
-    from vllm_omni.entrypoints.openai.api_server import _normalize_image
+    from vllm_omni.entrypoints.openai.images.helpers import _normalize_image
 
     # Test PIL Image input
     img = Image.new("RGB", (64, 64), color="red")
@@ -2163,7 +2165,7 @@ def test_extract_images_from_result():
     """Test _extract_images_from_result with various result formats"""
     import numpy as np
 
-    from vllm_omni.entrypoints.openai.api_server import _extract_images_from_result
+    from vllm_omni.entrypoints.openai.images.helpers import _extract_images_from_result
 
     # Test empty result
     class EmptyResult:

tests/entrypoints/openai_api/test_omni_sleep_wakeup.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
 import dataclasses
 
@@ -264,7 +264,7 @@ def pure_diffusion_app(pure_diffusion_engine, mocker):
     mocker.patch(
         "vllm_omni.entrypoints.openai.api_server.OmniOpenAIServingSpeech"
     ).for_diffusion.return_value = mocker.MagicMock()
-    mocker.patch("vllm_omni.entrypoints.openai.api_server._DiffusionServingModels")
+    mocker.patch("vllm_omni.entrypoints.openai.models.serving._DiffusionServingModels")
 
     loop = asyncio.new_event_loop()
     try:

tests/entrypoints/openai_api/test_serving_audio_generate.py

@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 import logging
 from inspect import Signature, signature
 from types import SimpleNamespace
@@ -18,6 +21,7 @@ from vllm_omni.entrypoints.openai.protocol.audio import (
 from vllm_omni.entrypoints.openai.serving_audio_generate import (
     OmniOpenAIServingAudioGenerate,
 )
+from vllm_omni.entrypoints.serve.utils import errors as serve_errors
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
 from vllm_omni.outputs import OmniRequestOutput
 
@@ -515,7 +519,7 @@ class TestErrorHandling:
         handler.create_audio_generate = AsyncMock(side_effect=exc)
         app = _make_api_server_test_app(handler)
 
-        with patch.object(api_server_module, "terminate_if_errored") as terminate_mock:
+        with patch.object(serve_errors, "terminate_if_errored") as terminate_mock:
             with TestClient(app) as client:
                 response = client.post("/v1/audio/generate", json={"input": "Hello"})
 

tests/entrypoints/openai_api/test_serving_speech.py

@@ -32,6 +32,7 @@ from vllm_omni.diffusion.sched.request_scheduler import RequestScheduler
 from vllm_omni.diffusion.sched.step_scheduler import StepScheduler
 from vllm_omni.entrypoints.omni_base import OmniEngineDeadError
 from vllm_omni.entrypoints.openai import api_server as api_server_module
+from vllm_omni.entrypoints.openai import errors as openai_errors
 from vllm_omni.entrypoints.openai import serving_speech as serving_speech_module
 from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin
 from vllm_omni.entrypoints.openai.protocol.audio import (
@@ -58,6 +59,7 @@ from vllm_omni.entrypoints.openai.tts_adapters.capabilities import load_supporte
 from vllm_omni.entrypoints.openai.tts_adapters.ming_tts import MingTTSAdapter
 from vllm_omni.entrypoints.openai.tts_adapters.qwen3_tts import Qwen3TTSAdapter, Qwen3TTSCodecLimitError
 from vllm_omni.entrypoints.openai.tts_adapters.voxtral import VoxtralTTSAdapter
+from vllm_omni.entrypoints.serve.utils import errors as serve_errors
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
 from vllm_omni.model_executor.models.fish_speech.prompt_utils import (
     FISH_TEXT_ONLY_SYSTEM_PROMPT,
@@ -3783,7 +3785,7 @@ def _patch_api_server_base(mocker: MockerFixture):
 
     fake_base = mocker.MagicMock()
     fake_base.create_error_response.side_effect = _fake_create_error_response
-    mocker.patch.object(api_server_module, "base", return_value=fake_base)
+    mocker.patch.object(openai_errors, "base", return_value=fake_base)
     return fake_base
 
 
@@ -4098,7 +4100,7 @@ def test_api_server_create_speech_engine_error_response_includes_request_and_sta
         )
     )
 
-    terminate_mock = mocker.patch.object(api_server_module, "terminate_if_errored")
+    terminate_mock = mocker.patch.object(serve_errors, "terminate_if_errored")
 
     raw_request = _make_api_server_request(handler, path="/v1/audio/speech")
     raw_request.app.state.args = SimpleNamespace(log_error_stack=False)
@@ -4130,8 +4132,8 @@ def test_omni_engine_error_handler_includes_request_and_stage_id(mocker: MockerF
     )
     app.state.server = SimpleNamespace()
 
-    terminate_mock = mocker.patch.object(api_server_module, "terminate_if_errored")
-    api_server_module._register_omni_exception_handlers(app)
+    terminate_mock = mocker.patch.object(serve_errors, "terminate_if_errored")
+    serve_errors._register_omni_exception_handlers(app)
 
     @app.get("/boom")
     async def boom(request: Request):

tests/entrypoints/openai_api/test_video_server.py

@@ -37,6 +37,12 @@ from vllm_omni.entrypoints.openai.protocol.videos import (
 from vllm_omni.entrypoints.openai.serving_video import OmniOpenAIServingVideo, ReferenceImage
 from vllm_omni.entrypoints.openai.storage import LocalStorageManager
 from vllm_omni.entrypoints.openai.stores import AsyncDictStore, TaskRegistry
+from vllm_omni.entrypoints.openai.video.generation import helpers as video_generation_helpers
+from vllm_omni.entrypoints.openai.video.generation.helpers import (
+    MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES,
+    _read_upload_limited,
+    _reference_video_decode_spec,
+)
 from vllm_omni.errors import GuardrailViolationError
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
 
@@ -192,6 +198,8 @@ def isolated_video_backends(tmp_path, monkeypatch):
     monkeypatch.setattr(api_server, "VIDEO_STORE", store)
     monkeypatch.setattr(api_server, "VIDEO_TASKS", tasks)
     monkeypatch.setattr(api_server, "STORAGE_MANAGER", storage)
+    monkeypatch.setattr(video_generation_helpers, "VIDEO_STORE", store)
+    monkeypatch.setattr(video_generation_helpers, "STORAGE_MANAGER", storage)
     return store, tasks, storage
 
 
@@ -245,7 +253,7 @@ async def test_server_worker_keeps_engine_alive_until_http_shutdown(monkeypatch)
     monkeypatch.setattr(api_server, "build_openai_app", lambda args, supported_tasks: FastAPI())
     monkeypatch.setattr(api_server, "serve_http", fake_serve_http)
     monkeypatch.setattr(api_server.STORAGE_MANAGER, "start", fake_storage_start)
-    monkeypatch.setattr(api_server, "_get_vllm_config", fake_get_vllm_config)
+    monkeypatch.setattr(api_server.openai_app_state, "_get_vllm_config", fake_get_vllm_config)
     monkeypatch.setattr(api_server, "omni_init_app_state", fake_init_app_state)
     monkeypatch.setattr(api_server, "get_uvicorn_log_config", lambda args: None)
 
@@ -978,7 +986,7 @@ def test_cosmos3_reference_video_limit_uses_v2v_condition_frames():
         extra_params={"condition_frame_indexes_vision": [0, 2]},
     )
 
-    spec = api_server._reference_video_decode_spec(request, _cosmos3_stage_configs())
+    spec = _reference_video_decode_spec(request, _cosmos3_stage_configs())
     assert spec.max_frames == 9
     assert spec.keep == "first"
 
@@ -990,7 +998,7 @@ def test_cosmos3_reference_video_limit_preserves_action_frames():
         extra_params={"action_mode": "inverse_dynamics", "action_chunk_size": 16},
     )
 
-    assert api_server._reference_video_decode_spec(request, _cosmos3_stage_configs()).max_frames == 17
+    assert _reference_video_decode_spec(request, _cosmos3_stage_configs()).max_frames == 17
 
 
 def test_cosmos3_reference_video_limit_caps_condition_frames_to_output_frames():
@@ -1000,7 +1008,7 @@ def test_cosmos3_reference_video_limit_caps_condition_frames_to_output_frames():
         extra_params={"condition_frame_indexes_vision": [0, 20]},
     )
 
-    assert api_server._reference_video_decode_spec(request, _cosmos3_stage_configs()).max_frames == 5
+    assert _reference_video_decode_spec(request, _cosmos3_stage_configs()).max_frames == 5
 
 
 def test_s2v_video_generation_with_audio_reference_form(test_client, mocker: MockerFixture):
@@ -1681,15 +1689,15 @@ def test_h3_multipart_maps_pillow_pixel_limit_error(field, test_client, monkeypa
 @pytest.mark.asyncio
 async def test_h3_upload_limit_checks_declared_size_before_read():
     class OversizedUpload:
-        size = api_server.MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES + 1
+        size = MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES + 1
 
         async def read(self, _size):
             raise AssertionError("the oversized upload must be rejected before reading")
 
     with pytest.raises(HTTPException, match="size limit"):
-        await api_server._read_upload_limited(
+        await _read_upload_limited(
             OversizedUpload(),
-            max_bytes=api_server.MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES,
+            max_bytes=MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES,
         )
 
 
@@ -2438,7 +2446,7 @@ def test_cosmos3_control_upload_rejects_existing_control_source(test_client):
 )
 def test_cosmos3_control_upload_rejects_invalid_size(control_bytes, message, test_client, monkeypatch):
     test_client.app.state.openai_serving_video._engine_client.model_class_name = "Cosmos3OmniDiffusersPipeline"
-    monkeypatch.setattr(api_server, "CONTROL_REFERENCE_MAX_BYTES", 3)
+    monkeypatch.setattr(video_generation_helpers, "CONTROL_REFERENCE_MAX_BYTES", 3)
 
     response = test_client.post(
         "/v1/videos/sync",

vllm_omni/config/endpoint_policy.py

@@ -2,16 +2,16 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Endpoint restriction policy for omni pipelines."""
 
-from collections.abc import Set
 from dataclasses import dataclass
 from enum import Enum
 from typing import NamedTuple
 
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
-from starlette.routing import Route
 from vllm.entrypoints.serve.exception_handling.error_response import create_error_response
 
+from vllm_omni.entrypoints.serve.utils.routes import remove_route_from_app
+
 
 class RouteTarget(NamedTuple):
     """A server path & supported methods."""
@@ -53,21 +53,6 @@ def build_rejection_handler(reason: str):
     return rejection_handler
 
 
-def remove_route_from_app(
-    app: FastAPI,
-    path: str,
-    methods: Set[str],
-) -> None:
-    """Remove routes matching a path and one of the given HTTP methods."""
-    routes_to_remove = [
-        route
-        for route in app.routes
-        if isinstance(route, Route) and route.path == path and route.methods is not None and route.methods & methods
-    ]
-    for route in routes_to_remove:
-        app.routes.remove(route)
-
-
 def shutdown_unsupported_routes(
     app: FastAPI,
     endpoint_restrictions: tuple[EndpointRestriction, ...],

vllm_omni/entrypoints/README.md

@@ -0,0 +1,21 @@
+# Entrypoints
+
+This package contains user-facing server, CLI, orchestration, and protocol
+entrypoints for vLLM-Omni.
+
+## What Belongs Here
+
+- Runtime orchestration entrypoints such as `AsyncOmni`, `Omni`, and
+  request-state helpers.
+- CLI integration under `cli/`.
+- Shared cross-entrypoint configuration and stage utilities.
+
+## What Does Not Belong Here
+
+- OpenAI endpoint-family route bodies and request helpers; those should move
+  toward `openai/<family>/` packages as part of #5227.
+- Generic model execution internals that belong under engine, worker, or model
+  executor packages.
+
+Some top-level utility files remain acceptable homes under Phase 0.2 (`P0.2`) of #5227.
+They are not part of the OpenAI helper migration unless explicitly referenced by #5227.

vllm_omni/entrypoints/cli/README.md

@@ -0,0 +1,21 @@
+# CLI Entrypoints
+
+This package owns vLLM-Omni command-line integration.
+
+## What Belongs Here
+
+- Omni CLI command registration and dispatch.
+- CLI argument parsing and validation for `vllm ... --omni`.
+- Headless/runtime launch wiring that ultimately calls server bootstrap code.
+- CLI-only presentation helpers such as the logo.
+
+## What Does Not Belong Here
+
+- FastAPI route bodies.
+- OpenAI endpoint request helpers.
+- Server app utilities that are shared outside CLI; those belong under
+  `entrypoints/serve`.
+
+The CLI currently imports `openai.api_server.omni_run_server`. If server
+bootstrap moves later, update this import, but do not fold route helpers into
+the CLI package. Related endpoint cleanup is tracked by #5227.

vllm_omni/entrypoints/cli/benchmark/README.md

@@ -0,0 +1,18 @@
+# Benchmark CLI
+
+This package owns Omni benchmark subcommands exposed through the CLI.
+
+## What Belongs Here
+
+- CLI parser extensions for benchmark commands.
+- Benchmark command dispatch.
+- Benchmark-only argument groups.
+
+## What Does Not Belong Here
+
+- Runtime server route helpers.
+- OpenAI endpoint implementation details.
+- Benchmark execution internals that belong under `vllm_omni.benchmarks`.
+
+This package is not part of the OpenAI endpoint helper migration. It may need
+import updates only if CLI/bootstrap ownership changes during #5227.

vllm_omni/entrypoints/duplex/warmup.py

@@ -0,0 +1,130 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Startup warmup probe for duplex ``/v1/realtime``.
+
+Moved out of ``api_server.py`` under P0.2 of #5227. The
+``omni_run_server_worker`` scheduling and ``/v1/realtime`` hold-clients
+preamble stay in ``api_server.py`` until P0.3.
+"""
+
+import asyncio
+import json
+import time
+import uuid
+
+import httpx
+import pybase64 as base64
+from vllm.logger import init_logger
+
+logger = init_logger(__name__)
+
+
+async def _warmup_duplex_realtime(app, args, warmup_frames: int) -> None:
+    """Run silent frames through a throwaway realtime session at startup.
+
+    Self-connects over the server's own websocket endpoint so the warmup
+    exercises the exact production path (serving adapter, duplex control
+    plane, every pipeline stage). One-time costs — kernel JIT compilation,
+    first prefill/decode paths, codec caches — are paid here instead of by
+    the first real client; ``/v1/realtime`` holds other connections until
+    the warmup finishes (see ``duplex_warmup_done``).
+    """
+    warmup_done = getattr(app.state, "duplex_warmup_done", None)
+    try:
+        try:
+            import websockets
+        except ImportError:
+            logger.warning("Duplex warmup skipped: the 'websockets' package is not installed.")
+            return
+        served = getattr(args, "served_model_name", None)
+        if isinstance(served, list | tuple) and served:
+            model_name = served[0]
+        elif isinstance(served, str) and served:
+            model_name = served
+        else:
+            model_name = args.model
+        adapter = getattr(getattr(app.state, "openai_serving_duplex", None), "_serving_runtime_adapter", None)
+        frame_samples = int(getattr(adapter, "silence_continuation_samples", 16000))
+        silence = base64.b64encode(bytes(frame_samples * 4)).decode("ascii")
+        from vllm_omni.experimental.fullduplex.client import build_realtime_url
+
+        url = (
+            build_realtime_url(f"ws://127.0.0.1:{args.port}/v1/realtime", model_name, autostart=False)
+            + "&vllm_omni_warmup=1"
+        )
+        logger.info("Duplex warmup starting: %d silent frames of %d samples.", warmup_frames, frame_samples)
+        start = time.time()
+        # The socket is bound before uvicorn finishes starting, so a plain
+        # connect can be accepted and then reset. Wait for /health first.
+        async with httpx.AsyncClient() as http:
+            for _ in range(120):
+                try:
+                    if (await http.get(f"http://127.0.0.1:{args.port}/health", timeout=2)).status_code == 200:
+                        break
+                except httpx.HTTPError:
+                    pass
+                await asyncio.sleep(1)
+            else:
+                logger.warning("Duplex warmup: /health never became ready; skipping warmup.")
+                return
+        async with websockets.connect(url, max_size=64 * 1024 * 1024, open_timeout=10) as ws:
+            await ws.send(
+                json.dumps(
+                    {
+                        "type": "session.update",
+                        "session": {
+                            "session_id": f"warmup-{uuid.uuid4().hex[:8]}",
+                            "model": model_name,
+                            "modalities": ["audio", "text"],
+                            "input_audio_format": "pcm_f32le",
+                            "output_audio_format": "pcm16",
+                            "idle_timeout_s": 60,
+                            "turn_detection": None,
+                            "extra_body": {"auto_response": True},
+                        },
+                    }
+                )
+            )
+            saw_audio = False
+            sent = 0
+
+            async def _recv_until(predicate, timeout_s: float) -> bool:
+                deadline = time.monotonic() + timeout_s
+                while time.monotonic() < deadline:
+                    try:
+                        raw = await asyncio.wait_for(ws.recv(), timeout=max(0.05, deadline - time.monotonic()))
+                    except (TimeoutError, asyncio.TimeoutError):
+                        return False
+                    event = json.loads(raw)
+                    if predicate(event):
+                        return True
+                return False
+
+            if not await _recv_until(lambda e: e.get("type") == "session.created", 30):
+                logger.warning("Duplex warmup: no session.created within 30 s; aborting warmup.")
+                return
+            while sent < warmup_frames:
+                await ws.send(json.dumps({"type": "input_audio_buffer.append", "audio": silence}))
+                sent += 1
+                await asyncio.sleep(0.08)
+            # Wait for the pipeline's first audio output so every stage ran.
+            saw_audio = await _recv_until(lambda e: e.get("type") == "response.audio.delta", 30)
+            # Close the session EXPLICITLY: a bare websocket close parks the
+            # session in its disconnect grace window, where it would keep
+            # occupying the max_sessions=1 slot and block the first client.
+            await ws.send(json.dumps({"type": "session.close"}))
+            await _recv_until(lambda e: e.get("type") == "session.closed", 10)
+        logger.info(
+            "Duplex warmup finished in %.1f s (%d silent frames, first audio %s).",
+            time.time() - start,
+            sent,
+            "received" if saw_audio else "NOT received",
+        )
+    except Exception:
+        logger.exception("Duplex warmup failed; continuing to serve (first session pays cold-start costs).")
+    finally:
+        if warmup_done is not None:
+            warmup_done.set()
+
+
+__all__ = ["_warmup_duplex_realtime"]

vllm_omni/entrypoints/openai/README.md

@@ -0,0 +1,58 @@
+# OpenAI Entrypoint Helpers
+
+This package contains OpenAI-compatible serving code for vLLM-Omni.
+
+## Hard Rule: `*_utils*` vs package `helpers.py`
+
+These are **different layers**. During the #5227 refactor they coexist on purpose.
+**Do not mix them. No ownership overlap.**
+
+| Layer | Files | Longevity | PUT HERE | DO NOT PUT HERE |
+| --- | --- | --- | --- | --- |
+| Shared media / serving utils | root `image_api_utils.py`, `video_api_utils.py`, `audio_utils_mixin.py`, `utils.py` | **Temporary.** TODO(#5227, Phase 1): tidy up / absorb into endpoint-family packages in Phase 1 modality PRs (P1.1-P1.3) | Codec / bytes / format / pure media conversion; cross-cutting stage/pipeline helpers reused by protocol + multiple serving paths | HTTP form parsing, FastAPI `Request` / app-state orchestration, job runners, endpoint response factories from `api_server.py` |
+| Endpoint-family helpers | `images/helpers.py`, `video/generation/helpers.py`, … | **Longer home** through helper (P0.2) + router (P0.3) stages; Phase 1 PRs may split further, not delete ownership | Request/job/runtime helpers peeled out of `api_server.py` for one endpoint family | Encode/decode/base64/media backends that already belong in root `*_utils*` |
+
+### Allowed dependency direction
+
+```text
+api_server / serving / protocol
+        │
+        ├──► package helpers.py     (endpoint orchestration; longer home)
+        │         │
+        │         └──► root *_utils*   (shared media / format; temporary)
+        │
+        └──► root *_utils*            (direct use OK)
+```
+
+- `helpers` **may import** `*_utils*`
+- `*_utils*` **must not import** package `helpers`
+- Never copy a symbol into both layers; pick one owner
+
+### Where to put new code (until family PRs finish)
+
+1. Pure image/video/audio bytes or format conversion with **no** FastAPI `Request` / job store / route semantics → root `*_utils*` only if it must stay shared; prefer not growing these files.
+2. Endpoint request parsing, validation against engine/app state, job lifecycle, or response shaping that used to sit beside routes in `api_server.py` → owning package `helpers.py` (**prefer this**).
+3. Cross-family OpenAI app-state / bootstrap → `app_state.py` / similar top-level OpenAI helpers — **not** into modality `*_utils*`.
+4. If unsure: **do not grow root `*_utils*`**. Put it in the owning package `helpers.py` and leave a TODO(#5227) for the modality PR.
+
+**TODO(#5227, Phase 1):** root `image_api_utils.py`, `video_api_utils.py`, `audio_utils_mixin.py`, and likely pieces of `utils.py` should be tidied, moved, or absorbed during Phase 1 audio / images / video family PRs (P1.1-P1.3). Package `helpers.py` files stay as the endpoint ownership surface until those PRs further split them.
+
+## What Belongs Here (top-level OpenAI helpers)
+
+- OpenAI protocol or response-shape helpers shared by multiple OpenAI endpoint
+  families.
+- App-state accessors for OpenAI serving objects, for example helpers that read
+  `request.app.state.openai_serving_*`.
+- Bootstrap helpers that are specific to OpenAI-compatible serving and are not
+  route bodies.
+
+## What Does Not Belong Here
+
+- FastAPI app/server mechanics that are not OpenAI-specific. Put those under
+  `vllm_omni.entrypoints.serve.utils`.
+- Modality-specific request parsing, validation, or response formatting. Put
+  those under the owning endpoint package, such as `images/` or
+  `video/generation/`.
+- Broad cleanup/rename of existing `image_api_utils.py`, `video_api_utils.py`,
+  `audio_utils_mixin.py`, and `utils.py` in this helper-only stage (P0.2) is out
+  of scope; tracked by #5227 Phase 1 modality PRs instead.

vllm_omni/entrypoints/openai/app_state.py

@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""OpenAI app-state accessors.
+
+This module owns small helpers that read initialized serving objects or engine
+configuration from ``request.app.state``."""
+
+from http import HTTPStatus
+from typing import Any
+
+from fastapi import HTTPException, Request
+from vllm.engine.protocol import EngineClient
+
+from vllm_omni.entrypoints.async_omni import AsyncOmni
+from vllm_omni.entrypoints.openai.batch_serving import OmniOpenAIServingChatBatch
+from vllm_omni.entrypoints.openai.serving_audio_generate import OmniOpenAIServingAudioGenerate
+from vllm_omni.entrypoints.openai.serving_chat import OmniOpenAIServingChat
+from vllm_omni.entrypoints.openai.serving_speech import OmniOpenAIServingSpeech
+from vllm_omni.entrypoints.openai.serving_video import OmniOpenAIServingVideo
+from vllm_omni.entrypoints.openai.utils import get_stage_type
+
+ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL = "endpoint-load-metrics-format"
+
+
+async def _get_vllm_config(engine_client: EngineClient) -> Any:
+    if hasattr(engine_client, "get_vllm_config"):
+        return await engine_client.get_vllm_config()
+    return getattr(engine_client, "vllm_config", None)
+
+
+def Omnivideo(request: Request) -> OmniOpenAIServingVideo | None:
+    return request.app.state.openai_serving_video
+
+
+def Omnichat(request: Request) -> OmniOpenAIServingChat | None:
+    return request.app.state.openai_serving_chat
+
+
+def OmniBatchChat(request: Request) -> OmniOpenAIServingChatBatch | None:
+    return request.app.state.openai_serving_chat_batch
+
+
+def Omnispeech(request: Request) -> OmniOpenAIServingSpeech | None:
+    return request.app.state.openai_serving_speech
+
+
+def OmniAudioGenerate(request: Request) -> OmniOpenAIServingAudioGenerate | None:
+    return getattr(request.app.state, "openai_serving_audio_generate", None)
+
+
+def _get_engine_and_model(raw_request: Request):
+    # Get engine client (AsyncOmni) from app state
+    engine_client: EngineClient | AsyncOmni | None = getattr(raw_request.app.state, "engine_client", None)
+    if engine_client is None:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE.value,
+            detail="Multi-stage engine not initialized. Start server with a multi-stage omni model.",
+        )
+
+    # Check if there's a diffusion stage.
+    # Prefer app state (compat layer populated at startup), then fall back to
+    # the engine client's stage configs for refactored AsyncOmni paths.
+    stage_configs = getattr(raw_request.app.state, "stage_configs", None)
+    if not stage_configs:
+        stage_configs = getattr(engine_client, "stage_configs", None)
+    if not stage_configs:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE.value,
+            detail="Stage configs not found. Start server with a multi-stage omni model.",
+        )
+
+    normalized_stage_configs = list(stage_configs)
+    has_diffusion_stage = any(get_stage_type(stage_cfg) == "diffusion" for stage_cfg in normalized_stage_configs)
+
+    if not has_diffusion_stage:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE.value,
+            detail="No diffusion stage found in multi-stage pipeline.",
+        )
+
+    # Get server's loaded model name
+    serving_models = getattr(raw_request.app.state, "openai_serving_models", None)
+    base_model_paths = getattr(serving_models, "base_model_paths", None) if serving_models else None
+    if base_model_paths:
+        model_name = base_model_paths[0].name
+    else:
+        model_name = "unknown"
+
+    return engine_client, model_name, normalized_stage_configs
+
+
+def _get_diffusion_od_config(raw_request: Request, engine_client: Any) -> Any:
+    diffusion_engine = getattr(raw_request.app.state, "diffusion_engine", None) or engine_client
+    get_diffusion_od_config = getattr(diffusion_engine, "get_diffusion_od_config", None)
+    return (
+        get_diffusion_od_config() if callable(get_diffusion_od_config) else getattr(diffusion_engine, "od_config", None)
+    )

vllm_omni/entrypoints/openai/audio_utils_mixin.py

@@ -1,5 +1,25 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Audio serving utility mixin.
+
+PUT HERE:
+  - AudioMixin helpers that convert audio tensors/bytes/formats for speech
+    and audio serving classes (shared by multiple serving paths).
+
+DO NOT PUT HERE:
+  - FastAPI route / form / job orchestration peeled from ``api_server.py``.
+    Put those under an audio package ``helpers.py`` (or audio serving modules)
+    when extracted.
+
+LONGEVITY:
+  - This root mixin is a **temporary shared home**.
+  - TODO(#5227, P1.1): tidy up / move with the audio/speech family split
+    in the Phase 1 audio PR; do not treat this file as the long-term owner.
+  - Endpoint-family helpers under an audio package are the longer home for
+    route-adjacent logic.
+
+See ``openai/README.md`` (utils vs helpers, no overlap).
+"""
 
 import math
 from io import BytesIO

vllm_omni/entrypoints/openai/chat_template.py

@@ -0,0 +1,55 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Chat-template loading helpers for OpenAI server bootstrap."""
+
+import json
+from pathlib import Path
+
+from vllm.logger import init_logger
+
+logger = init_logger(__name__)
+
+
+def _load_model_chat_template_json(model: str) -> str | None:
+    """Load a model-level chat_template.json from a local path or HF cache.
+
+    Some multimodal HF repos, including Qwen3-Omni, ship the chat template as a
+    separate file instead of embedding it in tokenizer_config.json. Transformers
+    4.44+ no longer supplies a default template, so serving must pass that model
+    template explicitly when the user did not provide --chat-template.
+    """
+    candidate = Path(model) / "chat_template.json"
+    template_path: str | None = str(candidate) if candidate.is_file() else None
+
+    if template_path is None:
+        try:
+            from vllm_omni.transformers_utils.repo_utils import hf_api
+
+            template_path = hf_api().hf_hub_download(
+                repo_id=model,
+                filename="chat_template.json",
+                local_files_only=True,
+            )
+        except Exception:
+            return None
+
+    try:
+        with open(template_path, encoding="utf-8") as f:
+            payload = json.load(f)
+    except Exception as exc:
+        logger.warning("Failed to load chat template from %s: %s", template_path, exc)
+        return None
+
+    if isinstance(payload, dict):
+        template = payload.get("chat_template")
+    elif isinstance(payload, str):
+        template = payload
+    else:
+        template = None
+
+    if not isinstance(template, str) or not template.strip():
+        logger.warning("Ignoring malformed chat template payload in %s", template_path)
+        return None
+
+    logger.info("Loaded chat template from %s", template_path)
+    return template

vllm_omni/entrypoints/openai/diffusion.py

@@ -0,0 +1,76 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""OpenAI diffusion-stage request helpers.
+
+These helpers are stage-based rather than image-format-specific."""
+
+import json
+from http import HTTPStatus
+from typing import Any, cast
+
+from fastapi import HTTPException
+
+from vllm_omni.entrypoints.async_omni import AsyncOmni
+from vllm_omni.entrypoints.openai.stage_params import (
+    build_stage_sampling_params_list,
+    get_default_sampling_params_list,
+)
+
+MAX_UINT32_SEED = 2**32 - 1
+
+
+async def _generate_with_async_omni(
+    engine_client: AsyncOmni | Any,
+    gen_params: Any,
+    stage_configs: list[Any],
+    **kwargs,
+):
+    engine_client = cast(AsyncOmni, engine_client)
+    result = None
+    normalized_stage_configs = list(stage_configs)
+    if not normalized_stage_configs:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE.value,
+            detail="Stage configs not found. Start server with a multi-stage omni model.",
+        )
+    sampling_params_list = build_stage_sampling_params_list(
+        normalized_stage_configs,
+        get_default_sampling_params_list(engine_client),
+        diffusion_params=gen_params,
+        replace_diffusion_params=True,
+    )
+
+    async for output in engine_client.generate(
+        sampling_params_list=sampling_params_list,
+        **kwargs,
+    ):
+        result = output
+
+    if result is None:
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+            detail="No output generated from multi-stage pipeline.",
+        )
+    return result
+
+
+def apply_stage_default_sampling_params(
+    default_params_json: str | None,
+    sampling_params: Any,
+    stage_key: str,
+) -> None:
+    """
+    Update a stage's sampling parameters with vLLM-Omni defaults.
+
+    Args:
+        default_params_json: JSON string of stage-keyed default parameters
+        sampling_params: The sampling parameters object to update
+        stage_key: The stage ID/key in the pipeline
+    """
+    if default_params_json is not None:
+        default_params_dict = json.loads(default_params_json)
+        if stage_key in default_params_dict:
+            stage_defaults = default_params_dict[stage_key]
+            for param_name, param_value in stage_defaults.items():
+                if hasattr(sampling_params, param_name):
+                    setattr(sampling_params, param_name, param_value)

vllm_omni/entrypoints/openai/errors.py

@@ -1,3 +1,53 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""OpenAI endpoint error helpers.
+
+This module owns helpers that are specific to OpenAI-compatible response
+shapes. Server-level exception handling belongs in
+``vllm_omni.entrypoints.serve.utils.errors``."""
+
+from http import HTTPStatus
+
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from vllm.entrypoints.serve.engine.protocol import ErrorResponse
+from vllm.entrypoints.serve.instrumentator.basic import base
+
+
 class InvalidInputReferenceError(ValueError):
     def __init__(self, message: str = "Invalid input reference.") -> None:
         super().__init__(message)
+
+
+def _error_response_to_json_response(
+    err: ErrorResponse,
+    *,
+    status_code: HTTPStatus | int | None = None,
+    default_status_code: HTTPStatus | int = HTTPStatus.BAD_REQUEST,
+) -> JSONResponse:
+    resolved_status = int(
+        status_code
+        if status_code is not None
+        else (err.error.code if err.error and err.error.code is not None else default_status_code)
+    )
+    payload = err.model_dump()
+    if err.error:
+        payload["error"]["code"] = resolved_status
+    return JSONResponse(content=payload, status_code=resolved_status)
+
+
+# TODO(#5227, P1.1): Relocate to audio/speech helpers when speech routes move out of api_server.py.
+def _create_speech_error_json_response(
+    raw_request: Request,
+    message: str,
+    *,
+    err_type: str = "BadRequestError",
+    status_code: HTTPStatus = HTTPStatus.BAD_REQUEST,
+) -> JSONResponse:
+    err = base(raw_request).create_error_response(
+        message=message,
+        err_type=err_type,
+        status_code=status_code,
+    )
+    return _error_response_to_json_response(err, status_code=status_code)

vllm_omni/entrypoints/openai/image_api_utils.py

@@ -1,11 +1,24 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """
-Shared helper utilities for OpenAI-compatible image generation API.
+Shared media utilities for OpenAI-compatible image APIs.
 
-This module provides common helper functions for the image generation endpoint.
-All functions work with plain Python types to maintain separation from the
-FastAPI HTTP layer.
+PUT HERE:
+  - Pure media helpers with no FastAPI Request / app-state / job semantics:
+    size parsing, image encode/base64, layered-layer validation.
+
+DO NOT PUT HERE:
+  - ``/v1/images*`` request/engine/job helpers — those go in ``images.helpers``.
+
+LONGEVITY:
+  - This root utils file is a **temporary shared home**.
+  - TODO(#5227, P1.2): tidy up / absorb into the images family (e.g. ``images/media.py``
+    or equivalent) in the P1.2 image modality PR; do not treat this file as the
+    long-term owner.
+  - ``images.helpers`` is the longer home for endpoint-specific logic through
+    P0.2/P0.3 until P1.2 further splits it.
+
+See ``openai/README.md`` and ``images/README.md`` (utils vs helpers, no overlap).
 """
 
 import base64

vllm_omni/entrypoints/openai/images/README.md

@@ -0,0 +1,31 @@
+# Image Endpoints
+
+This package owns OpenAI-compatible image generation and image edit endpoint
+logic.
+
+## `image_api_utils.py` vs `images/helpers.py` (no overlap)
+
+| | `openai/image_api_utils.py` | `openai/images/helpers.py` |
+| --- | --- | --- |
+| Layer | Shared media utils | Image endpoint helpers |
+| Longevity | **Temporary.** TODO(#5227, P1.2): absorb/tidy into this package (e.g. `media.py`) | **Longer home** for `/v1/images*` logic through P0.2/P0.3; P1.2 may split further |
+| PUT HERE | `parse_size`, encode/base64, layered-layer validation — plain Python, no FastAPI `Request` | Engine/app-state limits, input image loading from the edit request, Hunyuan edit extras, result extraction/normalization |
+| DO NOT PUT HERE | Request/engine/job helpers | Shared encode/`parse_size`/layered validation |
+| Typical importers | `protocol/images.py`, `serving_chat.py`, image routes | `api_server.py` image routes (later `images/api_router.py`) |
+| May import the other? | **No** — must not import `images.helpers` | **Yes** — may call `image_api_utils` |
+
+**Prefer growing `helpers.py`.** Do not grow root `image_api_utils.py` unless the symbol is clearly shared media with no endpoint semantics.
+
+## Put Here
+
+- Request parsing, validation, and response helpers used only by image
+  endpoints (`helpers.py`).
+- Image endpoint route bodies after route extraction.
+- Image-specific bridge/adapters introduced during the image refactor.
+
+## Do Not Put Here
+
+- Generic server utilities.
+- App-state accessors shared by multiple endpoint families.
+- Shared encode/size/layered validation already in `image_api_utils.py`.
+- Video or audio endpoint behavior.

vllm_omni/entrypoints/openai/images/helpers.py

@@ -0,0 +1,245 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Image endpoint helpers peeled from ``api_server.py``.
+
+PUT HERE:
+  - ``/v1/images*`` request/engine helpers: limits, input loading, edit extras,
+    result extraction/normalization, and other route-adjacent logic that used
+    to live in ``api_server.py``.
+  - May import ``image_api_utils`` for encode/validate (helpers → utils OK).
+
+DO NOT PUT HERE:
+  - Shared encode/size/layered validation — keep those in ``image_api_utils``.
+
+LONGEVITY:
+  - This package ``helpers.py`` is the **longer home** for image endpoint
+    logic through the helper (P0.2) + router (P0.3) stages.
+  - The later P1.2 image modality PR (#5227) may further split this file (e.g.
+    validation / editing / responses) and absorb ``image_api_utils`` into
+    this package; until then keep growing here, not in root utils.
+
+See ``images/README.md`` (utils vs helpers, no overlap).
+"""
+
+import io
+from http import HTTPStatus
+from numbers import Integral
+from typing import Any
+
+import httpx
+import numpy as np
+import pybase64 as base64
+from fastapi import HTTPException, Request
+from PIL import Image
+from vllm.logger import init_logger
+
+from vllm_omni.entrypoints.openai.app_state import _get_diffusion_od_config
+
+logger = init_logger(__name__)
+
+
+def _get_max_edit_input_images(raw_request: Request, engine_client: Any) -> int | None:
+    od_config = _get_diffusion_od_config(raw_request, engine_client)
+    if od_config is None:
+        # Preserve the existing compatibility behavior when the diffusion
+        # config is not exposed on the serving surface.
+        return None
+
+    supports_multimodal_inputs = getattr(od_config, "supports_multimodal_inputs", None)
+    if not isinstance(supports_multimodal_inputs, bool):
+        # Older serving surfaces and mocked engines may expose a placeholder
+        # object instead of a real diffusion config. Treat that as "unknown"
+        # so existing single-image flows keep working.
+        return None
+
+    if not supports_multimodal_inputs:
+        return 1
+
+    max_input_images = getattr(od_config, "max_multimodal_image_inputs", None)
+    if max_input_images is None:
+        return None
+    if isinstance(max_input_images, bool) or not isinstance(max_input_images, Integral):
+        return None
+    if max_input_images < 1:
+        return None
+    return int(max_input_images)
+
+
+def _check_max_generated_image_size(
+    app_state_args: Any,
+    width: int | None,
+    height: int | None,
+    resolution: int | None = None,
+) -> None:
+    """Raise 400 if the requested image size exceeds --max-generated-image-size."""
+    max_generated_image_size = getattr(app_state_args, "max_generated_image_size", None)
+    # Check max_generated_image_size
+    if max_generated_image_size is None:
+        return
+    if width is not None and height is not None:
+        if width * height > max_generated_image_size:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail=f"Requested image size {width}x{height} exceeds the maximum allowed "
+                f"size of {max_generated_image_size} pixels. You can reduce the requested size "
+                f"or increase the server's --max-generated-image-size limit.",
+            )
+    elif resolution is not None:
+        # When resolution is set, the output size is resolution * resolution
+        if resolution * resolution > max_generated_image_size:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail=f"Requested resolution {resolution} (max {resolution}x{resolution} pixels) "
+                f"exceeds the maximum allowed size of {max_generated_image_size} pixels. "
+                f"You can reduce the requested size or increase the server's --max-generated-image-size limit.",
+            )
+
+
+def _build_hunyuan_edit_extra_args(
+    *,
+    bot_task: str | None,
+    sys_type: str | None,
+    system_prompt: str | None,
+) -> dict[str, Any]:
+    """Map Hunyuan /v1/images/edits form fields to DiT ``extra_args``."""
+    extra_args: dict[str, Any] = {}
+    effective_use_system_prompt = sys_type
+    if effective_use_system_prompt is None and bot_task is not None:
+        from vllm_omni.diffusion.models.hunyuan_image3.prompt_utils import resolve_sys_type
+
+        effective_use_system_prompt = resolve_sys_type(bot_task)
+    if effective_use_system_prompt is not None:
+        extra_args["use_system_prompt"] = effective_use_system_prompt
+    if system_prompt is not None:
+        extra_args["system_prompt"] = system_prompt
+    if bot_task is not None:
+        extra_args["bot_task"] = bot_task
+    return extra_args
+
+
+def _update_if_not_none(object: Any, key: str, val: Any) -> None:
+    if val is not None:
+        setattr(object, key, val)
+
+
+def _normalize_image(image: Any) -> Any:
+    """Normalize a single image output to a PIL-compatible format."""
+    if isinstance(image, Image.Image):
+        return image
+    if not isinstance(image, np.ndarray):
+        raise ValueError(f"Unsupported image type: {type(image)}")
+    if not np.issubdtype(image.dtype, np.integer) and not np.issubdtype(image.dtype, np.floating):
+        raise ValueError(f"Unsupported dtype: {image.dtype}")
+    if isinstance(image, np.ndarray):
+        while image.ndim > 3:
+            image = image[0]
+        if image.min() < 0:
+            if image.min() < -1.01 or image.max() > 1.01:
+                logger.warning(
+                    f"Image float range [{image.min():.2f}, {image.max():.2f}] outside expected [-1, 1]. "
+                    f"Clipping to [-1, 1] before normalization."
+                )
+            image = np.clip(image, -1.0, 1.0) * 0.5 + 0.5
+        elif image.max() > 1.01:
+            logger.warning(
+                f"Image float range [{image.min():.2f}, {image.max():.2f}] outside expected [0, 1]. "
+                f"Clipping to [0, 1] before normalization."
+            )
+        image = (np.clip(image, 0.0, 1.0) * 255).astype(np.uint8)
+        image = Image.fromarray(image)
+    return image
+
+
+def _extract_images_from_result(result: Any) -> list[Any]:
+    images = []
+    if hasattr(result, "images") and result.images:
+        images = result.images
+    # Handle when generate more than one image
+    if images and isinstance(images[0], np.ndarray) and images[0].shape[0] > 1 and images[0].ndim == 5:
+        # Unwrap batch: (N, T, H, W, C) -> [img1, img2, ...]
+        images = list(images[0])
+    # Flatten nested lists (e.g., from layered models like Qwen-Image-Layered).
+    # Note: This only flattens one level deep. Deeper nesting is not supported.
+    flattened = []
+    for img in images:
+        if isinstance(img, list):
+            flattened.extend(img)
+        else:
+            flattened.append(img)
+    return [_normalize_image(img) for img in flattened]
+
+
+async def _load_input_images(
+    inputs: list[str],
+    *,
+    normalize_rgb: bool = True,
+) -> list[Image.Image]:
+    """
+    convert to PIL.Image.Image list
+    """
+    if isinstance(inputs, str):
+        inputs = [inputs]
+
+    images: list[Image.Image] = []
+
+    for inp in inputs:
+        # 1. URL + base64
+        if isinstance(inp, str) and inp.startswith("data:image"):
+            try:
+                _, b64_data = inp.split(",", 1)
+                image_bytes = base64.b64decode(b64_data)
+                img = Image.open(io.BytesIO(image_bytes))
+                images.append(img)
+            except Exception as e:
+                raise ValueError(f"Invalid base64 image: {e}")
+
+        # 2. URL
+        elif isinstance(inp, str) and inp.startswith("http"):
+            async with httpx.AsyncClient(timeout=60) as client:
+                try:
+                    resp = await client.get(inp)
+                    resp.raise_for_status()
+                    img = Image.open(io.BytesIO(resp.content))
+                    images.append(img)
+                except Exception as e:
+                    raise ValueError(f"Failed to download image from URL {inp}: {e}")
+
+        # 3. UploadFile
+        elif hasattr(inp, "file"):
+            try:
+                img_data = await inp.read()
+                img = Image.open(io.BytesIO(img_data))
+                images.append(img)
+            except Exception as e:
+                raise ValueError(f"Failed to open uploaded file: {e}")
+
+        else:
+            raise ValueError(f"Unsupported input: {inp}")
+
+    if not images:
+        raise ValueError("No valid input images found")
+
+    if not normalize_rgb:
+        return images
+
+    # Match the offline HunyuanImage3 image-edit example path, which eagerly
+    # normalizes input files with ``Image.open(...).convert("RGB")`` before
+    # they reach the AR stage. Keeping uploads as RGBA/P PIL objects makes
+    # online IT2I observe a different visual input than offline (for example
+    # transparent-logo uploads alpha-composited over white instead of black),
+    # which is enough for HunyuanImage3 AR recaption to diverge before DiT
+    # sees the request -- root cause of the "online 3 magnets vs offline 1
+    # magnet" systematic semantic mismatch.
+    return [img.convert("RGB") for img in images]
+
+
+def _choose_output_format(output_format: str | None, background: str | None) -> str:
+    # Normalize and choose extension
+    fmt = (output_format or "").lower()
+    if fmt in {"jpg", "png", "webp", "jpeg"}:
+        return fmt
+    # If transparency requested, prefer png
+    if (background or "auto").lower() == "transparent":
+        return "png"
+    # Default
+    return "jpeg"

vllm_omni/entrypoints/openai/lora.py

@@ -0,0 +1,38 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""OpenAI LoRA request helpers.
+
+Use this module for HTTP-facing LoRA parsing shared by OpenAI-compatible
+endpoint families."""
+
+import json
+from http import HTTPStatus
+from typing import Any
+
+from fastapi import HTTPException
+
+from vllm_omni.entrypoints.openai.utils import parse_lora_request
+
+
+def _get_lora_from_json_str(lora_body):
+    if lora_body is None:
+        return None
+    try:
+        lora_dict = json.loads(lora_body)
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=400, detail="Invalid LoRA JSON string")
+
+    if not isinstance(lora_dict, dict):
+        raise HTTPException(status_code=400, detail="LoRA must be a JSON object")
+
+    return lora_dict
+
+
+def _parse_lora_request(lora_body: dict[str, Any]):
+    try:
+        return parse_lora_request(lora_body)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=str(e),
+        ) from e

vllm_omni/entrypoints/openai/models/README.md

@@ -0,0 +1,19 @@
+# OpenAI Models
+
+This package mirrors upstream OpenAI model-list ownership.
+
+## Put Here
+
+- Helpers and serving shims that back `/v1/models`.
+- Route bodies for `/v1/models` after endpoint extraction.
+- Model-list protocol adapters that are OpenAI-specific.
+
+## Do Not Put Here
+
+- Actual model execution logic.
+- Image/video/audio endpoint helpers.
+- Generic app-state helpers.
+
+`_DiffusionServingModels` lives here because pure diffusion servers still need
+to satisfy the OpenAI `/v1/models` contract. The route body remains in
+`openai/api_server.py` until the route extraction tracked by #5227.

vllm_omni/entrypoints/openai/models/serving.py

@@ -0,0 +1,59 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""OpenAI models serving helpers.
+
+This module mirrors upstream's OpenAI models area for helpers that back
+``/v1/models`` behavior."""
+
+from vllm.entrypoints.openai.models.protocol import BaseModelPath
+from vllm.entrypoints.serve.engine.protocol import ModelCard, ModelList, ModelPermission
+
+
+class _DiffusionServingModels:
+    """Minimal OpenAIServingModels implementation for diffusion-only servers.
+
+    vLLM's /v1/models route expects `app.state.openai_serving_models` to expose
+    `show_available_models()`. In pure diffusion mode we don't initialize the
+    full OpenAIServingModels (it depends on LLM-specific processors), so we
+    provide a lightweight fallback.
+    """
+
+    class _NullModelConfig:
+        def __getattr__(self, name):
+            return None
+
+    class _Unsupported:
+        def __init__(self, name: str):
+            self.name = name
+
+        def __call__(self, *args, **kwargs):
+            raise NotImplementedError(f"{self.name} is not supported in diffusion mode")
+
+        def __getattr__(self, attr):
+            raise NotImplementedError(f"{self.name}.{attr} is not supported in diffusion mode")
+
+    def __init__(self, base_model_paths: list[BaseModelPath]) -> None:
+        self._base_model_paths = base_model_paths
+        self.model_config = self._NullModelConfig()
+
+    @property
+    def base_model_paths(self) -> list[BaseModelPath]:
+        return self._base_model_paths
+
+    def is_base_model(self, model_name: str) -> bool:
+        return any(p.name == model_name for p in self._base_model_paths)
+
+    def __getattr__(self, name):
+        return self._Unsupported(name)
+
+    async def show_available_models(self) -> ModelList:
+        return ModelList(
+            data=[
+                ModelCard(
+                    id=base_model.name,
+                    root=base_model.model_path,
+                    permission=[ModelPermission()],
+                )
+                for base_model in self._base_model_paths
+            ]
+        )

vllm_omni/entrypoints/openai/utils.py

@@ -1,4 +1,22 @@
 # SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Shared OpenAI serving utilities.
+
+PUT HERE:
+  - Cross-cutting stage-config / LoRA / pipeline helpers reused by multiple
+    OpenAI serving paths (not owned by a single endpoint family).
+
+DO NOT PUT HERE:
+  - Modality endpoint request/job helpers — those go in package ``helpers.py``
+    (e.g. ``images.helpers``, ``video.generation.helpers``).
+  - Pure image/video/audio media encode/decode — those stay in the matching
+    root ``*_utils*`` / mixin until family PRs absorb them.
+
+LONGEVITY:
+  - TODO(#5227, Phase 1): revisit and tidy this file during later OpenAI
+    entrypoint refactor stages; prefer growing endpoint-family helpers over this
+    root grab-bag when ownership is clear.
+"""
 
 from __future__ import annotations
 

vllm_omni/entrypoints/openai/video/README.md

@@ -0,0 +1,33 @@
+# Video Endpoints
+
+This package owns OpenAI-compatible video endpoint families.
+
+## `video_api_utils.py` vs `video/generation/helpers.py` (no overlap)
+
+| | `openai/video_api_utils.py` | `openai/video/generation/helpers.py` |
+| --- | --- | --- |
+| Layer | Shared video media utils | `/v1/videos*` endpoint helpers |
+| Longevity | **Temporary.** TODO(#5227, P1.3): move/tidy into family `media.py` | **Longer home** for generation resource logic through P0.2/P0.3; P1.3 may split further |
+| PUT HERE | Decode/encode URLs and bytes, frame/audio coercion, streaming encoders | Multipart form parsing, MiniMax upload limits, job runner/cleanup, `VideoResponse` factory, runtime context from app state |
+| DO NOT PUT HERE | Form/job/store orchestration | Reimplemented encode/decode |
+| Typical importers | `serving_video.py`, streaming output, generation helpers | `api_server.py` video routes (later `generation/api_router.py`) |
+| May import the other? | **No** — must not import `video.generation.helpers` | **Yes** — may call `video_api_utils` |
+
+**Prefer growing `generation/helpers.py`.** Do not grow root `video_api_utils.py` unless the symbol is clearly shared media.
+
+## Put Here
+
+- Video generation/resource endpoint code under `generation/`.
+- Generated video streaming-output code under a dedicated subpackage when that
+  route moves.
+- Video-specific request parsing, response formatting, and storage/job helpers
+  for the owning subpackage.
+
+## Do Not Put Here
+
+- Live streaming-input chat sessions such as `/v1/video/chat/stream`; those
+  should get their own route-owner package in the router split rather than
+  living under video generation.
+- Generic server utilities.
+- Shared media encode/decode that already lives in root `video_api_utils.py`.
+- Image or audio endpoint behavior.

vllm_omni/entrypoints/openai/video/generation/README.md

@@ -0,0 +1,33 @@
+# Video Generation
+
+This package owns the `/v1/videos` resource family.
+
+## Helpers vs root video utils
+
+See also `../README.md` and `../../README.md`.
+
+| | PUT HERE | DO NOT PUT HERE | Longevity |
+| --- | --- | --- | --- |
+| `helpers.py` | Form parse, upload validation, job run/cleanup, response factories, app-state runtime context for `/v1/videos*` | Shared encode/decode backends | **Longer home** through P0.2/P0.3; later P1.3 PR may split (form/jobs) |
+| `openai/video_api_utils.py` | Shared encode/decode/media backends only | Form/job/store orchestration | **Temporary.** TODO(#5227, P1.3): move into e.g. `media.py` |
+
+- Helpers call utils; utils never grow endpoint/job logic.
+- Prefer growing `helpers.py` over root utils.
+
+## Put Here
+
+- Multipart form parsing for video generation requests.
+- Async/sync video generation job helpers.
+- Video resource status, delete, retrieve, and download helpers.
+- The `api_router.py` for `/v1/videos*` once route bodies move out of
+  `api_server.py`.
+
+## Do Not Put Here
+
+- Live video-chat streaming input.
+- Generated video streaming-output sessions.
+- Generic video encoding/decoding helpers used outside this resource family
+  (keep those in `video_api_utils.py` until the P1.3 video family PR moves them).
+
+The P0.2 helper migration places non-route video generation code here first. Full
+route extraction (P0.3) and `video_api_utils.py` cleanup (P1.3) are tracked by #5227.

vllm_omni/entrypoints/openai/video/generation/helpers.py

@@ -0,0 +1,986 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""``/v1/videos*`` endpoint helpers peeled from ``api_server.py``.
+
+PUT HERE:
+  - Multipart form parsing, upload limits, job run/cleanup, response factories,
+    app-state runtime context, and other route-adjacent ``/v1/videos*`` logic
+    that used to live in ``api_server.py``.
+  - May import ``video_api_utils`` for decode/encode (helpers → utils OK).
+
+DO NOT PUT HERE:
+  - Shared media encode/decode backends — keep those in ``video_api_utils``.
+
+LONGEVITY:
+  - This package ``helpers.py`` is the **longer home** for video generation
+    endpoint logic through the helper (P0.2) + router (P0.3) stages.
+  - The later P1.3 video modality PR (#5227) may further split this file (e.g. form /
+    jobs) and move ``video_api_utils`` into ``media.py`` here; until then keep
+    growing here, not in root utils.
+
+See ``video/generation/README.md`` (utils vs helpers, no overlap).
+"""
+
+import asyncio
+import io
+import json
+import os
+import tempfile
+import time
+from collections.abc import Mapping, Sequence
+from contextlib import suppress
+from http import HTTPStatus
+from numbers import Integral
+from pathlib import Path
+from typing import Any, Literal, cast
+
+from fastapi import File, Form, HTTPException, Request, UploadFile
+from PIL import Image
+from vllm.entrypoints.launchers.launcher import terminate_if_errored
+from vllm.entrypoints.serve import create_error_response
+from vllm.logger import init_logger
+from vllm.v1.engine.exceptions import EngineDeadError, EngineGenerateError
+
+from vllm_omni.diffusion.models.interface import ReferenceVideoDecodeSpec
+from vllm_omni.entrypoints.openai.app_state import Omnivideo
+from vllm_omni.entrypoints.openai.errors import InvalidInputReferenceError
+from vllm_omni.entrypoints.openai.protocol.videos import (
+    SecondStr,
+    SizeStr,
+    VideoAction,
+    VideoError,
+    VideoGenerationRequest,
+    VideoGenerationStatus,
+    VideoResponse,
+)
+from vllm_omni.entrypoints.openai.serving_video import (
+    OmniOpenAIServingVideo,
+    ReferenceAudio,
+    ReferenceImage,
+    ReferenceVideo,
+)
+from vllm_omni.entrypoints.openai.storage import STORAGE_MANAGER
+from vllm_omni.entrypoints.openai.stores import VIDEO_STORE
+from vllm_omni.entrypoints.openai.utils import get_stage_type
+from vllm_omni.entrypoints.openai.video_api_utils import (
+    VideoFrames,
+    _decode_image_bytes,
+    _validate_image_pixel_limit,
+    decode_audio_url,
+    decode_input_reference,
+)
+from vllm_omni.errors import OmniClientError
+
+logger = init_logger(__name__)
+
+
+MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES = 30 * 1024 * 1024
+
+MINIMAX_H3_MAX_REFERENCE_VIDEO_BYTES = 50 * 1024 * 1024
+
+MINIMAX_H3_MAX_REFERENCE_AUDIO_BYTES = 15 * 1024 * 1024
+
+MINIMAX_H3_MAX_REFERENCE_COUNT = 12
+
+MINIMAX_H3_REFERENCE_IMAGE_FORMATS = frozenset({"jpeg", "png", "webp", "heic", "heif"})
+
+MINIMAX_H3_REFERENCE_VIDEO_SUFFIXES = frozenset({".mp4", ".mov"})
+
+MINIMAX_H3_REFERENCE_AUDIO_SUFFIXES = frozenset({".wav", ".mp3"})
+
+CONTROL_REFERENCE_IMAGE_SUFFIXES = frozenset({".bmp", ".gif", ".jpg", ".jpeg", ".png", ".tif", ".tiff", ".webp"})
+CONTROL_REFERENCE_VIDEO_SUFFIXES = frozenset({".mkv", ".mov", ".mp4", ".webm"})
+CONTROL_REFERENCE_MAX_BYTES = 512 * 1024 * 1024
+
+VIDEO_SYNC_TIMEOUT_S = float(os.environ.get("VLLM_OMNI_VIDEO_SYNC_TIMEOUT", 600.0))
+
+
+def _resolve_video_runtime_context(raw_request: Request) -> tuple[str | None, list[Any] | None]:
+    app_model_name = None
+    serving_models = getattr(raw_request.app.state, "openai_serving_models", None)
+    if serving_models and getattr(serving_models, "base_model_paths", None):
+        base_paths = serving_models.base_model_paths
+        if base_paths:
+            app_model_name = base_paths[0].name
+
+    app_stage_configs = getattr(raw_request.app.state, "stage_configs", None)
+    return app_model_name, app_stage_configs
+
+
+def _parse_form_json(value: str | None, expected_type: type | None = None) -> Any:
+    if value is None or value == "":
+        return None
+    try:
+        parsed = json.loads(value)
+    except json.JSONDecodeError as exc:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Invalid JSON in form field.",
+        ) from exc
+    if expected_type is not None and not isinstance(parsed, expected_type):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=f"Invalid JSON in form field: expected {expected_type.__name__}, got {type(parsed).__name__}.",
+        )
+    return parsed
+
+
+def _config_get(config: Any, key: str, default: Any = None) -> Any:
+    if isinstance(config, dict):
+        return config.get(key, default)
+    if hasattr(config, "get"):
+        try:
+            return config.get(key, default)
+        except Exception:
+            pass
+    return getattr(config, key, default)
+
+
+def _stage_engine_args(stage_cfg: Any) -> Any:
+    return _config_get(stage_cfg, "engine_args", {}) or {}
+
+
+def _diffusion_model_classes(stage_configs: list[Any] | None) -> list[type]:
+    if not stage_configs:
+        return []
+
+    from vllm_omni.diffusion.registry import DiffusionModelRegistry
+
+    model_classes: list[type] = []
+    for stage_cfg in stage_configs:
+        if get_stage_type(stage_cfg) != "diffusion":
+            continue
+        model_class_name = _config_get(_stage_engine_args(stage_cfg), "model_class_name")
+        if not model_class_name:
+            continue
+        model_cls = DiffusionModelRegistry._try_load_model_cls(model_class_name)
+        if model_cls is not None:
+            model_classes.append(model_cls)
+    return model_classes
+
+
+def _normalize_reference_video_decode_spec(spec: ReferenceVideoDecodeSpec) -> ReferenceVideoDecodeSpec:
+    max_frames = spec.max_frames
+    if max_frames is not None:
+        try:
+            max_frames = int(max_frames)
+        except (TypeError, ValueError) as exc:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail="Invalid reference video decode spec: max_frames must be an integer.",
+            ) from exc
+        if max_frames <= 0:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail="Invalid reference video decode spec: max_frames must be positive.",
+            )
+
+    keep = str(spec.keep or "first").strip().lower()
+    if keep not in {"first", "last"}:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Invalid reference video decode spec: keep must be either 'first' or 'last'.",
+        )
+    return ReferenceVideoDecodeSpec(max_frames=max_frames, keep=cast(Literal["first", "last"], keep))
+
+
+def _reference_video_decode_spec(
+    req: VideoGenerationRequest,
+    stage_configs: list[Any] | None,
+) -> ReferenceVideoDecodeSpec:
+    video_params = req.resolve_video_params()
+    extra_params = req.extra_params if isinstance(req.extra_params, dict) else {}
+    for model_cls in _diffusion_model_classes(stage_configs):
+        resolver = getattr(model_cls, "reference_video_decode_spec", None)
+        if resolver is None:
+            continue
+        try:
+            spec = resolver(num_frames=video_params.num_frames, extra_args=extra_params)
+        except (TypeError, ValueError) as exc:
+            raise HTTPException(status_code=HTTPStatus.BAD_REQUEST.value, detail=str(exc)) from exc
+        if spec is not None:
+            return _normalize_reference_video_decode_spec(spec)
+    return ReferenceVideoDecodeSpec(max_frames=video_params.num_frames, keep="first")
+
+
+def video_response_from_request(model_name: str, req: VideoGenerationRequest) -> VideoResponse:
+    video_params = req.resolve_video_params()
+    duration_s = None
+    if req.seconds is not None:
+        duration_s = float(req.seconds)
+    elif video_params.num_frames is not None and video_params.fps is not None:
+        duration_s = video_params.num_frames / video_params.fps
+
+    resp = VideoResponse(
+        model=model_name,
+        status=VideoGenerationStatus.QUEUED,
+        size=req.size,
+        prompt=req.prompt,
+        quality=req.quality or "default",
+        fps=video_params.fps,
+        num_frames=video_params.num_frames,
+        duration_s=duration_s,
+    )
+    resp.seconds = str(req.seconds or resp.seconds)
+    return resp
+
+
+def _status_code_for_video_failure(error: VideoError | None) -> int:
+    if error is None:
+        return HTTPStatus.INTERNAL_SERVER_ERROR.value
+
+    if isinstance(error.code, int):
+        if 400 <= error.code < 600:
+            return error.code
+        return HTTPStatus.INTERNAL_SERVER_ERROR.value
+
+    if error.code == "HTTPException":
+        status_text, _, _ = error.message.partition(":")
+        try:
+            status_code = int(status_text)
+        except ValueError:
+            return HTTPStatus.INTERNAL_SERVER_ERROR.value
+        if 400 <= status_code < 600:
+            return status_code
+        return HTTPStatus.INTERNAL_SERVER_ERROR.value
+
+    if error.code == "EngineDeadError":
+        return HTTPStatus.INTERNAL_SERVER_ERROR.value
+    if error.code == "EngineGenerateError":
+        return HTTPStatus.INTERNAL_SERVER_ERROR.value
+
+    return HTTPStatus.INTERNAL_SERVER_ERROR.value
+
+
+def _video_error_from_exception(exc: Exception) -> VideoError:
+    if isinstance(exc, HTTPException):
+        message = str(exc.detail) if exc.detail else str(exc)
+        return VideoError(code=exc.status_code, message=message)
+
+    if isinstance(exc, OmniClientError):
+        return VideoError(code=exc.status_code, message=exc.message)
+
+    if isinstance(exc, (EngineGenerateError, EngineDeadError)):
+        err = create_error_response(exc)
+        return VideoError(code=err.error.code, message=err.error.message)
+
+    return VideoError(
+        code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+        message=str(exc),
+    )
+
+
+async def _cleanup_video(video_id: str):
+    try:
+        await STORAGE_MANAGER.delete(video_id)
+    except Exception:
+        logger.warning("Failed to cleanup partial video file '%s'", video_id)
+
+
+def _cleanup_video_references(
+    reference_video: ReferenceVideo | None,
+    reference_audio: ReferenceAudio | None,
+    control_path: str | None = None,
+) -> None:
+    if reference_video is not None:
+        for path in reference_video.cleanup_paths:
+            if os.path.exists(path):
+                os.unlink(path)
+    if reference_audio is not None:
+        cleanup_paths = reference_audio.cleanup_paths or tuple(_reference_list(reference_audio.path))
+        for path in cleanup_paths:
+            if os.path.exists(path):
+                os.unlink(path)
+    if control_path is not None and os.path.exists(control_path):
+        os.unlink(control_path)
+
+
+def _unpack_video_generation_result(
+    result: Sequence[object],
+) -> tuple[bytes, dict[str, float], float, VideoAction | None, dict[str, object]]:
+    video_metadata: dict[str, object] = {}
+    if len(result) == 5:
+        video_bytes, stage_durations, peak_memory_mb, action, raw_metadata = result
+        if isinstance(raw_metadata, dict):
+            video_metadata = {str(key): value for key, value in raw_metadata.items()}
+    else:
+        video_bytes, stage_durations, peak_memory_mb, action = result
+    return (
+        cast(bytes, video_bytes),
+        cast(dict[str, float], stage_durations),
+        float(cast(float, peak_memory_mb)),
+        cast(VideoAction | None, action),
+        video_metadata,
+    )
+
+
+async def _run_video_generation_job(
+    handler: OmniOpenAIServingVideo,
+    request: VideoGenerationRequest,
+    video_id: str,
+    reference_image: ReferenceImage | None = None,
+    reference_video: ReferenceVideo | None = None,
+    reference_audio: ReferenceAudio | None = None,
+    control_path: str | None = None,
+    app_state: Any | None = None,
+) -> None:
+    job = await VIDEO_STORE.get(video_id)
+    if job is None:
+        logger.warning("Video job %s missing before generation task started; skipping", video_id)
+        _cleanup_video_references(reference_video, reference_audio, control_path)
+        return
+
+    await VIDEO_STORE.update_fields(video_id, {"status": VideoGenerationStatus.IN_PROGRESS})
+    started_at = time.perf_counter()
+    try:
+        video_bytes, stage_durations, peak_memory_mb, action, video_metadata = _unpack_video_generation_result(
+            await handler.generate_video_bytes(
+                request,
+                video_id,
+                reference_image=reference_image,
+                reference_video=reference_video,
+                reference_audio=reference_audio,
+            )
+        )
+
+        save_context = await STORAGE_MANAGER.save(video_bytes, video_id)
+        logger.info("Video request %s persisted %s output file.", video_id, save_context.key)
+
+        updated_fields = {
+            "status": VideoGenerationStatus.COMPLETED,
+            "progress": 100,
+            "file_name": f"{video_id}.{job.file_extension}",
+            "completed_at": save_context.created_at,
+            "inference_time_s": time.perf_counter() - started_at,
+            "stage_durations": stage_durations,
+            "peak_memory_mb": peak_memory_mb,
+            "action": action,
+        }
+        updated_fields.update({key: value for key, value in video_metadata.items() if value is not None})
+        if save_context.expires_at is not None:
+            updated_fields["expires_at"] = save_context.expires_at
+
+        await VIDEO_STORE.update_fields(video_id, updated_fields)
+    except (EngineGenerateError, EngineDeadError) as exc:
+        logger.exception("Video generation failed (engine error) for id=%s", video_id)
+
+        await _cleanup_video(video_id)
+        await VIDEO_STORE.update_fields(
+            video_id,
+            {
+                "status": VideoGenerationStatus.FAILED,
+                "completed_at": int(time.time()),
+                "error": _video_error_from_exception(exc),
+                "inference_time_s": time.perf_counter() - started_at,
+            },
+        )
+        # Background tasks can't propagate exceptions to FastAPI handlers.
+        # Actively signal shutdown when the engine is dead.
+        if app_state is not None and isinstance(exc, EngineDeadError):
+            terminate_if_errored(
+                server=app_state.server,
+                engine=app_state.engine_client,
+            )
+    except Exception as exc:
+        logger.exception("Video generation failed for id=%s", video_id)
+
+        await _cleanup_video(video_id)
+        await VIDEO_STORE.update_fields(
+            video_id,
+            {
+                "status": VideoGenerationStatus.FAILED,
+                "completed_at": int(time.time()),
+                "error": _video_error_from_exception(exc),
+                "inference_time_s": time.perf_counter() - started_at,
+            },
+        )
+    except asyncio.CancelledError:
+        await _cleanup_video(video_id)
+        await VIDEO_STORE.pop(video_id)
+        raise
+    finally:
+        _cleanup_video_references(reference_video, reference_audio, control_path)
+
+
+async def _persist_uploaded_video_references(uploads: list[UploadFile]) -> list[str]:
+    paths: list[str] = []
+    try:
+        for upload in uploads:
+            suffix = Path(upload.filename or "").suffix.lower()
+            if suffix not in {".mkv", ".mov", ".mp4", ".webm"}:
+                suffix = ".mp4"
+            fd, path = tempfile.mkstemp(prefix="vllm_omni_video_reference_", suffix=suffix)
+            paths.append(path)
+            with os.fdopen(fd, "wb") as output:
+                while chunk := await upload.read(1024 * 1024):
+                    output.write(chunk)
+    except Exception:
+        for path in paths:
+            if os.path.exists(path):
+                os.unlink(path)
+        raise
+    return paths
+
+
+async def _persist_uploaded_control_reference(
+    upload: UploadFile,
+    *,
+    max_bytes: int = CONTROL_REFERENCE_MAX_BYTES,
+) -> str:
+    """Stream one model control upload to request-scoped local storage."""
+    kind = _uploaded_media_kind(upload)
+    suffix = Path(upload.filename or "").suffix.lower()
+    supported_suffixes = CONTROL_REFERENCE_IMAGE_SUFFIXES | CONTROL_REFERENCE_VIDEO_SUFFIXES
+    if kind == "audio" or (suffix and suffix not in supported_suffixes):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="control_reference must be an image or video file.",
+        )
+    if not suffix:
+        suffix = ".png" if kind == "image" else ".mp4"
+
+    declared_size = getattr(upload, "size", None)
+    if isinstance(declared_size, Integral) and int(declared_size) > max_bytes:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=f"control_reference exceeds the {max_bytes // (1024 * 1024)} MiB size limit.",
+        )
+
+    fd, path = tempfile.mkstemp(prefix="vllm_omni_control_reference_", suffix=suffix)
+    size = 0
+    persisted = False
+    try:
+        with os.fdopen(fd, "wb") as output:
+            while chunk := await upload.read(min(1024 * 1024, max_bytes - size + 1)):
+                size += len(chunk)
+                if size > max_bytes:
+                    raise HTTPException(
+                        status_code=HTTPStatus.BAD_REQUEST.value,
+                        detail=f"control_reference exceeds the {max_bytes // (1024 * 1024)} MiB size limit.",
+                    )
+                output.write(chunk)
+        if size == 0:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail="control_reference must not be empty.",
+            )
+        persisted = True
+        return path
+    finally:
+        if not persisted:
+            with suppress(OSError):
+                os.unlink(path)
+
+
+def _validate_control_upload(
+    handler: OmniOpenAIServingVideo,
+    request: VideoGenerationRequest,
+    control_reference: UploadFile | None,
+    control_type: str | None,
+) -> str | None:
+    """Validate a generic uploaded control against model capabilities."""
+    if control_reference is None:
+        if control_type is not None:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail="control_type requires a control_reference upload.",
+            )
+        return None
+    if control_type is None or not control_type.strip():
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="control_reference requires control_type.",
+        )
+
+    normalized_type = control_type.strip().lower()
+    supported_types = frozenset(getattr(handler, "supported_control_upload_types", ()))
+    if normalized_type not in supported_types:
+        supported = ", ".join(sorted(supported_types)) or "none"
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=(
+                f"Control upload type '{normalized_type}' is not supported by this model. "
+                f"Supported control upload types: {supported}."
+            ),
+        )
+
+    existing = (request.extra_params or {}).get(normalized_type)
+    if existing is not None:
+        if not isinstance(existing, Mapping) and existing is not True:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail=f"extra_params.{normalized_type} must be an object when control_reference is uploaded.",
+            )
+        if isinstance(existing, Mapping) and any(existing.get(key) is not None for key in ("control", "control_path")):
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail=(
+                    "Provide either control_reference or "
+                    f"extra_params.{normalized_type}.control/control_path, not both."
+                ),
+            )
+    return normalized_type
+
+
+def _attach_control_upload(
+    request: VideoGenerationRequest,
+    control_type: str,
+    control_path: str | None = None,
+) -> None:
+    """Declare an uploaded control and attach its persisted path when available."""
+    extra_params = dict(request.extra_params or {})
+    existing = extra_params.get(control_type)
+    control_params = dict(existing) if isinstance(existing, Mapping) else {}
+    if control_path is not None:
+        control_params["control_path"] = control_path
+    extra_params[control_type] = control_params
+    request.extra_params = extra_params
+
+
+def _reference_list(value: Any) -> list[Any]:
+    if value is None:
+        return []
+    return list(value) if isinstance(value, list) else [value]
+
+
+def _uploaded_media_kind(upload: UploadFile) -> str:
+    content_type = (upload.content_type or "").lower()
+    if content_type.startswith("image/"):
+        return "image"
+    if content_type.startswith("audio/"):
+        return "audio"
+    suffix = Path(upload.filename or "").suffix.lower()
+    if suffix in {".png", ".jpg", ".jpeg", ".webp", ".heic", ".heif"}:
+        return "image"
+    if suffix in {".wav", ".mp3", ".m4a", ".aac", ".flac", ".ogg"}:
+        return "audio"
+    return "video"
+
+
+def _minimax_h3_upload_limit(upload: UploadFile) -> int:
+    kind = _uploaded_media_kind(upload)
+    if kind == "image":
+        return MINIMAX_H3_MAX_REFERENCE_IMAGE_BYTES
+    if kind == "audio":
+        return MINIMAX_H3_MAX_REFERENCE_AUDIO_BYTES
+    return MINIMAX_H3_MAX_REFERENCE_VIDEO_BYTES
+
+
+async def _read_upload_limited(upload: UploadFile, *, max_bytes: int | None = None) -> bytes:
+    """Read an upload with an optional hard byte limit."""
+    if max_bytes is None:
+        return await upload.read()
+
+    declared_size = getattr(upload, "size", None)
+    if isinstance(declared_size, Integral) and int(declared_size) > max_bytes:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=f"Uploaded reference exceeds the {max_bytes // (1024 * 1024)} MiB size limit.",
+        )
+
+    chunks: list[bytes] = []
+    total = 0
+    while True:
+        chunk = await upload.read(min(1024 * 1024, max_bytes - total + 1))
+        if not chunk:
+            break
+        total += len(chunk)
+        if total > max_bytes:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail=f"Uploaded reference exceeds the {max_bytes // (1024 * 1024)} MiB size limit.",
+            )
+        chunks.append(chunk)
+    return b"".join(chunks)
+
+
+def _validate_minimax_h3_image_payload(
+    payload: bytes,
+    *,
+    filename: str | None,
+    allow_non_image: bool = False,
+) -> None:
+    """Validate a H3 image before converting it to a format-less PIL image."""
+    try:
+        with Image.open(io.BytesIO(payload)) as image:
+            image_format = str(image.format or "").lower()
+            if image_format in MINIMAX_H3_REFERENCE_IMAGE_FORMATS:
+                _validate_image_pixel_limit(image)
+    except InvalidInputReferenceError as exc:
+        raise HTTPException(HTTPStatus.BAD_REQUEST.value, detail=str(exc)) from exc
+    except Image.DecompressionBombError as exc:
+        raise HTTPException(
+            HTTPStatus.BAD_REQUEST.value,
+            detail=f"Invalid uploaded image reference: {filename}; image exceeds the decoder pixel limit.",
+        ) from exc
+    except (OSError, ValueError) as exc:
+        if allow_non_image:
+            return
+        raise HTTPException(400, detail=f"Invalid uploaded image reference: {filename}") from exc
+
+    if image_format not in MINIMAX_H3_REFERENCE_IMAGE_FORMATS:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=(
+                "MiniMax H3 reference images must use JPG, JPEG, PNG, WEBP, HEIC, or HEIF; "
+                f"got {image_format or 'unknown'}."
+            ),
+        )
+
+
+async def _persist_uploaded_media_references(
+    uploads: list[UploadFile],
+) -> tuple[list[Image.Image], list[str], list[str]]:
+    """Persist a mixed MiniMax H3 multipart reference list.
+
+    Images are decoded in memory; videos and audio remain files because H3's
+    reference encoders need the original container streams (including video
+    soundtracks).
+    """
+    images: list[Image.Image] = []
+    videos: list[str] = []
+    audios: list[str] = []
+    paths: list[str] = []
+    if len(uploads) > MINIMAX_H3_MAX_REFERENCE_COUNT:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=f"MiniMax H3 accepts at most {MINIMAX_H3_MAX_REFERENCE_COUNT} total references.",
+        )
+    try:
+        for upload in uploads:
+            kind = _uploaded_media_kind(upload)
+            payload = await _read_upload_limited(upload, max_bytes=_minimax_h3_upload_limit(upload))
+            if kind == "image":
+                try:
+                    _validate_minimax_h3_image_payload(payload, filename=upload.filename)
+                    images.append(
+                        _decode_image_bytes(
+                            payload,
+                            source=f"uploaded image reference: {upload.filename}",
+                        )
+                    )
+                except InvalidInputReferenceError as exc:
+                    raise HTTPException(HTTPStatus.BAD_REQUEST.value, detail=str(exc)) from exc
+                continue
+            suffix = Path(upload.filename or "").suffix.lower()
+            if kind == "video" and suffix and suffix not in MINIMAX_H3_REFERENCE_VIDEO_SUFFIXES:
+                raise HTTPException(
+                    status_code=HTTPStatus.BAD_REQUEST.value,
+                    detail="MiniMax H3 reference videos must use an MP4 or MOV file.",
+                )
+            if kind == "audio" and suffix and suffix not in MINIMAX_H3_REFERENCE_AUDIO_SUFFIXES:
+                raise HTTPException(
+                    status_code=HTTPStatus.BAD_REQUEST.value,
+                    detail="MiniMax H3 reference audio must use a WAV or MP3 file.",
+                )
+            if not suffix or len(suffix) > 8:
+                suffix = ".mp3" if kind == "audio" else ".mp4"
+            fd, path = tempfile.mkstemp(prefix="vllm_omni_reference_", suffix=suffix)
+            paths.append(path)
+            with os.fdopen(fd, "wb") as output:
+                output.write(payload)
+            if kind == "audio":
+                audios.append(path)
+            else:
+                videos.append(path)
+    except Exception:
+        for path in paths:
+            if os.path.exists(path):
+                os.unlink(path)
+        raise
+    return images, videos, audios
+
+
+async def _parse_video_form(
+    raw_request: Request,
+    prompt: str = Form(...),
+    input_reference: UploadFile | None = File(default=None),
+    input_references: list[UploadFile] | None = File(default=None),
+    control_reference: UploadFile | None = File(default=None),
+    control_type: str | None = Form(default=None),
+    image_reference: str | None = Form(default=None),
+    video_reference: str | None = Form(default=None),
+    audio_reference: str | None = Form(default=None),
+    model: str | None = Form(default=None),
+    seconds: SecondStr | None = Form(default=None),
+    size: SizeStr | None = Form(default=None),
+    user: str | None = Form(default=None),
+    width: int | None = Form(default=None),
+    height: int | None = Form(default=None),
+    num_frames: int | None = Form(default=None),
+    fps: float | None = Form(default=None, ge=1, allow_inf_nan=False),
+    aspect_ratio: str | None = Form(default=None),
+    short_edge: int | None = Form(default=None, ge=1),
+    num_outputs_per_prompt: int = Form(default=1, ge=1, le=10),
+    start_time_seconds: float | None = Form(default=None, ge=0.0),
+    quality: str | None = Form(default=None),
+    num_inference_steps: int | None = Form(default=None),
+    guidance_scale: float | None = Form(default=None),
+    guidance_scale_2: float | None = Form(default=None),
+    boundary_ratio: float | None = Form(default=None),
+    flow_shift: float | None = Form(default=None),
+    true_cfg_scale: float | None = Form(default=None),
+    seed: int | None = Form(default=None),
+    generate_sound: bool | None = Form(default=None),
+    sound_duration: float | None = Form(default=None, gt=0.0),
+    negative_prompt: str | None = Form(default=None),
+    enable_frame_interpolation: bool | None = Form(default=None),
+    frame_interpolation_exp: int | None = Form(default=None, ge=1),
+    frame_interpolation_scale: float | None = Form(default=None, gt=0.0),
+    frame_interpolation_model_path: str | None = Form(default=None),
+    lora: str | None = Form(default=None),
+    extra_params: str | None = Form(default=None),
+    return_stage_metrics: bool | None = Form(default=None),
+) -> tuple[
+    VideoGenerationRequest,
+    "OmniOpenAIServingVideo",
+    str,
+    ReferenceImage | None,
+    ReferenceVideo | None,
+    ReferenceAudio | None,
+    str | None,
+]:
+    """FastAPI dependency that parses video form data, validates inputs,
+    resolves the handler, and decodes any reference image.
+
+    Used by both ``POST /v1/videos`` (async) and ``POST /v1/videos/sync``.
+    """
+    input_references = input_references or []
+    input_reference_bytes: bytes | None = None
+    parsed_image_reference = _parse_form_json(image_reference)
+    parsed_video_reference = _parse_form_json(video_reference)
+    parsed_audio_reference = _parse_form_json(audio_reference)
+
+    if input_references and any(
+        item is not None for item in (parsed_image_reference, parsed_video_reference, input_reference)
+    ):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Provide input_references alone, without input_reference, image_reference, or video_reference.",
+        )
+    if input_reference is not None and (parsed_image_reference is not None or parsed_video_reference is not None):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail=(
+                "Provide only one of input_reference, image_reference, or video_reference when using "
+                "input_reference; image_reference and video_reference may be combined."
+            ),
+        )
+
+    request_data: dict[str, Any] = {
+        "prompt": prompt,
+        "model": model,
+        "seconds": seconds,
+        "size": size,
+        "image_reference": parsed_image_reference,
+        "video_reference": parsed_video_reference,
+        "audio_reference": parsed_audio_reference,
+        "user": user,
+        "width": width,
+        "height": height,
+        "num_frames": num_frames,
+        "fps": fps,
+        "aspect_ratio": aspect_ratio,
+        "short_edge": short_edge,
+        "num_outputs_per_prompt": num_outputs_per_prompt,
+        "start_time_seconds": start_time_seconds,
+        "quality": quality,
+        "num_inference_steps": num_inference_steps,
+        "guidance_scale": guidance_scale,
+        "guidance_scale_2": guidance_scale_2,
+        "boundary_ratio": boundary_ratio,
+        "flow_shift": flow_shift,
+        "true_cfg_scale": true_cfg_scale,
+        "seed": seed,
+        "generate_sound": generate_sound,
+        "sound_duration": sound_duration,
+        "negative_prompt": negative_prompt,
+        "enable_frame_interpolation": enable_frame_interpolation,
+        "frame_interpolation_exp": frame_interpolation_exp,
+        "frame_interpolation_scale": frame_interpolation_scale,
+        "frame_interpolation_model_path": frame_interpolation_model_path,
+        "lora": _parse_form_json(lora, expected_type=dict),
+        "extra_params": _parse_form_json(extra_params, expected_type=dict),
+        "return_stage_metrics": return_stage_metrics,
+    }
+    request_data = {k: v for k, v in request_data.items() if v is not None}
+    request = VideoGenerationRequest(**request_data)
+
+    handler = Omnivideo(raw_request)
+    if handler is None:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE.value,
+            detail="Video generation handler not initialized.",
+        )
+    logger.info("Video generation handler: %s", type(handler).__name__)
+    try:
+        app_model_name, app_stage_configs = _resolve_video_runtime_context(raw_request)
+        effective_model_name = handler.model_name or app_model_name or request.model or "unknown"
+        if request.model is not None and effective_model_name is not None and request.model != effective_model_name:
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST.value,
+                detail=(
+                    f"Model mismatch: request specifies '{request.model}' but server is running "
+                    f"'{effective_model_name}'."
+                ),
+            )
+        handler.set_stage_configs_if_missing(app_stage_configs)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception("Video generation setup failed: %s", e)
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+            detail=f"Video generation setup failed: {str(e)}",
+        )
+
+    normalized_control_type = _validate_control_upload(handler, request, control_reference, control_type)
+    if normalized_control_type is not None:
+        # Make the selected transfer mode visible while choosing the model's
+        # reference-video decode policy. The upload itself stays unpersisted
+        # until reference parsing succeeds, preserving failure cleanup.
+        _attach_control_upload(request, normalized_control_type)
+
+    supports_mixed_reference_inputs = bool(getattr(handler, "supports_mixed_reference_inputs", False))
+    if input_reference is not None:
+        input_reference_bytes = await _read_upload_limited(
+            input_reference,
+            max_bytes=_minimax_h3_upload_limit(input_reference) if supports_mixed_reference_inputs else None,
+        )
+        if supports_mixed_reference_inputs:
+            input_reference_kind = _uploaded_media_kind(input_reference)
+            _validate_minimax_h3_image_payload(
+                input_reference_bytes,
+                filename=input_reference.filename,
+                allow_non_image=input_reference_kind != "image",
+            )
+
+    decode_spec = ReferenceVideoDecodeSpec()
+    if not input_references and (parsed_video_reference is not None or input_reference_bytes is not None):
+        stage_configs = (
+            handler.stage_configs
+            or app_stage_configs
+            or getattr(getattr(handler, "_engine_client", None), "stage_configs", None)
+        )
+        decode_spec = _reference_video_decode_spec(request, stage_configs)
+    reference_image = None
+    reference_video = None
+    reference_audio: ReferenceAudio | None = None
+    if input_references:
+        if not supports_mixed_reference_inputs:
+            video_paths = await _persist_uploaded_video_references(input_references)
+            reference_video = ReferenceVideo(data=video_paths, cleanup_paths=tuple(video_paths))
+            images, audio_paths = [], []
+        else:
+            images, video_paths, audio_paths = await _persist_uploaded_media_references(input_references)
+        if images:
+            reference_image = ReferenceImage(data=images if len(images) > 1 else images[0])
+        if video_paths:
+            reference_video = ReferenceVideo(data=video_paths, cleanup_paths=tuple(video_paths))
+        if audio_paths:
+            reference_audio = ReferenceAudio(path=audio_paths, cleanup_paths=tuple(audio_paths))
+    else:
+        video_paths: list[str] = []
+        try:
+            image_items = _reference_list(request.image_reference)
+            video_items = _reference_list(request.video_reference)
+            image_data = []
+            for item in image_items:
+                media_data = await decode_input_reference(item, None, None)
+                if not isinstance(media_data, Image.Image):
+                    raise InvalidInputReferenceError("image_reference did not decode to an image")
+                image_data.append(media_data)
+
+            video_frames: list[Image.Image] | None = None
+            for item in video_items:
+                media_data = await decode_input_reference(
+                    None,
+                    item,
+                    None,
+                    max_video_frames=decode_spec.max_frames,
+                    video_keep=decode_spec.keep,
+                )
+                if not isinstance(media_data, VideoFrames):
+                    raise InvalidInputReferenceError("video_reference did not decode to a video")
+                if media_data.source_path is not None:
+                    video_paths.append(media_data.source_path)
+                else:
+                    if len(video_items) != 1:
+                        raise InvalidInputReferenceError(
+                            "multiple video URL references must be downloadable source videos"
+                        )
+                    video_frames = list(media_data)
+
+            if input_reference_bytes is not None:
+                media_data = await decode_input_reference(
+                    None,
+                    None,
+                    input_reference_bytes,
+                    max_video_frames=decode_spec.max_frames,
+                    video_keep=decode_spec.keep,
+                )
+                if isinstance(media_data, Image.Image):
+                    image_data.append(media_data)
+                elif isinstance(media_data, VideoFrames):
+                    if media_data.source_path is not None:
+                        video_paths.append(media_data.source_path)
+                    else:
+                        video_frames = list(media_data)
+
+            if image_data:
+                reference_image = ReferenceImage(data=image_data if len(image_data) > 1 else image_data[0])
+            if video_paths:
+                reference_video = ReferenceVideo(data=video_paths, cleanup_paths=tuple(video_paths))
+            elif video_frames is not None:
+                reference_video = ReferenceVideo(data=video_frames)
+        except InvalidInputReferenceError as exc:
+            for path in video_paths:
+                if os.path.exists(path):
+                    os.unlink(path)
+            raise HTTPException(400, detail=str(exc) or "Invalid input reference.") from exc
+
+    audio_paths = [] if reference_audio is None else list(_reference_list(reference_audio.path))
+    if request.audio_reference is not None:
+        try:
+            for audio_reference in _reference_list(request.audio_reference):
+                audio_paths.append(await decode_audio_url(audio_reference.audio_url))
+        except InvalidInputReferenceError as exc:
+            _cleanup_video_references(reference_video, reference_audio)
+            cleanup_paths = set(() if reference_audio is None else reference_audio.cleanup_paths)
+            for path in audio_paths:
+                if path not in cleanup_paths and os.path.exists(path):
+                    os.unlink(path)
+            raise HTTPException(400, detail=str(exc)) from exc
+    if audio_paths:
+        cleanup_paths = (
+            tuple(audio_paths)
+            if reference_audio is None
+            else reference_audio.cleanup_paths
+            + tuple(path for path in audio_paths if path not in reference_audio.cleanup_paths)
+        )
+        reference_audio = ReferenceAudio(
+            path=audio_paths if len(audio_paths) > 1 else audio_paths[0],
+            cleanup_paths=cleanup_paths,
+        )
+
+    control_path: str | None = None
+    if control_reference is not None and normalized_control_type is not None:
+        try:
+            control_path = await _persist_uploaded_control_reference(
+                control_reference,
+                max_bytes=CONTROL_REFERENCE_MAX_BYTES,
+            )
+            _attach_control_upload(request, normalized_control_type, control_path)
+        except (asyncio.CancelledError, HTTPException, OSError, TypeError, ValueError):
+            _cleanup_video_references(reference_video, reference_audio, control_path)
+            raise
+
+    return (
+        request,
+        handler,
+        effective_model_name,
+        reference_image,
+        reference_video,
+        reference_audio,
+        control_path,
+    )

vllm_omni/entrypoints/openai/video_api_utils.py

@@ -1,7 +1,26 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """
-Shared helper utilities for OpenAI-compatible video generation API.
+Shared media utilities for OpenAI-compatible video APIs.
+
+PUT HERE:
+  - Shared media backends reused by generation and streaming serving:
+    decode/encode of image/video/audio references and frames, streaming
+    encoders, frame/audio coercion. No FastAPI Request / job-store orchestration.
+
+DO NOT PUT HERE:
+  - ``/v1/videos*`` multipart form parsing, upload limits, job runners,
+    cleanup, or response factories — those go in ``video.generation.helpers``.
+
+LONGEVITY:
+  - This root utils file is a **temporary shared home**.
+  - TODO(#5227, P1.3): tidy up / move into the video family (e.g.
+    ``video/generation/media.py``) in the P1.3 video modality PR; do not treat this
+    file as the long-term owner.
+  - ``video.generation.helpers`` is the longer home for ``/v1/videos*``
+    endpoint logic through P0.2/P0.3 until P1.3 further splits it.
+
+See ``openai/README.md`` and ``video/README.md`` (utils vs helpers, no overlap).
 """
 
 from __future__ import annotations

vllm_omni/entrypoints/serve/README.md

@@ -0,0 +1,22 @@
+# Serve Entrypoint Helpers
+
+This package contains server-level helpers for assembling and operating the
+HTTP/FastAPI serving process.
+
+## What Belongs Here
+
+- Helpers that are about the server app rather than a public OpenAI endpoint
+  family.
+- Operational route packages such as profiling and Omni control.
+- Shared server utilities that mirror upstream `vllm.entrypoints.serve.utils`
+  concepts.
+
+## What Does Not Belong Here
+
+- OpenAI protocol schemas or response bodies.
+- Image, video, audio, or streaming-input request parsing.
+- Serving model implementations.
+
+Endpoint-family route bodies are still staged in `openai/api_server.py` during
+the P0.2 helper migration. Route ownership (P0.3) and final folder cleanup (Phase 1)
+are tracked by #5227.

vllm_omni/entrypoints/serve/omni_control/README.md

@@ -0,0 +1,19 @@
+# Omni Control Routes
+
+This package owns operational Omni server-control routes such as sleep and
+wakeup.
+
+## Put Here
+
+- Request/response models for `/v1/omni/*` operational controls.
+- Route bodies for Omni control endpoints after endpoint extraction.
+- Small helpers that are specific to these operational controls.
+
+## Do Not Put Here
+
+- OpenAI modality endpoints.
+- Generic server utilities that are shared outside Omni control.
+- Engine implementation details for sleep or wakeup.
+
+The current route bodies still live in `openai/api_server.py` and should move
+here as part of #5227.

vllm_omni/entrypoints/serve/omni_control/protocol.py

@@ -0,0 +1,14 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Request models for Omni server control routes."""
+
+from pydantic import BaseModel
+
+
+class OmniSleepRequest(BaseModel):
+    stage_ids: list[int]
+    level: int = 2
+
+
+class OmniWakeupRequest(BaseModel):
+    stage_ids: list[int]

vllm_omni/entrypoints/serve/profile/README.md

@@ -0,0 +1,19 @@
+# Profile Routes
+
+This package owns profiling-control schemas and, after route extraction, should
+own the profiling route bodies.
+
+## Put Here
+
+- Request/response models for `/start_profile` and `/stop_profile`.
+- Profile-specific helper functions.
+- The profile `api_router.py` once route bodies move out of `api_server.py`.
+
+## Do Not Put Here
+
+- General server error handling.
+- Engine profiling implementation internals.
+- OpenAI modality endpoint logic.
+
+The route bodies still live in `openai/api_server.py` under Phase 0.2 (`P0.2`) of #5227.
+TODO(#5227, P0.3): move route bodies here during router extraction.

vllm_omni/entrypoints/serve/profile/protocol.py

@@ -0,0 +1,14 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Request models for Omni profile control routes."""
+
+from pydantic import BaseModel, Field
+
+
+class ProfileRequest(BaseModel):
+    """Request model for profiling endpoints."""
+
+    stages: list[int] | None = Field(
+        default=None,
+        description="List of stage IDs to profile. If None, profiles all stages.",
+    )

vllm_omni/entrypoints/serve/profile/utils.py

@@ -0,0 +1,27 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Utilities for Omni profile control routes."""
+
+
+def _should_enable_profiler_endpoints(stage_configs: list | None) -> bool:
+    """Check if any stage has profiler_config set in its engine_args."""
+    if not stage_configs:
+        return False
+    for stage in stage_configs:
+        engine_args = stage.get("engine_args") if isinstance(stage, dict) else getattr(stage, "engine_args", None)
+        if engine_args is None:
+            continue
+        profiler_config = (
+            engine_args.get("profiler_config")
+            if isinstance(engine_args, dict)
+            else getattr(engine_args, "profiler_config", None)
+        )
+        if profiler_config is not None:
+            profiler = (
+                profiler_config.get("profiler")
+                if isinstance(profiler_config, dict)
+                else getattr(profiler_config, "profiler", None)
+            )
+            if profiler is not None:
+                return True
+    return False

vllm_omni/entrypoints/serve/utils/README.md

@@ -0,0 +1,23 @@
+# Serve Utilities
+
+This directory is for server/app utilities shared by Omni entrypoints.
+
+## Put Here
+
+- FastAPI app/router mutation helpers (`routes.remove_route_from_app`,
+  `routes._remove_route_from_router`).
+- Server exception-handler registration.
+- Engine-failure helpers that are not owned by one endpoint family.
+
+## Do Not Put Here
+
+- Endpoint restriction *policy* (which routes a pipeline shuts down, and why).
+  That stays in `vllm_omni.config.endpoint_policy`; the policy module should
+  *call* `remove_route_from_app` from here, not redefine it.
+- OpenAI response-shape adapters. Those belong under
+  `vllm_omni.entrypoints.openai`.
+- Modality validation or payload parsing.
+- Generic helpers without a clear server/app responsibility.
+
+Keep this directory aligned with the spirit of upstream
+`vllm.entrypoints.serve.utils`. Broader utility cleanup is tracked by #5227.

vllm_omni/entrypoints/serve/utils/errors.py

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Server-level error helpers for vLLM-Omni entrypoints.
+
+Use this module for FastAPI/app exception handling and engine-failure response
+construction that is not owned by a single OpenAI endpoint family."""
+
+from typing import Any
+
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from vllm.entrypoints.launchers.launcher import terminate_if_errored
+from vllm.entrypoints.serve import create_error_response
+from vllm.logger import init_logger
+from vllm.v1.engine.exceptions import EngineDeadError, EngineGenerateError
+
+logger = init_logger(__name__)
+
+
+def _get_request_id_from_request(req: Request) -> str | None:
+    return req.state.request_metadata.request_id if hasattr(req.state, "request_metadata") else None
+
+
+def _build_engine_error_payload(
+    exc: EngineDeadError | EngineGenerateError,
+    *,
+    request_id: str | None,
+) -> tuple[dict[str, Any], int]:
+    err = create_error_response(exc)
+    payload = err.model_dump()
+    error_body = payload.get("error", {})
+
+    error_body["request_id"] = request_id
+    error_body["error_stage_id"] = getattr(exc, "error_stage_id", None)
+
+    return payload, err.error.code
+
+
+def _create_engine_error_json_response(
+    req: Request,
+    exc: EngineDeadError | EngineGenerateError,
+) -> JSONResponse:
+    request_id = _get_request_id_from_request(req)
+    error_stage_id = getattr(exc, "error_stage_id", None)
+    engine = req.app.state.engine_client
+
+    if isinstance(exc, EngineDeadError):
+        # Log Omni-specific diagnostic information for dead engines.
+        orchestrator_alive = engine.engine.is_alive() if hasattr(engine, "engine") else "N/A"
+        logger.error(
+            "EngineDeadError: orchestrator_alive=%s, errored=%s, request_id=%s, error_stage_id=%s",
+            orchestrator_alive,
+            engine.errored,
+            request_id,
+            error_stage_id,
+        )
+
+    terminate_if_errored(
+        server=req.app.state.server,
+        engine=engine,
+    )
+
+    payload, status_code = _build_engine_error_payload(exc, request_id=request_id)
+    return JSONResponse(content=payload, status_code=status_code)
+
+
+def _register_omni_exception_handlers(app) -> None:
+    """Override upstream vLLM exception handlers with Omni-aware versions.
+
+    The upstream ``engine_error_handler`` is designed for ``AsyncLLM`` (single
+    EngineCore process).  Omni uses a multi-stage orchestrator with different
+    health semantics, so we register our own handlers that:
+
+    - Log multi-stage diagnostic info (orchestrator liveness, per-stage health)
+      when an ``EngineDeadError`` is caught.
+    - Call ``terminate_if_errored``
+    - Return an OpenAI-compatible error JSON response.
+    """
+
+    async def omni_engine_error_handler(
+        req: Request,
+        exc: EngineDeadError | EngineGenerateError,
+    ):
+        request_id = _get_request_id_from_request(req)
+
+        if req.app.state.args.log_error_stack:
+            logger.exception("Engine Exception caught. Request id: %s", request_id)
+
+        return _create_engine_error_json_response(req, exc)
+
+    app.exception_handler(EngineGenerateError)(omni_engine_error_handler)
+    app.exception_handler(EngineDeadError)(omni_engine_error_handler)

vllm_omni/entrypoints/serve/utils/routes.py

@@ -0,0 +1,45 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Route-table helpers for Omni server assembly.
+
+Owns FastAPI app **and** router mutation used during server construction.
+``endpoint_policy`` imports app-level removal from here; it does not own
+route-table helpers itself.
+"""
+
+from collections.abc import Set
+
+from fastapi import APIRouter, FastAPI
+from starlette.routing import Route
+
+
+def remove_route_from_app(
+    app: FastAPI,
+    path: str,
+    methods: Set[str],
+) -> None:
+    """Remove app routes matching a path and one of the given HTTP methods."""
+    routes_to_remove = [
+        route
+        for route in app.routes
+        if isinstance(route, Route) and route.path == path and route.methods is not None and route.methods & methods
+    ]
+    for route in routes_to_remove:
+        app.routes.remove(route)
+
+
+def _remove_route_from_router(
+    router: APIRouter,
+    path: str,
+    methods: set[str] | None = None,
+) -> None:
+    """Remove routes from an ``APIRouter`` by path and optionally by methods."""
+    methods_set = {method.upper() for method in methods} if methods else None
+    for route in list(router.routes):
+        if getattr(route, "path", None) != path:
+            continue
+        if methods_set is not None:
+            route_methods = {method.upper() for method in (getattr(route, "methods", None) or set())}
+            if not (route_methods & methods_set):
+                continue
+        router.routes.remove(route)