fix(ingest): address review round 1 (reclaim queue + tests)

- Register the periodic stalled-job reclaim on a dedicated ingest-maintenance
  queue that every worker drains (any --tier), so reclaim still fires when the
  fast fleet is scaled to zero and only ocr workers run. procrastinate's
  periodic-defer dedup keeps it single-run across drainers.
- escalation: mark `unsupported`/`forced` reason labels as reserved (not raised).
- processor: note that options/progress_callback are intentionally not threaded
  through _parse_pdf_tier yet (symmetric with the inline path).
- tests: assert TieredEscalationStrategy backoff progression (4/8/16/…/300s);
  cover get_ingest_pending per-queue aggregation + the legacy job_counts
  fallback; add an external-path zero-page no-escalation case; use the canonical
  INGEST_QUEUE_FAST instead of the back-compat alias.

Deck #323.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

nextcloud_mcp_server/cli.py

@@ -384,6 +384,7 @@ def worker(concurrency: int | None, tier: str | None):
 
     from nextcloud_mcp_server.vector.queue.procrastinate import (  # noqa: PLC0415
         ALL_INGEST_QUEUES,
+        INGEST_QUEUE_MAINTENANCE,
         LEGACY_INGEST_QUEUE,
         TIER_QUEUES,
         apply_ingest_queue_schema,
@@ -392,11 +393,14 @@ def worker(concurrency: int | None, tier: str | None):
 
     # Which queues this process drains. A single tier -> just its queue; no tier
     # -> every tier queue PLUS the legacy single queue, so a rolling upgrade
-    # never strands jobs deferred under the pre-#323 name.
+    # never strands jobs deferred under the pre-#323 name. Every worker also
+    # drains the maintenance queue so the periodic stalled-job reclaim fires
+    # regardless of which tier(s) are scaled up (procrastinate dedups the
+    # periodic, so multiple drainers don't multiply the reclaim).
     if tier is not None:
-        queues = [TIER_QUEUES[tier]]
+        queues = [TIER_QUEUES[tier], INGEST_QUEUE_MAINTENANCE]
     else:
-        queues = [*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE]
+        queues = [*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE, INGEST_QUEUE_MAINTENANCE]
 
     # This is the consumer side of the distributed (postgres) ingest backend.
     # Unlike the in-process anyio pool, the worker talks to procrastinate's App

nextcloud_mcp_server/document_processors/escalation.py

@@ -53,8 +53,10 @@ class EscalateError(Exception):
     the junk text is never indexed, and it must never be swallowed by a broad
     ``except Exception`` on the indexing path.
 
-    ``reason`` uses the existing escalation label vocabulary:
+    ``reason`` uses the existing escalation label vocabulary. This PR raises
-    ``empty_text`` | ``low_confidence`` | ``unsupported`` | ``forced``.
+    ``empty_text`` (scanned / no text layer) and ``low_confidence`` (junk text
+    layer); ``unsupported`` and ``forced`` are reserved for future callers and
+    not raised yet.
     """
 
     def __init__(self, *, from_tier: str, to_tier: str, reason: str) -> None:

nextcloud_mcp_server/vector/processor.py

@@ -147,6 +147,10 @@ async def _parse_pdf_tier(
         EscalateError,
     )
 
+    # options / progress_callback are not threaded here -- the indexing caller
+    # passes neither today, and the inline path (registry.process) omits them
+    # too. Forward them if a tier processor ever needs per-call tuning (e.g. OCR
+    # DPI); keeping the two paths symmetric until then.
     result = await registry.process_tier(content, content_type, filename, tier)
     if result.success:
         decision = registry.evaluate_escalation(

nextcloud_mcp_server/vector/queue/procrastinate.py

@@ -87,6 +87,14 @@ LEGACY_INGEST_QUEUE = "ingest"
 # Back-compat alias for callers that imported the old single-queue constant.
 INGEST_QUEUE_NAME = DEFAULT_INGEST_QUEUE
 
+# Maintenance queue carrying ONLY the periodic stalled-job reclaim (no document
+# jobs). Every worker drains it regardless of --tier, so the reclaim fires even
+# in an asymmetric deployment where the fast fleet is scaled to zero and only
+# ocr workers run. procrastinate's periodic-defer dedup ensures exactly one
+# worker runs each tick even when many drain this queue. Kept off document
+# queues so tier isolation (which fleet processes which docs) is preserved.
+INGEST_QUEUE_MAINTENANCE = "ingest-maintenance"
+
 # Queues the job-count + reclaim helpers sweep (tier queues + the legacy one).
 _MANAGED_QUEUES: tuple[str, ...] = (*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE)
 
@@ -368,8 +376,12 @@ def _build_ingest_blueprint() -> Blueprint:
             max_transient_attempts=get_settings().ingest_transient_max_attempts
         ),
     )(process_document_task)
+    # Reclaim runs on the dedicated maintenance queue (every worker drains it),
+    # not a tier queue -- otherwise an ocr-only deployment (fast scaled to zero)
+    # would never fire the periodic and orphaned ``doing`` jobs would never be
+    # reclaimed. The task itself sweeps ALL queues (get_stalled_jobs(queue=None)).
     reclaim = bp.task(
-        name="reclaim_stalled_jobs", queue=DEFAULT_INGEST_QUEUE, pass_context=True
+        name="reclaim_stalled_jobs", queue=INGEST_QUEUE_MAINTENANCE, pass_context=True
     )(reclaim_stalled_ingest_jobs)
     bp.periodic(cron="*/5 * * * *", periodic_id="reclaim_stalled_ingest")(reclaim)
     return bp

tests/unit/test_registry_tiering.py

@@ -388,3 +388,15 @@ def test_evaluate_escalation_terminal_when_no_higher_tier(monkeypatch):
         processor="fast",
     )
     assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None
+
+
+def test_evaluate_escalation_zero_page_does_not_escalate(monkeypatch):
+    """A zero-page (empty/corrupt) PDF never escalates on the external path."""
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5))
+    res = ProcessingResult(
+        text="",
+        metadata={"page_count": 0, "page_boundaries": []},
+        processor="fast",
+    )
+    assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None

tests/unit/vector/test_ingest_status.py

@@ -11,21 +11,45 @@ pytestmark = pytest.mark.unit
 
 
 class TestGetIngestPending:
-    async def test_postgres_reads_job_counts(self):
+    async def test_postgres_aggregates_by_queue(self):
+        """Per-queue counts are summed into fleet-wide totals (Deck #323)."""
         producer = AsyncMock()
-        producer.job_counts.return_value = {"todo": 5, "doing": 2, "failed": 1}
+        producer.job_counts_by_queue.return_value = {
+            "ingest-fast": {"todo": 5, "doing": 1},
+            "ingest-ocr": {"doing": 1, "failed": 1},
+        }
 
         result = await get_ingest_pending(
             task_producer=producer,
             document_receive_stream=None,
             ingest_queue="postgres",
         )
-        assert result.pending == 7  # todo + doing
+        assert result.pending == 7  # todo(5) + doing(1+1)
         assert result.job_counts == {"todo": 5, "doing": 2, "failed": 1}
+        assert result.job_counts_by_queue == {
+            "ingest-fast": {"todo": 5, "doing": 1},
+            "ingest-ocr": {"doing": 1, "failed": 1},
+        }
+
+    async def test_postgres_falls_back_to_aggregate_counts(self):
+        """A producer without job_counts_by_queue uses the aggregated call."""
+
+        class LegacyProducer:
+            async def job_counts(self):
+                return {"todo": 5, "doing": 2, "failed": 1}
+
+        result = await get_ingest_pending(
+            task_producer=LegacyProducer(),
+            document_receive_stream=None,
+            ingest_queue="postgres",
+        )
+        assert result.pending == 7
+        assert result.job_counts == {"todo": 5, "doing": 2, "failed": 1}
+        assert result.job_counts_by_queue is None
 
     async def test_postgres_degrades_to_zero_on_error(self):
         producer = AsyncMock()
-        producer.job_counts.side_effect = RuntimeError("db down")
+        producer.job_counts_by_queue.side_effect = RuntimeError("db down")
 
         result = await get_ingest_pending(
             task_producer=producer,

tests/unit/vector/test_procrastinate_producer.py

@@ -48,7 +48,8 @@ class TestProcrastinateTaskProducer:
         assert len(jobs) == 1
         job = jobs[0]
         assert job["task_name"] == pq.INGEST_TASK_NAME
-        assert job["queue_name"] == pq.INGEST_QUEUE_NAME
+        # New jobs are deferred onto the cheapest tier's queue.
+        assert job["queue_name"] == pq.INGEST_QUEUE_FAST
         assert job["queueing_lock"] == "alice:note:42"
         assert job["lock"] is None  # no execution lock (crash-deadlock guard)
         assert job["args"]["doc_id"] == "42"

tests/unit/vector/test_tiered_escalation_strategy.py

@@ -5,6 +5,8 @@ and the procrastinate TieredEscalationStrategy that turns a raised exception
 into a queue-hop / same-tier retry / give-up decision.
 """
 
+from datetime import datetime, timezone
+
 import httpx
 import pytest
 from procrastinate.jobs import Job
@@ -87,6 +89,20 @@ class TestTieredEscalationStrategy:
         assert decision.queue is None
         assert decision.retry_at is not None
 
+    def test_transient_backoff_progression(self):
+        # min(4 * 2**(attempts-1), 300): 4, 8, 16, ... capped at 300s.
+        strat = self._strategy(max_transient=100)
+        for attempts, expected in [(1, 4), (2, 8), (3, 16), (4, 32), (20, 300)]:
+            decision = strat.get_retry_decision(
+                exception=httpx.ConnectError("x"), job=_job(attempts=attempts)
+            )
+            assert decision is not None and decision.retry_at is not None
+            delta = (decision.retry_at - datetime.now(timezone.utc)).total_seconds()
+            # retry_at = now + wait; allow a small window for execution time.
+            assert expected - 2 <= delta <= expected + 1, (
+                f"attempts={attempts}: delta={delta:.2f}s, expected≈{expected}s"
+            )
+
     def test_transient_gives_up_over_cap(self):
         decision = self._strategy(max_transient=5).get_retry_decision(
             exception=httpx.ConnectError("refused"), job=_job(attempts=5)