"""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 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 ) 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) 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) @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}' 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}}' 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)}") # 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" ) 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.}' 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} 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}." ) 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 _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 _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 # find the catalog column behind that name col = None for source in catalog.sources: for table in source.tables: for c in table.columns: if c.name.lower() == date_column.strip().lower(): col = c break 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." )