[KM-560][KM-561] drop catalog LLM enrichment + rename store to data_catalog + add /data-catalog index endpoint

- Remove CatalogEnricher entirely (cost cut). Planner reads stats +
sample rows + column names directly. Deletes src/catalog/enricher.py,
config/prompts/catalog_enricher.md, and the StructuredPipeline enrich
step. render_source moves to src/catalog/render.py.
- Rename jsonb table catalogs -> data_catalog. Class name unchanged;
table created fresh on next init_db (no rows in prod yet, no
migration needed).
- Add GET /api/v1/data-catalog/{user_id} returning list[CatalogIndexEntry]
(source_id, source_type, name, location_ref, table_count, updated_at).
Wired in main.py. Lightweight summary intended for the catalog
refresher.
- Update ARCHITECTURE.md / REPO_CONTEXT.md / PROGRESS.md: LLM call sites
now three (was four), ingestion flow is introspect -> validate -> upsert.

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

ARCHITECTURE.md

@@ -95,12 +95,11 @@ Each stage is its own module with typed input and typed output. No god classes.
 
 ### 4.5 Minimal LLM surface
 
-LLM calls happen in exactly four places:
+LLM calls happen in exactly three places (KM-557 removed `CatalogEnricher`; ingestion is now LLM-free — the planner reads column names, stats, and sample rows directly):
 
 1. **`IntentRouter`** — once per user message
-2. **`CatalogEnricher`** — once per source, at ingestion (not query time)
-3. **`QueryPlanner`** — once per structured query (produces the IR)
-4. **`ChatbotAgent`** — once per answer (formats the response)
+2. **`QueryPlanner`** — once per structured query (produces the IR)
+3. **`ChatbotAgent`** — once per answer (formats the response)
 
 Compiler and executors are pure code. No LLM in the hot path of query construction.
 
@@ -115,11 +114,9 @@ source upload / DB connect
     ↓
 introspect schema (DB: information_schema; tabular: file headers + sample rows)
     ↓
-CatalogEnricher  (1 LLM call per source — generates AI descriptions)
-    ↓
 validate (Pydantic)
     ↓
