[NOTICKET] cleanup PR: wire ChatHandler, drop Phase 1 remnants, implement catalog rebuild

- wire chat.py to ChatHandler; remove inline intent routing, retrieval, and answer generation
- drop Phase 1 dual-write in db_client /ingest: only on_db_registered runs now
- implement on_catalog_rebuild_requested + POST /api/v1/data-catalog/rebuild endpoint
- delete dead modules: agents/chatbot (Phase 1), pipeline/orchestrator, query/base, api/v1/knowledge, config/agents/
- rename answer_agent.py -> chatbot.py, AnswerAgent -> ChatbotAgent
- fix ChatHandler to use RetrievalRouter (restores Redis retrieval cache)
- remove dead _build_csv_documents/_build_excel_documents from processing_service
- add top_values to ColumnStats for low-cardinality columns (query planner context)
- replace deprecated @app .on_event("startup") with lifespan context manager

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

main.py

@@ -1,5 +1,7 @@
 """Main application entry point."""
 
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from src.middlewares.logging import configure_logging, get_logger
 from src.middlewares.cors import add_cors_middleware
@@ -9,7 +11,6 @@ from src.api.v1.document import router as document_router
 from src.api.v1.chat import router as chat_router
 from src.api.v1.room import router as room_router
 from src.api.v1.users import router as users_router
-from src.api.v1.knowledge import router as knowledge_router
 from src.api.v1.db_client import router as db_client_router
 from src.api.v1.data_catalog import router as data_catalog_router
 from src.db.postgres.init_db import init_db
@@ -19,11 +20,21 @@ import uvicorn
 configure_logging()
 logger = get_logger("main")
 
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logger.info("Starting application...")
+    await init_db()
+    logger.info("Database initialized")
+    yield
+
+
 # Create FastAPI app
 app = FastAPI(
     title="DataEyond Agentic Service",
     description="Multi-agent AI backend with RAG capabilities",
-    version="0.1.0"
+    version="0.1.0",
+    lifespan=lifespan,
 )
 
 # Add middleware
@@ -34,21 +45,12 @@ app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 # Include routers
 app.include_router(users_router)
 app.include_router(document_router)
-app.include_router(knowledge_router)
 app.include_router(room_router)
 app.include_router(chat_router)
 app.include_router(db_client_router)
 app.include_router(data_catalog_router)
 
 
-@app.on_event("startup")
-async def startup_event():
-    """Initialize database on startup."""
-    logger.info("Starting application...")
-    await init_db()
-    logger.info("Database initialized")
-
-
 @app.get("/")
 async def root():
     """Root endpoint."""

src/agents/answer_agent.py

