[NOTICKET][AI] perf(chat): PR-DB1 - memoizing catalog reader + shared ChatHandler

Cut redundant catalog-DB round-trips in the slow path and stop rebuilding the
Azure LLM clients per request.

- MemoizingCatalogReader (catalog/reader.py): request-scoped wrapper that caches
each read by source_hint. ChatHandler builds one per request and threads it into
the slow-path DataAccessToolInvoker, so the planner's catalog load is reused by
describe_source / query_structured instead of re-reading from the catalog DB.
Measured live: describe_source 3.5s -> ~2.0s; catalog reads/request ~5 -> ~2.
- chat.py: one module-level ChatHandler singleton instead of per-request, keeping
the Orchestrator/Chatbot Azure clients (and TLS pools) warm across requests.

Does not touch the external query_structured handshake (that is the engine-cache
work, DB2). Also logs the principal-review fix tracker (R1-R12, DB1-DB3) in PROGRESS.

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

.gitignore

@@ -46,6 +46,7 @@ test_tesseract.py
 # Windows binaries — installed via apt in Docker instead
 software/
 
+scripts/
 tests/
 .claude/
 migratego/

PROGRESS.md

@@ -2,11 +2,65 @@
 
 Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "Team — division of work". Update as PRs land. Future Claude Code sessions read this to know what's already done.
 
-**Last updated**: 2026-06-10 (tool layer complete + Langfuse tracing + gated slow-path wiring)
+**Last updated**: 2026-06-10 (principal review findings logged + PR-DB1 latency work started)
 **Current open PR**: `pr/2` — active.
 
 ---
 
