diff --git a/tests/entrypoints/openai_api/test_tts_adapter.py b/tests/entrypoints/openai_api/test_tts_adapter.py
new file mode 100644
index 000000000..2930a204d
--- /dev/null
+++ b/tests/entrypoints/openai_api/test_tts_adapter.py
@@ -0,0 +1,101 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Unit tests for the TTS serving adapter registry (RFC #4327).
+
+Pure-Python registry/resolution logic; no model or GPU resources are loaded.
+"""
+
+import pytest
+
+from vllm_omni.entrypoints.openai.tts_adapters import (
+    TTS_ADAPTER_REGISTRY,
+    ARTTSAdapter,
+    DiffusionTTSAdapter,
+    all_tts_model_types,
+    resolve_adapter,
+)
+from vllm_omni.entrypoints.openai.tts_adapters.qwen3_tts import Qwen3TTSAdapter
+
+# Every dedicated TTS model-type must have an adapter so the orchestrator's
+# uniform ``self._adapter.build(...)`` dispatch covers it.
+EXPECTED_MODEL_TYPES = {
+    "qwen3_tts",
+    "voxcpm2",
+    "voxtral_tts",
+    "fish_tts",
+    "cosyvoice3",
+    "omnivoice",
+    "covo_audio",
+    "ming_tts",
+    "moss_tts_nano",
+    "moss_tts",
+    "higgs_audio_v2",
+    "higgs_audio_v3",
+    "glm_tts",
+    "step_audio2",
+}
+
+
+def test_all_model_types_registered():
+    assert EXPECTED_MODEL_TYPES <= all_tts_model_types()
+
+
+def test_registry_keyed_by_name():
+    for name, cls in TTS_ADAPTER_REGISTRY.items():
+        assert cls.name == name
+
+
+def test_resolve_each_model_type():
+    for model_type in EXPECTED_MODEL_TYPES:
+        cls = resolve_adapter(model_type)
+        assert cls is not None, model_type
+        assert cls.name == model_type
+
+
+def test_resolve_qwen3_tts_class():
+    assert resolve_adapter("qwen3_tts") is Qwen3TTSAdapter
+
+
+def test_resolve_unknown_returns_none():
+    assert resolve_adapter("not_a_real_model") is None
+    assert resolve_adapter(None) is None
+
+
+def test_ming_flash_omni_not_migrated():
+    """ming_flash_omni is intentionally excluded from the adapter migration in
+    this PR; it stays on the legacy inline dispatch in serving_speech.py."""
+    assert resolve_adapter("ming_flash_omni_tts") is None
+
+
+def test_voxcpm2_resolves():
+    """VoxCPM2 (the served ``latent_generator`` model) resolves cleanly.
+
+    Detection never returns the legacy ``voxcpm`` type, so there is no shared
+    stage-key ambiguity to resolve.
+    """
+    assert resolve_adapter("voxcpm2") is not None
+    assert resolve_adapter("voxcpm") is None
+
+
+def test_all_adapters_are_ar_or_diffusion():
+    for cls in TTS_ADAPTER_REGISTRY.values():
+        assert issubclass(cls, (ARTTSAdapter, DiffusionTTSAdapter))
+        assert cls.backend in ("ar", "diffusion")
+
+
+def test_qwen3_tts_metadata():
+    assert Qwen3TTSAdapter.backend == "ar"
+    assert issubclass(Qwen3TTSAdapter, ARTTSAdapter)
+
+
+def test_diffusion_adapter_extra_body_params_fallback():
+    class _DiffAdapter(DiffusionTTSAdapter):
+        name = "diff_probe"
+
+        async def build(self, request, sampling_params_list):  # pragma: no cover
+            raise NotImplementedError
+
+    assert _DiffAdapter.extra_body_params() == frozenset()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])
diff --git a/tests/helpers/assertions.py b/tests/helpers/assertions.py
index f79c50173..da62ceab2 100644
--- a/tests/helpers/assertions.py
+++ b/tests/helpers/assertions.py
@@ -520,9 +520,25 @@ def assert_omni_response(response: Any, request_config: dict[str, Any], run_leve
 def assert_audio_speech_response(response: Any, request_config: dict[str, Any], run_level: str) -> None:
     """Validate speech API results from :class:`~tests.helpers.runtime.OmniResponse`.
 
-    Success path only. Use :meth:`OpenAIClientHandler.send_audio_speech_http_request` with ``err_code`` /
-    ``err_message`` for expected HTTP errors.
+    When ``request_config`` carries ``status_code`` and/or ``err_message``, the
+    request is expected to be rejected: assert it failed and that the HTTP status
+    / error text match. Otherwise the normal success-path checks run.
     """
+    expected_status = request_config.get("status_code")
+    expected_err = request_config.get("err_message")
+    if expected_status is not None or expected_err is not None:
+        assert not response.success, "Expected an error response, but the request succeeded."
+        if expected_status is not None:
+            allowed = expected_status if isinstance(expected_status, (list, tuple)) else (expected_status,)
+            assert response.status_code in allowed, f"Expected HTTP status in {allowed}, got {response.status_code}"
+        if expected_err is not None:
+            alternatives = expected_err if isinstance(expected_err, (list, tuple)) else (expected_err,)
+            error_text = response.error_message or ""
+            assert any(alt in error_text for alt in alternatives), (
+                f"Expected one of {alternatives} in error text, got: {error_text!r}"
+            )
+        return
+
     assert response.success, "The request failed."
 
     # Optional floor on decoded audio size (models with very short clips may use a lower value).
diff --git a/tests/helpers/runtime.py b/tests/helpers/runtime.py
index e87017df8..c12154f86 100644
--- a/tests/helpers/runtime.py
+++ b/tests/helpers/runtime.py
@@ -762,6 +762,10 @@ class OmniResponse:
     prompt_tokens: int | None = None
     cached_tokens: int | None = None
     logprobs: list | None = None
+    #: HTTP status + error text for the error-handling path (e.g. validator
+    #: rejections); populated when the OpenAI client raises an APIError.
+    status_code: int | None = None
+    error_message: str | None = None
 
 
 @dataclass
diff --git a/vllm_omni/entrypoints/openai/serving_speech.py b/vllm_omni/entrypoints/openai/serving_speech.py
index 92ebb0f9f..1f7a6b79d 100644
--- a/vllm_omni/entrypoints/openai/serving_speech.py
+++ b/vllm_omni/entrypoints/openai/serving_speech.py
@@ -44,6 +44,10 @@ from vllm_omni.entrypoints.openai.protocol.audio import (
     SpeechBatchItem,
     SpeechBatchItemResult,
 )
+from vllm_omni.entrypoints.openai.tts_adapters import (
+    SpeechServingContext,
+    resolve_adapter,
+)
 from vllm_omni.entrypoints.utils import coerce_param_message_types
 from vllm_omni.model_executor.models.fish_speech.prompt_utils import (
     build_fish_text_only_prompt_ids,
@@ -474,6 +478,36 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         self._build_fish_speech_prompt_async = make_async(self._build_fish_speech_prompt, executor=self._tts_executor)
         self._estimate_prompt_len_async = make_async(self._estimate_prompt_len, executor=self._tts_executor)
 
+        # Resolve the per-model serving adapter (RFC #4327), keyed on the
+        # detected model-type. Every dedicated TTS model has an adapter; the
+        # adapter owns request validation + prompt/param building. Sampling
+        # overrides and the model-type label remain in the orchestrator tail
+        # (keyed on ``_tts_model_type``) during this incremental migration.
+        self._adapter = None
+        if self._tts_stage is not None:
+            adapter_cls = resolve_adapter(self._tts_model_type)
+            if adapter_cls is not None:
+                ctx = SpeechServingContext(server=self, engine_client=self.engine_client)
+                self._adapter = adapter_cls(ctx)
+                logger.info("Resolved TTS serving adapter: %s", adapter_cls.__name__)
+
+    def _get_tts_adapter(self):
+        """Return the per-model serving adapter for the current ``_tts_model_type``.
+
+        Resolved lazily (rebuilt if ``_tts_model_type`` changed since the cached
+        instance was built) so callers that set ``_tts_model_type`` after
+        construction still dispatch to the matching adapter. In production
+        ``_tts_model_type`` is fixed at init, so the cached instance is reused.
+        """
+        adapter_cls = resolve_adapter(self._tts_model_type)
+        if adapter_cls is None:
+            self._adapter = None
+            return None
+        if self._adapter is None or type(self._adapter) is not adapter_cls:
+            ctx = SpeechServingContext(server=self, engine_client=self.engine_client)
+            self._adapter = adapter_cls(ctx)
+        return self._adapter
+
     async def warmup(self) -> None:
         """Run a synthetic speech request to trigger all first-request warmup.
 
@@ -3212,6 +3246,44 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
 
     # ---- Common speech generation helpers ----
 
+    async def _build_qwen3_tts_request(
+        self,
+        request: OpenAICreateSpeechRequest,
+    ) -> tuple[dict[str, Any], dict[str, Any], str | None]:
+        """Build prompt + tts_params for Qwen3-TTS.
+
+        Called from ``Qwen3TTSAdapter.build``. Returns
+        ``(prompt, tts_params, warmup_artifact_key)`` where the warmup key is the
+        Qwen3-TTS ref-audio artifact tracked after ``generate()``.
+        """
+        qwen3_ref_audio_warmup_artifact_key: str | None = None
+        tts_params = self._build_tts_params(request)
+        # Resolve ref_audio (explicit or auto-set for uploaded voices)
+        # to [[wav_list, sr]] so the model doesn't re-decode base64.
+        ref_audio_source = request.ref_audio
+        if ref_audio_source is None and isinstance(tts_params.get("ref_audio"), list):
+            # Uploaded voice: ref_audio was auto-set as [base64_data_url]
+            ref_audio_source = tts_params["ref_audio"][0]
+        if ref_audio_source is not None and isinstance(ref_audio_source, str):
+            wav_list, sr = await self._resolve_ref_audio(ref_audio_source)
+            artifact_key = self._get_resolved_ref_audio_artifact_key(ref_audio_source)
+            if artifact_key:
+                tts_params[_QWEN3_TTS_REF_AUDIO_CACHE_KEY] = [artifact_key]
+            ref_code_length = self._estimate_ref_code_len([wav_list, sr])
+            if ref_code_length is not None:
+                tts_params["ref_code_length"] = [int(ref_code_length)]
+            if self._qwen3_tts_can_use_ref_audio_artifact_only(tts_params, artifact_key):
+                logger.debug("Using Qwen3-TTS ref_audio artifact-only path: %s", artifact_key)
+            else:
+                tts_params["ref_audio"] = [[wav_list, sr]]
+                qwen3_ref_audio_warmup_artifact_key = artifact_key
+
+        ph_len = await self._estimate_prompt_len_async(tts_params)
+        prompt = tokens_input(prompt_token_ids=[1] * ph_len)
+        prompt["additional_information"] = tts_params
+        prompt["cache_salt"] = _conditioning_cache_salt(request, tts_params)
+        return prompt, tts_params, qwen3_ref_audio_warmup_artifact_key
+
     async def _prepare_speech_generation(
         self,
         request: OpenAICreateSpeechRequest,
@@ -3230,208 +3302,35 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         sampling_params_list = list(self.engine_client.default_sampling_params_list)
         sampling_params_list = coerce_param_message_types(sampling_params_list, request.stream)
 
-        # Resolve uploaded voice for non-Qwen3 models.
-        # Qwen3 TTS has its own uploaded voice handling in _build_tts_params().
+        # Build prompt + tts_params via the per-model adapter (RFC #4327). Every
+        # dedicated TTS model resolves to an adapter that owns its validation,
+        # uploaded-speaker handling, and prompt/param building. Sampling
+        # overrides and the model-type label remain in the orchestrator tail
+        # below (keyed on ``_tts_model_type``) during this incremental migration.
+        # Non-TTS deployments (no adapter) fall through to the rejection below.
+        # Capture inline-ref-audio status BEFORE validate(): several adapters
+        # apply uploaded speakers inside validate(), which sets request.ref_audio
+        # in place. The builders need to know whether the caller supplied audio
+        # inline vs. via an uploaded voice.
         has_inline_ref_audio = request.ref_audio is not None
-        if self._tts_model_type in ("fish_tts", "cosyvoice3", "moss_tts_nano", "glm_tts"):
-            err = self._apply_uploaded_speaker(request)
-            if err:
-                raise ValueError(err)
-
-        if self._tts_model_type == "step_audio2":
-            if not request.input or not request.input.strip():
-                raise ValueError("Input text cannot be empty")
-            prompt = self._build_step_audio2_prompt(request)
-            tts_params = {}
-        elif self._is_fish_speech:
-            validation_error = self._validate_fish_tts_request(request)
+        if self._tts_model_type == "ming_flash_omni_tts":
+            # ming_flash_omni is intentionally NOT migrated onto the adapter
+            # framework in this PR (it has no registered adapter); keep it on the
+            # legacy inline dispatch so serving still works.
+            validation_error = self._validate_ming_flash_omni_tts_request(request)
             if validation_error:
                 raise ValueError(validation_error)
-            ref_audio_data = None
-            if request.ref_audio is not None:
-                wav_list, sr = await self._resolve_ref_audio(request.ref_audio)
-                ref_audio_data = (wav_list, sr)
-            prompt = await self._build_fish_speech_prompt_async(
-                request, ref_audio_data=ref_audio_data, has_inline_ref_audio=has_inline_ref_audio
-            )
+            prompt = self._build_ming_flash_omni_prompt(request)
             tts_params = {}
-            # Fish Speech resolves conditioning entirely from raw request fields
-            # (text / ref_audio / ref_text rebuilt into inputs_embeds); unlike
-            # uploaded-voice models it has no resolved tts_params to fold in, so
-            # the salt is derived from the request alone.
-            prompt["cache_salt"] = _conditioning_cache_salt(request, tts_params)
-        elif self._tts_model_type == "omnivoice":
-            if not request.input or not request.input.strip():
-                raise ValueError("Input text cannot be empty")
-            err = self._apply_uploaded_speaker(request)
-            if err:
-                raise ValueError(err)
-            tts_params = {}
-            prompt: dict[str, Any] = {"input": request.input}
-            if request.ref_audio:
-                wav, sr = await self._resolve_ref_audio(request.ref_audio)
-                prompt["ref_audio"] = (np.asarray(wav, dtype=np.float32), sr)
-            if request.ref_text:
-                prompt["ref_text"] = request.ref_text
-            if request.voice:
-                voice_lower = request.voice.lower()
-                if voice_lower in self.uploaded_speakers and not has_inline_ref_audio:
-                    prompt["voice_name"] = voice_lower
-                    prompt["voice_created_at"] = self._voice_created_at(voice_lower)
-            if request.language:
-                prompt["lang"] = request.language
-            if request.instructions:
-                prompt["instruct"] = request.instructions
-        elif self._tts_model_type == "covo_audio":
-            prompt = self._build_covo_audio_prompt(request)
-            tts_params = {}
-        elif self._tts_model_type == "higgs_audio_v2":
-            # Explicit higgs_audio_v2 branch: do NOT fall through to the
-            # Qwen-style generic placeholder path; build prompt_token_ids
-            # directly via the upstream processor + the project's plain-text
-            # helper.
-            #
-            # Resolve uploaded voices first so a voice=<name> request (after
-            # POST /v1/audio/voices) populates ref_audio + ref_text from the
-            # cached speaker entry, mirroring the cosyvoice3 / voxcpm2 flow.
-            err = self._apply_uploaded_speaker(request)
-            if err:
-                raise ValueError(err)
-            validation_error = self._validate_higgs_audio_v2_request(request)
+            qwen3_ref_audio_warmup_artifact_key = None
+        elif (adapter := self._get_tts_adapter()) is not None:
+            validation_error = adapter.validate(request)
             if validation_error:
                 raise ValueError(validation_error)
-            prompt = await self._build_higgs_audio_v2_params(request)
-            if request.voice:
-                voice_lower = request.voice.lower()
-                if voice_lower in self.uploaded_speakers and not has_inline_ref_audio:
-                    additional = prompt.setdefault("additional_information", {})
-                    additional["voice_name"] = voice_lower
-                    additional["voice_created_at"] = self._voice_created_at(voice_lower)
-            tts_params = {}
-        elif self._tts_model_type == "higgs_audio_v3":
-            err = self._apply_uploaded_speaker(request)
-            if err:
-                raise ValueError(err)
-            validation_error = self._validate_higgs_audio_v3_request(request)
-            if validation_error:
-                raise ValueError(validation_error)
-            prompt = await self._build_higgs_audio_v3_params(request)
-            tts_params = {}
-        elif self._tts_model_type == "voxcpm2":
-            # voxcpm2 doesn't use `_apply_uploaded_speaker` because the prompt builder needs the
-            # raw waveform tuple for prefill-length accounting, not a base64 data URL.
-            validation_error = self._validate_voxcpm2_request(request)
-            if validation_error:
-                raise ValueError(validation_error)
-
-            uploaded_ref: tuple[np.ndarray, int] | None = None
-            if request.voice:
-                voice_lower = request.voice.lower()
-                if voice_lower in self.uploaded_speakers and not has_inline_ref_audio:
-                    if self.uploaded_speakers[voice_lower].get("embedding_source") == "direct":
-                        raise ValueError(
-                            f"Uploaded voice '{request.voice}' uses a speaker embedding (Qwen3-only). "
-                            f"Re-upload with an audio file for VoxCPM2."
-                        )
-                    if request.ref_audio is None:
-                        uploaded_ref = self._load_uploaded_audio(voice_lower)
-            prompt = await self._build_voxcpm2_prompt(request, uploaded_ref=uploaded_ref)
-            tts_params = {}
-            if request.voice:
-                voice_lower = request.voice.lower()
-                if voice_lower in self.uploaded_speakers or voice_lower in self.precomputed_speakers:
-                    additional = prompt.setdefault("additional_information", {})
-                    additional["voice_name"] = voice_lower
-                    additional["voice_created_at"] = self._voice_created_at(voice_lower)
-        elif self._is_tts:
-            validation_error = self._validate_tts_request(request)
-            if validation_error:
-                raise ValueError(validation_error)
-
-            if self._tts_model_type == "voxtral_tts":
-                prompt = await self._build_voxtral_prompt_async(request)
-                tts_params = {}
-            elif self._tts_model_type == "cosyvoice3":
-                prompt = await self._build_cosyvoice3_prompt(request, has_inline_ref_audio=has_inline_ref_audio)
-                tts_params = {}
-            elif self._tts_model_type == "glm_tts":
-                prompt = await self._build_glm_tts_prompt(request, has_inline_ref_audio=has_inline_ref_audio)
-                tts_params = {}
-            elif self._tts_model_type == "ming_flash_omni_tts":
-                prompt = self._build_ming_flash_omni_prompt(request)
-                tts_params = {}
-            elif self._tts_model_type == "ming_tts":
-                ref_audio_source = request.ref_audio
-                voice_lower = request.voice.lower() if isinstance(request.voice, str) else None
-                if ref_audio_source is None and voice_lower in self.uploaded_speakers:
-                    ref_audio_source = self._get_uploaded_audio_data(request.voice)
-                    if request.ref_text is None:
-                        request.ref_text = self.uploaded_speakers[voice_lower].get("ref_text")
-                ref_audio_data = None
-                if isinstance(ref_audio_source, list):
-                    ref_audio_data = await self._resolve_ref_audio_many(ref_audio_source)
-                    if request.speaker_embedding is None:
-                        request.speaker_embedding = self._extract_ming_speaker_embeddings_from_ref_audio(ref_audio_data)
-                elif ref_audio_source is not None and isinstance(ref_audio_source, str):
-                    wav_list, sr = await self._resolve_ref_audio(ref_audio_source)
-                    ref_audio_data = (wav_list, sr)
-                    if request.speaker_embedding is None:
-                        request.speaker_embedding = self._extract_ming_speaker_embeddings_from_ref_audio(
-                            [ref_audio_data]
-                        )[0]
-                prompt = self._build_ming_dense_prompt(request, ref_audio_data=ref_audio_data)
-                tts_params = prompt.get("additional_information", {})
-            elif self._tts_model_type in ("moss_tts_nano", "moss_tts"):
-                tts_params = await self._build_moss_tts_params(request)
-                if request.voice:
-                    voice_lower = request.voice.lower()
-                    if voice_lower in self.uploaded_speakers and not has_inline_ref_audio:
-                        tts_params["voice_name"] = [voice_lower]
-                        tts_params["voice_created_at"] = [self._voice_created_at(voice_lower)]
-                # Propagate seed from sampling params for deterministic generation.
-                # MOSS-TTS-Nano uses its own internal sampling inside
-                # inference_stream(), which reads seed from additional_information
-                # (not SamplingParams). Without this the model uses ambient RNG
-                # state and produces non-deterministic output.
-                if sampling_params_list and getattr(sampling_params_list[0], "seed", None) is not None:
-                    tts_params["seed"] = [sampling_params_list[0].seed]
-                # Full MOSS-TTS family returns real text_ids from the upstream
-                # processor; use them as prompt_token_ids. Nano keeps the
-                # ``[1]`` placeholder (it owns its own tokenisation internally).
-                if isinstance(tts_params.get("prompt_token_ids"), list):
-                    prompt_token_ids = tts_params.pop("prompt_token_ids")
-                    prompt = tokens_input(prompt_token_ids=prompt_token_ids)
-                else:
-                    prompt = tokens_input(prompt_token_ids=[1])
-                prompt["additional_information"] = tts_params
-                prompt["cache_salt"] = _conditioning_cache_salt(request, tts_params)
-            else:
-                tts_params = self._build_tts_params(request)
-                # Resolve ref_audio (explicit or auto-set for uploaded voices)
-                # to [[wav_list, sr]] so the model doesn't re-decode base64.
-                ref_audio_source = request.ref_audio
-                if ref_audio_source is None and isinstance(tts_params.get("ref_audio"), list):
-                    # Uploaded voice: ref_audio was auto-set as [base64_data_url]
-                    ref_audio_source = tts_params["ref_audio"][0]
-                if ref_audio_source is not None and isinstance(ref_audio_source, str):
-                    wav_list, sr = await self._resolve_ref_audio(ref_audio_source)
-                    artifact_key = self._get_resolved_ref_audio_artifact_key(ref_audio_source)
-                    if self._tts_model_type == "qwen3_tts" and artifact_key:
-                        tts_params[_QWEN3_TTS_REF_AUDIO_CACHE_KEY] = [artifact_key]
-                    ref_code_length = self._estimate_ref_code_len([wav_list, sr])
-                    if self._tts_model_type == "qwen3_tts" and ref_code_length is not None:
-                        tts_params["ref_code_length"] = [int(ref_code_length)]
-                    if self._qwen3_tts_can_use_ref_audio_artifact_only(tts_params, artifact_key):
-                        logger.debug("Using Qwen3-TTS ref_audio artifact-only path: %s", artifact_key)
-                    else:
-                        tts_params["ref_audio"] = [[wav_list, sr]]
-                        if self._tts_model_type == "qwen3_tts":
-                            qwen3_ref_audio_warmup_artifact_key = artifact_key
-
-                ph_len = await self._estimate_prompt_len_async(tts_params)
-                prompt = tokens_input(prompt_token_ids=[1] * ph_len)
-                prompt["additional_information"] = tts_params
-                prompt["cache_salt"] = _conditioning_cache_salt(request, tts_params)
+            prepared = await adapter.build(request, sampling_params_list, has_inline_ref_audio)
+            prompt = prepared.prompt
+            tts_params = prepared.tts_params
+            qwen3_ref_audio_warmup_artifact_key = prepared.warmup_artifact_key
         else:
             # Qwen omni models (Qwen3-Omni, Qwen2.5-Omni) use a "talker"
             # stage whose preprocess requires chat-templated tokens.  The
@@ -4047,6 +3946,11 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         merged_requests = [self._merge_batch_item(batch_request, item) for item in batch_request.items]
 
         async def _run_item(idx: int, req: OpenAICreateSpeechRequest) -> SpeechBatchItemResult:
+            # Batch validation still goes through _validate_tts_request directly
+            # (not the adapter). The single-request path validates via the
+            # adapter; both ultimately call the same per-model validators, so the
+            # surfaces stay in sync. Routing batch through the adapter (and its
+            # uploaded-speaker handling) is a follow-up (RFC #4327).
             validation_error = self._validate_tts_request(req)
             if validation_error is not None:
                 return SpeechBatchItemResult(index=idx, status="error", error=validation_error)
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/__init__.py b/vllm_omni/entrypoints/openai/tts_adapters/__init__.py
new file mode 100644
index 000000000..ca7b9612a
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/__init__.py
@@ -0,0 +1,80 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Registry of TTS serving adapters.
+
+Adapters register themselves by their ``name`` (the model-type discriminator
+returned by ``OmniOpenAIServingSpeech._detect_tts_model_type``) via
+``@register_tts_adapter``. Resolution is by that name: detection already encodes
+the (order-sensitive, partly ``model_arch``-based) stage logic, so keying the
+registry on the resolved type is both simpler and correct by construction.
+``stage_keys`` is retained on each adapter as documentation metadata.
+"""
+
+from vllm.logger import init_logger
+
+from vllm_omni.entrypoints.openai.tts_adapters.base import (
+    ARTTSAdapter,
+    DiffusionTTSAdapter,
+    OutputPolicy,
+    PreparedRequest,
+    SpeechServingContext,
+    TTSModelAdapter,
+)
+
+logger = init_logger(__name__)
+
+TTS_ADAPTER_REGISTRY: dict[str, type[TTSModelAdapter]] = {}
+
+
+def register_tts_adapter(cls: type[TTSModelAdapter]) -> type[TTSModelAdapter]:
+    """Class decorator: index ``cls`` under its ``name`` (model-type)."""
+    if cls.name in TTS_ADAPTER_REGISTRY:
+        raise ValueError(
+            f"TTS adapter name {cls.name!r} already registered to "
+            f"{TTS_ADAPTER_REGISTRY[cls.name].__qualname__}; {cls.__qualname__} conflicts."
+        )
+    TTS_ADAPTER_REGISTRY[cls.name] = cls
+    return cls
+
+
+def all_tts_model_types() -> frozenset[str]:
+    """All registered model-type names."""
+    return frozenset(TTS_ADAPTER_REGISTRY)
+
+
+def resolve_adapter(model_type: str | None) -> type[TTSModelAdapter] | None:
+    """Return the adapter for a detected model-type, or ``None``."""
+    if model_type is None:
+        return None
+    return TTS_ADAPTER_REGISTRY.get(model_type)
+
+
+# Import adapter modules for their registration side effects. Keep at the bottom
+# so the registry helpers above are defined first.
+from vllm_omni.entrypoints.openai.tts_adapters import (  # noqa: E402,F401
+    cosyvoice3,
+    covo_audio,
+    fish_speech,
+    glm_tts,
+    higgs_audio_v2,
+    higgs_audio_v3,
+    ming_tts,
+    moss_tts,
+    omnivoice,
+    qwen3_tts,
+    step_audio2,
+    voxcpm2,
+    voxtral,
+)
+
+__all__ = [
+    "ARTTSAdapter",
+    "DiffusionTTSAdapter",
+    "OutputPolicy",
+    "PreparedRequest",
+    "SpeechServingContext",
+    "TTSModelAdapter",
+    "TTS_ADAPTER_REGISTRY",
+    "all_tts_model_types",
+    "register_tts_adapter",
+    "resolve_adapter",
+]
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/base.py b/vllm_omni/entrypoints/openai/tts_adapters/base.py
new file mode 100644
index 000000000..b6d3d3e9f
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/base.py
@@ -0,0 +1,199 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Base contract for per-model TTS serving adapters.
+
+This package factors the per-model ``if self._tts_model_type == ...`` dispatch
+in ``serving_speech.py`` into one adapter class per model. Each adapter owns its
+model's request normalization, validation, prompt/param building, sampling
+overrides, and output policy, so adding a model means writing one adapter file
+instead of editing the shared serving module in ~10 scattered places.
+
+See the RFC for the full design (issue #4327). This is the foundation landed in
+the first migration PR; Qwen3-TTS is the first model routed through it while the
+remaining models stay on the legacy path until individually migrated.
+"""
+
+from abc import ABC, abstractmethod
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, ClassVar
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+_conditioning_cache_salt_fn: "Callable[..., str] | None" = None
+
+
+def conditioning_cache_salt(request: "OpenAICreateSpeechRequest", tts_params: dict) -> str:
+    """Return the conditioning cache salt for ``request`` + ``tts_params``.
+
+    Lazily imports and caches ``serving_speech._conditioning_cache_salt`` on first
+    use: the import is deferred to break the adapters<->serving_speech import
+    cycle, and cached so it resolves once instead of on every ``build()`` call.
+    """
+    global _conditioning_cache_salt_fn
+    if _conditioning_cache_salt_fn is None:
+        from vllm_omni.entrypoints.openai.serving_speech import _conditioning_cache_salt
+
+        _conditioning_cache_salt_fn = _conditioning_cache_salt
+    return _conditioning_cache_salt_fn(request, tts_params)
+
+
+@dataclass
+class OutputPolicy:
+    """How the orchestrator aggregates engine output for a model.
+
+    ``accumulate_nonstreaming`` enables MOSS-style cross-step accumulation in the
+    non-streaming path. Streaming cumulative/delta semantics stay engine-side,
+    keyed by request id, and are unaffected by this flag.
+    """
+
+    accumulate_nonstreaming: bool = False
+
+
+@dataclass
+class PreparedRequest:
+    """Everything the generic orchestrator needs to call ``<engine>.generate()``.
+
+    The fields mirror what ``_prepare_speech_generation`` assembled inline:
+    ``prompt`` is the engine prompt dict, ``tts_params`` the per-model parameter
+    dict, ``model_type`` the discriminator used for logging, and
+    ``output_policy`` controls non-streaming aggregation.
+    """
+
+    prompt: dict[str, Any]
+    tts_params: dict[str, Any] = field(default_factory=dict)
+    model_type: str = "generic"
+    output_policy: OutputPolicy = field(default_factory=OutputPolicy)
+    #: Cross-cutting per-request state the orchestrator still owns (e.g. the
+    #: Qwen3-TTS ref-audio warmup artifact key tracked after ``generate()``).
+    warmup_artifact_key: str | None = None
+
+
+@dataclass
+class SpeechServingContext:
+    """Shared state and helpers an adapter may use.
+
+    During the incremental migration this holds a back-reference to the owning
+    ``OmniOpenAIServingSpeech`` instance (``server``) so adapters can reuse the
+    existing, battle-tested helper implementations (ref-audio resolution,
+    uploaded-speaker handling, prompt-length estimation, speaker storage)
+    without relocating them. Follow-up PRs may narrow this to explicit fields as
+    more models migrate. Exactly one of ``engine_client`` / ``diffusion_engine``
+    is set, matching the deployment's serving backend.
+    """
+
+    server: Any
+    engine_client: Any | None = None
+    diffusion_engine: Any | None = None
+
+
+class TTSModelAdapter(ABC):
+    """Mandatory base class for a TTS model served via ``/v1/audio/speech``.
+
+    One concrete subclass per model, registered by stage key. The serving layer
+    resolves exactly one adapter at startup and dispatches all per-model
+    decisions to it. Adapters reuse shared helpers through ``ctx`` rather than
+    re-implementing them.
+    """
+
+    #: Stable discriminator string (the model-type from detection); registry key.
+    name: ClassVar[str]
+    #: Engine ``model_stage`` key(s) this model uses, for documentation only.
+    stage_keys: ClassVar[frozenset[str]] = frozenset()
+    #: Serving backend: ``"ar"`` (engine_client) or ``"diffusion"``.
+    backend: ClassVar[str] = "ar"
+
+    def __init__(self, ctx: SpeechServingContext) -> None:
+        self.ctx = ctx
+
+    def normalize(self, request: "OpenAICreateSpeechRequest") -> None:
+        """In-place request normalization/mutation (e.g. infer task type,
+        lowercase voice). Default: no-op."""
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        """Return an error string, or ``None`` if the request is valid.
+
+        Should be free of new side effects beyond what ``normalize`` did.
+        """
+        return None
+
+    @abstractmethod
+    async def build(
+        self,
+        request: "OpenAICreateSpeechRequest",
+        sampling_params_list: list,
+        has_inline_ref_audio: bool,
+    ) -> PreparedRequest:
+        """Build the engine prompt + tts_params for this request.
+
+        ``sampling_params_list`` is passed read-only for models (e.g. MOSS) that
+        fold the resolved seed into ``additional_information`` at build time.
+
+        ``has_inline_ref_audio`` is captured by the orchestrator *before*
+        ``validate()`` runs, because ``_apply_uploaded_speaker`` (invoked inside
+        several adapters' ``validate``) sets ``request.ref_audio`` in place.
+        Recomputing it here would misclassify uploaded voices as inline and drop
+        the ``voice_name`` / ``voice_created_at`` metadata.
+        """
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+    ) -> list:
+        """Apply model-specific sampling mutations.
+
+        The orchestrator guarantees the call order:
+        stream-coercion -> extra_params -> THIS -> seed. Default: identity.
+        """
+        return sampling_params_list
+
+
+class ARTTSAdapter(TTSModelAdapter):
+    """Adapter for models served through the AR ``engine_client`` path.
+
+    Covers pure-AR codec models as well as AR-base-LM + diffusion-side-computation
+    hybrids (e.g. VoxCPM2, Ming) whose internal diffusion is invisible to the
+    serving layer.
+    """
+
+    backend: ClassVar[str] = "ar"
+
+
+class DiffusionTTSAdapter(TTSModelAdapter):
+    """Adapter for pure-diffusion pipelines served through the diffusion engine.
+
+    In scope today: OmniVoice. Bridges to the diffusion pipeline parameter
+    contract (#3572) when present; see ``extra_body_params``.
+    """
+
+    backend: ClassVar[str] = "diffusion"
+
+    #: Backing diffusion pipeline class (for EXTRA_BODY_PARAMS lookup).
+    pipeline_cls: ClassVar[type | None] = None
+
+    @classmethod
+    def extra_body_params(cls) -> frozenset[str]:
+        """Fallback-safe access to the pipeline's declared body params.
+
+        Returns the pipeline's ``EXTRA_BODY_PARAMS`` if the #3572 contract is
+        present, else an empty frozenset (the adapter then uses its own inline
+        parameter logic).
+        """
+        params = getattr(cls.pipeline_cls, "EXTRA_BODY_PARAMS", None)
+        return frozenset(params) if params is not None else frozenset()
+
+
+# Re-exported here to avoid import cycles at call sites.
+__all__ = [
+    "ARTTSAdapter",
+    "DiffusionTTSAdapter",
+    "OutputPolicy",
+    "PreparedRequest",
+    "SpeechServingContext",
+    "TTSModelAdapter",
+]
+
+# Type alias kept for readability in adapter signatures.
+NormalizeFn = Callable[["OpenAICreateSpeechRequest"], None]
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/cosyvoice3.py b/vllm_omni/entrypoints/openai/tts_adapters/cosyvoice3.py
new file mode 100644
index 000000000..6a7a89702
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/cosyvoice3.py
@@ -0,0 +1,30 @@
+# SPDX-License-Identifier: Apache-2.0
+"""CosyVoice3 TTS serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class CosyVoice3Adapter(ARTTSAdapter):
+    stage_keys = frozenset({"cosyvoice3_talker"})
+    name = "cosyvoice3"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        err = self.ctx.server._apply_uploaded_speaker(request)
+        if err:
+            return err
+        return self.ctx.server._validate_cosyvoice3_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt = await self.ctx.server._build_cosyvoice3_prompt(request, has_inline_ref_audio=has_inline_ref_audio)
+        # NOTE: CosyVoice3 dynamic-token sampling stays in the orchestrator tail
+        # (keyed on _tts_model_type) during this incremental migration.
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="cosyvoice3")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/covo_audio.py b/vllm_omni/entrypoints/openai/tts_adapters/covo_audio.py
new file mode 100644
index 000000000..f98f89619
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/covo_audio.py
@@ -0,0 +1,22 @@
+# SPDX-License-Identifier: Apache-2.0
+"""CoVo-Audio serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class CovoAudioAdapter(ARTTSAdapter):
+    stage_keys = frozenset({"fused_thinker_talker"})
+    name = "covo_audio"
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt = self.ctx.server._build_covo_audio_prompt(request)
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="covo_audio")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/fish_speech.py b/vllm_omni/entrypoints/openai/tts_adapters/fish_speech.py
new file mode 100644
index 000000000..d216117b7
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/fish_speech.py
@@ -0,0 +1,37 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Fish Speech serving adapter (retires the legacy ``_is_fish_speech`` flag)."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, conditioning_cache_salt
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class FishSpeechAdapter(ARTTSAdapter):
+    stage_keys = frozenset({"fish_speech_slow_ar"})
+    name = "fish_tts"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        err = self.ctx.server._apply_uploaded_speaker(request)
+        if err:
+            return err
+        return self.ctx.server._validate_fish_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        server = self.ctx.server
+        ref_audio_data = None
+        if request.ref_audio is not None:
+            wav_list, sr = await server._resolve_ref_audio(request.ref_audio)
+            ref_audio_data = (wav_list, sr)
+        prompt = await server._build_fish_speech_prompt_async(
+            request, ref_audio_data=ref_audio_data, has_inline_ref_audio=has_inline_ref_audio
+        )
+        tts_params = {}
+        prompt["cache_salt"] = conditioning_cache_salt(request, tts_params)
+        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type="fish_speech")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/glm_tts.py b/vllm_omni/entrypoints/openai/tts_adapters/glm_tts.py
new file mode 100644
index 000000000..e9e8700e6
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/glm_tts.py
@@ -0,0 +1,30 @@
+# SPDX-License-Identifier: Apache-2.0
+"""GLM-TTS serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class GlmTTSAdapter(ARTTSAdapter):
+    stage_keys = frozenset({"glm_tts"})
+    name = "glm_tts"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        err = self.ctx.server._apply_uploaded_speaker(request)
+        if err:
+            return err
+        return self.ctx.server._validate_glm_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt = await self.ctx.server._build_glm_tts_prompt(request, has_inline_ref_audio=has_inline_ref_audio)
+        # GLM-TTS dynamic-token sampling stays in the orchestrator tail
+        # (keyed on _tts_model_type) during this incremental migration.
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="glm_tts")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/higgs_audio_v2.py b/vllm_omni/entrypoints/openai/tts_adapters/higgs_audio_v2.py
new file mode 100644
index 000000000..606b4bca1
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/higgs_audio_v2.py
@@ -0,0 +1,35 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Higgs-Audio v2 serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class HiggsAudioV2Adapter(ARTTSAdapter):
+    stage_keys = frozenset({"higgs_audio_v2"})
+    name = "higgs_audio_v2"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        err = self.ctx.server._apply_uploaded_speaker(request)
+        if err:
+            return err
+        return self.ctx.server._validate_higgs_audio_v2_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        server = self.ctx.server
+        prompt = await server._build_higgs_audio_v2_params(request)
+        if request.voice:
+            voice_lower = request.voice.lower()
+            if voice_lower in server.uploaded_speakers and not has_inline_ref_audio:
+                additional = prompt.setdefault("additional_information", {})
+                additional["voice_name"] = voice_lower
+                additional["voice_created_at"] = server._voice_created_at(voice_lower)
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="higgs_audio_v2")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/higgs_audio_v3.py b/vllm_omni/entrypoints/openai/tts_adapters/higgs_audio_v3.py
new file mode 100644
index 000000000..dde5e0815
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/higgs_audio_v3.py
@@ -0,0 +1,28 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Higgs-Audio v3 serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class HiggsAudioV3Adapter(ARTTSAdapter):
+    stage_keys = frozenset({"higgs_audio_v3"})
+    name = "higgs_audio_v3"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        err = self.ctx.server._apply_uploaded_speaker(request)
+        if err:
+            return err
+        return self.ctx.server._validate_higgs_audio_v3_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt = await self.ctx.server._build_higgs_audio_v3_params(request)
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="higgs_audio_v3")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/ming_tts.py b/vllm_omni/entrypoints/openai/tts_adapters/ming_tts.py
new file mode 100644
index 000000000..37318a521
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/ming_tts.py
@@ -0,0 +1,44 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Ming-TTS (dense) serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class MingTTSAdapter(ARTTSAdapter):
+    # Detected by model_arch (MingTTSForConditionalGeneration), not stage key.
+    name = "ming_tts"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        return self.ctx.server._validate_ming_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        server = self.ctx.server
+        ref_audio_source = request.ref_audio
+        voice_lower = request.voice.lower() if isinstance(request.voice, str) else None
+        if ref_audio_source is None and voice_lower in server.uploaded_speakers:
+            ref_audio_source = server._get_uploaded_audio_data(request.voice)
+            if request.ref_text is None:
+                request.ref_text = server.uploaded_speakers[voice_lower].get("ref_text")
+        ref_audio_data = None
+        if isinstance(ref_audio_source, list):
+            ref_audio_data = await server._resolve_ref_audio_many(ref_audio_source)
+            if request.speaker_embedding is None:
+                request.speaker_embedding = server._extract_ming_speaker_embeddings_from_ref_audio(ref_audio_data)
+        elif ref_audio_source is not None and isinstance(ref_audio_source, str):
+            wav_list, sr = await server._resolve_ref_audio(ref_audio_source)
+            ref_audio_data = (wav_list, sr)
+            if request.speaker_embedding is None:
+                request.speaker_embedding = server._extract_ming_speaker_embeddings_from_ref_audio([ref_audio_data])[0]
+        prompt = server._build_ming_dense_prompt(request, ref_audio_data=ref_audio_data)
+        tts_params = prompt.get("additional_information", {})
+        # Ming stop-token / max_tokens sampling stays in the orchestrator tail.
+        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type="ming_tts")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/moss_tts.py b/vllm_omni/entrypoints/openai/tts_adapters/moss_tts.py
new file mode 100644
index 000000000..62d0c6568
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/moss_tts.py
@@ -0,0 +1,58 @@
+# SPDX-License-Identifier: Apache-2.0
+"""MOSS-TTS serving adapters (Nano + full family).
+
+Both variants share the same build/validate flow (``_build_moss_tts_params``
+handles each); they are registered under distinct model-type names.
+"""
+
+from typing import TYPE_CHECKING
+
+from vllm.inputs import tokens_input
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, conditioning_cache_salt
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+class _MossTTSAdapterBase(ARTTSAdapter):
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        err = self.ctx.server._apply_uploaded_speaker(request)
+        if err:
+            return err
+        return self.ctx.server._validate_moss_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        server = self.ctx.server
+        tts_params = await server._build_moss_tts_params(request)
+        if request.voice:
+            voice_lower = request.voice.lower()
+            if voice_lower in server.uploaded_speakers and not has_inline_ref_audio:
+                tts_params["voice_name"] = [voice_lower]
+                tts_params["voice_created_at"] = [server._voice_created_at(voice_lower)]
+        # MOSS reads the resolved seed at build time (it samples internally).
+        if sampling_params_list and getattr(sampling_params_list[0], "seed", None) is not None:
+            tts_params["seed"] = [sampling_params_list[0].seed]
+        if isinstance(tts_params.get("prompt_token_ids"), list):
+            prompt_token_ids = tts_params.pop("prompt_token_ids")
+            prompt = tokens_input(prompt_token_ids=prompt_token_ids)
+        else:
+            prompt = tokens_input(prompt_token_ids=[1])
+        prompt["additional_information"] = tts_params
+        prompt["cache_salt"] = conditioning_cache_salt(request, tts_params)
+        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type=self.name)
+
+
+@register_tts_adapter
+class MossTTSNanoAdapter(_MossTTSAdapterBase):
+    stage_keys = frozenset({"moss_tts_nano"})
+    name = "moss_tts_nano"
+
+
+@register_tts_adapter
+class MossTTSAdapter(_MossTTSAdapterBase):
+    stage_keys = frozenset({"moss_tts", "moss_tts_codec"})
+    name = "moss_tts"
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/omnivoice.py b/vllm_omni/entrypoints/openai/tts_adapters/omnivoice.py
new file mode 100644
index 000000000..aa3ddd3b4
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/omnivoice.py
@@ -0,0 +1,49 @@
+# SPDX-License-Identifier: Apache-2.0
+"""OmniVoice serving adapter (AR engine_client path).
+
+OmniVoice can also be served through the pure-diffusion engine via
+``for_diffusion`` (``_create_diffusion_speech``); unifying that path is a
+follow-up. This adapter covers the AR-stage deployment.
+"""
+
+from typing import TYPE_CHECKING
+
+import numpy as np
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class OmniVoiceAdapter(ARTTSAdapter):
+    stage_keys = frozenset({"omnivoice_generator"})
+    name = "omnivoice"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        if not request.input or not request.input.strip():
+            return "Input text cannot be empty"
+        return self.ctx.server._apply_uploaded_speaker(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        server = self.ctx.server
+        prompt: dict = {"input": request.input}
+        if request.ref_audio:
+            wav, sr = await server._resolve_ref_audio(request.ref_audio)
+            prompt["ref_audio"] = (np.asarray(wav, dtype=np.float32), sr)
+        if request.ref_text:
+            prompt["ref_text"] = request.ref_text
+        if request.voice:
+            voice_lower = request.voice.lower()
+            if voice_lower in server.uploaded_speakers and not has_inline_ref_audio:
+                prompt["voice_name"] = voice_lower
+                prompt["voice_created_at"] = server._voice_created_at(voice_lower)
+        if request.language:
+            prompt["lang"] = request.language
+        if request.instructions:
+            prompt["instruct"] = request.instructions
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="omnivoice")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/qwen3_tts.py b/vllm_omni/entrypoints/openai/tts_adapters/qwen3_tts.py
new file mode 100644
index 000000000..0f62a169c
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/qwen3_tts.py
@@ -0,0 +1,56 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Qwen3-TTS serving adapter.
+
+First model migrated onto the adapter framework (RFC #4327). It is the dispatch
+surface for Qwen3-TTS: the serving orchestrator routes prompt/param building
+through ``build()`` instead of the inline ``else`` branch in
+``_prepare_speech_generation``.
+
+To stay behaviour-identical and avoid divergence from the batch path (which
+shares ``_validate_qwen_tts_request`` and ``_build_tts_params``), the adapter
+reuses the single-source helper implementations on the serving instance through
+``ctx.server`` rather than copying them. Subsequent PRs relocate those bodies as
+more models migrate and the batch path also routes through adapters.
+"""
+
+from typing import TYPE_CHECKING
+
+from vllm.logger import init_logger
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+logger = init_logger(__name__)
+
+
+@register_tts_adapter
+class Qwen3TTSAdapter(ARTTSAdapter):
+    """Adapter for Qwen3-TTS (AR ``engine_client`` backend)."""
+
+    stage_keys = frozenset({"qwen3_tts"})
+    name = "qwen3_tts"
+
+    def normalize(self, request: "OpenAICreateSpeechRequest") -> None:
+        """Qwen3-TTS normalization (Base-task inference, voice lowercasing) is
+        performed inside ``validate`` today; kept fused for a strict behaviour
+        match. See ``_validate_qwen_tts_request``."""
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        # Go through the shared dispatcher (which routes to
+        # _validate_qwen_tts_request for this model-type) rather than the leaf
+        # validator directly, matching the pre-adapter qwen3 path.
+        return self.ctx.server._validate_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt, tts_params, warmup_key = await self.ctx.server._build_qwen3_tts_request(request)
+        return PreparedRequest(
+            prompt=prompt,
+            tts_params=tts_params,
+            model_type=tts_params.get("task_type", ["unknown"])[0],
+            warmup_artifact_key=warmup_key,
+        )
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/step_audio2.py b/vllm_omni/entrypoints/openai/tts_adapters/step_audio2.py
new file mode 100644
index 000000000..dd1365204
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/step_audio2.py
@@ -0,0 +1,33 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Step-Audio2 serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class StepAudio2Adapter(ARTTSAdapter):
+    """Adapter for Step-Audio2 (AR ``engine_client`` backend).
+
+    Step-Audio2 runs a single thinker stage that emits audio tokens after a
+    ``<tts_start>`` marker; prompt building is a synchronous chat-template
+    construction with no uploaded-voice / ref-audio handling.
+    """
+
+    stage_keys = frozenset({"step_audio2_thinker"})
+    name = "step_audio2"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        # Dispatcher routes to the step_audio2 case (non-empty input check).
+        return self.ctx.server._validate_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt = self.ctx.server._build_step_audio2_prompt(request)
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="step_audio2")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/voxcpm2.py b/vllm_omni/entrypoints/openai/tts_adapters/voxcpm2.py
new file mode 100644
index 000000000..221e2f64b
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/voxcpm2.py
@@ -0,0 +1,49 @@
+# SPDX-License-Identifier: Apache-2.0
+"""VoxCPM2 serving adapter (AR base-LM + diffusion side-computation)."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class VoxCPM2Adapter(ARTTSAdapter):
+    """VoxCPM2 shares ``latent_generator`` with VoxCPM; selected when no ``vae``
+    stage is present (and/or via ``model_arch``)."""
+
+    stage_keys = frozenset({"latent_generator"})
+    name = "voxcpm2"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        return self.ctx.server._validate_voxcpm2_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        server = self.ctx.server
+        # VoxCPM2 needs the raw waveform tuple for prefill-length accounting, so
+        # it loads uploaded audio directly rather than via _apply_uploaded_speaker.
+        uploaded_ref = None
+        if request.voice:
+            voice_lower = request.voice.lower()
+            if voice_lower in server.uploaded_speakers and not has_inline_ref_audio:
+                if server.uploaded_speakers[voice_lower].get("embedding_source") == "direct":
+                    raise ValueError(
+                        f"Uploaded voice '{request.voice}' uses a speaker embedding (Qwen3-only). "
+                        f"Re-upload with an audio file for VoxCPM2."
+                    )
+                if request.ref_audio is None:
+                    uploaded_ref = server._load_uploaded_audio(voice_lower)
+        prompt = await server._build_voxcpm2_prompt(request, uploaded_ref=uploaded_ref)
+        tts_params = {}
+        if request.voice:
+            voice_lower = request.voice.lower()
+            if voice_lower in server.uploaded_speakers or voice_lower in server.precomputed_speakers:
+                additional = prompt.setdefault("additional_information", {})
+                additional["voice_name"] = voice_lower
+                additional["voice_created_at"] = server._voice_created_at(voice_lower)
+        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type="voxcpm2")
diff --git a/vllm_omni/entrypoints/openai/tts_adapters/voxtral.py b/vllm_omni/entrypoints/openai/tts_adapters/voxtral.py
new file mode 100644
index 000000000..72e18dc84
--- /dev/null
+++ b/vllm_omni/entrypoints/openai/tts_adapters/voxtral.py
@@ -0,0 +1,25 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Voxtral TTS serving adapter."""
+
+from typing import TYPE_CHECKING
+
+from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+
+if TYPE_CHECKING:
+    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
+
+
+@register_tts_adapter
+class VoxtralTTSAdapter(ARTTSAdapter):
+    stage_keys = frozenset({"audio_generation"})
+    name = "voxtral_tts"
+
+    def validate(self, request: "OpenAICreateSpeechRequest") -> str | None:
+        return self.ctx.server._validate_voxtral_tts_request(request)
+
+    async def build(
+        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
+    ) -> PreparedRequest:
+        prompt = await self.ctx.server._build_voxtral_prompt_async(request)
+        return PreparedRequest(prompt=prompt, tts_params={}, model_type="voxtral_tts")
diff --git a/vllm_omni/model_executor/models/moss_tts_nano/modeling_moss_tts_nano.py b/vllm_omni/model_executor/models/moss_tts_nano/modeling_moss_tts_nano.py
index 88352ec85..11aeada9e 100644
--- a/vllm_omni/model_executor/models/moss_tts_nano/modeling_moss_tts_nano.py
+++ b/vllm_omni/model_executor/models/moss_tts_nano/modeling_moss_tts_nano.py
@@ -21,6 +21,7 @@ to avoid OOM from vLLM's post-init KV-cache profiling. Mirrors qwen3_tts
 
 from __future__ import annotations
 
