From e31ac897ddac9b8396453b5e1eb70c5e0d8b6b07 Mon Sep 17 00:00:00 2001 From: 9O6 <3710814+9o6@users.noreply.github.com> Date: Mon, 31 Aug 2026 12:00:20 +0900 Subject: [PATCH] feat: detect PCM click and pop discontinuities --- docs/library-integration.md | 19 ++++ src/voxbench/control_plane/run_api.py | 69 +++++++++++++++ src/voxbench/observability/__init__.py | 2 + src/voxbench/observability/observer.py | 118 +++++++++++++++++++++++++ tests/test_observability.py | 87 ++++++++++++++++++ 5 files changed, 295 insertions(+) diff --git a/docs/library-integration.md b/docs/library-integration.md index 89cf447..c9f8445 100644 --- a/docs/library-integration.md +++ b/docs/library-integration.md @@ -253,3 +253,22 @@ The Control Plane endpoints used by the library are: During the run, the Web UI can fetch the normal run timeline and stage recording audio endpoints. Completing the run computes the existing VoxBench verifications. Failure aliases must not contain URLs, raw provider responses, SIP/SDP, or secrets. + +## PCM click/pop observation contract + +`observe_stage_audio` applies the deterministic `pcm16_adjacent_delta_v1` +detector to each stage's output. Its input contract is interleaved signed +16-bit little-endian PCM with the declared sample rate and channel count. It +compares adjacent samples independently per channel, including chunk boundaries, +and reports the strongest delta when it is at least 0.25 of PCM full scale and +either endpoint is at least 0.02 of full scale. Transitions where both endpoints +are below that minimum are treated as silence and ignored. Incidents less than +5 ms apart in the same stage are deduplicated. + +The observation contains only bounded metadata: stage, channel, cumulative +stage media time, normalized magnitude, and the detector contract. Media time is +not derived from the event's wall-clock timestamp. Queue-clear, interruption, or +barge-in events within 100 ms may be referenced as correlated evidence, but the +detector does not retain extra PCM, prove subjective audibility, identify the +cause, or prove remote playout. `remote_playout_observed` therefore remains +`false` unless a separate adapter supplies that evidence. diff --git a/src/voxbench/control_plane/run_api.py b/src/voxbench/control_plane/run_api.py index c2d1ffd..8c1c6d8 100644 --- a/src/voxbench/control_plane/run_api.py +++ b/src/voxbench/control_plane/run_api.py @@ -1157,6 +1157,7 @@ def rtp_collector_status(self) -> RtpCollectorStatusResponse: STAGE_SILENCE_SAMPLE_THRESHOLD_PCT = 98.0 STAGE_SILENCE_MIN_WINDOW_MS = 200.0 STAGE_SILENCE_MAX_OBSERVATION_GAP_MS = 100.0 +PCM_DISCONTINUITY_CORRELATION_WINDOW_MS = 100.0 ASSISTANT_OUTPUT_DEAD_AIR_MIN_OVERLAP_MS = 200.0 ASSISTANT_PLAYBACK_UNDERRUN_MIN_GAP_MS = 200.0 @@ -2904,6 +2905,74 @@ def _typed_timeline_incidents( item.stage: item for item in _stage_signal_evidence(run) } incidents: list[TimelineIncident] = [] + correlation_names = { + "playback_queue_cleared", + "barge_in_completed", + "provider_interrupt_requested", + "provider_interrupted", + } + for event in run.timeline_events: + if event.name != "stage.pcm_discontinuity_detected": + continue + nearby = sorted( + ( + candidate + for candidate in run.timeline_events + if candidate.name in correlation_names + and abs((candidate.ts - event.ts).total_seconds() * 1000.0) + <= PCM_DISCONTINUITY_CORRELATION_WINDOW_MS + ), + key=lambda candidate: ( + abs((candidate.ts - event.ts).total_seconds()), + candidate.event_id, + ), + ) + magnitude = _numeric_attribute(event.attributes, "magnitude") or 0.0 + media_time_ms = ( + _numeric_attribute(event.attributes, "media_time_ms") or 0.0 + ) + wall_time_ms = max( + 0.0, + _relative_seconds(event.ts, run.started_at) * 1000.0, + ) + evidence_refs = [] + artifact_id = artifact_by_stage.get(event.stage) + if artifact_id is not None: + evidence_refs.append(artifact_id) + evidence_refs.extend( + [event.event_id, *(candidate.event_id for candidate in nearby)] + ) + incidents.append( + TimelineIncident( + incident_id=f"pcm-discontinuity:{event.event_id}", + rule_id="pcm16_adjacent_delta_v1", + category="pipeline", + severity="warning", + title=( + f"PCM discontinuity suspected at " + f"{event.stage or 'unknown stage'}" + ), + summary=( + f"Adjacent PCM samples differed by {magnitude:.3g} full " + f"scale at media time {media_time_ms:.3f} ms" + ), + start_ms=wall_time_ms, + end_ms=wall_time_ms, + confidence="medium" if nearby else "low", + stage=event.stage, + observed={ + **event.attributes, + "temporally_correlated_event_count": len(nearby), + "remote_playout_observed": False, + }, + expected={ + "magnitude_below": event.attributes.get("threshold", 0.25), + "audibility": "not proven by this detector", + "remote_playout": "not observed", + }, + evidence_refs=evidence_refs, + ) + ) failure_index = 0 for verification_index, verification in enumerate(run.verifications): if verification.passed: diff --git a/src/voxbench/observability/__init__.py b/src/voxbench/observability/__init__.py index 09a3341..229d6d2 100644 --- a/src/voxbench/observability/__init__.py +++ b/src/voxbench/observability/__init__.py @@ -15,6 +15,7 @@ TimelineCategory, TimelineEvent, VoxBenchObserver, + detect_pcm_s16le_discontinuity, rtp_packet_from_datagram, ) @@ -33,5 +34,6 @@ "TimelineCategory", "TimelineEvent", "VoxBenchObserver", + "detect_pcm_s16le_discontinuity", "rtp_packet_from_datagram", ] diff --git a/src/voxbench/observability/observer.py b/src/voxbench/observability/observer.py index 1c66d4d..4d7bf03 100644 --- a/src/voxbench/observability/observer.py +++ b/src/voxbench/observability/observer.py @@ -21,6 +21,10 @@ PCM16_SILENCE_AMPLITUDE = round( PCM16_FULL_SCALE * 10 ** (PCM16_SILENCE_THRESHOLD_DBFS / 20.0) ) +PCM_DISCONTINUITY_CONTRACT_VERSION = "pcm16_adjacent_delta_v1" +PCM_DISCONTINUITY_THRESHOLD = 0.25 +PCM_DISCONTINUITY_MIN_SIGNAL_LEVEL = 0.02 +PCM_DISCONTINUITY_DEDUP_MS = 5.0 TimelineCategory = Literal[ "conversation", "signaling", @@ -476,6 +480,10 @@ def __init__(self, run_id: str, transport: ObservationTransport) -> None: self._timeline_events: list[TimelineEvent] = [] self._rtp_packet_ordinal = 0 self._rtp_capture_health_ordinal = 0 + self._discontinuity_ordinal = 0 + self._stage_pcm_state: dict[ + str, tuple[int, int, tuple[int, ...], int, float | None] + ] = {} self._lock = Lock() def observe_stage_audio( @@ -507,6 +515,15 @@ def observe_stage_audio( sample_count = len(output_pcm_s16le) // 2 frame_count = sample_count / channels chunk_duration_ms = frame_count / sample_rate_hz * 1000.0 + prior = self._stage_pcm_state.get(stage) + same_format = ( + prior is not None and prior[:2] == (sample_rate_hz, channels) + ) + discontinuity = detect_pcm_s16le_discontinuity( + output_pcm_s16le, + channels=channels, + previous_samples=prior[2] if same_format and prior[2] else None, + ) metrics = [ MetricPoint(stage=stage, name="input_rms", value=input_rms, ts=observed_at), MetricPoint(stage=stage, name="output_rms", value=output_rms, ts=observed_at), @@ -552,6 +569,64 @@ def observe_stage_audio( ) with self._lock: self._metrics.extend(metrics) + media_frame_offset = prior[3] if same_format else 0 + last_incident_ms = prior[4] if same_format else None + if discontinuity is not None: + media_time_ms = ( + media_frame_offset + discontinuity[0] + ) / sample_rate_hz * 1000.0 + if ( + last_incident_ms is None + or media_time_ms - last_incident_ms >= PCM_DISCONTINUITY_DEDUP_MS + ): + event_ordinal = self._discontinuity_ordinal + self._discontinuity_ordinal += 1 + self._timeline_events.append( + TimelineEvent( + event_id=f"pcm-discontinuity:{event_ordinal}", + category="pipeline", + name="stage.pcm_discontinuity_detected", + source="pcm_adjacent_delta_detector", + stage=stage, + correlation_alias=( + f"pcm-discontinuity:{stage}:{event_ordinal}" + ), + attributes={ + "media_time_ms": media_time_ms, + "magnitude": discontinuity[1], + "channel": discontinuity[2], + "detector_contract": ( + PCM_DISCONTINUITY_CONTRACT_VERSION + ), + "sample_format": "pcm_s16le", + "sample_rate_hz": sample_rate_hz, + "channels": channels, + "threshold": PCM_DISCONTINUITY_THRESHOLD, + "minimum_signal_level": ( + PCM_DISCONTINUITY_MIN_SIGNAL_LEVEL + ), + "silence_treatment": ( + "ignore_when_both_samples_below_minimum" + ), + "dedup_window_ms": PCM_DISCONTINUITY_DEDUP_MS, + "remote_playout_observed": False, + }, + ts=observed_at, + ) + ) + last_incident_ms = media_time_ms + final_samples = ( + _pcm_s16le_final_frame(output_pcm_s16le, channels) + if output_pcm_s16le + else (prior[2] if same_format else ()) + ) + self._stage_pcm_state[stage] = ( + sample_rate_hz, + channels, + final_samples, + media_frame_offset + int(frame_count), + last_incident_ms, + ) if record_output and output_pcm_s16le: self._audio_chunks.append( AudioChunk( @@ -873,6 +948,49 @@ def pcm_s16le_quality(pcm: bytes) -> tuple[float, float, float]: ) +def detect_pcm_s16le_discontinuity( + pcm: bytes, + *, + channels: int = 1, + previous_samples: tuple[int, ...] | None = None, +) -> tuple[int, float, int] | None: + """Return strongest suspicious delta as frame offset, magnitude, channel.""" + + if channels <= 0: + raise ValueError("channels must be positive") + if len(pcm) % (2 * channels): + raise ValueError("PCM16LE must contain complete channel frames") + if previous_samples is not None and len(previous_samples) != channels: + raise ValueError("previous_samples must contain one value per channel") + samples = [value for (value,) in struct.iter_unpack("= PCM_DISCONTINUITY_THRESHOLD + and signal_level >= PCM_DISCONTINUITY_MIN_SIGNAL_LEVEL + and (strongest is None or magnitude > strongest[1]) + ): + strongest = (frame_offset, magnitude, channel) + return strongest + + +def _pcm_s16le_final_frame(pcm: bytes, channels: int) -> tuple[int, ...]: + return tuple( + value[0] + for value in struct.iter_unpack(" float: if input_rms <= 0.0 or output_rms <= 0.0: return 0.0 diff --git a/tests/test_observability.py b/tests/test_observability.py index 01038d6..ef30852 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -19,6 +19,7 @@ SipEvent, TimelineEvent, VoxBenchObserver, + detect_pcm_s16le_discontinuity, rtp_packet_from_datagram, ) @@ -84,6 +85,92 @@ def _pcm(value: int, frame_count: int = 160) -> bytes: return struct.pack(" None: + observed = detect_pcm_s16le_discontinuity( + pcm, + previous_samples=previous, + ) + if expected is None: + assert observed is None + else: + assert observed is not None + assert observed[0] == expected[0] + assert observed[1] == pytest.approx(expected[1]) + assert observed[2] == expected[2] + + +def test_pcm_discontinuity_observation_uses_media_time_and_deduplicates() -> None: + transport = RecordingTransport() + observer = VoxBenchObserver("run", transport) + t0 = datetime(2026, 1, 1) + for index, value in enumerate((12000, -12000)): + observer.observe_stage_audio( + stage="serializer", + input_pcm_s16le=_pcm(value, 40), + output_pcm_s16le=_pcm(value, 40), + sample_rate_hz=8_000, + record_output=False, + ts=t0 + timedelta(seconds=index * 5), + ) + observer.flush() + + events = transport.batches[0].timeline_events + assert len(events) == 1 + assert events[0].attributes["media_time_ms"] == 5.0 + assert events[0].attributes["magnitude"] == pytest.approx(24000 / 32768) + assert events[0].attributes["remote_playout_observed"] is False + assert events[0].ts == t0 + timedelta(seconds=5) + + +def test_pcm_discontinuity_projects_bounded_incident(tmp_path: Path) -> None: + client = TestClient(create_app(artifact_root=tmp_path / "recordings")) + run_id = client.post( + "/runs/observed", + json=_observed_run_payload(), + ).json()["run_id"] + timeline = client.get(f"/runs/{run_id}/timeline").json() + t0 = datetime.fromisoformat(timeline["t0"]) + observer = VoxBenchObserver(run_id, ApiTestTransport(client)) + for index, value in enumerate((12000, -12000)): + observer.observe_stage_audio( + stage="serializer", + input_pcm_s16le=_pcm(value, 40), + output_pcm_s16le=_pcm(value, 40), + sample_rate_hz=8_000, + ts=t0 + timedelta(milliseconds=index * 5), + ) + observer.flush() + + lanes = client.get(f"/runs/{run_id}/timeline").json()["lanes"] + incident = next( + item + for item in lanes["incidents"] + if item["rule_id"] == "pcm16_adjacent_delta_v1" + ) + assert incident["stage"] == "serializer" + assert incident["confidence"] == "low" + assert incident["observed"]["media_time_ms"] == 5.0 + assert incident["observed"]["magnitude"] == pytest.approx(24000 / 32768) + assert incident["observed"]["remote_playout_observed"] is False + assert incident["expected"]["audibility"] == ( + "not proven by this detector" + ) + assert incident["evidence_refs"] == [ + "recording:0", + "pcm-discontinuity:0", + ] + + def _rtp_datagram( sequence_number: int, rtp_timestamp: int,