feat(telemetry): instrument processing lifecycle (#11540)

## Summary

- add accepted/terminal transcript lifecycle telemetry around the live
STT authority
- record speaker proposals and durable speaker-assignment confirmations
without person names or transcript content
- link extracted tasks to their source conversation and record bounded
assignee corrections
- measure authoritative decoded-audio duration, server-VAD speech
duration, and completed diarization outcomes

Processing-cost telemetry is intentionally excluded because this path
does not expose a reliable authoritative cost value.

## Verification

- `backend/test.sh` over the eight changed unit files plus the receiver
Opus regression file: passed
- `python scripts/scan_async_blockers.py --dirs routers utils`: passed
- `make preflight` against current `main` with this PR body: passed (24
diff-scoped checks)

The live provider WebSocket path was not exercised locally; lifecycle
behavior was verified through the production receiver/runtime seams with
injected telemetry clients.

Line-Count-Exception: backend/routers/conversations.py | 1538 -> 1605 |
speaker confirmation must emit after each authoritative durable
assignment mutation already owned by this router
Line-Count-Exception:
backend/utils/conversations/process_conversation.py | 2204 -> 2225 |
task extraction telemetry stays at the existing canonical and legacy
persistence boundary

## Product invariants affected

none

## Failure class (fixes)

Failure-Class: none


<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/BasedHardware/omi/pull/11540?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

Author: axAilotl

backend/routers/action_items.py

@@ -43,6 +43,7 @@
     PendingSyncResponse,
 )
 from utils.task_intelligence import task_links
+from utils.product_telemetry import emit_product_event
 
 router = APIRouter()
 
