Spaces:

fsanyoto
/

loopable

Running

File size: 124,741 Bytes

"""aios_grid — embed the AIOS React/glide Airtable-style grid inside Streamlit.



This is REUSABLE MODULE INFRASTRUCTURE: the same self-contained grid that runs as the

standalone aios-web app is inlined into a single HTML file and hosted inside a Streamlit

component. The React bundle reads its data from `window.__AIOS_GRID__` (an object

`{fields, rows}`) that we inject into the page <head> before the app's module script runs

— so there is NO backend and NO /api call in the Streamlit container; the browser only ever

sees the derived JSON we hand it.



Usage (from any page):

    import aios_grid

    aios_grid.render(aios_grid.rows_from_pool(pool_rows), aios_grid.FIELDS)



Design notes:

  * ZERO app-internal imports (no `modules.*` / `core.*`). FIELDS is a literal and

    `rows_from_pool` takes already-built rows — so this helper is tenant/module-agnostic and

    sidesteps deploy_hf.py's import guard entirely.

  * The built HTML ships as `aios_grid_embed.html` in THIS directory (added to

    deploy_hf.py INCLUDE). Build it with `npm run build:embed` in aios-web/web, then copy

    dist-embed/index.html -> platform/aios_grid_embed.html (see build_embed.py).

    For LOCAL dev before that copy, we fall back to reading dist-embed/index.html directly.

  * The preferred host is a Streamlit Components v1 bridge. It sends data/view/schema args

    into the React app and returns guarded events for persistence in the tenant store. The old

    injected-HTML path remains a read/local-write fallback when component assets are absent.

"""
import json
import re
from pathlib import Path

_HERE = Path(__file__).resolve().parent

# Where the inlined single-file build lives. FIRST match wins:
#   1. the shipped copy in this dir (what deploy_hf.py uploads to the Space)
#   2. the raw build output in the sibling aios-web tree (local dev, pre-copy)
_EMBED_CANDIDATES = [
    _HERE / "aios_grid_embed.html",
    _HERE.parent / "aios-web" / "web" / "dist-embed" / "index.html",
]
_COMPONENT_CANDIDATES = [
    _HERE / "aios_grid_component",
    _HERE.parent / "aios-web" / "web" / "dist-embed",
]
_DECLARED_COMPONENTS = {}

# --- the FIELD CONTRACT — loaded from the CANONICAL source `aios_grid_fields.json` in THIS
# directory (the SINGLE source of truth, shared with aios-web/api/main.py). source='odoo' is
# READ-ONLY; source='overlay' is the editable stratum (notes/tags) that lives OUTSIDE Odoo.
#
# Why a sibling JSON and NOT an import: aios_grid.py's contract is ZERO app-internal imports so
# it stays tenant/module-agnostic and sidesteps deploy_hf.py's import guard. A JSON next to the
# module preserves that exactly — no import, no guard interaction — while still single-sourcing
# the values (a build-time copy from another tree would reintroduce the drift we're removing).
# The file ships to the Space via deploy_hf.py INCLUDE. Edit the JSON, then run
# aios-web/verify_fields_contract.py. ---
_FIELDS_PATH = _HERE / "aios_grid_fields.json"


def _load_fields():
    if not _FIELDS_PATH.is_file():
        raise FileNotFoundError(
            f"aios_grid: canonical field contract missing at {_FIELDS_PATH}. It is the single "
            "source of truth for the grid schema and MUST ship (deploy_hf.py INCLUDE lists it)."
        )
    doc = json.loads(_FIELDS_PATH.read_text(encoding="utf-8"))
    fields = doc.get("fields") if isinstance(doc, dict) else doc
    if not isinstance(fields, list) or not fields:
        raise ValueError(f"aios_grid: {_FIELDS_PATH} has no 'fields' list.")
    return fields


FIELDS = _load_fields()


def product_fields():
    """The PRODUCT table's canonical contract (wave 15 C-TOPIC, `product_data` key in the same

    JSON). A separate accessor rather than a second module constant so the one file-read and the

    one failure mode stay shared with `FIELDS`."""
    doc = json.loads(_FIELDS_PATH.read_text(encoding="utf-8"))
    fields = (doc.get("product_data") or {}).get("fields")
    if not isinstance(fields, list) or not fields:
        raise ValueError(f"aios_grid: {_FIELDS_PATH} has no product_data.fields list.")
    return fields

# text/date fields pass through untouched; every OTHER odoo (numeric) field is rounded —
# mirrors aios-web/api/main.py _payload() exactly so embed == standalone byte-for-byte.


def _round(v):
    return round(v) if isinstance(v, (int, float)) and not isinstance(v, bool) else v


# Types a USER may create from the column menu (owner item 7, 2026-07-26). Mirrors
# customer-grid/types.ts CREATABLE_TYPES; verify_fields_contract.py holds the two in step.
#   select — a single-select with its own `options` (the "Status" a user wants to add; distinct
#            from the Odoo-sourced `status` lifecycle, which is not user-defined)
#   user   — an assignee, whose choices come from the HOST's real user list, never from here
#: `multiselect` (wave-2 item 5, 2026-07-27): the Airtable-style MULTI select — declared
#: options like `select`, but the cell holds a comma-joined SET and the row belongs to every
#: member of it (the `multi` grouping/cell contract the Cohorts column established).
#: Wave-5 item 11 (2026-07-27): `checkbox` (cell = bool; the overlay stores '1' or '') ·
#: `phone` / `email` / `url` (text-family with per-type rendering) · `rating` (top-level
#: `max`, SVG stars client-side — never emoji) · `created_time` (READ-ONLY, renders the row's
#: `_created`) · `formula` (READ-ONLY, client-computed from the row's other cells).
#: ⭐ Wave-19 R7 / contract C5: `image` — a PICTURE on a record. The cell holds a string
#: REFERENCE, never bytes: a product `code`, an `ed:<slug>` editorial asset, or `rec:<id>` for
#: something uploaded through the field (`POST /api/v1/assets/records`). Storing a reference is
#: what lets Royal's 1,142 existing SKU masters appear with nothing re-uploaded, and it keeps the
#: overlay stratum the size it is — base64 bytes in a JSON blob read on every render would be a
#: megabyte-per-row tax on the whole store. Editable by NATURE (deliberately NOT in
#: `READONLY_CUSTOM_TYPES`): the ref is what the upload endpoint hands back, and the client PATCHes
#: it through the ordinary overlay wall rather than the asset route writing cells behind it.
#: ⭐ WAVE 23 (C7) — `json` joined: a cell holding a whole DOCUMENT (an Instagram comment thread,
#: a webhook payload, a scraped blob) that opens in its own viewer instead of being flattened
#: into one unreadable line. It is EDITABLE by nature, like `image`: the value is still a plain
#: string on the wire, so it rides the ordinary overlay wall — what makes it a json field is that
#: `grid_events` REFUSES a write that does not parse (a column promising structure must not
#: silently hold something that isn't).
#: ⭐ 2026-08-07 — `link` and `rollup` JOINED (the relational wave). They ride here because
#: `UT_FIELD_TYPES` must stay a SUBSET of this set (gated in verify_api's W18-UT section) — the
#: surfaces that CREATE them are the user-table databases, where a relation between two tables is
#: a thing that exists. On the Odoo-backed Customer/Product grids there is no second user table to
#: point at, so the column menu there simply never offers one.
CUSTOM_FIELD_TYPES = {"text", "select", "multiselect", "user", "int", "currency", "pct", "date",
                      "checkbox", "phone", "email", "url", "rating", "created_time", "formula",
                      "automation", "image", "json", "link", "rollup", "code"}

#: ⭐ WAVE-27 item 13 (owner ruling R13) — the `code` field's LANGUAGES.
#:
#: R13 is explicit that this kind is "syntax-highlighted storage + language config, NO execution
#: engine". So a language is a RENDERING hint and nothing else: it selects a highlighter, it never
#: selects an interpreter, and no value here may ever grow a run path. The list is short on
#: purpose — every entry costs a highlighter the client actually has to implement, and an
#: unimplemented language would paint plain text under a label promising colour.
#:
#: `plain` is the default and the fallback, so it is never a second way to say nothing: it is the
#: honest answer for a snippet whose language the user has not chosen.
CODE_LANGUAGES = {"plain", "json", "sql", "python", "javascript", "typescript",
                  "html", "css", "markdown", "yaml", "xml", "shell"}


def _clean_code(raw):
    """The `code` field's config bag -> `{'language': ...}`, or None.



    Deliberately OPTIONAL rather than required (the `automation` posture, not `link`'s): a code

    column with no declared language is a legitimate state — it stores and highlights as plain

    text — so refusing the FIELD over a missing bag would block the ordinary create path. An

    unknown language falls back to `plain` rather than refusing, because the value is a rendering

    hint: dropping the user's column to punish a typo in a highlighter name would be the

    disproportionate half of the fail-closed rule.



    ⚠ `plain` RETURNS NONE, and that is what makes the control reversible rather than one-way.

    Absent already means plain, so storing `{'language': 'plain'}` would be the default wearing a

    second name (the `kanbanClamp` law). But the patch path resolves an OMITTED key to the

    previous value — so if plain were merely omitted by the client, switching a column back from

    SQL to Plain text would keep storing SQL and read as a control that does not save. Sending

    the bag explicitly and having it evaluate to None here means: omit = keep, plain = clear.

    """
    if not isinstance(raw, dict):
        return None
    lang = str(raw.get('language') or 'plain').strip().lower()
    if lang not in CODE_LANGUAGES or lang == 'plain':
        return None
    return {'language': lang}
#: User-created types whose CELLS are read-only: their values are computed (formula — client
#: side, any error degrades to BLANK) or system-owned (created_time = the row's `_created`).
#: Emitted with the cohort column's read-only mechanism — `source: 'odoo'` + `derived` — so
#: the client never offers an editor and the host never accepts a cell write for them.
READONLY_CUSTOM_TYPES = {"created_time", "formula"}
MAX_FIELD_OPTIONS = 50
MAX_FORMULA_LEN = 500
#: `rating` bounds. Airtable caps at 10; below 2 a rating is a checkbox.
RATING_MAX_DEFAULT, RATING_MAX_MIN, RATING_MAX_MAX = 5, 2, 10

#: Key prefix of a FORMULA-MEASURE column (owner item 7, 2026-07-27): a user-created field that
#: IS a measure over a window — `Sales · the last 90 days` as a column. Mirrors the client's
#: `measure_` keys in CustomerGrid.createField. Distinct from `custom_` because the two strata
#: could not be more different: `custom_` is the EDITABLE overlay (user-typed values), while a
#: measure field is READ-ONLY and its values are computed by the host per render.
MEASURE_FIELD_PREFIX = "measure_"
#: The numeric types a measure can render as (semantic._FORMAT_TYPE's range).
MEASURE_FIELD_TYPES = {"currency", "int", "pct"}


def _clean_options(raw):
    """Choices for a `select`: strings, trimmed, de-duplicated case-insensitively, capped.

    Mirrors types.ts parseOptions — a choice list that means one thing in the picker and

    another in the filter dropdown is a column with two vocabularies."""
    out, seen = [], set()
    for v in list(raw or [])[:MAX_FIELD_OPTIONS * 2]:
        if not isinstance(v, (str, int, float)) or isinstance(v, bool):
            continue
        s = str(v).strip()[:120]
        if not s or s.lower() in seen:
            continue
        seen.add(s.lower())
        out.append(s)
    return out[:MAX_FIELD_OPTIONS]


def _clean_option_colors(raw, options):
    """Choice-label -> #RRGGBB, limited to the field's canonical option vocabulary."""
    if not isinstance(raw, dict):
        return {}
    supplied = {}
    for label, color in raw.items():
        if not isinstance(label, str) or not isinstance(color, str):
            continue
        clean = color.strip().upper()
        if re.fullmatch(r"#[0-9A-F]{6}", clean):
            supplied[label.strip().lower()] = clean
    out = {}
    for option in options or []:
        color = supplied.get(str(option).strip().lower())
        if color:
            out[str(option)] = color
    return out


def _choice_appearance(raw, options):
    """Validated select-family appearance. Absent colour toggle means legacy-on."""
    if not isinstance(raw, dict):
        return {}
    out = {}
    if isinstance(raw.get("colorCodeOptions"), bool):
        out["colorCodeOptions"] = raw["colorCodeOptions"]
    colors = _clean_option_colors(raw.get("optionColors"), options)
    if colors:
        out["optionColors"] = colors
    return out


def _clean_rating_max(raw):
    """A rating's star count, bounded. Anything unusable is the default, not a refusal — the

    field still holds its 1..max integers either way."""
    try:
        return max(RATING_MAX_MIN, min(int(raw), RATING_MAX_MAX))
    except (TypeError, ValueError):
        return RATING_MAX_DEFAULT


#: The field types a number-style display format may apply to. `formula` is here because its
#: RESULT is a number the client renders; `pct` already renders in points and takes decimals.
_NUMBER_FORMAT_TYPES = {"int", "currency", "pct", "formula"}


def _clean_format(raw, ftype):
    """Per-type DISPLAY format (wave-5 item 10), fail-closed: unknown keys are DROPPED, wrong

    types return None (the property is simply absent). Rendering-only — a format can change how

    a value reads, never what it is, which is why this needs no parity gate of its own."""
    if not isinstance(raw, dict):
        return None
    out = {}
    if ftype in _NUMBER_FORMAT_TYPES:
        if isinstance(raw.get("thousands"), bool):
            out["thousands"] = raw["thousands"]
        if raw.get("decimals") is not None:
            try:
                d = int(raw["decimals"])
            except (TypeError, ValueError):
                d = None
            if d is not None and 0 <= d <= 4:
                out["decimals"] = d
        if isinstance(raw.get("abbrev"), bool):
            out["abbrev"] = raw["abbrev"]
    elif ftype in ("date", "created_time"):
        if isinstance(raw.get("time"), bool):
            out["time"] = raw["time"]
        if raw.get("tz") in ("local", "utc"):
            out["tz"] = raw["tz"]
    return out or None


def _clean_permissions(raw):
    """`{edit: 'everyone' | 'creator' | 'admins'}` or None (wave-5 item 1). WHO may set it is

    the host handler's business (creator/admin, enforced fail-closed there); this validates

    only the shape, like every other property here."""
    if isinstance(raw, dict) and raw.get("edit") in ("everyone", "creator", "admins"):
        return {"edit": raw["edit"]}
    return None


#: ⛔ THE CHARSET MUST ADMIT EVERY TOKEN THE CLIENT ENGINE PARSES, or a legal formula is
#: refused by a wall that is supposed to be structural (2026-08-03).
#:
#: This regex was written when a formula was arithmetic over refs. On 2026-07-31 the client
#: engine (owner item 2) gained STRING literals, `&` concatenation and `^` — CONCATENATE, TEXT,
#: LEFT/RIGHT/MID, and any `IF(cond, "yes", "no")`. This list was never widened to match, so
#: every such formula died here: `field_upsert` refused the create, and `fields_from_workspace`
#: dropped the column on read. Nothing went red — a refused create looks like a quiet failure
#: and a dropped column looks like a column nobody made.
#:
#: Found by trying to ship the owner's own Buy signal formula, which is `IF(..., "Buy now",
#: "OK")` and could not be created through the product UI at all.
#:
#: ⚠ IT IS STILL STRUCTURAL, and deliberately not a second grammar — that is the filter_sql-class
#: drift risk this file's docstring names. The charset says which characters may appear; the
#: engine says what they mean. `_quotes_balanced` below is the one structural rule the quote
#: character brings with it.
_FORMULA_CHARS = re.compile(r"^[\w\s{}()+\-*/.,<>=!^&\"]*$")
_FORMULA_REF = re.compile(r"\{([^{}]*)\}")


def _quotes_balanced(s):
    """An even number of `"` — the structural half of string support.



    Sound because the engine's own escape is Excel's: `""` inside a string is one quote, and it

    contributes TWO characters. So a well-formed expression always has an even count and an

    unterminated string always has an odd one. What a balanced pair MEANS is the engine's

    business, exactly as with parentheses.

    """
    return s.count('"') % 2 == 0


#: Wave-18 C5-AUTOFIELD. `kind` is a whitelist because an unknown kind would be a column that
#: silently never runs; `source` names where the run's subject comes from.
AUTOMATION_KINDS = {"instagram_profile"}
AUTOMATION_SOURCES = {"record_url_field", "self"}
MAX_AUTOMATION_SETTINGS = 12


def _clean_automation(raw, valid_keys=None, flow_ids=None):
    """Validate an `automation` config bag. Returns the clean dict, or None when there is

    nothing valid to store (the column then renders as unconfigured — never invented).



    ⛔ WAVE 22 (contract C8, owner item 5) — NO FIELD WITHOUT A FLOW. With `flow_ids` given

    (the WRITE path: grid_events / user_tables pass the tenant's automation-definition ids),

    the bag MUST carry a `flowId` naming one of them — absent or naming a deleted flow is

    refused, the same fail-closed direction as a formula ref that names no field. With

    `flow_ids=None` (the READ path, `fields_from_workspace`) the law is NOT applied: a column

    stored before the law must keep projecting — enforcement at read time would vaporise it,

    which is the `_clean_formula` write/read split exactly.

    """
    if not isinstance(raw, dict):
        return None
    kind = str(raw.get('kind') or '').strip()
    if kind not in AUTOMATION_KINDS:
        return None
    source = str(raw.get('source') or 'record_url_field').strip()
    if source not in AUTOMATION_SOURCES:
        source = 'record_url_field'
    out = {'kind': kind, 'source': source}
    flow = str(raw.get('flowId') or '').strip()[:40]
    if flow_ids is not None and (not flow or flow not in flow_ids):
        return None
    if flow:
        out['flowId'] = flow
    url_field = str(raw.get('urlField') or '').strip()[:80]
    # fail closed on a ref that does not exist, exactly as _clean_formula does at WRITE time
    if url_field and (valid_keys is None or url_field in valid_keys):
        out['urlField'] = url_field
    settings = {}
    for k, v in list((raw.get('settings') or {}).items())[:MAX_AUTOMATION_SETTINGS]:
        if isinstance(v, bool) or isinstance(v, (int, float)):
            settings[str(k)[:40]] = v
        elif isinstance(v, str):
            settings[str(k)[:40]] = v[:200]
    if settings:
        out['settings'] = settings
    return out


def _clean_formula(raw, valid_keys=None):
    """STRUCTURAL passthrough for a formula field's expression (wave-5 item 9).



    Meaning is NOT checked here: the CLIENT engine owns the grammar (arithmetic over `{field}`

    refs, ABS/ROUND/MIN/MAX/IF), and any evaluation error degrades to a BLANK cell — never a

    wrong number. That is the `_clean_window` split one stratum up, and deliberately NOT a

    Python mirror of the grammar: a second engine is the filter_sql-class drift risk.

    Structure IS checked — charset, length, balanced parens, BALANCED QUOTES, well-formed

    non-empty `{refs}` — and at WRITE time (`valid_keys` given) every ref must name a field this

    table has, fail closed. At READ time refs are left alone: a referenced field deleted later

    must blank the CELLS, not vaporise the column.



    ⚠ A `{ref}` INSIDE A STRING LITERAL is still checked against `valid_keys` at write time, so

    `IF(x, "see {notafield}", "")` is refused. That is a false rejection and it is the

    fail-closed direction: the alternative is teaching this function where strings begin and

    end, which is the second grammar the paragraph above refuses to write.

    """
    if not isinstance(raw, str):
        return None
    s = raw.strip()
    if not s or len(s) > MAX_FORMULA_LEN or not _FORMULA_CHARS.match(s):
        return None
    depth = 0
    for ch in s:
        if ch == "(":
            depth += 1
        elif ch == ")":
            depth -= 1
            if depth < 0:
                return None
    if depth:
        return None
    if not _quotes_balanced(s):
        return None
    refs = _FORMULA_REF.findall(s)
    leftover = _FORMULA_REF.sub("", s)
    if "{" in leftover or "}" in leftover:      # unbalanced / nested braces
        return None
    if any(not r.strip() for r in refs):        # a `{}` ref names nothing
        return None
    if valid_keys is not None and any(r not in valid_keys for r in refs):
        return None
    return s


def _field_extras(saved, ftype):
    """createdBy / permissions / format / scope — the validated passthrough the created strata

    share (wave 5). `createdBy` is only ever WRITTEN host-side (the handler stamps it); here it

    is carried so the client can gate its menus and the handler can enforce against it.

    `scope` (wave-6 item 9): 'cohort' marks a field created as cohort-specific — the handler

    stamps it at create (cohort page only) and preserves it like createdBy; carried here so the

    client can label the field, filtered OUT of other pages by fields_from_workspace."""
    out = {}
    who = saved.get("createdBy")
    if isinstance(who, str) and who.strip():
        out["createdBy"] = who.strip()[:80]
    perms = _clean_permissions(saved.get("permissions"))
    if perms:
        out["permissions"] = perms
    fmt = _clean_format(saved.get("format"), ftype)
    if fmt:
        out["format"] = fmt
    if saved.get("scope") == "cohort":
        out["scope"] = "cohort"
    corrected_from = saved.get("labelCorrectedFrom")
    correction_id = saved.get("labelCorrectionId")
    if (isinstance(corrected_from, str) and corrected_from.strip()
            and isinstance(correction_id, str) and correction_id.strip()):
        out["labelCorrectedFrom"] = corrected_from.strip()[:120]
        out["labelCorrectionId"] = correction_id.strip()[:180]
    return out


#: The DERIVED column listing the cohorts a customer is in (owner, 2026-07-27).
#:
#: NOT in `aios_grid_fields.json`, deliberately. That contract is per-TABLE and shared with the
#: standalone API and the dev sample; a cohort is per-USER, so the column exists exactly when the
#: caller has cohorts — the same condition under which the `__cohort__` FILTER field is offered.
#: Putting it in the canonical contract would mean an always-present column that is empty for
#: everyone else, plus three consumers to keep in step for a value none of them can produce.
COHORT_COLUMN = 'cohorts'


def cohort_field(label='Locked views'):
    """The derived membership column's descriptor.



    ⚠ WAVE 17 item 14 / C-STR — THE LABEL AND THE NOTE SPEAK THE NEW VOCABULARY; THE KEY DOES

    NOT. `COHORT_COLUMN` is still `'cohorts'` and the function is still `cohort_field`, because

    every stored view that shows or groups by this column names it by KEY, and every gate in

    two runtimes names the function. The owner renamed a CONCEPT ("we should stop calling it

    Cohort, but locked instead"), which is a change to what a reader sees — renaming the

    identifiers would break saved views to change a word nobody reads.



    `source: 'odoo'` is doing ONE job here and it is not provenance: the client keys editability

    off `source == 'overlay'`, and `_cl_handle_grid_event` accepts cell writes only for overlay

    keys. So 'odoo' is what makes this column READ-ONLY at both ends. `derived: True` is what

    stops the column menu calling it a "source field" on that basis.



    ⚠ `filterable: False`, and the replacement is the `Where [Cohort] […]` CONDITION, not another

    column. Text ops over a joined string would ALMOST work and disagree at the edges — `contains

    "VIP"` also matches a cohort called "VIP club" — and a filter that is nearly right is worse

    than one that is absent.

    """
    return {
        'key': COHORT_COLUMN, 'label': label, 'type': 'text', 'source': 'odoo',
        'default': False, 'filterable': False, 'derived': True, 'multi': True,
        'note': 'The locked views this customer is in, newest first. A locked view holds a SET, '
                'so grouping by this column puts a customer under EVERY view they are in — the '
                'group counts therefore add up to more than the record count, which stays the '
                'number of distinct customers. Read-only: membership changes only by adding or '
                'removing customers on the locked view itself.',
    }


def cohort_cells(cohorts, allowed_pids=None):
    """`{pid: {'cohorts': 'Q3 calls, Lost'}}` from `[{id,name,pids}, ...]`.



    Built per render from the caller's OWN cohorts and handed to `rows_from_pool` as `derived`,

    never merged into the cached pool rows — those are shared across users, and stamping one

    user's cohorts onto them would leak the membership to everybody else on the next render.

    """
    cells = {}
    for c in cohorts or []:
        # ⚠ The COMMA is the separator the client splits on to group a customer into EVERY
        # cohort they are in, so it cannot also occur inside a name. Cohort names are free text
        # ("Q3 calls, west" is a name somebody will type), so a comma is replaced here rather
        # than left to break the split silently — one group called "Q3 calls" and another called
        # "west" would be two lists that do not exist. The cost is cosmetic and confined to the
        # cell; the Cohort page still shows the name the user typed.
        name = str(c.get('name') or c.get('id') or '').replace(',', ' ').strip()
        for pid in c.get('pids') or ():
            if allowed_pids is not None and pid not in allowed_pids:
                continue
            cells.setdefault(pid, []).append(name)
    return {pid: {COHORT_COLUMN: ', '.join(names)} for pid, names in cells.items()}


def clean_measure_field(raw, offered):
    """Validate one UNTRUSTED formula-measure field (owner item 7) against the caller's OFFER.



    `offered` is `{measure key: {label, type}}` from `measure_filter.measure_fields(team_id)` —

    the same admission the condition builder uses, so a field can only name a measure this

    caller could also filter by. Returns the canonical stored shape, or None (fail closed).



    `source:'odoo'` + `derived:True` is the cohort column's read-only mechanism, reused:

    the client keys editability off `source == 'overlay'` and the host accepts cell writes only

    for overlay keys, so a measure column cannot be typed into at either end. `filterable:False`

    because the REPLACEMENT is the measure CONDITION with the same measure and window — the

    governed, gate-proved path (CG-8/CG-12) — not text ops over a derived cell.

    """
    if not isinstance(raw, dict):
        return None
    key = str(raw.get("key") or "")
    if not key.startswith(MEASURE_FIELD_PREFIX) or len(key) > 80:
        return None
    spec = raw.get("measure")
    if not isinstance(spec, dict):
        return None
    m = (offered or {}).get(spec.get("key"))
    if not m:
        return None                    # not admitted for this caller -> fail closed
    window = _clean_window(spec.get("window"))
    if window is None:
        return None                    # a measure column with no period is not a column
    mtype = m.get("type") if m.get("type") in MEASURE_FIELD_TYPES else "currency"
    return {
        "key": key,
        "label": str(raw.get("label") or m.get("label") or "Measure")[:120],
        "type": mtype,
        "source": "odoo",
        "default": True,
        "custom": True,
        "derived": True,
        "filterable": False,
        "agg": "sum" if mtype in ("currency", "int") else None,
        "note": str(raw.get("note") or "")[:2000],
        "measure": {"key": str(spec.get("key"))[:80], "window": window},
    }


def measure_fields_of(fields):
    """The formula-measure columns among `fields` — the ones whose values the host must compute

    per render (see `rows_from_pool`'s `derived`)."""
    return [f for f in fields or [] if isinstance(f.get("measure"), dict)]


