[KM-644] report v2: BQ answers, unresolved/excluded sections, evidence tables + curation/readiness endpoints

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. |
 
@@ -218,11 +220,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 +255,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 +293,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 +350,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 +387,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 would land under "Attempted, Unresolved" rather than the findings body. 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

@@ -62,9 +62,9 @@ base64-mangled from Go. Fix tasks (same status legend as §0):
 | 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 | ⬜ | flagged out of this sprint; request brief sent by Rifqi |
-| Q8 | Report v2: business-question answer section, unresolved/excluded sections, evidence tables from `results_snapshot`, caveat dedupe, single language | Rifqi/Sofhia | ⬜ | next up; adds one LLM call (button-triggered); new prompt → eval per §7B |
-| Q9 | Record-curation endpoint (`GET …/records` + `exclude_record_ids`) + readiness GET for the FE delta guard | Rifqi ↔ FE | ⬜ | contract addition → API_CONTRACT_BE_PYTHON.md |
+| 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)
 

REPO_STATUS.md

@@ -2,7 +2,7 @@
 
 **Audience:** teammates onboarding onto the Python repo (`Agentic-Service-Data-Eyond-Catalog`).
 **Scope:** what the code does **right now** (branch `pr/4`, ticket KM-652). Describes current state only — no roadmap or to-dos.
-**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Agent-quality fixes 2026-07-08 (pr/13):** from the scoped live-test review — the planner gains an explicit **infeasible** outcome (`TaskList.infeasible_reason` → deterministic EN/ID data-gap reply via `refusals.data_gap_message`; no more force-mapping absent measures like `pa` AS "revenue"), the IR validator rejects bare selects under `group_by` (self-corrects via the planner retry), `analyze_trend` handles integer year/month columns (was collapsing every row into one 1970-01 bucket), planner few-shots add top-N (Example G) + infeasible (Example H), numeric catalog `sample_values` are base64-decoded at read (`catalog/sample_decode.py` — stopgap for Go's byte-marshaling; primary fix is Go-side), traceability no longer emits null source rows for failed retrievals, and `check_data` hides `-1` row counts. **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. **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/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))
 
@@ -304,6 +472,25 @@ def _render_markdown(report: AnalysisReport) -> str:
             lines.append(f"- Open: {_mdx_escape(n.text)}")
         parts.append("\n".join(lines))
 
+    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:
@@ -363,24 +550,35 @@ 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 no longer dropped
+        # silently: they surface under "Attempted, Unresolved".
+        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 +590,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 +625,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/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/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/models/api/report.py

@@ -17,3 +17,31 @@ 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 would land under
+    "Attempted, Unresolved" rather than the findings body.
+    """
+
+    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).",
+    )