[Model] Integrate AURA into Full-Duplex Runtime (#7633)

tests/engine/duplex/test_aura_resumable_hooks.py

@@ -0,0 +1,725 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Public duplex hook: resumable defaults True; ephemeral ids when False."""
+
+from __future__ import annotations
+
+import pytest
+from vllm.sampling_params import SamplingParams
+
+from vllm_omni.engine.duplex.config import DuplexCapabilities
+from vllm_omni.engine.duplex.contracts import (
+    DuplexFence,
+    DuplexStageRequestContext,
+    DuplexStageSubmission,
+    duplex_ephemeral_stage_request_id,
+)
+from vllm_omni.engine.duplex.session.manager import DuplexSessionManager
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_submission_resumable_default_preserves_minicpm() -> None:
+    ctx = DuplexStageRequestContext(
+        request_id="r",
+        session_id="s",
+        fence=DuplexFence("s"),
+        stage_id=0,
+        final_stage_id=2,
+        config_generation=0,
+        sampling_params=(SamplingParams(max_tokens=1),),
+    )
+    submission = DuplexStageSubmission(
+        context=ctx,
+        prompt={"prompt_token_ids": [1, 2]},
+        already_submitted=False,
+    )
+    assert submission.resumable is True
+
+
+def test_stage_request_id_respects_resumable_flag() -> None:
+    fence = DuplexFence("sess", epoch=3, turn_id=9)
+    resumable_id = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=True)
+    ephemeral_id = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=False)
+    assert ephemeral_id == duplex_ephemeral_stage_request_id(fence, stage_id=0)
+    assert resumable_id != ephemeral_id
+    assert ephemeral_id.endswith("stage0-turn9")
+    from vllm_omni.engine.duplex.contracts import duplex_turn_id_from_request_id
+
+    assert duplex_turn_id_from_request_id(ephemeral_id) == 9
+    assert duplex_turn_id_from_request_id(resumable_id) is None
+
+
+def test_capabilities_expose_concurrent_turn_requests_default_false() -> None:
+    caps = DuplexCapabilities()
+    assert caps.supports_concurrent_turn_requests is False
+    assert caps.allows_video_without_audio() is False
+    assert caps.required_input_modalities == frozenset({"audio"})
+    assert caps.optional_input_modalities == frozenset({"video"})
+    assert caps.supports_core_resumable_request is False
+    payload = caps.as_dict()
+    assert payload["supports_concurrent_turn_requests"] is False
+    assert payload["required_input_modalities"] == ["audio"]
+    assert payload["optional_input_modalities"] == ["video"]
+
+
+def test_next_commit_allowed_soft_opens_on_r4_release() -> None:
+    from types import SimpleNamespace
+
+    from vllm_omni.engine.duplex.session import helpers
+
+    caps = DuplexCapabilities(supports_concurrent_turn_requests=True)
+    session = SimpleNamespace(active_response_id="resp", capabilities=caps)
+    tasks = SimpleNamespace(
+        active_response_task=None,
+        has_response_bound_append_tasks=lambda: False,
+    )
+    # Monkeypatch assistant_playback_active via response_in_progress path: active_response_id set.
+    assert helpers.response_in_progress(session, tasks) is True
+    assert helpers.next_commit_allowed(session, tasks, concurrent_turn_requests_released=False) is False
+    assert helpers.next_commit_allowed(session, tasks, concurrent_turn_requests_released=True) is True
+
+    caps_off = DuplexCapabilities(supports_concurrent_turn_requests=False)
+    session_off = SimpleNamespace(active_response_id="resp", capabilities=caps_off)
+    assert helpers.next_commit_allowed(session_off, tasks, concurrent_turn_requests_released=True) is False
+
+
+def test_active_response_accepts_own_turn_when_concurrent_turn_requests() -> None:
+    """Approach A: each response owns its turn; draining TTS uses request→response map."""
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+
+    session = DuplexEngineSession(
+        session_id="s",
+        config=DuplexSessionConfig(model="m", modalities=["text"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True),
+    )
+    session.begin_response(turn_id=3)
+    assert session.active_response_accepts_model_turn(3)
+    assert not session.active_response_accepts_model_turn(4)
+    assert not session.active_response_accepts_model_turn(2)
+    session.bind_draining_request("req-old", "resp-old")
+    assert session.response_id_for_request("req-old") == "resp-old"
+    assert session.is_draining_request("req-old")
+
+    # Ending the active response must not wipe older draining bindings.
+    session.bind_draining_request("req-newer-tts", session.active_response_id or "resp-active")
+    ended = session.active_response_id
+    session.end_response(commit_text=False)
+    assert session.is_draining_request("req-old")
+    assert session.response_id_for_request("req-old") == "resp-old"
+    assert not session.is_draining_request("req-newer-tts")
+    session.clear_draining_for_response("resp-old")
+    assert not session.is_draining_request("req-old")
+    del ended
+
+    strict = DuplexEngineSession(
+        session_id="s2",
+        config=DuplexSessionConfig(model="m", modalities=["text"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=False),
+    )
+    strict.begin_response(turn_id=3)
+    assert strict.active_response_accepts_model_turn(3)
+    assert not strict.active_response_accepts_model_turn(4)
+
+
+def test_draining_request_exempt_from_completed_turn_filter() -> None:
+    """Draining ownership must win over the model_turn_id < turn_id late-audio guard."""
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+
+    session = DuplexEngineSession(
+        session_id="s-drain",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True),
+    )
+    session.begin_response(turn_id=1)
+    r1 = session.active_response_id
+    assert r1 is not None
+    session.bind_draining_request("req-r1-tts", r1)
+    # Overlapping R2 becomes active, then finishes — must not wipe R1 draining.
+    session.begin_response(turn_id=2)
+    session.end_response(commit_text=False)
+    session.turn_id = 2
+    assert session.active_response_id is None
+    assert session.is_draining_request("req-r1-tts")
+    assert session.response_id_for_request("req-r1-tts") == r1
+    draining_response_id = session.response_id_for_request("req-r1-tts")
+    model_turn_id = 1
+    drop = (
+        draining_response_id is None
+        and session.active_response_id is None
+        and model_turn_id is not None
+        and model_turn_id < session.turn_id
+    )
+    assert drop is False
+
+
+def test_ephemeral_turn_id_parser_accepts_stage_turn_ids() -> None:
+    from vllm_omni.engine.duplex.contracts import duplex_turn_id_from_request_id
+
+    assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage0-turn12") == 12
+    assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage2-turn3") == 3
+    assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage0_t9") is None
+    assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage0") is None
+
+
+def test_same_turn_request_ids_skip_other_turns_and_epochs() -> None:
+    from vllm_omni.engine.duplex.contracts import duplex_same_turn_request_ids
+
+    stage3 = "duplex-s.x.e.0.r.stage3-turn4"
+    found = duplex_same_turn_request_ids(
+        stage3,
+        [
+            "duplex-s.x.e.0.r.stage0-turn4",
+            "duplex-s.x.e.0.r.stage1-turn4",
+            "duplex-s.x.e.0.r.stage2-turn4",
+            stage3,
+            "duplex-s.x.e.0.r.stage2-turn5",
+            "duplex-s.x.e.1.r.stage0-turn4",
+        ],
+    )
+    assert found == [
+        "duplex-s.x.e.0.r.stage0-turn4",
+        "duplex-s.x.e.0.r.stage1-turn4",
+        "duplex-s.x.e.0.r.stage2-turn4",
+    ]
+    from vllm_omni.engine.duplex.contracts import duplex_session_id_from_request_id
+
+    fence_session = "sess id"
+    import base64
+
+    encoded = base64.urlsafe_b64encode(fence_session.encode()).decode("ascii").rstrip("=")
+    assert duplex_session_id_from_request_id(f"duplex-s.{encoded}.e.0.r.stage1-turn2") == fence_session
+    assert duplex_session_id_from_request_id("not-a-request") is None
+
+
+def test_stale_keys_skip_already_draining_request_ids() -> None:
+    """Rebinding must not move an older draining Stage2/3 id onto the newer response."""
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+
+    session = DuplexEngineSession(
+        session_id="s-stale",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True),
+    )
+    session.begin_response(turn_id=1)
+    r1 = session.active_response_id
+    assert r1 is not None
+    session.bind_draining_request("req-r1-talker", r1)
+    session.begin_response(turn_id=2)
+    r2 = session.active_response_id
+    assert r2 is not None and r2 != r1
+    # Simulate the concurrent-turn rebind loop: already-draining ids keep R1.
+    for rid in ("req-r1-talker", "req-r2-talker"):
+        if session.is_draining_request(rid):
+            continue
+        session.bind_draining_request(rid, r2)
+    assert session.response_id_for_request("req-r1-talker") == r1
+    assert session.response_id_for_request("req-r2-talker") == r2
+
+
+def test_end_response_clears_only_own_draining_entries() -> None:
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+
+    session = DuplexEngineSession(
+        session_id="s-end",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True),
+    )
+    session.begin_response(turn_id=1)
+    r1 = session.active_response_id
+    session.bind_draining_request("req-r1", r1)
+    session.begin_response(turn_id=2)
+    r2 = session.active_response_id
+    session.bind_draining_request("req-r2", r2)
+    session.end_response(commit_text=False)
+    assert session.is_draining_request("req-r1")
+    assert session.response_id_for_request("req-r1") == r1
+    assert not session.is_draining_request("req-r2")
+
+
+def test_on_stage_failure_resolves_draining_response_before_active() -> None:
+    """Mirrors runner.on_stage_failure: draining request_id owns the failed done."""
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+
+    session = DuplexEngineSession(
+        session_id="s-fail",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True),
+    )
+    session.begin_response(turn_id=1)
+    r1 = session.active_response_id
+    assert r1 is not None
+    session.bind_draining_request("req-r1-talker", r1)
+    session.begin_response(turn_id=2)
+    r2 = session.active_response_id
+    assert r2 is not None and r2 != r1
+
+    request_id = "req-r1-talker"
+    draining_response_id = (
+        session.response_id_for_request(request_id) if session.is_draining_request(request_id) else None
+    )
+    response_id = draining_response_id or session.active_response_id
+    assert response_id == r1
+    assert response_id != r2
+    session.clear_draining_for_response(r1)
+    assert not session.is_draining_request(request_id)
+    assert session.active_response_id == r2
+
+
+def test_concurrent_turn_requests_released_resets_when_new_stage0_binds() -> None:
+    """After R4 release, a new ephemeral Stage0 bind must clear the gate flag."""
+    from types import SimpleNamespace
+
+    run = SimpleNamespace(concurrent_turn_requests_released=True)
+    # Simulate the concurrent-turn rebind arm in model_channel._append_via_data_plane.
+    concurrent_turn = True
+    if concurrent_turn:
+        run.concurrent_turn_requests_released = False
+    assert run.concurrent_turn_requests_released is False
+
+
+def test_draining_stage_ids_come_from_the_plugin() -> None:
+    from types import SimpleNamespace
+
+    from vllm_omni.engine.duplex.session.model_channel import ModelChannel
+
+    channel = ModelChannel.__new__(ModelChannel)
+    channel._ctx = SimpleNamespace(
+        plugin=SimpleNamespace(draining_stage_ids=lambda *, stage_count: {stage_count - 1}),
+        stage_port=SimpleNamespace(stage_count=5),
+    )
+    assert channel._draining_stage_ids() == frozenset({4})
+
+    channel._ctx.plugin = SimpleNamespace()
+    assert channel._draining_stage_ids() == frozenset()
+
+
+def test_draining_empty_eos_completes_owning_response_before_shortcut() -> None:
+    """R2 finishing first must not swallow R1's empty Stage3 EOS."""
+    import asyncio
+    from types import SimpleNamespace
+
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+    from vllm_omni.engine.duplex.session.model_channel import ModelChannel
+
+    session = DuplexEngineSession(
+        session_id="s-empty-eos",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"], extra_body={"auto_response": True}),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False),
+    )
+    session.begin_response(turn_id=1)
+    r1 = session.active_response_id
+    assert r1 is not None
+    r1_request = "duplex-s.x.e.0.r.stage3-turn1"
+    still_running = "duplex-s.x.e.0.r.stage1-turn2"
+    session.bind_draining_request(r1_request, r1)
+    session.begin_response(turn_id=2)
+    session.end_response(commit_text=False)
+    session.turn_id = 2
+    assert session.active_response_id is None
+
+    class _Plane:
+        def is_terminal(self, request_id: str | None) -> bool:
+            del request_id
+            return False
+
+        def close_stream(self, request_id: str) -> None:
+            del request_id
+
+        def mark_terminal(self, request_id: str) -> None:
+            del request_id
+
+    class _Port:
+        def __init__(self) -> None:
+            self.cleanups: list[list[str]] = []
+
+        async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None:
+            del abort
+            self.cleanups.append(list(request_ids))
+
+    class _Out:
+        def __init__(self) -> None:
+            self.events: list[dict[str, object]] = []
+
+        def auto_responds(self) -> bool:
+            return True
+
+        def emit(self, payload: dict[str, object]) -> None:
+            self.events.append(payload)
+
+    port = _Port()
+    out = _Out()
+
+    async def _noop(*_args: object, **_kwargs: object) -> None:
+        return None
+
+    async def _schedule(*_args: object, **_kwargs: object) -> bool:
+        return False
+
+    channel = ModelChannel(
+        SimpleNamespace(
+            session=session,
+            plugin=SimpleNamespace(data_plane=_Plane()),
+            stage_port=port,
+            model_state=SimpleNamespace(clear_continuation=lambda: None),
+            services=SimpleNamespace(spawn=lambda *_a, **_k: None),
+        ),
+        out,
+        close_from_runtime=_noop,
+        schedule_silence_continuation=_schedule,
+        abort_request=_noop,
+    )
+    asyncio.run(
+        channel._send_one_model_output_event(
+            {
+                "data_plane_request_id": r1_request,
+                "end_of_turn": True,
+                "model_turn_id": 1,
+                "text": "",
+                "stage_role": "tts",
+            }
+        )
+    )
+    done = [event for event in out.events if event.get("type") == "response.done"]
+    assert done and done[-1].get("response_id") == r1
+    assert port.cleanups == [[r1_request]]
+    assert all(still_running not in ids for ids in port.cleanups)
+    assert not session.is_draining_request(r1_request)
+
+
+def test_draining_completion_drops_finished_response_books_only() -> None:
+    """R1 drain finish drops R1 books and leaves the live R2 snapshot."""
+    import asyncio
+    from types import SimpleNamespace
+
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+    from vllm_omni.engine.duplex.session.model_channel import ModelChannel
+
+    session = DuplexEngineSession(
+        session_id="s-drain-books",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"], extra_body={"auto_response": True}),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False),
+    )
+    session.begin_response(turn_id=1)
+    r1 = session.active_response_id
+    assert r1 is not None
+    session.append_assistant_text("from-r1")
+    session.snapshot_active_response_for_drain()
+    r1_request = "duplex-s.x.e.0.r.stage3-turn1"
+    session.bind_draining_request(r1_request, r1)
+    session.begin_response(turn_id=2)
+    r2 = session.active_response_id
+    assert r2 is not None and r2 != r1
+    session.append_assistant_text("from-r2")
+    session.snapshot_active_response_for_drain()
+
+    class _Plane:
+        def is_terminal(self, request_id: str | None) -> bool:
+            del request_id
+            return False
+
+        def close_stream(self, request_id: str) -> None:
+            del request_id
+
+        def mark_terminal(self, request_id: str) -> None:
+            del request_id
+
+    class _Port:
+        async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None:
+            del request_ids, abort
+
+    class _Out:
+        def __init__(self) -> None:
+            self.events: list[dict[str, object]] = []
+
+        def auto_responds(self) -> bool:
+            return True
+
+        def emit(self, payload: dict[str, object]) -> None:
+            self.events.append(payload)
+
+    out = _Out()
+
+    async def _noop(*_args: object, **_kwargs: object) -> None:
+        return None
+
+    async def _schedule(*_args: object, **_kwargs: object) -> bool:
+        return False
+
+    channel = ModelChannel(
+        SimpleNamespace(
+            session=session,
+            plugin=SimpleNamespace(data_plane=_Plane()),
+            stage_port=_Port(),
+            model_state=SimpleNamespace(clear_continuation=lambda: None),
+            services=SimpleNamespace(spawn=lambda *_a, **_k: None),
+        ),
+        out,
+        close_from_runtime=_noop,
+        schedule_silence_continuation=_schedule,
+        abort_request=_noop,
+    )
+    asyncio.run(
+        channel._send_one_model_output_event(
+            {
+                "data_plane_request_id": r1_request,
+                "end_of_turn": True,
+                "model_turn_id": 1,
+                "text": "",
+                "stage_role": "tts",
+            }
+        )
+    )
+    assert r1 not in session._conversation.assistant_response_snapshots
+    assert r1 not in session._playback.by_response
+    assert f"item_{r1}" not in session._conversation.history_item_placeholders
+    assert r2 in session._conversation.assistant_response_snapshots
+    assert session.active_response_id == r2
+    done = [event for event in out.events if event.get("type") == "response.done"]
+    assert done and done[-1].get("response_id") == r1
+
+
+def test_stale_continue_does_not_close_the_new_response() -> None:
+    import asyncio
+    from types import SimpleNamespace
+
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+    from vllm_omni.engine.duplex.session.model_channel import ModelChannel
+
+    session = DuplexEngineSession(
+        session_id="s-stale-continue",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"], extra_body={"auto_response": True}),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False),
+    )
+    session.begin_response(turn_id=2)
+    live = session.active_response_id
+    session.epoch = 1
+
+    class _Out:
+        def __init__(self) -> None:
+            self.events: list[dict[str, object]] = []
+
+        def auto_responds(self) -> bool:
+            return True
+
+        def emit(self, payload: dict[str, object]) -> None:
+            self.events.append(payload)
+
+    out = _Out()
+    cleared: list[bool] = []
+
+    async def _noop(*_args: object, **_kwargs: object) -> None:
+        return None
+
+    async def _schedule(*_args: object, **_kwargs: object) -> bool:
+        return False
+
+    channel = ModelChannel(
+        SimpleNamespace(
+            session=session,
+            plugin=SimpleNamespace(data_plane=SimpleNamespace()),
+            stage_port=SimpleNamespace(),
+            model_state=SimpleNamespace(clear_continuation=lambda: cleared.append(True)),
+            services=SimpleNamespace(spawn=lambda *_a, **_k: None),
+            run=SimpleNamespace(closing=False),
+        ),
+        out,
+        close_from_runtime=_noop,
+        schedule_silence_continuation=_schedule,
+        abort_request=_noop,
+    )
+    asyncio.run(channel.maybe_continue_response(expected_epoch=0))
+    assert session.active_response_id == live
+    assert out.events == []
+    assert cleared == []
+
+    session.bind_draining_request("duplex-s.x.e.1.r.stage3-turn2", live or "")
+    asyncio.run(channel.maybe_continue_response(expected_epoch=1))
+    assert session.active_response_id == live
+    assert out.events == []
+
+    session.clear_draining_requests()
+    asyncio.run(channel.maybe_continue_response(expected_epoch=1))
+    assert session.active_response_id is None
+    assert [event.get("type") for event in out.events] == ["response.done"]
+    assert out.events[0].get("response_id") == live
+
+
+def test_silent_listen_with_continuation_releases_ephemeral_request() -> None:
+    import asyncio
+    from collections.abc import Coroutine
+    from types import SimpleNamespace
+    from typing import Any
+
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+    from vllm_omni.engine.duplex.session.model_channel import ModelChannel
+
+    session = DuplexEngineSession(
+        session_id="s-silent-continue",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False),
+    )
+    session.begin_response(turn_id=1)
+    response_id = session.active_response_id
+    request_id = "duplex-s.x.e.0.r.stage1-turn1"
+    session.bind_request(request_id)
+
+    class _Plane:
+        def __init__(self) -> None:
+            self.terminal: list[str] = []
+
+        def is_terminal(self, request_id: str | None) -> bool:
+            return request_id in self.terminal
+
+        def mark_terminal(self, request_id: str) -> None:
+            self.terminal.append(request_id)
+
+    class _Port:
+        def __init__(self) -> None:
+            self.cleanups: list[list[str]] = []
+
+        async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None:
+            del abort
+            self.cleanups.append(list(request_ids))
+
+    class _Out:
+        def __init__(self) -> None:
+            self.events: list[dict[str, object]] = []
+
+        def auto_responds(self) -> bool:
+            return False
+
+        def emit(self, payload: dict[str, object]) -> None:
+            self.events.append(payload)
+
+    port = _Port()
+    out = _Out()
+    spawned: list[Coroutine[Any, Any, None]] = []
+
+    async def _noop(*_args: object, **_kwargs: object) -> None:
+        return None
+
+    async def _schedule(*_args: object, **_kwargs: object) -> bool:
+        return False
+
+    channel = ModelChannel(
+        SimpleNamespace(
+            session=session,
+            plugin=SimpleNamespace(data_plane=_Plane()),
+            stage_port=port,
+            model_state=SimpleNamespace(
+                continuation_owner_id=None, continuation_units=0, clear_continuation=lambda: None
+            ),
+            services=SimpleNamespace(spawn=lambda coro, **_k: spawned.append(coro)),
+            run=SimpleNamespace(closing=False),
+        ),
+        out,
+        close_from_runtime=_noop,
+        schedule_silence_continuation=_schedule,
+        abort_request=_noop,
+    )
+    asyncio.run(
+        channel._send_one_model_output_event(
+            {
+                "data_plane_request_id": request_id,
+                "is_listen": True,
+                "end_of_turn": True,
+                "model_turn_id": 1,
+                "reason": "model_listen",
+            }
+        )
+    )
+    assert port.cleanups == [[request_id]]
+    assert session.active_request_id is None
+    assert spawned, "silent listen with unused continuation must schedule maybe_continue"
+    asyncio.run(spawned[0])
+    assert port.cleanups == [[request_id]]
+    assert session.active_response_id is None
+    assert [event.get("type") for event in out.events] == ["response.listen", "response.done"]
+    assert out.events[0].get("response_id") == response_id
+    assert out.events[1].get("response_id") == response_id
+
+
+def test_silent_listen_aborts_data_plane_request_with_full_id() -> None:
+    """abort_data_plane_request must pass [request_id], not a bare str."""
+    import asyncio
+    from types import SimpleNamespace
+
+    from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig
+    from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession
+    from vllm_omni.engine.duplex.session.model_channel import ModelChannel
+
+    session = DuplexEngineSession(
+        session_id="s-silent-abort",
+        config=DuplexSessionConfig(model="m", modalities=["text", "audio"]),
+        capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False),
+    )
+    request_id = "duplex-s.x.e.0.r.stage0-turn7"
+    session.bind_request(request_id)
+    aborts: list[list[str]] = []
+
+    class _Plane:
+        def is_terminal(self, request_id: str | None) -> bool:
+            return False
+
+        def mark_terminal(self, request_id: str) -> None:
+            del request_id
+
+    class _Port:
+        async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None:
+            del request_ids, abort
+
+    class _Out:
+        def auto_responds(self) -> bool:
+            return True
+
+        def emit(self, payload: dict[str, object]) -> None:
+            del payload
+
+    async def _noop(*_args: object, **_kwargs: object) -> None:
+        return None
+
+    async def _abort(request_ids: list[str], *, notify: bool = False) -> None:
+        del notify
+        aborts.append(list(request_ids))
+
+    channel = ModelChannel(
+        SimpleNamespace(
+            session=session,
+            plugin=SimpleNamespace(data_plane=_Plane()),
+            stage_port=_Port(),
+            model_state=SimpleNamespace(
+                continuation_owner_id=None, continuation_units=0, clear_continuation=lambda: None
+            ),
+            services=SimpleNamespace(spawn=lambda coro, **_k: None),
+            run=SimpleNamespace(closing=False),
+        ),
+        _Out(),
+        close_from_runtime=_noop,
+        schedule_silence_continuation=_noop,
+        abort_request=_abort,
+    )
+    asyncio.run(
+        channel._send_one_model_output_event(
+            {
+                "data_plane_request_id": request_id,
+                "is_listen": True,
+                "end_of_turn": True,
+                "model_turn_id": 1,
+                "reason": "model_listen",
+                "abort_data_plane_request": True,
+            }
+        )
+    )
+    assert aborts == [[request_id]]
+    assert all(len(item) > 1 for batch in aborts for item in batch), "must not split request id into chars"

