api/routes_shares.py

"""routes_shares.py — the manage-access surface (wave 20, owner ruling R10, contract C-SHARE).

    GET  /api/v1/share/{kind}/{oid}       -> {owner, entries:[{user,role}], mayAdminister, people}
    PUT  /api/v1/share/{kind}/{oid}       <- {entries:[{user,role}]}   (REPLACES the set)
    PUT  /api/v1/share/{kind}/{oid}/owner <- {owner: "<username>"}     (R6d — `field` ONLY)
    GET  /api/v1/share/mine               -> {view:[id], folder:[id], database:[id]}

`kind` ∈ view | folder | database | field. Roles are `view` | `edit` — the same two words the
view rail already speaks, now extended to folders, databases and COLUMNS so there is ONE
vocabulary in the UI (R10: "the same picker views use").

⭐⭐ **W38-T16 — `field` IS THE FOURTH KIND, AND ITS `oid` IS TOPIC-QUALIFIED: `"<table_key>:<field_key>"`**
(`shares.field_oid`). A bare column key repeats across databases — `notes` exists on a dozen —
so a grant stored under one would admit the grantee to every `notes` column in the tenant at
once. ⛔ THREE functions in this file branch on kind and ALL THREE need the new one, which is not
obvious because only two of them fail loudly: `_owns_object` (without it a column's own creator
is 404'd trying to share the thing they just made) and `_object_ref` (without it `route` is None,
`_notify_new_grantees` returns early, and the grantee is **granted and never told** — owner item
18's silent half, reopened one kind over). `_can_see_object` stays deliberately CLOSED for
anything that is not a view.

⭐⭐ **A "Can edit" GRANTEE MAY RE-SHARE A VIEW — OWNER RULING R4, BUILT AS W40-T02** (instruction
4: *"Edit View so a member can share a View as well, not just an admin"*). This REVERSES the flat
owner-or-admin sentence that stood here, and the reversal is bounded three ways, all enforced HERE
and not in the client:

  1. **ONLY `kind='view'`.** `shares.RESHARE_KINDS` is the one spelling of that. `folder`,
     `database` and `field` still require the owner or an admin, and the paragraph below is why
     the `database` kind in particular was never a candidate: a `ut_*` grant's blast radius is a
     whole table, where a view is one saved SELECTION over rows the receiver's own wall governs.
  2. **A RE-SHARE MAY NEVER EXCEED THE RE-SHARER'S OWN ROLE**, and the strict reading ships: an
     `edit` grantee may hand out `view` and NEVER `edit`. `shares.max_grantable_role` answers the
     ceiling, `put_share` enforces it on the DELTA (a name arriving at `edit`, or one raised to
     it) — never on every row of the body, because the `PUT` REPLACES and `ShareDialog.save`
     therefore re-sends the whole list with the re-sharer's own `edit` row inside it. Conferring
     `edit` stays the owner's alone.
  3. **OWNERSHIP NEVER MOVES.** A caller re-sharing rather than owning passes the EXISTING owner
     straight through (`put_share` below, stated rather than incidental).

⭐⭐ **AND R4 BOUNDED ONLY THE ROLE. TWO MORE BOUNDS ARRIVED WITH THE OWNER'S RULING OF
2026-08-24, BECAUSE THE GAPS WERE MEASURED ON THE SHIPPED BUILD:**

  4. **A RE-SHARER MAY SHARE ONLY WITH SPECIFIC PEOPLE (D-473).** Owner: *"a re-sharer can only
     share to specific people"*. Before this, `[fisch:edit, *:view]` from an `edit` grantee was a
     `200` — the role ceiling could not see it, because `*` at `view` never exceeds a `view`
     ceiling. `_audience_added` refuses an `EVERYONE` entry this caller is ADDING, and lets one
     the OWNER already placed ride along untouched, for the same delta reason as (2).
  5. **A RE-SHARER MAY UN-SHARE ONLY THE PEOPLE THEY SHARED TO (D-474).** Owner: *"a re-sharer
     can only unshare the people it shared to"*. Before this, an `edit` grantee PUTting a list
     that omitted another grantee got `200` and that person's access was gone; `PUT []` left the
     view shared with nobody. Answering this needs a fact the registry did not store, so each
     entry now carries `by` — who placed it — stamped by `shares.set_grants(granter=…)` and
     consulted by `_unremovable`. ⛔ An entry with NO `by` (every grant predating the change) is
     the owner's or an admin's to revoke and nobody else's: fail-closed, because from the door
     "unstamped" and "somebody else's" are the same observation.

⛔ **SO BOTH FEARS THE OLD SENTENCE NAMED ARE NOW ANSWERED, AND THE PARAGRAPH THAT PRICED THE
SECOND ONE IS GONE RATHER THAN SOFTENED.** It read *"'Widen a users-scoped view to everyone' is
NOT closed — an `edit` grantee may add `*` — it is CAPPED at Can view"*, and that was an accurate
account of R4 that the owner overruled the moment it was put in front of them. *"Grant themselves
ownership and lock you out"* stays closed by (3) and `set_grants`' sticky owner. A `view` grantee
still cannot share at all, and still gets `403`. The client greys the editor for
non-administrators and offers "Can edit" to anyone it greys in; that is a courtesy, and these
checks are the wall.

⚠ **WHAT `GET` NOW DISCLOSES.** The grant record carries `by`, and this route returns the record
verbatim — so anybody who may read a view's grant list also learns who added each person, a
`view` grantee included. Accepted deliberately (the alternative is a second, stripped read path
and therefore a second answer to one question), and stated here because this file states costs
rather than leaving them to be discovered.

⚠ **THE GRANT NEVER WIDENS PAST THE MODULE WALL — ON A GOVERNED MODULE.** `*` ("everyone") means
every account that can already open the surface: `require_session` plus the topic's own gate run
first, and for `customer_data` / `product_data` the receiver's own row scope and hidden-field
closure run BEFORE any foreign view is merged. Sharing there can only narrow-or-equal the set that
could already reach the data ([[aios-permissioning]]).

⛔⛔ **AND THAT SENTENCE IS FALSE FOR `kind='database'`, WHICH IS WHY IT NOW SAYS "ON A GOVERNED
MODULE" (W32-T26, audit S-8).** `routes_admin._PERM_MODULES` is `("customer_data","product_data")`
and `_clean_perms` **400s** on anything else, so **no row filter and no hidden field can even be
DECLARED for a `ut_*` database** — `routes_tables.py` makes zero `perm_scope` calls and passes
`hidden_keys=frozenset()`. There is no module wall behind a user table for a grant to be bounded
by: **this registry IS the wall.** So a `database` grant is ALL-OR-NOTHING — every row, every
column — and an `*` database grant admits every account in the tenant to all of it.
That is a real capability, deliberately kept; what was wrong was a docstring promising a second
wall that does not exist for this kind. Scoping user tables is booked, not done
(`waves/wave32/sharing-audit.md` S-8).

⚠ **TWO SYSTEMS ANSWER "IS THIS SHARED", AND THEY ARE NOT THE SAME ONE (audit S-4).** THIS
registry decides who appears in *"Shared with me"* and who may re-share. **`table_store.is_shared`
— the view's own `permissions` — is what actually decides who may OPEN a view.** A grant here
whose object is invisible under that one is a row in a list that opens a refusal, which is what
made item 18 worth auditing. `_entries_or_400` closes the common cause (a name nobody has), but
the two vocabularies are still two.
"""
from fastapi import APIRouter, Body, Depends

