platform/core/grid_events.py

"""core/grid_events.py — THE WRITE SEAM, lifted out of Streamlit (EXIT-1a, X1, 2026-07-30).

This is the grid's event handler: the code that takes an UNTRUSTED event object from the
browser component and decides what may be persisted. It was `app.py:_cl_handle_grid_event` /
`_cl_handle_one` and moved here as a PURE MOVE — no behavior change inside a move (MA principle
5). Every body below is the app.py body verbatim; the ONLY mechanical rewrites are the
Streamlit couplings the contract names:

    st.session_state['_cl_grid_seen_ids']   -> ctx.seen_ids       (adapter-owned dedup state)
    st.session_state['_cl_doc_payload']     -> ctx.out.doc        (EventResult.doc)
    st.session_state['_cohort_toast']       -> ctx.out.toast      (EventResult.toast)
    st.session_state['cl_table_workspace']  -> ctx.fallback_ws    (store-unavailable fallback)
    is_admin()  (the bare call inside doc_delete, and _cl_table_workspace's admin=None default)
                                            -> ctx.admin

There is NO `import streamlit` in this file and there must never be one — `verify_no_streamlit.py`
enforces that with a `sys.meta_path` block, which is the only proof that survives refactoring.
(That gate was `verify_seam.py` until EXIT-6 retired it with `app.py`; the block, its negative
control and the AST walk moved over intact.)

ONE ADAPTER, ONE HANDLER — it was TWO until 2026-08-04:
  * ⛔ the Streamlit adapter is GONE. `app.py::_cl_handle_grid_event` built an EventCtx whose
    `seen_ids` and `fallback_ws` were session-state-backed dicts and applied the Result to session
    state. EXIT-6 deleted it. The host-neutrality this file was written for is what let that
    happen without touching a line of the handler — which is the argument for the design, recorded
    at the moment it paid out.
  * the HTTP adapter — `aios-web/api` builds an EventCtx with `fallback_ws=None`. A None
    fallback with the store unavailable is a **503**, never a silent in-memory workspace: an
    API request has no durable session dict to fall back to, so pretending to persist would
    lose the write with a 200 on it.

⚠ LAYERING, stated honestly. `core/` is not supposed to import `modules/` (ARCHITECTURE §1's
one-directional rule). This handler genuinely straddles: it is the WRITE half of the customer
table and its persistence verbs live in `modules.customer_data` / `modules.cohort`. The
contracted path for the strangler is `core/grid_events.py` (both adapters import it from here),
so the import stays and the violation is RECORDED rather than hidden. It is the one file in
`core/` that points up; when EXIT-5 ports the surfaces, the persistence verbs are what move
down, and this note is the marker for that work.
"""
import datetime as dt
import json as _json_ev                 # wave 23 (C7): the json field's parse wall
from dataclasses import dataclass, field
from typing import MutableMapping, Optional

import core.store as store
import core.table_store as _tstore     # wave-9 I17: the shared-view authorisation predicates

#: ⭐ WAVE-27 item 5 (C7) — how many view ids one rail arrangement may name. Generous on purpose:
#: it bounds a hostile payload, not the user (the rail holds saved views + cohort projections +
#: shared views, and a busy Customer grid already carries dozens). An order longer than this is
#: not a person dragging rows.
MAX_VIEW_ORDER = 500
import core.user_tables as _ut_limits  # wave 23 (C7): MAX_JSON_CELL — one ceiling, one definer
import core.users as users
import harness.telemetry as _tel       # _tel.error() = THE error sink (all caught errors log)
import modules.cohort as cohort_mod
import modules.customer_data as cl_mod

#: C5 (owner item 12): customer documents. The cap is enforced HOST-side on the DECODED byte
#: count — the client checks the same number first, but a client check only spares the user a
#: pointless upload; this is what bounds what the tenant's dataset can be made to hold.
#: MOVED from app.py:1410–1412 with the handler (app.py re-imports these during the strangler).
_DOCS_KEY = 'customer_docs'
_DOC_MAX_BYTES = 5 * 1024 * 1024
_DOC_MAX_PER_ROW = 50

#: The props of a field def the CLIENT actually draws — the wave-6 no-blip comparison set.
#: Host-only stamps (createdBy, permissions) are deliberately absent: the client renders
#: nothing from its own copy of those, and a permissions refusal is answered explicitly.
_FIELD_ECHO_PROPS = ('label', 'type', 'note', 'options', 'colorCodeOptions',
                     'optionColors', 'max', 'formula', 'agg', 'scope', 'format', 'measure')


class StoreUnavailable(RuntimeError):
    """The store is down AND the caller offered no durable fallback (HTTP adapter → 503).

    The Streamlit adapter never sees this: it passes a session-state-backed `fallback_ws`, so
    its store-down path degrades to the same captioned in-memory workspace it always had. An
    API caller has no such dict, and the honest answer to "persist this" with nowhere to
    persist it is an error — not a 200 over a write that evaporates.
    """


@dataclass
class EventResult:
    """What the HANDLER needs to tell its adapter — the three things it cannot do itself.

    `rerender` is the old return value with its old meaning: "this page must re-render from
    server state", NOT "handled" (see `handle_one`'s docstring — the distinction is the
    difference between a snappy grid and a sluggish one).
    """
    rerender: bool = False
    doc: Optional[dict] = None       # doc_fetch's answer — the bridge has no response channel
    toast: Optional[str] = None      # add_to_list's confirmation


@dataclass
class EventCtx:
    """Everything the handler is allowed to know, and nothing about how it is being served.

    ⚠ `allowed_pids` is a SECURITY parameter: it is the pool this caller may touch at all, and
    every row-scoped event (overlay_patch, doc_*, cohort membership, add_to_list, memberPids)
    is intersected with it. `admin` and `uname` are likewise the HOST's statement of identity —
    never anything the browser said about itself.

    `seen_ids` is ADAPTER-OWNED dedup state (the event log resends its recent window on every
    emit, so consumed ids must stay consumed): Streamlit passes a session-state dict, HTTP
    passes the caller's per-session mapping.

    `fallback_ws` is the store-unavailable workspace home, the OUTER dict keyed by username
    (exactly what `st.session_state.setdefault('cl_table_workspace', {})` is). HTTP passes
    None, and None + store down raises StoreUnavailable (→ 503).

    `out` is the sink for the two parked answers, so `handle_one` can keep returning a bare
    bool exactly as it did in app.py while doc/toast still escape. (Additive amendment to X1's
    field list, 2026-07-30 — defaulted; recorded in the S1 mailbox.)
    """
    uname: str
    allowed_pids: frozenset
    fields: list
    measure_keys: frozenset = frozenset()
    resolved_ids: frozenset = frozenset()
    cohort_ids: frozenset = frozenset()
    measure_offer: tuple = ()
    visible_views: tuple = ()
    admin: bool = False
    #: ⛔ WAVE 15 C-PERM — the fields this caller's PERMANENT PERMISSIONS hide, transitive
    #: closure included (`core.perm_scope.hidden_keys`). A SECURITY parameter in the same class
    #: as `allowed_pids`, and here for the same reason: the read paths strip a hidden field from
    #: both wires, so a caller cannot SEE it — but without this they could still WRITE it by
    #: naming the key, because `overlay_keys` only asks whether the column exists. Empty means
    #: "nothing hidden", which is every legacy record and therefore every caller today.
    hidden_keys: frozenset = frozenset()
    scope_key: str = 'customer'
    #: Wave 16 C-TOPIC — the TABLE OPS this ctx reads and writes (a `core.table_store.make`
    #: object). None = the CUSTOMER table (`modules.customer_data.TABLE_OPS`), which keeps
    #: every existing caller byte-identical. The PRODUCT surface passes
    #: `modules.product_data.TABLE_OPS` so its views/fields/overlays land in the product
    #: bucket — never the customer one, whose pids are a different id space entirely.
    table: Optional[object] = None
    #: ⭐ WAVE 25 (R6b, closes DEBT D-16) — THE TENANT STORE HANDLE, and it is a RESIDENCY
    #: parameter in the same class as `allowed_pids` is a security one. Everything this handler
    #: could not reach through `ctx.table` used to go through the MODULE-LEVEL `core.store`,
    #: i.e. tenant #0's dataset repo: the documents family, `_store_of(ctx).available()`, and — through
    #: `_tops`' fallback — the whole customer and product table workspace. Owner, R6b:
    #: *"Tenant 0 should not intertwine anywhere."*
    #:
    #: None keeps the module default, which is byte-identical for tenant #0 (empty prefix,
    #: shared repo) and is what every non-HTTP caller still gets. The HTTP adapters pass
    #: `session.runtime`, so a Nurilab write resolves against Nurilab.
    st: Optional[object] = None
    seen_ids: Optional[MutableMapping] = None
    fallback_ws: Optional[MutableMapping] = None
    out: EventResult = field(default_factory=EventResult)

    def __post_init__(self):
        # A caller that forgets `seen_ids` gets per-call dedup rather than a crash — the SET is
        # what makes re-delivery idempotent, and its absence must not be the thing that decides
        # whether an event is processed at all.
        if self.seen_ids is None:
            self.seen_ids = {}


def _store_of(ctx):
    """THE store handle this ctx reads and writes through (wave 25, R6b / D-16). ONE resolver,
    so "whose repo does this land in" has exactly one answer per event.

    Order, and each step is deliberate: the ctx's own tenant handle · else the handle the TABLE
    OPS object was built with (`table_store.make(..., st=…)` keeps it on `.st`, which is how the
    `ut_` path has been tenant-correct since wave 18) · else the module default.

    ⛔ THE MODULE DEFAULT IS TENANT #0, AND THAT IS THE BUG D-16 NAMES — not a safe base case.
    It is kept because it is byte-identical for tenant #0 (empty prefix, shared repo) and every
    non-HTTP caller still relies on it; what changed is that the HTTP adapters now pass a real
    handle, so the fallback is no longer what serves a Nurilab request. `verify_scopes` proves
    that with a negative control: force this to fall through and the residency legs go RED.
    """
    st = getattr(ctx, 'st', None)
    if st is not None:
        return st
    tbl = getattr(ctx, 'table', None)
    inner = getattr(tbl, 'st', None) if tbl is not None else None
    return inner if inner is not None else store


def _blobs(ctx):
    """The BYTE store for this ctx — `_store_of`'s sibling, split out because it is NOT yet
    equivalent and pretending otherwise would hide a live residency hole.

    ⚠ `TenantRuntime` exposes `get/put/update/exists/available` and **no byte methods**
    (`harness/runtime.py`), so a tenant with its OWN repo has nowhere tenant-scoped to put a
    document's BYTES. The metadata half of D-16 is fixed here; the byte half needs
    `upload_bytes`/`download_bytes`/`delete_path` on the runtime, which is SESSION A's file —
    asked for in the mailbox and booked as debt rather than papered over with a fallback that
    reads as intentional. The moment those methods exist this resolver picks them up with no
    other change, which is why it is written as a capability test and not as a version check.
    """
    st = _store_of(ctx)
    return st if hasattr(st, 'upload_bytes') else store



def _ag_overview_id():
    """`aios_grid.IG_OVERVIEW_ID`, resolved defensively.

    Function-local import, this module's convention for the layer above it. Falls back to the
    literal rather than raising: a store blip must not turn "refuse to delete" into "allow", and
    an unguarded delete here is the one that reads as success and undoes itself.
    """
    try:
        import aios_grid as _ago
        return _ago.IG_OVERVIEW_ID
    except Exception:
        return 'tpl_overview'

def _tops(ctx):
    """The table-ops object this ctx operates on — the customer table unless the caller chose
    another topic. ONE resolution point so 'which bucket does this write land in' has exactly
    one answer per event.

    ⛔ WAVE 25 (R6b) — THIS FALLBACK IS STILL TENANT #0's, DELIBERATELY, AND IT IS THE ONE PIECE
    OF THE RESIDENCY FIX THAT DID NOT LAND. `cl_mod.TABLE_OPS` is a module-level singleton built
    with no store handle, so the customer and product table workspaces (views, fields, overlays,
    folders) resolve to tenant #0 for every caller. Scoping it here was written, measured and
    then REVERTED, because it would have been worse than the bug:

        the WRITE door (`routes_grid._ctx`) is this session's file and now carries a handle;
        the READ door is `routes_customers._ctx_for` / `routes_products` — files in NO session's
        fence this wave — and they carry none.

    Moving only the write side means a prefix tenant saves a view into `t/<slug>/…` and reads
    back from the unprefixed bucket forever: their own view vanishes the moment they save it.
    A latent split is worse than a stated residency error, and read and write must move in ONE
    change. Booked as an ask + debt with that exact shape. Tenant #0 is unaffected either way
    (empty namespace, default repo), which is why nothing today would have shown it.

    ⛔ Untouched by this wave regardless: the EMIT GUARD is D-40, deferred by ruling A-15.
    """
    if ctx.table is not None:
        return ctx.table
    return cl_mod.TABLE_OPS


def _cohorts(ctx):
    """The COHORT store this ctx operates on — `_tops`'s sibling, and here for the same reason.

    ⭐ WAVE 19 / R9: a cohort belongs to its database. Every cohort verb below used to name
    `cohort_mod` directly, so a list built on the Product surface landed in `customer_cohorts`
    beside somebody's real customer lists — with PRODUCT pids in it (CRC32 hashes of SKU codes),
    which the customer page would then try to resolve against partner ids. Wave 16 answered that
    by REFUSING cohorts on other topics (`with_cohorts=False`); R9 says scope-parameterize
    instead. `modules.cohort.scoped` returns the module itself for the customer topic, so the
    customer path is byte-unchanged.
    """
    return cohort_mod.scoped(ctx.scope_key)


#: The keys a stored document entry may put on the wire. ⛔ `path` IS NOT ONE OF THEM — it is the
#: byte address inside the tenant's repo, of no use to a browser and a needless disclosure of the
#: store's layout. `canDelete` is not stored either: it is a fact about the READER, computed per
#: request below, so one person's answer can never be cached into another's.
_DOC_WIRE_KEYS = ('id', 'name', 'mime', 'size', 'by', 'ts')


def docs_for(pids, scope_key='customer', uname='', admin=False, st=None):
    """`{"<pid>": [CustomerDoc, ...]}` for a SCOPED row set — the producer `payload.docs` lost.

    ⭐⭐ WAVE 30 / CONTRACT C4 (D-138). The write door has been live and UNREACHABLE since EXIT-6:
    `doc_add`/`doc_fetch`/`doc_delete` all still work, and the comment inside that handler says
    exactly why nobody noticed — *"the React shell cannot reach it yet (`payload.docs` is set only
    by app.py)"*. `app.py` was deleted, and the producer went with it. Every client half survived:
    `RecordDetail`, `SwipeView` and `Documents` are complete, and all six `onDoc*` handlers in
    `CustomerGrid.tsx` are written as `payload?.docs ? … : undefined`, so an ABSENT `docs` key is
    what has been switching the whole feature off. An empty dict is truthy and turns it on with an
    honest empty state.
    ⛔ THIS IS THE SIXTH INSTANCE OF THE REACHABILITY CLASS and the one `verify_reachability.py`
    cannot see — a payload key assembled through helpers and dict merges is not a text-matchable
    literal ([[reachable-is-not-the-same-as-built]], [[artifact-with-no-importer]]).

    ⛔ ONE READER, NOT A MATCHING PAIR. `routes_tables` (ut_*) and `routes_customers` (the customer
    scope) both call THIS — they do not each serialise. Two surfaces answering one question
    separately is how the same defect got reintroduced in the opposite direction inside a single
    commit ([[one-question-two-normalizers]]).

    ⚠ `pids` IS REQUIRED and is the row set the CALLER has already scoped — `shared_overlay.cells`'
    argument, for the same reason: there is no "every document in the tenant" call to reach for.
    A record with no documents gets `[]`, never a missing key, so a client can tell "none" from
    "not served".
    """
    wanted = {str(int(p)) for p in (pids or [])}
    if not wanted:
        return {}
    scope = str(scope_key or 'customer').strip().lower()
    key = _DOCS_KEY if scope in cohort_mod.LEGACY_SCOPES else f'{scope}_docs'
    try:
        stored = (st if st is not None else store).get(key) or {}
    except Exception:                                                 # noqa: BLE001
        stored = {}                       # a display read degrades; it never breaks the grid
    out = {}
    for pid in wanted:
        row = stored.get(pid)
        docs = []
        for d in (row if isinstance(row, list) else []):
            if not isinstance(d, dict) or not d.get('id'):
                continue
            wire = {k: d.get(k) for k in _DOC_WIRE_KEYS}
            # The delete wall, mirrored onto the wire so the button is absent rather than
            # offered-and-refused. `doc_delete` enforces the same test server-side; this is the
            # DISPLAY of that rule, never a substitute for it.
            wire['canDelete'] = bool(admin or (d.get('by') and d.get('by') == uname))
            docs.append(wire)
        out[pid] = docs
    return out


def _docs_key(ctx):
    """The DOCUMENTS bucket this ctx operates on — `_cohorts`'s sibling (wave 19, item 12's
    silent third case).

    ⛔ `_DOCS_KEY` STAYS the module constant with its shipped value: app.py re-imports it for the
    Streamlit docs payload and `verify_seam.py` asserts the two resolve to the same string. This
    resolves the CUSTOMER topic to exactly that constant and gives every other database its own
    bucket, so the pid keys inside one bucket all come from one id space.
    """
    scope = str(getattr(ctx, 'scope_key', '') or 'customer').strip().lower()
    if scope in cohort_mod.LEGACY_SCOPES:
        return _DOCS_KEY
    return f'{scope}_docs'


# ------------------------------------------------------------------ the workspace family (X1 amd.)
def table_workspace(ctx, allowed_pids=None, consume_corrections=True):
    """Durable Airtable-style view/schema/overlay state, with an honest session fallback.

    Wave-9 I17: the returned `views` are this user's OWN views MERGED with every view SHARED
    with them. Own views win a same-id collision — a personal view is the user's own object and
    must never be shadowed by somebody else's share.

    ⚠ `allowed_pids` is a SECURITY parameter, not a convenience. A shared view carries
    `memberPids`, which the writer's scope validated on the way IN — but the READER may have a
    narrower scope. Handing a Fisch-scoped user a view whose member list was built by a
    full-access user would leak the other BU's customers straight through the strict-isolation
    rule. So membership is re-filtered against the READER on the way OUT. Pass it wherever the
    result reaches a renderer or an event handler; omitting it drops memberPids entirely rather
    than trusting the stored list (fail-closed: a smaller view, never a wider one).

    MOVED from app.py:5323 (`_cl_table_workspace`). `admin` was a parameter defaulting to a live
    `is_admin()` call; it now always comes from `ctx.admin`, which is identical at the one call
    site because app.py computes `admin = bool(is_admin())` once per render. `allowed_pids`
    stays an EXPLICIT argument because the folder leg deliberately omits it.
    """
    uname = ctx.uname
    if _store_of(ctx).available():
        ws = _tops(ctx).workspace(
            uname, consume_corrections=bool(consume_corrections))
        try:
            shared = _tops(ctx).shared_views(uname, bool(ctx.admin))
        except Exception as _se:
            _tel.error('sharedviews:read', _se)
            shared = {}
        # ⭐ WAVE 21 (item 9, C1): the R10 registry's named-user grants, PROJECTED at last.
        # Wave 20 wrote view grants and no reader ever consulted them — "Share view…" recorded
        # a row and changed nothing on the receiver's screen.
        try:
            granted = _granted_views(ctx, uname)
        except Exception as _ge:
            _tel.error('sharedviews:granted', _ge)
            granted = {}
        if shared or granted:
            merged = dict(granted)
            merged.update(shared)                       # the everyone-bucket outranks a grant
            merged.update(ws.get('views') or {})        # own views win the id collision
            ws['views'] = {vid: scope_view(v, uname, allowed_pids)
                           for vid, v in merged.items()}
        return ws
    # Store down. Streamlit degrades to its captioned session workspace exactly as before; an
    # API caller has no durable dict and must be told, not quietly served memory.
    if ctx.fallback_ws is None:
        raise StoreUnavailable('table workspace unavailable: no store and no session fallback')
    return ctx.fallback_ws.setdefault(
        uname, {'views': {}, 'fields': {}, 'overlays': {}})


def _granted_views(ctx, uname):
    """{view_id: STAMPED view} for every view GRANTED to `uname` by name (wave 21, item 9, C1).

    The R10 registry stores bare ids, tenant-scoped through the SAME store handle this topic's
    bucket uses; the record itself is fetched out of the OWNER's personal stratum of THIS
    bucket. An id granted on another topic simply finds nothing here and contributes nothing
    (fail-closed). Each projected view is stamped `shared` / `sharedRole` / `owner` so the
    client can synthesise the "Shared with me" group and disable edit affordances for
    role=view — the stamps live on the PROJECTION only, never written back to the store.
    `role_for` is asked with is_admin=False deliberately: the stamp reports the GRANT (the
    registry listed only explicit entries), not the caller's rank.
    """
    import core.shares as shares
    tops = _tops(ctx)
    st = getattr(tops, 'st', None)
    ids = (shares.shared_with(uname, kind='view', st=st) or {}).get('view') or []
    out = {}
    for vid in ids:
        found = tops.find_view(vid) if hasattr(tops, 'find_view') else None
        if not found:
            continue
        owner, view = found
        if owner == uname:
            continue
        role = shares.role_for('view', vid, uname, st=st)
        if role not in ('view', 'edit'):
            continue
        v = dict(view)
        v['shared'] = True
        v['sharedRole'] = role
        v['owner'] = owner
        out[str(vid)] = v

    # ⭐ D-37 (2026-08-05) — THE FOLDER LEG. W20's R10 said "a folder share = its views ride
    # along"; W21 wired the view kind only, so a granted folder was accepted, stored and listed
    # under "Shared with me" while changing NOTHING on the receiver's screen. Two waves of a
    # control that reported success and did nothing — the same defect class the view leg above
    # was written to close, surviving in its sibling.
    #
    # ⚠ A DIRECT VIEW GRANT WINS over the folder it happens to sit in, and the ordering above is
    # what implements that: the direct pass has already filled `out`, so the folder pass skips
    # ids it finds there. The specific grant is the more deliberate statement of intent, and it
    # may carry a DIFFERENT role — a folder shared read-only must not downgrade a view somebody
    # was explicitly given edit on, nor upgrade one they were not.
    try:
        folder_ids = (shares.shared_with(uname, kind='folder', st=st) or {}).get('folder') or []
    except Exception:
        folder_ids = []
    for fid in folder_ids:
        found = tops.find_folder(fid) if hasattr(tops, 'find_folder') else None
        if not found:
            continue                      # a folder on another topic contributes nothing here
        owner, folder, inside = found
        if owner == uname:
            continue
        role = shares.role_for('folder', fid, uname, st=st)
        if role not in ('view', 'edit'):
            continue
        for vid, view in inside.items():
            if str(vid) in out:
                continue
            v = dict(view)
            v['shared'] = True
            v['sharedRole'] = role
            v['owner'] = owner
            # The folder's NAME rides so the client can group these together later; today it
            # renders in the same "Shared with me" group the view leg feeds. Stamps live on the
            # PROJECTION only and are never written back, exactly as above.
            v['sharedFolder'] = str(folder.get('name') or '')[:80]
            out[str(vid)] = v
    return out


def scope_view(view, uname, allowed_pids):
    """Re-scope a view to the READER. A no-op for a view the reader created.

    Two things travel inside a shared view that are SCOPE, not presentation:
      1. `memberPids` — a customer list. Re-filtered against the reader's permitted pool.
      2. cohort conditions — a cohort id is a permission handle owned by ONE user. For a reader
         who does not own it the leaf is UNANSWERABLE, and partial evaluation of an unanswerable
         leaf WIDENS `is none of` ([[cg-cohort-set-conditions]]) — i.e. it would show MORE rows,
         not fewer. So the whole filter tree is dropped for a foreign reader rather than
         evaluated with a hole in it. A visibly unfiltered shared view is recoverable; a
         silently widened one is a leak.
    """
    if not isinstance(view, dict) or view.get('createdBy') == uname:
        return view
    out = dict(view)
    cfg = dict(out.get('config') or {})
    mp = cfg.get('memberPids')
    if isinstance(mp, list):
        cfg['memberPids'] = ([p for p in mp if p in (allowed_pids or ())]
                             if allowed_pids is not None else [])
    if tree_has_cohort(cfg.get('filters')):
        cfg['filters'] = []
        cfg['_scopedOut'] = 'cohort'      # the client says so rather than pretending
    out['config'] = cfg
    return out


def tree_has_cohort(node):
    """True when a filter tree references the Cohorts column anywhere (any depth)."""
    import aios_grid as _agc
    col = _agc.COHORT_COLUMN
    if isinstance(node, list):
        return any(tree_has_cohort(n) for n in node)
    if not isinstance(node, dict):
        return False
    if node.get('colId') == col or node.get('field') == col:
        return True
    return tree_has_cohort(node.get('conditions') or node.get('children'))


def measure_leaves(nodes):
    """Every leaf of a filter tree that carries a date WINDOW — i.e. every measure condition.

    Keyed off the window rather than the column name on purpose: a window is the one thing no
    column condition has, so this cannot mistake `revenue_ytd > 5000` for a measure.
    """
    out = []
    for node in nodes or []:
        if not isinstance(node, dict):
            continue
        if isinstance(node.get('children'), list):
            out.extend(measure_leaves(node['children']))
        elif node.get('window') is not None:
            out.append(node)
    return out


# ------------------------------------------------------------------ private handler helpers
def _field_echo_equal(accepted, raw):
    """Does the ACCEPTED def match what the client already renders? (wave-6 item 3)

    The no-blip law: a field_upsert whose acceptance changed NOTHING the client draws needs no
    re-render — the client is the ORIGIN of the change, its optimistic copy is already right,
    and the rerun it would buy costs a full payload re-serialisation plus an iframe remount
    (the "blip" the owner named). STRICT per-prop equality after collapsing the empty forms
    ('' / [] / {} / absent all render identically): any normalisation a validator applied is a
    real difference the client must be SHOWN. Fail-closed — unsure means not equal means rerun.
    """
    empty = ('', [], {})
    for k in _FIELD_ECHO_PROPS:
        a, b = accepted.get(k), raw.get(k) if isinstance(raw, dict) else None
        if k == 'scope':
            # 'global' and absent are the SAME rendering (an unscoped field) — the event may
            # say 'global' explicitly while the stored def says it by omission.
            a = None if a == 'global' else a
            b = None if b == 'global' else b
        if (None if a in empty else a) != (None if b in empty else b):
            return False
    return True