+import contextlib
 import tempfile
 import threading
 from collections.abc import Iterable
@@ -38,6 +39,40 @@ from vllm_omni.model_executor.models.utils import reinit_rotary_inv_freq
 logger = init_logger(__name__)
 
 
+@contextlib.contextmanager
+def _hf_load_without_tp_warmup():
+    """Work around a transformers 5.8.x crash loading some trust_remote_code LMs.
+
+    ``caching_allocator_warmup`` does ``len(model._tp_plan)`` when
+    ``torch.distributed`` is initialized, but the MOSS-TTS-Nano remote model
+    leaves ``_tp_plan`` as ``None`` (it survives ``PreTrainedModel.__init__``),
+    so the warmup raises ``TypeError: object of type 'NoneType' has no len()``.
+    vLLM initializes ``torch.distributed`` even at TP=1, so the crash fires.
+
+    MOSS-TTS-Nano runs single-GPU (no tensor parallelism), so forcing the
+    warmup down its non-distributed branch is byte-for-byte equivalent and
+    sidesteps the ``None`` plan. Scoped to the wrapped ``from_pretrained`` call.
+
+    Targets the private ``transformers.modeling_utils._is_torch_distributed_initialized``.
+    Guarded by ``hasattr`` so that if a future transformers renames or removes it
+    (and presumably the crash with it), we no-op instead of raising AttributeError.
+    """
+    import transformers.modeling_utils as _tmu
+
+    if not hasattr(_tmu, "_is_torch_distributed_initialized"):
+        # Internal hook gone (transformers refactor); assume the upstream crash
+        # is gone too and run the load unpatched.
+        yield
+        return
+
+    original = _tmu._is_torch_distributed_initialized
+    _tmu._is_torch_distributed_initialized = lambda: False
+    try:
+        yield
+    finally:
+        _tmu._is_torch_distributed_initialized = original
+
+
 def _patch_torchaudio_load() -> None:
     """Patch torchaudio.load to use soundfile if torchcodec is unavailable."""
     try:
