fix(sync): never merge synced audio into a soft-deleted conversation

.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
 }

backend/database/conversations.py

@@ -1,4 +1,4 @@
 import copy
 import json
 import logging
 import uuid
@@ -1524,6 +1524,34 @@
 # ********************************
 
 
+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 @@
     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

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

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"

backend/utils/sync/pipeline.py

@@ -1,4 +1,4 @@
 """Sync local-files pipeline: decode → VAD → fair-use → STT → conversation merge.
 
 Extracted from routers/sync.py so the router stays thin and utils never imports routers.
 """
@@ -937,7 +937,7 @@
             self._cond.notify_all()
 
 
 def process_segment(
     path: str,
     uid: str,
     response: dict,
@@ -1059,9 +1059,9 @@
         # 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:
@@ -1488,7 +1488,7 @@
     return vad_errors, vad_ms
 
 
 async def _run_full_pipeline_background_async(
     job_id: str,
     uid: str,
     raw_paths: list,