Manual re-ingest + progress display + failure handling

app.py

@@ -8,7 +8,8 @@
 from src.generation.answerer import answer, build_source_elements
 from src.generation.condenser import Condenser
 from src.health_check import check_models, check_ollama
-from src.ingestion.ingest import ingest_folder
+from src.ingestion.ingest import IngestResult, ingest_folder
+from src.ingestion.scanner import scan_folder
 from src.retrieval.bm25_index import BM25Index
 from src.retrieval.embeddings import make_ollama_embed_fn
 from src.retrieval.hybrid import HybridRetriever
@@ -36,21 +37,25 @@ def _build_retriever(store: VectorStore, config) -> HybridRetriever:
     )
 
 
-async def _run_ingestion(store: VectorStore, config) -> int:
+async def _run_ingestion(store: VectorStore, config, **kwargs) -> int:
     """Ingest documents from configured folder. Returns chunk count."""
     folder = config.paths.documents
     if str(folder) and folder.exists():
-        ingested, skipped = ingest_folder(
+        result = ingest_folder(
             folder,
             store,
             recursive=config.scanning.recursive,
             chunk_size=config.chunking.chunk_size,
             chunk_overlap=config.chunking.chunk_overlap,
+            **kwargs,
         )
-        if ingested > 0:
-            logger.info("Ingested %d new files, skipped %d unchanged", ingested, skipped)
-        elif skipped > 0:
-            logger.info("All %d files unchanged, skipped", skipped)
+        if result.ingested > 0:
+            logger.info(
+                "Ingested %d new files, skipped %d, failed %d",
+                result.ingested, result.skipped, result.failed,
+            )
+        elif result.skipped > 0:
+            logger.info("All %d files unchanged, skipped", result.skipped)
 
     return store.count()
 
@@ -120,16 +125,89 @@ async def on_chat_start():
         ]
     ).send()
 
+    # Add re-ingest action button
+    actions = [
+        cl.Action(
+            name="reingest",
+            payload={"value": "reingest"},
+            label="Re-ingest Documents",
+            description="Force re-ingest all documents from the configured folder",
+        )
+    ]
+
     if doc_count > 0:
         await cl.Message(
-            content=f"{doc_count} chunks indexed. Ask me anything!"
+            content=f"{doc_count} chunks indexed. Ask me anything!",
+            actions=actions,
         ).send()
     else:
         await cl.Message(
-            content="No documents indexed. Configure your documents folder in the settings panel (gear icon) or `config.yaml`."
+            content="No documents indexed. Configure your documents folder in the settings panel (gear icon) or `config.yaml`.",
+            actions=actions,
         ).send()
 
 
+def _build_ingest_summary(result: IngestResult, total_files: int) -> str:
+    """Build a human-readable ingestion summary."""
+    parts = [f"**Ingestion complete.** {result.ingested}/{total_files} documents ingested."]
+
+    if result.skipped > 0:
+        parts.append(f"{result.skipped} unchanged (skipped).")
+
+    if result.failed > 0:
+        parts.append(f"{result.failed} failed.")
+        parts.append("\n<details><summary>Failed documents</summary>\n")
+        for filename, error in result.failures:
+            parts.append(f"- **{filename}**: {error}")
+        parts.append("\n</details>")
+
+    return " ".join(parts)
+
+
+@cl.action_callback("reingest")
+async def on_reingest(action: cl.Action):
+    """Handle manual re-ingest action button."""
+    config = cl.user_session.get("config")
+    store = cl.user_session.get("store")
+    folder = config.paths.documents
+
+    if not str(folder) or not folder.exists():
+        await cl.Message(
+            content="No documents folder configured. Set it in the settings panel first."
+        ).send()
+        return
+
+    # Show document count before starting
+    files = scan_folder(folder, recursive=config.scanning.recursive)
+    total_files = len(files)
+    await cl.Message(
+        content=f"Re-ingesting {total_files} documents from `{folder}`..."
+    ).send()
+
+    # Run ingestion with force=True and per-file progress steps
+    async with cl.Step(name="Re-ingestion Progress", type="tool") as step:
+        result = ingest_folder(
+            folder,
+            store,
+            recursive=config.scanning.recursive,
+            chunk_size=config.chunking.chunk_size,
+            chunk_overlap=config.chunking.chunk_overlap,
+            force=True,
+            on_progress=lambda name, status: logger.info("  %s: %s", name, status),
+        )
+        step.output = _build_ingest_summary(result, total_files)
+
+    # Rebuild retriever with updated store
+    retriever = _build_retriever(store, config)
+    cl.user_session.set("retriever", retriever)
+
+    summary = _build_ingest_summary(result, total_files)
+    doc_count = store.count()
+    await cl.Message(
+        content=f"{summary}\n\n{doc_count} chunks indexed."
+    ).send()
+
+
 @cl.on_settings_update
 async def on_settings_update(settings: dict):
     """Re-scan and ingest when settings change."""

