/fix v2 chat stream api update to analyses_messages

src/agents/chat_handler.py

@@ -215,13 +215,13 @@ class ChatHandler:
         from .gate import stub_analysis_state
 
         if not analysis_id:
-            return stub_analysis_state(problem_validated=False)
+            return stub_analysis_state()
         try:
             state = await self._get_state_store().get(analysis_id)
         except Exception as e:
             logger.warning("help state read failed — not-validated", error=str(e))
             state = None
-        return state if state is not None else stub_analysis_state(problem_validated=False)
+        return state if state is not None else stub_analysis_state()
 
     # ------------------------------------------------------------------
     # Public entry
@@ -320,7 +320,7 @@ class ChatHandler:
         #         intent,
         #         analysis_state
         #         if analysis_state is not None
-        #         else stub_analysis_state(problem_validated=False),
+        #         else stub_analysis_state(),
         #     )
 
         # The `intent` event is consumed by the endpoint (it gates response caching

src/agents/gate.py

@@ -20,7 +20,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from src.agents.orchestration import Intent
 from src.middlewares.logging import get_logger
@@ -29,17 +29,18 @@ logger = get_logger("gate")
 
 
 class AnalysisState(BaseModel):
-    """Per-analysis state the gate + Help skill read every turn (locked contract).
+    """Per-analysis state the Help skill + report layer read every turn.
 
