Merge pull request #904 from cbcoutinho/feat/tiered-escalation-procrastinate

feat(ingest): per-tier escalation via procrastinate queue-hop

nextcloud_mcp_server/api/management.py

@@ -338,6 +338,10 @@ async def get_vector_sync_status(request: Request) -> JSONResponse:
         if pending.job_counts is not None:
             # Per-status breakdown (todo/doing/failed/…) on the postgres backend.
             body["job_counts"] = pending.job_counts
+        if pending.job_counts_by_queue is not None:
+            # Per-tier-queue breakdown (Deck #323): where work sits across the
+            # ingest-fast / ingest-structured / ingest-ocr fleets.
+            body["job_counts_by_queue"] = pending.job_counts_by_queue
         return JSONResponse(body)
 
     except Exception as e:

nextcloud_mcp_server/cli.py

@@ -333,8 +333,18 @@ def _init_worker_observability(settings: Settings) -> None:
     default=None,
     help="Max concurrent jobs. Defaults to VECTOR_SYNC_PROCESSOR_WORKERS.",
 )
-def worker(concurrency: int | None):
+@click.option(
-    """Run the ingest worker (Deck #183).
+    "--tier",
+    type=click.Choice(["fast", "structured", "ocr"]),
+    default=None,
+    help=(
+        "Run only this extraction tier's queue (Deck #323). Omit to drain ALL "
+        "tier queues in one process (single-Deployment / dev); set it to run one "
+        "tier per Deployment so the fleets scale independently."
+    ),
+)
+def worker(concurrency: int | None, tier: str | None):
+    """Run the ingest worker (Deck #183, per-tier fleets #323).
 
     \b
     Drains the per-tenant Postgres ingest queue (procrastinate): for each
@@ -342,6 +352,13 @@ def worker(concurrency: int | None):
     embeds, and upserts into Qdrant. This is the scale-to-zero ``worker`` role of
     the api/worker split; run it as a separate Deployment from the API pod.
 
+    \b
+    With --tier the worker drains only that tier's queue (``ingest-<tier>``), so
+    a CPU-bound ``fast`` fleet, an in-cluster ``structured`` fleet, and a paid
+    ``ocr`` fleet scale independently. Without it, all tier queues are drained in
+    one process (handy for dev / a single Deployment). A low-quality parse hops
+    the job to the next tier's queue automatically (see TieredEscalationStrategy).
+
     \b
     Requires INGEST_QUEUE=postgres (a PostgreSQL DATABASE_URL); procrastinate is
     Postgres-only.
@@ -349,7 +366,7 @@ def worker(concurrency: int | None):
     \b
     Example:
       $ export DATABASE_URL=postgresql+asyncpg://mcp:mcp@db/mcp
-      $ nextcloud-mcp-server worker -c 4
+      $ nextcloud-mcp-server worker -c 4 --tier fast
     """
     import anyio  # noqa: PLC0415
 
@@ -366,11 +383,25 @@ def worker(concurrency: int | None):
     _init_worker_observability(settings)
 
     from nextcloud_mcp_server.vector.queue.procrastinate import (  # noqa: PLC0415
-        INGEST_QUEUE_NAME,
+        ALL_INGEST_QUEUES,
+        INGEST_QUEUE_MAINTENANCE,
+        LEGACY_INGEST_QUEUE,
+        TIER_QUEUES,
         apply_ingest_queue_schema,
         get_procrastinate_app,
     )
 
+    # 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. 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], INGEST_QUEUE_MAINTENANCE]
+    else:
+        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
     # directly (run_worker_async), so it does NOT go through IngestTransport —
@@ -397,13 +428,15 @@ def worker(concurrency: int | None):
             # Structured log (not click.echo) so it lands in the JSON / OTel
             # pipeline like every other startup message.
             logger.info(
-                "Ingest worker started: queue=%s concurrency=%s delete_succeeded=%s",
+                "Ingest worker started: tier=%s queues=%s concurrency=%s "
-                INGEST_QUEUE_NAME,
+                "delete_succeeded=%s",
+                tier or "all",
+                queues,
                 workers,
                 settings.ingest_delete_succeeded_jobs,
             )
             await app.run_worker_async(
-                queues=[INGEST_QUEUE_NAME],
+                queues=queues,
                 concurrency=workers,
                 install_signal_handlers=True,
                 # Drop succeeded jobs (default) so the queue table stays lean and

nextcloud_mcp_server/config.py

@@ -234,6 +234,29 @@ _DEFAULTS: dict[str, Any] = {
     # queue-depth metric clean). Set false to retain succeeded rows for audit
     # (note: indexing success is also recorded in logs/metrics regardless).
     "ingest_delete_succeeded_jobs": True,
+    # Per-tier escalation on the procrastinate (postgres) ingest path (Deck
+    # #323). When true, a document that a tier cannot parse well is requeued onto
+    # the next tier's queue (fast -> structured -> ocr) via a native procrastinate
+    # queue-hop. When false the ``fast`` tier is terminal -- reproduces the
+    # pre-#323 behaviour where the cheap tier's output is indexed as-is. No effect
+    # on the in-process ``memory`` backend, which keeps the inline escalation.
+    # HOT: re-read per job (process_document_task), so it takes effect on the next
+    # job -- unlike INGEST_TRANSIENT_MAX_ATTEMPTS, which is snapshotted at worker
+    # startup and needs a restart.
+    "ingest_escalation_enabled": True,
+    # Global cap on SAME-tier retries for transient infra errors (doc fetch /
+    # embed / Qdrant blips) on the procrastinate path. Parse-quality failures
+    # escalate (one parse attempt per tier) and do NOT consume this budget; only
+    # whitelisted transient exceptions retry in place. Shared across tiers because
+    # a queue-hop cannot reset a per-tier counter (see TieredEscalationStrategy).
+    # Snapshotted at worker startup (blueprint build); restart to change it.
+    "ingest_transient_max_attempts": 5,
+    # Delay (seconds) before a reclaimed stalled job is re-run. A stall is often
+    # systemic (Qdrant/embedding outage), so reclaiming every crashed job at
+    # now() would thundering-herd a recovering dependency every reclaim tick
+    # (*/5min), bypassing TieredEscalationStrategy's per-job backoff. A small
+    # fixed delay staggers the retry. 0 = immediate (legacy behaviour).
+    "ingest_reclaim_retry_delay_seconds": 30,
     "collection_metadata_source": "qdrant",  # qdrant | api
     # CP base URL for COLLECTION_METADATA_SOURCE=api (e.g. http://control-plane).
     # Required only when the source is api.
@@ -317,6 +340,8 @@ _dynaconf = Dynaconf(
         Validator("METRICS_PORT", gte=1, lte=65535),
         # Positive integers
         Validator("INGEST_STALLED_JOB_SECONDS", gte=1),
+        Validator("INGEST_TRANSIENT_MAX_ATTEMPTS", gte=1),
+        Validator("INGEST_RECLAIM_RETRY_DELAY_SECONDS", gte=0),
         Validator("VECTOR_SYNC_SCAN_INTERVAL", gte=1),
         Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1),
         Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1),
@@ -856,6 +881,9 @@ class Settings:
     mcp_role: str = "all"  # api | worker | all (Deck #183 two-pod model)
     ingest_stalled_job_seconds: int = 300  # crashed-worker reclaim threshold
     ingest_delete_succeeded_jobs: bool = True  # drop succeeded ingest jobs
+    ingest_escalation_enabled: bool = True  # per-tier queue-hop (Deck #323)
+    ingest_transient_max_attempts: int = 5  # same-tier transient-retry cap
+    ingest_reclaim_retry_delay_seconds: int = 30  # stagger reclaimed-job retries
     collection_metadata_source: str = "qdrant"  # qdrant | api
     collection_metadata_api_url: str | None = None  # CP URL when source=api
     embedding_gateway_url: str | None = None  # required when provider=gateway
@@ -1481,6 +1509,9 @@ def get_settings() -> Settings:
         "mcp_role": "MCP_ROLE",
         "ingest_stalled_job_seconds": "INGEST_STALLED_JOB_SECONDS",
         "ingest_delete_succeeded_jobs": "INGEST_DELETE_SUCCEEDED_JOBS",
+        "ingest_escalation_enabled": "INGEST_ESCALATION_ENABLED",
+        "ingest_transient_max_attempts": "INGEST_TRANSIENT_MAX_ATTEMPTS",
+        "ingest_reclaim_retry_delay_seconds": "INGEST_RECLAIM_RETRY_DELAY_SECONDS",
         "collection_metadata_source": "COLLECTION_METADATA_SOURCE",
         "collection_metadata_api_url": "COLLECTION_METADATA_API_URL",
         "embedding_gateway_url": "EMBEDDING_GATEWAY_URL",

nextcloud_mcp_server/document_processors/escalation.py

@@ -0,0 +1,69 @@
+"""Tier-escalation ladder + signal for the per-tier ingest fleet (Deck #323).
+
+The escalation ladder is the cheapest-first ordering of extraction tiers:
+
+    fast  ->  structured  ->  ocr   ( ->  llm, reserved)
+
+It mirrors the ``tier`` vocabulary documented on
+:meth:`DocumentProcessor.tier <.base.DocumentProcessor.tier>` and the
+observability label set. On the *external* (procrastinate) ingest path each tier
+runs on its own queue + worker fleet; a document that a tier cannot parse well is
+**requeued onto the next tier's queue** rather than escalated inline. The
+mechanism is a raised :class:`EscalateError` that the procrastinate retry
+strategy turns into a native ``RetryDecision(queue=<next-tier queue>)`` queue-hop
+(see ``vector/queue/procrastinate.py``).
+
+This module is deliberately free of any queue/transport dependency: it only
+knows the *tier* vocabulary and the escalation signal. The tier -> queue-name
+mapping lives in the queue layer, which imports :class:`EscalateError` from here
+(document_processors never imports vector.queue, so there is no import cycle).
+"""
+
+from __future__ import annotations
+
+# Cheapest-first. ``llm`` is reserved (see base.DocumentProcessor.tier) and not
+# wired yet, so it is intentionally absent from the live ladder.
+TIER_LADDER: tuple[str, ...] = ("fast", "structured", "ocr")
+
+
+def next_tier(current: str) -> str | None:
+    """The next tier above ``current`` in the ladder, or ``None`` if terminal.
+
+    Pure ordering only -- it does not consider whether the next tier is
+    *available* (a processor registered / OCR enabled). **Production routing uses
+    ``ProcessorRegistry.next_available_tier``**, which layers availability on top
+    of this ordering; ``next_tier`` itself is the underlying building block
+    (referenced directly by tests). A tier with no escalation target is terminal
+    and its result is indexed as-is.
+    """
+    try:
+        idx = TIER_LADDER.index(current)
+    except ValueError:
+        return None
+    nxt = idx + 1
+    return TIER_LADDER[nxt] if nxt < len(TIER_LADDER) else None
+
+
+class EscalateError(Exception):
+    """Raised when a tier's parse is too poor to index and a higher tier exists.
+
+    Carries the tiers + reason so the procrastinate retry strategy can hop the
+    job to the next tier's queue and record
+    ``astrolabe_document_escalation_total{from_tier,to_tier,reason}``. It is a
+    control-flow signal, NOT a failure: it must propagate *before* chunk/embed so
+    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. This PR raises
+    ``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:
+        self.from_tier = from_tier
+        self.to_tier = to_tier
+        self.reason = reason
+        super().__init__(
+            f"escalate {from_tier}->{to_tier} (reason={reason})",
+        )

nextcloud_mcp_server/document_processors/registry.py

@@ -14,7 +14,8 @@ from nextcloud_mcp_server.observability.metrics import (
 from nextcloud_mcp_server.observability.tracing import trace_operation
 
 from .base import DocumentProcessor, ProcessingResult, ProcessorError
-from .classifier import classify_from_text, image_coverage_per_page
+from .classifier import DocClassification, classify_from_text, image_coverage_per_page
+from .escalation import TIER_LADDER
 
 logger = logging.getLogger(__name__)
 
@@ -202,32 +203,9 @@ class ProcessorRegistry:
         """
         settings = get_settings()
 
-        # Pre-parse size guard: a pathologically large PDF (e.g. a 42 MB scanned
+        oversize = self._oversize_result(content, filename, settings)
-        # DUDE) burns the OCR timeout for 0 chars. Fail fast with an explicit
+        if oversize is not None:
-        # reason so the caller marks the placeholder "failed" instead of
+            return oversize
-        # retrying. 0 disables the cap. This lives on the auto-tiered path only:
-        # an explicit processor_name="ocr" override (registry.process) bypasses
-        # _process_pdf entirely and is intentionally not size-gated (power-user
-        # escape hatch). Returning here also skips _run_processor, so the
-        # rejection is counted on astrolabe_document_parse_failed_total{oversize}
-        # (via vector/processor.py) but deliberately not on the parse-duration
-        # histogram -- there is no parse to time.
-        max_pdf_mb = settings.document_max_pdf_size_mb
-        if max_pdf_mb > 0 and len(content) > max_pdf_mb * 1024 * 1024:
-            size_mb = len(content) / (1024 * 1024)
-            logger.warning(
-                "PDF %s is %.1f MB (> %.1f MB cap); failing fast as oversize",
-                filename or "<bytes>",
-                size_mb,
-                max_pdf_mb,
-            )
-            return ProcessingResult(
-                text="",
-                metadata={"parse_failed_reason": "oversize"},
-                processor="size_guard",
-                success=False,
-                error=(f"PDF exceeds size cap: {size_mb:.1f} MB > {max_pdf_mb:.1f} MB"),
-            )
 
         if settings.document_tier1_engine == "pymupdf":
             processor = self._pdf_processor_for_tier("structured")
@@ -263,46 +241,11 @@ class ProcessorRegistry:
         # Tier-0 classification from the extraction (cheap: text-only, no PDF
         # re-open). Scan detection (image analysis, re-opens the PDF) runs only
         # when OCR + detect_scanned are enabled, so its cost is paid by
-        # OCR-opted-in tenants only.
+        # OCR-opted-in tenants only. Shared with the external per-tier path via
-        classification = None
+        # _classify_result.
-        if settings.document_classify_enabled and result.success:
+        classification = self._classify_result(
-            try:
+            result, content, settings, record=True, filename=filename
-                image_coverage = None
+        )
-                if (
-                    settings.document_ocr_enabled
-                    and settings.document_ocr_detect_scanned
-                ):
-                    try:
-                        image_coverage = image_coverage_per_page(content)
-                    except Exception:
-                        # Best-effort: fall back to text-only signals. WARNING
-                        # (not DEBUG) so a systematic scan-detection failure on an
-                        # OCR-enabled tenant is visible at LOG_LEVEL=INFO.
-                        logger.warning(
-                            "Scan detection failed for %s; using text-only signals",
-                            filename or "<bytes>",
-                            exc_info=True,
-                        )
-                classification = classify_from_text(
-                    result.text,
-                    result.metadata.get("page_boundaries") or [],
-                    min_text_quality=settings.document_ocr_min_text_quality,
-                    min_page_chars=settings.document_ocr_min_page_chars,
-                    page_fraction=settings.document_ocr_page_fraction,
-                    image_coverage=image_coverage,
-                )
-                record_document_classification(
-                    classification.recommended_tier,
-                    classification.flags,
-                    classification.mean_text_quality,
-                    classification.ocr_page_fraction,
-                )
-            except Exception:
-                logger.warning(
-                    "Tier-0 classification failed for %s",
-                    filename or "<bytes>",
-                    exc_info=True,
-                )
 
         # Escalate scanned / no-text-layer PDFs to OCR (tier-3) when enabled and
         # a provider is registered. The fast tier is terminal otherwise. Note: a
@@ -355,6 +298,225 @@ class ProcessorRegistry:
 
         return result
 
+    def _oversize_result(
+        self, content: bytes, filename: str | None, settings: Any
+    ) -> ProcessingResult | None:
+        """Pre-parse size guard, shared by the inline and per-tier paths.
+
+        A pathologically large PDF (e.g. a 42 MB scanned DUDE) burns the OCR
+        timeout for 0 chars. Return an explicit ``oversize`` failure so the
+        caller marks the placeholder "failed" instead of retrying; 0 disables the
+        cap. An explicit ``processor_name`` override (``registry.process``)
+        bypasses tiering entirely and is intentionally not size-gated (power-user
+        escape hatch). Skipping ``_run_processor`` means the rejection is counted
+        on ``astrolabe_document_parse_failed_total{oversize}`` (via
+        ``vector/processor.py``) but deliberately not on the parse-duration
+        histogram -- there is no parse to time.
+        """
+        max_pdf_mb = settings.document_max_pdf_size_mb
+        if max_pdf_mb > 0 and len(content) > max_pdf_mb * 1024 * 1024:
+            size_mb = len(content) / (1024 * 1024)
+            logger.warning(
+                "PDF %s is %.1f MB (> %.1f MB cap); failing fast as oversize",
+                filename or "<bytes>",
+                size_mb,
+                max_pdf_mb,
+            )
+            return ProcessingResult(
+                text="",
+                metadata={"parse_failed_reason": "oversize"},
+                processor="size_guard",
+                success=False,
+                error=(f"PDF exceeds size cap: {size_mb:.1f} MB > {max_pdf_mb:.1f} MB"),
+            )
+        return None
+
+    def _classify_result(
+        self,
+        result: ProcessingResult,
+        content: bytes,
+        settings: Any,
+        *,
+        record: bool,
+        filename: str | None = None,
+    ) -> DocClassification | None:
+        """Tier-0 classification of a parse result (text-only, cheap).
+
+        Shared by the inline memory-backend pipeline (:meth:`_process_pdf`) and
+        the external per-tier path (:meth:`evaluate_escalation`). Returns
+        ``None`` when classification is disabled, the parse failed, or the
+        classifier raised -- best-effort, a classify failure must never break
+        indexing. ``record`` emits the classification metrics; set it only at the
+        FIRST classification of a document (the ``fast`` tier) so the per-doc
+        counters aren't multiplied across tiers.
+        """
+        if not (settings.document_classify_enabled and result.success):
+            return None
+        try:
+            image_coverage = None
+            if settings.document_ocr_enabled and settings.document_ocr_detect_scanned:
+                try:
+                    image_coverage = image_coverage_per_page(content)
+                except Exception:
+                    # Best-effort: fall back to text-only signals. WARNING (not
+                    # DEBUG) so a systematic scan-detection failure on an
+                    # OCR-enabled tenant is visible at LOG_LEVEL=INFO.
+                    logger.warning(
+                        "Scan detection failed for %s; using text-only signals",
+                        filename or "<bytes>",
+                        exc_info=True,
+                    )
+            classification = classify_from_text(
+                result.text,
+                result.metadata.get("page_boundaries") or [],
+                min_text_quality=settings.document_ocr_min_text_quality,
+                min_page_chars=settings.document_ocr_min_page_chars,
+                page_fraction=settings.document_ocr_page_fraction,
+                image_coverage=image_coverage,
+            )
+        except Exception:
+            logger.warning(
+                "Tier-0 classification failed for %s",
+                filename or "<bytes>",
+                exc_info=True,
+            )
+            return None
+        if record:
+            record_document_classification(
+                classification.recommended_tier,
+                classification.flags,
+                classification.mean_text_quality,
+                classification.ocr_page_fraction,
+            )
+        return classification
+
+    def _tier_available(self, tier: str, settings: Any) -> bool:
+        """Whether ``tier`` can actually run a PDF parse right now.
+
+        A tier is available when it has a registered PDF processor and is
+        enabled; the ``ocr`` tier additionally requires ``DOCUMENT_OCR_ENABLED``
+        (so OCR stays opt-in and a misconfigured tenant never escalates to a
+        backend it hasn't turned on).
+        """
+        if self._pdf_processor_for_tier(tier) is None:
+            return False
+        if tier == "ocr" and not settings.document_ocr_enabled:
+            return False
+        return True
+
+    def next_available_tier(
+        self, current_tier: str, settings: Any, *, minimum: str | None = None
+    ) -> str | None:
+        """First escalation target above ``current_tier`` that can actually run.
+
+        Walks the ladder strictly above ``current_tier`` (and not below
+        ``minimum``'s rung, when given) and returns the first
+        :meth:`_tier_available` tier. ``None`` means no higher tier can run --
+        ``current_tier`` is then terminal and its result is indexed as-is.
+        """
+        try:
+            cur_idx = TIER_LADDER.index(current_tier)
+        except ValueError:
+            return None
+        start_idx = cur_idx + 1
+        if minimum is not None:
+            try:
+                start_idx = max(start_idx, TIER_LADDER.index(minimum))
+            except ValueError:
+                pass
+        for tier in TIER_LADDER[start_idx:]:
+            if self._tier_available(tier, settings):
+                return tier
+        return None
+
+    async def process_tier(
+        self,
+        content: bytes,
+        content_type: str,
+        filename: str | None,
+        tier: str,
+        options: dict[str, Any] | None = None,
+        progress_callback: (
+            Callable[[float, float | None, str | None], Awaitable[None]] | None
+        ) = None,
+    ) -> ProcessingResult:
+        """Run exactly ONE extraction tier's processor on a PDF (external path).
+
+        The per-tier procrastinate fleet calls this for the tier matching the
+        job's queue. Escalation to the next tier is decided separately by
+        :meth:`evaluate_escalation` and effected by the queue's retry strategy as
+        a queue-hop -- never inline here. ``escalated`` is set for any tier above
+        the cheapest so the parse span/metrics reflect an escalated attempt.
+        """
+        oversize = self._oversize_result(content, filename, get_settings())
+        if oversize is not None:
+            return oversize
+        processor = self._pdf_processor_for_tier(tier)
+        if processor is None:
+            raise ProcessorError(
+                f"No '{tier}'-tier PDF processor registered "
+                f"(available: {', '.join(self.list_processors())})"
+            )
+        return await self._run_processor(
+            processor,
+            content,
+            content_type,
+            filename,
+            options,
+            progress_callback,
+            escalated=(tier != TIER_LADDER[0]),
+        )
+
+    def evaluate_escalation(
+        self,
+        result: ProcessingResult,
+        content: bytes,
+        current_tier: str,
+        settings: Any,
+        *,
+        filename: str | None = None,
+    ) -> tuple[str, str] | None:
+        """Decide whether ``current_tier``'s result must escalate (external path).
+
+        Returns ``(to_tier, reason)`` when the parse is too poor to index and a
+        higher tier can run, else ``None`` (index the result as-is). Reuses the
+        tier-0 classifier as the post-parse quality gate, so the escalation
+        signal is identical to the inline pipeline's.
+
+        A hard parse FAILURE (``result.success`` False) is never escalated: a
+        corrupt/encrypted PDF one engine can't open usually defeats the others
+        too (OCR reads the same bytes), so the caller marks it failed instead.
+
+        Routing of the target tier:
+
+        - ``total_chars == 0`` (scanned / no text layer) -> target the ``ocr``
+          tier directly. Text-extractor tiers (``structured``) cannot conjure
+          text from a pure raster scan, so a structured hop would just be wasted.
+        - low-confidence but non-empty layer -> escalate to the next rung, so a
+          different in-cluster extractor can try before paying for OCR.
+        """
+        classification = self._classify_result(
+            result,
+            content,
+            settings,
+            record=(current_tier == TIER_LADDER[0]),
+            filename=filename,
+        )
+        if classification is None or classification.recommended_tier != "ocr":
+            return None
+        # A zero-page (empty/corrupt) PDF gains nothing from any tier.
+        if classification.page_count <= 0:
+            return None
+        if classification.total_chars == 0:
+            to_tier = self.next_available_tier(current_tier, settings, minimum="ocr")
+            reason = "empty_text"
+        else:
+            to_tier = self.next_available_tier(current_tier, settings)
+            reason = "low_confidence"
+        if to_tier is None:
+            return None
+        return (to_tier, reason)
+
     async def _run_processor(
         self,
         processor: DocumentProcessor,

nextcloud_mcp_server/models/semantic.py

@@ -193,6 +193,15 @@ class VectorSyncStatusResponse(BaseResponse):
             "queue backend; None on the in-memory backend"
         ),
     )
+    job_counts_by_queue: dict[str, dict[str, int]] | None = Field(
+        default=None,
+        description=(
+            "Per-tier-queue ingest job counts {queue: {status: count}} on the "
+            "postgres backend (Deck #323), so an operator can see whether work is "
+            "backed up on ingest-fast vs waiting on ingest-structured/ingest-ocr; "
+            "None on the in-memory backend"
+        ),
+    )
 
 
 __all__ = [

nextcloud_mcp_server/observability/metrics.py

@@ -190,6 +190,21 @@ vector_sync_indexed_chunks = Gauge(
     "Total indexed chunks (non-placeholder points) in the vector store",
 )
 
+# Per-tier-queue ingest depth (Deck #323). One series per (queue, status) so an
+# operator can see where work sits -- a ``fast`` backlog, docs waiting on
+# ``ingest-structured``/``ingest-ocr``, or failures piling up per tier. KEDA
+# scales each tier Deployment off the queue's ``todo`` depth via direct SQL; this
+# gauge is the dashboard/alerting view of the same figures. Published by the
+# periodic vector_sync metrics task from the procrastinate per-queue job counts.
+ingest_queue_depth = Gauge(
+    "astrolabe_ingest_queue_depth",
+    "Ingest jobs per tier queue by status (todo/doing/failed)",
+    ["queue", "status"],
+)
+# The subset of statuses worth a gauge series; the rest (succeeded/cancelled/
+# aborted) are pruned from the queue table and uninteresting for operating.
+_INGEST_DEPTH_STATUSES = ("todo", "doing", "failed")
+
 qdrant_operations_total = Counter(
     "mcp_qdrant_operations_total",
     "Total Qdrant vector database operations",
@@ -637,6 +652,42 @@ def update_vector_sync_indexed_chunks(count: int) -> None:
     vector_sync_indexed_chunks.set(count)
 
 
+def update_ingest_queue_depth(by_queue: dict[str, dict[str, int]] | None) -> None:
+    """Set the per-tier-queue depth gauge from procrastinate job counts (#323).
+
+    ``by_queue`` is ``{queue_name: {status: count}}`` (see
+    ``queue.procrastinate.get_ingest_job_counts_by_queue``). No-op only on the
+    memory backend (``by_queue is None``); an empty dict (postgres backend with
+    every queue drained) still runs the pre-zero so the gauge reads 0.
+
+    Every managed queue is zeroed first: ``list_queues_async`` stops returning a
+    queue once it has no jobs, so a queue that drained to empty drops out of
+    ``by_queue`` entirely (and when ALL drain, ``by_queue`` is ``{}``). Without
+    the pre-zero its gauge series would stick at its last non-zero value (ghost
+    backlog in Grafana/alerts) instead of reading 0. The live counts then
+    overwrite the zeros for queues that still have work.
+    """
+    # ``is None`` not ``not by_queue``: an empty dict means "postgres, all queues
+    # drained" and MUST still zero the gauge -- only None (memory) is the no-op.
+    if by_queue is None:
+        return
+    # Lazy import to keep observability decoupled from the queue layer at module
+    # load (and sidestep any import cycle); both names are public constants.
+    from nextcloud_mcp_server.vector.queue.procrastinate import (  # noqa: PLC0415
+        ALL_INGEST_QUEUES,
+        LEGACY_INGEST_QUEUE,
+    )
+
+    for queue in (*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE):
+        for status in _INGEST_DEPTH_STATUSES:
+            ingest_queue_depth.labels(queue=queue, status=status).set(0)
+    for queue, per_status in by_queue.items():
+        for status in _INGEST_DEPTH_STATUSES:
+            ingest_queue_depth.labels(queue=queue, status=status).set(
+                per_status.get(status, 0)
+            )
+
+
 def record_document_parse(
     processor: str,
     tier: str,

nextcloud_mcp_server/server/semantic.py

@@ -1099,6 +1099,7 @@ def configure_semantic_tools(mcp: FastMCP):
                 enabled=True,
                 ingest_queue=settings.ingest_queue,
                 job_counts=pending.job_counts,
+                job_counts_by_queue=pending.job_counts_by_queue,
             )
 
         except Exception as e:

nextcloud_mcp_server/vector/ingest_status.py

@@ -29,6 +29,10 @@ class IngestPending:
     # Per-status counts (todo/doing/failed/…) on the postgres backend; None on
     # the memory backend, which has no durable per-status breakdown.
     job_counts: dict[str, int] | None = None
+    # Per-tier-queue breakdown ``{queue: {status: count}}`` on the postgres
+    # backend (Deck #323); None on the memory backend. Feeds the per-tier status
+    # surface + the astrolabe_ingest_queue_depth gauge.
+    job_counts_by_queue: dict[str, dict[str, int]] | None = None
 
 
 async def get_ingest_pending(
@@ -47,13 +51,27 @@ async def get_ingest_pending(
     """
     if ingest_queue == "postgres":
         counts: dict[str, int] = {}
-        if task_producer is not None and hasattr(task_producer, "job_counts"):
+        by_queue: dict[str, dict[str, int]] | None = None
+        # Prefer the per-queue breakdown (Deck #323) and aggregate from it, so the
+        # fleet-wide totals and the per-tier view always agree. Fall back to the
+        # aggregated call for any producer that predates job_counts_by_queue.
+        if task_producer is not None and hasattr(task_producer, "job_counts_by_queue"):
+            try:
+                by_queue = await task_producer.job_counts_by_queue()
+                for per_status in by_queue.values():
+                    for status, value in per_status.items():
+                        counts[status] = counts.get(status, 0) + value
+            except Exception as e:
+                logger.warning("Failed to read ingest job counts by queue: %s", e)
+        elif task_producer is not None and hasattr(task_producer, "job_counts"):
             try:
                 counts = await task_producer.job_counts()
             except Exception as e:
                 logger.warning("Failed to read ingest job counts: %s", e)
         pending = counts.get("todo", 0) + counts.get("doing", 0)
-        return IngestPending(pending=pending, job_counts=counts)
+        return IngestPending(
+            pending=pending, job_counts=counts, job_counts_by_queue=by_queue
+        )
 
     if document_receive_stream is None:
         return IngestPending(pending=0)

nextcloud_mcp_server/vector/metrics_publisher.py

@@ -29,6 +29,7 @@ from qdrant_client.models import FieldCondition, Filter, MatchValue
 
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.observability.metrics import (
+    update_ingest_queue_depth,
     update_vector_sync_indexed_chunks,
     update_vector_sync_indexed_documents,
     update_vector_sync_pending_documents,
@@ -96,6 +97,8 @@ async def publish_vector_sync_metrics(
         # Keep the legacy gauge meaningful on every consumer path, not just the
         # single-user one — existing dashboards/alerts reference it.
         update_vector_sync_queue_size(pending.pending)
+        # Per-tier-queue depth (Deck #323): None on the memory backend (no-op).
+        update_ingest_queue_depth(pending.job_counts_by_queue)
     except Exception as exc:  # noqa: BLE001 — metrics must not break ingest
         logger.warning("Failed to publish pending-documents gauge: %s", exc)
 

nextcloud_mcp_server/vector/processor.py

@@ -6,7 +6,7 @@ Processes documents from stream: fetches content, generates embeddings, stores i
 import logging
 import time
 import uuid
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import anyio
 import httpx
@@ -14,6 +14,12 @@ from anyio.abc import TaskStatus
 from anyio.streams.memory import MemoryObjectReceiveStream
 from qdrant_client.models import PointStruct
 
+if TYPE_CHECKING:
+    # Type-only: the document stack is heavy (pymupdf/_isolation) and must stay
+    # off processor.py's import path (#877); the runtime import is lazy.
+    from nextcloud_mcp_server.document_processors.base import ProcessingResult
+    from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry
+
 from nextcloud_mcp_server.acl_hash import compute_acl_hash
 from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.config import get_settings
@@ -21,6 +27,7 @@ from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_servi
 from nextcloud_mcp_server.models.deck import DeckCard
 from nextcloud_mcp_server.observability.metrics import (
     record_document_chunks,
+    record_document_escalation,
     record_document_parse_failed,
     record_embedding,
     record_embedding_tokens,
@@ -106,6 +113,63 @@ def _drop_reason(exc: BaseException) -> str:
     return "other"
 
 
+def _is_pdf(content_type: str) -> bool:
+    """Whether a MIME type is a PDF (parameter-tolerant)."""
+    return content_type.split(";")[0].strip().lower() == "application/pdf"
+
+
+async def _parse_pdf_tier(
+    registry: "ProcessorRegistry",
+    content: bytes,
+    content_type: str,
+    filename: str | None,
+    tier: str,
+    settings: Any,
+) -> "ProcessingResult":
+    """Run a single extraction tier and apply the post-parse escalation gate.
+
+    The external per-tier ingest path (Deck #323): the procrastinate worker for
+    ``tier`` parses with exactly that tier, then either returns the result to
+    index or raises ``EscalateError`` to hand the document to the next tier's
+    queue (the queue's retry strategy turns the raise into a native queue-hop).
+    The escalation metric is recorded here, at the decision point.
+
+    A hard parse failure (``result.success`` False) is returned as-is, not
+    escalated -- a corrupt/encrypted/oversize PDF that one engine can't open
+    usually defeats the others too; the caller marks it failed. This preserves
+    the "OCR is an enhancement, never worse than off" invariant: a tenant who has
+    not enabled a higher tier (or has no processor for it) simply indexes the
+    cheap tier's output.
+    """
+    # Lazy import: keep the document stack (pymupdf/_isolation) off the module
+    # load path; this runs only on the per-tier worker, which needs it anyway.
+    from nextcloud_mcp_server.document_processors.escalation import (  # noqa: PLC0415
+        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(
+            result, content, tier, settings, filename=filename
+        )
+        if decision is not None:
+            to_tier, reason = decision
+            record_document_escalation(tier, to_tier, reason)
+            logger.info(
+                "Escalating %s %s->%s (reason=%s)",
+                filename or "<bytes>",
+                tier,
+                to_tier,
+                reason,
+            )
+            raise EscalateError(from_tier=tier, to_tier=to_tier, reason=reason)
+    return result
+
+
 def assign_page_numbers(chunks, page_boundaries):
     """Assign page numbers to chunks based on page boundaries.
 
@@ -173,6 +237,7 @@ async def record_indexing_usage(
     token_count: int,
     total_chars: int,
     page_count: int | None,
+    pipeline_tier: str | None = None,
 ) -> None:
     """Record the billable usage events for one embedded document.
 
@@ -213,6 +278,11 @@ async def record_indexing_usage(
         "doc_type": doc_type,
         "user_id": user_id,
         "total_chars": total_chars,
+        # Which extraction tier produced the parsed pages (Deck #323). Carried so
+        # the CP rollup / a future per-tier price can attribute parsing cost to
+        # the tier that incurred it (paid OCR vs CPU-cheap fast). None for text
+        # doc types, which are never parsed.
+        "pipeline_tier": pipeline_tier,
     }
     try:
         store = await UsageEventStore.shared()
@@ -242,6 +312,18 @@ async def record_indexing_usage(
                 metadata=metadata,
                 enabled=True,
             )
+            # Paid-OCR pages are metered as a SEPARATE line (Deck #323) so the
+            # expensive tier's cost is billable independently of CPU-cheap parsing
+            # -- pages_embedded counts all parsed pages, pages_ocr only the OCR
+            # tier's. Gated on the tier so it's emitted exactly when the doc was
+            # actually OCR'd; the same page_count guard above applies.
+            if pipeline_tier == "ocr":
+                await store.record_usage_event(
+                    metric="pages_ocr",
+                    value=page_count,
+                    metadata=metadata,
+                    enabled=True,
+                )
     except Exception:
         # Reached only when shared()/store construction itself raises
         # (record_usage_event swallows its own write failures). Metering is on,
@@ -393,7 +475,11 @@ async def _reconcile_tag_event(
 
 
 async def process_document(
-    doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3
+    doc_task: DocumentTask,
+    nc_client: NextcloudClient,
+    *,
+    max_retries: int = 3,
+    tier: str | None = None,
 ):
     """
     Process a single document: fetch, tokenize, embed, store in Qdrant.
@@ -407,6 +493,11 @@ async def process_document(
             (3) suits the in-process SQLite pool, which has no durable retry. The
             procrastinate worker passes ``1`` so durable retry is owned by the
             queue (and survives worker crashes), avoiding compounding 3×N retries.
+        tier: Extraction tier to run for PDFs on the external per-tier path (Deck
+            #323) -- the procrastinate worker passes the tier matching its queue.
+            ``None`` (the default, used by the in-process/memory pool) runs the
+            inline tiered pipeline (``registry.process``: fast -> OCR escalation
+            in one call) and never raises ``EscalateError``.
 
     Retry layering: the embedding provider adds its own transient retry (5
     attempts, 2s→60s backoff — card 309) *inside* each of these attempts. On the
@@ -415,6 +506,20 @@ async def process_document(
     re-picked on the next scan; the procrastinate path (max_retries=1) caps it at
     one outer attempt (~30s) and defers. Don't stack a third retry layer here.
     """
+    # EscalateError is a control-flow signal that arises ONLY on the per-tier
+    # external path (tier set). Bind the class lazily here, and only when a tier
+    # is set, so the document stack is never imported at *module load* (the #877
+    # invariant) nor on the delete / text-doc call paths (file processing already
+    # imports it via get_registry regardless). When tier is None it can't be
+    # raised, so the guards below stay inert.
+    escalate_error_cls: type[BaseException] | None = None
+    if tier is not None:
+        from nextcloud_mcp_server.document_processors.escalation import (  # noqa: PLC0415
+            EscalateError,
+        )
+
+        escalate_error_cls = EscalateError
+
     start_time = time.time()
 
     logger.debug(
@@ -484,7 +589,9 @@ async def process_document(
 
             for attempt in range(max_retries):
                 try:
-                    indexed = await _index_document(doc_task, nc_client, qdrant_client)
+                    indexed = await _index_document(
+                        doc_task, nc_client, qdrant_client, tier=tier
+                    )
 
                     # A permanent parse failure returns False: it was already
                     # recorded (document_parse_failed_total + the registry's
@@ -506,6 +613,14 @@ async def process_document(
                     return  # Success
 
                 except Exception as e:
+                    # An escalation signal is control flow, not a failure:
+                    # propagate it untouched so the procrastinate retry strategy
+                    # can hop the job to the next tier's queue. Never retry it
+                    # in-process and never count it as a drop.
+                    if escalate_error_cls is not None and isinstance(
+                        e, escalate_error_cls
+                    ):
+                        raise
                     if attempt < max_retries - 1:
                         logger.warning(
                             "Retry %s/%s for %s_%s: %s",
@@ -556,7 +671,12 @@ async def process_document(
                         record_ingest_dropped(reason)
                         raise
 
-        except Exception:
+        except Exception as e:
+            # An escalation signal must reach the procrastinate retry strategy
+            # un-recorded -- it is neither a processing success nor an error
+            # (the hop is its own event, counted via record_document_escalation).
+            if escalate_error_cls is not None and isinstance(e, escalate_error_cls):
+                raise
             # Single processing-error call site: catches exhausted-retry
             # re-raises, delete failures, and setup errors (get_qdrant_client /
             # get_settings) — each counted exactly once. A failed delete is not
@@ -571,11 +691,20 @@ async def process_document(
 
 
 async def _index_document(
-    doc_task: DocumentTask, nc_client: NextcloudClient, qdrant_client
+    doc_task: DocumentTask,
+    nc_client: NextcloudClient,
+    qdrant_client,
+    *,
+    tier: str | None = None,
 ) -> bool | None:
     """
     Index a single document (called by process_document with retry).
 
+    ``tier`` selects the external per-tier PDF path (Deck #323): when set and the
+    file is a PDF, exactly that tier is parsed and a low-quality result raises
+    ``EscalateError`` to hand the document to the next tier's queue. ``None``
+    (default) runs the inline tiered pipeline (``registry.process``).
+
     Returns ``False`` when a permanent parse failure means nothing was indexed
     (the caller must then skip the success metrics); ``None`` otherwise.
 
@@ -800,22 +929,40 @@ async def _index_document(
                 "vector_sync.file_size": len(content_bytes),
             },
         ):
-            # The registry runs the tiered PDF pipeline (tier-0 classify ->
+            # The registry runs the tiered PDF pipeline and records
-            # tier-1 fast -> OCR escalation) and records classification metrics.
+            # classification metrics. Imported lazily so module import doesn't
-            # Imported lazily so module import doesn't pull in the document stack
+            # pull in the document stack (document_processors -> _isolation,
-            # (document_processors -> _isolation, Unix-only ``resource``; see #877).
+            # Unix-only ``resource``; see #877).
             from nextcloud_mcp_server.document_processors import (  # noqa: PLC0415
                 get_registry,
             )
+            from nextcloud_mcp_server.document_processors.escalation import (  # noqa: PLC0415
+                EscalateError,
+            )
 
             registry = get_registry()
 
             try:
-                result = await registry.process(
+                # External per-tier path (Deck #323): run only this worker's tier
-                    content=content_bytes,
+                # for PDFs and let a low-quality parse raise EscalateError (a
-                    content_type=content_type,
+                # queue-hop to the next tier). Everything else -- non-PDF files,
-                    filename=file_path,
+                # and the in-process/memory pool (tier is None) -- runs the inline
-                )
+                # tiered pipeline (fast -> OCR escalation in one call).
+                if tier is not None and _is_pdf(content_type):
+                    result = await _parse_pdf_tier(
+                        registry,
+                        content_bytes,
+                        content_type,
+                        file_path,
+                        tier,
+                        settings,
+                    )
+                else:
+                    result = await registry.process(
+                        content=content_bytes,
+                        content_type=content_type,
+                        filename=file_path,
+                    )
 
                 # A permanent parse failure (e.g. an isolated-worker OOM/timeout
                 # on a pathological PDF) returns success=False rather than
@@ -881,6 +1028,11 @@ async def _index_document(
                             )
                 else:
                     logger.debug("No page_boundaries in metadata for %s", file_path)
+            except EscalateError:
+                # Control-flow signal (per-tier path): re-raise untouched so the
+                # queue hops the job to the next tier. NOT a "failed to process"
+                # error -- don't log it as one.
+                raise
             except Exception as e:
                 logger.error("Failed to process file %s: %s", file_path, e)
                 raise
@@ -1037,6 +1189,14 @@ async def _index_document(
                     and not isinstance(raw_page_count, bool)
                     else None
                 ),
+                # Tier that produced the parsed pages (registry stamps it on the
+                # result metadata); text doc types stay "fast". Narrow defensively
+                # to str|None — file_metadata is loosely typed (Any values).
+                pipeline_tier=(
+                    pt
+                    if isinstance(pt := file_metadata.get("pipeline_tier"), str)
+                    else None
+                ),
             )
 
     async def generate_sparse_embeddings():

nextcloud_mcp_server/vector/queue/procrastinate.py

@@ -5,11 +5,14 @@ This replaces NATS JetStream and the old Postgres-queue stub. The MCP server now
 owns *both* sides of ingest:
 
 - **Producer** (API role / scanner) — :class:`ProcrastinateTaskProducer.send`
-  *defers* one ``ingest:process_document`` job per changed document into the
+  *defers* one ``ingest:process_document`` job per changed document onto the
-  per-tenant Postgres (the same app DB; procrastinate manages its own tables).
+  cheapest tier's queue (``ingest-fast``) in the per-tenant Postgres (the same
-- **Consumer** (worker role) — ``nextcloud-mcp-server worker`` runs
+  app DB; procrastinate manages its own tables).
-  :func:`procrastinate.App.run_worker`, which drains the ``ingest`` queue and
+- **Consumer** (worker role) — ``nextcloud-mcp-server worker [--tier T]`` runs
-  invokes the existing :func:`process_document` pipeline.
+  :func:`procrastinate.App.run_worker`, which drains its tier's queue and invokes
+  the existing :func:`process_document` pipeline. A parse too poor to index hops
+  the job to the next tier's queue (see :class:`TieredEscalationStrategy`), so
+  cheap CPU parsing and paid OCR run on independently-scaled fleets (Deck #323).
 
 Design notes:
 
@@ -31,13 +34,21 @@ from __future__ import annotations
 
 import logging
 from dataclasses import asdict
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from types import TracebackType
 from typing import TYPE_CHECKING
 
-from procrastinate import App, Blueprint, JobContext, PsycopgConnector, RetryStrategy
+from procrastinate import (
+    App,
+    BaseRetryStrategy,
+    Blueprint,
+    JobContext,
+    PsycopgConnector,
+    RetryDecision,
+)
 from procrastinate.connector import BaseConnector
 from procrastinate.exceptions import AlreadyEnqueued
+from procrastinate.jobs import Job
 
 from ...config import get_procrastinate_conninfo, get_settings
 from ..scanner import DocumentTask
@@ -47,14 +58,61 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-# Single queue for document ingest. KEDA scales the worker Deployment on the
+# One queue per extraction tier (Deck #323), aligned cheapest-first with
-# depth of this queue (``SELECT count(*) FROM procrastinate_jobs WHERE
+# document_processors.escalation.TIER_LADDER. Each queue is drained by its own
-# queue_name='ingest' AND status='todo'``).
+# worker Deployment + KEDA ScaledObject (``SELECT count(*) FROM
-INGEST_QUEUE_NAME = "ingest"
+# procrastinate_jobs WHERE queue_name=<queue> AND status='todo'``), so a
+# CPU-bound ``fast`` fleet, an in-cluster ``structured`` fleet, and a paid
+# network-bound ``ocr`` fleet scale (and fail) independently.
+INGEST_QUEUE_FAST = "ingest-fast"
+INGEST_QUEUE_STRUCTURED = "ingest-structured"
+INGEST_QUEUE_OCR = "ingest-ocr"
+
+# tier -> queue. The producer always defers onto the cheapest tier's queue; a
+# low-quality parse hops the job up the ladder via the retry strategy below.
+TIER_QUEUES: dict[str, str] = {
+    "fast": INGEST_QUEUE_FAST,
+    "structured": INGEST_QUEUE_STRUCTURED,
+    "ocr": INGEST_QUEUE_OCR,
+}
+_QUEUE_TIERS: dict[str, str] = {queue: tier for tier, queue in TIER_QUEUES.items()}
+ALL_INGEST_QUEUES: tuple[str, ...] = tuple(TIER_QUEUES.values())
+# New jobs start here; ``ocr`` is reached only by escalation.
+DEFAULT_INGEST_QUEUE = INGEST_QUEUE_FAST
+
+# Legacy single-queue name (pre-#323). A rolling upgrade may still have jobs
+# parked on it; a worker can be told to drain it alongside the tier queues, and
+# the job-count / reclaim helpers include it so nothing is stranded.
+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)
+
 # Blueprint namespace → registered task names are prefixed ``ingest:``.
 _NAMESPACE = "ingest"
 INGEST_TASK_NAME = f"{_NAMESPACE}:process_document"
 
+
+def tier_for_queue(queue: str | None) -> str:
+    """Tier a worker on ``queue`` should run. Unknown/legacy -> ``fast``.
+
+    The queue-aware task uses this to pick which single tier to parse with: the
+    job's current queue *is* its tier. A job on the legacy ``ingest`` queue (or
+    any unrecognised queue) defaults to the cheapest tier.
+    """
+    return _QUEUE_TIERS.get(queue or "", "fast")
+
+
 # A crashed worker leaves its job in ``doing``; reclaim it once its (per-worker)
 # heartbeat is this many seconds stale. The default is sized well above the
 # longest expected ``process_document`` (PDF render + embedding) so a slow-but-
@@ -69,6 +127,7 @@ INGEST_TASK_NAME = f"{_NAMESPACE}:process_document"
 # Blueprint cannot be added to more than one App — which the tests (in-memory +
 # real Postgres) and any re-init path require.
 async def process_document_task(
+    context: JobContext,
     *,
     user_id: str,
     doc_id: str,
@@ -80,12 +139,26 @@ async def process_document_task(
     etag: str | None = None,
     owner_id: str | None = None,
 ) -> None:
-    """Worker entry: rebuild the DocumentTask, resolve creds, run the pipeline."""
+    """Worker entry: rebuild the DocumentTask, resolve creds, run the pipeline.
+
+    Queue-aware (Deck #323): the tier this worker runs is the tier of the job's
+    current queue. A low-quality parse raises ``EscalateError``, which the
+    :class:`TieredEscalationStrategy` turns into a queue-hop to the next tier.
+    When per-tier escalation is disabled (``INGEST_ESCALATION_ENABLED=false``),
+    ``tier`` stays ``None`` and the inline pipeline runs (fast -> OCR in one
+    call), reproducing the pre-#323 single-queue behaviour.
+    """
     # Local imports avoid a heavy import chain at blueprint-definition time
     # (this module is also imported by the API pod just to defer jobs).
     from ..oauth_sync import NotProvisionedError  # noqa: PLC0415
     from ..processor import process_document  # noqa: PLC0415
 
+    tier = (
+        tier_for_queue(context.job.queue)
+        if get_settings().ingest_escalation_enabled
+        else None
+    )
+
     task = DocumentTask(
         user_id=user_id,
         doc_id=doc_id,
@@ -111,7 +184,7 @@ async def process_document_task(
 
     try:
         # Durable retry is procrastinate's job; disable the in-process loop.
-        await process_document(task, nc_client, max_retries=1)
+        await process_document(task, nc_client, max_retries=1, tier=tier)
     finally:
         await nc_client.close()
 
@@ -124,12 +197,22 @@ async def reclaim_stalled_ingest_jobs(context: JobContext, timestamp: int) -> No
     periodic-run marker (unused).
     """
     manager = context.app.job_manager
-    retry_at = datetime.now(tz=timezone.utc)
+    settings = get_settings()
-    stalled_after = get_settings().ingest_stalled_job_seconds
+    # Stagger the re-run rather than retry at now(): a stall is often systemic (a
+    # Qdrant / embedding outage stalls every in-flight job), so reclaiming the
+    # whole batch immediately every tick would thundering-herd a recovering
+    # dependency, bypassing TieredEscalationStrategy's per-job backoff. The fixed
+    # delay spreads them out; 0 restores the legacy immediate retry.
+    retry_at = datetime.now(tz=timezone.utc) + timedelta(
+        seconds=settings.ingest_reclaim_retry_delay_seconds
+    )
+    stalled_after = settings.ingest_stalled_job_seconds
     reclaimed = 0
-    for job in await manager.get_stalled_jobs(
+    # queue=None sweeps every queue, so an orphaned job on any tier queue is
-        queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=stalled_after
+    # reclaimed regardless of which tier's worker happens to run this periodic.
-    ):
+    # retry_job_by_id_async keeps the job on its own queue, so a stalled ``ocr``
+    # job re-runs on ``ingest-ocr`` (the ocr fleet), not the reclaiming worker's.
+    for job in await manager.get_stalled_jobs(seconds_since_heartbeat=stalled_after):
         if job.id is None:
             continue
         await manager.retry_job_by_id_async(job_id=job.id, retry_at=retry_at)
@@ -163,6 +246,128 @@ async def _resolve_client(user_id: str) -> NextcloudClient:
     return await get_user_client_basic_auth(user_id, host)
 
 
+def _first_leaf(exc: BaseException) -> BaseException:
+    """Descend nested ExceptionGroups to the first concrete leaf exception.
+
+    An anyio task group can wrap the real cause (and nest groups); the retry
+    strategy classifies on the leaf, mirroring ``processor._drop_reason``.
+    """
+    while isinstance(exc, BaseExceptionGroup) and exc.exceptions:
+        exc = exc.exceptions[0]
+    return exc
+
+
+def _is_transient_infra_error(exc: BaseException) -> bool:
+    """Whether ``exc`` is a transient infra blip worth a SAME-tier retry.
+
+    Mirrors the retryable subset of ``processor._drop_reason``: doc-fetch /
+    embed / Qdrant timeouts, connection drops, rate limits, and 5xx. A parse
+    that is merely *poor* never reaches here -- that path raises
+    ``EscalateError`` (handled separately) -- so this is purely about
+    infrastructure that should recover on its own. Imports are lazy: this only
+    runs in the worker, and the module is also imported by the API pod to defer.
+    """
+    import httpx  # noqa: PLC0415
+
+    if isinstance(exc, (httpx.TimeoutException, httpx.ConnectError)):
+        return True
+    try:
+        import openai  # noqa: PLC0415
+
+        if isinstance(
+            exc,
+            (
+                openai.APITimeoutError,
+                openai.APIConnectionError,
+                openai.RateLimitError,
+            ),
+        ):
+            return True
+        if isinstance(exc, openai.APIStatusError):
+            return exc.status_code >= 500
+    except ImportError:  # pragma: no cover -- openai is a hard dependency
+        pass
+    # Deliberately over-broad: this treats ALL qdrant_client exceptions as
+    # transient (not just timeouts/5xx). In a healthy cluster qdrant errors are
+    # transient, and a bounded same-tier retry is cheap; a genuinely permanent
+    # qdrant fault (e.g. schema mismatch) just exhausts the transient cap and
+    # then gives up. So unlike _drop_reason (which only *labels* the cause), this
+    # may add a few retries on a non-retriable qdrant error -- an acceptable
+    # trade for not having to enumerate qdrant's non-retriable status codes.
+    if type(exc).__module__.startswith("qdrant_client"):
+        return True
+    return False
+
+
+class TieredEscalationStrategy(BaseRetryStrategy):
+    """Native procrastinate retry that escalates across tier queues (Deck #323).
+
+    Three outcomes, decided from the raised exception:
+
+    - ``EscalateError`` -> ``RetryDecision(queue=<next tier's queue>)``: the SAME
+      job hops to the next fleet's queue and is parsed once by that tier. This is
+      how a document is "requeued on a failed parse" -- once per tier, with no
+      same-tier parse retry.
+    - a whitelisted transient infra error (doc fetch / embed / Qdrant blip) ->
+      same-queue exponential backoff, while under ``max_transient_attempts``.
+    - anything else, the transient cap is reached, or the target tier is unknown
+      -> ``None`` (no retry); the placeholder was already marked failed by the
+      pipeline, and the next scan re-picks the document.
+
+    Per-tier attempt accounting is intentionally approximate: a queue-hop can't
+    reset ``job.attempts`` (procrastinate has no per-tier counter), so parse
+    escalations *do* advance the same counter the transient cap reads. Because a
+    parse escalation hops (it never retries in place) the "once per parse per
+    tier" guarantee is structural; the cap is just a generous global ceiling on
+    transient churn across the whole lineage, not an exact per-tier count.
+    """
+
+    def __init__(self, *, max_transient_attempts: int) -> None:
+        self._max_transient_attempts = max_transient_attempts
+
+    def get_retry_decision(
+        self, *, exception: BaseException, job: Job
+    ) -> RetryDecision | None:
+        # Lazy import: EscalateError lives in the document stack, which the API
+        # pod (it also builds this App to defer) must not load. get_retry_decision
+        # runs only in the worker, where the stack is already imported.
+        from ...document_processors.escalation import EscalateError  # noqa: PLC0415
+
+        exc = _first_leaf(exception)
+
+        if isinstance(exc, EscalateError):
+            queue = TIER_QUEUES.get(exc.to_tier)
+            if queue is None:
+                # Unknown target tier: don't strand the job on a queue no worker
+                # drains -- stop and let the placeholder/next scan handle it.
+                logger.error(
+                    "ingest.escalate_unknown_tier from=%s to=%s",
+                    exc.from_tier,
+                    exc.to_tier,
+                )
+                return None
+            logger.info(
+                "ingest.escalate from=%s to=%s reason=%s queue=%s",
+                exc.from_tier,
+                exc.to_tier,
+                exc.reason,
+                queue,
+            )
+            # Immediate hop -- the next tier's fleet should pick it up at once.
+            return RetryDecision(queue=queue, retry_in={"seconds": 0})
+
+        if (
+            _is_transient_infra_error(exc)
+            and job.attempts < self._max_transient_attempts
+        ):
+            # 4, 8, 16, ... seconds, capped at 5 min. attempts is >=1 here (the
+            # failing attempt is counted), so attempts-1 makes the first wait 4s.
+            wait = min(4 * (2 ** max(0, job.attempts - 1)), 300)
+            return RetryDecision(retry_in={"seconds": wait})
+
+        return None
+
+
 def _build_ingest_blueprint() -> Blueprint:
     """Create a fresh Blueprint with the ingest tasks registered.
 
@@ -172,13 +377,29 @@ def _build_ingest_blueprint() -> Blueprint:
     bp = Blueprint()
     # Durable retry owned by the queue (survives worker crashes); the in-process
     # retry loop in process_document is disabled on this path via max_retries=1.
-    bp.task(
+    # The task's default queue is the cheapest tier; the producer defers there
+    # explicitly and the strategy hops a job up the ladder on a poor parse.
+    bp.task(  # type: ignore[no-matching-overload]
         name="process_document",
-        queue=INGEST_QUEUE_NAME,
+        queue=DEFAULT_INGEST_QUEUE,
-        retry=RetryStrategy(max_attempts=5, exponential_wait=4),
+        pass_context=True,
+        # procrastinate's RetryValue type only admits RetryStrategy, but a custom
+        # BaseRetryStrategy subclass is the documented extension point (and is
+        # accepted at runtime by get_retry_strategy). The annotation is just too
+        # narrow, hence the ignore.
+        # Settings are snapshotted here at blueprint-build time (build_app, first
+        # use), so a restart is needed to pick up INGEST_TRANSIENT_MAX_ATTEMPTS
+        # changes -- intentional: the strategy lives for the App's lifetime.
+        retry=TieredEscalationStrategy(
+            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=INGEST_QUEUE_NAME, 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
@@ -276,20 +497,43 @@ async def apply_ingest_queue_schema(
 _JOB_STATUSES = ("todo", "doing", "succeeded", "failed", "cancelled", "aborted")
 
 
-async def get_ingest_job_counts(app: App | None = None) -> dict[str, int]:
+async def get_ingest_job_counts_by_queue(
-    """Return ingest job counts by status (``todo``/``doing``/``failed``/…).
+    app: App | None = None,
+) -> dict[str, dict[str, int]]:
+    """Per-queue ingest job counts by status (Deck #323).
 
-    Reads procrastinate's per-queue stats via the manager API (not hand-written
+    Returns ``{queue_name: {status: count}}`` for the managed ingest queues (the
-    SQL) so a future schema bump doesn't silently break the status surface. The
+    per-tier queues + the legacy single queue) that have rows. Reads
-    manager flattens its per-status ``stats`` into top-level row keys, so we read
+    procrastinate's per-queue stats via the manager API (not hand-written SQL) so
-    the known status keys directly. Assumes the app's connector is already open.
+    a future schema bump doesn't silently break the status surface. Assumes the
+    app's connector is already open. Feeds the per-tier status surface + the
+    ``astrolabe_ingest_queue_depth`` gauge.
     """
     app = app or get_procrastinate_app()
-    counts: dict[str, int] = {}
+    by_queue: dict[str, dict[str, int]] = {}
-    for row in await app.job_manager.list_queues_async(queue=INGEST_QUEUE_NAME):
+    for row in await app.job_manager.list_queues_async():
+        name = row.get("name")
+        if name not in _MANAGED_QUEUES:
+            continue
+        per = by_queue.setdefault(name, {})
         for status in _JOB_STATUSES:
             if status in row:
-                counts[status] = counts.get(status, 0) + int(row[status])
+                per[status] = per.get(status, 0) + int(row[status])
+    return by_queue
+
+
+async def get_ingest_job_counts(app: App | None = None) -> dict[str, int]:
+    """Aggregate ingest job counts by status across all managed queues.
+
+    Fleet-wide totals summed over the per-tier queues + the legacy queue, so
+    ``pending = todo + doing`` reflects all outstanding ingest work regardless of
+    which tier a document currently sits on. Per-queue breakdown:
+    :func:`get_ingest_job_counts_by_queue`.
+    """
+    counts: dict[str, int] = {}
+    for per in (await get_ingest_job_counts_by_queue(app)).values():
+        for status, value in per.items():
+            counts[status] = counts.get(status, 0) + value
     return counts
 
 
@@ -336,7 +580,13 @@ class ProcrastinateTaskProducer:
 
     async def send(self, task: DocumentTask, /) -> None:
         key = _doc_queueing_lock(task)
-        deferrer = self._app.configure_task(INGEST_TASK_NAME, queueing_lock=key)
+        # Always defer onto the cheapest tier's queue; the escalation strategy
+        # hops the job up the ladder on a poor parse. queueing_lock is a global
+        # partial-unique on status='todo', so a doc mid-escalation on a higher
+        # tier still dedupes a fresh enqueue here -- no double-processing.
+        deferrer = self._app.configure_task(
+            INGEST_TASK_NAME, queue=DEFAULT_INGEST_QUEUE, queueing_lock=key
+        )
         try:
             await deferrer.defer_async(**asdict(task))
         except AlreadyEnqueued:
@@ -358,6 +608,10 @@ class ProcrastinateTaskProducer:
         """Ingest job counts by status (for the vector-sync status surface)."""
         return await get_ingest_job_counts(self._app)
 
+    async def job_counts_by_queue(self) -> dict[str, dict[str, int]]:
+        """Per-tier-queue ingest job counts by status (Deck #323)."""
+        return await get_ingest_job_counts_by_queue(self._app)
+
     def clone(self) -> ProcrastinateTaskProducer:
         return self
 

tests/unit/test_ingest_queue_depth_metric.py

@@ -0,0 +1,55 @@
+"""Unit test for the per-tier ingest-queue-depth gauge (Deck #323).
+
+Guards the round-2 fix: a queue that drains to empty (and so drops out of
+procrastinate's ``list_queues_async``) must read 0, not its last non-zero value.
+"""
+
+import pytest
+from pytest import approx
+
+from nextcloud_mcp_server.observability.metrics import update_ingest_queue_depth
+
+pytestmark = pytest.mark.unit
+
+_METRIC = "astrolabe_ingest_queue_depth"
+
+
+def test_drained_queue_zeroes_not_stale(metric_sample):
+    # ocr has a backlog this tick. (pytest.approx: the gauge sample is a float.)
+    update_ingest_queue_depth({"ingest-ocr": {"todo": 4}})
+    assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == approx(
+        4
+    )
+
+    # Next tick ocr has drained → procrastinate omits it from by_queue entirely.
+    update_ingest_queue_depth({"ingest-fast": {"todo": 1}})
+    # The gauge must read 0 for the drained queue, not the stale 4.
+    assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == approx(
+        0
+    )
+    assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == approx(
+        1
+    )
+
+
+def test_none_is_noop(metric_sample):
+    update_ingest_queue_depth({"ingest-fast": {"doing": 2}})
+    # Memory backend passes None → must not wipe the last published values.
+    update_ingest_queue_depth(None)
+    assert metric_sample(
+        _METRIC, {"queue": "ingest-fast", "status": "doing"}
+    ) == approx(2)
+
+
+def test_all_queues_drained_empty_dict_zeroes(metric_sample):
+    # postgres backend with every queue drained → get_ingest_job_counts_by_queue
+    # returns {} (list_queues_async drops empty queues). An empty dict is NOT the
+    # memory-backend no-op: it must still zero every managed queue's gauge.
+    update_ingest_queue_depth({"ingest-fast": {"todo": 9}})
+    assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == approx(
+        9
+    )
+    update_ingest_queue_depth({})
+    assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == approx(
+        0
+    )

tests/unit/test_processor_metering.py

@@ -176,3 +176,53 @@ async def test_store_failure_is_swallowed(monkeypatch):
         total_chars=9,
         page_count=2,
     )
+
+
+@pytest.mark.unit
+async def test_ocr_tier_records_pages_ocr(store_spy):
+    """OCR-tier pages are metered as a separate pages_ocr line (Deck #323)."""
+    await processor.record_indexing_usage(
+        enabled=True,
+        provider="mistral",
+        model="mistral-embed",
+        doc_type="file",
+        user_id="alice",
+        chunk_count=20,
+        token_count=900,
+        total_chars=40000,
+        page_count=8,
+        pipeline_tier="ocr",
+    )
+    by_metric = {
+        c.kwargs["metric"]: c.kwargs["value"]
+        for c in store_spy.record_usage_event.await_args_list
+    }
+    # pages_ocr fires IN ADDITION to pages_embedded for OCR-tier pages.
+    assert by_metric == {
+        "tokens_embedded": 900,
+        "pages_embedded": 8,
+        "pages_ocr": 8,
+    }
+    # pipeline_tier is threaded into the billing metadata for CP attribution.
+    for c in store_spy.record_usage_event.await_args_list:
+        assert c.kwargs["metadata"]["pipeline_tier"] == "ocr"
+
+
+@pytest.mark.unit
+async def test_fast_tier_does_not_record_pages_ocr(store_spy):
+    """A CPU-cheap fast-tier parse must NOT incur the paid pages_ocr line."""
+    await processor.record_indexing_usage(
+        enabled=True,
+        provider="mistral",
+        model="mistral-embed",
+        doc_type="file",
+        user_id="alice",
+        chunk_count=10,
+        token_count=500,
+        total_chars=20000,
+        page_count=4,
+        pipeline_tier="fast",
+    )
+    metrics = {c.kwargs["metric"] for c in store_spy.record_usage_event.await_args_list}
+    assert "pages_ocr" not in metrics
+    assert metrics == {"tokens_embedded", "pages_embedded"}

tests/unit/test_registry_tiering.py

@@ -241,3 +241,182 @@ async def test_no_ocr_escalation_when_disabled(monkeypatch):
     res = await r.process(b"%PDF-1.7", "application/pdf")
     # Fast tier is terminal when OCR is disabled.
     assert res.processor == "fast"
+
+
+# --- Per-tier external path (Deck #323) -------------------------------------
+
+
+async def test_process_tier_runs_named_tier(monkeypatch):
+    """process_tier runs exactly the requested tier's processor, not priority."""
+    monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings())
+    r = _registry(
+        (_Fake("fast", "fast"), 20),
+        (_Fake("structured", "structured"), 10),
+        (_Fake("ocr", "ocr"), 5),
+    )
+    res = await r.process_tier(b"%PDF-1.7", "application/pdf", "f.pdf", "structured")
+    assert res.processor == "structured"
+
+
+async def test_process_tier_unknown_tier_raises(monkeypatch):
+    from nextcloud_mcp_server.document_processors.base import ProcessorError
+
+    monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings())
+    r = _registry((_Fake("fast", "fast"), 20))
+    with pytest.raises(ProcessorError, match="structured"):
+        await r.process_tier(b"%PDF-1.7", "application/pdf", "f.pdf", "structured")
+
+
+async def test_process_tier_oversize_fails_fast(monkeypatch):
+    """The size guard applies on the per-tier path too (before any parse)."""
+    monkeypatch.setattr(
+        reg_mod, "get_settings", lambda: _Settings(max_pdf_size_mb=0.001)
+    )
+    r = _registry((_Fake("ocr", "ocr"), 5))
+    res = await r.process_tier(b"x" * 4096, "application/pdf", "big.pdf", "ocr")
+    assert res.success is False
+    assert res.metadata["parse_failed_reason"] == "oversize"
+
+
+def test_next_available_tier_walks_ladder():
+    r = _registry(
+        (_Fake("fast", "fast"), 20),
+        (_Fake("structured", "structured"), 10),
+        (_Fake("ocr", "ocr"), 5),
+    )
+    # ocr disabled -> structured is the only target above fast.
+    s = _Settings(ocr=False)
+    assert r.next_available_tier("fast", s) == "structured"
+    assert r.next_available_tier("structured", s) is None  # ocr gated off
+    # ocr enabled -> reachable; minimum skips the structured rung.
+    s_ocr = _Settings(ocr=True)
+    assert r.next_available_tier("structured", s_ocr) == "ocr"
+    assert r.next_available_tier("fast", s_ocr, minimum="ocr") == "ocr"
+
+
+def test_next_available_tier_skips_unregistered():
+    # No structured processor -> fast escalates straight to ocr.
+    r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5))
+    assert r.next_available_tier("fast", _Settings(ocr=True)) == "ocr"
+
+
+def test_evaluate_escalation_good_text_indexes(monkeypatch):
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    r = _registry(
+        (_Fake("fast", "fast", text="This is clean readable prose text."), 20),
+        (_Fake("ocr", "ocr"), 5),
+    )
+    res = ProcessingResult(
+        text="This is clean readable prose text.",
+        metadata={
+            "page_count": 1,
+            "page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 34}],
+        },
+        processor="fast",
+    )
+    assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None
+
+
+def test_evaluate_escalation_empty_jumps_to_ocr(monkeypatch):
+    """A scanned (no-text-layer) result targets ocr directly, skipping structured."""
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    r = _registry(
+        (_Fake("fast", "fast"), 20),
+        (_Fake("structured", "structured"), 10),
+        (_Fake("ocr", "ocr"), 5),
+    )
+    res = ProcessingResult(
+        text="",
+        metadata={
+            "page_count": 1,
+            "page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 0}],
+        },
+        processor="fast",
+    )
+    decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True))
+    assert decision == ("ocr", "empty_text")
+
+
+def test_evaluate_escalation_lowconf_goes_to_structured(monkeypatch):
+    """A junk-but-non-empty layer escalates to the next rung (structured)."""
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    junk = "x" * 40  # one long token, no whitespace -> quality ~0
+    r = _registry(
+        (_Fake("fast", "fast"), 20),
+        (_Fake("structured", "structured"), 10),
+        (_Fake("ocr", "ocr"), 5),
+    )
+    res = ProcessingResult(
+        text=junk,
+        metadata={
+            "page_count": 1,
+            "page_boundaries": [
+                {"page": 1, "start_offset": 0, "end_offset": len(junk)}
+            ],
+        },
+        processor="fast",
+    )
+    decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True))
+    assert decision == ("structured", "low_confidence")
+
+
+def test_evaluate_escalation_failure_not_escalated(monkeypatch):
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5))
+    res = ProcessingResult(
+        text="",
+        metadata={"parse_failed_reason": "error"},
+        processor="fast",
+        success=False,
+    )
+    assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) is None
+
+
+def test_evaluate_escalation_terminal_when_no_higher_tier(monkeypatch):
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    # Only fast registered -> nowhere to escalate even on junk text.
+    r = _registry((_Fake("fast", "fast"), 20))
+    junk = "y" * 40
+    res = ProcessingResult(
+        text=junk,
+        metadata={
+            "page_count": 1,
+            "page_boundaries": [
+                {"page": 1, "start_offset": 0, "end_offset": len(junk)}
+            ],
+        },
+        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
+
+
+def test_evaluate_escalation_lowconf_to_ocr_when_no_structured(monkeypatch):
+    """fast+ocr only: a low-confidence parse routes straight to ocr (skips the
+    unregistered structured rung), not to None."""
+    monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
+    junk = "z" * 40  # non-empty but junk -> recommended ocr, total_chars > 0
+    r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5))
+    res = ProcessingResult(
+        text=junk,
+        metadata={
+            "page_count": 1,
+            "page_boundaries": [
+                {"page": 1, "start_offset": 0, "end_offset": len(junk)}
+            ],
+        },
+        processor="fast",
+    )
+    decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True))
+    assert decision == ("ocr", "low_confidence")

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_parse_pdf_tier.py

@@ -0,0 +1,70 @@
+"""Unit tests for the per-tier PDF parse + escalation gate (Deck #323).
+
+``processor._parse_pdf_tier`` runs one tier and either returns the result to
+index or raises ``EscalateError`` (a queue-hop). These exercise the decision
+without standing up the full ingest pipeline.
+"""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from nextcloud_mcp_server.document_processors.base import ProcessingResult
+from nextcloud_mcp_server.document_processors.escalation import EscalateError
+from nextcloud_mcp_server.vector import processor
+
+pytestmark = pytest.mark.unit
+
+
+def _registry(result: ProcessingResult, decision):
+    reg = MagicMock()
+    reg.process_tier = AsyncMock(return_value=result)
+    reg.evaluate_escalation = MagicMock(return_value=decision)
+    return reg
+
+
+async def test_good_parse_returns_result(monkeypatch):
+    rec = MagicMock()
+    monkeypatch.setattr(processor, "record_document_escalation", rec)
+    result = ProcessingResult(text="clean", metadata={}, processor="fast")
+    reg = _registry(result, decision=None)
+    out = await processor._parse_pdf_tier(
+        reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object()
+    )
+    assert out is result
+    rec.assert_not_called()
+
+
+async def test_low_quality_parse_raises_escalate(monkeypatch):
+    rec = MagicMock()
+    monkeypatch.setattr(processor, "record_document_escalation", rec)
+    result = ProcessingResult(text="", metadata={}, processor="fast")
+    reg = _registry(result, decision=("ocr", "empty_text"))
+    with pytest.raises(EscalateError) as ei:
+        await processor._parse_pdf_tier(
+            reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object()
+        )
+    assert ei.value.from_tier == "fast"
+    assert ei.value.to_tier == "ocr"
+    assert ei.value.reason == "empty_text"
+    # The escalation is recorded at the decision point.
+    rec.assert_called_once_with("fast", "ocr", "empty_text")
+
+
+async def test_hard_failure_returns_result_without_escalating(monkeypatch):
+    rec = MagicMock()
+    monkeypatch.setattr(processor, "record_document_escalation", rec)
+    result = ProcessingResult(
+        text="",
+        metadata={"parse_failed_reason": "oversize"},
+        processor="size_guard",
+        success=False,
+    )
+    reg = _registry(result, decision=("ocr", "empty_text"))
+    out = await processor._parse_pdf_tier(
+        reg, b"%PDF", "application/pdf", "big.pdf", "fast", settings=object()
+    )
+    # success=False short-circuits: the gate is never consulted, no escalation.
+    assert out is result
+    reg.evaluate_escalation.assert_not_called()
+    rec.assert_not_called()

tests/unit/vector/test_procrastinate_producer.py

@@ -3,6 +3,7 @@
 Uses procrastinate's in-memory connector so no live Postgres is required.
 """
 
+from types import SimpleNamespace
 from typing import cast
 from unittest.mock import AsyncMock
 
@@ -15,6 +16,11 @@ from nextcloud_mcp_server.vector.scanner import DocumentTask
 pytestmark = pytest.mark.unit
 
 
+def _ctx(queue: str = pq.INGEST_QUEUE_FAST) -> JobContext:
+    """Minimal JobContext stand-in: the task only reads ``context.job.queue``."""
+    return cast(JobContext, SimpleNamespace(job=SimpleNamespace(queue=queue)))
+
+
 @pytest.fixture
 def app():
     """An App bound to the in-memory connector with the ingest tasks."""
@@ -42,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"
@@ -94,18 +101,21 @@ class TestProcessDocumentTask:
             captured["user_id"] = user_id
             return fake_client
 
-        async def fake_process(task, nc_client, *, max_retries):
+        async def fake_process(task, nc_client, *, max_retries, tier):
             captured["task"] = task
             captured["nc_client"] = nc_client
             captured["max_retries"] = max_retries
+            captured["tier"] = tier
 
         monkeypatch.setattr(pq, "_resolve_client", fake_resolve)
         monkeypatch.setattr(
             "nextcloud_mcp_server.vector.processor.process_document", fake_process
         )
 
-        # Calling the Task runs its wrapped function in-process.
+        # Calling the Task runs its wrapped function in-process. The job is on the
+        # ocr queue, so the queue-aware task must derive tier="ocr".
         await pq.process_document_task(
+            _ctx(pq.INGEST_QUEUE_OCR),
             user_id="alice",
             doc_id="42",
             doc_type="note",
@@ -120,6 +130,8 @@ class TestProcessDocumentTask:
         assert captured["task"].etag == "e1"
         # Worker disables the in-process retry loop; durable retry is the queue's.
         assert captured["max_retries"] == 1
+        # Tier is derived from the job's queue (escalation enabled by default).
+        assert captured["tier"] == "ocr"
         fake_client.close.assert_awaited_once()
 
     async def test_pipeline_error_propagates_and_closes_client(self, monkeypatch):
@@ -130,7 +142,7 @@ class TestProcessDocumentTask:
         async def fake_resolve(user_id):
             return fake_client
 
-        async def fake_process(task, nc_client, *, max_retries):
+        async def fake_process(task, nc_client, *, max_retries, tier):
             raise RuntimeError("transient qdrant failure")
 
         monkeypatch.setattr(pq, "_resolve_client", fake_resolve)
@@ -140,6 +152,7 @@ class TestProcessDocumentTask:
 
         with pytest.raises(RuntimeError, match="transient qdrant failure"):
             await pq.process_document_task(
+                _ctx(),
                 user_id="alice",
                 doc_id="42",
                 doc_type="note",
@@ -167,6 +180,7 @@ class TestProcessDocumentTask:
 
         # Returns cleanly (job succeeds as a no-op); pipeline never runs.
         await pq.process_document_task(
+            _ctx(),
             user_id="ghost",
             doc_id="9",
             doc_type="note",
@@ -178,9 +192,10 @@ class TestProcessDocumentTask:
 
 class TestReclaimStalledJobs:
     async def test_reclaims_each_stalled_job(self):
-        from datetime import datetime
+        from datetime import datetime, timezone
 
         retried: list[int] = []
+        retry_ats: list[datetime] = []
 
         class Job:
             def __init__(self, id):
@@ -188,12 +203,14 @@ class TestReclaimStalledJobs:
 
         class FakeManager:
             async def get_stalled_jobs(self, queue=None, seconds_since_heartbeat=0):
-                assert queue == pq.INGEST_QUEUE_NAME
+                # Reclaim sweeps EVERY queue (Deck #323), so no queue filter.
+                assert queue is None
                 return [Job(1), Job(2), Job(None)]  # None id is skipped
 
             async def retry_job_by_id_async(self, job_id, retry_at):
                 assert isinstance(retry_at, datetime)
                 retried.append(job_id)
+                retry_ats.append(retry_at)
 
         class FakeApp:
             job_manager = FakeManager()
@@ -201,34 +218,66 @@ class TestReclaimStalledJobs:
         class Ctx:
             app = FakeApp()
 
+        before = datetime.now(tz=timezone.utc)
         await pq.reclaim_stalled_ingest_jobs(cast(JobContext, Ctx()), timestamp=0)
         assert retried == [1, 2]
+        # Reclaimed jobs are staggered into the future (default 30s) rather than
+        # retried at now(), so a systemic outage doesn't thundering-herd.
+        assert all((ra - before).total_seconds() >= 25 for ra in retry_ats)
 
 
 class TestGetIngestJobCounts:
     async def test_aggregates_stats_rows(self):
         class FakeManager:
-            async def list_queues_async(self, queue=None):
+            async def list_queues_async(self, queue=None, **kwargs):
-                assert queue == pq.INGEST_QUEUE_NAME
+                # Counts now aggregate across all managed queues (Deck #323), so
+                # the helper lists every queue and filters by name itself.
+                assert queue is None
                 # procrastinate flattens per-status stats into top-level keys.
                 return [
                     {
-                        "name": "ingest",
+                        "name": "ingest-fast",
-                        "jobs_count": 6,
+                        "jobs_count": 4,
                         "todo": 3,
                         "doing": 1,
                         "succeeded": 0,
+                        "failed": 0,
+                        "cancelled": 0,
+                        "aborted": 0,
+                    },
+                    {
+                        "name": "ingest-ocr",
+                        "jobs_count": 2,
+                        "todo": 0,
+                        "doing": 0,
+                        "succeeded": 0,
                         "failed": 2,
                         "cancelled": 0,
                         "aborted": 0,
-                    }
+                    },
+                    {
+                        # An unmanaged queue must NOT pollute ingest counts.
+                        "name": "some-other-queue",
+                        "jobs_count": 9,
+                        "todo": 9,
+                        "doing": 0,
+                        "succeeded": 0,
+                        "failed": 0,
+                        "cancelled": 0,
+                        "aborted": 0,
+                    },
                 ]
 
         class FakeApp:
             job_manager = FakeManager()
 
         counts = await pq.get_ingest_job_counts(cast(App, FakeApp()))
-        assert counts["todo"] == 3
+        assert counts["todo"] == 3  # only ingest-* queues, not some-other-queue
         assert counts["doing"] == 1
         assert counts["failed"] == 2
         assert counts["succeeded"] == 0
+
+        by_queue = await pq.get_ingest_job_counts_by_queue(cast(App, FakeApp()))
+        assert set(by_queue) == {"ingest-fast", "ingest-ocr"}
+        assert by_queue["ingest-fast"]["todo"] == 3
+        assert by_queue["ingest-ocr"]["failed"] == 2

tests/unit/vector/test_tiered_escalation_strategy.py

@@ -0,0 +1,122 @@
+"""Unit tests for the per-tier escalation primitives (Deck #323).
+
+Covers the tier-ladder helpers + EscalateError (document_processors.escalation)
+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
+
+import nextcloud_mcp_server.vector.queue.procrastinate as pq
+from nextcloud_mcp_server.document_processors.escalation import (
+    TIER_LADDER,
+    EscalateError,
+    next_tier,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _job(queue: str = pq.INGEST_QUEUE_FAST, attempts: int = 1) -> Job:
+    return Job(
+        id=1,
+        queue=queue,
+        task_name=pq.INGEST_TASK_NAME,
+        lock=None,
+        queueing_lock=None,
+        attempts=attempts,
+    )
+
+
+class TestLadder:
+    def test_next_tier_ordering(self):
+        assert next_tier("fast") == "structured"
+        assert next_tier("structured") == "ocr"
+        assert next_tier("ocr") is None  # terminal
+        assert next_tier("unknown") is None
+
+    def test_ladder_is_cheapest_first(self):
+        assert TIER_LADDER == ("fast", "structured", "ocr")
+
+    def test_tier_for_queue(self):
+        assert pq.tier_for_queue(pq.INGEST_QUEUE_OCR) == "ocr"
+        assert pq.tier_for_queue(pq.INGEST_QUEUE_STRUCTURED) == "structured"
+        # Legacy / unknown / None all fall back to the cheapest tier.
+        assert pq.tier_for_queue(pq.LEGACY_INGEST_QUEUE) == "fast"
+        assert pq.tier_for_queue(None) == "fast"
+
+
+class TestTieredEscalationStrategy:
+    def _strategy(self, max_transient: int = 5):
+        return pq.TieredEscalationStrategy(max_transient_attempts=max_transient)
+
+    def test_escalate_hops_to_target_queue(self):
+        exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text")
+        decision = self._strategy().get_retry_decision(exception=exc, job=_job())
+        assert decision is not None
+        assert decision.queue == pq.INGEST_QUEUE_OCR
+
+    def test_escalate_to_structured(self):
+        exc = EscalateError(
+            from_tier="fast", to_tier="structured", reason="low_confidence"
+        )
+        decision = self._strategy().get_retry_decision(exception=exc, job=_job())
+        assert decision is not None
+        assert decision.queue == pq.INGEST_QUEUE_STRUCTURED
+
+    def test_escalate_unknown_tier_gives_up(self):
+        exc = EscalateError(from_tier="ocr", to_tier="bogus", reason="low_confidence")
+        decision = self._strategy().get_retry_decision(exception=exc, job=_job())
+        assert decision is None
+
+    def test_escalate_unwraps_exception_group(self):
+        exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text")
+        group = ExceptionGroup("wrapped", [exc])
+        decision = self._strategy().get_retry_decision(exception=group, job=_job())
+        assert decision is not None
+        assert decision.queue == pq.INGEST_QUEUE_OCR
+
+    def test_transient_retries_same_queue_under_cap(self):
+        decision = self._strategy(max_transient=5).get_retry_decision(
+            exception=httpx.ConnectError("refused"), job=_job(attempts=1)
+        )
+        assert decision is not None
+        # Same-tier retry: no queue override (stays on its current queue).
+        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.
+        # procrastinate sets retry_at = utcnow() + wait at call time. Bracketing
+        # the call with before/after makes the assertion exact and independent of
+        # runner load: with before <= call_now <= after, we have
+        #   (retry_at - after) <= wait <= (retry_at - before).
+        strat = self._strategy(max_transient=100)
+        for attempts, expected in [(1, 4), (2, 8), (3, 16), (4, 32), (20, 300)]:
+            before = datetime.now(timezone.utc)
+            decision = strat.get_retry_decision(
+                exception=httpx.ConnectError("x"), job=_job(attempts=attempts)
+            )
+            after = datetime.now(timezone.utc)
+            assert decision is not None and decision.retry_at is not None
+            lo = (decision.retry_at - after).total_seconds()
+            hi = (decision.retry_at - before).total_seconds()
+            assert lo <= expected <= hi, (
+                f"attempts={attempts}: expected={expected}s not in [{lo:.3f}, {hi:.3f}]"
+            )
+
+    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)
+        )
+        assert decision is None
+
+    def test_non_transient_error_gives_up(self):
+        decision = self._strategy().get_retry_decision(
+            exception=ValueError("permanent"), job=_job(attempts=1)
+        )
+        assert decision is None