@@ -1,170 +0,0 @@
-"""AnswerAgent — final answer formation. Phase 2 chatbot.
-
-Receives one of:
-  - a `QueryResult` (structured query path),
-  - a list of document chunks (unstructured path), or
-  - nothing (chat-only path: greeting, farewell, meta question).
-
-Streams the answer token-by-token so the chat handler can wrap each token
-into an SSE event. Conversation history is supported.
-
-Lives at `agents/answer_agent.py` rather than `agents/chatbot.py` to avoid
-colliding with the Phase 1 chatbot still imported by the legacy chat
-endpoint. PR7 cleanup will rename this to `chatbot.py` after Phase 1's
-chat endpoint is rewired to call this through `agents/chat_handler.py`.
-"""
-
-from __future__ import annotations
-
-from collections.abc import AsyncIterator
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any
-
-from langchain_core.messages import BaseMessage
-from langchain_core.output_parsers import StrOutputParser
-from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
-from langchain_core.runnables import Runnable
-from langchain_openai import AzureChatOpenAI
-
-from src.middlewares.logging import get_logger
-
-from ..query.executor.base import QueryResult
-
-logger = get_logger("answer_agent")
-
-
-_PROMPT_DIR = Path(__file__).resolve().parent.parent / "config" / "prompts"
-_SYSTEM_PROMPT_PATH = _PROMPT_DIR / "chatbot_system.md"
-_GUARDRAILS_PATH = _PROMPT_DIR / "guardrails.md"
-
-
-@dataclass
-class DocumentChunk:
-    """One retrieved document chunk for the unstructured path."""
-
-    content: str
-    filename: str | None = None
-    page_label: str | None = None
-
-
-def _load_system_prompt() -> str:
-    """Compose system prompt = chatbot_system.md + guardrails.md.
-
-    Guardrails appended last so they take precedence in conflict (matches
-    the docstring at the top of guardrails.md).
-    """
-    chatbot = _SYSTEM_PROMPT_PATH.read_text(encoding="utf-8")
-    guardrails = _GUARDRAILS_PATH.read_text(encoding="utf-8")
-    return f"{chatbot}\n\n{guardrails}"
-
-
-def _format_query_result(qr: QueryResult) -> str:
-    """Render a QueryResult as a compact context block for the LLM."""
-    if qr.error:
-        return (
-            f"[Query result — FAILED]\n"
-            f"source_id={qr.source_id}\n"
-            f"error: {qr.error}"
-        )
-    lines: list[str] = [
-        "[Query result]",
-        f"source_id: {qr.source_id}",
-        f"backend: {qr.backend}",
-        f"row_count: {qr.row_count}"
-        + (" (truncated)" if qr.truncated else ""),
-        f"elapsed_ms: {qr.elapsed_ms}",
-    ]
-    if qr.rows:
-        # Cap rendering at 25 rows; the LLM doesn't need the full set
-        cap = min(len(qr.rows), 25)
-        columns = list(qr.rows[0].keys())
-        lines.append("columns: " + ", ".join(columns))
-        lines.append("rows:")
-        for row in qr.rows[:cap]:
-            lines.append("  " + ", ".join(f"{k}={row[k]!r}" for k in columns))
-        if cap < len(qr.rows):
-            lines.append(f"  ... (+{len(qr.rows) - cap} more rows omitted from prompt)")
-    return "\n".join(lines)
-
-
-def _format_document_chunks(chunks: list[DocumentChunk]) -> str:
-    if not chunks:
-        return ""
-    blocks: list[str] = []
-    for c in chunks:
-        label_parts = [p for p in (c.filename, c.page_label) if p]
-        label = ", ".join(label_parts) if label_parts else "Unknown source"
-        blocks.append(f"[Source: {label}]\n{c.content}")
-    return "\n\n".join(blocks)
-
-
-def _build_context_block(
-    query_result: QueryResult | None,
-    chunks: list[DocumentChunk] | None,
-) -> str:
-    parts: list[str] = []
-    if query_result is not None:
-        parts.append(_format_query_result(query_result))
-    if chunks:
-        parts.append(_format_document_chunks(chunks))
-    return "\n\n".join(parts) if parts else "(no data context — answer conversationally)"
-
-
-def _build_default_chain() -> Runnable:
-    from src.config.settings import settings
-
-    llm = AzureChatOpenAI(
-        azure_deployment=settings.azureai_deployment_name_4o,
-        openai_api_version=settings.azureai_api_version_4o,
-        azure_endpoint=settings.azureai_endpoint_url_4o,
-        api_key=settings.azureai_api_key_4o,
-        temperature=0.3,
-    )
-    prompt = ChatPromptTemplate.from_messages(
-        [
-            ("system", _load_system_prompt()),
-            MessagesPlaceholder(variable_name="history", optional=True),
-            ("human", "{message}"),
-            ("system", "Data context for this turn:\n\n{context}"),
-        ]
-    )
-    return prompt | llm | StrOutputParser()
-
-
-class AnswerAgent:
-    """Formats and streams the final user-facing answer.
-
-    `chain` is injectable: tests pass a fake that yields canned tokens.
-    Default constructs the production Azure OpenAI streaming chain on
-    first use.
-    """
-
-    def __init__(self, chain: Runnable | None = None) -> None:
-        self._chain = chain
-
-    def _ensure_chain(self) -> Runnable:
-        if self._chain is None:
-            self._chain = _build_default_chain()
-        return self._chain
-
-    async def astream(
-        self,
-        message: str,
-        history: list[BaseMessage] | None = None,
-        query_result: QueryResult | None = None,
-        chunks: list[DocumentChunk] | None = None,
-    ) -> AsyncIterator[str]:
-        """Stream tokens of the final answer.
-
-        Caller wraps each token into the SSE format. Empty `history` and
-        no context = pure chat reply.
-        """
-        chain = self._ensure_chain()
-        payload: dict[str, Any] = {
-            "message": message,
-            "history": history or [],
-            "context": _build_context_block(query_result, chunks),
-        }
-        async for token in chain.astream(payload):
-            yield token

src/agents/chat_handler.py

@@ -4,11 +4,11 @@ End-to-end flow per user message:
 
   1. `IntentRouter.classify` → `chat` / `unstructured` / `structured`.
   2. Route:
-       - `chat`         → no context. Pass straight to AnswerAgent.
+       - `chat`         → no context. Pass straight to ChatbotAgent.
        - `structured`   → CatalogReader → QueryService → QueryResult.
        - `unstructured` → DocumentRetriever (placeholder, raises until TAB
                           ships) → list[DocumentChunk].
-  3. `AnswerAgent.astream` → yield text tokens.
+  3. `ChatbotAgent.astream` → yield text tokens.
   4. Wrap each step into an SSE-style event dict so the API endpoint can
      stream them as Server-Sent Events.
 
@@ -30,13 +30,13 @@ from langchain_core.messages import BaseMessage
 from src.middlewares.logging import get_logger
 
 from src.retrieval.base import RetrievalResult
-from .answer_agent import AnswerAgent, DocumentChunk
+from .chatbot import ChatbotAgent, DocumentChunk
 from .intent_router import IntentRouter
 
 if TYPE_CHECKING:
     from ..catalog.reader import CatalogReader
     from ..query.service import QueryService
-    from ..retrieval.document import DocumentRetriever
+    from ..retrieval.router import RetrievalRouter
 
 logger = get_logger("chat_handler")
 
@@ -55,10 +55,10 @@ class ChatHandler:
     def __init__(
         self,
         intent_router: IntentRouter | None = None,
-        answer_agent: AnswerAgent | None = None,
+        answer_agent: ChatbotAgent | None = None,
         catalog_reader: CatalogReader | None = None,
         query_service: QueryService | None = None,
-        document_retriever: DocumentRetriever | None = None,
+        document_retriever: RetrievalRouter | None = None,
     ) -> None:
         self._intent_router = intent_router
         self._answer_agent = answer_agent
@@ -75,9 +75,9 @@ class ChatHandler:
             self._intent_router = IntentRouter()
         return self._intent_router
 