import core.shares as shares
import core.users as users
from deps import Session, err, require_session
# ⭐ W32-T28 (C3) — the SHARE notification's topic word, imported from the module that CLASSIFIES
# it (`routes_alerts.notification_view`) rather than typed again here. The producer and the
# reader agreeing about one string is the whole difference between an Inbox row that opens the
# shared database and one that is quietly unclickable.
from routes_alerts import SHARE_TOPIC as _SHARE_TOPIC

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


def _kind_or_400(raw):
    try:
        return shares._check_kind(raw)
    except ValueError as e:
        raise err(400, "bad_kind", str(e))


# ── ⭐⭐ WAVE 32 · T26 (owner item 18, ruling R12) — THE WALL THIS FILE SAID IT HAD ─────────────
#
# `put_share`'s comment used to justify the first-claim rule with *"reaching this route at all
# means passing the surface's own wall"*. **There was no such wall.** `kind` and `oid` are free
# strings off the URL and the only dependency was `require_session`, so any signed-in account
# could `PUT` a grant on an id it had never seen. Because the 403 sat behind `if rec["owner"]`,
# an object with no grant record skipped the check entirely and the caller was stamped OWNER —
# sticky, so **the real creator was then refused on their own view, permanently.** Driven, not
# argued: `waves/wave32/sharing-audit.md` S-1 carries the four-step transcript.
#
# ⚠ AND IT WAS SILENT ON BOTH SIDES. The claimant does not even see the object in their own
# "Shared with me" (`shared_with` excludes what you own), so nothing appears anywhere until the
# victim next opens the dialog.

#: The built-in grid topics. A view or folder lives in `{topic}_table_workspace`, and the share
#: route is not told which topic — so resolving one means asking each.
_BUILTIN_TOPICS = ("customer", "product")


def _field_storage_keys(table_key):
    """Resolve the client-facing field topic to its durable stores and grant topic."""
    raw = str(table_key or "").strip()
    if raw in ("customer_data", "customer_table_workspace"):
        return "customer_table_workspace", "customer_table_workspace", "customer_data"
    if raw in ("product_data", "product_table_workspace"):
        return "product_table_workspace", "product_table_workspace", "product_data"
    if raw.startswith("ut_"):
        bare = raw[:-len("_table_workspace")] if raw.endswith("_table_workspace") else raw
        return f"{bare}_table_workspace", bare, bare
    workspace = raw if raw.endswith("_table_workspace") else f"{raw}_table_workspace"
    shared = raw[:-len("_table_workspace")] if raw.endswith("_table_workspace") else raw
    return workspace, shared, shared


def _field_definition(session, table_key, field_key):
    """Return the shared/private definition and the keys used by its write paths."""
    workspace_key, shared_key, grant_topic = _field_storage_keys(table_key)
    try:
        from core import shared_overlay
        shared = (shared_overlay.fields(shared_key, st=session.runtime) or {}).get(field_key)
        if isinstance(shared, dict):
            return shared, True, workspace_key, shared_key, grant_topic
        import core.table_store as table_store
        private = (table_store.make(workspace_key, st=session.runtime)
                   .workspace(session.uname).get("fields") or {}).get(field_key)
        if isinstance(private, dict):
            return private, False, workspace_key, shared_key, grant_topic
    except Exception:                                       # noqa: BLE001
        pass
    return None, False, workspace_key, shared_key, grant_topic


def _field_owner(session, definition, already_shared):
    """Resolve the creator for the share claim wall.

    A private field is already namespaced by the caller's own workspace.  Older field records
    from before the host-side creator stamp therefore remain safely claimable by that workspace
    owner, while a shared definition with no creator stays admin-only because its storage is
    tenant-wide and cannot identify an owner from residency alone.
    """
    owner = str((definition or {}).get("createdBy") or "").strip()
    if owner:
        return owner
    return str(session.uname or "").strip() if not already_shared else ""


def _topics(session):
    """Every topic whose workspace could hold a view or folder for this tenant.

    ⚠ `all_defs`, never `all_tables` — the latter is the whole 28.6 MB row payload (~703 ms on
    tenant #0) to answer a question about KEYS (D-185).
    """
    try:
        import core.user_tables as ut
        return (*_BUILTIN_TOPICS, *(ut.all_defs(st=session.runtime) or {}))
    except Exception:                                       # noqa: BLE001
        return _BUILTIN_TOPICS


def _owns_object(session, kind, oid):
    """May this caller CLAIM an object that has no grant record yet — i.e. do they own it?

    ⛔ THIS GUARDS THE CLAIM, NOT THE READ, AND THAT IS DELIBERATE. Resolving a view means asking
    each topic's workspace in turn, which is N store reads; making every share call pay that
    would put a loop on a route the manage-access dialog opens. The dangerous path is the one
    where a caller is about to be stamped OWNER of something nobody owns — so the resolution runs
    exactly there, and the common path (a record exists, `may_administer` decides) is untouched.
    """
    if kind == "cohort":
        scope, cohort_id = shares.split_cohort_oid(oid)
        source = shares.cohort_source(scope, cohort_id, st=session.runtime) if scope else None
        return bool(source and source["owner"] == str(session.uname or "").strip().lower())
    if session.admin:
        return True
    if kind == "field":
        # ⭐⭐ W38-T16 — A COLUMN'S OWNER IS ITS `createdBy`, WHICH THE CREATE DOOR ALREADY STAMPS
        # (`routes_tables.patch_shared_cell`) and the DELETE door already reads as its wall (R8 /
        # D-172: creator-or-admin). Read from the same place by all three, so a column cannot be
        # deletable by one person and shareable by another.
        # ⚠ THIS BRANCH IS NOT OPTIONAL AND ITS ABSENCE FAILS SILENTLY IN THE WORST DIRECTION:
        # a brand-new column has no grant record, so `put_share` falls to this predicate — and
        # without it the column's own creator is answered `404 no_object` on the first attempt to
        # share the thing they just made.
        table_key, field_key = shares.split_field_oid(oid)
        if not table_key:
            return False
        defn, _shared, _workspace, _shared_key, _grant_topic = _field_definition(
            session, table_key, field_key)
        owner = _field_owner(session, defn, _shared)
        return bool(defn) and owner.lower() == str(session.uname).strip().lower()
    if kind == "database":
        # ⚠ `may_open` is THE resolver for a user table (its own docstring says so) and already
        # admits creator, admin, or a `database` grantee. Re-implementing "who owns a table"
        # here would be the second definition this wave keeps finding.
        try:
            import core.user_tables as ut
            return bool(ut.may_open(oid, session.uname, is_admin=session.admin,
                                    st=session.runtime))
        except Exception:                                   # noqa: BLE001
            return False
    try:
        import core.table_store as table_store
    except Exception:                                       # noqa: BLE001
        return False
    for topic in _topics(session):
        try:
            ops = table_store.make(f"{topic}_table_workspace", st=session.runtime)
            hit = ops.find_view(oid) if kind == "view" else ops.find_folder(oid)
        except Exception:                                   # noqa: BLE001
            continue
        if hit:
            # `find_view`/`find_folder` answer `(owner_username, …)`. The claim belongs to the
            # person whose personal stratum holds it — anybody else reaching this line is
            # exactly the case S-1 describes.
            return str(hit[0]) == str(session.uname)
    return False


def _can_see_object(session, kind, oid):
    """May this caller READ an object's grant list — i.e. can they reach the object at all?

    ⛔⛔ THIS IS DELIBERATELY WIDER THAN {@link _owns_object}, AND CONFLATING THE TWO IS A
    REGRESSION I SHIPPED AND CAUGHT. The first version of T26 guarded BOTH doors with the
    ownership test, which reads sensibly and is wrong for the read, because **`find_view` searches
    PERSONAL STRATA ONLY** (its own docstring says so). So a view living in alice's stratum with
    `permissions.edit = "collaborative"` and no grant record yet — a view bob **can open and edit
    in the grid** — answered `404` when bob opened its manage-access dialog. Measured before
    fixing: `table_store._may_see(view, "bob") is True` while `GET /share/view/vc` said
    `404 no_object`.
    ⚠ THAT IS THE AUDIT'S OWN S-4 BITING THE AUDIT'S OWN FIX: two systems answer "is this shared",
    and the wall consulted the grant registry (system A) plus stratum ownership, never the view's
    `permissions` (system B) — which is the one that actually decides who may OPEN it.
    ⚠ And it hides the ANSWER, not just the editor. `ViewSidebar`'s Share row is deliberately not
    gated on edit rights because *"hiding the row from everyone else would hide the ANSWER too —
    'who has this?' is a fair question for anyone the view was shared with"*. A 404 there tells a
    legitimate collaborator their view does not exist.

    ⛔ THE CLAIM KEEPS THE NARROW TEST. Being able to SEE an object must not let you become its
    owner — that is S-1, and widening this predicate onto `put_share` would re-open it.
    """
    if _owns_object(session, kind, oid):
        return True
    if kind != "view":
        # A folder carries no per-object visibility flag of its own, and a database's `may_open`
        # (inside `_owns_object`) already admits grantees. Nothing wider to ask.
        # ⭐ W38-T16 — AND `field` KEEPS THIS CLOSED, DELIBERATELY. A grantee never reaches here:
        # `get_share` tests `role is None` first and a grant answers a role, so the only caller
        # left is an account with no relationship to the column at all. Widening it would let any
        # signed-in session enumerate who holds which column on a database they cannot open.
        return False
    try:
        import core.table_store as table_store
        for topic in _topics(session):
            hit = table_store.make(f"{topic}_table_workspace", st=session.runtime).find_view(oid)
            if hit:
                return bool(table_store._may_see(hit[1] if len(hit) > 1 else {},
                                                 session.uname, is_admin=session.admin))
    except Exception:                                       # noqa: BLE001
        return False
    return False


def _entries_or_400(session, entries, refusal=None):
    """Validate a grant list against the tenant's REAL, ACTIVE accounts — and refuse BY NAME.

    ⛔ `core.shares._clean_entries` silently drops junk, and its docstring argues that correctly:
    a UI mid-save must not lose the whole list to one malformed row. **But it validates the SHAPE
    of a string and the role word — never that the user EXISTS, is ACTIVE, or is in this tenant**,
    so a typo'd name is stored, reported as a successful save, and never reaches anybody. The
    sharer believes the person has access. That is item 18's plain reading.
    ⚠ The correct population is computed THREE FUNCTIONS BELOW and served to the picker
    (`_people`). One route, two populations, and the write door was the permissive one.
    ⚠ `*` (everyone) is not a user and is admitted deliberately — it is R10's vocabulary for
    "every account that can already open the surface". ⛔ A CALLER THAT NEEDS ONE REAL PERSON MUST
    THEREFORE REFUSE `*` ITSELF, BEFORE ASKING HERE — see `reassign_owner`, where "everyone" is
    not an answer to "who owns this column".

    ⭐ W41-T03 — `refusal` IS THE SENTENCE AFTER THE NAME, AND ITS DEFAULT IS BYTE-IDENTICAL to
    what this function has always printed. The tenant-registry check is the part that must not be
    written twice ([[one-evaluator-per-question]]); the advice at the end of it is the part that
    has to match the door the reader is standing at, because "Nothing was shared" is the wrong
    account of a refused OWNER change. One validator, one population, one code (`unknown_people`,
    which is what clients match on) — and a tail the caller owns.
    """
    known = {p["username"].strip().lower() for p in _people(session.tenant)}
    unknown = []
    for e in entries or ():
        if not isinstance(e, dict):
            continue
        user = str(e.get("user") or "").strip().lower()
        if user and user != shares.EVERYONE and user not in known:
            unknown.append(user)
    if unknown:
        raise err(400, "unknown_people",
                  "no active account in this workspace is named "
                  + ", ".join(sorted(set(unknown))) + ". "
                  + (refusal or "Nothing was shared. Pick people from the list rather than "
                                "typing a name."))


def _named(who):
    """The people in a refusal, in the words the person reading it uses.

    ONE spelling, used by all three re-share refusals below. `*` is never printed raw: the store's
    wildcard is a single character, and a 403 reading *"set * to Can view"* names nothing a person
    can find in the dialog they are looking at.
    """
    return ", ".join("everyone in this workspace" if w == shares.EVERYONE else w
                     for w in sorted(who))


def _audience_added(entries, held):
    """⭐⭐ D-473 — the `EVERYONE` grant this caller is ADDING, or the empty set.

    OWNER RULING 2026-08-24, verbatim: *"a re-sharer can only share to specific people"*. R4's
    ceiling bounded the ROLE a re-sharer may hand out and said nothing about the AUDIENCE, and the
    gap was measured rather than argued: an `edit` grantee PUTting `[fisch:edit, *:view]` was
    answered `200`, widening a two-person view to the whole tenant at `Can view`. The role check
    could not catch it, because `*` at `view` never exceeds a `view` ceiling.

    ⛔ IT IS THE ADDITION THAT IS REFUSED, NOT THE PRESENCE, and that is the same shape as the
    role check one arm above, for the same reason: the `PUT` REPLACES, so `ShareDialog.save`
    re-sends the WHOLE list every time. An `*` the OWNER placed rides along in every payload the
    re-sharer is able to produce, and refusing on presence would `403` every save on a view the
    owner had already opened to everyone — a re-sharer locked out of a list they may legitimately
    edit, with a message about a row they did not touch.

    ⛔ ROLE IS NOT CONSULTED HERE, DELIBERATELY. This answers "may this caller widen the
    AUDIENCE", and `*` held at `view` and resubmitted at `edit` is a ROLE escalation that the
    check above already refuses, by name. Two questions, two predicates
    ([[one-evaluator-per-question]]) — and separable is also what lets a gate disarm one of them
    in memory and prove the other still fires.

    ⚠ `_clean_entries`, NEVER THE RAW BODY: an entry with a junk role is dropped by the writer,
    so reading the raw list would refuse a widening that was never going to be stored.
    """
    if shares.EVERYONE in (held or {}):
        return set()
    return {e["user"] for e in shares._clean_entries(entries) if e["user"] == shares.EVERYONE}


