[KM-652] refactor(db): align analysis-family models to the dedorch schema

Repoint our SQLAlchemy models to the Go-owned dedorch tables (Go owns migrations;
we are consumer-only). Base tables already matched; only the analysis family differed.

- analysis_states -> `analysis`: table rename; id/report_id -> Postgres uuid.
- analysis_data_sources -> `data_sources`: richer shape (id, type, name,
reference_id, bound_by, bound_at, metadata, created_at). binding_store reads
`reference_id`; /analysis/create snapshots each source's type+name from the
catalog (fail-open). `metadata` mapped as `source_metadata` (reserved name).
- analysis_reports -> `reports`: flatten to title + content (rendered markdown) +
generated_at + version; drop user_id/jsonb data. ReportStore writes title+content
and rebuilds a minimal AnalysisReport on read (rendered_markdown = content;
structured fields empty — markdown-only per the 2026-06-23 checkpoint).

Deferred (pending lead): analysis_records has no dedorch table, so the report
generator / record persistence / readiness floor are untouched and POST /report
stays blocked until records get a home. The connstring cutover (+ SKIP_INIT_DB) is
a separate coordinated step — these renames only resolve against dedorch.

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

src/agents/binding_store.py

@@ -1,7 +1,7 @@
 """AnalysisDataSourceStore — read per-analysis data-source bindings (#10).
 
-The join table `analysis_data_sources(analysis_id, source_id)` records which catalog
-sources an analysis is scoped to. It's written atomically at `/analysis/create`; this
+The dedorch `data_sources` table records which catalog sources an analysis is scoped
+to (`reference_id` = the catalog source id). It's written at `/analysis/create`; this
 store is the read seam for the two consumers — `structured_flow` catalog scoping and
 the report's data-source appendix.
 
@@ -27,7 +27,7 @@ class AnalysisDataSourceStore:
     async def get(self, analysis_id: str) -> list[str]:
         async with AsyncSessionLocal() as session:
             result = await session.execute(
-                select(AnalysisDataSourceRow.source_id).where(
+                select(AnalysisDataSourceRow.reference_id).where(
                     AnalysisDataSourceRow.analysis_id == analysis_id
                 )
             )

src/agents/report/store.py

@@ -35,6 +35,27 @@ def _lock_key(analysis_id: str) -> int:
     return int.from_bytes(digest[:8], "big", signed=True)
 
 
+def _report_title(report: AnalysisReport) -> str:
+    """Title for the dedorch `reports.title` column — the goal, else a generic label."""
+    objective = (report.problem_statement.objective or "").strip()
+    return objective[:200] if objective else "Analysis Report"
+
+
+def _row_to_report(row) -> AnalysisReport:
+    """Rebuild a minimal AnalysisReport from the flat dedorch row.
+
+    dedorch stores markdown only, so structured fields (findings/caveats/…) come back
+    empty; `rendered_markdown` carries the content the FE renders/downloads.
+    """
+    return AnalysisReport(
+        report_id=row.id,
+        analysis_id=row.analysis_id,
+        version=row.version,
+        generated_at=row.generated_at,
+        rendered_markdown=row.content,
+    )
+
+
 class ReportStore:
     """Read/write versioned reports keyed by `analysis_id`."""
 
@@ -59,9 +80,10 @@ class ReportStore:
                     AnalysisReportRow(
                         id=report.report_id,
                         analysis_id=report.analysis_id,
-                        user_id=report.user_id,
+                        title=_report_title(report),
+                        content=report.rendered_markdown or "",
+                        generated_at=report.generated_at,
                         version=report.version,
-                        data=report.model_dump(mode="json"),
                     )
                 )
             # leaving session.begin() commits, which releases the advisory lock
@@ -76,17 +98,17 @@ class ReportStore:
     async def list_for_analysis(self, analysis_id: str) -> list[AnalysisReport]:
         async with AsyncSessionLocal() as session:
             result = await session.execute(
-                select(AnalysisReportRow.data)
+                select(AnalysisReportRow)
                 .where(AnalysisReportRow.analysis_id == analysis_id)
                 .order_by(AnalysisReportRow.version.asc())
             )
             rows = result.scalars().all()
-        return [AnalysisReport.model_validate(row) for row in rows]
+        return [_row_to_report(row) for row in rows]
 
     async def get(self, analysis_id: str, version: int) -> AnalysisReport | None:
         async with AsyncSessionLocal() as session:
             result = await session.execute(
-                select(AnalysisReportRow.data).where(
+                select(AnalysisReportRow).where(
                     AnalysisReportRow.analysis_id == analysis_id,
                     AnalysisReportRow.version == version,
                 )
@@ -94,4 +116,4 @@ class ReportStore:
             row = result.scalar_one_or_none()
         if row is None:
             return None
-        return AnalysisReport.model_validate(row)
+        return _row_to_report(row)

src/agents/state_store.py

@@ -35,7 +35,7 @@ def _row_to_state(row: AnalysisStateRow) -> AnalysisState:
 
 
 class AnalysisStateStore:
-    """Read/write `analysis_states` rows, keyed by the shared session id."""
+    """Read/write the dedorch `analysis` table, keyed by the shared session id."""
 
     async def get(self, analysis_id: str) -> AnalysisState | None:
         async with AsyncSessionLocal() as session:
@@ -113,7 +113,7 @@ class AnalysisStateStore:
             row = await session.get(AnalysisStateRow, analysis_id)
             if row is None:
                 logger.warning(
-                    "analysis_states row missing — update skipped",
+                    "analysis row missing — update skipped",
                     analysis_id=analysis_id,
                 )
                 return None

src/api/v1/analysis.py

@@ -40,13 +40,29 @@ def _serialize_state(row: AnalysisStateRow, data_source_ids: list[str]) -> dict:
 
 async def _bound_source_ids(db: AsyncSession, analysis_id: str) -> list[str]:
     result = await db.execute(
-        select(AnalysisDataSourceRow.source_id).where(
+        select(AnalysisDataSourceRow.reference_id).where(
             AnalysisDataSourceRow.analysis_id == analysis_id
         )
     )
     return list(result.scalars().all())
 
 
+async def _sources_by_id(user_id: str) -> dict:
+    """Catalog sources keyed by source_id, to resolve `type`/`name` on binding.
+
+    Never-throw: missing catalog / read error → empty map, and binding rows fall back
+    to type='unknown' / name=reference_id.
+    """
+    try:
+        from src.catalog.store import CatalogStore
+
+        catalog = await CatalogStore().get(user_id)
+    except Exception as e:  # noqa: BLE001 — binding must not fail on catalog read
+        logger.warning("analysis: catalog read failed for binding", user_id=user_id, error=str(e))
+        return {}
+    return {s.source_id: s for s in catalog.sources} if catalog else {}
+
+
 class CreateAnalysisRequest(BaseModel):
     user_id: str
     analysis_title: str = "New analysis"
@@ -63,10 +79,9 @@ async def create_analysis(
     """Create a new analysis session: one shared id for its state + chat room.
 
     Data-first gate (decision #2): an analysis requires >=1 bound data source.
-    The bound sources are persisted as `analysis_data_sources` rows (#10, Option A:
-    analysis-owned join table) in the same transaction as the state + room, so the
-    analysis is scoped to exactly the sources the user picked. `structured_flow` and
-    the report read this binding back.
+    The bound sources are persisted as dedorch `data_sources` rows (#10) in the same
+    transaction as the state + room, so the analysis is scoped to exactly the sources
+    the user picked. `structured_flow` and the report read this binding back.
     """
     if not request.data_source_ids:
         raise HTTPException(
@@ -86,11 +101,23 @@ async def create_analysis(
     )
     db.add(Room(id=analysis_id, user_id=request.user_id, title=request.analysis_title))
     db.add(state_row)
-    # dict.fromkeys dedupes while preserving order — composite PK (analysis_id,
-    # source_id) would otherwise reject a repeated source.
+    # dict.fromkeys dedupes while preserving order. Each binding row snapshots the
+    # source's type + name from the catalog (reference_id = catalog source id);
+    # bound_at/created_at default to now() in dedorch.
     bound_ids = list(dict.fromkeys(request.data_source_ids))
+    src_by_id = await _sources_by_id(request.user_id)
     for source_id in bound_ids:
-        db.add(AnalysisDataSourceRow(analysis_id=analysis_id, source_id=source_id))
+        src = src_by_id.get(source_id)
+        db.add(
+            AnalysisDataSourceRow(
+                id=str(uuid.uuid4()),
+                analysis_id=analysis_id,
+                type=src.source_type if src else "unknown",
+                name=src.name if src else source_id,
+                reference_id=source_id,
+                bound_by=request.user_id,
+            )
+        )
     await db.commit()
     await db.refresh(state_row)
 

src/db/postgres/models.py

@@ -10,9 +10,8 @@ from sqlalchemy import (
     Integer,
     String,
     Text,
-    UniqueConstraint,
 )
-from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
 
@@ -151,46 +150,42 @@ class AnalysisRecordRow(Base):
 
 
 class AnalysisReportRow(Base):
-    """One immutable row per generated report version (KM-644).
+    """One immutable row per generated report version — dedorch `reports` (Go-owned).
 
-    `data` holds the full Pydantic AnalysisReport
-    (src/agents/report/schemas.py:AnalysisReport) serialized via
-    `model_dump(mode="json")`; the read path rehydrates with
-    `AnalysisReport.model_validate(...)`. Versions accumulate per analysis session;
-    `(analysis_id, version)` is unique. Versioning is serialized by a per-analysis
-    advisory lock in `ReportStore` — the unique constraint is the backstop.
+    dedorch stores the rendered markdown `content` + `title` + `version` (no jsonb
+    snapshot — markdown-only per the 2026-06-23 checkpoint). The read path rebuilds a
+    minimal `AnalysisReport` (structured fields empty; `rendered_markdown` = content).
+    Versions accumulate per analysis; versioning is serialized by a per-analysis
+    advisory lock in `ReportStore`. Class name kept; table + shape changed for dedorch.
     """
-    __tablename__ = "analysis_reports"
+    __tablename__ = "reports"
 
-    id = Column(String, primary_key=True)  # AnalysisReport.report_id
-    analysis_id = Column(String, nullable=False, index=True)
-    user_id = Column(String, index=True)
+    id = Column(UUID(as_uuid=False), primary_key=True)  # AnalysisReport.report_id (uuid)
+    analysis_id = Column(UUID(as_uuid=False), nullable=False, index=True)
+    title = Column(String, nullable=False)
+    content = Column(Text, nullable=False)  # rendered markdown
+    generated_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
     version = Column(Integer, nullable=False)
-    data = Column(JSONB, nullable=False)
-    created_at = Column(DateTime(timezone=True), server_default=func.now())
-
-    __table_args__ = (
-        UniqueConstraint("analysis_id", "version", name="uq_analysis_version"),
-    )
 
 
 class AnalysisStateRow(Base):
-    """Per-analysis session state — shared id with `rooms` (analysis_id == room_id).
+    """Per-analysis session state — the dedorch `analysis` table (Go-owned migration).
 
-    One session = one analysis = one conversation: `id` is the same id as the chat
-    `rooms` row, so the existing `room_id` on the chat request doubles as the
-    `analysis_id`. The orchestrator gate + Help skill read this every turn;
+    One session = one analysis = one conversation; `id` is the shared session id
+    (canonical UUID). The orchestrator gate + Help skill read this every turn;
     `problem_validated` gates structured analysis; the Problem Statement skill flips
-    it; `report_id` is null until a report exists.
+    it; `report_id` is null until a report exists. `id`/`report_id` are Postgres
+    `uuid` in dedorch, so they bind as UUID (canonical-string in/out). Class name
+    kept as `AnalysisStateRow`; only the table + id types changed for dedorch.
     """
-    __tablename__ = "analysis_states"
+    __tablename__ = "analysis"
 
-    id = Column(String, primary_key=True)  # == rooms.id (shared session id)
+    id = Column(UUID(as_uuid=False), primary_key=True)  # shared session id (uuid)
     analysis_title = Column(String, nullable=False, default="New analysis")
     problem_statement = Column(Text, nullable=False, default="")
     problem_validated = Column(Boolean, nullable=False, default=False)
     owner_id = Column(String, nullable=False, index=True)
-    report_id = Column(String, nullable=True)
+    report_id = Column(UUID(as_uuid=False), nullable=True)
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(
         DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
@@ -198,16 +193,22 @@ class AnalysisStateRow(Base):
 
 
 class AnalysisDataSourceRow(Base):
-    """Per-analysis data-source binding (#10): which catalog sources an analysis is
-    scoped to.
+    """Per-analysis data-source binding (#10) — dedorch `data_sources` (Go-owned).
 
-    Written atomically with the state + room at `/analysis/create`; read by the
-    `structured_flow` catalog scoping (chat) and the report's data-source appendix.
-    Composite PK `(analysis_id, source_id)` — a source is bound at most once per
-    analysis. v1 binds whole sources (source-level granularity).
+    Which catalog sources an analysis is scoped to. `reference_id` is the catalog
+    `Source.source_id`; `type`/`name` snapshot the source kind + label. Written at
+    `/analysis/create`; read by `structured_flow` scoping + the report appendix.
+    `source_metadata` maps to the `metadata` column (`metadata` is reserved by the
+    declarative API). Class name kept; table + shape changed for dedorch.
     """
-    __tablename__ = "analysis_data_sources"
-
-    analysis_id = Column(String, primary_key=True, index=True)  # == analysis_states.id
-    source_id = Column(String, primary_key=True)  # == catalog Source.source_id
-    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    __tablename__ = "data_sources"
+
+    id = Column(UUID(as_uuid=False), primary_key=True)
+    analysis_id = Column(UUID(as_uuid=False), nullable=False, index=True)
+    type = Column(String, nullable=False)
+    name = Column(String, nullable=False)
+    reference_id = Column(String, nullable=False)  # == catalog Source.source_id
+    bound_by = Column(String, nullable=False)
+    bound_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    source_metadata = Column("metadata", JSONB, nullable=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())