fix/ check

REPO_STATUS.md

@@ -148,7 +148,7 @@ Two facts to internalise:
 - **Unstructured RAG** over PGVector.
 - **Analytics tools:** 4 registered composite `analyze_*` (descriptive, aggregate, correlation, trend) + 4 data-access tools (check_data, check_knowledge, retrieve_data, retrieve_knowledge). Four further composites (comparison, contribution, profile, segment) exist in code but are **not registered** with the Planner.
 - **Versioned report generation** from persisted records.
-- **Analysis sessions:** data-first creation gate (≥1 bound source), per-analysis data-source binding (#10).
+- **Analysis sessions:** data-first creation gate (≥1 bound source); each turn reads the analysis-scope catalog so it sees only that analysis's bound sources.
 - **Langfuse tracing** (PII-masked), **Redis caching**, **pooled DB engines** + speculative prewarm.
 
 ---
@@ -185,7 +185,7 @@ unless `SKIP_INIT_DB=true`.
 | `report_inputs` *(was `analysis_records`)* | jsonb `AnalysisRecord`, one per slow-path run; **Python-owned** | slow path | ReportGenerator, report readiness |
 | `analyses` *(dedorch, plural)* | uuid `id`, `user_id`, `analysis_title`, `objective`, `business_questions` jsonb, `status` (active\|inactive), `data_bind`(+`data_bind_version`), `report_id`, `report_collection` — **defined by Go migrations**; `problem_statement`/`problem_validated`/`owner_id` already **dropped** there (`0003`/`0004`) | Go `/api/v1/analyses`; Python state store | gate (no-op), Help, report |
 | `reports` *(dedorch)* | uuid, `analysis_id`, `user_id`, `title` + markdown `content` + `version` (UNIQUE per analysis) | Go + Python ReportStore | report API |
-| `data_sources` *(dedorch)* | per-analysis binding; `reference_id` = catalog source_id; `type ∈ document\|database` | Go `/analyses/{id}/data-bind` (+ Python `/analysis/create`) | structured-flow scoping, report appendix |
+| `data_sources` *(dedorch, Go-owned)* | per-analysis binding table. **Python no longer reads or writes it** — bindings live in Go's `analyses.data_bind`, which Go materializes into the analysis-scope `data_catalog` row; Python scopes off that row. The table exists (Go migration) but Python is fully decoupled — do **not** drop it manually | Go migration | — (unused by Python) |
 | `analyses_messages` *(dedorch)* | the analysis chat room (`role ∈ user\|ai`); replaces deprecated `rooms`/`chat_messages` | Go `/analyses/{id}/messages` | Python chat path **not yet migrated here** (§12) |
 
 > ✅ **Python ORM ↔ dedorch drift — reconciled 2026-07-01.** `AnalysisStateRow` (`analyses`) dropped
@@ -260,10 +260,15 @@ Fernet-decrypt creds (with owner check) → `asyncio.to_thread` (30s timeout)
 (read-only + statement_timeout) → 10k row cap. Defense-in-depth: IR validation + compiler whitelist
 + sqlglot guard + read-only session + LIMIT/timeout.
 
-### Data-source binding (#10) — `src/agents/binding_store.py`
-At `/analysis/create`, chosen `data_source_ids` become `data_sources` rows. On a `structured_flow`
-turn the catalog reader is wrapped so the Planner and the tools' re-reads see the same scoped
-catalog. **Fail-open**: empty/disjoint binding → whole catalog.
+### Analysis-scoped catalog reads — `src/catalog/reader.py::AnalysisScopedCatalogReader`
+An analysis is scoped to the sources the user picked by reading the **analysis-scope** catalog
+(`data_catalog` `scope_type='analysis'`, Go-materialized with the bound db + file sources under
+their real names). On a `structured_flow` turn the catalog reader is wrapped so the Planner and the
+tools' re-reads see the same analysis-scoped snapshot; `check` and the report's data-source appendix
+read it too. **Fail-open**: no analysis-scope row → user-scope catalog. The old `data_sources`
+binding table + `AnalysisDataSourceStore`/`_ScopedCatalogReader` (#10) were **removed** — the writer
+(`/analysis/create`) is Go-owned/unwired, so the table was always empty and its consumers fail-opened
+to the whole (mis-named) user catalog.
 
 ### Tool layer — `src/tools/data_access.py`, `src/agents/planner/registry.py`
 `DataAccessToolInvoker` implements the never-throw tool seam for the 4 data-access tools.
@@ -334,7 +339,7 @@ Python is consumer-only). State **re-verified against the Go source 2026-06-29**
   **Python-owned**; its finalized schema goes to Harry so the dedorch migration creates it post-cutover.
 - **Connection-string cutover DONE (2026-07-01).** Python's `postgres_connstring` now points at
   **dedorch** and reads the Go-migrated tables directly. Every ORM model Python reads (`analyses`,
-  `data_sources`, `analyses_messages`, `data_catalog`) has been reconciled to its dedorch shape.
+  `analyses_messages`, `data_catalog`) has been reconciled to its dedorch shape.
   **`init_db()` is now skipped by default** (`settings.skip_init_db` defaults **True**): its privileged
   DDL (`ALTER TABLE rooms …`, index creation) fails on Go-owned tables
   (`InsufficientPrivilegeError: must be owner of table rooms`). Skipping is safe — Go migration `0001`