tests/engine/duplex/test_engine_session.py

@@ -242,6 +242,19 @@ def test_history_commit_with_ack_only_policy_defers_unacknowledged_text():
     assert f"item_{response_id}" in session.pending_history_item_ids
 
 
+def test_history_commit_with_ack_only_policy_keeps_only_acknowledged_prefix():
+    session = _session(config=DuplexSessionConfig(playback_commit_policy="ack_only"))
+    session.begin_response()
+    session.append_assistant_text("hello world")
+    session.mark_audio_sent(duration_ms=1_000, text_chars=6)
+    session.mark_audio_sent(duration_ms=2_000, text_chars=11)
+    session.acknowledge_playback(played_ms=1_000, committed_ms=1_000)
+
+    committed = session.end_response(commit_text=True)
+
+    assert committed == {"role": "assistant", "content": "hello"}
+
+
 @pytest.mark.parametrize("audio_complete", [False, True])
 def test_unaligned_response_keeps_empty_turn_before_later_user_input(audio_complete):
     session = _session(config=DuplexSessionConfig(playback_commit_policy="ack_only"))
@@ -469,6 +482,36 @@ def test_cancel_fence_releases_stage_requests_and_advances_identity():
         session.cancel_fence(DuplexFence("sid-other", epoch=1, turn_id=0), DuplexFence("sid-cancel", epoch=2))
 
 
