Agentic-Service-Data-Eyond-Catalog

Running

Agentic-Service-Data-Eyond-Catalog / src /agents /planner /validator.py
[NOTICKET] fix(validator): reject an aggregate alias that names another column (#16)
db195e3 about 20 hours ago
59.2 kB
"""PlannerValidator — checks a TaskList before it reaches the TaskRunner.
Runs the 8 checks from AGENT_ARCHITECTURE_CONTEXT_new.md §7.3. On failure it
raises `PlannerValidationError` with a message specific enough that the planner
can be re-prompted to self-correct (the retry loop lives in service.py).
Check #1 (Pydantic parse) is enforced at the structured-output boundary — by the
time a `TaskList` reaches here it has already parsed; this validator additionally
rejects structurally-invalid plans (duplicate ids, dangling edges, cycles).
"""

from __future__ import annotations

import re
from collections.abc import Iterator

from pydantic import ValidationError

from src.middlewares.logging import get_logger

from ...catalog.models import Catalog
from ...query.ir.models import QueryIR
from ...query.ir.repair import IRRepairer
from ...query.ir.validator import IRValidationError, IRValidator
from ...tools.analytics.agg_vocab import normalise_agg
from ...tools.analytics.aggregation import SUPPORTED_AGGS
from .contracts import ToolRegistry
from .errors import PlannerValidationError
from .inputs import Constraints
from .schemas import PLACEHOLDER_RE, TaskList

logger = get_logger("ir_repair")

# Heuristic: a checkable success_criteria mentions a measurable signal.
#
# BOTH LANGUAGES, because the planner writes this field in the language of the
# question and the product is majority-Indonesian (A23 Bug 7). English-only tokens
# made the check unpassable on an ID turn except by luck: every Indonesian turn in
# the 2026-09-17 dry-run burned a retry here, and one degenerated further — told
# "not checkable" twice, the model dropped `success_criteria` and `expected_output`
# altogether and the turn died.
#
# This is parity, not loosening. Each Indonesian token mirrors an English one
# already trusted ("menghasilkan" = "produced", "jumlah"/"baris" = "count",
# "minimal"/"lebih"/"kurang" = "above"/"below"), so a criterion that would fail in
# English still fails in Indonesian.
_CHECKABLE_TOKENS = (
    # English. "rows"/"returned"/"records" restore parity with the Indonesian
    # "baris"/"menghasilkan" below: without them an English criterion phrased
    # "Returned rows containing component and breakdown_duration_total" failed
    # while its Indonesian twin passed, costing a retry on English turns only.
    "rate", "count", "match", "produced", "above", "below", "equal",
    "rows", "returned", "records",
    # Indonesian
    "tingkat", "jumlah", "baris", "cocok", "menghasilkan",
    "minimal", "lebih", "kurang", "sama", "persen",
)

# 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


class PlannerValidator:
    def __init__(
        self,
        ir_validator: IRValidator | None = None,
        ir_repairer: IRRepairer | None = None,
    ) -> None:
        self._ir_validator = ir_validator or IRValidator()
        self._ir_repairer = ir_repairer or IRRepairer()

    def validate(
        self,
        task_list: TaskList,
        registry: ToolRegistry,
        catalog: Catalog,
        constraints: Constraints,
    ) -> 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")
        if len(tasks) > constraints.max_tasks:
            raise PlannerValidationError(
                f"plan has {len(tasks)} tasks, exceeds max_tasks={constraints.max_tasks}"
            )

        ids = [t.id for t in tasks]
        if len(set(ids)) != len(ids):
            dupes = sorted({i for i in ids if ids.count(i) > 1})
            raise PlannerValidationError(f"duplicate task id(s): {dupes}")
        id_set = set(ids)
        tasks_by_id = {t.id: t for t in tasks}

        known_tools = registry.names()
        known_sources = {s.source_id for s in catalog.sources}

        for task in tasks:
            for call in task.tool_calls:
                # Check 2 — every tool exists in the registry.
                if call.tool not in known_tools:
                    raise PlannerValidationError(
                        f"task {task.id}: tool {call.tool!r} not in registry "
                        f"(known: {sorted(known_tools)})"
                    )
                spec = registry.get(call.tool)
                assert spec is not None  # guaranteed by the membership check above

                # Check 8a — args carry the required keys and no unknown keys.
                required = set(spec.input_schema.get("required", []))
                allowed = set(spec.input_schema.get("properties", {}).keys()) | required
                missing = required - set(call.args.keys())
                if missing:
                    raise PlannerValidationError(
                        f"task {task.id}: tool {call.tool!r} missing required arg(s): "
                        f"{sorted(missing)}"
                    )
                unknown = set(call.args.keys()) - allowed
                if unknown:
                    raise PlannerValidationError(
                        f"task {task.id}: tool {call.tool!r} has unknown arg(s): "
                        f"{sorted(unknown)} (allowed: {sorted(allowed)})"
                    )

                # Check 8e — analyze_importance: the target must actually VARY.
                #
                # Nothing can explain a number that never moves, and the tool says so
                # (ConstantTargetError) — but at execution time, which is too late for a
                # corrective retry, so the task reaches the Assembler as a dead end. The
                # 2026-09-17 dry-run lost its "what drives down hours" turn exactly that
                # way: the planner targeted `Total_Hours` (a fixed 24-hour day, and the
                # catalog knew `distinct_count=1`) instead of `Total_Down_Hour`.
                #
                # Catalog stats describe the FULL column, while the tool sees a filtered
                # subset — which is the safe direction and the reason this check has no
                # false positives: a column constant over every row is constant over any
                # subset of them. The converse is not true, so a subset that happens to
                # be constant still fails at the tool, as it must.
                #
                # Matched case-INSENSITIVELY on purpose: the planner writes column names
                # lowercased (the dry-run passed `total_hours` for `Total_Hours`), so an
                # exact match would silently never fire.
                # Check 8h — a bare month number needs its year retrieved with it.
                #
                # analyze_trend/analyze_forecast bucket on `date_column`. When that
                # column holds bare month numbers 1-12, _parse_date_column resolves
                # them against a sibling column literally named `year`, and raises
                # InvalidDateColumnError when none was retrieved -- correct, but at
                # EXECUTION, after the data has been fetched and too late for a
                # corrective retry. The 2026-09-17 pa_dummy run lost its trend turn
                # that way: `pa.year` exists and the plan simply did not select it.
                # The forecast turn on the same data DID select it, so this is the
                # planner being inconsistent, not incapable.
                if call.tool in ("analyze_trend", "analyze_forecast"):
                    _require_year_with_bare_month(
                        task.id, call, tasks_by_id, catalog
                    )
                    # Check 8l — the date column must be one the tool can read.
                    _reject_unreadable_period(task.id, call, tasks_by_id, catalog)

                if call.tool == "analyze_segmentation":
                    # Check 8n — band entities, not their period rows (2026-11-03).
                    _reject_raw_periodic_banding(task.id, call, tasks_by_id, catalog)

                if call.tool in _RAW_ROW_TOOLS:
                    # Check 8i — importance / an A/B test must not read a capped slice
                    # (2026-10-22; ab_test + period-only filters 2026-10-28).
                    _reject_capped_importance(
                        task.id, call, tasks_by_id, catalog, constraints
                    )
                if call.tool == "analyze_importance":
                    target = call.args.get("target_column")
                    if isinstance(target, str) and target.strip():
                        constant = _constant_columns(catalog)
                        if target.strip().lower() in constant:
                            raise PlannerValidationError(
                                f"task {task.id}: analyze_importance targets "
                                f"{target!r}, which holds one value in every row — "
                                "nothing can explain a column that never varies. "
                                "Target the measure the user actually asked about "
                                "(re-read the catalog for a column that moves), or "
                                "drop this task."
                            )

                # Check 8c — analyze_aggregate: every aggregation FUNCTION must be one
                # the tool supports. Check 8a only validates arg *names* (`aggregations`
                # is allowed); it never looks at the function *values* inside the dict,
                # so an unsupported func like `std` otherwise passes validation and only
                # fails at execution — too late for a corrective retry, so the task
                # reaches the Assembler as a silent failure. Catch it here so the planner
                # is re-prompted to degrade to a supported function (e.g. `mean`).
                if call.tool == "analyze_aggregate":
                    aggs = call.args.get("aggregations")
                    if isinstance(aggs, dict):
                        bad = sorted(
                            {
                                f
                                for funcs in aggs.values()
                                for f in ([funcs] if isinstance(funcs, str) else funcs or [])
                                # `avg`/`count_distinct` are the QueryIR's names for
                                # `mean`/`nunique`, and planner.md teaches them for the
                                # IR — so one plan legitimately carries both spellings.
                                # Rejecting the synonym here would fail a task the tool
                                # now runs fine. This narrows nothing: `std`, `var` and
                                # every other unsupported function are still rejected.
                                # (A23 Bug 2 -- see tools/analytics/agg_vocab.)
                                if normalise_agg(f) not in SUPPORTED_AGGS
                            }
                        )
                        if bad:
                            raise PlannerValidationError(
                                f"task {task.id}: analyze_aggregate has unsupported "
                                f"aggregation function(s) {bad} (supported: "
                                f"{sorted(SUPPORTED_AGGS)}). Use a supported function "
                                "(e.g. mean/median); for the spread of a whole column "
                                "use analyze_descriptive instead."
                            )

                    # Check 8d — group_by keys must be plain column-name strings. A
                    # derived grouping (a CASE/binning expression emitted as a dict)
                    # is unhashable and crashes df.groupby at execution ("unhashable
                    # type: 'dict'"). Reject it here so the planner is re-prompted;
                    # bucketing a numeric column into ranges belongs to
                    # analyze_segment, not smuggled through analyze_aggregate.
                    group_by = call.args.get("group_by")
                    if isinstance(group_by, list):
                        bad_keys = [g for g in group_by if not isinstance(g, str)]
                        if bad_keys:
                            raise PlannerValidationError(
                                f"task {task.id}: analyze_aggregate group_by must be "
                                f"column names (strings); got non-string entr(ies) "
                                f"{bad_keys}. Derived groupings (CASE/binning) are not "
                                "supported — group by an existing column."
                            )

                # Check 3 — concrete source_id args must exist in the catalog.
                src = call.args.get("source_id")
                if isinstance(src, str) and not _is_placeholder(src):
                    if src not in known_sources:
                        raise PlannerValidationError(
                            f"task {task.id}: tool {call.tool!r} references unknown "
                            f"source_id {src!r} (known: {sorted(known_sources)})"
                        )

                # Check 8b — inline retrieve_data IR validates against the catalog.
                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(
                    f"task {task.id}: success_criteria is not checkable — include a "
                    f"measurable signal (one of {list(_CHECKABLE_TOKENS)}); "
                    f"got {task.success_criteria!r}"
                )

        # Check 4 — DAG: edges resolve, placeholders resolve, no cycles.
        self._validate_dag(tasks_by_id, id_set)

        # Check 11 — a value-handoff `${t<id>.<column>}` must name a column its
        # upstream step actually outputs (D-5, 2026-09-23). After Check 4, so every
        # ref already resolves to a task.
        _reject_unknown_handoff_columns(tasks_by_id, catalog, registry)
        # Check 8k — a ranking the user reads is not cut to one row (2026-10-22).
        _reject_top1_rankings(tasks_by_id)

    def _validate_inline_ir(self, task_id: str, args: dict, catalog: Catalog) -> None:
        raw_ir = args.get("ir")
        if not isinstance(raw_ir, dict):
            raise PlannerValidationError(
                f"task {task_id}: retrieve_data.args.ir must be an inline QueryIR "
                f"object, got {type(raw_ir).__name__}"
            )
        try:
            ir = QueryIR.model_validate(raw_ir)
        except ValidationError as e:
            raise PlannerValidationError(
                f"task {task_id}: retrieve_data.args.ir is not a valid QueryIR: {e}"
            ) from e

        # Canonicalize near-miss ids (LLM dropped/mutated a char in an opaque
        # catalog id) before validating. On a successful repair, write the fixed
        # IR back into the tool call so the downstream executor runs the
        # corrected IR — not just the validator.
        ir, repairs = self._ir_repairer.repair(ir, catalog)
        if repairs:
            args["ir"] = ir.model_dump()
            for r in repairs:
                logger.info(
                    "repaired ir id",
                    task_id=task_id,
                    where=r.where,
                    from_id=r.from_id,
                    to_id=r.to_id,
                )

        try:
            self._ir_validator.validate(ir, catalog)
        except IRValidationError as e:
            raise PlannerValidationError(
                f"task {task_id}: retrieve_data IR failed catalog validation: {e}"
            ) from e

        _reject_misleading_aliases(task_id, ir, catalog)
        _reject_unmatchable_filter_values(task_id, ir, catalog)
        _reject_typed_lookup_ids(task_id, ir, catalog)
        _reject_raw_periodic_ranking(task_id, ir, catalog)

    @staticmethod
    def _validate_data_source(
        task_id: str, call, tasks_by_id: dict, registry: ToolRegistry
    ) -> None:
        """The `data`/`data_right` handoff (Pattern A) must be a '${t<id>}'
        placeholder pointing at an upstream data-producing task.
        Two failure modes are rejected here:
        1. The arg is present but is NOT a placeholder — e.g. the planner inlined a
           {table_id, source_id} table reference (or any literal) instead of
           chaining a retrieve_data output. analyze_* tools never self-fetch by
           source_id; they only consume an already-retrieved table, so such a plan
           can only blow up at execution ("unsupported 'data' type: dict"). Reject
           it here so the retry loop re-prompts the planner to add a retrieve_data
           task and pass its placeholder.
        2. The placeholder resolves to a metadata (check_data/check_knowledge) or
           documents (retrieve_knowledge) task. 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_right` is analyze_merge's second table input (KM-703) — same Pattern A
        handoff, so it gets the same guard. An arg that is absent is left to Check 8a
        (required-arg presence).
        """
        for arg_name in ("data", "data_right"):
            if arg_name not in call.args:
                continue
            data_arg = call.args[arg_name]
            match = (
                PLACEHOLDER_RE.fullmatch(data_arg.strip())
                if isinstance(data_arg, str)
                else None
            )
            if match is None:
                raise PlannerValidationError(
                    f"task {task_id}: tool {call.tool!r} arg {arg_name!r} must be a "
                    f"'${{t<id>}}' placeholder referencing a retrieve_data output "
                    f"(Pattern A), got {data_arg!r}. analyze_* tools do not fetch data "
                    "themselves — add a retrieve_data task and pass its output."
                )
            ref_task = tasks_by_id.get(match.group(1))
            if ref_task is None or not ref_task.tool_calls:
                continue  # 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 {arg_name!r} from "
                    f"task {match.group(1)} ({ref_tool!r}, category "
                    f"{ref_spec.category!r}), which produces metadata/documents — not "
                    "analyzable data rows. Feed analyze_* from a data-producing tool "
                    "(e.g. retrieve_data)."
                )
            # Check 10 — render_chart path shape (SPINE_V2_PLAN §4.3): the chart's
            # `data` must come from a task whose tool yields a TABLE. A stats- or
            # series-kind upstream passes the category denylist above but cannot
            # be materialized into a DataFrame at execution, so the chart could
            # only fail late; reject pre-run so the retry re-prompts toward a
            # grouped retrieve_data or a table-producing analyze_*.
            if (
                call.tool == "render_chart"
                and ref_spec is not None
                and ref_spec.output_kind != "table"
            ):
                raise PlannerValidationError(
                    f"task {task_id}: render_chart takes its {arg_name!r} from task "
                    f"{match.group(1)} ({ref_tool!r}), which yields "
                    f"{ref_spec.output_kind!r} output — a chart is drawn from a TABLE. "
                    "Feed it a table-producing task (a grouped retrieve_data, "
                    "analyze_aggregate, analyze_merge, ...)."
                )

    @staticmethod
    def _validate_dag(tasks_by_id: dict, id_set: set[str]) -> None:
        for task in tasks_by_id.values():
            for dep in task.depends_on:
                if dep not in id_set:
                    raise PlannerValidationError(
                        f"task {task.id}: depends_on references unknown task {dep!r}"
                    )
                if dep == task.id:
                    raise PlannerValidationError(
                        f"task {task.id}: depends_on includes itself"
                    )

        cycle = _find_cycle(tasks_by_id)
        if cycle:
            raise PlannerValidationError(f"cycle detected in depends_on: {' -> '.join(cycle)}")

        # A `${...}` that is not a task reference (`${sdj_site_id}`) is never
        # resolved by the TaskRunner: it reached the database as a literal string and
        # the step died there (SDJ FE test 2026-10-27, "invalid input syntax for type
        # bigint"). Reject it here, where the planner can still correct it.
        for task in tasks_by_id.values():
            for call in task.tool_calls:
                for value in _iter_arg_strings(call.args):
                    for inner in _ANY_PLACEHOLDER_RE.findall(value):
                        if not PLACEHOLDER_RE.fullmatch("${" + inner + "}"):
                            raise PlannerValidationError(
                                f"task {task.id}: '${{{inner}}}' is not a task reference "
                                "— a placeholder is '${t<id>}' or '${t<id>.<alias>}' of "
                                "an earlier task; to use a looked-up value, select it in "
                                "that task and reference it by its alias"
                            )

        # A placeholder to an existing task that is not upstream is the planner
        # writing the handoff and forgetting its depends_on edge (EMA replay
        # 2026-10-27: "${t3.site_id}" with no "t3" dependency — a whole retry spent
        # on bookkeeping). Add the edge in place when it creates no cycle; the
        # target then runs first, which is all the edge means. A cycle reverts it,
        # and the check below rejects as before.
        for task in tasks_by_id.values():
            for ref in sorted(_placeholder_refs(task)):
                if ref not in id_set or ref == task.id:
                    continue
                if ref in _all_ancestors(tasks_by_id)[task.id]:
                    continue
                task.depends_on.append(ref)
                if _find_cycle(tasks_by_id):
                    task.depends_on.remove(ref)
                else:
                    logger.info("added missing depends_on", task_id=task.id, ref=ref)

        # On an acyclic graph, a placeholder is safe iff its target is a
        # transitive ancestor — i.e. guaranteed to have completed before this
        # task runs. Requiring a *direct* depends_on would wrongly reject valid
        # plans that depend on the target through an intermediate task.
        ancestors = _all_ancestors(tasks_by_id)
        for task in tasks_by_id.values():
            for ref in _placeholder_refs(task):
                if ref not in id_set:
                    raise PlannerValidationError(
                        f"task {task.id}: placeholder '${{{ref}}}' references unknown task"
                    )
                if ref not in ancestors[task.id]:
                    raise PlannerValidationError(
                        f"task {task.id}: placeholder '${{{ref}}}' used but {ref!r} is "
                        f"not a (transitive) dependency — add it to depends_on"
                    )


