diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 0853ad2d..ee12ac2a 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/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: diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index 9f1d2283..a169ffaa 100644 --- a/nextcloud_mcp_server/cli.py +++ b/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): - """Run the ingest worker (Deck #183). +@click.option( + "--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-``), 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,21 @@ 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, + 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. + if tier is not None: + queues = [TIER_QUEUES[tier]] + else: + queues = [*ALL_INGEST_QUEUES, LEGACY_INGEST_QUEUE] + # 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 +424,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_QUEUE_NAME, + "Ingest worker started: tier=%s queues=%s concurrency=%s " + "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 diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 3afe22ce..397ae470 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -234,6 +234,19 @@ _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. + "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). + "ingest_transient_max_attempts": 5, "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 +330,7 @@ _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("VECTOR_SYNC_SCAN_INTERVAL", gte=1), Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1), Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1), @@ -856,6 +870,8 @@ 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 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 +1497,8 @@ 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", "collection_metadata_source": "COLLECTION_METADATA_SOURCE", "collection_metadata_api_url": "COLLECTION_METADATA_API_URL", "embedding_gateway_url": "EMBEDDING_GATEWAY_URL", diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py new file mode 100644 index 00000000..c3ec30bd --- /dev/null +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -0,0 +1,66 @@ +"""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=)`` 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). Callers that need + availability resolve it against the registry + settings (see + ``ProcessorRegistry.next_available_tier``); 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: + ``empty_text`` | ``low_confidence`` | ``unsupported`` | ``forced``. + """ + + 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})", + ) diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index b6c181a5..ff4b678e 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/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 - # DUDE) burns the OCR timeout for 0 chars. Fail fast with an explicit - # reason so the caller marks the placeholder "failed" instead of - # 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 "", - 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"), - ) + oversize = self._oversize_result(content, filename, settings) + if oversize is not None: + return oversize 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. - classification = None - if settings.document_classify_enabled and result.success: - 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 "", - 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 "", - exc_info=True, - ) + # OCR-opted-in tenants only. Shared with the external per-tier path via + # _classify_result. + classification = self._classify_result( + result, content, settings, record=True, filename=filename + ) # 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 "", + 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 "", + 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 "", + 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, diff --git a/nextcloud_mcp_server/models/semantic.py b/nextcloud_mcp_server/models/semantic.py index 35f67ddf..ff39493c 100644 --- a/nextcloud_mcp_server/models/semantic.py +++ b/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__ = [ diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index d6a37299..f506afbc 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/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,23 @@ 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``). A queue missing a + status is set to 0 so a drained queue reads zero rather than going stale at + its last non-zero value. No-op on the memory backend (``by_queue`` is None). + """ + if not by_queue: + return + 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, diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 382236a4..0696fbcc 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/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: diff --git a/nextcloud_mcp_server/vector/ingest_status.py b/nextcloud_mcp_server/vector/ingest_status.py index 6f470525..f1cd7689 100644 --- a/nextcloud_mcp_server/vector/ingest_status.py +++ b/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) diff --git a/nextcloud_mcp_server/vector/metrics_publisher.py b/nextcloud_mcp_server/vector/metrics_publisher.py index 87998201..2958ff72 100644 --- a/nextcloud_mcp_server/vector/metrics_publisher.py +++ b/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) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index ddc5da73..6da08d79 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/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,59 @@ 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, + ) + + 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 "", + 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 +233,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 +274,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 +308,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 +471,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 +489,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 +502,19 @@ 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 there so the in-process / + # memory path never pulls the document stack at call time (mirrors the lazy + # get_registry import; see #877). 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 +584,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 +608,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 +666,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 +686,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 +924,40 @@ async def _index_document( "vector_sync.file_size": len(content_bytes), }, ): - # The registry runs the tiered PDF pipeline (tier-0 classify -> - # tier-1 fast -> OCR escalation) and records classification metrics. - # Imported lazily so module import doesn't pull in the document stack - # (document_processors -> _isolation, Unix-only ``resource``; see #877). + # The registry runs the tiered PDF pipeline and records + # classification metrics. Imported lazily so module import doesn't + # pull in the document stack (document_processors -> _isolation, + # 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( - content=content_bytes, - content_type=content_type, - filename=file_path, - ) + # External per-tier path (Deck #323): run only this worker's tier + # for PDFs and let a low-quality parse raise EscalateError (a + # queue-hop to the next tier). Everything else -- non-PDF files, + # 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 +1023,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 +1184,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(): diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index e7b13f92..d37b5af6 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/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 - per-tenant Postgres (the same app DB; procrastinate manages its own tables). -- **Consumer** (worker role) — ``nextcloud-mcp-server worker`` runs - :func:`procrastinate.App.run_worker`, which drains the ``ingest`` queue and - invokes the existing :func:`process_document` pipeline. + *defers* one ``ingest:process_document`` job per changed document onto the + cheapest tier's queue (``ingest-fast``) in the per-tenant Postgres (the same + app DB; procrastinate manages its own tables). +- **Consumer** (worker role) — ``nextcloud-mcp-server worker [--tier T]`` runs + :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: @@ -35,9 +38,17 @@ from datetime import datetime, 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,53 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Single queue for document ingest. KEDA scales the worker Deployment on the -# depth of this queue (``SELECT count(*) FROM procrastinate_jobs WHERE -# queue_name='ingest' AND status='todo'``). -INGEST_QUEUE_NAME = "ingest" +# One queue per extraction tier (Deck #323), aligned cheapest-first with +# document_processors.escalation.TIER_LADDER. Each queue is drained by its own +# worker Deployment + KEDA ScaledObject (``SELECT count(*) FROM +# procrastinate_jobs WHERE queue_name= 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 + +# 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 +119,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 +131,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 +176,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() @@ -127,9 +192,11 @@ async def reclaim_stalled_ingest_jobs(context: JobContext, timestamp: int) -> No retry_at = datetime.now(tz=timezone.utc) stalled_after = get_settings().ingest_stalled_job_seconds reclaimed = 0 - for job in await manager.get_stalled_jobs( - queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=stalled_after - ): + # queue=None sweeps every queue, so an orphaned job on any tier queue is + # 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 +230,121 @@ 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 + 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=)``: 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 +354,22 @@ 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, - retry=RetryStrategy(max_attempts=5, exponential_wait=4), + queue=DEFAULT_INGEST_QUEUE, + 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. + retry=TieredEscalationStrategy( + max_transient_attempts=get_settings().ingest_transient_max_attempts + ), )(process_document_task) reclaim = bp.task( - name="reclaim_stalled_jobs", queue=INGEST_QUEUE_NAME, pass_context=True + name="reclaim_stalled_jobs", queue=DEFAULT_INGEST_QUEUE, pass_context=True )(reclaim_stalled_ingest_jobs) bp.periodic(cron="*/5 * * * *", periodic_id="reclaim_stalled_ingest")(reclaim) return bp @@ -276,20 +467,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]: - """Return ingest job counts by status (``todo``/``doing``/``failed``/…). +async def get_ingest_job_counts_by_queue( + 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 - SQL) so a future schema bump doesn't silently break the status surface. The - manager flattens its per-status ``stats`` into top-level row keys, so we read - the known status keys directly. Assumes the app's connector is already open. + Returns ``{queue_name: {status: count}}`` for the managed ingest queues (the + per-tier queues + the legacy single queue) that have rows. Reads + procrastinate's per-queue stats via the manager API (not hand-written SQL) so + 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] = {} - for row in await app.job_manager.list_queues_async(queue=INGEST_QUEUE_NAME): + by_queue: dict[str, dict[str, int]] = {} + 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 +550,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 +578,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 diff --git a/tests/unit/test_processor_metering.py b/tests/unit/test_processor_metering.py index 1b9dedc7..01f612f1 100644 --- a/tests/unit/test_processor_metering.py +++ b/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"} diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index c111763d..0c0a4da6 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -241,3 +241,150 @@ 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 diff --git a/tests/unit/vector/test_parse_pdf_tier.py b/tests/unit/vector/test_parse_pdf_tier.py new file mode 100644 index 00000000..cc445ea3 --- /dev/null +++ b/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() diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index f472044a..156ce073 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/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.""" @@ -94,18 +100,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 +129,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 +141,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 +151,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 +179,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", @@ -188,7 +201,8 @@ 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): @@ -208,27 +222,55 @@ class TestReclaimStalledJobs: class TestGetIngestJobCounts: async def test_aggregates_stats_rows(self): class FakeManager: - async def list_queues_async(self, queue=None): - assert queue == pq.INGEST_QUEUE_NAME + async def list_queues_async(self, queue=None, **kwargs): + # 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", - "jobs_count": 6, + "name": "ingest-fast", + "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 diff --git a/tests/unit/vector/test_tiered_escalation_strategy.py b/tests/unit/vector/test_tiered_escalation_strategy.py new file mode 100644 index 00000000..215eafc6 --- /dev/null +++ b/tests/unit/vector/test_tiered_escalation_strategy.py @@ -0,0 +1,100 @@ +"""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. +""" + +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_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