diff --git a/tests/engine/test_duplex_orchestrator.py b/tests/engine/test_duplex_orchestrator.py index 1bffda919..2b0c646e6 100644 --- a/tests/engine/test_duplex_orchestrator.py +++ b/tests/engine/test_duplex_orchestrator.py @@ -307,6 +307,24 @@ async def test_append_submits_the_resumable_stage0_request_and_counts_it_running assert counter.value == 0 +@pytest.mark.asyncio +async def test_resumable_append_prompt_does_not_reach_prewarmed_stages() -> None: + # A resumable append carries Stage0's duplex buffer. The async-chunk + # prewarm copies ``request_state.prompt`` into downstream placeholders, + # so that buffer must not end up there (#7962). + orchestrator, clients, rpc_q, _ = _build(stages=2) + orchestrator.async_chunk = True + orchestrator._stage_receives_async_chunks = lambda stage_id: stage_id > 0 # type: ignore[method-assign] + await _open(orchestrator, rpc_q) + + await _submit(orchestrator, _append_audio()) + + assert clients[0].add_request_calls[0][0].model_intermediate_buffer + assert len(clients[1].add_request_calls) == 1 + assert clients[1].add_request_calls[0][0].model_intermediate_buffer is None + await orchestrator.session_manager.shutdown() + + @pytest.mark.asyncio async def test_session_update_refreshes_the_next_append_sampling_params() -> None: orchestrator, clients, rpc_q, _ = _build() diff --git a/vllm_omni/engine/duplex_orchestrator.py b/vllm_omni/engine/duplex_orchestrator.py index f3d1472aa..e0b18061e 100644 --- a/vllm_omni/engine/duplex_orchestrator.py +++ b/vllm_omni/engine/duplex_orchestrator.py @@ -357,9 +357,6 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): if not isinstance(request_state, DuplexOrchestratorRequestState): raise RuntimeError(f"duplex request was not preregistered: {context.request_id}") request_state.streaming.enabled = submission.resumable - # Keep raw Stage0 prompt (additional_information / multi_modal_data) for - # stage input processors via process_engine_inputs. - request_state.prompt = dict(submission.prompt) if submission.resumable: request = build_engine_core_request_from_tokens( request_id=context.request_id, @@ -369,6 +366,11 @@ class DuplexOrchestrator(Orchestrator, DuplexStagePort): resumable=True, ) else: + # Keep raw Stage0 prompt (additional_information / multi_modal_data) for + # stage input processors via process_engine_inputs. Resumable requests + # leave it unset: their prompt is one append with Stage0's duplex + # buffer, which the async-chunk prewarm would copy downstream. + request_state.prompt = dict(submission.prompt) # Use the ordinary multimodal input processor for turn-model plugins. # Its CPU preprocessing runs off the session/orchestrator event loop. request = await asyncio.to_thread(