diff --git a/nextcloud_mcp_server/providers/base.py b/nextcloud_mcp_server/providers/base.py index 9dad34c4..7149b13a 100644 --- a/nextcloud_mcp_server/providers/base.py +++ b/nextcloud_mcp_server/providers/base.py @@ -75,6 +75,12 @@ class Provider(ABC): usage from their embedding response override this. Used by the usage-metering hooks (Deck #67) to bill ``embeddings_queries`` by tokens rather than by operation count. + + IMPORTANT (recursion invariant): this default calls ``self.embed``. A + provider that overrides ``embed()`` to delegate to ``embed_with_usage()`` + (to avoid duplicating request logic) MUST also override this method, or + the two will call each other forever. The shipped providers that use + that delegation (Bedrock) do override both — keep that pairing. """ embedding = await self.embed(text) return embedding, self._estimate_tokens([text]) @@ -86,6 +92,11 @@ class Provider(ABC): Returns ``(embeddings, token_count)``; the default estimates. See :meth:`embed_with_usage`. + + IMPORTANT (recursion invariant): this default calls ``self.embed_batch``. + A provider that overrides ``embed_batch()`` to delegate to + ``embed_batch_with_usage()`` (Mistral, OpenAI, Ollama do) MUST also + override this method, or the two recurse infinitely. Keep the pairing. """ embeddings = await self.embed_batch(texts) return embeddings, self._estimate_tokens(texts) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 6dd7cac2..4a14fd73 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -112,6 +112,65 @@ def should_use_page_aware( return page_aware_enabled and doc_type == "file" and bool(page_boundaries) +async def record_indexing_usage( + *, + enabled: bool, + provider: str, + model: str, + doc_type: str, + user_id: str, + chunk_count: int, + token_count: int, + total_chars: int, +) -> None: + """Record the two billable usage events for one embedded document. + + ``pages_chunks`` is the volume (chunks embedded); ``embeddings_queries`` is + the embedding request's token count — the same metric search records, so the + meter bills embedding tokens whether they were incurred indexing a document + or embedding a query (Deck #67). + + Best-effort and flag-gated: a metering failure is logged and never breaks + indexing. No-op when metering is disabled or the document produced no chunks + (an empty batch embeds nothing and would only write zero-value rows). + + Privacy note: ``user_id`` stays tenant-local — the CP rollup aggregates + GROUP BY (day, metric) into ``usage_daily`` (no metadata column), so nothing + here reaches Stripe; it is retained only to keep Deck #67's future per-user + attribution derivable from the app DB without a re-migration. + """ + if not enabled or chunk_count == 0: + return + + metadata = { + "provider": provider, + "model": model, + "doc_type": doc_type, + "user_id": user_id, + "total_chars": total_chars, + } + try: + store = await UsageEventStore.shared() + # enabled=True: the guard above already confirmed the flag, so the store + # skips a second uncached Settings build per record (ADR-024). + await store.record_usage_event( + metric="pages_chunks", value=chunk_count, metadata=metadata, enabled=True + ) + await store.record_usage_event( + metric="embeddings_queries", + value=token_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, + # so warn rather than hide the "enabled but no billing data" case. + logger.warning( + "usage metering hook (indexing embeddings) skipped", exc_info=True + ) + + async def processor_task( worker_id: int, receive_stream: MemoryObjectReceiveStream[DocumentTask], @@ -825,55 +884,21 @@ async def _index_document( chunks=len(chunk_texts), chars=total_chars, ) - # Usage metering (Deck #67): record chunks embedded as a billable - # 'pages_chunks' event. Best-effort and gated on the flag so the - # off-path (OSS default) touches no storage; placed after the - # embedding succeeds so it can never affect the indexing path. - # - # Privacy note: user_id stays tenant-local — the CP rollup - # aggregates GROUP BY (day, metric) into usage_daily (no metadata - # column), so nothing here reaches Stripe; it is retained only to - # keep Deck #67's future per-user attribution derivable from the - # app DB without a re-migration. - if settings.usage_metering_enabled: - # Two billable events per indexed document: 'pages_chunks' is - # the volume (chunks embedded); 'embeddings_queries' is the - # token count of the embedding request — the same metric search - # records, so the meter bills embedding tokens whether they were - # incurred indexing a document or embedding a query (Deck #67). - metering_metadata = { - "provider": provider, - "model": settings.get_embedding_model_name(), - "doc_type": doc_task.doc_type, - "user_id": doc_task.user_id, - "total_chars": total_chars, - } - try: - store = await UsageEventStore.shared() - await store.record_usage_event( - metric="pages_chunks", - value=len(chunk_texts), - metadata=metering_metadata, - # The outer guard already confirmed the flag, so pass - # enabled=True directly — the store then skips a second - # uncached Settings build here (ADR-024). - enabled=True, - ) - await store.record_usage_event( - metric="embeddings_queries", - value=embed_tokens, - metadata=metering_metadata, - enabled=True, - ) - except Exception: - # Reached only when shared()/store construction itself - # raises (record_usage_event swallows its own write - # failures). Metering is on, so warn rather than hide the - # "enabled but no billing data" case in DEBUG logs. - logger.warning( - "usage metering hook (indexing embeddings) skipped", - exc_info=True, - ) + # Usage metering (Deck #67): record the chunk volume + + # embedding-token count for this document. Best-effort and + # flag-gated; placed after the embedding succeeds so it can never + # affect the indexing path. See record_indexing_usage for the + # metric/privacy details. + await record_indexing_usage( + enabled=settings.usage_metering_enabled, + provider=provider, + model=settings.get_embedding_model_name(), + doc_type=doc_task.doc_type, + user_id=doc_task.user_id, + chunk_count=len(chunk_texts), + token_count=embed_tokens, + total_chars=total_chars, + ) async def generate_sparse_embeddings(): """Generate sparse embeddings (BM25 for keyword matching).""" diff --git a/tests/unit/providers/test_gateway_provider.py b/tests/unit/providers/test_gateway_provider.py index 24b7fbc3..3a834358 100644 --- a/tests/unit/providers/test_gateway_provider.py +++ b/tests/unit/providers/test_gateway_provider.py @@ -331,8 +331,9 @@ def test_trailing_slash_base_url_normalized(): async def test_gateway_embed_with_usage_forwards_after_bearer(monkeypatch): """embed_with_usage refreshes the bearer, then returns the (embedding, token_count) from the inherited OpenAI implementation.""" + # https mock host (never contacted — the OpenAI client is patched below). provider = GatewayProvider( - base_url="http://gw:8083/v1", embedding_model="mistral/mistral-embed" + base_url="https://gw:8083/v1", embedding_model="mistral/mistral-embed" ) order: list[str] = [] @@ -364,8 +365,9 @@ async def test_gateway_embed_with_usage_forwards_after_bearer(monkeypatch): async def test_gateway_embed_batch_with_usage_forwards_after_bearer(monkeypatch): """embed_batch_with_usage also refreshes the bearer before delegating.""" + # https mock host (never contacted — the OpenAI client is patched below). provider = GatewayProvider( - base_url="http://gw:8083/v1", embedding_model="mistral/mistral-embed" + base_url="https://gw:8083/v1", embedding_model="mistral/mistral-embed" ) ensured = {"n": 0} diff --git a/tests/unit/providers/test_ollama.py b/tests/unit/providers/test_ollama.py index 0a457107..94505dde 100644 --- a/tests/unit/providers/test_ollama.py +++ b/tests/unit/providers/test_ollama.py @@ -15,8 +15,9 @@ from nextcloud_mcp_server.providers.ollama import OllamaProvider @pytest.fixture def ollama_provider(): # Construct with no models so __init__ skips _check_model_is_loaded (no - # network call), then enable embeddings post-construction. - provider = OllamaProvider(base_url="http://ollama:11434") + # network call), then enable embeddings post-construction. https mock host + # (never contacted — client.post is patched in each test). + provider = OllamaProvider(base_url="https://ollama:11434") provider.embedding_model = "nomic-embed-text" return provider diff --git a/tests/unit/test_processor_metering.py b/tests/unit/test_processor_metering.py new file mode 100644 index 00000000..9a9630ef --- /dev/null +++ b/tests/unit/test_processor_metering.py @@ -0,0 +1,104 @@ +"""Unit tests for the indexing-path usage-metering helper (Deck #67). + +``record_indexing_usage`` records the two billable events (``pages_chunks`` + +``embeddings_queries``) after a document's chunks are embedded. These cover the +value mapping, the flag/zero-chunk no-ops, and the best-effort failure path +without standing up the full document pipeline. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nextcloud_mcp_server.vector import processor + + +@pytest.fixture +def store_spy(monkeypatch): + """Patch UsageEventStore.shared() to return a spy store.""" + store = MagicMock() + store.record_usage_event = AsyncMock() + monkeypatch.setattr( + processor.UsageEventStore, "shared", AsyncMock(return_value=store) + ) + return store + + +@pytest.mark.unit +async def test_records_pages_chunks_and_token_count(store_spy): + """Both events fire: pages_chunks = chunk count, embeddings_queries = tokens.""" + await processor.record_indexing_usage( + enabled=True, + provider="mistral", + model="mistral-embed", + doc_type="file", + user_id="alice", + chunk_count=110, + token_count=4242, + total_chars=170826, + ) + + calls = store_spy.record_usage_event.await_args_list + by_metric = {c.kwargs["metric"]: c.kwargs["value"] for c in calls} + assert by_metric == {"pages_chunks": 110, "embeddings_queries": 4242} + for c in calls: + # Hot-path fast-gate + tenant-local attribution metadata. + assert c.kwargs["enabled"] is True + assert c.kwargs["metadata"]["provider"] == "mistral" + assert c.kwargs["metadata"]["model"] == "mistral-embed" + assert c.kwargs["metadata"]["user_id"] == "alice" + assert c.kwargs["metadata"]["doc_type"] == "file" + + +@pytest.mark.unit +async def test_disabled_is_noop(store_spy): + """Flag off → no store access, no events.""" + await processor.record_indexing_usage( + enabled=False, + provider="mistral", + model="mistral-embed", + doc_type="file", + user_id="alice", + chunk_count=10, + token_count=20, + total_chars=5, + ) + store_spy.record_usage_event.assert_not_awaited() + + +@pytest.mark.unit +async def test_zero_chunks_is_noop(store_spy): + """A document with no chunks records nothing (no zero-value rows).""" + await processor.record_indexing_usage( + enabled=True, + provider="mistral", + model="mistral-embed", + doc_type="file", + user_id="alice", + chunk_count=0, + token_count=0, + total_chars=0, + ) + store_spy.record_usage_event.assert_not_awaited() + + +@pytest.mark.unit +async def test_store_failure_is_swallowed(monkeypatch): + """A store-construction failure is logged, never raised into indexing.""" + monkeypatch.setattr( + processor.UsageEventStore, + "shared", + AsyncMock(side_effect=RuntimeError("boom")), + ) + + # Must not raise. + await processor.record_indexing_usage( + enabled=True, + provider="mistral", + model="mistral-embed", + doc_type="file", + user_id="alice", + chunk_count=3, + token_count=7, + total_chars=9, + )