api/routes_tables.py

"""routes_tables.py — USER TABLES over the wire (wave 18, contract C3-UT).

The runtime-database primitive: `core/user_tables.py` (wave-9 C6, host-only until now) served
through the SAME grid machinery every other topic rides — `table_store` for the per-user
workspace strata, `aios_grid.workspace_wire` for the wire shape, `core.grid_events` for every
durable write except rows. Rows are the one genuinely new channel: the events seam has no row
event types (the user_tables docstring's `row_add` gate was described, never built), so row
add/delete/patch are REST endpoints here, walled by `user_tables.is_user_table` +
`user_tables.may_open` — a connector-backed table can never accept an invented row.

TENANCY: every store touch goes through `session.runtime` (the tenant's store handle), so a
Nurilab admin's tables live under Nurilab's prefix/repo, and the isolation gate's proof #3
covers them for free.

VISIBILITY is `user_tables.may_open` — creator or admin, fail-closed. There is no module grant
to check because a user table is not a module; the per-table wall is the whole wall, and it is
applied in `_defn_or_refuse` before any payload is built.
"""
import threading
import json
import time

from fastapi import Body, Depends
from fastapi import APIRouter

from deps import Session, err, require_session

router = APIRouter(prefix="/api/v1")


def _ut():
    import core.user_tables as user_tables
    return user_tables


def _ops(session, table_key):
    import core.table_store as table_store
    return table_store.make(f"{table_key}_table_workspace", st=session.runtime)


#: WAVE 27 item 2 (contract C2 / amendment A2) — the relation refresh is COALESCED and runs OFF
#: the request path. `{tenant: True}` while a pass is queued; the worker holds the lock.
_REL_LOCK = threading.Lock()
_REL_DIRTY = {}
_REL_RUNNING = {}


def _refresh_relations(session):
    """Mark this tenant's Links/Rollups stale and refresh them AFTER the response, once.

    ⛔ WHY THIS IS NOT A DIRECT CALL ANY MORE, and the numbers are the argument. The owner's
    item 2 was "adding a new record visually takes too long, I need to be able to spam it", and
    this function was the largest single cost inside `POST /tables/{key}/rows`:
    `engine.refresh_relations` DEEP-COPIES every table and every row in the tenant before it can
    decide whether anything needs doing (`automation_engine.py:9383-9388`), and when something
    does it commits with `flush="sync"` — a store round trip, i.e. an HF Dataset commit on the
    default backend. Every added record paid a whole-tenant snapshot plus a synchronous commit
    before the 201 came back.

    ⛔ AND BACKGROUNDING ALONE WOULD HAVE BEEN A WORSE BUG. Twenty rapid adds would queue twenty
    whole-tenant snapshots, each one re-reading a store the previous one just wrote — the spam
    the item asks us to support is exactly the load that would melt it. So this COALESCES: at
    most one pass runs, and at most one is queued behind it.

    ⚠ THE DIRTY FLAG IS THE LOAD-BEARING PART, not the lock. A write landing WHILE a pass is in
    flight must still get a pass afterwards, because the in-flight one snapshotted before that
    write existed. Without the flag the LAST add in a burst is precisely the one whose rollups
    never update — the failure nobody would notice until a total was quietly wrong.

    Eventual consistency is the accepted trade and was already the documented posture: the tick
    repairs materialised cells regardless, and answering 503 here would invite the browser to
    repeat a mutation that already succeeded.
    """
    tenant = str(getattr(session, "tenant", "") or "")
    rt = session.runtime
    with _REL_LOCK:
        _REL_DIRTY[tenant] = True
        if _REL_RUNNING.get(tenant):
            return                     # a worker is live; it will see the flag and loop
        _REL_RUNNING[tenant] = True

    def _worker():
        import automation_engine as engine
        try:
            while True:
                with _REL_LOCK:
                    if not _REL_DIRTY.get(tenant):
                        _REL_RUNNING.pop(tenant, None)
                        return
                    _REL_DIRTY.pop(tenant, None)
                try:
                    engine.refresh_relations(rt, log=lambda *_args: None)
                except Exception as exc:                          # noqa: BLE001
                    # The source write already landed and the response is already sent. Log and
                    # let the tick repair it; never retry in a tight loop.
                    print(f"[tables] relation refresh deferred: {type(exc).__name__}: {exc}")
        finally:
            # Belt for an unexpected raise on the bookkeeping itself: a tenant left marked
            # RUNNING would never refresh again for the life of the process.
            with _REL_LOCK:
                if _REL_RUNNING.get(tenant) and not _REL_DIRTY.get(tenant):
                    _REL_RUNNING.pop(tenant, None)

    threading.Thread(target=_worker, daemon=True,
                     name=f"rel-refresh:{tenant or 'default'}").start()


def _defn_or_refuse(session, table_key, st=None):
    """The per-table wall: 404 for a key that does not exist, 403 for one this session may not
    open. 404-before-403 leaks nothing useful — ut keys are guessable slugs, and 'exists but
    not yours' is exactly what may_open is for.

    ⭐ `st` LETS A CALLER LEND THE SNAPSHOT IT IS ALREADY HOLDING (W31 QA), the same `lend()` the
    nav and `/tables` use since W31-T10/T12. The WALL is untouched — the same `may_open`, asked
    about the same table — it is simply not re-reading a 28.5 MB document to ask it.
    """
    ut = _ut()
    # ⭐⭐ W31 QA — THE SWEEP, AND IT IS ONE LINE BECAUSE IT IS DONE HERE RATHER THAN PER ROUTE.
    # Owner: *"this is just one case, I need you to check and apply the fix everywhere too."*
    # This guard has FOURTEEN call sites and cost TWO whole-document deep copies at every one
    # (`get`, then `may_open`) — on tenant #0 that is 2 x 28.5 MB (D-185) to answer two questions
    # about ONE table. Lending here fixes every caller at once, including the eight routes the
    # sweep enumerated (`PATCH /shared/{pid}` · `DELETE /shared/fields/{k}` · `GET|POST /rows` ·
    # `POST /rows/import` · `POST|PATCH /fields` · `PATCH /rows/{pid}`).
    # ⛔ WHY IT IS SAFE HERE AND WOULD NOT BE IN THE ROUTES: a lend is a PRE-WRITE snapshot. The
    # guard runs before any mutation and returns only the DEFINITION, so the snapshot never
    # survives to serve a read-back. Blanket-replacing `st=session.runtime` inside the routes
    # would hand that stale snapshot to `patch_row`'s and `add_row`'s post-write read-back, which
    # is [[refetch-eats-its-own-write]] with the sign flipped — a write that reports the value it
    # replaced. The remaining in-route reads are deliberately untouched and booked instead.
    st = st if st is not None else ut.lend(session.runtime)
    defn = ut.get(table_key, st=st)
    if not defn:
        raise err(404, "unknown_table", "that database does not exist")
    if not ut.may_open(table_key, session.uname, session.admin, st=st):
        raise err(403, "forbidden", "that database belongs to another user")
    return defn


def _records_or_refuse(session, table_key, st=None):
    """The human record-write wall for a database the automation engine owns."""
    # One lend for BOTH questions this wall asks — the definition wall and the record-mode wall —
    # so the two are answered from one read instead of two. Same reasoning as `_defn_or_refuse`.
    st = st if st is not None else _ut().lend(session.runtime)
    defn = _defn_or_refuse(session, table_key, st=st)
    if not _ut().records_mutable(table_key, st=st):
        raise err(403, "records_read_only",
                  "records in this automation-owned database are read-only — add Instagram "
                  "handles in a Profile database and let enrichment populate this database")
    return defn


#: ⛔ THE PER-USER CANDIDATE WALL IS RETIRED (WAVE 27, DEBT D-72). **THE TENANT IS THE UNIT, NOT
#: THE USER** — one profile is ONE row, and everyone who may open the database sees all of it.
#:
#: WHY IT HAD TO GO, and it is not a preference: wave 26's R4 made the candidate identity
#: `(platform, handle)` and `_merge_candidates` stamps only the FIRST finder, later finders never
#: overwriting. Combined with a wall that then showed a non-admin only `created_by == me`, the
#: two were SILENT DATA LOSS — the second finder's row was merged away into the first finder's,
#: and the wall then hid the survivor from the person who just found it. They searched, they
#: paid, and the screen said nothing arrived.
#:
#: ⚠ AND THE PAIR WAS MUTUALLY MASKING, which is why it stayed green for a wave: the wall named
#: `ut_ig_candidates`, and a READ-ONLY census of all four tenant stores
#: (`ops/w26_candidate_census.py`) proved that table exists in NO tenant — since wave 25's R2 the
#: write target is whatever database the user points the Create-record action at. So the wall
#: governed a table nobody writes, and fixing EITHER half alone would have armed the other
#: ([[defects-that-mask-each-other]]).
#:
#: The register offered two exits and R4 already implied this one. Restoring per-user visibility
#: instead would have required R4's merge to stop crossing users — a bigger change, against the
#: ruling, to bring back a wall that never governed anything real.
#:
#: ⛔ DO NOT RE-ADD THIS BY INFERRING THE RULE FROM A `created_by` COLUMN. Every
#: automation-written table has one (a scraped row says `automation`), so inference would hide
#: every scraped row from every non-admin — the same disappearance defect, one table wide instead
#: of one table narrow. `created_by` survives as W26/R4's informational "Found by" stamp ONLY.


def _too_big():
    """`routes_odoo_tables.TooBigToMaterialise`, imported lazily — one name, two policies below."""
    import routes_odoo_tables
    return routes_odoo_tables.TooBigToMaterialise


def _read_through_rows(table_key, field_keys, rt=None):
    """The mirror's rows for one read-through grid, projected to this table's declared columns.

    ⭐⭐ W31-T45 / D-169 — `rt` IS THE SESSION'S TENANT RUNTIME AND IT IS PASSED THROUGH (D's ask,
    `mailbox/D.md` D-2). `_defn_or_refuse` above answers *"may this SESSION open this DATABASE"*;
    the guard `whole_pool` fires on `rt` answers a DIFFERENT question — *"is the DuckDB file this
    process has open THIS TENANT's"* — and R2 gives GTM Lab connected tables in its own document,
    which is exactly the shape that satisfies the first wall while failing the second.
    `datastore.ro_con()` reads a process-global `DB_PATH` and one Space process serves every
    tenant, so the two walls are not substitutes for each other.

    ⛔ ONE FETCH, TWO POLICIES — and the split is the whole of W31-T20. Both `scoped_pool` (which
    owes a caller every row) and `scoped_pids` (which owes only the row SET) reach the mirror
    through this function, so the pid set the cheap path answers with is IDENTICAL to the pool's
    by construction rather than by a second query that agrees today. A pid-only `SELECT` would be
    cheaper and would also be a SECOND statement of what a row of this table is
    ([[one-question-two-normalizers]]) — the two callers differ in what they do with the REFUSAL,
    never in how they ask.

    Raises `TooBigToMaterialise` when the population exceeds one window; the caller decides
    whether that is a 409 or an unresolved pid scope.
    """
    import routes_odoo_tables

    rows_src = [{k: v for k, v in r.items() if k in field_keys or k == "pid"}
                for r in routes_odoo_tables.whole_pool(table_key, rt=rt)]
    rows_src.sort(key=lambda r: r["pid"])
    return rows_src


#: ⭐⭐ W31-T20 / D-174 — R6's SECOND SENTENCE FOR A PID SCOPE THAT CANNOT BE RESOLVED.
#:
#: Wave 30 un-capped the ROW door and left the WORKSPACE envelope, so `GET /workspace?scope=
#: ut_odoo_gl_lines` answered `409 window_required` six times out of six on the live deploy and the
#: grid painted an ERROR PAGE with a Retry button — a whole shipped feature nobody could open.
#: The cause: an envelope needs no row, but it asked for every one of 963,783 of them to derive a
#: pid set it uses for exactly two things (cohort membership and a shared view's `memberPids`).
#:
#: ⛔ SO THE ENVELOPE STOPS ASKING, AND SAYS SO. An empty pid set is not "no rows" — it is
#: "membership is unresolved on this grid", which is a different claim and has to be made out
#: loud, on the wire, or it is the silent truncation R6 is actually about. Both consequences are
#: fail-closed: a stored cohort's members are reported MISSING by `aios_grid.workspace_wire`
#: rather than silently dropped, and any WRITE that names a pid is refused by `routes_grid`.
_PID_SCOPE_LIMIT = {
    "subject": "pids", "effect": "unresolved",
    "recommendation": "cohorts and shared-view membership are resolved per page on this grid; "
                      "filter and read it a window at a time (`/odoo-tables/{key}/rows`), where "
                      "every total is a SQL count over the whole table",
}


def scoped_pool(session: Session, table_key: str):
    """`(pids, rows_src, fields_base, defn)` — THE USER-TABLE WALL, on its own.

    `routes_products.scoped_pool`'s sibling, extracted for the same reason (wave 19, item 12): a
    caller that only needs "which rows of this database may this session touch" — record comments
    — must ask through `_defn_or_refuse` (404/403, the whole wall) rather than growing a second
    idea of what a user table's pool is.
    """
    defn = _defn_or_refuse(session, table_key)
    fields_base = [dict(f) for f in (defn.get("fields") or [])]
    field_keys = {f["key"] for f in fields_base}
    # ⭐⭐ W30-T31 / D-87 — A READ-THROUGH DATABASE HAS NO ROWS HERE, SO THEY COME FROM THE MIRROR.
    #
    # This is the one function that turns "what is stored" into "what this session may see", which
    # is exactly why the read-through arm belongs HERE and nowhere else: the rows route, the events
    # route, the comments wall and the assembly all reach rows through it, so they all convert
    # together or not at all. ⛔ Reading `defn["rows"]` for such a table would find `{}` and serve
    # an EMPTY GRID — correct-looking, wrong, and silent.
    # ⚠ The wall above has already run. This adds no scope of its own and takes none away.
    if not _ut().materialises(table_key, st=session.runtime, defn=defn):
        try:
            rows_src = _read_through_rows(table_key, field_keys, rt=session.runtime)
        except _too_big() as e:
            # R6's second sentence: a limit that cannot be met is a SENTENCE, never a short grid.
            # ⚠ THE ROWS PATH STILL REFUSES, AND THAT IS CORRECT (W31-T20 changed the ENVELOPE, not
            # this): a caller that asked for every row of a 963,783-row grid cannot be served a
            # short one. `scoped_pids` below takes the same refusal and answers a different
            # question with it, because an envelope needs no row.
            raise err(409, "window_required", str(e))
        except RuntimeError as e:
            raise err(503, "store_not_ready", str(e))
        return frozenset(r["pid"] for r in rows_src), rows_src, fields_base, defn
    rows_src = []
    for rid, row in (defn.get("rows") or {}).items():
        if not str(rid).isdigit():
            continue
        # WAVE 27 / D-72: no per-row owner filter. The table-level wall above
        # (`_defn_or_refuse` -> `may_open`) is the WHOLE wall — see the retirement note on
        # PER_USER_TABLES' former home. A row this session can reach is a row the tenant owns.
        r = {k: v for k, v in (row or {}).items() if k in field_keys}
        r["pid"] = int(rid)
        rows_src.append(r)
    rows_src.sort(key=lambda r: r["pid"])
    return frozenset(r["pid"] for r in rows_src), rows_src, fields_base, defn


@router.patch("/tables/{table_key}/shared/{pid}")
def patch_shared_cell(table_key: str, pid: int, body: dict = Body(default=None),
                      session: Session = Depends(require_session)):
    """Write a cell into the TENANT-WIDE overlay — the product door `core/shared_overlay.py` has
    been waiting for since it shipped (W29-T62, wave 30 T28).

    ⭐ WHY A SEPARATE STRATUM AT ALL, restated because it is the whole feature and it is not
    "sharing would be nice": a user-created column and its values live PER USER, so a shared view
    filtering on one names a column other accounts do not have — and an unknown column is an
    INACTIVE condition in the tri-state engine, which IGNORES it and therefore WIDENS. The buy
    list would silently show the whole catalogue to everyone but its author. A column whose value
    is the same for every reader is the precondition for editing it at all.

    ⛔ THE WALL IS `_defn_or_refuse`, AND THE STRATUM IS NOT ONE. `shared_overlay` refuses no
    reader and no writer by design; "may this session open this surface" is answered HERE, where
    the session is. Do not push the question down there.
    ⚠ A tenant-wide write is not a private one: every account that may open this database sees it.
    That is the point, and it is why this door declares the column too — a value with no
    definition is a cell nobody can find.
    """
    body = body if isinstance(body, dict) else {}
    key = str(body.get("field") or "").strip()
    if not key:
        raise err(400, "bad_request", "a field key is required")
    _defn_or_refuse(session, table_key)
    from core import shared_overlay
    if not shared_overlay.is_shared(table_key, key, st=session.runtime):
        shared_overlay.put_field(table_key, key, {
            "key": key, "label": str(body.get("label") or key), "source": "overlay",
            "type": str(body.get("type") or "text"), "shared": True,
            "createdBy": session.uname}, st=session.runtime)
    try:
        # ⚠ `put_cell`, not `put_cells` — this door writes exactly ONE cell, and the singular is
        # the API that says so. It delegates to the plural, so both stay reachable through the one
        # caller; before this, the singular had no caller at all and `verify_reachability` LENS 2
        # named it (the same lens that found `drop_field` had no door either).
        stored = {key: shared_overlay.put_cell(table_key, pid, key, body.get("value"),
                                               st=session.runtime)}
    except ValueError as e:
        # A non-scalar RAISES in the stratum rather than being dropped; relay it as the answer.
        raise err(400, "bad_value", str(e))
    return {"ok": True, "pid": pid, "cells": stored,
            "fields": list(shared_overlay.fields(table_key, st=session.runtime))}


