From 09ba8b0782ef34c77e0b9f32a246185c1dd53ed4 Mon Sep 17 00:00:00 2001 From: engkimo Date: Thu, 6 Aug 2026 00:18:13 +0900 Subject: [PATCH] feat: automate matched real-call RTCP experiments --- README.md | 56 ++++++++++++---- docs/demo-live-softphone.md | 72 ++++++++++++++++----- scripts/asterisk-local | 11 ++++ src/voxbench/cli/main.py | 89 ++++++++++++++++++++++++++ src/voxbench/live_demo/observed_run.py | 24 ++++++- src/voxbench/telephony/audiosocket.py | 9 +++ tests/test_asterisk_local_example.py | 3 + tests/test_audiosocket.py | 66 +++++++++++++++++++ 8 files changed, 300 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 8ac6bb9..53648ef 100644 --- a/README.md +++ b/README.md @@ -244,28 +244,58 @@ speaker-to-microphone echo. The bridge records local queue disposal and provider chunk correlation, but `remote_playout_observed: false` means it cannot claim exactly what the caller heard without a caller-side recording. -### 7. Collect RTCP during the next call +### 7. Run the matched real-call experiment with RTCP An empty **RTP quality** panel does not prove the network was clean. It means no -normalized RTP/RTCP evidence reached that run. Start the collector in terminal 4 -while the real call is active, using the new run ID: +normalized RTP/RTCP evidence reached that run. For a matched experiment, run one +condition at a time. The bridge automatically binds the read-only AMI collector +to the run ID created for each call. + +First use a headset and let Gemini finish without speaking over it: + +```fish +./scripts/asterisk-local gemini \ + --collect-rtcp \ + --experiment-condition no-interruption +``` + +Call `7000`, keep the call active for 20–30 seconds, hang up, and stop the +bridge. Then repeat with one deliberate interruption at a repeatable point: + +```fish +./scripts/asterisk-local gemini \ + --collect-rtcp \ + --experiment-condition intentional-barge-in +``` + +The local helper supplies the loopback-only AMI account from the Asterisk +development configuration. When invoking the CLI directly, keep credentials in +environment variables: ```fish set -gx VOXBENCH_AMI_USERNAME voxbench-rtcp set -gx VOXBENCH_AMI_SECRET voxbench-ami-local-only -voxbench asterisk-ami-rtcp \ - --run-id '' \ +voxbench audiosocket-realtime \ + --provider gemini-live \ --control-plane-url http://127.0.0.1:8001 \ - --host 127.0.0.1 \ - --port 5038 \ - --clock-rate-hz 8000 + --collect-rtcp \ + --ami-host 127.0.0.1 \ + --ami-port 5038 \ + --ami-clock-rate-hz 8000 \ + --experiment-condition no-interruption ``` -Keep the call active for 20–30 seconds so Asterisk has time to emit RTCP -reports. Aggregate RTCP can show loss, jitter, and RTT; it cannot identify an -exact missing RTP sequence number. Use the library packet-observation adapter -when packet-level proof is required. +Keep only one test call active. AMI reports aggregate Asterisk RTCP events and +does not retain the Channel identifier, so concurrent calls cannot be safely +attributed to separate runs. Aggregate RTCP can show loss, jitter, and RTT; it +cannot identify an exact missing RTP sequence number. Use the library +packet-observation adapter when packet-level proof is required. + +For an audible-quality conclusion, retain a caller-side reference recording. +The simplest independent setup is a second device recording the headset output; +label it with the run ID printed by the bridge. VoxBench stage WAVs prove the +local processing path, but not what the remote softphone ultimately played. ## Local demo troubleshooting @@ -282,7 +312,7 @@ when packet-level proof is required. | **Fetch** or the circular refresh button appears to do nothing | Refreshing the same completed run does not create new evidence or change immutable recordings. Select a different run, keep an active call running, or create a new call. | | **Readiness** shows unchecked/incomplete items | Readiness is an evidence checklist, not an automatic failure count. A live bridge can complete while deployment-specific checklist fields remain unknown. | | Audio is choppy or clicks | Compare all four stage WAVs, click the barge-in incidents, repeat once with a headset and no overlap, then repeat with RTCP collection. Do not attribute the symptom to packet loss until transport evidence exists. | -| **RTP quality** is empty | Start `asterisk-ami-rtcp` during the active call and keep the call long enough for RTCP. No points means unobserved transport quality, not confirmed zero loss. | +| **RTP quality** is empty | Restart the Gemini bridge with `--collect-rtcp`, keep exactly one call active, and keep it connected long enough for RTCP. The standalone `asterisk-ami-rtcp` command remains available for manual collection. No points means unobserved transport quality, not confirmed zero loss. | Detailed operator and library integration references are available in [`docs/demo-live-softphone.md`](docs/demo-live-softphone.md) and diff --git a/docs/demo-live-softphone.md b/docs/demo-live-softphone.md index 61a6029..09c6da3 100644 --- a/docs/demo-live-softphone.md +++ b/docs/demo-live-softphone.md @@ -296,25 +296,65 @@ export VOXBENCH_AMI_USERNAME=voxbench-rtcp export VOXBENCH_AMI_SECRET='REPLACE_WITH_LOCAL_SECRET' ``` -Start the AudioSocket bridge and place a call. Copy the run ID printed by the -bridge, then start the collector in a third terminal while that run is active: +The recommended local workflow attaches the collector automatically when each +AudioSocket run is created: -```bash -voxbench asterisk-ami-rtcp \ - --run-id '' \ - --host 127.0.0.1 \ - --port 5038 \ - --clock-rate-hz 8000 +```fish +./scripts/asterisk-local gemini \ + --collect-rtcp \ + --experiment-condition no-interruption ``` -`--clock-rate-hz` is the RTP codec clock, not necessarily an audio output sample -rate. Use 8000 for PCMU/PCMA and configure the actual clock for other codecs. Stop -the collector when the call/run ends. The Web RTP quality panel shows direction, -jitter, loss, RTT, MOS when separately supplied, and each point's relative time. -Live preview also shows an RTP collector block: `connected` after AMI login, -`collecting` after the first normalized RTCP event, and `failed` after a safe -collector failure observation. Its event count is additive across collector -restarts for the same run. +The helper supplies the development AMI credentials through environment +variables. The bridge starts one collector after provider connection succeeds, +binds it to the new run ID, and cancels it before that run is completed. Collector +failure is reported as operational evidence but does not fail an otherwise valid +voice call. + +To invoke the bridge directly, set `VOXBENCH_AMI_USERNAME` and +`VOXBENCH_AMI_SECRET`, then pass `--collect-rtcp`, `--ami-host`, `--ami-port`, and +`--ami-clock-rate-hz`. The clock rate is the RTP codec clock, not necessarily an +audio output sample rate. Use 8000 for PCMU/PCMA and configure the actual clock +for other codecs. + +The standalone `voxbench asterisk-ami-rtcp --run-id ...` command remains +available for integrations that manage run and collector lifecycles separately. + +The Web RTP quality panel shows direction, jitter, loss, RTT, MOS when separately +supplied, and each point's relative time. Live preview also shows an RTP collector +block: `connected` after AMI login, `collecting` after the first normalized RTCP +event, and `failed` after a safe collector failure observation. Its event count +is additive across collector restarts for the same run. + +Keep only one call active while automatic collection is enabled. The privacy-safe +collector intentionally drops Asterisk Channel, address, and SSRC fields, so AMI +events from concurrent calls cannot be attributed safely. + +## Matched Choppy-Audio Experiment + +Use identical provider model, codec, route, gain, noise floor, caller phrase, and +call duration for both conditions. + +1. Use a headset to prevent speaker audio from re-entering the microphone. +2. Start `no-interruption` with automatic RTCP collection, call `7000`, speak one + fixed prompt, and remain silent until Gemini finishes. +3. Keep the call active for 20–30 seconds, hang up, and retain the printed run ID. +4. Restart the bridge with `--experiment-condition intentional-barge-in`. +5. Repeat the same prompt, but interrupt Gemini once at the same relative phrase. +6. Retain an independent caller-side recording for each call and name it with the + corresponding run ID. +7. Select the no-interruption run as **Primary** and the interruption run as + **Compare**. + +Confirm all of the following before attributing choppy audio: + +- **RTP quality** contains received or sent points rather than remaining empty. +- The experiment condition is present in the run environment tags/operator note. +- The same provider model appears on both runs. +- Stage WAVs identify the first local stage containing the audible defect. +- Barge-in evidence explains any queued audio disposal. +- The caller-side recording confirms that the candidate instant was actually + audible after Asterisk RTP and softphone playout. Relevant Asterisk references: diff --git a/scripts/asterisk-local b/scripts/asterisk-local index 6de31fe..9043521 100755 --- a/scripts/asterisk-local +++ b/scripts/asterisk-local @@ -133,6 +133,7 @@ case "$command" in shift gemini_model="${VOXBENCH_GEMINI_MODEL:-gemini-3.1-flash-live-preview}" previous_arg="" + collect_rtcp=0 gemini_args=() for arg in "$@"; do if [[ "$previous_arg" == "--model" ]]; then @@ -147,6 +148,10 @@ case "$command" in --model=*) gemini_model="${arg#--model=}" ;; + --collect-rtcp) + collect_rtcp=1 + gemini_args+=("$arg") + ;; *) gemini_args+=("$arg") ;; @@ -204,6 +209,12 @@ EOF echo "Starting Gemini Live bridge on 127.0.0.1:9019." echo "Model: $gemini_model" echo "Control Plane: $CONTROL_PLANE_URL" + if [[ "$collect_rtcp" == "1" ]]; then + export VOXBENCH_AMI_USERNAME="${VOXBENCH_AMI_USERNAME:-voxbench-rtcp}" + export VOXBENCH_AMI_SECRET="${VOXBENCH_AMI_SECRET:-$AMI_PASSWORD}" + echo "Asterisk RTCP auto-collection: enabled on 127.0.0.1:$AMI_PORT" + echo "Keep only one test call active so aggregate AMI events stay attributable." + fi echo "Call 7000 after the listening message appears. Press Ctrl+C to stop." exec "$ROOT/.venv/bin/voxbench" audiosocket-realtime \ --control-plane-url "$CONTROL_PLANE_URL" \ diff --git a/src/voxbench/cli/main.py b/src/voxbench/cli/main.py index 34fe74b..9ec5682 100644 --- a/src/voxbench/cli/main.py +++ b/src/voxbench/cli/main.py @@ -15,6 +15,7 @@ from voxbench.live_demo.observed_run import ( LiveDemoProvider, build_audiosocket_observed_run_payload, + normalize_experiment_condition, ) from voxbench.observability import HttpObservationTransport, VoxBenchObserver from voxbench.realtime_providers import ( @@ -163,6 +164,34 @@ def audiosocket_realtime( float, typer.Option("--connect-backoff-seconds", min=0.0, max=30.0), ] = 0.5, + experiment_condition: Annotated[ + str | None, + typer.Option( + "--experiment-condition", + help="Safe alias recorded on each run, such as no-interruption.", + ), + ] = None, + collect_rtcp: Annotated[ + bool, + typer.Option( + "--collect-rtcp", + help="Attach aggregate Asterisk AMI RTCP observations to each call run.", + ), + ] = False, + ami_host: Annotated[str, typer.Option("--ami-host")] = "127.0.0.1", + ami_port: Annotated[int, typer.Option("--ami-port", min=1, max=65535)] = 5038, + ami_clock_rate_hz: Annotated[ + int, + typer.Option("--ami-clock-rate-hz", min=1, max=384_000), + ] = 8_000, + ami_username_env: Annotated[ + str, + typer.Option("--ami-username-env"), + ] = "VOXBENCH_AMI_USERNAME", + ami_secret_env: Annotated[ + str, + typer.Option("--ami-secret-env"), + ] = "VOXBENCH_AMI_SECRET", ) -> None: """Bridge Asterisk AudioSocket PCM to a realtime AI provider.""" @@ -178,6 +207,30 @@ def audiosocket_realtime( detail = f"set {env_vars} and install the live extra: pip install -e '.[live]'" raise typer.BadParameter(detail) + if experiment_condition is not None: + try: + experiment_condition = normalize_experiment_condition(experiment_condition) + except ValueError as exc: + raise typer.BadParameter(str(exc)) from None + + ami_username: str | None = None + ami_secret: str | None = None + if collect_rtcp: + ami_username = os.environ.get(ami_username_env) + ami_secret = os.environ.get(ami_secret_env) + missing = [ + name + for name, value in ( + (ami_username_env, ami_username), + (ami_secret_env, ami_secret), + ) + if not value + ] + if missing: + raise typer.BadParameter( + f"set required environment variable(s): {', '.join(missing)}" + ) + transport = HttpObservationTransport(control_plane_url) async def create_session(call_uuid: UUID) -> RealtimeCallSession: @@ -190,6 +243,7 @@ async def create_session(call_uuid: UUID) -> RealtimeCallSession: noise_floor=noise_floor, mode="provider", model=selected_model, + experiment_condition=experiment_condition, ) run = await asyncio.to_thread(transport.start_run, payload) observer = VoxBenchObserver(run["run_id"], transport) @@ -245,6 +299,40 @@ def report_retry(failed_attempt: int, delay: float) -> None: f"AudioSocket call {call_id} -> {provider}/{selected_model} " f"-> run {run['run_id']}" ) + if experiment_condition is not None: + typer.echo(f"Experiment condition: {experiment_condition}") + background_tasks: tuple[asyncio.Task[object], ...] = () + if collect_rtcp: + assert ami_username is not None + assert ami_secret is not None + rtcp_observer = VoxBenchObserver(run["run_id"], transport) + collector = AmiRtcpCollector( + host=ami_host, + port=ami_port, + username=ami_username, + secret=ami_secret, + clock_rate_hz=ami_clock_rate_hz, + ) + + async def collect_call_rtcp() -> None: + try: + await collector.collect(rtcp_observer) + except asyncio.CancelledError: + raise + except AmiError as exc: + rtcp_observer.observe_metric("asterisk_ami_rtcp_failures", 1.0) + with suppress(Exception): + await asyncio.to_thread(rtcp_observer.flush) + typer.echo( + f"Asterisk RTCP collection failed for run {run['run_id']}: {exc}", + err=True, + ) + + background_tasks = (asyncio.create_task(collect_call_rtcp()),) + typer.echo( + f"Asterisk RTCP collection attached to run {run['run_id']} " + f"on {ami_host}:{ami_port}" + ) return RealtimeCallSession( call_id=call_id, observer=observer, @@ -257,6 +345,7 @@ def report_retry(failed_attempt: int, delay: float) -> None: target_rms=target_rms, max_gain=max_gain, noise_floor=noise_floor, + background_tasks=background_tasks, ) server = AudioSocketRealtimeServer( diff --git a/src/voxbench/live_demo/observed_run.py b/src/voxbench/live_demo/observed_run.py index a1b1e32..8ce16d7 100644 --- a/src/voxbench/live_demo/observed_run.py +++ b/src/voxbench/live_demo/observed_run.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re from copy import deepcopy from pathlib import Path from typing import Any, Literal @@ -18,6 +19,19 @@ "manifests/processor/limiter.json", "manifests/processor/serializer.json", ) +EXPERIMENT_CONDITION_PATTERN = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}") + + +def normalize_experiment_condition(value: str) -> str: + """Return a safe, stable alias for a matched live-call condition.""" + + normalized = value.strip().lower() + if EXPERIMENT_CONDITION_PATTERN.fullmatch(normalized) is None: + raise ValueError( + "experiment_condition must be a 1-64 character lowercase alias using " + "letters, digits, dot, underscore, or dash" + ) + return normalized def build_audiosocket_observed_run_payload( @@ -29,6 +43,7 @@ def build_audiosocket_observed_run_payload( noise_floor: float, mode: AudioSocketDemoMode = "loopback", model: str | None = None, + experiment_condition: str | None = None, ) -> dict[str, Any]: config = deepcopy(_load_json(f"configs/live-demo-{provider}.json")) selected_model = model or str(config["spec"]["ai"]["model"]) @@ -57,6 +72,13 @@ def build_audiosocket_observed_run_payload( else "Bidirectional AudioSocket media connected to the selected provider." ) provider_note = f"{provider_note} Provider model: {selected_model}." + tags = ["live-demo", "audiosocket", mode, provider] + if experiment_condition is not None: + experiment_condition = normalize_experiment_condition(experiment_condition) + provider_note = ( + f"{provider_note} Experiment condition: {experiment_condition}." + ) + tags.append(f"experiment-{experiment_condition}") return { "config_name": config["meta"]["name"], "configs": [config], @@ -71,7 +93,7 @@ def build_audiosocket_observed_run_payload( "integration_target_alias": f"{provider}:{selected_model}:{mode}", "started_from": f"voxbench-audiosocket-{mode}", "operator_note": provider_note, - "tags": ["live-demo", "audiosocket", mode, provider], + "tags": tags, "secret_ref_names": ( ["OPENAI_API_KEY"] if provider == "openai-realtime" else ["GOOGLE_API_KEY"] ) diff --git a/src/voxbench/telephony/audiosocket.py b/src/voxbench/telephony/audiosocket.py index 11e2654..912e017 100644 --- a/src/voxbench/telephony/audiosocket.py +++ b/src/voxbench/telephony/audiosocket.py @@ -379,6 +379,10 @@ class RealtimeCallSession: limiter_ceiling: float = 0.7 telephony_rate: int = 8_000 flush_interval_seconds: float = 0.1 + background_tasks: tuple[asyncio.Task[object], ...] = field( + default_factory=tuple, + repr=False, + ) _closed: bool = False _input_resampler: Pcm16MonoStreamResampler | None = None _output_resampler: Pcm16MonoStreamResampler | None = None @@ -1254,6 +1258,11 @@ async def close(self, failure_alias: str | None = None) -> None: self.mark_output_ended(stop_reason="call_closed") if self._flush_task is not None: await self._flush_task + for task in self.background_tasks: + if not task.done(): + task.cancel() + if self.background_tasks: + await asyncio.gather(*self.background_tasks, return_exceptions=True) await asyncio.to_thread(self.observer.flush) if failure_alias is not None and self.fail_run is not None: await asyncio.to_thread(self.fail_run, failure_alias) diff --git a/tests/test_asterisk_local_example.py b/tests/test_asterisk_local_example.py index e8e8e8d..3a529ea 100644 --- a/tests/test_asterisk_local_example.py +++ b/tests/test_asterisk_local_example.py @@ -67,3 +67,6 @@ def test_local_launcher_distinguishes_gemini_from_loopback() -> None: assert '[[ -z "${GEMINI_API_KEY:-}" ]]' in launcher assert "audiosocket-realtime" in launcher assert "--provider gemini-live" in launcher + assert "--collect-rtcp" in launcher + assert 'VOXBENCH_AMI_USERNAME:-voxbench-rtcp' in launcher + assert 'VOXBENCH_AMI_SECRET:-$AMI_PASSWORD' in launcher diff --git a/tests/test_audiosocket.py b/tests/test_audiosocket.py index 8184aee..fb3409e 100644 --- a/tests/test_audiosocket.py +++ b/tests/test_audiosocket.py @@ -214,6 +214,36 @@ def test_audiosocket_observed_payload_resolves(tmp_path: Path) -> None: assert response.json()["status"] == "running" +def test_audiosocket_observed_payload_records_experiment_condition() -> None: + payload = build_audiosocket_observed_run_payload( + provider="gemini-live", + call_id=str(uuid4()), + target_rms=3000.0, + max_gain=8.0, + noise_floor=200.0, + mode="provider", + experiment_condition=" Intentional-Barge-In ", + ) + + environment = payload["environment"] + assert "experiment-intentional-barge-in" in environment["tags"] + assert environment["operator_note"].endswith( + "Experiment condition: intentional-barge-in." + ) + + +def test_audiosocket_observed_payload_rejects_unsafe_experiment_condition() -> None: + with pytest.raises(ValueError, match="experiment_condition"): + build_audiosocket_observed_run_payload( + provider="gemini-live", + call_id=str(uuid4()), + target_rms=3000.0, + max_gain=8.0, + noise_floor=200.0, + experiment_condition="https://example.invalid/call", + ) + + def test_realtime_call_session_resamples_provider_audio_and_observes_stages() -> None: async def scenario(): transport = CapturingTransport() @@ -256,6 +286,42 @@ def complete() -> None: assert any(metric.name == "provider_input_rms" for metric in batch.metrics) +def test_realtime_call_session_stops_background_tasks_before_completion() -> None: + async def scenario() -> tuple[bool, bool]: + transport = CapturingTransport() + provider = FakeProviderSession() + background_started = asyncio.Event() + background_stopped = False + completed_after_stop = False + + async def background() -> None: + nonlocal background_stopped + background_started.set() + try: + await asyncio.Event().wait() + finally: + background_stopped = True + + task = asyncio.create_task(background()) + await background_started.wait() + + def complete() -> None: + nonlocal completed_after_stop + completed_after_stop = background_stopped + + session = RealtimeCallSession( + call_id="call-with-collector", + observer=VoxBenchObserver("run-1", transport), + provider_session=provider, + complete_run=complete, + background_tasks=(task,), + ) + await session.close() + return background_stopped, completed_after_stop + + assert asyncio.run(scenario()) == (True, True) + + def test_realtime_call_session_drops_buffered_audio_on_barge_in() -> None: async def scenario(): transport = CapturingTransport()