[Core] Graduate MiniCPM-o 4.5 and PersonaPlex full-duplex serving out of experimental (#6196)

Signed-off-by: chickeyton <ngton2014@gmail.com>
Signed-off-by: Gao Han <hgaoaf@connect.ust.hk>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Gao Han <hgaoaf@connect.ust.hk>

examples/offline_inference/personaplex/personaplex_offline.py

@@ -1,77 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Offline PersonaPlex inference through the vllm-omni fullduplex backend.
-
-This is the runnable end-to-end proof for the integration: it drives a user WAV
-through :class:`PersonaPlexEngine` + :class:`PersonaPlexSession` (our wrappers
-around the Moshi/Mimi lockstep loop) and writes the agent's audio + inner-monologue
-text. It mirrors ``python -m moshi.offline`` but goes through our ``FrameStepper``
-seam, so a green run validates that the session/adapter plumbing is faithful.
-
-Prereqs: HF access to ``nvidia/personaplex-7b-v1`` (``export HF_TOKEN=...``). The
-stack is moshi-free (native temporal + depformer + streaming Mimi via
-``transformers``); decoding is greedy.
-
-Example:
-    python examples/offline_inference/personaplex/personaplex_offline.py \\
-        --input-wav input_assistant.wav --output-wav out.wav \\
-        --voice-prompt NATF2.pt --persona "You are a wise and friendly teacher."
-"""
-
-from __future__ import annotations
-
-import argparse
-import json
-
-import numpy as np
-
-from vllm_omni.experimental.fullduplex.personaplex import (
-    PersonaPlexConfig,
-    PersonaPlexEngine,
-    PersonaPlexSession,
-)
-from vllm_omni.experimental.fullduplex.personaplex.config import DEFAULT_PERSONA
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--input-wav", required=True, help="User-side input WAV")
-    parser.add_argument("--output-wav", required=True, help="Agent audio output WAV")
-    parser.add_argument("--output-text", default=None, help="Optional JSON for agent text")
-    parser.add_argument("--voice-prompt", default="NATF2.pt", help="Voice clone (bundled .pt or .wav)")
-    parser.add_argument("--persona", default=DEFAULT_PERSONA, help="System role text")
-    parser.add_argument("--hf-repo", default="nvidia/personaplex-7b-v1")
-    parser.add_argument("--device", default="cuda")
-    parser.add_argument("--cpu-offload", action="store_true")
-    args = parser.parse_args()
-
-    import sphn  # local import: only the runnable driver needs the audio I/O dep
-
-    config = PersonaPlexConfig(
-        hf_repo=args.hf_repo,
-        voice_prompt=args.voice_prompt,
-        persona=args.persona,
-        device=args.device,
-        cpu_offload=args.cpu_offload,
-    )
-
-    engine = PersonaPlexEngine(config).load()
-    session = PersonaPlexSession(engine, config)
-    session.open()
-
-    # Load exactly like the reference (moshi.lm.load_audio): native read then
-    # explicit resample, so greedy runs stay bit-faithful to `python -m moshi.offline`.
-    user_pcm, src_sr = sphn.read(args.input_wav)
-    user_pcm = sphn.resample(user_pcm, src_sample_rate=src_sr, dst_sample_rate=config.sample_rate)
-    audio, text = session.run(np.asarray(user_pcm[0], dtype=np.float32))
-
-    sphn.write_wav(args.output_wav, audio, config.sample_rate)
-    print(f"wrote {audio.shape[0] / config.sample_rate:.2f}s of agent audio to {args.output_wav}")
-    print(f"agent text: {text!r}")
-    if args.output_text:
-        with open(args.output_text, "w") as f:
-            json.dump({"text": text}, f, ensure_ascii=False)
-
-
-if __name__ == "__main__":
-    main()

examples/online_serving/barge_in_client.py

@@ -0,0 +1,272 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Full-duplex use case: talk over the assistant (barge-in).
+
+Scenario (see ``barge_in_client_flow.md`` for the diagram)
+---------------------------------------------------------------
+1. The user asks a question (``--question-wav``); the model answers in speech.
+2. Mid-answer, the user starts talking again (``--interrupt-wav``) — a
+   follow-up or correction — while assistant audio is still streaming.
+3. The model handles the overlap: depending on the session overlap policy and
+   speech length, the in-flight response is barged in (cancelled) or the
+   commit is deferred to the turn end, and a new response answers the
+   interruption. Either way the user never waits for the first answer.
+
+The demo is model-neutral: it drives any duplex model through the public
+:class:`vllm_omni.clients.duplex.DuplexClient` over
+``/v1/realtime?duplex=1``. Model-specific session shape comes from the
+per-model presets under ``vllm_omni.clients.<model>`` selected with
+``--preset`` (default ``minicpmo_4_5``, whose model decides listen/speak on
+its own — no VAD).
+
+Run the server first (for MiniCPM-o 4.5: ``bash minicpmo/barge_in_serve.sh``, which
+starts the ``vllm_omni/deploy/minicpmo_4_5.yaml`` profile with ``session_mode: duplex``),
+then:
+
+    python barge_in_client.py \
+        --url ws://127.0.0.1:8099/v1/realtime \
+        --model openbmb/MiniCPM-o-4_5 \
+        --ref-audio /path/to/reference_voice.wav \
+        --question-wav question_16k.wav \
+        --interrupt-wav follow_up_16k.wav \
+        --output-dir ./duplex_out
+
+Input WAVs must be mono 16 kHz PCM16; when a preset's session takes a
+different capture format (the PersonaPlex preset streams 24 kHz float32),
+the audio is converted before streaming. ``--ref-audio`` is required by the
+MiniCPM-o preset (without it the session is rejected with
+``ref_audio_required``). Outputs land in ``--output-dir`` as one WAV per
+response — ``response_1_cancelled.wav`` for a barged-in answer,
+``response_1_completed.wav`` for one that finished — plus ``summary.json``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from vllm_omni.clients.duplex import (  # noqa: E402
+    AudioFormat,
+    DuplexClient,
+    EventCollector,
+    SessionConfig,
+    acknowledge_collected_playback,
+    audio_data_url,
+    read_pcm16_wav,
+    wait_for_condition,
+    write_pcm16_wav,
+)
+
+_TRANSCRIPT_DELTA_EVENTS = {
+    "response.audio_transcript.delta",
+    "response.output_text.delta",
+    "response.text.delta",
+}
+
+# Input WAVs are mono 16 kHz PCM16; each preset's session may negotiate a
+# different capture format (PersonaPlex takes 24 kHz float32) and a different
+# model-unit-aligned chunk size (PersonaPlex frames are exactly 80 ms).
+_PRESET_DEFAULT_CHUNK_MS = {"minicpmo_4_5": 100, "personaplex": 80, "none": 100}
+
+
+def _convert_to_session_format(pcm16_16k: bytes, target: AudioFormat) -> bytes:
+    """Convert mono 16 kHz PCM16 to the session's input format."""
+    if target.encoding == "pcm16" and target.sample_rate_hz == 16_000:
+        return pcm16_16k
+    import numpy as np
+
+    samples = np.frombuffer(pcm16_16k, dtype=np.int16).astype(np.float32) / 32768.0
+    if target.sample_rate_hz != 16_000 and len(samples) > 1:
+        target_len = round(len(samples) * target.sample_rate_hz / 16_000)
+        positions = np.linspace(0.0, len(samples) - 1, num=target_len)
+        samples = np.interp(positions, np.arange(len(samples)), samples).astype(np.float32)
+    if target.encoding == "pcm_f32le":
+        return samples.tobytes()
+    if target.encoding == "pcm16":
+        return (np.clip(samples, -1.0, 1.0) * 32767.0).astype(np.int16).tobytes()
+    raise ValueError(f"Unsupported session input encoding: {target.encoding!r}")
+
+
+def _session_config(preset: str, *, ref_audio_path: str | None) -> SessionConfig:
+    ref_audio = audio_data_url(Path(ref_audio_path).expanduser()) if ref_audio_path else None
+    if preset == "minicpmo_4_5":
+        from vllm_omni.clients.minicpmo_4_5 import create_duplex_session_config
+
+        return create_duplex_session_config(ref_audio=ref_audio)
+    if preset == "personaplex":
+        from vllm_omni.clients.personaplex import create_duplex_session_config
+
+        return create_duplex_session_config()
+    return SessionConfig(ref_audio=ref_audio)
+
+
+def _fold_responses(collector: EventCollector) -> dict[str, dict[str, object]]:
+    """Fold the event stream into per-response audio, text, and status."""
+    out: dict[str, dict[str, object]] = {}
+    for response_id in collector.response_ids:
+        out[response_id] = {
+            "audio": collector.audio_bytes(response_id),
+            "text": "",
+            "status": "in_progress",
+            "sample_rate_hz": collector.output_sample_rate_hz,
+        }
+
+    for event in collector.events:
+        etype = event.get("type")
+        response_id = collector.response_id(event)
+        item = out.get(response_id) if isinstance(response_id, str) else None
+        if etype in _TRANSCRIPT_DELTA_EVENTS and item is not None:
+            delta = event.get("delta")
+            if isinstance(delta, str):
+                item["text"] += delta
+        elif etype == "response.done" and item is not None:
+            response = event.get("response")
+            status = response.get("status") if isinstance(response, dict) else None
+            if item["status"] == "in_progress":
+                item["status"] = "cancelled" if status == "cancelled" else "completed"
+        elif etype == "audio.cancelled":
+            targets = [item] if item is not None else list(out.values())
+            for target in targets:
+                if target["status"] == "in_progress":
+                    target["status"] = "cancelled"
+    return out
+
+
+async def run(args: argparse.Namespace) -> int:
+    config = _session_config(args.preset, ref_audio_path=args.ref_audio)
+    question = _convert_to_session_format(read_pcm16_wav(Path(args.question_wav)), config.input_audio)
+    interrupt = _convert_to_session_format(read_pcm16_wav(Path(args.interrupt_wav)), config.input_audio)
+    chunk_ms = args.chunk_ms if args.chunk_ms is not None else _PRESET_DEFAULT_CHUNK_MS.get(args.preset, 100)
+    output_dir = Path(args.output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    collector = EventCollector()
+    client = DuplexClient(
+        args.url,
+        model=args.model,
+        config=config,
+        reconnect=None,
+        heartbeat_interval_s=None,
+        handshake_timeout_s=args.timeout_s,
+    )
+    # GREET: open the voice session (handshake happens on enter).
+    async with client:
+        consume_task = asyncio.create_task(collector.consume(client))
+
+        # ASK: stream the question at realtime pace, then commit the turn.
+        # The model decides on its own to answer (no VAD).
+        print("[1] streaming question ...", file=sys.stderr)
+        await client.stream_pcm(question, chunk_ms=chunk_ms)
+        await client.commit(create_response=False)
+        await wait_for_condition(
+            lambda: collector.count("response.created") > 0,
+            timeout_s=args.timeout_s,
+            label="first response.created",
+        )
+        print("[1] assistant is answering", file=sys.stderr)
+
+        # Let the answer play for a moment, as a listening user would.
+        await asyncio.sleep(args.listen_s)
+
+        # INTERRUPT: talk over the assistant while its audio still streams.
+        # The serving overlap policy decides: enough overlapped speech barges
+        # in (the first response is cancelled) or a short remark is deferred
+        # until the model yields the turn.
+        print("[2] interrupting mid-answer ...", file=sys.stderr)
+        ids_before_interrupt = set(collector.response_ids)
+        await client.stream_pcm(interrupt, chunk_ms=chunk_ms)
+        await client.commit(create_response=False)
+
+        # ANSWER AGAIN: wait for a response created *after* the interruption
+        # to finish. The first answer may still be streaming here; matching on
+        # any response.done would let its completion satisfy the wait and cut
+        # off the actual follow-up answer at close.
+        def follow_up_answer_done() -> bool:
+            return any(
+                event.get("type") == "response.done"
+                and collector.response_id(event) is not None
+                and collector.response_id(event) not in ids_before_interrupt
+                for event in collector.events
+            )
+
+        await wait_for_condition(
+            follow_up_answer_done,
+            timeout_s=args.timeout_s,
+            label="response.done for the response answering the interruption",
+        )
+
+        # HANG UP: report playback, close the session cleanly.
+        await acknowledge_collected_playback(client, collector)
+        await client.close(timeout_s=args.timeout_s)
+        try:
+            await asyncio.wait_for(consume_task, timeout=5.0)
+        except (asyncio.TimeoutError, asyncio.CancelledError):
+            consume_task.cancel()
+
+    # Report: one WAV per response; statuses show what the overlap did.
+    responses = _fold_responses(collector)
+    summary: list[dict[str, object]] = []
+    for index, (response_id, item) in enumerate(responses.items(), start=1):
+        audio = item["audio"]
+        assert isinstance(audio, bytes)
+        sample_rate_hz = int(item["sample_rate_hz"])  # type: ignore[arg-type]
+        wav_path = output_dir / f"response_{index}_{item['status']}.wav"
+        if audio:
+            write_pcm16_wav(wav_path, audio, sample_rate_hz=sample_rate_hz)
+        summary.append(
+            {
+                "response_id": response_id,
+                "status": item["status"],
+                "audio_s": round(len(audio) / (sample_rate_hz * 2), 2),
+                "text": item["text"],
+                "wav": wav_path.name if audio else None,
+            }
+        )
+    (output_dir / "summary.json").write_text(json.dumps(summary, indent=2, ensure_ascii=False))
+    (output_dir / "events.jsonl").write_text(
+        "".join(json.dumps(event, ensure_ascii=False) + "\n" for event in collector.events),
+        encoding="utf-8",
+    )
+    print(json.dumps(summary, indent=2, ensure_ascii=False))
+    cancelled = sum(1 for item in summary if item["status"] == "cancelled")
+    print(
+        f"\n{len(summary)} responses ({cancelled} interrupted); outputs in {output_dir}",
+        file=sys.stderr,
+    )
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--url", default="ws://127.0.0.1:8099/v1/realtime")
+    parser.add_argument("--model", default="openbmb/MiniCPM-o-4_5")
+    parser.add_argument(
+        "--preset",
+        choices=["minicpmo_4_5", "personaplex", "none"],
+        default="minicpmo_4_5",
+        help="per-model session preset from vllm_omni.clients.<model>",
+    )
+    parser.add_argument("--ref-audio", help="reference voice WAV (the MiniCPM-o preset requires it)")
+    parser.add_argument("--question-wav", required=True, help="mono 16 kHz PCM16 question")
+    parser.add_argument("--interrupt-wav", required=True, help="mono 16 kHz PCM16 follow-up spoken mid-answer")
+    parser.add_argument("--output-dir", default="./duplex_out")
+    parser.add_argument(
+        "--chunk-ms",
+        type=int,
+        default=None,
+        help="append chunk size; defaults to the preset's model-unit-aligned size (minicpmo_4_5: 100, personaplex: 80)",
+    )
+    parser.add_argument("--listen-s", type=float, default=2.0, help="how long to listen before interrupting")
+    parser.add_argument("--timeout-s", type=float, default=120.0)
+    return asyncio.run(run(parser.parse_args()))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/online_serving/barge_in_client_flow.md

@@ -0,0 +1,54 @@
+# `barge_in_client.py` — what happens, at a glance
+
+The demo plays one everyday conversation pattern against the full-duplex
+server: ask something, get talked-over mid-answer, and see the assistant
+cope. High-level flow below — for the exact wire events, read the
+docstring in `barge_in_client.py`.
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant U as user (client script)
+    participant A as assistant (server + model)
+
+    Note over U,A: GREET
+    U->>A: start a voice session (with the reference voice)
+    A-->>U: ready
+
+    Note over U,A: ASK
+    U->>A: ask a question (speak the question WAV)
+    Note right of A: decides on its own to answer<br/>(no VAD — the model chooses)
+    A-->>U: starts answering aloud
+
+    Note over U,A: INTERRUPT
+    Note over U: listens for ~2 seconds
+    par both talk at once
+        U->>A: interrupts with a follow-up
+    and
+        A-->>U: still speaking the first answer
+    end
+
+    alt the interruption is substantial
+        Note right of A: stops mid-sentence (barge-in),<br/>first answer is cut off
+    else it was just a short remark
+        Note right of A: finishes the first answer,<br/>then takes the follow-up
+    end
+
+    Note over U,A: ANSWER AGAIN
+    A-->>U: answers the follow-up aloud
+
+    Note over U,A: HANG UP
+    U->>A: goodbye (close the session)
+    Note over U: saves one recording per answer<br/>plus a summary of what happened
+```
+
+Outcome: the output directory shows which branch ran — a cut-off first
+answer is saved as `response_1_cancelled.wav`, a completed one as
+`response_1_completed.wav`, and `summary.json` records status, text, and
+duration per answer.
+
+Companion: the runtime architecture lives in
+[`docs/design/fullduplex.md`](../../docs/design/fullduplex.md); the client
+API is `vllm_omni.clients.duplex.DuplexClient`, documented with the wire
+protocol in
+[`docs/serving/realtime_duplex_api.md`](../../docs/serving/realtime_duplex_api.md).

examples/online_serving/minicpmo/README.md

@@ -4,9 +4,11 @@ This directory contains MiniCPM-o 4.5 online serving demos for vLLM-Omni.
 Inputs can include text, image, audio, or video; outputs are text and optional
 24 kHz speech.
 
-For the experimental native duplex runtime architecture, lifecycle invariants,
+For the native duplex runtime architecture, lifecycle invariants,
 capability boundary, and validation scope, see
-[`vllm_omni/experimental/fullduplex/DESIGN.md`](../../../vllm_omni/experimental/fullduplex/DESIGN.md).
+[`docs/design/fullduplex.md`](../../../docs/design/fullduplex.md); for the
+`DuplexClient` API and the `/v1/realtime?duplex=1` wire protocol, see
+[`docs/serving/realtime_duplex_api.md`](../../../docs/serving/realtime_duplex_api.md).
 
 ## Installation
 
@@ -231,7 +233,7 @@ Start the duplex generation server:
 
 ```bash
 vllm serve openbmb/MiniCPM-o-4_5 --omni --trust-remote-code \
-    --deploy-config vllm_omni/deploy/minicpmo_4_5_duplex.yaml \
+    --deploy-config vllm_omni/deploy/minicpmo_4_5.yaml \
     --served-model-name openbmb/MiniCPM-o-4_5 \
     --host 0.0.0.0 --port 8099
 ```

examples/online_serving/minicpmo/barge_in_serve.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+# Server side for barge_in_client.py: start MiniCPM-o 4.5 in native
+# full-duplex mode.
+#
+# The duplex deploy profile registers the WS /v1/duplex endpoint (with
+# /v1/realtime?duplex=1 projecting onto the same handler), enables the
+# engine duplex control plane, and bounds the Thinker/Talker/Code2Wav
+# stages to four live duplex sessions on one GPU. See
+# vllm_omni/deploy/minicpmo_4_5.yaml (session_mode: duplex) for the session
+# limits; every other duplex_session knob keeps its runtime default
+# (idle TTL 300s, disconnect grace 30s, 16MiB pending input per session).
+set -euo pipefail
+
+MODEL="${MODEL:-openbmb/MiniCPM-o-4_5}"
+PORT="${PORT:-8099}"
+
+exec vllm-omni serve "$MODEL" \
+    --omni \
+    --deploy-config vllm_omni/deploy/minicpmo_4_5.yaml \
+    --trust-remote-code \
+    --host 0.0.0.0 --port "$PORT"
+# duplex endpoints: ws://<host>:$PORT/v1/duplex and /v1/realtime?duplex=1

examples/online_serving/minicpmo/realtime_duplex_demo.py

@@ -23,23 +23,26 @@ REPO_ROOT = Path(__file__).resolve().parents[3]
 if str(REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(REPO_ROOT))
 
-from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
+from vllm_omni.clients.duplex import (  # noqa: E402
     PCM16_BYTES_PER_SAMPLE,
     PCM16_SAMPLE_RATE,
-    RealtimeDuplexClient,
-    RealtimeEventCollector,
-    build_realtime_url,
+    DuplexClient,
+    EventCollector,
+    acknowledge_collected_playback,
     read_pcm16_wav,
-    wait_for,
+    wait_for_condition,
     write_pcm16_wav,
 )
-from vllm_omni.experimental.fullduplex.client import chunk_period_ms as _chunk_period_ms  # noqa: E402
-from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
+from vllm_omni.clients.duplex import chunk_period_ms as _chunk_period_ms  # noqa: E402
+from vllm_omni.clients.duplex import (  # noqa: E402
     has_residual_model_unit as _has_residual_model_unit,
 )
-from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
+from vllm_omni.clients.duplex import (  # noqa: E402
     reference_audio_data_url as _ref_audio_data_url,
 )
+from vllm_omni.clients.minicpmo_4_5 import (  # noqa: E402
+    create_duplex_session_config,
+)
 from vllm_omni.experimental.fullduplex.video_stacking import (  # noqa: E402
     concat_frames_b64,
     unit_subframe_offsets,
@@ -96,7 +99,7 @@ class _StreamingOutputWriter:
         )
 
 
-class _StreamingEventCollector(RealtimeEventCollector):
+class _StreamingEventCollector(EventCollector):
     def __init__(self, writer: _StreamingOutputWriter) -> None:
         super().__init__()
         self._writer = writer
@@ -155,7 +158,7 @@ def _response_in_progress(events: list[dict[str, object]]) -> bool:
     )
 
 
-async def _commit_input_after_playback_checkpoint(client: RealtimeDuplexClient) -> float:
+async def _commit_input_after_playback_checkpoint(client: DuplexClient, collector: EventCollector) -> float:
     """Commit input only after preserving already-played assistant history.
 
     Native duplex can finish several responses while the user is still
@@ -166,7 +169,7 @@ async def _commit_input_after_playback_checkpoint(client: RealtimeDuplexClient)
     original history position.  A second acknowledgement after the commit can
     then safely advance response-local playback that drained in the meantime.
     """
-    await client.acknowledge_playback()
+    await acknowledge_collected_playback(client, collector)
     commit_sent_at_s = time.monotonic()
     await client.commit()
     return commit_sent_at_s
@@ -434,87 +437,95 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
     if not input_pcm16:
         raise ValueError("input WAV has no audio")
 
-    url = build_realtime_url(
+    collector = _StreamingEventCollector(stream_writer)
+    client = DuplexClient(
         args.url,
-        args.model,
-        autostart=False if args.ref_audio else None,
-        session_id=args.session_id,
-    )
-    client = RealtimeDuplexClient(url)
-    client.events = _StreamingEventCollector(stream_writer)
-    async with client:
-        await client.configure(
-            args.model,
+        model=args.model,
+        config=create_duplex_session_config(
             ref_audio=_ref_audio_data_url(args.ref_audio),
-            session_id=args.session_id,
             temperature=args.temperature,
-            timeout_s=args.timeout_s,
-        )
-        stream_event_cursor = len(client.events.events)
-        frames_sent = await client.stream_pcm16(
+        ),
+        session_id=args.session_id,
+        reconnect=None,
+        heartbeat_interval_s=None,
+        handshake_timeout_s=args.timeout_s,
+    )
+    async with client:
+        consume_task = asyncio.create_task(collector.consume(client))
+        # The handshake consumed session.created before the collector
+        # subscribed; seed it so counts and capability lookups still work.
+        collector.add({"type": "session.created", "session": client.session_info})
+        stream_event_cursor = len(collector.events)
+        frames_sent = await client.stream_pcm(
             input_pcm16,
             chunk_ms=args.chunk_ms,
             realtime=not args.no_realtime_pacing,
             video_frames=video_frames,
             stacked_video_frames=stacked_video_frames,
         )
-        commit_event_cursor = len(client.events.events)
-        stream_decision = _latest_model_decision(client.events.events, stream_event_cursor)
+        commit_event_cursor = len(collector.events)
+        stream_decision = _latest_model_decision(collector.events, stream_event_cursor)
         input_has_residual_model_unit = _has_residual_model_unit(
             input_pcm16,
-            chunk_period_ms=_chunk_period_ms(client.events.events),
+            chunk_period_ms=_chunk_period_ms(collector.events),
         )
         wait_for_post_commit_decision = False
-        commit_sent_at_s = await _commit_input_after_playback_checkpoint(client)
+        commit_sent_at_s = await _commit_input_after_playback_checkpoint(client, collector)
         wait_error: str | None = None
         committed_index: int | None = None
         post_commit_decision: str | None = None
         try:
-            await wait_for(
-                lambda: _input_committed_index(client.events.events, commit_event_cursor) is not None,
+            await wait_for_condition(
+                lambda: _input_committed_index(collector.events, commit_event_cursor) is not None,
                 timeout_s=args.timeout_s,
                 label="input_audio_buffer.committed",
             )
-            committed_index = _input_committed_index(client.events.events, commit_event_cursor)
-            stream_decision = _latest_model_decision(client.events.events[: committed_index + 1], stream_event_cursor)
+            committed_index = _input_committed_index(collector.events, commit_event_cursor)
+            stream_decision = _latest_model_decision(collector.events[: committed_index + 1], stream_event_cursor)
             wait_for_post_commit_decision = input_has_residual_model_unit or _response_in_progress(
-                client.events.events[: committed_index + 1]
+                collector.events[: committed_index + 1]
             )
             if wait_for_post_commit_decision:
-                await wait_for(
-                    lambda: _post_commit_model_decision(client.events.events, committed_index) is not None,
+                await wait_for_condition(
+                    lambda: _post_commit_model_decision(collector.events, committed_index) is not None,
                     timeout_s=args.timeout_s,
                     label="post-commit model decision or response drain",
                 )
-                post_commit_decision = _post_commit_model_decision(client.events.events, committed_index)
+                post_commit_decision = _post_commit_model_decision(collector.events, committed_index)
         except TimeoutError as exc:
             wait_error = str(exc)
-        await client.acknowledge_playback()
+        await acknowledge_collected_playback(client, collector)
         close_error: str | None = None
+        await client.close(timeout_s=args.timeout_s)
+        # close() returns when the client's reader observed session.closed;
+        # the collector is a separate subscriber task, so wait for it to
+        # drain before checking, or a normal close reads as a failure.
         try:
-            await client.close_session(timeout_s=args.timeout_s)
-        except TimeoutError as exc:
-            close_error = str(exc)
+            await asyncio.wait_for(consume_task, timeout=5.0)
+        except (asyncio.TimeoutError, asyncio.CancelledError):
+            consume_task.cancel()
+        if collector.count("session.closed") == 0:
+            close_error = "Timed out waiting for session.closed"
 
-        audio = client.events.audio_bytes()
-        first_text_at_s = client.events.first_received_at(
+        audio = collector.audio_bytes()
+        first_text_at_s = collector.first_received_at(
             "response.audio_transcript.delta",
             "response.output_text.delta",
             after_s=commit_sent_at_s,
         )
-        first_audio_at_s = client.events.first_received_at(
+        first_audio_at_s = collector.first_received_at(
             "response.audio.delta",
             after_s=commit_sent_at_s,
         )
-        response_created_at_s = client.events.first_received_at(
+        response_created_at_s = collector.first_received_at(
             "response.created",
             after_s=commit_sent_at_s,
         )
-        response_done_at_s = client.events.first_received_at(
+        response_done_at_s = collector.first_received_at(
             "response.done",
             after_s=commit_sent_at_s,
         )
-        audio_duration_s = len(audio) / (client.events.output_sample_rate_hz * 2)
+        audio_duration_s = len(audio) / (collector.output_sample_rate_hz * 2)
         response_generation_s = (
             response_done_at_s - response_created_at_s
             if response_done_at_s is not None and response_created_at_s is not None
@@ -522,27 +533,27 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
         )
         transcript_deltas = [
             str(event.get("delta", ""))
-            for event in client.events.events
+            for event in collector.events
             if event.get("type")
             in {
                 "response.audio_transcript.delta",
                 "response.output_text.delta",
             }
         ]
-        response_id = client.events.response_ids[0] if client.events.response_ids else None
-        timing = client.events.timing_summary(
+        response_id = collector.response_ids[0] if collector.response_ids else None
+        timing = collector.timing_summary(
             after_s=commit_sent_at_s,
             input_committed_at_s=commit_sent_at_s,
             response_id=response_id,
         )
-        errors = client.events.errors()
+        errors = collector.errors()
         if wait_error:
             errors.append({"type": "client.timeout", "message": wait_error})
         if close_error:
             errors.append({"type": "client.timeout", "message": close_error})
         model_decision = post_commit_decision or stream_decision
         (output_dir / "events.jsonl").write_text(
-            "".join(json.dumps(event, ensure_ascii=False) + "\n" for event in client.events.events),
+            "".join(json.dumps(event, ensure_ascii=False) + "\n" for event in collector.events),
             encoding="utf-8",
         )
         (output_dir / "output.pcm").write_bytes(audio)
@@ -550,13 +561,13 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
             write_pcm16_wav(
                 output_dir / "output.wav",
                 audio,
-                sample_rate_hz=client.events.output_sample_rate_hz,
+                sample_rate_hz=collector.output_sample_rate_hz,
             )
 
         result = {
             "ok": (
-                client.events.count("session.created") > 0
-                and client.events.count("session.closed") > 0
+                collector.count("session.created") > 0
+                and collector.count("session.closed") > 0
                 and not errors
                 and model_decision is not None
                 and (bool(audio) or not args.require_audio)
@@ -568,12 +579,12 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
                 "decision_required": wait_for_post_commit_decision,
                 "input_had_residual_model_unit": input_has_residual_model_unit,
                 "response_listen_count": _event_count_after(
-                    client.events.events,
+                    collector.events,
                     "response.listen",
                     committed_index,
                 ),
                 "response_done_count": _event_count_after(
-                    client.events.events,
+                    collector.events,
                     "response.done",
                     committed_index,
                 ),
@@ -585,7 +596,7 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
             "audio_bytes": len(audio),
             "audio_chunk_count": len(stream_writer.audio_chunk_paths),
             "audio_chunk_files": [str(path) for path in stream_writer.audio_chunk_paths],
-            "output_sample_rate_hz": client.events.output_sample_rate_hz,
+            "output_sample_rate_hz": collector.output_sample_rate_hz,
             "latency": {
                 "ttft_ms": (
                     round((first_text_at_s - commit_sent_at_s) * 1000, 2) if first_text_at_s is not None else None
@@ -611,7 +622,7 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
                 "measurement_origin": "input_audio_buffer.commit send",
             },
             "timing": timing,
-            "response_ids": client.events.response_ids,
+            "response_ids": collector.response_ids,
             "transcript": "".join(transcript_deltas),
             "errors": errors,
             "output_dir": str(output_dir),

examples/online_serving/minicpmo/realtime_web/app/static/app.js

@@ -86,7 +86,7 @@
     url.protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
     url.searchParams.set('duplex', '1');
     url.searchParams.set('model', config.model || 'openbmb/MiniCPM-o-4_5');
-    url.searchParams.set('minicpmo45_native_duplex', '1');
+    url.searchParams.set('native_duplex', '1');
     url.searchParams.set('autostart', '0');
     return url.toString();
   }
@@ -458,7 +458,7 @@
         settled = true;
         const extraBody = {
           auto_response: true,
-          minicpmo45_native_duplex: true,
+          native_duplex: true,
         };
         const session = {
           modalities: ['audio', 'text'],

examples/online_serving/personaplex/README.md

@@ -1,47 +1,40 @@
 # PersonaPlex full-duplex online serving
 
 Serve [`nvidia/personaplex-7b-v1`](https://huggingface.co/nvidia/personaplex-7b-v1)
-(a Moshi-based full-duplex speech-to-speech model) with the native vLLM-Omni engine.
+(a Moshi-based full-duplex speech-to-speech model) with the native vLLM-Omni engine
+through the unified duplex serving stack (`/v1/duplex` and `/v1/realtime?duplex=1`).
 
-The server hosts the **official PersonaPlex web client** at `/` (auto-downloaded from
-the model repo) and implements its WebSocket protocol at `/api/chat`, so you talk to the
-model live in the browser — mic in, agent speech + inner-monologue text out.
-
-> Experimental. Requires a GPU and Hugging Face access to the gated repo
+> Requires a GPU and Hugging Face access to the gated repo
 > (`HF_TOKEN` with access to `nvidia/personaplex-7b-v1`).
 
 ## Start the server
 
-```bash
-HF_TOKEN=... CUDA_VISIBLE_DEVICES=0 bash run_server.sh
-# or directly:
-HF_TOKEN=... CUDA_VISIBLE_DEVICES=0 python -m \
-    vllm_omni.experimental.fullduplex.personaplex.serving.server --port 8124
-```
-
-## Talk to it
-
-- **Browser (recommended):** open `http://localhost:8124/` and allow the microphone.
-  Use headphones so the agent does not hear itself.
-- **Headless:** stream a 24 kHz mono WAV and save the reply:
-
-  ```bash
-  python duplex_client.py --url ws://localhost:8124/api/chat --input user.wav --out reply.wav
-  ```
-
-## Unified Realtime API
-
-The default `vllm_omni/deploy/personaplex.yaml` also enables main's engine-owned
-full-duplex control plane:
+The default `vllm_omni/deploy/personaplex.yaml` enables the engine-owned
+full-duplex control plane (`session_mode: duplex`):
 
 ```bash
-CUDA_VISIBLE_DEVICES=0 python -m vllm_omni.entrypoints.cli.main serve \
+HF_TOKEN=... CUDA_VISIBLE_DEVICES=0 python -m vllm_omni.entrypoints.cli.main serve \
   /path/to/personaplex-7b-v1 \
   --omni \
   --deploy-config vllm_omni/deploy/personaplex.yaml
 ```
 
-Validate the actual `/v1/realtime?duplex=1` scheduler path with paced 24 kHz
+This exposes:
+
+- `WS /v1/duplex` — the native duplex session dialect
+  (`session.create` / `input_audio_buffer.append` / `response.output_audio.delta` ...);
+- `WS /v1/realtime?duplex=1` — the same sessions projected onto the
+  OpenAI Realtime event vocabulary (client API and wire protocol:
+  [`docs/serving/realtime_duplex_api.md`](../../../docs/serving/realtime_duplex_api.md)).
+
+PersonaPlex is a pure-lockstep model: every session on a PersonaPlex deployment is
+native duplex (`is_enabled()` is unconditionally true), audio flows continuously in
+both directions, and there are no client commits or external turn signals
+(`supports_client_commit=false`, `supports_external_turn_signal=false`).
+
+## Validate the serving path
+
+Validate the `/v1/realtime?duplex=1` scheduler path with paced 24 kHz
 PCM, two concurrent sessions, overflow admission, per-session slot recycling,
 and non-silent output:
 
@@ -52,33 +45,19 @@ python tests/e2e/online_serving/personaplex_realtime_duplex.py \
   --output-dir /tmp/personaplex-realtime-duplex
 ```
 
-This is distinct from the standalone `/api/chat` compatibility server above.
 The unified endpoint advertises `supports_barge_in=false`: overlapping speech
 is native model behavior, but destructive output interruption and model-state
 rewind have not been validated for PersonaPlex.
 
-## Protocol (`/api/chat`)
-
-Binary WebSocket messages, first byte is a tag (identical to `moshi.server`):
-
-| Direction | Tag | Payload |
-| --- | --- | --- |
-| server → client | `\x00` | ready handshake (after the system-prompt prefill) |
-| client → server | `\x01` | Opus-encoded mic audio (24 kHz) |
-| server → client | `\x01` | Opus-encoded agent audio |
-| server → client | `\x02` | UTF-8 inner-monologue text |
-
-Connect with query params `text_prompt` (persona) and `voice_prompt` (voice file).
-A raw-PCM endpoint (`/v1/audio/duplex`, JSON `open` + float32 frames) is also available
-for clients that do not want an Opus dependency.
-
 ## Notes
 
-- **Run the client near the server.** Real-time 80 ms audio is sensitive to network
-  latency/jitter; over a high-latency remote link playback can stutter regardless of
-  engine speed. On localhost it is smooth.
-- Greedy decoding by default. One conversation per server by default (a second
-  concurrent connection is rejected with WS close 1013 until the first ends);
-  pass `--batch-size N` to host N concurrent conversations on one engine
-  (elastic batching; connections beyond N are rejected until a slot frees).
+- **Run the client near the server.** Real-time 80 ms frame audio is sensitive to
+  network latency/jitter; over a high-latency remote link playback can stutter
+  regardless of engine speed. On localhost it is smooth.
+- The earlier standalone Moshi-web compatibility server (browser client at `/`,
+  binary WS protocol at `/api/chat`, raw-PCM `/v1/audio/duplex`) was demo-only
+  and has been removed; use the unified endpoints above.
+- Session config (voice / persona / sampling) is passed per session via
+  `extra_body`; see
+  `vllm_omni/model_executor/models/personaplex/duplex/serving_adapter.py`.
 - Full runbook: [`recipes/NVIDIA/PersonaPlex.md`](../../../recipes/NVIDIA/PersonaPlex.md).

examples/online_serving/personaplex/duplex_client.py

@@ -1,114 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Headless client for the PersonaPlex duplex server (official Moshi protocol).
-
-Streams a 24 kHz mono WAV to ``/api/chat`` exactly as the browser client does
-(Opus over WebSocket, tags ``\\x00`` ready / ``\\x01`` audio / ``\\x02`` text),
-collects the agent's reply audio + inner-monologue text, and writes a WAV.
-
-The browser UI at ``http://<host>:<port>/`` is the primary demo; this script is
-for headless testing / scripting.
-
-    python duplex_client.py --url ws://localhost:8124/api/chat \
-        --input user.wav --out reply.wav
-
-Deps: ``websockets``, ``sphn``, ``soundfile``, ``numpy``.
-"""
-
-from __future__ import annotations
-
-import argparse
-import asyncio
-
-import numpy as np
-import soundfile as sf
-import sphn
-
-SR = 24000  # PersonaPlex / Mimi codec rate
-
-
-async def converse(url: str, pcm: np.ndarray) -> tuple[np.ndarray, str]:
-    import websockets
-
-    reader = sphn.OpusStreamReader(SR)  # decode agent audio
-    writer = sphn.OpusStreamWriter(SR)  # encode our mic audio
-    text: list[str] = []
-    ready = asyncio.Event()
-    last = {"t": 0.0}
-
-    async with websockets.connect(url, max_size=None) as ws:
-
-        async def receive() -> None:
-            loop = asyncio.get_event_loop()
-            while True:
-                try:
-                    msg = await ws.recv()
-                except Exception:
-                    return
-                if not isinstance(msg, (bytes, bytearray)) or not msg:
-                    continue
-                tag, payload = msg[0], msg[1:]
-                if tag == 0:
-                    ready.set()
-                elif tag == 1:
-                    reader.append_bytes(bytes(payload))
-                    last["t"] = loop.time()
-                elif tag == 2:
-                    text.append(payload.decode("utf8", errors="replace"))
-
-        recv_task = asyncio.create_task(receive())
-        await asyncio.wait_for(ready.wait(), timeout=120)  # waits for system-prompt prefill
-
-        frame = 1920  # 80 ms
-        for i in range(0, len(pcm), frame):
-            chunk = pcm[i : i + frame]
-            if len(chunk) < frame:
-                chunk = np.concatenate([chunk, np.zeros(frame - len(chunk), dtype=np.float32)])
-            writer.append_pcm(chunk)
-            data = writer.read_bytes()
-            if data:
-                await ws.send(b"\x01" + data)
-            await asyncio.sleep(frame / SR)  # pace at real time
-
-        loop = asyncio.get_event_loop()
-        last["t"] = loop.time()
-        while loop.time() - last["t"] < 2.5:  # drain until the agent stops talking
-            await asyncio.sleep(0.1)
-        recv_task.cancel()
-
-    chunks: list[np.ndarray] = []
-    while True:
-        out = reader.read_pcm()
-        if out is None or out.shape[-1] == 0:
-            break
-        chunks.append(np.asarray(out, dtype=np.float32))
-    audio = np.concatenate(chunks) if chunks else np.zeros(0, dtype=np.float32)
-    return audio, "".join(text)
-
-
-def main() -> None:
-    ap = argparse.ArgumentParser()
-    ap.add_argument("--url", default="ws://localhost:8124/api/chat")
-    ap.add_argument("--input", required=True, help="24 kHz mono WAV of the user turn")
-    ap.add_argument("--out", default="personaplex_reply.wav")
-    ap.add_argument("--tail", type=float, default=10.0, help="trailing silence (s) for the agent reply")
-    ap.add_argument("--persona", default="")
-    ap.add_argument("--voice", default="NATF2.pt")
-    args = ap.parse_args()
-
-    wav, sr = sf.read(args.input, dtype="float32")
-    if wav.ndim > 1:
-        wav = wav.mean(axis=1)
-    if sr != SR:
-        raise SystemExit(f"input must be {SR} Hz mono; got {sr} Hz. Resample it first.")
-    wav = np.concatenate([wav, np.zeros(int(args.tail * SR), dtype=np.float32)])
-
-    url = f"{args.url}?text_prompt={args.persona}&voice_prompt={args.voice}"
-    audio, text = asyncio.run(converse(url, wav))
-    sf.write(args.out, audio if audio.size else np.zeros(SR, dtype=np.float32), SR)
-    print(f"agent reply: {audio.size / SR:.2f}s -> {args.out}")
-    print(f"inner-monologue: {text.strip()!r}")
-
-
-if __name__ == "__main__":
-    main()

examples/online_serving/personaplex/run_server.sh

@@ -1,15 +0,0 @@
-#!/bin/bash
-# PersonaPlex full-duplex online serving.
-#
-# Serves the OFFICIAL PersonaPlex web client at / (auto-downloaded from
-# nvidia/personaplex-7b-v1) and the Moshi WebSocket protocol at /api/chat,
-# backed by the native vLLM-Omni engine. Requires HF access to the gated repo.
-
-HOST="${HOST:-0.0.0.0}"
-PORT="${PORT:-8124}"
-
-echo "Starting PersonaPlex duplex server on ${HOST}:${PORT}..."
-echo "Open http://localhost:${PORT}/ for the official web client (run it near the server)."
-
-python -m vllm_omni.experimental.fullduplex.personaplex.serving.server \
-    --host "$HOST" --port "$PORT"