diff --git a/apps/api/app/api/v1/routes/retrieval.py b/apps/api/app/api/v1/routes/retrieval.py index 3c9202df..c7d79595 100644 --- a/apps/api/app/api/v1/routes/retrieval.py +++ b/apps/api/app/api/v1/routes/retrieval.py @@ -7,7 +7,7 @@ from app.api.dependencies.current_user import with_current_user from app.services.rate_limit.data_structures import CurrentUser from fastapi import APIRouter, Depends -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, model_validator from sqlalchemy.ext.asyncio import AsyncSession from shared.core.database import get_db @@ -60,10 +60,17 @@ class RetrievalQueryRequest(BaseModel): ) channels: list[str] = Field( default_factory=list, - description="Channels to run (empty=all). Options: path, content, term", + description=( + "Deprecated and unsupported by the persisted map-unit route. " + "Leave empty; explicit channel selection is rejected." + ), ) channel_weights: dict[str, float] = Field( - default_factory=dict, description="Per-channel weight overrides" + default_factory=dict, + description=( + "Deprecated and unsupported by the persisted map-unit route. " + "Leave empty; explicit overrides are rejected." + ), ) rerank: bool = Field(False, description="Enable LLM reranking after RRF fusion") threshold: float = Field(0.0, ge=0.0, description="Minimum RRF score threshold") @@ -112,6 +119,20 @@ def validate_chunk_types(cls, v: list[str] | None) -> list[str] | None: def normalize_namespace(cls, namespace: str | None) -> str: return normalize_retrieval_namespace(namespace) + @model_validator(mode="after") + def reject_unsupported_channel_controls(self) -> "RetrievalQueryRequest": + if self.channels: + raise ValueError( + "channels is deprecated and unsupported; omit it and use the " + "persisted path/content map-unit scorer" + ) + if self.channel_weights: + raise ValueError( + "channel_weights is deprecated and unsupported; omit it and use " + "the persisted path/content map-unit scorer" + ) + return self + class RetrievalQueryResponse(BaseModel): namespace: str diff --git a/apps/api/tests/contract/test_retrieval_contract.py b/apps/api/tests/contract/test_retrieval_contract.py index 78dffed9..90dc7e4f 100644 --- a/apps/api/tests/contract/test_retrieval_contract.py +++ b/apps/api/tests/contract/test_retrieval_contract.py @@ -306,6 +306,28 @@ async def test_should_return_request_validation_failure_for_an_invalid_channel( assert "Invalid channel" in cast(str, violations[0]["description"]) +async def test_should_reject_legacy_channel_controls( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + response = await api_client.post( + "/api/v1/retrieval/query", + json={ + "namespace": "default", + "query": "alpha", + "channels": ["content"], + }, + ) + + assert response.status_code == 400 + response_json = cast(dict[str, object], response.json()) + error = cast(dict[str, object], response_json["error"]) + assert error["code"] == "INVALID_ARGUMENT" + assert "deprecated and unsupported" in str(error) + + async def test_should_exclude_matching_document_ids_from_the_response( developer_api_client_factory: Callable[ [], AbstractAsyncContextManager[AsyncClient] diff --git a/apps/api/tests/contract/test_retrieval_lazy_snapshot_quality_contract.py b/apps/api/tests/contract/test_retrieval_lazy_snapshot_quality_contract.py index b80fb8ae..c796e2a1 100644 --- a/apps/api/tests/contract/test_retrieval_lazy_snapshot_quality_contract.py +++ b/apps/api/tests/contract/test_retrieval_lazy_snapshot_quality_contract.py @@ -165,10 +165,7 @@ def test_lazy_provider_preserves_score_units_and_scores() -> None: lazy_scores = compute_corpus_map_and_unit_scores( lazy, doc_ids=["doc"], query="alpha retrieval" ) - assert eager_scores[1] == {} - assert lazy_scores[1] == {} - assert all(score == 0.0 for score in eager_scores[0].values()) - assert all(score == 0.0 for score in lazy_scores[0].values()) + assert eager_scores == lazy_scores lazy_provider = lazy._provider self_units = getattr(lazy_provider, "self_units") @@ -245,7 +242,7 @@ def build_stats(search_field: str) -> PersistedBm25Stats: assert scored["beta evidence"]["unit-b"] > scored["beta evidence"]["unit-a"] -def test_missing_index_does_not_read_chunk_payloads() -> None: +def test_missing_index_falls_back_to_legacy_payload_scoring() -> None: _eager, lazy, store = _providers() queries: list[str] = ["alpha retrieval", "supporting image"] store.section_loads = 0 @@ -256,8 +253,8 @@ def test_missing_index_does_not_read_chunk_payloads() -> None: ) assert set(actual) == set(queries) - assert all(unit_scores == {} for _map_scores, unit_scores in actual.values()) - assert store.section_loads == 0 + assert any(unit_scores for _map_scores, unit_scores in actual.values()) + assert store.section_loads > 0 def test_missing_index_is_empty_across_documents() -> None: @@ -276,8 +273,8 @@ def test_missing_index_is_empty_across_documents() -> None: ) assert actual == expected - assert all(unit_scores == {} for _map_scores, unit_scores in actual.values()) - assert store.section_loads == 0 + assert any(unit_scores for _map_scores, unit_scores in actual.values()) + assert store.section_loads > 0 def test_native_chunk_store_strips_async_driver_from_database_url( diff --git a/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py b/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py index 5c653722..e95efd23 100644 --- a/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py +++ b/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py @@ -276,8 +276,8 @@ def reject_payload_read( assert any(score > 0.0 for score in actual_scores[1].values()) assert select_map_highlights(actual_scores[1], k=3) - assert fallback_scores[1] == {} - assert all(score == 0.0 for score in fallback_scores[0].values()) + assert any(score > 0.0 for score in fallback_scores[1].values()) + assert any(score > 0.0 for score in fallback_scores[0].values()) async def test_lazy_snapshot_defers_selected_asset_reference_metadata( @@ -476,8 +476,8 @@ def test_incomplete_index_returns_empty_scores() -> None: lazy, doc_ids=["doc-a", "doc-b"], query="alpha beta" ) - assert actual[1] == {} - assert expected[1] == {} + assert set(actual[1]) == {"leaf-a", "leaf-b"} + assert set(expected[1]) == {"leaf-a", "leaf-b"} assert all(score == 0.0 for score in actual[0].values()) assert store.persisted_loads == 1 diff --git a/packages/shared-python/shared/services/retrieval/cache_service.py b/packages/shared-python/shared/services/retrieval/cache_service.py index da76068d..c327eb46 100644 --- a/packages/shared-python/shared/services/retrieval/cache_service.py +++ b/packages/shared-python/shared/services/retrieval/cache_service.py @@ -10,6 +10,7 @@ _RETRIEVAL_CACHE_TTL_SECONDS = 300 _VERSION_FALLBACK = 0 +_INDEX_READINESS_TTL_SECONDS = 60 def _namespace_version_key(*, user_id: str, namespace: str) -> str: @@ -17,6 +18,37 @@ def _namespace_version_key(*, user_id: str, namespace: str) -> str: return f"retrieval:version:{user_id}:{namespace}" +def _namespace_index_readiness_key(*, user_id: str, namespace: str) -> str: + namespace = normalize_retrieval_namespace(namespace) + return f"retrieval:index-readiness:{user_id}:{namespace}" + + +async def record_retrieval_index_readiness( + *, + user_id: str, + namespace: str, + ready: bool, + expected_revisions: int, + indexed_revisions: int, +) -> None: + """Publish a short-lived index readiness signal for operators and callers. + + Redis is deliberately only a status cache. PostgreSQL generations and + serving rows remain the source of truth, and retrieval must continue to + work if Redis is unavailable. + """ + redis_service = RedisServiceFactory.get_service() + await redis_service.set( + _namespace_index_readiness_key(user_id=user_id, namespace=namespace), + { + "ready": bool(ready), + "expected_revisions": int(expected_revisions), + "indexed_revisions": int(indexed_revisions), + }, + ex=_INDEX_READINESS_TTL_SECONDS, + ) + + def _normalize_exclude_sections(exclude_sections: list[dict[str, str]]) -> list[str]: normalized: list[str] = [] for item in exclude_sections: diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py b/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py index 716ff1ec..dfbad871 100644 --- a/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py +++ b/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py @@ -300,6 +300,7 @@ def load_persisted_score_corpus( comes from ``document_map_unit_indexes`` (written at index time). """ from shared.services.retrieval.nav.persisted_score_load import ( + average_idf_from_namespace_stats, build_channel_bm25_stats, combine_average_idf, ) @@ -359,12 +360,51 @@ def load_persisted_score_corpus( revision_key ] else: - average_idf_path = combine_average_idf( - [(float(row[4] or 0.0), int(row[3] or 0)) for row in index_rows] - ) - average_idf_content = combine_average_idf( - [(float(row[5] or 0.0), int(row[3] or 0)) for row in index_rows] - ) + total_unit_count = sum(int(row[3] or 0) for row in index_rows) + try: + cur.execute( + "SELECT tokens.channel, tokens.token, " + "COUNT(DISTINCT tokens.map_unit_id) " + "FROM document_map_unit_tokens AS tokens " + "JOIN document_map_units AS units ON units.id = tokens.map_unit_id " + f"JOIN (VALUES {values_sql}) AS revisions(document_id, job_result_id) " + "ON units.document_id = revisions.document_id " + "AND units.job_result_id = revisions.job_result_id " + "WHERE tokens.channel = ANY(%s) " + "GROUP BY tokens.channel, tokens.token", + [*revision_params, ["path", "content"]], + ) + namespace_token_dfs: dict[str, list[int]] = { + "path": [], + "content": [], + } + for channel, _token, document_frequency in cur.fetchall(): + if str(channel) in namespace_token_dfs: + namespace_token_dfs[str(channel)].append( + int(document_frequency) + ) + average_idf_path = average_idf_from_namespace_stats( + unit_count=total_unit_count, + token_document_frequencies=namespace_token_dfs["path"], + ) + average_idf_content = average_idf_from_namespace_stats( + unit_count=total_unit_count, + token_document_frequencies=namespace_token_dfs["content"], + ) + except Exception as exc: + _logger.warning( + "exact namespace IDF load failed; using revision averages: %s", + exc, + ) + average_idf_path = combine_average_idf( + [(float(row[4] or 0.0), int(row[3] or 0)) for row in index_rows] + ) + average_idf_content = combine_average_idf( + [ + (float(row[5] or 0.0), int(row[3] or 0)) + for row in index_rows + ] + ) self._score_average_idf_cache[revision_key] = ( average_idf_path, average_idf_content, diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py b/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py index c646070a..2458661b 100644 --- a/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py +++ b/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py @@ -8,12 +8,89 @@ build_content_search_text, build_path_search_text, build_term_search_text, + PersistedScoreCorpus, + PersistedScoreUnit, score_persisted_corpus_many, ) +from .persisted_score_load import ( + average_idf_from_unit_dfs, + build_channel_bm25_stats, +) _logger = logging.getLogger(__name__) +def _build_legacy_score_corpus(ts: Any, doc_ids: Sequence[str]) -> PersistedScoreCorpus: + """Build the retired in-memory scorer input when persisted indexes are absent.""" + raw_units: List[dict] = [] + for doc_id in doc_ids: + raw_units.extend(build_score_units(ts, doc_id)) + frequencies: Dict[Tuple[str, str], Dict[str, int]] = {} + unit_rows: List[dict] = [] + path_dfs: Dict[str, int] = {} + content_dfs: Dict[str, int] = {} + for unit in raw_units: + unit_id = str(unit.get("chunk_id") or "").strip() + if not unit_id: + continue + path_tokens = str(unit.get("path_search_text") or "").split() + content_tokens = str(unit.get("content_search_text") or "").split() + path_freq: Dict[str, int] = {} + content_freq: Dict[str, int] = {} + for token in path_tokens: + path_freq[token] = path_freq.get(token, 0) + 1 + for token in content_tokens: + content_freq[token] = content_freq.get(token, 0) + 1 + frequencies[(unit_id, "path")] = path_freq + frequencies[(unit_id, "content")] = content_freq + for token in path_freq: + path_dfs[token] = path_dfs.get(token, 0) + 1 + for token in content_freq: + content_dfs[token] = content_dfs.get(token, 0) + 1 + unit_rows.append( + { + "unit_id": unit_id, + "path_length": len(path_tokens), + "content_length": len(content_tokens), + } + ) + unit_count = len(unit_rows) + return PersistedScoreCorpus( + units=[ + PersistedScoreUnit( + unit_id=str(row["unit_id"]), + path_length=int(row["path_length"]), + content_length=int(row["content_length"]), + path_frequencies=frequencies[(str(row["unit_id"]), "path")], + content_frequencies=frequencies[(str(row["unit_id"]), "content")], + ) + for row in unit_rows + ], + path_stats=build_channel_bm25_stats( + unit_rows=unit_rows, + map_unit_id_field="unit_id", + length_field="path_length", + channel="path", + query_tokens=list(path_dfs), + frequencies=frequencies, + average_idf=average_idf_from_unit_dfs( + unit_count=unit_count, token_document_frequency=path_dfs + ), + ), + content_stats=build_channel_bm25_stats( + unit_rows=unit_rows, + map_unit_id_field="unit_id", + length_field="content_length", + channel="content", + query_tokens=list(content_dfs), + frequencies=frequencies, + average_idf=average_idf_from_unit_dfs( + unit_count=unit_count, token_document_frequency=content_dfs + ), + ), + ) + + def _children_ids(ts: Any, section_id: str, doc_id: str) -> List[str]: children_fn = getattr(ts, "_children_for_section_path", None) if not callable(children_fn): @@ -346,6 +423,13 @@ def compute_corpus_map_and_unit_scores_many( time.perf_counter() - loader_started, persisted_corpus is not None, ) + if persisted_corpus is None: + _logger.warning( + "retrieval map index unavailable; using bounded legacy in-memory scorer " + "documents=%d", + len(valid_doc_ids), + ) + persisted_corpus = _build_legacy_score_corpus(ts, valid_doc_ids) score_started = time.perf_counter() unit_scores_by_query = ( score_persisted_corpus_many(persisted_corpus, unique_queries) diff --git a/packages/shared-python/shared/services/retrieval/nav/persisted_score_load.py b/packages/shared-python/shared/services/retrieval/nav/persisted_score_load.py index 9ce10c2e..7f1eb202 100644 --- a/packages/shared-python/shared/services/retrieval/nav/persisted_score_load.py +++ b/packages/shared-python/shared/services/retrieval/nav/persisted_score_load.py @@ -38,6 +38,29 @@ def combine_average_idf(parts: Sequence[tuple[float, int]]) -> float: ) +def average_idf_from_namespace_stats( + *, + unit_count: int, + token_document_frequencies: Sequence[int], +) -> float: + """Compute the exact namespace-level average IDF used by rank_bm25. + + Namespace token statistics already contain one document frequency per + token. Computing the mean from those rows avoids the incorrect + per-revision-average approximation when a namespace contains revisions + with different token distributions. + """ + if unit_count <= 0: + return 0.0 + idfs = [ + math.log(unit_count - int(frequency) + 0.5) + - math.log(int(frequency) + 0.5) + for frequency in token_document_frequencies + if 0 < int(frequency) <= unit_count + ] + return sum(idfs) / len(idfs) if idfs else 0.0 + + def build_channel_bm25_stats( *, unit_rows: Sequence[Mapping[str, Any]], diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py index 58fa98dd..84b3b2d2 100644 --- a/packages/shared-python/shared/services/retrieval/publication_service.py +++ b/packages/shared-python/shared/services/retrieval/publication_service.py @@ -200,6 +200,14 @@ def _publish_document_state_for_job( namespace=str(existing_namespace), document_id=document.document_id, ) + # A namespace move mutates both namespace snapshots. Advance the + # old namespace generation as well so request-scoped/process-local + # snapshot caches cannot reuse the pre-move generation. + advance_namespace_generation( + db, + user_id=scope.user_id, + namespace=str(existing_namespace), + ) advance_namespace_generation( db, user_id=scope.user_id, diff --git a/packages/shared-python/shared/services/retrieval/search/map_unit_discovery.py b/packages/shared-python/shared/services/retrieval/search/map_unit_discovery.py index 34d3d74f..97f5d11e 100644 --- a/packages/shared-python/shared/services/retrieval/search/map_unit_discovery.py +++ b/packages/shared-python/shared/services/retrieval/search/map_unit_discovery.py @@ -37,9 +37,12 @@ tokenize_query_for_ranker, ) from shared.services.retrieval.nav.persisted_score_load import ( + average_idf_from_namespace_stats, build_channel_bm25_stats, combine_average_idf, ) +from shared.services.retrieval.serving_manifest import decode_serving_manifest +from shared.services.retrieval.cache_service import record_retrieval_index_readiness from shared.services.retrieval.search.scoring import normalize_row_scores from shared.services.retrieval.search.section_filters import is_excluded_section from shared.services.retrieval.settings import ASSET_CHUNK_TYPES @@ -81,6 +84,80 @@ class DiscoveryResult: error: str | None = None +async def _load_exact_namespace_average_idf( + db: AsyncSession, + *, + user_id: str, + namespace: str, +) -> tuple[float, float] | None: + """Load exact namespace IDF floors from the published aggregate tables.""" + generation_row = ( + await db.execute( + text( + "SELECT generation FROM retrieval_namespace_generations " + "WHERE user_id = :user_id AND namespace = :namespace" + ), + {"user_id": user_id, "namespace": namespace}, + ) + ).first() + if generation_row is None: + return None + generation = int(generation_row[0]) + stat_row = ( + await db.execute( + text( + "SELECT payload_zlib, checksum, format_version " + "FROM retrieval_namespace_stats " + "WHERE user_id = :user_id AND namespace = :namespace " + "AND generation = :generation" + ), + { + "user_id": user_id, + "namespace": namespace, + "generation": generation, + }, + ) + ).first() + if stat_row is None: + return None + payload = decode_serving_manifest( + stat_row[0], checksum=str(stat_row[1]), format_version=int(stat_row[2]) + ) + unit_count = int(payload.get("unit_count") or 0) + if unit_count <= 0: + return None + token_rows = ( + await db.execute( + text( + "SELECT channel, document_frequency " + "FROM retrieval_namespace_token_stats " + "WHERE user_id = :user_id AND namespace = :namespace " + "AND generation = :generation AND channel = ANY(:channels)" + ), + { + "user_id": user_id, + "namespace": namespace, + "generation": generation, + "channels": ["path", "content"], + }, + ) + ).all() + frequencies: dict[str, list[int]] = {"path": [], "content": []} + for channel, frequency in token_rows: + if str(channel) in frequencies: + frequencies[str(channel)].append(int(frequency)) + return ( + average_idf_from_namespace_stats( + unit_count=unit_count, + token_document_frequencies=frequencies["path"], + ), + average_idf_from_namespace_stats( + unit_count=unit_count, + token_document_frequencies=frequencies["content"], + ), + ) + + def _build_revision_scope( revision_pins: Mapping[str, str] | None, ) -> tuple[str, str, dict[str, Any]]: @@ -251,15 +328,84 @@ async def map_unit_discovery( (float(path_idf or 0.0), float(content_idf or 0.0), int(unit_count or 0)) for path_idf, content_idf, unit_count in index_result.all() ] - average_idf_path = combine_average_idf( - [(path_idf, unit_count) for path_idf, _content_idf, unit_count in index_parts] - ) - average_idf_content = combine_average_idf( - [ - (content_idf, unit_count) - for _path_idf, content_idf, unit_count in index_parts - ] + expected_revisions = { + (str(row["document_id"]), str(row["job_result_id"])) for row in unit_rows + } + unfiltered_scope = not any( + ( + chunk_types, + signal_paths, + exclude_sections, + exclude_document_ids, + ) ) + index_unit_count_mismatch = unfiltered_scope and sum( + unit_count for _path_idf, _content_idf, unit_count in index_parts + ) != len(unit_rows) + if len(index_parts) != len(expected_revisions) or index_unit_count_mismatch: + try: + await record_retrieval_index_readiness( + user_id=user_id, + namespace=namespace, + ready=False, + expected_revisions=len(expected_revisions), + indexed_revisions=len(index_parts), + ) + except Exception as exc: + logger.warning("retrieval index readiness publish failed: %s", exc) + logger.warning( + "retrieval map index incomplete user_id=%s namespace=%s " + "expected_revisions=%d indexed_revisions=%d fallback=legacy_fts", + user_id, + namespace, + len(expected_revisions), + len(index_parts), + ) + return await _legacy_chunk_discovery( + db, + user_id=user_id, + namespace=namespace, + query=query, + top_k=top_k, + exclude_document_ids=exclude_document_ids, + exclude_sections=exclude_sections, + chunk_types=chunk_types, + signal_paths=signal_paths or [], + filter_mode=filter_mode, + revision_pins=revision_pins, + ) + try: + await record_retrieval_index_readiness( + user_id=user_id, + namespace=namespace, + ready=True, + expected_revisions=len(expected_revisions), + indexed_revisions=len(index_parts), + ) + except Exception as exc: + logger.warning("retrieval index readiness publish failed: %s", exc) + exact_namespace_idf = None + if revision_pins is None: + exact_namespace_idf = await _load_exact_namespace_average_idf( + db, + user_id=user_id, + namespace=namespace, + ) + if exact_namespace_idf is not None: + average_idf_path, average_idf_content = exact_namespace_idf + else: + average_idf_path = combine_average_idf( + [ + (path_idf, unit_count) + for path_idf, _content_idf, unit_count in index_parts + ] + ) + average_idf_content = combine_average_idf( + [ + (content_idf, unit_count) + for _path_idf, content_idf, unit_count in index_parts + ] + ) path_stats = build_channel_bm25_stats( unit_rows=unit_rows, @@ -337,6 +483,103 @@ async def map_unit_discovery( ) +async def _legacy_chunk_discovery( + db: AsyncSession, + *, + user_id: str, + namespace: str, + query: str, + top_k: int, + exclude_document_ids: list[str], + exclude_sections: list[dict[str, str]], + chunk_types: set[str] | None, + signal_paths: list[str], + filter_mode: str, + revision_pins: Mapping[str, str] | None, +) -> DiscoveryResult: + """Bounded lexical fallback used while a serving index is incomplete.""" + clauses = [ + "d.user_id = :user_id", + "d.namespace = :namespace", + "d.status = 'active'", + ] + params: dict[str, Any] = { + "user_id": user_id, + "namespace": namespace, + "query": query, + "limit": max(1, int(top_k)), + } + if revision_pins is None: + clauses.append("d.current_job_result_id = dc.job_result_id") + else: + pairs = [ + (str(document_id), str(job_result_id)) + for document_id, job_result_id in revision_pins.items() + ] + if not pairs: + return DiscoveryResult(status="discovery_done", payload={"fused_rows": []}) + placeholders = [] + for index, (document_id, job_result_id) in enumerate(pairs): + document_key = f"_legacy_doc_{index}" + revision_key = f"_legacy_revision_{index}" + placeholders.append(f"(:{document_key}, :{revision_key})") + params[document_key] = document_id + params[revision_key] = job_result_id + clauses.append(f"(dc.document_id, dc.job_result_id) IN ({', '.join(placeholders)})") + if exclude_document_ids: + clauses.append("d.document_id <> ALL(:excluded_doc_ids)") + params["excluded_doc_ids"] = exclude_document_ids + if chunk_types: + type_keys = [] + for index, chunk_type in enumerate(sorted(chunk_types)): + key = f"_legacy_type_{index}" + type_keys.append(f":{key}") + params[key] = chunk_type + clauses.append(f"LOWER(dc.chunk_type) IN ({', '.join(type_keys)})") + if signal_paths: + signal_parts = [] + for index, signal in enumerate(signal_paths): + key = f"_legacy_signal_{index}" + signal_parts.append("LOWER(COALESCE(ds.section_path, '')) LIKE :" + key) + params[key] = f"%{signal.lower()}%" + combined = " OR ".join(signal_parts) + clauses.append(f"({combined})" if filter_mode == "keep" else f"NOT ({combined})") + for index, item in enumerate(exclude_sections): + document_id = str(item.get("document_id") or "").strip() + section_path = str(item.get("section_path") or "").strip() + if not document_id or not section_path: + continue + doc_key = f"_legacy_exclude_doc_{index}" + path_key = f"_legacy_exclude_path_{index}" + params[doc_key] = document_id + params[path_key] = section_path + clauses.append( + "NOT (dc.document_id = :" + doc_key + " AND (" + "COALESCE(ds.section_path, '') = :" + path_key + " OR " + "POSITION(:" + path_key + " || ' / ' IN COALESCE(ds.section_path, '')) = 1))" + ) + where_sql = " AND ".join(clauses) + statement = text( + "SELECT dc.chunk_id, dc.document_id, dc.section_id, dc.chunk_type, " + "dc.content, dc.source_chunk_path, dc.file_path, dc.chunk_metadata, " + "dc.job_result_id, dc.sort_order, ds.section_path, d.source_file_name, " + "jr.job_id, GREATEST(ts_rank_cd(dc.path_search_tsv, plainto_tsquery('simple', :query)), " + "2 * ts_rank_cd(dc.content_search_tsv, plainto_tsquery('simple', :query))) AS score " + "FROM document_chunks dc JOIN documents d ON d.document_id = dc.document_id " + "LEFT JOIN document_sections ds ON ds.section_id = dc.section_id " + "LEFT JOIN job_results jr ON jr.id = dc.job_result_id " + f"WHERE {where_sql} AND (dc.path_search_tsv @@ plainto_tsquery('simple', :query) " + "OR dc.content_search_tsv @@ plainto_tsquery('simple', :query) " + "OR LOWER(COALESCE(dc.term_search_text, '')) LIKE LOWER(:term_query)) " + "ORDER BY score DESC, dc.sort_order, dc.chunk_id LIMIT :limit" + ) + params["term_query"] = f"%{query}%" + rows = [dict(row._mapping) for row in (await db.execute(statement, params)).all()] + if rows: + normalize_row_scores(rows, source_field="score", target_field="discovery_score", default=0.5) + return DiscoveryResult(status="discovery_done", payload={"fused_rows": rows}) + + def _as_metadata_dict(value: object) -> dict[str, Any]: if isinstance(value, dict): return value diff --git a/packages/shared-python/shared/services/retrieval/serving_manifest.py b/packages/shared-python/shared/services/retrieval/serving_manifest.py index 88811320..467d55c9 100644 --- a/packages/shared-python/shared/services/retrieval/serving_manifest.py +++ b/packages/shared-python/shared/services/retrieval/serving_manifest.py @@ -4,6 +4,8 @@ import hashlib import json +import logging +import time import zlib from typing import Any @@ -26,6 +28,7 @@ from shared.services.retrieval.publication_models import DocumentPublicationScope SERVING_MANIFEST_FORMAT_VERSION = 1 +_logger = logging.getLogger(__name__) def build_revision_serving_payload( @@ -230,6 +233,7 @@ def rebuild_namespace_serving_statistics( Callers hold the namespace generation lock. The aggregate is prepared for the generation that the caller will publish next. """ + started = time.perf_counter() generation = db.execute( select(RetrievalNamespaceGeneration) .where(RetrievalNamespaceGeneration.user_id == user_id) @@ -328,6 +332,17 @@ def rebuild_namespace_serving_statistics( ] ) db.flush() + _logger.info( + "retrieval namespace statistics rebuilt user_id=%s namespace=%s " + "generation=%d documents=%d units=%d token_stats=%d seconds=%.3f", + user_id, + namespace, + target_generation, + aggregate["document_count"], + aggregate["unit_count"], + len(document_frequencies), + time.perf_counter() - started, + ) return target_generation