[KM-556] rewire chat stream endpoint to Phase 2 IntentRouter + AnswerAgent + QueryService

PROGRESS.md

@@ -2,7 +2,7 @@
 
 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-05-08 (item 41 done; item 16 done; item 31 done; item 35 done — upload endpoint wired to on_tabular_uploaded)
+**Last updated**: 2026-05-08 (item 41 done; item 16 done; item 31 done; item 35 done; item 36 done — chat endpoint rewired to Phase 2 QueryService)
 **Current open PR**: none — all Phase 2 contracts shipped on `pr/1`. Cleanup PR pending (API rewiring + Phase 1 removal).
 
 ---
@@ -119,7 +119,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 |---|---|---|---|---|
 | 34 | DB client endpoints (`api/v1/db_client.py`) | DB | `[ ]` | Phase 1 endpoint exists — rewire `/ingest` to call `pipeline.triggers.on_db_registered`. Trigger is ready as of PR2a; deferred to a later PR until both teammates ack. |
 | 35 | Document/tabular upload endpoints (`api/v1/document.py`) | TAB | `[x]` | Rewired `/document/process` — after processing CSV/XLSX, calls `on_tabular_uploaded(document_id, user_id)`. Catalog ingestion failure is logged but does not fail the request (document already ingested to vector store). |
-| 36 | Chat stream endpoint (`api/v1/chat.py`) | B | `[ ]` | Phase 2 handler module ready (`agents/chat_handler.py`); rewiring of the actual `/chat/stream` endpoint deferred to cleanup PR to avoid breaking Phase 1 during the migration. |
+| 36 | Chat stream endpoint (`api/v1/chat.py`) | B | `[x]` | Rewired `/chat/stream` — replaced `query_executor.execute()` (Phase 1) with `CatalogReader + QueryService` (Phase 2). Kept Phase 1 structure: Redis cache, message persistence, fast intent, orchestrator, retriever, chatbot. Only query execution block swapped. |
 | 37 | Room / users endpoints (`api/v1/room.py`, `api/v1/users.py`) | B | `[ ]` | No catalog work; only touch if auth flow changes |
 
 ### Tests + eval

src/api/v1/chat.py

@@ -1,17 +1,17 @@
 """Chat endpoint with streaming support."""
 
-import asyncio
 import uuid
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
 from src.db.postgres.connection import get_db
 from src.db.postgres.models import ChatMessage, MessageSource
-from src.agents.orchestration import orchestrator
-from src.agents.chatbot import chatbot
+from src.agents.intent_router import IntentRouter
+from src.agents.answer_agent import AnswerAgent, DocumentChunk
 from src.retrieval.router import retrieval_router as retriever
 from src.retrieval.base import RetrievalResult
-from src.query.query_executor import query_executor
-from src.query.base import QueryResult
+from src.catalog.reader import CatalogReader
+from src.catalog.store import CatalogStore
+from src.query.service import QueryService
 from src.db.redis.connection import get_redis
 from src.config.settings import settings
 from src.middlewares.logging import get_logger, log_execution
@@ -26,17 +26,16 @@ _GREETINGS = frozenset(["hi", "hello", "hey", "halo", "hai", "hei"])
 _GOODBYES = frozenset(["bye", "goodbye", "thanks", "thank you", "terima kasih", "sampai jumpa"])
 
 
-def _fast_intent(message: str) -> Optional[dict]:
-    """Bypass LLM orchestrator for obvious greetings and farewells."""
+def _fast_intent(message: str) -> Optional[str]:
+    """Return a direct response string for obvious greetings/farewells, else None."""
     lower = message.lower().strip().rstrip("!.,?")
     if lower in _GREETINGS:
-        return {"intent": "greeting", "needs_search": False,
-                "direct_response": "Hello! How can I assist you today?", "search_query": ""}
+        return "Hello! How can I assist you today?"
     if lower in _GOODBYES:
