Merge commit 'refs/pr/1' of https://huggingface.co/spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog into pr/1

PROGRESS.md

@@ -0,0 +1,205 @@
+# Progress — Phase 2 catalog-driven build
+
+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-07 (PR2a — enricher + structured pipeline shipped)
+**Current open PR**: PR2a (DB owner — CatalogEnricher + StructuredPipeline + on_db_registered trigger + FK extension)
+
+---
+
+## Legend
+
+- `[x]` done and merged
+- `[~]` in progress (open PR or active branch)
+- `[ ]` not started
+- **DB** / **TAB** / **B** — ownership (from REPO_CONTEXT.md)
+
+---
+
+## PR sequence
+
+| PR | Status | Owner(s) | Scope |
+|---|---|---|---|
+| PR1 | `[x]` merged | DB | Contract locks + catalog plumbing + DB introspector + IR validator + tests |
+| PR1-tab | `[ ]` | TAB | Tabular introspector + golden IR examples for tabular |
+| PR2a | `[~]` open | DB | CatalogEnricher + StructuredPipeline + on_db_registered trigger + FK extension on Table |
+| PR2b | `[ ]` | B | IntentRouter + planner prompt (pair) + planner LLM service |
+| PR3 | `[ ]` | B (split) | SQL compiler + DB executor (DB), pandas compiler + tabular executor (TAB) |
+| PR4 | `[ ]` | B (pair) | ExecutorDispatcher + QueryService + chat stream endpoint integration |
+| PR5 | `[ ]` | B | Retry/self-correction loop on execution failure |
+| PR6 | `[ ]` | B | Eval harness (golden question→IR→result examples) |
+| PR7 | `[ ]` | B | Auto PII tagging review + ChatbotAgent rewrite + API rewiring |
+| Cleanup | `[ ]` | B | Remove Phase 1 (rag/, query/executors/, database_client/, …) once Phase 2 has feature parity |
+
+---
+
+## All items
+
+### Contracts (B — shared)
+
+| # | Item | Status | Notes |
+|---|---|---|---|
+| 1 | Catalog Pydantic models (`catalog/models.py`) | `[x]` | PR1 added `location_ref` URI-scheme docstring; PR2a added `ForeignKey` model + `Table.foreign_keys` field |
+| 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 |
+| — | `QueryResult` shape (`query/executor/base.py`) | `[x]` | Pre-existing scaffold; revisit in PR3 if `column_types` needed |
+| — | `Source.location_ref` URI scheme | `[x]` | PR1 documented in `catalog/models.py` docstring |
+
+### Ingestion — introspection
+
+| # | Item | Owner | Status | Notes |
+|---|---|---|---|---|
+| 5 | DB introspector (`catalog/introspect/database.py`) | DB | `[x]` | PR1 — reuses Phase 1 `database_client_service`, `db_credential_encryption`, `db_pipeline_service.engine_scope`, `extractor.get_schema/profile_column/get_row_count`. PR2a wired FK extraction (was discarded before). |
+| 6 | Tabular introspector (`catalog/introspect/tabular.py`) | TAB | `[ ]` | XLSX sheet → one Table; `Source.source_id = document_id` |
+| 7 | `BaseIntrospector` ABC (`catalog/introspect/base.py`) | B | `[x]` | Pre-existing; signature locked |
+
+### Ingestion — shared catalog plumbing
+
+| # | 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. |
+| 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 |
+| 12 | PII detector (`catalog/pii_detector.py`) | B | `[x]` | PR1 (DB owner picked up) — name + value matching, bias toward over-flag |
+
+### Ingestion — pipelines
+
+| # | 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. |
+| 14 | Triggers (`pipeline/triggers.py`) | B | `[~]` | PR2a — `on_db_registered` implemented (DB owner). `on_tabular_uploaded`, `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 | `[ ]` | Tabular-adjacent (file uploads). Phase 1 implementation exists at `pipeline/document_pipeline/document_pipeline.py` — reuse or rewrite TBD |
+
+### Query — shared spine
+
+| # | Item | Owner | Status | Notes |
+|---|---|---|---|---|
+| 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 | `[ ]` | PR2b |
+| 19 | Planner prompt (`query/planner/prompt.py`, `config/prompts/query_planner.md`) | B | `[ ]` | PR2b — **pair-program**; must render DB and tabular sources uniformly. Can reuse `catalog.enricher.render_source` as a starting point. |
+| 20 | Intent router (`agents/intent_router.py`, `config/prompts/intent_router.md`) | B | `[ ]` | PR2b |
+| 21 | Executor base + `QueryResult` (`query/executor/base.py`) | B | `[x]` | Pre-existing scaffold |
+| 22 | Executor dispatcher (`query/executor/dispatcher.py`) | B | `[ ]` | PR4 — `(Catalog, IR) → BaseExecutor` |
+| 23 | Compiler base ABC (`query/compiler/base.py`) | B | `[x]` | Pre-existing scaffold |
+| 24 | Top-level QueryService (`query/service.py`) | B | `[ ]` | PR4 — wires planner → validator → compiler → executor |
+
+### Query — DB path
+
+| # | Item | Status | Notes |
+|---|---|---|---|
+| 25 | SQL compiler (`query/compiler/sql.py`) | `[ ]` | PR3 — IR → (sql, params); identifiers from catalog (quoted), values parameterized |
+| 26 | DB executor (`query/executor/db.py`) | `[ ]` | PR3 — asyncpg/pymysql; sqlglot SELECT-only guard; RO txn; 30s timeout |
+| 27 | Credential encryption (`security/credentials.py`) | `[ ]` | Stub exists; PR1 reused Phase 1 `utils/db_credential_encryption.py` instead. Move in cleanup PR |
+| 28 | User-DB connection management | `[ ]` | Reused Phase 1 `db_pipeline_service.engine_scope` in PR1; new helper in PR3 if needed |
+
+### Query — Tabular path
+
+| # | Item | Status | Notes |
+|---|---|---|---|
+| 29 | Pandas compiler (`query/compiler/pandas.py`) | `[ ]` | PR3 — IR → callable on DataFrame |
+| 30 | Tabular executor (`query/executor/tabular.py`) | `[ ]` | PR3 — eager pandas first; pyarrow/polars later if file size demands |
+| 31 | Parquet upload/download wrapper | `[ ]` | Phase 1 has `knowledge/parquet_service.py` — reuse or move to `storage/` in cleanup |
+
+### Agents + chat
+
+| # | Item | Status | Notes |
+|---|---|---|---|
+| 32 | Chatbot agent + prompt (`agents/chatbot.py`, `config/prompts/chatbot_system.md`) | `[ ]` | PR7 — receives `QueryResult` or Cu chunks |
+| 33 | Guardrails prompt (`config/prompts/guardrails.md`) | `[ ]` | PR7 |
+
+### API surface
+
+| # | Item | Owner | Status | Notes |
+|---|---|---|---|---|
+| 34 | DB client endpoints (`api/v1/db_client.py`) | DB | `[ ]` | Phase 1 endpoint exists — rewire `/ingest` to call `pipeline.triggers.on_db_registered`. Trigger is ready as of PR2a; deferred to a later PR until both teammates ack. |
+| 35 | Document/tabular upload endpoints (`api/v1/document.py`) | TAB | `[ ]` | Phase 1 endpoint exists — rewire after enricher |
+| 36 | Chat stream endpoint (`api/v1/chat.py`) | B | `[ ]` | PR4 — pair on dispatch logic; SSE event sequence stays |
+| 37 | Room / users endpoints (`api/v1/room.py`, `api/v1/users.py`) | B | `[ ]` | No catalog work; only touch if auth flow changes |
+
+### Tests + eval
+
+| # | Item | Owner | Status | Notes |
+|---|---|---|---|---|
+| 38 | DB compiler golden tests (`tests/query/compiler/test_sql.py`) | DB | `[ ]` | PR3 — pure-Python, no LLM |
+| 39 | Pandas compiler golden tests (`tests/query/compiler/test_pandas.py`) | TAB | `[ ]` | PR3 — pure-Python, no LLM |
+| 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 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 | `[ ]` | TAB to add when introspector lands |
+| 41 | Planner eval (`tests/query/planner/`) | B | `[ ]` | PR6 — golden question → IR examples; each side contributes |
+| 42 | E2E smoke tests (`tests/e2e/`) | B | `[ ]` | PR4 — pair |
+| — | Golden IR fixtures (`tests/fixtures/golden_irs.json`) | B | `[~]` | PR1 seeded with 5 DB-targeting examples; TAB extends in PR1-tab |
+| — | Shared `sample_catalog` fixture (`tests/conftest.py`) | B | `[x]` | PR1 — DB-shaped; TAB may add tabular sibling |
+
+---
+
+## What just shipped (PR2a — DB owner)
+
+**Files implemented**:
+- `src/catalog/enricher.py` — Azure OpenAI GPT-4o + structured output (`EnrichmentResponse`), `render_source` (reusable by planner prompt later), `apply_descriptions` merger, injectable `structured_chain` for tests
+- `src/pipeline/structured_pipeline.py` — `StructuredPipeline` orchestrator + `default_structured_pipeline()` factory with lazy production-dep imports
+- `src/pipeline/triggers.py` — `on_db_registered` wired; tabular/document/rebuild stubs preserved with implementation notes
+
+**Files extended**:
+- `src/catalog/models.py` — added `ForeignKey` model, `Table.foreign_keys: list[ForeignKey] = []`
+- `src/catalog/introspect/database.py` — `_extract_foreign_keys` populates `Table.foreign_keys` from extractor data
+- `src/config/prompts/catalog_enricher.md` — full system prompt with style rules and one few-shot example
+
+**Tests added** (14 new, all passing — total now 64):
+- `tests/catalog/test_enricher.py` — render / apply / end-to-end with fake chain (10 tests)
+- `tests/pipeline/test_structured_pipeline.py` — orchestration with stub deps (4 tests)
+
+**Lint**: `ruff check` clean on all Phase 2 paths. Phase 1 files (`pipeline/db_pipeline/`, `pipeline/document_pipeline/`) have pre-existing ruff issues — out of scope for this PR.
+
+---
+
+## What shipped previously (PR1 — DB owner's first chunk)
+
+**Files implemented** (was `NotImplementedError`):
+- `src/catalog/pii_detector.py`, `src/catalog/validator.py`, `src/catalog/store.py`, `src/catalog/reader.py`
+- `src/catalog/introspect/database.py` (FK extraction added in PR2a)
+- `src/query/ir/validator.py`
+
+**Files extended**:
+- `src/query/ir/operators.py` — `TYPE_COMPATIBILITY` matrix
+- `src/catalog/models.py` — `location_ref` URI-scheme docstring
+- `src/db/postgres/models.py` — `Catalog` SQLAlchemy table; `init_db.py` imports it
+
+**Tests**: 50 unit tests + 1 integration (gated on `RUN_INTEGRATION_TESTS=1`).
+
+**Reused Phase 1 utilities** (cleanup deferred):
+- `src/database_client/database_client_service.py:get`
+- `src/utils/db_credential_encryption.py:decrypt_credentials_dict`
+- `src/pipeline/db_pipeline/db_pipeline_service.py:engine_scope`
+- `src/pipeline/db_pipeline/extractor.py:get_schema/profile_column/get_row_count`
+
+---
+
+## Open contract items (not yet locked)
+
+- **Joins in IR** — currently single-table only (ARCHITECTURE.md §7); DB owner accepted the constraint for v1, will revisit in PR3 if it's blocking real queries
+- **`updated_at` on Source vs `generated_at` on Catalog** — Pydantic models have both; introspector sets per-Source; CatalogStore preserves both
+- **Catalog refresh trigger** (open question §3) — default policy is rebuild-on-upload-or-connect; auto-refresh deferred
+- **Unstructured catalog entries** (open question §2) — currently empty filter for `source_hint="unstructured"`; revisit when adding doc descriptions
+- **PII handling for `sample_values`** (open question §5) — currently nulls them out (skip); mask/synthesize deferred
+- **Dialect priority for SQL compiler** — PR3 will land Postgres first, MySQL second; BigQuery/Snowflake/SQL Server later
+
+---
+
+## How to update this file
+
+When a PR lands:
+1. Flip status from `[ ]` or `[~]` to `[x]`
+2. Add a short note (file paths, scope cuts, surprises)
+3. Bump "Last updated" at the top
+4. If a new contract decision lands, move it from "Open contract items" to the relevant inline note
+
+When opening a PR:
+1. Flip status to `[~]` and add yourself as the active owner in the PR row
+2. Don't promise items in the PR description that aren't in the table

REPO_CONTEXT.md

@@ -227,16 +227,19 @@ Catalog
 └── sources[]
     └── Source { source_id, source_type, name, description, location_ref, updated_at }
         └── tables[]
-            └── Table { table_id, name, description, row_count }
-                └── columns[]
-                    └── Column { column_id, name, data_type, description,
-                                  nullable, pii_flag, sample_values[]|null, stats|null }
+            └── Table { table_id, name, description, row_count, foreign_keys[] }
+                ├── columns[]
+                │   └── Column { column_id, name, data_type, description,
+                │                 nullable, pii_flag, sample_values[]|null, stats|null }
+                └── foreign_keys[]
+                    └── ForeignKey { column_id, target_table_id, target_column_id }
 ```
 
 `source_type ∈ {schema, tabular, unstructured}`.
 `data_type ∈ {int, decimal, string, datetime, date, bool, json}`.