def _unremovable(held_rows, entries, uname):
    """⭐⭐ D-474 — the people this caller is dropping from the list but may NOT revoke.

    OWNER RULING 2026-08-24, verbatim: *"a re-sharer can only unshare the people it shared to"*.
    R4 bounded what a re-sharer may HAND OUT and left what they may TAKE AWAY unbounded, and both
    halves of that were measured: an `edit` grantee PUTting a list that omits another grantee was
    answered `200` and that person's access was gone; `PUT []` left the view shared with nobody.

    ⛔⛔ AN ENTRY WITH NO `by` IS NOT REMOVABLE BY A RE-SHARER — ONLY BY THE OWNER OR AN ADMIN,
    AND THAT IS THE FAIL-CLOSED DIRECTION RATHER THAN AN OVERSIGHT. Every grant written before
    provenance existed carries no stamp, so "no `by`" and "granted by somebody else" are
    indistinguishable from here. Reading absence as *"nobody claims it, so anyone may take it"*
    would hand every re-sharer the power to revoke the entire pre-existing grant set on day one
    of this change, which is the exact capability the ruling withholds. A re-sharer must not be
    able to revoke a grant they cannot PROVE they made [[aios-permissioning]].

    ⚠ SO A RE-SHARER CANNOT REMOVE THEMSELVES EITHER, AND THAT IS STATED BECAUSE IT LOOKS LIKE A
    BUG. Their own row was placed by the owner, so it carries the owner's `by` and lands in this
    set. Read literally, the ruling says a re-sharer unshares only who THEY shared to, and their
    own grant is not one of those. Leaving the view is the owner's to do, like every other
    revocation of an owner-placed grant. ⛔ Do not carve an exception here without a ruling: the
    carve-out is indistinguishable from "a re-sharer may revoke any row whose `by` names the
    owner", which is the wall itself.

    ⚠ `_clean_entries`, NEVER THE RAW BODY, AND THIS IS THE HOLE THAT SHAPE CLOSES. A role the
    writer rejects is a row that will NOT be stored, so `{user: victim, role: "nonsense"}` looks
    present in the raw payload and is a silent REVOCATION once written. Asking the same
    normaliser the store uses is what makes "submitted" mean the same thing at both ends.
    """
    me = str(uname or "").strip().lower()
    submitted = {e["user"] for e in shares._clean_entries(entries)}
    stuck = set()
    for user, row in (held_rows or {}).items():
        if user in submitted:
            continue
        stamp = str((row or {}).get("by") or "").strip().lower()
        if not me or not stamp or stamp != me:
            stuck.add(user)
    return stuck


@router.get("/share/mine")
def my_shares(session: Session = Depends(require_session)):
    """Everything shared WITH me, by kind — the "Shared with me" rail section (R10).

    Registered before `/share/{kind}/{oid}` so the literal path wins the match; FastAPI resolves
    in declaration order and `mine` would otherwise be read as a `kind`, answering 400 for a URL
    that is not malformed at all.
    """
    return shares.shared_with(session.uname, st=session.runtime)


@router.get("/cohorts/{scope}")
def list_cohorts(scope: str, session: Session = Depends(require_session)):
    """The authorized Cohort list for one database.

    Unlike the former owner-bucket projection, this calls the same predicate as the Cohort
    detail below.  An account without a grant receives no entry at all; it never receives an
    id/name shell whose membership has been stripped.
    """
    try:
        records = shares.visible_cohorts(scope, session.uname, st=session.runtime)
    except ValueError as exc:
        raise err(400, "bad_cohort_scope", str(exc))
    return {"cohorts": [{"id": cid, **record} for cid, record in sorted(records.items())]}


@router.get("/cohorts/{scope}/{cohort_id}")
def get_cohort(scope: str, cohort_id: str, session: Session = Depends(require_session)):
    """One authorized Cohort, with an absent-permission answer indistinguishable from absence."""
    try:
        record = shares.cohort_record(scope, cohort_id, session.uname, st=session.runtime)
    except ValueError as exc:
        raise err(400, "bad_cohort_scope", str(exc))
    if record is None:
        raise err(404, "no_cohort", "no such cohort, or it is not shared with this account")
    return {"id": record["id"], **record["cohort"]}


@router.get("/share/{kind}/{oid}")
def get_share(kind: str, oid: str, session: Session = Depends(require_session)):
    kind = _kind_or_400(kind)
    if kind == "field":
        table_key, field_key = shares.split_field_oid(oid)
        if table_key and field_key:
            oid = shares.field_oid(_field_storage_keys(table_key)[2], field_key)
    rec = shares.grants(kind, oid, st=session.runtime)
    # A Cohort is visible only to its source owner or an explicit grant.  In particular, an
    # administrator-shaped Farhan session must not turn Shantal's absent grant into a record.
    # Other share kinds retain their established administrator behaviour.
    _admin_share = bool(session.admin) and kind != "cohort"
    role = shares.role_for(kind, oid, session.uname, is_admin=_admin_share, st=session.runtime)
    may_admin = shares.may_administer(kind, oid, session.uname, is_admin=_admin_share,
                                      st=session.runtime)
    # W39-T29 — the dialog opens with GET before its first PUT.  Until that PUT exists the grant
    # registry has no owner to return from `may_administer`, even though the same caller may safely
    # claim their own object through the PUT path below.  Reflect that exact claim predicate here:
    # a Member who owns an unshared View receives the people picker; a collaborator still does not.
    if not rec["owner"] and not may_admin and _owns_object(session, kind, oid):
        may_admin = True
    # ⭐ W32-T26 (audit S-3) — A STRANGER LEARNS NOTHING. This route used to answer for ANY id:
    # who owns it, everyone it is granted to, and the tenant's whole username↔name directory —
    # to any signed-in session, about objects it cannot open. Now a caller with no role on an
    # object must prove they can reach it, and gets a 404 otherwise: the same answer a
    # non-existent id gives, so the route cannot be used to probe which ids are real.
    # ⚠ `role is None` is the cheap pre-test, so the N-topic resolution below runs only for a
    # caller who has no relationship with the object at all.
    if role is None and not _can_see_object(session, kind, oid):
        raise err(404, "no_object", "no such item, or it is not shared with this account")
    return {
        **rec,
        "role": role,
        "mayAdminister": may_admin,
        # ⚠ WAVE 21 (C1 identity fix): grant entries BIND on USERNAMES, so the picker must carry
        # them. `assignable_people` serves bare display names because `user`-kind CELLS store
        # display names — that list's shape cannot change without migrating cell values — so
        # this route serves objects of its own. Existing grants that were written as lowercased
        # display names are normalised by the wave-21 cleanup script.
        # ⭐ W32-T26 (audit S-3) — the roster is the EDITOR's data, so it rides only for a caller
        # who may open the editor. A read-only grantee gets the grant list (their fair question is
        # "who else has this?") and not a directory of every account in the workspace.
        # ⭐⭐ W40-T02 / R4 — AND THAT RULE IS WHY THIS LINE NEEDED NO EDIT. `may_administer` now
        # answers True for an `edit` grantee on a VIEW, which MOVES that account into "may open
        # the editor" — so the picker they need arrives by the roster riding on the same flag it
        # always did. Gating it on anything else (owner, `role == 'owner'`, a fresh predicate)
        # would be a second answer to a question this file already answers once, and would leave
        # the new grantee with an editor and no people to put in it. A `view` grantee is still
        # `may_admin=False` here and still gets `[]`.
        "people": _people(session.tenant) if may_admin else [],
    }


def _people(tenant):
    """[{username, name}] for this tenant — same population as `assignable_people`, with the
    BINDING identity alongside the display one."""
    try:
        reg = users.registry() or {}
    except Exception:
        return []
    want = str(tenant or '').strip().lower()
    out = []
    for uname, u in reg.items():
        if not isinstance(u, dict) or u.get('active') is False:
            continue
        if want and str(u.get('tenant') or 'royal-imports').strip().lower() != want:
            continue
        out.append({"username": str(uname), "name": str(u.get('name') or uname)})
    return sorted(out, key=lambda p: p["name"].lower())


