api/routes_odoo_tables.py

"""routes_odoo_tables.py — the door to the Odoo relational spawn (wave 27 item 17, contract C8).

Two endpoints and no cleverness: refresh the locked Odoo databases, and report their freshness.
The work itself lives in `odoo_relational.py`; this file only decides WHO may ask and turns a
refusal into a status code.

⛔ THE MOUNT IS DONE (`main.py` includes this router) and the wave-23 scar it was written against
— three finished routers shipping 404-dead behind green gates — is covered by `verify_api`
enumerating `main.app.routes`. ⚠ BUT WAVE 28 PROVED THAT CONTROL IS ONLY HALF OF THE QUESTION:
being mounted is not being CALLABLE. This router was mounted, enumerated, green, and answered a
plain-text 500 to every request for a day because of an attribute typo in the admin check
(`session.is_admin`, which does not exist). A route-existence check cannot see that; only calling
it can. `verify_api` now does both.

ADMIN-GATED, and not for tidiness: a refresh REWRITES four locked databases for the whole tenant
and deletes the rows that left the population. That is an operator action.
⭐ WAVE 30 (R6/R7, contract C2) ADDS A THIRD ENDPOINT AND IT IS A DIFFERENT KIND OF THING: the
READ-THROUGH WINDOW. The two above operate on the materialised copy; `/{table_key}/rows` does not
read the copy at all. See the block above `GRID_SOURCES` for why that had to change.
"""
import json

from fastapi import Depends
from fastapi import APIRouter, Query

from deps import Session, err, require_session

router = APIRouter(prefix="/api/v1")


def _rel():
    import odoo_relational
    return odoo_relational


def _rt():
    from harness import runtime
    return runtime


# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⭐⭐ W31-T45 / D-169 — THE CROSS-TENANT MIRROR GUARD, AND WHY EVERY DOOR BELOW GOES THROUGH IT.
# ═════════════════════════════════════════════════════════════════════════════════════════════
#
# `harness/datastore` holds ONE process-wide DuckDB connection over a process-global `DB_PATH`,
# and ONE Space process serves EVERY tenant. `TenantRuntime.assert_datastore_matches` was written
# for exactly the failure that arrangement produces — its own docstring names it: *"an analytical
# read served from another tenant's file returns real, plausible, wrong rows … and the number
# reconciles against the wrong book"* — and until this wave its only caller anywhere was
# `verify_api.py`. The guard was correct, tested, and governed nothing.
#
# ⛔ WHAT ACTUALLY HELD THE BOUNDARY WAS THE CUSTOMER LIST, NOT THE CODE. Three of the four doors
# below are reachable only by a tenant whose own `user_tables` document declares the table
# (`_defn_or_refuse`) or whose slug passes `is_royal` — so no second tenant could reach the mirror
# because no second tenant had mirror databases. R2 puts Meta Ads on this same mirror for GTM Lab,
# which ends that fact. A boundary that holds because of who our customers are is not a boundary.
#
# ⚠ AND ONE DOOR WAS ALREADY THROUGH IT TODAY — measured, not theorised: `/odoo-tables/status`
# takes a mirror cursor with NO tenant gate at all (`is_royal` appears only as an `applicable`
# FIELD in its response), so any authenticated tenant reached tenant #0's file and was served row
# counts from it. That is D-169 firing in production, and it is what `_mirror_cur` closes.
def _mirror_cur(session):
    """A mirror cursor for THIS session's tenant, or a refusal that says which failure it is.

    Returns `(cursor, mismatch_sentence)` — exactly one of the two is None, so a caller cannot
    accidentally treat "refused" as "empty".

    ⛔ IT CATCHES ONLY THE MISMATCH AND LETS THE OTHER `RuntimeError` THROUGH, because the two mean
    opposite actions and every caller already has a policy for the second one:

      * `DatastoreMismatch` — this process has ANOTHER tenant's file open. It never fixes itself by
        waiting; the deployment is mis-pinned. Returned as a SENTENCE so each door can choose its
        own status (the row doors raise, the status door reports — R6's second sentence).
      * a plain `RuntimeError` — the mirror is completing its first sync, and retrying works. It
        PROPAGATES, so `odoo_table_rows` keeps answering `503 store_not_ready` and the status door
        keeps degrading to "no size half", exactly as each did before this guard existed.

    ⚠ Collapsing them was the tempting simplification and it is the defect: `DatastoreMismatch`
    SUBCLASSES `RuntimeError`, so a single `except RuntimeError` here would render every
    cross-tenant refusal as a retry banner. Catching the subclass first is a correctness rule.
    """
    rtm = _rt()
    try:
        return rtm.mirror_cursor(session.runtime), None
    except rtm.DatastoreMismatch as e:
        return None, str(e)


@router.post("/odoo-tables/refresh")
def refresh_odoo_tables(session: Session = Depends(require_session)):
    """Rebuild all four locked Odoo databases from the tenant's mirror — customers, products,
    invoices and orders (it was invoices + customers until 2026-08-09).

    Idempotent: row ids are Odoo ids, so a re-run updates in place. Returns what MOVED, because
    "refreshed" with no counts is indistinguishable from a no-op over an empty mirror.
    """
    # ⛔ `deps.err()` RETURNS an HTTPException, it does not raise one — so it must be `raise
    # err(...)`. `return err(...)` serialises the exception object with a **200**, which is the
    # wave-23 shape exactly: a finished route, green everything, wrong on the wire. The house
    # convention is 243 `raise err` against 6 strays; this file uses `raise`.
    rel = _rel()
    # ⛔⛔ THIS LINE WAS `session.is_admin` AND IT IS WHY D-107 LOOKED LIKE A STORE PROBLEM FOR A
    # DAY. `Session` is a plain dataclass with ONE admin accessor, the `.admin` property; there is
    # no `is_admin` and no `__getattr__`, so the attribute lookup raised `AttributeError` — SEVEN
    # LINES ABOVE the `try:` below, for every caller, admin or not. That is the whole explanation
    # for the measured signature: a bare `500` carrying Starlette's default PLAIN-TEXT body
    # instead of our JSON envelope, on a route whose own handler had just been taught to name the
    # exception. The refresh never ran, never reached `plan()`, never touched the mirror.
    # ⚠ The tell was in the diagnosis all along and was read as evidence about the STORE: "it is
    # not a timeout, the same operation takes 16.8 s from a laptop". Correct, and the reason was
    # that the request never got as far as doing any work at all.
    # ⭐ It was the SOLE `session.is_admin` in the repo against ~40 `session.admin` — an
    # unmounted-shaped defect that no gate could see, because `verify_api` pinned these two routes
    # as MOUNTED and never CALLED them. That gate now calls them; see `verify_api`'s odoo-tables
    # section and its negative control.
    if not session.admin:
        raise err(403, "forbidden", "Refreshing the Odoo databases is an admin action.")
    if not rel.is_royal(session.tenant):
        # A plain 400 with the reason: this tenant has no Odoo mirror, and spawning two empty
        # locked databases it can never fill would be worse than refusing.
        raise err(400, "not_applicable",
                  "These databases are built from the Royal Imports Odoo mirror (R1).")
    try:
        out = rel.refresh(session.runtime, session.tenant, username=session.user)
        # ⭐⭐ W30-T31 — AND THEN THE ROWS THAT SHOULD NOT BE HERE LEAVE AGAIN.
        #
        # The spawn writes every bucket's rows straight into the tenant document by mutating it
        # inside its own updater, so no guard in `core.user_tables` is on that path (measured, not
        # assumed: `_ensure_table_inplace` never calls a function there). Stripping AFTER the write
        # is what makes "a read-through grid stores no rows" true rather than intended — and the
        # refresh is the only moment a stripped table can come back.
        # ⚠ Reported in the response, because a silent 7 MB moving in or out of a tenant's
        # document is exactly the kind of thing an operator should be able to see happening.
        sync_read_through()
        moved = _ut().strip_materialised(st=session.runtime)
        return {"ok": True, **out, "unmaterialised": moved}
    except rel.Refused as e:
        # A refusal is the ANSWER, not a crash: the caller must see WHY nothing was written.
        raise err(409, "refused", str(e))
    except RuntimeError as e:
        # `datastore.ro_con()` raises this while the store is completing its first sync. Relaying
        # it beats writing a partial table from a half-synced mirror.
        raise err(503, "store_not_ready", str(e))
    except Exception as e:                                            # noqa: BLE001
        # ⛔ AN UNEXPECTED FAILURE MUST STILL SAY WHAT IT WAS. Measured live 2026-08-09: this
        # route answered a bare `500` and the operator had no way to learn why — the reason lived
        # only in a container log nobody can reach from the product. A spawn that rewrites four
        # locked databases is exactly the operation whose failure needs a sentence.
        # ⚠ The TYPE is included deliberately: `BinderException: Referenced column "agent_id" not
        # found` is a different action (wait for the column backfill) from a timeout or an auth
        # failure, and "500" cannot tell them apart.
        raise err(500, "refresh_failed", f"{type(e).__name__}: {e}")


@router.get("/odoo-tables/status")
def odoo_tables_status(session: Session = Depends(require_session)):
    """Row counts + the newest `refreshed` stamp per table.

    ⚠ The stamp is the whole point while the resync wiring is outstanding: a locked collections
    worklist that quietly stopped updating is a worklist that lies, so its age must be readable
    without anybody running a refresh to find out.
    """
    rel, out = _rel(), {}
    import core.user_tables as user_tables
    # ⭐ W30-T31 — THE ELIGIBILITY PASS RUNS HERE TOO, and the mirror cursor is taken best-effort:
    # the freshness surface must answer on a box with no mirror (that is what it is FOR), so a
    # store that is not ready costs the size half of the question, never the whole endpoint.
    # ⭐⭐ W31-T45 / D-169 — THIS IS THE DOOR THAT WAS ALREADY LEAKING, and it leaked as COUNTS
    # rather than as rows, which is why nothing caught it. There is no tenant gate above this
    # line: `rel.is_royal(session.tenant)` appears once, as an `applicable` FIELD in the response
    # 60 lines below. So a nurilab or gtmlab session took a cursor on tenant #0's `royal.duckdb`,
    # `_source_for` read its columns and `_ds.window` counted its rows — real, plausible numbers
    # belonging to another customer, returned 200 OK.
    # ⛔ IT REPORTS RATHER THAN RAISING, deliberately, and the distinction is R6's second sentence:
    # "which Odoo databases exist for me" is a legitimate question for any tenant and its honest
    # answer here is *none* — a 409 would be refusing the question instead of the leak. So the
    # mirror half is refused, `rowsFrom` says so per table, and the refusal rides the payload with
    # its cause. A count that silently became `-1` would be the silent truncation R6 bans.
    cur, mirror_refused = None, None
    try:
        cur, mirror_refused = _mirror_cur(session)
    except Exception:                                                 # noqa: BLE001
        pass                    # mid-first-sync (or no mirror at all) — the document half answers
    eligible, why_not = sync_read_through(cur)
    # ⚠ ITERATES `rel.TABLES`, NEVER A LITERAL PAIR. It read `(INVOICES_KEY, CUSTOMERS_KEY)`
    # while those were the only two; the 2026-08-09 widening added products and orders, and a
    # hard-coded list here would have reported "everything is fine" over two databases it had
    # stopped knowing about ([[gate-answers-the-wrong-question]] — the hard-coded count).
    for _bucket, key, _label, _fields in rel.TABLES:
        table = user_tables.get(key, st=session.runtime) or {}
        rows = (table.get("rows") or {})
        stamps = [str((r or {}).get("refreshed") or "") for r in rows.values()]
        # ⛔⛔ A FRESHNESS SURFACE THAT LIES IS WORSE THAN NO FRESHNESS SURFACE, and this endpoint
        # was one step from becoming one. Both numbers below are derived from `table["rows"]`, so
        # the moment a grid stops materialising they would read `rows: 0, refreshed: ""` — an
        # operator would see a database that looks EMPTY and STALE on a route whose own docstring
        # says a worklist that quietly stopped updating must be legible without a refresh. So a
        # read-through table is counted from the MIRROR and says where its count came from.
        materialised = user_tables.materialises(key, st=session.runtime)
        if not materialised:
            try:
                rows = {}
                stamps = []
                spec = _source_for(cur, key) if cur is not None else None
                if spec is not None:
                    from harness import datastore as _ds
                    frm = ({"from_sql": spec["from_sql"]} if spec.get("from_sql")
                           else {"table": spec["table"]})
                    n = _ds.window(select="1", where=spec.get("where") or "",
                                   order_by=spec["id"], limit=1, cur=cur, **frm)["total"]
                else:
                    n = -1                    # unknown, and never reported as zero
            except Exception:                                         # noqa: BLE001
                n = -1
        out[key] = {
            "exists": bool(table),
            "rows": (len(rows) if materialised else n),
            # ⭐ W31-T45: a refused mirror says `refused`, never `mirror` — a size that came from
            # nowhere must not be labelled with the source it did not come from.
            "rowsFrom": ("document" if materialised
                         else ("refused" if mirror_refused else "mirror")),
            "materialised": materialised,
            # why this grid still keeps its rows in the tenant document, when it does not have to
            "materialisedBecause": (why_not.get(key, "") if materialised else ""),
            "refreshed": (max([s for s in stamps if s], default="") if materialised
                          else "live — read through the mirror"),
            "locked": table.get("recordMode") == user_tables.AUTOMATION_RECORD_MODE,
            # W30/R7: whether this database is served THROUGH the mirror rather than from the copy
            # above. Reported per table, because they convert one at a time and an operator
            # reading `rows: 0` needs to know whether that means "empty" or "not stored here".
            "readThrough": key in _sources(),
        }
    # ⭐ W30-T32 / R6's SECOND SENTENCE, APPLIED TO OUR OWN HALF-BUILT STATE. A read-through
    # binding whose FIELD DECLARATION has not landed yet answers 404 on the rows route, and an
    # operator would read that as "the grid does not exist" rather than as "half of it shipped".
    # These are the two line grains: bound here, declared in `odoo_relational` by W30-T35.
    # ⚠ It reports the KEYS, never a field list — inventing a contract here is exactly the second
    # source of truth the ticket forbids.
    pending = {k: {"readThrough": True, "declared": False,
                   "cause": "this connected grid has a read-through binding but no field "
                            "declaration in odoo_relational yet, so it cannot be opened",
                   "recommendation": "declare its fields + a TABLES row (W30-T35); the binding "
                                     "and the window are already live"}
               for k in _sources() if k not in out}
    return {"ok": True, "applicable": rel.is_royal(session.tenant), "tables": out,
            "bound_not_declared": pending,
            # ⭐ W31-T45 / D-169 — R6's SECOND SENTENCE, ON A GUARD RATHER THAN ON A ROW CAP. A
            # limit that genuinely cannot be removed must be REPORTED with its cause and a
            # recommended fix; silence is the violation. `null` when the mirror answered.
            "mirrorRefused": (
                {"cause": mirror_refused,
                 "effect": "sizes_unavailable",
                 "recommendation": "this deployment has another tenant's analytical store open; "
                                   "sizes are withheld rather than read from it. Pin the process "
                                   "to this tenant (AIOS_DUCKDB_PATH / datastore.use_path) to "
                                   "restore them."}
                if mirror_refused else None),
            # ⭐ W31-T46 / D-160 — WHY THERE IS NO MIRROR, when there is none. The last time this
            # was silent it read as a pinned tag and a store problem for days. `null` when the
            # container has one; a sentence with a cause and a fix when it does not.
            "mirrorSeed": _seed_state()}


def _seed_state():
    """`main.MIRROR_SEED` as a reportable block, or None when the mirror is present.

    ⚠ IMPORTED LAZILY AND FAIL-QUIET: `main` imports this router, so a module-level import here
    would be a cycle, and a freshness surface must never 500 because a diagnostic was unavailable.
    """
    try:
        import main as _main
        from harness import datastore as _ds
        if _ds.DB_PATH.exists():
            return None
        state = dict(_main.MIRROR_SEED)
        if not state.get("cause"):
            state["cause"] = "this container has no analytical mirror yet"
            state["recommendation"] = ("the boot seed runs independently of AIOS_PREWARM; if this "
                                       "persists, check HF_TOKEN on the deployment")
        return state
    except Exception:                                                 # noqa: BLE001
        return None


# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⭐⭐ THE READ-THROUGH WINDOW — owner ruling R6 ("no cap on connected-source data") via R7.
# ═════════════════════════════════════════════════════════════════════════════════════════════
#
# THE PROBLEM IT REPLACES, in the owner's numbers. `core.user_tables.MAX_ROWS = 60_000` bounds a
# `ut_*` table because ALL of a tenant's tables live in ONE JSON document that `Store.get`
# deep-copies per request (D-87: four Odoo databases = 20.7 MB, ~370 ms a copy). Orders at
# **32,826** is already 55% of that ceiling; order lines (**255,286**) are 4.3× over it and GL
# lines (**971,034**) 16× over, so those two could never be grids at all — `odoo_relational.plan`
# refuses them rather than truncating, which is the correct refusal of the wrong architecture.
#
# ⭐ "NO CAP" IS NOT A BIGGER NUMBER. Raising `MAX_ROWS` would make every request slower for every
# tenant, including the ones who never open an Odoo grid. The mirror ALREADY holds all of it
# uncapped — measured on this box: 963,783 GL lines counted in ~24 ms — so the fix is to stop
# copying and serve a WINDOW: the requested slice, plus a `SELECT count(*)` that tells the truth
# about the whole.
#
# ⛔⛔ `total` IS NEVER `len(rows)`. It comes from `datastore.window`'s own count statement over
# the SAME predicate. A window whose count is its own length is a fabricated aggregate wearing an
# authoritative face ([[no-unverifiable-aggregates]]).
#
# ⛔ PREDICATES PUSH DOWN, and this is the half a client cannot be trusted with. A filter chip
# evaluated over the 200 rows that happen to be in memory would report "40 matches" out of
# 971,034 — [[one-question-two-normalizers]] at scale. `harness/filter_sql.py` compiles OUR filter
# vocabulary to SQL and is the same evaluator the TS engine is held in step with, so the fold and
# the display answer one question.
#
# ⭐ THE SPEC BELOW IS A SECOND STATEMENT OF SOMETHING `odoo_relational`'s READER ALREADY SAYS,
# AND THAT IS THE REAL RISK HERE — not SQL injection. It is gated rather than trusted:
# `verify_scopes.section_read_through` runs the REAL reader over the REAL mirror as an ORACLE and
# asserts the windowed path agrees ROW FOR ROW and TOTAL FOR TOTAL. A binding that drifts from its
# reader goes red; a bucket with no binding is REPORTED, never silently served from the copy as
# though it were read-through ([[one-evaluator-per-question]], [[gate-answers-the-wrong-question]]).
#
# ⚠ ONE BUCKET IS BOUND HERE (orders). The other seven still serve from the materialised copy and
# say so on the wire (`readThrough: false`), because R6's second sentence — *"if there is lag or
# it can't be done, you need to explicitly tell me why and recommend a fix"* — makes an unconverted
# grid something to REPORT, not something to leave looking converted. Adding one is a spec row plus
# a green oracle check.

#: `ut_*` key -> how to read that grid straight out of the mirror.
#:
#: `where`/`select` are SQL WE author (never a request value); every request value is bound through
#: `params` by `filter_sql`. `cols` maps a FIELD KEY (what `odoo_relational.<x>_fields()` declares,
#: and what a saved view's filters name) to `(sql expression, coercion)`. The coercion reproduces
#: the reader's own python cast, by CALLING the reader's helpers where one exists — `_as_date` and
#: `_in_scope` are imported, not re-implemented.
#: Keys a reader does not produce (link columns, `refreshed`) are absent here on purpose: they are
#: filled by the grid at render time exactly as they are on the materialised path.
GRID_SOURCES = {}

#: ⭐⭐ W31-T49 / C4 — THE SOURCE PROVIDERS. Odoo builds its specs below; ANY OTHER connector adds
#: its own by registering a builder here, and `_sources()` folds them into the one registry every
#: door already reads.
#:
#: ⛔ THIS IS THE SEAM THAT KEEPS "ONE REGISTRY" TRUE WHILE THE FILE STAYS ODOO-NAMED. The
#: alternative — a second `META_GRID_SOURCES` consulted beside this one — would give the platform
#: two answers to *"how is this connected grid read?"*, and every consumer (`_source_for`,
#: `whole_pool`, `population`, `odoo_tables_status`, `routes_tables.scoped_pids`) would have to
#: learn both or silently serve one. That is [[one-question-two-normalizers]] on the read path.
#: ⚠ A provider registers a CALLABLE, not a dict, so its module is not imported until the first
#: read — the same lazy rule `_sources` already follows for `odoo_relational`.
_SOURCE_PROVIDERS = []


def register_source_provider(fn):
    """Add a `() -> {table_key: spec}` builder to the connected-grid registry.

    Additive and idempotent by identity, so a re-import cannot double-register. Returns the number
    of providers now known — a caller that wants to assert its registration took has a number.
    """
    if callable(fn) and fn not in _SOURCE_PROVIDERS:
        _SOURCE_PROVIDERS.append(fn)
    return len(_SOURCE_PROVIDERS)


def _sources():
    """Build `GRID_SOURCES` lazily by asking EVERY registered provider — Odoo is one of them.

    ⛔⛔ ODOO GOES THROUGH THE SEAM TOO, AND THAT IS THE POINT OF W31-T49 RATHER THAN A FLOURISH.
    The first version built Odoo's specs inline here and folded other providers in afterwards,
    which quietly says "Odoo is the registry and everyone else is an addendum" — two mechanisms
    for one question, with the second one exercised by nobody until a connector arrives. It also
    left `register_source_provider` with no production caller at all, which `verify_reachability`
    correctly reddened as a capability behind no door ([[artifact-with-no-importer]]). Registering
    the incumbent through its own seam makes the seam load-bearing from the first request.

    ⚠ STILL LAZY, for the original reason: `_odoo_sources` names `odoo_relational` constants, and
    importing that module at file-import time would drag the Odoo layer into every process that
    mounts a router. A provider is a CALLABLE precisely so it stays unimported until first read.
    """
    if GRID_SOURCES:
        return GRID_SOURCES
    for build in list(_SOURCE_PROVIDERS):
        try:
            extra = build() or {}
        except Exception:                                             # noqa: BLE001
            # ⛔ ONE PROVIDER'S FAILURE COSTS ITS OWN GRIDS, NEVER THE PAGE. The Odoo grids must
            # not go dark because a newer connector's module is unhappy — and vice versa.
            continue
        if extra:
            GRID_SOURCES.update(extra)
            _ut().register_connected(*extra)
    return GRID_SOURCES


def _odoo_sources():
    """Odoo's read-through bindings — `{table_key: spec}`. Registered as a provider below.

    ⚠ IT RETURNS a dict rather than mutating the module global: `_sources` owns the merge, so a
    provider that half-built its specs and raised cannot leave a partial registry behind.
    """
    rel = _rel()
    # ⭐ R6 / W30-T29 — TELL THE STORE LAYER WHICH DATABASES ARE CONNECTED, so `MAX_ROWS` stops
    # being a fact about them. `core` never imports up, so the declaration goes this way round.
    # ⚠ ALL EIGHT, not just the read-through-bound ones — every row in these tables comes from
    # Odoo, which is what R6 is about; being served from the stored copy today is our conversion
    # state, not a property of the data. `_sources`' own `register_connected` covers only the keys
    # a provider RETURNS, so this call is not redundant with it and must not be folded into it.
    _ut().register_connected(*[key for _b, key, _l, _f in rel.TABLES])
    GRID_SOURCES = {}                       # the LOCAL registry this builder fills and returns
    _s, _i, _n = ((lambda v: str(v or "")), (lambda v: int(v or 0)),
                  (lambda v: float(v or 0.0)))
    GRID_SOURCES[rel.ORDERS_KEY] = {
        "table": "sale_order",
        # ⛔ IMPORTED, NOT RETYPED. `_CONFIRMED` is the fixed wholesale scope; if it ever changes,
        # this window changes with it and the oracle check proves it did.
        "where": f"{rel._CONFIRMED} AND partner_id IS NOT NULL",
        "id": "id",
        "needs_excluded": True,          # `wholesale_scope` is resolved against the excluded set
        "cols": {
            "order_no": ("name", _s),
            "odoo_id": ("id", _i),
            "customer": ("partner_name", _s),
            rel.JOIN_KEY: ("partner_id", _i),
            "order_date": ("date_order", rel._as_date),
            "amount_untaxed": ("amount_untaxed", _n),
            "team": ("team_name", _s),
            "state": ("state", _s),
            "invoice_status": ("invoice_status", _s),
            # ⭐ THE SCOPE COLUMN BECOMES REAL SQL, which is the point. On the materialised path it
            # is `_in_scope(pid, excluded)` — a python set test, and a filter on it therefore could
            # not push down. Inlining the ids (ints, from our own query) makes it a column the
            # mirror can filter and sort on, so the R6 "limit" it would otherwise have earned does
            # not exist. `{excluded}` is substituted by `_source_for` below.
            "wholesale_scope": ("CASE WHEN partner_id IN ({excluded}) THEN '' ELSE '1' END", _s),
        },
    }

    # ═════════════════════════════════════════════════════════════════════════════════════════
    # ⭐⭐ W30-T32 — THE TWO LINE GRAINS. These are the grids R6 exists for: they have never had a
    # `ut_*` table and never can, at any cap. MEASURED on this box's mirror, warm:
    #   sale_order_line     254,189 in the confirmed scope (256,810 unscoped) — 63.9 MB as JSON
    #   account_move_line   963,783                                           — ~240 MB as JSON
    # Against `MAX_ROWS = 60_000` that is 4.2x and 16x, and against the 32 MB per-table document
    # budget it is 2x and 7.5x. Read THROUGH, both serve a page in 134–166 ms.
    #
    # ⛔ THE FIELD KEYS BELOW ARE HALF OF A CONTRACT AND `odoo_relational` OWNS THE OTHER HALF
    # (W30-T35, session E). `cols` binds a field key to SQL; the field's label, type and order are
    # DECLARED THERE, once. Until that declaration lands the route answers 404 for these two keys
    # (`rel.TABLES` has no entry), and `odoo_tables_status` REPORTS them as bound-not-declared
    # rather than leaving them invisible — R6's second sentence applied to our own conversion.
    # ⚠ A key here with no declaration there is a silent NO-CELL (`rows_from_pool` projects
    # strictly); a key there with no binding here is an INACTIVE filter leaf, which WIDENS. The
    # two lists are checked against each other by `verify_scopes.section_line_grids`.
    ol_key = getattr(rel, "ORDER_LINES_KEY", "ut_odoo_order_lines")
    gl_key = getattr(rel, "GL_LINES_KEY", "ut_odoo_gl_lines")
    _ut().register_connected(ol_key, gl_key)
    # ⛔ THE SCOPE IS THE ORDER'S, AND THE LINE TABLE CANNOT ANSWER IT ALONE: `sale_order_line`
    # carries no `state` (12 columns, measured), so the confirmed-order scope — and `order_date`,
    # and the order NAME a person reads the grid by — only exist across the join. MEASURED, warm,
    # best of two: the JOIN beats `order_id IN (SELECT id FROM sale_order WHERE …)` at both depths
    # (166 / 483 ms against 237 / 565 ms at offset 0 / 200,000), so the shape is chosen on a
    # number rather than on taste.
    # ⚠ `_CONFIRMED` is IMPORTED and QUALIFIED, never retyped — it opens with the bare column
    # `state`, which `sale_order_line` does not have, so the prefix is what keeps it unambiguous
    # if that table ever gains one. `section_line_grids` counts the same population a second way
    # (a subquery, not a join) and the two must agree, which is what catches a mis-qualification.
    GRID_SOURCES[ol_key] = {
        "from_sql": "(sale_order_line sol JOIN sale_order so ON so.id = sol.order_id)",
        "tables": {"sol": "sale_order_line", "so": "sale_order"},
        "where": f"so.{rel._CONFIRMED}",
        "id": "sol.id",
        "needs_excluded": True,
        "cols": {
            "odoo_id": ("sol.id", _i),
            "order_no": ("so.name", _s),
            "order_id": ("sol.order_id", _i),
            "customer": ("sol.order_partner_name", _s),
            rel.JOIN_KEY: ("sol.order_partner_id", _i),
            "product": ("sol.product_name", _s),
            rel.PRODUCT_JOIN_KEY: ("sol.product_id", _i),
            "qty": ("sol.product_uom_qty", _n),
            "price_subtotal": ("sol.price_subtotal", _n),
            "margin": ("sol.margin", _n),
            "purchase_price": ("sol.purchase_price", _n),
            "order_date": ("so.date_order", rel._as_date),
            "state": ("so.state", _s),
            "wholesale_scope": (
                "CASE WHEN sol.order_partner_id IN ({excluded}) THEN '' ELSE '1' END", _s),
        },
    }
    # ⚠ UNSCOPED ON PURPOSE, and it is a decision rather than an omission: every other grid here
    # carries a fixed scope, but a GENERAL LEDGER whose draft and cancelled entries are invisible
    # is a ledger that cannot be reconciled. `parent_state` rides as a column so a person filters
    # in SQL over all 963,783 rows instead of us choosing for them. (Posted-only is 944,846.)
    # The join to `account_account` is what makes `account_code` — the key `ut_odoo_accounts` is
    # linked on — available at all; the mirror flattens `account_id`/`account_name` onto the line
    # but not the CODE, and 192 accounts hash-join for free.
    GRID_SOURCES[gl_key] = {
        "from_sql": ("(account_move_line aml LEFT JOIN account_account aa "
                     "ON aa.id = aml.account_id)"),
        "tables": {"aml": "account_move_line", "aa": "account_account"},
        "where": "",
        "id": "aml.id",
        "needs_excluded": True,
        "cols": {
            "odoo_id": ("aml.id", _i),
            "entry": ("aml.move_name", _s),
            "move_id": ("aml.move_id", _i),
            "account": ("aml.account_name", _s),
            rel.ACCOUNT_JOIN_KEY: ("aa.code", _s),
            "customer": ("aml.partner_name", _s),
            rel.JOIN_KEY: ("aml.partner_id", _i),
            "date": ("aml.date", rel._as_date),
            "debit": ("aml.debit", _n),
            "credit": ("aml.credit", _n),
            "balance": ("aml.balance", _n),
            "line_type": ("aml.display_type", _s),
            "move_type": ("aml.move_type", _s),
            "parent_state": ("aml.parent_state", _s),
            "wholesale_scope": (
                "CASE WHEN aml.partner_id IN ({excluded}) THEN '' ELSE '1' END", _s),
        },
    }
    # ⭐ W30-T31 — THE GL ACCOUNT REGISTRY, BOUND BECAUSE IT IS THE ONE GRID THAT CAN ACTUALLY
    # STOP MATERIALISING TODAY. 192 rows, nothing folds it, nothing links at it, and it fits
    # inside one window — the three conditions `_unmaterialisable` checks. It is small, and that
    # is the point: it is the first shipped database whose rows are NOT in the tenant document,
    # so the stratum is proven on a real table instead of on a mechanism with no subject.
    # ⚠ `account_fields()` declares six columns and `read_accounts` is one `cur.execute` over
    # `account_account`; `section_read_through`'s differential oracle holds this binding to it.
    # ⚠ `is_expense` REPRODUCES `read_accounts`' predicate IN SQL rather than inventing one, and
    # that predicate is itself a copy of the semantic layer's `gl_lines` scope. Three statements of
    # one rule is two too many, but the reader's own comment explains why it is copied rather than
    # imported, and `section_read_through`'s differential oracle is what keeps this one honest.
    GRID_SOURCES[rel.ACCOUNTS_KEY] = {
        "table": "account_account",
        "where": "",
        "id": "id",
        "cols": {
            rel.ACCOUNT_JOIN_KEY: ("code", _s),
            "account_name": ("name", _s),
            "odoo_id": ("id", _i),
            "account_type": ("account_type", _s),
            "is_expense": ("CASE WHEN account_type IN ('expense','expense_depreciation') "
                           "THEN '1' ELSE '' END", _s),
        },
    }
    return GRID_SOURCES


#: Odoo registers itself, at import, exactly as any other connector does. ⚠ The order providers
#: are registered in is the order their specs land; keys are namespaced by connector (`ut_odoo_`,
#: `ut_meta_`), so a later provider cannot shadow an earlier one's grid.
register_source_provider(_odoo_sources)


def sync_read_through(cur=None):
    """Register every grid that may stop storing rows, and REPORT why the rest may not.

    Returns `({key: eligible}, {key: reason})`. Idempotent, cheap, and safe to call from any door:
    registration is additive and `strip_materialised` is a no-op once a table is empty.

    ⛔ IT IS ALSO THE ONLY PLACE THAT MAY CALL `register_read_through`, because the eligibility
    question needs BOTH halves — `odoo_relational`'s field declarations (for the fold matrix) and
    the mirror (for the size) — and `core` may import neither.
    """
    eligible, reasons = _unmaterialisable(cur)
    keys = [k for k, ok in eligible.items() if ok]
    if keys:
        _ut().register_read_through(*keys)
    return eligible, reasons


#: ⚠ A mirror can be `ready()` and still lack a column (`ready()` reads entity PHASES; column
#: backfills checkpoint separately) — the gap that already cost a live 500. Every projected
#: expression is checked against the real column list and degraded to a literal, exactly as
#: `odoo_relational._col` does for the reader, so a fresh Space serves a blank cell rather than a
#: DuckDB Binder error.
#:
#: ⛔ W30-T32 — IT TAKES AN ALIAS MAP NOW, AND WITHOUT THAT THE GUARD WAS ABOUT TO GO BLIND. The
#: line-grain grids project `sol.price_subtotal` / `aml.parent_state`, and a dotted string is not
#: `isalnum()`, so the old single-table version returned EVERY qualified expression unchecked —
#: the same "expression, nothing to check" branch that correctly skips a CASE. Both of those
#: columns are 2026-07-28 backfills that a mirror can genuinely be missing, so the blind spot
#: would have surfaced as a bare DuckDB Binder error on a fresh Space, which is precisely the
#: failure this helper exists to prevent ([[gate-answers-the-wrong-question]]).
# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⭐⭐ W30-T31 / D-87 — WHICH CONNECTED GRIDS MAY STOP MATERIALISING, AND WHY MOST MAY NOT YET.
# ═════════════════════════════════════════════════════════════════════════════════════════════
#
# The prize is real: all of a tenant's `ut_*` tables live in ONE document that `Store.get`
# deep-copies on EVERY call, hit or miss, and the four original Odoo grids are 20.7 MB / ~370 ms
# of it. Every permission check in the app pays that.
#
# ⛔ AND YOU CANNOT SIMPLY DELETE THE ROWS, WHICH IS THE FINDING THIS FUNCTION ENCODES. Measured
# mechanically across the eight field declarations, not by eye:
#   * `ut_odoo_invoices` and `ut_odoo_orders` — 16.2 of those 20.7 MB — are folded by SIX link
#     rollups on `ut_odoo_customers` (`ar_outstanding`, `open_invoices`, `oldest_due`,
#     `invoiced_all_time`, `order_count`, `last_order`). `automation_engine.compute_relation_cells`
#     answers those by reading the RAW document, so with the rows gone it writes zeros — silently.
#   * every table with a `link` pointing AT it (agents, vendors, bills, customers) has its link
#     CELLS materialised the same way, from the target's stored rows.
#   * `ut_odoo_customers` and `ut_odoo_products` carry SOURCE rollups, and `rollup_sql.compute`
#     writes those cells INTO their own stored rows: no rows, no cells.
#   * and while the grid client still asks for a whole table (F's W30-T42 is what changes this),
#     a population larger than one window could only be served by TRUNCATING it — which R6
#     forbids more strongly than it forbids a cap.
#
# ⭐ SO THE PREDICATE IS DERIVED FROM THE DECLARATIONS RATHER THAN LISTED. The day a lane converts
# those six link rollups to SOURCE rollups (`orders_ytd` on that same table is the precedent), or
# the day the client pages, the affected grids become eligible here with NO code change — and
# until then each one's reason is reported per table on the status door, which is R6's second
# sentence applied to our own conversion state.
def _fold_reasons(rel):
    """`{table_key: "why its stored rows are still read by something else"}`.

    Read out of the FIELD DECLARATIONS themselves — one pass over `rel.TABLES`. A table absent
    from this map is folded by nothing.
    """
    reasons = {}

    def _add(key, why):
        reasons.setdefault(str(key), []).append(why)

    for _bucket, key, _label, mk in rel.TABLES:
        try:
            fields = mk()
        except Exception:                                             # noqa: BLE001
            continue
        by_key = {f.get("key"): f for f in fields if isinstance(f, dict)}
        short = str(key).replace("ut_odoo_", "")
        for f in fields:
            if not isinstance(f, dict):
                continue
            if f.get("type") == "link" and (f.get("link") or {}).get("table"):
                _add(f["link"]["table"], f"{short}.{f['key']} is a link whose cells are built "
                                         f"from these rows")
            if f.get("type") != "rollup":
                continue
            bag = f.get("rollup") or {}
            if isinstance(bag.get("source"), dict):
                _add(key, f"{short}.{f['key']} is a source rollup and its cells are written "
                          f"into these rows")
                continue
            tgt = ((by_key.get(str(bag.get("link") or "")) or {}).get("link") or {}).get("table")
            if tgt:
                _add(tgt, f"{short}.{f['key']} folds these rows")
    return {k: "; ".join(v) for k, v in reasons.items()}


def _unmaterialisable(cur=None):
    """`({key: eligible}, {key: reason})` — who may stop storing rows, and why the rest may not.

    ⛔ SIZE FORCES READ-THROUGH; IT NEVER BLOCKS IT — and getting that backwards was a real bug in
    the first cut of this function. A table too big for the tenant document has NO materialised
    option at all, so making it ineligible would have handed `odoo_relational.plan` a `row_limit`
    of None and invited it to build 963,783 python dicts. The window ceiling is a different, much
    softer thing: it only limits what the whole-table CLIENT door can serve today.

        rows > MAX_ROWS            -> read-through REQUIRED (the document cannot hold it)
        else if something folds it -> stays materialised (a fold over no rows writes ZEROS)
        else if rows > WINDOW_MAX  -> stays materialised until the client pages (W30-T42)
        else                       -> eligible

    ⚠ Without a mirror cursor the size questions cannot be asked, so only a grid that was never
    part of the materialised spawn is eligible — a table is never freed by a question we skipped.
    """
    rel = _rel()
    folds, eligible, reasons = _fold_reasons(rel), {}, {}
    # ⛔ "WAS THIS PART OF THE MATERIALISED SPAWN?", and being in `TABLES` STOPPED ANSWERING IT.
    # W30-T35 declared the two line grains, which have no python row builder anywhere — their
    # absence from `_READERS` IS that statement — so a table the spawn could never materialise
    # started reading as spawned, and on a mirror-less deployment fell through to "its size could
    # not be read" and came back INELIGIBLE. That inverts this function's own first law (size
    # forces read-through; it never blocks it). Ask the question through the reader map, which is
    # what actually decides whether a row could ever have been built.
    spawned = {key for _b, key, _l, _f in rel.TABLES if _b in getattr(rel, "_READERS", {})}
    from harness import datastore
    ut = _ut()
    for key, spec in _sources().items():
        total = None
        if cur is not None:
            try:
                frm = ({"from_sql": spec["from_sql"]} if spec.get("from_sql")
                       else {"table": spec["table"]})
                total = datastore.window(select="1", where=spec.get("where") or "",
                                         order_by=spec["id"], limit=1, cur=cur, **frm)["total"]
            except Exception:                                         # noqa: BLE001
                total = None
        if total is not None and total > ut.MAX_ROWS:
            eligible[key] = True                  # no other home exists; this is not a choice
            continue
        if total is None and key not in spawned:
            eligible[key] = True                  # never materialised, so the mirror is its home
            continue
        why = []
        if total is None:
            why.append("its size could not be read from the mirror on this deployment")
        if key in folds:
            why.append(folds[key])
        if total is not None and total > datastore.WINDOW_MAX:
            why.append(f"its {total:,} rows exceed the {datastore.WINDOW_MAX:,}-row window and "
                       f"the grid client still asks for whole tables, so un-materialising it "
                       f"today could only truncate")
        eligible[key] = not why
        if why:
            reasons[key] = "; ".join(why)
    return eligible, reasons


def _degrade(expr, have_by_alias, default_alias=""):
    bare = expr.strip()
    alias, _, col = bare.partition(".")
    if col and alias.replace("_", "").isalnum() and col.replace("_", "").isalnum():
        have = have_by_alias.get(alias)
        if have is None:
            return expr                   # an alias we do not own — leave it to the caller's SQL
        return expr if col.lower() in have else "NULL"
    if not bare.replace("_", "").isalnum():
        return expr                       # an expression, not a bare column — nothing to check
    # ⚠ A BARE COLUMN IN A JOINED SPEC RESOLVES THE WAY SQL RESOLVES IT — against every table in
    # the FROM, not against nothing. Checking it against `have_by_alias[""]`, which a joined spec
    # does not have, would degrade every such column to NULL: a silent blank cell, which is the
    # failure this helper exists to avoid rather than to cause.
    have = have_by_alias.get(default_alias)
    if have is None:
        have = set().union(*have_by_alias.values()) if have_by_alias else set()
    return expr if bare.lower() in have else "NULL"


def _source_for(cur, table_key):
    """The resolved spec for one connected grid, or None when this grid is not read-through yet."""
    spec = _sources().get(str(table_key or ""))
    if not spec:
        return None
    from harness import datastore
    # ⚠ `tables` maps the SQL alias a projection uses to the mirror table behind it. A single-table
    # spec declares none and its columns are bare, so it degrades against `table` as before.
    aliases = dict(spec.get("tables") or {})
    if spec.get("table"):
        aliases.setdefault("", spec["table"])
    have_by_alias = {}
    for alias, tname in aliases.items():
        cols = datastore.columns_of(tname, cur=cur)
        if not cols:
            return None                   # the mirror has no such table on this deployment
        have_by_alias[alias] = cols
    excluded = ""
    if spec.get("needs_excluded"):
        ids = sorted(int(p) for p in _rel().excluded_ids(cur))
        # `-1` keeps the IN-list non-empty and matches no Odoo id, so the SQL shape is constant
        # whether or not this tenant excludes a channel.
        excluded = ", ".join(str(i) for i in ids) or "-1"
    cols = {}
    for key, (expr, cast) in spec["cols"].items():
        cols[key] = (_degrade(expr.format(excluded=excluded) if "{excluded}" in expr else expr,
                              have_by_alias), cast)
    return {**spec, "cols": cols}


#: How deep a page has to be before the walk is worth a sentence. Derived from the measurement in
#: the route below, not chosen: 100,000 is still 71 ms, 200,000 is 483 ms, and the second half of
#: the GL table is where it passes a second. Reporting from 100,000 puts the sentence in front of
#: the person BEFORE the wait rather than after it.
_DEEP_PAGE = 100_000


#: `filter_sql` speaks the CLIENT's type vocabulary (`types.ts isNumericType`), and two of our
#: field kinds are spelled differently there. Mapped in one place so the pushdown and the grid
#: cannot disagree about whether a column is text or a number.
_FILTER_TYPE = {"select": "status", "checkbox": "text"}


def _filter_columns(spec, fields):
    """`{colId: {sql, type, aggregate}}` — what `filter_sql` needs to compile a predicate.

    Only columns with a real mirror expression are offered. An omitted column is UNKNOWN to the
    compiler, which skips it — and that is the one behaviour that must be REPORTED rather than
    accepted, because an ignored condition WIDENS (the tri-state engine's inactive-leaf rule).
    `_unpushable` below turns every such skip into an R6 sentence.
    """
    by_key = {f["key"]: f for f in fields}
    out = {}
    for key, (expr, _cast) in spec["cols"].items():
        ftype = str((by_key.get(key) or {}).get("type") or "text")
        out[key] = {"sql": expr, "type": _FILTER_TYPE.get(ftype, ftype), "aggregate": False}
    return out


def _leaf_cols(nodes):
    """Every `colId` a filter tree names, at any depth."""
    seen = set()
    for n in (nodes or []):
        if not isinstance(n, dict):
            continue
        if n.get("children") is not None:
            seen |= _leaf_cols(n.get("children"))
        elif n.get("colId"):
            seen.add(str(n["colId"]))
            rhs = n.get("rhs")
            if isinstance(rhs, dict) and rhs.get("colId"):
                seen.add(str(rhs["colId"]))
    return seen


def _json_arg(raw, what):
    """Decode a JSON query argument, or refuse. ⛔ NEVER degrade to "no filter": a filter that
    silently fails to parse WIDENS the answer, and the caller sees a plausible bigger number."""
    if raw in (None, ""):
        return None
    try:
        val = json.loads(raw)
    except Exception:                                                 # noqa: BLE001
        raise err(400, "bad_argument", f"{what} must be JSON")
    return val


@router.get("/odoo-tables/{table_key}/rows")
def odoo_table_rows(table_key: str,
                    offset: int = Query(default=0, ge=0),
                    limit: int = Query(default=0),
                    filters: str = Query(default=None),
                    filterConj: str = Query(default="and"),
                    sorts: str = Query(default=None),
                    search: str = Query(default=None),
                    session: Session = Depends(require_session)):
    """ONE WINDOW over a connected grid, read straight from the mirror (contract C2).

    `{fields, rows, total, totalUnfiltered, offset, limit, limits, identity, recordsMutable}` —
    `rows` is the requested slice and `total` is the count of everything the CURRENT PREDICATE
    matches, from its own `SELECT count(*)`. `rows.length < total` is the normal case.

    ⛔ `total` is never `len(rows)`; `totalUnfiltered` is the population with the predicate
    dropped, so a client can render "N of M" without inventing either number.
    ⛔ The filter, the sort and the search all resolve in SQL against the whole table. Anything
    that CANNOT (a column with no mirror expression, an op the compiler refuses) is reported in
    `limits` with its cause and a recommendation — R6's second sentence — and never silently
    dropped, because an ignored condition widens.
    """
    import aios_grid
    from harness import datastore
    from routes_tables import _defn_or_refuse

    rel = _rel()
    # THE WALL FIRST, and it is the SAME one the materialised path uses — 404 for a key that does
    # not exist, 403 for one this session may not open. Reused rather than re-stated: a second
    # idea of "may this session open this database" is a permission bug waiting to happen.
    defn = _defn_or_refuse(session, table_key)
    fields = None
    for _bucket, key, _label, mk_fields in rel.TABLES:
        if key == table_key:
            fields = mk_fields()
            break
    if fields is None:
        raise err(404, "not_connected", "that database is not a connected Odoo grid")

    # ⭐⭐ W31-T45 / D-169 — THE SECOND WALL, and it asks a question `_defn_or_refuse` above cannot.
    # That wall asks "may this SESSION open this DATABASE"; this one asks "is the file this process
    # has open the one this TENANT's rows live in". A definition wall is satisfied the moment a
    # tenant's own document declares a connected table — which is exactly what R2's Meta Ads spawn
    # gives GTM Lab — and it would then serve that session a window over whatever DuckDB file the
    # process happens to be pinned to. Rows, not counts. So this door RAISES.
    # ⛔ 409, NOT 503, and the two were one line from being confused: `DatastoreMismatch` subclasses
    # `RuntimeError`, and the handler directly below turns a `RuntimeError` into
    # `503 store_not_ready` — "the store is completing its first sync, retry in a few minutes". A
    # mis-pinned process never becomes un-mis-pinned by waiting, so relaying it as a retry would
    # turn the loudest refusal in the system into a spinner. The guard therefore runs ABOVE the
    # try, not inside it.
    try:
        cur, mismatch = _mirror_cur(session)
    except RuntimeError as e:
        # ⚠ Reached ONLY by the mid-first-sync case: `_mirror_cur` has already consumed the
        # mismatch subclass, which is what makes this broad clause safe to keep here.
        raise err(503, "store_not_ready", str(e))
    if mismatch:
        raise err(409, "cross_tenant_store", mismatch)

    spec = _source_for(cur, table_key)
    if spec is None:
        raise err(409, "not_read_through",
                  "this connected database is still served from its stored copy; it has no "
                  "read-through binding on this deployment yet")

    cols = _filter_columns(spec, fields)
    limits, tree = [], _json_arg(filters, "filters")
    sort_spec = _json_arg(sorts, "sorts") or []

    # ── the predicate, pushed down ────────────────────────────────────────────────────────────
    named = _leaf_cols(tree)
    missing = sorted(named - set(cols))
    if missing:
        limits.append({
            "subject": ", ".join(missing), "effect": "filter_ignored",
            "cause": "these columns have no expression in the mirror (a link, a rollup, or a "
                     "column this deployment's mirror has not backfilled), so a condition on "
                     "them cannot be answered in SQL",
            "recommendation": "filter on the id column the link is built from, or open the "
                              "linked database directly"})
    # ⭐ R6, MEASURED, AND IT IS THE KIND OF LIMIT THE RULING EXISTS FOR — a difference in the
    # ANSWER, not in the speed.
    #
    # `filter_sql._value_sql` compiles every numeric comparison as `round_even(x, 0)`, and says
    # why: *"reproduces `aios_grid._round` … The grid displays rounded values; filters must agree
    # with what is on screen."* That is true of `source: "odoo"` columns, which `rows_from_pool`
    # rounds. It is FALSE here — these columns are `source: "overlay"` (a storage choice, not a
    # display one) so `rows_from_pool` passes the exact value through, and the TS engine
    # (`useVisibleRows.toNum`) does not round either. So the pushdown compares at whole units while
    # the cell beside it carries cents.
    #
    # MEASURED on the live orders mirror (32,700 rows, 47.9% with a non-integer amount):
    #   > 1000     python 4,476  vs SQL 4,473   (-3)
    #   > 173.6    python 24,711 vs SQL 24,726  (+15)
    #   > 500.25   python 11,028 vs SQL 11,026  (-2)
    # Small, and NOT nothing. Rounding the wire to match would have been the other fix and it was
    # rejected on measurement: it changes 96 of every 200 money cells (173.55 -> 174) to remove a
    # 0.05% counting difference — a visible product regression traded for an invisible one.
    # ⛔ SO IT IS REPORTED INSTEAD. Booked for the owner of `harness/filter_sql.py`, which is not
    # this fence; see mailbox/D.md.
    numeric = sorted(k for k in (named & set(cols))
                     if cols[k]["type"] in ("currency", "int", "pct"))
    if numeric:
        limits.append({
            "subject": ", ".join(numeric), "effect": "precision",
            "cause": "a number condition is evaluated in SQL at whole-unit precision "
                     "(`filter_sql` rounds to match the grids whose values the server rounds), "
                     "while these cells carry their exact value — so a row within half a unit of "
                     "the threshold can fall on the other side of it",
            "recommendation": "compare against a whole number, or use a range that does not sit "
                              "on a fractional boundary"})

    where, params = spec["where"], []
    try:
        pred = _fs().compile_filter_tree(
            tree, conj=(filterConj if filterConj in ("and", "or") else "and"), columns=cols,
            today=_today())
    except ValueError as e:
        # ⛔ A REFUSAL IS AN ANSWER, NOT A CRASH — and it must not become "no filter". Ranking ops
        # ("top 10") have no SQL form in this compiler; saying so beats returning every row.
        raise err(400, "filter_unsupported", str(e))
    if pred is not None:
        if pred.uses_aggregate:
            raise err(400, "filter_unsupported",
                      "a condition on an aggregate column belongs in HAVING, and that path is "
                      "deliberately not built for windowed grids")
        where = f"({where}) AND {pred.sql}" if where else pred.sql
        params.extend(pred.params)
    if str(search or "").strip():
        got = _fs().compile_search(search.strip(), cols)
        if got is not None:
            where = f"({where}) AND {got.sql}" if where else got.sql
            params.extend(got.params)

    # ── the order, made TOTAL ─────────────────────────────────────────────────────────────────
    # ⛔ `tiebreak_sql` is not optional here: without a total order, LIMIT/OFFSET may return one
    # row on two pages and drop another entirely — a duplicate the user sees with no error
    # anywhere. `compile_order_by`'s own docstring says so.
    order = _fs().compile_order_by(sort_spec, cols, tiebreak_sql=spec["id"]) or spec["id"]

    select = ", ".join(f'{expr} AS "{key}"' for key, (expr, _c) in spec["cols"].items())
    # ⚠ EXACTLY ONE of `table`/`from_sql` — `window` raises if both or neither arrive, so the
    # spec's own shape decides and a malformed spec fails loudly instead of serving a wrong FROM.
    frm = {"from_sql": spec["from_sql"]} if spec.get("from_sql") else {"table": spec["table"]}
    win = datastore.window(select=f'{spec["id"]} AS "_pid", {select}',
                           where=where, params=tuple(params), order_by=order,
                           offset=offset, limit=(limit or None), cur=cur, **frm)
    if win["clamped"]:
        limits.append({
            "subject": "limit", "effect": "window_clamped",
            "cause": f"one response carries at most {datastore.WINDOW_MAX} rows so a single "
                     f"request cannot exhaust memory for every other tenant on the process",
            "recommendation": "page with `offset`; `total` already reports the whole population, "
                              "and no row is unreachable"})
    # ⭐ R6's SECOND SENTENCE, ON THE ONE LIMIT THAT SURVIVES THE CONVERSION. Removing the row cap
    # does not make every row equally cheap: `LIMIT/OFFSET` WALKS the offset, so the deeper the
    # page the longer the scan. MEASURED warm on 963,783 GL lines — offset 0: 134 ms · 100,000:
    # 71 ms · 900,000: **1,450 ms**; sorted by date rather than by id, offset 500,000: 1,645 ms.
    # It is a real cost, it is nobody's mistake, and the owner asked to be told rather than to
    # discover it: say so with the fix, which is a CURSOR the client has to send.
    if offset >= _DEEP_PAGE:
        limits.append({
            "subject": "offset", "effect": "slow",
            "cause": f"a page {offset:,} rows deep is reached by walking every row before it "
                     f"(SQL OFFSET has no other meaning), which costs about a second past "
                     f"half a million rows",
            "recommendation": "jump with a filter or a sort instead of scrolling, or ask for "
                              "keyset paging (`id > <last row>`), which is O(page) at any depth "
                              "and needs the client to send the last row it holds"})

    # ── the wire rows, through the SAME serialiser the materialised path uses ──────────────────
    # ⭐ D-155 — the freshness stamp rides the SAME builder, so both read-through doors answer the
    # `refreshed` column identically instead of one of them leaving it blank.
    rows_src = _pool_from_window(spec, win,
                                 stamp=mirror_stamp(spec, cur, table_key,
                                                    session.runtime))
    overlays = {str(r["pid"]): {k: v for k, v in r.items() if k != "pid"} for r in rows_src}

    # ⭐⭐ W30-T28 — THE TENANT-WIDE OVERLAY, ON A READ-THROUGH GRID.
    #
    # A user-added column on a connected grid has nowhere per-user to live: there is no `ut_*` row
    # to hang it on any more, and `table_store`'s per-user strata would make a SHARED view name a
    # column other accounts do not have — an unknown column is an INACTIVE condition in the
    # tri-state engine, which WIDENS. `core/shared_overlay.py` was built for exactly this and has
    # had no product door since it shipped (W29-T62).
    #
    # ⛔ `cells(table_key, pids)` TAKES THE PIDS AND THERE IS NO "EVERYTHING" CALL — and here that
    # is an asset rather than a chore: **the window IS the scoped row set**, already narrowed by
    # the wall and the predicate, so the argument it demands is the list we just fetched.
    # ⚠ It is NOT a permission wall (its header says so twice); `_defn_or_refuse` above already
    # answered "may this session open this surface".
    shared_defs = _so().fields(table_key, st=session.runtime)
    if shared_defs:
        fields = list(fields) + [dict(f, source="overlay") for f in shared_defs.values()]
        for pid, cells in _so().cells(table_key, [r["pid"] for r in rows_src],
                                      st=session.runtime).items():
            overlays.setdefault(pid, {}).update(cells)
    rows = aios_grid.rows_from_pool(rows_src, fields, overlays)

    unfiltered = win["total"] if where == spec["where"] else datastore.window(
        select="1", where=spec["where"], order_by=spec["id"], limit=1, cur=cur, **frm)["total"]

    return {"ok": True, "fields": fields, "rows": rows,
            # ⛔ FROM THE COUNT STATEMENT. Never `len(rows)`.
            "total": win["total"], "totalUnfiltered": unfiltered,
            "offset": win["offset"], "limit": win["limit"], "limits": limits,
            "today": _today(), "identity": {"pid": "pid"},
            "scope": {"table": table_key, "readThrough": True},
            "recordsMutable": bool(defn.get("recordMode") != _ut().AUTOMATION_RECORD_MODE)}


def mirror_stamp(spec, cur=None, table_key="", rt=None):
    """⭐ D-155 — WHEN THIS READ-THROUGH GRID'S DATA WAS LAST RECONCILED AGAINST ODOO, as a date.

    ⛔ THE COLUMN EXISTED AND THE BINDING DID NOT, which is the whole of D-155. Every Odoo table
    declares `refreshed` (`odoo_relational._refreshed_field`), and a MATERIALISED grid gets a
    per-row stamp written by the spawn. A read-through grid has no stored row to carry one, so
    `_pool_from_window` produced no `refreshed` key at all and `rows_from_pool` projected it as
    `""` — a freshness column that used to say something and now silently says nothing, on the two
    biggest grids in the product. Nothing goes red for that; it just quietly stops being true.

    ⭐ THE HONEST VALUE ALREADY EXISTS — the mirror keeps a per-entity sync stamp in `_sync_state`,
    which is the same fact one level up: these rows are as fresh as the last sync of the table
    they are read through. It is per-TABLE, not per-row, and that is not a downgrade — for a grid
    that stores no rows, "when did this table last sync" IS the per-row answer.

    ⚠ READ ON THE CALLER'S CURSOR, never `datastore.status()`. That opens a SECOND connection to a
    file DuckDB holds exclusively, so a freshness column would turn a working grid into an
    IOException — a cosmetic fix taking out the feature it decorates.
    ⚠ Date only (`YYYY-MM-DD`), because `refreshed` is declared `type: "date"` and the grid's date
    renderer is what reads it; handing it a full timestamp renders the ISO string a user should
    never see (the wave-26 `copyData` scar, one column over).

    ⛔⛔ THE TABLE COMES FROM THE **ID COLUMN'S ALIAS**, AND THE FIRST DRAFT GOT THIS WRONG IN THE
    ONLY WAY THAT MATTERS. It read `spec["table"]` with a `tables[""]` fallback — and **neither of
    the two grids D-155 is about has a bare `table` key**: `ut_odoo_order_lines` and
    `ut_odoo_gl_lines` are join-shaped (`from_sql` + `tables: {"sol": …, "so": …}` /
    `{"aml": …, "aa": …}`), so the lookup returned `""` and the fix would have shipped doing
    nothing on exactly the two grids it was written for — green gate, unchanged product. It was
    caught only because the gate's fixture used `ut_odoo_orders`, a MATERIALISED grid that never
    had the defect [[gate-answers-the-wrong-question]].
    ⭐ The id column IS the grain: `sol.id` means this grid is one row per `sale_order_line`, and
    the freshness of a joined lookup table (`sale_order`, `account_account`) is not this grid's
    freshness. So the alias is read off `spec["id"]`, and a bare `id` degrades to `table` — which
    is exactly the single-table case.

    ⛔ AND IT ANSWERS `""` FOR A **MATERIALISED** GRID, ON PURPOSE. Those eight tables store a
    per-ROW `refreshed` written by the spawn, which is strictly better than one table-level date —
    and this value arrives as an OVERLAY, so returning a stamp here would quietly overwrite eight
    working grids' per-row stamps while fixing two blank ones. D-155's subject is the grid that has
    NOWHERE to keep a row; a fix that also rewrites the grids that do is a different, unrequested
    change [[reuse-and-delete-are-hypotheses]].
    """
    # the materialised carve-out above, made structural. ⚠ Fail-QUIET to `""`: when we cannot
    # tell whether this grid stores rows, the safe answer is the behaviour that shipped (blank),
    # never a stamp that might overwrite a per-row one.
    if table_key:
        try:
            if _ut().materialises(str(table_key), st=rt):
                return ""
        except Exception:                                              # noqa: BLE001
            return ""
    ident = str(spec.get("id") or "")
    alias = ident.split(".", 1)[0] if "." in ident else ""
    table = str((spec.get("tables") or {}).get(alias) or spec.get("table")
                or (spec.get("tables") or {}).get("") or "")
    if not table or cur is None:
        return ""
    try:
        row = cur.execute("SELECT updated_at FROM _sync_state WHERE entity = ?",
                          [table]).fetchone()
    except Exception:                                                  # noqa: BLE001
        return ""                     # no mirror bookkeeping ⇒ blank, exactly as before
    return str((row or [""])[0] or "")[:10]


def _pool_from_window(spec, win, stamp=""):
    """`[{pid, **cells}]` — THE one place a mirror window becomes product rows.

    ⛔ ONE BUILDER, TWO CALLERS, and that is deliberate rather than tidy: the windowed route and
    the whole-table read-through below would otherwise each cast the same columns their own way,
    and a cell that renders differently depending on which door served it is this repo's recorded
    defect class ([[one-question-two-normalizers]]).

    ⚠ `stamp` (D-155) rides here for that same reason: both doors must produce the same
    `refreshed` cell, and a caller that forgot it would give one door a freshness column and the
    other a blank one. Empty when the caller cannot cheaply know — blank is what shipped, so the
    degradation is the previous behaviour rather than a new wrong value.
    """
    keys = list(spec["cols"])
    extra = {"refreshed": stamp} if stamp else {}
    return [{"pid": int(r[0]), **extra,
             **{k: spec["cols"][k][1](v) for k, v in zip(keys, r[1:])}} for r in win["rows"]]


class TooBigToMaterialise(Exception):
    """A read-through table asked for WHOLE exceeds one window — refuse, never truncate."""


def population(table_key, cur=None, rt=None):
    """How many rows a read-through grid HAS, without fetching one — or None if it is not bound.

    ⭐⭐ W31 / B's ask (mailbox/B.md B-2). `routes_tables.scoped_pids` learned "this grid is too big
    to list" the only way that existed: call `whole_pool()` and catch `TooBigToMaterialise`. That
    pulls a full 5,000-row window out of DuckDB and throws it away on every `/workspace` for
    `ut_odoo_gl_lines` and `ut_odoo_order_lines` — MEASURED by B at ~1,600 ms of a ~3,460 ms
    in-proc envelope. This asks the SIZE instead: one `SELECT count(*)` over the same predicate.

    ⛔ IT ANSWERS A DIFFERENT QUESTION, NOT THE SAME ONE MORE CHEAPLY, and that distinction is what
    keeps it safe. B's constraint is that the pid set must stay IDENTICAL to `scoped_pool`'s BY
    CONSTRUCTION rather than by two queries that agree today ([[one-question-two-normalizers]]) —
    so this returns a COUNT and never a pid. A caller uses it to decide whether to ask for pids at
    all; when it does ask, the pids still come from the one `whole_pool` fetch they always did.

    ⚠ `None` means "no read-through binding on this deployment", which is NOT `0`. A caller that
    collapses them reports an empty grid where it should report an unbound one — the exact
    distinction `odoo_tables_status` already reports as `bound_not_declared`.
    ⚠ `total` is the count over the spec's OWN `where`, i.e. the same population `whole_pool`
    would return — the fixed scope is not dropped for being cheaper.
    """
    from harness import datastore
    if rt is not None:
        rt.assert_datastore_matches()          # W31-T45: the same wall the row doors carry
    cur = cur if cur is not None else datastore.ro_con()
    spec = _source_for(cur, table_key)
    if spec is None:
        return None
    frm = {"from_sql": spec["from_sql"]} if spec.get("from_sql") else {"table": spec["table"]}
    return datastore.window(select="1", where=spec.get("where") or "", order_by=spec["id"],
                            limit=1, cur=cur, **frm)["total"]


def whole_pool(table_key, cur=None, rt=None):
    """Every row of a read-through grid, in `scoped_pool`'s shape — or a refusal.

    ⚠ THIS EXISTS FOR THE CLIENT WE HAVE, NOT THE ONE WE WANT. The grid still fetches whole
    tables (`GET /tables/{key}/rows`); paging is F's W30-T42. So a database whose rows have left
    this tenant's document has to be servable to that client somehow, and the honest answer for a
    small one is "read all of it from the mirror". `_unmaterialisable` only ever registers a table
    that fits, so the refusal below is a guard against the population GROWING past the window
    later — at which point R6 requires a sentence, not a quietly shorter grid.

    ⭐⭐ W31-T45 / D-169 — `rt` IS THE TENANT RUNTIME, AND IT IS OPTIONAL FOR A STATED REASON
    RATHER THAN A LAZY ONE. This function serves ROWS, so it is the door where a cross-tenant read
    would be worst — but its only production caller is `routes_tables.py::_read_through_rows`,
    which is in another lane's fence THIS WAVE (B's W31-T20 is rewriting that exact path) and does
    not thread a session down to here. Making `rt` required would break that caller on import; a
    second predicate invented locally would be one question with two normalizers, a recorded
    defect class here. So the guard fires when a caller passes the runtime it already holds, and
    the ask to thread `session.runtime` through `_read_through_rows` is posted to B in
    `mailbox/D.md`. ⛔ Until that lands this door is guarded only by `_defn_or_refuse` upstream —
    stated here rather than left for a reader to discover, because an unguarded row door is
    precisely what D-169 is about.
    """
    from harness import datastore
    if rt is not None:
        cur = cur if cur is not None else _rt().mirror_cursor(rt)
        rt.assert_datastore_matches()          # also covers a cursor the CALLER opened and lent
    cur = cur if cur is not None else datastore.ro_con()
    spec = _source_for(cur, table_key)
    if spec is None:
        raise TooBigToMaterialise(
            f"{table_key}: this database is served through the mirror but has no read-through "
            f"binding on this deployment, so its rows cannot be read at all")
    frm = {"from_sql": spec["from_sql"]} if spec.get("from_sql") else {"table": spec["table"]}
    select = ", ".join(f'{expr} AS "{key}"' for key, (expr, _c) in spec["cols"].items())
    win = datastore.window(select=f'{spec["id"]} AS "_pid", {select}', where=spec.get("where"),
                           order_by=spec["id"], limit=datastore.WINDOW_MAX, cur=cur, **frm)
    if win["total"] > len(win["rows"]):
        # ⛔ REFUSE, NEVER TRIM. A short grid that says nothing is exactly the silent truncation
        # R6's second sentence is about — and the caller turns this into a message with a cause.
        raise TooBigToMaterialise(
            f"{table_key}: {win['total']:,} rows is more than one {datastore.WINDOW_MAX:,}-row "
            f"window, and this database no longer stores rows in the tenant document; it can "
            f"only be read a page at a time (`/odoo-tables/{table_key}/rows`)")
    return _pool_from_window(spec, win, stamp=mirror_stamp(spec, cur, table_key, rt))


def _fs():
    from harness import filter_sql
    return filter_sql


def _so():
    import core.shared_overlay as shared_overlay
    return shared_overlay


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


def _today():
    import time
    return time.strftime("%Y-%m-%d")