+`ForeignKey` references are within the SAME `Source` only; cross-source FKs are not modeled.
 
-Deferred Column fields (add when justified): `description_human`, `synonyms[]`, `tags[]`, `primary_key`, `foreign_keys`, `unit`, `semantic_type`, `example_questions[]`, `schema_hash`, `enrichment_status`.
+Deferred Column fields (add when justified): `description_human`, `synonyms[]`, `tags[]`, `primary_key`, `unit`, `semantic_type`, `example_questions[]`, `schema_hash`, `enrichment_status`.
 
 ---
 

src/catalog/enricher.py

@@ -1,16 +1,196 @@
 """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.
+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
+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.
 """
 
+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.
+
+    Public so tests and the planner-prompt builder can reuse the same format.
+    """
+    lines: list[str] = [
+        f"Source: {source.name} ({source.source_type})",
+        f"Source ID: {source.source_id}",
+        "",
+        "Tables:",
+    ]
+
+    tables_by_id = {t.table_id: t for t in source.tables}
+    col_names_by_id = {
+        t.table_id: {c.column_id: c.name for c in t.columns} for t in source.tables
+    }
+
+    for table in source.tables:
+        rc = table.row_count
+        rc_str = f"({rc:,} rows) " if rc is not None else ""
+        lines.append("")
+        lines.append(f"  Table: {table.name} {rc_str}— id={table.table_id}")
+        lines.append("  Columns:")
+        for col in table.columns:
+            samples = "PII (suppressed)" if col.pii_flag else (col.sample_values or [])
+            stats_parts: list[str] = []
+            if col.stats:
+                if col.stats.min is not None:
+                    stats_parts.append(f"min={col.stats.min}")
+                if col.stats.max is not None:
+                    stats_parts.append(f"max={col.stats.max}")
+                if col.stats.distinct_count is not None:
+                    stats_parts.append(f"distinct={col.stats.distinct_count}")
+            stats_str = (", " + ", ".join(stats_parts)) if stats_parts else ""
+            lines.append(
+                f"    - {col.name} [{col.data_type}]: samples={samples}{stats_str} "
+                f"— id={col.column_id}"
+            )
+        if table.foreign_keys:
+            lines.append("  Foreign keys:")
+            cols_in_this_table = {c.column_id: c.name for c in table.columns}
+            for fk in table.foreign_keys:
+                src_col_name = cols_in_this_table.get(fk.column_id, fk.column_id)
+                tgt_table = tables_by_id.get(fk.target_table_id)
+                tgt_table_name = tgt_table.name if tgt_table else fk.target_table_id
+                tgt_col_name = col_names_by_id.get(fk.target_table_id, {}).get(
+                    fk.target_column_id, fk.target_column_id
+                )
+                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."""
+    """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:
-        raise NotImplementedError
+        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/catalog/introspect/database.py

@@ -27,7 +27,7 @@ from src.pipeline.db_pipeline.extractor import (
 )
 from src.utils.db_credential_encryption import decrypt_credentials_dict
 
-from ..models import Column, ColumnStats, DataType, Source, Table
+from ..models import Column, ColumnStats, DataType, ForeignKey, Source, Table
 from ..pii_detector import PIIDetector
 from .base import BaseIntrospector
 
@@ -171,6 +171,8 @@ class DatabaseIntrospector(BaseIntrospector):
                         continue
                     columns.append(self._to_column(table_name, col, profile))
 
+                foreign_keys = self._extract_foreign_keys(table_name, cols)
+
                 tables.append(
                     Table(
                         table_id=_stable_id("t_", table_name),
@@ -178,10 +180,36 @@ class DatabaseIntrospector(BaseIntrospector):
                         description="",
                         row_count=row_count,
                         columns=columns,
+                        foreign_keys=foreign_keys,
                     )
                 )
         return tables
 
+    @staticmethod
+    def _extract_foreign_keys(
+        table_name: str, cols: list[dict[str, Any]]
+    ) -> list[ForeignKey]:
+        """Convert extractor's `foreign_key: 'target_table.target_col'` strings
+        into ForeignKey objects with stable IDs (derived deterministically from
+        names — same scheme used to generate table_id / column_id elsewhere).
+        """
+        fks: list[ForeignKey] = []
+        for col in cols:
+            fk_str = col.get("foreign_key")
+            if not fk_str:
+                continue
+            target_table, _, target_col = fk_str.partition(".")
+            if not target_table or not target_col:
+                continue
+            fks.append(
+                ForeignKey(
+                    column_id=_stable_id("c_", table_name, col["name"]),
+                    target_table_id=_stable_id("t_", target_table),
+                    target_column_id=_stable_id("c_", target_table, target_col),
+                )
+            )
+        return fks
+
     def _to_column(
         self, table_name: str, col: dict[str, Any], profile: dict[str, Any]
     ) -> Column:

src/catalog/models.py

@@ -48,12 +48,25 @@ class Column(BaseModel):
     stats: ColumnStats | None = None
 
 
+class ForeignKey(BaseModel):
+    """A FK edge from one column in this table to a column in another table.
+
+    All references use stable IDs derived from source/table/column names so
+    edges survive renames at the `name` level. The target table must belong
+    to the SAME `Source` — cross-source FKs are not modeled in v1.
+    """
+    column_id: str           # the column in this table that holds the FK
+    target_table_id: str     # referenced table_id, within the same Source
+    target_column_id: str    # referenced column_id
+
+
 class Table(BaseModel):
     table_id: str
     name: str
     description: str
     row_count: int | None = None
     columns: list[Column]
+    foreign_keys: list[ForeignKey] = Field(default_factory=list)
 
 
 class Source(BaseModel):

src/config/prompts/catalog_enricher.md

@@ -1,21 +1,62 @@
-# Catalog Enricher Prompt
+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.
 
-Takes raw introspected schema (table names, column names, types, sample values)
-and produces AI-generated descriptions for the source, each table, and each column.
+## Your task
 
-One LLM call per source at ingestion time. Used by `src/catalog/enricher.py`.
+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.
 
-## System prompt
+## Style rules
 
-(to be written)
+- **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.
 
-## Output schema
+## Source-level description
 
-Strict JSON matching the catalog Pydantic models in `src/catalog/models.py`.
-Validated by `CatalogValidator` before being persisted.
+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.
 
-## Notes
+## Table-level description
 
-- Descriptions should be one-line and factual.
-- Do NOT invent column meanings beyond what the sample values support.
-- For `pii_flag` columns, do NOT include sample values in the 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/pipeline/structured_pipeline.py

@@ -1,13 +1,98 @@
 """StructuredPipeline — runs catalog enrichment for DB / tabular sources.
 