-    def _get_answer_agent(self) -> AnswerAgent:
+    def _get_answer_agent(self) -> ChatbotAgent:
         if self._answer_agent is None:
-            self._answer_agent = AnswerAgent()
+            self._answer_agent = ChatbotAgent()
         return self._answer_agent
 
     def _get_catalog_reader(self) -> CatalogReader:
@@ -95,11 +95,11 @@ class ChatHandler:
             self._query_service = QueryService()
         return self._query_service
 
-    def _get_document_retriever(self) -> DocumentRetriever:
+    def _get_document_retriever(self) -> RetrievalRouter:
         if self._document_retriever is None:
-            from ..retrieval.document import DocumentRetriever
+            from ..retrieval.router import RetrievalRouter
 
-            self._document_retriever = DocumentRetriever()
+            self._document_retriever = RetrievalRouter()
         return self._document_retriever
 
     # ------------------------------------------------------------------

src/agents/chatbot.py

@@ -1,85 +1,165 @@
-"""Chatbot agent with RAG capabilities."""
+"""ChatbotAgent — final answer formation. Phase 2 chatbot.
 
-import tiktoken
-from langchain_openai import AzureChatOpenAI
-from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+Receives one of:
+  - a `QueryResult` (structured query path),
+  - a list of document chunks (unstructured path), or
+  - nothing (chat-only path: greeting, farewell, meta question).
+
+Streams the answer token-by-token so the chat handler can wrap each token
+into an SSE event. Conversation history is supported.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from langchain_core.messages import BaseMessage
 from langchain_core.output_parsers import StrOutputParser
-from src.config.settings import settings
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.runnables import Runnable
+from langchain_openai import AzureChatOpenAI
+
 from src.middlewares.logging import get_logger
-from langchain_core.messages import HumanMessage, AIMessage
+
+from ..query.executor.base import QueryResult
 
 logger = get_logger("chatbot")
 
-_enc = tiktoken.get_encoding("cl100k_base")
 
+_PROMPT_DIR = Path(__file__).resolve().parent.parent / "config" / "prompts"
+_SYSTEM_PROMPT_PATH = _PROMPT_DIR / "chatbot_system.md"
+_GUARDRAILS_PATH = _PROMPT_DIR / "guardrails.md"
 
-def _count_tokens(messages: list, context: str) -> dict:
-    msg_tokens = sum(len(_enc.encode(m.content)) for m in messages)
-    ctx_tokens = len(_enc.encode(context))
-    return {"messages_tokens": msg_tokens, "context_tokens": ctx_tokens, "total": msg_tokens + ctx_tokens}
 
+@dataclass
+class DocumentChunk:
+    """One retrieved document chunk for the unstructured path."""
 
-class ChatbotAgent:
-    """Chatbot agent with RAG capabilities."""
-
-    def __init__(self):
-        self.llm = AzureChatOpenAI(
-            azure_deployment=settings.azureai_deployment_name_4o,
-            openai_api_version=settings.azureai_api_version_4o,
-            azure_endpoint=settings.azureai_endpoint_url_4o,
-            api_key=settings.azureai_api_key_4o,
-            temperature=0.7
+    content: str
+    filename: str | None = None
+    page_label: str | None = None
+
+
+def _load_system_prompt() -> str:
+    """Compose system prompt = chatbot_system.md + guardrails.md.
+
+    Guardrails appended last so they take precedence in conflict (matches
+    the docstring at the top of guardrails.md).
+    """
+    chatbot = _SYSTEM_PROMPT_PATH.read_text(encoding="utf-8")
+    guardrails = _GUARDRAILS_PATH.read_text(encoding="utf-8")
+    return f"{chatbot}\n\n{guardrails}"
+
+
+def _format_query_result(qr: QueryResult) -> str:
+    """Render a QueryResult as a compact context block for the LLM."""
+    if qr.error:
+        return (
+            f"[Query result — FAILED]\n"
+            f"source_id={qr.source_id}\n"
+            f"error: {qr.error}"
         )
+    lines: list[str] = [
+        "[Query result]",
+        f"source_id: {qr.source_id}",
+        f"backend: {qr.backend}",
+        f"row_count: {qr.row_count}"
+        + (" (truncated)" if qr.truncated else ""),
+        f"elapsed_ms: {qr.elapsed_ms}",
+    ]
+    if qr.rows:
+        # Cap rendering at 25 rows; the LLM doesn't need the full set
+        cap = min(len(qr.rows), 25)
+        columns = list(qr.rows[0].keys())
+        lines.append("columns: " + ", ".join(columns))
+        lines.append("rows:")
+        for row in qr.rows[:cap]:
+            lines.append("  " + ", ".join(f"{k}={row[k]!r}" for k in columns))
+        if cap < len(qr.rows):
+            lines.append(f"  ... (+{len(qr.rows) - cap} more rows omitted from prompt)")
+    return "\n".join(lines)
+
+
+def _format_document_chunks(chunks: list[DocumentChunk]) -> str:
+    if not chunks:
+        return ""
+    blocks: list[str] = []
+    for c in chunks:
+        label_parts = [p for p in (c.filename, c.page_label) if p]
+        label = ", ".join(label_parts) if label_parts else "Unknown source"
+        blocks.append(f"[Source: {label}]\n{c.content}")
+    return "\n\n".join(blocks)
+
+
+def _build_context_block(
+    query_result: QueryResult | None,
+    chunks: list[DocumentChunk] | None,
+) -> str:
+    parts: list[str] = []
+    if query_result is not None:
+        parts.append(_format_query_result(query_result))
+    if chunks:
+        parts.append(_format_document_chunks(chunks))
+    return "\n\n".join(parts) if parts else "(no data context — answer conversationally)"
+
+
+def _build_default_chain() -> Runnable:
+    from src.config.settings import settings
+
+    llm = AzureChatOpenAI(
+        azure_deployment=settings.azureai_deployment_name_4o,
+        openai_api_version=settings.azureai_api_version_4o,
+        azure_endpoint=settings.azureai_endpoint_url_4o,
+        api_key=settings.azureai_api_key_4o,
+        temperature=0.3,
+    )
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", _load_system_prompt()),
+            MessagesPlaceholder(variable_name="history", optional=True),
+            ("human", "{message}"),
+            ("system", "Data context for this turn:\n\n{context}"),
+        ]
+    )
+    return prompt | llm | StrOutputParser()
+
+
+class ChatbotAgent:
+    """Formats and streams the final user-facing answer.
 
-        # Read system prompt
-        try:
-            with open("src/config/agents/system_prompt.md", "r") as f:
-                system_prompt = f.read()
-        except FileNotFoundError:
-            system_prompt = "You are a helpful AI assistant with access to user's uploaded documents."
+    `chain` is injectable: tests pass a fake that yields canned tokens.
+    Default constructs the production Azure OpenAI streaming chain on
+    first use.
+    """
 
-        # Create prompt template
-        self.prompt = ChatPromptTemplate.from_messages([
-            ("system", system_prompt),
-            MessagesPlaceholder(variable_name="messages"),
-            ("system", "Relevant documents:\n{context}")
-        ])
+    def __init__(self, chain: Runnable | None = None) -> None:
+        self._chain = chain
 
-        # Create chain
-        self.chain = self.prompt | self.llm | StrOutputParser()
+    def _ensure_chain(self) -> Runnable:
+        if self._chain is None:
+            self._chain = _build_default_chain()
+        return self._chain
 
-    async def generate_response(
+    async def astream(
         self,
-        messages: list,
-        context: str = ""
-    ) -> str:
-        """Generate response with optional RAG context."""
-        try:
-            logger.info("Generating chatbot response")
-
-            # Generate response
-            response = await self.chain.ainvoke({
-                "messages": messages,
-                "context": context
-            })
-
-            logger.info(f"Generated response: {response[:100]}...")
-            return response
-
-        except Exception as e:
-            logger.error("Response generation failed", error=str(e))
-            raise
-
-    async def astream_response(self, messages: list, context: str = ""):
-        """Stream response tokens as they are generated."""
-        try:
-            token_counts = _count_tokens(messages, context)
-            logger.info("LLM input tokens", **token_counts)
-            async for token in self.chain.astream({"messages": messages, "context": context}):
-                yield token
-        except Exception as e:
-            logger.error("Response streaming failed", error=str(e))
-            raise
-
-
-chatbot = ChatbotAgent()
+        message: str,
+        history: list[BaseMessage] | None = None,
+        query_result: QueryResult | None = None,
+        chunks: list[DocumentChunk] | None = None,
+    ) -> AsyncIterator[str]:
+        """Stream tokens of the final answer.
+
+        Caller wraps each token into the SSE format. Empty `history` and
+        no context = pure chat reply.
+        """
+        chain = self._ensure_chain()
+        payload: dict[str, Any] = {
+            "message": message,
+            "history": history or [],
+            "context": _build_context_block(query_result, chunks),
+        }
+        async for token in chain.astream(payload):
+            yield token

src/api/v1/chat.py

@@ -1,33 +1,33 @@
 """Chat endpoint with streaming support."""
 
 import uuid
+import json
+from typing import List, Dict, Any, Optional
+
 from fastapi import APIRouter, Depends, HTTPException
+from langchain_core.messages import HumanMessage, AIMessage
+from pydantic import BaseModel
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sse_starlette.sse import EventSourceResponse
+
+from src.agents.chat_handler import ChatHandler
+from src.config.settings import settings
 from src.db.postgres.connection import get_db
 from src.db.postgres.models import ChatMessage, MessageSource
-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.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
-from sse_starlette.sse import EventSourceResponse
-from langchain_core.messages import HumanMessage, AIMessage
-from sqlalchemy import select
-from pydantic import BaseModel
-from typing import List, Dict, Any, Optional
-import json
+
+logger = get_logger("chat_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Chat"])
 
 _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[str]:
-    """Return a direct response string for obvious greetings/farewells, else None."""
+    """Return a direct response for obvious greetings/farewells, else None."""
     lower = message.lower().strip().rstrip("!.,?")
     if lower in _GREETINGS:
         return "Hello! How can I assist you today?"
@@ -36,62 +36,12 @@ def _fast_intent(message: str) -> Optional[str]:
     return None
 
 
-logger = get_logger("chat_api")
-
-router = APIRouter(prefix="/api/v1", tags=["Chat"])
-
-
 class ChatRequest(BaseModel):
     user_id: str
     room_id: str
     message: str
 
 
-def _extract_sources(results: List[RetrievalResult]) -> List[Dict[str, Any]]:
-    """Extract deduplicated source references from retrieval results."""
-    seen = set()
-    sources = []
-    for result in results:
-        meta = result.metadata
-        data = meta.get("data", {})
-        if "document_id" in data:
-            key = (data.get("document_id"), data.get("page_label"))
-            if key not in seen:
-                seen.add(key)
-                sources.append({
-                    "document_id": data.get("document_id"),
-                    "filename": data.get("filename", "Unknown"),
-                    "page_label": data.get("page_label", "Unknown"),
-                })
-        else:
-            key = (data.get("table_name"), data.get("column_name"))
-            if key not in seen:
-                seen.add(key)
-                table_name = data.get("table_name")
-                user_id = meta.get("user_id")
-                sources.append({
-                    "document_id": f"{user_id}_{table_name}",
-                    "filename": data.get("table_name", "Unknown"),
-                    "page_label": data.get("column_name", "Unknown"),
-                })
-    logger.debug(f"Extracted sources: {sources}")
-    return sources
-
-
-def _to_document_chunks(results: List[RetrievalResult]) -> List[DocumentChunk]:
-    """Convert Phase 1 RetrievalResult list to Phase 2 DocumentChunk list."""
-    chunks = []
-    for r in results:
-        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]:
     cached = await redis.get(cache_key)
     if cached:
@@ -147,7 +97,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
     """Chat endpoint with streaming response.
 
     SSE event sequence:
