Drop defensive branches

haiku_rag_slim/haiku/rag/config/models.py

@@ -367,6 +367,9 @@ class FSSourceConfig(_SourceBase):
 
 class HTTPSourceConfig(_SourceBase):
     type: Literal["http"]
+    # HTTP sources have no natural key to derive an id from (a list of urls
+    # has no canonical representation), so require one.
+    id: str
     urls: list[str] = []
     headers: dict[str, str] = Field(default_factory=dict)
 
@@ -384,6 +387,9 @@ class WebDAVSourceConfig(_SourceBase):
     are discovered via PROPFIND on `base_url`; fetch is plain HTTP GET."""
 
     type: Literal["webdav"]
+    # base_url can be deep + opaque (long URL paths, credentials embedded);
+    # require an explicit short id for the queue and logs.
+    id: str
     base_url: str
     username: str | None = None
     password: str | None = None

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

@@ -34,8 +34,6 @@ def build_source(
             source_id=cfg.id,
         )
     if isinstance(cfg, HTTPSourceConfig):
-        if cfg.id is None:  # pragma: no cover - config-validation guard
-            raise ValueError("HTTPSourceConfig.id is required")
         return HTTPSource(source_id=cfg.id, urls=cfg.urls, headers=cfg.headers)
     if isinstance(cfg, S3SourceConfig):
         return S3Source(
@@ -47,8 +45,6 @@ def build_source(
             source_id=cfg.id,
         )
     if isinstance(cfg, WebDAVSourceConfig):
-        if cfg.id is None:  # pragma: no cover - config-validation guard
-            raise ValueError("WebDAVSourceConfig.id is required")
         return WebDAVSource(
             source_id=cfg.id,
             base_url=cfg.base_url,

haiku_rag_slim/haiku/rag/ingester/sources/fs.py

@@ -135,10 +135,6 @@ class FSSource:
             if not self.filter.include_file(str(path)):
                 continue
             uri = path.as_uri()
-            if uri in seen:
-                # A symlink and its target (or two symlinks to the same file)
-                # both walked through; the first wins so the queue sees one job.
-                continue
             revision = str(path.stat().st_mtime_ns)
             seen.add(uri)
             previous = snapshot.get(uri)

haiku_rag_slim/haiku/rag/ingester/workers/pool.py

@@ -108,13 +108,7 @@ class WorkerPool:
 
     async def _worker_loop(self, worker_id: str) -> None:
         while not self._stop.is_set():
-            try:
+            job = await self._jobs.claim_next(worker_id)
-                job = await self._jobs.claim_next(worker_id)
-            except Exception:  # pragma: no cover - defensive against DB hiccups
-                logger.exception("claim_next failed in %s", worker_id)
-                await self._sleep_or_stop(self._poll_idle_s)
-                continue
-
             if job is None:
                 await self._sleep_or_stop(self._poll_idle_s)
                 continue
@@ -127,12 +121,9 @@ class WorkerPool:
             await self._sleep_or_stop(self._reaper_interval_s)
             if self._stop.is_set():
                 return
-            try:
+            reset = await self._jobs.reap_stale(self._claim_timeout_s)
-                reset = await self._jobs.reap_stale(self._claim_timeout_s)
+            if reset:
-                if reset:
+                logger.info("Reaper reset %d stale claim(s)", reset)
-                    logger.info("Reaper reset %d stale claim(s)", reset)
-            except Exception:  # pragma: no cover - defensive against DB hiccups
-                logger.exception("reaper failed")
 
     async def _sleep_or_stop(self, seconds: float) -> None:
         try:
@@ -198,12 +189,6 @@ class WorkerPool:
                 e,
             )
             return
-        except Exception as e:  # pragma: no cover - pipeline classifier net
-            # Defensive: pipeline classifier should have caught everything.
-            await self._jobs.mark_dead(job.id, f"unclassified: {e!r}", worker_id)
-            logger.exception("Unclassified error in job %s", job.id)
-            return
-
         # Guard against the reaper race: if our claim was reset and another
         # worker re-claimed the job, mark_succeeded is a no-op. Don't write
         # sync_state in that case — the new worker will write it when it

tests/ingester/test_fs_source.py

@@ -219,17 +219,16 @@ async def test_fs_source_discover_follows_within_root_symlinks(fs_root: Path):
     """A symlink whose target lives inside root is legitimate — supports/
     head/fetch all accept it (resolve-then-check), so discover() must too,
     otherwise an ad-hoc add-src on a link works but the poller never picks
-    it up. The emitted URI is the resolved target's, and the seen-set
+    it up. We emit the resolved target's URI, never the alias's; redundant
-    dedupes when both the link and its target are walked."""
+    yields from walking both alias and target are absorbed by the queue's
+    unique index on (source_id, uri, op)."""
     target = fs_root / "real.md"
     target.write_text("real content")
     (fs_root / "alias.md").symlink_to(target)
     src = FSSource(root=fs_root, supported_extensions=[".md"])
     events = [e async for e in src.discover(since=None)]
     uris = [e.uri for e in events]
-    # The resolved target appears exactly once even though both `alias.md`
+    assert target.as_uri() in uris
-    # and `real.md` are encountered during the walk.
-    assert uris.count(target.as_uri()) == 1
     # The alias's own URI is not emitted — we normalise to the target.
     assert (fs_root / "alias.md").as_uri() not in uris