/fix data_catalog model + CatalogStore to dedorch schema (data → catalog_payload)

Python's Catalog ORM still had the old Python-owned shape (user_id PK + `data`
jsonb), so every catalog read ran `SELECT data_catalog.data` and 500'd against
the Go-migrated dedorch table (post-cutover). Surfaced by the `check` skill:
"what data do I have" → check_data/check_knowledge → UndefinedColumnError.

- Reconcile `Catalog` ORM to dedorch: id PK, scope_type, user_id, analysis_id,
catalog_payload, + partial unique indexes (mirrors AnalysisStateRow).
- CatalogStore.get reads catalog_payload WHERE scope_type='user' (matches Go's
catalog.Service); upsert reconciled + marked legacy (Go owns catalog writes).
- REPO_STATUS §8/§12: drift resolved, data_catalog shape, cutover done.

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

REPO_STATUS.md

@@ -2,7 +2,7 @@
 
 **Audience:** teammates onboarding onto the Python repo (`Agentic-Service-Data-Eyond-Catalog`).
 **Scope:** what the code does **right now** (branch `pr/4`, ticket KM-652). Describes current state only — no roadmap or to-dos.
-**Snapshot date:** 2026-06-25. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
+**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
 the **Go source** (`Orchestrator-Agent-Service`), not its docs. The Go service has moved well past its
 own (uncommitted, stale) design docs: it now hosts the **dedorch SQL migrations** in-repo and a full
 **`/api/v1/analyses` + `/api/v1/skills`** REST surface. Go does **not** call Python yet — those skills
@@ -178,7 +178,7 @@ unless `SKIP_INIT_DB=true`.
 |---|---|---|---|
 | `users`, `rooms`, `chat_messages`, `message_sources` | base app | chat endpoint, Go | chat history |
 | `documents`, `databases` | uploads + DB creds (Fernet-encrypted) | Go ingestion | executor cred resolution |
-| `data_catalog` | per-user jsonb `Catalog` (Source → Table → Column) | Go ingestion / Python pipeline | CatalogReader, planner, tools |
+| `data_catalog` *(dedorch, Go-owned)* | `id` uuid, `scope_type` ('user'\|'analysis'), `user_id`, `analysis_id`, **`catalog_payload`** jsonb (the `Catalog`: Source → Table → Column), schema_version, generated_at, updated_at; partial-unique on `user_id WHERE scope_type='user'` | **Go `catalog.Service`** (all writes: DB/file ingestion) | CatalogReader → CatalogStore (**read-only**), planner, tools |
 | `langchain_pg_embedding` | PGVector document chunks | Go ingestion | DocumentRetriever |
 | `report_inputs` *(was `analysis_records`)* | jsonb `AnalysisRecord`, one per slow-path run; **Python-owned** | slow path | ReportGenerator, report readiness |
 | `analyses` *(dedorch, plural)* | uuid `id`, `user_id`, `analysis_title`, `objective`, `business_questions` jsonb, `status` (active\|inactive), `data_bind`(+`data_bind_version`), `report_id`, `report_collection` — **defined by Go migrations**; `problem_statement`/`problem_validated`/`owner_id` already **dropped** there (`0003`/`0004`) | Go `/api/v1/analyses`; Python state store | gate (no-op), Help, report |
@@ -186,12 +186,15 @@ unless `SKIP_INIT_DB=true`.
 | `data_sources` *(dedorch)* | per-analysis binding; `reference_id` = catalog source_id; `type ∈ document\|database` | Go `/analyses/{id}/data-bind` (+ Python `/analysis/create`) | structured-flow scoping, report appendix |
 | `analyses_messages` *(dedorch)* | the analysis chat room (`role ∈ user\|ai`); replaces deprecated `rooms`/`chat_messages` | Go `/analyses/{id}/messages` | Python chat path **not yet migrated here** (§12) |
 
