diff --git a/tests/model_executor/models/registry.py b/tests/model_executor/models/registry.py
new file mode 100644
index 000000000..98f9aefb5
--- /dev/null
+++ b/tests/model_executor/models/registry.py
@@ -0,0 +1,179 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""
+Test metadata for Omni models.
+
+Mirrors vllm's ``tests/models/registry.py`` (_HfExamplesInfo) but adapted for
+the Omni multi-stage architecture.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+
+import pytest
+from packaging.version import Version
+from transformers import __version__ as TRANSFORMERS_VERSION
+
+
+@dataclass(frozen=True)
+class _OmniExamplesInfo:
+    """Metadata used by processing-correctness tests for Omni models."""
+
+    default: str
+    """The default HF model ID to use for testing this architecture."""
+
+    model_stage: str = "thinker"
+    """Stage type identifier, e.g. ``"thinker"`` or ``"talker"``."""
+
+    hf_config_name: str | None = None
+    """
+    Name of the sub-config inside the top-level HF config
+    (e.g. ``"thinker_config"``).  ``None`` means use the root config.
+    """
+
+    has_multimodal_processor: bool = False
+    """Whether the architecture registers a multimodal processor."""
+
+    min_transformers_version: str | None = None
+    """Minimum ``transformers`` version required to load this model."""
+
+    max_transformers_version: str | None = None
+    """Maximum ``transformers`` version that this model supports."""
+
+    is_available_online: bool = True
+    """Set to ``False`` if the model is no longer hosted on HF Hub."""
+
+    trust_remote_code: bool = False
+    """Whether ``trust_remote_code=True`` is needed to load the model."""
+
+    max_model_len: int | None = None
+    """
+    An explicit ``max_model_len`` override.  Useful when the default value
+    from the HF config is too large for CI.
+    """
+
+    # ----- helper methods (same API as _HfExamplesInfo) --------------------
+
+    def check_transformers_version(
+        self,
+        *,
+        on_fail: Literal["error", "skip", "return"],
+    ) -> str | None:
+        """Skip / error / return a message when the installed
+        ``transformers`` does not satisfy the version constraints."""
+        if self.min_transformers_version is None and self.max_transformers_version is None:
+            return None
+
+        current = TRANSFORMERS_VERSION
+        cur_base = Version(current).base_version
+        min_ver = self.min_transformers_version
+        max_ver = self.max_transformers_version
+
+        msg = f"`transformers=={current}` installed, but `transformers"
+        if min_ver and Version(cur_base) < Version(min_ver):
+            msg += f">={min_ver}` is required to run this model."
+        elif max_ver and Version(cur_base) > Version(max_ver):
+            msg += f"<={max_ver}` is required to run this model."
+        else:
+            return None
+
+        if on_fail == "error":
+            raise RuntimeError(msg)
+        elif on_fail == "skip":
+            pytest.skip(msg)
+        return msg
+
+    def check_available_online(
+        self,
+        *,
+        on_fail: Literal["error", "skip"],
+    ) -> None:
+        """Skip / error when the model is not available online."""
+        if not self.is_available_online:
+            msg = "Model is not available online"
+            if on_fail == "error":
+                raise RuntimeError(msg)
+            else:
+                pytest.skip(msg)
+
+
+# ---------------------------------------------------------------------------
+# Example model registry for tests
+# ---------------------------------------------------------------------------
+
+_OMNI_EXAMPLE_MODELS: dict[str, _OmniExamplesInfo] = {
+    # ---- Qwen2.5-Omni ----
+    "Qwen2_5OmniForConditionalGeneration": _OmniExamplesInfo(
+        default="Qwen/Qwen2.5-Omni-7B",
+        model_stage="thinker",
+        has_multimodal_processor=True,
+    ),
+    "Qwen2_5OmniThinkerModel": _OmniExamplesInfo(
+        default="Qwen/Qwen2.5-Omni-7B",
+        model_stage="thinker",
+        hf_config_name="thinker_config",
+        has_multimodal_processor=True,
+    ),
+    "Qwen2_5OmniTalkerModel": _OmniExamplesInfo(
+        default="Qwen/Qwen2.5-Omni-7B",
+        model_stage="talker",
+        hf_config_name="talker_config",
+        has_multimodal_processor=True,
+    ),
+    "Qwen2_5OmniToken2WavModel": _OmniExamplesInfo(
+        default="Qwen/Qwen2.5-Omni-7B",
+        model_stage="token2wav",
+    ),
+    "Qwen2_5OmniToken2WavDiTModel": _OmniExamplesInfo(
+        default="Qwen/Qwen2.5-Omni-7B",
+        model_stage="token2wav",
+    ),
+    # Internal sub-model key hardcoded by Qwen2_5OmniTalkerForConditionalGeneration
+    # to load its language model component (see qwen2_5_omni_talker.py).
+    "Qwen2ForCausalLM_old": _OmniExamplesInfo(
+        default="Qwen/Qwen2.5-Omni-7B",
+        model_stage="thinker",
+    ),
+    # ---- Qwen3-Omni MoE  ----
+    "Qwen3OmniMoeForConditionalGeneration": _OmniExamplesInfo(
+        default="Qwen/Qwen3-Omni-30B-A3B-Instruct",
+        model_stage="thinker",
+        has_multimodal_processor=True,
+    ),
+    "Qwen3OmniMoeThinkerForConditionalGeneration": _OmniExamplesInfo(
+        default="Qwen/Qwen3-Omni-30B-A3B-Instruct",
+        model_stage="thinker",
+        hf_config_name="thinker_config",
+        has_multimodal_processor=True,
+    ),
+    "Qwen3OmniMoeTalkerForConditionalGeneration": _OmniExamplesInfo(
+        default="Qwen/Qwen3-Omni-30B-A3B-Instruct",
+        model_stage="talker",
+        hf_config_name="talker_config",
+    ),
+    "Qwen3OmniMoeCode2Wav": _OmniExamplesInfo(
+        default="Qwen/Qwen3-Omni-30B-A3B-Instruct",
+        model_stage="code2wav",
+    ),
+    # ---- Qwen3-TTS ----
+    "Qwen3TTSForConditionalGeneration": _OmniExamplesInfo(
+        default="Qwen/Qwen3-TTS",
+        model_stage="talker",
+    ),
+    "Qwen3TTSTalkerForConditionalGeneration": _OmniExamplesInfo(
+        default="Qwen/Qwen3-TTS",
+        model_stage="talker",
+    ),
+    "Qwen3TTSCode2Wav": _OmniExamplesInfo(
+        default="Qwen/Qwen3-TTS",
+        model_stage="code2wav",
+    ),
+}
+
+# Only architectures that register a multimodal processor are relevant for
+# processing-correctness tests.
+_MULTIMODAL_OMNI_EXAMPLE_MODELS: dict[str, _OmniExamplesInfo] = {
+    arch: info for arch, info in _OMNI_EXAMPLE_MODELS.items() if info.has_multimodal_processor
+}
diff --git a/tests/model_executor/models/test_omni_processing.py b/tests/model_executor/models/test_omni_processing.py
new file mode 100644
index 000000000..fe9c63b82
--- /dev/null
+++ b/tests/model_executor/models/test_omni_processing.py
@@ -0,0 +1,327 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""
+Test multimodal processing correctness for Omni models.
+
+Tests that:
+1. Cached vs non-cached processor outputs are identical.
+2. Text prompt vs token prompt inputs produce identical results after processing.
+
+Adapted from vllm/tests/models/multimodal/processing/test_common.py
+"""
+
+from functools import partial
+
+import numpy as np
+import pytest
+from PIL import Image
+from vllm.config.multimodal import (
+    AudioDummyOptions,
+    BaseDummyOptions,
+    ImageDummyOptions,
+    VideoDummyOptions,
+)
+from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalDataDict
+from vllm.multimodal.cache import MultiModalProcessorOnlyCache
+from vllm.multimodal.inputs import MultiModalInputs, batched_tensors_equal
+from vllm.multimodal.processing import BaseMultiModalProcessor, InputProcessingContext
+from vllm.tokenizers import TokenizerLike, cached_tokenizer_from_config
+
+from tests.model_executor.models.registry import (
+    _MULTIMODAL_OMNI_EXAMPLE_MODELS,
+    _OmniExamplesInfo,
+)
+from vllm_omni.config import OmniModelConfig
+
+
+def random_image(rng: np.random.RandomState, min_wh: int, max_wh: int):
+    w, h = rng.randint(min_wh, max_wh, size=(2,))
+    arr = rng.randint(0, 255, size=(w, h, 3), dtype=np.uint8)
+    return Image.fromarray(arr)
+
+
+def random_video(
+    rng: np.random.RandomState,
+    min_frames: int,
+    max_frames: int,
+    min_wh: int,
+    max_wh: int,
+):
+    num_frames = rng.randint(min_frames, max_frames)
+    w, h = rng.randint(min_wh, max_wh, size=(2,))
+    return rng.randint(0, 255, size=(num_frames, w, h, 3), dtype=np.uint8)
+
+
+def random_audio(
+    rng: np.random.RandomState,
+    min_len: int,
+    max_len: int,
+    sr: int,
+):
+    audio_len = rng.randint(min_len, max_len)
+    return rng.rand(audio_len), sr
+
+
+_IGNORE_MM_KEYS: dict[str, set[str]] = {}
+
+
+def _get_model_archs_to_test() -> list[str]:
+    """Return architecture strings for parametrization."""
+    return list(_MULTIMODAL_OMNI_EXAMPLE_MODELS.keys())
+
+
+def get_text_token_prompts(
+    processor: BaseMultiModalProcessor,
+    mm_data: MultiModalDataDict,
+):
+    """Return ``(text_prompt, token_prompt)`` for the given multimodal data."""
+    dummy_inputs = processor.dummy_inputs
+    tokenizer: TokenizerLike = processor.info.get_tokenizer()
+    model_config = processor.info.ctx.model_config
+
+    parsed_data = processor.info.parse_mm_data(mm_data)
+    mm_counts = {k: len(vs) for k, vs in parsed_data.items()}
+
+    inputs = dummy_inputs.get_dummy_processor_inputs(
+        model_config.max_model_len,
+        mm_counts,
+        mm_options={},
+    )
+
+    text_prompt: str | None
+    token_prompt: list[int]
+    if isinstance(inputs.prompt, list):
+        text_prompt = None
+        token_prompt = inputs.prompt
+    elif isinstance(inputs.prompt, str):
+        text_prompt = inputs.prompt
+        token_prompt = tokenizer.encode(
+            text_prompt,
+            **processor.info.get_default_tok_params().get_encode_kwargs(),
+        )
+    else:
+        raise TypeError(type(inputs.prompt))
+
+    return text_prompt, token_prompt
+
+
+def _build_model_config(model_arch: str, info: _OmniExamplesInfo) -> OmniModelConfig:
+    """Create an ``OmniModelConfig`` suitable for processor testing."""
+    kwargs: dict = dict(
+        model=info.default,
+        tokenizer=info.default,
+        tokenizer_mode="auto",
+        trust_remote_code=info.trust_remote_code,
+        model_arch=model_arch,
+        model_stage=info.model_stage,
+        max_model_len=info.max_model_len,
+        enforce_eager=True,
+        dtype="auto",
+    )
+    if info.hf_config_name is not None:
+        kwargs["hf_config_name"] = info.hf_config_name
+
+    model_config = OmniModelConfig(**kwargs)
+    # Ensure cache is effectively unlimited to prevent eviction during test
+    model_config.multimodal_config.mm_processor_cache_gb = 2048
+    return model_config
+
+
+def _test_processing_correctness(
+    model_arch: str,
+    hit_rate: float,
+    num_batches: int,
+    simplify_rate: float,
+):
+    """Run the full processing-correctness test for one Omni model."""
+    info = _MULTIMODAL_OMNI_EXAMPLE_MODELS[model_arch]
+    info.check_transformers_version(on_fail="skip")
+    info.check_available_online(on_fail="skip")
+
+    model_config = _build_model_config(model_arch, info)
+
+    model_cls = MULTIMODAL_REGISTRY._get_model_cls(model_config)
+    factories = model_cls._processor_factory
+    ctx = InputProcessingContext(
+        model_config,
+        tokenizer=cached_tokenizer_from_config(model_config),
+    )
+    cache = MultiModalProcessorOnlyCache(model_config)
+
+    processing_info = factories.info(ctx)
+    supported_mm_limits = processing_info.get_supported_mm_limits()
+    limit_mm_per_prompt_ints = {
+        modality: 3 if limit is None else limit for modality, limit in supported_mm_limits.items()
+    }
+
+    def _to_dummy_options(modality: str, count: int) -> BaseDummyOptions:
+        if modality == "video":
+            return VideoDummyOptions(count=count)
+        if modality == "image":
+            return ImageDummyOptions(count=count)
+        if modality == "audio":
+            return AudioDummyOptions(count=count)
+        return BaseDummyOptions(count=count)
+
+    model_config.get_multimodal_config().limit_per_prompt = {
+        modality: _to_dummy_options(modality, count) for modality, count in limit_mm_per_prompt_ints.items()
+    }
+
+    baseline_processor = factories.build_processor(ctx, cache=None)
+    cached_processor = factories.build_processor(ctx, cache=cache)
+
+    rng = np.random.RandomState(0)
+
+    input_to_hit = {
+        "image": Image.new("RGB", size=(128, 128)),
+        "video": np.zeros((4, 128, 128, 3), dtype=np.uint8),
+        "audio": (np.zeros((512,)), 16000),
+    }
+    input_factory = {
+        "image": partial(random_image, rng, min_wh=128, max_wh=256),
+        "video": partial(random_video, rng, min_frames=2, max_frames=16, min_wh=128, max_wh=256),
+        "audio": partial(random_audio, rng, min_len=512, max_len=1024, sr=16000),
+    }
+
+    for batch_idx in range(num_batches):
+        mm_data = {
+            k: [
+                (input_to_hit[k] if rng.rand() < hit_rate else input_factory[k]())
+                for _ in range(rng.randint(limit + 1))
+            ]
+            for k, limit in limit_mm_per_prompt_ints.items()
+        }
+
+        # Drop unnecessary keys and test single -> multi conversion
+        if rng.rand() < simplify_rate:
+            for k in list(mm_data.keys()):
+                if not mm_data[k]:
+                    del mm_data[k]
+                elif len(mm_data[k]) == 1:
+                    mm_data[k] = mm_data[k][0]
+
+        _test_processing_correctness_one(
+            model_config,
+            mm_data,
+            baseline_processor,
+            cached_processor,
+            batch_idx,
+        )
+
+
+def _test_processing_correctness_one(
+    model_config: OmniModelConfig,
+    mm_data: MultiModalDataDict,
+    baseline_processor: BaseMultiModalProcessor,
+    cached_processor: BaseMultiModalProcessor,
+    batch_idx: int,
+):
+    model_type = model_config.hf_config.model_type
+
+    text_prompt, token_prompt = get_text_token_prompts(baseline_processor, mm_data)
+    mm_items = baseline_processor.info.parse_mm_data(mm_data)
+    ignore_mm_keys = _IGNORE_MM_KEYS.get(model_type, set[str]())
+
+    baseline_tokenized_result = baseline_processor(
+        token_prompt,
+        mm_items=mm_items,
+        hf_processor_mm_kwargs={},
+    )
+
+    cached_tokenized_result = cached_processor(
+        token_prompt,
+        mm_items=mm_items,
+        hf_processor_mm_kwargs={},
+    )
+
+    _assert_inputs_equal(
+        baseline_tokenized_result,
+        cached_tokenized_result,
+        ignore_mm_keys=ignore_mm_keys,
+        msg=f"Failed ({batch_idx=}, {token_prompt=}, {mm_data=})",
+    )
+
+    if text_prompt is not None:
+        baseline_text_result = baseline_processor(
+            text_prompt,
+            mm_items=mm_items,
+            hf_processor_mm_kwargs={},
+        )
+        cached_text_result = cached_processor(
+            text_prompt,
+            mm_items=mm_items,
+            hf_processor_mm_kwargs={},
+        )
+
+        _assert_inputs_equal(
+            baseline_text_result,
+            cached_text_result,
+            ignore_mm_keys=ignore_mm_keys,
+            msg=f"Failed ({batch_idx=}, {text_prompt=}, {mm_data=})",
+        )
+
+        _assert_inputs_equal(
+            baseline_text_result,
+            baseline_tokenized_result,
+            ignore_mm_keys=ignore_mm_keys,
+            msg=f"Failed ({batch_idx=}, {text_prompt=}, {token_prompt=}, {mm_data=})",
+        )
+
+        _assert_inputs_equal(
+            cached_text_result,
+            cached_tokenized_result,
+            ignore_mm_keys=ignore_mm_keys,
+            msg=f"Failed ({batch_idx=}, {text_prompt=}, {token_prompt=}, {mm_data=})",
+        )
+
+
+@pytest.mark.core_model
+@pytest.mark.omni
+@pytest.mark.cpu
+@pytest.mark.parametrize("model_arch", _get_model_archs_to_test())
+@pytest.mark.parametrize("hit_rate", [0.3, 0.5, 1.0])
+@pytest.mark.parametrize("num_batches", [32])
+@pytest.mark.parametrize("simplify_rate", [1.0])
+def test_omni_processing_correctness(
+    model_arch: str,
+    hit_rate: float,
+    num_batches: int,
+    simplify_rate: float,
+):
+    """
+    For each Omni model verify that:
+    - cached and non-cached processor outputs are identical.
+    - text-prompt and token-prompt inputs yield the same result.
+    """
+    _test_processing_correctness(
+        model_arch,
+        hit_rate=hit_rate,
+        num_batches=num_batches,
+        simplify_rate=simplify_rate,
+    )
+
+
+def _assert_inputs_equal(
+    a: MultiModalInputs,
+    b: MultiModalInputs,
+    *,
+    ignore_mm_keys: set[str] | None = None,
+    msg: str = "",
+):
+    if ignore_mm_keys is None:
+        ignore_mm_keys = set()
+
+    ignore_prompt_keys = ("prompt", "mm_kwargs")
+    a_rest = {k: v for k, v in a.items() if k not in ignore_prompt_keys}
+    b_rest = {k: v for k, v in b.items() if k not in ignore_prompt_keys}
+
+    assert a_rest == b_rest, msg
+
+    a_data = a["mm_kwargs"].get_data()
+    b_data = b["mm_kwargs"].get_data()
+
+    for key in ignore_mm_keys:
+        a_data.pop(key, None)
+        b_data.pop(key, None)
+
+    assert batched_tensors_equal(a_data, b_data), msg