src/ingestion/ingest.py

@@ -1,6 +1,8 @@
-"""Incremental document ingestion with change detection."""
+"""Incremental document ingestion with change detection and failure handling."""
 
 import logging
+from collections.abc import Callable
+from dataclasses import dataclass, field
 from pathlib import Path
 
 from src.ingestion.chunker import chunk_documents
@@ -10,6 +12,20 @@
 
 logger = logging.getLogger(__name__)
 
+ProgressCallback = Callable[[str, str], None]
+"""Callback(filename, status) called per file. Status: 'processing', 'ingested', 'skipped', 'failed'."""
+
+
+@dataclass
+class IngestResult:
+    """Summary of an ingestion run."""
+
+    ingested: int = 0
+    skipped: int = 0
+    failed: int = 0
+    failures: list[tuple[str, str]] = field(default_factory=list)
+    """List of (filename, error_message) for failed files."""
+
 
 def ingest_folder(
     folder: Path,
@@ -18,50 +34,79 @@ def ingest_folder(
     recursive: bool = True,
     chunk_size: int = 512,
     chunk_overlap: int = 100,
-) -> tuple[int, int]:
+    force: bool = False,
+    on_progress: ProgressCallback | None = None,
+) -> IngestResult:
     """Scan a folder and ingest new or changed documents.
 
-    Returns (ingested_count, skipped_count) — files ingested vs unchanged.
+    Args:
+        folder: Path to the documents folder.
+        store: Vector store to add chunks to.
+        recursive: Whether to scan subdirectories.
+        chunk_size: Maximum chunk size in characters.
+        chunk_overlap: Overlap between chunks.
+        force: If True, re-ingest all files regardless of hash.
+        on_progress: Optional callback for per-file progress.
+
+    Returns:
+        IngestResult with counts of ingested, skipped, and failed files.
     """
     files = scan_folder(folder, recursive=recursive)
+    result = IngestResult()
 
     if not files:
-        return 0, 0
+        return result
 
-    existing_hashes = _get_stored_file_hashes(store)
-
-    ingested = 0
-    skipped = 0
+    existing_hashes = set() if force else _get_stored_file_hashes(store)
 
     for file_path in files:
-        file_hash = compute_file_hash(file_path)
-        if file_hash in existing_hashes:
-            skipped += 1
-            continue
+        filename = file_path.name
 
-        docs = load_file(file_path, folder)
-        if not docs:
-            continue
-
-        # Attach file_hash to document metadata before chunking
-        for doc in docs:
-            doc.metadata["file_hash"] = file_hash
-
-        chunks = chunk_documents(
-            docs,
-            chunk_size=chunk_size,
-            chunk_overlap=chunk_overlap,
-        )
+        if on_progress:
+            on_progress(filename, "processing")
 
-        # Propagate file_hash to chunk metadata
-        for chunk in chunks:
-            chunk.metadata["file_hash"] = file_hash
-
-        store.add_chunks(chunks)
-        ingested += 1
-        logger.info("Ingested %s (%d chunks)", file_path.name, len(chunks))
+        file_hash = compute_file_hash(file_path)
+        if not force and file_hash in existing_hashes:
+            result.skipped += 1
+            if on_progress:
+                on_progress(filename, "skipped")
+            continue
 
-    return ingested, skipped
+        try:
+            docs = load_file(file_path, folder)
+            if not docs:
+                result.skipped += 1
+                if on_progress:
+                    on_progress(filename, "skipped")
+                continue
+
+            for doc in docs:
+                doc.metadata["file_hash"] = file_hash
+
+            chunks = chunk_documents(
+                docs,
+                chunk_size=chunk_size,
+                chunk_overlap=chunk_overlap,
+            )
+
+            for chunk in chunks:
+                chunk.metadata["file_hash"] = file_hash
+
+            store.add_chunks(chunks)
+            result.ingested += 1
+            if on_progress:
+                on_progress(filename, "ingested")
+            logger.info("Ingested %s (%d chunks)", filename, len(chunks))
+
+        except Exception as e:
+            result.failed += 1
+            error_msg = f"{type(e).__name__}: {e}"
+            result.failures.append((filename, error_msg))
+            if on_progress:
+                on_progress(filename, "failed")
+            logger.warning("Failed to ingest %s: %s", filename, error_msg)
+
+    return result
 
 
 def _get_stored_file_hashes(store: VectorStore) -> set[str]:

