diff --git a/backend/config/firestore_quick_reads.py b/backend/config/firestore_quick_reads.py new file mode 100644 index 00000000000..afdf1054a5e --- /dev/null +++ b/backend/config/firestore_quick_reads.py @@ -0,0 +1,22 @@ +# LIFECYCLE: permanent +"""Default-on structural optimizations; independent emergency kill switches.""" + +import os + + +def smart_merge_partition_enabled() -> bool: + return os.getenv('FSR_SMART_MERGE_PARTITION', 'on').strip().lower() != 'off' + + +def kg_citation_index_enabled() -> bool: + return os.getenv('FSR_KG_CITATION_INDEX', 'on').strip().lower() != 'off' + + +def indexed_string_value_safe(value: object) -> bool: + """Only query identifiers fully represented by Firestore's UTF-8 index value.""" + if not isinstance(value, str): + return False + try: + return len(value.encode('utf-8')) <= 1500 + except UnicodeEncodeError: + return False diff --git a/backend/database/firestore_index_registry.py b/backend/database/firestore_index_registry.py index 059954e7509..adfbc28dd35 100644 --- a/backend/database/firestore_index_registry.py +++ b/backend/database/firestore_index_registry.py @@ -1935,6 +1935,24 @@ def _contains(field_path: str) -> FirestoreIndexField: index_fields=(_asc('discarded'), _asc('source'), _asc('status'), _desc('created_at'), _desc('__name__')), ) +CONVERSATIONS_SMART_MERGE_PARTITION_QUERY = FirestoreQuerySpec( + identifier='conversations_smart_merge_device_preceding', + collection_group='conversations', + query_scope='COLLECTION', + filters=( + *CONVERSATIONS_SMART_MERGE_PRECEDING_QUERY.filters, + FirestoreQueryFilter('client_device_id', '==', 'client_device_id'), + ), + index_fields=( + _asc('discarded'), + _asc('source'), + _asc('status'), + _asc('client_device_id'), + _desc('created_at'), + _desc('__name__'), + ), +) + CONVERSATION_PHOTOS_NAME_RANGE_QUERY = FirestoreQuerySpec( identifier='conversation_photos_name_range_export', @@ -2028,6 +2046,7 @@ def _contains(field_path: str) -> FirestoreIndexField: DAY3_REENGAGEMENT_RETURNED_CONVERSATIONS_QUERY, DAILY_SUMMARY_RECIPIENTS_QUERY, CONVERSATIONS_SMART_MERGE_PRECEDING_QUERY, + CONVERSATIONS_SMART_MERGE_PARTITION_QUERY, ) _INDEX_ONLY_REQUIREMENT_SIGNATURES = frozenset(requirement.signature for requirement in INDEX_ONLY_REQUIREMENTS) diff --git a/backend/database/knowledge_graph.py b/backend/database/knowledge_graph.py index 388fc7247ee..3c648dba78f 100644 --- a/backend/database/knowledge_graph.py +++ b/backend/database/knowledge_graph.py @@ -1,6 +1,7 @@ from datetime import datetime, timezone from typing import Any, Dict, Iterable, List, Optional, Tuple, TypedDict, cast import uuid +import logging from google.cloud.firestore_v1 import FieldFilter @@ -10,6 +11,12 @@ from ._client import db from .read_boundary import parse_snapshot_or_none +from config.firestore_quick_reads import kg_citation_index_enabled, indexed_string_value_safe +from database.firestore_read_origin import read_origin +from google.api_core.exceptions import FailedPrecondition +from utils.observability.fallback import record_fallback + +logger = logging.getLogger(__name__) users_collection = 'users' knowledge_nodes_collection = 'knowledge_nodes' @@ -861,6 +868,38 @@ def _batch_delete(coll_ref: Any) -> None: _batch_delete(edges_ref) +def _citation_nodes(nodes_ref: Any, memory_ids: List[str]) -> List[Any]: + """Use Firestore's writer-maintained reverse index, including historical rows. + + Native array indexes cover every successful write and are backfilled by + Firestore before READY. Never infer coverage from an empty side collection. + Materialize all queries before mutations, so fallback cannot repeat writes. + """ + if not kg_citation_index_enabled(): + return list(nodes_ref.stream()) + ids = set(memory_ids) + # A node can match every chunk of a large retraction. Keep the original + # single scan for bulk work rather than multiply billed reads for it. + if len(ids) > 30 or not all(indexed_string_value_safe(value) for value in ids): + return list(nodes_ref.stream()) + try: + query = nodes_ref.where(filter=FieldFilter('memory_ids', 'array_contains_any', sorted(ids))) + return list(query.stream()) + except FailedPrecondition as error: + if 'index' not in str(error).lower(): + raise + record_fallback( + component='other', + from_mode='indexed_query', + to_mode='legacy_query', + reason='unavailable', + outcome='recovered', + log=logger, + ) + return list(nodes_ref.stream()) + + +@read_origin('task:kg_citation_prune') def prune_memory_citations_from_kg(uid: str, memory_ids: List[str], *, db_client: Any = None) -> int: """Remove memory_ids from KG nodes/edges; delete entities with no remaining citations.""" if not memory_ids: @@ -872,7 +911,7 @@ def prune_memory_citations_from_kg(uid: str, memory_ids: List[str], *, db_client edges_ref = user_ref.collection(knowledge_edges_collection) pruned = 0 - for doc in nodes_ref.stream(): + for doc in _citation_nodes(nodes_ref, memory_ids): node_doc: KnowledgeNodeDoc = cast(KnowledgeNodeDoc, _typed_doc(doc)) existing_ids = set(node_doc.get("memory_ids") or []) if not existing_ids.intersection(retracted): diff --git a/backend/database/smart_merge.py b/backend/database/smart_merge.py index fd41e03f1ae..d72a589f388 100644 --- a/backend/database/smart_merge.py +++ b/backend/database/smart_merge.py @@ -23,18 +23,24 @@ from google.cloud import firestore from google.cloud.firestore_v1 import FieldFilter +from google.api_core.exceptions import FailedPrecondition from database import conversations as conversations_db from database import smart_merge_audit as audit_db from database.quiet_check_retirement import stop_quiet_check_for_conversation from database._client import get_firestore_client, run_transactional -from database.firestore_index_registry import CONVERSATIONS_SMART_MERGE_PRECEDING_QUERY +from database.firestore_index_registry import ( + CONVERSATIONS_SMART_MERGE_PRECEDING_QUERY, + CONVERSATIONS_SMART_MERGE_PARTITION_QUERY, +) from database.people_stats_cache import invalidate_people_stats_cache from config.sync_live_dedupe import sync_live_dedupe_active_for from utils.conversations.smart_merge_policy import rebase_donor_segments from utils.sync.bridge_observation import log_bridge_observation from config import merge_ancestry from config.conversation_smart_merge import smart_merge_flatten_enabled +from config.firestore_quick_reads import smart_merge_partition_enabled, indexed_string_value_safe +from database.firestore_read_origin import read_origin from utils.conversations.size_limit import ( CONVERSATION_ABSORPTION_BYTE_CAP, projected_document_bytes, @@ -119,6 +125,7 @@ def find_preceding_conversations( discarded: bool = False, transaction: Any = None, firestore_client: Any = None, + client_device_id: Optional[str] = None, ) -> list[dict[str, Any]]: """Newest-first rows of one source and discard flag created before ``created_before``. @@ -142,13 +149,59 @@ def find_preceding_conversations( query = query.order_by('created_at', direction=firestore.Query.DESCENDING).select(list(_PRECEDING_FIELDS)) if limit is not None: query = query.limit(limit) - stream = query.stream(transaction=transaction) if transaction is not None else query.stream() - rows = [] - for snapshot in stream: - data = snapshot.to_dict() or {} - data['id'] = snapshot.id - rows.append(data) - return rows + + def collect(selected): + stream = selected.stream(transaction=transaction) if transaction is not None else selected.stream() + rows = [] + for snapshot in stream: + data = snapshot.to_dict() or {} + data['id'] = snapshot.id + rows.append(data) + return rows + + # A filter before a finite limit changes the global six-row window. Keep + # that lookup unchanged; the expensive discarded-barrier scan is unbounded. + # Missing device fields are not matched by equality-to-null in Firestore. + if ( + limit is None + and isinstance(client_device_id, str) + and client_device_id + and smart_merge_partition_enabled() + and indexed_string_value_safe(client_device_id) + ): + with read_origin('task:smart_merge_partition_lookup'): + try: + partition_query = ( + CONVERSATIONS_SMART_MERGE_PARTITION_QUERY.build( + _collection(client, uid), + { + 'discarded': discarded, + 'source': source, + 'statuses': list(_ALL_STATUSES), + 'created_before': created_before, + 'client_device_id': client_device_id, + }, + field_filter_factory=FieldFilter, + ) + .order_by('created_at', direction=firestore.Query.DESCENDING) + .select(list(_PRECEDING_FIELDS)) + ) + return collect(partition_query) + except FailedPrecondition as error: + if 'index' not in str(error).lower(): + raise + # The lead reconciles the index before shipping, but a rolling + # deploy must remain correct while the index is still building. + record_fallback( + component='other', + from_mode='indexed_query', + to_mode='legacy_query', + reason='unavailable', + outcome='recovered', + log=logger, + ) + return collect(query) + return collect(query) def _epoch(value: Any) -> Optional[float]: @@ -219,6 +272,7 @@ def has_intervening_discarded( discarded=True, transaction=transaction, firestore_client=firestore_client, + client_device_id=donor.get('client_device_id'), ): if row.get('client_device_id') != donor.get('client_device_id'): continue diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 35b6b83399f..ee36e129b45 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -433,7 +433,9 @@ and an explicit empty literal renders as `''`. | `EPISODES_ENABLED` | Default on unless set to a non-on value; live episodes overlay (per-moment gpt-5-nano update, provisional items, /v1/episodes capability). Off reports enabled=false and apps keep today's per-conversation UI | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `FAIR_USE_ENABLED` | Enforce fair-use metering; set false to stop enforcement | backend | env | closed | — | true (backend-listen (chart)) | true (backend-listen (chart)) | — | keep | keep | — | unowned | | `FAIR_USE_KILL_SWITCH` | Emergency stop for fair-use enforcement | backend | env | inverted | — | false (backend-listen (chart)) | false (backend-listen (chart)) | — | keep | keep | — | unowned | +| `FSR_KG_CITATION_INDEX` | Default on; query native reverse citations for KG node pruning; off restores the node scan | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `FSR_MAINTENANCE_COORDINATION_ENABLED` | Default on in prod independently of read-lever rollback; explicit false restores per-replica maintenance | backend | env | open | — | — | — | — | keep | keep | — | dazheng | +| `FSR_SMART_MERGE_PARTITION` | Default on; partition the unbounded smart-merge discarded-barrier scan by explicit device; off restores the original scan | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `JOBSTORE_SHADOW_WRITES` | Default on unless 0/false/off/no; off stops the A2 Postgres shadow mirror (and its parity reads) while BATCH_JOBSTORE_MODE=shadow | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `LISTEN_FINALIZATION_BYOK_ABANDONMENT_ENABLED` | Abandon finalization after BYOK failure | backend | env | open | true | true (backend-listen (chart), gke/backend-listen) | true (backend-listen (chart), gke/backend-listen) | — | keep | keep | — | unowned | | `LISTEN_FINALIZATION_DURABLE_ATTEMPT_CAP_ENABLED` | Cap durable listen worker attempts while retaining transcript-only conversations | backend | env | open | — | — | true | — | keep | keep | — | dazheng | @@ -843,7 +845,9 @@ their code default (`fail` tells you which way a missing value resolves). - `FIRESTORE_CACHE_ENABLED` — Enable Firestore response cache (fail: closed) - `FIRST_SUMMARY_DEADLINE_SECONDS` — First required summary dispatch budget; default 120 seconds, bounded 60 to 120 (fail: closed) - `FIRST_SUMMARY_UID_ALLOWLIST` — Optional summary canary owner allowlist; empty admits only the positive percentage cohort when global gate is on (fail: closed) +- `FSR_KG_CITATION_INDEX` — Default on; query native reverse citations for KG node pruning; off restores the node scan (fail: open) - `FSR_MAINTENANCE_COORDINATION_ENABLED` — Default on in prod independently of read-lever rollback; explicit false restores per-replica maintenance (fail: open) +- `FSR_SMART_MERGE_PARTITION` — Default on; partition the unbounded smart-merge discarded-barrier scan by explicit device; off restores the original scan (fail: open) - `HELD_CONTENT_COHORT` — Fallback tokens uid:,pct:<0-100> or explicit all, used only after rollout_cohorts/held-content is confirmed absent; empty, unset, or malformed admits nobody (fail: closed) - `HELD_CONTENT_ENABLED` — Default-off at-cap choice API; additionally requires X-Omi-Held-Content: 1. Capture custody and recovery must ship before enablement. (fail: closed) - `JOBSTORE_SHADOW_WRITES` — Default on unless 0/false/off/no; off stops the A2 Postgres shadow mirror (and its parity reads) while BATCH_JOBSTORE_MODE=shadow (fail: open) diff --git a/backend/docs/runbooks/firestore-quick-reads.md b/backend/docs/runbooks/firestore-quick-reads.md new file mode 100644 index 00000000000..1a9e38e3bd7 --- /dev/null +++ b/backend/docs/runbooks/firestore-quick-reads.md @@ -0,0 +1,68 @@ +# Firestore structural query reads + +PR #23442 now retains two default-on structural optimizations. The public catalog +and request-cohort cache changes were dropped during the 2026-10-27 main merge: +main already has public-list single-flight/Redis reuse and deadline-bounded shared +cohort observations. This PR adds no request or cross-request cache lifetime. + +| Kill switch (`off` restores the prior path) | Scope | Origin | +| --- | --- | --- | +| `FSR_SMART_MERGE_PARTITION` | Unbounded discarded-barrier scan for explicit devices | `task:smart_merge_partition_lookup` | +| `FSR_KG_CITATION_INDEX` | Reverse-citation node lookup during KG pruning | `task:kg_citation_prune` | + +## Index and policy gates + +The lead must reconcile the generated manifest and verify +`conversations_smart_merge_device_preceding` READY before serving this source. +Use the existing schema approval/readiness path; this lane performs no applies, +access changes, production writes or deployment. A missing/building-index +`FailedPrecondition` falls back to the old query before returning rows; other +errors propagate. Both attempts retain the caller's transaction. + +The finite predecessor lookup keeps its global six-row window. Missing or empty +device IDs retain the old scan so absent fields remain visible. Discarded barriers +stay unbounded within the same device, including deleted barriers. Only values +fully represented by the index are eligible: device and citation identifiers +over 1,500 UTF-8 bytes retain the original scan. Firestore documents +[index-value truncation and inconsistent query results](https://docs.cloud.google.com/firestore/quotas) +above that limit; this is optimization eligibility, not a new input restriction. +Main's donor +rebasing, source/lineage checks, open-loop extraction and retirement paths remain. + +Firestore's native `memory_ids` array index is the reverse-citation writer: it +updates with existing node writes and covers historical data before serving. +There is no separate side collection or backfill. One query for at most 30 distinct memory IDs is materialized before mutations. +Larger retractions keep the original single scan, so overlapping citation chunks +cannot amplify reads. A missing index falls back before any write. Global surviving-node and orphan-edge scans remain: the +operation must still remove unrelated pre-existing dangling edges. No canonical +promotion, deletion/hold authority, transcription, speaker or Free-plan change. + +## Measurement and economics + +The 10-26 fleet caller pools are **16.749388M reads/day** for +`smart_merge:find_preceding_conversations` and **8.915616M/day** for +`knowledge_graph:prune_memory_citations_from_kg`. At $0.43–0.47/M and 30 days: + +- Smart merge's entire caller pool is $216–236/month. The removable fraction is + only explicit-device unbounded barriers multiplied by the off-device rows they + currently download. The saved aggregate does not identify either fraction. + **$0–236/month gross ceiling; no defensible point forecast.** The synthetic + 121-to-1 lookup is mechanism evidence, not the fleet traffic mix. +- KG removes only the first of two full node scans; the edge scan remains. Its + full caller pool is $115–126/month, but this change's structural ceiling is + **$0–63/month**, before empty-query floors and the unknown fraction of bulk retractions. The 206-to-106 + synthetic fixture corresponds to $56–61/month only if that graph shape holds + fleet-wide; it is not a measured forecast. + +The retained bundle's **$0–299/month** is an upper bound, before added composite +index storage and fallback overhead. Existing writes maintain the native indexes; +no new application document writes are introduced. New verified/realized savings +are **$0**. Earlier 22.5M/day predictions are retired, not booked. + +Compare old and new callers together: smart merge's inner collector is attributed +to `smart_merge:collect` (including the unchanged finite query), KG's new lookup to +`knowledge_graph:_citation_nodes`, and remaining sweeps to the old prune caller. +Track both new origins, legacy callers, fallback counts, query/RPC latency, +merge/barrier correctness and graph-cleanup equivalence. After lead serving, +compare matched windows against the 10-24/25 billing baseline and confirm net +reads/writes/storage at D+1 in the billing export. A renamed caller is not saving. diff --git a/backend/tests/support/firestore_query_driver_registry.py b/backend/tests/support/firestore_query_driver_registry.py index 1ae85eccdd9..0d782b2d2f6 100644 --- a/backend/tests/support/firestore_query_driver_registry.py +++ b/backend/tests/support/firestore_query_driver_registry.py @@ -1561,6 +1561,13 @@ def _seed_attached_frame_request(client, combo, trial): 'database.knowledge_graph.prune_memory_citations_from_kg', base={'uid': UID, 'memory_ids': ['mem-1', 'mem-2']} ) ) +_add( + CoveredByEntry( + 'database.knowledge_graph._citation_nodes', + covered_by=('database.knowledge_graph.prune_memory_citations_from_kg',), + reason='The prune driver exercises the native reverse-citation query before global orphan cleanup.', + ) +) _add( DriverEntry( 'database.knowledge_graph.upsert_knowledge_node', base={'uid': UID, 'node_data': {'label': 'shape-node'}} @@ -2245,7 +2252,11 @@ def _seed_notification_recipient_with_tokens(client, combo, trial): DriverEntry( 'database.smart_merge.find_preceding_conversations', base={'uid': UID, 'source': 'omi', 'created_before': T0}, - domains={'discarded': [False, True], 'transaction': [None, ref_transaction()]}, + domains={ + 'discarded': [False, True], + 'transaction': [None, ref_transaction()], + 'client_device_id': [None, 'pendant-1'], + }, neutrals={'limit': _LIMIT_OPT}, ) ) diff --git a/backend/tests/unit/test_conversation_smart_merge.py b/backend/tests/unit/test_conversation_smart_merge.py index ef7c6b80a92..a4f274b392a 100644 --- a/backend/tests/unit/test_conversation_smart_merge.py +++ b/backend/tests/unit/test_conversation_smart_merge.py @@ -131,7 +131,16 @@ def get(self, uid, cid, **_): return row def preceding( - self, uid, *, source, created_before, limit=None, discarded=False, transaction=None, firestore_client=None + self, + uid, + *, + source, + created_before, + limit=None, + discarded=False, + transaction=None, + firestore_client=None, + client_device_id=None, ): if transaction is not None: assert_read_allowed = getattr(transaction, '_assert_read_allowed', None) @@ -148,6 +157,7 @@ def preceding( and value.get('source') == source and value.get('status') in smart_merge_db._ALL_STATUSES and value['created_at'] < created_before + and (not client_device_id or limit is not None or value.get('client_device_id') == client_device_id) ] rows.sort(key=lambda row: row['created_at'], reverse=True) projected = [{k: v for k, v in row.items() if k != 'transcript_segments'} for row in rows] diff --git a/backend/tests/unit/test_conversation_smart_merge_intervening_discarded.py b/backend/tests/unit/test_conversation_smart_merge_intervening_discarded.py index 46fdc23ae84..9107a246bd0 100644 --- a/backend/tests/unit/test_conversation_smart_merge_intervening_discarded.py +++ b/backend/tests/unit/test_conversation_smart_merge_intervening_discarded.py @@ -387,6 +387,7 @@ def run_query(*, request, **kwargs): source = next((value for op, value in filters.get('source', []) if op == 'EQUAL'), None) statuses = next((value for op, value in filters.get('status', []) if op == 'IN'), None) created_before = next((value for op, value in filters.get('created_at', []) if op == 'LESS_THAN'), None) + device = next((value for op, value in filters.get('client_device_id', []) if op == 'EQUAL'), None) docs = [] for path, row in world.store.rows.items(): if path[:3] != ('users', UID, 'conversations'): @@ -395,6 +396,8 @@ def run_query(*, request, **kwargs): continue if source is not None and row.get('source') != source: continue + if device is not None and row.get('client_device_id') != device: + continue if statuses is not None and row.get('status') not in statuses: continue if created_before is not None and not row['created_at'] < created_before: diff --git a/backend/tests/unit/test_firestore_quick_partition.py b/backend/tests/unit/test_firestore_quick_partition.py new file mode 100644 index 00000000000..831ee2db72e --- /dev/null +++ b/backend/tests/unit/test_firestore_quick_partition.py @@ -0,0 +1,89 @@ +"""Exercise real SDK query serialization with offline RPCs and read counts.""" + +from datetime import timedelta + +import pytest +from google.api_core.exceptions import FailedPrecondition + +from database import smart_merge as storage +from tests.unit.test_conversation_smart_merge import T0, UID +from tests.unit.test_conversation_smart_merge_intervening_discarded import sdk_world, _field_filters + + +def _lookup(test, **kwargs): + return storage.find_preceding_conversations( + UID, + source='omi', + created_before=T0 + timedelta(days=10), + discarded=True, + firestore_client=test.client, + **kwargs, + ) + + +def test_partition_filters_unbounded_scan_without_hiding_deleted_barrier(sdk_world, monkeypatch): + test = sdk_world + for i in range(120): + test.world.add(f'other-{i}', 20 + i, 1, discarded=True, client_device_id='other-device') + test.world.add('barrier', 11, 2, discarded=True, deleted=True) + legacy = _lookup(test) + indexed = _lookup(test, client_device_id='pendant-1') + assert len(legacy) == 121 + assert [row['id'] for row in indexed] == ['barrier'] + assert indexed == [row for row in legacy if row.get('client_device_id') == 'pendant-1'] + assert ('client_device_id', 'EQUAL', 'pendant-1') in _field_filters(test.queries[-1], test.client) + + +@pytest.mark.parametrize('device', [None, '']) +def test_unknown_device_retains_rows_with_absent_field(sdk_world, device): + test = sdk_world + test.world.add('missing-device', 11, 2, discarded=True) + test.world.store.rows[('users', UID, 'conversations', 'missing-device')].pop('client_device_id') + assert [row['id'] for row in _lookup(test, client_device_id=device)] == ['missing-device'] + assert not any(field == 'client_device_id' for field, _, _ in _field_filters(test.queries[-1], test.client)) + + +def test_finite_window_and_kill_switch_preserve_old_query(sdk_world, monkeypatch): + test = sdk_world + test.world.add('other-newer', 18, 1, discarded=True, client_device_id='other') + test.world.add('matching-older', 11, 1, discarded=True) + assert [row['id'] for row in _lookup(test, limit=1, client_device_id='pendant-1')] == ['other-newer'] + monkeypatch.setenv('FSR_SMART_MERGE_PARTITION', 'off') + assert len(_lookup(test, client_device_id='pendant-1')) == 2 + + +def test_index_build_falls_back_before_returning_partial_rows(sdk_world, monkeypatch): + test = sdk_world + test.world.add('barrier', 11, 1, discarded=True) + actual = test.api.run_query.side_effect + + def indexed_failure(*, request, **kwargs): + if any(field == 'client_device_id' for field, _, _ in _field_filters(request, test.client)): + raise FailedPrecondition('query requires an index; index is building') + return actual(request=request, **kwargs) + + monkeypatch.setattr(test.api.run_query, 'side_effect', indexed_failure) + assert [row['id'] for row in _lookup(test, client_device_id='pendant-1')] == ['barrier'] + assert test.api.run_query.call_count == 2 + + +def test_unrelated_precondition_error_is_not_swallowed(sdk_world): + sdk_world.api.run_query.side_effect = FailedPrecondition('transaction is no longer active') + with pytest.raises(FailedPrecondition, match='transaction'): + _lookup(sdk_world, client_device_id='pendant-1') + + +@pytest.mark.parametrize('device', ['a' * 1501, '\u00e9' * 751]) +def test_oversized_device_retains_barriers_without_truncated_index_lookup(sdk_world, device): + test = sdk_world + test.world.add('barrier', 11, 2, discarded=True, deleted=True, client_device_id=device) + assert [row['id'] for row in _lookup(test, client_device_id=device)] == ['barrier'] + assert not any(field == 'client_device_id' for field, _, _ in _field_filters(test.queries[-1], test.client)) + + +@pytest.mark.parametrize('device', ['a' * 1500, '\u00e9' * 750]) +def test_device_at_index_byte_limit_can_use_partition_query(sdk_world, device): + test = sdk_world + test.world.add('barrier', 11, 2, discarded=True, client_device_id=device) + assert [row['id'] for row in _lookup(test, client_device_id=device)] == ['barrier'] + assert ('client_device_id', 'EQUAL', device) in _field_filters(test.queries[-1], test.client) diff --git a/backend/tests/unit/test_kg_prune_memory_citations.py b/backend/tests/unit/test_kg_prune_memory_citations.py index 47889db8943..74b8ea94e73 100644 --- a/backend/tests/unit/test_kg_prune_memory_citations.py +++ b/backend/tests/unit/test_kg_prune_memory_citations.py @@ -7,6 +7,7 @@ import sys import types from unittest.mock import MagicMock, patch +from google.api_core.exceptions import FailedPrecondition import pytest @@ -58,6 +59,9 @@ def __init__(self, db, path: str): def document(self, doc_id: str): return _DocRef(self._db, f"{self.path}/{doc_id}") + def where(self, *, filter): + return _Query(self._db, self.path, filter) + def stream(self): prefix = f"{self.path}/" depth = self.path.count("/") @@ -66,12 +70,35 @@ def stream(self): continue if path.count("/") != depth + 1: continue + self._db.reads += 1 yield _StreamDoc(self._db, path) +class _Query(_CollectionRef): + def __init__(self, db, path, filter): + super().__init__(db, path) + self.filter = filter + + def stream(self): + self._db.query_count += 1 + if self._db.fail_query_at == self._db.query_count: + raise FailedPrecondition('query requires an index') + assert self.filter.field_path == 'memory_ids' + assert self.filter.op_string == 'array_contains_any' + assert len(self.filter.value) <= 30 + for path, row in sorted(self._db.docs.items()): + if path.startswith(self.path + '/') and path.count('/') == self.path.count('/') + 1: + if set(row.get('memory_ids') or []).intersection(self.filter.value): + self._db.reads += 1 + yield _StreamDoc(self._db, path) + + class _KgFakeDb: def __init__(self, docs=None): self.docs = dict(docs or {}) + self.reads = 0 + self.query_count = 0 + self.fail_query_at = None def collection(self, name: str): return _CollectionRef(self, name) @@ -91,7 +118,9 @@ def kg_module(): cloud_mod.firestore = firestore_mod cloud_mod.firestore_v1 = firestore_v1_mod firestore_mod.Client = MagicMock() - firestore_v1_mod.FieldFilter = MagicMock() + firestore_v1_mod.FieldFilter = lambda field, op, value: types.SimpleNamespace( + field_path=field, op_string=op, value=value + ) def transactional(func): return func @@ -146,3 +175,93 @@ def test_prune_memory_citations_removes_dangling_edges_when_node_deleted(kg_modu assert node_path not in fake_db.docs assert edge_path not in fake_db.docs assert pruned >= 2 + + +def test_reverse_index_reduces_reads_and_preserves_global_orphan_cleanup(kg_module, monkeypatch): + kg_mod, indexed = kg_module + uid = 'synthetic-kg' + prefix = f'users/{uid}' + indexed.docs.update( + {f'{prefix}/knowledge_nodes/node-{i}': {'id': f'node-{i}', 'memory_ids': ['keep']} for i in range(100)} + ) + indexed.docs[f'{prefix}/knowledge_nodes/removed'] = {'id': 'removed', 'memory_ids': ['old']} + indexed.docs[f'{prefix}/knowledge_nodes/partial'] = {'id': 'partial', 'memory_ids': ['old', 'keep']} + indexed.docs[f'{prefix}/knowledge_edges/incident'] = { + 'source_id': 'removed', + 'target_id': 'node-1', + 'memory_ids': ['keep'], + } + indexed.docs[f'{prefix}/knowledge_edges/unrelated-orphan'] = { + 'source_id': 'never-existed', + 'target_id': 'node-1', + 'memory_ids': ['keep'], + } + indexed.docs[f'{prefix}/knowledge_edges/survives'] = { + 'source_id': 'node-1', + 'target_id': 'node-2', + 'memory_ids': ['keep', 'old'], + } + legacy = _KgFakeDb({path: dict(row) for path, row in indexed.docs.items()}) + assert kg_mod.prune_memory_citations_from_kg(uid, ['old'], db_client=indexed) == 5 + monkeypatch.setenv('FSR_KG_CITATION_INDEX', 'off') + assert kg_mod.prune_memory_citations_from_kg(uid, ['old'], db_client=legacy) == 5 + # updated_at is intentionally current in each call. + for store in (indexed, legacy): + for row in store.docs.values(): + row.pop('updated_at', None) + assert indexed.docs == legacy.docs + assert indexed.reads == 106 + assert legacy.reads == 206 + + +def test_bulk_citations_retain_legacy_scan_without_read_amplification(kg_module, monkeypatch): + kg_mod, store = kg_module + uid = 'synthetic-kg' + prefix = f'users/{uid}/knowledge_nodes' + path = f'{prefix}/node' + ids = [f'memory-{i:02d}' for i in range(65)] + store.docs[path] = {'id': 'node', 'memory_ids': ids + ['keep']} + store.docs[f'{prefix}/unrelated'] = {'id': 'unrelated', 'memory_ids': ['keep']} + legacy = _KgFakeDb({path: dict(row) for path, row in store.docs.items()}) + assert kg_mod.prune_memory_citations_from_kg(uid, ids, db_client=store) == 1 + monkeypatch.setenv('FSR_KG_CITATION_INDEX', 'off') + assert kg_mod.prune_memory_citations_from_kg(uid, ids, db_client=legacy) == 1 + for db in (store, legacy): + for row in db.docs.values(): + row.pop('updated_at', None) + assert store.docs == legacy.docs + assert store.query_count == 0 + assert store.reads == legacy.reads == 4 + + +def test_thirty_distinct_citations_use_one_query_even_with_duplicate_inputs(kg_module): + kg_mod, store = kg_module + uid = 'synthetic-kg' + path = f'users/{uid}/knowledge_nodes/node' + ids = [f'memory-{i:02d}' for i in range(30)] + store.docs[path] = {'id': 'node', 'memory_ids': ids + ['keep']} + assert kg_mod.prune_memory_citations_from_kg(uid, ids + ids, db_client=store) == 1 + assert store.query_count == 1 + assert store.docs[path]['memory_ids'] == ['keep'] + + +def test_missing_native_index_falls_back_before_writes(kg_module): + kg_mod, store = kg_module + uid = 'synthetic-kg' + path = f'users/{uid}/knowledge_nodes/node' + store.docs[path] = {'id': 'node', 'memory_ids': ['old']} + store.fail_query_at = 1 + assert kg_mod.prune_memory_citations_from_kg(uid, ['old'], db_client=store) == 1 + assert store.query_count == 1 + assert path not in store.docs + + +@pytest.mark.parametrize('memory_id', ['a' * 1501, '\u00e9' * 751]) +def test_oversized_citation_id_retains_legacy_pruning_without_index_truncation(kg_module, memory_id): + kg_mod, store = kg_module + uid = 'synthetic-kg' + path = f'users/{uid}/knowledge_nodes/node' + store.docs[path] = {'id': 'node', 'memory_ids': [memory_id, 'keep']} + assert kg_mod.prune_memory_citations_from_kg(uid, [memory_id], db_client=store) == 1 + assert store.query_count == 0 + assert store.docs[path]['memory_ids'] == ['keep'] diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index a118801f341..dbe0dc48dd0 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -1,5 +1,7 @@ flags: - {"key":"FSR_MAINTENANCE_COORDINATION_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on in prod independently of read-lever rollback; explicit false restores per-replica maintenance","fail":"open","owner":"dazheng","created":"2026-10-27","terminal":"keep","decision":"keep","notes":"Retains existing Redis lease cadence, ownership fencing, bounded takeover, publisher eligibility and outage fail-open. Non-production including stage-prod staging keeps the legacy singleton_loops cohort unless explicitly overridden. No scan policy or job claims change. Lead owns deployment; verify publisher availability first. See backend/docs/runbooks/firestore-spend-guard.md."} + - {"key":"FSR_SMART_MERGE_PARTITION","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; partition the unbounded smart-merge discarded-barrier scan by explicit device; off restores the original scan","fail":"open","owner":"dazheng","created":"2026-10-23","terminal":"keep","decision":"keep","notes":"Uses the registered device composite index with missing-index fallback. Unknown devices, the global six-row predecessor window, all statuses and transactional barrier checks retain their original semantics. Reconcile the index before shipping. Attribution task:smart_merge_partition_lookup."} + - {"key":"FSR_KG_CITATION_INDEX","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; query native reverse citations for KG node pruning; off restores the node scan","fail":"open","owner":"dazheng","created":"2026-10-23","terminal":"keep","decision":"keep","notes":"Firestore maintains native array and adjacency indexes for existing and new writes; missing index falls back before any mutations. Global surviving-node and dangling-edge sweeps remain. Attribution task:kg_citation_prune."} - {"key":"BATCH_PIPELINE_LANGUAGE_INHERITANCE_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on unless false; batch pieces inherit a known bound conversation language, else the user language, before per-piece LID","fail":"open","owner":"dazheng","created":"2026-10-27","terminal":"keep","decision":"keep","notes":"Known means a recognized STT language code, excluding multi/auto/und/unknown. Short ambient-dominated WAL clips never override known context. Only when both are unknown does the existing self/vendor LID run. Legacy eligibility, durable custody and unsupported-language worker barriers stay in force. false restores former single-language-preference input. New intakes only; rollback does not rewrite frozen work."} - {"key":"BATCH_ADMISSION_LANGUAGE_EXCLUSION_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on unless false; exclude new unsupported-language batch uploads and non-single-language users without a supported conversation language in every legacy mode","fail":"open","owner":"dazheng","created":"2026-10-27","terminal":"keep","decision":"keep","notes":"Lead ruling 6a783e: legacy ON dispatches excluded uploads through legacy; draining/dark retains the existing retry fence, never a dark bypass. Before legacy drain, multilingual and unsupported speech needs a batch provider route (Soniox). Runs before re-home, decode, LID or batch custody; admitted batch receipts and durable conversation owners retain their route. Supported conversation language becomes explicit LID input. False restores the former admission and input selection exactly. Log event=batch_admission_language with bounded reason; existing language-route and shared fallback metrics. Compatible with language inheritance; unsupported inherited codes are excluded."} - {"key":"STT_IDLE_REOPEN_COST_RESELECT_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default off after the 2026-10-27 first-text regression; explicit true re-enables managed cost reselection at Soniox idle onset","fail":"open","owner":"dazheng","created":"2026-10-27","terminal":"keep","decision":"keep","notes":"Rolled default-off 2026-10-27: default-on idle reselect retired the serving leg at every idle onset and rebuilt through the full chain, saturating the window cap and terminating sessions (listenlat RCA). True/1/on restores managed reselection with current stages, capability and account admission; false/0/off (default) keeps the stored Soniox factory and the in-place reopen. Planned retirements retain onset PCM and use ordinary capture replay/adoption without spending a failure episode or benching Soniox. Wire and connected seconds carry bounded origin initial|idle_reopened|replacement. Lead owns bundled CR/listen deployment. See backend/docs/listen_pusher_pipeline.mdx."} diff --git a/firestore.indexes.json b/firestore.indexes.json index 0852c7017bc..a338ab4d726 100644 --- a/firestore.indexes.json +++ b/firestore.indexes.json @@ -4687,6 +4687,36 @@ "order": "ASCENDING" } ] + }, + { + "collectionGroup": "conversations", + "queryScope": "COLLECTION", + "fields": [ + { + "fieldPath": "discarded", + "order": "ASCENDING" + }, + { + "fieldPath": "source", + "order": "ASCENDING" + }, + { + "fieldPath": "status", + "order": "ASCENDING" + }, + { + "fieldPath": "client_device_id", + "order": "ASCENDING" + }, + { + "fieldPath": "created_at", + "order": "DESCENDING" + }, + { + "fieldPath": "__name__", + "order": "DESCENDING" + } + ] } ], "fieldOverrides": [