-write to catalog store (Postgres jsonb, keyed by user_id)
+write to catalog store (Postgres jsonb in `data_catalog`, keyed by user_id)
 ```
 
 For unstructured files: chunk + embed → PGVector.

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 (items 16,31,35,36,41 done; Phase 1 remnants deleted: query/executors/, query_executor.py, agents/orchestration.py)
+**Last updated**: 2026-05-11 ([KM-557] catalog LLM enrichment removed; jsonb table renamed `catalogs` → `data_catalog`; new `GET /api/v1/data-catalog/{user_id}` index endpoint)
 **Current open PR**: none — all Phase 2 contracts shipped on `pr/1`. Cleanup PR pending (API rewiring + Phase 1 removal).
 
 ---
@@ -22,7 +22,8 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 |---|---|---|---|
 | PR1 | `[x]` merged | DB | Contract locks + catalog plumbing + DB introspector + IR validator + tests |
 | PR1-tab | `[x]` shipped | TAB | Tabular introspector + on_tabular_uploaded trigger + 31 unit tests |
-| PR2a | `[x]` merged | DB | CatalogEnricher + StructuredPipeline + on_db_registered trigger + FK extension on Table |
+| PR2a | `[x]` merged | DB | CatalogEnricher + StructuredPipeline + on_db_registered trigger + FK extension on Table (enricher later removed in KM-557) |
+| KM-557 | `[x]` shipped | DB | Drop CatalogEnricher entirely (cost cut — planner uses stats + sample rows directly); rename jsonb table `catalogs` → `data_catalog`; add `GET /api/v1/data-catalog/{user_id}` index endpoint for catalog refresher |
 | PR2b | `[x]` shipped | DB-solo (B-review) | IntentRouter + planner prompt + planner LLM service |
 | PR3-DB | `[x]` shipped | DB | SqlCompiler (Postgres) + DbExecutor (sqlglot guard, RO + statement_timeout, asyncio.to_thread) + 36 golden IR→SQL tests |
 | PR3-TAB | `[x]` shipped | TAB | PandasCompiler + TabularExecutor + 43+12 golden IR→DataFrame tests |
@@ -44,7 +45,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | 2 | IR Pydantic models (`query/ir/models.py`) | `[x]` | Pre-existing scaffold |
 | 3 | IR operator whitelists (`query/ir/operators.py`) | `[x]` | PR1 filled `TYPE_COMPATIBILITY` matrix |
 | 4 | PII patterns / regex (`security/pii_patterns.py`) | `[x]` | Pre-existing |
-| — | `catalogs` Postgres jsonb table (`db/postgres/models.py`) | `[x]` | PR1 added `Catalog` SQLAlchemy class + `init_db.py` import |
+| — | `data_catalog` Postgres jsonb table (`db/postgres/models.py`) | `[x]` | PR1 added `Catalog` SQLAlchemy class + `init_db.py` import. KM-557 renamed `__tablename__` from `catalogs` → `data_catalog`; created fresh (no migration) |
 | — | `QueryResult` shape (`query/executor/base.py`) | `[x]` | Pre-existing scaffold; `columns: list[str]` added (TAB owner, PR1-tab) — DbExecutor updated to populate it. |
 | — | `Source.location_ref` URI scheme | `[x]` | PR1 documented in `catalog/models.py` docstring |
 
@@ -60,7 +61,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 
 | # | Item | Owner | Status | Notes |
 |---|---|---|---|---|
-| 8 | Catalog enricher + prompt (`catalog/enricher.py`, `config/prompts/catalog_enricher.md`) | B | `[x]` | PR2a (DB owner picked up) — Azure OpenAI GPT-4o, structured output (flat `EnrichmentResponse` keyed by stable IDs), source-type-agnostic prompt with PII suppression and FK rendering. LLM is constructor-injectable for tests. |
+| 8 | ~~Catalog enricher + prompt~~ | B | **REMOVED in KM-557** | Cost optimization — planner reads stats + sample rows + column names directly. `catalog/enricher.py` + `config/prompts/catalog_enricher.md` deleted. `render_source` (the only piece still needed) moved to `src/catalog/render.py`. Tests moved to `tests/catalog/test_render.py`. |
 | 9 | Catalog validator (`catalog/validator.py`) | B | `[x]` | PR1 (DB owner picked up) — uniqueness invariants |
 | 10 | Catalog store — Postgres jsonb (`catalog/store.py`) | B | `[x]` | PR1 (DB owner picked up) — `INSERT ... ON CONFLICT` |
 | 11 | Catalog reader (`catalog/reader.py`) | B | `[x]` | PR1 (DB owner picked up) — filters by source_hint, empty on miss |
@@ -70,7 +71,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 
 | # | Item | Owner | Status | Notes |
 |---|---|---|---|---|
-| 13 | Structured pipeline (`pipeline/structured_pipeline.py`) | B | `[x]` | PR2a (DB owner) — `introspect → enrich → merge with existing → validate → upsert`. Source-type-agnostic: caller supplies the introspector. `default_structured_pipeline()` factory wires production deps lazily so tests can inject mocks without `Settings()` construction. |
+| 13 | Structured pipeline (`pipeline/structured_pipeline.py`) | B | `[x]` | PR2a (DB owner) — Source-type-agnostic: caller supplies the introspector. `default_structured_pipeline()` factory wires production deps lazily so tests can inject mocks without `Settings()` construction. **KM-557**: enrich step removed; pipeline is now `introspect → merge with existing → validate → upsert`. Constructor no longer takes `enricher`. |
 | 14 | Triggers (`pipeline/triggers.py`) | B | `[~]` | PR2a — `on_db_registered` implemented (DB owner). PR1-tab — `on_tabular_uploaded` implemented (TAB owner). `on_document_uploaded`, `on_catalog_rebuild_requested` still stubs. |
 | 15 | Ingestion orchestrator (`pipeline/orchestrator.py`) | B | `[ ]` | Likely redundant — StructuredPipeline already takes the introspector at run() time. Revisit if a higher-level routing layer is needed. |
 | 16 | Document pipeline (`pipeline/document_pipeline.py`) | TAB | `[x]` | Flattened `pipeline/document_pipeline/document_pipeline.py` (folder) → `pipeline/document_pipeline.py` (file). Updated import in `api/v1/document.py`. |
@@ -81,7 +82,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 |---|---|---|---|---|
 | 17 | IR validator (`query/ir/validator.py`) | B | `[x]` | PR1 (DB owner) — full rule set; descriptive errors for planner retry |
 | 18 | Planner LLM service (`query/planner/service.py`) | B | `[x]` | PR2b — Azure OpenAI structured output → `QueryIR`. Injectable chain. Supports retry via `previous_error` argument. |
-| 19 | Planner prompt (`query/planner/prompt.py`, `config/prompts/query_planner.md`) | B | `[x]` | PR2b — system prompt with hard constraints + few-shot for DB and tabular sources. `build_planner_prompt(question, catalog, previous_error)` reuses `catalog.enricher.render_source` so both LLM call sites see the same source format. |
+| 19 | Planner prompt (`query/planner/prompt.py`, `config/prompts/query_planner.md`) | B | `[x]` | PR2b — system prompt with hard constraints + few-shot for DB and tabular sources. `build_planner_prompt(question, catalog, previous_error)` calls `catalog.render.render_source` (renamed from `catalog.enricher.render_source` in KM-557). |
 | 20 | Intent router (`agents/intent_router.py`, `config/prompts/intent_router.md`) | B | `[x]` | PR2b — single LLM call → `IntentRouterDecision(needs_search, source_hint, rewritten_query)`. Supports conversation history. |
 | 21 | Executor base + `QueryResult` (`query/executor/base.py`) | B | `[x]` | Pre-existing scaffold |
 | 22 | Executor dispatcher (`query/executor/dispatcher.py`) | B | `[x]` | PR4 — picks DbExecutor / TabularExecutor by `source.source_type`. Lazy imports of production executors keep import side-effect-free for tests. Caches per source_type. |
@@ -121,6 +122,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | 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 | `[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 |
+| — | Data catalog index endpoint (`api/v1/data_catalog.py`) | DB | `[x]` | **KM-557** — `GET /api/v1/data-catalog/{user_id}` → `list[CatalogIndexEntry]` (source_id, source_type, name, location_ref, table_count, updated_at). Lightweight summary used by the catalog refresher; full catalog rows are not exposed. Router wired in `main.py`. Response model in `src/models/api/catalog.py:CatalogIndexEntry`. |
 
 ### Tests + eval
 
@@ -131,6 +133,7 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 | 40 | IR validator tests (`tests/query/ir/test_validator.py`) | B | `[x]` | PR1 — 19 tests, all rules covered |
 | — | PII detector tests (`tests/catalog/test_pii_detector.py`) | B | `[x]` | PR1 — 26 tests (parametrized) |
 | — | Catalog validator tests (`tests/catalog/test_validator.py`) | B | `[x]` | PR1 — 5 tests |
+| — | Catalog render tests (`tests/catalog/test_render.py`) | B | `[x]` | **KM-557** — 5 tests (renamed from `test_enricher.py`; LLM enrichment tests dropped, render-only tests kept). |
 | — | Catalog store integration test (`tests/catalog/test_store.py`) | DB | `[x]` | PR1 — module-level skip without `RUN_INTEGRATION_TESTS=1` |
 | — | DB introspector test | DB | `[ ]` | Deferred to PR2 — needs Postgres testcontainer or fixture infra |
 | — | Tabular introspector test | TAB | `[x]` | PR1-tab — 31 unit tests (CSV/XLSX/Parquet, stats, PII, error paths). No DB/blob I/O — mocks injected via constructor. |
@@ -141,7 +144,45 @@ Persistent tracker mirroring the 42-item ownership table in `REPO_CONTEXT.md` "T
 
 ---
 
-## What just shipped (PR2b/4/5/6/7-bundle — DB owner solo, teammate reviews)
+## What just shipped (KM-557 — DB owner)
+
+After lead review of the catalog ingestion cost: dropped LLM enrichment,
+renamed the storage table, and exposed a lightweight index endpoint for
+the upcoming catalog refresher.
+
+**Files deleted**:
+- `src/catalog/enricher.py` — entire CatalogEnricher + EnrichmentResponse + apply_descriptions removed
+- `src/config/prompts/catalog_enricher.md` — dead prompt
+- `tests/catalog/test_enricher.py` — replaced by `test_render.py`
+
+**Files added**:
+- `src/catalog/render.py` — new home for `render_source` (the only piece of the old enricher still needed; consumed by `query/planner/prompt.py`)
+- `src/api/v1/data_catalog.py` — `GET /api/v1/data-catalog/{user_id}` returns `list[CatalogIndexEntry]`
+- `tests/catalog/test_render.py` — 5 tests (same coverage as the old render block)
+
+**Files modified**:
+- `src/db/postgres/models.py` — `__tablename__ = "data_catalog"` (was `"catalogs"`). Class name unchanged
+- `src/pipeline/structured_pipeline.py` — `StructuredPipeline(validator, store)` (was `(enricher, validator, store)`); pipeline is now `introspect → merge → validate → upsert`; `default_structured_pipeline()` no longer constructs an enricher
+- `src/pipeline/triggers.py` — docstrings updated; `on_catalog_rebuild_requested` docstring rewritten for the refresher use case
+- `src/query/planner/prompt.py` — import now `from ...catalog.render import render_source`
+- `src/catalog/introspect/{base,database,tabular}.py` — docstring scrubs (no behavior changes)
+- `src/models/api/catalog.py` — added `CatalogIndexEntry`; simplified `CatalogRebuildResponse` to `sources_rebuilt`
+- `main.py` — registered `data_catalog_router`
+- `src/security/README.md` — one stale wording fix
+
+**No migration**: the `data_catalog` table is created from scratch on first `init_db()`. The old `catalogs` table was never deployed against production data, so no rename SQL is needed.
+
+**Tests**: all 4 `test_structured_pipeline.py` tests reworked to construct `StructuredPipeline(validator=, store=)` without `enricher`. 5 `test_render.py` tests cover render_source standalone.
+
+**Lint**: `ruff check` clean on modified Phase 2 paths.
+
+**Open follow-ups left for the lead**:
+- `on_catalog_rebuild_requested` body — the refresher will iterate the index endpoint and call this trigger per source
+- `api/v1/db_client.py` `/ingest` still doesn't call `on_db_registered` — same blocker as before, untouched by KM-557
+
+---
+
+## What shipped previously (PR2b/4/5/6/7-bundle — DB owner solo, teammate reviews)
 
 **Files implemented**:
 - `src/agents/intent_router.py` — `IntentRouter.classify(message, history) → IntentRouterDecision`. Pydantic model for structured output. History-aware query rewriting.

REPO_CONTEXT.md

@@ -68,9 +68,9 @@ src/                   — all application code
 | `catalog/introspect/base.py` | `BaseIntrospector.introspect(location_ref) -> Source` |
 | `catalog/introspect/database.py` | `information_schema` + ~100 row sample → draft Source |
 | `catalog/introspect/tabular.py` | Parquet/CSV/XLSX header reader + sample (one Table per sheet for XLSX) |
-| `catalog/enricher.py` | one LLM call per source — adds AI descriptions at source/table/column |
+| `catalog/render.py` | renders a `Source` as the canonical text block consumed by the planner (KM-557; LLM enrichment removed — planner reads stats + samples directly) |
 | `catalog/validator.py` | invariants beyond Pydantic shape (unique IDs, FK refs) |
-| `catalog/store.py` | persist as Postgres `jsonb` row keyed by user_id (`get/upsert/delete`) |
+| `catalog/store.py` | persist as Postgres `jsonb` row keyed by user_id (`get/upsert/delete`) — table `data_catalog` |
 | `catalog/reader.py` | load + filter catalog by source_hint (returns full catalog for ≤50 tables) |
 | `catalog/pii_detector.py` | flag PII columns at ingestion → suppresses `sample_values` |
 
@@ -97,14 +97,16 @@ src/                   — all application code
 | `retrieval/document.py` | `DocumentRetriever` over PGVector chunks |
 | `retrieval/router.py` | dispatches the `unstructured` route (the `chat` and `structured` routes do not pass through here) |
 
-### Agents — the four LLM call sites
+### Agents — the three LLM call sites
 
 | Path | Role |
 |---|---|
 | `agents/intent_router.py` | classify message → `needs_search`, `source_hint ∈ {chat, unstructured, structured}` |
 | `agents/chatbot.py` | final answer formation (receives Cu chunks or QueryResult); SSE-streamed |
 
-(`CatalogEnricher` + `QueryPlanner` are the other two LLM call sites — both live under `catalog/` and `query/planner/`.)
+(`QueryPlanner` is the third LLM call site, under `query/planner/`. The
+fourth — `CatalogEnricher` — was removed in KM-557; ingestion no longer
+makes any LLM calls.)
 
 ### Pipelines — ingestion coordinators
 
@@ -139,7 +141,7 @@ src/                   — all application code
 | `observability/langfuse/langfuse.py` | trace helper |
 | `config/settings.py` | pydantic-settings; `.env` uses double-underscore aliases |
 | `config/env_constant.py` | env file path constant |
-| `config/prompts/*.md` | prompt templates: `intent_router`, `catalog_enricher`, `query_planner`, `chatbot_system`, `guardrails` |
+| `config/prompts/*.md` | prompt templates: `intent_router`, `query_planner`, `chatbot_system`, `guardrails` (KM-557 removed `catalog_enricher`) |
 
 ---
 
@@ -153,8 +155,7 @@ src/                   — all application code
 
 4. **Pipeline stage isolation.** Each stage (`IntentRouter`, `CatalogReader`, `QueryPlanner`, `IRValidator`, `QueryCompiler`, `QueryExecutor`, `ChatbotAgent`) is its own module with typed input and typed output. No god classes.
 
-5. **Minimal LLM surface.** Only four LLM call sites in the system:
-   - `CatalogEnricher` — once per source, **at ingestion** (not query time)
+5. **Minimal LLM surface.** Only three LLM call sites in the system (KM-557 dropped `CatalogEnricher` — ingestion is now LLM-free; the planner reads stats + sample rows + column names directly):
    - `IntentRouter` — once per user message
    - `QueryPlanner` — once per structured query
    - `ChatbotAgent` — once per answer (formatting)
@@ -179,9 +180,8 @@ source upload / DB connect
     │
     └── structured (DB schema or tabular file)
           → introspect (information_schema or file headers + sample rows)
-          → CatalogEnricher (1 LLM call per source — AI descriptions)
           → CatalogValidator (Pydantic + unique-IDs + FK refs)
-          → CatalogStore.upsert(user_id jsonb row)
+          → CatalogStore.upsert(user_id jsonb row in `data_catalog`)
 ```
 
 ### Query (per user message)
@@ -313,7 +313,7 @@ The service is built by two engineers; many modules are source-type-agnostic and
 | 6 | Tabular introspector (CSV/XLSX/Parquet headers + sample) | `catalog/introspect/tabular.py` | TAB | Each XLSX sheet → one Table |
 | 7 | `BaseIntrospector` ABC | `catalog/introspect/base.py` | B | Confirm signature returns the same `Source` shape |
 | **Ingestion — shared catalog plumbing** | | | | |
-| 8 | Catalog enricher + prompt | `catalog/enricher.py`, `config/prompts/catalog_enricher.md` | B | Whoever picks it up first; the other reviews. Prompt must work uniformly across source types |
+| 8 | ~~Catalog enricher + prompt~~ | — | **REMOVED in KM-557.** Cost optimization — planner reads stats + sample rows directly. `catalog/render.py` keeps the source-rendering helper. |
 | 9 | Catalog validator | `catalog/validator.py` | B | Type-agnostic |
 | 10 | Catalog store (Postgres jsonb) | `catalog/store.py` | B | Recommend DB (Postgres expertise) |
 | 11 | Catalog reader | `catalog/reader.py` | B | Type-agnostic |

main.py

@@ -11,6 +11,7 @@ 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
 import uvicorn
 
@@ -37,6 +38,7 @@ 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")

src/api/v1/data_catalog.py

@@ -0,0 +1,67 @@
+"""API endpoints for the per-user data catalog index.
+
+The index is a lightweight summary of every structured source registered
+by a user (DB connections and tabular files). It is intended to be
+consumed by the catalog refresher and by frontend listings — full
+catalog payloads (tables + columns + samples + stats) are not exposed
+here on purpose.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, HTTPException, status
+
+from src.catalog.store import CatalogStore
+from src.middlewares.logging import get_logger, log_execution
+from src.models.api.catalog import CatalogIndexEntry
+
+logger = get_logger("data_catalog_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Data Catalog"])
+
+
+@router.get(
+    "/data-catalog/{user_id}",
+    response_model=List[CatalogIndexEntry],
+    summary="List the user's data catalog index",
+    response_description="One entry per registered structured source.",
+    responses={
+        200: {"description": "Returns an empty list if the user has no registered sources."},
+        500: {"description": "Internal server error while reading the catalog."},
+    },
+)
+@log_execution(logger)
+async def list_data_catalog_index(user_id: str):
+    """
+    Return a lightweight index of every structured source registered by the user.
+
+    One entry per source (DB connection or tabular file), including the
+    `source_id`, `source_type`, display `name`, `location_ref`, current
+    `table_count`, and `updated_at` timestamp.
+
+    Used by the catalog refresher to decide which sources need to be
+    rebuilt. Returns an empty list if the user has no catalog yet.
+    """
+    try:
+        catalog = await CatalogStore().get(user_id)
+    except Exception as e:
+        logger.error("Failed to read catalog index", user_id=user_id, error=str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to read catalog index: {e}",
+        )
+
+    if catalog is None:
+        return []
+
+    return [
+        CatalogIndexEntry(
+            source_id=s.source_id,
+            source_type=s.source_type,
+            name=s.name,
+            location_ref=s.location_ref,
+            table_count=len(s.tables),
+            updated_at=s.updated_at,
+        )
+        for s in catalog.sources
+    ]

src/catalog/introspect/base.py

@@ -1,7 +1,8 @@
 """BaseIntrospector — contract for source-specific schema readers.
 
-Subclasses produce a draft Source object with raw schema (names + types +
-sample values). The CatalogEnricher then adds descriptions in a separate step.
+Subclasses produce a Source object with raw schema (names, types, sample
+values, stats). The planner consumes this directly — descriptions are not
+LLM-generated.
 """
 
 from abc import ABC, abstractmethod

src/catalog/introspect/database.py

@@ -1,8 +1,8 @@
 """Database schema introspection (Postgres / MySQL / Supabase).
 
 Reads information_schema for tables/columns/types, samples ~100 rows per table
-for `sample_values` and basic stats. Does NOT generate descriptions
-(that happens in CatalogEnricher).
+for `sample_values` and basic stats. Description fields are left empty —
+the planner relies on names + samples + stats directly.
 
 Reuses Phase 1 utilities (`database_client_service`, `db_credential_encryption`,
 `db_pipeline_service.engine_scope`, `extractor.get_schema/profile_column/get_row_count`)

src/catalog/introspect/tabular.py

@@ -75,8 +75,7 @@ class TabularIntrospector(BaseIntrospector):
     """Read column names, dtypes, and sample values from Parquet/CSV/XLSX.
 
     Heavy I/O dependencies (`fetch_doc`, `fetch_blob`) are injectable so unit
-    tests can pass mocks without triggering Settings or DB construction — same
-    pattern as CatalogEnricher's `structured_chain` parameter.
+    tests can pass mocks without triggering Settings or DB construction.
     """
 
     def __init__(

src/catalog/{enricher.py → render.py}

@@ -1,92 +1,16 @@
-"""CatalogEnricher — runs 1 LLM call per source to generate AI descriptions.
-
-Input: a draft Source produced by an introspector (raw schema, no descriptions).
-Output: the same Source enriched with description fields at source / table /
-column level. Other fields (sample_values, stats, foreign_keys, ids, etc.) are
-preserved verbatim — the LLM only emits descriptions keyed by stable IDs and
-they are merged back in.
-
-Prompt: `src/config/prompts/catalog_enricher.md`.
-LLM: Azure OpenAI GPT-4o by default. The structured-output runnable is
-injectable via the constructor for testability.
-"""
+"""Render a `Source` into the canonical text block consumed by the planner."""
 
 from __future__ import annotations
 
-from pathlib import Path
-
-from langchain_core.prompts import ChatPromptTemplate
-from langchain_core.runnables import Runnable
-from langchain_openai import AzureChatOpenAI
-from pydantic import BaseModel, Field
-
-from src.middlewares.logging import get_logger
-
 from .models import Source
 
-logger = get_logger("catalog_enricher")
-
-_PROMPT_PATH = (
-    Path(__file__).resolve().parent.parent / "config" / "prompts" / "catalog_enricher.md"
-)
-
-
-class _IdDescription(BaseModel):
-    """One description keyed by a stable identifier from the input Source.
-
-    `target_id` MUST match a `source_id`, `table_id`, or `column_id` exactly
-    as it appeared in the rendered input — do not regenerate or transform.
-    """
-
-    target_id: str = Field(
-        ...,
-        description="source_id, table_id, or column_id — copied verbatim from the input.",
-    )
-    description: str = Field(
-        ...,
-        description="One-line factual description grounded in the input.",
-    )
-
-
-class EnrichmentResponse(BaseModel):
-    """Structured output: a flat list of (id, description) pairs."""
-
-    descriptions: list[_IdDescription] = Field(default_factory=list)
-
-
-def _load_prompt_text() -> str:
-    return _PROMPT_PATH.read_text(encoding="utf-8")
-
-
-def _build_default_chain() -> Runnable:
-    """Construct the production LangChain runnable.
-
-    `settings` is imported lazily so importing this module is side-effect-free
-    for tests that inject a mock `structured_chain` (otherwise
-    `Settings()` would fail without a populated `.env`).
-    """
-    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.2,
-    )
-    prompt = ChatPromptTemplate.from_messages(
-        [
-            ("system", _load_prompt_text()),
-            ("human", "{source_text}"),
-        ]
-    )
-    return prompt | llm.with_structured_output(EnrichmentResponse)
-
 
 def render_source(source: Source) -> str:
-    """Render a Source as the text that the enricher prompt expects.
+    """Render a Source as the canonical text block consumed by the planner.
 
-    Public so tests and the planner-prompt builder can reuse the same format.
+    Includes stable IDs (so the LLM can echo them back), per-column data
+    type, sample values (or `PII (suppressed)` for flagged columns), basic
+    stats, and resolved-by-name foreign keys.
     """
     lines: list[str] = [
         f"Source: {source.name} ({source.source_type})",
@@ -133,64 +57,3 @@ def render_source(source: Source) -> str:
                 )
                 lines.append(f"    - {src_col_name} -> {tgt_table_name}.{tgt_col_name}")
     return "\n".join(lines)
-
-
-def apply_descriptions(source: Source, response: EnrichmentResponse) -> Source:
-    """Merge LLM-emitted descriptions back into the Source by stable ID.
-
-    Items the LLM omitted retain their existing description (usually ""
-    from the introspector). All other fields are preserved verbatim.
-    """
-    by_id = {d.target_id: d.description for d in response.descriptions}
-
-    new_tables = []
-    for table in source.tables:
-        new_columns = [
-            col.model_copy(
-                update={"description": by_id.get(col.column_id, col.description)}
-            )
-            for col in table.columns
-        ]
-        new_tables.append(
-            table.model_copy(
-                update={
-                    "description": by_id.get(table.table_id, table.description),
-                    "columns": new_columns,
-                }
-            )
-        )
-
-    return source.model_copy(
-        update={
-            "description": by_id.get(source.source_id, source.description),
-            "tables": new_tables,
-        }
-    )
-
-
-class CatalogEnricher:
-    """Adds AI-generated descriptions to a freshly introspected source.
-
-    Inject `structured_chain` for tests; default builds an Azure OpenAI
-    GPT-4o chain wired to `with_structured_output(EnrichmentResponse)`.
-    """
-
-    def __init__(self, structured_chain: Runnable | None = None) -> None:
-        self._chain = structured_chain
-
-    def _ensure_chain(self) -> Runnable:
-        if self._chain is None:
-            self._chain = _build_default_chain()
-        return self._chain
-
-    async def enrich(self, source: Source) -> Source:
-        rendered = render_source(source)
-        chain = self._ensure_chain()
-        response: EnrichmentResponse = await chain.ainvoke({"source_text": rendered})
-        logger.info(
-            "catalog source enriched",
-            source_id=source.source_id,
-            descriptions=len(response.descriptions),
-            tables=len(source.tables),
-        )
-        return apply_descriptions(source, response)

src/config/prompts/catalog_enricher.md

@@ -1,62 +0,0 @@
-You are a senior data analyst writing concise, factual descriptions for a user's data catalog. The catalog is consumed by an AI agent that helps the same user query their data, so descriptions must accurately convey what each source / table / column **is**, not what an analyst might guess it could be.
-
-## Your task
-
-Given a single data source rendered below, produce a description for the source itself, each table inside it, and each column inside each table. Return your output as structured JSON matching the requested schema; identifiers (`source_id`, `table_id`, `column_id`) must be copied **verbatim** from the input.
-
-## Style rules
-
-- **One factual sentence per item.** Two only if a second is genuinely necessary (e.g., to call out a unit or important caveat).
-- **Ground every claim in the evidence shown** — column names, sample values, stats, foreign keys, table names. Do not invent semantics that the inputs do not support.
-- **Do not restate the data type.** ("`amount` (decimal)" is redundant; say what `amount` represents.)
-- **Mention obvious units / scales** when sample values make them clear (e.g., "in cents", "in milliseconds", "0–100 score").
-- **PII columns have `samples=PII (suppressed)`.** Describe their role from the column name only — do not speculate about content. Example: "Customer's email address." (correct), not "Email address, e.g. alice@example.com" (wrong, invented).
-- **Foreign keys**: when present, mention the relationship at the table level (e.g., "Each row links to a customer via `customer_id`"). Don't repeat the FK in the column description unless it's the primary point.
-- **Do not include sample values verbatim in any description.** Use them as evidence to infer meaning, not as content to quote.
-- **No markdown formatting in descriptions** (no bold, lists, code fences). Plain text only.
-
-## Source-level description
-
-One or two sentences describing what kind of data this source holds and what the user might use it for. If the source has only one table, the source and table descriptions can overlap but should not be identical word-for-word.
-
-## Table-level description
-
-What real-world entity or event each row represents. Mention the grain (one row per …) if non-obvious.
-
-## Column-level description
-
-What the column **means**, not what it stores. Read sample values, ranges, and the column name together to triangulate. If genuinely ambiguous, write a description that captures the ambiguity rather than picking one interpretation arbitrarily.
-
-## Few-shot example
-
-Input (rendered source):
-
-```
-Source: prod_db (schema)
-
-Tables:
-
-  Table: orders (12,453 rows)
-  Columns:
-    - id [int]: samples=[1, 2, 3], min=1, max=12453, distinct=12453
-    - customer_id [int]: samples=[42, 17, 99], min=1, max=8200, distinct=8200
-    - total_cents [int]: samples=[2499, 4999, 1999], min=99, max=999900, distinct=4321
-    - status [string]: samples=[completed, pending, refunded], distinct=4
-    - created_at [datetime]: samples=[2026-04-01T08:12:00Z, 2026-04-01T08:14:00Z], distinct=12440
-  Foreign keys:
-    - customer_id -> customers.id
-```
-
-Expected descriptions:
-
-- Source: "Production order ledger for the storefront — one row per checkout, used for reporting on revenue, fulfillment, and customer purchase history."
-- Table `orders`: "One row per completed or attempted checkout; links each order to the customer who placed it."
-- Column `id`: "Unique order identifier."
-- Column `customer_id`: "Identifier of the customer who placed the order; references customers.id."
-- Column `total_cents`: "Order total in cents (USD), inclusive of taxes and discounts."
-- Column `status`: "Lifecycle state of the order — values include completed, pending, and refunded."
-- Column `created_at`: "Timestamp when the order was placed (UTC)."
-
-## Output
-
-Return strict JSON matching the schema requested by the structured-output binding. The shape mirrors the input source: a flat map of identifiers to descriptions is acceptable but the full nested form is preferred. Identifiers must match the input exactly — do not regenerate them.

src/db/postgres/init_db.py

@@ -22,7 +22,7 @@ async def init_db():
         await conn.execute(text("SELECT pg_advisory_xact_lock(1573678846307946496)"))
         await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
 
-        # Create application tables
+        # Create application tables (includes `data_catalog`)
         await conn.run_sync(Base.metadata.create_all)
 
         # Schema migrations (idempotent — safe to run on every startup)

src/db/postgres/models.py

@@ -104,8 +104,11 @@ class Catalog(Base):
     `data` holds the full Pydantic Catalog (src/catalog/models.py:Catalog)
     serialized via `model_dump(mode="json")`. Read path uses
     `Catalog.model_validate(...)` to rehydrate.
+
+    Dedicated table — kept separate from `langchain_pg_embedding` so unstructured
+    embeddings and structured-catalog metadata never share storage.
     """
-    __tablename__ = "catalogs"
+    __tablename__ = "data_catalog"
 
     user_id = Column(String, primary_key=True)
     data = Column(JSONB, nullable=False)

src/models/api/catalog.py

@@ -1,6 +1,8 @@
-"""Request / response models for catalog-related routes (e.g. /knowledge/rebuild)."""
+"""Request / response models for catalog-related routes."""
 
-from pydantic import BaseModel
+from datetime import datetime
+
+from pydantic import BaseModel, Field
 
 
 class CatalogRebuildRequest(BaseModel):
@@ -9,6 +11,17 @@ class CatalogRebuildRequest(BaseModel):
 
 class CatalogRebuildResponse(BaseModel):
     user_id: str
-    sources_enriched: int
-    tables_enriched: int
-    columns_enriched: int
+    sources_rebuilt: int
+
+
+class CatalogIndexEntry(BaseModel):
+    """One row in the per-user catalog index — used by the refresher to decide
+    which sources to rebuild and by the UI to list registered sources.
+    """
+
+    source_id: str = Field(..., description="Stable internal source identifier.")
+    source_type: str = Field(..., description="schema | tabular | unstructured.")
+    name: str = Field(..., description="Display name (DB name or filename).")
+    location_ref: str = Field(..., description="URI: dbclient://… or az_blob://…")
+    table_count: int = Field(..., description="Number of tables/sheets in this source.")
+    updated_at: datetime = Field(..., description="Last time this source was (re)introspected.")

src/pipeline/structured_pipeline.py

@@ -1,18 +1,17 @@
-"""StructuredPipeline — runs catalog enrichment for DB / tabular sources.
+"""StructuredPipeline — builds a catalog for DB / tabular sources.
 
 Steps (per source, end-to-end):
   1. introspect (caller-supplied — DatabaseIntrospector or TabularIntrospector)
-  2. enrich    (catalog/enricher.py — 1 LLM call per source)
-  3. merge     (replace any existing source with the same source_id)
-  4. validate  (catalog/validator.py)
-  5. upsert    (catalog/store.py)
+  2. merge     (replace any existing source with the same source_id)
+  3. validate  (catalog/validator.py)
+  4. upsert    (catalog/store.py)
+
+LLM-driven enrichment was removed: the planner relies on stats + sample
+rows + column names directly. Source/table/column `description` fields stay
+in the model but are not populated by this pipeline.
 
 Source-type-agnostic: the caller picks the introspector. Triggers in
 `pipeline/triggers.py` know which one to use based on the upload event.
-
-Heavy production deps (`CatalogEnricher`, `CatalogStore`) are imported lazily
-inside `default_structured_pipeline()` so unit tests that inject mocks never
-trigger `Settings()` construction at module import.
 """
 
 from __future__ import annotations
@@ -25,7 +24,6 @@ from src.catalog.models import Catalog, Source
 from src.middlewares.logging import get_logger
 
 if TYPE_CHECKING:
-    from src.catalog.enricher import CatalogEnricher
     from src.catalog.store import CatalogStore
     from src.catalog.validator import CatalogValidator
 
@@ -33,7 +31,7 @@ logger = get_logger("structured_pipeline")
 
 
 class StructuredPipeline:
-    """Orchestrates introspect → enrich → merge → validate → store.
+    """Orchestrates introspect → merge → validate → store.
 
     Dependencies are injected (no concrete imports at class-definition time)
     so tests can pass mocks without constructing Settings or opening DB
@@ -42,11 +40,9 @@ class StructuredPipeline:
 
     def __init__(
         self,
-        enricher: CatalogEnricher,
         validator: CatalogValidator,
         store: CatalogStore,
     ) -> None:
-        self._enricher = enricher
         self._validator = validator
         self._store = store
 
@@ -57,18 +53,17 @@ class StructuredPipeline:
         user_id: str,
     ) -> Source:
         source = await introspector.introspect(location_ref)
-        enriched = await self._enricher.enrich(source)
-        merged = await self._merge_with_existing(user_id, enriched)
+        merged = await self._merge_with_existing(user_id, source)
         self._validator.validate(merged)
         await self._store.upsert(merged)
         logger.info(
             "structured pipeline complete",
             user_id=user_id,
-            source_id=enriched.source_id,
-            source_type=enriched.source_type,
-            tables=len(enriched.tables),
+            source_id=source.source_id,
+            source_type=source.source_type,
+            tables=len(source.tables),
         )
-        return enriched
+        return source
 
     async def _merge_with_existing(self, user_id: str, new_source: Source) -> Catalog:
         existing = await self._store.get(user_id)
@@ -87,12 +82,10 @@ def default_structured_pipeline() -> StructuredPipeline:
     Lazy imports keep `from src.pipeline.structured_pipeline import …` cheap
     and side-effect-free for tests.
     """
-    from src.catalog.enricher import CatalogEnricher
     from src.catalog.store import CatalogStore
     from src.catalog.validator import CatalogValidator
 
     return StructuredPipeline(
-        enricher=CatalogEnricher(),
         validator=CatalogValidator(),
         store=CatalogStore(),
     )

src/pipeline/triggers.py

@@ -19,8 +19,7 @@ async def on_db_registered(database_client_id: str, user_id: str) -> None:
     Called by `/api/v1/database-clients/{id}/ingest` (after rewiring in a
     later PR). The DatabaseIntrospector resolves the client_id to a
     DatabaseClient row, decrypts credentials, connects, and produces a Source.
-    The CatalogEnricher then fills in descriptions, the catalog is validated
-    and upserted.
+    The catalog is then validated and upserted (no LLM enrichment step).
     """
     from src.catalog.introspect.database import database_introspector
     from src.pipeline.structured_pipeline import default_structured_pipeline
@@ -40,8 +39,8 @@ async def on_tabular_uploaded(document_id: str, user_id: str) -> None:
 
     Called after a CSV/XLSX/Parquet file has been processed and its Parquet
     blob(s) uploaded. The TabularIntrospector downloads the original blob,
-    profiles each column, and produces a Source. The CatalogEnricher then fills
-    in descriptions, the catalog is validated and upserted.
+    profiles each column, and produces a Source. The catalog is then validated
+    and upserted (no LLM enrichment step).
     """
     from src.catalog.introspect.tabular import tabular_introspector
     from src.pipeline.structured_pipeline import default_structured_pipeline
@@ -65,6 +64,10 @@ 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, useful after enricher prompt
-    changes. Implemented when the bulk re-enrichment script lands."""
+    """Stub — re-runs every source for a user (catalog refresher).
+
+    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.
+    """
     raise NotImplementedError

src/query/planner/prompt.py

@@ -2,16 +2,12 @@
 
 Renders the catalog into a compact textual form that fits the LLM context
 window. For users with ≤50 tables the full catalog goes in verbatim.
-
-Reuses `catalog.enricher.render_source` so the planner sees the same
-source-rendering format as the enricher does at ingestion time — keeping
-catalog descriptions consistent across both LLM call sites.
 """
 
 from __future__ import annotations
 
-from ...catalog.enricher import render_source
 from ...catalog.models import Catalog
+from ...catalog.render import render_source
 
 
 def render_catalog(catalog: Catalog) -> str:

src/security/README.md

@@ -3,6 +3,6 @@
 Cross-cutting security primitives:
 - credential encryption (Fernet) for stored DB credentials
 - authentication / password / JWT helpers
-- PII detection patterns used by the catalog enricher
+- PII detection patterns used by the catalog introspectors
 
 Consolidates utilities previously split between `utils/` and `users/`.