feat: meter Pinecone query read units by family and caller

backend/database/pinecone_partition_index.py

@@ -14,6 +14,7 @@
 from typing import Any, Mapping, Sequence, cast
 
 from database import pinecone_partition_state as state
+from database import pinecone_query_metrics
 from database.legal_holds import external_write_fence
 from database.pinecone_provider_errors import barrier_record, is_missing_namespace
 
@@ -76,9 +77,24 @@ def _check_filter(self, value: Any) -> None:
         if not _owner_filter(value, self.uid):
             raise ValueError('Pinecone filter must constrain the authenticated owner')
 
-    def query(self, *, namespace: str, **kwargs: Any) -> Any:
+    def query(self, *, namespace: str, caller: str = 'other', **kwargs: Any) -> Any:
         self._check_filter(kwargs.get('filter'))
-        return self.index.query(namespace=self._route(namespace), **kwargs)
+        routed_namespace = self._route(namespace)
+        response: Any = None
+        succeeded = False
+        try:
+            response = self.index.query(namespace=routed_namespace, **kwargs)
+            succeeded = True
+            return response
+        finally:
+            self._meter_query(namespace, routed_namespace, caller, response, succeeded=succeeded)
+
+    @staticmethod
+    def _meter_query(family: str, routed_namespace: str, caller: str, response: Any, *, succeeded: bool) -> None:
+        try:
+            pinecone_query_metrics.record_query(family, routed_namespace, caller, response, succeeded=succeeded)
+        except Exception:
+            pinecone_query_metrics.record_metering_error()
 
     def fetch(self, *, ids: list[str], namespace: str, **kwargs: Any) -> Any:
         result = self.index.fetch(ids=ids, namespace=self._route(namespace), **kwargs)
@@ -225,13 +241,26 @@ def verify_empty_all_copies(self, *, namespace: str, filter: dict) -> None:
         dimension = _field(self.index.describe_index_stats(), 'dimension')
         for target in (key.family, key.partition_namespace):
             for attempt in range(6):
-                result, _, headers = self.index._query(
-                    namespace=target,
-                    vector=[1.0] + [0.0] * (dimension - 1),
-                    filter=filter,
-                    top_k=1,
-                    _return_http_data_only=False,
-                )
+                response: Any = None
+                succeeded = False
+                try:
+                    response = self.index._query(
+                        namespace=target,
+                        vector=[1.0] + [0.0] * (dimension - 1),
+                        filter=filter,
+                        top_k=1,
+                        _return_http_data_only=False,
+                    )
+                    succeeded = True
+                finally:
+                    self._meter_query(
+                        namespace,
+                        target,
+                        'account_purge_verify',
+                        response[0] if succeeded else None,
+                        succeeded=succeeded,
+                    )
+                result, _, headers = response
                 indexed = int(headers.get('x-pinecone-max-indexed-lsn', 0 if lsns[target] == 0 else -1))
                 if indexed >= lsns[target] and not _field(result, 'matches'):
                     break

backend/database/pinecone_query_metrics.py

@@ -0,0 +1,78 @@
+"""Best-effort, process-local Pinecone query accounting with closed label sets.
+
+Kept independent of utils.metrics so the provider boundary does not import the
+application metrics graph. All application exporters use the default registry.
+"""
+
+from collections.abc import Mapping
+from typing import Any
+
+from prometheus_client import Counter
+
+FAMILIES = frozenset({'ns1', 'ns2', 'ns3', 'ns4', 'ns_x', 'ns_tchunks', 'workstream-association-v1'})
+CALLERS = frozenset(
+    {
+        'query_vectors',
+        'query_vectors_by_metadata',
+        'query_workstream_association_candidates',
+        'find_similar_memories',
+        'query_memory_vector_candidates',
+        'find_similar_x_posts',
+        'search_screen_activity_vectors',
+        'search_action_items_by_vector',
+        'find_similar_action_items',
+        'search_transcript_chunks',
+        'account_purge_verify',
+    }
+)
+
+QUERIES = Counter(
+    'omi_pinecone_queries_total',
+    'Pinecone provider query attempts by shared namespace family, routed mode and bounded caller',
+    ['family', 'route_mode', 'caller'],
+)
+READ_UNITS = Counter(
+    'omi_pinecone_read_units_total',
+    'Pinecone read units reported on successful query responses',
+    ['family', 'route_mode', 'caller'],
+)
+MISSING_USAGE = Counter(
+    'omi_pinecone_query_missing_usage_total',
+    'Successful Pinecone queries without read-unit usage; no units are inferred',
+    ['family', 'route_mode', 'caller'],
+)
+METERING_ERRORS = Counter('omi_pinecone_query_metering_errors_total', 'Swallowed Pinecone query metering errors')
+
+
+def record_metering_error() -> None:
+    try:
+        METERING_ERRORS.inc()
+    except Exception:
+        pass
+
+
+def _field(value: Any, name: str) -> Any:
+    return value.get(name) if isinstance(value, Mapping) else getattr(value, name, None)
+
+
+def record_query(family: str, routed_namespace: str, caller: str, response: Any, *, succeeded: bool) -> None:
+    """No network, logs, retries or owner labels; never affect provider outcomes."""
+    try:
+        labels = (
+            family if family in FAMILIES else 'other',
+            'legacy' if routed_namespace == family else 'partitioned',
+            caller if caller in CALLERS else 'other',
+        )
+        QUERIES.labels(*labels).inc()
+        if not succeeded:
+            return
+        units = _field(_field(response, 'usage'), 'read_units')
+        if units is None:
+            MISSING_USAGE.labels(*labels).inc()
+            return
+        # The pinned 7.3.0 SDK's Usage.read_units schema is integer-only.
+        if isinstance(units, bool) or not isinstance(units, int) or units < 0:
+            raise ValueError('Invalid Pinecone read-unit usage')
+        READ_UNITS.labels(*labels).inc(units)
+    except Exception:
+        record_metering_error()

