/fix planner and report

API_CONTRACT_BE_PYTHON.md

@@ -30,6 +30,8 @@ The frontend uses this service during the analysis conversation flow:
 | `POST` | `/api/v1/tools/help` | Stream contextual help for the current analysis conversation. |
 | `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}/records` | List analysis records for report curation (added 2026-07-09). |
+| `GET` | `/api/v1/tools/report/{analysis_id}/readiness` | Report-readiness signal for the Generate-Report button (added 2026-07-09). |
 | `GET` | `/api/v1/tools/report/{analysis_id}/{version}` | Retrieve one report version. |
 | `GET` | `/api/v1/traceability` | Retrieve provenance for one assistant answer. |
 
@@ -39,7 +41,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 traceability.
+- `message_id`: assistant answer identifier used to correlate chat streaming with traceability. **Server-minted, never accepted from the client.** **Updated 2026-07-09:** it is a UUID string (e.g. `77f06761-0fdf-4cc5-84f8-5f81bcbb6f84`), matching the shape of Go's `analyses_messages.id`. The `msg_…` values in the examples below are illustrative placeholders only.
 
 ### Server-Sent Events
 
@@ -71,7 +73,6 @@ Request body:
 {
   "user_id": "u_1a2b3c",
   "analysis_id": "an_42",
-  "message_id": "msg_88f1",
   "message": "What were total sales by region last quarter?"
 }
 ```
@@ -82,7 +83,7 @@ Fields:
 | --- | --- | --- |
 | `user_id` | Yes | User identifier. |
 | `analysis_id` | Yes | Analysis conversation identifier. |
-| `message_id` | No | Assistant answer id for traceability correlation. If omitted, Python returns one in `done`. |
+| ~~`message_id`~~ | — | **Updated 2026-07-09:** not a request field. Python always mints the id server-side and returns it on `done`; any caller-sent value is ignored (server-authoritative — open-Q #1). |
 | `message` | Yes | User message text. |
 
 Response: `text/event-stream`.
@@ -218,11 +219,13 @@ Query params:
 | --- | --- | --- |
 | `analysis_id` | Yes | Analysis identifier. |
 | `user_id` | Yes | User identifier. |
+| `exclude_record_ids` | No | Record ids to leave out of this version (repeat the param per id). Added 2026-07-09; get ids from `GET /tools/report/{analysis_id}/records`. Excluded runs are listed in the report's "Excluded Analyses" section. Excluding every substantive record returns `409`. |
 
 Example:
 
 ```text
 POST /api/v1/tools/report?analysis_id=an_42&user_id=u_1a2b3c
+POST /api/v1/tools/report?analysis_id=an_42&user_id=u_1a2b3c&exclude_record_ids=rec_a1&exclude_record_ids=rec_c3
 ```
 
 Status codes:
@@ -251,6 +254,20 @@ Response `201`:
   },
   "record_ids": ["rec_a1", "rec_b2"],
   "executive_summary": "Revenue is concentrated in the Central region (38% of total). The West was the only region to contract, down 12% QoQ, the main driver of the Q1 dip.",
+  "bq_answers": [
+    {
+      "question": "Which regions contribute most to total revenue?",
+      "answer": "The Central region leads with 38% of total revenue.",
+      "status": "answered",
+      "record_ids": ["rec_a1"]
+    },
+    {
+      "question": "Did any region decline quarter-over-quarter?",
+      "answer": "Yes — the West region fell 12% QoQ.",
+      "status": "answered",
+      "record_ids": ["rec_b2"]
+    }
+  ],
   "findings": [
     {
       "text": "Central region contributed 38% of total revenue, the largest share.",
@@ -275,6 +292,23 @@ Response `201`:
       "record_ids": ["rec_b2"]
     }
   ],
+  "unresolved": [
+    {
+      "text": "Correlate churn with tenure — churn column not found in the source.",
+      "record_ids": ["rec_d4"]
+    }
+  ],
+  "excluded": [],
+  "evidence_tables": {
+    "rec_a1": [
+      {
+        "title": "Aggregate revenue by region",
+        "columns": ["region", "total_revenue"],
+        "rows": [["Central", "18321"], ["West", "9954"]],
+        "truncated": false
+      }
+    ]
+  },
   "data_sources": [
     {
       "source_id": "src_sales_db",
@@ -315,6 +349,13 @@ Response `409`:
 }
 ```
 
+Report v2 fields (added 2026-07-09; all default-empty, so older stored reports read back unchanged):
+
+- `bq_answers` — one entry per business question. `status` is `answered` | `partial` | `unanswered`; `record_ids` cite the backing analyses. Written in the analysis's language (Indonesian objective → Indonesian answers).
+- `unresolved` — runs that were attempted but produced no usable evidence (every `analyze_*` step failed). Not part of the findings body.
+- `excluded` — runs the caller excluded via `exclude_record_ids`.
+- `evidence_tables` — `record_id` → small result tables copied from the run's stored outputs (max 3 tables per record, max 10 rows each; `truncated: true` when rows were capped). Rendered as markdown tables under the matching Key Findings group in `rendered_markdown`.
+
 Precondition:
 
 - Reports require at least one completed analysis record for the session.
@@ -345,6 +386,48 @@ Response `200`:
 
 If no reports exist, returns `[]`.
 
+### `GET /api/v1/tools/report/{analysis_id}/records` (added 2026-07-09)
+
+Lists the persisted analysis runs a report would be built from, oldest first. The frontend shows this before generating so the user can deselect runs; the chosen ids go to `POST /tools/report` as `exclude_record_ids`.
+
+Response `200`:
+
+```json
+[
+  {
+    "record_id": "rec_a1",
+    "goal_restated": "Rank regions by total revenue",
+    "created_at": "2026-06-30T08:55:02Z",
+    "substantive": true,
+    "findings_count": 2
+  },
+  {
+    "record_id": "rec_d4",
+    "goal_restated": "Correlate churn with tenure",
+    "created_at": "2026-06-30T09:01:47Z",
+    "substantive": false,
+    "findings_count": 1
+  }
+]
+```
+
+`substantive: false` means no `analyze_*` step succeeded — that run is listed in the report's `unresolved` JSON field rather than the findings body. (Since 2026-07-09 the rendered markdown is compact and no longer includes "Attempted, Unresolved" / "Notes & Limitations" / "How This Was Analyzed" sections; the JSON fields `unresolved` / `caveats` / `open_questions` / `method_steps` are unchanged.) If no runs exist, returns `[]`.
+
+### `GET /api/v1/tools/report/{analysis_id}/readiness` (added 2026-07-09)
+
+Deterministic report-readiness signal for the Generate-Report button — the same producer as Help's readiness signal, including the advisory delta-since-report check, so the button, Help, and this endpoint never disagree.
+
+Response `200`:
+
+```json
+{
+  "ready": false,
+  "missing": ["a new analysis since the last report"]
+}
+```
+
+Note: `POST /tools/report` itself only enforces the floor (`at least one completed analysis`) — a new version is always allowed. The delta gap in `missing` is a soft warning the frontend can surface ("nothing new since the last report") without blocking the button.
+
 ### `GET /api/v1/tools/report/{analysis_id}/{version}`
 
 Returns one report version. Shape is the same as the `201` response from `POST /api/v1/tools/report`.

DEV_PLAN.md

@@ -34,7 +34,7 @@ the endpoint contract *before* coding the tools. Status legend: ⬜ not started
 | **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. |
+| **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. **Updated 2026-07-09 (pr/13):** id format changed from `msg_<hex>` to a canonical UUID string (`str(uuid.uuid4())`, both mint sites) to mirror Go's `analyses_messages.id` shape — the value is still independently Python-minted (not the real row id), only format-compatible for a future swap. |
 | **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. |
