"""routes_nav.py — X2's `GET /api/v1/nav`: the registry, filtered by what this session may open.

SERVER-FILTERED, not client-filtered. The client renders what it is given and never decides who
may see what — a nav that hides a link the API would still serve is a UI courtesy, not a
permission. `core.perms.nav_pages` is the single predicate (shared with `may_open`'s page gate),
so the nav and the 403 can never disagree about a grant.

The full rule set — archived is invisible to everyone, `group_only` rows are excluded so no
`parent` reference can dangle, `nav: False` surfaces ship as `chrome: 'utility'`, and the
`prefs.json` Library preference is deliberately NOT applied — is documented on
`core.perms.nav_pages`, which is where it belongs: one place, both callers.
"""
import time

from fastapi import APIRouter, Body, Depends

from deps import Session, err, perms, require_session

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

#: Wave 2026-08-02 (C-SCHEMA): per-user folders over the database list, the view-folder
#: pattern applied to the nav. Cosmetic per-user state — placement never grants or hides a
#: surface (the server-filtered nav still decides what exists).
_NAV_PREFS_KEY = "nav_prefs"
_MAX_NAV_FOLDERS = 16

#: WAVE 19 (R8 / contract C1): the database's NAME and ICON overrides.
#:
#: ⚠ TENANT-WIDE, which is the whole difference from `nav_prefs` above and the reason it is a
#: separate bucket rather than another field in that one. Folders and placement are one
#: person's arrangement of their own rail — per-user by definition. What a database is CALLED
#: and what it looks like are facts about the database: a workspace where two people call the
#: same table different things has no shared vocabulary left to discuss it in. Same store, two
#: buckets, because they answer to two different owners.
#:
#: Shape: {"<pageKey>": {"icon": {"shape": <FolderShape>, "tone": <FolderTone>},
#:                       "name": "<override, ut_* keys only>"}}
_NAV_META_KEY = "nav_meta"

#: The grid's folder-icon vocabulary, mirrored — 12 shapes x 5 tones. The CLIENT imports these
#: from `customer-grid/types` rather than redefining them; this end cannot import TypeScript,
#: so it is the one place the list is written twice.
#:
#: ⛔ WHAT AN UNKNOWN VALUE MUST NOT DO IS BE STORED. `FolderMark` indexes its path table by
#: shape and maps the result, so a shape this whitelist let through and the renderer does not
#: know is `undefined.map()` — a blank rail, from a stored preference, for every user in the
#: tenant until somebody edits the store by hand. Refusing at the door is the cheap end of
#: that. If the two lists ever drift the symptom is an icon that silently reverts to the
#: default, which is the loudest SAFE failure available here.
_ICON_SHAPES = frozenset({"folder", "star", "flag", "tag", "bookmark", "grid",
                          "chart", "map", "users", "clock", "heart", "bolt"})
_ICON_TONES = frozenset({"neutral", "blue", "green", "yellow", "red"})
_MAX_NAV_NAME = 60

#: WAVE 23 (contract C10 / ruling R7) — the Home landing's RECENTS.
#:
#: PER-USER, like `nav_prefs` two buckets up and unlike `nav_meta`: what I opened last is
#: nobody else's business, and a tenant-wide "recently opened" would be a surveillance feature
#: rather than a convenience one.
#:
#: ⛔ A MAP KEYED BY PAGE, NOT AN APPEND LOG, and the difference is the whole feature.
#: `{username: {pageKey: <epoch seconds>}}` — re-opening a database OVERWRITES its stamp. An
#: append-only list capped at 50 fills with fifty copies of the same ten databases inside one
#: working session, and the cap then evicts OLDEST-FIRST: the tenth database you touched falls
#: off the list while forty slots hold repeat visits to the first. Keying by page makes
#: "recent" mean what the word means, and makes the cap bound the number of DATABASES
#: remembered rather than the number of clicks.
_NAV_RECENTS_KEY = "nav_recents"
_MAX_RECENTS = 50

#: ⚠ EPOCH SECONDS (UTC by definition), never a formatted stamp — and this is a correction of a
#: precedent, not a preference. `user_tables.create` writes
#: `datetime.now().strftime('%Y-%m-%dT%H:%M:%S')`: naive LOCAL time, no offset. A browser parses
#: that string as its OWN local time, so on a UTC host read by a non-UTC reader "opened 30
#: minutes ago" renders as "opened 7 hours ago" and the Today / Past-7-days buckets misfile —
#: with nothing to go red, because both ends are internally consistent. An integer instant has
#: no such reading. (D-18 made the same correction for notification stamps AFTER the defect
#: shipped; this is that lesson applied before it.)
def _now() -> int:
    return int(time.time())


def _clean_nav_prefs(raw, page_keys, *, keep_unknown_ut=False):
    """Validated wholesale replacement, the `clean_folders` posture: prune, never invent.

    `page_keys` is the set of TOP-LEVEL keys this session may see; a placement of an unknown
    or invisible key is dropped (it can return when the grant does — placement is cosmetic,
    so pruning is loss-free). Unknown folder refs drop the placement, not the folder.

    `keep_unknown_ut` is the STORE-BLIP escape hatch — see `_placeable_top_keys`. When the
    `ut_*` listing could not be read, a `ut_` key absent from `page_keys` is KEPT rather than
    pruned: keeping a placement whose database may since have been deleted is cosmetically
    harmless, while pruning a live one is the silent data loss this function just stopped
    causing. Never widened to non-`ut_` keys — those come from the compiled registry, which
    cannot fail to enumerate.
    """
    raw = raw if isinstance(raw, dict) else {}
    folders, seen = [], set()
    for f in (raw.get("folders") or [])[:_MAX_NAV_FOLDERS]:
        if not isinstance(f, dict):
            continue
        fid = str(f.get("id") or "").strip()[:40]
        name = " ".join(str(f.get("name") or "").split())[:40]
        if not fid or fid in seen or not name:
            continue
        seen.add(fid)
        folders.append({"id": fid, "name": name})
    ids = {f["id"] for f in folders}
    placement = {}
    src = raw.get("placement")
    if isinstance(src, dict):
        for k, v in src.items():
            k, v = str(k)[:60], str(v)[:40]
            if v not in ids:
                continue                       # the folder is gone; the placement goes with it
            if page_keys is None or k in page_keys:
                placement[k] = v
            elif keep_unknown_ut and k.startswith("ut_"):
                placement[k] = v
    return {"folders": folders, "placement": placement}


def _placeable_top_keys(session):
    """`(keys, enumerated)` — the keys a placement may name, INCLUDING this tenant's databases.

    ⛔ THIS IS THE ITEM-6 FIX (wave 27, contract C1), and the bug it closes was invisible by
    construction. The old version read `perms.nav_pages` alone — the compiled REGISTRY — and
    `perms.py` contains no `ut_` or `user_tables` reference at all, because a tenant's
    user-created databases are merged into the nav payload by `nav()` BELOW the permission wall.
    So every `ut_*` key was absent from this set, and `_clean_nav_prefs` pruned every placement
    naming one — on READ and on WRITE.

    The symptom was "a database I drag into a folder falls out of it later", never "it does not
    work": the write answered **200 OK** having stored `{}`, the client kept its optimistic copy
    (`Shell.commitPrefs` reverts only on `!ok`), and the folder held for the rest of the session.
    The next reload served the pruned copy and the database was back at root. Built-in modules
    (`customer_data`, `product_data`) ARE in the registry and persisted fine, which is exactly
    why it read as intermittent rather than as a missing feature.

    ⚠ THE SAME MISTAKE, ALREADY MADE AND ALREADY FIXED ONE FUNCTION AWAY. `nav()`'s recents
    block (see its `allowed` comment) hit this in wave 23 and solved it by pruning against the
    ASSEMBLED page list. This is that lesson applied to the second consumer, which the wave-23
    fix did not reach.

    `enumerated` is False when the `ut_` listing raised — the caller must then NOT treat this set
    as authoritative for `ut_` keys (`keep_unknown_ut`). Pruning a person's whole arrangement
    because the store blinked would be the original defect wearing a different cause.
    """
    pages = perms.nav_pages(session.user) or []
    keys = {p["key"] for p in pages if not p.get("parent")}
    try:
        import core.user_tables as user_tables
        # The SAME listing `nav()` renders from — `may_open`-filtered, so a placement can never
        # name a database this session cannot see, and the nav and this door agree by sharing
        # one predicate rather than by two lists being kept in step.
        # ⭐ W31-T10 (C1/D-175): `nav_entries` lends its own read to `may_open`, so this door is
        # ONE document copy rather than `1 + N`. It is not a footnote here — `Shell.tsx` fires
        # `/nav/prefs` CONCURRENTLY with `/nav` on the same `Store._lock`, and it measured
        # **8,807 ms live / 17,945 ms in-process** on tenant #0, i.e. roughly half the wait the
        # owner reports as "the Automation row arrives ten seconds late".
        for e in user_tables.nav_entries(viewer=session.uname, is_admin=session.admin,
                                         st=session.runtime):
            keys.add(str(e.get("key") or ""))
    except Exception:
        return keys, False
    return keys, True


def _clean_icon(raw):
    """A stored/incoming icon, or None. Prune, never invent — `clean_folders`' posture."""
    if not isinstance(raw, dict):
        return None
    shape, tone = raw.get("shape"), raw.get("tone")
    if shape not in _ICON_SHAPES or tone not in _ICON_TONES:
        return None
    return {"shape": shape, "tone": tone}


def _read_nav_meta(runtime):
    """The tenant's `nav_meta`, validated on the way OUT as well as in.

    Re-validating a read looks redundant and is not: the bucket outlives this code, an older
    build may have written a shape this one no longer accepts, and the whitelist is mirrored
    from a vocabulary that lives in another language. A row whose icon does not survive
    validation renders the default mark — never a crash, and never a half-drawn glyph.
    """
    try:
        stored = runtime.get(_NAV_META_KEY) or {}
    except Exception:
        return {}                      # a store blip must not take the nav down with it
    if not isinstance(stored, dict):
        return {}
    out = {}
    for key, meta in stored.items():
        if not isinstance(meta, dict):
            continue
        entry = {}
        icon = _clean_icon(meta.get("icon"))
        if icon:
            entry["icon"] = icon
        # ⛔ THE `ut_` RULE IS RE-APPLIED ON READ, not trusted from the record. The write
        # route refuses a name for a built-in key, but a record written by an older build (or
        # by hand) could still carry one — and honouring it would let the store rename
        # `customer_data` on screen while `core/registry.py` and every other reader went on
        # calling it Customer. A name that could not be written today is not served today.
        name = meta.get("name")
        if str(key).startswith("ut_") and isinstance(name, str) and name.strip():
            entry["name"] = " ".join(name.split())[:_MAX_NAV_NAME]
        if entry:
            out[str(key)] = entry
    return out


def _read_recents(runtime, uname, allowed):
    """This user's recents, newest first, PRUNED to what they may currently see.

    ⚠ PRUNED ON READ, never on write, and both halves of that are deliberate:

      · the WRITE happens on every route open — it is the one hot path this file has — so it
        must not build the whole nav to validate one key;
      · a key whose grant was REVOKED must stop being offered without anyone running a
        migration, and must come back if the grant does. That is `_clean_nav_prefs`' own
        prune-never-invent posture, applied to a second bucket for the same reason.

    A key that is not in `allowed` therefore reaches the store and never reaches a screen —
    which also means a stuffed key cannot be used to discover what exists: it comes back only
    if the session could already see it.
    """
    try:
        stored = (runtime.get(_NAV_RECENTS_KEY) or {}).get(uname) or {}
    except Exception:
        return []                      # a store blip must not take the nav down with it
    if not isinstance(stored, dict):
        return []
    out = []
    for key, at in stored.items():
        key = str(key)
        if allowed is not None and key not in allowed:
            continue
        try:
            at = int(at)
        except (TypeError, ValueError):
            continue                   # a stamp this build cannot read is not a stamp
        if at <= 0:
            continue
        out.append({"key": key, "at": at})
    out.sort(key=lambda r: r["at"], reverse=True)
    return out[:_MAX_RECENTS]


@router.get("/nav")
def nav(session: Session = Depends(require_session)):
    """`{pages: [{key,label,source?,chrome}], landing}`.

    Note what this does NOT do: it never returns an empty `pages` list as a way of saying "you
    are not allowed". A session that may open nothing at all is a misconfigured account, and it
    gets an explicit 403 — an empty 200 is indistinguishable from "the registry is empty" and is
    how a permission bug hides in plain sight (X2's never-an-empty-200 rule).
    """
    pages = list(perms.nav_pages(session.user) or [])
    # Wave 18 C1-TENANT: the REGISTRY is the product's module catalogue — tenant #0's world.
    # A tenant record may enable a subset (`modules: [...]`); absent/'all' means everything
    # (royal-imports and the compiled builders). A blank tenant enables NOTHING: its nav is
    # its own databases, which is what "different set of databases at later waves" means.
    tcfg = getattr(session.runtime.tenant, "config", None) or {}
    tmods = tcfg.get("modules", "all")
    # ⭐⭐ W31-T11 (owner item 6b) — WHAT THE CATALOGUE FILTER REMOVED, SAID OUT LOUD.
    #
    # Every provisioned tenant carries a restricted list today (`gtmlab`/`loopable`/`nurilab` all
    # `['analyst','automation']` — census in `proto/nav-omission-census.md`), so this filter runs
    # on every request that is not tenant #0's. It is the INTENDED catalogue, not a defect. But a
    # row it removes and a row a store failure dropped are the SAME absence on the wire, and the
    # client renders `null` for both — pixel-identical to still-loading. Naming the removals is
    # what lets the shell tell "not part of this workspace" from "we could not read it".
    _omitted = []
    if tmods != "all":
        allowed = {str(k) for k in (tmods or [])}
        _omitted = sorted({str(p.get("key")) for p in pages
                           if p.get("key") not in allowed
                           and (p.get("parent") or "") not in allowed})
        pages = [p for p in pages
                 if p.get("key") in allowed or (p.get("parent") or "") in allowed]
    # Wave 18 C3-UT: this tenant's user-created databases, merged AFTER the registry rows —
    # the host does the same at app.py:7796. Filtered by the per-table wall (`may_open`), so
    # the nav cannot offer a row the table routes would refuse.
    # ⭐⭐ W31-T10 (contract C1, D-175) — THE DOCUMENT IS READ ONCE FOR THE WHOLE REQUEST.
    #
    # This route was `2 + N` full deep copies of a 35.8 MB-ceiling document: `nav_entries` took one
    # and then `may_open` took another PER TABLE, and the `manage`/`canDelete`/`locked` loop below
    # took a SECOND independent one. Median `GET /nav` on tenant #0: **9,548 ms live, 24,741 ms
    # in-process** — the second figure is the one that matters, because with the network gone and
    # the document resident this route and `/nav/prefs` are the ONLY two that stay slow while every
    # other collapses to 20–161 ms. That is per-call CPU, and this is it.
    # ⚠ `_ut_defs` is read HERE, above the merge, so the two former reads become one; the locked/
    # manage loop below consumes this same dict.
    _ut_defs, _ut_locked_mode, _degraded = {}, "", []
    try:
        import core.user_tables as user_tables
        # ⭐⭐ W32-T02 (R8/D-175/D-185) — AND THAT ONE READ IS A PROJECTION NOW. The block below
        # reads `label`, `source`, `createdBy` and `recordMode`; `may_open` reads `createdBy` and
        # the shares registry. Nothing on this route has ever opened a row — and on tenant #0 the
        # rows are **99.89% of the document** (28,551,441 bytes; the definitions are 31,220 of
        # them). Measured: `all_tables` 1,750 ms -> `all_defs` 1.4 ms, and `GET /nav` 1,921 ms
        # median -> see the ticket. THAT is owner item 5: the rail rows did not arrive seconds
        # apart because of CSS or ordering, they arrived when their route finished copying.
        # ⛔ `all_defs` REFUSES `rows` rather than answering `{}` — if you add something here that
        # needs a row, it raises with the reason instead of painting an empty grid. Use
        # `all_tables` for that, and know you are buying the whole copy back.
        _ut_defs = user_tables.all_defs(st=session.runtime) or {}
        # ⛔ NEVER DEFAULT THIS TO `None`. `_ut_defs.get(k)` is `{}` for an unknown key, so its
        # `.get("recordMode")` is None too — and `None == None` would mark EVERY database locked
        # the moment this import failed. A sentinel that can equal real data is not a sentinel.
        _ut_locked_mode = str(user_tables.AUTOMATION_RECORD_MODE)
        _lent = user_tables.lend(session.runtime, **{user_tables.STORE_KEY: _ut_defs})
        for e in user_tables.nav_entries(viewer=session.uname, is_admin=session.admin,
                                         st=_lent):
            pages.append({"key": e["key"], "label": e["label"],
                          "source": e.get("source") or "Blank", "chrome": "main"})
    except Exception:
        # ⭐⭐ W31-T11 (owner item 6b) — STILL SWALLOWED, NO LONGER SILENT.
        #
        # A store blip must not take the whole nav down with it, and that half stands. What
        # changed is that it used to answer **200 OK with every database missing** and say
        # nothing — so a client cannot tell "this tenant has no databases" from "we could not
        # read them", and the rail renders the same complete-looking thing either way. That is
        # the owner's *"Connectors and Automation module still disappears"* class of report:
        # the payload is a claim about what exists, and a claim it could not verify has to be
        # marked as such. `degraded` is that mark; the shell renders it in the affected slot.
        _degraded.append("databases")
        pass
    if not pages:
        if tmods != "all":
            # A provisioned tenant with no modules and no databases YET is a legitimate empty
            # state, not a misconfigured account — the client renders "create your first
            # database", and X2's never-an-empty-200 rule is honoured by saying WHY it is
            # empty rather than leaving 200-[] ambiguous.
            return {"pages": [], "landing": None, "empty": "no_databases"}
        raise err(403, "no_surfaces",
                  "your account has no dashboards assigned — ask an administrator")
    # WAVE 19 (R8 / C1): the tenant's name + icon overrides, merged LAST — after the registry
    # rows, after the tenant module filter, after the user tables. Merged HERE rather than
    # applied by the client for one reason: `label` is what every reader of this payload shows,
    # including the Settings modal's `moduleLabels` map, so a client-side merge would put the
    # override in one door and the registry label in the other.
    #
    # ⛔ THIS MUTATES `pages` IN PLACE, WHICH IS SAFE ONLY BECAUSE `perms.nav_pages` BUILDS A
    # FRESH `row = {...}` PER CALL (perms.py:194) and the ut_ rows are built fresh here. If
    # either ever starts handing back cached or module-level dicts, this loop would write one
    # tenant's chosen label into the object the NEXT tenant's request reads — a cross-tenant
    # leak with no symptom until two tenants rename the same registry key. Copy the rows before
    # merging on the day that invariant changes.
    meta = _read_nav_meta(session.runtime)
    # Wave 21 (C3): the definitions, once — `manage`/`canDelete` below answer from `createdBy`.
    # WAVE 27 (C9): `locked` answers from `recordMode`, off the same one read.
    # ⭐ W31-T10: that read is now the SAME one the merge above did — it used to be a second,
    # independent `all_tables`, which is why D-175 called this route `2 + N` rather than `1 + N`.
    # ⚠ The fail-closed defaults still hold: both are initialised before the try above, so an
    # import or store failure leaves `_ut_locked_mode` empty and nothing is marked locked.
    for p in pages:
        # `manage` (R14): may THIS session change this row's icon/name? Answered HERE because
        # the server is the only end that knows — the client cannot see who created a user
        # table. Additive and FAIL-CLOSED (absent reads as "no"), so the rail offers a control
        # only where the write would actually land, and the route re-checks it regardless.
        #
        # ⛔ WAVE 21 (C3/W-5): the "presence IS the answer" shortcut DIED in wave 20 — the
        # de5037f share-grant admission widened `may_open`, so a row's presence now includes
        # databases merely SHARED to this viewer. `manage` (rename/icon) and `canDelete` are
        # therefore answered from the DEFINITION: creator-or-admin strictly, matching the
        # walls the PATCH and DELETE routes actually enforce. A grantee sees the row and no
        # controls — the honest shape (the old code offered rename to users the route 403'd).
        key = str(p.get("key", ""))
        if key.startswith("ut_"):
            _creator = str((_ut_defs.get(key) or {}).get("createdBy") or "")
            _mine = bool(session.admin or (_creator and _creator == session.uname))
            p["manage"] = _mine
            p["canDelete"] = _mine
            # ⭐ WAVE 27 item 3 (contract C9) — A LOCKED DATABASE SAYS SO IN THE RAIL.
            #
            # "Locked" is the owner's item-4 vocabulary and it means exactly ONE thing (DESIGN.md
            # §4, THE THREE LOCKS): RECORDS cannot be added, deleted or edited — **fields still
            # can**. The automation-owned IG child datasets (posts, comments, snapshots) are the
            # live example; their rows arrive from the engine, so a "+" row there could only
            # refuse, which R8 calls a fake affordance.
            #
            # ⚠ THE AUTHORITY IS `user_tables.records_mutable`, NOT THIS LINE. The comparison is
            # inlined only because `_ut_defs` is already in hand — calling the predicate per row
            # would be one store read per database on every nav request — and the VALUE comes
            # from the module's own constant rather than a copied string, so the two cannot drift
            # to different answers. If that predicate ever grows a second condition, this must
            # become a call.
            #
            # ⚠ ABSENT READS AS UNLOCKED, and that is the safe direction here even though it is
            # the opposite of `manage`'s fail-closed: the lock ICON is an affordance hint, while
            # the actual refusal is `routes_tables._records_or_refuse`'s 403. A store blip costs
            # a missing hint, never a write that should not have landed.
            if (_ut_locked_mode
                    and (_ut_defs.get(key) or {}).get("recordMode") == _ut_locked_mode):
                p["locked"] = True
        elif session.admin:
            p["manage"] = True
        entry = meta.get(p.get("key")) if meta else None
        if not entry:
            continue
        if entry.get("icon"):
            p["icon"] = entry["icon"]
        if entry.get("name"):
            p["label"] = entry["name"]
    landing = perms.landing_page(session.user)
    if landing and not any(p.get("key") == landing for p in pages):
        landing = pages[0].get("key")
    # WAVE 23 (C10 / R7) — the Home landing's recents, on the payload the client already asks
    # for. A second round trip for a list this short, computed from a bucket this route is
    # already holding the store open for, would be a request per page load for no gain.
    #
    # ⛔ `allowed` IS THE ASSEMBLED PAGE LIST — the rows this route just built, `ut_*` databases
    # included. Pruning against `perms.nav_pages` alone would silently drop every user database
    # from Home's recents: the exact surface R7 is about, invisible, with every gate green.
    #
    # ⚠ WAVE 27 (item 6): the OTHER consumer of that narrower set — the folder-placement door —
    # had the identical bug and nobody connected the two for four waves. It is fixed at the
    # source now (`_placeable_top_keys`, which merges the same `may_open`-filtered listing), so
    # both doors finally agree on what a placeable key is. This block keeps using the assembled
    # list because it already holds it: re-enumerating here would be a second store read for an
    # answer sitting in a local variable.
    recents = _read_recents(session.runtime, session.uname,
                            {str(p.get("key", "")) for p in pages})
    # ⭐ W31-T11 — TWO KINDS OF ABSENCE, NAMED SEPARATELY, and both keys are ALWAYS PRESENT.
    #   `omitted`  — this workspace's catalogue does not include these modules. Deliberate.
    #   `degraded` — a part of this payload could not be read. NOT deliberate, and the shell says
    #                so in the affected slot instead of rendering a confident nothing.
    # ⚠ A key a consumer has to test for is a key a consumer forgets to test for; both ship as
    # `[]` rather than being omitted when empty, which is the same rule `limits` follows.
    return {"pages": pages, "landing": landing, "recents": recents,
            "omitted": _omitted, "degraded": _degraded}