+## Principal architecture review (2026-06-10) — findings + fix tracker
+
+A full external review (read the context docs + the slow path, tool layer, query
+spine, catalog plumbing, chat endpoint, config/connection layers) landed. It confirmed
+the DB-latency diagnosis and surfaced several gaps **not previously tracked here**.
+Verified against code before logging. Severity: **critical** / important / nice-to-have.
+
+**Runtime / latency (the original problem):**
+- DB connection handling is the anomaly, NOT cold start. `DbExecutor._run_sync`
+  (`db.py:192`) → `engine_scope` does `create_engine → connect (TCP+TLS+SCRAM) → 2×SET
+  → dispose` on EVERY query. Measured ~6–8s for 60 rows; a 2nd warm-session query was
+  still ~6.6s → per-call handshake, never amortized. `engine_scope`'s connect-once-dispose
+  semantics were designed for the ingestion pipeline and wrongly inherited by the query path.
+- `describe_source` ~3.5s is **planner-induced waste**: every few-shot (`examples.py`)
+  opens with a `describe_source` task, so the LLM always plans a tool that re-reads from
+  the catalog DB the same catalog already rendered into its prompt. Its impl does 2
+  sequential full-catalog reads (`data_access.py:127-128`). Total catalog reads/request ~5×.
+- Azure LLM clients rebuilt per request: `ChatHandler(enable_tracing=True)` is constructed
+  per request (`chat.py:172`) → fresh Orchestrator/Chatbot → fresh AzureChatOpenAI → fresh
+  TLS to Azure each call. Planner/Assembler correctly use module singletons; the other two don't.
+- Tokens (~13k/request) are NORMAL for this design — do not optimize for $.
+- **Reject the scheduled DB-warmer idea**: targets cold start (~1.8s slice) not the per-call
+  handshake, keeps serverless user DBs awake 24/7 (their compute bill), and decrypts every
+  tenant's creds on a cron (attack surface). Strictly dominated by an engine cache +
+  request-scoped pre-connect.
+
+**Fix tracker (new):**
+
+| # | Fix | Severity | Owner | Status |
+|---|---|---|---|---|
+| R1 | **AuthN/AuthZ** on data endpoints — derive `user_id` from JWT (`security/auth.py` helpers exist), reject body-supplied IDs. `/chat/stream` has none (`chat.py:40,128`); tenant isolation is client honesty. **Gates the engine-cache work.** | **critical** | DB/B | `[ ]` |
+| R2 | **Always compile a LIMIT** — `sql.py:281` emits no LIMIT when `ir.limit is None`; `db.py:93` materializes ALL rows before the 10k cap. Compile `LIMIT min(ir.limit or CAP, CAP)+1`. | **critical** | DB | `[ ]` |
+| R3 | **Commit `tests/` + minimal CI** — `tests/` is gitignored; the 200+ tests cited as done exist only on laptops (already caused rename rot). GitHub origin carries tests; HF Space gets the Docker build (already doesn't COPY tests). | **critical (process)** | shared | `[ ]` |
+| DB1 | **In-memory `describe_source`** (request-scoped `MemoizingCatalogReader`, `reader.py`) + **LLM-client hoist** (shared module-level `ChatHandler` in `chat.py`). Measured live: `describe_source` 3.5s→~2.0s (structured read now served from the planner's cached snapshot; only the unstructured read remains a round-trip), catalog reads/request ~5→~2. External `query_structured` handshake unchanged (DB2's job) so total slow path is ~flat until DB2. Tests: `tests/catalog/test_reader.py`. | important | agent | `[x]` |
+| DB2 | **Keyed engine cache** (LRU, small pool, `pool_recycle`, `pool_pre_ping`) replacing per-call `engine_scope`; read-only + statement_timeout via `connect_args options` (zero SET round-trips, read-only-at-birth); invalidate on client update/delete. Extract to `src/database_client/engine.py`. **Land with/after R1.** | important | DB | `[ ]` |
+| DB3 | **Speculative pre-connect** overlapped with the ~4s Planner call (only if DB2 still shows first-touch handshake). | nice-to-have | DB | `[ ]` |
+| R4 | **Stream the Assembler + emit TaskRunner progress events** — slow path streams nothing for ~20s (`chat_handler.py:321` single chunk); proxy idle-timeout + UX risk. | important | agent | `[ ]` |
+| R5 | **Response cache**: key on `user_id` + catalog version; invalidate on ingest. Today `chat:{room_id}:{message}`, 24h TTL, no user (`chat.py:138`) → cross-room replay + stale answers. | important | B | `[ ]` |
+| R6 | **Hard time budget** — wrap `coordinator.run()` in `asyncio.wait_for` (60–90s). `Constraints.time_budget_seconds` is rendered but not enforced. | important | agent | `[ ]` |
+| R7 | **Root-task-failure short-circuit** before the Assembler (templated/fast-path fallback, NOT replanning) — stops paying ~2k tok to narrate an empty RunState. | important | agent | `[ ]` |
+| R8 | **Catalog upsert race** — per-user advisory lock around read-merge-upsert (`store.py`); concurrent uploads can drop a source. | important | DB | `[ ]` |
+| R9 | **`extra="ignore"`** in `settings.py:15` (currently `allow` → typo'd env vars silently swallowed); require Azure keys in prod. | nice-to-have | B | `[ ]` |
+| R10 | **Read-only enforcement is session-state, not a server role.** `REPO_CONTEXT.md` counts "read-only DB credentials" as a defense layer but nothing requests/verifies a read-only role. Either request read-only creds at registration (verify via `SELECT current_setting(...)`) or drop the claim. | important | DB | `[ ]` |
+| R11 | **De-duplicate** `_PLACEHOLDER_RE` (`task_runner.py:31` vs validator) and `_DATA_ACCESS_TOOLS` (invoker vs planner registry) — import one from the other; comments aren't a sync mechanism. | nice-to-have | agent/tool | `[ ]` |
+| R12 | **Doc/process hygiene** — gitignored canonical docs (`AGENT_ARCHITECTURE_CONTEXT_new.md`, `PROJECT_SUMMARY.md`) are cited by code docstrings but absent on disk; `CLAUDE.md` lists deleted modules (enricher, `pipeline/orchestrator.py`); `main` is 38 commits behind on a dead architecture. | nice-to-have | agent | `[ ]` |
+
+**Architecture verdict:** fundamentally sound (catalog-driven IR + deterministic compiler
++ static plan is the right call). Debt is transitional duplication (two planners/registries/
+contract modules — documented, owned) and `ChatHandler` drifting toward a god object
+(extract the slow-path composition root + the SSE `_build_sources`/`_normalize_chunks`
+mappers when convenient).
+
+---
+
 ## What just shipped (2026-06-09/10 — tool layer, tracing, slow-path wiring)
 
 Big stretch since the slow-path workers landed. The tool layer (teammate-owned) is

src/agents/chat_handler.py

@@ -156,10 +156,17 @@ class ChatHandler:
         # ---- 2. Route ------------------------------------------------
         if decision.source_hint == "structured":
             try:
-                catalog = await self._get_catalog_reader().read(user_id, "structured")
+                # One memoizing reader per request: the same catalog is otherwise
+                # 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())
+                catalog = await req_reader.read(user_id, "structured")
                 if self._enable_slow_path:
                     async for event in self._run_slow_path(
-                        user_id, rewritten, catalog, tracer
+                        user_id, rewritten, catalog, tracer, req_reader
                     ):
                         yield event
                     return
@@ -238,7 +245,7 @@ class ChatHandler:
         return RequestTracer.start(user_id=user_id, question=question)
 
     def _get_slow_path_coordinator(
-        self, user_id: str, tracer: Any = None
+        self, user_id: str, tracer: Any = None, catalog_reader: CatalogReader | None = None
     ) -> SlowPathCoordinator:
         """Build the per-request slow-path coordinator (composition root).
 
@@ -260,7 +267,7 @@ class ChatHandler:
         from .slow_path.task_runner import TaskRunner
 
         invoker: Any = CompositeToolInvoker(
-            DataAccessToolInvoker(user_id, self._get_catalog_reader()),
+            DataAccessToolInvoker(user_id, catalog_reader or self._get_catalog_reader()),
             AnalyticsToolInvoker(),
         )
         if tracer is not None and getattr(tracer, "active", False):
@@ -285,6 +292,7 @@ class ChatHandler:
         query: str,
         catalog: Any,
         tracer: Any = None,
+        catalog_reader: CatalogReader | None = None,
     ) -> AsyncIterator[dict[str, Any]]:
         """Run the slow path and stream its assembled answer as SSE events.
 
@@ -301,7 +309,7 @@ class ChatHandler:
 
             tracer = NullTracer()
 
-        coordinator = self._get_slow_path_coordinator(user_id, tracer)
+        coordinator = self._get_slow_path_coordinator(user_id, tracer, catalog_reader)
         context = await get_business_context(user_id)
         pc = tracer.callbacks()              # planner: PII-safe, full capture
         ac = tracer.callbacks(masked=True)   # assembler: sees real rows -> masked

src/api/v1/chat.py

@@ -22,6 +22,12 @@ logger = get_logger("chat_api")
 
 router = APIRouter(prefix="/api/v1", tags=["Chat"])
 
+# One shared ChatHandler for the process. It holds no per-request state (user_id
+# is passed into handle()), and lazily builds + caches the Orchestrator/Chatbot
+# chains — so reusing it keeps the Azure OpenAI clients (and their httpx/TLS pools)
+# warm across requests instead of re-handshaking on the first call of every request.
+_chat_handler = ChatHandler(enable_tracing=True)
+
 _GREETINGS = frozenset(["hi", "hello", "hey", "halo", "hai", "hei"])
 _GOODBYES = frozenset(["bye", "goodbye", "thanks", "thank you", "terima kasih", "sampai jumpa"])
 
@@ -169,7 +175,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
             return EventSourceResponse(stream_direct())
 
         history = await load_history(db, request.room_id, limit=10)
-        handler = ChatHandler(enable_tracing=True)
+        handler = _chat_handler
 
         async def stream_response():
             logger.info("stream_response started", room_id=request.room_id, user_id=request.user_id)

src/catalog/reader.py

@@ -38,3 +38,30 @@ class CatalogReader:
             filtered = [s for s in catalog.sources if s.source_type == "unstructured"]
 
         return catalog.model_copy(update={"sources": filtered})
+
+
+class MemoizingCatalogReader(CatalogReader):
+    """Request-scoped CatalogReader that caches each ``read`` by source_hint.
+
+    One per request. The same per-user catalog is otherwise fetched from the
+    catalog DB 4-5x during a single slow-path run (planner load, then
+    describe_source's structured+unstructured reads, then query_structured's
+    structured read). Wrapping the base reader collapses those to one round-trip
+    per distinct source_hint and pins a single consistent snapshot for the whole
+    request (plan-time and execution-time catalogs can no longer diverge).
+    """
+
+    def __init__(self, inner: CatalogReader) -> None:
+        # `read` is fully overridden below and delegates to `inner`, so the parent's
+        # `_store` is never used — carry it through only so this stays a real
+        # CatalogReader (any inner with a `read` works, including test fakes).
+        super().__init__(getattr(inner, "_store", None))
+        self._inner = inner
+        self._cache: dict[SourceHint, Catalog] = {}
+
+    async def read(self, user_id: str, source_hint: SourceHint) -> Catalog:
+        cached = self._cache.get(source_hint)
+        if cached is None:
+            cached = await self._inner.read(user_id, source_hint)
+            self._cache[source_hint] = cached
+        return cached