[Feature] PersonaPlex (Moshi-based full-duplex S2S): native vLLM port + duplex serving (#4771)

Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
Signed-off-by: Sy0307 <24841857+Sy0307@users.noreply.github.com>
Co-authored-by: Sy0307 <24841857+Sy0307@users.noreply.github.com>

tests/e2e/features/fullduplex/personaplex/test_import_boundary.py

@@ -0,0 +1,30 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+import subprocess
+import sys
+
+
+def test_batched_runtime_import_does_not_require_optional_opus_dependency() -> None:
+    script = """
+import builtins
+
+real_import = builtins.__import__
+
+def guarded_import(name, *args, **kwargs):
+    if name == "sphn":
+        raise ModuleNotFoundError("blocked optional dependency: sphn")
+    return real_import(name, *args, **kwargs)
+
+builtins.__import__ = guarded_import
+from vllm_omni.experimental.fullduplex.personaplex.serving.batched import BatchedSessionManager
+assert BatchedSessionManager is not None
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+
+    assert result.returncode == 0, result.stderr

tests/e2e/features/fullduplex/personaplex/test_stage0_runtime.py

@@ -0,0 +1,303 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+import base64
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import torch
+
+from vllm_omni.experimental.fullduplex.personaplex.policy import (
+    SILENCE_TOKENS,
+    SINE_TOKENS,
+)
+from vllm_omni.experimental.fullduplex.personaplex.stage0 import (
+    PersonaPlexStage0DuplexRuntime,
+)
+from vllm_omni.model_executor.models.personaplex.personaplex_talker import (
+    PersonaPlexTalkerForConditionalGeneration,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class _FakeCodec:
+    def __init__(self) -> None:
+        self.encode_calls = 0
+        self.reset_calls = 0
+
+    def streaming_init(self, batch_size: int) -> None:
+        assert batch_size == 1
+
+    def encode_frame(self, pcm: torch.Tensor) -> torch.Tensor:
+        assert pcm.shape == (1, 1920)
+        self.encode_calls += 1
+        return torch.full((1, 8), self.encode_calls, dtype=torch.long)
+
+    def reset_streaming(self) -> None:
+        self.reset_calls += 1
+
+
+class _FakeTalker:
+    def __init__(self) -> None:
+        self.device = torch.device("cpu")
+        self.dtype = torch.float32
+        self.frame_calls: list[dict[str, torch.Tensor | None]] = []
+
+    def _build_prefill_embed(
+        self,
+        tokens,
+        offset,
+        span,
+        device,
+        silence=None,
+        user_sine=None,
+    ):
+        del offset, silence, user_sine
+        values = tokens[:span].to(device=device, dtype=torch.float32)
+        return values[:, None].expand(-1, 4).contiguous()
+
+    def _build_frame_embed(
+        self,
+        text_token,
+        last_agent,
+        prev_agent,
+        device,
+        user_d0=None,
+        user_d1=None,
+    ):
+        self.frame_calls.append(
+            {
+                "text_token": text_token.clone(),
+                "last_agent": None if last_agent is None else last_agent.clone(),
+                "prev_agent": None if prev_agent is None else prev_agent.clone(),
+                "user_d0": None if user_d0 is None else user_d0.clone(),
+                "user_d1": None if user_d1 is None else user_d1.clone(),
+            }
+        )
+        value = float(user_d0[0].item()) if user_d0 is not None else -1.0
+        return torch.full((1, 4), value, device=device)
+
+
+def _duplex_info(*, seq: int, session_id: str = "session", incarnation: int = 1):
+    pcm = np.zeros(1920, dtype="<f4")
+    return {
+        "data_plane": True,
+        "session_id": session_id,
+        "incarnation": incarnation,
+        "epoch": 0,
+        "seq": seq,
+        "payload": {
+            "format": "pcm_f32le",
+            "sample_rate_hz": 24000,
+            "audio": base64.b64encode(pcm.tobytes()).decode("ascii"),
+        },
+        "runtime_config": {
+            "personaplex_model_path": "/unused",
+            "personaplex_voice_prompt": "NATF2.pt",
+            "personaplex_persona": "Be concise.",
+        },
+    }
+
+
+def _runtime(*codecs: _FakeCodec) -> PersonaPlexStage0DuplexRuntime:
+    voice_embeddings = torch.arange(8, dtype=torch.float32).reshape(2, 1, 1, 4)
+    codec_args = {"codec": codecs[0]}
+    if len(codecs) > 1:
+        available = iter(codecs)
+        codec_args = {
+            "codec_factory": lambda: next(available),
+            "max_sessions": len(codecs),
+        }
+    return PersonaPlexStage0DuplexRuntime(
+        _FakeTalker(),
+        model_path="/unused",
+        device="cpu",
+        **codec_args,
+        tokenizer=lambda _text: [7, 8, 9],
+        voice_loader=lambda _voice: {
+            "embeddings": voice_embeddings,
+            "cache": torch.zeros((1, 17, 4), dtype=torch.long),
+        },
+    )
+
+
+def test_first_append_prepends_voice_and_persona_once() -> None:
+    codec = _FakeCodec()
+    runtime = _runtime(codec)
+
+    first = runtime.prepare_append(_duplex_info(seq=1), prompt_len=18)
+    second = runtime.prepare_append(_duplex_info(seq=2), prompt_len=18)
+
+    assert first.prefill_applied is True
+    assert first.prompt_offset == 0
+    assert first.user_codes.shape == (1, 8)
+    assert first.inputs_embeds.shape == (18, 4)
+    assert second.prefill_applied is False
+    assert second.prompt_offset == 17
+    assert second.user_codes.shape == (2, 8)
+    assert second.inputs_embeds.shape == (1, 4)
+    assert codec.encode_calls == 2
+
+
+@pytest.mark.parametrize(
+    ("append_count", "expected_tokens", "expected_provided"),
+    [
+        (
+            1,
+            [*SILENCE_TOKENS, 1, *SINE_TOKENS[1:]],
+            [False, *([True] * 15)],
+        ),
+        (
+            2,
+            [*SILENCE_TOKENS, 2, *([1] * 7)],
+            [*([False] * 8), *([True] * 8)],
+        ),
+    ],
+)
+def test_prepare_append_preserves_native_depformer_teacher_forcing(
+    append_count: int,
+    expected_tokens: list[int],
+    expected_provided: list[bool],
+) -> None:
+    runtime = _runtime(_FakeCodec())
+
+    result = None
+    for seq in range(1, append_count + 1):
+        result = runtime.prepare_append(_duplex_info(seq=seq), prompt_len=18)
+
+    assert result is not None
+    assert result.info_update["pplex_depformer_audio_tokens"].tolist() == expected_tokens
+    assert result.info_update["pplex_depformer_audio_provided"].tolist() == expected_provided
+
+
+def test_frame_embed_uses_previous_effective_agent_frame() -> None:
+    fake_talker = SimpleNamespace(
+        config=SimpleNamespace(
+            num_audio_codebooks=16,
+            audio_vocab_size=2048,
+        ),
+        input_embeddings=lambda stack: stack,
+    )
+    last_agent = torch.arange(10, 18)
+
+    embeds = PersonaPlexTalkerForConditionalGeneration._build_frame_embed(
+        fake_talker,
+        torch.tensor([3]),
+        last_agent,
+        torch.arange(8),
+        torch.device("cpu"),
+    )
+
+    assert embeds.shape == (1, 17)
+    assert torch.equal(embeds[0, 1:9], last_agent)
+
+
+def test_repeated_append_identity_does_not_advance_codec() -> None:
+    codec = _FakeCodec()
+    runtime = _runtime(codec)
+
+    info = _duplex_info(seq=1)
+    first = runtime.prepare_append(info, prompt_len=18, request_id="req")
+    retry = runtime.prepare_append(info, prompt_len=18, request_id="req")
+
+    assert codec.encode_calls == 1
+    assert torch.equal(first.inputs_embeds, retry.inputs_embeds)
+    assert torch.equal(first.user_codes, retry.user_codes)
+
+
+def test_next_append_uses_prior_sample_and_causally_delayed_user_frame() -> None:
+    runtime = _runtime(_FakeCodec())
+
+    runtime.prepare_append(_duplex_info(seq=1), prompt_len=18, request_id="req")
+    first_agent = torch.arange(8, dtype=torch.long)
+    runtime.record_sample(request_id="req", text_token=torch.tensor(101), agent_codes=first_agent)
+    runtime.prepare_append(_duplex_info(seq=2), prompt_len=19, request_id="req")
+    second_agent = torch.arange(10, 18, dtype=torch.long)
+    runtime.record_sample(request_id="req", text_token=torch.tensor(102), agent_codes=second_agent)
+    runtime.prepare_append(_duplex_info(seq=3), prompt_len=20, request_id="req")
+
+    silence = torch.tensor(SILENCE_TOKENS, dtype=torch.long)
+    sine = torch.tensor(SINE_TOKENS, dtype=torch.long)
+    first_call = runtime.stage_model.frame_calls[0]
+    assert all(torch.equal(first_call[key], silence) for key in ("last_agent", "prev_agent"))
+    assert all(torch.equal(first_call[key], sine) for key in ("user_d0", "user_d1"))
+
+    second_call = runtime.stage_model.frame_calls[1]
+    assert second_call["text_token"].tolist() == [101]
+    expected_first_effective = torch.cat([first_agent[:1], silence[1:]])
+    assert torch.equal(second_call["last_agent"], expected_first_effective)
+    assert torch.equal(second_call["prev_agent"], expected_first_effective)
+    assert torch.equal(second_call["user_d0"], torch.full((8,), 1, dtype=torch.long))
+    assert torch.equal(second_call["user_d1"], sine)
+
+    third_call = runtime.stage_model.frame_calls[2]
+    assert third_call["text_token"].tolist() == [102]
+    assert torch.equal(third_call["last_agent"], second_agent)
+    assert torch.equal(third_call["prev_agent"], second_agent)
+    assert torch.equal(third_call["user_d0"], torch.full((8,), 2, dtype=torch.long))
+    assert torch.equal(third_call["user_d1"], torch.full((8,), 1, dtype=torch.long))
+
+
+def test_decoded_pcm_is_writable_for_torch_zero_copy() -> None:
+    pcm = PersonaPlexStage0DuplexRuntime._decode_pcm(_duplex_info(seq=1)["payload"])
+
+    assert pcm.flags.writeable
+
+
+def _prepare_two_sessions(
+    runtime: PersonaPlexStage0DuplexRuntime,
+) -> tuple[object, object, object, object]:
+    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),
+        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),
+        prompt_len=19,
+    )
+    return first_1, second_1, first_2, second_2
+
+
+def test_live_sessions_keep_independent_streaming_encoders() -> None:
+    first_codec = _FakeCodec()
+    second_codec = _FakeCodec()
+    runtime = _runtime(first_codec, second_codec)
+
+    first_1, second_1, first_2, second_2 = _prepare_two_sessions(runtime)
+
+    assert first_1.user_codes[:, 0].tolist() == [1]
+    assert second_1.user_codes[:, 0].tolist() == [1]
+    assert first_2.user_codes[:, 0].tolist() == [1, 2]
+    assert second_2.user_codes[:, 0].tolist() == [1, 2]
+
+
+def test_stage0_session_capacity_fails_before_codec_state_is_shared() -> None:
+    runtime = _runtime(_FakeCodec(), _FakeCodec())
+    _prepare_two_sessions(runtime)
+
+    with pytest.raises(RuntimeError, match="capacity 2"):
+        runtime.prepare_append(
+            _duplex_info(seq=1, session_id="third", incarnation=3),
+            prompt_len=18,
+        )
+
+
+def test_close_session_resets_and_reuses_released_codec() -> None:
+    first_codec = _FakeCodec()
+    second_codec = _FakeCodec()
+    runtime = _runtime(first_codec, second_codec)
+    _prepare_two_sessions(runtime)
+
+    runtime.close_session("session", 1)
+
+    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),
+        prompt_len=18,
+    )
+    assert replacement.user_codes[:, 0].tolist() == [3]

tests/e2e/features/fullduplex/personaplex/test_streaming_code2wav.py

@@ -0,0 +1,316 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from types import SimpleNamespace
+
+import pytest
+import torch
+from torch import nn
+
+from vllm_omni.model_executor.models.personaplex import (
+    personaplex_code2wav,
+    personaplex_mimi,
+)
+from vllm_omni.model_executor.models.personaplex.personaplex_code2wav import (
+    PersonaPlexCode2Wav,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class _FakeStreamingMimi(nn.Module):
+    def __init__(self, stream_value: int = 0) -> None:
+        super().__init__()
+        self.stream_value = stream_value
+        self.decode_frame_calls = 0
+        self.reset_calls = 0
+
+    def decode(self, codes: torch.Tensor, return_dict: bool = True):
+        assert return_dict
+        frames = int(codes.shape[-1])
+        return SimpleNamespace(audio_values=torch.arange(frames * 4, dtype=torch.float32).reshape(1, 1, -1))
+
+    def streaming_init(self, batch_size: int) -> None:
+        assert batch_size == 1
+
+    def decode_frame(self, codes: torch.Tensor) -> torch.Tensor:
+        assert codes.shape == (1, 2)
+        self.decode_frame_calls += 1
+        value = self.stream_value + self.decode_frame_calls
+        return torch.full((1, 4), float(value), dtype=torch.float32)
+
+    def reset_streaming(self) -> None:
+        self.reset_calls += 1
+
+
+class _FakeChunkedMimi(_FakeStreamingMimi):
+    def __init__(self) -> None:
+        super().__init__()
+        self.decode_frames_calls = 0
+        self.decode_frames_shapes: list[tuple[int, ...]] = []
+
+    def decode_frames(self, codes: torch.Tensor) -> torch.Tensor:
+        self.decode_frames_calls += 1
+        self.decode_frames_shapes.append(tuple(codes.shape))
+        return torch.arange(codes.shape[-1] * 4, dtype=torch.float32).reshape(1, -1)
+
+
+class _FakeMimiModel(nn.Module):
+    def __init__(self, _config) -> None:
+        super().__init__()
+        self.encoder_transformer = nn.Linear(1, 1, bias=False)
+        self.decoder_transformer = nn.Linear(1, 1, bias=False)
+        self.quantizer = nn.Linear(1, 1, bias=False)
+        self.encoder = SimpleNamespace(layers=[])
+        self.downsample = SimpleNamespace(conv=nn.Conv1d(1, 1, 1))
+        self.upsample = SimpleNamespace(conv=nn.ConvTranspose1d(1, 1, 1))
+        self.decoder = SimpleNamespace(layers=[])
+
+    def load_state_dict(self, _state_dict, strict: bool = True):
+        assert not strict
+        return SimpleNamespace(
+            missing_keys=[
+                "encoder_transformer.weight",
+                "decoder_transformer.weight",
+            ],
+            unexpected_keys=[],
+        )
+
+
+class _FakeMimiTransformer(nn.Module):
+    def __init__(self) -> None:
+        super().__init__()
+        self.weight = nn.Parameter(torch.zeros(1))
+
+    def load_weights(self, _state_dict, _prefix: str) -> int:
+        return 80
+
+
+def test_moshi_checkpoint_mapping_drops_replaced_transformer_weights(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    mapper = getattr(personaplex_mimi, "_map_moshi_codec_weights", None)
+    assert callable(mapper), "PersonaPlex Mimi must load its bundled codec weights without Hugging Face"
+
+    source = {
+        "encoder.model.3.conv.conv.weight": torch.ones(1),
+        "downsample.conv.conv.conv.weight": torch.ones(1),
+        "quantizer.rvq_first.vq.layers.0._codebook.embedding_sum": torch.ones(1),
+        "quantizer.rvq_rest.vq.layers.3._codebook._initialized": torch.ones(1),
+        "encoder_transformer.transformer.layers.0.norm1.weight": torch.ones(1),
+    }
+
+    mapped = mapper(source)
+
+    assert set(mapped) == {
+        "encoder.layers.3.conv.weight",
+        "downsample.conv.weight",
+        "quantizer.semantic_residual_vector_quantizer.layers.0.codebook.embed_sum",
+        "quantizer.acoustic_residual_vector_quantizer.layers.3.codebook.initialized",
+    }
+    assert mapped["encoder.layers.3.conv.weight"] is source["encoder.model.3.conv.conv.weight"]
+    monkeypatch.setattr("transformers.MimiConfig", lambda: object())
+    monkeypatch.setattr("transformers.MimiModel", _FakeMimiModel)
+    monkeypatch.setattr("safetensors.torch.load_file", lambda *_args, **_kwargs: {})
+    monkeypatch.setattr(personaplex_mimi, "_MimiStreamingTransformer", _FakeMimiTransformer)
+
+    codec = personaplex_mimi.PersonaPlexMimiCodec(
+        checkpoint="/unused",
+        device="cpu",
+    )
+
+    assert not hasattr(codec.model, "encoder_transformer")
+    assert not hasattr(codec.model, "decoder_transformer")
+    assert not any(
+        name.startswith(("model.encoder_transformer.", "model.decoder_transformer."))
+        for name, _ in codec.named_parameters()
+    )
+
+
+def test_mimi_full_stream_reset_reuses_all_state_storage_and_clears_offsets() -> None:
+    codec = personaplex_mimi.PersonaPlexMimiCodec.__new__(personaplex_mimi.PersonaPlexMimiCodec)
+    nn.Module.__init__(codec)
+    codec.device = torch.device("cpu")
+    codec.dtype = torch.float32
+    codec._enc_stages = [("conv", personaplex_mimi._StreamConv1d(nn.Conv1d(1, 1, 3)))]
+    codec._dec_stages = []
+    codec._downsample = personaplex_mimi._StreamConv1d(nn.Conv1d(1, 1, 3))
+    codec._upsample = personaplex_mimi._StreamConvTr1d(nn.ConvTranspose1d(1, 1, 3))
+    codec.encoder_transformer = personaplex_mimi._MimiStreamingTransformer(
+        num_layers=1,
+        dim=4,
+        num_heads=1,
+        context=3,
+    )
+    codec.decoder_transformer = personaplex_mimi._MimiStreamingTransformer(
+        num_layers=1,
+        dim=4,
+        num_heads=1,
+        context=3,
+    )
+    codec.streaming_init(batch_size=2)
+
+    conv_states = list(codec._conv_states())
+    conv_buffers = [
+        state.prev if isinstance(state, personaplex_mimi._StreamConv1d) else state.partial for state in conv_states
+    ]
+    transformers = [codec.encoder_transformer, codec.decoder_transformer]
+    kv_states = [kv for transformer in transformers for kv in transformer._kv]
+    state_tensors = [
+        *conv_buffers,
+        *(kv.cache for kv in kv_states),
+        *(kv.end_offset for kv in kv_states),
+        *(kv.start_offset for kv in kv_states),
+        *(transformer._offset for transformer in transformers),
+    ]
+    for tensor in state_tensors:
+        tensor.fill_(1)
+    for state in conv_states:
+        state._fresh.zero_()
+    storage = [tensor.data_ptr() for tensor in state_tensors]
+
+    codec.reset_streaming()
+
+    assert storage == [tensor.data_ptr() for tensor in state_tensors]
+    cleared = [
+        *conv_buffers,
+        *(kv.end_offset for kv in kv_states),
+        *(kv.start_offset for kv in kv_states),
+        *(transformer._offset for transformer in transformers),
+    ]
+    assert all(not tensor.any() for tensor in cleared)
+    assert all(state._fresh.all() for state in conv_states)
+
+
+def _model(*, max_sessions: int = 1) -> tuple[PersonaPlexCode2Wav, _FakeStreamingMimi]:
+    mimi_config = SimpleNamespace(num_codebooks=2, sample_rate=24000, samples_per_frame=4, mimi_name=None)
+    config = SimpleNamespace(mimi_config=mimi_config, mimi_name=None)
+    vllm_config = SimpleNamespace(
+        model_config=SimpleNamespace(
+            model="/unused",
+            hf_config=config,
+            duplex_max_sessions=max_sessions,
+        ),
+        device_config=SimpleNamespace(device="cpu"),
+    )
+    model = PersonaPlexCode2Wav(vllm_config=vllm_config)
+    mimi = _FakeStreamingMimi()
+    model.mimi = mimi
+    model._mimi_device = torch.device("cpu")
+    return model, mimi
+
+
+def _codes(frames: int, *, start: int = 0) -> torch.Tensor:
+    return torch.stack([torch.arange(start + offset, start + offset + frames) for offset in (0, 100)]).reshape(-1)
+
+
+def _audio(output) -> torch.Tensor:
+    return output.multimodal_outputs["model_outputs"][0]
+
+
+@pytest.mark.parametrize("second_codes", [_codes(3), _codes(1, start=100)])
+def test_resumable_codes_emit_only_new_pcm(second_codes: torch.Tensor) -> None:
+    model, mimi = _model()
+
+    first = model(input_ids=_codes(2), request_ids=["req"])
+    second = model(input_ids=second_codes, request_ids=["req"])
+
+    assert _audio(first).numel() == 8
+    assert _audio(second).numel() == 4
+    assert mimi.decode_frame_calls == 3
+
+
+@pytest.mark.parametrize(
+    ("frames", "expected_shapes"),
+    [
+        (5, [(1, 2, 5)]),
+        (12, [(1, 2, 5), (1, 2, 5), (1, 2, 2)]),
+    ],
+)
+def test_streaming_decode_uses_bounded_multi_frame_calls(
+    frames: int,
+    expected_shapes: list[tuple[int, ...]],
+) -> None:
+    model, _ = _model()
+    mimi = _FakeChunkedMimi()
+    model.mimi = mimi
+
+    output = model(input_ids=_codes(frames), request_ids=["req"])
+
+    assert _audio(output).numel() == frames * 4
+    assert mimi.decode_frames_shapes == expected_shapes
+    assert mimi.decode_frame_calls == 0
+
+
+def test_profile_inputs_skip_decode_while_malformed_online_input_warns(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    model, mimi = _model()
+    runtime_info = [{"meta": {"personaplex_dummy_profile": True}}]
+    warnings: list[tuple] = []
+    monkeypatch.setattr(personaplex_code2wav.logger, "warning", lambda *args: warnings.append(args))
+
+    output = model(
+        input_ids=torch.arange(3),
+        runtime_additional_information=runtime_info,
+    )
+
+    assert _audio(output).numel() == 0
+    assert mimi.decode_frame_calls == 0
+    assert warnings == []
+
+    output = model(input_ids=torch.arange(3))
+
+    assert _audio(output).numel() == 0
+    assert mimi.decode_frame_calls == 0
+    assert len(warnings) == 1
+    assert "not divisible by" in warnings[0][0]
+    assert model.get_dummy_runtime_additional_information(2) == [
+        {"meta": {"personaplex_dummy_profile": True}},
+        {"meta": {"personaplex_dummy_profile": True}},
+    ]
+
+
+def test_request_id_falls_back_to_runtime_information() -> None:
+    model, _ = _model()
+    info = [{"request_id": "runtime-req"}]
+
+    model(input_ids=_codes(2), runtime_additional_information=info)
+    second = model(input_ids=_codes(3), runtime_additional_information=info)
+
+    assert _audio(second).numel() == 4
+
+
+def test_decoder_slot_lifecycle_isolated_capacity_and_reuse() -> None:
+    model, _ = _model(max_sessions=2)
+    first = _FakeStreamingMimi(stream_value=10)
+    second = _FakeStreamingMimi(stream_value=20)
+    model._set_mimi_codecs([first, second])
+    assert model._max_codec_sessions == 2
+
+    initial = model(
+        input_ids=torch.cat([_codes(1), _codes(1)]),
+        request_ids=["first", "second"],
+        seq_token_counts=[2, 2],
+    )
+    continued = model(
+        input_ids=torch.cat([_codes(1, start=10), _codes(1, start=20)]),
+        request_ids=["first", "second"],
+        seq_token_counts=[2, 2],
+    )
+
+    initial_audio = initial.multimodal_outputs["model_outputs"]
+    continued_audio = continued.multimodal_outputs["model_outputs"]
+    assert initial_audio[0].tolist() == [11.0] * 4
+    assert initial_audio[1].tolist() == [21.0] * 4
+    assert continued_audio[0].tolist() == [12.0] * 4
+    assert continued_audio[1].tolist() == [22.0] * 4
+    assert model._request_codec_slots == {"first": 0, "second": 1}
+    with pytest.raises(RuntimeError, match="decoder capacity 2 is exhausted"):
+        model._codec_for_request("overflow")
+
+    model.on_requests_finished(["first"])
+    assert first.reset_calls == 1
+    assert second.reset_calls == 0
+    replacement = model(input_ids=_codes(1), request_ids=["replacement"])
+    assert _audio(replacement).tolist() == [13.0] * 4

tests/e2e/features/fullduplex/personaplex/test_unified_runtime.py

@@ -0,0 +1,364 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+import base64
+from pathlib import Path
+from types import SimpleNamespace
+
+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.experimental.fullduplex.engine.duplex_runtime import (
+    DuplexInputMode,
+)
+from vllm_omni.experimental.fullduplex.engine.messages import DuplexFence
+from vllm_omni.experimental.fullduplex.personaplex.data_plane import (
+    PersonaPlexDataPlaneContext,
+    PersonaPlexDataPlaneSession,
+)
+from vllm_omni.experimental.fullduplex.personaplex.input import (
+    PersonaPlexPcmAppendBuffer,
+)
+from vllm_omni.experimental.fullduplex.personaplex.runtime_extension import (
+    PersonaPlexDuplexRuntimeExtension,
+)
+from vllm_omni.experimental.fullduplex.personaplex.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.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.experimental.fullduplex.personaplex.runtime_extension.PersonaPlexDuplexRuntimeExtension",
+        ),
+        (
+            "duplex_serving_adapter",
+            "vllm_omni.experimental.fullduplex.personaplex.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_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)

tests/e2e/features/fullduplex/test_core_contracts.py

@@ -0,0 +1,122 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Contract tests for the fullduplex core (PR #3907 alignment).
+
+Covers the DuplexFence identity value, the session identity rules (epoch only
+advances on barge-in, response_index once per begin_response), stale-output
+dropping in the turn lifecycle. The lockstep (single-eternal-response)
+lifecycle is model-owned (see
+tests/e2e/features/fullduplex/test_personaplex_adapter.py).
+"""
+
+import asyncio
+import dataclasses
+
+import pytest
+
+from vllm_omni.experimental.fullduplex.core import protocol as ev
+from vllm_omni.experimental.fullduplex.core.adapter import (
+    DuplexAdapter,
+    DuplexCapability,
+    OutputChunk,
+)
+from vllm_omni.experimental.fullduplex.core.runtime import DuplexRuntime
+from vllm_omni.experimental.fullduplex.core.session import (
+    DuplexSession,
+    DuplexSessionConfig,
+    DuplexState,
+)
+from vllm_omni.experimental.fullduplex.engine.messages import DuplexFence
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_fence_is_frozen_value_identity():
+    f = DuplexFence(session_id="s1")
+    assert (f.epoch, f.turn_id, f.response_seq) == (0, 0, 0)
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        f.epoch = 1  # type: ignore[misc]
+    assert DuplexFence("s1", 1, 2, 3) == DuplexFence("s1", 1, 2, 3)
+    assert DuplexFence("s1", 1, 2, 3) != DuplexFence("s1", 2, 2, 3)
+
+
+def test_session_identity_rules():
+    s = DuplexSession(session_id="test", config=DuplexSessionConfig())
+    r1, e1 = s.begin_response()
+    r2, e2 = s.begin_response()
+    assert r2 == r1 + 1, "response index advances exactly once per begin"
+    assert e1 == e2 == 0, "epoch does not move on response begin"
+    s.end_response()
+    assert s.epoch == 0, "epoch does not move on response end"
+    new_epoch = s.barge_in()
+    assert new_epoch == 1, "epoch advances only on barge-in"
+    assert s.is_stale(e1) and not s.is_stale(new_epoch)
+
+
+class _TurnAdapter(DuplexAdapter):
+    """Emits chunks until cancelled; counts respond() lifecycles."""
+
+    def __init__(self):
+        self.responses = 0
+        self.barge_ins = 0
+
+    def capabilities(self) -> DuplexCapability:
+        return DuplexCapability(
+            input_modalities=frozenset({"audio"}),
+            output_modalities=frozenset({"audio"}),
+            proactive=True,
+        )
+
+    async def on_input(self, session, modality, data) -> None:
+        pass
+
+    def should_respond(self, session) -> bool:
+        return True
+
+    async def on_barge_in(self, session) -> None:
+        self.barge_ins += 1
+
+    async def respond(self, session):
+        self.responses += 1
+        for i in range(100):
+            await asyncio.sleep(0.001)
+            yield OutputChunk("audio", i)
+
+
+async def _drive(runtime, events):
+    async def gen():
+        for e in events:
+            await asyncio.sleep(0.005)
+            yield e
+
+    out = []
+
+    async def emit(e):
+        out.append(e)
+
+    await runtime.run(gen(), emit)
+    return out
+
+
+def test_turn_mode_supersede_and_stale_drop():
+    adapter = _TurnAdapter()
+    session = DuplexSession(session_id="test", config=DuplexSessionConfig(proactive=True))
+    runtime = DuplexRuntime(session, adapter)
+    out = asyncio.run(
+        _drive(
+            runtime,
+            [
+                {"type": ev.RESPONSE_CREATE},
+                {"type": ev.RESPONSE_CREATE},  # supersedes the first
+                {"type": ev.CLOSE},
+            ],
+        )
+    )
+    assert adapter.responses == 2
+    assert adapter.barge_ins == 1, "superseding an in-flight response barges in"
+    created = [e for e in out if e["type"] == ev.RESPONSE_CREATED]
+    done = [e for e in out if e["type"] == ev.RESPONSE_DONE]
+    assert len(created) == 2
+    assert len(done) == 1, "stale first response must not emit done"
+    assert done[0]["response_index"] == created[1]["response_index"]
+    assert session.state is DuplexState.CLOSED

tests/e2e/features/fullduplex/test_personaplex_adapter.py

@@ -0,0 +1,262 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""PersonaPlex adapter driven through the model-owned lockstep runtime.
+
+PersonaPlexDuplexRuntime starts ONE response for the whole session, streams one
+agent frame per user frame, and drains on close (core stays verbatim upstream;
+the lockstep lifecycle is model policy) — all with a GPU-free stub stepper.
+"""
+
+import asyncio
+
+import numpy as np
+import pytest
+
+from vllm_omni.experimental.fullduplex.core import protocol as ev
+from vllm_omni.experimental.fullduplex.core.session import (
+    DuplexSession,
+    DuplexSessionConfig,
+    DuplexState,
+)
+from vllm_omni.experimental.fullduplex.personaplex.adapter import (
+    PersonaPlexDuplexAdapter,
+    PersonaPlexDuplexRuntime,
+)
+from vllm_omni.experimental.fullduplex.personaplex.config import FRAME_SIZE
+from vllm_omni.experimental.fullduplex.personaplex.engine import FrameOutput
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class StubStepper:
+    """Deterministic FrameStepper: frame i -> audio filled with i, text 't{i}' on odds."""
+
+    sample_rate = 24000
+    frame_size = FRAME_SIZE
+
+    def __init__(self) -> None:
+        self.opened: tuple[str | None, str | None] | None = None
+        self.steps = 0
+
+    def open_session(self, voice_prompt=None, persona=None) -> None:
+        self.opened = (voice_prompt, persona)
+        self.steps = 0
+
+    def step(self, user_pcm):
+        assert user_pcm.shape[0] == self.frame_size
+        self.steps += 1
+        i = self.steps
+        return FrameOutput(
+            audio=np.full(self.frame_size, float(i), dtype=np.float32),
+            text=(f"t{i}" if i % 2 else None),
+        )
+
+
+class RecordingAdapter(PersonaPlexDuplexAdapter):
+    def __init__(self, stepper: StubStepper) -> None:
+        super().__init__(stepper)
+        self.closed = False
+
+    async def on_close(self, session: DuplexSession) -> None:
+        self.closed = True
+        await super().on_close(session)
+
+
+def _collector():
+    out: list[dict] = []
+
+    async def emit(event: dict) -> None:
+        out.append(event)
+
+    return out, emit
+
+
+async def _feed(events):
+    for e in events:
+        yield e
+
+
+async def _raising_feed(frame):
+    yield {"type": ev.INPUT_APPEND, "modality": "audio", "data": frame}
+    await asyncio.sleep(0)
+    raise RuntimeError("input failed")
+
+
+def _respond_tasks() -> list[asyncio.Task]:
+    current = asyncio.current_task()
+    return [
+        task
+        for task in asyncio.all_tasks()
+        if task is not current and "_respond_once" in getattr(task.get_coro(), "__qualname__", "")
+    ]
+
+
+def _session():
+    cfg = DuplexSessionConfig(
+        input_modalities=("audio",),
+        output_modalities=("audio", "text"),
+    )
+    return DuplexSession("s", cfg)
+
+
+@pytest.mark.asyncio
+async def test_lockstep_one_agent_frame_per_user_frame():
+    stub = StubStepper()
+    adapter = PersonaPlexDuplexAdapter(stub, voice_prompt="NATF2.pt", persona="be terse")
+    rt = PersonaPlexDuplexRuntime(_session(), adapter)
+    out, emit = _collector()
+
+    frame = np.zeros(FRAME_SIZE, dtype=np.float32)
+    events = [{"type": ev.INPUT_APPEND, "modality": "audio", "data": frame} for _ in range(3)]
+    events.append({"type": ev.CLOSE})
+    await rt.run(_feed(events), emit)
+
+    types = [e["type"] for e in out]
+    # exactly one response for the whole session, drained cleanly on close
+    assert types.count(ev.RESPONSE_CREATED) == 1
+    assert types.count(ev.RESPONSE_DONE) == 1
+    audio = [e for e in out if e["type"] == ev.RESPONSE_DELTA and e["modality"] == "audio"]
+    text = [e["data"] for e in out if e["type"] == ev.RESPONSE_DELTA and e["modality"] == "text"]
+    assert len(audio) == 3  # one agent frame per user frame
+    assert [int(e["data"][0]) for e in audio] == [1, 2, 3]  # lockstep order
+    assert text == ["t1", "t3"]
+    assert stub.steps == 3
+    assert stub.opened == ("NATF2.pt", "be terse")
+
+
+@pytest.mark.asyncio
+async def test_unaligned_chunks_are_reframed_to_80ms():
+    stub = StubStepper()
+    adapter = PersonaPlexDuplexAdapter(stub)
+    rt = PersonaPlexDuplexRuntime(_session(), adapter)
+    out, emit = _collector()
+
+    # 2.5 frames in one chunk + 0.5 frame in another => 3 whole frames
+    events = [
+        {"type": ev.INPUT_APPEND, "modality": "audio", "data": np.zeros(FRAME_SIZE * 5 // 2, dtype=np.float32)},
+        {"type": ev.INPUT_APPEND, "modality": "audio", "data": np.zeros(FRAME_SIZE // 2, dtype=np.float32)},
+        {"type": ev.CLOSE},
+    ]
+    await rt.run(_feed(events), emit)
+    audio = [e for e in out if e["type"] == ev.RESPONSE_DELTA and e["modality"] == "audio"]
+    assert len(audio) == 3
+    assert stub.steps == 3
+
+
+@pytest.mark.asyncio
+async def test_close_without_input_is_clean():
+    adapter = PersonaPlexDuplexAdapter(StubStepper())
+    rt = PersonaPlexDuplexRuntime(_session(), adapter)
+    out, emit = _collector()
+    await rt.run(_feed([{"type": ev.CLOSE}]), emit)
+    # no input -> response never started -> no created/done, no error
+    assert not any(e["type"] == ev.ERROR for e in out)
+    assert not any(e["type"] == ev.RESPONSE_CREATED for e in out)
+
+
+@pytest.mark.asyncio
+async def test_iterator_eof_without_close_drains_and_stops():
+    # Transport drop: the input iterator ends WITHOUT an explicit CLOSE event.
+    # The runtime must still signal on_close so the eternal response drains
+    # instead of blocking on its inbox forever.
+    stub = StubStepper()
+    adapter = PersonaPlexDuplexAdapter(stub)
+    rt = PersonaPlexDuplexRuntime(_session(), adapter)
+    out, emit = _collector()
+    frame = np.zeros(FRAME_SIZE, dtype=np.float32)
+    events = [{"type": ev.INPUT_APPEND, "modality": "audio", "data": frame}]
+    await asyncio.wait_for(rt.run(_feed(events), emit), timeout=5.0)
+    types = [e["type"] for e in out]
+    assert types.count(ev.RESPONSE_DONE) == 1
+    assert stub.steps == 1
+
+
+@pytest.mark.asyncio
+async def test_input_exception_closes_response_and_runtime() -> None:
+    stub = StubStepper()
+    adapter = RecordingAdapter(stub)
+    rt = PersonaPlexDuplexRuntime(_session(), adapter)
+    out, emit = _collector()
+    frame = np.zeros(FRAME_SIZE, dtype=np.float32)
+
+    with pytest.raises(RuntimeError, match="input failed"):
+        await rt.run(_raising_feed(frame), emit)
+
+    leaked = _respond_tasks()
+    for task in leaked:
+        task.cancel()
+    if leaked:
+        await asyncio.gather(*leaked, return_exceptions=True)
+
+    assert adapter.closed is True
+    assert rt.session.state is DuplexState.CLOSED
+    assert not leaked
+
+
+@pytest.mark.asyncio
+async def test_default_session_config_gets_lockstep_lifecycle():
+    # The lifecycle comes from the model-owned runtime class, not a session
+    # flag: a plain default-constructed session config runs lockstep.
+    stub = StubStepper()
+    adapter = PersonaPlexDuplexAdapter(stub)
+    session = DuplexSession(
+        "s",
+        DuplexSessionConfig(input_modalities=("audio",), output_modalities=("audio", "text")),
+    )
+    rt = PersonaPlexDuplexRuntime(session, adapter)
+    out, emit = _collector()
+    frame = np.zeros(FRAME_SIZE, dtype=np.float32)
+    events = [{"type": ev.INPUT_APPEND, "modality": "audio", "data": frame}, {"type": ev.CLOSE}]
+    await rt.run(_feed(events), emit)
+    types = [e["type"] for e in out]
+    assert types.count(ev.RESPONSE_CREATED) == 1
+    assert types.count(ev.RESPONSE_DONE) == 1
+    assert stub.steps == 1
+
+
+@pytest.mark.asyncio
+async def test_partial_tail_is_padded_and_stepped_on_close():
+    # A trailing sub-frame chunk is zero-padded and stepped at close (same
+    # contract as PersonaPlexSession.flush) rather than silently dropped.
+    stub = StubStepper()
+    adapter = PersonaPlexDuplexAdapter(stub)
+    rt = PersonaPlexDuplexRuntime(_session(), adapter)
+    out, emit = _collector()
+    events = [
+        {"type": ev.INPUT_APPEND, "modality": "audio", "data": np.ones(FRAME_SIZE // 2, dtype=np.float32)},
+        {"type": ev.CLOSE},
+    ]
+    await rt.run(_feed(events), emit)
+    audio = [e for e in out if e["type"] == ev.RESPONSE_DELTA and e["modality"] == "audio"]
+    assert len(audio) == 1
+    assert stub.steps == 1
+
+
+@pytest.mark.asyncio
+async def test_single_eternal_response_across_barge_in():
+    # Moved from the core contract tests when the lockstep lifecycle became
+    # model-owned: ONE response for the whole session, RESPONSE_CANCEL bumps the
+    # epoch fence (playback cursor) but the stream keeps flowing, and close
+    # drains via on_close.
+    stub = StubStepper()
+    adapter = PersonaPlexDuplexAdapter(stub)
+    session = _session()
+    rt = PersonaPlexDuplexRuntime(session, adapter)
+    out, emit = _collector()
+
+    frame = np.zeros(FRAME_SIZE, dtype=np.float32)
+    events = [
+        {"type": ev.INPUT_APPEND, "modality": "audio", "data": frame},
+        {"type": ev.RESPONSE_CANCEL},  # barge-in: epoch bumps, stream continues
+        {"type": ev.INPUT_APPEND, "modality": "audio", "data": frame},
+        {"type": ev.CLOSE},
+    ]
+    await rt.run(_feed(events), emit)
+
+    types = [e["type"] for e in out]
+    assert types.count(ev.RESPONSE_CREATED) == 1, "exactly one eternal response"
+    assert types.count(ev.RESPONSE_DONE) == 1
+    audio = [e for e in out if e["type"] == ev.RESPONSE_DELTA and e["modality"] == "audio"]
+    assert len(audio) == 2, "stream keeps flowing across barge-in"
+    assert session.epoch == 1, "cancel still advanced the epoch fence"
+    assert stub.steps == 2

tests/e2e/features/fullduplex/test_personaplex_batched.py

@@ -0,0 +1,178 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""GPU-free tests for the batched PersonaPlex slot manager (stub engine)."""
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from vllm_omni.experimental.fullduplex.personaplex.engine import FrameOutput, PrefillStep
+from vllm_omni.experimental.fullduplex.personaplex.serving.batched import (
+    BatchedSessionManager,
+    SlotPhase,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+FRAME = 4
+
+
+class StubEngine:
+    """Records the slot-API call sequence; echoes each row's pcm as its output."""
+
+    frame_size = FRAME
+
+    def __init__(self, batch_size: int = 4, prefill_len: int = 2) -> None:
+        self.config = SimpleNamespace(batch_size=batch_size)
+        self.prefill_len = prefill_len
+        self.recycled: list[int] = []
+        self.voice_ended: list[int] = []
+        self.prefill_done: list[int] = []
+        self.steps: list[tuple[np.ndarray, dict | None]] = []
+
+    def prefill_steps(self, persona=None):
+        steps = [PrefillStep("sacrifice"), PrefillStep("voice", embedding=None), PrefillStep("voice_end")]
+        steps += [PrefillStep("tokens", moshi_tokens=None, text_token=3, user_sine=True)] * self.prefill_len
+        return steps
+
+    def recycle_slot(self, b):
+        self.recycled.append(b)
+
+    def voice_end(self, b):
+        self.voice_ended.append(b)
+
+    def end_prefill(self, b):
+        self.prefill_done.append(b)
+
+    def step_batch(self, pcm, prefill=None):
+        self.steps.append((pcm.copy(), dict(prefill) if prefill else None))
+        return [FrameOutput(audio=pcm[b].copy(), text=None) for b in range(self.config.batch_size)]
+
+
+def test_all_idle_does_not_step():
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng)
+    assert mgr.tick() is False
+    assert eng.steps == []
+
+
+def test_pristine_slot_is_live_immediately():
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng)
+    slot, ready_now = mgr.acquire()
+    assert ready_now is True
+    assert slot.phase is SlotPhase.LIVE
+    assert eng.recycled == []
+
+
+def test_custom_persona_prefills_and_recycles_on_tick_thread():
+    eng = StubEngine(prefill_len=2)
+    mgr = BatchedSessionManager(eng)
+    slot, ready_now = mgr.acquire(persona="custom")
+    assert ready_now is False
+    assert slot.phase is SlotPhase.PREFILL
+    assert eng.recycled == []  # deferred: reset must run between engine steps
+
+    ready = []
+    slot.on_ready = lambda: ready.append(True)
+
+    mgr.tick()  # recycle + sacrifice
+    assert eng.recycled == [slot.index]
+    assert eng.steps[-1][1][slot.index].kind == "sacrifice"
+
+    mgr.tick()  # voice frame
+    assert eng.steps[-1][1][slot.index].kind == "voice"
+    assert eng.voice_ended == []
+
+    mgr.tick()  # voice_end executed BETWEEN ticks, then first tokens frame
+    assert eng.voice_ended == [slot.index]
+    assert eng.steps[-1][1][slot.index].kind == "tokens"
+
+    mgr.tick()  # second tokens frame
+    assert not ready
+    mgr.tick()  # prefill exhausted -> live
+    assert eng.prefill_done == [slot.index]
+    assert slot.phase is SlotPhase.LIVE
+    assert ready == [True]
+
+
+def test_pristine_slot_after_ticking_requires_recycle():
+    # step_batch advances EVERY row each tick, so a pristine idle row that has
+    # lived through ticks has drifted from its boot prefill and must be
+    # recycled before a client may join it.
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng)
+    first, ready_now = mgr.acquire()
+    assert ready_now is True  # claimed before any tick: truly pristine
+    mgr.tick()
+    second, ready_now2 = mgr.acquire()
+    assert second.index != first.index
+    assert ready_now2 is False
+    assert second.phase is SlotPhase.PREFILL
+
+
+def test_used_slot_requires_recycle_even_with_default_persona():
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng)
+    slot, ready_now = mgr.acquire()
+    assert ready_now
+    mgr.release(slot)
+    slot2, ready_now2 = mgr.acquire()
+    assert slot2.index == slot.index
+    assert ready_now2 is False
+    assert slot2.phase is SlotPhase.PREFILL
+
+
+def test_acquire_returns_none_when_full():
+    eng = StubEngine(batch_size=2)
+    mgr = BatchedSessionManager(eng)
+    assert mgr.acquire() is not None
+    assert mgr.acquire() is not None
+    assert mgr.acquire() is None
+
+
+def test_feed_frames_epoch_guard_and_backlog_cap():
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng, max_buffered_frames=3)
+    slot, _ = mgr.acquire()
+    mgr.feed(slot, np.arange(FRAME * 2, dtype=np.float32), slot.epoch)
+    assert len(slot.inq) == 2
+    mgr.feed(slot, np.zeros(3, dtype=np.float32), slot.epoch)  # partial tail buffered
+    assert len(slot.inq) == 2 and slot.pending.shape[0] == 3
+    mgr.feed(slot, np.zeros(FRAME * 5, dtype=np.float32), slot.epoch)
+    assert len(slot.inq) == 3  # oldest dropped beyond the cap
+    mgr.feed(slot, np.zeros(FRAME, dtype=np.float32), slot.epoch - 1)  # stale epoch ignored
+    assert len(slot.inq) == 3
+
+
+def test_live_slot_consumes_input_and_emits_output():
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng)
+    slot, _ = mgr.acquire()
+    got = []
+    slot.on_output = got.append
+    frame = np.full(FRAME, 0.5, dtype=np.float32)
+    mgr.feed(slot, frame, slot.epoch)
+    mgr.tick()
+    assert len(got) == 1
+    np.testing.assert_array_equal(got[0].audio, frame)
+    # queue drained: next tick feeds zeros
+    mgr.tick()
+    np.testing.assert_array_equal(got[1].audio, np.zeros(FRAME, dtype=np.float32))
+
+
+def test_release_mid_flight_stops_emission():
+    eng = StubEngine()
+    mgr = BatchedSessionManager(eng)
+    slot, _ = mgr.acquire()
+    got = []
+    slot.on_output = got.append
+    mgr.tick()
+    assert len(got) == 1
+    mgr.release(slot)
+    # another slot keeps the batch non-idle
+    other, _ = mgr.acquire()  # same slot index, now PREFILL (used)
+    assert other.phase is SlotPhase.PREFILL
+    mgr.tick()
+    assert len(got) == 1  # no further output for the released epoch