def fields_from_workspace(workspace=None, cohorts=False, scope_key=None, fields_base=None):
    """Overlay persisted notes/custom fields onto the immutable source-field contract.



    `cohorts=True` appends the derived cohort column — see `cohort_field`. Appended LAST and

    `default: False`, so it never displaces a column somebody already reads; the Fields menu is

    where you turn it on.



    Three saved strata pass through: notes on base fields, `custom_` overlay fields (editable),

    and `measure_` formula-measure fields (read-only, host-computed — owner item 7). A measure

    field was validated against the caller's measure OFFER when it was written

    (`clean_measure_field`); here only its SHAPE is re-checked, because this module has zero

    app-internal imports and cannot know the offer. A measure that has since become

    unanswerable (a BU scope on a company-level measure) degrades to a BLANK column at value

    time, never to an error.



    `scope_key` (wave-6 item 9) names the PAGE doing the asking ('customer' / 'cohort'). A

    saved def carrying `scope` is emitted only when it matches — so a cohort-specific field

    never appears on the Customer table. FAIL-CLOSED: a caller that passes no scope_key sees

    only unscoped (global) fields; base contract fields are never scoped.



    `fields_base` (wave 16 C-TOPIC) — the canonical contract to overlay onto. Absent = the

    CUSTOMER contract (`FIELDS`), byte-identical to before the parameter existed; the product

    surface passes `product_fields()`. The workspace dict a caller hands in must already be the

    matching table object's bucket — this function cannot tell a customer overlay from a

    product one, which is exactly why the buckets are separate stores.

    """
    saved = dict((workspace or {}).get("fields") or {})
    out = []
    base_fields = fields_base if fields_base is not None else FIELDS
    base_keys = {field["key"] for field in base_fields}
    for base in base_fields:
        meta = saved.get(base["key"]) or {}
        field = dict(base)
        if isinstance(meta.get("note"), str):
            field["note"] = meta["note"][:2000]
        # Wave-5 item 10: a saved DISPLAY format on any base field (a preset included) — how a
        # number or date READS, per user. Rendering-only, so this is the whole acceptance.
        fmt = _clean_format(meta.get("format"), base.get("type"))
        if fmt:
            field["format"] = fmt
        # ⭐ W29-T83 — the saved COLUMN SUMMARY, the read half of the write door in
        # `grid_events.field_upsert`. Without this the value round-trips into the store and is
        # never served back, which looks exactly like a write that never happened
        # ([[read-path-cannot-witness-write-path]]). Absent = whatever the contract declares.
        if meta.get("agg") in FIELD_AGGS:
            field["agg"] = meta["agg"]
        # PRESET measure fields (wave-2 item 8): a preset+measure base may take a saved
        # window/label override. Wave 6 deleted every preset member (the owner's
        # no-buildable-presets rule) so this branch is currently MEMBERLESS — kept as the
        # measure_ path's twin for any future preset-carrying contract, and because deleting
        # it would silently change what a re-added preset means.
        if base.get("preset") and isinstance(base.get("measure"), dict):
            saved_measure = meta.get("measure") if isinstance(meta.get("measure"), dict) else {}
            window = _clean_window(saved_measure.get("window"))
            if window is not None:
                field["measure"] = {"key": base["measure"]["key"], "window": window}
                if isinstance(meta.get("label"), str) and meta["label"].strip():
                    field["label"] = meta["label"][:120]
        out.append(field)
    for key, field in saved.items():
        if key in base_keys or not isinstance(field, dict):
            continue
        if field.get("scope") and field.get("scope") != scope_key:
            continue          # a cohort-specific field on another page (wave-6 item 9)
        if (str(key).startswith(MEASURE_FIELD_PREFIX)
                and isinstance(field.get("measure"), dict)):
            window = _clean_window(field["measure"].get("window"))
            mkey = str(field["measure"].get("key") or "")
            if window is None or not mkey:
                continue
            mtype = field.get("type") if field.get("type") in MEASURE_FIELD_TYPES else "currency"
            out.append({
                "key": str(key)[:80],
                "label": str(field.get("label") or "Measure")[:120],
                "type": mtype,
                "source": "odoo",
                "default": bool(field.get("default", True)),
                "custom": True,
                "derived": True,
                "filterable": False,
                "agg": "sum" if mtype in ("currency", "int") else None,
                "note": str(field.get("note") or "")[:2000],
                "measure": {"key": mkey[:80], "window": window},
                **_field_extras(field, mtype),
            })
            continue
        if not str(key).startswith("custom_"):
            continue
        ftype = field.get("type")
        if ftype not in CUSTOM_FIELD_TYPES:
            continue
        if ftype in READONLY_CUSTOM_TYPES:
            # Wave-5 items 9/11: the read-only user-created pair. Emitted with the cohort
            # column's mechanism (source 'odoo' + derived) so the client never offers an
            # editor and the host's overlay-write guard excludes them by construction.
            # FILTERABLE since wave 6 (owner item 6): their values live client-side
            # (formula computes over the row, created_time renders `_created`), this
            # table's counts are client-mode, and the windowed count path never sees
            # these tables — so the client engine answers them soundly.
            entry = {
                "key": str(key)[:80],
                "label": str(field.get("label") or "Untitled")[:120],
                "type": ftype,
                "source": "odoo",
                "derived": True,
                "filterable": True,
                "default": bool(field.get("default", True)),
                "custom": True,
                "note": str(field.get("note") or "")[:2000],
                **_field_extras(field, ftype),
            }
            if ftype == "formula":
                formula = _clean_formula(field.get("formula"))
                if formula is None:
                    continue          # a formula field without a formula is nothing
                entry["formula"] = formula
            out.append(entry)
            continue
        if field.get("source") != "overlay":
            continue
        options = (_clean_options(field.get("options"))
                   if ftype in ("select", "multiselect") else [])
        if ftype in ("select", "multiselect") and not options:
            # A select with no surviving choices can never hold a value. Dropping the COLUMN
            # would lose the user's data; degrading it to text keeps every stored value
            # readable and lets them re-add choices.
            ftype = "text"
        out.append({
            "key": str(key)[:80],
            "label": str(field.get("label") or "Untitled")[:120],
            "type": ftype,
            "source": "overlay",
            "default": bool(field.get("default", True)),
            "custom": True,
            # WAVE-29 C7: the whole vocabulary, not the `{"sum"}` literal that was here — a
            # picker offering Average against a projection that only passes Sum through is the
            # silent half of this feature.
            "agg": field.get("agg") if field.get("agg") in FIELD_AGGS else None,
            "note": str(field.get("note") or "")[:2000],
            **({"options": options} if ftype in ("select", "multiselect") else {}),
            **(_choice_appearance(field, options)
               if ftype in ("select", "multiselect") else {}),
            # comma-joined SET semantics (the Cohorts column's contract): the row belongs to
            # every member, groups count it under each, the toolbar count stays distinct.
            **({"multi": True} if ftype == "multiselect" else {}),
            **({"max": _clean_rating_max(field.get("max"))} if ftype == "rating" else {}),
            **({"automation": _clean_automation(field.get("automation"))}
               if ftype == "automation" and _clean_automation(field.get("automation")) else {}),
            # ⭐ WAVE-27 item 13 (R13) — the code column's language rides the wire, because the
            # highlighter is chosen per column and the client cannot infer a language from a
            # string. Absent = `plain`, which is what an unconfigured code column renders as.
            **({"code": _clean_code(field.get("code"))}
               if ftype == "code" and _clean_code(field.get("code")) else {}),
            **_field_extras(field, ftype),
        })
    if cohorts and not any(f.get('key') == COHORT_COLUMN for f in out):
        # ⚠ The emptiness check is WAVE 19's, and it is about the topics R9 opened this column to.
        # A user table's field keys are slugged from whatever its creator typed, so a column
        # literally called "Cohorts" produces the key `cohorts` — and appending here unguarded
        # would put TWO fields with one key on the wire. The client indexes fields by key, so the
        # duplicate does not error: it silently paints one column's values under the other's
        # header. The user's own column wins; the derived one steps aside rather than shadowing it.
        out.append(cohort_field())
    return out


def rows_from_pool(pool_rows, fields=None, overlays=None, derived=None):
    """Map customer_data.pool() dicts -> the API row shape the grid expects:

    pid + each Odoo field (numeric fields rounded, text/date passed through) + the

    persisted external overlay. Mirrors the standalone API payload contract.



    `derived` is `{pid: {key: value}}` for columns the HOST computes per render rather than

    reads off the pool row — today just the cohort column. A separate argument from `overlays`

    on purpose: `overlays` is the PERSISTED user stratum, and putting a value there that is

    never written back would make the dict mean two things.

    """
    fields = fields or FIELDS
    odoo_fields = [field for field in fields if field["source"] == "odoo"]
    overlay_fields = [field for field in fields if field["source"] == "overlay"]
    derived_keys = [field["key"] for field in fields if field.get("derived")]
    overlays = overlays or {}
    derived = derived or {}
    out = []
    for r in pool_rows:
        pid = r.get("pid")
        # `_created` (wave-5 item 11) rides every row like `pid` does — the datum the
        # `created_time` field type renders, regardless of that field's own key. Not a Field:
        # it has no column of its own until a user creates one. `lat`/`lon` (wave-7 W11) ride
        # the same way: the Map VIEW's data, nullable, deliberately not a column.
        row = {"pid": pid, "_created": r.get("_created") or "",
               "lat": r.get("lat"), "lon": r.get("lon")}
        for field in odoo_fields:
            k = field["key"]
            if field.get("derived"):
                continue          # not on the pool row — filled from `derived` below
            v = r.get(k)
            row[k] = v if field["type"] in {"text", "status", "date"} else _round(v)
        saved = overlays.get(str(pid), {}) or {}
        for field in overlay_fields:
            row[field["key"]] = saved.get(field["key"], "")
        got = derived.get(pid) or {}
        for k in derived_keys:
            # '' not None: a customer in no cohort has an EMPTY cohort list, and `is empty` on a
            # text column is the question somebody will ask of it.
            row[k] = got.get(k, "")
        out.append(row)
    return out


# --- the FILTER-TREE contract (mirrors customer-grid/types.ts) ---------------
# Ops the Airtable-parity condition builder can emit. isEmpty/isNotEmpty are
# VALUE-FREE (they legitimately carry no value and must never be dropped for it).
FILTER_OPS = {'contains', 'doesNotContain', 'eq', 'neq', 'isEmpty', 'isNotEmpty',
              'gt', 'gte', 'lt', 'lte', 'between', 'within',
              # Wave 2026-08-02 (C-OPS): RANK operators — evaluated as a SET pass over the
              # sibling-filtered domain by the client engine (useVisibleRows). The validator
              # accepts them like any op (structural, not semantic); filter_sql REFUSES to
              # compile them to row SQL (a per-row WHERE cannot express Top-N). aboveAvg /
              # belowAvg are VALUE-FREE; the rest encode their argument in `value` as a
              # string int (topN/bottomN 1..10000, inTopPct/inBottomPct 1..100,
              # inQuartile 1..4, inDecile 1..10). Deliberately NOT in MEASURE_OPS: on a
              # measure-carrying column they rank the field's own derived values.
              'topN', 'bottomN', 'inTopPct', 'inBottomPct',
              'aboveAvg', 'belowAvg', 'inQuartile', 'inDecile'}
#: The RESERVED pseudo-column of a cohort-membership leaf (owner item 5, 2026-07-26). It is not
#: a Field and never will be — a cohort is a hand-curated SET, so making it a column would mean
#: a cell per row per cohort. Mirrors customer-grid/types.ts COHORT_FIELD.
COHORT_FIELD = '__cohort__'
#: Ops a cohort leaf may carry (owner, 2026-07-27): set operators over a SET of cohorts.
#: Anything else is dropped. Deliberately DISJOINT from FILTER_OPS — see types.ts COHORT_OPS:
#: a set op reaching a column leaf would fall through the client engine's switch to "no
#: narrowing", so keeping the vocabularies apart makes the existing fail-closed drop do the work.
COHORT_OPS = {'anyOf', 'allOf', 'noneOf'}
#: The single-cohort ops this leaf shipped with, kept as PERMANENT aliases and REWRITTEN here:
#: `is part of [one]` is `is any of [that one]`, so a saved view keeps answering and upgrades the
#: next time it is written. Mirrors types.ts COHORT_OP_ALIASES.
COHORT_OP_ALIASES = {'eq': 'anyOf', 'neq': 'noneOf'}
#: How many cohorts one condition may name. Mirrors types.ts MAX_COHORT_IDS.
MAX_COHORT_IDS = 20


def parse_cohort_ids(value):
    """The cohorts a leaf names, parsed out of `value`. Mirrors types.ts `cohortIds()`.



    Comma-separated in one string because `FilterRule.value` is what all four layers persist and

    round-trip, and a one-element list is byte-identical to what the single-cohort leaf already

    stored — so every shipped view parses with no migration. Safe because a cohort id is built

    from `[a-z0-9_]` only (modules/cohort.new_id), so a comma cannot occur inside one.

    """
    out = []
    for raw in ('' if value is None else str(value)).split(','):
        cid = raw.strip()[:120]
        if not cid or cid in out:
            continue
        out.append(cid)
        if len(out) >= MAX_COHORT_IDS:
            break
    return out
# Airtable allows 3 nesting levels (root conditions -> group -> group), then grays
# the button out. MAX_FILTER_DEPTH in types.ts must stay in lock-step with this.
MAX_FILTER_DEPTH = 3
MAX_FILTER_NODES = 100      # total nodes across the whole tree
MAX_FILTER_SIBLINGS = 50    # per level


#: Shape of a measure condition's date window. aios_grid has ZERO app-internal imports by
#: design, so it does NOT know the window VOCABULARY — `harness/windows.py` owns that, mirrored
#: in `customer-grid/windows.ts`, and a third copy here is exactly the drift those two already
#: need a gate to prevent. This validates SHAPE only.
WINDOW_MAX_N = 3650


def _clean_window(raw):
    """Structural passthrough for a measure condition's `{kind, n?, from?, to?}` window.



    Meaning is NOT checked here: an unrecognised `kind` survives this function and is REFUSED by

    `harness.measure_filter.resolve_rule`, the layer that owns the vocabulary. Splitting it this

    way keeps the grid module reusable and keeps one definition of what "last quarter" means.

    """
    if not isinstance(raw, dict):
        return None
    kind = raw.get('kind')
    if not isinstance(kind, str) or not kind or len(kind) > 40:
        return None
    out = {'kind': kind}
    if raw.get('n') is not None:
        try:
            out['n'] = max(1, min(int(raw['n']), WINDOW_MAX_N))
        except (TypeError, ValueError):
            return None
    for side in ('from', 'to'):
        if raw.get(side) not in (None, ''):
            out[side] = str(raw[side])[:32]
    return out


def _clean_rhs(raw, valid_keys):
    """CG-9 — validate `{kind, colId, window?}`, the "compare against another attribute" side.



    SHAPE and KEY only: `colId` must be something this table has (the caller widens `valid_keys`

    with the measure keys, exactly as it does for the left side), and a measure rhs must carry a

    window. What the window MEANS is `harness/windows.py`'s business, same split as `_clean_window`.

    """
    if not isinstance(raw, dict):
        return None
    kind = raw.get('kind')
    if kind not in ('field', 'measure', 'stat'):
        return None
    if kind == 'stat':
        # A STATISTIC carries no column: the population is the comparand. Shape only — which
        # statistics exist is `harness/measure_filter.STATS`'s business, and an unrecognised one
        # is REFUSED there rather than guessed at, exactly like an unrecognised window kind.
        stat = raw.get('stat')
        if not isinstance(stat, str) or not stat or len(stat) > 24:
            return None
        return {'kind': 'stat', 'stat': stat}
    col = raw.get('colId')
    if col not in valid_keys:
        return None
    out = {'kind': kind, 'colId': col}
    if kind == 'measure':
        window = _clean_window(raw.get('window'))
        if window is None:
            return None            # a measure comparand with no period is not a question
        out['window'] = window
    return out


#: View DISPLAY MODES beside the grid (wave-6 item 10; 'map' wave-7 W11; 'dashboard' wave-8
#: I19). Mirrors customer-grid/types.ts DISPLAY_MODES; 'grid' is what an absent/unknown
#: display means, so it is never stored.
#: ⚠ 'dashboard' is RETAINED FOREVER (wave-9 I10, contract C2). The owner renamed the mode to
#: "Chart" (a Dashboard MODULE is coming and the two would collide), but this set is the
#: gatekeeper for a STORED value: `_clean_display` DROPS an unknown mode, so removing
#: 'dashboard' here would silently downgrade every already-saved chart view to grid — and live
#: views are sitting in mode:'dashboard' right now (wave 8's own close-out records one). The
#: rename is therefore a stored-value MIGRATION, not a constant rename: accept 'dashboard' on
#: READ forever, only ever WRITE 'chart'.
#: ⭐ WAVE-27 item 8 (owner ruling R2, contract C3): 'swipe' — a DECK of the records whose bound
#: single-select is EMPTY, triaged one at a time by swiping left or right into two of that
#: field's options. Landed here FIRST and in the same change as the client registry, which is
#: the whole reason the two modes above it needed a staged hold: `_clean_display` DROPS an
#: unknown mode, so a client that offers a mode this set does not carry lets a user build a view
#: that silently reverts to a grid on the next read.
#: ⭐⭐ WAVE-29 R6/R7 (owner item 10, contract C3) — 'form', which CLOSES D-90. The client has
#: carried `form` in its union, with an icon, a label and a tone, since wave 23; this set never
#: did, so `_clean_display` DROPPED both the mode and the `display.form` spec on every write —
#: while `routes_forms.py` reads exactly that key to serve the public submit door. The public door
#: has therefore been live and UNREACHABLE for two waves: not broken, just impossible to point at
#: anything. The mirror is one name, and it is the half nobody could see was missing because
#: BOTH sides were individually consistent.
#: ⚠ Being a legal stored mode is NOT the same as being offered: `form` is deliberately held out
#: of the client's `CREATABLE_MODES` until `CustomerGrid` mounts a renderer for it (the hold law
#: written into `iconShapes.ts`, and now machine-enforced in BOTH directions by
#: `verify_icons.py::mode_parity` — offering an unmounted mode is red, and mounting an unoffered
#: one is red too, so the hold cannot outlive its reason the way wave 27's did).
DISPLAY_MODES = {'grid', 'list', 'calendar', 'kanban', 'map', 'dashboard', 'chart',
                 'timeseries', 'catalog', 'swipe', 'form'}

#: ⭐⭐ WAVE-29 R7 (owner item 10, contracts C3/C4) — THE FORM INTERFACE's stored spec, at
#: `views[<id>].config.display.form`. The public door (`aios-web/api/routes_forms.py`) has read
#: exactly this key since wave 23 and NOTHING HAS EVER BEEN ABLE TO WRITE IT: `form` was not a
#: legal mode and this function had no branch for the key, so every spec a client sent was dropped
#: on the way in. That is D-90 stated precisely — not a broken feature, an unreachable one.
#:
#: ⛔ THE TOKEN IS NOT HERE, AND IT NEVER WILL BE. A share token that rides the wire is a token a
#: browser can CHOOSE, and `routes_forms._resolve` walks tenants and answers with the FIRST match —
#: so one tenant setting its token to another tenant's value would silently receive that tenant's
#: submissions. The token is minted server-side and lives in a bucket no client write can reach;
#: `_clean_form` drops any `token` key that arrives here, rather than validating its shape.
FORM_ACCESS = ('public', 'emails')
MAX_FORM_FIELDS = 60
MAX_FORM_EMAILS = 200
MAX_FORM_TITLE, MAX_FORM_DESC, MAX_FORM_SUBMIT = 120, 1000, 60
MAX_FORM_EMAIL = 254

#: The field types a form may COLLECT, as an ALLOW-LIST rather than a list of exclusions — the
#: fail-closed direction, because the cost of the two mistakes is not symmetric. A type missing
#: from here is a question the builder cannot ask yet; a type wrongly present is a public door
#: writing values the column cannot mean (an `image` with no upload channel, a `json` document
#: typed into a text box, a `link` naming a record id a stranger guessed).
#: ⚠ NOT sufficient on its own, and the reason is a shape this codebase has been bitten by before:
#: a METRIC bag rides ANY field type (`core/user_tables.py` — `metric` is not a field kind), so an
#: `int` column can be machine-computed while passing this list. `routes_forms` therefore asks
#: `user_tables.is_computed_cell` as well — one evaluator for "is this computed", reused rather
#: than re-derived ([[one-evaluator-per-question]]).
#: Client mirror: `customer-grid/FormInterface.tsx` FORM_FIELD_TYPES; `verify_forms.py` compares
#: the two files name-for-name.
FORM_FIELD_TYPES = ('text', 'select', 'multiselect', 'int', 'currency', 'pct', 'date',
                    'checkbox', 'phone', 'email', 'url', 'rating')

#: Deliberately looser than a full RFC parse and stricter than `routes_forms._clean_values`' "@ in
#: it": this list decides who MAY SUBMIT, so a typo that silently locks a colleague out is the
#: expensive failure, not an odd address that gets in.
_FORM_EMAIL = re.compile(r'^[^@\s,;]+@[^@\s,;]+\.[^@\s,;]+$')


def _clean_form(raw, valid_keys):
    """One form spec, fail-closed. Returns None when nothing is configured.



    ⚠ FIELD ORDER IS THE FORM'S OWN and is preserved here, not re-derived from the schema: the

    builder let somebody arrange these questions, and sorting them by column order would silently

    rearrange a live form every time a column was added (`_public_form` states the same rule from

    the serving end).



    ⚠ PARTIAL-DROP, not whole-key drop, and the asymmetry against `swipe` above is deliberate. A

    swipe binding is one three-part machine: two of its parts is not a degraded deck, it is a deck

    that can never write. A form is a LIST of questions — losing the column behind question three

    costs the asker question three, and taking the whole form away because one field was deleted

    would be a far larger loss than the one that happened.

    """
    if not isinstance(raw, dict):
        return None
    fields, seen = [], set()
    for k in (raw.get('fields') or [])[:MAX_FORM_FIELDS]:
        if k in valid_keys and k not in seen:
            seen.add(k)
            fields.append(k)
    out = {}
    if fields:
        out['fields'] = fields
        # A required flag on a question the form no longer asks is not a rule, it is a trap: the
        # submitter can never satisfy it and the sentence names a field they cannot see.
        req = [k for k in dict.fromkeys(raw.get('required') or []) if k in seen]
        if req:
            out['required'] = req
    for key, cap in (('title', MAX_FORM_TITLE), ('desc', MAX_FORM_DESC),
                     ('submitLabel', MAX_FORM_SUBMIT)):
        text = str(raw.get(key) or '').strip()[:cap]
        if text:
            out[key] = text
    # `public` is the ABSENT default (the `kanbanClamp` law: one way to say one thing), so only
    # the restrictive value is ever stored. ⇒ a spec that loses its `access` key fails OPEN, which
    # is why `emails` is what gets written rather than a `public: false`.
    if raw.get('access') == 'emails':
        out['access'] = 'emails'
    emails = []
    for e in (raw.get('emails') or [])[:MAX_FORM_EMAILS]:
        e = str(e or '').strip().lower()[:MAX_FORM_EMAIL]
        if _FORM_EMAIL.match(e) and e not in emails:
            emails.append(e)
    # Kept even while `access` is public: a person toggling the door open to test it and back
    # again must not lose the list of people they typed. It is never served publicly.
    if emails:
        out['emails'] = emails
    return out or None


#: C3 — the swipe binding's option cap. `leftOption`/`rightOption` are stored VALUES of a
#: single-select, and `_clean_options` trims every choice to 120 chars, so this is that same
#: number rather than a second opinion about it: a longer string cannot name a real option, and
#: a SHORTER cap here would silently refuse a binding to a legal one.
MAX_SWIPE_OPTION = 120

#: C-DISP (wave 2026-08-02): the time-series view's bucket vocabulary and caps, plus the
#: calendar-summary metric cap. types.ts mirrors these as TS_BUCKETS / TS_MAX_LAST_N /
#: TS_MAX_FIELDS / MAX_CALENDAR_METRICS, and cleanDisplay applies the same per-entry drops,
#: so an accepted save reads back byte-identically on both engines.
TS_BUCKETS = {'week', 'month', 'quarter', 'year'}
TS_MAX_LAST_N = 120
TS_MAX_FIELDS = 12
MAX_CALENDAR_METRICS = 4
_ISO_DAY = re.compile(r'^\d{4}-\d{2}-\d{2}$')

#: C6-CATALOG (wave 18) — the catalog view's vocabulary and caps. types.ts mirrors every name
#: below, and `cleanDisplay` applies the same drops in the SAME ORDER, so an accepted save reads
#: back identically on both engines. The code budget is the order-sensitive one — see
#: `_clean_catalogs`.
MAX_CATALOGS = 12
MAX_CATALOG_PAGES = 40
MAX_CATALOG_CODES = 500          # cumulative across ONE catalog's pages, spent in PAGE ORDER
CATALOG_PAPERS = {'letter', 'a4', 'tabloid'}
CATALOG_ORIENTATIONS = {'portrait', 'landscape'}
CATALOG_QUALITIES = {'web', 'print'}
CATALOG_PAGE_KINDS = {'cover', 'intro', 'section', 'gallery'}
CATALOG_COLS = (2, 3, 4)
CATALOG_ID_MAX, CATALOG_NAME_MAX = 40, 80
CATALOG_TITLE_MAX, CATALOG_BODY_MAX, CATALOG_CODE_MAX = 120, 2000, 60
_HEX6 = re.compile(r'^#[0-9A-Fa-f]{6}$')

#: Wave 14 C-ACC ([[loopable-wave14-split]]; rulings R2/R3). Mirrored by types.ts
#: TS_DELTA_KINDS / TS_MAX_CUSTOM_ROWS / TS_MAX_STYLES — the C-DISP byte-identical law.
TS_DELTA_KINDS = ('abs', 'pct', 'yoy', 'ytd')
TS_MAX_CUSTOM_ROWS = 12
TS_MAX_STYLES = 200
#: R2 — a formula row's `expr` is stored VERBATIM and NEVER parsed here (evaluation is client
#: law; the client refuses unknown refs/cycles/div-zero itself). The charset wall is the whole
#: server-side contract: row refs `[...]`, arithmetic, numbers — no markup, no control chars.
_TS_EXPR_OK = re.compile(r'^[A-Za-z0-9_ .+\-*/()\[\]]+$')

#: Old wire value -> the value we store today. Applied AFTER the membership test so an unknown
#: mode is still rejected rather than accidentally aliased.
_LEGACY_MODES = {'dashboard': 'chart'}

#: Chart kinds a chart-mode view may hold (wave-8 I19, contract C2). Mirrors the client's
#: union. Deliberately small: the owner asked to "start with simple charts" and expand, and a
#: kind the client cannot draw is worse than one that does not exist yet.
#: Wave-16 C-CHARTCAP: + 'table' — the group-by aggregate table (by-rep / by-BU /
#: top-customers, the third Sales block shape). Client renderer: DashboardView's
#: GroupTableView over salesParity.tableFromSpec.
CHART_KINDS = {'bar', 'line', 'area', 'donut', 'kpi', 'table'}
CHART_AGGS = {'sum', 'avg', 'count', 'min', 'max'}

#: ⭐ WAVE-29 C7 (item 17) — THE COLUMN-SUMMARY vocabulary: what a FIELD's `agg` may be, which is
#: what the grid's totals row and its per-group subtotals compute. ORDERED, because the order is
#: the picker's order; membership tests read it as a tuple perfectly well.
#:
#: ⛔ IT IS NOT `CHART_AGGS` AND THE TWO MUST NOT BE MERGED, however alike they look. `CHART_AGGS`
#: gatekeeps a STORED value with live data behind it (`charts[].agg`, `calendarMetrics[].agg`):
#: `_clean_chart` falls back to 'sum' on an unknown agg and `_clean_display` DROPS a whole
#: calendarMetrics entry, so renaming its 'avg' would silently turn every saved chart into a sum
#: and delete calendar cards, with nothing red. A chart's aggregation and a column's summary are
#: also different questions — one reduces a SERIES, the other a COLUMN — and one list serving both
#: would have to be the intersection of what each can express.
#:
#: ⭐ `average`, NOT `avg`, and the tie is broken by the vocabulary we cannot rename: `ROLLUP_FNS`
#: (`core/user_tables.py`, 16 names, 47 rollups live in production) already spells it `average`,
#: and it is the aggregate vocabulary a user actually reads today. Spelling it `avg` here would
#: give the product two words for one operation on two menus a click apart.
#:
#: ⚠ `median` is net-new: it is in NEITHER `CHART_AGGS` nor `ROLLUP_FNS`, so a Median column
#: summary has no rollup equivalent and this list is NOT a subset of either of its neighbours.
#:
#: ⚠ `count` counts ROWS in the scope (the group, or every matched row) — not non-blank cells.
#: `ROLLUP_FNS` splits that hair three ways (count / counta / countall); a column summary does not,
#: and must not grow a second spelling of it.
#:
#: Client mirror: `customer-grid/iconShapes.ts` FIELD_AGGS — ONE client list, imported by
#: `aggregations.ts` and the field editor rather than re-declared, so the only boundary left to
#: police is this one. `verify_icons.py::agg_parity` reads BOTH FILES and compares them.
FIELD_AGGS = ('sum', 'average', 'median', 'min', 'max', 'count')
MAX_CHARTS = 12                 #: per view — a dashboard, not an unbounded render loop
MAX_CHART_TITLE = 60

#: Wave-9 I11 (contract C2) — chart customisation, host-validated.
#:
#: `palette` names a colour JOB, never a colour. A browser must not be able to post a raw hex:
#: the four names below map to the four jobs a palette can do (identity / magnitude / polarity)
#: and resolve to brand ramps client-side, so a tenant restyle cannot be defeated by a stored
#: literal. STATUS colours (good/warning/serious/critical) are deliberately NOT selectable —
#: they are reserved signal, and reusing them as "series 4" is how a chart starts lying.
CHART_PALETTES = {'brand', 'categorical', 'sequential', 'diverging'}
CHART_FORMATS = {'auto', 'number', 'currency', 'percent', 'compact'}
MAX_AXIS_LABEL = 40
#: `size` is the I10 drag. Width is in GRID COLUMNS (a 12-column board), height in px.
CHART_W_RANGE = (1, 12)
CHART_H_RANGE = (120, 800)


def _clean_chart(raw, valid_keys):
    """One dashboard chart, fail-closed. Returns None if the chart cannot be drawn.



    A chart's `y` is optional (absent = count of rows, which is what "how many customers per

    state" means). `x` is NOT: a chart with no category axis has nothing to plot against, and

    silently keeping it would put an empty card on the dashboard with no way to tell why.

    """
    if not isinstance(raw, dict):
        return None
    kind = raw.get('kind')
    if kind not in CHART_KINDS:
        return None
    x = raw.get('x')
    if x not in valid_keys:
        return None                     # dead category ref -> the chart goes, not the board
    cid = raw.get('id')
    if not isinstance(cid, str) or not cid.strip():
        return None                     # the client owns chart ids; an unidentified card
        # cannot be edited or removed, so it must not persist
    out = {'id': cid.strip()[:64], 'kind': kind, 'x': x,
           'agg': raw.get('agg') if raw.get('agg') in CHART_AGGS else 'sum'}
    if raw.get('y') in valid_keys:
        out['y'] = raw['y']
    else:
        # no measurable column -> the only honest aggregation left is "how many rows"
        out['agg'] = 'count'
    title = raw.get('title')
    if isinstance(title, str) and title.strip():
        out['title'] = title.strip()[:MAX_CHART_TITLE]

    # ── wave-9 I11 (contract C2): customisation ────────────────────────────────────────────
    # `splitBy` is the field whose values become the SERIES. It is deliberately not called
    # `colorBy`: that name already means two other things here (`config.colorBy` = row
    # colouring, `display.colorField` = map pin colour) and a third sense would be unreadable.
    if raw.get('splitBy') in valid_keys and raw['splitBy'] != x:
        out['splitBy'] = raw['splitBy']
    # Stacking is only a question once there are series to stack, and only for the two kinds
    # that can express it. Anywhere else it is dropped rather than stored as a lie the client
    # would have to re-decide.
    if out.get('splitBy') and kind in ('bar', 'area') and raw.get('stacked') is True:
        out['stacked'] = True
    if raw.get('palette') in CHART_PALETTES:
        out['palette'] = raw['palette']

    axis = raw.get('axis')
    if isinstance(axis, dict):
        # ⛔ ONE y-scale, always. There is no second-axis key here and there must never be:
        # two y-scales on one frame can manufacture any correlation you like by rescaling, and
        # the honest alternatives are two charts, small multiples, or indexing to a common base.
        # Ruled explicitly in contract C2 against the "Tableau versatility" brief.
        clean_axis = {}
        for side in ('x', 'y'):
            spec = axis.get(side)
            if not isinstance(spec, dict):
                continue
            one = {}
            lab = spec.get('label')
            if isinstance(lab, str) and lab.strip():
                one['label'] = lab.strip()[:MAX_AXIS_LABEL]
            if spec.get('format') in CHART_FORMATS:
                one['format'] = spec['format']
            if one:
                clean_axis[side] = one
        if clean_axis:
            out['axis'] = clean_axis

    size = raw.get('size')
    if isinstance(size, dict):
        one = {}
        for key, (lo, hi) in (('w', CHART_W_RANGE), ('h', CHART_H_RANGE)):
            try:
                one[key] = max(lo, min(hi, int(size[key])))
            except (KeyError, TypeError, ValueError):
                pass            # a partial size is fine: the client defaults the missing axis
        if one:
            out['size'] = one

    # ── Wave 14 R3 ([[loopable-wave14-split]]): a METRIC chart may carry a PERIOD — the
    # trend-over-buckets encoding. Kept only when the chart's value field is measure-backed:
    # a category column has no time dimension, and a stored period on it would promise a
    # trend the TS channel must refuse. `span` is meaningful only beside `bucket`.
    if isinstance(out.get('y'), str) and out['y'].startswith('measure_'):
        if raw.get('bucket') in TS_BUCKETS:
            out['bucket'] = raw['bucket']
            sp = raw.get('span')
            if isinstance(sp, dict):
                n = sp.get('lastN')
                if (isinstance(n, int) and not isinstance(n, bool)
                        and 1 <= n <= TS_MAX_LAST_N):
                    out['span'] = {'lastN': n}
        # ── Wave-16 C-CHARTCAP: the YoY companion. Kept ONLY where it can mean something —
        # beside a kept bucket (the compare series) or on a sum-of-metric KPI (the delta
        # line). Anything else is a stored claim the renderer would have to re-refuse.
        # Mirrors the client's cleanCharts rule key for key.
        if raw.get('compare') == 'prior_year' and (
                out.get('bucket') or (kind == 'kpi' and out.get('agg') == 'sum')):
            out['compare'] = 'prior_year'
    return out


def _clean_catalog_page(raw, budget):
    """C6-CATALOG — one page of a catalog. Returns `(page | None, codes_spent)`.



    `budget` is what is LEFT of the catalog's 500-code allowance. Codes are deduped WITHIN a

    page and not across the catalog: a product legitimately appears on a gallery page and again

    in its section listing, and de-duplicating globally would silently delete the second

    appearance. The budget is spent in page order, so a catalog that runs out loses the TAIL of

    its last pages — never a random scatter, and never a page (a page with no products is a

    heading the user can still see and fix).

    """
    if not isinstance(raw, dict):
        return None, 0
    page_id = str(raw.get('id') or '')[:CATALOG_ID_MAX]
    kind = raw.get('kind')
    if not page_id or kind not in CATALOG_PAGE_KINDS:
        return None, 0
    out = {'id': page_id, 'kind': kind}
    for key, cap in (('title', CATALOG_TITLE_MAX), ('body', CATALOG_BODY_MAX),
                     ('imageCode', CATALOG_CODE_MAX)):
        v = raw.get(key)
        if isinstance(v, str) and v:
            out[key] = v[:cap]
    products = raw.get('products')
    if isinstance(products, list) and budget > 0:
        clean_p, seen_p = [], set()
        for c in products:
            if not isinstance(c, str) or not c:
                continue
            c = c[:CATALOG_CODE_MAX]
            if c in seen_p:
                continue
            seen_p.add(c)
            clean_p.append(c)
            if len(clean_p) >= budget:
                break
        if clean_p:
            out['products'] = clean_p
    layout = raw.get('layout')
    if isinstance(layout, dict):
        clean_l = {}
        cols = layout.get('cols')
        if isinstance(cols, int) and not isinstance(cols, bool) and cols in CATALOG_COLS:
            clean_l['cols'] = cols
        # The kanbanClamp/tsSparkline asymmetry, one per direction: pack and colour SHOW by
        # default (the 2027 catalogue shows both), price does NOT (it shows no prices at all).
        # So only the opt-OUT is storable for the first two and only the opt-IN for the third —
        # a second spelling of a default is how a round trip starts churning.
        if layout.get('showPack') is False:
            clean_l['showPack'] = False
        if layout.get('showColor') is False:
            clean_l['showColor'] = False
        if layout.get('showPrice') is True:
            clean_l['showPrice'] = True
        if clean_l:
            out['layout'] = clean_l
    return out, len(out.get('products') or ())


def _clean_catalogs(raw, valid_keys):
    """C6-CATALOG (wave 18) — `display.catalogs`, fail-closed. Returns a list or None.



    A catalog is a PRINT artifact, so the two structural keys that decide how it paginates

    (`paper`, `orientation`) are NORMALISED WITH A DEFAULT rather than dropped: a page box with

    no size is not a smaller catalog, it is an unrenderable one. Everything else follows the

    house rules — unknown keys dropped, per-entry drops never cost the neighbours, empty

    sub-objects omitted entirely (`brand`, `fields`, `layout`, `products`) so an absent key and

    an empty one are not two spellings of the same nothing.



    `fields` binds the listing lines to real columns (the 2027 listing prints description / SKU /

    pack / colour, and `product_data` carries no pack or colour of its own — the user binds

    custom fields). Refs are checked against `valid_keys` HERE and not on the client, the same

    split `dateField`/`stackField` already run.

    """
    if not isinstance(raw, list):
        return None
    out = []
    for c in raw:
        if len(out) >= MAX_CATALOGS:
            break
        if not isinstance(c, dict):
            continue
        cat_id = str(c.get('id') or '')[:CATALOG_ID_MAX]
        name = c.get('name')
        # An EMPTY name is legal (the user cleared the box and will type again) — an ABSENT one
        # is a malformed record. The `tsRows` label rule, same reasoning.
        if not cat_id or not isinstance(name, str):
            continue
        cat = {'id': cat_id, 'name': name[:CATALOG_NAME_MAX]}
        cat['paper'] = c['paper'] if c.get('paper') in CATALOG_PAPERS else 'letter'
        cat['orientation'] = (c['orientation']
                              if c.get('orientation') in CATALOG_ORIENTATIONS else 'portrait')
        if c.get('quality') in CATALOG_QUALITIES:
            cat['quality'] = c['quality']
        brand = c.get('brand')
        if isinstance(brand, dict):
            clean_b = {}
            for k in ('primary', 'accent'):
                v = brand.get(k)
                if isinstance(v, str) and _HEX6.match(v):
                    clean_b[k] = v
            company = brand.get('company')
            if isinstance(company, str) and company:
                clean_b['company'] = company[:CATALOG_NAME_MAX]
            # An asset CODE (resolved through C2-ASSET), never a URL: an arbitrary host inside
            # print CSS is exactly the tokens-not-values rule this contract carries.
            logo = brand.get('logo')
            if isinstance(logo, str) and logo:
                clean_b['logo'] = logo[:CATALOG_CODE_MAX]
            if clean_b:
                cat['brand'] = clean_b
        binds = c.get('fields')
        if isinstance(binds, dict):
            clean_bind = {k: binds[k] for k in ('name', 'pack', 'color', 'price')
                          if binds.get(k) in valid_keys}
            if clean_bind:
                cat['fields'] = clean_bind
        pages, budget = [], MAX_CATALOG_CODES
        raw_pages = c.get('pages')
        if isinstance(raw_pages, list):
            for p in raw_pages:
                if len(pages) >= MAX_CATALOG_PAGES:
                    break
                page, spent = _clean_catalog_page(p, budget)
                if page is None:
                    continue
                budget -= spent
                pages.append(page)
        # ALWAYS emitted, even empty: a catalog with no pages yet is the state every catalog
        # starts in, and dropping the key would make "new" and "corrupt" the same wire value.
        cat['pages'] = pages
        out.append(cat)
    return out or None


def _clean_display(raw, valid_keys):
    """Structural passthrough for a view's `config.display` (wave-6 item 10), fail-closed.



    `{mode, dateField?, stackField?, titleField?, colorField?, sizeField?, charts?}` — mode

    must be a known non-grid mode (grid is the absent default, so storing it would be a second

    way to say nothing); every field ref must name a field this table has (a ref to a deleted

    field is DROPPED and the client falls back to its per-mode default); unknown keys are

    dropped. What each mode MEANS — calendar wants a date-family field, kanban a select-family

    stack, map a single-select to colour by and a numeric to size by — is the client's

    business: it is the only layer that renders them, and a wrong-typed ref degrades to that

    surface's default rather than to an error (the `_clean_window` split).



    Wave-8 (contract C2) adds the map encodings (`colorField` I3, `sizeField` I5) and

    dashboard `charts` (I19). A chart whose x/y names a deleted field is dropped INDIVIDUALLY —

    never the whole array, because losing one column should not cost the user a dashboard they

    spent time building.

    """
    if not isinstance(raw, dict):
        return None
    mode = raw.get('mode')
    if mode not in DISPLAY_MODES or mode == 'grid':
        return None
    # Wave-9 I10 (C2): normalise the legacy wire value AFTER the membership test, so an unknown
    # mode is still rejected rather than accidentally aliased into a real one. Every already
    # saved 'dashboard' view reads back as 'chart' from here on; nothing writes 'dashboard'.
    mode = _LEGACY_MODES.get(mode, mode)
    out = {'mode': mode}
    for ref in ('dateField', 'stackField', 'titleField', 'colorField', 'sizeField'):
        if raw.get(ref) in valid_keys:
            out[ref] = raw[ref]
    # ── C-DISP (wave 2026-08-02) ─────────────────────────────────────────────────────────
    # kanbanClamp: stored ONLY as the literal opt-OUT. Absent means clamped — the new
    # standardized default — so storing True would be a second way to say nothing (the same
    # rule that keeps mode:'grid' out of the store).
    if raw.get('kanbanClamp') is False:
        out['kanbanClamp'] = False
    if raw.get('calendarMode') in ('records', 'summary'):
        out['calendarMode'] = raw['calendarMode']
    metrics = raw.get('calendarMetrics')
    if isinstance(metrics, list):
        clean_m, seen_m = [], set()
        for m in metrics[:MAX_CALENDAR_METRICS]:
            # Dropped INDIVIDUALLY (the charts precedent): one dead metric must not cost the
            # user the summary card they configured around it.
            if not isinstance(m, dict):
                continue
            mid = str(m.get('id') or '')[:40]
            if (not mid or mid in seen_m or m.get('field') not in valid_keys
                    or m.get('agg') not in CHART_AGGS):
                continue
            seen_m.add(mid)
            clean_m.append({'id': mid, 'field': m['field'], 'agg': m['agg']})
        if clean_m:
            out['calendarMetrics'] = clean_m
    if raw.get('tsBucket') in TS_BUCKETS:
        out['tsBucket'] = raw['tsBucket']
    span = raw.get('tsSpan')
    if isinstance(span, dict):
        clean_span = {}
        n = span.get('lastN')
        if isinstance(n, int) and not isinstance(n, bool) and 1 <= n <= TS_MAX_LAST_N:
            clean_span['lastN'] = n
        else:
            f, t = span.get('from'), span.get('to')
            f = f if isinstance(f, str) and _ISO_DAY.match(f) else None
            t = t if isinstance(t, str) and _ISO_DAY.match(t) else None
            if f and t and f > t:
                f, t = t, f
            if f:
                clean_span['from'] = f
            if t:
                clean_span['to'] = t
        if clean_span:
            out['tsSpan'] = clean_span
    ts_fields = raw.get('tsFields')
    if isinstance(ts_fields, list):
        clean_f, seen_f = [], set()
        for k in ts_fields[:TS_MAX_FIELDS]:
            if k in valid_keys and k not in seen_f:
                seen_f.add(k)
                clean_f.append(k)
        if clean_f:
            out['tsFields'] = clean_f
    # ── Wave 14 C-ACC ([[loopable-wave14-split]] R2; items 17/18) ────────────────────────
    deltas = raw.get('tsDeltas')
    if isinstance(deltas, list):
        clean_d, seen_d = [], set()
        for d in deltas:
            if d in TS_DELTA_KINDS and d not in seen_d:
                seen_d.add(d)
                clean_d.append(d)
        if clean_d:
            out['tsDeltas'] = clean_d
    # Gridlines: stored ONLY as the literal opt-OUT (absent = shown), sparkline ONLY as the
    # literal opt-IN (absent = off) — the kanbanClamp asymmetry, one per direction.
    if raw.get('tsGridlines') is False:
        out['tsGridlines'] = False
    if raw.get('tsSparkline') is True:
        out['tsSparkline'] = True
    rows = raw.get('tsRows')
    if isinstance(rows, list):
        clean_r, seen_r = [], set()
        for r in rows[:TS_MAX_CUSTOM_ROWS]:
            if not isinstance(r, dict):
                continue
            rid = str(r.get('id') or '')[:40]
            r_kind = r.get('kind')
            if not rid or rid in seen_r or r_kind not in ('note', 'formula'):
                continue
            label = r.get('label')
            if not isinstance(label, str):
                continue    # ABSENT label = malformed; an EMPTY one is a legal spacer row
                # (GRID's dated asymmetry amendments, 2026-08-02)
            one = {'id': rid, 'kind': r_kind, 'label': label.strip()[:120]}
            if r_kind == 'formula':
                expr = r.get('expr')
                if (isinstance(expr, str) and expr.strip()
                        and len(expr) <= 200 and _TS_EXPR_OK.match(expr)):
                    one['expr'] = expr
                # else: keep the ROW, drop the EXPR — it renders "—". Vanishing the row
                # would delete the user's label to punish their arithmetic (GRID's dated
                # asymmetry amendment; the calendarMetrics per-entry-drop precedent).
            seen_r.add(rid)
            clean_r.append(one)
        if clean_r:
            out['tsRows'] = clean_r
    styles = raw.get('tsStyles')
    if isinstance(styles, dict):
        clean_s = {}
        for s_key, s_val in styles.items():
            if len(clean_s) >= TS_MAX_STYLES:
                break               # capped, not truncated silently: the gate names this
            if not isinstance(s_key, str) or not s_key or len(s_key) > 96:
                continue            # key = rowId or "rowId:colKey" — the client's grammar
            if not isinstance(s_val, dict):
                continue
            one = {}
            if s_val.get('bold') is True:
                one['bold'] = True
            if s_val.get('line') is True:
                one['line'] = True
            if one:
                clean_s[s_key] = one
        if clean_s:
            out['tsStyles'] = clean_s
    charts = raw.get('charts')
    if isinstance(charts, list):
        clean = [c for c in (_clean_chart(x, valid_keys) for x in charts[:MAX_CHARTS]) if c]
        # de-dupe by id: two cards sharing an id are one card as far as the client's keyed
        # render is concerned, and the second would silently shadow the first
        seen, uniq = set(), []
        for c in clean:
            if c['id'] in seen:
                continue
            seen.add(c['id'])
            uniq.append(c)
        if uniq:
            out['charts'] = uniq
    # ── C6-CATALOG (wave 18) ─────────────────────────────────────────────────────────────
    catalogs = _clean_catalogs(raw.get('catalogs'), valid_keys)
    if catalogs:
        out['catalogs'] = catalogs
    # ── ⭐ WAVE-27 C3 (item 8 / R2): the swipe binding ────────────────────────────────────
    # `{fieldKey, leftOption, rightOption}` — WHOLE-KEY drop, never a partial one, and that
    # asymmetry against `charts`/`calendarMetrics` above is the point rather than an oversight.
    # Those are LISTS of independent cards, so losing one entry costs the user one card. This is
    # a single three-part BINDING: a swipe view holding a fieldKey with one option, or two
    # options and no field, is not a degraded swipe view — it is a deck that can never write
    # anything, rendered as though it were configured. Dropping the key entirely puts the view
    # back in its honest unconfigured state, which is the one state the client has a UI for.
    #
    # ⚠ What this CANNOT check, deliberately, and why the client must: whether `fieldKey` names
    # a SELECT, and whether the two options are still in that select's vocabulary. `valid_keys`
    # is a key set, and the docstring above draws this exact line — "what each mode MEANS ... is
    # the client's business". So SwipeView owns three losses this function is blind to (field
    # deleted, field retyped away from select, option removed) and must SHOW each one rather
    # than fall back to the first option, per the `viewModes.tsx` house rule.
    swipe = raw.get('swipe')
    if isinstance(swipe, dict):
        f_key = swipe.get('fieldKey')
        left, right = swipe.get('leftOption'), swipe.get('rightOption')
        ok = (f_key in valid_keys
              and isinstance(left, str) and isinstance(right, str))
        if ok:
            left, right = left.strip()[:MAX_SWIPE_OPTION], right.strip()[:MAX_SWIPE_OPTION]
            # Both non-empty, and DISTINCT: one option on both sides is a deck whose two
            # gestures do the same thing, which is two spellings of one state (the
            # `kanbanClamp` law) wearing a control that promises a choice.
            if left and right and left.casefold() != right.casefold():
                out['swipe'] = {'fieldKey': f_key, 'leftOption': left, 'rightOption': right}
    # ── ⭐⭐ WAVE-29 R7 (item 10): the FORM spec — see `_clean_form` for why the token is not here.
    form = _clean_form(raw.get('form'), valid_keys)
    if form:
        out['form'] = form
    return out


#: FOLDERS over the saved views / cohorts sidebars (wave-8 I11, contract C4).
#:
#: ⚠ Folder membership is stored as a SIDE MAP (`itemFolders`), not as a `folderId` ON each
#: view or cohort — a deliberate amendment to C4's first wording, recorded in the split doc.
#: Two reasons. (1) A cohort lives in a DIFFERENT store (`customer_cohorts`, keyed by cohort
#: id) and adding a `folders` key beside those ids would collide with a cohort whose generated
#: id happened to be 'folders'. (2) Folder placement is a per-user ORGANISING act, not part of
#: what a view IS: keeping it out of the view config means duplicating or exporting a view does
#: not drag a folder reference along with it. One map, one home, both surfaces.
FOLDER_SURFACES = {"views", "cohorts"}
MAX_FOLDERS = 60
MAX_FOLDER_NAME = 80

#: Wave-9 I15 (contract C5) — a user-chosen folder icon, as {shape, tone}.
#:
#: Both halves are WHITELISTS, never free values: `shape` names geometry the client already
#: draws (one source, `iconShapes.ts`, read by both painters) and `tone` names a palette token,
#: not a colour — so a browser cannot post a hex and defeat a tenant restyle, and a shape the
#: client cannot render can never reach the store.
#: ⚠ MIRRORS CLIENT'S `iconShapes.ts` ENUMERATION EXACTLY (C5: CLIENT enumerates, HOST mirrors —
#: posted in the split doc 2026-07-29, HOST adopted it the same day, replacing a provisional
#: 12-shape guess of mine that contained shapes the client cannot draw). Do not extend this set
#: without the matching client geometry: an unknown shape falls back to the default folder mark,
#: which is also I14's "existing folders get the folder icon" for every pre-wave-9 folder.
FOLDER_ICON_SHAPES = {"folder", "star", "flag", "tag", "bookmark", "box", "circle", "square"}
#: Tones are the C1 pastels — FILLS ONLY, never text (the standing palette rule). 'grey' is the
#: default, and is CLIENT's key name: not 'neutral', which is what HOST first guessed.
FOLDER_ICON_TONES = {"blue", "green", "yellow", "red", "grey"}
FOLDER_ICON_DEFAULT_TONE = "grey"


#: Wave-9 I17 (contract C4) — who may EDIT a saved view.
#:
#: ⚠ READ THIS BEFORE BUILDING ON IT. Views are stored PER USER today
#: (`core/table_store.TableStore.workspace` reads `store.get(table_key)[username]`), so one
#: user's views are invisible to every other user and "collaborative" has nothing to act on
#: yet. This validator is therefore CORRECT-BUT-INERT plumbing: it makes the setting durable
#: and fail-closed now, so that when shared views land the permission does not need a data
#: migration and no saved view is retro-restricted. It does NOT make anything shared, and
#: nothing in the app currently reads it to grant or deny cross-user access.
#: Recorded as the C4 amendment in .claude/wiki/research/grid-wave9-split.md.
VIEW_EDIT_MODES = {'personal', 'collaborative', 'users'}
MAX_VIEW_USERS = 50


def clean_view_permissions(raw, default, known_users=None):
    """{edit, users?} — fail-closed on both halves.



    `default` is supplied by the CALLER because it splits by path, and that split is a

    permission rule rather than a formatting one: absent on a view that already exists means a

    pre-wave-9 view and must stay 'collaborative' (retro-restricting somebody's saved view is a

    silent takeaway), while absent on CREATE must be 'personal' (a new view must never be

    anyone-can-edit purely by omission).



    `known_users` (when given) is the real account list: an unknown name is DROPPED, and an

    'users' grant left with nobody in it collapses to 'personal' rather than to everyone.

    """
    mode = (raw or {}).get('edit') if isinstance(raw, dict) else None
    if mode not in VIEW_EDIT_MODES:
        mode = default if default in VIEW_EDIT_MODES else 'personal'
    if mode != 'users':
        return {'edit': mode}
    names, seen = [], set()
    for u in list((raw or {}).get('users') or [])[:MAX_VIEW_USERS]:
        u = str(u or '').strip()
        if not u or u.lower() in seen:
            continue
        if known_users is not None and u not in known_users:
            continue                      # fail-closed: a name we cannot resolve grants nothing
        seen.add(u.lower())
        names.append(u)
    if not names:
        return {'edit': 'personal'}       # an empty grant is NOT "everyone"
    return {'edit': 'users', 'users': names}


def clean_folder_icon(raw):
    """{shape, tone} or None. Fail-closed on both halves, independently.



    A folder with a valid shape but a junk tone keeps the shape and defaults the tone rather

    than losing the icon entirely — losing a user's pick because one half was wrong is the kind

    of silent data loss the folder events already avoid elsewhere.

    """
    if not isinstance(raw, dict):
        return None
    shape = raw.get("shape")
    if shape not in FOLDER_ICON_SHAPES:
        return None
    tone = raw.get("tone")
    return {"shape": shape,
            "tone": tone if tone in FOLDER_ICON_TONES else FOLDER_ICON_DEFAULT_TONE}


def clean_folders(raw):
    """Validate the per-surface folder lists, fail-closed. {surface: [{id, name, order}]}."""
    out = {}
    for surface in FOLDER_SURFACES:
        items, seen = [], set()
        for f in list((raw or {}).get(surface) or [])[:MAX_FOLDERS]:
            if not isinstance(f, dict):
                continue
            fid = str(f.get("id") or "").strip()[:80]
            name = str(f.get("name") or "").strip()[:MAX_FOLDER_NAME]
            if not fid or not name or fid in seen:
                continue          # an unidentified or unnamed folder cannot be shown or edited
            seen.add(fid)
            try:
                order = int(f.get("order", len(items)))
            except (TypeError, ValueError):
                order = len(items)
            row = {"id": fid, "name": name, "order": order}
            icon = clean_folder_icon(f.get("icon"))      # wave-9 I15 (C5); absent = default mark
            if icon:
                row["icon"] = icon
            items.append(row)
        items.sort(key=lambda x: x["order"])
        for i, f in enumerate(items):
            f["order"] = i                     # re-index so `order` is always dense and total
        if items:
            out[surface] = items
    return out


#: ⭐⭐ WAVE 32 · OWNER ITEM 20 (`W32-T27`, raised by SESSION C as ASK C-16) — "FILED AT ROOT".
#:
#: ⛔ THE DEFECT IS THAT ROOT WAS REPRESENTED BY *ABSENCE*, AND ABSENCE CANNOT HOLD TWO FACTS.
#: "this arrived by grant and was never filed" and "the receiver deliberately dragged this OUT of
#: the Shared group" were the same stored state — nothing — so the client had to GUESS, and
#: `folders.ts::groupByFolder` guessed "Shared". That is why only folder→folder moves appeared to
#: work: **the root bucket was unreachable for a shared view by construction.**
#:
#: ⚠ A RESERVED FOLDER ID, NOT A NEW FIELD, deliberately. The placement map is `{itemId: folderId}`
#: and every reader on both sides already understands it; a parallel "filedAtRoot" set would be a
#: second source of truth for one question, and the two would disagree the first time one of them
#: was written without the other. This id names no folder BY DESIGN and is therefore exempt from
#: the folder-exists test below — it is the one value that means "no folder, on purpose".
#: ⚠ Spelled `ROOT_FOLDER_ID` on the client (`customer-grid/folders.ts`, C's file). Two spellings
#: of one constant is [[a-constant-two-features-share]]; `verify_folders`/`verify_api` assert they
#: agree rather than a comment asking nicely.
ROOT_PLACEMENT = "__root__"