@@ -158,11 +193,12 @@ class MossTTSNanoForGeneration(nn.Module):
         logger.info("Loading MOSS-TTS-Nano LM from %s (dtype=%s)", self.model_path, tts_dtype)
         from transformers import AutoModelForCausalLM
 
-        lm = AutoModelForCausalLM.from_pretrained(
-            self.model_path,
-            trust_remote_code=True,
-            torch_dtype=tts_dtype,
-        )
+        with _hf_load_without_tp_warmup():
+            lm = AutoModelForCausalLM.from_pretrained(
+                self.model_path,
+                trust_remote_code=True,
+                torch_dtype=tts_dtype,
+            )
         if device.type == "cuda":
             try:
                 import flash_attn  # noqa: F401
@@ -197,11 +233,12 @@ class MossTTSNanoForGeneration(nn.Module):
         logger.info("Loading MOSS-Audio-Tokenizer-Nano from %s", codec_path)
         from transformers import AutoModel
 
-        audio_tokenizer = AutoModel.from_pretrained(
-            codec_path,
-            trust_remote_code=True,
-            torch_dtype=torch.float32,
-        )
+        with _hf_load_without_tp_warmup():
+            audio_tokenizer = AutoModel.from_pretrained(
+                codec_path,
+                trust_remote_code=True,
+                torch_dtype=torch.float32,
+            )
         audio_tokenizer.to(device=device)
         audio_tokenizer.eval()
         self._audio_tokenizer: nn.Module = audio_tokenizer