# Any `${...}`, task reference or not — the first is checked against PLACEHOLDER_RE.
_ANY_PLACEHOLDER_RE = re.compile(r"\$\{([^}]*)\}")


def _is_placeholder(value: str) -> bool:
    return bool(PLACEHOLDER_RE.fullmatch(value.strip()))


def _placeholder_refs(task) -> set[str]:
    """Task ids referenced by any placeholder in a task's args — including ones
    NESTED inside a retrieve_data IR (e.g. a '${t2.customer_id}' value-handoff
    filter). A '${t<id>.<col>}' ref contributes its TASK id ('t2'); the column
    suffix is stripped so the DAG check still enforces the dependency (task ids
    contain no '.')."""
    refs: set[str] = set()
    for call in task.tool_calls:
        for value in _iter_arg_strings(call.args):
            for ref in PLACEHOLDER_RE.findall(value):
                refs.add(ref.split(".", 1)[0])
    return refs


def _iter_arg_strings(value: object) -> Iterator[str]:
    """Yield every string reachable in a tool-arg value, recursing into dicts and
    lists so placeholders nested inside an IR are seen — not just top-level string
    args (which is all `data`/`data_right` Pattern-A handoffs used to be)."""
    if isinstance(value, str):
        yield value
    elif isinstance(value, dict):
        for v in value.values():
            yield from _iter_arg_strings(v)
    elif isinstance(value, list):
        for v in value:
            yield from _iter_arg_strings(v)


def _is_checkable(text: str) -> bool:
    low = text.lower()
    return any(tok in low for tok in _CHECKABLE_TOKENS)


def _find_cycle(tasks_by_id: dict) -> list[str] | None:
    color = {tid: _WHITE for tid in tasks_by_id}
    stack: list[str] = []

    def dfs(node: str) -> list[str] | None:
        color[node] = _GREY
        stack.append(node)
        for dep in tasks_by_id[node].depends_on:
            if color.get(dep) == _GREY:
                idx = stack.index(dep)
                return stack[idx:] + [dep]
            if color.get(dep) == _WHITE:
                found = dfs(dep)
                if found:
                    return found
        stack.pop()
        color[node] = _BLACK
        return None

    for tid in tasks_by_id:
        if color[tid] == _WHITE:
            found = dfs(tid)
            if found:
                return found
    return None


def _all_ancestors(tasks_by_id: dict) -> dict[str, set[str]]:
    """ancestors[id] = all tasks reachable by following depends_on edges."""
    cache: dict[str, set[str]] = {}

    def visit(node: str, seen: set[str]) -> set[str]:
        if node in cache:
            return cache[node]
        acc: set[str] = set()
        for dep in tasks_by_id[node].depends_on:
            if dep in seen or dep not in tasks_by_id:
                continue
            acc.add(dep)
            acc |= visit(dep, seen | {dep})
        cache[node] = acc
        return acc

    return {tid: visit(tid, {tid}) for tid in tasks_by_id}