@router.post("/nav/opened")
def nav_opened(body: dict = Body(default=None),
               session: Session = Depends(require_session)):
    """WAVE 23 (C10) — stamp a page as JUST OPENED. Fire-and-forget from the client.

    The client calls this on every route commit, so this is the only write in this file on a
    hot path, and three things follow from that:

      · `flush='async'` — the coalescing mode ([[store-async-flush]]). A blocking upload per
        page open against an HF-Dataset-backed store would put a network round trip inside every
        navigation. `nav_prefs`/`nav_meta` stay `sync` because a folder rename is not a hot path;
        this is.
      · NO VALIDATION OF THE KEY against the nav. Building the page list to check one string
        would make the stamp cost more than the navigation that triggered it — and it would buy
        nothing, because the READ prunes to what the session may currently see. An unknown or
        revoked key is stored and never served back.
      · THE MAP IS CAPPED HERE TOO. Read-side capping alone would let a hostile or buggy client
        grow one user's document without bound; `_MAX_RECENTS` entries survive, oldest first to
        go, which is the same rule the read applies.
    """
    body = body if isinstance(body, dict) else {}
    key = str(body.get("key") or "").strip()[:60]
    if not key:
        raise err(400, "bad_request", "no page was named")
    if not session.runtime.available():
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — nothing was recorded")
    stamp, uname = _now(), session.uname

    def _up(data):
        data = data if isinstance(data, dict) else {}
        mine = dict(data.get(uname) or {}) if isinstance(data.get(uname), dict) else {}
        mine[key] = stamp
        if len(mine) > _MAX_RECENTS:
            # Oldest first. `int(v)` guarded: a stamp an older build wrote in another shape
            # sorts as 0 and is the first thing evicted, which is the right answer for a value
            # this route can no longer read.
            def _at(item):
                try:
                    return int(item[1])
                except (TypeError, ValueError):
                    return 0
            mine = dict(sorted(mine.items(), key=_at, reverse=True)[:_MAX_RECENTS])
        data[uname] = mine
        return data

    try:
        session.runtime.update(_NAV_RECENTS_KEY, _up, flush='async')
    except Exception:
        raise err(503, "store_unavailable",
                  "the tenant store refused the write — nothing was recorded")
    return {"key": key, "at": stamp}


