feat(api): retry indexing saved documents without duplicate chunks

.tmp/proof/document-index-retry.txt

@@ -0,0 +1,124 @@
+$ env -i HOME="$HOME" PATH="$PATH" UV_PYTHON=3.13 UV_FROZEN=1 MISE_YES=1 SLACK_BOT_TOKEN=xoxb-fixture-only SLACK_SIGNING_SECRET=fixture-only DATABASE_URL=sqlite:///.tmp-tests.db PYTEST_ADDOPTS=-rP mise run test-backend
+tests/test_document_indexing.py::test_create_and_retry_keep_one_owned_chunk_and_preserve_saved_record PASSED [ 94%]
+tests/test_document_indexing.py::test_retry_recovers_saved_document_after_embedding_failure_with_transcript PASSED [ 94%]
+tests/test_document_indexing.py::test_retry_rechunks_and_removes_only_obsolete_owned_chunks PASSED [ 95%]
+tests/test_document_indexing.py::test_retry_reads_latest_saved_content_title_and_source PASSED [ 95%]
+tests/test_document_indexing.py::test_retry_failure_keeps_previous_chunks_and_saved_content[lookup] PASSED [ 96%]
+tests/test_document_indexing.py::test_retry_failure_keeps_previous_chunks_and_saved_content[embed] PASSED [ 96%]
+tests/test_document_indexing.py::test_retry_failure_keeps_previous_chunks_and_saved_content[upsert] PASSED [ 97%]
+tests/test_document_indexing.py::test_cleanup_failure_is_reported_and_retry_finishes_without_duplicates PASSED [ 97%]
+tests/test_document_indexing.py::test_missing_saved_document_does_not_initialise_vector_store PASSED [ 98%]
+tests/test_document_indexing.py::test_invalid_legacy_document_is_rejected_without_changing_chunks[content- \n\t ] PASSED [ 98%]
+tests/test_document_indexing.py::test_invalid_legacy_document_is_rejected_without_changing_chunks[content-None] PASSED [ 99%]
+tests/test_document_indexing.py::test_invalid_legacy_document_is_rejected_without_changing_chunks[title-] PASSED [ 99%]
+tests/test_document_indexing.py::test_invalid_legacy_document_is_rejected_without_changing_chunks[source-None] PASSED [100%]
+====================== 209 passed, 928 warnings in 6.48s =======================
+
+Captured stdout from test_retry_recovers_saved_document_after_embedding_failure_with_transcript
+{
+  "query": "How do I reset VPN?",
+  "document_id": 1,
+  "top_k": 5,
+  "before_retry": [],
+  "after_two_retries": [
+    {
+      "content": "Reset VPN via IT portal.",
+      "metadata": {
+        "chunk": 0,
+        "document_id": 1,
+        "source": "api",
+        "title": "VPN guide"
+      },
+      "distance": 1.0
+    }
+  ],
+  "chunk_count": 1,
+  "note": "Fixed-vector squared L2 distances prove recovery and identity, not production relevance."
+}
+
+$ mise run dev-backend (with a local SQLite URL, fixed Chroma embeddings and a Slack-denying sitecustomize hook)
+INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
+INFO:     Application startup complete.
+INFO:     Application shutdown complete.
+
+$ curl -sS --max-time 10 -i http://127.0.0.1:8000/health
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:52 GMT
+server: uvicorn
+content-length: 20
+content-type: application/json
+
+{"status":"healthy"}
+
+$ curl -sS --max-time 10 -i http://127.0.0.1:8000/api/retrieval/preview -H 'Content-Type: application/json' -d '{"question": "How do I reset VPN?"}'
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:52 GMT
+server: uvicorn
+content-length: 93
+content-type: application/json
+
+{"question":"How do I reset VPN?","channel_id":null,"top_k":5,"max_distance":null,"items":[]}
+
+$ curl -sS --max-time 10 -i http://127.0.0.1:8000/api/documents -H 'Content-Type: application/json' -d '{"title": "VPN guide", "content": "Reset VPN via IT portal."}'
+HTTP/1.1 503 Service Unavailable
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 83
+content-type: application/json
+
+{"detail":{"message":"Document saved but indexing is unavailable","document_id":1}}
+
+$ curl -sS --max-time 10 -i http://127.0.0.1:8000/api/documents/1
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 164
+content-type: application/json
+
+{"id":1,"title":"VPN guide","content":"Reset VPN via IT portal.","source":"api","created_at":"2026-11-15T02:22:53.268264","updated_at":"2026-11-15T02:22:53.268271"}
+
+$ curl -sS --max-time 10 -i -X POST http://127.0.0.1:8000/api/documents/1/index
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 50
+content-type: application/json
+
+{"id":1,"message":"Document indexed successfully"}
+
+$ curl -sS --max-time 10 -i -X POST http://127.0.0.1:8000/api/documents/1/index
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 50
+content-type: application/json
+
+{"id":1,"message":"Document indexed successfully"}
+
+$ curl -sS --max-time 10 -i http://127.0.0.1:8000/api/retrieval/preview -H 'Content-Type: application/json' -d '{"question": "How do I reset VPN?"}'
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 220
+content-type: application/json
+
+{"question":"How do I reset VPN?","channel_id":null,"top_k":5,"max_distance":null,"items":[{"content":"Reset VPN via IT portal.","metadata":{"chunk":0,"document_id":1,"source":"api","title":"VPN guide"},"distance":1.0}]}
+
+$ curl -sS --max-time 10 -i http://127.0.0.1:8000/api/documents
+HTTP/1.1 200 OK
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 174
+content-type: application/json
+
+{"items":[{"id":1,"title":"VPN guide","source":"api","created_at":"2026-11-15T02:22:53.268264","content_preview":"Reset VPN via IT portal."}],"total":1,"limit":50,"offset":0}
+
+$ curl -sS --max-time 10 -i -X POST http://127.0.0.1:8000/api/documents/999/index
+HTTP/1.1 404 Not Found
+date: Sun, 15 Nov 2026 02:22:53 GMT
+server: uvicorn
+content-length: 31
+content-type: application/json
+
+{"detail":"Document not found"}
+