-    `problem_validated` is the gate driver; `report_id` is null until a report
-    exists. Field names mirror the `analysis_states` table so the DB read swaps in
-    without touching readers.
+    Field names mirror the dedorch `analyses` table so the DB read swaps in without
+    touching readers. The goal is the user-entered `objective` + `business_questions`
+    (set at onboarding by Go); the old `problem_statement`/`problem_validated` gate fields
+    were dropped (dedorch #3 / KM-652). `report_id` is null until a report exists.
     """
 
     id: str
     analysis_title: str
-    problem_statement: str
-    problem_validated: bool = False
+    objective: str = ""
+    business_questions: list[str] = Field(default_factory=list)
     user_id: str
     report_id: str | None = None
     created_at: datetime
@@ -66,18 +67,18 @@ def gate(intent: Intent, state: AnalysisState) -> Intent:
     return intent
 
 
-def stub_analysis_state(*, problem_validated: bool = False) -> AnalysisState:
-    """Hardcoded Analysis State for building/testing before the DB lands (#9).
+def stub_analysis_state() -> AnalysisState:
+    """Hardcoded Analysis State for never-throw fallbacks / tests.
 
-    Shared fixture so the gate, the Help skill, and tests all exercise the same
-    shape. `problem_validated=True` simulates a passed interview.
+    Shared fixture so the gate seam, the Help skill, and tests all exercise the same
+    shape when a real row is missing or a read fails.
     """
     now = datetime.now(UTC)
     return AnalysisState(
         id="stub-analysis",
         analysis_title="Stub analysis",
-        problem_statement="Stub problem statement" if problem_validated else "",
-        problem_validated=problem_validated,
+        objective="",
+        business_questions=[],
         user_id="stub-user",
         report_id=None,
         created_at=now,
@@ -104,8 +105,8 @@ async def get_analysis_state(analysis_id: str) -> AnalysisState:
             analysis_id=analysis_id,
             error=str(exc),
         )
-        return stub_analysis_state(problem_validated=False)
+        return stub_analysis_state()
     if state is None:
         logger.debug("analysis_state missing — default not-validated", analysis_id=analysis_id)
-        return stub_analysis_state(problem_validated=False)
+        return stub_analysis_state()
     return state

src/agents/state_store.py

@@ -25,8 +25,8 @@ def _row_to_state(row: AnalysisStateRow) -> AnalysisState:
     return AnalysisState(
         id=row.id,
         analysis_title=row.analysis_title,
-        problem_statement=row.problem_statement,
-        problem_validated=row.problem_validated,
+        objective=row.objective or "",
+        business_questions=list(row.business_questions or []),
         user_id=row.user_id,
         report_id=row.report_id,
         created_at=row.created_at,
@@ -58,14 +58,17 @@ class AnalysisStateStore:
         no source bindings — binding scoping fail-opens to the whole catalog.
         """
         async with AsyncSessionLocal() as session:
+            # Bare get-or-create row. objective/business_questions are always supplied
+            # (empty) so the INSERT satisfies dedorch whether or not they are NOT NULL;
+            # the real goal is set at onboarding by Go and preserved by ON CONFLICT DO NOTHING.
             stmt = (
                 insert(AnalysisStateRow)
                 .values(
                     id=analysis_id,
                     user_id=user_id,
                     analysis_title=analysis_title,
-                    problem_statement="",
-                    problem_validated=False,
+                    objective="",
+                    business_questions=[],
                 )
                 .on_conflict_do_nothing(index_elements=[AnalysisStateRow.id])
             )
@@ -80,7 +83,8 @@ class AnalysisStateStore:
         analysis_id: str,
         user_id: str,
         analysis_title: str = "New analysis",
-        problem_statement: str = "",
+        objective: str = "",
+        business_questions: list[str] | None = None,
     ) -> AnalysisState:
         """Create the state row for a new analysis (id shared with its chat room)."""
         async with AsyncSessionLocal() as session:
@@ -88,8 +92,8 @@ class AnalysisStateStore:
                 id=analysis_id,
                 user_id=user_id,
                 analysis_title=analysis_title,
-                problem_statement=problem_statement,
-                problem_validated=False,
+                objective=objective,
+                business_questions=business_questions or [],
             )
             session.add(row)
             await session.commit()
@@ -100,14 +104,13 @@ class AnalysisStateStore:
         self,
         analysis_id: str,
         *,
-        problem_statement: str | None = None,
-        problem_validated: bool | None = None,
         report_id: str | None = None,
     ) -> AnalysisState | None:
         """Patch the given fields (only non-None args are written). Returns the row.
 
-        Used by the Problem Statement skill (`problem_validated`) and the report
-        flow (`report_id`). Returns None if the analysis doesn't exist.
+        Used by the report flow (`report_id` write-back). The analysis goal
+        (`objective`/`business_questions`) is set at onboarding by Go, not patched here.
+        Returns None if the analysis doesn't exist.
         """
         async with AsyncSessionLocal() as session:
             row = await session.get(AnalysisStateRow, analysis_id)
@@ -117,10 +120,6 @@ class AnalysisStateStore:
                     analysis_id=analysis_id,
                 )
                 return None
-            if problem_statement is not None:
-                row.problem_statement = problem_statement
-            if problem_validated is not None:
-                row.problem_validated = problem_validated
             if report_id is not None:
                 row.report_id = report_id
             await session.commit()

src/api/v1/chat.py

@@ -14,7 +14,7 @@ from sse_starlette.sse import EventSourceResponse
 from src.agents.chat_handler import ChatHandler
 from src.config.settings import settings
 from src.db.postgres.connection import get_db
-from src.db.postgres.models import ChatMessage, MessageSource
+from src.db.postgres.models import AnalysesMessageRow
 from src.db.redis.connection import get_redis
 from src.middlewares.logging import get_logger, log_execution
 
@@ -100,12 +100,16 @@ async def cache_response(redis, cache_key: str, response: str, sources: list):
     )
 
 
-async def load_history(db: AsyncSession, room_id: str, limit: int = 10) -> list:
-    """Load recent chat messages for a room as LangChain message objects (oldest-first)."""
+async def load_history(db: AsyncSession, analysis_id: str, limit: int = 10) -> list:
+    """Load recent conversation messages for an analysis as LangChain messages (oldest-first).
+
+    Reads the dedorch `analyses_messages` table (`role ∈ user|ai`), which replaced the
+    deprecated `rooms`/`chat_messages`.
+    """
     result = await db.execute(
-        select(ChatMessage)
-        .where(ChatMessage.room_id == room_id)
-        .order_by(ChatMessage.created_at.asc())
+        select(AnalysesMessageRow)
+        .where(AnalysesMessageRow.analysis_id == analysis_id)
+        .order_by(AnalysesMessageRow.created_at.asc())
         .limit(limit)
     )
     rows = result.scalars().all()
@@ -117,24 +121,25 @@ async def load_history(db: AsyncSession, room_id: str, limit: int = 10) -> list:
 
 async def save_messages(
     db: AsyncSession,
-    room_id: str,
+    analysis_id: str,
+    user_id: str,
     user_content: str,
     assistant_content: str,
-    sources: Optional[List[Dict[str, Any]]] = None,
 ):
-    """Persist user and assistant messages, and attach sources to the assistant message."""
-    db.add(ChatMessage(id=str(uuid.uuid4()), room_id=room_id, role="user", content=user_content))
-    assistant_id = str(uuid.uuid4())
-    db.add(ChatMessage(id=assistant_id, room_id=room_id, role="assistant", content=assistant_content))
-    for src in (sources or []):
-        page = src.get("page_label")
-        db.add(MessageSource(
-            id=str(uuid.uuid4()),
-            message_id=assistant_id,
-            document_id=src.get("document_id"),
-            filename=src.get("filename"),
-            page_label=str(page) if page is not None else None,
-        ))
+    """Persist the user turn + AI answer to dedorch `analyses_messages` (`role` user|ai).
+
+    Python writes this Go-owned table as a consumer (it does not create it). RAG source
+    citations are streamed to the client but not persisted — the old `message_sources`
+    table is deprecated along with `chat_messages`.
+    """
+    db.add(AnalysesMessageRow(
+        id=str(uuid.uuid4()), analysis_id=analysis_id, user_id=user_id,
+        role="user", content=user_content,
+    ))
+    db.add(AnalysesMessageRow(
+        id=str(uuid.uuid4()), analysis_id=analysis_id, user_id=user_id,
+        role="ai", content=assistant_content,
+    ))
     await db.commit()
 
 
@@ -187,7 +192,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
         logger.info("Returning cached response")
         cached_text = cached["response"]
         cached_sources = cached["sources"]
-        await save_messages(db, request.room_id, request.message, cached_text, sources=cached_sources)
+        await save_messages(db, request.room_id, request.user_id, request.message, cached_text)
 
         async def stream_cached():
             yield {"event": "sources", "data": json.dumps(cached_sources)}
@@ -202,7 +207,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
         direct = _fast_intent(request.message)
         if direct:
             await cache_response(redis, cache_key, direct, sources=[])
-            await save_messages(db, request.room_id, request.message, direct, sources=[])
+            await save_messages(db, request.room_id, request.user_id, request.message, direct)
 
             async def stream_direct():
                 yield {"event": "sources", "data": json.dumps([])}
@@ -244,7 +249,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
                         await cache_response(redis, cache_key, full_response, sources=sources)
                     logger.info("saving messages", sources_count=len(sources), sources=sources)
                     try:
-                        await save_messages(db, request.room_id, request.message, full_response, sources=sources)
+                        await save_messages(db, request.room_id, request.user_id, request.message, full_response)
                     except Exception as e:
                         logger.error("save_messages failed", room_id=request.room_id, error=str(e))
                     yield event

src/api/v1/report.py

@@ -128,7 +128,7 @@ async def generate_report(
 
     state = await _load_state(analysis_id)
     floor_missing, _ = await report_floor(
-        analysis_id, state or stub_analysis_state(problem_validated=False)
+        analysis_id, state or stub_analysis_state()
     )
     if floor_missing:
         raise HTTPException(

src/api/v2/chat.py

@@ -88,7 +88,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
         logger.info("Returning cached response")
         cached_text = cached["response"]
         cached_sources = cached["sources"]
-        await save_messages(db, analysis_id, request.message, cached_text, sources=cached_sources)
+        await save_messages(db, analysis_id, request.user_id, request.message, cached_text)
 
         async def stream_cached():
             yield {"event": "sources", "data": json.dumps(cached_sources)}
@@ -103,7 +103,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
         direct = _fast_intent(request.message)
         if direct:
             await cache_response(redis, cache_key, direct, sources=[])
-            await save_messages(db, analysis_id, request.message, direct, sources=[])
+            await save_messages(db, analysis_id, request.user_id, request.message, direct)
 
             async def stream_direct():
                 yield {"event": "sources", "data": json.dumps([])}
@@ -144,7 +144,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
                         await cache_response(redis, cache_key, full_response, sources=sources)
                     try:
                         await save_messages(
-                            db, analysis_id, request.message, full_response, sources=sources
+                            db, analysis_id, request.user_id, request.message, full_response
                         )
                     except Exception as e:
                         logger.error("save_messages failed", analysis_id=analysis_id, error=str(e))

src/db/postgres/models.py

@@ -3,7 +3,6 @@
 from uuid import uuid4
 
 from sqlalchemy import (
-    Boolean,
     Column,
     DateTime,
     ForeignKey,
@@ -186,27 +185,23 @@ class AnalysisStateRow(Base):
     One session = one analysis = one conversation; `id` is the shared session id
     (canonical UUID). Verified against the dedorch DB 2026-06-25.
 
-    dedorch `analyses` ACTUAL columns: `id` (uuid), `analysis_title`, `user_id` (text),
-    `report_id` (uuid), `created_at`, `updated_at`, `problem_statement`,
-    `problem_validated`, `status` (text 'active'|'inactive' — soft-delete),
-    `data_bind` (jsonb), `data_bind_version` (int), `report_collection` (jsonb).
+    dedorch `analyses` columns (reconciled 2026-07-01 — Harry's #3 landed): `id` (uuid),
+    `analysis_title`, `objective` (text), `business_questions` (jsonb), `user_id` (text),
+    `report_id` (uuid), `status` (text 'active'|'inactive' — soft-delete), `data_bind` (jsonb),
+    `data_bind_version` (int), `report_collection` (jsonb), `created_at`, `updated_at`.
 
-    Reconciled to that shape (#4, 2026-06-26): `user_id` (was `owner_id`) + `status`/`data_bind`/
-    `data_bind_version`/`report_collection` added. dedorch still carries `problem_statement`/
-    `problem_validated` and does NOT yet have `objective`/`business_questions` — Harry's #3 drops
-    the former + adds the latter; the report layer reads the goal getattr-tolerantly so that swap
-    stays non-breaking. The new FE/Go columns are stored to match dedorch but NOT surfaced in the
-    `AnalysisState` pydantic contract (no Python reader needs them yet).
-
-    `analysis` (singular) is the deprecated DUPLICATE table Harry will drop — never use it.
-    Class name kept as `AnalysisStateRow`.
+    `problem_statement`/`problem_validated` were DROPPED in dedorch (#3) and removed here;
+    `objective` + `business_questions` (the user-entered goal, set at onboarding by Go) replace
+    them. The FE/Go columns (`status`/`data_bind*`/`report_collection`) are carried to match
+    dedorch but are NOT surfaced in the `AnalysisState` pydantic contract. `analysis` (singular)
+    is the deprecated DUPLICATE table Harry will drop — never use it. Class name kept.
     """
     __tablename__ = "analyses"
 
     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)
+    objective = Column(Text, nullable=False, default="")
+    business_questions = Column(JSONB, nullable=False, default=list)
     user_id = Column(String, nullable=False, index=True)  # was owner_id (dedorch uses user_id)
     report_id = Column(UUID(as_uuid=False), nullable=True)
     # dedorch `analyses` columns (FE/Go concerns; carried so create_all matches dedorch).
@@ -240,3 +235,23 @@ class AnalysisDataSourceRow(Base):
     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())
+
+
+class AnalysesMessageRow(Base):
+    """One conversation message — dedorch `analyses_messages` (Go-owned table).
+
+    The analysis chat room (user question + AI answer), replacing the deprecated
+    `rooms`/`chat_messages`. Python is a **consumer/writer** here: it INSERTs and reads
+    rows but does NOT own the table (Go's migration creates it). Shape mirrors the Go
+    contract (`API_CONTRACT_BE_GOLANG.md` §Analysis Messages): `role ∈ user|ai`. RAG source
+    citations are NOT persisted here — the old `message_sources` table is deprecated along
+    with `chat_messages`.
+    """
+    __tablename__ = "analyses_messages"
+
+    id = Column(UUID(as_uuid=False), primary_key=True)
+    analysis_id = Column(UUID(as_uuid=False), nullable=False, index=True)
+    user_id = Column(String, nullable=False, index=True)
+    role = Column(String, nullable=False)  # user | ai
+    content = Column(Text, nullable=False)
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())