/feat traceability

API_CONTRACT_BE_PYTHON.md

@@ -1,6 +1,6 @@
 # Backend Agentic Service API Contract
 
-This document describes the Python agentic backend used by the frontend for AI chat, help/report tools, and observability data shown alongside chat answers.
+This document describes the Python agentic backend used by the frontend for AI chat, help/report tools, and traceability data shown alongside chat answers.
 
 Base path examples use relative URLs. Configure the frontend with the deployed Python service base URL.
 
@@ -11,14 +11,14 @@ The Python backend owns the generative AI interaction surface:
 1. Stream chat answers from the AI agent.
 2. Execute tool-style actions for help and report generation.
 3. Return report versions and report details.
-4. Return observability/provenance for a completed assistant answer.
+4. Return traceability/provenance for a completed assistant answer.
 
 The frontend uses this service during the analysis conversation flow:
 
 1. User sends a chat message.
 2. Frontend calls `POST /api/v2/chat/stream` and renders the streamed answer.
 3. When the stream emits `done`, frontend stores or reads the returned `message_id`.
-4. Frontend calls `GET /api/v1/observability` for planning, tool calls, and source provenance.
+4. Frontend calls `GET /api/v1/traceability` for planning, tool calls, and source provenance.
 5. Frontend calls `/api/v1/tools/help` for guided help and `/api/v1/tools/report` for report generation.
 
 ## Endpoint Summary
@@ -31,7 +31,7 @@ The frontend uses this service during the analysis conversation flow:
 | `POST` | `/api/v1/tools/report` | Generate and persist a new report version. |
 | `GET` | `/api/v1/tools/report/{analysis_id}` | List report versions for an analysis. |
 | `GET` | `/api/v1/tools/report/{analysis_id}/{version}` | Retrieve one report version. |
-| `GET` | `/api/v1/observability` | Retrieve provenance for one assistant answer. |
+| `GET` | `/api/v1/traceability` | Retrieve provenance for one assistant answer. |
 
 ## Common Concepts
 
@@ -39,7 +39,7 @@ The frontend uses this service during the analysis conversation flow:
 
 - `user_id`: user identifier passed by the frontend.
 - `analysis_id`: analysis conversation identifier.
-- `message_id`: assistant answer identifier used to correlate chat streaming with observability.
+- `message_id`: assistant answer identifier used to correlate chat streaming with traceability.
 
 ### Server-Sent Events
 
@@ -51,13 +51,13 @@ Common event types:
 
 | Event | Data | Meaning |
 | --- | --- | --- |
-| `sources` | JSON array | Sources available early in the stream. May be empty. |
+| `sources` | JSON array | Always `[]` — sources moved to `GET /api/v1/traceability` (KM-691). Event kept for backward-compat; read `sources[]` from the traceability call. |
 | `status` | text | Optional progress update for slower paths. |
 | `chunk` | text | Answer text fragment. Concatenate chunks in order. |
 | `done` | JSON object | Terminal success event. Includes `message_id`. |
 | `error` | text | Terminal error event. Stream stops after this. |
 
-The stream carries answer text only. Planning, tool call details, and full provenance are fetched from `GET /api/v1/observability` after the stream is done.
+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.
 
 ## Chat
 
@@ -82,7 +82,7 @@ Fields:
 | --- | --- | --- |
 | `user_id` | Yes | User identifier. |
 | `analysis_id` | Yes | Analysis conversation identifier. |
-| `message_id` | No | Assistant answer id for observability correlation. If omitted, Python returns one in `done`. |
+| `message_id` | No | Assistant answer id for traceability correlation. If omitted, Python returns one in `done`. |
 | `message` | Yes | User message text. |
 
 Response: `text/event-stream`.
@@ -91,7 +91,7 @@ Example structured answer:
 
 ```text
 event: sources
-data: [{"document_id":"u_1a2b3c_orders","filename":"orders","page_label":null}]
+data: []
 
 event: status
 data: Planning analysis...
@@ -127,7 +127,7 @@ Behavior notes:
 - Greeting and farewell messages may use a fast canned path.
 - Stateless `chat` intent may use a 1-hour Redis response cache.
 - The router may classify messages into intents such as `chat`, `help`, `check`, `unstructured_flow`, or `structured_flow`.
-- `sources` can be empty for chat/help/error paths.
+- `sources` in the stream is **always `[]`** (KM-691) — read the real `sources[]` from `GET /api/v1/traceability` after `done`.
 - `status` events are optional and should be safe for the frontend to ignore.
 
 ## Tools
@@ -357,13 +357,15 @@ Response `404`:
 }
 ```
 
-## Observability
+## Traceability
+
+> Renamed from `observability` (team decision 2026-07-06) so it is never confused with the internal Langfuse *observability* stack (engineering telemetry, PII-masked). Traceability is **user-facing** provenance — real tool args, output previews, and the executed query — shown alongside the answer.
 
-### `GET /api/v1/observability`
+### `GET /api/v1/traceability`
 
-Returns Responsible AI provenance for one assistant answer.
+Returns user-facing provenance for one assistant answer.
 
-The frontend should call this after the chat/help stream emits `done`, using the `message_id` from the `done` event. If the row is not ready yet, the frontend may poll until `200` or stop on `404` according to product behavior.
+The frontend should call this after the chat/help stream emits `done`, using the `message_id` from the `done` event. The row is written **before** `done`, so an immediate GET returns `200` (no polling race). A `404` means the id is unknown or the turn errored before completing (error turns never produce a row).
 
 Query params:
 
@@ -375,15 +377,19 @@ Query params:
 Example:
 
 ```text
-GET /api/v1/observability?analysis_id=an_42&message_id=msg_88f1
+GET /api/v1/traceability?analysis_id=an_42&message_id=msg_88f1
 ```
 
+`intent` values the frontend may see: `chat` · `help` · `check` · `unstructured_flow` · `structured_flow` · `out_of_scope` · `blocked` (`blocked` = input-guard or Azure content-filter refusal; `chat` also covers the greeting fast-path and cache replays).
+
 Field rules:
 
-- `planning`: present only when the planner ran; otherwise `null`.
-- `thinking`: optional reasoning summary; `null` if unavailable.
-- `tool_calls`: every invoked tool with input, output, and status; empty for pure chat or greeting paths.
-- `sources`: required for retrieval flows; empty for chat/help paths that do not reference data.
+- `planning`: present only when the planner ran (`structured_flow`); otherwise `null`.
+- `thinking`: **always `null` in v1** — our agents are plain chat completions with no native reasoning output, and synthesizing it post-hoc would be unfaithful. The field stays in the payload so it can be populated later without a contract change.
+- `tool_calls`: every invoked tool with `input`, `output`, `status`, `task_id` (nullable), and `error` (nullable); empty for chat / help / greeting / refusal paths.
+- `sources`: required for retrieval flows; empty for chat / help / refusal paths and for `check`.
+- The payload also carries an internal `user_id` (ownership); the frontend may ignore it.
+- Truncation: `preview` ≤ 5 rows; any string inside `input`/`output`/`preview`/`snippet` ≤ 300 chars; rows beyond the preview are dropped (`row_count` is preserved).
 
 Response `200` for `structured_flow`:
 
@@ -391,49 +397,58 @@ Response `200` for `structured_flow`:
 {
   "analysis_id": "an_42",
   "message_id": "msg_88f1",
+  "user_id": "user_7",
   "intent": "structured_flow",
-  "generated_at": "2026-06-30T03:21:09.114Z",
+  "generated_at": "2026-07-06T03:21:09.114Z",
   "planning": {
     "goal_restated": "Find which regions drive revenue and why Q1 dipped.",
-    "assumptions": ["'last quarter' = Q1 2026"],
+    "assumptions": [],
     "steps": [
       {
         "step": 1,
         "stage": "data_understanding",
-        "objective": "Inventory the sales source"
+        "objective": "Inventory the sales source",
+        "status": "success",
+        "tools_used": ["check_data"]
       },
       {
         "step": 2,
         "stage": "modeling",
-        "objective": "Aggregate revenue by region"
+        "objective": "Aggregate revenue by region",
+        "status": "success",
+        "tools_used": ["retrieve_data", "analyze_aggregate"]
       }
     ]
   },
-  "thinking": "The question needs a per-region breakdown plus a cause, so I inventory the source, aggregate revenue by region, then compare quarters.",
+  "thinking": null,
   "tool_calls": [
     {
       "order": 1,
+      "task_id": null,
       "name": "check_data",
       "input": { "source_hint": "structured" },
-      "output": { "kind": "table", "summary": "1 source, 1 table, 48,213 rows" },
-      "status": "success"
+      "output": {
+        "kind": "table",
+        "columns": ["source_id", "name", "source_type", "table_count"],
+        "row_count": 1,
+        "preview": [["src_sales_db", "orders", "schema", 1]]
+      },
+      "status": "success",
+      "error": null
     },
     {
       "order": 2,
+      "task_id": null,
       "name": "retrieve_data",
-      "input": {
-        "source_id": "src_sales_db",
-        "table_id": "orders",
-        "select": ["region", "amount"],
-        "group_by": ["region"]
-      },
+      "input": { "ir": { "source_id": "src_sales_db", "table_id": "orders", "select": ["region", "amount"], "group_by": ["region"] } },
       "output": {
         "kind": "table",
         "columns": ["region", "total"],
         "row_count": 4,
         "preview": [["Central", 1210000], ["East", 740000]]
       },
-      "status": "success"
+      "status": "success",
+      "error": null
     }
   ],
   "sources": [
@@ -443,37 +458,36 @@ Response `200` for `structured_flow`:
       "name": "orders",
       "query": "SELECT region, SUM(amount) AS total FROM orders GROUP BY region",
       "detail": {
-        "tables": ["orders"],
-        "row_count": 48213
+        "table": "orders",
+        "row_count": 4
       }
     }
   ]
 }
 ```
 
+> Note: `retrieve_data`'s real `input` is the compiled query IR under an `ir` key (the planner builds an IR, never raw SQL). The executed SQL/rendered query appears on the corresponding `sources[].query`.
+
 Response `200` for `unstructured_flow`:
 
 ```json
 {
   "analysis_id": "an_42",
   "message_id": "msg_55",
+  "user_id": "user_7",
   "intent": "unstructured_flow",
-  "generated_at": "2026-06-30T03:40:02.001Z",
+  "generated_at": "2026-07-06T03:40:02.001Z",
   "planning": null,
   "thinking": null,
   "tool_calls": [
     {
       "order": 1,
+      "task_id": null,
       "name": "retrieve_knowledge",
-      "input": {
-        "query": "technology stack used in this project",
-        "top_k": 4
-      },
-      "output": {
-        "kind": "documents",
-        "row_count": 4
-      },
-      "status": "success"
+      "input": { "query": "technology stack used in this project" },
+      "output": { "kind": "documents", "row_count": 4 },
+      "status": "success",
+      "error": null
     }
   ],
   "sources": [
@@ -490,14 +504,15 @@ Response `200` for `unstructured_flow`:
 }
 ```
 
-Response `200` for simple chat or greeting:
+Response `200` for chat / greeting / help / refusals (`out_of_scope`, `blocked`):
 
 ```json
 {
   "analysis_id": "an_42",
   "message_id": "msg_12",
+  "user_id": "user_7",
   "intent": "chat",
-  "generated_at": "2026-06-30T03:05:00.000Z",
+  "generated_at": "2026-07-06T03:05:00.000Z",
   "planning": null,
   "thinking": null,
   "tool_calls": [],
@@ -509,13 +524,13 @@ Response `404`:
 
 ```json
 {
-  "detail": "No observability for message 'msg_88f1' yet."
+  "detail": "No traceability for message 'msg_88f1' yet."
 }
 ```
 
 Frontend rendering guidance:
 
-- Render observability separately from the streamed answer.
+- Render traceability separately from the streamed answer.
 - Default state can be collapsed.
 - Show planning, tool calls, and sources as separate sections.
 - Treat `planning: null`, `tool_calls: []`, and `sources: []` as valid states.

DEV_PLAN.md

@@ -30,20 +30,19 @@ the endpoint contract *before* coding the tools. Status legend: ⬜ not started
 | **2 — v2 + regroup** | Tools list → `/api/v1/tools/list` | Sofhia | ✅ | Renamed route `GET /api/v1/tools` → `GET /api/v1/tools/list` ([tools.py:133](src/api/v1/tools.py:133)). |
 | **2 — v2 + regroup** | FE: slash menu = `/help` only; report = right-side button | Mentor (FE) | ⬜ | Coordination note, not Python work. |
 | **3 — tools + obs** | Finish `help` so it actually **calls** (not just lists) + test | Sofhia | ⬜ | Mentor: help currently only lists tools. Core #2 after chat. |
-| **3 — tools + obs** | Observability **scratchpad** (decorator) accumulating in the chat agent | Rifqi + Sofhia | ⬜ | Capture planning / tool I/O / sources during the run; flush one record on `done`. |
-| **3 — tools + obs** | Audit `report_inputs` — covers planning + tool I/O + source? add cols / new store | Rifqi | ⬜ | **Rec:** dedicated provenance store = 1 JSONB row per message (logical 3 sections); keep Langfuse for engineering. |
-| **3 — tools + obs** | Build `GET /api/v1/observability` (one merged response) | Rifqi | ⬜ | Intent-based source rules (greeting/help = none; retrieve = required). Richness path-dependent (full planning only on slow path). |
-| **3 — tools + obs** | Keep stream **text-only**; observability is a separate parallel call | Rifqi | ⬜ | Per mentor — don't slow the stream. |
-| **3 — tools + obs** | Resolve `message_id` correlation (stream ↔ observability) with Harry | Rifqi ↔ Harry | ✅ | **RESOLVED (pr/6):** Python is the **sole minter** — `message_id` dropped from the `/api/v2/chat/stream` + `/api/v1/tools/help` request bodies; always minted server-side (server-authoritative, FE-security) and returned on `done`. Any caller-sent `message_id` is ignored. Contract open-Q #1 closed. |
+| **3 — tools + obs** | Traceability **scratchpad** accumulating in the chat agent | Rifqi + Sofhia | ✅ | **KM-691.** `TraceabilityScratchpad` + `TraceabilityToolInvoker` (`src/traceability/`) capture planning / tool I/O / sources during the run; flushed one row before every `done` (all 8 sites; error turns = no row). Renamed observability→traceability (vs. Langfuse). |
+| **3 — tools + obs** | Audit `report_inputs` — covers planning + tool I/O + source? add cols / new store | Rifqi | ✅ | **KM-691.** Chose a dedicated store: `message_traceability` = 1 JSONB row per message (Python-owned, like `report_inputs`; DDL run manually against dedorch, handed to Harry). Langfuse kept for engineering. |
+| **3 — tools + obs** | Build `GET /api/v1/traceability` (one merged response) | Rifqi | ✅ | **KM-691.** `src/api/v1/traceability.py` → store.get → payload/404. Intent-based source rules (greeting/help/refusals = none; retrieve = required); full planning only on slow path. Contract §7 updated. |
+| **3 — tools + obs** | Keep stream **text-only**; traceability is a separate parallel call | Rifqi | ✅ | **KM-691.** No trace data in the SSE stream; the FE fetches `/traceability` on `done`. |
+| **3 — tools + obs** | Resolve `message_id` correlation (stream ↔ traceability) with Harry | Rifqi ↔ Harry | ✅ | **RESOLVED (pr/6):** Python is the **sole minter** — `message_id` dropped from the `/api/v2/chat/stream` + `/api/v1/tools/help` request bodies; always minted server-side (server-authoritative, FE-security) and returned on `done`. Any caller-sent `message_id` is ignored. Contract open-Q #1 closed. |
 | **4 — biz questions** | Get Go folder; confirm `business_questions` in create-analysis (max 5); sync Python | Harry/Mentor → Rifqi | ⬜ | Go currently missing the field ("lagi difixing"). Python already models objective + business_questions. |
 | **deferred** | Report formats: PPT (preferred) / PDF / infographic on download | — | ⏸️ | MD is fine for the FE preview stage now. |
 | **deferred** | Charts (Plotly→JSON) + images tables | — | ⏸️ | Carried from §4 #26/#27. |
 
 **Next up:** Phase 2 Python work is **done** (chat→v2 `analysis_id`; `help`/`report`/`list` regrouped
 under `/api/v1/tools/`). The `message_id` correlation contract is now settled in **pr/6** (Python sole
-minter, stream-only). The full **Phase 3 observability build** — scratchpad + `GET /api/v1/observability`
-(shape sketched in the contract §7) — is slated for a **later PR**, then **Phase 4** (business
-questions, Go-blocked).
+minter, stream-only). The **Phase 3 traceability build** — scratchpad + `GET /api/v1/traceability`
+(contract §7) — is **done (KM-691)**. Next is **Phase 4** (business questions, Go-blocked).
 
 ---
 

REPO_STATUS.md

@@ -18,7 +18,7 @@ are placeholders (see §12).
 > [DEV_PLAN §0](DEV_PLAN.md)). **Python is becoming a generation/AI-only service** — Go owns the full
 > analysis lifecycle *and* the data-plane endpoints. Scope:
 > - **Unwired from `main` + Swagger** (router files kept, *not* deleted): `analysis` CRUD, `room`, `db_client`, `document`, `data_catalog`, `users`/login. **✅ DONE — KM-686, commit `0b2d678`** (so the §7 rows for these are now commented out of `main.py`).
-> - **AI surface that stays live:** `chat` → **`POST /api/v2/chat/stream`** (explicit **`analysis_id`**, not `room_id`); the skills regroup under **`/api/v1/tools/`** (`list` · `help` · `report`); plus a **new `GET /api/v1/observability`** (Responsible-AI provenance per answer, backed by a provenance store — shape TBD). **⬜ pending.**
+> - **AI surface that stays live:** `chat` → **`POST /api/v2/chat/stream`** (explicit **`analysis_id`**, not `room_id`); the skills regroup under **`/api/v1/tools/`** (`list` · `help` · `report`); plus a **new `GET /api/v1/traceability`** (user-facing provenance per answer, backed by a Python-owned `message_traceability` store — renamed from `observability`, KM-691). **✅ built.**
 > - **Only `chat/stream` moves to `/api/v2`;** everything else stays `/api/v1`.
 >
 > §2/§4/§7 below still describe the **pre-restructure wiring** except the unwire above, which has landed.
@@ -50,7 +50,7 @@ streaming.
 
 > **» pr/5 (decided, not yet in code):** Python's non-AI endpoints (analysis CRUD, `room`, `document`,
 > `db_client`, `data_catalog`, `users`/login) are being **unwired** — Python keeps only the
-> generation/AI surface (chat, tools: `help`/`report`/`list`, observability). See the Direction-update banner.
+> generation/AI surface (chat, tools: `help`/`report`/`list`, traceability). See the Direction-update banner.
 
 Shared infra: **Postgres** (app tables + `data_catalog` jsonb + PGVector `langchain_pg_embedding`), **Azure Blob**, and (Python-only) **Redis**.
 
@@ -78,8 +78,8 @@ Entry: `POST /api/v1/chat/stream` (`src/api/v1/chat.py`) → `ChatHandler.handle
 (`src/agents/chat_handler.py`). One shared `ChatHandler` per process keeps the Azure clients warm.
 
 > **» pr/5:** this endpoint moves to **`POST /api/v2/chat/stream`** with an explicit **`analysis_id`**
-> field (replacing `room_id`), and the observability detail (planning / tool I/O / sources) moves out of
-> the stream to a separate `GET /api/v1/observability` call. See the Direction-update banner.
+> field (replacing `room_id`), and the traceability detail (planning / tool I/O / sources) moves out of
+> the stream to a separate `GET /api/v1/traceability` call. See the Direction-update banner.
 
 ```
 POST /chat/stream { user_id, room_id, message }
@@ -156,7 +156,7 @@ Two facts to internalise:
 ## 7. API surface (this repo, all under `/api/v1`)
 
 > **» 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/observability`; and the
+> 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.
 

main.py

@@ -21,6 +21,7 @@ from slowapi.errors import RateLimitExceeded
 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.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
@@ -67,6 +68,7 @@ app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 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(chat_v2_router)     # pr/5 Phase 2: POST /api/v2/chat/stream (analysis_id)
 
 

src/agents/chat_handler.py

@@ -38,6 +38,7 @@ from langchain_core.messages import BaseMessage
 
 from src.middlewares.logging import get_logger
 from src.retrieval.base import RetrievalResult
+from src.traceability import TraceabilityScratchpad, TraceabilityToolInvoker
 
 from .chatbot import ChatbotAgent, DocumentChunk
 from .guard import InputGuard
@@ -54,6 +55,7 @@ from .refusals import blocked_message, out_of_scope_message
 if TYPE_CHECKING:
     from ..catalog.reader import CatalogReader
     from ..retrieval.router import RetrievalRouter
+    from ..traceability.store import TraceabilityStore
     from .gate import AnalysisState
     from .slow_path.coordinator import SlowPathCoordinator
     from .slow_path.store import ReportInputStore
@@ -102,6 +104,7 @@ class ChatHandler:
             Callable[[str], SlowPathCoordinator] | None
         ) = None,
         analysis_store: ReportInputStore | None = None,
+        traceability_store: TraceabilityStore | None = None,
         check_invoker_factory: Callable[[str], Any] | None = None,
         ps_agent: ProblemStatementAgent | None = None,
         help_agent: HelpAgent | None = None,
@@ -123,6 +126,9 @@ class ChatHandler:
         # factory + store are injectable for tests.
         self._slow_path_factory = slow_path_coordinator_factory
         self._analysis_store = analysis_store
+        # 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
         # `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
@@ -251,6 +257,7 @@ class ChatHandler:
         analysis_id: str | None,
         history: list[BaseMessage] | None = None,
         message: str | None = None,
+        message_id: str | None = None,
     ) -> AsyncIterator[dict[str, Any]]:
         """Deterministic `help` dispatch for the dedicated `/api/v1/tools/help` endpoint.
 
@@ -262,6 +269,12 @@ class ChatHandler:
         documents), `chunk`*, then `done` (data left empty; the endpoint stamps the
         `message_id`). On failure, yields a terminal `error` event.
         """
+        # Traceability (KM-691): a help turn has no planning/tools/sources — an
+        # empty payload stamped `help`, flushed before `done`.
+        pad = TraceabilityScratchpad()
+        pad.message_id = message_id
+        pad.set_intent("help")
+
         # Load (or lazily create) the analysis state; fail closed to a not-validated
         # stub so help degrades gracefully on a missing row / read error / legacy id.
         state: AnalysisState | None = None
@@ -292,6 +305,7 @@ class ChatHandler:
             logger.error("help streaming failed", user_id=user_id, error=str(e))
             yield {"event": "error", "data": f"Help generation failed: {e}"}
             return
+        await self._flush_trace(pad, analysis_id, user_id)
         yield {"event": "done", "data": ""}
 
     async def handle(
@@ -300,8 +314,14 @@ class ChatHandler:
         user_id: str,
         history: list[BaseMessage] | None = None,
         analysis_id: str | None = None,
+        message_id: str | None = None,
     ) -> AsyncIterator[dict[str, Any]]:
         tracer = self._make_tracer(user_id, message)
+        # Traceability (KM-691): per-request accumulator, flushed before EVERY `done`
+        # (§5 matrix). Default intent `chat` until the router classifies; the two
+        # refusal branches below stamp `blocked` explicitly.
+        pad = TraceabilityScratchpad()
+        pad.message_id = message_id
 
         # ---- 0. Input guard ------------------------------------------
         # Deliberate input-filtering layer BEFORE the router: screen for prompt-
@@ -319,6 +339,8 @@ class ChatHandler:
             yield {"event": "sources", "data": json.dumps([])}
             yield {"event": "chunk", "data": blocked_message(message)}
             tracer.end()
+            pad.set_intent("blocked")
+            await self._flush_trace(pad, analysis_id, user_id)
             yield {"event": "done", "data": ""}
             return
 
@@ -335,6 +357,8 @@ class ChatHandler:
                 yield {"event": "sources", "data": json.dumps([])}
                 yield {"event": "chunk", "data": blocked_message(message)}
                 tracer.end()
+                pad.set_intent("blocked")
+                await self._flush_trace(pad, analysis_id, user_id)
                 yield {"event": "done", "data": ""}
                 return
             logger.error("intent classification failed", error=repr(e))
@@ -345,6 +369,7 @@ class ChatHandler:
             return
 
         intent = decision.intent
+        pad.set_intent(intent)  # traceability: chat | check | *_flow | out_of_scope | help
         # ---- 1a. Ensure session state row (T-A) ----------------------
         # Rooms created via /room/create have no `analysis` row. Without one, Help and
         # the report_id write-back silently no-op. Lazily get-or-create it (idempotent).
@@ -392,6 +417,7 @@ class ChatHandler:
             yield {"event": "sources", "data": json.dumps([])}
             yield {"event": "chunk", "data": out_of_scope_message(message)}
             tracer.end()
+            await self._flush_trace(pad, analysis_id, user_id)
             yield {"event": "done", "data": ""}
             return
         if intent == "structured_flow":
@@ -416,7 +442,8 @@ class ChatHandler:
                 # make the assembled answer English for an Indonesian question).
                 reply_language = detect_reply_language(history, message=message)
                 async for event in self._run_slow_path(
-                    user_id, rewritten, catalog, tracer, reader, analysis_id, reply_language
+                    user_id, rewritten, catalog, tracer, reader, analysis_id,
+                    reply_language, pad,
                 ):
                     yield event
                 return
@@ -434,6 +461,14 @@ class ChatHandler:
                     rewritten, user_id
                 )
                 chunks = _normalize_chunks(raw_chunks)
+                # Traceability (KM-691): retrieval bypasses the tool invoker, so synth
+                # the retrieve_knowledge call (input = rewritten query) + document sources.
+                pad.record_tool_call(
+                    "retrieve_knowledge",
+                    {"query": rewritten},
+                    {"kind": "documents", "row_count": len(raw_chunks or [])},
+                )
+                pad.add_document_sources(raw_chunks, rewritten)
             except Exception as e:
                 logger.error(
                     "unstructured route failed", user_id=user_id, error=str(e)
@@ -442,13 +477,19 @@ class ChatHandler:
                 return
         elif intent == "check":
             try:
-                invoker = self._get_check_invoker(user_id)
-                text = await run_check(rewritten, invoker)
+                # Wrap the check invoker so its check_* tool calls land in the trace.
+                invoker = TraceabilityToolInvoker(self._get_check_invoker(user_id), pad)
+                # Detect from the ORIGINAL message (not `rewritten`, which the
+                # router normalizes to English) so the deterministic check reply
+                # matches the user's language like the other paths.
+                reply_language = detect_reply_language(history, message=message)
+                text = await run_check(rewritten, invoker, reply_language)
             except Exception as e:
                 logger.error("check route failed", user_id=user_id, error=str(e))
                 yield {"event": "error", "data": f"Lookup failed: {e}"}
                 return
             yield {"event": "chunk", "data": text}
+            await self._flush_trace(pad, analysis_id, user_id)
             yield {"event": "done", "data": ""}
             return
         # problem_statement dispatch removed 2026-06-24 (skill unwired; intent no longer
@@ -500,19 +541,17 @@ class ChatHandler:
                 yield {"event": "error", "data": f"Help generation failed: {e}"}
                 return
             tracer.end()
+            await self._flush_trace(pad, analysis_id, user_id)
             yield {"event": "done", "data": ""}
             return
         # else: chat path — no context
 
         # ---- 2b. Emit sources ---------------------------------------
-        sources = _build_sources(intent, user_id, query_result, raw_chunks)
-        logger.info(
-            "built sources",
-            intent=intent,
-            sources_count=len(sources),
-            raw_chunks_count=len(raw_chunks) if raw_chunks else 0,
-        )
-        yield {"event": "sources", "data": json.dumps(sources)}
+        # Sources moved to traceability (KM-691): the stream stays text-only. The FE
+        # reads sources from GET /api/v1/traceability (richer + intent-consistent; the
+        # document sources for this turn are captured on the pad above). The empty
+        # `sources` event is kept for SSE backward-compat.
+        yield {"event": "sources", "data": json.dumps([])}
 
         # ---- 3. Stream answer ----------------------------------------
         # masked: the answer call sees real query rows / doc chunks (possible PII).
@@ -533,6 +572,8 @@ class ChatHandler:
             return
 
         tracer.end()
+        # chat: empty payload; unstructured_flow: synth retrieve_knowledge + doc sources.
+        await self._flush_trace(pad, analysis_id, user_id)
         yield {"event": "done", "data": ""}
 
     # ------------------------------------------------------------------
@@ -550,7 +591,11 @@ class ChatHandler:
         return RequestTracer.start(user_id=user_id, question=question)
 
     def _get_slow_path_coordinator(
-        self, user_id: str, tracer: Any = None, catalog_reader: CatalogReader | None = None
+        self,
+        user_id: str,
+        tracer: Any = None,
+        catalog_reader: CatalogReader | None = None,
+        pad: TraceabilityScratchpad | None = None,
     ) -> SlowPathCoordinator:
         """Build the per-request slow-path coordinator (composition root).
 
@@ -579,6 +624,10 @@ class ChatHandler:
             from ..observability.langfuse.tracing import TracingToolInvoker
 
             invoker = TracingToolInvoker(invoker, tracer)
+        # Traceability outermost: records the SAME real I/O the tools return (both
+        # wrappers see it; order is immaterial). KM-691.
+        if pad is not None:
+            invoker = TraceabilityToolInvoker(invoker, pad)
         registry = default_registry()
         return SlowPathCoordinator(
             PlannerService(), TaskRunner(invoker, registry), Assembler(), registry
@@ -591,6 +640,30 @@ class ChatHandler:
             self._analysis_store = PostgresReportInputStore()
         return self._analysis_store
 
+    def _get_traceability_store(self) -> TraceabilityStore:
+        if self._traceability_store is None:
+            from ..traceability import PostgresTraceabilityStore
+
+            self._traceability_store = PostgresTraceabilityStore()
+        return self._traceability_store
+
+    async def _flush_trace(
+        self, pad: TraceabilityScratchpad, analysis_id: str | None, user_id: str
+    ) -> None:
+        """Persist the turn's traceability row right before `done` (KM-691).
+
+        No-op without a `message_id` (tests / unwired callers never flush). The
+        store save is itself never-throw, but we also guard here: a trace failure
+        must never break — or delay past `done` — the user's answer.
+        """
+        if pad.message_id is None:
+            return
+        try:
+            payload = pad.build(analysis_id or "", user_id, pad.message_id)
+            await self._get_traceability_store().save(payload)
+        except Exception as e:  # noqa: BLE001 — never break the answer on a trace slip
+            logger.warning("traceability flush failed", error=str(e))
+
     async def _run_slow_path(
         self,
         user_id: str,
@@ -600,6 +673,7 @@ class ChatHandler:
         catalog_reader: CatalogReader | None = None,
         analysis_id: str | None = None,
         reply_language: str | None = None,
+        pad: TraceabilityScratchpad | None = None,
     ) -> AsyncIterator[dict[str, Any]]:
         """Run the slow path and stream its assembled answer as SSE events.
 
@@ -617,7 +691,7 @@ class ChatHandler:
 
             tracer = NullTracer()
 
-        coordinator = self._get_slow_path_coordinator(user_id, tracer, catalog_reader)
+        coordinator = self._get_slow_path_coordinator(user_id, tracer, catalog_reader, pad)
         context = await get_business_context(user_id)
 
         # DB3: warm the user's DB connection in parallel with planning so the
@@ -673,7 +747,9 @@ class ChatHandler:
             yield {"event": "error", "data": f"Analysis failed: {e}"}
             return
 
-        yield {"event": "sources", "data": json.dumps([])}  # TODO: derive from record
+        # Sources live in traceability now (KM-691), derived from the run's
+        # retrieve_data calls; the stream stays text-only.
+        yield {"event": "sources", "data": json.dumps([])}
         yield {"event": "chunk", "data": result.chat_answer}
         try:
             # Stamp identity from the request scope: owner + the shared session id
@@ -684,9 +760,15 @@ class ChatHandler:
                 update={"user_id": user_id, "analysis_id": analysis_id}
             )
             await self._get_analysis_store().save(record)
+            if pad is not None:
+                # Traceability planning = goal_restated + tasks_run from the record;
+                # tool_calls were already recorded by the wrapped invoker.
+                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))
         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)
         yield {"event": "done", "data": ""}
 
 
@@ -715,54 +797,6 @@ class _ScopedCatalogReader:
         return catalog.model_copy(update={"sources": scoped or catalog.sources})
 
 
-def _build_sources(
-    intent: str,
-    user_id: str,
-    query_result: Any,
-    raw_chunks: Any,
-) -> list[dict[str, Any]]:
-    """Build the sources payload for the SSE `sources` event.
-
-    - structured_flow: one entry per executed table (table_name only).
-    - unstructured_flow: deduped by (document_id, page_label), Phase 1 shape.
-    - chat or error: empty list.
-    """
-    if intent == "structured_flow":
-        if query_result is None or getattr(query_result, "error", None):
-            return []
-        table_name = getattr(query_result, "table_name", "") or ""
-        if not table_name:
-            return []
-        return [{
-            "document_id": f"{user_id}_{table_name}",
-            "filename": table_name,
-            "page_label": None,
-        }]
-
-    if intent == "unstructured_flow" and raw_chunks:
-        seen: set[tuple[Any, Any]] = set()
-        sources: list[dict[str, Any]] = []
-        for item in raw_chunks:
-            if isinstance(item, RetrievalResult):
-                data = item.metadata.get("data", {})
-            elif isinstance(item, dict):
-                data = item
-            else:
-                continue
-            key = (data.get("document_id"), data.get("page_label"))
-            if key in seen or key == (None, None):
-                continue
-            seen.add(key)
-            sources.append({
-                "document_id": data.get("document_id"),
-                "filename": data.get("filename", "Unknown"),
-                "page_label": data.get("page_label", "Unknown"),
-            })
-        return sources
-
-    return []
-
-
 def _normalize_chunks(raw: Any) -> list[DocumentChunk]:
     """Convert whatever the retriever returns into list[DocumentChunk].
 

src/agents/handlers/check.py

@@ -14,23 +14,27 @@ from __future__ import annotations
 
 import asyncio
 import re
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from src.tools.contracts import ToolOutput
 
 if TYPE_CHECKING:
     from src.agents.slow_path.invoker import ToolInvoker
 
-# Cues that point at documents rather than structured data.
+# Cues that point at documents rather than structured data. Deliberately
+# document-SPECIFIC: generic words like "file" and "uploaded" are excluded
+# because users say "file/data yang aku upload" about tabular datasets too, and
+# forcing those to the document-only branch made "what data have I uploaded?"
+# answer "nothing" for a user whose data is tabular (lives under check_data).
+# A bare inventory question with no specific cue falls through to the
+# helicopter view (both structured + documents) instead.
 _KNOWLEDGE_CUES = (
     "document",
     "docs",
     "doc ",
-    "file",
     "pdf",
     "docx",
     ".txt",
-    "uploaded",
     "knowledge",
     "dokumen",
 )
@@ -50,6 +54,88 @@ _DATA_CUES = (
 )
 
 
+# Internal id columns — meaningful to the system (drill-down keys off them via
+# the ToolOutput rows), but noise in a user-facing answer. Hidden from the
+# rendered table; the raw ToolOutput still carries them for _matched_source_ids.
+# `table_count` is kept: always 1 for a tabular file, but the real table count
+# for a database, which is worth showing.
+_HIDDEN_COLUMNS = frozenset({"source_id", "table_id", "column_id"})
+
+# Friendly, localized labels for the catalog column ids the tools emit. Any
+# column not listed falls back to its raw name.
+_HEADER_LABELS = {
+    "English": {
+        "name": "Name", "source_type": "Type", "table_count": "Tables",
+        "table_name": "Table", "table_row_count": "Rows",
+        "column_name": "Column", "data_type": "Data type",
+        "nullable": "Nullable", "pii_flag": "PII",
+    },
+    "Indonesian": {
+        "name": "Nama", "source_type": "Jenis", "table_count": "Jumlah tabel",
+        "table_name": "Tabel", "table_row_count": "Jumlah baris",
+        "column_name": "Kolom", "data_type": "Tipe data",
+        "nullable": "Boleh kosong", "pii_flag": "PII",
+    },
+}
+
+# Friendly values for the raw `source_type` enum ("schema"/"tabular"/"unstructured").
+_SOURCE_TYPE_LABELS = {
+    "English": {"tabular": "Tabular file", "schema": "Database", "unstructured": "Document"},
+    "Indonesian": {"tabular": "File tabular", "schema": "Database", "unstructured": "Dokumen"},
+}
+
+
+# User-facing fixed strings, localized by reply language (detect_reply_language
+# returns "Indonesian"/"English"). Table headers/values are humanized via the
+# maps above; only the prose the user reads is translated here.
+_STRINGS = {
+    "English": {
+        "no_match": "Nothing registered yet — I don't see any matching sources.",
+        "none": "Nothing registered yet — I don't see any sources or documents.",
+        "structured": "Structured data",
+        "documents": "Documents",
+        "lookup_error": "Sorry, I couldn't look that up: {error}",
+    },
+    "Indonesian": {
+        "no_match": "Belum ada yang terdaftar — aku tidak menemukan sumber data yang cocok.",
+        "none": "Belum ada yang terdaftar — aku tidak menemukan sumber data atau dokumen.",
+        "structured": "Data terstruktur",
+        "documents": "Dokumen",
+        "lookup_error": "Maaf, aku tidak bisa mencarinya: {error}",
+    },
+}
+
+
+def _s(reply_language: str) -> dict[str, str]:
+    """Localized string bundle; defaults to English for an unknown language."""
+    return _STRINGS.get(reply_language, _STRINGS["English"])
+
+
+# Natural-language lead-ins so a check reply reads like an answer, not a raw
+# table dump. `{n}` = item count, `{s}` = English plural suffix, `{name}` = source.
+_LEADS = {
+    "English": {
+        "opener": "Here's what you have:",
+        "structured": "You have {n} structured data source{s}:",
+        "documents": "You have {n} document{s}:",
+        "schema": "Here's what's inside {name}:",
+    },
+    "Indonesian": {
+        "opener": "Berikut data yang kamu punya:",
+        "structured": "Kamu punya {n} sumber data terstruktur:",
+        "documents": "Kamu punya {n} dokumen:",
+        "schema": "Berikut isi {name}:",
+    },
+}
+
+
+def _lead(kind: str, reply_language: str, n: int = 0, name: str = "") -> str:
+    """Build a localized lead-in sentence (English pluralizes on {n})."""
+    bundle = _LEADS.get(reply_language, _LEADS["English"])
+    suffix = "s" if (reply_language == "English" and n != 1) else ""
+    return bundle[kind].format(n=n, s=suffix, name=name)
+
+
 def _intent(message: str) -> str:
     """Return 'knowledge', 'data', or 'both' (helicopter view) from keyword cues."""
     lowered = message.lower()
@@ -62,22 +148,75 @@ def _intent(message: str) -> str:
     return "both"
 
 
-def render_tool_output(out: ToolOutput) -> str:
+def render_tool_output(out: ToolOutput, reply_language: str = "English") -> str:
     """Render a `check_*` ToolOutput table into a markdown string, or '' if empty."""
     if out.kind == "error":
-        return f"Sorry, I couldn't look that up: {out.error}"
+        return _s(reply_language)["lookup_error"].format(error=out.error)
     columns = out.columns or []
     rows = out.rows or []
     if not rows:
         return ""
-    header = "| " + " | ".join(columns) + " |"
-    separator = "| " + " | ".join("---" for _ in columns) + " |"
+
+    # Drop internal id columns, humanize the rest of the headers, and translate
+    # the source_type enum — the user sees names, not UUIDs and raw enums.
+    keep = [i for i, c in enumerate(columns) if c not in _HIDDEN_COLUMNS] or list(
+        range(len(columns))
+    )
+    labels = _HEADER_LABELS.get(reply_language, _HEADER_LABELS["English"])
+    type_labels = _SOURCE_TYPE_LABELS.get(reply_language, _SOURCE_TYPE_LABELS["English"])
+    type_idx = columns.index("source_type") if "source_type" in columns else -1
+
+    def _cell(i: int, row: list[Any]) -> str:
+        if i == type_idx:
+            return type_labels.get(str(row[i]), str(row[i]))
+        return str(row[i])
+
+    header = "| " + " | ".join(labels.get(columns[i], columns[i]) for i in keep) + " |"
+    separator = "| " + " | ".join("---" for _ in keep) + " |"
     body = "\n".join(
-        "| " + " | ".join(str(cell) for cell in row) + " |" for row in rows
+        "| " + " | ".join(_cell(i, row) for i in keep) + " |" for row in rows
     )
     return f"{header}\n{separator}\n{body}"
 
 
+def _render_source_list(out: ToolOutput, reply_language: str) -> str:
+    """Render a check_data/check_knowledge *listing* as a bullet list, not a table.
+
+    One bullet per source: `- name — Type (N tables)`. The type + table-count
+    annotation is only added for structured sources (file vs database); documents
+    are all "unstructured", so the section header already says so — just the name.
+    Returns '' when there are no rows. (The column-level schema drill-down still
+    renders as a table via `render_tool_output` — that data is genuinely tabular.)
+    """
+    if out.kind == "error":
+        return _s(reply_language)["lookup_error"].format(error=out.error)
+    columns = out.columns or []
+    rows = out.rows or []
+    if not rows:
+        return ""
+
+    idx = {c: i for i, c in enumerate(columns)}
+    type_labels = _SOURCE_TYPE_LABELS.get(reply_language, _SOURCE_TYPE_LABELS["English"])
+
+    def _table_word(n: int) -> str:
+        if reply_language == "English":
+            return "table" if n == 1 else "tables"
+        return "tabel"
+
+    items: list[str] = []
+    for row in rows:
+        name = str(row[idx["name"]]) if "name" in idx else ""
+        st = str(row[idx["source_type"]]) if "source_type" in idx else ""
+        annotation = ""
+        if st and st != "unstructured":
+            annotation = type_labels.get(st, st)
+            if "table_count" in idx:
+                tc = row[idx["table_count"]]
+                annotation += f" ({tc} {_table_word(int(tc))})"
+        items.append(f"- {name} — {annotation}" if annotation else f"- {name}")
+    return "\n".join(items)
+
+
 def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
     """All source_ids whose name appears as a whole word in the message.
 
@@ -106,60 +245,325 @@ def _matched_source_ids(message: str, inventory: ToolOutput) -> list[str]:
     return matched
 
 
-def _render_helicopter(data_out: ToolOutput, knowledge_out: ToolOutput) -> str:
+def _render_helicopter(
+    data_out: ToolOutput, knowledge_out: ToolOutput, reply_language: str = "English"
+) -> str:
     """Stitch structured + document inventory into one helicopter-view reply."""
+    strings = _s(reply_language)
     parts: list[str] = []
 
-    data_table = render_tool_output(data_out)
-    if data_table:
-        parts.append(f"**Data terstruktur**\n{data_table}")
+    data_list = _render_source_list(data_out, reply_language)
+    if data_list:
+        parts.append(f"{strings['structured']}:\n{data_list}")
 
-    knowledge_table = render_tool_output(knowledge_out)
-    if knowledge_table:
-        parts.append(f"**Dokumen**\n{knowledge_table}")
+    knowledge_list = _render_source_list(knowledge_out, reply_language)
+    if knowledge_list:
+        parts.append(f"{strings['documents']}:\n{knowledge_list}")
 
     if not parts:
-        return "Nothing registered yet — I don't see any sources or documents."
+        return strings["none"]
+
+    opener = _lead("opener", reply_language)
+    return opener + "\n\n" + "\n\n".join(parts)
+
+
+# ---------------------------------------------------------------------------
+# Schema-detail view (columns + types)
+# ---------------------------------------------------------------------------
+
+# Cues meaning "show me the columns/types", not just "what sources exist". When
+# present, the handler drills into the schema of the named source(s) — or ALL
+# structured sources when none is named — instead of listing. This is what lets
+# "kolom apa aja yang aku miliki dan tipenya?" resolve straight to the data
+# without the user having to type a full filename.
+_SCHEMA_CUES = (
+    "kolom", "column", "tipe data", "data type", "tipe", "dtype",
+    "struktur", "structure", "schema", "skema", "field",
+)
+
+# A database with more tables than this is summarised (table names + counts)
+# rather than dumping every column — otherwise a big DB is a wall of text.
+_DB_TABLE_THRESHOLD = 3
+
+# Friendly, localized data-type words (normalized from the catalog's raw type).
+_TYPE_WORDS = {
+    "English": {"string": "text", "integer": "integer", "float": "decimal",
+                "date": "date", "boolean": "boolean", "other": "other"},
+    "Indonesian": {"string": "teks", "integer": "bilangan bulat", "float": "desimal",
+                   "date": "tanggal", "boolean": "boolean", "other": "lainnya"},
+}
+
+# Localized strings for the schema view + the per-source Ringkasan.
+_SCHEMA_STR = {
+    "English": {
+        "lead": "Here are the columns and data types across your {n} source{s}:",
+        "summary": "Summary:",
+        "cat": "Categorical", "int": "Numeric (integer)",
+        "float": "Numeric (decimal)", "date": "Date",
+        "note_date": "still text, consider converting to datetime",
+        "note_id": "likely an identifier",
+        "rows_word": "rows", "table_word": "Table",
+        "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:",
+        "summary": "Ringkasan:",
+        "cat": "Kategorikal", "int": "Numerik (bilangan bulat)",
+        "float": "Numerik (desimal)", "date": "Tanggal",
+        "note_date": "masih teks, sebaiknya dikonversi ke datetime",
+        "note_id": "kemungkinan identifier",
+        "rows_word": "baris", "table_word": "Tabel",
+        "yes": "Ya", "no": "—",
+        "db_tables": "{name} punya {n} tabel:",
+        "db_item": "- {table} ({cols} kolom{rows})",
+        "db_hint": "Sebut nama tabelnya untuk lihat kolomnya.",
+    },
+}
+
+
+def _sc(reply_language: str) -> dict[str, str]:
+    """Localized schema/summary string bundle; defaults to English."""
+    return _SCHEMA_STR.get(reply_language, _SCHEMA_STR["English"])
+
+
+def _wants_schema(message: str) -> bool:
+    """True when the message asks about columns/types (schema detail)."""
+    lowered = message.lower()
+    return any(cue in lowered for cue in _SCHEMA_CUES)
 
+
+def _type_category(raw: str) -> str:
+    """Normalize a catalog `data_type` string into a coarse category."""
+    r = (raw or "").lower()
+    if "bool" in r:
+        return "boolean"
+    if any(k in r for k in ("float", "double", "decimal", "numeric", "real")):
+        return "float"
+    if "int" in r:
+        return "integer"
+    if any(k in r for k in ("date", "time", "timestamp")):
+        return "date"
+    if any(k in r for k in ("char", "text", "string", "object", "varchar", "str")):
+        return "string"
+    return "other"
+
+
+def _looks_date(name: str) -> bool:
+    return any(k in name for k in ("date", "tanggal", "tgl"))
+
+
+def _looks_id(name: str) -> bool:
+    tokens = re.split(r"[^a-z0-9]+", name)
+    return "id" in tokens or "email" in tokens or name.endswith("id")
+
+
+def _bucket_and_note(name: str, data_type: str, reply_language: str) -> tuple[str, str]:
+    """Assign a column to a Ringkasan bucket + an optional heuristic note.
+
+    Grouping is by data type, with two name-based overrides that mirror how a
+    human reads a schema: a date-named column goes to Date even when stored as
+    text (with a 'convert to datetime' note), and a numeric column whose name
+    looks like an id gets an 'identifier' note (shouldn't be treated as a
+    quantity). Deterministic — heuristics only, no LLM.
+    """
+    sc = _sc(reply_language)
+    cat = _type_category(data_type)
+    lname = name.lower()
+    if _looks_date(lname) or cat == "date":
+        return "date", (sc["note_date"] if cat != "date" else "")
+    if cat == "integer":
+        return "int", (sc["note_id"] if _looks_id(lname) else "")
+    if cat == "float":
+        return "float", (sc["note_id"] if _looks_id(lname) else "")
+    return "cat", ""
+
+
+def _schema_rows(out: ToolOutput) -> list[dict[str, Any]]:
+    """Flatten a check_data(source_id) ToolOutput into per-column dicts."""
+    cols = out.columns or []
+    idx = {c: i for i, c in enumerate(cols)}
+
+    def _get(row: list[Any], key: str, default: object = None) -> object:
+        return row[idx[key]] if key in idx else default
+
+    rows: list[dict[str, Any]] = []
+    for r in out.rows or []:
+        rows.append({
+            "table": str(_get(r, "table_name", "")),
+            "table_rows": _get(r, "table_row_count"),
+            "column": str(_get(r, "column_name", "")),
+            "data_type": str(_get(r, "data_type", "")),
+            "nullable": bool(_get(r, "nullable", False)),
+            "pii": bool(_get(r, "pii_flag", False)),
+        })
+    return rows
+
+
+def _columns_table(cols: list[dict[str, Any]], reply_language: str) -> str:
+    """Lean per-table markdown: Column · Data type (+ Nullable if mixed, + PII if any)."""
+    labels = _HEADER_LABELS.get(reply_language, _HEADER_LABELS["English"])
+    types = _TYPE_WORDS.get(reply_language, _TYPE_WORDS["English"])
+    sc = _sc(reply_language)
+    # Nullable only carries signal when a table mixes nullable + non-nullable;
+    # PII only when at least one column is flagged. Otherwise the column is noise.
+    show_nullable = len({c["nullable"] for c in cols}) > 1
+    show_pii = any(c["pii"] for c in cols)
+
+    headers = [labels["column_name"], labels["data_type"]]
+    if show_nullable:
+        headers.append(labels["nullable"])
+    if show_pii:
+        headers.append(labels["pii_flag"])
+    lines = [
+        "| " + " | ".join(headers) + " |",
+        "| " + " | ".join("---" for _ in headers) + " |",
+    ]
+    for c in cols:
+        cells = [c["column"], types.get(_type_category(c["data_type"]), c["data_type"])]
+        if show_nullable:
+            cells.append(sc["yes"] if c["nullable"] else sc["no"])
+        if show_pii:
+            cells.append(sc["yes"] if c["pii"] else sc["no"])
+        lines.append("| " + " | ".join(cells) + " |")
+    return "\n".join(lines)
+
+
+def _render_summary(cols: list[dict[str, Any]], reply_language: str) -> str:
+    """Adaptive per-source Ringkasan: only buckets that actually have columns."""
+    sc = _sc(reply_language)
+    buckets: dict[str, list[str]] = {"cat": [], "date": [], "int": [], "float": []}
+    for c in cols:
+        bucket, note = _bucket_and_note(c["column"], c["data_type"], reply_language)
+        buckets[bucket].append(f"{c['column']} ({note})" if note else c["column"])
+    lines = [sc["summary"]]
+    emitted = False
+    for key in ("cat", "date", "int", "float"):
+        if buckets[key]:
+            emitted = True
+            lines.append(f"- {sc[key]}: " + ", ".join(buckets[key]))
+    return "\n".join(lines) if emitted else ""
+
+
+def _render_schema_source(out: ToolOutput, reply_language: str) -> str:
+    """Render one source's schema: columns per table + a Ringkasan.
+
+    A database with more than `_DB_TABLE_THRESHOLD` tables is summarised at the
+    table level instead (name + column/row counts) so it doesn't become a wall
+    of text; the user names a table to drill into its columns.
+    """
+    if out.kind == "error":
+        return _s(reply_language)["lookup_error"].format(error=out.error)
+    rows = _schema_rows(out)
+    if not rows:
+        return ""
+    meta = out.meta or {}
+    name = str(meta.get("source_name") or "source")
+    source_type = str(meta.get("source_type") or "")
+    sc = _sc(reply_language)
+
+    tables: dict[str, list[dict[str, Any]]] = {}
+    for r in rows:
+        tables.setdefault(r["table"], []).append(r)
+
+    def _rows_suffix(rc: object) -> str:
+        return f", {rc} {sc['rows_word']}" if rc else ""
+
+    if source_type == "schema" and len(tables) > _DB_TABLE_THRESHOLD:
+        lines = [sc["db_tables"].format(name=name, n=len(tables))]
+        for tname, tcols in tables.items():
+            rc = tcols[0]["table_rows"]
+            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] = []
+    if len(tables) == 1:
+        tname, tcols = next(iter(tables.items()))
+        rc = tcols[0]["table_rows"]
+        head = f"**{name}**" + (f" ({rc} {sc['rows_word']})" if rc else "")
+        parts.append(f"{head}\n{_columns_table(tcols, reply_language)}")
+    else:
+        parts.append(f"**{name}**")
+        for tname, tcols in tables.items():
+            rc = tcols[0]["table_rows"]
+            sub = f"{sc['table_word']} {tname}" + (f" ({rc} {sc['rows_word']})" if rc else "") + ":"
+            parts.append(f"{sub}\n{_columns_table(tcols, reply_language)}")
+
+    summary = _render_summary(rows, reply_language)
+    if summary:
+        parts.append(summary)
     return "\n\n".join(parts)
 
 
-async def run_check(message: str, invoker: ToolInvoker) -> str:
-    """Route to check_data, check_knowledge, or both (helicopter view) based on cues."""
+def _render_schemas(schemas: list[ToolOutput], reply_language: str) -> str:
+    """Stitch one or more source schemas under a single lead-in sentence."""
+    blocks = [b for b in (_render_schema_source(o, reply_language) for o in schemas) if b]
+    if not blocks:
+        return ""
+    n = len(blocks)
+    suffix = "s" if (reply_language == "English" and n != 1) else ""
+    lead = _sc(reply_language)["lead"].format(n=n, s=suffix)
+    return lead + "\n\n" + "\n\n".join(blocks)
+
+
+async def run_check(
+    message: str, invoker: ToolInvoker, reply_language: str = "English"
+) -> str:
+    """Route to check_data, check_knowledge, or both (helicopter view) based on cues.
+
+    `reply_language` ("Indonesian"/"English", from `detect_reply_language`)
+    localizes the fixed prose so this deterministic path answers in the user's
+    language like the LLM paths do.
+    """
     intent = _intent(message)
 
-    _no_match = "Nothing registered yet — I don't see any matching sources."
+    _no_match = _s(reply_language)["no_match"]
 
     if intent == "knowledge":
         out = await invoker.invoke("check_knowledge", {})
-        return render_tool_output(out) or _no_match
+        listing = _render_source_list(out, reply_language)
+        if not listing:
+            return _no_match
+        return _lead("documents", reply_language, len(out.rows or [])) + "\n\n" + listing
 
-    if intent == "data":
+    # Schema-detail question ("kolom apa aja + tipe datanya"): drill into the
+    # schema of the named source(s), or ALL structured sources when none is
+    # named — so the user never has to type a full filename to see their columns
+    # and types. This supersedes name-matched drill-down (a source referred to as
+    # "dataset itu" now resolves via the single-/all-source fallback).
+    if _wants_schema(message):
         inventory = await invoker.invoke("check_data", {})
         if inventory.kind == "error":
-            return render_tool_output(inventory)
-        # Drill down to the schema of each source the user named; if they named
-        # none, return the source listing.
-        source_ids = _matched_source_ids(message, inventory)
-        if not source_ids:
-            return render_tool_output(inventory) or _no_match
+            return render_tool_output(inventory, reply_language)
+        if not (inventory.rows or []):
+            return _no_match
+        named = _matched_source_ids(message, inventory)
+        cols = inventory.columns or []
+        sid_i = cols.index("source_id") if "source_id" in cols else 0
+        ids = named or [str(r[sid_i]) for r in (inventory.rows or [])]
         schemas = await asyncio.gather(
-            *(invoker.invoke("check_data", {"source_id": sid}) for sid in source_ids)
+            *(invoker.invoke("check_data", {"source_id": sid}) for sid in ids)
         )
-        if len(schemas) == 1:
-            return render_tool_output(schemas[0]) or _no_match
-        # Multiple named sources → one labelled section per source.
-        sections: list[str] = []
-        for out in schemas:
-            table = render_tool_output(out)
-            if table:
-                label = (out.meta or {}).get("source_name") or "source"
-                sections.append(f"**{label}**\n{table}")
-        return "\n\n".join(sections) or _no_match
+        return _render_schemas(schemas, reply_language) or _no_match
+
+    if intent == "data":
+        inventory = await invoker.invoke("check_data", {})
+        if inventory.kind == "error":
+            return render_tool_output(inventory, reply_language)
+        listing = _render_source_list(inventory, reply_language)
+        if not listing:
+            return _no_match
+        n = len(inventory.rows or [])
+        return _lead("structured", reply_language, n) + "\n\n" + listing
 
     # broad / ambiguous → helicopter view: call both concurrently
     data_out, knowledge_out = await asyncio.gather(
         invoker.invoke("check_data", {}),
         invoker.invoke("check_knowledge", {}),
     )
-    return _render_helicopter(data_out, knowledge_out)
+    return _render_helicopter(data_out, knowledge_out, reply_language)

src/api/v1/help.py

@@ -62,6 +62,7 @@ async def help_stream(request: HelpRequest, db: AsyncSession = Depends(get_db)):
                 request.analysis_id,
                 history=history,
                 message=None,
+                message_id=message_id,
             ):
                 if event["event"] == "done":
                     # Stamp the turn id so the FE can fetch /observability for it.

src/api/v1/traceability.py

@@ -0,0 +1,43 @@
+"""Traceability endpoint — user-facing per-turn provenance (KM-691).
+
+`GET /api/v1/traceability?analysis_id=&message_id=` returns the provenance record for
+one assistant turn: planning / tool calls (with real inputs + outputs) / data sources
+(with the executed query). One JSONB row per assistant `message_id`, written by the
+chat pipeline right before the `done` SSE event; the FE fires this GET on `done`.
+
+Renamed from the contracted `/api/v1/observability` (team decision 2026-07-06) so it is
+never confused with the Langfuse *observability* stack (engineering-only, PII-masked).
+No auth — Go fronts Python.
+"""
+
+from fastapi import APIRouter, HTTPException, Query
+
+from src.middlewares.logging import get_logger, log_execution
+from src.traceability import PostgresTraceabilityStore, TraceabilityPayload
+
+logger = get_logger("traceability_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Traceability"])
+
+# Warm, process-shared store (mirrors the chat endpoints' module-level instances).
+_store = PostgresTraceabilityStore()
+
+
+@router.get("/traceability", response_model=TraceabilityPayload)
+@log_execution(logger)
+async def get_traceability(
+    analysis_id: str = Query(..., description="Analysis/session id"),
+    message_id: str = Query(
+        ..., description="Assistant turn id, taken from the `done` SSE event"
+    ),
+) -> TraceabilityPayload:
+    """Fetch one turn's provenance record. 404 while the turn is still running or if
+    the id is unknown (the FE never gets a `message_id` for error turns, so 404 is
+    the correct answer there)."""
+    payload = await _store.get(analysis_id, message_id)
+    if payload is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No traceability for message '{message_id}' yet.",
+        )
+    return payload

src/api/v2/chat.py

@@ -44,11 +44,29 @@ from src.db.postgres.connection import get_db
 from src.db.redis.connection import get_redis
 from src.middlewares.logging import get_logger, log_execution
 from src.middlewares.rate_limit import limiter
+from src.traceability import PostgresTraceabilityStore, TraceabilityScratchpad
 
 logger = get_logger("chat_api_v2")
 
 router = APIRouter(prefix="/api/v2", tags=["Chat"])
 
+# Module-level store (mirrors the warm, process-shared `_chat_handler`): the greeting
+# fast-path and cache-replay branches never enter `ChatHandler.handle`, so they write
+# their (empty `chat`) traceability row directly here. KM-691.
+_traceability_store = PostgresTraceabilityStore()
+
+
+async def _save_empty_chat_trace(analysis_id: str, user_id: str, message_id: str) -> None:
+    """Persist an empty `chat` traceability row for a turn that bypassed the handler
+    (greeting / cache replay), so the FE's GET on `done` returns a payload, not a 404."""
+    try:
+        pad = TraceabilityScratchpad()
+        pad.message_id = message_id
+        pad.set_intent("chat")
+        await _traceability_store.save(pad.build(analysis_id, user_id, message_id))
+    except Exception as e:  # noqa: BLE001 — never break the reply on a trace slip
+        logger.warning("traceability direct save failed", message_id=message_id, error=str(e))
+
 
 def _mint_message_id() -> str:
     """Mint the assistant turn id. Server-authoritative — never accepted from the caller
@@ -75,11 +93,11 @@ async def chat_stream(
     """Chat endpoint with streaming response (v2 — keyed on `analysis_id`).
 
     SSE event sequence:
-      1. sources  — JSON array of source refs (table for structured; deduped
-                    document_id/page_label for unstructured; [] for chat/help/error)
+      1. sources  — always `[]` (KM-691): sources moved to GET /api/v1/traceability;
+                    the stream stays text-only. Event kept for backward-compat.
       2. status   — slow-path progress pings (optional)
       3. chunk    — text fragments of the answer
-      4. done     — {"message_id": "..."} for the observability lookup
+      4. done     — {"message_id": "..."} for the traceability lookup
     """
     analysis_id = body.analysis_id
     message_id = _mint_message_id()
@@ -103,6 +121,8 @@ async def chat_stream(
                 yield {"event": "chunk", "data": cached_text[i:i + 50]}
             yield done_event
 
+        # Write the row BEFORE the stream so the FE's GET on `done` can't race a 404.
+        await _save_empty_chat_trace(analysis_id, body.user_id, message_id)
         return EventSourceResponse(stream_cached())
 
     try:
@@ -116,6 +136,8 @@ async def chat_stream(
                 yield {"event": "chunk", "data": direct}
                 yield done_event
 
+            # Write the row BEFORE the stream so the FE's GET on `done` can't race a 404.
+            await _save_empty_chat_trace(analysis_id, body.user_id, message_id)
             return EventSourceResponse(stream_direct())
 
         history = await load_history(db, analysis_id, limit=10)
@@ -127,7 +149,8 @@ async def chat_stream(
             sources: list[dict[str, Any]] = []
             effective_intent: str | None = None
             async for event in handler.handle(
-                body.message, body.user_id, history, analysis_id=analysis_id
+                body.message, body.user_id, history,
+                analysis_id=analysis_id, message_id=message_id,
             ):
                 if event["event"] == "intent":
                     # consumed internally (not forwarded); gates caching below.

src/config/prompts/planner.md

@@ -84,6 +84,21 @@ only a `TaskList` object that conforms to the provided schema.
   - For `in`/`not_in` (value is a list) or `between` (value is `[low, high]`), it
     is still the ELEMENT type — a list of names is `"string"`, a date range is
     `"date"`. It is **never** `"list"`.
+- **Filters are ANDed — never stack two disjoint ranges on one column.** Every
+  `filters[]` entry combines with AND, so two non-overlapping `between` ranges on the
+  SAME column (e.g. a date in Q1 2025 AND in Q1 2026) match **zero rows**. Never AND
+  two ranges on one column expecting an OR.
+- **Period-over-period comparison = one query per period.** To compare two time
+  windows (e.g. Q1 2025 vs Q1 2026), emit a SEPARATE `retrieve_data` (+ its own
+  `analyze_*`) task for EACH period, each carrying its own `between` filter for that
+  window; the final answer compares the per-period results. Do NOT attempt it in one
+  query — there is no way to bucket a date into a quarter/month (no derived columns;
+  `group_by` is over EXISTING columns only), so a single spanning range collapses the
+  periods into one number and the comparison is lost.
+- **Time-bound a metric by the event's OWN date.** Filter a measure by the column
+  that dates the event you are measuring — revenue/sales by the order/transaction
+  date, NOT by an entity-creation date like `products.created_at` or
+  `customers.registered_at`. Join to the table that holds that date if needed.
 - **Use only the `args` a tool lists** — e.g. `analyze_aggregate` takes only
   `data`/`aggregations`/`group_by`, so never add `order_by`/`limit` to it.
 - **Top-N ("top/most/least N by <metric>") is a single `retrieve_data` query**,

src/db/postgres/models.py

@@ -278,3 +278,31 @@ class AnalysesMessageRow(Base):
     role = Column(String, nullable=False)  # user | ai
     content = Column(Text, nullable=False)
     created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+
+class MessageTraceabilityRow(Base):
+    """One row per assistant turn — user-facing provenance (KM-691).
+
+    `data` holds the full Pydantic `TraceabilityPayload`
+    (src\\traceability\\schemas.py:TraceabilityPayload) serialized via
+    `model_dump(mode="json", by_alias=True)`; the read path rehydrates with
+    `TraceabilityPayload.model_validate(...)`. One row per assistant `message_id`
+    (the Python-minted turn id "msg_<12hex>"), written before the `done` SSE event
+    and served by `GET /api/v1/traceability`.
+
+    OWNERSHIP / HANDOFF (KM-691): **Python-owned for now**, the same pattern as
+    `report_inputs` (ReportInputRow). 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-06); the finalized schema goes to Harry so the dedorch
+    migration creates it later (`message_id`/`analysis_id` gain the FK to
+    `analyses(id)` there). Distinct from Langfuse observability — this is unmasked,
+    user-facing provenance, not engineering telemetry.
+    """
+    __tablename__ = "message_traceability"
+
+    message_id = Column(String, primary_key=True)  # Python-minted turn id ("msg_<12hex>")
+    analysis_id = Column(UUID(as_uuid=False), nullable=False, index=True)  # analysis session id
+    user_id = Column(String, nullable=False)
+    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())

src/query/executor/base.py

@@ -20,6 +20,9 @@ class QueryResult:
     table_id: str = ""
     table_name: str = ""
     source_name: str = ""
+    # The executed query string (compiled SQL for db, rendered pandas chain for
+    # tabular), surfaced to traceability. None on error/unset paths. KM-691.
+    query: str | None = None
 
 
 class BaseExecutor(ABC):

src/query/executor/db.py

@@ -114,6 +114,7 @@ class DbExecutor(BaseExecutor):
                 table_id=ir.table_id,
                 table_name=table_name,
                 source_name=source_name,
+                query=compiled.sql,  # executed SQL, for traceability (KM-691)
             )
 
         except Exception as e:

src/query/executor/tabular.py

@@ -83,7 +83,8 @@ class TabularExecutor(BaseExecutor):
                 )
 
             compiled = self._compiler.compile(ir)
-            logger.info("pandas query", query=_render_query(ir, {c.column_id: c for c in table.columns}))
+            rendered_query = _render_query(ir, {c.column_id: c for c in table.columns})
+            logger.info("pandas query", query=rendered_query)
             blob_name = _resolve_blob_name(source, table)
             blob_bytes = await self._fetch_blob(blob_name)
 
@@ -113,6 +114,7 @@ class TabularExecutor(BaseExecutor):
                 table_id=ir.table_id,
                 table_name=table_name,
                 source_name=source_name,
+                query=rendered_query,  # rendered pandas chain, for traceability (KM-691)
             )
 
         except Exception as e:

src/query/ir/validator.py

@@ -5,6 +5,8 @@ is re-prompted with the error context (max 3 retries) — error messages
 must therefore be specific enough that the LLM can self-correct.
 """
 