backend/app/main.py

@@ -6,7 +6,7 @@
 
 from fastapi import Depends, FastAPI, HTTPException, Query
 from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel, StringConstraints, field_validator
+from pydantic import BaseModel, StringConstraints, ValidationError, field_validator
 from sqlalchemy import func, or_
 from sqlalchemy.orm import Session
 from starlette.concurrency import run_in_threadpool
@@ -260,18 +260,29 @@ async def add_document(doc: DocumentRequest, db: Session = Depends(get_db)):
         db.add(db_doc)
         db.commit()
         db.refresh(db_doc)
-
-        # Index in vector store
-        rag_engine.index_document(doc.title, doc.content, doc.source)
-
-        return {
-            "id": db_doc.id,
-            "message": "Document added successfully"
-        }
+
     except Exception as e:
+        db.rollback()
         logger.error(f"Error adding document: {e}")
         raise HTTPException(status_code=500, detail=str(e))
 
+    await _index_saved_document(db_doc.id, doc)
+    return {"id": db_doc.id, "message": "Document added successfully"}
+
+
+async def _index_saved_document(document_id: int, doc: DocumentRequest):
+    try:
+        await run_in_threadpool(
+            rag_engine.index_document, doc.title, doc.content, doc.source,
+            document_id=document_id,
+        )
+    except Exception:
+        logger.exception("Indexing saved document %s failed", document_id)
+        raise HTTPException(status_code=503, detail={
+            "message": "Document saved but indexing is unavailable",
+            "document_id": document_id,
+        })
+
 
 @app.get("/api/documents", response_model=DocumentListResponse)
 async def list_documents(
@@ -327,6 +338,22 @@ async def get_document(document_id: int, db: Session = Depends(get_db)):
     return doc
 
 
+@app.post("/api/documents/{document_id}/index")
+async def index_saved_document(document_id: int, db: Session = Depends(get_db)):
+    """Retry indexing saved content without creating another document record."""
+    doc = db.query(Document).filter(Document.id == document_id).first()
+    if doc is None:
+        raise HTTPException(status_code=404, detail="Document not found")
+    try:
+        validated = DocumentRequest(title=doc.title, content=doc.content, source=doc.source)
+    except ValidationError:
+        raise HTTPException(
+            status_code=422, detail="Saved document has invalid title, content or source",
+        )
+    await _index_saved_document(doc.id, validated)
+    return {"id": doc.id, "message": "Document indexed successfully"}
+
+
 @app.get("/api/messages", response_model=List[MessageResponse])
 async def list_messages(
     limit: int = 100,

backend/app/rag.py

@@ -204,14 +204,16 @@ def index_slack_thread(self, messages: List[Dict[str, Any]], channel_id: str):
                 ids=list(chunks),
             )
 
-    def index_document(self, title: str, content: str, source: str = "docs"):
+    def index_document(self, title: str, content: str, source: str = "docs", *,
+                       document_id: int | None = None):
         """
         Index a document for retrieval
 
         Args:
             title: Document title
             content: Document content
             source: Source of the document
+            document_id: Saved record ID for retry-safe replacement of its chunks
         """
         # Split content into chunks if needed
         chunks = self._split_text(content)
@@ -225,7 +227,10 @@ def index_document(self, title: str, content: str, source: str = "docs"):
             for i in range(len(chunks))
         ]
 
-        vector_store.add_documents(chunks, metadatas)
+        if document_id is None:
+            vector_store.add_documents(chunks, metadatas)
+        else:
+            vector_store.replace_document(document_id, chunks, metadatas)
 
     def _split_text(self, text: str) -> List[str]:
         """Split into overlapping character windows, stopping at the final chunk."""

backend/app/vector_store.py

@@ -68,6 +68,28 @@ def add_documents(self, texts: List[str], metadatas: List[Dict[str, Any]] = None
 
         options = {"ids": ids} if ids is not None else {}
         return self.vector_store.add_texts(texts=texts, metadatas=metadatas, **options)
+
+    def replace_document(self, document_id: int, texts: List[str],
+                         metadatas: List[Dict[str, Any]]):
+        """Upsert owned chunks before removing obsolete chunks for a saved document.
+
+        Existing chunks are not deleted before the upsert succeeds. Cleanup
+        failures propagate so the caller can retry the same stable IDs.
+        Unowned legacy chunks and other documents are never removed.
+        """
+        if not texts:
+            raise ValueError("A document must contain at least one chunk")
+        self._ensure_initialized()
+        if self.vector_store is None:
+            raise RuntimeError("Vector store not initialized")
+
+        previous = self.vector_store.get(where={"document_id": document_id}, include=[])
+        ids = [f"document:{document_id}:{i}" for i in range(len(texts))]
+        owned_metadata = [{**metadata, "document_id": document_id} for metadata in metadatas]
+        self.add_documents(texts, owned_metadata, ids=ids)
+        obsolete = sorted(set(previous["ids"]) - set(ids))
+        if obsolete:
+            self.vector_store.delete(ids=obsolete)
 
     def similarity_search(self, query: str, k: int = None, *,
                           channel_id: str = None) -> List[Dict[str, Any]]: