[NOTICKET] feat(slow-path): persist AnalysisRecord via PostgresAnalysisStore

Groundwork for generate_report - the slow path now persists a real, citable
record that the report (next) renders from.

Contract gaps (schemas.py):
- stage: CrispStage added to TaskResult + TaskSummary, populated at all 3
TaskResult build sites (task_runner) + copied in assembler._build_record,
so the report can group its method appendix by CRISP-DM phase.
- AnalysisRecord gained identity: record_id (auto uuid), analysis_id/user_id
(optional; stamped at persist time).

Persistence:
- PostgresAnalysisStore (store.py): save() never-throw idempotent upsert +
list_for_analysis() oldest-first (the report render order). NullAnalysisStore
kept for tests. AnalysisStore Protocol gained list_for_analysis.
- analysis_records jsonb table (models.py), indexed by analysis_id + user_id;
registered in init_db, created by create_all on startup (no migration).
- chat_handler default store -> PostgresAnalysisStore; user_id stamped at save.

Open: analysis_id is NULL until the Analysis State reaches the slow path
(session-id handoff from Harry needed to group records per analysis).

Docs: checkpoint plan + PROGRESS updated.

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

CHECKPOINT_PLAN_2026-06-17.md

@@ -22,6 +22,8 @@ Dated snapshot of what landed this session. Live task status (incl. what's left)
 - ✅ **Planner tool list verified** against the trimmed registry — no references to old tool names or deferred analytics anywhere in `src/` (Rifqi).
 - 📌 **Decision:** `tests/` stays gitignored — team decided not to push tests to origin (closes PROGRESS.md R3 as won't-do).
 - 📌 **Ownership:** Rifqi owns `generate_report` development + the `analysis_records` table / real `AnalysisStore` (contract still co-designed with Sofhia).
+- ✅ **R5 cache fix** (Rifqi, `b701e95`): chat cache scoped by `user_id`, TTL 24h→1h.
+- ✅ **AnalysisRecord persistence landed** (Rifqi): `stage` now flows to the record (CRISP-DM grouping for the report) + identity fields (`record_id`/`analysis_id`/`user_id`); `PostgresAnalysisStore` + `analysis_records` table replace `NullAnalysisStore`, wired into `ChatHandler`. Unblocks the `generate_report` renderer and the DoD "record persisted" step. Open: `analysis_id` handoff from Harry's Analysis State.
 
 ---
 
@@ -68,8 +70,12 @@ Dated snapshot of what landed this session. Live task status (incl. what's left)
 
 ### Rifqi
 - [x] **Redis Cloud free tier** (~30–50 MB): create instance, set TTL = 1 h, share env vars in the group. (done 12 Jun)
-- [ ] **Own `generate_report` development** (contract co-designed with Sofhia, see §3) — includes the `analysis_records` table + real `AnalysisStore` (replaces `NullAnalysisStore`); required for the DoD "record persisted" step.
+- [x] **R5 cache fix**: chat cache key scoped by `user_id`, TTL 24h→1h (urgent on shared Redis). (12 Jun, commit `b701e95`)
+- [x] **AnalysisRecord contract gaps closed**: `stage` (CRISP-DM) now flows Task→TaskResult→TaskSummary so the report can group the method appendix; `AnalysisRecord` gained `record_id`/`analysis_id`/`user_id` identity fields. (12 Jun)
+- [x] **`analysis_records` table + real `AnalysisStore`**: `PostgresAnalysisStore` (save + `list_for_analysis`, never-throw) replaces `NullAnalysisStore`; wired into `ChatHandler`, `user_id` stamped at save. Satisfies the DoD "record persisted" step. (12 Jun)
+- [ ] **Own `generate_report` development** (contract co-designed with Sofhia, see §3) — reads `analysis_records` + Problem Statement; versioned, bumps `report_status`. *(persistence half done above; renderer + contract doc next)*
 - [x] Verify planner tool list matches the trimmed registry (4 analytics + 4 data/knowledge) and few-shots don't reference removed tools. (verified 12 Jun — no stale tool names in `src/`)
+- ⚠️ **Blocked-on-Harry**: `analysis_id` is `NULL` on persisted records until the Analysis State reaches the slow path — need the session-ID handoff so `generate_report` can group records per analysis.
 
 ### Shared (Sofhia + Rifqi)
 - [ ] `generate_report` design + skeleton: input = AnalysisRecords for the session + Problem Statement from Analysis State; output = versioned artifact; bumps `report_status`. Agree on the contract even if rendering is stubbed for Wednesday. (Development: Rifqi.)

PROGRESS.md

@@ -2,11 +2,35 @@
 
 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-06-12 (Redis Cloud live; R3 closed as won't-do per team decision; Rifqi owns `analysis_records`/`AnalysisStore` + `generate_report`)
+**Last updated**: 2026-06-12 (Redis Cloud live; R3 closed as won't-do; R5 cache fix; AnalysisRecord persistence landed — `PostgresAnalysisStore` + `analysis_records` table)
 **Current open PR**: `pr/3` — active.
 
 ---
 
+## What just shipped (2026-06-12 — AnalysisRecord persistence, Rifqi)
+
+Groundwork for `generate_report`. The slow path now persists a real, citable
+record; the report (next) renders from it.
+
+- **Contract gaps closed** (`agents/slow_path/schemas.py`): `stage: CrispStage`
+  added to `TaskResult` + `TaskSummary` and populated at all 3 `TaskResult` build
+  sites in `task_runner.py` + copied in `assembler._build_record` — so the report
+  can group its method appendix by CRISP-DM phase. `AnalysisRecord` gained identity:
+  `record_id` (auto uuid), `analysis_id`/`user_id` (optional; stamped at persist).
+- **Real store** (`agents/slow_path/store.py`): `PostgresAnalysisStore` —
+  `save()` (never-throw, idempotent upsert) + `list_for_analysis()` (oldest-first,
+  the report's render order). `NullAnalysisStore` kept (tests / disabled persistence).
+  `AnalysisStore` Protocol gained `list_for_analysis`.
+- **Table** (`db/postgres/models.py`): `analysis_records` jsonb table (one row per
+  run, indexed by `analysis_id` + `user_id`); registered in `init_db.py`, created by
+  `create_all` on startup (no migration — `data_catalog` precedent).
+- **Wired** (`agents/chat_handler.py`): default store flipped to `PostgresAnalysisStore`;
+  `user_id` stamped onto the record at the save site (in scope there).
+- **Open**: `analysis_id` is `NULL` until Harry's Analysis State reaches the slow
+  path (session-ID handoff needed to group records per analysis).
+
+---
+
 ## Principal architecture review (2026-06-10) — findings + fix tracker
 
 A full external review (read the context docs + the slow path, tool layer, query

src/agents/chat_handler.py

@@ -282,9 +282,9 @@ class ChatHandler:
 
     def _get_analysis_store(self) -> AnalysisStore:
         if self._analysis_store is None:
-            from .slow_path.store import NullAnalysisStore
+            from .slow_path.store import PostgresAnalysisStore
 
-            self._analysis_store = NullAnalysisStore()
+            self._analysis_store = PostgresAnalysisStore()
         return self._analysis_store
 
     async def _run_slow_path(
@@ -368,7 +368,10 @@ class ChatHandler:
         yield {"event": "sources", "data": json.dumps([])}  # TODO: derive from record
         yield {"event": "chunk", "data": result.chat_answer}
         try:
-            await self._get_analysis_store().save(result.analysis_record)
+            # Stamp the owner now (in scope here); analysis_id is wired once the
+            # Analysis State reaches the slow path. The store is also never-throw.
+            record = result.analysis_record.model_copy(update={"user_id": user_id})
+            await self._get_analysis_store().save(record)
         except Exception as e:  # persistence must never break the user's answer
             logger.error("analysis_record persist failed", user_id=user_id, error=str(e))
         tracer.end()  # output omitted (chat_answer may contain PII on Cloud)

src/agents/slow_path/assembler.py

@@ -120,6 +120,7 @@ def _build_record(narrative: AssemblerNarrative, run_state: RunState) -> Analysi
     tasks_run = [
         TaskSummary(
             task_id=task_id,
+            stage=result.stage,
             objective=result.objective,
             status=result.status,
             tools_used=[o.tool for o in result.outputs],

src/agents/slow_path/schemas.py

@@ -21,10 +21,12 @@ from __future__ import annotations
 
 from datetime import datetime
 from typing import Literal
+from uuid import uuid4
 
 from pydantic import BaseModel, Field
 
 from ..planner.contracts import ToolOutput
+from ..planner.schemas import CrispStage
 
 TaskStatus = Literal["success", "partial", "failure"]
 
@@ -36,6 +38,7 @@ TaskStatus = Literal["success", "partial", "failure"]
 
 class TaskResult(BaseModel):
     task_id: str
+    stage: CrispStage  # copied from the plan Task; carries CRISP-DM grouping to the report
     status: TaskStatus
     objective: str
     outputs: list[ToolOutput] = Field(default_factory=list)  # one per tool_call
@@ -57,12 +60,21 @@ class RunState(BaseModel):
 
 class TaskSummary(BaseModel):
     task_id: str
+    stage: CrispStage  # lets the report group the method appendix by CRISP-DM phase
     objective: str
     status: TaskStatus
     tools_used: list[str] = Field(default_factory=list)
 
 
 class AnalysisRecord(BaseModel):
+    # Identity. `record_id` is the unit the report cites and snapshots
+    # (`record_ids`); `analysis_id`/`user_id` scope the record to one analysis
+    # session + owner and are stamped by the composition root / AnalysisStore at
+    # persist time (they depend on the Analysis State that lives outside the slow
+    # path), so they default to None when the Assembler first builds the record.
+    record_id: str = Field(default_factory=lambda: uuid4().hex)
+    analysis_id: str | None = None
+    user_id: str | None = None
     # Narrative fields — authored by the Assembler LLM.
     goal_restated: str
     findings: list[str] = Field(default_factory=list)

src/agents/slow_path/store.py

@@ -2,21 +2,28 @@
 
 The Assembler produces an `AnalysisRecord` (the faithful, structured record of a
 run — §8.3, INV-4). Persisting it is a separate concern from streaming the answer,
-so it sits behind this one-method seam.
+so it sits behind this seam. `generate_report` later reads records back by
+`analysis_id` (oldest-first) and renders from them — never from chat history.
 
-`NullAnalysisStore` is the default: it logs that a record was produced but stores
-nothing, because the backing table does not exist yet. The plan is to store records
-in the **same catalog DB** (Neon `dataeyond`, `settings.postgres_connstring`).
+- `NullAnalysisStore` logs and stores nothing (kept for tests / when persistence
+  is intentionally disabled).
+- `PostgresAnalysisStore` writes one `analysis_records` row per run in the catalog
+  DB (Neon `dataeyond`, `settings.postgres_connstring`).
 
-TODO(persistence): add a Postgres-backed `AnalysisStore` writing an
-`analysis_records` table in the catalog DB, keyed on
-(business_context_id, plan_id, created_at), then inject it into ChatHandler.
+`save` must never raise on the caller's path — a persistence failure must not break
+the user's answer (§8.3). `list_for_analysis` is a read for the report generator and
+is allowed to surface errors to its caller.
 """
 
 from __future__ import annotations
 
 from typing import Protocol, runtime_checkable
 
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert
+
+from src.db.postgres.connection import AsyncSessionLocal
+from src.db.postgres.models import AnalysisRecordRow
 from src.middlewares.logging import get_logger
 
 from .schemas import AnalysisRecord
@@ -26,19 +33,78 @@ logger = get_logger("analysis_store")
 
 @runtime_checkable
 class AnalysisStore(Protocol):
-    """Persist a completed analysis. Implementations must never raise on the
-    caller's path — a persistence failure must not break the user's answer."""
+    """Persist + read completed analyses.
+
+    `save` must never raise on the caller's path. `list_for_analysis` returns the
+    records for one analysis session, oldest-first (the order the report renders in).
+    """
 
     async def save(self, record: AnalysisRecord) -> None: ...
 
+    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisRecord]: ...
+
 
 class NullAnalysisStore:
-    """Default no-op store: logs the record, persists nothing (no table yet)."""
+    """No-op store: logs the record, persists nothing. Reads return empty."""
 
     async def save(self, record: AnalysisRecord) -> None:
         logger.info(
-            "analysis_record produced (not persisted — no store configured)",
+            "analysis_record produced (not persisted — NullAnalysisStore)",
+            record_id=record.record_id,
             plan_id=record.plan_id,
-            business_context_id=record.business_context_id,
             n_tasks=len(record.tasks_run),
         )
+
+    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisRecord]:
+        return []
+
+
+class PostgresAnalysisStore:
+    """Writes/reads `analysis_records` jsonb rows in the catalog DB.
+
+    Mirrors `CatalogStore`: each call opens its own `AsyncSession`. One row per
+    record (vs. one-per-user for the catalog) since records accumulate per analysis.
+    """
+
+    async def save(self, record: AnalysisRecord) -> None:
+        try:
+            payload = record.model_dump(mode="json")
+            async with AsyncSessionLocal() as session:
+                stmt = insert(AnalysisRecordRow).values(
+                    id=record.record_id,
+                    analysis_id=record.analysis_id,
+                    user_id=record.user_id,
+                    plan_id=record.plan_id,
+                    data=payload,
+                    created_at=record.created_at,
+                )
+                # Re-running the same plan id-collides only if record_id repeats;
+                # treat that as idempotent (overwrite) rather than erroring the user.
+                stmt = stmt.on_conflict_do_update(
+                    index_elements=[AnalysisRecordRow.id],
+                    set_={"data": stmt.excluded.data},
+                )
+                await session.execute(stmt)
+                await session.commit()
+            logger.info(
+                "analysis_record persisted",
+                record_id=record.record_id,
+                analysis_id=record.analysis_id,
+                user_id=record.user_id,
+            )
+        except Exception as exc:  # never break the user's answer (§8.3)
+            logger.error(
+                "analysis_record persist failed",
+                record_id=record.record_id,
+                error=str(exc),
+            )
+
+    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisRecord]:
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(AnalysisRecordRow.data)
+                .where(AnalysisRecordRow.analysis_id == analysis_id)
+                .order_by(AnalysisRecordRow.created_at.asc())
+            )
+            rows = result.scalars().all()
+        return [AnalysisRecord.model_validate(row) for row in rows]

src/agents/slow_path/task_runner.py

@@ -53,6 +53,7 @@ class TaskRunner:
                 for tid in list(remaining):
                     results[tid] = TaskResult(
                         task_id=tid,
+                        stage=tasks_by_id[tid].stage,
                         status="failure",
                         objective=tasks_by_id[tid].objective,
                         error="unresolved dependency; task could not run",
@@ -68,6 +69,7 @@ class TaskRunner:
                 if failed:
                     results[tid] = TaskResult(
                         task_id=tid,
+                        stage=task.stage,
                         status="failure",
                         objective=task.objective,
                         error=f"skipped: upstream {failed} did not succeed",
@@ -110,6 +112,7 @@ class TaskRunner:
             error = errs[0] if errs else "all tool calls failed"
         return TaskResult(
             task_id=task.id,
+            stage=task.stage,
             status=status,
             objective=task.objective,
             outputs=outputs,

src/db/postgres/init_db.py

@@ -3,6 +3,7 @@
 from sqlalchemy import text
 from src.db.postgres.connection import engine, Base
 from src.db.postgres.models import (
+    AnalysisRecordRow,
     Catalog,
     ChatMessage,
     DatabaseClient,

src/db/postgres/models.py

@@ -115,4 +115,25 @@ class Catalog(Base):
     schema_version = Column(String, nullable=False, default="1.0")
     generated_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
- 
+
+
+class AnalysisRecordRow(Base):
+    """One row per completed slow-path analysis (the report's source of truth).
+
+    `data` holds the full Pydantic AnalysisRecord
+    (src/agents/slow_path/schemas.py:AnalysisRecord) serialized via
+    `model_dump(mode="json")`; the read path rehydrates with
+    `AnalysisRecord.model_validate(...)`. Many records accumulate per analysis
+    session — `generate_report` reads them by `analysis_id`, oldest-first.
+
+    `analysis_id` is nullable until the Analysis State (owned upstream) is wired
+    into the slow path; records still persist (and carry `user_id`) before then.
+    """
+    __tablename__ = "analysis_records"
+
+    id = Column(String, primary_key=True)  # AnalysisRecord.record_id
+    analysis_id = Column(String, index=True)  # FK to the analysis session (nullable for now)
+    user_id = Column(String, nullable=False, index=True)
+    plan_id = Column(String, nullable=False)
+    data = Column(JSONB, nullable=False)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())