PHASE1_TO_PHASE2_REPORT.md

# Phase 1 → Phase 2 Migration Report

A walkthrough of what changed between the original retrieval-style backend (Phase 1) and the current catalog-driven backend (Phase 2). Intended as a hand-off for the lead.

---

## 1. The conceptual change

**Phase 1** was a single retrieval-style RAG pipeline. Every question — whether it pointed at a database, a spreadsheet, or a PDF — went through the same primitive: **chunk + embed + top-K** over PGVector. Schema and tabular columns were embedded as chunks and ranked alongside prose. When the question needed SQL, the LLM **wrote the SQL string directly** (via `query_executor`).

**Phase 2** splits the system into two paths governed by an LLM router:

| Path | Primitive | Why |
|---|---|---|
| Unstructured (PDF / DOCX / TXT) | Dense similarity over prose chunks (PGVector) | Right primitive for free text |
| Structured (DB / CSV / XLSX / Parquet) | **Per-user data catalog** → LLM emits a **JSON IR** of intent → deterministic **compiler** → **executor** (SQL or pandas) | A column lookup shouldn't go through a similarity ranking lottery; the LLM emits intent, never SQL syntax |

Three explicit LLM call sites only:

1. **Intent router** (classifies the user message into `chat` / `unstructured` / `structured`)
2. **Query planner** (turns the question + catalog into a Pydantic-validated `QueryIR`)
3. **Chatbot agent** (formats the final answer, streamed over SSE)

Everything else — IR validation, SQL/pandas compilation, execution — is deterministic Python.

---

## 2. File-by-file changes

### 2.1 Deleted (Phase 1 only)

| Phase 1 path | Reason it was removed |
|---|---|
| `src/rag/base.py`, `src/rag/retriever.py`, `src/rag/router.py` | Replaced by `src/retrieval/` |
| `src/rag/retrievers/baseline.py`, `schema.py`, `document.py` | Schema retrieval gone (catalog replaces it); document retriever rewritten in `src/retrieval/document.py` |
| `src/tools/search.py` (whole `tools/` folder) | Only consumer was `rag/router.py` |
| `src/query/base.py` | Duplicate of `query/executor/base.py` |
| `src/query/query_executor.py` | Replaced by `src/query/service.py` |
| `src/query/executors/db_executor.py` | Replaced by `src/query/executor/db.py` |
| `src/query/executors/tabular.py` | Replaced by `src/query/executor/tabular.py` |
| `src/agents/chatbot.py` (Phase 1 LangChain chatbot) | Phase 2 `ChatbotAgent` lives at the same path now — see §2.2 |
| `src/api/v1/knowledge.py` | Fake `/knowledge/rebuild` endpoint, never wired |
| `src/config/agents/system_prompt.md`, `guardrails_prompt.md` | Replaced by `src/config/prompts/{chatbot_system,guardrails}.md` |
| `src/models/structured_output.py` (`IntentClassification`) | Replaced by `IntentRouterDecision` Pydantic model inside `agents/orchestration.py` |
| `src/models/sql_query.py` | LLM no longer emits SQL; IR replaces it |
| `src/pipeline/orchestrator.py` (empty stub) | Redundant — `StructuredPipeline` takes the introspector at `run()` time |

### 2.2 Renamed / moved (same role, new home)

| Phase 1 location | Phase 2 location | Notes |
|---|---|---|
| `src/agents/chatbot.py` (Phase 1) → deleted, then `src/agents/answer_agent.py` (`AnswerAgent`) → renamed | `src/agents/chatbot.py::ChatbotAgent` | Final answer formation; streams via `astream` |
| `src/knowledge/parquet_service.py` | `src/storage/parquet.py` | Parquet upload/download helper |
| `src/pipeline/document_pipeline/document_pipeline.py` (folder) | `src/pipeline/document_pipeline.py` (flat) | Single module |
| `src/rag/retrievers/document.py` | `src/retrieval/document.py` | `DocumentRetriever` migrated; tabular file types filtered out of results |
| `src/rag/router.py` | `src/retrieval/router.py` | `RetrievalRouter`, Redis-cached, unstructured-only; dead `db: AsyncSession` + `source_hint` params removed |
| `src/rag/base.py` (`RetrievalResult`, `BaseRetriever`) | `src/retrieval/base.py` | Same dataclass + ABC |