def _constant_columns(catalog: Catalog) -> set[str]:
    """Lower-cased names of every column the catalog reports as single-valued.
    Names, not ids: tool args reference columns by the name they carry in the
    retrieved frame, never by `column_id`. A name colliding across two tables is
    harmless here — both would have to be constant for the check to fire.
    """
    return {
        col.name.lower()
        for source in catalog.sources
        for table in source.tables
        for col in table.columns
        if col.stats is not None and col.stats.distinct_count == 1
    }


# --------------------------------------------------------------------------- #
# Checks 8f/8g — the IR can lie about WHAT it selected (A23 Bugs 10 and 11).
#
# Both were found on the 2026-09-17 pa_dummy run, and both produced a confident
# answer rather than an error, which is why neither existed before: nothing in the
# pipeline was looking.
# --------------------------------------------------------------------------- #


def _source_of(ir: QueryIR, catalog: Catalog):
    return next((s for s in catalog.sources if s.source_id == ir.source_id), None)


def _columns_of(source) -> dict:
    """column_id -> Column, across every table of the source."""
    return {c.column_id: c for t in source.tables for c in t.columns}


# --------------------------------------------------------------------------- #
# Checks 8m/8n — rows of a per-period table are not entities (audit P0-2, 2026-11-03).
#
# "equipment mana yang PA-nya paling rendah tahun 2026?" was planned as
# `SELECT eqm_no, pa … ORDER BY pa LIMIT 10` on the monthly view: the top row was one
# April month of 8 days (EXHC36007 6.36%), presented as the unit's 2026 PA. planner.md
# says "aggregate per entity first" (Rankings) and the 10-28 cut deleted the sentence
# against ranking raw rows; prose alone did not hold. An ungrouped query on a table with
# period columns is about PERIODS — fine when it shows the period, wrong otherwise.
# --------------------------------------------------------------------------- #

# Grain columns only: a table with just a `date` is usually an event log (one row per
# breakdown), where "10 terlama" ranks events legitimately (review 2026-11-03).
_PERIOD_NAMES = frozenset({"year", "month", "week", "quarter"})


def _period_column_ids(ir: QueryIR, catalog: Catalog) -> set[str]:
    """Ids of the period columns (by name) of the tables the IR reads."""
    source = _source_of(ir, catalog)
    if source is None:
        return set()
    in_play = {ir.table_id, *(j.target_table_id for j in ir.joins)}
    return {
        c.column_id for t in source.tables if t.table_id in in_play
        for c in t.columns if c.name.lower() in _PERIOD_NAMES
    }


def _raw_period_rows(ir: QueryIR, catalog: Catalog) -> bool:
    """True when the IR returns ungrouped rows of a per-period table without a period column."""
    if ir.group_by or any(getattr(s, "kind", "") == "agg" for s in ir.select):
        return False
    periods = _period_column_ids(ir, catalog)
    if not periods:
        return False
    return not any(getattr(s, "column_id", None) in periods for s in ir.select)


