[NOTICKET] perf(slow-path): DB3 pre-connect + R4 progress events

DB3 (DB owner): DbExecutor.prewarm(catalog, user_id) warms the pooled engine for
schema sources, fired (best-effort, never raises) at slow-path entry so the cold
first-query TCP+TLS+auth handshake overlaps the ~4s Planner LLM call. Gated to the
default path (skipped when a coordinator factory is injected, so tests stay
hermetic). Reuses DbExecutor's existing client resolution + the DB2 engine cache.

R4 (agent): SlowPathCoordinator.run gained an optional `progress` callback;
ChatHandler bridges it to SSE `status` events via an asyncio queue, and chat.py
forwards them. The slow path no longer streams ~13s of silence — live timeline now
shows Planning -> Running N steps -> Composing, max wire gap ~4.6s, fixing proxy
idle-timeout + UX. Status events only appear when the coordinator calls back, so the
existing wiring test's event sequence is unchanged (fake coordinator just accepts
**kwargs). Token-streaming the Assembler answer is deferred: it would require
splitting the Assembler into a streamed prose call + a structured-record call,
doubling its LLM calls (cost/latency) - a separate decision.

Verified live through ChatHandler.handle against the real user DB. 114 tests pass.

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

PROGRESS.md

@@ -42,8 +42,8 @@ Verified against code before logging. Severity: **critical** / important / nice-
 | 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** — `src/database_client/engine.py::UserEngineCache` (process singleton): pooled engines keyed by `client_id + creds-hash` (rotation auto-invalidates), bounded LRU (50) + 600s idle TTL, `pool_pre_ping` + `pool_recycle=300`. `DbExecutor._run_sync` reuses the warm connection instead of `create_engine→connect→dispose` per query (postgres/supabase only; other db_types keep the legacy path — no regression). **Live-measured: warm `query_structured` 6.6–9.4s → ~2.5s** (the residual is the per-call catalog-DB client fetch + pre-ping, not the external handshake). **Finding:** Neon's transaction pooler REJECTS `default_transaction_read_only` as a libpq startup `option` — caught live; moved read-only + statement_timeout to a per-connection `connect` event (best-effort; authoritative read-only is the SELECT-only compiler + sqlglot guard, see R10). Per-request ownership/active check kept. Proceeded ahead of R1 per owner decision (marginal security delta over the existing no-auth state; auth tracked separately). Tests: `tests/database_client/test_engine.py`. First query/process still cold → DB3. | important | DB | `[x]` |
-| 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 | `[ ]` |
+| DB3 | **Speculative pre-connect** — `DbExecutor.prewarm(catalog, user_id)` warms the pooled engine for schema sources (fire-and-forget at slow-path entry) so the cold first-query handshake overlaps the ~4s Planner call. Best-effort, never raises; gated to the default path (skipped when a coordinator factory is injected). Verified live through `ChatHandler.handle`. | nice-to-have | DB | `[x]` |
+| R4 | **Per-stage progress events** — `SlowPathCoordinator.run` gained an optional `progress` callback; `ChatHandler` bridges it to SSE `status` events (`chat.py` forwards them). Live: stream now shows `Planning…`→`Running N steps…`→`Composing…` (max wire gap ~4.6s, was ~13s of silence) → fixes proxy idle-timeout + UX. **Deferred:** token-streaming the Assembler answer needs splitting it into a streamed prose call + a structured-record call — that doubles the Assembler LLM calls (cost/latency), so it's a separate decision; the answer is still emitted as one chunk after the (fast ~2.5s) Assembler. Test: `test_chat_handler_wiring.py`. | 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 | `[ ]` |

src/agents/chat_handler.py

@@ -22,6 +22,7 @@ inject mocks).
 
 from __future__ import annotations
 
+import asyncio
 import json
 from collections.abc import AsyncIterator, Callable
 from typing import TYPE_CHECKING, Any
@@ -311,6 +312,15 @@ class ChatHandler:
 
         coordinator = self._get_slow_path_coordinator(user_id, tracer, catalog_reader)
         context = await get_business_context(user_id)
+
+        # DB3: warm the user's DB connection in parallel with planning so the
+        # handshake overlaps the ~4s Planner call. Default path only — an injected
+        # coordinator factory (tests / custom) may not use the real DbExecutor.
+        if self._slow_path_factory is None:
+            from ..query.executor.db import DbExecutor
+
+            asyncio.create_task(DbExecutor.prewarm(catalog, user_id))  # noqa: RUF006
+
         pc = tracer.callbacks()              # planner: PII-safe, full capture
         ac = tracer.callbacks(masked=True)   # assembler: sees real rows -> masked
         run_kw: dict[str, Any] = {}
@@ -318,8 +328,38 @@ class ChatHandler:
             run_kw["planner_callbacks"] = pc
         if ac:
             run_kw["assembler_callbacks"] = ac
