[KM-691] Develop 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
 
@@ -57,7 +57,7 @@ Common event types:
 | `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`.
@@ -357,13 +357,15 @@ Response `404`:
 }
 ```
 
-## Observability
+## Traceability
 
-### `GET /api/v1/observability`
+> 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.
 
-Returns Responsible AI provenance for one assistant answer.
+### `GET /api/v1/traceability`
 
-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.
+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. 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,7 +477,8 @@ class ChatHandler:
                 return
         elif intent == "check":
             try:
-                invoker = self._get_check_invoker(user_id)
+                # 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.
@@ -453,6 +489,7 @@ class ChatHandler:
                 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
@@ -504,6 +541,7 @@ 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
@@ -537,6 +575,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": ""}
 
     # ------------------------------------------------------------------
@@ -554,7 +594,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).
 
@@ -583,6 +627,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
@@ -595,6 +643,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,
@@ -604,6 +676,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.
 
@@ -621,7 +694,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
@@ -688,9 +761,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": ""}
 
 

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
@@ -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/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/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,213 @@
+"""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 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
+
+
+def _truncate(obj: Any) -> Any:
+    """Recursively cap any string to CAP_STR; leave numbers/None untouched."""
+    if isinstance(obj, str):
+        return obj[:CAP_STR]
+    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": _truncate(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 (mirrors `chat_handler._build_sources`), stamped with the query."""
+        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)