@router.post("/nav/meta")
def save_nav_meta(body: dict = Body(default=None),
                  session: Session = Depends(require_session)):
    """WAVE 19 (R8 / C1) — set one database's icon and/or name, tenant-wide.

    A PATCH OF ONE KEY, not the wholesale replace `/nav/prefs` uses two routes up, and the
    asymmetry is deliberate. Prefs are one user's complete picture of their own rail, so
    replacing the document whole is what makes a deleted folder stay deleted. This bucket is
    shared by every admin in the tenant: a wholesale write here means whoever saves last
    silently erases what the other one named while their tab was open.

    THREE WALLS, all fail-closed:
      · the SESSION must be able to open the key at all (the same predicate the nav uses, so
        the rail and this route cannot disagree about what exists);
      · the WRITE is admin-only — renaming a database is a change every user in the tenant
        sees, which is the definition of an administrative act here;
      · `name` is refused outright for a non-`ut_` key. Built-in labels are compiled registry
        literals: honouring an override would leave this payload and `core/registry.py`
        calling the same module two different things, and the wave-16 lesson is that the
        client must not be the place that decides what a payload meant.

    `icon: null` CLEARS. An absent field is untouched — which is what makes a rename and an
    icon change two independent writes rather than a race between them.
    """
    body = body if isinstance(body, dict) else {}
    key = str(body.get("key") or "").strip()
    if not key:
        raise err(400, "bad_request", "no database was named")
    # ── THE WALL (ruling R14, 2026-08-04) ────────────────────────────────────────────────────
    # TWO DOORS, because a `ut_` database and a built-in module are owned by different people.
    #
    #   · `ut_*` — the table's CREATOR or a tenant admin, which is exactly what
    #     `user_tables.may_open` already means. A database you made is yours to name; requiring
    #     an admin for that was the C1 consequence B flagged and R14 resolved. `session.require`
    #     is deliberately NOT used here: it is the MODULE gate and would 403 every ut_ key,
    #     because a user table is not a module. Same split `/nav/schema/{key}` makes.
    #   · everything else — admin only, and icon only. There is no owner of `customer_data` to
    #     defer to, and its label is a compiled registry literal (refused below regardless).
    if key.startswith("ut_"):
        import core.user_tables as user_tables
        if not user_tables.get(key, st=session.runtime) or not user_tables.may_open(
                key, session.uname, session.admin, st=session.runtime):
            raise err(403, "forbidden", "that database belongs to another user")
    else:
        if not session.admin:
            raise err(403, "forbidden",
                      "only an administrator can change a built-in database's icon")
        session.require(key)

    patch = {}
    if "icon" in body:
        icon = _clean_icon(body.get("icon"))
        if body.get("icon") is not None and icon is None:
            # Loud, not silent. A shape this build does not know is a CLIENT that has drifted
            # from this whitelist, and answering 200 to a write that stored nothing is how
            # that drift stays invisible until a user reports "my icon keeps resetting".
            raise err(400, "bad_icon", "that icon is not one of the available shapes and tones")
        patch["icon"] = icon
    if "name" in body:
        if not key.startswith("ut_"):
            raise err(400, "name_not_allowed",
                      "only a database you created can be renamed — this one's name comes "
                      "from the module registry")
        name = " ".join(str(body.get("name") or "").split())[:_MAX_NAV_NAME]
        if not name:
            raise err(400, "bad_request", "a database needs a name")
        patch["name"] = name
    if not patch:
        raise err(400, "bad_request", "nothing to change")

    if not session.runtime.available():
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — nothing was saved")

    def _up(data):
        data = data if isinstance(data, dict) else {}
        entry = dict(data.get(key) or {}) if isinstance(data.get(key), dict) else {}
        for field, value in patch.items():
            if value is None:
                entry.pop(field, None)     # an explicit null CLEARS
            else:
                entry[field] = value
        # An entry with nothing left in it is removed rather than stored empty: the read side
        # skips empties anyway, and a bucket that accumulates `{}` per key is a document that
        # grows forever and says nothing.
        if entry:
            data[key] = entry
        else:
            data.pop(key, None)
        return data

    try:
        session.runtime.update(_NAV_META_KEY, _up)
    except Exception:
        raise err(503, "store_unavailable",
                  "the change was not saved — the store refused the write")
    return {"key": key, "meta": _read_nav_meta(session.runtime).get(key, {})}


