[KM-626][AI] Slow-path wiring prep: context + persistence seams, retrieve_documents few-shot

Three small steps that de-couple the live wiring from the remaining blockers so
flipping enable_slow_path on later is a one-liner.

- BusinessContext seam: new get_business_context(user_id) (planner/business_context.py)
is the single place context is read — returns a stub today, swap the body when the
lead's real source lands. ChatHandler now reads through it.
- AnalysisStore seam: new slow_path/store.py (Protocol + NullAnalysisStore no-op that
logs only). ChatHandler persists the analysis_record through it after streaming
(never breaks the answer on failure). Real Postgres-backed store in the catalog DB
(Neon dataeyond) is a TODO — table not created yet.
- Planner few-shot: add Example C (retrieve_documents) — a mixed structured +
unstructured plan showing an independent document-retrieval branch (NL query, no
${t} placeholder, runs in parallel). The Planner had no retrieve_documents example.

Verified: slow_path + planner suites green (52 passed); live RUN_PLANNER_EVAL still
passes with the new few-shot (no regression).

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

src/agents/chat_handler.py

@@ -38,8 +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
+    from .slow_path.store import AnalysisStore
 
 logger = get_logger("chat_handler")
 
@@ -69,6 +69,7 @@ class ChatHandler:
         slow_path_coordinator_factory: (
             Callable[[str], SlowPathCoordinator] | None
         ) = None,
+        analysis_store: AnalysisStore | None = None,
     ) -> None:
         self._intent_router = intent_router
         self._answer_agent = answer_agent
@@ -76,11 +77,12 @@ class ChatHandler:
         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.
+        # gated until the lead's real BusinessContext lands. When True, `structured`
+        # intents route here instead of the single-query QueryService path. The
+        # factory + store are injectable for tests.
         self._enable_slow_path = enable_slow_path
         self._slow_path_factory = slow_path_coordinator_factory
+        self._analysis_store = analysis_store
 
     # ------------------------------------------------------------------
     # Lazy default-dep builders
@@ -238,6 +240,13 @@ class ChatHandler:
             PlannerService(), TaskRunner(invoker, registry), Assembler(), registry
         )
 
+    def _get_analysis_store(self) -> AnalysisStore:
+        if self._analysis_store is None:
+            from .slow_path.store import NullAnalysisStore
+
+            self._analysis_store = NullAnalysisStore()
+        return self._analysis_store
+
     async def _run_slow_path(
         self,
         user_id: str,
@@ -246,15 +255,16 @@ class ChatHandler:
     ) -> 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).
