anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

[Model][PersonaPlex] Add paced multi-session validation (#7428)

Browse Source 
Signed-off-by: luozijian <luozijian0924@gmail.com>
This commit is contained in:
LOGO127 authored and GitHub committed 2026-09-16 23:04:37 -07:00
1 parent 4200768f59
commit 224152566a
3 files changed
+1155 -88
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/design/fullduplex-personaplex.md
+108
View File
@@ -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.
tests/e2e/online_serving/personaplex_realtime_duplex.py
+479 -88
View File
@@ -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="<f4")
await client.send(
{
"type": "input_audio_buffer.append",
"audio": base64.b64encode(frame).decode("ascii"),
"format": "pcm_f32le",
"sample_rate_hz": SAMPLE_RATE_HZ,
"duration_ms": 80,
"audio_end_ms": (seq + 1) * 80,
"is_speech": bool(np.sqrt(np.mean(np.square(frame))) >= 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="<f4")
return {
"type": "input_audio_buffer.append",
"audio": base64.b64encode(frame).decode("ascii"),
"format": "pcm_f32le",
"sample_rate_hz": SAMPLE_RATE_HZ,
"duration_ms": 80,
"audio_end_ms": (seq + 1) * 80,
"is_speech": bool(np.sqrt(np.mean(np.square(frame))) >= 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="<i2").astype(np.float32) / 32768.0
rms = float(np.sqrt(np.mean(np.square(pcm))))
stats["audio_rms"] = rms
# This low floor only catches corrupt, non-finite, or effectively zero PCM.
# Audible speech is checked per frame below with the higher 1e-3 default.
if not np.isfinite(pcm).all() or rms <= 1e-5:
raise AssertionError(f"output audio is non-finite or silent: samples={pcm.size}, rms={rms}")
rates = {
int(event["sample_rate_hz"])
for event in _events(client, "response.output_audio.delta")
if isinstance(event.get("sample_rate_hz"), int)
}
raise _AudioAcceptanceError(
"audio_rms", f"output audio is non-finite or silent: samples={pcm.size}, rms={rms}", stats
)
rates = {rate for event in _audio_events(client) if isinstance(rate := event.get("sample_rate_hz"), int)}
if rates != {SAMPLE_RATE_HZ}:
raise AssertionError(f"unexpected output sample rates: {sorted(rates)}")
stats = _audio_frame_stats(
raw,
input_frames=input_frames,
voiced_frame_rms_threshold=args.voiced_frame_rms_threshold,
)
if not 0 <= int(stats["frame_deficit"]) <= args.max_frame_deficit:
raise AssertionError(f"invalid synchronous frame accounting: input={input_frames}, stats={stats}")
raise _AudioAcceptanceError("sample_rate", f"unexpected output sample rates: {sorted(rates)}", stats)
if input_frames < 1:
raise _AudioAcceptanceError("input_frames", f"audio received without input frames: input={input_frames}", stats)
try:
frame_stats = _audio_frame_stats(
raw,
input_frames=input_frames,
voiced_frame_rms_threshold=args.voiced_frame_rms_threshold,
)
except AssertionError as exc:
# _audio_frame_stats only formats local sample counts, not wire data.
raise _AudioAcceptanceError("whole_codec_frames", str(exc), stats) from exc
stats.update(frame_stats)
if chunks < minimum_chunks:
raise _AudioAcceptanceError(
"audio_chunks", f"insufficient audio chunks: chunks={chunks}, minimum={minimum_chunks}", stats
)
if not 0 <= int(frame_stats["frame_deficit"]) <= args.max_frame_deficit:
raise _AudioAcceptanceError(
"frame_deficit",
f"invalid synchronous frame accounting: input={input_frames}, maximum_deficit={args.max_frame_deficit}, "
f"stats={stats}",
stats,
)
min_voiced_frames = int(args.min_voiced_frames)
if min_voiced_frames < 1:
raise ValueError(f"min_voiced_frames must be positive, got {min_voiced_frames}")
if int(stats["voiced_frames"]) < min_voiced_frames:
raise AssertionError(
f"output contained insufficient audible speech: minimum_voiced_frames={min_voiced_frames}, stats={stats}"
if int(frame_stats["voiced_frames"]) < min_voiced_frames:
raise _AudioAcceptanceError(
"voiced_frames",
f"output contained insufficient audible speech: minimum_voiced_frames={min_voiced_frames}, stats={stats}",
stats,
)
runtime_metadata = [
metadata["vllm_omni"]
for event in _events(client, "response.output_audio.delta")
for event in _audio_events(client)
if isinstance((metadata := event.get("metadata")), dict) and isinstance(metadata.get("vllm_omni"), dict)
]
if not any(
@@ -298,18 +378,19 @@ def _session_result(
and item.get("runner_kv_backed") is True
for item in runtime_metadata
):
raise AssertionError(f"audio events did not prove the scheduler data plane: {runtime_metadata}")
raise _AudioAcceptanceError(
"scheduler_data_plane", "audio events did not prove the scheduler data plane", stats
)
response_ids = _response_ids(client)
if len(response_ids) != 1:
raise AssertionError(f"continuous stream used multiple responses: {sorted(response_ids)}")
raise _AudioAcceptanceError(
"response_identity", f"continuous stream requires one response: count={len(response_ids)}", stats
)
return (
raw,
response_ids,
{
**stats,
"audio_chunks": chunks,
"audio_samples": len(raw) // 2,
"audio_rms": rms,
"response_ids": sorted(response_ids),
},
)
@@ -332,6 +413,8 @@ def _save(output_dir: Path, name: str, client: RawRealtimeProbe, raw: bytes) ->
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__":
tests/e2e/online_serving/test_personaplex_load_driver.py
+568
View File
@@ -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="<i2").tobytes())
return driver.parse_args(
[
"--model",
"fixture-only",
"--input-wav",
str(wav_path),
"--output-dir",
str(tmp_path / "out"),
"--sessions",
"2",
"--load-frames",
"2",
"--tail-s",
"0",
"--drain-s",
"0.04",
"--timeout-s",
"3",
"--cleanup-timeout-s",
"1",
"--min-voiced-frames",
"1",
"--minimum-audio-chunks",
"1",
"--max-frame-deficit",
"0",
*extra,
]
)
def _audio(response_id="r", samples=1920, event_type="response.output_audio.delta"):
return {
"type": event_type,
"response_id": response_id,
"delta": base64.b64encode(np.full(samples, 2000, dtype="<i2").tobytes()).decode("ascii"),
"sample_rate_hz": driver.SAMPLE_RATE_HZ,
"metadata": {
"vllm_omni": {
"runtime_impl": "scheduler_data_plane",
"uses_model_runner_scheduler": True,
"runner_kv_backed": True,
}
},
}
@asynccontextmanager
async def _server(mode="success"):
states: list[_ServerState] = []
async def handle(ws):
index = len(states)
state: _ServerState = {"index": index, "closed": False, "frames": 0, "close_requested": False}
states.append(state)
response_id = "shared" if mode == "shared_response" else f"r-{index}"
try:
async for raw in ws:
message = json.loads(raw)
if message["type"] == "session.update":
if mode == "reject_second" and index == 1:
await ws.send(json.dumps({"type": "error", "error": {"code": "resource_exhausted"}}))
continue
if mode == "missing_capabilities":
await ws.send(
json.dumps(
{
"type": "session.created",
"session": {"resume_token": "must-not-be-exported"},
}
)
)
continue
await ws.send(
json.dumps(
{
"type": "session.created",
"session": {
"resume_token": "must-not-be-exported",
"capabilities": {
"chunk_period_ms": 80,
"supports_multi_session_same_replica": True,
},
},
}
)
)
await ws.send(json.dumps({"type": "response.created", "response": {"id": response_id}}))
elif message["type"] == "input_audio_buffer.append":
state["frames"] += 1
if mode == "short_audio" and state["frames"] == 2:
continue
if mode == "disconnect":
await ws.close()
return
if mode == "bad_json":
await ws.send("{")
elif mode not in ("silent", "late_on_close"):
event_type = (
"response.audio.delta" if mode == "legacy_audio_event" else "response.output_audio.delta"
)
packet = _audio(response_id, event_type=event_type)
if mode == "private_metadata":
packet["metadata"] = {"vllm_omni": {"private": "remote-detail-must-not-be-exported"}}
if mode == "odd_packet":
raw_audio = base64.b64decode(packet["delta"])
raw_audio = raw_audio + b"\x00" if state["frames"] == 1 else raw_audio[:-1]
packet["delta"] = base64.b64encode(raw_audio).decode("ascii")
if state["frames"] == 2:
if mode == "missing_packet_rate":
packet.pop("sample_rate_hz")
elif mode == "string_packet_rate":
packet["sample_rate_hz"] = "24000"
elif mode == "invalid_base64":
packet["delta"] = "!" + packet["delta"]
elif mode == "orphan_audio":
orphan = _audio(response_id)
orphan.pop("response_id")
await ws.send(json.dumps(orphan))
await ws.send(json.dumps(packet))
elif message["type"] == "session.close":
state["close_requested"] = True
if mode == "late_on_close":
await ws.send(json.dumps(_audio(response_id, samples=driver.FRAME_SAMPLES * state["frames"])))
if mode == "error_on_close":
await ws.send(json.dumps({"type": "error", "error": {"code": "cleanup_failed"}}))
if mode != "no_close_ack":
await ws.send(json.dumps({"type": "session.closed"}))
finally:
state["closed"] = True
async with websockets.serve(handle, "127.0.0.1", 0) as server:
port = server.sockets[0].getsockname()[1]
yield f"ws://127.0.0.1:{port}/v1/realtime", states
def test_omitted_sessions_preserves_lifecycle_mode():
args = driver.parse_args(["--model", "unused", "--input-wav", "unused.wav"])
assert args.sessions is None
@pytest.mark.parametrize(
"options",
[
["--sessions", "0"],
["--sessions", "65"],
["--load-frames", "0"],
["--load-frames", "10001"],
["--drain-s", "nan"],
["--tail-s", "-1"],
["--timeout-s", "0"],
["--timeout-s", "inf"],
["--cleanup-timeout-s", "0"],
["--cleanup-timeout-s", "nan"],
["--max-client-rtf", "-1"],
["--max-frame-deficit", "-1"],
["--minimum-audio-chunks", "0"],
["--min-voiced-frames", "0"],
["--voiced-frame-rms-threshold", "nan"],
],
)
def test_load_rejects_invalid_configuration(tmp_path, options):
with pytest.raises(SystemExit):
_args(tmp_path, *options)
def test_metric_clock_and_packet_frame_distinction():
client = driver.RawRealtimeProbe("ws://unused")
client.events.add({"type": "session.created", "session": {"resume_token": "secret"}}, received_at_s=7)
for when in (10.4, 10.8, 11.6):
event = _audio(samples=driver.FRAME_SAMPLES * 5)
event["_client_received_at_s"] = -999.0
client.events.add(event, received_at_s=when)
report = driver._load_metrics(client, [(10, 10, 10.01)])
intervals = report["client_audio_packet_interval_ms"]
assert isinstance(intervals, dict)
assert intervals["count"] == 2
assert intervals["median"] == pytest.approx(600)
assert intervals["p99"] == pytest.approx(796)
assert intervals["max"] == pytest.approx(800)
assert intervals["argmax_index"] == 1
assert report["client_first_audio_after_stream_start_ms"] == pytest.approx(400)
assert report["output_samples"] == 15 * driver.FRAME_SAMPLES
assert report["audio_underrun_s"] == pytest.approx(0.4)
assert report["audio_continuity_ok"] is False
underruns = report["audio_underrun_event_count"]
assert isinstance(underruns, int) and underruns >= 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="<f4"),
epoch=10.0,
timeout_s=1.0,
sends=sends,
clock=clock.monotonic,
sleep=clock.sleep,
)
assert len(sends) == 3
assert sends[1][1] >= 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)