Backfill ingester test gaps and drop unneeded retry clamp

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

@@ -202,7 +202,9 @@ class WorkerPool:
                 )
                 return
             delay = compute_backoff(job.attempts, self._retry)
-            if not await self._jobs.reschedule(job.id, delay, str(e), worker_id):
+            if not await self._jobs.reschedule(  # pragma: no cover - reaper race
+                job.id, delay, str(e), worker_id
+            ):
                 logger.warning(
                     "Job %s lost claim before reschedule (likely reaper race); "
                     "letting the re-claiming worker drive retry instead",

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

@@ -24,8 +24,6 @@ def compute_backoff(
 
     `rng` is injectable for deterministic tests.
     """
-    if attempt < 1:
-        attempt = 1
     raw = min(policy.base_delay_s * (2 ** (attempt - 1)), policy.max_delay_s)
     r = rng if rng is not None else random
     j = 1.0 + (r.random() * 2 - 1) * policy.jitter

tests/ingester/test_api.py

@@ -352,6 +352,13 @@ async def test_dlq_retry_resurrects(state, jobs):
     assert resp.json()["status"] == JobStatus.QUEUED.value
 
 
+@pytest.mark.asyncio
+async def test_dlq_retry_404_on_missing_job(state):
+    async with _client(state) as client:
+        resp = await client.post("/dlq/missing/retry")
+    assert resp.status_code == 404
+
+
 # --- /sources ---
 
 
@@ -549,6 +556,40 @@ async def test_providers_probes_each_docling_serve_url(state, monkeypatch):
     assert "Name or service not known" in body["docling_serve"][1]["error"]
 
 
+@pytest.mark.asyncio
+async def test_providers_probe_with_real_httpx_transport():
+    """End-to-end through the actual _probe — MockTransport drives the
+    branches: 200, non-2xx, and a transport error all map to the right
+    ProviderEndpoint shape."""
+    import httpx
+
+    from haiku.rag.ingester.api.routes.providers import _probe
+
+    def _handler(request: httpx.Request) -> httpx.Response:
+        path = str(request.url)
+        if "ok" in path:
+            return httpx.Response(200, json={"status": "ok"})
+        if "bad" in path:
+            return httpx.Response(503)
+        raise httpx.ConnectError("boom")
+
+    transport = httpx.MockTransport(_handler)
+    async with httpx.AsyncClient(transport=transport) as client:
+        good = await _probe(client, "http://ok:5001")
+        assert good.reachable is True
+        assert good.status_code == 200
+        assert good.error is None
+
+        unhealthy = await _probe(client, "http://bad:5001")
+        assert unhealthy.reachable is False
+        assert unhealthy.status_code == 503
+
+        dead = await _probe(client, "http://dead:5001")
+        assert dead.reachable is False
+        assert dead.status_code is None
+        assert dead.error is not None and "boom" in dead.error
+
+
 @pytest.mark.asyncio
 async def test_providers_requires_auth(state):
     async with _client(state, auth_token="secret") as client:

tests/ingester/test_pipeline.py

@@ -196,6 +196,23 @@ async def test_connect_error_classified_transient():
         await run_job(client, _job())
 
 
+@pytest.mark.parametrize(
+    "exc_factory",
+    [
+        lambda: TimeoutError("slow"),
+        lambda: OSError("io"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_timeout_and_io_errors_classified_transient(exc_factory):
+    """TimeoutError / OSError both classify as transient. Without this
+    branch they'd fall through to the generic "unexpected" wrapper."""
+    client = _mock_client()
+    client.create_document_from_source.side_effect = exc_factory()
+    with pytest.raises(TransientError, match="timeout/io"):
+        await run_job(client, _job())
+
+
 @pytest.mark.parametrize(
     "exc_factory",
     [

tests/ingester/test_queue.py

@@ -590,6 +590,9 @@ async def test_list_jobs_with_filters(jobs):
     by_status = await jobs.list_jobs(status=JobStatus.QUEUED)
     assert {j.id for j in by_status} == {j2.id, j3.id}
 
+    by_uri = await jobs.list_jobs(uri="u3")
+    assert {j.id for j in by_uri} == {j3.id}
+
 
 @pytest.mark.asyncio
 async def test_counts_by_status(jobs):

tests/ingester/test_workers.py

@@ -457,6 +457,22 @@ async def test_drain_pending_releases_with_no_orphans_is_noop(client, jobs, sync
     assert await pool.drain_pending_releases() == 0
 
 
+@pytest.mark.asyncio
+async def test_live_workers_drops_when_a_worker_finishes(client, jobs, sync):
+    """live_workers powers /health's degraded signal. When a worker task
+    has completed (crashed or exited), it must no longer count."""
+    pool = _pool(client, jobs, sync, worker_count=2)
+    await pool.start()
+    try:
+        assert pool.live_workers == 2
+        # Cancel one worker directly to simulate a crash.
+        pool._workers[0].cancel()
+        await asyncio.gather(pool._workers[0], return_exceptions=True)
+        assert pool.live_workers == 1
+    finally:
+        await pool.stop()
+
+
 @pytest.mark.asyncio
 async def test_double_start_raises(client, jobs, sync):
     pool = _pool(client, jobs, sync, worker_count=1)