backend/database/vector_db.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
 import json
 import logging
@@ -230,7 +230,7 @@
 
     xq = query_vector if query_vector is not None else embeddings.embed_query(query)
     xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
-        vector=xq, top_k=k, include_metadata=False, filter=filter_data, namespace="ns1"
+        caller='query_vectors', vector=xq, top_k=k, include_metadata=False, filter=filter_data, namespace="ns1"
     )
     matches: List[Any] = xc['matches']
     return [item['id'].replace(f'{uid}-', '') for item in matches]
@@ -268,7 +268,13 @@
         )
 
     xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
-        vector=vector, filter=filter_data, namespace="ns1", include_values=False, include_metadata=True, top_k=1000
+        caller='query_vectors_by_metadata',
+        vector=vector,
+        filter=filter_data,
+        namespace="ns1",
+        include_values=False,
+        include_metadata=True,
+        top_k=1000,
     )
     if not xc['matches']:
         # Relax-retry when the structured people/topics/entities $or clause produced no hits, dropping
@@ -280,6 +286,7 @@
             and_clauses.pop(1)
             logger.warning(f'query_vectors_by_metadata retrying without structured filters: {json.dumps(filter_data)}')
             xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
+                caller='query_vectors_by_metadata',
                 vector=vector,
                 filter=filter_data,
                 namespace="ns1",
@@ -373,6 +380,7 @@
     if index is None or not summary.strip():
         return []
     response = owner_index(index, uid, index_name=_pinecone_index_name).query(
+        caller='query_workstream_association_candidates',
         vector=embeddings.embed_query(summary),
         top_k=max(1, min(limit, 20)),
         include_metadata=True,
@@ -559,7 +567,12 @@
     filter_data = build_legacy_memory_vector_filter(uid, subject_entity_id=subject_entity_id)
 
     xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
-        vector=vector, top_k=limit, include_metadata=True, filter=filter_data, namespace=MEMORIES_NAMESPACE
+        caller='find_similar_memories',
+        vector=vector,
+        top_k=limit,
+        include_metadata=True,
+        filter=filter_data,
+        namespace=MEMORIES_NAMESPACE,
     )
 
     results: List[Dict[str, Any]] = []
@@ -768,6 +781,7 @@
             else build_default_memory_vector_filter(uid)
         )
     response = owner_index(index, uid, index_name=_pinecone_index_name).query(
+        caller='query_memory_vector_candidates',
         vector=vector,
         top_k=bounded_limit,
         include_metadata=True,
@@ -849,7 +863,12 @@
         return []
     vector = embeddings.embed_query(content)
     xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
-        vector=vector, top_k=limit, include_metadata=True, filter={'uid': uid}, namespace=X_POSTS_NAMESPACE
+        caller='find_similar_x_posts',
+        vector=vector,
+        top_k=limit,
+        include_metadata=True,
+        filter={'uid': uid},
+        namespace=X_POSTS_NAMESPACE,
     )
     matches: List[Any] = xc.get('matches', [])
     return [
@@ -944,6 +963,7 @@
         filter_data['appName'] = app_filter
 
     xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
+        caller='search_screen_activity_vectors',
         vector=query_vector,
         top_k=k,
         include_metadata=True,
@@ -1087,7 +1107,12 @@
         filter_data: Dict[str, Any] = {'uid': uid}
 
         xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
-            vector=vector, top_k=limit, include_metadata=True, filter=filter_data, namespace=ACTION_ITEMS_NAMESPACE
+            caller='search_action_items_by_vector',
+            vector=vector,
+            top_k=limit,
+            include_metadata=True,
+            filter=filter_data,
+            namespace=ACTION_ITEMS_NAMESPACE,
         )
 
         matches: List[Any] = xc.get('matches', [])
@@ -1132,6 +1157,7 @@
     try:
         vector = embeddings.embed_query(prepared_query)
         xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
