platform/modules/product_data.py

"""modules/product_data.py — the PRODUCT table's pool (wave 15 item 9/10, contract C-TOPIC).

The second object on the table-page factory: same grid, same engine, same permission wall — the
only difference is the field schema and the identity. `modules/customer_data.pool()` is the
template and this deliberately mirrors its signature and its row shape.

⛔ THREE DECISIONS THIS FILE HAD TO MAKE, EACH RECORDED BECAUSE A LATER READER WILL WONDER.

1. **THE IDENTITY IS A SKU CODE, WHICH IS A STRING, AND THE GRID WANTS AN INTEGER `pid`.**
   Cohort membership, `allowed_pids`, the measure channel and `rows_from_pool` all key on an
   integer. So each row carries BOTH: `pid` (a stable CRC32 of the code, so the same SKU gets the
   same id on every pull and across processes — never an enumeration index, which would reshuffle
   whenever the catalogue changes) and `code`, the real business key, as a visible column.
   `_assert_no_pid_collision` fails the BUILD rather than the read: two SKUs sharing a pid would
   silently merge in every downstream set operation, and a loud build failure is the only version
   of that anyone would notice.

2. **THE SCOPE RULE — inventory columns are CONSOLIDATED and are therefore OMITTED for a
   BU-scoped caller.** `products.directory(t, team_id)` is brand-shaped; `inventory.sku_inventory`
   is explicitly NOT (its own docstring: "on-hand stock is one physical warehouse, not
   brand-tagged"). Joining them for a Fisch-only user would put BU-shaped revenue beside
   company-wide stock IN THE SAME ROW — the mixed-scope value defect wave 15's amendment 3 exists
   for, arriving through a different door. The honest options were "omit" or "label the columns
   company-wide"; omit is the fail-closed one, and a column that is absent asks a question,
   whereas a column that is silently company-wide answers one wrongly.

3. **WHAT IS NOT HERE, NAMED RATHER THAN QUIETLY MISSING.** R4 lists ~20 fields. Vendor and
   COUNTRY are not in Odoo at all — they live in the inventory WORKBOOK
   ([[odoo-vendor-country-origin]]) and need a loader this module deliberately does not invent.
   Margin % comes from `modules/pricing.table`, which is a heavier build (channel-rate cost
   allocation) and is left for the wave that needs it. `validate()` reconciles what SHIPS; it
   does not pretend to cover columns that are absent.

4. **THE POOL IS CATALOGUE-FIRST, WITH REVENUE LEFT-JOINED** (wave 29, owner item 22 / R12,
   2026-08-11). It was `for r in products.directory(...)` — and `directory()`'s row set IS the
   union of two revenue read-groups, so **a SKU that never sold could not exist** and the grid
   showed **2,717** of **5,875** active products. Three things that will be re-derived otherwise:

     · ⛔ THE CAUSE IS A JOIN, NOT A LIMIT. No row cap exists on this path. Removing the date
       window from `directory()` would ALSO be wrong twice: it breaks four SKU-health metrics
       that legitimately want a sales window, and it lands at 3,327 (all-time-sold), because
       ~2,550 active SKUs have never sold in wholesale scope at all.
     · ⭐ REVENUE ON A NEVER-SOLD ROW IS **BLANK, NOT $0** (R12). A row that appears in the
       revenue universe carries measured numbers INCLUDING a real 0.0 — it sold last year and
       not this one, and that zero is a measurement. A row that appears in NO revenue read
       carries `None`, which the wire keeps (`aios_grid._round` passes None through) and the
       client renders as an empty cell. Blank is an admission; zero is a measurement.
     · ⚠ A BU-SCOPED CALLER NOW SEES THE WHOLE CATALOGUE, with ITS OWN revenue and blanks where
       that BU never sold. That is not decision 2's mixed-scope defect: the catalogue is the ROW
       UNIVERSE, not a company-wide VALUE sitting beside a BU-shaped one. `product.product`
       carries no team, so a catalogue cannot be BU-shaped at all — which is exactly why the two
       sources are JOINED rather than merged.
"""
import json
import zlib
from pathlib import Path

import core.odoo as O
import core.periods as P
import core.shared_overlay as shared_overlay
import core.table_store as table_store
import modules.products as products

#: POOL-ROW KEYS that exist ONLY on a consolidated pull. See decision 2.
#: `cover_gap_d` / `buy_now` (wave 17) join it because both are computed FROM `dos`, and a buy
#: signal built on a stock number the caller cannot see would be a recommendation nobody could
#: check ([[no-unverifiable-aggregates]]).
#:
#: ⚠ ROW KEYS, NOT FIELDS, and the distinction started mattering on 2026-08-03. Two readers:
#: `verify_perm_scope` asserts the shape of a POOL ROW against this, and `routes_products`
#: narrows the FIELD list with it. Since owner item 5 `buy_now` is a row key with no Field —
#: computed as `validate()`'s oracle, projected onto no wire — so it belongs here for the first
#: reader and is inert for the second. Removing it would stop the scope gate proving that a
#: BU-scoped pull withholds it.
#:
#: ⭐⭐ OWNER RULING 2026-08-11 — THIS SET IS NOW EMPTY, AND THAT IS THE POINT. Verbatim:
#: *"Just scope any inventory with Sales from Fisch, leave the rest."* Withholding the whole
#: inventory block from a BU-scoped reader meant a Fisch salesperson could not see whether
#: anything was IN STOCK, which is the first question they ask. The ruling splits the block by
#: what a business unit can actually shape:
#:
#:   * `on_hand` / `unit_cost` / `inv_value` — ONE physical warehouse, no Fisch shelf and no
#:     Royal shelf. Served UNSCOPED to everybody, identical in both pulls. ("leave the rest")
#:   * `qty_ltm` / `dos` / `stock_bucket` / `cover_gap_d` / `buy_now` — all SALES-derived, so a
#:     BU-scoped pull recomputes them from THAT unit's LTM units. ("scope any inventory with
#:     Sales from Fisch")
#:
#: This deliberately relaxes decision 2's "never a company-wide value beside a BU-shaped one":
#: stock is not a value a BU can own, and a column that is absent for a Fisch reader asks a
#: question they cannot answer anywhere else in the product. `verify_perm_scope` no longer
#: asserts the columns are WITHHELD — it asserts the split above, which is a stronger claim and
#: cannot pass on an empty scoped payload (the shape the old rule and an outage share).
CONSOLIDATED_ONLY = ()

#: ⛔ POOL-ROW KEYS THAT DELIBERATELY HAVE NO FIELD — they must never reach a browser.
#:
#: `rows_from_pool` projects strictly through the field contract, so "no Field" already means
#: "no cell on the wire". This tuple makes that a CHECKED fact rather than a consequence nobody
#: is watching: `verify_perm_scope` asserts every member is absent from BOTH product contracts,
#: and that every OTHER `CONSOLIDATED_ONLY` key is present in the consolidated one.
#:
#: Written because the two tuples silently diverged the moment owner item 5 retired `buy_now`'s
#: Field while keeping its computation, and the gate — which was asserting a FIELD property from
#: a ROW-key list — went red with no way to tell a deliberate divergence from a dropped column.
UNSHIPPED_ROW_KEYS = ("buy_now",)

#: Wave 17 (owner item 13, ruling R3) — the SUPPLIER MASTER, from the curated mastersheet map
#: `procurement_suppliers.json` (3,963 SKUs; supplier on 3,624, lead time on 3,563). NOT Odoo:
#: the owner confirmed this data came from the Inventory System Mastersheet, which is why
#: `modules/product_data`'s header said Vendor/Country were "NOT HERE, named rather than
#: quietly missing" and needed "a loader this module deliberately does not invent". This is
#: that loader.
#:
#: ⛔ THESE ARE CONTRACT COLUMNS, NOT USER-CREATED FIELDS, AND THE REASON IS A PERMISSION FACT.
#: A user-created field and its values live in the PER-USER strata (`table_store.workspace`
#: reads `store.get(key)[username]`); only VIEWS are shared. So a shared "Buy list" view that
#: filtered on a user-created column would, for every OTHER account, name a column that does not
#: exist — and an unknown column is an INACTIVE condition in the tri-state engine, which IGNORES
#: it and therefore WIDENS. The buy list would silently show the whole catalogue to everyone but
#: its author. Contract columns are identical for every reader, so the view means one thing.
#:
#: ⭐⭐ 2026-08-12 (W30-T36) — THE OWNER'S ASK IS NOW DELIVERED, AND THE PARAGRAPH ABOVE IS WHY IT
#: TOOK THREE WAVES. Owner: *"turn this Excel sheet into a User created Field that we can edit."*
#: It was parked because a per-user column silently WIDENS a shared view — not because editing was
#: hard. Wave 29's `core/shared_overlay.py` (whose header quotes this very comment) is the stratum
#: that removes the objection: **one value per (row, column) for the whole tenant**, so the column
#: still means ONE thing to every reader and a shared view still filters honestly.
#: ⇒ The four columns below are now `source: "overlay"` + `shared: true` in the canonical
#: contract, their values live in `<TABLE_KEY>__shared`, and the master map is what SEEDS an
#: unedited cell rather than what freezes it. See `SHARED_KEYS` and `_ProductTableStore`.
_SUPPLIER_MAP_PATH = Path(__file__).resolve().parent.parent / "procurement_suppliers.json"
_SUPPLIER_CACHE = {}


def supplier_master():
    """`{code: {supplier, lead_days, origin_country, first_cost}}`, read once per process.

    Degrades to `{}` when the file is unreadable, matching `_inventory_by_code`: a product table
    that will not render because a master map is missing is a worse failure than one with blank
    supplier columns.
    """
    if _SUPPLIER_CACHE:
        return _SUPPLIER_CACHE
    try:
        raw = json.loads(_SUPPLIER_MAP_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {}
    for code, meta in (raw or {}).items():
        if not isinstance(meta, dict):
            continue
        lead = meta.get("lead")
        _SUPPLIER_CACHE[str(code)] = {
            "supplier": (meta.get("vendor") or "") or None,
            "lead_days": int(lead) if isinstance(lead, (int, float)) else None,
            "origin_country": (meta.get("country") or "") or None,
            "first_cost": meta.get("first_cost"),
        }
    return _SUPPLIER_CACHE

#: Wave 16 C-TOPIC — the PRODUCT table's OWN workspace bucket. ⛔ Never the customer one:
#: product pids are CRC32 hashes of SKU codes and customer pids are Odoo partner ids, so in a
#: SHARED overlay bucket a hash collision would silently write a product note onto somebody's
#: customer (or the reverse). Separate store keys make that structurally impossible, which is
#: the whole reason the table-page factory exists ("a new table object gets its own
#: table_store.make('<its>_table_workspace')").
TABLE_KEY = 'product_table_workspace'

_GRID_FIELDS_PATH = Path(__file__).resolve().parent.parent / 'aios_grid_fields.json'
_SHARED_KEYS = None


def SHARED_KEYS():
    """The product columns whose values are TENANT-WIDE — derived from the canonical contract's
    own `shared: true`, never typed out a second time.

    ⛔ IT DELIBERATELY DOES NOT SWALLOW A READ FAILURE. Degrading to `()` would send a shared
    write into the per-user stratum with nothing going wrong anywhere — the widening defect
    reappearing silently, which is the one outcome this whole mechanism exists to prevent. If the
    canonical contract is unreadable the product grid cannot render at all (`pd_fields` parses the
    same file with no guard), so a raise here costs nothing that was still working.
    """
    global _SHARED_KEYS
    if _SHARED_KEYS is None:
        doc = json.loads(_GRID_FIELDS_PATH.read_text(encoding='utf-8'))
        _SHARED_KEYS = tuple(f['key'] for f in (doc.get('product_data') or {}).get('fields') or []
                             if f.get('shared'))
    return _SHARED_KEYS


class _ProductTableStore(table_store.TableStore):
    """The product workspace, with the SHARED columns routed to the tenant-wide stratum.

    ⭐⭐ THIS SUBCLASS IS THE WHOLE OF W30-T36's WRITE PATH, AND THE REASON IT LIVES HERE RATHER
    THAN AT A ROUTE IS THAT **THE BROWSER NEVER CALLS `PATCH /products/{pid}`** — measured, zero
    call sites in `aios-web/web/src`. A cell edit travels `POST /grid/events` → `grid_events.
    handle_one` → `_tops(ctx).patch_overlay(...)`, and `_tops` returns `ctx.table`, which
    `routes_grid._ctx` sets to `TABLE_OPS` for the product scope. So this object IS the seam both
    doors pass through; intercepting at either route would have left the other one writing a
    per-user value that only its author could see.

    ⚠ `st=self.st`, NEVER the module default. The shared stratum must resolve to the SAME store
    handle as the per-user one it sits beside — `_tops`' own comment explains that a split, where
    one side is tenant-scoped and the other is not, is worse than a stated residency error because
    a user's value would vanish the moment they saved it. Reading `self.st` means both strata move
    together the day that singleton gains a tenant handle.
    """

    def patch_overlay(self, username, pid, updates):
        clean = dict(updates or {})
        if not clean:
            return
        keys = set(SHARED_KEYS())
        shared = {k: v for k, v in clean.items() if k in keys}
        personal = {k: v for k, v in clean.items() if k not in keys}
        if shared:
            shared_overlay.put_cells(TABLE_KEY, pid, shared, st=self.st)
        if personal:
            super().patch_overlay(username, pid, personal)


TABLE_OPS = _ProductTableStore(TABLE_KEY)


def shared_cells(pids, st=None):
    """`{"<pid>": {key: value}}` for the SHARED columns of the rows named by `pids`.

    ⚠ `pids` is required and positional all the way down — `shared_overlay.cells` refuses to serve
    "everything" by design, and the caller here always holds an already-scoped pool.
    """
    return shared_overlay.cells(TABLE_KEY, pids, st=st if st is not None else TABLE_OPS.st)


def sku_pid(code):
    """A stable integer id for a SKU code. CRC32, masked to 31 bits so it is always positive and
    always JSON-safe. Stable across processes and pulls, which an enumeration index is not."""
    return zlib.crc32(str(code).encode("utf-8")) & 0x7FFFFFFF


def _assert_no_pid_collision(rows):
    """Two SKUs sharing a pid would MERGE in every set operation downstream — cohort membership,
    allowed_pids, the measure channel — and nothing would report it. Fail the build instead."""
    seen = {}
    for r in rows:
        prior = seen.get(r["pid"])
        if prior is not None and prior != r["code"]:
            raise ValueError(
                f"product_data: pid collision — {prior!r} and {r['code']!r} both hash to "
                f"{r['pid']}. Downstream set operations would merge them silently; widen the id "
                f"before shipping this catalogue.")
        seen[r["pid"]] = r["code"]


def _inventory_by_code(t):
    """`{code: {...}}` from the inventory module, or `{}` if it cannot be read.

    Degrades to empty rather than raising, matching `customer_data._pool_build`'s treatment of
    its own slow families: a product table that will not render because inventory is momentarily
    unreachable is a worse failure than one with blank stock columns.
    """
    try:
        import modules.inventory as inventory
        return inventory.sku_inventory(t=t) or {}
    except Exception:
        return {}


def _bu_ltm_share(t, team_id):
    """`{code: 0.0..1.0}` — this unit's SHARE of the SKU's last-twelve-months units.

    ⭐ OWNER 2026-08-11: *"scope any inventory with Sales from Fisch"*. The velocity half of the
    inventory block has to be re-shaped by a BU fact, and this is that fact.

    ⛔ A SHARE, NOT THE UNIT COUNT ITSELF — AND THE REASON IS A MEASURED IMPOSSIBILITY. My first
    version returned `products._sku_rev(...)['qty']` and used it directly as the scoped `qty_ltm`,
    leaving the consolidated column on `inventory.sku_inventory`'s own figure. Two readers of one
    question ([[one-question-two-normalizers]]): they disagree, and on 5 SKUs of 5,871 the live
    check found **Fisch's LTM units EXCEEDING the company's** — a subset larger than its superset,
    which no reader could explain and no reconciliation could survive.

    Both halves of the ratio come from the SAME read here, so the share is in [0, 1] by
    construction and the scoped figure can never exceed the consolidated one. It also leaves the
    consolidated column exactly as it was — the Inventory page and the Product grid still agree
    about company-wide units, which is what "leave the rest" asked for.

    `all_qty == 0` implies `bu_qty == 0` (same reader), so a share of 0 is the honest answer for
    "this unit never sold it": `_bucket` turns that into 'No recent sales', not zero cover.

    Degrades to `{}` on any failure, matching `_inventory_by_code` — and a missing code then takes
    the `_DEFAULT_SHARE` below rather than a silent 0.
    """
    try:
        f, to = P.ltm(t)
        allq = {code: (r.get('qty') or 0.0)
                for code, r in (products._sku_rev(f, to, None) or {}).items()}
        buq = {code: (r.get('qty') or 0.0)
               for code, r in (products._sku_rev(f, to, team_id) or {}).items()}
    except Exception:
        return {}
    out = {}
    for code, total in allq.items():
        out[code] = min(1.0, max(0.0, (buq.get(code, 0.0) / total))) if total > 0 else 0.0
    return out


#: What a SKU absent from the LTM sales read is worth to a business unit. ZERO — it did not sell
#: in anybody's book over the window, so no unit can claim its velocity. Named rather than
#: inlined so the choice is visible: the alternative (1.0, "assume it is all ours") would print
#: company-wide cover on a BU grid, which is the mixed-scope defect this whole rule avoids.
_DEFAULT_SHARE = 0.0


def _rescope_inventory(e, share):
    """`(qty_ltm, dos, bucket)` recomputed for ONE business unit's sales rate.

    The formulas are `inventory.sku_inventory`'s, applied to a BU-shaped numerator — NOT a second
    idea of what days-of-supply means. `_bucket` is imported from there for the same reason: two
    copies of a threshold table is how the Product grid and the Inventory page start disagreeing
    about which SKUs are dead.

    ⚠ `on_hand` is whatever the warehouse holds, unscoped — so a Fisch reader's `dos` answers
    "how long does ALL our stock last at Fisch's rate", which is the question a Fisch salesperson
    actually has. It is deliberately NOT a pro-rated share of the shelf: there is no such shelf,
    and inventing one would put a number on screen that no Odoo query could reproduce.
    """
    on_hand = e.get("on_hand")
    if on_hand is None:
        return None, None, None          # no inventory row for this SKU: blank, never zero
    qty = float(e.get("qty_ltm") or 0.0) * float(share or 0.0)
    daily = qty / 365.0
    if daily > 0:
        dos_raw = on_hand / daily
    else:
        dos_raw = float('inf') if on_hand > 0 else 0.0
    try:
        import modules.inventory as inventory
        bucket = inventory._bucket(dos_raw, on_hand, qty)
    except Exception:
        bucket = None
    return qty, (None if dos_raw == float('inf') else round(float(dos_raw), 0)), bucket


def _catalogue_by_code():
    """`{code: {'product', 'category'}}` — the CATALOGUE universe this pool is built from.

    ⛔ UNLIKE `_inventory_by_code`, THIS ONE RAISES, and the asymmetry is the whole point.
    Inventory degrades to `{}` because a product table with blank stock columns beats one that
    will not render. The CATALOGUE is not a column — it is the ROW SET. A catalogue read that
    failed quietly would drop the grid straight back to the sold-only 2,717 with every gate
    green and nothing on screen saying so, which is the defect this seam exists to end
    ([[gate-can-report-green-on-nothing]]). `routes_products._pool_for` already turns the raise
    into a 503 that names the cause, so the loud failure has somewhere honest to land.

    It exists as a `pd`-level function rather than an inline `products.catalogue()` call for the
    same reason `_inventory_by_code` does: it is the seam `verify_perm_scope`'s section H stubs
    to build a pool without Odoo.
    """
    return products.catalogue()


def _pricelist_by_code():
    """`({code: {price_*: price}}, report)` from `products.pricelist_by_code`, or `({}, …)`.

    A SEAM for the same two reasons `_inventory_by_code` is one: it degrades rather than raises
    (these are columns, not the row set), and `verify_perm_scope`'s section H stubs it to build a
    pool without Odoo. ⛔ Unstubbed there, section H would reach live Odoo through the back door
    and the whole file would stop being runnable offline.
    """
    try:
        return products.pricelist_by_code()
    except Exception as e:
        return {}, {"error": f"{type(e).__name__}: {str(e)[:200]}"}


def pool(team_id=None, t=None):
    """One row per ACTIVE product — the PRODUCT analogue of `customer_data.pool`.

    CATALOGUE-FIRST, REVENUE LEFT-JOINED (R12 — see decision 4 in the module header). The row set
    is `products.catalogue()`; `products.directory()` supplies the revenue columns for the SKUs
    that sold in its window and contributes NO rows of its own.

    `team_id` shapes the revenue columns exactly as it does for customers (`products.directory`
    passes it into `_sku_rev`), which is why `core.perm_scope.derive_pool_scope` must keep
    driving it rather than a post-filter deciding the BU. It does NOT shape the row set: a
    catalogue has no team.
    """
    t = t or P.today()
    consolidated = team_id is None
    # ⭐ OWNER 2026-08-11: read inventory on EVERY pull, not just a consolidated one. The stock
    # itself is company-wide; only the sales-derived half is re-scoped, by `_rescope_inventory`.
    inv = _inventory_by_code(t)
    bu_share = {} if consolidated else _bu_ltm_share(t, team_id)
    sup = supplier_master()
    prices, _price_report = _pricelist_by_code()
    cat = _catalogue_by_code()
    # The LEFT side of the join, indexed by the same code key. ⛔ A code here that the catalogue
    # does not carry belongs to a product no ACTIVE record claims — archived, and R12 keeps those
    # out. `validate()` asserts that of Odoo rather than assuming it, and reports the revenue
    # that therefore sits outside the grid (MEASURED 2026-08-11: 3 codes, $0.00 YTD / $294.50 LY).
    rev = {r["code"]: r for r in products.directory(t=t, team_id=team_id)}

    rows = []
    for code, meta in cat.items():
        r = rev.get(code)
        row = {
            "pid": sku_pid(code),
            "code": code,
            # ⭐ ONE source for the name and the category, the CATALOGUE — not the sale line's
            # m2o. For a re-SKUed code the line's name can be the ARCHIVED record's; the active
            # record's `display_name` is the current truth, and it is the same string for every
            # SKU that is not re-SKUed. `directory()` derives the category identically.
            "product": meta.get("product") or code,
            "category": meta.get("category") or "(uncategorized)",
            # ⭐ R12 — BLANK, NEVER $0, on a SKU no revenue read ever saw. A row that IS in `rev`
            # keeps its measured numbers including a real 0.0 (it sold last year, not this one).
            "rev_ytd": r.get("rev_ytd", 0.0) if r else None,
            "rev_ly": r.get("rev_ly", 0.0) if r else None,
            "yoy_pct": r.get("yoy_pct") if r else None,
            "qty_ytd": r.get("qty_ytd", 0.0) if r else None,
            "orders_ytd": r.get("orders_ytd", 0) if r else None,
        }
        # ⭐ WAVE 30 W30-T34 — the PRICELIST stratum, one column per declared list. Catalogue
        # data like the supplier block, so it rides every pull, scoped or not: a price book is
        # not a thing a business unit owns a slice of, and a Fisch reader who cannot see the
        # Fisch price is exactly who this column is for.
        #
        # ⛔ `None`, NEVER 0 AND NEVER A FALLBACK, for the 1,883 SKUs no list prices. That is
        # W29-T52's own negative control: the single `3_global` rule computes over `list_price`,
        # which is 1.00 on 5,817 of 5,871 products, so the fallback is not a cheaper answer —
        # it is a wrong one wearing a currency sign.
        p = prices.get(code) or {}
        for _col, _name in products.PRICELIST_COLUMNS:
            row[_col] = p.get(_col)
        # Wave 17 R3 — the supplier master, on every pull (it is catalogue data, not stock).
        s = sup.get(code) or {}
        row.update({
            "supplier": s.get("supplier") or "(none)",
            "lead_days": s.get("lead_days"),
            "origin_country": s.get("origin_country") or "(none)",
            "first_cost": s.get("first_cost"),
        })
        e = inv.get(code) or {}
        if consolidated:
            qty_ltm, dos, bucket = e.get("qty_ltm"), e.get("dos"), e.get("bucket")
        else:
            qty_ltm, dos, bucket = _rescope_inventory(e, bu_share.get(code, _DEFAULT_SHARE))
        row.update({
            # UNSCOPED on purpose (owner 2026-08-11): one warehouse, no per-brand shelf.
            "on_hand": e.get("on_hand"),
            "unit_cost": e.get("unit_cost"),
            "inv_value": e.get("inv_value"),
            # SCOPED: these three are functions of how fast THIS unit sells the SKU.
            "qty_ltm": qty_ltm,
            "dos": dos,
            "stock_bucket": bucket,
        })
        # THE BUY TRIGGER (R3), ported from `modules/procurement`'s lead-time-cover rule:
        # buy when the shelf runs out before a reorder could land.
        #
        # ⚠ BLANK, NOT "OK", WHEN EITHER INPUT IS MISSING. 339 SKUs have no supplier and 400
        # no lead time; `dos` is null for anything that never sells through. "We don't know"
        # and "you're fine" are different sentences, and only one of them is safe to print
        # beside a purchasing decision.
        #
        # ⛔ `buy_now` SHIPS NOWHERE ANY MORE, and is computed anyway. OWNER 2026-08-03: it
        # is not a preset field — it is a formula over the two columns beside it, and the
        # platform has a formula field type for exactly that. Its Field is gone from
        # `aios_grid_fields.json`, and `rows_from_pool` projects rows STRICTLY through that
        # contract, so no Field means no cell on the wire. Nothing renders this key.
        #
        # It stays computed because deleting it would delete the PROOF. `validate()` below
        # reconciles the formula's predicate against this one, row for row, which is the only
        # thing that makes "the same exact figures" a claim rather than an assertion — and
        # this repo has the scar already (wave 17: archiving `ar` "would have DELETED the
        # proof"). Two comparisons per SKU is what that costs.
        #
        # ⭐ READS THE SCOPED `dos`, NOT `e['dos']`. On a Fisch pull the cover gap must answer
        # "does the shelf outlast a reorder AT FISCH'S RATE" — reading the consolidated figure
        # here would print a buy signal computed from both units' velocity beside a days-of-supply
        # computed from one, and the two columns would disagree on the same row.
        lead = s.get("lead_days")
        if isinstance(dos, (int, float)) and isinstance(lead, (int, float)) and lead > 0:
            row["cover_gap_d"] = int(round(dos - lead))
            row["buy_now"] = "Buy now" if dos < lead else "OK"
        else:
            row["cover_gap_d"] = None
            row["buy_now"] = None
        rows.append(row)

    # `or 0.0` reads a BLANK as zero FOR SORTING ONLY — never for the cell. A never-sold SKU
    # settles among the zero-revenue ones at the bottom, which is where it belongs; the stored
    # value stays None so nothing downstream can mistake "we never saw a sale" for "we measured
    # nothing sold".
    rows.sort(key=lambda r: -(r["rev_ytd"] or 0.0))
    _assert_no_pid_collision(rows)
    return rows


#: ⭐ THE BUY SIGNAL, AS THE FORMULA FIELD EVALUATES IT (owner, 2026-08-03).
#:
#: The owner's ruling was that "Buy signal" is not a preset field — it is a formula over two
#: columns the product table already carries, and the platform has a formula field type for it.
#: The formula, verbatim, is what `_seed_wave17.BUY_SIGNAL_FORMULA` creates and what the JSON
#: contract's `_product_removed_buy_now` note records:
#:
#:     IF({lead_days} > 0, IF({dos} < {lead_days}, "Buy now", "OK"), "")
#:
#: This function is a PORT of how `customer-grid/formulaEngine.ts` evaluates that tree, not a
#: restatement of the business rule — that is the whole point, because the two could drift and
#: `validate()` is where the drift must show. Three engine behaviours it reproduces exactly:
#:
#:   · a `ref` to a missing/non-numeric cell is None (`case "ref"` returns null for anything
#:     that is not a finite number or a string);
#:   · `cmp` returns BLANK unless BOTH sides read as numbers — it never coerces a blank to 0,
#:     which is the difference between this and a filter engine's `toNum(null) === 0`;
#:   · `IF` with a non-boolean condition returns BLANK ("no truthiness"), so a blank comparison
#:     propagates out as a blank cell rather than taking the false branch.
BUY_SIGNAL_FORMULA = 'IF({lead_days} > 0, IF({dos} < {lead_days}, "Buy now", "OK"), "")'


def _buy_signal_formula(row):
    """Evaluate `BUY_SIGNAL_FORMULA` over one pool row -> 'Buy now' | 'OK' | '' (blank)."""
    def num(v):
        # The engine's `ref` + `asNumber`: booleans are not values here, and a non-finite
        # number is null. `isinstance(True, int)` is True in Python, so bool is excluded first.
        if isinstance(v, bool) or not isinstance(v, (int, float)):
            return None
        return v if v == v and v not in (float('inf'), float('-inf')) else None

    lead, dos = num(row.get("lead_days")), num(row.get("dos"))
    if lead is None:                 # `{lead_days} > 0` is blank -> IF(blank, …) is blank
        return ""
    if not lead > 0:
        return ""                    # the formula's own else-branch
    if dos is None:                  # `{dos} < {lead_days}` is blank -> IF(blank, …) is blank
        return ""
    return "Buy now" if dos < lead else "OK"


def validate(team_id=None, t=None):
    """Reconcile what SHIPS to an independent aggregate — the platform's own rule that a number
    which does not tie to Odoo does not ship.

    ⛔⛔ THE POPULATION LEG IS NEW, AND IT IS HERE BECAUSE THIS FUNCTION USED TO BE SELF-SEALING
    (wave 29, contract C8). It reconciled Σ per-SKU YTD revenue against `products._sku_rev` — the
    same function `pool()` built its rows from. BOTH SIDES CAME FROM THE JOIN, so the oracle could
    not see a missing row, and a grid holding 2,717 of 5,875 active products passed for a wave.
    No gate anywhere asserted a pool row COUNT against an independent Odoo `search_count` either.
    That is the leg below, and it is the control that would have caught it
    ([[no-unverifiable-aggregates]]).

    ⚠ The revenue leg was NOT deleted with it — it answers a different question ("did the join
    lose money?") and is still the right shape for that one. What changed is that it now has to
    account for revenue belonging to codes NO ACTIVE PRODUCT CARRIES, because R12 keeps archived
    products out of the grid; the total is decomposed rather than compared loosely, so a growing
    "outside the grid" figure shows up as a number rather than as slack in a tolerance.

    Still deliberately NOT covered, and said rather than implied: the inventory columns are a
    different module's reconciliation, and they are absent on a scoped pull anyway.
    """
    t = t or P.today()
    rows = pool(team_id=team_id, t=t)

    # ⭐⭐ THE POPULATION, AGAINST AN ORACLE THAT CANNOT SEE OUR JOIN. A bare Odoo count of active
    # products, asked fresh — not a `len()` over anything this module or `products.catalogue()`
    # built. NC: drop one row from `pool()` and this goes red; that is the whole point of it.
    n_active = products.catalogue_count()
    checks = [{
        "check": "the pool holds one row per ACTIVE product (R12) — row count == an INDEPENDENT "
                 "Odoo search_count('product.product', active=True)",
        "ours": len(rows), "theirs": n_active, "ok": len(rows) == n_active,
    }]

    yf, yt = P.ytd(t)
    sku_rev = products._sku_rev(yf, yt, team_id) or {}
    in_pool = {r["code"] for r in rows}
    outside = {c: v for c, v in sku_rev.items() if c not in in_pool}
    ours = round(sum(r["rev_ytd"] or 0.0 for r in rows), 2)
    theirs = round(sum(v.get("rev", 0.0) for v in sku_rev.values()), 2)
    dropped = round(sum(v.get("rev", 0.0) for v in outside.values()), 2)
    checks.append({
        "check": "Σ per-SKU YTD revenue in the grid + Σ YTD revenue of codes no ACTIVE product "
                 "carries == modules.products._sku_rev over the same window",
        "ours": round(ours + dropped, 2), "theirs": theirs,
        "ok": abs((ours + dropped) - theirs) < 0.01,
        "detail": {"in_the_grid": ours, "outside_the_grid": dropped,
                   "codes_outside": len(outside)},
    })
    # ⛔ AND THE EXCLUSION IS ASSERTED, NOT ASSUMED. "Everything I dropped was archived" is
    # trivially true when checked against the dict that did the dropping — that is the
    # self-sealing shape all over again. So it is asked of ODOO. A code here that a live ACTIVE
    # product carries means `products.catalogue()` missed a row that has revenue, which is the
    # 2,717 defect returning in a smaller costume.
    # ⚠ `pid:N` keys are not `default_code`s and are skipped: an UNCODED active product is in the
    # catalogue under its own `pid:N` key and therefore cannot be in `outside` at all.
    coded_outside = sorted(c for c in outside if not str(c).startswith("pid:"))
    still_active = (O.get_odoo().search_count(
        'product.product', [('active', '=', True), ('default_code', 'in', coded_outside)])
        if coded_outside else 0)
    checks.append({
        "check": "every SKU with revenue but NO grid row is genuinely ARCHIVED (R12: archived "
                 "stay out) — asked of Odoo, never of the catalogue that did the dropping",
        "ours": still_active, "theirs": 0, "ok": still_active == 0,
        "detail": {"codes": coded_outside[:10], "n_codes": len(outside)},
    })
    # ── ⭐⭐ WAVE 30 W30-T34 — THE PRICELIST COLUMNS, AGAINST ORACLES THAT CANNOT SEE OUR JOIN ──
    #
    # Three legs, because the column can fail in three different ways and only one of them is a
    # count. `products.pricelist_by_code` builds a `{code: {col: price}}` map by expanding each
    # rule over the products it names, IN PYTHON — that expansion is the fragile part, so every
    # leg below re-asks ODOO instead of re-reading the map.
    price_report = _pricelist_by_code()[1]
    _pl_rows = O.search_read('product.pricelist', [], ['id', 'name'])
    _pl_id = {}
    for _p in _pl_rows:
        _pl_id.setdefault(str(_p.get('name') or '').strip(), _p['id'])
    _today = P.today().isoformat()

    _targets_memo = {}

    def _rule_targets(list_name):
        """(variant_ids, template_ids) a pricelist prices today — read FRESH from Odoo.

        Memoised for the LIFE OF THIS CALL only: legs 1 and 2 both need all three lists, and
        without this `validate()` makes six identical round trips instead of three (measured:
        ~48s of the run). Deliberately NOT an `lru_cache` — an oracle that survives the process
        is an oracle reading yesterday's Odoo.
        """
        if list_name in _targets_memo:
            return _targets_memo[list_name]
        var, tmpl = set(), set()
        plid = _pl_id.get(list_name)
        if plid is None:
            return var, tmpl
        for r in O.search_read(
                'product.pricelist.item',
                [('pricelist_id', '=', plid), ('compute_price', '=', 'fixed'),
                 ('applied_on', 'in', ['0_product_variant', '1_product'])],
                ['product_id', 'product_tmpl_id', 'applied_on', 'fixed_price',
                 'date_start', 'date_end']):
            ds, de = str(r.get('date_start') or '')[:10], str(r.get('date_end') or '')[:10]
            if (ds and ds > _today) or (de and de < _today):
                continue
            fp = r.get('fixed_price')
            if not isinstance(fp, (int, float)) or fp <= 0:
                continue
            if r.get('applied_on') == '0_product_variant' and r.get('product_id'):
                var.add(O.m2o_id(r['product_id']))
            elif r.get('product_tmpl_id'):
                tmpl.add(O.m2o_id(r['product_tmpl_id']))
        _targets_memo[list_name] = (var, tmpl)
        return var, tmpl

    # LEG 1 — COVERAGE PER LIST. Ours: cells we filled. Theirs: a bare Odoo `search_count` of
    # ACTIVE products a fresh read of that list's rules reaches. The rule set is shared (it IS
    # the data) but the EXPANSION is not, and the expansion is what breaks.
    for _col, _name in products.PRICELIST_COLUMNS:
        _var, _tmpl = _rule_targets(_name)
        _dom = [('active', '=', True), '|', ('id', 'in', sorted(_var)),
                ('product_tmpl_id', 'in', sorted(_tmpl))]
        theirs = O.get_odoo().search_count('product.product', _dom) if (_var or _tmpl) else 0
        ours = sum(1 for r in rows if isinstance(r.get(_col), (int, float)))
        checks.append({
            "check": f"{_name} pricelist: SKUs priced in the grid == an INDEPENDENT Odoo count "
                     f"of ACTIVE products its date-valid fixed rules reach",
            "ours": ours, "theirs": theirs, "ok": ours == theirs,
            "detail": {"column": _col, "variant_rules": len(_var), "template_rules": len(_tmpl)},
        })

    # LEG 2 — ⛔ THE TICKET'S OWN NEGATIVE CONTROL, AS A LEG. A SKU with no specific item must
    # render BLANK. Asked of ODOO, never of the map that produced the blank — "everything I left
    # empty was genuinely unpriced" is trivially true when checked against the dict that emptied
    # it, which is the self-sealing shape the population leg above exists to end.
    _unpriced = [r for r in rows
                 if not any(isinstance(r.get(c), (int, float))
                            for c, _n in products.PRICELIST_COLUMNS)]
    _codes = {r["code"] for r in _unpriced}
    _stray = 0
    if _codes:
        # Resolve those codes back to Odoo ids and ask whether ANY declared list prices them.
        _ids, _tmpls = set(), set()
        for _p in O.search_read('product.product',
                                [('active', '=', True),
                                 ('default_code', 'in', sorted(c for c in _codes
                                                               if not c.startswith("pid:")))],
                                ['id', 'product_tmpl_id']):
            _ids.add(_p['id'])
            _tmpls.add(O.m2o_id(_p.get('product_tmpl_id')))
        for _col, _name in products.PRICELIST_COLUMNS:
            _v, _t = _rule_targets(_name)
            _stray += len((_v & _ids) | ({t for t in _tmpls if t in _t}))
    checks.append({
        "check": "every SKU rendering a BLANK price is genuinely unpriced on all three declared "
                 "lists (W29-T52's NC: no fallback dressed as a price) — asked of Odoo",
        "ours": _stray, "theirs": 0, "ok": _stray == 0,
        "detail": {"blank_skus": len(_unpriced), "priced_skus": len(rows) - len(_unpriced)},
    })

    # LEG 3 — ⭐ R6's SECOND SENTENCE, AS A NUMBER. *"If there is lag or it can't be done, you
    # need to explicitly tell me why and recommend a fix."* Everything the reader cannot see is
    # counted here rather than dropped. The ASSERTION is the one thing that must never be true —
    # a price cell that is present and not a positive number, i.e. a "free" SKU — while the
    # rest rides `detail` so it is reported without reddening an honest day's data.
    _bad = [r["code"] for r in rows
            for c, _n in products.PRICELIST_COLUMNS
            if r.get(c) is not None and not (isinstance(r.get(c), (int, float)) and r[c] > 0)]
    # ⛔ MEASURED 2026-08-12 AND NOT FIXABLE FROM THIS FENCE: `aios_grid.rows_from_pool` sends
    # every non-text field through `_round(v)` = `round(v)` with no ndigits, i.e. to a whole
    # dollar. 52.1% of fixed prices carry cents and the median relative error is 5.26%, so this
    # column ties to the cent HERE and ships rounded. Reported, never silently enforced.
    _cents = sum(1 for r in rows for c, _n in products.PRICELIST_COLUMNS
                 if isinstance(r.get(c), (int, float)) and abs(r[c] - round(r[c])) > 1e-9)
    _to_zero = sum(1 for r in rows for c, _n in products.PRICELIST_COLUMNS
                   if isinstance(r.get(c), (int, float)) and round(r[c]) == 0)
    checks.append({
        "check": "no price cell is present-but-not-a-positive-number (a 0 would read as FREE; "
                 "an unset rule must be blank), and what the reader cannot see is COUNTED",
        "ours": len(_bad), "theirs": 0, "ok": not _bad,
        "detail": {"bad_cells": _bad[:10], "reader_report": price_report,
                   "wire_rounding_loses_cents_on": _cents,
                   "wire_rounding_to_zero_dollars": _to_zero},
    })

    # WAVE 17 R3 — the BUY SIGNAL must be a total, exact partition of the catalogue, and every
    # member of it must be re-derivable from the two columns beside it. A signal somebody buys
    # stock on cannot be "mostly right": the failure that matters is a row that says OK because
    # an input was missing, so the blank leg is checked as hard as the other two.
    if team_id is None:                      # consolidated only — the inputs exist only there
        buy = [r for r in rows if r.get("buy_now") == "Buy now"]
        ok_rows = [r for r in rows if r.get("buy_now") == "OK"]
        blank = [r for r in rows if r.get("buy_now") is None]
        mis = sum(1 for r in buy if not (r["dos"] < r["lead_days"]))
        mis += sum(1 for r in ok_rows if not (r["dos"] >= r["lead_days"]))
        # A blank must be UNKNOWN — never a row we could have answered and quietly did not.
        mis += sum(1 for r in blank
                   if isinstance(r.get("dos"), (int, float))
                   and isinstance(r.get("lead_days"), (int, float)) and r["lead_days"] > 0)
        checks.append({
            "check": "Buy signal partitions the catalogue (buy + ok + unknown == rows, none "
                     "misclassified)",
            "ours": len(buy) + len(ok_rows) + len(blank) - mis, "theirs": len(rows),
            "ok": mis == 0 and len(buy) + len(ok_rows) + len(blank) == len(rows),
            "detail": {"buy_now": len(buy), "ok": len(ok_rows), "unknown": len(blank),
                       "misclassified": mis},
        })
        # ⭐ OWNER 2026-08-03 — "use the formula fields to come up to the same EXACT figures".
        #
        # This is that sentence, as a check. The retired preset column and the formula that
        # replaces it must agree on every SKU, in all three states, or the replacement is not a
        # replacement. `_buy_signal_formula` is a line-by-line port of the client formula
        # engine's evaluation of
        #
        #     IF({lead_days} > 0, IF({dos} < {lead_days}, "Buy now", "OK"), "")
        #
        # including the part that is easy to get wrong: a comparison against a BLANK is blank,
        # never a coerced 0. (The client engine's `cmp` returns null unless both sides read as
        # numbers, and `IF` refuses a non-boolean condition — so a missing `dos` yields "" and
        # not "Buy now". A filter engine would have said `0 < 30` and swept in every SKU that
        # never sells through; the formula engine does not, and this check is what holds it.)
        #
        # ⚠ It compares SETS OF SKUs, not counts. Two different partitions can share a shape.
        disagree = sorted(r["code"] for r in rows
                          if (r.get("buy_now") or "") != _buy_signal_formula(r))
        checks.append({
            "check": 'Buy signal as a FORMULA field == the retired preset column, per SKU '
                     '(IF({lead_days} > 0, IF({dos} < {lead_days}, "Buy now", "OK"), ""))',
            "ours": len(rows) - len(disagree), "theirs": len(rows),
            "ok": not disagree,
            "detail": {"disagreeing_skus": disagree[:10], "n_disagree": len(disagree)},
        })
        # The other half of "the same figures": the SHARED Buy list view can no longer filter on
        # the retired column, and its replacement conditions must select the same SKUs. They are
        # `dos isNotEmpty AND lead_days isNotEmpty AND lead_days > 0 AND dos < lead_days`
        # (_seed_wave17.views), so this reproduces exactly that conjunction.
        #
        # ⛔ THE NEGATIVE CONTROL IS WHY THIS IS NOT `cover_gap_d < 0`, which reads like the
        # obvious filter and is WRONG: `cover_gap_d` is `int(round(dos - lead))`, so a genuine
        # gap of -0.4 days rounds to 0 and that SKU drops off a buy list it belongs on.
        view_rows = {r["code"] for r in rows
                     if isinstance(r.get("dos"), (int, float))
                     and isinstance(r.get("lead_days"), (int, float))
                     and r["lead_days"] > 0 and r["dos"] < r["lead_days"]}
        signal_rows = {r["code"] for r in buy}
        rounding_would_miss = sorted(
            c for c in signal_rows
            if next((r for r in rows if r["code"] == c), {}).get("cover_gap_d") == 0)
        checks.append({
            "check": "Buy list view conditions (dos/lead_days, no retired column) select "
                     "exactly the Buy-now SKUs",
            "ours": len(view_rows), "theirs": len(signal_rows),
            "ok": view_rows == signal_rows,
            "detail": {"only_in_view": sorted(view_rows - signal_rows)[:10],
                       "only_in_signal": sorted(signal_rows - view_rows)[:10],
                       # Reported, not asserted: how many SKUs a `cover_gap_d < 0` filter would
                       # have silently dropped. 0 today does not make that filter correct.
                       "cover_gap_rounds_to_zero": len(rounding_would_miss)},
        })
    return checks