fix(desktop): detect Telegram calls and gate meeting treatment at finalization (#11832)

## What and why

- Add Telegram Desktop's shipping bundle ID (`com.tdesktop.Telegram`,
normalized to lowercase) and legacy `ru.keepcoder.telegram` ID to native
call detection. The proactive chat-app policy consumes the same Telegram
ID set so the catalogs cannot drift independently.
- Keep `conversation_role` as OPEN-time session provenance. Introduce a
separate backend-owned `meeting_treatment_eligible` finalization result,
persist it with the durable finalization job, and expose it through both
the finalization-status and synchronous from-segments responses.
- Gate the meeting Chat arrival and desktop completion notification on
that finalization result. Short or mostly silent calls still finalize as
ordinary conversations without rewriting their role or provenance.
- Count actual speech as the union of non-empty transcript intervals,
preventing simultaneous microphone and system-audio transcript twins
from roughly doubling the speech measure.

Meeting treatment requires at least five minutes of wall-clock duration
and at least 60 seconds of deduplicated transcribed speech. Five minutes
filters quick calls that should remain ordinary conversations. The
smaller speech floor still admits listening-heavy meetings and natural
pauses while rejecting long hold/silence captures with only a brief
accidental transcription. Interval union makes the floor independent of
whether one or both desktop audio streams transcribed the same remote
speech.

## Line-count exception

Line-Count-Exception: backend/routers/developer.py | 2111 -> 2115 |
propagates the new meeting_treatment_eligible field through the existing
conversation response model and its two projection sites; splitting this
router is unrelated to this fix and would make the change unreviewable.

## Failure class

Failure-Class: none

No registered failure class matches this macOS native-process catalog
omission or the separate finalization-time treatment boundary.
`FC-meeting-trigger-title-identity-drift` is a Windows
browser-title/event identity contract. The production `isNativeCallApp`
regression and shared Telegram ID set guard the catalog boundary;
backend behavioral tests guard duration, deduplicated speech,
finalization persistence, Chat arrival, and response projection.

## Verification

-
`PYTHON=/Users/dazheng/workspace/omi/upstream-keep-clean/backend/.venv/bin/python
bash test-preflight.sh` (from `backend/`) — 17 passed, 9 optional
warnings, 0 failed.
-
`PYTHON=/Users/dazheng/workspace/omi/upstream-keep-clean/backend/.venv/bin/python
BACKEND_PYTEST_WORKERS=6
BACKEND_UNIT_TEST_FILE_LIST=../.backend-meeting-test-files.txt bash
test.sh` (from `backend/`; the temporary selector listed the six
changed/new backend test files and was then deleted) — 195 passed,
including the four dedicated meeting-treatment policy tests.
-
`PYTHON=/Users/dazheng/workspace/omi/upstream-keep-clean/backend/.venv/bin/python
BACKEND_PYTEST_WORKERS=10 bash test.sh` (from `backend/`) — all 856
backend unit-test files were selected and the runner progressed through
nearly the entire suite, but one remaining file produced no output for
several minutes and the file-isolated runner has no per-file timeout.
The run was interrupted and did not produce a terminal pass/fail
summary; full-suite status is therefore unverified.
- `./scripts/run-swift-ci.sh --test` (from `desktop/macos/`) — could not
start because the pinned `/Applications/Xcode_16.4.app` toolchain is
unavailable in this environment.
-
`CLANG_MODULE_CACHE_PATH=/Volumes/scratch/tmp/meeting-detect-swift-cache/clang
SWIFTPM_MODULECACHE_OVERRIDE=/Volumes/scratch/tmp/meeting-detect-swift-cache/clang
xcrun swift test --jobs 1 -Xswiftc -j1 --disable-index-store
--disable-sandbox --skip-update --disable-build-manifest-caching
--manifest-cache none --cache-path
/Volumes/scratch/tmp/meeting-detect-swift-cache/swiftpm-cache
--config-path
/Volumes/scratch/tmp/meeting-detect-swift-cache/swiftpm-config
--security-path
/Volumes/scratch/tmp/meeting-detect-swift-cache/swiftpm-security
--scratch-path /private/tmp/meeting-swift-build --package-path
desktop/macos/Desktop --filter
'ConferencingAppsTests|TranscriptionFinalizationStateMachineTests|ProactiveAssistantOrchestrationPolicyTests|APIClientConversationCountTests'`
— SwiftPM reached the debug build and then failed with opaque `error:
fatalError`; tests did not execute.
-
`CLANG_MODULE_CACHE_PATH=/Volumes/scratch/tmp/meeting-detect-swift-cache/harness-clang
SWIFTPM_MODULECACHE_OVERRIDE=/Volumes/scratch/tmp/meeting-detect-swift-cache/harness-clang
xcrun swiftc -module-cache-path
/Volumes/scratch/tmp/meeting-detect-swift-cache/harness-clang
desktop/macos/Desktop/Sources/ConferencingApps.swift
.meeting-detect-catalog-harness.swift -o
/private/tmp/meeting-detect-catalog-harness &&
/private/tmp/meeting-detect-catalog-harness` — directly compiled the
production classifier; both Telegram IDs classified as native call apps
and Omi did not. Printed `Telegram catalog classification passed`; the
temporary source was deleted.
- `python3 scripts/check_desktop_test_quality.py` (from
`desktop/macos/`) — passed with no new source-reading tests, trapping
dictionary initializers, or wall-clock waits.
- `scripts/swift-format-wrapper.sh lint <nine changed Swift files>`
(from `desktop/macos/`) — passed.
- `make preflight` — all 11 manifest checks passed. Because the changes
were still uncommitted, the preflight runner reported `files=0`; rerun
after the commit so diff-scoped selection is meaningful.
- `git diff --check` — passed.

