Skip to content

Commit 332f52d

Browse files
feat(observability): count live unnamed speaker ordinals per session (#23556)
#23533 (85f080f) defers a live "Speaker N" ordinal until an unnamed voice holds five seconds of non-overlapping placed speech, but nothing measured whether per-session ordinal churn actually fell after the listen roll - David's consecutive Speaker 4-8 lines had no metric. Add omi_live_speaker_ordinals_per_session, a histogram observed once per listen session at teardown. LiveSpeakerOrdinalTracker rides the same final segment stream the collapse monitor sees (routers/listen/transcripts _speaker_detection, after scope stamping and speaker-id assignment) and applies the display deferral's own qualification: a voice key (speaker_id_scope, speaker_id) mints one ordinal at MIN_EVIDENCE_SECONDS of unioned placed speech, and a voice the session named (owner or person, per-segment or in the matcher's final map) never mints one. The companion omi_live_speaker_scope_rotations_total counts scope/epoch rotations over the same sessions, so ordinal churn is separable from real multi-speaker conversations. Both metrics are label-free; the tracker is bounded (128 voices, 512 intervals per voice), never feeds speaker decisions, and its recording never raises into teardown. The pipeline doc's live-speaker-epochs section names the PromQL to watch after listen rolls. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent cbf2224 commit 332f52d

5 files changed

Lines changed: 350 additions & 0 deletions

File tree

‎backend/docs/listen_pusher_pipeline.mdx‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6876,3 +6876,15 @@ Unplaced segments still bypass matching. Their capture-placement repair belongs
68766876
to the live timeline lane; this change never invents offsets to obtain evidence.
68776877
No production configuration flip is required or performed. Backend/listen images
68786878
and a Flutter build are required to deliver the corresponding safe changes.
6879+
6880+
Per-session ordinal churn is exported where the deferral is enforced:
6881+
`omi_live_speaker_ordinals_per_session` (histogram, observed once per listen
6882+
session at teardown by `utils/observability/live_speaker_ordinals.py`) counts the
6883+
distinct unnamed scoped voices that cleared the five-second placed-speech floor,
6884+
with voices the session named excluded; `omi_live_speaker_scope_rotations_total`
6885+
counts scope/epoch rotations over the same sessions. **Watch after listen rolls**:
6886+
`histogram_quantile(0.95, rate(omi_live_speaker_ordinals_per_session_bucket[1h]))`
6887+
and its mean (`_sum/_count`) should fall while rotations per session
6888+
(`rate(omi_live_speaker_scope_rotations_total[1h]) / rate(omi_listen_ws_close_total[1h])`)
6889+
holds — ordinals falling with rotations flat means the deferral is working, not
6890+
that conversations gained fewer speakers.

‎backend/routers/listen/runtime.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
from utils.observability.journeys import ClientJourneyAttempt
6565
from utils.observability.listen_close import CLOSE_FRAME_DRAIN_SECONDS, note_ws_close, record_listen_ws_close
6666
from utils.observability.listen_loop import ensure_loop_lag_probe
67+
from utils.observability.live_speaker_ordinals import LiveSpeakerOrdinalTracker, record_live_speaker_ordinals
6768
from utils.observability.routing_cohort import RoutingCohort, current_routing_cohort
6869
from utils.observability.transcription import (
6970
LiveSTTAttempt,
@@ -223,6 +224,7 @@ def __init__(self, request: ListenRequest):
223224
self.transcripts: Any = None
224225
self.conversations: Any = None
225226
self.parity_capture = ListenParityCapture(None)
227+
self.speaker_ordinal_tracker = LiveSpeakerOrdinalTracker()
226228

227229
def _build_components(self) -> None:
228230
channels = build_channel_config(self.request.source or 'phone_call') if self.is_multi_channel else []
@@ -1447,6 +1449,9 @@ async def _teardown_components(self) -> None:
14471449
await self.language_observations.summarize(None if owner_persistence_blocked else self.request.uid)
14481450
if not owner_persistence_blocked:
14491451
await self.task_supervisor.drain_all(timeout=5.0, cancel=True)
1452+
# After the transcript loop drained and before the matcher's maps clear:
1453+
# the ordinal ledger needs both the final segments and final naming.
1454+
record_live_speaker_ordinals(getattr(self, 'speaker_ordinal_tracker', None), self.speakers)
14501455
self.receiver.clear()
14511456
self.transcripts.clear()
14521457
self.speakers.clear()

‎backend/routers/listen/transcripts.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1486,11 +1486,14 @@ async def _speaker_detection(
14861486
"""
14871487
speaker = self.host.speakers
14881488
observer = getattr(speaker, 'observe_segment', None)
1489+
ordinal_tracker = getattr(self.host, 'speaker_ordinal_tracker', None)
14891490
if observer is not None:
14901491
observed = queue_from_raw if queue_from_raw is not None else [s.model_dump() for s in segments]
14911492
for raw in observed:
14921493
if raw.get('speaker_id') is not None:
14931494
observer(raw['speaker_id'], raw.get('speaker_id_scope') or '', raw.get('id') or '')
1495+
if ordinal_tracker is not None:
1496+
ordinal_tracker.observe(raw)
14941497
for segment in segments:
14951498
segment_id = cast(str, segment.id)
14961499
if queue_from_raw is None:
Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
"""Live speaker ordinal churn telemetry mirrors the display deferral, once per session.
2+
3+
The tracker must reproduce #23533's qualification exactly enough to be a churn
4+
signal: five seconds of non-overlapping placed speech on an unnamed scoped voice
5+
mint one ordinal, named voices mint none, and scope rotations are counted
6+
separately so churn is readable against provider-epoch and reconnect rotations.
7+
"""
8+
9+
from types import SimpleNamespace
10+
11+
from models.transcript_segment import TranscriptSegment
12+
from routers.listen.transcripts import TranscriptProcessor
13+
from utils.observability.live_speaker_ordinals import (
14+
MAX_TRACKED_VOICES,
15+
OMI_LIVE_SPEAKER_ORDINALS_PER_SESSION,
16+
OMI_LIVE_SPEAKER_SCOPE_ROTATIONS_TOTAL,
17+
LiveSpeakerOrdinalTracker,
18+
record_live_speaker_ordinals,
19+
)
20+
21+
22+
def voice(sid, speaker, scope, start, end, *, is_user=False, person_id=None):
23+
return dict(
24+
id=sid,
25+
text='Testing one two',
26+
speaker_id=speaker,
27+
speaker_id_scope=scope,
28+
start=start,
29+
end=end,
30+
is_user=is_user,
31+
person_id=person_id,
32+
)
33+
34+
35+
def test_deferral_floor_mints_no_ordinal_below_five_placed_seconds():
36+
tracker = LiveSpeakerOrdinalTracker()
37+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 4.9))
38+
assert tracker.close() == (0, 0)
39+
40+
41+
def test_five_seconds_of_one_voice_mints_exactly_one_ordinal():
42+
tracker = LiveSpeakerOrdinalTracker()
43+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 2.0))
44+
tracker.observe(voice('s2', 0, 'scope:0', 3.0, 6.0))
45+
assert tracker.close() == (1, 0)
46+
47+
48+
def test_overlapping_and_replayed_spans_union_instead_of_double_counting():
49+
tracker = LiveSpeakerOrdinalTracker()
50+
# 4s + 4s overlapping in a 1s window is a 7s union: one qualified voice,
51+
# and a replay of the first span adds nothing.
52+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 4.0))
53+
tracker.observe(voice('s1', 0, 'scope:0', 3.0, 7.0))
54+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 4.0))
55+
assert tracker.close() == (1, 0)
56+
57+
# Adjacent bursts that never reach the floor across two voices mint nothing.
58+
adjacent = LiveSpeakerOrdinalTracker()
59+
adjacent.observe(voice('a1', 0, 'scope:0', 0.0, 2.5))
60+
adjacent.observe(voice('a2', 1, 'scope:0', 2.5, 5.0))
61+
assert adjacent.close() == (0, 0)
62+
63+
64+
def test_same_speaker_in_new_scope_is_a_new_ordinal_and_one_rotation():
65+
tracker = LiveSpeakerOrdinalTracker()
66+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 6.0))
67+
tracker.observe(voice('s2', 1, 'scope:1', 10.0, 16.0))
68+
tracker.observe(voice('s3', 2, 'scope:2', 20.0, 26.0))
69+
assert tracker.close() == (3, 2)
70+
71+
72+
def test_named_voices_never_mint_ordinals():
73+
tracker = LiveSpeakerOrdinalTracker()
74+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 6.0, is_user=True))
75+
tracker.observe(voice('s2', 1, 'scope:0', 0.0, 6.0, person_id='person'))
76+
# A voice named only at close, through the matcher's final map.
77+
tracker.observe(voice('s3', 2, 'scope:0', 0.0, 6.0))
78+
tracker.note_named_voices(SimpleNamespace(speaker_to_person={2: ('person', 'Ada')}, _voice_scopes={2: 'scope:0'}))
79+
assert tracker.close() == (0, 0)
80+
81+
82+
def test_unplaced_or_malformed_spans_add_no_seconds():
83+
tracker = LiveSpeakerOrdinalTracker()
84+
tracker.observe(voice('s1', 0, 'scope:0', -1.0, -1.0))
85+
tracker.observe(voice('s2', 0, 'scope:0', 4.0, 4.0))
86+
tracker.observe(voice('s3', 0, 'scope:0', float('nan'), 6.0))
87+
tracker.observe(voice('s4', 0, 'scope:0', None, '6.0'))
88+
assert tracker.close() == (0, 0)
89+
90+
91+
def test_segments_without_usable_speaker_id_are_ignored():
92+
tracker = LiveSpeakerOrdinalTracker()
93+
tracker.observe({'id': 's1', 'start': 0.0, 'end': 9.0})
94+
tracker.observe(voice('s2', 'zero', 'scope:0', 0.0, 9.0))
95+
tracker.observe(voice('s3', True, 'scope:0', 0.0, 9.0))
96+
assert tracker.close() == (0, 0)
97+
98+
99+
def test_voice_inventory_is_bounded():
100+
beyond = LiveSpeakerOrdinalTracker()
101+
for index in range(MAX_TRACKED_VOICES):
102+
beyond.observe(voice(f's{index}', index, 'scope:0', 0.0, 1.0))
103+
# The (MAX_TRACKED_VOICES + 1)-th distinct voice is dropped, never a crash.
104+
beyond.observe(voice('late', MAX_TRACKED_VOICES, 'scope:0', 0.0, 60.0))
105+
assert beyond.close() == (0, 0)
106+
107+
108+
def test_record_observes_histogram_and_rotation_counter():
109+
before_observations = OMI_LIVE_SPEAKER_ORDINALS_PER_SESSION._sum.get()
110+
before_rotations = OMI_LIVE_SPEAKER_SCOPE_ROTATIONS_TOTAL._value.get()
111+
tracker = LiveSpeakerOrdinalTracker()
112+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 6.0))
113+
tracker.observe(voice('s2', 1, 'scope:1', 8.0, 14.0))
114+
tracker.observe(voice('s3', 2, 'scope:2', 20.0, 26.0))
115+
record_live_speaker_ordinals(tracker, SimpleNamespace(speaker_to_person={}, _voice_scopes={}))
116+
assert OMI_LIVE_SPEAKER_ORDINALS_PER_SESSION._sum.get() == before_observations + 3
117+
assert OMI_LIVE_SPEAKER_SCOPE_ROTATIONS_TOTAL._value.get() == before_rotations + 2
118+
119+
120+
def test_record_never_raises_and_close_resets_state():
121+
record_live_speaker_ordinals(None, None)
122+
record_live_speaker_ordinals(LiveSpeakerOrdinalTracker(), object())
123+
tracker = LiveSpeakerOrdinalTracker()
124+
tracker.observe(voice('s1', 0, 'scope:0', 0.0, 6.0))
125+
assert tracker.close() == (1, 0)
126+
assert tracker.close() == (0, 0)
127+
128+
129+
async def test_speaker_detection_feeds_the_session_tracker():
130+
observed = []
131+
host = SimpleNamespace(
132+
limits=SimpleNamespace(max_segment_buffer_size=10, max_photo_buffer_size=10),
133+
translation_language=None,
134+
language='en',
135+
speakers=SimpleNamespace(
136+
observe_segment=lambda speaker_id, scope, segment_id: observed.append((speaker_id, scope, segment_id)),
137+
person_embeddings={},
138+
speaker_to_person={},
139+
),
140+
state=SimpleNamespace(speaker_id_enabled=False, current_conversation_id='c1'),
141+
speaker_ordinal_tracker=LiveSpeakerOrdinalTracker(),
142+
)
143+
processor = TranscriptProcessor(host)
144+
segments = [
145+
TranscriptSegment(
146+
id='s1',
147+
text='Testing one two',
148+
speaker_id=0,
149+
speaker_id_scope='scope:0',
150+
start=0.0,
151+
end=6.0,
152+
is_user=False,
153+
speech_profile_processed=True,
154+
),
155+
TranscriptSegment(
156+
id='s2',
157+
text='Testing three four',
158+
speaker_id=1,
159+
speaker_id_scope='scope:0',
160+
start=6.0,
161+
end=8.0,
162+
is_user=False,
163+
speech_profile_processed=True,
164+
),
165+
]
166+
await processor._speaker_detection(segments, 0.0)
167+
assert observed == [(0, 'scope:0', 's1'), (1, 'scope:0', 's2')]
168+
# One qualified voice plus one still below the deferral floor, one scope.
169+
assert host.speaker_ordinal_tracker.close() == (1, 0)
Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
"""Live per-session unnamed-speaker ordinal churn, observed once when a listen session closes.
2+
3+
#23533 defers a live ordinal until an unnamed voice holds MIN_EVIDENCE_SECONDS of
4+
non-overlapping placed speech; before that floor the voice renders "Unknown
5+
speaker" and mints no ordinal. This metric answers whether that deferral actually
6+
reduced the consecutive "Speaker 4, 5, 6, 7, 8" lines seen live: how many distinct
7+
visible unnamed ordinals one session still mints, and how many speaker scope/epoch
8+
rotations (provider transitions, reconnect epochs) the session spent. Rotations
9+
re-key the same humans and re-mint their ordinals, so the pair separates label
10+
churn from real multi-speaker conversations.
11+
12+
Qualification mirrors ``utils.speakers.display.display_participants`` with
13+
``defer_short_voices=True``: the same voice key ``(speaker_id_scope, speaker_id)``,
14+
the same five-second union of placed speech, and a voice the session named (owner
15+
or person identity) never mints an ordinal. This is display bookkeeping only; it
16+
never feeds speaker decisions.
17+
18+
Labels: none. Never uid, session or conversation identifiers.
19+
"""
20+
21+
from __future__ import annotations
22+
23+
import logging
24+
from math import isfinite
25+
from typing import Any, Mapping, Optional
26+
27+
from prometheus_client import Counter, Histogram
28+
29+
from utils.speakers.policy import MIN_EVIDENCE_SECONDS
30+
31+
logger = logging.getLogger(__name__)
32+
33+
# Voice inventory ceiling, matching routers.listen.speakers.MAX_SPEAKER_VOICES;
34+
# a session minting more unnamed voices than this is already collapsed to "Others"
35+
# by the display layer's MAX_UNNAMED_VOICES guard.
36+
MAX_TRACKED_VOICES = 128
37+
# Per-voice merged placed-speech intervals while a voice is still below the
38+
# evidence floor. Past this many disjoint bursts the tracker stops extending the
39+
# union (under-counting errs toward "Unknown speaker", the same side as deferral).
40+
MAX_VOICE_INTERVALS = 512
41+
42+
OMI_LIVE_SPEAKER_ORDINALS_PER_SESSION = Histogram(
43+
'omi_live_speaker_ordinals_per_session',
44+
'Distinct visible unnamed speaker ordinals one live listen session minted, '
45+
'after the #23533 short-voice deferral floor',
46+
buckets=(0, 1, 2, 3, 4, 5, 6, 7, 8, 10, 13, 16, 24, 32),
47+
)
48+
OMI_LIVE_SPEAKER_SCOPE_ROTATIONS_TOTAL = Counter(
49+
'omi_live_speaker_scope_rotations_total',
50+
'Speaker scope/epoch rotations inside live listen sessions (provider transitions and '
51+
'reconnects), summed per session at close',
52+
)
53+
54+
55+
def _merge_interval(intervals: list[tuple[float, float]], span: tuple[float, float]) -> None:
56+
"""Insert one span into a sorted, disjoint interval list, merging overlaps in place."""
57+
if len(intervals) >= MAX_VOICE_INTERVALS:
58+
return
59+
start, end = span
60+
merged: list[tuple[float, float]] = []
61+
placed = False
62+
for low, high in intervals:
63+
if high < start:
64+
merged.append((low, high))
65+
elif end < low:
66+
if not placed:
67+
merged.append((start, end))
68+
placed = True
69+
merged.append((low, high))
70+
else:
71+
start = min(start, low)
72+
end = max(end, high)
73+
if not placed:
74+
merged.append((start, end))
75+
intervals[:] = merged
76+
77+
78+
class LiveSpeakerOrdinalTracker:
79+
"""Per-session ordinal ledger. Event-loop confined, like the collapse monitor.
80+
81+
Observe every final live segment (post scope-stamp and speaker-id assignment);
82+
close once at session teardown. Unplaced or malformed spans add no speech
83+
seconds, exactly like the display deferral's placement gate.
84+
"""
85+
86+
def __init__(self, min_evidence_seconds: float = MIN_EVIDENCE_SECONDS) -> None:
87+
self._min_evidence_seconds = min_evidence_seconds
88+
self._intervals: dict[tuple[str, int], list[tuple[float, float]]] = {}
89+
self._qualified: set[tuple[str, int]] = set()
90+
self._named: set[tuple[str, int]] = set()
91+
self._scopes: set[str] = set()
92+
93+
def observe(self, segment: Mapping[str, Any]) -> None:
94+
speaker_id = segment.get('speaker_id')
95+
if isinstance(speaker_id, bool) or not isinstance(speaker_id, int):
96+
return
97+
scope = segment.get('speaker_id_scope')
98+
scope = scope if isinstance(scope, str) else ''
99+
if scope:
100+
self._scopes.add(scope)
101+
key = (scope, speaker_id)
102+
if segment.get('is_user') or segment.get('person_id'):
103+
self._named.add(key)
104+
if key in self._qualified:
105+
return
106+
start, end = segment.get('start'), segment.get('end')
107+
if (
108+
isinstance(start, bool)
109+
or isinstance(end, bool)
110+
or not isinstance(start, (int, float))
111+
or not isinstance(end, (int, float))
112+
):
113+
return
114+
start, end = float(start), float(end)
115+
if not (isfinite(start) and isfinite(end) and 0 <= start < end):
116+
return
117+
intervals = self._intervals.get(key)
118+
if intervals is None:
119+
if len(self._intervals) >= MAX_TRACKED_VOICES:
120+
return
121+
intervals = self._intervals[key] = []
122+
_merge_interval(intervals, (start, end))
123+
if sum(high - low for low, high in intervals) >= self._min_evidence_seconds:
124+
# Qualified: the ordinal is minted unless the voice is named by close.
125+
self._qualified.add(key)
126+
del self._intervals[key]
127+
128+
def note_named_voices(self, matcher: Any) -> None:
129+
"""Fold the matcher's final voice-to-person map into the named set."""
130+
mappings = getattr(matcher, 'speaker_to_person', None)
131+
if not isinstance(mappings, Mapping):
132+
return
133+
scopes = getattr(matcher, '_voice_scopes', None)
134+
scopes = scopes if isinstance(scopes, Mapping) else {}
135+
for voice in mappings:
136+
if isinstance(voice, int) and not isinstance(voice, bool):
137+
self._named.add((scopes.get(voice) or '', voice))
138+
139+
def close(self) -> tuple[int, int]:
140+
"""(minted unnamed ordinals, scope rotations) for the session; resets state."""
141+
ordinals = len(self._qualified - self._named)
142+
rotations = max(0, len(self._scopes) - 1)
143+
self._intervals.clear()
144+
self._qualified.clear()
145+
self._named.clear()
146+
self._scopes.clear()
147+
return ordinals, rotations
148+
149+
150+
def record_live_speaker_ordinals(tracker: Optional[LiveSpeakerOrdinalTracker], matcher: Any) -> None:
151+
"""Observe one ended listen session. Never raises: telemetry must not fail teardown."""
152+
try:
153+
if tracker is None:
154+
return
155+
tracker.note_named_voices(matcher)
156+
ordinals, rotations = tracker.close()
157+
OMI_LIVE_SPEAKER_ORDINALS_PER_SESSION.observe(ordinals)
158+
if rotations:
159+
OMI_LIVE_SPEAKER_SCOPE_ROTATIONS_TOTAL.inc(rotations)
160+
except Exception as error:
161+
logger.warning('Live speaker ordinal metric failed type=%s', type(error).__name__)

0 commit comments

Comments
 (0)