Merge pull request #875 from cbcoutinho/feat/meter-embedding-tokens

feat(usage): meter embedding tokens (tokens_embedded/pages_embedded) on both paths + Prometheus export

nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py

@@ -55,7 +55,7 @@ def upgrade() -> None:
             postgresql.TIMESTAMP(timezone=True) if is_pg else sa.TIMESTAMP(),
             nullable=False,
         ),
-        # Catalog metric: 'embeddings_queries' or 'pages_chunks'. Deliberately
+        # Catalog metric: 'tokens_embedded' or 'pages_embedded'. Deliberately
         # an unconstrained Text (no CHECK/enum) — the metric catalog lives in
         # control-plane config, not the app-DB schema. If a third metric is
         # ever added, the CP-side catalog must learn it too, or its rollup will

nextcloud_mcp_server/embedding/gateway_client.py

@@ -217,10 +217,21 @@ class GatewayProvider(OpenAIProvider):
                 exc,
             )
 
+    # Bearer-refresh override topology. OpenAIProvider routes embed_batch(),
+    # embed_with_usage() and embed_batch_with_usage() all through
+    # embed_batch_with_usage(); only embed() (single) is self-contained. So we
+    # override exactly two methods to refresh the bearer exactly once on every
+    # path: embed() (its own entrypoint) and embed_batch_with_usage() (the
+    # shared funnel for the other three). Overriding embed_batch() as well would
+    # double-call _ensure_bearer() (override → super().embed_batch() →
+    # self.embed_batch_with_usage() → override again).
+
     async def embed(self, text: str) -> list[float]:
         await self._ensure_bearer()
         return await super().embed(text)
 
-    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
+    async def embed_batch_with_usage(
+        self, texts: list[str]
+    ) -> tuple[list[list[float]], int]:
         await self._ensure_bearer()
-        return await super().embed_batch(texts)
+        return await super().embed_batch_with_usage(texts)

nextcloud_mcp_server/embedding/service.py