## Verification limits

- A real Telegram call and named desktop bundle were not exercised
because the pinned desktop toolchain is unavailable and SwiftPM cannot
complete whole-module emission with the available toolchain.
- The full backend suite did not reach a terminal summary; the
changed-area component run is green, but the remaining backend files are
not claimed as verified.

## Follow-ups

- Add Tier-2 detection for unknown CoreAudio processes only after a
telemetry-only phase builds an observed allowlist from real process
topologies. `isRunningOutput` means an active output stream, not audible
speech; games, DAWs, and media apps can satisfy it, while split-process
Electron/browser calls can fail a same-process input+output rule.
- Design idle-based conversation rotation through the existing
serialized meeting-boundary path. It must never become a second rotation
authority: a failed meeting rotation can call `stopTranscription()`, so
an independent idle race could stop always-on capture.

🤖 Generated with [Claude Code](https://claude.com/claude-code)


<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/BasedHardware/omi/pull/11832?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: Git-on-my-level

backend/database/conversation_finalization_jobs.py

@@ -731,6 +731,7 @@ def _mark_finalization_fanout_completed_txn(
     dispatch_generation: int,
     lease_epoch: int,
     now: datetime,
+    meeting_treatment_eligible: bool,
 ) -> bool:
     snapshot = job_ref.get(transaction=transaction)
     if not getattr(snapshot, 'exists', False):
@@ -745,6 +746,7 @@ def _mark_finalization_fanout_completed_txn(
         {
             'fanout_status': 'completed',
             'fanout_completed_at': now,
+            'meeting_treatment_eligible': meeting_treatment_eligible,
             'updated_at': now,
         },
     )
@@ -756,12 +758,20 @@ def mark_finalization_fanout_completed(
     dispatch_generation: int,
     lease_epoch: int,
     *,
+    meeting_treatment_eligible: bool,
     firestore_client: Any = None,
 ) -> bool:
     client = _client(firestore_client)
     transaction = client.transaction()
     transactional = firestore.transactional(_mark_finalization_fanout_completed_txn)
-    return transactional(transaction, _job_ref(client, job_id), dispatch_generation, lease_epoch, _now())
+    return transactional(
+        transaction,
+        _job_ref(client, job_id),
+        dispatch_generation,
+        lease_epoch,
+        _now(),
+        meeting_treatment_eligible,
+    )
 
 
 def _mark_finalization_retryable_txn(

backend/models/conversation.py

@@ -348,6 +348,7 @@ class ConversationFinalizationStatusResponse(BaseModel):
     retryable: bool
     attempt_count: int
     task_retry_count: int
+    meeting_treatment_eligible: bool = False
 
 
 # MIGRATE: For backward compatibility with the old memories routes and app

backend/routers/developer.py

@@ -56,6 +56,7 @@
 from utils.conversations.process_conversation import process_conversation
 from utils.conversations import lifecycle as lifecycle_service
 from utils.conversations.location import resolve_geolocation
+from utils.conversations.meeting_treatment import is_meeting_treatment_eligible
 from utils.executors import postprocess_executor
 from utils.request_validation import HistoryDays
 from utils.llm.memories import identify_category_for_memory
@@ -1084,6 +1085,7 @@ class ConversationResponse(BaseModel):
     id: str
     status: str
     discarded: bool
+    meeting_treatment_eligible: bool = False
 
 
 class UpdateConversationRequest(BaseModel):
@@ -1474,6 +1476,7 @@ def _conversation_response_from_data(conversation: dict) -> ConversationResponse
         id=conversation['id'],
         status=status,
         discarded=bool(conversation.get('discarded', False)),
+        meeting_treatment_eligible=is_meeting_treatment_eligible(conversation),
     )
 
 
@@ -1670,6 +1673,7 @@ def _create_conversation_from_segments(
         id=conversation.id,
         status=conversation.status.value if conversation.status else 'completed',
         discarded=conversation.discarded,
+        meeting_treatment_eligible=is_meeting_treatment_eligible(conversation),
     )
 
 

backend/tests/unit/test_chat_first_proactive_engine.py

@@ -1,6 +1,6 @@
 """Failure-isolated, content-free proactive-judgment contracts."""
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 import logging
 from types import SimpleNamespace
 from unittest.mock import MagicMock
@@ -15,6 +15,12 @@
     QuestionOption,
 )
 from utils.task_intelligence.chat_first_eligibility import ChatFirstEligibility