def _unique_name(wanted, existing, *, fallback='Untitled', max_len=120):
    """Return a durable, case-insensitively unique field/view display name."""
    limit = max(1, int(max_len))

    def _clean(value):
        return ' '.join(str(value or '').split())

    base = (_clean(wanted) or _clean(fallback) or 'Untitled')[:limit].rstrip()
    taken = {_clean(value).casefold() for value in existing if _clean(value)}
    if base.casefold() not in taken:
        return base
    index = 2
    while True:
        suffix = f' {index}'
        stem = base[:max(0, limit - len(suffix))].rstrip()
        candidate = f'{stem}{suffix}' if stem else str(index)[-limit:]
        if candidate.casefold() not in taken:
            return candidate
        index += 1


def _known_usernames():
    """Real account names, for C4's fail-closed `users` grant. None = "could not resolve".

    None means DO NOT FILTER, deliberately. Returning an empty set on a transient registry
    error would silently collapse a user's stored 'specific users' grant to 'personal' on the
    very next autosave — destroying a setting because a read blipped. The grant is inert today
    anyway (views are per-user; nothing reads it to allow cross-user access), so the honest
    trade is: never invent access, never destroy a stored intent either.
    """
    try:
        reg = set(users.registry() or {})
    except Exception as _ue:
        _tel.error('viewperms:users', _ue)
        return None
    # ⚠ EMPTY IS "UNRESOLVED", NOT "NOBODY EXISTS". registry() returns {} rather than raising
    # when the user store cannot be read (a store blip logs `store:get:users` and hands back an
    # empty dict) — and a working app ALWAYS has at least the bootstrap admin. Treating {} as a
    # real answer would filter every name out of a stored grant and collapse it to 'personal'
    # on the next autosave. Caught by testing this in BARE mode with no store token.
    return reg or None


# ------------------------------------------------------------------ THE SEAM
def handle_events(events, ctx):
    """The event LOG entry point → one EventResult. The `[-24:]` window lives here.

    ⚠ THE VALUE IS AN EVENT LOG `{'events': [...]}`, NOT ONE EVENT (2026-07-27). A component
    has exactly one value slot, and two quick emits share it: creating a field emits
    `field_upsert` and the insertColumn autosave emits `view_upsert` 420ms later — with the
    server still inside the first rerun, the second write clobbered the slot before any run
    read it, and the field creation VANISHED with no error anywhere (measured live; the same
    race sat latent under overlay-field creation, masked by localStorage). The client now
    sends its recent window of events; each is processed exactly once, keyed by its id. The
    single-event shape is still accepted — it is what an older client in a stale tab sends.

    Accepts everything the component value can be: the log dict, a bare list of events (what
    the HTTP body's `events` array unwraps to), or one legacy single-event object.

    `EventResult.rerender` is True when ANY processed event demands a re-render from server
    state — the same OR-fold `_cl_handle_grid_event` did.
    """
    if isinstance(events, dict) and isinstance(events.get('events'), list):
        seq = events['events'][-24:]
    elif isinstance(events, list):
        seq = events[-24:]
    else:
        seq = [events]
    rerun = False
    for one in seq:
        rerun = handle_one(one, ctx) or rerun
    ctx.out.rerender = rerun
    return ctx.out


