fix(sync): fence post-merge audio finalization

.github/scripts/product_file_line_count_ratchet_baseline/backend-utils.json

@@ -5,12 +5,12 @@
     "backend/utils/memory_ingestion/pipeline.py": 1711,
     "backend/utils/other/storage.py": 1584,
     "backend/utils/retrieval/tools/calendar_tools.py": 1513,
-    "backend/utils/sync/pipeline.py": 2389
+    "backend/utils/sync/pipeline.py": 2467
   },
   "raise_justifications": {
     "backend/utils/apps.py": "get_available_app_model_by_id: the set-preferred-app availability authority as a validated App model, so the summary path honors what the setter admitted instead of re-deciding with the installed slice (#10074).",
     "backend/utils/other/storage.py": "delete_app_logo now requires an exact app-logo bucket prefix (startswith) so a foreign URL embedding the prefix later cannot delete an unrelated object on this deletion path (David review on #9873).",
-    "backend/utils/sync/pipeline.py": "Plan-aware soft-cap evaluation and its default-tier fallback remain in a small helper beside Sync's idempotent fresh-speech metering rather than overgrowing the durable worker coordinator. The same coordinator owns lifecycle-fence detection and the run-token-fenced superseded finalizer that prevents stale WAL retries."
+    "backend/utils/sync/pipeline.py": "Plan-aware soft-cap evaluation and its default-tier fallback remain in a small helper beside Sync's idempotent fresh-speech metering rather than overgrowing the durable worker coordinator. A lifecycle-fenced deferred reprocess must durably tombstone its conversation before audio side effects; retry hydration must preserve that ownership boundary. Fenced IDs are now discarded from new_memories too, because a conversation created in an earlier segment can be merged into by a later segment."
   },
   "threshold": 1500
 }

backend/testing/e2e/test_core_flow_expansion.py