-      1. sources  — JSON array of {document_id, filename, page_label}
+      1. sources  — JSON array of source references (empty until retrieval is wired into ChatHandler)
       2. chunk    — text fragments of the answer
       3. done     — signals end of stream
     """
@@ -181,49 +131,24 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
 
             return EventSourceResponse(stream_direct())
 
-        # 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)
+        history = await load_history(db, request.room_id, limit=10)
+        handler = ChatHandler()
 
-        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)
-
-        # Phase 2: AnswerAgent streams answer tokens
         async def stream_response():
             full_response = ""
-            yield {"event": "sources", "data": json.dumps(sources)}
-            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": ""}
-            await cache_response(redis, cache_key, full_response)
-            await save_messages(db, request.room_id, request.message, full_response, sources=sources)
+            yield {"event": "sources", "data": json.dumps([])}
+            async for event in handler.handle(request.message, request.user_id, history):
+                if event["event"] == "chunk":
+                    full_response += event["data"]
+                    yield event
+                elif event["event"] == "done":
+                    await cache_response(redis, cache_key, full_response)
+                    await save_messages(db, request.room_id, request.message, full_response, sources=[])
+                    yield event
+                elif event["event"] == "error":
+                    yield event
+                    return
+                # "intent" event: consumed internally, not forwarded to frontend
 
         return EventSourceResponse(stream_response())
 

src/api/v1/data_catalog.py

@@ -9,11 +9,12 @@ here on purpose.
 
 from typing import List
 
-from fastapi import APIRouter, HTTPException, status
+from fastapi import APIRouter, HTTPException, Query, status
 
 from src.catalog.store import CatalogStore
 from src.middlewares.logging import get_logger, log_execution
 from src.models.api.catalog import CatalogIndexEntry
+from src.pipeline.triggers import on_catalog_rebuild_requested
 
 logger = get_logger("data_catalog_api")
 
@@ -65,3 +66,35 @@ async def list_data_catalog_index(user_id: str):
         )
         for s in catalog.sources
     ]
+
+
+@router.post(
+    "/data-catalog/rebuild",
+    status_code=status.HTTP_200_OK,
+    summary="Rebuild the catalog for a user",
+    response_description="Confirmation that the rebuild was triggered.",
+    responses={
+        200: {"description": "Rebuild completed. Per-source errors are logged but do not fail this request."},
+        500: {"description": "Unexpected error before the rebuild loop started."},
+    },
+)
+@log_execution(logger)
+async def rebuild_data_catalog(
+    user_id: str = Query(..., description="ID of the user whose catalog should be rebuilt."),
+):
+    """
+    Re-introspect every source in the user's catalog and upsert the results.
+
+    Each source (DB connection or tabular file) is processed independently.
+    A failure on one source is logged but does not abort the remaining sources.
+    If the user has no catalog yet, returns success with no-op.
+    """
+    try:
+        await on_catalog_rebuild_requested(user_id)
+    except Exception as e:
+        logger.error("catalog rebuild failed", user_id=user_id, error=str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Catalog rebuild failed: {e}",
+        )
+    return {"status": "success", "user_id": user_id}

src/api/v1/db_client.py

@@ -27,9 +27,7 @@ from src.models.credentials import (  # noqa: F401 — re-exported for Swagger s
     SqlServerCredentials,
     SupabaseCredentials,
 )
-from src.pipeline.db_pipeline import db_pipeline_service
 from src.pipeline.triggers import on_db_registered
-from src.utils.db_credential_encryption import decrypt_credentials_dict
 
 logger = get_logger("database_client_api")
 
@@ -414,14 +412,14 @@ async def delete_database_client(
 @router.post(
     "/database-clients/{client_id}/ingest",
     status_code=status.HTTP_200_OK,
-    summary="Ingest schema from a registered database into the vector store",
-    response_description="Count of chunks ingested.",
+    summary="Build the catalog for a registered database connection",
+    response_description="Confirmation that the catalog was built.",
     responses={
-        200: {"description": "Ingestion completed successfully."},
+        200: {"description": "Catalog built successfully."},
         403: {"description": "Access denied — user_id does not own this connection."},
         404: {"description": "Connection not found."},
-        501: {"description": "The connection's db_type is not yet supported by the pipeline."},
-        500: {"description": "Ingestion failed (connection error, profiling error, etc.)."},
+        409: {"description": "Connection is inactive."},
+        500: {"description": "Catalog build failed."},
     },
 )
 @limiter.limit("5/minute")
@@ -433,11 +431,9 @@ async def ingest_database_client(
     db: AsyncSession = Depends(get_db),
 ):
     """
