diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 97fb8338..952a46e2 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -56,6 +56,64 @@ logger = logging.getLogger(__name__) _USAGE_METADATA_MAX_DOC_TYPES = 16 +async def record_search_usage( + *, + enabled: bool, + user_id: str, + fusion: str, + doc_types: list[str] | None, + token_count: int | None, +) -> None: + """Record the billable ``embeddings_queries`` event for one semantic search. + + The value is the query embedding's token count (provider-reported or + estimated) — the unit upstream providers bill on, and the same metric the + indexing path records for chunk embeddings (Deck #67). ``nc_semantic_search`` + and ``nc_semantic_search_answer`` (which reuses it) both flow through here — + do not add a second hook. + + Best-effort and flag-gated: a metering failure is logged and never breaks + the search. Unlike the indexing path's chunk-count guard, a 0-token query is + still recorded (the query embedding ran); a zero-value row is a no-op at the + Stripe ``sum`` aggregation. + + Privacy note: ``user_id`` stays tenant-local — the CP rollup aggregates + GROUP BY (day, metric) into ``usage_daily`` (no metadata column), so nothing + here propagates to Stripe; it is retained only to keep Deck #67's future + per-user attribution derivable from app-DB metadata without a re-migration. + """ + if not enabled: + return + try: + store = await UsageEventStore.shared() + await store.record_usage_event( + metric="embeddings_queries", + value=token_count or 0, + metadata={ + "user_id": user_id, + "fusion": fusion, + # Bounded copy — see _USAGE_METADATA_MAX_DOC_TYPES. Both None and + # [] normalize to null so a future metadata->'doc_types' IS NULL + # query counts the all-types case consistently. + "doc_types": ( + doc_types[:_USAGE_METADATA_MAX_DOC_TYPES] if doc_types else None + ), + }, + # The caller already confirmed the flag, so pass enabled=True + # directly — the store then skips a second uncached Settings build on + # this hot query path (ADR-024). + 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 — a silent DEBUG line would hide "operator enabled metering + # but gets no data". + logger.warning( + "usage metering hook (embeddings_queries) skipped", exc_info=True + ) + + def configure_semantic_tools(mcp: FastMCP): """Configure semantic search tools for MCP server.""" @@ -527,60 +585,21 @@ def configure_semantic_tools(mcp: FastMCP): logger.info("Returning %d results from BM25 hybrid search", len(results)) - # Usage metering (Deck #67): one billable 'embeddings_queries' - # event per successful search. The value is the query embedding's - # token count (provider-reported, or estimated) — the unit upstream - # providers bill on, and the same metric the indexing path records - # for chunk embeddings. Best-effort and gated on the flag so the - # off-path touches no storage. nc_semantic_search_answer reuses this - # tool, so it records here too — do not add a second hook there. - # - # query_token_count is set by BM25HybridSearchAlgorithm during the - # search() above. The doc_types loop reuses one search_algo instance - # for the same query, and the algorithm caches the dense embedding - # per query, so the query is embedded — and metered — exactly once - # regardless of how many doc_types were searched. Falls back to 0 - # only if the embedding never ran (e.g. a pre-embed error). - # - # Privacy note: user_id stays tenant-local. The CP rollup - # aggregates GROUP BY (day, metric) into usage_daily, which has no - # metadata column, so nothing here propagates to Stripe; the value - # is retained only so Deck #67's "per-user attribution derivable - # from app-DB metadata later" stays possible without a re-migration. - if settings.usage_metering_enabled: - try: - store = await UsageEventStore.shared() - await store.record_usage_event( - metric="embeddings_queries", - value=search_algo.query_token_count or 0, - metadata={ - "user_id": username, - "fusion": fusion, - # Bounded copy — see _USAGE_METADATA_MAX_DOC_TYPES. - # Both None and [] normalize to null so a future - # metadata->'doc_types' IS NULL query counts the - # all-types case consistently. - "doc_types": ( - doc_types[:_USAGE_METADATA_MAX_DOC_TYPES] - if doc_types - else None - ), - }, - # The outer guard already confirmed the flag, so pass - # enabled=True directly — the store then skips a second - # uncached Settings build on this hot query path - # (ADR-024). - 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 — a silent DEBUG line - # would hide "operator enabled metering but gets no data". - logger.warning( - "usage metering hook (embeddings_queries) skipped", - exc_info=True, - ) + # Usage metering (Deck #67): record the query embedding's token + # count as a billable 'embeddings_queries' event. query_token_count + # is set by BM25HybridSearchAlgorithm during the search() above; the + # doc_types loop reuses one search_algo instance for the same query + # and the algorithm caches the dense embedding per query, so the + # query is embedded — and metered — exactly once regardless of how + # many doc_types were searched. See record_search_usage for the + # metric/privacy details. + await record_search_usage( + enabled=settings.usage_metering_enabled, + user_id=username, + fusion=fusion, + doc_types=doc_types, + token_count=search_algo.query_token_count, + ) return SemanticSearchResponse( results=results, diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 4a14fd73..a44efbef 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -657,9 +657,13 @@ async def _index_document( doc_type="file", user_id=doc_task.user_id, ) + # No embedding ran, so no usage is recorded here — stated + # explicitly so a "fewer embeddings_queries rows than expected" + # audit lands on the dedup path rather than reconstructing it + # from Qdrant claim logs. logger.info( "Dedup hit for file %s (etag=%s); claimed for user %s " - "without reprocessing", + "without reprocessing (no embedding/usage recorded)", doc_task.doc_id, doc_task.etag, doc_task.user_id, diff --git a/tests/unit/server/__init__.py b/tests/unit/server/__init__.py new file mode 100644 index 00000000..1d5caa11 --- /dev/null +++ b/tests/unit/server/__init__.py @@ -0,0 +1 @@ +"""Unit tests for server-layer MCP tools.""" diff --git a/tests/unit/server/test_semantic_metering.py b/tests/unit/server/test_semantic_metering.py new file mode 100644 index 00000000..e93a79a0 --- /dev/null +++ b/tests/unit/server/test_semantic_metering.py @@ -0,0 +1,109 @@ +"""Unit tests for the search-path usage-metering helper (Deck #67). + +``record_search_usage`` records the billable ``embeddings_queries`` event for a +semantic search. These pin the value mapping (query token count), the flag-off +no-op, the doc_types metadata bounding, and the best-effort failure path — +covering the server-tool metering wiring without standing up the full +``nc_semantic_search`` tool. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nextcloud_mcp_server.server import semantic + + +@pytest.fixture +def store_spy(monkeypatch): + """Patch UsageEventStore.shared() to return a spy store.""" + store = MagicMock() + store.record_usage_event = AsyncMock() + monkeypatch.setattr( + semantic.UsageEventStore, "shared", AsyncMock(return_value=store) + ) + return store + + +@pytest.mark.unit +async def test_records_query_token_count(store_spy): + """The event value is the query embedding's token count.""" + await semantic.record_search_usage( + enabled=True, + user_id="alice", + fusion="rrf", + doc_types=["note", "file"], + token_count=42, + ) + + store_spy.record_usage_event.assert_awaited_once() + kwargs = store_spy.record_usage_event.await_args.kwargs + assert kwargs["metric"] == "embeddings_queries" + assert kwargs["value"] == 42 + assert kwargs["enabled"] is True + assert kwargs["metadata"]["user_id"] == "alice" + assert kwargs["metadata"]["fusion"] == "rrf" + assert kwargs["metadata"]["doc_types"] == ["note", "file"] + + +@pytest.mark.unit +async def test_disabled_is_noop(store_spy): + """Flag off → no store access, no event.""" + await semantic.record_search_usage( + enabled=False, + user_id="alice", + fusion="rrf", + doc_types=None, + token_count=10, + ) + store_spy.record_usage_event.assert_not_awaited() + + +@pytest.mark.unit +async def test_none_token_count_records_zero(store_spy): + """A missing token count (pre-embed error) records value 0, not None.""" + await semantic.record_search_usage( + enabled=True, + user_id="alice", + fusion="dbsf", + doc_types=None, + token_count=None, + ) + kwargs = store_spy.record_usage_event.await_args.kwargs + assert kwargs["value"] == 0 + # None and [] both normalize to null for consistent IS NULL counting. + assert kwargs["metadata"]["doc_types"] is None + + +@pytest.mark.unit +async def test_doc_types_metadata_is_bounded(store_spy): + """A large doc_types list is truncated to the metadata cap.""" + many = [f"type-{i}" for i in range(40)] + await semantic.record_search_usage( + enabled=True, + user_id="alice", + fusion="rrf", + doc_types=many, + token_count=5, + ) + recorded = store_spy.record_usage_event.await_args.kwargs["metadata"]["doc_types"] + assert recorded == many[: semantic._USAGE_METADATA_MAX_DOC_TYPES] + + +@pytest.mark.unit +async def test_store_failure_is_swallowed(monkeypatch): + """A store-construction failure is logged, never raised into the search.""" + monkeypatch.setattr( + semantic.UsageEventStore, + "shared", + AsyncMock(side_effect=RuntimeError("boom")), + ) + + # Must not raise. + await semantic.record_search_usage( + enabled=True, + user_id="alice", + fusion="rrf", + doc_types=None, + token_count=7, + )