tests/test_ingest.py

@@ -1,10 +1,11 @@
 """Tests for incremental ingestion."""
 
+import hashlib
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from src.ingestion.ingest import ingest_folder
+from src.ingestion.ingest import IngestResult, ingest_folder
 
 
 @pytest.fixture
@@ -24,16 +25,14 @@ def mock_store():
 
 class TestIngestFolder:
     def test_ingests_new_file(self, doc_folder, mock_store):
-        ingested, skipped = ingest_folder(
+        result = ingest_folder(
             doc_folder, mock_store, chunk_size=512, chunk_overlap=100
         )
-        assert ingested == 1
-        assert skipped == 0
+        assert result.ingested == 1
+        assert result.skipped == 0
         mock_store.add_chunks.assert_called_once()
 
     def test_skips_unchanged_file(self, doc_folder, mock_store):
-        # Simulate file already in store by returning its hash
-        import hashlib
         file_hash = hashlib.md5(
             (doc_folder / "notes.txt").read_bytes()
         ).hexdigest()
@@ -43,9 +42,9 @@ def test_skips_unchanged_file(self, doc_folder, mock_store):
             [{"file_hash": file_hash}],
         )
 
-        ingested, skipped = ingest_folder(doc_folder, mock_store)
-        assert ingested == 0
-        assert skipped == 1
+        result = ingest_folder(doc_folder, mock_store)
+        assert result.ingested == 0
+        assert result.skipped == 1
         mock_store.add_chunks.assert_not_called()
 
     def test_chunks_have_file_hash(self, doc_folder, mock_store):
@@ -55,13 +54,60 @@ def test_chunks_have_file_hash(self, doc_folder, mock_store):
             assert "file_hash" in chunk.metadata
 
     def test_empty_folder(self, tmp_path, mock_store):
-        ingested, skipped = ingest_folder(tmp_path, mock_store)
-        assert ingested == 0
-        assert skipped == 0
+        result = ingest_folder(tmp_path, mock_store)
+        assert result.ingested == 0
+        assert result.skipped == 0
 
     def test_relative_path_in_metadata(self, doc_folder, mock_store):
         ingest_folder(doc_folder, mock_store)
         chunks = mock_store.add_chunks.call_args[0][0]
         for chunk in chunks:
             rel_path = chunk.metadata.get("relative_path", "")
             assert not rel_path.startswith("/")
+
+    def test_force_ignores_existing_hashes(self, doc_folder, mock_store):
+        """Force mode ingests even when file hash already exists."""
+        file_hash = hashlib.md5(
+            (doc_folder / "notes.txt").read_bytes()
+        ).hexdigest()
+        mock_store.count.return_value = 5
+        mock_store.get_all_texts_and_metadatas.return_value = (
+            ["text"],
+            [{"file_hash": file_hash}],
+        )
+
+        result = ingest_folder(doc_folder, mock_store, force=True)
+        assert result.ingested == 1
+        assert result.skipped == 0
+        mock_store.add_chunks.assert_called_once()
+
+    def test_failure_skips_bad_file_continues(self, tmp_path, mock_store):
+        """A file that fails to load is skipped; other files still ingested."""
+        (tmp_path / "good.txt").write_text("Good content")
+        (tmp_path / "bad.pdf").write_bytes(b"not a real pdf")
+
+        result = ingest_folder(tmp_path, mock_store)
+        # good.txt should succeed, bad.pdf may fail
+        assert result.ingested >= 1 or result.failed >= 1
+        assert result.ingested + result.failed + result.skipped == 2
+
+    def test_failure_records_error_details(self, tmp_path, mock_store):
+        """Failed files are recorded with filename and error message."""
+        (tmp_path / "bad.pdf").write_bytes(b"not a real pdf")
+
+        result = ingest_folder(tmp_path, mock_store)
+        if result.failed > 0:
+            assert len(result.failures) == result.failed
+            assert result.failures[0][0] == "bad.pdf"
+            assert isinstance(result.failures[0][1], str)
+
+    def test_progress_callback_called(self, doc_folder, mock_store):
+        """Progress callback is called for each file."""
+        progress_calls = []
+        ingest_folder(
+            doc_folder,
+            mock_store,
+            on_progress=lambda name, status: progress_calls.append((name, status)),
+        )
+        assert len(progress_calls) >= 1
+        assert progress_calls[0][0] == "notes.txt"