[KM-715] charts: render_chart tool + message_charts store + GET /api/v1/charts + planner viz slice

W2 of SPINE_V2_PLAN: deterministic Plotly-JSON spec builder (dataeyond.chart.v1,
no plotly dep), ToolOutput.kind "chart", never-throw ChartStore mirroring
traceability (rows written before done), FE fetches GET /charts on done (empty
list = valid 200; done event unchanged). Planner: recipe table + explicit-ask
rule, Examples J (viz tail) + K (viz-infeasible), validator Check 10
(render_chart.data must be table-kind). Traceability records charts compactly
(point_count, not raw arrays). Live e2e vs dedorch: ALL PASS.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

API_CONTRACT_BE_PYTHON.md

@@ -34,6 +34,7 @@ The frontend uses this service during the analysis conversation flow:
 | `GET` | `/api/v1/tools/report/{analysis_id}/readiness` | Report-readiness signal for the Generate-Report button (added 2026-07-09). |
 | `GET` | `/api/v1/tools/report/{analysis_id}/{version}` | Retrieve one report version. |
 | `GET` | `/api/v1/traceability` | Retrieve provenance for one assistant answer. |
+| `GET` | `/api/v1/charts` | Retrieve chart(s) produced by `render_chart` for one assistant answer (added 2026-07-13). |
 
 ## Common Concepts
 
@@ -61,6 +62,8 @@ Common event types:
 
 The stream carries answer text only. Planning, tool call details, and full provenance are fetched from `GET /api/v1/traceability` after the stream is done.
 
+**Charts (added 2026-07-13, SPINE_V2_PLAN §4.5):** the `done` event is unchanged by charts — no additive `chart_count`/`chart_ids` field yet (open question owned by Harry). Until that's resolved, the frontend fetches `GET /api/v1/charts` unconditionally on every `done`, the same fetch-on-`done` pattern already used for traceability. SSE order and every existing field stay exactly as documented above; `sources` stays `[]`.
+
 ## Chat
 
 ### `POST /api/v2/chat/stream`
@@ -643,3 +646,91 @@ Frontend rendering guidance:
 - Default state can be collapsed.
 - Show planning, tool calls, and sources as separate sections.
 - Treat `planning: null`, `tool_calls: []`, and `sources: []` as valid states.
+
+## Charts
+
+> Added 2026-07-13 (SPINE_V2_PLAN §4.4/§4.5, S2 visualization). A chart is a `render_chart` tool output, planner-selected only when the user explicitly asks to plot/visualize. Delivery mirrors traceability: a Python-owned store plus a dedicated `GET` endpoint; the streamed answer (SSE) stays text-only — charts are never embedded in `chunk`.
+
+### `GET /api/v1/charts`
+
+Returns every chart produced by `render_chart` during one assistant answer.
+
+The frontend should call this after the chat stream emits `done`, using the `analysis_id` + `message_id` from the `done` event — the same fetch-on-`done` pattern as traceability; the row(s) are written **before** `done`, so there is no polling race. Unlike `GET /api/v1/traceability`, **an empty list is a valid `200`** — most turns don't ask for a chart, and once `done` has fired there is no "not ready yet" state to distinguish from "no charts this turn", so this endpoint never 404s on a chartless turn.
+
+Query params:
+
+| Query | Required | Description |
+| --- | --- | --- |
+| `analysis_id` | Yes | Analysis identifier. |
+| `message_id` | Yes | Assistant answer identifier returned by the stream. |
+
+Example:
+
+```text
+GET /api/v1/charts?analysis_id=an_42&message_id=msg_88f1
+```
+
+The `dataeyond.chart.v1` envelope (the shape of `charts[].spec`, verbatim from `render_chart`):
+
+```json
+{
+  "schema": "dataeyond.chart.v1",
+  "chart_type": "bar",
+  "title": "Revenue by region",
+  "plotly": {
+    "data":   [{ "type": "bar", "x": ["A", "B"], "y": [1, 2], "name": "revenue" }],
+    "layout": { "title": {"text": "Revenue by region"}, "xaxis": {"title": {"text": "region"}},
+                "yaxis": {"title": {"text": "revenue"}} }
+  }
+}
+```
+
+Frontend renders it with `Plotly.newPlot(el, spec.plotly.data, spec.plotly.layout)`. v1 chart types: `bar`, `line`, `pie`, `scatter`.
+
+Response `200` (one chart):
+
+```json
+{
+  "count": 1,
+  "charts": [
+    {
+      "chart_id": "3fbd8e2e-8e21-4d4b-9b21-9e6b6a0a6a6e",
+      "chart_type": "bar",
+      "title": "Revenue by region",
+      "spec": {
+        "schema": "dataeyond.chart.v1",
+        "chart_type": "bar",
+        "title": "Revenue by region",
+        "plotly": {
+          "data":   [{ "type": "bar", "x": ["A", "B"], "y": [1, 2], "name": "revenue" }],
+          "layout": { "title": {"text": "Revenue by region"}, "xaxis": {"title": {"text": "region"}},
+                      "yaxis": {"title": {"text": "revenue"}} }
+        }
+      },
+      "created_at": "2026-07-13T03:21:09.114Z"
+    }
+  ]
+}
+```
+
+Response `200` (chartless turn — the common case; not an error):
+
+```json
+{
+  "count": 0,
+  "charts": []
+}
+```
+
+Field rules:
+
+- `spec` is the full `dataeyond.chart.v1` envelope, unmodified — it is the source of truth, not a projection; render straight from it.
+- `chart_type` / `title` are copied out of `spec` for convenience (list rendering without parsing `spec`); `title` may be `null`.
+- A turn can produce more than one chart (multiple `render_chart` calls in the same plan); `charts` is ordered by creation time.
+- The payload carries no `user_id` (unlike traceability) — charts are keyed by `analysis_id` + `message_id` only.
+
+Frontend rendering guidance:
+
+- Fetch unconditionally on every `done` — see the SSE note above (no `chart_count` hint yet); an empty `charts[]` means render nothing extra.
+- Render each chart under the assistant message it belongs to, via `Plotly.newPlot`.
+- "Chart iteration" v1 = a follow-up chat turn (e.g. "make it a line chart") — the planner re-emits `render_chart` with patched args. There is no separate edit endpoint.

