[KM-691] Make sources trace-authoritative; drop stream sources

Sources move fully to GET /api/v1/traceability. The SSE sources event
now always emits [] (kept for backward-compat); removed the dead
_build_sources helper and the structured-path source TODO. Updates the
contract + v2 chat docstring accordingly.

API_CONTRACT_BE_PYTHON.md

@@ -51,7 +51,7 @@ 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`. |
@@ -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

src/agents/chat_handler.py

@@ -547,14 +547,11 @@ class ChatHandler:
         # 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).
@@ -750,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
@@ -798,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/api/v2/chat.py

@@ -93,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()

src/traceability/scratchpad.py

@@ -148,7 +148,8 @@ class TraceabilityScratchpad:
 
     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."""
+        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", {})