@@ -221,7 +221,7 @@ def fake_process_segment(
     monkeypatch.setattr(sync_pipeline, "retrieve_vad_segments", fake_retrieve_vad_segments)
     monkeypatch.setattr(sync_pipeline, "get_wav_duration", lambda path: 2.0)
     monkeypatch.setattr(sync_pipeline, "process_segment", fake_process_segment)
-    monkeypatch.setattr(sync_pipeline, "_reprocess_merged_conversations", lambda uid, response: None)
+    monkeypatch.setattr(sync_pipeline, "_reprocess_merged_conversations", lambda uid, response, on_fenced=None: None)
     monkeypatch.setattr(sync_pipeline, "build_person_embeddings_cache", lambda uid: {})
     monkeypatch.setattr(sync_router, "get_hard_restriction_status", lambda uid: (False, None))
     monkeypatch.setattr(sync_router, "has_transcription_credits", lambda uid: True)

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
@@ -1006,10 +1006,11 @@
         assert 'build_person_embeddings_cache' in _get_pipeline_async_function_body('_load_sync_segment_context')
 
     def test_no_thread_pool_slot_held_for_coordinator(self):
-        """Async coordinator must NOT use submit_with_context or hold a thread pool slot."""
+        """Async coordinator must not submit itself to a thread pool slot."""
         body = self._get_bg_func_body()
         assert 'submit_with_context' not in body, "Async coordinator must not use submit_with_context"
-        assert '.result(' not in body, "Async coordinator must not call future.result()"
+        assert body.count('.result(') == 1, "Only the sync-worker fence callback may synchronously await a future"
+        assert 'asyncio.run_coroutine_threadsafe(' in body
 
 
 # ---------------------------------------------------------------------------
@@ -1273,7 +1274,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 = {}
@@ -1527,11 +1528,33 @@
         pipeline._reprocess_conversation_after_update = MagicMock(
             side_effect=[ConversationPersistenceFenced('conversation replaced'), None]
         )
-        response = {'_merged': {'replaced-conversation': 'en', 'current-conversation': 'fr'}}
+        response = {
+            '_merged': {'replaced-conversation': 'en', 'current-conversation': 'fr'},
+            'updated_memories': {'replaced-conversation', 'current-conversation'},
+            # A conversation can appear in both new_memories (created in an
+            # earlier segment) and updated_memories (merged into by a later
+            # segment).  The fence must discard it from both.
+            'new_memories': {'replaced-conversation'},
+        }
+        checkpointed_fences = []
 
-        pipeline._reprocess_merged_conversations('uid', response)
+        pipeline._reprocess_merged_conversations(
+            'uid',
+            response,
+            on_fenced=lambda: checkpointed_fences.append(
+                {
+                    'fenced': set(response['_fenced_conversation_ids']),
+                    'updated': set(response['updated_memories']),
+                }
+            ),
+        )
 
-        assert response == {}
+        assert response == {
+            '_fenced_conversation_ids': {'replaced-conversation'},
+            'updated_memories': {'current-conversation'},
+            'new_memories': set(),
+        }
+        assert checkpointed_fences == [{'fenced': {'replaced-conversation'}, 'updated': {'current-conversation'}}]
         assert pipeline._reprocess_conversation_after_update.call_args_list == [
             unittest.mock.call('uid', 'replaced-conversation', 'en'),
             unittest.mock.call('uid', 'current-conversation', 'fr'),
@@ -1542,6 +1565,20 @@
         )
         pipeline.logger.error.assert_not_called()
 
+        audio_file = MagicMock()
+        audio_file.model_dump.return_value = {'path': 'current.opus'}
+        pipeline.conversations_db = MagicMock()
+        pipeline.conversations_db.create_audio_files_from_chunks.return_value = [audio_file]
+        pipeline.precache_conversation_audio = MagicMock()
+        pipeline.is_audio_merge_dispatch_enabled = MagicMock(return_value=False)
+
+        pipeline._finalize_sync_audio_files('uid', response)
+
+        pipeline.conversations_db.create_audio_files_from_chunks.assert_called_once_with('uid', 'current-conversation')
+        pipeline.conversations_db.update_conversation.assert_called_once_with(
+            'uid', 'current-conversation', {'audio_files': [{'path': 'current.opus'}]}
+        )
+
     @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."""
@@ -2581,6 +2618,189 @@
         finally:
             self._cleanup(stubs['saved_modules'])
 
+    @pytest.mark.asyncio
+    async def test_retry_hydration_excludes_durably_fenced_conversation_from_audio_finalization(self):
+        """A fence recorded before a retry cannot let its stale checkpoint finalize audio."""
+        module, stubs = self._load_sync_module()
+        try:
+            segment_path = '/tmp/seg_1700000001.wav'
+            pipeline = stubs['pipeline']
+            pipeline.decode_files_to_wav = MagicMock(return_value=['/tmp/w.wav'])
+            pipeline._cleanup_files = MagicMock()
+            pipeline.retrieve_vad_segments = lambda _path, paths, _errors: paths.add(segment_path)
+            pipeline.get_wav_duration = MagicMock(return_value=5.0)
+            pipeline.users_db.get_user_transcription_preferences = MagicMock(return_value={})
+            pipeline.users_db.get_user_private_cloud_sync_enabled = MagicMock(return_value=True)
+            pipeline.users_db.get_data_protection_level = MagicMock(return_value=None)
+            pipeline.build_person_embeddings_cache = MagicMock(return_value={})
+            pipeline.get_sync_job = MagicMock(
+                return_value={
+                    'partial_result': {
+                        'updated_memories': ['fenced-conversation', 'current-conversation'],
+                        'new_memories': ['fenced-conversation'],
+                    }
+                }
+            )
+            pipeline.get_sync_content_partial_result = MagicMock(
+                return_value={'fenced_conversation_ids': ['fenced-conversation']}
+            )
+            pipeline.get_processed_sync_segment_ids = MagicMock(return_value={'segment-1'})
+            pipeline.compute_sync_segment_id = MagicMock(return_value='segment-1')
+            pipeline.process_segment = MagicMock()
+            pipeline._reprocess_merged_conversations = MagicMock()
+            pipeline._finalize_sync_audio_files = MagicMock()
+
+            await module._run_full_pipeline_background_async(
+                'job-retry-fenced',
+                'uid',
+                ['/tmp/f.opus'],
+                'omi',
+                False,
+                '/tmp/job-retry-fenced',
+                task_mode=True,
+                content_id='content-retry-fenced',
+                content_run_bound=True,
+                ledger_fence_active=False,
+            )
+
+            pipeline.process_segment.assert_not_called()
+            pipeline._finalize_sync_audio_files.assert_called_once()
+            finalized_response = pipeline._finalize_sync_audio_files.call_args.args[1]
+            assert finalized_response['updated_memories'] == {'current-conversation'}
+            assert 'fenced-conversation' not in finalized_response['updated_memories']
+            assert 'fenced-conversation' not in finalized_response['new_memories']
+        finally:
+            self._cleanup(stubs['saved_modules'])
+
+    @pytest.mark.asyncio
+    async def test_fence_checkpoint_handoffs_to_db_executor_before_audio_finalization(self):
+        """A sync-worker fence persists ledger then job state through the coordinator loop."""
+        module, stubs = self._load_sync_module()
+        sync_worker = ThreadPoolExecutor(max_workers=1)
+        try:
+            segment_path = '/tmp/seg_1700000001.wav'
+            pipeline = stubs['pipeline']
+            checkpoint_events = []
+            db_offload_calls = []
+            pipeline.decode_files_to_wav = MagicMock(return_value=['/tmp/w.wav'])
+            pipeline._cleanup_files = MagicMock()
+            pipeline.retrieve_vad_segments = lambda _path, paths, _errors: paths.add(segment_path)
+            pipeline.get_wav_duration = MagicMock(return_value=5.0)
+            pipeline.users_db.get_user_transcription_preferences = MagicMock(return_value={})
+            pipeline.users_db.get_user_private_cloud_sync_enabled = MagicMock(return_value=True)
+            pipeline.users_db.get_data_protection_level = MagicMock(return_value=None)
+            pipeline.build_person_embeddings_cache = MagicMock(return_value={})
+            pipeline.get_sync_job = MagicMock(return_value={'partial_result': {}})
+            pipeline.get_sync_content_partial_result = MagicMock(return_value={})
+            pipeline.get_processed_sync_segment_ids = MagicMock(return_value=set())
+            pipeline.compute_sync_segment_id = MagicMock(return_value='segment-1')
+            pipeline.process_segment = MagicMock(return_value=False)
+            pipeline.checkpoint_sync_content_partial_result = MagicMock(
+                side_effect=lambda *_args, **_kwargs: checkpoint_events.append('durable') or True
+            )
+
+            def _record_job_partial(*args):
+                if 'partial_result' in args[2]:
+                    checkpoint_events.append('job')
+
+            pipeline._update_sync_job_for_run = MagicMock(side_effect=_record_job_partial)
+
+            def _fence_in_sync_worker(_uid, response, on_fenced):
+                response['_fenced_conversation_ids'].add('fenced-conversation')
+                on_fenced()
+
+            pipeline._reprocess_merged_conversations = _fence_in_sync_worker
+            pipeline._finalize_sync_audio_files = MagicMock()
+
+            async def _threaded_run_blocking(executor, fn, *args, **kwargs):
+                if executor is pipeline.sync_executor:
+                    return await _production_run_blocking(sync_worker, fn, *args, **kwargs)
+                if executor is pipeline.db_executor and (
+                    fn is pipeline.checkpoint_sync_content_partial_result
+                    or (fn is pipeline._update_sync_job_for_run and 'partial_result' in args[2])
+                ):
+                    db_offload_calls.append(fn)
+                return fn(*args, **kwargs)
+
+            pipeline.run_blocking = _threaded_run_blocking
+
+            await module._run_full_pipeline_background_async(
+                'job-fence-checkpoint',
+                'uid',
+                ['/tmp/f.opus'],
+                'omi',
+                False,
+                '/tmp/job-fence-checkpoint',
+                task_mode=True,
+                content_id='content-fence-checkpoint',
+                content_run_bound=True,
+            )
+
+            assert db_offload_calls == [
+                pipeline.checkpoint_sync_content_partial_result,
+                pipeline._update_sync_job_for_run,
+            ]
+            assert checkpoint_events == ['durable', 'job']
+            pipeline._finalize_sync_audio_files.assert_called_once()
+        finally:
+            sync_worker.shutdown(wait=True)
+            self._cleanup(stubs['saved_modules'])
+
+    @pytest.mark.asyncio
+    async def test_fence_checkpoint_lease_loss_stops_audio_finalization(self):
+        """A durable fence checkpoint failure reaches the sync callback before side effects."""
+        module, stubs = self._load_sync_module()
+        sync_worker = ThreadPoolExecutor(max_workers=1)
+        try:
+            segment_path = '/tmp/seg_1700000001.wav'
+            pipeline = stubs['pipeline']
+            pipeline.decode_files_to_wav = MagicMock(return_value=['/tmp/w.wav'])
+            pipeline._cleanup_files = MagicMock()
+            pipeline.retrieve_vad_segments = lambda _path, paths, _errors: paths.add(segment_path)
+            pipeline.get_wav_duration = MagicMock(return_value=5.0)
+            pipeline.users_db.get_user_transcription_preferences = MagicMock(return_value={})
+            pipeline.users_db.get_user_private_cloud_sync_enabled = MagicMock(return_value=True)
+            pipeline.users_db.get_data_protection_level = MagicMock(return_value=None)
+            pipeline.build_person_embeddings_cache = MagicMock(return_value={})
+            pipeline.get_sync_job = MagicMock(return_value={'partial_result': {}})
+            pipeline.get_sync_content_partial_result = MagicMock(return_value={})
+            pipeline.get_processed_sync_segment_ids = MagicMock(return_value=set())
+            pipeline.compute_sync_segment_id = MagicMock(return_value='segment-1')
+            pipeline.process_segment = MagicMock(return_value=False)
+            pipeline.checkpoint_sync_content_partial_result = MagicMock(return_value=False)
+
+            def _fence_in_sync_worker(_uid, response, on_fenced):
+                response['_fenced_conversation_ids'].add('fenced-conversation')
+                on_fenced()
+
+            pipeline._reprocess_merged_conversations = _fence_in_sync_worker
+            pipeline._finalize_sync_audio_files = MagicMock()
+
+            async def _threaded_run_blocking(executor, fn, *args, **kwargs):
+                if executor is pipeline.sync_executor:
+                    return await _production_run_blocking(sync_worker, fn, *args, **kwargs)
+                return fn(*args, **kwargs)
+
+            pipeline.run_blocking = _threaded_run_blocking
+
+            with pytest.raises(pipeline.SyncJobRunLeaseLost):
+                await module._run_full_pipeline_background_async(
+                    'job-fence-lease-lost',
+                    'uid',
+                    ['/tmp/f.opus'],
+                    'omi',
+                    False,
+                    '/tmp/job-fence-lease-lost',
+                    task_mode=True,
+                    content_id='content-fence-lease-lost',
+                    content_run_bound=True,
+                )
+
+            pipeline._finalize_sync_audio_files.assert_not_called()
+        finally:
+            sync_worker.shutdown(wait=True)
+            self._cleanup(stubs['saved_modules'])
+
     @pytest.mark.asyncio
     async def test_person_embeddings_fallback(self):
         """Person embeddings failure must fall back to empty dict, not crash."""
@@ -2742,7 +2962,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()