fix(conversations): reject soft-deleted conversations from merge

backend/tests/unit/test_merge_deleted_source_race.py

@@ -0,0 +1,77 @@
+"""A source soft-deleted after admission must not be merged by the background worker.
+
+validate_merge_compatibility rejects a soft-deleted source at admission, but
+perform_merge_async re-fetches the sources later in the background task. If a source
+becomes a soft-deleted tombstone between admission and that fetch (the delete-vs-merge
+race), the worker must abort — otherwise it resurrects the deleted source's
+transcript/photos/audio into a new visible conversation and then re-deletes the sources.
+"""
+
+import os
+
+os.environ.setdefault("ENCRYPTION_SECRET", "omi_ZwB2ZNqB2HHpMK6wStk7sTpavJiPTFg7gXUHnc4tFABPU6pZ2c2DKgehtfgi4RZv")
+os.environ.setdefault("OPENAI_API_KEY", "sk-test-not-real")
+
+from unittest.mock import MagicMock
+
+import pytest
+
+import utils.conversations.merge_conversations as merge
+
+
+@pytest.fixture(scope='module', autouse=True)
+def _warm_merge_imports():
+    """perform_merge_async does heavy local imports (process_conversation) at call
+    time. Charge them to module setup so the fast-unit CPU-time guard measures only
+    the call phase (mirrors test_modulate_stt.py's warm fixture)."""
+    import utils.conversations.process_conversation  # noqa: F401
+    import utils.notifications  # noqa: F401
+
+
+def _install(monkeypatch, sources):
+    monkeypatch.setattr(merge.conversations_db, "get_conversation", lambda uid, cid: sources.get(cid))
+    fail = MagicMock()
+    monkeypatch.setattr(merge, "_handle_merge_failure", fail)
+    create = MagicMock()
+    monkeypatch.setattr(merge.lifecycle_service, "create_processing_conversation", create)
+    return fail, create
+
+
+def test_merge_aborts_when_a_source_is_deleted_after_admission(monkeypatch):
+    # c1 became a soft-deleted tombstone between admission and this background fetch.
+    fail, create = _install(
+        monkeypatch,
+        {
+            'c1': {'id': 'c1', 'status': 'completed', 'deleted': True},
+            'c2': {'id': 'c2', 'status': 'completed'},
+        },
+    )
+
+    merge.perform_merge_async('u1', ['c1', 'c2'])
+
+    fail.assert_called_once()
+    # The deleted source's content never reached a new visible conversation.
+    create.assert_not_called()
+
+
+def test_merge_does_not_abort_when_no_source_is_deleted(monkeypatch):
+    # The guard must not reject a legitimate merge: two live sources reach the build step
+    # (create_processing_conversation) rather than _handle_merge_failure. Downstream build
+    # is stubbed to raise right there so the test stays on the guard boundary.
+    fail, create = _install(
+        monkeypatch,
+        {
+            'c1': {'id': 'c1', 'status': 'completed', 'started_at': None},
+            'c2': {'id': 'c2', 'status': 'completed', 'started_at': None},
+        },
+    )
+    monkeypatch.setattr(merge, "_normalize_conversation_timestamps", lambda convs: convs)
+    monkeypatch.setattr(merge, "_merge_transcript_segments", lambda convs: [])
+    monkeypatch.setattr(merge, "_collect_all_photos", lambda uid, convs: [])
+    monkeypatch.setattr(merge, "_copy_audio_chunks_for_merge", lambda uid, convs, new_id: [])
+    monkeypatch.setattr(merge, "Conversation", lambda **kw: MagicMock(**{'model_dump.return_value': {}}))
+    create.side_effect = RuntimeError("stop at build; the guard already let us through")
+
+    merge.perform_merge_async('u1', ['c1', 'c2'])
+
+    create.assert_called_once()  # got past the deleted guard into the build step

backend/tests/unit/test_merge_validation.py

@@ -187,13 +187,14 @@ def test_unsupported_type_returns_none(self, merge):
 # ---------------------------------------------------------------------------
 
 
-def _conv(conv_id="c1", started=None, finished=None, status="completed", locked=False):
+def _conv(conv_id="c1", started=None, finished=None, status="completed", locked=False, deleted=False):
     return {
         "id": conv_id,
         "started_at": started,
         "finished_at": finished,
         "status": status,
         "is_locked": locked,
+        "deleted": deleted,
     }
 
 
@@ -239,6 +240,22 @@ def test_rejects_locked_conversation(self, merge):
         assert ok is False
         assert "locked" in err.lower()
 
+    def test_rejects_deleted_conversation(self, merge):
+        # A soft-deleted tombstone must never be a merge source: merging it
+        # resurrects deleted content into a new visible conversation (#10119
+        # guards the sync merge path; this guards the user-initiated merge).
+        convs = [_conv("c1", deleted=True), _conv("c2")]
+        ok, err, warn = merge.validate_merge_compatibility(convs)
+        assert ok is False
+        assert "deleted" in err.lower()
+        assert warn is None
+
+    def test_allows_non_deleted_conversations(self, merge):
+        # Baseline: the deleted guard must not reject ordinary sources.
+        ok, err, warn = merge.validate_merge_compatibility([_conv("c1"), _conv("c2")])
+        assert ok is True
+        assert err is None
+
     def test_rejects_non_completed_conversation(self, merge):
         convs = [_conv("c1"), _conv("c2", status="processing")]
         ok, err, warn = merge.validate_merge_compatibility(convs)

backend/utils/conversations/merge_conversations.py

@@ -107,6 +107,7 @@
 
     Rejection criteria (hard failures):
     - Less than 2 conversations
+    - Any conversation is a soft-deleted tombstone
     - Any conversation is locked
     - Any conversation is not completed (processing/merging/in_progress)
 
@@ -116,6 +117,17 @@
     if len(conversations) < 2:
         return False, "At least 2 conversations required to merge", None
 
+    # Check none are soft-deleted. A soft-deleted tombstone is invisible to the
+    # user, so merging it resurrects deleted content into a new visible
+    # conversation — the inverse of the tombstone contract the sync merge path
+    # already enforces (see conversations_db.eligible_merge_target, #10119).
+    # `get_conversation` returns tombstones unfiltered and the /merge endpoint
+    # only 404s on a missing (None) doc, so a deleted id passed by an API client
+    # or a delete-vs-merge race would otherwise flow straight through.
+    for conv in conversations:
+        if conv.get('deleted'):
+            return False, "Cannot merge a deleted conversation.", None
+
     # Check none are locked
     for conv in conversations:
         if conv.get('is_locked', False):
@@ -144,7 +156,7 @@
     return True, None, warning_msg
 
 
 def perform_merge_async(
     uid: str,
     conversation_ids: List[str],
     reprocess: bool = True,
@@ -182,6 +194,15 @@
             _handle_merge_failure(uid, conversation_ids)
             return
 
+        # A source can be soft-deleted between admission (validate_merge_compatibility
+        # at the endpoint) and this background re-fetch — the delete-vs-merge race. Re-check
+        # here, before reading any content: merging a tombstone would resurrect its deleted
+        # transcript/photos/audio into a new visible conversation. Abort rather than merge.
+        if any(conv.get('deleted') for conv in conversations):
+            logger.error(f"Merge aborted: a source was deleted after admission uid={uid}")
+            _handle_merge_failure(uid, conversation_ids)
+            return
+
         # Normalise timestamp fields once so the sort key, max() reducer,
         # .isoformat() metadata, and _merge_transcript_segments arithmetic
         # below can all assume tz-aware datetimes regardless of how each