+from typing import Any
+
 from ...catalog.models import Catalog, Column, Source, Table
 from .models import QueryIR
 from .operators import (
@@ -80,6 +82,8 @@ class IRValidator:
                         f"(allowed: {sorted(allowed)})"
                     )
 
+        self._reject_contradictory_ranges(ir)
+
         for i, col_id in enumerate(ir.group_by):
             self._require_column(columns_by_id, col_id, f"group_by[{i}]")
 
@@ -100,6 +104,36 @@ class IRValidator:
                     f"limit {ir.limit} exceeds hard cap {LIMIT_HARD_CAP}"
                 )
 
+    @staticmethod
+    def _reject_contradictory_ranges(ir: QueryIR) -> None:
+        """Reject disjoint BETWEEN ranges on the same column.
+
+        All filters in an IR are ANDed, so two non-overlapping BETWEEN ranges on one
+        column (e.g. Q1 2025 AND Q1 2026 on order_date) match no rows — a common LLM
+        slip when it means a multi-period comparison. Rejecting lets the planner's
+        re-prompt retry correct it (one spanning range + group-by the period, or one
+        query per period). Best-effort: incomparable values are skipped, not raised.
+        """
+        ranges: dict[str, list[tuple[Any, Any]]] = {}
+        for f in ir.filters:
+            if f.op == "between" and isinstance(f.value, list) and len(f.value) == 2:
+                ranges.setdefault(f.column_id, []).append((f.value[0], f.value[1]))
+        for col_id, rs in ranges.items():
+            if len(rs) < 2:
+                continue
+            try:
+                # AND of ranges = intersection [max(lo), min(hi)]; empty when lo > hi.
+                empty = max(lo for lo, _ in rs) > min(hi for _, hi in rs)
+            except TypeError:
+                continue  # incomparable value types — skip (fail-open)
+            if empty:
+                raise IRValidationError(
+                    f"filters place {len(rs)} non-overlapping BETWEEN ranges on column "
+                    f"{col_id!r}; because all filters are ANDed this matches no rows. "
+                    "For a multi-period comparison use a single spanning range and group "
+                    "by the period, or emit one query per period."
+                )
+
     def _validate_joins(
         self, ir: QueryIR, source: Source, columns_by_id: dict[str, Column]
     ) -> None:

src/tools/data_access.py

@@ -277,6 +277,8 @@ class DataAccessToolInvoker:
                 "row_count": result.row_count,
                 "truncated": result.truncated,
                 "elapsed_ms": result.elapsed_ms,
+                # Executed query for traceability (KM-691); None if unavailable.
+                "query": result.query,
             },
         )
 

src/traceability/__init__.py

@@ -0,0 +1,32 @@
+"""Traceability (KM-691): user-facing per-turn provenance.
+
+Captured by a per-request `TraceabilityScratchpad` inside `ChatHandler`, persisted
+through a `TraceabilityStore` before the `done` SSE event, and served by
+`GET /api/v1/traceability`. Kept separate from Langfuse observability
+(`src/observability/langfuse/`), which is engineering-only and PII-masked.
+"""
+
+from .schemas import (
+    PlanningInfo,
+    PlanStep,
+    ToolCallInfo,
+    TraceabilityPayload,
+)
+from .scratchpad import TraceabilityScratchpad, TraceabilityToolInvoker
+from .store import (
+    NullTraceabilityStore,
+    PostgresTraceabilityStore,
+    TraceabilityStore,
+)
+
+__all__ = [
+    "NullTraceabilityStore",
+    "PlanStep",
+    "PlanningInfo",
+    "PostgresTraceabilityStore",
+    "ToolCallInfo",
+    "TraceabilityPayload",
+    "TraceabilityScratchpad",
+    "TraceabilityStore",
+    "TraceabilityToolInvoker",
+]

src/traceability/schemas.py

