[NOTICKET] test: planner eval harness (rule-compliance, --selfcheck/--rescore)

Add eval/planner/ — a regression net for the live planner, grounded in the
PA Data Dummy catalog. Scores each golden question by RULE-COMPLIANCE
assertions on the emitted plan/IR (count uses count(); entity ranking uses
group_by + agg + order + limit; fuzzy model filter never enumerates from
samples) instead of exact-IR match, since many IRs are valid. Includes
carried_over regression cases + counter-examples (raw-row listing,
exact-match filters) so a planner.md / examples.py change can be gated
before deploy.

Assertions are grouping-aware — they honor the analyze_aggregate tool
(group_by/aggregations args, mean==avg), not only IR group_by — and resolve
analyze_aggregate aliases back to columns via the retrieve_data select.

Runner modes: --selfcheck (validate the scorer offline against synthetic
good/buggy plans, no LLM) and --rescore RESULTS.json (re-score a saved run's
persisted facts after editing assertions, no LLM). Baseline on the current
planner: 27/27.

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

eval/planner/README.md

@@ -0,0 +1,54 @@
+# Planner eval (E-planner)
+
+Scores the **live planner** (`PlannerService.plan`) on golden questions against a
+fixture catalog grounded in `PA Data Dummy.xlsx`. Purpose: a **regression net**
+for changes to `planner.md` / `examples.py` — before/after any prompt tweak, run
+this and confirm the target cases improve while `carried_over` cases stay green.
+
+## Why rule-compliance scoring (not exact-match)
+
+A question has **many valid IRs**, so we don't compare IRs verbatim. Each case
+pins only the **properties that matter** (`expect` assertions): does a count
+question use a `count` aggregate? does entity ranking `group_by` the entity? does
+a fuzzy model filter use `like` instead of an enumerated `in`? See `_expect_keys`
+in `planner_dataset.json` for the full assertion vocabulary.
+
+## Run
+
+```bash
+uv run python -m eval.planner.run_eval            # full run (needs Azure creds)
+uv run python -m eval.planner.run_eval --limit 6  # smoke test
+uv run python -m eval.planner.run_eval --selfcheck  # test the scorer, no LLM
+```
+
+Each run writes `results/planner_result_<timestamp>.json` (never overwritten).
+`id` is stable per case, so runs diff case-by-case over time.
+
+## What's covered
+
+| category | targets |
+|---|---|
+| `count` | scalar count → `count` aggregate (shipped fix) |
+| `ranking` | top/bottom-N entities → `group_by` + `avg` + `order_by` + `limit` (**Bug 1**) |
+| `fuzzy_filter` | partial model ref → `like`, never enumerate from samples (**Bug 2**) |
+| `aggregate`, `descriptive`, `correlation`, `trend`, `merge` | believed-correct baselines |
+| `counter_raw_rows` | "show N records" must stay raw rows (guards Bug 1 fix from over-aggregating) |
+| `counter_exact_filter` | exact filters stay exact (guards Bug 2 fix from over-`like`ing) |
+| `infeasible` | measures absent from the catalog → `infeasible_reason` |
+
+`carried_over: true` = behavior believed correct today (regression guard);
+`false` = the known bugs. **Expected baseline (before the planner fixes):** the
+`ranking` and `fuzzy_filter` (777) cases FAIL, everything else green — that gap is
+exactly what the planner fixes should close, without turning any `carried_over`
+case red.
+
+## Files
+
+- `planner_dataset.json` — cases (question + `expect` assertions)
+- `catalog_fixture.py` — the `PA Data Dummy` catalog the planner plans against
+- `run_eval.py` — runner + deterministic scorer (`--selfcheck`)
+
+> Date columns are typed `date` in the fixture (the *post-fix* catalog). The live
+> system currently mis-types Excel date serials as `int` — an **ingest** bug, not
+> a planner one — so the fixture types them correctly to keep this eval about
+> planner logic.

eval/planner/catalog_fixture.py