-    Decrypt the stored credentials, connect to the user's database, introspect
-    its schema, profile each column, embed the descriptions, and store them in
-    the shared PGVector collection tagged with `source_type="database"`.
-
-    Chunks become retrievable via the same retriever used for document chunks.
+    Introspect the registered database and build (or rebuild) the catalog entry
+    for this connection. The catalog is stored in `data_catalog` and used by
+    the query pipeline to plan structured queries.
     """
     client = await database_client_service.get(db, client_id)
 
@@ -454,26 +450,12 @@ async def ingest_database_client(
         )
 
     try:
-        creds = decrypt_credentials_dict(client.credentials)
-        with db_pipeline_service.engine_scope(
-            db_type=client.db_type,
-            credentials=creds,
-        ) as engine:
-            total = await db_pipeline_service.run(user_id=user_id, client_id=client_id, engine=engine)
-    except NotImplementedError as e:
-        raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=str(e))
+        await on_db_registered(client_id, user_id)
     except Exception as e:
-        logger.error(
-            f"Ingestion failed for client {client_id}", user_id=user_id, error=str(e)
-        )
+        logger.error("catalog build failed", client_id=client_id, error=str(e))
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Ingestion failed: {e}",
+            detail=f"Catalog build failed: {e}",
         )
 
-    try:
-        await on_db_registered(client_id, user_id)
-    except Exception as e:
-        logger.error("catalog build failed after ingest", client_id=client_id, error=str(e))
-
-    return {"status": "success", "client_id": client_id, "chunks_ingested": total}
+    return {"status": "success", "client_id": client_id}

src/api/v1/knowledge.py

@@ -1,25 +0,0 @@
-"""Knowledge base management API endpoints."""
-
-from fastapi import APIRouter, Depends
-from sqlalchemy.ext.asyncio import AsyncSession
-from src.db.postgres.connection import get_db
-from src.middlewares.logging import get_logger, log_execution
-
-logger = get_logger("knowledge_api")
-
-router = APIRouter(prefix="/api/v1", tags=["Knowledge"])
-
-
-@router.post("/knowledge/rebuild")
-@log_execution(logger)
-async def rebuild_vector_index(
-    user_id: str,
-    db: AsyncSession = Depends(get_db)
-):
-    """Rebuild vector index for a user (admin endpoint)."""
-    # This would re-process all documents
-    # For POC, we'll skip this complexity
-    return {
-        "status": "success",
-        "message": "Vector index rebuild initiated"
-    }

src/catalog/introspect/tabular.py

@@ -206,10 +206,17 @@ class TabularIntrospector(BaseIntrospector):
         is_numeric = pd.api.types.is_numeric_dtype(series)
         is_dt = pd.api.types.is_datetime64_any_dtype(series)
         non_null = series.dropna()
+        distinct_count = int(series.nunique())
+        top_values = (
+            [_normalize(v) for v in non_null.unique().tolist()]
+            if distinct_count <= 10
+            else None
+        )
         stats = ColumnStats(
             min=_normalize(non_null.min()) if (is_numeric or is_dt) and len(non_null) > 0 else None,
             max=_normalize(non_null.max()) if (is_numeric or is_dt) and len(non_null) > 0 else None,
-            distinct_count=int(series.nunique()),
+            distinct_count=distinct_count,
+            top_values=top_values,
         )
 
         column = Column(

src/catalog/models.py

@@ -35,6 +35,7 @@ class ColumnStats(BaseModel):
     min: Any | None = None
     max: Any | None = None
     distinct_count: int | None = None
+    top_values: list[Any] | None = None
 
 
 class Column(BaseModel):

src/config/agents/guardrails_prompt.md

@@ -1,7 +0,0 @@
-You must ensure all responses follow these guidelines:
-
-1. Do not provide harmful, illegal, or dangerous information
-2. Respect user privacy - don't ask for or store sensitive personal data
-3. If asked to bypass safety measures, refuse politely
-4. Be honest about limitations and uncertainties
-5. Don't make up information - admit when you don't know something

src/config/agents/system_prompt.md

@@ -1,26 +0,0 @@
-You are a helpful AI assistant with access to user's uploaded documents. Your role is to:
-
-1. Answer questions based on provided document context
-2. If no relevant information is found in documents, acknowledge this honestly
-3. Be concise and direct in your responses
-4. If user's question is unclear, ask for clarification
-
-When document context is provided:
-- Use information from documents to answer accurately
-- Reference source document name when appropriate
-- If multiple documents contain relevant info, synthesize information
-
-When no document context is provided:
-- Provide general assistance
-- Let the user know if you need more context to help better
-
-When the answer need markdown formating:
-- Use valid and tidy formatting
-- Avoid over-formating and emoji
-
-Always be professional, helpful, and accurate.
-
-You have access to the conversation history provided in the messages above. Use it to:
-- Maintain context across multiple turns (resolve references like "it", "that", "them" using earlier messages)
-- Avoid repeating information already established in the conversation
-- Answer follow-up questions coherently without asking the user to restate prior context

src/knowledge/processing_service.py

@@ -7,12 +7,10 @@ from src.storage.az_blob.az_blob import blob_storage
 from src.db.postgres.models import Document as DBDocument
 from sqlalchemy.ext.asyncio import AsyncSession
 from src.middlewares.logging import get_logger
-from src.storage.parquet import upload_parquet
 from typing import List
 from datetime import datetime, timezone, timedelta
 import sys
 import docx
-import pandas as pd
 import pytesseract
 from pdf2image import convert_from_bytes
 from io import BytesIO
@@ -44,10 +42,6 @@ class KnowledgeProcessingService:
 
             if db_doc.file_type == "pdf":
                 documents = await self._build_pdf_documents(content, db_doc)
-            elif db_doc.file_type == "csv":
-                documents = await self._build_csv_documents(content, db_doc)
-            elif db_doc.file_type == "xlsx":
-                documents = await self._build_excel_documents(content, db_doc)
             else:
                 text = self._extract_text(content, db_doc.file_type)
                 if not text.strip():
@@ -121,106 +115,6 @@ class KnowledgeProcessingService:
 
         return documents
 
-    def _profile_dataframe(
-        self, df: pd.DataFrame, source_name: str, db_doc: DBDocument
-    ) -> List[LangChainDocument]:
-        """Profile each column of a dataframe → one chunk per column."""
-        documents = []
-        row_count = len(df)
-
-        for col_name in df.columns:
-            col = df[col_name]
-            is_numeric = pd.api.types.is_numeric_dtype(col)
-            null_count = int(col.isnull().sum())
-            distinct_count = int(col.nunique())
-            distinct_ratio = distinct_count / row_count if row_count > 0 else 0
-
-            text = f"Source: {source_name} ({row_count} rows)\n"
-            text += f"Column: {col_name} ({col.dtype})\n"
-            text += f"Null count: {null_count}\n"
-            text += f"Distinct count: {distinct_count} ({distinct_ratio:.1%})\n"
-
-            if is_numeric:
-                text += f"Min: {col.min()}, Max: {col.max()}\n"
-                text += f"Mean: {col.mean():.4f}, Median: {col.median():.4f}\n"
-
-            if 0 < distinct_ratio <= 0.05:
-                top_values = col.value_counts().head(10)
-                top_str = ", ".join(f"{v} ({c})" for v, c in top_values.items())
-                text += f"Top values: {top_str}\n"
-
-            text += f"Sample values: {col.dropna().head(5).tolist()}"
-
-            documents.append(LangChainDocument(
-                page_content=text,
-                metadata={
-                    "user_id": db_doc.user_id,
-                    "source_type": "document",
-                    "chunk_level": "column",
-                    "updated_at": datetime.now(_JAKARTA_TZ).isoformat(),
-                    "data": {
-                        "document_id": db_doc.id,
-                        "filename": db_doc.filename,
-                        "file_type": db_doc.file_type,
-                        "source": source_name,
-                        "column_name": col_name,
-                        "column_type": str(col.dtype),
-                    }
-                }
-            ))
-        return documents
-
-    def _to_sheet_document(
-        self, df: pd.DataFrame, db_doc: DBDocument, sheet_name: str | None, source_name: str
-    ) -> LangChainDocument:
-        col_summary = ", ".join(f"{c} ({df[c].dtype})" for c in df.columns)
-        text = (
-            f"Source: {source_name} ({len(df)} rows)\n"
-            f"Columns ({len(df.columns)}): {col_summary}"
-        )
-        return LangChainDocument(
-            page_content=text,
-            metadata={
-                "user_id": db_doc.user_id,
-                "source_type": "document",
-                "chunk_level": "sheet",
-                "updated_at": datetime.now(_JAKARTA_TZ).isoformat(),
-                "data": {
-                    "document_id": db_doc.id,
-                    "filename": db_doc.filename,
-                    "file_type": db_doc.file_type,
-                    "sheet_name": sheet_name,
-                    "column_names": list(df.columns),
-                    "row_count": len(df),
-                },
-            },
-        )
-
-    async def _build_csv_documents(self, content: bytes, db_doc: DBDocument) -> List[LangChainDocument]:
-        """Profile each column of a CSV file and upload Parquet to Azure Blob."""
-        df = pd.read_csv(BytesIO(content))
-        await upload_parquet(df, db_doc.user_id, db_doc.id)
-        logger.info(f"Uploaded Parquet for CSV {db_doc.id}")
-        docs = self._profile_dataframe(df, db_doc.filename, db_doc)
-        docs.append(self._to_sheet_document(df, db_doc, sheet_name=None, source_name=db_doc.filename))
-        return docs
-
-    async def _build_excel_documents(self, content: bytes, db_doc: DBDocument) -> List[LangChainDocument]:
-        """Profile each column of every sheet in an Excel file and upload one Parquet per sheet."""
-        sheets = pd.read_excel(BytesIO(content), sheet_name=None)
-        documents = []
-        for sheet_name, df in sheets.items():
-            source_name = f"{db_doc.filename} / sheet: {sheet_name}"
-            docs = self._profile_dataframe(df, source_name, db_doc)
-            for doc in docs:
-                doc.metadata["data"]["sheet_name"] = sheet_name
-                doc.metadata["chunk_level"] = "column"
-            documents.extend(docs)
-            documents.append(self._to_sheet_document(df, db_doc, sheet_name, source_name))
-            await upload_parquet(df, db_doc.user_id, db_doc.id, sheet_name)
-            logger.info(f"Uploaded Parquet for sheet '{sheet_name}' of {db_doc.id}")
-        return documents
-
     def _extract_text(self, content: bytes, file_type: str) -> str:
         """Extract text from DOCX or TXT content."""
         if file_type == "docx":

src/pipeline/orchestrator.py

@@ -1,11 +0,0 @@
-"""IngestionOrchestrator — top-level coordinator for ingestion.
-
-Routes uploads / DB connections to the right pipeline:
-- unstructured (pdf/docx/txt) → DocumentPipeline
-- schema or tabular            → StructuredPipeline (which writes to the catalog)
-"""
-
-
-class IngestionOrchestrator:
-    async def ingest(self, source_ref: str, source_type: str, user_id: str) -> None:
-        raise NotImplementedError

src/pipeline/triggers.py

@@ -70,10 +70,30 @@ async def on_document_uploaded(document_id: str, user_id: str) -> None:
 
 
 async def on_catalog_rebuild_requested(user_id: str) -> None:
-    """Stub — re-runs every source for a user (catalog refresher).
+    """Re-introspect every source in the user's catalog and upsert the result.
 
-    Implemented when the bulk refresh script lands. Expected to iterate over
-    every Source in the user's current catalog, re-introspect it, and upsert
-    the refreshed result.
+    Iterates all Sources in the current catalog. Each source is re-run through
+    its original trigger (on_db_registered for schema, on_tabular_uploaded for
+    tabular). Per-source failures are logged but do not abort the remaining
+    sources.
     """
