[KM-624][AI] Planner: realign stub registry + examples to composite analyze_* tools

Team decision: v1 uses composite "family" tools (analyze_*), not the atomic
compute_* set. Realign the planner-facing stub to the real KM-624 inventory so
the Planner plans against tools that exist.

- registry.py: replace the 9 atomic entries (compute_median/stddev/percentile/
mode, date_trunc, ...) with 12 composite entries -- 4 data-access
(query_structured, retrieve_documents, list_sources, describe_source) + 8
analyze_* (descriptive, aggregate, comparison, contribution, profile,
correlation, segment, trend). Each analyze_* takes a `data` "${t<id>}"
placeholder (Pattern A, assumed pending the tool team).
- examples.py: Example A -> analyze_contribution; Example B -> analyze_trend
(drops the removed date_trunc/compute_stddev chain).
- planner.md: rewrite the "compute_* tools" bullet as data-access vs analytics.

Validator/prompt/service unchanged (generic over the registry). Planner tests
updated locally (tests/ is gitignored): 32 passing + 1 gated, ruff clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/planner/examples.py

@@ -2,7 +2,11 @@
 
 Two illustrative (question -> TaskList) pairs that teach the OUTPUT SHAPE:
 stages, dependency edges, parallelism, ordered tool-call chains, inline QueryIR,
-and "${t<id>}" placeholders. They reference a hypothetical sales catalog
+"${t<id>}" placeholders, and the assumed data-flow convention — `query_structured`
+pulls rows, then a composite `analyze_*` tool consumes them via a `data` placeholder
+referencing the upstream result's column aliases (Pattern A; the tool team may
+instead pick self-fetch by `source_id`, in which case these examples are reshaped
+to match — see registry.py). They reference a hypothetical sales catalog
 (`src_sales` / `t_orders`); these ids are part of the illustration and are not
 validated against the user's real catalog. v1 is descriptive/diagnostic — no
 modeling tasks.
@@ -17,6 +21,9 @@ from .schemas import Task, TaskList, ToolCall
 # --------------------------------------------------------------------------- #
 # Example A — exploratory, no modeling.
 # "Which product categories drove last quarter's revenue?"
+# Shows: query_structured pulls rows -> analyze_contribution computes each
+# category's share of the total in one call (no manual per-category + total
+# queries).
 # --------------------------------------------------------------------------- #
 
 _EXAMPLE_A = TaskList(
@@ -38,8 +45,8 @@ _EXAMPLE_A = TaskList(
         ),
         Task(
             id="t2",
-            stage="evaluation",
-            objective="Sum last quarter's revenue per category, ranked high to low.",
+            stage="data_preparation",
+            objective="Pull last quarter's order-level category and revenue rows.",
             tool_calls=[
                 ToolCall(
                     tool="query_structured",
@@ -49,12 +56,7 @@ _EXAMPLE_A = TaskList(
                             "table_id": "t_orders",
                             "select": [
                                 {"kind": "column", "column_id": "c_category", "alias": "category"},
-                                {
-                                    "kind": "agg",
-                                    "fn": "sum",
-                                    "column_id": "c_revenue",
-                                    "alias": "revenue",
-                                },
+                                {"kind": "column", "column_id": "c_revenue", "alias": "revenue"},
                             ],
                             "filters": [
                                 {
@@ -64,54 +66,36 @@ _EXAMPLE_A = TaskList(
                                     "value_type": "date",
                                 }
                             ],
-                            "group_by": ["c_category"],
-                            "order_by": [{"column_id": "revenue", "dir": "desc"}],
-                            "limit": 20,
+                            "limit": 10000,
                         }
                     },
                 )
             ],
-            expected_output="revenue_by_category",
-            success_criteria="Produced a ranked revenue figure per category.",
+            expected_output="quarter_rows",
+            success_criteria="Produced last quarter's order rows with category and revenue.",
             depends_on=["t1"],
-            parallelizable_with=["t3"],
-            estimated_cost="low",
+            parallelizable_with=[],
+            estimated_cost="medium",
         ),
         Task(
             id="t3",
             stage="evaluation",
-            objective="Get total last-quarter revenue to contextualize each category's share.",
+            objective="Rank each category's revenue share of the quarter total.",
             tool_calls=[
                 ToolCall(
-                    tool="query_structured",
+                    tool="analyze_contribution",
                     args={
-                        "ir": {
-                            "source_id": "src_sales",
-                            "table_id": "t_orders",
-                            "select": [
-                                {
-                                    "kind": "agg",
-                                    "fn": "sum",
-                                    "column_id": "c_revenue",
-                                    "alias": "total_revenue",
-                                }
-                            ],
-                            "filters": [
-                                {
-                                    "column_id": "c_order_date",
-                                    "op": "between",
-                                    "value": ["2026-01-01", "2026-03-31"],
-                                    "value_type": "date",
-                                }
-                            ],
-                        }
+                        "data": "${t2}",
+                        "dimension": "category",
+                        "value_column": "revenue",
+                        "agg": "sum",
                     },
                 )
             ],
-            expected_output="total_revenue",
-            success_criteria="Produced a single total revenue figure for the quarter.",
-            depends_on=["t1"],
-            parallelizable_with=["t2"],
+            expected_output="category_contribution",
+            success_criteria="Produced each category's revenue share, ranked high to low.",
+            depends_on=["t2"],
+            parallelizable_with=[],
             estimated_cost="low",
         ),
     ],
