diff --git "a/api/routes_grid.py" "b/api/routes_grid.py"
--- "a/api/routes_grid.py"
+++ "b/api/routes_grid.py"
@@ -1,1083 +1,1083 @@
-"""routes_grid.py — X2's write seam over HTTP: the SECOND adapter on `core.grid_events` (EXIT-1c).
-
-`POST /api/v1/grid/events` takes the component's event objects VERBATIM — the same objects the
-Streamlit host receives through the component value slot, unchanged — and runs them through the
-same `core.grid_events.handle_events` the Streamlit adapter runs. That is the whole point of
-EXIT-1a: one implementation of every permission wall, two transports. A route that re-validated
-anything here would be a second wall to keep in step, and the two would drift on the first change.
-
-DEDUP IS PER REQUEST, and that is a deliberate limit, not an oversight. The Streamlit adapter's
-`seen_ids` lives in `st.session_state` — genuinely per user session — because the client resends
-its recent 24-event window on every emit inside one page session. A stateless API has no such
-dict, and inventing a per-user server-side one would be exactly the resident per-tenant state
-EXIT-4a exists to remove. So: ids are deduped WITHIN a request body (the resend window's whole
-purpose — a batch that repeats an id processes it once), and a genuinely replayed request is
-handled by the operations being idempotent. The one event where a replay is observable is
-`add_to_list` (it would add the same pids to the same cohort twice — a set union, so the
-membership is unchanged, but the toast count repeats). Noted rather than papered over; a
-server-side idempotency key belongs with D2's session mirror (C-2).
-
-STORE DOWN = 503. `fallback_ws` is None on this adapter, so `core.grid_events` raises
-`StoreUnavailable` rather than writing to an in-memory workspace no API request could ever read
-back. A 200 over a write that evaporated is the failure this rule exists to prevent.
-"""
-import datetime as dt
-import time
-
-from fastapi import APIRouter, Body, Depends
-
-from deps import Session, err, module_gate, perms, require_session
-
-router = APIRouter(prefix="/api/v1")
-
-MODULE = "customer_data"
-
-#: The client's own resend window (`app.py`'s `[-24:]`). A body larger than this is not a
-#: legitimate client — refuse it rather than doing 500 store writes on one request.
-_MAX_EVENTS = 24
-
-
-def _ctx(session: Session, fields, pids, **kw):
-    from core import grid_events
-    import aios_grid
-    import core.perm_scope as perm_scope
-
-    # Wave 16 C-TOPIC: the ctx is TOPIC-SHAPED. The product scope swaps all three of the
-    # things a write is validated against — the module the permission wall reads, the canonical
-    # contract the hidden-field closure runs over, and the TABLE OPS the write lands in.
-    # Getting any one of them from the other topic is the "validated against the wrong field
-    # contract" near-miss the wave-15 routes_products header warned about.
-    scope_key = str(kw.get("scope_key") or "")
-    if scope_key == "product":
-        import modules.product_data as pd
-        from routes_products import MODULE as _PMOD, pd_fields
-
-        module, canonical = _PMOD, pd_fields(consolidated=True)
-        kw.setdefault("table", pd.TABLE_OPS)
-        hidden = perm_scope.hidden_keys(session.user, module, canonical)
-    elif scope_key.startswith("ut_"):
-        # Wave 18 C3-UT: a user table has no module in the permission wall (its wall is
-        # `user_tables.may_open`, already applied by the assembly this route ran first), so
-        # the hidden-field closure is EMPTY rather than borrowed from another topic's contract.
-        import core.table_store as table_store
-
-        kw.setdefault("table",
-                      table_store.make(f"{scope_key}_table_workspace", st=session.runtime))
-        hidden = frozenset()
-    else:
-        module, canonical = MODULE, aios_grid.FIELDS
-        hidden = perm_scope.hidden_keys(session.user, module, canonical)
-    return grid_events.EventCtx(
-        uname=session.uname, allowed_pids=pids, fields=fields, admin=session.admin,
-        # C-PERM: the write wall's field half, on the EVENTS transport too. This route takes the
-        # component's event objects verbatim, so a hidden key would otherwise arrive here with
-        # nothing between it and the store.
-        hidden_keys=hidden,
-        # ⭐ WAVE 25 (R6b, closes D-16) — THE TENANT HANDLE, on EVERY scope. This is the line
-        # D-16's exit condition names, and without it the rest of the fix is inert: the handler
-        # falls back to the module-global `core.store`, which is tenant #0's repo, so a Nurilab
-        # user's documents, cohorts and — through `_tops` — their whole customer/product table
-        # workspace were written into Royal Imports' dataset. Note it is set for the CUSTOMER
-        # and PRODUCT branches too, not only `ut_`: those two never passed a scoped `table`, so
-        # they were the ones actually resolving to tenant #0 on every request.
-        st=session.runtime,
-        fallback_ws=None, seen_ids={}, **kw)
-
-
-#: The surfaces the ONE grid serves. `cohort` is the same table over hand-curated SETS rather
-#: than over the whole scoped pool — `app.py:page_cohort` calls the same `_table_grid` with
-#: `cohort_mode=True, scope_key='cohort'`. `product` (wave 16 C-TOPIC) is the SKU table:
-#: same engine, its own pool, field contract and workspace BUCKET (routes_products).
-_SCOPES = ("customer", "cohort", "product")
-
-
-def _scope_or_400(raw):
-    """⛔ REFUSE AN UNKNOWN SCOPE, never default it. A typo silently served as `customer` would
-    hand a user the whole book on a page they opened to see one cohort — the widening direction,
-    which is the one that must fail closed.
-
-    Wave 18 (C3-UT): a `ut_`-prefixed scope names a USER TABLE and passes through here —
-    existence and the per-table wall are enforced by `routes_tables.ut_assembly` (404/403),
-    which every consumer of such a scope goes through. Passing an unknown ut key therefore
-    still fails closed, just one layer down where the store can actually be consulted."""
-    scope = (raw or "customer").strip().lower()
-    if scope.startswith("ut_"):
-        return scope
-    if scope not in _SCOPES:
-        raise err(400, "bad_scope",
-                  f"scope must be one of {', '.join(_SCOPES)} — refusing to guess")
-    return scope
-
-
-@router.get("/workspace")
-def workspace(scope: str = "customer",
-              session: Session = Depends(require_session)):
-    """The durable table workspace for this session: views, fields, overlays, folders.
-
-    ⛔ WAVE 21 (C4): the wall is TOPIC-SHAPED, so the DEPENDENCY is session-only and each scope
-    asserts its own gate below. The old `module_gate("customer_data")` dependency 403'd a
-    `ut_*` workspace for any tenant whose catalogue omits the customer module (loopable/
-    nurilab/gtmlab ship modules w/o it) — and the client then fell back to the shared
-    localStorage bucket + demo data, which is exactly the "new database shows RI's customer
-    fields" defect. A user table's real wall is `user_tables.may_open`, enforced inside
-    `ut_assembly` (404/403, one layer down where the store can be consulted).
-
-    ⚠ WHY THIS EXISTS (S1↔S2, 2026-07-30 — an X2 AMENDMENT, see the split doc). X2 fixes
-    `/customers` at the exact shape `verify_fields_contract.py` referees, so the workspace cannot
-    ride along in it without forking the thing that gate keeps single-sourced. Without a
-    workspace route the standalone shell's write path would be WRITE-ONLY: events persist
-    server-side and nothing reads them back on reload, so a saved view looks lost to the user
-    even though the store has it. This route closes that read-back gap at its own URL.
-
-    `allowed_pids` is passed so a SHARED view's `memberPids` are re-scoped to THIS reader — the
-    wave-9 leak rule. Omitting it would hand a Fisch-scoped user a member list built by a
-    full-access user.
-    """
-    from core import grid_events
-    from routes_customers import grid_assembly
-
-    scope = _scope_or_400(scope)
-    # Wave 21 C4 — the per-scope gate the dependency no longer asserts: customer/cohort need
-    # the customer module; product asserts PRODUCT_MODULE in its branch; ut_* needs only a
-    # session (its wall is the table's own, inside the assembly).
-    if not (scope == "product" or scope.startswith("ut_")):
-        session.require(MODULE)
-    # ── Wave 18 C3-UT: a USER TABLE'S workspace — the third topic through the one wire. The
-    # storage key has no (bu, agent) because a user table has no Odoo scope; per-user by the
-    # table's own wall (creator or admin, `user_tables.may_open`).
-    if scope.startswith("ut_"):
-        from routes_tables import ut_assembly
-
-        storage_key = f"{session.tenant}:{scope}:{session.uname}"
-        try:
-            # ⭐⭐ W31-T20 / D-174 — `with_rows=False`. THIS ROUTE RENDERS NO ROW: the grid fetches
-            # them from `/tables/{key}/rows` or `/odoo-tables/{key}/rows` in the same paint, and
-            # `useCustomerData.load` fires both calls in one `Promise.all`. Building the table here
-            # bought nothing and cost everything — it is why `?scope=ut_odoo_gl_lines` answered
-            # `409 window_required` six times out of six on the live deploy while the ROWS door for
-            # the same table served a page in ~150 ms, i.e. a grid that could not be opened at all.
-            g = ut_assembly(session, scope, storage_key=storage_key, with_rows=False)
-        except grid_events.StoreUnavailable:
-            raise err(503, "store_unavailable", "the tenant store is unavailable")
-        workspace = g["workspace"]
-        workspace["overlays"] = g["ws"].get("overlays") or {}
-        # The field contract rides the cheap re-read on EVERY topic — see the customer branch
-        # below for why (a user table's first cohort changes its contract too: `_cohorts(ctx)`
-        # is scope-parameterized, so `ut_` surfaces have their own sets).
-        workspace["fields"] = g["fields"]
-        workspace["measures"] = g["measures"]
-        workspace["measureSets"] = g["measure_sets"]
-        # ⚠ `g["derived"]`, NOT `{}` (2026-08-04). R9 gave every topic its own cohort sets, and
-        # `ut_assembly` has been building this topic's cohort CELLS since — but this route threw
-        # them away, so the Locked-views column arrived on the cheap re-read with permanently
-        # empty values. A column that exists and can never have one is worse than an absent
-        # column: it reads as "this record is in no locked view", which is a claim.
-        workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()}
-        workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
-        try:
-            from core import users as _users
-            workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
-            workspace["userAvatars"] = {k: v for k, v in
-                                        _users.avatar_map(tenant=session.tenant).items()
-                                        if str(v).startswith("data:image/")}
-        except Exception:
-            workspace["userOptions"] = []
-            workspace["userAvatars"] = {}
-        workspace["scopeKey"] = scope
-        # ⭐ W31-T20 — R6's SECOND SENTENCE REACHES THE BROWSER. On a read-through grid too large
-        # for one window the pid set is empty, so cohort membership and a shared view's
-        # `memberPids` are UNRESOLVED here (both fail closed, and `workspace_wire` already stamps
-        # `missing` on a shortened cohort). An unannounced empty scope is the silent limit R6
-        # forbids; this is the announcement, and W31-T22 is the ticket that renders it.
-        workspace["limits"] = g.get("limits") or []
-        return {"workspace": workspace}
-
-    # ── Wave 16 C-TOPIC: the PRODUCT surface has its own assembly (own pool, own field
-    # contract, own workspace BUCKET) and the contract's own storage key. It branches before
-    # the customer derivation because the customer storage key embeds (bu, agent) while the
-    # product one is deliberately `all:all` — one product workspace per user (the wave doc's
-    # C-TOPIC line), since BU narrows the product VALUES, not which workspace you own.
-    if scope == "product":
-        from routes_products import MODULE as PRODUCT_MODULE, product_assembly
-
-        # ⛔ THE GATE CHANGES WITH THE SCOPE (wave 21 C4: the dependency is session-only now,
-        # so this line IS the product wall — not a second one on top of customer_data).
-        session.require(PRODUCT_MODULE)
-        storage_key = f"{session.tenant}:product-list:{session.uname}:all:all"
-        try:
-            g = product_assembly(session, scope=scope, storage_key=storage_key)
-        except grid_events.StoreUnavailable:
-            raise err(503, "store_unavailable", "the tenant store is unavailable")
-        workspace = g["workspace"]
-        workspace["overlays"] = g["ws"].get("overlays") or {}
-        # R9 made cohorts per-topic, so the PRODUCT surface has its own sets and its own
-        # first-cohort contract change. Same reason as the customer branch below.
-        workspace["fields"] = g["fields"]
-        # The measure channel is EMPTY on this topic (customer-grain descope) — stated
-        # explicitly so the client's pickers grey rather than guess.
-        workspace["measures"] = g["measures"]
-        workspace["measureSets"] = g["measure_sets"]
-        # `g["derived"]` — the same correction as the user-table branch above, for the same
-        # reason: `product_assembly` builds this topic's cohort cells and this route discarded
-        # them. The measure half stays empty on this topic by descope, which is a different
-        # statement and one the offer already makes.
-        workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()}
-        workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
-        try:
-            from core import users as _users
-            workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
-            workspace["userAvatars"] = {k: v for k, v in
-                                        _users.avatar_map(tenant=session.tenant).items()
-                                        if str(v).startswith("data:image/")}
-        except Exception:
-            workspace["userOptions"] = []
-            workspace["userAvatars"] = {}
-        workspace["scopeKey"] = scope
-        return {"workspace": workspace}
-
-    # The storage key mirrors the host's own convention byte-for-byte (app.py's `_table_grid`
-    # call sites) with the session's TENANT in the host's hardcoded slot, so localStorage state
-    # carries across embed ⇄ standalone on the same browser.
-    # Wave 15 R1 — the SAME derivation the pool uses (`routes_customers._team_agent`), so the
-    # storage key cannot drift from the scope it names. Value-identical for every migrated
-    # record (verify_perm_scope section D proves the derivation reproduces the legacy scope
-    # exactly), so nobody's saved localStorage state moves when the migration runs.
-    import core.perm_scope as perm_scope
-    team_id, agent = perm_scope.derive_pool_scope(session.user, MODULE)
-    bu = team_id if team_id is not None else "all"
-    if scope == "cohort":
-        storage_key = f"{session.tenant}:cohort:{session.uname}:{bu}"
-    else:
-        storage_key = f"{session.tenant}:customer-list:{session.uname}:{bu}:{agent or 'all'}"
-
-    # ⛔ THE WIRE SHAPE IS THE SHARED PROJECTION (`aios_grid.workspace_wire`), not the store
-    # shape. The first version returned the store dict with no `storageKey` — and the client
-    # validator requires one, so the standalone shell DISCARDED the whole workspace: saved views
-    # never rendered, `cohortMode` never arrived, and the Cohort route silently drew the Customer
-    # surface. One projection for both servers is the fix that cannot drift.
-    try:
-        g = grid_assembly(session, scope=scope, storage_key=storage_key)
-    except grid_events.StoreUnavailable:
-        raise err(503, "store_unavailable", "the tenant store is unavailable")
-    workspace = g["workspace"]
-    # ADDITIVE to the wire: the caller's own overlay cells. The grid reads overlays from the
-    # /customers rows, but this route is the cheap read-back a write can be verified against
-    # (verify_api's overlay probe) without paying the pool call. Per-user by construction —
-    # `table_workspace` is this session's workspace.
-    workspace["overlays"] = g["ws"].get("overlays") or {}
-    # ⭐ THE FIELD CONTRACT, 2026-08-04 — and it is NOT decoration.
-    #
-    # `fields_from_workspace(ws, cohorts=bool(cohort_lists))` appends the derived "Locked
-    # views" column ONLY when the caller owns at least one cohort. So a user's FIRST cohort
-    # CHANGES THE FIELD CONTRACT — and the rows call that used to be the only carrier of
-    # `fields` is deliberately never re-fetched on a write (it is a 15-minute-cached Odoo
-    # pull; this route is the cheap re-read). The client therefore had no way to learn about
-    # that column short of a remount, which is the second half of the owner's "it only shows
-    # up when I switch modules and come back".
-    #
-    # ⚠ `g["fields"]` is the SAME list `_payload` serves on `/customers` — same assembly,
-    # same permission wall (`hidden_keys` already applied) — so the two wires cannot disagree
-    # about what a column is. Taking it from anywhere else would fork the contract that
-    # verify_fields_contract.py exists to keep single-sourced.
-    workspace["fields"] = g["fields"]
-    # ── the STANDALONE measure channel (owner item 1, 2026-07-31) ────────────────────────────
-    # The embed receives these as top-level render args; standalone lifts them off THIS route
-    # (useCustomerData merges them into the payload slots the grid already reads). They ride
-    # the workspace rather than /customers because a durable write re-reads exactly this route
-    # (WORKSPACE_STALE), so a new measure column populates without refetching the heavy pool.
-    workspace["measures"] = g["measures"]
-    workspace["measureSets"] = g["measure_sets"]
-    # Derived cells (cohort column + measure columns), keyed by pid. JSON object keys are
-    # strings; the client indexes with String(pid).
-    workspace["derived"] = {str(pid): cells for pid, cells in g["derived"].items()}
-    # Who is looking (permissions verdicts) + the assignee choices for `user`-typed columns —
-    # the two other host-only render args the shell was missing (fail-closed without them:
-    # restricted fields uneditable, assignee picker empty).
-    workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
-    try:
-        from core import users as _users
-        workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
-        # Wave 14 C-AVATAR — the options vocabulary's companion: display name -> data URL.
-        # Absent entries fall back to the client's initials disc; a non-data: value is
-        # dropped (defence in depth beside the write-side wall in routes_auth).
-        workspace["userAvatars"] = {k: v for k, v in
-                                    _users.avatar_map(tenant=session.tenant).items()
-                                    if str(v).startswith("data:image/")}
-    except Exception:
-        workspace["userOptions"] = []
-        workspace["userAvatars"] = {}
-
-    # THE SURFACE STAMP — a MIRROR of the host's own (`_table_grid`'s hide/cohort stamps).
-    # ⚠ Deliberately NOT `hideViews`: `cohortMode` is what swaps the Views panel for the cohort
-    # list, and two switches for one behaviour would drift.
-    workspace["scopeKey"] = scope
-    if scope == "cohort":
-        workspace["cohortMode"] = True
-        # The create pane offers "this cohort table only" (default) vs "all customer tables".
-        workspace["scopeChoice"] = True
-    return {"workspace": workspace}
-
-
-# ─────────────────────────────────────────────── the TIME-SERIES channel (C-TS, 2026-08-02)
-_TS_BUCKETS = ("week", "month", "quarter", "year")
-_TS_MAX_BUCKETS = 120
-_TS_MAX_FIELDS = 12
-_TS_MAX_PIDS = 5000
-_TS_MAX_LAST_N = 120
-#: C-TSWIN (wave 14): sliding windows cost ONE aggregate query per (metric, bucket) — the
-#: price of "bucket N == the grid cell at bucket N's end". The product cap keeps a legal
-#: 120-bucket × 12-metric ask from becoming 1,440 queries in one request; typical panels
-#: (12 × 3) sit two orders of magnitude under it. Dated amendment in the split doc.
-_TS_MAX_CELLS = 720
-
-
-def _ts_start_of(bucket, d):
-    """The calendar START of the bucket holding `d` (week = Monday, the vocabulary rule)."""
-    if bucket == "week":
-        return d - dt.timedelta(days=d.weekday())
-    if bucket == "month":
-        return d.replace(day=1)
-    if bucket == "quarter":
-        return d.replace(month=((d.month - 1) // 3) * 3 + 1, day=1)
-    return d.replace(month=1, day=1)
-
-
-def _ts_next(bucket, d):
-    if bucket == "week":
-        return d + dt.timedelta(days=7)
-    if bucket == "year":
-        return d.replace(year=d.year + 1)
-    step = 1 if bucket == "month" else 3
-    m = d.month + step
-    return dt.date(d.year + (m - 1) // 12, (m - 1) % 12 + 1, 1)
-
-
-def _ts_prev(bucket, d):
-    if bucket == "week":
-        return d - dt.timedelta(days=7)
-    if bucket == "year":
-        return d.replace(year=d.year - 1)
-    step = 1 if bucket == "month" else 3
-    y, m = d.year, d.month - step
-    while m < 1:
-        m += 12
-        y -= 1
-    return dt.date(y, m, 1)
-
-
-def _ts_label(bucket, start):
-    if bucket == "week":
-        return f"Wk of {start.strftime('%b %d')}"
-    if bucket == "month":
-        return start.strftime("%b %Y")
-    if bucket == "quarter":
-        return f"Q{(start.month - 1) // 3 + 1} {start.year}"
-    return str(start.year)
-
-
-@router.post("/grid/timeseries")
-def grid_timeseries(body: dict = Body(default=None), scope: str = "customer",
-                    session: Session = Depends(module_gate(MODULE))):
-    """Pooled measure values per calendar bucket — the time-series view's data channel.
-
-    C-TSWIN (wave 14, ruling R1): AS-OF semantics. Each metric's OWN stored window is
-    re-resolved per bucket with `today := min(bucket end, real today)` — `ytd` is cumulative
-    from Jan 1, `last_90_days` trailing, `all_time` cumulative ever; bucket N equals what the
-    grid's measure column would show if today were bucket N's end. Fixed-range (`custom`)
-    windows cannot slide and are dropped per field as `window_fixed`. Rows carry
-    `window: {kind, label}` so a cumulative row cannot be misread as periodic, and there is
-    NO `total` column — sliding windows overlap, so a sum of columns would double-count.
-
-    The client sends the pids its view currently matches (the filter IS the scope); the server
-    intersects them with the session's book, so the request can only ever NARROW. Values are
-    POOLED aggregates computed by the semantic layer's own expression (the additivity law: an
-    average is computed at pool grain, never averaged over per-customer answers). A bucket
-    with no rows is 0 for a sum/count and null for anything else; a bucket that has not
-    STARTED yet is null for every kind — an unstarted period's YTD is unanswered, not 0.
-    """
-    from core import measure_resolve
-    from harness import windows as _wn
-    from routes_customers import _pool_stamp, _team_agent, allowed_pids
-
-    # Wave 16 C-TOPIC: the channel is CUSTOMER-GRAIN (measure_resolve's own grain), so the
-    # product surface is refused in words rather than answered with an id-space accident —
-    # product pids are CRC32 hashes and would mostly fall outside the customer book anyway,
-    # but "mostly" is not a wall. The client hides the mode for this topic; this is the
-    # server's half of the same refusal.
-    _sc = _scope_or_400(scope)
-    if _sc == "product" or _sc.startswith("ut_"):
-        raise err(400, "bad_timeseries",
-                  "this surface has no time-series channel — measures are customer-grain")
-
-    body = body or {}
-    bucket = body.get("bucket")
-    if bucket not in _TS_BUCKETS:
-        raise err(400, "bad_timeseries", "bucket must be one of week, month, quarter, year")
-    raw_fields = body.get("fields")
-    if not isinstance(raw_fields, list) or not raw_fields:
-        raise err(400, "bad_timeseries", "fields must be a non-empty list of field keys")
-    if len(raw_fields) > _TS_MAX_FIELDS:
-        raise err(400, "bad_timeseries", f"at most {_TS_MAX_FIELDS} fields per request")
-    raw_pids = body.get("pids")
-    if not isinstance(raw_pids, list) or not raw_pids:
-        raise err(400, "bad_timeseries", "pids must be a non-empty list")
-    if len(raw_pids) > _TS_MAX_PIDS:
-        raise err(400, "bad_timeseries", f"at most {_TS_MAX_PIDS} pids per request")
-    try:
-        wanted = {int(p) for p in raw_pids}
-    except (TypeError, ValueError):
-        raise err(400, "bad_timeseries", "pids must be integers")
-    pool = wanted & {int(p) for p in allowed_pids(session)}
-    if not pool:
-        # 403, not an empty 200: the caller asked about customers outside their book, and an
-        # all-zero series over nobody would read as "no activity", not "not yours".
-        raise err(403, "out_of_scope", "none of those customers are in your book")
-
-    span = body.get("span")
-    if not isinstance(span, dict):
-        raise err(400, "bad_timeseries", "span must be {'lastN': n} or {'from': .., 'to': ..}")
-    today = time.strftime("%Y-%m-%d")
-    t = dt.date.fromisoformat(today)
-    n = span.get("lastN")
-    starts = []
-    if n is not None:
-        if not isinstance(n, int) or isinstance(n, bool) or not (1 <= n <= _TS_MAX_LAST_N):
-            raise err(400, "bad_timeseries", f"lastN must be 1..{_TS_MAX_LAST_N}")
-        cur = _ts_start_of(bucket, t)
-        starts = [cur]
-        for _ in range(n - 1):
-            cur = _ts_prev(bucket, cur)
-            starts.append(cur)
-        starts.reverse()
-    else:
-        try:
-            d_from = dt.date.fromisoformat(str(span.get("from")))
-            d_to = dt.date.fromisoformat(str(span.get("to")))
-        except (TypeError, ValueError):
-            raise err(400, "bad_timeseries", "span.from/to must be ISO dates (YYYY-MM-DD)")
-        if d_from > d_to:
-            d_from, d_to = d_to, d_from
-        cur = _ts_start_of(bucket, d_from)
-        while cur <= d_to:
-            starts.append(cur)
-            if len(starts) > _TS_MAX_BUCKETS:
-                raise err(400, "bad_timeseries",
-                          f"that span is more than {_TS_MAX_BUCKETS} {bucket} buckets - "
-                          f"narrow it")
-            cur = _ts_next(bucket, cur)
-        if not starts:
-            raise err(400, "bad_timeseries", "the span holds no buckets")
-    ends = [_ts_next(bucket, s) - dt.timedelta(days=1) for s in starts]
-
-    rt = session.runtime
-    if not rt.available():
-        raise err(503, "store_unavailable", "the tenant store is unavailable")
-    import modules.customer_data as cl_mod
-    # Read-only route: it must not consume a one-shot label-correction ack the browser has
-    # not seen yet (the same rule event validation follows).
-    ws = cl_mod.table_workspace(session.uname, consume_corrections=False)
-    fdefs = ws.get("fields") or {}
-    keys, dropped, seen = [], [], set()
-    for k in raw_fields:
-        k = str(k or "")
-        if not k or k in seen:
-            continue
-        seen.add(k)
-        fd = fdefs.get(k)
-        if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict):
-            # Named, never silent: v1 eligibility is measure-backed fields only (C-TS).
-            dropped.append({"field": k, "reason": "not_a_measure_field"})
-            continue
-        keys.append(k)
-    # ⚡ WAVE 17 R9 (amendment 2026-08-03, GRID's cross-fence ask) — A SHEET WITH NO MEASURE
-    # FIELDS IS NO LONGER A 400. It answers with the BUCKET GRID and no rows.
-    #
-    # Why this is the honest direction and not a loosening: the bucket starts and their labels
-    # are SERVER math (`_ts_start_of` / `_ts_next` / `_ts_label`), and R9 has the client
-    # synthesizing snapshot rows for preset columns that have no window to slide. Those rows
-    # must be painted under the SAME headings as everything else, so the client needs the
-    # columns even when the server has no series to put in them. Refusing the whole request
-    # meant the panel could offer nothing at all on this tenant — the shipped contract has zero
-    # measure-backed presets (`aios_grid.py`: "this branch is currently MEMBERLESS"), which is
-    # what item 5 is actually about.
-    #
-    # Nothing is invented by this: `rows` is empty and `meta.dropped` NAMES every field and why.
-    # A 400 is still returned for a request that is malformed (bad bucket, bad span, no fields
-    # at all) — this only stops treating "I asked about columns you cannot serve" as an error.
-    mfields, slid_keys = [], []
-    for k in keys:
-        m = dict(fdefs[k]["measure"])
-        if (m.get("window") or {}).get("kind") == "custom":
-            # C-TSWIN: a fixed date range cannot slide across buckets — named, never silent,
-            # the same posture as not_a_measure_field.
-            dropped.append({"field": k, "reason": "window_fixed"})
-            continue
-        mfields.append({"key": k, "measure": m})
-        slid_keys.append(k)
-    # Same amendment as above: every metric being fixed-range is a sheet with no SERIES, not a
-    # broken request. The columns still stand, and `window_fixed` still names each refusal.
-    if len(starts) * len(mfields) > _TS_MAX_CELLS:
-        raise err(400, "bad_timeseries",
-                  "that ask is too wide - narrow the span or pick fewer metrics")
-
-    team_id, agent = _team_agent(session)
-    stamp = _pool_stamp(rt, team_id, agent)
-    problems = []
-    bucket_bounds = [(s.isoformat(), e.isoformat()) for s, e in zip(starts, ends)]
-    # No slidable metrics = nothing to resolve. Skipping the call rather than asking the
-    # resolver about an empty list keeps the memo free of a meaningless key.
-    answers = measure_resolve.series_values(
-        mfields, bucket, bucket_bounds,
-        team_id, frozenset(pool), today, stamp, rt.series_memo,
-        on_error=lambda tag, e: problems.append(str(e)[:200])) if mfields else {}
-
-    columns = []
-    for s, e in zip(starts, ends):
-        col = {"key": s.isoformat(), "label": _ts_label(bucket, s),
-               "from": s.isoformat(), "to": e.isoformat()}
-        if e > t:
-            col["partial"] = True
-        columns.append(col)
-    rows = []
-    for k in slid_keys:
-        ans = answers.get(k)
-        if ans is None:
-            dropped.append({"field": k, "reason": "unresolvable"})
-            continue
-        vals_by = ans.get("values") or {}
-        agg_kind = str(ans.get("agg") or "sum")
-        zero_fill = agg_kind in ("sum", "count")
-        vals = []
-        for s in starts:
-            if s > t:
-                # C-TSWIN: an unstarted period is unanswered for EVERY agg kind — a future
-                # month's YTD zero-filled to 0 would read as "the year reset".
-                vals.append(None)
-            else:
-                vals.append(vals_by.get(s.isoformat(), 0 if zero_fill else None))
-        wspec = (fdefs[k].get("measure") or {}).get("window")
-        wnorm = _wn.normalize(wspec) or {}
-        rows.append({"field": k, "label": str(fdefs[k].get("label") or k)[:120],
-                     "agg": agg_kind, "values": vals,
-                     "window": {"kind": str(wnorm.get("kind") or ""),
-                                "label": _wn.label(wspec)}})
-    meta = {"pool": len(pool), "today": today, "bucket": bucket}
-    if dropped:
-        meta["dropped"] = dropped
-    if problems:
-        meta["problems"] = problems[:5]
-    return {"columns": columns, "rows": rows, "meta": meta}
-
-
-#: C-CAL caps (wave 17, owner item 9 / ruling R4). A month of days, a handful of metrics.
-#: Every one of these is a 400 WITH ITS REASON, never a silent trim: a calendar that quietly
-#: answered 20 of the 31 days asked about would paint eleven blank cells that look like days
-#: with no activity.
-_CAL_MAX_GROUPS = 31
-_CAL_MAX_FIELDS = 6
-_CAL_MAX_CELLS = 186
-
-
-@router.post("/grid/calendar_metrics")
-def grid_calendar_metrics(body: dict = Body(default=None), scope: str = "customer",
-                          session: Session = Depends(module_gate(MODULE))):
-    """Per-DAY measure values with each metric's own window slid to that day (C-CAL / R4).
-
-    ⛔ WHY THIS EXISTS AT ALL. The calendar's summary cells used to aggregate the row VALUES the
-    grid already held — which for a measure column means "every member's YTD **as of today**",
-    summed and printed under a date in March. The number was arithmetically fine and semantically
-    a lie: it answered a question about today while sitting in a cell labelled with another day.
-    R4: a metric in a day cell is computed AS OF THAT DAY.
-
-    The shape differs from the time-series channel in exactly one way, and it is the reason this
-    is a separate route rather than a parameter: **every group carries its OWN pid set**. A
-    calendar day holds the records the date field placed there, so day-to-day the subject
-    changes. One `allowed_pids` for the whole request — the TS channel's shape — would compute
-    each day over everybody, which is a different question again.
-
-    Static (non-measure) fields are NOT served here. They have no window to slide, so the
-    client's own per-day aggregation over row values stays correct for them; sending them would
-    invite a second implementation of arithmetic that already works.
-    """
-    from core import measure_resolve
-    from harness import windows as _wn
-    from routes_customers import _pool_stamp, _team_agent, allowed_pids
-
-    # The same refusal the TS channel makes, for the same reason: measures are customer-grain
-    # and product pids are CRC32 hashes of SKU codes. "Mostly outside the book" is not a wall.
-    _sc_cal = _scope_or_400(scope)
-    if _sc_cal == "product" or _sc_cal.startswith("ut_"):
-        raise err(400, "bad_calendar_metrics",
-                  "the product surface has no measure channel yet — measures are customer-grain")
-
-    body = body or {}
-    raw_groups = body.get("groups")
-    if not isinstance(raw_groups, list) or not raw_groups:
-        raise err(400, "bad_calendar_metrics",
-                  "groups must be a non-empty list of {key: 'YYYY-MM-DD', pids: [...]}")
-    if len(raw_groups) > _CAL_MAX_GROUPS:
-        raise err(400, "bad_calendar_metrics",
-                  f"at most {_CAL_MAX_GROUPS} days per request (one month)")
-    raw_fields = body.get("fields")
-    if not isinstance(raw_fields, list) or not raw_fields:
-        raise err(400, "bad_calendar_metrics", "fields must be a non-empty list of field keys")
-    if len(raw_fields) > _CAL_MAX_FIELDS:
-        raise err(400, "bad_calendar_metrics", f"at most {_CAL_MAX_FIELDS} metrics per request")
-    if len(raw_groups) * len(raw_fields) > _CAL_MAX_CELLS:
-        raise err(400, "bad_calendar_metrics",
-                  "that ask is too wide - fewer days or fewer metrics")
-
-    book = {int(p) for p in allowed_pids(session)}
-    groups, seen_days = [], set()
-    for g in raw_groups:
-        if not isinstance(g, dict):
-            raise err(400, "bad_calendar_metrics", "every group must be an object")
-        day = str(g.get("key") or "")
-        try:
-            d = dt.date.fromisoformat(day)
-        except (TypeError, ValueError):
-            raise err(400, "bad_calendar_metrics",
-                      "every group key must be an ISO date (YYYY-MM-DD)")
-        if day in seen_days:
-            raise err(400, "bad_calendar_metrics", f"day {day} appears twice")
-        seen_days.add(day)
-        raw_pids = g.get("pids")
-        if not isinstance(raw_pids, list):
-            raise err(400, "bad_calendar_metrics", "every group needs a pids list")
-        try:
-            wanted = {int(p) for p in raw_pids}
-        except (TypeError, ValueError):
-            raise err(400, "bad_calendar_metrics", "pids must be integers")
-        # NARROW-ONLY, per group. The client sends what its calendar placed; the server can
-        # only ever remove from that, never add.
-        groups.append((day, d, frozenset(wanted & book)))
-    if sum(len(p) for _, _, p in groups) > _TS_MAX_PIDS:
-        raise err(400, "bad_calendar_metrics",
-                  f"at most {_TS_MAX_PIDS} customer references per request")
-
-    rt = session.runtime
-    if not rt.available():
-        raise err(503, "store_unavailable", "the tenant store is unavailable")
-    import modules.customer_data as cl_mod
-    ws = cl_mod.table_workspace(session.uname, consume_corrections=False)
-    fdefs = ws.get("fields") or {}
-    keys, dropped, seen = [], [], set()
-    for k in raw_fields:
-        k = str(k or "")
-        if not k or k in seen:
-            continue
-        seen.add(k)
-        fd = fdefs.get(k)
-        if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict):
-            # The TS channel's vocabulary, deliberately reused rather than re-coined: the client
-            # already knows how to say these three words to a reader.
-            dropped.append({"field": k, "reason": "not_a_measure_field"})
-            continue
-        if ((fd["measure"].get("window") or {}).get("kind") == "custom"):
-            dropped.append({"field": k, "reason": "window_fixed"})
-            continue
-        keys.append(k)
-    if not keys:
-        raise err(400, "bad_calendar_metrics",
-                  "none of the requested fields are measure fields with a window that can "
-                  "slide to a day")
-
-    today = time.strftime("%Y-%m-%d")
-    t = dt.date.fromisoformat(today)
-    team_id, agent = _team_agent(session)
-    stamp = _pool_stamp(rt, team_id, agent)
-    problems = []
-    values = {k: {} for k in keys}
-    for day, d, pids in groups:
-        # A day that has not happened is unanswered for EVERY aggregate kind — the unstarted
-        # bucket law at day grain. Answering 0 would say "we sold nothing", which is a claim
-        # about a day nobody has lived through yet.
-        if d > t or not pids:
-            for k in keys:
-                values[k][day] = None
-            continue
-        answers = measure_resolve.series_values(
-            [{"key": k, "measure": dict(fdefs[k]["measure"])} for k in keys],
-            "day", [(day, day)], team_id, pids, today, stamp, rt.series_memo,
-            on_error=lambda tag, e: problems.append(str(e)[:200]))
-        for k in keys:
-            ans = answers.get(k)
-            if ans is None:
-                values[k][day] = None
-                continue
-            vals_by = ans.get("values") or {}
-            zero_fill = str(ans.get("agg") or "sum") in ("sum", "count")
-            values[k][day] = vals_by.get(day, 0 if zero_fill else None)
-
-    for k in keys:
-        if all(v is None for v in values[k].values()):
-            # Every day unanswerable is a FIELD-level failure, and saying so is the difference
-            # between "no activity that month" and "this metric could not be computed".
-            dropped.append({"field": k, "reason": "unresolvable"})
-    out = {"values": values, "today": today,
-           "windows": {k: {"kind": str((_wn.normalize(
-               (fdefs[k].get("measure") or {}).get("window")) or {}).get("kind") or ""),
-               "label": _wn.label((fdefs[k].get("measure") or {}).get("window"))}
-               for k in keys}}
-    if dropped:
-        out["dropped"] = dropped
-    if problems:
-        out["problems"] = problems[:5]
-    return out
-
-
-#: ⭐⭐ THE BULK CELL DOOR'S OWN CEILING, and it is REPORTED rather than silently enforced. The
-#: standing no-cap rule governs data read from a connected source; this is a WRITE of team-typed
-#: values, so a bound is right — but R6's second sentence still binds, which is why exceeding it
-#: is a 400 naming the number and the count, never a truncation.
-#: Sized against the real job with headroom: the 2027 catalog is 1,397 rows.
-MAX_BULK_ROWS = 10_000
-
-
-@router.post("/grid/bulk-cells")
-def bulk_cells(body: dict = Body(default=None),
-               session: Session = Depends(require_session)):
-    """`{scopeKey, rows: {"<pid>": {key: value}}}` → shared cells on many rows, ONE transaction.
-
-    ⭐⭐ WHY THIS EXISTS AT ALL, because "we already have `/grid/events`" is the obvious objection.
-    That door is capped at `_MAX_EVENTS = 24` (it is sized for the client's resend window), so a
-    1,397-row import is ~59 sequential POSTs against ONE JSON document — and this repo has a
-    MEASURED scar for exactly that: 18 writes against one document under the store's coalescing
-    single-flight landed **zero** while answering 200 eighteen times. Raising `_MAX_EVENTS` would
-    have widened the browser's own resend window to fix a script's problem. One door, one
-    transaction, is the honest shape.
-
-    ⛔ TENANT-WIDE KEYS ONLY, AND THAT IS THE POINT RATHER THAN A LIMITATION. It writes through
-    `shared_overlay`, which every account in the tenant reads. A per-user bulk write would be a
-    contradiction: nobody imports 1,397 rows of the team's work so that one account can see it —
-    that is precisely the failure owner ruling R6 was made to avoid. A key that is not declared
-    `shared: true` in the canonical contract is REFUSED BY NAME.
-
-    ⛔ ADMIN ONLY. This mutates data every account in the tenant reads, in bulk, in one call.
-    D-172 already records that the shared stratum has an asymmetric wall (anyone may create a
-    tenant-wide column, only the creator or an admin may delete one); a new door does not get to
-    inherit the loose half of an asymmetry somebody already flagged.
-    """
-    import core.shared_overlay as shared_overlay
-    import modules.product_data as pd
-    import core.perm_scope as perm_scope
-    from routes_products import MODULE as PRODUCT_MODULE, scoped_pool
-
-    scope = _scope_or_400((body or {}).get("scopeKey"))
-    if scope != "product":
-        # Fail-closed with the reason: the shared stratum is a PRODUCT-topic mechanism today.
-        raise err(400, "scope_not_bulk_writable",
-                  f"bulk cell writes are available on the product database; '{scope}' has no "
-                  f"tenant-wide cell stratum")
-    session.require(PRODUCT_MODULE)
-    if not session.admin:
-        raise err(403, "admin_only",
-                  "a bulk write changes values every account in this workspace reads")
-
-    rows = (body or {}).get("rows")
-    if not isinstance(rows, dict):
-        raise err(400, "bad_rows", "expected {rows: {\"<pid>\": {field: value}}}")
-    if len(rows) > MAX_BULK_ROWS:
-        raise err(400, "too_many_rows",
-                  f"at most {MAX_BULK_ROWS} rows per request; this one carried {len(rows)}")
-
-    pids, _team, _src, fields_base = scoped_pool(session)
-    allowed = {int(p) for p in pids}
-    shared_keys = set(pd.SHARED_KEYS())
-    hidden = perm_scope.hidden_keys(session.user, PRODUCT_MODULE, fields_base)
-
-    clean, unknown_pid, refused_keys = {}, [], {}
-    for raw_pid, values in rows.items():
-        try:
-            pid = int(raw_pid)
-        except (TypeError, ValueError):
-            unknown_pid.append(str(raw_pid)[:40])
-            continue
-        # ⚠ THE POOL IS THE WALL. `scoped_pool` is the same predicate the read door uses, so a
-        # caller cannot write a row they could not see — including a row in another BU.
-        if pid not in allowed:
-            unknown_pid.append(str(raw_pid)[:40])
-            continue
-        keep = {}
-        for key, value in dict(values or {}).items():
-            key = str(key)
-            if key not in shared_keys:
-                refused_keys.setdefault("not_shared", set()).add(key)
-                continue
-            if key in hidden:
-                refused_keys.setdefault("hidden_by_permissions", set()).add(key)
-                continue
-            if not isinstance(value, (str, int, float)) or isinstance(value, bool):
-                refused_keys.setdefault("unsupported_value", set()).add(key)
-                continue
-            keep[key] = value
-        if keep:
-            clean[pid] = keep
-
-    written = shared_overlay.put_rows(pd.TABLE_KEY, clean, st=pd.TABLE_OPS.st) if clean else {}
-    out = {
-        "rows_written": len(written),
-        "cells_written": sum(len(v) for v in written.values()),
-        "rows_requested": len(rows),
-    }
-    # ⭐ R6's second sentence: what did NOT land, and why. A bulk door that reports only its
-    # successes is how 701 unmatched SKUs disappear quietly.
-    if unknown_pid:
-        out["rows_not_in_your_pool"] = {"count": len(unknown_pid),
-                                        "sample": sorted(unknown_pid)[:10]}
-    if refused_keys:
-        out["refused_fields"] = {reason: sorted(keys)
-                                 for reason, keys in refused_keys.items()}
-    return out
-
-
-@router.post("/grid/events")
-def grid_events_route(body: dict = Body(default=None),
-                      session: Session = Depends(require_session)):
-    """`{events: [<component event objects, verbatim>]}` → `{results, doc?, toast?}`.
-
-    Wave 21 C4: session-only dependency, per-scope gate below — the write door must admit the
-    same sessions the read door (`/workspace`) admits, or a tenant without the customer module
-    can SEE its own user tables and not write to them."""
-    from core import grid_events
-    from routes_customers import grid_assembly
-
-    events = (body or {}).get("events")
-    if events is None and isinstance(body, dict) and body.get("type"):
-        events = [body]                    # a single event object, the legacy shape
-    if not isinstance(events, list):
-        raise err(400, "bad_events", "expected {events: [...]}")
-    if len(events) > _MAX_EVENTS:
-        raise err(400, "too_many_events",
-                  f"at most {_MAX_EVENTS} events per request (the client's resend window)")
-
-    # Validated with the SAME predicate as the read route. An unrecognised scopeKey used to be
-    # passed through verbatim, and `core.grid_events` only ever compares it to 'cohort' — so a
-    # typo degraded silently to customer-scope behaviour on a WRITE. Read and write must agree on
-    # what a scope is, or the surface you read is not the surface you wrote.
-    scope = _scope_or_400((body or {}).get("scopeKey"))
-    # Wave 21 C4 — same per-scope gate as /workspace (read and write doors must agree).
-    if not (scope == "product" or scope.startswith("ut_")):
-        session.require(MODULE)
-    # This assembly validates the write; it is not a payload the browser will render. Leave
-    # one-shot field-name correction acks queued for the subsequent /workspace refresh.
-    # Wave 16 C-TOPIC: the PRODUCT topic gets the product assembly — product field contract,
-    # product pids, and (below) the product TABLE OPS, so a product event is validated against
-    # and lands in the product bucket. The measure/cohort context is honestly EMPTY there:
-    # `clean_measure_field` refuses measure creates on this surface by construction (the
-    # customer-grain descope), which is the fail-closed shape, not an accident.
-    if scope == "product":
-        from routes_products import MODULE as PRODUCT_MODULE, product_assembly
-
-        session.require(PRODUCT_MODULE)   # the product wall (dependency is session-only, C4)
-        g = product_assembly(session, consume_corrections=False)
-    elif scope.startswith("ut_"):
-        # Wave 18 C3-UT — the user-table wall (creator/admin) is inside the assembly; the
-        # measure/cohort context is honestly EMPTY (customer-grain machinery, no meaning here).
-        # ⭐⭐ WAVE 30 / W30-T30 (owner item 2: *"when I click hide fields it crash… no matter the
-        # size"*). This used to be `ut_assembly(...)`, which builds the whole table to validate a
-        # write it then throws away: `scoped_pool` allocates a dict per row and sorts them, so ONE
-        # hide-fields checkbox rebuilt ~33k order rows before the event was even dispatched.
-        # ⛔ NOTHING IS VALIDATED LESS. The comment above still holds — this assembly is the
-        # permission wall and the admission context — and `ut_write_ctx` returns the SAME six keys
-        # this route reads, with a pid set derived from exactly the row ids `scoped_pool` would
-        # have kept. What it does not do is materialise the rows nobody here looks at.
-        from routes_tables import ut_write_ctx
-
-        g = ut_write_ctx(session, scope)
-        # ⭐⭐ W31-T20 / D-174 — A WRITE THAT NEEDS THE ROW SET REFUSES OUT LOUD, and this is the
-        # half that is easy to skip because the code already "fails closed" without it. When a
-        # read-through grid's population exceeds one window the pid set is EMPTY, and every
-        # pid-bearing handler in `grid_events` then answers `False` — `overlay_patch` and
-        # `add_to_list` both `return False` for a pid not in `allowed_pids`. On the wire that is
-        # `rerender: false` and HTTP 200: a write the user watched succeed, that did nothing
-        # ([[lost-write-looks-like-failed-read]]). Schema-only events — hide a field, save a view,
-        # rename a column — name no pid and are untouched, which is D-170.
-        # ⚠ THE TEST IS THE EVENT'S OWN KEYS, not a kind list: a new pid-bearing event type would
-        # otherwise inherit the silent no-op the day it is added.
-        if g.get("limits"):
-            named = [e for e in events
-                     if isinstance(e, dict) and (e.get("pid") is not None or e.get("pids"))]
-            if named:
-                lim = g["limits"][0]
-                raise err(409, "pid_scope_unresolved",
-                          f"this database is served through the connector mirror and its rows "
-                          f"cannot be listed in one window, so a change addressed to particular "
-                          f"records ({len(named)} of {len(events)} here) cannot be admitted — "
-                          f"{lim.get('cause') or 'the row set is unresolved'}. "
-                          f"{lim.get('recommendation') or ''}".strip())
-    else:
-        g = grid_assembly(session, scope=scope, consume_corrections=False)
-    # ⚠ THE MEASURE CONTEXT IS NOT OPTIONAL (2026-07-31). Without `measure_offer`,
-    # `clean_measure_field` had an empty admission list and every measure-column create over
-    # HTTP was silently refused; without `measure_keys`, `clean_filter_tree` stripped every
-    # measure CONDITION out of a saved view. The embed always passed these; the API adapter
-    # simply had not been given them — the standalone shell could read measures it could
-    # never write.
-    ctx = _ctx(session, g["fields"], g["pids"], scope_key=scope,
-               measure_keys=frozenset(m["key"] for m in g["measures"]),
-               resolved_ids=frozenset(g["measure_sets"]),
-               cohort_ids=frozenset(c["id"] for c in g["lists"]),
-               measure_offer=tuple(g["measures"]),
-               visible_views=tuple(g["views"]))
-
-    # Per-event results so the client can tell which of a batch landed — the component's own
-    # bridge has no response channel at all, so this is strictly more than the embed gets.
-    results = []
-    try:
-        for one in events:
-            eid = str(one.get("id") or "") if isinstance(one, dict) else ""
-            # ⭐⭐ D-291 — WHICH event was refused, not merely THAT something was. The handler
-            # appends to a shared list, so the refusals belonging to THIS event are exactly the
-            # ones that appeared across THIS call. Reading the list once after the loop would
-            # answer "the batch was refused" and leave the caller to guess which member, which is
-            # the same class of unfalsifiable answer the channel exists to end.
-            _before = len(ctx.out.refusals)
-            rerender = grid_events.handle_one(one, ctx)
-            row = {"id": eid, "rerender": bool(rerender)}
-            mine = ctx.out.refusals[_before:]
-            if mine:
-                # `refused` is the SHAPE a caller branches on; the first reason is the one that
-                # stopped this write (a handler returns at its first refusal).
-                row["refused"] = mine[0]
-            results.append(row)
-    except grid_events.StoreUnavailable:
-        raise err(503, "store_unavailable",
-                  "the tenant store is unavailable — none of your changes were saved")
-
-    out = {"results": results, "rerender": any(r["rerender"] for r in results)}
-    if ctx.out.refusals:
-        # ⚠ ALSO AT THE TOP LEVEL, because a batch that was wholly refused must not read as a
-        # batch that wholly landed. A caller that only checks the envelope still learns something
-        # is wrong, and a caller that walks `results` learns exactly which member.
-        out["refusals"] = list(ctx.out.refusals)
-    if ctx.out.doc is not None:
-        # ⭐ C4 / W30-T27 — `docPayload` IS THE NAME THE CLIENT ALREADY DECLARES. `types.ts` has
-        # carried `docPayload?: {pid, docId, name, mime, data_b64}` since C5, and `Documents.tsx`
-        # matches it against the fetch it is waiting on — while this route has been answering
-        # `doc`, which `apiBridge.ts` deliberately drops. One object, emitted under the name the
-        # consumer looks for, so F's wiring needs no translation step to get wrong.
-        # ⚠ `doc` stays for one wave: nothing in the client reads it, but a gate might, and
-        # removing a key to save six bytes is not worth a red nobody predicted.
-        out["doc"] = out["docPayload"] = ctx.out.doc
-    if ctx.out.toast is not None:
-        out["toast"] = ctx.out.toast
-
-    # ── ⭐ owner item 2 (2026-08-03): THE NEW MEASURE COLUMN'S VALUES, ONE ROUND TRIP SOONER ──
-    #
-    # Creating a measure column cost the browser TWO sequential trips before a single number
-    # appeared: this one to persist the field, then a whole `/workspace` to compute it. The
-    # second cannot start until the first lands (the resolver reads the PERSISTED field), so the
-    # wait was structural, not slow code — the owner's "it takes some time for the data to
-    # populate". The values are computed here instead, immediately after the write, and ride
-    # this response.
-    #
-    # ⚠ IT COSTS NOTHING EXTRA TO COMPUTE. The expensive part is one DuckDB aggregate over the
-    # book, and `rt.measure_memo` is keyed on (pool stamp, scope, pool, measure, window) — so
-    # the `/workspace` re-read that still follows HITS the memo instead of doing this work. The
-    # query happens once either way; only its position moved.
-    #
-    # ⚠ NARROW ON PURPOSE. Gated to an actual measure-column write, so an overlay edit or a
-    # cohort add — the overwhelming majority of events — never pays for a second assembly.
-    #
-    # ⚠ AND IT IS A SHORTCUT, NOT A PATH. Any failure is swallowed: `WORKSPACE_STALE` still
-    # fires from `rerender`, and the re-read still delivers these values exactly as it does
-    # today. Nothing depends on this having worked.
-    if out["rerender"] and scope in ("customer", "cohort") and any(
-            isinstance(e, dict) and e.get("type") == "field_upsert"
-            and str(((e.get("field") or {}) if isinstance(e.get("field"), dict) else {})
-                    .get("key") or "").startswith("measure_")
-            for e in events):
-        try:
-            fresh = grid_assembly(session, scope=scope, consume_corrections=False)
-            out["derived"] = {str(pid): cells for pid, cells in fresh["derived"].items()}
-        except Exception:
-            pass
-    # ⚠ NOTHING IS INVALIDATED HERE, on purpose. The runtime cache holds ONLY the scope-shaped
-    # Odoo pool (see `routes_customers._pool_rows`), and no event on this route can change an
-    # Odoo column — Odoo is read-only. Everything an event DOES change (overlays, fields, views,
-    # folders, cohorts) is re-read from the store on the next request. An earlier version cleared
-    # `pool_cache` after an `overlay_patch`, which threw away an expensive Odoo pull to refresh
-    # data that was never in it.
-    return out
-
-
-# ── CONTRACT C1 (W36-T20): THE REGISTRY TOPICS' ROW READERS ───────────────────────────────────
-# ⭐⭐ R6 — *"EVERY database gets the same permission logic, always."* `core.perm_scope.scoped_table`
-# is the ONE door to any database's rows, and it cannot import a topic's pool builder: `core` never
-# imports up (`platform/ARCHITECTURE.md`) and these pools are built by `modules/` + `aios_grid`
-# behind this layer's per-tenant cache. So the app layer DECLARES its readers, exactly as
-# `routes_odoo_tables` declares connected tables to `user_tables.register_connected`.
-#
-# ⛔ REGISTERED HERE RATHER THAN IN `routes_customers`/`routes_products` because those two files
-# are outside wave 36's lane-C fence. The readers themselves are three lines each and call the
-# SAME `_pool_for` + `derive_pool_scope` pair those routes call, so there is no second pool and no
-# second scope derivation — only a second CALLER of the one that exists.
-#
-# ⚠ AND THE TOPIC ROUTES STILL HAVE THEIR OWN DOOR TODAY. Contract C1 says `apply_row_scope` +
-# `visible_fields` "move behind" `scoped_table`; moving `routes_customers.grid_assembly` and
-# `routes_products.scoped_pool` is booked as a PENDING row (mailbox/C.md, C-1) rather than done
-# here, because neither file is in this fence. What ships now is the arm the wave is load-bearing
-# on — every `ut_*` database, plus E's sandbox — and a topic arm that is REAL rather than stubbed,
-# so `scoped_table`'s topic leg is exercised by the product instead of only by a gate.
-def _topic_rt(st, module):
-    """The tenant runtime a topic pool must be built against, or a REPORTED refusal.
-
-    ⛔ A topic pool is per TENANT (`rt.pool_cache`), so `st=None` cannot be resolved to "the
-    default" without picking a tenant at random — which on this box is tenant #0's PRODUCTION
-    data. Standing rule 1's second sentence: say why, and say what to do instead.
-    """
-    if st is None:
-        import core.perm_scope as perm_scope
-        raise perm_scope.Unresolvable(
-            subject="rows", effect="unreadable",
-            cause=f"'{module}' is a registry topic whose pool is built per tenant and no tenant "
-                  f"runtime was passed",
-            recommendation="pass the session's runtime as `st=`. A topic pool cannot be "
-                           "resolved without knowing which tenant is asking")
-    return st
-
-
-def _customer_rows(table_key, user, st):
-    """`(fields, rows)` for the customer topic — the SAME derivation `_team_agent` uses."""
-    import aios_grid
-    import core.perm_scope as perm_scope
-    from routes_customers import _pool_for
-
-    rt = _topic_rt(st, table_key)
-    team_id, agent = perm_scope.derive_pool_scope(user, table_key)
-    return list(aios_grid.FIELDS), _pool_for(rt, team_id, agent)
-
-
-def _product_rows(table_key, user, st):
-    """`(fields, rows)` for the product topic. `consolidated=` follows the derived scope, so a
-    BU-pinned reader gets that BU's field contract rather than the consolidated one."""
-    import core.perm_scope as perm_scope
-    from routes_products import _pool_for, pd_fields
-
-    rt = _topic_rt(st, table_key)
-    team_id, _agent = perm_scope.derive_pool_scope(user, table_key)
-    return pd_fields(consolidated=team_id is None), _pool_for(rt, team_id)
-
-
-def _register_topic_rows():
-    """Declare both topic readers to C1. Called at import; returns the registered key set.
-
-    ⚠ THE KEYS ARE LITERALS AND THE ROUTE IMPORTS ARE INSIDE THE READERS, on purpose: this runs at
-    module import, and `from routes_products import MODULE` here would pull a sibling router in
-    before its own imports have settled. Every other cross-router reference in this file is lazy
-    for the same reason. The literals are held to their sources by `verify_scopes`, so they cannot
-    drift into naming a topic that does not exist.
-    """
-    import core.perm_scope as perm_scope
-
-    perm_scope.register_rows(_customer_rows, MODULE)
-    return perm_scope.register_rows(_product_rows, _PRODUCT_MODULE)
-
-
-#: ⚠ `_`-prefixed, because three functions in this file already bind the name `PRODUCT_MODULE`
-#: LOCALLY from `routes_products`. A module-level twin of that spelling would read as the same
-#: thing and be a different one — [[constant-two-features-share]] waiting to happen.
-_PRODUCT_MODULE = "product_data"
-
-_C1_ROW_SOURCES = _register_topic_rows()
+"""routes_grid.py — X2's write seam over HTTP: the SECOND adapter on `core.grid_events` (EXIT-1c).
+
+`POST /api/v1/grid/events` takes the component's event objects VERBATIM — the same objects the
+Streamlit host receives through the component value slot, unchanged — and runs them through the
+same `core.grid_events.handle_events` the Streamlit adapter runs. That is the whole point of
+EXIT-1a: one implementation of every permission wall, two transports. A route that re-validated
+anything here would be a second wall to keep in step, and the two would drift on the first change.
+
+DEDUP IS PER REQUEST, and that is a deliberate limit, not an oversight. The Streamlit adapter's
+`seen_ids` lives in `st.session_state` — genuinely per user session — because the client resends
+its recent 24-event window on every emit inside one page session. A stateless API has no such
+dict, and inventing a per-user server-side one would be exactly the resident per-tenant state
+EXIT-4a exists to remove. So: ids are deduped WITHIN a request body (the resend window's whole
+purpose — a batch that repeats an id processes it once), and a genuinely replayed request is
+handled by the operations being idempotent. The one event where a replay is observable is
+`add_to_list` (it would add the same pids to the same cohort twice — a set union, so the
+membership is unchanged, but the toast count repeats). Noted rather than papered over; a
+server-side idempotency key belongs with D2's session mirror (C-2).
+
+STORE DOWN = 503. `fallback_ws` is None on this adapter, so `core.grid_events` raises
+`StoreUnavailable` rather than writing to an in-memory workspace no API request could ever read
+back. A 200 over a write that evaporated is the failure this rule exists to prevent.
+"""
+import datetime as dt
+import time
+
+from fastapi import APIRouter, Body, Depends
+
+from deps import Session, err, module_gate, perms, require_session
+
+router = APIRouter(prefix="/api/v1")
+
+MODULE = "customer_data"
+
+#: The client's own resend window (`app.py`'s `[-24:]`). A body larger than this is not a
+#: legitimate client — refuse it rather than doing 500 store writes on one request.
+_MAX_EVENTS = 24
+
+
+def _ctx(session: Session, fields, pids, **kw):
+    from core import grid_events
+    import aios_grid
+    import core.perm_scope as perm_scope
+
+    # Wave 16 C-TOPIC: the ctx is TOPIC-SHAPED. The product scope swaps all three of the
+    # things a write is validated against — the module the permission wall reads, the canonical
+    # contract the hidden-field closure runs over, and the TABLE OPS the write lands in.
+    # Getting any one of them from the other topic is the "validated against the wrong field
+    # contract" near-miss the wave-15 routes_products header warned about.
+    scope_key = str(kw.get("scope_key") or "")
+    if scope_key == "product":
+        import modules.product_data as pd
+        from routes_products import MODULE as _PMOD, pd_fields
+
+        module, canonical = _PMOD, pd_fields(consolidated=True)
+        kw.setdefault("table", pd.TABLE_OPS)
+        hidden = perm_scope.hidden_keys(session.user, module, canonical)
+    elif scope_key.startswith("ut_"):
+        # Wave 18 C3-UT: a user table has no module in the permission wall (its wall is
+        # `user_tables.may_open`, already applied by the assembly this route ran first), so
+        # the hidden-field closure is EMPTY rather than borrowed from another topic's contract.
+        import core.table_store as table_store
+
+        kw.setdefault("table",
+                      table_store.make(f"{scope_key}_table_workspace", st=session.runtime))
+        hidden = frozenset()
+    else:
+        module, canonical = MODULE, aios_grid.FIELDS
+        hidden = perm_scope.hidden_keys(session.user, module, canonical)
+    return grid_events.EventCtx(
+        uname=session.uname, allowed_pids=pids, fields=fields, admin=session.admin,
+        # C-PERM: the write wall's field half, on the EVENTS transport too. This route takes the
+        # component's event objects verbatim, so a hidden key would otherwise arrive here with
+        # nothing between it and the store.
+        hidden_keys=hidden,
+        # ⭐ WAVE 25 (R6b, closes D-16) — THE TENANT HANDLE, on EVERY scope. This is the line
+        # D-16's exit condition names, and without it the rest of the fix is inert: the handler
+        # falls back to the module-global `core.store`, which is tenant #0's repo, so a Nurilab
+        # user's documents, cohorts and — through `_tops` — their whole customer/product table
+        # workspace were written into Royal Imports' dataset. Note it is set for the CUSTOMER
+        # and PRODUCT branches too, not only `ut_`: those two never passed a scoped `table`, so
+        # they were the ones actually resolving to tenant #0 on every request.
+        st=session.runtime,
+        fallback_ws=None, seen_ids={}, **kw)
+
+
+#: The surfaces the ONE grid serves. `cohort` is the same table over hand-curated SETS rather
+#: than over the whole scoped pool — `app.py:page_cohort` calls the same `_table_grid` with
+#: `cohort_mode=True, scope_key='cohort'`. `product` (wave 16 C-TOPIC) is the SKU table:
+#: same engine, its own pool, field contract and workspace BUCKET (routes_products).
+_SCOPES = ("customer", "cohort", "product")
+
+
+def _scope_or_400(raw):
+    """⛔ REFUSE AN UNKNOWN SCOPE, never default it. A typo silently served as `customer` would
+    hand a user the whole book on a page they opened to see one cohort — the widening direction,
+    which is the one that must fail closed.
+
+    Wave 18 (C3-UT): a `ut_`-prefixed scope names a USER TABLE and passes through here —
+    existence and the per-table wall are enforced by `routes_tables.ut_assembly` (404/403),
+    which every consumer of such a scope goes through. Passing an unknown ut key therefore
+    still fails closed, just one layer down where the store can actually be consulted."""
+    scope = (raw or "customer").strip().lower()
+    if scope.startswith("ut_"):
+        return scope
+    if scope not in _SCOPES:
+        raise err(400, "bad_scope",
+                  f"scope must be one of {', '.join(_SCOPES)} — refusing to guess")
+    return scope
+
+
+@router.get("/workspace")
+def workspace(scope: str = "customer",
+              session: Session = Depends(require_session)):
+    """The durable table workspace for this session: views, fields, overlays, folders.
+
+    ⛔ WAVE 21 (C4): the wall is TOPIC-SHAPED, so the DEPENDENCY is session-only and each scope
+    asserts its own gate below. The old `module_gate("customer_data")` dependency 403'd a
+    `ut_*` workspace for any tenant whose catalogue omits the customer module (loopable/
+    nurilab/gtmlab ship modules w/o it) — and the client then fell back to the shared
+    localStorage bucket + demo data, which is exactly the "new database shows RI's customer
+    fields" defect. A user table's real wall is `user_tables.may_open`, enforced inside
+    `ut_assembly` (404/403, one layer down where the store can be consulted).
+
+    ⚠ WHY THIS EXISTS (S1↔S2, 2026-07-30 — an X2 AMENDMENT, see the split doc). X2 fixes
+    `/customers` at the exact shape `verify_fields_contract.py` referees, so the workspace cannot
+    ride along in it without forking the thing that gate keeps single-sourced. Without a
+    workspace route the standalone shell's write path would be WRITE-ONLY: events persist
+    server-side and nothing reads them back on reload, so a saved view looks lost to the user
+    even though the store has it. This route closes that read-back gap at its own URL.
+
+    `allowed_pids` is passed so a SHARED view's `memberPids` are re-scoped to THIS reader — the
+    wave-9 leak rule. Omitting it would hand a Fisch-scoped user a member list built by a
+    full-access user.
+    """
+    from core import grid_events
+    from routes_customers import grid_assembly
+
+    scope = _scope_or_400(scope)
+    # Wave 21 C4 — the per-scope gate the dependency no longer asserts: customer/cohort need
+    # the customer module; product asserts PRODUCT_MODULE in its branch; ut_* needs only a
+    # session (its wall is the table's own, inside the assembly).
+    if not (scope == "product" or scope.startswith("ut_")):
+        session.require(MODULE)
+    # ── Wave 18 C3-UT: a USER TABLE'S workspace — the third topic through the one wire. The
+    # storage key has no (bu, agent) because a user table has no Odoo scope; per-user by the
+    # table's own wall (creator or admin, `user_tables.may_open`).
+    if scope.startswith("ut_"):
+        from routes_tables import ut_assembly
+
+        storage_key = f"{session.tenant}:{scope}:{session.uname}"
+        try:
+            # ⭐⭐ W31-T20 / D-174 — `with_rows=False`. THIS ROUTE RENDERS NO ROW: the grid fetches
+            # them from `/tables/{key}/rows` or `/odoo-tables/{key}/rows` in the same paint, and
+            # `useCustomerData.load` fires both calls in one `Promise.all`. Building the table here
+            # bought nothing and cost everything — it is why `?scope=ut_odoo_gl_lines` answered
+            # `409 window_required` six times out of six on the live deploy while the ROWS door for
+            # the same table served a page in ~150 ms, i.e. a grid that could not be opened at all.
+            g = ut_assembly(session, scope, storage_key=storage_key, with_rows=False)
+        except grid_events.StoreUnavailable:
+            raise err(503, "store_unavailable", "the tenant store is unavailable")
+        workspace = g["workspace"]
+        workspace["overlays"] = g["ws"].get("overlays") or {}
+        # The field contract rides the cheap re-read on EVERY topic — see the customer branch
+        # below for why (a user table's first cohort changes its contract too: `_cohorts(ctx)`
+        # is scope-parameterized, so `ut_` surfaces have their own sets).
+        workspace["fields"] = g["fields"]
+        workspace["measures"] = g["measures"]
+        workspace["measureSets"] = g["measure_sets"]
+        # ⚠ `g["derived"]`, NOT `{}` (2026-08-04). R9 gave every topic its own cohort sets, and
+        # `ut_assembly` has been building this topic's cohort CELLS since — but this route threw
+        # them away, so the Locked-views column arrived on the cheap re-read with permanently
+        # empty values. A column that exists and can never have one is worse than an absent
+        # column: it reads as "this record is in no locked view", which is a claim.
+        workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()}
+        workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
+        try:
+            from core import users as _users
+            workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
+            workspace["userAvatars"] = {k: v for k, v in
+                                        _users.avatar_map(tenant=session.tenant).items()
+                                        if str(v).startswith("data:image/")}
+        except Exception:
+            workspace["userOptions"] = []
+            workspace["userAvatars"] = {}
+        workspace["scopeKey"] = scope
+        # ⭐ W31-T20 — R6's SECOND SENTENCE REACHES THE BROWSER. On a read-through grid too large
+        # for one window the pid set is empty, so cohort membership and a shared view's
+        # `memberPids` are UNRESOLVED here (both fail closed, and `workspace_wire` already stamps
+        # `missing` on a shortened cohort). An unannounced empty scope is the silent limit R6
+        # forbids; this is the announcement, and W31-T22 is the ticket that renders it.
+        workspace["limits"] = g.get("limits") or []
+        return {"workspace": workspace}
+
+    # ── Wave 16 C-TOPIC: the PRODUCT surface has its own assembly (own pool, own field
+    # contract, own workspace BUCKET) and the contract's own storage key. It branches before
+    # the customer derivation because the customer storage key embeds (bu, agent) while the
+    # product one is deliberately `all:all` — one product workspace per user (the wave doc's
+    # C-TOPIC line), since BU narrows the product VALUES, not which workspace you own.
+    if scope == "product":
+        from routes_products import MODULE as PRODUCT_MODULE, product_assembly
+
+        # ⛔ THE GATE CHANGES WITH THE SCOPE (wave 21 C4: the dependency is session-only now,
+        # so this line IS the product wall — not a second one on top of customer_data).
+        session.require(PRODUCT_MODULE)
+        storage_key = f"{session.tenant}:product-list:{session.uname}:all:all"
+        try:
+            g = product_assembly(session, scope=scope, storage_key=storage_key)
+        except grid_events.StoreUnavailable:
+            raise err(503, "store_unavailable", "the tenant store is unavailable")
+        workspace = g["workspace"]
+        workspace["overlays"] = g["ws"].get("overlays") or {}
+        # R9 made cohorts per-topic, so the PRODUCT surface has its own sets and its own
+        # first-cohort contract change. Same reason as the customer branch below.
+        workspace["fields"] = g["fields"]
+        # The measure channel is EMPTY on this topic (customer-grain descope) — stated
+        # explicitly so the client's pickers grey rather than guess.
+        workspace["measures"] = g["measures"]
+        workspace["measureSets"] = g["measure_sets"]
+        # `g["derived"]` — the same correction as the user-table branch above, for the same
+        # reason: `product_assembly` builds this topic's cohort cells and this route discarded
+        # them. The measure half stays empty on this topic by descope, which is a different
+        # statement and one the offer already makes.
+        workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()}
+        workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
+        try:
+            from core import users as _users
+            workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
+            workspace["userAvatars"] = {k: v for k, v in
+                                        _users.avatar_map(tenant=session.tenant).items()
+                                        if str(v).startswith("data:image/")}
+        except Exception:
+            workspace["userOptions"] = []
+            workspace["userAvatars"] = {}
+        workspace["scopeKey"] = scope
+        return {"workspace": workspace}
+
+    # The storage key mirrors the host's own convention byte-for-byte (app.py's `_table_grid`
+    # call sites) with the session's TENANT in the host's hardcoded slot, so localStorage state
+    # carries across embed ⇄ standalone on the same browser.
+    # Wave 15 R1 — the SAME derivation the pool uses (`routes_customers._team_agent`), so the
+    # storage key cannot drift from the scope it names. Value-identical for every migrated
+    # record (verify_perm_scope section D proves the derivation reproduces the legacy scope
+    # exactly), so nobody's saved localStorage state moves when the migration runs.
+    import core.perm_scope as perm_scope
+    team_id, agent = perm_scope.derive_pool_scope(session.user, MODULE)
+    bu = team_id if team_id is not None else "all"
+    if scope == "cohort":
+        storage_key = f"{session.tenant}:cohort:{session.uname}:{bu}"
+    else:
+        storage_key = f"{session.tenant}:customer-list:{session.uname}:{bu}:{agent or 'all'}"
+
+    # ⛔ THE WIRE SHAPE IS THE SHARED PROJECTION (`aios_grid.workspace_wire`), not the store
+    # shape. The first version returned the store dict with no `storageKey` — and the client
+    # validator requires one, so the standalone shell DISCARDED the whole workspace: saved views
+    # never rendered, `cohortMode` never arrived, and the Cohort route silently drew the Customer
+    # surface. One projection for both servers is the fix that cannot drift.
+    try:
+        g = grid_assembly(session, scope=scope, storage_key=storage_key)
+    except grid_events.StoreUnavailable:
+        raise err(503, "store_unavailable", "the tenant store is unavailable")
+    workspace = g["workspace"]
+    # ADDITIVE to the wire: the caller's own overlay cells. The grid reads overlays from the
+    # /customers rows, but this route is the cheap read-back a write can be verified against
+    # (verify_api's overlay probe) without paying the pool call. Per-user by construction —
+    # `table_workspace` is this session's workspace.
+    workspace["overlays"] = g["ws"].get("overlays") or {}
+    # ⭐ THE FIELD CONTRACT, 2026-08-04 — and it is NOT decoration.
+    #
+    # `fields_from_workspace(ws, cohorts=bool(cohort_lists))` appends the derived "Locked
+    # views" column ONLY when the caller owns at least one cohort. So a user's FIRST cohort
+    # CHANGES THE FIELD CONTRACT — and the rows call that used to be the only carrier of
+    # `fields` is deliberately never re-fetched on a write (it is a 15-minute-cached Odoo
+    # pull; this route is the cheap re-read). The client therefore had no way to learn about
+    # that column short of a remount, which is the second half of the owner's "it only shows
+    # up when I switch modules and come back".
+    #
+    # ⚠ `g["fields"]` is the SAME list `_payload` serves on `/customers` — same assembly,
+    # same permission wall (`hidden_keys` already applied) — so the two wires cannot disagree
+    # about what a column is. Taking it from anywhere else would fork the contract that
+    # verify_fields_contract.py exists to keep single-sourced.
+    workspace["fields"] = g["fields"]
+    # ── the STANDALONE measure channel (owner item 1, 2026-07-31) ────────────────────────────
+    # The embed receives these as top-level render args; standalone lifts them off THIS route
+    # (useCustomerData merges them into the payload slots the grid already reads). They ride
+    # the workspace rather than /customers because a durable write re-reads exactly this route
+    # (WORKSPACE_STALE), so a new measure column populates without refetching the heavy pool.
+    workspace["measures"] = g["measures"]
+    workspace["measureSets"] = g["measure_sets"]
+    # Derived cells (cohort column + measure columns), keyed by pid. JSON object keys are
+    # strings; the client indexes with String(pid).
+    workspace["derived"] = {str(pid): cells for pid, cells in g["derived"].items()}
+    # Who is looking (permissions verdicts) + the assignee choices for `user`-typed columns —
+    # the two other host-only render args the shell was missing (fail-closed without them:
+    # restricted fields uneditable, assignee picker empty).
+    workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
+    try:
+        from core import users as _users
+        workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
+        # Wave 14 C-AVATAR — the options vocabulary's companion: display name -> data URL.
+        # Absent entries fall back to the client's initials disc; a non-data: value is
+        # dropped (defence in depth beside the write-side wall in routes_auth).
+        workspace["userAvatars"] = {k: v for k, v in
+                                    _users.avatar_map(tenant=session.tenant).items()
+                                    if str(v).startswith("data:image/")}
+    except Exception:
+        workspace["userOptions"] = []
+        workspace["userAvatars"] = {}
+
+    # THE SURFACE STAMP — a MIRROR of the host's own (`_table_grid`'s hide/cohort stamps).
+    # ⚠ Deliberately NOT `hideViews`: `cohortMode` is what swaps the Views panel for the cohort
+    # list, and two switches for one behaviour would drift.
+    workspace["scopeKey"] = scope
+    if scope == "cohort":
+        workspace["cohortMode"] = True
+        # The create pane offers "this cohort table only" (default) vs "all customer tables".
+        workspace["scopeChoice"] = True
+    return {"workspace": workspace}
+
+
+# ─────────────────────────────────────────────── the TIME-SERIES channel (C-TS, 2026-08-02)
+_TS_BUCKETS = ("week", "month", "quarter", "year")
+_TS_MAX_BUCKETS = 120
+_TS_MAX_FIELDS = 12
+_TS_MAX_PIDS = 5000
+_TS_MAX_LAST_N = 120
+#: C-TSWIN (wave 14): sliding windows cost ONE aggregate query per (metric, bucket) — the
+#: price of "bucket N == the grid cell at bucket N's end". The product cap keeps a legal
+#: 120-bucket × 12-metric ask from becoming 1,440 queries in one request; typical panels
+#: (12 × 3) sit two orders of magnitude under it. Dated amendment in the split doc.
+_TS_MAX_CELLS = 720
+
+
+def _ts_start_of(bucket, d):
+    """The calendar START of the bucket holding `d` (week = Monday, the vocabulary rule)."""
+    if bucket == "week":
+        return d - dt.timedelta(days=d.weekday())
+    if bucket == "month":
+        return d.replace(day=1)
+    if bucket == "quarter":
+        return d.replace(month=((d.month - 1) // 3) * 3 + 1, day=1)
+    return d.replace(month=1, day=1)
+
+
+def _ts_next(bucket, d):
+    if bucket == "week":
+        return d + dt.timedelta(days=7)
+    if bucket == "year":
+        return d.replace(year=d.year + 1)
+    step = 1 if bucket == "month" else 3
+    m = d.month + step
+    return dt.date(d.year + (m - 1) // 12, (m - 1) % 12 + 1, 1)
+
+
+def _ts_prev(bucket, d):
+    if bucket == "week":
+        return d - dt.timedelta(days=7)
+    if bucket == "year":
+        return d.replace(year=d.year - 1)
+    step = 1 if bucket == "month" else 3
+    y, m = d.year, d.month - step
+    while m < 1:
+        m += 12
+        y -= 1
+    return dt.date(y, m, 1)
+
+
+def _ts_label(bucket, start):
+    if bucket == "week":
+        return f"Wk of {start.strftime('%b %d')}"
+    if bucket == "month":
+        return start.strftime("%b %Y")
+    if bucket == "quarter":
+        return f"Q{(start.month - 1) // 3 + 1} {start.year}"
+    return str(start.year)
+
+
+@router.post("/grid/timeseries")
+def grid_timeseries(body: dict = Body(default=None), scope: str = "customer",
+                    session: Session = Depends(module_gate(MODULE))):
+    """Pooled measure values per calendar bucket — the time-series view's data channel.
+
+    C-TSWIN (wave 14, ruling R1): AS-OF semantics. Each metric's OWN stored window is
+    re-resolved per bucket with `today := min(bucket end, real today)` — `ytd` is cumulative
+    from Jan 1, `last_90_days` trailing, `all_time` cumulative ever; bucket N equals what the
+    grid's measure column would show if today were bucket N's end. Fixed-range (`custom`)
+    windows cannot slide and are dropped per field as `window_fixed`. Rows carry
+    `window: {kind, label}` so a cumulative row cannot be misread as periodic, and there is
+    NO `total` column — sliding windows overlap, so a sum of columns would double-count.
+
+    The client sends the pids its view currently matches (the filter IS the scope); the server
+    intersects them with the session's book, so the request can only ever NARROW. Values are
+    POOLED aggregates computed by the semantic layer's own expression (the additivity law: an
+    average is computed at pool grain, never averaged over per-customer answers). A bucket
+    with no rows is 0 for a sum/count and null for anything else; a bucket that has not
+    STARTED yet is null for every kind — an unstarted period's YTD is unanswered, not 0.
+    """
+    from core import measure_resolve
+    from harness import windows as _wn
+    from routes_customers import _pool_stamp, _team_agent, allowed_pids
+
+    # Wave 16 C-TOPIC: the channel is CUSTOMER-GRAIN (measure_resolve's own grain), so the
+    # product surface is refused in words rather than answered with an id-space accident —
+    # product pids are CRC32 hashes and would mostly fall outside the customer book anyway,
+    # but "mostly" is not a wall. The client hides the mode for this topic; this is the
+    # server's half of the same refusal.
+    _sc = _scope_or_400(scope)
+    if _sc == "product" or _sc.startswith("ut_"):
+        raise err(400, "bad_timeseries",
+                  "this surface has no time-series channel — measures are customer-grain")
+
+    body = body or {}
+    bucket = body.get("bucket")
+    if bucket not in _TS_BUCKETS:
+        raise err(400, "bad_timeseries", "bucket must be one of week, month, quarter, year")
+    raw_fields = body.get("fields")
+    if not isinstance(raw_fields, list) or not raw_fields:
+        raise err(400, "bad_timeseries", "fields must be a non-empty list of field keys")
+    if len(raw_fields) > _TS_MAX_FIELDS:
+        raise err(400, "bad_timeseries", f"at most {_TS_MAX_FIELDS} fields per request")
+    raw_pids = body.get("pids")
+    if not isinstance(raw_pids, list) or not raw_pids:
+        raise err(400, "bad_timeseries", "pids must be a non-empty list")
+    if len(raw_pids) > _TS_MAX_PIDS:
+        raise err(400, "bad_timeseries", f"at most {_TS_MAX_PIDS} pids per request")
+    try:
+        wanted = {int(p) for p in raw_pids}
+    except (TypeError, ValueError):
+        raise err(400, "bad_timeseries", "pids must be integers")
+    pool = wanted & {int(p) for p in allowed_pids(session)}
+    if not pool:
+        # 403, not an empty 200: the caller asked about customers outside their book, and an
+        # all-zero series over nobody would read as "no activity", not "not yours".
+        raise err(403, "out_of_scope", "none of those customers are in your book")
+
+    span = body.get("span")
+    if not isinstance(span, dict):
+        raise err(400, "bad_timeseries", "span must be {'lastN': n} or {'from': .., 'to': ..}")
+    today = time.strftime("%Y-%m-%d")
+    t = dt.date.fromisoformat(today)
+    n = span.get("lastN")
+    starts = []
+    if n is not None:
+        if not isinstance(n, int) or isinstance(n, bool) or not (1 <= n <= _TS_MAX_LAST_N):
+            raise err(400, "bad_timeseries", f"lastN must be 1..{_TS_MAX_LAST_N}")
+        cur = _ts_start_of(bucket, t)
+        starts = [cur]
+        for _ in range(n - 1):
+            cur = _ts_prev(bucket, cur)
+            starts.append(cur)
+        starts.reverse()
+    else:
+        try:
+            d_from = dt.date.fromisoformat(str(span.get("from")))
+            d_to = dt.date.fromisoformat(str(span.get("to")))
+        except (TypeError, ValueError):
+            raise err(400, "bad_timeseries", "span.from/to must be ISO dates (YYYY-MM-DD)")
+        if d_from > d_to:
+            d_from, d_to = d_to, d_from
+        cur = _ts_start_of(bucket, d_from)
+        while cur <= d_to:
+            starts.append(cur)
+            if len(starts) > _TS_MAX_BUCKETS:
+                raise err(400, "bad_timeseries",
+                          f"that span is more than {_TS_MAX_BUCKETS} {bucket} buckets - "
+                          f"narrow it")
+            cur = _ts_next(bucket, cur)
+        if not starts:
+            raise err(400, "bad_timeseries", "the span holds no buckets")
+    ends = [_ts_next(bucket, s) - dt.timedelta(days=1) for s in starts]
+
+    rt = session.runtime
+    if not rt.available():
+        raise err(503, "store_unavailable", "the tenant store is unavailable")
+    import modules.customer_data as cl_mod
+    # Read-only route: it must not consume a one-shot label-correction ack the browser has
+    # not seen yet (the same rule event validation follows).
+    ws = cl_mod.table_workspace(session.uname, consume_corrections=False)
+    fdefs = ws.get("fields") or {}
+    keys, dropped, seen = [], [], set()
+    for k in raw_fields:
+        k = str(k or "")
+        if not k or k in seen:
+            continue
+        seen.add(k)
+        fd = fdefs.get(k)
+        if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict):
+            # Named, never silent: v1 eligibility is measure-backed fields only (C-TS).
+            dropped.append({"field": k, "reason": "not_a_measure_field"})
+            continue
+        keys.append(k)
+    # ⚡ WAVE 17 R9 (amendment 2026-08-03, GRID's cross-fence ask) — A SHEET WITH NO MEASURE
+    # FIELDS IS NO LONGER A 400. It answers with the BUCKET GRID and no rows.
+    #
+    # Why this is the honest direction and not a loosening: the bucket starts and their labels
+    # are SERVER math (`_ts_start_of` / `_ts_next` / `_ts_label`), and R9 has the client
+    # synthesizing snapshot rows for preset columns that have no window to slide. Those rows
+    # must be painted under the SAME headings as everything else, so the client needs the
+    # columns even when the server has no series to put in them. Refusing the whole request
+    # meant the panel could offer nothing at all on this tenant — the shipped contract has zero
+    # measure-backed presets (`aios_grid.py`: "this branch is currently MEMBERLESS"), which is
+    # what item 5 is actually about.
+    #
+    # Nothing is invented by this: `rows` is empty and `meta.dropped` NAMES every field and why.
+    # A 400 is still returned for a request that is malformed (bad bucket, bad span, no fields
+    # at all) — this only stops treating "I asked about columns you cannot serve" as an error.
+    mfields, slid_keys = [], []
+    for k in keys:
+        m = dict(fdefs[k]["measure"])
+        if (m.get("window") or {}).get("kind") == "custom":
+            # C-TSWIN: a fixed date range cannot slide across buckets — named, never silent,
+            # the same posture as not_a_measure_field.
+            dropped.append({"field": k, "reason": "window_fixed"})
+            continue
+        mfields.append({"key": k, "measure": m})
+        slid_keys.append(k)
+    # Same amendment as above: every metric being fixed-range is a sheet with no SERIES, not a
+    # broken request. The columns still stand, and `window_fixed` still names each refusal.
+    if len(starts) * len(mfields) > _TS_MAX_CELLS:
+        raise err(400, "bad_timeseries",
+                  "that ask is too wide - narrow the span or pick fewer metrics")
+
+    team_id, agent = _team_agent(session)
+    stamp = _pool_stamp(rt, team_id, agent)
+    problems = []
+    bucket_bounds = [(s.isoformat(), e.isoformat()) for s, e in zip(starts, ends)]
+    # No slidable metrics = nothing to resolve. Skipping the call rather than asking the
+    # resolver about an empty list keeps the memo free of a meaningless key.
+    answers = measure_resolve.series_values(
+        mfields, bucket, bucket_bounds,
+        team_id, frozenset(pool), today, stamp, rt.series_memo,
+        on_error=lambda tag, e: problems.append(str(e)[:200])) if mfields else {}
+
+    columns = []
+    for s, e in zip(starts, ends):
+        col = {"key": s.isoformat(), "label": _ts_label(bucket, s),
+               "from": s.isoformat(), "to": e.isoformat()}
+        if e > t:
+            col["partial"] = True
+        columns.append(col)
+    rows = []
+    for k in slid_keys:
+        ans = answers.get(k)
+        if ans is None:
+            dropped.append({"field": k, "reason": "unresolvable"})
+            continue
+        vals_by = ans.get("values") or {}
+        agg_kind = str(ans.get("agg") or "sum")
+        zero_fill = agg_kind in ("sum", "count")
+        vals = []
+        for s in starts:
+            if s > t:
+                # C-TSWIN: an unstarted period is unanswered for EVERY agg kind — a future
+                # month's YTD zero-filled to 0 would read as "the year reset".
+                vals.append(None)
+            else:
+                vals.append(vals_by.get(s.isoformat(), 0 if zero_fill else None))
+        wspec = (fdefs[k].get("measure") or {}).get("window")
+        wnorm = _wn.normalize(wspec) or {}
+        rows.append({"field": k, "label": str(fdefs[k].get("label") or k)[:120],
+                     "agg": agg_kind, "values": vals,
+                     "window": {"kind": str(wnorm.get("kind") or ""),
+                                "label": _wn.label(wspec)}})
+    meta = {"pool": len(pool), "today": today, "bucket": bucket}
+    if dropped:
+        meta["dropped"] = dropped
+    if problems:
+        meta["problems"] = problems[:5]
+    return {"columns": columns, "rows": rows, "meta": meta}
+
+
+#: C-CAL caps (wave 17, owner item 9 / ruling R4). A month of days, a handful of metrics.
+#: Every one of these is a 400 WITH ITS REASON, never a silent trim: a calendar that quietly
+#: answered 20 of the 31 days asked about would paint eleven blank cells that look like days
+#: with no activity.
+_CAL_MAX_GROUPS = 31
+_CAL_MAX_FIELDS = 6
+_CAL_MAX_CELLS = 186
+
+
+@router.post("/grid/calendar_metrics")
+def grid_calendar_metrics(body: dict = Body(default=None), scope: str = "customer",
+                          session: Session = Depends(module_gate(MODULE))):
+    """Per-DAY measure values with each metric's own window slid to that day (C-CAL / R4).
+
+    ⛔ WHY THIS EXISTS AT ALL. The calendar's summary cells used to aggregate the row VALUES the
+    grid already held — which for a measure column means "every member's YTD **as of today**",
+    summed and printed under a date in March. The number was arithmetically fine and semantically
+    a lie: it answered a question about today while sitting in a cell labelled with another day.
+    R4: a metric in a day cell is computed AS OF THAT DAY.
+
+    The shape differs from the time-series channel in exactly one way, and it is the reason this
+    is a separate route rather than a parameter: **every group carries its OWN pid set**. A
+    calendar day holds the records the date field placed there, so day-to-day the subject
+    changes. One `allowed_pids` for the whole request — the TS channel's shape — would compute
+    each day over everybody, which is a different question again.
+
+    Static (non-measure) fields are NOT served here. They have no window to slide, so the
+    client's own per-day aggregation over row values stays correct for them; sending them would
+    invite a second implementation of arithmetic that already works.
+    """
+    from core import measure_resolve
+    from harness import windows as _wn
+    from routes_customers import _pool_stamp, _team_agent, allowed_pids
+
+    # The same refusal the TS channel makes, for the same reason: measures are customer-grain
+    # and product pids are CRC32 hashes of SKU codes. "Mostly outside the book" is not a wall.
+    _sc_cal = _scope_or_400(scope)
+    if _sc_cal == "product" or _sc_cal.startswith("ut_"):
+        raise err(400, "bad_calendar_metrics",
+                  "the product surface has no measure channel yet — measures are customer-grain")
+
+    body = body or {}
+    raw_groups = body.get("groups")
+    if not isinstance(raw_groups, list) or not raw_groups:
+        raise err(400, "bad_calendar_metrics",
+                  "groups must be a non-empty list of {key: 'YYYY-MM-DD', pids: [...]}")
+    if len(raw_groups) > _CAL_MAX_GROUPS:
+        raise err(400, "bad_calendar_metrics",
+                  f"at most {_CAL_MAX_GROUPS} days per request (one month)")
+    raw_fields = body.get("fields")
+    if not isinstance(raw_fields, list) or not raw_fields:
+        raise err(400, "bad_calendar_metrics", "fields must be a non-empty list of field keys")
+    if len(raw_fields) > _CAL_MAX_FIELDS:
+        raise err(400, "bad_calendar_metrics", f"at most {_CAL_MAX_FIELDS} metrics per request")
+    if len(raw_groups) * len(raw_fields) > _CAL_MAX_CELLS:
+        raise err(400, "bad_calendar_metrics",
+                  "that ask is too wide - fewer days or fewer metrics")
+
+    book = {int(p) for p in allowed_pids(session)}
+    groups, seen_days = [], set()
+    for g in raw_groups:
+        if not isinstance(g, dict):
+            raise err(400, "bad_calendar_metrics", "every group must be an object")
+        day = str(g.get("key") or "")
+        try:
+            d = dt.date.fromisoformat(day)
+        except (TypeError, ValueError):
+            raise err(400, "bad_calendar_metrics",
+                      "every group key must be an ISO date (YYYY-MM-DD)")
+        if day in seen_days:
+            raise err(400, "bad_calendar_metrics", f"day {day} appears twice")
+        seen_days.add(day)
+        raw_pids = g.get("pids")
+        if not isinstance(raw_pids, list):
+            raise err(400, "bad_calendar_metrics", "every group needs a pids list")
+        try:
+            wanted = {int(p) for p in raw_pids}
+        except (TypeError, ValueError):
+            raise err(400, "bad_calendar_metrics", "pids must be integers")
+        # NARROW-ONLY, per group. The client sends what its calendar placed; the server can
+        # only ever remove from that, never add.
+        groups.append((day, d, frozenset(wanted & book)))
+    if sum(len(p) for _, _, p in groups) > _TS_MAX_PIDS:
+        raise err(400, "bad_calendar_metrics",
+                  f"at most {_TS_MAX_PIDS} customer references per request")
+
+    rt = session.runtime
+    if not rt.available():
+        raise err(503, "store_unavailable", "the tenant store is unavailable")
+    import modules.customer_data as cl_mod
+    ws = cl_mod.table_workspace(session.uname, consume_corrections=False)
+    fdefs = ws.get("fields") or {}
+    keys, dropped, seen = [], [], set()
+    for k in raw_fields:
+        k = str(k or "")
+        if not k or k in seen:
+            continue
+        seen.add(k)
+        fd = fdefs.get(k)
+        if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict):
+            # The TS channel's vocabulary, deliberately reused rather than re-coined: the client
+            # already knows how to say these three words to a reader.
+            dropped.append({"field": k, "reason": "not_a_measure_field"})
+            continue
+        if ((fd["measure"].get("window") or {}).get("kind") == "custom"):
+            dropped.append({"field": k, "reason": "window_fixed"})
+            continue
+        keys.append(k)
+    if not keys:
+        raise err(400, "bad_calendar_metrics",
+                  "none of the requested fields are measure fields with a window that can "
+                  "slide to a day")
+
+    today = time.strftime("%Y-%m-%d")
+    t = dt.date.fromisoformat(today)
+    team_id, agent = _team_agent(session)
+    stamp = _pool_stamp(rt, team_id, agent)
+    problems = []
+    values = {k: {} for k in keys}
+    for day, d, pids in groups:
+        # A day that has not happened is unanswered for EVERY aggregate kind — the unstarted
+        # bucket law at day grain. Answering 0 would say "we sold nothing", which is a claim
+        # about a day nobody has lived through yet.
+        if d > t or not pids:
+            for k in keys:
+                values[k][day] = None
+            continue
+        answers = measure_resolve.series_values(
+            [{"key": k, "measure": dict(fdefs[k]["measure"])} for k in keys],
+            "day", [(day, day)], team_id, pids, today, stamp, rt.series_memo,
+            on_error=lambda tag, e: problems.append(str(e)[:200]))
+        for k in keys:
+            ans = answers.get(k)
+            if ans is None:
+                values[k][day] = None
+                continue
+            vals_by = ans.get("values") or {}
+            zero_fill = str(ans.get("agg") or "sum") in ("sum", "count")
+            values[k][day] = vals_by.get(day, 0 if zero_fill else None)
+
+    for k in keys:
+        if all(v is None for v in values[k].values()):
+            # Every day unanswerable is a FIELD-level failure, and saying so is the difference
+            # between "no activity that month" and "this metric could not be computed".
+            dropped.append({"field": k, "reason": "unresolvable"})
+    out = {"values": values, "today": today,
+           "windows": {k: {"kind": str((_wn.normalize(
+               (fdefs[k].get("measure") or {}).get("window")) or {}).get("kind") or ""),
+               "label": _wn.label((fdefs[k].get("measure") or {}).get("window"))}
+               for k in keys}}
+    if dropped:
+        out["dropped"] = dropped
+    if problems:
+        out["problems"] = problems[:5]
+    return out
+
+
+#: ⭐⭐ THE BULK CELL DOOR'S OWN CEILING, and it is REPORTED rather than silently enforced. The
+#: standing no-cap rule governs data read from a connected source; this is a WRITE of team-typed
+#: values, so a bound is right — but R6's second sentence still binds, which is why exceeding it
+#: is a 400 naming the number and the count, never a truncation.
+#: Sized against the real job with headroom: the 2027 catalog is 1,397 rows.
+MAX_BULK_ROWS = 10_000
+
+
+@router.post("/grid/bulk-cells")
+def bulk_cells(body: dict = Body(default=None),
+               session: Session = Depends(require_session)):
+    """`{scopeKey, rows: {"<pid>": {key: value}}}` → shared cells on many rows, ONE transaction.
+
+    ⭐⭐ WHY THIS EXISTS AT ALL, because "we already have `/grid/events`" is the obvious objection.
+    That door is capped at `_MAX_EVENTS = 24` (it is sized for the client's resend window), so a
+    1,397-row import is ~59 sequential POSTs against ONE JSON document — and this repo has a
+    MEASURED scar for exactly that: 18 writes against one document under the store's coalescing
+    single-flight landed **zero** while answering 200 eighteen times. Raising `_MAX_EVENTS` would
+    have widened the browser's own resend window to fix a script's problem. One door, one
+    transaction, is the honest shape.
+
+    ⛔ TENANT-WIDE KEYS ONLY, AND THAT IS THE POINT RATHER THAN A LIMITATION. It writes through
+    `shared_overlay`, which every account in the tenant reads. A per-user bulk write would be a
+    contradiction: nobody imports 1,397 rows of the team's work so that one account can see it —
+    that is precisely the failure owner ruling R6 was made to avoid. A key that is not declared
+    `shared: true` in the canonical contract is REFUSED BY NAME.
+
+    ⛔ ADMIN ONLY. This mutates data every account in the tenant reads, in bulk, in one call.
+    D-172 already records that the shared stratum has an asymmetric wall (anyone may create a
+    tenant-wide column, only the creator or an admin may delete one); a new door does not get to
+    inherit the loose half of an asymmetry somebody already flagged.
+    """
+    import core.shared_overlay as shared_overlay
+    import modules.product_data as pd
+    import core.perm_scope as perm_scope
+    from routes_products import MODULE as PRODUCT_MODULE, scoped_pool
+
+    scope = _scope_or_400((body or {}).get("scopeKey"))
+    if scope != "product":
+        # Fail-closed with the reason: the shared stratum is a PRODUCT-topic mechanism today.
+        raise err(400, "scope_not_bulk_writable",
+                  f"bulk cell writes are available on the product database; '{scope}' has no "
+                  f"tenant-wide cell stratum")
+    session.require(PRODUCT_MODULE)
+    if not session.admin:
+        raise err(403, "admin_only",
+                  "a bulk write changes values every account in this workspace reads")
+
+    rows = (body or {}).get("rows")
+    if not isinstance(rows, dict):
+        raise err(400, "bad_rows", "expected {rows: {\"<pid>\": {field: value}}}")
+    if len(rows) > MAX_BULK_ROWS:
+        raise err(400, "too_many_rows",
+                  f"at most {MAX_BULK_ROWS} rows per request; this one carried {len(rows)}")
+
+    pids, _team, _src, fields_base = scoped_pool(session)
+    allowed = {int(p) for p in pids}
+    shared_keys = set(pd.SHARED_KEYS())
+    hidden = perm_scope.hidden_keys(session.user, PRODUCT_MODULE, fields_base)
+
+    clean, unknown_pid, refused_keys = {}, [], {}
+    for raw_pid, values in rows.items():
+        try:
+            pid = int(raw_pid)
+        except (TypeError, ValueError):
+            unknown_pid.append(str(raw_pid)[:40])
+            continue
+        # ⚠ THE POOL IS THE WALL. `scoped_pool` is the same predicate the read door uses, so a
+        # caller cannot write a row they could not see — including a row in another BU.
+        if pid not in allowed:
+            unknown_pid.append(str(raw_pid)[:40])
+            continue
+        keep = {}
+        for key, value in dict(values or {}).items():
+            key = str(key)
+            if key not in shared_keys:
+                refused_keys.setdefault("not_shared", set()).add(key)
+                continue
+            if key in hidden:
+                refused_keys.setdefault("hidden_by_permissions", set()).add(key)
+                continue
+            if not isinstance(value, (str, int, float)) or isinstance(value, bool):
+                refused_keys.setdefault("unsupported_value", set()).add(key)
+                continue
+            keep[key] = value
+        if keep:
+            clean[pid] = keep
+
+    written = shared_overlay.put_rows(pd.TABLE_KEY, clean, st=pd.TABLE_OPS.st) if clean else {}
+    out = {
+        "rows_written": len(written),
+        "cells_written": sum(len(v) for v in written.values()),
+        "rows_requested": len(rows),
+    }
+    # ⭐ R6's second sentence: what did NOT land, and why. A bulk door that reports only its
+    # successes is how 701 unmatched SKUs disappear quietly.
+    if unknown_pid:
+        out["rows_not_in_your_pool"] = {"count": len(unknown_pid),
+                                        "sample": sorted(unknown_pid)[:10]}
+    if refused_keys:
+        out["refused_fields"] = {reason: sorted(keys)
+                                 for reason, keys in refused_keys.items()}
+    return out
+
+
+@router.post("/grid/events")
+def grid_events_route(body: dict = Body(default=None),
+                      session: Session = Depends(require_session)):
+    """`{events: [<component event objects, verbatim>]}` → `{results, doc?, toast?}`.
+
+    Wave 21 C4: session-only dependency, per-scope gate below — the write door must admit the
+    same sessions the read door (`/workspace`) admits, or a tenant without the customer module
+    can SEE its own user tables and not write to them."""
+    from core import grid_events
+    from routes_customers import grid_assembly
+
+    events = (body or {}).get("events")
+    if events is None and isinstance(body, dict) and body.get("type"):
+        events = [body]                    # a single event object, the legacy shape
+    if not isinstance(events, list):
+        raise err(400, "bad_events", "expected {events: [...]}")
+    if len(events) > _MAX_EVENTS:
+        raise err(400, "too_many_events",
+                  f"at most {_MAX_EVENTS} events per request (the client's resend window)")
+
+    # Validated with the SAME predicate as the read route. An unrecognised scopeKey used to be
+    # passed through verbatim, and `core.grid_events` only ever compares it to 'cohort' — so a
+    # typo degraded silently to customer-scope behaviour on a WRITE. Read and write must agree on
+    # what a scope is, or the surface you read is not the surface you wrote.
+    scope = _scope_or_400((body or {}).get("scopeKey"))
+    # Wave 21 C4 — same per-scope gate as /workspace (read and write doors must agree).
+    if not (scope == "product" or scope.startswith("ut_")):
+        session.require(MODULE)
+    # This assembly validates the write; it is not a payload the browser will render. Leave
+    # one-shot field-name correction acks queued for the subsequent /workspace refresh.
+    # Wave 16 C-TOPIC: the PRODUCT topic gets the product assembly — product field contract,
+    # product pids, and (below) the product TABLE OPS, so a product event is validated against
+    # and lands in the product bucket. The measure/cohort context is honestly EMPTY there:
+    # `clean_measure_field` refuses measure creates on this surface by construction (the
+    # customer-grain descope), which is the fail-closed shape, not an accident.
+    if scope == "product":
+        from routes_products import MODULE as PRODUCT_MODULE, product_assembly
+
+        session.require(PRODUCT_MODULE)   # the product wall (dependency is session-only, C4)
+        g = product_assembly(session, consume_corrections=False)
+    elif scope.startswith("ut_"):
+        # Wave 18 C3-UT — the user-table wall (creator/admin) is inside the assembly; the
+        # measure/cohort context is honestly EMPTY (customer-grain machinery, no meaning here).
+        # ⭐⭐ WAVE 30 / W30-T30 (owner item 2: *"when I click hide fields it crash… no matter the
+        # size"*). This used to be `ut_assembly(...)`, which builds the whole table to validate a
+        # write it then throws away: `scoped_pool` allocates a dict per row and sorts them, so ONE
+        # hide-fields checkbox rebuilt ~33k order rows before the event was even dispatched.
+        # ⛔ NOTHING IS VALIDATED LESS. The comment above still holds — this assembly is the
+        # permission wall and the admission context — and `ut_write_ctx` returns the SAME six keys
+        # this route reads, with a pid set derived from exactly the row ids `scoped_pool` would
+        # have kept. What it does not do is materialise the rows nobody here looks at.
+        from routes_tables import ut_write_ctx
+
+        g = ut_write_ctx(session, scope)
+        # ⭐⭐ W31-T20 / D-174 — A WRITE THAT NEEDS THE ROW SET REFUSES OUT LOUD, and this is the
+        # half that is easy to skip because the code already "fails closed" without it. When a
+        # read-through grid's population exceeds one window the pid set is EMPTY, and every
+        # pid-bearing handler in `grid_events` then answers `False` — `overlay_patch` and
+        # `add_to_list` both `return False` for a pid not in `allowed_pids`. On the wire that is
+        # `rerender: false` and HTTP 200: a write the user watched succeed, that did nothing
+        # ([[lost-write-looks-like-failed-read]]). Schema-only events — hide a field, save a view,
+        # rename a column — name no pid and are untouched, which is D-170.
+        # ⚠ THE TEST IS THE EVENT'S OWN KEYS, not a kind list: a new pid-bearing event type would
+        # otherwise inherit the silent no-op the day it is added.
+        if g.get("limits"):
+            named = [e for e in events
+                     if isinstance(e, dict) and (e.get("pid") is not None or e.get("pids"))]
+            if named:
+                lim = g["limits"][0]
+                raise err(409, "pid_scope_unresolved",
+                          f"this database is served through the connector mirror and its rows "
+                          f"cannot be listed in one window, so a change addressed to particular "
+                          f"records ({len(named)} of {len(events)} here) cannot be admitted — "
+                          f"{lim.get('cause') or 'the row set is unresolved'}. "
+                          f"{lim.get('recommendation') or ''}".strip())
+    else:
+        g = grid_assembly(session, scope=scope, consume_corrections=False)
+    # ⚠ THE MEASURE CONTEXT IS NOT OPTIONAL (2026-07-31). Without `measure_offer`,
+    # `clean_measure_field` had an empty admission list and every measure-column create over
+    # HTTP was silently refused; without `measure_keys`, `clean_filter_tree` stripped every
+    # measure CONDITION out of a saved view. The embed always passed these; the API adapter
+    # simply had not been given them — the standalone shell could read measures it could
+    # never write.
+    ctx = _ctx(session, g["fields"], g["pids"], scope_key=scope,
+               measure_keys=frozenset(m["key"] for m in g["measures"]),
+               resolved_ids=frozenset(g["measure_sets"]),
+               cohort_ids=frozenset(c["id"] for c in g["lists"]),
+               measure_offer=tuple(g["measures"]),
+               visible_views=tuple(g["views"]))
+
+    # Per-event results so the client can tell which of a batch landed — the component's own
+    # bridge has no response channel at all, so this is strictly more than the embed gets.
+    results = []
+    try:
+        for one in events:
+            eid = str(one.get("id") or "") if isinstance(one, dict) else ""
+            # ⭐⭐ D-291 — WHICH event was refused, not merely THAT something was. The handler
+            # appends to a shared list, so the refusals belonging to THIS event are exactly the
+            # ones that appeared across THIS call. Reading the list once after the loop would
+            # answer "the batch was refused" and leave the caller to guess which member, which is
+            # the same class of unfalsifiable answer the channel exists to end.
+            _before = len(ctx.out.refusals)
+            rerender = grid_events.handle_one(one, ctx)
+            row = {"id": eid, "rerender": bool(rerender)}
+            mine = ctx.out.refusals[_before:]
+            if mine:
+                # `refused` is the SHAPE a caller branches on; the first reason is the one that
+                # stopped this write (a handler returns at its first refusal).
+                row["refused"] = mine[0]
+            results.append(row)
+    except grid_events.StoreUnavailable:
+        raise err(503, "store_unavailable",
+                  "the tenant store is unavailable — none of your changes were saved")
+
+    out = {"results": results, "rerender": any(r["rerender"] for r in results)}
+    if ctx.out.refusals:
+        # ⚠ ALSO AT THE TOP LEVEL, because a batch that was wholly refused must not read as a
+        # batch that wholly landed. A caller that only checks the envelope still learns something
+        # is wrong, and a caller that walks `results` learns exactly which member.
+        out["refusals"] = list(ctx.out.refusals)
+    if ctx.out.doc is not None:
+        # ⭐ C4 / W30-T27 — `docPayload` IS THE NAME THE CLIENT ALREADY DECLARES. `types.ts` has
+        # carried `docPayload?: {pid, docId, name, mime, data_b64}` since C5, and `Documents.tsx`
+        # matches it against the fetch it is waiting on — while this route has been answering
+        # `doc`, which `apiBridge.ts` deliberately drops. One object, emitted under the name the
+        # consumer looks for, so F's wiring needs no translation step to get wrong.
+        # ⚠ `doc` stays for one wave: nothing in the client reads it, but a gate might, and
+        # removing a key to save six bytes is not worth a red nobody predicted.
+        out["doc"] = out["docPayload"] = ctx.out.doc
+    if ctx.out.toast is not None:
+        out["toast"] = ctx.out.toast
+
+    # ── ⭐ owner item 2 (2026-08-03): THE NEW MEASURE COLUMN'S VALUES, ONE ROUND TRIP SOONER ──
+    #
+    # Creating a measure column cost the browser TWO sequential trips before a single number
+    # appeared: this one to persist the field, then a whole `/workspace` to compute it. The
+    # second cannot start until the first lands (the resolver reads the PERSISTED field), so the
+    # wait was structural, not slow code — the owner's "it takes some time for the data to
+    # populate". The values are computed here instead, immediately after the write, and ride
+    # this response.
+    #
+    # ⚠ IT COSTS NOTHING EXTRA TO COMPUTE. The expensive part is one DuckDB aggregate over the
+    # book, and `rt.measure_memo` is keyed on (pool stamp, scope, pool, measure, window) — so
+    # the `/workspace` re-read that still follows HITS the memo instead of doing this work. The
+    # query happens once either way; only its position moved.
+    #
+    # ⚠ NARROW ON PURPOSE. Gated to an actual measure-column write, so an overlay edit or a
+    # cohort add — the overwhelming majority of events — never pays for a second assembly.
+    #
+    # ⚠ AND IT IS A SHORTCUT, NOT A PATH. Any failure is swallowed: `WORKSPACE_STALE` still
+    # fires from `rerender`, and the re-read still delivers these values exactly as it does
+    # today. Nothing depends on this having worked.
+    if out["rerender"] and scope in ("customer", "cohort") and any(
+            isinstance(e, dict) and e.get("type") == "field_upsert"
+            and str(((e.get("field") or {}) if isinstance(e.get("field"), dict) else {})
+                    .get("key") or "").startswith("measure_")
+            for e in events):
+        try:
+            fresh = grid_assembly(session, scope=scope, consume_corrections=False)
+            out["derived"] = {str(pid): cells for pid, cells in fresh["derived"].items()}
+        except Exception:
+            pass
+    # ⚠ NOTHING IS INVALIDATED HERE, on purpose. The runtime cache holds ONLY the scope-shaped
+    # Odoo pool (see `routes_customers._pool_rows`), and no event on this route can change an
+    # Odoo column — Odoo is read-only. Everything an event DOES change (overlays, fields, views,
+    # folders, cohorts) is re-read from the store on the next request. An earlier version cleared
+    # `pool_cache` after an `overlay_patch`, which threw away an expensive Odoo pull to refresh
+    # data that was never in it.
+    return out
+
+
+# ── CONTRACT C1 (W36-T20): THE REGISTRY TOPICS' ROW READERS ───────────────────────────────────
+# ⭐⭐ R6 — *"EVERY database gets the same permission logic, always."* `core.perm_scope.scoped_table`
+# is the ONE door to any database's rows, and it cannot import a topic's pool builder: `core` never
+# imports up (`platform/ARCHITECTURE.md`) and these pools are built by `modules/` + `aios_grid`
+# behind this layer's per-tenant cache. So the app layer DECLARES its readers, exactly as
+# `routes_odoo_tables` declares connected tables to `user_tables.register_connected`.
+#
+# ⛔ REGISTERED HERE RATHER THAN IN `routes_customers`/`routes_products` because those two files
+# are outside wave 36's lane-C fence. The readers themselves are three lines each and call the
+# SAME `_pool_for` + `derive_pool_scope` pair those routes call, so there is no second pool and no
+# second scope derivation — only a second CALLER of the one that exists.
+#
+# ⚠ AND THE TOPIC ROUTES STILL HAVE THEIR OWN DOOR TODAY. Contract C1 says `apply_row_scope` +
+# `visible_fields` "move behind" `scoped_table`; moving `routes_customers.grid_assembly` and
+# `routes_products.scoped_pool` is booked as a PENDING row (mailbox/C.md, C-1) rather than done
+# here, because neither file is in this fence. What ships now is the arm the wave is load-bearing
+# on — every `ut_*` database, plus E's sandbox — and a topic arm that is REAL rather than stubbed,
+# so `scoped_table`'s topic leg is exercised by the product instead of only by a gate.
+def _topic_rt(st, module):
+    """The tenant runtime a topic pool must be built against, or a REPORTED refusal.
+
+    ⛔ A topic pool is per TENANT (`rt.pool_cache`), so `st=None` cannot be resolved to "the
+    default" without picking a tenant at random — which on this box is tenant #0's PRODUCTION
+    data. Standing rule 1's second sentence: say why, and say what to do instead.
+    """
+    if st is None:
+        import core.perm_scope as perm_scope
+        raise perm_scope.Unresolvable(
+            subject="rows", effect="unreadable",
+            cause=f"'{module}' is a registry topic whose pool is built per tenant and no tenant "
+                  f"runtime was passed",
+            recommendation="pass the session's runtime as `st=`. A topic pool cannot be "
+                           "resolved without knowing which tenant is asking")
+    return st
+
+
+def _customer_rows(table_key, user, st):
+    """`(fields, rows)` for the customer topic — the SAME derivation `_team_agent` uses."""
+    import aios_grid
+    import core.perm_scope as perm_scope
+    from routes_customers import _pool_for
+
+    rt = _topic_rt(st, table_key)
+    team_id, agent = perm_scope.derive_pool_scope(user, table_key)
+    return list(aios_grid.FIELDS), _pool_for(rt, team_id, agent)
+
+
+def _product_rows(table_key, user, st):
+    """`(fields, rows)` for the product topic. `consolidated=` follows the derived scope, so a
+    BU-pinned reader gets that BU's field contract rather than the consolidated one."""
+    import core.perm_scope as perm_scope
+    from routes_products import _pool_for, pd_fields
+
+    rt = _topic_rt(st, table_key)
+    team_id, _agent = perm_scope.derive_pool_scope(user, table_key)
+    return pd_fields(consolidated=team_id is None), _pool_for(rt, team_id)
+
+
+def _register_topic_rows():
+    """Declare both topic readers to C1. Called at import; returns the registered key set.
+
+    ⚠ THE KEYS ARE LITERALS AND THE ROUTE IMPORTS ARE INSIDE THE READERS, on purpose: this runs at
+    module import, and `from routes_products import MODULE` here would pull a sibling router in
+    before its own imports have settled. Every other cross-router reference in this file is lazy
+    for the same reason. The literals are held to their sources by `verify_scopes`, so they cannot
+    drift into naming a topic that does not exist.
+    """
+    import core.perm_scope as perm_scope
+
+    perm_scope.register_rows(_customer_rows, MODULE)
+    return perm_scope.register_rows(_product_rows, _PRODUCT_MODULE)
+
+
+#: ⚠ `_`-prefixed, because three functions in this file already bind the name `PRODUCT_MODULE`
+#: LOCALLY from `routes_products`. A module-level twin of that spelling would read as the same
+#: thing and be a different one — [[constant-two-features-share]] waiting to happen.
+_PRODUCT_MODULE = "product_data"
+
+_C1_ROW_SOURCES = _register_topic_rows()