[Core][Model] Fit PersonaPlex into the Unified Full-duplex Framework (RFC #7181 PR 2) (#7695)

Signed-off-by: chickeyton <ngton2014@gmail.com>
Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Yueqian Lin <linyueqian@outlook.com>

tests/model_executor/executor_common/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project

tests/model_executor/executor_common/audio/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project

tests/model_executor/executor_common/audio/test_pcm.py

@@ -0,0 +1,63 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import base64
+
+import numpy as np
+import pytest
+
+from vllm_omni.model_executor.common.audio.pcm import (
+    PCM_F32LE_BYTES_PER_SAMPLE,
+    decode_pcm_f32le_base64,
+    pcm_f32le_sample_count,
+    pcm_f32le_samples,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _b64(samples: np.ndarray) -> str:
+    return base64.b64encode(np.ascontiguousarray(samples, dtype="<f4").tobytes()).decode("ascii")
+
+
+def test_decode_returns_the_raw_bytes_of_whole_finite_samples() -> None:
+    samples = np.arange(4, dtype=np.float32)
+
+    raw = decode_pcm_f32le_base64(_b64(samples))
+
+    assert raw == samples.tobytes()
+    assert pcm_f32le_sample_count(raw) == 4
+    assert PCM_F32LE_BYTES_PER_SAMPLE == 4
+
+
+def test_decode_accepts_an_empty_chunk() -> None:
+    assert decode_pcm_f32le_base64("") == b""
+    assert pcm_f32le_sample_count(b"") == 0
+
+
+@pytest.mark.parametrize(
+    ("encoded", "match"),
+    [
+        (None, "must be base64 pcm_f32le"),
+        (b"bytes", "must be base64 pcm_f32le"),
+        ("not*base64", "not valid base64"),
+        (base64.b64encode(b"\x00\x00\x00").decode("ascii"), "divisible by four"),
+        (_b64(np.array([0.0, np.nan], dtype=np.float32)), "must be finite"),
+        (_b64(np.array([np.inf], dtype=np.float32)), "must be finite"),
+    ],
+)
+def test_decode_rejects_malformed_audio_naming_the_model(encoded: object, match: str) -> None:
+    with pytest.raises(ValueError, match=match) as excinfo:
+        decode_pcm_f32le_base64(encoded, model="TestModel")
+
+    assert str(excinfo.value).startswith("TestModel ")
+
+
+def test_samples_are_a_writable_contiguous_float32_copy() -> None:
+    raw = np.arange(3, dtype="<f4").tobytes()
+
+    samples = pcm_f32le_samples(raw)
+
+    assert samples.dtype == np.float32
+    assert samples.flags.writeable and samples.flags.c_contiguous
+    assert samples.tolist() == [0.0, 1.0, 2.0]

tests/model_executor/executor_common/duplex/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project

tests/model_executor/executor_common/duplex/test_data_plane.py

@@ -0,0 +1,104 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from vllm_omni.engine.duplex.contracts import DuplexFence, duplex_resource_request_id
+from vllm_omni.engine.duplex.plugin import DuplexDataPlane, DuplexDataPlaneContext
+from vllm_omni.model_executor.common.duplex.data_plane import CumulativeAudioTextDataPlane
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _encode_audio(audio, _sample_rate, _response_format, _speed):
+    if audio is None:
+        return None
+    size = int(np.asarray(audio, dtype=np.float32).size)
+    return f"audio-{size}" if size else None
+
+
+def _output(request_id: str, *, text: str, samples: int, sr: int = 24000) -> SimpleNamespace:
+    return SimpleNamespace(
+        request_id=request_id,
+        outputs=[SimpleNamespace(text=text, multimodal_output={})],
+        multimodal_output={"model_outputs": np.arange(samples, dtype=np.float32), "sr": sr},
+        finished=False,
+    )
+
+
+class _TestDataPlane(CumulativeAudioTextDataPlane):
+    stage_role = "codec"
+    default_sample_rate_hz = 16000
+
+
+def test_is_a_duplex_data_plane_and_projects_cumulative_output_as_deltas() -> None:
+    plane = CumulativeAudioTextDataPlane(_encode_audio)
+    plane.begin_request("req")
+    context = DuplexDataPlaneContext(response_format="wav", modalities=("audio", "text"))
+
+    projected = list(
+        plane.project(
+            {"data_plane_outputs": [_output("req", text="he", samples=4), _output("req", text="hello", samples=6)]},
+            context=context,
+        )
+    )
+
+    assert isinstance(plane, DuplexDataPlane)
+    assert [item["text"] for item in projected] == ["he", "llo"]
+    assert [item["audio_data"] for item in projected] == ["audio-4", "audio-2"]
+    assert [item["audio_duration_ms"] for item in projected] == [0, 0]
+    assert all(item["sample_rate_hz"] == 24000 for item in projected)
+    assert all(item["data_plane_request_id"] == "req" for item in projected)
+    assert all(item["is_listen"] is False and item["end_of_turn"] is False for item in projected)
+    assert projected[0]["stage_role"] == "tts"
+    assert projected[0]["runtime_impl"] == "scheduler_data_plane"
+    assert projected[0]["audio_format"] == "wav"
+
+
+def test_subclass_constants_and_default_sample_rate_reach_the_projection() -> None:
+    plane = _TestDataPlane(_encode_audio)
+    output = SimpleNamespace(
+        request_id="req",
+        outputs=[SimpleNamespace(text="", multimodal_output={})],
+        multimodal_output={"audio": np.zeros(160, dtype=np.float32)},
+        finished=False,
+    )
+
+    [item] = list(plane.project({"data_plane_outputs": [output]}))
+
+    assert item["stage_role"] == "codec"
+    assert item["sample_rate_hz"] == 16000
+    assert item["audio_duration_ms"] == 10
+
+
+def test_nothing_new_yields_nothing_and_non_results_are_ignored() -> None:
+    plane = CumulativeAudioTextDataPlane(_encode_audio)
+    same = _output("req", text="hi", samples=4)
+
+    assert len(list(plane.project({"data_plane_outputs": [same, same]}))) == 1
+    assert list(plane.project("not a dict")) == []
+    assert list(plane.project({"data_plane_outputs": None})) == []
+
+
+def test_terminal_and_close_bookkeeping() -> None:
+    plane = CumulativeAudioTextDataPlane(_encode_audio)
+    session_request = duplex_resource_request_id(DuplexFence("sid", epoch=0), "stage0")
+    plane.begin_request(session_request)
+    plane.begin_request("other-req")
+
+    assert plane.is_terminal(None) is False
+    assert plane.is_terminal(session_request) is False
+    plane.mark_terminal(session_request)
+    assert plane.is_terminal(session_request) is True
+    plane.begin_request(session_request)
+    assert plane.is_terminal(session_request) is False
+
+    plane.close_session("sid", active_request_id="other-req")
+    assert plane._requests == {}
+
+    plane.mark_terminal("late")
+    plane.close_stream("late")
+    assert plane.is_terminal("late") is False

tests/model_executor/executor_common/duplex/test_payload.py

@@ -0,0 +1,60 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import base64
+
+import numpy as np
+import pytest
+
+from vllm_omni.model_executor.common.duplex.payload import (
+    decode_pcm_f32le_payload,
+    payload_audio,
+    payload_sample_count,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _payload(samples: int, *, sample_rate_hz: int = 24000, fmt: str = "pcm_f32le", key: str = "audio") -> dict:
+    audio = base64.b64encode(np.zeros(samples, dtype="<f4").tobytes()).decode("ascii")
+    return {"type": "audio", "format": fmt, "sample_rate_hz": sample_rate_hz, key: audio}
+
+
+def test_decode_returns_the_pcm_of_a_matching_payload() -> None:
+    raw = decode_pcm_f32le_payload(_payload(1920), sample_rate_hz=24000, exact_samples=1920)
+
+    assert len(raw) == 1920 * 4
+
+
+def test_decode_accepts_the_older_data_key() -> None:
+    assert len(decode_pcm_f32le_payload(_payload(8, key="data"), sample_rate_hz=24000)) == 32
+    assert payload_audio({"data": "x"}) == "x"
+    assert payload_audio("not a mapping") is None
+
+
+@pytest.mark.parametrize(
+    ("payload", "match"),
+    [
+        ("frame", "must be a mapping"),
+        (_payload(1920, fmt="pcm16"), "format must be pcm_f32le"),
+        (_payload(1920, sample_rate_hz=16000), "sample_rate_hz must be 24000"),
+        (_payload(1919), "exactly 1920 samples"),
+        ({"format": "pcm_f32le", "sample_rate_hz": 24000, "audio": 12}, "must be base64"),
+    ],
+)
+def test_decode_rejects_payloads_that_do_not_match_the_model_unit(payload: object, match: str) -> None:
+    with pytest.raises(ValueError, match=match) as excinfo:
+        decode_pcm_f32le_payload(payload, sample_rate_hz=24000, exact_samples=1920, model="TestModel")
+
+    assert str(excinfo.value).startswith("TestModel ")
+
+
+def test_decode_without_exact_samples_accepts_any_whole_number_of_samples() -> None:
+    assert len(decode_pcm_f32le_payload(_payload(7), sample_rate_hz=24000)) == 28
+
+
+def test_payload_sample_count_is_lenient() -> None:
+    assert payload_sample_count(_payload(16000, sample_rate_hz=16000)) == 16000
+    assert payload_sample_count(_payload(16000, fmt="pcm16")) is None
+    assert payload_sample_count({"format": "pcm_f32le", "audio": "not*base64"}) is None
+    assert payload_sample_count(None) is None

tests/model_executor/executor_common/duplex/test_pcm_buffer.py

@@ -0,0 +1,185 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import base64
+
+import numpy as np
+import pytest
+
+from vllm_omni.engine.duplex.plugin import PcmAppendBuffer, PcmAppendReservation
+from vllm_omni.model_executor.common.duplex.pcm_buffer import (
+    FixedFramePcmAppendBuffer,
+    FixedFramePcmAppendReservation,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+FRAME = 1920
+RATE = 24000
+PERIOD = 80
+
+
+def _pcm_payload(samples: np.ndarray, *, sample_rate_hz: int = RATE, fmt: str = "pcm_f32le") -> dict[str, object]:
+    samples = np.ascontiguousarray(samples, dtype="<f4")
+    return {
+        "type": "audio",
+        "format": fmt,
+        "sample_rate_hz": sample_rate_hz,
+        "audio": base64.b64encode(samples.tobytes()).decode("ascii"),
+    }
+
+
+def _buffer() -> FixedFramePcmAppendBuffer:
+    return FixedFramePcmAppendBuffer(sample_rate_hz=RATE, frame_samples=FRAME, chunk_period_ms=PERIOD, model="Test")
+
+
+def _decode(payload: dict[str, object]) -> np.ndarray:
+    return np.frombuffer(base64.b64decode(str(payload["audio"])), dtype="<f4")
+
+
+def test_buffer_implements_the_framework_contracts() -> None:
+    buffer = _buffer()
+
+    assert isinstance(buffer, PcmAppendBuffer)
+    assert buffer.frame_bytes == FRAME * 4
+    reservation = buffer.prepare_append(
+        _pcm_payload(np.arange(FRAME)), operation_id="op", chunk_period_ms=PERIOD, allow_emit=True
+    )
+    assert isinstance(reservation, FixedFramePcmAppendReservation)
+    assert isinstance(reservation, PcmAppendReservation)
+
+
+def test_one_whole_frame_is_reserved_transactionally() -> None:
+    buffer = _buffer()
+
+    reservation = buffer.prepare_append(
+        _pcm_payload(np.arange(FRAME)), operation_id="op-1", chunk_period_ms=PERIOD, allow_emit=True
+    )
+
+    assert reservation is not None
+    assert reservation.active
+    assert reservation.byte_count == FRAME * 4
+    assert reservation.payload is not None and reservation.payload["sample_rate_hz"] == RATE
+    assert _decode(reservation.payload).tolist() == list(range(FRAME))
+    assert buffer.pending_byte_count == 0
+    assert buffer.has_reserved() and not buffer.has_pending()
+
+    reservation.rollback()
+
+    assert not reservation.active
+    assert buffer.pending_byte_count == FRAME * 4
+    assert buffer.has_pending() and not buffer.has_reserved()
+
+
+def test_partial_chunks_accumulate_until_a_whole_frame_exists() -> None:
+    buffer = _buffer()
+    half = _pcm_payload(np.zeros(FRAME // 2))
+
+    assert buffer.prepare_append(half, operation_id="op-1", chunk_period_ms=PERIOD, allow_emit=True) is None
+    assert buffer.pending_byte_count == FRAME * 2
+    reservation = buffer.prepare_append(half, operation_id="op-2", chunk_period_ms=PERIOD, allow_emit=True)
+
+    assert reservation is not None
+    assert reservation.byte_count == FRAME * 4
+    assert buffer.pending_byte_count == 0
+
+
+def test_appends_are_not_emitted_while_emission_is_disallowed() -> None:
+    buffer = _buffer()
+
+    assert (
+        buffer.prepare_append(
+            _pcm_payload(np.zeros(FRAME)), operation_id="op", chunk_period_ms=PERIOD, allow_emit=False
+        )
+        is None
+    )
+    assert buffer.pending_byte_count == FRAME * 4
+
+
+def test_commit_flushes_a_padded_final_frame_and_an_empty_buffer_reserves_nothing() -> None:
+    buffer = _buffer()
+    buffer.prepare_append(_pcm_payload(np.ones(10)), operation_id="op-1", chunk_period_ms=PERIOD, allow_emit=False)
+
+    reservation = buffer.prepare_commit(operation_id="commit", chunk_period_ms=PERIOD)
+
+    assert reservation.payload is not None
+    assert reservation.payload["final"] is True
+    decoded = _decode(reservation.payload)
+    assert decoded.size == FRAME
+    assert decoded[:10].tolist() == [1.0] * 10 and not decoded[10:].any()
+    assert reservation.byte_count == 40
+    reservation.commit()
+    assert not buffer.has_reserved()
+
+    empty = buffer.prepare_commit(operation_id="commit-2", chunk_period_ms=PERIOD)
+    assert empty.payload is None and empty.byte_count == 0
+    empty.commit()
+    assert buffer.flush(chunk_period_ms=PERIOD) is None
+
+
+def test_rolling_back_an_earlier_reservation_restores_later_ones_in_order() -> None:
+    buffer = _buffer()
+    first = buffer.prepare_append(
+        _pcm_payload(np.full(FRAME, 1.0)), operation_id="op-1", chunk_period_ms=PERIOD, allow_emit=True
+    )
+    second = buffer.prepare_append(
+        _pcm_payload(np.full(FRAME, 2.0)), operation_id="op-2", chunk_period_ms=PERIOD, allow_emit=True
+    )
+    assert first is not None and second is not None
+
+    first.rollback()
+
+    assert not first.active and not second.active
+    assert buffer.pending_byte_count == 2 * FRAME * 4
+    replay = buffer.prepare_append(
+        _pcm_payload(np.zeros(0)), operation_id="op-3", chunk_period_ms=PERIOD, allow_emit=True
+    )
+    assert replay is not None and _decode(replay.payload)[0] == 1.0
+
+
+def test_clear_drops_pending_audio_and_deactivates_reservations() -> None:
+    buffer = _buffer()
+    reservation = buffer.prepare_append(
+        _pcm_payload(np.zeros(FRAME + 8)), operation_id="op-1", chunk_period_ms=PERIOD, allow_emit=True
+    )
+    assert reservation is not None and buffer.pending_byte_count == 32
+
+    buffer.clear()
+    buffer.clear_force_listen()
+
+    assert not reservation.active
+    assert buffer.pending_byte_count == 0
+    reservation.rollback()  # no-op after clear
+    assert buffer.pending_byte_count == 0
+
+
+@pytest.mark.parametrize(
+    ("payload", "match"),
+    [
+        (_pcm_payload(np.zeros(8), sample_rate_hz=16000), "sample_rate_hz must be 24000"),
+        (_pcm_payload(np.zeros(8), fmt="pcm16"), "format must be pcm_f32le"),
+        (_pcm_payload(np.array([0.0, np.nan])), "finite"),
+    ],
+)
+def test_malformed_audio_is_rejected_and_nothing_is_buffered(payload: dict[str, object], match: str) -> None:
+    buffer = _buffer()
+
+    with pytest.raises(ValueError, match=match):
+        buffer.prepare_append(payload, operation_id="op", chunk_period_ms=PERIOD, allow_emit=True)
+
+    assert buffer.pending_byte_count == 0
+
+
+def test_chunk_period_and_duplicate_operations_are_refused() -> None:
+    buffer = _buffer()
+
+    with pytest.raises(ValueError, match="chunk_period_ms must be 80"):
+        buffer.prepare_append(_pcm_payload(np.zeros(8)), operation_id="op", chunk_period_ms=1000, allow_emit=True)
+    buffer.prepare_append(_pcm_payload(np.zeros(FRAME)), operation_id="op", chunk_period_ms=PERIOD, allow_emit=True)
+    with pytest.raises(ValueError, match="duplicate active operation_id"):
+        buffer.prepare_append(_pcm_payload(np.zeros(8)), operation_id="op", chunk_period_ms=PERIOD, allow_emit=True)
+
+
+def test_constructor_validates_its_parameters() -> None:
+    with pytest.raises(ValueError, match="positive"):
+        FixedFramePcmAppendBuffer(sample_rate_hz=0, frame_samples=FRAME, chunk_period_ms=PERIOD)

tests/model_executor/executor_common/test_request_outputs.py

@@ -0,0 +1,139 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import torch
+
+from vllm_omni.model_executor.common.request_outputs import (
+    audio_sample_count,
+    audio_value,
+    coerce_int,
+    coerce_int_list,
+    first_completion,
+    multimodal_output,
+    sample_rate_hz,
+    slice_audio_delta,
+    text_delta,
+    text_value,
+    unwrap_request_output,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_unwrap_prefers_the_wrapped_request_output_and_its_first_completion() -> None:
+    completion = SimpleNamespace(text="hi", multimodal_output={})
+    inner = SimpleNamespace(request_id="req", outputs=[completion, SimpleNamespace(text="second")])
+    wrapper = SimpleNamespace(request_output=inner)
+
+    output, first = unwrap_request_output(wrapper)
+
+    assert output is inner
+    assert first is completion
+    assert first_completion(SimpleNamespace(outputs=[])) is None
+    assert first_completion(SimpleNamespace()) is None
+
+
+def test_multimodal_output_takes_the_first_non_empty_mapping_as_a_copy() -> None:
+    completion = SimpleNamespace(multimodal_output={"audio": [1.0]})
+    output = SimpleNamespace(multimodal_output={})
+
+    result = multimodal_output(output, completion)
+
+    assert result == {"audio": [1.0]}
+    assert result is not completion.multimodal_output
+    assert multimodal_output(SimpleNamespace(multimodal_output=None), None) == {}
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (4, 4),
+        ("11", 11),
+        (2.9, 2),
+        (True, 1),
+        (None, None),
+        ("x", None),
+        (object(), None),
+        ([1], None),
+        (torch.tensor([7]), 7),
+        (torch.tensor(9), 9),
+        (torch.tensor([], dtype=torch.long), None),
+        (np.array([5], dtype=np.int32), 5),
+        (np.array([], dtype=np.int32), None),
+    ],
+)
+def test_coerce_int_handles_scalars_strings_tensors_and_arrays(value: object, expected: int | None) -> None:
+    assert coerce_int(value) == expected
+
+
+def test_coerce_int_list_flattens_tensors_and_drops_non_integers() -> None:
+    assert coerce_int_list(torch.tensor([[1, 2], [3, 4]])) == [1, 2, 3, 4]
+    assert coerce_int_list(np.array([5, 6])) == [5, 6]
+    assert coerce_int_list([1, "2", "x", None, torch.tensor(3)]) == [1, 2, 3]
+    assert coerce_int_list(None) == []
+    assert coerce_int_list("12") == []
+
+
+def test_audio_value_reads_the_known_keys_and_unwraps_a_single_item_list() -> None:
+    assert audio_value({"model_outputs": [np.zeros(2)]}).shape == (2,)
+    assert audio_value({"latent": "x"}) == "x"
+    assert audio_value({"audio": [1, 2]}) == [1, 2]
+    assert audio_value({}) is None
+
+
+def test_text_value_prefers_multimodal_text_then_llm_text_then_the_completion() -> None:
+    completion = SimpleNamespace(text="from completion")
+    assert text_value({"text": "mm"}, completion) == "mm"
+    assert text_value({"llm_output_text": "llm"}, completion) == "llm"
+    assert text_value({}, completion) == "from completion"
+    assert text_value({"text": ""}, None) == ""
+
+
+@pytest.mark.parametrize(
+    ("text", "previous", "expected"),
+    [("", "he", ""), ("he", "he", ""), ("hello", "he", "llo"), ("new", "old", "new"), ("abc", "", "abc")],
+)
+def test_text_delta(text: str, previous: str, expected: str) -> None:
+    assert text_delta(text, previous) == expected
+
+
+def test_slice_audio_delta_returns_only_new_samples_for_cumulative_audio() -> None:
+    cumulative = np.arange(6, dtype=np.float32)
+
+    assert slice_audio_delta(cumulative, 4).tolist() == [4.0, 5.0]
+    assert slice_audio_delta(cumulative, 6) is None
+    assert slice_audio_delta(cumulative, 0).tolist() == cumulative.tolist()
+    # A shorter waveform than the cursor means the stream restarted: send it whole.
+    assert slice_audio_delta(np.arange(3, dtype=np.float32), 4).tolist() == [0.0, 1.0, 2.0]
+    assert slice_audio_delta(None, 2) is None
+    assert slice_audio_delta(np.zeros(0, dtype=np.float32), 0) is None
+
+
+def test_slice_audio_delta_keeps_tensors_as_contiguous_tensors() -> None:
+    delta = slice_audio_delta(torch.arange(6, dtype=torch.float32).reshape(2, 3), 4)
+
+    assert isinstance(delta, torch.Tensor)
+    assert delta.is_contiguous()
+    assert delta.tolist() == [4.0, 5.0]
+    assert audio_sample_count(torch.zeros(2, 3)) == 6
+    assert audio_sample_count("not audio") is None
+
+
+@pytest.mark.parametrize(
+    ("multimodal", "expected"),
+    [
+        ({"sr": 16000}, 16000),
+        ({"sample_rate_hz": 22050}, 22050),
+        ({"sr": [48000]}, 48000),
+        ({"sr": np.array([8000])}, 8000),
+        ({"sr": "bad"}, 24000),
+        ({"sr": 0}, 24000),
+        ({}, 24000),
+    ],
+)
+def test_sample_rate_hz_falls_back_to_the_default(multimodal: dict, expected: int) -> None:
+    assert sample_rate_hz(multimodal, default=24000) == expected

tests/model_executor/models/personaplex/duplex/conftest.py

@@ -1,21 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
-"""PersonaPlex duplex is not on the duplex plugin framework yet.
-
-RFC vllm-omni#7181 splits the port: this PR lands the unified framework with
-MiniCPM-o 4.5, and the follow-up PR moves PersonaPlex onto ``DuplexModelPlugin``.
-Until then the modules under test still import the pre-framework duplex
-runtime / serving adapter that this PR removed, so collecting them would fail
-at import time and abort the whole pytest session. The tests themselves are
-unchanged and come back with the port.
-
-Listed per file rather than globbed, so the coverage loss is visible and shrinks
-file by file as the port lands.
-"""
-
-#: ``test_unified_runtime`` imports ``engine.duplex.runtime`` and
-#: ``entrypoints.duplex.runtime_adapter`` directly. ``test_stage0_runtime`` fails
-#: transitively: ``personaplex.duplex.runtime_extension`` imports ``DuplexFence``
-#: from ``engine.duplex.messages``, which no longer exports it.
-collect_ignore = ["test_stage0_runtime.py", "test_unified_runtime.py"]

tests/model_executor/models/personaplex/duplex/test_plugin.py

@@ -0,0 +1,489 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""PersonaPlex on the unified full-duplex framework: the plugin contract and the E2E driver helpers."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+from pathlib import Path
+from types import SimpleNamespace
+from typing import cast
+
+import numpy as np
+import pytest
+from vllm.sampling_params import SamplingParams
+
+from tests.e2e.online_serving import personaplex_realtime_duplex as e2e_driver
+from vllm_omni.config.stage_config import load_deploy_config, merge_pipeline_deploy
+from vllm_omni.engine.duplex.config import DuplexSessionConfig
+from vllm_omni.engine.duplex.contracts import DuplexFence
+from vllm_omni.engine.duplex.plugin import (
+    DefaultDuplexModelSessionState,
+    DuplexDataPlaneContext,
+    DuplexRuntimeConfigError,
+    load_duplex_plugin,
+    validate_duplex_plugin_sampling,
+)
+from vllm_omni.model_executor.models.personaplex.duplex import stage0
+from vllm_omni.model_executor.models.personaplex.duplex.config import DEFAULT_PERSONA, FRAME_SIZE, SAMPLE_RATE
+from vllm_omni.model_executor.models.personaplex.duplex.data_plane import PersonaPlexDataPlaneSession
+from vllm_omni.model_executor.models.personaplex.duplex.input import PersonaPlexPcmAppendBuffer
+from vllm_omni.model_executor.models.personaplex.duplex.plugin import (
+    PRIVATE_RUNTIME_CONFIG_KEYS,
+    PersonaPlexDuplexPlugin,
+    PersonaPlexSessionState,
+)
+from vllm_omni.model_executor.models.personaplex.pipeline import PERSONAPLEX_PIPELINE
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+DEPLOY_PATH = Path(__file__).parents[5] / "vllm_omni" / "deploy" / "personaplex.yaml"
+
+
+def _pcm_payload(samples: np.ndarray, *, sample_rate_hz: int = SAMPLE_RATE) -> dict[str, object]:
+    samples = np.ascontiguousarray(samples, dtype="<f4")
+    return {
+        "type": "audio",
+        "format": "pcm_f32le",
+        "sample_rate_hz": sample_rate_hz,
+        "audio": base64.b64encode(samples.tobytes()).decode("ascii"),
+    }
+
+
+def _encode_audio(audio, _sample_rate, _response_format, _speed):
+    if audio is None:
+        return None
+    size = int(np.asarray(audio, dtype=np.float32).size)
+    return f"audio-{size}" if size else None
+
+
+def _plugin() -> PersonaPlexDuplexPlugin:
+    return PersonaPlexDuplexPlugin(_encode_audio)
+
+
+# --------------------------------------------------------------------------- #
+# Pipeline binding                                                            #
+# --------------------------------------------------------------------------- #
+
+
+def test_pipeline_declares_the_duplex_plugin_and_no_pre_framework_wiring() -> None:
+    assert PERSONAPLEX_PIPELINE.duplex_plugin == (
+        "vllm_omni.model_executor.models.personaplex.duplex.plugin.PersonaPlexDuplexPlugin"
+    )
+    assert PERSONAPLEX_PIPELINE.duplex_runtime_extension is None
+    assert PERSONAPLEX_PIPELINE.duplex_serving_adapter is None
+    assert PERSONAPLEX_PIPELINE.duplex_control_enabled is False
+
+    plugin = load_duplex_plugin(PERSONAPLEX_PIPELINE.duplex_plugin, _encode_audio)
+
+    assert isinstance(plugin, PersonaPlexDuplexPlugin)
+    assert plugin.plugin_id == "personaplex"
+    assert isinstance(plugin.data_plane, PersonaPlexDataPlaneSession)
+    validate_duplex_plugin_sampling(plugin, sampling_defaults=(SamplingParams(), SamplingParams()))
+
+
+@pytest.mark.parametrize(
+    ("engine_arg", "expected"),
+    [("skip_tokenizer_init", True), ("enable_prefix_caching", False)],
+)
+def test_personaplex_stage_engine_args(engine_arg: str, expected: bool) -> None:
+    stages = merge_pipeline_deploy(PERSONAPLEX_PIPELINE, load_deploy_config(DEPLOY_PATH))
+
+    assert [stage.yaml_engine_args.get(engine_arg) for stage in stages] == [expected, expected]
+
+
+def test_personaplex_deploy_is_duplex_and_propagates_capacity_to_all_model_stages() -> None:
+    deploy = load_deploy_config(DEPLOY_PATH)
+    stages = merge_pipeline_deploy(PERSONAPLEX_PIPELINE, deploy)
+
+    assert deploy.session_mode == "duplex"
+    assert deploy.duplex_session.max_sessions == 2
+    assert [stage.yaml_engine_args.get("duplex_max_sessions") for stage in stages] == [2, 2]
+    assert "personaplex_codec_max_sessions" not in deploy.connectors["connector_of_shared_memory"]["extra"]
+
+
+# --------------------------------------------------------------------------- #
+# Session policy                                                              #
+# --------------------------------------------------------------------------- #
+
+
+def test_capabilities_are_honest() -> None:
+    single = _plugin().capabilities(max_sessions=1)
+    multi = _plugin().capabilities(max_sessions=2)
+
+    assert multi.as_dict()["implementation_level"] == "model_native_duplex"
+    assert multi.as_dict()["input_modes"] == ["append_audio_chunk"]
+    assert multi.chunk_period_ms == 80
+    assert single.supports_multi_session is False
+    assert multi.supports_multi_session is True
+    assert multi.supports_multi_session_same_replica is True
+    assert multi.supports_barge_in is False
+    assert multi.supports_client_commit is False
+    assert multi.supports_external_turn_signal is False
+    assert multi.supports_session_resume is False
+    assert multi.supports_audio_truncate is False
+    assert multi.supports_chat_completions is False
+    assert multi.session_admission_mode == "engine_managed"
+
+
+def test_session_state_is_the_framework_default_with_the_personaplex_buffer() -> None:
+    state = _plugin().create_session_state()
+
+    assert isinstance(state, PersonaPlexSessionState)
+    assert isinstance(state, DefaultDuplexModelSessionState)
+    assert isinstance(state.audio_buffer, PersonaPlexPcmAppendBuffer)
+    state.retain_committed_audio({"audio": ""}, operation_id="op", reserved_bytes=8)
+    assert state.clear_committed_audio() == 8
+    assert state.committed_audio_payload is None
+
+
+def test_private_runtime_keys_are_rejected_in_extra_body() -> None:
+    plugin = _plugin()
+
+    assert PRIVATE_RUNTIME_CONFIG_KEYS == {
+        "personaplex_prefill_slots",
+        "personaplex_model_path",
+        "personaplex_voice_prompt",
+        "personaplex_persona",
+    }
+    plugin.validate_client_extra_body({"auto_response": True})
+    plugin.validate_client_extra_body(None)
+    with pytest.raises(DuplexRuntimeConfigError, match="personaplex_persona") as excinfo:
+        plugin.validate_client_extra_body({"personaplex_persona": "x", "auto_response": True})
+    assert excinfo.value.code == "invalid_duplex_runtime_config"
+
+
+@pytest.mark.asyncio
+async def test_prepare_runtime_config_resolves_voice_and_prefill_slots_once(monkeypatch) -> None:
+    calls: list[tuple[str, str, str]] = []
+
+    def fake_prefill_slots(model_path: str, voice: str, persona: str) -> int:
+        calls.append((model_path, voice, persona))
+        return 42
+
+    monkeypatch.setattr(stage0, "personaplex_prefill_slots", fake_prefill_slots)
+    plugin = _plugin()
+    model_config = SimpleNamespace(model="/models/personaplex")
+
+    runtime_config = await plugin.prepare_runtime_config(
+        DuplexSessionConfig(voice="NATM1.pt", instructions="Be brief."),
+        model_config=model_config,
+    )
+    again = await plugin.prepare_runtime_config(
+        DuplexSessionConfig(voice="NATM1.pt", instructions="Be brief."),
+        model_config=model_config,
+    )
+    defaults = await plugin.prepare_runtime_config(DuplexSessionConfig(), model_config=model_config)
+
+    assert runtime_config == {
+        "personaplex_model_path": "/models/personaplex",
+        "personaplex_voice_prompt": "NATM1.pt",
+        "personaplex_persona": "Be brief.",
+        "personaplex_prefill_slots": 42,
+    }
+    assert again == runtime_config
+    assert defaults["personaplex_voice_prompt"] == "NATF2.pt"
+    assert defaults["personaplex_persona"] == DEFAULT_PERSONA
+    # The voice bundle and tokenizer are read once per (model, voice, persona).
+    assert calls == [
+        ("/models/personaplex", "NATM1.pt", "Be brief."),
+        ("/models/personaplex", "NATF2.pt", DEFAULT_PERSONA),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_runtime_config_reports_prefill_failures_as_runtime_config_errors(monkeypatch) -> None:
+    def failing_prefill_slots(model_path: str, voice: str, persona: str) -> int:
+        raise FileNotFoundError("voices.tgz missing")
+
+    monkeypatch.setattr(stage0, "personaplex_prefill_slots", failing_prefill_slots)
+
+    with pytest.raises(DuplexRuntimeConfigError, match="voices.tgz missing") as excinfo:
+        await _plugin().prepare_runtime_config(DuplexSessionConfig(), model_config=SimpleNamespace(model="/m"))
+    assert excinfo.value.code == "prefill_unavailable"
+
+    with pytest.raises(DuplexRuntimeConfigError, match="model path") as excinfo:
+        await _plugin().prepare_runtime_config(DuplexSessionConfig(), model_config=None)
+    assert excinfo.value.code == "model_path_unavailable"
+
+
+@pytest.mark.parametrize("voice", ["../NATF2.pt", "voices/NATF2.pt", "NATF2.wav", "/abs/NATF2.pt"])
+@pytest.mark.asyncio
+async def test_prepare_runtime_config_rejects_voices_that_are_not_bundled_basenames(voice: str) -> None:
+    with pytest.raises(DuplexRuntimeConfigError, match="bundled .pt basename") as excinfo:
+        await _plugin().prepare_runtime_config(
+            DuplexSessionConfig(voice=voice), model_config=SimpleNamespace(model="/m")
+        )
+    assert excinfo.value.code == "invalid_voice"
+
+
+def test_runtime_config_update_rejects_changed_persona() -> None:
+    plugin = _plugin()
+    short_persona = "You are a helpful assistant."
+    current = {"personaplex_persona": short_persona, "personaplex_voice_prompt": "NATF2.pt"}
+
+    with pytest.raises(DuplexRuntimeConfigError, match="persona") as excinfo:
+        plugin.runtime_config_for_update(DuplexSessionConfig(instructions="You are a pirate."), current)
+    assert excinfo.value.code == "persona_update_unsupported"
+
+    unchanged = plugin.runtime_config_for_update(DuplexSessionConfig(instructions=short_persona), current)
+    assert unchanged["personaplex_persona"] == short_persona
+    assert unchanged is not current
+
+
+def test_runtime_config_update_rejects_changed_voice() -> None:
+    plugin = _plugin()
+    current = {"personaplex_persona": DEFAULT_PERSONA, "personaplex_voice_prompt": "NATF2.pt"}
+
+    with pytest.raises(DuplexRuntimeConfigError, match="voice") as excinfo:
+        plugin.runtime_config_for_update(DuplexSessionConfig(voice="NATM1.pt"), current)
+    assert excinfo.value.code == "voice_update_unsupported"
+
+    unchanged = plugin.runtime_config_for_update(DuplexSessionConfig(voice="NATF2.pt"), current)
+    assert unchanged["personaplex_voice_prompt"] == "NATF2.pt"
+
+
+def test_data_plane_context_is_the_framework_default() -> None:
+    context = _plugin().data_plane_context(
+        epoch=1,
+        turn_id=2,
+        active_response_turn_id=None,
+        active_response_id="resp",
+        auto_responds=True,
+        response_format="pcm16",
+        speed=None,
+        modalities=("audio",),
+    )
+
+    assert isinstance(context, DuplexDataPlaneContext)
+    assert context.response_format == "pcm16" and context.epoch == 1
+
+
+# --------------------------------------------------------------------------- #
+# Engine policy                                                               #
+# --------------------------------------------------------------------------- #
+
+
+def test_sampling_params_are_greedy_one_token_on_stage0_only() -> None:
+    defaults = (SamplingParams(temperature=0.8, top_k=10, max_tokens=10), SamplingParams(max_tokens=1024))
+
+    configured = _plugin().configure_sampling_params(runtime_config={}, defaults=defaults)
+
+    assert configured[0].temperature == 0.0
+    assert configured[0].top_k == 1
+    assert configured[0].max_tokens == 1
+    assert configured[1] is defaults[1]
+    assert defaults[0].max_tokens == 10
+    assert _plugin().configure_sampling_params(runtime_config={}, defaults=()) == ()
+
+
+def test_plan_append_reserves_one_slot_plus_prefill_on_the_first_seq() -> None:
+    plugin = _plugin()
+    fence = DuplexFence("session", epoch=3, turn_id=1)
+    common = {
+        "request_id": "req",
+        "fence": fence,
+        "session_config": {"instructions": "Be concise."},
+        "runtime_config": {"personaplex_prefill_slots": 4},
+        "turn_seq": 1,
+        "payload": _pcm_payload(np.zeros(FRAME_SIZE, np.float32)),
+        "final": False,
+        "sampling_params": SamplingParams(),
+    }
+
+    first = plugin.plan_append(seq=1, **common)
+    second = plugin.plan_append(seq=2, **common)
+
+    assert len(first.prompt["prompt_token_ids"]) == 5
+    assert len(second.prompt["prompt_token_ids"]) == 1
+    duplex = first.prompt["model_intermediate_buffer"]["duplex"]
+    assert duplex["session_id"] == "session"
+    assert duplex["epoch"] == 3
+    assert duplex["turn_id"] == 1
+    assert duplex["seq"] == 1
+    assert duplex["data_plane"] is True
+    assert duplex["fence"] == fence
+    assert duplex["scheduler_token_budget"] == 5
+    assert duplex["runtime_config"] == {"personaplex_prefill_slots": 4}
+    assert "incarnation" not in duplex
+    assert first.prompt["model_intermediate_buffer"]["global_request_id"] == ["session"]
+
+
+@pytest.mark.parametrize(
+    ("payload", "match"),
+    [
+        (_pcm_payload(np.zeros(FRAME_SIZE - 1, np.float32)), "exactly 1920 samples"),
+        (_pcm_payload(np.zeros(FRAME_SIZE, np.float32), sample_rate_hz=16000), "sample_rate_hz must be 24000"),
+        ({"format": "pcm16", "sample_rate_hz": SAMPLE_RATE, "audio": ""}, "format must be pcm_f32le"),
+        ("frame", "must be a mapping"),
+    ],
+)
+def test_plan_append_rejects_anything_but_one_24k_frame(payload: object, match: str) -> None:
+    with pytest.raises(ValueError, match=match):
+        _plugin().plan_append(
+            request_id="req",
+            fence=DuplexFence("session"),
+            session_config={},
+            runtime_config={},
+            seq=1,
+            turn_seq=1,
+            payload=payload,
+            final=False,
+            sampling_params=None,
+        )
+
+
+def test_plan_append_rejects_a_malformed_prefill_slot_count() -> None:
+    with pytest.raises(ValueError, match="personaplex_prefill_slots"):
+        _plugin().plan_append(
+            request_id="req",
+            fence=DuplexFence("session"),
+            session_config={},
+            runtime_config={"personaplex_prefill_slots": "many"},
+            seq=1,
+            turn_seq=1,
+            payload=_pcm_payload(np.zeros(FRAME_SIZE, np.float32)),
+            final=False,
+            sampling_params=None,
+        )
+
+
+def test_decide_output_never_decides() -> None:
+    assert (
+        _plugin().decide_output(
+            stage_id=0,
+            final_stage_id=1,
+            segment_finished=True,
+            segment_token_ids=(1,),
+            segment_output_metadata={},
+            output=SimpleNamespace(outputs=[]),
+        )
+        is None
+    )
+
+
+def test_silence_unit_is_one_24k_frame_that_plan_append_accepts() -> None:
+    plugin = _plugin()
+
+    unit = plugin.silence_unit_payload()
+
+    assert unit["format"] == "pcm_f32le"
+    assert unit["sample_rate_hz"] == SAMPLE_RATE
+    assert len(base64.b64decode(unit["audio"])) == FRAME_SIZE * 4
+    plan = plugin.plan_append(
+        request_id="req",
+        fence=DuplexFence("session"),
+        session_config={},
+        runtime_config={},
+        seq=7,
+        turn_seq=7,
+        payload=unit,
+        final=False,
+        sampling_params=None,
+    )
+    assert len(plan.prompt["prompt_token_ids"]) == 1
+
+
+# --------------------------------------------------------------------------- #
+# E2E driver helpers (CPU-checked parts of the GPU driver)                    #
+# --------------------------------------------------------------------------- #
+
+
+def _audio_client(frame_count: int = 10, *, voiced_frames: int = 0) -> e2e_driver.RawRealtimeProbe:
+    """A probe double carrying only the collected events the driver's result builder reads."""
+    frames = np.full((frame_count, e2e_driver.FRAME_SAMPLES), 7, dtype="<i2")
+    frames[:voiced_frames] = 1000
+    raw = frames.tobytes()
+    event = {
+        "type": "response.output_audio.delta",
+        "response_id": "response-1",
+        "sample_rate_hz": e2e_driver.SAMPLE_RATE_HZ,
+        "metadata": {
+            "vllm_omni": {
+                "runtime_impl": "scheduler_data_plane",
+                "uses_model_runner_scheduler": True,
+                "runner_kv_backed": True,
+            }
+        },
+    }
+    events = SimpleNamespace(events=[event], response_audio={"response-1": [raw]}, audio_bytes=lambda: raw)
+    return cast(e2e_driver.RawRealtimeProbe, SimpleNamespace(events=events))
+
+
+def test_realtime_audio_frame_stats_separate_voiced_and_silent_frames() -> None:
+    silent = np.zeros(e2e_driver.FRAME_SAMPLES, dtype="<i2")
+    voiced = np.full(e2e_driver.FRAME_SAMPLES, 1000, dtype="<i2")
+
+    stats = e2e_driver._audio_frame_stats(
+        silent.tobytes() + voiced.tobytes(),
+        input_frames=3,
+        voiced_frame_rms_threshold=1e-4,
+    )
+
+    assert stats == {
+        "output_frames": 2,
+        "voiced_frames": 1,
+        "silent_frames": 1,
+        "frame_deficit": 1,
+        "frame_coverage_ratio": pytest.approx(2 / 3),
+    }
+
+
+def test_realtime_audio_frame_stats_reject_partial_codec_frame() -> None:
+    partial_frame = np.zeros(e2e_driver.FRAME_SAMPLES - 1, dtype="<i2")
+
+    with pytest.raises(AssertionError, match="whole 80 ms codec frames"):
+        e2e_driver._audio_frame_stats(partial_frame.tobytes(), input_frames=1, voiced_frame_rms_threshold=1e-4)
+
+
+@pytest.mark.parametrize("frame_count", [10, 40])
+def test_e2e_driver_rejects_inaudible_sessions(frame_count: int) -> None:
+    args = argparse.Namespace(max_frame_deficit=4, voiced_frame_rms_threshold=1e-3, min_voiced_frames=5)
+
+    with pytest.raises(AssertionError, match="audible speech"):
+        e2e_driver._session_result(_audio_client(frame_count), input_frames=frame_count, args=args, minimum_chunks=1)
+
+
+def test_e2e_driver_uses_absolute_audible_floor() -> None:
+    args = argparse.Namespace(max_frame_deficit=4, voiced_frame_rms_threshold=1e-3, min_voiced_frames=5)
+
+    e2e_driver._session_result(_audio_client(200, voiced_frames=5), input_frames=200, args=args, minimum_chunks=1)
+
+
+def test_e2e_driver_defaults_replacement_to_full_input() -> None:
+    args = e2e_driver.parse_args(["--model", "/model", "--input-wav", "/input.wav"])
+
+    assert args.replacement_frames == 0
+    assert args.min_voiced_frames == 5
+
+
+def test_e2e_driver_records_and_validates_input_sha256(tmp_path: Path) -> None:
+    input_wav = tmp_path / "input.wav"
+    input_wav.write_bytes(b"fixture-wav")
+
+    identity = e2e_driver._input_identity(
+        input_wav,
+        expected_sha256="bade4b3c163edde390ff391207d34a887257d8a9cc3b621cc8c618b6e6761304",
+    )
+
+    assert identity == {
+        "path": str(input_wav.resolve()),
+        "sha256": "bade4b3c163edde390ff391207d34a887257d8a9cc3b621cc8c618b6e6761304",
+    }
+
+
+def test_e2e_driver_rejects_unexpected_input_sha256(tmp_path: Path) -> None:
+    input_wav = tmp_path / "input.wav"
+    input_wav.write_bytes(b"fixture-wav")
+
+    with pytest.raises(ValueError, match="input WAV SHA-256 mismatch"):
+        e2e_driver._input_identity(input_wav, expected_sha256="0" * 64)
+
+
+def test_e2e_driver_lets_the_server_allocate_the_session_id() -> None:
+    url = e2e_driver._realtime_url("ws://127.0.0.1:8099/v1/realtime?duplex=1", "nvidia/personaplex-7b-v1")
+
+    assert "session_id=" not in url
+    assert "duplex=1" in url and "autostart=0" in url

tests/model_executor/models/personaplex/duplex/test_stage0_runtime.py

@@ -14,6 +14,7 @@ from vllm_omni.model_executor.models.personaplex.duplex.policy import (
 )
 from vllm_omni.model_executor.models.personaplex.duplex.stage0 import (
     PersonaPlexStage0DuplexRuntime,
+    PersonaPlexStage0PreparedAppend,
 )
 from vllm_omni.model_executor.models.personaplex.personaplex_talker import (
     PersonaPlexTalkerForConditionalGeneration,
@@ -80,13 +81,12 @@ class _FakeTalker:
         return torch.full((1, 4), value, device=device)
 
 
-def _duplex_info(*, seq: int, session_id: str = "session", incarnation: int = 1):
+def _duplex_info(*, seq: int, session_id: str = "session", epoch: int = 0):
     pcm = np.zeros(1920, dtype="<f4")
     return {
         "data_plane": True,
         "session_id": session_id,
-        "incarnation": incarnation,
-        "epoch": 0,
+        "epoch": epoch,
         "seq": seq,
         "payload": {
             "format": "pcm_f32le",
@@ -103,7 +103,7 @@ def _duplex_info(*, seq: int, session_id: str = "session", incarnation: int = 1)
 
 def _runtime(*codecs: _FakeCodec) -> PersonaPlexStage0DuplexRuntime:
     voice_embeddings = torch.arange(8, dtype=torch.float32).reshape(2, 1, 1, 4)
-    codec_args = {"codec": codecs[0]}
+    codec_args: dict[str, object] = {"codec": codecs[0]}
     if len(codecs) > 1:
         available = iter(codecs)
         codec_args = {
@@ -248,15 +248,20 @@ def test_decoded_pcm_is_writable_for_torch_zero_copy() -> None:
 
 def _prepare_two_sessions(
     runtime: PersonaPlexStage0DuplexRuntime,
-) -> tuple[object, object, object, object]:
+) -> tuple[
+    PersonaPlexStage0PreparedAppend,
+    PersonaPlexStage0PreparedAppend,
+    PersonaPlexStage0PreparedAppend,
+    PersonaPlexStage0PreparedAppend,
+]:
     first_1 = runtime.prepare_append(_duplex_info(seq=1), prompt_len=18)
     second_1 = runtime.prepare_append(
-        _duplex_info(seq=1, session_id="other", incarnation=2),
+        _duplex_info(seq=1, session_id="other"),
         prompt_len=18,
     )
     first_2 = runtime.prepare_append(_duplex_info(seq=2), prompt_len=19)
     second_2 = runtime.prepare_append(
-        _duplex_info(seq=2, session_id="other", incarnation=2),
+        _duplex_info(seq=2, session_id="other"),
         prompt_len=19,
     )
     return first_1, second_1, first_2, second_2
@@ -281,7 +286,7 @@ def test_stage0_session_capacity_fails_before_codec_state_is_shared() -> None:
 
     with pytest.raises(RuntimeError, match="capacity 2"):
         runtime.prepare_append(
-            _duplex_info(seq=1, session_id="third", incarnation=3),
+            _duplex_info(seq=1, session_id="third"),
             prompt_len=18,
         )
 
@@ -292,12 +297,55 @@ def test_close_session_resets_and_reuses_released_codec() -> None:
     runtime = _runtime(first_codec, second_codec)
     _prepare_two_sessions(runtime)
 
-    runtime.close_session("session", 1)
+    runtime.close_session("session", 0)
 
     assert first_codec.reset_calls == 1
     assert second_codec.reset_calls == 0
     replacement = runtime.prepare_append(
-        _duplex_info(seq=1, session_id="replacement", incarnation=4),
+        _duplex_info(seq=1, session_id="replacement"),
         prompt_len=18,
     )
     assert replacement.user_codes[:, 0].tolist() == [3]
+
+
+def test_a_new_epoch_replays_the_prefill_and_recycles_the_codec() -> None:
+    codec = _FakeCodec()
+    runtime = _runtime(codec)
+    runtime.prepare_append(_duplex_info(seq=1), prompt_len=18, request_id="req-e0")
+    runtime.prepare_append(_duplex_info(seq=2), prompt_len=19, request_id="req-e0")
+
+    # A cancel advanced the fence: the next append is seq 1 of epoch 1 on a
+    # fresh Stage 0 request, so the voice/persona prefill is replayed and the
+    # earlier epoch's lockstep state is released first.
+    restarted = runtime.prepare_append(_duplex_info(seq=1, epoch=1), prompt_len=18, request_id="req-e1")
+
+    assert restarted.prefill_applied is True
+    assert restarted.prompt_offset == 0
+    assert restarted.user_codes.shape == (1, 8)
+    assert list(runtime.sessions) == [("session", 1)]
+    assert runtime.request_sessions == {"req-e1": ("session", 1)}
+    assert codec.reset_calls == 1
+    assert runtime.sessions[("session", 1)].codec is codec
+
+
+def test_a_late_finish_of_the_old_epoch_request_does_not_close_the_new_state() -> None:
+    runtime = _runtime(_FakeCodec())
+    runtime.prepare_append(_duplex_info(seq=1), prompt_len=18, request_id="req-e0")
+    runtime.prepare_append(_duplex_info(seq=1, epoch=1), prompt_len=18, request_id="req-e1")
+
+    runtime.close_request("req-e0")
+
+    assert list(runtime.sessions) == [("session", 1)]
+    runtime.close_request("req-e1")
+    assert runtime.sessions == {}
+
+
+def test_stage0_capacity_counts_live_epochs_not_superseded_ones() -> None:
+    runtime = _runtime(_FakeCodec(), _FakeCodec())
+    runtime.prepare_append(_duplex_info(seq=1), prompt_len=18)
+    runtime.prepare_append(_duplex_info(seq=1, session_id="other"), prompt_len=18)
+
+    # Restarting one session must not need a third codec.
+    runtime.prepare_append(_duplex_info(seq=1, epoch=1), prompt_len=18)
+
+    assert sorted(runtime.sessions) == [("other", 0), ("session", 1)]

tests/model_executor/models/personaplex/duplex/test_unified_runtime.py

@@ -1,420 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
-import base64
-from pathlib import Path
-from types import SimpleNamespace
-
-import numpy as np
-import pytest
-from vllm.sampling_params import SamplingParams
-from vllm_omni.engine.duplex.runtime import (
-    DuplexInputMode,
-)
-from vllm_omni.entrypoints.duplex.runtime_adapter import (
-    ServingRuntimeConfigError,
-)
-
-from tests.e2e.online_serving import personaplex_realtime_duplex as e2e_driver
-from vllm_omni.config.stage_config import (
-    load_deploy_config,
-    merge_pipeline_deploy,
-)
-from vllm_omni.engine.duplex.messages import DuplexFence
-from vllm_omni.model_executor.models.personaplex.duplex.config import DEFAULT_PERSONA
-from vllm_omni.model_executor.models.personaplex.duplex.data_plane import (
-    PersonaPlexDataPlaneContext,
-    PersonaPlexDataPlaneSession,
-)
-from vllm_omni.model_executor.models.personaplex.duplex.input import (
-    PersonaPlexPcmAppendBuffer,
-)
-from vllm_omni.model_executor.models.personaplex.duplex.runtime_extension import (
-    PersonaPlexDuplexRuntimeExtension,
-)
-from vllm_omni.model_executor.models.personaplex.duplex.serving_adapter import (
-    PersonaPlexServingRuntimeAdapter,
-)
-from vllm_omni.model_executor.models.personaplex.pipeline import (
-    PERSONAPLEX_PIPELINE,
-)
-
-pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
-
-
-def _pcm_payload(samples: np.ndarray, *, sample_rate_hz: int = 24000) -> dict[str, object]:
-    samples = np.ascontiguousarray(samples, dtype="<f4")
-    return {
-        "type": "audio",
-        "format": "pcm_f32le",
-        "sample_rate_hz": sample_rate_hz,
-        "audio": base64.b64encode(samples.tobytes()).decode("ascii"),
-    }
-
-
-def _audio_client(
-    frame_count: int = 10,
-    *,
-    voiced_frames: int = 0,
-) -> SimpleNamespace:
-    frames = np.full((frame_count, e2e_driver.FRAME_SAMPLES), 7, dtype="<i2")
-    frames[:voiced_frames] = 1000
-    raw = frames.tobytes()
-    event = {
-        "type": "response.output_audio.delta",
-        "response_id": "response-1",
-        "sample_rate_hz": e2e_driver.SAMPLE_RATE_HZ,
-        "metadata": {
-            "vllm_omni": {
-                "runtime_impl": "scheduler_data_plane",
-                "uses_model_runner_scheduler": True,
-                "runner_kv_backed": True,
-            }
-        },
-    }
-    events = SimpleNamespace(
-        events=[event],
-        response_audio={"response-1": [raw]},
-        audio_bytes=lambda: raw,
-    )
-    return SimpleNamespace(events=events)
-
-
-def test_realtime_audio_frame_stats_separate_voiced_and_silent_frames() -> None:
-    silent = np.zeros(e2e_driver.FRAME_SAMPLES, dtype="<i2")
-    voiced = np.full(e2e_driver.FRAME_SAMPLES, 1000, dtype="<i2")
-
-    stats = e2e_driver._audio_frame_stats(
-        silent.tobytes() + voiced.tobytes(),
-        input_frames=3,
-        voiced_frame_rms_threshold=1e-4,
-    )
-
-    assert stats == {
-        "output_frames": 2,
-        "voiced_frames": 1,
-        "silent_frames": 1,
-        "frame_deficit": 1,
-        "frame_coverage_ratio": pytest.approx(2 / 3),
-    }
-
-
-def test_realtime_audio_frame_stats_reject_partial_codec_frame() -> None:
-    partial_frame = np.zeros(e2e_driver.FRAME_SAMPLES - 1, dtype="<i2")
-
-    with pytest.raises(AssertionError, match="whole 80 ms codec frames"):
-        e2e_driver._audio_frame_stats(
-            partial_frame.tobytes(),
-            input_frames=1,
-            voiced_frame_rms_threshold=1e-4,
-        )
-
-
-@pytest.mark.parametrize("frame_count", [10, 40])
-def test_e2e_driver_rejects_inaudible_sessions(frame_count: int) -> None:
-    args = SimpleNamespace(
-        max_frame_deficit=4,
-        voiced_frame_rms_threshold=1e-3,
-        min_voiced_frames=5,
-    )
-
-    with pytest.raises(AssertionError, match="audible speech"):
-        e2e_driver._session_result(
-            _audio_client(frame_count),
-            input_frames=frame_count,
-            args=args,
-            minimum_chunks=1,
-        )
-
-
-def test_e2e_driver_uses_absolute_audible_floor() -> None:
-    args = SimpleNamespace(
-        max_frame_deficit=4,
-        voiced_frame_rms_threshold=1e-3,
-        min_voiced_frames=5,
-    )
-
-    e2e_driver._session_result(
-        _audio_client(200, voiced_frames=5),
-        input_frames=200,
-        args=args,
-        minimum_chunks=1,
-    )
-
-
-def test_e2e_driver_defaults_replacement_to_full_input() -> None:
-    args = e2e_driver.parse_args(
-        [
-            "--model",
-            "/model",
-            "--input-wav",
-            "/input.wav",
-        ]
-    )
-
-    assert args.replacement_frames == 0
-    assert args.min_voiced_frames == 5
-
-
-def test_e2e_driver_records_and_validates_input_sha256(tmp_path: Path) -> None:
-    input_wav = tmp_path / "input.wav"
-    input_wav.write_bytes(b"fixture-wav")
-
-    identity = e2e_driver._input_identity(
-        input_wav,
-        expected_sha256="bade4b3c163edde390ff391207d34a887257d8a9cc3b621cc8c618b6e6761304",
-    )
-
-    assert identity == {
-        "path": str(input_wav.resolve()),
-        "sha256": "bade4b3c163edde390ff391207d34a887257d8a9cc3b621cc8c618b6e6761304",
-    }
-
-
-def test_e2e_driver_rejects_unexpected_input_sha256(tmp_path: Path) -> None:
-    input_wav = tmp_path / "input.wav"
-    input_wav.write_bytes(b"fixture-wav")
-
-    with pytest.raises(ValueError, match="input WAV SHA-256 mismatch"):
-        e2e_driver._input_identity(input_wav, expected_sha256="0" * 64)
-
-
-@pytest.mark.parametrize(
-    ("attribute", "expected"),
-    [
-        (
-            "duplex_runtime_extension",
-            "vllm_omni.model_executor.models.personaplex.duplex.runtime_extension.PersonaPlexDuplexRuntimeExtension",
-        ),
-        (
-            "duplex_serving_adapter",
-            "vllm_omni.model_executor.models.personaplex.duplex.serving_adapter.PersonaPlexServingRuntimeAdapter",
-        ),
-    ],
-)
-def test_personaplex_pipeline_registers_unified_duplex_components(
-    attribute: str,
-    expected: str,
-) -> None:
-    assert PERSONAPLEX_PIPELINE.duplex_control_enabled is True
-    assert getattr(PERSONAPLEX_PIPELINE, attribute) == expected
-
-
-@pytest.mark.parametrize(
-    ("engine_arg", "expected"),
-    [
-        ("skip_tokenizer_init", True),
-        ("enable_prefix_caching", False),
-    ],
-)
-def test_personaplex_stage_engine_args(
-    engine_arg: str,
-    expected: bool,
-) -> None:
-    deploy_path = Path(__file__).parents[5] / "vllm_omni" / "deploy" / "personaplex.yaml"
-    stages = merge_pipeline_deploy(
-        PERSONAPLEX_PIPELINE,
-        load_deploy_config(deploy_path),
-    )
-
-    assert [stage.yaml_engine_args.get(engine_arg) for stage in stages] == [
-        expected,
-        expected,
-    ]
-
-
-def test_personaplex_duplex_capacity_is_propagated_to_all_model_stages() -> None:
-    deploy_path = Path(__file__).parents[5] / "vllm_omni" / "deploy" / "personaplex.yaml"
-    deploy = load_deploy_config(deploy_path)
-    stages = merge_pipeline_deploy(PERSONAPLEX_PIPELINE, deploy)
-
-    assert deploy.duplex_session.max_sessions == 2
-    assert [stage.yaml_engine_args.get("duplex_max_sessions") for stage in stages] == [2, 2]
-    assert "personaplex_codec_max_sessions" not in deploy.connectors["connector_of_shared_memory"]["extra"]
-
-
-def test_personaplex_capabilities_are_honest() -> None:
-    single = PersonaPlexServingRuntimeAdapter(lambda *_: None).capabilities(max_sessions=1)
-    multi = PersonaPlexServingRuntimeAdapter(lambda *_: None).capabilities(max_sessions=2)
-
-    assert multi.implementation_level == "model_native_duplex"
-    assert multi.input_modes == ["append_audio_chunk"]
-    assert multi.chunk_period_ms == 80
-    assert single.supports_multi_session is False
-    assert multi.supports_multi_session is True
-    assert multi.supports_multi_session_same_replica is True
-    assert multi.supports_barge_in is False
-
-
-def test_personaplex_runtime_config_update_rejects_changed_persona() -> None:
-    adapter = PersonaPlexServingRuntimeAdapter(lambda *_: None)
-    short_persona = "You are a helpful assistant."
-    longer_persona = (
-        "You are now a swashbuckling pirate captain who speaks entirely in "
-        "nautical slang, references buried treasure constantly, and never "
-        "breaks character no matter what the user asks."
-    )
-    current = {"personaplex_persona": short_persona, "personaplex_voice_prompt": "NATF2.pt"}
-
-    with pytest.raises(ServingRuntimeConfigError, match="persona"):
-        adapter.runtime_config_for_update(
-            SimpleNamespace(instructions=longer_persona, extra_body={}),
-            current,
-        )
-
-    unchanged = adapter.runtime_config_for_update(
-        SimpleNamespace(instructions=short_persona, extra_body={}),
-        current,
-    )
-    assert unchanged["personaplex_persona"] == short_persona
-
-
-def test_personaplex_runtime_config_update_rejects_changed_voice() -> None:
-    adapter = PersonaPlexServingRuntimeAdapter(lambda *_: None)
-    current = {"personaplex_persona": DEFAULT_PERSONA, "personaplex_voice_prompt": "NATF2.pt"}
-
-    with pytest.raises(ServingRuntimeConfigError, match="voice"):
-        adapter.runtime_config_for_update(
-            SimpleNamespace(instructions=None, voice="OtherVoice.pt", extra_body={}),
-            current,
-        )
-
-    unchanged = adapter.runtime_config_for_update(
-        SimpleNamespace(instructions=None, voice="NATF2.pt", extra_body={}),
-        current,
-    )
-    assert unchanged["personaplex_voice_prompt"] == "NATF2.pt"
-
-
-def test_personaplex_rejects_native_duplex_opt_out_flag() -> None:
-    # PersonaPlex is always model-native; the per-session opt-in knob is
-    # server-owned there, under both the canonical name and its deprecated
-    # model-prefixed alias.
-    adapter = PersonaPlexServingRuntimeAdapter(lambda *_: None)
-
-    with pytest.raises(ServingRuntimeConfigError, match="native_duplex"):
-        adapter.validate_client_extra_body({"native_duplex": False})
-    with pytest.raises(ServingRuntimeConfigError, match="minicpmo45_native_duplex"):
-        adapter.validate_client_extra_body({"minicpmo45_native_duplex": False})
-
-
-def test_pcm_buffer_emits_one_80ms_frame_transactionally() -> None:
-    buffer = PersonaPlexPcmAppendBuffer()
-    reservation = buffer.prepare_append(
-        _pcm_payload(np.arange(1920, dtype=np.float32)),
-        operation_id="op-1",
-        chunk_period_ms=80,
-        allow_emit=True,
-    )
-
-    assert reservation is not None
-    assert reservation.byte_count == 1920 * 4
-    assert buffer.pending_byte_count == 0
-    reservation.rollback()
-    assert buffer.pending_byte_count == 1920 * 4
-
-
-def test_pcm_buffer_rejects_changed_sample_rate() -> None:
-    buffer = PersonaPlexPcmAppendBuffer()
-    assert (
-        buffer.prepare_append(
-            _pcm_payload(np.zeros(100, np.float32)),
-            operation_id="op-1",
-            chunk_period_ms=80,
-            allow_emit=False,
-        )
-        is None
-    )
-
-    with pytest.raises(ValueError, match="sample_rate"):
-        buffer.prepare_append(
-            _pcm_payload(np.zeros(100, np.float32), sample_rate_hz=16000),
-            operation_id="op-2",
-            chunk_period_ms=80,
-            allow_emit=False,
-        )
-
-
-def test_pcm_buffer_rejects_non_finite_samples() -> None:
-    buffer = PersonaPlexPcmAppendBuffer()
-
-    with pytest.raises(ValueError, match="finite"):
-        buffer.prepare_append(
-            _pcm_payload(np.array([0.0, np.nan], np.float32)),
-            operation_id="op-1",
-            chunk_period_ms=80,
-            allow_emit=False,
-        )
-
-
-def test_runtime_extension_builds_one_frame_scheduler_append() -> None:
-    extension = PersonaPlexDuplexRuntimeExtension()
-    defaults = (
-        SamplingParams(temperature=0.8, top_k=10, max_tokens=10),
-        SamplingParams(max_tokens=1024),
-    )
-
-    configured = extension.configure_sampling_params(runtime_config={}, defaults=defaults)
-    assert configured[0].temperature == 0.0
-    assert configured[0].top_k == 1
-    assert configured[0].max_tokens == 1
-    assert configured[1] is defaults[1]
-
-    plan = extension.plan_append(
-        request_id="req",
-        fence=DuplexFence("session", incarnation=2, epoch=3),
-        session_config={"instructions": "Be concise."},
-        runtime_config={"personaplex_prefill_slots": 4},
-        seq=1,
-        turn_seq=1,
-        mode=DuplexInputMode.APPEND_AUDIO_CHUNK,
-        payload=_pcm_payload(np.zeros(1920, np.float32)),
-        final=False,
-        sampling_params=configured[0],
-    )
-    assert len(plan.prompt["prompt_token_ids"]) == 5
-    duplex = plan.prompt["model_intermediate_buffer"]["duplex"]
-    assert duplex["session_id"] == "session"
-    assert duplex["incarnation"] == 2
-    assert duplex["epoch"] == 3
-    assert duplex["seq"] == 1
-    assert duplex["data_plane"] is True
-
-
-def test_projector_emits_only_cumulative_audio_and_text_suffixes() -> None:
-    encoded_sizes: list[int] = []
-
-    def encode_audio(audio, _sample_rate, _response_format, _speed):
-        size = int(np.asarray(audio, dtype=np.float32).size)
-        encoded_sizes.append(size)
-        return f"audio-{size}"
-
-    projector = PersonaPlexDataPlaneSession(encode_audio)
-    projector.begin_request("req")
-    context = PersonaPlexDataPlaneContext(response_format="wav")
-
-    first = SimpleNamespace(
-        request_id="req",
-        outputs=[SimpleNamespace(text="he", multimodal_output={})],
-        multimodal_output={"model_outputs": np.arange(4, dtype=np.float32), "sr": 24000},
-        finished=False,
-    )
-    second = SimpleNamespace(
-        request_id="req",
-        outputs=[SimpleNamespace(text="hello", multimodal_output={})],
-        multimodal_output={"model_outputs": np.arange(6, dtype=np.float32), "sr": 24000},
-        finished=False,
-    )
-
-    projected = list(
-        projector.project(
-            {"data_plane_outputs": [first, second]},
-            context=context,
-        )
-    )
-
-    assert encoded_sizes == [4, 2]
-    assert [item["text"] for item in projected] == ["he", "llo"]
-    assert [item["audio_data"] for item in projected] == ["audio-4", "audio-2"]
-    assert all(item["sample_rate_hz"] == 24000 for item in projected)
-    assert all(item["end_of_turn"] is False for item in projected)