def _reject_raw_periodic_ranking(task_id: str, ir: QueryIR, catalog: Catalog) -> None:
    """Check 8m — an ordered, limited query on raw rows of a per-period table."""
    if not ir.order_by or ir.limit is None or not _raw_period_rows(ir, catalog):
        return
    raise PlannerValidationError(
        f"task {task_id}: ranks raw rows of a table with one row per period, so one "
        "entity can appear many times and a single period's value is read as its "
        "total. Group by the entity and aggregate (avg for rates and scores, sum for "
        "totals), with `count` as n_periods and `max` of the period as last_period. "
        "If the user means the top PERIODS, select the period column."
    )


def _reject_raw_periodic_banding(
    task_id: str, call, tasks_by_id: dict, catalog: Catalog
) -> None:
    """Check 8n — analyze_segmentation fed raw rows of a per-period table."""
    data = call.args.get("data")
    if not isinstance(data, str) or not (data.startswith("${") and data.endswith("}")):
        return
    upstream = tasks_by_id.get(data[2:-1])
    if upstream is None:
        return
    for up_call in upstream.tool_calls:
        raw = up_call.args.get("ir")
        if up_call.tool != "retrieve_data" or not isinstance(raw, dict):
            continue
        try:
            ir = QueryIR.model_validate(raw)
        except ValidationError:
            return  # Check 8b reports it
        if _raw_period_rows(ir, catalog):
            raise PlannerValidationError(
                f"task {task_id}: analyze_segmentation bands raw rows of a table with "
                "one row per period, so each entity is counted once per period. Group "
                "the upstream retrieve_data by the entity and average the measure first, "
                "then band; select the period column only if the user asks about periods."
            )


def _reject_misleading_aliases(task_id: str, ir: QueryIR, catalog: Catalog) -> None:
    """An alias may not rename a column into some OTHER thing that exists here.
    Asked whether *backlog* relates to bad actor score — two tables the schema
    declares no FK between, so unjoinable — the planner emitted
        SELECT badactor.bad_actor_score, badactor.ma AS "backlog" FROM badactor
    and the answer read "Korelasi Pearson antara backlog dan bad actor score adalah
    -0,029". `ma` is Mechanical Availability. It is not backlog, and `backlog` is a
    separate TABLE it could not reach. Rather than refuse, the plan renamed a
    different column to match the user's word, and the alias made the wrong answer
    read as the right one — every layer downstream, including the narrative, saw
    only "backlog".
    This is the force-mapping the feasibility gate (recipes R6/R15) exists to stop,
    taking a route the gate never watched: not a wrong COLUMN CHOICE, a wrong NAME
    on a defensible column.
    The rule is narrow on purpose. An alias is rejected only when it collides with
    something real in this source — another table's name, or another column's name —
    because that is when a reader is actively misled about which thing they are
    looking at. Aliases that merely tidy (`total_mo` -> `total_maintenance_orders`)
    collide with nothing and pass untouched.
    """
    source = _source_of(ir, catalog)
    if source is None:  # unknown source: the IR validator already rejects it
        return
    table_names = {t.name.lower() for t in source.tables}
    cols = _columns_of(source)
    col_names = {c.name.lower() for c in cols.values()}

    for item in ir.select:
        alias = getattr(item, "alias", None)
        column_id = getattr(item, "column_id", None)
        if not alias or not isinstance(alias, str):
            continue
        low = alias.strip().lower()
        own = cols.get(column_id)
        if own is not None and low == own.name.lower():
            continue  # aliasing a column to its own name is a no-op
        if low in table_names:
            raise PlannerValidationError(
                f"task {task_id}: column {own.name if own else column_id!r} is aliased "
                f"as {alias!r}, which is the name of a TABLE in this source. That "
                f"renames one thing into another and the answer would describe "
                f"{alias!r} while reporting "
                f"{own.name if own else 'a different column'}. If the question is "
                f"about {alias!r}, query that table; if it cannot be reached from "
                f"here, return an infeasible plan saying so."
            )
        if low in col_names:
            raise PlannerValidationError(
                f"task {task_id}: column {own.name if own else column_id!r} is aliased "
                f"as {alias!r}, which is the name of a DIFFERENT column in this "
                f"source. Alias to a name that is not already taken, or select the "
                f"column that is actually called {alias!r}."
            )
        # Test demo #16: `AVG(mtbf) AS avg_mtbs` passed the exact-name check above and
        # the answer reported MTBF as MTBS. Strip one aggregate affix and look again.
        stem = _alias_stem(low)
        if (
            own is not None
            and stem is not None
            and stem != own.name.lower()
            and stem in col_names
        ):
            raise PlannerValidationError(
                f"task {task_id}: column {own.name!r} is aliased as {alias!r}, which "
                f"names the DIFFERENT column {stem!r} in this source. Select "
                f"{stem!r} if that is the measure asked for; otherwise alias "
                f"{own.name!r} by its own name."
            )


# Aggregate words a planner wraps around a column in an alias (`avg_mtbs`, `pa_mean`).
_ALIAS_AGG_PREFIXES = (
    "avg_", "average_", "mean_", "sum_", "total_", "min_", "max_", "median_", "count_",
    "rata_rata_",
)
_ALIAS_AGG_SUFFIXES = ("_avg", "_mean", "_sum", "_total", "_min", "_max", "_median", "_count")


def _alias_stem(alias: str) -> str | None:
    """The alias with one aggregate prefix or suffix removed, or None if it has none."""
    for prefix in _ALIAS_AGG_PREFIXES:
        if alias.startswith(prefix) and len(alias) > len(prefix):
            return alias[len(prefix):]
    for suffix in _ALIAS_AGG_SUFFIXES:
        if alias.endswith(suffix) and len(alias) > len(suffix):
            return alias[: -len(suffix)]
    return None


