[Model][JoyAI-VL-Interaction] Add native multi-stage pipeline (#5352)

Signed-off-by: Shuolei Wang <shuoleiwang123@gmail.com>

recipes/JD/JoyAI-VL-Interaction.md

@@ -219,3 +219,51 @@ the audio-track caveat.
   5 chunks, `key_frames=0` = summarize all chunk frames) are aligned to the JoyVL
   reference adapter so per-tick behavior matches the released model; the framework only
   supplies the serving structure.
+
+## Native multi-stage pipeline (experimental)
+
+An experimental native speech path is available alongside the existing Day-0 serving path:
+
+```text
+one request with image/video and text
+  -> JoyAI complete action text
+  -> Qwen3-TTS Talker
+  -> Code2Wav
+  -> 24 kHz audio
+```
+
+It runs behind one `vllm serve --omni` endpoint:
+
+```bash
+vllm serve jdopensource/JoyAI-VL-Interaction-Preview --omni \
+  --deploy-config vllm_omni/deploy/joyai_vl_interaction.yaml \
+  --port 8091
+```
+
+With `modalities: ["text", "audio"]`, its behavior is:
+
+| JoyAI action | Native output |
+|---|---|
+| `</response> <text>` | action text + speech for `<text>` |
+| `</silence>` | action text + empty audio output (zero samples) |
+| `</response> <note> </delegation> <question>` | action text + speech for `<note>` only |
+
+Native speech supports the Qwen3-TTS `CustomVoice` task. The request-level `voice` and
+`language` fields select the speaker and language; when omitted, they default to
+`Vivian` and `Auto`. Values supplied as `tts_speaker` and `tts_language` inside
+`additional_information` take priority.
+
+This experimental native multi-stage pipeline is request-scoped, stateless, and
+all-sync. It currently does not include:
+
+- continuous frame sessions, persistent standing instructions, or session memory;
+- audio input / ASR, full-duplex interaction, or barge-in;
+- background Agent execution for delegated questions;
+- async-chunk support from Talker to Code2Wav;
+- integration with the existing WebUI.
+
+For actions with spoken text, the Talker starts only after the complete JoyAI action is
+available, and Code2Wav starts only after the Talker completes. Silence actions skip both
+TTS stages and return an empty audio output with zero samples. All three stages are placed
+on GPU 0 by the provided
+[`deploy config`](../../vllm_omni/deploy/joyai_vl_interaction.yaml).

tests/config/test_omni_config.py

@@ -532,6 +532,17 @@ def test_from_pipeline_config_dispatches_async_chunk_processors_without_mutating
     assert pipeline.get_stage(1).custom_process_input_func is None
 
 
+def test_joyai_code2wav_waits_for_full_payload():
+    config = _from_pipeline_key("joyai_vl_interaction")
+    talker = config.stage_by_id(1)
+    code2wav = config.stage_by_id(2)
+
+    assert talker.custom_process_next_stage_input_func.endswith("talker2code2wav_full_payload")
+    assert code2wav.custom_process_input_func.endswith("talker2code2wav_token_only")
+    assert code2wav.connector_config.async_chunk is False
+    assert code2wav.model_config.requires_full_payload_input is True
+
+
 def test_vllm_omni_stage_config_public_fields_use_typed_stage_realizations():
     assert not hasattr(BaseVllmOmniStageConfig, "from_stage_config")
     assert not hasattr(BaseVllmOmniStageConfig, "to_legacy_stage_config")

tests/engine/test_orchestrator_error_handling.py

@@ -16,7 +16,7 @@ from types import SimpleNamespace
 
 import janus
 import pytest
-from vllm.sampling_params import SamplingParams
+from vllm.sampling_params import RequestOutputKind, SamplingParams
 from vllm.v1.engine.exceptions import EngineDeadError
 from vllm.v1.serial_utils import MsgpackEncoder
 
@@ -33,11 +33,16 @@ from vllm_omni.engine.orchestrator import (
     cleanup_request_artifact_dirs,
 )
 from vllm_omni.engine.stage_pool import StageUnavailableError