@router.delete("/tables/{table_key}/shared/fields/{field_key}")
def delete_shared_field(table_key: str, field_key: str,
                        session: Session = Depends(require_session)):
    """Remove a TENANT-WIDE column and every value in it.

    ⛔ WHY THIS EXISTS AT ALL, said plainly: W30-T28 shipped the door that CREATES a shared column
    and none that removes one, so a column anybody added was permanent for the whole tenant. The
    reachability gate found it from the other end — `shared_overlay.drop_field` was complete,
    correct, gated, and callable by nothing but its own gate ([[reachable-is-not-the-same-as-built]]).

    ⛔ AND THIS ONE IS CREATOR-OR-ADMIN, WHICH THE WRITE DOOR IS NOT. Writing a cell changes a
    value; dropping the column deletes that value for EVERY account at once, so it is the
    destructive-op wall this repo already uses for a database delete — not `editRole`, which
    governs renaming and is not a value wall ([[schema-role-is-not-a-value-wall]]).
    ⚠ `createdBy` is stamped by the write door above; a column stored before that stamp existed
    is admin-only, which is the safe direction.
    """
    _defn_or_refuse(session, table_key)
    from core import shared_overlay
    defn = (shared_overlay.fields(table_key, st=session.runtime) or {}).get(str(field_key))
    if not defn:
        raise err(404, "unknown_field", "that column is not a shared column on this database")
    owner = str(defn.get("createdBy") or "")
    if not session.admin and owner != session.uname:
        raise err(403, "forbidden",
                  f"a tenant-wide column can be removed by its creator or an admin — this one "
                  f"was added by {owner or 'somebody else'}, and dropping it would delete the "
                  f"value for every account")
    dropped = shared_overlay.drop_field(table_key, str(field_key), st=session.runtime)
    return {"ok": True, "dropped": bool(dropped),
            "fields": list(shared_overlay.fields(table_key, st=session.runtime))}


def scoped_pids(session: Session, table_key: str, limits=None):
    """`(pids, fields_base, defn)` — the SAME wall and the SAME row set as `scoped_pool`, without
    building a row.

    ⭐⭐ WAVE 30 / W30-T30 — THIS IS WHY ONE HIDE-FIELDS CHECKBOX WAS EXPENSIVE. A view write
    (`view_upsert`) reaches `grid_events_route`, which built a FULL assembly purely to validate
    it: `scoped_pool` allocates a fresh dict per row and then sorts them — ~33k order rows, on
    every toggle — and the six keys the events route actually reads from that assembly
    (`fields`, `pids`, `measures`, `measure_sets`, `lists`, `views`) contain no row at all.
    `rows_src` was computed and discarded.

    ⛔ THE PID SET IS IDENTICAL, NOT MERELY EQUIVALENT, and that is the whole safety argument:
    `scoped_pool` derives its pids as `frozenset(r["pid"] for r in rows_src)` over exactly the
    row ids that pass `str(rid).isdigit()`, which is this comprehension with a dict build in the
    middle. The row WALL is unchanged — a narrower or wider set here would be a permission
    change, and this is a performance change.

    ⚠ It does NOT make the write cheap on its own: `_defn_or_refuse` still costs two whole-document
    deep copies (`ut.get` then `may_open`), which is D-87 and W30-T31. This removes the row pass.

    ⭐⭐ W31-T20 / D-174 — `limits` IS AN OUT-PARAMETER, AND IT IS THE POINT OF THE TICKET. Pass a
    list and this function APPENDS R6's sentence to it when the pid set could not be resolved (a
    read-through grid whose population exceeds one window). The pid set is then EMPTY, and every
    consumer of an empty pid set is fail-closed — but "fail-closed and unannounced" is exactly the
    silent limit R6's second sentence forbids, so a caller that renders an envelope or admits a
    write is expected to carry the sentence through. Omitting the list means the caller accepts an
    unannounced empty scope, which is only ever right for a caller that does not use the pids.
    """
    defn = _defn_or_refuse(session, table_key)
    fields_base = [dict(f) for f in (defn.get("fields") or [])]
    # ⚠ W30-T31: on a read-through database the stored `rows` is `{}` by construction, so the
    # comprehension below would answer an EMPTY pid set — and the promise this function makes is
    # that its set is IDENTICAL to `scoped_pool`'s, not merely cheaper. It reaches the mirror
    # through the SAME fetch that function uses rather than growing a second idea of the row set;
    # the saving W30-T30 bought stays on every materialised table, which is all of the big ones.
    if not _ut().materialises(table_key, st=session.runtime, defn=defn):
        try:
            rows = _read_through_rows(table_key, {f["key"] for f in fields_base},
                                      rt=session.runtime)
            return frozenset(r["pid"] for r in rows), fields_base, defn
        except _too_big() as e:
            # ⛔ THE REFUSAL BECOMES AN ANSWER HERE, WHICH IT MUST NOT ON THE ROWS PATH. Turning
            # this into a 409 is what made both line grids unopenable: the envelope was refused
            # over rows it never renders. The scope is empty and SAID to be empty.
            if limits is not None:
                limits.append({**_PID_SCOPE_LIMIT, "cause": str(e)})
            return frozenset(), fields_base, defn
        except RuntimeError as e:
            raise err(503, "store_not_ready", str(e))
    pids = frozenset(int(rid) for rid in (defn.get("rows") or {}) if str(rid).isdigit())
    return pids, fields_base, defn


def ut_write_ctx(session: Session, table_key: str):
    """The g-dict a WRITE needs — same keys as `ut_assembly`, no rows.

    Returns the six keys `routes_grid.grid_events_route` reads, so the events route consumes this
    or a full assembly interchangeably. `rows_src` is `[]` on purpose rather than absent: a caller
    that starts needing rows should fail on an empty list it can see, not on a KeyError.
    """
    import aios_grid
    from core import grid_events

    limits = []
    pids, fields_base, defn = scoped_pids(session, table_key, limits=limits)
    ctx = grid_events.EventCtx(
        uname=session.uname, allowed_pids=pids, fields=[], hidden_keys=frozenset(),
        admin=session.admin, fallback_ws=None, seen_ids={}, st=session.runtime,
        scope_key=table_key, table=_ops(session, table_key))
    ws = grid_events.table_workspace(ctx, allowed_pids=pids, consume_corrections=False)
    workspace, fields, views, lists = aios_grid.workspace_wire(
        ws, session.uname, set(pids), defs={}, scope_key=table_key, storage_key="",
        fields_base=fields_base)
    return {"rows_src": [], "pids": pids, "ws": ws, "workspace": workspace,
            "fields": fields, "views": views, "lists": lists,
            "derived": aios_grid.cohort_cells(lists),
            "measures": [], "measure_sets": {}, "today": time.strftime("%Y-%m-%d"),
            # ⭐ W31-T20 — the write door reads this to refuse a PID-BEARING event loudly rather
            # than letting `allowed_pids` swallow it as a no-op. See `routes_grid`'s ut_ branch.
            "limits": limits, "defn": defn}


def ut_assembly(session: Session, table_key: str, storage_key: str = "",
                consume_corrections: bool = True, with_rows: bool = True):
    """The user-table mirror of `grid_assembly` / `product_assembly` — SAME g-dict keys, so
    `/workspace` and the events route consume any of the three interchangeably.

    Honest absence: `measures`/`measure_sets` are EMPTY — `core.measure_resolve` is
    customer-grain, so there is nothing to offer over user rows.

    ⭐ WAVE 19 / R9 — `lists` IS NO LONGER EMPTY. "For ANY database new/old": a user table gets
    cohorts like every other database, out of its OWN bucket (`ut_<slug>_cohorts`), holding its
    own row ids. The wave-18 refusal was correct while there was one customer-keyed bucket and
    wrong the moment the store learned about topics.

    ⭐⭐ W31-T20 / D-174 — `with_rows=False` BUILDS THE ENVELOPE AND NOT THE TABLE, and the
    caller that wants it is `/workspace`, which renders no row at all (the grid fetches rows from
    `/tables/{key}/rows` or `/odoo-tables/{key}/rows` beside it). Two things follow:
      * the read-through line grains become OPENABLE — `scoped_pool` refused their envelope over
        963,783 rows nobody was going to look at, which is D-174 in one sentence;
      * every materialised `ut_*` database stops allocating a dict per row and sorting them on a
        route whose payload has no rows in it — `ut_odoo_orders` was rebuilding 32,826 of them per
        database switch (owner item 7).
    ⛔ NOTHING IS VALIDATED LESS. `scoped_pids` runs the SAME `_defn_or_refuse` wall and answers
    the identical pid set; the flag removes work, never a check — the shape W30-T30 already proved
    on the write door.
    """
    import aios_grid
    from core import grid_events

    limits = []
    if with_rows:
        pids, rows_src, fields_base, defn = scoped_pool(session, table_key)
    else:
        pids, fields_base, defn = scoped_pids(session, table_key, limits=limits)
        rows_src = []

    ops = _ops(session, table_key)
    ctx = grid_events.EventCtx(
        uname=session.uname, allowed_pids=pids, fields=[], hidden_keys=frozenset(),
        admin=session.admin, fallback_ws=None, seen_ids={},
        # R6b (D-16): the tenant handle rides every ctx this layer builds, not just the ones
        # that happen to carry a scoped `table`.
        st=session.runtime,
        scope_key=table_key, table=ops)
    ws = grid_events.table_workspace(ctx, allowed_pids=pids,
                                     consume_corrections=consume_corrections)
    workspace, fields, views, lists = aios_grid.workspace_wire(
        ws, session.uname, set(pids), defs={}, scope_key=table_key,
        storage_key=storage_key, fields_base=fields_base)

    return {"rows_src": rows_src, "pids": pids, "ws": ws, "workspace": workspace,
            "fields": fields, "views": views, "lists": lists,
            # R9: the Cohorts column's cells from this table's own lists.
            "derived": aios_grid.cohort_cells(lists),
            "measures": [], "measure_sets": {}, "today": time.strftime("%Y-%m-%d"),
            # ⚠ ALWAYS PRESENT, EMPTY WHEN THERE IS NOTHING TO SAY — a key a consumer has to test
            # for is a key a consumer forgets to test for, and this one carries a refusal.
            "limits": limits, "defn": defn}