@router.put("/share/{kind}/{oid}")
def put_share(kind: str, oid: str, body: dict = Body(default=None),
              session: Session = Depends(require_session)):
    kind = _kind_or_400(kind)
    if kind == "field":
        table_key, field_key = shares.split_field_oid(oid)
        if table_key and field_key:
            oid = shares.field_oid(_field_storage_keys(table_key)[2], field_key)
    body = body or {}
    rec = shares.grants(kind, oid, st=session.runtime)
    # `claiming` is the "no owner yet, and this caller may become one" branch, hoisted to a name
    # because TWO decisions below need it: the ceiling (a claimant is about to be the owner, so
    # their ceiling is an owner's) and the owner written back (D3 — see `set_grants` at the end).
    claiming = False
    # An object with NO grant record yet has no owner — the first person to share it claims it.
    # That is safe because reaching this route at all means passing the surface's own wall, and
    # the alternative (refusing until somebody seeds an owner) would make a brand-new folder
    # unshareable by the person who just made it.
    if rec["owner"]:
        if not shares.may_administer(kind, oid, session.uname,
                                     is_admin=bool(session.admin) and kind != "cohort",
                                     st=session.runtime):
            # ⭐ R4 / W40-T02 — `may_administer` now also admits an `edit` grantee on a VIEW, so
            # the population refused here is narrower than the code word `not_owner` suggests: a
            # `view` grantee, or an account with an `edit` role on a kind outside
            # `shares.RESHARE_KINDS`. The code string is kept because clients match on it.
            raise err(403, "not_owner",
                      "only the owner of this item (or an administrator) can change who it is "
                      "shared with")
    # ⛔⛔ W32-T26 (audit S-1) — THE CLAIM NOW HAS A PRECONDITION. An object with no grant record
    # is still claimed by the first person to share it — that rule is right, and refusing until
    # somebody seeds an owner would make a brand-new folder unshareable by the person who just
    # made it. What was missing is the half the old comment ASSERTED and the code never did: the
    # claimant has to be able to reach the object. Without this, any signed-in account could
    # stamp itself owner of an id it had never seen and lock the real creator out for good.
    elif not _owns_object(session, kind, oid):
        raise err(404, "no_object", "no such item, or it is not shared with this account")
    else:
        claiming = True
    entries = body.get("entries")
    if not isinstance(entries, list):
        raise err(400, "bad_entries",
                  "entries must be a list of {user, role}. Send [] to un-share, which is how "
                  "revoking is expressed")
    _entries_or_400(session, entries)
    # ⭐⭐ R4 / W40-T02 — THE CEILING. `may_administer` above now opens this door to an `edit`
    # grantee on a VIEW, so R4's other half ("a re-share may never exceed the role the re-sharer
    # holds") needs a check of its own: that caller may hand out `view`, and conferring `edit`
    # stays the owner's or an administrator's.
    #
    # ⛔ AFTER THE ADMISSION, NEVER BEFORE, AND THAT ORDER IS A SECURITY PROPERTY. A caller with
    # no role at all must keep receiving `404 no_object` (audit S-1/S-3: a stranger learns
    # nothing, so this route cannot be used to probe which ids are real). A ceiling raised first
    # would answer that caller `403` and turn the one route hardened against id-probing back into
    # an oracle that confirms an id exists. It also runs before the `field` promotion below, so a
    # refusal cannot leave a column promoted with no grant written.
    #
    # ⛔ AND IT IS THE DELTA, NOT EVERY ROW OF THE BODY — read off the shipped client, not
    # assumed. `ShareDialog.save` PUTs the WHOLE list every time ("a body assembled from a delta
    # would revoke everyone it failed to mention"), so the re-sharer's OWN `{user, role: "edit"}`
    # row rides in every payload they are able to produce. Refusing per-entry would `403` the
    # exact re-share this ticket exists to enable, and the only body that would pass is one that
    # revokes the re-sharer. So what is refused is edit access this caller is CREATING: a name
    # arriving at `edit`, or an existing `view` grantee raised to it. A row that already stood at
    # `edit` was the OWNER's decision, and is not this caller's to be refused for.
    #
    # ⚠ A CLAIMANT IS AN OWNER. The branch above admits a Member who owns an object that has no
    # grant record yet, and `set_grants` is about to stamp them owner — asking the registry for
    # their role here would answer `None` (no record exists to hold one) and refuse the first
    # `edit` grant on every newly created view. Same predicate as the door, one line apart.
    ceiling = "edit" if claiming else shares.max_grantable_role(
        kind, oid, session.uname, is_admin=session.admin, st=session.runtime)
    #
    # ⭐⭐ OWNER RULING 2026-08-24 (D-473 + D-474) — AND THE CEILING IS NOW ONE OF THREE WALLS IN
    # THIS BLOCK, NOT THE WALL. R4 bounded the ROLE a re-sharer may hand out and was silent on the
    # other two questions a re-share asks, so both gaps shipped and both were measured on the
    # build: an `edit` grantee could PUT `[fisch:edit, *:view]` and widen a two-person view to the
    # whole tenant (200), and could PUT a list omitting another grantee — or `[]` — and revoke
    # people they never granted (200). The owner's answer settles both in one sentence: *"no a
    # re-sharer can only share to specific people and a re-sharer can only unshare the people it
    # shared to"*.
    #
    # ⛔ THREE PREDICATES, THREE FUNCTIONS, ONE ORDER: role, then audience, then revocation. They
    # are separate because they answer separate questions and because a wall that cannot be
    # disarmed ALONE cannot be proven alone — `verify_scopes.section_reshare_bounds` patches each
    # one in memory and requires exactly its own leg to go red, which a single fused `if` would
    # make impossible ([[a-declared-gate-is-an-unchecked-claim]]).
    # ⚠ ROLE AND AUDIENCE COMPOSE, AND THE ORDER DECIDES WHICH REFUSAL A PERSON READS. `*`
    # submitted at `edit` while held at `view` is BOTH an escalation and (if unheld) a widening;
    # the role check runs first and names the fix that is actually available to this caller
    # ("set it to Can view"), which is the more useful of the two sentences.
    if ceiling != "edit":
        # The prior ROWS, not just their roles: the revocation wall needs each entry's `by`, and
        # reading it from a second place would be a second answer to "what does the store hold".
        held_rows = {e.get("user"): e for e in (rec["entries"] or ())
                     if isinstance(e, dict) and e.get("user")}
        held = {u: e.get("role") for u, e in held_rows.items()}
        noun = {"view": "view", "folder": "folder",
                "database": "database", "field": "column"}.get(kind, "item")
        raised = set()
        for e in entries:
            if not isinstance(e, dict):
                continue
            who = str(e.get("user") or "").strip().lower()
            if who and str(e.get("role") or "").strip().lower() == "edit" \
                    and held.get(who) != "edit":
                raised.add(who)
        if raised:
            raise err(403, "grant_exceeds_role",
                      "you can share this " + noun + " at Can view, which is as far as your own "
                      "access reaches. Only its owner (or an administrator) can give somebody "
                      "Can edit, so nothing was saved. Set " + _named(raised)
                      + " to Can view and save again.")
        # ⭐⭐ D-473 — THE AUDIENCE. A re-sharer names PEOPLE; reaching "everyone" is the owner's.
        if _audience_added(entries, held):
            raise err(403, "grant_exceeds_audience",
                      "you can share this " + noun + " with specific people, which is as far as "
                      "your own access reaches. Only its owner (or an administrator) can open it "
                      "to everyone in this workspace, so nothing was saved. Remove Everyone from "
                      "the list, add the people you meant by name, and save again.")
        # ⭐⭐ D-474 — THE REVOCATION. An omission IS a revocation on a replacing PUT, so this is
        # the only place a removal can be refused. ⛔ REFUSED WHOLE: `set_grants` has not run, so
        # a payload carrying a legitimate addition ALONGSIDE a forbidden removal saves neither.
        # That is deliberate and it is what the message promises ("nothing was saved") — a
        # half-applied permission change is worse than a refused one, because the person reading
        # the toast has no way to tell which half took.
        stuck = _unremovable(held_rows, entries, session.uname)
        if stuck:
            raise err(403, "revoke_not_yours",
                      "you can remove the people you shared this " + noun + " with, and this "
                      "workspace has no record of you sharing it with " + _named(stuck)
                      + ". Only its owner (or an administrator) can remove them, so nothing was "
                      "saved. Put them back on the list and save again.")
    if kind == "field":
        # A field grant is a visibility and edit wall. Promote a private custom
        # field exactly once, then keep the requested Share field role as the
        # authoritative override for the legacy permissions bag.
        from core import field_permissions, shared_overlay
        table_key, field_key = shares.split_field_oid(oid)
        defn, already_shared, workspace_key, shared_key, grant_topic = _field_definition(
            session, table_key, field_key)
        if not isinstance(defn, dict):
            raise err(404, "no_object", "no such field, or it is not shared with this account")
        if not already_shared:
            defn = field_permissions.promote_field(
                workspace_key, shared_key, grant_topic,
                session.uname, defn, st=session.runtime)
        stamped = dict(defn)
        stamped["shared"] = True
        stamped["granted"] = True
        shared_overlay.put_field(shared_key, field_key, stamped, st=session.runtime)
        oid = shares.field_oid(grant_topic, field_key)
    # ⭐⭐ R4 / W40-T02 (D3) — OWNERSHIP NEVER MOVES ON A RE-SHARE, AND IT IS SAID HERE RATHER
    # THAN LEFT TO FALL OUT. `set_grants`' owner is sticky, so the old `rec["owner"] or
    # session.uname` already happened not to transfer ownership — incidentally, as a property of
    # the callee. R4 names ownership transfer as one of the two halves of the old protection that
    # SURVIVES the widening, and a rule that survives by accident is one the next edit deletes
    # without noticing. So the branch is explicit: a claimant becomes the owner, and everybody
    # else — an owner re-saving, an admin, and now an `edit` grantee re-sharing — passes the
    # EXISTING owner straight back through. `claiming` is the same flag the admission set, so
    # there is no second answer to "is this person taking ownership".
    # ⭐⭐ D-474 — `granter` IS THIS SESSION, ON EVERY SAVE INCLUDING THE OWNER'S. Provenance is
    # recorded for whoever adds a person, not only for a re-sharer: an owner-placed grant carrying
    # NO stamp is indistinguishable from a pre-provenance one, and `_unremovable` would then be
    # deciding on the store's AGE rather than on who granted what. `set_grants` stamps only
    # entries that are NEW to the record and never re-stamps an existing one, so an owner
    # re-saving a list does not quietly take provenance off the re-sharer who built it.
    # ⛔⛔ THE THREE REFUSALS ABOVE WERE DECIDED AGAINST `rec`, WHICH WAS READ AT THE TOP OF
    # THIS FUNCTION. Handing `expect` to the writer is what makes them true at the moment of the
    # write rather than at the moment of the read: a concurrent save lands between the two, and
    # a wave-40 adversarial probe drove a re-sharer's PUT being ACCEPTED while the grant the
    # owner had just added disappeared. See `shares.set_grants`' own note.
    try:
        out = shares.set_grants(kind, oid, entries,
                                owner=session.uname if claiming else rec["owner"],
                                granter=session.uname,
                                st=session.runtime,
                                expect=rec["entries"])
    except shares.GrantsChanged as exc:
        raise err(409, "grants_changed", str(exc))
    _notify_new_grantees(session, kind, oid, before=rec["entries"], after=out.get("entries") or [])
    return out


