fix(llm): scope chunk deletes to collection; honest resume docstring (refs #567)

src/llm/index.py

@@ -1,10 +1,12 @@
 """Incremental, resumable embedding indexer.
 
 Per doc: compare ``content_hash`` against ``index_state``; skip when
-unchanged, else delete the item's old chunks (by ``cmetadata->>
-'item_key'``), embed via the host pool, upsert with deterministic ids,
-and record the new hash — one transaction per doc, so a killed run
-resumes exactly where it stopped.
+unchanged, else delete the item's old chunks in this collection (by
+``cmetadata->>'item_key'`` + ``collection_id``), embed via the host
+pool, upsert with deterministic ids, and record the new hash — three
+commits per doc, ordered embed → delete → add → record-state, so a
+killed run reprocesses any half-done doc on the next pass (its hash was
+never recorded) rather than losing work.
 
 The metadata delete runs after ``vectorstore()`` has already constructed
 the ``PGVector`` store: the installed ``langchain-postgres==0.0.17``
@@ -59,15 +61,25 @@ def _state(engine: Engine, collection: str) -> dict[str, str]:
     return dict(rows)
 
 
-def _delete_old_chunks(engine: Engine, item_key: str) -> None:
+def _delete_old_chunks(engine: Engine, item_key: str, collection: str) -> None:
+    """Delete *item_key*'s chunks in *collection* only.
+
+    ``index_state``'s PK is (item_key, collection): the same item may be
+    indexed into several collections, so an unscoped delete-by-item_key
+    would silently drop another collection's chunks. Column names match
+    langchain-postgres 0.0.17: ``langchain_pg_collection.uuid`` is the PK
+    that ``langchain_pg_embedding.collection_id`` references.
+    """
     try:
         with engine.begin() as conn:
             conn.execute(
                 text(
                     "DELETE FROM langchain_pg_embedding "
-                    "WHERE cmetadata->>'item_key' = :k"
+                    "WHERE cmetadata->>'item_key' = :k "
+                    "AND collection_id = (SELECT uuid "
+                    "FROM langchain_pg_collection WHERE name = :c)"
                 ),
-                {"k": item_key},
+                {"k": item_key, "c": collection},
             )
     except ProgrammingError:
         # Fresh database, store tables not created yet — nothing to delete.
@@ -99,7 +111,7 @@ def index_docs(
             stats["skipped"] += 1
             continue
         vectors = embed_texts(pool, cfg.embed_model, [c.text for c in chunks])
-        _delete_old_chunks(engine, doc.key)
+        _delete_old_chunks(engine, doc.key, collection)
         store.add_embeddings(
             texts=[c.text for c in chunks],
             embeddings=vectors,

tests/llm/test_index.py

@@ -60,8 +60,20 @@ class TestIndexDocs:
     def test_changed_doc_deletes_old_chunks_first(self):
         stats, store, conn = _run([DOC], state_rows=[("K1", "stalehash")])
         assert stats["indexed"] == 1
-        deletes = " ".join(str(c.args[0]) for c in conn.execute.call_args_list)
-        assert "item_key" in deletes  # DELETE ... cmetadata->>'item_key'
+        deletes = [
+            c
+            for c in conn.execute.call_args_list
+            if "DELETE FROM langchain_pg_embedding" in str(c.args[0])
+        ]
+        assert len(deletes) == 1
+        sql = str(deletes[0].args[0])
+        assert "item_key" in sql  # DELETE ... cmetadata->>'item_key'
+        # Scoped to the target collection, not a global delete by item_key.
+        assert "collection_id" in sql
+        assert "langchain_pg_collection" in sql
+        params = deletes[0].args[1]
+        assert params["k"] == "K1"
+        assert params["c"] == "comments"
 
     def test_force_reembeds_unchanged(self):
         h = content_hash(DOC.text)