[NOTICKET] check: scope to analysis catalog + list database tables

Scope the `check` skill's structured reads to the analysis catalog instead of
the user catalog, and expand databases to their table names in the inventory.

Why: at user scope a database shows its auto-generated `postgres_<hash>` name
and every source the user ever registered — not this room's. The analysis-scope
catalog row carries the real name (e.g. "xl test") and only the bound sources.

- store: `get_by_analysis()` reads the `scope_type='analysis'` catalog row
(additive; the user-scope `get()` is untouched, so other tools/paths are
unaffected).
- reader: `AnalysisScopedCatalogReader` serves analysis-scope for STRUCTURED
reads (the naming/scope gap is DB-only), passing documents + any miss through
to the user-scope reader so unbound/legacy rooms behave exactly as before.
- chat_handler: the check branch builds the invoker with that reader.
- check: the inventory listing drills each database for its table names and
nests them (uncapped) under the source, so a DB is no longer an opaque
"N tables" and a "what are the other tables?" follow-up is already answered.

Note: not yet verified end-to-end against a live DB (unit-tested with fakes).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/chat_handler.py

@@ -182,13 +182,17 @@ class ChatHandler:
             self._document_retriever = RetrievalRouter()
         return self._document_retriever
 
-    def _get_check_invoker(self, user_id: str) -> Any:
-        """Build the per-request data-access invoker for the `check` skill."""
+    def _get_check_invoker(self, user_id: str, catalog_reader: Any = None) -> Any:
+        """Build the per-request data-access invoker for the `check` skill.
+
+        `catalog_reader` lets the caller scope the read (e.g. to the analysis
+        catalog); defaults to the user-scope reader.
+        """
         if self._check_invoker_factory is not None:
             return self._check_invoker_factory(user_id)
         from ..tools.data_access import DataAccessToolInvoker
 
-        return DataAccessToolInvoker(user_id, self._get_catalog_reader())
+        return DataAccessToolInvoker(user_id, catalog_reader or self._get_catalog_reader())
 
     def _get_ps_agent(self) -> ProblemStatementAgent:
         if self._ps_agent is None:
@@ -477,8 +481,19 @@ class ChatHandler:
                 return
         elif intent == "check":
             try:
+                # Scope check to the analysis catalog: it holds only this room's
+                # bound sources and their real names (a DB shows as "xl test", not
+                # the user-scope `postgres_<hash>` placeholder). Falls back to the
+                # user-scope reader when the analysis has no catalog row.
+                from ..catalog.reader import AnalysisScopedCatalogReader
+
+                scoped_reader = AnalysisScopedCatalogReader(
+                    self._get_catalog_reader(), analysis_id
+                )
                 # Wrap the check invoker so its check_* tool calls land in the trace.
-                invoker = TraceabilityToolInvoker(self._get_check_invoker(user_id), pad)
+                invoker = TraceabilityToolInvoker(
+                    self._get_check_invoker(user_id, scoped_reader), pad
+                )
                 # Detect from the ORIGINAL message (not `rewritten`, which the
                 # router normalizes to English) so the deterministic check reply
                 # matches the user's language like the other paths.

src/agents/handlers/check.py

@@ -179,14 +179,19 @@ def render_tool_output(out: ToolOutput, reply_language: str = "English") -> str:
     return f"{header}\n{separator}\n{body}"
 
 
-def _render_source_list(out: ToolOutput, reply_language: str) -> str:
+def _render_source_list(
+    out: ToolOutput,
+    reply_language: str,
+    db_tables: dict[str, list[tuple[str, Any]]] | None = None,
+) -> str:
     """Render a check_data/check_knowledge *listing* as a bullet list, not a table.
 
     One bullet per source: `- name — Type (N tables)`. The type + table-count
     annotation is only added for structured sources (file vs database); documents
     are all "unstructured", so the section header already says so — just the name.
-    Returns '' when there are no rows. (The column-level schema drill-down still
-    renders as a table via `render_tool_output` — that data is genuinely tabular.)
+    When `db_tables` supplies a database's table names, they are nested as sub-
+    bullets (capped at `_INVENTORY_TABLE_CAP`) so a DB isn't an opaque "N tables".
+    Returns '' when there are no rows.
     """
     if out.kind == "error":
         return _s(reply_language)["lookup_error"].format(error=out.error)
@@ -197,6 +202,8 @@ def _render_source_list(out: ToolOutput, reply_language: str) -> str:
 
     idx = {c: i for i, c in enumerate(columns)}
     type_labels = _SOURCE_TYPE_LABELS.get(reply_language, _SOURCE_TYPE_LABELS["English"])
+    sc = _sc(reply_language)
+    db_tables = db_tables or {}
 
     def _table_word(n: int) -> str:
         if reply_language == "English":
@@ -207,16 +214,70 @@ def _render_source_list(out: ToolOutput, reply_language: str) -> str:
     for row in rows:
         name = str(row[idx["name"]]) if "name" in idx else ""
         st = str(row[idx["source_type"]]) if "source_type" in idx else ""
+        sid = str(row[idx["source_id"]]) if "source_id" in idx else ""
         annotation = ""
         if st and st != "unstructured":
             annotation = type_labels.get(st, st)
             if "table_count" in idx:
                 tc = row[idx["table_count"]]
                 annotation += f" ({tc} {_table_word(int(tc))})"
-        items.append(f"- {name} — {annotation}" if annotation else f"- {name}")
+
+        head = f"- {name} — {annotation}" if annotation else f"- {name}"
+        tables = db_tables.get(sid) if st == "schema" else None
+        if tables:
+            # Uncapped for now: list every table of the database, so a follow-up
+            # "what are the other tables?" is inherently already answered.
+            lines = [head + ":"]
+            for tname, rc in tables:
+                suffix = f" ({rc} {sc['rows_word']})" if rc else ""
+                lines.append(f"    - {tname}{suffix}")
+            items.append("\n".join(lines))
+        else:
+            items.append(head)
     return "\n".join(items)
 
 
+def _distinct_tables(out: ToolOutput) -> list[tuple[str, Any]]:
+    """(table_name, row_count) pairs from a check_data(source_id) output, in order."""
+    if out.kind != "table":
+        return []
+    cols = out.columns or []
+    idx = {c: i for i, c in enumerate(cols)}
+    if "table_name" not in idx:
+        return []
+    seen: dict[str, Any] = {}
+    for r in out.rows or []:
+        tname = str(r[idx["table_name"]])
+        if tname not in seen:
+            seen[tname] = r[idx["table_row_count"]] if "table_row_count" in idx else None
+    return list(seen.items())
+
+
+async def _fetch_db_tables(
+    inventory: ToolOutput, invoker: ToolInvoker
+) -> dict[str, list[tuple[str, Any]]]:
+    """Drill each database source in an inventory for its table names + row counts.
+
+    Only `schema` (database) sources are drilled — tabular files are always a
+    single table, so their `(N tabel)` line already says everything. Returns a
+    map keyed by source_id for `_render_source_list` to nest.
+    """
+    cols = inventory.columns or []
+    idx = {c: i for i, c in enumerate(cols)}
+    if "source_type" not in idx or "source_id" not in idx:
+        return {}
+    dbs = [r for r in (inventory.rows or []) if str(r[idx["source_type"]]) == "schema"]
+    if not dbs:
+        return {}
+    outs = await asyncio.gather(
+        *(invoker.invoke("check_data", {"source_id": str(r[idx["source_id"]])}) for r in dbs)
+    )
+    return {
+        str(r[idx["source_id"]]): _distinct_tables(o)
+        for r, o in zip(dbs, outs, strict=True)
+    }
+
+
 def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
     """All source_ids whose name appears as a whole word in the message.
 
@@ -246,13 +307,16 @@ def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
 
 
 def _render_helicopter(
-    data_out: ToolOutput, knowledge_out: ToolOutput, reply_language: str = "English"
+    data_out: ToolOutput,
+    knowledge_out: ToolOutput,
+    reply_language: str = "English",
+    db_tables: dict[str, list[tuple[str, Any]]] | None = None,
 ) -> str:
     """Stitch structured + document inventory into one helicopter-view reply."""
     strings = _s(reply_language)
     parts: list[str] = []
 
-    data_list = _render_source_list(data_out, reply_language)
+    data_list = _render_source_list(data_out, reply_language, db_tables)
     if data_list:
         parts.append(f"{strings['structured']}:\n{data_list}")
 
@@ -555,7 +619,8 @@ async def run_check(
         inventory = await invoker.invoke("check_data", {})
         if inventory.kind == "error":
             return render_tool_output(inventory, reply_language)
-        listing = _render_source_list(inventory, reply_language)
+        db_tables = await _fetch_db_tables(inventory, invoker)
+        listing = _render_source_list(inventory, reply_language, db_tables)
         if not listing:
             return _no_match
         n = len(inventory.rows or [])
@@ -566,4 +631,5 @@ async def run_check(
         invoker.invoke("check_data", {}),
         invoker.invoke("check_knowledge", {}),
     )
-    return _render_helicopter(data_out, knowledge_out, reply_language)
+    db_tables = await _fetch_db_tables(data_out, invoker)
+    return _render_helicopter(data_out, knowledge_out, reply_language, db_tables)

src/catalog/reader.py

@@ -7,12 +7,24 @@ Catalog-level search is added later if catalog grows past the limit.
 from datetime import UTC, datetime
 from typing import Literal
 
-from .models import Catalog
+from .models import Catalog, Source
 from .store import CatalogStore
 
 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.
 
@@ -30,14 +42,7 @@ class CatalogReader:
         if catalog is None:
             return Catalog(user_id=user_id, generated_at=datetime.now(UTC))
 
-        if source_hint == "chat":
-            filtered: list = []
-        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})
+        return _filter_sources(catalog, source_hint)
 
 
 class MemoizingCatalogReader(CatalogReader):
@@ -66,3 +71,38 @@ class MemoizingCatalogReader(CatalogReader):
             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 AND their real names — a database shows as
+    "xl test" (analysis-scope) instead of the auto-generated `postgres_<hash>`
+    placeholder stored in the user-scope row. When the analysis has no catalog
+    row (legacy / not yet bound) or the read fails, it degrades to the wrapped
+    user-scope reader, so unbound rooms behave exactly as before.
+    """
+
+    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:
+        # Only STRUCTURED reads get analysis scope — that's where the naming/scope
+        # problem lives (databases named `postgres_<hash>` in user-scope vs their
+        # real name in analysis-scope). Documents pass through to the user-scope
+        # reader: they carry no scope-name gap, and we don't assume Go populates
+        # the analysis-scope catalog with unstructured sources (scoping them could
+        # wrongly hide a user's documents in a room).
+        if source_hint == "structured" and self._analysis_id:
+            try:
+                catalog = await self._store.get_by_analysis(self._analysis_id)
+            except Exception:  # noqa: BLE001 — never block check on the analysis read
+                catalog = None
+            if catalog is not None:
+                return _filter_sources(catalog, source_hint)
+        return await self._inner.read(user_id, source_hint)

