From 8f6ba917104fe98a6d4cc6f3b51bc31082d1500f Mon Sep 17 00:00:00 2001
From: Aakash Kattelu <aakash@plasticlabs.ai>
Date: Wed, 19 Aug 2026 11:48:08 -0400
Subject: [PATCH] fix(deriver): harden create_documents candidate hoist and
 test isolation

Skip empty embeddings on the external-store path, isolate per-document
resolve failures, and keep replacement times_derived in the in-batch
ledger. Patch get_external_vector_store in the hoist test and cover
in-loop SQLAlchemyError abort.
---
 src/crud/document.py        |  8 +++--
 tests/crud/test_document.py | 60 ++++++++++++++++++++++++++++++++++++-
 2 files changed, 64 insertions(+), 4 deletions(-)

diff --git a/src/crud/document.py b/src/crud/document.py
index 2261479f..892925c4 100644
--- a/src/crud/document.py
+++ b/src/crud/document.py
@@ -544,7 +544,7 @@ async def create_documents(
 
         async def _resolve_candidates(index: int, doc: schemas.DocumentCreate) -> None:
             filters = _semantic_dup_filters(doc)
-            if filters is None:
+            if filters is None or not doc.embedding:
                 return
             semantic_candidates[index] = await query_external_vector_document_ids(
                 workspace_name=workspace_name,
@@ -557,7 +557,8 @@ async def create_documents(
             )
 
         await asyncio.gather(
-            *(_resolve_candidates(i, doc) for i, doc in enumerate(documents))
+            *(_resolve_candidates(i, doc) for i, doc in enumerate(documents)),
+            return_exceptions=True,
         )
 
     # exact-content dedup (independent of `deduplicate`): pre-fetch
@@ -674,6 +675,7 @@ async def create_documents(
                         existing_dup.id, existing_dup.times_derived
                     )
                     doc.times_derived = max(doc.times_derived, current_td + 1)
+                    pending_times_derived[existing_dup.id] = doc.times_derived
                     row_ops.append(_DocumentRowOp("replace", existing_dup.id))
                     semantic_dup_replaced_count += 1
                 elif (
@@ -1312,7 +1314,7 @@ async def _semantic_dup_decision(
             filters=filters,
             max_distance=_SEMANTIC_DUP_MAX_DISTANCE,
             top_k=_SEMANTIC_DUP_TOP_K,
-            embedding=doc.embedding,
+            embedding=doc.embedding or None,
         )
 
     if not similar_docs:
diff --git a/tests/crud/test_document.py b/tests/crud/test_document.py
index 65059791..2e4e4e94 100644
--- a/tests/crud/test_document.py
+++ b/tests/crud/test_document.py
@@ -1499,7 +1499,7 @@ class TestCreateDocumentsErrorHandling:
         )
 
     @pytest.mark.asyncio
-    async def test_db_error_in_loop_aborts_batch(
+    async def test_db_error_on_row_update_flush_aborts_batch(
         self,
         db_session: AsyncSession,
         sample_data: tuple[models.Workspace, models.Peer],
@@ -1558,6 +1558,60 @@ class TestCreateDocumentsErrorHandling:
         assert [d.content for d in docs] == ["existing fact"]
         assert docs[0].times_derived == 1
 
+    @pytest.mark.asyncio
+    async def test_db_error_in_loop_aborts_batch(
+        self,
+        db_session: AsyncSession,
+        sample_data: tuple[models.Workspace, models.Peer],
+    ):
+        """A DB error during per-document classification raises and commits nothing."""
+        test_workspace, test_peer = sample_data
+        test_peer2, test_session = await self._setup(
+            db_session, test_workspace, test_peer
+        )
+        workspace_name = test_workspace.name
+        observer = test_peer.name
+        observed = test_peer2.name
+        session_name = test_session.name
+
+        class FakePGError(Exception):
+            sqlstate: str = "40P01"
+
+        deadlock = OperationalError("SELECT documents", {}, FakePGError())
+        with (
+            patch(
+                "src.crud.document._semantic_dup_decision",
+                AsyncMock(side_effect=deadlock),
+            ),
+            pytest.raises(OperationalError),
+        ):
+            await crud.create_documents(
+                db_session,
+                [
+                    self._doc("a brand new fact", session_name),
+                    self._doc("another new fact", session_name),
+                ],
+                workspace_name=workspace_name,
+                observer=observer,
+                observed=observed,
+                deduplicate=True,
+            )
+
+        docs = (
+            (
+                await db_session.execute(
+                    select(models.Document).where(
+                        models.Document.workspace_name == workspace_name,
+                        models.Document.observer == observer,
+                        models.Document.observed == observed,
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert docs == []
+
     @pytest.mark.asyncio
     async def test_per_document_error_still_skips_only_that_document(
         self,
@@ -1671,6 +1725,10 @@ class TestExternalCandidateHoist:
                 "src.crud.document.query_external_vector_document_ids",
                 side_effect=fake_resolve,
             ),
+            patch(
+                "src.crud.document.get_external_vector_store",
+                return_value=None,
+            ),
         ):
             result = await crud.create_documents(
                 db_session,