@router.put("/share/{kind}/{oid}/owner")
def reassign_owner(kind: str, oid: str, body: dict = Body(default=None),
                   session: Session = Depends(require_session)):
    """⭐⭐ W41-T03 / RULING R6(d) — HAND ONE COLUMN TO A NEW OWNER. `{"owner": "<username>"}`.

    ⛔ A SEPARATE DOOR, NOT AN `owner` MEMBER ON `PUT /share/{kind}/{oid}`, AND THE REASON IS THE
    FAILURE MODE RATHER THAN TIDINESS. `GET` answers the stored record — `{owner, entries, ...}` —
    and `ShareDialog` parses the whole of it into its state. The moment any client re-serialises
    that state on save (today `shareModel.sharePutBody` returns `{entries}` alone, which is one
    refactor away from `{...state}`), an `owner` member on the share body would REASSIGN THE COLUMN
    because the client echoed back a key it had always been given. A door nobody can walk through
    by accident is the whole point [[a-guard-authorises-a-destination-not-a-payload]].

    ⛔ AND THE TWO DOORS DO NOT SHARE A WALL, WHICH IS THE SECOND AND STRONGER REASON. `put_share`
    admits whoever `shares.may_administer` admits, and since R4 that INCLUDES an `edit` grantee on
    a view. Ownership is the owner's and an administrator's alone, so this asks
    `shares.role_for(...) == 'owner'` instead. Fusing the two would put two different walls behind
    the presence of one key in one body ([[one-evaluator-per-question]]).

    ⭐⭐ `field` ONLY, AND THE REFUSAL FOR THE OTHER THREE KINDS IS A REAL FINDING RATHER THAN
    SCOPE-TRIMMING. This registry's `owner` is AUTHORITATIVE for a column: `field_permissions.
    field_class` (contract C1) reads it and falls back to `createdBy` only when there is no record,
    and `user_tables.delete_field_refusal` then derives R6's delete right from that bag. For the
    other three kinds it is NOT: a view or folder is found by `table_store.find_view`/`find_folder`
    searching PERSONAL STRATA, and a database resolves through `user_tables.may_open` — so moving
    the registry owner there would move the badge and leave both the rights and the residency
    behind, i.e. ship the half that is visible and not the half that is true. That is a bigger
    ticket than this one, so this door says no rather than pretending.

    ⚠ ORDER IS A SECURITY PROPERTY HERE, EXACTLY AS IN `put_share`: the wall runs BEFORE the new
    owner is validated, because validating a username consults `_people` — the tenant's whole
    account directory — and a stranger must not be able to use this route as a roster oracle or an
    id oracle (audit S-1/S-3). A caller with no relationship to the column keeps getting the same
    `404` a made-up id gets.

    ⚠ AND THE PREVIOUS OWNER IS NOT KEPT ON THE LIST. Reassignment moves the seat; it does not
    leave a consolation grant behind, because W41-T03's `done-when` requires the previous owner to
    STOP passing `may_delete_field`. An administrator who wants them to keep access adds them as an
    ordinary entry through the share door, which is the one place grants are decided.
    """
    kind = _kind_or_400(kind)
    if kind != "field":
        raise err(400, "kind_not_reassignable",
                  "only a column's owner can be changed here. A view, a folder and a database "
                  "each record their owner outside this list, so nothing was changed.")
    table_key, field_key = shares.split_field_oid(oid)
    if table_key and field_key:
        oid = shares.field_oid(_field_storage_keys(table_key)[2], field_key)
    rec = shares.grants(kind, oid, st=session.runtime)
    # ⛔⛔ THE WALL: AN ADMINISTRATOR OR THE CURRENT OWNER, AND NOBODY ELSE. `role_for` answers
    # `'owner'` for both (an admin reads as owner by its own rule), and it answers `'edit'` — not
    # `'owner'` — for a grantee. That distinction is load-bearing on this kind in particular:
    # `field_permissions.migrate_legacy_fields` writes `[{user: '*', role: 'edit'}]` on every
    # promoted legacy column, so a wall spelled `may_edit` or `may_administer` would hand EVERY
    # account in the tenant the right to reassign every migrated column.
    role = shares.role_for(kind, oid, session.uname, is_admin=session.admin, st=session.runtime)
    if rec["owner"]:
        if role != "owner":
            raise err(403, "not_owner",
                      "only the owner of this column (or an administrator) can hand it to "
                      "somebody else, so nothing was changed.")
    # ⚠ NO RECORD YET MEANS THE OWNER IS THE COLUMN'S CREATOR, and `_owns_object` is the one place
    # that reads `createdBy` for this question — the same predicate `put_share` claims through, one
    # function apart. A stranger falls out here as `404`, never `403`: a `403` would confirm the id
    # is real, which is the oracle S-3 closed.
    elif role != "owner" and not _owns_object(session, kind, oid):
        raise err(404, "no_object", "no such item, or it is not shared with this account")
    # ⛔ FAIL CLOSED ON A COLUMN THIS DOOR CANNOT RESOLVE, AND NAME WHICH HALF FAILED.
    defn, already_shared = None, False
    if table_key and field_key:
        defn, already_shared = _field_definition(session, table_key, field_key)[:2]
    if not isinstance(defn, dict):
        raise err(404, "no_object",
                  "this column could not be found on that database, so its owner was not changed.")
    # ⛔⛔ A PRIVATE COLUMN CANNOT BE HANDED OVER, AND REFUSING IS THE ONLY HONEST ANSWER. An
    # unpromoted definition lives in ONE account's own workspace stratum, so a reassignment would
    # write a registry owner who can never see the column while the creator — no longer the
    # registry owner — loses the share door on it. That is precisely the unmanageable object
    # `shares.set_grants`' sticky-owner note exists to prevent, arriving through a new door.
    # ⛔ AND THE FIX IS NOT TO PROMOTE IT HERE. `put_share` promotes because sharing is what the
    # caller asked for; promotion stamps the definition tenant-wide and would move contract C1's
    # `audience` badge from `private` to `everyone`. An ownership change must never widen an
    # audience as a side effect.
    if not already_shared:
        raise err(409, "field_not_shared",
                  "this column is still private to the person who made it, so it cannot be handed "
                  "to somebody else yet. Share it first, then change its owner.")
    body = body or {}
    new_owner = str(body.get("owner") or "").strip().lower()
    if not new_owner:
        raise err(400, "bad_owner",
                  "name the account that should own this column. Nothing was changed.")
    # ⚠ `_entries_or_400` ADMITS `*` DELIBERATELY (it is R10's word for "everyone"), so the one
    # refusal it cannot make for us is made here: a column is owned by a PERSON, and an owner of
    # `*` is an owner nobody can log in as.
    if new_owner == shares.EVERYONE:
        raise err(400, "bad_owner",
                  "a column is owned by one person, not by everyone in this workspace. Pick an "
                  "account by name. Nothing was changed.")
    # ⛔ THE SAME VALIDATOR THE SHARE DOOR USES, NEVER A SECOND ONE. A reassignment to a name no
    # active account in this tenant answers to would strand the column with an owner who cannot
    # sign in — the ownership-shaped version of exactly what item 18 was about.
    _entries_or_400(session, [{"user": new_owner, "role": "view"}],
                    refusal="Nothing was changed. Pick the new owner from the list rather than "
                            "typing a name.")
    # ⭐⭐ D-474 — `granter` IS DELIBERATELY NOT PASSED, AND THAT IS NOT AN OMISSION. This call
    # adds NOBODY: it writes back the entry list exactly as it was read, so every user is already
    # in the prior record and `set_grants` keeps each stored `by` VERBATIM. Naming a granter would
    # be inert on that path today and WRONG on the rebase path `set_grants` warns about — where a
    # user reclassified from "prior" to "new" would silently have their provenance transferred to
    # whoever reassigned the column, handing an administrator the right to revoke people a
    # re-sharer had granted. Absent, such a user gets no stamp at all, which is the fail-closed
    # direction `_unremovable` already reads as "the owner's alone to revoke".
    # ⛔ `expect` IS PASSED FOR THE REASON `put_share` PASSES IT: this write REPLACES the entry set
    # with the snapshot read at the top of this function, so a grant somebody added in between
    # would be silently deleted by an operation that is supposed to touch only the owner. The
    # compare-and-set turns that into a 409 the person can act on.
    # ⚠ WHAT `expect` DOES NOT COVER IS THE OWNER ITSELF — it compares `(user, role)` pairs only.
    # Two concurrent reassignments therefore both land, last write winning. Both callers held the
    # seat when they decided, so this is a lost update between authorised callers rather than an
    # escalation; widening the compare-and-set to the owner is a change to a shared primitive and
    # is booked rather than smuggled in here.
    try:
        return shares.set_grants(kind, oid, rec["entries"], owner=new_owner,
                                 st=session.runtime, expect=rec["entries"])
    except shares.GrantsChanged as exc:
        raise err(409, "grants_changed", str(exc))