def handle_one(event, ctx):
    """Validate ONE untrusted component event.

    Odoo-source fields are never accepted as cell updates. View and schema objects are
    user-scoped; row patches additionally require the pid to be in the currently permitted
    Customer List pool.

    RETURN VALUE = "this page must re-render from server state", NOT "handled".
    The distinction is the difference between a snappy grid and a sluggish one.
    Persisting is not by itself a reason to rerun: Streamlit has ALREADY re-run the
    script once simply to deliver the component value, and the client is the ORIGIN
    of the change, so its own state is correct before the server hears about it. An
    extra st.rerun() therefore buys nothing and costs a second full page run.
    Measured on the 1,550-row book (2026-07-25): one filter tweak produced TWO
    page_customer_data runs ~2.8s apart, each re-serialising the whole ~576 KB row
    payload — and the handler on the second run is a guaranteed no-op, because the
    event-id dedupe two lines below has already consumed that id.

    THE WAVE-6 NO-BLIP LAW (item 3, 2026-07-27) — True ONLY when the next render
    genuinely differs from what the client already draws:
      field_upsert    — False when the accepted def ECHOES the client's copy
                        (_field_echo_equal); True on any validator mutation, any
                        permissions refusal, and ALWAYS when a measure window changed
                        (values must be re-resolved server-side).
      overlay_patch   — False when every requested update was accepted verbatim;
                        True the moment anything was refused or truncated.
      field_delete    — False: the client removed the column optimistically and the
                        store scrub is invisible to the next render.
      field_duplicate — True only when host-held VALUES must flow back (custom_ cell
                        copies, measure_ recomputes); formula/created_time clones
                        compute client-side.
      view_delete     — False (the deleted view simply stops being served).
      view_upsert     — unchanged: False, except an unresolved measure condition.
      cohort_* / add_to_list — True: the lists payload is server truth, and these are
                        not the hot path.

    The id dedupe is a SET (not a last-id slot): the event log resends the whole recent
    window on every emit, so every id must stay consumed, not just the latest one.
    """
    # The ctx fields the moved body reads as plain locals — bound once here so every branch
    # below is textually the app.py body (which is what makes this move reviewable).
    uname = ctx.uname
    allowed_pids = ctx.allowed_pids
    fields = ctx.fields
    measure_keys = ctx.measure_keys
    resolved_ids = ctx.resolved_ids
    cohort_ids = ctx.cohort_ids
    measure_offer = ctx.measure_offer
    visible_views = ctx.visible_views
    admin = ctx.admin
    scope_key = ctx.scope_key

    if not isinstance(event, dict):
        return False
    event_id = str(event.get('id') or '')[:180]
    seen = ctx.seen_ids
    if not event_id or event_id in seen:
        return False
    seen[event_id] = True
    if len(seen) > 400:                            # a session-lifetime set, kept bounded
        for stale in list(seen)[:200]:
            seen.pop(stale, None)
    kind = event.get('type')
    field_by_key = {f['key']: f for f in fields}
    valid_keys = set(field_by_key)
    overlay_keys = {f['key'] for f in fields if f.get('source') == 'overlay'}
    # Event validation is not a render payload. It must not consume a one-shot label
    # correction before the browser has had a chance to receive it.
    ws = table_workspace(
        ctx, allowed_pids=allowed_pids, consume_corrections=False)

    def _session_ready():
        ws.setdefault('views', {})
        ws.setdefault('fields', {})
        ws.setdefault('overlays', {})

    if kind == 'view_select':
        # Owner item 3 (2026-07-31): remember WHERE THE USER IS so a fresh browser resumes
        # there instead of the system default view. Presentation state only — the read side
        # re-validates the id against what the caller may see, so nothing here needs an
        # authorisation wall beyond the length cap. Never a re-render: the client is already
        # on that view.
        vid = str(event.get('viewId') or '').strip()[:120]
        if vid and _store_of(ctx).available():
            try:
                _tops(ctx).save_active_view(uname, vid)
            except Exception as _ae:
                _tel.error('grid:view-select', _ae)
        return False

    if kind == 'record_layout':
        # Wave 2026-08-02 (C-LAYOUT): the per-user record-detail field order. Presentation
        # state for one surface; grid column order is untouched. Unknown keys are pruned
        # against the LIVE field set, dupes collapse, and the wire re-validates at serve
        # time — so a client racing a field delete degrades to default order, never an error.
        raw_order = event.get('order')
        if not isinstance(raw_order, list):
            return False
        order, seen = [], set()
        for k in raw_order[:200]:
            k = str(k or '').strip()[:80]
            if k and k in valid_keys and k not in seen:
                seen.add(k)
                order.append(k)
        if _store_of(ctx).available():
            _tops(ctx).save_record_layout(uname, order)
        else:
            _session_ready()
            if order:
                ws['recordLayout'] = {'order': order}
            else:
                ws.pop('recordLayout', None)
        # Autosave hot path, like view_select: the client already renders the order it asked
        # for, and the wire echoes it back on the next workspace read.
        return False

    if kind == 'view_upsert':
        raw = event.get('view')
        if not isinstance(raw, dict) or not isinstance(raw.get('config'), dict):
            return False
        view_id = str(raw.get('id') or '').strip()[:120]
        name = str(raw.get('name') or '').strip()[:120]
        requested_name = name
        if not view_id or not name:
            return False
        cfg = raw['config']
        # The filter TREE (leaf conditions and/or nested condition groups) is
        # validated by the module-agnostic grid contract helper — imported lazily
        # here for the same reason the page does it: a missing embed build must
        # never break page import.
        import aios_grid as _ag
        # A MEASURE key is valid in a FILTER and nowhere else (CG-8): it may not be ordered,
        # shown, grouped or coloured by, because it is not a column of this table. Hence the
        # widened key set here and the plain `valid_keys` everywhere below.
        # `cohort_ids` is a PERMISSION list, not a convenience: a cohort rule naming anything
        # else is dropped here. Leaving it in would persist a condition the engine can only
        # answer with "nothing", so `Cohort is not [a list I deleted]` would show an empty table
        # forever with nothing on screen explaining why.
        filters = _ag.clean_filter_tree(cfg.get('filters'),
                                        valid_keys | set(measure_keys),
                                        cohort_ids=set(cohort_ids))
        sorts = []
        for rule in list(cfg.get('sorts') or [])[:20]:
            if (isinstance(rule, dict) and rule.get('colId') in valid_keys
                    and rule.get('dir') in ('asc', 'desc')):
                sorts.append({'colId': rule['colId'], 'dir': rule['dir']})
        order = list(dict.fromkeys(
            key for key in list(cfg.get('order') or []) if key in valid_keys))
        order += [field['key'] for field in fields if field['key'] not in order]
        visible = list(dict.fromkeys(
            key for key in list(cfg.get('visible') or []) if key in valid_keys))
        widths = {}
        raw_widths = cfg.get('widths') if isinstance(cfg.get('widths'), dict) else {}
        for key, value in raw_widths.items():
            if key in valid_keys:
                try:
                    widths[key] = max(60, min(600, round(float(value))))
                except (TypeError, ValueError):
                    pass
        # C-LOCK (wave 2026-08-02, item 12's mechanism): a view may be LOCKED to one cohort.
        # `cohort_ids` is the same permission list the cohort filter leaf uses — a lock naming
        # a cohort this session cannot see is DROPPED at write, exactly like a cohort rule.
        # (Readers of a SHARED locked view resolve the id against their OWN visible sets; an
        # unresolvable lock matches NOTHING client-side — the cohort-leaf law, never a widen.)
        cohort_lock = str(cfg.get('cohortLock') or '').strip()[:120]
        # ⛔ WAVE 17 R1 (C-LOCKV guard a) — A PROJECTED LOCKED VIEW RE-STAMPS ITS OWN LOCK.
        # A locked view IS its set: its id and its lock are the same fact. The client rebuilds
        # `config` from scratch on every autosave (a column resize is enough), so taking the
        # lock from the browser would mean one omitted key turns "the 40 accounts we agreed to
        # call" into the whole book, under the same name, with nothing going red. The read side
        # re-stamps too (aios_grid.views_from_defs) — this keeps the STORE from accumulating the
        # wrong record in the first place.
        _is_locked_view = view_id in set(cohort_ids)
        if _is_locked_view:
            cohort_lock = view_id
        config = {
            'filters': filters,
            # conjunction joining the ROOT-level conditions; anything but an
            # explicit 'or' means 'and' (legacy views had an implicit AND).
            'filterConj': 'or' if cfg.get('filterConj') == 'or' else 'and',
            **({'cohortLock': cohort_lock}
               if cohort_lock and cohort_lock in set(cohort_ids) else {}),
            # ⭐⭐ WAVE 32 · R5 / CONTRACT C4 — "Mark important" (raised by SESSION C as ASK C-12).
            #
            # ⛔ WITHOUT THIS LINE THE MARK IS A LIE THAT SURVIVES UNTIL RELOAD. This dict is
            # REBUILT FROM AN ALLOWLIST on every autosave — a column resize is enough — so a key
            # the client sends and this literal does not name is silently dropped. The badge would
            # paint, the write would answer 200, and the mark would be gone on the next read: the
            # exact shape T24's done-when says "persists across a reload" to catch, and
            # [[a-migration-that-runs-on-the-next-write]] on a write path.
            #
            # ⚠ UNCONDITIONAL, NOT SPREAD-CONDITIONAL LIKE `cohortLock` ABOVE, and the asymmetry is
            # the whole point: R5's mark must be REMOVABLE. A key written only when present means
            # unmarking sends `important: false`, the allowlist drops it, and the stored `true`
            # survives — a mark you can set and never clear.
            # ⚠ `is True`, not truthy: the client sends a real boolean, and a string `"false"` must
            # not become a mark.
            # ⚠ NO PERMISSION GATE, deliberately. R5 makes this a personal legibility mark, not a
            # lock; `_may_administer` already decides who may write this view at all, and adding a
            # second wall here would be the two-normalizers shape this wave is already fixing
            # elsewhere.
            'important': cfg.get('important') is True,
            'sorts': sorts,
            'groupBy': cfg.get('groupBy') if cfg.get('groupBy') in valid_keys else None,
            'colorBy': cfg.get('colorBy') if cfg.get('colorBy') in valid_keys else None,
            'rowHeightMode': (cfg.get('rowHeightMode')
                              if cfg.get('rowHeightMode') in ('short', 'medium', 'tall')
                              else 'short'),
            'order': order,
            'visible': visible,
            'widths': widths,
            'memberPids': [pid for pid in list(cfg.get('memberPids') or [])
                           if isinstance(pid, int) and pid in allowed_pids],
        }
        # Wave-6 item 10: a non-grid display mode rides the view config, validated
        # structurally (mode whitelist, refs must be this table's fields).
        display = _ag._clean_display(cfg.get('display'), valid_keys)
        if display:
            config['display'] = display
        # Wave-6 item 11 (pin): how many leading columns stay frozen while scrolling
        # sideways. Clamped 1..8 — 0 would unfreeze the identity column the whole
        # surface pivots on; absent = the legacy single frozen column.
        if cfg.get('frozenCount') is not None:
            try:
                config['frozenCount'] = max(1, min(8, int(cfg['frozenCount'])))
            except (TypeError, ValueError):
                pass
        # ── wave-9 C3 (lock) + C4 (permissions) ────────────────────────────────────────────
        # The PRIOR stored view decides both: whether this is a create or an update (C4's
        # default splits on exactly that) and who is allowed to change the lock (C3).
        prior = (ws.get('views') or {}).get(view_id) or {}
        # ⚠ THE SHARED WALL (wave-9 I17). `ws` is already visibility-filtered, so a view the
        # caller may not SEE reads as absent — which would look like a CREATE and let anybody
        # overwrite somebody else's shared view by guessing its id. Ask the store for the raw
        # record and refuse before deciding anything else. Fail-closed on both branches.
        _raw_shared = _tops(ctx).shared_view(view_id) if _store_of(ctx).available() else None
        if _raw_shared is not None and not prior:
            return False                       # exists, not visible to this caller: refuse
        if _raw_shared is not None and not _tstore._may_edit(_raw_shared, uname, admin):
            return False                       # visible but not editable by this caller
        is_new = not prior
        # C4: the creator is HOST-STAMPED and never read from the browser — the same rule as
        # the field-level createdBy. An existing view keeps whoever made it.
        creator = prior.get('createdBy') or uname
        # Only the creator or an admin may change the PERMISSIONS themselves. A collaborator who
        # could rewrite them could grant themselves sole ownership of someone else's view, or
        # quietly widen a users-scoped view to everyone — privilege escalation by edit.
        if not is_new and not _tstore._may_administer(prior, uname, admin):
            raw = dict(raw)
            raw['permissions'] = prior.get('permissions')
            raw['locked'] = prior.get('locked')
        # ── I12/C3 BELT (2026-07-31): a LOCKED view's DISPLAY MODE is frozen HOST-SIDE too.
        # The client's switcher already refuses (setDisplayMode), and its comment has claimed
        # "the host refuses too" since wave 9 — this makes that claim true for direct API
        # callers. ⚠ MIRRORS types.ts `isModeFrozen` EXACTLY: SYSTEM views and the undeletable
        # list:/cohort: projections are EXEMPT — their `locked` is legacy "undeletable", not
        # mode-frozen, and freezing them would trap the default view in whatever mode it last
        # saved (the owner's stuck-in-Map report is that trap, client-made). An upsert that
        # UNLOCKS in the same write (authorised above) may change display freely.
        _mode_frozen = (prior.get('locked') and raw.get('locked') is not False
                        and prior.get('kind') != 'system'
                        and view_id != 'all-customers')
        if _mode_frozen:
            if prior.get('config', {}).get('display'):
                config['display'] = dict(prior['config']['display'])
            else:
                config.pop('display', None)
        # ⚠ A SYSTEM VIEW IS PER-USER STATE AND MUST NEVER BE SHARED. Caught in the wave-9
        # integration pass, before deploy: `all-customers` is every user's DEFAULT view and it
        # carries their own column widths, visible fields and row height. C4's read-default
        # ('collaborative' when permissions are absent, so nobody's saved view is
        # retro-restricted) is right for a view somebody AUTHORED — but applied to the system
        # view it routed `all-customers` into the shared bucket, and the first user to resize a
        # column would have published their layout to everyone. Verified in the store:
        # shared_table_views() was returning ['all-customers'].
        # System views are pinned personal, so they stay in each user's own workspace exactly as
        # before this wave. `kind` is host-validated two lines below; read it here from `raw` and
        # treat the stored kind as authoritative for an existing view.
        _kind = (prior.get('kind') if not is_new
                 else (raw.get('kind') if raw.get('kind') in ('system', 'list', 'custom')
                       else 'custom'))
        if _kind == 'system' or view_id == 'all-customers':
            perms = {'edit': 'personal'}
        else:
            perms = _ag.clean_view_permissions(
                raw.get('permissions'),
                default='personal' if is_new else 'collaborative',
                known_users=_known_usernames())
        # C3: only the creator or an admin may SET or CLEAR the lock; anyone else keeps whatever
        # is already stored. Enforced here rather than hidden in the client. This composes with
        # C4 and does not override it: C4 governs who may edit the CONTENTS, C3 who may freeze
        # the display MODE.
        # ⚠ Inert in practice TODAY — a user can only ever write their own per-user workspace,
        # so `uname == creator` always holds. Correct plumbing for shared views, not a live gate.
        may_lock = admin or uname == creator
        locked = bool(raw.get('locked')) if may_lock else bool(prior.get('locked'))
        # C3: the LOCK's actual job — a locked view's display mode is frozen, so a locked Kanban
        # stays Kanban. This half IS live per-user and is the part the owner asked for. Filters,
        # sorts and columns stay editable by design.
        if _mode_frozen:
            was = (prior.get('config') or {}).get('display')
            now = config.get('display')
            if (was or {}).get('mode') != (now or {}).get('mode'):
                if was:
                    config['display'] = was
                else:
                    config.pop('display', None)
        # WAVE 17 R1 (C-LOCKV guard b) — ONE THING, ONE NAME. Renaming a locked view renames
        # the SET it is, in the store that owns it; the view record must not grow a second title
        # that the read side would then have to arbitrate. Routed rather than duplicated.
        if _is_locked_view and _store_of(ctx).available():
            _sets = _cohorts(ctx)                     # R9: the SET lives in this topic's bucket
            _cur = (_sets.all_for(uname).get(view_id) or {}).get('name')
            if name and name != _cur:
                try:
                    _sets.rename(uname, view_id, name)
                except ValueError as _e:
                    _tel.error('cohort:rename-via-view', _e)
        view = {'id': view_id, 'name': name,
                # ⚠ DERIVED FROM THE ID, never taken from the browser: `kind` is what paints the
                # lock mark, so a client claiming it for an ordinary view would be claiming a
                # guarantee the engine is not making about that view's rows.
                'kind': _ag.LOCKED_VIEW_KIND if _is_locked_view
                else raw.get('kind') if raw.get('kind') in ('system', 'list', 'custom')
                else 'custom',
                'locked': locked,
                'createdBy': creator,
                'permissions': perms,
                'note': str(raw.get('note') or '')[:2000],
                'config': config}
        # System/list projections are not all persisted in this table store, but custom views
        # may not impersonate their visible names. Persisted view names are allocated
        # tenant-globally by TableStore inside the write transaction.
        _reserved_view_names = [
            value.get('name') for value in visible_views
            if (isinstance(value, dict) and value.get('id') != view_id
                and value.get('kind') in ('system', 'list'))
        ]
        # Wave 21 (item 3, R6): the reserved SYSTEM name is PER-TOPIC — a ut_ database's system
        # view is "All records", and reserving the customer literal there uniquified a user's
        # honest "All customers" view name against a string that never appears on that topic.
        _scope = str(getattr(ctx, 'scope_key', '') or 'customer').strip().lower()
        _sys_name = ('All products' if _scope == 'product'
                     else 'All records' if _scope.startswith('ut_') else 'All customers')
        if view_id != 'all-customers':
            _reserved_view_names.append(_sys_name)
        # ⭐ WAVE 21 (item 9, C1) — THE FOREIGN-VIEW WALL. A granted view is projected into the
        # receiver's rail, so their autosave can now emit an upsert carrying an id they do not
        # own. Silently saving that into the RECEIVER's stratum would FORK the view (own views
        # win the merge, so the receiver sees their private copy forever and calls sharing
        # broken). Role `edit` (or admin) writes THROUGH to the owner's record — config only,
        # never identity or reach: `createdBy`/`permissions` are pinned from the stored record.
        # Role `view` (or no grant) refuses with a toast instead of forking.
        _save_as = uname
        if _store_of(ctx).available():
            _foreign = _tops(ctx).find_view(view_id)
            if _foreign and _foreign[0] != uname:
                _o_uname, _stored = _foreign
                if not admin:
                    import core.shares as _shares
                    if _shares.role_for('view', view_id, uname,
                                        st=getattr(_tops(ctx), 'st', None)) != 'edit':
                        ctx.out.toast = ('That view is shared with you read-only — ask its '
                                         'owner for edit access.')
                        return False
                view['createdBy'] = _stored.get('createdBy', _o_uname)
                if _stored.get('permissions') is not None:
                    view['permissions'] = _stored.get('permissions')
                else:
                    view.pop('permissions', None)
                _save_as = _o_uname
        if _store_of(ctx).available():
            view = (_tops(ctx).save_view(
                _save_as, view, reserved_names=_reserved_view_names, is_admin=admin) or view)
        else:
            _session_ready()
            _view_names = list(_reserved_view_names)
            _view_names.extend(
                value.get('name')
                for candidate_id, value in ws['views'].items()
                if candidate_id != view_id and isinstance(value, dict)
            )
            view['name'] = _unique_name(view.get('name'), _view_names)
            ws['views'][view_id] = view
        name_changed = view.get('name') != requested_name
        # PERSISTED, but deliberately NO rerun — see the return-value contract in
        # this function's docstring. This is the hot path: every filter, sort,
        # group, colour, column-resize and column-toggle autosaves through here.
        #
        # CG-8 is the ONE exception, and it is self-limiting. A measure condition is answered
        # server-side, so if this view carries one whose answer was NOT in the payload we just
        # shipped, the client is showing a pending marker with nothing coming. Ask for exactly
        # one more run — and only then, so the common case (a column filter) still costs zero.
        # If the top-of-run read of the component's own value already covered it, this is False
        # and nothing extra happens; if that read is ever stale, this is what stops the
        # condition hanging unresolved.
        #
        # ⚠ COMPLETE rules only (wave-6 item 3, found by the live roundtrip on this rev):
        # picking a measure FIELD mints a valueless rule whose autosave used to land here as
        # "unresolved" and trigger a rerun that could resolve NOTHING — `rule_complete` refuses
        # half-asked questions — so the only effect was a remount that closed the builder
        # popover mid-build (the wave-5 "gate flake" runs, now explained). A rule still being
        # typed stays client-side PENDING by design; the run that matters fires when its VALUE
        # arrives and the question becomes answerable.
        from harness import measure_filter as _mfc
        unresolved = {str(r.get('id') or '') for r in measure_leaves(filters)
                      if _mfc.rule_complete(r)} - set(resolved_ids)
        return name_changed or bool(unresolved - {''})

    if kind == 'view_delete':
        view_id = str(event.get('viewId') or '')[:120]
        # ⭐ WAVE-27 item 27 (R8) — `IG_OVERVIEW_ID` joins the system view here, and the comment
        # eight lines down already explains why in the LOCKED-view case: dropping the saved
        # config leaves the PROJECTION, which rebuilds the row on the very next render. Overview
        # is injected on every read of an Instagram database, so a delete would look like it
        # worked and the view would be back after a reload — which is precisely the
        # reads-as-success failure the client's own `UNDELETABLE_VIEW_IDS` note describes.
        if not view_id or view_id in ('all-customers', _ag_overview_id()):
            return False
        # WAVE 17 R1 (C-LOCKV guard c) — deleting a LOCKED view deletes the set it is. Without
        # this the store would drop the saved config and the projection would rebuild the view
        # on the very next render: the row would come back, minus the user's sort and columns,
        # which reads as a corruption rather than as a refusal. Own sets only (`all_for`), the
        # same wall `cohort_delete` rides.
        if _store_of(ctx).available() and view_id in _cohorts(ctx).all_for(uname):
            _cohorts(ctx).delete(uname, view_id)
            _tops(ctx).delete_view(uname, view_id)
            # True, unlike an ordinary view_delete: the lists payload IS server truth and the
            # client cannot know the membership channel changed.
            return True
        # Wave-9 I17: deleting a SHARED view is narrower than editing it — creator or admin
        # only. A collaborator can change what a shared view shows; they cannot destroy
        # everybody else's copy of it, because there is only one copy.
        if _store_of(ctx).available():
            _raw_shared = _tops(ctx).shared_view(view_id)
            if _raw_shared is not None and not _tstore._may_administer(
                    _raw_shared, uname, admin):
                return False
            # ⭐ WAVE 21 (item 9, C1) — the foreign wall, delete half. Contract: a grant of
            # `edit` may delete (there is only one copy and the owner handed over its content);
            # `view` or no grant refuses. The dead grant record is emptied afterwards so the
            # receiver's "Shared with me" heals instead of listing a ghost id forever.
            _foreign = _tops(ctx).find_view(view_id)
            if _foreign and _foreign[0] != uname:
                import core.shares as _shares
                _sst = getattr(_tops(ctx), 'st', None)
                if not admin and _shares.role_for('view', view_id, uname,
                                                  st=_sst) != 'edit':
                    ctx.out.toast = 'That view is shared with you read-only.'
                    return False
                _tops(ctx).delete_view(_foreign[0], view_id)
                try:
                    _shares.set_grants('view', view_id, [], st=_sst)
                except Exception:
                    pass
                return False
            _tops(ctx).delete_view(uname, view_id)
        else:
            _session_ready()
            ws['views'].pop(view_id, None)
        # Wave-6 item 3: the client removed the view optimistically; a deleted view simply
        # stops being served, so the next render differs in nothing the client still shows.
        return False

    if kind == 'field_upsert':
        raw = event.get('field')
        if not isinstance(raw, dict):
            return False
        key = str(raw.get('key') or '')[:80]
        import aios_grid as _ag2
        if key.startswith(_ag2.MEASURE_FIELD_PREFIX):
            # A FORMULA-MEASURE column (owner item 7). Validated against THIS caller's measure
            # offer — the same admission the condition builder uses — so a field can only name a
            # measure this user could also filter by. Fail closed on anything else.
            field = _ag2.clean_measure_field(raw, {m['key']: m for m in measure_offer})
            if field is None:
                return False
            fmt = _ag2._clean_format(raw.get('format'), field.get('type'))
            if fmt:
                field['format'] = fmt
        elif key in field_by_key and not key.startswith('custom_'):
            base = field_by_key[key]
            field = {**base, 'note': str(raw.get('note') or '')[:2000]}
            # ⭐ W29-T83 — THE COLUMN SUMMARY ON A SOURCE-BACKED COLUMN. This branch rebuilds the
            # accepted def from the CONTRACT and adds only what a user may override, and `agg` was
            # not in that list — so Average on Customer's `Overdue days` was accepted with a 200
            # and dropped here, before it ever reached the store. Validated against the shared
            # vocabulary rather than a literal (see the custom branch below for what a literal
            # costs). An ABSENT or unknown `agg` clears it: the client sends the whole field on
            # every edit, so a missing key is a user who unset the summary, and `{**base}` has
            # already re-applied whatever the contract itself declares.
            _agg = raw.get('agg')
            if _agg in _ag2.FIELD_AGGS:
                field['agg'] = _agg
            else:
                field.pop('agg', None)
            # Wave-5 item 10: a DISPLAY format on any base field (presets included) — how a
            # number/date reads, per user. Rendering-only; validated fail-closed per type.
            fmt = _ag2._clean_format(raw.get('format'), base.get('type'))
            if fmt:
                field['format'] = fmt
            else:
                field.pop('format', None)      # cleared -> back to the type's default render
            # PRESET measure fields (wave-2 item 8): the browser may change the PERIOD (and the
            # auto-renamed label) of revenue_ytd / revenue_ly / orders_24m — nothing else. The
            # whitelist IS the base contract: only a field that shipped with preset+measure can
            # take a window here, and the measure KEY is pinned to the base's (fail closed).
            if (base.get('preset') and isinstance(base.get('measure'), dict)
                    and isinstance(raw.get('measure'), dict)):
                window = _ag2._clean_window(raw['measure'].get('window'))
                if window is not None:
                    field['measure'] = {'key': base['measure']['key'], 'window': window}
                    if isinstance(raw.get('label'), str) and raw['label'].strip():
                        field['label'] = str(raw['label'])[:120]
        elif key.startswith('custom_') and raw.get('type') in _ag2.CUSTOM_FIELD_TYPES:
            ftype = raw['type']
            # ⭐⭐ 2026-08-09 — THE RELATIONAL PAIR HAS NO HOME IN A PER-USER OVERLAY, and this
            # door used to accept one silently. It builds the stored field from a fixed key list
            # and adds only the bags it knows (`options`, `max`, `formula`, `automation`, `code`)
            # — so a `link`/`rollup` arriving here was stored WITHOUT its bag, in one user's
            # stratum, where `compute_relation_cells` (which walks `user_tables` definitions)
            # never sees it. Measured on nurilab as `custom_video_views_90j26`: a Rollup column
            # that rendered, could not be filled by anything, and reported no error.
            # ⛔ REFUSED RATHER THAN REPAIRED. These two belong to the table's SHARED schema
            # (`POST /tables/{key}/fields`), which is where the client now sends them; storing a
            # second, private, non-computing copy here would be a second definition of the same
            # column. A refusal repaints, so a stale client says something rather than nothing.
            if ftype in ('link', 'rollup'):
                return False
            readonly = ftype in _ag2.READONLY_CUSTOM_TYPES
            # The editable family must SAY it is overlay (the client's own contract); the
            # read-only pair (formula / created_time) is stored under the cohort column's
            # mechanism instead, so no cell write can ever be accepted for it.
            if not readonly and raw.get('source') != 'overlay':
                return False
            field = {'key': key, 'label': str(raw.get('label') or 'Untitled')[:120],
                     'type': ftype, 'source': 'odoo' if readonly else 'overlay',
                     'default': True, 'custom': True,
                     'note': str(raw.get('note') or '')[:2000]}
            if readonly:
                field['derived'] = True
                # FILTERABLE since wave 6 (owner item 6): formula/created_time values live
                # client-side, and this table's counts are client-mode — the client engine
                # answers conditions on them soundly. Mirrors fields_from_workspace.
                field['filterable'] = True
            # ⛔ W29-T83 — THE VOCABULARY, NOT THE LITERAL `'sum'`. This is the write gatekeeper
            # for an overlay column's summary and it accepted exactly one name, while the READ
            # projection (`aios_grid.py:679`) passes all six — so a user who chose Median got a
            # 200, a menu that said Median, and no summary after the next reload. The same shape
            # as the `agg:"sum"` defect this repo already shipped, where one door kept it and the
            # other dropped it ([[read-path-cannot-witness-write-path]]).
            _agg = raw.get('agg')
            if _agg in _ag2.FIELD_AGGS and not readonly:
                field['agg'] = _agg
            # A `select`/`multiselect` carries its own vocabulary; a `user` does NOT — its
            # choices are the tenant's accounts, resolved fresh on every render, so persisting
            # a snapshot here would go stale the moment somebody joins or leaves.
            if ftype in ('select', 'multiselect'):
                field['options'] = _ag2._clean_options(raw.get('options'))
                if not field['options']:
                    return False        # a select with no choices can never hold a value
                field.update(_ag2._choice_appearance(raw, field['options']))
            if ftype == 'rating':
                field['max'] = _ag2._clean_rating_max(raw.get('max'))
            if ftype == 'formula':
                # WRITE-time is where refs must exist (fail closed); read-time leaves them
                # alone so a later-deleted ref blanks the CELLS, not the column.
                formula = _ag2._clean_formula(raw.get('formula'), valid_keys)
                if formula is None:
                    return False
                field['formula'] = formula
            if ftype == 'automation':
                # C5-AUTOFIELD (wave 18): persist the validated config bag. No `return False`
                # when absent — an unconfigured automation column is a legitimate intermediate
                # state (create the column, then configure it in the gear).
                # ⛔ C8 (wave 22): a CONFIGURED bag must name an existing flow — the write door
                # reads the tenant's definition ids and `_clean_automation` refuses a bag
                # without one (or naming a deleted one). Refused = the whole field write is
                # refused, loudly (repaint), because storing the field while dropping its bag
                # would be a column that silently never runs.
                if isinstance(raw.get('automation'), dict):
                    try:
                        _flow_ids = frozenset((ctx.table.st if ctx.table is not None
                                               else _tops(ctx).st).get('automations') or {})
                    except Exception:
                        _flow_ids = frozenset()
                    auto = _ag2._clean_automation(raw.get('automation'), valid_keys,
                                                  flow_ids=_flow_ids)
                    if auto is None:
                        return False
                    field['automation'] = auto
            # ⭐ WAVE-27 item 13 (R13): the code column's language. OPTIONAL, like the
            # `automation` bag above and unlike `formula` — an unconfigured code column is a
            # legitimate intermediate state (create it, pick the language in the gear), and it
            # stores and renders as plain text meanwhile. `_clean_code` never refuses over an
            # unknown language; it falls back to `plain`, because the value chooses a
            # HIGHLIGHTER, and dropping a user's column to punish a typo in one would be the
            # disproportionate half of fail-closed.
            if ftype == 'code' and isinstance(raw.get('code'), dict):
                code_bag = _ag2._clean_code(raw.get('code'))
                if code_bag:
                    field['code'] = code_bag
            fmt = _ag2._clean_format(raw.get('format'), ftype)
            if fmt:
                field['format'] = fmt
        else:
            return False
        # Wave-5 item 1 — createdBy + permissions on the CREATED strata (custom_/measure_).
        # `createdBy` is stamped HOST-SIDE at create and preserved on every later write: the
        # client never supplies it, so a browser cannot claim someone else's field. A
        # permissions CHANGE is accepted only from the creator or an admin; a legacy field
        # with no createdBy is admin-only (fail closed, both directions). The table workspace
        # is per-user today, so this wall becomes load-bearing when sharing arrives — but the
        # admin path and the legacy path are enforceable (and NC'd) right now.
        needs_values = False
        refused_perms = False
        if key.startswith('custom_') or key.startswith(_ag2.MEASURE_FIELD_PREFIX):
            prior = ws.get('fields', {}).get(key)
            prior = prior if isinstance(prior, dict) else None
            if prior is None:
                field['createdBy'] = uname
                # Wave-6 item 9: scope is chosen at CREATE, on the cohort page only —
                # 'cohort' marks the field cohort-specific; 'global' or absent stays
                # unscoped, so an old bundle that sends no scope keeps today's behavior.
                if scope_key == 'cohort' and raw.get('scope') == 'cohort':
                    field['scope'] = 'cohort'
            else:
                if isinstance(prior.get('createdBy'), str) and prior['createdBy'].strip():
                    field['createdBy'] = prior['createdBy']
                # Scope is PRESERVED like createdBy: the client copy never moves an
                # existing field between scopes.
                if prior.get('scope') == 'cohort':
                    field['scope'] = 'cohort'
            want = _ag2._clean_permissions(raw.get('permissions'))
            have = _ag2._clean_permissions((prior or {}).get('permissions'))
            owner = field.get('createdBy')
            if want is not None and (admin or (owner and owner == uname)):
                field['permissions'] = want
            else:
                if have is not None:
                    field['permissions'] = have
                # Asked for a permissions change and did not get it: the client's
                # optimistic copy is a lie that must be repainted (wave-6 item 3).
                refused_perms = want is not None and want != have
            # A NEW measure column, or an existing one whose window moved: the VALUES
            # are host-computed, so the next render genuinely differs.
            if key.startswith(_ag2.MEASURE_FIELD_PREFIX):
                needs_values = prior is None or prior.get('measure') != field.get('measure')
        # Canonical names are reserved outside the durable workspace. Persisted field names are
        # allocated by TableStore from the transaction's current state, not this stale snapshot.
        _workspace_field_keys = set((ws.get('fields') or {}))
        _reserved_field_names = [
            value.get('label') for candidate_key, value in field_by_key.items()
            if (candidate_key != key and candidate_key not in _workspace_field_keys
                and isinstance(value, dict))
        ]
        if _store_of(ctx).available():
            field = (_tops(ctx).save_field(
                uname, field, reserved_names=_reserved_field_names,
                correction_id=event_id) or field)
        else:
            _session_ready()
            _field_names = list(_reserved_field_names)
            _field_names.extend(
                value.get('label')
                for candidate_key, value in ws['fields'].items()
                if candidate_key != key and isinstance(value, dict)
            )
            requested_label = ' '.join(
                str(field.get('label') or 'Untitled').split())[:120].rstrip()
            field.pop('labelCorrectedFrom', None)
            field.pop('labelCorrectionId', None)
            field['label'] = _unique_name(requested_label, _field_names)
            if field['label'] != requested_label and event_id:
                field['labelCorrectedFrom'] = requested_label
                field['labelCorrectionId'] = str(event_id)[:180]
            # ⚠ W29-T83 — THE SAME PREDICATE the store uses, called rather than restated. This
            # fallback carried its own shorter copy (note only), so a format or measure override
            # survived one path and not the other, and `agg` survived neither.
            if _tstore.source_override_is_empty(field):
                ws['fields'].pop(key, None)
            else:
                ws['fields'][key] = field
        if needs_values or refused_perms:
            return True
        # THE NO-BLIP RETURN (wave-6 item 3): an accepted def that echoes what the client
        # already renders needs no rerun — see _field_echo_equal and the docstring's law.
        return not _field_echo_equal(field, raw)

    if kind in ('cohort_add', 'cohort_remove', 'cohort_rename', 'cohort_delete'):
        # Cohort membership/lifecycle edits from the grid's cohort panel (wave-2 item 2 —
        # replaces the retired Manage popover, so rename/delete keep a surface). Fail-closed:
        # only the caller's OWN cohorts (all_for), and member pids are intersected with the
        # caller's permitted pool — the same guard add_to_list rides.
        if not _store_of(ctx).available():
            return False
        sets = _cohorts(ctx)                          # R9: this DATABASE's lists, never another's
        cohort_id = str(event.get('cohortId') or '').strip()[:120]
        if not cohort_id or cohort_id not in sets.all_for(uname):
            return False
        try:
            if kind == 'cohort_rename':
                name = str(event.get('name') or '').strip()[:120]
                if not name:
                    return False
                sets.rename(uname, cohort_id, name)
            elif kind == 'cohort_delete':
                sets.delete(uname, cohort_id)
            else:
                raw_pids = event.get('pids')
                if not isinstance(raw_pids, list):
                    return False
                pids = [p for p in raw_pids if isinstance(p, int) and p in allowed_pids]
                if not pids:
                    return False
                if kind == 'cohort_add':
                    sets.add_members(uname, cohort_id, pids)
                else:
                    sets.remove_members(uname, cohort_id, pids)
        except ValueError as e:
            _tel.error('cohort:panel-edit', e)
            return False
        # True: the lists payload (names, membership, counts) must reflect this on the next
        # render — the panel is showing server truth, not local optimism.
        return True

    if kind in ('doc_add', 'doc_delete', 'doc_fetch'):
        # DOCUMENTS on a row (owner item 12, contract C5). Metadata in the store key
        # `customer_docs`; the BYTES live at their own path in the tenant dataset, never in a
        # JSON blob — see core.store.upload_bytes for why that distinction is load-bearing.
        #
        # ⭐ WAVE 19 / R9 — item 1's SILENT THIRD CASE, and the one nobody reported because the
        # React shell cannot reach it yet (`payload.docs` is set only by app.py). Both homes are
        # keyed by pid and both were topic-blind: the metadata bucket AND the byte path
        # `docs/<pid>/…`. Two topics' pid spaces overlap by construction (a CRC32 hash lands
        # anywhere an Odoo partner id can), so a product's file could occupy the path of a
        # customer's — one upload silently shadowing another tenant-visible document. Per-topic
        # bucket AND per-topic prefix; the customer topic keeps both of its shipped names.
        import base64 as _b64c
        import re as _re_mod
        if not _store_of(ctx).available():
            return False
        docs_key = _docs_key(ctx)
        docs_dir = 'docs' if docs_key == _DOCS_KEY else f'docs/{docs_key[:-5]}'
        pid = event.get('pid')
        doc_id = str(event.get('docId') or '').strip()[:80]
        # The pid wall is the SAME one every other row-scoped event rides: a caller may only
        # touch rows inside their own permitted pool. Without it, documents would be the
        # one surface where a scoped agent could read another book's rows.
        if not isinstance(pid, int) or pid not in allowed_pids or not doc_id:
            return False
        # ⭐ WAVE 25 (R6b / D-16) — THE DOCUMENTS FAMILY WAS THE LAST MODULE-GLOBAL READER HERE.
        # Every line below said `store.…`, so an R2 tenant's document metadata (and its bytes)
        # landed in tenant #0's repo. Wave 19 made this family per-TOPIC; per-TENANT is the axis
        # it was still missing, and the two failures look identical from the app.
        _st_docs = _store_of(ctx)
        meta_all = _st_docs.get(docs_key) or {}
        row = list((meta_all.get(str(pid)) or []))

        if kind == 'doc_add':
            raw_b64 = event.get('data_b64')
            name = str(event.get('name') or 'document')[:200]
            if not isinstance(raw_b64, str) or not raw_b64:
                return False
            try:
                blob = _b64c.b64decode(raw_b64, validate=True)
            except Exception:
                return False              # not base64 -> not a file we will store
            # Re-check the DECODED size host-side. The client checks it too, but a client check
            # is a courtesy: this is the number that bounds what the tenant's repo can be made
            # to hold by anyone who can reach the endpoint.
            if not blob or len(blob) > _DOC_MAX_BYTES:
                return False
            if len(row) >= _DOC_MAX_PER_ROW or any(d.get('id') == doc_id for d in row):
                return False
            safe = _re_mod.sub(r'[^A-Za-z0-9._-]', '_', name)[:80] or 'file'
            path = f'{docs_dir}/{pid}/{doc_id}_{safe}'
            try:
                _blobs(ctx).upload_bytes(
                    path, blob, message=f'doc {doc_id} for {scope_key} {pid}')
            except Exception as _ue:
                _tel.error('docs:upload', _ue)
                return False
            row.append({'id': doc_id, 'name': name,
                        'mime': str(event.get('mime') or 'application/octet-stream')[:120],
                        'size': len(blob), 'by': uname,
                        'ts': dt.datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
                        'path': path})
        elif kind == 'doc_delete':
            hit = next((d for d in row if d.get('id') == doc_id), None)
            if not hit:
                return False
            # THE WALL: the uploader or an admin. Enforced here regardless of what the client
            # chose to render — `canDelete` in the payload only hides a button.
            if hit.get('by') != uname and not admin:
                return False
            _blobs(ctx).delete_path(hit.get('path') or '')
            row = [d for d in row if d.get('id') != doc_id]
        else:                                    # doc_fetch
            hit = next((d for d in row if d.get('id') == doc_id), None)
            if not hit:
                return False
            blob = _blobs(ctx).download_bytes(hit.get('path') or '')
            if blob is None:
                return False
            # The bridge is a fire-and-forget event LOG with no response channel, so the answer
            # is PARKED on the Result and rides the next render's payload (C5).
            ctx.out.doc = {
                'pid': pid, 'docId': doc_id, 'name': hit.get('name') or 'document',
                'mime': hit.get('mime') or 'application/octet-stream',
                'data_b64': _b64c.b64encode(blob).decode('ascii')}
            return True                          # re-render to deliver it

        def _write(cur):
            cur[str(pid)] = row
            return cur
        # ⚠ `docs_key`, NOT `_DOCS_KEY`. The read above and the byte path were both made
        # per-topic and this line was not — so a product document's BYTES landed correctly under
        # `docs/product/…` while its METADATA went into `customer_docs` beside the customer rows,
        # keyed by a CRC32 hash. Found by `verify_scopes.py`'s own leg on its first run, which is
        # the argument for asserting the ABSENCE from the legacy bucket and not just the presence
        # in the new one: every "it saved" check here was green.
        _st_docs.update(docs_key, _write, flush='async')
        return True

    if kind in ('folder_create', 'folder_rename', 'folder_delete', 'folder_duplicate',
                'item_move', 'folder_reorder'):
        # FOLDERS over the Views / Cohorts sidebars (owner item 11, contract C4). All five
        # events touch ONE home — the table workspace's `folders` + `itemFolders` — so a cohort
        # can be filed without the cohort STORE learning about folders at all.
        import aios_grid as _agf
        if not _store_of(ctx).available():
            return False
        surface = str(event.get('surface') or '')
        if surface not in _agf.FOLDER_SURFACES:
            return False
        # The ids this user may actually file: their own views, their own cohorts. Fail-closed —
        # an item_move naming somebody else's cohort is dropped by clean_item_folders anyway,
        # but refusing here means we never write it in the first place.
        def _own_ids(sfc, cur_ws):
            if sfc == 'views':
                return set((cur_ws.get('views') or {}))
            return set(_cohorts(ctx).all_for(uname))

        def _mutate(cur_ws):
            folders = _agf.clean_folders(cur_ws.get('folders'))
            placed = dict(cur_ws.get('itemFolders') or {})
            lst = list(folders.get(surface) or [])
            # ⚠ the wire field is `folderId` on EVERY folder event (CLIENT's types.ts is the
            # built contract; the split doc's first draft said `id`). Converged here rather
            # than asking the client to re-emit — see the C4 reconciliation note in the doc.
            fid = str(event.get('folderId') or '').strip()[:80]
            if kind == 'folder_create':
                name = str(event.get('name') or '').strip()[:_agf.MAX_FOLDER_NAME]
                if not fid or not name or len(lst) >= _agf.MAX_FOLDERS:
                    return None
                if any(f['id'] == fid for f in lst):
                    return None
                row = {'id': fid, 'name': name, 'order': len(lst)}
                icon = _agf.clean_folder_icon(event.get('icon'))    # wave-9 I15 (C5)
                if icon:
                    row['icon'] = icon
                lst.append(row)
            elif kind == 'folder_rename':
                name = str(event.get('name') or '').strip()[:_agf.MAX_FOLDER_NAME]
                hit = next((f for f in lst if f['id'] == fid), None)
                if not hit or not name:
                    return None
                hit['name'] = name
                # wave-9 I15 (C5): the rename pane is also where the icon is changed. An event
                # that carries NO icon key leaves the existing one alone (a rename must not
                # silently strip a chosen icon); an explicit null clears it back to default.
                if 'icon' in event:
                    icon = _agf.clean_folder_icon(event.get('icon'))
                    if icon:
                        hit['icon'] = icon
                    else:
                        hit.pop('icon', None)
            elif kind == 'folder_delete':
                if not any(f['id'] == fid for f in lst):
                    return None
                lst = [f for f in lst if f['id'] != fid]
                # CONTENTS MOVE TO ROOT, never delete. Dropping the placements is exactly that:
                # an item with no placement renders at the top level.
                placed[surface] = {k: v for k, v in (placed.get(surface) or {}).items()
                                   if v != fid}
            elif kind == 'folder_reorder':
                # ⭐ WAVE 20 (owner item 19, contract C-FOLDER-REORDER) — the user's own folder
                # ORDER in the rail. Views could already be dragged between folders; the folders
                # themselves could not be dragged past each other.
                #
                # THE WIRE CARRIES THE FULL ORDER, NEVER A DELTA, and that is the contract's
                # doing: a partial list cannot say where an UNNAMED folder went, so a "moved B
                # after C" message would leave every other folder's position to be re-derived by
                # two sides that can disagree. The client sends the list it is looking at.
                #
                # ⚠ IDS THIS USER DOES NOT OWN ARE IGNORED, NOT REJECTED. A stale tab can emit an
                # order containing a folder that has since been deleted; dropping the whole event
                # would make the rail un-reorderable until reload, while dropping the unknown id
                # reorders exactly what still exists. Anything the payload omits keeps its
                # relative position AFTER the named ones — an order that silently deleted a
                # folder it merely failed to mention would be a data loss dressed as a sort.
                order = [str(i).strip()[:80] for i in (event.get('order') or [])]
                if not order:
                    return None
                known = {f['id']: f for f in lst}
                seen, ranked = set(), []
                for fid_ in order:
                    if fid_ in known and fid_ not in seen:
                        seen.add(fid_)
                        ranked.append(known[fid_])
                ranked.extend(f for f in lst if f['id'] not in seen)
                if len(ranked) != len(lst):          # cannot happen; refuse rather than truncate
                    return None
                for i, f in enumerate(ranked):
                    f['order'] = i
                lst = ranked
            elif kind == 'folder_duplicate':
                new_id = str(event.get('newId') or '').strip()[:80]
                src = next((f for f in lst if f['id'] == fid), None)
                if not src or not new_id or len(lst) >= _agf.MAX_FOLDERS:
                    return None
                if any(f['id'] == new_id for f in lst):
                    return None
                lst.append({'id': new_id, 'name': f"{src['name']} copy"[:_agf.MAX_FOLDER_NAME],
                            'order': len(lst)})
                # ⚠ The folder's CONTENTS are duplicated by the client emitting the ordinary
                # per-item duplicate events it already owns (view_upsert with a fresh id, the
                # cohort copy path) and then item_move-ing them here. Copying views/cohorts
                # inside this handler would be a SECOND implementation of "duplicate an item",
                # and the two would drift the first time either one changed.
            else:                                   # item_move
                item_id = str(event.get('itemId') or '').strip()[:120]
                target = event.get('folderId')
                if not item_id or item_id not in _own_ids(surface, cur_ws):
                    return None
                cur = dict(placed.get(surface) or {})
                if target is None:
                    # ⚠ "UNFILE" — the client says nothing about where it went. Popping is right
                    # here and always was: no placement = render at the top level, and that is the
                    # state a brand-new item is already in.
                    cur.pop(item_id, None)          # back to the root
                elif str(target) == _agf.ROOT_PLACEMENT:
                    # ⭐⭐ WAVE 32 · OWNER ITEM 20 (`W32-T27`, ASK C-16) — "FILED AT ROOT", STORED.
                    #
                    # ⛔ THE THIRD CASE, AND ITS ABSENCE IS THE WHOLE DEFECT. Popping (above) and
                    # this branch produce the SAME rendering for an ordinary view, which is why
                    # nobody missed it for nine waves — but they are different FACTS, and for a
                    # SHARED view the difference is the entire feature: "never filed" sends it to
                    # the Shared group, "the receiver dragged it out" must not. Absence cannot hold
                    # two facts, so the second one gets a value.
                    #
                    # ⛔ AND IT IS DELIBERATELY EXEMPT FROM THE FOLDER-EXISTS TEST BELOW. This id
                    # names no folder by design; running it through `any(f['id'] == tid ...)` would
                    # refuse every root filing, which is the same silent no-op this ticket exists
                    # to remove. `aios_grid.clean_item_folders` carries the matching exemption on
                    # the READ side — the two are one change (see its note).
                    cur[item_id] = _agf.ROOT_PLACEMENT
                else:
                    tid = str(target)[:80]
                    if not any(f['id'] == tid for f in lst):
                        return None
                    cur[item_id] = tid
                placed[surface] = cur
            folders[surface] = lst
            clean = _agf.clean_folders(folders)
            return clean, _agf.clean_item_folders(
                placed, clean,
                {sfc: _own_ids(sfc, cur_ws) for sfc in _agf.FOLDER_SURFACES})

        # ⚠ NO allowed_pids ON PURPOSE — this is app.py's bare `_cl_table_workspace(uname)`
        # call at the old :6268, preserved exactly. Omitting the pool drops memberPids from
        # any foreign shared view (fail-closed), which is harmless here because the folder
        # legs read only view IDS and the folder maps. Passing the pool would be a behavior
        # change inside a pure move.
        _res = _mutate(table_workspace(ctx, consume_corrections=False))
        if _res is None:
            return False                      # refused: nothing is written
        _tops(ctx).save_folders(uname, _res[0], _res[1])
        # True: the sidebars render from server state (which folder holds what), not from local
        # optimism — the same contract the cohort panel's membership edits ride.
        return True

    if kind == 'view_reorder':
        # ⭐ WAVE-27 item 5 (contract C7) — the user drags views up and down the rail.
        #
        # THE WIRE CARRIES THE FULL ORDER, NEVER A DELTA — the `folder_reorder` contract above,
        # verbatim, and for its reason: a partial list cannot say where an unnamed view went.
        #
        # ⚠ NO OWNERSHIP FILTER, and that is the difference from `item_move`. Filing a view into
        # a folder writes a placement keyed by a view this user must own; ARRANGING the rail is
        # about the reader's own screen, and the rail legitimately contains views this user does
        # not own — the system view, cohort projections, R8's injected Overview, and anything
        # SHARED with them. Refusing ids they do not own would make exactly those un-draggable,
        # which is the half of sharing people notice.
        #
        # ⚠ Unresolvable ids are harmless by construction rather than by filtering here: the READ
        # side (`aios_grid.views_from_defs`) ranks by this list and appends anything unranked, so
        # an id naming a deleted view simply never matches. Validating against today's view set
        # at WRITE time would also drop an id whose view is merely not visible in this request.
        raw_order = event.get('order')
        if not isinstance(raw_order, list):
            return False        # a non-list order is malformed, not an empty arrangement
        order, seen_v = [], set()
        for vid in raw_order[:MAX_VIEW_ORDER]:
            vid = str(vid or '').strip()[:120]
            if vid and vid not in seen_v:
                seen_v.add(vid)
                order.append(vid)
        # ⚠ THE SESSION FALLBACK IS `record_layout`'s, COPIED DELIBERATELY, and it is not just a
        # testability convenience. Both keys are per-user PRESENTATION state for one surface, and
        # both should keep working while the store is briefly unavailable — a rail that refuses
        # to reorder because a write failed is a frozen control with no explanation. It also
        # makes the round trip reachable from `verify_grid_events`, which is headless and where
        # `store.available()` is False: the store-backed branch of every OTHER stratum writer is
        # covered only by the live pass, and this one did not have to be.
        if _store_of(ctx).available():
            _tops(ctx).save_view_order(uname, order)
        else:
            _session_ready()
            if order:
                ws['viewOrder'] = order
            else:
                ws.pop('viewOrder', None)     # absent = server order, never a stored []
        # True: the rail renders from server state, like the folder sidebars it sits beside.
        return True

    if kind == 'field_delete':
        # Delete a USER-CREATED column outright (owner gap closed 2026-07-27). The prefix IS
        # the permission: only the created strata (`custom_` overlay fields, `measure_` formula
        # columns) are deletable — a base field's key fails the prefix test no matter what the
        # browser claims, so the table's contract cannot be deleted from the client. The store
        # scrubs the field's stored cell values with it (delete_table_field); views naming the
        # key self-heal on their next autosave, the rule every stale colId already rides.
        key = str(event.get('key') or '')[:80]
        import aios_grid as _ag3
        if not (key.startswith('custom_') or key.startswith(_ag3.MEASURE_FIELD_PREFIX)):
            return False
        if _store_of(ctx).available():
            _tops(ctx).delete_field(uname, key)
        else:
            _session_ready()
            ws['fields'].pop(key, None)
            for _row in ws['overlays'].values():
                if isinstance(_row, dict):
                    _row.pop(key, None)
        # Wave-6 item 3: the client dropped the column optimistically; the store scrub is
        # invisible to the next render (the values leave WITH the column).
        return False

    if kind == 'choice_rename':
        # ⭐ WAVE 20 (owner item 15, contract C-RENAME) — RENAMING AN OPTION CARRIES ITS VALUES.
        #
        # Owner: *"if I change the option from 'Aaron' to 'Aron' the existing field that has the
        # Selected 'Aaron' doesn't change to 'Aron'."* Correct, and it is a data-model fact rather
        # than an oversight: a select cell stores the option's LABEL, so renaming the option in
        # the field definition orphans every cell holding the old string. They do not render as
        # the new name; they render as a value that is no longer an option.
        #
        # ⛔ THE WIRE CARRIES AN EXPLICIT {from, to} MAPPING, NEVER A DIFF OF THE CHOICE LISTS.
        # Diffing is what makes this unsafe: rename A→B while also DELETING C and adding D, and a
        # differ sees two removals and two additions with no way to know which pairs up. It would
        # cheerfully rewrite every C cell to D. The client knows which row of its editor the user
        # typed in; that intent travels, and nothing here has to guess it.
        #
        # THREE PLACES HOLD THE OLD STRING, and missing any one is its own visible bug:
        #   1. the CELLS (`overlays`) — the owner's complaint;
        #   2. the FIELD's own `choices` list — else the picker still offers the old name;
        #   3. every VIEW that FILTERS or COLOURS BY that value — a saved view filtering
        #      `Agent is "Aaron"` silently matches NOTHING the moment the value moves, which is
        #      the failure that looks like data loss ([[aios-shared-views]]).
        key = str(event.get('key') or '')[:80]
        pairs = []
        for r in (event.get('renames') or ()):
            if not isinstance(r, dict):
                continue
            a, b = str(r.get('from') or '')[:200], str(r.get('to') or '')[:200]
            if a and b and a != b:
                pairs.append((a, b))
        field = field_by_key.get(key)
        if not key or not pairs or not isinstance(field, dict):
            return False
        if str(field.get('type') or '') not in ('select', 'multiselect', 'status'):
            return False
        # Only a field this caller may edit. A rename is a write to the field CONTRACT, so it
        # rides the same wall `field_upsert` does rather than a looser one of its own.
        if key in (ctx.hidden_keys or frozenset()):
            return False
        mapping = dict(pairs)

        def _rename_in(ws_):
            fields_ = ws_.setdefault('fields', {})
            fld = fields_.get(key)
            if isinstance(fld, dict) and isinstance(fld.get('choices'), list):
                seen, out = set(), []
                for c in fld['choices']:
                    if isinstance(c, dict):
                        nm = mapping.get(str(c.get('name') or ''), None)
                        if nm is not None:
                            c = {**c, 'name': nm}
                        ident = str(c.get('name') or '')
                    else:
                        c = mapping.get(str(c), str(c))
                        ident = str(c)
                    # A rename ONTO an existing option MERGES rather than duplicating it: the
                    # user typed a name that already exists, and two identical options is a
                    # picker bug. Their cells merge too, which is what the value rewrite below
                    # does anyway.
                    if ident in seen:
                        continue
                    seen.add(ident)
                    out.append(c)
                fld['choices'] = out
                fields_[key] = fld
            for row in (ws_.get('overlays') or {}).values():
                if not isinstance(row, dict) or key not in row:
                    continue
                val = row[key]
                if isinstance(val, list):        # multiselect
                    merged, out2 = set(), []
                    for v in val:
                        nv = mapping.get(str(v), str(v))
                        if nv not in merged:
                            merged.add(nv)
                            out2.append(nv)
                    row[key] = out2
                elif isinstance(val, str) and val in mapping:
                    row[key] = mapping[val]
            # Views: filter rule values + colour-by keys naming the old option.
            for view in (ws_.get('views') or {}).values():
                if not isinstance(view, dict):
                    continue
                cfg = view.get('config') if isinstance(view.get('config'), dict) else view

                def _walk(nodes):
                    for n in nodes or ():
                        if not isinstance(n, dict):
                            continue
                        if isinstance(n.get('children'), list):
                            _walk(n['children'])
                            continue
                        if str(n.get('colId') or '') != key:
                            continue
                        v = n.get('value')
                        if isinstance(v, str) and v in mapping:
                            n['value'] = mapping[v]
                        elif isinstance(v, list):
                            n['value'] = [mapping.get(str(x), x) for x in v]

                _walk(cfg.get('filters') or [])
                # ⚠ `colorBy` NEEDS NO REWRITE, and the reason is worth stating because the
                # opposite looks obvious: it stores a COLUMN KEY (a string, validated against
                # `valid_keys` at `view_upsert`), not a {value: colour} map. Per-value colours
                # ride the CHOICE OBJECTS themselves, which this function renames in place —
                # `{**c, 'name': nm}` keeps every other key on the choice, colour included — so
                # a renamed option keeps its swatch for free. An earlier version of this handler
                # tried to rewrite `colorBy['rules']` and crashed the whole event pipeline with
                # `TypeError: unhashable type: 'dict'` on the FIRST view it touched.
            return ws_

        if _store_of(ctx).available():
            _tops(ctx).rename_choice_values(uname, _rename_in)
        else:
            _session_ready()
            _rename_in(ws)
        # True = the host must re-read: the client dropped its optimistic copy of the CHOICES,
        # but the rewritten cells and view filters only exist server-side until the next payload.
        return True

    if kind == 'field_duplicate':
        # Wave-5 item 1. The DESTINATION key is client-generated (the established pattern —
        # the client must know it synchronously to place the clone right of the source), and
        # both keys must sit in the SAME created stratum: base odoo fields offer no Duplicate,
        # because duplicating the source-of-truth into an editable copy would need a
        # value-snapshot mechanism nobody asked for. Overlay VALUES are copied only for
        # `custom_` sources (a measure clone recomputes; a formula clone re-evaluates).
        # The duplicator becomes the clone's creator — it is a NEW field they made.
        src_key = str(event.get('sourceKey') or '')[:80]
        new_key = str(event.get('key') or '')[:80]
        import aios_grid as _ag5
        prefix = ('custom_' if src_key.startswith('custom_')
                  else _ag5.MEASURE_FIELD_PREFIX
                  if src_key.startswith(_ag5.MEASURE_FIELD_PREFIX) else None)
        if (prefix is None or not new_key.startswith(prefix) or new_key == src_key
                or new_key in field_by_key):
            return False
        src = ws.get('fields', {}).get(src_key)
        if not isinstance(src, dict):
            return False
        clone = dict(src)
        clone['key'] = new_key
        label = str(event.get('label') or '').strip()[:120]
        requested_label = label or (str(src.get('label') or 'Untitled')[:110] + ' copy')
        _workspace_field_keys = set((ws.get('fields') or {}))
        _reserved_field_names = [
            value.get('label') for candidate_key, value in field_by_key.items()
            if (candidate_key != new_key and candidate_key not in _workspace_field_keys
                and isinstance(value, dict))
        ]
        clone['label'] = requested_label
        clone['createdBy'] = uname
        # Wave-6 item 9: the clone inherits the source's scope via dict(src); on the cohort
        # page the duplicate surface may explicitly place it instead (same create-time rule
        # as field_upsert — a page that is not the cohort page cannot mint a cohort field).
        if scope_key == 'cohort' and event.get('scope') in ('cohort', 'global'):
            if event['scope'] == 'cohort':
                clone['scope'] = 'cohort'
            else:
                clone.pop('scope', None)
        if _store_of(ctx).available():
            clone = (_tops(ctx).duplicate_field(
                uname, src_key, new_key, clone,
                reserved_names=_reserved_field_names,
                correction_id=event_id) or clone)
        else:
            _session_ready()
            _field_names = list(_reserved_field_names)
            _field_names.extend(
                value.get('label')
                for candidate_key, value in ws['fields'].items()
                if candidate_key != new_key and isinstance(value, dict)
            )
            clone.pop('labelCorrectedFrom', None)
            clone.pop('labelCorrectionId', None)
            clone['label'] = _unique_name(requested_label, _field_names)
            if clone['label'] != requested_label and event_id:
                clone['labelCorrectedFrom'] = requested_label
                clone['labelCorrectionId'] = str(event_id)[:180]
            ws['fields'][new_key] = clone
            if src_key.startswith('custom_'):
                for _row in ws['overlays'].values():
                    if isinstance(_row, dict) and src_key in _row:
                        _row[new_key] = _row[src_key]
        # Wave-6 item 3: True only when host-held VALUES must flow back — copied overlay
        # cells (custom_ sources) or host-computed measure values. formula/created_time
        # clones compute client-side, so the optimistic insert is already complete.
        label_changed = clone['label'] != requested_label
        return label_changed or src.get('type') not in ('formula', 'created_time')

    if kind == 'add_to_list':
        # "Add to list" (owner item 6): a view's current matches become fixed cohort members.
        #
        # ⚠ `pids` arrives from the BROWSER. It is intersected with `allowed_pids` — the pool this
        # user may actually see — before anything is stored. The client computing the set is a
        # correctness decision (it is the only engine that already knows the answer, and asking a
        # second engine the same question is the drift verify_filter_engine.py exists to prevent);
        # it is NOT a trust decision.
        raw_pids = event.get('pids')
        if not isinstance(raw_pids, list):
            return False
        pids = [p for p in raw_pids if isinstance(p, int) and p in allowed_pids]
        if not pids:
            return False
        cohort_id = str(event.get('cohortId') or '').strip()[:120]
        name = str(event.get('name') or '').strip()[:120]
        if not _store_of(ctx).available():
            return False
        # ⭐ WAVE 19 / R9 — THE BUG THIS RULING NAMES. This branch is where a list is BORN, and it
        # named `cohort_mod` directly: a list created from the Product surface was written into
        # `customer_cohorts` with CRC32 product pids in it, then rendered on the Customer page as
        # a list of customers nobody recognised. The set now belongs to the database it was
        # created on, and `ctx.scope_key` is the only thing that decides which.
        sets = _cohorts(ctx)
        try:
            if not cohort_id:
                if not name:
                    return False
                cohort_id = sets.create(uname, name, pids)
            else:
                if cohort_id not in sets.all_for(uname):
                    return False        # not this user's cohort — fail closed, do not create
                sets.add_members(uname, cohort_id, pids)
        except ValueError as e:
            _tel.error('cohort:add_to_list', e)
            return False
        # True: the Cohort page's list and counts must reflect this, and the toast below only
        # renders on a fresh run.
        #
        # ⚠ The NOUN is the TOPIC's, not the literal "customer" this line carried since wave 2.
        # "Added 12 customers" over a set of SKUs is the surface stating, in words, that the list
        # went somewhere it did not — the user-visible half of the same leak.
        _noun = sets.noun
        ctx.out.toast = (
            f"Added {len(pids):,} {_noun}{'' if len(pids) == 1 else 's'} to "
            f"{name or 'the list'}.")
        return True

    if kind == 'overlay_patch':
        pid = event.get('pid')
        if not isinstance(pid, int) or pid not in allowed_pids:
            return False
        updates = {}
        refused = False
        stage_moved = False
        row_events = []            # C3 (wave 22): admitted writes, emitted AFTER they apply
        raw_updates = event.get('updates') if isinstance(event.get('updates'), dict) else {}
        for key, value in raw_updates.items():
            if key not in overlay_keys or not isinstance(value, (str, int, float)):
                refused = True
                continue
            # C-PERM: a field this caller's permissions HIDE cannot be written, only refused.
            # `overlay_keys` asks whether the column exists, not whether this person may touch
            # it — so without this line a restricted user who knows a hidden key could PATCH a
            # value they are not allowed to read back. Refused LOUDLY (repaint) so the client
            # reverts rather than showing a value the store never took.
            if key in (ctx.hidden_keys or frozenset()):
                refused = True
                continue
            # Wave-5 item 1: `permissions.edit` is enforced HERE, per key — the client hiding
            # its editor is courtesy, this is the wall. Fail-closed: a restricted field with
            # no readable createdBy admits only admins.
            perms = (field_by_key.get(key) or {}).get('permissions') or {}
            edit = perms.get('edit')
            if edit == 'admins' and not admin:
                refused = True
                continue
            if edit == 'creator' and not (
                    admin or (field_by_key.get(key) or {}).get('createdBy') == uname):
                refused = True
                continue
            # C5-AUTOFIELD (wave 18): an automation cell is MACHINE-WRITTEN. The client marks
            # the type read-only (READONLY_CELL_TYPES) but that is courtesy — this is the wall.
            # Without it the UI hides the editor while the API still takes a PATCH, which is
            # exactly the shape a "read-only" type must not have.
            if (field_by_key.get(key) or {}).get('type') == 'automation':
                refused = True
                continue
            # C2 (wave 22): a STAGE FIELD is engine-written EXCEPT at a review gate, where a
            # human may advance a card to a member of the review stage's `next` and nothing
            # else. The law rides the field definition (`automation.humanMoves` / `arrive`,
            # stamped by the engine that derives the stages) because this module must not
            # import the engine — the engine states the law, this door enforces it
            # mechanically. An ADMITTED move writes THROUGH to the definition row, the shared
            # truth the engine and every viewer read: parked in this user's overlay stratum it
            # would be a board position only its author could see. Everything else about the
            # field is refused — the same wall the automation-cell type gets, extended, not
            # replaced.
            if str(ctx.scope_key or '').startswith('ut_'):
                import core.user_tables as _ut_mod
                _tdef = _ut_mod.get(ctx.scope_key, st=ctx.table.st) if ctx.table is not None \
                    else None
                _fdef = next((f for f in ((_tdef or {}).get('fields') or [])
                              if f.get('key') == key), None)
                # C7 (wave 22): a METRIC cell is engine-computed from the master series — a
                # human write would be an invented measurement, refused the way an automation
                # cell is (the wall, not the courtesy).
                # ⭐ 2026-08-07 — EXTENDED TO THE RELATIONAL PAIR, and read through the field
                # layer's own predicate rather than re-tested here. A `rollup` cell is an
                # aggregate the server computes; a DERIVED `link` cell is a join the server
                # resolves. Both would be overwritten by the next refresh anyway, so accepting
                # the write would not merely be wrong — it would look like it worked and then
                # silently revert, which is worse than a refusal.
                # ⚠ An ORDINARY (user-picked) link is deliberately NOT refused: `is_computed_cell`
                # answers per FIELD, not per type, because one kind is read-only in one mode and
                # editable in the other.
                if _ut_mod.is_computed_cell(_fdef):
                    refused = True
                    continue
                _auto = (_fdef or {}).get('automation')
                # An ordinary Link is a shared database relationship. Validate its selected
                # target ids and write it through to the definition row; a personal overlay
                # would be invisible to reciprocal links and Rollups.
                if (_fdef or {}).get('type') == 'link':
                    _linked = _ut_mod.patch_link_cell(
                        ctx.scope_key, pid, key, value, st=ctx.table.st)
                    if _linked is None:
                        refused = True
                        continue
                    if _linked != str(value):
                        refused = True
                    row_events.append((key, _linked))
                    continue
                # ⭐ WAVE 25 (C3 + amendment C3-A1, owner rulings R7/R6) — THE PROFILE CELL.
                # Two things happen here and neither can live anywhere else:
                #
                #   1. IT WRITES THROUGH TO THE DEFINITION ROW, the stage field's mechanism two
                #      branches up, for the identical reason. MEASURED before it was written: an
                #      ordinary `ut_*` cell edit lands in the TYPIST'S OWN overlay stratum, and
                #      the automation engine reads `t['rows']` — so a handle somebody typed
                #      would be invisible to the enrich action and to every other viewer of the
                #      same database. The flag would be decorative.
                #   2. BLANKING IT CLEARS THAT ROW'S PRESET CELLS IN THE SAME WRITE (R6) —
                #      `patch_profile_cell` does both inside ONE store update, because a blank
                #      that commits without its clear leaves a row with no handle and a live
                #      follower count, i.e. stale numbers attributed to nobody.
                #
                # ⛔ NOT VIA THE EMIT GUARD (HARD RULE 5 / D-40). Nothing here emits; this is
                # `user_tables`' own write door, reached from the branch that already knows the
                # scope is `ut_`. ⚠ `editRole` is deliberately NOT consulted: it governs the
                # SCHEMA, and C3 is explicit that a profile field's VALUES stay fully editable —
                # that is the whole point of item 5a.
                if isinstance((_fdef or {}).get('profile'), dict):
                    _pw = _ut_mod.patch_profile_cell(
                        ctx.scope_key, pid, key, value, st=ctx.table.st)
                    if _pw is None:
                        # Not a handle we can enrich. Refused LOUDLY so the client repaints to
                        # the stored truth instead of showing a value the store never took.
                        refused = True
                        continue
                    # A normalisation (`@Name` -> `name`, a pasted URL -> its handle) is a real
                    # difference from what the client is drawing, so it must repaint. An
                    # untouched value echoes and stays on the no-blip path.
                    if _pw['handle'] != str(value):
                        refused = True
                    if _pw['cleared']:
                        stage_moved = True    # cells the client never typed changed: re-read
                    row_events.append((key, _pw['handle']))
                    continue
                # ⭐ OWNER ITEM 4 (2026-08-06) — A COLUMN AN AUTOMATION FILLS IS NOT A COLUMN A
                # PERSON TYPES IN. Owner, verbatim: *"I should not be able to edit any of the
                # Pre-set fields on instagram automation, we talked about this extensively."*
                #
                # ⛔ IT HAD NEVER BEEN ENFORCED ANYWHERE. `editRole: 'admins'` rides on every
                # preset column and reads exactly like this wall, which is why it was believed to
                # be one — but it governs the SCHEMA (`user_tables.py:535`): who may rename or
                # retype the column. The VALUES stayed open to anyone with grid access, and the
                # owner typed into one to prove it. The metric branch above is the same rule for
                # the same reason, written for a different column kind; this is its sibling.
                #
                # ⚠ REACHED ONLY AFTER stage AND profile have `continue`d, and the order is the
                # rule rather than an accident: both of those columns carry the identical tag and
                # both are cells a human is SUPPOSED to drive — the stage because moving a card
                # IS the decision, the profile because it is where enrichment gets its subject.
                # Refusing on the tag alone would freeze every kanban.
                #
                # `flowId` and not merely `automation`: a user-configured `automation` COLUMN
                # carries its own gear bag (kind/source/urlField/settings) and no flowId, and it
                # is not this ruling's subject.
                if isinstance(_auto, dict) and (_auto.get('flowId') or _auto.get('preset')):
                    refused = True
                    continue
            # ⭐ WAVE 23 (C7) — the json wall, BEFORE the generic 10 000-char truncation, which
            # would otherwise cut a document in half and store the halves as valid text. A json
            # column promises its readers a parseable document; a write that is not one is
            # REFUSED with the cell unchanged rather than silently kept as a string nothing can
            # open. Blank stays legal — an empty cell is "no document", not a malformed one.
            if (field_by_key.get(key) or {}).get('type') == 'json':
                raw_j = value if isinstance(value, str) else _json_ev.dumps(value)
                if not str(raw_j).strip():
                    updates[key] = ''
                    continue
                if len(str(raw_j)) > _ut_limits.MAX_JSON_CELL:
                    refused = True
                    continue
                try:
                    _json_ev.loads(raw_j)
                except (ValueError, TypeError):
                    refused = True
                    continue
                updates[key] = str(raw_j)
                continue
            clean = str(value)[:10000]
            if isinstance(value, str) and clean != value:
                # Truncated: the stored truth is shorter than the client's copy. (An
                # int/float arriving as its string form renders identically — not a refusal.)
                refused = True
            definition = field_by_key.get(key) or {}
            if definition.get('type') in ('select', 'multiselect') and clean:
                options = [str(v) for v in definition.get('options') or []]
                canonical = {v.strip().lower(): v for v in options}
                if definition.get('type') == 'select':
                    normalized = canonical.get(clean.strip().lower())
                    if normalized is None:
                        refused = True
                        continue
                else:
                    normalized_parts = []
                    seen = set()
                    invalid = False
                    for part in clean.split(','):
                        match = canonical.get(part.strip().lower())
                        if match is None:
                            invalid = True
                            break
                        if match.lower() not in seen:
                            seen.add(match.lower())
                            normalized_parts.append(match)
                    if invalid:
                        refused = True
                        continue
                    normalized = ','.join(normalized_parts)
                if normalized != clean:
                    refused = True
                    clean = normalized
            updates[key] = clean
        if updates:
            if str(ctx.scope_key or '').startswith('ut_') and ctx.table is not None:
                # User-table cells are collaborative record data. Views/layout remain personal,
                # but record values live in the definition rows so Links, Rollups, automations,
                # and every viewer observe one truth.
                import core.user_tables as _ut_values
                _ut_values.patch_cells(ctx.scope_key, pid, updates, st=ctx.table.st)
            elif _store_of(ctx).available():
                _tops(ctx).patch_overlay(uname, pid, updates)
            else:
                _session_ready()
                ws['overlays'].setdefault(str(pid), {}).update(updates)
        # C3 (wave 22): the admitted writes become ROW EVENTS — emitted AFTER they applied, on
        # the ut scopes the event triggers watch this wave, through the seam in user_tables so
        # this module never imports the engine. Human door only, which IS the loop law.
        if str(ctx.scope_key or '').startswith('ut_') and ctx.table is not None:
            import core.user_tables as _ut_ev
            for _k, _v in list(updates.items()) + row_events:
                _ut_ev.emit_row_event({'type': 'event_field', 'table': str(ctx.scope_key),
                                       'rowId': str(pid), 'field': _k, 'after': _v,
                                       'st': ctx.table.st, 'user': uname})
        # Wave-6 item 3: the client's cell already shows the typed value — rerun ONLY when
        # some of what it shows was refused or clamped and must be repainted to the truth.
        # A stage move reruns too: its `arrive` side-effects (the tracked flag) changed cells
        # the client never typed.
        return refused or stage_moved

    return False