def ut_label(defn, key, meta=None):
    """THE name of a user table, resolved ONCE (wave 20, item 6a).

    `nav_meta`'s rename wins, then the definition's own label, then the key. Every surface that
    shows a database name reads through here, because the alternative is what wave 20 found: the
    rail showed the renamed name (it reads `nav_meta`) while the automation editor's picker
    showed the original (it reads the definition), and neither looked broken.

    ⚠ `set_label` now writes the DEFINITION too, so the two agree at the source. This resolver
    stays because it makes every row already stored — renamed before that fix landed — read
    correctly today, without a migration.
    """
    return ((meta or {}).get(key, {}).get("name")
            or (defn or {}).get("label") or key)


def nav_meta(session):
    """The tenant's nav_meta bucket, read defensively. A store blip must not take a list down."""
    try:
        got = session.runtime.get("nav_meta")
        return got if isinstance(got, dict) else {}
    except Exception:                                             # noqa: BLE001
        return {}


@router.get("/tables")
def list_tables(session: Session = Depends(require_session)):
    """This session's user tables — the list the '+ New database' surface renders.

    ⭐⭐ W31-T12 (contract C1, D-175's third instance) — ONE DOCUMENT READ, NOT `1 + 2N`. This
    route was never ticketed and has the same shape `/nav` and `/automations` were fixed for:
    `all_tables` once, then `may_open` per key (another whole-document deep copy each, 28.6 MB on
    tenant #0 measured, 703 ms warm) and `records_mutable` per key on top of that — so a tenant
    with ten databases paid twenty-one copies to list them. The wall is UNCHANGED and still asked
    about every table; it is handed the document this function already holds. See
    `user_tables.lend`'s own note for why inlining the predicate is the one fix that is not
    available.
    """
    ut = _ut()
    meta = nav_meta(session)
    out = []
    tables = ut.all_tables(st=session.runtime)
    lent = ut.lend(session.runtime, **{ut.STORE_KEY: tables})
    for key, t in sorted(tables.items(),
                         key=lambda kv: (ut_label(kv[1], kv[0], meta) or "").lower()):
        if not ut.may_open(key, session.uname, session.admin, st=lent):
            continue
        out.append({"key": key, "label": ut_label(t, key, meta),
                    "source": t.get("source") or "Blank",
                    "recordsMutable": ut.records_mutable(key, st=lent),
                    "createdBy": t.get("createdBy") or "",
                    "created": t.get("created") or "",
                    "fields": [dict(f) for f in (t.get("fields") or [])],
                    "rowCount": len(t.get("rows") or {})})
    return {"tables": out}


#: ⭐⭐ THE ROLLUP SOURCE OFFER — what makes the read-through rollup a FEATURE rather than a
#: capability. Owner 2026-08-09: *"Full field editor: pick topic → metric → window."*
#:
#: ⛔ THE ENGINE SHIPPED WITH NO WRITER. `_clean_rollup` has accepted a `source` bag since
#: 2026-08-09 and `rollup_sql` answers it in one grouped query, but the ONLY thing in the product
#: that ever produced one was a hard-coded field in `odoo_relational.py`. Nothing in the client
#: could make one, so the whole read-through path was reachable by editing Python — the
#: [[artifact-with-no-importer]] shape, twice burned in this repo already.
#:
#: ⚠ DERIVED FROM THE MODEL FILES, NEVER A HAND-WRITTEN LIST. `model/topics/*.yml` and
#: `model/metrics/*.yml` already state which measures belong to which topic and which dims that
#: topic can group by; a second list here would be a second definition of the same fact, and the
#: two would drift the day somebody adds a metric. Same argument `rollup_sql` makes for binding
#: to a metric KEY instead of carrying SQL.
_ROLLUP_CACHE = {}


def _rollup_source_offer():
    """`{topics:[…], windows:[…]}` — every (topic, measure, dim) the engine can actually answer.

    ⛔ ONLY COMBINATIONS THAT CAN RESOLVE ARE OFFERED, because a rollup that refuses at COMPUTE
    time refuses silently — the cells are simply left blank, hours later, on a column that looks
    configured. Two exclusions do real work:
      * a topic with NO dims cannot be grouped at all, so it can never key a parent row;
      * a CROSS-TOPIC metric (`aov`, `margin_pct`, `returns_pct` — `agg: ratio`/`derived` whose
        inputs live elsewhere) is refused by `store_query` the moment a `group_by` is present:
        *"cross-topic measures are scalar-only"*. Offering one would mint a column that can only
        ever error.
    Each dim also declares HOW it keys — by an Odoo id or by its own value — because that is what
    the user is matching their own column against, and `payment_state` (a value) and
    `partner` (an id) are matched to very different columns.
    """
    if _ROLLUP_CACHE.get("offer"):
        return _ROLLUP_CACHE["offer"]
    from harness import semantic as sem
    from harness import windows as W
    ut = _ut()

    topics, metrics = sem.topics(), sem.metrics()
    by_topic = {}
    for key, m in metrics.items():
        # A ratio/derived metric whose parts sit on another topic cannot be grouped — see above.
        if m.get("agg") in ("ratio", "derived"):
            continue
        by_topic.setdefault(m["topic"], []).append(
            {"key": key, "label": m.get("label") or key, "format": m.get("format") or "usd",
             "description": m.get("description") or ""})

    out = []
    for tkey, t in sorted(topics.items()):
        dims = ((t.get("store") or {}).get("dims") or {})
        measures = by_topic.get(tkey) or []
        if not dims or not measures:
            continue
        out.append({
            "key": tkey,
            "label": t.get("label") or tkey,
            "grain": t.get("grain") or "",
            "dims": [{"key": dkey,
                      "label": d.get("label") or dkey,
                      # `store_query` emits `<dim>_id` only when the dim carries a display name
                      # alongside the key; otherwise the value IS the key. `rollup_sql` handles
                      # both, and the editor says which so the user matches the right column.
                      "keyedBy": "id" if d.get("name_col") else "value"}
                     for dkey, d in dims.items()],
            "measures": sorted(measures, key=lambda m: m["label"].lower()),
        })

    # ⚠ THE WINDOW LIST IS `core.user_tables`' OWN, not `harness.windows`'. `ROLLUP_SOURCE_WINDOWS`
    # is the validator's closed set and is deliberately NARROWER (it omits the parameterised kinds
    # like `last_n_days`, which have nowhere in the bag to carry their `n`). Offering a kind the
    # validator refuses would let the editor build a field the save door rejects.
    windows = [{"key": k, "label": W.WINDOW_LABELS.get(k, k).format(n="N")}
               for k in ut.ROLLUP_SOURCE_WINDOWS]
    offer = {"topics": out, "windows": windows}
    _ROLLUP_CACHE["offer"] = offer
    return offer


@router.get("/tables/rollup-sources")
def rollup_sources(session: Session = Depends(require_session)):
    """The topic → metric → dim → window offer the rollup field editor renders.

    ⚠ DECLARED ABOVE EVERY `/tables/{table_key}/…` ROUTE, and kept there deliberately. FastAPI
    matches in declaration order, so the day somebody adds a bare `GET /tables/{table_key}` below
    this line it still resolves; added ABOVE it, this endpoint would silently start arriving as
    `table_key='rollup-sources'` and 404 from the table wall. There is no such route today —
    this is the cheap ordering that keeps it from mattering.

    ⛔ TENANT-SCOPED, AND IT WAS NOT WHEN FIRST WRITTEN. `sem.topics()` reads the GLOBAL model
    files, so nurilab and gtmlab were served the full Odoo offer — they would have seen "Live
    Odoo data" in the field editor and been able to build a column that can only ever be blank,
    because there is no mirror behind it. Three comments (here, in `apiBridge` and on the
    `rollupSourceOffer` prop) each asserted that a tenant with nothing connected receives `[]`,
    and the mode switch's "render only when there is a choice" guard is built on that promise.
    ⚠ `odoo_relational.is_royal` is the authority, reused rather than re-decided: it is already
    what `refresh` consults to decide whether these tables may exist at all, and a second copy of
    the rule would be a second answer the day a tenant gains a mirror.
    """
    import odoo_relational
    if not odoo_relational.is_royal(session.tenant):
        return {"topics": [], "windows": []}
    return _rollup_source_offer()


@router.post("/tables", status_code=201)
def create_table(body: dict = Body(default=None),
                 session: Session = Depends(require_session)):
    ut = _ut()
    body = body or {}
    label = str(body.get("label") or "").strip()
    if not label:
        raise err(400, "bad_label", "give the database a name")
    if not session.runtime.available():
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — nothing was created")
    source = body.get("source")
    try:
        key = ut.create(label, session.uname, fields=body.get("fields"),
                        source=source, st=session.runtime)
    except Exception:
        raise err(503, "store_unavailable",
                  "the tenant store refused the write — nothing was created")
    if not key:
        raise err(400, "refused",
                  f"could not create it — the name may be empty or this tenant already has "
                  f"{ut.MAX_TABLES} databases")
    return {"key": key}