-        return {"intent": "goodbye", "needs_search": False,
-                "direct_response": "Goodbye! Have a great day!", "search_query": ""}
+        return "Goodbye! Have a great day!"
     return None
 
+
 logger = get_logger("chat_api")
 
 router = APIRouter(prefix="/api/v1", tags=["Chat"])
@@ -48,18 +47,6 @@ class ChatRequest(BaseModel):
     message: str
 
 
-def _format_context(results: List[RetrievalResult]) -> str:
-    """Format retrieval results as context string for the LLM."""
-    lines = []
-    for result in results:
-        data = result.metadata.get("data", {})
-        filename = data.get("filename", "Unknown")
-        page = data.get("page_label")
-        source_label = f"{filename}, p.{page}" if page else filename
-        lines.append(f"[Source: {source_label}]\n{result.content}\n")
-    return "\n".join(lines)
-
-
 def _extract_sources(results: List[RetrievalResult]) -> List[Dict[str, Any]]:
     """Extract deduplicated source references from retrieval results."""
     seen = set()
@@ -87,25 +74,22 @@ def _extract_sources(results: List[RetrievalResult]) -> List[Dict[str, Any]]:
                     "filename": data.get("table_name", "Unknown"),
                     "page_label": data.get("column_name", "Unknown"),
                 })
-
     logger.debug(f"Extracted sources: {sources}")
     return sources
 
 
-def _format_query_results(results: list[QueryResult]) -> str:
-    if not results:
-        return ""
-    lines = []
+def _to_document_chunks(results: List[RetrievalResult]) -> List[DocumentChunk]:
+    """Convert Phase 1 RetrievalResult list to Phase 2 DocumentChunk list."""
+    chunks = []
     for r in results:
-        name = r.metadata.get("client_name", r.source_id)
-        lines.append(f"[Query result — {name}, tables: {r.table_or_file}]")
-        lines.append(f"SQL: {r.metadata.get('sql', '')}")
-        if r.columns and r.rows:
-            lines.append(" | ".join(r.columns))
-            for row in r.rows[:20]:
-                lines.append(" | ".join(str(row.get(c, "")) for c in r.columns))
-        lines.append(f"({r.row_count} rows total)\n")
-    return "\n".join(lines)
+        data = r.metadata.get("data", {})
+        page = data.get("page_label")
+        chunks.append(DocumentChunk(
+            content=r.content,
+            filename=data.get("filename"),
+            page_label=str(page) if page is not None else None,
+        ))
+    return chunks
 
 
 async def get_cached_response(redis, cache_key: str) -> Optional[str]:
