cover s3 uri override, unsupported extension, watcher error path

tests/test_s3_monitor.py

@@ -317,6 +317,37 @@ async def test_s3_watcher_invalid_uri_rejected():
         )
 
 
+@pytest.mark.asyncio
+async def test_s3_watcher_upsert_failure_does_not_abort_sweep(s3_listing):
+    """A failing upsert doesn't propagate; the refresh keeps processing siblings."""
+    set_batches, _ = s3_listing
+    set_batches([[_meta("incoming/bad.txt", "abc"), _meta("incoming/good.txt", "def")]])
+
+    from haiku.rag.monitor import S3Watcher
+
+    rag = AsyncMock(spec=HaikuRAG)
+    rag.list_documents.return_value = []
+
+    good_doc = Document(
+        id="good-id", content="...", uri="s3://my-bucket/incoming/good.txt"
+    )
+
+    async def maybe_fail(uri, **_):
+        if uri.endswith("bad.txt"):
+            raise RuntimeError("boom")
+        return good_doc
+
+    rag.create_document_from_source.side_effect = maybe_fail
+
+    watcher = S3Watcher(client=rag, entry=_entry(), supported_extensions=[".txt"])
+
+    # The failing upsert must not propagate out of refresh().
+    await watcher.refresh()
+
+    # Both objects were attempted — the first failure didn't abort the sibling.
+    assert rag.create_document_from_source.await_count == 2
+
+
 @pytest.mark.asyncio
 async def test_serve_starts_one_s3_task_per_entry(monkeypatch, s3_listing):
     """`serve` wires one S3Watcher task per MonitorConfig.s3 entry."""

tests/test_s3_source.py

@@ -169,3 +169,37 @@ async def test_create_document_from_s3_rejects_invalid_uri(
     async with HaikuRAG(temp_db_path, create=True) as client:
         with pytest.raises(ValueError, match="Invalid S3 URI"):
             await client.create_document_from_source("s3://only-bucket-no-key")
+
+
+@pytest.mark.asyncio
+async def test_create_document_from_s3_rejects_unsupported_extension(
+    fake_obstore_io, temp_db_path
+):
+    head_async, _ = fake_obstore_io
+    head_async.return_value = _meta('"abc"')
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        with pytest.raises(ValueError, match="Unsupported content type"):
+            await client.create_document_from_source("s3://my-bucket/file.unsupported")
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+async def test_create_document_from_s3_uri_override(fake_obstore_io, temp_db_path):
+    """`uri=` kwarg overrides the s3:// URL as the stored document identifier."""
+    head_async, get_async = fake_obstore_io
+    head_async.return_value = _meta('"abc"')
+    get_async.return_value = _get_result(b"override target content")
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        doc = await client.create_document_from_source(
+            "s3://my-bucket/file.txt", uri="arxiv:2401.00001"
+        )
+
+        assert doc.uri == "arxiv:2401.00001"
+        # The override is the canonical identifier — lookup by it must hit the doc.
+        looked_up = await client.get_document_by_uri("arxiv:2401.00001")
+        assert looked_up is not None
+        assert looked_up.id == doc.id
+        # The s3:// URL is NOT a stored identifier.
+        assert await client.get_document_by_uri("s3://my-bucket/file.txt") is None