diff --git a/docs/user_guide/diffusion/quantization/autoround.md b/docs/user_guide/diffusion/quantization/autoround.md
index 48df176b0..d06627d40 100644
--- a/docs/user_guide/diffusion/quantization/autoround.md
+++ b/docs/user_guide/diffusion/quantization/autoround.md
@@ -72,6 +72,8 @@ At load time:
 | Model | HF Checkpoint | Scheme | Group Size | Backend |
 |-------|--------------|--------|------------|---------|
 | FLUX.1-dev | `vllm-project-org/FLUX.1-dev-AutoRound-w4a16` | W4A16 | 128 | GPTQ-Marlin |
+| Qwen2.5-Omni-7B | `Intel/Qwen2.5-Omni-7B-int4-AutoRound` | W4A16 | 128 | GPTQ-Marlin |
+| Qwen3-Omni-30B-A3B-Instruct | `Intel/Qwen3-Omni-30B-A3B-Instruct-int4-AutoRound` | W4A16 | 128 | GPTQ-Marlin |
 
 ## Creating a Quantized Checkpoint
 
diff --git a/examples/offline_inference/qwen2_5_omni/end2end.py b/examples/offline_inference/qwen2_5_omni/end2end.py
index dfe124700..a65c554a9 100644
--- a/examples/offline_inference/qwen2_5_omni/end2end.py
+++ b/examples/offline_inference/qwen2_5_omni/end2end.py
@@ -5,6 +5,7 @@ This example shows how to use vLLM-Omni for running offline inference
 with the correct prompt format on Qwen2.5-Omni
 """
 
+import json
 import os
 import time
 from typing import NamedTuple
@@ -289,7 +290,10 @@ query_map = {
 
 
 def main(args):
-    model_name = "Qwen/Qwen2.5-Omni-7B"
+    model_name = args.model
+    quantization_config = None
+    if args.quantization_config is not None:
+        quantization_config = json.loads(args.quantization_config)
 
     # Get paths from args
     video_path = getattr(args, "video_path", None)
@@ -320,6 +324,7 @@ def main(args):
         query_result = query_func(audio_path=audio_path, sampling_rate=sampling_rate)
     else:
         query_result = query_func()
+    args.quantization_config = quantization_config
     omni = Omni.from_cli_args(args, model=model_name)
     thinker_sampling_params = SamplingParams(
         temperature=0.0,  # Deterministic - no randomness
@@ -424,6 +429,18 @@ def main(args):
 
 def parse_args():
     parser = FlexibleArgumentParser(description="Demo on using vLLM for offline inference with audio language models")
+    parser.add_argument(
+        "--model",
+        type=str,
+        default="Qwen/Qwen2.5-Omni-7B",
+        help="Model name or local path.",
+    )
+    parser.add_argument(
+        "--quantization-config",
+        type=str,
+        default=None,
+        help="Optional JSON string forwarded to Omni(quantization_config=...).",
+    )
     parser.add_argument(
         "--query-type",
         "-q",
diff --git a/tests/diffusion/quantization/test_component_routing.py b/tests/diffusion/quantization/test_component_routing.py
new file mode 100644
index 000000000..c8b3837e2
--- /dev/null
+++ b/tests/diffusion/quantization/test_component_routing.py
@@ -0,0 +1,408 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Tests for component routing for quantization."""
+
+from unittest.mock import MagicMock
+
+import pytest
+import torch
+from vllm.model_executor.layers.quantization.base_config import (
+    QuantizationConfig,
+)
+from vllm.model_executor.models.utils import WeightsMapper
+
+from vllm_omni.model_executor.models.qwen3_omni.qwen3_omni_moe_thinker import (
+    PRE_QUANTIZED_METHODS,
+)
+from vllm_omni.quantization.component_config import (
+    ComponentQuantizationConfig,
+)
+from vllm_omni.quantization.inc_config import OmniINCConfig
+
+pytestmark = [pytest.mark.core_model]
+
+
+# ---------------------------------------------------------------------------
+# Helpers: lightweight mock quant configs
+# ---------------------------------------------------------------------------
+
+
+class _MockQuantConfig(QuantizationConfig):
+    """Minimal mock that only implements get_name()."""
+
+    def __init__(self, name: str, **attrs):
+        self._name = name
+        for k, v in attrs.items():
+            setattr(self, k, v)
+
+    def get_name(self) -> str:
+        return self._name
+
+    def get_quant_method(self, layer, prefix):
+        return MagicMock()
+
+    @classmethod
+    def get_supported_act_dtypes(cls):
+        return [torch.bfloat16, torch.float16]
+
+    def get_min_capability(self):
+        return 0
+
+    @classmethod
+    def from_config(cls, config):
+        raise NotImplementedError
+
+    def get_config_filenames(self):
+        return []
+
+
+def _make_inc_config(block_names="thinker.model.layers,talker.model.layers", extra_config=None):
+    """Create a real OmniINCConfig with block_name_to_quantize."""
+    return OmniINCConfig(
+        weight_bits=4,
+        group_size=128,
+        sym=True,
+        block_name_to_quantize=block_names,
+        extra_config=extra_config or {},
+    )
+
+
+THINKER_MAPPER = WeightsMapper(
+    orig_to_new_prefix={
+        "thinker.lm_head.": "language_model.lm_head.",
+        "thinker.model.": "language_model.model.",
+        "thinker.": "",
+    }
+)
+
+TALKER_MAPPER = WeightsMapper(
+    orig_to_new_prefix={
+        "talker.codec_head.": "language_model.lm_head.",
+        "talker.model.": "language_model.model.",
+        "talker.thinker_to_talker_proj.": "thinker_to_talker_proj.",
+        "talker.": "",
+    }
+)
+
+
+# ===================================================================
+# 1. OmniINCConfig.apply_vllm_mapper
+# ===================================================================
+
+
+class TestApplyVllmMapper:
+    def test_inc_csv_string_normalized_to_list(self):
+        """CSV string block_name_to_quantize is split into a list."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        assert isinstance(cfg.block_name_to_quantize, list)
+
+    def test_thinker_blocks_remapped(self):
+        """thinker.model.layers -> language_model.model.layers after apply_vllm_mapper."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        assert any("language_model.model.layers" in b for b in cfg.block_name_to_quantize)
+
+    def test_cross_stage_blocks_kept_unchanged(self):
+        """Blocks not matching any mapper prefix are kept unchanged (harmless)."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        # talker.model.layers doesn't match any thinker mapper prefix → stays as-is
+        assert "talker.model.layers" in cfg.block_name_to_quantize
+
+    def test_talker_remap(self):
+        """talker.model.layers -> language_model.model.layers with talker mapper."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(TALKER_MAPPER)
+        assert any("language_model.model.layers" in b for b in cfg.block_name_to_quantize)
+        # thinker.model.layers doesn't match talker mapper → stays as-is
+        assert "thinker.model.layers" in cfg.block_name_to_quantize
+
+    def test_extra_config_keys_remapped(self):
+        """Regex keys in extra_config get their escaped-dot prefixes remapped."""
+        extra = {
+            r".*thinker\.model\.layers\.0\.mlp\.gate.*": {"bits": 16, "data_type": "float"},
+        }
+        cfg = _make_inc_config("thinker.model.layers", extra_config=extra)
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        # The key should now reference the vLLM runtime path
+        assert any("language_model" in k for k in cfg.extra_config)
+        # Original thinker\.model prefix should be replaced
+        assert not any(r"thinker\.model" in k for k in cfg.extra_config)
+
+    def test_single_block_name(self):
+        """Only one block name (not CSV) still works."""
+        cfg = _make_inc_config("thinker.model.layers")
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        assert any("language_model.model.layers" in b for b in cfg.block_name_to_quantize)
+
+    def test_already_list_block_names(self):
+        """block_name_to_quantize already a list (not CSV string) works."""
+        cfg = _make_inc_config(["thinker.model.layers", "talker.model.layers"])
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        assert isinstance(cfg.block_name_to_quantize, list)
+        assert any("language_model.model.layers" in b for b in cfg.block_name_to_quantize)
+
+    def test_mutates_in_place(self):
+        """apply_vllm_mapper mutates the config in place (same as upstream INCConfig)."""
+        cfg = _make_inc_config("thinker.model.layers")
+        original_id = id(cfg)
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        assert id(cfg) == original_id
+
+    # -- Stage prefix tests (runtime prefix = container + internal name) --
+
+    def test_thinker_block_has_stage_prefix(self):
+        """Mapped block name must start with 'thinker.' so runtime startswith() works."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        assert "thinker.language_model.model.layers" in cfg.block_name_to_quantize
+
+    def test_talker_block_has_stage_prefix(self):
+        """Mapped block name must start with 'talker.' so runtime startswith() works."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(TALKER_MAPPER)
+        assert "talker.language_model.model.layers" in cfg.block_name_to_quantize
+
+    def test_thinker_block_matches_runtime_prefix(self):
+        """Simulates get_layer_config's startswith() check for FusedMoE layers."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        runtime_prefix = "thinker.language_model.model.layers.0.mlp.experts"
+        assert any(runtime_prefix.startswith(b) for b in cfg.block_name_to_quantize)
+
+    def test_talker_block_matches_runtime_prefix(self):
+        """Simulates get_layer_config's startswith() check for talker FusedMoE."""
+        cfg = _make_inc_config("thinker.model.layers,talker.model.layers")
+        cfg.apply_vllm_mapper(TALKER_MAPPER)
+        runtime_prefix = "talker.language_model.model.layers.0.mlp.experts"
+        assert any(runtime_prefix.startswith(b) for b in cfg.block_name_to_quantize)
+
+    def test_extra_config_plain_key_has_stage_prefix(self):
+        """Plain extra_config keys are remapped with stage prefix."""
+        extra = {
+            "talker.model.layers.0.mlp.shared_expert_gate": {"bits": 16},
+        }
+        cfg = _make_inc_config("talker.model.layers", extra_config=extra)
+        cfg.apply_vllm_mapper(TALKER_MAPPER)
+        assert "talker.language_model.model.layers.0.mlp.shared_expert_gate" in cfg.extra_config
+
+    def test_extra_config_regex_key_still_works(self):
+        """Regex extra_config keys use re.search so no stage prefix needed."""
+        import re
+
+        extra = {
+            r".*thinker\.model\.layers\.0\.mlp\.gate.*": {"bits": 16},
+        }
+        cfg = _make_inc_config("thinker.model.layers", extra_config=extra)
+        cfg.apply_vllm_mapper(THINKER_MAPPER)
+        runtime_name = "thinker.language_model.model.layers.0.mlp.gate"
+        matched = any(re.search(k, runtime_name) for k in cfg.extra_config)
+        assert matched
+
+
+# ===================================================================
+# 2. OmniINCConfig upgrade helpers
+# ===================================================================
+
+
+class TestOmniINCConfigUpgrade:
+    def test_maybe_upgrade_none(self):
+        assert OmniINCConfig.maybe_upgrade(None) is None
+
+    def test_maybe_upgrade_non_inc(self):
+        """Non-INC configs are passed through unchanged."""
+        cfg = _MockQuantConfig("fp8")
+        assert OmniINCConfig.maybe_upgrade(cfg) is cfg
+
+    def test_maybe_upgrade_already_omni(self):
+        """Already OmniINCConfig is returned as-is."""
+        cfg = _make_inc_config()
+        assert OmniINCConfig.maybe_upgrade(cfg) is cfg
+
+    def test_maybe_upgrade_vanilla_inc(self):
+        """Vanilla INCConfig is promoted to OmniINCConfig."""
+        from vllm.model_executor.layers.quantization.inc import INCConfig
+
+        vanilla = INCConfig(weight_bits=4, group_size=128, sym=True)
+        upgraded = OmniINCConfig.maybe_upgrade(vanilla)
+        assert isinstance(upgraded, OmniINCConfig)
+        assert upgraded.weight_bits == 4
+        assert upgraded.group_size == 128
+
+
+# ===================================================================
+# 2. Three-branch thinker routing (simulated)
+# ===================================================================
+
+
+def _simulate_thinker_routing(quant_config):
+    """Simulate the three-branch routing in thinker __init__.
+
+    Returns (visual_quant_config, language_quant_config, wrapped_vllm_quant).
+    """
+    if isinstance(quant_config, ComponentQuantizationConfig):
+        visual_quant_config = quant_config.resolve("visual")
+        language_quant_config = quant_config.resolve("language_model")
+        return visual_quant_config, language_quant_config, quant_config
+    elif quant_config is not None:
+        if quant_config.get_name() in PRE_QUANTIZED_METHODS:
+            return quant_config, quant_config, quant_config
+        else:
+            language_quant_config = quant_config
+            wrapped = ComponentQuantizationConfig(
+                component_configs={"language_model": quant_config},
+                default_config=None,
+            )
+            return None, language_quant_config, wrapped
+    else:
+        return None, None, None
+
+
+class TestThinkerRouting:
+    def test_none(self):
+        vis, lang, wrapped = _simulate_thinker_routing(None)
+        assert vis is None
+        assert lang is None
+        assert wrapped is None
+
+    @pytest.mark.parametrize("method", ["modelopt", "modelopt_fp4", "modelopt_mxfp8"])
+    def test_pre_quantized_all_components(self, method):
+        """Pre-quantized methods pass config to all components."""
+        cfg = _MockQuantConfig(method)
+        vis, lang, wrapped = _simulate_thinker_routing(cfg)
+        assert vis is cfg
+        assert lang is cfg
+        assert wrapped is cfg
+
+    def test_fp8_dynamic_language_only(self):
+        """fp8 dynamic: visual=None, language gets original config."""
+        cfg = _MockQuantConfig("fp8")
+        vis, lang, wrapped = _simulate_thinker_routing(cfg)
+        assert vis is None
+        assert lang is cfg
+        assert isinstance(wrapped, ComponentQuantizationConfig)
+        assert wrapped.resolve("language_model") is cfg
+        assert wrapped.resolve("visual") is None
+
+    def test_inc_autoround_language_only(self):
+        """INC/AutoRound: not in _PRE_QUANTIZED_METHODS -> wrapped like fp8."""
+        cfg = _MockQuantConfig("inc")
+        vis, lang, wrapped = _simulate_thinker_routing(cfg)
+        assert vis is None
+        assert lang is cfg
+        assert isinstance(wrapped, ComponentQuantizationConfig)
+
+    def test_component_config_passthrough(self):
+        """Explicit ComponentQuantizationConfig is used directly."""
+        inner_fp8 = _MockQuantConfig("fp8")
+        inner_modelopt = _MockQuantConfig("modelopt")
+        cqc = ComponentQuantizationConfig(
+            component_configs={
+                "visual": inner_modelopt,
+                "language_model": inner_fp8,
+            }
+        )
+        vis, lang, wrapped = _simulate_thinker_routing(cqc)
+        assert vis is inner_modelopt
+        assert lang is inner_fp8
+        assert wrapped is cqc
+
+
+# ===================================================================
+# 3. Talker visual routing (init_multi_modal guard)
+# ===================================================================
+
+
+def _simulate_talker_visual_routing(quant_config):
+    """Simulate the talker init_multi_modal visual routing."""
+    if quant_config is not None and quant_config.get_name() in PRE_QUANTIZED_METHODS:
+        return quant_config
+    return None
+
+
+class TestTalkerVisualRouting:
+    def test_none(self):
+        assert _simulate_talker_visual_routing(None) is None
+
+    @pytest.mark.parametrize("method", ["modelopt", "modelopt_fp4", "modelopt_mxfp8"])
+    def test_pre_quantized_passes_through(self, method):
+        """Pre-quantized methods pass quant config to visual."""
+        cfg = _MockQuantConfig(method)
+        assert _simulate_talker_visual_routing(cfg) is cfg
+
+    def test_fp8_blocked(self):
+        """fp8 dynamic must NOT be passed to visual."""
+        cfg = _MockQuantConfig("fp8")
+        assert _simulate_talker_visual_routing(cfg) is None
+
+    def test_inc_blocked(self):
+        """INC/AutoRound must NOT be passed to visual (not in _PRE_QUANTIZED_METHODS)."""
+        cfg = _MockQuantConfig("inc")
+        assert _simulate_talker_visual_routing(cfg) is None
+
+
+# ===================================================================
+# 4. ComponentQuantizationConfig.resolve
+# ===================================================================
+
+
+class TestComponentResolve:
+    def test_longest_prefix_match(self):
+        a = _MockQuantConfig("a")
+        b = _MockQuantConfig("b")
+        cqc = ComponentQuantizationConfig(component_configs={"language_model": a, "language_model.model": b})
+        assert cqc.resolve("language_model.model.layers.0") is b
+        assert cqc.resolve("language_model.lm_head") is a
+
+    def test_no_match_returns_default(self):
+        a = _MockQuantConfig("a")
+        default = _MockQuantConfig("default")
+        cqc = ComponentQuantizationConfig(
+            component_configs={"language_model": a},
+            default_config=default,
+        )
+        assert cqc.resolve("visual") is default
+
+    def test_no_match_no_default_returns_none(self):
+        a = _MockQuantConfig("a")
+        cqc = ComponentQuantizationConfig(
+            component_configs={"language_model": a},
+        )
+        assert cqc.resolve("visual") is None
+
+    def test_get_name(self):
+        cqc = ComponentQuantizationConfig(component_configs={})
+        assert cqc.get_name() == "component"
+
+    def test_get_quant_method_delegates(self):
+        """get_quant_method dispatches to the resolved config."""
+        inner = _MockQuantConfig("fp8")
+        cqc = ComponentQuantizationConfig(
+            component_configs={"language_model": inner},
+        )
+        layer = MagicMock()
+        result = cqc.get_quant_method(layer, "language_model.model.layers.0.mlp")
+        assert result is not None  # delegates to inner.get_quant_method
+
+    def test_get_quant_method_returns_none_for_unmatched(self):
+        """get_quant_method returns None when no config matches."""
+        inner = _MockQuantConfig("fp8")
+        cqc = ComponentQuantizationConfig(
+            component_configs={"language_model": inner},
+        )
+        layer = MagicMock()
+        result = cqc.get_quant_method(layer, "visual.blocks.0.mlp")
+        assert result is None
+
+    def test_min_capability(self):
+        a = _MockQuantConfig("a")
+        a.get_min_capability = lambda: 80
+        b = _MockQuantConfig("b")
+        b.get_min_capability = lambda: 70
+        cqc = ComponentQuantizationConfig(component_configs={"x": a, "y": b})
+        assert cqc.get_min_capability() == 70
+
+    def test_min_capability_empty(self):
+        cqc = ComponentQuantizationConfig(component_configs={})
+        assert cqc.get_min_capability() == 0
diff --git a/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16.py b/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16.py
new file mode 100644
index 000000000..e59a8c7d7
--- /dev/null
+++ b/tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16.py
@@ -0,0 +1,170 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""E2E tests for Qwen2.5-Omni AutoRound W4A16 quantized inference.
+
+These tests cover text, audio, image, video, and mixed-modality inputs
+to verify multimodal understanding with quantized weights.
+
+Requirements:
+  - CUDA GPUs (4x L4 / 24 GB or equivalent)
+  - The quantized model checkpoint (Intel/Qwen2.5-Omni-7B-int4-AutoRound)
+"""
+
+import os
+
+import pytest
+
+from tests.helpers.mark import hardware_test
+from tests.helpers.media import (
+    generate_synthetic_audio,
+    generate_synthetic_image,
+    generate_synthetic_video,
+)
+from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config
+
+QUANTIZED_MODEL = "Intel/Qwen2.5-Omni-7B-int4-AutoRound"
+BASELINE_MODEL = "Qwen/Qwen2.5-Omni-7B"
+
+# Allow overriding via environment for local testing
+QUANTIZED_MODEL = os.environ.get("QWEN2_5_OMNI_AUTOROUND_MODEL", QUANTIZED_MODEL)
+BASELINE_MODEL = os.environ.get("QWEN2_5_OMNI_BASELINE_MODEL", BASELINE_MODEL)
+
+_CI_DEPLOY = get_deploy_config_path("ci/qwen2_5_omni.yaml")
+
+
+def _get_stage_config():
+    """Build a CI-friendly stage config with eager mode."""
+    return modify_stage_config(
+        _CI_DEPLOY,
+        updates={
+            "stages": {
+                0: {"enforce_eager": True},
+                1: {"enforce_eager": True},
+            },
+        },
+    )
+
+
+stage_config = _get_stage_config()
+
+# Parametrise: (model, stage_config)
+quant_params = [(QUANTIZED_MODEL, stage_config)]
+
+
+# ------------------------------------------------------------------
+# Test: text-only input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "L4"}, num_cards=4)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_text_to_text(omni_runner, omni_runner_handler):
+    """Text input → text output with W4A16 quantized Qwen2.5-Omni."""
+    request_config = {
+        "prompts": "What is the capital of China?",
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: audio input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "L4"}, num_cards=4)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_audio_to_text(omni_runner, omni_runner_handler):
+    """Audio input → text output with W4A16 quantized Qwen2.5-Omni."""
+    audio = generate_synthetic_audio(1, 1, 16000)["np_array"]
+    if len(audio.shape) == 2:
+        audio = audio.squeeze()
+
+    request_config = {
+        "prompts": "What is the content of this audio?",
+        "audios": (audio, 16000),
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: image input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "L4"}, num_cards=4)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_image_to_text(omni_runner, omni_runner_handler):
+    """Image input → text output with W4A16 quantized Qwen2.5-Omni."""
+    image = generate_synthetic_image(16, 16)["np_array"]
+
+    request_config = {
+        "prompts": "Describe what you see in this image.",
+        "images": image,
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: video input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "L4"}, num_cards=4)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_video_to_text(omni_runner, omni_runner_handler):
+    """Video input → text output with W4A16 quantized Qwen2.5-Omni."""
+    video = generate_synthetic_video(16, 16, 30)["np_array"]
+
+    request_config = {
+        "prompts": "Describe the video briefly.",
+        "videos": video,
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: mixed modality (audio + image + video) → audio output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "L4"}, num_cards=4)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_mix_to_audio(omni_runner, omni_runner_handler):
+    """Mixed-modality input → audio output with W4A16 quantized Qwen2.5-Omni."""
+    video = generate_synthetic_video(16, 16, 30)["np_array"]
+    image = generate_synthetic_image(16, 16)["np_array"]
+    audio = generate_synthetic_audio(1, 1, 16000)["np_array"]
+    if len(audio.shape) == 2:
+        audio = audio.squeeze()
+
+    request_config = {
+        "prompts": "What is recited in the audio? What is in this image? Describe the video briefly.",
+        "videos": video,
+        "images": image,
+        "audios": (audio, 16000),
+        "modalities": ["audio"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
diff --git a/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16.py b/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16.py
new file mode 100644
index 000000000..3a3c874b6
--- /dev/null
+++ b/tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16.py
@@ -0,0 +1,205 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""E2E tests for Qwen3-Omni AutoRound W4A16 quantized inference.
+
+These tests cover text, audio, image, video, and mixed-modality inputs
+to verify multimodal understanding with quantized weights.
+
+Requirements:
+  - CUDA GPUs (2x H100-80G or equivalent)
+  - The quantized model checkpoint (Intel/Qwen3-Omni-30B-A3B-Instruct-int4-AutoRound)
+"""
+
+import os
+
+import pytest
+
+from tests.helpers.mark import hardware_test
+from tests.helpers.media import (
+    generate_synthetic_audio,
+    generate_synthetic_image,
+    generate_synthetic_video,
+)
+from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config
+
+QUANTIZED_MODEL = "Intel/Qwen3-Omni-30B-A3B-Instruct-int4-AutoRound"
+BASELINE_MODEL = "Qwen/Qwen3-Omni-30B-A3B-Instruct"
+
+# Allow overriding via environment for local testing
+QUANTIZED_MODEL = os.environ.get("QWEN3_OMNI_AUTOROUND_MODEL", QUANTIZED_MODEL)
+BASELINE_MODEL = os.environ.get("QWEN3_OMNI_BASELINE_MODEL", BASELINE_MODEL)
+
+_CI_DEPLOY = get_deploy_config_path("ci/qwen3_omni_moe.yaml")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _qwen3_omni_env():
+    """Set env vars required by multi-stage worker spawning.
+
+    Must run before CUDA context init.  Reverted after every test module
+    so that values do not leak into unrelated test files.
+    """
+    with pytest.MonkeyPatch.context() as mp:
+        mp.setenv("VLLM_WORKER_MULTIPROC_METHOD", "spawn")
+        mp.setenv("VLLM_TEST_CLEAN_GPU_MEMORY", "0")
+        yield
+
+
+def _get_stage_config():
+    """Build a CI-friendly stage config with eager mode."""
+    return modify_stage_config(
+        _CI_DEPLOY,
+        updates={
+            "stages": {
+                0: {"enforce_eager": True},
+                1: {"enforce_eager": True},
+            },
+        },
+    )
+
+
+stage_config = _get_stage_config()
+
+# Parametrise: (model, stage_config)
+quant_params = [(QUANTIZED_MODEL, stage_config)]
+
+
+# ------------------------------------------------------------------
+# Test: text-only input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_text_to_text(omni_runner, omni_runner_handler):
+    """Text input → text output with W4A16 quantized Qwen3-Omni."""
+    request_config = {
+        "prompts": "What is the capital of France?",
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: audio input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_audio_to_text(omni_runner, omni_runner_handler):
+    """Audio input → text output with W4A16 quantized Qwen3-Omni."""
+    audio = generate_synthetic_audio(1, 1, 16000)["np_array"]
+    if len(audio.shape) == 2:
+        audio = audio.squeeze()
+
+    request_config = {
+        "prompts": "What is the content of this audio?",
+        "audios": (audio, 16000),
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: image input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_image_to_text(omni_runner, omni_runner_handler):
+    """Image input → text output with W4A16 quantized Qwen3-Omni."""
+    image = generate_synthetic_image(16, 16)["np_array"]
+
+    request_config = {
+        "prompts": "Describe what you see in this image.",
+        "images": image,
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: video input → text output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_video_to_text(omni_runner, omni_runner_handler):
+    """Video input → text output with W4A16 quantized Qwen3-Omni."""
+    video = generate_synthetic_video(224, 224, 300)["np_array"]
+
+    request_config = {
+        "prompts": "Describe the video briefly.",
+        "videos": video,
+        "modalities": ["text"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+    assert response.text_content and len(response.text_content.strip()) > 0
+
+
+# ------------------------------------------------------------------
+# Test: video input → audio output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_video_to_audio(omni_runner, omni_runner_handler):
+    """Video input → audio output with W4A16 quantized Qwen3-Omni."""
+    video = generate_synthetic_video(224, 224, 300)["np_array"]
+
+    request_config = {
+        "prompts": "Describe the video briefly.",
+        "videos": video,
+        "modalities": ["audio"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
+
+
+# ------------------------------------------------------------------
+# Test: mixed modality (audio + image + video) → audio output
+# ------------------------------------------------------------------
+
+
+@pytest.mark.advanced_model
+@pytest.mark.omni
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_runner", quant_params, indirect=True)
+def test_mix_to_audio(omni_runner, omni_runner_handler):
+    """Mixed-modality input → audio output with W4A16 quantized Qwen3-Omni."""
+    video = generate_synthetic_video(224, 224, 300)["np_array"]
+    image = generate_synthetic_image(16, 16)["np_array"]
+    audio = generate_synthetic_audio(1, 1, 16000)["np_array"]
+    if len(audio.shape) == 2:
+        audio = audio.squeeze()
+
+    request_config = {
+        "prompts": "What is recited in the audio? What is in this image? Describe the video briefly.",
+        "videos": video,
+        "images": image,
+        "audios": (audio, 16000),
+        "modalities": ["audio"],
+    }
+    response = omni_runner_handler.send_request(request_config)
+    assert response.success, f"Request failed: {response.error_message}"
diff --git a/vllm_omni/config/model.py b/vllm_omni/config/model.py
index af89184a6..217a600b4 100644
--- a/vllm_omni/config/model.py
+++ b/vllm_omni/config/model.py
@@ -49,6 +49,25 @@ class OmniModelArchConfigConvertor(ModelArchConfigConvertorBase):
                     if quant_cfg is not None:
                         return quant_cfg
 
+            # Fall back to top-level quantization_config
+            top_quant = super().get_quantization_config()
+            if top_quant is not None:
+                block_names = top_quant.get("block_name_to_quantize")
+                if block_names is not None:
+                    # NOTE: This assumes stage_config_name follows the HF
+                    # ``<stage>_config`` convention (e.g. thinker_config →
+                    # prefix "thinker.").  removesuffix is a no-op when
+                    # the suffix doesn't match, so a non-standard name
+                    # would just use itself as prefix — safe but worth
+                    # verifying if new stage names are introduced.
+                    hf_prefix = self.stage_config_name.removesuffix("_config") + "."
+                    if isinstance(block_names, str):
+                        block_names = [b.strip() for b in block_names.split(",")]
+                    if isinstance(block_names, list) and not any(b.startswith(hf_prefix) for b in block_names):
+                        # This stage is not listed → no quantization.
+                        return None
+                return top_quant
+
             # For non-thinker stages (talker, code2wav) whose text_config
             # has no quantization_config, return None so quantization is
             # not applied to stages that were not quantized.
diff --git a/vllm_omni/engine/stage_init_utils.py b/vllm_omni/engine/stage_init_utils.py
index 94b9faa80..cc7676ba5 100644
--- a/vllm_omni/engine/stage_init_utils.py
+++ b/vllm_omni/engine/stage_init_utils.py
@@ -14,7 +14,7 @@ import multiprocessing as mp
 import os
 import time
 from collections.abc import Callable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Any, Literal
 
 from vllm.logger import init_logger
@@ -29,6 +29,7 @@ from vllm_omni.entrypoints.stage_utils import _to_dict, set_stage_devices
 from vllm_omni.entrypoints.utils import filter_dataclass_kwargs, resolve_model_config_path
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams, OmniSamplingParams
 from vllm_omni.platforms import current_omni_platform
+from vllm_omni.quantization.inc_config import OmniINCConfig
 
 logger = init_logger(__name__)
 
@@ -459,6 +460,11 @@ def build_vllm_config(
     )
     executor_class = Executor.get_class(vllm_config)
 
+    # Upgrade vanilla INCConfig to OmniINCConfig for multi-stage models.
+    upgraded = OmniINCConfig.maybe_upgrade(vllm_config.quant_config)
+    if upgraded is not vllm_config.quant_config:
+        vllm_config = replace(vllm_config, quant_config=upgraded)
+
     return vllm_config, executor_class
 
 
diff --git a/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_talker.py b/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_talker.py
index bb491d01b..28f49918f 100644
--- a/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_talker.py
+++ b/vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_talker.py
@@ -100,11 +100,23 @@ class Qwen3OmniMoeTalkerForConditionalGeneration(
     def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
         super().__init__()
         talker_config: Qwen3OmniMoeTalkerConfig = vllm_config.model_config.hf_config
-        rope_params = talker_config.text_config.rope_scaling
+        rope_params = getattr(talker_config.text_config, "rope_scaling", None)
         if rope_params is None:
-            # Newer transformers use rope_parameters instead of rope_scaling
             rope_params = getattr(talker_config.text_config, "rope_parameters", None) or {}
-        rope_params["rope_theta"] = talker_config.text_config.rope_theta
+        rope_params = dict(rope_params)
+        # In transformers <5.0.0, rope_theta is a top-level config attribute
+        # (e.g. config.text_config.rope_theta = 1000000.0).
+        # In transformers >=5.0.0 (PR #39847), rope_theta moved inside the
+        # rope_parameters dict (e.g. config.text_config.rope_parameters =
+        # {"rope_theta": 1000000.0, "rope_type": "default"}).
+        # Use setdefault so we never overwrite a value already present.
+        # Precedence: rope_params["rope_theta"] (already set)
+        #           > text_config.rope_theta (transformers <5.0.0 top-level attr)
+        #           > 1000000 (Qwen3 Omni default)
+        rope_params.setdefault(
+            "rope_theta",
+            getattr(talker_config.text_config, "rope_theta", 1000000),
+        )
         talker_config.text_config.rope_parameters = rope_params
         quant_config = vllm_config.quant_config
         if isinstance(quant_config, ComponentQuantizationConfig):
diff --git a/vllm_omni/quantization/__init__.py b/vllm_omni/quantization/__init__.py
index b50fe8607..a709dc74d 100644
--- a/vllm_omni/quantization/__init__.py
+++ b/vllm_omni/quantization/__init__.py
@@ -13,6 +13,7 @@ Adds per-component quantization for multi-stage models.
 
 from .component_config import ComponentQuantizationConfig
 from .factory import SUPPORTED_QUANTIZATION_METHODS, build_quant_config
+from .inc_config import OmniINCConfig
 
 # DiffusionGGUFConfig is NOT imported here to avoid pulling in
 # GGUF -> fused_moe -> pynvml at module load time.
@@ -20,5 +21,6 @@ from .factory import SUPPORTED_QUANTIZATION_METHODS, build_quant_config
 __all__ = [
     "build_quant_config",
     "ComponentQuantizationConfig",
+    "OmniINCConfig",
     "SUPPORTED_QUANTIZATION_METHODS",
 ]
diff --git a/vllm_omni/quantization/factory.py b/vllm_omni/quantization/factory.py
index f85589d69..5af47a1f7 100644
--- a/vllm_omni/quantization/factory.py
+++ b/vllm_omni/quantization/factory.py
@@ -43,16 +43,16 @@ def _build_int8(**kw: Any) -> QuantizationConfig:
 
 def _build_inc(**kw: Any) -> QuantizationConfig:
     """Lazy import for INC/AutoRound config with checkpoint kwarg normalization."""
-    from vllm.model_executor.layers.quantization.inc import INCConfig
+    from .inc_config import OmniINCConfig
 
     # Map checkpoint key 'bits' to INCConfig's 'weight_bits'
     if "bits" in kw and "weight_bits" not in kw:
         kw["weight_bits"] = kw.pop("bits")
 
     # Filter to only valid INCConfig params
-    valid = set(inspect.signature(INCConfig.__init__).parameters) - {"self"}
+    valid = set(inspect.signature(OmniINCConfig.__init__).parameters) - {"self"}
     filtered = {k: v for k, v in kw.items() if k in valid}
-    return INCConfig(**filtered)
+    return OmniINCConfig(**filtered)
 
 
 _OVERRIDES: dict[str, Callable[..., QuantizationConfig]] = {
diff --git a/vllm_omni/quantization/inc_config.py b/vllm_omni/quantization/inc_config.py
new file mode 100644
index 000000000..fb4851cca
--- /dev/null
+++ b/vllm_omni/quantization/inc_config.py
@@ -0,0 +1,140 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Extended INC/AutoRound config for multi-stage omni models."""
+
+from __future__ import annotations
+
+from os.path import commonprefix
+from typing import TYPE_CHECKING, Any
+
+from vllm.model_executor.layers.quantization.inc import INCConfig
+from vllm.model_executor.models.utils import WeightsMapper
+
+if TYPE_CHECKING:
+    from vllm.model_executor.layers.quantization.base_config import (
+        QuantizationConfig,
+    )
+
+_REGEX_SPECIAL_CHARS = frozenset(r"*+?^$()[]{}|\\")
+
+
+def _stage_prefix(prefix_map: dict[str, str | None]) -> str:
+    """Derive the container/stage prefix from mapper source keys."""
+    cp = commonprefix(list(prefix_map.keys()))
+    dot = cp.rfind(".")
+    return cp[: dot + 1] if dot >= 0 else ""
+
+
+def _map_with_stage_prefix(
+    items: list[str],
+    prefix_map: dict[str, str | None],
+    stage: str,
+) -> list[str]:
+    """Apply *prefix_map* to each item and prepend *stage* to mapped items."""
+    sorted_keys = sorted(prefix_map, key=len, reverse=True)
+    result: list[str] = []
+    for item in items:
+        new_item = item
+        for orig in sorted_keys:
+            if item.startswith(orig):
+                new_val = prefix_map[orig] or ""
+                new_item = stage + new_val + item[len(orig) :]
+                break
+        result.append(new_item)
+    return result
+
+
+class OmniINCConfig(INCConfig):
+    """INCConfig extended with multi-stage prefix remapping."""
+
+    # ------------------------------------------------------------------
+    # Core integration: called by vLLM's configure_quant_config()
+    # ------------------------------------------------------------------
+
+    def apply_vllm_mapper(self, hf_to_vllm_mapper: WeightsMapper) -> None:
+        """Remap HF checkpoint names to vLLM runtime prefixes."""
+        prefix_map = getattr(hf_to_vllm_mapper, "orig_to_new_prefix", None) or {}
+        stage = _stage_prefix(prefix_map) if prefix_map else ""
+
+        # -- Normalize CSV string -----------------------------------------
+        if isinstance(self.block_name_to_quantize, str):
+            self.block_name_to_quantize = [b.strip() for b in self.block_name_to_quantize.split(",") if b.strip()]
+
+        # -- block_name_to_quantize ----------------------------------------
+        if self.block_name_to_quantize is not None:
+            if prefix_map and stage:
+                self.block_name_to_quantize = _map_with_stage_prefix(
+                    self.block_name_to_quantize,
+                    prefix_map,
+                    stage,
+                )
+            else:
+                self.block_name_to_quantize = hf_to_vllm_mapper.apply_list(self.block_name_to_quantize)
+
+        # -- extra_config --------------------------------------------------
+        if self.extra_config is not None and prefix_map:
+            new_extra: dict[str, Any] = {}
+            sorted_keys = sorted(prefix_map, key=len, reverse=True)
+
+            # Build escaped-dot map for regex pattern keys
+            escaped_map: dict[str, str] = {}
+            for orig, new in prefix_map.items():
+                escaped_map[orig.replace(".", r"\.")] = (new or "").replace(".", r"\.")
+            escaped_sorted = sorted(escaped_map, key=len, reverse=True)
+
+            for key, val in self.extra_config.items():
+                is_regex = any(c in _REGEX_SPECIAL_CHARS for c in key)
+                if is_regex:
+                    # Regex keys: escaped-dot substring replacement.
+                    # re.search matches anywhere so no stage prefix needed.
+                    new_key = key
+                    for esc_orig in escaped_sorted:
+                        if esc_orig in new_key:
+                            new_key = new_key.replace(
+                                esc_orig,
+                                escaped_map[esc_orig],
+                                1,
+                            )
+                            break
+                else:
+                    # Plain keys: prefix replacement + stage prefix.
+                    new_key = key
+                    for orig in sorted_keys:
+                        if key.startswith(orig):
+                            new_val = prefix_map[orig] or ""
+                            new_key = stage + new_val + key[len(orig) :]
+                            break
+                new_extra[new_key] = val
+            self.extra_config = new_extra
+        elif self.extra_config is not None:
+            self.extra_config = hf_to_vllm_mapper.apply_dict(self.extra_config)
+
+    # ------------------------------------------------------------------
+    # Upgrading a vanilla INCConfig created by vLLM
+    # ------------------------------------------------------------------
+
+    @classmethod
+    def from_inc_config(cls, inc: INCConfig) -> OmniINCConfig:
+        """Promote a vanilla :class:`INCConfig` to :class:`OmniINCConfig`.
+
+        Copies all attributes so that the new instance is a drop-in
+        replacement.
+        """
+        omni = object.__new__(cls)
+        omni.__dict__.update(inc.__dict__)
+        return omni
+
+    @classmethod
+    def maybe_upgrade(cls, quant_config: QuantizationConfig | None) -> QuantizationConfig | None:
+        """Upgrade *quant_config* to :class:`OmniINCConfig` if applicable.
+
+        Returns the original config unchanged when it is not an INC
+        config or is already an :class:`OmniINCConfig`.
+        """
+        if quant_config is None:
+            return None
+        if isinstance(quant_config, cls):
+            return quant_config
+        if isinstance(quant_config, INCConfig):
+            return cls.from_inc_config(quant_config)
+        return quant_config