-    raise NotImplementedError
+    from src.catalog.store import CatalogStore
+
+    catalog = await CatalogStore().get(user_id)
+    if catalog is None:
+        logger.info("no catalog to rebuild", user_id=user_id)
+        return
+
+    logger.info("on_catalog_rebuild_requested triggered", user_id=user_id, source_count=len(catalog.sources))
+    for source in catalog.sources:
+        try:
+            if source.source_type == "schema":
+                client_id = source.location_ref.split("://")[1]
+                await on_db_registered(client_id, user_id)
+            elif source.source_type == "tabular":
+                document_id = source.location_ref.split("://")[1].split("/")[1]
+                await on_tabular_uploaded(document_id, user_id)
+            else:
+                logger.warning("unsupported source_type for rebuild", source_type=source.source_type, source_id=source.source_id)
+        except Exception as e:
+            logger.error("rebuild failed for source", source_id=source.source_id, source_type=source.source_type, error=str(e))

src/query/base.py

@@ -1,32 +0,0 @@
-"""Shared contract for query executors."""
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from src.retrieval.base import RetrievalResult
-
-
-@dataclass
-class QueryResult:
-    source_type: str        # "database" or "document"
-    source_id: str          # database_client_id or document_id
-    table_or_file: str
-    columns: list[str]
-    rows: list[dict]
-    row_count: int
-    metadata: dict = field(default_factory=dict)
-    # metadata should include "column_types": {"col_name": "dtype"} when available
-
-
-class BaseExecutor(ABC):
-    @abstractmethod
-    async def execute(
-        self,
-        results: list[RetrievalResult],
-        user_id: str,
-        db: AsyncSession,
-        question: str,
-        limit: int = 100,
-    ) -> list[QueryResult]: ...

