fix/ check and new chart tool

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`
@@ -353,10 +356,24 @@ Report v2 fields (added 2026-07-09; all default-empty, so older stored reports r
 - `unresolved` — runs that were attempted but produced no usable evidence (every `analyze_*` step failed). Not part of the findings body.
 - `excluded` — runs the caller excluded via `exclude_record_ids`.
 - `evidence_tables` — `record_id` → small result tables copied from the run's stored outputs (max 3 tables per record, max 10 rows each; `truncated: true` when rows were capped). Rendered as markdown tables under the matching Key Findings group in `rendered_markdown`.
+- `charts` *(added 2026-07-14)* — `record_id` → `dataeyond.chart.v1` envelopes (see §Charts) copied verbatim from the run's stored outputs (max 3 per record). `rendered_markdown` gains an `## EDA` section where each chart appears as a fenced block the frontend renders with plotly.js:
+
+  ````text
+  ```plotly
+  {
+    "schema": "dataeyond.chart.v1",
+    "chart_type": "bar",
+    "title": "…",
+    "plotly": { "data": [ … ], "layout": { … } }
+  }
+  ```
+  ````
+
+  The fence content is the **full envelope** (same shape as `charts[].spec` on `GET /api/v1/charts`) — the FE hook parses it and renders `Plotly.newPlot(el, parsed.plotly.data, parsed.plotly.layout)`. A bold caption line (chart title) precedes each fence.
 
 Precondition:
 
-- Reports require at least one completed analysis record for the session.
+- Reports require at least one completed analysis record for the session (*updated 2026-07-14:* a run whose `render_chart` succeeded counts — a chart-only session can generate a report).
 - If slow-path analysis recording is disabled, report generation can return `409` by design.
 
 ### `GET /api/v1/tools/report/{analysis_id}`
@@ -643,3 +660,114 @@ 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 `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.
+
+> **Updated 2026-07-13 (lead review):** lookup is now by **`message_id` alone** (it is a server-minted UUID — globally unique; `analysis_id` was removed from the query), and every response is **HTTP 200 with an explicit `status` marker** instead of a bare list: `success` (≥1 chart), `empty` (the turn completed but produced no charts — the common case), `not_found` (no completed turn is known for this `message_id`: mistyped/stale id, or an error turn). *(Note the asymmetry: `GET /api/v1/traceability` still takes `analysis_id` + `message_id` — aligning it is a separate change if wanted.)*
+
+Query params:
+
+| Query | Required | Description |
+| --- | --- | --- |
+| `message_id` | Yes | Assistant answer identifier returned by the stream's `done` event. |
+
+Example:
+
+```text
+GET /api/v1/charts?message_id=88f10c3a-6f03-4204-bf98-41ffc20388b2
+```
+
+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` (`status: success` — ≥1 chart):
+
+```json
+{
+  "status": "success",
+  "message": "1 chart(s) for this message.",
+  "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` (`status: empty` — chartless turn, the common case; not an error):
+
+```json
+{
+  "status": "empty",
+  "message": "This message completed without producing charts.",
+  "count": 0,
+  "charts": []
+}
+```
+
+Response `200` (`status: not_found` — no completed turn known for this id):
+
+```json
+{
+  "status": "not_found",
+  "message": "No completed turn is known for this message_id.",
+  "count": 0,
+  "charts": []
+}
+```
+
+Field rules:
+
+- `status` is the outcome marker: `success` | `empty` | `not_found` (always HTTP 200 — the FE fetches unconditionally, so a missing turn is a data state, not a transport failure). `empty` vs `not_found` is decided against the turn's traceability row (written before `done`), so `not_found` reliably means "this id never completed a turn".
+- `message` is a human-readable line for logs/debugging — do not parse it; branch on `status`.
+- `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` / `analysis_id` — charts are keyed by `message_id` alone.
+
+> **DDL note (Harry / dedorch migration):** the original manual index is `(analysis_id, message_id)`, which does not serve a `message_id`-only lookup. Additive index for the migration (also safe to run manually now):
+> ```sql
+> CREATE INDEX IF NOT EXISTS idx_message_charts_message ON message_charts (message_id);
+> ```
+
+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.

DEV_PLAN.md

@@ -68,6 +68,30 @@ base64-mangled from Go. Fix tasks (same status legend as §0):
 | Q10 | Traceability `data_used` layer — resolve IR ids → real names for the FE (users couldn't map `c_…`/`t_…` ids back to their data; aggregate aliases like `total_revenue` looked like real columns) | Rifqi ↔ FE | ✅ | done 2026-07-13 (pr/15): new `src/traceability/resolve.py` builds `data_used[]` (real source/table/column names; joins; plain-language filters; `columns_read` vs `output_columns` with `computed`+`formula`), `tool_calls[].summary`, `sources[]` gains `source_name`+all tables; **ids kept but machine-only (FE must not render)**; deterministic no-LLM, never-throw; catalog threaded to the scratchpad at the composition root. Contract + `TRACEABILITY_FE_HANDOFF.md` updated; FE wiring pending (Rifqi → FE). Additive/non-breaking |
 | Q11 | IR wart: `OrderByClause.column_id` may hold a SELECT **alias** (a computed output), not a catalog column_id | Rifqi | 🔎 | surfaced by Q10 — the resolver tolerates it (`kind: "computed"` fallback), but the IR field name is misleading. Consider an explicit `by_alias` field or renaming. Low priority; no functional bug |
 
+## 0.6. pr/16 sprint — Spine v2: W2 charts + W1 checkpoint (SPINE_V2_PLAN, approved 2026-07-13)
+
+Scope approved 2026-07-13 (Rifqi + Sofia): build W2 (`render_chart` + chart store + `GET /charts`
++ planner viz slice) and W1 (S1a quality checkpoint). **W3 (activate deferred `analyze_*`) deferred
+at approval — do not start until Rifqi re-opens.** W4 (S1b repair) stays gated on INV-6 sign-off +
+S1a telemetry. Design + handoff source: `SPINE_V2_PLAN.md`.
+
+| # | Task | Owner | Status | Note |
+|---|---|---|---|---|
+| V1 | `render_chart` tool slice: `visualization.py` (Plotly-JSON `dataeyond.chart.v1` envelope, no plotly dep) + `ToolOutput.kind` `"chart"` + registry + invoker | Rifqi (Sofia signed off on the tool-layer edit) | ✅ | done 2026-07-13; deterministic spec builder (bar/line/pie/scatter, fixed style preset), traceability scratchpad summarizes chart outputs compactly (point_count, not the raw arrays) |
+| V2 | Chart store + API: `MessageChartRow` (`message_charts`) + `src/charts/store.py` (never-throw save) + write site in `_run_slow_path` + `GET /api/v1/charts` + contract §charts | Rifqi | ✅ | done 2026-07-13; empty list = valid 200; `done` event unchanged (no `chart_count` — open, Harry); FE fetches unconditionally on `done` |
+| V3 | Planner viz slice: recipe table + "Charts only on explicit ask" rule (planner.md), Example J (viz tail) + Example K (viz-infeasible), validator Check 10 (`render_chart.data` must be table-kind), assembler chart one-liner guard | Rifqi | ✅ | done 2026-07-13; behavioral matrix verified in-process (real LLM): explicit ask EN/ID → chart tail; plain question → no chart; absent dimension → infeasible (Example K + prompt guard added after the first smoke force-mapped `status` AS "region") |
+| V4 | W1 S1a quality checkpoint: `slow_path/checkpoint.py` (CK1–CK6) + `RunAssessment` schemas + coordinator call site + "Execution assessment" block in the assembler input + `refusals.run_failure_message` | Rifqi | ✅ | done 2026-07-13; 13 local tests (one per CK rule + never-throw + prompt + coordinator); CK1 all-failed → deterministic honest failure, **no assembler call**; clean run renders nothing (zero behavior change); every flag logs `repair_candidate` (S1b evidence) |
+| V5 | `message_charts` DDL: run manually against dedorch (block for the live e2e chart test), then hand the schema to Harry for the dedorch migration | Rifqi → Harry | ✅ | Rifqi ran the DDL 2026-07-13; **live e2e ALL PASS** same day (real v2 endpoint: viz turn → chart row keyed by `done` message_id → `GET /charts` valid v1 envelope; chartless → 200 empty; injected `render_chart` failure → answer streams as a table, no row). **Remaining: send Harry the migration handoff** (schema + contract §charts) |
+| V6 | Restore the `eval.chat_sim` harness — `eval/chat_sim/*.py` is missing from disk AND git (only `__pycache__` remains); §7B prompt gate can't run | Rifqi | ✅ | restored by Rifqi 2026-07-13 (accidental delete). ⚠️ its hard-coded `DEFAULT_USER_ID`/`TITANIC_SOURCE_ID` are stale (that user has no catalog; the Titanic blobs are gone) — update the constants before the next full run |
+| V7 | Local `.env` lagged Go's Supabase-S3 data plane: `storage_provider=azure_blob` + empty `supabase_s3_*` made EVERY local tabular retrieve fail `BlobNotFound` | Rifqi | ✅ | found during the e2e (masked as "data not available" by the honest-degrade path); Rifqi set the six values 2026-07-13. Gotcha documented in REPO_STATUS §13 |
+| V8 | Lead review of `GET /charts`: lookup by `message_id` alone + tri-state response marker (`status: success \| empty \| not_found` + `message`) instead of a bare list | Rifqi (lead ask) | ✅ | done 2026-07-14; `not_found` vs `empty` decided against the turn's traceability row (PK lookup); always HTTP 200; contract §charts updated. ⚠️ additive DDL for the new lookup: `CREATE INDEX IF NOT EXISTS idx_message_charts_message ON message_charts (message_id);` (run manually + include in Harry's migration). Traceability GET still takes both params — aligning it is open |
+| V9 | Report chart embedding: `AnalysisReport.charts` (verbatim envelopes per record) + `## EDA` section with ` ```plotly ` fences (content = the **full v1 envelope**, pretty-printed — the FE hook's verified shape); `has_successful_analysis` extended so a successful `render_chart` counts (chart-only sessions satisfy the report floor) | Rifqi | ✅ | done 2026-07-14; first cut emitted bare `{data, layout}` — FE test showed the hook parses the full envelope, fixed same day; live reports v3 (wrong fence) → **v4 (correct)** for analysis `7be50846…` (3 charts embedded); suite **381 passed, 7 skipped** (+5 chart-embed tests; same 2 pre-existing failures) |
+
+Full-suite evidence for this sprint: **376 passed, 7 skipped** (+13 new checkpoint tests; the 2
+failures — `test_chat_handler::test_structured_flow_runs_slow_path`,
+`test_reader::test_structured_read_falls_back_to_user_scope_when_no_analysis_row` — reproduce at
+HEAD before this diff, i.e. pre-existing). Ruff clean on all touched paths; `import main` OK.
+
 ## 1. The direction change (locked decisions from 2026-06-24)
 
 1. **"Problem statement" is replaced by two user-entered fields: `objective` + `business_questions`.**
@@ -195,7 +219,7 @@ Status legend: ⬜ not started · 🔄 in progress · ✅ done · ⛔ blocked ·
 | 23 | Report markdown formatting: tables, **bold**, *italic*, horizontal separators | Sofhia | ✅ | Done 2026-06-25. Added `---` separators between header + each section in `_render_markdown`. Tables (EDA) / bold (method labels) / italic (meta + citations) already emitted. Relaxed `report_summary.md` to allow inline `**bold**`/`*italic*` for emphasis (kept no-headings/no-bullets so it doesn't duplicate the section structure / Key Findings). Compile + ruff clean |
 | 24 | Clarify report input contract: records table (+ `last_report` for edit mode?) | Rifqi/Sofhia ↔ Harry | ⬜ new | Edit-mode input left open at the checkpoint |
 | 25 | Migrate Python chat path to Go `analyses_messages` (+ `analyses`) | Rifqi ↔ Harry | ✅ | Done 2026-07-02. Read path already on `analyses_messages` (commit `0066161`). This change makes Python **read-only**: removed the `save_messages` calls from `/api/v2/chat/stream` so **Go is the sole writer** — fixes the double-write both Go+Python were producing. `load_history` still reads `analyses_messages`. v1 `/chat/stream` is unwired so left untouched |
-| 26 | **Charts (DEFERRED):** store Plotly JSON in a future `chart` table (not matplotlib PNG) | — | ⏸️ | After the markdown path is done end-to-end |
+| 26 | **Charts (DEFERRED):** store Plotly JSON in a future `chart` table (not matplotlib PNG) | — | 🔎 | **Landed 2026-07-13 (pr/16, §0.6 V1–V3):** `render_chart` + `message_charts` + `GET /api/v1/charts`, Plotly JSON as decided. 🔎 pending the dedorch DDL run (V5) + live e2e |
 | 27 | **Images (DEFERRED):** image table (id, analysis_id, msg/report ref, order) + originals in a bucket | — | ⏸️ | Maintenance-heavy; parked |
 | 28 | **UI research** (FE): new-analysis form, knowledge menu (user vs analysis level), report artifacts + version selector | Team | ⬜ new | No dedicated UI person; interview + old analysis UI removed |
 
@@ -218,3 +242,8 @@ Status legend: ⬜ not started · 🔄 in progress · ✅ done · ⛔ blocked ·
 - **Report input for edit mode** — does Python need the last report content? (#24)
 - ~~`report_inputs` write scope — every agent call vs slow-path-only? (#21)~~ RESOLVED: one row per slow-path run (telemetry stays Langfuse).
 - **Python history source** — confirm Go's `analysis_message` (#25).
+- **`done.chart_count`** — additive field on the SSE `done` event so the FE can skip `GET /charts`
+  on chartless turns? (Harry; SPINE_V2_PLAN §4.5.) Until decided the FE fetches unconditionally.
+- **W3 re-open timing** (deferred `analyze_*` activation) — Rifqi (deferred at the 2026-07-13 approval).
+- **INV-6 relaxation for S1b targeted repair** — team, only after S1a `repair_candidate` telemetry
+  shows a meaningful hit-rate (SPINE_V2_PLAN §6).

REPO_STATUS.md

@@ -2,7 +2,7 @@
 
 **Audience:** teammates onboarding onto the Python repo (`Agentic-Service-Data-Eyond-Catalog`).
 **Scope:** what the code does **right now** (branch `pr/4`, ticket KM-652). Describes current state only — no roadmap or to-dos.
-**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Agent-quality fixes 2026-07-08 (pr/13):** from the scoped live-test review — the planner gains an explicit **infeasible** outcome (`TaskList.infeasible_reason` → deterministic EN/ID data-gap reply via `refusals.data_gap_message`; no more force-mapping absent measures like `pa` AS "revenue"), the IR validator rejects bare selects under `group_by` (self-corrects via the planner retry), `analyze_trend` handles integer year/month columns (was collapsing every row into one 1970-01 bucket), planner few-shots add top-N (Example G) + infeasible (Example H), numeric catalog `sample_values` are base64-decoded at read (`catalog/sample_decode.py` — stopgap for Go's byte-marshaling; primary fix is Go-side), traceability no longer emits null source rows for failed retrievals, and `check_data` hides `-1` row counts. **Report v2 + analyze_merge planner support 2026-07-09 (pr/13):** Sofia's `analyze_merge` tool (8abf635, KM-703) is now planner-supported (`_validate_data_source` guards `data_right`, two-retrieve→merge few-shot Example I, planner.md "Two measures per entity" rule); the report gains per-business-question answers (`bq_answers` — drafted by the SAME single LLM call, index-based record refs, deterministic fallback unchanged), "Attempted, Unresolved" + "Excluded Analyses" sections (failed runs are no longer silently dropped), evidence tables copied from `results_snapshot` (table-kind outputs, ≤3/record ≤10 rows ≤8 cols, `check_*` skipped), normalized caveat dedupe with caps (12/10), and single-language output via `detect_reply_language`; the report surface adds `GET /tools/report/{analysis_id}/records` (curation list), `GET …/readiness` (FE delta guard), and `exclude_record_ids` on POST — see API_CONTRACT_BE_PYTHON.md. **Report compaction 2026-07-09 (pr/13):** the rendered markdown drops the "Notes & Limitations", "Attempted, Unresolved", and "How This Was Analyzed" sections (team decision — compact report; render blocks commented out in `report/generator.py`, not deleted). The JSON body keeps `caveats`/`open_questions`/`unresolved`/`method_steps` and the curation/records endpoints are unchanged. **Traceability `data_used` layer 2026-07-13 (pr/15):** `GET /api/v1/traceability` gains a resolved, user-facing `data_used[]` block (one per `retrieve_data` call) — real source/table/column names, joins, plain-language filters, and result columns split into read-from-data vs `computed` (with `formula`, e.g. `total_revenue` = `SUM(line_total)`, so an alias is never shown as a real column). Ids are kept but **machine-only (FE must not render)**. Also adds `tool_calls[].summary`, and `sources[]` now carry `source_name` + every table touched. Deterministic catalog resolution (new `src/traceability/resolve.py`), no LLM, never-throw; catalog threaded to the scratchpad at the slow-path composition root. Additive/non-breaking; contract + `TRACEABILITY_FE_HANDOFF.md` updated. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
+**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Agent-quality fixes 2026-07-08 (pr/13):** from the scoped live-test review — the planner gains an explicit **infeasible** outcome (`TaskList.infeasible_reason` → deterministic EN/ID data-gap reply via `refusals.data_gap_message`; no more force-mapping absent measures like `pa` AS "revenue"), the IR validator rejects bare selects under `group_by` (self-corrects via the planner retry), `analyze_trend` handles integer year/month columns (was collapsing every row into one 1970-01 bucket), planner few-shots add top-N (Example G) + infeasible (Example H), numeric catalog `sample_values` are base64-decoded at read (`catalog/sample_decode.py` — stopgap for Go's byte-marshaling; primary fix is Go-side), traceability no longer emits null source rows for failed retrievals, and `check_data` hides `-1` row counts. **Report v2 + analyze_merge planner support 2026-07-09 (pr/13):** Sofia's `analyze_merge` tool (8abf635, KM-703) is now planner-supported (`_validate_data_source` guards `data_right`, two-retrieve→merge few-shot Example I, planner.md "Two measures per entity" rule); the report gains per-business-question answers (`bq_answers` — drafted by the SAME single LLM call, index-based record refs, deterministic fallback unchanged), "Attempted, Unresolved" + "Excluded Analyses" sections (failed runs are no longer silently dropped), evidence tables copied from `results_snapshot` (table-kind outputs, ≤3/record ≤10 rows ≤8 cols, `check_*` skipped), normalized caveat dedupe with caps (12/10), and single-language output via `detect_reply_language`; the report surface adds `GET /tools/report/{analysis_id}/records` (curation list), `GET …/readiness` (FE delta guard), and `exclude_record_ids` on POST — see API_CONTRACT_BE_PYTHON.md. **Report compaction 2026-07-09 (pr/13):** the rendered markdown drops the "Notes & Limitations", "Attempted, Unresolved", and "How This Was Analyzed" sections (team decision — compact report; render blocks commented out in `report/generator.py`, not deleted). The JSON body keeps `caveats`/`open_questions`/`unresolved`/`method_steps` and the curation/records endpoints are unchanged. **Traceability `data_used` layer 2026-07-13 (pr/15):** `GET /api/v1/traceability` gains a resolved, user-facing `data_used[]` block (one per `retrieve_data` call) — real source/table/column names, joins, plain-language filters, and result columns split into read-from-data vs `computed` (with `formula`, e.g. `total_revenue` = `SUM(line_total)`, so an alias is never shown as a real column). Ids are kept but **machine-only (FE must not render)**. Also adds `tool_calls[].summary`, and `sources[]` now carry `source_name` + every table touched. Deterministic catalog resolution (new `src/traceability/resolve.py`), no LLM, never-throw; catalog threaded to the scratchpad at the slow-path composition root. Additive/non-breaking; contract + `TRACEABILITY_FE_HANDOFF.md` updated. **Spine v2 W2+W1 2026-07-13 (pr/16):** §6/§7/§8/§9/§12 — `render_chart` tool lands (first of the `render_*` family: deterministic Plotly-JSON `dataeyond.chart.v1` envelope, hand-built, **no plotly dependency**; planner-selected only on an explicit chart ask, EN/ID) + Python-owned `message_charts` store + `GET /api/v1/charts` (FE fetches on `done`, same pattern as traceability; empty list = valid 200); planner gains a named **recipe table** + viz few-shots (Example J tail, Example K viz-infeasible) + validator Check 10 (`render_chart.data` must reference a table-producing task); and the slow path gains the **S1a quality checkpoint** (`slow_path/checkpoint.py`, 0 LLM, never-throw, between runner and assembler: CK1 all-failed → deterministic honest-failure answer with **no** assembler call, CK2 empty retrieve + downstream, CK3 10k-cap truncation, CK4 single trend bucket, CK5 all-null column, CK6 chart-spec sanity; flags render as an "Execution assessment" block in the assembler input and every flag logs `repair_candidate` — the S1b evidence base). Design + handoff doc: `SPINE_V2_PLAN.md`. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
 the **Go source** (`Orchestrator-Agent-Service`), not its docs. The Go service has moved well past its
 own (uncommitted, stale) design docs: it now hosts the **dedorch SQL migrations** in-repo and a full
 **`/api/v1/analyses` + `/api/v1/skills`** REST surface. Go does **not** call Python yet — those skills
@@ -141,33 +141,36 @@ Two facts to internalise:
 
 ## 6. Feature list (what's built)
 
-- **5-intent handler router** (`chat`/`help`/`check`/`unstructured_flow`/`structured_flow`) with history-aware query rewriting (EN/ID).
+- **6-intent handler router** (`chat`/`help`/`check`/`unstructured_flow`/`structured_flow`/`out_of_scope`, the last added 2026-07-03) with history-aware query rewriting (EN/ID).
 - **Skills:** `help` (LLM, state-aware next-step guidance), `check` (no-LLM data/document inventory). *(The `problem_statement` skill and the `problem_validated` gate were removed 2026-06-25 — KM-652; `gate.py` kept as a no-op seam, `problem_statement.py` kept but unwired.)*
-- **Slow analytical path:** Planner → TaskRunner → Assembler (static plan, degrade-and-continue, 3 LLM calls fixed).
+- **Slow analytical path:** Planner → TaskRunner → **S1a quality checkpoint** (0 LLM, added 2026-07-13) → Assembler (static plan, degrade-and-continue, 3 LLM calls fixed; an all-failed run now short-circuits to a deterministic honest-failure answer — 2 calls).
 - **Structured query engine:** catalog-driven JSON IR → deterministic SQL/pandas compiler → read-only executor, with **single-level FK joins** (DB sources only).
 - **Unstructured RAG** over PGVector.
-- **Analytics tools:** 4 registered composite `analyze_*` (descriptive, aggregate, correlation, trend) + 4 data-access tools (check_data, check_knowledge, retrieve_data, retrieve_knowledge). Four further composites (comparison, contribution, profile, segment) exist in code but are **not registered** with the Planner.
+- **Analytics tools:** 6 registered (5 composite `analyze_*` — descriptive, aggregate, correlation, trend, merge — plus `render_chart`, added 2026-07-13) + 4 data-access tools (check_data, check_knowledge, retrieve_data, retrieve_knowledge). Four further composites (comparison, contribution, profile, segment) exist in code but are **not registered** with the Planner (W3, deferred).
+- **Charts (S2, 2026-07-13; updated 2026-07-14):** planner-selected `render_chart` (only on an explicit chart ask, EN/ID) builds a `dataeyond.chart.v1` Plotly-JSON envelope; persisted to Python-owned `message_charts` before `done`, fetched via `GET /api/v1/charts?message_id=` (tri-state `status` marker: success/empty/not_found). SSE stays text-only. **Reports embed charts too** (2026-07-14): the generator copies chart envelopes from `results_snapshot` into an `## EDA` section as ` ```plotly ` fenced blocks (FE hook renders them); a successful `render_chart` now counts toward the report floor (`has_successful_analysis`).
 - **Versioned report generation** from persisted records.
 - **Analysis sessions:** data-first creation gate (≥1 bound source); each turn reads the analysis-scope catalog so it sees only that analysis's bound sources.
 - **Langfuse tracing** (PII-masked), **Redis caching**, **pooled DB engines** + speculative prewarm.
 
 ---
 
-## 7. API surface (this repo, all under `/api/v1`)
+## 7. API surface (this repo)
 
-> **» pr/5 (decided, not yet in code):** chat → `/api/v2/chat/stream` (`analysis_id`); `/tools` splits
-> into `/tools/list` + `/tools/help` + `/tools/report`; new `/api/v1/traceability`; and the
-> analysis-CRUD / `room` / `users` / `document` / `db_client` / `data_catalog` rows are unwired from
-> `main` + Swagger. See the Direction-update banner.
+> ✅ **pr/5 restructure IN CODE (table refreshed 2026-07-13).** The banner that stood here
+> ("decided, not yet in code") is done: chat lives at `/api/v2/chat/stream`, the skills regrouped
+> under `/api/v1/tools/*`, `traceability` and (2026-07-13) `charts` are mounted, and the analysis-CRUD
+> / `room` / `users` / `document` / `db_client` / `data_catalog` routers are unwired from `main` +
+> Swagger (files kept, commented mounts). Table below is the **live** surface (`main.py` mounts).
 
 | Endpoint | Purpose | Caller |
 |---|---|---|
-| `POST /chat/stream` | Main chat SSE (router → dispatch) | FE → Go → Python (the only FE→Python call today) |
-| `DELETE /chat/cache` · `/chat/cache/room/{id}` · `/retrieval/cache/{user_id}` | Cache management | internal / ops |
-| `POST /analysis/create` · `GET /analysis` · `GET /analysis/{id}` | Analysis-session CRUD (state + room + bindings created atomically) | intended FE → Go |
-| `POST /report` · `GET /report/{id}` · `GET /report/{id}/{ver}` | Report generate / list / fetch | FE → Go (report button) |
-| `GET /tools` | Slash-command catalog (static, cacheable) | Go caches it for the FE "/" menu |
-| `users` · `room` · `document` · `db_client` · `data_catalog` routers | Phase-1 legacy; functionally migrated to Go | mostly dormant |
+| `POST /api/v2/chat/stream` | Main chat SSE (`analysis_id`; router → dispatch) | FE → Go → Python |
+| `GET /api/v1/tools/list` | Slash-command catalog (static, cacheable) | Go caches it for the FE "/" menu |
+| `POST /api/v1/tools/help` | State-aware help skill | FE → Go → Python |
+| `POST /api/v1/tools/report` (+ `GET …/records` · `…/readiness` · `…/{analysis_id}/{version}` GETs) | Report generate / curate / readiness / fetch | FE → Go (report button) |
+| `GET /api/v1/traceability` | Per-turn provenance (fetched on `done`) | FE → Go → Python |
+| `GET /api/v1/charts?message_id=` | Per-turn `render_chart` envelopes (fetched on `done`); always 200 with `status: success\|empty\|not_found` — added 2026-07-13, reshaped per lead review 2026-07-14 | FE → Go → Python |
+| `users` · `room` · `document` · `db_client` · `data_catalog` · v1 `chat` · analysis-CRUD routers | Unwired (files kept in tree, not mounted) | — |
 
 ---
 
@@ -187,6 +190,8 @@ unless `SKIP_INIT_DB=true`.
 | `reports` *(dedorch)* | uuid, `analysis_id`, `user_id`, `title` + markdown `content` + `version` (UNIQUE per analysis) | Go + Python ReportStore | report API |
 | `data_sources` *(dedorch, Go-owned)* | per-analysis binding table. **Python no longer reads or writes it** — bindings live in Go's `analyses.data_bind`, which Go materializes into the analysis-scope `data_catalog` row; Python scopes off that row. The table exists (Go migration) but Python is fully decoupled — do **not** drop it manually | Go migration | — (unused by Python) |
 | `analyses_messages` *(dedorch)* | the analysis chat room (`role ∈ user\|ai`); replaces deprecated `rooms`/`chat_messages` | Go `/analyses/{id}/messages` | Python chat path **not yet migrated here** (§12) |
