diff --git a/.github/scripts/product_file_line_count_ratchet_baseline/backend-database.json b/.github/scripts/product_file_line_count_ratchet_baseline/backend-database.json index cdf493da682..a31c6d89802 100644 --- a/.github/scripts/product_file_line_count_ratchet_baseline/backend-database.json +++ b/.github/scripts/product_file_line_count_ratchet_baseline/backend-database.json @@ -1,10 +1,10 @@ { "files": { - "backend/database/conversations.py": 1580, + "backend/database/conversations.py": 1601, "backend/database/users.py": 1943 }, "raise_justifications": { - "backend/database/conversations.py": "Public shared-chat uses the existing conversation storage format for bounded, safe transcript decoding; extraction would duplicate crypto/storage invariants." + "backend/database/conversations.py": "Public shared-chat uses the existing conversation storage format for bounded, safe transcript decoding; extraction would duplicate crypto/storage invariants. +21 for the #10033 merge-target eligibility predicate and pure closest-match selector beside the query they filter." }, "threshold": 1500 } diff --git a/backend/database/conversations.py b/backend/database/conversations.py index e5662a24e3d..07809f8af35 100644 --- a/backend/database/conversations.py +++ b/backend/database/conversations.py @@ -1524,6 +1524,34 @@ def _store(transaction) -> bool: # ******************************** +def eligible_merge_target(conversation: Optional[dict]) -> bool: + """Whether synced audio may merge into this conversation (#10033). + + A soft-deleted tombstone must never absorb new segments: the user cannot + see it, so merged audio disappears — recordings that "never create a + conversation". Discarded rows stay eligible; the merge path reprocesses + and revives them. + """ + return bool(conversation) and not conversation.get('deleted') + + +def select_closest_conversation(conversations, start_timestamp: int, end_timestamp: int) -> Optional[dict]: + """Pure closest-by-boundary choice among eligible merge targets (#10033).""" + closest_conversation = None + min_diff = float('inf') + for conversation in conversations: + if not eligible_merge_target(conversation): + continue + conversation_start_timestamp = conversation['started_at'].timestamp() + conversation_end_timestamp = conversation['finished_at'].timestamp() + diff1 = abs(conversation_start_timestamp - start_timestamp) + diff2 = abs(conversation_end_timestamp - end_timestamp) + if diff1 < min_diff or diff2 < min_diff: + min_diff = min(diff1, diff2) + closest_conversation = conversation + return closest_conversation + + @prepare_for_read(decrypt_func=_prepare_conversation_for_read) @with_photos(get_conversation_photos) def get_closest_conversation_to_timestamps(uid: str, start_timestamp: int, end_timestamp: int) -> Optional[dict]: @@ -1548,17 +1576,10 @@ def get_closest_conversation_to_timestamps(uid: str, start_timestamp: int, end_t for conversation in conversations: logger.info(f"- {conversation['id']} {conversation['started_at']} {conversation['finished_at']}") - # get the conversation that has the closest start timestamp or end timestamp - closest_conversation = None - min_diff = float('inf') - for conversation in conversations: - conversation_start_timestamp = conversation['started_at'].timestamp() - conversation_end_timestamp = conversation['finished_at'].timestamp() - diff1 = abs(conversation_start_timestamp - start_timestamp) - diff2 = abs(conversation_end_timestamp - end_timestamp) - if diff1 < min_diff or diff2 < min_diff: - min_diff = min(diff1, diff2) - closest_conversation = conversation + closest_conversation = select_closest_conversation(conversations, start_timestamp, end_timestamp) + if closest_conversation is None: + logger.info('get_closest_conversation_to_timestamps: no eligible merge target (deleted rows excluded)') + return None logger.info(f"get_closest_conversation_to_timestamps closest_conversation: {closest_conversation['id']}") return closest_conversation diff --git a/backend/tests/fast_unit_duration_allowlist.txt b/backend/tests/fast_unit_duration_allowlist.txt index ac5b81aeb4c..b4a6a93cf09 100644 --- a/backend/tests/fast_unit_duration_allowlist.txt +++ b/backend/tests/fast_unit_duration_allowlist.txt @@ -207,3 +207,6 @@ tests/unit/test_inv_mem_1_guard.py::TestInvMemSourceRatchet::test_default_read_p tests/unit/test_task_intelligence_contract_freeze.py tests/unit/test_lock_bypass_fixes.py tests/unit/test_byok_security.py::TestCacheRouting::test_cached_openai_chat_no_raw_key_in_cache +# Behavior test drives the real process_segment target-attach fallback (#10033/#10119 review ask); +# it is the sole heavy test in its file, so it amortizes the sync pipeline import into its call phase. +tests/unit/test_sync_merge_target_selection.py::test_sync_target_attach_fallback_to_closest diff --git a/backend/tests/unit/test_sync_merge_target_selection.py b/backend/tests/unit/test_sync_merge_target_selection.py new file mode 100644 index 00000000000..5355f4d5f60 --- /dev/null +++ b/backend/tests/unit/test_sync_merge_target_selection.py @@ -0,0 +1,79 @@ +"""Synced audio must never merge into a soft-deleted conversation (#10033). + +`get_closest_conversation_to_timestamps` matched any row in the ±2min window — +including `deleted: True` tombstones — so offline audio recorded near a +conversation the user later deleted merged into the tombstone and vanished: +the "recordings never create a conversation" symptom. The closest-match choice +is now a pure selector over eligible merge targets, and the auto-sync +target-attach path consults the same predicate. +""" + +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch, MagicMock + +from database.conversations import eligible_merge_target, select_closest_conversation + +_BASE = datetime(2026, 7, 20, 12, 0, tzinfo=timezone.utc) + + +def _conversation(conversation_id: str, *, offset_seconds: int = 0, deleted: bool = False) -> dict: + started = _BASE + timedelta(seconds=offset_seconds) + return { + 'id': conversation_id, + 'started_at': started, + 'finished_at': started + timedelta(seconds=60), + 'deleted': deleted, + } + + +def test_deleted_tombstone_is_never_a_merge_target(): + target = int(_BASE.timestamp()) + only_deleted = [_conversation('gone', deleted=True)] + assert select_closest_conversation(only_deleted, target, target + 60) is None + + # A deleted row closer than a live one must lose to the live one. + rows = [_conversation('gone', deleted=True), _conversation('live', offset_seconds=90)] + chosen = select_closest_conversation(rows, target, target + 60) + assert chosen is not None and chosen['id'] == 'live' + + +def test_closest_live_conversation_wins_by_boundary_distance(): + target = int((_BASE + timedelta(seconds=95)).timestamp()) + rows = [_conversation('near', offset_seconds=90), _conversation('far', offset_seconds=600)] + chosen = select_closest_conversation(rows, target, target + 60) + assert chosen is not None and chosen['id'] == 'near' + assert select_closest_conversation([], target, target + 60) is None + + +def test_eligible_merge_target_predicate(): + assert eligible_merge_target(_conversation('live')) is True + assert eligible_merge_target(_conversation('gone', deleted=True)) is False + assert eligible_merge_target(None) is False + # Discarded rows stay eligible: the merge path reprocesses and revives them. + discarded = _conversation('quiet') + discarded['discarded'] = True + assert eligible_merge_target(discarded) is True + + +@patch('utils.sync.pipeline.get_syncing_file_temporal_signed_url', return_value='url') +@patch('utils.sync.pipeline.schedule_syncing_temporal_file_deletion') +@patch('utils.sync.pipeline.prerecorded', return_value=(['word'], 'en')) +@patch('utils.sync.pipeline.postprocess_words') +@patch('utils.sync.pipeline.conversations_db.get_conversation', return_value={'id': 'deleted', 'deleted': True}) +@patch('utils.sync.pipeline.get_timestamp_from_path', return_value=123) +@patch('utils.sync.pipeline.get_closest_conversation_to_timestamps', side_effect=RuntimeError("FALLBACK_TAKEN")) +def test_sync_target_attach_fallback_to_closest( + mock_closest, mock_timestamp, mock_get_conv, mock_postprocess, mock_prerecorded, mock_schedule, mock_signed_url +): + """Behavior-level test: if the specified target conversation is deleted/ineligible, + the pipeline must fall back to the timestamp-based closest match.""" + from utils.sync.pipeline import process_segment + + mock_postprocess.return_value = [MagicMock(end=1.0)] + + # process_segment catches exceptions internally, so it will swallow our RuntimeError and return False. + result = process_segment('seg_123.wav', 'uid', {'segments': []}, MagicMock(), [], target_conversation_id='deleted') + + assert result is False + assert mock_closest.called, "Pipeline did not fall back to get_closest_conversation_to_timestamps" diff --git a/backend/utils/sync/pipeline.py b/backend/utils/sync/pipeline.py index 959ff6f0bdb..43f9243c445 100644 --- a/backend/utils/sync/pipeline.py +++ b/backend/utils/sync/pipeline.py @@ -1059,9 +1059,9 @@ def process_segment( # attach segments to it directly instead of searching by timestamp. if target_conversation_id: closest_memory = conversations_db.get_conversation(uid, target_conversation_id) - if not closest_memory: + if not conversations_db.eligible_merge_target(closest_memory): logger.warning( - f'Target conversation {target_conversation_id} not found, falling back to timestamp lookup' + f'Target conversation {target_conversation_id} not found or deleted, falling back to timestamp lookup' ) closest_memory = get_closest_conversation_to_timestamps(uid, timestamp, segment_end_timestamp) else: