api/odoo_relational.py

"""odoo_relational.py — Odoo entities as LOCKED relational databases.

Owner ruling R1 / contract C8: spawn preset Odoo databases for Royal Imports, give them preset
Link + Rollup fields, and prove each rollup against the `measure_` column it will eventually
replace.

⭐⭐ 2026-08-09 — THE POPULATIONS WIDENED FROM "OPEN AR" TO **EVERY ODOO ID**, which is the
owner's item: *"make sure we have all the Unique ID in Odoo in Database for the Royal Imports
tenant."* Before this change there were two tables holding 438 partners and 1,228 invoices — the
partners who owed money — so most Odoo ids were simply absent, and the missing rows were the
reason a Rollup could not answer a sales question. Four tables now, keyed on the Odoo id itself:

    ut_odoo_customers   2,465 rows   0.43 MB     every partner with a confirmed order or a
                                                 posted customer document
    ut_odoo_products    5,829 rows   1.20 MB     every active product carrying a SKU code
    ut_odoo_invoices   31,418 rows   8.68 MB     EVERY posted customer invoice + refund
    ut_odoo_orders     32,700 rows   7.50 MB     every confirmed sale order

⛔ WHAT MADE THAT LEGAL, AND IT WAS NOT A BIGGER NUMBER. `MAX_ROWS` was 5,000 and this module's
own `plan()` refused above it — but the cap was never a property of the store (see the measured
banner on `core.user_tables.MAX_ROWS`; `ig_master` has run a 500,000-row bucket the whole time).
The cap is now 60,000, DERIVED from what a row actually weighs (⚠ this line said 100,000 until
wave 28 — that was the FIRST candidate and its own derivation REJECTED it for clearing the memory
budget by 0.6%; the prose was written before the number lost, and two sibling files said it too).
The four tables together are
17.81 MB in one `user_tables` document — real, bounded, and booked: the per-table row-key split
is D-87's next increment. ⛔ ORDER LINES REMAIN OUT (256,810 rows / 63.9 MB / 2.57 s per copy);
they are answered by the read-through rollup, which never copies a row.

⭐ THE EXCLUDED CHANNEL IS NOW A COLUMN, NOT A DELETION. `core.odoo.EXCLUDE_PARTNER_NAMES` puts
GIFTWARE DEALS (partner 6369 — the Amazon channel) outside WHOLESALE scope, and the old tables
dropped its rows entirely. Dropping them contradicts "every Odoo id", so the rows are kept and
carry **`wholesale_scope`** instead. ⚠⚠ READ THIS BEFORE COMPARING ANY TOTAL: that one partner
holds **$1,755,779.95 of the $2,347,608.49** raw open balance — 75% of it — across 25 invoices.
Wholesale open AR is $591,828.54. So a column total here will not equal the AR page unless you
filter `wholesale_scope`, and that is the scope difference, not a defect. `read_open_ar` keeps
excluding, because `modules/ar` is its oracle and an oracle answers ONE question.

⭐ AR SURVIVED THE WIDENING UNCHANGED, AND THAT IS MEASURED, NOT ASSUMED. `sum(residual)` over
ALL posted customer documents equals `sum(residual)` over `modules/ar._open_docs`' own predicate
**to the cent** ($2,347,608.49): zero posted rows carry a non-zero residual outside
`payment_state IN ('not_paid','partial')`, and zero rows inside it carry a residual of 0. So the
`ar_outstanding` rollup needs no condition. ⛔ THE COUNT AND THE DATE DO — `countall` over the
wider link would count 31,418 documents and call them open invoices, so those two rollups carry
the oracle's predicate as an explicit `payment_state` condition pair.
"""
import datetime as _dt

#: Royal Imports only (R1). A tenant slug that is not this one gets a refusal, never a spawn:
#: nurilab has no Odoo mirror behind these tables and would get empty locked databases.
RI_SLUGS = ("", "royal-imports")

INVOICES_KEY = "ut_odoo_invoices"
CUSTOMERS_KEY = "ut_odoo_customers"
ORDERS_KEY = "ut_odoo_orders"
PRODUCTS_KEY = "ut_odoo_products"
#: ⭐ WAVE 28 (owner R1): *"ALL of Unique ID in Odoo is a database e.g. Customers/Products/Agents,
#: etc. Including expenses and GL codes."* Four more DOCUMENT/REGISTRY grains, each measured to
#: fit far inside `MAX_ROWS` (19 / 192 / 6,538 / 393 against 60,000).
AGENTS_KEY = "ut_odoo_agents"
ACCOUNTS_KEY = "ut_odoo_accounts"
BILLS_KEY = "ut_odoo_bills"
VENDORS_KEY = "ut_odoo_vendors"

#: ⭐⭐ WAVE 30 / R7 / W30-T35 — THE TWO LINE GRAINS, AND THEY ARRIVE THE ONLY WAY THEY EVER COULD.
#:
#: ⚠ THE PARAGRAPH THAT STOOD HERE SAID THESE WERE "DELIBERATELY NOT HERE … at any cap", and it
#: was RIGHT ABOUT THE CAP AND WRONG ABOUT THE CONCLUSION — which is exactly why it is replaced
#: rather than left standing beside its own contradiction. The obstacle was never the number of
#: rows; it was that every row had to be COPIED into the shared `user_tables` document. MEASURED
#: on this box's mirror 2026-08-12: 254,189 order lines in the confirmed scope (256,810 unscoped)
#: and 963,783 GL lines — 4.2x and 16x `MAX_ROWS`, 63.9 MB and ~240 MB as JSON. Owner ruling R6
#: settles what that means: *"there is no cap in how many data from the API source … can be pulled
#: into the app"*, so the answer is a different residency, never a bigger ceiling.
#:
#: ⛔ THESE TWO TABLES STORE NO ROWS HERE AND NEVER WILL. `routes_odoo_tables` binds them to the
#: DuckDB mirror (`GRID_SOURCES`) and `core.user_tables.row_limit` answers **0** for them — "this
#: database stores no rows HERE", which is a different statement from `None` ("connected and
#: uncapped") and from `MAX_ROWS` ("the editable substrate"). `plan()` below reads that evaluator
#: and builds no python row for either grain: 963,783 dicts in one process is the dangerous work
#: the answer exists to prevent. What DOES get written is the DEFINITION — a locked database with
#: fields, a label, grants and a nav entry, and zero rows. A definition with no rows is a working
#: grid; that is the whole shape of the conversion.
ORDER_LINES_KEY = "ut_odoo_order_lines"
GL_LINES_KEY = "ut_odoo_gl_lines"

#: The join column the partner-grain tables carry. Derived links resolve through it (`on`/`from`).
JOIN_KEY = "partner_id"
#: The product-grain equivalent.
PRODUCT_JOIN_KEY = "product_id"
#: ⚠ AN AGENT IS A `res.partner`, so its id shares the partner namespace with a customer's — but
#: it is a DIFFERENT COLUMN on the customer row (`agent_id`, the customer's assigned agent) and the
#: two must never be joined through `JOIN_KEY`, which would link every customer to itself.
AGENT_JOIN_KEY = "agent_id"
#: A vendor is also a `res.partner`; same reasoning, its own column.
VENDOR_JOIN_KEY = "vendor_id"
ACCOUNT_JOIN_KEY = "account_code"

#: The oracle's own predicate — `modules/ar._open_docs`, copied rather than re-derived so the two
#: cannot drift. It now selects a SUBSET of the invoices table rather than defining it.
_AR_OPEN = "payment_state IN ('not_paid','partial')"
_POSTED_DOCS = "state = 'posted' AND move_type IN ('out_invoice','out_refund')"
_AR_WHERE = f"{_POSTED_DOCS} AND {_AR_OPEN}"
_CONFIRMED = "state IN ('sale','done')"

#: A refresh that would delete more than this share of a table's stored rows REFUSES instead.
#: ⛔ THE GUARD ONLY BECAME NECESSARY WHEN THE TABLES GOT BIG. `_ensure_table` removes rows that
#: left the population, which is right — a reversed invoice must not keep inflating a total. But
#: the population comes from the DuckDB mirror, and a mirror caught mid-resync (or one seeded
#: against an empty store) answers with FEWER rows and no error. At 1,228 rows that was a visible
#: mistake; at 31,418 it is a silent one. Odoo history does not halve, so a halving is a bad read.
MAX_SHRINK = 0.5


def _ut():
    import core.user_tables as user_tables
    return user_tables


def _iso_today():
    return _dt.date.today().strftime("%Y-%m-%d")


def _preset(field, flow="odoo_relational"):
    """Stamp a field as machine-owned + preset — the `ut_ensure` lock_fields convention, so the
    grid renders it grey and the preset walls refuse a rename or a delete."""
    field = dict(field)
    field["automation"] = {"flowId": flow, "preset": True}
    return field


#: The two conditions that reproduce `modules/ar`'s open-document predicate inside a rollup.
#: ⚠ Two `eq` legs joined by OR, not one `in` — `ROLLUP_CONDITION_OPS` has no `in`, and inventing
#: one here would be a second condition vocabulary beside `_clean_rollup`'s.
_OPEN_ONLY = {"conditions": [{"field": "payment_state", "op": "eq", "value": "not_paid"},
                             {"field": "payment_state", "op": "eq", "value": "partial"}],
              "conditionConj": "or"}


# ---------------------------------------------------------------------------------------------
# FIELD CONTRACTS
# ---------------------------------------------------------------------------------------------
# ⚠ Every type here must be in `core.user_tables.UT_FIELD_TYPES`, and `_clean_field` returns None
# for an unknown one — which DELETES the column silently on the next read rather than erroring.
def _scope_field():
    return {"key": "wholesale_scope", "label": "In wholesale scope", "type": "checkbox",
            "source": "overlay", "default": False,
            "description": "Unticked = the GIFTWARE DEALS / Amazon channel, which every wholesale "
                           "metric in this product excludes. The row is kept so no Odoo id is "
                           "missing; filter on this column to reconcile against the AR page."}


def _refreshed_field():
    return {"key": "refreshed", "label": "Refreshed", "type": "date", "source": "overlay",
            "default": False, "description": "When this row was last reconciled against Odoo."}