@@ -181,30 +165,28 @@ _EXAMPLE_B = TaskList(
         Task(
             id="t3",
             stage="evaluation",
-            objective="Bucket the order dates into months to form the monthly trend.",
+            objective="Bucket revenue into months and summarize the trend and movement.",
             tool_calls=[
                 ToolCall(
-                    tool="date_trunc",
-                    args={"values": "${t2}", "granularity": "month"},
+                    tool="analyze_trend",
+                    args={
+                        "data": "${t2}",
+                        "date_column": "order_date",
+                        "value_column": "revenue",
+                        "freq": "month",
+                        "agg": "sum",
+                    },
                 )
             ],
-            expected_output="monthly_series",
-            success_criteria="Produced a per-month revenue series.",
+            expected_output="monthly_trend",
+            success_criteria=(
+                "Produced a per-month revenue series with direction and change rate to "
+                "flag months above/below the typical level."
+            ),
             depends_on=["t2"],
             parallelizable_with=[],
             estimated_cost="low",
         ),
-        Task(
-            id="t4",
-            stage="evaluation",
-            objective="Quantify month-to-month spread to flag unusual months.",
-            tool_calls=[ToolCall(tool="compute_stddev", args={"values": "${t3}"})],
-            expected_output="monthly_volatility",
-            success_criteria="Produced a stddev figure that flags months above the typical spread.",
-            depends_on=["t3"],
-            parallelizable_with=[],
-            estimated_cost="low",
-        ),
     ],
 )
 

src/agents/planner/registry.py

@@ -1,19 +1,44 @@
-"""STUB v1 P0 tool registry.
+"""STUB v1 tool registry — composite ("family") tools.
 
 This is the agent team's local stand-in for the tool team's inventory (KM-608)
-so the planner is buildable and testable before the real tools land. The tools
-here are *contracts only* — there is no implementation behind them; the planner
-plans against the registry and never names a tool outside it (INV-7).
+so the planner is buildable and testable before the real wrapper layer lands.
+The tools here are *contracts only* — the compute logic for the `analyze_*`
+family already exists in `src/tools/analytics/` (KM-624), but the wrapper layer
+(source/placeholder -> DataFrame fetch, the `ToolOutput` envelope, never-throw
+error handling, ToolSpec registration) is still pending the Planner seam
+(KM-418 / AGENT_ARCHITECTURE_CONTEXT_new.md §8.4). The planner plans against the
+registry and never names a tool outside it (INV-7).
 