@@ -369,9 +374,10 @@ records-based report; floor: ≥1 `analyze_*` success). Wiring Go → Python is
   `cryptography.fernet.InvalidToken` — whose `str()` is **empty**, so it logged as `error=""` and
   masqueraded as a DB-connection failure (the executor now logs `repr(e)` to expose it). Tell-apart:
   a valid-but-wrong key → `InvalidToken`; a malformed key → a non-empty `ValueError` at cipher build.
-- **Never-throw seams** are pervasive (tool invoker, query service, executors, state/binding reads,
+- **Never-throw seams** are pervasive (tool invoker, query service, executors, state/catalog reads,
   record persistence, report summary). Failures degrade into soft output rather than raising — good
-  for UX, but they can mask real breakage (e.g. a binding silently fail-opening to the full catalog).
+  for UX, but they can mask real breakage (e.g. a missing analysis-scope catalog silently falling
+  back to the whole user catalog).
 - **Prompts** live in `src/config/prompts/*.md`. `chatbot_system.md` has `guardrails.md` appended so
   guardrails win on conflict.
 - **Tests** are gitignored (team decision) — run them locally.

main.py

@@ -15,7 +15,7 @@ from slowapi.errors import RateLimitExceeded
 # from src.api.v1.users import router as users_router                # unwired: login moved off Python
 # from src.api.v1.db_client import router as db_client_router        # unwired: Go registers DB client
 # from src.api.v1.data_catalog import router as data_catalog_router  # unwired: Go handles the catalog
-# from src.api.v1.analysis import router as analysis_router          # unwired: Go owns create/update analysis
+# NOTE: src.api.v1.analysis was DELETED (Go owns analysis + its data_sources binding).
 # from src.api.v1.chat import router as chat_router  # unwired: replaced by /api/v2/chat/stream
 # NOTE: src.api.v1.chat module still imported by v2 chat + /tools/help — keep the file.
 from src.api.v1.report import router as report_router
@@ -63,7 +63,6 @@ app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 # app.include_router(room_router)          # unwired: replaced by analysis_id
 # app.include_router(db_client_router)     # unwired: Go registers DB client
 # app.include_router(data_catalog_router)  # unwired: Go handles the catalog
-# app.include_router(analysis_router)      # unwired: Go owns create/update analysis
 # app.include_router(chat_router)  # unwired: v2 chat replaces it (drops v1 cache ops routes)
 app.include_router(report_router)
 app.include_router(tools_router)

src/agents/binding_store.py

@@ -1,34 +0,0 @@
-"""AnalysisDataSourceStore — read per-analysis data-source bindings (#10).
-
-The dedorch `data_sources` table records which catalog sources an analysis is scoped
-to (`reference_id` = the catalog source id). It's written at `/analysis/create`; this
-store is the read seam for the two consumers — `structured_flow` catalog scoping and
-the report's data-source appendix.
-
-Fail-open by convention at the call sites: an empty binding (legacy room, or the FE
-not yet sending ids) means "no restriction" — fall back to the whole catalog. Mirrors
-`AnalysisStateStore`: each call opens its own `AsyncSession`.
-"""
-
-from __future__ import annotations
-
-from sqlalchemy import select
-
-from src.db.postgres.connection import AsyncSessionLocal
-from src.db.postgres.models import AnalysisDataSourceRow
-from src.middlewares.logging import get_logger
-
-logger = get_logger("binding_store")
-
-
-class AnalysisDataSourceStore:
-    """Read the bound catalog `source_id`s for an analysis."""
-
-    async def get(self, analysis_id: str) -> list[str]:
-        async with AsyncSessionLocal() as session:
-            result = await session.execute(
-                select(AnalysisDataSourceRow.reference_id).where(
-                    AnalysisDataSourceRow.analysis_id == analysis_id
-                )
-            )
-            return list(result.scalars().all())

src/agents/chat_handler.py

@@ -109,7 +109,6 @@ class ChatHandler:
         ps_agent: ProblemStatementAgent | None = None,
         help_agent: HelpAgent | None = None,
         state_store: Any | None = None,
-        binding_store: Any | None = None,
         input_guard: InputGuard | None = None,
         enable_gate: bool = False,
         enable_tracing: bool = False,
@@ -138,9 +137,6 @@ class ChatHandler:
         # `help` skill: LLM guide that reads the Analysis State + chat history.
         self._help_agent = help_agent
         self._state_store = state_store
-        # `#10` data-source binding: scopes structured_flow's catalog to the sources
-        # the analysis is bound to. Injectable for tests; fail-open when absent.
-        self._binding_store = binding_store
         # Input guard: screens each message for prompt-injection / secret-extraction /
         # abuse BEFORE the router. Injectable for tests; lazily built in production.
         self._input_guard = input_guard
@@ -182,13 +178,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:
@@ -207,29 +207,6 @@ class ChatHandler:
             self._state_store = AnalysisStateStore()
         return self._state_store
 
-    def _get_binding_store(self) -> Any:
-        if self._binding_store is None:
-            from .binding_store import AnalysisDataSourceStore
-
-            self._binding_store = AnalysisDataSourceStore()
-        return self._binding_store
-
-    async def _bound_source_ids(self, analysis_id: str | None) -> set[str]:
-        """#10: the catalog source_ids this analysis is bound to (empty = unscoped).
-
-        Fail-open: no analysis_id, no binding rows (legacy room / FE not sending
-        ids), or a read error → empty set, which the caller treats as "whole
-        catalog". Used to build a `_ScopedCatalogReader` so the Planner AND the
-        data-access tools (which re-read the catalog themselves) see the same scope.
-        """
-        if not analysis_id:
-            return set()
-        try:
-            return set(await self._get_binding_store().get(analysis_id))
-        except Exception as e:  # noqa: BLE001 — never block the query on this
-            logger.warning("binding read failed — unscoped", analysis_id=analysis_id, error=str(e))
-            return set()
-
     async def _load_analysis_state(self, analysis_id: str | None) -> AnalysisState:
         """Load Analysis State for the Help skill; fail closed to a not-validated stub.
 
@@ -426,14 +403,20 @@ class ChatHandler:
                 # re-fetched from the catalog DB 4-5x across the slow-path run. This
                 # collapses those to one round-trip per source_hint and pins a single
                 # consistent snapshot for plan + execution.
-                from ..catalog.reader import MemoizingCatalogReader
-
-                req_reader = MemoizingCatalogReader(self._get_catalog_reader())
-                # #10: scope every catalog read — the Planner's AND the data-access
-                # tools' own re-reads — to the analysis's bound sources, so binding
-                # is a boundary, not just a planner hint (T-B). Fail-open (T-C).
-                bound = await self._bound_source_ids(analysis_id)
-                reader = _ScopedCatalogReader(req_reader, bound) if bound else req_reader
+                from ..catalog.reader import (
+                    AnalysisScopedCatalogReader,
+                    MemoizingCatalogReader,
+                )
+
+                # Scope every catalog read — the Planner's AND the data-access tools'
+                # own re-reads — to the analysis-scope catalog: Go materializes it with
+                # exactly this analysis's bound db + file sources under their real
+                # names. Falls back to the user-scope catalog when no analysis row
+                # exists. Memoized so plan + execution share one snapshot.
+                scoped = AnalysisScopedCatalogReader(
+                    self._get_catalog_reader(), analysis_id
+                )
+                reader = MemoizingCatalogReader(scoped)
                 catalog = await reader.read(user_id, "structured")
                 # structured_flow always runs the slow analytical path (the
                 # ENABLE_SLOW_PATH flag was removed 2026-07-02).
@@ -477,8 +460,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.
@@ -772,31 +766,6 @@ class ChatHandler:
         yield {"event": "done", "data": ""}
 
 
-class _ScopedCatalogReader:
-    """Wraps a CatalogReader, restricting `structured` reads to an analysis's bound
-    sources (#10).
-
-    Scoping lives here — not at a single call site — so the Planner AND the
-    data-access tools (which re-read the catalog themselves) see the same scoped
-    view; otherwise binding is only a hint to the Planner while the executor runs
-    against the full catalog. Fail-open: an empty or fully-disjoint binding yields
-    the whole catalog, so a stale / cross-source binding degrades instead of
-    emptying the catalog. Only `structured` reads are scoped (all #10 binds today);
-    `unstructured` / retrieval reads pass through.
-    """
-
-    def __init__(self, inner: Any, bound: set[str]) -> None:
-        self._inner = inner
-        self._bound = bound
-
-    async def read(self, user_id: str, source_hint: str) -> Any:
-        catalog = await self._inner.read(user_id, source_hint)
-        if not self._bound or source_hint != "structured":
-            return catalog
-        scoped = [s for s in catalog.sources if s.source_id in self._bound]
-        return catalog.model_copy(update={"sources": scoped or catalog.sources})
-
-
 def _normalize_chunks(raw: Any) -> list[DocumentChunk]:
     """Convert whatever the retriever returns into list[DocumentChunk].
 

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/agents/planner/validator.py

@@ -19,6 +19,7 @@ from ...catalog.models import Catalog
 from ...query.ir.models import QueryIR
 from ...query.ir.repair import IRRepairer
 from ...query.ir.validator import IRValidationError, IRValidator
+from ...tools.analytics.aggregation import SUPPORTED_AGGS
 from .contracts import ToolRegistry
 from .errors import PlannerValidationError
 from .inputs import Constraints
@@ -105,6 +106,33 @@ class PlannerValidator:
                         f"{sorted(unknown)} (allowed: {sorted(allowed)})"
                     )
 
+                # Check 8c — analyze_aggregate: every aggregation FUNCTION must be one
+                # the tool supports. Check 8a only validates arg *names* (`aggregations`
+                # is allowed); it never looks at the function *values* inside the dict,
+                # so an unsupported func like `std` otherwise passes validation and only
+                # fails at execution — too late for a corrective retry, so the task
+                # reaches the Assembler as a silent failure. Catch it here so the planner
+                # is re-prompted to degrade to a supported function (e.g. `mean`).
+                if call.tool == "analyze_aggregate":
+                    aggs = call.args.get("aggregations")
+                    if isinstance(aggs, dict):
+                        bad = sorted(
+                            {
+                                f
+                                for funcs in aggs.values()
+                                for f in ([funcs] if isinstance(funcs, str) else funcs or [])
+                                if f not in SUPPORTED_AGGS
+                            }
+                        )
+                        if bad:
+                            raise PlannerValidationError(
+                                f"task {task.id}: analyze_aggregate has unsupported "
+                                f"aggregation function(s) {bad} (supported: "
+                                f"{sorted(SUPPORTED_AGGS)}). Use a supported function "
+                                "(e.g. mean/median); for the spread of a whole column "
+                                "use analyze_descriptive instead."
+                            )
+
                 # Check 3 — concrete source_id args must exist in the catalog.
                 src = call.args.get("source_id")
                 if isinstance(src, str) and not _is_placeholder(src):

src/agents/report/generator.py

@@ -133,15 +133,14 @@ def _collect_method_steps(records: list[AnalysisRecord]) -> list[TaskSummary]:
 
 
 def _build_data_sources(
-    records: list[AnalysisRecord], catalog, bound_ids: list[str] | None = None
+    records: list[AnalysisRecord], catalog
 ) -> list[DataSourceRef]:
     """Freeze real catalog metadata for the sources this analysis used.
 
-    When the analysis has a data-source binding (#10), the candidate set is scoped
-    to the bound sources first (fail-open if the binding doesn't intersect the
-    catalog). Within that set, matches catalog sources against the records'
-    (narrative) `data_used` by name/id; falls back to all (bound) sources, then to
-    bare `data_used` strings if no catalog is available — so the section is always
+    `catalog` is the analysis-scope catalog — already restricted to this analysis's
+    bound sources — so every source in it is a candidate. Matches candidates against
+    the records' (narrative) `data_used` by name/id; falls back to all sources, then
+    to bare `data_used` strings if no catalog is available — so the section is always
     populated, best-effort.
     """
     if catalog is None or not catalog.sources:
@@ -153,9 +152,6 @@ def _build_data_sources(
         return [DataSourceRef(source_id=d, name=d, source_type="", detail={}) for d in seen]
 
     candidates = catalog.sources
-    if bound_ids:
-        scoped = [s for s in candidates if s.source_id in set(bound_ids)]
-        candidates = scoped or candidates  # fail-open if binding doesn't match catalog
 
     def _ref(s) -> DataSourceRef:
         return DataSourceRef(
@@ -337,12 +333,10 @@ class ReportGenerator:
         record_store=None,
         structured_chain: Runnable | None = None,
         catalog_store=None,
-        binding_store=None,
     ) -> None:
         self._record_store = record_store
         self._chain = structured_chain
         self._catalog_store = catalog_store
-        self._binding_store = binding_store
 
     def _ensure_record_store(self):
         if self._record_store is None:
@@ -383,9 +377,8 @@ class ReportGenerator:
         caveats = _collect_notes(records, "caveats")
         open_questions = _collect_notes(records, "open_questions")
         method_steps = _collect_method_steps(records)
-        bound_ids = await self._read_binding(analysis_id)
         data_sources = _build_data_sources(
-            records, await self._read_catalog(user_id), bound_ids
+            records, await self._read_catalog(user_id, analysis_id)
         )
         executive_summary = await self._summarize(ps, findings, caveats)
 
@@ -414,30 +407,21 @@ class ReportGenerator:
         )
         return report
 
-    async def _read_catalog(self, user_id: str | None):
-        if not user_id:
-            return None
+    async def _read_catalog(self, user_id: str | None, analysis_id: str | None):
+        """Prefer the analysis-scope catalog (this analysis's bound sources + their
+        real names); fall back to the user-scope catalog when the analysis has no row
+        (legacy / unbound)."""
         try:
-            return await self._ensure_catalog_store().get(user_id)
+            store = self._ensure_catalog_store()
+            if analysis_id:
+                cat = await store.get_by_analysis(analysis_id)
+                if cat is not None:
+                    return cat
+            return await store.get(user_id) if user_id else None
         except Exception as exc:  # data_sources falls back; never break the report
             logger.warning("catalog read failed; data_sources will fall back", error=str(exc))
             return None
 
-    def _ensure_binding_store(self):
-        if self._binding_store is None:
-            from ..binding_store import AnalysisDataSourceStore
-
-            self._binding_store = AnalysisDataSourceStore()
-        return self._binding_store
-
-    async def _read_binding(self, analysis_id: str) -> list[str]:
-        """Bound source ids for the analysis (#10). Never-throw → [] (unscoped)."""
-        try:
-            return await self._ensure_binding_store().get(analysis_id)
-        except Exception as exc:  # data_sources falls back to whole catalog
-            logger.warning("binding read failed; data_sources unscoped", error=str(exc))
-            return []
-
     async def _summarize(
         self, ps: ProblemStatement, findings: list[ReportFinding], caveats: list[AttributedNote]
     ) -> str:

src/api/v1/analysis.py

@@ -1,174 +0,0 @@
-"""Analysis session API — create a new analysis (the per-session workspace).
-
-An analysis IS the chat session: the `analysis_states` row and the chat `rooms`
-row share one id (`analysis_id == room_id`), so the existing `room_id` on the chat
-request doubles as the `analysis_id`. Creating an analysis enforces the data-first
-gate (>=1 bound source) and seeds the state with a title + an optional problem
-statement (validated later by the Problem Statement skill).
-"""
-
-import uuid
-
-from fastapi import APIRouter, Depends, HTTPException
-from pydantic import BaseModel, Field
-from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from src.db.postgres.connection import get_db
-from src.db.postgres.models import AnalysisDataSourceRow, AnalysisStateRow, Room
-from src.middlewares.logging import get_logger, log_execution
-
-logger = get_logger("analysis_api")
-
-router = APIRouter(prefix="/api/v1", tags=["Analysis"])
-
-
-def _serialize_state(row: AnalysisStateRow, data_source_ids: list[str]) -> dict:
-    """The full analysis payload: the 8 state fields + the bound source ids."""
-    return {
-        "id": row.id,
-        "analysis_title": row.analysis_title,
-        "problem_statement": row.problem_statement,
-        "problem_validated": row.problem_validated,
-        "user_id": row.user_id,
-        "report_id": row.report_id,
-        "data_source_ids": data_source_ids,
-        "created_at": row.created_at.isoformat() if row.created_at else None,
-        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
-    }
-
-
-async def _bound_source_ids(db: AsyncSession, analysis_id: str) -> list[str]:
-    result = await db.execute(
-        select(AnalysisDataSourceRow.reference_id).where(
-            AnalysisDataSourceRow.analysis_id == analysis_id
-        )
-    )
-    return list(result.scalars().all())
-
-
-async def _sources_by_id(user_id: str) -> dict:
-    """Catalog sources keyed by source_id, to resolve `type`/`name` on binding.
-
-    Never-throw: missing catalog / read error → empty map, and binding rows fall back
-    to type='unknown' / name=reference_id.
-    """
-    try:
-        from src.catalog.store import CatalogStore
-
-        catalog = await CatalogStore().get(user_id)
-    except Exception as e:  # noqa: BLE001 — binding must not fail on catalog read
-        logger.warning("analysis: catalog read failed for binding", user_id=user_id, error=str(e))
-        return {}
-    return {s.source_id: s for s in catalog.sources} if catalog else {}
-
-
-class CreateAnalysisRequest(BaseModel):
-    user_id: str
-    analysis_title: str = "New analysis"
-    problem_statement: str = ""
-    data_source_ids: list[str] = Field(default_factory=list)
-
-
-@router.post("/analysis/create")
-@log_execution(logger)
-async def create_analysis(
-    request: CreateAnalysisRequest,
-    db: AsyncSession = Depends(get_db),
-):
-    """Create a new analysis session: one shared id for its state + chat room.
-
-    Data-first gate (decision #2): an analysis requires >=1 bound data source.
-    The bound sources are persisted as dedorch `data_sources` rows (#10) in the same
-    transaction as the state + room, so the analysis is scoped to exactly the sources
-    the user picked. `structured_flow` and the report read this binding back.
-    """
-    if not request.data_source_ids:
-        raise HTTPException(
-            status_code=400,
-            detail="An analysis requires at least one bound data source.",
-        )
-
-    analysis_id = str(uuid.uuid4())
-    # The analysis IS the session: state row + chat room + source bindings share one
-    # id, created atomically in one transaction.
-    state_row = AnalysisStateRow(
-        id=analysis_id,
-        user_id=request.user_id,
-        analysis_title=request.analysis_title,
-        problem_statement=request.problem_statement,
-        problem_validated=False,
-    )
-    db.add(Room(id=analysis_id, user_id=request.user_id, title=request.analysis_title))
-    db.add(state_row)
-    # dict.fromkeys dedupes while preserving order. Each binding row snapshots the
-    # source's type + name from the catalog (reference_id = catalog source id);
-    # bound_at/created_at default to now() in dedorch.
-    bound_ids = list(dict.fromkeys(request.data_source_ids))
-    src_by_id = await _sources_by_id(request.user_id)
-    for source_id in bound_ids:
-        src = src_by_id.get(source_id)
-        db.add(
-            AnalysisDataSourceRow(
-                id=str(uuid.uuid4()),
-                analysis_id=analysis_id,
-                type=src.source_type if src else "unknown",
-                name=src.name if src else source_id,
-                reference_id=source_id,
-                bound_by=request.user_id,
-            )
-        )
-    await db.commit()
-    await db.refresh(state_row)
-
-    logger.info(
-        "analysis created",
-        analysis_id=analysis_id,
-        user_id=request.user_id,
-        sources=len(bound_ids),
-    )
-    return {
-        "status": "success",
-        "message": "Analysis created successfully",
-        "data": _serialize_state(state_row, bound_ids),
-    }
-
-
-@router.get("/analysis")
-@log_execution(logger)
-async def list_analyses(user_id: str, db: AsyncSession = Depends(get_db)):
-    """List a user's analyses, most-recently-updated first (Analysis sidebar).
-
-    Summary fields only (no per-row source bindings — fetch those via the detail
-    endpoint) to keep the list a single query.
-    """
-    result = await db.execute(
-        select(AnalysisStateRow)
-        .where(AnalysisStateRow.user_id == user_id)
-        .order_by(AnalysisStateRow.updated_at.desc())
-    )
-    rows = result.scalars().all()
-    return {
-        "status": "success",
-        "data": [
-            {
-                "id": r.id,
-                "analysis_title": r.analysis_title,
-                "problem_validated": r.problem_validated,
-                "report_id": r.report_id,
-                "updated_at": r.updated_at.isoformat() if r.updated_at else None,
-            }
-            for r in rows
-        ],
-    }
-
-
-@router.get("/analysis/{analysis_id}")
-@log_execution(logger)
-async def get_analysis(analysis_id: str, db: AsyncSession = Depends(get_db)):
-    """Read one analysis's state + bound data sources (the FE workspace render)."""
-    row = await db.get(AnalysisStateRow, analysis_id)
-    if row is None:
-        raise HTTPException(status_code=404, detail=f"Analysis {analysis_id!r} not found.")
-    data_source_ids = await _bound_source_ids(db, analysis_id)
-    return {"status": "success", "data": _serialize_state(row, data_source_ids)}

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,42 @@ 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 — 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). 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:
+        # Read analysis-scope for BOTH structured and unstructured. Verified via
+        # the dedorch `data_catalog` table: the analysis-scope rows carry the real
+        # DB names AND the room's documents (`source_type='unstructured'`), whereas
+        # the user-scope rows hold only structured sources with `postgres_<hash>`
+        # placeholder names and NO documents at all — so reading documents from
+        # user-scope always returned empty ("not listed"). Fall back to the
+        # user-scope reader only when the analysis has no catalog row (legacy /
+        # unbound room).
+        if 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)