def agent_fields():
    """One row per SALES AGENT, keyed on the `res.partner` id.

    ⭐ THE POPULATION IS A UNION OF TWO DISAGREEING SOURCES, and the disagreement is the reason it
    is a union rather than a pick. MEASURED 2026-08-09: 16 partners carry commission lines, 17
    carry `res_partner.agent = TRUE`, and the union is 19 — so **2 agents earn commission without
    the flag and 3 are flagged with no commission yet**. Either source alone silently drops real
    agents. Same shape as `read_customers`' two document universes, for the same reason.
    """
    return [_preset(f) for f in (
        {"key": "agent", "label": "Agent", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `res.partner` id. Also this row's id."},
        {"key": AGENT_JOIN_KEY, "label": "Odoo agent id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "flagged", "label": "Flagged in Odoo", "type": "checkbox", "source": "overlay",
         "default": True,
         "description": "Ticked = `res.partner.agent` is set. Unticked agents were found by "
                        "their commission lines instead - both are real, which is why this "
                        "table is the union of the two."},
        {"key": "commissioned", "label": "Has commission lines", "type": "checkbox",
         "source": "overlay", "default": True},
        # ⭐ THE INVERSE HALF: the customers whose `agent_id` names this agent. MEASURED: 2,093
        # customers carry one and ALL 2,093 resolve to a row in this table (zero dangling).
        {"key": "customers", "label": "Customers", "type": "link", "source": "overlay",
         "default": True,
         "link": {"table": CUSTOMERS_KEY, "on": AGENT_JOIN_KEY, "from": AGENT_JOIN_KEY}},
        _refreshed_field(),
    )]


def account_fields():
    """One row per `account.account` — the GL chart, the owner's "GL codes".

    ⚠ NO LINK COLUMN, and that is a finding rather than an omission. A GL account meets the rest
    of this schema only at LINE grain (963,783 `account_move_line` rows, 154,917 of them on
    expense-type accounts), and a `ut_*` link folds rows that live in the store. The honest
    binding is a read-through rollup naming a governed topic, or the mirror grid (R2) — never a
    link into a table that does not exist. Declaring one here would render a permanently blank
    column, which is the exact trap D-87 warns about from the value site.
    """
    return [_preset(f) for f in (
        {"key": "account_code", "label": "Code", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": "account_name", "label": "Account", "type": "text", "source": "overlay",
         "default": True},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `account.account` id. Also this row's id."},
        # ⚠ 15 DISTINCT VALUES MEASURED IN THE MIRROR, all declared. A `select` storing a value its
        # options omit is wave-26 item 24: the filter panel answers with a list that cannot match
        # what is stored.
        {"key": "account_type", "label": "Type", "type": "select", "source": "overlay",
         "default": True,
         "options": ["expense", "expense_direct_cost", "expense_depreciation", "income",
                     "income_other", "asset_cash", "asset_current", "asset_receivable",
                     "asset_fixed", "asset_non_current", "asset_prepayments",
                     "liability_current", "liability_payable", "liability_credit_card",
                     "liability_non_current", "equity", "equity_unaffected", "off_balance"]},
        {"key": "is_expense", "label": "Expense account", "type": "checkbox", "source": "overlay",
         "default": True,
         "description": "Ticked for the expense family - the same predicate the semantic layer's "
                        "gl_lines topic uses, so this column and that topic cannot disagree."},
        _refreshed_field(),
    )]


def vendor_fields():
    """One row per partner we have POSTED a vendor bill to, keyed on the `res.partner` id.

    ⚠ A VENDOR IS NOT A CUSTOMER TABLE ROW, even though both are `res.partner`. MEASURED: 393
    vendors, of which only 9 also appear in the customer population. Pointing bills at
    `ut_odoo_customers` would have dangled 384 of 393 links — the failure would have been a mostly
    empty column, not an error.
    """
    return [_preset(f) for f in (
        {"key": "vendor", "label": "Vendor", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `res.partner` id. Also this row's id."},
        {"key": VENDOR_JOIN_KEY, "label": "Odoo vendor id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "country", "label": "Country", "type": "text", "source": "overlay",
         "default": True},
        {"key": "bills", "label": "Bills", "type": "link", "source": "overlay", "default": True,
         "link": {"table": BILLS_KEY, "on": VENDOR_JOIN_KEY, "from": VENDOR_JOIN_KEY}},
        _refreshed_field(),
    )]


def bill_fields():
    """One row per POSTED vendor bill or refund — the owner's "expenses", at DOCUMENT grain.

    ⚠ DOCUMENT GRAIN IS A CHOICE AND IT IS THE ONLY ONE THAT FITS: 6,538 bills against 154,917
    expense GL lines. What a person calls "expenses" is both, and they are different tables - the
    bill is what you pay, the line is what it was coded to. This is the payable; the line ledger
    is the read-through mirror grid (R2).
    """
    return [_preset(f) for f in (
        {"key": "bill_no", "label": "Bill", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `account.move` id. Also this row's id."},
        {"key": "vendor", "label": "Vendor", "type": "text", "source": "overlay", "default": True},
        {"key": VENDOR_JOIN_KEY, "label": "Odoo vendor id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "invoice_date", "label": "Bill date", "type": "date", "source": "overlay",
         "default": True},
        {"key": "due_date", "label": "Due date", "type": "date", "source": "overlay",
         "default": True},
        # ⚠ SIGNED, like the customer side: Odoo's `_signed` fields already carry the refund's
        # direction, so a refund reduces a total without anybody re-deriving a sign here.
        {"key": "amount_untaxed", "label": "Billed $", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        {"key": "residual", "label": "Outstanding $", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        {"key": "payment_state", "label": "Payment state", "type": "select", "source": "overlay",
         "default": True,
         "options": ["not_paid", "partial", "in_payment", "paid", "reversed"]},
        {"key": "move_type", "label": "Document", "type": "select", "source": "overlay",
         "default": False, "options": ["in_invoice", "in_refund"]},
        {"key": "vendor_link", "label": "Vendor record", "type": "link", "source": "overlay",
         "default": False,
         "link": {"table": VENDORS_KEY, "on": VENDOR_JOIN_KEY, "from": VENDOR_JOIN_KEY}},
        _refreshed_field(),
    )]


def invoice_fields():
    """One row per POSTED customer invoice or refund — the full history, not just what is open."""
    return [_preset(f) for f in (
        {"key": "invoice_no", "label": "Invoice", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `account.move` id. Also this row's id."},
        {"key": "customer", "label": "Customer", "type": "text", "source": "overlay",
         "default": True},
        {"key": JOIN_KEY, "label": "Odoo partner id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "invoice_date", "label": "Invoice date", "type": "date", "source": "overlay",
         "default": True},
        {"key": "due_date", "label": "Due date", "type": "date", "source": "overlay",
         "default": True},
        {"key": "residual", "label": "Outstanding $", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum",
         "description": "Odoo's signed residual. Exactly 0 on every settled document, which is "
                        "why AR rollups need no filter."},
        {"key": "amount_untaxed", "label": "Invoiced $", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        # ⛔ THE OPTION LIST WIDENED WITH THE POPULATION. It read ['not_paid','partial'] while the
        # table held open AR only; the full posted history also carries paid / in_payment /
        # reversed. A select holding a value its options do not declare is the wave-26 item-24
        # defect — the filter panel answers with a list that cannot match what is stored.
        {"key": "payment_state", "label": "Payment state", "type": "select", "source": "overlay",
         "default": True,
         "options": ["not_paid", "partial", "in_payment", "paid", "reversed"]},
        {"key": "move_type", "label": "Document", "type": "select", "source": "overlay",
         "default": False, "options": ["out_invoice", "out_refund"]},
        _scope_field(),
        # ⭐ THE RECIPROCAL HALF (owner item 2, 2026-08-09). DERIVED (`on` declared), exactly like
        # its twin, so the engine owns the cell and no human can edit a relation Odoo decided.
        {"key": "customer_link", "label": "Customer record", "type": "link", "source": "overlay",
         "default": False,
         "link": {"table": CUSTOMERS_KEY, "on": JOIN_KEY, "from": JOIN_KEY}},
        # ⭐ DEBT D-88, closed 2026-08-09. `invoice_origin` carries the ORDER NAME an invoice was
        # raised from, and the mirror did not sync it until this wave — so order->invoice was a
        # two-hop join through 963,783 `account_move_line` rows and wave 27 shipped no link at all.
        # ⚠ IT IS A NAME, NOT AN ID, and Odoo writes free text there (a manual invoice can hold
        # anything; a merged one can hold several origins space-separated). The link resolves
        # against `order_no` and finds nothing when the text is not an order name — the honest
        # outcome, and the reason this is a join HINT rather than a foreign key.
        {"key": "origin_order", "label": "Source order", "type": "text", "source": "overlay",
         "default": False,
         "description": "Odoo's `invoice_origin` - usually the order name, sometimes blank."},
        {"key": "order_link", "label": "Order record", "type": "link", "source": "overlay",
         "default": False,
         "link": {"table": ORDERS_KEY, "on": "order_no", "from": "origin_order"}},
        _refreshed_field(),
    )]


def order_fields():
    """One row per CONFIRMED sale order — `state in (sale, done)`, the fixed wholesale scope."""
    return [_preset(f) for f in (
        {"key": "order_no", "label": "Order", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `sale.order` id. Also this row's id."},
        {"key": "customer", "label": "Customer", "type": "text", "source": "overlay",
         "default": True},
        {"key": JOIN_KEY, "label": "Odoo partner id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "order_date", "label": "Order date", "type": "date", "source": "overlay",
         "default": True},
        {"key": "amount_untaxed", "label": "Order $", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        {"key": "team", "label": "Business unit", "type": "select", "source": "overlay",
         "default": True, "options": ["Fisch", "Royal", "Sales", "Giftware Deals"]},
        {"key": "state", "label": "State", "type": "select", "source": "overlay",
         "default": False, "options": ["sale", "done"]},
        {"key": "invoice_status", "label": "Invoice status", "type": "select", "source": "overlay",
         "default": True, "options": ["invoiced", "to invoice", "upselling", "no"]},
        _scope_field(),
        {"key": "customer_link", "label": "Customer record", "type": "link", "source": "overlay",
         "default": False,
         "link": {"table": CUSTOMERS_KEY, "on": JOIN_KEY, "from": JOIN_KEY}},
        # The reciprocal of the invoice's `order_link` (D-88): the invoices raised from THIS
        # order, matched on the order's own name.
        {"key": "invoices", "label": "Invoices", "type": "link", "source": "overlay",
         "default": True,
         "link": {"table": INVOICES_KEY, "on": "origin_order", "from": "order_no"}},
        _refreshed_field(),
    )]


# ═════════════════════════════════════════════════════════════════════════════════════════════
# THE TWO READ-THROUGH GRAINS (W30-T35). Their rows are SERVED FROM THE MIRROR, never stored.
# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⛔⛔ THE KEY SET IS HALF OF A CONTRACT AND `routes_odoo_tables.GRID_SOURCES[…]["cols"]` IS THE
# OTHER HALF. It binds a key to SQL; the label, type and order are declared HERE, once. The two
# lists must name the SAME columns, and both failure directions are silent:
#   * a bound key with no declaration -> a silent NO-CELL (the row projection is strict);
#   * a declared key with no binding  -> an INACTIVE filter leaf, which WIDENS the result set.
# `verify_scopes.section_line_grids` compares the two sets, which is why neither side may "just
# add a column".
#
# ⛔ NO LINK COLUMN AND NO `refreshed` STAMP ON EITHER TABLE, and both absences are findings
# rather than omissions:
#   * a LINK folds the rows the TARGET table STORES (`compute_relation_cells` reads the raw
#     document, not the mirror), so a link at a read-through grain resolves against nothing and
#     renders a permanently blank column — the trap `account_fields` names from the other end.
#     The join ids ride as ordinary filterable columns instead, so the relationships are all
#     still reachable by a person and by SQL.
#   * `refreshed` means "when this row was last reconciled against Odoo", and it is stamped by
#     `_ensure_table_inplace` onto rows it WRITES. Nothing here is ever written, so the column
#     would be blank for every row forever. `section_line_grids` tolerates the key; the honest
#     thing is not to declare it.


def order_line_fields():
    """One row per `sale.order.line` on a CONFIRMED order — the same `state in (sale, done)`
    scope every wholesale metric in this product uses.

    MEASURED on the mirror 2026-08-12: **254,189 lines in scope** of 256,810 (the 2,621 excluded
    sit on draft/sent/cancelled orders). Every line in scope is on a `sale` order — zero `done` —
    but `done` stays in the option list because it is in the SCOPE, and a filter offering only
    what happens to be stored today goes stale the first time an order is marked done.

    ⛔ THE SCOPE, THE ORDER DATE AND THE ORDER NAME ALL LIVE ACROSS A JOIN. `sale_order_line`
    carries no `state` at all (12 columns, measured), so the binding is a join to `sale_order` —
    which is also what makes `order_no` a readable primary cell instead of a line id.

    ⚠ `qty` IS `int`, NOT `currency`, AND THAT IS A MEASUREMENT. 3,888 of 256,810 lines carry a
    FRACTIONAL quantity (0.2, 0.4, 0.5, 1.66 …) and the minimum is -1.0, so the question "does the
    type truncate?" had to be answered rather than assumed: it does not. `int` and `currency` both
    render through the client's `numberText`, which rounds nothing without a `format.decimals`
    bag — the only difference is the `$` a `currency` column prepends. A quantity is not money, so
    it takes the type that does not paint one.
    """
    return [_preset(f) for f in (
        {"key": "order_no", "label": "Order", "type": "text", "source": "overlay",
         "default": True, "pinned": True,
         "description": "The sale order this line belongs to. Zero orders have a blank name, "
                        "which is why it is the primary cell rather than the line id."},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `sale.order.line` id. Also this row's id."},
        {"key": "order_id", "label": "Odoo order id", "type": "int", "source": "overlay",
         "default": False,
         "description": "The `sale.order` id — the key `ut_odoo_orders` is keyed on."},
        {"key": "customer", "label": "Customer", "type": "text", "source": "overlay",
         "default": True},
        {"key": JOIN_KEY, "label": "Odoo partner id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "product", "label": "Product", "type": "text", "source": "overlay",
         "default": True,
         "description": "Blank on the 101 section and note lines, which carry no product."},
        {"key": PRODUCT_JOIN_KEY, "label": "Odoo product id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "qty", "label": "Qty", "type": "int", "source": "overlay", "default": True,
         "agg": "sum",
         "description": "Ordered quantity. 3,888 lines carry a fraction and some are negative "
                        "(returns), so nothing here is rounded."},
        {"key": "price_subtotal", "label": "Line $", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        {"key": "margin", "label": "Margin $", "type": "currency", "source": "overlay",
         "default": False, "agg": "sum",
         "description": "Odoo's own line margin. Populated on every line."},
        {"key": "purchase_price", "label": "Unit cost", "type": "currency", "source": "overlay",
         "default": False,
         "description": "The cost Odoo priced this line's margin against, per unit."},
        {"key": "order_date", "label": "Order date", "type": "date", "source": "overlay",
         "default": True},
        {"key": "state", "label": "State", "type": "select", "source": "overlay",
         "default": False, "options": ["sale", "done"]},
        _scope_field(),
    )]


def gl_line_fields():
    """One row per `account.move.line` — the general ledger, and the owner's "expenses" at the
    grain a person can actually browse.

    MEASURED 2026-08-12: **963,783 lines**, of which 944,846 posted, 18,885 cancelled and 52
    draft. ⛔ UNSCOPED ON PURPOSE — a general ledger whose draft and cancelled entries are
    invisible is a ledger that cannot be reconciled, so `parent_state` rides as a COLUMN and the
    reader chooses. That is the same decision the binding states from the SQL side.

    ⚠ TWO COLUMNS ARE LEGITIMATELY BLANK ON REAL ROWS, named here so neither reads as a defect:
    **61,911 lines carry no partner** (journal entries that are not about a customer), and
    **1,727 carry no account** at all, which is also why `account_code` — the key
    `ut_odoo_accounts` is keyed on — is blank on exactly those 1,727 and the join that supplies
    it is a LEFT one.

    ⚠ `move_type` IS `text`, NOT `select`, and it is the `product_fields.category` argument:
    five values exist today (`out_invoice` 515,634 · `entry` 414,992 · `in_invoice` 18,901 ·
    `out_refund` 14,061 · `in_refund` 195) and Odoo's enum is longer than what we happen to hold.
    A select whose options go stale answers a filter with a list that cannot match a stored value
    (wave-26 item 24). `line_type` and `parent_state` ARE selects because their option lists were
    measured COMPLETE against the whole table.
    """
    return [_preset(f) for f in (
        {"key": "entry", "label": "Entry", "type": "text", "source": "overlay",
         "default": True, "pinned": True,
         "description": "The journal entry this line belongs to. Never blank."},
        {"key": "odoo_id", "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `account.move.line` id. Also this row's id."},
        {"key": "move_id", "label": "Odoo entry id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "account", "label": "Account", "type": "text", "source": "overlay",
         "default": True},
        {"key": ACCOUNT_JOIN_KEY, "label": "Account code", "type": "text", "source": "overlay",
         "default": True,
         "description": "The GL code, from the joined chart of accounts — the key "
                        "`ut_odoo_accounts` is keyed on. Blank on the 1,727 lines with no "
                        "account."},
        {"key": "customer", "label": "Partner", "type": "text", "source": "overlay",
         "default": True,
         "description": "Blank on the 61,911 lines that are not about a partner."},
        {"key": JOIN_KEY, "label": "Odoo partner id", "type": "int", "source": "overlay",
         "default": False},
        {"key": "date", "label": "Date", "type": "date", "source": "overlay", "default": True},
        {"key": "debit", "label": "Debit", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        {"key": "credit", "label": "Credit", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum"},
        {"key": "balance", "label": "Balance", "type": "currency", "source": "overlay",
         "default": True, "agg": "sum",
         "description": "Debit minus credit, as Odoo stores it. Sums to zero over a whole entry."},
        {"key": "line_type", "label": "Line type", "type": "select", "source": "overlay",
         "default": False,
         "options": ["product", "cogs", "payment_term", "line_note", "line_section"]},
        {"key": "move_type", "label": "Document type", "type": "text", "source": "overlay",
         "default": False},
        {"key": "parent_state", "label": "Entry state", "type": "select", "source": "overlay",
         "default": True, "options": ["draft", "posted", "cancel"]},
        _scope_field(),
    )]


def product_fields():
    """One row per `product.product`, keyed on its id — EVERY product, archived ones included.

    ⚠ THE ROW ID IS THE PRODUCT ID, NOT THE SKU CODE, and the difference is measurable: 12 codes
    map to more than one product id (re-SKU / merge history). The code is what a human reads and
    the id is what `sales_lines.product` groups by, so both are columns and only the id is the
    identity.

    ⛔ THE PINNED COLUMN IS THE NAME, NOT THE SKU, and that is not a style choice. 62 products
    carry no `default_code` at all (UBER CHARGE, Delivery Charges, PICK UP …) while ZERO carry a
    blank name — measured. Pinning `code` would give those rows a blank primary cell, which is
    exactly D-80: a first column nothing populates quietly becoming the row's identity
    ([[fallback-that-became-the-rule]]).
    """
    return [_preset(f) for f in (
        {"key": "product", "label": "Product", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": PRODUCT_JOIN_KEY, "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `product.product` id. Also this row's id, and what "
                                          "every product rollup groups by."},
        {"key": "code", "label": "SKU", "type": "text", "source": "overlay",
         "default": True, "description": "Odoo's `default_code`. Blank on the 62 charge/service "
                                         "products that are not stocked SKUs."},
        {"key": "active", "label": "Active in Odoo", "type": "checkbox", "source": "overlay",
         "default": True,
         "description": "Unticked = archived. Archived products are kept because they still "
                        "carry sales history — two of them sold this year."},
        # ⚠ TEXT, NOT SELECT. 71 categories exist today and Odoo gains them without telling us; a
        # select whose options go stale answers a filter with a list that cannot match a stored
        # value (wave-26 item 24). Text filters honestly and never goes out of date.
        {"key": "category", "label": "Category", "type": "text", "source": "overlay",
         "default": True},
        {"key": "product_type", "label": "Type", "type": "select", "source": "overlay",
         "default": False, "options": ["product", "consu", "service"]},
        {"key": "standard_price", "label": "Standard cost", "type": "currency",
         "source": "overlay", "default": True},
        # ⭐ SOURCE-BACKED (read-through) — the product-grain half of "compute all of the data in
        # Odoo". It names a governed TOPIC + METRIC KEY and one grouped query answers every SKU;
        # `sales_lines` holds 256,810 rows that are never copied into this table.
        {"key": "sales_ytd", "label": "Sales YTD", "type": "rollup", "source": "overlay",
         "default": True, "agg": "sum",
         "rollup": {"source": {"topic": "sales_lines", "measure": "revenue",
                               "groupBy": "product", "on": PRODUCT_JOIN_KEY, "window": "ytd"}}},
        {"key": "units_ytd", "label": "Units YTD", "type": "rollup", "source": "overlay",
         "default": True, "agg": "sum",
         "rollup": {"source": {"topic": "sales_lines", "measure": "units",
                               "groupBy": "product", "on": PRODUCT_JOIN_KEY, "window": "ytd"}}},
        {"key": "margin_ytd", "label": "Gross margin YTD $", "type": "rollup", "source": "overlay",
         "default": True, "agg": "sum",
         "rollup": {"source": {"topic": "sales_lines", "measure": "margin",
                               "groupBy": "product", "on": PRODUCT_JOIN_KEY, "window": "ytd"}}},
        _refreshed_field(),
    )]


def customer_fields():
    """One row per partner Odoo has transacted with, keyed on the `res.partner` id.

    ⭐ Two DERIVED links (`on` declared), so the engine owns both cells and a human cannot edit a
    relation Odoo already decided. The rollups come in two kinds on purpose:
      * LINK rollups fold the rows in `ut_odoo_invoices` / `ut_odoo_orders` — they can answer
        anything about a document the table holds, including a date rank;
      * SOURCE rollups name a governed topic + metric and are answered by ONE grouped SQL query
        over the whole mirror — they can answer a DATE-WINDOWED money question, which a link
        rollup cannot, because a condition can only compare against a literal and a literal year
        start is right until 1 January.
    """
    return [_preset(f) for f in (
        {"key": "customer", "label": "Customer", "type": "text", "source": "overlay",
         "default": True, "pinned": True},
        {"key": JOIN_KEY, "label": "Odoo ID", "type": "int", "source": "overlay",
         "default": False, "description": "The `res.partner` id. Also this row's id."},
        {"key": "city", "label": "City", "type": "text", "source": "overlay", "default": True},
        {"key": "state", "label": "State", "type": "text", "source": "overlay", "default": True},
        {"key": "country", "label": "Country", "type": "text", "source": "overlay",
         "default": False},
        {"key": "agent", "label": "Sales agent", "type": "text", "source": "overlay",
         "default": True,
         "description": "The customer's assigned agent (res.partner.agent_ids[0] — the "
                        "Customers-module convention)."},
        # ⭐ WAVE 28 — the agent's ID beside its NAME, because a link joins on an id and this
        # table carried only the display string. ⚠ It is `agent_id`, NEVER `partner_id`: both are
        # `res.partner` ids, and joining agents through `JOIN_KEY` would link every customer to
        # itself and look plausible doing it.
        {"key": AGENT_JOIN_KEY, "label": "Odoo agent id", "type": "int", "source": "overlay",
         "default": False},
        _scope_field(),

        # --- the relations -------------------------------------------------------------------
        {"key": "invoices", "label": "Invoices", "type": "link", "source": "overlay",
         "default": True,
         "link": {"table": INVOICES_KEY, "on": JOIN_KEY, "from": JOIN_KEY}},
        {"key": "orders", "label": "Orders", "type": "link", "source": "overlay",
         "default": True,
         "link": {"table": ORDERS_KEY, "on": JOIN_KEY, "from": JOIN_KEY}},
        # MEASURED: 2,093 customers carry an `agent_id` and all 2,093 resolve to a row in the
        # agents table — zero dangling, which is why this ships as a link rather than a lookup.
        {"key": "agent_link", "label": "Agent record", "type": "link", "source": "overlay",
         "default": False,
         "link": {"table": AGENTS_KEY, "on": AGENT_JOIN_KEY, "from": AGENT_JOIN_KEY}},

        # --- link rollups over the invoice history --------------------------------------------
        # ⭐ NO CONDITION, and that is measured rather than assumed: a settled document's residual
        # is exactly 0, so summing the full history gives the open balance to the cent.
        # ⚠ THE LABEL SAYS "ALL CHANNELS" BECAUSE THE COLUMN TOTAL DOES NOT MATCH THE AR PAGE.
        # Per customer this is exactly right. Summed down the column it is $2,347,608.49 while
        # `Settings → AR` shows $591,828.54 — a 4x gap that is entirely the GIFTWARE DEALS /
        # Amazon partner, which wholesale scope excludes and this table deliberately keeps. Two
        # numbers with one name, 4x apart, in one product is how a correct figure gets reported
        # as a bug; the scope belongs in the label, not only in a column somebody has to filter.
        {"key": "ar_outstanding", "label": "AR outstanding $ - all channels", "type": "rollup",
         "source": "overlay", "default": True, "agg": "sum",
         "description": "Open balance across every posted document, INCLUDING the Amazon "
                        "channel. Filter `In wholesale scope` to reconcile with the AR page.",
         "rollup": {"link": "invoices", "field": "residual", "fn": "sum"}},
        {"key": "invoiced_all_time", "label": "Invoiced $ - all time", "type": "rollup",
         "source": "overlay", "default": False, "agg": "sum",
         "rollup": {"link": "invoices", "field": "amount_untaxed", "fn": "sum"}},
        # ⛔ THESE TWO DO NEED THE PREDICATE. Over the widened link a bare `countall` counts every
        # document ever posted and labels it "open invoices" — the wrong-number-that-looks-right
        # this module refuses everywhere else.
        {"key": "open_invoices", "label": "Open invoices #", "type": "rollup",
         "source": "overlay", "default": True,
         "rollup": {"link": "invoices", "fn": "countall", **_OPEN_ONLY}},
        # ⛔ NOT `min`. `_rollup_fold`'s min/max are NUMERIC folds (`_lane_num`), so `min` over a
        # date column finds no numbers and returns BLANK — a column that renders empty forever
        # while looking configured. Ranking a DATE is what `latest` + `sortBy` is for.
        {"key": "oldest_due", "label": "Oldest due date", "type": "rollup", "source": "overlay",
         "default": True,
         "rollup": {"link": "invoices", "field": "due_date", "fn": "latest",
                    "sortBy": "due_date", "sortDir": "asc", **_OPEN_ONLY}},

        # --- link rollups over the order history ----------------------------------------------
        {"key": "order_count", "label": "Orders #", "type": "rollup", "source": "overlay",
         "default": True,
         "rollup": {"link": "orders", "fn": "countall"}},
        {"key": "last_order", "label": "Last order date", "type": "rollup", "source": "overlay",
         "default": True,
         "rollup": {"link": "orders", "field": "order_date", "fn": "latest",
                    "sortBy": "order_date", "sortDir": "desc"}},

        # --- source-backed (read-through) rollups ---------------------------------------------
        # ⛔ THESE DO NOT AND CANNOT COME FROM THE `invoices` LINK. `ut_odoo_invoices` is posted
        # BILLING; `revenue_invoiced` is ORDER-LINE revenue narrowed by the order's fully-invoiced
        # flag. Different grain, different question — the metric KEY carries the distinction,
        # which is the whole reason a rollup may not carry SQL of its own.
        {"key": "sales_ytd", "label": "Sales YTD - invoiced", "type": "rollup",
         "source": "overlay", "default": True, "agg": "sum",
         "rollup": {"source": {"topic": "sales_lines", "measure": "revenue_invoiced",
                               "groupBy": "order_partner", "on": JOIN_KEY, "window": "ytd"}}},
        {"key": "sales_ltm", "label": "Sales LTM", "type": "rollup", "source": "overlay",
         "default": True, "agg": "sum",
         "rollup": {"source": {"topic": "sales_lines", "measure": "revenue",
                               "groupBy": "order_partner", "on": JOIN_KEY, "window": "ltm"}}},
        {"key": "margin_ytd", "label": "Gross margin YTD $", "type": "rollup", "source": "overlay",
         "default": False, "agg": "sum",
         "rollup": {"source": {"topic": "sales_lines", "measure": "margin",
                               "groupBy": "order_partner", "on": JOIN_KEY, "window": "ytd"}}},
        {"key": "orders_ytd", "label": "Orders YTD #", "type": "rollup", "source": "overlay",
         "default": False,
         "rollup": {"source": {"topic": "sales_orders", "measure": "orders",
                               "groupBy": "partner", "on": JOIN_KEY, "window": "ytd"}}},
        _refreshed_field(),
    )]


# ---------------------------------------------------------------------------------------------
# READING THE MIRROR
# ---------------------------------------------------------------------------------------------
def excluded_names():
    """The partner names out of wholesale scope, from the ONE place that defines them.

    Read through `core.odoo` rather than re-listed here: a second literal is a second scope, and
    the day somebody adds a channel this module would keep answering the old question.
    """
    try:
        import core.odoo as odoo
        names = getattr(odoo, "EXCLUDE_PARTNER_NAMES", None) or set()
        return {str(n).strip().lower() for n in names if str(n).strip()}
    except Exception:                                                 # noqa: BLE001
        return set()


def excluded_ids(cur, names=None):
    """The out-of-scope partner IDS, resolved against the MIRROR.

    ⭐ IDS, NOT THE DENORMALISED NAME ON THE DOCUMENT, for two reasons that both bite.
    `account_move.partner_name` is a copy taken when the document was written, and this module's
    own `customers_from` says so out loud — *"a partner's name can differ across documents
    (renames land on new invoices only)"*. So a rename would put some of one partner's documents
    in scope and the rest out, silently, and the totals would stop reconciling with nothing to
    point at. `modules/ar`, the oracle these numbers answer to, has always excluded by ID.

    ⛔ RESOLVED FROM THE MIRROR, NOT `core.odoo.excluded_partner_ids()`. That function issues a
    LIVE `search_read`, so importing it here would make spawning four locked databases fail
    whenever Odoo is unreachable — including on this developer machine, where the handshake dies
    on an expired certificate ([[local-odoo-ssl-quirk]]). Same names, same answer, no network.

    ⚠ MATCHED CASE- AND WHITESPACE-INSENSITIVELY, and a NULL name simply does not match — which
    is the correct direction. 44 transacting partners carry no name at all; treating an
    unanswerable name as "excluded" would drop $7,734.83 of real open AR out of scope.
    """
    names = names if names is not None else excluded_names()
    if not names:
        return set()
    rows = cur.execute("SELECT id, name FROM res_partner WHERE name IS NOT NULL").fetchall()
    return {int(pid) for pid, name in rows if str(name).strip().lower() in names}


def columns(cur, table):
    """The column names a mirror table actually has, lowercased. `set()` if the table is absent.

    ⛔ WHY THIS EXISTS, AND IT COST A LIVE 500. `harness.datastore.ready()` gates on ENTITY
    phases, and a Space hydrates its mirror from `store_seed/royal.duckdb` — a SNAPSHOT. Columns
    added to `ENTITIES` after that snapshot was taken (`res_partner.agent_id`,
    `account_move_line.product_id`, …) are backfilled by `sync_all()` under their OWN `_sync_state`
    keys, which `ready()` does not read. So there is a real window, right after a boot, where the
    store reports READY and a column this module names does not exist yet — and DuckDB answers a
    missing identifier with a Binder error, which reached the operator as a bare `500`.
    ⚠ The absent columns are all DISPLAY ones (an agent name, a category, a team). Refusing the
    whole spawn over a cosmetic column would be worse than the gap it is reporting, so the readers
    degrade the COLUMN to blank and still write every id.
    """
    try:
        rows = cur.execute(f"SELECT * FROM {table} LIMIT 0")
        return {str(d[0]).lower() for d in rows.description}
    except Exception:                                                 # noqa: BLE001
        return set()


def _col(have, name, default="NULL"):
    """`name` when the mirror has it, else a literal that keeps the SELECT's arity intact."""
    return name if str(name).split(".")[-1].lower() in have else default


def _as_date(value):
    """ISO date string, or ''. The grid renders `date` cells itself (W26: `Aug 5, 2026`), so the
    STORED value stays ISO — a formatted string in the cell is a value the filters cannot sort."""
    if not value:
        return ""
    return str(value)[:10]


def _in_scope(pid, excluded):
    """`'1'` | `''` — the `checkbox` cell convention (`aios_grid`: the overlay stores '1' or '')."""
    return "" if int(pid) in excluded else "1"


def read_invoices(cur, excluded=None, open_only=False):
    """[(row dict)] — posted customer invoices and refunds, keyed on the `account.move` id.

    ONE reader, two projections. `open_only` applies the AR oracle's predicate and drops the
    out-of-scope channel, which is what `read_open_ar` wants; the default keeps every row and
    TAGS the channel instead. Two queries would be two populations, and they drift the moment
    either is edited.

    Takes a CURSOR so a gate can hand it a fixture connection; no global store binding here.
    """
    excluded = excluded if excluded is not None else excluded_ids(cur)
    where = _AR_WHERE if open_only else _POSTED_DOCS
    # ⚠ `invoice_origin` (D-88) is read through `_col` DELIBERATELY. It was added to `ENTITIES` in
    # this same wave, so a Space whose mirror is still hydrating from a pre-wave seed snapshot does
    # not have the column yet — and DuckDB answers a missing identifier with a Binder error that
    # reaches the operator as a bare 500. This is the exact class `columns()` was written for: the
    # link degrades to blank for one sync cycle instead of refusing the whole spawn.
    have = columns(cur, "account_move")
    sql = ("SELECT id, name, partner_id, partner_name, invoice_date, invoice_date_due, "
           "       amount_untaxed_signed, amount_residual_signed, payment_state, move_type, "
           f"       {_col(have, 'invoice_origin', chr(39) + chr(39))} "
           f"FROM account_move WHERE {where} AND partner_id IS NOT NULL")
    out = []
    for r in cur.execute(sql).fetchall():
        (mid, name, pid, pname, inv_date, due, untaxed, residual, pay_state, mtype, origin) = r
        scope = _in_scope(pid, excluded)
        if open_only and not scope:
            continue
        out.append({
            "_id": str(mid),
            "invoice_no": str(name or ""),
            "odoo_id": int(mid),
            "customer": str(pname or ""),
            JOIN_KEY: int(pid),
            "invoice_date": _as_date(inv_date),
            "due_date": _as_date(due),
            "residual": float(residual or 0.0),
            "amount_untaxed": float(untaxed or 0.0),
            "payment_state": str(pay_state or ""),
            "move_type": str(mtype or ""),
            "origin_order": str(origin or "").strip(),
            "wholesale_scope": scope,
        })
    return out


def read_open_ar(cur, excluded=None):
    """The OPEN, wholesale-scoped subset — `modules/ar._open_docs`' own population.

    Kept as its own door because `modules/ar` is this module's oracle for the AR numbers, and an
    oracle answers exactly one question. It is a projection of `read_invoices`, never a second
    query.
    """
    return read_invoices(cur, excluded=excluded, open_only=True)


def read_orders(cur, excluded=None):
    """[(row dict)] — confirmed sale orders, keyed on the `sale.order` id."""
    excluded = excluded if excluded is not None else excluded_ids(cur)
    have = columns(cur, "sale_order")
    sql = (f"SELECT id, name, date_order, partner_id, partner_name, {_col(have, 'team_name')}, "
           f"       state, amount_untaxed, {_col(have, 'invoice_status')} "
           f"FROM sale_order WHERE {_CONFIRMED} AND partner_id IS NOT NULL")
    out = []
    for r in cur.execute(sql).fetchall():
        (oid, name, when, pid, pname, team, state, untaxed, inv_status) = r
        out.append({
            "_id": str(oid),
            "order_no": str(name or ""),
            "odoo_id": int(oid),
            "customer": str(pname or ""),
            JOIN_KEY: int(pid),
            "order_date": _as_date(when),
            "amount_untaxed": float(untaxed or 0.0),
            "team": str(team or ""),
            "state": str(state or ""),
            "invoice_status": str(inv_status or ""),
            "wholesale_scope": _in_scope(pid, excluded),
        })
    return out


def read_products(cur):
    """[(row dict)] — EVERY `product.product`, keyed on its id.

    ⛔ NO `active` AND NO `default_code` FILTER, and both exclusions were measured before they
    were dropped. Filtering to active-and-coded gave 5,829 of 5,948 rows and left FIVE products
    that sold this very year with no row at all: two archived SKUs (`9SAT-FY`, `2GSTY`) and three
    uncoded charge lines (`UBER CHARGE`, `[Delivery_009] Delivery Charges`, `PICK UP`). A product
    grouped by `sales_lines.product` that has no parent row is a rollup value with nowhere to
    land — silently. 119 extra rows is the whole cost of the claim being literally true.
    """
    have = columns(cur, "product_product")
    sql = (f"SELECT id, default_code, name, {_col(have, 'categ_name')}, type, "
           f"       {_col(have, 'standard_price', '0')}, {_col(have, 'active', 'TRUE')} "
           "FROM product_product")
    out = []
    for r in cur.execute(sql).fetchall():
        (prid, code, name, categ, ptype, cost, active) = r
        out.append({
            "_id": str(prid),
            "product": str(name or ""),
            PRODUCT_JOIN_KEY: int(prid),
            "code": str(code or ""),
            "active": "1" if active else "",
            "category": str(categ or ""),
            "product_type": str(ptype or ""),
            "standard_price": float(cost or 0.0),
        })
    return out


def read_customers(cur, excluded=None):
    """[(row dict)] — every CUSTOMER partner, keyed on the `res.partner` id.

    ⭐ THE POPULATION IS A UNION OF THREE LEGS, and every one of them is load-bearing.

    The two DOCUMENT legs are the original pair: Amazon books as direct invoices with no sale
    order (the `odoo-api` gotcha), so a sale-order leg alone would silently drop a real customer.

    ⭐⭐ THE THIRD IS `customer_rank > 0 AND active` (wave 29, item 22 / R12 via finding F2 —
    the owner's *"never an arbitrary limit… applies to ALL connected database"*). The old
    docstring said partners with no document are *"left out on purpose — a row that can never
    appear in any topic has nothing to roll up"*; that reasoning is RETIRED. It is the same
    join-drop class as the Product grid's 2,717, and it dropped **~1,149 real customer records**
    (MEASURED 2026-08-11: `rank>0 active` = 3,617 against a document union of ~2,000). A customer
    a salesperson has not sold to yet is exactly the row a prospecting view needs.

    ⛔ IT IS A UNION AND NOT A REPLACEMENT, AND THAT IS MEASURED, NOT TIDINESS. Swapping the
    document legs for the rank leg would drop **16 partners that hold posted documents** (7
    archived, 9 active with rank <= 0), and `ut_odoo_invoices` / `ut_odoo_orders` rows carry
    `partner_id` LINKS straight back here — so those links would dangle with nothing reporting
    it. When other tables point AT a population, a widening must be a SUPERSET.

    ⚠ THE RANK LEG IS SKIPPED WHEN THE MIRROR HAS NO `customer_rank` COLUMN, which is the same
    `columns()`/`_col` discipline every other optional column here uses — but note the difference
    honestly: an absent `agent_id` blanks a CELL, while an absent `customer_rank` narrows the
    POPULATION back to the document union. It degrades to today's behaviour rather than to an
    empty or a wrong table, and `verify_odoo_relational` carries a check that goes RED while the
    column is missing so the narrowing can never pass for done.

    ⛔ NOT FROM LIVE ODOO, THOUGH `customer_rank` IS TRIVIAL TO ASK IT. `excluded_ids` above
    states the rule for this module and it applies with more force to a POPULATION than to a name
    list: a live call makes the spawn fail whenever Odoo is unreachable, and a Space hydrates its
    mirror from a SNAPSHOT at boot. The population would then be "whichever source answered this
    time" — swinging ~45% against `MAX_SHRINK`'s 50% refusal, deleting and re-adding rows on the
    weather. One source, always present at spawn time: the mirror.

    ⛔ NOT DERIVED FROM THE INVOICE ROWS. `customers_from` did that when the table WAS the open-AR
    partners; sourcing a customer registry from its own receivables is what kept most Odoo ids
    out of the store in the first place.
    """
    excluded = excluded if excluded is not None else excluded_ids(cur)
    have = columns(cur, "res_partner")
    # ⚠ THE AGENT JOIN IS DROPPED WHOLE when `agent_id` is absent, not merely NULL-ed: the join
    # itself names the column, so `_col` on the SELECT list alone would still fail to bind.
    agent = ("ag.name" if "agent_id" in have else "NULL")
    agent_id_col = ("p.agent_id" if "agent_id" in have else "NULL")
    join = ("LEFT JOIN res_partner ag ON ag.id = p.agent_id " if "agent_id" in have else "")
    # ⚠ BOTH columns must be present, not just `customer_rank`: `active` is what keeps an
    # archived prospect out, and a rank test without it would re-admit the 47 archived partners
    # the mirror carries. Absent ⇒ the leg is dropped WHOLE, exactly like the agent join above.
    rank_leg = ("   OR (p.customer_rank > 0 AND p.active) "
                if {"customer_rank", "active"} <= have else "")
    sql = (f"SELECT p.id, p.name, {_col(have, 'p.city')}, {_col(have, 'p.state_name')}, "
           f"       {_col(have, 'p.country_name')}, {agent}, {agent_id_col} "
           "FROM res_partner p "
           f"{join}"
           "WHERE p.id IN ("
           f"  SELECT partner_id FROM sale_order WHERE {_CONFIRMED} AND partner_id IS NOT NULL "
           "  UNION "
           "  SELECT partner_id FROM account_move "
           f"  WHERE {_POSTED_DOCS} AND partner_id IS NOT NULL)"
           f"{rank_leg}")
    out = []
    for r in cur.execute(sql).fetchall():
        (pid, name, city, state, country, agent, agent_id) = r
        out.append({
            "_id": str(pid),
            "customer": str(name or ""),
            JOIN_KEY: int(pid),
            "city": str(city or ""),
            "state": str(state or ""),
            "country": str(country or ""),
            "agent": str(agent or ""),
            AGENT_JOIN_KEY: int(agent_id) if agent_id else "",
            "wholesale_scope": _in_scope(pid, excluded),
        })
    return out


def read_agents(cur):
    """[(row dict)] — the UNION of both agent sources, keyed on the `res.partner` id.

    ⛔ `res_partner.agent` is a BOOLEAN and `datastore.BOOL_FIELDS` lists it for a measured reason:
    Odoo returns False both for "empty" and for "boolean false", so a bool missing from that list
    silently becomes NULL and every row would read "not an agent" indistinguishably from
    "unknown". Read it as a truth value, never as a presence test.
    """
    have = columns(cur, "res_partner")
    if "id" not in have:
        return []
    flagged = "p.agent" if "agent" in have else "FALSE"
    # ⛔⛔ THE COMMISSION TABLE IS GUARDED AS A **TABLE**, not just as a column, and that
    # distinction is the whole point of this block. `columns()` was written for a missing COLUMN
    # (a backfill that has not run yet); `account_invoice_line_agent` is an OCA module entity that
    # a mirror hydrated from an older seed snapshot may not have AT ALL. A SELECT naming an absent
    # table is a DuckDB Binder error, and this reader runs inside `plan()` — so one missing table
    # would fail the WHOLE eight-table spawn and reach the operator as a bare 500. That is
    # precisely D-107's shape, and it would have arrived on the first deploy of this feature.
    # ⚠ DEGRADE, NEVER REFUSE, which is the posture `columns()`'s own docstring sets: without the
    # commission table the population falls back to the FLAGGED partners alone and `commissioned`
    # reads blank for every row — fewer agents and an honestly empty column, rather than no spawn.
    has_comm = bool(columns(cur, "account_invoice_line_agent"))
    commissioned = ("(p.id IN (SELECT agent_id FROM account_invoice_line_agent "
                    "          WHERE agent_id IS NOT NULL))" if has_comm else "FALSE")
    union_leg = ("  SELECT agent_id FROM account_invoice_line_agent WHERE agent_id IS NOT NULL "
                 "  UNION " if has_comm else "")
    sql = (f"SELECT p.id, p.name, {flagged}, {commissioned} AS commissioned "
           "FROM res_partner p WHERE p.id IN ("
           f"{union_leg}SELECT id FROM res_partner WHERE {flagged})")
    out = []
    for (aid, name, flag, comm) in cur.execute(sql).fetchall():
        out.append({
            "_id": str(aid),
            "agent": str(name or ""),
            "odoo_id": int(aid),
            AGENT_JOIN_KEY: int(aid),
            "flagged": "1" if flag else "",
            "commissioned": "1" if comm else "",
        })
    return out


def read_accounts(cur):
    """[(row dict)] — the whole GL chart, keyed on the `account.account` id.

    ⚠ THE EXPENSE PREDICATE IS THE SEMANTIC LAYER'S, copied rather than invented:
    `harness/semantic.py`'s `gl_lines` topic scopes expenses as
    `account_type in ('expense','expense_depreciation')`. A second definition here is how a
    column and a topic start disagreeing about the same word.
    ⛔ `account.account` has NO `active` column in this Odoo version (a domain naming it 500s), so
    there is nothing to filter and every account is a row.
    """
    have = columns(cur, "account_account")
    if not have:
        return []
    sql = (f"SELECT id, {_col(have, 'code', chr(39) + chr(39))}, "
           f"       {_col(have, 'name', chr(39) + chr(39))}, "
           f"       {_col(have, 'account_type', chr(39) + chr(39))} FROM account_account")
    out = []
    for (aid, code, name, atype) in cur.execute(sql).fetchall():
        t = str(atype or "")
        out.append({
            "_id": str(aid),
            ACCOUNT_JOIN_KEY: str(code or ""),
            "account_name": str(name or ""),
            "odoo_id": int(aid),
            "account_type": t,
            "is_expense": "1" if t in ("expense", "expense_depreciation") else "",
        })
    return out


_VENDOR_DOCS = "state = 'posted' AND move_type IN ('in_invoice','in_refund')"


def read_bills(cur):
    """[(row dict)] — posted vendor bills and refunds, keyed on the `account.move` id."""
    have = columns(cur, "account_move")
    if not have:
        return []
    sql = ("SELECT id, name, partner_id, partner_name, invoice_date, invoice_date_due, "
           f"       {_col(have, 'amount_untaxed_signed', '0')}, "
           f"       {_col(have, 'amount_residual_signed', '0')}, "
           f"       {_col(have, 'payment_state', chr(39) + chr(39))}, move_type "
           f"FROM account_move WHERE {_VENDOR_DOCS} AND partner_id IS NOT NULL")
    out = []
    for r in cur.execute(sql).fetchall():
        (mid, name, pid, pname, when, due, untaxed, residual, pay, mtype) = r
        out.append({
            "_id": str(mid),
            "bill_no": str(name or ""),
            "odoo_id": int(mid),
            "vendor": str(pname or ""),
            VENDOR_JOIN_KEY: int(pid),
            "invoice_date": _as_date(when),
            "due_date": _as_date(due),
            "amount_untaxed": float(untaxed or 0.0),
            "residual": float(residual or 0.0),
            "payment_state": str(pay or ""),
            "move_type": str(mtype or ""),
        })
    return out


def read_vendors(cur):
    """[(row dict)] — every partner carrying a posted vendor bill, keyed on the `res.partner` id.

    ⚠ DERIVED FROM THE BILLS, unlike `read_customers` which is deliberately NOT derived from its
    invoices. The asymmetry is intentional and the reason is what that function's own comment
    says: a customer registry sourced from receivables is what kept most Odoo ids out of the store.
    There is no second document universe for vendors — a partner with no bill has no payable
    history to show — so the bill IS the population, and MEASURED it dangles nothing (0 bills
    carry a null partner; all 393 vendors resolve in `res_partner`).
    """
    have = columns(cur, "res_partner")
    if not have or not columns(cur, "account_move"):
        return []
    sql = (f"SELECT p.id, p.name, {_col(have, 'p.country_name', chr(39) + chr(39))} "
           "FROM res_partner p WHERE p.id IN "
           f"  (SELECT partner_id FROM account_move WHERE {_VENDOR_DOCS} "
           "    AND partner_id IS NOT NULL)")
    out = []
    for (pid, name, country) in cur.execute(sql).fetchall():
        out.append({
            "_id": str(pid),
            "vendor": str(name or ""),
            "odoo_id": int(pid),
            VENDOR_JOIN_KEY: int(pid),
            "country": str(country or ""),
        })
    return out


def customers_from(invoice_rows):
    """The partners carrying the given invoice rows — the pre-2026-08-09 population builder.

    ⚠ NO LONGER WHAT SPAWNS `ut_odoo_customers` (that is `read_customers`). Kept because it is a
    pure function over rows and the gate uses it to prove the FOLD against a fixture without a
    mirror; deleting it would cost a test its independence from the SQL.
    """
    out = {}
    for row in invoice_rows:
        pid = row[JOIN_KEY]
        entry = out.setdefault(str(pid), {"_id": str(pid), "customer": row["customer"],
                                          JOIN_KEY: pid})
        # A partner's name can differ across documents (renames land on new invoices only);
        # the newest non-empty one wins so the locked table shows what Odoo shows today.
        if row["customer"]:
            entry["customer"] = row["customer"]
    return list(out.values())


# ---------------------------------------------------------------------------------------------
# THE SPAWN
# ---------------------------------------------------------------------------------------------
class Refused(Exception):
    """A refusal a caller should SHOW, not swallow. Every raise names what would otherwise have
    been written wrong."""


#: `plan()` bucket -> (store key, nav label, field contract). ⭐ ONE ROW PER TABLE is the whole
#: point: adding an Odoo entity is a spec row plus a reader, not a fifth copy of the spawn code.
#: ⚠ ORDER MATTERS ONLY FOR THE REFUSAL MESSAGE; `plan` checks every cap before anything commits.
TABLES = (
    ("customers", CUSTOMERS_KEY, "Odoo customers", customer_fields),
    ("products", PRODUCTS_KEY, "Odoo products", product_fields),
    ("invoices", INVOICES_KEY, "Odoo invoices", invoice_fields),
    ("orders", ORDERS_KEY, "Odoo orders", order_fields),
    # ⭐ WAVE 28 / R1. Measured populations: 19 / 192 / 6,538 / 393 — every one of them two orders
    # of magnitude inside `MAX_ROWS`, which is why the answer to "every unique id is a database"
    # is four more spec rows and four readers rather than a new substrate.
    ("agents", AGENTS_KEY, "Odoo agents", agent_fields),
    ("accounts", ACCOUNTS_KEY, "Odoo GL accounts", account_fields),
    ("vendors", VENDORS_KEY, "Odoo vendors", vendor_fields),
    ("bills", BILLS_KEY, "Odoo vendor bills", bill_fields),
    # ⭐⭐ W30-T35 / R7 — the two READ-THROUGH grains. They are spec rows like any other, and that
    # is the point: `apply_plan` creates their DEFINITION (label, fields, lock, nav entry, grants)
    # exactly as it does for the eight above, and `plan` hands them ZERO rows. Leaving them out of
    # this tuple was the alternative and it is the wrong one — the route 404s on a key `TABLES`
    # does not name, so the grids would be bound to the mirror and unreachable, which is this
    # wave's own [[reachable-is-not-the-same-as-built]] shape.
    ("order_lines", ORDER_LINES_KEY, "Odoo order lines", order_line_fields),
    ("gl_lines", GL_LINES_KEY, "Odoo GL lines", gl_line_fields),
)

#: The bucket -> reader map. ⛔ ITS ABSENCES ARE LOAD-BEARING: a bucket with no reader has no
#: python row builder ANYWHERE, which is what makes "never materialised" structural rather than a
#: policy `plan()` could forget. The two line grains are absent for that reason and no other.
_READERS = {
    "customers": lambda cur, excluded: read_customers(cur, excluded=excluded),
    "products": lambda cur, excluded: read_products(cur),
    "invoices": lambda cur, excluded: read_invoices(cur, excluded=excluded),
    "orders": lambda cur, excluded: read_orders(cur, excluded=excluded),
    "agents": lambda cur, excluded: read_agents(cur),
    "accounts": lambda cur, excluded: read_accounts(cur),
    "vendors": lambda cur, excluded: read_vendors(cur),
    "bills": lambda cur, excluded: read_bills(cur),
}

#: The table keys this module can never materialise — DERIVED from the absence of a reader, never
#: typed out, so it cannot drift from the fact it describes.
#:
#: ⛔⛔ IT IS STAMPED ONTO THE DEFINITION AT SPAWN, AND THAT IS NOT BELT-AND-BRACES — IT IS THE
#: ONLY WAY THESE TWO TABLES EVER GET THE DURABLE FLAG. `core.user_tables.materialises` reads a
#: process-global registry first and falls back to a stored `readThrough` stamp, "which is what a
#: cold process reads" — but the only writer of that stamp is `strip_materialised`, and it stamps
#: exclusively tables it found rows on (`if isinstance(t, dict) and t.get('rows')`, after an early
#: return when nothing is fat). A table that was BORN read-through has no rows to strip, so it is
#: never stamped, so a process that cannot reach the mirror reads `rows: {}` and calls that the
#: answer — an EMPTY GRID with nothing going red, which is the exact failure that docstring names.
#: The conversion writes the stamp; a table that needs no conversion still needs the statement.
READ_THROUGH_KEYS = frozenset(key for bucket, key, _l, _f in TABLES if bucket not in _READERS)


class _LentDoc:
    """A store handle that serves the ONE `user_tables` document `plan()` has ALREADY read.

    ⛔⛔ THIS IS NOT A MICRO-OPTIMISATION AND IT IS NOT OPTIONAL. `core.user_tables.row_limit`
    resolves through `materialises` → `get` → `all_tables(st)`, and every one of those is a WHOLE
    20 MB document read, deep-copied under `Store._lock`. `plan()` asks the evaluator once per
    table per loop, so passing the live handle would have added ~16 full document copies to a
    function that already reads it exactly once — and with `st=None` (the gate's fixture posture,
    and any dry run) those reads resolve to the MODULE-GLOBAL store, i.e. a Hugging Face dataset
    fetch per table, on a path that has no business touching the network at all.
    `materialises`' own docstring asks callers to lend the definition they are holding; `row_limit`
    takes `st` rather than `defn`, so the lending happens one level up, here.

    ⚠ It answers ONLY the user-tables document and `None` for anything else, deliberately: a shim
    that quietly proxied other keys would be a second store with a partial view, which is worse
    than one that says what it knows.
    """

    def __init__(self, doc, key):
        self._doc, self._key = doc if isinstance(doc, dict) else {}, key

    def get(self, name):
        return self._doc if name == self._key else None


# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⭐⭐ W32-T15/T16/T17 / CONTRACT C2 / RULINGS R9, R10, R11 — THE CONNECTOR'S OWN CONFIGURATION
# ═════════════════════════════════════════════════════════════════════════════════════════════
#
# The owner opened the Odoo connector and found nothing to configure: no key, no server database,
# no choice of which grids to materialise, no sync cadence, no way off. R9/R10/R11 answer all
# four, and the state lives HERE rather than in the route because `refresh()` is what has to obey
# it — a config the route knows and the sync path does not is a switch that flips nothing.
#
#: `{"grids": {table_key: bool}, "syncEvery": "<preset>", "frozen": bool, "frozenAt": "<iso>"}`
CONFIG_KEY = "odoo_connector_config"

#: ⭐ R11's presets, and the FLOOR IS THE POINT. Owner: *"30m / 1h / 4h / daily / manual. The
#: floor is 30 minutes, server enforced; no free-text interval."* An interval box would let a
#: tenant ask for 60 s against an ERP over XML-RPC and a Hugging Face free-tier container.
#: ⚠ `manual` is not "very slow" — it is NO scheduled sync at all, which is why it maps to None
#: rather than to a large number. A caller that treats None as a duration gets a TypeError rather
#: than a silent once-a-century schedule.
SYNC_PRESETS = {"30m": 1800, "1h": 3600, "4h": 14400, "daily": 86400, "manual": None}
SYNC_FLOOR_SECONDS = 1800
DEFAULT_SYNC = "30m"


def read_config(rt):
    """This tenant's stored connector config, defaulted. Never raises — a store hiccup must not
    make a connector look disconnected."""
    try:
        cur = rt.get(CONFIG_KEY) if rt is not None else None
    except Exception:                                                  # noqa: BLE001
        cur = None
    cur = cur if isinstance(cur, dict) else {}
    grids = cur.get("grids") if isinstance(cur.get("grids"), dict) else {}
    every = cur.get("syncEvery")
    return {"grids": {str(k): bool(v) for k, v in grids.items()},
            "syncEvery": every if every in SYNC_PRESETS else DEFAULT_SYNC,
            "frozen": bool(cur.get("frozen")),
            "frozenAt": str(cur.get("frozenAt") or "")}


def grid_choices(rt):
    """`[{key, label, enabled}]` for every grid this connector can materialise — DERIVED from
    `TABLES`, never a second hand-typed list (contract C2's parity leg asserts exactly that).

    ⚠ ABSENT MEANS ENABLED. A tenant that has never opened the panel has every grid, which is
    what they have today; only an explicit untick turns one off. The alternative — an empty
    config meaning "nothing enabled" — would silently unspawn ten live databases on deploy.
    """
    chosen = read_config(rt)["grids"]
    return [{"key": key, "label": label, "enabled": bool(chosen.get(key, True))}
            for _bucket, key, label, _fields in TABLES]


def enabled_buckets(rt):
    """The BUCKET names `plan()` speaks, for the grids this tenant has left ticked."""
    chosen = read_config(rt)["grids"]
    return {bucket for bucket, key, _l, _f in TABLES if chosen.get(key, True)}


def sync_seconds(rt):
    """How often this tenant's Odoo mirror should resync, or None for `manual` (R11).

    ⛔ THE FLOOR IS ENFORCED HERE AS WELL AS AT THE WRITE DOOR, deliberately. A stored value that
    predates the preset list, or one written by any path that is not the route, must still not be
    able to ask this loop for a 60-second cycle — a limit with only one enforcer is a limit that
    holds until somebody finds the second way in [[limit-with-no-enforcer]].
    """
    secs = SYNC_PRESETS.get(read_config(rt)["syncEvery"], SYNC_PRESETS[DEFAULT_SYNC])
    if secs is None:
        return None
    return max(int(secs), SYNC_FLOOR_SECONDS)


def frozen(rt):
    """Has this tenant DISCONNECTED Odoo (R10)? Frozen grids keep every row and every field and
    stop being refreshed — distinct from PAUSED, which is temporary and keeps the credential."""
    return bool(read_config(rt)["frozen"])


def plan(cur, rt=None):
    """The rows that WOULD be written, plus the refusals that apply — no store WRITE at all.

    Separated from `apply_plan` so a route, a gate and a dry run all measure the same thing, and
    so **every cap is checked before anything is committed**.

    Returns one key per bucket plus two that are not buckets: `problems` (refusals — a non-empty
    list makes `apply_plan` raise before it writes anything) and, since W30-T35, **`limits`** —
    `{table_key: limit_report}` for every table this plan did not fully materialise, which is R6's
    second sentence carried as data rather than left for a reader to infer from an empty list.

    ⛔ `rt` IS WHAT MAKES THE TABLE-COUNT CHECK HONEST, and leaving it out was a real half-spawn
    bug. This spawn writes FOUR tables in four updater passes; a tenant near `MAX_TABLES` would
    create some and refuse the rest — leaving a locked invoices database with no rollup host,
    while the route answered as though nothing had happened. A partial spawn is worse than a
    refused one, so the count is checked against the tables that ALREADY exist, before the first
    write. `rt` also carries the stored row counts the shrink guard compares against.
    """
    ut = _ut()
    # ⭐ R6, AND WITHOUT THIS LINE THE RULE IS ONLY ACCIDENTALLY TRUE. `is_connected` answers from
    # three places in falling authority: the registry, a stored `connected: True`, then the
    # `ut_odoo_` naming convention — and that last leg needs the table to ALREADY EXIST. So on a
    # FIRST spawn, in a process that has not yet built `routes_odoo_tables.GRID_SOURCES`, every one
    # of these tables reads as unconnected and earns `MAX_ROWS`: R6's cap removal would silently
    # not apply on exactly the run that creates the databases. This module DECLARES these keys, so
    # it is the honest place to say what they are. Idempotent (a set add), and it fills the
    # evaluator's input rather than becoming a second evaluator.
    ut.register_connected(*[key for _b, key, _l, _f in TABLES])
    excluded = excluded_ids(cur)
    # ⭐⭐ W30-T35 / R6 / R7 — WHICH BUCKETS ARE BUILT AT ALL IS NOW ASKED, NOT ASSUMED, and it is
    # `core.user_tables.row_limit` that answers: 0 = "stores no rows HERE" (read-through), None =
    # "connected and uncapped", MAX_ROWS = "the editable substrate". Its own docstring names this
    # function as the caller that reads it, which is the seam working as designed — one evaluator,
    # so the spawn, the write doors and the wire cannot disagree about whether a table is capped
    # ([[one-evaluator-per-question]]).
    #
    # ⛔ TWO DIFFERENT REASONS NOT TO BUILD, AND THEY ARE KEPT SEPARATE ON PURPOSE:
    #   * no reader at all — structural, permanent, and the case that must not depend on a store
    #     read succeeding (a cold process with no mirror still must not try to build 963,783 rows);
    #   * a reader exists but the table has already been converted to read-through — the
    #     `ut_odoo_accounts` case. Building 192 rows and letting `strip_materialised` delete them
    #     again on the next pass "works", and it is exactly the wasted, dangerous work `row_limit`
    #     was built to prevent. It also stops a refresh from silently RE-MATERIALISING a table
    #     D-87's conversion had already emptied.
    #
    # ⚠ THE DOCUMENT IS READ **ONCE**, HERE, AND LENT TO THE EVALUATOR — see `_LentDoc`. It used
    # to be read after the build loop; it moved up because `row_limit` needs it and reading it per
    # table per loop is ~16 more whole-document deep copies (or, with `st=None`, a Hugging Face
    # fetch per table on a path that must never touch the network).
    existing = {}
    if rt is not None:
        try:
            existing = dict(rt.get(ut.STORE_KEY) or {})
        except Exception:                                             # noqa: BLE001
            existing = {}
    # ⛔ THE LENT DOCUMENT CARRIES THE `readThrough` STAMP THIS MODULE IS RESPONSIBLE FOR, and
    # without it R6's report is silently absent on the run that matters most — the FIRST spawn.
    # MEASURED: with an empty store, `row_limit` finds no registry entry and no stored stamp, so
    # it answers `None` ("connected and uncapped") for a grain that stores nothing at all, and
    # `limit_report` answers None with it — so `plan()["limits"]` came back EMPTY and the grids
    # were skipped with no stated reason. The structural `reader is None` guard still did its job;
    # what went missing was the half of R6 that has to SAY WHY.
    #
    # ⚠ THE OBVIOUS FIX IS THE ONE I DID NOT TAKE: `ut.register_read_through(*READ_THROUGH_KEYS)`
    # would work in one line, and `core.user_tables` explicitly reserves that registrar —
    # *"IT IS ALSO THE ONLY PLACE THAT MAY CALL `register_read_through`"* — for
    # `routes_odoo_tables.sync_read_through`, because ELIGIBILITY needs the mirror and the fold
    # matrix. That reasoning does not apply to a grain with no reader (there is nothing it could
    # be eligible FOR), but the law is written without an exception, so this lends the evaluator
    # the definition instead of taking one. `_ensure_table_inplace` writes exactly this stamp, so
    # what is lent is the document as it stands the moment this plan applies.
    lent = _LentDoc({**existing,
                     **{k: {**(existing.get(k) or {}), "readThrough": True}
                        for k in READ_THROUGH_KEYS}},
                    ut.STORE_KEY)

    built, limits = {}, {}
    caps = {}
    for bucket, key, _label, _fields in TABLES:
        reader = _READERS.get(bucket)
        caps[key] = cap = ut.row_limit(key, st=lent)
        if reader is None or cap == 0:
            built[bucket] = []
            report = ut.limit_report(key, st=lent)
            if report:
                limits[key] = report
            continue
        built[bucket] = reader(cur, excluded)
    problems = []

    if rt is not None:
        needed = [k for _b, k, _l, _f in TABLES if k not in existing]
        if needed and len(existing) + len(needed) > ut.MAX_TABLES:
            problems.append(
                f"tenant holds {len(existing)} of MAX_TABLES={ut.MAX_TABLES} user tables and "
                f"needs {len(needed)} more ({', '.join(needed)}); refusing rather than creating "
                f"part of a linked set")

    for bucket, key, _label, _fields in TABLES:
        rows = built[bucket]
        cap = caps[key]          # asked ONCE per table, above — never re-read per loop
        # ⭐⭐ R6: THE `MAX_ROWS` REFUSAL IS GONE FOR A CONNECTED SOURCE, AND THE SENTENCE IT USED
        # TO PRINT IS NOW `limit_report`'s STRUCTURED ANSWER. Owner, verbatim: *"there is no cap in
        # how many data from the API source (as long as its from a connected source like Odoo) that
        # can be pulled into the app… Now if there is lag or it can't be done, you need to
        # explicitly tell me why and recommend a fix."* Both halves are here: a connected table
        # answers `None` and is never refused for its size, and anything that IS still bounded is
        # reported with its cause and its recommendation instead of a hand-typed line.
        #
        # ⛔ THE `cap and` GUARD IS THE WHOLE CHANGE AND ITS TWO FALSY CASES MEAN OPPOSITE THINGS:
        # `None` = connected, uncapped, build every row Odoo has; `0` = stores no rows here, and
        # the loop above already handed it an empty list. Neither may reach the refusal. The
        # editable substrate still gets `MAX_ROWS` and is still REFUSED, never truncated — a capped
        # table understates every total it feeds while looking exactly like a complete one.
        if cap and len(rows) > cap:
            report = ut.limit_report(key, st=lent) or {}
            limits[key] = report
            problems.append(
                f"{key}: {len(rows):,} rows exceeds the {cap:,}-row ceiling; refusing "
                f"({report.get('cause', 'a truncated table understates every rollup it feeds')}). "
                f"{report.get('recommendation', '')}".strip())
        # ⚠ THE SHRINK GUARD SKIPS A READ-THROUGH GRAIN, and without this it would refuse every
        # spawn after the first conversion: zero rows against a stored population is the INTENDED
        # end state there, not the partial mirror read this guard exists to catch.
        if cap == 0:
            continue
        stored = len(((existing.get(key) or {}).get("rows")) or {})
        if stored and len(rows) < stored * MAX_SHRINK:
            problems.append(
                f"{key}: the mirror answered {len(rows)} rows against {stored} stored — a drop of "
                f"more than {int((1 - MAX_SHRINK) * 100)}% is a bad read, not Odoo history "
                f"shrinking; refusing rather than deleting rows that still exist")
    built["problems"] = problems
    # R6's second sentence as DATA rather than prose: every table whose rows this plan did not
    # (or may not) materialise, with the cause and the recommendation `core.user_tables` derives.
    # ⚠ NOT a bucket — `apply_plan` iterates `TABLES` and asks `if bucket in built`, so a key that
    # is not a bucket name is inert there, exactly as `problems` has always been.
    built["limits"] = limits
    return built


def apply_plan(rt, built, username="automation", today=None):
    """Create-or-merge every table in `built` and its rows. Idempotent by construction.

    Row ids ARE the Odoo ids, so a re-run updates in place and never appends a second copy of the
    same record — which is also what makes "every Odoo unique id is in the database" a checkable
    statement rather than a hopeful one.

    ⚠ ONLY THE BUCKETS PRESENT ARE WRITTEN, so a caller (or a gate) may hand in a subset.
    """
    if built.get("problems"):
        raise Refused("; ".join(built["problems"]))
    stamp = today or _iso_today()
    written = {}
    plans = [(key, label, fields(), built[bucket])
             for bucket, key, label, fields in TABLES if bucket in built]

    # ⭐⭐ ONE SYNC WRITE FOR ALL FOUR TABLES, not one per table — measured, not tidied.
    #
    # ⛔ A `flush="sync"` update of `user_tables` is a FULL DOWNLOAD of the document plus a full
    # UPLOAD of it (`Store.update` -> `_read_strict` -> `put`). Four of them against the 20.6 MB
    # document these tables produce is ~165 MB of Hugging Face traffic and four dataset commits
    # EVERY resync — and `main.py` runs this at boot and after every `sync_all()` (~30 min).
    # Composed into one pass it is ~41 MB and one commit: the same rows, a quarter of the bill.
    #
    # ⭐ AND IT IS ATOMIC, WHICH IS THE BIGGER WIN. `_ensure_table_inplace` raises `Refused` at
    # `MAX_TABLES`; with four separate writes that refusal landed AFTER earlier tables had
    # already been committed, leaving exactly the half-spawn `plan()` opens by refusing to
    # create. Inside one updater, a raise aborts before anything is persisted.
    def _apply_all(cur):
        cur = cur if isinstance(cur, dict) else {}
        for key, label, fields, rows in plans:
            written[key] = _ensure_table_inplace(cur, key, label, fields, rows, username, stamp)
        return cur

    rt.update(_ut().STORE_KEY, _apply_all, flush="sync")
    return written


def _ensure_table(rt, key, label, fields, rows, username, stamp):
    """One table, written on its own. Kept because the gate drives a single table directly, and
    because a caller with one table to reconcile should not have to compose an updater."""
    written = {}

    def _one(cur):
        cur = cur if isinstance(cur, dict) else {}
        written["counts"] = _ensure_table_inplace(cur, key, label, fields, rows, username, stamp)
        return cur

    rt.update(_ut().STORE_KEY, _one, flush="sync")
    return written["counts"]


def _ensure_table_inplace(cur, key, label, fields, rows, username, stamp):
    """One table INSIDE a caller's updater: definition merged, rows reconciled, dict mutated.

    ⚠ ROWS THAT LEFT THE POPULATION ARE REMOVED, and that stayed correct through the widening —
    but only because the populations widened to "everything Odoo has". While `ut_odoo_customers`
    was built FROM open invoices, removal meant a customer who paid their bill vanished from the
    registry; now a partner leaves only when their last document does. The shrink guard in
    `plan()` is the backstop for the case this policy cannot distinguish: a partial mirror read.
    """
    ut = _ut()
    wanted = {r["_id"]: {k: v for k, v in r.items() if k != "_id"} for r in rows}
    for row in wanted.values():
        row["refreshed"] = stamp
    counts = {"added": 0, "updated": 0, "removed": 0, "rows": len(wanted)}
    table = cur.get(key)
    if table is None:
        if len(cur) >= ut.MAX_TABLES:
            # ⚠ `ut_ensure` returns silently at this cap; a silent no-op here would report a
            # successful refresh over a table that does not exist.
            raise Refused(f"{key}: tenant is at MAX_TABLES={ut.MAX_TABLES}; nothing created")
        table = cur[key] = {
            "key": key, "label": label, "source": ut.AUTOMATION_SOURCE,
            "createdBy": username, "created": stamp, "fields": [], "rows": {},
            # recordMode = a LOCKED database (item-3 nomenclature): no human may add or
            # delete records, while fields stay addable. Odoo owns this population.
            "recordMode": ut.AUTOMATION_RECORD_MODE,
        }
    table.setdefault("recordMode", ut.AUTOMATION_RECORD_MODE)
    # W30-T35 — the durable "my rows are not in this document" statement, on the tables no
    # conversion will ever stamp (see `READ_THROUGH_KEYS`). Written on every pass, not
    # `setdefault`: it is derived from the code's own structure, so the code is what it must agree
    # with, and a definition that somehow lost the flag should regain it rather than keep serving
    # an empty grid.
    if key in READ_THROUGH_KEYS:
        table["readThrough"] = True
    have = {str(f.get("key")): f for f in (table.get("fields") or [])}
    for field in fields:
        fkey = str(field.get("key"))
        if fkey not in have:
            table.setdefault("fields", []).append(dict(field))
            continue
        # ⭐ A PRESET FIELD'S CONTRACT IS FORWARD-MIGRATED, not merely created once. The
        # widening moved `payment_state`'s option list and every rollup's conditions; a
        # create-only merge would have left the LIVE table declaring the old contract
        # forever, so the column would render but its filter could not match what is stored.
        # ⚠ Only machine-owned keys are touched — `automation.preset` is the wall — so a
        # column a user added to a locked database is never rewritten.
        stored = have[fkey]
        # ⭐⭐ 2026-08-09 — a column a human has taken over keeps its own definition. Same stamp,
        # same reader (`user_tables.user_edited`) and the same reason as the IG reconciler: the
        # loop below overwrites `rollup` from the shipped contract, so an edited preset rollup on
        # an Odoo database would silently revert at the next boot rebuild.
        if _ut().user_edited(stored):
            continue
        if (stored.get("automation") or {}).get("preset"):
            for prop in ("label", "type", "options", "link", "rollup", "description",
                         "agg", "pinned", "default"):
                if prop in field:
                    stored[prop] = field[prop]
                else:
                    stored.pop(prop, None)
    stored_rows = table.setdefault("rows", {})
    for rid, values in wanted.items():
        current = stored_rows.get(rid)
        if current is None:
            stored_rows[rid] = dict(values)
            counts["added"] += 1
        elif any(str(current.get(k, "")) != str(v) for k, v in values.items()
                 if k != "refreshed"):
            current.update(values)
            counts["updated"] += 1
        else:
            current["refreshed"] = values["refreshed"]
    for rid in [r for r in stored_rows if r not in wanted]:
        stored_rows.pop(rid, None)
        counts["removed"] += 1
    return counts


def is_royal(tenant):
    return str(tenant or "").strip().lower() in RI_SLUGS


def refresh(rt, tenant, username="automation", cur=None, today=None):
    """THE entry point — the store-resync path and the route both call this.

    ⚠ It must be CALLED on resync by something outside this file. If it is not wired, every row
    still carries a `refreshed` stamp, so a stale worklist is at least LEGIBLE rather than
    silently authoritative.
    """
    if not is_royal(tenant):
        raise Refused(f"tenant {tenant!r} has no Odoo mirror behind these tables (R1: Royal "
                      f"Imports only); refusing to spawn empty locked databases")
    # ⭐⭐ W31-T45 / D-169 — THE SLUG GATE ABOVE AND THE FILE GATE HERE ANSWER DIFFERENT QUESTIONS,
    # and this is the one place in the codebase where that is easy to miss. `is_royal` asks "is
    # this tenant ENTITLED to Odoo databases"; it says nothing about WHICH DuckDB file this process
    # has open. A worker pinned to another tenant's store (AIOS_DUCKDB_PATH, or a `use_path` in a
    # provisioning script) passes `is_royal("royal-imports")` and then WRITES tenant #0's locked
    # databases from another customer's rows — a spawn, not a read, so the wrong numbers become
    # durable. Entitlement is not residency.
    # ⚠ It runs when `cur` is LENT too, not only when we open one: the resync loop and the boot
    # rebuild both hand a cursor in, and a lent cursor is exactly the case where nobody re-checks.
    if rt is not None:
        rt.assert_datastore_matches()
    # ⛔⛔ W32-T16 / R10 — A DISCONNECTED CONNECTOR DOES NOT REFRESH, AND THAT IS THE WHOLE FREEZE.
    # R10: *"Disconnect removes the credential and FREEZES the grids as static data."* Removing
    # the credential alone is not a freeze — this function is also reached by the boot rebuild and
    # the resync loop, and for tenant #0 the ENVIRONMENT still holds Odoo credentials, so a
    # disconnected workspace would silently re-materialise from `.env` on the next tick and the
    # "disconnect" would last until the container restarted. The refusal is a REPORT, not a raise:
    # the resync loop calling this every cycle must not be handed an exception as a status.
    if rt is not None and frozen(rt):
        return {"tables": {}, "frozen": True,
                "note": "this workspace has disconnected Odoo; its databases are frozen as "
                        "static data and are not being refreshed"}
    if cur is None:
        from harness import datastore
        cur = datastore.ro_con()
    built = plan(cur, rt=rt)
    # ⭐ W32-T15 / R9 — THE GRID PICKER, ENFORCED WHERE IT COUNTS. `apply_plan` writes only the
    # buckets present in `built`, so dropping an unticked one here is the whole of "unticking a
    # grid stops it materialising on the next sync". Done AFTER `plan` rather than inside it so
    # every cap, refusal and limit report is still computed over the full set — a config must not
    # be able to hide a problem by hiding the table that has it.
    # ⚠ It does NOT delete a grid that was already spawned. Unticking stops the next refresh from
    # rewriting it; dropping the rows a tenant already has is `disconnect`'s job, and it does not
    # do that either (R10 keeps them). Silent data deletion behind a checkbox is not on offer.
    if rt is not None:
        keep = enabled_buckets(rt)
        skipped = sorted(key for bucket, key, _l, _f in TABLES if bucket not in keep)
        for bucket, _key, _l, _f in TABLES:
            if bucket not in keep:
                built.pop(bucket, None)
    else:
        skipped = []
    written = apply_plan(rt, built, username=username, today=today)
    return {"tables": written,
            # R6's second sentence: a set that was deliberately not built SAYS SO, with the keys.
            **({"skipped": skipped} if skipped else {}),
            **{bucket: len(built[bucket]) for bucket, _k, _l, _f in TABLES if bucket in built}}