Tighten HTTP config-removal handling

haiku_rag_slim/haiku/rag/ingester/queue/repository.py

@@ -399,12 +399,15 @@ class SyncStateRepo:
         # connection. See JobRepo for the SQLite cursor + commit constraint.
         self._lock = lock or asyncio.Lock()
 
-    async def get_snapshot(self, source_id: str) -> dict[str, str]:
+    async def get_snapshot(self, source_id: str) -> dict[str, str | None]:
-        """uri -> revision map for the source. Drops rows where revision is
+        """uri -> revision map for the source. Revision is None for URIs
-        NULL (the poller can't compare against an absent revision)."""
+        known to the source but never successfully ingested with a revision
+        (HTTP responses without ETag / Last-Modified, or a worker that DLQ'd
+        before completion). Sources compare with `previous == current`; None
+        compares as not-equal so the URI is treated as changed."""
         async with self._lock:
             async with self._conn.execute(
-                "SELECT uri, revision FROM sync_state WHERE source_id=? AND revision IS NOT NULL",
+                "SELECT uri, revision FROM sync_state WHERE source_id=?",
                 (source_id,),
             ) as cursor:
                 rows = await cursor.fetchall()

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

@@ -9,7 +9,7 @@ from pydantic import BaseModel, Field
 # uri -> revision. Captures what revisions of which URIs we had last seen
 # for a given source. Passed to discover() so the source can yield only
 # UPSERT / UNCHANGED / DELETE deltas instead of a full re-scan.
-RevisionSnapshot = Mapping[str, str]
+RevisionSnapshot = Mapping[str, str | None]
 
 
 class SourceEventKind(StrEnum):

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

@@ -104,7 +104,7 @@ class FSSource:
     async def discover(
         self, since: RevisionSnapshot | None = None
     ) -> AsyncIterator[SourceEvent]:
-        snapshot: dict[str, str] = dict(since) if since else {}
+        snapshot: dict[str, str | None] = dict(since) if since else {}
         now = datetime.now(UTC)
         seen: set[str] = set()
 

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

@@ -95,7 +95,7 @@ class HTTPSource:
         # ambiguous (transient outage, misconfigured URL, auth blip), so we
         # fall back to UPSERT with no revision and let the worker decide
         # via GET.
-        snapshot: dict[str, str] = dict(since) if since else {}
+        snapshot: dict[str, str | None] = dict(since) if since else {}
         now = datetime.now(UTC)
         configured = set(self.urls)
 

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

@@ -25,19 +25,22 @@ def resolve_fetcher(
     source whose ``supports(uri)`` returns True wins, then a scheme-based
     ad-hoc adapter.
     """
+    if source_id is not None:
+        for src in sources or ():
+            if src.source_id == source_id:
+                if not src.supports(uri):
+                    raise UnsupportedSourceError(
+                        f"Source {source_id!r} doesn't support URI {uri!r}"
+                    )
+                return src
+        raise UnsupportedSourceError(
+            f"No configured source with id {source_id!r} for URI {uri!r}"
+        )
+
     if sources:
-        if source_id is not None:
+        for src in sources:
-            for src in sources:
+            if src.supports(uri):
-                if src.source_id == source_id:
+                return src
-                    if not src.supports(uri):
-                        raise UnsupportedSourceError(
-                            f"Source {source_id!r} doesn't support URI {uri!r}"
-                        )
-                    return src
-        else:
-            for src in sources:
-                if src.supports(uri):
-                    return src
 
     scheme = urlparse(uri).scheme
     if scheme in ("", "file"):

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

@@ -109,7 +109,7 @@ class S3Source:
 
         from haiku.rag.s3 import make_s3_store
 
-        snapshot: dict[str, str] = dict(since) if since else {}
+        snapshot: dict[str, str | None] = dict(since) if since else {}
         now = datetime.now(UTC)
         seen: set[str] = set()
         store = make_s3_store(self.bucket, self.storage_options)

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

@@ -253,7 +253,7 @@ class WebDAVSource:
     async def discover(
         self, since: RevisionSnapshot | None = None
     ) -> AsyncIterator[SourceEvent]:
-        snapshot: dict[str, str] = dict(since) if since else {}
+        snapshot: dict[str, str | None] = dict(since) if since else {}
         now = datetime.now(UTC)
         seen: set[str] = set()
 

tests/ingester/test_http_source.py

@@ -296,3 +296,25 @@ async def test_discover_emits_delete_for_removed_url():
     assert by_uri["https://example.com/a.md"].kind is not SourceEventKind.DELETE
     assert by_uri["https://example.com/gone.md"].kind is SourceEventKind.DELETE
     assert by_uri["https://example.com/gone.md"].revision is None
+
+
+@pytest.mark.asyncio
+async def test_discover_emits_delete_for_removed_url_with_none_revision():
+    """Same as above but the snapshot entry's revision is None — an HTTP
+    response without ETag or Last-Modified produces that state, and these
+    rows must still trigger config-removal DELETE."""
+    transport = _transport(
+        {
+            ("HEAD", "https://example.com/a.md"): httpx.Response(
+                200, headers={"etag": '"abc"'}
+            ),
+        }
+    )
+    src = HTTPSource(
+        source_id="x", urls=["https://example.com/a.md"], transport=transport
+    )
+    events = [
+        e async for e in src.discover(since={"https://example.com/no-etag.md": None})
+    ]
+    by_uri = {e.uri: e for e in events}
+    assert by_uri["https://example.com/no-etag.md"].kind is SourceEventKind.DELETE

tests/ingester/test_pollers.py

@@ -130,9 +130,10 @@ async def test_upsert_event_enqueues_job_and_touches_sync_state(fs_config, jobs,
     assert queued[0].revision == "r1"
 
     # Pollers DO NOT write revision to sync_state — the worker does that
-    # after a successful ingest. But last_seen_at is bumped.
+    # after a successful ingest. The URI shows up in the snapshot with
+    # revision=None until then.
     snapshot = await sync.get_snapshot("src")
-    assert snapshot == {}  # revision left empty by the poller
+    assert snapshot == {"file:///a.md": None}
 
 
 @pytest.mark.asyncio

tests/ingester/test_queue.py

@@ -750,6 +750,17 @@ async def test_sync_state_snapshot_scoped_per_source(sync):
     assert await sync.get_snapshot("s2") == {"u": "def"}
 
 
+@pytest.mark.asyncio
+async def test_sync_state_snapshot_includes_null_revision_rows(sync):
+    """A URI known to a source but never successfully ingested with a
+    revision (HTTP without ETag/Last-Modified, or a worker DLQ'd before
+    completion) is still in the snapshot — with revision=None. Needed so
+    config-removal DELETE detection sees these rows."""
+    await sync.upsert("s", "u", revision=None, content_hash=None)
+    snapshot = await sync.get_snapshot("s")
+    assert snapshot == {"u": None}
+
+
 @pytest.mark.asyncio
 async def test_sync_state_upsert_preserves_revision_when_none(sync):
     """upsert(revision=None) leaves an existing revision in place."""

tests/ingester/test_resolve_fetcher.py

@@ -83,3 +83,20 @@ def test_source_id_raises_when_matched_source_rejects_uri():
     fs = FSSource(root=Path("/tmp"), source_id="local")
     with pytest.raises(ValueError, match="doesn't support URI"):
         resolve_fetcher("https://example.com/x", sources=[fs], source_id="local")
+
+
+def test_source_id_raises_when_no_source_matches():
+    """A worker job carries source_id from when it was enqueued. If the
+    operator renamed/removed that source, falling through to an ad-hoc
+    unauthenticated adapter would silently drop credentials. Raise instead
+    so the job DLQs and the misconfiguration is visible."""
+    arxiv = HTTPSource(source_id="arxiv", headers={"Authorization": "Bearer A"})
+    with pytest.raises(ValueError, match="No configured source with id 'gone'"):
+        resolve_fetcher("https://example.com/x", sources=[arxiv], source_id="gone")
+
+
+def test_source_id_with_no_sources_list_still_raises():
+    """Same strict mode when sources is None: source_id is a strong claim
+    that must match a configured source."""
+    with pytest.raises(ValueError, match="No configured source with id"):
+        resolve_fetcher("https://example.com/x", source_id="arxiv")