-> ⚠️ **Python ORM ↔ dedorch drift (verified 2026-06-29).** Python's `AnalysisStateRow` + `state_store.py`
-> still model **`problem_statement` / `problem_validated`** and do **not** carry `objective` /
-> `business_questions`, but the Go migrations have already dropped the former and added the latter.
-> Pre-cutover this is harmless (Python runs `create_all` on its own copy); **post-`SKIP_INIT_DB`**, when
-> Python reads dedorch directly, ORM column selection on the dropped columns will break. Reconcile the
-> Python model before the connection-string cutover.
+> ✅ **Python ORM ↔ dedorch drift — reconciled 2026-07-01.** `AnalysisStateRow` (`analyses`) dropped
+> `problem_statement`/`problem_validated` and added `objective`/`business_questions` (Harry's #3);
+> `data_catalog` was the last stale model. Its `Catalog` ORM (old `user_id`-PK + `data` jsonb) is now
+> the dedorch shape (`id` PK, `scope_type`, **`catalog_payload`**), and `CatalogStore` reads
+> `catalog_payload WHERE scope_type='user'` (matching Go's `catalog.Service`). This closed a **live
+> bug**: the `check` skill / `CatalogReader` still selected the dropped `data_catalog.data` column, so
+> every catalog read 500'd after the cutover ("what data do I have" → *"Sorry, I couldn't look that up:
+> column data_catalog.data does not exist"*). Python's catalog **write** methods (`upsert`/
+> `remove_source`/`StructuredPipeline`) were reconciled but are now **legacy** — Go owns ingestion.
 
 **Catalog shape** (the jsonb in `data_catalog`):
 `Catalog → Source[ {source_id, source_type ∈ schema|tabular|unstructured, name, location_ref} → Table[ {table_id, name, row_count, foreign_keys[]} → Column[ {column_id, name, data_type, nullable, pii_flag, sample_values|null, stats} ] ] ]`. PII columns have `sample_values: null` so real values never enter prompts.
@@ -309,8 +312,8 @@ copies disagree with the current code on:
 
 ## 12. dedorch migration — current state
 
-The Python DB is moving from `dataeyond` → **dedorch** (Go owns dedorch migrations; Python is
-consumer-only). State **re-verified against the Go source 2026-06-29**:
+The Python DB has moved from `dataeyond` → **dedorch** (cutover 2026-07-01; Go owns dedorch migrations;
+Python is consumer-only). State **re-verified against the Go source 2026-06-29**:
 
 - **The dedorch migrations now live IN the Go repo** — embedded SQL at
   `internal/repository/postgres/migrations/0001_create_core_schema.sql … 0004_replace_chat_with_analysis_scope.sql`,
@@ -325,8 +328,11 @@ consumer-only). State **re-verified against the Go source 2026-06-29**:
   `rooms`/`chat_messages`/`interview_*` tables to `zdeprecated_*`.
 - **`report_inputs`** (the slow-path structured output, formerly `analysis_records`) stays
   **Python-owned**; its finalized schema goes to Harry so the dedorch migration creates it post-cutover.
-- The connection-string cutover (paired with `SKIP_INIT_DB`) **has not happened yet**; Python still
-  runs `create_all` on its own models until then.
+- **Connection-string cutover DONE (2026-07-01).** Python's `postgres_connstring` now points at
+  **dedorch** and reads the Go-migrated tables directly. Every ORM model Python reads (`analyses`,
+  `data_sources`, `analyses_messages`, `data_catalog`) has been reconciled to its dedorch shape.
+  `create_all` still runs (no `SKIP_INIT_DB` yet) but is now harmless — `checkfirst` skips the
+  Go-owned tables. `report_inputs` stays Python-created until Harry's migration lands.
 
 **⚠️ Integration gap (verified — the big one).** Go's `/api/v1/analyses` and `/api/v1/skills`
 (`help` / `report`) are **placeholders that return dummy data** — the `SendMessage` / `GenerateReport`

src/catalog/store.py

@@ -1,7 +1,9 @@
-"""CatalogStore — persists per-user catalogs as Postgres jsonb rows.
+"""CatalogStore — reads the per-user catalog from the dedorch `data_catalog` table.
 
