platform/harness/semantic.py

"""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.

      <dim>      -> 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.
      <dim>_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.
      <metric>   -> 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