+from vllm_omni.errors import OmniClientError
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
+from vllm_omni.model_executor.stage_input_processors import (
+    joyai_vl_interaction as joyai_bridge,
+)
 from vllm_omni.outputs import OmniRequestOutput
 
 from .test_orchestrator import (
     FakeOutputProcessor,
+    FakeRunningCounter,
     FakeStageClient,
     OrchestratorFixture,
     _build_harness,
@@ -45,6 +50,7 @@ from .test_orchestrator import (
     _build_stage_pools,
     _engine_core_outputs,
     _enqueue_add_request,
+    _get_output_message,
     _wait_for,
 )
 
@@ -935,6 +941,139 @@ async def test_diffusion_client_error_output_propagates_status_code(orchestrator
         orchestrator_fixture.thread.join(timeout=5)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("event_driven", [False, True], ids=["legacy", "event-driven"])
+@pytest.mark.parametrize("first_speaker", ["Ryan", "not-a-real-speaker"])
+async def test_joyai_tts_request_validation_keeps_orchestrator_running(
+    orchestrator_factory,
+    monkeypatch: pytest.MonkeyPatch,
+    first_speaker: str,
+    event_driven: bool,
+) -> None:
+    """Check that the same orchestrator can handle the next valid request."""
+    monkeypatch.setenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", "1" if event_driven else "0")
+
+    # Use a fixed token count so the test does not need tokenizer files.
+    class FakeTokenizer:
+        def encode(self, _text: str, *, add_special_tokens: bool) -> list[int]:
+            return list(range(12))
+
+    talker_model_config = SimpleNamespace(
+        max_model_len=64,
+        hf_config=SimpleNamespace(
+            talker_config=SimpleNamespace(spk_id={"vivian": 0, "ryan": 1}, codec_language_id={}, spk_is_dialect={})
+        ),
+    )
+    monkeypatch.setattr(joyai_bridge, "cached_tokenizer_from_config", lambda _: FakeTokenizer())
+
+    # Use the real bridge so speaker validation is part of the test.
+    class JoyAITalkerStage(FakeStageClient):
+        def process_engine_inputs(self, source_outputs, prompt=None, streaming_context=None):
+            self.bridge_inputs = joyai_bridge.joyai_action_to_tts(
+                source_outputs, prompt, self.requires_multimodal_data, target_model_config=talker_model_config
+            )
+            return self.bridge_inputs
+
+    stage0 = FakeStageClient(stage_type="llm", final_output=False)
+    stage1 = JoyAITalkerStage(stage_type="llm", final_output=True)
+    stage0_processor, stage1_processor = FakeOutputProcessor(), FakeOutputProcessor()
+    fixture = orchestrator_factory(
+        [stage0, stage1],
+        output_processors=[stage0_processor, stage1_processor],
+        stage_vllm_configs=[
+            SimpleNamespace(model_config=SimpleNamespace(max_model_len=64)),
+            SimpleNamespace(model_config=talker_model_config),
+        ],
+    )
+    counter = FakeRunningCounter()
+    fixture.orchestrator._running_counter = counter
+    # With a valid speaker, task_type should be changed from Base to CustomVoice.
+    # An unknown speaker should cause a request error.
+    first_info = {"tts_task_type": "Base", "speaker": "Vivian", "tts_speaker": first_speaker}
+    talker_submissions = 0
+    # Send both requests to the same orchestrator. The second uses the default Vivian voice.
+    for index, (request_id, info) in enumerate((("first", first_info), ("good", {})), start=1):
+        await _enqueue_add_request(
+            fixture,
+            request_id=request_id,
+            prompt=SimpleNamespace(request_id=request_id, prompt_token_ids=[1, 2]),
+            original_prompt={"additional_information": info},
+            sampling_params_list=[_sampling_params(), _sampling_params()],
+            final_stage_id=1,
+        )
+        await _wait_for(lambda: len(stage0.add_request_calls) == index and counter.value == 1)
+        stage0_processor.request_outputs = [_build_request_output(request_id, text="</response> Hello.")]
+        stage0.push_engine_core_outputs(_engine_core_outputs(f"{request_id}-raw", float(index)))
+
+        if request_id == "first" and first_speaker == "not-a-real-speaker":
+            # Reject this request before Talker runs, while keeping the engine alive.
+            error = await _wait_for_error_message(fixture, request_id=request_id)
+            assert error.status_code == 400 and error.error_type == "BadRequestError"
+            assert error.fatal is False and error.stage_id == 1
+            assert "not-a-real-speaker" in error.error
+            assert stage1.add_request_calls == []
+        else:
+            talker_submissions += 1
+            await _wait_for(lambda: len(stage1.add_request_calls) == talker_submissions or fixture.result_future.done())
+            assert not fixture.result_future.done()
+            assert stage1.add_request_calls[-1][0].request_id == request_id
+            metadata = stage1.bridge_inputs[0]["additional_information"]
+            assert metadata["task_type"] == ["CustomVoice"]
+            assert metadata["speaker"] == [first_speaker if request_id == "first" else "Vivian"]
+            # Finish the valid request and check its final output.
+            stage1_processor.request_outputs = [_build_request_output(request_id, text="Completed.")]
+            stage1.push_engine_core_outputs(_engine_core_outputs(f"{request_id}-talker", float(index)))
+            output = await _get_output_message(fixture)
+            assert output.request_id == request_id and output.stage_id == 1
+            assert output.finished is True
+            assert output.engine_outputs.outputs[0].text == "Completed."
+
+        # Wait for full cleanup before sending the next request.
+        await _wait_for(lambda: request_id not in fixture.orchestrator.request_states and counter.value == 0)
+        assert all(pool.get_bound_replica_id(request_id) is None for pool in fixture.orchestrator.stage_pools)
+        assert fixture.thread.is_alive() and not fixture.result_future.done()
+
+
+@pytest.mark.asyncio
+async def test_streaming_input_processor_client_error_does_not_forward_terminal_update() -> None:
+    """Do not send a second update after the first one fails and clears the request."""
+    processor_calls = []
+
+    class FailingStage(FakeStageClient):
+        def process_engine_inputs(self, *args, **kwargs):
+            processor_calls.append((args, kwargs))
+            raise OmniClientError("Input rejected")
+
+    orchestrator, queues = _build_bare_orchestrator(
+        _build_stage_pools([[FakeStageClient()], [FailingStage(final_output=True)]])
+    )
+    state = OrchestratorRequestState(
+        request_id="bad-stream",
+        prompt={"prompt_token_ids": [1]},
+        sampling_params_list=[SamplingParams(output_kind=RequestOutputKind.CUMULATIVE), _sampling_params()],
+        final_stage_id=1,
+    )
+    state.streaming.enabled = True
+    assert state.duplex_identity is None
+    orchestrator.request_states[state.request_id] = state
+    try:
+        await orchestrator._route_output(0, 0, _build_request_output(state.request_id), state, stage_metrics=None)
+
+        assert len(processor_calls) == 1
+        error = queues[1].sync_q.get_nowait()
+        assert isinstance(error, ErrorMessage)
+        assert error.request_id == state.request_id
+        assert error.stage_id == 1
+        assert error.status_code == 400
+        assert error.fatal is False
+        assert queues[1].sync_q.empty()
+        assert state.request_id not in orchestrator.request_states
+        assert all(pool.get_bound_replica_id(state.request_id) is None for pool in orchestrator.stage_pools)
+    finally:
+        for q in queues:
+            q.close()
+
+
 @pytest.mark.asyncio
 async def test_duplex_input_processor_failure_is_request_scoped(orchestrator_factory, monkeypatch) -> None:
     class FailingStage(FakeStageClient):

tests/model_executor/stage_input_processors/test_joyai_vl_interaction.py

@@ -0,0 +1,98 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Tests for JoyAI action-to-TTS conversion."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from vllm_omni.engine.stage_engine_core_client import StageEngineCoreClient
+from vllm_omni.model_executor.stage_input_processors import (
+    joyai_vl_interaction as joyai_bridge,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_joyai_action_to_tts_bridge(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Convert each JoyAI action through the real stage-client entry point."""
+    target_model_config = SimpleNamespace(
+        hf_config=SimpleNamespace(talker_config=SimpleNamespace(spk_id={"vivian": 1, "ryan": 2}))
+    )
+    prompt_length_calls: list[tuple[dict[str, list[str]], object]] = []
+    prompt_lengths = {
+        "Colored squares are visible.": 12,
+        "I am checking this request now.": 18,
+    }
+
+    def compute_prompt_length(metadata, model_config):
+        prompt_length_calls.append((metadata, model_config))
+        return prompt_lengths[metadata["text"][0]]
+
+    monkeypatch.setattr(
+        joyai_bridge,
+        "_compute_talker_prompt_length",
+        compute_prompt_length,
+    )
+
+    client = object.__new__(StageEngineCoreClient)
+    client.custom_process_input_func = joyai_bridge.joyai_action_to_tts
+    client.requires_multimodal_data = False
+    client.vllm_config = SimpleNamespace(model_config=target_model_config)
+
+    cases = [
+        (
+            "</response> Colored squares are visible.",
+            None,
+            {
+                "tts_task_type": "Base",
+                "language": ["Chinese"],
+                "speaker": ["Ryan"],
+            },
+            ("Colored squares are visible.", "Chinese", "Ryan"),
+        ),
+        ("</silence>", None, {}, None),
+        (
+            "",
+            (
+                "</response> I am checking this request now. "
+                "</delegation> Why is the purple elephant dancing beside a volcano?"
+            ),
+            {
+                "language": "Chinese",
+                "speaker": "Ryan",
+                "tts_language": "English",
+                "tts_speaker": "Vivian",
+            },
+            ("I am checking this request now.", "English", "Vivian"),
+        ),
+    ]
+
+    for text, cumulative_text, additional_information, expected in cases:
+        completion = SimpleNamespace(text=text)
+        if cumulative_text is not None:
+            completion.cumulative_text = cumulative_text
+
+        call_count = len(prompt_length_calls)
+        talker_inputs = client.process_engine_inputs(
+            [SimpleNamespace(outputs=[completion])],
+            prompt={"additional_information": additional_information},
+        )
+
+        if expected is None:
+            assert talker_inputs == []
+            assert len(prompt_length_calls) == call_count
+            continue
+
+        spoken_text, language, speaker = expected
+        [talker_input] = talker_inputs
+        metadata = talker_input["additional_information"]
+        assert talker_input["prompt_token_ids"] == [0] * prompt_lengths[spoken_text]
+        assert metadata == {
+            "task_type": ["CustomVoice"],
+            "language": [language],
+            "speaker": [speaker],
+            "instruct": [""],
+            "text": [spoken_text],
+        }
+        assert prompt_length_calls[-1] == (metadata, target_model_config)

vllm_omni/config/pipeline_registry.py

@@ -67,6 +67,9 @@ from vllm_omni.model_executor.models.indextts2.pipeline import (
     INDEXTTS2_PIPELINE,
     INDEXTTS25_PIPELINE,
 )
+from vllm_omni.model_executor.models.joyai_vl_interaction.pipeline import (
+    JOYAI_VL_INTERACTION_PIPELINE,
+)
 from vllm_omni.model_executor.models.lance.pipeline import LANCE_PIPELINE
 from vllm_omni.model_executor.models.lingbot_world.pipeline import LINGBOT_WORLD_PIPELINE
 from vllm_omni.model_executor.models.mammoth_moda2.pipeline import (
@@ -122,6 +125,7 @@ PipelineResolverFunc: TypeAlias = Callable[[PretrainedConfig | None], PipelineCo
 # --- Multi-stage omni pipelines (LLM-centric; audio / video I/O) ---
 OMNI_PIPELINES: dict[str, PipelineConfig | PipelineResolverFunc] = {
     "aura_omni": AURA_OMNI_PIPELINE,
+    "joyai_vl_interaction": JOYAI_VL_INTERACTION_PIPELINE,
     "qwen2_5_omni": QWEN2_5_OMNI_PIPELINE,
     "qwen2_5_omni_thinker_only": QWEN2_5_OMNI_THINKER_ONLY_PIPELINE,
     "personaplex": PERSONAPLEX_PIPELINE,

vllm_omni/deploy/joyai_vl_interaction.yaml

@@ -0,0 +1,79 @@
+# Single-GPU JoyAI-VL-Interaction -> Qwen3-TTS all-sync profile.
+pipeline: joyai_vl_interaction
+async_chunk: false
+trust_remote_code: true
+
+connectors:
+  connector_of_shared_memory:
+    name: SharedMemoryConnector
+    extra:
+      codec_streaming: true
+      connector_get_sleep_s: 0.01
+      connector_get_max_wait_first_chunk: 3000
+      connector_get_max_wait: 300
+      codec_chunk_frames: 25
+      codec_left_context_frames: 72
+      initial_codec_chunk_frames: 1
+      # Qwen3-TTS Code2Wav buckets: 25, 72+1, 72+25, 72+72+25, and 300+25 frames.
+      decode_cudagraph_capture_sizes: [25, 73, 97, 169, 325]
+
+stages:
+  # GPU 0: JoyAI action model.
+  - stage_id: 0
+    model: jdopensource/JoyAI-VL-Interaction-Preview
+    devices: "0"
+    gpu_memory_utilization: 0.5
+    max_num_seqs: 4
+    # Some OmniInteract video prompts contain more than 12K input tokens.
+    max_model_len: 16384
+    max_num_batched_tokens: 16384
+    limit_mm_per_prompt: {image: 256, video: 1}
+    enable_prefix_caching: true
+    default_sampling_params:
+      temperature: 0.8
+      top_p: 0.9
+      top_k: 40
+      max_tokens: 256
+      repetition_penalty: 1.1
+      presence_penalty: 1.5
+
+  # GPU 0: Qwen3-TTS Talker.
+  - stage_id: 1
+    model: Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice
+    devices: "0"
+    gpu_memory_utilization: 0.2
+    max_num_seqs: 8
+    max_model_len: 4096
+    max_num_batched_tokens: 512
+    enable_prefix_caching: false
+    output_connectors:
+      to_stage_2: connector_of_shared_memory
+    default_sampling_params:
+      temperature: 0.9
+      top_k: 50
+      min_tokens: 2
+      max_tokens: 4096
+      repetition_penalty: 1.05
+    subtalker_sampling_params:
+      do_sample: true
+      temperature: 0.9
+      top_k: 50
+      top_p: 1.0
+
+  # GPU 0: Qwen3-TTS Code2Wav.
+  - stage_id: 2
+    model: Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice
+    devices: "0"
+    gpu_memory_utilization: 0.2
+    max_num_seqs: 8
+    max_model_len: 65536
+    max_num_batched_tokens: 65536
+    enable_prefix_caching: false
+    input_connectors:
+      from_stage_1: connector_of_shared_memory
+    default_sampling_params:
+      temperature: 0.0
+      top_p: 1.0
+      top_k: -1
+      max_tokens: 65536
+      repetition_penalty: 1.0

vllm_omni/engine/orchestrator.py

@@ -75,7 +75,7 @@ from vllm_omni.engine.messages import (
 from vllm_omni.engine.orchestrator_monitor import create_orch_monitor, replica_key
 from vllm_omni.engine.serialization import serialize_additional_information
 from vllm_omni.engine.stage_pool import StagePool, StageUnavailableError
-from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE
+from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE, OmniClientError
 from vllm_omni.metrics import definitions as metric_defs
 from vllm_omni.metrics.prometheus import OmniRequestCounter
 from vllm_omni.metrics.stat_logger import OmniPrometheusStatLogger
@@ -1647,9 +1647,11 @@ class Orchestrator:
         stage_id: int,
         error: str,
         *,
+        status_code: int = HTTPStatus.BAD_REQUEST.value,
+        error_type: str = DEFAULT_CLIENT_ERROR_TYPE,
         close_duplex_sessions: bool = False,
     ) -> None:
-        """Fail one request with a non-fatal 400 (bad input, engine survives).
+        """Fail one request with a non-fatal client error (400 by default).
 
         The non-fatal counterpart of `_fail_request_dead_stage`: emits a
         client-error ErrorMessage (default `fatal=False`) so the engine keeps
@@ -1658,8 +1660,8 @@ class Orchestrator:
         await self.output_async_queue.put(
             ErrorMessage(
                 error=error,
-                status_code=HTTPStatus.BAD_REQUEST.value,
-                error_type=DEFAULT_CLIENT_ERROR_TYPE,
+                status_code=status_code,
+                error_type=error_type,
                 request_id=req_id,
                 stage_id=stage_id,
             )
@@ -2073,6 +2075,8 @@ class Orchestrator:
                 )
                 if (
                     req_state.streaming.enabled
+                    # A failed forward may already have cleaned up this request.
+                    and self.request_states.get(req_id) is req_state
                     and finished
                     and not final_only_finished
                     and not self._is_duplex_session_request(req_state)
@@ -2734,6 +2738,16 @@ class Orchestrator:
                 req_state.prompt,
                 streaming_context=req_state.streaming,
             )
+        except OmniClientError as exc:
+            await self._fail_request_client_error(
+                req_id,
+                next_logical,
+                str(exc),
+                status_code=exc.status_code,
+                error_type=exc.error_type,
+                close_duplex_sessions=self._is_duplex_session_request(req_state),
+            )
+            return
         except Exception as exc:
             logger.exception(
                 "[Orchestrator] req=%s process_engine_inputs FAILED for stage-%s",

vllm_omni/engine/stage_engine_core_client.py

@@ -396,36 +396,54 @@ class StageEngineCoreClientBase(StageClientBase):
         and the original prompt.
         """
         if self.custom_process_input_func is not None:
-            signature = inspect.signature(self.custom_process_input_func)
-            extra_kwargs: dict[str, Any] = {}
-            if "next_stage_hf_config" in signature.parameters:
-                # Let a processor size the next stage's prompt from that
-                # stage's model config (e.g. a talker whose engine positions
-                # must cover a speaker-prompt prefill).
-                extra_kwargs["next_stage_hf_config"] = self._stage_hf_config
-            # Match the context parameter by name, including the
-            # underscore-prefixed spelling some processors use (e.g.
-            # MiniCPM-o's ``llm2tts(..., _streaming_context)``), so bridge
-            # state keeps flowing to them.
-            if "streaming_context" in signature.parameters or "_streaming_context" in signature.parameters:
-                return self.custom_process_input_func(
-                    source_outputs,
-                    prompt,
-                    self.requires_multimodal_data,
-                    streaming_context,
-                    **extra_kwargs,
-                )
-            return self.custom_process_input_func(
-                source_outputs,
-                prompt,
-                self.requires_multimodal_data,
-                **extra_kwargs,
-            )
+            return self._call_custom_process_input(source_outputs, prompt, streaming_context)
 
         if not self.engine_input_source:
             raise ValueError(f"engine_input_source empty for stage {self.stage_id}")
         return _default_process_engine_inputs(source_outputs, prompt, self.requires_multimodal_data)
 
+    def _call_custom_process_input(
+        self,
+        source_outputs: list[Any],
+        prompt: Any,
+        streaming_context: Any | None,
+    ) -> list[OmniTokensPrompt]:
+        """Call a stage input processor with its explicitly requested context."""
+        processor = self.custom_process_input_func
+        assert processor is not None
+        signature = inspect.signature(processor)
+        extra_kwargs: dict[str, Any] = {}
+        if "next_stage_hf_config" in signature.parameters:
+            # Let a processor size the next stage's prompt from that
+            # stage's model config (e.g. a talker whose engine positions
+            # must cover a speaker-prompt prefill).
+            extra_kwargs["next_stage_hf_config"] = self._stage_hf_config
+        target_model_config = signature.parameters.get("target_model_config")
+        if target_model_config is not None:
+            # The JoyAI bridge needs the Talker tokenizer and model config
+            # to calculate the exact prompt length.
+            if target_model_config.kind is not inspect.Parameter.KEYWORD_ONLY:
+                raise TypeError("target_model_config must be a keyword-only parameter")
+            extra_kwargs["target_model_config"] = self.vllm_config.model_config
+        # Match the context parameter by name, including the
+        # underscore-prefixed spelling some processors use (e.g.
+        # MiniCPM-o's ``llm2tts(..., _streaming_context)``), so bridge
+        # state keeps flowing to them.
+        if "streaming_context" in signature.parameters or "_streaming_context" in signature.parameters:
+            return processor(
+                source_outputs,
+                prompt,
+                self.requires_multimodal_data,
+                streaming_context,
+                **extra_kwargs,
+            )
+        return processor(
+            source_outputs,
+            prompt,
+            self.requires_multimodal_data,
+            **extra_kwargs,
+        )
+
     async def collective_rpc_async(
         self,
         method: str,

vllm_omni/model_executor/models/joyai_vl_interaction/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""JoyAI-VL-Interaction native pipeline."""

vllm_omni/model_executor/models/joyai_vl_interaction/pipeline.py

@@ -0,0 +1,65 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""JoyAI-VL-Interaction -> Qwen3-TTS native all-sync pipeline."""
+
+from vllm_omni.config.stage_config import (
+    PipelineConfig,
+    StageExecutionType,
+    StagePipelineConfig,
+)
+
+_JOYAI_INPUT_PROCESSOR = "vllm_omni.model_executor.stage_input_processors.joyai_vl_interaction"
+_QWEN3_TTS_INPUT_PROCESSOR = "vllm_omni.model_executor.stage_input_processors.qwen3_tts"
+
+
+JOYAI_VL_INTERACTION_PIPELINE = PipelineConfig(
+    model_type="joyai_vl_interaction",
+    model_arch="Qwen3VLForConditionalGeneration",
+    default_deploy_config_name="joyai_vl_interaction.yaml",
+    stages=(
+        StagePipelineConfig(
+            stage_id=0,
+            model_stage="joyvl",
+            execution_type=StageExecutionType.LLM_AR,
+            input_sources=(),
+            final_output=True,
+            final_output_type="text",
+            owns_tokenizer=True,
+            requires_multimodal_data=True,
+            engine_output_type="text",
+            sampling_constraints={
+                "detokenize": True,
+                "skip_special_tokens": False,
+            },
+        ),
+        StagePipelineConfig(
+            stage_id=1,
+            model_stage="qwen3_tts",
+            execution_type=StageExecutionType.LLM_AR,
+            input_sources=(0,),
+            owns_tokenizer=True,
+            engine_output_type="latent",
+            model_arch="Qwen3TTSTalkerForConditionalGeneration",
+            custom_process_input_func=f"{_JOYAI_INPUT_PROCESSOR}.joyai_action_to_tts",
+            custom_process_next_stage_input_func=f"{_QWEN3_TTS_INPUT_PROCESSOR}.talker2code2wav_full_payload",
+            sampling_constraints={
+                "detokenize": False,
+                "stop_token_ids": [2150],
+            },
+        ),
+        StagePipelineConfig(
+            stage_id=2,
+            model_stage="code2wav",
+            execution_type=StageExecutionType.LLM_GENERATION,
+            input_sources=(1,),
+            final_output=True,
+            final_output_type="audio",
+            engine_output_type="audio",
+            model_arch="Qwen3TTSCode2Wav",
+            sync_process_input_func=f"{_QWEN3_TTS_INPUT_PROCESSOR}.talker2code2wav_token_only",
+            requires_full_payload_input=True,
+            sampling_constraints={"detokenize": True},
+            extras={"tts_args": {"max_instructions_length": 500}},
+        ),
+    ),
+)

vllm_omni/model_executor/stage_input_processors/joyai_vl_interaction.py

@@ -0,0 +1,129 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Build Qwen3-TTS Talker inputs from completed JoyAI actions."""
+
+from typing import Any
+
+from vllm.outputs import RequestOutput
+from vllm.tokenizers import cached_tokenizer_from_config
+
+from vllm_omni.errors import OmniClientError
+from vllm_omni.experimental.fullduplex.joyvl.decision.output_parser import (
+    parse_action,
+)
+from vllm_omni.inputs.data import OmniTokensPrompt
+from vllm_omni.model_executor.models.qwen3_tts.prompt_embeds_builder import (
+    Qwen3TTSPromptEmbedsBuilder,
+    first_value,
+)
+
+# Match the standalone Qwen3-TTS default attributes
+_DEFAULT_TTS_TASK_TYPE = "CustomVoice"
+_DEFAULT_TTS_LANGUAGE = "Auto"
+_DEFAULT_TTS_SPEAKER = "Vivian"
+
+
+def _extract_completed_action_text(joyai_output: RequestOutput) -> str:
+    """Return the final JoyAI action text, using cumulative text when available."""
+    completion = joyai_output.outputs[0]
+    action_text = getattr(completion, "cumulative_text", None) or completion.text
+    return action_text if isinstance(action_text, str) else ""
+
+
+def _build_tts_metadata(request_prompt: object, spoken_text: str) -> dict[str, list[str]]:
+    """Build Qwen3-TTS metadata for the text JoyAI chose to speak."""
+    prompt_dict = request_prompt if isinstance(request_prompt, dict) else {}
+    raw_additional_info = prompt_dict.get("additional_information")
+    additional_info = raw_additional_info if isinstance(raw_additional_info, dict) else {}
+
+    # The native pipeline loads CustomVoice checkpoints for both TTS stages.
+    # Keep client metadata from selecting an incompatible Qwen3-TTS task.
+    task_type = _DEFAULT_TTS_TASK_TYPE
+    request_language = first_value(additional_info.get("language"), _DEFAULT_TTS_LANGUAGE)
+    language = first_value(additional_info.get("tts_language"), request_language)
+    request_speaker = first_value(additional_info.get("speaker"), _DEFAULT_TTS_SPEAKER)
+    speaker = first_value(additional_info.get("tts_speaker"), request_speaker)
+    instruction = first_value(additional_info.get("tts_instruct"), "")
+
+    return {
+        "task_type": [str(task_type)],
+        "language": [str(language)],
+        "speaker": [str(speaker)],
+        "instruct": [str(instruction)],
+        "text": [spoken_text],
+    }
+
+
+def _validate_talker_speaker(tts_metadata: dict[str, list[str]], talker_model_config: Any) -> None:
+    """Validate the selected voice using the destination Talker's speaker table."""
+    talker_config = getattr(talker_model_config.hf_config, "talker_config", None)
+    if talker_config is None:
+        raise ValueError("The target stage is not a Qwen3-TTS Talker.")
+
+    # Match CustomVoice prompt construction, including case and whitespace.
+    speaker = str(first_value(tts_metadata.get("speaker"), "")).lower().strip()
+    speakers = {name.lower() for name in (getattr(talker_config, "spk_id", None) or {})}
+    if not speaker or speaker not in speakers:
+        raise OmniClientError(f"Unsupported speaker: {speaker!r}")
+
+
+def _compute_talker_prompt_length(
+    tts_metadata: dict[str, list[str]],
+    talker_model_config: Any,
+) -> int:
+    """Compute the exact number of prompt positions used by the Talker."""
+    task_type = str(first_value(tts_metadata.get("task_type"), _DEFAULT_TTS_TASK_TYPE))
+    if task_type != _DEFAULT_TTS_TASK_TYPE:
+        raise ValueError("JoyAI native TTS supports only CustomVoice.")
+
+    talker_tokenizer = cached_tokenizer_from_config(talker_model_config)
+    if talker_tokenizer is None:
+        raise ValueError("The Qwen3-TTS Talker must have an initialized tokenizer.")
+
+    talker_config = getattr(talker_model_config.hf_config, "talker_config", None)
+    if talker_config is None:
+        raise ValueError("The target stage is not a Qwen3-TTS Talker.")
+
+    def tokenize_prompt(text: str) -> list[int]:
+        return talker_tokenizer.encode(text, add_special_tokens=True)
+
+    return Qwen3TTSPromptEmbedsBuilder.estimate_prompt_len_from_additional_information(
+        additional_information=tts_metadata,
+        task_type=task_type,
+        tokenize_prompt=tokenize_prompt,
+        codec_language_id=getattr(talker_config, "codec_language_id", None),
+        spk_is_dialect=getattr(talker_config, "spk_is_dialect", None),
+    )
+
+
+def joyai_action_to_tts(
+    source_outputs: list[RequestOutput],
+    prompt: object | None = None,
+    requires_multimodal_data: bool = False,
+    *,
+    target_model_config: Any,
+) -> list[OmniTokensPrompt]:
+    """Build Talker inputs for JoyAI actions that contain text to speak."""
+    del requires_multimodal_data
+
+    request_prompts = prompt if isinstance(prompt, list) else [prompt] * len(source_outputs)
+    talker_inputs: list[OmniTokensPrompt] = []
+    for request_index, joyai_output in enumerate(source_outputs):
+        parsed_action = parse_action(_extract_completed_action_text(joyai_output))
+        if not parsed_action.spoke or not parsed_action.text:
+            continue
+
+        request_prompt = request_prompts[request_index] if request_index < len(request_prompts) else None
+        tts_metadata = _build_tts_metadata(request_prompt, parsed_action.text)
+        _validate_talker_speaker(tts_metadata, target_model_config)
+        talker_prompt_length = _compute_talker_prompt_length(tts_metadata, target_model_config)
+        talker_inputs.append(
+            OmniTokensPrompt(
+                # OmniTokensPrompt reserves vllm scheduler positions with zero IDs;
+                # the Talker builds the actual prompt from tts_metadata.
+                prompt_token_ids=[0] * talker_prompt_length,
+                additional_information=tts_metadata,
+            )
+        )
+
+    return talker_inputs