Detect click/pop discontinuities with bounded PCM evidence

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.

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:

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",
 ]

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("<h", pcm)]
+    strongest: tuple[int, float, int] | None = None
+    for frame_offset in range(len(samples) // channels):
+        for channel in range(channels):
+            current = samples[frame_offset * channels + channel]
+            if frame_offset:
+                previous = samples[(frame_offset - 1) * channels + channel]
+            elif previous_samples is not None:
+                previous = previous_samples[channel]
+            else:
+                continue
+            magnitude = abs(current - previous) / PCM16_FULL_SCALE
+            signal_level = max(abs(current), abs(previous)) / PCM16_FULL_SCALE
+            if (
+                magnitude >= 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("<h", pcm[-2 * channels :])
+    )
+
+
 def _delta_db(input_rms: float, output_rms: float) -> float:
     if input_rms <= 0.0 or output_rms <= 0.0:
         return 0.0

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("<h", value) * frame_count
 
 
+@pytest.mark.parametrize(
+    ("pcm", "previous", "expected"),
+    [
+        (struct.pack("<hhhh", 1000, 1100, 1200, 1300), None, None),
+        (struct.pack("<hhh", -100, 0, 100), None, None),
+        (struct.pack("<hh", 12000, -12000), None, (1, 24000 / 32768, 0)),
+        (_pcm(0, 4), (0,), None),
+        (struct.pack("<hh", 32767, -32768), None, (1, 65535 / 32768, 0)),
+    ],
+)
+def test_pcm_discontinuity_detector_contract(pcm, previous, expected) -> 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,