diff --git "a/platform/modules/products.py" "b/platform/modules/products.py"
--- "a/platform/modules/products.py"
+++ "b/platform/modules/products.py"
@@ -1,1005 +1,1077 @@
-"""Products / SKU module — SKU health: YoY movers (risers & decliners), zombie SKUs
-(catalog rot — sellable, formerly selling, now dead), new winners, coverage collapse
-(SKUs losing customer breadth — the FFS-recovery early-warning signal), and velocity leaders.
-
-Per-SKU margin already lives in the Financial module (low_margin_skus); not duplicated here.
-Basket / co-purchase (MBA) is intentionally deferred — the existing client app computes it
-runtime-side, and a local co-occurrence pull over 74k LTM lines is the kind of heavy job the
-project guardrails keep off the local PC. (See BACKLOG.)
-
-All line-level (sale.order.line), RI+FFS scope, excluded accounts removed — reusing sale_line_domain.
-Coverage (distinct customers per SKU) uses a 2-level read_group and is a touch slow (~15s/
-window), so it's its own function the app calls lazily and caches.
-"""
-import sys
-import datetime as dt
-from pathlib import Path
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from functools import lru_cache
-import core.odoo as O
-import core.periods as P
-import modules.sales as sales_mod
-
-# SKU-health views are about real products — exclude service/delivery pseudo-SKUs
-# (Delivery Charges otherwise dominate movers/winners). Dot-path FILTER works (groupby
-# on the dot-path does not). Validation reconciles on this same non-service universe.
-_NO_SVC = [('product_id.type', '!=', 'service')]
-
-
-@lru_cache(maxsize=1)
-def _code_map():
-    """product_id → SKU code. Multiple product records can share a default_code (re-SKUing
-    /duplicates); keying by code merges them so a re-coded item doesn't read as a fake
-    decliner + fake riser. Products without a code fall back to a per-id key. Archived
-    products are INCLUDED — re-SKUing typically archives the old record and creates a new
-    one under the same code, and the old record still carries last-year sales."""
-    prods = O.search_read('product.product', [('active', 'in', [True, False])], ['id', 'default_code'],
-                          limit=50000)
-    return {p['id']: (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
-            for p in prods}
-
-
-def _sku_rev(date_from, date_to, team_id=None):
-    """{sku_code: {'name','rev','qty','orders'}} over a window (services excluded,
-    duplicate product records merged by SKU code)."""
-    g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC),
-                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False)
-    codes = _code_map()
-    out = {}
-    for r in g:
-        pid = O.m2o_id(r.get('product_id'))
-        if not pid:
-            continue
-        key = codes.get(pid, f"pid:{pid}")
-        e = out.setdefault(key, {'name': O.m2o_name(r.get('product_id')),
-                                 'rev': 0.0, 'qty': 0.0, 'orders': 0})
-        e['rev'] += r.get('price_subtotal') or 0.0
-        e['qty'] += r.get('product_uom_qty') or 0.0
-        e['orders'] += r.get('__count') or 0
-    return out
-
-
-def yoy_movers(t=None, limit=20, team_id=None):
-    """Top SKU risers and decliners by YTD-vs-same-period-LY revenue change."""
-    t = t or P.today()
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    this = _sku_rev(yf, yt, team_id)
-    last = _sku_rev(lf, lt, team_id)
-    rows = []
-    for pid in set(this) | set(last):
-        tr = this.get(pid, {'rev': 0.0, 'name': last.get(pid, {}).get('name', '')})
-        lr = last.get(pid, {'rev': 0.0})
-        name = this.get(pid, {}).get('name') or last.get(pid, {}).get('name') or ''
-        rows.append({'code': pid, 'product': name, 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'],
-                     'change': tr['rev'] - lr['rev']})
-    risers = sorted([r for r in rows if r['change'] > 0], key=lambda x: -x['change'])[:limit]
-    decliners = sorted([r for r in rows if r['change'] < 0], key=lambda x: x['change'])[:limit]
-    return {'risers': risers, 'decliners': decliners}
-
-
-def zombie_skus(t=None, limit=30, min_prior=1500.0, team_id=None):
-    """Catalog rot: SKUs that sold materially last year but are ~dead this year."""
-    t = t or P.today()
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    this = _sku_rev(yf, yt, team_id)
-    last = _sku_rev(lf, lt, team_id)
-    rows = []
-    for pid, lr in last.items():
-        if lr['rev'] < min_prior:
-            continue
-        tr = this.get(pid, {'rev': 0.0})
-        if tr['rev'] > 0.05 * lr['rev']:   # still selling at >5% of prior → not a zombie
-            continue
-        rows.append({'code': pid, 'product': lr['name'], 'rev_ly': lr['rev'], 'rev_ytd': tr['rev'],
-                     'lost': lr['rev'] - tr['rev']})
-    rows.sort(key=lambda x: -x['lost'])
-    return rows[:limit]
-
-
-def new_winners(t=None, limit=20, min_this=1500.0, team_id=None):
-    """SKUs that barely sold last year but are selling well this year."""
-    t = t or P.today()
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    this = _sku_rev(yf, yt, team_id)
-    last = _sku_rev(lf, lt, team_id)
-    rows = []
-    for pid, tr in this.items():
-        if tr['rev'] < min_this:
-            continue
-        lr = last.get(pid, {'rev': 0.0})
-        if lr['rev'] > 0.05 * tr['rev']:
-            continue
-        rows.append({'code': pid, 'product': tr['name'], 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'],
-                     'gained': tr['rev'] - lr['rev']})
-    rows.sort(key=lambda x: -x['gained'])
-    return rows[:limit]
-
-
-def velocity_leaders(t=None, limit=25, team_id=None):
-    """Top SKUs by LTM unit velocity (units/month) and revenue."""
-    t = t or P.today()
-    lf, lt = P.ltm(t)
-    sku = _sku_rev(lf, lt, team_id)
-    rows = [{'code': k, 'product': v['name'], 'units_ltm': v['qty'], 'units_per_mo': v['qty'] / 12.0,
-             'rev_ltm': v['rev'], 'orders_ltm': v['orders']} for k, v in sku.items()]
-    rows.sort(key=lambda x: -x['units_ltm'])
-    return rows[:limit]
-
-
-def _code_category():
-    """code -> category name. Built from _code_map (pid->code) + the sales category map
-    (pid->category); the first record carrying a code sets that code's category."""
-    codes = _code_map()
-    cats = sales_mod._product_cat()
-    out = {}
-    for pid, code in codes.items():
-        if code not in out and pid in cats:
-            out[code] = cats[pid]
-    return out
-
-
-def directory(t=None, team_id=None):
-    """Every SKU that sold this YTD or the same period last year, with the fields the SKU
-    drill-down filters/sorts on: category, YTD/LY revenue, YoY %, units and orders. Mirrors the
-    customer directory — the full list (filtering happens in the UI), sorted by YTD revenue."""
-    t = t or P.today()
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    this = _sku_rev(yf, yt, team_id)
-    last = _sku_rev(lf, lt, team_id)
-    code_cat = _code_category()
-    rows = []
-    for code in set(this) | set(last):
-        tr = this.get(code, {})
-        lr = last.get(code, {})
-        rev_ytd = tr.get('rev', 0.0)
-        rev_ly = lr.get('rev', 0.0)
-        rows.append({'code': code, 'product': tr.get('name') or lr.get('name') or code,
-                     'category': code_cat.get(code, '(uncategorized)'),
-                     'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'change': rev_ytd - rev_ly,
-                     'yoy_pct': P.yoy_pct(rev_ytd, rev_ly),
-                     'qty_ytd': tr.get('qty', 0.0), 'orders_ytd': tr.get('orders', 0)})
-    rows.sort(key=lambda r: -r['rev_ytd'])
-    return rows
-
-
-def catalogue():
-    """`{sku_code: {'product', 'category', 'id', 'discontinued', 'incoming'}}` — EVERY ACTIVE
-    product, sold or not.
-
-    ⛔ THIS IS THE CATALOGUE UNIVERSE, AND IT IS DELIBERATELY NOT `directory()`. `directory()`'s
-    row set IS the union of two revenue `read_group`s over `sale.order.line` (`:155`), so a SKU
-    that never sold cannot exist in it. That is CORRECT for its own callers — `yoy_movers`,
-    `zombie_skus`, `new_winners` and `categories` all legitimately want a sales-window universe —
-    and it is wrong for the PRODUCT GRID, which is a catalogue and was therefore showing 2,717 of
-    5,875 SKUs (wave 29, owner item 22 / ruling R12). The fix is this function plus a LEFT JOIN in
-    the consumer, never a window removed from `directory()`: removing the window alone lands at
-    3,327 (all-time-sold), because ~2,550 active SKUs have never sold in wholesale scope at all.
-
-    ⚠ NOT BU-SHAPED, and it cannot be: `product.product` carries no team. A catalogue is one
-    catalogue. `directory()` stays the BU-shaped half, which is why the two are joined rather than
-    merged — a scoped caller gets every SKU with ITS OWN revenue, blank where that BU never sold.
-
-    Keyed exactly like `_code_map()` — the `default_code`, or a `pid:N` fallback for the 33 active
-    records that carry none — so the join against `directory()` is code-for-code with no
-    normaliser. Names come from `display_name` (`[CODE] NAME`), which is what `O.m2o_name` yields
-    off a sale line, so a never-sold row wears the same format as a sold one.
-
-    ⛔ RAISES on a truncated read rather than returning a short catalogue. A silently short pull
-    would put the grid back at a plausible wrong number with every gate green — the exact failure
-    this function exists to end ([[no-unverifiable-aggregates]], and the same truncation guard
-    `modules/backorders.py:161` already uses).
-    """
-    dom = [('active', '=', True)]
-    # ⭐ `product_tag_ids` and `incoming_qty` RIDE THIS READ rather than paying for their own.
-    # This function already pulls every active product once; a second full pull for either column
-    # costs ~6 s on somebody's first page load for data that is already in flight.
-    # ⚠ `incoming_qty` is an UNSTORED computed field. It reads fine per record, but Odoo refuses
-    # to aggregate it (`Fault 2: Cannot aggregate field 'incoming_qty'`), so nothing downstream
-    # may push a filter or a read_group on it back to Odoo. It is materialised into the pool here
-    # and summed on our side; `validate()` reconciles that sum against purchase-order lines.
-    prods = O.search_read('product.product', dom,
-                          ['id', 'default_code', 'display_name', 'name',
-                           'product_tag_ids', 'incoming_qty'],
-                          limit=50000)
-    n = O.get_odoo().search_count('product.product', dom)
-    if len(prods) != n:
-        raise ValueError(
-            f"products.catalogue: the product pull is TRUNCATED — read {len(prods)} rows against "
-            f"a search_count of {n}. A short catalogue renders as a plausible smaller grid with "
-            f"nothing reporting it; raise the limit before shipping this.")
-    code_cat = _code_category()
-    disc_tag = discontinued_tag_id()
-    out = {}
-    for p in prods:
-        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
-        # First record wins, matching `_code_category`'s own convention.
-        # ⛔ THE 2026-08-11 MEASUREMENT ("zero active products share a code, so this branch is a
-        # guard, not a merge policy") HAS EXPIRED. RE-MEASURED 2026-08-19: **5,873 active products
-        # carry 5,872 distinct codes** because `2112-12` is now on two records. So this IS a merge
-        # policy today, and it is the standing reason `validate()`'s row-count leg reads 5872
-        # against an Odoo `search_count` of 5873, along with all three pricelist legs being short
-        # by exactly one. Those reds are this duplicate, not a truncated read.
-        # ⚠ Which is also why `incoming` ACCUMULATES below instead of taking the first record's
-        # value: a merge policy that keeps one record's name must still sum the other's stock.
-        # ⭐⭐ W33-T43 (R2 / amendment A2) — `id` RIDES THE ROW. Odoo's `product.product` id is
-        # already in hand (the `pid:{id}` fallback above uses it and then threw it away), and it is
-        # the ONE column `ut_odoo_products` had that `product_data` lacked. R2 merges onto the
-        # legacy key and keeps every data column, so retiring the twin without this would lose a
-        # real identifier rather than a navigational link.
-        # ⚠ A product's grid pid is `crc32(default_code)`, NOT the Odoo id — unlike a CUSTOMER row,
-        # whose pid IS the partner id. That asymmetry is exactly why this cannot be derived
-        # downstream in `aios_grid.py` and has to be carried from the source read.
-        row = out.setdefault(code, {'product': p.get('display_name') or p.get('name') or code,
-                                    'category': code_cat.get(code, '(uncategorized)'),
-                                    'id': p['id'],
-                                    'discontinued': 'No', 'incoming': 0.0})
-        # ⛔ ACROSS EVERY RECORD SHARING THE CODE, not just the first one that won above. A SKU
-        # whose variants are separate records carries its inbound on whichever record the PO named,
-        # and 'first record wins' would drop the rest — a buy list that under-counts what is
-        # already on the water tells you to re-order stock you have already bought.
-        row['incoming'] += float(p.get('incoming_qty') or 0.0)
-        if disc_tag is not None and disc_tag in (p.get('product_tag_ids') or ()):
-            row['discontinued'] = 'Yes'
-    return out
-
-
-#: ⭐ The Odoo product tag that marks a SKU withdrawn from the line. Owner, 2026-08-19: *"we have
-#: a status Field called discontinued from Odoo. Surface it right now so we can filter correctly."*
-#: MEASURED that day: it is not a FIELD at all — `product.product` has no `discontinued` column and
-#: no selection carrying the word. It is `product.tag`, on **497 of 5,873 active products**, every
-#: one of them `active=True` and `sale_ok=True`, so nothing else in Odoo distinguishes them and the
-#: buy list was quoting them as live SKUs.
-DISCONTINUED_TAG = 'Discontinued'
-
-
-@lru_cache(maxsize=1)
-def discontinued_tag_id():
-    """The `product.tag` id for `DISCONTINUED_TAG`, or `None` if the tag is not there.
-
-    ⚠ RESOLVED BY NAME, NOT BY ID, for the reason `PRICELIST_COLUMNS` gives: an id hard-coded here
-    keeps pointing at whatever inherits it if the tag is deleted and re-made, and every product
-    would be silently mislabelled. A missing tag yields `None`, which marks the whole catalogue
-    `No` — so `validate()` reconciles the count against Odoo rather than trusting this, and goes
-    RED on the difference instead of shipping a column that quietly says nobody is discontinued.
-    """
-    rows = O.search_read('product.tag', [('name', '=ilike', DISCONTINUED_TAG)], ['id'], limit=2)
-    return rows[0]['id'] if rows else None
-
-
-def incoming_from_po(prods=None):
-    """`({code: open_po_units}, report)` — inbound derived from PURCHASE ORDER LINES.
-
-    ⛔ THIS IS THE ORACLE, NOT THE SOURCE. The shipped `incoming` column comes from
-    `product.product.incoming_qty` (see `catalogue()`); this derives the same quantity a second,
-    independent way so `validate()` can hold them against each other rather than reconciling a
-    number with itself — the self-sealing failure `product_data.validate` already carries a scar
-    for ([[gate-and-nc-must-not-share-a-binding]]).
-
-    ⛔⛔ AND THE TWO ARE NOT EXPECTED TO BE EQUAL, WHICH IS WHY THE SOURCE IS THE ONE IT IS.
-    MEASURED 2026-08-19: totals 179,979 vs 173,775 units, differing on **107 of ~490 codes**, and
-    the differences are a UNIT OF MEASURE gap, not an error. `purchase.order.line.product_qty` is
-    in the line's PURCHASE uom while `incoming_qty` is in the product's STOCK uom, so the `SP-*`
-    family reads exactly 4x apart (400 vs 100, 320 vs 80, 240 vs 60). `on_hand` and `qty_ltm` are
-    both in the STOCK uom, so `incoming_qty` is the only one of the two that can be ADDED to a
-    shelf quantity. A buy list built on the purchase uom would over-count inbound fourfold on
-    those SKUs and tell the team not to re-order stock that is not coming.
-
-    So the check this feeds asserts SHAPE and DIRECTION (same SKUs carry inbound, totals within a
-    stated band) and REPORTS the per-code gap. Asserting equality would go red forever on a
-    difference that is correct.
-    """
-    lines = O.search_read('purchase.order.line', [('state', '=', 'purchase')],
-                          ['product_id', 'product_qty', 'qty_received'], limit=50000)
-    n = O.get_odoo().search_count('purchase.order.line', [('state', '=', 'purchase')])
-    by_id = {p['id']: p for p in (prods if prods is not None else [])}
-    if not by_id:
-        for p in O.search_read('product.product', [('active', '=', True)],
-                               ['id', 'default_code'], limit=50000):
-            by_id[p['id']] = p
-    out, off_catalogue = {}, 0.0
-    for line in lines:
-        pid = (line.get('product_id') or [None])[0]
-        open_q = float(line.get('product_qty') or 0.0) - float(line.get('qty_received') or 0.0)
-        if open_q <= 0:
-            continue
-        rec = by_id.get(pid)
-        if rec is None:                 # ordered against an ARCHIVED product: real, but off-grid
-            off_catalogue += open_q
-            continue
-        code = (str(rec['default_code']).strip() if rec.get('default_code') else f"pid:{pid}")
-        out[code] = out.get(code, 0.0) + open_q
-    return out, {'lines_read': len(lines), 'lines_total': n,
-                 'truncated': len(lines) != n,
-                 'units_on_archived_products': round(off_catalogue, 1)}
-
-
-#: ⭐⭐ WAVE 30 / W30-T34 (the carried W29-T52) — THE PRICELIST STRATUM, PER LIST.
-#:
-#: MEASURED LIVE 2026-08-12, and every one of these numbers shaped the design rather than
-#: decorating it:
-#:
-#:   · **Five pricelists, three of them material.** Date-valid fixed rules: Fisch 5,139,
-#:     Royal 1 2,702, Royal 2 2,605, Public Pricelist 17, Giftware Deals 1. At SKU level that
-#:     is Fisch 3,971 · Royal 1 1,920 · Royal 2 1,857 · Public 6 · Giftware 0.
-#:   · **3,988 of 5,871 active products carry a usable fixed rule; 1,883 carry NONE.**
-#:   · **1,853 SKUs are priced on all three lists AND THE LISTS DISAGREE** — median relative
-#:     spread 16%, max 82%. That is why this is three columns rather than one "the price" or a
-#:     low/high pair: neither can tell a Fisch rep what Fisch sells the SKU for, which is the
-#:     only question the column exists to answer.
-#:   · ⛔ **`list_price` is 1.00 on 5,817 of the 5,871** — so the single `3_global` rule
-#:     computed over it is meaningless, exactly as W29-T52 said. **This reader never falls
-#:     through to it**, and that is the ticket's own negative control: a SKU with no specific
-#:     item renders BLANK, never a fallback dressed as a price.
-#:
-#: ⚠ RESOLVED BY NAME, NOT BY ID. A pricelist renamed or deleted in Odoo makes its column go
-#: blank and makes `report["lists_missing"]` name it — a hard-coded id would keep pointing at
-#: whatever inherited it and mislabel every cell silently.
-PRICELIST_COLUMNS = (
-    ("price_fisch", "Fisch"),
-    ("price_royal_1", "Royal 1"),
-    ("price_royal_2", "Royal 2"),
-)
-
-
-def pricelist_by_code():
-    """`({code: {column_key: price}}, report)` — the date-valid FIXED base-tier price for each
-    declared pricelist, keyed EXACTLY as `catalogue()` keys its rows.
-
-    ⛔ THE KEYING IS LOAD-BEARING, not a detail. `catalogue()` keys on `default_code` with a
-    `pid:{id}` fallback for the ~33 active records that carry none. Keying this map any other
-    way would leave those SKUs permanently blank while an oracle counting active products
-    counted them — a red gate on a working build, or worse, a silent hole nobody counts.
-
-    **The base tier, deliberately.** `pricecomp._tier_for` picks the highest `min_quantity` at
-    or below an order's quantity, because it is pricing a LINE that has one. A catalogue column
-    has no quantity in hand, so it takes the LOWEST `min_quantity` — the price at qty 1. Rules
-    above that break are a bulk price, and `report["qty_break_only"]` counts the SKUs whose only
-    rule sits on one (MEASURED: 10 rules of 10,464 carry a break at all).
-
-    **Variant rules beat template rules**, matching `pricecomp._tier_for`: a `0_product_variant`
-    rule is the more specific statement about this exact SKU.
-
-    ⚠ DEGRADES TO `({}, report)` on a read failure, matching `_inventory_by_code` rather than
-    `catalogue()`: these are COLUMNS, and a product grid that will not render because pricing is
-    momentarily unreachable is a worse failure than one with blank price columns. The blanks are
-    not silent — `product_data.validate()`'s coverage leg reconciles each column against a fresh
-    Odoo count and goes red at zero.
-
-    ⭐ **The report exists because R6's second sentence is law** (*"if it can't be done, you need
-    to explicitly tell me why and recommend a fix"*). Everything this reader CANNOT see is
-    counted rather than dropped: rules on pricelists the contract does not declare, non-fixed
-    (`formula`/`percent`) rules, the `3_global` fallback, and prices that only exist above a
-    quantity break.
-    """
-    # ⚠ THE COUNTERS ARE DISJOINT AND ORDER-DEPENDENT, and saying so is the difference between a
-    # report and a misleading one. A rule is classified ONCE, by the first reason it is skipped:
-    # undeclared list → out of date → not fixed → global → zero price. So `rules_not_fixed: 0`
-    # means "no formula rule on a list we declare", NOT "this Odoo has no formula rules"
-    # (MEASURED 2026-08-12: it has exactly one, and it sits on Public Pricelist, which the
-    # contract does not declare — so it lands in `rules_undeclared_list`).
-    report = {"lists_missing": [], "rules_total": 0, "rules_undeclared_list": 0,
-              "rules_not_fixed": 0, "rules_global": 0, "rules_out_of_date": 0,
-              "rules_zero_price": 0, "qty_break_only": 0}
-    try:
-        pl_rows = O.search_read('product.pricelist', [], ['id', 'name'])
-        by_name = {}
-        for p in pl_rows:
-            by_name.setdefault(str(p.get('name') or '').strip(), p['id'])
-        wanted = {}
-        for col, name in PRICELIST_COLUMNS:
-            pid = by_name.get(name)
-            if pid is None:
-                report["lists_missing"].append(name)
-            else:
-                wanted[pid] = col
-
-        # ⭐⭐ FILTERED SERVER-SIDE, AND THE REASON IS A MEASUREMENT, NOT A STYLE PREFERENCE.
-        # Reading all 10,464 rules with 9 fields and sorting them in Python costs **30.1s** on
-        # this connection; the same rules under a server-side domain with 6 fields cost
-        # **13.6s** (measured 2026-08-12, back to back). `pool()` runs this on a scope's
-        # first-ever build, so that 16.5s is 16.5s of somebody's page load.
-        #
-        # ⚠ The domain reproduces the Python predicate EXACTLY, and the null legs are the part
-        # that is easy to get wrong: an absent `date_start` is `False`, not a past date, so
-        # `('date_start','<=',today)` ALONE would drop every open-ended rule — which is almost
-        # all of them.
-        today = P.today().isoformat()
-        _declared = sorted(wanted)
-        _kinds = ['0_product_variant', '1_product']
-        _live = [('pricelist_id', 'in', _declared), ('compute_price', '=', 'fixed'),
-                 ('applied_on', 'in', _kinds),
-                 '|', ('date_start', '=', False), ('date_start', '<=', today),
-                 '|', ('date_end', '=', False), ('date_end', '>=', today),
-                 ('fixed_price', '>', 0)]
-        rules = O.search_read(
-            'product.pricelist.item', _live,
-            ['pricelist_id', 'product_id', 'product_tmpl_id', 'applied_on', 'fixed_price',
-             'min_quantity']) if _declared else []
-
-        # ⭐ R6's SECOND SENTENCE, PAID FOR WITH `search_count` RATHER THAN A WIDER READ. Six
-        # counts cost ~1.2s together; the rules they count would cost 16s to read. What this
-        # reader cannot see is still REPORTED — it is just no longer transferred.
-        def _n(extra):
-            try:
-                return O.get_odoo().search_count('product.pricelist.item', extra)
-            except Exception:
-                return -1                      # -1 reads as "not measured", never as zero
-        _dated = ['|', ('date_start', '=', False), ('date_start', '<=', today),
-                  '|', ('date_end', '=', False), ('date_end', '>=', today)]
-        report["rules_total"] = _n([])
-        report["rules_undeclared_list"] = _n([('pricelist_id', 'not in', _declared)]) \
-            if _declared else report["rules_total"]
-        if _declared:
-            _on = [('pricelist_id', 'in', _declared)]
-            # `percent_price` and formula rules are read NOWHERE in this repo. A rule we cannot
-            # price is one the operator is TOLD about, never one that quietly becomes a blank.
-            report["rules_not_fixed"] = _n(_on + [('compute_price', '!=', 'fixed')])
-            report["rules_global"] = _n(_on + [('compute_price', '=', 'fixed'),
-                                               ('applied_on', 'not in', _kinds)])
-            report["rules_out_of_date"] = (
-                _n(_on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)])
-                - _n(_on + [('compute_price', '=', 'fixed'),
-                            ('applied_on', 'in', _kinds)] + _dated))
-            # A zero price is not "free" — it is an unset rule. Blank says so; 0 does not.
-            report["rules_zero_price"] = _n(
-                _on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)]
-                + _dated + [('fixed_price', '<=', 0)])
-
-        by_var, by_tmpl = {}, {}
-        for r in rules:
-            col = wanted[O.m2o_id(r.get('pricelist_id'))]
-            if r.get('applied_on') == '0_product_variant' and r.get('product_id'):
-                by_var.setdefault((col, O.m2o_id(r['product_id'])), []).append(r)
-            elif r.get('product_tmpl_id'):
-                by_tmpl.setdefault((col, O.m2o_id(r['product_tmpl_id'])), []).append(r)
-
-        dom = [('active', '=', True)]
-        prods = O.search_read('product.product', dom,
-                              ['id', 'default_code', 'product_tmpl_id'], limit=50000)
-        n = O.get_odoo().search_count('product.product', dom)
-        if len(prods) != n:
-            # Same guard, same reason as `catalogue()`: a short pull renders as a plausible
-            # smaller set of priced SKUs with nothing reporting it.
-            raise ValueError(
-                f"products.pricelist_by_code: the product pull is TRUNCATED — read {len(prods)} "
-                f"rows against a search_count of {n}.")
-    except Exception as e:
-        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
-        return {}, report
-
-    out = {}
-    for p in prods:
-        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
-        tmpl = O.m2o_id(p.get('product_tmpl_id'))
-        cells = {}
-        for col, _name in PRICELIST_COLUMNS:
-            cands = by_var.get((col, p['id'])) or by_tmpl.get((col, tmpl))
-            if not cands:
-                continue
-            base = min(cands, key=lambda r: r.get('min_quantity') or 0.0)
-            if (base.get('min_quantity') or 0.0) > 1.0:
-                report["qty_break_only"] += 1
-            cells[col] = base.get('fixed_price')
-        if cells:
-            out.setdefault(code, {}).update(cells)
-    return out, report
-
-
-def active_products(limit=50000):
-    """`[{id, default_code, product_tmpl_id}]` for every ACTIVE product — read ONCE and shared.
-
-    ⛔ RAISES ON A SHORT PULL, like `catalogue()` and for the same reason: a truncated product read
-    renders as a plausible smaller set of priced SKUs with nothing reporting it.
-    ⭐ It exists so `tier_prices_by_code` and `packagings_by_code` can be called from ONE pool
-    build without each paying for its own copy of the same 5,873-row read — measured at ~4 s each
-    on this connection, on a path that is somebody's first page load.
-    """
-    dom = [('active', '=', True)]
-    prods = O.search_read('product.product', dom, ['id', 'default_code', 'product_tmpl_id'],
-                          limit=limit)
-    n = O.get_odoo().search_count('product.product', dom)
-    if len(prods) != n:
-        raise ValueError(f"products.active_products: the product pull is TRUNCATED. Read "
-                         f"{len(prods)} rows against a search_count of {n}.")
-    return prods
-
-
-def tier_prices_by_code(prods=None):
-    """`({code: [{pricelist, unit_price}]}, report)` — EVERY live price a SKU really has.
-
-    ⭐⭐ W37-T14 (owner: multiple prices per SKU). `pricelist_by_code` above answers a DIFFERENT
-    question and both are needed: it fills three DECLARED columns (`price_fisch`, `price_royal_1`,
-    `price_royal_2`) and therefore cannot show a price on a list the contract does not name. This
-    one is the honest set — one entry per pricelist that actually prices this SKU today.
-
-    ⛔ IT READS EVERY LIVE PRICELIST, not `PRICELIST_COLUMNS`. Hardcoding this tenant's three list
-    names into "how many prices does a SKU have" is exactly what the ticket forbids, and it is what
-    would make the answer wrong for tenant #1 on the day they are onboarded.
-
-    ⛔ ROWS ARE TEMPLATE-SCOPED (`proto/P2-pricing-uom.md`): joining on `product_id` drops 99.9% of
-    price rows, because `applied_on` is `1_product` on 10,454 of 10,468 items and `1_product` means
-    the TEMPLATE. A variant rule (`0_product_variant`, 13 rows) is the more specific statement and
-    wins, matching `pricecomp._tier_for` and `pricelist_by_code`.
-
-    ⛔ FILTER BY `pricelist_id`, NEVER BY `active`: the default `product.pricelist.item` count hides
-    2,580 archived items, so an `active` filter reads as a smaller, plausible, wrong set.
-
-    ⚠ THE BASE TIER, at qty 1, for the same reason `pricelist_by_code` gives: a catalogue cell has
-    no quantity in hand. Measured: only 10 of 10,468 items carry a quantity break at all, so this
-    is very nearly the whole story rather than a simplification.
-    ⚠ CARDINALITY 1..3 TODAY, mode 3 — and `Royal 2` carries 2,605 price rows against **0 customers
-    and 0 orders**, so a SKU reading "3 tiers" is catalogue-true and commercially misleading. The
-    entry keeps the list NAME so a reader can see which tier it is rather than a bare count.
-    """
-    report = {"lists": [], "rules_total": 0, "rules_not_fixed": 0, "rules_out_of_date": 0,
-              "rules_global": 0, "skus_with_no_price": 0, "identity_breaks": 0}
-    try:
-        pls = {p['id']: str(p.get('name') or '').strip()
-               for p in O.search_read('product.pricelist', [], ['id', 'name'])}
-        report["lists"] = sorted(pls.values())
-        today = P.today().isoformat()
-        dom = [('pricelist_id', 'in', sorted(pls)), ('compute_price', '=', 'fixed'),
-               ('applied_on', 'in', ['0_product_variant', '1_product']),
-               '|', ('date_start', '=', False), ('date_start', '<=', today),
-               '|', ('date_end', '=', False), ('date_end', '>=', today),
-               ('fixed_price', '>', 0)]
-        rules = O.search_read('product.pricelist.item', dom,
-                              ['pricelist_id', 'product_id', 'product_tmpl_id', 'applied_on',
-                               'fixed_price', 'min_quantity'])
-
-        # R6's second sentence: what this reader cannot see is COUNTED, never dropped.
-        def _n(extra):
-            try:
-                return O.get_odoo().search_count('product.pricelist.item', extra)
-            except Exception:                                          # noqa: BLE001
-                return -1                     # -1 reads as "not measured", never as zero
-        report["rules_total"] = _n([])
-        report["rules_not_fixed"] = _n([('compute_price', '!=', 'fixed')])
-        report["rules_global"] = _n([('applied_on', '=', '3_global')])
-        report["rules_out_of_date"] = _n(['|', ('date_end', '!=', False),
-                                          ('date_start', '!=', False)])
-
-        by_var, by_tmpl = {}, {}
-        for r in rules:
-            plid = O.m2o_id(r.get('pricelist_id'))
-            if r.get('applied_on') == '0_product_variant' and r.get('product_id'):
-                by_var.setdefault((plid, O.m2o_id(r['product_id'])), []).append(r)
-            elif r.get('product_tmpl_id'):
-                by_tmpl.setdefault((plid, O.m2o_id(r['product_tmpl_id'])), []).append(r)
-
-        prods = active_products() if prods is None else prods
-    except Exception as e:                                             # noqa: BLE001
-        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
-        return {}, report
-
-    out = {}
-    for p in prods:
-        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
-        tmpl = O.m2o_id(p.get('product_tmpl_id'))
-        tiers = []
-        for plid, name in sorted(pls.items(), key=lambda kv: kv[1].lower()):
-            cands = by_var.get((plid, p['id'])) or by_tmpl.get((plid, tmpl))
-            if not cands:
-                continue
-            base = min(cands, key=lambda r: r.get('min_quantity') or 0.0)
-            tiers.append({"pricelist": name, "unit_price": round(base.get('fixed_price') or 0.0, 2)})
-        if tiers:
-            # ⚠ A code carried by TWO active products (D-309: `2112-12`) MERGES here, because the
-            # grid is keyed by code and one code is one row. The prices are UNIONED rather than
-            # one silently winning — a SKU that really does have two different Fisch prices should
-            # show both, and hiding one is how a $60 price disappeared behind a $24 one.
-            prior = out.get(code)
-            if prior is None:
-                out[code] = tiers
-            else:
-                seen = {(t["pricelist"], t["unit_price"]) for t in prior}
-                for t in tiers:
-                    if (t["pricelist"], t["unit_price"]) not in seen:
-                        prior.append(t)
-                        report["identity_breaks"] += 1
-        else:
-            report["skus_with_no_price"] += 1
-    return out, report
-
-
-def packagings_by_code(prods=None):
-    """`({code: [{name, qty}]}, report)` — the UNITS a SKU is really sold in (W37-T15).
-
-    ⛔ IT COMES FROM `product.packaging` ALONE, and `proto/P2-pricing-uom.md` measured why the
-    obvious alternative is a dead end: the `uom.uom` Config category holds 26 conversion-bearing
-    units (`Case-Packed 12` … `Pallet-Packed 4,800`) referenced by **0 products, 0 sale lines and
-    0 stock moves**, and every unit actually in use has `factor_inv = 1`. Building on `uom.uom`
-    conversions yields a column of 1s.
-
-    ⛔ `qty > 1` IS A FILTER, NOT A TIDY-UP. 6,283 of 7,471 packaging rows are `qty = 1.0` — a
-    packaging that packs one of something is not a unit tier — plus real junk (one-character
-    names). Without it "units per SKU" reads **5,859** SKUs instead of **1,150**.
-
-    ⛔ 133 ROWS ARE ORPHANS (`product_id = False`) and would collapse under a `groupby(product_id)`
-    into ONE fake product carrying 133 packagings. Dropped, and counted.
-
-    ⚠ HONEST EMPTY: only **1,150 of 5,873 SKUs (19.6%)** have any unit tier, so this cell is
-    legitimately blank for 80% of the catalogue — "this SKU is sold in one unit", never "missing".
-    The 19.6% is not decoration: units are transacted on 77.9% of confirmed sale lines.
-    ⚠ `qty` is denominated in the product's OWN `uom_id` (matched on 7,338/7,338).
-    """
-    report = {"rows_total": 0, "rows_orphan": 0, "rows_qty_le_1": 0, "skus_with_units": 0}
-    try:
-        rows = O.search_read('product.packaging', [], ['id', 'name', 'qty', 'product_id'])
-    except Exception as e:                                             # noqa: BLE001
-        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
-        return {}, report
-    report["rows_total"] = len(rows)
-    by_pid = {}
-    for r in rows:
-        pid = O.m2o_id(r.get('product_id'))
-        if not pid:
-            report["rows_orphan"] += 1
-            continue
-        if (r.get('qty') or 0) <= 1:
-            report["rows_qty_le_1"] += 1
-            continue
-        by_pid.setdefault(pid, []).append(
-            {"name": str(r.get('name') or '').strip(), "qty": float(r.get('qty') or 0)})
-    try:
-        prods = active_products() if prods is None else prods
-    except Exception as e:                                             # noqa: BLE001
-        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
-        return {}, report
-    out = {}
-    for p in prods:
-        got = by_pid.get(p['id'])
-        if not got:
-            continue
-        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
-        out.setdefault(code, []).extend(sorted(got, key=lambda u: u["qty"]))
-    report["skus_with_units"] = len(out)
-    return out, report
-
-
-def catalogue_count():
-    """The INDEPENDENT population oracle: Odoo's own count of active products.
-
-    Deliberately a bare `search_count` and not a `len()` over anything this module built — the
-    2,717 defect shipped silently for a wave because `product_data.validate()` derived BOTH sides
-    of its reconciliation from `_sku_rev`, so the oracle could never see a missing row.
-    """
-    return O.get_odoo().search_count('product.product', [('active', '=', True)])
-
-
-def categories(t=None, team_id=None):
-    """Sorted distinct category names present in the SKU directory (for the drill-down filter)."""
-    return sorted({r['category'] for r in directory(t, team_id)})
-
-
-def _coverage(date_from, date_to, team_id=None):
-    """{sku_code: distinct_customer_count} via 2-level read_group (slow-ish), merged by code.
-    (A customer buying two records sharing a code can count twice; duplicates are rare, and
-    code-level avoids the bigger error of a re-SKUed product reading as full coverage loss.)"""
-    g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC),
-                     ['__count'], ['product_id', 'order_partner_id'], lazy=False)
-    codes = _code_map()
-    cov = {}
-    names = {}
-    for r in g:
-        pid = O.m2o_id(r.get('product_id'))
-        if not pid:
-            continue
-        key = codes.get(pid, f"pid:{pid}")
-        cov[key] = cov.get(key, 0) + 1
-        names.setdefault(key, O.m2o_name(r.get('product_id')))
-    return cov, names
-
-
-def coverage_collapse(t=None, limit=25, min_prior_custs=8, team_id=None):
-    """SKUs that lost the most customer breadth YoY — early warning a SKU is dying even if
-    revenue hasn't fully cratered yet."""
-    t = t or P.today()
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    cov_t, names_t = _coverage(yf, yt, team_id)
-    cov_l, names_l = _coverage(lf, lt, team_id)
-    rows = []
-    for pid, lc in cov_l.items():
-        if lc < min_prior_custs:
-            continue
-        tc = cov_t.get(pid, 0)
-        drop = lc - tc
-        if drop <= 0:
-            continue
-        rows.append({'code': pid, 'product': names_l.get(pid) or names_t.get(pid) or '',
-                     'custs_ly': lc, 'custs_ytd': tc, 'lost_custs': drop,
-                     'pct_drop': drop / lc * 100})
-    rows.sort(key=lambda x: (-x['lost_custs'], -x['pct_drop']))
-    return rows[:limit]
-
-
-# ====================================================================== SKU DRAWER (mirror of customer drawer)
-def _sku_product_ids(code):
-    """All product.product ids sharing this SKU code (merged variants / archived records)."""
-    return [pid for pid, c in _code_map().items() if c == code]
-
-
-def _sku_dom(pids, date_from, date_to, team_id=None):
-    return O.sale_line_domain(date_from, date_to, team_id, extra=[('product_id', 'in', pids)] + _NO_SVC)
-
-
-def _sku_name_category(pids):
-    rows = O.search_read('product.product', [('id', 'in', pids)], ['name', 'categ_id'])
-    name = rows[0]['name'] if rows else '(unknown)'
-    catmap = sales_mod._product_cat()
-    main = next((catmap.get(p) for p in pids if catmap.get(p)), '(uncategorized)')
-    return name, main
-
-
-def _sku_buyers(pids, date_from, date_to, team_id=None):
-    """{partner_id: {'name','rev','qty'}} for buyers of this SKU over a window."""
-    g = O.read_group('sale.order.line', _sku_dom(pids, date_from, date_to, team_id),
-                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False)
-    out = {}
-    for r in g:
-        pid = O.m2o_id(r.get('order_partner_id'))
-        if pid:
-            out[pid] = {'name': O.m2o_name(r.get('order_partner_id')),
-                        'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0}
-    return out
-
-
-def sku_detail(code, t=None, team_id=None, n_months=13, allsku=None):
-    """KPIs (rev/qty/buyers YoY, GM%), monthly trend, rank & % of BU for one SKU code.
-    Pass `allsku` (a cached _sku_rev YTD map) to skip the ~all-SKU rank read."""
-    t = t or P.today()
-    pids = _sku_product_ids(code)
-    if not pids:
-        return None
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    mf, mt = P.ltm(t)
-    name, category = _sku_name_category(pids)
-
-    def s(df, dtt, field='price_subtotal'):
-        return O.sum_field('sale.order.line', _sku_dom(pids, df, dtt, team_id), field)
-    rev_ytd, rev_ly = s(yf, yt), s(lf, lt)
-    qty_ytd, qty_ly = s(yf, yt, 'product_uom_qty'), s(lf, lt, 'product_uom_qty')
-    buyers_ytd = len(_sku_buyers(pids, yf, yt, team_id))
-    buyers_ly = len(_sku_buyers(pids, lf, lt, team_id))
-    g = O.read_group('sale.order.line', _sku_dom(pids, mf, mt, team_id),
-                     ['price_subtotal:sum', 'margin:sum'], [], lazy=False)
-    line_rev = (g[0].get('price_subtotal') if g else 0) or 0.0
-    margin = (g[0].get('margin') if g else 0) or 0.0
-
-    # Monthly trend: bin lines by their order's month in Python (dot-path month groupby is rejected
-    # on sale.order.line), in 2 reads instead of 26 point queries.
-    range_start = dt.date(t.year - 2, t.month, 1).isoformat()
-    lines = O.search_read('sale.order.line', _sku_dom(pids, range_start, t.isoformat(), team_id),
-                          ['price_subtotal', 'order_id'])
-    oids = list({O.m2o_id(line['order_id']) for line in lines if line.get('order_id')})
-    odate = {}
-    for i in range(0, len(oids), 1000):
-        for o in O.search_read('sale.order', [('id', 'in', oids[i:i + 1000])], ['date_order']):
-            if o.get('date_order'):
-                odate[o['id']] = str(o['date_order'])[:7]
-    mrev = {}
-    for line in lines:
-        ym = odate.get(O.m2o_id(line.get('order_id')))
-        if ym:
-            mrev[ym] = mrev.get(ym, 0.0) + (line.get('price_subtotal') or 0.0)
-    monthly = []
-    for ym, start, end in P.month_starts(n_months, t):
-        y, m = int(ym[:4]) - 1, int(ym[5:7])
-        monthly.append({'month': ym, 'revenue': mrev.get(ym, 0.0),
-                        'revenue_ly': mrev.get(f'{y:04d}-{m:02d}', 0.0)})
-
-    allsku = allsku if allsku is not None else _sku_rev(yf, yt, team_id)
-    total = sum(v['rev'] for v in allsku.values()) or 1.0
-    rank = next((i + 1 for i, (c, _v) in enumerate(sorted(allsku.items(), key=lambda kv: -kv[1]['rev']))
-                 if c == code), None)
-    return {
-        'code': code, 'name': name, 'category': category,
-        'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'rev_yoy_pct': P.yoy_pct(rev_ytd, rev_ly),
-        'qty_ytd': qty_ytd, 'qty_ly': qty_ly, 'qty_yoy_pct': P.yoy_pct(qty_ytd, qty_ly),
-        'buyers_ytd': buyers_ytd, 'buyers_ly': buyers_ly, 'buyers_delta': buyers_ytd - buyers_ly,
-        'gm_pct': (margin / line_rev * 100) if line_rev else 0.0, 'gm_dollars': margin,
-        'rank': rank, 'n_skus': len(allsku), 'pct_of_bu': rev_ytd / total * 100,
-        'monthly': monthly,
-    }
-
-
-def sku_buyer_bridge(code, t=None, team_id=None, top=12):
-    """Who drives the SKU's YoY: retained / new / churned buyers + a $-ranked churned call list."""
-    t = t or P.today()
-    pids = _sku_product_ids(code)
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id)
-    tset, lset = set(this), set(last)
-    retained, new, churned = tset & lset, tset - lset, lset - tset
-    churned_list = sorted([{'pid': p, 'customer': last[p]['name'], 'ly_rev': last[p]['rev']}
-                           for p in churned], key=lambda x: -x['ly_rev'])[:top]
-    return {'retained': {'n': len(retained), 'rev': sum(this[p]['rev'] for p in retained)},
-            'new': {'n': len(new), 'rev': sum(this[p]['rev'] for p in new)},
-            'churned': {'n': len(churned), 'rev': sum(last[p]['rev'] for p in churned)},
-            'buyer_retention_pct': (len(retained) / len(lset) * 100) if lset else 0.0,
-            'churned_buyers': churned_list, 'buyers_this': len(tset), 'buyers_last': len(lset),
-            'this_total': sum(v['rev'] for v in this.values())}
-
-
-def sku_concentration(code, t=None, team_id=None):
-    """Buyer-concentration risk: top-1/top-3 share, Herfindahl index, effective buyer count (LTM)."""
-    t = t or P.today()
-    mf, mt = P.ltm(t)
-    buyers = _sku_buyers(_sku_product_ids(code), mf, mt, team_id)
-    revs = sorted([v['rev'] for v in buyers.values()], reverse=True)
-    total = sum(revs) or 1.0
-    hhi = sum((r / total) ** 2 for r in revs)
-    return {'n_buyers': len(revs), 'top1_pct': (revs[0] / total * 100) if revs else 0.0,
-            'top3_pct': (sum(revs[:3]) / total * 100) if revs else 0.0,
-            'hhi': hhi, 'eff_buyers': (1 / hhi) if hhi else 0.0}
-
-
-def sku_price_dispersion(code, t=None, team_id=None, cap=60):
-    """Realized $/unit per buyer (LTM) vs the volume-weighted average; recoverable $ on below-VWAP
-    accounts. list_price is unreliable, so the dispersion among actual buyers is the margin lever."""
-    t = t or P.today()
-    mf, mt = P.ltm(t)
-    g = O.read_group('sale.order.line', _sku_dom(_sku_product_ids(code), mf, mt, team_id),
-                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False)
-    rows, tot_rev, tot_qty = [], 0.0, 0.0
-    for r in g:
-        pid = O.m2o_id(r.get('order_partner_id'))
-        rev = r.get('price_subtotal') or 0.0
-        qty = r.get('product_uom_qty') or 0.0
-        if not pid or qty <= 0:
-            continue
-        rows.append({'pid': pid, 'customer': O.m2o_name(r.get('order_partner_id')),
-                     'price': rev / qty, 'qty': qty, 'rev': rev})
-        tot_rev += rev
-        tot_qty += qty
-    vwap = (tot_rev / tot_qty) if tot_qty else 0.0
-    for r in rows:
-        r['recoverable'] = max(0.0, vwap - r['price']) * r['qty']
-    rows.sort(key=lambda x: -x['recoverable'])
-    return {'vwap': vwap, 'n_buyers': len(rows),
-            'recoverable_total': sum(r['recoverable'] for r in rows), 'rows': rows[:cap]}
-
-
-def sku_top_buyers(code, t=None, team_id=None, top=15):
-    """Ranked buyers of this SKU (YTD) with YoY — clickable to open the customer drawer."""
-    t = t or P.today()
-    pids = _sku_product_ids(code)
-    yf, yt = P.ytd(t)
-    lf, lt = P.ytd_last_year(t)
-    this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id)
-    rows = [{'pid': p, 'customer': v['name'], 'rev': v['rev'], 'qty': v['qty'],
-             'yoy_pct': P.yoy_pct(v['rev'], last.get(p, {}).get('rev', 0.0))} for p, v in this.items()]
-    rows.sort(key=lambda x: -x['rev'])
-    return rows[:top]
-
-
-_BUYER_TIERS = [('Whale (≥$25k)', 25000.0), ('Large ($10–25k)', 10000.0),
-                ('Mid ($2–10k)', 2000.0), ('Small (<$2k)', 0.0)]
-
-
-def sku_customer_analysis(code, t=None, team_id=None, top=15):
-    """WHO buys this SKU (LTM), as customers: the value-tier mix of its buyers (by each buyer's TOTAL
-    spend), and the accounts most DEPENDENT on it (this SKU as a share of their spend — who gets hurt
-    most if it stocks out / who to protect)."""
-    t = t or P.today()
-    lf, lt = P.ltm(t)
-    pids = _sku_product_ids(code)
-    empty = {'segments': [], 'dependency': [], 'n_buyers': 0, 'avg_dependency': 0.0}
-    if not pids:
-        return empty
-    buyers = _sku_buyers(pids, lf, lt, team_id)        # {pid: {name, rev(this SKU), qty}}
-    if not buyers:
-        return empty
-    bpids = list(buyers)
-    g = O.read_group('sale.order', sales_mod.order_domain(lf, lt, team_id) + [('partner_id', 'in', bpids)],
-                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
-    total = {O.m2o_id(r['partner_id']): (r.get('amount_untaxed') or 0.0) for r in g if r.get('partner_id')}
-
-    def tier(rev):
-        for nm, lo in _BUYER_TIERS:
-            if rev >= lo:
-                return nm
-        return _BUYER_TIERS[-1][0]
-    seg = {nm: {'tier': nm, 'buyers': 0, 'sku_rev': 0.0} for nm, _ in _BUYER_TIERS}
-    dep = []
-    for p, v in buyers.items():
-        ct = total.get(p, v['rev']) or v['rev']
-        s = seg[tier(ct)]
-        s['buyers'] += 1
-        s['sku_rev'] += v['rev']
-        dep.append({'customer': v['name'], 'pid': p, 'sku_rev': v['rev'], 'cust_total': ct,
-                    'dependency_pct': (v['rev'] / ct * 100) if ct else None})
-    tot = sum(s['sku_rev'] for s in seg.values()) or 1.0
-    segments = []
-    for nm, _ in _BUYER_TIERS:
-        s = seg[nm]
-        s['rev_share'] = s['sku_rev'] / tot * 100
-        s['avg_per_buyer'] = (s['sku_rev'] / s['buyers']) if s['buyers'] else 0.0
-        segments.append(s)
-    dep.sort(key=lambda x: -x['sku_rev'])
-    deps_known = [d['dependency_pct'] for d in dep if d['dependency_pct'] is not None]
-    return {'segments': segments, 'dependency': dep[:top], 'n_buyers': len(buyers),
-            'avg_dependency': (sum(deps_known) / len(deps_known)) if deps_known else 0.0}
-
-
-def sku_whitespace(code, t=None, team_id=None, top=15):
-    """Customers who buy this SKU's category but NOT this SKU — ranked prospect list."""
-    import modules.customers as cust_mod
-    t = t or P.today()
-    pids = _sku_product_ids(code)
-    _name, category = _sku_name_category(pids)
-    cat_buyers = cust_mod.category_buyers(category, team_id=team_id) or set()
-    mf, mt = P.ltm(t)
-    prospects = list(cat_buyers - set(_sku_buyers(pids, mf, mt, team_id)))
-    if not prospects:
-        return {'category': category, 'rows': []}
-    g = O.read_group('sale.order', sales_mod.order_domain(mf, mt, team_id) + [('partner_id', 'in', prospects)],
-                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
-    rows = [{'pid': O.m2o_id(r['partner_id']), 'customer': O.m2o_name(r['partner_id']),
-             'total_spend': r.get('amount_untaxed') or 0.0} for r in g if r.get('partner_id')]
-    rows.sort(key=lambda x: -x['total_spend'])
-    return {'category': category, 'rows': rows[:top]}
-
-
-def sku_drawer_bundle(code, t=None, team_id=None, allsku=None):
-    """The whole SKU drawer's first paint in one cached unit: sku_detail first (the not-found gate),
-    then the other six pulls CONCURRENTLY (O.parallel). Cuts a ~6-call cold open to ~max(call)."""
-    detail = sku_detail(code, t=t, team_id=team_id, allsku=allsku)
-    if detail is None:
-        return {'detail': None}
-    bridge, conc, price, buyers, white, ca = O.parallel([
-        lambda: sku_buyer_bridge(code, t, team_id),
-        lambda: sku_concentration(code, t, team_id),
-        lambda: sku_price_dispersion(code, t, team_id),
-        lambda: sku_top_buyers(code, t, team_id),
-        lambda: sku_whitespace(code, t, team_id),
-        lambda: sku_customer_analysis(code, t, team_id),
-    ])
-    return {'detail': detail, 'bridge': bridge, 'conc': conc, 'price': price,
-            'buyers': buyers, 'white': white, 'ca': ca}
-
-
-def validate(t=None, team_id=None):
-    """Reconcile SKU metrics to independent Odoo aggregates. When team_id is set (a single BU
-    selected) every check runs SCOPED to that BU, so the validation panel never reconciles
-    against — or exposes — the other BU's numbers. All checks here are BU-scopeable (no cross-BU
-    mirror), so team_id threads straight through."""
-    t = t or P.today()
-    yf, yt = P.ytd(t)
-    checks = []
-    sku = _sku_rev(yf, yt, team_id)
-    sku_sum = sum(v['rev'] for v in sku.values())
-    line_total = O.sum_field('sale.order.line',
-                             O.sale_line_domain(yf, yt, team_id, extra=_NO_SVC), 'price_subtotal')
-    checks.append({'check': 'SKU rev: Σ(per-SKU) == total line revenue, ex-services (YTD)',
-                   'a': round(sku_sum, 2), 'b': round(line_total, 2),
-                   'gap': round(sku_sum - line_total, 2),
-                   'ok': abs(sku_sum - line_total) <= 1.0})
-
-    m = yoy_movers(t, limit=10**9, team_id=team_id)
-    movers_sum = sum(r['change'] for r in m['risers']) + sum(r['change'] for r in m['decliners'])
-    lf, lt = P.ytd_last_year(t)
-    last_total = O.sum_field('sale.order.line',
-                             O.sale_line_domain(lf, lt, team_id, extra=_NO_SVC), 'price_subtotal')
-    checks.append({'check': 'SKU movers: Σ(Δ) == (YTD − LY) total',
-                   'a': round(movers_sum, 2), 'b': round(line_total - last_total, 2),
-                   'gap': round(movers_sum - (line_total - last_total), 2),
-                   'ok': abs(movers_sum - (line_total - last_total)) <= 1.0})
-
-    # SKU drawer: buyer-bridge reconciles — retained + new buyer revenue == this-period SKU revenue
-    top_code = max(sku.items(), key=lambda kv: kv[1]['rev'])[0] if sku else None
-    if top_code:
-        bb = sku_buyer_bridge(top_code, t, team_id=team_id)
-        recon = bb['retained']['rev'] + bb['new']['rev']
-        checks.append({'check': 'SKU drawer: retained+new buyer rev == SKU YTD revenue',
-                       'a': round(recon, 2), 'b': round(bb['this_total'], 2),
-                       'gap': round(recon - bb['this_total'], 2),
-                       'ok': abs(recon - bb['this_total']) <= 1.0})
-    return checks
+"""Products / SKU module — SKU health: YoY movers (risers & decliners), zombie SKUs
+(catalog rot — sellable, formerly selling, now dead), new winners, coverage collapse
+(SKUs losing customer breadth — the FFS-recovery early-warning signal), and velocity leaders.
+
+Per-SKU margin already lives in the Financial module (low_margin_skus); not duplicated here.
+Basket / co-purchase (MBA) is intentionally deferred — the existing client app computes it
+runtime-side, and a local co-occurrence pull over 74k LTM lines is the kind of heavy job the
+project guardrails keep off the local PC. (See BACKLOG.)
+
+All line-level (sale.order.line), RI+FFS scope, excluded accounts removed — reusing sale_line_domain.
+Coverage (distinct customers per SKU) uses a 2-level read_group and is a touch slow (~15s/
+window), so it's its own function the app calls lazily and caches.
+"""
+import sys
+import datetime as dt
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from functools import lru_cache
+import core.odoo as O
+import core.periods as P
+import modules.sales as sales_mod
+
+# SKU-health views are about real products — exclude service/delivery pseudo-SKUs
+# (Delivery Charges otherwise dominate movers/winners). Dot-path FILTER works (groupby
+# on the dot-path does not). Validation reconciles on this same non-service universe.
+_NO_SVC = [('product_id.type', '!=', 'service')]
+
+
+@lru_cache(maxsize=1)
+def _code_map():
+    """product_id → SKU code. Multiple product records can share a default_code (re-SKUing
+    /duplicates); keying by code merges them so a re-coded item doesn't read as a fake
+    decliner + fake riser. Products without a code fall back to a per-id key. Archived
+    products are INCLUDED — re-SKUing typically archives the old record and creates a new
+    one under the same code, and the old record still carries last-year sales."""
+    prods = O.search_read('product.product', [('active', 'in', [True, False])], ['id', 'default_code'],
+                          limit=50000)
+    return {p['id']: (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
+            for p in prods}
+
+
+def _sku_rev(date_from, date_to, team_id=None, all_channels=False):
+    """{sku_code: {'name','rev','qty','orders'}} over a window (services excluded,
+    duplicate product records merged by SKU code).
+
+    ⚠ `all_channels=True` keeps the excluded house accounts (GIFTWARE DEALS, the Amazon channel)
+    IN the read. Only the replenishment caller wants that; see `sale_line_domain`'s own note."""
+    g = O.read_group('sale.order.line',
+                     O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC,
+                                        all_channels=all_channels),
+                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False)
+    codes = _code_map()
+    out = {}
+    for r in g:
+        pid = O.m2o_id(r.get('product_id'))
+        if not pid:
+            continue
+        key = codes.get(pid, f"pid:{pid}")
+        e = out.setdefault(key, {'name': O.m2o_name(r.get('product_id')),
+                                 'rev': 0.0, 'qty': 0.0, 'orders': 0})
+        e['rev'] += r.get('price_subtotal') or 0.0
+        e['qty'] += r.get('product_uom_qty') or 0.0
+        e['orders'] += r.get('__count') or 0
+    return out
+
+
+#: ⭐⭐ THE BUY LIST'S DEMAND HORIZON (owner ruling, 2026-08-19). The owner described the cover gap
+#: as *"based on the next 8 months Unit sales"*; it was in fact a trailing-twelve-month daily rate,
+#: and asked to choose, the owner ruled FORWARD 8 MONTHS. This is that horizon, in days.
+FORWARD_MONTHS = 8
+FORWARD_DAYS = 243                        # 8 months of 30.4 days, rounded to whole days
+
+
+def forward_demand_by_code(t=None, team_id=None):
+    """`({code: units}, report)` — units this SKU is expected to sell over the NEXT 8 months.
+
+    ⭐⭐ SEASONAL NAIVE, AND FOR THIS TENANT THAT IS THE POINT RATHER THAN A SHORTCUT. The estimate
+    for the next 8 months is **the units the SKU actually sold in the same 8 calendar months one
+    year ago**. Royal Imports is a floral-supply wholesaler whose year is Valentine's, Mother's Day
+    and Christmas; a flat trailing-twelve-month rate spreads those peaks evenly across the year and
+    tells a buyer to hold the same stock in January as in early February. Reading the matching
+    window last year keeps the shape of the year in the number.
+
+    ⛔ THE FALLBACK IS NOT OPTIONAL AND IT IS NOT SILENT. A SKU introduced since that window has no
+    history in it, and a pure seasonal read would estimate ZERO demand and quietly guarantee the
+    newest products are never re-ordered — a buy list's worst possible blind spot, because nothing
+    on screen would look wrong. So a SKU with no reference-window sales but real LTM sales falls
+    back to its LTM rate scaled to the horizon, and `report` COUNTS how many rows took each road so
+    the mix is visible rather than assumed.
+
+    ⚠ BU-SHAPED THROUGH `team_id`, exactly as `directory()` is: a Fisch reader's buy quantity must
+    answer "what will FISCH sell", and the window reads through the same `sale_line_domain`.
+    """
+    t = t or P.today()
+    # ⛔ THE WINDOW IS THE NEXT 8 MONTHS SHIFTED BACK A YEAR, NOT THE 8 MONTHS BEFORE THIS DATE
+    # LAST YEAR, and the two are easy to swap. Today is the ANCHOR: `t - 365d` is where the
+    # forecast period starts and the window runs FORWARD from there. Written the other way round
+    # (`ref_to = t - 365d`, counting backwards) the estimate reads the wrong half of the year and
+    # is at its most wrong exactly where seasonality is strongest — it would price the next
+    # Christmas off last spring. Caught on the first run of this function, 2026-08-19.
+    ref_from = t - dt.timedelta(days=365)
+    ref_to = ref_from + dt.timedelta(days=FORWARD_DAYS)
+    # ⭐⭐ ALL CHANNELS, INCLUDING AMAZON (owner, 2026-08-19). Those units ship off the same shelf,
+    # so a reorder quantity that omits them under-buys every SKU the Amazon channel moves. This is
+    # the ONE demand read in this module that opts in; revenue and margin stay wholesale-scoped.
+    seasonal = _sku_rev(ref_from.isoformat(), ref_to.isoformat(), team_id, all_channels=True)
+    lf, lt = P.ltm(t)
+    ltm = _sku_rev(lf, lt, team_id, all_channels=True)
+
+    out, from_seasonal, from_ltm = {}, 0, 0
+    for code, e in seasonal.items():
+        qty = float(e.get('qty') or 0.0)
+        if qty > 0:
+            out[code] = qty
+            from_seasonal += 1
+    for code, e in ltm.items():
+        if code in out:
+            continue
+        qty = float(e.get('qty') or 0.0)
+        if qty > 0:
+            # Scale the trailing rate onto the horizon. Flat by construction: there is no
+            # seasonal shape to borrow for a SKU that was not there last year.
+            out[code] = qty * (FORWARD_DAYS / 365.0)
+            from_ltm += 1
+    return out, {
+        'reference_window': [ref_from.isoformat(), ref_to.isoformat()],
+        'horizon_days': FORWARD_DAYS,
+        'codes_from_seasonal': from_seasonal,
+        'codes_from_ltm_fallback': from_ltm,
+        'codes_with_no_demand': 0,     # filled by the caller against the catalogue it holds
+    }
+
+
+def yoy_movers(t=None, limit=20, team_id=None):
+    """Top SKU risers and decliners by YTD-vs-same-period-LY revenue change."""
+    t = t or P.today()
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    this = _sku_rev(yf, yt, team_id)
+    last = _sku_rev(lf, lt, team_id)
+    rows = []
+    for pid in set(this) | set(last):
+        tr = this.get(pid, {'rev': 0.0, 'name': last.get(pid, {}).get('name', '')})
+        lr = last.get(pid, {'rev': 0.0})
+        name = this.get(pid, {}).get('name') or last.get(pid, {}).get('name') or ''
+        rows.append({'code': pid, 'product': name, 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'],
+                     'change': tr['rev'] - lr['rev']})
+    risers = sorted([r for r in rows if r['change'] > 0], key=lambda x: -x['change'])[:limit]
+    decliners = sorted([r for r in rows if r['change'] < 0], key=lambda x: x['change'])[:limit]
+    return {'risers': risers, 'decliners': decliners}
+
+
+def zombie_skus(t=None, limit=30, min_prior=1500.0, team_id=None):
+    """Catalog rot: SKUs that sold materially last year but are ~dead this year."""
+    t = t or P.today()
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    this = _sku_rev(yf, yt, team_id)
+    last = _sku_rev(lf, lt, team_id)
+    rows = []
+    for pid, lr in last.items():
+        if lr['rev'] < min_prior:
+            continue
+        tr = this.get(pid, {'rev': 0.0})
+        if tr['rev'] > 0.05 * lr['rev']:   # still selling at >5% of prior → not a zombie
+            continue
+        rows.append({'code': pid, 'product': lr['name'], 'rev_ly': lr['rev'], 'rev_ytd': tr['rev'],
+                     'lost': lr['rev'] - tr['rev']})
+    rows.sort(key=lambda x: -x['lost'])
+    return rows[:limit]
+
+
+def new_winners(t=None, limit=20, min_this=1500.0, team_id=None):
+    """SKUs that barely sold last year but are selling well this year."""
+    t = t or P.today()
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    this = _sku_rev(yf, yt, team_id)
+    last = _sku_rev(lf, lt, team_id)
+    rows = []
+    for pid, tr in this.items():
+        if tr['rev'] < min_this:
+            continue
+        lr = last.get(pid, {'rev': 0.0})
+        if lr['rev'] > 0.05 * tr['rev']:
+            continue
+        rows.append({'code': pid, 'product': tr['name'], 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'],
+                     'gained': tr['rev'] - lr['rev']})
+    rows.sort(key=lambda x: -x['gained'])
+    return rows[:limit]
+
+
+def velocity_leaders(t=None, limit=25, team_id=None):
+    """Top SKUs by LTM unit velocity (units/month) and revenue."""
+    t = t or P.today()
+    lf, lt = P.ltm(t)
+    sku = _sku_rev(lf, lt, team_id)
+    rows = [{'code': k, 'product': v['name'], 'units_ltm': v['qty'], 'units_per_mo': v['qty'] / 12.0,
+             'rev_ltm': v['rev'], 'orders_ltm': v['orders']} for k, v in sku.items()]
+    rows.sort(key=lambda x: -x['units_ltm'])
+    return rows[:limit]
+
+
+def _code_category():
+    """code -> category name. Built from _code_map (pid->code) + the sales category map
+    (pid->category); the first record carrying a code sets that code's category."""
+    codes = _code_map()
+    cats = sales_mod._product_cat()
+    out = {}
+    for pid, code in codes.items():
+        if code not in out and pid in cats:
+            out[code] = cats[pid]
+    return out
+
+
+def directory(t=None, team_id=None):
+    """Every SKU that sold this YTD or the same period last year, with the fields the SKU
+    drill-down filters/sorts on: category, YTD/LY revenue, YoY %, units and orders. Mirrors the
+    customer directory — the full list (filtering happens in the UI), sorted by YTD revenue."""
+    t = t or P.today()
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    this = _sku_rev(yf, yt, team_id)
+    last = _sku_rev(lf, lt, team_id)
+    code_cat = _code_category()
+    rows = []
+    for code in set(this) | set(last):
+        tr = this.get(code, {})
+        lr = last.get(code, {})
+        rev_ytd = tr.get('rev', 0.0)
+        rev_ly = lr.get('rev', 0.0)
+        rows.append({'code': code, 'product': tr.get('name') or lr.get('name') or code,
+                     'category': code_cat.get(code, '(uncategorized)'),
+                     'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'change': rev_ytd - rev_ly,
+                     'yoy_pct': P.yoy_pct(rev_ytd, rev_ly),
+                     'qty_ytd': tr.get('qty', 0.0), 'orders_ytd': tr.get('orders', 0)})
+    rows.sort(key=lambda r: -r['rev_ytd'])
+    return rows
+
+
+def catalogue():
+    """`{sku_code: {'product', 'category', 'id', 'discontinued', 'incoming'}}` — EVERY ACTIVE
+    product, sold or not.
+
+    ⛔ THIS IS THE CATALOGUE UNIVERSE, AND IT IS DELIBERATELY NOT `directory()`. `directory()`'s
+    row set IS the union of two revenue `read_group`s over `sale.order.line` (`:155`), so a SKU
+    that never sold cannot exist in it. That is CORRECT for its own callers — `yoy_movers`,
+    `zombie_skus`, `new_winners` and `categories` all legitimately want a sales-window universe —
+    and it is wrong for the PRODUCT GRID, which is a catalogue and was therefore showing 2,717 of
+    5,875 SKUs (wave 29, owner item 22 / ruling R12). The fix is this function plus a LEFT JOIN in
+    the consumer, never a window removed from `directory()`: removing the window alone lands at
+    3,327 (all-time-sold), because ~2,550 active SKUs have never sold in wholesale scope at all.
+
+    ⚠ NOT BU-SHAPED, and it cannot be: `product.product` carries no team. A catalogue is one
+    catalogue. `directory()` stays the BU-shaped half, which is why the two are joined rather than
+    merged — a scoped caller gets every SKU with ITS OWN revenue, blank where that BU never sold.
+
+    Keyed exactly like `_code_map()` — the `default_code`, or a `pid:N` fallback for the 33 active
+    records that carry none — so the join against `directory()` is code-for-code with no
+    normaliser. Names come from `display_name` (`[CODE] NAME`), which is what `O.m2o_name` yields
+    off a sale line, so a never-sold row wears the same format as a sold one.
+
+    ⛔ RAISES on a truncated read rather than returning a short catalogue. A silently short pull
+    would put the grid back at a plausible wrong number with every gate green — the exact failure
+    this function exists to end ([[no-unverifiable-aggregates]], and the same truncation guard
+    `modules/backorders.py:161` already uses).
+    """
+    dom = [('active', '=', True)]
+    # ⭐ `product_tag_ids` and `incoming_qty` RIDE THIS READ rather than paying for their own.
+    # This function already pulls every active product once; a second full pull for either column
+    # costs ~6 s on somebody's first page load for data that is already in flight.
+    # ⚠ `incoming_qty` is an UNSTORED computed field. It reads fine per record, but Odoo refuses
+    # to aggregate it (`Fault 2: Cannot aggregate field 'incoming_qty'`), so nothing downstream
+    # may push a filter or a read_group on it back to Odoo. It is materialised into the pool here
+    # and summed on our side; `validate()` reconciles that sum against purchase-order lines.
+    prods = O.search_read('product.product', dom,
+                          ['id', 'default_code', 'display_name', 'name',
+                           'product_tag_ids', 'incoming_qty'],
+                          limit=50000)
+    n = O.get_odoo().search_count('product.product', dom)
+    if len(prods) != n:
+        raise ValueError(
+            f"products.catalogue: the product pull is TRUNCATED — read {len(prods)} rows against "
+            f"a search_count of {n}. A short catalogue renders as a plausible smaller grid with "
+            f"nothing reporting it; raise the limit before shipping this.")
+    code_cat = _code_category()
+    disc_tag = discontinued_tag_id()
+    out = {}
+    for p in prods:
+        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
+        # First record wins, matching `_code_category`'s own convention.
+        # ⛔ THE 2026-08-11 MEASUREMENT ("zero active products share a code, so this branch is a
+        # guard, not a merge policy") HAS EXPIRED. RE-MEASURED 2026-08-19: **5,873 active products
+        # carry 5,872 distinct codes** because `2112-12` is now on two records. So this IS a merge
+        # policy today, and it is the standing reason `validate()`'s row-count leg reads 5872
+        # against an Odoo `search_count` of 5873, along with all three pricelist legs being short
+        # by exactly one. Those reds are this duplicate, not a truncated read.
+        # ⚠ Which is also why `incoming` ACCUMULATES below instead of taking the first record's
+        # value: a merge policy that keeps one record's name must still sum the other's stock.
+        # ⭐⭐ W33-T43 (R2 / amendment A2) — `id` RIDES THE ROW. Odoo's `product.product` id is
+        # already in hand (the `pid:{id}` fallback above uses it and then threw it away), and it is
+        # the ONE column `ut_odoo_products` had that `product_data` lacked. R2 merges onto the
+        # legacy key and keeps every data column, so retiring the twin without this would lose a
+        # real identifier rather than a navigational link.
+        # ⚠ A product's grid pid is `crc32(default_code)`, NOT the Odoo id — unlike a CUSTOMER row,
+        # whose pid IS the partner id. That asymmetry is exactly why this cannot be derived
+        # downstream in `aios_grid.py` and has to be carried from the source read.
+        row = out.setdefault(code, {'product': p.get('display_name') or p.get('name') or code,
+                                    'category': code_cat.get(code, '(uncategorized)'),
+                                    'id': p['id'],
+                                    'discontinued': 'No', 'incoming': 0.0})
+        # ⛔ ACROSS EVERY RECORD SHARING THE CODE, not just the first one that won above. A SKU
+        # whose variants are separate records carries its inbound on whichever record the PO named,
+        # and 'first record wins' would drop the rest — a buy list that under-counts what is
+        # already on the water tells you to re-order stock you have already bought.
+        row['incoming'] += float(p.get('incoming_qty') or 0.0)
+        if disc_tag is not None and disc_tag in (p.get('product_tag_ids') or ()):
+            row['discontinued'] = 'Yes'
+    return out
+
+
+#: ⭐ The Odoo product tag that marks a SKU withdrawn from the line. Owner, 2026-08-19: *"we have
+#: a status Field called discontinued from Odoo. Surface it right now so we can filter correctly."*
+#: MEASURED that day: it is not a FIELD at all — `product.product` has no `discontinued` column and
+#: no selection carrying the word. It is `product.tag`, on **497 of 5,873 active products**, every
+#: one of them `active=True` and `sale_ok=True`, so nothing else in Odoo distinguishes them and the
+#: buy list was quoting them as live SKUs.
+DISCONTINUED_TAG = 'Discontinued'
+
+
+@lru_cache(maxsize=1)
+def discontinued_tag_id():
+    """The `product.tag` id for `DISCONTINUED_TAG`, or `None` if the tag is not there.
+
+    ⚠ RESOLVED BY NAME, NOT BY ID, for the reason `PRICELIST_COLUMNS` gives: an id hard-coded here
+    keeps pointing at whatever inherits it if the tag is deleted and re-made, and every product
+    would be silently mislabelled. A missing tag yields `None`, which marks the whole catalogue
+    `No` — so `validate()` reconciles the count against Odoo rather than trusting this, and goes
+    RED on the difference instead of shipping a column that quietly says nobody is discontinued.
+    """
+    rows = O.search_read('product.tag', [('name', '=ilike', DISCONTINUED_TAG)], ['id'], limit=2)
+    return rows[0]['id'] if rows else None
+
+
+def incoming_from_po(prods=None):
+    """`({code: open_po_units}, report)` — inbound derived from PURCHASE ORDER LINES.
+
+    ⛔ THIS IS THE ORACLE, NOT THE SOURCE. The shipped `incoming` column comes from
+    `product.product.incoming_qty` (see `catalogue()`); this derives the same quantity a second,
+    independent way so `validate()` can hold them against each other rather than reconciling a
+    number with itself — the self-sealing failure `product_data.validate` already carries a scar
+    for ([[gate-and-nc-must-not-share-a-binding]]).
+
+    ⛔⛔ AND THE TWO ARE NOT EXPECTED TO BE EQUAL, WHICH IS WHY THE SOURCE IS THE ONE IT IS.
+    MEASURED 2026-08-19: totals 179,979 vs 173,775 units, differing on **107 of ~490 codes**, and
+    the differences are a UNIT OF MEASURE gap, not an error. `purchase.order.line.product_qty` is
+    in the line's PURCHASE uom while `incoming_qty` is in the product's STOCK uom, so the `SP-*`
+    family reads exactly 4x apart (400 vs 100, 320 vs 80, 240 vs 60). `on_hand` and `qty_ltm` are
+    both in the STOCK uom, so `incoming_qty` is the only one of the two that can be ADDED to a
+    shelf quantity. A buy list built on the purchase uom would over-count inbound fourfold on
+    those SKUs and tell the team not to re-order stock that is not coming.
+
+    So the check this feeds asserts SHAPE and DIRECTION (same SKUs carry inbound, totals within a
+    stated band) and REPORTS the per-code gap. Asserting equality would go red forever on a
+    difference that is correct.
+    """
+    lines = O.search_read('purchase.order.line', [('state', '=', 'purchase')],
+                          ['product_id', 'product_qty', 'qty_received'], limit=50000)
+    n = O.get_odoo().search_count('purchase.order.line', [('state', '=', 'purchase')])
+    by_id = {p['id']: p for p in (prods if prods is not None else [])}
+    if not by_id:
+        for p in O.search_read('product.product', [('active', '=', True)],
+                               ['id', 'default_code'], limit=50000):
+            by_id[p['id']] = p
+    out, off_catalogue = {}, 0.0
+    for line in lines:
+        pid = (line.get('product_id') or [None])[0]
+        open_q = float(line.get('product_qty') or 0.0) - float(line.get('qty_received') or 0.0)
+        if open_q <= 0:
+            continue
+        rec = by_id.get(pid)
+        if rec is None:                 # ordered against an ARCHIVED product: real, but off-grid
+            off_catalogue += open_q
+            continue
+        code = (str(rec['default_code']).strip() if rec.get('default_code') else f"pid:{pid}")
+        out[code] = out.get(code, 0.0) + open_q
+    return out, {'lines_read': len(lines), 'lines_total': n,
+                 'truncated': len(lines) != n,
+                 'units_on_archived_products': round(off_catalogue, 1)}
+
+
+#: ⭐⭐ WAVE 30 / W30-T34 (the carried W29-T52) — THE PRICELIST STRATUM, PER LIST.
+#:
+#: MEASURED LIVE 2026-08-12, and every one of these numbers shaped the design rather than
+#: decorating it:
+#:
+#:   · **Five pricelists, three of them material.** Date-valid fixed rules: Fisch 5,139,
+#:     Royal 1 2,702, Royal 2 2,605, Public Pricelist 17, Giftware Deals 1. At SKU level that
+#:     is Fisch 3,971 · Royal 1 1,920 · Royal 2 1,857 · Public 6 · Giftware 0.
+#:   · **3,988 of 5,871 active products carry a usable fixed rule; 1,883 carry NONE.**
+#:   · **1,853 SKUs are priced on all three lists AND THE LISTS DISAGREE** — median relative
+#:     spread 16%, max 82%. That is why this is three columns rather than one "the price" or a
+#:     low/high pair: neither can tell a Fisch rep what Fisch sells the SKU for, which is the
+#:     only question the column exists to answer.
+#:   · ⛔ **`list_price` is 1.00 on 5,817 of the 5,871** — so the single `3_global` rule
+#:     computed over it is meaningless, exactly as W29-T52 said. **This reader never falls
+#:     through to it**, and that is the ticket's own negative control: a SKU with no specific
+#:     item renders BLANK, never a fallback dressed as a price.
+#:
+#: ⚠ RESOLVED BY NAME, NOT BY ID. A pricelist renamed or deleted in Odoo makes its column go
+#: blank and makes `report["lists_missing"]` name it — a hard-coded id would keep pointing at
+#: whatever inherited it and mislabel every cell silently.
+PRICELIST_COLUMNS = (
+    ("price_fisch", "Fisch"),
+    ("price_royal_1", "Royal 1"),
+    ("price_royal_2", "Royal 2"),
+)
+
+
+def pricelist_by_code():
+    """`({code: {column_key: price}}, report)` — the date-valid FIXED base-tier price for each
+    declared pricelist, keyed EXACTLY as `catalogue()` keys its rows.
+
+    ⛔ THE KEYING IS LOAD-BEARING, not a detail. `catalogue()` keys on `default_code` with a
+    `pid:{id}` fallback for the ~33 active records that carry none. Keying this map any other
+    way would leave those SKUs permanently blank while an oracle counting active products
+    counted them — a red gate on a working build, or worse, a silent hole nobody counts.
+
+    **The base tier, deliberately.** `pricecomp._tier_for` picks the highest `min_quantity` at
+    or below an order's quantity, because it is pricing a LINE that has one. A catalogue column
+    has no quantity in hand, so it takes the LOWEST `min_quantity` — the price at qty 1. Rules
+    above that break are a bulk price, and `report["qty_break_only"]` counts the SKUs whose only
+    rule sits on one (MEASURED: 10 rules of 10,464 carry a break at all).
+
+    **Variant rules beat template rules**, matching `pricecomp._tier_for`: a `0_product_variant`
+    rule is the more specific statement about this exact SKU.
+
+    ⚠ DEGRADES TO `({}, report)` on a read failure, matching `_inventory_by_code` rather than
+    `catalogue()`: these are COLUMNS, and a product grid that will not render because pricing is
+    momentarily unreachable is a worse failure than one with blank price columns. The blanks are
+    not silent — `product_data.validate()`'s coverage leg reconciles each column against a fresh
+    Odoo count and goes red at zero.
+
+    ⭐ **The report exists because R6's second sentence is law** (*"if it can't be done, you need
+    to explicitly tell me why and recommend a fix"*). Everything this reader CANNOT see is
+    counted rather than dropped: rules on pricelists the contract does not declare, non-fixed
+    (`formula`/`percent`) rules, the `3_global` fallback, and prices that only exist above a
+    quantity break.
+    """
+    # ⚠ THE COUNTERS ARE DISJOINT AND ORDER-DEPENDENT, and saying so is the difference between a
+    # report and a misleading one. A rule is classified ONCE, by the first reason it is skipped:
+    # undeclared list → out of date → not fixed → global → zero price. So `rules_not_fixed: 0`
+    # means "no formula rule on a list we declare", NOT "this Odoo has no formula rules"
+    # (MEASURED 2026-08-12: it has exactly one, and it sits on Public Pricelist, which the
+    # contract does not declare — so it lands in `rules_undeclared_list`).
+    report = {"lists_missing": [], "rules_total": 0, "rules_undeclared_list": 0,
+              "rules_not_fixed": 0, "rules_global": 0, "rules_out_of_date": 0,
+              "rules_zero_price": 0, "qty_break_only": 0}
+    try:
+        pl_rows = O.search_read('product.pricelist', [], ['id', 'name'])
+        by_name = {}
+        for p in pl_rows:
+            by_name.setdefault(str(p.get('name') or '').strip(), p['id'])
+        wanted = {}
+        for col, name in PRICELIST_COLUMNS:
+            pid = by_name.get(name)
+            if pid is None:
+                report["lists_missing"].append(name)
+            else:
+                wanted[pid] = col
+
+        # ⭐⭐ FILTERED SERVER-SIDE, AND THE REASON IS A MEASUREMENT, NOT A STYLE PREFERENCE.
+        # Reading all 10,464 rules with 9 fields and sorting them in Python costs **30.1s** on
+        # this connection; the same rules under a server-side domain with 6 fields cost
+        # **13.6s** (measured 2026-08-12, back to back). `pool()` runs this on a scope's
+        # first-ever build, so that 16.5s is 16.5s of somebody's page load.
+        #
+        # ⚠ The domain reproduces the Python predicate EXACTLY, and the null legs are the part
+        # that is easy to get wrong: an absent `date_start` is `False`, not a past date, so
+        # `('date_start','<=',today)` ALONE would drop every open-ended rule — which is almost
+        # all of them.
+        today = P.today().isoformat()
+        _declared = sorted(wanted)
+        _kinds = ['0_product_variant', '1_product']
+        _live = [('pricelist_id', 'in', _declared), ('compute_price', '=', 'fixed'),
+                 ('applied_on', 'in', _kinds),
+                 '|', ('date_start', '=', False), ('date_start', '<=', today),
+                 '|', ('date_end', '=', False), ('date_end', '>=', today),
+                 ('fixed_price', '>', 0)]
+        rules = O.search_read(
+            'product.pricelist.item', _live,
+            ['pricelist_id', 'product_id', 'product_tmpl_id', 'applied_on', 'fixed_price',
+             'min_quantity']) if _declared else []
+
+        # ⭐ R6's SECOND SENTENCE, PAID FOR WITH `search_count` RATHER THAN A WIDER READ. Six
+        # counts cost ~1.2s together; the rules they count would cost 16s to read. What this
+        # reader cannot see is still REPORTED — it is just no longer transferred.
+        def _n(extra):
+            try:
+                return O.get_odoo().search_count('product.pricelist.item', extra)
+            except Exception:
+                return -1                      # -1 reads as "not measured", never as zero
+        _dated = ['|', ('date_start', '=', False), ('date_start', '<=', today),
+                  '|', ('date_end', '=', False), ('date_end', '>=', today)]
+        report["rules_total"] = _n([])
+        report["rules_undeclared_list"] = _n([('pricelist_id', 'not in', _declared)]) \
+            if _declared else report["rules_total"]
+        if _declared:
+            _on = [('pricelist_id', 'in', _declared)]
+            # `percent_price` and formula rules are read NOWHERE in this repo. A rule we cannot
+            # price is one the operator is TOLD about, never one that quietly becomes a blank.
+            report["rules_not_fixed"] = _n(_on + [('compute_price', '!=', 'fixed')])
+            report["rules_global"] = _n(_on + [('compute_price', '=', 'fixed'),
+                                               ('applied_on', 'not in', _kinds)])
+            report["rules_out_of_date"] = (
+                _n(_on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)])
+                - _n(_on + [('compute_price', '=', 'fixed'),
+                            ('applied_on', 'in', _kinds)] + _dated))
+            # A zero price is not "free" — it is an unset rule. Blank says so; 0 does not.
+            report["rules_zero_price"] = _n(
+                _on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)]
+                + _dated + [('fixed_price', '<=', 0)])
+
+        by_var, by_tmpl = {}, {}
+        for r in rules:
+            col = wanted[O.m2o_id(r.get('pricelist_id'))]
+            if r.get('applied_on') == '0_product_variant' and r.get('product_id'):
+                by_var.setdefault((col, O.m2o_id(r['product_id'])), []).append(r)
+            elif r.get('product_tmpl_id'):
+                by_tmpl.setdefault((col, O.m2o_id(r['product_tmpl_id'])), []).append(r)
+
+        dom = [('active', '=', True)]
+        prods = O.search_read('product.product', dom,
+                              ['id', 'default_code', 'product_tmpl_id'], limit=50000)
+        n = O.get_odoo().search_count('product.product', dom)
+        if len(prods) != n:
+            # Same guard, same reason as `catalogue()`: a short pull renders as a plausible
+            # smaller set of priced SKUs with nothing reporting it.
+            raise ValueError(
+                f"products.pricelist_by_code: the product pull is TRUNCATED — read {len(prods)} "
+                f"rows against a search_count of {n}.")
+    except Exception as e:
+        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
+        return {}, report
+
+    out = {}
+    for p in prods:
+        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
+        tmpl = O.m2o_id(p.get('product_tmpl_id'))
+        cells = {}
+        for col, _name in PRICELIST_COLUMNS:
+            cands = by_var.get((col, p['id'])) or by_tmpl.get((col, tmpl))
+            if not cands:
+                continue
+            base = min(cands, key=lambda r: r.get('min_quantity') or 0.0)
+            if (base.get('min_quantity') or 0.0) > 1.0:
+                report["qty_break_only"] += 1
+            cells[col] = base.get('fixed_price')
+        if cells:
+            out.setdefault(code, {}).update(cells)
+    return out, report
+
+
+def active_products(limit=50000):
+    """`[{id, default_code, product_tmpl_id}]` for every ACTIVE product — read ONCE and shared.
+
+    ⛔ RAISES ON A SHORT PULL, like `catalogue()` and for the same reason: a truncated product read
+    renders as a plausible smaller set of priced SKUs with nothing reporting it.
+    ⭐ It exists so `tier_prices_by_code` and `packagings_by_code` can be called from ONE pool
+    build without each paying for its own copy of the same 5,873-row read — measured at ~4 s each
+    on this connection, on a path that is somebody's first page load.
+    """
+    dom = [('active', '=', True)]
+    prods = O.search_read('product.product', dom, ['id', 'default_code', 'product_tmpl_id'],
+                          limit=limit)
+    n = O.get_odoo().search_count('product.product', dom)
+    if len(prods) != n:
+        raise ValueError(f"products.active_products: the product pull is TRUNCATED. Read "
+                         f"{len(prods)} rows against a search_count of {n}.")
+    return prods
+
+
+def tier_prices_by_code(prods=None):
+    """`({code: [{pricelist, unit_price}]}, report)` — EVERY live price a SKU really has.
+
+    ⭐⭐ W37-T14 (owner: multiple prices per SKU). `pricelist_by_code` above answers a DIFFERENT
+    question and both are needed: it fills three DECLARED columns (`price_fisch`, `price_royal_1`,
+    `price_royal_2`) and therefore cannot show a price on a list the contract does not name. This
+    one is the honest set — one entry per pricelist that actually prices this SKU today.
+
+    ⛔ IT READS EVERY LIVE PRICELIST, not `PRICELIST_COLUMNS`. Hardcoding this tenant's three list
+    names into "how many prices does a SKU have" is exactly what the ticket forbids, and it is what
+    would make the answer wrong for tenant #1 on the day they are onboarded.
+
+    ⛔ ROWS ARE TEMPLATE-SCOPED (`proto/P2-pricing-uom.md`): joining on `product_id` drops 99.9% of
+    price rows, because `applied_on` is `1_product` on 10,454 of 10,468 items and `1_product` means
+    the TEMPLATE. A variant rule (`0_product_variant`, 13 rows) is the more specific statement and
+    wins, matching `pricecomp._tier_for` and `pricelist_by_code`.
+
+    ⛔ FILTER BY `pricelist_id`, NEVER BY `active`: the default `product.pricelist.item` count hides
+    2,580 archived items, so an `active` filter reads as a smaller, plausible, wrong set.
+
+    ⚠ THE BASE TIER, at qty 1, for the same reason `pricelist_by_code` gives: a catalogue cell has
+    no quantity in hand. Measured: only 10 of 10,468 items carry a quantity break at all, so this
+    is very nearly the whole story rather than a simplification.
+    ⚠ CARDINALITY 1..3 TODAY, mode 3 — and `Royal 2` carries 2,605 price rows against **0 customers
+    and 0 orders**, so a SKU reading "3 tiers" is catalogue-true and commercially misleading. The
+    entry keeps the list NAME so a reader can see which tier it is rather than a bare count.
+    """
+    report = {"lists": [], "rules_total": 0, "rules_not_fixed": 0, "rules_out_of_date": 0,
+              "rules_global": 0, "skus_with_no_price": 0, "identity_breaks": 0}
+    try:
+        pls = {p['id']: str(p.get('name') or '').strip()
+               for p in O.search_read('product.pricelist', [], ['id', 'name'])}
+        report["lists"] = sorted(pls.values())
+        today = P.today().isoformat()
+        dom = [('pricelist_id', 'in', sorted(pls)), ('compute_price', '=', 'fixed'),
+               ('applied_on', 'in', ['0_product_variant', '1_product']),
+               '|', ('date_start', '=', False), ('date_start', '<=', today),
+               '|', ('date_end', '=', False), ('date_end', '>=', today),
+               ('fixed_price', '>', 0)]
+        rules = O.search_read('product.pricelist.item', dom,
+                              ['pricelist_id', 'product_id', 'product_tmpl_id', 'applied_on',
+                               'fixed_price', 'min_quantity'])
+
+        # R6's second sentence: what this reader cannot see is COUNTED, never dropped.
+        def _n(extra):
+            try:
+                return O.get_odoo().search_count('product.pricelist.item', extra)
+            except Exception:                                          # noqa: BLE001
+                return -1                     # -1 reads as "not measured", never as zero
+        report["rules_total"] = _n([])
+        report["rules_not_fixed"] = _n([('compute_price', '!=', 'fixed')])
+        report["rules_global"] = _n([('applied_on', '=', '3_global')])
+        report["rules_out_of_date"] = _n(['|', ('date_end', '!=', False),
+                                          ('date_start', '!=', False)])
+
+        by_var, by_tmpl = {}, {}
+        for r in rules:
+            plid = O.m2o_id(r.get('pricelist_id'))
+            if r.get('applied_on') == '0_product_variant' and r.get('product_id'):
+                by_var.setdefault((plid, O.m2o_id(r['product_id'])), []).append(r)
+            elif r.get('product_tmpl_id'):
+                by_tmpl.setdefault((plid, O.m2o_id(r['product_tmpl_id'])), []).append(r)
+
+        prods = active_products() if prods is None else prods
+    except Exception as e:                                             # noqa: BLE001
+        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
+        return {}, report
+
+    out = {}
+    for p in prods:
+        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
+        tmpl = O.m2o_id(p.get('product_tmpl_id'))
+        tiers = []
+        for plid, name in sorted(pls.items(), key=lambda kv: kv[1].lower()):
+            cands = by_var.get((plid, p['id'])) or by_tmpl.get((plid, tmpl))
+            if not cands:
+                continue
+            base = min(cands, key=lambda r: r.get('min_quantity') or 0.0)
+            tiers.append({"pricelist": name, "unit_price": round(base.get('fixed_price') or 0.0, 2)})
+        if tiers:
+            # ⚠ A code carried by TWO active products (D-309: `2112-12`) MERGES here, because the
+            # grid is keyed by code and one code is one row. The prices are UNIONED rather than
+            # one silently winning — a SKU that really does have two different Fisch prices should
+            # show both, and hiding one is how a $60 price disappeared behind a $24 one.
+            prior = out.get(code)
+            if prior is None:
+                out[code] = tiers
+            else:
+                seen = {(t["pricelist"], t["unit_price"]) for t in prior}
+                for t in tiers:
+                    if (t["pricelist"], t["unit_price"]) not in seen:
+                        prior.append(t)
+                        report["identity_breaks"] += 1
+        else:
+            report["skus_with_no_price"] += 1
+    return out, report
+
+
+def packagings_by_code(prods=None):
+    """`({code: [{name, qty}]}, report)` — the UNITS a SKU is really sold in (W37-T15).
+
+    ⛔ IT COMES FROM `product.packaging` ALONE, and `proto/P2-pricing-uom.md` measured why the
+    obvious alternative is a dead end: the `uom.uom` Config category holds 26 conversion-bearing
+    units (`Case-Packed 12` … `Pallet-Packed 4,800`) referenced by **0 products, 0 sale lines and
+    0 stock moves**, and every unit actually in use has `factor_inv = 1`. Building on `uom.uom`
+    conversions yields a column of 1s.
+
+    ⛔ `qty > 1` IS A FILTER, NOT A TIDY-UP. 6,283 of 7,471 packaging rows are `qty = 1.0` — a
+    packaging that packs one of something is not a unit tier — plus real junk (one-character
+    names). Without it "units per SKU" reads **5,859** SKUs instead of **1,150**.
+
+    ⛔ 133 ROWS ARE ORPHANS (`product_id = False`) and would collapse under a `groupby(product_id)`
+    into ONE fake product carrying 133 packagings. Dropped, and counted.
+
+    ⚠ HONEST EMPTY: only **1,150 of 5,873 SKUs (19.6%)** have any unit tier, so this cell is
+    legitimately blank for 80% of the catalogue — "this SKU is sold in one unit", never "missing".
+    The 19.6% is not decoration: units are transacted on 77.9% of confirmed sale lines.
+    ⚠ `qty` is denominated in the product's OWN `uom_id` (matched on 7,338/7,338).
+    """
+    report = {"rows_total": 0, "rows_orphan": 0, "rows_qty_le_1": 0, "skus_with_units": 0}
+    try:
+        rows = O.search_read('product.packaging', [], ['id', 'name', 'qty', 'product_id'])
+    except Exception as e:                                             # noqa: BLE001
+        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
+        return {}, report
+    report["rows_total"] = len(rows)
+    by_pid = {}
+    for r in rows:
+        pid = O.m2o_id(r.get('product_id'))
+        if not pid:
+            report["rows_orphan"] += 1
+            continue
+        if (r.get('qty') or 0) <= 1:
+            report["rows_qty_le_1"] += 1
+            continue
+        by_pid.setdefault(pid, []).append(
+            {"name": str(r.get('name') or '').strip(), "qty": float(r.get('qty') or 0)})
+    try:
+        prods = active_products() if prods is None else prods
+    except Exception as e:                                             # noqa: BLE001
+        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
+        return {}, report
+    out = {}
+    for p in prods:
+        got = by_pid.get(p['id'])
+        if not got:
+            continue
+        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
+        out.setdefault(code, []).extend(sorted(got, key=lambda u: u["qty"]))
+    report["skus_with_units"] = len(out)
+    return out, report
+
+
+def catalogue_count():
+    """The INDEPENDENT population oracle: Odoo's own count of active products.
+
+    Deliberately a bare `search_count` and not a `len()` over anything this module built — the
+    2,717 defect shipped silently for a wave because `product_data.validate()` derived BOTH sides
+    of its reconciliation from `_sku_rev`, so the oracle could never see a missing row.
+    """
+    return O.get_odoo().search_count('product.product', [('active', '=', True)])
+
+
+def categories(t=None, team_id=None):
+    """Sorted distinct category names present in the SKU directory (for the drill-down filter)."""
+    return sorted({r['category'] for r in directory(t, team_id)})
+
+
+def _coverage(date_from, date_to, team_id=None):
+    """{sku_code: distinct_customer_count} via 2-level read_group (slow-ish), merged by code.
+    (A customer buying two records sharing a code can count twice; duplicates are rare, and
+    code-level avoids the bigger error of a re-SKUed product reading as full coverage loss.)"""
+    g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC),
+                     ['__count'], ['product_id', 'order_partner_id'], lazy=False)
+    codes = _code_map()
+    cov = {}
+    names = {}
+    for r in g:
+        pid = O.m2o_id(r.get('product_id'))
+        if not pid:
+            continue
+        key = codes.get(pid, f"pid:{pid}")
+        cov[key] = cov.get(key, 0) + 1
+        names.setdefault(key, O.m2o_name(r.get('product_id')))
+    return cov, names
+
+
+def coverage_collapse(t=None, limit=25, min_prior_custs=8, team_id=None):
+    """SKUs that lost the most customer breadth YoY — early warning a SKU is dying even if
+    revenue hasn't fully cratered yet."""
+    t = t or P.today()
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    cov_t, names_t = _coverage(yf, yt, team_id)
+    cov_l, names_l = _coverage(lf, lt, team_id)
+    rows = []
+    for pid, lc in cov_l.items():
+        if lc < min_prior_custs:
+            continue
+        tc = cov_t.get(pid, 0)
+        drop = lc - tc
+        if drop <= 0:
+            continue
+        rows.append({'code': pid, 'product': names_l.get(pid) or names_t.get(pid) or '',
+                     'custs_ly': lc, 'custs_ytd': tc, 'lost_custs': drop,
+                     'pct_drop': drop / lc * 100})
+    rows.sort(key=lambda x: (-x['lost_custs'], -x['pct_drop']))
+    return rows[:limit]
+
+
+# ====================================================================== SKU DRAWER (mirror of customer drawer)
+def _sku_product_ids(code):
+    """All product.product ids sharing this SKU code (merged variants / archived records)."""
+    return [pid for pid, c in _code_map().items() if c == code]
+
+
+def _sku_dom(pids, date_from, date_to, team_id=None):
+    return O.sale_line_domain(date_from, date_to, team_id, extra=[('product_id', 'in', pids)] + _NO_SVC)
+
+
+def _sku_name_category(pids):
+    rows = O.search_read('product.product', [('id', 'in', pids)], ['name', 'categ_id'])
+    name = rows[0]['name'] if rows else '(unknown)'
+    catmap = sales_mod._product_cat()
+    main = next((catmap.get(p) for p in pids if catmap.get(p)), '(uncategorized)')
+    return name, main
+
+
+def _sku_buyers(pids, date_from, date_to, team_id=None):
+    """{partner_id: {'name','rev','qty'}} for buyers of this SKU over a window."""
+    g = O.read_group('sale.order.line', _sku_dom(pids, date_from, date_to, team_id),
+                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False)
+    out = {}
+    for r in g:
+        pid = O.m2o_id(r.get('order_partner_id'))
+        if pid:
+            out[pid] = {'name': O.m2o_name(r.get('order_partner_id')),
+                        'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0}
+    return out
+
+
+def sku_detail(code, t=None, team_id=None, n_months=13, allsku=None):
+    """KPIs (rev/qty/buyers YoY, GM%), monthly trend, rank & % of BU for one SKU code.
+    Pass `allsku` (a cached _sku_rev YTD map) to skip the ~all-SKU rank read."""
+    t = t or P.today()
+    pids = _sku_product_ids(code)
+    if not pids:
+        return None
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    mf, mt = P.ltm(t)
+    name, category = _sku_name_category(pids)
+
+    def s(df, dtt, field='price_subtotal'):
+        return O.sum_field('sale.order.line', _sku_dom(pids, df, dtt, team_id), field)
+    rev_ytd, rev_ly = s(yf, yt), s(lf, lt)
+    qty_ytd, qty_ly = s(yf, yt, 'product_uom_qty'), s(lf, lt, 'product_uom_qty')
+    buyers_ytd = len(_sku_buyers(pids, yf, yt, team_id))
+    buyers_ly = len(_sku_buyers(pids, lf, lt, team_id))
+    g = O.read_group('sale.order.line', _sku_dom(pids, mf, mt, team_id),
+                     ['price_subtotal:sum', 'margin:sum'], [], lazy=False)
+    line_rev = (g[0].get('price_subtotal') if g else 0) or 0.0
+    margin = (g[0].get('margin') if g else 0) or 0.0
+
+    # Monthly trend: bin lines by their order's month in Python (dot-path month groupby is rejected
+    # on sale.order.line), in 2 reads instead of 26 point queries.
+    range_start = dt.date(t.year - 2, t.month, 1).isoformat()
+    lines = O.search_read('sale.order.line', _sku_dom(pids, range_start, t.isoformat(), team_id),
+                          ['price_subtotal', 'order_id'])
+    oids = list({O.m2o_id(line['order_id']) for line in lines if line.get('order_id')})
+    odate = {}
+    for i in range(0, len(oids), 1000):
+        for o in O.search_read('sale.order', [('id', 'in', oids[i:i + 1000])], ['date_order']):
+            if o.get('date_order'):
+                odate[o['id']] = str(o['date_order'])[:7]
+    mrev = {}
+    for line in lines:
+        ym = odate.get(O.m2o_id(line.get('order_id')))
+        if ym:
+            mrev[ym] = mrev.get(ym, 0.0) + (line.get('price_subtotal') or 0.0)
+    monthly = []
+    for ym, start, end in P.month_starts(n_months, t):
+        y, m = int(ym[:4]) - 1, int(ym[5:7])
+        monthly.append({'month': ym, 'revenue': mrev.get(ym, 0.0),
+                        'revenue_ly': mrev.get(f'{y:04d}-{m:02d}', 0.0)})
+
+    allsku = allsku if allsku is not None else _sku_rev(yf, yt, team_id)
+    total = sum(v['rev'] for v in allsku.values()) or 1.0
+    rank = next((i + 1 for i, (c, _v) in enumerate(sorted(allsku.items(), key=lambda kv: -kv[1]['rev']))
+                 if c == code), None)
+    return {
+        'code': code, 'name': name, 'category': category,
+        'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'rev_yoy_pct': P.yoy_pct(rev_ytd, rev_ly),
+        'qty_ytd': qty_ytd, 'qty_ly': qty_ly, 'qty_yoy_pct': P.yoy_pct(qty_ytd, qty_ly),
+        'buyers_ytd': buyers_ytd, 'buyers_ly': buyers_ly, 'buyers_delta': buyers_ytd - buyers_ly,
+        'gm_pct': (margin / line_rev * 100) if line_rev else 0.0, 'gm_dollars': margin,
+        'rank': rank, 'n_skus': len(allsku), 'pct_of_bu': rev_ytd / total * 100,
+        'monthly': monthly,
+    }
+
+
+def sku_buyer_bridge(code, t=None, team_id=None, top=12):
+    """Who drives the SKU's YoY: retained / new / churned buyers + a $-ranked churned call list."""
+    t = t or P.today()
+    pids = _sku_product_ids(code)
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id)
+    tset, lset = set(this), set(last)
+    retained, new, churned = tset & lset, tset - lset, lset - tset
+    churned_list = sorted([{'pid': p, 'customer': last[p]['name'], 'ly_rev': last[p]['rev']}
+                           for p in churned], key=lambda x: -x['ly_rev'])[:top]
+    return {'retained': {'n': len(retained), 'rev': sum(this[p]['rev'] for p in retained)},
+            'new': {'n': len(new), 'rev': sum(this[p]['rev'] for p in new)},
+            'churned': {'n': len(churned), 'rev': sum(last[p]['rev'] for p in churned)},
+            'buyer_retention_pct': (len(retained) / len(lset) * 100) if lset else 0.0,
+            'churned_buyers': churned_list, 'buyers_this': len(tset), 'buyers_last': len(lset),
+            'this_total': sum(v['rev'] for v in this.values())}
+
+
+def sku_concentration(code, t=None, team_id=None):
+    """Buyer-concentration risk: top-1/top-3 share, Herfindahl index, effective buyer count (LTM)."""
+    t = t or P.today()
+    mf, mt = P.ltm(t)
+    buyers = _sku_buyers(_sku_product_ids(code), mf, mt, team_id)
+    revs = sorted([v['rev'] for v in buyers.values()], reverse=True)
+    total = sum(revs) or 1.0
+    hhi = sum((r / total) ** 2 for r in revs)
+    return {'n_buyers': len(revs), 'top1_pct': (revs[0] / total * 100) if revs else 0.0,
+            'top3_pct': (sum(revs[:3]) / total * 100) if revs else 0.0,
+            'hhi': hhi, 'eff_buyers': (1 / hhi) if hhi else 0.0}
+
+
+def sku_price_dispersion(code, t=None, team_id=None, cap=60):
+    """Realized $/unit per buyer (LTM) vs the volume-weighted average; recoverable $ on below-VWAP
+    accounts. list_price is unreliable, so the dispersion among actual buyers is the margin lever."""
+    t = t or P.today()
+    mf, mt = P.ltm(t)
+    g = O.read_group('sale.order.line', _sku_dom(_sku_product_ids(code), mf, mt, team_id),
+                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False)
+    rows, tot_rev, tot_qty = [], 0.0, 0.0
+    for r in g:
+        pid = O.m2o_id(r.get('order_partner_id'))
+        rev = r.get('price_subtotal') or 0.0
+        qty = r.get('product_uom_qty') or 0.0
+        if not pid or qty <= 0:
+            continue
+        rows.append({'pid': pid, 'customer': O.m2o_name(r.get('order_partner_id')),
+                     'price': rev / qty, 'qty': qty, 'rev': rev})
+        tot_rev += rev
+        tot_qty += qty
+    vwap = (tot_rev / tot_qty) if tot_qty else 0.0
+    for r in rows:
+        r['recoverable'] = max(0.0, vwap - r['price']) * r['qty']
+    rows.sort(key=lambda x: -x['recoverable'])
+    return {'vwap': vwap, 'n_buyers': len(rows),
+            'recoverable_total': sum(r['recoverable'] for r in rows), 'rows': rows[:cap]}
+
+
+def sku_top_buyers(code, t=None, team_id=None, top=15):
+    """Ranked buyers of this SKU (YTD) with YoY — clickable to open the customer drawer."""
+    t = t or P.today()
+    pids = _sku_product_ids(code)
+    yf, yt = P.ytd(t)
+    lf, lt = P.ytd_last_year(t)
+    this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id)
+    rows = [{'pid': p, 'customer': v['name'], 'rev': v['rev'], 'qty': v['qty'],
+             'yoy_pct': P.yoy_pct(v['rev'], last.get(p, {}).get('rev', 0.0))} for p, v in this.items()]
+    rows.sort(key=lambda x: -x['rev'])
+    return rows[:top]
+
+
+_BUYER_TIERS = [('Whale (≥$25k)', 25000.0), ('Large ($10–25k)', 10000.0),
+                ('Mid ($2–10k)', 2000.0), ('Small (<$2k)', 0.0)]
+
+
+def sku_customer_analysis(code, t=None, team_id=None, top=15):
+    """WHO buys this SKU (LTM), as customers: the value-tier mix of its buyers (by each buyer's TOTAL
+    spend), and the accounts most DEPENDENT on it (this SKU as a share of their spend — who gets hurt
+    most if it stocks out / who to protect)."""
+    t = t or P.today()
+    lf, lt = P.ltm(t)
+    pids = _sku_product_ids(code)
+    empty = {'segments': [], 'dependency': [], 'n_buyers': 0, 'avg_dependency': 0.0}
+    if not pids:
+        return empty
+    buyers = _sku_buyers(pids, lf, lt, team_id)        # {pid: {name, rev(this SKU), qty}}
+    if not buyers:
+        return empty
+    bpids = list(buyers)
+    g = O.read_group('sale.order', sales_mod.order_domain(lf, lt, team_id) + [('partner_id', 'in', bpids)],
+                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
+    total = {O.m2o_id(r['partner_id']): (r.get('amount_untaxed') or 0.0) for r in g if r.get('partner_id')}
+
+    def tier(rev):
+        for nm, lo in _BUYER_TIERS:
+            if rev >= lo:
+                return nm
+        return _BUYER_TIERS[-1][0]
+    seg = {nm: {'tier': nm, 'buyers': 0, 'sku_rev': 0.0} for nm, _ in _BUYER_TIERS}
+    dep = []
+    for p, v in buyers.items():
+        ct = total.get(p, v['rev']) or v['rev']
+        s = seg[tier(ct)]
+        s['buyers'] += 1
+        s['sku_rev'] += v['rev']
+        dep.append({'customer': v['name'], 'pid': p, 'sku_rev': v['rev'], 'cust_total': ct,
+                    'dependency_pct': (v['rev'] / ct * 100) if ct else None})
+    tot = sum(s['sku_rev'] for s in seg.values()) or 1.0
+    segments = []
+    for nm, _ in _BUYER_TIERS:
+        s = seg[nm]
+        s['rev_share'] = s['sku_rev'] / tot * 100
+        s['avg_per_buyer'] = (s['sku_rev'] / s['buyers']) if s['buyers'] else 0.0
+        segments.append(s)
+    dep.sort(key=lambda x: -x['sku_rev'])
+    deps_known = [d['dependency_pct'] for d in dep if d['dependency_pct'] is not None]
+    return {'segments': segments, 'dependency': dep[:top], 'n_buyers': len(buyers),
+            'avg_dependency': (sum(deps_known) / len(deps_known)) if deps_known else 0.0}
+
+
+def sku_whitespace(code, t=None, team_id=None, top=15):
+    """Customers who buy this SKU's category but NOT this SKU — ranked prospect list."""
+    import modules.customers as cust_mod
+    t = t or P.today()
+    pids = _sku_product_ids(code)
+    _name, category = _sku_name_category(pids)
+    cat_buyers = cust_mod.category_buyers(category, team_id=team_id) or set()
+    mf, mt = P.ltm(t)
+    prospects = list(cat_buyers - set(_sku_buyers(pids, mf, mt, team_id)))
+    if not prospects:
+        return {'category': category, 'rows': []}
+    g = O.read_group('sale.order', sales_mod.order_domain(mf, mt, team_id) + [('partner_id', 'in', prospects)],
+                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
+    rows = [{'pid': O.m2o_id(r['partner_id']), 'customer': O.m2o_name(r['partner_id']),
+             'total_spend': r.get('amount_untaxed') or 0.0} for r in g if r.get('partner_id')]
+    rows.sort(key=lambda x: -x['total_spend'])
+    return {'category': category, 'rows': rows[:top]}
+
+
+def sku_drawer_bundle(code, t=None, team_id=None, allsku=None):
+    """The whole SKU drawer's first paint in one cached unit: sku_detail first (the not-found gate),
+    then the other six pulls CONCURRENTLY (O.parallel). Cuts a ~6-call cold open to ~max(call)."""
+    detail = sku_detail(code, t=t, team_id=team_id, allsku=allsku)
+    if detail is None:
+        return {'detail': None}
+    bridge, conc, price, buyers, white, ca = O.parallel([
+        lambda: sku_buyer_bridge(code, t, team_id),
+        lambda: sku_concentration(code, t, team_id),
+        lambda: sku_price_dispersion(code, t, team_id),
+        lambda: sku_top_buyers(code, t, team_id),
+        lambda: sku_whitespace(code, t, team_id),
+        lambda: sku_customer_analysis(code, t, team_id),
+    ])
+    return {'detail': detail, 'bridge': bridge, 'conc': conc, 'price': price,
+            'buyers': buyers, 'white': white, 'ca': ca}
+
+
+def validate(t=None, team_id=None):
+    """Reconcile SKU metrics to independent Odoo aggregates. When team_id is set (a single BU
+    selected) every check runs SCOPED to that BU, so the validation panel never reconciles
+    against — or exposes — the other BU's numbers. All checks here are BU-scopeable (no cross-BU
+    mirror), so team_id threads straight through."""
+    t = t or P.today()
+    yf, yt = P.ytd(t)
+    checks = []
+    sku = _sku_rev(yf, yt, team_id)
+    sku_sum = sum(v['rev'] for v in sku.values())
+    line_total = O.sum_field('sale.order.line',
+                             O.sale_line_domain(yf, yt, team_id, extra=_NO_SVC), 'price_subtotal')
+    checks.append({'check': 'SKU rev: Σ(per-SKU) == total line revenue, ex-services (YTD)',
+                   'a': round(sku_sum, 2), 'b': round(line_total, 2),
+                   'gap': round(sku_sum - line_total, 2),
+                   'ok': abs(sku_sum - line_total) <= 1.0})
+
+    m = yoy_movers(t, limit=10**9, team_id=team_id)
+    movers_sum = sum(r['change'] for r in m['risers']) + sum(r['change'] for r in m['decliners'])
+    lf, lt = P.ytd_last_year(t)
+    last_total = O.sum_field('sale.order.line',
+                             O.sale_line_domain(lf, lt, team_id, extra=_NO_SVC), 'price_subtotal')
+    checks.append({'check': 'SKU movers: Σ(Δ) == (YTD − LY) total',
+                   'a': round(movers_sum, 2), 'b': round(line_total - last_total, 2),
+                   'gap': round(movers_sum - (line_total - last_total), 2),
+                   'ok': abs(movers_sum - (line_total - last_total)) <= 1.0})
+
+    # SKU drawer: buyer-bridge reconciles — retained + new buyer revenue == this-period SKU revenue
+    top_code = max(sku.items(), key=lambda kv: kv[1]['rev'])[0] if sku else None
+    if top_code:
+        bb = sku_buyer_bridge(top_code, t, team_id=team_id)
+        recon = bb['retained']['rev'] + bb['new']['rev']
+        checks.append({'check': 'SKU drawer: retained+new buyer rev == SKU YTD revenue',
+                       'a': round(recon, 2), 'b': round(bb['this_total'], 2),
+                       'gap': round(recon - bb['this_total'], 2),
+                       'ok': abs(recon - bb['this_total']) <= 1.0})
+    return checks