@router.patch("/tables/{table_key}")
def patch_table(table_key: str, body: dict = Body(default=None),
                session: Session = Depends(require_session)):
    """Rename a database — IN ITS DEFINITION (wave 20, item 6a).

    ⚠ THE RENAME DOOR IN THE NAV WRITES `nav_meta` AND MUST ALSO CALL THIS. `nav_meta` is the
    nav's display layer; the definition is what the automation editor's database picker, the
    schema drawer and every future reader see. A rename that lands in only one of them leaves a
    picker that is confidently wrong rather than obviously stale. Posted to the wave doc as an
    amendment for whoever owns that door.
    """
    defn = _defn_or_refuse(session, table_key)
    ut = _ut()
    if not (session.admin or defn.get("createdBy") == session.uname):
        raise err(403, "forbidden", "only the database's creator or an admin can rename it")
    label = ut.set_label(table_key, (body or {}).get("label"), st=session.runtime)
    if not label:
        raise err(400, "bad_label", "give the database a name")
    return {"key": table_key, "label": label}


@router.get("/tables/{table_key}/footprint")
def table_footprint(table_key: str, session: Session = Depends(require_session)):
    """What dies with this database — the confirm dialog's disclosure (wave 21, item 6a / C3).

    Counts drill to the SAME buckets `user_tables.delete` cleans; a dialog listing categories
    without numbers would break [[no-unverifiable-aggregates]] at the scariest moment. Walled
    like the delete itself: only someone who could delete may case the joint."""
    defn = _defn_or_refuse(session, table_key)
    if not (session.admin or defn.get("createdBy") == session.uname):
        raise err(403, "forbidden", "only the database's creator or an admin can delete it")
    s = session.runtime
    views, fields = set(), len(defn.get("fields") or [])
    try:
        bucket = s.get(f"{table_key}_table_workspace") or {}
        for _u, ws in bucket.items():
            if isinstance(ws, dict):
                views |= set((ws.get("views") or {}).keys())
                fields += len(ws.get("fields") or {})     # per-user custom/measure strata
    except Exception:
        pass
    import core.shares as shares
    g = shares.grants("database", table_key, st=s)
    auto = []
    try:
        import automation_engine as engine
        for aid, d in (engine.all_definitions(s) or {}).items():
            if (d.get("config") or {}).get("targetTable") == str(table_key):
                auto.append({"id": str(aid), "name": d.get("name") or str(aid)})
    except Exception:
        pass
    return {"rows": len(defn.get("rows") or {}), "fields": fields, "views": len(views),
            "sharedUsers": len(g.get("entries") or []),
            "automations": sorted(auto, key=lambda a: a["name"].lower())}


@router.delete("/tables/{table_key}")
def delete_table(table_key: str, session: Session = Depends(require_session)):
    """CREATOR OR ADMIN — checked explicitly (wave 21, item 6a / C3).

    ⛔ The wave-20 docstring said "the same actors may_open admits" and that stopped being the
    creator-or-admin set the day de5037f taught `may_open` to admit share GRANTEES: a view-role
    grantee could reach this route and delete the database somebody shared with them. The wall
    is now the definition's own `createdBy`, the same check the rename route always had.

    Deletion cleans the artifact families server-side (`user_tables.delete` lists them) and
    DISABLES bound automations with a status note — never deletes them. The client's confirm
    dialog disclosed `/footprint` first; the server cannot tell a click from a plan, so the
    dialog is a product requirement, not a formality."""
    defn = _defn_or_refuse(session, table_key)
    if not (session.admin or defn.get("createdBy") == session.uname):
        raise err(403, "forbidden", "only the database's creator or an admin can delete it")
    try:
        import automation_engine as engine
        engine.disable_for_table(session.runtime, table_key)
    except Exception:
        pass
    try:
        _ut().delete(table_key, st=session.runtime)
    except Exception:
        raise err(503, "store_unavailable", "the delete did not land — try again")
    return {"ok": True}


#: ⭐ 2026-08-07 — tenants whose Instagram tables THIS PROCESS has already brought forward.
_IG_FORWARDED = set()


def _ig_forward(session):
    """Bring this tenant's Instagram tables onto the current schema, at most once per process.

    ⛔ WHY A MIGRATION RUNS ON A READ AT ALL, when the module's own rule is that it rides the WRITE
    path. `ut_ensure` calling it is right for a schema an automation is about to append to, and
    useless for a change a PERSON is waiting to see: the owner's report was *"the first field is
    still blank"*, and "re-save the automation and it will fix itself" is not an answer to that.
    The write path stays exactly as it was — this is a second door to the same idempotent call,
    not a replacement for it.

    ⚠ BOUNDED THREE WAYS, because a write on a read is otherwise how a grid gets slow: once per
    tenant per process; `migrate_ig_tables` returns without a write when every table is already
    current (the common case after the first read); and a failure is SWALLOWED — a migration must
    never be the reason a database will not open.

    ⚠ THE TENANT IS MARKED BEFORE THE ATTEMPT, deliberately. A migration that raises must not be
    retried on every subsequent read of every table for the life of the process — the write path is
    still the backstop, so the cost of skipping is a delay, while the cost of retrying is a failing
    store call on the hot path of a grid that is trying to render.
    """
    tenant = str(getattr(session, "tenant", "") or "")
    if tenant in _IG_FORWARDED:
        return
    _IG_FORWARDED.add(tenant)
    try:
        import automation_engine as engine
        engine.migrate_ig_tables(session.runtime, log=lambda *_a: None)
    except Exception as e:                                            # noqa: BLE001
        print(f"[tables] ig forward-migration skipped: {type(e).__name__}: {e}")



#: Above this many characters a `json` cell is replaced by a stand-in in the LIST envelope. Sized
#: so an ordinary config document (a few hundred bytes) is untouched while a vendor response is
#: not — the shape this exists for is one already-paid provider payload per row.
JSON_LIST_MAX = 400


def _thin_json(fields, merged, table_key=""):
    """Replace oversized `json` cells with a stand-in for the LIST response. Pure; returns a copy.

    ⚠ THE STAND-IN IS ITSELF VALID JSON and carries the byte count, so the grid preview reads
    `{...} 3 keys` rather than a broken brace, and a reader can see the column holds something
    large rather than something empty. `_truncated` is what the viewer keys its fetch on.

    ⭐ THE STAND-IN CARRIES ITS OWN `_url`, which is what keeps this change small AND correct: the
    viewer needs no table key, no record id and no new props threaded down through three
    components to find the document — the route that removed the value says where it went. One
    writer of that address instead of a server rule and a client rule that must agree forever.
    """
    json_keys = [str(f.get("key")) for f in (fields or [])
                 if str(f.get("type") or "") == "json"]
    if not json_keys:
        return merged
    out = {}
    for pid, cells in (merged or {}).items():
        row = cells
        for key in json_keys:
            raw = cells.get(key)
            if isinstance(raw, str) and len(raw) > JSON_LIST_MAX:
                if row is cells:
                    row = dict(cells)
                row[key] = json.dumps({
                    "_truncated": True, "bytes": len(raw),
                    "_url": f"/api/v1/tables/{table_key}/rows/{pid}/fields/{key}"})
        out[pid] = row
    return out