> **Heads-up on the intent router**: the Phase 1 file `src/agents/orchestration.py` and its class `OrchestratorAgent` were **kept in place** for Phase 2 — but the body was fully rewritten. The class now emits `IntentRouterDecision(needs_search, source_hint ∈ {chat, unstructured, structured}, rewritten_query)`. The prompt file and test file use the `intent_router` name (`config/prompts/intent_router.md`, `tests/agents/test_intent_router.py`), but **the source module is still `orchestration.py` and the class is still `OrchestratorAgent`**. Existing imports continue to work; only the behavior changed.

### 2.3 Added (Phase 2 new)

**Catalog subsystem (whole new concept)**

| Path | Role |
|---|---|
| `src/catalog/models.py` | Pydantic: `Catalog → Source[] → Table[] → Column[]`, `ForeignKey`, `ColumnStats.top_values` |
| `src/catalog/introspect/base.py` | `BaseIntrospector` ABC |
| `src/catalog/introspect/database.py` | DB introspector — wraps Phase 1 `db_pipeline/extractor.py` (`get_schema`, `profile_column`, `get_row_count`) |
| `src/catalog/introspect/tabular.py` | CSV / XLSX / Parquet introspector — one `Table` per XLSX sheet |
| `src/catalog/render.py` | Renders a `Source` for the planner prompt |
| `src/catalog/validator.py` | Unique-ID + foreign-key-ref invariants |
| `src/catalog/store.py` | Postgres `jsonb` upsert keyed by `user_id` (table `data_catalog`) |
| `src/catalog/reader.py` | Loads + filters catalog by `source_hint` |
| `src/catalog/pii_detector.py` | Flags PII columns at ingestion → suppresses `sample_values` |
| `src/security/pii_patterns.py` | Name patterns + value regex used by the detector |

**JSON IR + query subsystem**

| Path | Role |
|---|---|
| `src/query/ir/models.py` | `QueryIR` Pydantic schema |
| `src/query/ir/operators.py` | `ALLOWED_FILTER_OPS`, `ALLOWED_AGG_FNS`, `LIMIT_HARD_CAP`, `TYPE_COMPATIBILITY` |
| `src/query/ir/validator.py` | Catalog-aware IR validation (rejects unknown column ids, bad ops, type mismatches, oversize limits) |
| `src/query/planner/service.py` | `QueryPlannerService.plan(question, catalog, previous_error)` — Azure OpenAI structured output → `QueryIR` |
| `src/query/planner/prompt.py` | Builds the planner prompt from catalog text |
| `src/query/compiler/base.py` | Compiler ABC |
| `src/query/compiler/sql.py` | `SqlCompiler` (Postgres) — all 12 filter ops, params as a dict |
| `src/query/compiler/pandas.py` | `PandasCompiler` — returns `CompiledPandas(apply, output_columns)` |
| `src/query/executor/base.py` | `BaseExecutor` + `QueryResult` |
| `src/query/executor/db.py` | `DbExecutor` — sqlglot SELECT-only guard, RO txn, 30 s `statement_timeout`, 10 k row cap |
| `src/query/executor/tabular.py` | `TabularExecutor` — Parquet via blob, `asyncio.to_thread`, 10 k cap |
| `src/query/executor/dispatcher.py` | `ExecutorDispatcher.pick(ir)` — picks by `source.source_type` |
| `src/query/service.py` | `QueryService.run(user_id, question, catalog)` — plan → validate → retry (max 3) → dispatch → execute |

**Agents**