@@ -46,6 +46,26 @@ minter, stream-only). The **Phase 3 traceability build** — scratchpad + `GET /
 
 ---
 
+## 0.5. pr/13 sprint — agent-quality fixes (2026-07-08 live-test review)
+
+Findings from the scoped live sessions (mining analysis, 2026-07-07/08 traces): the planner
+force-mapped absent measures (`pa` aliased as "revenue"), top-N ranked raw rows (duplicate models),
+`analyze_trend` collapsed integer months into a single 1970-01 bucket, an invalid grouped IR reached
+Postgres, failed retrievals wrote all-null traceability sources, and numeric catalog samples arrive
+base64-mangled from Go. Fix tasks (same status legend as §0):
+
+| # | Task | Owner | Status | Note |
+|---|---|---|---|---|
+| Q1 | IR validator: reject bare selects under `group_by` (planner retry self-corrects) | Rifqi | ✅ | `query/ir/validator.py` |
+| Q2 | Planner **infeasible** path: `TaskList.infeasible_reason` + deterministic EN/ID data-gap reply | Rifqi | ✅ | schemas/validator/coordinator/refusals + planner.md "When the catalog cannot answer"; refusal wording → Rifqi to review |
+| Q3 | `analyze_trend`: integer year/month handling (epoch-parse bug) | Rifqi | ✅ | `temporal.py` + 5 local tests |
+| Q4 | Planner few-shots: top-N (Example G) + infeasible (Example H) + entity-vs-row ranking rule | Rifqi | ✅ | live-tested 2026-07-08: backlog top-3 correct via single-IR group+sum; "best PA performance" correct in-process (avg-per-model, assumption recorded). Stale-server trace was a false alarm |
+| Q5 | Catalog numeric `sample_values` base64-decode stopgap (`catalog/sample_decode.py`) | Rifqi | ✅ | self-disabling; **primary fix = Go marshaling — DDL-free handoff to Harry** |
+| Q6 | Traceability null-source suppression + `check_data` `-1` row-count hiding | Rifqi | ✅ | `scratchpad.py` / `data_access.py` |
+| Q7 | `analyze_merge` two-table combine tool (unblocks "worst A + biggest B" questions) | tool owner | ✅ | tool shipped by Sofia (8abf635, KM-703); planner slice done 2026-07-09: `_validate_data_source` guards `data_right`, two-retrieve→merge few-shot (Example I), planner.md "Two measures per entity" bullet |
+| Q8 | Report v2: business-question answer section, unresolved/excluded sections, evidence tables from `results_snapshot`, caveat dedupe, single language | Rifqi/Sofhia | ✅ | done 2026-07-09: still exactly ONE LLM call (extended to also draft `bq_answers`, index-based record refs, deterministic fallback = v1 behavior); evidence tables from table-kind outputs (≤3/record, ≤10 rows, ≤8 cols, `check_*` skipped); reply language via `detect_reply_language` on objective+BQs; verified in-process against live analysis 935a091e |
+| Q9 | Record-curation endpoint (`GET …/records` + `exclude_record_ids`) + readiness GET for the FE delta guard | Rifqi ↔ FE | ✅ | done 2026-07-09: `GET /tools/report/{analysis_id}/records` + `/readiness` (registered before `/{version}` — int-coercion route-order trap), `exclude_record_ids` on POST; contract updated same change; FE wiring pending (Rifqi → FE) |
+
 ## 1. The direction change (locked decisions from 2026-06-24)
 
 1. **"Problem statement" is replaced by two user-entered fields: `objective` + `business_questions`.**

REPO_STATUS.md

@@ -2,7 +2,7 @@
 
 **Audience:** teammates onboarding onto the Python repo (`Agentic-Service-Data-Eyond-Catalog`).
 **Scope:** what the code does **right now** (branch `pr/4`, ticket KM-652). Describes current state only — no roadmap or to-dos.
-**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
+**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Agent-quality fixes 2026-07-08 (pr/13):** from the scoped live-test review — the planner gains an explicit **infeasible** outcome (`TaskList.infeasible_reason` → deterministic EN/ID data-gap reply via `refusals.data_gap_message`; no more force-mapping absent measures like `pa` AS "revenue"), the IR validator rejects bare selects under `group_by` (self-corrects via the planner retry), `analyze_trend` handles integer year/month columns (was collapsing every row into one 1970-01 bucket), planner few-shots add top-N (Example G) + infeasible (Example H), numeric catalog `sample_values` are base64-decoded at read (`catalog/sample_decode.py` — stopgap for Go's byte-marshaling; primary fix is Go-side), traceability no longer emits null source rows for failed retrievals, and `check_data` hides `-1` row counts. **Report v2 + analyze_merge planner support 2026-07-09 (pr/13):** Sofia's `analyze_merge` tool (8abf635, KM-703) is now planner-supported (`_validate_data_source` guards `data_right`, two-retrieve→merge few-shot Example I, planner.md "Two measures per entity" rule); the report gains per-business-question answers (`bq_answers` — drafted by the SAME single LLM call, index-based record refs, deterministic fallback unchanged), "Attempted, Unresolved" + "Excluded Analyses" sections (failed runs are no longer silently dropped), evidence tables copied from `results_snapshot` (table-kind outputs, ≤3/record ≤10 rows ≤8 cols, `check_*` skipped), normalized caveat dedupe with caps (12/10), and single-language output via `detect_reply_language`; the report surface adds `GET /tools/report/{analysis_id}/records` (curation list), `GET …/readiness` (FE delta guard), and `exclude_record_ids` on POST — see API_CONTRACT_BE_PYTHON.md. **Report compaction 2026-07-09 (pr/13):** the rendered markdown drops the "Notes & Limitations", "Attempted, Unresolved", and "How This Was Analyzed" sections (team decision — compact report; render blocks commented out in `report/generator.py`, not deleted). The JSON body keeps `caveats`/`open_questions`/`unresolved`/`method_steps` and the curation/records endpoints are unchanged. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
 the **Go source** (`Orchestrator-Agent-Service`), not its docs. The Go service has moved well past its
 own (uncommitted, stale) design docs: it now hosts the **dedorch SQL migrations** in-repo and a full
 **`/api/v1/analyses` + `/api/v1/skills`** REST surface. Go does **not** call Python yet — those skills

src/agents/planner/examples.py

@@ -524,6 +524,212 @@ _EXAMPLE_F = TaskList(
 )
 
 
+# --------------------------------------------------------------------------- #
+# Example G — top-N ranking.
+# "Top 3 product categories by total revenue."
+# Shows: top-N is ONE retrieve_data query — group by the entity, aggregate the
+# measure with an alias, order by that alias, limit N. NEVER a bare
+# order-by-measure + limit (that ranks raw rows, so the same entity can appear
+# twice — observed in production: "top 3 models" returned one model twice).
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_G = TaskList(
+    plan_id="example_g",
+    goal_restated="Rank product categories by total revenue and return the top 3.",
+    assumptions=[],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes category and revenue.",
+            tool_calls=[ToolCall(tool="check_data", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria=(
+                "Produced the orders table schema; category and revenue columns "
+                "are present."
+            ),
+            depends_on=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Aggregate revenue per category, rank descending, keep the top 3.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_data",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_category", "alias": "category"},
+                                {
+                                    "kind": "agg",
+                                    "fn": "sum",
+                                    "column_id": "c_revenue",
+                                    "alias": "total_revenue",
+                                },
+                            ],
+                            "group_by": ["c_category"],
+                            "order_by": [{"column_id": "total_revenue", "dir": "desc"}],
+                            "limit": 3,
+                        }
+                    },
+                )
+            ],
+            expected_output="top3_categories",
+            success_criteria=(
+                "Produced at most 3 rows, one distinct category each, ranked by "
+                "total revenue."
+            ),
+            depends_on=["t1"],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+# --------------------------------------------------------------------------- #
+# Example H — infeasible question (see planner.md "When the catalog cannot
+# answer"). "What is our customer churn rate?" against a sales catalog with no
+# subscription/churn data: no task list is forced onto unrelated columns;
+# instead `infeasible_reason` states the gap + the nearest available data.
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_H = TaskList(
+    plan_id="example_h",
+    goal_restated="Measure the customer churn rate.",
+    assumptions=[],
+    open_questions=[],
+    tasks=[],
+    infeasible_reason=(
+        "The connected source has no churn or subscription-status data — the "
+        "orders table only carries order-level category, revenue, quantity, and "
+        "dates. Nearest available analyses: repeat-purchase behaviour or revenue "
+        "per customer over time."
+    ),
+)
+
+
+# --------------------------------------------------------------------------- #
+# Example I — combine two measures per entity (KM-703).
+# "Which category has both the highest revenue and the highest average order
+# quantity?" Shows: each measure is computed in its OWN grouped retrieve_data
+# task (a "${t<id>}" placeholder resolves to a task's LAST output, so the two
+# retrievals must be separate tasks), then analyze_merge aligns them on the
+# shared entity alias. The merged table answers "both A and B" questions that
+# a single query cannot express.
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_I = TaskList(
+    plan_id="example_i",
+    goal_restated=(
+        "Identify the product category with both the highest total revenue and the "
+        "highest average order quantity."
+    ),
+    assumptions=[],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes category, revenue, and quantity.",
+            tool_calls=[ToolCall(tool="check_data", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria=(
+                "Produced the orders table schema; category, revenue, and quantity "
+                "columns are present."
+            ),
+            depends_on=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Total revenue per category.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_data",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_category", "alias": "category"},
+                                {
+                                    "kind": "agg",
+                                    "fn": "sum",
+                                    "column_id": "c_revenue",
+                                    "alias": "total_revenue",
+                                },
+                            ],
+                            "group_by": ["c_category"],
+                        }
+                    },
+                )
+            ],
+            expected_output="revenue_per_category",
+            success_criteria="Produced one total-revenue row per category.",
+            depends_on=["t1"],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t3",
+            stage="data_preparation",
+            objective="Average order quantity per category.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_data",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_category", "alias": "category"},
+                                {
+                                    "kind": "agg",
+                                    "fn": "avg",
+                                    "column_id": "c_quantity",
+                                    "alias": "avg_quantity",
+                                },
+                            ],
+                            "group_by": ["c_category"],
+                        }
+                    },
+                )
+            ],
+            expected_output="quantity_per_category",
+            success_criteria="Produced one average-quantity row per category.",
+            depends_on=["t1"],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t4",
+            stage="evaluation",
+            objective="Align both measures per category to find the category leading on both.",
+            tool_calls=[
+                ToolCall(
+                    tool="analyze_merge",
+                    args={
+                        "data": "${t2}",
+                        "data_right": "${t3}",
+                        "on": ["category"],
+                    },
+                )
+            ],
+            expected_output="combined_measures",
+            success_criteria=(
+                "Produced one row per category carrying both total_revenue and "
+                "avg_quantity."
+            ),
+            depends_on=["t2", "t3"],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+
 EXAMPLES: list[tuple[str, TaskList]] = [
     ("Which product categories drove last quarter's revenue?", _EXAMPLE_A),
     ("How has monthly revenue trended by region this year, and what's unusual?", _EXAMPLE_B),
@@ -531,6 +737,13 @@ EXAMPLES: list[tuple[str, TaskList]] = [
     ("What is the average and total order value per region?", _EXAMPLE_D),
     ("Total revenue for the East and West regions, counting orders of at least 100.", _EXAMPLE_E),
     ("Give me the summary statistics for order revenue and quantity.", _EXAMPLE_F),
+    ("Which 3 product categories have the best revenue performance?", _EXAMPLE_G),
+    ("What is our customer churn rate?", _EXAMPLE_H),
+    (
+        "Which product category has both the highest revenue and the highest average "
+        "order quantity?",
+        _EXAMPLE_I,
+    ),
 ]
 
 

src/agents/planner/schemas.py

@@ -58,3 +58,18 @@ class TaskList(BaseModel):
     assumptions: list[str] = Field(default_factory=list)
     open_questions: list[str] = Field(default_factory=list)
     tasks: list[Task] = Field(default_factory=list)
+    # Infeasible sentinel (planner.md "When the catalog cannot answer"): set with
+    # an EMPTY `tasks` list when no catalog column plausibly holds the requested
+    # measure/entity. Explains what is missing and names the nearest available
+    # data. The coordinator renders it as an honest data-gap answer instead of
+    # running the pipeline — the alternative was the planner force-mapping
+    # unrelated columns (observed: `pa` aliased as "revenue").
+    infeasible_reason: str | None = Field(
+        None,
+        description=(
+            "Set ONLY when the question cannot be answered from the catalog: no "
+            "column plausibly holds the requested measure or entity. State what "
+            "is missing and the nearest data that IS available. Leave tasks "
+            "empty when set."
+        ),
+    )

src/agents/planner/validator.py

@@ -61,6 +61,15 @@ class PlannerValidator:
     ) -> None:
         tasks = task_list.tasks
 
+        # Infeasible sentinel (planner.md "When the catalog cannot answer"): an
+        # empty plan carrying `infeasible_reason` is a VALID outcome — the
+        # coordinator renders it as an honest data-gap answer instead of the
+        # planner force-mapping the question onto unrelated columns. A non-empty
+        # plan keeps normal validation and the reason is ignored (a real plan
+        # wins over a hedge).
+        if task_list.infeasible_reason and not tasks:
+            return
+
         # Check 6 — plan non-empty and within the task cap.
         if not tasks:
             raise PlannerValidationError("plan is empty: at least one task is required")
@@ -211,24 +220,28 @@ class PlannerValidator:
         requested columns. Resolving points at the referenced task's representative
         output — its last tool call (matches TaskRunner's `outputs[-1]`).
         """
-        data_arg = call.args.get("data")
-        if not isinstance(data_arg, str):
-            return
-        match = PLACEHOLDER_RE.fullmatch(data_arg.strip())
-        if not match:
-            return
-        ref_task = tasks_by_id.get(match.group(1))
-        if ref_task is None or not ref_task.tool_calls:
-            return  # a dangling placeholder is reported by the DAG check
-        ref_tool = ref_task.tool_calls[-1].tool
-        ref_spec = registry.get(ref_tool)
-        if ref_spec is not None and ref_spec.category in _NON_DATA_SOURCE_CATEGORIES:
-            raise PlannerValidationError(
-                f"task {task_id}: tool {call.tool!r} takes its 'data' from task "
-                f"{match.group(1)} ({ref_tool!r}, category {ref_spec.category!r}), "
-                "which produces metadata/documents — not analyzable data rows. Feed "
-                "analyze_* from a data-producing tool (e.g. retrieve_data)."
-            )
+        # `data_right` is analyze_merge's second table input (KM-703) — same
+        # Pattern A handoff, so it gets the same guard.
+        for arg_name in ("data", "data_right"):
+            data_arg = call.args.get(arg_name)
+            if not isinstance(data_arg, str):
+                continue
+            match = PLACEHOLDER_RE.fullmatch(data_arg.strip())
+            if not match:
+                continue
+            ref_task = tasks_by_id.get(match.group(1))
+            if ref_task is None or not ref_task.tool_calls:
+                continue  # a dangling placeholder is reported by the DAG check
+            ref_tool = ref_task.tool_calls[-1].tool
+            ref_spec = registry.get(ref_tool)
+            if ref_spec is not None and ref_spec.category in _NON_DATA_SOURCE_CATEGORIES:
+                raise PlannerValidationError(
+                    f"task {task_id}: tool {call.tool!r} takes its {arg_name!r} from "
+                    f"task {match.group(1)} ({ref_tool!r}, category "
+                    f"{ref_spec.category!r}), which produces metadata/documents — not "
+                    "analyzable data rows. Feed analyze_* from a data-producing tool "
+                    "(e.g. retrieve_data)."
+                )
 
     @staticmethod
     def _validate_dag(tasks_by_id: dict, id_set: set[str]) -> None:

src/agents/refusals.py

@@ -56,6 +56,39 @@ _BLOCKED = {
 }
 
 
+# Data-gap: the planner judged the bound sources cannot answer the question
+# (planner.md "When the catalog cannot answer"). Deterministic wrapper on
+# purpose — the model that declined to plan is not re-asked to prose it up.
+# Keyed on the pipeline's reply_language ("Indonesian"/"English"), not marker
+# detection: the upstream language decision is authoritative here.
+_DATA_GAP = {
+    "en": (
+        "I can't answer that from the data sources connected to this analysis. "
+        "{reason}You can bind a source that holds this data, or ask me what's "
+        "available (try /help or \"what data do I have?\")."
+    ),
+    "id": (
+        "Saya tidak bisa menjawab itu dari sumber data yang terhubung ke "
+        "analisis ini. {reason}Anda bisa menambahkan sumber yang memuat data "
+        "tersebut, atau tanyakan data apa yang tersedia (coba /help atau "
+        "\"data apa yang saya punya?\")."
+    ),
+}
+
+
+def data_gap_message(reason: str | None, reply_language: str | None = None) -> str:
+    """Answer for an infeasible analysis: the bound sources lack the asked-for data.
+
+    `reason` is the planner's `infeasible_reason` (may be None/empty);
+    `reply_language` is the pipeline's detected language ("Indonesian"/"English").
+    """
+    detail = (reason or "").strip()
+    if detail and not detail.endswith((".", "!", "?")):
+        detail += "."
+    lang = "id" if reply_language == "Indonesian" else "en"
+    return _DATA_GAP[lang].format(reason=f"{detail} " if detail else "")
+
+
 def out_of_scope_message(message: str) -> str:
     """Refusal for a benign but out-of-scope request (the `out_of_scope` intent)."""
     return _OUT_OF_SCOPE["id" if _is_indonesian(message) else "en"]

src/agents/report/generator.py

@@ -24,13 +24,17 @@ from langchain_openai import AzureChatOpenAI
 
 from src.middlewares.logging import get_logger
 
+from ..language import detect_reply_language
 from ..slow_path.schemas import AnalysisRecord, TaskSummary
 from .errors import ReportError
 from .readiness import has_successful_analysis
 from .schemas import (
     AnalysisReport,
     AttributedNote,
+    BQAnswerDraft,
+    BusinessQuestionAnswer,
     DataSourceRef,
+    EvidenceTable,
     ProblemStatement,
     ReportFinding,
     ReportSummaryNarrative,
@@ -40,6 +44,15 @@ logger = get_logger("report_generator")
 
 _FALLBACK_SUMMARY = "Automated summary unavailable — see the findings below."
 
+# Caps keeping the deterministic sections readable on multi-record analyses.
+_MAX_CAVEATS = 12
+_MAX_OPEN_QUESTIONS = 10
+_EVIDENCE_MAX_ROWS = 10
+_EVIDENCE_MAX_TABLES = 3  # per record
+# Wider tables are raw analysis *inputs* (e.g. a 19-column correlation pull), not
+# presentable evidence — grouped/top-N/merge results are always narrow.
+_EVIDENCE_MAX_COLS = 8
+
 # CRISP-DM phases in narrative order, with human labels for the method appendix.
 _STAGE_LABELS: list[tuple[str, str]] = [
     ("data_understanding", "Data understanding"),
@@ -48,6 +61,13 @@ _STAGE_LABELS: list[tuple[str, str]] = [
     ("evaluation", "Evaluation"),
 ]
 
+# Human labels for BusinessQuestionAnswer.status in the rendered markdown.
+_BQ_STATUS_LABELS: dict[str, str] = {
+    "answered": "Answered",
+    "partial": "Partially answered",
+    "unanswered": "Unanswered",
+}
+
 # Friendly labels for the catalog's internal source_type enum, shown in Data Sources.
 _SOURCE_TYPE_LABELS: dict[str, str] = {
     "schema": "Database",
@@ -113,16 +133,88 @@ def _collect_findings(records: list[AnalysisRecord]) -> list[ReportFinding]:
     return out
 
 
-def _collect_notes(records: list[AnalysisRecord], field: str) -> list[AttributedNote]:
-    # Caveats / open_questions are deduped by text; a merged note cites every
-    # record it came from (plural record_ids).
-    merged: dict[str, list[str]] = {}
+def _note_key(text: str) -> str:
+    # Dedupe key: collapse whitespace, drop trailing punctuation, casefold — so the
+    # Assembler's near-identical rephrasings ("Data is capped at 500 rows." vs
+    # "data is capped at 500 rows") merge into one note.
+    return " ".join(text.split()).rstrip(".!").casefold()
+
+
+def _collect_notes(records: list[AnalysisRecord], field: str, cap: int) -> list[AttributedNote]:
+    # Caveats / open_questions are deduped by normalized text; a merged note keeps
+    # the first phrasing seen and cites every record it came from (plural
+    # record_ids). Capped so a many-record analysis stays readable.
+    merged: dict[str, AttributedNote] = {}
     for rec in records:
         for text in getattr(rec, field):
-            ids = merged.setdefault(text, [])
-            if rec.record_id not in ids:
-                ids.append(rec.record_id)
-    return [AttributedNote(text=text, record_ids=ids) for text, ids in merged.items()]
+            key = _note_key(text)
+            if not key:
+                continue
+            note = merged.setdefault(key, AttributedNote(text=text))
+            if rec.record_id not in note.record_ids:
+                note.record_ids.append(rec.record_id)
+    return list(merged.values())[:cap]
+
+
+def _fmt_cell(value) -> str:
+    if value is None:
+        return "—"
+    if isinstance(value, float):
+        return f"{value:g}"  # 1234.5 not 1234.5000000001; no trailing zeros
+    return str(value)
+
+
+def _collect_evidence(records: list[AnalysisRecord]) -> dict[str, list[EvidenceTable]]:
+    """Copy small result tables out of each record's `results_snapshot` (INV-4).
+
+    Table-kind tool outputs only — the copy-paste-able slices (top-N rankings,
+    grouped aggregates, merges). `check_*` outputs are skipped (catalog metadata,
+    not evidence). Rows and tables-per-record are capped so a wide retrieval
+    can't balloon the report.
+    """
+    out: dict[str, list[EvidenceTable]] = {}
+    for rec in records:
+        tables: list[EvidenceTable] = []
+        for result in rec.results_snapshot.values():
+            for output in result.outputs:
+                if len(tables) >= _EVIDENCE_MAX_TABLES:
+                    break
+                if output.tool in ("check_data", "check_knowledge"):
+                    continue
+                if output.kind != "table" or not output.columns or not output.rows:
+                    continue
+                if len(output.columns) > _EVIDENCE_MAX_COLS:
+                    continue
+                tables.append(
+                    EvidenceTable(
+                        title=result.objective,
+                        columns=[str(c) for c in output.columns],
+                        rows=[
+                            [_fmt_cell(v) for v in row]
+                            for row in output.rows[:_EVIDENCE_MAX_ROWS]
+                        ],
+                        truncated=len(output.rows) > _EVIDENCE_MAX_ROWS,
+                    )
+                )
+        if tables:
+            out[rec.record_id] = tables
+    return out
+
+
+def _unresolved_note(rec: AnalysisRecord) -> AttributedNote:
+    # Goal + the record's own first caveat as the "why" — both Assembler-authored,
+    # nothing new is synthesized here.
+    text = rec.goal_restated or "Analysis run"
+    reason = next(iter(rec.caveats), None)
+    if reason:
+        text += f" — {reason}"
+    return AttributedNote(text=text, record_ids=[rec.record_id])
+
+
+def _excluded_note(rec: AnalysisRecord) -> AttributedNote:
+    return AttributedNote(
+        text=rec.goal_restated or rec.record_id, record_ids=[rec.record_id]
+    )
 
 
 def _collect_method_steps(records: list[AnalysisRecord]) -> list[TaskSummary]:
@@ -175,24 +267,72 @@ def _build_data_sources(
 
 
 def _build_human_content(
-    ps: ProblemStatement, findings: list[ReportFinding], caveats: list[AttributedNote]
+    ps: ProblemStatement,
+    records: list[AnalysisRecord],
+    caveats: list[AttributedNote],
+    reply_language: str,
 ) -> str:
+    # Questions and analyses are NUMBERED so the model can reference them by index
+    # in `bq_answers` (question_index / analysis_indexes) — it never reproduces ids.
     sections = []
     if ps.objective:
         sections.append("# Objective\n" + ps.objective)
     if ps.business_questions:
         sections.append(
-            "# Business questions\n" + "\n".join(f"- {q}" for q in ps.business_questions)
+            "# Business questions\n"
+            + "\n".join(f"{i}. {q}" for i, q in enumerate(ps.business_questions, 1))
         )
-    sections.append(
-        "# Findings (already finalized — synthesize, do not add numbers)\n"
-        + "\n".join(f"- {f.text}" for f in findings)
-    )
+    lines = ["# Analyses (findings already finalized — synthesize, do not add numbers)"]
+    for i, rec in enumerate(records, 1):
+        lines.append(f"Analysis {i}: {rec.goal_restated}")
+        seen: set[str] = set()
+        for text in rec.findings:
+            if text in seen:
+                continue
+            seen.add(text)
+            lines.append(f"- {text}")
+    sections.append("\n".join(lines))
     if caveats:
         sections.append("# Caveats\n" + "\n".join(f"- {c.text}" for c in caveats))
+    sections.append("# Reply language\n" + reply_language)
     return "\n\n".join(sections)
 
 
+def _resolve_bq_answers(
+    drafts: list[BQAnswerDraft],
+    questions: list[str],
+    records: list[AnalysisRecord],
+) -> list[BusinessQuestionAnswer]:
+    """Map the LLM's index-based drafts onto real question text and record ids.
+
+    Every question gets a row (unanswered when the model skipped it);
+    out-of-range indexes are silently dropped.
+    """
+    if not questions:
+        return []
+    by_index = {d.question_index: d for d in drafts}
+    out: list[BusinessQuestionAnswer] = []
+    for i, question in enumerate(questions, 1):
+        draft = by_index.get(i)
+        if draft is None:
+            out.append(BusinessQuestionAnswer(question=question))
+            continue
+        record_ids = [
+            records[j - 1].record_id
+            for j in draft.analysis_indexes
+            if 1 <= j <= len(records)
+        ]
+        out.append(
+            BusinessQuestionAnswer(
+                question=question,
+                answer=draft.answer,
+                status=draft.status,
+                record_ids=record_ids,
+            )
+        )
+    return out
+
+
 # Inline code spans (one or more backticks). Content inside is already literal in
 # Markdown/MDX, so escaping within them would only surface a visible backslash
 # (e.g. `product\_id`). We keep code spans verbatim and escape only around them.
@@ -254,6 +394,17 @@ def _render_markdown(report: AnalysisReport) -> str:
     if report.executive_summary:
         parts.append("## Executive Summary\n" + report.executive_summary)
 
+    if report.bq_answers:
+        lines = ["## Answers to Business Questions"]
+        for i, a in enumerate(report.bq_answers, 1):
+            label = _BQ_STATUS_LABELS.get(a.status, a.status)
+            entry = f"{i}. **{_mdx_escape(a.question)}** — *{label}*"
+            if a.answer:
+                # LLM prose (same authorship as the executive summary): not escaped.
+                entry += f"\n   {a.answer}"
+            lines.append(entry)
+        parts.append("\n".join(lines))
+
     if report.findings:
         # Group findings by their originating analysis (record) so results from
         # different questions read as separate analyses, not one flat, seemingly
@@ -275,6 +426,23 @@ def _render_markdown(report: AnalysisReport) -> str:
             if grouped:
                 block.append(f"### {_mdx_escape(report.record_goals.get(rid) or 'Analysis')}")
             block.extend(f"{i}. {_mdx_escape(f.text)}" for i, f in enumerate(group, 1))
+            # Evidence tables (copied result slices) under the findings they back,
+            # so the numbers are copy-paste-ready next to the claims.
+            for tbl in report.evidence_tables.get(rid, []):
+                if not tbl.columns:
+                    continue
+                block.append("")  # blank line: terminate the list before the table
+                if tbl.title:
+                    block.append(f"**{_mdx_escape(tbl.title)}**")
+                    block.append("")
+                block.append("| " + " | ".join(_mdx_escape(c) for c in tbl.columns) + " |")
+                block.append("|" + "---|" * len(tbl.columns))
+                block.extend(
+                    "| " + " | ".join(_mdx_escape(c) for c in row) + " |"
+                    for row in tbl.rows
+                )
+                if tbl.truncated:
+                    block.append(f"\n*(first {len(tbl.rows)} rows shown)*")
             blocks.append("\n".join(block))
         parts.append("\n\n".join(blocks))
 
@@ -296,26 +464,54 @@ def _render_markdown(report: AnalysisReport) -> str:
             )
         parts.append("\n".join(lines))
 
-    if report.caveats or report.open_questions:
-        lines = ["## Notes & Limitations"]
-        for n in report.caveats:
-            lines.append(f"- {_mdx_escape(n.text)}")
-        for n in report.open_questions:
-            lines.append(f"- Open: {_mdx_escape(n.text)}")
+    # ## Notes & Limitations — dropped from the rendered report 2026-07-09 (team
+    # decision: compact report). caveats/open_questions still populate the
+    # AnalysisReport JSON body; only the markdown section is gone.
+    # if report.caveats or report.open_questions:
+    #     lines = ["## Notes & Limitations"]
+    #     for n in report.caveats:
+    #         lines.append(f"- {_mdx_escape(n.text)}")
+    #     for n in report.open_questions:
+    #         lines.append(f"- Open: {_mdx_escape(n.text)}")
+    #     parts.append("\n".join(lines))
+
+    # ## Attempted, Unresolved — dropped from the rendered report 2026-07-09 (team
+    # decision: compact report). Failed runs still populate `report.unresolved`
+    # (JSON body) and the /records curation list; only the markdown section is gone.
+    # if report.unresolved:
+    #     lines = [
+    #         "## Attempted, Unresolved",
+    #         "*These analyses ran but produced no usable evidence;"
+    #         " they are not reflected in the findings above.*",
+    #         "",
+    #     ]
+    #     lines.extend(f"- {_mdx_escape(n.text)}" for n in report.unresolved)
+    #     parts.append("\n".join(lines))
+
+    if report.excluded:
+        lines = [
+            "## Excluded Analyses",
+            "*Excluded from this report at generation time.*",
+            "",
+        ]
+        lines.extend(f"- {_mdx_escape(n.text)}" for n in report.excluded)
         parts.append("\n".join(lines))
 
-    if report.method_steps:
-        lines = ["## How This Was Analyzed"]
-        for stage_key, label in _STAGE_LABELS:
-            steps = [s for s in report.method_steps if s.stage == stage_key]
-            if not steps:
-                continue
-            rendered = "; ".join(
-                f"{', '.join(_mdx_escape(t) for t in s.tools_used) or '—'} ({s.status})"
-                for s in steps
-            )
-            lines.append(f"**{label}** — {rendered}")
-        parts.append("\n".join(lines))
+    # ## How This Was Analyzed — dropped from the rendered report 2026-07-09 (team
+    # decision: compact report). method_steps (and _STAGE_LABELS above) stay for the
+    # AnalysisReport JSON body; only the markdown section is gone.
+    # if report.method_steps:
+    #     lines = ["## How This Was Analyzed"]
+    #     for stage_key, label in _STAGE_LABELS:
+    #         steps = [s for s in report.method_steps if s.stage == stage_key]
+    #         if not steps:
+    #             continue
+    #         rendered = "; ".join(
+    #             f"{', '.join(_mdx_escape(t) for t in s.tools_used) or '—'} ({s.status})"
+    #             for s in steps
+    #         )
+    #         lines.append(f"**{label}** — {rendered}")
+    #     parts.append("\n".join(lines))
 
     return "\n\n---\n\n".join(parts)
 
@@ -363,24 +559,36 @@ class ReportGenerator:
         user_id: str | None = None,
         problem_statement: ProblemStatement | None = None,
         user_name: str | None = None,
+        exclude_record_ids: list[str] | None = None,
     ) -> AnalysisReport:
-        records = await self._ensure_record_store().list_for_analysis(analysis_id)
-        # The report reflects only substantive runs — those with a successful
+        all_records = await self._ensure_record_store().list_for_analysis(analysis_id)
+        excluded_ids = set(exclude_record_ids or [])
+        excluded = [r for r in all_records if r.record_id in excluded_ids]
+        kept = [r for r in all_records if r.record_id not in excluded_ids]
+        # The report body reflects only substantive runs — those with a successful
         # analysis step (the same set the report floor validates). Fully-failed runs
-        # are dropped so their failure narration can't contradict the real findings.
-        records = [r for r in records if has_successful_analysis(r)]
+        # can't contradict the real findings, but they are not dropped silently
+        # either: they surface in the JSON `unresolved` list and the /records
+        # curation endpoint (the rendered markdown section was dropped 2026-07-09).
+        records = [r for r in kept if has_successful_analysis(r)]
+        unresolved_records = [r for r in kept if not has_successful_analysis(r)]
         if not records:
             raise ReportError(f"no analyses recorded for {analysis_id!r} yet")
 
         ps = problem_statement or ProblemStatement()
+        reply_language = detect_reply_language(
+            None, goal_texts=[ps.objective, *ps.business_questions]
+        )
         findings = _collect_findings(records)
-        caveats = _collect_notes(records, "caveats")
-        open_questions = _collect_notes(records, "open_questions")
+        caveats = _collect_notes(records, "caveats", _MAX_CAVEATS)
+        open_questions = _collect_notes(records, "open_questions", _MAX_OPEN_QUESTIONS)
         method_steps = _collect_method_steps(records)
         data_sources = _build_data_sources(
             records, await self._read_catalog(user_id, analysis_id)
         )
-        executive_summary = await self._summarize(ps, findings, caveats)
+        executive_summary, bq_answers = await self._summarize(
+            ps, records, caveats, reply_language
+        )
 
         report = AnalysisReport(
             analysis_id=analysis_id,
@@ -392,9 +600,13 @@ class ReportGenerator:
             record_ids=[r.record_id for r in records],
             record_goals={r.record_id: r.goal_restated for r in records},
             executive_summary=executive_summary,
+            bq_answers=bq_answers,
             findings=findings,
             caveats=caveats,
             open_questions=open_questions,
+            unresolved=[_unresolved_note(r) for r in unresolved_records],
+            excluded=[_excluded_note(r) for r in excluded],
+            evidence_tables=_collect_evidence(records),
             data_sources=data_sources,
             method_steps=method_steps,
         )
@@ -423,14 +635,20 @@ class ReportGenerator:
             return None
 
     async def _summarize(
-        self, ps: ProblemStatement, findings: list[ReportFinding], caveats: list[AttributedNote]
-    ) -> str:
-        human_content = _build_human_content(ps, findings, caveats)
+        self,
+        ps: ProblemStatement,
+        records: list[AnalysisRecord],
+        caveats: list[AttributedNote],
+        reply_language: str,
+    ) -> tuple[str, list[BusinessQuestionAnswer]]:
+        human_content = _build_human_content(ps, records, caveats, reply_language)
         try:
             narrative: ReportSummaryNarrative = await self._ensure_chain().ainvoke(
                 {"human_content": human_content}
             )
-            return narrative.executive_summary
         except Exception as exc:  # D1: degrade, don't fail the whole report
-            logger.warning("report summary LLM failed; using fallback", error=str(exc))
-            return _FALLBACK_SUMMARY
+            logger.warning("report summary LLM failed; using fallback", error=repr(exc))
+            return _FALLBACK_SUMMARY, []
+        return narrative.executive_summary, _resolve_bq_answers(
+            narrative.bq_answers, ps.business_questions, records
+        )

src/agents/report/schemas.py

@@ -14,12 +14,15 @@ See CHECKPOINT_PLAN_2026-06-17.md decision #8.
 from __future__ import annotations
 
 from datetime import datetime
+from typing import Literal
 from uuid import uuid4
 
 from pydantic import BaseModel, Field
 
 from ..slow_path.schemas import TaskSummary
 
+BQStatus = Literal["answered", "partial", "unanswered"]
+
 
 class ProblemStatement(BaseModel):
     """The analysis goal, frozen into each report at generation time.
@@ -66,10 +69,53 @@ class AttributedNote(BaseModel):
     record_ids: list[str] = Field(default_factory=list)
 
 
+class BusinessQuestionAnswer(BaseModel):
+    """A per-business-question answer, grounded only in the records' findings.
+
+    Drafted by the same single LLM call that authors the executive summary; the
+    question text and `record_ids` are resolved from the model's index-based
+    references by code, so the model never has to reproduce an id verbatim.
+    """
+
+    question: str
+    answer: str = ""
+    status: BQStatus = "unanswered"
+    record_ids: list[str] = Field(default_factory=list)  # records backing the answer
+
+
+class EvidenceTable(BaseModel):
+    """A small result table copied verbatim from a record's `results_snapshot`.
+
+    Deterministic pass-through (INV-4): code stringifies the cells and caps the
+    rows; nothing here is LLM-authored. `truncated` marks that rows were dropped
+    by the cap.
+    """
+
+    title: str = ""  # the producing task's objective
+    columns: list[str] = Field(default_factory=list)
+    rows: list[list[str]] = Field(default_factory=list)
+    truncated: bool = False
+
+
+class BQAnswerDraft(BaseModel):
+    """One business-question answer as the LLM emits it — index-based references.
+
+    `question_index` / `analysis_indexes` are 1-based positions into the numbered
+    lists shown in the human message; the generator maps them back to real
+    question text and record ids (out-of-range indexes are dropped).
+    """
+
+    question_index: int
+    answer: str = ""
+    status: BQStatus = "unanswered"
+    analysis_indexes: list[int] = Field(default_factory=list)
+
+
 class ReportSummaryNarrative(BaseModel):
     """The ONLY LLM-authored part of the report (with_structured_output target)."""
 
     executive_summary: str
+    bq_answers: list[BQAnswerDraft] = Field(default_factory=list)
 
 
 class AnalysisReport(BaseModel):
@@ -88,10 +134,18 @@ class AnalysisReport(BaseModel):
     record_goals: dict[str, str] = Field(default_factory=dict)
     # LLM-authored.
     executive_summary: str = ""
+    bq_answers: list[BusinessQuestionAnswer] = Field(default_factory=list)
     # Deterministic pass-through from records.
     findings: list[ReportFinding] = Field(default_factory=list)
     caveats: list[AttributedNote] = Field(default_factory=list)
     open_questions: list[AttributedNote] = Field(default_factory=list)
+    # Honesty sections: runs that produced no usable evidence (attempted but every
+    # analyze step failed) and records the report author excluded at generation.
+    # Both are outside `record_ids`/`record_goals` — they contribute no findings.
+    unresolved: list[AttributedNote] = Field(default_factory=list)
+    excluded: list[AttributedNote] = Field(default_factory=list)
+    # record_id -> small result tables copied from that record's results_snapshot.
+    evidence_tables: dict[str, list[EvidenceTable]] = Field(default_factory=dict)
     data_sources: list[DataSourceRef] = Field(default_factory=list)
     method_steps: list[TaskSummary] = Field(default_factory=list)  # carries `stage`
     rendered_markdown: str = ""

src/agents/slow_path/coordinator.py

@@ -11,13 +11,16 @@ See AGENT_ARCHITECTURE_CONTEXT_new.md §5.2 / §6.1.
 from __future__ import annotations
 
 from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
 
 from ...catalog.models import Catalog
 from ..planner.contracts import BusinessContext, ToolRegistry
 from ..planner.inputs import Constraints
+from ..planner.schemas import TaskList
 from ..planner.service import PlannerService
+from ..refusals import data_gap_message
 from .assembler import Assembler
-from .schemas import AssembledOutput
+from .schemas import AnalysisRecord, AssembledOutput
 from .task_runner import TaskRunner
 
 
@@ -54,6 +57,13 @@ class SlowPathCoordinator:
         task_list = await self._planner.plan(
             context, catalog, self._registry, query, constraints, **plan_kw
         )
+        if task_list.infeasible_reason and not task_list.tasks:
+            # Honest data-gap outcome (planner.md "When the catalog cannot
+            # answer"): nothing to execute, and the refusal is deliberately
+            # deterministic — not LLM-prosed. The record carries no tasks, so it
+            # is non-substantive: it can never satisfy the report floor or leak
+            # into a report.
+            return _infeasible_output(task_list, context, reply_language)
         if progress:
             await progress(f"Running {len(task_list.tasks)} analysis steps…")
         run_state = await self._task_runner.run(
@@ -65,3 +75,25 @@ class SlowPathCoordinator:
         return await self._assembler.assemble(
             run_state, context, question=query, reply_language=reply_language, **asm_kw
         )
+
+
+def _infeasible_output(
+    task_list: TaskList, context: BusinessContext, reply_language: str | None
+) -> AssembledOutput:
+    """Build the data-gap answer + a faithful (non-substantive) record."""
+    reason = task_list.infeasible_reason or ""
+    return AssembledOutput(
+        chat_answer=data_gap_message(reason, reply_language),
+        analysis_record=AnalysisRecord(
+            goal_restated=task_list.goal_restated,
+            findings=[],
+            caveats=[reason] if reason else [],
+            data_used=[],
+            open_questions=list(task_list.open_questions),
+            tasks_run=[],
+            results_snapshot={},
+            plan_id=task_list.plan_id,
+            business_context_id=context.project_id,
+            created_at=datetime.now(UTC),
+        ),
+    )

src/api/v1/help.py

@@ -51,8 +51,9 @@ async def help_stream(request: HelpRequest, db: AsyncSession = Depends(get_db)):
       2. chunk    — text fragments of the guidance
       3. done     — `{"message_id": "..."}` for the observability lookup
     """
-    # Server-authoritative turn id — never accepted from the caller (keys /observability).
-    message_id = f"msg_{uuid.uuid4().hex[:12]}"
+    # Server-authoritative turn id — never accepted from the caller (keys traceability).
+    # Canonical UUID string, matching Go's `analyses_messages.id` shape (mirrors v2 chat).
+    message_id = str(uuid.uuid4())
     try:
         history = await load_history(db, request.analysis_id, limit=10)
 

src/api/v1/report.py

@@ -2,9 +2,11 @@
 
 NOT a chat route. The frontend button calls these endpoints directly (pr/5: regrouped
 under /tools — Go owns the analysis lifecycle, Python only generates):
-  POST /api/v1/tools/report                       generate a new version for a session
-  GET  /api/v1/tools/report/{analysis_id}         list a session's report versions
-  GET  /api/v1/tools/report/{analysis_id}/{ver}   fetch one version
+  POST /api/v1/tools/report                          generate a new version for a session
+  GET  /api/v1/tools/report/{analysis_id}            list a session's report versions
+  GET  /api/v1/tools/report/{analysis_id}/records    list analysis records (curation)
+  GET  /api/v1/tools/report/{analysis_id}/readiness  readiness signal (FE delta guard)
+  GET  /api/v1/tools/report/{analysis_id}/{ver}      fetch one version
 
 Generation reads persisted AnalysisRecords + Problem Statement, makes one LLM call
 (the executive summary), and persists an immutable versioned artifact. The
@@ -24,7 +26,11 @@ from src.agents.report.generator import ReportGenerator
 from src.agents.report.schemas import AnalysisReport, ProblemStatement
 from src.agents.report.store import ReportStore
 from src.middlewares.logging import get_logger, log_execution
-from src.models.api.report import ReportVersionEntry
+from src.models.api.report import (
+    AnalysisRecordEntry,
+    ReportReadinessResponse,
+    ReportVersionEntry,
+)
 
 logger = get_logger("report_api")
 
@@ -114,6 +120,12 @@ async def _record_report_on_state(analysis_id: str, report_id: str) -> None:
 async def generate_report(
     analysis_id: str = Query(..., description="The analysis session to report on."),
     user_id: str = Query(..., description="Owner of the analysis session."),
+    exclude_record_ids: list[str] = Query(
+        default=[],
+        description="Record ids to leave out of this version (curation; repeat the "
+        "param per id). Excluded runs are listed in the report's Excluded Analyses "
+        "section. Get the ids from GET /tools/report/{analysis_id}/records.",
+    ),
 ):
     """Generate, persist, and return a new report version.
 
@@ -121,7 +133,8 @@ async def generate_report(
     Problem Statement it used. Server-side gate: the report **floor** — a validated
     goal + ≥1 substantive analysis — the same floor Help's readiness signal uses, so
     the button and Help can't disagree (T-D). The delta-since-report check is NOT
-    applied here: a new version is always allowed (decision 4A).
+    applied here: a new version is always allowed (decision 4A). Excluding every
+    substantive record 409s (nothing left to report).
     """
     from src.agents.gate import stub_analysis_state
     from src.agents.report.readiness import report_floor
@@ -142,7 +155,11 @@ async def generate_report(
         problem_statement = _problem_statement_from(state)
         user_name = await _resolve_user_name(user_id)
         report = await _generator.generate(
-            analysis_id, user_id, problem_statement=problem_statement, user_name=user_name
+            analysis_id,
+            user_id,
+            problem_statement=problem_statement,
+            user_name=user_name,
+            exclude_record_ids=exclude_record_ids,
         )
     except ReportError as e:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) from e
@@ -203,6 +220,72 @@ async def list_report_versions(analysis_id: str):
     ]
 
 
+# ⚠️ Route order: these two literal-suffix routes MUST stay registered BEFORE
+# `/report/{analysis_id}/{version}` — FastAPI matches in registration order, and the
+# `{version}` route would swallow `/records` / `/readiness` and 422 on int coercion
+# (no fall-through to a later route).
+
+
+@router.get(
+    "/report/{analysis_id}/records",
+    response_model=list[AnalysisRecordEntry],
+    summary="List a session's analysis records (for report curation)",
+    response_description="Persisted analysis runs, oldest-first. Empty if none yet.",
+)
+@log_execution(logger)
+async def list_analysis_records(analysis_id: str):
+    """Return the persisted analysis runs a report would be built from.
+
+    The FE shows this list before generating so the user can deselect runs; the
+    chosen ids go to POST /tools/report as `exclude_record_ids`.
+    """
+    from src.agents.report.readiness import has_successful_analysis
+    from src.agents.slow_path.store import PostgresReportInputStore
+
+    try:
+        records = await PostgresReportInputStore().list_for_analysis(analysis_id)
+    except Exception as e:
+        logger.error("record list failed", analysis_id=analysis_id, error=str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to list analysis records: {e}",
+        ) from e
+
+    return [
+        AnalysisRecordEntry(
+            record_id=r.record_id,
+            goal_restated=r.goal_restated,
+            created_at=r.created_at,
+            substantive=has_successful_analysis(r),
+            findings_count=len(r.findings),
+        )
+        for r in records
+    ]
+
+
+@router.get(
+    "/report/{analysis_id}/readiness",
+    response_model=ReportReadinessResponse,
+    summary="Report-readiness signal for an analysis session",
+    response_description="Whether a report can be generated now, with the gaps if not.",
+)
+@log_execution(logger)
+async def get_report_readiness(analysis_id: str):
+    """Deterministic readiness signal for the FE's Generate-Report button.
+
+    Same producer as Help's readiness signal (`is_report_ready`), including the
+    advisory delta-since-report check — so the button, Help, and this endpoint can
+    never disagree. POST itself only enforces the floor (a new version is always
+    allowed, decision 4A); `missing` here may name the delta gap as a soft warning.
+    """
+    from src.agents.gate import stub_analysis_state
+    from src.agents.report.readiness import is_report_ready
+
+    state = await _load_state(analysis_id)
+    readiness = await is_report_ready(analysis_id, state or stub_analysis_state())
+    return ReportReadinessResponse(ready=readiness.ready, missing=readiness.missing)
+
+
 @router.get(
     "/report/{analysis_id}/{version}",
     response_model=AnalysisReport,

src/api/v2/chat.py

@@ -70,8 +70,11 @@ async def _save_empty_chat_trace(analysis_id: str, user_id: str, message_id: str
 
 def _mint_message_id() -> str:
     """Mint the assistant turn id. Server-authoritative — never accepted from the caller
-    (it keys the future /observability lookup). Returned on `done`; open-Q #1 resolved."""
-    return f"msg_{uuid.uuid4().hex[:12]}"
+    (it keys the GET /api/v1/traceability lookup). Returned on `done`; open-Q #1 resolved.
+
+    A canonical UUID string, matching Go's `analyses_messages.id` shape, so the value stays
+    format-compatible if we later swap to the real message-row id (still Python-minted now)."""
+    return str(uuid.uuid4())
 
 
 class ChatRequest(BaseModel):

src/catalog/sample_decode.py

@@ -0,0 +1,119 @@
+"""Base64-sample-value decoding for catalogs affected by a dedorch bug.
+
+Go's introspection JSON-marshals numeric sample bytes as base64 (a `[]byte`
+serialization quirk), so every decimal/int-typed column's `sample_values`
+currently arrive as base64 strings (e.g. ``'OTUuMA=='`` for ``"95.0"``)
+instead of the plain numeric text the planner LLM expects. The planner then
+sees gibberish instead of value ranges for exactly the columns it filters and
+aggregates on. Until Go fixes the marshaling, decode these at catalog read
+time.
+
+Conservative by design (a wrong decode silently corrupts planner context):
+  - only numeric-typed columns are considered
+  - every non-null sample in the column must pass a strict base64 gate
+    (valid base64, decodes to printable ASCII, parses as a float) — a single
+    non-conforming entry leaves the WHOLE column untouched (mixed content is
+    suspicious, never guessed)
+  - columns with `sample_values is None` (e.g. PII-flagged columns, which
+    carry no samples by design) are skipped cleanly
+  - self-disabling: once Go ships real numeric samples (plain ``"95.0"`` or
+    actual numbers), the gate fails — plain digit strings are either not
+    base64-padded correctly or don't decode to printable numeric text — so
+    the pass becomes a no-op with no further changes needed here
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+
+from src.middlewares.logging import get_logger
+
+from .models import Catalog
+
+logger = get_logger("sample_decode")
+
+_NUMERIC_TYPES = {
+    "int",
+    "integer",
+    "bigint",
+    "decimal",
+    "numeric",
+    "float",
+    "double",
+    "number",
+}
+
+
+def _decode_one(value: str) -> str | None:
+    """Return the decoded numeric text for `value`, or None if it fails the gate."""
+    if len(value) < 2 or len(value) % 4 != 0:
+        return None
+    try:
+        decoded = base64.b64decode(value, validate=True)
+    except (binascii.Error, ValueError):
+        return None
+    try:
+        text = decoded.decode("ascii")
+    except UnicodeDecodeError:
+        return None
+    if not text.isprintable():
+        return None
+    try:
+        float(text)
+    except ValueError:
+        return None
+    return text
+
+
+def _decode_column_samples(samples: list) -> tuple[list, int] | None:
+    """Return (decoded list, count decoded) if every non-null entry passes the gate.
+
+    Returns None if any entry fails the gate (mixed content is left untouched).
+    """
+    decoded_values = []
+    count = 0
+    for entry in samples:
+        if entry is None:
+            decoded_values.append(None)
+            continue
+        if not isinstance(entry, str):
+            return None
+        decoded = _decode_one(entry)
+        if decoded is None:
+            return None
+        count += 1
+        decoded_values.append(decoded)
+    if not count:
+        return None
+    return decoded_values, count
+
+
+def decode_sample_values(catalog: Catalog) -> int:
+    """Decode base64-encoded numeric sample_values in place. Returns count decoded.
+
+    Never raises: any unexpected shape (wrong types, malformed entries) leaves
+    the offending column's values untouched.
+    """
+    total = 0
+    try:
+        for source in catalog.sources:
+            for table in source.tables:
+                for col in table.columns:
+                    if col.data_type.lower() not in _NUMERIC_TYPES:
+                        continue
+                    samples = col.sample_values
+                    if not samples:
+                        continue
+                    result = _decode_column_samples(samples)
+                    if result is None:
+                        continue
+                    decoded_values, count = result
+                    col.sample_values = decoded_values
+                    total += count
+    except Exception as e:
+        logger.error("sample decode failed", error=repr(e))
+        return total
+    if total:
+        logger.info("decoded base64 sample values", user_id=catalog.user_id, count=total)
+    return total

src/catalog/store.py

@@ -15,6 +15,7 @@ from src.middlewares.logging import get_logger
 
 from .fk_inference import infer_foreign_keys
 from .models import Catalog
+from .sample_decode import decode_sample_values
 
 logger = get_logger("catalog_store")
 
@@ -41,7 +42,12 @@ class CatalogStore:
         # dedorch catalogs ship no foreign_keys (Go introspection drops them),
         # but the IR validator only allows FK-backed joins. Infer the obvious
         # edges so the planner and validator agree. No-op once Go emits real FKs.
-        return infer_foreign_keys(Catalog.model_validate(row))
+        catalog = infer_foreign_keys(Catalog.model_validate(row))
+        # dedorch also JSON-marshals numeric sample bytes as base64 (Go bug) —
+        # decode them so the planner sees value ranges, not gibberish.
+        # No-op once Go emits plain numeric samples.
+        decode_sample_values(catalog)
+        return catalog
 
     async def get_by_analysis(self, analysis_id: str) -> Catalog | None:
         """Read the `scope_type='analysis'` catalog row for an analysis.
@@ -63,7 +69,9 @@ class CatalogStore:
             row = result.scalar_one_or_none()
         if row is None:
             return None
-        return infer_foreign_keys(Catalog.model_validate(row))
+        catalog = infer_foreign_keys(Catalog.model_validate(row))
+        decode_sample_values(catalog)
+        return catalog
 
     async def upsert(self, catalog: Catalog) -> None:
         # Legacy: Go's catalog.Service owns catalog writes now. Kept working (and

src/config/prompts/planner.md

@@ -21,6 +21,12 @@ only a `TaskList` object that conforms to the provided schema.
    id lookup, so a paraphrased name fails.
 5. **No modeling in v1.** There are no modeling tools. Do not emit `modeling`
    tasks. The product is descriptive/diagnostic only — no predictions, no charts.
+6. **Never re-purpose a column as a different business measure.** A column means
+   what the catalog says it means — do not alias one concept as another to force
+   an answer (e.g. selecting an availability percentage AS "revenue", or a
+   0-1 ratio AS a percentage metric). If no column plausibly holds the measure
+   or entity the question asks about, the plan is **infeasible** — see "When the
+   catalog cannot answer".
 
 # How to plan
 
@@ -61,6 +67,14 @@ only a `TaskList` object that conforms to the provided schema.
   `orders.total_amount`); if they genuinely aren't linked, say the data isn't
   connected rather than guessing. Prefer an existing measure column over
   recomputing. Joins are database-only — not available for tabular/file sources.
+- **Two measures per entity ("which X has both the worst A and the biggest B").**
+  Compute each measure in its OWN grouped `retrieve_data` task (one aggregate per
+  entity each), then align them with `analyze_merge`:
+  `{"data": "${tA}", "data_right": "${tB}", "on": ["<entity alias>"]}`.
+  The two retrievals MUST be separate tasks — a `"${t<id>}"` placeholder resolves
+  to a task's LAST output, so two retrievals inside one task lose the first
+  table. The merged table (one row per entity, both measures) answers the
+  question, or feeds a further `analyze_*` step.
 - **Mixing structured + unstructured.** If qualitative context helps, add a
   `retrieve_knowledge` task against an unstructured source listed in the catalog.
 - **CRISP-DM stages.** Tag each task with the stage it serves:
@@ -68,9 +82,29 @@ only a `TaskList` object that conforms to the provided schema.
 - **success_criteria is a reporting signal**, not a control trigger. State, in
   checkable terms (counts, rates, "produced", "above"/"below"), what a good
   result looks like. It never causes a retry.
-- **Surface uncertainty, don't guess.** If the question is ambiguous or the
-  catalog can't fully answer it, record it in `open_questions` and plan the best
-  defensible analysis anyway. Record interpretation choices in `assumptions`.
+- **Surface uncertainty, don't guess.** If the question is *ambiguous* — the
+  catalog can answer it but a term needs interpreting (which period, which
+  metric variant) — record the interpretation in `assumptions`, anything
+  unresolved in `open_questions`, and plan the best defensible analysis. This
+  never licenses re-purposing columns: when the requested measure itself is
+  absent from the catalog, the question is not ambiguous, it is **infeasible**
+  (next section).
+
+# When the catalog cannot answer
+
+Some questions ask for a measure or entity the connected sources simply do not
+hold (e.g. "sales revenue" against a maintenance database, "churn rate" with no
+subscription data). For those:
+
+- Return `tasks: []` and set **`infeasible_reason`**: one short paragraph naming
+  (a) what the question needs that no column provides, and (b) the nearest
+  analyses the catalog CAN support, so the user knows what to ask instead.
+- Do NOT emit a plan that maps the question onto semantically unrelated columns
+  just because their types fit — a confidently wrong number is worse than an
+  honest gap.
+- The test: could you point at a specific catalog column whose *meaning* (name,
+  sample values, table context) matches the requested measure? If not,
+  it is infeasible.
 
 # Writing a retrieve_data QueryIR
 
@@ -109,6 +143,14 @@ only a `TaskList` object that conforms to the provided schema.
   select the product column + `sum(revenue)` aliased `total_revenue`, with
   `group_by: ["<product_col_id>"]`,
   `order_by: [{"column_id": "total_revenue", "dir": "desc"}]`, `limit: 3`.
+  This applies to EVERY entity-ranking phrasing — "top/best/worst/highest/lowest
+  N <entities>", "<entities> with the best <measure> performance", "which
+  <entities> perform best" — the unit being ranked is the ENTITY, so the measure
+  MUST be aggregated per entity first (`group_by` the entity column). Ranking
+  raw rows can return the same entity twice, which is never a valid entity
+  ranking. Choose the aggregate by measure type: additive measures (revenue,
+  counts, backlog) → `sum`; ratio/percentage/rate metrics (availability,
+  utilization, scores) → `avg`. Record the choice in `assumptions`.
 
 # Output
 

src/config/prompts/report_summary.md

@@ -1,6 +1,12 @@
-You are a senior data analyst writing the **executive summary** of an analysis report.
+You are a senior data analyst writing the narrative parts of an analysis report.
 
-You are given the analysis Objective and its Business questions, plus a list of already-finalized findings and caveats drawn from completed analyses. Write a concise executive summary (3–5 sentences) that synthesizes those findings in relation to the objective and, where the findings allow, the business questions.
+You are given the analysis Objective, its numbered Business questions, and a numbered list of Analyses whose findings are already finalized, plus their caveats. You emit a structured object with two parts: `executive_summary` and `bq_answers`.
+
+Write ALL prose in the language named under "# Reply language".
+
+## executive_summary
+
+Write a concise executive summary (3–5 sentences) that synthesizes the findings in relation to the objective and, where the findings allow, the business questions.
 
 Rules:
 - Synthesize and prioritize — lead with the most decision-relevant finding.
@@ -9,3 +15,13 @@ Rules:
 - If the findings are thin or inconclusive, say so plainly rather than overstating.
 - Plain business language. Write **prose only — no headings, no bullet lists** (the report already supplies the section structure and a Key Findings list below this summary; do not duplicate them).
 - You MAY use light inline markdown for emphasis within the prose — `**bold**` for the most decision-relevant figure or term, `*italic*` sparingly. Keep it subtle; do not bold whole sentences.
+
+## bq_answers
+
+One entry per numbered business question. If no Business questions section is given, return an empty list.
+
+For each business question:
+- `question_index`: the question's number exactly as given.
+- `answer`: 1–3 sentences answering that question using ONLY the finalized findings — the same no-new-numbers rule as the summary. If nothing addresses it, one short sentence saying what the completed analyses do not cover.
+- `status`: `"answered"` when the findings fully answer the question, `"partial"` when they address only part of it (say which part in the answer), `"unanswered"` when no finding addresses it.
+- `analysis_indexes`: the numbers of the Analyses whose findings support the answer (empty when unanswered). Use only numbers that appear in the Analyses list.

src/db/postgres/models.py

@@ -265,7 +265,7 @@ class MessageTraceabilityRow(Base):
     (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
+    (the Python-minted turn id, a UUID string), 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
@@ -278,7 +278,7 @@ class MessageTraceabilityRow(Base):
     """
     __tablename__ = "message_traceability"
 
-    message_id = Column(String, primary_key=True)  # Python-minted turn id ("msg_<12hex>")
+    message_id = Column(String, primary_key=True)  # Python-minted turn id (UUID string)
     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)

src/models/api/report.py

@@ -17,3 +17,32 @@ class ReportVersionEntry(BaseModel):
     version: int = Field(..., description="Monotonic version (V1, V2, …).")
     generated_at: datetime = Field(..., description="When this version was generated.")
     record_count: int = Field(..., description="Number of AnalysisRecords it was built from.")
+
+
+class AnalysisRecordEntry(BaseModel):
+    """One persisted analysis run, listed for report curation (KM-644 report v2).
+
+    The FE shows these before generating so the user can exclude runs
+    (`exclude_record_ids` on POST /tools/report). `substantive` mirrors the
+    report's own inclusion rule: non-substantive runs land in the report's
+    `unresolved` list rather than the findings body (the "Attempted, Unresolved"
+    markdown section was dropped 2026-07-09; the JSON field remains).
+    """
+
+    record_id: str = Field(..., description="Id to pass in exclude_record_ids.")
+    goal_restated: str = Field("", description="The run's question, as the agent restated it.")
+    created_at: datetime = Field(..., description="When the run was recorded.")
+    substantive: bool = Field(
+        ..., description="True if an analyze_* step succeeded (counts toward the report floor)."
+    )
+    findings_count: int = Field(0, description="Number of findings the run recorded.")
+
+
+class ReportReadinessResponse(BaseModel):
+    """Deterministic report-readiness signal (same producer as Help's signal)."""
+
+    ready: bool = Field(..., description="Whether generating a report now makes sense.")
+    missing: list[str] = Field(
+        default_factory=list,
+        description="Human-readable gaps when not ready (e.g. the delta-since-report check).",
+    )

src/query/ir/validator.py

@@ -87,6 +87,22 @@ class IRValidator:
         for i, col_id in enumerate(ir.group_by):
             self._require_column(columns_by_id, col_id, f"group_by[{i}]")
 
+        # A grouped query must not select bare columns that aren't in group_by —
+        # the database rejects it only at execution ("must appear in the GROUP BY
+        # clause"), which is past the planner's corrective-retry window. Catching
+        # it here turns a failed turn into a self-correcting re-prompt.
+        if ir.group_by:
+            grouped = set(ir.group_by)
+            for i, item in enumerate(ir.select):
+                if item.kind == "column" and item.column_id not in grouped:
+                    raise IRValidationError(
+                        f"select[{i}].column_id {item.column_id!r} is selected bare "
+                        "while group_by is present — every selected column must "
+                        "either appear in group_by or be wrapped in an aggregate "
+                        f'(e.g. {{"kind": "agg", "fn": "sum", '
+                        f'"column_id": {item.column_id!r}}})'
+                    )
+
         for i, ob in enumerate(ir.order_by):
             if ob.column_id not in columns_by_id and ob.column_id not in select_aliases:
                 raise IRValidationError(

src/tools/analytics/merge.py

@@ -0,0 +1,136 @@
+"""analyze_merge — combine TWO upstream tables on shared keys (KM-608).
+
+The only analytics "family" tool with a SECOND data input. In ONE call it joins
+two already-materialized tables (`data` = LEFT, `data_right` = RIGHT) on one or
+more shared key columns and returns the combined rows. This is what unlocks the
+"which X has BOTH the worst A and the biggest B" question shape: A and B come
+from two separate `retrieve_data` pulls (e.g. PA-by-section and backlog-by-section)
+that must be aligned per X before either can be judged against the other. Without
+a two-input combine the run dies with ColumnNotFoundError because no single tool
+ever sees both metrics.
+
+Pattern A, extended: it takes TWO `"${t<id>}"` placeholders. The invoker
+materializes BOTH into DataFrames before calling this function (no self-fetch);
+the `on` key(s) reference the column aliases the upstream queries produced.
+
+STATUS: compute layer only — takes two already-materialized DataFrames. The
+wrapper layer (the ToolOutput envelope, dual-arg materialization, ToolSpec
+registration) lives in src/tools/invoker.py + registry.py. Keeping compute
+separate from data-fetching keeps this easy to unit-test and stable when wrapped.
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+
+from src.tools.analytics.descriptive import ColumnNotFoundError
+
+# Join types the tool understands. Whitelisted so an unknown `how` fails loudly
+# instead of silently doing the wrong thing. "cross" is deliberately excluded —
+# it ignores `on` and is never the right tool for the "align two metrics" shape.
+SUPPORTED_HOWS = ("inner", "left", "right", "outer")
+
+
+class UnsupportedJoinError(ValueError):
+    """Requested join type is not in SUPPORTED_HOWS (maps to error_code UNSUPPORTED_JOIN)."""
+
+
+def _clean(value: object) -> object:
+    """Coerce a scalar to a JSON-clean Python value.
+
+    An outer/left/right join introduces `NaN` for non-matching rows, and numpy /
+    pandas scalars (numpy.int64, pandas.Timestamp) are not JSON-serializable —
+    normalise all three so the returned rows are clean.
+    """
+    if isinstance(value, pd.Timestamp):
+        return value.isoformat()
+    if value is None:
+        return None
+    try:
+        if pd.isna(value):
+            return None
+    except (TypeError, ValueError):
+        pass  # non-scalar / unhashable — leave as-is
+    if hasattr(value, "item"):
+        return value.item()
+    return value
+
+
+# Prompt-style description read by the Planner to decide WHEN to pick this tool.
+DESCRIPTION = """\
+Summary: Combine TWO upstream tables into one by joining on shared key column(s) \
+(a pandas merge). `data` is the LEFT table, `data_right` is the RIGHT table; `on` \
+is the shared column alias(es) present in BOTH. Returns the combined rows, one \
+per matched key (join type controlled by `how`, default inner).
+
+USE WHEN a question needs TWO different metrics per the SAME entity and those \
+metrics come from two separate pulls — the tell-tale shape is "which X has BOTH \
+A and B" (e.g. "which section has the worst PA AND the biggest backlog", "top \
+customers by revenue that also have the most complaints"). Plan it as two \
+retrieve_data tasks (one per metric, each keyed by X), then analyze_merge on X.
+
+SETTING KEYS: `on` must be column alias(es) that exist in BOTH tables (the entity \
+you align on, e.g. section_id). Use `suffixes` (default ["_left","_right"]) to \
+disambiguate non-key columns that share a name across the two tables. `how`: \
+inner (only matched keys), left/right (keep one side), outer (keep all).
+
+DON'T USE WHEN:
+  - both metrics can be pulled in ONE retrieve_data query -> just retrieve_data
+  - it groups/aggregates a single table -> analyze_aggregate
+
+Example questions:
+  - "which section has the worst PA and the biggest maintenance backlog"
+  - "regions in the top 10 for sales that are also bottom 10 for margin"
+  - "products low on stock that also have high demand"
+"""
+
+
+def analyze_merge(
+    df: pd.DataFrame,
+    data_right: pd.DataFrame,
+    on: list[str] | str,
+    how: str = "inner",
+    suffixes: tuple[str, str] | list[str] = ("_left", "_right"),
+) -> list[dict[str, object]]:
+    """Join two already-materialized tables on shared key column(s).
+
+    Args:
+        df: LEFT table (in the real system the invoker materializes this from the
+            `data` placeholder).
+        data_right: RIGHT table (materialized from the `data_right` placeholder).
+        on: shared key column alias(es) present in BOTH tables. A bare string is
+            treated as a single key.
+        how: join type — one of SUPPORTED_HOWS (default "inner").
+        suffixes: 2-element (left, right) suffixes applied to non-key columns that
+            collide by name across the two tables.
+
+    Returns:
+        list[dict]: one row per merged record, values JSON-clean (NaN -> None).
+
+    Raises:
+        ColumnNotFoundError: if `on` is empty or a key is absent from either side.
+        UnsupportedJoinError: if `how` is not supported.
+        ValueError: if `suffixes` is not a 2-element sequence.
+    """
+    keys = [on] if isinstance(on, str) else list(on)
+    if not keys:
+        raise ColumnNotFoundError("merge 'on' must name at least one shared key column")
+    if how not in SUPPORTED_HOWS:
+        raise UnsupportedJoinError(
+            f"unsupported join '{how}'; supported: {list(SUPPORTED_HOWS)}"
+        )
+
+    missing_left = [c for c in keys if c not in df.columns]
+    missing_right = [c for c in keys if c not in data_right.columns]
+    if missing_left or missing_right:
+        raise ColumnNotFoundError(
+            f"join key(s) not found — left missing {missing_left}, "
+            f"right missing {missing_right}"
+        )
+
+    suf = tuple(suffixes)
+    if len(suf) != 2:
+        raise ValueError(f"suffixes must be a 2-element (left, right) sequence, got {suffixes!r}")
+
+    merged = df.merge(data_right, on=keys, how=how, suffixes=suf)
+    return [{k: _clean(v) for k, v in rec.items()} for rec in merged.to_dict("records")]

src/tools/analytics/temporal.py

@@ -41,6 +41,11 @@ class UnsupportedAggregationError(ValueError):
     """The requested aggregation is not supported (maps to error_code UNSUPPORTED_AGG)."""
 
 
+class InvalidDateColumnError(ValueError):
+    """date_column holds numeric values that aren't a recognizable date/year/month
+    (maps to error_code INVALID_DATE_COLUMN)."""
+
+
 def _clean(value: object) -> object:
     """Convert numpy scalars to plain Python; NaN -> None for JSON-clean output."""
     if value is None:
@@ -53,6 +58,63 @@ def _clean(value: object) -> object:
     return value
 
 
+def _parse_date_column(df: pd.DataFrame, date_column: str) -> pd.Series:
+    """Parse date_column into datetimes, guarding against numeric epoch misparsing.
+
+    pd.to_datetime() treats bare numeric input as epoch-nanoseconds, so bare
+    month numbers (1-12) or calendar years (e.g. 2025) silently collapse to a
+    single 1970 timestamp instead of raising. Numeric columns are resolved
+    explicitly here rather than falling through to pd.to_datetime().
+    """
+    col = df[date_column]
+    if not pd.api.types.is_numeric_dtype(col):
+        return pd.to_datetime(col)
+
+    non_null = col.dropna()
+    is_whole = non_null.empty or (non_null == non_null.astype(int)).all()
+
+    if is_whole and non_null.between(1, 12).all():
+        year_col = next((c for c in df.columns if c.lower() == "year"), None)
+        year_series = df[year_col] if year_col is not None else None
+        year_non_null = year_series.dropna() if year_series is not None else pd.Series(dtype=float)
+        year_ok = (
+            year_series is not None
+            and pd.api.types.is_numeric_dtype(year_series)
+            and not year_non_null.empty
+            and (year_non_null == year_non_null.astype(int)).all()
+            and year_non_null.between(1900, 2100).all()
+        )
+        if not year_ok:
+            raise InvalidDateColumnError(
+                f"date_column '{date_column}' holds bare month numbers (1-12) and no "
+                "'year' column is present in the data — retrieve a year column "
+                "alongside month, or use a real date column."
+            )
+        valid = col.notna() & year_series.notna()
+        result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
+        result.loc[valid] = pd.to_datetime(
+            {
+                "year": year_series.loc[valid].astype(int),
+                "month": col.loc[valid].astype(int),
+                "day": 1,
+            }
+        )
+        return result
+
+    if is_whole and non_null.between(1900, 2100).all():
+        result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
+        valid = col.notna()
+        result.loc[valid] = pd.to_datetime(
+            col.loc[valid].astype(int).astype(str), format="%Y"
+        )
+        return result
+
+    raise InvalidDateColumnError(
+        f"date_column '{date_column}' is numeric but is not a recognizable date, "
+        "year, or month column."
+    )
+
+
 def _period_label(ts: pd.Timestamp, freq: str) -> str:
     """Human-readable period label keyed off the friendly frequency name."""
     if freq == "month":
@@ -119,6 +181,8 @@ def analyze_trend(
         ColumnNotFoundError: if date_column or value_column is absent.
         InvalidFrequencyError: if freq is not a known period.
         UnsupportedAggregationError: if agg is not supported.
+        InvalidDateColumnError: if date_column is numeric but not a recognizable
+            date, year, or bare month number (needing a companion 'year' column).
     """
     missing = [c for c in (date_column, value_column) if c not in df.columns]
     if missing:
@@ -134,7 +198,7 @@ def analyze_trend(
 
     # Build a clean datetime-indexed series, then resample into periods.
     s = df[[date_column, value_column]].copy()
-    s[date_column] = pd.to_datetime(s[date_column])
+    s[date_column] = _parse_date_column(df, date_column)
     s = s.dropna(subset=[date_column]).set_index(date_column).sort_index()
     resampled = s[value_column].resample(FREQ_MAP[freq]).agg(agg)
 

src/tools/data_access.py

@@ -154,7 +154,9 @@ class DataAccessToolInvoker:
             [
                 t.table_id,
                 t.name,
-                t.row_count,
+                # dedorch catalogs mark an uncounted table as -1; surface None so
+                # the planner prompt never sees a nonsensical "-1 rows".
+                t.row_count if (t.row_count or 0) >= 0 else None,
                 c.column_id,
                 c.name,
                 c.data_type,

src/tools/invoker.py

@@ -31,6 +31,7 @@ from src.tools.analytics import (
     comparison,
     decomposition,
     descriptive,
+    merge,
     quality,
     relationship,
     segmentation,
@@ -52,6 +53,7 @@ _DISPATCH: dict[str, tuple[Callable[..., Any], str]] = {
     "analyze_correlation": (relationship.analyze_correlation, "stats"),
     "analyze_segment": (segmentation.analyze_segment, "table"),
     "analyze_trend": (temporal.analyze_trend, "series"),
+    "analyze_merge": (merge.analyze_merge, "table"),
 }
 
 
@@ -73,6 +75,19 @@ class AnalyticsToolInvoker:
             return ToolOutput(tool=tool_name, kind="error", error=err)
 
         kwargs = {k: v for k, v in args.items() if k != "data"}
+
+        # Second data input (Pattern A, extended): analyze_merge takes a
+        # `data_right` placeholder the TaskRunner has resolved to another upstream
+        # ToolOutput. Materialize it the same way as `data` and hand the compute fn
+        # a DataFrame, not the raw envelope.
+        if "data_right" in kwargs:
+            df_right, err = _materialize(kwargs["data_right"])
+            if err is not None:
+                err = f"data_right: {err}"
+                logger.warning("tool returned error", tool=tool_name, error=err)
+                return ToolOutput(tool=tool_name, kind="error", error=err)
+            kwargs["data_right"] = df_right
+
         try:
             result = fn(df, **kwargs)
         except Exception as exc:  # noqa: BLE001 — never-throw seam (§8.4)

src/tools/registry.py

@@ -29,6 +29,7 @@ from src.tools.analytics import (
     comparison,
     decomposition,
     descriptive,
+    merge,
     quality,
     relationship,
     segmentation,
@@ -96,6 +97,22 @@ ACTIVE_ANALYTICS_TOOLS: list[ToolSpec] = [
         output_kind="series",
         description=temporal.DESCRIPTION,
     ),
+    ToolSpec(
+        name="analyze_merge",
+        category="analytics.combine",
+        input_schema={
+            "required": ["data", "data_right", "on"],
+            "properties": {
+                "data": {"type": "string"},
+                "data_right": {"type": "string"},
+                "on": {"type": "array"},
+                "how": {"type": "string"},
+                "suffixes": {"type": "array"},
+            },
+        },
+        output_kind="table",
+        description=merge.DESCRIPTION,
+    ),
 ]
 
 # Deferred this round — specs kept intact for easy re-activation, NOT exposed to

src/traceability/scratchpad.py

@@ -124,13 +124,17 @@ class TraceabilityScratchpad:
                 error=out_dict.get("error"),
             )
         )
-        if name == "retrieve_data":
+        if name == "retrieve_data" and status == "success":
             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)
+        if not meta.get("source_id"):
+            # A failed/aborted retrieval carries no provenance meta — emitting it
+            # anyway produced all-null source rows in the payload.
+            return
         query = meta.get("query")
         table = meta.get("table_name") or meta.get("table_id")
         self._db_sources.append({