src/catalog/store.py

@@ -43,6 +43,28 @@ class CatalogStore:
         # edges so the planner and validator agree. No-op once Go emits real FKs.
         return infer_foreign_keys(Catalog.model_validate(row))
 
+    async def get_by_analysis(self, analysis_id: str) -> Catalog | None:
+        """Read the `scope_type='analysis'` catalog row for an analysis.
+
+        Distinct from `get()` (which reads the user-scope row): the analysis-scope
+        payload carries the sources actually bound to this analysis AND their
+        real names (a database is named e.g. "xl test" here, vs the auto-generated
+        `postgres_<hash>` placeholder in the user-scope row). Returns None when the
+        analysis has no catalog row (legacy / not yet bound), so callers fall back
+        to the user-scope catalog.
+        """
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(CatalogRow.catalog_payload).where(
+                    CatalogRow.analysis_id == analysis_id,
+                    CatalogRow.scope_type == "analysis",
+                )
+            )
+            row = result.scalar_one_or_none()
+        if row is None:
+            return None
+        return infer_foreign_keys(Catalog.model_validate(row))
+
     async def upsert(self, catalog: Catalog) -> None:
         # Legacy: Go's catalog.Service owns catalog writes now. Kept working (and
         # reconciled to the dedorch shape) but no longer on any live Python path.
@@ -86,7 +108,9 @@ class CatalogStore:
             return
         filtered = [s for s in existing.sources if s.source_id != source_id]
         if len(filtered) == len(existing.sources):
-            logger.info("remove_source: source not in catalog", user_id=user_id, source_id=source_id)
+            logger.info(
+                "remove_source: source not in catalog", user_id=user_id, source_id=source_id
+            )
             return
         await self.upsert(existing.model_copy(update={"sources": filtered}))
         logger.info("remove_source: source removed", user_id=user_id, source_id=source_id)