@@ -0,0 +1,96 @@
+"""Fixture catalog for the planner eval — grounded in Sofhia's real test file
+`PA Data Dummy.xlsx` (mining equipment physical-availability data, 9729 daily
+rows, single site, April 2026).
+
+Column ids are readable (`c_<snake_name>`) on purpose: the planner only echoes
+whatever ids the summary gives it, and readable ids make the result files easy
+to diff. `name_to_id()` maps a column NAME back to its id so the assertions in
+`run_eval.py` can be written against names.
+
+NOTE: date columns are typed `date` here — i.e. the *post-fix* catalog. The live
+system currently mis-types Excel date serials as `int` (a Go-ingest bug, see the
+date-handling note), which is an INGEST issue, not a planner one. Typing them
+correctly here keeps this eval about planner logic, not the ingest bug.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from src.catalog.models import Catalog, Column, ColumnStats, DataType, Source, Table
+
+SOURCE_ID = "src_pa"
+TABLE_ID = "t_pa"
+ROW_COUNT = 9729
+
+# (name, data_type, sample_values, top_values)
+_COLUMNS: list[tuple[str, DataType, list[Any] | None, list[Any] | None]] = [
+    ("KeyId", "int", [895272, 895245, 895285], None),
+    ("Month_ID", "int", [202604], [202604]),
+    ("Site_ID", "int", [2009], [2009]),
+    ("From_Date", "date", ["2026-04-06", "2026-04-25"], None),
+    ("To_Date", "date", ["2026-04-06", "2026-04-25"], None),
+    ("Time_Description", "string", ["Daily"], ["Daily"]),
+    # High-cardinality but NOT unique — a valid ranking/group dimension.
+    ("Model_Unit", "string", ["777E", "777D", "777", "785D", "789C", "HD785-7", "PC2000-8", "EX3600-6"], None),
+    # Per-unit identifier (high cardinality). Ranking BY unit must still group by it.
+    ("Equipment_Number", "string", ["HDCT77457", "HDCT77455", "EXHC36007"], None),
+    ("Equipment_Group_ID", "string", ["Main Hauler", "Main Loader"], ["Main Hauler", "Main Loader"]),
+    ("Unit_Status", "string", ["INPR"], ["INPR"]),
+    ("Total_Breakdown_Schedule_Hour", "decimal", [19.416, 0.0], None),
+    ("Total_Breakdown_Unschedule_Hour", "decimal", [0.0, 3.728], None),
+    ("Total_Adj_Breakdown_Schedule_Hour", "decimal", [19.416, 0.0], None),
+    ("Total_Adj_Breakdown_Unschedule_Hour", "decimal", [0.0, 2.982], None),
+    ("Total_MTC_Hour", "decimal", [19.4164, 0.0], None),
+    ("Total_Down_Hour", "decimal", [19.4164, 0.0], None),
+    ("Total_Frequency_Breakdown_Schedule", "int", [1, 0], None),
+    ("Total_Frequency_Breakdown_Unschedule", "int", [0, 3], None),
+    ("Total_Frequency_Maintenance", "int", [1, 3], None),
+    ("Total_Frequency_Tire", "int", [0], None),
+    ("Total_Frequency_Down", "int", [1, 3], None),
+    ("Total_Hours", "int", [24], [24]),
+    ("Total_INPR_Hour", "int", [24, 0], None),
+    ("Total_Record_HM_Hour", "decimal", [0.0], None),
+    ("Total_HM_Mtc_Down_Hour", "int", [0, 20], None),
+    ("Plan_PA_Percent", "decimal", [100.0, 91.46], None),
+    ("PA_Percent", "decimal", [19.0984, 100.0, 84.4676], None),
+    ("MTBS", "decimal", [0.0, 5.4667], None),
+    ("MTTR", "decimal", [19.4164, 1.2426, 0.0], None),
+    ("SM_Percent", "decimal", [100.0, 0.0], None),
+    ("Unschedule_SM_Percent", "decimal", [0.0, 100.0], None),
+    ("IsDeleted", "int", [0], [0]),
+    ("Section", "string", ["OB HAULER", "OB LOADER"], ["OB HAULER", "OB LOADER"]),
+    ("Week_ID", "int", [202617], None),
+    ("Plan_PA_Percent_2", "decimal", [88.0, 91.0], None),
+    ("Updated_Date", "datetime", ["2026-04-25T00:45:17"], None),
+]
+
+
+def name_to_id() -> dict[str, str]:
+    return {name: f"c_{name.lower()}" for name, *_ in _COLUMNS}
+
+
+def build_pa_catalog() -> Catalog:
+    columns = [
+        Column(
+            column_id=f"c_{name.lower()}",
+            name=name,
+            data_type=dtype,
+            nullable=False,
+            pii_flag=False,
+            sample_values=samples,
+            stats=ColumnStats(distinct_count=len(top) if top else None, top_values=top),
+        )
+        for name, dtype, samples, top in _COLUMNS
+    ]
+    table = Table(table_id=TABLE_ID, name="PA Data Dummy", row_count=ROW_COUNT, columns=columns, foreign_keys=[])
+    source = Source(
+        source_id=SOURCE_ID,
+        source_type="tabular",
+        name="PA Data Dummy.xlsx",
+        location_ref="object_storage://eval/pa",
+        updated_at=datetime.now(UTC),
+        tables=[table],
+    )
+    return Catalog(user_id="eval-user", sources=[source], generated_at=datetime.now(UTC))

eval/planner/planner_dataset.json

@@ -0,0 +1,234 @@
+{
+  "_about": "Golden planner dataset (E-planner eval — runs against the LIVE planner LLM, PlannerService.plan). Each case is a question + `expect` = a set of IR/plan ASSERTIONS (rule-compliance scoring, NOT exact IR match — many IRs are valid, we only pin the properties that matter). Grounded in `PA Data Dummy.xlsx` via eval/planner/catalog_fixture.py. `id` is a stable per-case handle so timestamped runs (results/planner_result_<ts>.json) diff case-by-case. `carried_over`=true marks behavior believed CORRECT today (regression guard); the rest target the 3 known planner bugs. Balanced ID/EN because users code-switch.",
+  "_expect_keys": {
+    "has_tool": "some task tool_call uses this tool name",
+    "no_tool": "no task uses this tool",
+    "select_agg": "some retrieve_data IR select has an agg with this fn (count/sum/avg/min/max/count_distinct)",
+    "group_by": "true = some IR has a non-empty group_by; false = NO IR has group_by (raw-row guard)",
+    "group_by_col": "some IR group_by contains this column NAME",
+    "filter_op": "some IR filter uses this op",
+    "no_filter_op": "NO IR filter uses this op",
+    "has_filter": "true = at least one IR has a filter",
+    "order_dir": "some IR order_by uses this dir (asc/desc)",
+    "limit": "some IR has exactly this limit",
+    "infeasible": "true = plan has no tasks / an infeasible_reason (measure not in catalog)"
+  },
+  "cases": [
+    {
+      "id": "count_zero_pa",
+      "category": "count",
+      "lang": "en",
+      "question": "how many records have PA_Percent = 0?",
+      "expect": {"select_agg": "count", "has_filter": true, "group_by": false},
+      "carried_over": true
+    },
+    {
+      "id": "count_mttr_gt20_id",
+      "category": "count",
+      "lang": "id",
+      "question": "berapa banyak record dengan MTTR di atas 20?",
+      "expect": {"select_agg": "count", "has_filter": true},
+      "carried_over": true
+    },
+    {
+      "id": "count_section_hauler",
+      "category": "count",
+      "lang": "en",
+      "question": "how many rows are in section OB HAULER?",
+      "expect": {"select_agg": "count", "has_filter": true},
+      "carried_over": true
+    },
+    {
+      "id": "rank_units_worst_pa_id",
+      "category": "ranking",
+      "lang": "id",
+      "question": "5 unit dengan PA terburuk?",
+      "expect": {"group_by": true, "group_by_col": "Equipment_Number", "select_agg": "avg", "order_dir": "asc", "limit": 5},
+      "carried_over": false
+    },
+    {
+      "id": "rank_models_top_mttr_id",
+      "category": "ranking",
+      "lang": "id",
+      "question": "top 3 model dengan MTTR tertinggi?",
+      "expect": {"group_by": true, "group_by_col": "Model_Unit", "select_agg": "avg", "order_dir": "desc", "limit": 3},
+      "carried_over": true
+    },
+    {
+      "id": "rank_sections_lowest_pa_en",
+      "category": "ranking",
+      "lang": "en",
+      "question": "which section has the lowest average PA?",
+      "expect": {"group_by": true, "group_by_col": "Section", "select_agg": "avg"},
+      "carried_over": true
+    },
+    {
+      "id": "rank_units_most_breakdown_id",
+      "category": "ranking",
+      "lang": "id",
+      "question": "unit mana yang paling sering breakdown?",
+      "expect": {"group_by": true, "group_by_col": "Equipment_Number", "order_dir": "desc"},
+      "carried_over": false
+    },
+    {
+      "id": "rank_units_worst_pa_en",
+      "category": "ranking",
+      "lang": "en",
+      "question": "list the 10 worst units by availability",
+      "expect": {"group_by": true, "group_by_col": "Equipment_Number", "select_agg": "avg", "order_dir": "asc", "limit": 10},
+      "carried_over": false
+    },
+    {
+      "id": "fuzzy_model_777_id",
+      "category": "fuzzy_filter",
+      "lang": "id",
+      "question": "berapa banyak model 777?",
+      "expect": {"select_agg": "count", "no_filter_op": "in"},
+      "carried_over": false
+    },
+    {
+      "id": "fuzzy_model_hd785_id",
+      "category": "fuzzy_filter",
+      "lang": "id",
+      "question": "berapa banyak unit HD785?",
+      "expect": {"select_agg": "count", "no_filter_op": "in"},
+      "carried_over": true
+    },
+    {
+      "id": "fuzzy_model_ex_en",
+      "category": "fuzzy_filter",
+      "lang": "en",
+      "question": "how many EX excavator units are there?",
+      "expect": {"no_filter_op": "in"},
+      "carried_over": false
+    },
+    {
+      "id": "agg_pa_per_section_id",
+      "category": "aggregate",
+      "lang": "id",
+      "question": "berapa rata-rata PA per section?",
+      "expect": {"group_by": true, "group_by_col": "Section", "select_agg": "avg"},
+      "carried_over": true
+    },
+    {
+      "id": "agg_mttr_per_model_en",
+      "category": "aggregate",
+      "lang": "en",
+      "question": "what is the average MTTR per model unit?",
+      "expect": {"group_by": true, "group_by_col": "Model_Unit", "select_agg": "avg"},
+      "carried_over": true
+    },
+    {
+      "id": "agg_downhour_per_group_id",
+      "category": "aggregate",
+      "lang": "id",
+      "question": "total down hour per equipment group?",
+      "expect": {"group_by": true, "group_by_col": "Equipment_Group_ID", "select_agg": "sum"},
+      "carried_over": true
+    },
+    {
+      "id": "desc_mttr_stats_id",
+      "category": "descriptive",
+      "lang": "id",
+      "question": "berikan ringkasan statistik MTTR",
+      "expect": {"has_tool": "analyze_descriptive"},
+      "carried_over": true
+    },
+    {
+      "id": "desc_pa_stats_en",
+      "category": "descriptive",
+      "lang": "en",
+      "question": "give me the summary statistics for PA_Percent",
+      "expect": {"has_tool": "analyze_descriptive"},
+      "carried_over": true
+    },
+    {
+      "id": "corr_mttr_pa_id",
+      "category": "correlation",
+      "lang": "id",
+      "question": "apakah ada korelasi antara MTTR dan PA?",
+      "expect": {"has_tool": "analyze_correlation"},
+      "carried_over": true
+    },
+    {
+      "id": "corr_freq_pa_en",
+      "category": "correlation",
+      "lang": "en",
+      "question": "is breakdown frequency correlated with availability?",
+      "expect": {"has_tool": "analyze_correlation"},
+      "carried_over": true
+    },
+    {
+      "id": "trend_pa_daily_id",
+      "category": "trend",
+      "lang": "id",
+      "question": "bagaimana trend PA harian?",
+      "expect": {"has_tool": "analyze_trend"},
+      "carried_over": true
+    },
+    {
+      "id": "trend_downhour_en",
+      "category": "trend",
+      "lang": "en",
+      "question": "show the trend of total down hours over time",
+      "expect": {"has_tool": "analyze_trend"},
+      "carried_over": true
+    },
+    {
+      "id": "merge_worst_pa_and_mttr_id",
+      "category": "merge",
+      "lang": "id",
+      "question": "model mana yang PA-nya paling buruk sekaligus MTTR-nya paling tinggi?",
+      "expect": {"group_by": true, "group_by_col": "Model_Unit"},
+      "carried_over": true
+    },
+    {
+      "id": "raw_rows_low_pa_id",
+      "category": "counter_raw_rows",
+      "lang": "id",
+      "question": "tampilkan 10 record dengan PA di bawah 50",
+      "expect": {"group_by": false, "has_filter": true, "limit": 10},
+      "carried_over": true
+    },
+    {
+      "id": "raw_rows_head_en",
+      "category": "counter_raw_rows",
+      "lang": "en",
+      "question": "show me the first 5 rows of the data",
+      "expect": {"group_by": false},
+      "carried_over": true
+    },
+    {
+      "id": "exact_model_777d_id",
+      "category": "counter_exact_filter",
+      "lang": "id",
+      "question": "berapa banyak record untuk model 777D?",
+      "expect": {"select_agg": "count", "has_filter": true},
+      "carried_over": true
+    },
+    {
+      "id": "exact_section_loader_en",
+      "category": "counter_exact_filter",
+      "lang": "en",
+      "question": "how many records are in the OB LOADER section?",
+      "expect": {"select_agg": "count", "has_filter": true},
+      "carried_over": true
+    },
+    {
+      "id": "infeasible_churn_id",
+      "category": "infeasible",
+      "lang": "id",
+      "question": "berapa churn rate pelanggan?",
+      "expect": {"infeasible": true},
+      "carried_over": true
+    },
+    {
+      "id": "infeasible_profit_en",
+      "category": "infeasible",
+      "lang": "en",
+      "question": "what is the monthly profit margin?",
+      "expect": {"infeasible": true},
+      "carried_over": true
+    }
+  ]
+}

eval/planner/run_eval.py

@@ -0,0 +1,391 @@
+"""Planner eval runner (E-planner).
+
+Feeds each golden case in `planner_dataset.json` to the LIVE planner
+(`PlannerService.plan`) against the `PA Data Dummy` fixture catalog, then scores
+each case by RULE-COMPLIANCE assertions on the emitted plan/IR (not exact IR
+match — many IRs are valid; we only pin the properties that matter). Records
+latency + token usage, prints a per-case + aggregate summary, and writes a
+timestamped JSON report under `results/` (never overwritten — diff runs over
+time).
+
+Run before any deploy that touches planner.md or examples.py:
+
+    uv run python -m eval.planner.run_eval
+    uv run python -m eval.planner.run_eval --limit 6     # quick smoke test
+
+Needs Azure OpenAI creds in the env (same as the live planner). The scoring
+layer is deterministic and unit-tested via `--selfcheck` (no LLM call).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import statistics
+import time
+from dataclasses import asdict, dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from langchain_core.callbacks import BaseCallbackHandler
+from langchain_core.outputs import LLMResult
+
+from src.agents.planner.contracts import BusinessContext
+from src.agents.planner.inputs import Constraints
+from src.agents.planner.registry import default_registry
+from src.agents.planner.service import PlannerService
+
+from .catalog_fixture import build_pa_catalog, name_to_id
+
+_HERE = Path(__file__).resolve().parent
+DATASET = _HERE / "planner_dataset.json"
+RESULTS_DIR = _HERE / "results"
+
+_CONTEXT = BusinessContext(
+    project_id="eval-planner",
+    industry="mining",
+    completeness="partial",
+    business_description=(
+        "Physical Availability (PA) analysis of heavy mining equipment — haulers "
+        "and loaders — from daily operational records."
+    ),
+    scale_and_scope="9,729 daily equipment records, single site, April 2026.",
+)
+
+
+# --------------------------------------------------------------------------- #
+# Plan introspection + assertion scoring (deterministic — no LLM)
+# --------------------------------------------------------------------------- #
+
+# Grouping/agg can be expressed EITHER in the retrieve_data IR (group_by + agg
+# select) OR via the analyze_aggregate tool (group_by + aggregations args) — both
+# are valid (planner.md R2). Checks below look at both. IR uses fn "avg"; the
+# aggregate tool uses "mean" — treat as synonyms.
+_AGG_SYN = {
+    "avg": {"avg", "mean"}, "mean": {"avg", "mean"}, "sum": {"sum"},
+    "count": {"count"}, "min": {"min"}, "max": {"max"},
+    "count_distinct": {"count_distinct", "nunique"},
+}
+
+
+def extract_facts(task_list: Any) -> dict:
+    """Flatten a TaskList into the plain facts the scorer needs (JSON-safe, so a
+    run can be re-scored offline via --rescore without re-calling the LLM)."""
+    irs: list[dict] = []
+    agg_args: list[dict] = []
+    tools: set[str] = set()
+    for t in task_list.tasks:
+        for c in t.tool_calls:
+            tools.add(c.tool)
+            if c.tool == "retrieve_data" and isinstance(c.args.get("ir"), dict):
+                irs.append(c.args["ir"])
+            elif c.tool == "analyze_aggregate":
+                agg_args.append(c.args)
+    return {
+        "tools": sorted(tools),
+        "irs": irs,
+        "agg_args": agg_args,
+        "infeasible": (not task_list.tasks) or bool(getattr(task_list, "infeasible_reason", None)),
+    }
+
+
+def _ir_agg_fns(ir: dict) -> list[str]:
+    return [s.get("fn") for s in ir.get("select", []) if isinstance(s, dict) and s.get("kind") == "agg"]
+
+
+def _all_agg_fns(f: dict) -> list[str]:
+    fns = [fn for ir in f["irs"] for fn in _ir_agg_fns(ir)]
+    for a in f["agg_args"]:
+        for lst in (a.get("aggregations") or {}).values():
+            fns += lst if isinstance(lst, list) else [lst]
+    return fns
+
+
+def _group_by_ids(f: dict) -> list[str]:
+    return [g for ir in f["irs"] for g in (ir.get("group_by") or [])]
+
+
+def _group_by_aliases(f: dict) -> list[str]:
+    return [str(g) for a in f["agg_args"] for g in (a.get("group_by") or [])]
+
+
+def _alias_to_id(f: dict) -> dict[str, str]:
+    """Map each retrieve_data SELECT alias -> its column_id, so an
+    analyze_aggregate group_by (which references aliases) resolves back to a
+    real column even when the planner aliases it differently from the name."""
+    m: dict[str, str] = {}
+    for ir in f["irs"]:
+        for s in ir.get("select", []):
+            if isinstance(s, dict) and s.get("alias") and s.get("column_id"):
+                m[s["alias"]] = s["column_id"]
+    return m
+
+
+def _filter_ops(f: dict) -> list[str]:
+    return [flt.get("op") for ir in f["irs"] for flt in ir.get("filters", []) if isinstance(flt, dict)]
+
+
+def evaluate_facts(f: dict, expect: dict, n2id: dict[str, str]) -> list[tuple[str, bool, str]]:
+    """Return [(check, passed, detail)] for every assertion. Grouping/agg checks
+    honor BOTH the IR and the analyze_aggregate tool."""
+    irs, tools = f["irs"], set(f["tools"])
+    grouped = bool(_group_by_ids(f) or _group_by_aliases(f))
+    res: list[tuple[str, bool, str]] = []
+
+    for key, want in expect.items():
+        if key == "has_tool":
+            res.append((f"has_tool={want}", want in tools, f"tools={sorted(tools)}"))
+        elif key == "no_tool":
+            res.append((f"no_tool={want}", want not in tools, f"tools={sorted(tools)}"))
+        elif key == "select_agg":
+            syn = _AGG_SYN.get(want, {want})
+            got = _all_agg_fns(f)
+            res.append((f"select_agg={want}", any(g in syn for g in got), f"aggs={got}"))
+        elif key == "group_by":
+            res.append(("group_by" if want else "no_group_by", grouped == want, f"grouped={grouped}"))
+        elif key == "group_by_col":
+            col_id = n2id.get(want, want)
+            a2id = _alias_to_id(f)
+            resolved = _group_by_ids(f) + [a2id.get(a) for a in _group_by_aliases(f)]
+            hit = col_id in resolved or want.lower() in [a.lower() for a in _group_by_aliases(f)]
+            res.append((f"group_by_col={want}", hit, f"ids={_group_by_ids(f)} aliases={_group_by_aliases(f)} resolved={[r for r in resolved if r]}"))
+        elif key == "filter_op":
+            got = _filter_ops(f)
+            res.append((f"filter_op={want}", want in got, f"ops={got}"))
+        elif key == "no_filter_op":
+            got = _filter_ops(f)
+            res.append((f"no_filter_op={want}", want not in got, f"ops={got}"))
+        elif key == "has_filter":
+            has = any(ir.get("filters") for ir in irs)
+            res.append(("has_filter", has == want, f"filter_present={has}"))
+        elif key == "order_dir":
+            got = [o.get("dir", "asc") for ir in irs for o in ir.get("order_by", []) if isinstance(o, dict)]
+            res.append((f"order_dir={want}", want in got, f"dirs={got}"))
+        elif key == "limit":
+            got = [ir.get("limit") for ir in irs]
+            res.append((f"limit={want}", want in got, f"limits={got}"))
+        elif key == "infeasible":
+            res.append(("infeasible" if want else "feasible", f["infeasible"] == want, f"infeasible={f['infeasible']}"))
+        else:
+            res.append((f"UNKNOWN:{key}", False, "unknown expect key"))
+    return res
+
+
+# --------------------------------------------------------------------------- #
+# Token callback (parity with intent eval)
+# --------------------------------------------------------------------------- #
+
+class _TokenCounter(BaseCallbackHandler):
+    def __init__(self) -> None:
+        self.total = 0
+
+    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
+        for gen_list in response.generations:
+            for gen in gen_list:
+                msg = getattr(gen, "message", None)
+                usage = getattr(msg, "usage_metadata", None) if msg else None
+                if usage:
+                    self.total += usage.get("total_tokens", 0)
+
+
+# --------------------------------------------------------------------------- #
+# Runner
+# --------------------------------------------------------------------------- #
+
+@dataclass
+class CaseResult:
+    id: str
+    category: str
+    lang: str
+    carried_over: bool
+    question: str
+    passed: bool
+    checks: list[dict] = field(default_factory=list)
+    facts: dict = field(default_factory=dict)  # extracted plan (for offline --rescore)
+    error: str | None = None
+    latency_ms: int = 0
+    tokens: int = 0
+
+
+async def _run_case(planner: PlannerService, catalog: Any, tools: Any, n2id: dict, case: dict) -> CaseResult:
+    tok = _TokenCounter()
+    started = time.perf_counter()
+    facts: dict = {}
+    try:
+        task_list = await planner.plan(
+            _CONTEXT, catalog, tools, case["question"], Constraints(), callbacks=[tok]
+        )
+        facts = extract_facts(task_list)
+        checks = evaluate_facts(facts, case["expect"], n2id)
+        passed = all(ok for _, ok, _ in checks)
+        err = None
+    except Exception as e:  # planner failure = case fails (record why)
+        checks, passed, err = [], False, f"{type(e).__name__}: {e}"
+    latency = int((time.perf_counter() - started) * 1000)
+    return CaseResult(
+        id=case["id"], category=case["category"], lang=case["lang"],
+        carried_over=case.get("carried_over", False), question=case["question"],
+        passed=passed, error=err, latency_ms=latency, tokens=tok.total, facts=facts,
+        checks=[{"check": c, "ok": ok, "detail": d} for c, ok, d in checks],
+    )
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--limit", type=int, default=None, help="run only the first N cases")
+    ap.add_argument("--selfcheck", action="store_true", help="test the scorer on a synthetic plan (no LLM)")
+    ap.add_argument("--rescore", metavar="RESULTS.json", help="re-score a saved run's facts with the current assertions (no LLM)")
+    args = ap.parse_args()
+
+    if args.selfcheck:
+        _selfcheck()
+        return
+    if args.rescore:
+        _rescore(Path(args.rescore))
+        return
+
+    data = json.loads(DATASET.read_text(encoding="utf-8"))
+    cases = data["cases"][: args.limit] if args.limit else data["cases"]
+    catalog, tools, n2id = build_pa_catalog(), default_registry(), name_to_id()
+    planner = PlannerService()
+
+    results: list[CaseResult] = []
+    for case in cases:
+        r = await _run_case(planner, catalog, tools, n2id, case)
+        mark = "PASS" if r.passed else ("ERR " if r.error else "FAIL")
+        print(f"[{mark}] {r.id:<28} {r.lang}  {r.latency_ms:>5}ms  {r.tokens:>5}tok")
+        if not r.passed:
+            if r.error:
+                print(f"        error: {r.error}")
+            for c in r.checks:
+                if not c["ok"]:
+                    print(f"        ✗ {c['check']}  ({c['detail']})")
+        results.append(r)
+
+    _summarize(results)
+    _write(results, data)
+
+
+def _summarize(results: list[CaseResult]) -> None:
+    total = len(results)
+    passed = sum(r.passed for r in results)
+    print("\n" + "=" * 60)
+    print(f"OVERALL: {passed}/{total} passed ({passed / total:.0%})" if total else "no cases")
+
+    def rate(subset: list[CaseResult]) -> str:
+        return f"{sum(r.passed for r in subset)}/{len(subset)}" if subset else "0/0"
+
+    cats = sorted({r.category for r in results})
+    print("\nby category:")
+    for c in cats:
+        print(f"  {c:<22} {rate([r for r in results if r.category == c])}")
+    print("\nregression guard:")
+    print(f"  carried_over (must stay green) {rate([r for r in results if r.carried_over])}")
+    print(f"  new (target bugs)              {rate([r for r in results if not r.carried_over])}")
+    lat = [r.latency_ms for r in results if r.latency_ms]
+    if lat:
+        print(f"\nlatency ms: median={statistics.median(lat):.0f} max={max(lat)}")
+    print(f"tokens total: {sum(r.tokens for r in results)}")
+
+
+def _write(results: list[CaseResult], dataset: dict) -> None:
+    RESULTS_DIR.mkdir(exist_ok=True)
+    ts = datetime.now().strftime("%Y-%m-%d_%H%M%S")
+    out = RESULTS_DIR / f"planner_result_{ts}.json"
+    payload = {
+        "timestamp": ts,
+        "total": len(results),
+        "passed": sum(r.passed for r in results),
+        "cases": [asdict(r) for r in results],
+    }
+    out.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
+    print(f"\nwrote {out}")
+
+
+# --------------------------------------------------------------------------- #
+# Selfcheck — verifies the scorer without an LLM call
+# --------------------------------------------------------------------------- #
+
+def _rescore(path: Path) -> None:
+    """Re-score a saved run's persisted `facts` with the CURRENT assertions —
+    iterate on assertions/dataset without spending another LLM run."""
+    saved = json.loads(path.read_text(encoding="utf-8"))
+    expect_by_id = {c["id"]: c["expect"] for c in json.loads(DATASET.read_text(encoding="utf-8"))["cases"]}
+    n2id = name_to_id()
+    results: list[CaseResult] = []
+    for c in saved["cases"]:
+        facts, exp = c.get("facts") or {}, expect_by_id.get(c["id"], {})
+        if c.get("error") or not facts:
+            checks, passed = [], False
+        else:
+            t = evaluate_facts(facts, exp, n2id)
+            checks = [{"check": ck, "ok": ok, "detail": d} for ck, ok, d in t]
+            passed = all(x["ok"] for x in checks)
+        r = CaseResult(
+            id=c["id"], category=c["category"], lang=c["lang"], carried_over=c["carried_over"],
+            question=c["question"], passed=passed, checks=checks, facts=facts, error=c.get("error"),
+        )
+        mark = "PASS" if r.passed else ("ERR " if r.error else "FAIL")
+        print(f"[{mark}] {r.id:<28} {r.lang}")
+        if not r.passed:
+            if r.error:
+                print(f"        error: {r.error}")
+            for x in r.checks:
+                if not x["ok"]:
+                    print(f"        ✗ {x['check']}  ({x['detail']})")
+        results.append(r)
+    _summarize(results)
+    print(f"\n(re-scored {path.name} — no LLM calls)")
+
+
+def _selfcheck() -> None:
+    from types import SimpleNamespace as NS
+    n2id = name_to_id()
+
+    def plan(tasks_tcs: list[list[tuple[str, dict]]], infeasible: str | None = None):
+        tasks = [NS(tool_calls=[NS(tool=t, args=a) for t, a in tcs]) for tcs in tasks_tcs]
+        return NS(tasks=tasks, infeasible_reason=infeasible)
+
+    def ev(tl: Any, expect: dict) -> bool:
+        return all(ok for _, ok, _ in evaluate_facts(extract_facts(tl), expect, n2id))
+
+    exp_rank = {"group_by": True, "group_by_col": "Equipment_Number", "select_agg": "avg", "order_dir": "asc", "limit": 5}
+
+    # ranking via IR group_by — passes
+    good_ir = plan([[("retrieve_data", {"ir": {
+        "select": [{"kind": "agg", "fn": "avg", "column_id": "c_pa_percent"}],
+        "group_by": ["c_equipment_number"],
+        "order_by": [{"column_id": "avg_pa", "dir": "asc"}], "limit": 5}})]])
+    assert ev(good_ir, exp_rank), "IR-group ranking should pass"
+
+    # grouping via analyze_aggregate tool (aliases + 'mean') must ALSO count
+    agg_tool = plan([
+        [("retrieve_data", {"ir": {"select": [
+            {"kind": "column", "column_id": "c_section", "alias": "section"},
+            {"kind": "column", "column_id": "c_pa_percent", "alias": "pa"}]}})],
+        [("analyze_aggregate", {"group_by": ["section"], "aggregations": {"pa": ["mean"]}})],
+    ])
+    assert ev(agg_tool, {"group_by": True, "group_by_col": "Section", "select_agg": "avg"}), \
+        "analyze_aggregate grouping (mean==avg) should pass"
+
+    # buggy: raw rows, no grouping anywhere — must FAIL
+    bad = plan([[("retrieve_data", {"ir": {"select": [{"kind": "column", "column_id": "c_equipment_number"}],
+                 "order_by": [{"column_id": "c_pa_percent", "dir": "asc"}], "limit": 5}})]])
+    assert not ev(bad, exp_rank), "raw-row ranking should fail"
+
+    # fuzzy: enumerated 'in' fails; non-enumerated (like/=) passes
+    in_ir = plan([[("retrieve_data", {"ir": {"select": [{"kind": "agg", "fn": "count"}],
+                   "filters": [{"column_id": "c_model_unit", "op": "in", "value": ["777E", "777D"]}]}})]])
+    assert not ev(in_ir, {"no_filter_op": "in"}), "enumerated 'in' should fail"
+    ok_ir = plan([[("retrieve_data", {"ir": {"select": [{"kind": "agg", "fn": "count"}],
+                   "filters": [{"column_id": "c_model_unit", "op": "like", "value": "777%"}]}})]])
+    assert ev(ok_ir, {"no_filter_op": "in"}), "non-enumerated filter should pass"
+
+    assert ev(NS(tasks=[], infeasible_reason="no churn data"), {"infeasible": True})
+    print("selfcheck OK — scorer distinguishes good vs buggy plans (IR + analyze_aggregate paths)")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())