def _reject_unmatchable_filter_values(task_id: str, ir: QueryIR, catalog: Catalog) -> None:
    """Reject a filter value that CANNOT match, when the catalog can prove it.
    "Compare PA between the two sites" produced
        WHERE site_id IN (...)   ->   0 rows
    even though the catalog carries `site_id` samples ['SITE_A', 'SITE_B'] and says
    distinct_count = 2. The planner had both real values in front of it and filtered
    on something else; the turn then reported that the comparison "cannot be
    completed", which reads like a data limitation rather than the planning slip it
    was.
    Only fires when the catalog's samples are PROVABLY the complete value set --
    `distinct_count` equal to the number of samples -- so "not in the samples" really
    does mean "not in the column". For any column where the samples are a partial
    peek (the normal case) this stays silent, because absence from a sample is not
    evidence of absence from the data.
    PII columns are skipped: their samples are nulled by design, so an empty sample
    list there says nothing at all.
    """
    source = _source_of(ir, catalog)
    if source is None:
        return
    cols = _columns_of(source)

    for f in ir.filters:
        if f.op not in ("=", "in"):
            continue
        col = cols.get(f.column_id)
        if col is None or col.pii_flag or col.stats is None:
            continue
        samples = col.sample_values
        distinct = col.stats.distinct_count
        if not samples or distinct is None or distinct != len(samples):
            continue  # samples are not provably exhaustive -> say nothing
        known = {str(v).strip().lower() for v in samples}
        wanted = f.value if isinstance(f.value, list | tuple | set) else [f.value]
        missing = [v for v in wanted if str(v).strip().lower() not in known]
        if missing and len(missing) == len(list(wanted)):
            raise PlannerValidationError(
                f"task {task_id}: filter on {col.name!r} uses value(s) {missing!r}, "
                f"none of which exist in that column — its complete set of values is "
                f"{samples!r}. This filter matches nothing. Use the real values, or "
                f"drop the filter."
            )


def _is_bare_month(col) -> bool:
    """True when a column holds month NUMBERS rather than dates.
    Matches _parse_date_column's own test: whole numbers inside 1-12. The name is
    a secondary signal only -- a column called `month` holding real dates is fine,
    and one called `periode` holding 1-12 is not.
    """
    if col is None or col.stats is None:
        return False
    lo, hi = col.stats.min, col.stats.max
    if lo is None or hi is None:
        return False
    try:
        return 1 <= float(lo) and float(hi) <= 12 and col.data_type in ("int", "integer")
    except (TypeError, ValueError):
        return False


def _retrieve_output_names(task, catalog: Catalog) -> set[str] | None:
    """Exact column names a task's LAST call outputs, when that call is a
    `retrieve_data` whose every select item has a certain name (an alias, or a
    bare catalog column). None when unknown — an unaliased aggregate's output name
    is the compiler's choice, and a non-retrieve last step is not a table the
    runtime can read a column from by name."""
    if not task.tool_calls or task.tool_calls[-1].tool != "retrieve_data":
        return None
    raw = task.tool_calls[-1].args.get("ir")
    if not isinstance(raw, dict):
        return None
    source = next((s for s in catalog.sources if s.source_id == raw.get("source_id")), None)
    if source is None:
        return None
    by_id = {c.column_id: c for t in source.tables for c in t.columns}
    names: set[str] = set()
    for item in raw.get("select") or []:
        if not isinstance(item, dict):
            return None
        if item.get("alias"):
            names.add(str(item["alias"]))
        elif item.get("fn") is None and item.get("column_id") in by_id:
            names.add(by_id[item["column_id"]].name)
        else:
            return None
    return names


def _reject_unknown_handoff_columns(
    tasks_by_id: dict, catalog: Catalog, registry: ToolRegistry | None = None
) -> None:
    """Check 11 (D-5): `${t<id>.<column>}` must name a column t<id> outputs.
    At runtime an unknown column resolves to an EMPTY list (`task_runner.
    _column_values`), which is indistinguishable from a real empty result: `in`
    then matches nothing and `not_in` matches EVERYTHING — an anti-join with a
    typo'd column returns the whole table as "customers who never ordered".
    Caught here, the planner gets a retry naming the columns that exist. The name
    must match EXACTLY, as the runtime lookup does, so a case-only miss is
    rejected too, with the right spelling. Unresolvable upstreams are skipped —
    this never manufactures an error it cannot prove.
    """
    for task in tasks_by_id.values():
        for call in task.tool_calls:
            for value in _iter_arg_strings(call.args):
                match = PLACEHOLDER_RE.fullmatch(value.strip())
                if not match:
                    continue
                ref_id, _, column = match.group(1).partition(".")
                upstream = tasks_by_id.get(ref_id)
                if not column or upstream is None:
                    continue
                # A handoff from a metadata/documents task (check_data, check_knowledge,
                # retrieve_knowledge) has no data column to read: `${t1.site_id}` off a
                # check_data step resolves to an EMPTY list, so `in` silently matches
                # nothing (EMA eval 2026-10-22: a site lookup that never happened).
                last = upstream.tool_calls[-1].tool if upstream.tool_calls else None
                spec = registry.get(last) if (registry is not None and last) else None
                if spec is not None and spec.category in _NON_DATA_SOURCE_CATEGORIES:
                    raise PlannerValidationError(
                        f"task {task.id}: value-handoff '${{{ref_id}.{column}}}' reads "
                        f"from {ref_id}, a {last!r} step, which returns metadata or "
                        "documents, not data rows. Look the value up with its own "
                        "retrieve_data task (e.g. select the id where the code matches) "
                        "and hand THAT task's column on."
                    )
                names = _retrieve_output_names(upstream, catalog)
                if names is None or column in names:
                    continue
                near = [n for n in names if n.lower() == column.lower()]
                hint = f" (did you mean {near[0]!r}?)" if near else ""
                raise PlannerValidationError(
                    f"task {task.id}: value-handoff '${{{ref_id}.{column}}}' names a "
                    f"column {ref_id} does not output{hint}; {ref_id} outputs "
                    f"{sorted(names)}. An unknown column resolves to an EMPTY list, "
                    "so `in` would match nothing and `not_in` would match everything."
                )