def _notify_new_grantees(session, kind, oid, before, after):
    """⭐⭐ W32-T28 (owner item 18's last clause, contract C3) — tell the RECEIVER, in their Inbox.

    Owner item 18 ends *"being shared a database notifies the receiver"*. Until now sharing was
    silent: the grant landed in a rail section the receiver had to notice on their own, which is
    why "I shared it with you" and "I never saw it" were both true.

    ⛔ WRITTEN ON THE SHARE, NEVER POLLED. `/notifications` re-evaluates view-ALERTS on read
    because an alert is a live question about rows; a share is an EVENT that happened once, and
    polling for it would mean re-deriving "was this new?" on every inbox open — the diff below
    only exists here, at the moment the set changes.

    ⚠ ONLY THE NEWLY ADDED. `PUT` REPLACES the whole entry set (revoking is expressed by absence),
    so every save re-sends everyone who was already there. Diffing against `before` is what stops
    a rename or a role change from ringing the bell for people whose access did not change.
    ⚠ `*` IS NOT NOTIFIED: there is no user to name, and minting one notification per account in
    the tenant on a single click is a broadcast nobody asked for. The rail still shows it.
    ⚠ IT NEVER RAISES. A notification that fails must not fail the share that triggered it — the
    grant is the user's actual intent, and `core.alerts.notify` writes with `flush='async'`.
    """
    try:
        was = {e.get("user") for e in (before or ()) if isinstance(e, dict)}
        fresh = [str(e.get("user")) for e in (after or ())
                 if isinstance(e, dict) and e.get("user") not in was
                 and e.get("user") != shares.EVERYONE]
        if not fresh:
            return
        import core.alerts as alerts

        label, route, view_id = _object_ref(session, kind, oid)
        if not route:
            # ⛔ NO ROUTE, NO NOTIFICATION — the receiver would get a row that opens nothing, and
            # `notification_view` would have to invent a target. Silence is the honest answer
            # here; the rail still shows the grant under "Shared with me".
            return
        sharer = str(session.user.get("name") or session.uname)
        for user in fresh:
            # ⚠ THE SHAPE IS `routes_alerts.notification_view`'s SHARE BRANCH, and the two must
            # agree or the Inbox row is unclickable: `topic` selects the branch and `key` becomes
            # `alertId`, which that branch reads as the id to open. Both constants are IMPORTED
            # from there rather than typed again — one vocabulary, one owner.
            # ⭐⭐ W33-T28 (`ASK C-14`, answered) — `actor` IS THE SENDER, AND IT IS THE ONLY WAY
            # THE INBOX CAN NAME ONE. An alert and an automation have no person behind them and
            # are honestly named by their machine; a SHARE has a real person, and only this call
            # site knows who. ⛔ It is passed as its OWN field rather than recovered from the
            # `detail` prose below: a sender parsed out of "<name> shared this with you" breaks
            # the first time the sentence is reworded, silently, in the header
            # [[grep-output-is-not-source]]. The prose stays as the body; this is the From.
            alerts.notify(user, label, topic=_SHARE_TOPIC, key=route, row_id=view_id,
                          detail=f"{sharer} shared this with you", actor=sharer,
                          st=session.runtime)
    except Exception:                                       # noqa: BLE001
        return