src/retrieval/router.py

@@ -13,8 +13,6 @@ import hashlib
 import json
 from dataclasses import asdict
 
-from sqlalchemy.ext.asyncio import AsyncSession
-
 from src.db.redis.connection import get_redis
 from src.middlewares.logging import get_logger
 from src.retrieval.base import RetrievalResult
@@ -24,7 +22,6 @@ logger = get_logger("retrieval_router")
 
 _CACHE_TTL = 3600
 _CACHE_KEY_PREFIX = "retrieval"
-_UNSTRUCTURED_HINTS = frozenset({"document", "unstructured", "both"})
 
 
 class RetrievalRouter:
@@ -40,22 +37,17 @@ class RetrievalRouter:
         self,
         query: str,
         user_id: str,
-        db: AsyncSession,
         k: int = 5,
-        source_hint: str = "both",
     ) -> list[RetrievalResult]:
-        if source_hint not in _UNSTRUCTURED_HINTS:
-            return []
-
         redis = await get_redis()
         query_hash = hashlib.md5(query.encode()).hexdigest()
-        cache_key = f"{_CACHE_KEY_PREFIX}:{user_id}:{source_hint}:{query_hash}:{k}"
+        cache_key = f"{_CACHE_KEY_PREFIX}:{user_id}:{query_hash}:{k}"
 
         cached = await redis.get(cache_key)
         if cached:
             try:
                 raw = json.loads(cached)
-                logger.info("returning cached retrieval results", source_hint=source_hint)
+                logger.info("returning cached retrieval results")
                 return [RetrievalResult(**r) for r in raw]
             except Exception:
                 logger.warning("corrupted retrieval cache, fetching fresh")
@@ -66,9 +58,6 @@ class RetrievalRouter:
             logger.error("retrieval failed", error=str(e))
             return []
 
-        if not results and source_hint == "both":
-            logger.warning("empty retrieval result for source_hint='both'")
-
         await redis.setex(
             cache_key,
             _CACHE_TTL,