+from utils.conversations.meeting_treatment import (
+    MIN_MEETING_DURATION_SECONDS,
+    MIN_TRANSCRIBED_SPEECH_SECONDS,
+    deduplicated_transcribed_speech_seconds,
+    is_meeting_treatment_eligible,
+)
 
 NOW = datetime(2026, 7, 15, 12, tzinfo=timezone.utc)
 SUBJECT = ChatFirstSubject(kind='goal', id='goal-1')
@@ -223,6 +229,9 @@ def test_desktop_meeting_adapter_uses_stored_role_and_skips_non_meeting_or_rotat
         'source': 'desktop',
         'status': 'completed',
         'discarded': False,
+        'started_at': NOW,
+        'finished_at': NOW + timedelta(seconds=MIN_MEETING_DURATION_SECONDS),
+        'transcript_segments': [{'text': 'A substantive exchange', 'start': 0, 'end': MIN_TRANSCRIBED_SPEECH_SECONDS}],
         'structured': {'title': 'Ambient capture'},
         'external_data': {'conversation_role': 'ambient'},
     }
@@ -254,6 +263,35 @@ def test_desktop_meeting_adapter_uses_stored_role_and_skips_non_meeting_or_rotat
     persist.assert_not_called()
 
 
+def test_meeting_treatment_requires_five_minutes_and_deduplicated_speech():
+    eligible = {
+        'source': 'desktop',
+        'discarded': False,
+        'started_at': NOW,
+        'finished_at': NOW + timedelta(seconds=MIN_MEETING_DURATION_SECONDS),
+        'external_data': {'conversation_role': 'meeting'},
+        'transcript_segments': [
+            {'text': 'first exchange', 'start': 0, 'end': 35},
+            {'text': 'second exchange', 'start': 35, 'end': MIN_TRANSCRIBED_SPEECH_SECONDS},
+        ],
+    }
+    assert is_meeting_treatment_eligible(eligible) is True
+
+    short_call = {**eligible, 'finished_at': NOW + timedelta(seconds=MIN_MEETING_DURATION_SECONDS - 1)}
+    assert is_meeting_treatment_eligible(short_call) is False
+
+    duplicate_streams = {
+        **eligible,
+        'finished_at': NOW + timedelta(minutes=20),
+        'transcript_segments': [
+            {'text': 'remote stream from mic', 'start': 0, 'end': 45},
+            {'text': 'same remote stream from system audio', 'start': 0, 'end': 45},
+        ],
+    }
+    assert deduplicated_transcribed_speech_seconds(duplicate_streams['transcript_segments']) == 45
+    assert is_meeting_treatment_eligible(duplicate_streams) is False
+
+
 def test_proactive_failure_logs_redact_authenticated_uid(monkeypatch, caplog):
     uid = 'sensitive-user-123456'
     monkeypatch.setattr(

backend/tests/unit/test_conversation_finalization_jobs.py

@@ -493,7 +493,8 @@ def test_finalization_completion_requires_durable_fanout_completion():
     ref.data = ref.data | fanout.updates[0][1]
 
     completed_fanout = _Transaction()
-    assert jobs._mark_finalization_fanout_completed_txn(completed_fanout, ref, 1, 4, now) is True
+    assert jobs._mark_finalization_fanout_completed_txn(completed_fanout, ref, 1, 4, now, True) is True
+    assert completed_fanout.updates[0][1]['meeting_treatment_eligible'] is True
     ref.data = ref.data | completed_fanout.updates[0][1]
 
     completed = _Transaction()

backend/tests/unit/test_conversation_search_date_validation.py

@@ -696,6 +696,7 @@ def test_finalization_status_endpoint_exposes_retryable_durable_state():
         'retryable': True,
         'attempt_count': 2,
         'task_retry_count': 1,
+        'meeting_treatment_eligible': False,
     }
     with (
         patch.object(conv.conversations_db, 'get_conversation', return_value={'id': 'conv-1'}),

backend/tests/unit/test_developer_from_segments_idempotency.py

@@ -173,6 +173,25 @@ def _request(**overrides):
     return developer.CreateConversationFromTranscriptRequest.model_validate(data)
 
 
+def _eligible_meeting_request(**overrides):
+    data = {
+        'transcript_segments': [
+            {
+                'text': 'substantive meeting discussion',
+                'speaker': 'SPEAKER_00',
+                'is_user': True,
+                'start': 0.0,
+                'end': 60.0,
+            }
+        ],
+        'started_at': NOW,
+        'finished_at': NOW + timedelta(minutes=5),
+        'conversation_role': 'meeting',
+    }
+    data.update(overrides)
+    return _request(**data)
+
+
 def test_no_client_session_id_preserves_create_conversation_path(monkeypatch):
     captured = {}
 
@@ -200,7 +219,7 @@ def _process(uid, language, conversation):
 
     arrival = MagicMock()
     monkeypatch.setattr(proactive_engine, 'persist_capture_arrival_intent', arrival)
-    response = developer._create_conversation_from_segments('uid1', _request(conversation_role='meeting'))
+    response = developer._create_conversation_from_segments('uid1', _eligible_meeting_request())
 
     assert response.id == 'random-process-id'
     assert isinstance(captured['conversation'], CreateConversation)
@@ -281,10 +300,11 @@ def _process(_uid, _language, conversation):
     monkeypatch.setattr(proactive_engine, 'persist_capture_arrival_intent', arrival)
 
     response = developer._create_conversation_from_segments(
-        'uid1', _request(client_session_id='meeting-session-1', conversation_role='meeting')
+        'uid1', _eligible_meeting_request(client_session_id='meeting-session-1')
     )
 
     assert response.id == expected_id
+    assert response.meeting_treatment_eligible is True
     arrival.assert_called_once_with(
         'uid1', conversation_id=expected_id, summary='Design review', is_desktop_meeting=True
     )
@@ -342,6 +362,9 @@ def test_completed_desktop_meeting_retry_repairs_missing_arrival(monkeypatch):
                 'source': 'desktop',
                 'status': 'completed',
                 'discarded': False,
+                'started_at': NOW,
+                'finished_at': NOW + timedelta(minutes=5),
+                'transcript_segments': [{'text': 'substantive meeting discussion', 'start': 0.0, 'end': 60.0}],
                 'structured': {'title': 'Design review'},
                 'external_data': {'conversation_role': 'meeting'},
             }