-Storage shape: one row per user in a `catalogs` table with columns
-(user_id PK, data jsonb, schema_version, generated_at, updated_at).
+Storage shape (Go-owned): one row per scope in `data_catalog`
+(id, scope_type, user_id, analysis_id, catalog_payload jsonb, schema_version,
+generated_at, updated_at). Python reads the user-scoped row (scope_type='user');
+Go's `catalog.Service` owns all writes, so `upsert`/`remove_source` are legacy.
 """
 
 from sqlalchemy import case, delete, func, select
@@ -27,7 +29,10 @@ class CatalogStore:
     async def get(self, user_id: str) -> Catalog | None:
         async with AsyncSessionLocal() as session:
             result = await session.execute(
-                select(CatalogRow.data).where(CatalogRow.user_id == user_id)
+                select(CatalogRow.catalog_payload).where(
+                    CatalogRow.user_id == user_id,
+                    CatalogRow.scope_type == "user",
+                )
             )
             row = result.scalar_one_or_none()
         if row is None:
@@ -35,22 +40,29 @@ class CatalogStore:
         return Catalog.model_validate(row)
 
     async def upsert(self, catalog: Catalog) -> None:
+        # Legacy: Go's catalog.Service owns catalog writes now. Kept working (and
+        # reconciled to the dedorch shape) but no longer on any live Python path.
         payload = catalog.model_dump(mode="json")
         async with AsyncSessionLocal() as session:
             stmt = insert(CatalogRow).values(
+                scope_type="user",
                 user_id=catalog.user_id,
-                data=payload,
+                catalog_payload=payload,
                 schema_version=catalog.schema_version,
                 generated_at=catalog.generated_at,
                 updated_at=func.now(),
             )
             stmt = stmt.on_conflict_do_update(
                 index_elements=[CatalogRow.user_id],
+                index_where=CatalogRow.scope_type == "user",
                 set_={
-                    "data": stmt.excluded.data,
+                    "catalog_payload": stmt.excluded.catalog_payload,
                     "schema_version": stmt.excluded.schema_version,
                     "updated_at": case(
-                        (stmt.excluded.data != CatalogRow.data, func.now()),
+                        (
+                            stmt.excluded.catalog_payload != CatalogRow.catalog_payload,
+                            func.now(),
+                        ),
                         else_=CatalogRow.updated_at,
                     ),
                 },

src/db/postgres/models.py

@@ -6,9 +6,11 @@ from sqlalchemy import (
     Column,
     DateTime,
     ForeignKey,
+    Index,
     Integer,
     String,
     Text,
+    text,
 )
 from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import relationship
@@ -108,23 +110,44 @@ class DatabaseClient(Base):
 
 
 class Catalog(Base):
-    """Per-user data catalog stored as a single jsonb row.
+    """Data catalog — dedorch **`data_catalog`** (Go-owned; reconciled 2026-07-01).
 
-    `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.
+    Mirrors Go migration `0001`/`0002`. One jsonb `catalog_payload` per scope:
+    `scope_type='user'` rows are keyed by `user_id` (partial unique index),
+    `scope_type='analysis'` rows by `analysis_id`. Python is **consumer-only** —
+    Go's `catalog.Service` owns all writes (DB/file ingestion); `CatalogStore`
+    reads the user-scoped catalog and its write methods are legacy.
 
-    Dedicated table — kept separate from `langchain_pg_embedding` so unstructured
-    embeddings and structured-catalog metadata never share storage.
+    `catalog_payload` holds the full Pydantic Catalog (src/catalog/models.py:Catalog)
+    serialized via `model_dump(mode="json")`; the read path rehydrates with
+    `Catalog.model_validate(...)`. Go writes the same shape (json tags match).
     """
     __tablename__ = "data_catalog"
 
-    user_id = Column(String, primary_key=True)
-    data = Column(JSONB, nullable=False)
+    id = Column(UUID(as_uuid=False), primary_key=True, default=lambda: str(uuid4()))
+    scope_type = Column(String, nullable=False, default="user")  # 'user' | 'analysis'
+    user_id = Column(String, nullable=False, index=True)
+    analysis_id = Column(UUID(as_uuid=False), nullable=True)
+    catalog_payload = Column(JSONB, nullable=False)
     schema_version = Column(String, nullable=False, default="1.0")
-    generated_at = Column(DateTime(timezone=True), server_default=func.now())
+    generated_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
 
+    __table_args__ = (
+        Index(
+            "idx_data_catalog_user_scope",
+            "user_id",
+            unique=True,
+            postgresql_where=text("scope_type = 'user'"),
+        ),
+        Index(
+            "idx_data_catalog_analysis_scope",
+            "analysis_id",
+            unique=True,
+            postgresql_where=text("scope_type = 'analysis'"),
+        ),
+    )
+
 
 class ReportInputRow(Base):
     """One row per completed slow-path analysis (the report's source of truth).