"""harness/semantic.py — the semantic seam (OM-0a, 2026-07-11). The ONE governed registry of topics + metrics (platform/model/*.yml) and the resolver that turns a metric key into a number through the SAME proven data layer the validated modules use (core/odoo domain builders + aggregate helpers) — so scope parity is by construction, and every metric carries its definition, ai_context, and an independent validate contract. Consumers (the point of the seam): the app's pages, the metric dictionary (OM-0c), the saved- dashboard viewer (OM-3; the Explore page was retired 2026-07-23), the AI Analyst's tools (OM-4: list_topics / describe_topic / run_semantic_query), and the MCP server (OM-6). Plan: .claude/wiki/research/omni-adoption.md (Part IV + the productization addendum). Design constraints honored (from /odoo-api): - read_group dot-path GROUPBY fails → scalar aggregates here use sum_field/distinct_count; grouped queries (OM-3+) must group by direct m2o fields only. - Scope is never hand-rolled: topics BIND to a named domain builder (sale_line_domain, ...); the YAML `scope:` block is the declarative statement of what that builder bakes in. - READ-ONLY on Odoo, always. """ import ast import functools import operator from pathlib import Path import yaml import core.odoo as O MODEL_DIR = Path(__file__).resolve().parents[1] / "model" # Topic scope binds to a NAMED, reviewed domain builder — never a YAML-assembled domain (one # source of truth for scope until OM-2 makes topics executable against the store). def _order_domain(*a, **kw): # lazy: order scope currently lives in modules/sales.py; OM-2 lifts it into core/the store import modules.sales as S return S.order_domain(*a, **kw) def _customer_move_domain(move_type): """A posted customer document domain, matching `modules/returns.py`'s `_mdom` exactly. ⚠ COMPANY-LEVEL, and it REFUSES a team rather than ignoring one. Credit notes all carry team_id = 1, so a BU filter on the document is meaningless — silently dropping the argument would hand a business-unit user the company's number under their own name, which is the widening sin in a different currency. """ def build(date_from=None, date_to=None, team_id=None, **kw): if team_id: raise ModelError( f"customer {move_type} documents are COMPANY-LEVEL — credit notes and invoices " f"carry no trustworthy business-unit tag, so a team-scoped total cannot be " f"produced. Refusing rather than returning the company number.") dom = [("move_type", "=", move_type), ("state", "=", "posted")] if date_from: dom.append(("invoice_date", ">=", date_from)) if date_to: dom.append(("invoice_date", "<=", date_to)) return dom return build _BUILDERS = { "sale_line_domain": O.sale_line_domain, "sale_order_domain": _order_domain, "credit_note_domain": _customer_move_domain("out_refund"), "customer_invoice_domain": _customer_move_domain("out_invoice"), } class ModelError(Exception): """A model-file problem (unknown key, bad reference, unparseable expr).""" # ---------------------------------------------------------------- loading @functools.lru_cache(maxsize=1) def _tenant(): f = MODEL_DIR / "tenant.yml" return yaml.safe_load(f.read_text(encoding="utf-8")) if f.exists() else {"params": {}} def _sql_value(v): """Render a tenant param into SQL safely (params are versioned config, not user input — but escape anyway): ints as-is, lists comma-joined, strings single-quoted + escaped.""" if isinstance(v, bool): raise ModelError("boolean tenant params not supported in scope_sql") if isinstance(v, (int, float)): return str(v) if isinstance(v, (list, tuple)): return ", ".join(_sql_value(x) for x in v) return "'" + str(v).replace("'", "''") + "'" def render_scope(sql): """Substitute {param} placeholders in a store scope/filter SQL fragment from tenant.yml.""" params = _tenant().get("params") or {} out = sql for k, v in params.items(): out = out.replace("{" + k + "}", _sql_value(v)) if "{" in out and "}" in out: import re as _re missing = _re.findall(r"\{([a-z_]+)\}", out) if missing: raise ModelError(f"scope_sql references unknown tenant params: {missing}") return out @functools.lru_cache(maxsize=1) def _model(): topics, metrics = {}, {} for f in sorted((MODEL_DIR / "topics").glob("*.yml")): d = yaml.safe_load(f.read_text(encoding="utf-8")) if not d or "key" not in d: raise ModelError(f"topic file {f.name} lacks a 'key'") if d.get("domain_builder") and d["domain_builder"] not in _BUILDERS: raise ModelError(f"topic {d['key']}: unknown domain_builder {d.get('domain_builder')!r}") # a topic may be STORE-ONLY (no live-path builder yet, e.g. gl_lines) — metric() raises # a clear error if a live resolution is attempted on it topics[d["key"]] = d for f in sorted((MODEL_DIR / "metrics").glob("*.yml")): d = yaml.safe_load(f.read_text(encoding="utf-8")) topic = d.get("topic") if topic not in topics: raise ModelError(f"metrics file {f.name}: unknown topic {topic!r}") for m in d.get("metrics", []): if "key" not in m: raise ModelError(f"metrics file {f.name}: a metric lacks a 'key'") if m["key"] in metrics: raise ModelError(f"duplicate metric key {m['key']!r}") m.setdefault("topic", topic) # a metric may name its own topic (rare) if m["topic"] not in topics: raise ModelError(f"metric {m['key']!r}: unknown topic {m['topic']!r}") metrics[m["key"]] = m return {"topics": topics, "metrics": metrics} def reload_model(): _model.cache_clear() _tenant.cache_clear() return _model() def topics(): return dict(_model()["topics"]) def metrics(): return dict(_model()["metrics"]) def describe(key): """Full metadata for a metric (the dictionary page / AI describe_topic feed).""" m = _model()["metrics"].get(key) if not m: raise ModelError(f"unknown metric {key!r}") return {**m, "topic_def": _model()["topics"][m["topic"]]} # ---------------------------------------------------------------- resolving def _domain(topic_def, date_from, date_to, team_id): b = topic_def.get("domain_builder") if not b: raise ModelError(f"topic {topic_def['key']!r} is store-only — use store_query() " f"(no live-path domain builder bound yet)") return _BUILDERS[b](date_from, date_to, team_id=team_id) _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv} def _eval_expr(expr, resolve): """Safely evaluate a derived-metric expression: metric keys, numbers, + - * / and parens.""" def ev(node): if isinstance(node, ast.Expression): return ev(node.body) if isinstance(node, ast.BinOp) and type(node.op) in _OPS: return _OPS[type(node.op)](ev(node.left), ev(node.right)) if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub): return -ev(node.operand) if isinstance(node, ast.Num): # py<3.8 compat name; Constant below return node.n if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return node.value if isinstance(node, ast.Name): return resolve(node.id) raise ModelError(f"disallowed token in derived expr: {ast.dump(node)}") try: return ev(ast.parse(expr, mode="eval")) except SyntaxError as e: raise ModelError(f"unparseable derived expr {expr!r}: {e}") def metric(key, date_from=None, date_to=None, team_id=None, _seen=None): """Resolve a registered metric to a number for a window (+optional BU). The only public compute path — every surface that shows this metric goes through here.""" _seen = _seen or set() if key in _seen: raise ModelError(f"circular metric reference at {key!r}") _seen = _seen | {key} m = _model()["metrics"].get(key) if not m: raise ModelError(f"unknown metric {key!r}") agg = m.get("agg") if agg == "ratio": num = metric(m["numerator"], date_from, date_to, team_id, _seen) den = metric(m["denominator"], date_from, date_to, team_id, _seen) return (num / den) if den else 0.0 if agg == "derived": return _eval_expr(m["expr"], lambda k: metric(k, date_from, date_to, team_id, _seen)) t = _model()["topics"][m["topic"]] dom = _domain(t, date_from, date_to, team_id) # Wave 21 R2 — the LIVE twin of `store_filter_sql`. A filtered metric must filter BOTH # paths, or store_parity would compare a narrowed store number against an unfiltered live # one and the "parity failure" would read as a data problem. YAML shape: # live_domain: [[field, op, value], ...] — appended verbatim to the topic domain. for c in (m.get("live_domain") or []): dom = dom + [tuple(c)] if agg == "sum": # ⚠ `negate` applies HERE too, not only in `_measure_sql`. It existed for GL income on a # STORE-ONLY topic, so the live path never met it — the first metric that is both negated # and parity-checked (`returns`) would otherwise compare a negative live number against a # positive store one, and the "parity failure" would look like a data problem. got = O.sum_field(t["entity"], dom, m["field"]) return -got if m.get("negate") else got if agg == "count_distinct": return O.distinct_count(t["entity"], dom, m["field"]) if agg == "count": return O.get_odoo().search_count(t["entity"], dom) raise ModelError(f"metric {key!r}: unknown agg {agg!r}") # ---------------------------------------------------------------- validate contracts def _v_order_level_revenue(date_from, date_to, team_id): """Independent check: line-level Σ price_subtotal == order-level Σ amount_untaxed.""" import modules.sales as S line = metric("revenue", date_from, date_to, team_id) order = O.sum_field("sale.order", S.order_domain(date_from, date_to, team_id), "amount_untaxed") return line, order def _v_order_count(date_from, date_to, team_id): """Exact reconciliation: order-level count == distinct line-parent orders + line-LESS orders. (Confirmed orders with zero lines exist in the data — 2 found at first proof, 2026-07-11; counting them explicitly makes the identity exact instead of hiding them in a tolerance, and surfaces them as a data-health signal.)""" import modules.sales as S order_level = metric("orders", date_from, date_to, team_id) dom = S.order_domain(date_from, date_to, team_id) from_lines = O.distinct_count("sale.order.line", O.sale_line_domain(date_from, date_to, team_id), "order_id") empty = O.get_odoo().search_count("sale.order", dom + [("order_line", "=", False)]) return order_level, from_lines + empty def _v_gl_opex(date_from, date_to, team_id): """Store-path opex == the live Odoo aggregate with the Expenses-module domain (posted, expense-type accounts), to the cent. team_id N/A (company-level).""" res = store_query("gl_lines", ["opex"], date_from=date_from, date_to=date_to) ours = res["rows"][0]["opex"] if res["rows"] else 0 dom = [("move_id.state", "=", "posted"), ("account_id.account_type", "in", ["expense", "expense_depreciation"])] if date_from: dom.append(("date", ">=", date_from)) if date_to: dom.append(("date", "<=", date_to)) live = O.sum_field("account.move.line", dom, "balance") return ours, live def _v_ar_outstanding(date_from, date_to, team_id): """Store-path AR outstanding == live Odoo aggregate (posted out_invoice/out_refund residuals). No date window — outstanding is an as-of-now balance.""" res = store_query("receivables", ["ar_outstanding"]) ours = res["rows"][0]["ar_outstanding"] if res["rows"] else 0 live = O.sum_field("account.move", [("state", "=", "posted"), ("move_type", "in", ["out_invoice", "out_refund"])], "amount_residual_signed") return ours, live def _v_returns_sign(date_from, date_to, team_id): """Independent check on the NEGATION: `returns` must equal the ABSOLUTE untaxed total of the same posted credit notes, read straight from Odoo without going through the metric. A lost negation makes returns negative; a doubled one makes it negative again. Both read as "no returns" rather than as an error, and neither is visible to any self-consistency check — the store and the live path would agree with each other perfectly while both being wrong. """ ours = metric("returns", date_from, date_to, team_id) raw = O.sum_field("account.move", _BUILDERS["credit_note_domain"](date_from, date_to, team_id), "amount_untaxed_signed") return ours, abs(raw) def _v_line_vs_move_grain(date_from, date_to, team_id): """Independent check on the invoice-LINE topic: the store's netted product-line total must equal the same figure read straight from LIVE Odoo. Two things this catches that self-consistency cannot. (1) A LOST CREDIT-NOTE NEGATION — `price_subtotal` is stored POSITIVE on a refund line, so a dropped subtraction inflates revenue while every internal check still agrees. (2) A WRONG `display_type` FILTER — the invoice tables carry 'cogs' and 'payment_term' rows alongside 'product' ones, and including them changes the total silently rather than erroring. Company-level (team_id None) uses the strongest possible reference: the MOVE-grain `amount_untaxed_signed`, a different table entirely. Per-BU there is no move-grain analogue (account.move carries no business unit — every invoice sits on team 1), so the reference is the live line-level aggregate reached through the sale-order dot path: still an independent engine and query path from the DuckDB store, which is what the contract requires. """ ours = (store_query("invoice_lines", ["invoiced_line_sales"], date_from=date_from, date_to=date_to, team_id=team_id) .get("rows") or [{}])[0].get("invoiced_line_sales") or 0.0 if team_id is None: raw = O.sum_field("account.move", [("state", "=", "posted"), ("move_type", "in", ["out_invoice", "out_refund"]), ("invoice_date", ">=", date_from), ("invoice_date", "<=", date_to)], "amount_untaxed_signed") return ours, raw dom = [("parent_state", "=", "posted"), ("display_type", "=", "product"), ("date", ">=", date_from), ("date", "<=", date_to), ("sale_line_ids.order_id.team_id", "=", team_id)] gross = O.sum_field("account.move.line", dom + [("move_type", "=", "out_invoice")], "price_subtotal") refunds = O.sum_field("account.move.line", dom + [("move_type", "=", "out_refund")], "price_subtotal") return ours, (gross or 0) - (refunds or 0) _VALIDATORS = { "returns_sign": _v_returns_sign, "line_vs_move_grain": _v_line_vs_move_grain, "order_level_revenue": _v_order_level_revenue, "order_count": _v_order_count, "gl_opex": _v_gl_opex, "ar_outstanding": _v_ar_outstanding, } def validate_metric(key, date_from=None, date_to=None, team_id=None, tol=0.01): """Run a metric's declared independent cross-check. Returns the standard check dict (the validate.py convention: check/ok/gap).""" m = _model()["metrics"].get(key) if not m: raise ModelError(f"unknown metric {key!r}") contract = m.get("validate") if not contract: return {"check": f"semantic.{key}: no independent contract declared", "ok": True, "gap": 0.0, "note": "declare one in model/metrics when an independent aggregate exists"} fn = _VALIDATORS.get(contract["method"]) if not fn: raise ModelError(f"metric {key!r}: unknown validate method {contract['method']!r}") ours, independent = fn(date_from, date_to, team_id) gap = abs((ours or 0) - (independent or 0)) return {"check": f"semantic.{key} == {contract['method']} ({date_from}..{date_to}, team={team_id})", "ok": gap <= tol, "gap": round(gap, 4), "ours": ours, "independent": independent} def validate(pre=None): """Module-convention validate(): run every declared contract over YTD, all-BU + per-BU.""" import core.periods as P yf, yt = P.ytd() out = [] for key, m in _model()["metrics"].items(): if not m.get("validate"): continue out.append(validate_metric(key, yf, yt, None)) topic = _model()["topics"][m["topic"]] if (topic.get("store") or {}).get("team_col"): # company-level topics have no BU runs for tid in O.TEAM_IDS: out.append(validate_metric(key, yf, yt, tid)) return out # ---------------------------------------------------------------- store path (OM-2) # The compiler behind run_semantic_query: registered metrics + whitelisted dims → DuckDB SQL over # the tenant store (harness/datastore.py). EVERY identifier comes from the versioned model files # (trusted); every VALUE is parameterized — the small model only ever picks keys, so there is no # injection surface. Live XML-RPC stays the validate() path (the store never validates itself). def _store_con(): import harness.datastore as DS if not DS.ready(): # a mid-backfill store returns PARTIAL totals silently raise ModelError("the data cache is still warming up after a restart (a one-time first " "sync) — the dashboards work now from live data; ask me again in a " "moment and I'll have it") return DS.ro_cursor() # cursor on the shared instance — no sync-writer contention def _measure_sql(m, alias): """SQL for a base measure. Supports FILTERED measures (`store_filter_sql` — the Omni filtered-measure concept: sum(CASE WHEN … )) and `negate: true` (e.g. GL income, stored credit-negative, reported positive).""" agg, f = m.get("agg"), m.get("field") flt = render_scope(m["store_filter_sql"]) if m.get("store_filter_sql") else None if agg == "sum": core = f"sum(CASE WHEN {flt} THEN {alias}.{f} ELSE 0 END)" if flt else f"sum({alias}.{f})" elif agg == "count_distinct": core = (f"count(DISTINCT CASE WHEN {flt} THEN {alias}.{f} END)" if flt else f"count(DISTINCT {alias}.{f})") elif agg == "count": core = f"count(CASE WHEN {flt} THEN 1 END)" if flt else "count(*)" else: raise ModelError(f"metric {m['key']!r}: agg {agg!r} has no store expression") return f"-({core})" if m.get("negate") else core def _expand_measures(keys): """Expand ratio/derived metrics into the base measures SQL must aggregate, keeping the requested keys for post-compute. Returns (base_keys, requested_keys).""" mts = _model()["metrics"] base, seen = [], set() def need(k): if k in seen: return seen.add(k) m = mts.get(k) if not m: raise ModelError(f"unknown metric {k!r}") if m["agg"] == "ratio": need(m["numerator"]); need(m["denominator"]) elif m["agg"] == "derived": import ast as _ast for node in _ast.walk(_ast.parse(m["expr"], mode="eval")): if isinstance(node, _ast.Name): need(node.id) else: if k not in base: base.append(k) for k in keys: need(k) return base, list(keys) def _post_compute(row, key): m = _model()["metrics"][key] if m["agg"] == "ratio": num, den = _post_compute(row, m["numerator"]), _post_compute(row, m["denominator"]) return (num / den) if den else 0.0 if m["agg"] == "derived": return _eval_expr(m["expr"], lambda k: _post_compute(row, k)) return row.get(key) or 0.0 # A metric's declared `format` -> the grid field type the compiler and the cells both read. # One mapping, so a new metric cannot arrive as an untyped column. _FORMAT_TYPE = {"usd": "currency", "int": "int", "pct": "pct"} def _clean_tree(tree, cols): """Run a filter tree through the SAME validator the client's view state goes through. `filter_sql`'s documented input contract is "feed me CLEANED trees" — it is faithful to the TS engine rather than fail-closed on garbage, because rejecting garbage is the validator's job. The UI path always cleans (view state is sanitised on the way into the store), but store_query/store_rows are callable directly, and an uncleaned tree is where the two diverge: `{"value": null}` is ACTIVE in TS (and throws on a text field) and INACTIVE here. Normalising at the entry point makes that unreachable rather than merely unlikely, and brings the depth / width / node-count caps to the server side too. `aios_grid` is a leaf module with zero imports of its own, so this cannot cycle. """ from aios_grid import (clean_filter_tree, COHORT_FIELD as _COHORT_FIELD, MAX_FILTER_DEPTH, MAX_FILTER_NODES, MAX_FILTER_SIBLINGS) # REFUSE an over-large tree rather than TRUNCATE it. clean_filter_tree caps siblings per # level, total nodes and depth by DROPPING the excess — correct on the client, where the # cap is a DoS guard on a per-row render loop and a dropped condition is the lesser evil. # On the server it is not: dropping AND-conditions WIDENS the result, and store_rows then # reports a row_count and scope-wide totals that look authoritative for a query the caller # never asked for. That is precisely the silent [:N] that [[no-unverifiable-aggregates]] # forbids. A caller that exceeds the caps gets an error, not a quietly different answer. total = 0 def _walk(nodes, depth): nonlocal total nodes = list(nodes or []) if len(nodes) > MAX_FILTER_SIBLINGS: raise ModelError(f"filter tree has {len(nodes)} conditions at one level; the limit " f"is {MAX_FILTER_SIBLINGS}. Refusing rather than dropping the " f"excess, which would silently widen the result.") for node in nodes: if not isinstance(node, dict): continue total += 1 if isinstance(node.get('children'), list): if depth >= MAX_FILTER_DEPTH: raise ModelError(f"filter tree nests deeper than {MAX_FILTER_DEPTH} levels; " f"refusing rather than dropping the deepest group.") _walk(node['children'], depth + 1) _walk(tree, 1) if total > MAX_FILTER_NODES: raise ModelError(f"filter tree has {total} nodes; the limit is {MAX_FILTER_NODES}. " f"Refusing rather than truncating, which would silently widen it.") # CG-8. A MEASURE condition ("Sales in the last 90 days > 5,000") is answered by # harness/measure_filter.py, which resolves it to a SET OF CUSTOMER IDS. It has no meaning # on this path, and both ways it could arrive here are silent: # - on a topic without that column, `clean_filter_tree` DROPS it as unknown — widening the # query while store_rows reports an authoritative row_count for something nobody asked # for (the same class as the sibling/depth caps above); # - on sales_lines at row grain, `revenue` IS a real column, so it would compile as a # per-LINE predicate with the window silently discarded — a different question answered # confidently, which is worse than dropping it. # The discriminator is the `window`, not the column name: no column condition has one. windowed = [] cohorts = [] def _find(nodes): for node in nodes or []: if not isinstance(node, dict): continue if isinstance(node.get('children'), list): _find(node['children']) elif node.get('window') is not None: windowed.append(str(node.get('colId'))) elif node.get('colId') == _COHORT_FIELD: cohorts.append(str(node.get('value'))) _find(tree) # Owner item 5, and the same refusal for the same reason. A cohort leaf names a # hand-curated set of CUSTOMERS held per user in the tenant store — this path knows nothing # about it, and `clean_filter_tree` below would drop it as an unknown key, widening the # query while store_rows still reported an authoritative row_count. If a caller ever needs # it here, the fix is to pass the membership in, not to let it fall through. if cohorts: raise ModelError( f"filter tree carries cohort-membership condition(s) {sorted(set(cohorts))} — " f"cohort membership is host state, not a column of this topic. Refusing rather than " f"dropping it, which would silently widen the result under an authoritative count.") if windowed: raise ModelError( f"filter tree carries measure condition(s) over a date window {sorted(set(windowed))}" f" — those are resolved to a set of ids by harness.measure_filter, not compiled into " f"this query. Refusing rather than dropping or mis-compiling them, either of which " f"would silently answer a different question under an authoritative count.") return clean_filter_tree(tree, set(cols)) def _measure_row_sql(m, alias): """A sum-metric's per-ROW value: the same expression `_measure_sql` aggregates, without the aggregate wrapper. At line grain `revenue` IS `l.price_subtotal` for that line — deriving it from the metric rather than hand-writing the column is what keeps ONE definition of the number (the drift the semantic layer exists to prevent). Only `sum` has a row value; a count/count_distinct metric is a property of a SET, not of a row.""" if m.get("agg") != "sum": return None f = m.get("field") flt = render_scope(m["store_filter_sql"]) if m.get("store_filter_sql") else None core = f"CASE WHEN {flt} THEN {alias}.{f} ELSE 0 END" if flt else f"{alias}.{f}" return f"-({core})" if m.get("negate") else core def store_columns(topic, include_measures=True, grain="aggregate"): """The `{colId: {sql, type, aggregate}}` spec `harness.filter_sql` compiles against (CG-1). `grain="row"` is the LINE-grain projection (CG-2): sum-metrics resolve to their own raw field with `aggregate=False`, so a line table filters and sorts on real row values in WHERE rather than needing HAVING. count-style metrics are dropped — they have no row value. This is the seam between the grid's field contract and the semantic model: the grid speaks field KEYS, the compiler needs SQL expressions and a field TYPE, and only the model may say what a key resolves to. Nothing about a topic leaks into filter_sql itself. -> the dim's NAME column (what the grid displays and what a user means when they type "contains floral"), filtered PRE-aggregation in WHERE. Correct even in a grouped query: every row of a group shares its group's name. _id -> the raw id, for exact machine-keyed filtering. date -> the topic's date column, so a grid can filter/sort it like any other field. -> the aggregate expression, marked `aggregate` so the caller routes it to HAVING rather than WHERE (see store_query — that path is not built yet). """ t = _model()["topics"].get(topic) if not t or "store" not in t: raise ModelError(f"topic {topic!r} has no store binding") s = t["store"] cols = {} for key, d in (s.get("dims") or {}).items(): cols[key] = {"sql": d.get("name_col") or d["col"], "type": "text", "aggregate": False} # Only expose a separate id column when the dim actually HAS one. `city` is its own # name (col == name_col), so a `city_id` would be a text column typed int — filtering # it numerically would quietly compare 0 against 0 for every row. if d.get("name_col") and d["name_col"] != d["col"]: cols[f"{key}_id"] = {"sql": d["col"], "type": "int", "aggregate": False} if s.get("date_col"): cols["date"] = {"sql": s["date_col"], "type": "date", "aggregate": False} if include_measures: for k, m in _model()["metrics"].items(): if m.get("topic") != topic or m.get("agg") in ("ratio", "derived"): continue # ratio/derived are post-computed, not SQL # The field TYPE comes from the metric's declared format, never a guess: `units` is # format:int, and typing it currency would render 791,960 units as dollars. ftype = _FORMAT_TYPE.get(m.get("format"), "currency") if grain == "row": row_sql = _measure_row_sql(m, s["alias"]) if row_sql: cols[k] = {"sql": row_sql, "type": ftype, "aggregate": False} else: cols[k] = {"sql": _measure_sql(m, s["alias"]), "type": ftype, "aggregate": True} return cols #: The ceiling for a GROUPED `store_query`. One row per group, so this bounds DIMENSION #: CARDINALITY, not payload — a tenant would need 200,000 distinct customers (or products, or #: cities) to reach it. Deliberately NOT unbounded: a runaway group-by should fail, not swap. MAX_GROUPS = 200_000 def store_query(topic, measures, group_by=None, grain=None, date_from=None, date_to=None, team_id=None, filters=None, sort=None, limit=1000, exclude_services=False, filter_tree=None, filter_conj="and", today=None): """The semantic query over the tenant store — the engine behind the Analyst's run_semantic_query tool and the saved-view re-runner. All keys whitelisted against the model.""" t = _model()["topics"].get(topic) if not t or "store" not in t: raise ModelError(f"topic {topic!r} has no store binding") s = t["store"] alias = s["alias"] dims = s.get("dims") or {} # ⭐ A GROUPED QUERY IS BOUNDED BY GROUPS, NOT BY ROWS — and the 5,000 row ceiling applied to # both, which made it a SILENT TRUNCATION of the answer rather than of a payload. # # ⛔ MEASURED 2026-08-09 on the Royal mirror, grouping 256,810 order lines by customer: # limit=100 -> 100 groups, total 1,644,181.65 # limit=1000 -> 1000 groups, total 9,042,614.10 # limit=5000 -> 1748 groups, total 14,567,929.72 # The TOTAL MOVES WITH THE CAP. A row window is honest because counts and totals are computed # over the full scope beside it (`store_rows`' whole design); a GROUP window has no such # companion — the groups ARE the answer, so dropping one is dropping data with nothing to # notice. Royal has 1,943 customers so it passes today and would have passed every test, # then gone quietly wrong for the first tenant with more ([[no-unverifiable-aggregates]]). # # ⚠ The ceiling is not removed, because unbounded is its own failure. It is raised to a bound # no realistic dimension reaches, and — the load-bearing half — truncation is now a FACT the # caller can read rather than something it must infer from `len(rows)`. grouped = bool(group_by) limit = max(1, min(int(limit or 1000), MAX_GROUPS if grouped else 5000)) base_keys, requested = _expand_measures(list(measures or [])) if not base_keys: raise ModelError("at least one measure required") # Cross-topic base metrics (e.g. aov = revenue ÷ orders, where orders lives on sales_orders): # in a SCALAR query they resolve via a nested scalar query on their HOME topic (correct grain — # never count(*) on the wrong table); in a GROUPED query they are refused (split the request). foreign = [k for k in base_keys if _model()["metrics"][k]["topic"] != topic] if foreign and (group_by or grain): raise ModelError(f"metrics {foreign} belong to another topic — cross-topic measures are " f"scalar-only; run them against their own topic when grouping") base_keys = [k for k in base_keys if k not in foreign] foreign_vals = {} for k in foreign: sub = store_query(_model()["metrics"][k]["topic"], [k], date_from=date_from, date_to=date_to, team_id=team_id) foreign_vals[k] = sub["rows"][0][k] if sub["rows"] else 0 if not base_keys: # purely foreign scalar request return {"topic": topic, "rows": [foreign_vals], "row_count": 1, "sql": "(nested)", "measures": requested, "group_by": [], "grain": None} select, group_cols, params = [], [], [] gb = [g for g in (group_by or []) if g] for g in gb: if g not in dims: raise ModelError(f"group_by {g!r} not a dim of {topic!r} (allowed: {list(dims)})") d = dims[g] select.append(f"{d['col']} AS {g}_id" if d.get("name_col") else f"{d['col']} AS {g}") group_cols.append(d["col"]) if d.get("name_col"): select.append(f"max({d['name_col']}) AS {g}") if grain: if grain not in ("month", "week", "day"): raise ModelError("grain must be month|week|day") select.insert(0, f"date_trunc('{grain}', CAST({s['date_col']} AS TIMESTAMP)) AS period") group_cols.insert(0, f"date_trunc('{grain}', CAST({s['date_col']} AS TIMESTAMP))") for k in base_keys: select.append(f"{_measure_sql(_model()['metrics'][k], alias)} AS {k}") where = [render_scope(s["scope_sql"].strip())] # TYPE-CORRECT date bounds (the 2025-07-01 lesson): columns hold ISO strings in TWO shapes — # datetimes ('YYYY-MM-DD HH:MM:SS', sale date_order) and bare dates ('YYYY-MM-DD', GL date). # Lexical string comparison EXCLUDES the window's first day for bare dates ('2025-07-01' < # '2025-07-01 00:00:00'), which silently dropped a whole day of GL ($49,467). Cast both sides. if date_from: where.append(f"CAST({s['date_col']} AS TIMESTAMP) >= ?") params.append(f"{date_from} 00:00:00") if date_to: where.append(f"CAST({s['date_col']} AS TIMESTAMP) <= ?") params.append(f"{date_to} 23:59:59") if team_id: if not s.get("team_col"): raise ModelError(f"topic {topic!r} is company-level — it has no business-unit filter") where.append(f"{s['team_col']} = ?"); params.append(int(team_id)) if exclude_services and s.get("service_filter"): where.append(render_scope(s["service_filter"])) for dim_key, vals in (filters or {}).items(): if dim_key not in dims: raise ModelError(f"filter dim {dim_key!r} not allowed (use: {list(dims)})") # id dims filter by int; TEXT dims (e.g. city) filter by the value itself — either way # every VALUE stays parameterized (the whitelist covers identifiers only) def _coerce(v): try: return int(v) except (TypeError, ValueError): return str(v) ids = [_coerce(v) for v in (vals if isinstance(vals, (list, tuple)) else [vals])] where.append(f"{dims[dim_key]['col']} IN ({','.join('?' for _ in ids)})") params.extend(ids) # The grid's filter TREE (CG-1). `filters` above stays the Analyst's flat dim=IN shape; # this is the nested, 11-operator contract the table UI emits. Compiled by # harness/filter_sql, which is held in lock-step with the TS engine and the validator by # aios-web/verify_filter_engine.py. Appended AFTER the dim filters so params stay in # positional order with `where`. if filter_tree: from harness import filter_sql as _fs cols = store_columns(topic) pred = _fs.compile_filter_tree(_clean_tree(filter_tree, cols), filter_conj, cols, today=today) if pred is not None: if pred.uses_aggregate: bad = sorted(c for c in pred.columns_used if cols[c].get("aggregate")) raise ModelError( f"filter_tree references measure(s) {bad} — a measure filter has to land in " f"HAVING, and that path is deliberately NOT built: CG-1 ships the WHERE path " f"because its consumer (CG-2, the line-grain sales table) does not aggregate. " f"Filter on dims, or aggregate first and filter the result.") where.append(pred.sql) params.extend(pred.params) sql = (f"SELECT {', '.join(select)} FROM {s['table']} {alias} {s.get('join','')} " f"WHERE {' AND '.join(where)}") if group_cols: sql += f" GROUP BY {', '.join(group_cols)}" if sort: key = sort.lstrip("-") if key not in base_keys + gb + (["period"] if grain else []): raise ModelError(f"sort {sort!r} must reference a selected measure/dim") sql += f" ORDER BY {key} {'DESC' if sort.startswith('-') else 'ASC'}" elif grain: sql += " ORDER BY period" # ⚠ `limit + 1` — ONE extra row, so "did this truncate" is a FACT and not the guess # `len(rows) == limit` makes (which is wrong exactly when the count lands on the cap). sql += f" LIMIT {limit + 1}" con = _store_con() try: cur = con.execute(sql, params) cols = [d[0] for d in cur.description] rows = [dict(zip(cols, r)) for r in cur.fetchall()] finally: con.close() for row in rows: # ratio/derived post-compute per group row.update(foreign_vals) # scalar cross-topic components for k in requested: if _model()["metrics"][k]["agg"] in ("ratio", "derived"): row[k] = _post_compute(row, k) if "period" in row and row["period"] is not None: row["period"] = str(row["period"])[:10] truncated = len(rows) > limit if truncated: rows = rows[:limit] return {"topic": topic, "rows": rows, "row_count": len(rows), "sql": sql, "measures": requested, "group_by": gb, "grain": grain, # ⛔ A CALLER THAT AGGREGATES THESE ROWS MUST CHECK THIS. For a grouped query the # groups ARE the answer, so a truncated result is a WRONG NUMBER, not a short list. "truncated": truncated} def store_rows(topic, date_from=None, date_to=None, team_id=None, filter_tree=None, filter_conj="and", search=None, sorts=None, limit=200, offset=0, aggregates=None, exclude_services=False, id_sql=None, member_ids=None, today=None): """ROW-grain fetch — the line-grain counterpart to store_query's aggregate path (CG-2). store_query answers "what is the number"; this answers "which rows". At line grain the payload is the binding constraint (254,837 sale_order_lines against 1,550 customers — the whole book would be ~96 MB at the customer table's measured 377 B/row), so rows come back WINDOWED. A window is only honest if the counts and totals are not windowed with it ([[no-unverifiable-aggregates]]). So this returns THREE independent numbers, each from its own query over the FULL scope, never from `len(rows)`: row_count rows matching the filter across the whole scope -> the N of "N of M" total_count rows in the scope with no filter at all -> the M aggregates the topic's OWN metric definitions, summed over the whole FILTERED scope That is the difference between a windowed fetch and a silent `[:N]` cap: the user is told how many rows exist and shown totals for all of them, while receiving only the page they can see. Filters/sorts/search compile through harness.filter_sql, so a line table means EXACTLY what the client engine means at customer grain — that equivalence is what verify_filter_engine.py exists to hold. Dates are returned already truncated to the 10-char ISO shape the filter compares against, so what is displayed and what is filtered are the same string. Numerics come back RAW; the payload layer rounds them (aios_grid._round), and the compiler mirrors that rounding. """ t = _model()["topics"].get(topic) if not t or "store" not in t: raise ModelError(f"topic {topic!r} has no store binding") from harness import filter_sql as _fs s = t["store"] alias = s["alias"] cols = store_columns(topic, grain="row") id_sql = id_sql or f"{alias}.id" limit = max(1, min(int(limit or 200), 5000)) offset = max(0, int(offset or 0)) # --- the scope every one of the three queries shares ------------------------------- scope, scope_params = [render_scope(s["scope_sql"].strip())], [] if date_from: scope.append(f"CAST({s['date_col']} AS TIMESTAMP) >= ?") scope_params.append(f"{date_from} 00:00:00") if date_to: scope.append(f"CAST({s['date_col']} AS TIMESTAMP) <= ?") scope_params.append(f"{date_to} 23:59:59") if team_id: if not s.get("team_col"): raise ModelError(f"topic {topic!r} is company-level — it has no business-unit filter") scope.append(f"{s['team_col']} = ?") scope_params.append(int(team_id)) if exclude_services and s.get("service_filter"): scope.append(render_scope(s["service_filter"])) # --- the user's narrowing, on top of the scope -------------------------------------- narrow, narrow_params = [], [] pred = _fs.compile_filter_tree(_clean_tree(filter_tree or [], cols), filter_conj, cols, today=today, member_ids=member_ids or None, id_sql=id_sql) if pred is not None: if pred.uses_aggregate: # cannot happen at grain="row"; guard anyway raise ModelError(f"row-grain filter referenced an aggregate: {sorted(pred.columns_used)}") narrow.append(pred.sql) narrow_params.extend(pred.params) spred = _fs.compile_search(search, cols) if spred is not None: narrow.append(spred.sql) narrow_params.extend(spred.params) frm = f"{s['table']} {alias} {s.get('join', '')}" scope_where = " AND ".join(scope) all_where = " AND ".join(scope + narrow) con = _store_con() try: # 1. the WINDOW of rows select = [f"{id_sql} AS _rid"] + [ (f"SUBSTR(CAST({c['sql']} AS VARCHAR), 1, 10) AS {k}" if c["type"] == "date" else f"{c['sql']} AS {k}") for k, c in cols.items()] order = _fs.compile_order_by(sorts, cols, tiebreak_sql=id_sql) or f"{id_sql} ASC" sql = (f"SELECT {', '.join(select)} FROM {frm} WHERE {all_where} " f"ORDER BY {order} LIMIT {limit} OFFSET {offset}") cur = con.execute(sql, scope_params + narrow_params) names = [d[0] for d in cur.description] rows = [dict(zip(names, r)) for r in cur.fetchall()] # 2. the two counts — over the WHOLE scope, never len(rows) count_sql = f"SELECT count(*) FROM {frm} WHERE {all_where}" row_count = con.execute(count_sql, scope_params + narrow_params).fetchone()[0] total_count = con.execute(f"SELECT count(*) FROM {frm} WHERE {scope_where}", scope_params).fetchone()[0] # 3. the aggregates — the topic's OWN metric definitions over the whole FILTERED scope mts = _model()["metrics"] keys = [k for k in (aggregates if aggregates is not None else [k for k, m in mts.items() if m.get("topic") == topic and m.get("agg") == "sum"])] aggs = {} if keys: for k in keys: if k not in mts or mts[k].get("topic") != topic: raise ModelError(f"aggregate {k!r} is not a metric of topic {topic!r}") agg_sql = ("SELECT " + ", ".join(f"{_measure_sql(mts[k], alias)} AS {k}" for k in keys) + f" FROM {frm} WHERE {all_where}") cur = con.execute(agg_sql, scope_params + narrow_params) aggs = dict(zip([d[0] for d in cur.description], cur.fetchone())) finally: con.close() return { "topic": topic, "rows": rows, "columns": {k: {"type": c["type"]} for k, c in cols.items()}, "window": {"offset": offset, "limit": limit, "returned": len(rows)}, "row_count": row_count, # N — matches the filter, across the WHOLE scope "total_count": total_count, # M — the unfiltered scope "aggregates": aggs, # over the whole FILTERED scope, not the window "sql": sql, "count_sql": count_sql, } def store_field_values(topic, dim, search=None, limit=25): """Resolve real filterable values for a dim (the Analyst's get_field_values tool — fixes 'NYC Florist' vs the actual record name BEFORE querying).""" t = _model()["topics"].get(topic) if not t or "store" not in t: raise ModelError(f"topic {topic!r} has no store binding") d = (t["store"].get("dims") or {}).get(dim) if not d: raise ModelError(f"unknown dim {dim!r} for topic {topic!r}") if not d.get("name_col"): raise ModelError(f"dim {dim!r} has no name column (filter by id)") s = t["store"] sql = (f"SELECT DISTINCT {d['col']} AS id, {d['name_col']} AS name " f"FROM {s['table']} {s['alias']} {s.get('join','')} " f"WHERE {render_scope(s['scope_sql'].strip())}") params = [] if search: sql += f" AND {d['name_col']} ILIKE ?" params.append(f"%{search}%") sql += f" ORDER BY name LIMIT {max(1, min(int(limit), 100))}" con = _store_con() try: return [{"id": r[0], "name": r[1]} for r in con.execute(sql, params).fetchall()] finally: con.close() def store_parity(date_from=None, date_to=None): """THE OM-2 gate: the store path must equal the live path to the cent, per metric per scope.""" import core.periods as P if not date_from: date_from, date_to = P.ytd() out = [] for key in ("revenue", "margin", "units", "orders", "customers", "returns", "invoiced", "revenue_invoiced", "orders_invoiced"): # wave 21 R2 — both filtered paths m = _model()["metrics"][key] # ⚠ A COMPANY-LEVEL topic is checked at company scope ONLY. `store_query` RAISES when a # team is asked of a topic with no team_col, and the live builder refuses too — so # looping the BUs here would fail the gate for a topic that is correct. Parity for a # company-level number means company-level parity; pretending otherwise would either # crash or, worse, compare two numbers that silently ignored the team. _store = (_model()["topics"].get(m["topic"]) or {}).get("store") or {} scopes = (None, *O.TEAM_IDS) if _store.get("team_col") else (None,) for tid in scopes: live = metric(key, date_from, date_to, tid) res = store_query(m["topic"], [key], date_from=date_from, date_to=date_to, team_id=tid) ours = res["rows"][0][key] if res["rows"] else 0 gap = abs((ours or 0) - (live or 0)) out.append({"check": f"store.{key} == live.{key} (team={tid})", "ok": gap <= 0.01, "gap": round(gap, 4), "store": ours, "live": live}) # The agent dim (2026-07-17): store agent-filtered revenue must equal live revenue over the # agent's FIRST-agent book — a fully independent path (live partner read + line-domain sum) # against the store's res_partner.agent_id attribution. Largest book = the sharpest check. recs = O.search_read("res.partner", [("active", "in", [True, False]), ("agent_ids", "!=", False)], ["agent_ids"], limit=100000) books = {} for r in recs: a = (r.get("agent_ids") or [None])[0] if a: books.setdefault(a, []).append(r["id"]) if books: top = max(books, key=lambda k: len(books[k])) live = O.sum_field("sale.order.line", O.sale_line_domain(date_from, date_to, partner_ids=books[top]), "price_subtotal") res = store_query("sales_lines", ["revenue"], date_from=date_from, date_to=date_to, filters={"agent": [top]}) ours = res["rows"][0]["revenue"] if res["rows"] else 0 gap = abs((ours or 0) - (live or 0)) out.append({"check": f"store.revenue[agent={top}] == live book sum (first-agent, " f"{len(books[top])} customers)", "ok": gap <= 0.01, "gap": round(gap, 4), "store": ours, "live": live}) return out