+            caller='find_similar_action_items',
             vector=vector,
             top_k=limit,
             include_metadata=True,
@@ -1306,6 +1332,7 @@
     if timeout_seconds is not None:
         query_kwargs['_request_timeout'] = timeout_seconds
     xc = owner_index(index, uid, index_name=_pinecone_index_name).query(
+        caller='search_transcript_chunks',
         vector=vector,
         top_k=limit,
         include_metadata=True,

backend/deploy/cloud_run_gmp_sync.yaml

@@ -37,10 +37,11 @@ spec:
         # C5 exact-input reuse (2026-10-26): the dedupe hit/miss/checkpoint counters
         # are the soak readout for backend-sync/backfill/backend-integration.
         # Bridge measurement: sync_.* includes absorbed and published-row counters.
+        # Pinecone: bounded query/read-unit counters attribute retrieval spend on workers.
         - action: keep
           sourceLabels:
             - __name__
-          regex: 'omi_(playback_writes_total|sync_.*|journey_.*|client_journey_.*|owner_recognition_conversations_total|conversation_duration_seconds(_.*)?|conversation_relevance_decision_total|conversation_summary_vector_upserts_total|conversation_relevance_decision_total|firestore_documents_per_operation(_.*)?|capture_finalization_.*|queue_oldest_ready_.*|batch_pipeline_.*|batch_intake_outcome_total|batch_admission_sample_window_total|batch_ramp_heavy_owner_excluded_total|batch_shadow_.*|accept_to_summary_seconds(_.*)?|summary_terminal_total|summary_debt_.*|live_session_transcript_outcome_total|fallback_total|parakeet_request_seconds(_.*)?|firestore_billed_reads_by_(caller|origin)_total|firestore_caller_label_overflow_total|firestore_billed_writes(_by_caller)?_total|firestore_write_caller_label_overflow_total|firestore_ambiguous_commit_writes_total|firestore_noop_writes_by_caller_total|memory_released_projection_total|fsr_leader_lease_total|held_content_.*|held_recovery_pacing_.*|stt_cap_recovery_reconciliation_total|stt_cap_recovery_cohort_narrowed|stt_cap_cohort_.*|stt_cap_launch_config_errors.*|stt_cap_dispatch_decisions_total|stt_cap_provider_audio_seconds_total|legacy_sync_.*|legacy_dark_.*|jobstore_.*|capture_coverage_.*|capture_dedupe_shadow_total|capture_retry_cap_total|speaker_evidence_.*|speaker_evidence_confident_owner_total|lifecycle_refactor_shadow_total|executor_task_(active|queued)|clients_refactor_shadow_total|notes_refactor_shadow_total|notes_refactor_results_total|stt_language_gate_.*|prerecorded_vendor_.*|episode_compose_.*)|conversation_llm_input_dedupe_.*'
+          regex: 'omi_(pinecone_.*|playback_writes_total|sync_.*|journey_.*|client_journey_.*|owner_recognition_conversations_total|conversation_duration_seconds(_.*)?|conversation_relevance_decision_total|conversation_summary_vector_upserts_total|conversation_relevance_decision_total|firestore_documents_per_operation(_.*)?|capture_finalization_.*|queue_oldest_ready_.*|batch_pipeline_.*|batch_intake_outcome_total|batch_admission_sample_window_total|batch_ramp_heavy_owner_excluded_total|batch_shadow_.*|accept_to_summary_seconds(_.*)?|summary_terminal_total|summary_debt_.*|live_session_transcript_outcome_total|fallback_total|parakeet_request_seconds(_.*)?|firestore_billed_reads_by_(caller|origin)_total|firestore_caller_label_overflow_total|firestore_billed_writes(_by_caller)?_total|firestore_write_caller_label_overflow_total|firestore_ambiguous_commit_writes_total|firestore_noop_writes_by_caller_total|memory_released_projection_total|fsr_leader_lease_total|held_content_.*|held_recovery_pacing_.*|stt_cap_recovery_reconciliation_total|stt_cap_recovery_cohort_narrowed|stt_cap_cohort_.*|stt_cap_launch_config_errors.*|stt_cap_dispatch_decisions_total|stt_cap_provider_audio_seconds_total|legacy_sync_.*|legacy_dark_.*|jobstore_.*|capture_coverage_.*|capture_dedupe_shadow_total|capture_retry_cap_total|speaker_evidence_.*|speaker_evidence_confident_owner_total|lifecycle_refactor_shadow_total|executor_task_(active|queued)|clients_refactor_shadow_total|notes_refactor_shadow_total|notes_refactor_results_total|stt_language_gate_.*|prerecorded_vendor_.*|episode_compose_.*)|conversation_llm_input_dedupe_.*'
         # These eagerly initialized live-STT combinations dominate empty series
         # on sync workers. Keep fallbacks from sync/summary/storage components.
         - action: drop