@@ -353,16 +376,46 @@ def test_completed_desktop_meeting_retry_repairs_missing_arrival(monkeypatch):
     monkeypatch.setattr(proactive_engine, 'persist_capture_arrival_intent', arrival)
 
     response = developer._create_conversation_from_segments(
-        'uid1', _request(client_session_id='meeting-session-1', conversation_role='meeting')
+        'uid1', _eligible_meeting_request(client_session_id='meeting-session-1')
     )
 
     assert response.id == expected_id
+    assert response.meeting_treatment_eligible is True
     process.assert_not_called()
     arrival.assert_called_once_with(
         'uid1', conversation_id=expected_id, summary='Design review', is_desktop_meeting=True
     )
 
 
+def test_short_desktop_meeting_stays_ordinary_conversation(monkeypatch):
+    monkeypatch.setattr(conversations_db, 'get_conversation', MagicMock())
+    monkeypatch.setattr(developer.lifecycle_service, 'create_processing_conversation', MagicMock())
+
+    def _process(_uid, _language, conversation):
+        return Conversation(
+            id='short-meeting',
+            created_at=NOW,
+            started_at=conversation.started_at,
+            finished_at=conversation.finished_at,
+            source=conversation.source,
+            language=conversation.language,
+            structured={'title': 'Short call'},
+            transcript_segments=conversation.transcript_segments,
+            external_data=conversation.external_data,
+            status=ConversationStatus.completed,
+        )
+
+    monkeypatch.setattr(developer, 'process_conversation', _process)
+    arrival = MagicMock()
+    monkeypatch.setattr(proactive_engine, 'persist_capture_arrival_intent', arrival)
+
+    response = developer._create_conversation_from_segments('uid1', _request(conversation_role='meeting'))
+
+    assert response.status == 'completed'
+    assert response.meeting_treatment_eligible is False
+    arrival.assert_not_called()
+
+
 def test_completed_ambient_retry_cannot_reclassify_conversation_as_meeting(monkeypatch):
     expected_id = developer._from_segments_conversation_id('uid1', 'ambient-session-1')
     monkeypatch.setattr(

backend/tests/unit/test_listen_finalization_cloud_tasks.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import json
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 import runpy
 from types import SimpleNamespace
@@ -315,17 +316,20 @@ def test_finalization_status_exposes_retry_and_terminal_state(monkeypatch):
         'retryable': True,
         'attempt_count': 2,
         'task_retry_count': 0,
+        'meeting_treatment_eligible': False,
     }
 
     job['status'] = 'dead_letter'
     job['task_retry_count'] = 3