@router.get("/nav/prefs")
def nav_prefs(session: Session = Depends(require_session)):
    """This user's database-list folders, re-validated at serve time against what they may
    currently see (a revoked page's placement vanishes with the page, and returns with it)."""
    try:
        stored = (session.runtime.get(_NAV_PREFS_KEY) or {}).get(session.uname) or {}
    except Exception:
        stored = {}
    keys, enumerated = _placeable_top_keys(session)
    return {"prefs": _clean_nav_prefs(stored, keys, keep_unknown_ut=not enumerated)}


@router.post("/nav/prefs")
def save_nav_prefs(body: dict = Body(default=None),
                   session: Session = Depends(require_session)):
    """Wholesale replace, like the table folder stratum — the client sent its complete
    picture, validated here; a partial merge would resurrect deleted folders forever."""
    keys, enumerated = _placeable_top_keys(session)
    clean = _clean_nav_prefs(body or {}, keys, keep_unknown_ut=not enumerated)
    if not session.runtime.available():
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — nothing was saved")

    def _up(data):
        data = data if isinstance(data, dict) else {}
        if clean["folders"] or clean["placement"]:
            data[session.uname] = clean
        else:
            data.pop(session.uname, None)
        return data

    try:
        session.runtime.update(_NAV_PREFS_KEY, _up)
    except Exception:
        raise err(503, "store_unavailable",
                  "the folder change was not saved — the store refused the write")
    return {"prefs": clean}