@router.get("/tables/{table_key}/rows")
def table_rows(table_key: str, session: Session = Depends(require_session)):
    """The `/customers`-shaped envelope for one user table: `{fields, rows, today, pulled_at,
    identity}` — so the client's generic topic fetch consumes it with zero new parsing.

    ⚠ THE MERGE ORDER IS THE CONTRACT. `rows_from_pool` sources an overlay-typed field's cell
    from the OVERLAY stratum only (that is what makes a custom column render standalone) — a
    user table's base values live in its DEFINITION rows, so they are layered UNDER the user's
    overlay edits here: base first, overlay wins. Without this every base cell reads empty
    (found by this route's own gate check, not by luck)."""
    import aios_grid

    _ig_forward(session)
    g = ut_assembly(session, table_key)
    # ⛔ THE OVERLAY WAS NEVER ACTUALLY MERGED, and the docstring above has described the merge
    # this line does not perform since the route was written (owner item 3, 2026-08-09:
    # *"Using the swipe, fast doesn't register the CHANGE. I went back and it all got reseted"*).
    #
    # `merged` was built from `rows_src` ALONE — the DEFINITION rows. But `rows_from_pool`
    # sources an overlay-typed field's cell from this dict, and a CUSTOM column on a `ut_*` table
    # is overlay-typed by construction, so it looked up a key that could not be there and every
    # such cell rendered blank.
    #
    # ⭐ THE WRITES WERE NEVER LOST — MEASURED. `ws['overlays']` holds
    # `{"1": {"custom_geography_yf6vi": "Jakarta"}, ...}` for ten rows: the owner's swipes landed
    # in the store exactly as they should. Only the READ-BACK dropped them, which is why the
    # value survived the gesture, vanished on reload, and looked like "it reset itself" — and
    # why `patch_row`'s `_took()` then reported a perfectly good write as `refused`.
    #
    # ⚠ OVERLAY WINS, base underneath — the order the docstring already specifies. A definition
    # value must not shadow an edit the user has made on top of it.
    # ⚠ AND IT IS THIS USER'S OWN OVERLAY (`table_workspace` is keyed by `ctx.uname`), so this
    # widens what a caller can SEE by exactly their own edits and nothing else.
    _overlays = (g.get("ws") or {}).get("overlays") or {}
    merged = {}
    for _r in g["rows_src"]:
        _pid = str(_r["pid"])
        _cells = {k: v for k, v in _r.items() if k != "pid"}
        _ov = _overlays.get(_pid)
        if isinstance(_ov, dict):
            _cells.update(_ov)
        merged[_pid] = _cells
    # ⭐⭐ 2026-08-10 (owner: *"wth is going on, why does it take forever to load now?"*) — THE
    # JSON DOCUMENTS DO NOT RIDE THE LIST.
    #
    # MEASURED on nurilab, and the numbers are the whole argument: `source_payload` is **95.6%
    # to 98.5%** of every IG grid's bytes — `ut_ig_post_snapshots` shipped **11.6 MB of a 12.2 MB
    # response**, `ut_ig_snapshots` 1.85 MB of 2.03 MB, the profile table 1.83 MB of 2.11 MB. The
    # grid renders those cells as a SIXTY-CHARACTER preview (`display.jsonPreview`), so the whole
    # vendor response crossed the wire, was parsed by the browser and held in memory purely so a
    # clipped first line could be drawn.
    #
    # ⚠ IT IS NOT A CAP AND NOTHING IS LOST. The full document is served by
    # `GET /tables/{key}/rows/{pid}/fields/{fkey}`, which the JSON viewer fetches when it opens —
    # the one place a person actually reads it, for the one row they opened. The cell that rides
    # the list is a VALID small document saying what it stands for, so the preview renders
    # honestly instead of showing half a truncated brace.
    # ⛔ ONLY `json` COLUMNS, and only over the threshold: a small document still travels whole,
    # so a tenant using `json` for a short config sees no change at all.
    rows = aios_grid.rows_from_pool(
        g["rows_src"], g["fields"], _thin_json(g["fields"], merged, table_key), derived=g["derived"])
    # ⭐ R6's SECOND SENTENCE, ON THE WIRE (W30-T29). *"if there is lag or it can't be done, you
    # need to explicitly tell me why and recommend a fix."* A ceiling that still applies to this
    # database says so here, with its cause and the recommendation, rather than waiting to be
    # discovered as a refused paste. `None` for a connected source, and an EMPTY LIST is the
    # honest answer for a table nothing limits — never an absent key, which a client cannot tell
    # apart from an older server.
    _report = _ut().limit_report(table_key, st=session.runtime)
    # ⭐ C4 / D-138 — THE DOCUMENTS PRODUCER. Absent since EXIT-6 deleted `app.py`, which was the
    # only thing that ever set this key; the write door never stopped working and every client
    # half is complete, but all six `onDoc*` handlers read `payload?.docs ? … : undefined`, so a
    # missing key has been silently switching the feature off. ⛔ ONE shared serialiser with the
    # customer scope — `grid_events.docs_for` — never a twin here.
    from core import grid_events as _ge
    return {"fields": g["fields"], "rows": rows, "today": g["today"],
            "docs": _ge.docs_for(g["pids"], scope_key=table_key, uname=session.uname,
                                 admin=session.admin, st=session.runtime),
            "pulled_at": time.strftime("%Y-%m-%d %H:%M"),
            "identity": {"pid": "pid"},
            "scope": {"table": table_key, "rowCount": len(rows)},
            "limits": [_report] if _report else [],
            "recordsMutable": _ut().records_mutable(table_key, st=session.runtime)}


@router.get("/tables/{table_key}/rows/{pid}/fields/{fkey}")
def table_cell(table_key: str, pid: str, fkey: str,
               session: Session = Depends(require_session)):
    """ONE cell, whole — the other half of `_thin_json`.

    ⛔ WITHOUT THIS THE THINNING WOULD BE A CAP, and a cap on data somebody already paid a vendor
    for is exactly what this product refuses everywhere else. The list ships a stand-in; the JSON
    viewer opens this for the one row a person is actually reading.

    ⚠ SAME PERMISSION WALL AS THE LIST, reached the same way (`ut_assembly` resolves the session's
    view of the table), so this cannot become a side door onto a table the caller may not open —
    which is the failure a "just fetch the raw cell" helper invites.
    ⚠ THE OVERLAY WINS, exactly as it does in `table_rows`: a user who has typed over a cell must
    read back what they typed, not the definition value underneath it.
    """
    g = ut_assembly(session, table_key)
    field = next((f for f in (g.get("fields") or []) if str(f.get("key")) == str(fkey)), None)
    if field is None:
        raise err(404, "unknown field", f"{fkey!r} is not a column on this database")
    row = next((r for r in g["rows_src"] if str(r.get("pid")) == str(pid)), None)
    if row is None:
        raise err(404, "unknown record", f"no record {pid!r} in this database")
    overlay = ((g.get("ws") or {}).get("overlays") or {}).get(str(pid)) or {}
    value = overlay.get(fkey, row.get(fkey))
    return {"table": table_key, "pid": str(pid), "field": str(fkey),
            "value": "" if value is None else str(value)}


@router.post("/tables/{table_key}/rows", status_code=201)
def add_row(table_key: str, body: dict = Body(default=None),
            session: Session = Depends(require_session)):
    """Append a row — or RESTORE one under its old id (contract C-ADDROW / C-UNDO).

    ⚠ THE ANSWER IS ALWAYS THE ID THAT WAS STORED, never the one that was asked for. An undo
    that requested `rid: 7` and got 12 because 7 had been re-used must find that out from the
    response rather than assume; the client re-anchors on what came back.
    """
    _records_or_refuse(session, table_key)
    ut = _ut()
    values = (body or {}).get("values") or {}
    if not isinstance(values, dict):
        raise err(400, "bad_values", "values must be an object of {fieldKey: value}")
    try:
        rid = ut.add_row(table_key, values, session.uname, st=session.runtime,
                         rid=(body or {}).get("rid"))
    except Exception:
        raise err(503, "store_unavailable", "the row was not saved — the store refused")
    if rid is None:
        # C3 (wave 25): `add_row` also refuses a profile cell that is not a handle, so the cap
        # sentence alone would misdirect — the reader would go and count rows. Ask the same
        # validator the law used rather than re-deciding here (one rule, two voices).
        pf = ut.profile_field(table_key, st=session.runtime)
        if pf and pf["key"] in values:
            _h, ok = ut.normalize_profile(values[pf["key"]], pf["profile"].get("source"))
            if not ok:
                raise err(400, "refused",
                          f"{str(values[pf['key']])[:80]!r} is not an Instagram profile — "
                          f"{pf.get('label') or pf['key']!r} takes a handle (@name) or a "
                          f"profile link (instagram.com/name)")
        raise err(400, "refused",
                  f"row refused — the table may be at its {ut.MAX_ROWS}-row cap")
    _refresh_relations(session)
    return {"rid": rid, "pid": int(rid)}


@router.post("/tables/{table_key}/rows/import", status_code=201)
def import_rows(table_key: str, body: dict = Body(default=None),
                session: Session = Depends(require_session)):
    """⭐⭐ WAVE-29 T25 (owner item 6) — the IMPORT door: N mapped rows, ONE store write.

    Body: `{"rows": [{fieldKey: value, ...}, ...]}` — already MAPPED by the client's dialog, so a
    spreadsheet column name never reaches the store. APPEND-ONLY in v1: every row is a new record,
    nothing is matched or overwritten, and the dialog says so before the button is pressed.

    ⛔ COMPUTED COLUMNS ARE REFUSED HERE, NOT FILTERED. `is_computed_cell` is the same predicate
    the cell wall uses (one evaluator for one question), and a rollup or formula key arriving in
    an import is not a stray to be tidied away — it means the client offered a target it should
    not have, and silently dropping it would leave the user looking for a column of values that
    never arrived. The refusal names the column.

    ⛔ ATOMIC. `add_rows` writes nothing unless the whole batch fits under `MAX_ROWS` and every
    profile cell validates, because a half-imported file is the worst outcome available: the user
    cannot tell which rows landed without reconciling the spreadsheet by hand.
    """
    _records_or_refuse(session, table_key)
    ut = _ut()
    rows_in = (body or {}).get("rows")
    if not isinstance(rows_in, list) or not rows_in:
        raise err(400, "bad_rows", "rows must be a non-empty array of {fieldKey: value} objects")
    if any(not isinstance(r, dict) for r in rows_in):
        raise err(400, "bad_rows", "every row must be an object of {fieldKey: value}")
    defn = _defn_or_refuse(session, table_key)
    by_key = {f["key"]: f for f in (defn.get("fields") or [])}
    asked = {k for r in rows_in for k in r}
    unknown = sorted(k for k in asked if k not in by_key)
    if unknown:
        raise err(400, "unknown_field",
                  f"this database has no column {unknown[0]!r}")
    computed = sorted(k for k in asked if ut.is_computed_cell(by_key[k]))
    if computed:
        label = by_key[computed[0]].get("label") or computed[0]
        raise err(400, "computed_field",
                  f"{label!r} is worked out from other columns, so it cannot be imported into")
    # ⛔ W29-T81 — THE TYPE WALL, AND IT LIVES HERE BECAUSE THE ONLY OTHER ONE IS IN THE BROWSER.
    # `coerceClipboardValue` refuses "seventeen-ish" at an `int` column in the dialog; curl, a
    # second client, or a future importer met no wall at all and the string landed verbatim in a
    # typed column, where the grid then painted it as a fabricated `0`. Refused whole and BEFORE
    # `add_rows`, matching this door's own atomicity: a half-imported file is the outcome every
    # rule on this route exists to prevent. The sentence names the row and the column, because
    # "invalid value" sends somebody hunting through 2,000 lines of spreadsheet.
    for index, row in enumerate(rows_in):
        for key, value in row.items():
            why = ut.cell_type_refusal(by_key[key], value)
            if why:
                raise err(400, "bad_value", f"row {index + 1}: {why} — nothing was imported")
    try:
        made = ut.add_rows(table_key, rows_in, session.uname, st=session.runtime)
    except Exception:
        raise err(503, "store_unavailable", "nothing was imported — the store refused")
    if made is None:
        raise err(400, "refused",
                  f"nothing was imported — {len(rows_in)} rows would take this database past "
                  f"its {ut.MAX_ROWS}-row cap, or a profile column rejected a value")
    _refresh_relations(session)
    return {"imported": len(made), "pids": [int(r) for r in made]}


# ---------------------------------------------------------------------------------------------
# THE SHARED FIELD SCHEMA (contract C-FIELD, owner ruling R2)
# ---------------------------------------------------------------------------------------------
# R2: a `ut_*` table's fields are the TABLE'S schema — everyone with access sees the same
# columns, the creator or an admin edits them, and a per-field `editRole` can open ONE column's
# definition to everyone without handing over the table. This supersedes wave 17's "fields are
# per-user" law for this path only; the connector scopes keep their own model.
#
# ⚠ WHY IT MATTERS BEYOND TIDINESS: a grid add-field on a `ut_` scope used to land in the
# per-user workspace overlay, which is why the automation editor's "Automation column" picker
# could not see a column the user had just created — it reads the DEFINITION. Same defect shape
# as the rename (item 6a): two places to look, and the surfaces disagreed silently.

def _field_or_refuse(session, table_key, fkey=""):
    """The schema wall. `_defn_or_refuse` first (404/403 on the table), then the per-field rule."""
    defn = _defn_or_refuse(session, table_key)
    ut = _ut()
    if not ut.is_user_table(table_key, st=session.runtime):
        raise err(400, "not_a_user_table",
                  "only a user-created database has an editable schema — a connected source "
                  "owns its own columns")
    if fkey and not ut.may_edit_field(table_key, fkey, session.uname, session.admin,
                                      st=session.runtime):
        field = next((f for f in (defn.get("fields") or [])
                      if f.get("key") == str(fkey)), None)
        if isinstance((field or {}).get("automation"), dict) \
                and field["automation"].get("preset") is True:
            # ⚠ THIS BRANCH NARRATES `may_edit_field`, it does not re-decide (the `and not
            # ut.may_edit_field` above is the wall). Said out loud because the sentence itself
            # went stale on 2026-08-09: preset ROLLUPS became editable by owner ruling, so a
            # blanket "pre-set fields are locked" would now be the server explaining a refusal
            # it did not make — `preset_editable` is the one predicate that answers this.
            raise err(403, "preset_field_locked",
                      "this is a pre-set column, so its name and type are fixed; you may sort, "
                      "filter or hide it, edit any Rollup column, and add your own columns")
        raise err(403, "forbidden", "that column can only be changed by the database's creator "
                                    "or an admin")
    if not fkey and not (session.admin or defn.get("createdBy") == session.uname):
        raise err(403, "forbidden", "only the database's creator or an admin can add a column")
    return defn


@router.post("/tables/{table_key}/fields", status_code=201)
def add_field(table_key: str, body: dict = Body(default=None),
              session: Session = Depends(require_session)):
    _field_or_refuse(session, table_key)
    ut = _ut()
    field = ut.add_field(table_key, body or {}, st=session.runtime)
    if not field:
        # ⭐ D-46 CLOSED (wave 23) — the C8 flow law gets its OWN sentence. `add_field` answers
        # None for every refusal, so this route said "check the name and type" to somebody whose
        # name and type were fine and whose automation column named a flow that does not exist.
        # A refusal that misdirects is worse than a bare 400: it sends the reader to look at the
        # one thing that was never wrong. Checked HERE, in the route's own words, because the
        # law itself stays enforced in `user_tables.flow_bound` — this narrates it, never
        # re-implements it (a second copy of the rule is how two doors start disagreeing).
        raise err(400, "refused",
                  _refusal_sentence(ut, session, body or {}, table_key=table_key))
    if field.get("type") == "link":
        synced = ut.sync_reciprocal_link(table_key, field["key"], st=session.runtime)
        field = synced.get("field") or field
    # ⭐⭐ 2026-08-09 — `rollup` REFRESHES TOO, and the omission was invisible until this route
    # became reachable for one. It was gated on `link` alone, while `patch_field` and
    # `delete_field` next door refresh unconditionally — so a newly created Rollup got its first
    # fold from `_store_resync_loop`, which sleeps 1800 s BEFORE its first pass (D-107's shape).
    # The user would have created the column, watched a 201 come back, and read a blank cell for
    # half an hour: *"the Rollup doesn't work"*, arriving through the door opened to fix it.
    # ⚠ Still conditional rather than unconditional: a pass deep-copies every table and row in
    # the tenant, and adding a text column has nothing to fold. The condition is now "is this
    # field relational", which is the question that was always meant.
    if field.get("type") in ("link", "rollup"):
        _refresh_relations(session)
    return {"field": field}


def _refusal_sentence(ut, session, body, table_key="", fkey=""):
    """Why was this column refused? The specific reason when we can name one, the general list
    otherwise — never a specific-sounding guess."""
    # ⭐ C3 (wave 25, R7): the one-profile-per-table refusal NAMES THE EXISTING COLUMN, which is
    # what the contract asks for and what makes it actionable — "at most one" sends the reader
    # hunting through a 40-column schema for a flag they cannot see from the header.
    if isinstance((body or {}).get("profile"), dict):
        # ⚠ THE TYPE IS NAMED FIRST, and the order is the point. Both refusals can be true at
        # once (an `int` profile column on a table that already has a profile column), and the
        # TYPE is the one that is wrong about what the caller just sent — unconditionally, no
        # matter what else is on the table. Answering "you already have one" to somebody whose
        # real mistake was the column type sends them to fix the wrong thing, which is the
        # misdirection D-46 closed one door over.
        if str((body or {}).get("type") or "text").strip().lower() != "text":
            return ("a profile column is a flag on an ordinary TEXT column — it validates what "
                    "is typed into it, which it can only do for text")
        existing = ut.profile_field(table_key, st=session.runtime) if table_key else None
        if existing and existing.get("key") != str(fkey):
            return (f"this database already has a profile column — "
                    f"{existing.get('label') or existing.get('key')!r}. A database has at most "
                    f"one, so the automation knows which handle to enrich; edit that column, or "
                    f"take the flag off it first")
    bag = (body or {}).get("automation")
    if isinstance(bag, dict):
        flow = str(bag.get("flowId") or "").strip()
        if not flow:
            return ("an automation column has to name the automation that fills it — pick a "
                    "flow, or make this an ordinary column")
        if not ut.flow_bound(bag, st=session.runtime):
            return (f"this column names automation {flow!r}, which does not exist in this "
                    f"workspace — it may have been deleted; pick a flow that is still there")
    kind = str((body or {}).get("type") or "").strip()
    if kind and kind not in ut.UT_FIELD_TYPES:
        return (f"{kind!r} is not a column type here (types: "
                f"{', '.join(sorted(ut.UT_FIELD_TYPES))})")
    return (f"the column was refused — check the name and type, or the table may be at its "
            f"{ut.MAX_FIELDS}-column cap (types: {', '.join(sorted(ut.UT_FIELD_TYPES))})")


@router.patch("/tables/{table_key}/fields/{fkey}")
def patch_field(table_key: str, fkey: str, body: dict = Body(default=None),
                session: Session = Depends(require_session)):
    """Edit one column's definition, and MIGRATE its values when options are renamed.

    ⛔ A CHOICE RENAME IS AN EXPLICIT MAPPING, NEVER A DIFF (contract C-RENAME). `{renames:
    [{from, to}]}` arrives alongside the new options list, because a diff cannot tell "renamed
    Blue to Navy" from "deleted Blue, added Navy" — and guessing wrong empties the column and
    every saved view that filtered on it.
    """
    _field_or_refuse(session, table_key, fkey)
    ut = _ut()
    body = body or {}
    migrated = None
    renames = body.get("renames")
    if renames:
        try:
            migrated = ut.rename_choice_values(table_key, fkey, renames, st=session.runtime)
        except Exception:
            raise err(503, "store_unavailable", "the rename did not land — try again")
        # The per-user workspace strata and any view filter naming the old value are the OTHER
        # half of C-RENAME and belong to `core.table_store`. Called only if it is there: an
        # enumerator's mirror waits for its counterpart rather than guessing at its shape, and a
        # missing counterpart must not lose the half that DID land.
        try:
            import core.table_store as table_store
            fn = getattr(table_store, "rename_choice_values", None)
            if callable(fn):
                fn(table_key, fkey, renames, st=session.runtime)
                migrated = dict(migrated or {}, workspace=True)
        except Exception:                                         # noqa: BLE001
            migrated = dict(migrated or {}, workspace=False)
    field = ut.patch_field(table_key, fkey, body, st=session.runtime)
    if not field:
        # C3: the same narrator as `add_field`. Flagging a SECOND column is the same act as
        # adding one, so it must get the same sentence naming the column that already holds the
        # flag — a patch that answered "check the name and type" would send the reader to the
        # one thing that was never wrong (the D-46 lesson, one door over).
        raise err(400, "refused",
                  _refusal_sentence(ut, session, body, table_key=table_key, fkey=fkey))
    synced = ut.sync_reciprocal_link(table_key, fkey, st=session.runtime)
    field = synced.get("field") or field
    _refresh_relations(session)
    out = {"field": field}
    if migrated is not None:
        out["migrated"] = migrated
    return out