main.py

@@ -22,6 +22,7 @@ from src.api.v1.report import router as report_router
 from src.api.v1.tools import router as tools_router
 from src.api.v1.help import router as help_router  # pr/5 Phase 2: dedicated /tools/help
 from src.api.v1.traceability import router as traceability_router  # KM-691
+from src.api.v1.charts import router as charts_router  # W2: GET /api/v1/charts (SPINE_V2_PLAN §4.5)
 from src.api.v2.chat import router as chat_v2_router  # pr/5 Phase 2: v2 chat pilot (analysis_id)
 from src.db.postgres.init_db import init_db
 from src.config.settings import settings
@@ -68,6 +69,7 @@ app.include_router(report_router)
 app.include_router(tools_router)
 app.include_router(help_router)
 app.include_router(traceability_router)  # KM-691: GET /api/v1/traceability
+app.include_router(charts_router)  # W2: GET /api/v1/charts (SPINE_V2_PLAN §4.5)
 app.include_router(chat_v2_router)     # pr/5 Phase 2: POST /api/v2/chat/stream (analysis_id)
 
 

src/agents/chat_handler.py

@@ -54,6 +54,7 @@ from .refusals import blocked_message, out_of_scope_message
 
 if TYPE_CHECKING:
     from ..catalog.reader import CatalogReader
+    from ..charts.store import ChartStore
     from ..retrieval.router import RetrievalRouter
     from ..traceability.store import TraceabilityStore
     from .gate import AnalysisState
@@ -105,6 +106,7 @@ class ChatHandler:
         ) = None,
         analysis_store: ReportInputStore | None = None,
         traceability_store: TraceabilityStore | None = None,
+        chart_store: ChartStore | None = None,
         check_invoker_factory: Callable[[str], Any] | None = None,
         ps_agent: ProblemStatementAgent | None = None,
         help_agent: HelpAgent | None = None,
@@ -128,6 +130,9 @@ class ChatHandler:
         # Traceability (KM-691): user-facing per-turn provenance store. Injectable for
         # tests; lazily built (Postgres) in production. Distinct from Langfuse tracing.
         self._traceability_store = traceability_store
+        # Charts (S2, SPINE_V2_PLAN §4.4): `render_chart` output store. Injectable for
+        # tests; lazily built (Postgres) in production.
+        self._chart_store = chart_store
         # `check` skill: builds the data-access invoker (check_data/check_knowledge)
         # per request with the authenticated user_id. Injectable for tests.
         self._check_invoker_factory = check_invoker_factory
@@ -644,6 +649,13 @@ class ChatHandler:
             self._traceability_store = PostgresTraceabilityStore()
         return self._traceability_store
 
+    def _get_chart_store(self) -> ChartStore:
+        if self._chart_store is None:
+            from ..charts import PostgresChartStore
+
+            self._chart_store = PostgresChartStore()
+        return self._chart_store
+
     async def _flush_trace(
         self, pad: TraceabilityScratchpad, analysis_id: str | None, user_id: str
     ) -> None:
@@ -763,6 +775,22 @@ class ChatHandler:
                 pad.set_planning_from_record(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))
+        # SPINE_V2_PLAN §4.4: chart rows are written before `done`; the FE fetches
+        # GET /api/v1/charts unconditionally on every `done` (no polling race).
+        try:
+            if pad is not None and pad.message_id:
+                for task_result in result.analysis_record.results_snapshot.values():
+                    for output in task_result.outputs:
+                        if output.kind == "chart" and isinstance(output.value, dict):
+                            await self._get_chart_store().save(
+                                message_id=pad.message_id,
+                                analysis_id=analysis_id or "",
+                                user_id=user_id,
+                                record_id=result.analysis_record.record_id,
+                                envelope=output.value,
+                            )
+        except Exception as e:  # chart persist must never break the user's answer
+            logger.error("chart persist failed", user_id=user_id, error=str(e))
         tracer.end()  # output omitted (chat_answer may contain PII on Cloud)
         if pad is not None:
             await self._flush_trace(pad, analysis_id, user_id)

src/agents/planner/examples.py

@@ -730,6 +730,119 @@ _EXAMPLE_I = TaskList(
 )
 
 