-Steps:
-1. introspect (catalog/introspect/database.py or catalog/introspect/tabular.py)
-2. enrich    (catalog/enricher.py — 1 LLM call per source)
-3. validate  (catalog/validator.py)
-4. write     (catalog/store.py)
+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)
+
+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
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from src.catalog.introspect.base import BaseIntrospector
+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
+
+logger = get_logger("structured_pipeline")
+
 
 class StructuredPipeline:
-    async def run(self, source_ref: str, source_type: str, user_id: str) -> None:
-        raise NotImplementedError
+    """Orchestrates introspect → enrich → merge → validate → store.
+
+    Dependencies are injected (no concrete imports at class-definition time)
+    so tests can pass mocks without constructing Settings or opening DB
+    connections.
+    """
+
+    def __init__(
+        self,
+        enricher: CatalogEnricher,
+        validator: CatalogValidator,
+        store: CatalogStore,
+    ) -> None:
+        self._enricher = enricher
+        self._validator = validator
+        self._store = store
+
+    async def run(
+        self,
+        introspector: BaseIntrospector,
+        location_ref: str,
+        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)
+        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),
+        )
+        return enriched
+
+    async def _merge_with_existing(self, user_id: str, new_source: Source) -> Catalog:
+        existing = await self._store.get(user_id)
+        now = datetime.now(UTC)
+        if existing is None:
+            return Catalog(user_id=user_id, generated_at=now, sources=[new_source])
+        kept = [s for s in existing.sources if s.source_id != new_source.source_id]
+        return existing.model_copy(
+            update={"sources": [*kept, new_source], "generated_at": now}
+        )
+
+
+def default_structured_pipeline() -> StructuredPipeline:
+    """Build the production pipeline with default deps.
+
+    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

@@ -1,21 +1,63 @@
 """Pipeline trigger entry points called from API routes / event handlers.
 