+| `message_traceability` *(Python-owned)* | one jsonb `TraceabilityPayload` per assistant turn (PK `message_id`); flushed before `done` | chat pipeline (KM-691) | `GET /api/v1/traceability` |
+| `message_charts` *(Python-owned, added 2026-07-13)* | one row per `render_chart` chart — `spec` jsonb holds the full `dataeyond.chart.v1` envelope; keyed (`analysis_id`, `message_id`), multiple rows per turn allowed; written before `done`, never-throw | slow-path chart persist (`chat_handler._run_slow_path`) | `GET /api/v1/charts` |
 
 > ✅ **Python ORM ↔ dedorch drift — reconciled 2026-07-01.** `AnalysisStateRow` (`analyses`) dropped
 > `problem_statement`/`problem_validated` and added `objective`/`business_questions` (Harry's #3);
@@ -243,13 +248,22 @@ dedorch state migration (#3/#4) renames it.
 - **TaskRunner** (`slow_path/task_runner.py`) — deterministic, 0 LLM. Wave-based execution,
   `${t<id>}` placeholder resolution (Pattern A), never-throw invocation, **degrade-and-continue**
   (failed task → dependents skipped, independent branches run). No replanning.
+- **Quality checkpoint (S1a)** (`slow_path/checkpoint.py`, added 2026-07-13) — deterministic,
+  0 LLM, never-throw inspection between runner and assembler. CK1 all-failed → the coordinator
+  returns a deterministic honest-failure answer (`refusals.run_failure_message`, EN/ID) with **no**
+  assembler call and a non-substantive record; CK2 empty retrieve (+ transitive dependents),
+  CK3 10k-cap truncation, CK4 single trend bucket, CK5 all-null column consumed, CK6 chart-spec
+  sanity (§4.6 of SPINE_V2_PLAN). Degraded flags render as an "# Execution assessment" block in the
+  assembler's human content; every flag logs `repair_candidate` via structlog (the gated-S1b
+  evidence base). A clean run renders nothing — zero behavior change.
 - **Assembler** (`slow_path/assembler.py`) — 1 LLM call authoring only the narrative; code copies
   the structured `results_snapshot` / `tasks_run` from the run state into the `AnalysisRecord`
   (the report's source of truth).
 
 Streaming + persistence: `chat_handler._run_slow_path` bridges per-stage progress to SSE `status`
 events, prewarms the DB engine in parallel with planning, emits the answer, then persists the
-record stamped with `user_id` + `analysis_id`.
+record stamped with `user_id` + `analysis_id`, and (2026-07-13) any `kind="chart"` outputs to
+`message_charts` — both never-throw, both before `done`.
 
 ### Structured query engine — `src/query/`
 `QueryService.run` (`query/service.py`): plan → validate → retry(3) → dispatch → execute; **never
@@ -275,14 +289,23 @@ to the whole (mis-named) user catalog.
 `retrieve_data` runs a pre-built IR (validate → dispatch → execute, skipping the planner) and
 coerces `Decimal`→`float` — the Pattern A handoff the `analyze_*` tools consume. The planner
 registry composes a local data-access spec stub (name-checked against `DATA_ACCESS_TOOLS`) with the
-real `analytics_registry()`.
+real `analytics_registry()`. **2026-07-13:** `analytics_registry()` also exposes `render_chart`
+(`src/tools/analytics/visualization.py`, category `analytics.visualization`, `output_kind="chart"`
+— `ToolOutput.kind` gained `"chart"`): a pure spec builder mapping a table to a Plotly-JSON
+envelope (bar/line/pie/scatter, fixed house style preset, **no plotly import**); the planner
+validator's Check 10 forces its `data` to reference a table-producing task.
 
 ### Report — `src/agents/report/`
 `generator.py` reads records, deterministically assembles structured fields, 1 LLM call for the
 executive summary; `store.py` versions under an advisory lock and persists markdown to dedorch
-`reports`; `readiness.py` defines the **report floor** (≥1 successful `analyze_*`; the
-`problem_validated` precondition was dropped 2026-06-25) shared by the report API and the Help
-readiness signal so the two can't disagree.
+`reports`; `readiness.py` defines the **report floor** (≥1 successful `analyze_*` **or**, since
+2026-07-14, `render_chart` — a chart-only session is substantive; the `problem_validated`
+precondition was dropped 2026-06-25) shared by the report API and the Help readiness signal so the
+two can't disagree. **2026-07-14:** the report embeds charts — `_collect_charts` copies
+`dataeyond.chart.v1` envelopes verbatim (INV-4) from `results_snapshot` into
+`AnalysisReport.charts`, rendered as ` ```plotly ` fenced blocks in the `## EDA` section
+(fence content = the **full v1 envelope**, pretty-printed — the shape the FE's fence hook parses,
+verified 2026-07-14).
 
 ### Observability — Langfuse
 The endpoint's `ChatHandler` runs with `enable_tracing=True`. One trace per request groups
@@ -337,6 +360,9 @@ Python is consumer-only). State **re-verified against the Go source 2026-06-29**
   `rooms`/`chat_messages`/`interview_*` tables to `zdeprecated_*`.
 - **`report_inputs`** (the slow-path structured output, formerly `analysis_records`) stays
   **Python-owned**; its finalized schema goes to Harry so the dedorch migration creates it post-cutover.
+  Same pattern for **`message_traceability`** (created manually 2026-07-06) and **`message_charts`**
+  (created manually 2026-07-13, DDL in `SPINE_V2_PLAN.md` §4.4; live e2e verified same day —
+  Harry's migration handoff for both is still the open item).
 - **Connection-string cutover DONE (2026-07-01).** Python's `postgres_connstring` now points at
   **dedorch** and reads the Go-migrated tables directly. Every ORM model Python reads (`analyses`,
   `analyses_messages`, `data_catalog`) has been reconciled to its dedorch shape.
@@ -374,6 +400,14 @@ records-based report; floor: ≥1 `analyze_*` success). Wiring Go → Python is
   `cryptography.fernet.InvalidToken` — whose `str()` is **empty**, so it logged as `error=""` and
   masqueraded as a DB-connection failure (the executor now logs `repr(e)` to expose it). Tell-apart:
   a valid-but-wrong key → `InvalidToken`; a malformed key → a non-empty `ValueError` at cipher build.
+- **Storage-provider parity with Go (gotcha, found 2026-07-13).** Go's data plane uploads tabular
+  parquet to **Supabase S3** and writes `location_ref: object_storage://…`; Python's
+  `TabularExecutor` picks its download backend from `settings.storage_provider`
+  (`azure_blob` | `supabase_s3`, blank → Azure legacy). If the `.env` still says `azure_blob`,
+  **every tabular `retrieve_data` fails with an Azure `BlobNotFound`** — and the never-throw path
+  degrades it into an honest "data not available" answer, so it masquerades as a data problem.
+  Tell-apart: `BlobNotFound` + `location_ref` starting `object_storage://` ⇒ env gap; set
+  `storage_provider=supabase_s3` + the five `supabase_s3_*` values (match Go's data plane).
 - **Never-throw seams** are pervasive (tool invoker, query service, executors, state/catalog reads,
   record persistence, report summary). Failures degrade into soft output rather than raising — good
   for UX, but they can mask real breakage (e.g. a missing analysis-scope catalog silently falling

SPINE_V2_PLAN.md

@@ -0,0 +1,315 @@
+# SPINE_V2_PLAN.md — Analysis Spine v2 + Visualization Tool
+
+**Status:** ✅ **APPROVED 2026-07-13** by Rifqi, with Sofia's sign-off (covers implementing the tool-layer
+slice — `ToolOutput.kind` + `render_chart` — in this repo). **Delta at approval:** W3 (activate deferred
+`analyze_*` tools) is **deferred to a later cycle — do not start**. W4 stays gated, W5 stays future-only.
+FE reminder (plotly rendering) owned by Rifqi, timing TBD.
+**Delta 2026-07-13 (pr/16, same-day build):** **W2 and W1 are DONE** — see DEV_PLAN §0.6 (V1–V7)
+for the task rows and evidence. Rifqi ran the §4.4 DDL against dedorch and the **live e2e gate
+passed in full** (real `POST /api/v2/chat/stream`: viz turn → `message_charts` row keyed by the
+`done` message_id → `GET /charts` 200 with a valid v1 envelope; chartless turn → 200 empty;
+injected `render_chart` failure → answer still streams with the data as a table, no row). W1 ✅
+(13 local tests, one per CK rule). Planner viz behavior verified in-process EN+ID incl. the
+viz-infeasible case (Example K was added when the first smoke force-mapped a stand-in column).
+Two env finds along the way: `eval.chat_sim` was missing from disk+git (restored by Rifqi — its
+hard-coded user/source ids are still stale) and the local `.env` lagged Go's Supabase-S3 data
+plane (`storage_provider` — fixed by Rifqi; REPO_STATUS §13 gotcha).
+**Owner:** Rifqi. Contributors named per workstream (Sofia = tool layer, Harry = Go/dedorch/contract, mentor = FE).
+**Companions:** [REPO_STATUS.md](REPO_STATUS.md) (built state) · [DEV_PLAN.md](DEV_PLAN.md) (sprint tracker) ·
+[API_CONTRACT_BE_PYTHON.md](API_CONTRACT_BE_PYTHON.md) (live contract). This file is the design + handoff
+source for the Spine-v2 work; when a workstream lands, DEV_PLAN gets the task rows and this doc gets dated deltas.
+
+Status legend (house): ⬜ not started · 🔄 in progress · ✅ done · ⛔ blocked · 🔎 verify · ⏸️ deferred.
+
+---
+
+## 0. Executive summary
+
+We evolve the slow path from a **one-shot pipeline** (plan → run → narrate) into a **staged workflow with one
+bounded self-correction point**, and we grow the tool registry into **four families on one spine**:
+`check_*`/`retrieve_*` (built) · `analyze_*` (built) · `render_*` (this plan) · `model_*` (future).
+
+Inspiration is CoDA (Google, ICLR 2026) and the Julius.ai comparison, with one governing rule: **CoDA's phases,
+not CoDA's agents.** CoDA pays ~15 LLM calls per output because every phase (viz mapping, design, codegen,
+debugging, visual evaluation) is its own LLM agent operating on generated code. We keep our declarative engine —
+the same phases exist here, but only planning is an LLM decision; mapping lives in the planner's existing single
+call, styling is a fixed preset, "generation" is a deterministic spec builder, and evaluation is deterministic
+code. Happy path stays **3 LLM calls per turn** (router + planner + assembler); the only growth is an
+**evidence-gated repair pass (max +1)** that is NOT built until telemetry justifies it.
+
+What ships, in order:
+
+- **S1a — Quality checkpoint** (0 LLM): after execution, before composition, deterministic code inspects the run
+  (failed branches, empty/degenerate results, later chart-spec sanity). Verdicts: proceed · honest degrade ·
+  log-a-repair-candidate. Converts today's silent degrade-and-continue into *explained* degradation and produces
+  the data for the S1b decision. No invariant touched.
+- **S2 — Visualization tool** (`render_chart`): planner-selected, Pattern A, emits **Plotly JSON** (locked
+  decision, DEV_PLAN §0 deferred row #26 — no matplotlib PNGs, no new dependency). Delivery mirrors traceability:
+  Python-owned `message_charts` store + `GET /api/v1/charts`, SSE stays text-only. Includes formalizing planner
+  **path recipes** (the few-shots become named workflows; viz = any recipe + a `render_chart` tail).
+- **S1b — Targeted repair (GATED)**: one bounded re-plan of only the failed subgraph, triggered by the
+  checkpoint. Relaxes INV-6 ("no mid-run LLM") → requires team sign-off *and* S1a telemetry showing it pays.
+- **S3 — `model_*` family (FUTURE, sketch only)**: regression/forecast/segmentation as fixed, audited compute
+  functions on the same spine. Named here so the architecture anticipates it; no build this cycle.
+
+What is deliberately **not changing**: no generated code ever executes; INV-4 (LLM never authors numbers); the
+five query-defense layers; never-throw seams; text-only SSE (`sources` stays `[]`); records-based versioned
+reports; Go's ownership of dedorch DDL and `analyses_messages`.
+
+---
+
+## 1. Target architecture
+
+```
+POST /api/v2/chat/stream  (structured_flow turn)
+   │
+   ├── Input guard ─► Router (1 LLM) ─► analysis-scoped catalog read (metadata only, PII-safe)
+   ▼
+ PLAN      Planner (1 LLM) — classifies the question into a RECIPE, instantiates a staged Task DAG
+   │  ▲       └ validator: 8 existing checks + path-shape checks, re-prompt ≤3 (all PRE-run)
+   │  │
+   │  └────────────── (S1b, GATED: targeted repair, ≤1 pass, failed subgraph only) ──┐
+   ▼                                                                                 │
+ EXECUTE   TaskRunner (0 LLM) — waves, ${t<id>} handoffs, degrade-and-continue       │
+   │          tool families (fixed registry, all never-throw):                       │
+   │          check_* / retrieve_*  ·  analyze_*  ·  render_* (S2)  ·  model_* (S3)  │
+   ▼                                                                                 │
+ CHECK     Quality checkpoint (S1a, 0 LLM, never-throw)                              │
+   │          ok → continue · repairable → repair request ────────────────────────────┘
+   │          unfixable → honest degrade (specific, not "couldn't compute")
+   ▼
+ COMPOSE   Assembler (1 LLM — narrative only, INV-4; structured fields copied by code)
+   ▼
+ DELIVER   SSE: sources[] → status* → chunk → done{message_id}
+              writes: report_inputs · message_charts (S2) · message_traceability
+              FE follow-ups on done: GET /api/v1/traceability · GET /api/v1/charts (S2)
+              later: POST /api/v1/tools/report (records-based, unchanged)
+```
+
+**Recipes (the "predetermined workflow" made explicit).** The planner few-shots already are path templates;
+S2 names them and the validator enforces their shape:
+
+| Recipe | Chain | Exists today as |
+|---|---|---|
+| R1 descriptive | `retrieve_data → analyze_descriptive` | few-shot A/B |
+| R2 aggregate / top-N | single grouped-IR `retrieve_data` (± `analyze_aggregate`) | Example G |
+| R3 trend | `retrieve_data → analyze_trend` | few-shot |
+| R4 correlation | `retrieve_data → analyze_correlation` | few-shot |
+| R5 two-metric merge | `retrieve_data ×2 → analyze_merge → …` | Example I |
+| R6 infeasible | no tasks + `infeasible_reason` → deterministic refusal | Example H |
+| viz tail (S2) | any of R1–R5 **+ `render_chart`** when the user explicitly asks to plot/visualize | NEW — Example J |
+
+CoDA-role mapping (why no new LLM agents): VizMapping → planner args in the existing call · Design Explorer →
+fixed style preset inside `render_chart` · Code Generator/Debug → deterministic spec builder (nothing to debug) ·
+Visual Evaluator → checkpoint spec checks.
+
+**LLM budget:** happy path 3 calls (unchanged) · S1b failure path max 4 · report unchanged at 1.
+
+---
+
+## 2. Workstreams & sequencing
+
+| # | Workstream | Owner | Status | Gate |
+|---|---|---|---|---|
+| W1 | S1a quality checkpoint (+ telemetry for the S1b decision) | Rifqi | ✅ 2026-07-13 | none — code-only, 0 LLM |
+| W2 | S2 `render_chart` tool + chart store + `GET /charts` + planner slice (recipes + Example J) | Rifqi + Sofia | ✅ 2026-07-13 (DDL run + live e2e ALL PASS) | Harry: migration handoff + `done.chart_count` still open · FE: Plotly render (Rifqi to remind) |
+| W3 | Activate the 4 deferred `analyze_*` tools (comparison, contribution, profile, segment) | Sofia + Rifqi | ⏸️ | **Deferred at approval (2026-07-13) — do not start until Rifqi re-opens** |
+| W4 | S1b targeted repair pass | Rifqi | ⏸️ | **team sign-off on INV-6 relaxation + S1a telemetry evidence** |
+| W5 | S3 `model_*` family | — | ⏸️ | future design review; sketch in §7 |
+
+Sequencing: **W2 leads** (visible product value), **W1 lands alongside or immediately after** (invisible to
+users, unblocks nothing, improves everything). W3 deferred at approval. W4 only with evidence. Parallel
+dispatch is safe: W1 and W2 are file-disjoint except `coordinator.py` (W1) — coordinate that one file.
+
+---
+
+## 3. W1 — S1a Quality checkpoint
+
+**What it is:** a deterministic, never-throw inspection of `RunState` between `TaskRunner.run` and
+`Assembler.assemble`. No LLM. No user-visible new surface — only better inputs to the Assembler and structured
+logs.
+
+**Files:**
+- NEW `src/agents/slow_path/checkpoint.py` — `assess(run_state, task_list) -> RunAssessment`.
+- `src/agents/slow_path/coordinator.py` — call site between runner and assembler; pass the assessment into
+  `assemble(...)`.
+- `src/agents/slow_path/prompt.py` — render the assessment as a short "execution assessment" block in the
+  assembler's human content so the narrative names *what specifically* failed/degraded instead of generic
+  caveats.
+- `src/agents/slow_path/schemas.py` — `RunAssessment` model (pydantic): per-task verdicts + overall verdict
+  (`ok | degraded | failed`) + `repair_candidates: list[RepairCandidate{task_id, reason}]`.
+
+**v1 checks (all deterministic):**
+- CK1 all tasks failed → overall `failed`; coordinator returns a deterministic honest-failure answer (mirrors
+  the existing infeasible path shape; record stays non-substantive so it can't hit the report floor).
+- CK2 `retrieve_data` returned 0 rows and a dependent consumed it → flag task + downstream.
+- CK3 table output truncated at the 10k cap → flag (answer must say "based on the first 10,000 rows").
+- CK4 `analyze_trend` produced a single bucket → flag (the pr/13 1970-bucket class).
+- CK5 all-null column consumed by an `analyze_*` task → flag.
+- CK6 (lands with W2) chart-spec sanity — see §4.6.
+- Every flag logs `repair_candidate` + reason via structlog → the S1b evidence base.
+
+**Explicit non-goals in S1a:** no re-planning, no LLM, no change to degrade-and-continue semantics, no change to
+never-throw seams. INV-6 untouched.
+
+---
+
+## 4. W2 — S2 Visualization tool (`render_chart`)
+
+### 4.1 Tool (Sofia's slice — tool layer is tool-team-owned)
+- NEW `src/tools/analytics/visualization.py` — `render_chart(df, chart_type, x, y, series=None, title=None)`.
+  Pure, deterministic, **no LLM, no new dependency**: builds a Plotly-conformant dict by hand (Plotly JSON is a
+  documented schema; the FE renders with plotly.js — Python never imports plotly). v1 chart types: `bar`,
+  `line`, `pie`, `scatter`. `pie` maps `x`→labels, `y`→values. House style preset (colors, axis config) is a
+  module constant — style is not a HAL decision.
+- `src/tools/contracts.py` — `ToolOutput.kind` Literal gains `"chart"` (one line; tool-team-owned file).
+- `src/tools/registry.py` — `ToolSpec(name="render_chart", category="analytics.visualization",
+  input_schema={"required": ["data", "chart_type", "x", "y"], "properties": {…, "series", "title"}},
+  output_kind="chart", description=visualization.DESCRIPTION)` in `ACTIVE_ANALYTICS_TOOLS`.
+- `src/tools/invoker.py` — `_DISPATCH["render_chart"] = (visualization.render_chart, "chart")`. Never-throw
+  comes free; a failed chart degrades the task, the answer still assembles.
+
+### 4.2 Chart-spec envelope (v1, FE contract)
+```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"}} }
+  }
+}
+```
+FE renders with `Plotly.newPlot(el, spec.plotly.data, spec.plotly.layout)`. The envelope (not raw plotly) is
+what `ToolOutput.value` carries and what the store persists.
+
+### 4.3 Planner slice (Rifqi)
+- `src/agents/planner/examples.py` — **Example J**: "plot revenue by region" → R2 retrieve → `render_chart`
+  with `data="${t1}"`.
+- `src/config/prompts/planner.md` — the recipe table (§1) + rule: chart **only when the user explicitly asks**
+  (plot/visualize/chart/grafik/buatkan diagram); never speculative; `render_chart` is always a tail, fed by a
+  table-kind upstream.
+- `src/agents/planner/validator.py` — path-shape check: `render_chart.data` must reference a task whose tool
+  yields `table` output.
+- `src/config/prompts/intent_router.md` — only if EN/ID viz phrasings misroute in the smoke; keep few-shots ↔
+  eval dataset mirrored if touched.
+- `src/agents/slow_path/prompt.py` — assembler renderer gets a `kind == "chart"` branch: one-line summary
+  (type, title, point count). Without this the full spec (x/y arrays) floods the assembler prompt
+  ([prompt.py:52](src/agents/slow_path/prompt.py:52) currently dumps `output.value` verbatim for non-table kinds).
+- `src/agents/report/generator.py` — no change needed: `_collect_evidence` copies table-kind only; charts cannot
+  corrupt the markdown report. Report embedding of charts stays **deferred** (standing decision).
+
+### 4.4 Chart store (Rifqi) — mirrors `message_traceability`
+- NEW `src/charts/store.py` — `ChartStore` Protocol + `PostgresChartStore` (never-throw `save`, `list` read) +
+  `NullChartStore`. ORM row in `src/db/postgres/models.py`.
+- Write site: `chat_handler._run_slow_path`, after the record persist — scan `record.results_snapshot` outputs
+  for `kind == "chart"`, save each stamped with `pad.message_id`, `analysis_id`, `user_id`, `record_id`.
+  Never-throw; a chart-persist failure must not break the answer.
+- **Zero DDL from Python** (§2.2). DDL below is run manually against dedorch now and handed to Harry for the
+  migration:
+```sql
+CREATE TABLE IF NOT EXISTS message_charts (
+    id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    message_id  text        NOT NULL,
+    analysis_id uuid        NOT NULL REFERENCES analyses(id),
+    user_id     text        NOT NULL,
+    record_id   text,
+    chart_type  text        NOT NULL,
+    title       text,
+    spec        jsonb       NOT NULL,   -- the dataeyond.chart.v1 envelope (§4.2)
+    created_at  timestamptz NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_message_charts_lookup
+    ON message_charts (analysis_id, message_id);
+```
+
+### 4.5 API surface + contract delta (Rifqi → Harry)
+- NEW `GET /api/v1/charts?analysis_id=&message_id=` (`src/api/v1/charts.py`, mounted in `main.py`) →
+  `{"count": n, "charts": [{chart_id, chart_type, title, spec, created_at}]}`; empty list is a valid 200 (FE:
+  no charts this turn). Same fetch-on-`done` pattern as traceability; the row is written before `done`, no
+  polling race.
+- `done` event: **proposed additive field** `chart_count` (or `chart_ids`) so the FE can skip the GET on
+  chartless turns. Harry's call; fallback = FE fetches unconditionally. SSE order and all existing fields
+  unchanged; `sources` stays `[]`.
+- `API_CONTRACT_BE_PYTHON.md` updated in the same change (new §charts + `done` note). FE ask (via Rifqi):
+  render `spec.plotly` with plotly.js under the assistant message; "chart iteration" v1 = a follow-up chat turn
+  (the planner re-emits `render_chart` with patched args — no new endpoint).
+
+### 4.6 Checkpoint spec checks (CK6, with W1)
+Empty series · `len(x) != len(y)` · bar/pie with > 30 / > 8 categories → flag (S1a: honest note; S1b later:
+re-pick args or fall back to table) · non-numeric y for bar/line/scatter.
+
+### 4.7 PII / tracing notes
+Chart specs contain the same result rows the user already receives in tables — no new exposure class. Langfuse:
+tool spans stay metadata-only; the traceability scratchpad records the `render_chart` call like any tool (verify
+the `TraceabilityToolInvoker` summary handles `kind="chart"` sanely — row_count is absent; use point count).
+
+---
+
+## 5. W3 — Activate deferred analytics tools (⏸️ DEFERRED 2026-07-13 — kept for the later cycle)
+
+Move `analyze_comparison`, `analyze_contribution`, `analyze_profile`, `analyze_segment` from
+`DEFERRED_ANALYTICS_TOOLS` to `ACTIVE_ANALYTICS_TOOLS` (`src/tools/registry.py` — compute fns and invoker
+mappings already exist). Per the registry's own note: each re-activated tool needs planner few-shot coverage
+(`examples.py`) — that is the real work. Recipes gain R7 comparison / R8 contribution / R9 segment rows.
+Prompt change → full §7B eval gauntlet. Closes the breadth gap vs Julius at near-zero engineering cost.
+
+---
+
+## 6. W4 — S1b targeted repair (GATED — do not build yet)
+
+Trigger: checkpoint returns repair candidates AND the team has approved the INV-6 relaxation AND S1a telemetry
+shows a meaningful hit-rate (proposal: revisit after 2 weeks of S1a logs). Design (for the future PR):
+`PlannerService.repair(task_list, run_state, reasons)` → patched subgraph for the failed task ids only; runner
+re-executes only those tasks + skipped dependents; **max 1 pass per turn**; repair prompt gets the failure
+evidence (error strings, row counts). Budget: +1 LLM call, failure paths only. If rejected: S1a's honest-degrade
+stands alone and this section moves to an ADR graveyard note — the checkpoint is still worth it.
+
+---
+
+## 7. W5 — S3 `model_*` family (future sketch, no build)
+
+Same spine, fourth family: fixed, audited compute functions (e.g. `model_regression`, `model_forecast`,
+`model_cluster` — sklearn/statsmodels behind ToolSpecs; **dependency additions need the §6.4 ask**). Recipe:
+`retrieve → prepare → model_* → evaluate` — CRISP-DM's modeling/evaluation stages become real. Outputs are
+`stats`/`table`/`series` + optional `render_chart` tails. Reports gain model sections then, not before.
+Explicitly NOT codegen — if open-ended modeling demand materializes, a sandboxed codegen tool is a separate
+§6.3 guardrail conversation.
+
+---
+
+## 8. Verification gates (per workstream — house §7)
+
+- Every W: `ruff check src/` clean · full local pytest with **exact counts** vs previous ·
+  `PYTHONPATH=. ./.venv/Scripts/python.exe -c "import main"` exits 0 · no never-throw seam pierced.
+- W1: unit tests for every CK rule (fixture RunStates); in-process slow-path run showing the assessment block in
+  the assembler input; no behavior change on a clean run.
+- W2 prompts: `eval.chat_sim` smoke ≥ last committed score (name both files); `eval.intent` only if router
+  prompt touched; new timestamped results committed, never overwriting.
+- W2 e2e: live in-process `structured_flow` turn with an explicit viz ask → chart row in `message_charts` →
+  `GET /charts` 200 with a valid v1 envelope; a chartless turn → 200 empty list; a failing `render_chart` →
+  answer still streams, no chart row.
+- W3: eval smokes for each new few-shot; EN+ID phrasing per eval conventions (§7F).
+- Contract changes: `API_CONTRACT_BE_PYTHON.md` same-change; REPO_STATUS/DEV_PLAN deltas dated (§7E).
+
+## 9. Decisions log & open questions
+
+Locked (approved 2026-07-13): Plotly JSON not PNG (re-affirms DEV_PLAN #26) · no per-phase LLM agents ·
+S1a before/with S2, S1b evidence-gated · ~~charts chat-only for now (report embedding deferred)~~
+RESOLVED 2026-07-14: the FE fenced-block hook exists (lead verified), so report embedding LANDED exactly
+as sketched — ` ```plotly ` fences in `rendered_markdown`'s EDA section, emitted by the generator from
+`results_snapshot` chart outputs, INV-4 copy-verbatim; `has_successful_analysis` now also counts a
+successful `render_chart`, so a chart-only session satisfies the report floor ·
+`render_chart` only on explicit user ask · ~~ToolOutput.kind + render_chart ownership split~~ RESOLVED:
+Sofia signed off, implemented in-repo by the dev session · **W3 deferred to a later cycle**.
+
+Open → owner: `done.chart_count` additive field yes/no → Harry · FE chart placement (under message) +
+plotly.js availability + report fenced-block hook → mentor/FE (Rifqi to remind) · INV-6 relaxation for W4 →
+team, after S1a telemetry · W3 re-open timing → Rifqi.
+
+## 10. Doc-sync checklist on landing (per workstream)
+
+REPO_STATUS §6/§9 (tool count, new checkpoint stage, charts surface) · DEV_PLAN new task rows under a dated
+sprint section · API_CONTRACT §charts + `done` delta · this file: status flips + dated delta banner.

eval/chat_sim/README.md

@@ -0,0 +1,55 @@
+# Chat simulator (end-to-end, in-process)
+
+Drives the **real** `ChatHandler.handle()` in-process to simulate a user chatting in
+one analysis session — from creation through a report — and prints **what each step
+does**: router decision, slow-path status, every tool call, every LLM call (output +
+tokens + latency), the streamed answer, and a final report built from the run's
+`report_inputs`.
+
+In-process (not via the HTTP server) on purpose: the SSE endpoint hides the internal
+LLM outputs (router/planner/assembler), so a script that consumes `/chat/stream` can't
+show them. This hooks the same tracer seam the handler threads its Langfuse callbacks
++ tool spans through (`ScriptTracer.active=True`) — **no source changes**.
+
+## Run
+
+Module mode (`-m`) so `src` imports resolve; needs a populated `.env` (Azure OpenAI +
+Postgres + Azure Blob for the Titanic Parquet). `ENABLE_SLOW_PATH` is forced on here.
+
+```bash
+uv run python -m eval.chat_sim.run_chat                  # scripted Titanic convo + report
+uv run python -m eval.chat_sim.run_chat --interactive    # type your own messages; 'report' / 'exit'
+uv run python -m eval.chat_sim.run_chat --max-turns 1 --no-report   # cheap smoke test
+uv run python -m eval.chat_sim.run_chat --no-bind        # planner sees the whole catalog (not just Titanic)
+uv run python -m eval.chat_sim.run_chat --plain          # no ANSI colors
+```
+
+(Or `./.venv/Scripts/python.exe -m eval.chat_sim.run_chat ...` on Windows.)
+
+## What it does each run
+
+1. **Creates a fresh analysis** (`AnalysisStateStore.create`) with an objective, and
+   **scopes it to the Titanic source** by seeding an analysis-scope `data_catalog` row
+   (the user catalog restricted to Titanic) so `structured_flow` is scoped to one
+   source — same as `/analysis/create` (in production Go materializes this row from
+   `analyses.data_bind`).
+2. **Runs each turn** through `handle()` and prints the router decision, slow-path
+   status pings, the tool table (kind / rows / latency / error), the LLM table
+   (in/out/total tokens + ms + a prompt/output snippet), and the answer + sources.
+3. **Generates a report** (mirrors `POST /report`: floor check → `ReportGenerator` →
+   `ReportStore`) from the `report_inputs` the structured turns persisted.
+
+## Notes
+
+- **Default user** is `4b5d1bac-…` whose playground catalog has the Titanic CSV
+  (tabular) + a "dummy" Postgres (schema). Override with `--user-id`.
+- **Writes to the DB the `.env` points at** — a fresh `analyses` row + `report_inputs`
+  + a `reports` row per run. Point `.env` at the playground DB. Each run is a new
+  `analysis_id` (printed at the end) so runs don't collide.
+- **"output: <no text content — structured / tool-call output>"** is expected for the
+  router, planner, and assembler — they use structured/function-call output, so the
+  LLM message has no plain text. Their result shows up in the ROUTER line, the plan,
+  and the streamed ANSWER respectively. The chatbot/help calls stream text, so their
+  output is shown (annotated `masked→cloud` where the cloud trace would redact it).
+- The first scripted question may route to `chat` rather than `help` — that's the live
+  router's real call, shown transparently.

eval/chat_sim/run_chat.py

@@ -0,0 +1,528 @@
+"""End-to-end chat simulator with full step transparency (in-process).
+
+Simulates a user chatting inside ONE analysis session, from creation onward, and
+prints what each step of the pipeline does:
+
+  - the ROUTER decision (intent / rewritten query / confidence)
+  - slow-path STATUS pings (Planning… / Running N steps…)
+  - every TOOL call the slow path makes (check_data / retrieve_data / analyze_* —
+    with result kind, row count, latency, error)
+  - every LLM call (router / planner / assembler / chatbot / help) with
+    input-token / output-token / latency, and a snippet of the raw model output
+  - the streamed ANSWER + its SOURCES
+  - a per-turn timing + token summary
+  - finally, a REPORT generated from the slow-path report_inputs the run produced
+
+It calls `ChatHandler.handle()` IN-PROCESS (no server) so it can see the internal
+LLM outputs the SSE endpoint hides. Transparency is captured by injecting a custom
+tracer (`ScriptTracer`) into the exact seam the handler already threads its Langfuse
+callbacks + tool spans through — no source changes.
+
+Run as a module from the repo root (so `src` imports resolve):
+
+    uv run python -m eval.chat_sim.run_chat                 # predefined Titanic convo + report
+    uv run python -m eval.chat_sim.run_chat --interactive   # you type the messages
+    uv run python -m eval.chat_sim.run_chat --no-report     # skip the report capstone
+    uv run python -m eval.chat_sim.run_chat --no-bind       # don't scope to Titanic (whole catalog)
+
+Needs a populated `.env` (Azure OpenAI + Postgres + Azure Blob for the Titanic
+Parquet). Writes to the DB the `.env` points at (analysis state + report_inputs +
+report) — point it at the playground DB. ENABLE_SLOW_PATH is forced on here.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+
+# --- Windows: psycopg3 async needs the selector loop (mirrors run.py). Set BEFORE
+# anything touches asyncio.
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+# Windows consoles default to cp1252 and choke on the box-drawing glyphs below.
+for _stream in (sys.stdout, sys.stderr):
+    try:
+        _stream.reconfigure(encoding="utf-8")  # type: ignore[union-attr]
+    except Exception:
+        pass
+
+from langchain_core.callbacks import BaseCallbackHandler  # noqa: E402
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage  # noqa: E402
+
+from src.agents.chat_handler import ChatHandler  # noqa: E402
+
+# This user's catalog (verified in the playground DB):
+#   tabular  source 9b565bc8-… = Titanic-Dataset.csv (891 rows)
+#   schema   source aaa0a4c6-… = "dummy" postgres (orders/customers/products/…)
+DEFAULT_USER_ID = "4b5d1bac-7211-490f-9a3d-66fed0168d5a"
+TITANIC_SOURCE_ID = "9b565bc8-ccc4-4d10-9382-0bad416a091b"
+TITANIC_NAME = "Titanic-Dataset.csv"
+
+OBJECTIVE = "Understand what drove passenger survival on the Titanic — by sex, class, and fare."
+
+# Default scripted conversation. Chosen to exercise every router intent against the
+# real Titanic columns (Survived, Sex, Pclass, Age, Fare, Embarked).
+DEFAULT_TURNS = [
+    "What can you help me do in this analysis?",                                   # -> help
+    "What data do I have available here?",                                         # -> check
+    "What was the overall passenger survival rate, and how did it differ "
+    "between male and female passengers?",                                         # -> structured_flow
+    "Did higher passenger class (Pclass) come with a higher average fare and "
+    "a higher survival rate?",                                                     # -> structured_flow
+]
+
+# ANSI (Windows Terminal / VS Code support it). Disable with --plain.
+_C = {
+    "h": "\033[1;36m", "u": "\033[1;33m", "ai": "\033[1;32m",
+    "dim": "\033[2m", "warn": "\033[1;31m", "r": "\033[0m",
+}
+
+
+def c(key: str, text: str) -> str:
+    return f"{_C.get(key, '')}{text}{_C['r']}" if _C.get("_on", True) else text
+
+
+# ───────────────────────── transparency capture ──────────────────────────────
+
+
+@dataclass
+class LlmCall:
+    idx: int
+    ms: int | None
+    tin: int
+    tout: int
+    ttot: int
+    prompt_preview: str
+    output_preview: str
+    masked: bool
+
+
+@dataclass
+class ToolCall:
+    tool: str
+    arg_keys: list[str]
+    kind: str | None
+    rows: int | None
+    error: str | None
+    ms: int
+
+
+@dataclass
+class Sink:
+    """Per-turn collector shared by all StepLoggers + spans of that turn."""
+    llm: list[LlmCall] = field(default_factory=list)
+    tools: list[ToolCall] = field(default_factory=list)
+
+
+def _usage(response: Any) -> tuple[int, int, int]:
+    """Sum token usage off an LLMResult (usage_metadata, legacy fallback)."""
+    tin = tout = ttot = 0
+    for gens in getattr(response, "generations", []) or []:
+        for g in gens:
+            msg = getattr(g, "message", None)
+            um = getattr(msg, "usage_metadata", None) if msg else None
+            if um:
+                tin += um.get("input_tokens", 0)
+                tout += um.get("output_tokens", 0)
+                ttot += um.get("total_tokens", 0)
+    if ttot == 0 and getattr(response, "llm_output", None):
+        u = response.llm_output.get("token_usage") or {}
+        tin += u.get("prompt_tokens", 0)
+        tout += u.get("completion_tokens", 0)
+        ttot += u.get("total_tokens", 0)
+    return tin, tout, ttot
+
+
+def _out_text(response: Any) -> str:
+    try:
+        gens = response.generations
+        g = gens[0][0]
+        msg = getattr(g, "message", None)
+        return (getattr(msg, "content", None) or getattr(g, "text", "") or "").strip()
+    except Exception:
+        return ""
+
+
+def _preview(text: str, n: int = 240) -> str:
+    text = " ".join(str(text).split())
+    return text if len(text) <= n else text[: n - 1] + "…"
+
+
+class StepLogger(BaseCallbackHandler):
+    """One per `tracer.callbacks()` call; all share the turn's Sink.
+
+    Captures each LLM call's latency + tokens + a snippet of prompt/output. Matches
+    start->end by run_id so concurrent/streamed calls don't cross wires.
+    """
+
+    def __init__(self, sink: Sink, masked: bool = False) -> None:
+        self.sink = sink
+        self.masked = masked
+        self._t0: dict[Any, float] = {}
+        self._prompt: dict[Any, str] = {}
+
+    def on_chat_model_start(self, serialized, messages, *, run_id=None, **kw):  # type: ignore[override]
+        self._t0[run_id] = time.perf_counter()
+        try:
+            flat = [m for grp in messages for m in grp]
+            self._prompt[run_id] = _preview(
+                next((getattr(m, "content", "") for m in flat
+                      if m.__class__.__name__.startswith("System")), ""
+                     ) or (flat[-1].content if flat else ""), 120
+            )
+        except Exception:
+            self._prompt[run_id] = ""
+
+    def on_llm_start(self, serialized, prompts, *, run_id=None, **kw):  # type: ignore[override]
+        self._t0[run_id] = time.perf_counter()
+        self._prompt[run_id] = _preview(prompts[0] if prompts else "", 120)
+
+    def on_llm_end(self, response, *, run_id=None, **kw):  # type: ignore[override]
+        t0 = self._t0.pop(run_id, None)
+        ms = round((time.perf_counter() - t0) * 1000) if t0 else None
+        tin, tout, ttot = _usage(response)
+        self.sink.llm.append(LlmCall(
+            idx=len(self.sink.llm) + 1, ms=ms, tin=tin, tout=tout, ttot=ttot,
+            prompt_preview=self._prompt.pop(run_id, ""),
+            output_preview=_preview(_out_text(response)),
+            masked=self.masked,
+        ))
+
+
+class ScriptSpan:
+    """Mirrors tracing._ToolSpan: a metadata-only span around one slow-path tool call."""
+
+    def __init__(self, sink: Sink, tool: str, args: dict) -> None:
+        self.sink = sink
+        self.tool = tool
+        self.args = args
+        self.t0 = time.perf_counter()
+
+    def end(self, out: Any) -> None:
+        kind = getattr(out, "kind", None)
+        rows = len(getattr(out, "rows", None) or []) if kind == "table" else None
+        err = getattr(out, "error", None)
+        self.sink.tools.append(ToolCall(
+            tool=self.tool,
+            arg_keys=sorted(self.args) if isinstance(self.args, dict) else [],
+            kind=kind, rows=rows,
+            error=_preview(err, 160) if err else None,
+            ms=round((time.perf_counter() - self.t0) * 1000),
+        ))
+
+
+class ScriptTracer:
+    """Drop-in for RequestTracer/NullTracer. active=True so the slow path wraps its
+    ToolInvoker in TracingToolInvoker and routes tool spans here."""
+
+    active = True
+
+    def __init__(self, sink: Sink) -> None:
+        self.sink = sink
+
+    def callbacks(self, *, masked: bool = False) -> list:
+        return [StepLogger(self.sink, masked)]
+
+    def tool_span(self, tool: str, args: dict) -> Any:
+        return ScriptSpan(self.sink, tool, args)
+
+    def end(self, *, output: Any = None) -> None:
+        return None
+
+
+class InstrumentedChatHandler(ChatHandler):
+    """ChatHandler that emits our ScriptTracer instead of Langfuse/Null, so every
+    LLM + tool step of a turn lands in `self.sink`."""
+
+    def __init__(self, *a, **k) -> None:
+        super().__init__(*a, **k)
+        self.sink = Sink()
+
+    def _make_tracer(self, user_id: str, question: str) -> Any:  # type: ignore[override]
+        return ScriptTracer(self.sink)
+
+
+# ───────────────────────────── pretty printing ───────────────────────────────
+
+
+def banner(text: str, ch: str = "═") -> None:
+    print(f"\n{c('h', ch * 78)}\n{c('h', text)}\n{c('h', ch * 78)}")
+
+
+def _llm_labels(intent: str | None, n: int) -> list[str]:
+    """Best-effort name per LLM call, by the path's known call order."""
+    seq = {
+        "structured_flow": ["router", "planner", "assembler"],
+        "help": ["router", "help"],
+        "unstructured_flow": ["router", "chatbot"],
+        "chat": ["router", "chatbot"],
+        "check": ["router"],
+    }.get(intent or "", ["router"])
+    out = []
+    for i in range(n):
+        if i < len(seq) - 1:
+            out.append(seq[i])
+        elif i == n - 1:
+            out.append(seq[-1])           # last call = final author
+        else:
+            out.append(f"{seq[1] if len(seq) > 1 else 'llm'}·retry")
+    return out
+
+
+def print_turn_steps(sink: Sink, intent: str | None, total_ms: int) -> None:
+    if sink.tools:
+        print(c("dim", "\n  tool calls (slow path):"))
+        for t in sink.tools:
+            tag = c("warn", "ERROR") if t.error else (t.kind or "ok")
+            extra = f" rows={t.rows}" if t.rows is not None else ""
+            print(f"    • {t.tool:<18} {tag:<7}{extra:<10} {t.ms:>5}ms"
+                  f"  args={t.arg_keys}")
+            if t.error:
+                print(c("warn", f"        ↳ {t.error}"))
+
+    if sink.llm:
+        labels = _llm_labels(intent, len(sink.llm))
+        print(c("dim", "\n  llm calls (output / tokens / latency):"))
+        print(c("dim", f"    {'#':<2} {'step':<14} {'in':>6} {'out':>6} {'tot':>6} {'ms':>6}"))
+        for call, label in zip(sink.llm, labels):
+            ms = f"{call.ms}" if call.ms is not None else "?"
+            print(f"    {call.idx:<2} {label:<14} {call.tin:>6} {call.tout:>6} "
+                  f"{call.ttot:>6} {ms:>6}")
+            print(c("dim", f"        prompt: {call.prompt_preview}"))
+            # Local tool over your own data → show output regardless of the masked
+            # flag (masking only matters for Langfuse Cloud). Note when it's a
+            # cloud-masked call or has no text (structured / tool-call output).
+            out = call.output_preview or "<no text content — structured / tool-call output>"
+            tag = " (masked→cloud)" if call.masked else ""
+            print(c("dim", f"        output{tag}: {out}"))
+
+    tin = sum(c_.tin for c_ in sink.llm)
+    tout = sum(c_.tout for c_ in sink.llm)
+    print(c("dim", f"\n  ── turn: {total_ms}ms · {len(sink.llm)} llm call(s) · "
+                   f"{len(sink.tools)} tool call(s) · {tin}+{tout} tokens"))
+
+
+# ───────────────────────────── setup / turns ─────────────────────────────────
+
+
+async def setup_analysis(user_id: str, bind_titanic: bool) -> str:
+    """Create a fresh analysis session (state row) + optionally bind it to Titanic.
+
+    Mirrors what `/analysis/create` does: a state row carrying the goal, plus an
+    analysis-scope `data_catalog` row (B) restricting the analysis to one source, so
+    structured_flow is scoped deterministically. Returns the analysis_id (== room_id).
+    """
+    from src.agents.state_store import AnalysisStateStore
+
+    analysis_id = str(uuid.uuid4())
+    await AnalysisStateStore().create(
+        analysis_id=analysis_id,
+        user_id=user_id,
+        analysis_title="Titanic survival analysis (sim)",
+        objective=OBJECTIVE,
+    )
+    print(f"  created analysis {c('h', analysis_id)}")
+    print(f"  objective: {OBJECTIVE}")
+
+    if bind_titanic:
+        try:
+            from datetime import UTC, datetime
+
+            from src.catalog.models import Catalog as CatalogModel
+            from src.catalog.store import CatalogStore
+            from src.db.postgres.connection import AsyncSessionLocal
+            from src.db.postgres.models import Catalog as CatalogRow
+
+            # Scope structured_flow by seeding the analysis-scope catalog (B): the
+            # user's catalog restricted to Titanic. structured_flow reads this row via
+            # CatalogStore.get_by_analysis (the data_sources binding table was removed;
+            # in production Go materializes B from analyses.data_bind).
+            user_cat = await CatalogStore().get(user_id)
+            titanic = [
+                s for s in (user_cat.sources if user_cat else [])
+                if s.source_id == TITANIC_SOURCE_ID
+            ]
+            if not titanic:
+                print(c("warn", f"  Titanic source {TITANIC_SOURCE_ID} not in user "
+                                "catalog — running unscoped (whole catalog)"))
+            else:
+                scoped = CatalogModel(
+                    user_id=user_id, generated_at=datetime.now(UTC), sources=titanic,
+                )
+                async with AsyncSessionLocal() as s:
+                    s.add(CatalogRow(
+                        scope_type="analysis", user_id=user_id, analysis_id=analysis_id,
+                        catalog_payload=scoped.model_dump(mode="json"),
+                    ))
+                    await s.commit()
+                print(f"  bound source: {TITANIC_NAME} ({TITANIC_SOURCE_ID})  "
+                      f"{c('dim', '→ structured_flow scoped to Titanic (analysis catalog)')}")
+        except Exception as e:  # noqa: BLE001 — fail-open to whole catalog
+            print(c("warn", f"  binding skipped ({type(e).__name__}: {e}) — "
+                            f"fail-open to whole catalog"))
+    else:
+        print(c("dim", "  no binding → structured_flow sees the whole catalog"))
+    return analysis_id
+
+
+async def run_turn(
+    handler: InstrumentedChatHandler,
+    user_id: str,
+    analysis_id: str,
+    message: str,
+    history: list[BaseMessage],
+) -> None:
+    handler.sink = Sink()
+    banner(f"USER  ▸  {message}", "─")
+
+    answer = ""
+    sources: list[dict] = []
+    intent: str | None = None
+    t0 = time.perf_counter()
+
+    async for ev in handler.handle(message, user_id, history, analysis_id=analysis_id):
+        kind, data = ev["event"], ev["data"]
+        if kind == "intent":
+            try:
+                d = json.loads(data)
+                intent = d.get("intent")
+                print(f"  {c('h', 'ROUTER')} → intent={c('h', intent)}  "
+                      f"confidence={d.get('confidence')}")
+                rq = d.get("rewritten_query")
+                if rq and rq != message:
+                    print(c("dim", f"         rewritten: {rq}"))
+            except Exception:
+                pass
+        elif kind == "status":
+            print(c("dim", f"  · {data}"))
+        elif kind == "sources":
+            try:
+                sources = json.loads(data) or []
+            except Exception:
+                sources = []
+        elif kind == "chunk":
+            answer += data
+        elif kind == "error":
+            print(c("warn", f"  ERROR: {data}"))
+
+    total_ms = round((time.perf_counter() - t0) * 1000)
+    print(f"\n  {c('ai', 'ANSWER')} ▾")
+    for line in (answer or "(empty)").splitlines() or ["(empty)"]:
+        print(f"    {line}")
+    if sources:
+        print(c("dim", f"\n  sources ({len(sources)}): "
+                       + ", ".join(s.get("filename") or s.get("document_id", "?")
+                                   for s in sources)))
+    print_turn_steps(handler.sink, intent, total_ms)
+
+    history.append(HumanMessage(content=message))
+    history.append(AIMessage(content=answer))
+
+
+async def generate_report(user_id: str, analysis_id: str) -> None:
+    """Mirror POST /report: floor check → ReportGenerator → ReportStore → print."""
+    banner("REPORT  ▸  generating from accumulated report_inputs")
+    from src.agents.gate import stub_analysis_state
+    from src.agents.report.generator import ReportGenerator
+    from src.agents.report.readiness import report_floor
+    from src.agents.report.schemas import ProblemStatement
+    from src.agents.report.store import ReportStore
+    from src.agents.state_store import AnalysisStateStore
+
+    state = await AnalysisStateStore().get(analysis_id)
+    missing, _ = await report_floor(
+        analysis_id, state or stub_analysis_state(problem_validated=False)
+    )
+    if missing:
+        print(c("warn", f"  floor not met (409 in the API): {', '.join(missing)}"))
+        print(c("dim", "  → need ≥1 successful slow-path analysis first "
+                       "(did the structured turns run analyze_* tools?)"))
+        return
+
+    objective = (getattr(state, "objective", "") or
+                 getattr(state, "problem_statement", "") or "")
+    ps = ProblemStatement(
+        objective=objective,
+        business_questions=list(getattr(state, "business_questions", []) or []),
+    )
+    t0 = time.perf_counter()
+    report = await ReportGenerator().generate(
+        analysis_id, user_id, problem_statement=ps, user_name=None
+    )
+    saved = await ReportStore().save(report)
+    print(f"  generated v{saved.version} in {round((time.perf_counter()-t0)*1000)}ms "
+          f"· report_id={saved.report_id} · built from {len(saved.record_ids)} record(s)\n")
+    print(c("dim", "  ── rendered markdown ──"))
+    for line in saved.rendered_markdown.splitlines():
+        print(f"  {line}")
+
+
+# ──────────────────────────────── main ───────────────────────────────────────
+
+
+async def amain(args: argparse.Namespace) -> None:
+    if args.plain:
+        _C["_on"] = False
+
+    banner("DATA EYOND — end-to-end chat simulator (in-process)")
+    print(f"  user_id: {args.user_id}")
+    print(f"  slow_path: ON   tracing→terminal: ON   db: (from .env)")
+
+    handler = InstrumentedChatHandler(
+        enable_tracing=False, enable_gate=False
+    )
+    analysis_id = await setup_analysis(args.user_id, bind_titanic=not args.no_bind)
+    history: list[BaseMessage] = []
+
+    if args.interactive:
+        print(c("dim", "\n  interactive mode — type a message, 'report' to generate, "
+                       "'exit' to quit.\n"))
+        loop = asyncio.get_event_loop()
+        while True:
+            try:
+                msg = (await loop.run_in_executor(None, input, "you ▸ ")).strip()
+            except (EOFError, KeyboardInterrupt):
+                break
+            if not msg:
+                continue
+            if msg.lower() in {"exit", "quit"}:
+                break
+            if msg.lower() == "report":
+                await generate_report(args.user_id, analysis_id)
+                continue
+            await run_turn(handler, args.user_id, analysis_id, msg, history)
+    else:
+        turns = DEFAULT_TURNS[: args.max_turns] if args.max_turns else DEFAULT_TURNS
+        for msg in turns:
+            await run_turn(handler, args.user_id, analysis_id, msg, history)
+
+    if not args.no_report:
+        await generate_report(args.user_id, analysis_id)
+
+    banner("DONE")
+    print(f"  analysis_id (== room_id): {analysis_id}")
+    print(c("dim", "  state, report_inputs, and report were written to the .env DB."))
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="End-to-end chat simulator with step transparency")
+    p.add_argument("--user-id", default=DEFAULT_USER_ID)
+    p.add_argument("--interactive", action="store_true", help="type messages yourself")
+    p.add_argument("--no-report", action="store_true", help="skip the report capstone")
+    p.add_argument("--no-bind", action="store_true",
+                   help="don't scope to Titanic (planner sees the whole catalog)")
+    p.add_argument("--max-turns", type=int, default=0,
+                   help="run only the first N scripted turns (cheap smoke test)")
+    p.add_argument("--plain", action="store_true", help="disable ANSI colors")
+    asyncio.run(amain(p.parse_args()))
+
+
+if __name__ == "__main__":
+    main()

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/handlers/check.py

@@ -278,6 +278,18 @@ async def _fetch_db_tables(
     }
 
 
+def _name_in_message(name: str, message: str) -> bool:
+    """Whole-word, case-insensitive match of `name` in `message`.
+
+    `\\b` treats `_` as a word char, so "sales" won't hit "sales_archive" and the
+    short table name "pa" won't hit "pareto" — the same rule used for source names,
+    reused here for table names.
+    """
+    return bool(name) and (
+        re.search(rf"\b{re.escape(name)}\b", message, re.IGNORECASE) is not None
+    )
+
+
 def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
     """All source_ids whose name appears as a whole word in the message.
 
@@ -285,9 +297,9 @@ def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
     the tool needs exact `source_id`s. We resolve them against the inventory
     rows (kind="table", columns include "source_id" + "name") instead of an LLM
     — a cheap match against catalog metadata already in hand. Whole-word match
-    (`\\b`) avoids nuisance hits ("orders" inside "reorders") and treats `_` as
-    part of the word, so "sales" won't pick up "sales_archive". Multiple named
-    sources all match, so the caller can show each schema.
+    avoids nuisance hits ("orders" inside "reorders") and treats `_` as part of
+    the word, so "sales" won't pick up "sales_archive". Multiple named sources all
+    match, so the caller can show each schema.
     """
     if inventory.kind != "table" or not inventory.rows:
         return []
@@ -301,7 +313,7 @@ def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
     matched: list[str] = []
     for row in inventory.rows:
         name = str(row[name_idx])
-        if name and re.search(rf"\b{re.escape(name)}\b", message, re.IGNORECASE):
+        if _name_in_message(name, message):
             matched.append(str(row[id_idx]))
     return matched
 
@@ -370,7 +382,6 @@ _SCHEMA_STR = {
         "yes": "Yes", "no": "—",
         "db_tables": "{name} has {n} tables:",
         "db_item": "- {table} ({cols} columns{rows})",
-        "db_hint": "Name a table to see its columns.",
     },
     "Indonesian": {
         "lead": "Berikut kolom dan tipe data di {n} sumber yang kamu punya:",
@@ -383,7 +394,6 @@ _SCHEMA_STR = {
         "yes": "Ya", "no": "—",
         "db_tables": "{name} punya {n} tabel:",
         "db_item": "- {table} ({cols} kolom{rows})",
-        "db_hint": "Sebut nama tabelnya untuk lihat kolomnya.",
     },
 }
 
@@ -542,7 +552,6 @@ def _render_schema_source(out: ToolOutput, reply_language: str) -> str:
             lines.append(
                 sc["db_item"].format(table=tname, cols=len(tcols), rows=_rows_suffix(rc))
             )
-        lines.append(sc["db_hint"])
         return "\n".join(lines)
 
     parts: list[str] = []
@@ -580,6 +589,41 @@ def _render_schemas(schemas: list[ToolOutput], reply_language: str) -> str:
     return lead + "\n\n" + "\n\n".join(blocks)
 
 
+def _filter_to_named_tables(
+    schemas: list[ToolOutput], message: str
+) -> list[ToolOutput] | None:
+    """Narrow drilled schemas to the specific table(s) named in the message.
+
+    Lets "what columns are in badactor" answer with badactor's columns instead of
+    the whole-DB table list. When the message names table(s) present in a source,
+    that source is narrowed to just those tables; sources with none of the named
+    tables are dropped, so a table-specific question doesn't dump every other
+    source's schema. Returns None when the message names no table at all, so the
+    caller keeps the default whole-source view. Table names are matched
+    whole-word (same rule as source names) against each drilled schema's rows.
+    """
+    narrowed: list[ToolOutput] = []
+    any_named = False
+    for out in schemas:
+        if out.kind != "table":
+            continue
+        cols = out.columns or []
+        if "table_name" not in cols:
+            continue
+        tn_idx = cols.index("table_name")
+        keep = {
+            str(r[tn_idx])
+            for r in (out.rows or [])
+            if _name_in_message(str(r[tn_idx]), message)
+        }
+        if not keep:
+            continue
+        any_named = True
+        rows = [r for r in (out.rows or []) if str(r[tn_idx]) in keep]
+        narrowed.append(out.model_copy(update={"rows": rows}))
+    return narrowed if any_named else None
+
+
 async def run_check(
     message: str, invoker: ToolInvoker, reply_language: str = "English"
 ) -> str:
@@ -618,7 +662,12 @@ async def run_check(
         schemas = await asyncio.gather(
             *(invoker.invoke("check_data", {"source_id": sid}) for sid in ids)
         )
-        return _render_schemas(schemas, reply_language) or _no_match
+        # Table-level drill: if the message names specific table(s) — e.g. "what
+        # columns are in badactor" — narrow to those so we answer with that table's
+        # columns instead of a whole-DB table list. Falls back to the full view when
+        # no table is named.
+        narrowed = _filter_to_named_tables(schemas, message)
+        return _render_schemas(narrowed or schemas, reply_language) or _no_match
 
     if intent == "data":
         inventory = await invoker.invoke("check_data", {})

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

@@ -11,6 +11,8 @@ rejects structurally-invalid plans (duplicate ids, dangling edges, cycles).
 
 from __future__ import annotations
 
+from collections.abc import Iterator
+
 from pydantic import ValidationError
 
 from src.middlewares.logging import get_logger
@@ -281,6 +283,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:
@@ -322,14 +342,33 @@ def _is_placeholder(value: str) -> bool:
 
 
 def _placeholder_refs(task) -> set[str]:
+    """Task ids referenced by any placeholder in a task's args — including ones
+    NESTED inside a retrieve_data IR (e.g. a '${t2.customer_id}' value-handoff
+    filter). A '${t<id>.<col>}' ref contributes its TASK id ('t2'); the column
+    suffix is stripped so the DAG check still enforces the dependency (task ids
+    contain no '.')."""
     refs: set[str] = set()
     for call in task.tool_calls:
-        for value in call.args.values():
-            if isinstance(value, str):
-                refs.update(PLACEHOLDER_RE.findall(value))
+        for value in _iter_arg_strings(call.args):
+            for ref in PLACEHOLDER_RE.findall(value):
+                refs.add(ref.split(".", 1)[0])
     return refs
 
 
+def _iter_arg_strings(value: object) -> Iterator[str]:
+    """Yield every string reachable in a tool-arg value, recursing into dicts and
+    lists so placeholders nested inside an IR are seen — not just top-level string
+    args (which is all `data`/`data_right` Pattern-A handoffs used to be)."""
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for v in value.values():
+            yield from _iter_arg_strings(v)
+    elif isinstance(value, list):
+        for v in value:
+            yield from _iter_arg_strings(v)
+
+
 def _is_checkable(text: str) -> bool:
     low = text.lower()
     return any(tok in low for tok in _CHECKABLE_TOKENS)

src/agents/refusals.py

@@ -76,6 +76,33 @@ _DATA_GAP = {
 }
 
 
+# Run failure: the plan was feasible but EVERY execution step failed (S1a CK1,
+# SPINE_V2_PLAN §3). Deterministic like the data-gap answer — a run that just
+# failed is not re-asked to prose up its own failure. Same reply_language keying.
+_RUN_FAILURE = {
+    "en": (
+        "I ran the analysis, but every step failed, so I don't have a result I can "
+        "stand behind. {reason}You can try rephrasing the question, or check that "
+        "the connected data source is reachable."
+    ),
+    "id": (
+        "Saya sudah menjalankan analisisnya, tetapi semua langkah gagal, jadi belum "
+        "ada hasil yang bisa saya laporkan. {reason}Coba ubah pertanyaannya, atau "
+        "periksa apakah sumber data yang terhubung dapat diakses."
+    ),
+}
+
+
+def run_failure_message(reason: str | None, reply_language: str | None = None) -> str:
+    """Answer for an all-steps-failed run (S1a CK1). `reason` is the most specific
+    failure detail available (may be None); `reply_language` as in data_gap_message."""
+    detail = (reason or "").strip()
+    if detail and not detail.endswith((".", "!", "?")):
+        detail += "."
+    lang = "id" if reply_language == "Indonesian" else "en"
+    return _RUN_FAILURE[lang].format(reason=f"{detail} " if detail else "")
+
+
 def data_gap_message(reason: str | None, reply_language: str | None = None) -> str:
     """Answer for an infeasible analysis: the bound sources lack the asked-for data.
 

src/agents/report/generator.py

@@ -13,6 +13,7 @@ Chain construction mirrors `agents/slow_path/assembler.py`.
 
 from __future__ import annotations
 
+import json
 import re
 from datetime import UTC, datetime
 from pathlib import Path
@@ -201,6 +202,33 @@ def _collect_evidence(records: list[AnalysisRecord]) -> dict[str, list[EvidenceT
     return out
 
 
+_CHARTS_MAX_PER_RECORD = 3  # mirrors _EVIDENCE_MAX_TABLES
+
+
+def _collect_charts(records: list[AnalysisRecord]) -> dict[str, list[dict]]:
+    """Copy chart envelopes out of each record's `results_snapshot` (INV-4).
+
+    `render_chart` outputs only — the full `dataeyond.chart.v1` envelope, verbatim
+    (the snapshot trimmer never touches chart-kind outputs, so the spec is whole).
+    Rendered as ```plotly fenced blocks in the markdown EDA section.
+    """
+    out: dict[str, list[dict]] = {}
+    for rec in records:
+        charts: list[dict] = []
+        for result in rec.results_snapshot.values():
+            for output in result.outputs:
+                if len(charts) >= _CHARTS_MAX_PER_RECORD:
+                    break
+                if output.kind != "chart" or not isinstance(output.value, dict):
+                    continue
+                if not isinstance(output.value.get("plotly"), dict):
+                    continue
+                charts.append(output.value)
+        if charts:
+            out[rec.record_id] = charts
+    return out
+
+
 def _unresolved_note(rec: AnalysisRecord) -> AttributedNote:
     # Goal + the record's own first caveat as the "why" — both Assembler-authored,
     # nothing new is synthesized here.
@@ -446,8 +474,26 @@ def _render_markdown(report: AnalysisReport) -> str:
             blocks.append("\n".join(block))
         parts.append("\n\n".join(blocks))
 
-    # ## EDA — reserved for future exploratory visuals; charts wrapped as MDX
-    # components will render here. Emitted only when it has content; omitted today.
+    # ## EDA — charts (W2, 2026-07-14). Each chart a `render_chart` task produced
+    # is emitted as a ```plotly fenced block the FE's fence hook renders with
+    # plotly.js. Fence content is the FULL dataeyond.chart.v1 envelope, verbatim
+    # (INV-4) and pretty-printed — the shape the FE hook parses (verified against
+    # the FE 2026-07-14; it reads spec.plotly internally). Section omitted when no
+    # record produced a chart.
+    if report.charts:
+        lines = ["## EDA"]
+        ordered_rids = [rid for rid in report.record_ids if rid in report.charts]
+        ordered_rids += [rid for rid in report.charts if rid not in ordered_rids]
+        for rid in ordered_rids:
+            for envelope in report.charts[rid]:
+                caption = envelope.get("title") or report.record_goals.get(rid) or "Chart"
+                lines.append(f"**{_mdx_escape(str(caption))}**")
+                lines.append("")
+                lines.append("```plotly")
+                lines.append(json.dumps(envelope, ensure_ascii=False, indent=2))
+                lines.append("```")
+                lines.append("")
+        parts.append("\n".join(lines).rstrip())
 
     if report.data_sources:
         lines = ["## Data Sources", "| source | type | detail |", "|---|---|---|"]
@@ -607,6 +653,7 @@ class ReportGenerator:
             unresolved=[_unresolved_note(r) for r in unresolved_records],
             excluded=[_excluded_note(r) for r in excluded],
             evidence_tables=_collect_evidence(records),
+            charts=_collect_charts(records),
             data_sources=data_sources,
             method_steps=method_steps,
         )

src/agents/report/readiness.py

@@ -72,15 +72,18 @@ def _is_newer(a: datetime, b: datetime) -> bool:
 
 
 def has_successful_analysis(record) -> bool:
-    """True if the record has at least one *analysis* task that succeeded.
+    """True if the record has at least one *result-producing* task that succeeded.
 
     A failed run still writes findings (narrating the failure) and its data-access
     tasks (check_/retrieve_) succeed, so we can't key on findings or on "any task
-    succeeded". An analysis tool (analyze_*) completing is the real "we produced a
-    result" signal.
+    succeeded". A completed analysis tool (analyze_*) — or, since W2 charts
+    (2026-07-14), a completed `render_chart`, whose viz-tail upstream necessarily
+    computed the numbers being charted — is the real "we produced a result"
+    signal. A chart-only session therefore satisfies the report floor.
     """
     return any(
-        t.status == "success" and any(tool.startswith("analyze") for tool in t.tools_used)
+        t.status == "success"
+        and any(tool.startswith("analyze") or tool == "render_chart" for tool in t.tools_used)
         for t in record.tasks_run
     )
 

src/agents/report/schemas.py

@@ -146,6 +146,10 @@ class AnalysisReport(BaseModel):
     excluded: list[AttributedNote] = Field(default_factory=list)
     # record_id -> small result tables copied from that record's results_snapshot.
     evidence_tables: dict[str, list[EvidenceTable]] = Field(default_factory=dict)
+    # record_id -> dataeyond.chart.v1 envelopes, copied verbatim from that record's
+    # results_snapshot (W2 charts, 2026-07-14). The markdown renders each as a
+    # ```plotly fenced block in the EDA section; the FE's fence hook draws it.
+    charts: dict[str, list[dict]] = Field(default_factory=dict)
     data_sources: list[DataSourceRef] = Field(default_factory=list)
     method_steps: list[TaskSummary] = Field(default_factory=list)  # carries `stage`
     rendered_markdown: str = ""

src/agents/slow_path/assembler.py

@@ -33,6 +33,7 @@ from .schemas import (
     AnalysisRecord,
     AssembledOutput,
     AssemblerNarrative,
+    RunAssessment,
     RunState,
     TaskResult,
     TaskSummary,
@@ -96,6 +97,7 @@ class Assembler:
         question: str | None = None,
         reply_language: str | None = None,
         callbacks: list | None = None,
+        assessment: RunAssessment | None = None,
     ) -> AssembledOutput:
         chain = self._ensure_chain()
         # `reply_language` is detected upstream from the ORIGINAL user message. Fall back
@@ -103,7 +105,9 @@ class Assembler:
         # rewritten_query, which may be normalized to English, so the caller should pass it.
         if reply_language is None:
             reply_language = detect_reply_language([], message=question)
-        human_content = build_assembler_prompt(run_state, context, question, reply_language)
+        human_content = build_assembler_prompt(
+            run_state, context, question, reply_language, assessment=assessment
+        )
         try:
             if callbacks:
                 narrative: AssemblerNarrative = await chain.ainvoke(

src/agents/slow_path/checkpoint.py

@@ -0,0 +1,233 @@
+"""Quality checkpoint (S1a) — deterministic run inspection (SPINE_V2_PLAN §3, W1).
+
+Sits between `TaskRunner.run` and `Assembler.assemble` in the coordinator. 0 LLM,
+never-throw: `assess` inspects the completed `RunState` against the plan and turns
+today's *silent* degrade-and-continue into an *explained* one — the Assembler gets
+told what specifically failed/degraded, and every flag is structlog'd as a
+`repair_candidate` so the S1b (targeted repair, gated) decision has an evidence
+base. INV-6 untouched: nothing here re-plans or calls a model.
+
+v1 checks:
+  CK1  all tasks failed            -> overall "failed" (coordinator answers honestly, no LLM)
+  CK2  empty retrieve consumed     -> flag the retrieve + its transitive dependents
+  CK3  table truncated at the cap  -> flag (answer must scope itself to the first 10k rows)
+  CK4  trend collapsed to 1 bucket -> flag (the pr/13 1970-bucket class)
+  CK5  all-null column consumed    -> flag the consuming analyze_* task
+  CK6  chart-spec sanity (§4.6)    -> flag empty/mismatched/overcrowded/non-numeric charts
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict, deque
+from typing import Any
+
+from src.middlewares.logging import get_logger
+
+from ..planner.schemas import PLACEHOLDER_RE, TaskList
+from .schemas import RepairCandidate, RunAssessment, RunState, TaskAssessment
+
+logger = get_logger("slow_path_checkpoint")
+
+_TABLE_ROW_CAP = 10_000  # the query pipeline's LIMIT defense cap (CK3)
+_BAR_CATEGORY_CAP = 30   # §4.6 — a bar chart beyond this is unreadable
+_PIE_CATEGORY_CAP = 8    # §4.6 — a pie chart beyond this is unreadable
+
+
+def assess(run_state: RunState, task_list: TaskList) -> RunAssessment:
+    """Inspect a completed run. NEVER raises — on an internal slip the checkpoint
+    degrades to today's behavior (an all-ok pass-through), never blocks the answer."""
+    try:
+        return _assess(run_state, task_list)
+    except Exception as exc:  # noqa: BLE001 — never-throw seam (house pattern)
+        logger.error("checkpoint failed", error=repr(exc))
+        return RunAssessment(overall="ok")
+
+
+def _assess(run_state: RunState, task_list: TaskList) -> RunAssessment:
+    results = run_state.results
+    notes: dict[str, list[str]] = {tid: [] for tid in results}
+    candidates: list[RepairCandidate] = []
+
+    def flag(task_id: str, check: str, reason: str, *, repairable: bool) -> None:
+        if task_id not in notes:
+            return
+        notes[task_id].append(reason)
+        # Telemetry for the S1b decision (SPINE_V2_PLAN §3): every flag is logged,
+        # repairable or not — the hit-rate decides whether a repair pass pays.
+        logger.info("repair_candidate", task_id=task_id, check=check, reason=reason)
+        if repairable:
+            candidates.append(RepairCandidate(task_id=task_id, reason=reason))
+
+    # CK1 — everything failed: no partial result worth prose. The coordinator
+    # returns a deterministic honest-failure answer (0 LLM), mirroring the
+    # infeasible path; the record stays non-substantive.
+    if results and all(r.status == "failure" for r in results.values()):
+        logger.info(
+            "repair_candidate",
+            task_id="*",
+            check="CK1",
+            reason="all tasks failed",
+        )
+        return RunAssessment(
+            overall="failed",
+            tasks=[
+                TaskAssessment(
+                    task_id=tid, verdict="failed", notes=[r.error] if r.error else []
+                )
+                for tid, r in results.items()
+            ],
+        )
+
+    dependents = _transitive_dependents(task_list)
+
+    for tid, result in results.items():
+        for out in result.outputs:
+            # CK2 — an empty retrieval, with downstream consumers flagged too.
+            if out.tool == "retrieve_data" and out.kind == "table" and not (out.rows or []):
+                flag(tid, "CK2", "retrieve_data returned 0 rows (the filters "
+                     "matched nothing)", repairable=True)
+                for dep in sorted(dependents.get(tid, ())):
+                    flag(dep, "CK2", f"consumed the empty result of {tid}",
+                         repairable=False)
+            # CK3 — the retrieval cap was hit; findings only cover the cap window.
+            elif out.kind == "table" and out.rows is not None and len(out.rows) >= _TABLE_ROW_CAP:
+                flag(tid, "CK3", f"table hit the {_TABLE_ROW_CAP:,}-row retrieval "
+                     "cap — findings describe only the first "
+                     f"{_TABLE_ROW_CAP:,} rows", repairable=False)
+
+            # CK4 — a single trend bucket (e.g. every date parsed into one period).
+            if out.tool == "analyze_trend" and out.kind == "series" and isinstance(out.value, dict):
+                points = out.value.get("points")
+                if isinstance(points, list) and len(points) == 1:
+                    freq = out.value.get("freq", "period")
+                    flag(tid, "CK4", f"trend collapsed into a single {freq} bucket "
+                         "(1 point) — no movement can be described",
+                         repairable=True)
+
+            # CK6 — chart-spec sanity (§4.6; lands with W2's render_chart).
+            if out.kind == "chart" and isinstance(out.value, dict):
+                for issue in _chart_issues(out.value):
+                    flag(tid, "CK6", issue, repairable=True)
+
+    # CK5 — an analyze_* consumed a table whose column(s) are entirely null.
+    # Consumption is read from the PLAN (the `data`/`data_right` placeholders);
+    # the runner resolves the same references at execution time.
+    for task in task_list.tasks:
+        for call in task.tool_calls:
+            if not call.tool.startswith("analyze_"):
+                continue
+            for arg_name in ("data", "data_right"):
+                ref = _placeholder_ref(call.args.get(arg_name))
+                table = _last_table_output(results.get(ref)) if ref else None
+                if table is None:
+                    continue
+                null_cols = _all_null_columns(table)
+                if null_cols:
+                    flag(task.id, "CK5",
+                         f"input column(s) {null_cols} from {ref} are entirely "
+                         "null — results based on them are meaningless",
+                         repairable=True)
+
+    assessments: list[TaskAssessment] = []
+    for tid, result in results.items():
+        if result.status == "failure":
+            verdict = "failed"
+        elif notes[tid] or result.status == "partial":
+            verdict = "degraded"
+        else:
+            verdict = "ok"
+        assessments.append(TaskAssessment(task_id=tid, verdict=verdict, notes=notes[tid]))
+
+    overall = "degraded" if any(a.verdict != "ok" for a in assessments) else "ok"
+    return RunAssessment(overall=overall, tasks=assessments, repair_candidates=candidates)
+
+
+def _transitive_dependents(task_list: TaskList) -> dict[str, set[str]]:
+    """dependents[id] = every task downstream of id via depends_on edges."""
+    direct: dict[str, set[str]] = defaultdict(set)
+    for task in task_list.tasks:
+        for dep in task.depends_on:
+            direct[dep].add(task.id)
+    out: dict[str, set[str]] = {}
+    for tid in list(direct):
+        seen: set[str] = set()
+        queue = deque(direct[tid])
+        while queue:
+            node = queue.popleft()
+            if node in seen:
+                continue
+            seen.add(node)
+            queue.extend(direct.get(node, ()))
+        out[tid] = seen
+    return out
+
+
+def _placeholder_ref(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    match = PLACEHOLDER_RE.fullmatch(value.strip())
+    return match.group(1) if match else None
+
+
+def _last_table_output(result: Any) -> Any | None:
+    """The referenced task's representative output (matches TaskRunner's
+    `outputs[-1]` resolution), if it is a non-empty table."""
+    if result is None or not result.outputs:
+        return None
+    out = result.outputs[-1]
+    if out.kind != "table" or not out.columns or not out.rows:
+        return None
+    return out
+
+
+def _all_null_columns(table: Any) -> list[str]:
+    cols: list[str] = []
+    for i, name in enumerate(table.columns):
+        if all(row[i] is None for row in table.rows if i < len(row)):
+            cols.append(name)
+    return cols
+
+
+def _chart_issues(envelope: dict[str, Any]) -> list[str]:
+    """§4.6 spec checks on a dataeyond.chart.v1 envelope. Purely structural —
+    the spec builder is deterministic, so an issue here means the *plan args*
+    picked a bad shape (wrong column, oversized category set), not a tool bug."""
+    issues: list[str] = []
+    chart_type = envelope.get("chart_type", "chart")
+    plotly = envelope.get("plotly")
+    traces = plotly.get("data") if isinstance(plotly, dict) else None
+    if not isinstance(traces, list) or not traces:
+        return [f"{chart_type} chart has no data traces"]
+
+    for trace in traces:
+        if not isinstance(trace, dict):
+            continue
+        name = trace.get("name")
+        label = f"series {name!r}" if name else "a series"
+        xs = trace.get("x") if trace.get("x") is not None else trace.get("labels")
+        ys = trace.get("y") if trace.get("y") is not None else trace.get("values")
+        xs = xs if isinstance(xs, list) else []
+        ys = ys if isinstance(ys, list) else []
+        if not xs or not ys or all(v is None for v in ys):
+            issues.append(f"{chart_type} chart: {label} is empty")
+            continue
+        if len(xs) != len(ys):
+            issues.append(
+                f"{chart_type} chart: {label} has mismatched lengths "
+                f"(x={len(xs)}, y={len(ys)})"
+            )
+        if chart_type == "bar" and len(xs) > _BAR_CATEGORY_CAP:
+            issues.append(
+                f"bar chart: {label} has {len(xs)} categories "
+                f"(> {_BAR_CATEGORY_CAP}) — unreadable; aggregate or top-N first"
+            )
+        if chart_type == "pie" and len(xs) > _PIE_CATEGORY_CAP:
+            issues.append(
+                f"pie chart has {len(xs)} slices (> {_PIE_CATEGORY_CAP}) — "
+                "unreadable; group the tail into 'other' or use a bar chart"
+            )
+        if chart_type in ("bar", "line", "scatter") and any(
+            v is not None and not isinstance(v, int | float) for v in ys
+        ):
+            issues.append(f"{chart_type} chart: {label} has non-numeric y values")
+    return issues

src/agents/slow_path/coordinator.py

@@ -18,9 +18,10 @@ from ..planner.contracts import BusinessContext, ToolRegistry
 from ..planner.inputs import Constraints
 from ..planner.schemas import TaskList
 from ..planner.service import PlannerService
-from ..refusals import data_gap_message
+from ..refusals import data_gap_message, run_failure_message
 from .assembler import Assembler
-from .schemas import AnalysisRecord, AssembledOutput
+from .checkpoint import assess
+from .schemas import AnalysisRecord, AssembledOutput, RunState
 from .task_runner import TaskRunner
 
 
@@ -70,14 +71,52 @@ class SlowPathCoordinator:
         run_state = await self._task_runner.run(
             task_list, business_context_id=context.project_id
         )
+        # S1a quality checkpoint (SPINE_V2_PLAN §3): deterministic, 0 LLM,
+        # never-throw. `failed` short-circuits to an honest deterministic answer
+        # (no Assembler call); `degraded` flags flow into the Assembler's input so
+        # the narrative names what specifically was affected; `ok` changes nothing.
+        assessment = assess(run_state, task_list)
+        if assessment.overall == "failed":
+            return _run_failure_output(task_list, run_state, context, reply_language)
         if progress:
             await progress("Composing the answer…")
         asm_kw = {"callbacks": assembler_callbacks} if assembler_callbacks else {}
         return await self._assembler.assemble(
-            run_state, context, question=query, reply_language=reply_language, **asm_kw
+            run_state, context, question=query, reply_language=reply_language,
+            assessment=assessment, **asm_kw
         )
 
 
+def _run_failure_output(
+    task_list: TaskList,
+    run_state: RunState,
+    context: BusinessContext,
+    reply_language: str | None,
+) -> AssembledOutput:
+    """Every task failed (S1a CK1): answer honestly and deterministically — same
+    shape as the infeasible path. The record carries no tasks_run/results, so it
+    is non-substantive: it can never satisfy the report floor."""
+    errors = [r.error for r in run_state.results.values() if r.error]
+    # The first error is usually the root cause (later tasks fail on the missing
+    # upstream); one specific reason beats a list of cascaded ones.
+    reason = errors[0] if errors else None
+    return AssembledOutput(
+        chat_answer=run_failure_message(reason, reply_language),
+        analysis_record=AnalysisRecord(
+            goal_restated=task_list.goal_restated,
+            findings=[],
+            caveats=errors or ["all analysis steps failed"],
+            data_used=[],
+            open_questions=list(task_list.open_questions),
+            tasks_run=[],
+            results_snapshot={},
+            plan_id=task_list.plan_id,
+            business_context_id=context.project_id,
+            created_at=datetime.now(UTC),
+        ),
+    )
+
+
 def _infeasible_output(
     task_list: TaskList, context: BusinessContext, reply_language: str | None
 ) -> AssembledOutput:

src/agents/slow_path/prompt.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from ..planner.contracts import BusinessContext, ToolOutput
 from ..planner.prompt import render_business_context
-from .schemas import RunState, TaskResult
+from .schemas import RunAssessment, RunState, TaskResult
 
 _MAX_ROWS = 20
 
@@ -48,20 +48,56 @@ def _render_output(output: ToolOutput) -> str:
         )
         more = "" if len(rows) <= _MAX_ROWS else f" … (+{len(rows) - _MAX_ROWS} more rows)"
         return f"({output.tool}) table [{header}]: {preview}{more}"
+    if output.kind == "chart" and isinstance(output.value, dict):
+        # W2 (SPINE_V2_PLAN §4.3): one-line summary only — the raw spec's x/y
+        # arrays would flood the prompt. The chart itself reaches the user via
+        # GET /api/v1/charts; the narrative only refers to it.
+        chart_type = output.value.get("chart_type", "chart")
+        title = output.value.get("title") or ""
+        plotly = output.value.get("plotly")
+        traces = plotly.get("data") if isinstance(plotly, dict) else None
+        n_traces = len(traces) if isinstance(traces, list) else 0
+        return (
+            f"({output.tool}) chart: {chart_type} \"{title}\" ({n_traces} series) — "
+            "shown to the user alongside this answer; refer to it, do not restate "
+            "its data points"
+        )
     meta = f" meta={output.meta}" if output.meta else ""
     return f"({output.tool}) {output.kind}: {output.value}{meta}"
 
 
+def render_assessment(assessment: RunAssessment) -> str | None:
+    """The S1a checkpoint's flags as a short block (SPINE_V2_PLAN §3) — only tasks
+    that carry specific notes; a clean run renders nothing (no behavior change)."""
+    flagged = [t for t in assessment.tasks if t.notes]
+    if not flagged:
+        return None
+    lines = [f"Overall execution: {assessment.overall}."]
+    for t in flagged:
+        lines.extend(f"- [{t.verdict}] {t.task_id}: {note}" for note in t.notes)
+    lines.append(
+        "Name these limitations specifically in the answer — say what was affected "
+        "and how it scopes the findings. Do not present partial results as complete, "
+        "and never fall back to a generic \"couldn't compute\"."
+    )
+    return "\n".join(lines)
+
+
 def build_assembler_prompt(
     run_state: RunState,
     context: BusinessContext,
     question: str | None = None,
     reply_language: str | None = None,
+    assessment: RunAssessment | None = None,
 ) -> str:
     sections = [
         f"# Business context\n\n{render_business_context(context)}",
         f"# Analysis results\n\n{render_run_state(run_state)}",
     ]
+    if assessment is not None:
+        block = render_assessment(assessment)
+        if block:
+            sections.append(f"# Execution assessment\n\n{block}")
     if question:
         sections.append(f"# Original question\n\n{question}")
     if reply_language:

src/agents/slow_path/schemas.py

@@ -53,6 +53,41 @@ class RunState(BaseModel):
     open_questions: list[str] = Field(default_factory=list)
 
 
+# --------------------------------------------------------------------------- #
+# Quality checkpoint (S1a) — checkpoint.assess() -> Assembler (SPINE_V2_PLAN §3)
+# --------------------------------------------------------------------------- #
+
+AssessmentVerdict = Literal["ok", "degraded", "failed"]
+
+
+class RepairCandidate(BaseModel):
+    """One checkpoint flag a future targeted repair (S1b, gated) could act on.
+
+    Until S1b is approved this is telemetry: every candidate is also structlog'd
+    so the team can measure whether a repair pass would pay for itself.
+    """
+
+    task_id: str
+    reason: str
+
+
+class TaskAssessment(BaseModel):
+    task_id: str
+    verdict: AssessmentVerdict
+    notes: list[str] = Field(default_factory=list)  # specific, human-readable flags
+
+
+class RunAssessment(BaseModel):
+    """Deterministic post-execution verdict (0 LLM). `overall` drives the
+    coordinator: `failed` -> honest deterministic failure answer (no Assembler
+    call); `degraded` -> the flags are rendered into the Assembler's input so the
+    narrative names *what specifically* was affected; `ok` -> no behavior change."""
+
+    overall: AssessmentVerdict
+    tasks: list[TaskAssessment] = Field(default_factory=list)
+    repair_candidates: list[RepairCandidate] = Field(default_factory=list)
+
+
 # --------------------------------------------------------------------------- #
 # Assembled output (Assembler -> Orchestrator / memory) — §8.3
 # --------------------------------------------------------------------------- #

src/agents/slow_path/task_runner.py

@@ -126,16 +126,28 @@ class TaskRunner:
 
     @staticmethod
     def _resolve_value(value: Any, results: dict[str, TaskResult]) -> Any:
-        # A data arg is exactly a "${t<id>}" placeholder (Pattern A); resolve it to
-        # the referenced task's representative output (its last ToolOutput).
-        # Materializing that envelope into a DataFrame is the invoker's job.
+        # Resolve "${t<id>}" placeholders from upstream results, recursing into
+        # containers so a placeholder NESTED inside a tool arg is resolved too —
+        # not just a whole-value "${t2}" data arg. This is what lets a
+        # "${t2.customer_id}" filter value inside a retrieve_data IR become a real
+        # list (value-handoff): "cari-lalu-lookup" / anti-join across two steps.
         if isinstance(value, str):
             match = PLACEHOLDER_RE.fullmatch(value.strip())
             if match:
-                upstream = results.get(match.group(1))
+                # group(1) is "t<id>" (whole output → Pattern A, materialized to a
+                # DataFrame by the invoker) or "t<id>.<column>" (value-handoff →
+                # the list of that column's upstream values).
+                task_id, _, column = match.group(1).partition(".")
+                upstream = results.get(task_id)
                 if upstream is None or not upstream.outputs:
-                    return None
-                return upstream.outputs[-1]
+                    return [] if column else None
+                out = upstream.outputs[-1]
+                return _column_values(out, column) if column else out
+            return value
+        if isinstance(value, dict):
+            return {k: TaskRunner._resolve_value(v, results) for k, v in value.items()}
+        if isinstance(value, list):
+            return [TaskRunner._resolve_value(v, results) for v in value]
         return value
 
     def _validate_args(self, tool: str, resolved: dict[str, Any]) -> str | None:
@@ -156,6 +168,27 @@ class TaskRunner:
             return ToolOutput(tool=tool, kind="error", error=f"invoker raised: {exc}")
 
 
+def _column_values(out: ToolOutput, column: str) -> list[Any]:
+    """Ordered, de-duplicated values of `column` from a table ToolOutput.
+
+    Backs value-handoff: a "${t<id>.<column>}" reference resolves to the SET of
+    that column's upstream values so an `in`/`not_in` filter can look up / anti-
+    join by a prior step's result (e.g. "customers who never ordered" =
+    `CustomerID not_in ${orders.customer_id}`). Empty list when the column is
+    absent or the output isn't a table — an `in` then matches nothing and a
+    `not_in` matches everything, the correct set semantics for an empty reference.
+    """
+    columns = out.columns or []
+    if column not in columns:
+        return []
+    idx = columns.index(column)
+    seen: dict[Any, None] = {}
+    for row in out.rows or []:
+        if idx < len(row):
+            seen.setdefault(row[idx], None)
+    return list(seen)
+
+
 def _label(outputs: list[ToolOutput]) -> TaskStatus:
     if not outputs:
         return "failure"

src/api/v1/charts.py

@@ -0,0 +1,72 @@
+"""Charts endpoint — user-facing chart retrieval for one turn (S2 visualization,
+SPINE_V2_PLAN §4.4/§4.5).
+
+`GET /api/v1/charts?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` with the `message_id` from that event.
+Lookup is by `message_id` alone (Python-minted UUID4 — lead decision 2026-07-13).
+
+Every response is HTTP 200 with an explicit `status` marker (lead ask 2026-07-13):
+  - `success`   — ≥1 chart; render them.
+  - `empty`     — the turn completed but produced no charts (the common case).
+  - `not_found` — no completed turn is known for this message_id (mistyped/stale
+                  id, or an error turn — those never write a row). Distinguished
+                  from `empty` via the turn's traceability row.
+
+No auth — Go fronts Python.
+"""
+
+from typing import Literal
+
+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):
+    status: Literal["success", "empty", "not_found"]
+    message: str
+    count: int
+    charts: list[ChartRecord]
+
+
+@router.get("/charts", response_model=ChartsResponse)
+@log_execution(logger)
+async def get_charts(
+    message_id: str = Query(
+        ..., description="Assistant turn id, taken from the `done` SSE event"
+    ),
+) -> ChartsResponse:
+    """Fetch every chart for one turn. Always 200 — the `status` field carries the
+    outcome, so the FE can call this unconditionally on every `done`."""
+    charts = await _store.list_for_message(message_id)
+    if charts:
+        return ChartsResponse(
+            status="success",
+            message=f"{len(charts)} chart(s) for this message.",
+            count=len(charts),
+            charts=charts,
+        )
+    if await _store.turn_exists(message_id):
+        return ChartsResponse(
+            status="empty",
+            message="This message completed without producing charts.",
+            count=0,
+            charts=[],
+        )
+    return ChartsResponse(
+        status="not_found",
+        message="No completed turn is known for this message_id.",
+        count=0,
+        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,172 @@
+"""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, MessageTraceabilityRow
+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);
+    the lookup is by `message_id` alone (Python-minted UUID4, globally unique —
+    lead decision 2026-07-13). `turn_exists` says whether the turn flushed a
+    traceability row, so the endpoint can tell "no charts" from "unknown id".
+    """
+
+    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, message_id: str) -> list[ChartRecord]: ...
+
+    async def turn_exists(self, message_id: str) -> bool: ...
+
+
+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, message_id: str) -> list[ChartRecord]:
+        return []
+
+    async def turn_exists(self, message_id: str) -> bool:
+        return False
+
+
+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, message_id: str) -> list[ChartRecord]:
+        # message_id-only lookup (lead decision 2026-07-13). NOTE for the Harry
+        # migration: the manual DDL's composite index (analysis_id, message_id)
+        # does not serve this predicate — an additive index on (message_id) is
+        # part of the handoff.
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(MessageChartRow)
+                .where(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
+        ]
+
+    async def turn_exists(self, message_id: str) -> bool:
+        """True iff the turn flushed its traceability row (written before `done`).
+
+        Lets the endpoint tri-state a zero-chart GET: `empty` (completed turn, no
+        charts — the common case) vs `not_found` (unknown/mistyped id, or an error
+        turn, which never writes traceability). PK lookup — cheap.
+        """
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(MessageTraceabilityRow.message_id).where(
+                    MessageTraceabilityRow.message_id == message_id
+                )
+            )
+            return result.scalar_one_or_none() is not None

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`.
@@ -74,6 +91,24 @@ only a `TaskList` object that conforms to the provided schema.
   `orders.total_amount`); if they genuinely aren't linked, say the data isn't
   connected rather than guessing. Prefer an existing measure column over
   recomputing. Joins are database-only — not available for tabular/file sources.
+- **Look up or exclude by a prior step's values (value-handoff).** To filter one
+  task by the values another task produced — WITHOUT a join — put
+  `"${t<id>.<alias>}"` as a filter `value` and use op `in` (keep matches) or
+  `not_in` (exclude them). `<alias>` is the SELECT alias of the referenced task's
+  column; it resolves to the LIST of that column's values. `value_type` stays the
+  element type (e.g. `"string"`), never `"list"`. Set `depends_on`. This works on
+  tabular AND database sources, and is the ONLY way to relate two tables on a
+  tabular/file source (no joins there). Two patterns:
+  - **Attribute of the top/bottom entity** ("name of the customer with the most
+    orders"): t1 = group by the id + aggregate + `order_by` + `limit: 1` to get the
+    winning id; t2 = `retrieve_data` selecting the attribute (e.g. the name) where
+    `{"op": "in", "value": "${t1.<id_alias>}", "value_type": "string", ...}`.
+  - **"Never / not among" (anti-join)** ("customers who never ordered"): t1 =
+    select the id column from the transaction table (e.g. all `Orders.CustomerID`);
+    t2 = select from the entity table where
+    `{"op": "not_in", "value": "${t1.<id_alias>}", "value_type": "string", ...}`.
+    Do NOT use `analyze_merge` (a left merge) for this — it fans out to one row per
+    match and does not isolate the non-matching rows, so the result is wrong.
 - **Two measures per entity ("which X has both the worst A and the biggest B").**
   Compute each measure in its OWN grouped `retrieve_data` task (one aggregate per
   entity each), then align them with `analyze_merge`:
@@ -82,6 +117,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/query/compiler/pandas.py

@@ -136,6 +136,19 @@ def _like_to_regex(pattern: str) -> str:
     return "".join(parts)
 
 
+def _as_list(val: Any) -> list[Any]:
+    """Coerce a scalar to a single-element list for `isin`.
+
+    `in`/`not_in` values are normally lists, but a value-handoff (or a planner
+    single-value filter) can arrive as a scalar; pandas `.isin()` rejects a bare
+    scalar ("only list-like objects are allowed to be passed to isin()"). Wrapping
+    keeps both shapes working.
+    """
+    if isinstance(val, list | tuple | set):
+        return list(val)
+    return [val]
+
+
 def _apply_filters(
     df: pd.DataFrame,
     filters: list[FilterClause],
@@ -161,9 +174,9 @@ def _apply_filters(
         elif op == ">=":
             mask &= series >= val
         elif op == "in":
-            mask &= series.isin(val)
+            mask &= series.isin(_as_list(val))
         elif op == "not_in":
-            mask &= ~series.isin(val)
+            mask &= ~series.isin(_as_list(val))
         elif op == "is_null":
             mask &= series.isna()
         elif op == "is_not_null":

src/query/executor/tabular.py

@@ -193,12 +193,22 @@ def _resolve_blob_name(source: Source, table: Table) -> str:
     return parquet_blob_name(user_id, document_id, sheet_name)
 
 
+def _render_value(v: object, cap: int = 5) -> str:
+    """Render a filter value for the human-readable query string, truncating a
+    long list so a value-handoff `in`/`not_in` set doesn't dump thousands of ids
+    into the logs AND the user-facing traceability record (KM-691)."""
+    if isinstance(v, list | tuple) and len(v) > cap:
+        head = ", ".join(repr(x) for x in v[:cap])
+        return f"[{head}, … +{len(v) - cap} more]"
+    return repr(v)
+
+
 def _render_query(ir: QueryIR, cols_by_id: dict) -> str:
     from ..ir.models import AggSelect, ColumnSelect
     parts = ["df"]
     if ir.filters:
         conds = " & ".join(
-            f'(df["{cols_by_id[f.column_id].name}"] {f.op} {f.value!r})'
+            f'(df["{cols_by_id[f.column_id].name}"] {f.op} {_render_value(f.value)})'
             for f in ir.filters
         )
         parts.append(f"[{conds}]")

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("_", " ")