tests/e2e/features/fullduplex/test_personaplex_elastic.py

@@ -0,0 +1,240 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""GPU e2e for PersonaPlex elastic batching (bit-exact isolation + recycle).
+
+Greedy decoding makes every cross-slot leak observable as a bit difference:
+
+1. Isolation (bit-exact): recycling one slot mid-flight leaves every other
+   slot's output bit-identical to a run that never recycled.
+2. Replay exactness: a recycled slot's opening live ticks match a pristine boot
+   at numerical-noise level (a broken KV mask / staging roll garbles frame 0).
+3. Coherence (behavioral): a recycled slot fed a real spoken question produces
+   a real reply in its inner-monologue text stream.
+
+Run (moshi-free native stack; ~30 GB GPU + HF access to the gated repo; set
+PPLEX_QUESTION_WAV to a spoken-question wav to enable the coherence test):
+    pytest tests/e2e/features/fullduplex/test_personaplex_elastic.py -v -s
+
+Bench (B sweep, ms/tick vs the 80 ms budget):
+    python tests/e2e/features/fullduplex/test_personaplex_elastic.py --bench 2 4 8 16
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+torch = pytest.importorskip("torch")
+
+# slow + H100 + omni routes this module into the weekly Buildkite E2E group
+# ("Omni · H100" in .buildkite/cuda/test-weekly.yml: -m "slow and H100 and
+# omni" --run-level full_model); the ready/merge CPU sweeps exclude it by mark.
+pytestmark = [pytest.mark.full_model, pytest.mark.omni, pytest.mark.slow, pytest.mark.H100]
+
+if not torch.cuda.is_available():  # noqa: E402
+    pytest.skip("needs CUDA", allow_module_level=True)
+
+from vllm_omni.experimental.fullduplex.personaplex.config import PersonaPlexConfig  # noqa: E402
+from vllm_omni.experimental.fullduplex.personaplex.runtime import (  # noqa: E402
+    PersonaPlexEngine,
+)
+
+B = 4
+LIVE_TICKS = 40  # live frames compared per claim
+RECYCLE_AT = 10
+
+
+def _make_engine() -> PersonaPlexEngine:
+    cfg = PersonaPlexConfig(batch_size=B)  # native stepper is greedy-only
+    try:
+        return PersonaPlexEngine(cfg).load()
+    except Exception as exc:  # gated-repo access is environment-dependent
+        name = type(exc).__name__
+        if "Gated" in name or "401" in str(exc) or "403" in str(exc):
+            pytest.skip(f"no access to the gated nvidia/personaplex-7b-v1 repo: {name}")
+        raise
+
+
+@pytest.fixture(scope="module")
+def engine() -> PersonaPlexEngine:
+    return _make_engine()
+
+
+def _slot_stream(slot: int, tick: int, frame: int) -> np.ndarray:
+    """Deterministic per-slot input: distinct low-amplitude tones."""
+    sr = 24000
+    freq = 180.0 + 90.0 * slot
+    t0 = tick * frame
+    t = (np.arange(frame) + t0) / sr
+    return (0.12 * np.sin(2 * np.pi * freq * t)).astype(np.float32)
+
+
+def _run(engine, total_ticks: int, recycle: bool):
+    """Run the lockstep loop; optionally recycle slot 1 at RECYCLE_AT.
+
+    Returns per-slot lists of (audio | None) per live tick index, where the
+    recycled slot's list restarts its input stream at its post-prefill tick.
+    """
+    frame = engine.frame_size
+    outs: list[list[np.ndarray | None]] = [[] for _ in range(B)]
+    stream_tick = [0] * B  # per-slot conversation clock driving its input
+    prefill = None
+    tick = 0
+    while tick < total_ticks:
+        if recycle and tick == RECYCLE_AT and prefill is None:
+            engine.recycle_slot(1)
+            prefill = list(engine.prefill_steps(None))
+        pcm = np.zeros((B, frame), dtype=np.float32)
+        step_for_1 = None
+        if prefill:
+            while prefill and prefill[0].kind == "voice_end":
+                engine.voice_end(1)
+                prefill.pop(0)
+            if prefill:
+                step_for_1 = prefill.pop(0)
+            else:
+                engine.end_prefill(1)
+                prefill = None
+                stream_tick[1] = 0  # restart slot 1's input stream from its own t=0
+        for b in range(B):
+            if b == 1 and step_for_1 is not None:
+                continue  # engine overrides the user channel for prefill rows
+            pcm[b] = _slot_stream(b, stream_tick[b], frame)
+            stream_tick[b] += 1
+        frame_outs = engine.step_batch(pcm, {1: step_for_1} if step_for_1 is not None else None)
+        for b in range(B):
+            if b == 1 and step_for_1 is not None:
+                continue  # prefill tick: nothing conversational to record
+            frame_out = frame_outs[b]
+            outs[b].append(None if frame_out is None else frame_out.audio)
+        tick += 1
+    return outs
+
+
+def test_recycle_isolation_and_fresh_equivalence(engine):
+    engine.open_batch()
+    prefill_len = sum(1 for s in engine.prefill_steps(None) if s.kind != "voice_end")
+    total = RECYCLE_AT + prefill_len + LIVE_TICKS
+
+    base = _run(engine, total, recycle=False)
+
+    engine.open_batch()  # deterministic re-boot (greedy, fixed voice/persona)
+    rec = _run(engine, total, recycle=True)
+
+    # Claim 1 -- isolation: untouched slots bit-identical to the no-recycle run.
+    for b in (0, 2, 3):
+        assert len(base[b]) == total and len(rec[b]) == total
+        for i, (x, y) in enumerate(zip(base[b], rec[b])):
+            assert (x is None) == (y is None), f"slot {b} tick {i} validity diverged"
+            if x is not None:
+                assert np.array_equal(x, y), f"slot {b} tick {i} audio diverged (recycle leaked)"
+
+    # Claim 2 -- replay exactness: the recycled slot's opening live ticks match a
+    # pristine boot at numerical-noise level. A broken KV mask / staging-ring
+    # roll / Mimi reset garbles the audio from the very first frame (diffs of
+    # 0.1-1.0); a correct replay differs only by decoder-state rounding (~1e-4).
+    #
+    # Full-length bitwise identity is NOT expected, for two structural reasons:
+    # the replayed system prompt sits at a different ABSOLUTE RoPE offset (bf16
+    # rounds differently there), and the recycled Mimi ENCODER cannot restore a
+    # fresh stream's start padding (a batch-shared scalar), so from the first
+    # non-silent user frame it emits different-but-equally-valid codes for the
+    # same audio. Once any near-tie flips a token, the two equally valid
+    # trajectories legitimately diverge; coherence of the recycled slot is
+    # asserted behaviorally in test_recycled_slot_converses.
+    fresh = [a for a in base[1] if a is not None][:LIVE_TICKS]
+    reborn = [a for a in rec[1][RECYCLE_AT:] if a is not None][:LIVE_TICKS]
+    assert len(reborn) >= LIVE_TICKS // 2, "recycled slot produced too few live frames"
+    diffs = [float(np.max(np.abs(x - y))) for x, y in zip(fresh, reborn)]
+    first_flip = next((i for i, d in enumerate(diffs) if d > 5e-2), len(diffs))
+    print(
+        f"[elastic] replay exactness: {first_flip}/{len(diffs)} opening ticks at noise level "
+        f"(head diffs: {[f'{d:.1e}' for d in diffs[:6]]})"
+    )
+    assert first_flip >= 4, (
+        f"recycled slot diverged from fresh behavior at tick {first_flip} "
+        f"(diffs {diffs[: first_flip + 2]}); the system-prompt replay is broken"
+    )
+    assert max(diffs[:4]) < 2e-2, f"reborn audio not at noise level: {max(diffs[:4])}"
+
+
+def test_recycled_slot_converses(engine):
+    """Behavioral guarantee: a recycled slot holds a coherent conversation.
+
+    Feed a real spoken question into a slot that was recycled mid-flight (while
+    other slots keep running); the model's own inner-monologue text stream must
+    produce a real reply. This is the end-user property the encoder-state
+    caveats above cannot break.
+    """
+    import os
+
+    import sphn
+
+    wav_path = os.environ.get("PPLEX_QUESTION_WAV")
+    if not wav_path:
+        pytest.skip("set PPLEX_QUESTION_WAV to a spoken-question wav to run")
+    wav, _ = sphn.read(wav_path, sample_rate=24000)
+    wav = wav[0].astype(np.float32)
+    frame = engine.frame_size
+    pad = (-len(wav)) % frame
+    wav = np.concatenate([wav, np.zeros(pad, dtype=np.float32)])
+    q_frames = [wav[i : i + frame] for i in range(0, len(wav), frame)]
+
+    engine.open_batch()
+    pcm = np.zeros((B, frame), dtype=np.float32)
+    for _ in range(5):  # everyone live briefly
+        engine.step_batch(pcm)
+    engine.recycle_slot(1)
+    prefill = list(engine.prefill_steps(None))
+    while prefill:
+        while prefill and prefill[0].kind == "voice_end":
+            engine.voice_end(1)
+            prefill.pop(0)
+        if not prefill:
+            break
+        engine.step_batch(pcm, {1: prefill.pop(0)})
+    engine.end_prefill(1)
+
+    text = ""
+    total = len(q_frames) + 120  # question + ~10 s to answer
+    for i in range(total):
+        pcm = np.zeros((B, frame), dtype=np.float32)
+        if i < len(q_frames):
+            pcm[1] = q_frames[i]
+        outs = engine.step_batch(pcm)
+        if outs[1] is not None and outs[1].text:
+            text += outs[1].text
+    alpha = sum(c.isalpha() for c in text)
+    print(f"[elastic] recycled slot inner monologue: {text!r}")
+    assert alpha >= 15, f"recycled slot produced no coherent reply: {text!r}"
+
+
+def _bench(batch_sizes: list[int]) -> None:
+    import time
+
+    for bs in batch_sizes:
+        cfg = PersonaPlexConfig(batch_size=max(bs, 2))
+        eng = PersonaPlexEngine(cfg).load()
+        eng.open_batch()
+        frame = eng.frame_size
+        pcm = np.zeros((cfg.batch_size, frame), dtype=np.float32)
+        for _ in range(10):  # warm
+            eng.step_batch(pcm)
+        torch.cuda.synchronize()
+        t0 = time.monotonic()
+        n = 100
+        for _ in range(n):
+            eng.step_batch(pcm)
+        torch.cuda.synchronize()
+        ms = (time.monotonic() - t0) / n * 1000
+        mem = torch.cuda.max_memory_allocated() / 2**30
+        print(f"B={cfg.batch_size:3d}  {ms:6.1f} ms/tick  RTF/stream {ms / 80:.3f}  {mem:.1f} GiB")
+        del eng
+        torch.cuda.empty_cache()
+
+
+if __name__ == "__main__":
+    import sys
+
+    sizes = [int(x) for x in sys.argv[2:]] if len(sys.argv) > 2 and sys.argv[1] == "--bench" else [2, 4, 8]
+    _bench(sizes)