+# --------------------------------------------------------------------------- #
+# Example J — visualization tail (SPINE_V2_PLAN §4.3, recipe "viz tail").
+# "Show me a bar chart of total revenue per product category."
+# Shows: render_chart is planned ONLY on an explicit ask ("bar chart"), and is
+# ALWAYS a tail step — its `data` consumes a TABLE-producing upstream (here the
+# grouped top-N-style retrieve from Example G, without the limit), never stats/
+# series/metadata. `x`/`y` reference that table's column aliases; the chart
+# carries the already-aggregated rows (one bar per category), never raw order
+# rows. The `assumptions` line carries the feasibility check on purpose: a chart
+# ask never licenses aliasing a stand-in column (see Example K).
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_J = TaskList(
+    plan_id="example_j",
+    goal_restated="Chart total revenue per product category as a bar chart.",
+    assumptions=[
+        "The chart dimension exists in the catalog (c_category) — a chart ask is "
+        "planned only when the asked-for dimension and measure have real catalog "
+        "columns; otherwise it is infeasible, chart or no chart."
+    ],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes category and revenue.",
+            tool_calls=[ToolCall(tool="check_data", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria="Produced the orders table schema; category and revenue present.",
+            depends_on=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Aggregate total revenue per product category.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_data",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_category", "alias": "category"},
+                                {
+                                    "kind": "agg",
+                                    "fn": "sum",
+                                    "column_id": "c_revenue",
+                                    "alias": "total_revenue",
+                                },
+                            ],
+                            "group_by": ["c_category"],
+                        }
+                    },
+                )
+            ],
+            expected_output="revenue_per_category",
+            success_criteria="Produced one total-revenue row per category.",
+            depends_on=["t1"],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t3",
+            stage="evaluation",
+            objective="Render the per-category revenue table as a bar chart.",
+            tool_calls=[
+                ToolCall(
+                    tool="render_chart",
+                    args={
+                        # `data` references the TABLE task (t2) — a chart is always a
+                        # tail on an aggregated table, never on raw rows or stats.
+                        "data": "${t2}",
+                        "chart_type": "bar",
+                        "x": "category",
+                        "y": "total_revenue",
+                        "title": "Total revenue by product category",
+                    },
+                )
+            ],
+            expected_output="revenue_bar_chart",
+            success_criteria="Produced a bar-chart spec with one bar per category.",
+            depends_on=["t2"],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+
+# --------------------------------------------------------------------------- #
+# Example K — a chart ask that is INFEASIBLE (SPINE_V2_PLAN §4.3 + planner.md
+# "Charts ... never relaxes feasibility"). "Plot the customer churn rate by
+# month as a line chart" against the sales catalog: churn does not exist, and
+# the explicit chart request does NOT license mapping some other column into
+# the asked-for measure — the verdict is the same infeasible_reason Example H
+# would give, chart or no chart.
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_K = TaskList(
+    plan_id="example_k",
+    goal_restated="Chart the monthly customer churn rate as a line chart.",
+    assumptions=[],
+    open_questions=[],
+    tasks=[],
+    infeasible_reason=(
+        "The connected source has no churn or subscription-status data — the "
+        "orders table only carries order-level category, revenue, quantity, and "
+        "dates — so there is nothing to chart. Nearest chartable analyses: "
+        "monthly revenue trend, or order counts per category."
+    ),
+)
+
+
 EXAMPLES: list[tuple[str, TaskList]] = [
     ("Which product categories drove last quarter's revenue?", _EXAMPLE_A),
     ("How has monthly revenue trended by region this year, and what's unusual?", _EXAMPLE_B),
@@ -744,6 +857,8 @@ EXAMPLES: list[tuple[str, TaskList]] = [
         "order quantity?",
         _EXAMPLE_I,
     ),
+    ("Show me a bar chart of total revenue per product category.", _EXAMPLE_J),
+    ("Plot the customer churn rate by month as a line chart.", _EXAMPLE_K),
 ]
 
 

src/agents/planner/validator.py

@@ -281,6 +281,24 @@ class PlannerValidator:
                     "analyzable data rows. Feed analyze_* from a data-producing tool "
                     "(e.g. retrieve_data)."
                 )
+            # Check 10 — render_chart path shape (SPINE_V2_PLAN §4.3): the chart's
+            # `data` must come from a task whose tool yields a TABLE. A stats- or
+            # series-kind upstream passes the category denylist above but cannot
+            # be materialized into a DataFrame at execution, so the chart could
+            # only fail late; reject pre-run so the retry re-prompts toward a
+            # grouped retrieve_data or a table-producing analyze_*.
+            if (
+                call.tool == "render_chart"
+                and ref_spec is not None
+                and ref_spec.output_kind != "table"
+            ):
+                raise PlannerValidationError(
+                    f"task {task_id}: render_chart takes its {arg_name!r} from task "
+                    f"{match.group(1)} ({ref_tool!r}), which yields "
+                    f"{ref_spec.output_kind!r} output — a chart is drawn from a TABLE. "
+                    "Feed it a table-producing task (a grouped retrieve_data, "
+                    "analyze_aggregate, analyze_merge, ...)."
+                )
 
     @staticmethod
     def _validate_dag(tasks_by_id: dict, id_set: set[str]) -> None:

src/api/v1/charts.py