def _upstream_select_names(call, tasks_by_id: dict, catalog: Catalog) -> set[str] | None:
    """Names (alias or catalog name) the retrieve feeding this call produced.
    Returns None when the upstream cannot be resolved -- an unknown shape must not
    manufacture a validation error.
    """
    data = call.args.get("data")
    if not isinstance(data, str) or not (data.startswith("${") and data.endswith("}")):
        return None
    upstream = tasks_by_id.get(data[2:-1])
    if upstream is None:
        return None
    names: set[str] = set()
    for up_call in upstream.tool_calls:
        raw = up_call.args.get("ir")
        if not isinstance(raw, dict):
            continue
        source = next(
            (s for s in catalog.sources if s.source_id == raw.get("source_id")), None
        )
        if source is None:
            continue
        by_id = {c.column_id: c for t in source.tables for c in t.columns}
        for item in raw.get("select") or []:
            if not isinstance(item, dict):
                continue
            alias = item.get("alias")
            col = by_id.get(item.get("column_id"))
            if alias:
                names.add(str(alias).lower())
            elif col is not None:
                names.add(col.name.lower())
    return names or None


def _upstream_column(call, tasks_by_id: dict, catalog: Catalog):
    """The catalog column the `data` retrieve selected as `date_column`, else None.
    Matched on the select item's alias, or its column name when it has none, among
    PLAIN columns only — an aggregate's output does not carry the column's stats.
    """
    date_column = call.args.get("date_column")
    data = call.args.get("data")
    if not isinstance(date_column, str) or not isinstance(data, str):
        return None
    if not (data.startswith("${") and data.endswith("}")):
        return None
    upstream = tasks_by_id.get(data[2:-1])
    if upstream is None:
        return None
    name = date_column.strip().lower()
    col = None
    for up_call in upstream.tool_calls:
        raw = up_call.args.get("ir")
        if not isinstance(raw, dict):
            continue
        source = next(
            (s for s in catalog.sources if s.source_id == raw.get("source_id")), None
        )
        if source is None:
            continue
        by_id = {c.column_id: c for t in source.tables for c in t.columns}
        for item in raw.get("select") or []:
            if not isinstance(item, dict) or item.get("fn") is not None:
                continue
            c = by_id.get(item.get("column_id"))
            if c is not None and str(item.get("alias") or c.name).lower() == name:
                col = c
    return col


def _require_year_with_bare_month(
    task_id: str, call, tasks_by_id: dict, catalog: Catalog
) -> None:
    date_column = call.args.get("date_column")
    if not isinstance(date_column, str):
        return
    names = _upstream_select_names(call, tasks_by_id, catalog)
    if names is None or date_column.strip().lower() not in names:
        return  # can't resolve, or the column isn't from this retrieve
    # The exact column the retrieve selected. A lookup by NAME across the catalog
    # took the last `month` it met — on the EMA source a decimal one in another view
    # — so the check never fired and the trend failed at execution (2026-11-03).
    col = _upstream_column(call, tasks_by_id, catalog)
    if not _is_bare_month(col) or "year" in names:
        return
    raise PlannerValidationError(
        f"task {task_id}: {call.tool} buckets on {date_column!r}, which holds bare "
        "month numbers (1-12), but the upstream retrieve_data did not select a "
        "`year` column. Month numbers alone cannot be ordered across years. Add the "
        "year column to that retrieve's select list."
    )


def _reject_unreadable_period(
    task_id: str, call, tasks_by_id: dict, catalog: Catalog
) -> None:
    """Check 8l — trend/forecast on whole numbers that are no month and no year.
    `_parse_date_column` reads a numeric date column only as month numbers (1-12,
    with a year) or years (1900-2100); anything else — a week number 1-52, a
    `yyyyww` code — raises InvalidDateColumnError at EXECUTION, too late for a retry.
    The demo test of 2026-11-02 lost its monthly MTTR trend that way: the plan read
    the weekly view and bucketed on `week`. Decided from the catalog stats, which
    mirror the tool's own test; a column without stats is left to the tool.
    """
    date_column = call.args.get("date_column")
    col = _upstream_column(call, tasks_by_id, catalog)
    if col is None or col.stats is None or col.data_type not in ("int", "integer"):
        return
    try:
        lo, hi = float(col.stats.min), float(col.stats.max)
    except (TypeError, ValueError):
        return
    if (1 <= lo and hi <= 12) or (1900 <= lo and hi <= 2100):
        return
    raise PlannerValidationError(
        f"task {task_id}: {call.tool} buckets on {date_column!r}, which holds whole "
        f"numbers from {lo:g} to {hi:g} — not a date, a month number (1-12) or a year, "
        "so the tool cannot place it on a calendar. For a trend by month, quarter or "
        "year, retrieve from the month-grain table and bucket on its `month` column "
        "(select `year` with it)."
    )


# Check 8i — analyze_importance must not rank a capped slice (EMA dry run 2026-10-22).
#
# An UNFILTERED raw retrieve of a table larger than `row_budget` is capped by
# construction, and the compiler orders a capped raw pull by its selected columns
# (D-2), so the slice is the first `row_budget` rows in that order — not a sample.
# "bad actor di pengaruhi apa aja?" pulled 10,000 of 21,124 rows sorted by
# `is_bad_actor`: every row was 0, and the tool failed at execution with
# ConstantTargetError, too late for a retry. planner.md says to compare the groups
# in the query instead; the model stated that rule in its assumptions and still
# planned importance, so it is enforced here.
#
# Narrow on purpose, so it has no false positives: it fires only when the feeding
# IR has NO aggregation, the table's catalog row count exceeds the budget, and its
# filters (if any) are on PERIOD columns only. A filter on an entity or category may
# shrink the pull below the budget (its size is unknown here), so it passes; a year or
# month filter on a multi-year table still keeps most of it (2026-10-28: "apa yang
# membedakan unit bad actor dengan unit lain di 2026" planned analyze_ab_test on a
# `year = 2026` pull of a 21,124-row table, half the time, against every prompt
# wording tried — so the rule lives here). Also covers analyze_ab_test, which reads
# the raw rows the same way.
_RAW_ROW_TOOLS = frozenset({"analyze_importance", "analyze_ab_test"})
_PERIOD_COLUMN_RE = re.compile(
    r"^(year|month|week|quarter|day|date|period)(_id)?$|_(date|year|month|week)$",
    re.IGNORECASE,
)


def _period_only_filters(filters: list, catalog: Catalog) -> bool:
    names = {c.column_id: c.name for s in catalog.sources for t in s.tables for c in t.columns}
    for f in filters or []:
        col = f.get("column_id") if isinstance(f, dict) else None
        if not col or not _PERIOD_COLUMN_RE.search(names.get(col, "")):
            return False
    return True