+        Context comes from the `get_business_context` seam (a stub today); the
+        `analysis_record` is persisted via the `AnalysisStore` seam (a no-op today).
+        `chat_answer` is emitted as a single `chunk` (the Assembler returns the whole
+        object — true token streaming is a later step).
         """
+        from .planner.business_context import get_business_context
         from .planner.inputs import Constraints
 
         coordinator = self._get_slow_path_coordinator(user_id)
-        context = _stub_business_context(user_id)
+        context = await get_business_context(user_id)
         try:
             result = await coordinator.run(context, catalog, query, Constraints())
         except Exception as e:
@@ -264,28 +274,13 @@ class ChatHandler:
 
         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.
+        try:
+            await self._get_analysis_store().save(result.analysis_record)
+        except Exception as e:  # persistence must never break the user's answer
+            logger.error("analysis_record persist failed", user_id=user_id, error=str(e))
         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,
     user_id: str,

src/agents/planner/business_context.py

@@ -0,0 +1,31 @@
+"""BusinessContext reader — the single seam the slow path reads context through.
+
+`get_business_context(user_id)` is the one place the live flow obtains a
+`BusinessContext`. Today it returns a minimal STUB so the slow path runs end to
+end; when the lead's real Business Understanding source lands, swap the body here
+(read the interview / stored context) and nothing upstream changes.
+
+See AGENT_ARCHITECTURE_CONTEXT_new.md §7.1.
+"""
+
+from __future__ import annotations
+
+from .contracts import BusinessContext
+
+
+async def get_business_context(user_id: str) -> BusinessContext:
+    """Return the user's BusinessContext.
+
+    STUB until the lead's real source lands. `project_id` flows through as
+    `RunState.business_context_id`. Async so the real implementation (a DB / store
+    read) fits without changing this signature.
+
+    TODO(lead): replace the body with the real read (Business Understanding store).
+    """
+    return BusinessContext(
+        project_id=user_id,
+        industry="unknown",
+        completeness="partial",
+        business_description="(not yet captured — BusinessContext source pending)",
+        scale_and_scope="(unknown)",
+    )

src/agents/planner/examples.py

@@ -191,9 +191,120 @@ _EXAMPLE_B = TaskList(
 )
 
 
+# --------------------------------------------------------------------------- #
+# Example C — mixed structured + unstructured.
+# "Revenue dipped in Q1 — what happened?"
+# Shows: a structured branch (query -> analyze_trend) runs alongside an
+# INDEPENDENT retrieve_documents branch that pulls qualitative context. Note
+# retrieve_documents takes a natural-language `query` (NOT a `${t<id>}` data
+# placeholder — it is a source, not a consumer) and can run in parallel; the
+# Assembler folds the document context into the explanation.
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_C = TaskList(
+    plan_id="example_c",
+    goal_restated="Explain Q1's revenue dip using both the numbers and the qualitative record.",
+    assumptions=["'Q1' = 2026-01-01 to 2026-03-31."],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes order date and revenue.",
+            tool_calls=[ToolCall(tool="describe_source", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria="describe_source returns the orders table with date and revenue.",
+            depends_on=[],
+            parallelizable_with=["t4"],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Pull Q1 order dates and revenue.",
+            tool_calls=[
+                ToolCall(
+                    tool="query_structured",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {
+                                    "kind": "column",
+                                    "column_id": "c_order_date",
+                                    "alias": "order_date",
+                                },
+                                {"kind": "column", "column_id": "c_revenue", "alias": "revenue"},
+                            ],
+                            "filters": [
+                                {
+                                    "column_id": "c_order_date",
+                                    "op": "between",
+                                    "value": ["2026-01-01", "2026-03-31"],
+                                    "value_type": "date",
+                                }
+                            ],
+                            "limit": 10000,
+                        }
+                    },
+                )
+            ],
+            expected_output="q1_rows",
+            success_criteria="Produced Q1 order rows with date and revenue.",
+            depends_on=["t1"],
+            parallelizable_with=[],
+            estimated_cost="medium",
+        ),
+        Task(
+            id="t3",
+            stage="evaluation",
+            objective="Summarize the Q1 monthly revenue trend to locate the dip.",
+            tool_calls=[
+                ToolCall(
+                    tool="analyze_trend",
+                    args={
+                        "data": "${t2}",
+                        "date_column": "order_date",
+                        "value_column": "revenue",
+                        "freq": "month",
+                        "agg": "sum",
+                    },
+                )
+            ],
+            expected_output="q1_trend",
+            success_criteria="Produced a per-month revenue series showing where revenue fell.",
+            depends_on=["t2"],
+            parallelizable_with=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t4",
+            stage="data_understanding",
+            objective="Retrieve qualitative context on Q1 operational events behind a dip.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_documents",
+                    args={
+                        "query": "operational issues, outages, or notable events in Q1 2026",
+                        "top_k": 5,
+                    },
+                )
+            ],
+            expected_output="q1_context_chunks",
+            success_criteria="Produced relevant document chunks about Q1 operations.",
+            depends_on=[],
+            parallelizable_with=["t1"],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+
 EXAMPLES: list[tuple[str, TaskList]] = [
     ("Which product categories drove last quarter's revenue?", _EXAMPLE_A),
     ("How has monthly revenue trended by region this year, and what's unusual?", _EXAMPLE_B),
+    ("Revenue dipped in Q1 — what happened?", _EXAMPLE_C),
 ]
 
 

src/agents/slow_path/store.py

@@ -0,0 +1,44 @@
+"""AnalysisStore — the seam the slow path persists its AnalysisRecord through.
+
+The Assembler produces an `AnalysisRecord` (the faithful, structured record of a
+run — §8.3, INV-4). Persisting it is a separate concern from streaming the answer,
+so it sits behind this one-method seam.
+
+`NullAnalysisStore` is the default: it logs that a record was produced but stores
+nothing, because the backing table does not exist yet. The plan is to store records
+in the **same catalog DB** (Neon `dataeyond`, `settings.postgres_connstring`).
+
+TODO(persistence): add a Postgres-backed `AnalysisStore` writing an
+`analysis_records` table in the catalog DB, keyed on
+(business_context_id, plan_id, created_at), then inject it into ChatHandler.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from src.middlewares.logging import get_logger
+
+from .schemas import AnalysisRecord
+
+logger = get_logger("analysis_store")
+
+
+@runtime_checkable
+class AnalysisStore(Protocol):
+    """Persist a completed analysis. Implementations must never raise on the
+    caller's path — a persistence failure must not break the user's answer."""
+
+    async def save(self, record: AnalysisRecord) -> None: ...
+
+
+class NullAnalysisStore:
+    """Default no-op store: logs the record, persists nothing (no table yet)."""
+
+    async def save(self, record: AnalysisRecord) -> None:
+        logger.info(
+            "analysis_record produced (not persisted — no store configured)",
+            plan_id=record.plan_id,
+            business_context_id=record.business_context_id,
+            n_tasks=len(record.tasks_run),
+        )