@@ -0,0 +1,46 @@
+"""Charts endpoint — user-facing chart retrieval for one turn (S2 visualization,
+SPINE_V2_PLAN §4.4/§4.5).
+
+`GET /api/v1/charts?analysis_id=&message_id=` returns every chart a `render_chart`
+tool call produced during one assistant turn, as `dataeyond.chart.v1` envelopes
+(SPINE_V2_PLAN §4.2). Rows are written by the chat pipeline right before the `done`
+SSE event; the FE fires this GET on `done`.
+
+Unlike `GET /api/v1/traceability`, an **empty list is a valid 200** here (most turns
+don't ask for a chart) rather than a 404 — once `done` has fired there is no "not
+ready yet" state to distinguish from "no charts this turn".
+
+No auth — Go fronts Python.
+"""
+
+from fastapi import APIRouter, Query
+from pydantic import BaseModel
+
+from src.charts import ChartRecord, PostgresChartStore
+from src.middlewares.logging import get_logger, log_execution
+
+logger = get_logger("charts_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Charts"])
+
+# Warm, process-shared store (mirrors the chat endpoints' module-level instances).
+_store = PostgresChartStore()
+
+
+class ChartsResponse(BaseModel):
+    count: int
+    charts: list[ChartRecord]
+
+
+@router.get("/charts", response_model=ChartsResponse)
+@log_execution(logger)
+async def get_charts(
+    analysis_id: str = Query(..., description="Analysis/session id"),
+    message_id: str = Query(
+        ..., description="Assistant turn id, taken from the `done` SSE event"
+    ),
+) -> ChartsResponse:
+    """Fetch every chart for one turn. Empty list (not 404) when the turn produced
+    none — the FE can call this unconditionally on every `done`."""
+    charts = await _store.list_for_message(analysis_id, message_id)
+    return ChartsResponse(count=len(charts), charts=charts)

src/charts/__init__.py

@@ -0,0 +1,21 @@
+"""Charts (S2 visualization, SPINE_V2_PLAN §4.4): `render_chart` output persistence.
+
+Chart envelopes produced by the `render_chart` tool (`dataeyond.chart.v1`, SPINE_V2_PLAN
+§4.2) are persisted through a `ChartStore` right before the `done` SSE event, and served
+by `GET /api/v1/charts`. SSE stays text-only — charts are never embedded in the streamed
+answer.
+"""
+
+from .store import (
+    ChartRecord,
+    ChartStore,
+    NullChartStore,
+    PostgresChartStore,
+)
+
+__all__ = [
+    "ChartRecord",
+    "ChartStore",
+    "NullChartStore",
+    "PostgresChartStore",
+]

src/charts/store.py

