platform/core/user_tables.py

"""USER-CREATED TABLES (wave-9 I8 / contract C6) — databases that are not a connector.

The owner's "+ New" flow has two branches: connect a SOURCE (Odoo today) or *"create a completely
new database blank, where this time the user can actually insert row that they want."* The first
branch is the connector seam and lives in `harness/datastore.py`; THIS module is the second.

WHY A SEPARATE STORE AND NOT A REGISTRY ROW. `core/registry.py` is a static Python list read at
import time — it is the catalogue of modules the PRODUCT ships, and a tenant cannot append to it
at runtime without editing code. A user-created table is tenant data, so it lives in the tenant
store and is MERGED into the nav beside the registry rows. That also keeps the archived/allowed
machinery honest: a user table is never "archived", it is deleted, and it is never in
`allowed_modules` because it is not a module.

SHAPE (one store key, `user_tables`):

    {table_key: {'key', 'label', 'source', 'createdBy', 'created', 'fields': [...], 'rows': {...}}}

`source` is deliberately NOT 'Odoo' — the nav badge is generated from it, so a blank table reads
"Blank" and can never be mistaken for connected data. `rows` is `{row_id: {field_key: value}}`,
which is the same overlay shape `TableStore` already uses, so the grid's write path needs no new
storage concept.

⚠ ROW WRITES ARE ONLY EVER LEGAL HERE. An Odoo-sourced table is READ-ONLY at the source and the
row endpoints refuse unless `is_user_table(key)` — a user must not be able to invent a customer
in Odoo by typing into a grid. (Wave 18 note: the events seam has NO row event types; row
add/delete are REST endpoints on `aios-web/api/routes_tables.py`, and this predicate is their
wall.)

WAVE 18 (C3-UT): every function takes an optional `st` STORE HANDLE — anything exposing
`get(name)` / `update(name, fn, flush=)` / `exists(name)`. Default = `core.store` (tenant #0,
the Streamlit host, unchanged). The API passes the session's `TenantRuntime`, whose accessors
apply the tenant prefix / repo binding, so a Nurilab table lands in Nurilab's store.
"""
import datetime as _dt
import hashlib
import json as _json
import re

import core.store as store

STORE_KEY = 'user_tables'
MAX_TABLES = 40
MAX_LABEL = 60

#: ⭐ THE PER-TABLE ROW CEILING — 5,000 until 2026-08-09, and it was never a property of the
#: substrate. Owner: *"Why do we keep having this MAX_ROWS problem, 5000 is too little. We need to
#: exceed it… solve the rootcause."*
#:
#: ⛔ WHAT 5,000 ACTUALLY WAS. Two things wore the same number and neither justified it:
#:   * `add_row`'s human-insert door (below) — a guard against a runaway paste, not a measurement;
#:   * `odoo_relational.plan()`'s own refusal, which is what kept the Odoo databases at OPEN AR
#:     only and is the reason the owner could not find every Odoo id in them.
#: `_ensure_table` writes rows straight through `rt.update(STORE_KEY, …)` and never crosses either,
#: so the cap was never even enforced on the population it was blamed for. And `ig_master.py` has
#: run a **500,000**-row bucket the whole time, which settles the question of whether the store can
#: hold more than five thousand of anything.
#:
#: ⭐ THE REAL CONSTRAINT, MEASURED 2026-08-09 against tenant #0's own mirror, is the one this
#: number now expresses: every `ut_*` table lives in ONE `user_tables` document that
#: `Store.get`/`Store.update` copy with `json.loads(json.dumps(...))` on every call. Built from
#: real Odoo rows, one table costs:
#:
#:     customers   2,465 rows   0.43 MB     8 ms        invoices   31,418 rows   8.68 MB   158 ms
#:     products    5,829 rows   1.20 MB    26 ms        orders     32,700 rows   7.50 MB   129 ms
#:
#: ⭐ THE NUMBER IS DERIVED FROM TWO HEADROOMS, and both are checked rather than asserted by
#: `verify_odoo_relational._prove_row_ceiling` ([[rules-need-gates]]):
#:   * BYTES — the widest table the product ships measures **0.318 KB/row** (`ut_odoo_invoices`,
#:     13 columns), so 60,000 rows is **19.1 MB** against a 32 MB per-table budget: 40% spare.
#:   * GROWTH — the largest shipped population is 32,700 confirmed orders, growing ~8k/year, so
#:     the cap is 83% above what exists and does not bind for roughly three years.
#: ⚠ 100,000 WAS THE FIRST ANSWER AND IT WAS WRONG BY ITS OWN GATE: at the measured row width it
#: is 31.8 MB against the same budget — 0.6% of headroom, i.e. a ceiling that one extra column
#: turns red. A cap chosen so it *just* passes its own control is a cap chosen to look justified.
#: ⛔ WHEN THIS BINDS, THE FIX IS NOT A BIGGER NUMBER. It is D-87's per-table row-key split, after
#: which the budget applies to a table on its own rather than to a share of one shared document.
#:
#: ⚠ IT IS A PER-TABLE CEILING AND THE TENANT PAYS THE SUM. The four Odoo databases together are
#: 20.6 MB / ≈ 370 ms per composite copy — real, and the reason the per-table row KEY split is
#: booked as the next increment (DEBT D-87) rather than claimed here. Order lines (256,810 rows,
#: 63.9 MB, 2.57 s) stay out of the store under ANY JSON substrate; they are answered by the
#: read-through rollup, which never copies a row.
MAX_ROWS = 60_000

# ─────────────────────────────────────────────────────────────────────────────────────────────
# ⭐⭐ WAVE 30 / OWNER RULING R6 — THE CEILING ABOVE DOES NOT APPLY TO A CONNECTED SOURCE.
#
# Verbatim: *"there is no cap in how many data from the API source (as long as its from a
# connected source like Odoo) that can be pulled into the app. Make this a standing rule. Now if
# there is lag or it can't be done, you need to explicitly tell me why and recommend a fix."*
#
# ⭐ BOTH SENTENCES ARE THE RULE. The first removes the cap; the second says a cap that CANNOT be
# removed must be reported with its cause and a recommendation — `limit_report()` below is that
# sentence, and it is served on the wire rather than left in a comment. A silent limit is the
# failure mode, not a limit.
#
# ⚠ AND THE EDITABLE SUBSTRATE KEEPS ITS BOUND. `MAX_ROWS` is not a rule about data, it is a
# measurement of THIS document (the header above derives it from 0.318 KB/row against a 32 MB
# per-table budget). A `ut_*` database somebody types into still lives here and still pays that
# cost. What changes is that a CONNECTED table's rows stop living here at all — they are served
# read-through from the mirror (R7, `routes_odoo_tables`), where 971,034 GL lines already sit and
# are counted in ~24 ms. Conflating the two turns R6 into a rewrite of user tables.

#: Table keys DECLARED connected by the app layer. Populated by `routes_odoo_tables._sources()`;
#: `core` never imports up, so the app tells this layer rather than being interrogated by it.
_CONNECTED = set()


def register_connected(*table_keys):
    """Declare table keys as CONNECTED-SOURCE — their rows come from a connector and are served
    read-through, so this document's ceiling is not a fact about them."""
    for key in table_keys:
        k = str(key or '').strip()
        if k:
            _CONNECTED.add(k)
    return frozenset(_CONNECTED)


#: ⭐⭐ W31-T25 / CONTRACT C2 — the KEY-CONVENTION leg's prefixes are a REGISTRY, not a literal.
#:
#: `is_connected`'s third leg used to read `key.startswith('ut_odoo_')` inline. Meta Ads is the
#: second connector to spawn locked databases (`ut_meta_*`), and the alternative was a second
#: predicate in the app layer answering the same question — one question, two normalizers, which
#: is a recorded defect class here ([[one-question-two-normalizers]]). So the connector layer
#: DECLARES its prefix and this module keeps the single evaluator.
#:
#: ⛔ `'ut_odoo_'` IS SEEDED RATHER THAN REGISTERED, AND THAT IS THE POINT OF THE LEG. The whole
#: reason a naming convention sits below the registry is to answer in a process where nothing has
#: registered anything — a cold worker that never imported `routes_odoo_tables` still owes the
#: right answer for `ut_odoo_gl_lines`, and an empty seed would have it fall through to
#: `MAX_ROWS`, i.e. re-cap a connected source against R6. A newly registered prefix carries the
#: weaker guarantee that `register_connected` already carries: it holds once its registrar has run.
_CONNECTED_PREFIXES = {'ut_odoo_'}


def register_connected_prefix(*prefixes):
    """Declare a table-key prefix whose automation-sourced, records-locked tables are CONNECTED.

    The connector layer calls this beside `register_connected` (which names exact keys): the
    prefix answers for tables this process has not enumerated, the exact keys answer for the ones
    it has. Returns the whole registered set, so a caller can assert what it just joined.

    ⚠ A PREFIX ALONE NEVER MAKES A TABLE CONNECTED. `is_connected`'s third leg still demands
    `source == AUTOMATION_SOURCE` and `recordMode == AUTOMATION_RECORD_MODE`, because a person who
    names a database "Meta ads spend" gets a `ut_meta_…`-shaped key too — and exempting it from
    the ceiling on the strength of its NAME would be exactly the silent hole the fence exists for.
    """
    for prefix in prefixes:
        p = str(prefix or '').strip()
        if p:
            _CONNECTED_PREFIXES.add(p)
    return connected_prefixes()


def connected_prefixes():
    """The registered key prefixes — the ONE list a caller may read to name the same set.

    C2's parity gate reads this rather than re-spelling `ut_odoo_` / `ut_meta_` on the far side:
    a second literal is a second normalizer wearing an assertion's clothes.

    ⚠ AND IT IS THE REGISTRAR'S OWN RETURN VALUE, which is not tidiness — `verify_reachability`
    went RED the hour this shipped, correctly: a public function whose only callers are `verify_*`
    files is a feature no user can reach ([[reachable-is-not-the-same-as-built]]), and "another
    lane's gate will call it" is not a caller. Routing the write door's answer through the read
    door makes one construction site of the set instead of two, and makes the accessor
    load-bearing rather than decorative.
    """
    return frozenset(_CONNECTED_PREFIXES)


def is_connected(table_key, st=None):
    """Is this database backed by a connected source rather than by typing?

    THREE ANSWERS, in falling order of authority, and the third is fenced deliberately:
      1. the app layer registered it (`register_connected`) — the live path;
      2. the stored definition says so (`connected: True`) — survives a process with no registrar;
      3. ⚠ a REGISTERED KEY PREFIX (`register_connected_prefix`; seeded `ut_odoo_`, and Meta Ads
         joins it as `ut_meta_`), AND ONLY on a table that is automation-sourced with records
         locked. A person who names a database "Odoo foo" gets exactly that key, and exempting it
         from the ceiling on the strength of its name would be a silent hole. Their table is
         `Blank`-sourced and records-mutable, so it cannot reach this leg.
    """
    key = str(table_key or '').strip()
    if key in _CONNECTED:
        return True
    table = get(key, st) or {}
    if not table:
        return False
    if table.get('connected') is True:
        return True
    return (any(key.startswith(p) for p in _CONNECTED_PREFIXES)
            and table.get('source') == AUTOMATION_SOURCE
            and table.get('recordMode') == AUTOMATION_RECORD_MODE)


#: ⭐⭐ W30-T31 (D-87) — table keys whose rows DO NOT LIVE HERE AT ALL.
#:
#: A subset of `_CONNECTED`, and the distinction is the whole ticket. *Connected* says where the
#: DATA came from, so `MAX_ROWS` is not a fact about it (R6). *Read-through* says where the rows
#: are SERVED FROM, so they are not in this document — which is the only thing that makes the
#: per-request copy cheaper. The eight Odoo grids are all connected; only the ones the app layer
#: has actually bound to the mirror AND proven safe to un-materialise are registered here.
#:
#: ⛔ THE APP LAYER DECIDES, and it is not merely "is there a binding". `routes_odoo_tables`
#: checks three conditions, because getting any one of them wrong is a silent wrong answer rather
#: than an error: a binding must exist; NOTHING may fold or link at the table's stored rows (a
#: rollup over a table with no rows writes ZEROS, and `compute_relation_cells` reads the raw
#: document, not this module); and the population must fit a window while the client still asks
#: for whole tables. Every table that fails one is REPORTED, per table, on the status door.
_READ_THROUGH = set()


def register_read_through(*table_keys):
    """Declare that these databases are SERVED from the mirror and store no rows here."""
    for key in table_keys:
        k = str(key or '').strip()
        if k:
            _READ_THROUGH.add(k)
            _CONNECTED.add(k)          # read-through implies connected; never one without the other
    return frozenset(_READ_THROUGH)


def materialises(table_key, st=None, defn=None):
    """Does this database keep its rows in the shared `user_tables` document?

    True for everything except a read-through grid — including every editable table, every
    automation table, and a connected table whose conversion has not happened yet.

    ⛔ THE STORED FLAG IS NOT A CACHE OF THE REGISTRY, IT IS THE DURABLE HALF, and without it this
    predicate would be a live hazard rather than a lookup. `_READ_THROUGH` is filled by an app-layer
    route; a worker process that has never served one would answer True for a table whose rows were
    stripped weeks ago, read `rows: {}` and serve an EMPTY GRID with nothing going red. So the strip
    STAMPS the definition and the stamp is what a cold process reads. Same three-tier shape as
    `is_connected` above, minus the naming convention — there is no safe way to guess this one.

    ⚠ `defn` LETS A CALLER LEND THE DEFINITION IT IS ALREADY HOLDING, and on the row path that is
    not a micro-optimisation: every `get()` here is another whole-document deep copy under
    `Store._lock`, so asking this question inside `scoped_pool` without lending would have added a
    third copy to the very request this ticket exists to make cheaper.
    """
    key = str(table_key or '').strip()
    if key in _READ_THROUGH:
        return False
    if defn is None:
        defn = get(key, st) or {}
    return not bool((defn or {}).get('readThrough'))


def strip_materialised(st=None):
    """Remove the stored rows of every read-through database. Idempotent; returns what moved.

    ⭐ THIS IS THE HALF THAT MAKES THE INVARIANT TRUE RATHER THAN INTENDED. The spawn
    (`odoo_relational._ensure_table_inplace`) writes rows by mutating this document inside its own
    `rt.update(STORE_KEY, …)`, so it never passes through any function here — a guard on the write
    doors below would simply not be on that path. Re-running this after a refresh is therefore the
    enforcement, and being idempotent is what lets every door call it without coordinating.

    ⚠ The definition STAYS. A read-through table is a real database — its fields, label, views,
    grants and lock all live here; only the rows are elsewhere. Deleting the definition would take
    the permission wall and the nav entry with it.
    """
    out = {'dropped': {}, 'before': 0, 'after': 0}
    if not _READ_THROUGH:
        return out
    s = _st(st)
    try:
        cur = s.get(STORE_KEY) or {}
    except Exception:                                             # noqa: BLE001
        return out
    fat = {k: len((cur.get(k) or {}).get('rows') or {}) for k in _READ_THROUGH}
    fat = {k: n for k, n in fat.items() if n}
    if not fat:
        return out                    # the common case: nothing to do, and nothing measured
    # Only now is the measurement worth its own cost — `json.dumps` of a 20 MB document is not
    # free, and this exists to be reported (R6's second sentence), not to run on every request.
    out['before'] = len(_json.dumps(cur, default=str))

    def _strip(doc):
        for key in list(_READ_THROUGH):
            t = (doc or {}).get(key)
            if isinstance(t, dict) and t.get('rows'):
                out['dropped'][key] = len(t['rows'])
                t['rows'] = {}
                # ⛔ THE STAMP IS THE POINT — see `materialises`. Without it the next process to
                # boot reads an empty `rows` and calls that the answer.
                t['readThrough'] = True
        return doc

    s.update(STORE_KEY, _strip, flush='sync')
    try:
        out['after'] = len(_json.dumps(s.get(STORE_KEY) or {}, default=str))
    except Exception:                                             # noqa: BLE001
        out['after'] = 0
    return out


def row_limit(table_key, st=None):
    """The row ceiling that applies to this database: 0, None, or `MAX_ROWS`.

    ONE evaluator, so the write doors, the spawn's refusal and the wire's report cannot disagree
    about whether a table is capped ([[one-evaluator-per-question]]).

    ⛔ THE THREE ANSWERS ARE THREE DIFFERENT STATEMENTS AND 0 IS NOT "A VERY SMALL CAP":
      * **0** — this database stores no rows HERE at all; it is served read-through from the
        mirror. A caller that builds rows for it is doing wasted, dangerous work: 963,783 GL lines
        is ~240 MB of python dicts in one process. `odoo_relational.plan` reads this to decide
        whether to build a bucket at all.
      * **None** — connected, materialised, and UNCAPPED (R6): however many rows Odoo has.
      * **MAX_ROWS** — the editable substrate, which really is bounded by this document.
    """
    if not materialises(table_key, st):
        return 0
    return None if is_connected(table_key, st) else MAX_ROWS


def limit_report(table_key, st=None):
    """R6's second sentence as data: `{subject, cause, recommendation, effect}` or None.

    ⛔ `effect` is never `truncated`. Every enforcement site REFUSES — a truncated table
    understates every total it feeds while looking exactly like a complete one, which is the
    failure this whole module is arranged against.
    """
    cap = row_limit(table_key, st)
    if cap is None:
        return None
    if cap == 0:
        return {
            'subject': 'rows', 'limit': 0, 'effect': 'read_through',
            'cause': 'this database is served THROUGH the connector mirror, so its rows are '
                     'never copied into this tenant document and no ceiling applies to them',
            'recommendation': 'read it with a window (`/odoo-tables/{key}/rows`); the row count '
                              'you see there is a SQL count over the whole table',
        }
    return {
        'subject': 'rows', 'limit': cap, 'effect': 'refused',
        'cause': f'this database is EDITABLE, so its rows live in the shared `{STORE_KEY}` '
                 f'document that every request copies; at the widest shipped row (0.318 KB) '
                 f'{cap:,} rows is 19.1 MB against a 32 MB per-table budget',
        'recommendation': 'connect it to a source instead — connected data is served '
                          'read-through from the mirror and is not bounded (R6); for typed data '
                          'the increment is D-87, a per-table row key, not a bigger number',
    }


MAX_FIELDS = 60
#: the badge the nav paints for a table with no connector behind it
BLANK_SOURCE = 'Blank'
#: automation-created tables carry their maker instead (wave 18 C4-AUTO)
AUTOMATION_SOURCE = 'Automation'
AUTOMATION_RECORD_MODE = 'automation'
#: every user table's key is prefixed, so it can never collide with a registry module key
KEY_PREFIX = 'ut_'


#: ⭐⭐ W31 QA — KEYS WHOSE RECORDS ARE MACHINE-OWNED **BY DECLARATION**, registered by the layer
#: that owns the connector (`automation_engine` at import), exactly as `register_connected_prefix`
#: is. `core` never imports up, so the engine pushes its list down here rather than this module
#: reaching for it.
#:
#: ⛔ WHY A REGISTRY AND NOT THE STORED FLAG ALONE — THE OWNER FOUND THIS IN PRODUCTION.
#: `recordMode` is stamped by `ut_ensure`, so a table only became locked THE NEXT TIME AN
#: AUTOMATION WROTE TO IT. W31-T32 proved that migration works and read it as done; it means every
#: TikTok database already sitting in a tenant stayed unlocked — offering "+ New record" — until
#: somebody re-ran a TikTok automation, which nobody had. Owner, verbatim: *"the databases for
#: Tiktok do not have the small lock icon as I asked."* A lock that arrives on the next write is a
#: lock that is absent exactly while a person is looking at it.
#: ⚠ THE REGISTRY IS THE AUTHORITY AND THE STORED FLAG STILL COUNTS — either one locks. The flag
#: keeps working for tables nobody declares, and the declaration needs no store write, no boot
#: order and no per-tenant migration, so it is true for every tenant the moment the module loads.
#: ⚠ LOCKED DATABASE, NOT READ-ONLY (DESIGN.md §4): this predicate governs RECORDS. Adding a FIELD
#: must still work, and `may_edit_field` is a different wall.
_LOCKED_RECORD_KEYS = set()


def register_locked_records(*keys):
    """Declare table keys whose records are machine-owned. Idempotent; -> the registered set."""
    for k in keys:
        for one in (k if isinstance(k, (set, frozenset, list, tuple)) else (k,)):
            if str(one or '').strip():
                _LOCKED_RECORD_KEYS.add(str(one).strip())
    return frozenset(_LOCKED_RECORD_KEYS)



def records_mutable(table_key, st=None):
    """May a human add/edit/delete records in this database?

    Ordinary and Profile databases default open. Automation-owned child datasets opt out — by
    DECLARATION (`register_locked_records`, which needs no write) or by the stored table-level
    mode; engine writers use their direct coalesced path and are intentionally not routed through
    this human-door predicate.
    """
    table = get(table_key, st) or {}
    if not table:
        return False
    if str(table_key) in _LOCKED_RECORD_KEYS:
        return False
    return table.get('recordMode') != AUTOMATION_RECORD_MODE

#: The field types a user table may declare. ⚠ Kept a SUBSET of
#: `aios_grid.CUSTOM_FIELD_TYPES` (gated in verify_api's W18-UT section). `created_time` stays
#: out (a row-datum kind — a base column of it would have no author);
#: `image` is unchanged this wave. `automation` JOINED in WAVE 21 (item 7a,
#: contract C2): the engine and the automation-column picker read the DEFINITION, so a
#: grid-created automation column must be able to live there — the column's CELLS stay
#: machine-written (the engine is the writer). Before this, the grid's automation columns
#: landed in the per-user workspace stratum and the picker could not see them — the owner's
#: "Choose a column..." stays empty. Local literal rather than an aios_grid import so this
#: module stays dependency-light for the API's boot path.
#: ⭐ `json` JOINED IN WAVE 23 (C7) — and it belongs here rather than staying grid-only the way
#: `image` did, because the databases that need it are exactly these: a scrape lands a comment
#: thread or a webhook payload against a row, and the automation engine writes those cells.
#: ⭐ 2026-08-07 — `link` and `rollup` JOINED (the relational wave). They belong here for
#: `metric`'s reason, not `formula`'s: their cells are MACHINE-COMPUTED SERVER-SIDE and
#: materialised into the row, because a rollup reads ANOTHER TABLE'S ROWS — data the client has
#: not loaded and must not have to.
#: ⭐⭐ 2026-08-10 — `formula` JOINED, and the argument that kept it out was answered rather than
#: overruled. "A base column of it would have no author" was about its VALUES, and that half is
#: still true: a formula cell is computed in the browser from the row's other cells and is stored
#: NOWHERE, here or anywhere else. What joins this set is the DEFINITION — the expression, the
#: label, the column's existence. Excluded, those landed in `<table>_table_workspace`, the
#: PER-USER overlay: measured on nurilab as `Trimmed reel views`, a column only its creator could
#: see on a database four people share. That is not a narrower feature, it is the same shape as
#: the rollup defect one stratum over ([[rollup-lives-in-the-definition-not-the-overlay]]) with a
#: different consequence — not "nothing can compute it" but "nobody else can SEE it".
#: ⛔ SAY THE LIMIT OUT LOUD, because this line will be read as a bigger promise than it is: the
#: values still do not exist server-side, so an automation, a rollup, a `where` clause and an
#: export still cannot read a formula column. Making the definition shared does not make the
#: number stored. A number the server must read is a `rollup` (link fold) or a `metric`.
#: ⚠ Its cells are therefore in `is_computed_cell` — the WRITE wall. Without that, `grid_events`
#: would accept a paste into a column whose renderer overwrites it on the next paint.
#: ⛔ `image` WAS MISSING FROM THIS SET UNTIL WAVE 27, AND THAT WAS A LIVE DEFECT — recorded
#: here because the shape repeats and the next kind will be added by somebody reading this line.
#: Wave-19 R7 landed `image` on FOUR of the five surfaces (the `FieldType` union, `CREATABLE_TYPES`,
#: `aios_grid.CUSTOM_FIELD_TYPES`, the shape/label tables) and missed this one. The client
#: therefore offered an Image column in EVERY column menu, including on `ut_*` databases, while
#: `_clean_field` returned None for it — so on a user database the column was created, named,
#: configured, and GONE on the next read, with nothing anywhere going red. It survived three
#: waves because `verify_fields_contract` diffed the client's offer against the ODOO grid's
#: acceptance and never against this set. Wave-27 item 13 adds that derived leg, which is what
#: found it (measured, not reviewed: `_clean_field({'type':'image'})` returned None).
UT_FIELD_TYPES = {'text', 'select', 'multiselect', 'user', 'int', 'currency', 'pct', 'date',
                  'checkbox', 'phone', 'email', 'url', 'rating', 'automation', 'json',
                  'link', 'rollup', 'code', 'image', 'formula'}
#: A cell is still a scalar on the wire (`Row` values are strings/numbers) — what this bounds is
#: the DOCUMENT inside it.  Source data is one already-paid provider response; keep it whole so a
#: provider field is never silently discarded merely because the response is rich. 32 MiB matches
#: the automation transport's guarded maximum response size.
MAX_JSON_CELL = 32 * 1024 * 1024

# ---------------------------------------------------------------------------------------------
# THE PROFILE FLAG (wave 25, contract C3 / owner ruling R7)
# ---------------------------------------------------------------------------------------------
# R7: *"the Profile field is a FLAG on an ordinary text field, not a new field kind"* — so it
# works on databases that already exist, and every reader that does not know about it goes on
# rendering a text column. `profile: {source: 'instagram'}` present = flagged; absent = ordinary.
#
# ⛔ THE FLAG IS ONLY WORTH HAVING BECAUSE OF THE TWO MITIGATIONS R7 ACCEPTED AS OPEN HOLES, and
# they are what this block is: AT MOST ONE per table (`_profile_of`, enforced at both write
# doors) and the flag VALIDATES ITS CELL (`normalize_profile`). Without the first, "the profile
# column" is not a thing the enrich action can resolve; without the second, the flag is a label.

#: Where a profile handle points. A CLOSED vocabulary, for the reason every other closed enum
#: here is closed: a source nobody wrote a reader for renders as a column that promises a link.
#:
#: ⭐ WAVE-29 CONTRACT C2 (owner R1/R2, 2026-08-11): `tiktok` joins it, so a `ut_tt_*` row can
#: carry a profile flag the enrich action resolves. **A source is not a NAME, it is three rules**
#: (below) — the vocabulary and the rules move together, because adding the word alone would
#: accept a TikTok handle and then hand it an `instagram.com` link from `profile_url`.
PROFILE_SOURCES = ('instagram', 'tiktok')

#: Hosts whose `/<handle>/` path IS an Instagram profile. `www.` is stripped before the lookup.
_IG_HOSTS = ('instagram.com', 'instagr.am')

#: URL path segments that are NOT handles. `instagram.com/p/Cxyz` is a POST and
#: `instagram.com/explore/tags/silk` is a search — both are things a person pastes, and both
#: would otherwise normalise to a handle (`p`, `explore`) that can never resolve.
_IG_RESERVED = frozenset({'p', 'reel', 'reels', 'tv', 'stories', 'explore', 'accounts',
                          'direct', 'about', 'developer', 'legal', 'privacy', 'terms'})

#: Instagram's own handle rule: 1–30 of letters, digits, period, underscore.
_IG_HANDLE_RE = re.compile(r'^[A-Za-z0-9._]{1,30}$')

#: TikTok's profile hosts. ⛔ `vm.tiktok.com` is deliberately ABSENT: those are short links that
#: usually resolve to a VIDEO, and accepting one would store a "handle" (the opaque code) that can
#: never enrich — the same failure `_IG_RESERVED` exists to prevent, one layer earlier.
_TT_HOSTS = ('tiktok.com', 'm.tiktok.com')

#: TikTok's non-handle path segments. `tiktok.com/@user/video/123` is already refused by the
#: one-segment rule; these are the ONE-segment paths that are still not people.
_TT_RESERVED = frozenset({'video', 'tag', 'music', 'discover', 'search', 'foryou', 'explore',
                          'live', 'upload', 'about', 'legal', 'privacy', 'terms', 'effect',
                          'business', 'ads', 'node', 'login', 'signup'})

#: TikTok's own handle rule: 2–24 of letters, digits, period, underscore. ⚠ NOT Instagram's —
#: the charset matches but the bounds do not, and a shared regex would accept a 30-character
#: TikTok "handle" that the vendor answers nothing for.
_TT_HANDLE_RE = re.compile(r'^[A-Za-z0-9._]{2,24}$')

#: ⭐ C2 — the per-source rules, ONE table, so `PROFILE_SOURCES` cannot grow a member that no
#: normaliser knows. Every function below reads THIS rather than branching on the word: a source
#: with no entry refuses at the door instead of silently taking Instagram's rules and Instagram's
#: URL, which is what "add the word and ship" would have produced.
#:
#: `at_handle` = the profile path wears a leading `@` (TikTok's `/@name`; Instagram's `/name`).
#: It is stripped before validation and re-added by `profile_url`, so the STORED truth is the bare
#: handle in both families and a filter on the column cannot miss half of it.
_PROFILE_RULES = {
    'instagram': {
        'hosts': _IG_HOSTS,
        'reserved': _IG_RESERVED,
        'handle': _IG_HANDLE_RE,
        'at_handle': False,
        'url': 'https://www.instagram.com/{handle}/',
    },
    'tiktok': {
        'hosts': _TT_HOSTS,
        'reserved': _TT_RESERVED,
        'handle': _TT_HANDLE_RE,
        'at_handle': True,
        'url': 'https://www.tiktok.com/@{handle}',
    },
}


def _profile_rules(source):
    """The rule bag for a source, or None — the ONE resolution of "is this source real"."""
    key = str(source or '').strip().lower()
    if key not in PROFILE_SOURCES:
        return None
    return _PROFILE_RULES.get(key)

#: The C1 preset keys a profile field OWNS — the cells R6 clears when its handle is blanked.
#:
#: MIRRORED from the engine's C1 set as local literals, the same discipline (and for the same
#: boot-path reason) as `METRIC_MEASURES` below; the pair is held in step by a parity check
#: rather than by an import, because `core/` must stay importable without the API layer.
#:
#: Human-written columns are deliberately absent. `ut_ensure` stamps automation provenance on
#: generated fields, so clearing by that tag alone would erase unrelated machine-owned columns.
#: This is therefore an intersection of the known Instagram preset keys and the provenance tag.
PROFILE_PRESET_KEYS = (
    # the profile facts themselves (CANDIDATE_FIELDS + the enrichment run_field_instagram pulls)
    'full_name', 'followers', 'following', 'posts_count', 'avg_engagement', 'bio',
    'external_url', 'verified', 'category', 'business_category', 'is_business',
    'is_professional', 'is_private', 'highlights_count', 'bio_hashtags',
    'pronouns', 'ig_id', 'profile_url',
    # ⭐ 2026-08-07 — the rest of the Bright Data profile schema, promoted to preset columns by
    # owner instruction. They are profile FACTS like the nineteen above, so blanking the handle
    # clears them for the same reason (R6): they describe an account this row no longer names.
    'profile_name', 'is_joined_recently', 'has_channel', 'partner_id', 'external_url_title',
    'fbid', 'related_accounts', 'country_code', 'source_payload',
    # relational projections and summaries owned by the Instagram graph
    'posts_link', 'profile_snapshots_link', 'post_snapshots_link', 'comments_link',
    'avg_views_12', 'avg_plays_12', 'avg_likes_12', 'avg_comments_12',
    'posts_captured', 'profile_reads', 'post_measurements_captured', 'comments_captured',
    # R3's LATEST-value stamp. The full series stays in `ut_ig_snapshots` — one store for one
    # series — which is why clearing here can never be a history delete.
    'enriched_at',
)


def _clean_profile(raw):
    """One `profile` bag → the stored shape, or None (refused).

    Shaped exactly like `_clean_metric`: a mis-shaped flag REFUSES THE FIELD rather than being
    dropped, because a column stored without the flag it was created with is a column whose
    cells nothing will ever validate — the silent half of the same mistake.
    """
    if not isinstance(raw, dict):
        return None
    source = str(raw.get('source') or '').strip().lower()
    if source not in PROFILE_SOURCES:
        return None
    return {'source': source}


def normalize_profile(value, source='instagram'):
    """A typed profile cell → `(handle, ok)`. THE cell validator the flag promises (C3).

    Accepts a bare handle (`name`), an at-handle (`@name`) or a profile URL of THIS SOURCE in any
    of the forms a person actually pastes (`https://www.instagram.com/name/?hl=en`,
    `instagram.com/name`, `https://www.tiktok.com/@name`). Returns the BARE HANDLE — the URL form
    is derived for display, so the stored truth has exactly one spelling and a filter on it cannot
    miss half the column.

    ⛔ C2: the host, the reserved segments and the handle rule all come from `_PROFILE_RULES[source]`.
    A TikTok URL pasted into an Instagram-flagged column is REFUSED, and vice versa — a cell that
    names the wrong network is not a handle this row's automation can enrich, and quietly accepting
    it would defer the failure to a run that then answers nothing.

    `('', True)` for a blank: clearing a profile cell is legal and is the whole of R6.
    `(None, False)` for anything else, and the caller refuses the write with a sentence — a
    value that is neither is not a handle we can enrich, and storing it would put the failure
    off until the automation runs and answers nothing.
    """
    rules = _profile_rules(source)
    if rules is None:
        return (None, False)
    raw = str(value if value is not None else '').strip()
    if not raw:
        return ('', True)                      # the clear — R6's whole subject
    cand = raw.lstrip('@')
    # ⚠ THE URL TEST IS `/`, AND NOTHING CLEVERER. An earlier version also treated a DOT as a
    # sign of a URL, which refused `@nuri.lab_1` — a period is legal in an Instagram handle and
    # common in real ones, so that test rejected a whole class of valid input while every URL
    # case still passed. A bare handle can never contain a slash; every URL form does.
    if '/' in cand:
        # Parse rather than regex it: a hand-rolled pattern is how
        # `instagram.com.evil.test/name` gets read as Instagram.
        probe = cand if '//' in cand else 'https://' + cand.lstrip('/')
        try:
            from urllib.parse import urlsplit
            parts = urlsplit(probe)
        except Exception:
            return (None, False)
        host = (parts.netloc or '').split('@')[-1].split(':')[0].strip().lower()
        if host.startswith('www.'):
            host = host[4:]
        if host in rules['hosts']:
            segs = [s for s in (parts.path or '').split('/') if s]
            # C2: TikTok's profile path is `/@name`; Instagram's is `/name`. The `@` is stripped
            # here and re-added by `profile_url`, so both families STORE a bare handle.
            first = segs[0].lstrip('@') if segs else ''
            if len(segs) != 1 or not first or first.lower() in rules['reserved']:
                # Two segments is a post/reel/video, zero is the site itself. Refused rather
                # than taking the first segment, which is what turns `/p/Cxyz` into handle `p`.
                return (None, False)
            cand = first
        elif host:
            return (None, False)               # a URL, but not one of THIS source's
    if not rules['handle'].match(cand) or cand.lower() in rules['hosts']:
        # The host check catches a bare `instagram.com` — it matches the handle charset, so
        # without this it would normalise to a "handle" that can only ever resolve to nothing.
        return (None, False)
    return (cand.lower(), True)


def profile_url(handle, source='instagram'):
    """The DERIVED display form of a stored handle (C3: *"the URL form is derived for display"*).
    Derived and never stored, so the two spellings cannot drift apart.

    ⛔ C2 — the template comes from the SOURCE's own rules. This function used to hardcode
    `instagram.com` while its guard merely checked membership of `PROFILE_SOURCES`, so the moment
    that tuple grew a second member it would have handed every TikTok handle an Instagram link —
    a guard that passes and a body that lies ([[hardcoded-fallback-not-hardcoded-key]]).
    """
    h = str(handle or '').strip().lstrip('@')
    rules = _profile_rules(source)
    if not h or rules is None:
        return ''
    return rules['url'].format(handle=h)


def _profile_of(fields, exclude=None):
    """The profile-flagged field in this list, or None. `exclude` skips one key — a PATCH of the
    profile field itself must not read as "a second one already exists"."""
    for f in (fields or []):
        if (isinstance(f, dict) and isinstance(f.get('profile'), dict)
                and f.get('key') != exclude):
            return f
    return None


def profile_field(table_key, st=None):
    """THE profile column of one table, or None. ONE resolver, so "which column holds the
    handle" has exactly one answer for the write door, the clear, the grid and the engine."""
    return _profile_of((get(table_key, st) or {}).get('fields') or [])


def _st(st):
    return st if st is not None else store


#: Wave 22 (contract C3) — the ROW-EVENT seam. The API layer APPENDS listeners (the automation
#: engine's trigger hook, registered by `routes_automation` — the one module that imports both
#: sides); platform code only ever EMITS. This is what keeps the Notion loop-prevention law
#: structural: the emit sites are the HUMAN write doors (`add_row` here, `overlay_patch` in
#: grid_events) and nothing the engine itself writes through, so an automation's write cannot
#: fire an event trigger — its own or a sibling's — by construction.
ROW_HOOKS = []


def emit_row_event(evt):
    """Fan one row event out to every registered listener. A listener failure is swallowed —
    a broken trigger must never break typing into a cell — but it is swallowed LOUDLY."""
    for hook in list(ROW_HOOKS):
        try:
            hook(evt)
        except Exception as e:                                    # noqa: BLE001
            print(f'[user-tables] row hook failed: {type(e).__name__}: {e}')


def _slug(label):
    s = re.sub(r'[^a-z0-9]+', '_', str(label or '').strip().lower()).strip('_')
    return (s or 'table')[:40]


def _ag_formula(raw):
    """`aios_grid._clean_formula`, reached the way this module reaches everything one layer up.

    ⚠ ONE call site for the function-local import, shared by BOTH field doors (`clean_fields` and
    `_clean_field`). `core` must not import `aios_grid` at module level — the API's boot path
    depends on this module staying dependency-light — and the `code` branch already pays that
    cost twice. A third and fourth copy of `import aios_grid as _agX` is how one of them ends up
    calling a different validator.

    ⚠ `valid_keys` IS DELIBERATELY NOT PASSED. See the note at `_clean_field`'s formula branch:
    neither door has reliable sibling-field context, and a second, weaker copy of the ref rule is
    worse than leaving refs to the read-time behaviour `_clean_formula` already documents.
    """
    import aios_grid as _agf
    return _agf._clean_formula(raw)


#: Display formats a field may declare. ⛔ DISPLAY ONLY — none of these touches the stored value,
#: which stays the scalar the fold or the mapper wrote. That separation is the whole safety of the
#: feature: a `thousands` setting can never make a number wrong, only easier to read.
#: ⚠ `tz`/`time` belong to the date family and `thousands`/`decimals`/`abbrev` to the number family;
#: they are validated together because ONE bag rides one field and the renderer already reads only
#: the keys its own type understands (`display.numberText` / `display.dateTimeText`).
FORMAT_TZ = ('local', 'utc')
FORMAT_MAX_DECIMALS = 4


#: ⭐ WAVE-29 CONTRACT C7 — the COLUMN-SUMMARY vocabulary a field's `agg` may take.
#:
#: MIRRORED from `aios_grid.FIELD_AGGS` (session C publishes it) and from the client's
#: `aggregations.FIELD_AGGS`, as a local literal for the reason `METRIC_MEASURES` and
#: `PROFILE_PRESET_KEYS` are local literals here: `core/` must stay importable without the layers
#: above it. Three copies, one law, held in step by C's parity gate rather than by an import.
#:
#: ⛔ NOT `CHART_AGGS` (which spells it `avg` and gatekeeps STORED chart values with live data)
#: and NOT `ROLLUP_FNS` (16 fold names, overlapping but not the same list). Merging any two of the
#: three silently turns live stored charts into sums.
FIELD_AGGS = ('sum', 'average', 'median', 'min', 'max', 'count')

#: A choice colour, as the client's editor spells one. `#RRGGBB`, upper-cased on the way in so a
#: stored value has exactly ONE spelling and two fields cannot disagree about `#ebf6ef`.
_OPTION_COLOR_RE = re.compile(r'#[0-9A-F]{6}')


def _clean_option_colors(raw, options):
    """`{option label → #RRGGBB}`, capped to the field's own option vocabulary (wave-29 C1).

    ⛔ MIRRORED from `aios_grid._clean_option_colors` as a LOCAL function rather than imported, for
    the reason `METRIC_MEASURES` and `PROFILE_PRESET_KEYS` are local literals here: `core/` must
    stay importable without the layers above it, and `aios_grid` sits above this module. The pair
    is held in step by a gate leg, never by an import.

    ⚠ THE CAP IS THE POINT, not tidiness: colours are keyed by LABEL, so a renamed or deleted
    option would otherwise leave a colour behind that nothing renders and no editor can reach —
    and on the next rename that orphan could re-attach to a different option with the same name.
    Case-insensitive on the way in, canonical on the way out, so `Done` and `done` cannot both
    claim the same option.
    """
    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 _OPTION_COLOR_RE.fullmatch(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 _clean_format(raw):
    """One `format` bag → the stored shape, or None.

    ⛔⛔ THIS FUNCTION EXISTS BECAUSE THE BAG HAD NO STORAGE AT ALL. The column menu has shipped a
    "Field format" pane (thousands separator, decimal places, abbreviate) since wave 5, and on a
    `ut_*` database BOTH write doors built their entry dict key-by-key and never copied `format` —
    so the pane saved, the request succeeded, and the setting was gone by the next read. It looked
    like a rendering bug and was a persistence one. Same shape as
    [[flag-shipped-without-its-writer]] from the other end: here the writer exists and the STORE
    does not.

    ⚠ EVERY KEY IS OPTIONAL AND AN EMPTY BAG IS None, so a field that declares no format is stored
    exactly as it was before this existed — the wave-5 note's own parity rule.
    """
    if not isinstance(raw, dict):
        return None
    out = {}
    if isinstance(raw.get('thousands'), bool):
        out['thousands'] = raw['thousands']
    if isinstance(raw.get('abbrev'), bool):
        out['abbrev'] = raw['abbrev']
    if isinstance(raw.get('time'), bool):
        out['time'] = raw['time']
    decimals = raw.get('decimals')
    # `bool` is an `int` subclass, so `True` would otherwise store as 1 decimal place — a checkbox
    # arriving in a numeric slot should be refused, not interpreted (the `sigmas` lesson).
    if isinstance(decimals, int) and not isinstance(decimals, bool):
        if 0 <= decimals <= FORMAT_MAX_DECIMALS:
            out['decimals'] = decimals
    tz = str(raw.get('tz') or '').strip().lower()
    if tz in FORMAT_TZ:
        out['tz'] = tz
    return out or None


def clean_fields(raw):
    """Validate a caller-supplied field list into the stored shape — refuse junk, never store
    it. Returns a list of `{key,label,type,source}` dicts (source is always 'overlay': every
    base column of a user table is user-editable by construction), or None when nothing
    survives. Unknown types are DROPPED, not coerced — a column silently retyped is a lie."""
    out, seen = [], set()
    for f in (raw or [])[:MAX_FIELDS]:
        if not isinstance(f, dict):
            continue
        label = ' '.join(str(f.get('label') or f.get('key') or '').split())[:80]
        key = re.sub(r'[^a-z0-9_]+', '_', str(f.get('key') or _slug(label)).strip().lower())
        key = key.strip('_')[:60]
        ftype = str(f.get('type') or 'text').strip().lower()
        if not key or key in seen or ftype not in UT_FIELD_TYPES or not label:
            continue
        seen.add(key)
        entry = {'key': key, 'label': label, 'type': ftype, 'source': 'overlay'}
        # ⭐ 2026-08-10 — the DISPLAY format, carried through the CREATE door. See `_clean_format`:
        # this is the fifth time an entry built key-by-key has silently dropped a bag the editor
        # sends, and the fourth of those was `agg` on this very line one wave ago.
        fmt = _clean_format(f.get('format'))
        if fmt:
            entry['format'] = fmt
        description = ' '.join(str(f.get('description') or '').split())[:300]
        if description:
            entry['description'] = description
        if ftype in ('select', 'multiselect'):
            opts = [' '.join(str(o).split())[:60] for o in (f.get('options') or [])
                    if str(o).strip()][:50]
            if opts:
                entry['options'] = opts
        if f.get('default') is True:
            entry['default'] = True
        # ⭐ WAVE-27 item 19 (R15) + D-80 — THE PRIMARY COLUMN, CARRIED THROUGH THE CREATE DOOR.
        #
        # `_clean_field` has honoured `pinned` since 2026-08-07; THIS door silently dropped it,
        # and the split the comment below names is why. It did not matter while `create()` minted
        # one column — with nothing pinned, `fields.find(f => f.pinned) ?? fields[0]` made the
        # only column the primary either way. R15 seeds FOUR at once, so "whichever came first"
        # stops being a coincidence that happens to be right, and D-80 is the register entry for
        # exactly that fallback quietly being the whole rule. MEASURED before fixing:
        # `clean_fields([{...'pinned': True}])` returned the field without the key.
        if f.get('pinned') is True:
            entry['pinned'] = True
        # ⭐ WAVE-27 item 13 — the code language, for the same reason: a `code` column created
        # through THIS door would otherwise arrive without the bag `_clean_field` preserves, and
        # a column whose language depends on which door made it is the divergence the note below
        # warns about, arriving through a different key.
        if ftype == 'code' and f.get('code') is not None:
            import aios_grid as _agc
            c = _agc._clean_code(f.get('code'))
            if c:
                entry['code'] = c
        # C3 (wave 25): the profile flag rides the CREATE path too, and the one-per-table law
        # holds inside a single list. ⚠ This validator and `_clean_field` are two doors onto one
        # contract (a pre-existing split — `create`/`set_fields` come through here, `add_field`/
        # `patch_field` through the other), so a rule taught to only one of them is a rule the
        # other silently permits: a table could be CREATED with two profile columns and then
        # refuse to accept a third, which is the shape a user reads as random.
        if f.get('profile') is not None:
            p = _clean_profile(f.get('profile'))
            if p is None or ftype != 'text' or _profile_of(out):
                continue                   # refused: dropped, never stored unvalidated
            entry['profile'] = p
        # ⭐ 2026-08-09 — THE RELATIONAL PAIR RIDES THIS DOOR TOO, and it is the same divergence
        # `pinned` and `code` were fixed for above, arriving through a third key. `_clean_field`
        # has carried `link`/`rollup` since 2026-08-07; THIS validator dropped both, so a `link`
        # or `rollup` column created through `create()` / `set_fields()` arrived with its bag
        # stripped — a column that renders, is typed `rollup`, and computes nothing forever.
        # ⚠ IT WAS LATENT, NOT LIVE (no API route calls `set_fields`, and `odoo_relational` writes
        # its definitions directly), and it stops being latent the moment a user can BUILD a
        # rollup from the field editor — which is exactly what this wave ships. Caught by
        # `verify_odoo_relational`'s own NOTE line, which had been printing the list for two days.
        # ⛔ THE TYPE/BAG PAIRING IS ENFORCED BOTH WAYS, as it is in `_clean_field`: a `link` with
        # no bag can never resolve a row, and a bag on a column of another type is a declaration
        # nothing reads. Either half alone is the silent kind of broken.
        for bag_key, cleaner in (('link', _clean_link), ('rollup', _clean_rollup)):
            bag = cleaner(f.get(bag_key)) if f.get(bag_key) is not None else None
            if ftype == bag_key and bag is None:
                entry = None
                break
            if bag is not None:
                if ftype != bag_key:
                    entry = None
                    break
                entry[bag_key] = bag
        # ⭐⭐ 2026-08-10 — the FORMULA expression rides this door too, for the fourth time in the
        # same seam (`pinned`, `code`, `link`/`rollup`, now this). A formula column created through
        # `create()`/`set_fields()` with its expression dropped is a permanently blank column that
        # renders, sorts and filters — the exact silent shape the paragraph above describes.
        if entry is not None:
            fx = (_ag_formula(f.get('formula')) if f.get('formula') is not None else None)
            if (ftype == 'formula') != (fx is not None):
                entry = None
            elif fx is not None:
                entry['formula'] = fx
        if entry is None:
            seen.discard(key)
            continue
        # `agg` decides HOW the grid's totals row summarises the column. A rollup that declares it
        # through one door and not the other summarises differently depending on who made it.
        # ⭐ WAVE-29 C7: the vocabulary is `FIELD_AGGS`, not the single word `sum` — see its note.
        agg_in = str(f.get('agg') or '').strip()
        if agg_in in FIELD_AGGS:
            entry['agg'] = agg_in
        out.append(entry)
    return out or None


def all_tables(st=None):
    """{key: definition} — every user-created table in this tenant.

    ⚠ THIS READS THE ROWS TOO, and on tenant #0 that is 28.6 MB / ~703 ms per call. Callers that
    only need DEFINITIONS (labels, `createdBy`, `recordMode`, `fields`) want `all_defs` below.
    """
    try:
        return dict(_st(st).get(STORE_KEY) or {})
    except Exception:
        return {}


#: The keys `all_defs` refuses to read. `rows` is 99.9% of tenant #0's document by bytes and is
#: touched by no permission check, no nav render and no workspace envelope (D-185, W32-T01).
DEF_ONLY_DROPS = ('rows',)


def all_defs(st=None):
    """{key: definition WITHOUT `rows`} — the same listing, for readers that never open a row.

    ⭐⭐ W32-T01 (D-185). `all_tables` is the whole 28.6 MB deep copy; this is the same answer for
    every definition key at ~0.1% of the bytes. `/nav` is the caller that made it worth building:
    `nav_entries`, `may_open` and the manage/canDelete/locked loop read `label`, `source`,
    `createdBy` and `recordMode` — and nothing else.

    ⛔ **A DROPPED KEY RAISES, IT DOES NOT READ AS EMPTY.** `defn.get('rows')` on one of these
    raises `KeyError` naming the projection, because the alternative — `{}` — is indistinguishable
    from a database that genuinely has no rows, and the caller least able to notice is the one that
    wanted rows. If you get that error you wanted `all_tables`.

    ⚠ Falls back to `all_tables` on ANY failure, including a store handle too old to project (a
    fake store installed by a gate, a `_Lent` handle serving a document somebody else read). The
    fallback is CORRECT-but-slow by construction — it can only ever return MORE than was asked for
    — which is the one direction a fallback here is allowed to be wrong in.
    """
    try:
        return dict(_st(st).get_projection(STORE_KEY, drop=DEF_ONLY_DROPS) or {})
    except AttributeError:
        return all_tables(st)
    except Exception:
        return {}


def get(table_key, st=None):
    return all_tables(st).get(str(table_key))


def is_user_table(table_key, st=None):
    """THE permission predicate for row writes. Cheap and total: a key that is not in this store
    is not a user table, so it cannot accept invented rows."""
    return str(table_key or '').startswith(KEY_PREFIX) and get(table_key, st) is not None


def create(label, username, fields=None, source=None, st=None):
    """Create a blank table. Returns its key, or None if refused.

    Refuses rather than raises on the ordinary conditions (no name, duplicate, cap reached) —
    the caller is a UI form and a refusal is a message, not an exception.
    """
    label = str(label or '').strip()[:MAX_LABEL]
    if not label:
        return None
    existing = all_tables(st)
    if len(existing) >= MAX_TABLES:
        return None
    key = KEY_PREFIX + _slug(label)
    if key in existing:                       # a second table of the same name gets a suffix
        for n in range(2, 50):
            if f'{key}_{n}' not in existing:
                key = f'{key}_{n}'
                break
        else:
            return None
    # A blank table still needs ONE column to be a table at all: a row with no fields cannot be
    # displayed, edited or identified. 'name' is the identity column, the same role `customer`
    # plays in the Customer table.
    defn = {
        'key': key,
        'label': label,
        'source': (source if source in (BLANK_SOURCE, AUTOMATION_SOURCE) else BLANK_SOURCE),
        'createdBy': username,
        'created': _dt.datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
        'fields': clean_fields(fields) or [{'key': 'name', 'label': 'Name', 'type': 'text',
                                            'source': 'overlay', 'default': True}],
        'rows': {},
    }

    def _add(cur):
        cur[key] = defn
        return cur

    _st(st).update(STORE_KEY, _add, flush='sync')   # a creation is not a hot path; commit it
    return key


def delete(table_key, st=None):
    """Delete a user table AND its artifact families (wave 21, item 6a / C3).

    Wave 20 popped the definition alone and booked ten orphan families; the owner then ruled a
    real Delete (R3) with the footprint disclosed in the confirm dialog first, so the cleanup
    is no longer optional. Cleaned here, in one place: definition (`user_tables`) · the
    workspace bucket (`<key>_table_workspace`, every stratum incl. `__shared__` — views,
    custom fields, overlays, folders) · cohorts (`<key>_cohorts`) · record comments
    (`<key>_record_comments`) · docs metadata (`<key>_docs`) AND the dataset BYTES each
    metadata row names (D-36, closed wave 23 — wave 21 left those unreachable-but-present) ·
    `nav_meta[<key>]` · share grants (the database grant plus a view grant per view that
    lived in this bucket, via `shares.drop_objects`) · alert definitions on this topic.
    Bound AUTOMATIONS are the API layer's to disable (platform must not import the engine).
    Every family after the definition is best-effort: a store blip mid-sweep must not resurrect
    the table, and an orphaned empty bucket is residue, not a leak."""
    key = str(table_key)
    s = _st(st)

    def _drop(cur):
        cur.pop(key, None)
        return cur
    s.update(STORE_KEY, _drop, flush='sync')

    ws_key = f'{key}_table_workspace'
    view_ids = set()
    try:
        bucket = s.get(ws_key) or {}
        for _u, ws in bucket.items():
            if isinstance(ws, dict):
                view_ids |= set((ws.get('views') or {}).keys())
    except Exception:
        pass
    # ⭐ D-36 CLOSED (wave 23) — the document BYTES, not just their metadata. Wave 21 cleaned
    # `<key>_docs` and disclosed that the blobs under `docs/<key>/…` were left on the dataset;
    # "unreachable" is not "deleted", and for a tenant asking us to delete a database the
    # difference is the whole promise. Read the metadata FIRST (it carries each blob's exact
    # stored `path`, so nothing is guessed from a naming convention that could drift), then
    # clear the bucket, then delete the blobs. Best-effort per file: a storage blip must not
    # resurrect the table or abort the rest of the sweep — an undeleted blob is residue we can
    # sweep again, an aborted delete is a table the user asked to be gone.
    doc_paths = []
    try:
        for _pid, _rows in (s.get(f'{key}_docs') or {}).items():
            for _doc in (_rows or []):
                p = str((_doc or {}).get('path') or '')
                if p:
                    doc_paths.append(p)
    except Exception:
        pass
    # ⚠ `<key>__shared` JOINED THIS LIST IN WAVE 30 AND ITS ABSENCE WAS A REAL LEAK. The
    # tenant-wide overlay (W30-T28) is a bucket like the four beside it, and the docstring above
    # promises the artifact FAMILIES go with the database. Left out, a deleted database's shared
    # columns and their values survived in the store — invisible, and resurrected wholesale if
    # anybody ever re-created a database under the same key, which is exactly the reuse
    # `drop_field` scrubs cells to prevent one column at a time.
    try:
        import core.shared_overlay as _so
        shared_key = _so.bucket(key)
    except Exception:                                             # noqa: BLE001
        shared_key = f'{key}__shared'
    for b in (ws_key, f'{key}_cohorts', f'{key}_record_comments', f'{key}_docs', shared_key):
        try:
            s.update(b, lambda cur: {}, flush='async')
        except Exception:
            pass
    for p in doc_paths:
        try:
            import core.store as _store_bytes
            _store_bytes.delete_path(p)
        except Exception:
            pass

    def _drop_meta(cur):
        if isinstance(cur, dict):
            cur.pop(key, None)
        return cur
    try:
        s.update('nav_meta', _drop_meta, flush='async')
    except Exception:
        pass
    try:
        import core.shares as shares
        shares.drop_objects([('database', key)] + [('view', v) for v in view_ids], st=s)
    except Exception:
        pass
    try:
        import core.alerts as alerts
        alerts.drop_topic(key, st=s)
    except Exception:
        pass


def set_fields(table_key, fields, st=None):
    """Replace a table's base-field contract (wave 18 — the automation engine grows tables it
    creates). Cleaned with the same validator as `create`; refuses to leave a table fieldless."""
    clean = clean_fields(fields)
    if not clean or not is_user_table(table_key, st):
        return False

    def _set(cur):
        t = cur.get(str(table_key))
        if t is not None:
            t['fields'] = clean
        return cur

    _st(st).update(STORE_KEY, _set, flush='sync')
    return True


def add_row(table_key, values=None, username=None, st=None, rid=None):
    """Append one row. Returns the new row id, or None if refused.

    `rid` RESTORES A ROW UNDER ITS OLD ID (contract C-ADDROW / C-UNDO). Undo has to put a deleted
    row back where it was: a restore under a fresh id would break every cohort, comment and view
    filter that named the original, so the id is part of what is being undone. It is honoured
    only when that id is FREE — an undo can never overwrite a row somebody has since created in
    the gap, and it never invents a non-numeric id, because `scoped_pool` reads row ids as ints.
    """
    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
        return None                            # ⚠ never on a connector-backed table
    defn = get(table_key, st) or {}
    rows = defn.get('rows') or {}
    # R6: `row_limit` is None for a connected source. `add_row` cannot reach one anyway — the
    # `records_mutable` guard above refuses first — but reading the ceiling through the ONE
    # evaluator is what stops a later relaxation of that guard quietly re-imposing this cap.
    cap = row_limit(table_key, st)
    if cap is not None and len(rows) >= cap:
        return None
    want = str(rid or '').strip()
    if want and want.isdigit() and want not in rows:
        rid = want
    else:
        rid = str(max((int(r) for r in rows if str(r).isdigit()), default=0) + 1)
    valid = {f['key'] for f in (defn.get('fields') or [])}
    clean = {k: str(v) for k, v in (values or {}).items() if k in valid}
    # C3 (wave 25): a PROFILE cell is validated and normalised on the insert door too, not only
    # on the edit door. A row born with `https://instagram.com/p/Cxyz` in its handle column is a
    # row the enrich action can never answer, and the failure would otherwise surface as an
    # automation that quietly returns nothing rather than as a refused write.
    pf = _profile_of(defn.get('fields') or [])
    if pf and pf['key'] in clean:
        handle, ok = normalize_profile(clean[pf['key']], pf['profile'].get('source'))
        if not ok:
            return None
        clean[pf['key']] = handle

    def _add(cur):
        t = cur.get(str(table_key))
        if t is not None:
            t.setdefault('rows', {})[rid] = clean
        return cur

    _st(st).update(STORE_KEY, _add, flush='async')
    # C3 (wave 22): a HUMAN-door insert is a row event — the record_created trigger's whole
    # substrate. The engine's own writers never come through here, which is the loop law.
    emit_row_event({'type': 'record_created', 'table': str(table_key), 'rowId': rid,
                    'st': _st(st), 'user': str(username or '')})
    return rid


def add_rows(table_key, rows_in, username=None, st=None):
    """⭐⭐ WAVE-29 T25 (owner item 6) — APPEND MANY ROWS IN ONE STORE WRITE. Returns the list of
    new row ids, or None if the table refuses records at all.

    ⛔ WHY THIS EXISTS RATHER THAN A LOOP OVER `add_row`. Every `add_row` is a read-modify-write of
    the WHOLE table document plus a row event; importing a 2,000-row spreadsheet that way is 2,000
    full-document copies under one lock, which is the same shape as the 1.4 s-per-bucket problem
    item 20 exists to avoid — on the one uvicorn process this product runs. One `update()` writes
    them all.

    ⛔ AND WHY NOT `automation_engine.upsert_rows`, which this module's own note (below) says to
    reuse: that function is UPSERT-BY-KEY shaped, and v1 of the import door is deliberately
    APPEND-ONLY (no key to merge on, no silent overwrite of a row somebody edited). It also lives
    in the API layer, and `core/` may not import upward. When the import door grows an "update
    matching rows" mode, the engine's is the one to lift — the note stands, it just is not this.

    ⚠ THE CAP IS CHECKED AGAINST THE WHOLE BATCH, not per row: a partial import that stops at
    `MAX_ROWS` leaves the user reconciling a spreadsheet against a table, which is exactly what
    the refusal-with-a-sentence exists to prevent. Nothing is written if the batch does not fit.

    ⚠ Ids continue the table's own sequence, and a PROFILE column validates every cell exactly as
    `add_row` does — a row born with an unusable handle is a row the enrich action can never
    answer, and the failure would otherwise surface as an automation quietly returning nothing.
    """
    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
        return None
    batch = [dict(r) for r in (rows_in or []) if isinstance(r, dict)]
    if not batch:
        return []
    defn = get(table_key, st) or {}
    rows = defn.get('rows') or {}
    cap = row_limit(table_key, st)        # R6: None for a connected source
    if cap is not None and len(rows) + len(batch) > cap:
        return None
    valid = {f['key'] for f in (defn.get('fields') or [])}
    pf = _profile_of(defn.get('fields') or [])
    next_id = max((int(r) for r in rows if str(r).isdigit()), default=0)
    made = []
    for values in batch:
        clean = {k: str(v) for k, v in values.items() if k in valid}
        if pf and pf['key'] in clean:
            handle, ok = normalize_profile(clean[pf['key']], pf['profile'].get('source'))
            if not ok:
                return None                    # atomic: one bad handle refuses the FILE
            clean[pf['key']] = handle
        next_id += 1
        made.append((str(next_id), clean))

    def _add(cur):
        t = cur.get(str(table_key))
        if t is not None:
            bucket = t.setdefault('rows', {})
            for rid, clean in made:
                bucket[rid] = clean
        return cur

    _st(st).update(STORE_KEY, _add, flush='sync')
    # ⚠ ONE EVENT PER ROW, deliberately: `record_created` is a trigger substrate, and an import of
    # 500 rows genuinely IS 500 created records. The engine's own flood hold is what decides
    # whether that runs an automation 500 times — a decision that belongs there, not to a writer
    # that quietly emits less than it did.
    for rid, _clean in made:
        emit_row_event({'type': 'record_created', 'table': str(table_key), 'rowId': rid,
                        'st': _st(st), 'user': str(username or '')})
    return [rid for rid, _ in made]


# ⛔ `upsert_rows` USED TO LIVE HERE, AND DELETING IT IS THE FIX (DEBT D-6, wave 20).
#
# There were TWO bulk-upsert implementations — this one and `automation_engine.upsert_rows` —
# with different counts (`{updated, inserted, orphans}` vs the engine's seven, including the
# `capped` count that makes a full table LOUD), a different signature (a dict keyed by value vs
# a list of rows) and a different cap story. D-6 called it a "silent drift risk"; the honest
# measurement is worse and simpler: **this one had zero callers, anywhere in the repo.** It was
# not drifting from the engine, it was a second answer nobody had ever asked.
#
# So there is now one implementation because there is one implementation — no parity gate to
# maintain, no second definition to keep in step. If a host-side bulk upsert is ever wanted, the
# engine's is the one that has been exercised (verify_automation section B) and it is PURE:
# `(existing, incoming, key_field, cap) -> (rows, counts)`, so it can be lifted here without its
# store half coming along. Do not re-derive a new one.


def set_label(table_key, label, st=None):
    """Rename a table IN ITS DEFINITION. Returns the stored label, or None if refused.

    ⚠ WAVE 20, item 6a — THE RENAME USED TO WRITE ONLY `nav_meta`. That bucket is the nav's
    display layer, so the rail showed the new name while everything reading the DEFINITION —
    the automation editor's database picker above all — went on showing the old one. A rename
    that only some surfaces can see is worse than no rename: the picker was not stale-looking,
    it was confidently wrong, and a user choosing "Influencers" there could be choosing the
    table they had renamed to something else months earlier.
    """
    label = ' '.join(str(label or '').split())[:MAX_LABEL]
    if not label or not is_user_table(table_key, st):
        return None

    def _set(cur):
        t = cur.get(str(table_key))
        if t is not None:
            t['label'] = label
        return cur

    _st(st).update(STORE_KEY, _set, flush='sync')
    return label


# ---------------------------------------------------------------------------------------------
# THE SHARED FIELD SCHEMA (contract C-FIELD, owner ruling R2)
# ---------------------------------------------------------------------------------------------
# R2 reverses wave 17's "fields are per-user" law FOR THIS PATH: a `ut_*` table's fields are the
# TABLE'S SCHEMA, the way they are in Airtable — everyone with access to the database sees the
# same columns, and the creator or an admin edits them. Per-field `editRole` narrows or widens
# who may edit ONE column's definition without handing over the whole table.
#
# ⚠ `editRole` GOVERNS THE SCHEMA, NEVER THE VALUES. 'everyone' means anybody with access may
# rename this column or change its options; it does not decide who may type in its cells. Those
# are different questions and conflating them would let a column's own settings quietly become a
# data-permission system nobody wrote.

#: Who may edit ONE field's definition. Fail-closed default: admins (= creator or admin).
FIELD_EDIT_ROLES = ('admins', 'everyone')

#: Wave 22 (C7) — the metric-field vocabulary, MIRRORED from `automation_engine` as local
#: literals for the boot-path reason `MACHINE_OWNERS` states; the pair is held in step by a
#: verify_automation parity check, the same discipline.
METRIC_MEASURES = ('followers', 'avg_engagement', 'likes', 'comments')
METRIC_WINDOWS = ('latest', 'last_3_posts', 'last_7d', 'last_30d')
METRIC_PROFILE_MEASURES = ('followers', 'avg_engagement')
METRIC_AGGS = ('avg', 'sum', 'latest')


def _clean_metric(raw):
    """One `metric` bag → the stored shape, or None (refused). A profile count over a
    post-count window — or a SUM of a follower count — is a question the series cannot
    answer; refused at write, never bent into a number that looks plausible (C7)."""
    if not isinstance(raw, dict):
        return None
    measure = str(raw.get('measure') or '').strip()
    window = str(raw.get('window') or '').strip()
    agg = str(raw.get('agg') or '').strip()
    if measure not in METRIC_MEASURES or window not in METRIC_WINDOWS:
        return None
    if measure in METRIC_PROFILE_MEASURES and (window == 'last_3_posts' or agg == 'sum'):
        return None
    if agg and agg not in METRIC_AGGS:
        return None
    out = {'source': 'ig', 'measure': measure, 'window': window}
    if agg:
        out['agg'] = agg
    return out


# ---------------------------------------------------------------------------------------------
# ⭐⭐ THE RELATIONAL PAIR (2026-08-07, owner instruction) — `link` and `rollup`
# ---------------------------------------------------------------------------------------------
# Owner: *"adding relational database function with links and rollups… exactly like how Airtable
# does it… and this rollup needs to have formula that we can use to calculate things like average
# Views over last N posts."*
#
# Airtable's own contract, fetched 2026-08-07 (SPEC):
#   multipleRecordLinks.options : linkedTableId · isReversed · prefersSingleRecordLink ·
#                                 inverseLinkFieldId · viewIdForRecordSelection
#   rollup.options              : recordLinkFieldId · fieldIdInLinkedTable · result ·
#                                 referencedFieldIds · isValid
#   17 functions, and a rollup is exactly ONE HOP.
#
# ⛔ THE CELL IS A SCALAR, WHICH IS NOT A PREFERENCE. Airtable's link cell is an ARRAY of record
# ids; the `Row` contract here is scalars end to end and `grid_events` refuses a non-scalar write.
# So a link cell holds a COMMA-JOINED list of linked row ids — the shape `multiselect` already
# uses for a set, so the grouping/filter/copy paths already know what to do with it.
#
# ⭐ THREE MODES, ONE KIND:
#   `on` DECLARED  -> a DERIVED link. The linked rows are those whose `on` column equals this
#                    row's `from` column. Machine-maintained, read-only, recomputed on the same
#                    pass that already recomputes `metric` cells. THIS IS THE INSTAGRAM CASE, and
#                    it is why the mode exists: `ut_ig_posts.influencer_key` ALREADY holds the
#                    relation and is rewritten by the engine on every pull, so a second
#                    user-editable copy of the same relation could only ever drift away from it.
#   `on` ABSENT    -> an ORDINARY link. The user picks records; the cell is editable; `single` is
#                    Airtable's `prefersSingleRecordLink`. Its value is the source of truth.
#   `inverse`       -> the COMPUTED reciprocal of one ordinary source link. It never fans writes
#                    into target rows: the relation pass derives it from the source cells, so the
#                    same shared-row truth powers both sides and every dependent Rollup.

#: How many linked ids one link cell may carry. Row ids are short numeric strings, so 500 ids is
#: ~3 KB — comfortably inside a text cell, and `MAX_ROWS` bounds the absolute worst case anyway.
LINK_MAX_IDS = 500

#: Airtable's 17 aggregations, minus the two that are meaningless over scalar cells.
#: ⛔ `arrayflatten` and `arrayslice` are NOT here: both operate on NESTED arrays, and a cell in
#: this product is a scalar — they could only ever return their own input, so offering them would
#: mint columns that compute nothing. `arrayslice`'s real use ("just the first N") is what `limit`
#: does below, honestly and by declared rank.
#: ⭐ `stdev` is the ONE addition beyond Airtable's list (wave 28, owner ruling R4), and it is here
#: for a request no combination of the other fourteen can express: *"the average of the last 10
#: posts, with anything beyond 2 standard deviations removed"*. `average` alone cannot say which
#: rows are outliers, and a client that computed the threshold itself would be a second definition
#: of the same statistic over a row set only the server can see (the fold receives the UNCAPPED
#: linked set; a link CELL shows at most `LINK_MAX_IDS`).
#: ⚠ SAMPLE standard deviation (n-1) — see `_rollup_fold`. The population form (n) is the wrong
#: default here: a rollup folds the rows that happen to be linked, which is a sample of an
#: account's posting history, not its entirety.
ROLLUP_FNS = ('sum', 'average', 'stdev', 'min', 'max', 'latest', 'count', 'counta', 'countall',
              'and', 'or', 'xor', 'concatenate', 'arrayjoin', 'arraycompact', 'arrayunique')
ROLLUP_SORT_DIRS = ('asc', 'desc')
ROLLUP_CONDITION_OPS = ('eq', 'neq', 'contains', 'not_contains', 'is_empty', 'is_not_empty',
                        'gt', 'gte', 'lt', 'lte')
ROLLUP_CONDITION_CONJ = ('and', 'or')
ROLLUP_MAX_CONDITIONS = 20

#: ⭐⭐ THE SET-STATISTIC THRESHOLD (wave 28, R4 — contract C1). A condition may compare a field
#: against a LITERAL (`value`) or against a statistic OF THE SCOPED SET ITSELF
#: (`ref: {sigmas: k}` → mean + k*stdev of that same field). It is what makes "beyond 2 sigma"
#: expressible as configuration instead of code, and it composes with everything already here:
#: the scope (`sortBy` + `limit`) picks the window, THEN the threshold is computed over that
#: window, THEN the conditions filter it.
#:
#: ⛔ ORDER IS THE WHOLE CONTRACT AND IT IS NOT NEGOTIABLE: the statistic is computed over the
#: SCOPED set (after ranking and `limit`, before filtering). Computing it over the unfiltered
#: table would answer "2 sigma of everything this account ever posted" while the column says "of
#: the last 10", and both readings produce a plausible number — which is the failure this module
#: refuses everywhere else.
#:
#: ⛔ ORDERING OPS ONLY. `eq`/`neq` against a computed float is a coin flip on floating-point
#: representation, and `contains` against a number is meaningless — offering either would mint a
#: condition that silently never matches.
ROLLUP_REF_OPS = ('gt', 'gte', 'lt', 'lte')
#: A sanity bound, not a statistical one. Beyond ±10 sigma nothing is being selected on any real
#: distribution, so a value out here is a typo or a unit mistake, and refusing it is kinder than
#: storing a filter that can only ever match nothing. ⭐ It also does the NaN/infinity work for
#: free: `not (-10 <= NaN <= 10)` is True, and infinity fails the same comparison.
ROLLUP_MAX_SIGMAS = 10.0

#: The date-window KINDS a source-backed rollup may name. MIRRORS `harness/windows.py`'s
#: `WINDOW_KINDS`, deliberately as a local literal — `core` stays dependency-light and does not
#: import up, the same reason `UT_FIELD_TYPES` is a literal rather than an `aios_grid` import.
#: ⚠ This list VALIDATES; it never RESOLVES. `harness.windows.resolve(spec, today)` owns turning a
#: kind into a date pair, so there is exactly one implementation of what "ytd" means and this
#: module cannot drift into a second one. A kind added there and not here is simply not offerable
#: on a rollup yet — the fail-closed direction.
ROLLUP_SOURCE_WINDOWS = (
    'all_time', 'today', 'yesterday', 'this_week', 'last_week', 'this_month', 'last_month',
    'this_quarter', 'last_quarter', 'this_year', 'last_year', 'ytd', 'ytd_last_year',
    'ltm', 'past_week', 'past_month', 'past_year',
)
#: ⛔ DELIBERATELY ABSENT: `last_n_days`, `next_n_days`, `custom`. Each needs a PARAMETER (an `n`,
#: or a date pair) that this bag has nowhere to carry, and a parameterised kind accepted without
#: its parameter would resolve to something arbitrary — a window that looks configured and
#: measures the wrong period. They land when the bag learns to carry the parameter, not before.
#: The ceiling on `limit`. Not a performance bound — `MAX_ROWS` is that — but a refusal to let a
#: column claim a window bigger than a table can hold. It therefore MOVES WITH the cap, which is
#: why it is written as `MAX_ROWS` and not as a number.
#: ⚠ THIS COMMENT SAID "100,000" UNTIL 2026-08-09 (wave 28) while sitting eight lines above
#: `ROLLUP_MAX_LIMIT = MAX_ROWS`, where `MAX_ROWS` is 60,000 — a doc contradicting the constant it
#: annotates, in the same screenful. Two sibling comments said it too (`odoo_relational`,
#: `rollup_sql`). 100,000 was a candidate cap REJECTED during its own derivation for clearing the
#: memory budget by 0.6%; the prose was written before the number lost.
#: ⛔ IT IS NOT `LINK_MAX_IDS`, and the near-miss is worth keeping. 500 looks like the real bound —
#: a rollup folds what a link names, and a link cell holds at most 500 ids. But that cap is a
#: DISPLAY cap: `compute_relation_cells` projects the first 500 ids into the cell and hands every
#: rollup the UNCAPPED `resolved` set, so a customer with 3,000 invoices shows 500 and still sums
#: all 3,000. Tightening this to 500 would have silently truncated the fold's declared window on
#: exactly the tables this wave exists to build ([[measure-the-real-call]]).
ROLLUP_MAX_LIMIT = MAX_ROWS


def _field_key(raw):
    """A field/table key as this module spells them, or '' — the same normalisation
    `_clean_field` applies to `key`, so a bag can never name a key the field layer would store
    differently."""
    k = re.sub(r'[^a-z0-9_]+', '_', str(raw or '').strip().lower()).strip('_')[:60]
    return k


def _clean_link(raw):
    """One `link` bag → the stored shape, or None (refused).

    Same posture as `_clean_metric`: a mis-shaped bag REFUSES THE FIELD. A link column stored
    without a usable target is a column promising a relation that nothing can ever resolve.
    """
    if not isinstance(raw, dict):
        return None
    table = _field_key(raw.get('table'))
    # ⛔ It must NAME A USER TABLE. A link into a connector-backed module key (`customer`,
    # `product`) would promise a join across two strata with different row-id namespaces and
    # different permission walls — a different feature, not a smaller version of this one.
    if not table.startswith(KEY_PREFIX) or len(table) <= len(KEY_PREFIX):
        return None
    out = {'table': table}
    on = _field_key(raw.get('on'))
    frm = _field_key(raw.get('from'))
    inverse = _field_key(raw.get('inverse'))
    reciprocal = _field_key(raw.get('reciprocal'))
    if inverse:
        # Airtable's reciprocal field: rows in `table` whose ordinary `inverse` link includes
        # this row. It is computed, so `on`/`from` cannot simultaneously configure another join.
        if on or frm:
            return None
        out['inverse'] = inverse
    elif on:
        out['on'] = on
        # `from` is OPTIONAL and resolved at compute time (the profile-flagged column, then the
        # pinned one) — which is what makes an Instagram database link up with no configuration
        # at all, i.e. the "automatically" in the instruction.
        if frm:
            out['from'] = frm
    elif frm:
        # ⛔ A `from` with no `on` configures NOTHING — there is no join to drive. Refused rather
        # than dropped: a bag half of which is silently ignored is [[wrong-parent-not-broken-control]]
        # with the control still on screen.
        return None
    if reciprocal:
        out['reciprocal'] = reciprocal
    if raw.get('single') is True:
        out['single'] = True
    return out


def _clean_rollup(raw):
    """One `rollup` bag → the stored shape, or None (refused).

    ⭐ `limit` + `sortBy` ARE THE DELIBERATE SUPERSET OF AIRTABLE, and they are the owner's
    headline: *"average Views over last N posts"*. Airtable's rollup conditions filter by
    PREDICATE, never by RANK, so "the last 12" is not expressible there at all.

    ⛔ `limit` REQUIRES `sortBy`, and that refusal is the whole care in this function. "The last
    N" with no declared order is not a measurement — it is whichever N rows happen to sit first
    in the store, which for an append table is capture order and for a merged one is nothing at
    all. A wrong number that looks right is the failure this module refuses everywhere else.
    """
    if not isinstance(raw, dict):
        return None
    # ⭐ A SOURCE-BACKED ROLLUP — the read-through kind (owner 2026-08-09: *"make rollup be able to
    # capture up to 1 million rows of a linked database … ok to push rollup filter etc to SQL"*).
    #
    # ⛔ WHY IT IS A DIFFERENT SHAPE RATHER THAN A BIGGER `link`. A linked rollup folds ROWS THAT
    # EXIST IN THE STORE, and `MAX_ROWS` bounds them because every `ut_*` table lives in one JSON
    # blob that is parsed and copied per request. ⚠ RAISING THE CAP TO 60,000 DOES NOT RETIRE THIS
    # KIND, and the arithmetic is why: 256,810 order lines are 63.9 MB and 2.57 s PER COPY, so they
    # cannot live there under any JSON substrate — 4.3x the cap and 8x a tolerable copy. This kind
    # never COPIES the rows at all: it names a governed semantic TOPIC and a METRIC KEY, and one
    # grouped SQL query answers every parent row at once (MEASURED: 1,748 customers over 256,810
    # lines in 232 ms).
    #
    # ⚠ A METRIC KEY, NEVER A FILTER FRAGMENT, and that is the load-bearing decision. `model/
    # metrics/*.yml` already carries each metric's scope, its `store_filter_sql` AND the matching
    # `live_domain` — whose own comment says *"BOTH or store_parity compares two different
    # questions"*. Binding to the key inherits the scope and the live-parity oracle; letting a
    # rollup carry SQL would mint a second definition of a number the semantic layer exists to
    # define once.
    #
    # ⚠ `window` is a date-window KIND (`ytd`, `this_month`, …) resolved against `today` at
    # COMPUTE time, never a literal date pair — a hand-computed year start is correct until
    # 1 January and wrong after it, with nothing to notice ([[date-window-vocabulary]]).
    src = raw.get('source')
    if isinstance(src, dict):
        topic = _field_key(src.get('topic'))
        measure = _field_key(src.get('measure'))
        group_by = _field_key(src.get('groupBy'))
        on = _field_key(src.get('on'))
        if not (topic and measure and group_by and on):
            return None
        bag = {'topic': topic, 'measure': measure, 'groupBy': group_by, 'on': on}
        window = str(src.get('window') or '').strip().lower()
        if window:
            if window not in ROLLUP_SOURCE_WINDOWS:
                return None
            bag['window'] = window
        return {'source': bag}
    link = _field_key(raw.get('link'))
    field = _field_key(raw.get('field'))
    fn = str(raw.get('fn') or '').strip().lower()
    if not link or fn not in ROLLUP_FNS:
        return None
    # `countall` counts LINKED RECORDS, so it alone needs no field to read (Airtable's rule).
    if not field and fn != 'countall':
        return None
    out = {'link': link, 'fn': fn}
    if field:
        out['field'] = field
    try:
        limit = int(str(raw.get('limit') or '0').strip() or 0)
    except (TypeError, ValueError):
        return None
    sort_by = _field_key(raw.get('sortBy'))
    distinct_by = _field_key(raw.get('distinctBy'))
    sort_dir = str(raw.get('sortDir') or '').strip().lower()
    if limit < 0 or limit > ROLLUP_MAX_LIMIT:
        return None
    if (limit or fn == 'latest') and not sort_by:
        return None
    if sort_dir and sort_dir not in ROLLUP_SORT_DIRS:
        return None
    if sort_by:
        out['sortBy'] = sort_by
        # Newest-first is the default because every window this feature exists for is "the LAST
        # N", and a default that silently means "the first N" would answer a different question
        # with the same column name.
        out['sortDir'] = sort_dir or 'desc'
        if limit:
            out['limit'] = limit
    elif sort_dir:
        return None
    if distinct_by:
        # A rollup may cross a table that already contains duplicate logical records. Keep the
        # first row after ranking for each identity so "last 12 posts" means twelve POSTS, while
        # a snapshot table can still keep every timestamped observation by omitting this option.
        out['distinctBy'] = distinct_by
    # ⭐⭐ 2026-08-09 (owner) — THE PRE-FILTER, and it is a SECOND list rather than a flag on the
    # first. Owner: *"instead of last 12 posts, we also want to make it so its last N record,
    # where the record's Status is video."*
    #
    # ⛔ WHY TWO LISTS AND NOT ONE. `conditions` is applied AFTER the window (C1 fixed that order
    # deliberately, so a `ref: {sigmas}` threshold is a statistic OF the rows being folded). That
    # makes "the last 10 REELS" inexpressible with it: the window would take the last 10 posts of
    # any kind and the filter would then keep whichever of those happened to be video — on this
    # tenant's live data, 12 captured posts per profile of which ~8 are video, so the column would
    # silently answer "the reels among the last 12" and wear the name "the last 10 reels".
    # `where` runs BEFORE the ranking, so the window is spent on rows that already qualify.
    #
    # ⛔ `ref` IS REFUSED HERE, and that refusal is what keeps C1 coherent. A sigma computed over
    # the pre-filtered-but-unwindowed set is a statistic of a DIFFERENT set from the one the fold
    # reports on — the exact incoherence C1's own note describes ("the set being described and the
    # set doing the describing would be different sets"). Fail closed: the threshold belongs in
    # the post-filter, where the window is already fixed.
    for bag_key, conj_key, allow_ref in (('where', 'whereConj', False),
                                         ('conditions', 'conditionConj', True)):
        conditions = []
        raw_conditions = raw.get(bag_key) or []
        if not isinstance(raw_conditions, list) or len(raw_conditions) > ROLLUP_MAX_CONDITIONS:
            return None
        for condition in raw_conditions:
            if not isinstance(condition, dict):
                return None
            condition_field = _field_key(condition.get('field'))
            op = str(condition.get('op') or '').strip().lower()
            if not condition_field or op not in ROLLUP_CONDITION_OPS:
                return None
            item = {'field': condition_field, 'op': op}
            if op not in ('is_empty', 'is_not_empty'):
                ref = condition.get('ref')
                if ref is not None:
                    # ⭐ THE SET-STATISTIC THRESHOLD (R4 / contract C1). `ref` replaces `value`;
                    # the number this leaf compares against is computed by the FOLD, over the
                    # scoped set, at compute time — so it moves with the data instead of freezing
                    # a threshold that was true the day somebody typed it.
                    # ⛔ NEVER BOTH. A leaf carrying `value` AND `ref` has two answers to one
                    # question and whichever the fold happened to read would be silent. Refuse
                    # the field.
                    if not allow_ref:
                        return None
                    if not isinstance(ref, dict) or condition.get('value') is not None:
                        return None
                    if op not in ROLLUP_REF_OPS:
                        return None
                    sigmas = ref.get('sigmas')
                    # `bool` is an `int` subclass, so `True` would otherwise validate as 1.0
                    # sigma — a checkbox arriving in a numeric slot should be refused, not
                    # interpreted.
                    if isinstance(sigmas, bool) or not isinstance(sigmas, (int, float)):
                        return None
                    sigmas = float(sigmas)
                    if not (-ROLLUP_MAX_SIGMAS <= sigmas <= ROLLUP_MAX_SIGMAS):
                        return None  # also catches NaN and +/-inf: every comparison is False
                    item['ref'] = {'sigmas': sigmas}
                else:
                    value = condition.get('value')
                    if not isinstance(value, (str, int, float, bool)):
                        return None
                    item['value'] = str(value)[:1000]
            conditions.append(item)
        if conditions:
            conj = str(raw.get(conj_key) or 'and').strip().lower()
            if conj not in ROLLUP_CONDITION_CONJ:
                return None
            out[bag_key] = conditions
            out[conj_key] = conj
    return out


def _clean_field(raw, previous=None):
    """One field dict → the stored shape, or None. The single validator for create AND patch, so
    a column cannot be typed one way on the way in and another on the way back."""
    prev = previous or {}
    raw = raw if isinstance(raw, dict) else {}
    label = ' '.join(str(raw.get('label') or prev.get('label') or '').split())[:80]
    key = re.sub(r'[^a-z0-9_]+', '_',
                 str(raw.get('key') or prev.get('key') or _slug(label)).strip().lower())
    key = key.strip('_')[:60]
    ftype = str(raw.get('type') or prev.get('type') or 'text').strip().lower()
    if not key or not label or ftype not in UT_FIELD_TYPES:
        return None
    out = {'key': key, 'label': label, 'type': ftype, 'source': 'overlay'}
    # ⭐ 2026-08-10 — the DISPLAY format, on the add/patch door too. ⚠ `'format' in raw` rather than
    # `raw.get('format')`, exactly like `description` below: an explicit `{}`/None CLEARS the
    # format, while omitting the key keeps what is stored. A PATCH that sends only a label must not
    # silently strip a formatting choice — that is how `agg` stopped totalling a money column.
    fmt = _clean_format(raw.get('format') if 'format' in raw else prev.get('format'))
    if fmt:
        out['format'] = fmt
    # Canonical descriptions are schema metadata, not the user's per-view note. Preserve them
    # through every field write so product-owned presets can explain themselves in the header
    # tooltip and schema drawer. An explicit blank clears a user-created field's description;
    # omission keeps the previous declaration.
    description_raw = (raw.get('description') if 'description' in raw
                       else prev.get('description'))
    description = ' '.join(str(description_raw or '').split())[:300]
    if description:
        out['description'] = description
    opts_raw = raw.get('options') if 'options' in raw else prev.get('options')
    if ftype in ('select', 'multiselect'):
        opts = [' '.join(str(o).split())[:60] for o in (opts_raw or []) if str(o).strip()][:50]
        if opts:
            out['options'] = opts
    # ⭐⭐ WAVE-29 CONTRACT C1 (owner item 13) — **THE OPTION COLOURS, WHICH THIS DOOR DROPPED.**
    #
    # The client has carried `optionColors`/`colorCodeOptions` since wave 14 (`types.ts`, the
    # `BRAND_SWATCHES` editor, `cells.optionTint`, the kanban card) and `aios_grid._clean_field`
    # keeps both — but THIS validator, the one every `ut_*` create and patch goes through, had
    # zero occurrences of either key. So the owner picked a colour on a user database, the pane
    # saved, and the colour was gone on the next read: the same crossed-validator shape as item 2a,
    # in a different pair of doors. `grid_events._FIELD_ECHO_PROPS` already carried both, so the
    # echo wire was plumbed for a value nothing stored.
    #
    # ⛔ SIBLING KEYS, and `options` IS NOT RESHAPED. Every consumer treats options as `string[]`
    # (this function flattens each to a bare `str` twelve lines up); folding colour into the option
    # entries would be a wire change across the grid, the kanban, the editor and the filter panel.
    #
    # ⛔ OMIT-WHEN-ABSENT, exactly like `format`/`metric`/`link`, because `verify_api.py` and
    # `verify_automation.py` assert `_clean_field(f) == f` over every engine-seeded field list —
    # a branch that emitted `optionColors: {}` would redden both gates for every preset column in
    # the product. ([[default-must-pass-its-own-guard]])
    #
    # ⚠ INHERITED ACROSS A PATCH, for `agg`'s reason: a PATCH that sends only a label must not
    # silently strip the colours somebody chose, which is precisely how a money column stopped
    # totalling.
    #
    # ⚠ A HEX IS VALIDATED; BRAND MEMBERSHIP IS NOT, and that is deliberate. `_clean_option_colors`
    # (`aios_grid.py:200`) — the sibling this mirrors — accepts any `#RRGGBB`, and the client
    # renders a stored colour through `choiceColors.brandedTint`, which maps it to its NEAREST
    # brand swatch at DISPLAY time while leaving what is stored byte-stable. Refusing a non-brand
    # hex here would refuse values the other validator accepts on the same UI, and would break a
    # column that moved between the two.
    if ftype in ('select', 'multiselect'):
        colors_raw = (raw.get('optionColors') if 'optionColors' in raw
                      else prev.get('optionColors'))
        colors = _clean_option_colors(colors_raw, out.get('options') or [])
        if colors:
            out['optionColors'] = colors
        color_code = (raw.get('colorCodeOptions') if 'colorCodeOptions' in raw
                      else prev.get('colorCodeOptions'))
        if isinstance(color_code, bool):
            out['colorCodeOptions'] = color_code
    # ⭐ WAVE-29 T22's own repair: a `rating` column's star count. Grid-created columns on a `ut_*`
    # database now come through this door rather than the per-user overlay, and this validator had
    # no `max` — so a 3-star or 10-star column silently became 5-star on the way in. Bounded rather
    # than refused, mirroring `aios_grid._clean_rating_max`: the field still holds its 1..max
    # integers whatever the number was, so an unusable value is a default and never a lost column.
    if ftype == 'rating':
        max_raw = raw.get('max') if 'max' in raw else prev.get('max')
        if max_raw is not None:
            try:
                out['max'] = max(2, min(int(max_raw), 10))
            except (TypeError, ValueError):
                pass
    role = str(raw.get('editRole') or prev.get('editRole') or 'admins').strip().lower()
    out['editRole'] = role if role in FIELD_EDIT_ROLES else 'admins'
    if prev.get('default') is True or raw.get('default') is True:
        out['default'] = True
    # ⭐ 2026-08-07 (owner ruling) — THE PRIMARY COLUMN, DECLARED RATHER THAN POSITIONAL.
    #
    # The grid's identity column is `fields.find(f => f.pinned)?.key ?? fields[0]?.key`
    # (`useGridColumns.reconcileOrder` / `permsModel.identityKey`), forced to position 0 and
    # force-shown. No `ut_*` field had ever declared `pinned`, so the fallback was the whole rule:
    # whichever column happened to be created FIRST became the locked primary forever. On a
    # database somebody made by hand and later pointed at an Instagram automation, that is the
    # `name` column `create()` mints — which nothing ever writes, so the table's identity column
    # was permanently blank while `handle` sat six columns to the right.
    #
    # ⚠ STICKY ACROSS A PATCH, like `default` one line up: the engine declares it (the C1 preset
    # set pins `handle`), and a client PATCH that simply omits the key must not silently unpin the
    # column it is editing. Clearing it is deliberate work, not a side effect of renaming a header.
    if prev.get('pinned') is True or raw.get('pinned') is True:
        out['pinned'] = True
    # ⭐ 2026-08-09 (wave 28) — `agg` DECIDES WHETHER THE GRID'S TOTALS ROW SUMS THE COLUMN, and
    # this door silently dropped it while `clean_fields` (:402) kept it. Two validators, one
    # question, opposite answers — D-91's exact class, found from the other end: the register's row
    # described `clean_fields` as the lax door, and by the time it was read that half had been
    # fixed while THIS one still diverged.
    # ⛔ THE CONSEQUENCE IS QUIET, WHICH IS WHY IT SURVIVED. Every currency and money-rollup column
    # in the four Odoo databases declares `agg: "sum"`; the spawn writes definitions straight to
    # the store, so the totals row worked — until anyone PATCHED one of those fields (a rename, a
    # width, a description), at which point the column silently stopped totalling and nothing
    # anywhere went red. Caught by a gate leg written to assert the LINK bag survives this door,
    # which found `agg` instead.
    # ⚠ Sticky across a patch for the same reason as `default` and `pinned`: a client PATCH that
    # omits the key must not un-total a column as a side effect of renaming its header.
    # ⭐ WAVE-29 C7 — the vocabulary widened from the single word `sum` to `FIELD_AGGS`. The
    # STICKINESS is unchanged and now applies to every name: a PATCH that omits `agg` keeps
    # whatever the column had, because dropping it is how a money column silently stopped
    # totalling. ⛔ An unknown name is DROPPED rather than stored — a summary nothing can compute
    # would render blank under a picker that says otherwise.
    # ⚠ `'agg' in raw`, not `raw.get('agg') or …` — the same idiom `format` and `description` use
    # two screens up, and it is what makes the difference between OMITTING the key (keep what is
    # stored: the wave-28 stickiness that stops a rename un-totalling a money column) and sending
    # it EMPTY (clear the summary, which the picker's "None" row has to be able to do). Written
    # the loose way, "None" would have been a no-op with the menu cheerfully showing the change.
    agg_name = str((raw.get('agg') if 'agg' in raw else prev.get('agg')) or '').strip()
    if agg_name in FIELD_AGGS:
        out['agg'] = agg_name
    # An automation column carries the binding the engine reads; it is preserved across a patch
    # rather than re-declared, because the automation editor owns it and this route does not.
    auto = raw.get('automation') if 'automation' in raw else prev.get('automation')
    if isinstance(auto, dict):
        out['automation'] = auto
    # C7 (wave 22): a metric bag is validated or the FIELD is refused — a mis-shaped metric
    # column stored anyway would be a column of invented blanks wearing a number's name.
    metric_raw = raw.get('metric') if 'metric' in raw else prev.get('metric')
    if metric_raw is not None:
        m = _clean_metric(metric_raw)
        if m is None:
            return None
        out['metric'] = m
    # C3 (wave 25, R7): the PROFILE FLAG. Same posture as the metric bag one line up — a
    # mis-shaped flag refuses the FIELD, and an explicit `profile: None` CLEARS it, so the flag
    # can be taken off a column without deleting the column. ⚠ `type: 'text'` is enforced here
    # rather than left to the caller: R7 is *"a flag on an ordinary text field"*, and a flag on
    # a select/int would promise a validated handle to a cell this module never validates.
    profile_raw = raw.get('profile') if 'profile' in raw else prev.get('profile')
    if profile_raw is not None:
        p = _clean_profile(profile_raw)
        if p is None or ftype != 'text':
            return None
        out['profile'] = p
    # ⭐ 2026-08-07 — the relational pair. Same posture as the two bags above: a mis-shaped bag
    # refuses the FIELD, and the bag is required for its own type. ⛔ THE TYPE/BAG PAIRING IS
    # ENFORCED BOTH WAYS — a `link` with no `link` bag is a column that can never resolve a row,
    # and a `link` bag on an `int` column is a declaration nothing reads. Either half alone is the
    # silent kind of broken: the column exists, renders, and means nothing.
    # ⚠ AN INHERITED BAG IS ONLY INHERITED WHILE THE TYPE STILL WANTS IT (2026-08-09, owner:
    # *"everything is custom and changeable always"*). `prev`'s bag used to be picked up on ANY
    # patch, so retyping a rollup column to text sent `{'type': 'text'}`, inherited the rollup bag
    # from `prev`, hit the `ftype != 'rollup'` arm and returned None — i.e. the server refused to
    # let a rollup become anything else, with a sentence about the name and type. Changing the
    # type is the one act that MEANS "drop this bag", so a mismatched inherited bag is dropped.
    # ⛔ AN EXPLICIT bag on a mismatched type STILL REFUSES: that is a caller declaring something
    # nothing will ever read, which is the silent-broken half this pairing exists to catch.
    prev_type = str(prev.get('type') or '').strip().lower()
    link_raw = (raw.get('link') if 'link' in raw
                else (prev.get('link') if prev_type == ftype else None))
    if link_raw is not None:
        lk = _clean_link(link_raw)
        if lk is None or ftype != 'link':
            return None
        out['link'] = lk
    elif ftype == 'link':
        return None
    rollup_raw = (raw.get('rollup') if 'rollup' in raw
                  else (prev.get('rollup') if prev_type == ftype else None))
    if rollup_raw is not None:
        rl = _clean_rollup(rollup_raw)
        if rl is None or ftype != 'rollup':
            return None
        out['rollup'] = rl
    elif ftype == 'rollup':
        return None
    # ⭐ WAVE-27 item 13 (owner ruling R13) — the `code` column's language bag.
    #
    # ⚠ Deliberately the WEAKEST posture of the four bags above, and the asymmetry is the whole
    # point rather than an inconsistency. `metric`/`link`/`rollup` bags DEFINE what their column
    # is: without one the cell can never resolve a value, so a missing or mis-shaped bag refuses
    # the field. A `language` chooses a HIGHLIGHTER over a string the column already holds
    # perfectly well — so it is optional (an unconfigured code column renders plain), it survives
    # a patch that omits it (the `automation` rule: the gear owns it, a header rename must not
    # clear it), and an unknown language degrades to `plain` instead of destroying the column.
    # R13's "NO execution engine" is why this can be so relaxed: the value selects a renderer and
    # can never select an interpreter.
    # ⚠ Function-local import, the convention `grid_events` already uses for the same module
    # (`import aios_grid as _ag2`, five call sites): `core` must not import the layer above it at
    # MODULE level. One vocabulary, one validator — declaring the language list a second time
    # here is exactly the drift the fields contract exists to catch.
    # ⚠ OMIT = KEEP, `plain` = CLEAR. `_clean_code` returns None for plain (see its docstring),
    # so an explicit `{'language': 'plain'}` from the editor drops the key here while a patch
    # that never mentions `code` inherits the previous bag. Without that split, switching a
    # column back to Plain text would be a control that silently does not save.
    code_raw = raw.get('code') if 'code' in raw else prev.get('code')
    if code_raw is not None and ftype == 'code':
        import aios_grid as _agc
        c = _agc._clean_code(code_raw)
        if c:
            out['code'] = c
    # ⭐⭐ 2026-08-10 — THE FORMULA EXPRESSION. Posture: `link`/`rollup`'s, not `code`'s — a
    # formula column with no expression is a column that can never show a value, so the pairing is
    # enforced BOTH WAYS and either half alone refuses the field.
    #
    # ⚠ `valid_keys` IS DELIBERATELY OMITTED, and it is the one real asymmetry with the overlay
    # door. `grid_events` passes the workspace's own key set, so a `{ref}` naming no column is
    # refused at write time there. This function is handed ONE field with no sibling context —
    # `add_field`/`patch_field` could supply the table's keys, but `clean_fields` validates a list
    # mid-construction where half the referenced columns do not exist yet, and `clean_machine_
    # fields` judges a list that has no table at all. A second, weaker copy of the ref rule in one
    # of the three would be the divergence this module keeps paying for. STRUCTURE is checked here
    # (charset, length, balanced parens and quotes, well-formed refs); an unresolvable ref blanks
    # the CELL client-side, which is `_clean_formula`'s own documented read-time behaviour and the
    # same answer a later-deleted column already produces.
    # ⚠ Reached through `_ag_formula` so BOTH doors call one validator through one import.
    formula_raw = (raw.get('formula') if 'formula' in raw
                   else (prev.get('formula') if prev_type == ftype else None))
    if formula_raw is not None:
        fx = _ag_formula(formula_raw)
        if fx is None or ftype != 'formula':
            return None
        out['formula'] = fx
    elif ftype == 'formula':
        return None
    return out


def is_derived_link(field):
    """Is this a link whose cell the ENGINE owns? (an `on` join, not a user-picked set)

    ⛔ ONE definition, read by the write wall, the compute pass and the wire. The alternative is
    three places independently asking "does the bag have an `on`", which is how one of them
    silently keeps taking writes after the other two stop.
    """
    lk = (field or {}).get('link')
    return isinstance(lk, dict) and bool(lk.get('on') or lk.get('inverse'))


def is_computed_cell(field):
    """Is this field's cell computed by the SERVER, so a human write is an invented value?

    The `ut_*` twin of `aios_grid.READONLY_CUSTOM_TYPES`, and deliberately a PREDICATE rather
    than a type set: a `link` is read-only when it is derived and editable when it is not, which
    a set of type names cannot express. `grid_events` is the wall that reads it.

    ⭐⭐ 2026-08-10 — `formula` JOINS, and it is the one member here whose value the server does
    NOT compute. Every other member is refused because the server owns the number; a formula cell
    is refused because NOBODY owns a stored number — the browser recomputes the column on every
    paint from the row's other cells, so a value written here is overwritten on screen before it
    is ever read and sits in the store as a fossil no reader agrees with. The client already
    declines to offer an editor (`READONLY_CELL_TYPES`) and declines the paste
    (`coerceClipboardValue`); both are courtesy, and this is the wall
    ([[schema-role-is-not-a-value-wall]] — a courtesy in front of no wall is the shape that
    keeps shipping).
    """
    f = field or {}
    return (isinstance(f.get('metric'), dict) or f.get('type') in ('rollup', 'formula')
            or is_derived_link(f))


#: The numeric family and the date family, named once. Membership decides which cells the
#: refusal below can speak about at all — a `text` column takes any string by definition.
_NUMERIC_CELL_TYPES = ('int', 'currency', 'pct')
_DATE_CELL_TYPES = ('date',)


def cell_type_refusal(field, value):
    """Why this value cannot go in this column, or None if it can. ⛔ REFUSES; never rewrites.

    ⭐⭐ W29-T81 — THE SERVER-SIDE HALF of "can this text be this field's value". Until now the
    only evaluator was `coerceClipboardValue`, which lives in the BROWSER — so the import door
    took `{"qa_count": "seventeen-ish"}` at an `int` column, answered `201 {"imported": 1}`, and
    stored it verbatim. Measured against production on `bac40c2`. A rule enforced only by the
    client is a rule that holds for exactly one client ([[one-evaluator-per-question]]).

    ⚠ DELIBERATELY NARROWER THAN THE CLIENT'S COERCER, and it must stay that way: this asks only
    whether a NUMBER is a number and a DATE is a date, the two families where an unusable string
    is later rendered as a fabricated figure. It accepts everything the client's coercer emits
    (canonical ISO dates, plain decimals) plus the human spellings that coercer also takes
    ("$1,234.50"), so no UI path can trip a wall the UI cannot see. Choice vocabularies stay
    client-side for now: their options change under a stored row, and refusing an import against
    a vocabulary edited yesterday would reject data that is merely out of date.

    ⚠ BLANK IS ALWAYS ACCEPTABLE. An empty cell is an empty cell in every column; the import
    plan already omits them, and refusing one would make a ragged spreadsheet unimportable.
    """
    f = field or {}
    ftype = f.get('type')
    text = '' if value is None else str(value).strip()
    if not text:
        return None
    label = f.get('label') or f.get('key') or 'that column'
    if ftype in _NUMERIC_CELL_TYPES:
        try:
            float(text.replace('$', '').replace(',', '').replace('%', '').replace(' ', ''))
        except (TypeError, ValueError):
            return f'{text[:40]!r} is not a number, and {label!r} holds numbers'
        return None
    if ftype == 'rating':
        try:
            n = int(text)
        except (TypeError, ValueError):
            return f'{text[:40]!r} is not a whole number, and {label!r} holds a rating'
        top = f.get('max') if isinstance(f.get('max'), int) and f.get('max') else 5
        if not 1 <= n <= top:
            return f'{label!r} takes a rating from 1 to {top}, and {text[:40]!r} is outside it'
        return None
    if ftype in _DATE_CELL_TYPES:
        # ISO-prefixed, which is exactly what the client's coercer produces and what every
        # reader of a stored date assumes. A month name is not refused as invalid English — it
        # is refused because nothing downstream would parse it either.
        if not re.match(r'^\d{4}-\d{2}-\d{2}', text):
            return (f'{text[:40]!r} is not a date, and {label!r} holds dates '
                    f'(write it as YYYY-MM-DD)')
        return None
    return None


#: ⭐⭐ 2026-08-09 (owner) — THE STAMP THAT SURVIVES A RECONCILER.
#:
#: Owner: *"No rollup field should be uneditable, everything is custom and changeable always."*
#: Unlocking `may_edit_field` alone would have been a lie with a delay on it: BOTH preset
#: reconcilers (`automation_engine._reconcile_ig_graph_fields` and `odoo_relational`'s forward
#: migration) OVERWRITE every contract key — `label`, `type`, `link`, `rollup` — on any field
#: still carrying `automation.preset`. So a reconfigured "Avg views · last 12 posts" would have
#: saved, rendered, recomputed, and then silently reverted to the shipped contract at the next
#: enrichment run or store resync. The edit would look like it worked, which is worse than a
#: refusal ([[flag-shipped-without-its-writer]]).
#:
#: ⛔ IT DOES NOT CLEAR `preset`, and that is deliberate. `preset` records PROVENANCE (this column
#: was spawned by the product, and `ut_ensure` re-stamps it on every run anyway — clearing it
#: would be undone within one tick). This stamp records CUSTODY: a human has taken this column
#: over, so the machine stops rewriting its definition while still owning its VALUES.
USER_EDITED_KEY = 'userEdited'


def user_edited(field):
    """Has a human taken over this preset column's DEFINITION? (the reconcilers' skip signal)"""
    auto = (field or {}).get('automation')
    return isinstance(auto, dict) and auto.get(USER_EDITED_KEY) is True


def preset_editable(field):
    """Does the pre-set lock let this column's DEFINITION be changed anyway?

    ⛔ ONE PREDICATE FOR ONE QUESTION, read by all three walls that enforce the lock
    (`may_edit_field` here, `routes_tables._field_or_refuse`, and the client's `schemaLocked`).
    Three independent copies of "is this preset?" is how one of them keeps refusing after the
    other two stop — the shape D-107 wore, and the reason `is_derived_link` above is a function
    rather than an expression repeated three times.

    Today it answers YES for exactly one thing: a ROLLUP. The owner's ruling names rollups, and
    the argument is specific to them rather than a general softening of the lock — a rollup holds
    no data of its own. It is a QUESTION asked of other rows ("average views over the last N
    posts"), it can be re-asked at any time, and re-asking it costs nothing because the answer is
    recomputed from the authoritative store on the next pass. Retyping a preset `followers`
    column, by contrast, would strand real measurements in a column that can no longer read them.
    """
    return str((field or {}).get('type') or '') == 'rollup'


def clean_machine_fields(fields):
    """`(clean, refused)` — the SANCTIONED way for a machine writer to seed a table's schema
    (wave 25, doc order step 2). `clean` is the list as this module would store it; `refused`
    names every field that did not survive, so a caller can be LOUD rather than short.

    ⛔ WHY THIS EXISTS. `automation_engine.ut_ensure` writes field dicts STRAIGHT into the
    `user_tables` bucket with `rt.update(UT_STORE_KEY, …)` — it never routes through this
    module, so **the field layer's own validator has never judged a single column the engine has
    ever spawned.** That is [[default-must-pass-its-own-guard]] with a live subject: the product
    seeds fields against a law it does not run. Nothing has broken yet only because the drift is
    one key wide and in the fail-closed direction (a field with no `editRole` reads as
    not-`everyone`), which is luck, not design.

    ⚠ THIS IS THE "IMPOSSIBLE" HALF AND IT IS OPT-IN; the LOUD half is `verify_api`'s derived
    section, which walks every field list the engine declares and refuses to let one through
    that this validator would change in any way but adding `editRole`. A helper the caller may
    ignore is not a control — the gate is the control, and it covers a list nobody remembered
    to route through here.
    """
    clean, refused = [], []
    for f in (fields or []):
        got = _clean_field(f)
        if got is None:
            refused.append(str((f or {}).get('key') or f))
        else:
            clean.append(got)
    return clean, refused


def may_edit_field(table_key, fkey, viewer, is_admin=False, st=None):
    """May `viewer` change THIS column's definition? Creator/admin always; others only when the
    field itself says `editRole: 'everyone'`. Fail-closed on an unknown field."""
    table = get(table_key, st) or {}
    field = next((f for f in (table.get('fields') or []) if f.get('key') == str(fkey)), None)
    # Instagram's pre-set schema is product contract, not tenant configuration. Even an admin
    # may sort/filter/hide it, but cannot rename, retype, duplicate or delete it. Ordinary fields
    # on the same database remain user-owned, including user-created Links and Rollups.
    if isinstance((field or {}).get('automation'), dict) \
            and field['automation'].get('preset') is True \
            and not preset_editable(field):
        return False
    if may_open(table_key, viewer, is_admin, st) and (
            bool(is_admin) or table.get('createdBy') == viewer):
        return True
    for f in (table.get('fields') or []):
        if f.get('key') == str(fkey):
            return f.get('editRole') == 'everyone'
    return False


def flow_bound(bag, st=None):
    """C8 (wave 22, owner item 5) — may this `automation` bag be STORED? Only when its
    `flowId` names an automation definition that exists in this tenant's `automations` bucket.
    Fail-closed both ways: no flowId is a column that silently never runs, and a flowId naming
    a deleted flow is the same thing one delete later. (Read tolerance for pre-law fields
    lives in `aios_grid._clean_automation` — this guards the WRITE doors only.)"""
    if not isinstance(bag, dict):
        return True                        # no bag, no law — an ordinary column
    flow = str(bag.get('flowId') or '').strip()
    if not flow:
        return False
    try:
        return flow in (_st(st).get('automations') or {})
    except Exception:
        return False


def add_field(table_key, raw, st=None):
    """Append one column to the shared schema. Returns the stored field, or None if refused."""
    if not is_user_table(table_key, st):
        return None
    defn = get(table_key, st) or {}
    have = [f for f in (defn.get('fields') or [])]
    if len(have) >= MAX_FIELDS:
        return None
    field = _clean_field(raw)
    if not field or any(f.get('key') == field['key'] for f in have):
        return None
    if not flow_bound(field.get('automation'), st):
        return None                        # C8: no field without a flow (400 at the route)
    # C3 (wave 25, R7's first mitigation): AT MOST ONE profile field per table. Without it the
    # flag means nothing — "the profile column" stops being resolvable the moment there are two,
    # and the enrich action would have to guess which handle it was pointed at. The route names
    # the existing column in its refusal (`_refusal_sentence`); the LAW is here.
    if field.get('profile') and _profile_of(have):
        return None

    def _add(cur):
        t = cur.get(str(table_key))
        if t is not None:
            t.setdefault('fields', []).append(field)
        return cur

    _st(st).update(STORE_KEY, _add, flush='sync')
    return field


def _reciprocal_link_key(table_key, field_key):
    digest = hashlib.sha1(f'{table_key}:{field_key}'.encode('utf-8')).hexdigest()[:10]
    return f'linked_{digest}'


def sync_reciprocal_link(table_key, field_key, st=None):
    """Create/repair Airtable's reciprocal link field for one ordinary link.

    The source cell stores picked target row ids. The reciprocal is a computed inverse link on
    the target database; it lists source rows that include the current target id. Retargeting or
    retyping the source removes the obsolete inverse in the same store update.
    """
    table_key, field_key = str(table_key), str(field_key)
    result = {'field': None, 'reciprocal': None}

    def _sync(cur):
        cur = cur if isinstance(cur, dict) else {}
        source = cur.get(table_key) or {}
        source_field = next((f for f in (source.get('fields') or [])
                             if f.get('key') == field_key), None)
        # Remove every old inverse for this source identity first. This makes retargeting and
        # deleting deterministic instead of leaving a live-looking backlink on the old table.
        for candidate in cur.values():
            if not isinstance(candidate, dict):
                continue
            candidate['fields'] = [f for f in (candidate.get('fields') or [])
                                   if not (isinstance(f.get('link'), dict)
                                           and f['link'].get('table') == table_key
                                           and f['link'].get('inverse') == field_key)]
        if not source_field or source_field.get('type') != 'link':
            return cur
        bag = dict(source_field.get('link') or {})
        if bag.get('on') or bag.get('inverse'):
            return cur
        target_key = str(bag.get('table') or '')
        target = cur.get(target_key)
        if target is None:
            return cur
        # ⛔ 2026-08-10 — THE COLUMN CAP APPLIES TO A COLUMN NOBODY ASKED FOR, TOO.
        #
        # `add_field` refuses at `MAX_FIELDS`; this function appended straight past it, and the
        # field it appends lands on a table the user is not even looking at. The IG preset set is
        # 41 columns against a cap of 60, so a Posts or Comments database is genuinely reachable —
        # and the failure mode is the worst kind: the source link saves, the target silently grows
        # its 61st column, and every LATER `add_field` on that table is refused with a sentence
        # about a cap the user never crossed on purpose.
        # ⚠ THE SOURCE LINK STILL SAVES. Refusing the whole relation because the OTHER database is
        # full would be a refusal the user cannot act on from where they are standing; an ordinary
        # link with no reciprocal is exactly what every link was before this function existed, and
        # it still resolves its own cell. `reciprocal` is simply not stamped on the bag, so the
        # next `sync` on a table with room creates it — this is a skip, not a tombstone.
        if len(target.get('fields') or []) >= MAX_FIELDS:
            return cur
        reciprocal_key = _reciprocal_link_key(table_key, field_key)
        reciprocal = _clean_field({
            'key': reciprocal_key,
            'label': str(source.get('label') or table_key)[:80],
            'type': 'link',
            'link': {'table': table_key, 'inverse': field_key,
                     'reciprocal': field_key},
            'editRole': 'admins',
        })
        if reciprocal is None:
            return cur
        target.setdefault('fields', []).append(reciprocal)
        bag['reciprocal'] = reciprocal_key
        source_field['link'] = bag
        result['field'] = dict(source_field)
        result['reciprocal'] = dict(reciprocal)
        return cur

    _st(st).update(STORE_KEY, _sync, flush='sync')
    return result


def patch_field(table_key, fkey, raw, st=None):
    """Edit one column's definition IN PLACE. Returns the stored field, or None if refused.

    ⚠ THE KEY NEVER MOVES. A field's key is what every stored cell is filed under, so accepting
    a new one here would orphan every value in the column while looking like a rename. The
    LABEL is the renameable thing; the key is identity.
    """
    if not is_user_table(table_key, st):
        return None
    fields = [dict(f) for f in ((get(table_key, st) or {}).get('fields') or [])]
    idx = next((i for i, f in enumerate(fields) if f.get('key') == str(fkey)), -1)
    if idx < 0:
        return None
    merged = dict(raw or {})
    merged['key'] = str(fkey)
    field = _clean_field(merged, fields[idx])
    if not field:
        return None
    if 'automation' in merged and not flow_bound(field.get('automation'), st):
        return None                        # C8: a patch cannot un-bind a column either
    # C3: the one-per-table law on the PATCH door too — flagging a second column is the same act
    # as adding one. `exclude=fkey` so re-saving the profile column does not collide with itself.
    if field.get('profile') and _profile_of(fields, exclude=str(fkey)):
        return None
    # ⭐⭐ 2026-08-09 — THE CUSTODY STAMP IS WRITTEN HERE, at the ONE door a human reaches.
    #
    # `USER_EDITED_KEY`'s note has the why. It is written here rather than inside `_clean_field`
    # because `_clean_field` is also the machine's validator (`clean_fields`, `clean_machine_
    # fields`): stamping there would mark the product's OWN spawn as user-edited and switch the
    # reconcilers off for every preset column on the first tick, which is the inverse of the bug
    # it exists to prevent.
    # ⚠ Only when the definition ACTUALLY differs. A no-op save (open the editor, press Save)
    # must not quietly take a column out of the product's contract.
    if user_edited(fields[idx]) or (field != fields[idx]
                                    and isinstance(fields[idx].get('automation'), dict)
                                    and fields[idx]['automation'].get('preset') is True):
        automation = dict(field.get('automation') or fields[idx].get('automation') or {})
        automation[USER_EDITED_KEY] = True
        field['automation'] = automation

    def _set(cur):
        t = cur.get(str(table_key))
        if t is not None:
            for i, f in enumerate(t.get('fields') or []):
                if f.get('key') == str(fkey):
                    t['fields'][i] = field
                    break
        return cur

    _st(st).update(STORE_KEY, _set, flush='sync')
    return field


def delete_field(table_key, fkey, st=None):
    """Drop one column from the shared schema. Refuses to leave a table fieldless.

    ⚠ THE CELLS ARE LEFT IN THE ROWS ON PURPOSE. A deleted column whose values were also
    scrubbed makes an accidental delete unrecoverable; the values are invisible without a field
    declaring them, and re-adding the column with the same key brings them back. Same reasoning
    as `routes_tables.delete_table` leaving the workspace bucket in place. Booked, not hidden.
    """
    if not is_user_table(table_key, st):
        return False
    fields = (get(table_key, st) or {}).get('fields') or []
    if len(fields) <= 1 or not any(f.get('key') == str(fkey) for f in fields):
        return False
    doomed = next((f for f in fields if f.get('key') == str(fkey)), {})
    doomed_link = doomed.get('link') if isinstance(doomed.get('link'), dict) else {}
    inverse_source = None
    if doomed_link.get('inverse'):
        source_table = str(doomed_link.get('table') or '')
        source_field = str(doomed_link.get('inverse') or '')
        source = get(source_table, st) or {}
        if len(source.get('fields') or []) <= 1:
            return False
        inverse_source = (source_table, source_field)

    def _drop(cur):
        t = cur.get(str(table_key))
        if t is not None:
            t['fields'] = [f for f in (t.get('fields') or []) if f.get('key') != str(fkey)]
        if inverse_source:
            source_table, source_field = inverse_source
            source = cur.get(source_table)
            if isinstance(source, dict):
                source['fields'] = [f for f in (source.get('fields') or [])
                                    if f.get('key') != source_field]
        # An ordinary link owns its reciprocal field. Delete the reciprocal in the same schema
        # update so no target database can retain a live-looking backlink to a missing source.
        for candidate in cur.values():
            if not isinstance(candidate, dict):
                continue
            candidate['fields'] = [f for f in (candidate.get('fields') or [])
                                   if not (isinstance(f.get('link'), dict)
                                           and ((f['link'].get('table') == str(table_key)
                                                 and f['link'].get('inverse') == str(fkey))
                                                or (inverse_source
                                                    and f['link'].get('table') == inverse_source[0]
                                                    and f['link'].get('inverse') == inverse_source[1])))]
        return cur

    _st(st).update(STORE_KEY, _drop, flush='sync')
    return True


def rename_choice_values(table_key, fkey, renames, st=None):
    """Rename select/multiselect OPTIONS **and migrate every stored cell** (contract C-RENAME).

    `renames` = `[{'from': old, 'to': new}, …]` — an EXPLICIT MAPPING, never a diff. A diff
    cannot tell "renamed Blue to Navy" from "deleted Blue and added Navy", and guessing wrong
    empties a column silently.

    Returns `{'options': n, 'cells': n}`. This is the DEFINITION half — the base options list and
    the base row values. A user's per-user overlay stratum and any view filter naming the old
    value are `core.table_store`'s half of the same contract; the caller runs both.
    """
    pairs = []
    for r in (renames or [])[:50]:
        if not isinstance(r, dict):
            continue
        a = ' '.join(str(r.get('from') or '').split())[:60]
        b = ' '.join(str(r.get('to') or '').split())[:60]
        if a and b and a != b:
            pairs.append((a, b))
    if not pairs or not is_user_table(table_key, st):
        return {'options': 0, 'cells': 0}
    mapping = dict(pairs)
    counts = {'options': 0, 'cells': 0}

    def _apply(cur):
        t = cur.get(str(table_key))
        if t is None:
            return cur
        multi = False
        for f in (t.get('fields') or []):
            if f.get('key') != str(fkey):
                continue
            multi = f.get('type') == 'multiselect'
            opts = f.get('options') or []
            new_opts, seen = [], set()
            for o in opts:
                v = mapping.get(o, o)
                if v not in seen:          # a rename ONTO an existing option merges them
                    seen.add(v)
                    new_opts.append(v)
                if o in mapping:
                    counts['options'] += 1
            if opts:
                f['options'] = new_opts
        for row in (t.get('rows') or {}).values():
            if not isinstance(row, dict) or str(fkey) not in row:
                continue
            cell = str(row.get(str(fkey)) or '')
            if not cell:
                continue
            if multi:
                # ⚠ MULTISELECT CELLS ARE COMMA-JOINED LABEL STRINGS, so a rename has to walk the
                # parts. Rewriting the whole string would only ever hit a single-value cell.
                parts = [p.strip() for p in cell.split(',')]
                nxt = [mapping.get(p, p) for p in parts]
                if nxt != parts:
                    row[str(fkey)] = ', '.join(dict.fromkeys(nxt))
                    counts['cells'] += 1
            elif cell in mapping:
                row[str(fkey)] = mapping[cell]
                counts['cells'] += 1
        return cur

    _st(st).update(STORE_KEY, _apply, flush='sync')
    return counts


def patch_cells(table_key, row_id, values, st=None):
    """Write shared definition-row cells for trusted server-side writers."""
    if not is_user_table(table_key, st):
        return False
    rows = (get(table_key, st) or {}).get('rows') or {}
    if str(row_id) not in rows:
        return False

    def _set(cur):
        t = cur.get(str(table_key))
        if t is not None:
            t.setdefault('rows', {}).setdefault(str(row_id), {}).update(
                {str(k): str(v) for k, v in (values or {}).items()})
        return cur

    _st(st).update(STORE_KEY, _set, flush='sync')
    return True


def patch_link_cell(table_key, row_id, fkey, value, st=None):
    """Persist one user-picked Link cell in the shared row and return its canonical id string.

    A Link is a relationship in the database schema, not one user's visual overlay. Persisting
    it here makes the reciprocal field and every Rollup see the same source of truth. Derived
    joins and inverse fields are engine-owned and are refused by this door.
    """
    table_key, row_id, fkey = str(table_key), str(row_id), str(fkey)
    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
        return None
    table = get(table_key, st) or {}
    if row_id not in (table.get('rows') or {}):
        return None
    field = next((f for f in (table.get('fields') or []) if f.get('key') == fkey), None)
    bag = (field or {}).get('link')
    if (field or {}).get('type') != 'link' or not isinstance(bag, dict) \
            or bag.get('on') or bag.get('inverse'):
        return None
    target = get(str(bag.get('table') or ''), st) or {}
    valid = set((target.get('rows') or {}).keys())
    raw_ids = value if isinstance(value, (list, tuple, set)) else str(value or '').split(',')
    picked, seen = [], set()
    for raw_id in raw_ids:
        rid = str(raw_id).strip()
        if not rid or rid in seen:
            continue
        if rid not in valid or len(picked) >= LINK_MAX_IDS:
            return None
        seen.add(rid)
        picked.append(rid)
        if bag.get('single'):
            break
    canonical = ','.join(picked)

    def _set(cur):
        current = cur.get(table_key)
        if current is not None and row_id in (current.get('rows') or {}):
            current['rows'][row_id][fkey] = canonical
        return cur

    _st(st).update(STORE_KEY, _set, flush='sync')
    return canonical


def patch_profile_cell(table_key, row_id, fkey, value, st=None):
    """Write a PROFILE cell — and, when it is blanked, clear that row's preset cells IN THE SAME
    WRITE (wave 25, contract C3 + owner ruling R6). Returns
    `{'handle', 'cleared': [key, …]}`, or None when the write is refused.

    ⭐ THIS IS THE `ut_*` WRITE DOOR R6 NAMES, and it is deliberately NOT the event seam.
    HARD RULE 5: item 5a's clearing *looks* exactly like an event-trigger feature, and
    implementing it by widening `grid_events`' emit guard is D-40 — deferred by ruling A-15
    because customer/product `EventCtx` carries no scoped handle and a nurilab user's edit would
    fire against royal-imports' automations. Nothing here emits anything.

    ⚠ IT WRITES THE DEFINITION ROW, NOT AN OVERLAY (amendment C3-A1, MEASURED). An ordinary
    `ut_*` cell edit lands in the typist's own overlay stratum, where the automation engine —
    which reads `t['rows']` — can never see it. A handle nobody can enrich is the flag being
    decorative, so a profile cell writes THROUGH to the shared row, exactly as the stage field
    does for exactly the same reason.

    ⚠ ONE `update`, not two. "In the same write" is the contract, and it is also the only safe
    shape: a blank committed without its clear leaves a row whose handle is gone and whose
    follower count still reads 41,000 — stale numbers attributed to nobody, which is worse than
    either the old row or the empty one.

    ⛔ IT DELETES NO FIELDS AND TOUCHES NO HISTORY. The columns stay (they are the table's
    schema, and the next handle refills them); `ut_ig_snapshots` is a different table this
    module never opens. R3's *"one store for one series"* is why that separation holds — the
    cleared cells are the LATEST-value stamp, and the series itself was never in them.
    """
    if not is_user_table(table_key, st):
        return None
    defn = get(table_key, st) or {}
    rows = defn.get('rows') or {}
    if str(row_id) not in rows:
        return None
    fields = defn.get('fields') or []
    fdef = next((f for f in fields if f.get('key') == str(fkey)), None)
    if not isinstance(fdef, dict) or not isinstance(fdef.get('profile'), dict):
        return None                        # not a profile column — the caller's wall failed open
    handle, ok = normalize_profile(value, fdef['profile'].get('source'))
    if not ok:
        return None                        # refused with a sentence at the route
    # Clear only the known Instagram presets that the automation declared, never every field with
    # automation provenance.
    cleared = []
    if not handle:
        cleared = [f['key'] for f in fields
                   if f.get('key') in PROFILE_PRESET_KEYS
                   and isinstance(f.get('automation'), dict)
                   and str(rows.get(str(row_id), {}).get(f['key']) or '') != '']

    def _set(cur):
        t = cur.get(str(table_key))
        if t is not None:
            row = t.setdefault('rows', {}).setdefault(str(row_id), {})
            row[str(fkey)] = handle
            for k in cleared:
                row[k] = ''
        return cur

    _st(st).update(STORE_KEY, _set, flush='sync')
    return {'handle': handle, 'cleared': cleared}


def delete_row(table_key, row_id, st=None):
    """⛔ SYNC, AND IT WAS `async` — owner report 2026-08-10: *"it took me forever to delete a
    record… it took me like 3 tries"*.

    `flush='async'` applies the drop to the in-process CACHE and defers the upload by
    `store._FLUSH_DELAY` (2 s) so a burst coalesces into one commit. That trade was made for
    `add_row`, and made there for a stated reason — the owner's *"adding a new record visually
    takes too long, I need to be able to spam it"*. **This function inherited the flag and none of
    the argument.** Nobody spams a delete: it is rare, it is destructive, and it is the one
    mutation where "eventually consistent" is indistinguishable from "it did not work". Inside
    that window the row is gone from this process and still present in the store, so anything that
    re-reads the hub copy — a restart, an eviction, another reader — brings it back, and the user
    deletes it again.

    ⚠ ONE COMMIT PER DELETE is the whole cost, and it is the correct one to pay: the alternative
    is a destructive action whose durability depends on the container staying up for two seconds.
    Adds keep their async path untouched — the spam case is real and is theirs.
    ⚠ THIS IS NOT THE WHOLE OF D-118. A sync commit closes the lost-delete window; it does not make
    two writers safe, which is D-4's job.
    """
    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
        return False

    def _drop(cur):
        t = cur.get(str(table_key))
        if t is not None:
            (t.get('rows') or {}).pop(str(row_id), None)
        return cur

    _st(st).update(STORE_KEY, _drop, flush='sync')
    return True


#: Names the ENGINE uses for itself when no human is on the other end of a run. ⛔ A table
#: stamped with one of these has NO HUMAN OWNER, so `may_open` can only admit an admin — which
#: is why the automation engine must never mint one (see `ut_ensure`'s owner rule, wave 20).
#: Named here rather than inline so the two modules that care about it read the SAME list.
MACHINE_OWNERS = ('automation', 'scheduler')


def may_open(table_key, viewer, is_admin=False, st=None):
    """FAIL-CLOSED visibility: the creator, or an admin. Nothing else. **THE one resolver.**

    A user table cannot be gated by `allowed_modules` — a module grant written last month cannot
    describe a table created this morning — so it needs its own rule, and the safe rule is the
    narrow one. Tables are stored tenant-wide under one key, so without this ANY user (including
    a BU-scoped sales agent) could open a table somebody else created just by knowing its key.
    Sharing a user table with named colleagues is a follow-on, and it should reuse the shared-
    VIEW vocabulary rather than inventing a second one.

    ⛔ WAVE 20 — THERE WAS A SECOND, WIDER RULE, AND THE TWO DISAGREED. `routes_automation`'s
    table picker admitted `createdBy in (uname, 'automation', 'scheduler')`, so a non-admin could
    SEE an automation-created database in the picker and then be refused when they opened it,
    edited it, or tried to delete it. Two ideas of who owns a table is the same defect class as
    the two ideas of an agent's book (D-30): both surfaces look right in isolation and only
    disagree in front of a user. The picker now calls THIS function, and the engine stamps a
    real owner (see `MACHINE_OWNERS`) so nothing legitimate is narrowed by the merge.
    """
    t = get(table_key, st)
    if not t:
        return False
    if bool(is_admin) or t.get('createdBy') == viewer:
        return True
    # ⭐ WAVE 20 (R10, closes DEBT D-32) — A GRANT ADMITS TOO. `core/shares.py` is the ONE registry
    # for view/folder/database grants, and this is the line that makes a shared DATABASE actually
    # open: without it the share was RECORDED and the receiver's nav never changed, which reads as
    # "sharing is broken" rather than as a missing wall.
    #
    # ⚠ PURELY ADDITIVE, and deliberately last: it can only ever admit somebody a grant NAMES, and
    # can never deny anyone the creator-or-admin rule already admits. `role_for` is fail-closed on
    # a junk bucket or an unreadable store, so an unreachable registry degrades to today's rule
    # rather than to an open door.
    try:
        import core.shares as shares
        return shares.may_see('database', table_key, viewer, is_admin=is_admin, st=st)
    except Exception:
        return False


#: ⭐⭐ W31-T10 / CONTRACT C1 — THE TWO BUCKETS `may_open` READS, LENT FOR THE LENGTH OF ONE PASS.
#:
#: `core/store.py:Store.get` deep-copies the WHOLE tenant document on every call, hit or miss,
#: under `self._lock` — a documented 35.8 MB ceiling. So asking `may_open` about N databases cost
#: `1 + N` full copies of `user_tables`, plus up to another N of `object_shares` (its last leg),
#: and `GET /nav` measured a median of **9,548 ms live / 24,741 ms in-process** on tenant #0.
#:
#: ⛔ THE OBVIOUS FIX IS THE FORBIDDEN ONE, and `may_open`'s own docstring is the record of why:
#: wave 20 found TWO ideas of who owns a table (`routes_automation` admitted
#: `createdBy in (uname, 'automation', 'scheduler')`), so a non-admin saw a database in a picker
#: and was refused on opening it. Inlining creator-or-admin at a call site re-creates exactly
#: that. **`may_open` stays THE one resolver, unmodified and still asked about every table** —
#: it is simply no longer charged for documents the caller is already holding.
#:
#: ⚠ WHY THE SHARES BUCKET IS IN HERE TOO, when `routes_automation._LentTables` intercepts only
#: `user_tables`: `may_open`'s admit-by-grant leg calls `shares.may_see` → `role_for` → `grants`,
#: which is `st.get('object_shares')` PER TABLE. It is a different key, so a `user_tables`-only
#: lend passes it straight through — the exact residual C1 says needs its own answer. It is read
#: LAZILY (an admin, or a table's creator, is admitted before that leg runs and never pays for
#: it) and memoised for the pass.
#:
#: ⚠ THE ALLOW-LIST IS THE SAFETY ARGUMENT, not the read-only-ness. Any other key passes through
#: to the real runtime uncached, `fresh=True` always bypasses, and every WRITE verb reaches the
#: runtime through `__getattr__` — so this can serve a stale value for exactly two keys, for
#: exactly one pass, to code proven not to mutate them (`all_tables` shallow-copies the top level
#: before anybody touches it; `grants`/`_clean_entries` build new objects; `may_open` only READS
#: `createdBy`).
_LENDABLE = (STORE_KEY, 'object_shares')


class _Lent:
    """A read-only `st` that serves already-read documents and passes everything else through."""

    def __init__(self, runtime, docs):
        self._runtime, self._docs = runtime, dict(docs or {})

    def get(self, name, fresh=False):
        if fresh or name not in _LENDABLE:
            return self._runtime.get(name, fresh) if fresh else self._runtime.get(name)
        if name not in self._docs:
            self._docs[name] = self._runtime.get(name)      # lazily, once, for this pass
        return self._docs[name]

    def get_projection(self, name, drop=()):
        """⭐ W32-T01: serve the LENT document here too, or a projection re-reads what the pass
        already holds — `__getattr__` would forward this straight to the runtime and the lend
        would buy nothing for exactly the callers built to use it.

        ⚠ THE TWO MISMATCHES ARE BOTH SAFE, AND IN OPPOSITE WAYS — which is why this serves the
        lent document as-is rather than re-projecting it:
        • lent WHOLE, asked for a projection ⇒ the caller gets MORE than it asked for. Correct,
          just not cheap — the one direction a shortcut here is allowed to be wrong in.
        • lent PROJECTED, asked for the whole thing via `get` ⇒ the caller gets a `_Projected`,
          and reading a dropped key RAISES with the reason. Loud, not silent.
        A re-projection would cost a second copy to convert the first case and change nothing
        about the second.
        """
        if name not in _LENDABLE:
            return self._runtime.get_projection(name, drop=drop)
        if name not in self._docs:
            self._docs[name] = self._runtime.get_projection(name, drop=drop)
        return self._docs[name]

    def __getattr__(self, name):
        return getattr(self._runtime, name)


def lend(st=None, **docs):
    """A handle that serves `docs` for `_LENDABLE` keys and defers everything else to `st`.

    `lend(rt, user_tables=doc)` — keyword names are STORE KEYS. Call it once per read pass and
    hand the result to every predicate in that pass; see the note above for what makes it safe.
    """
    return _Lent(_st(st), {k: v for k, v in docs.items() if v is not None})


def nav_entries(viewer=None, is_admin=False, st=None):
    """Registry-SHAPED dicts so the nav can render user tables beside real modules with no
    special-casing: the same keys the flat nav reads (`key`, `label`, `source`).

    Filtered by `may_open`, so the nav cannot offer a row the dispatcher would refuse.

    ⭐ W31-T10 (C1): ONE document read for the whole listing. The filter is unchanged — the same
    `may_open`, asked about the same tables, in the same order — it is handed the document this
    function has already read instead of taking a fresh 35.8 MB copy per table. A caller that
    already holds the document passes its own `lend(...)` as `st` and this pays nothing at all.

    ⭐⭐ W32-T02 (R8/D-175): and that ONE read is now a PROJECTED one. This function reads `label`,
    `source` and `key`, and `may_open` reads `createdBy` (then the shares registry) — **no caller
    of this listing has ever touched a row.** On tenant #0 that turns a 1,750 ms copy into 1.4 ms,
    for both consumers: `nav()` and `_placeable_top_keys`, i.e. `GET /nav` and `GET /nav/prefs` —
    the two that `Shell.tsx` fires CONCURRENTLY on the same `Store._lock`, which is why they added
    up rather than overlapped and why the rail rows arrived seconds apart (owner item 5).
    """
    tables = all_defs(st)
    lent = lend(st, **{STORE_KEY: tables})
    out = []
    for key, t in sorted(tables.items(),
                         key=lambda kv: (kv[1].get('label') or '').lower()):
        if viewer is not None and not may_open(key, viewer, is_admin, st=lent):
            continue
        out.append({'key': key, 'label': t.get('label') or key,
                    'source': t.get('source') or BLANK_SOURCE,
                    'user_table': True, 'nav': True, 'validate': False})
    return out