src/config/prompts/assembler.md

@@ -25,9 +25,17 @@ You produce two things in one structured object:
    and values present in the task results. **Never invent, estimate, or extrapolate
    a number** that is not in the results. If the data does not answer part of the
    question, say so.
-2. **Report what failed.** Some tasks may have `status: partial` or `failure`. Do
-   not pretend they succeeded. Briefly state what could not be completed and how it
-   limits the answer; put unresolved items in `open_questions`.
+2. **Report what failed — in plain terms, and still answer.** Some tasks may have
+   `status: partial` or `failure`. Do not pretend they succeeded — but do not lead with
+   the failure either. **First** give the most useful answer the SUCCESSFUL tasks
+   support; **then** state, in business language, what could not be determined and how
+   it limits the answer; put unresolved items in `open_questions`. **Never expose the
+   internal cause** of a failure — no "the tool failed", "could not compute", "technical
+   error", task ids, or function/tool names. Describe the limit by what it *means for the
+   reader*, not by what broke internally. E.g. write "a formal significance test was not
+   run, so this shows the difference in averages but not whether it is statistically
+   significant" — NOT "the calculation of the score distribution failed". A narrower,
+   honest answer beats an apology.
 3. **Render, don't recompute.** Build markdown tables from the structured task
    outputs as they are. Do not do your own arithmetic beyond trivially restating a
    value already computed.