+def test_request_resource_keys_are_stage_id_and_request_id():
+    session = _session("sid-keys")
+    session.bind_stage_request(0, "req-a", fence=session.fence)
+    session.bind_stage_request(1, "req-b", fence=session.fence)
+    session.bind_stage_request(2, "req-tts", fence=session.fence)
+    stale_keys = list(session.request_resources.keys())
+    stale_ids = list(dict.fromkeys(rid for _, rid in stale_keys))
+    assert stale_ids == ["req-a", "req-b", "req-tts"]
+    for sid, rid in stale_keys:
+        if sid < 2:
+            session.request_resources.pop((sid, rid), None)
+    assert session.resource_request_ids() == ["req-tts"]
+    with pytest.raises(TypeError, match="unhashable"):
+        dict.fromkeys(rid for _, rid in session.request_resources.items())
+
+
+def test_release_resources_for_request_ids_keeps_other_ids_on_the_same_fence():
+    session = _session("sid-drain-release")
+    fence = session.fence
+    session.bind_stage_request(2, "drain", fence=fence)
+    session.bind_stage_request(3, "drain", fence=fence)
+    session.bind_stage_request(0, "live", fence=fence)
+
+    released = session.release_resources_for_request_ids(["drain"])
+
+    assert released == ["drain"]
+    assert session.resource_request_ids() == ["live"]
+    assert session.release_resources_for_request_ids([]) == []
+
+
 # ---- public view ----
 
 
