[NOTICKET] fix(planner): reject analyze_* fed from metadata/documents tools

The planner sometimes wired an analyze_* tool's 'data' to check_data (catalog metadata) instead of retrieve_data (rows); the plan passed structural validation but analyze_* then failed to find its columns at runtime (silent ColumnNotFoundError). Add PlannerValidator Check 9: an analyze_* 'data' placeholder must not reference a catalog.introspection (check_data/check_knowledge) or retrieval.documents (retrieve_knowledge) task — a category denylist, so retrieve_data and table-producing analyze_* chains stay valid. Reinforce with prompt guidance (planner.md: never feed analyze_* from check_data) + a new analyze_descriptive few-shot (Example F). Also log a compact plan dump on 'analysis planned' for diagnosability.

src/agents/planner/examples.py

@@ -448,12 +448,89 @@ _EXAMPLE_E = TaskList(
 )
 
 
+# --------------------------------------------------------------------------- #
+# Example F — descriptive statistics (analyze_descriptive).
+# "Give me the summary statistics for order revenue and quantity."
+# Shows: retrieve_data pulls the numeric columns -> analyze_descriptive summarizes
+# them. The `data` comes from the retrieve_data task (t2), NEVER from the check_data
+# inspection step (t1) — check_data returns column metadata, not data rows.
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_F = TaskList(
+    plan_id="example_f",
+    goal_restated="Summarize the distribution of order revenue and quantity.",
+    assumptions=[],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes order 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; the needed columns are present.",
+            depends_on=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Pull the order revenue and quantity rows.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_data",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_revenue", "alias": "revenue"},
+                                {"kind": "column", "column_id": "c_quantity", "alias": "quantity"},
+                            ],
+                            "limit": 10000,
+                        }
+                    },
+                )
+            ],
+            expected_output="order_rows",
+            success_criteria="Produced order-level rows with revenue and quantity.",
+            depends_on=["t1"],
+            estimated_cost="medium",
+        ),
+        Task(
+            id="t3",
+            stage="evaluation",
+            objective="Summarize the distribution of revenue and quantity.",
+            tool_calls=[
+                ToolCall(
+                    tool="analyze_descriptive",
+                    args={
+                        # `data` references t2 (retrieve_data rows), NOT t1 (check_data).
+                        "data": "${t2}",
+                        # column refs are the retrieve_data output aliases.
+                        "column_ids": ["revenue", "quantity"],
+                    },
+                )
+            ],
+            expected_output="summary_stats",
+            success_criteria=(
+                "Produced mean/median/std/quartiles for revenue and quantity, above/below "
+                "the typical range."
+            ),
+            depends_on=["t2"],
+            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),
     ("Revenue dipped in Q1 — what happened?", _EXAMPLE_C),
     ("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),
 ]
 
 

src/agents/planner/service.py

@@ -142,6 +142,25 @@ class PlannerService:
                 plan_id=task_list.plan_id,
                 n_tasks=len(task_list.tasks),
                 retry=attempt > 1,
+                # Compact plan dump: task id, its tools, deps, and each tool's arg keys
+                # (+ any analyze_* column refs) — enough to see how retrieve_data feeds
+                # the analyze_* tools without dumping full inline IRs.
+                plan=[
+                    {
+                        "id": t.id,
+                        "depends_on": t.depends_on,
+                        "tools": [
+                            {
+                                "tool": c.tool,
+                                "args": sorted(c.args.keys()),
+                                "cols": c.args.get("column_ids") or c.args.get("column"),
+                                "data": c.args.get("data"),
+                            }
+                            for c in t.tool_calls
+                        ],
+                    }
+                    for t in task_list.tasks
+                ],
             )
             return task_list
 

src/agents/planner/validator.py

@@ -29,6 +29,15 @@ logger = get_logger("ir_repair")
 # Heuristic: a checkable success_criteria mentions a measurable signal.
 _CHECKABLE_TOKENS = ("rate", "count", "match", "produced", "above", "below", "equal")
 
+# Tool categories whose output is NOT analyzable data rows: `catalog.introspection`
+# (check_data/check_knowledge → catalog metadata) and `retrieval.documents`
+# (retrieve_knowledge → prose chunks). An analyze_* `data` handoff must not come from
+# one of these — their output is kind="table"/documents so the structural checks pass,
+# but feeding metadata rows to analyze_* makes it fail to find the requested columns.
+# A denylist (not an allowlist) so any data/table-producing tool — retrieve_data AND a
+# table-producing analyze_* chained into another analyze_* — stays valid.
+_NON_DATA_SOURCE_CATEGORIES = frozenset({"catalog.introspection", "retrieval.documents"})
+
 # DFS colors for cycle detection.
 _WHITE, _GREY, _BLACK = 0, 1, 2
 
@@ -109,6 +118,11 @@ class PlannerValidator:
                 if call.tool == "retrieve_data":
                     self._validate_inline_ir(task.id, call.args, catalog)
 
+                # Check 9 — a `data` handoff (Pattern A) must reference a task that
+                # produces analyzable data rows, not one producing catalog metadata
+                # (check_data/check_knowledge) or documents (retrieve_knowledge).
+                self._validate_data_source(task.id, call, tasks_by_id, registry)
+
             # Check 7 — success_criteria is checkable.
             if not _is_checkable(task.success_criteria):
                 raise PlannerValidationError(
@@ -157,6 +171,37 @@ class PlannerValidator:
                 f"task {task_id}: retrieve_data IR failed catalog validation: {e}"
             ) from e
 
+    @staticmethod
+    def _validate_data_source(
+        task_id: str, call, tasks_by_id: dict, registry: ToolRegistry
+    ) -> None:
+        """A `data` placeholder must reference a data-producing task, not a
+        metadata (check_data/check_knowledge) or documents (retrieve_knowledge) one.
+
+        Those pass the structural checks (check_* also returns kind="table"), but
+        their rows are catalog schema, so a downstream analyze_* fails to find the
+        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)."
+            )
+
     @staticmethod
     def _validate_dag(tasks_by_id: dict, id_set: set[str]) -> None:
         for task in tasks_by_id.values():

src/config/prompts/planner.md

@@ -39,6 +39,12 @@ only a `TaskList` object that conforms to the provided schema.
   profiling — run `retrieve_data` to fetch the rows, then pass its output to
   the matching composite `analyze_*` tool via a `"${t<id>}"` `data` argument
   (referencing the upstream result's column aliases).
+  An `analyze_*` tool's `data` MUST reference a `retrieve_data` task (or a
+  table-producing `analyze_*`) — **NEVER `check_data` or `check_knowledge`**.
+  Those return catalog **metadata** (a listing of columns/types), not data rows,
+  so an `analyze_*` fed from them finds no columns to analyze and fails.
+  `check_data` is only for inspecting *what exists*; always `retrieve_data` to
+  pull the rows before analyzing them.
 - **Measure by a dimension in another table (joins).** When the number you are
   aggregating and the grouping dimension live in DIFFERENT tables of the same
   database source, add a `joins` entry to the `retrieve_data` IR. **Join ONLY on a