diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py index a49bf6df..7e39d765 100644 --- a/nextcloud_mcp_server/document_processors/escalation.py +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -30,10 +30,11 @@ 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. + *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) diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index f506afbc..71dc4728 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -656,12 +656,27 @@ def update_ingest_queue_depth(by_queue: dict[str, dict[str, int]] | None) -> Non """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). + ``queue.procrastinate.get_ingest_job_counts_by_queue``). No-op on the memory + backend (``by_queue`` is None). + + 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. 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. """ if not by_queue: 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( diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index 9f926da6..bc511f44 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -279,6 +279,13 @@ def _is_transient_infra_error(exc: BaseException) -> bool: 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 diff --git a/tests/unit/test_ingest_queue_depth_metric.py b/tests/unit/test_ingest_queue_depth_metric.py new file mode 100644 index 00000000..9ff5a5d5 --- /dev/null +++ b/tests/unit/test_ingest_queue_depth_metric.py @@ -0,0 +1,32 @@ +"""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 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. + update_ingest_queue_depth({"ingest-ocr": {"todo": 4}}) + assert metric_sample(_METRIC, {"queue": "ingest-ocr", "status": "todo"}) == 4.0 + + # 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"}) == 0.0 + assert metric_sample(_METRIC, {"queue": "ingest-fast", "status": "todo"}) == 1.0 + + +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"}) == 2.0 diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index 1c0f6dc4..7c343b14 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -400,3 +400,23 @@ def test_evaluate_escalation_zero_page_does_not_escalate(monkeypatch): 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")