def clean_item_folders(raw, folders, valid_ids):
    """{surface: {itemId: folderId}} — dropping any placement whose ITEM or FOLDER is gone.



    This is what makes a deleted folder's contents fall back to the root rather than vanish:

    nothing stores "this item is in no folder", so an unresolvable placement simply disappears

    and the item renders at the top level. Same for an item that was deleted elsewhere — its

    stale placement can never resurrect it, because the sidebars render ITEMS and consult this

    map, never the other way round.



    ⭐⭐ WAVE 32 — THE PARAGRAPH ABOVE STATES THE FEATURE AND THE BUG IN ONE SENTENCE, and it took

    owner item 20 to notice they were the same mechanism. *"Nothing stores 'this item is in no

    folder', so an unresolvable placement simply disappears"* is exactly right for a DELETED FOLDER

    (its contents should fall to the root) and exactly wrong for a SHARED VIEW (falling back means

    falling back INTO the Shared group, which is where it started). `ROOT_PLACEMENT` is the value

    that survives this function so the second case can be said out loud.

    """
    out = {}
    for surface in FOLDER_SURFACES:
        fids = {f["id"] for f in (folders or {}).get(surface, [])}
        ok = {}
        for item_id, fid in ((raw or {}).get(surface) or {}).items():
            if not isinstance(item_id, str) or not isinstance(fid, str):
                continue
            # ⛔ `fid == ROOT_PLACEMENT` FIRST, and it is NOT in `fids` — it names no folder, which
            # is the whole point. Without this clause the value is written by `item_move` and
            # scrubbed here on the way back out, so the mark would be stored and instantly lost:
            # the two halves are ONE change and shipping either alone is worse than shipping
            # neither ([[lost-write-looks-like-failed-read]]).
            if item_id in (valid_ids or {}).get(surface, ()) and (fid == ROOT_PLACEMENT
                                                                  or fid in fids):
                ok[item_id[:120]] = fid
        if ok:
            out[surface] = ok
    return out


def clean_filter_tree(raw, valid_keys, depth=1, budget=None, cohort_ids=None):
    """Recursively validate an UNTRUSTED filter tree (conditions + nested groups).



    Returns a clean tree of leaf conditions ({colId, op, value, value2}) and

    groups ({conj, children}). Module-agnostic on purpose: any module embedding

    the grid validates its own view state through this one function.



    Fail-closed PER NODE: anything unrecognised is DROPPED rather than raised —

    the same contract the rest of the view sanitiser follows, so one bad rule can

    never cost a user their whole saved view. Depth, per-level width and total

    node count are all capped: the tree is re-evaluated for every row on every

    render, so an unbounded structure would be a persistent client-side DoS.

    Empty groups are dropped (they carry no meaning once persisted).



    `cohort_ids` is the set of cohorts the CALLER may see. A cohort leaf naming anything else is

    DROPPED here rather than left for the engine — a deleted cohort would otherwise leave a

    condition that can only match nothing, so `List is not [deleted]` would show an empty table

    forever with no way to tell why. `None` means this host has no cohorts, and then every

    cohort leaf is dropped: fail-closed, like every other unknown key.

    """
    if budget is None:
        budget = [MAX_FILTER_NODES]
    out = []
    for node in list(raw or [])[:MAX_FILTER_SIBLINGS]:
        if budget[0] <= 0:
            break
        if not isinstance(node, dict):
            continue
        if isinstance(node.get('children'), list):            # a condition GROUP
            if depth >= MAX_FILTER_DEPTH:
                continue                                       # too deep -> drop
            budget[0] -= 1
            children = clean_filter_tree(node['children'], valid_keys,
                                        depth + 1, budget, cohort_ids)
            if children:
                out.append({'conj': 'or' if node.get('conj') == 'or' else 'and',
                            'children': children})
            continue
        if node.get('colId') == COHORT_FIELD:                 # a cohort-membership leaf
            op = COHORT_OP_ALIASES.get(node.get('op'), node.get('op'))
            named = parse_cohort_ids(node.get('value'))
            # ALL of them, or the leaf goes. A set that quietly lost a member asks a DIFFERENT
            # question, and for `noneOf` a strictly wider one: `is none of [A, B]` degrading to
            # `is none of [A]` would show every row in B under a count nobody would doubt. This
            # is the same all-or-nothing the single-cohort leaf already had, extended to a set.
            if op in COHORT_OPS and named and all(c in (cohort_ids or ()) for c in named):
                budget[0] -= 1
                out.append({'colId': COHORT_FIELD, 'op': op,
                            'value': ','.join(named), 'value2': ''})
            continue
        if node.get('colId') in valid_keys and node.get('op') in FILTER_OPS:
            budget[0] -= 1
            # `or ''` would be wrong here: it maps every FALSY value to '', and '' is the
            # signal for "inactive". A numeric 0 (or 0.0, or False) is a real value the client
            # treats as active — `0 === ""` is false in TS — so `revenue = 0` would silently
            # stop filtering and show every row instead of the zero-revenue ones.
            val, val2 = node.get('value'), node.get('value2')
            leaf = {'colId': node['colId'], 'op': node['op'],
                    'value': ('' if val is None else str(val))[:500],
                    'value2': ('' if val2 is None else str(val2))[:500]}
            # CG-8. A MEASURE condition ("Sales, in the last 90 days, > 5,000") carries two
            # extra members: a stable client-generated `id`, which is how the server's answer
            # finds its way back to the condition that asked (positional matching silently
            # re-associates every answer the moment a user deletes a condition), and the
            # `window`. Emitted ONLY when the input has them — a column condition's cleaned
            # shape is unchanged, so every persisted view deserialises byte-identically and
            # `clean_filter_tree` stays idempotent (verify_filter_engine.py asserts that by
            # exact structural comparison).
            rid = node.get('id')
            if rid not in (None, ''):
                leaf['id'] = str(rid)[:64]
            window = _clean_window(node.get('window'))
            if window is not None:
                leaf['window'] = window
            # Owner items 3 + 4, carried under the SAME rule as CG-8's `id`/`window`: emitted
            # only when the input has them, so a plain column condition's cleaned shape is
            # byte-identical to what it always was and `clean_filter_tree` stays idempotent
            # (verify_filter_engine.py asserts that by exact structural comparison). Drop the
            # carry-through and the next autosave silently strips a date condition back to a
            # bare comparison against an empty value — i.e. back to INACTIVE.
            date_window = _clean_window(node.get('dateWindow'))
            if date_window is not None:
                leaf['dateWindow'] = date_window
            mode = node.get('dateMode')
            # SHAPE only. An unrecognised mode survives here and is refused by
            # `windows.resolve_anchor`, which returns None and makes the condition match
            # NOTHING — the same split as `_clean_window`, and the reason this module can stay
            # free of the date vocabulary it would otherwise have to keep in step.
            if isinstance(mode, str) and 0 < len(mode) <= 40:
                leaf['dateMode'] = mode
            rhs = _clean_rhs(node.get('rhs'), valid_keys)
            if rhs is not None:
                leaf['rhs'] = rhs
            out.append(leaf)
    return out


def _default_view_config(fields):
    # ⭐⭐ W30-T41's SERVER HALF (F's ask F-1, answered by D — this file is D's fence).
    #
    # ⛔ THE SECOND ARM USED TO BE `or field["source"] == "overlay"`, AND IT SWALLOWED THE FIRST
    # ONE FOR EVERY CONNECTED COLUMN. `user_tables._clean_field` stamps `source: "overlay"` on
    # every `ut_` field, so on an Odoo grid the arm was true for ALL of them and `default: False`
    # meant nothing: `odoo_id`, `state`, `customer_link` and `partner_id` opened SHOWN however
    # they were declared. The exception had become the rule ([[fallback-that-became-the-rule]]),
    # and it is the same predicate `useGridColumns.isDefaultVisible` carried on the client.
    #
    # ⚠ AND THE TWO HALVES MUST MOVE TOGETHER, which is why this is not cosmetic. `CustomerGrid`
    # compares the stored view against its own `defaultViewConfig` by JSON equality; with the
    # client fixed (T41) and this left alone, the system view would differ from the client's
    # default on every render — a view that looks permanently dirty and autosaves forever, which
    # is the failure `verify_filter_engine`'s key-ORDER check exists to prevent, one level down.
    #
    # ⚠ A USER-CREATED COLUMN IS UNAFFECTED, and that is why the fix is a DELETION rather than a
    # carve-out for the four Odoo keys: it carries no `default` key at all, so `is not False`
    # keeps it visible. On the main Customer grid exactly one field moves — `notes`, which asks
    # to be hidden in its own declaration and was being shown against it.
    shown = [field["key"] for field in fields if field.get("default") is not False]
    hidden = [field["key"] for field in fields if field["key"] not in shown]
    return {
        # `filters` is the ROOT of the filter tree: leaf conditions and/or nested
        # condition groups ({conj, children}); `filterConj` joins the root level.
        "filters": [], "filterConj": "and",
        "sorts": [], "groupBy": None, "colorBy": None,
        "rowHeightMode": "short", "order": shown + hidden, "visible": shown,
        "widths": {}, "memberPids": [],
    }


#: Wave 17 R1 / C-LOCKV — the `kind` a PROJECTED locked view wears. A cohort is not a separate
#: kind of object any more: it is a saved view whose rows are a hand-curated set.
LOCKED_VIEW_KIND = 'locked'


def locked_view_projection(entry, base_config):
    """One cohort -> the saved-view row that IS it (wave 17 R1, contract C-LOCKV).



    ⛔ THE LOCK IS THE VIEW'S IDENTITY, NOT ITS CONFIGURATION. `config.cohortLock` names the

    view's OWN id, which is what makes the shipped engine law (`useVisibleRows`: intersect the

    named set FIRST, unconditionally, and match NOTHING when the membership is unresolvable) do

    all the work with no second mechanism. Membership itself is NEVER copied in here — it stays

    in `customer_cohorts` and travels as `workspace.lists`, because a per-reader-scoped

    collection inside a client-writable `config` is deleted by the next autosave (see the

    contract's reason 2).



    ⚠ `locked: True` is the LEGACY "undeletable/mode-frozen" flag and is deliberately NOT set:

    these views are ordinary in every respect the owner asked for — reorder, folder, sort,

    filter, change display mode. The lock mark in the rail is driven by `kind`.

    """
    return {
        'id': entry['id'],
        'name': entry.get('name') or entry['id'],
        'kind': LOCKED_VIEW_KIND,
        'config': {**base_config, 'cohortLock': entry['id']},
    }


#: ⭐ WAVE-27 item 27 (owner ruling R8) — the IG "Overview" view's CURATED COLUMNS, in the
#: owner's own order: handle, followers, engagement, location, last enriched.
#:
#: Written as candidates rather than as a requirement. The template registry REFUSES a template
#: whose columns the target lacks (`view_templates.missing_columns`) because applying one writes
#: the user's own views and a filter on a missing column silently WIDENS. This view is INJECTED,
#: not applied, and it filters nothing — so the proportionate rule is the opposite one: take the
#: columns the table has, in this order, and skip the rest. An IG database that predates a
#: column simply shows the other four.
#:
#: ⚠ `location_guess` is SESSION B's item-16 column and may not exist yet. That is exactly why
#: this list is intersected rather than asserted: a hard requirement here would make the whole
#: view vanish (or the assembly refuse) on every tenant until B lands, and then appear by
#: surprise. `profile_url` closes the list as the click-through, which is what makes the view
#: usable rather than merely informative.
IG_OVERVIEW_COLUMNS = ('handle', 'full_name', 'followers', 'avg_engagement',
                       'location_guess', 'enriched_at', 'profile_url')

#: The id is PINNED, the `view_templates` discipline: re-assembling must update the same view
#: rather than mint "Overview 2". It also lets a user's own edits overlay it through the saved
#: -config loop below, exactly as a cohort projection does.
IG_OVERVIEW_ID = 'tpl_overview'

#: How this function recognises an IG preset database WITHOUT importing the engine: two of the
#: profile preset columns is a stronger signal than any single one (a hand-made table could
#: plausibly own a column called `followers`; owning `followers` AND `avg_engagement` AND
#: `handle` is the preset set). `core/` must stay importable without the API layer, so this
#: mirrors `user_tables.PROFILE_PRESET_KEYS` the way that module mirrors the engine's.
_IG_SIGNATURE = ('handle', 'followers', 'avg_engagement')


def _ig_overview_view(fields, base):
    """R8's curated Overview, or None when this table is not an Instagram one."""
    keys = {f['key'] for f in fields}
    if not all(k in keys for k in _IG_SIGNATURE):
        return None
    visible = [k for k in IG_OVERVIEW_COLUMNS if k in keys]
    return {
        'id': IG_OVERVIEW_ID,
        'name': 'Overview',
        'kind': 'system',
        # NOT `locked`. The system view is locked because it is the identity of the table ("show
        # me everything"); this one is a STARTING LAYOUT, and R8 calls it curated rather than
        # fixed. A user who wants a sixth column should get one.
        'note': 'The five things worth seeing first on a creator. Sorted by reach.',
        'config': {
            **dict(base),
            'visible': visible,
            'order': visible + [k for k in (f['key'] for f in fields) if k not in visible],
            'sorts': ([{'colId': 'followers', 'dir': 'desc'}]
                      if 'followers' in keys else []),
        },
    }