@@ -52,6 +52,22 @@ class EmbeddingService:
         """
         return await self.provider.embed_batch(texts)
 
+    async def embed_with_usage(self, text: str) -> tuple[list[float], int]:
+        """Embed one text and report the request's token count.
+
+        Returns ``(embedding, token_count)`` for usage metering (Deck #67).
+        """
+        return await self.provider.embed_with_usage(text)
+
+    async def embed_batch_with_usage(
+        self, texts: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Embed multiple texts and report the total token count.
+
+        Returns ``(embeddings, token_count)`` for usage metering (Deck #67).
+        """
+        return await self.provider.embed_batch_with_usage(texts)
+
     def get_dimension(self) -> int:
         """
         Get embedding dimension.

nextcloud_mcp_server/observability/metrics.py

@@ -338,6 +338,22 @@ embedding_chars_total = Counter(
     ["kind", "provider"],
 )
 
+# Token consumption — the billed cost unit (mirrors the tokens_embedded billing
+# measure, Deck #67). On a dedicated counter (not folded into the chunk/request
+# metrics above) so query embeds don't inflate indexing dashboards; labelled by
+# operation = index | query. Always emitted, independent of USAGE_METERING_ENABLED.
+#
+# Dashboard note: operation="query" is recorded at embed time (before Qdrant /
+# verify-on-read), whereas the billing-store tokens_embedded row is written only
+# after the search fully succeeds. So this counter can legitimately exceed the
+# billing aggregate when a search fails post-embed — don't alert on that gap as
+# a divergence bug.
+embedding_tokens_total = Counter(
+    "astrolabe_embedding_tokens_total",
+    "Total embedding tokens consumed (provider-reported or estimated)",
+    ["provider", "operation"],  # operation: index | query
+)
+
 # --- Chunking & indexed-by-type -----------------------------------------------
 
 document_chunks_total = Counter(
@@ -727,6 +743,25 @@ def record_embedding(
             embedding_chars_total.labels(kind=kind, provider=provider).inc(chars)
 
 
+def record_embedding_tokens(provider: str, operation: str, tokens: int) -> None:
+    """Export embedding token consumption to Prometheus.
+
+    Mirrors the ``tokens_embedded`` billing measure (Deck #67) as an always-on
+    observability signal — emitted regardless of ``USAGE_METERING_ENABLED`` so
+    OSS/self-host deployments still see token cost in Grafana.
+
+    Args:
+        provider: Provider family (mistral | openai | bedrock | ollama | simple).
+        operation: ``"index"`` (chunk-batch embedding) or ``"query"`` (search
+            query embedding).
+        tokens: Token count for this embedding request (no-op when ``<= 0``).
+    """
+    if tokens > 0:
+        embedding_tokens_total.labels(provider=provider, operation=operation).inc(
+            tokens
+        )
+
+
 def record_document_chunks(doc_type: str, count: int) -> None:
     """
     Record the number of chunks produced for a document.

nextcloud_mcp_server/providers/base.py

@@ -1,5 +1,6 @@
 """Unified provider interface for embeddings and text generation."""
 
+import math
 from abc import ABC, abstractmethod
 
 
@@ -55,6 +56,51 @@ class Provider(ABC):
         """
         pass
 
+    @staticmethod
+    def _estimate_tokens(texts: list[str]) -> int:
+        """Best-effort token estimate when a provider returns no usage data.
+
+        Uses a coarse ~4-chars-per-token heuristic so the billable token
+        value stays non-zero and monotone with input size for local/dev
+        providers (Simple, Ollama without ``prompt_eval_count``). Real
+        providers override ``*_with_usage`` to report exact counts.
+        """
+        return math.ceil(sum(len(t) for t in texts) / 4)
+
+    async def embed_with_usage(self, text: str) -> tuple[list[float], int]:
+        """Embed one text and report the request's token count.
+
+        Returns ``(embedding, token_count)``. The default delegates to
+        :meth:`embed` and estimates the tokens; providers that surface real
+        usage from their embedding response override this. Used by the
+        usage-metering hooks (Deck #67) to bill ``tokens_embedded`` 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])
+
+    async def embed_batch_with_usage(
+        self, texts: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Embed multiple texts and report the total token count.
+
+        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)
+
     @abstractmethod
     def get_dimension(self) -> int:
         """

nextcloud_mcp_server/providers/bedrock.py

@@ -164,6 +164,16 @@ class BedrockProvider(Provider):
             NotImplementedError: If embeddings not enabled (no embedding_model)
             ClientError: If Bedrock API call fails
         """
+        embedding, _ = await self.embed_with_usage(text)
+        return embedding
+
+    async def embed_with_usage(self, text: str) -> tuple[list[float], int]:
+        """Embed one text, reporting the request's token count.
+
+        Titan Embed responses carry ``inputTextTokenCount``; for Cohere /
+        unknown models (no token field) this falls back to a char-based
+        estimate. Used by the usage-metering hooks (Deck #67).
+        """
         if not self.supports_embeddings:
             raise NotImplementedError(
                 "Embedding not supported - no embedding_model configured"
@@ -182,7 +192,13 @@ class BedrockProvider(Provider):
             response_body = json.loads(response["body"].read())
             embedding = self._parse_embedding_response(response_body)
 
-            return embedding
+            token_count = response_body.get("inputTextTokenCount")
+            tokens = (
+                round(token_count)
+                if isinstance(token_count, (int, float))
+                else self._estimate_tokens([text])
+            )
+            return embedding, tokens
 
         except (BotoCoreError, ClientError) as e:
             logger.error("Bedrock embedding error: %s", e)
@@ -205,16 +221,30 @@ class BedrockProvider(Provider):
             NotImplementedError: If embeddings not enabled (no embedding_model)
             ClientError: If Bedrock API call fails
         """
+        embeddings, _ = await self.embed_batch_with_usage(texts)
+        return embeddings
+
+    async def embed_batch_with_usage(
+        self, texts: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Embed multiple texts, summing the per-call token counts.
+
+        Bedrock has no batch embedding API, so requests run sequentially and
+        the token total is the sum of each call's ``inputTextTokenCount``
+        (Titan) or estimate (Cohere/unknown).
+        """
         if not self.supports_embeddings:
             raise NotImplementedError(
                 "Embedding not supported - no embedding_model configured"
             )
 
-        embeddings = []
+        embeddings: list[list[float]] = []
+        total_tokens = 0
         for text in texts:
-            embedding = await self.embed(text)
+            embedding, tokens = await self.embed_with_usage(text)
             embeddings.append(embedding)
-        return embeddings
+            total_tokens += tokens
+        return embeddings, total_tokens
 
     async def _detect_dimension(self):
         """

nextcloud_mcp_server/providers/mistral.py

@@ -122,17 +122,42 @@ class MistralProvider(Provider):
 
     async def embed_batch(self, texts: list[str]) -> list[list[float]]:
         """Generate embeddings for multiple texts, chunking by ``BATCH_SIZE``."""
+        embeddings, _ = await self.embed_batch_with_usage(texts)
+        return embeddings
+
+    async def embed_with_usage(self, text: str) -> tuple[list[float], int]:
+        """Embed one text, reporting the Mistral request's token count."""
+        embeddings, tokens = await self.embed_batch_with_usage([text])
+        if not embeddings:
+            raise RuntimeError(
+                f"Mistral embeddings API returned no embedding for model "
+                f"{self.embedding_model}"
+            )
+        return embeddings[0], tokens
+
+    async def embed_batch_with_usage(
+        self, texts: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Embed multiple texts, summing the Mistral-reported token usage.
+
+        Returns ``(embeddings, total_tokens)`` where ``total_tokens`` is the
+        sum of ``response.usage.total_tokens`` across the ``BATCH_SIZE`` sub-
+        requests (the unit Mistral bills on). Used by the usage-metering hooks
+        to record ``tokens_embedded`` by tokens (Deck #67).
+        """
         if not self.supports_embeddings:
             raise NotImplementedError(_NO_EMBEDDING_MODEL_MSG)
 
         if not texts:
-            return []
+            return [], 0
 
         all_embeddings: list[list[float]] = []
+        total_tokens = 0
         for i in range(0, len(texts), BATCH_SIZE):
             batch = texts[i : i + BATCH_SIZE]
-            batch_embeddings = await self._embed_batch_request(batch)
+            batch_embeddings, batch_tokens = await self._embed_batch_request(batch)
             all_embeddings.extend(batch_embeddings)
+            total_tokens += batch_tokens
 
             if self._dimension is None and batch_embeddings:
                 self._dimension = len(batch_embeddings[0])
@@ -142,11 +167,18 @@ class MistralProvider(Provider):
                     self.embedding_model,
                 )
 
-        return all_embeddings
+        return all_embeddings, total_tokens
 
     @_retry_429
-    async def _embed_batch_request(self, batch: list[str]) -> list[list[float]]:
-        """Single batch request with rate-limit retry."""
+    async def _embed_batch_request(
+        self, batch: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Single batch request with rate-limit retry.
+
+        Returns ``(embeddings, token_count)``; ``token_count`` comes from the
+        response's ``usage.total_tokens`` and falls back to a char-based
+        estimate if the API omits usage.
+        """
         assert self.embedding_model is not None
         response = await self.client.embeddings.create_async(
             model=self.embedding_model,
@@ -170,7 +202,18 @@ class MistralProvider(Provider):
                 f"Mistral embeddings API returned {len(result)} embeddings "
                 f"for {len(batch)} inputs"
             )
-        return result
+
+        usage = getattr(response, "usage", None)
+        total_tokens = getattr(usage, "total_tokens", None) if usage else None
+        # Guard on numeric type (not just ``is not None``): a real response
+        # gives an int, but test doubles / partial responses can surface a
+        # non-numeric attribute — fall back to the estimate there.
+        tokens = (
+            round(total_tokens)
+            if isinstance(total_tokens, (int, float))
+            else self._estimate_tokens(batch)
+        )
+        return result, tokens
 
     def get_dimension(self) -> int:
         if not self.supports_embeddings:

nextcloud_mcp_server/providers/ollama.py

@@ -82,17 +82,12 @@ class OllamaProvider(Provider):
         Raises:
             NotImplementedError: If embeddings not enabled (no embedding_model)
         """
-        if not self.supports_embeddings:
-            raise NotImplementedError(
-                "Embedding not supported - no embedding_model configured"
-            )
-
-        response = await self.client.post(
-            f"{self.base_url}/api/embeddings",
-            json={"model": self.embedding_model, "prompt": text},
-        )
-        response.raise_for_status()
-        return response.json()["embedding"]
+        # Delegate to embed_with_usage so single and batch embeds use the same
+        # /api/embed endpoint (the legacy /api/embeddings differs in payload and
+        # omits prompt_eval_count). _detect_dimension() and other embed() callers
+        # therefore stay consistent with the search/indexing path.
+        embedding, _ = await self.embed_with_usage(text)
+        return embedding
 
     async def embed_batch(
         self, texts: list[str], batch_size: int = 32
@@ -116,12 +111,44 @@ class OllamaProvider(Provider):
         Raises:
             NotImplementedError: If embeddings not enabled (no embedding_model)
         """
+        embeddings, _ = await self.embed_batch_with_usage(texts, batch_size=batch_size)
+        return embeddings
+
+    async def embed_with_usage(self, text: str) -> tuple[list[float], int]:
+        """Embed one text, reporting the request's token count.
+
+        Routes through ``/api/embed`` (which carries ``prompt_eval_count``)
+        rather than the legacy ``/api/embeddings`` so a token count is
+        available; falls back to a char-based estimate when the field is
+        absent. Used by the usage-metering hooks (Deck #67).
+        """
+        embeddings, tokens = await self.embed_batch_with_usage([text])
+        if not embeddings:
+            raise RuntimeError(
+                "Ollama embeddings API returned no embedding for model "
+                f"{self.embedding_model}"
+            )
+        return embeddings[0], tokens
+
+    async def embed_batch_with_usage(
+        self, texts: list[str], batch_size: int = 32
+    ) -> tuple[list[list[float]], int]:
+        """Embed multiple texts, summing ``prompt_eval_count`` token usage.
+
+        Returns ``(embeddings, total_tokens)``. Ollama's ``/api/embed`` may
+        omit ``prompt_eval_count`` (older versions); a char-based estimate is
+        used per batch when it does.
+        """
         if not self.supports_embeddings:
             raise NotImplementedError(
                 "Embedding not supported - no embedding_model configured"
             )
 
-        all_embeddings = []
+        if not texts:
+            return [], 0
+
+        all_embeddings: list[list[float]] = []
+        total_tokens = 0
         for i in range(0, len(texts), batch_size):
             batch = texts[i : i + batch_size]
             response = await self.client.post(
@@ -129,9 +156,29 @@ class OllamaProvider(Provider):
                 json={"model": self.embedding_model, "input": batch},
             )
             response.raise_for_status()
-            all_embeddings.extend(response.json()["embeddings"])
+            data = response.json()
+            all_embeddings.extend(data["embeddings"])
 
-        return all_embeddings
+            # Cache the dimension inline (mirrors OpenAI/Mistral) so it is set
+            # via any embed path, not only an explicit _detect_dimension() call.
+            if self._dimension is None and data["embeddings"]:
+                self._dimension = len(data["embeddings"][0])
+
+            # ``prompt_eval_count`` is assumed to be the batch-level total for a
+            # multi-input /api/embed call. Ollama's API docs aren't explicit
+            # about batch aggregation; if a version reports only the last
+            # input's tokens this understates the batch. Unverified against a
+            # live instance — Ollama isn't the Cloud billing provider (Mistral
+            # is). If it proves last-item-only, switch to per-item requests and
+            # sum. The char-based estimate covers versions that omit the field.
+            prompt_eval = data.get("prompt_eval_count")
+            total_tokens += (
+                round(prompt_eval)
+                if isinstance(prompt_eval, (int, float))
+                else self._estimate_tokens(batch)
+            )
+
+        return all_embeddings, total_tokens
 
     async def _detect_dimension(self):
         """

nextcloud_mcp_server/providers/openai.py

@@ -153,19 +153,49 @@ class OpenAIProvider(Provider):
                 "Embedding not supported - no embedding_model configured"
             )
 
+        embeddings, _ = await self.embed_batch_with_usage(texts)
+        return embeddings
+
+    async def embed_with_usage(self, text: str) -> tuple[list[float], int]:
+        """Embed one text, reporting the request's token count."""
+        embeddings, tokens = await self.embed_batch_with_usage([text])
+        if not embeddings:
+            raise RuntimeError(
+                "OpenAI embeddings API returned no embedding for model "
+                f"{self.embedding_model}"
+            )
+        return embeddings[0], tokens
+
+    async def embed_batch_with_usage(
+        self, texts: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Embed multiple texts, summing the API-reported token usage.
+
+        Returns ``(embeddings, total_tokens)`` where ``total_tokens`` sums
+        ``response.usage.total_tokens`` across the sub-requests (the unit the
+        provider bills on). Used by the usage-metering hooks (Deck #67). Also
+        serves the gateway path via :class:`GatewayProvider`.
+        """
+        if not self.supports_embeddings:
+            raise NotImplementedError(
+                "Embedding not supported - no embedding_model configured"
+            )
+
         if not texts:
-            return []
+            return [], 0
 
         # OpenAI supports batches up to 2048, but use smaller batches for safety
         batch_size = 100
         all_embeddings: list[list[float]] = []
+        total_tokens = 0
 
         for i in range(0, len(texts), batch_size):
             batch = texts[i : i + batch_size]
 
             # Use helper method with retry logic for each batch
-            batch_embeddings = await self._embed_batch_request(batch)
+            batch_embeddings, batch_tokens = await self._embed_batch_request(batch)
             all_embeddings.extend(batch_embeddings)
+            total_tokens += batch_tokens
 
             # Update dimension if not set
             if self._dimension is None and batch_embeddings:
@@ -176,11 +206,18 @@ class OpenAIProvider(Provider):
                     self.embedding_model,
                 )
 
-        return all_embeddings
+        return all_embeddings, total_tokens
 
     @_retry_429
-    async def _embed_batch_request(self, batch: list[str]) -> list[list[float]]:
-        """Make a single batch embedding request with retry logic."""
+    async def _embed_batch_request(
+        self, batch: list[str]
+    ) -> tuple[list[list[float]], int]:
+        """Make a single batch embedding request with retry logic.
+
+        Returns ``(embeddings, token_count)``; ``token_count`` comes from the
+        response's ``usage.total_tokens`` and falls back to a char-based
+        estimate if the API omits usage.
+        """
         assert self.embedding_model is not None  # Type narrowing
         response = await self.client.embeddings.create(
             input=batch,
@@ -188,7 +225,19 @@ class OpenAIProvider(Provider):
         )
         # Sort by index to maintain order
         sorted_data = sorted(response.data, key=lambda x: x.index)
-        return [item.embedding for item in sorted_data]
+        embeddings = [item.embedding for item in sorted_data]
+
+        usage = getattr(response, "usage", None)
+        total_tokens = getattr(usage, "total_tokens", None) if usage else None
+        # Guard on numeric type (not just ``is not None``): a real response
+        # gives an int, but test doubles / partial responses can surface a
+        # non-numeric attribute — fall back to the estimate there.
+        tokens = (
+            round(total_tokens)
+            if isinstance(total_tokens, (int, float))
+            else self._estimate_tokens(batch)
+        )
+        return embeddings, tokens
 
     def get_dimension(self) -> int:
         """

nextcloud_mcp_server/search/algorithms.py

@@ -285,9 +285,25 @@ class SearchAlgorithm(ABC):
         query_embedding: The query embedding generated during the last search.
             Available after search() completes for algorithms that use embeddings.
             Can be reused by callers to avoid redundant embedding generation.
+        query_token_count: Token count of the query embedding request from the
+            last search (provider-reported, or estimated). Set by algorithms
+            that embed the query so the usage-metering hook can bill
+            ``tokens_embedded`` by tokens (Deck #67). The instance is
+            per-request, so this side-channel is concurrency-safe.
     """
 
+    # Class-level defaults are a safety net; __init__ shadows them per instance.
     query_embedding: list[float] | None = None
+    query_token_count: int | None = None
+
+    def __init__(self) -> None:
+        # Set the query-embedding side-channel as instance attributes so
+        # concurrent SearchAlgorithm instances never share it through the
+        # class-level defaults above — per-request isolation by construction,
+        # not just by the convention that each subclass redeclares them.
+        # Subclasses with their own __init__ should call super().__init__().
+        self.query_embedding: list[float] | None = None
+        self.query_token_count: int | None = None
 
     @abstractmethod
     async def search(

nextcloud_mcp_server/search/bm25_hybrid.py

@@ -9,7 +9,10 @@ from qdrant_client.models import Filter
 
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service
-from nextcloud_mcp_server.observability.metrics import record_qdrant_operation
+from nextcloud_mcp_server.observability.metrics import (
+    record_embedding_tokens,
+    record_qdrant_operation,
+)
 from nextcloud_mcp_server.observability.tracing import trace_operation
 from nextcloud_mcp_server.search.access_filter import build_base_filter_conditions
 from nextcloud_mcp_server.search.algorithms import (
@@ -53,9 +56,16 @@ class BM25HybridSearchAlgorithm(SearchAlgorithm):
                 f"Invalid fusion algorithm '{fusion}'. Must be 'rrf' or 'dbsf'"
             )
 
+        # super() sets the per-instance query_embedding / query_token_count
+        # side-channel; this adds the cache key for it.
+        super().__init__()
         self.score_threshold = score_threshold
         self.fusion = models.Fusion.RRF if fusion == "rrf" else models.Fusion.DBSF
         self.fusion_name = fusion
+        # ``_embedded_query`` is the query string whose dense embedding is held
+        # in ``query_embedding`` — repeated search() calls on this per-request
+        # instance (the doc_types loop) reuse it instead of re-embedding.
+        self._embedded_query: str | None = None
 
     @property
     def name(self) -> str:
@@ -128,13 +138,35 @@ class BM25HybridSearchAlgorithm(SearchAlgorithm):
             self.fusion_name,
         )
 
-        # Generate dense embedding for semantic search
+        # Generate dense embedding for semantic search. Cache it per query on
+        # this (per-request) instance: nc_semantic_search calls search() once
+        # per doc_type with the same query, so re-embedding each time would make
+        # N redundant API calls and bill the query's tokens N times (Deck #67).
+        # Reuse the first call's embedding + token count so the query is embedded
+        # — and metered — exactly once.
         with trace_operation("search.get_embedding_service"):
             embedding_service = get_embedding_service()
         with trace_operation("search.dense_embedding"):
-            dense_embedding = await embedding_service.embed(query)
-        # Store for reuse by callers (e.g., viz_routes PCA visualization)
-        self.query_embedding = dense_embedding
+            if self.query_embedding is not None and self._embedded_query == query:
+                dense_embedding = self.query_embedding
+            else:
+                (
+                    dense_embedding,
+                    query_tokens,
+                ) = await embedding_service.embed_with_usage(query)
+                # Store for reuse by callers (e.g., viz_routes PCA
+                # visualization) and for the usage-metering hook in
+                # server/semantic.py (token count).
+                self.query_embedding = dense_embedding
+                self.query_token_count = query_tokens
+                self._embedded_query = query
+                # Export query-embedding token cost to Prometheus
+                # (operation=query), mirroring the per-search billing record in
+                # server/semantic.py. Inside the cache-miss branch so a reused
+                # embedding isn't double-counted.
+                record_embedding_tokens(
+                    settings.get_embedding_provider_family(), "query", query_tokens
+                )
         logger.debug("Generated dense embedding (dimension=%s)", len(dense_embedding))
 
         # Generate sparse embedding for BM25 keyword search

nextcloud_mcp_server/search/semantic.py

@@ -35,6 +35,7 @@ class SemanticSearchAlgorithm(SearchAlgorithm):
         Args:
             score_threshold: Minimum similarity score (0-1, default: 0.7)
         """
+        super().__init__()
         self.score_threshold = score_threshold
 
     @property

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 ``tokens_embedded`` 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. ``nc_semantic_search_answer`` exposes no
+    ``doc_types`` parameter, so its searches always meter with
+    ``doc_types=None``.
+
+    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="tokens_embedded",
+            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 (tokens_embedded) skipped", exc_info=True)
+
+
 def configure_semantic_tools(mcp: FastMCP):
     """Configure semantic search tools for MCP server."""
 
@@ -527,51 +585,29 @@ 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 query embedding is the metered
-            # cost). 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.
+            # Usage metering (Deck #67): record the query embedding's token
+            # count as a billable 'tokens_embedded' 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.
             #
-            # 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=1,
-                        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,
-                    )
+            # NOTE (v1 billing gap): this fires only on a fully successful
+            # search. If the query embed succeeded (provider billed the tokens,
+            # and Prometheus recorded them via record_embedding_tokens) but a
+            # later step (Qdrant/verify) raised, no tokens_embedded row is
+            # written — the embed cost is real but absent from the billing
+            # ledger. Acceptable for v1 (search failures are rare and the meter
+            # is not billed today); revisit if billing accuracy needs it.
+            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,

nextcloud_mcp_server/usage/store.py

@@ -102,8 +102,8 @@ class UsageEventStore:
         logged and swallowed — this must never break the caller's operation.
 
         Args:
-            metric: Catalog metric, e.g. ``"embeddings_queries"`` or
-                ``"pages_chunks"``.
+            metric: Catalog metric, e.g. ``"tokens_embedded"`` or
+                ``"pages_embedded"``.
             value: Count/quantity for this event.
             occurred_at: Operation completion time; defaults to now (UTC).
             metadata: Optional rawest-unit context (provider, model, tokens,

nextcloud_mcp_server/vector/processor.py

@@ -23,6 +23,7 @@ from nextcloud_mcp_server.observability.metrics import (
     record_document_chunks,
     record_document_parse_failed,
     record_embedding,
+    record_embedding_tokens,
     record_qdrant_operation,
     record_vector_sync_processing,
     update_vector_sync_queue_size,
@@ -112,6 +113,80 @@ 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_embedded`` is the buyer-facing "pages indexed" dimension;
+    ``tokens_embedded`` 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).
+
+    TODO(#282): ``pages_embedded`` currently carries the raw chunk count
+    (``len(chunk_texts)``) as an interim value. The real normalized "pages
+    indexed" count — real pages for paginated types (PDF/DOCX/PPT), a fixed
+    chars/tokens-per-page constant otherwise — is deferred to instrumentation
+    card #282; this code (card #284) only lands the metric name/contract.
+
+    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).
+        # record_usage_event swallows its own write failures, so the two records
+        # are independent; if pages_embedded somehow raised mid-way,
+        # tokens_embedded would be skipped, leaving an unmatched pages_embedded
+        # row — acceptable under the (day, metric) SUM-aggregation billing model.
+        await store.record_usage_event(
+            # TODO(#282): value is the interim chunk count; switch to normalized
+            # real-page count when the per-page constant lands.
+            metric="pages_embedded",
+            value=chunk_count,
+            metadata=metadata,
+            enabled=True,
+        )
+        await store.record_usage_event(
+            metric="tokens_embedded",
+            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],
@@ -598,9 +673,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 tokens_embedded 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,
@@ -809,7 +888,10 @@ async def _index_document(
             embedding_service = get_embedding_service()
             embed_start = time.time()
             try:
-                dense_embeddings = await embedding_service.embed_batch(chunk_texts)
+                (
+                    dense_embeddings,
+                    embed_tokens,
+                ) = await embedding_service.embed_batch_with_usage(chunk_texts)
             except Exception:
                 record_embedding(
                     "dense", provider, time.time() - embed_start, status="error"
@@ -822,42 +904,24 @@ 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:
-                try:
-                    store = await UsageEventStore.shared()
-                    await store.record_usage_event(
-                        metric="pages_chunks",
-                        value=len(chunk_texts),
-                        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,
-                        },
-                        # 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,
-                    )
-                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 (pages_chunks) skipped", exc_info=True
-                    )
+            # Export token consumption to Prometheus (always-on, independent of
+            # the billing flag) so Grafana sees indexing token cost.
+            record_embedding_tokens(provider, "index", embed_tokens)
+            # 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)."""