@@ -688,3 +731,65 @@ def test_minicpmo_native_capabilities_do_not_overclaim_single_session_deployment
 
     assert caps["supports_multi_session"] is False
     assert caps["supports_multi_session_same_replica"] is False
+
+
+def test_draining_response_stays_ack_admissible_after_next_begin_response() -> None:
+    from vllm_omni.engine.duplex.events import ErrorEvent
+    from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack
+
+    session = _session()
+    first = session.begin_response(turn_id=1)
+    session.append_assistant_text("hello")
+    session.mark_audio_sent(400, text_chars=5)
+    session.snapshot_active_response_for_drain()
+    second = session.begin_response(turn_id=2)
+    assert second != first
+    assert session.playback.sent_ms == 0
+    events = apply_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": first,
+            "item_id": f"item_{first}",
+            "played_ms": 100,
+            "committed_ms": 100,
+        },
+    )
+    assert not any(isinstance(event, ErrorEvent) and event.code == "playback_item_not_found" for event in events)
+    session.mark_audio_sent(900, text_chars=5, response_id=first)
+    assert session.playback_for_response(first).sent_ms == 900
+    assert session.playback.sent_ms == 0
+
+
+def test_finished_drain_keeps_sent_audio_ackable() -> None:
+    from vllm_omni.engine.duplex.events import ErrorEvent
+    from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack
+
+    session = _session()
+    first = session.begin_response(turn_id=1)
+    session.append_assistant_text("hello")
+    session.mark_audio_sent(400, text_chars=5)
+    session.snapshot_active_response_for_drain()
+    second = session.begin_response(turn_id=2)
+    session.release_finished_drain_response(first)
+    assert first in session._conversation.assistant_response_snapshots
+    assert f"item_{first}" in session._conversation.history_item_placeholders
+    events = apply_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": first,
+            "item_id": f"item_{first}",
+            "played_ms": 400,
+            "committed_ms": 400,
+        },
+    )
+    assert not any(isinstance(event, ErrorEvent) for event in events)
+    assert session.active_response_id == second
+
+    silent = session.begin_response(turn_id=3)
+    session.snapshot_active_response_for_drain()
+    session.begin_response(turn_id=4)
+    session.release_finished_drain_response(silent)
+    assert silent not in session._conversation.assistant_response_snapshots
+    assert f"item_{silent}" not in session._conversation.history_item_placeholders

