diff --git a/README.md b/README.md index 53648ef..6facc3a 100644 --- a/README.md +++ b/README.md @@ -616,6 +616,23 @@ Open `http://127.0.0.1:5173/`. The Web UI can inspect a run timeline, compare tw runs, play stage recordings, watch live run status, and start an async run from an example payload. +### Guided real-call reproduction + +Select an observed baseline run and open **Ask VoxBench** in the right rail. The +guided experiment keeps the main UI to three steps: baseline, intentional +barge-in, and evidence. Add the second run ID and select **Compare evidence**. +VoxBench validates that the runs are distinct observed AudioSocket calls, carry +the expected experiment-condition tags, last at least 20 seconds, use the same +provider/configuration, and contain RTCP evidence. It then contrasts barge-in +incidents and returns one recommended next action. + +Setup commands are available under the collapsed **Call setup and commands** +section. Starting Docker/Asterisk, supplying a Gemini API key, placing the +Telephone call, and speaking the controlled interruption remain explicit +operator actions; the diagnostic endpoint never receives or stores the key and +does not execute shell commands. This keeps the comparison reproducible without +giving the Web agent access to the host or credentials. + For the shortest product walkthrough, use **Diagnose a call in 3 seconds** at the top of the page: diff --git a/src/voxbench/control_plane/real_call_diagnostics.py b/src/voxbench/control_plane/real_call_diagnostics.py new file mode 100644 index 0000000..44847d5 --- /dev/null +++ b/src/voxbench/control_plane/real_call_diagnostics.py @@ -0,0 +1,291 @@ +"""Bounded diagnostics for the guided real-call reproduction experiment.""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class RealCallExperimentRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + primary_run_id: str = Field(min_length=1, max_length=128) + compare_run_id: str | None = Field(default=None, min_length=1, max_length=128) + + @field_validator("primary_run_id", "compare_run_id") + @classmethod + def normalize_run_id(cls, value: str | None) -> str | None: + if value is None: + return None + normalized = value.strip() + if not normalized or any( + marker in normalized.lower() for marker in ("http://", "https://") + ): + raise ValueError("run IDs must be safe identifiers") + return normalized + + +class ExperimentCriterion(BaseModel): + key: str + label: str + status: Literal["pass", "fail", "missing"] + detail: str + + +class ExperimentFinding(BaseModel): + classification: Literal["observed", "derived", "unknown", "recommended"] + title: str + detail: str + run_role: Literal["primary", "compare", "both"] + evidence_refs: list[str] = Field(default_factory=list) + + +class RunExperimentSnapshot(BaseModel): + run_id: str + role: Literal["no-interruption", "intentional-barge-in", "unknown"] + observed_call: bool + duration_ms: float + provider: str + config_hash: str + rtp_report_count: int + incident_count: int + barge_in_count: int + recording_stages: list[str] + + +class RealCallExperimentResponse(BaseModel): + status: Literal["needs-compare", "ready", "inconclusive"] + summary: str + primary: RunExperimentSnapshot + compare: RunExperimentSnapshot | None = None + criteria: list[ExperimentCriterion] + findings: list[ExperimentFinding] + next_actions: list[str] + + +def analyze_real_call_experiment( + primary: Any, + compare: Any | None, +) -> RealCallExperimentResponse: + primary_snapshot = _snapshot(primary) + if compare is None: + return RealCallExperimentResponse( + status="needs-compare", + summary=( + "Baseline run selected. Add the intentional barge-in run to compare the experiment." + ), + primary=primary_snapshot, + criteria=[ + _criterion( + "primary-role", + "Baseline condition", + primary_snapshot.role == "no-interruption", + f"Detected role: {primary_snapshot.role}", + ), + ExperimentCriterion( + key="compare-present", + label="Barge-in comparison", + status="missing", + detail="Select the intentional-barge-in run as Compare.", + ), + ], + findings=[], + next_actions=[ + "Complete one intentional-barge-in call, then select its run as Compare." + ], + ) + + compare_snapshot = _snapshot(compare) + same_provider = primary.provider == compare.provider + same_config = primary.config_hash == compare.config_hash + criteria = [ + _criterion( + "different-runs", + "Two distinct runs", + primary.run_id != compare.run_id, + "Primary and Compare must differ.", + ), + _criterion( + "primary-role", + "Primary is no-interruption", + primary_snapshot.role == "no-interruption", + f"Detected role: {primary_snapshot.role}", + ), + _criterion( + "compare-role", + "Compare is intentional-barge-in", + compare_snapshot.role == "intentional-barge-in", + f"Detected role: {compare_snapshot.role}", + ), + _criterion( + "observed-calls", + "Both are observed calls", + primary_snapshot.observed_call and compare_snapshot.observed_call, + "Synthetic runs cannot prove the real-call path.", + ), + _criterion( + "duration", + "Both calls are at least 20 seconds", + primary_snapshot.duration_ms >= 20_000 and compare_snapshot.duration_ms >= 20_000, + ( + f"Primary {primary_snapshot.duration_ms / 1000:.1f}s · " + f"Compare {compare_snapshot.duration_ms / 1000:.1f}s" + ), + ), + _criterion( + "provider", "Same provider", same_provider, f"{primary.provider} · {compare.provider}" + ), + _criterion( + "config", + "Same configuration", + same_config, + "Configuration hashes match." + if same_config + else "Configuration hashes differ; treat the comparison as confounded.", + ), + _criterion( + "rtcp", + "RTCP evidence on both runs", + primary_snapshot.rtp_report_count > 0 and compare_snapshot.rtp_report_count > 0, + ( + f"Primary {primary_snapshot.rtp_report_count} · " + f"Compare {compare_snapshot.rtp_report_count} reports" + ), + ), + _criterion( + "baseline-clear", + "No barge-in in baseline", + primary_snapshot.barge_in_count == 0, + f"Primary {primary_snapshot.barge_in_count} barge-in incident(s)", + ), + _criterion( + "compare-barge-in", + "Barge-in reproduced in Compare", + compare_snapshot.barge_in_count > 0, + f"Compare {compare_snapshot.barge_in_count} barge-in incident(s)", + ), + ] + findings = _findings(primary_snapshot, compare_snapshot) + required = { + "different-runs", + "primary-role", + "compare-role", + "observed-calls", + "duration", + "provider", + "rtcp", + "baseline-clear", + "compare-barge-in", + } + ready = all(item.status == "pass" for item in criteria if item.key in required) + next_actions: list[str] = [] + if compare_snapshot.barge_in_count == 0: + next_actions.append("Repeat the Compare call and interrupt once after the second sentence.") + if primary_snapshot.barge_in_count > 0: + next_actions.append( + "Check headset isolation and microphone sensitivity before repeating the baseline." + ) + if primary_snapshot.rtp_report_count == 0 or compare_snapshot.rtp_report_count == 0: + next_actions.append( + "Repeat with --collect-rtcp and keep each call active for 20-30 seconds." + ) + if not same_config: + next_actions.append( + "Repeat both calls with the same provider model and pipeline configuration." + ) + if not next_actions: + next_actions.append( + "Inspect the Compare barge-in incident and listen to each stage at the shared cursor." + ) + return RealCallExperimentResponse( + status="ready" if ready else "inconclusive", + summary=( + "The controlled real-call comparison has the required evidence." + if ready + else "The runs were compared, but one or more required controls are missing." + ), + primary=primary_snapshot, + compare=compare_snapshot, + criteria=criteria, + findings=findings, + next_actions=next_actions, + ) + + +def _snapshot(run: Any) -> RunExperimentSnapshot: + timeline = run.to_timeline() + tags = {tag.lower() for tag in run.environment.tags} + role: Literal["no-interruption", "intentional-barge-in", "unknown"] = "unknown" + if "intentional-barge-in" in tags or "experiment-intentional-barge-in" in tags: + role = "intentional-barge-in" + elif "no-interruption" in tags or "experiment-no-interruption" in tags: + role = "no-interruption" + incidents = timeline.lanes.incidents + barge_incidents = [item for item in incidents if item.rule_id == "barge_in_sequence"] + duration_ms = max( + [item.duration_ms for item in timeline.lanes.recordings] + + ( + [max(0.0, (run.ended_at - run.started_at).total_seconds() * 1000)] + if run.ended_at + else [0.0] + ) + ) + return RunExperimentSnapshot( + run_id=run.run_id, + role=role, + observed_call=( + "audiosocket" in tags + or (run.environment.started_from or "").startswith("voxbench-audiosocket-") + ), + duration_ms=duration_ms, + provider=run.provider, + config_hash=run.config_hash, + rtp_report_count=len(timeline.lanes.rtp_quality), + incident_count=len(incidents), + barge_in_count=len(barge_incidents), + recording_stages=[item.stage for item in timeline.lanes.recordings], + ) + + +def _criterion(key: str, label: str, passed: bool, detail: str) -> ExperimentCriterion: + return ExperimentCriterion( + key=key, label=label, status="pass" if passed else "fail", detail=detail + ) + + +def _findings( + primary: RunExperimentSnapshot, compare: RunExperimentSnapshot +) -> list[ExperimentFinding]: + findings = [ + ExperimentFinding( + classification="derived", + title="Barge-in contrast", + detail=( + f"Primary has {primary.barge_in_count} barge-in incident(s); " + f"Compare has {compare.barge_in_count}." + ), + run_role="both", + ), + ExperimentFinding( + classification="observed" + if primary.rtp_report_count and compare.rtp_report_count + else "unknown", + title="Transport coverage", + detail=( + f"RTCP reports: Primary {primary.rtp_report_count}, " + f"Compare {compare.rtp_report_count}." + ), + run_role="both", + ), + ] + if compare.barge_in_count > 0 and primary.barge_in_count == 0: + findings.append( + ExperimentFinding( + classification="derived", + title="Controlled interruption reproduced", + detail="Barge-in evidence appears only in the intentional interruption condition.", + run_role="both", + ) + ) + return findings diff --git a/src/voxbench/control_plane/run_api.py b/src/voxbench/control_plane/run_api.py index c2d1ffd..fd2482e 100644 --- a/src/voxbench/control_plane/run_api.py +++ b/src/voxbench/control_plane/run_api.py @@ -64,6 +64,11 @@ from voxbench.control_plane.models import ( Verification as VerificationRow, ) +from voxbench.control_plane.real_call_diagnostics import ( + RealCallExperimentRequest, + RealCallExperimentResponse, + analyze_real_call_experiment, +) from voxbench.control_plane.repository_config import ( RepositoryReadiness, memory_repository_readiness, @@ -4849,6 +4854,35 @@ async def live_websocket( except WebSocketDisconnect: return + @router.post( + "/diagnostics/real-call-experiment", + response_model=RealCallExperimentResponse, + ) + async def diagnose_real_call_experiment( + request: RealCallExperimentRequest, + api_state: RunApiStateDependency, + ) -> RealCallExperimentResponse: + primary = api_state.repository.get(request.primary_run_id) + if primary is None: + raise HTTPException( + status_code=404, + detail=f"unknown primary run '{request.primary_run_id}'", + ) + compare = None + if request.compare_run_id is not None: + if request.compare_run_id == request.primary_run_id: + raise HTTPException( + status_code=409, + detail="Primary and Compare must be different runs", + ) + compare = api_state.repository.get(request.compare_run_id) + if compare is None: + raise HTTPException( + status_code=404, + detail=f"unknown compare run '{request.compare_run_id}'", + ) + return analyze_real_call_experiment(primary, compare) + @router.get("/runs/{run_id}", response_model=RunResponse) async def get_run( run_id: str, diff --git a/tests/test_real_call_diagnostics.py b/tests/test_real_call_diagnostics.py new file mode 100644 index 0000000..fa4e06d --- /dev/null +++ b/tests/test_real_call_diagnostics.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +from fastapi.testclient import TestClient + +from voxbench.control_plane.app import create_app +from voxbench.control_plane.real_call_diagnostics import analyze_real_call_experiment + + +class FakeRun: + def __init__( + self, + *, + run_id: str, + condition: str, + barge_in_count: int, + rtp_report_count: int = 2, + duration_seconds: int = 25, + config_hash: str = "same-config", + ) -> None: + self.run_id = run_id + self.provider = "gemini-live" + self.config_hash = config_hash + self.started_at = datetime.now(UTC) + self.ended_at = self.started_at + timedelta(seconds=duration_seconds) + self.environment = SimpleNamespace( + tags=["live-demo", "audiosocket", f"experiment-{condition}"], + started_from="voxbench-audiosocket-realtime", + ) + self._timeline = SimpleNamespace( + lanes=SimpleNamespace( + recordings=[ + SimpleNamespace(stage="serializer", duration_ms=duration_seconds * 1000) + ], + rtp_quality=[SimpleNamespace() for _ in range(rtp_report_count)], + incidents=[ + SimpleNamespace(rule_id="barge_in_sequence") for _ in range(barge_in_count) + ], + ) + ) + + def to_timeline(self): + return self._timeline + + +def test_controlled_real_call_experiment_is_ready() -> None: + baseline = FakeRun(run_id="baseline", condition="no-interruption", barge_in_count=0) + interrupted = FakeRun( + run_id="interrupted", + condition="intentional-barge-in", + barge_in_count=1, + ) + + result = analyze_real_call_experiment(baseline, interrupted) + + assert result.status == "ready" + assert all( + criterion.status == "pass" for criterion in result.criteria if criterion.key != "config" + ) + assert any(finding.title == "Controlled interruption reproduced" for finding in result.findings) + + +def test_missing_compare_returns_one_next_step() -> None: + baseline = FakeRun(run_id="baseline", condition="no-interruption", barge_in_count=0) + + result = analyze_real_call_experiment(baseline, None) + + assert result.status == "needs-compare" + assert result.compare is None + assert result.next_actions == [ + "Complete one intentional-barge-in call, then select its run as Compare." + ] + + +def test_missing_rtcp_and_baseline_barge_in_are_inconclusive() -> None: + baseline = FakeRun( + run_id="baseline", + condition="no-interruption", + barge_in_count=1, + rtp_report_count=0, + ) + interrupted = FakeRun( + run_id="interrupted", + condition="intentional-barge-in", + barge_in_count=0, + rtp_report_count=0, + ) + + result = analyze_real_call_experiment(baseline, interrupted) + + assert result.status == "inconclusive" + assert any(item.key == "rtcp" and item.status == "fail" for item in result.criteria) + assert any(item.key == "compare-barge-in" and item.status == "fail" for item in result.criteria) + assert any("headset" in action.lower() for action in result.next_actions) + assert any("--collect-rtcp" in action for action in result.next_actions) + + +def test_real_call_diagnostic_endpoint_validates_run_scope(tmp_path: Path) -> None: + client = TestClient(create_app(artifact_root=tmp_path / "recordings")) + created = client.post( + "/runs/live-demo/simulated", + json={"provider": "gemini-live", "scenario": "clean", "duration_ms": 3000}, + ) + run_id = created.json()["run_id"] + + baseline_only = client.post( + "/diagnostics/real-call-experiment", + json={"primary_run_id": run_id}, + ) + duplicate = client.post( + "/diagnostics/real-call-experiment", + json={"primary_run_id": run_id, "compare_run_id": run_id}, + ) + + assert baseline_only.status_code == 200 + assert baseline_only.json()["status"] == "needs-compare" + assert duplicate.status_code == 409 diff --git a/web/src/App.tsx b/web/src/App.tsx index d24a5ca..dd036b4 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -27,6 +27,7 @@ import { } from 'react' import WaveSurfer from 'wavesurfer.js' +import { RealCallAgentPanel } from './RealCallAgentPanel' import type { CrossSessionTrend, EnvironmentProfile, @@ -973,6 +974,17 @@ export function App() {