[KM-652] fix(orchestrator): un-brick sessions + enforce data-source binding

Three critical fixes found tracing the request lifecycle:

- T-A: rooms from /room/create have no analysis_states row, so with the gate
on the turn redirect-looped (gate fail-closed + a silent no-op on the
problem_validated write) and gate-off it spammed duplicate report versions
(report_id write-back no-op'd). AnalysisStateStore.update now warns instead
of silently no-op'ing; new idempotent ensure() (INSERT ON CONFLICT DO
NOTHING) get-or-creates the row; ChatHandler ensures it once per turn and the
gate reuses that state (also drops the gate/help double-read).

- T-B: #10 binding only scoped the Planner's catalog; the data-access tools
re-read the full catalog themselves, so binding was a hint, not a boundary.
Scope at the reader (_ScopedCatalogReader) so the Planner AND the tools share
one scoped view. Fast path was already correct.

- T-C: _scope_to_binding failed CLOSED on a disjoint binding (empty catalog ->
hard fail). The scoped reader fails open (scoped or whole catalog), matching
the report generator.

Caveat: lazily-created legacy rows carry no bindings, so those analyses run
unscoped (whole catalog) until sessions converge on /analysis/create.

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

src/agents/chat_handler.py

@@ -186,24 +186,21 @@ class ChatHandler:
             self._binding_store = AnalysisDataSourceStore()
         return self._binding_store
 
-    async def _scope_to_binding(self, catalog: Any, analysis_id: str | None) -> Any:
-        """#10: restrict the catalog to the analysis's bound sources.
+    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 → return the catalog unscoped (whole catalog).
+        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 catalog
+            return set()
         try:
-            bound = set(await self._get_binding_store().get(analysis_id))
+            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 catalog
-        if not bound:
-            return catalog
-        return catalog.model_copy(
-            update={"sources": [s for s in catalog.sources if s.source_id in bound]}
-        )
+            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.
@@ -246,20 +243,30 @@ class ChatHandler:
             return
 
         intent = decision.intent
+        # ---- 1a. Ensure session state row (T-A) ----------------------
+        # Rooms created via /room/create have no `analysis_states` row. Without one
+        # the gate redirect-loops and problem_statement / report_id writes silently
+        # no-op. Lazily get-or-create it (idempotent) so any session is gate-ready.
+        analysis_state: AnalysisState | None = None
+        if analysis_id:
+            try:
+                analysis_state = await self._get_state_store().ensure(analysis_id, user_id)
+            except Exception as e:
+                logger.warning(
+                    "analysis state ensure failed", analysis_id=analysis_id, error=str(e)
+                )
+
         # ---- 1b. Gate (deterministic, post-router) -------------------
         # Redirect structured_flow -> problem_statement until the analysis is
-        # validated. Fails closed (not-validated) on a missing row / read error.
+        # validated. Fails closed (not-validated) when the state row is unavailable.
         if self._enable_gate and analysis_id:
             from .gate import gate, stub_analysis_state
 
-            try:
-                state = await self._get_state_store().get(analysis_id)
-            except Exception as e:
-                logger.warning("gate state read failed — not-validated", error=str(e))
-                state = None
             intent = gate(
                 intent,
-                state if state is not None else stub_analysis_state(problem_validated=False),
+                analysis_state
+                if analysis_state is not None
+                else stub_analysis_state(problem_validated=False),
             )
 
         # Back-compat: the frontend still reads `source_hint` off the intent event.
@@ -284,11 +291,15 @@ class ChatHandler:
                 from ..catalog.reader import MemoizingCatalogReader
 
                 req_reader = MemoizingCatalogReader(self._get_catalog_reader())
-                catalog = await req_reader.read(user_id, "structured")
-                catalog = await self._scope_to_binding(catalog, analysis_id)
+                # #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
+                catalog = await reader.read(user_id, "structured")
                 if self._enable_slow_path:
                     async for event in self._run_slow_path(
-                        user_id, rewritten, catalog, tracer, req_reader, analysis_id
+                        user_id, rewritten, catalog, tracer, reader, analysis_id
                     ):
                         yield event
                     return
@@ -344,7 +355,7 @@ class ChatHandler:
             return
         elif intent == "help":
             try:
-                state = await self._load_analysis_state(analysis_id)
+                state = analysis_state or await self._load_analysis_state(analysis_id)
             except Exception as e:
                 logger.error("help route failed", user_id=user_id, error=str(e))
                 yield {"event": "error", "data": f"Help failed: {e}"}
@@ -561,6 +572,31 @@ 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 _intent_to_source_hint(intent: str) -> str:
     """Map a router intent to the legacy `source_hint` value (frontend back-compat)."""
     if intent == "structured_flow":

src/agents/state_store.py

@@ -10,6 +10,8 @@ Mirrors `PostgresAnalysisStore`: each call opens its own `AsyncSession`.
 
 from __future__ import annotations
 
+from sqlalchemy.dialects.postgresql import insert
+
 from src.agents.gate import AnalysisState
 from src.db.postgres.connection import AsyncSessionLocal
 from src.db.postgres.models import AnalysisStateRow
@@ -40,6 +42,38 @@ class AnalysisStateStore:
             row = await session.get(AnalysisStateRow, analysis_id)
             return _row_to_state(row) if row is not None else None
 
+    async def ensure(
+        self,
+        analysis_id: str,
+        owner_id: str,
+        analysis_title: str = "New analysis",
+    ) -> AnalysisState:
+        """Get-or-create the state row for a session (idempotent, race-safe).
+
+        Sessions born from `/room/create` have no `analysis_states` row; without
+        one the gate redirect-loops and `problem_statement` / `report_id` writes
+        silently no-op. Called per turn (analysis_id == room_id) so any session is
+        gate-ready. `INSERT ... ON CONFLICT DO NOTHING` makes concurrent first
+        turns safe; the row is then read back. Legacy rows created this way carry
+        no source bindings — binding scoping fail-opens to the whole catalog.
+        """
+        async with AsyncSessionLocal() as session:
+            stmt = (
+                insert(AnalysisStateRow)
+                .values(
+                    id=analysis_id,
+                    owner_id=owner_id,
+                    analysis_title=analysis_title,
+                    problem_statement="",
+                    problem_validated=False,
+                )
+                .on_conflict_do_nothing(index_elements=[AnalysisStateRow.id])
+            )
+            await session.execute(stmt)
+            await session.commit()
+            row = await session.get(AnalysisStateRow, analysis_id)
+            return _row_to_state(row)
+
     async def create(
         self,
         *,
@@ -78,6 +112,10 @@ class AnalysisStateStore:
         async with AsyncSessionLocal() as session:
             row = await session.get(AnalysisStateRow, analysis_id)
             if row is None:
+                logger.warning(
+                    "analysis_states row missing — update skipped",
+                    analysis_id=analysis_id,
+                )
                 return None
             if problem_statement is not None:
                 row.problem_statement = problem_statement