tests/engine/duplex/test_input_transcribed.py

@@ -0,0 +1,49 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import pytest
+
+from vllm_omni.engine.duplex.realtime_events import (
+    RealtimeProjectionState,
+    project_internal_event,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_input_transcribed_is_a_user_transcript_event() -> None:
+    events = project_internal_event(
+        RealtimeProjectionState(session_id="s"),
+        {"type": "input.transcribed", "transcript": "  你好  "},
+    )
+    assert len(events) == 1
+    assert events[0].wire_type == "conversation.item.input_audio_transcription.completed"
+    assert events[0].transcript == "你好"
+
+
+def test_input_transcribed_updates_the_committed_item() -> None:
+    state = RealtimeProjectionState(session_id="s")
+    project_internal_event(
+        state,
+        {
+            "type": "input.committed",
+            "realtime_item_id": "item_user_1",
+            "is_speech": True,
+            "message": {"role": "user"},
+        },
+    )
+    events = project_internal_event(
+        state,
+        {"type": "input.transcribed", "transcript": "你好", "realtime_item_id": "item_user_1"},
+    )
+    assert events[0].item_id == "item_user_1"
+    item = state.conversation_items["item_user_1"]
+    assert item["content"][0]["transcript"] == "你好"
+
+
+def test_blank_input_transcribed_emits_nothing() -> None:
+    events = project_internal_event(
+        RealtimeProjectionState(session_id="s"),
+        {"type": "input.transcribed", "transcript": "  "},
+    )
+    assert events == []

tests/engine/duplex/test_overlap_policy.py

@@ -79,6 +79,39 @@ def test_an_explicit_client_overlap_action_requests_barge_in() -> None:
     assert decision["reason"] == "client_overlap_action"
 
 
+def test_turn_mode_keeps_silent_vision_follow_during_overlap() -> None:
+    """Vision-follow is is_speech=False + a frame; must not drop while TTS plays."""
+    session = _session()
+    session.capabilities = type(session.capabilities)(
+        required_input_modalities=frozenset({"video"}),
+        optional_input_modalities=frozenset({"audio"}),
+    )
+    payload = {"format": "pcm16", "audio": _pcm16(0.0), "video_frames": ["frame"], "is_speech": False}
+    decision = overlap_policy.decide(session, {"is_speech": False}, payload, auto_responds=False)
+    assert decision["action"] == "listen"
+    assert decision["reason"] == "vision_follow"
+    assert decision["buffer_audio"] is True
+    assert decision["defer_runtime_append"] is False
+    assert decision["force_listen"] is False
+
+
+def test_turn_mode_drops_silent_vision_without_capability() -> None:
+    session = _session()
+    assert session.capabilities.allows_video_without_audio() is False
+    payload = {"format": "pcm16", "audio": _pcm16(0.0), "video_frames": ["frame"], "is_speech": False}
+    decision = overlap_policy.decide(session, {"is_speech": False}, payload, auto_responds=False)
+    assert decision["action"] == "drop"
+    assert decision["reason"] == "silence_or_noise"
+
+
+def test_turn_mode_still_drops_silent_overlap_without_vision() -> None:
+    session = _session()
+    payload = {"format": "pcm16", "audio": _pcm16(0.0), "is_speech": False}
+    decision = overlap_policy.decide(session, {"is_speech": False}, payload, auto_responds=False)
+    assert decision["action"] == "drop"
+    assert decision["reason"] == "silence_or_noise"
+
+
 def test_barge_in_is_deferred_when_the_session_cannot_support_it() -> None:
     session = _session()
     session.capabilities = type(session.capabilities)(supports_barge_in=False)

tests/engine/duplex/test_session_manager.py

@@ -196,7 +196,11 @@ class FakePlugin(DuplexModelPlugin):
 
     def capabilities(self, *, max_sessions: int) -> DuplexCapabilities:
         del max_sessions
-        return DuplexCapabilities(supports_input_append=True, supports_core_resumable_request=True)
+        # Resident Stage0 ids (MiniCPM-shaped); AURA opts out via supports_core_resumable_request=False.
+        return DuplexCapabilities(
+            supports_input_append=True,
+            supports_core_resumable_request=True,
+        )
 
     def validate_client_extra_body(self, extra_body: object) -> None:
         pass
@@ -409,7 +413,8 @@ async def test_open_answers_with_capabilities_and_emits_session_created() -> Non
         assert result.session_id == "sid-open"
         assert result.lease_generation == 0
         assert result.capabilities == DuplexCapabilities(
-            supports_input_append=True, supports_core_resumable_request=True
+            supports_input_append=True,
+            supports_core_resumable_request=True,
         )
         assert result.public_session is not None
         assert result.public_session["id"] == "sid-open"
@@ -1028,6 +1033,40 @@ async def test_append_bytes_are_reserved_at_admission_until_the_runner_dequeues(
         gate.set()
 
 
+async def test_append_admission_counts_audio_and_video_frame_bytes() -> None:
+    """Manager reserves len(audio)+Σlen(frame); video bytes count toward the same limit."""
+    async with Harness.create(max_sessions=1, max_pending_input_bytes_per_session=20) as harness:
+        await harness.open("sid-av")
+        harness.events()
+        session = harness.session("sid-av")
+        runner = harness.manager.runners["sid-av"]
+        gate = asyncio.Event()
+        runner._mailbox.put_nowait(_Internal("wait", {"gate": gate}))
+        runner._on_internal = lambda item: gate.wait()  # type: ignore[method-assign]
+        await asyncio.sleep(0)
+
+        # Default caps require audio; attach video to a non-empty audio unit.
+        frame_a = "aaaa"
+        frame_b = "bbbbbb"
+        audio = b"1234"
+        expected = len(audio) + len(frame_a) + len(frame_b)
+        harness.command(
+            "sid-av",
+            AppendAudio(audio=audio, video_frames=(frame_a, frame_b), event_id="evt-av"),
+        )
+        assert session.pending_input_bytes == expected
+        # Second append that would exceed the limit is backpressured.
+        harness.command(
+            "sid-av",
+            AppendAudio(audio=b"x" * 10, video_frames=("yyyyyyyyyy",), event_id="evt-over"),
+        )
+        errors = [event for event in harness.events("sid-av") if isinstance(event, ErrorEvent)]
+        assert [error.code for error in errors] == ["input_backpressure"]
+        assert errors[0].related_event_id == "evt-over"
+        assert session.pending_input_bytes == expected
+        gate.set()
+
+
 async def test_expired_session_retains_the_admission_slot_until_cleanup_succeeds() -> None:
     async with Harness.create(max_sessions=1, idle_ttl_s=1.0) as harness:
         await harness.open("sid-expired")

tests/engine/duplex/test_session_runner.py

@@ -180,7 +180,10 @@ class Harness:
             segment_token_ids=tuple(segment_token_ids),
             segment_output_metadata=dict(segment_output_metadata or {}),
         )
-        return self.runner.on_stage_output(stage_id, output, metrics, request_id=output.request_id, context=context)
+        request_id = getattr(output, "request_id", None)
+        if not isinstance(request_id, str):
+            raise AssertionError("stage output is missing request_id")
+        return self.runner.on_stage_output(stage_id, output, metrics, request_id=request_id, context=context)
 
     async def deliver_and_settle(self, output: object, **kwargs: Any) -> list[DuplexEvent]:
         self.deliver(output, **kwargs)
@@ -520,6 +523,28 @@ async def test_commit_without_audio_is_rejected() -> None:
         await close_harness(h)
 
 
+@pytest.mark.asyncio
+async def test_empty_silent_commit_acks_without_opening_response() -> None:
+    """#3: empty is_speech=False commit keeps the silent-ack pair, even auto-respond."""
+    h = await open_harness(auto_response=True)
+    try:
+        # resolve_commit only admits empty silent commits after non-speech was seen.
+        projector = h.runner._require_projector()
+        projector.input_audio_buffer_had_non_speech = True
+        events = await h.run(commands.Commit())
+        assert "error" not in types(events), types(events)
+        assert "input_audio_buffer.committed" in types(events), types(events)
+        assert "response.listen" in types(events), types(events)
+        listen = next(event for event in events if event.type == "response.listen")
+        assert listen.details["reason"] == "silence_or_noise"
+        committed = next(event for event in events if event.type == "input_audio_buffer.committed")
+        assert committed.details.get("empty") is True
+        assert committed.details.get("no_response") is True
+        assert h.port.submissions == []
+    finally:
+        await close_harness(h)
+
+
 @pytest.mark.asyncio
 async def test_commit_projects_the_user_item_and_does_not_resubmit_consumed_audio() -> None:
     h = await open_harness()
@@ -674,6 +699,45 @@ async def test_stale_epoch_output_is_dropped_after_barge_in() -> None:
         await close_harness(h)
 
 
+async def test_barge_in_aborts_draining_tts_as_well_as_the_active_request() -> None:
+    h = await open_harness()
+    try:
+        await h.run(append_audio())
+        request_id = h.stage0_request_id()
+        await h.deliver_and_settle(tts_output(request_id, samples=24000, text="he"))
+        h.session.bind_draining_request("duplex-drain-tts", "resp-old")
+        # Draining TTS sits on an older turn fence. cancel_fence only drops
+        # the fence being cancelled, so these bindings must be popped by id.
+        from vllm_omni.engine.duplex.contracts import DuplexFence
+        from vllm_omni.engine.duplex.session.engine_session import DuplexRequestResource
+
+        older = DuplexFence(h.session.session_id, epoch=h.session.epoch, turn_id=h.session.turn_id + 5)
+        h.session.request_resources[(2, "duplex-drain-tts")] = DuplexRequestResource(
+            stage_id=2, request_id="duplex-drain-tts", fence=older, submitted=True
+        )
+        h.session.request_resources[(3, "duplex-drain-tts")] = DuplexRequestResource(
+            stage_id=3, request_id="duplex-drain-tts", fence=older, submitted=True
+        )
+        h.session.request_resources[(1, "still-live")] = DuplexRequestResource(
+            stage_id=1, request_id="still-live", fence=older, submitted=True
+        )
+        events = await h.run(commands.BargeIn())
+        assert h.port.aborts == [[request_id, "duplex-drain-tts"]]
+        assert not h.session.is_draining_request("duplex-drain-tts")
+        done_ids = [
+            event.response["id"]
+            for event in events
+            if getattr(event, "wire_type", "") == "response.done" and isinstance(getattr(event, "response", None), dict)
+        ]
+        assert "resp-old" in done_ids
+        assert (2, "duplex-drain-tts") not in h.session.request_resources
+        assert (3, "duplex-drain-tts") not in h.session.request_resources
+        assert (0, request_id) not in h.session.request_resources
+        assert (1, "still-live") in h.session.request_resources
+    finally:
+        await close_harness(h)
+
+
 @pytest.mark.asyncio
 async def test_cancel_response_without_active_response_is_rejected() -> None:
     h = await open_harness()
@@ -778,6 +842,46 @@ async def test_listen_decision_on_a_resumable_request_closes_the_bounded_respons
         await close_harness(h)
 
 
+@pytest.mark.asyncio
+async def test_direct_response_listen_still_emits_response_done_after_continuation_clear() -> None:
+    """AURA-like silent DIRECT_RESPONSE: after continuation budget is spent, emit response.done.
+
+    Non-resumable turn-commit still closes the active response on a listen/direct
+    decision so the client is not left without a terminal event.
+    """
+    from dataclasses import replace
+
+    h = await open_harness()
+    try:
+        h.session.capabilities = replace(h.session.capabilities, supports_core_resumable_request=False)
+        await h.run(append_audio())
+        assert h.port.submissions, "ephemeral Stage0 must submit"
+        request_id = h.port.submissions[-1].context.request_id
+        assert "-turn" in request_id
+        await h.deliver_and_settle(tts_output(request_id, samples=24000, text="ok"))
+        response_id = h.session.active_response_id
+        assert response_id is not None
+        model_state = h.runner.model_state
+        model_state.continuation_owner_id = f"response:{response_id}"
+        model_state.continuation_units = h.runner.model._AUTO_RESPONSE_MAX_CONTINUATION_UNITS
+
+        listen = listen_output(request_id)
+        listen.finished = True
+        events = await h.deliver_and_settle(
+            listen,
+            stage_id=0,
+            segment_finished=True,
+            segment_token_ids=[11, 12, LISTEN_TOKEN_ID],
+            segment_output_metadata={"meta.listen_token_id": LISTEN_TOKEN_ID},
+        )
+        assert "response.listen" in types(events)
+        done = find(events, "response.done")
+        assert done.response_id == response_id
+        assert h.session.active_response_id is None
+    finally:
+        await close_harness(h)
+
+
 @pytest.mark.asyncio
 async def test_listen_decision_on_a_resumable_request_keeps_the_turn_going() -> None:
     """Same unfinished listen, budget left: it must schedule the next unit.
@@ -860,6 +964,38 @@ async def test_stage_failure_fails_the_active_response() -> None:
         await close_harness(h)
 
 
+@pytest.mark.asyncio
+async def test_stage_failure_on_draining_request_fails_that_response_only() -> None:
+    """Overlapped: a draining older request failure must not wipe the active response."""
+    from dataclasses import replace
+
+    h = await open_harness()
+    try:
+        h.session.capabilities = replace(h.session.capabilities, supports_concurrent_turn_requests=True)
+        await h.run(append_audio())
+        request_id = h.stage0_request_id()
+        await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hello"))
+        r1 = h.session.active_response_id
+        assert r1 is not None
+        draining_req = "req-r1-talker-drain"
+        h.session.bind_draining_request(draining_req, r1)
+        # Open a newer active response while R1 TTS is still draining.
+        r2 = h.session.begin_response(turn_id=(h.session.turn_id or 0) + 1)
+        assert h.session.active_response_id == r2
+        assert r2 != r1
+
+        h.runner.on_stage_failure(2, RuntimeError("drain-boom"), request_id=draining_req)
+        events = await h.settle()
+        assert types(events)[0] == "error"
+        done = find(events, "response.done")
+        assert done.response_id == r1
+        assert done.status == "failed"
+        assert h.session.active_response_id == r2
+        assert not h.session.is_draining_request(draining_req)
+    finally:
+        await close_harness(h)
+
+
 # --------------------------------------------------------------------------- #
 # Turn mode (no auto response)                                                #
 # --------------------------------------------------------------------------- #
@@ -877,6 +1013,91 @@ async def test_turn_mode_skips_silent_chunks() -> None:
         await close_harness(h)
 
 
+@pytest.mark.asyncio
+async def test_turn_mode_keeps_silent_chunks_with_video_frames() -> None:
+    """Engine vision-follow buffers silent+frames; MiniCPM commit stays speech-gated."""
+    from dataclasses import replace
+
+    h = await open_harness(auto_response=False)
+    try:
+        h.session.capabilities = replace(
+            h.session.capabilities,
+            required_input_modalities=frozenset({"video"}),
+            optional_input_modalities=frozenset({"audio"}),
+        )
+        # Minimal 1x1 JPEG (base64) — wire validation only checks non-empty str.
+        frame = (
+            "/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkS"
+            "Ew8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/2wBDAQkJ"
+            "CQwLDBgNDRgyIRwhMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIy"
+            "MjIyMjIyMjIyMjIyMjL/wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAA"
+            "AAAAAAAAAAj/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/8QAFQEBAQAAAAAAAAAAAAAA"
+            "AAAAAAD/xAAUEQEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIQAxAAAAGfAP/EABQQ"
+            "AQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEAAQUCf//EABQRAQAAAAAAAAAAAAAAAAAA"
+            "AAD/2gAIAQMBAT8Bf//EABQRAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQIBAT8Bf//E"
+            "ABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEABj8Cf//EABQQAQAAAAAAAAAAAAAA"
+            "AAAAAAD/2gAIAQEAAT8hf//Z"
+        )
+        cmd = commands.AppendAudio(
+            audio=pcm_f32(160, value=0.0),
+            format="pcm_f32le",
+            sample_rate_hz=16000,
+            is_speech=False,
+            video_frames=(frame,),
+        )
+        events = await h.run(cmd)
+        assert "response.listen" not in types(events)
+        assert h.runner.model_state.audio_buffer.has_pending()
+        events = await h.run(commands.Commit(create_response=True))
+        # MiniCPM prepare_commit is speech-gated (no frames-only Stage0).
+        assert "input_audio_buffer.committed" in types(events), types(events)
+        assert h.port.submissions == []
+    finally:
+        await close_harness(h)
+
+
+@pytest.mark.asyncio
+async def test_turn_mode_keeps_silent_vision_while_response_in_progress() -> None:
+    """TTS still playing: vision-follow must buffer, not drop as silence_or_noise."""
+    from dataclasses import replace
+
+    h = await open_harness(auto_response=False)
+    try:
+        h.session.capabilities = replace(
+            h.session.capabilities,
+            required_input_modalities=frozenset({"video"}),
+            optional_input_modalities=frozenset({"audio"}),
+        )
+        h.session._response.active_response_id = "resp-tts"
+        frame = (
+            "/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkS"
+            "Ew8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/2wBDAQkJ"
+            "CQwLDBgNDRgyIRwhMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIy"
+            "MjIyMjIyMjIyMjIyMjL/wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAA"
+            "AAAAAAAAAAj/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/8QAFQEBAQAAAAAAAAAAAAAA"
+            "AAAAAAD/xAAUEQEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIQAxAAAAGfAP/EABQQ"
+            "AQAAAAAAAAAAAAAAAAAAAAD/2gAIAQMBAT8Bf//EABQRAQAAAAAAAAAAAAAAAAAA"
+            "AAD/2gAIAQIBAT8Bf//EABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEABj8Cf//E"
+            "ABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEAAT8hf//Z"
+        )
+        events = await h.run(
+            commands.AppendAudio(
+                audio=pcm_f32(160, value=0.0),
+                format="pcm_f32le",
+                sample_rate_hz=16000,
+                is_speech=False,
+                video_frames=(frame,),
+            )
+        )
+        listen_reasons = [
+            getattr(event, "details", {}) or {} for event in events if getattr(event, "type", None) == "response.listen"
+        ]
+        assert all(details.get("reason") != "silence_or_noise" for details in listen_reasons)
+        assert h.runner.model_state.audio_buffer.has_pending()
+    finally:
+        await close_harness(h)
+
+
 @pytest.mark.asyncio
 async def test_turn_mode_commit_with_response_create_starts_one_response() -> None:
     h = await open_harness(auto_response=False)
@@ -1033,7 +1254,7 @@ async def test_conversation_items_can_be_injected_and_deleted() -> None:
         await close_harness(h)
 
 
-def _stage_metrics_of(event: object) -> dict[str, dict[str, object]]:
+def _stage_metrics_of(event: DuplexEvent) -> dict[str, dict[str, object]]:
     """Per-stage engine metrics as the client reads them off one wire event."""
     payload = event.to_realtime()
     metadata = payload.get("metadata")
@@ -1081,6 +1302,40 @@ async def test_stage0_metrics_reach_the_response_even_though_its_output_feeds_tt
         await close_harness(h)
 
 
+@pytest.mark.asyncio
+async def test_projected_stage1_metrics_reach_the_spoken_audio_event() -> None:
+    """Thinker text is projected, but its TTFT/TPOT still have to ride the audio event."""
+    h = await open_harness(stage_count=3)
+    try:
+        plugin = h.runner.model._ctx.plugin
+
+        def project_stage1(*, stage_id: int, output: object, context: object) -> bool:
+            del output, context
+            return stage_id == 1
+
+        plugin.project_intermediate_output = project_stage1
+        await h.run(append_audio())
+        request_id = h.stage0_request_id()
+        thinker = SimpleNamespace(
+            request_id=request_id,
+            finished=True,
+            outputs=[SimpleNamespace(text="hello", token_ids=[7, 8], multimodal_output={})],
+            multimodal_output={},
+        )
+        forwarded = h.deliver(
+            thinker,
+            stage_id=1,
+            segment_finished=False,
+            metrics=stage_stats(stage_id=1, request_id=request_id, num_tokens_out=2, itls_ms=[6.0]),
+        )
+        assert forwarded is False
+        events = await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hello"), stage_id=2)
+        stage_metrics = _stage_metrics_of(find(events, "response.output_audio.delta"))
+        assert stage_metrics["1"]["num_tokens_out"] == 2
+    finally:
+        await close_harness(h)
+
+
 @pytest.mark.asyncio
 async def test_stage0_metrics_from_several_units_are_summed_into_one_response() -> None:
     """Two pass-through segments before the first audio are one response's tokens."""

tests/engine/duplex/test_session_runner_ephemeral_bind.py

@@ -0,0 +1,28 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Ephemeral Stage0 bind uses r.stage0-turn{N}, not resumable r.stage0."""
+
+from __future__ import annotations
+
+from vllm_omni.engine.duplex.contracts import (
+    DuplexFence,
+    duplex_ephemeral_stage_request_id,
+    duplex_resource_request_id,
+)
+from vllm_omni.engine.duplex.session.manager import DuplexSessionManager
+
+
+def test_stage0_request_id_ephemeral_when_not_resumable() -> None:
+    fence = DuplexFence("duplex-test", epoch=0, turn_id=3)
+    rid = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=False)
+    expected = duplex_ephemeral_stage_request_id(fence, stage_id=0)
+    assert rid == expected
+    assert rid.endswith(".r.stage0-turn3")
+    assert not rid.endswith(".r.stage0")
+
+
+def test_stage0_request_id_resumable_keeps_stable_role() -> None:
+    fence = DuplexFence("duplex-test", epoch=1, turn_id=3)
+    rid = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=True)
+    expected = duplex_resource_request_id(fence, "stage0")
+    assert rid == expected
+    assert rid.endswith(".r.stage0")