@@ -0,0 +1,76 @@
+"""Traceability payload schemas (KM-691).
+
+User-facing provenance for one assistant turn: what the AI planned, which tools it
+called (with real inputs + outputs), and which data sources it read (with the
+executed query). One `TraceabilityPayload` is built per assistant `message_id`,
+stored as a JSONB row, and served by `GET /api/v1/traceability`.
+
+Distinct from Langfuse *observability* (`src/observability/langfuse/`): that is
+engineering-only, PII-masked (arg keys + row counts). Traceability shows the user
+their own data's provenance — real args, output previews, executed SQL — so it is
+NOT masked. Truncation caps (in `scratchpad.py`) bound the payload size instead.
+
+`thinking` is always `null` in v1 (our agents are plain chat completions with no
+native reasoning output; synthesizing it post-hoc would be unfaithful). The field
+stays in the payload so adding it later is contract-compatible.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class PlanStep(BaseModel):
+    """One CRISP-DM step the Planner ran (derived from an AnalysisRecord task)."""
+
+    step: int
+    stage: str
+    objective: str
+    status: str
+    tools_used: list[str] = Field(default_factory=list)
+
+
+class PlanningInfo(BaseModel):
+    """Planner output for a slow-path turn; `null` on every other turn type."""
+
+    goal_restated: str
+    assumptions: list[str] = Field(default_factory=list)
+    steps: list[PlanStep] = Field(default_factory=list)
+
+
+class ToolCallInfo(BaseModel):
+    """One tool invocation with its real input + output (both truncation-capped).
+
+    The field is spelled `input` in the wire contract (the FE reads it), which
+    shadows the `input` builtin — hence the alias + `populate_by_name` so callers
+    can pass `input=` on construction and `model_dump(by_alias=True)` emits `input`.
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    order: int
+    task_id: str | None = None
+    name: str
+    input_: dict[str, Any] = Field(default_factory=dict, alias="input")
+    output: dict[str, Any] = Field(default_factory=dict)
+    status: Literal["success", "error"] = "success"
+    error: str | None = None
+
+
+class TraceabilityPayload(BaseModel):
+    """The full provenance record for one assistant `message_id`."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    analysis_id: str
+    message_id: str
+    user_id: str  # ownership column on the row; the FE may ignore it
+    intent: str
+    generated_at: datetime
+    planning: PlanningInfo | None = None
+    thinking: str | None = None
+    tool_calls: list[ToolCallInfo] = Field(default_factory=list)
+    sources: list[dict[str, Any]] = Field(default_factory=list)

src/traceability/scratchpad.py

@@ -0,0 +1,230 @@
+"""Per-request traceability accumulator + tool-invoker wrapper (KM-691).
+
+`TraceabilityScratchpad` is a mutable, per-request blackboard that `ChatHandler`
+fills while answering a turn, then `build()`s into a `TraceabilityPayload` right
+before the `done` SSE event. `TraceabilityToolInvoker` wraps the real tool invoker
+so every tool call on the slow path / check branch records its full I/O into the
+scratchpad (mirrors `TracingToolInvoker` in `src/observability/langfuse/tracing.py`,
+whose name is taken — this one records real I/O, not masked metadata).
+
+Everything here is best-effort: recording must never break the user's answer.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+
+from src.middlewares.logging import get_logger
+
+from .schemas import PlanningInfo, PlanStep, ToolCallInfo, TraceabilityPayload
+
+logger = get_logger("traceability")
+
+# Truncation caps (bound the JSONB payload) — see plan §3.
+CAP_PREVIEW_ROWS = 5
+CAP_STR = 300
+# Executed queries get a higher cap than free strings: the SQL/query is the point of
+# the feature, and 300 chars mangles all but the smallest statements.
+CAP_QUERY = 2000
+
+
+def _truncate(obj: Any) -> Any:
+    """Recursively cap strings to CAP_STR and summarize embedded tool results.
+
+    A Pattern-A `analyze_*` input carries its upstream `retrieve_data` result as a
+    `ToolOutput` (a BaseModel). Left untouched that would re-embed the FULL upstream
+    table (all rows + the untruncated query) into the payload, defeating the caps —
+    so a tool-result-shaped model is summarized via `_output_to_dict` (preview ≤ 5
+    rows, `row_count` kept), and any other BaseModel is dumped then recursed.
+    """
+    if isinstance(obj, str):
+        return obj[:CAP_STR]
+    if isinstance(obj, BaseModel):
+        if hasattr(obj, "kind") and hasattr(obj, "rows"):  # a ToolOutput-shaped result
+            return _output_to_dict(obj)
+        return _truncate(obj.model_dump(mode="json"))
+    if isinstance(obj, dict):
+        return {k: _truncate(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_truncate(v) for v in obj]
+    return obj
+
+
+def _output_to_dict(output: Any) -> dict[str, Any]:
+    """Normalize a tool result (`ToolOutput` or a synth dict) to the wire shape:
+    kind/columns/row_count/preview/value/error, all truncation-capped."""
+    if isinstance(output, dict):
+        return _truncate(output)
+
+    kind = getattr(output, "kind", None)
+    result: dict[str, Any] = {"kind": kind}
+    rows = getattr(output, "rows", None)
+    if rows is not None:
+        result["row_count"] = len(rows)
+        columns = getattr(output, "columns", None)
+        if columns is not None:
+            result["columns"] = list(columns)
+        result["preview"] = [
+            [_truncate(cell) for cell in row] for row in rows[:CAP_PREVIEW_ROWS]
+        ]
+    value = getattr(output, "value", None)
+    if value is not None:
+        result["value"] = _truncate(value)
+    error = getattr(output, "error", None)
+    if error is not None:
+        result["error"] = _truncate(error)
+    return result
+
+
+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):
+        meta = output.get("meta")
+    else:
+        meta = getattr(output, "meta", None)
+    return meta if isinstance(meta, dict) else {}
+
+
+class TraceabilityScratchpad:
+    """Mutable per-request accumulator; `build()` freezes it into a payload."""
+
+    def __init__(self) -> None:
+        self.message_id: str | None = None  # set at handler entry; None => no flush
+        self.intent: str = "chat"  # default until the router classifies
+        self._planning: PlanningInfo | None = None
+        self._tool_calls: list[ToolCallInfo] = []
+        self._db_sources: list[dict[str, Any]] = []
+        self._doc_sources: list[dict[str, Any]] = []
+        self._doc_seen: set[tuple[Any, Any]] = set()
+
+    def set_intent(self, intent: str) -> None:
+        self.intent = intent
+
+    def record_tool_call(
+        self,
+        name: str,
+        args: dict[str, Any],
+        output: Any,
+        task_id: str | None = None,
+    ) -> None:
+        """Append one tool call (input + normalized output). For `retrieve_data`,
+        also derive a database source from the args + executed query in `meta`."""
+        out_dict = _output_to_dict(output)
+        status = "error" if out_dict.get("kind") == "error" else "success"
+        self._tool_calls.append(
+            ToolCallInfo(
+                order=len(self._tool_calls) + 1,
+                task_id=task_id,
+                name=name,
+                input=_truncate(dict(args)),
+                output=out_dict,
+                status=status,
+                error=out_dict.get("error"),
+            )
+        )
+        if name == "retrieve_data":
+            self._record_db_source(output)
+
+    def _record_db_source(self, output: Any) -> None:
+        # retrieve_data's args are {"ir": ...}; the reliable source_id/table/query
+        # live on the tool OUTPUT meta (see tools/data_access.py::_retrieve_data).
+        meta = _meta_of(output)
+        query = meta.get("query")
+        table = meta.get("table_name") or meta.get("table_id")
+        self._db_sources.append({
+            "type": "database",
+            "source_id": meta.get("source_id"),
+            "name": table,
+            "query": query[:CAP_QUERY] if isinstance(query, str) else None,
+            "detail": {"table": table, "row_count": meta.get("row_count")},
+        })
+
+    def set_planning_from_record(self, record: Any) -> None:
+        """Map an `AnalysisRecord` (goal_restated + tasks_run) to `PlanningInfo`."""
+        try:
+            steps = [
+                PlanStep(
+                    step=i + 1,
+                    stage=str(getattr(task, "stage", "")),
+                    objective=getattr(task, "objective", ""),
+                    status=str(getattr(task, "status", "")),
+                    tools_used=list(getattr(task, "tools_used", []) or []),
+                )
+                for i, task in enumerate(getattr(record, "tasks_run", []) or [])
+            ]
+            self._planning = PlanningInfo(
+                goal_restated=getattr(record, "goal_restated", "") or "",
+                assumptions=[],  # AnalysisRecord carries no assumptions field (honest: empty)
+                steps=steps,
+            )
+        except Exception as exc:  # never break the answer on a mapping slip
+            logger.warning("traceability planning mapping failed", error=str(exc))
+
+    def add_document_sources(self, raw_chunks: Any, query: str) -> None:
+        """Dedupe retrieved chunks by (document_id, page_label) into document
+        sources, stamped with the query. Sole source of document provenance now
+        that the stream no longer emits sources (KM-691)."""
+        for item in raw_chunks or []:
+            if hasattr(item, "metadata"):
+                data = item.metadata.get("data", {})
+            elif isinstance(item, dict):
+                data = item
+            else:
+                continue
+            key = (data.get("document_id"), data.get("page_label"))
+            if key == (None, None) or key in self._doc_seen:
+                continue
+            self._doc_seen.add(key)
+            source: dict[str, Any] = {
+                "type": "document",
+                "document_id": data.get("document_id"),
+                "filename": data.get("filename", "Unknown"),
+                "page_label": data.get("page_label"),
+                "query": _truncate(query),
+            }
+            snippet = data.get("snippet") or data.get("content") or data.get("text")
+            if isinstance(snippet, str):
+                source["snippet"] = snippet[:CAP_STR]
+            score = data.get("score")
+            if score is not None:
+                source["score"] = score
+            self._doc_sources.append(source)
+
+    def build(self, analysis_id: str, user_id: str, message_id: str) -> TraceabilityPayload:
+        """Freeze the accumulated state into a `TraceabilityPayload`."""
+        from datetime import UTC, datetime
+
+        return TraceabilityPayload(
+            analysis_id=analysis_id,
+            message_id=message_id,
+            user_id=user_id,
+            intent=self.intent,
+            generated_at=datetime.now(UTC),
+            planning=self._planning,
+            thinking=None,
+            tool_calls=list(self._tool_calls),
+            sources=self._doc_sources + self._db_sources,
+        )
+
+
+class TraceabilityToolInvoker:
+    """Wraps a ToolInvoker to record each call's full I/O into a scratchpad.
+
+    Implements the ToolInvoker protocol (`async invoke(tool_name, args)`). Recording
+    is never-throw so a trace slip can't break the tool run. Distinct from
+    `TracingToolInvoker` (Langfuse, masked-metadata-only) — that name is taken.
+    """
+
+    def __init__(self, inner: Any, pad: TraceabilityScratchpad) -> None:
+        self._inner = inner
+        self._pad = pad
+
+    async def invoke(self, tool_name: str, args: dict[str, Any]) -> Any:
+        out = await self._inner.invoke(tool_name, args)
+        try:
+            self._pad.record_tool_call(tool_name, args, out)
+        except Exception as exc:  # never break the tool run
+            logger.warning("traceability tool record failed", tool=tool_name, error=str(exc))
+        return out

src/traceability/store.py

@@ -0,0 +1,110 @@
+"""TraceabilityStore — the seam the chat pipeline persists provenance through (KM-691).
+
+`ChatHandler` (and the v2 chat endpoint's greeting/cache branches) flush one
+`TraceabilityPayload` per assistant turn through this seam, right before the `done`
+SSE event; `GET /api/v1/traceability` reads it back by (analysis_id, message_id).
+
+- `NullTraceabilityStore` logs and stores nothing (tests / disabled persistence).
+- `PostgresTraceabilityStore` writes one `message_traceability` row per turn
+  (dedorch, `AsyncSessionLocal`), mirroring `PostgresReportInputStore`.
+
+`save` must NEVER raise on the caller's path — a persistence failure must not break
+the user's answer. `get` is the endpoint read and returns `None` on a miss (→ 404).
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert
+
+from src.db.postgres.connection import AsyncSessionLocal
+from src.db.postgres.models import MessageTraceabilityRow
+from src.middlewares.logging import get_logger
+
+from .schemas import TraceabilityPayload
+
+logger = get_logger("traceability_store")
+
+
+@runtime_checkable
+class TraceabilityStore(Protocol):
+    """Persist + read one provenance record per assistant `message_id`.
+
+    `save` must never raise on the caller's path. `get` returns the payload for one
+    turn, or `None` if none exists yet (the turn is still running or the id is unknown).
+    """
+
+    async def save(self, payload: TraceabilityPayload) -> None: ...
+
+    async def get(
+        self, analysis_id: str, message_id: str
+    ) -> TraceabilityPayload | None: ...
+
+
+class NullTraceabilityStore:
+    """No-op store: logs the payload, persists nothing. Reads return `None`."""
+
+    async def save(self, payload: TraceabilityPayload) -> None:
+        logger.info(
+            "traceability produced (not persisted — NullTraceabilityStore)",
+            message_id=payload.message_id,
+            intent=payload.intent,
+            n_tool_calls=len(payload.tool_calls),
+        )
+
+    async def get(
+        self, analysis_id: str, message_id: str
+    ) -> TraceabilityPayload | None:
+        return None
+
+
+class PostgresTraceabilityStore:
+    """Writes/reads `message_traceability` jsonb rows. Upsert on `message_id`."""
+
+    async def save(self, payload: TraceabilityPayload) -> None:
+        try:
+            data = payload.model_dump(mode="json", by_alias=True)
+            async with AsyncSessionLocal() as session:
+                stmt = insert(MessageTraceabilityRow).values(
+                    message_id=payload.message_id,
+                    analysis_id=payload.analysis_id,
+                    user_id=payload.user_id,
+                    intent=payload.intent,
+                    data=data,
+                )
+                # Idempotent: a re-flushed turn overwrites its own row.
+                stmt = stmt.on_conflict_do_update(
+                    index_elements=[MessageTraceabilityRow.message_id],
+                    set_={"data": stmt.excluded.data, "intent": stmt.excluded.intent},
+                )
+                await session.execute(stmt)
+                await session.commit()
+            logger.info(
+                "traceability persisted",
+                message_id=payload.message_id,
+                analysis_id=payload.analysis_id,
+                intent=payload.intent,
+            )
+        except Exception as exc:  # never break the user's answer
+            logger.error(
+                "traceability persist failed",
+                message_id=payload.message_id,
+                error=str(exc),
+            )
+
+    async def get(
+        self, analysis_id: str, message_id: str
+    ) -> TraceabilityPayload | None:
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(
+                select(MessageTraceabilityRow.data).where(
+                    MessageTraceabilityRow.message_id == message_id,
+                    MessageTraceabilityRow.analysis_id == analysis_id,
+                )
+            )
+            row = result.scalar_one_or_none()
+        if row is None:
+            return None
+        return TraceabilityPayload.model_validate(row)