"""Top-level semantic reconciliation and intent assertion. `reconcile_plan` orchestrates all evidence extraction, field/entity resolution, and repairs; `assert_intent_preserved` is the final gate that rejects a validated plan which dropped a material user requirement. """ from __future__ import annotations from app.planning.reconciliation.text import _WORD_RE, _norm, _singularize, _tokens from app.planning.reconciliation.evidence import _AGGREGATE_OPERATOR_TOKENS, _COMPARATOR_PHRASE_BOUNDARIES, _DATE_CONTEXT_TOKENS, _NON_METRIC_WORDS, _NUMERIC_CONSTRAINT_PATTERNS, _find_aggregate_cue_operation, _find_grouping_dimension_word, _find_most_fewest_dimension_word, _find_numeric_constraint, COMPARATIVE_RANK_RE, _infer_sort_direction, has_sort_intent, requested_limit from app.planning.reconciliation.field_projection import _explicit_field_list_request, _infer_explicit_field_list, _projection_with_requested_entity_names from app.planning.reconciliation.field_metrics import _resolve_grouping_dimension_column from app.planning.reconciliation.entities import _figure_table_behind_a_dimension_mention, _find_most_fewest_metric_table, _plan_is_locally_grounded, _resolve_entity_from_metric_evidence, explicit_table_mentions from app.planning.reconciliation.field_repairs import _repair_hallucinated_display_name_field, _repair_hallucinated_field_names, _repair_unrequested_related_fields from app.planning.reconciliation.filter_repairs import _ensure_aging, _ensure_open_when_falling_due, _ensure_overdue_condition, _ensure_category_filter, _match_names_loosely, _ensure_boolean_flag_filters, _ensure_counterparty_name_filter, _ensure_described_value_filters, _ensure_explicit_year_filter, _ensure_named_dimension_filter, _ensure_numeric_constraint_filter, _ensure_numeric_range_filter, _ensure_possessed_amount_filter, _ensure_relative_date_filter, _repair_unrequested_filters from app.planning.reconciliation.aggregate_repairs import _ensure_figure_total, _chronological_period_target, _detect_computed_metric_breakdown, _ensure_additional_aggregate, _ensure_comparative_status_grouping, _ensure_computed_metric_resolution, _ensure_grouped_aggregate, _ensure_scalar_superlative_aggregate, _repair_additional_aggregate_fields, _sort_target_group_key, repair_aggregate_field, repair_missing_count_operation, repair_ranking_sort, repair_unrequested_count from app.planning.models import RawQueryPlan from app.planning.models import SortSpec from app.planning.operators import AGGREGATE_OPERATIONS from app.planning.operators import Operation from app.schema.registry import SchemaSnapshot from app.schema.vocabulary import VocabularyIndex def assert_intent_preserved( question: str, raw_plan: RawQueryPlan, validated_plan: object, snapshot: SchemaSnapshot, vocab: VocabularyIndex, ) -> None: """Reject a validated plan that dropped a material user requirement. This is the final semantic gate before SQL construction. It does not repair or infer a plan: a missing entity, metric, or comparison is an error that returns to the normal LLM clarification/retry path. Keeping this check here makes every HTTP and worker execution use the same contract after reconciliation and schema validation have completed. """ from app.core.exceptions import QueryPlanValidationError if getattr(validated_plan, "settlement", None) is not None: # A settlement analysis states its whole requirement - the declared # invoice/payment link, the rule, the period, the name - on its spec # (`reconciliation.settlements`); there are no model-authored parts to check. return if getattr(validated_plan, "party_activity", None) is not None: # Repeat / lapsed parties: the declared source, the period and the # minimum are all on the spec (`reconciliation.row_shapes`). return if getattr(validated_plan, "duplicates", None) is not None: # Possible duplicates: the declared key and the period are on the spec. return if getattr(validated_plan, "aging", False) and getattr(validated_plan, "aging_buckets", None): # A due schedule: its windows are the buckets (`reconciliation.due_schedule`). return primary_table = getattr(validated_plan, "primary_table") filters = getattr(validated_plan, "filters") requested_operation = _find_aggregate_cue_operation(question) # "vendor with the most bills" implies COUNT just as much as "count of # bills by vendor" does - it just says so without any of the literal # aggregate words `_find_aggregate_cue_operation` looks for. Recognizing # the same idiom here (not just in `_ensure_grouped_aggregate`) keeps # this final gate from rejecting the very plan that idiom produces. requests_count = requested_operation == Operation.COUNT or _find_most_fewest_dimension_word(question) is not None # "daily average number of transactions": a count, averaged per day. requests_count = requests_count or getattr(validated_plan, "per_period_average", None) is not None if raw_plan.operation == Operation.COUNT and not requests_count: raise QueryPlanValidationError( f"The plan introduced {raw_plan.operation.value} even though the question does not request that aggregate." ) # A default projection is never evidence for an entity. If the model # selected a table which the question does not name and no field/metric # grounding survived planning, it is an arbitrary fallback. mentions = explicit_table_mentions(question, snapshot, vocab) question_terms = _tokens(question) - _NON_METRIC_WORDS referenced_fields = list(getattr(validated_plan, "select_fields", [])) referenced_fields.extend(field for field, _operator, _value, _value2 in filters) referenced_fields.extend(getattr(validated_plan, "group_by", [])) aggregate_field = getattr(validated_plan, "aggregate_field", None) if aggregate_field is not None: referenced_fields.append(aggregate_field) # A field is evidence only when its real schema identifier is actually # represented in the user request. This prevents a model's default # `users.id, users.name` projection from legitimizing an unrelated table. has_plan_grounding = any( _tokens(field.column.replace("_", " ")) & question_terms for field in referenced_fields ) is_unqualified_flat_list = ( raw_plan.operation in {Operation.LIST, Operation.GET, Operation.SEARCH} and not filters and getattr(validated_plan, "aggregate_field", None) is None and not getattr(validated_plan, "group_by", []) and getattr(validated_plan, "sort", None) is None ) if primary_table not in mentions and is_unqualified_flat_list and not has_plan_grounding: raise QueryPlanValidationError( f"The plan selected '{primary_table}' without schema-backed entity or metric evidence from the question." ) numeric_constraint = _find_numeric_constraint(question) if numeric_constraint is not None: required_operator, required_value = numeric_constraint matching = [ field for field, operator, value, _value2 in filters if operator == required_operator and value == required_value ] if not matching: raise QueryPlanValidationError( "The question contains a numeric comparison that is missing from the validated plan." ) # Metric words appearing in a real column identifier must survive on # the comparison field. This is schema-driven (derived from the live # columns) and therefore catches dropped requirements such as a # payable threshold without special-casing a business table. comparison_match = next( (pattern.search(question.lower()) for pattern, _operator in _NUMERIC_CONSTRAINT_PATTERNS if pattern.search(question.lower())), None, ) # Only words the comparison DIRECTLY attaches to are metric evidence: # the noun phrase immediately before the comparator ("expense amount # above 10000" -> "expense amount"; "for unpaid invoices above 50000" # -> "unpaid invoices"). Scanning the whole pre-comparator text # instead would make projection/status words ("customer name, total, # and balance for ... invoices above 50000" -> "name"/"balance"; # "... in the draft or void status above ..." -> "status") mandatory # evidence and reject the schema-correct filter on `total`. Units # after the number (for example "dollars") are never scanned at all. question_tokens: set[str] = set() before_text = question[:comparison_match.start()].lower() if comparison_match else question.lower() for match in reversed(list(_WORD_RE.finditer(before_text))): word = match.group(0) if word in _COMPARATOR_PHRASE_BOUNDARIES: break question_tokens.add(word) entity_tokens = { token for table in snapshot.tables for token in _tokens(table.replace("_", " ")) } entity_tokens |= {token + "s" for token in entity_tokens} # Evidence is scoped to the plan's OWN primary table's numeric # columns. A token like "status" from "... void status above 50000" # must never become mandatory because some unrelated table in the # schema happens to have an integer `status_code` column - only the # metric words that exist on the table the comparison actually runs # against are required to survive on its field. primary_table = getattr(validated_plan, "primary_table", None) required_metric_tokens: set[str] = set() if primary_table in snapshot.tables: required_metric_tokens = { token for column_name, column in snapshot.tables[primary_table].columns.items() for token in _tokens(column_name.replace("_", " ")) if len(token) >= 4 and token in question_tokens and token not in entity_tokens and column.type in {"integer", "decimal"} and token not in _NON_METRIC_WORDS and token not in _AGGREGATE_OPERATOR_TOKENS and token not in _DATE_CONTEXT_TOKENS } # "amount" is the generic monetary term that the shared canonical # resolution (_canonical_monetary_column) maps to the table's # authoritative monetary field (balance, total, or the literal # amount column). When the question says "amount above 5000" the # required metric token is "amount" but the resolved filter field # may be `balance` or `total` — neither contains the word "amount" # in its name. Accept the canonical monetary fields as valid # matches so this schema-correct resolution is not rejected. canonical_monetary = {"amount", "total", "balance"} def _metric_tokens_ok(field_col: str) -> bool: col_tokens = _tokens(field_col.replace("_", " ")) if required_metric_tokens & col_tokens: return True if required_metric_tokens <= {"amount"} and col_tokens & canonical_monetary: return True return False if required_metric_tokens and not any( _metric_tokens_ok(field.column) for field in matching ): raise QueryPlanValidationError( "The numeric comparison is not attached to the schema-backed metric requested by the user." ) def _drop_conditions_naming_the_metric_itself(raw_plan: RawQueryPlan, snapshot: SchemaSnapshot) -> None: """"Cash used in investing activities": a model condition such as `account_type_code = 'investing'` on a ledger figure only restates the figure's own name (investing cash flow) - the figure already selects its accounts. Such a condition is dropped; any other one is left for validation to judge (a ledger figure cannot be narrowed by a name).""" field = raw_plan.aggregate_field table = raw_plan.entity if not field or table not in snapshot.tables: return column = field.rsplit(".", 1)[-1] if not snapshot.is_computed(table, column): return descriptor = snapshot.tables[table].columns[column] own_words = set() for phrase in (column.replace("_", " "), *descriptor.aliases): own_words |= {_singularize(word) for word in _norm(phrase).split()} def restates(value) -> bool: values = value if isinstance(value, list) else [value] words = {_singularize(word) for item in values if isinstance(item, str) for word in _norm(item.replace("_", " ")).split()} return bool(words) and words <= own_words raw_plan.filters = [ f for f in raw_plan.filters if snapshot.column_type(table, f.field.rsplit(".", 1)[-1]) in ("date", "datetime") or not restates(f.value) ] def reconcile_plan( question: str, raw_plan: RawQueryPlan, snapshot: SchemaSnapshot, vocab: VocabularyIndex, forced_entity: str | None = None, platform: str | None = None, ) -> RawQueryPlan: """Pipeline stage "semantic reconciliation": repair a plan (the model's intent or a deterministic seed) against the question's explicit evidence and the schema, in place, and return it (the reconciled plan). This is where every deterministic repair runs - entity, declared metrics and shapes, operation, projection, grouping, filters, ranking, limit and sort - in the order documented in docs/architecture.md. It never adds a requirement the question did not state, and names no business table or question: everything comes from the schema and `schema_customizations.json`. `forced_entity` is an explicit caller scope that always wins. `platform` is the same normalized, request-level accounting-platform context (`planning.planner.build_query_plan`'s `platform_scope`, one of Xero/QuickBooks/Zoho or None) that `validation.query_validator.validate_query_plan` applies as a filter. It is threaded in here too so it is available throughout semantic reconciliation/planning - not just at the very end, at validation time - the same "question + platform" context a human analyst would have from the start of reasoning about the request. It remains, here as everywhere else, a plain data-source *context* constraint: reconciliation never uses it to choose, override, or repair an entity, table, column, join, or field. Every existing repair in this function (entity correction, filter/field pruning, ranking/aggregate repair, relative-date resolution) is driven purely by the question text and the schema - identical whether `platform` is omitted or set to any of the 3 supported values (see `tests/unit/test_platform_entity_resolution_regression.py`, section 1a). Once the entity/fields/filters/aggregate are resolved from the question alone, `platform` is applied, unchanged, as a plain filter on whichever already-chosen table's curated authoritative platform field is known - see `validation.query_validator._apply_platform_scope` - and silently becomes a no-op on a table with no such curated field, exactly as before. """ # Shapes only reconciliation decides - never taken from the model's JSON. raw_plan.aging = False raw_plan.aging_as_of = None raw_plan.aging_also = [] raw_plan.aging_buckets = None raw_plan.overdue_split = False raw_plan.ratio = None raw_plan.party_facet = None raw_plan.settlement = None raw_plan.party_activity = None raw_plan.duplicates = None raw_plan.per_period_average = None raw_plan.metric_category = None raw_plan.metric_set = None raw_plan.metric_set_difference = False raw_plan.days_overdue = False explicit_tables = explicit_table_mentions(question, snapshot, vocab) # Checked before any entity-mention/table-correction logic below: a # multi-table computed-metric breakdown ("revenue vs expenses for each # month") names no real table at all in its own wording (the tables # only exist inside the metric's own schema-declared components), so # the ordinary entity-resolution evidence below has nothing to work # from and would otherwise land on whichever single table happens to # score best against a bare word like "expenses" - losing the other # side of the comparison entirely. `forced_entity` (an explicit, # caller-supplied scope) still always wins outright. is_breakdown_request = ( not forced_entity and _detect_computed_metric_breakdown(raw_plan, question, snapshot) ) if forced_entity: raw_plan.entity = forced_entity elif is_breakdown_request: pass # entity/operation/aggregate_field/group_by already fully set above else: # The counted noun in "X with the most/fewest Y" is checked first, # ahead of the generic single-explicit-table-mention path below: for # that idiom, X (the grouping dimension) is very often *also* a real # table name in its own right ("customer" -> `customers`), so the # question ends up naming two real tables at once and the # single-match path below never uniquely fires. The counted noun is # unambiguous, stronger evidence for the entity holding the # countable rows in exactly this construction; everywhere else, # falls straight through to the existing single-mention evidence. most_fewest_table = _find_most_fewest_metric_table(question, snapshot, vocab) # "top 5 customers by revenue": the one table the question names is # the grouping dimension, and the figure lives on another table # that carries that dimension as a column. The figure's table is # the entity; the named table becomes the GROUP BY (see # `_ensure_grouped_aggregate` below). figure_table = ( None if most_fewest_table else _figure_table_behind_a_dimension_mention(question, raw_plan, snapshot, explicit_tables) ) candidate_table = most_fewest_table or figure_table or ( next(iter(explicit_tables)) if len(explicit_tables) == 1 else None ) # The single-mention path above cannot resolve a grouped-aggregate # question that names two real tables at once (the grouping dimension # "each customer" and the metric table "invoices" both resolve to # tables). When the model's entity is not locally grounded - its # fields/aggregates don't exist there - the mention that actually # carries the question's own metric/grouping evidence wins instead of # letting a hallucinated plan fail validation or silently degrade to # a flat LIST. See `_resolve_entity_from_metric_evidence`. if candidate_table is None: candidate_table = _resolve_entity_from_metric_evidence( question, raw_plan, snapshot, explicit_tables, ) if candidate_table: skip_override = False original_entity = raw_plan.entity # A plan that is already completely grounded in its declared entity is # stronger evidence than a lexical table mention. In particular, # aliases such as "payable" may describe a metric on a vendor contact # record rather than request a bill row. Replacing that entity here # used to turn valid projections/filters into a different business # question. Entity correction remains available for an ungrounded # (hallucinated or incomplete) plan below. if _plan_is_locally_grounded(raw_plan, original_entity, snapshot) and candidate_table != figure_table: skip_override = True if original_entity in snapshot.tables and original_entity != candidate_table: # A bare grouping-dimension word ("by department", "for each # vendor") can coincidentally also be a table's own name/alias # ("departments"). If the model's own entity already has a real # local column for that exact dimension, the word is describing # a grouping attribute of the model's entity, not naming a # different primary table - overriding here would silently # swap a working entity for an unrelated one that can't even # answer the rest of the plan (its filters/aggregate field). # But when the explicitly-mentioned table resolves the SAME # dimension just as well, the coincidence argument doesn't hold # - the literal mention is then genuine, stronger evidence (per # the rule below), not a false-positive table-name clash, so it # must still win. dimension_word = _find_grouping_dimension_word(question) if ( dimension_word is not None and _resolve_grouping_dimension_column(dimension_word, original_entity, snapshot) is not None and _resolve_grouping_dimension_column(dimension_word, candidate_table, snapshot) is None ): skip_override = True if not skip_override: # An explicitly named primary entity is stronger evidence than a # model guess. This is the main protection against unrelated # table drift. raw_plan.entity = candidate_table # A schema-declared computed metric (e.g. "profit") is a first-class # semantic field: the question's own name/alias for it is identity # evidence exactly like a real column's. Runs after entity resolution and # component-breakdown detection (so an explicit caller scope and a # breakdown both win) but BEFORE `primary_table` is derived below, so the # metric's own declared anchor table is what every following repair # (filters, dates, grouping, projection) resolves against - and so a plan # the model left incomplete or flagged `needs_clarification` still # resolves to the metric the user actually asked for. _ensure_computed_metric_resolution(raw_plan, question, snapshot, forced_entity=forced_entity) _drop_conditions_naming_the_metric_itself(raw_plan, snapshot) # "the net cash change and its drivers": the metric declared as its drivers. from app.planning.reconciliation.aggregate_repairs import apply_drivers_metric apply_drivers_metric(raw_plan, question, snapshot) # A metric that decides its own shape is grouped only when asked to be. from app.planning.reconciliation.aggregate_repairs import settle_shape_owning_metric settle_shape_owning_metric(raw_plan, question, snapshot) # "insurance expense trend": the ledger metric narrowed to those accounts. from app.planning.reconciliation.ledger_categories import apply_metric_category apply_metric_category(raw_plan, question, snapshot) # "what caused the variance in operating expenses": by account. from app.planning.reconciliation.ledger_categories import apply_driver_grouping apply_driver_grouping(raw_plan, question, snapshot) # "gross profit and net profit", "KPI scorecard": several declared # metrics answered together (`reconciliation.metric_sets`). if not forced_entity: from app.planning.reconciliation.metric_sets import apply_metric_set apply_metric_set(raw_plan, question, snapshot) # "largest outstanding receivable by customer", "which customer owes us # the most": a declared party facet asked of every party - grouped on the # facet's own table (`reconciliation.party_facets.apply_party_ranking_facet`). if not forced_entity and not is_breakdown_request: from app.planning.reconciliation.party_facets import apply_party_ranking_facet apply_party_ranking_facet(raw_plan, question, snapshot) # "active customers this month", "freight charges billed to customers": # the table that records the figure, not the one named as the party # (`reconciliation.row_shapes`). if not forced_entity and raw_plan.party_facet is None and not raw_plan.metric_set: from app.planning.reconciliation.row_shapes import apply_active_parties, apply_overdue_parties, retarget_to_named_figure if not apply_overdue_parties(raw_plan, question, snapshot) and not apply_active_parties(raw_plan, question, snapshot): retarget_to_named_figure(raw_plan, question, snapshot) primary_table = raw_plan.entity if raw_plan.entity in snapshot.tables else None if primary_table: # Operation-level repair first: "how many X" that the model answered # as a bare row listing is a lost requirement, not a field-level # ambiguity, so it is corrected before any of the below repairs # (projection, grouping, filters) build on the wrong operation. repair_unrequested_count(raw_plan, question) repair_missing_count_operation(raw_plan, question, primary_table, snapshot) _repair_unrequested_related_fields( raw_plan, question, primary_table, explicit_tables, snapshot, entity_is_user_selected=bool(forced_entity or len(explicit_tables) == 1), ) # Must run before `_ensure_grouped_aggregate`: a " vs # " comparison names two of a column's own described VALUES, # not a dimension noun, so it needs its own detector - but once it # fires, `_ensure_grouped_aggregate` must see the resulting plan as # already grouped/aggregate-shaped and leave it alone rather than # re-deriving a different dimension from the same question text. _ensure_comparative_status_grouping(raw_plan, question, primary_table, snapshot) _ensure_grouped_aggregate(raw_plan, question, primary_table, snapshot) # Ungrouped counterpart of the repair just above: a bare "the # highest/lowest X" with no ranking subject and no grouping # dimension can only mean one row's extreme value (rule 69), never # an unlimited per-row listing sorted by it. Must run after # `_ensure_grouped_aggregate` so a genuinely grouped request is # already shaped (and skipped here) by the time this runs. _ensure_scalar_superlative_aggregate(raw_plan, question, primary_table, snapshot) # A question that names a figure ("revenue for customer Acme in # August") is answered with that figure, not with the rows behind # it. After the grouped/superlative shapes above, so a request that # is really one of those is already shaped and skipped here. _ensure_figure_total(raw_plan, question, primary_table, snapshot) _ensure_additional_aggregate(raw_plan, question, primary_table, snapshot) _repair_additional_aggregate_fields(raw_plan, primary_table, snapshot) # An explicit field projection ("show the invoice number, customer # name, total, and balance ...") must survive even when the model # returned an empty field list - otherwise validation falls back to # the table's default projection and the user's requested columns # are silently broadened. Only row-listing operations qualify # (aggregate/grouped plans select their own expressions), and only # when the question itself reads like an explicit projection (see # `_explicit_field_list_request`) - a bare sort/filter reference # never triggers this. if ( raw_plan.operation in {Operation.LIST, Operation.GET, Operation.SEARCH} and not raw_plan.select_all and not raw_plan.fields ): inferred_fields = _infer_explicit_field_list(question, primary_table, snapshot) # An explicit projection may also name the entity itself only by # its business word ("each customer's name") when the table # carries no column literally named `_name` - its display # name lives in a generic name-shaped column (`contact_name`, # `company_name`, ...). Without this, "show the customer's name, # outstanding receivable amount, and status" recovers only the # two verbatim-named columns and the requested name silently # disappears from the SELECT. projection_fields = _projection_with_requested_entity_names( question, primary_table, snapshot, inferred_fields ) if _explicit_field_list_request(question, projection_fields): raw_plan.fields = projection_fields elif ( raw_plan.operation in {Operation.LIST, Operation.GET, Operation.SEARCH} and not raw_plan.select_all and raw_plan.fields ): # The model may ALSO return the projection already filled in but # write the requested entity name as the bare word ("name") that # is not a real column on the entity. Repair just that word with # the table's own generic display-name columns (identity-based), # leaving every other listed field untouched. _repair_hallucinated_display_name_field( raw_plan, question, primary_table, snapshot ) _repair_hallucinated_field_names(raw_plan, primary_table, snapshot) _repair_unrequested_filters(raw_plan, question, primary_table, explicit_tables, snapshot) _ensure_relative_date_filter(raw_plan, question, primary_table, snapshot) _ensure_explicit_year_filter(raw_plan, question, primary_table, snapshot) _ensure_described_value_filters(raw_plan, question, primary_table, snapshot) _ensure_named_dimension_filter(raw_plan, question, primary_table, snapshot) _ensure_category_filter(raw_plan, question, primary_table, snapshot) _ensure_counterparty_name_filter(raw_plan, question, primary_table, snapshot) # Last among the filter repairs: whoever wrote a name condition, it # matches the name rather than equating it (see `_match_names_loosely`). _match_names_loosely(raw_plan, question, primary_table, snapshot) _ensure_boolean_flag_filters(raw_plan, question, primary_table, snapshot) _ensure_numeric_constraint_filter(raw_plan, question, primary_table, snapshot) _ensure_numeric_range_filter(raw_plan, question, primary_table, snapshot) _ensure_possessed_amount_filter(raw_plan, question, primary_table, snapshot) if repair_ranking_sort(raw_plan, question, primary_table, snapshot): # A genuine tie remains unresolved because the helper returns # False; clear only an LLM's spurious ambiguity flag here. raw_plan.needs_clarification = False raw_plan.clarification_question = None if repair_aggregate_field(raw_plan, question, primary_table, snapshot): raw_plan.needs_clarification = False raw_plan.clarification_question = None # Last: an aging analysis, else "overdue" as a date condition on # whatever shape the plan now has. _ensure_aging(raw_plan, question, primary_table, snapshot) if not raw_plan.aging: _ensure_overdue_condition(raw_plan, question, primary_table, snapshot) _ensure_open_when_falling_due(raw_plan, question, primary_table, snapshot) from app.planning.reconciliation.filter_repairs import apply_days_overdue apply_days_overdue(raw_plan, question, primary_table, snapshot) # A rate is shown row by row, never summed; "and for what reasons" is # the table's own reason column (`reconciliation.row_shapes`). from app.planning.reconciliation.row_shapes import apply_rate_listing, apply_reason_grouping apply_rate_listing(raw_plan, question, primary_table, snapshot) apply_reason_grouping(raw_plan, question, primary_table, snapshot) # "X as a percentage of Y": two figures over one period. Only the # period survives from the plan; each side carries its own category. from app.planning.reconciliation.ratios import apply_ratio # One named customer's / vendor's declared facet ("payment history for # vendor Acme"): the facet's own tables, the name and the period. from app.planning.reconciliation.party_facets import apply_party_facet # How customers paid - collection time, late-payment behaviour # (`reconciliation.settlements`) - from the declared invoice/payment link. from app.planning.reconciliation.settlements import apply_settlement from app.core.config import get_settings from app.planning.reconciliation.row_shapes import apply_duplicates, apply_party_activity, apply_per_period_average from app.planning.reconciliation.due_schedule import apply_due_schedule if apply_settlement(raw_plan, question, snapshot): pass elif not forced_entity and apply_due_schedule(raw_plan, question, snapshot): # What is scheduled to come in / go out, by due date - the aging # query with the schedule's own buckets. pass elif apply_duplicates(raw_plan, question, primary_table, snapshot, get_settings().app_timezone): # Possible duplicate documents (same party, amount, date): `sql.builder`'s. pass elif not forced_entity and apply_party_activity(raw_plan, question, snapshot, get_settings().app_timezone): # Repeat / lapsed customers or vendors: the rows are `sql.builder`'s. pass elif ( raw_plan.ratio is None and not raw_plan.metric_set and (raw_plan.classification is None or raw_plan.classification.side in ("in", "out")) and raw_plan.party_facet is None and not raw_plan.aging and apply_per_period_average(raw_plan, question, primary_table, snapshot, explicit_tables) ): # "daily average number of transactions": one figure per day/month/... pass elif not apply_party_facet(raw_plan, question, snapshot) and apply_ratio(raw_plan, question, snapshot): from app.planning.reconciliation.field_dates import _infer_date_filter_field as _date_field date_column = _date_field(raw_plan.entity, snapshot) raw_plan.filters = [ f.model_copy(update={"field": date_column}) for f in raw_plan.filters if date_column is not None and snapshot.column_type( f.field.split(".", 1)[0] if "." in f.field else primary_table, f.field.rsplit(".", 1)[-1] ) in ("date", "datetime") ][:1] # User-intent authority for optional SQL clauses. # Ordinal record-position requests are handled deterministically by # `record_positions`; they are not row limits. if raw_plan.record_positions: raw_plan.limit = None raw_plan.sort = None else: # A stated N, or exactly one row for the singular-superlative idiom # ("which customer has the highest outstanding balance?") - the # limit is decided here, once, from the question and the plan # shape, never from the model's own `limit`. raw_plan.limit = requested_limit(question, raw_plan) if not has_sort_intent(question): # Never add sorting to an otherwise unsorted query - whatever # the model (or an upstream repair) set is discarded. raw_plan.sort = None elif _chronological_period_target(question, raw_plan) is not None: # "ordered chronologically" on a time-bucketed aggregate # ("each month with the total bill amount, ordered # chronologically") orders by the bucket's own ascending # sequence - never by the per-group aggregate value, which a # generic ranking reading would produce. The bucket period word # already stored in `group_by` is the sort key; the validator # resolves it to the same DATE_FORMAT expression the GROUP BY # uses (see `validation.query_validator._resolve_sort`). raw_plan.sort = SortSpec( field=_chronological_period_target(question, raw_plan), direction="ASC" ) elif raw_plan.sort is not None: # The sort *field* above may come from the model directly, from # `_ensure_grouped_aggregate`, or from `repair_ranking_sort` - # but `direction` is never trusted from any of them. It is # always the single, deterministic, question-driven mapping # (`_infer_sort_direction`), applied last and unconditionally, # so a model that names the right field but reverses "ascending" # vs "descending" (or "earliest" vs "latest", etc.) can never # make it into the final plan - direction cannot be lost or # reversed between here and SQL generation. sort_field = raw_plan.sort.field if ( sort_field == "result" and raw_plan.group_by and raw_plan.group_by.fields and raw_plan.operation in ({Operation.GROUP} | AGGREGATE_OPERATIONS) ): # `_ensure_grouped_aggregate` defaulted the sort to the # aggregate result, but the question may explicitly name # the group key as the ordering target ("ordered by # month", "ordered by vendor name") - then the bucket/ # dimension column is the requested sort, never the # per-group result. group_key = _sort_target_group_key(question, raw_plan, primary_table, snapshot) if group_key is not None: sort_field = group_key if ( raw_plan.group_by and raw_plan.group_by.fields and raw_plan.operation in ({Operation.GROUP} | AGGREGATE_OPERATIONS) and sort_field != "result" and sort_field not in raw_plan.group_by.fields ): # In a grouped-aggregate query, ranking language ("ordered # by total invoice amount", "top 5 vendors by balance") # refers to the per-group aggregate itself - never to an # arbitrary raw row value inside each group. A raw sort # column here (e.g. `total` beside `SUM(total)`, or any # unrelated column beside a grouped COUNT) would order by # whichever single row MySQL happens to surface per group, # silently discarding the requested ranking. The group key # itself ("ordered by vendor name") is the one legitimate # raw sort target and is left untouched. sort_field = "result" if ( raw_plan.operation in ({Operation.GROUP} | AGGREGATE_OPERATIONS) and not raw_plan.fields and not (raw_plan.group_by and raw_plan.group_by.fields) and sort_field != "result" ): # A scalar aggregate (SUM/AVG/MIN/MAX/COUNT over a single # result row, with no entity display fields) has no per-row # column to order by: "top 5 invoices by total amount" on a # SUM plan must order by the aggregate result itself # (`result`), never by a raw column - a raw sort would be # invalid under ONLY_FULL_GROUP_BY and, for a top-N limit, # would rank nothing meaningful. The one exception is an # entity+value MAX/MIN request ("which customer has the # highest balance"), which carries entity display fields and # deliberately sorts by the raw metric column - the # `not raw_plan.fields` guard keeps that shape untouched. sort_field = "result" raw_plan.sort = SortSpec(field=sort_field, direction=_infer_sort_direction(question)) elif raw_plan.group_by and raw_plan.group_by.fields and raw_plan.aggregate_field: # The model can also produce an already-correct grouped # aggregate (group_by + aggregate_field) without ever setting # `sort` at all - `_ensure_grouped_aggregate` only adds this # when it builds the grouping itself, not when the model # already supplied one. Ordering by the aggregate result is # exactly what "highest/lowest/ascending/..." means for a # per-group total, so it must not be silently dropped here - # unless the question explicitly names the group key as the # ordering target instead. sort_field = "result" if raw_plan.operation in ({Operation.GROUP} | AGGREGATE_OPERATIONS): group_key = _sort_target_group_key(question, raw_plan, primary_table, snapshot) if group_key is not None: sort_field = group_key raw_plan.sort = SortSpec(field=sort_field, direction=_infer_sort_direction(question)) elif ( raw_plan.operation in ({Operation.GROUP} | AGGREGATE_OPERATIONS) and not raw_plan.fields and not (raw_plan.group_by and raw_plan.group_by.fields) and raw_plan.limit is not None ): # A scalar aggregate with an explicit top/bottom-N limit ("Show # the top 5 invoices by total amount") is a ranking of the # aggregate result even when the model never set a sort at all: # without an ORDER BY, the LIMIT would arbitrarily cut the # single-row result. Bind the ordering to the aggregate result # with the question's own direction (top -> DESC, bottom -> # ASC), exactly as the grouped path does above. raw_plan.sort = SortSpec(field="result", direction=_infer_sort_direction(question)) # A yes/no flag stays a yes/no condition (never a number, a count or a # dated status) - see `filter_repairs.settle_flag_conditions`. from app.planning.reconciliation.filter_repairs import settle_flag_conditions if raw_plan.entity in snapshot.tables: settle_flag_conditions(raw_plan, question, raw_plan.entity, snapshot) # "which entity has the higher net margin": a comparative names a ranked # winner exactly as a superlative does - the metric per member, largest # (smallest) first, one row unless the question states a number. comparative = COMPARATIVE_RANK_RE.search(_norm(question)) aggregate = raw_plan.aggregate_field.rsplit(".", 1)[-1] if raw_plan.aggregate_field else None if ( comparative is not None and raw_plan.group_by is not None and raw_plan.group_by.fields and aggregate is not None and snapshot.is_computed(raw_plan.entity, aggregate) ): low = comparative.group(2) in {"lower", "smaller", "worse", "fewer", "less"} raw_plan.sort = SortSpec(field="result", direction="ASC" if low else "DESC") raw_plan.limit = requested_limit(question, raw_plan) return raw_plan