def _reject_capped_importance(
    task_id: str, call, tasks_by_id: dict, catalog: Catalog, constraints: Constraints
) -> None:
    data = call.args.get("data")
    if not isinstance(data, str) or not (data.startswith("${") and data.endswith("}")):
        return
    upstream = tasks_by_id.get(data[2:-1])
    if upstream is None:
        return
    for up_call in upstream.tool_calls:
        raw = up_call.args.get("ir")
        if up_call.tool != "retrieve_data" or not isinstance(raw, dict):
            continue
        if not _period_only_filters(raw.get("filters"), catalog) or raw.get("group_by") or any(
            isinstance(s, dict) and s.get("kind") == "agg" for s in raw.get("select") or []
        ):
            continue
        table = next(
            (
                t
                for s in catalog.sources
                if s.source_id == raw.get("source_id")
                for t in s.tables
                if t.table_id == raw.get("table_id")
            ),
            None,
        )
        if table is None or not table.row_count or table.row_count <= constraints.row_budget:
            continue
        scope = "a period-filtered" if raw.get("filters") else "an unfiltered"
        raise PlannerValidationError(
            f"task {task_id}: {call.tool} reads {scope} raw pull of "
            f"{table.name!r} ({table.row_count:,} rows), which is over the "
            f"{constraints.row_budget:,}-row budget — it would see only a capped, "
            "ordered slice, where the target may never vary. Compare the groups in "
            "the query instead: for a flag or category target, one retrieve_data that "
            "groups by the outcome column and averages each candidate measure (plus a "
            "count). For a numeric target, filter the pull to the slice the question "
            "is about, or correlate two named columns with analyze_correlation."
        )


# Check 8j — a key that points at a lookup table is not filtered by a TYPED id
# (EMA dry run 2026-10-22).
#
# Users name sites, plants and units by CODE ("SDJ", "IPR"); the measure tables hold
# only the numeric key (`site_id`). With no lookup the planner fills the id in from
# memory or from position: the replay filtered "kalo sdj?" as site 2001 (another
# site) and labelled the result SDJ, and the eval saw EST -> 3003 guessed from the
# order of sample values. A wrong id returns another site's data under the right
# name, so a typed literal on such a key is rejected and the retry looks it up — by
# joining the lookup and filtering its code column, or by a value-handoff
# (`in "${t1.site_id}"`, which passes because it is not a literal).
#
# Fires only when the filtered column is a foreign key (declared or inferred) — or
# the target's own key reached through a join — and the target table has a CODE
# column (`code`, `*_code`, `kode*`): users name those rows by that code. A key into
# a table with only descriptive text (orders.status, customers.email — review
# 2026-10-22: "any string column" rejected `equipment_id = 11` and
# `order_id = 1001`), a placeholder value, and `is_null`-style ops all pass.
_CODE_COLUMN_RE = re.compile(r"(^|_)(code|kode)($|_)|^kode", re.IGNORECASE)


def _reject_typed_lookup_ids(task_id: str, ir: QueryIR, catalog: Catalog) -> None:
    source = _source_of(ir, catalog)
    if source is None:
        return
    tables = {t.table_id: t for t in source.tables}
    in_query = [ir.table_id, *[j.target_table_id for j in ir.joins]]
    fk_target = {
        fk.column_id: fk.target_table_id
        for tid in in_query
        if tid in tables
        for fk in tables[tid].foreign_keys
    }
    # The lookup's OWN key is just as much a typed id once the lookup is joined:
    # `vw_dim_site.site_id = 2001` after a join is the same guess (review finding 2).
    for j in ir.joins:
        fk_target.setdefault(j.right_column_id, j.target_table_id)
    for flt in ir.filters:
        target = tables.get(fk_target.get(flt.column_id, ""))
        if target is None or flt.op not in ("=", "!=", "in", "not_in", "between"):
            continue
        values = flt.value if isinstance(flt.value, list) else [flt.value]
        if not values or any(isinstance(v, str) and v.startswith("${") for v in values):
            continue
        codes = [c.name for c in target.columns
                 if c.data_type == "string" and _CODE_COLUMN_RE.search(c.name)]
        if not codes:
            continue
        col = next(
            (c.name for t in source.tables for c in t.columns if c.column_id == flt.column_id),
            flt.column_id,
        )
        raise PlannerValidationError(
            f"task {task_id}: filters {col!r} by a typed id ({flt.value!r}), but "
            f"{col!r} points at {target.name!r}, whose rows carry a code "
            f"({', '.join(codes)}). Never type an id you did not look up: join "
            f"{target.name!r} and filter its code column, or look the id up in a "
            "separate retrieve_data and hand it on with `in \"${t<id>.<column>}\"`."
        )


# Check 8k — a ranking the user reads is not cut to one row (EMA dry run 2026-10-22).
#
# "Which unit had the most bad-actor months?" planned as group + order + `limit: 1`
# returned one unit while 21 tied at 6 of 6 — the answer named a sole leader that
# did not exist. planner.md asks for 10 rows when no number is given; the model
# followed it in one run and not the next, so it is enforced here.
#
# `limit: 1` stays valid where it belongs: a lookup whose single value a later task
# reads by value-handoff (`${t1.<column>}`), and an ungrouped / unordered pull.
def _reject_top1_rankings(tasks_by_id: dict) -> None:
    handed_off = {
        match.group(1).partition(".")[0]
        for task in tasks_by_id.values()
        for call in task.tool_calls
        for value in _iter_arg_strings(call.args)
        if (match := PLACEHOLDER_RE.fullmatch(value.strip())) and "." in match.group(1)
    }
    for task in tasks_by_id.values():
        if task.id in handed_off:
            continue
        for call in task.tool_calls:
            raw = call.args.get("ir") if call.tool == "retrieve_data" else None
            if not isinstance(raw, dict):
                continue
            if raw.get("limit") == 1 and raw.get("group_by") and raw.get("order_by"):
                raise PlannerValidationError(
                    f"task {task.id}: a grouped, ordered ranking is cut to `limit: 1`, "
                    "which hides entities tied at the top. Use the user's N, or 10 when "
                    "they gave none, so the answer can say who shares the top value."
                )