-These thin functions are what the FastAPI routes invoke; they delegate
-to IngestionOrchestrator.
+These thin functions are what the FastAPI routes invoke; they delegate to the
+appropriate pipeline (StructuredPipeline for DB/tabular, DocumentPipeline for
+unstructured).
+
+Errors propagate from the pipelines — the caller decides whether to surface
+them as HTTP 4xx/5xx or quietly fail. The trigger itself does not catch.
 """
 
+from src.middlewares.logging import get_logger
 
-async def on_document_uploaded(document_id: str, user_id: str) -> None:
-    raise NotImplementedError
+logger = get_logger("pipeline_triggers")
 
 
 async def on_db_registered(database_client_id: str, user_id: str) -> None:
-    raise NotImplementedError
+    """Build a dbclient:// location_ref and run the structured pipeline.
+
+    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.
+    """
+    from src.catalog.introspect.database import database_introspector
+    from src.pipeline.structured_pipeline import default_structured_pipeline
+
+    location_ref = f"dbclient://{database_client_id}"
+    logger.info(
+        "on_db_registered triggered",
+        user_id=user_id,
+        database_client_id=database_client_id,
+    )
+    pipeline = default_structured_pipeline()
+    await pipeline.run(database_introspector, location_ref, user_id)
 
 
 async def on_tabular_uploaded(document_id: str, user_id: str) -> None:
+    """Stub — implemented by TAB owner once `catalog/introspect/tabular.py` lands.
+
+    Expected pattern (mirrors `on_db_registered`):
+
+        from src.catalog.introspect.tabular import tabular_introspector
+        from src.pipeline.structured_pipeline import default_structured_pipeline
+        location_ref = f"az_blob://{user_id}/{document_id}"
+        pipeline = default_structured_pipeline()
+        await pipeline.run(tabular_introspector, location_ref, user_id)
+    """
+    raise NotImplementedError
+
+
+async def on_document_uploaded(document_id: str, user_id: str) -> None:
+    """Stub — for unstructured (PDF/DOCX/TXT). Owned by TAB / document pipeline.
+
+    Expected to call DocumentPipeline (extract → chunk → embed → PGVector).
+    """
     raise NotImplementedError
 
 
 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."""
     raise NotImplementedError