"""CatalogReader — loads + filters catalog by source_hint. For typical users (≤50 tables), returns the FULL catalog with no slicing. Catalog-level search is added later if catalog grows past the limit. """ from datetime import UTC, datetime from typing import Literal from src.middlewares.logging import get_logger from .models import Catalog, Source from .store import CatalogStore logger = get_logger("catalog_reader") SourceHint = Literal["chat", "unstructured", "structured"] def _filter_sources(catalog: Catalog, source_hint: SourceHint) -> Catalog: """Return a copy of `catalog` keeping only the sources matching `source_hint`.""" filtered: list[Source] if source_hint == "chat": filtered = [] elif source_hint == "structured": filtered = [s for s in catalog.sources if s.source_type in {"schema", "tabular"}] else: # "unstructured" filtered = [s for s in catalog.sources if s.source_type == "unstructured"] return catalog.model_copy(update={"sources": filtered}) class CatalogReader: """Loads the user's catalog and filters by source_hint. On miss, returns an empty Catalog (never raises) — query path is responsible for handling "no data registered yet" gracefully. Returned Catalog is always a copy; the underlying stored catalog is never mutated. """ def __init__(self, store: CatalogStore) -> None: self._store = store async def read(self, user_id: str, source_hint: SourceHint) -> Catalog: catalog = await self._store.get(user_id) if catalog is None: return Catalog(user_id=user_id, generated_at=datetime.now(UTC)) return _filter_sources(catalog, source_hint) class MemoizingCatalogReader(CatalogReader): """Request-scoped CatalogReader that caches each ``read`` by source_hint. One per request. The same per-user catalog is otherwise fetched from the catalog DB 4-5x during a single slow-path run (planner load, then check_data's structured read + check_knowledge's unstructured read, then retrieve_data's structured read). Wrapping the base reader collapses those to one round-trip per distinct source_hint and pins a single consistent snapshot for the whole request (plan-time and execution-time catalogs can no longer diverge). """ def __init__(self, inner: CatalogReader) -> None: # `read` is fully overridden below and delegates to `inner`, so the parent's # `_store` is never used — carry it through only so this stays a real # CatalogReader (any inner with a `read` works, including test fakes). super().__init__(getattr(inner, "_store", None)) self._inner = inner self._cache: dict[SourceHint, Catalog] = {} async def read(self, user_id: str, source_hint: SourceHint) -> Catalog: cached = self._cache.get(source_hint) if cached is None: cached = await self._inner.read(user_id, source_hint) self._cache[source_hint] = cached return cached class AnalysisScopedCatalogReader(CatalogReader): """Reads the analysis-scope catalog, falling back to the user-scope reader. Used by the `check` skill so "what data do I have" inside a room reflects that analysis's bound sources — structured AND documents — with their real names. A database shows as "xl test" (analysis-scope) instead of the auto-generated `postgres_` placeholder, and documents show at all (the user-scope catalog holds no `unstructured` sources, so reading them from user-scope always came back empty). Fallback rule (tightened 2026-07-13): the user-scope reader is used ONLY when there is no `analysis_id` at all (a legacy room with no analysis concept). When an `analysis_id` IS present but its catalog row is missing (legacy analysis created before catalog materialization, or Go hasn't rebuilt the binding yet) or the read fails, this returns an EMPTY catalog — NOT the user-scope catalog. Previously it degraded to user-scope, which silently surfaced sources that are NOT bound to this analysis as if they were (e.g. a room bound to a CSV answered "check" with the account's unrelated XLSX). An empty result reads correctly as "nothing bound yet — re-save / rebuild the binding to materialize the analysis catalog". Every outcome is logged so a miss is diagnosable instead of silent. """ def __init__(self, inner: CatalogReader, analysis_id: str | None) -> None: # `inner` is a real CatalogReader (constructed at the check call site), so # its `_store` is the live CatalogStore we need for the analysis read. super().__init__(inner._store) self._inner = inner self._analysis_id = analysis_id async def read(self, user_id: str, source_hint: SourceHint) -> Catalog: # No analysis_id at all → legacy room with no analysis-scope concept; fall # back to the user-scope reader (unchanged behavior). if not self._analysis_id: return await self._inner.read(user_id, source_hint) # Analysis-scoped room: read its OWN catalog. Analysis-scope rows carry the # real DB names AND the room's documents (`source_type='unstructured'`), # unlike the user-scope rows (`postgres_` names, no documents). try: catalog = await self._store.get_by_analysis(self._analysis_id) except Exception as e: # noqa: BLE001 — never block check on the analysis read logger.warning( "analysis catalog read failed — returning empty", analysis_id=self._analysis_id, error=str(e), ) catalog = None if catalog is not None: logger.info( "analysis catalog hit", analysis_id=self._analysis_id, sources=len(catalog.sources), ) return _filter_sources(catalog, source_hint) # analysis_id present but no catalog row (or read failed): return EMPTY, # NOT the user-scope catalog. Surfacing user-scope sources here reads as # "these are your bound sources" when they are not (the misleading-XLSX bug). logger.info( "analysis catalog miss — returning empty (no user-scope fallback)", analysis_id=self._analysis_id, ) return Catalog(user_id=user_id, generated_at=datetime.now(UTC))