+
+        # R4: bridge the coordinator's per-stage progress callback to SSE `status`
+        # events so the stream isn't silent for ~12s (and proxies don't drop the
+        # idle connection). Status events only appear if the coordinator calls back.
+        progress_q: asyncio.Queue[str] = asyncio.Queue()
+
+        async def _progress(stage: str) -> None:
+            await progress_q.put(stage)
+
+        run_task = asyncio.create_task(
+            coordinator.run(
+                context, catalog, query, Constraints(), progress=_progress, **run_kw
+            )
+        )
+        getter: asyncio.Task = asyncio.create_task(progress_q.get())
+        pending: set[asyncio.Task] = {run_task, getter}
+        while True:
+            done, pending = await asyncio.wait(
+                pending, return_when=asyncio.FIRST_COMPLETED
+            )
+            if getter in done:
+                yield {"event": "status", "data": getter.result()}
+                getter = asyncio.create_task(progress_q.get())
+                pending = pending | {getter}
+            if run_task in done:
+                getter.cancel()
+                while not progress_q.empty():
+                    yield {"event": "status", "data": progress_q.get_nowait()}
+                break
+
         try:
-            result = await coordinator.run(context, catalog, query, Constraints(), **run_kw)
+            result = run_task.result()
         except Exception as e:
             logger.error("slow path failed", user_id=user_id, error=str(e))
             yield {"event": "error", "data": f"Analysis failed: {e}"}

src/agents/slow_path/coordinator.py

@@ -10,6 +10,8 @@ See AGENT_ARCHITECTURE_CONTEXT_new.md §5.2 / §6.1.
 
 from __future__ import annotations
 
+from collections.abc import Awaitable, Callable
+
 from ...catalog.models import Catalog
 from ..planner.contracts import BusinessContext, ToolRegistry
 from ..planner.inputs import Constraints
@@ -40,14 +42,24 @@ class SlowPathCoordinator:
         constraints: Constraints,
         planner_callbacks: list | None = None,
         assembler_callbacks: list | None = None,
+        progress: Callable[[str], Awaitable[None]] | None = None,
     ) -> AssembledOutput:
+        # `progress` (optional) surfaces per-stage status to the caller so a long
+        # slow-path run isn't a silent ~12s on the wire. Each stage is a single
+        # awaitable, so the most granular signal we can emit is at stage boundaries.
+        if progress:
+            await progress("Planning the analysis…")
         plan_kw = {"callbacks": planner_callbacks} if planner_callbacks else {}
         task_list = await self._planner.plan(
             context, catalog, self._registry, query, constraints, **plan_kw
         )
+        if progress:
+            await progress(f"Running {len(task_list.tasks)} analysis steps…")
         run_state = await self._task_runner.run(
             task_list, business_context_id=context.project_id
         )
+        if progress:
+            await progress("Composing the answer…")
         asm_kw = {"callbacks": assembler_callbacks} if assembler_callbacks else {}
         return await self._assembler.assemble(
             run_state, context, question=query, **asm_kw

src/api/v1/chat.py

@@ -199,6 +199,10 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
                     except Exception as e:
                         logger.error("save_messages failed", room_id=request.room_id, error=str(e))
                     yield event
+                elif event["event"] == "status":
+                    # slow-path progress ("Planning…", "Running N steps…"): forward
+                    # so the client shows activity and the SSE connection stays alive.
+                    yield event
                 elif event["event"] == "error":
                     yield event
                     return

src/query/executor/db.py

@@ -197,9 +197,9 @@ class DbExecutor(BaseExecutor):
         engine = user_engine_cache.get_engine(client_id, db_type, creds)
         if engine is not None:
             # Pooled, reused engine (postgres-like). Read-only + statement_timeout
-            # are pinned at connection establishment (connect_args options), so no
-            # per-query SET round-trips and no dispose — the connection returns to
-            # the pool warm for the next query.
+            # are set once per physical connection (connect event in UserEngineCache),
+            # so no per-query SET round-trips and no dispose — the connection returns
+            # to the pool warm for the next query.
             with engine.connect() as conn:
                 result = conn.execute(text(compiled.sql), compiled.params)
                 return list(result.keys()), [dict(row) for row in result.mappings()]
@@ -210,3 +210,38 @@ class DbExecutor(BaseExecutor):
             with eng.connect() as conn:
                 result = conn.execute(text(compiled.sql), compiled.params)
                 return list(result.keys()), [dict(row) for row in result.mappings()]
+
+    # ------------------------------------------------------------------
+    # Speculative pre-connect (DB3)
+    # ------------------------------------------------------------------
+
+    @classmethod
+    async def prewarm(cls, catalog: Catalog, user_id: str) -> None:
+        """Best-effort: warm pooled engines for the catalog's schema sources.
+
+        Called at slow-path entry so the TCP+TLS+auth handshake overlaps the ~4s
+        Planner LLM call — by the time `query_structured` runs, the connection is
+        already established. Warming is an optimization, never a requirement, so
+        this never raises and per-source failures are swallowed.
+        """
+        for source in catalog.sources:
+            if source.source_type != "schema":
+                continue
+            try:
+                client_id = cls._parse_client_id(source.location_ref)
+                client = await cls._fetch_client(client_id)
+                if client.user_id != user_id:
+                    continue
+                creds = decrypt_credentials_dict(client.credentials)
+                await asyncio.to_thread(cls._warm_sync, client_id, client.db_type, creds)
+            except Exception as exc:  # noqa: BLE001 — best-effort warming
+                logger.info("prewarm skipped", source_id=source.source_id, error=str(exc))
+
+    @staticmethod
+    def _warm_sync(client_id: str, db_type: str, creds: dict) -> None:
+        engine = user_engine_cache.get_engine(client_id, db_type, creds)
+        if engine is not None:
+            # Open + return a pooled physical connection: forces the handshake and
+            # runs the connect-event session SETs, leaving the pool warm.
+            with engine.connect():
+                pass