File size: 6,506 Bytes
6bff5d9 32abc41 b9dfc76 6bff5d9 32abc41 6bff5d9 b9dfc76 6bff5d9 b9dfc76 81e5fe7 0721bb4 81e5fe7 b9dfc76 32abc41 b9dfc76 32abc41 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """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_<hash>` 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_<hash>` 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))
|