[KM-626][AI] ChatHandler: gated structured->slow-path wiring

Add an off-by-default slow analytical path to ChatHandler. When
enable_slow_path=True, a `structured` intent routes to SlowPathCoordinator
(Planner -> TaskRunner -> Assembler) instead of the single-query QueryService
path, streaming the assembled chat_answer as SSE.

- New keyword-only ctor params: enable_slow_path (default False) and an
injectable slow_path_coordinator_factory for tests.
- _get_slow_path_coordinator is the composition root: builds a per-request
CompositeToolInvoker (data-access needs user_id + CatalogReader) so the
slow_path/ package stays tool-agnostic (INV-7).
- Flag defaults off and chat.py constructs ChatHandler() unchanged, so live
behavior is identical.

Gated on two stubs before flipping on: _stub_business_context (-> lead's real
BusinessContext) and analysis_record persistence (no store yet).

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

src/agents/chat_handler.py

@@ -23,7 +23,7 @@ inject mocks).
 from __future__ import annotations
 
 import json
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from typing import TYPE_CHECKING, Any
 
 from langchain_core.messages import BaseMessage
@@ -38,6 +38,8 @@ if TYPE_CHECKING:
     from ..catalog.reader import CatalogReader
     from ..query.service import QueryService
     from ..retrieval.router import RetrievalRouter
+    from .planner.contracts import BusinessContext
+    from .slow_path.coordinator import SlowPathCoordinator
 
 logger = get_logger("chat_handler")
 
@@ -62,12 +64,23 @@ class ChatHandler:
         catalog_reader: CatalogReader | None = None,
         query_service: QueryService | None = None,
         document_retriever: RetrievalRouter | None = None,
+        *,
+        enable_slow_path: bool = False,
+        slow_path_coordinator_factory: (
+            Callable[[str], SlowPathCoordinator] | None
+        ) = None,
     ) -> None:
         self._intent_router = intent_router
         self._answer_agent = answer_agent
         self._catalog_reader = catalog_reader
         self._query_service = query_service
         self._document_retriever = document_retriever
+        # Slow analytical path (Planner -> TaskRunner -> Assembler). OFF by default:
+        # gated until the lead's real BusinessContext + analysis_record persistence
+        # land. When True, `structured` intents route here instead of the single-query
+        # QueryService path. The factory is injectable for tests.
+        self._enable_slow_path = enable_slow_path
+        self._slow_path_factory = slow_path_coordinator_factory
 
     # ------------------------------------------------------------------
     # Lazy default-dep builders
@@ -134,6 +147,10 @@ class ChatHandler:
         if decision.source_hint == "structured":
             try:
                 catalog = await self._get_catalog_reader().read(user_id, "structured")
+                if self._enable_slow_path:
+                    async for event in self._run_slow_path(user_id, rewritten, catalog):
+                        yield event
+                    return
                 query_result = await self._get_query_service().run(
                     user_id, rewritten, catalog
                 )
@@ -189,6 +206,85 @@ class ChatHandler:
 
         yield {"event": "done", "data": ""}
 
+    # ------------------------------------------------------------------
+    # Slow analytical path (gated, off by default)
+    # ------------------------------------------------------------------
+
+    def _get_slow_path_coordinator(self, user_id: str) -> SlowPathCoordinator:
+        """Build the per-request slow-path coordinator (composition root).
+
+        The data-access tools need the authenticated `user_id` + `CatalogReader`,
+        so the `CompositeToolInvoker` is constructed per request. The slow-path
+        agent code stays tool-agnostic (INV-7) — only here, the composition root,
+        do we name concrete tool implementations.
+        """
+        if self._slow_path_factory is not None:
+            return self._slow_path_factory(user_id)
+
+        from ..tools.data_access import DataAccessToolInvoker
+        from ..tools.invoker import AnalyticsToolInvoker, CompositeToolInvoker
+        from .planner.registry import default_registry
+        from .planner.service import PlannerService
+        from .slow_path.assembler import Assembler
+        from .slow_path.coordinator import SlowPathCoordinator
+        from .slow_path.task_runner import TaskRunner
+
+        invoker = CompositeToolInvoker(
+            DataAccessToolInvoker(user_id, self._get_catalog_reader()),
+            AnalyticsToolInvoker(),
+        )
+        registry = default_registry()
+        return SlowPathCoordinator(
+            PlannerService(), TaskRunner(invoker, registry), Assembler(), registry
+        )
+
+    async def _run_slow_path(
+        self,
+        user_id: str,
+        query: str,
+        catalog: Any,
+    ) -> AsyncIterator[dict[str, Any]]:
+        """Run the slow path and stream its assembled answer as SSE events.
+
+        STUB `BusinessContext` until the lead's real source lands; `analysis_record`
+        persistence is deferred (no store yet). `chat_answer` is emitted as a single
+        `chunk` (the Assembler returns the whole object — true token streaming is a
+        later step).
+        """
+        from .planner.inputs import Constraints
+
+        coordinator = self._get_slow_path_coordinator(user_id)
+        context = _stub_business_context(user_id)
+        try:
+            result = await coordinator.run(context, catalog, query, Constraints())
+        except Exception as e:
+            logger.error("slow path failed", user_id=user_id, error=str(e))
+            yield {"event": "error", "data": f"Analysis failed: {e}"}
+            return
+
+        yield {"event": "sources", "data": json.dumps([])}  # TODO: derive from record
+        yield {"event": "chunk", "data": result.chat_answer}
+        # TODO(persistence): persist result.analysis_record once a memory store exists.
+        yield {"event": "done", "data": ""}
+
+
+def _stub_business_context(user_id: str) -> BusinessContext:
+    """Minimal stand-in BusinessContext until the lead's real source lands.
+
+    The slow path requires a BusinessContext; `project_id` flows through as
+    `RunState.business_context_id`. TODO(lead): replace with a real
+    `get_business_context(user_id)` reader.
+    """
+    from .planner.contracts import BusinessContext
+
+    return BusinessContext(
+        project_id=user_id,
+        industry="unknown",
+        completeness="partial",
+        business_description="(not yet captured — BusinessContext source pending)",
+        scale_and_scope="(unknown)",
+    )
+
 
 def _build_sources(
     source_hint: str,