| Path | Role |
|---|---|
| `src/agents/orchestration.py` | `OrchestratorAgent` — Phase 1 file/class name preserved; Phase 2 body. Emits `IntentRouterDecision` |
| `src/agents/chatbot.py` | `ChatbotAgent` — formerly `AnswerAgent` in `agents/answer_agent.py`; renamed in Cleanup PR |
| `src/agents/chat_handler.py` | `ChatHandler.handle(...)` — top-level orchestrator; yields `intent` / `chunk` / `done` / `error` SSE events |

**Pipelines & API**

| Path | Role |
|---|---|
| `src/pipeline/structured_pipeline.py` | DB / tabular ingestion: introspect → merge → validate → upsert |
| `src/pipeline/triggers.py` | `on_db_registered`, `on_tabular_uploaded`, `on_document_uploaded`, `on_catalog_rebuild_requested` |
| `src/api/v1/data_catalog.py` | `GET /api/v1/data-catalog/{user_id}` + `POST /api/v1/data-catalog/rebuild` |
| `src/models/api/catalog.py` | Catalog request/response models |
| `src/config/prompts/intent_router.md`, `query_planner.md`, `chatbot_system.md`, `guardrails.md` | New prompts. `guardrails.md` is appended to `chatbot_system.md` at load time |
| `src/db/postgres/models.py` (added `Catalog` SQLAlchemy class) | Stores the per-user jsonb document in `data_catalog` |

### 2.4 Rewired API endpoints

| Endpoint | Phase 1 wiring | Phase 2 wiring |
|---|---|---|
| `POST /api/v1/chat/stream` | Inline in `chat.py`: `OrchestratorAgent` → `retriever` → `query_executor` → `chatbot` | Delegates to `ChatHandler.handle()`. Redis cache, fast intent, history load, and message persistence stay in the endpoint |
| `POST /api/v1/database-clients/{id}/ingest` | Called `db_pipeline_service.run()` and dual-wrote vectors | Calls **only** `on_db_registered` (catalog build). Failure → HTTP 500 |
| `POST /api/v1/document/process` | Always pushed to vector store | PDF/DOCX/TXT → `knowledge_processor` (vectors); CSV/XLSX → `on_tabular_uploaded` (catalog only, **no vector embedding**) |
| `POST /api/v1/document/upload` | Storage + DB row | Same, plus `on_document_uploaded` trigger |
| `POST /api/v1/data-catalog/rebuild` | — | New: iterates all sources, re-runs per-source trigger |
| `GET /api/v1/data-catalog/{user_id}` | — | New: returns `list[CatalogIndexEntry]` |

### 2.5 Phase 1 files still in production use

These were **not rewritten** — Phase 2 imports them directly:

- `src/database_client/database_client_service.py`
- `src/utils/db_credential_encryption.py` (`decrypt_credentials_dict`) — `src/security/credentials.py` is still a stub
- `src/pipeline/db_pipeline/db_pipeline_service.py` (`engine_scope` context manager — used by both the introspector and `DbExecutor`)
- `src/pipeline/db_pipeline/extractor.py` (`get_schema`, `profile_column`, `get_row_count`)
- `src/knowledge/processing_service.py` (PDF / DOCX / TXT extraction + embedding)
- `src/db/postgres/{connection,init_db,vector_store}.py`, `src/storage/az_blob/`, `src/middlewares/`, `src/security/auth.py`

---

## 3. End-to-end flow (current state)

### 3.1 Ingestion

```
User action                Pipeline                                Storage
──────────────             ────────────────────────────             ─────────────────
upload PDF/DOCX/TXT  →   DocumentPipeline                       →  Azure Blob + PGVector
                          (extract → chunk → embed)                 (table: langchain_pg_embedding)
                          + on_document_uploaded                    + retrieval cache invalidate

upload CSV/XLSX     →    TabularIntrospector                   →  Azure Blob (Parquet)
                          (sheets / columns + sample + stats)       + data_catalog jsonb row
                          → CatalogValidator → CatalogStore         (NO vector store — catalog only)
                          via on_tabular_uploaded

register DB         →    DatabaseIntrospector                  →  data_catalog jsonb row
                          (information_schema + sample + FKs)
                          → validate → store
                          via on_db_registered
```