def views_from_defs(defs, saved_views, fields, system_name="All customers", locked_lists=None,

                    view_order=None):
    """Convert legacy list formulas into the shared serializable SavedView contract.



    `system_name` (wave 16 C-TOPIC) labels the system view per TOPIC ("All products" on the

    product surface). The ID stays "all-customers" on every topic — the client pins it

    (UNDELETABLE_VIEW_IDS, the landing default), and an id that varies by surface would fork

    that contract for a label's sake.



    `locked_lists` (wave 17 R1) are the caller's cohorts, each PROJECTED as a saved view whose

    id IS the cohort id — so every stored reference to that id (a `cohortLock` on another view,

    an `is part of` condition, a folder placement) keeps pointing at the same thing and no

    rewrite map is needed. Saved config OVERLAYS the projection through the same mechanism the

    `list:` views have always used, which is what gives a locked view its own sort, filter,

    columns and display mode with no new storage."""
    base = _default_view_config(fields)
    views = [{
        "id": "all-customers", "name": system_name, "kind": "system",
        "locked": True, "config": dict(base),
    }]
    # ⭐ WAVE-27 item 27 (R8) — the IG Overview, ABOVE All records.
    #
    # ⚠ INJECTED, not seeded into the store, and that is what makes "existing IG databases gain
    # it too" true with no migration and no write on a read path. It is the same mechanism the
    # system view above has always used; the pinned id means a user's own edits overlay it
    # through the saved-config loop below rather than forking a second view.
    _overview = _ig_overview_view(fields, base)
    if _overview:
        views.insert(0, _overview)
    op_map = {">=": "gte", ">": "gt", "<=": "lte", "<": "lt", "=": "eq",
              "contains": "contains"}
    for name, definition in (defs or {}).items():
        filters = []
        for rule in definition.get("rules") or []:
            if rule.get("field") not in {field["key"] for field in fields}:
                continue
            filters.append({
                "colId": rule["field"],
                "op": op_map.get(rule.get("op"), "eq"),
                "value": str(rule.get("value") if rule.get("value") is not None else ""),
            })
        sort = str(definition.get("sort") or "")
        sorts = ([{"colId": sort.lstrip("-"),
                   "dir": "desc" if sort.startswith("-") else "asc"}]
                 if sort.lstrip("-") in {field["key"] for field in fields} else [])
        views.append({
            "id": "list:" + str(name),
            "name": str(name),
            "kind": "list",
            "note": str(definition.get("note") or ""),
            "config": {
                **base, "filters": filters, "sorts": sorts,
                "memberPids": [int(pid) for pid in definition.get("members") or []
                               if isinstance(pid, int) or str(pid).isdigit()],
            },
        })
    # Wave 17 R1 — the cohorts, as ordinary views. Appended BEFORE the saved-config overlay
    # below so a user's own edits to a locked view (its sort, its columns, its display mode)
    # land on the projection instead of creating a second row with the same id.
    for _entry in (locked_lists or []):
        if isinstance(_entry, dict) and _entry.get('id'):
            views.append(locked_view_projection(_entry, base))
    index = {view["id"]: i for i, view in enumerate(views)}
    for view_id, saved in (saved_views or {}).items():
        if not isinstance(saved, dict) or not isinstance(saved.get("config"), dict):
            continue
        clean = dict(saved)
        clean["id"] = str(view_id)
        if view_id in index:
            views[index[view_id]] = clean
        else:
            views.append(clean)
    # ⛔ WAVE 17 R1 — RE-STAMP THE LOCK AFTER THE OVERLAY. The loop above REPLACES a projected
    # view with its saved record, and a saved record that omits `cohortLock` would hand back a
    # view that shows the WHOLE BOOK under a locked view's name. That is not hypothetical: the
    # client rebuilds `config` on every autosave (a column resize is enough), and the lock is
    # identity here, not something the browser is the source of truth for. Read-side rather than
    # write-side-only on purpose — this also repairs any record already written by another path.
    _locked_ids = {e['id']: e for e in (locked_lists or [])
                   if isinstance(e, dict) and e.get('id')}
    if _locked_ids:
        for _v in views:
            _entry = _locked_ids.get(_v.get('id'))
            if not _entry:
                continue
            _v['kind'] = LOCKED_VIEW_KIND
            _v['config'] = {**(_v.get('config') or {}), 'cohortLock': _v['id']}
            # One thing, one name: the cohort store owns it (the rename event routes there), so
            # a stale `name` on the saved record can never fork into a second title.
            _v['name'] = _entry.get('name') or _v['id']
    # ── ⭐ WAVE-27 item 5, contract C7: the PER-USER VIEW ORDER ───────────────────────────────
    #
    # `view_order` is a list of view ids this user dragged into place. Applied LAST, over the
    # finished list, so it reorders whatever the assembly produced without having to know how any
    # of it got there (system, list:, cohort projection, saved, injected Overview).
    #
    # ⛔ THE SYSTEM VIEW STAYS AT INDEX 0 (C7), and it is re-pinned here rather than trusted to
    # sort correctly: `all-customers` is the client's landing default and one of its
    # UNDELETABLE_VIEW_IDS, so a stored order that happened to omit it — or list it third —
    # would move the rail's home row. ⚠ R8's Overview is the ONE thing allowed above it, because
    # the owner put it there; it is re-pinned with the system view so a drag cannot bury it
    # either. Both are facts about the table rather than the user's arrangement of it.
    #
    # ⚠ UNKNOWN IDS APPEND IN SERVER ORDER (C7). A view created since this order was stored, or
    # one shared to this user yesterday, must APPEAR — dropping it would make sharing look
    # broken, and the failure would be invisible to whoever shared it. Ids in the stored order
    # that no longer resolve are simply skipped.
    if view_order:
        _rank = {vid: i for i, vid in enumerate(view_order) if isinstance(vid, str)}
        _pinned = [v for v in views if v.get('id') in (IG_OVERVIEW_ID, 'all-customers')]
        _rest = [v for v in views if v.get('id') not in (IG_OVERVIEW_ID, 'all-customers')]
        # A stable sort over a rank that DEFAULTS TO THE END keeps unranked views in their
        # server order behind the ranked ones, rather than interleaving them by accident.
        _rest.sort(key=lambda v: _rank.get(v.get('id'), len(_rank) + 1))
        views = _pinned + _rest
    return views


def workspace_wire(ws, uname, pool_pids, defs=None, scope_key='customer', storage_key=None,

                   fields_base=None, with_cohorts=True):
    """The client's `GridWorkspace` WIRE SHAPE from the stored table workspace — the ONE

    projection, shared by both servers (app.py's `_table_grid` and the API's `/workspace`).



    ⛔ WHY THIS EXISTS (2026-07-30). The API route used to return the STORE shape with no

    `storageKey` — and the client validator (`fetchWorkspace`) requires one, so the standalone

    shell silently discarded the whole workspace: saved views never rendered and `cohortMode`

    never arrived (the Cohort route drew the Customer surface). Duplicating the host's inline

    projection into the route would have re-created the same drift one wave later; extracting it

    means the wire can only be one thing.



    Returns `(workspace, fields, views, cohort_lists)` — the extra three because the host

    interleaves further work (docs, derived cells, measure sets) that consumes them.



    HOST-ONLY extras stay with the host: `docs`/`docPayload`, `pool`, `hideViews`,

    `cohortMode`/`scopeChoice` (the API stamps its own from `?scope=`).



    Wave 16 C-TOPIC: `fields_base` selects the canonical contract (absent = customer,

    byte-identical).



    ⭐ WAVE 19 / R9 — `with_cohorts` NO LONGER MEANS "customer only". Wave 16 set it False on the

    product surface because cohorts were a single customer-keyed bucket, so resolving them against

    product pids would have intersected two unrelated id spaces and printed a plausible,

    meaningless member count. `modules.cohort` is scope-parameterized now: the lists come from

    THIS topic's bucket (`cohort_mod.scoped(scope_key)`), so their ids are this topic's ids and

    the intersection with `pool_pids` is the ordinary one. The flag survives as an honest OFF

    switch for a surface that wants no membership channel at all — it is not a topic wall.

    """
    import modules.cohort as cohort_mod

    cohort_lists = []
    if with_cohorts:
        for cid, c in sorted(cohort_mod.scoped(scope_key).visible(uname, pool_pids).items(),
                             key=lambda kv: (kv[1].get('name') or '').lower()):
            members = [p for p in (c.get('members') or []) if p in pool_pids]
            entry = {'id': cid, 'name': c.get('name') or cid, 'pids': members}
            # Rule 8b: a member can drop out of the 24-month pool without the cohort being
            # wrong, and a silently smaller cohort is exactly what the unverifiable-count rule
            # forbids.
            missing = len(c.get('members') or []) - len(members)
            if missing:
                entry['missing'] = missing
            cohort_lists.append(entry)

    fields = fields_from_workspace(ws, cohorts=bool(cohort_lists), scope_key=scope_key,
                                   fields_base=fields_base)
    views = views_from_defs(defs or {}, ws.get('views'), fields,
                            # Wave 21 (item 3, R6): the system view's name is TOPIC-DERIVED. A
                            # user database's default view used to read "All customers" — a
                            # compiled customer literal minted on every topic, one half of the
                            # owner's "my new database looks like RI's Customer table". The ID
                            # stays 'all-customers' everywhere (pinned client+server — the
                            # client's UNDELETABLE set and the view pin both name it).
                            system_name=("All products" if scope_key == 'product'
                                         else "All records"
                                         if str(scope_key or '').startswith('ut_')
                                         else "All customers"),
                            locked_lists=cohort_lists,
                            # ⭐ WAVE-27 item 5 (C7) — this user's own rail arrangement, from
                            # their own stratum. Read here rather than sorted by the client so
                            # the ORDER a request answers with is the order that was stored:
                            # sorting client-side would make the rail settle after a paint on
                            # every load, and shared views would land in server order first.
                            view_order=ws.get('viewOrder'))
    workspace = {'storageKey': storage_key, 'views': views, 'lists': cohort_lists}
    # Owner item 3 (2026-07-31): where this user left off. The client's own localStorage copy
    # wins when present; this is the server's answer for a FRESH browser, which used to fall
    # all the way to the system default view (and whatever display mode was stored on it).
    if ws.get('activeViewId'):
        workspace['activeViewId'] = str(ws['activeViewId'])

    # FOLDERS (owner item 11, contract C4), re-validated at SERVE time: a view or cohort can be
    # deleted by a path that knows nothing about folders, and the placement map must not outlive
    # the thing it points at.
    _folders = clean_folders(ws.get('folders'))
    _view_ids = {v['id'] for v in (views or []) if isinstance(v, dict) and v.get('id')}
    # ── WAVE 17 R1 (C-LOCKV amendment 2026-08-03): the two folder surfaces become ONE, AT
    # SERVE TIME rather than by a store migration. A locked view is an ordinary view now, so
    # its folder has to be an ordinary view folder — but rewriting the stored map would be a
    # one-shot write that has to be got right once, while this is a projection that is right
    # every time it runs. New drags write to `views` anyway (the client only knows that
    # surface), so `cohorts` drains on its own and never needs a second pass.
    # ⚠ A cohorts-surface folder whose id ALREADY names a views folder is DROPPED, not merged:
    # re-parenting somebody's list into a folder that merely shares an id is a worse outcome
    # than the list appearing at the root, where it is visible and one drag from home.
    _cf = list(_folders.get('cohorts') or [])
    if _cf:
        _vf = list(_folders.get('views') or [])
        _taken = {f['id'] for f in _vf}
        _order = len(_vf)
        for _f in _cf:
            if _f['id'] in _taken:
                continue
            _vf.append({**_f, 'order': _order})
            _order += 1
        _folders['views'] = _vf
    _raw_item_folders = dict(ws.get('itemFolders') or {})
    if _raw_item_folders.get('cohorts'):
        # Cohort placements now describe VIEWS (same ids — that is the point of preserving them).
        # A placement already stored on the views surface WINS: it is the more recent act.
        _raw_item_folders['views'] = {**dict(_raw_item_folders.get('cohorts') or {}),
                                      **dict(_raw_item_folders.get('views') or {})}
    _placed = clean_item_folders(
        _raw_item_folders, _folders,
        {'views': _view_ids, 'cohorts': {c['id'] for c in cohort_lists}})
    if _folders.get('views'):
        workspace['folders'] = _folders['views']
    # ⛔ `cohortFolders` IS NO LONGER EMITTED. The rail has no cohorts section to fold, and a
    # wire that still described one would invite a second rendering of rows that are now views.
    _vplaced = _placed.get('views') or {}
    for _v in (views or []):
        if isinstance(_v, dict) and _v.get('id') in _vplaced:
            _v['folderId'] = _vplaced[_v['id']]
    _cplaced = _placed.get('cohorts') or {}
    for _c in cohort_lists:
        if _c['id'] in _cplaced:
            _c['folderId'] = _cplaced[_c['id']]

    # RECORD LAYOUT (wave 2026-08-02, C-LAYOUT): the per-user record-detail field order,
    # re-validated at SERVE time exactly like folders — a field can be deleted by a path
    # that knows nothing about this stratum, and a stale key must not outlive its field.
    _rl = ws.get('recordLayout')
    if isinstance(_rl, dict) and isinstance(_rl.get('order'), list):
        _fkeys = {f['key'] for f in fields if isinstance(f, dict) and f.get('key')}
        _order, _seen = [], set()
        for _k in _rl['order'][:200]:
            _k = str(_k or '')
            if _k and _k in _fkeys and _k not in _seen:
                _seen.add(_k)
                _order.append(_k)
        if _order:
            workspace['recordLayout'] = {'order': _order}

    return workspace, fields, views, cohort_lists


def embed_html_path():
    """The first existing candidate path for the inlined single-file build, or None."""
    for p in _EMBED_CANDIDATES:
        if p.is_file():
            return p
    return None


def scope_counts(shown, matched, total):
    """The honest 'N of M' a SERVER-WINDOWED table must carry (CG-2).



    `matched` and `total` MUST come from their own queries over the whole scope. Never pass

    `len(rows)` as `matched` — that is the silent [:N] this exists to prevent: the page would

    report the window size as though it were the result size.



    Refuses the shapes that could only be a mistake, because a wrong count here is invisible on

    screen (it looks like a smaller dataset, not like an error).

    """
    shown, matched, total = int(shown), int(matched), int(total)
    if matched > total:
        raise ValueError(f"matched ({matched}) exceeds total ({total}) — a filter cannot match "
                         f"more rows than the scope holds")
    if shown > matched:
        raise ValueError(f"shown ({shown}) exceeds matched ({matched}) — the window cannot hold "
                         f"more rows than the filter matched")
    return {"shown": shown, "matched": matched, "total": total, "windowed": True}



# ⛔ EXIT-6 (2026-08-04): `build_html`, `component_dir`, `render` and `_DECLARED_COMPONENTS` WERE
# HERE, and they are gone with Streamlit. They were the EMBED HOST — the path that declared the
# prebuilt bundle as a `streamlit.components.v1` custom component (or injected the single-file
# HTML build as a fallback) so the React grid could be drawn inside a Streamlit page.
#
# THIS MODULE ITSELF SURVIVES, and that distinction is the whole point: `aios_grid.py` is imported
# at 11 sites across `aios-web/api/` plus `harness/semantic.py` — it owns the canonical field
# contract, the workspace wire and the count envelope. Only the ~95 lines that knew about a HOST
# went; the rest never did. Its one and only `import streamlit` lived inside `render`, lazily, and
# left with it. `api/verify_no_streamlit.py` now gates that nothing here re-imports it.
#
# Deleted with them: `aios_grid_embed.html` + `aios_grid_component/index.html` (a 2.1 MB prebuilt
# bundle), `build_embed.py` that produced them, and `deploy_hf.py`'s embed-staleness guard. The
# React app is now served directly by the FastAPI container — there is no twin to keep fresh, so
# the entire class of "the code shipped but the bundle did not" is retired rather than guarded.