@@ -168,8 +152,9 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
       3. done     — signals end of stream
     """
     redis = await get_redis()
-
     cache_key = f"{settings.redis_prefix}chat:{request.room_id}:{request.message}"
+
+    # Redis cache hit
     cached = await get_cached_response(redis, cache_key)
     if cached:
         logger.info("Returning cached response")
@@ -183,91 +168,56 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
         return EventSourceResponse(stream_cached())
 
     try:
-        # Step 1: Fast local intent check (skips LLM for greetings/farewells)
-        intent_result = _fast_intent(request.message)
-
-        context = ""
-        sources: List[Dict[str, Any]] = []
-
-        if intent_result is None:
-            # Step 2: Launch retrieval and history loading in parallel, then run orchestrator.
-            # k=5
-            # tables — db_executor's FK expansion is one-hop and cannot bridge
-            # 2-hop gaps (e.g. customers -> order_items -> products) on its own.
-            retrieval_task = asyncio.create_task(
-                retriever.retrieve(request.message, request.user_id, db, k=5)
-            )
-            history_task = asyncio.create_task(
-                load_history(db, request.room_id, limit=6)  # 6 msgs (3 pairs) for orchestrator
-            )
-            history = await history_task  # fast DB query (<100ms), done before orchestrator finishes
-            intent_result = await orchestrator.analyze_message(request.message, history)
-
-            search_query = intent_result.get("search_query", request.message) or request.message
-            if not intent_result.get("needs_search"):
-                retrieval_task.cancel()
-                try:
-                    await retrieval_task
-                except asyncio.CancelledError:
-                    pass
-                raw_results = []
-            else:
-                logger.info(f"Searching for: {search_query}")
-                if search_query != request.message:
-                    retrieval_task.cancel()
-                    try:
-                        await retrieval_task
-                    except asyncio.CancelledError:
-                        pass
-                    raw_results = await retriever.retrieve(
-                        query=search_query,
-                        user_id=request.user_id,
-                        db=db,
-                        k=5,
-                        source_hint=intent_result.get("source_hint", "both"),
-                    )
-                else:
-                    raw_results = await retrieval_task
-
-            context = _format_context(raw_results)
-            sources = _extract_sources(raw_results)
-
-            source_hint = intent_result.get("source_hint", "both")
-            if source_hint in ("schema", "both"):
-                # Use search_query (orchestrator's standalone rewrite) so follow-up
-                # messages like "dive deeper" or "show me last year" resolve correctly.
-                # For first-turn questions search_query == request.message, so no change.
-                query_results = await query_executor.execute(
-                    results=raw_results,
-                    user_id=request.user_id,
-                    db=db,
-                    question=search_query,
-                )
-                query_context = _format_query_results(query_results)
-                if query_context:
-                    context = query_context + "\n\n" + context
-
-        # Step 3: Direct response for greetings / non-document intents
-        if intent_result.get("direct_response"):
-            response = intent_result["direct_response"]
-            await cache_response(redis, cache_key, response)
-            await save_messages(db, request.room_id, request.message, response, sources=[])
+        # Fast intent: greetings/farewells bypass LLM entirely
+        direct = _fast_intent(request.message)
+        if direct:
+            await cache_response(redis, cache_key, direct)
+            await save_messages(db, request.room_id, request.message, direct, sources=[])
 
             async def stream_direct():
                 yield {"event": "sources", "data": json.dumps([])}
-                yield {"event": "message", "data": response}
+                yield {"event": "message", "data": direct}
 
             return EventSourceResponse(stream_direct())
 
-        # Step 4: Stream answer token-by-token as LLM generates it
-        # Load full history (10 msgs) for chatbot — richer context than the 6 used by orchestrator
+        # Load history for intent routing
+        history = await load_history(db, request.room_id, limit=6)
+
+        # Phase 2: IntentRouter classifies message
+        decision = await IntentRouter().classify(request.message, history)
+        rewritten = decision.rewritten_query or request.message
+
+        query_result = None
+        chunks: List[DocumentChunk] | None = None
+        sources: List[Dict[str, Any]] = []
+
+        if decision.source_hint == "structured":
+            catalog = await CatalogReader(CatalogStore()).read(request.user_id, "structured")
+            query_result = await QueryService().run(request.user_id, rewritten, catalog)
+
+        elif decision.source_hint == "unstructured":
+            raw_results = await retriever.retrieve(
+                query=rewritten,
+                user_id=request.user_id,
+                db=db,
+                k=5,
+            )
+            chunks = _to_document_chunks(raw_results)
+            sources = _extract_sources(raw_results)
+
+        # Load full history for answer generation
         full_history = await load_history(db, request.room_id, limit=10)
-        messages = full_history + [HumanMessage(content=request.message)]
 
+        # Phase 2: AnswerAgent streams answer tokens
         async def stream_response():
             full_response = ""
             yield {"event": "sources", "data": json.dumps(sources)}
-            async for token in chatbot.astream_response(messages, context):
+            async for token in AnswerAgent().astream(
+                request.message,
+                history=full_history,
+                query_result=query_result,
+                chunks=chunks,
+            ):
                 full_response += token
                 yield {"event": "chunk", "data": token}
             yield {"event": "done", "data": ""}