### 3.2 Query (per user message → SSE stream)

```
POST /api/v1/chat/stream
        │
        ├── Redis cache check (24h TTL) — hit returns cached stream
        ├── _fast_intent (greetings / goodbyes) — bypass LLM
        ├── load history from chat_messages
        │
        └── ChatHandler.handle(message, user_id, history)         [src/agents/chat_handler.py]
                │
                ├─ OrchestratorAgent.classify()                    [agents/orchestration.py]
                │     → needs_search, source_hint, rewritten_query
                │
                ├── source_hint == "chat"
                │     → ChatbotAgent.astream() → yield chunk events
                │
                ├── source_hint == "unstructured"
                │     → RetrievalRouter.retrieve()                 [retrieval/router.py, Redis-cached]
                │         → DocumentRetriever (PGVector MMR/cosine/etc.)
                │     → ChatbotAgent.astream(chunks=...)
                │
                └── source_hint == "structured"
                      → CatalogReader.read(user_id, "structured")  [catalog/reader.py]
                      → QueryService.run(user_id, question, catalog)   [query/service.py]
                           │
                           ├─ QueryPlannerService.plan(...)        [query/planner/service.py]
                           │    LLM(catalog, question, prev_error?) → QueryIR
                           │
                           ├─ IRValidator.validate(ir, catalog)    [query/ir/validator.py]
                           │    fail → loop back to planner with error context (max 3)
                           │
                           ├─ ExecutorDispatcher.pick(ir)          [query/executor/dispatcher.py]
                           │    schema source  → DbExecutor
                           │    tabular source → TabularExecutor
                           │
                           ├─ DbExecutor.run(ir):                  [query/executor/db.py]
                           │    SqlCompiler → (sql, params)
                           │      → sqlglot SELECT-only guard
                           │      → engine_scope (Phase 1 utility) in asyncio.to_thread
                           │      → RO txn + statement_timeout=30s + 10k cap
                           │
                           ├─ TabularExecutor.run(ir):             [query/executor/tabular.py]
                           │    resolve Parquet blob path
                           │      → download → PandasCompiler.apply(df)
                           │      → asyncio.to_thread → 10k cap
                           │
                           └─ QueryResult { rows, columns, row_count,
                                            truncated, source_id, error?, elapsed_ms }
                      →
                      ChatbotAgent.astream(query_result=...)
                            → yield chunk events
                │
                └── final events: done / error
        │
        └── persist user + assistant messages to chat_messages
        └── populate Redis cache
```

**Safety invariants for the structured path** (read-only at every layer):

1. IR validated against the catalog before reaching the compiler
2. Identifiers come from the catalog (trusted; inlined as quoted identifiers)
3. Values from `IR.filters` are always parameterized
4. Compiler is deterministic — no LLM in the hot path
5. sqlglot rejects anything that isn't a pure SELECT
6. DB connection is read-only with a 30 s `statement_timeout`
7. Hard 10 000 row cap on both executors; neither raises — errors go in `QueryResult.error`

---

## 4. Summary table for review