@@ -486,6 +487,25 @@ def update_action_item(
     updated_item = action_items_db.get_action_item(uid, action_item_id)
     if updated_item is None:
         raise HTTPException(status_code=500, detail="Updated action item could not be loaded")
+
+    if request.owner is not None:
+        previous_owner_value = existing_item.get('owner') or 'unknown'
+        previous_owner = getattr(previous_owner_value, 'value', str(previous_owner_value))
+        next_owner = request.owner.value
+        if previous_owner != next_owner:
+            emit_product_event(
+                uid=uid,
+                event='Task Assignee Corrected',
+                properties={
+                    'action_item_id': action_item_id,
+                    'conversation_id': updated_item.get('conversation_id'),
+                    'previous_assignee': (
+                        previous_owner if previous_owner in {'user', 'other', 'unknown'} else 'unknown'
+                    ),
+                    'new_assignee': next_owner,
+                    'field_changed': 'owner',
+                },
+            )
     _wake_task_changes(uid, [action_item_id], updated_item.get('updated_at'))
 
     # Reconcile the client-scheduled reminder when completion or due date changed, using the final

backend/routers/conversations.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 
 from fastapi import APIRouter, Depends, HTTPException, Query, BackgroundTasks
 from typing import Any, Dict, List, Optional
@@ -65,6 +66,7 @@
 from utils.other.storage import get_conversation_recording_if_exists
 from utils.app_integrations import trigger_external_integrations
 from utils.request_validation import NonNegativeOffset, PositiveLimit
+from utils.product_telemetry import emit_product_event
 from utils.conversations.calendar_linking import (
     get_overlapping_calendar_event,
     write_conversation_link_to_calendar_event,
@@ -91,6 +93,45 @@ def _get_valid_conversation_by_id(uid: str, conversation_id: str) -> dict:
     return conversation
 
 
+def _speaker_assignment(segment: TranscriptSegment) -> str:
+    if segment.is_user:
+        return 'self'
+    if segment.person_id:
+        return f"person:{hashlib.sha256(str(segment.person_id).encode('utf-8')).hexdigest()[:16]}"
+    return 'unassigned'
+
+
+def _speaker_assignment_kind(assignment: str) -> str:
+    return 'person' if assignment.startswith('person:') else assignment
+
+
+def _emit_speaker_identity_confirmed(
+    *,
+    uid: str,
+    conversation_id: str,
+    scope: str,
+    before: List[str],
+    after: List[str],
+) -> None:
+    if not after:
+        return
+    assignment_kinds = [_speaker_assignment_kind(value) for value in after]
+    properties = {
+        'conversation_id': conversation_id,
+        'confirmation': 'accepted' if before == after else 'corrected',
+        'assignment': assignment_kinds[0] if len(set(assignment_kinds)) == 1 else 'mixed',
+        'scope': scope,
+        'affected_segment_count': len(after),
+    }
+    if len(set(after)) == 1 and assignment_kinds[0] == 'person':
+        properties['assignment_id'] = after[0]
+    emit_product_event(
+        uid=uid,
+        event='Speaker Identity Confirmed',
+        properties=properties,
+    )
+
+
 def _enrich_deferred_conversation(uid: str, conversation: dict) -> dict:
     """First open of a lazily-deferred desktop conversation. The LLM enrichment (summary, action
     items, memories, embeddings, app results) takes ~10s, so we run it in the BACKGROUND and return
@@ -1021,6 +1062,7 @@ def set_assignee_conversation_segment(
 
     is_unassigning = value is None or value is False
 
+    before = [_speaker_assignment(conversation.transcript_segments[segment_idx])]
     if assign_type == 'is_user':
         conversation.transcript_segments[segment_idx].is_user = bool(value) if value is not None else False
         conversation.transcript_segments[segment_idx].person_id = None
@@ -1034,6 +1076,13 @@ def set_assignee_conversation_segment(
     conversations_db.update_conversation_segments(
         uid, conversation_id, [segment.model_dump() for segment in conversation.transcript_segments]
     )
+    _emit_speaker_identity_confirmed(
+        uid=uid,
+        conversation_id=conversation_id,
+        scope='segment',
+        before=before,
+        after=[_speaker_assignment(conversation.transcript_segments[segment_idx])],
+    )
     # thinh's note: disabled for now
     # segment_words = len(conversation.transcript_segments[segment_idx].text.split(' '))
     # # TODO: can do this async
@@ -1090,6 +1139,9 @@ def set_assignee_conversation_segment(
 
     is_unassigning = value is None or value is False
 
+    targeted_segments = [segment for segment in conversation.transcript_segments if segment.speaker_id == speaker_id]
+    before = [_speaker_assignment(segment) for segment in targeted_segments]
+
     if assign_type == 'is_user':
         for segment in conversation.transcript_segments:
             if segment.speaker_id == speaker_id:
@@ -1108,6 +1160,13 @@ def set_assignee_conversation_segment(
     conversations_db.update_conversation_segments(
         uid, conversation_id, [segment.model_dump() for segment in conversation.transcript_segments]
     )
+    _emit_speaker_identity_confirmed(
+        uid=uid,
+        conversation_id=conversation_id,
+        scope='speaker',
+        before=before,
+        after=[_speaker_assignment(segment) for segment in targeted_segments],
+    )
     # This will be used when we setup recording for conversations, not used for now
     # get the segment with the most words with the speaker_id
     # segment_idx = 0
@@ -1153,6 +1212,7 @@ def assign_segments_bulk(
 
     segment_indices = _resolve_bulk_segment_indices(conversation, data.segment_ids)
     resolved_segment_ids = [conversation.transcript_segments[index].id for index in segment_indices]
+    before = [_speaker_assignment(conversation.transcript_segments[index]) for index in segment_indices]
 
     for index in segment_indices:
         segment = conversation.transcript_segments[index]
@@ -1166,6 +1226,13 @@ def assign_segments_bulk(
     conversations_db.update_conversation_segments(
         uid, conversation_id, [segment.model_dump() for segment in conversation.transcript_segments]
     )
+    _emit_speaker_identity_confirmed(
+        uid=uid,
+        conversation_id=conversation_id,
+        scope='bulk',
+        before=before,
+        after=[_speaker_assignment(conversation.transcript_segments[index]) for index in segment_indices],
+    )
 
     # Trigger speaker sample extraction when assigning to a person
     if data.assign_type == 'person_id' and value:

backend/routers/listen/receiver.py

@@ -70,6 +70,7 @@
 from utils.log_sanitizer import sanitize
 from utils.listen_audio import ChannelConfig, mix_n_channel_buffers, resample_pcm
 from utils.observability.fallback import record_fallback
+from utils.product_telemetry import emit_product_event
 
 logger = logging.getLogger(__name__)
 
@@ -516,11 +517,11 @@ async def _flush_stt_buffer(self, buffer: bytearray, *, force: bool = False) ->
             self._capture('capture_outbound_stt', outbound_audio)
             self.host.state.dg_usage_ms_pending += decision.dg_usage_ms
 
-    async def _handle_multi_channel_audio(self, data: bytes) -> None:
+    async def _handle_multi_channel_audio(self, data: bytes) -> int:
         request = self.host.request
         channel_index = self.channel_id_to_index.get(data[0])
         if channel_index is None:
-            return
+            return 0
         audio = data[1:]
         if request.codec == 'opus' and self.multi_opus_decoders[channel_index]:
             try:
@@ -529,10 +530,10 @@ async def _handle_multi_channel_audio(self, data: bytes) -> None:
                 )
             except Exception as error:
                 self._record_decode_failure('opus', error, len(audio), channel=channel_index)
-                return
+                return 0
             self.decode_failure_streak = 0
             if not audio:
-                return
+                return 0
         pcm = resample_pcm(bytes(audio), request.sample_rate, TARGET_SAMPLE_RATE)
         self._capture('capture_client_audio', pcm)
         # Custom-STT clients own transcript production.  Their channel sockets are intentionally
@@ -575,6 +576,8 @@ async def _handle_multi_channel_audio(self, data: bytes) -> None:
                 for buffer in self.channel_mix_buffers:
                     del buffer[: decision.min_len]
 
+        return len(audio)
+
     async def _handle_text(self, message: str) -> None:
         try:
             loaded = json.loads(message)
@@ -649,6 +652,7 @@ async def _handle_speaker_assigned(self, payload: Dict[str, Any]) -> None:
     async def receive_data(self) -> None:
         request = self.host.request
         buffer = bytearray()
+        decoded_audio_bytes = 0
         self.host.state.last_audio_received_time = time.time()
         self.host.state.last_activity_time = self.host.state.last_audio_received_time
         try:
@@ -674,7 +678,7 @@ async def receive_data(self) -> None:
                         self.host.state.last_usage_record_timestamp = now
                         self.host.start_live_transcription()
                     if self.host.is_multi_channel:
-                        await self._handle_multi_channel_audio(data)
+                        decoded_audio_bytes += await self._handle_multi_channel_audio(data)
                         continue
                     try:
                         decoded: bytes = data
@@ -694,6 +698,7 @@ async def receive_data(self) -> None:
                     self.decode_failure_streak = 0
                     if not decoded:
                         continue
+                    decoded_audio_bytes += len(decoded)
                     self._capture('capture_client_audio', decoded)
                     if self.host.state.audio_ring_buffer is not None:
                         self.host.state.audio_ring_buffer.write(decoded, now)
@@ -710,8 +715,35 @@ async def receive_data(self) -> None:
             logger.error('Listen receive failure type=%s', type(error).__name__)
             self.host.state.close_code = 1011
         finally:
+            if decoded_audio_bytes:
+                sample_rate = max(1, int(getattr(request, 'sample_rate', 16000)))
+                emit_product_event(
+                    uid=str(getattr(request, 'uid', '') or ''),
+                    event='Encoded Audio Duration Measured',
+                    properties={
+                        'recording_id': getattr(self.host, 'recording_session_id', None),
+                        'conversation_id': getattr(self.host.state, 'current_conversation_id', None),
+                        'codec': request.codec,
+                        'decoded_audio_bytes': decoded_audio_bytes,
+                        'duration_seconds': decoded_audio_bytes / (sample_rate * 2),
+                    },
+                )
             if self.vad_gate is not None:
+                vad_metrics = self.vad_gate.get_metrics()
                 logger.info(json.dumps(self.vad_gate.to_json_log()))
+                speech_ms = max(0, int(vad_metrics.get('speech_ms_total') or 0))
+                if speech_ms:
+                    emit_product_event(
+                        uid=str(getattr(request, 'uid', '') or ''),
+                        event='Speech Positive Duration Measured',
+                        properties={
+                            'recording_id': getattr(self.host, 'recording_session_id', None),
+                            'conversation_id': getattr(self.host.state, 'current_conversation_id', None),
+                            'duration_seconds': speech_ms / 1000,
+                            'measurement': 'server_vad',
+                            'vad_mode': vad_metrics.get('mode') or 'unknown',
+                        },
+                    )
             if not self.host.use_custom_stt:
                 await self._flush_stt_buffer(buffer, force=True)
             await self._drain_stt_sockets()

backend/routers/listen/runtime.py

@@ -47,6 +47,7 @@
 from utils.onboarding import OnboardingHandler
 from utils.observability.transcription import LiveSTTAttempt
 from utils.pusher import PusherCircuitBreakerOpen
+from utils.product_telemetry import emit_product_event
 from utils.stt.streaming import get_stt_service_for_language
 from utils.subscription import get_remaining_transcription_seconds, is_trial_paywalled
 from utils.transcribe_decisions import (
@@ -183,6 +184,18 @@ def send_event(self, event: MessageEvent) -> None:
             self.spawn(self.asend_event(event), name='message_event')
 
     def emit_speaker_suggestion(self, speaker_id: int, person_id: str, person_name: str, segment_id: str) -> None:
+        emit_product_event(
+            uid=self.request.uid,
+            event='Speaker Identity Proposed',
+            properties={
+                'recording_id': self.recording_session_id,
+                'conversation_id': self.state.current_conversation_id,
+                'speaker_id': speaker_id,
+                'matched_existing_person': bool(person_id),
+                'auto_assign_enabled': self.request.speaker_auto_assign_enabled,
+                'proposal_source': 'live_speaker_identification',
+            },
+        )
         self.send_event(
             SpeakerLabelSuggestionEvent(
                 speaker_id=speaker_id,
@@ -204,6 +217,12 @@ def start_live_transcription(self) -> None:
             self.state.live_transcription_attempt = LiveSTTAttempt(
                 provider=getattr(self.stt_service, 'value', self.stt_service),
                 platform=self.client_device_context.platform,
+                uid=self.request.uid,
+                recording_id=self.recording_session_id,
+                conversation_id=self.state.current_conversation_id,
+                source=self.request.source,
+                model=self.stt_model,
+                language=self.stt_language,
             )
 
     def capture_client_audio(self, audio: bytes) -> None:

backend/routers/listen/transcripts.py

@@ -38,6 +38,7 @@
 from utils.translation import TranslationService
 from utils.translation_cache import ConversationLanguageState, TranscriptSegmentLanguageCache
 from utils.translation_coordinator import TranslationCoordinator
+from utils.product_telemetry import emit_product_event
 
 logger = logging.getLogger(__name__)
 
@@ -263,12 +264,18 @@ async def _deliver_segments(self, client_segments: List[Dict[str, Any]]) -> bool
         return False
 
     async def process_loop(self) -> None:
+        diarized_speaker_ids_by_conversation: Dict[str, set[int]] = {}
         while self.host.state.active or self.segment_buffer or self.photo_buffer:
             if await self.host.wait(0.6) and not (self.segment_buffer or self.photo_buffer):
                 break
             if not self.segment_buffer and not self.photo_buffer:
                 continue
             raw_segments = sort_segments_by_start(list(self.segment_buffer))
+            conversation_id = self.host.state.current_conversation_id
+            if conversation_id:
+                diarized_speaker_ids_by_conversation.setdefault(conversation_id, set()).update(
+                    int(segment['speaker_id']) for segment in raw_segments if isinstance(segment.get('speaker_id'), int)
+                )
             self.segment_buffer.clear()
             photos = list(self.photo_buffer)
             self.photo_buffer.clear()
@@ -358,6 +365,19 @@ async def process_loop(self) -> None:
             logger.warning('Timed out waiting for listen speaker identification to finish')
         await self.host.speakers.drain(timeout=10, label='listen_speaker_final')
         await self.flush_speaker_assignments(self.host.state.current_conversation_id)
+        for conversation_id, diarized_speaker_ids in diarized_speaker_ids_by_conversation.items():
+            if not diarized_speaker_ids:
+                continue
+            emit_product_event(
+                uid=self.host.request.uid,
+                event='Diarization Completed',
+                properties={
+                    'recording_id': getattr(self.host, 'recording_session_id', None),
+                    'conversation_id': conversation_id,
+                    'speaker_count': len(diarized_speaker_ids),
+                    'source': 'stt_provider',
+                },
+            )
 
     async def _write_fresh(
         self,

backend/tests/unit/test_action_item_canonical_contract.py

@@ -109,6 +109,38 @@ def test_released_macos_request_shapes_remain_compatible_at_route_boundary():
     assert 'clear_due_at' not in update.storage_payload()
 
 
+def test_owner_change_emits_bounded_assignee_correction_after_durable_update(monkeypatch):
+    existing = {'id': 'task-1', 'description': 'Send budget', 'owner': 'unknown', 'completed': False}
+    updated = {**existing, 'owner': 'user', 'conversation_id': 'conversation-1'}
+    emitted = []
+    monkeypatch.setattr(action_items_router, '_get_valid_action_item', lambda *_: existing)
+    monkeypatch.setattr(action_items_router.action_items_db, 'update_action_item', lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(action_items_router.action_items_db, 'get_action_item', lambda *_: updated)
+    monkeypatch.setattr(action_items_router.task_links, 'validate_task_links', lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(action_items_router, 'emit_product_event', lambda **event: emitted.append(event))
+
+    result = action_items_router.update_action_item(
+        'task-1',
+        ActionItemUpdateRequest(owner='user'),
+        uid='user-1',
+    )
+
+    assert result.owner == 'user'
+    assert emitted == [
+        {
+            'uid': 'user-1',
+            'event': 'Task Assignee Corrected',
+            'properties': {
+                'action_item_id': 'task-1',
+                'conversation_id': 'conversation-1',
+                'previous_assignee': 'unknown',
+                'new_assignee': 'user',
+                'field_changed': 'owner',
+            },
+        }
+    ]
+
+
 @pytest.mark.parametrize(
     'payload',
     [