tests/e2e/features/fullduplex/test_personaplex_server_lease.py

@@ -0,0 +1,101 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Single-session lease teardown: a cancelled engine call must be drained.
+
+Cancelling an asyncio task blocked on a worker-thread engine call returns
+immediately while the thread keeps running (executor futures cannot be
+interrupted). The server therefore tracks the in-flight future on the session
+state and waits it out before releasing the capacity-one lease; otherwise a
+fast reconnect could call ``open()`` on the shared engine while the previous
+connection's final ``step()`` is still mutating it.
+"""
+
+import asyncio
+import contextlib
+import sys
+import threading
+import types
+
+import pytest
+
+try:
+    import sphn  # noqa: F401
+except ImportError:
+    # These lease tests never encode/decode Opus. Keep the optional serving
+    # dependency from hiding dependency-free teardown regressions.
+    sys.modules["sphn"] = types.ModuleType("sphn")
+
+from vllm_omni.experimental.fullduplex.personaplex.config import PersonaPlexConfig
+from vllm_omni.experimental.fullduplex.personaplex.serving.server import DuplexServer
+from vllm_omni.experimental.fullduplex.personaplex.session import PersonaPlexServingSessionState
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+@pytest.mark.asyncio
+async def test_cancelled_engine_call_is_drained_before_lease_release():
+    server = DuplexServer(PersonaPlexConfig())  # engine never loaded: stub calls only
+    state = PersonaPlexServingSessionState()
+
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def slow_engine_op():
+        started.set()
+        release.wait(timeout=5.0)
+        finished.set()
+        return []
+
+    task = asyncio.create_task(server._engine_call(state, slow_engine_op))
+    await asyncio.to_thread(started.wait, 5.0)
+
+    # Cancel while the worker thread is mid-call: the await returns at once,
+    # the thread keeps running, and its completion event stays tracked on the
+    # state (the asyncio future itself reads as cancelled/done here, which is
+    # exactly why the drain must not rely on it).
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+    assert state.inflight is not None
+    assert not state.inflight.is_set()
+
+    # The drain must genuinely wait for the worker, not just return.
+    drain = asyncio.create_task(server._drain_inflight(state))
+    await asyncio.sleep(0.05)
+    assert not drain.done()
+
+    release.set()
+    await asyncio.wait_for(drain, timeout=5.0)
+    assert finished.is_set()
+    assert state.inflight.is_set()
+
+
+@pytest.mark.asyncio
+async def test_completed_engine_call_clears_inflight():
+    server = DuplexServer(PersonaPlexConfig())
+    state = PersonaPlexServingSessionState()
+    result = await server._engine_call(state, lambda: "ok")
+    assert result == "ok"
+    assert state.inflight is None
+    await server._drain_inflight(state)  # no-op on a clean state
+
+
+@pytest.mark.asyncio
+async def test_drain_timeout_keeps_lease_until_worker_finishes():
+    server = DuplexServer(PersonaPlexConfig())
+    server._active = True
+    state = PersonaPlexServingSessionState()
+    state.inflight = threading.Event()
+
+    drained = await server._release_lease_after_drain(state, timeout=0.01)
+
+    assert drained is False
+    assert server._active is True
+
+    state.inflight.set()
+    for _ in range(100):
+        if not server._active:
+            break
+        await asyncio.sleep(0.01)
+    assert server._active is False

tests/e2e/features/fullduplex/test_personaplex_session.py

@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Lockstep-contract tests for the PersonaPlex session driver (no GPU / no moshi).
+
+A stub :class:`FrameStepper` stands in for the real Moshi backend so we can assert
+the framing + 1-frame-in / 1-frame-out invariants without weights.
+"""
+
+import numpy as np
+import pytest
+
+from vllm_omni.experimental.fullduplex.personaplex.config import FRAME_SIZE, PersonaPlexConfig
+from vllm_omni.experimental.fullduplex.personaplex.engine import FrameOutput, FrameStepper
+from vllm_omni.experimental.fullduplex.personaplex.session import PersonaPlexSession
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class StubStepper:
+    """Deterministic FrameStepper: frame i -> audio filled with i, text 't{i}' on odds."""
+
+    sample_rate = 24000
+    frame_size = FRAME_SIZE
+
+    def __init__(self) -> None:
+        self.opened: tuple[str | None, str | None] | None = None
+        self.steps = 0
+
+    def open_session(self, voice_prompt=None, persona=None) -> None:
+        self.opened = (voice_prompt, persona)
+        self.steps = 0
+
+    def step(self, user_pcm):
+        assert user_pcm.shape[0] == self.frame_size
+        self.steps += 1
+        i = self.steps
+        return FrameOutput(
+            audio=np.full(self.frame_size, float(i), dtype=np.float32),
+            text=(f"t{i}" if i % 2 else None),
+        )
+
+
+def test_stub_satisfies_protocol():
+    assert isinstance(StubStepper(), FrameStepper)
+
+
+def test_feed_emits_one_output_per_complete_frame():
+    s = PersonaPlexSession(StubStepper(), PersonaPlexConfig())
+    s.open(voice_prompt="NATF2.pt", persona="be terse")
+    out = s.feed(np.zeros(FRAME_SIZE * 3, dtype=np.float32))
+    assert len(out) == 3
+    # lockstep order preserved: frame i -> audio filled with i
+    assert [int(o.audio[0]) for o in out] == [1, 2, 3]
+    assert [o.text for o in out] == ["t1", None, "t3"]
+
+
+def test_partial_frames_are_buffered_until_complete():
+    stub = StubStepper()
+    s = PersonaPlexSession(stub, PersonaPlexConfig())
+    s.open()
+    assert s.feed(np.zeros(FRAME_SIZE // 2, dtype=np.float32)) == []  # nothing yet
+    assert stub.steps == 0
+    out = s.feed(np.zeros(FRAME_SIZE, dtype=np.float32))  # 1.5 frames total -> 1 frame
+    assert len(out) == 1
+    assert stub.steps == 1
+
+
+def test_flush_pads_the_tail():
+    stub = StubStepper()
+    s = PersonaPlexSession(stub, PersonaPlexConfig())
+    s.open()
+    s.feed(np.ones(FRAME_SIZE + 10, dtype=np.float32))  # 1 frame + 10-sample tail
+    assert stub.steps == 1
+    tail = s.flush()
+    assert len(tail) == 1 and stub.steps == 2
+    assert s.flush() == []  # nothing left
+
+
+def test_run_concatenates_audio_and_joins_text():
+    s = PersonaPlexSession(StubStepper(), PersonaPlexConfig())
+    s.open()
+    audio, text = s.run(np.zeros(FRAME_SIZE * 4, dtype=np.float32))
+    assert audio.shape[0] == FRAME_SIZE * 4
+    assert text == "t1t3"  # frames 1..4, text on odds
+
+
+def test_open_forwards_voice_and_persona():
+    stub = StubStepper()
+    s = PersonaPlexSession(stub, PersonaPlexConfig())
+    s.open(voice_prompt="NATM1.pt", persona="pirate")
+    assert stub.opened == ("NATM1.pt", "pirate")
+
+
+def test_feed_before_open_raises():
+    s = PersonaPlexSession(StubStepper(), PersonaPlexConfig())
+    with pytest.raises(RuntimeError):
+        s.feed(np.zeros(FRAME_SIZE, dtype=np.float32))