| Concern | Phase 1 — where it lived | Phase 2 — where it lives | Change type |
|---|---|---|---|
| Intent classification | `agents/orchestration.py::OrchestratorAgent` (free-text intent) | **Same path + same class name** — body rewritten to emit `IntentRouterDecision` | Body rewrite only |
| Top-level chat orchestration | Inline in `api/v1/chat.py` | `agents/chat_handler.py::ChatHandler` | Extracted to a reusable module |
| Final answer formation | `agents/chatbot.py` (Phase 1 LangChain) | `agents/chatbot.py::ChatbotAgent` (was `AnswerAgent` in `answer_agent.py` mid-cycle) | Rewritten + renamed |
| Schema retrieval (DB / tabular) | `rag/retrievers/schema.py` + PGVector chunks | **Removed**. Replaced by catalog (`catalog/store.py` jsonb) loaded verbatim into planner prompt | Whole concept replaced |
| Doc retrieval (PDF / DOCX / TXT) | `rag/retrievers/document.py`, `rag/router.py` | `retrieval/document.py`, `retrieval/router.py` | Moved; Redis cache restored; tabular files filtered |
| Query writing | `query/query_executor.py` + `models/sql_query.py` (LLM writes SQL) | `query/planner/service.py` (LLM writes IR) + `query/compiler/sql.py` (deterministic) | LLM emits intent, not SQL |
| DB execution | `query/executors/db_executor.py` | `query/executor/db.py::DbExecutor` | Folder renamed (`executors` → `executor`); sqlglot guard + RO txn + 30 s timeout kept |
| Tabular execution | `query/executors/tabular.py` | `query/executor/tabular.py::TabularExecutor` | Parquet-only; pandas compiler split out |
| Executor selection | Hard-coded in `query_executor.py` | `query/executor/dispatcher.py::ExecutorDispatcher` | New; routes by `source.source_type` |
| Catalog (NEW) | — | `catalog/` (models, introspect/, validator, store, reader, pii_detector, render) | New subsystem |
| Catalog persistence | (data was embedded in PGVector) | Postgres jsonb table `data_catalog`, keyed by `user_id` | New table |
| Ingestion triggers | Inline in API endpoints | `pipeline/triggers.py` (`on_db_registered`, `on_tabular_uploaded`, `on_document_uploaded`, `on_catalog_rebuild_requested`) | Centralized event entry points |
| Structured pipeline | `pipeline/db_pipeline/db_pipeline_service.py` (still present for `engine_scope` + extractor reuse) | `pipeline/structured_pipeline.py` (orchestrator) — reuses Phase 1 extractor | New orchestrator wraps Phase 1 introspection helpers |
| Document pipeline | `pipeline/document_pipeline/document_pipeline.py` (folder) | `pipeline/document_pipeline.py` (file) | Flattened; CSV / XLSX now skip the vector store |
| Parquet helper | `knowledge/parquet_service.py` | `storage/parquet.py` | Moved into `storage/` |
| Prompts | `config/agents/system_prompt.md`, `guardrails_prompt.md` | `config/prompts/{intent_router,query_planner,chatbot_system,guardrails}.md` | Folder renamed; split into four files; guardrails appended to `chatbot_system` at load |
| PII detection | — | `catalog/pii_detector.py` + `security/pii_patterns.py` | New. Columns flagged `pii_flag=true` get `sample_values: null` so PII never enters prompts |
| Chat endpoint | `api/v1/chat.py` (does everything inline) | `api/v1/chat.py` (cache + history + persistence) → delegates to `ChatHandler` | Slimmed; SSE event shape is `intent` / `chunk` / `done` / `error` |
| DB ingest endpoint | `api/v1/db_client.py::ingest` (Phase 1 `db_pipeline_service.run()`) | `api/v1/db_client.py::ingest` (calls `on_db_registered` only) | Phase 1 dual-write removed |
| Document process endpoint | `api/v1/document.py::process` (always vectorize) | `api/v1/document.py::process` (PDF/DOCX/TXT → vectors; CSV/XLSX → catalog via `on_tabular_uploaded`) | Routing by file type |
| Catalog management API | — | `api/v1/data_catalog.py` (GET index + POST rebuild) | New |

**Bottom line.** Every Phase 1 file under `src/rag/`, `src/tools/`, `src/query/executors/`, `src/query/query_executor.py`, `src/query/base.py`, `src/api/v1/knowledge.py`, and `src/config/agents/` is gone. Phase 1 introspection helpers under `src/pipeline/db_pipeline/` and `src/database_client/` are still imported by Phase 2 — they were not rewritten, just wrapped. The three LLM call sites are now explicit and the SQL-writing one no longer exists; the planner emits a Pydantic-validated `QueryIR` instead.

The one filename gotcha to remember: the **intent router** still lives at `src/agents/orchestration.py` as class `OrchestratorAgent` (Phase 1 name kept for import-site compatibility, Phase 2 body). The matching prompt and tests use the `intent_router` name, but the source module does not.