+    job['meeting_treatment_eligible'] = True
     assert lifecycle_service.get_finalization_status('uid-1', 'conversation-1') == {
         'job_id': 'job-1',
         'status': 'dead_letter',
         'terminal': True,
         'retryable': False,
         'attempt_count': 2,
         'task_retry_count': 3,
+        'meeting_treatment_eligible': True,
     }
 
 
@@ -1054,6 +1058,9 @@ async def inline_run_blocking(_executor, func, *args, **kwargs):
         source=SimpleNamespace(value=source),
         external_data=external_data,
         discarded=discarded,
+        started_at=datetime(2026, 8, 18, 12, tzinfo=timezone.utc),
+        finished_at=datetime(2026, 8, 18, 12, tzinfo=timezone.utc) + timedelta(minutes=10),
+        transcript_segments=[SimpleNamespace(text='substantive exchange', start=0, end=60)],
         structured=SimpleNamespace(title='Captured title', overview='Captured overview'),
     )
     integrations = AsyncMock(return_value=[])
@@ -1097,7 +1104,12 @@ async def inline_run_blocking(_executor, func, *args, **kwargs):
     else:
         extracted.assert_called_once_with('uid-1', conversation)
     assert disposition == ConversationFinalizationDisposition.completed
-    completed.assert_called_once_with('job-1', 2, 3)
+    completed.assert_called_once_with(
+        'job-1',
+        2,
+        3,
+        meeting_treatment_eligible=(source == 'desktop' and expected_intent_kwargs is not None),
+    )
     if expected_intent_kwargs is None:
         capture_arrival.assert_not_called()
     else:
@@ -1440,7 +1452,7 @@ def contract_faithful_process(_uid, _lang, conv, **kwargs):
     assert disposition == ConversationFinalizationDisposition.completed
     derived_runner.assert_called_once()
     integrations.assert_awaited_once()
-    complete.assert_called_once_with('job-1', 2, 3)
+    complete.assert_called_once_with('job-1', 2, 3, meeting_treatment_eligible=False)
 
 
 @pytest.mark.anyio
@@ -1520,7 +1532,7 @@ def contract_faithful_process(_uid, _lang, conv, **kwargs):
     )
 
     assert disposition == ConversationFinalizationDisposition.completed
-    complete.assert_called_once_with('job-1', 2, 3)
+    complete.assert_called_once_with('job-1', 2, 3, meeting_treatment_eligible=False)
     safe_target.assert_called_once_with('https://app.test/hook?uid=uid-1')
     webhook_client.post.assert_awaited_once_with(
         pinned_url,