fix(sync): isolate fenced merged-conversation reprocessing

backend/tests/unit/test_sync_v2.py

@@ -1,4 +1,4 @@
 """
 Tests for v2 async sync-local-files endpoints (#5941, #7281).
 
 v2 saves raw files and returns 202 immediately, then runs the full pipeline
@@ -1273,7 +1273,7 @@
     mocked dependencies. Verifies actual call sequences and outcomes."""
 
     @staticmethod
     def _load_sync_module():
         """Load routers/sync.py with all heavy deps stubbed, return (module, stubs)."""
         saved_modules = {}
         stubs = {}
@@ -1514,6 +1514,34 @@
         finally:
             self._cleanup(stubs['saved_modules'])
 
+    def test_merged_reprocess_fence_isolated_to_its_conversation(self, fenced_worker_module):
+        """One replaced merge target must not supersede its sibling conversations."""
+        _module, stubs = fenced_worker_module
+        pipeline = stubs['pipeline']
+
+        class ConversationPersistenceFenced(RuntimeError):
+            pass
+
+        pipeline.SyncConversationPersistenceFenced = ConversationPersistenceFenced
+        pipeline.logger = MagicMock()
+        pipeline._reprocess_conversation_after_update = MagicMock(
+            side_effect=[ConversationPersistenceFenced('conversation replaced'), None]
+        )
+        response = {'_merged': {'replaced-conversation': 'en', 'current-conversation': 'fr'}}
+
+        pipeline._reprocess_merged_conversations('uid', response)
+
+        assert response == {}
+        assert pipeline._reprocess_conversation_after_update.call_args_list == [
+            unittest.mock.call('uid', 'replaced-conversation', 'en'),
+            unittest.mock.call('uid', 'current-conversation', 'fr'),
+        ]
+        pipeline.logger.info.assert_called_once_with(
+            'event=sync_conversation_reprocess outcome=fenced conversation_id=%s',
+            'replaced-conversation',
+        )
+        pipeline.logger.error.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_durable_completion_offloads_epoch_and_terminal_metric(self, fenced_worker_module):
         """Redis-backed fencing and telemetry never run on the async coordinator loop."""
@@ -2714,7 +2742,7 @@
     """Execute v2 endpoints using FastAPI TestClient with mocked dependencies."""
 
     @staticmethod
     def _build_test_app():
         """Build a minimal FastAPI app with the sync router, all deps mocked."""
         saved_modules = {}
         mock_sync_jobs = MagicMock()

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.
 """
@@ -990,7 +990,7 @@
             self._cond.notify_all()
 
 
 def process_segment(
     path: str,
     uid: str,
     response: dict,
@@ -1288,7 +1288,7 @@
         try:
             _reprocess_conversation_after_update(uid, conversation_id, language)
         except SyncConversationPersistenceFenced:
-            raise
+            logger.info('event=sync_conversation_reprocess outcome=fenced conversation_id=%s', conversation_id)
         except Exception as e:
             logger.error(f'sync: failed to reprocess merged conversation {conversation_id}: {e}')
 
@@ -1550,7 +1550,7 @@
     return vad_errors, vad_ms
 
 
 async def _run_full_pipeline_background_async(  # pyright: ignore[reportGeneralTypeIssues] — legacy coordinator exceeds Pyright's analyzer complexity ceiling
     job_id: str,
     uid: str,
     raw_paths: list,