[KM-644] planner: infeasible plan outcome + entity-ranking few-shots

TaskList.infeasible_reason lets the planner decline questions no catalog
column can answer (was force-mapping, e.g. pa AS revenue); coordinator
short-circuits to a deterministic EN/ID data-gap reply and a
non-substantive record. Few-shots add top-N entity ranking (G) and an
infeasible example (H); planner.md gains hard rule 6 + 'When the catalog
cannot answer'. Live-verified 2026-07-08 on the PA analysis.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/agents/planner/examples.py

@@ -524,6 +524,94 @@ _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."
+    ),
+)
+
+
 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 +619,8 @@ 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),
 ]
 
 

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")

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/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/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
 
@@ -68,9 +74,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 +135,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