def _object_ref(session, kind, oid):
    """`(label, route, view_id)` — what to CALL the shared thing, and where it OPENS.

    ⛔ THE ROUTE IS RESOLVED HERE, NOT SHAPED IN THE CONSUMER, AND THE FIRST VERSION GOT IT
    WRONG: it put the raw `oid` in the notification's key, so a shared VIEW produced
    `target: {module: "database", id: "view_42"}` — an instruction to open a database named
    `view_42`. It read perfectly in the payload and would have opened nothing. **A view is not
    addressable on its own; it is a SELECTION inside a topic's grid**, so the pair is what has to
    travel. Caught by looking at the notification the driver actually produced, not by reading
    the code back.

    ⚠ `label` never falls back to a raw id. A notification headed `ut_leads_3f2a` tells the
    receiver nothing they can act on, and the id is already in the target.
    ⚠ An unresolvable object answers `route=None`, and the caller then sends NOTHING rather than
    a row that opens nowhere.
    """
    try:
        if kind == "cohort":
            scope, cohort_id = shares.split_cohort_oid(oid)
            source = shares.cohort_source(scope, cohort_id, st=session.runtime) if scope else None
            if source is None:
                return ("A cohort", None, "")
            from routes_alerts import route_for_topic
            label = str(source["cohort"].get("name") or "").strip() or "A cohort"
            return (label, route_for_topic(source["scope"]) or None, source["id"])
        if kind == "field":
            # ⭐⭐ W38-T16 — A COLUMN IS NOT ADDRESSABLE ON ITS OWN, exactly as a view is not: it
            # is a column INSIDE a database, so the target that travels is the DATABASE. Without
            # this branch the function falls through to the view/folder loop, finds nothing,
            # answers `route=None` — and `_notify_new_grantees` returns EARLY. The grant lands and
            # the receiver is never told, which is the silent half of owner item 18 reopened one
            # kind over.
            from routes_alerts import route_for_topic
            table_key, field_key = shares.split_field_oid(oid)
            if not table_key:
                return ("A column", None, "")
            try:
                defn, _shared, _workspace, shared_key, _grant_topic = _field_definition(
                    session, table_key, field_key)
            except Exception:                               # noqa: BLE001
                defn = None
            label = str((defn or {}).get("label") or "").strip() or field_key
            # ⚠ TWO SPELLINGS REACH THIS LINE AND ONE MAP ANSWERS BOTH. `shared_overlay` is keyed
            # by whatever the calling door already held: a `ut_*` database uses its bare key,
            # while a registry topic uses `<topic>_table_workspace` (`product_data.TABLE_KEY`).
            # `route_for_topic` speaks the GRID SCOPE vocabulary (`customer`, not
            # `customer_data`), so the suffix comes off before it is asked — rather than a second
            # route table being written here, which is how the two come apart.
            _WS = "_table_workspace"
            scope = {"customer_data": "customer", "product_data": "product"}.get(table_key)
            if scope is None:
                scope = table_key[:-len(_WS)] if table_key.endswith(_WS) else table_key
            return (label, route_for_topic(scope) or None, "")
        if kind == "database":
            import core.user_tables as ut
            defn = (ut.all_defs(st=session.runtime) or {}).get(str(oid)) or {}
            # A user table IS its own route key in both vocabularies (`route_for_topic`).
            return (str(defn.get("label") or "").strip() or "A database", str(oid), "")
        import core.table_store as table_store
        from routes_alerts import route_for_topic
        for topic in _topics(session):
            ops = table_store.make(f"{topic}_table_workspace", st=session.runtime)
            hit = ops.find_view(oid) if kind == "view" else ops.find_folder(oid)
            if not hit:
                continue
            route = route_for_topic(topic)
            if not route:
                break
            row = hit[1] if len(hit) > 1 else {}
            name = str((row or {}).get("name") or "").strip()
            # ⚠ Only a VIEW carries a selection. A folder is a rail grouping, so the target opens
            # the grid and stops there rather than naming a view the receiver did not get.
            return (name or ("A view" if kind == "view" else "A folder"),
                    route, str(oid) if kind == "view" else "")
    except Exception:                                       # noqa: BLE001
        pass
    return ({"view": "A view", "folder": "A folder",
             "field": "A column", "cohort": "A cohort"}.get(kind, "An item"), None, "")


def _cohort_visible(scope, username, allowed_pids=None, st=None):
    """Adapt the Cohort module's existing list projection to the one registry predicate.

    ``workspace_wire`` is the live Cohort list/open path.  It intentionally receives only a
    username and a pid envelope, so the adapter remains at this API boundary and reuses the
    module's membership cleaner rather than creating a second Cohort representation in
    ``core.shares``.  ``visible_cohorts`` returns no unauthorized id, which prevents the blank
    locked-view placeholder the old per-user read created.
    """
    import modules.cohort as cohort_mod

    records = shares.visible_cohorts(scope, username, st=st)
    return {cid: {**record,
                  "members": cohort_mod._clean_members(record.get("members"), allowed_pids)}
            for cid, record in records.items()}


def _install_cohort_visibility():
    """Route all API Cohort list/detail reads through the normal share predicate once."""
    try:
        import modules.cohort as cohort_mod

        def _customer_visible(username, allowed_pids=None):
            return _cohort_visible("customer", username, allowed_pids)

        def _scoped_visible(self, username, allowed_pids=None):
            return _cohort_visible(self.scope, username, allowed_pids,
                                   st=getattr(self, 'st', None))

        cohort_mod.visible = _customer_visible
        cohort_mod.CohortStore.visible = _scoped_visible
    except Exception:
        # The API route still has its direct, fail-closed list/detail paths if the retired module
        # is unavailable during an import-only tool invocation.
        return


_install_cohort_visibility()