Scope structured_flow + report to analysis catalog; drop dead data_sources binding

Read the analysis-scope data_catalog row (bound db+file sources, real names)
for structured_flow and the report appendix, with a user-scope fallback —
matching Sofia's check change. Fixes structured_flow being silently unscoped
(the data_sources binding table was never written: /analysis/create is Go-owned
and unwired, so it was always empty and fail-opened to the full user catalog)
and the synthetic postgres_<hash> names in traceability sources + the report.

Removes the now-dead #10 binding infra: data_sources table/model,
AnalysisDataSourceStore, _ScopedCatalogReader, _bound_source_ids, _read_binding,
and the unwired src/api/v1/analysis.py. Drop the empty table in Neon:
DROP TABLE IF EXISTS data_sources

REPO_STATUS.md

@@ -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`~~ *(dropped from Python)* | per-analysis binding table; **no longer read or written by Python** — the writer (`/analysis/create`) was Go-owned/unwired so it was always empty. Scoping now reads the analysis-scope `data_catalog` row instead | — | — |
 | `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.

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
@@ -211,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.
 
@@ -430,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).
@@ -787,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/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/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).