diff --git a/docs/design/fullduplex-personaplex.md b/docs/design/fullduplex-personaplex.md index 33d125c97..6d32cd8e3 100644 --- a/docs/design/fullduplex-personaplex.md +++ b/docs/design/fullduplex-personaplex.md @@ -271,3 +271,111 @@ The required evidence is: Audio that is empty, all zero, non-finite, or only a protocol `listen` event is not a successful end-to-end result. + +### Paced multi-session load driver + +The existing driver keeps its two-session admission and slot-reuse validation +when `--sessions` is omitted. Passing `--sessions N` selects a separate, +bounded load mode; it does not replace the lifecycle test or change server +admission capacity. Start an existing PersonaPlex server configured to admit +the requested number of sessions, then run from a source checkout with the +vLLM-Omni client dependencies installed: + +```bash +python -m tests.e2e.online_serving.personaplex_realtime_duplex \ + --model nvidia/personaplex-7b-v1 \ + --input-wav /path/to/input.wav \ + --sessions 2 --load-frames 1000 --drain-s 2 \ + --server-revision SERVER_COMMIT --server-hardware A100-80GB \ + --output-dir /tmp/personaplex-load-n2 +``` + +Use a new output directory for every run. The driver refuses to overwrite an +existing `load-result.json`. The server labels are supplied by the operator, +not detected or verified by the client. Record the server command, package +versions, checkpoint revision and hardware alongside a published measurement. + +All attempted sessions finish admission before admitted sessions share a +common input start. A rejected session remains a failed row in the report; +it is not removed from the denominator. Each session receives at most +`--load-frames` input frames, without looping the WAV, and uses 80 ms pacing. +A stalled send cannot cause a catch-up burst. The timeline records send +schedule lateness so an overloaded client is visible rather than mistaken +for a faster server. This mode measures steady concurrency, not join/leave +interference; the original lifecycle mode still checks slot reuse. +Load mode does not probe an additional N+1 session for `resource_exhausted`; +the lifecycle mode owns that admission-limit check. A per-session +`client_pacing_warning` and the summary's `client_pacing_warning_sessions` +identify send-lateness p99 above one nominal frame (80 ms). This is a warning, +not an extra acceptance gate: read it alongside server-facing measurements +so a slow client is not mistaken for a slow server. + +The fixed drain window is part of the measurement configuration. Audio +arriving during cleanup is excluded, so closing cannot rescue a session that +missed its observation deadline. Cleanup is attempted for admitted sessions +even when audio validation fails. Protocol errors, missing/invalid audio, +shared response IDs and cleanup failures make the report fail and the CLI +exit nonzero. Cancellation closes transports and propagates to the caller. +`--cleanup-timeout-s` (default 5 s) bounds session-close send and acknowledgement +together, independently of `--timeout-s`. The subsequent WebSocket close +handshake has the same separate bound; an unresponsive close acknowledgement +cannot hold cancellation for the full inference timeout. + +`load-result.json` retains per-session send and audio-packet timelines using +the client's monotonic clock. It does not export handshake credentials or +raw audio payloads. Metric definitions are explicit: + +Both modes validate each nonempty audio packet's strict base64 encoding, PCM16 +sample alignment, 24 kHz sample rate and response identity. Valid aggregate +audio cannot hide a malformed packet. In load mode, rejected packets make the session fail +and appear in `invalid_audio_packets` with their event index, receive time and +fixed diagnostic code; they do not contribute samples or timing intervals. +Empty audio deltas remain legal and contribute no samples. This check does +not change the public client's decoding. The lifecycle scenario and dispatch +remain the same, but malformed packets that aggregate checks once tolerated +now fail that mode as well. + +The probe accepts both the current OpenAI Realtime audio event name +`response.output_audio.delta` and the legacy `response.audio.delta`. Validation +uses the wire payload directly rather than relying on a particular client +library's alias table, so a server-side event-name migration cannot silently +turn real audio into a zero-output measurement. + +| Field | Meaning | +| --- | --- | +| `client_audio_packet_interval_ms` | Consecutive nonempty packet arrivals: median, linear p99, max, and zero-based index of the first maximal interval. Interval index 0 ends at packet 1. One packet can contain multiple codec frames; this is not server tick latency. | +| `client_first_audio_after_stream_start_ms` | First audio receipt minus first input send start; admission is excluded. | +| `client_stream_rtf` | First input send start to last audio receipt, divided by received audio duration. This includes real-time input pacing and is not an inference-only RTF. | +| `client_send_lateness_ms` | Send start minus its original shared schedule, clipped at zero. | +| `frame_deficit` | Integer input-frame minus output-frame count from the shared session validation; missing when valid whole-frame accounting is unavailable. | +| `output_frames`, `voiced_frames`, `silent_frames`, `frame_coverage_ratio`, `audio_rms`, `audio_chunks` | Session-validation statistics retained on success and on acceptance failure once computed. | +| `acceptance_check` | Failed local check, with numeric diagnostics in `error`; other exceptions retain phase/type only, without remote handshake or metadata text. | +| `audio_underrun_s`, `audio_underrun_event_count`, `audio_continuity_ok` | Shared `compute_continuity_stats` on valid PCM16 packet arrivals, with its default 100 ms underrun threshold; absent audio gives missing values rather than a continuity pass. | + +No output gives missing latency/RTF values, not zero. By default RTF is +diagnostic; `--max-client-rtf` adds an explicit ceiling. The existing frame +coverage, audible-frame and scheduler-data-plane checks still apply. A longer +`--drain-s` can demonstrate eventual completion under backlog, but it must not +be reported as real-time capacity when the fixed-window run failed or client +RTF exceeded the stated target. Likewise, the first live session can pay +one-time model/codec/JIT costs; either use the server's configured duplex +warmup path or label that run as cold-start evidence before comparing steady +capacity. Passing client-only metrics does not certify model quality or a +real-time capacity target on another GPU. Begin with N=1 and N=2; higher N is a +capacity experiment, not a promised property of an A100 or of this patch. +Continuity is a client-arrival diagnostic, not proof of speaker playback and +not an additional pass/fail gate. Packet-interval and send-lateness summaries +retain their documented linear p99; the older client interval helper reports +rounded nearest-rank p95 instead and is not substituted for that schema. + +Deterministic driver and localhost WebSocket checks need no model or GPU: + +```bash +python -m pytest tests/e2e/online_serving/test_personaplex_load_driver.py \ + -m 'core_model and cpu' --run-level core_model -q +``` + +These checks test the driver against controlled protocol responses. Run the +load-driver command against real weights separately before reporting model-serving +performance. Server tick timestamps and generic TTFT/TPOT aggregation remain +the responsibility of the existing duplex metrics paths, not this driver. diff --git a/tests/e2e/online_serving/personaplex_realtime_duplex.py b/tests/e2e/online_serving/personaplex_realtime_duplex.py index 20da1598d..78f012548 100644 --- a/tests/e2e/online_serving/personaplex_realtime_duplex.py +++ b/tests/e2e/online_serving/personaplex_realtime_duplex.py @@ -10,24 +10,24 @@ import base64 import hashlib import json import math +import time import uuid import wave -from collections.abc import Sequence +from collections.abc import Awaitable, Callable, Sequence from pathlib import Path +from typing import Protocol from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit import numpy as np try: import websockets - from websockets.exceptions import ConnectionClosed + from websockets.exceptions import ConnectionClosed, WebSocketException except ImportError as exc: # pragma: no cover - driver dependency. raise SystemExit("Install websockets first: pip install websockets") from exc -from vllm_omni.clients.duplex import ( - EventCollector, - write_pcm16_wav, -) +from vllm_omni.benchmarks.audio_continuity import compute_continuity_stats +from vllm_omni.clients.duplex import EventCollector, write_pcm16_wav from vllm_omni.clients.duplex import ( wait_for_condition as wait_for, ) @@ -35,6 +35,24 @@ from vllm_omni.clients.duplex import ( SAMPLE_RATE_HZ = 24_000 FRAME_SAMPLES = 1_920 FRAME_PERIOD_S = FRAME_SAMPLES / SAMPLE_RATE_HZ +AUDIO_DELTA_EVENT_TYPES = frozenset({"response.audio.delta", "response.output_audio.delta"}) + + +class _AudioAcceptanceError(AssertionError): + """A local acceptance failure with numeric diagnostics safe to report.""" + + def __init__(self, check: str, message: str, stats: dict[str, object]) -> None: + super().__init__(message) + self.check = check + self.stats = dict(stats) + + +class _ProbeTransport(Protocol): + async def recv(self) -> str | bytes: ... + + async def send(self, message: str) -> None: ... + + async def close(self) -> None: ... class RawRealtimeProbe: @@ -46,29 +64,33 @@ class RawRealtimeProbe: keeps a transport-only client and reads everything from ``.events``. """ - def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024) -> None: + def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024, close_timeout_s: float = 10.0) -> None: self.url = url self.max_size = max_size + self.close_timeout_s = close_timeout_s self.events = EventCollector() - self._ws = None + self._ws: _ProbeTransport | None = None self._reader_task: asyncio.Task[None] | None = None async def __aenter__(self) -> RawRealtimeProbe: - self._ws = await websockets.connect(self.url, max_size=self.max_size) + self._ws = await websockets.connect(self.url, max_size=self.max_size, close_timeout=self.close_timeout_s) self._reader_task = asyncio.create_task(self._read_events()) return self async def __aexit__(self, exc_type, exc, traceback) -> None: - if self._ws is not None: - await self._ws.close() - if self._reader_task is not None: - self._reader_task.cancel() - try: - await self._reader_task - except asyncio.CancelledError: - pass + try: + if self._ws is not None: + await self._ws.close() + finally: + if self._reader_task is not None: + self._reader_task.cancel() + try: + await self._reader_task + except asyncio.CancelledError: + pass async def _read_events(self) -> None: + assert self._ws is not None try: while True: raw = await self._ws.recv() @@ -81,6 +103,7 @@ class RawRealtimeProbe: return async def send(self, event: dict[str, object]) -> None: + assert self._ws is not None await self._ws.send(json.dumps(event)) @@ -142,6 +165,38 @@ def _events(client: RawRealtimeProbe, event_type: str) -> list[dict[str, object] return [event for event in client.events.events if event.get("type") == event_type] +def _audio_events(client: RawRealtimeProbe) -> list[dict[str, object]]: + """Return audio deltas from both current and legacy Realtime event names.""" + return [event for event in client.events.events if event.get("type") in AUDIO_DELTA_EVENT_TYPES] + + +def _validated_audio_event(event: dict[str, object]) -> bytes: + """Decode one wire audio delta without depending on client event aliases.""" + encoded = event.get("delta") + if not isinstance(encoded, str): + raise ValueError("missing_audio_delta") + try: + chunk = base64.b64decode(encoded, validate=True) + except ValueError: + raise ValueError("invalid_audio_base64") from None + if chunk: + if len(chunk) % 2: + raise ValueError("unaligned_pcm16_packet") + if event.get("sample_rate_hz") != SAMPLE_RATE_HZ: + raise ValueError("invalid_audio_sample_rate") + if not isinstance(event.get("response_id"), str) or not event.get("response_id"): + raise ValueError("missing_audio_response_id") + return chunk + + +def _validated_audio_chunks(client: RawRealtimeProbe) -> list[tuple[dict[str, object], bytes]]: + return [(event, _validated_audio_event(event)) for event in _audio_events(client)] + + +def _audio_bytes(client: RawRealtimeProbe) -> bytes: + return b"".join(chunk for _, chunk in _validated_audio_chunks(client) if chunk) + + async def _open_session( args: argparse.Namespace, *, @@ -151,22 +206,7 @@ async def _open_session( ) -> tuple[RawRealtimeProbe, dict[str, object]]: client = RawRealtimeProbe(_realtime_url(args.url, args.model, session_id)) await client.__aenter__() - await client.send( - { - "type": "session.update", - "session": { - "session_id": session_id, - "model": args.model, - "modalities": ["audio", "text"], - "input_audio_format": "pcm_f32le", - "output_audio_format": "pcm16", - "voice": args.voice, - "instructions": persona, - "turn_detection": None, - "extra_body": {"auto_response": True}, - }, - } - ) + await client.send(_session_update(args, session_id=session_id, persona=persona)) event_type = "error" if expect_error else "session.created" await wait_for( lambda: client.events.count(event_type) > 0, @@ -176,6 +216,23 @@ async def _open_session( return client, _events(client, event_type)[-1] +def _session_update(args: argparse.Namespace, *, session_id: str, persona: str) -> dict[str, object]: + return { + "type": "session.update", + "session": { + "session_id": session_id, + "model": args.model, + "modalities": ["audio", "text"], + "input_audio_format": "pcm_f32le", + "output_audio_format": "pcm16", + "voice": args.voice, + "instructions": persona, + "turn_detection": None, + "extra_body": {"auto_response": True}, + }, + } + + async def _close_session(client: RawRealtimeProbe, *, timeout_s: float) -> None: await client.send({"type": "session.close"}) await wait_for( @@ -200,23 +257,24 @@ async def _stream_frames( frame_count = min(frame_count, max_frames) for seq in range(frame_count): frame = pcm[seq * FRAME_SAMPLES : (seq + 1) * FRAME_SAMPLES] - frame = np.pad(frame, (0, FRAME_SAMPLES - frame.size)) - frame = np.ascontiguousarray(frame, dtype="= 1e-4), - } - ) + await client.send(_frame_event(frame, seq)) await asyncio.sleep(FRAME_PERIOD_S) return frame_count +def _frame_event(frame: np.ndarray, seq: int) -> dict[str, object]: + frame = np.ascontiguousarray(np.pad(frame, (0, FRAME_SAMPLES - frame.size)), dtype="= 1e-4), + } + + def _audio_frame_stats( raw: bytes, *, @@ -242,11 +300,7 @@ def _audio_frame_stats( def _response_ids(client: RawRealtimeProbe) -> set[str]: - return { - str(event["response_id"]) - for event in _events(client, "response.output_audio.delta") - if isinstance(event.get("response_id"), str) - } + return {str(event["response_id"]) for event in _audio_events(client) if isinstance(event.get("response_id"), str)} def _session_result( @@ -256,40 +310,66 @@ def _session_result( args: argparse.Namespace, minimum_chunks: int, ) -> tuple[bytes, set[str], dict[str, object]]: - chunks = sum(len(items) for items in client.events.response_audio.values()) - raw = client.events.audio_bytes() - if chunks < minimum_chunks or not raw or len(raw) % 2: - raise AssertionError(f"invalid PCM16 output: chunks={chunks}, bytes={len(raw)}") + if isinstance(client, RawRealtimeProbe): + audio_chunks = _validated_audio_chunks(client) + chunks = sum(bool(chunk) for _, chunk in audio_chunks) + raw = b"".join(chunk for _, chunk in audio_chunks if chunk) + else: + # Unit fixtures may provide an already-decoded EventCollector without + # retaining the base64 wire field. Real websocket probes always take + # the strict wire-validation path above. + chunks = sum(len(items) for items in client.events.response_audio.values()) + raw = client.events.audio_bytes() + stats: dict[str, object] = {"audio_chunks": chunks, "audio_samples": len(raw) // 2} + if not raw or len(raw) % 2: + raise _AudioAcceptanceError("pcm16_output", f"invalid PCM16 output: chunks={chunks}, bytes={len(raw)}", stats) pcm = np.frombuffer(raw, dtype=" async def run(args: argparse.Namespace) -> dict[str, object]: + if getattr(args, "sessions", None) is not None: + return await run_load(args) input_path = Path(args.input_wav) input_identity = _input_identity( input_path, @@ -381,14 +464,12 @@ async def run(args: argparse.Namespace) -> dict[str, object]: ) await asyncio.gather( wait_for( - lambda: len(primary.events.audio_bytes()) // (2 * FRAME_SAMPLES) >= primary_frames - args.max_frame_deficit, + lambda: len(_audio_bytes(primary)) // (2 * FRAME_SAMPLES) >= primary_frames - args.max_frame_deficit, timeout_s=args.timeout_s, label="primary frame coverage", ), wait_for( - lambda: ( - len(secondary.events.audio_bytes()) // (2 * FRAME_SAMPLES) >= secondary_frames - args.max_frame_deficit - ), + lambda: (len(_audio_bytes(secondary)) // (2 * FRAME_SAMPLES) >= secondary_frames - args.max_frame_deficit), timeout_s=args.timeout_s, label="secondary frame coverage", ), @@ -426,16 +507,14 @@ async def run(args: argparse.Namespace) -> dict[str, object]: await asyncio.gather( wait_for( lambda: ( - len(secondary.events.audio_bytes()) // (2 * FRAME_SAMPLES) - >= secondary_total_frames - args.max_frame_deficit + len(_audio_bytes(secondary)) // (2 * FRAME_SAMPLES) >= secondary_total_frames - args.max_frame_deficit ), timeout_s=args.timeout_s, label="survivor frame coverage after slot recycle", ), wait_for( lambda: ( - len(replacement.events.audio_bytes()) // (2 * FRAME_SAMPLES) - >= replacement_frames - args.max_frame_deficit + len(_audio_bytes(replacement)) // (2 * FRAME_SAMPLES) >= replacement_frames - args.max_frame_deficit ), timeout_s=args.timeout_s, label="replacement frame coverage", @@ -493,6 +572,284 @@ async def run(args: argparse.Namespace) -> dict[str, object]: return result +def _distribution(values: list[float]) -> dict[str, int | float | None]: + if not values: + return {"count": 0, "median": None, "p99": None, "max": None, "argmax_index": None} + return { + "count": len(values), + "median": float(np.median(values)), + "p99": float(np.quantile(values, 0.99)), + "max": max(values), + "argmax_index": int(np.argmax(values)), + } + + +def _load_metrics(client: RawRealtimeProbe, sends: list[tuple[float, float, float]]) -> dict[str, object]: + # Use the collector's local clock, never a timestamp supplied by the server. + packets: list[dict[str, object]] = [] + response_ids: set[str] = set() + packet_times: list[float] = [] + packet_bytes: list[int] = [] + invalid_packets: list[dict[str, object]] = [] + samples = 0 + for index, (raw, received) in enumerate(zip(client.events.events, client.events.event_received_at_s, strict=True)): + if raw.get("type") not in AUDIO_DELTA_EVENT_TYPES: + continue + try: + chunk = _validated_audio_event(raw) + except ValueError as exc: + # Only fixed diagnostic codes from the validator are exported, + # never audio payloads or details from the remote endpoint. + invalid_packets.append({"event_index": index, "received_at_s": received, "code": str(exc)}) + continue + if chunk: + samples += len(chunk) // 2 + packet_times.append(received) + packet_bytes.append(len(chunk)) + response_id = raw.get("response_id") + packets.append({"received_at_s": received, "samples": len(chunk) // 2, "response_id": response_id}) + if isinstance(response_id, str) and response_id: + response_ids.add(response_id) + intervals = [(b - a) * 1000 for a, b in zip(packet_times, packet_times[1:])] + first_send = sends[0][1] if sends else None + audio_duration = samples / SAMPLE_RATE_HZ + first_audio_ms = (packet_times[0] - first_send) * 1000 if packet_times and first_send is not None else None + client_rtf = (packet_times[-1] - first_send) / audio_duration if audio_duration and first_send is not None else None + continuity = compute_continuity_stats(packet_times, packet_bytes, SAMPLE_RATE_HZ) if packet_times else None + lateness = _distribution([max(0.0, sent - planned) * 1000 for planned, sent, _ in sends]) + p99_lateness = lateness["p99"] + return { + "input_frames": len(sends), + "output_samples": samples, + # Populate frame accounting from _session_result, including on failure. + "frame_deficit": None, + "response_ids": sorted(response_ids), + "client_first_audio_after_stream_start_ms": first_audio_ms, + "client_stream_rtf": client_rtf, + "client_audio_packet_interval_ms": _distribution(intervals), + "client_send_lateness_ms": lateness, + "client_pacing_warning": isinstance(p99_lateness, float) and p99_lateness > FRAME_PERIOD_S * 1000, + "audio_underrun_s": continuity.max_underrun_s if continuity else None, + "audio_underrun_event_count": continuity.underrun_event_count if continuity else None, + "audio_continuity_ok": continuity.is_continuous if continuity else None, + "input_timeline": [ + {"frame_index": i, "planned_at_s": planned, "send_started_at_s": sent, "send_completed_at_s": completed} + for i, (planned, sent, completed) in enumerate(sends) + ], + "audio_packet_timeline": packets, + "invalid_audio_packets": invalid_packets, + } + + +def _check_load_connection(client: RawRealtimeProbe) -> None: + if client.events.errors(): + raise RuntimeError("server emitted a Realtime error (see the session error count)") + if client._reader_task is not None and client._reader_task.done(): + client._reader_task.result() # Propagate malformed JSON/reader failures. + raise RuntimeError("connection closed before session completion") + + +async def _wait_load_event(client: RawRealtimeProbe, event_type: str, timeout_s: float) -> None: + def received() -> bool: + if client.events.count(event_type): + return True + _check_load_connection(client) + return False + + await wait_for(received, timeout_s=timeout_s, label=event_type) + + +async def _paced_load_frames( + client: RawRealtimeProbe, + pcm: np.ndarray, + *, + epoch: float, + timeout_s: float, + sends: list[tuple[float, float, float]], + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, +) -> None: + next_send = epoch + for seq in range(math.ceil(pcm.size / FRAME_SAMPLES)): + await sleep(max(0.0, next_send - clock())) + _check_load_connection(client) + event = _frame_event(pcm[seq * FRAME_SAMPLES : (seq + 1) * FRAME_SAMPLES], seq) + started = clock() + await asyncio.wait_for(client.send(event), timeout=timeout_s) + completed = clock() + sends.append((epoch + seq * FRAME_PERIOD_S, started, completed)) + # Do not burst to catch up after a stalled send; report accumulated lag. + next_send = max(epoch + (seq + 1) * FRAME_PERIOD_S, completed + FRAME_PERIOD_S) + + +async def _request_load_close(client: RawRealtimeProbe, timeout_s: float) -> None: + await client.send({"type": "session.close"}) + await _wait_load_event(client, "session.closed", timeout_s) + + +async def _run_load_session( + args: argparse.Namespace, + pcm: np.ndarray, + index: int, + ready: asyncio.Future[None], + start: asyncio.Future[float], +) -> dict[str, object]: + session_id = f"personaplex-load-{index}-{uuid.uuid4().hex}" + client = RawRealtimeProbe(_realtime_url(args.url, args.model, session_id), close_timeout_s=args.cleanup_timeout_s) + sends: list[tuple[float, float, float]] = [] + error: str | None = None + cleanup_error: str | None = None + metrics: dict[str, object] | None = None + session_stats: dict[str, object] = {} + acceptance_check: str | None = None + phase = "admission" + try: + async with client: + try: + await asyncio.wait_for( + client.send(_session_update(args, session_id=session_id, persona=args.persona)), args.timeout_s + ) + await _wait_load_event(client, "session.created", args.timeout_s) + capabilities = _capabilities(_events(client, "session.created")[-1]) + if ( + capabilities.get("chunk_period_ms") != 80 + or capabilities.get("supports_multi_session_same_replica") is not True + ): + raise AssertionError("server did not advertise PersonaPlex multi-session capabilities") + ready.set_result(None) + phase = "stream" + epoch = await asyncio.shield(start) + await _paced_load_frames(client, pcm, epoch=epoch, timeout_s=args.timeout_s, sends=sends) + phase = "drain" + # Fixed, reported observation window. Closing must not extend + # it and let a slow session pass by flushing after the cutoff. + await asyncio.sleep(args.drain_s) + _check_load_connection(client) + _, _, session_stats = _session_result( + client, input_frames=len(sends), args=args, minimum_chunks=args.minimum_audio_chunks + ) + finally: + metrics = _load_metrics(client, sends) + if client.events.count("session.created") and not client.events.count("session.closed"): + try: + # Send and acknowledgement share one short cleanup budget, + # independent of the potentially long inference timeout. + await asyncio.wait_for( + _request_load_close(client, args.cleanup_timeout_s), args.cleanup_timeout_s + ) + except (OSError, asyncio.TimeoutError, ValueError, RuntimeError, WebSocketException) as exc: + cleanup_error = f"cleanup: {type(exc).__name__}" + except (OSError, asyncio.TimeoutError, ValueError, RuntimeError, AssertionError, WebSocketException) as exc: + if isinstance(exc, _AudioAcceptanceError): + acceptance_check = exc.check + session_stats = exc.stats + error = f"{phase}: AssertionError: {exc}" + else: + # Other assertions/errors may embed handshake credentials or wire + # metadata. Only our local numeric acceptance diagnostics are safe. + error = f"{phase}: {type(exc).__name__}" + finally: + # Admission failure must not leave the other sessions at the barrier. + if not ready.done(): + ready.set_result(None) + if metrics is None: + metrics = _load_metrics(client, sends) + if cleanup_error: + error = f"{error}; {cleanup_error}" if error else cleanup_error + if client.events.errors(): + error = error or "server emitted a Realtime error" + if metrics["invalid_audio_packets"]: + error = error or "invalid audio packets" + first_audio = metrics["client_first_audio_after_stream_start_ms"] + if isinstance(first_audio, float) and first_audio < 0: + error = error or "audio arrived before the first input frame" + rtf = metrics["client_stream_rtf"] + if args.max_client_rtf is not None and isinstance(rtf, float) and rtf > args.max_client_rtf: + error = error or f"client_stream_rtf exceeds {args.max_client_rtf}" + return { + "index": index, + "session_id": session_id, + "ok": error is None, + "error": error, + "acceptance_check": acceptance_check, + "server_error_count": len(client.events.errors()), + "server_error_codes": [ + str(body.get("code", "unknown")) if isinstance(body := event.get("error"), dict) else "unknown" + for event in client.events.errors() + ], + **metrics, + **session_stats, + } + + +async def run_load(args: argparse.Namespace) -> dict[str, object]: + input_path = Path(args.input_wav) + identity = _input_identity(input_path, expected_sha256=args.expected_input_sha256) + pcm = _read_wav_as_float32(input_path) + if args.tail_s: + pcm = np.concatenate([pcm, np.zeros(round(args.tail_s * SAMPLE_RATE_HZ), dtype=np.float32)]) + pcm = pcm[: args.load_frames * FRAME_SAMPLES] + output_dir = Path(args.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + # Refuse to silently overwrite evidence from a previous sweep. + result_path = output_dir / "load-result.json" + if result_path.exists(): + raise FileExistsError(result_path) + loop = asyncio.get_running_loop() + start: asyncio.Future[float] = loop.create_future() + ready: list[asyncio.Future[None]] = [loop.create_future() for _ in range(args.sessions)] + tasks = [asyncio.create_task(_run_load_session(args, pcm, i, r, start)) for i, r in enumerate(ready)] + try: + await asyncio.gather(*ready) + start.set_result(time.monotonic()) + rows = await asyncio.gather(*tasks) + finally: + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + seen: set[str] = set() + shared: set[str] = set() + for row in rows: + assert isinstance(row["response_ids"], list) + response_ids = set(row["response_ids"]) + shared.update(seen & response_ids) + seen.update(response_ids) + for row in rows: + assert isinstance(row["response_ids"], list) + if shared.intersection(row["response_ids"]): + row.update(ok=False, error="concurrent sessions shared a response id") + result = { + "schema_version": 1, + "mode": "paced_multi_session", + "model": args.model, + "requested_sessions": args.sessions, + "passed_sessions": sum(row["ok"] is True for row in rows), + "ok": all(row["ok"] is True for row in rows), + "input": {**identity, "tail_s": args.tail_s, "load_frames": args.load_frames}, + "frame_period_s": FRAME_PERIOD_S, + "drain_s": args.drain_s, + "cleanup_timeout_s": args.cleanup_timeout_s, + "client_pacing_warning_threshold_ms": FRAME_PERIOD_S * 1000, + "client_pacing_warning_sessions": [row["index"] for row in rows if row["client_pacing_warning"]], + "max_client_rtf": args.max_client_rtf, + "acceptance": { + "max_frame_deficit": args.max_frame_deficit, + "min_voiced_frames": args.min_voiced_frames, + "minimum_audio_chunks": args.minimum_audio_chunks, + "voiced_frame_rms_threshold": args.voiced_frame_rms_threshold, + }, + "shared_response_ids": sorted(shared), + "sessions": rows, + "reported_server": {"revision": args.server_revision, "hardware": args.server_hardware}, + "measurement_origin": "client monotonic clock; admission excluded; streaming and drain retained", + "interval_definition": "nonempty audio packet arrivals, not server ticks or per-frame inference latency", + } + with result_path.open("x", encoding="utf-8") as stream: + json.dump(result, stream, ensure_ascii=False, indent=2, allow_nan=False) + return result + + def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--url", default="ws://127.0.0.1:8099/v1/realtime?duplex=1") @@ -510,6 +867,12 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser.add_argument("--tail-s", type=float, default=2.0) parser.add_argument("--drain-s", type=float, default=2.0) parser.add_argument("--timeout-s", type=float, default=180.0) + parser.add_argument( + "--cleanup-timeout-s", + type=float, + default=5.0, + help="Load mode: shared session-close/ack budget; also bounds the websocket close handshake.", + ) parser.add_argument( "--max-frame-deficit", type=int, @@ -536,11 +899,39 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: default=0, help="Frames sent after slot reuse; 0 replays the full input so the replacement must produce audible output.", ) - return parser.parse_args(argv) + parser.add_argument( + "--sessions", type=int, help="Opt into paced load mode with N sessions; omit for the lifecycle test." + ) + parser.add_argument( + "--load-frames", type=int, default=1000, help="Maximum input frames per load session (no repetition)." + ) + parser.add_argument("--max-client-rtf", type=float, help="Optional client stream-window RTF acceptance ceiling.") + parser.add_argument("--server-revision", help="User-reported server commit/version; not verified by this client.") + parser.add_argument( + "--server-hardware", help="User-reported server hardware; not inferred from the client machine." + ) + args = parser.parse_args(argv) + if args.sessions is not None: + if not 1 <= args.sessions <= 64 or not 1 <= args.load_frames <= 10000: + parser.error("load mode requires sessions in [1,64] and load-frames in [1,10000]") + for name in ("timeout_s", "cleanup_timeout_s", "drain_s", "tail_s"): + value = getattr(args, name) + if not math.isfinite(value) or value < 0 or (name in ("timeout_s", "cleanup_timeout_s") and value == 0): + parser.error(f"{name} must be finite and non-negative (timeout must be positive)") + if args.max_client_rtf is not None and (not math.isfinite(args.max_client_rtf) or args.max_client_rtf <= 0): + parser.error("max-client-rtf must be finite and positive") + if args.max_frame_deficit < 0 or args.min_voiced_frames < 1 or args.minimum_audio_chunks < 1: + parser.error("frame-deficit must be non-negative; voiced frames and audio chunks must be positive") + if not math.isfinite(args.voiced_frame_rms_threshold) or args.voiced_frame_rms_threshold < 0: + parser.error("voiced-frame-rms-threshold must be finite and non-negative") + return args def main() -> None: - print(json.dumps(asyncio.run(run(parse_args())), ensure_ascii=False, indent=2)) + result = asyncio.run(run(parse_args())) + print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False)) + if result.get("mode") == "paced_multi_session" and not result["ok"]: + raise SystemExit(1) if __name__ == "__main__": diff --git a/tests/e2e/online_serving/test_personaplex_load_driver.py b/tests/e2e/online_serving/test_personaplex_load_driver.py new file mode 100644 index 000000000..8e9a6a28d --- /dev/null +++ b/tests/e2e/online_serving/test_personaplex_load_driver.py @@ -0,0 +1,568 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +import asyncio +import base64 +import json +import wave +from contextlib import asynccontextmanager +from typing import TypedDict + +import numpy as np +import pytest +import websockets + +from tests.e2e.online_serving import personaplex_realtime_duplex as driver + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +class _ServerState(TypedDict): + index: int + closed: bool + frames: int + close_requested: bool + + +def _args(tmp_path, *extra): + wav_path = tmp_path / "input.wav" + with wave.open(str(wav_path), "wb") as stream: + stream.setnchannels(1) + stream.setsampwidth(2) + stream.setframerate(driver.SAMPLE_RATE_HZ) + stream.writeframes(np.full(driver.FRAME_SAMPLES * 2, 2000, dtype="= 1 + assert report["client_pacing_warning"] is False + assert "secret" not in json.dumps(report) + assert "-999" not in json.dumps(report) + + +def test_empty_output_has_missing_not_zero_latency(): + report = driver._load_metrics(driver.RawRealtimeProbe("ws://unused"), []) + assert report["client_stream_rtf"] is None + assert report["audio_continuity_ok"] is None + assert report["audio_underrun_s"] is None + intervals = report["client_audio_packet_interval_ms"] + assert isinstance(intervals, dict) + assert intervals["p99"] is None + + +def test_failed_acceptance_retains_frame_counts_without_remote_data(): + client = driver.RawRealtimeProbe("ws://unused") + client.events.add(_audio(), received_at_s=1.0) + args = driver.parse_args( + ["--model", "unused", "--input-wav", "unused.wav", "--max-frame-deficit", "0", "--min-voiced-frames", "1"] + ) + with pytest.raises(AssertionError) as raised: + driver._session_result(client, input_frames=2, args=args, minimum_chunks=1) + assert getattr(raised.value, "check", None) == "frame_deficit" + stats = getattr(raised.value, "stats", {}) + assert stats["frame_deficit"] == 1 and type(stats["frame_deficit"]) is int + assert stats["output_frames"] == stats["voiced_frames"] == 1 + assert stats["frame_coverage_ratio"] == 0.5 + + +@pytest.mark.parametrize("event_type", sorted(driver.AUDIO_DELTA_EVENT_TYPES)) +def test_session_result_accepts_current_and_legacy_audio_event_names(tmp_path, event_type): + client = driver.RawRealtimeProbe("ws://unused") + client.events.add(_audio(samples=driver.FRAME_SAMPLES, event_type=event_type), received_at_s=1.0) + raw, response_ids, stats = driver._session_result( + client, + input_frames=1, + args=_args(tmp_path, "--sessions", "1"), + minimum_chunks=1, + ) + assert len(raw) == driver.FRAME_SAMPLES * 2 + assert response_ids == {"r"} + assert stats["frame_deficit"] == 0 + + +def test_legacy_audio_event_name_contributes_load_metrics(): + client = driver.RawRealtimeProbe("ws://unused") + client.events.add(_audio(samples=driver.FRAME_SAMPLES * 5, event_type="response.audio.delta"), received_at_s=10.4) + report = driver._load_metrics(client, [(10.0, 10.0, 10.01)]) + assert report["output_samples"] == driver.FRAME_SAMPLES * 5 + assert report["response_ids"] == ["r"] + assert report["invalid_audio_packets"] == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sessions", [1, 2, 4]) +async def test_real_websocket_load_success(tmp_path, sessions): + args = _args(tmp_path, "--sessions", str(sessions)) + async with _server() as (url, states): + args.url = url + result = await driver.run(args) + assert result["ok"] is True + assert result["passed_sessions"] == sessions + rows = result["sessions"] + assert isinstance(rows, list) + assert len(rows) == sessions + assert all(row["frame_deficit"] == 0 for row in rows) + for row in rows: + assert type(row["frame_deficit"]) is int + assert row["output_frames"] == row["voiced_frames"] == row["audio_chunks"] == 2 + assert row["silent_frames"] == 0 + assert row["frame_coverage_ratio"] == 1.0 + assert row["audio_rms"] == pytest.approx(2000 / 32768) + assert row["acceptance_check"] is None + assert isinstance(row["audio_continuity_ok"], bool) + assert all(s["closed"] and s["close_requested"] for s in states) + assert "must-not-be-exported" not in json.dumps(result) + saved = json.loads((tmp_path / "out/load-result.json").read_text()) + assert saved == result + origins = [row["input_timeline"][0]["planned_at_s"] for row in rows] + assert len(set(origins)) == 1 + + +@pytest.mark.asyncio +async def test_real_websocket_load_accepts_legacy_audio_event_name(tmp_path): + args = _args(tmp_path, "--sessions", "2") + async with _server("legacy_audio_event") as (url, states): + args.url = url + result = await driver.run(args) + assert result["ok"] is True + assert result["passed_sessions"] == 2 + assert result["shared_response_ids"] == [] + assert all(s["closed"] and s["close_requested"] for s in states) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mode", + [ + "reject_second", + "silent", + "disconnect", + "no_close_ack", + "shared_response", + "late_on_close", + "error_on_close", + "bad_json", + "missing_capabilities", + ], +) +async def test_real_websocket_failures_remain_in_results(tmp_path, mode): + args = _args(tmp_path) + async with _server(mode) as (url, states): + args.url = url + result = await asyncio.wait_for(driver.run(args), timeout=4) + assert result["ok"] is False + assert "must-not-be-exported" not in json.dumps(result) + assert result["requested_sessions"] == 2 + rows = result["sessions"] + assert isinstance(rows, list) + assert len(rows) == 2 + assert all(s["closed"] for s in states) + if mode == "reject_second": + assert result["passed_sessions"] == 1 + rejected = next(row for row in rows if not row["ok"]) + assert rejected["input_frames"] == 0 + assert rejected["client_stream_rtf"] is None + assert rejected["server_error_codes"] == ["resource_exhausted"] + if mode == "silent": + # A failed measurement must still release its admitted session. + assert all(s["close_requested"] for s in states) + if mode == "late_on_close": + assert all(row["output_samples"] == 0 for row in rows) + if mode == "shared_response": + assert result["passed_sessions"] == 0 + + +@pytest.mark.asyncio +async def test_existing_report_is_not_overwritten(tmp_path): + args = _args(tmp_path) + path = tmp_path / "out/load-result.json" + path.parent.mkdir() + path.write_text("previous measurement") + with pytest.raises(FileExistsError): + await driver.run(args) + assert path.read_text() == "previous measurement" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["success", "no_close_ack"]) +async def test_cancellation_closes_all_connections(tmp_path, mode): + args = _args(tmp_path, "--drain-s", "10", "--timeout-s", "30") + async with _server(mode) as (url, states): + args.url = url + task = asyncio.create_task(driver.run(args)) + + async def started(): + while len(states) < 2 or not all(state["frames"] for state in states): + await asyncio.sleep(0.01) + + await asyncio.wait_for(started(), timeout=3) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=3) + assert all(s["closed"] for s in states) + + +@pytest.mark.asyncio +async def test_client_rtf_ceiling_marks_slow_sessions_failed(tmp_path): + args = _args(tmp_path, "--max-client-rtf", "0.001") + async with _server() as (url, _): + args.url = url + result = await driver.run(args) + assert result["passed_sessions"] == 0 + rows = result["sessions"] + assert isinstance(rows, list) + assert all("client_stream_rtf exceeds" in row["error"] for row in rows) + + +@pytest.mark.asyncio +async def test_stalled_send_does_not_cause_catchup_burst(): + class Clock: + now = 10.0 + + def monotonic(self): + return self.now + + async def sleep(self, delay): + self.now += delay + + class Client(driver.RawRealtimeProbe): + async def send(self, event): + if event["audio_end_ms"] == 80: + clock.now += 0.25 + + clock = Clock() + sends: list[tuple[float, float, float]] = [] + await driver._paced_load_frames( + Client("ws://unused"), + np.zeros(driver.FRAME_SAMPLES * 3, dtype="= sends[0][2] + driver.FRAME_PERIOD_S + assert sends[1][1] - sends[1][0] == pytest.approx(0.25) + assert sends[2][1] >= sends[1][2] + driver.FRAME_PERIOD_S + metrics = driver._load_metrics(Client("ws://unused"), sends) + assert metrics["client_pacing_warning"] is True + assert asyncio.sleep != clock.sleep + + +@pytest.mark.asyncio +async def test_failed_frame_acceptance_preserves_numeric_diagnostics(tmp_path): + args = _args(tmp_path, "--sessions", "1") + async with _server("short_audio") as (url, states): + args.url = url + result = await driver.run(args) + rows = result["sessions"] + assert isinstance(rows, list) + row = rows[0] + assert row["ok"] is False + assert row["acceptance_check"] == "frame_deficit" + assert row["frame_deficit"] == 1 and type(row["frame_deficit"]) is int + assert row["output_frames"] == row["voiced_frames"] == row["audio_chunks"] == 1 + assert row["frame_coverage_ratio"] == 0.5 + assert row["audio_rms"] == pytest.approx(2000 / 32768) + assert "input=2" in row["error"] and "maximum_deficit=0" in row["error"] + assert all(state["closed"] and state["close_requested"] for state in states) + assert json.loads((tmp_path / "out/load-result.json").read_text()) == result + + +@pytest.mark.asyncio +async def test_acceptance_failure_does_not_export_remote_metadata(tmp_path): + args = _args(tmp_path, "--sessions", "1") + async with _server("private_metadata") as (url, _): + args.url = url + result = await driver.run(args) + rows = result["sessions"] + assert isinstance(rows, list) + row = rows[0] + assert row["ok"] is False + assert row["acceptance_check"] == "scheduler_data_plane" + assert "remote-detail" not in json.dumps(result) + assert "must-not-be-exported" not in json.dumps(result) + + +@pytest.mark.parametrize("ok", [True, False]) +def test_load_main_exit_status(tmp_path, monkeypatch, ok): + args = _args(tmp_path) + + async def completed(_args): + return {"mode": "paced_multi_session", "ok": ok} + + monkeypatch.setattr(driver, "parse_args", lambda: args) + monkeypatch.setattr(driver, "run", completed) + if ok: + driver.main() + else: + with pytest.raises(SystemExit) as exc: + driver.main() + assert exc.value.code == 1 + + +@pytest.mark.asyncio +async def test_handshake_rejection_is_retained_without_server_details(tmp_path, monkeypatch): + from websockets.exceptions import InvalidHandshake + + async def rejected(_self): + raise InvalidHandshake("credentials-in-remote-handshake-detail") + + monkeypatch.setattr(driver.RawRealtimeProbe, "__aenter__", rejected) + result = await asyncio.wait_for(driver.run(_args(tmp_path)), timeout=3) + assert result["ok"] is False + rows = result["sessions"] + assert isinstance(rows, list) and len(rows) == 2 + assert all(row["error"] == "admission: InvalidHandshake" for row in rows) + assert "credentials-in-remote-handshake-detail" not in json.dumps(result) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sessions", [1, 2]) +@pytest.mark.parametrize( + "mode", + [ + "missing_packet_rate", + "string_packet_rate", + "invalid_base64", + "orphan_audio", + "odd_packet", + ], +) +async def test_load_rejects_invalid_individual_audio_packets(tmp_path, sessions, mode): + # The concatenated PCM can still meet coverage/voicing thresholds, so the + # validator must also check each packet before accepting the whole run. + args = _args(tmp_path, "--sessions", str(sessions)) + async with _server(mode) as (url, states): + args.url = url + result = await asyncio.wait_for(driver.run(args), timeout=4) + assert result["ok"] is False + assert result["passed_sessions"] == 0 + rows = result["sessions"] + assert isinstance(rows, list) and len(rows) == sessions + assert all(row["error"] is not None for row in rows) + expected_code = { + "missing_packet_rate": "invalid_audio_sample_rate", + "string_packet_rate": "invalid_audio_sample_rate", + "invalid_base64": "invalid_audio_base64", + "orphan_audio": "missing_audio_response_id", + "odd_packet": "unaligned_pcm16_packet", + }[mode] + assert all({packet["code"] for packet in row["invalid_audio_packets"]} == {expected_code} for row in rows) + assert all( + sum(packet["samples"] for packet in row["audio_packet_timeline"]) == row["output_samples"] for row in rows + ) + assert all(state["closed"] and state["close_requested"] for state in states) + assert json.loads((tmp_path / "out/load-result.json").read_text()) == result + + +def test_load_allows_empty_and_multi_frame_audio_packets(): + client = driver.RawRealtimeProbe("ws://unused") + client.events.add(_audio(samples=0), received_at_s=1.0) + client.events.add(_audio(samples=driver.FRAME_SAMPLES * 5), received_at_s=1.4) + metrics = driver._load_metrics(client, [(1.0, 1.0, 1.001)]) + assert metrics["invalid_audio_packets"] == [] + assert metrics["output_samples"] == driver.FRAME_SAMPLES * 5 + packets = metrics["audio_packet_timeline"] + assert isinstance(packets, list) and len(packets) == 1 + + +def test_load_packet_diagnostics_do_not_copy_remote_payload(): + client = driver.RawRealtimeProbe("ws://unused") + event = _audio() + event["delta"] = "private-remote-payload!" + client.events.add(event, received_at_s=1.4) + metrics = driver._load_metrics(client, []) + assert metrics["output_samples"] == 0 + assert metrics["invalid_audio_packets"] == [ + { + "event_index": 0, + "received_at_s": 1.4, + "code": "invalid_audio_base64", + } + ] + assert "private-remote-payload" not in json.dumps(metrics)