@router.delete("/tables/{table_key}/fields/{fkey}")
def delete_field(table_key: str, fkey: str, session: Session = Depends(require_session)):
    _field_or_refuse(session, table_key, fkey)
    if not _ut().delete_field(table_key, fkey, st=session.runtime):
        raise err(400, "refused",
                  "that column could not be removed — a database must keep at least one")
    _refresh_relations(session)
    return {"deleted": fkey}


@router.delete("/tables/{table_key}/rows/{rid}")
def delete_row(table_key: str, rid: str, session: Session = Depends(require_session)):
    # ⭐⭐ W31 QA — ONE SNAPSHOT FOR THE WHOLE DELETE, and the owner reported what it cost.
    # Owner, verbatim (2026-08-13): *"it still takes forever to delete a record from TT Profile."*
    # A single DELETE was FIVE whole-document deep copies before the commit even began — three in
    # the guard (`get` + `may_open` + `records_mutable`) and two more inside
    # `core.user_tables.delete_row` (`is_user_table` + `records_mutable` again). MEASURED: 2 reads
    # cost 45 ms on an 0.8 MB fixture, and tenant #0's `user_tables` document is **28.5 MB**
    # (D-185), so the guard alone was seconds of copying to answer questions about one row.
    # ⚠ THE LEND IS READ-ONLY AND THE WRITE STILL GOES THROUGH THE REAL RUNTIME — `_Lent`
    # `__getattr__`-passes `update` straight to it, and `_drop` runs against the LIVE document
    # under the store lock, so a lent snapshot can never be the thing written back.
    # ⚠ `flush='sync'` IS DELIBERATELY UNTOUCHED (D-118): nobody spams a delete, and an eventually
    # consistent delete is indistinguishable from one that did not work. This makes the guard
    # cheap; it does not make the commit optimistic.
    lent = _ut().lend(session.runtime)
    _records_or_refuse(session, table_key, st=lent)
    try:
        ok = _ut().delete_row(table_key, rid, st=lent)
    except Exception:
        raise err(503, "store_unavailable", "the delete did not land — try again")
    if not ok:
        raise err(400, "refused", "rows can only be deleted from user-created databases")
    _refresh_relations(session)
    return {"ok": True}


@router.patch("/tables/{table_key}/rows/{pid}")
def patch_row(table_key: str, pid: int, body: dict = Body(default=None),
              session: Session = Depends(require_session)):
    """Cell edits — the products PATCH on the user-table ctx. Routed through
    `core.grid_events.handle_one` so truncation and permission rules stay ONE implementation;
    the accepted values are read BACK from the bucket, never echoed from the request."""
    from core import grid_events

    updates = dict(body or {})
    if not updates:
        raise err(400, "empty_patch", "no fields to update")
    _records_or_refuse(session, table_key)
    g = ut_assembly(session, table_key, consume_corrections=False)
    if pid not in g["pids"]:
        raise err(403, "out_of_scope", "that row is not in this database")
    ctx = grid_events.EventCtx(
        uname=session.uname, allowed_pids=g["pids"], fields=g["fields"],
        admin=session.admin, fallback_ws=None, seen_ids={}, hidden_keys=frozenset(),
        st=session.runtime,                       # R6b (D-16)
        scope_key=table_key, table=_ops(session, table_key))
    try:
        grid_events.handle_one(
            {"id": f"patch:{table_key}:{pid}:{time.time_ns()}", "type": "overlay_patch",
             "pid": pid, "updates": updates}, ctx)
    except grid_events.StoreUnavailable:
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — your change was not saved")
    # ⚠ THE READ-BACK IS THE DEFINITION ROW, AND ON A `ut_` SCOPE THAT IS THE WHOLE OF IT.
    #
    # ⛔ CORRECTED, wave-29 T22 (owner item 2a): this note used to say "THE READ-BACK SPANS BOTH
    # STRATA, and it has to (wave 25, C3-A1)" while the two lines under it read exactly one bucket.
    # It was true of the wave-25 world it was written in — an ordinary cell landed in the caller's
    # OVERLAY and only a PROFILE cell wrote through — and it stayed after `grid_events` began
    # routing EVERY accepted cell on a `ut_` scope to `user_tables.patch_cells`
    # (`grid_events.py:1979-1984`). There is no second stratum this read is missing; a docstring
    # claiming otherwise is what makes the next reader look for a merge bug that is not here.
    # ⚠ Legacy overlay values, written before that routing existed, are still merged for DISPLAY
    # by `table_rows` (:587-595) — display only, and deliberately not re-asserted here: `_took`
    # asks whether THIS write landed, and this write goes to the definition.
    stored = dict(((_ut().get(table_key, st=session.runtime) or {}).get("rows") or {})
                  .get(str(pid)) or {})
    accepted = {k: stored.get(k) for k in updates if k in stored}

    def _took(k):
        """Did the cell TAKE this write? Normally that is "stored == asked".

        ⚠ A PROFILE COLUMN CANONICALISES, so "stored != asked" is its NORMAL success: `@Nurilab`
        and `instagram.com/nurilab` both store `nurilab`. Reporting those as refused would tell
        the client to roll back a write that landed. But it cannot simply be exempted either —
        a junk handle leaves the OLD value sitting in `stored`, which would then read as
        accepted. So the question asked is the exact one: **is what is stored the canonical form
        of what was asked?** Anything else is a genuine refusal.
        """
        if k not in accepted:
            return False
        want = str(updates[k])
        if stored.get(k) == want:
            return True
        pf = _ut().profile_field(table_key, st=session.runtime)
        if pf and pf["key"] == k:
            handle, ok = _ut().normalize_profile(want, pf["profile"].get("source"))
            return bool(ok) and stored.get(k) == handle
        return False

    refused = sorted(k for k in updates if not _took(k))
    out = {"pid": pid, "updates": accepted}
    if refused:
        out["refused"] = refused
    # ⭐ R6: the cells the SERVER changed that the client never typed — the preset cells a
    # profile blank cleared. Without this the grid keeps painting a stale follower count under
    # an empty handle until something else forces a refetch, which is the NO-BLIP LAW's other
    # half: the client may keep only what the server actually took, and must be TOLD what else
    # moved. Derived by diffing this row against what was asked for, so it cannot drift from
    # whatever the clear rule decides to touch.
    also = {k: v for k, v in stored.items() if k not in updates and str(v or "") == ""}
    cleared = sorted(k for k in also if k in _ut().PROFILE_PRESET_KEYS)
    if cleared:
        out["cleared"] = cleared
    # ⭐⭐ R9's SECOND RE-ARM DOOR — the one call that makes `engine.clear_gone` live (wave 28,
    # amendment A5; SESSION B built and gated the function and correctly declared it INERT until
    # this line existed, citing [[flag-shipped-without-its-writer]]).
    #
    # ⛔ R9 makes a `not_found` handle a TOMBSTONE, not a 30-day backoff: nothing re-buys a dead
    # account on a timer any more. Door 1 — correcting the handle — needs no wiring, because the
    # verdict is keyed on `(platform, handle)` and a corrected handle simply is not the verdict we
    # recorded. THIS is door 2: a human re-typing the SAME handle, which is how somebody says "try
    # it again, the account is back". Without this call that person has no way back at all, and
    # the failure costs nothing and raises nothing — so no spend-shaped test would ever find it.
    #
    # ⚠ GATED ON `updates`, NOT ON `accepted`: re-typing the identical value is the whole case this
    # door exists for, and a no-op write can be filtered out of `accepted`. What matters is that a
    # human touched the handle cell.
    # ⚠ NOT a bare `except: pass`. A swallowed AttributeError here would be exactly the optional-
    # prop silence this wiring exists to prevent — if the engine ever loses `clear_gone`, that must
    # be readable in the log rather than degrade into "the re-arm quietly stopped working".
    _pf = _ut().profile_field(table_key, st=session.runtime)
    if _pf and _pf["key"] in updates:
        try:
            import automation_engine as _engine
            _engine.clear_gone(session.runtime, table_key, stored.get(_pf["key"]))
        except Exception as e:                                    # noqa: BLE001
            print(f"[aios-api] clear_gone failed: {type(e).__name__}: {e}")
    _refresh_relations(session)
    return out