-`input_schema` is a lightweight JSON-schema-ish dict consumed by the planner
-validator (validator.py check #8): it carries `required` (list of arg names) and
-`properties` (allowed arg names). Arg *values* may be "${t<id>}" placeholders the
-TaskRunner resolves at execution time, so the validator checks arg *keys*, not
-value types — except `query_structured.args["ir"]`, whose inline QueryIR is
-validated against the catalog by the existing IRValidator.
+**Taxonomy decision (2026-06-08):** v1 uses **composite/family** tools, not the
+atomic `compute_*` set the earlier draft assumed. One `analyze_*` call does a
+whole analytical job (e.g. `analyze_descriptive` returns mean/median/mode/std/
+quartiles/skew/null_rate at once, replacing four atomic `compute_*` tools). See
+§9.3 / the decisions table in the architecture doc.
 
-When KM-608 ships, replace `default_registry()` with the real registry import.
-See AGENT_ARCHITECTURE_CONTEXT_new.md §9.2 / §9.3.
+**Ownership (revised 2026-06-08): the tool team owns ALL tools** — compute,
+data-access (`query_structured`/`retrieve_documents`/`list_sources`/
+`describe_source`), the wrapper/invoker, and tests. This file is purely the agent
+team's local scaffold for building/testing the Planner (and later the TaskRunner/
+Assembler against mocks) until the real registry lands; replace it then.
+
+**Data-flow convention (Pattern A — assumed, but the tool team's call, still open):**
+this stub assumes the `analyze_*` tools do NOT self-fetch by `source_id`; each
+takes a `data` argument that is a `"${t<id>}"` placeholder pointing at an upstream
+`query_structured` table output, resolved to a DataFrame at execution time. Column
+arguments (`column_ids`, `dimension`, `value_column`, `date_column`, …) reference
+the *aliases* the upstream query produced. If the tool team instead picks Pattern B
+(self-fetch by `source_id`), reshape this stub + the few-shot examples to match —
+the agent code does not change either way (INV-7).
+
+`input_schema` is the lightweight JSON-schema-ish dict the planner validator
+(validator.py check #8) consumes: `required` (list of arg names) + `properties`
+(allowed arg names). Arg *values* may be `"${t<id>}"` placeholders resolved at
+execution time, so the validator checks arg *keys*, not value types — except
+`query_structured.args["ir"]`, whose inline QueryIR is validated against the
+catalog by the existing IRValidator.
+
+When KM-608/KM-418 ship, replace `default_registry()` with the real registry
+import. See AGENT_ARCHITECTURE_CONTEXT_new.md §9.2 / §9.3.
 """
 
 from __future__ import annotations
@@ -21,6 +46,10 @@ from __future__ import annotations
 from .contracts import ToolRegistry, ToolSpec
 
 _P0_TOOLS: list[ToolSpec] = [
+    # ----------------------------------------------------------------------- #
+    # Data access + catalog introspection (agent-team owned; wrap existing
+    # Phase 2 infra — QueryService / RetrievalRouter / CatalogReader).
+    # ----------------------------------------------------------------------- #
     ToolSpec(
         name="query_structured",
         category="analytics.query",
@@ -30,11 +59,13 @@ _P0_TOOLS: list[ToolSpec] = [
             "Run one validated, single-table query against a structured source (DB "
             "schema or tabular file) and return rows. The `ir` argument is an inline "
             "QueryIR (the JSON intent: source_id, table_id, select, filters, group_by, "
-            "order_by, limit) — never SQL. Use this for any selection, filtering, "
-            "grouping, or built-in aggregation (count/sum/avg/min/max/count_distinct). "
-            "Do NOT use it for medians/percentiles/modes/stddev (use the compute_* "
-            "tools on its output) and do NOT use it to read documents (use "
-            "retrieve_documents)."
+            "order_by, limit) — never SQL. This is the data-access entry point: use it "
+            "to select, filter, and pull the rows the analytics (`analyze_*`) tools "
+            "then consume. It also does simple built-in aggregation the IR can express "
+            "(count/sum/avg/min/max/count_distinct). Do NOT use it for richer statistics "
+            "(median/percentile/mode/stddev/skew → analyze_descriptive), trends "
+            "(analyze_trend), correlation, segmentation, or share-of-total; and do NOT "
+            "use it to read documents (use retrieve_documents)."
         ),
     ),
     ToolSpec(
@@ -81,79 +112,194 @@ _P0_TOOLS: list[ToolSpec] = [
             "before querying it. Do NOT use it to fetch data rows (use query_structured)."
         ),
     ),
+    # ----------------------------------------------------------------------- #
+    # Analytics family (KM-624 compute; wrapper pending). Each takes `data` =
+    # a "${t<id>}" placeholder for an upstream query_structured table output.
+    # ----------------------------------------------------------------------- #
     ToolSpec(
-        name="compute_median",
-        category="analytics.aggregation",
-        input_schema={"required": ["values"], "properties": {"values": {"type": "array"}}},
-        output_kind="scalar",
+        name="analyze_descriptive",
+        category="analytics.descriptive",
+        input_schema={
+            "required": ["data", "column_ids"],
+            "properties": {
+                "data": {"type": "string"},
+                "column_ids": {"type": "array"},
+                "metrics": {"type": "array"},
+            },
+        },
+        output_kind="stats",
         description=(
-            "Compute the median of a numeric series. `values` is typically a "
-            "'${t<id>}' placeholder referencing an upstream query_structured output "
-            "column. Use this because SQL/pandas median is not exposed via the IR. Do "
-            "NOT use it on categorical data (use compute_mode)."
+            "Single/multi-column EDA in one call: count, mean, median, mode, std, "
+            "variance, quartiles (q1/q3), min, max, skew, null_count, null_rate for each "
+            "of `column_ids`. `data` is a '${t<id>}' placeholder for an upstream "
+            "query_structured result; `column_ids` are that result's column aliases. "
+            "This replaces the atomic compute_median/mode/stddev/percentile tools — ask "
+            "for the whole profile, not one statistic at a time. Do NOT use it for "
+            "group-by aggregates (analyze_aggregate) or time trends (analyze_trend)."
         ),
     ),
     ToolSpec(
-        name="compute_stddev",
+        name="analyze_aggregate",
         category="analytics.aggregation",
-        input_schema={"required": ["values"], "properties": {"values": {"type": "array"}}},
-        output_kind="scalar",
+        input_schema={
+            "required": ["data", "aggregations"],
+            "properties": {
+                "data": {"type": "string"},
+                "aggregations": {"type": "object"},
+                "group_by": {"type": "array"},
+            },
+        },
+        output_kind="table",
         description=(
-            "Compute the standard deviation of a numeric series (`values`, usually a "
-            "'${t<id>}' placeholder from an upstream query). Use to quantify spread or "
-            "to flag outliers. Do NOT use on non-numeric data."
+            "Group-by aggregation over an already-materialized result: per group, "
+            "compute `aggregations` like {\"revenue\": [\"sum\", \"mean\"], "
+            "\"order_id\": [\"count\"]} (sum/mean/count/min/max/median/nunique). `data` "
+            "is a '${t<id>}' placeholder; `group_by` columns and aggregated columns are "
+            "that result's aliases. Prefer query_structured for simple group-by the IR "
+            "can already express; use this to aggregate a derived/joined/intermediate "
+            "result, or for median per group (the IR cannot)."
         ),
     ),
     ToolSpec(
-        name="compute_percentile",
-        category="analytics.aggregation",
+        name="analyze_comparison",
+        category="analytics.comparison",
         input_schema={
-            "required": ["values", "percentile"],
+            "required": ["data", "dimension", "value_column", "group_a", "group_b"],
             "properties": {
-                "values": {"type": "array"},
-                "percentile": {"type": "number"},
+                "data": {"type": "string"},
+                "dimension": {"type": "string"},
+                "value_column": {"type": "string"},
+                "group_a": {},
+                "group_b": {},
+                "agg": {"type": "string"},
             },
         },
-        output_kind="scalar",
+        output_kind="stats",
         description=(
-            "Compute a given `percentile` (0-100) of a numeric series `values` "
-            "(usually a '${t<id>}' placeholder). Use for p90/p95-style thresholds. Do "
-            "NOT use for the median alone (use compute_median)."
+            "Compare one aggregated metric between two groups of a dimension (e.g. "
+            "region 'A' vs 'B'): returns each group's value, absolute and percent "
+            "difference, and direction (higher/lower/equal); group_a is the baseline. "
+            "`data` is a '${t<id>}' placeholder; `dimension`/`value_column` are aliases; "
+            "`agg` defaults to sum. Use for exactly TWO groups. For many categories' "
+            "share of a total use analyze_contribution; for movement over time use "
+            "analyze_trend."
         ),
     ),
     ToolSpec(
-        name="compute_mode",
-        category="analytics.aggregation",
-        input_schema={"required": ["values"], "properties": {"values": {"type": "array"}}},
-        output_kind="scalar",
+        name="analyze_contribution",
+        category="analytics.decomposition",
+        input_schema={
+            "required": ["data", "dimension", "value_column"],
+            "properties": {
+                "data": {"type": "string"},
+                "dimension": {"type": "string"},
+                "value_column": {"type": "string"},
+                "agg": {"type": "string"},
+                "top_n": {"type": "integer"},
+            },
+        },
+        output_kind="table",
+        description=(
+            "Share-of-total breakdown: each category's value, share, and running "
+            "cumulative share, largest first — the tool for 'which categories drive "
+            "most of X?' and Pareto (80/20) reasoning. `data` is a '${t<id>}' "
+            "placeholder; `dimension`/`value_column` are aliases; `agg` defaults to sum; "
+            "`top_n` lumps the tail into an 'Others' row. Use for a single snapshot of "
+            "many categories. Do NOT use it to compare exactly two groups "
+            "(analyze_comparison) or to trend over time (analyze_trend)."
+        ),
+    ),
+    ToolSpec(
+        name="analyze_profile",
+        category="analytics.quality",
+        input_schema={
+            "required": ["data"],
+            "properties": {"data": {"type": "string"}, "column_ids": {"type": "array"}},
+        },
+        output_kind="stats",
+        description=(
+            "Per-column data-quality profile: dtype, inferred type, completeness "
+            "(null_count/null_rate), cardinality (distinct_count/rate, is_constant), and "
+            "for numeric columns min/max/mean plus an IQR-based outlier_count (top value "
+            "for non-numeric). `data` is a '${t<id>}' placeholder; `column_ids` defaults "
+            "to all columns. Use in data_understanding to judge whether data is clean "
+            "enough before deeper analysis. Do NOT use it for the analytical answer "
+            "itself — it describes data health, not the business metric."
+        ),
+    ),
+    ToolSpec(
+        name="analyze_correlation",
+        category="analytics.relationship",
+        input_schema={
+            "required": ["data"],
+            "properties": {
+                "data": {"type": "string"},
+                "column_ids": {"type": "array"},
+                "method": {"type": "string"},
+            },
+        },
+        output_kind="stats",
+        description=(
+            "Pairwise correlation across numeric columns: returns the full matrix plus "
+            "column pairs ranked by strength. `data` is a '${t<id>}' placeholder; "
+            "`column_ids` defaults to all numeric columns; `method` is pearson "
+            "(default), spearman, or kendall. Use for 'does X relate to Y?'. Needs at "
+            "least two numeric columns. Correlation is not causation — it does not "
+            "explain why, and is not a model."
+        ),
+    ),
+    ToolSpec(
+        name="analyze_segment",
+        category="analytics.segmentation",
+        input_schema={
+            "required": ["data", "column", "bins"],
+            "properties": {
+                "data": {"type": "string"},
+                "column": {"type": "string"},
+                "bins": {},
+                "method": {"type": "string"},
+                "labels": {"type": "array"},
+                "value_column": {"type": "string"},
+                "agg": {"type": "string"},
+            },
+        },
+        output_kind="table",
         description=(
-            "Compute the most frequent value(s) of a series `values` (usually a "
-            "'${t<id>}' placeholder). Works on categorical or numeric data. Use to find "
-            "the typical category. Do NOT use it for an average (use query_structured "
-            "avg)."
+            "Bucket rows by binning a numeric `column` and report how rows distribute "
+            "across segments (count, and optionally an aggregate of `value_column` per "
+            "segment). `method` 'edges' takes explicit boundaries in `bins` (e.g. "
+            "[0,18,35,60]); 'quantile' takes an integer bucket count (e.g. 4 for "
+            "quartiles). `data` is a '${t<id>}' placeholder; columns are aliases. Use "
+            "for age brackets, value tiers, etc. The binned column must be numeric."
         ),
     ),
     ToolSpec(
-        name="date_trunc",
+        name="analyze_trend",
         category="analytics.timeseries",
         input_schema={
-            "required": ["values", "granularity"],
+            "required": ["data", "date_column", "value_column"],
             "properties": {
-                "values": {"type": "array"},
-                "granularity": {"type": "string"},
+                "data": {"type": "string"},
+                "date_column": {"type": "string"},
+                "value_column": {"type": "string"},
+                "freq": {"type": "string"},
+                "agg": {"type": "string"},
             },
         },
         output_kind="series",
         description=(
-            "Truncate a datetime series `values` (usually a '${t<id>}' placeholder) to "
-            "a `granularity` ('day' | 'week' | 'month' | 'quarter' | 'year') so results "
-            "can be grouped into time buckets for trend analysis. Do NOT use it to "
-            "filter by date — put a date filter in the query_structured IR instead."
+            "Time-series trend in one call: bucket rows into periods (`freq` = "
+            "day/week/month/quarter/year), aggregate `value_column` per period (`agg` "
+            "defaults to sum), and summarize movement (per-period points, first vs last, "
+            "absolute/percent change, direction, linear slope). `data` is a '${t<id>}' "
+            "placeholder; `date_column`/`value_column` are aliases from the upstream "
+            "query. This replaces the atomic date_trunc tool. Do NOT use it to filter by "
+            "date — put the date filter in the query_structured IR instead."
         ),
     ),
 ]
 
 
 def default_registry() -> ToolRegistry:
-    """The v1 P0 stub registry (a fresh instance per call)."""
+    """The v1 stub registry (a fresh instance per call)."""
     return ToolRegistry(tools=list(_P0_TOOLS))

src/config/prompts/planner.md

@@ -31,10 +31,14 @@ only a `TaskList` object that conforms to the provided schema.
 - **Wire data between tasks with placeholders.** When a task needs an upstream
   task's output as an argument, use the string `"${t<id>}"` (e.g. `"${t2}"`) as
   the argument value. Set `depends_on` accordingly.
-- **Built-in aggregation vs compute_* tools.** Use `query_structured` for
-  count/sum/avg/min/max/count_distinct, filtering, and grouping. For statistics
-  the IR cannot express (median, percentile, mode, standard deviation), run
-  `query_structured` to fetch the series, then a `compute_*` tool on its output.
+- **Data access vs analytics tools.** `query_structured` is the data-access entry
+  point: use it to select, filter, and pull rows (and simple built-in
+  count/sum/avg/min/max/count_distinct the IR can express). For anything richer —
+  descriptive statistics (median/percentile/mode/std/skew), time trends, group
+  comparisons, share-of-total, correlation, segmentation, or data-quality
+  profiling — run `query_structured` 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).
 - **Mixing structured + unstructured.** If qualitative context helps, add a
   `retrieve_documents` task against an unstructured source listed in the catalog.
 - **Parallelism.** List sibling tasks that have no data dependency on each other