src/config/prompts/help.md

@@ -1,4 +1,9 @@
-<!-- help.md · v4 · Help skill prompt.
+<!-- help.md · v5 · Help skill prompt.
+     v5 (2026-07-07): added the "Capability boundary" section — Help now only suggests
+     analyses the live tools can actually deliver (descriptive, group-by, correlation, trend)
+     and must NOT suggest significance tests, forecasting/modeling, causal claims, clustering/
+     segmentation, or share-of-total. Fixes Help recommending a statistical-significance
+     question the system has no tool for (the user copy-pasted the suggestion → dead end).
      v4 (2026-07-03): reply language relaxed from hard-"only" to DEFAULT + explicit-request
      exception — an explicit user request ("jawab dalam bahasa Inggris") now overrides the
      detected [Reply language]; anti-drift default (incl. synthetic-trigger protection) is
@@ -75,6 +80,35 @@ Do not over-promise the report's depth.
 > chat skill to fix it; gently suggest they set the objective + business questions in the New
 > Analysis form.
 
+## Capability boundary — only suggest analyses the tools can actually do
+
+Every question you propose must be answerable by the system's live analysis capabilities.
+Suggesting an analysis the tools cannot perform sets the user up to fail: they copy your
+suggestion, ask it, and hit a dead end. Stay strictly inside this list.
+
+**You MAY suggest** (supported):
+- Descriptive summaries of a column — average, median, spread, min/max, distribution.
+- Group-by breakdowns — a total, average, or count of a metric **per category**. This is also
+  how to compare groups (e.g. "the average retention for online vs offline").
+- Correlation / relationship between numeric columns ("which factors relate to exam score?").
+- Trends over time.
+- Inventory — what data / tables / documents exist.
+
+**You must NOT suggest** (no tool exists — do not propose these even when they fit the goal):
+- **Statistical significance / hypothesis tests** — never use "significant", "statistically
+  significant", "significant difference", t-test, ANOVA, or p-value. To compare groups, suggest
+  a group-by average that shows the gap (e.g. "compare the average retention of online vs
+  offline"), NOT "is the difference significant".
+- Predictive modeling, forecasting, or regression models.
+- Causal claims — avoid "cause", "impact of", "effect of" framed as causation; keep it to
+  relationship / correlation.
+- Clustering or segmentation into discovered groups.
+- Share-of-total / contribution breakdowns.
+
+When a goal naturally invites a forbidden analysis (e.g. the user wants to know if a gap is
+"real"), degrade the suggestion to the nearest supported one — the group-by average that shows
+the gap — rather than promising the unsupported analysis.
+
 ## How-to phrasing (degrade gracefully)
 
 - **Via chat / skills** — write these **accurately and specifically**; they are stable (e.g. "type your question in the chat", "run `/report`").
@@ -102,6 +136,7 @@ spam, no overselling. A few sentences is usually enough.
 - Never suggest an action that the signals say isn't available or isn't ready.
 - One step at a time — give the next step, not the whole roadmap.
 - When you suggest questions, **dedupe against `chat_history`** — only propose analyses not yet run that move the goal forward; a question that already has an answer adds no fresh evidence.
+- **Stay inside the Capability boundary above** — never propose significance tests, forecasting, modeling, causal claims, clustering/segmentation, or share-of-total; there is no tool for them, so suggesting them sends the user into a dead end.
 - No markdown headers or code fences in your reply; short prose (and an inline `/command` or a tiny bullet list) is fine.
 
 ## Examples

src/db/postgres/init_db.py

@@ -3,7 +3,6 @@
 from sqlalchemy import text
 from src.db.postgres.connection import engine, Base
 from src.db.postgres.models import (
-    AnalysisDataSourceRow,
     ReportInputRow,
     AnalysisReportRow,
     AnalysisStateRow,

src/db/postgres/models.py

@@ -238,28 +238,6 @@ class AnalysisStateRow(Base):
     )
 
 
-class AnalysisDataSourceRow(Base):
-    """Per-analysis data-source binding (#10) — dedorch `data_sources` (Go-owned).
-
-    Which catalog sources an analysis is scoped to. `reference_id` is the catalog
-    `Source.source_id`; `type`/`name` snapshot the source kind + label. Written at
-    `/analysis/create`; read by `structured_flow` scoping + the report appendix.
-    `source_metadata` maps to the `metadata` column (`metadata` is reserved by the
-    declarative API). Class name kept; table + shape changed for dedorch.
-    """
-    __tablename__ = "data_sources"
-
-    id = Column(UUID(as_uuid=False), primary_key=True)
-    analysis_id = Column(UUID(as_uuid=False), nullable=False, index=True)
-    type = Column(String, nullable=False)
-    name = Column(String, nullable=False)
-    reference_id = Column(String, nullable=False)  # == catalog Source.source_id
-    bound_by = Column(String, nullable=False)
-    bound_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
-    source_metadata = Column("metadata", JSONB, nullable=True)
-    created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
-
-
 class AnalysesMessageRow(Base):
     """One conversation message — dedorch `analyses_messages` (Go-owned table).
 

src/tools/analytics/aggregation.py

@@ -44,8 +44,12 @@ def _clean(value: object) -> object:
 # Final destination is ToolSpec.description once the wrapper layer is built.
 DESCRIPTION = """\
 Summary: Group-by aggregation. Splits rows by one or more key columns and \
-computes aggregates per group (sum, mean, count, min, max, median, nunique). \
-Returns one row per group.
+computes aggregates per group. Returns one row per group.
+
+SUPPORTED FUNCTIONS — use ONLY these: sum, mean, count, min, max, median, \
+nunique. Standard deviation / variance are NOT available here: for the spread of \
+a whole column use analyze_descriptive (whole-column only — spread PER GROUP is \
+not available in v1). NEVER pass std / var / stdev to this tool; it will fail.
 
 USE WHEN the question groups a metric by a category — the tell-tale sign is \
 "per"/"each"/"by" a dimension. Trigger words: "per/each" (per/tiap), "by" \
@@ -57,6 +61,15 @@ DON'T USE WHEN:
   - it splits a single total into shares -> analyze_contribution
   - the grouping is over time periods -> analyze_trend
 
+GRACEFUL DEGRADE — if a tool named above (e.g. analyze_comparison) is NOT present \
+in the "Available tools" list, do NOT fake it with an unsupported aggregation \
+(like std) and do NOT abandon the task. Degrade to an answer that stays on the \
+SAME question: group the measure by the group column with `mean` (optionally \
+min/max/median) so each group's level and their difference are still shown. Do \
+NOT switch to an unrelated tool just to produce output. The Assembler will note \
+that a significance test / per-group spread was not computed — a narrower, true \
+answer beats a failed one.
+
 Example questions:
   - "total revenue per region"
   - "average order value by customer segment"