@router.get("/nav/schema/{key}")
def nav_schema(key: str, session: Session = Depends(require_session)):
    """The database's schema drawer payload: its field contract + the semantic measures this
    session may build with. Fail-closed on the SAME predicate as the nav — a key the session
    may not open answers 403, never a redacted schema."""
    # Wave 18 C3-UT: a user table's schema is its own definition, walled by ITS predicate
    # (`may_open`) rather than the module grant machinery — `session.require` would 403 every
    # ut key because a user table is deliberately not a module.
    if key.startswith("ut_"):
        import core.user_tables as user_tables
        defn = user_tables.get(key, st=session.runtime)
        if not defn or not user_tables.may_open(key, session.uname, session.admin,
                                                st=session.runtime):
            raise err(403, "forbidden", "that database belongs to another user")
        # WAVE 19 (R8) — the drawer wears the RENAMED name. A rail that says one thing and a
        # schema panel opened from it that says another is the drift a rename is supposed to
        # remove, not create.
        return {"key": key,
                "label": (_read_nav_meta(session.runtime).get(key, {}).get("name")
                          or defn.get("label") or key),
                "source": defn.get("source") or "Blank",
                "fields": [{"key": f["key"], "label": f["label"], "type": f["type"],
                            "source": f.get("source") or "overlay",
                            "description": str(f.get("description") or "")}
                           for f in (defn.get("fields") or [])],
                "measures": []}
    session.require(key)
    pages = perms.nav_pages(session.user) or []
    page = next((p for p in pages if p.get("key") == key), None)
    if page is None:
        raise err(404, "unknown_page", f"{key!r} is not a database this session can see")
    fields = []
    if key in ("customer_data", "cohort", "customers"):
        try:
            import aios_grid
            for f in aios_grid.FIELDS:
                entry = {"key": f["key"], "label": f["label"], "type": f["type"],
                         "source": f["source"],
                         "description": str(f.get("description") or "")}
                if f.get("options"):
                    entry["options"] = list(f["options"])
                fields.append(entry)
        except Exception:
            fields = []
    measures = []
    try:
        from core import measure_resolve
        team_id = perms.scope_team_id(session.user)
        for m in measure_resolve.offer(team_id) or []:
            measures.append({"key": str(m.get("key") or ""),
                             "label": str(m.get("label") or m.get("key") or ""),
                             "type": str(m.get("type") or "")})
    except Exception:
        measures = []
    out = {"key": key, "label": page.get("label") or key,
           "source": page.get("source") or "", "fields": fields, "measures": measures}
    if not fields:
        # Honest, never a mock: a database whose contract is not yet published says so.
        out["note"] = "This database has not published a field contract yet."
    return out