@@ -0,0 +1,154 @@
+"""ChartStore — the seam the chat pipeline persists `render_chart` outputs through
+(S2 visualization, SPINE_V2_PLAN §4.4).
+
+`ChatHandler._run_slow_path` scans a completed `AnalysisRecord.results_snapshot` for
+`ToolOutput(kind="chart")` entries and flushes one `MessageChartRow` per chart through
+this seam, right before the `done` SSE event (mirrors the traceability flush). Unlike
+traceability's one row per turn, a turn with multiple `render_chart` calls writes
+multiple rows sharing one `message_id`. `GET /api/v1/charts` reads them back by
+(analysis_id, message_id).
+
+- `NullChartStore` logs the envelope and stores nothing (tests / disabled persistence).
+- `PostgresChartStore` writes one `message_charts` row per chart (dedorch,
+  `AsyncSessionLocal`), mirroring `PostgresTraceabilityStore`.
+
+`save` must NEVER raise on the caller's path — a chart-persist failure must not break
+the user's answer. `list_for_message` is the endpoint read; an empty list is a valid
+result (chartless turn), not an error.
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Protocol, runtime_checkable
+
+from pydantic import BaseModel
+from sqlalchemy import select
+
+from src.db.postgres.connection import AsyncSessionLocal
+from src.db.postgres.models import MessageChartRow
+from src.middlewares.logging import get_logger
+
+logger = get_logger("charts_store")
+
+
+class ChartRecord(BaseModel):
+    """One persisted chart, as served by `GET /api/v1/charts`."""
+
+    chart_id: str
+    chart_type: str
+    title: str | None = None
+    spec: dict
+    created_at: datetime
+
+
+@runtime_checkable
+class ChartStore(Protocol):
+    """Persist + read `render_chart` outputs for one assistant `message_id`.
+
+    `save` must never raise on the caller's path. `list_for_message` returns every
+    chart for one turn (possibly empty — a turn need not have asked for a chart).
+    """
+
+    async def save(
+        self,
+        *,
+        message_id: str,
+        analysis_id: str,
+        user_id: str,
+        record_id: str | None,
+        envelope: dict,
+    ) -> None: ...
+
+    async def list_for_message(
+        self, analysis_id: str, message_id: str
+    ) -> list[ChartRecord]: ...
+
+
+class NullChartStore:
+    """No-op store: logs the envelope, persists nothing. Reads return `[]`."""
+
+    async def save(
+        self,
+        *,
+        message_id: str,
+        analysis_id: str,
+        user_id: str,
+        record_id: str | None,
+        envelope: dict,
+    ) -> None:
+        logger.info(
+            "chart produced (not persisted — NullChartStore)",
+            message_id=message_id,
+            chart_type=envelope.get("chart_type", "unknown"),
+        )
+
+    async def list_for_message(
+        self, analysis_id: str, message_id: str
+    ) -> list[ChartRecord]:
+        return []
+
+
+class PostgresChartStore:
+    """Writes/reads `message_charts` rows. One insert per chart (not an upsert)."""
+
+    async def save(
+        self,
+        *,
+        message_id: str,
+        analysis_id: str,
+        user_id: str,
+        record_id: str | None,
+        envelope: dict,
+    ) -> None:
+        try:
+            async with AsyncSessionLocal() as session:
+                row = MessageChartRow(
+                    id=str(uuid.uuid4()),
+                    message_id=message_id,
+                    analysis_id=analysis_id,
+                    user_id=user_id,
+                    record_id=record_id,
+                    chart_type=envelope.get("chart_type", "unknown"),
+                    title=envelope.get("title"),
+                    spec=envelope,
+                )
+                session.add(row)
+                await session.commit()
+            logger.info(
+                "chart persisted",
+                message_id=message_id,
+                analysis_id=analysis_id,
+                chart_type=envelope.get("chart_type", "unknown"),
+            )
+        except Exception as exc:  # never break the user's answer
+            logger.error(
+                "chart persist failed",
+                message_id=message_id,
+                error=str(exc),
+            )
+
+    async def list_for_message(
+        self, analysis_id: str, message_id: str
+    ) -> list[ChartRecord]:
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(MessageChartRow)
+                .where(
+                    MessageChartRow.analysis_id == analysis_id,
+                    MessageChartRow.message_id == message_id,
+                )
+                .order_by(MessageChartRow.created_at)
+            )
+            rows = result.scalars().all()
+        return [
+            ChartRecord(
+                chart_id=row.id,
+                chart_type=row.chart_type,
+                title=row.title,
+                spec=row.spec,
+                created_at=row.created_at,
+            )
+            for row in rows
+        ]

src/config/prompts/planner.md

@@ -20,7 +20,9 @@ only a `TaskList` object that conforms to the provided schema.
    section. Copy the stable ids verbatim — downstream validation does a literal
    id lookup, so a paraphrased name fails.
 5. **No modeling in v1.** There are no modeling tools. Do not emit `modeling`
-   tasks. The product is descriptive/diagnostic only — no predictions, no charts.
+   tasks. The product is descriptive/diagnostic only — no predictions. A chart
+   exists ONLY as a `render_chart` tail on an explicit user ask (see "Charts"
+   under "How to plan") — never speculatively.
 6. **Never re-purpose a column as a different business measure.** A column means
    what the catalog says it means — do not alias one concept as another to force
    an answer (e.g. selecting an availability percentage AS "revenue", or a
@@ -35,6 +37,21 @@ only a `TaskList` object that conforms to the provided schema.
    is noise, not helpfulness. A multi-part task list is correct ONLY when the
    question itself has multiple parts (e.g. "trend by region AND what's unusual").
 
+# Recipes — the named workflows
+
+Classify the question into ONE recipe, then instantiate it. Most questions are a
+recipe verbatim; a genuinely multi-part question composes recipes.
+
+| Recipe | The question asks for… | Chain |
+|---|---|---|
+| R1 descriptive | a summary/distribution of columns | `retrieve_data` → `analyze_descriptive` |
+| R2 aggregate / top-N | totals or averages per group, "top N by …" | ONE grouped `retrieve_data` IR (± `analyze_aggregate`) |
+| R3 trend | movement over time | `retrieve_data` → `analyze_trend` |
+| R4 correlation | the relationship between numeric columns | `retrieve_data` → `analyze_correlation` |
+| R5 two-metric merge | "which X has both A and B" | `retrieve_data` ×2 → `analyze_merge` → … |
+| R6 infeasible | a measure/entity no catalog column holds | `tasks: []` + `infeasible_reason` |
+| viz tail | …any of the above, WITH an explicit ask to chart it | R1–R5 + a `render_chart` tail |
+
 # How to plan
 
 - **Smallest plan that answers the question.** Do not exceed `Constraints.max_tasks`.
@@ -82,6 +99,20 @@ only a `TaskList` object that conforms to the provided schema.
   to a task's LAST output, so two retrievals inside one task lose the first
   table. The merged table (one row per entity, both measures) answers the
   question, or feeds a further `analyze_*` step.
+- **Charts (`render_chart`) only on an explicit ask.** Add a `render_chart` tail
+  ONLY when the user explicitly asks to see a chart — "plot", "chart", "graph",
+  "visualize", "diagram" (ID: "grafik", "diagram", "visualisasikan", "buatkan
+  grafik/diagram") — NEVER speculatively on a plain data question. It is always
+  the LAST step: its `data` takes a `"${t<id>}"` from a task that yields a
+  **table** (a grouped `retrieve_data` or a table-producing `analyze_*` — never
+  stats/series/metadata), and `x`/`y` are that table's column aliases. Chart the
+  already-aggregated table (one row per category/period), not raw rows. Pick
+  `chart_type` by the question: `bar` (magnitude per category), `line` (over
+  time), `pie` (share of a small whole), `scatter` (two numeric columns).
+  A chart ask NEVER relaxes feasibility (rule 6): if the asked-for dimension or
+  measure has no catalog column, the question is **infeasible** — never chart a
+  stand-in column aliased under the asked-for name (e.g. never select a status
+  column AS "region" because the user asked for a chart by region).
 - **Mixing structured + unstructured.** If qualitative context helps, add a
   `retrieve_knowledge` task against an unstructured source listed in the catalog.
 - **CRISP-DM stages.** Tag each task with the stage it serves:

src/db/postgres/models.py

@@ -284,3 +284,36 @@ class MessageTraceabilityRow(Base):
     intent = Column(String, nullable=False)
     data = Column(JSONB, nullable=False)  # full TraceabilityPayload (source of truth)
     created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+
+class MessageChartRow(Base):
+    """One row per chart produced by `render_chart` — S2 visualization (SPINE_V2_PLAN §4.4).
+
+    `spec` holds the full `dataeyond.chart.v1` envelope (SPINE_V2_PLAN §4.2) exactly as
+    returned by the tool — `{schema, chart_type, title, plotly: {data, layout}}` — it is
+    the source of truth the FE renders with
+    `Plotly.newPlot(el, spec.plotly.data, spec.plotly.layout)`. Unlike
+    `MessageTraceabilityRow` (one row per turn), **multiple rows can share one
+    `message_id`** — one row per chart a turn produced. Written before the `done` SSE
+    event and served by `GET /api/v1/charts` (SPINE_V2_PLAN §4.4/§4.5).
+
+    OWNERSHIP / HANDOFF (SPINE_V2_PLAN §4.4, 2026-07-13): **Python-owned for now**, the
+    same pattern as `report_inputs` / `message_traceability`. Post-cutover `init_db` no
+    longer runs `create_all`, so the table is created by a one-time manual DDL run
+    against dedorch (Rifqi, 2026-07-13); the finalized schema goes to Harry so the
+    dedorch migration re-creates it later (`analysis_id` gains the FK to `analyses(id)`
+    there).
+    """
+    __tablename__ = "message_charts"
+
+    # Client-minted default — the DDL's `gen_random_uuid()` default never fires from
+    # Python (we always pass `id` explicitly, but the default is here for parity).
+    id = Column(UUID(as_uuid=False), primary_key=True, default=lambda: str(uuid4()))
+    message_id = Column(String, nullable=False, index=True)  # turn id; multiple rows per turn
+    analysis_id = Column(UUID(as_uuid=False), nullable=False, index=True)  # analysis session id
+    user_id = Column(String, nullable=False)
+    record_id = Column(String, nullable=True)
+    chart_type = Column(String, nullable=False)
+    title = Column(String, nullable=True)
+    spec = Column(JSONB, nullable=False)  # full dataeyond.chart.v1 envelope (source of truth)
+    created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())

src/tools/analytics/visualization.py

@@ -0,0 +1,239 @@
+"""render_chart — declarative chart-spec builder (SPINE_V2_PLAN §4.1, S2).
+
+First tool of the `render_*` family: turns an already-materialized DataFrame into
+a Plotly-conformant JSON spec wrapped in the `dataeyond.chart.v1` envelope
+(SPINE_V2_PLAN §4.2). The FE renders it with plotly.js
+(`Plotly.newPlot(el, spec.plotly.data, spec.plotly.layout)`).
+
+Deliberately NOT a code generator and NOT a plotly import (locked decision,
+DEV_PLAN deferred row #26): the spec is a hand-built dict, so no generated code
+ever executes and no new dependency lands. Style (colors, axis chrome) is the
+fixed module preset below — never a planner/LLM decision (CoDA's "Design
+Explorer" phase collapsed to a constant).
+
+Pattern A, same as the `analyze_*` family: `data` is resolved to a DataFrame by
+the invoker from an upstream table-kind output; this function never fetches.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+from decimal import Decimal
+
+import numpy as np
+import pandas as pd
+
+from src.tools.analytics.descriptive import ColumnNotFoundError
+
+# v1 chart types (SPINE_V2_PLAN §4.1). `pie` maps x -> labels, y -> values.
+CHART_TYPES = ("bar", "line", "pie", "scatter")
+
+
+class UnsupportedChartTypeError(ValueError):
+    """The requested chart_type is not in CHART_TYPES (error_code UNSUPPORTED_CHART_TYPE)."""
+
+
+# --------------------------------------------------------------------------- #
+# House style preset — fixed by design, not a planner argument.
+#
+# Colorway = the dataviz reference categorical palette (8 slots, light mode).
+# The slot ORDER is the colorblind-safety mechanism (validated: worst adjacent
+# CVD deltaE 24.2 vs the >=12 target) — series take slots in this fixed order,
+# never cycled or re-picked. Backgrounds are transparent so the chart inherits
+# the FE surface; inks/gridlines are the palette's chrome roles.
+# --------------------------------------------------------------------------- #
+COLORWAY = [
+    "#2a78d6",  # blue
+    "#1baf7a",  # aqua
+    "#eda100",  # yellow
+    "#008300",  # green
+    "#4a3aa7",  # violet
+    "#e34948",  # red
+    "#e87ba4",  # magenta
+    "#eb6834",  # orange
+]
+
+_INK_SECONDARY = "#52514e"  # body text
+_INK_MUTED = "#898781"      # axis titles / tick labels (mode-neutral)
+_GRIDLINE = "#e1e0d9"       # hairline grid
+_BASELINE = "#c3c2b7"       # axis line
+
+LAYOUT_PRESET: dict[str, object] = {
+    "colorway": COLORWAY,
+    "font": {
+        "family": 'system-ui, -apple-system, "Segoe UI", sans-serif',
+        "size": 13,
+        "color": _INK_SECONDARY,
+    },
+    "paper_bgcolor": "rgba(0,0,0,0)",
+    "plot_bgcolor": "rgba(0,0,0,0)",
+    "margin": {"t": 48, "r": 16, "b": 48, "l": 56},
+    "bargap": 0.25,  # keeps adjacent bars visually separated (mark-spec gap)
+}
+
+# Mark specs: thin marks — 2px lines, 8px markers.
+_LINE_STYLE = {"width": 2}
+_MARKER_STYLE = {"size": 8}
+
+
+def _axis(label: str) -> dict[str, object]:
+    """Recessive axis chrome + the column name as the axis title."""
+    return {
+        "title": {"text": label, "font": {"color": _INK_MUTED}},
+        "tickfont": {"color": _INK_MUTED},
+        "gridcolor": _GRIDLINE,
+        "linecolor": _BASELINE,
+        "automargin": True,
+    }
+
+
+def _clean(value: object) -> object:
+    """One JSON-safe scalar: numpy -> Python, NaN/NaT -> None, dates -> ISO strings.
+
+    Chart specs are persisted as JSONB and shipped to the FE, so every value in
+    the trace arrays must survive json.dumps. Upstream `_materialize` already
+    normalizes fully-numeric columns, but mixed columns can still carry Decimal,
+    and date columns carry Timestamps.
+    """
+    if value is None or value is pd.NaT:
+        return None
+    if isinstance(value, float) and np.isnan(value):
+        return None
+    if hasattr(value, "item"):  # numpy scalar (incl. np.datetime64 via Timestamp below)
+        value = value.item()
+        if isinstance(value, float) and np.isnan(value):
+            return None
+    if isinstance(value, Decimal):
+        return float(value)
+    if isinstance(value, pd.Timestamp):
+        return value.isoformat()
+    if isinstance(value, _dt.datetime | _dt.date):
+        return value.isoformat()
+    if isinstance(value, bool | int | float | str):
+        return value
+    return str(value)
+
+
+def _values(col: pd.Series) -> list[object]:
+    return [_clean(v) for v in col.tolist()]
+
+
+def _series_label(value: object) -> str:
+    """Trace name for one series group; null groups get an explicit label."""
+    cleaned = _clean(value)
+    return "(missing)" if cleaned is None else str(cleaned)
+
+
+# Prompt-style description read by the Planner to decide WHEN to pick this tool.
+DESCRIPTION = """\
+Summary: Build a chart specification (bar, line, pie, scatter) from an \
+already-retrieved table, for the app to render. Maps one x column and one y \
+column (pie: x -> slice labels, y -> slice values); an optional `series` column \
+splits bar/line/scatter into one trace per category. Produces a spec only — it \
+computes no numbers.
+
+USE WHEN the user EXPLICITLY asks to see a chart — "plot", "chart", "graph", \
+"visualize", "diagram" (ID: "grafik", "diagram", "plot", "visualisasikan", \
+"buatkan grafik/diagram"). ALWAYS a tail step: `data` consumes the output of an \
+upstream task that yields a table (retrieve_data or a table-kind analyze_*).
+
+DON'T USE WHEN:
+  - the user did not explicitly ask for a chart -> answer with tables/stats only \
+(never add a speculative chart)
+  - the numbers still need computing -> run the analyze_*/retrieve step first, \
+then chart its table output
+  - the upstream output is stats- or series-kind -> feed it a table-kind task \
+(e.g. a grouped retrieve or analyze_aggregate)
+
+Example questions:
+  - "plot revenue by region as a bar chart"
+  - "buatkan grafik penjualan per kategori"
+  - "show a pie chart of orders by sales channel"
+  - "visualize monthly revenue" (chain a month-grouped table first, then line)
+"""
+
+
+def render_chart(
+    df: pd.DataFrame,
+    chart_type: str,
+    x: str,
+    y: str,
+    series: str | None = None,
+    title: str | None = None,
+) -> dict[str, object]:
+    """Build a `dataeyond.chart.v1` envelope from a table (SPINE_V2_PLAN §4.2).
+
+    Args:
+        df: already-materialized data (the invoker resolves `data` upstream).
+        x: column for the x axis (pie: slice labels).
+        y: column for the y axis (pie: slice values).
+        series: optional column splitting bar/line/scatter into one trace per
+            distinct value (fixed-order colorway slots). Ignored for pie.
+        title: chart title; defaults to "<y> by <x>".
+
+    Returns:
+        The envelope dict: {schema, chart_type, title, plotly: {data, layout}}.
+
+    Raises:
+        UnsupportedChartTypeError: chart_type not in CHART_TYPES.
+        ColumnNotFoundError: x, y, or series absent from the data.
+    """
+    if chart_type not in CHART_TYPES:
+        raise UnsupportedChartTypeError(
+            f"unsupported chart_type '{chart_type}'; supported: {list(CHART_TYPES)}"
+        )
+    needed = [x, y] if chart_type == "pie" or series is None else [x, y, series]
+    missing = [c for c in needed if c not in df.columns]
+    if missing:
+        raise ColumnNotFoundError(f"columns not found: {missing}")
+
+    chart_title = title or f"{y} by {x}"
+
+    if chart_type == "pie":
+        traces: list[dict[str, object]] = [
+            {"type": "pie", "labels": _values(df[x]), "values": _values(df[y])}
+        ]
+    else:
+        plotly_type = "bar" if chart_type == "bar" else "scatter"
+        mode = {"line": "lines+markers", "scatter": "markers"}.get(chart_type)
+
+        def _trace(frame: pd.DataFrame, name: str) -> dict[str, object]:
+            t: dict[str, object] = {
+                "type": plotly_type,
+                "x": _values(frame[x]),
+                "y": _values(frame[y]),
+                "name": name,
+            }
+            if mode is not None:
+                t["mode"] = mode
+                t["line"] = dict(_LINE_STYLE)
+                t["marker"] = dict(_MARKER_STYLE)
+            return t
+
+        if series is None:
+            traces = [_trace(df, y)]
+        else:
+            # sort=True keeps trace order (and therefore colorway slot
+            # assignment) deterministic across runs of the same data.
+            traces = [
+                _trace(group, _series_label(value))
+                for value, group in df.groupby(series, dropna=False, sort=True)
+            ]
+
+    layout: dict[str, object] = {
+        **LAYOUT_PRESET,
+        "title": {"text": chart_title},
+        # Legend only when there is more than one thing to identify: multi-series
+        # traces, or pie slices (a single named series is titled, not legended).
+        "showlegend": len(traces) > 1 or chart_type == "pie",
+    }
+    if chart_type != "pie":
+        layout["xaxis"] = _axis(x)
+        layout["yaxis"] = _axis(y)
+
+    return {
+        "schema": "dataeyond.chart.v1",
+        "chart_type": chart_type,
+        "title": chart_title,
+        "plotly": {"data": traces, "layout": layout},
+    }

src/tools/contracts.py

@@ -68,7 +68,7 @@ class ToolRegistry(BaseModel):
 
 class ToolOutput(BaseModel):
     tool: str
-    kind: Literal["scalar", "table", "stats", "series", "documents", "error"]
+    kind: Literal["scalar", "table", "stats", "series", "documents", "chart", "error"]
     value: Any | None = None
     columns: list[str] | None = None
     rows: list[list[Any]] | None = None

src/tools/invoker.py

@@ -36,6 +36,7 @@ from src.tools.analytics import (
     relationship,
     segmentation,
     temporal,
+    visualization,
 )
 from src.tools.contracts import ToolOutput
 from src.tools.data_access import DATA_ACCESS_TOOLS, DataAccessToolInvoker
@@ -54,6 +55,7 @@ _DISPATCH: dict[str, tuple[Callable[..., Any], str]] = {
     "analyze_segment": (segmentation.analyze_segment, "table"),
     "analyze_trend": (temporal.analyze_trend, "series"),
     "analyze_merge": (merge.analyze_merge, "table"),
+    "render_chart": (visualization.render_chart, "chart"),
 }
 
 

src/tools/registry.py

@@ -34,10 +34,12 @@ from src.tools.analytics import (
     relationship,
     segmentation,
     temporal,
+    visualization,
 )
 from src.tools.contracts import ToolRegistry, ToolSpec
 
-# Active this round — the four analytics tools the Planner may select.
+# Active this round — the analytics (+ render_chart, SPINE_V2_PLAN §4.1) tools
+# the Planner may select.
 ACTIVE_ANALYTICS_TOOLS: list[ToolSpec] = [
     ToolSpec(
         name="analyze_descriptive",
@@ -113,6 +115,23 @@ ACTIVE_ANALYTICS_TOOLS: list[ToolSpec] = [
         output_kind="table",
         description=merge.DESCRIPTION,
     ),
+    ToolSpec(
+        name="render_chart",
+        category="analytics.visualization",
+        input_schema={
+            "required": ["data", "chart_type", "x", "y"],
+            "properties": {
+                "data": {"type": "string"},
+                "chart_type": {"type": "string"},
+                "x": {"type": "string"},
+                "y": {"type": "string"},
+                "series": {"type": "string"},
+                "title": {"type": "string"},
+            },
+        },
+        output_kind="chart",
+        description=visualization.DESCRIPTION,
+    ),
 ]
 
 # Deferred this round — specs kept intact for easy re-activation, NOT exposed to

src/traceability/scratchpad.py

@@ -71,13 +71,40 @@ def _output_to_dict(output: Any) -> dict[str, Any]:
         ]
     value = getattr(output, "value", None)
     if value is not None:
-        result["value"] = _truncate(value)
+        if kind == "chart" and isinstance(value, dict):
+            # The full dataeyond.chart.v1 envelope is persisted in message_charts
+            # (SPINE_V2_PLAN §4.4); traceability keeps a compact summary — the
+            # spec's x/y arrays are numeric, so _truncate's string caps would let
+            # them through whole.
+            result.update(_chart_summary(value))
+        else:
+            result["value"] = _truncate(value)
     error = getattr(output, "error", None)
     if error is not None:
         result["error"] = _truncate(error)
     return result
 
 
+def _chart_summary(envelope: dict[str, Any]) -> dict[str, Any]:
+    """Compact wire shape for a chart-kind output: chart_type/title/trace_count/
+    point_count (charts have no rows, so `row_count` is absent — §4.7)."""
+    plotly = envelope.get("plotly")
+    traces = plotly.get("data") if isinstance(plotly, dict) else None
+    traces = traces if isinstance(traces, list) else []
+    points = 0
+    for trace in traces:
+        if isinstance(trace, dict):
+            axis = trace.get("x") if trace.get("x") is not None else trace.get("labels")
+            if isinstance(axis, list):
+                points += len(axis)
+    return {
+        "chart_type": envelope.get("chart_type"),
+        "title": envelope.get("title"),
+        "trace_count": len(traces),
+        "point_count": points,
+    }
+
+
 def _meta_of(output: Any) -> dict[str, Any]:
     """Best-effort read of a tool result's `meta` dict (ToolOutput or plain dict)."""
     if isinstance(output, dict):
@@ -109,6 +136,11 @@ def _summarize(name: str, out_dict: dict[str, Any], meta: dict[str, Any]) -> str
     if name.startswith("analyze_"):
         pretty = name.removeprefix("analyze_").replace("_", " ")
         return f"Ran {pretty} analysis on {ncol} columns" if ncol else f"Ran {pretty} analysis"
+    if name == "render_chart":
+        ctype = out_dict.get("chart_type") or "chart"
+        points = out_dict.get("point_count")
+        summary = f"Built a {ctype} chart"
+        return f"{summary} ({points} data points)" if points else summary
     return name.replace("_", " ")