Surface failed discovery sweeps in run-batch

docs/ingester.md

@@ -322,8 +322,9 @@ haiku-ingester run-batch --db rag.lancedb
 ```
 
 Orphan deletion compares each source against `sync_state` in the queue DB,
-so persist `ingester.db` between runs for deletions to be detected. It
+so persist `ingester.db` between runs for deletions to be detected. It exits
-exits non-zero if any job dead-letters.
+non-zero if any job dead-letters or a source's discovery sweep does not
+complete.
 
 ### The queue
 

haiku_rag_slim/haiku/rag/ingester/app.py

@@ -23,10 +23,11 @@ logger = logging.getLogger(__name__)
 
 class BatchReport(BaseModel):
     """Outcome of a one-shot batch run: terminal job counts after the queue
-    drained."""
+    drained, plus any sources whose discovery sweep did not complete."""
 
     succeeded: int = 0
     dead: int = 0
+    failed_sweeps: list[str] = []
 
 
 class IngesterApp:
@@ -198,7 +199,7 @@ class IngesterApp:
             started_at = datetime.now(UTC)
             await self._pool.start()
             try:
-                await self._pollers.sweep_all()
+                failed_sweeps = await self._pollers.sweep_all()
                 while True:
                     counts = await self._jobs.counts_by_status()
                     if not counts.get("queued") and not counts.get("claimed"):
@@ -208,6 +209,7 @@ class IngesterApp:
                 return BatchReport(
                     succeeded=completed.get("succeeded", 0),
                     dead=completed.get("dead", 0),
+                    failed_sweeps=failed_sweeps,
                 )
             finally:
                 await self._stop_pool()

haiku_rag_slim/haiku/rag/ingester/cli.py

@@ -167,7 +167,8 @@ def run_batch(
 ) -> None:
     """Run one discover sweep across every configured source, drain the queue,
     then exit. New and changed resources are ingested, resources that vanished
-    from a source are deleted. Exits non-zero if any job dead-letters."""
+    from a source are deleted. Exits non-zero if any job dead-letters or a
+    source's sweep does not complete."""
     asyncio.run(_run_batch(get_config(), db))
 
 
@@ -176,5 +177,10 @@ async def _run_batch(app_config: AppConfig, db_path: Path | None) -> None:
     app = IngesterApp(config=app_config, db_path=db)
     report = await app.run_batch()
     typer.echo(f"Batch complete: {report.succeeded} succeeded, {report.dead} dead")
-    if report.dead:
+    if report.failed_sweeps:
+        typer.echo(
+            f"Sources that failed to sweep: {', '.join(report.failed_sweeps)}",
+            err=True,
+        )
+    if report.dead or report.failed_sweeps:
         raise typer.Exit(1)

haiku_rag_slim/haiku/rag/ingester/pollers/manager.py

@@ -77,12 +77,17 @@ class PollerManager:
             poller._stop.clear()
             self._tasks.append(asyncio.create_task(poller.run()))
 
-    async def sweep_all(self) -> None:
+    async def sweep_all(self) -> list[str]:
         """Run one discover() sweep on every poller, sequentially. Used by
         one-shot batch runs that drive discovery explicitly rather than
-        through the periodic loop."""
+        through the periodic loop. Returns the source ids whose sweep did not
+        complete (discovery failed, circuit open, or pending work already
+        queued) so callers can treat a one-shot run as failed."""
+        failed: list[str] = []
         for poller in self._pollers:
-            await poller._sweep_once()
+            if not await poller._sweep_once():
+                failed.append(poller.source_id)
+        return failed
 
     async def stop(self) -> None:
         for poller in self._pollers:

tests/ingester/test_cli.py

@@ -35,3 +35,12 @@ def test_run_batch_exits_nonzero_when_dead(monkeypatch):
 
     assert result.exit_code == 1
     assert "2 dead" in result.output
+
+
+def test_run_batch_exits_nonzero_when_sweep_fails(monkeypatch):
+    _fake_app(BatchReport(succeeded=2, dead=0, failed_sweeps=["docs"]), monkeypatch)
+
+    result = runner.invoke(cli, ["run-batch", "--db", "x.lancedb"])
+
+    assert result.exit_code == 1
+    assert "failed to sweep: docs" in result.output

tests/ingester/test_run_batch.py

@@ -189,6 +189,34 @@ async def test_run_batch_recovered_doc_is_not_counted_as_dead(tmp_path, use_clie
     assert second.succeeded == 1
 
 
+@pytest.mark.asyncio
+async def test_run_batch_reports_failed_sweep(
+    tmp_path, use_client, monkeypatch, caplog
+):
+    """A source whose discover() raises is reported in failed_sweeps so the
+    run can be treated as failed rather than a silent empty success."""
+    (tmp_path / "a.md").write_text("hello")
+    use_client(_mock_client())
+
+    async def _failing_discover(self, **kwargs):
+        raise RuntimeError("discover blew up")
+        yield  # unreachable; makes this an async generator
+
+    monkeypatch.setattr(
+        "haiku.rag.ingester.sources.fs.FSSource.discover", _failing_discover
+    )
+
+    with caplog.at_level("ERROR", logger="haiku.rag.ingester.pollers.base"):
+        report = await IngesterApp(
+            config=_config(tmp_path), db_path=tmp_path / "db.lancedb"
+        ).run_batch()
+
+    assert report.failed_sweeps == ["local"]
+    assert report.succeeded == 0
+    assert report.dead == 0
+    assert "discover() failed" in caplog.text
+
+
 @pytest.mark.asyncio
 async def test_run_batch_empty_source_returns_immediately(tmp_path, use_client):
     client = _mock_client()