api/routes_alerts.py

"""routes_alerts.py — the Alerts module (wave 20, owner item 25, contract C-ALERT).

    GET    /api/v1/alerts                  -> {alerts:[...]}
    POST   /api/v1/alerts                  <- {viewId, topic, label?}
    DELETE /api/v1/alerts/{alert_id}
    POST   /api/v1/alerts/{alert_id}/run   -> evaluate now (the pane's manual refresh)
    GET    /api/v1/notifications           -> {unread, items:[...]}
    POST   /api/v1/notifications/read      <- {ids:[...]|null, read?:bool}

The semantics — an alert is a view plus a remembered matched set, a notification is a NEW
ENTRANT, and the first evaluation seeds silently — live in `core.alerts` with the reasoning.
This file owns the two things a route must: WHO may do it, and HOW the view gets evaluated.

⭐ **THE EVALUATION RUNS AS THE ALERT'S OWNER, NOT AS THE CALLER.** `_run_alert` builds the pool
for `rec['owner']`, never for whoever tripped the write hook. Any other choice leaks: a
full-access admin editing a cell would otherwise evaluate a BU-scoped user's alert over the whole
book, and the notification would name customers that user may not see — a permission leak wearing
a notification's clothes. The owner's own scope is the only correct basis for their alert.

⚠ **AN ALERT IS NOT A SECOND READ PATH.** It resolves rows through the same
`routes_customers.grid_assembly` / `routes_tables.ut_assembly` the grid uses, so a row that an
alert can see is by construction a row its owner could open. Re-implementing the filter here
would be a second definition of "matches", and those two would drift.
"""
import re

from fastapi import APIRouter, Body, Depends

import core.alerts as alerts
from deps import Session, err, require_session

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

#: The alert-bearing surfaces. `ut_` tables are admitted by prefix, like everywhere else.
_TOPICS = ("customer", "product")

# ── ⭐⭐ WAVE 32 · T20 · CONTRACT C3 — THE INBOX SHAPE, DERIVED ON READ ────────────────────────
#
# `GET /notifications` gains `subject`, `kind` and `target` per item (`read` was always there).
#
# ⛔ DERIVED, NEVER STORED, AND THAT IS THE WHOLE OF WHY THIS WAVE EXISTS. Stamping the three
# keys onto the record at write time would give them to notifications minted AFTER the deploy and
# to nothing else — every notification already sitting in every tenant's inbox would open nothing,
# and the feature would be correct in the source and absent from the product
# ([[a-migration-that-runs-on-the-next-write]], D-201). A read-side derivation reaches a
# notification queued last month. It also keeps the store shape out of `core/alerts.py`, which is
# another lane's file this wave — but that is the convenience, not the reason.
#
# ⚠ TWO PRODUCERS WRITE TWO SHAPES into one inbox, and the vocabulary below is what tells them
# apart. `_queue` (a record ENTERED a watched view) sets `topic`+`viewId`. `notify()` sets
# `topic='automation'` and puts the producer's key in `alertId`, leaving `viewId` empty. Deciding
# here means the client branches on ONE field instead of re-deriving the same split.
#
# ⛔ **D-101 IS CLOSED HERE, BY SUBTRACTION.** There was a THIRD shape — `kind='automation_review'`
# + `autoId`, a card arriving at a review stage — and its producer `notify_review` was deleted by
# W27/R3 with the review lanes. `automation_engine.py`'s own tombstone (search `notify_review`)
# records the 2026-08-12 sweep: **no `.py` file anywhere produces one**, while the client branch,
# its route and three gate legs stayed fully alive. D-101's exit condition is *"the client review
# branch is deleted in the same change as any remaining residue, OR `notify_review` gains its real
# caller"* — the residue is zero, so the branch goes. It is not carried into the Inbox: a stored
# review notification (if any survives in a tenant from the wave-23 era) derives as an ordinary
# `alert` with no target, i.e. an honest unclickable row, which is correct — the board it pointed
# at was deleted two waves ago.

#: C3's `kind` vocabulary. Plain strings on the wire — the client must never union over them
#: (alertsModel's wave-9 law: a client union turns "the server grew a kind" into a dropped row).
NOTIF_KIND_ALERT = "alert"
NOTIF_KIND_AUTOMATION = "automation"
NOTIF_KIND_SHARE = "share"

#: C3's `target.module` vocabulary, and the automation sub-selection.
TARGET_MODULE_DATABASE = "database"
TARGET_MODULE_AUTOMATION = "automation"
TARGET_TAB_RUNS = "runs"

#: The topic `notify()` carries for a SHARE (W32-T28 writes it; nothing does yet, and a kind with
#: no producer is a string that reads as a feature — the reason this constant is named here and
#: cited from `routes_shares` rather than typed twice).
SHARE_TOPIC = "share"

#: `core.alerts.notify`'s default topic for a run outcome. Mirrors `inboxModel.AUTOMATION_TOPIC`.
AUTOMATION_TOPIC = "automation"

_UT_TOPIC = re.compile(r"ut_[A-Za-z0-9_]+\Z")


def route_for_topic(topic):
    """A grid SCOPE key -> the registry route that renders it, or None.

    ⛔ THE SAME TABLE AS `alertsModel.routeForTopic`, and the parity is GATED
    (`verify_alerts.py`'s vocabulary scan) rather than trusted. The two built-ins are the only
    pair that differ — the registry names the surface (`customer_data`) while the grid names the
    scope (`customer`) — so a topic passed through as a route sends every click to a page that
    does not exist. `None` for anything else: a target this product cannot resolve must be ABSENT
    rather than plausible, because an absent target renders as a row that does not pretend to be
    clickable, and a wrong one renders as a click that silently goes nowhere.
    """
    t = str(topic or "").strip()
    if t == "customer":
        return "customer_data"
    if t == "product":
        return "product_data"
    if _UT_TOPIC.match(t):
        return t
    return None


def _refusal_code(exc):
    """The `error.code` an `HTTPException` raised by `deps.err()` carries, or `""`.

    ⭐ W32-T22. Four refusals travel up the assembly chain — `unknown_table` (404), `forbidden`
    (403), `window_required` (409) and `store_not_ready` (503) — and each already names its own
    cause. Anything that reduces all four to one word is throwing away the only information the
    reader could have acted on. Returns `""` for a plain exception, so a caller can tell
    "refused, and here is why" apart from "broke, and we do not know why".
    """
    detail = getattr(exc, "detail", None)
    if isinstance(detail, dict):
        inner = detail.get("error")
        if isinstance(inner, dict):
            return str(inner.get("code") or "")
    return ""


def notification_view(item):
    """One STORED notification -> the shape the Inbox renders. PURE, and total.

    Never raises and never drops a row: an item it cannot classify comes back as an `alert` with
    no `target`, which the client renders as an unclickable row rather than hiding. An inbox that
    silently omits what it does not understand is the one failure a reader cannot detect.
    """
    if not isinstance(item, dict):
        return item
    topic = str(item.get("topic") or "").strip()
    alert_id = str(item.get("alertId") or "").strip()

    # ⛔ THE ID TEST IS HALF OF EVERY BRANCH, and it is the load-bearing half. A row whose topic
    # says `automation` but whose producer key never arrived (a truncated payload, a server
    # mid-deploy) would otherwise be handed a target naming NOTHING — a click that appears to work
    # and silently does not, which is this repo's most-repeated failure shape. Failing the test
    # drops it to the `alert` branch, where `route_for_topic` refuses out loud by answering None.
    if topic == AUTOMATION_TOPIC and alert_id:
        kind = NOTIF_KIND_AUTOMATION
        target = {"module": TARGET_MODULE_AUTOMATION, "id": alert_id, "tab": TARGET_TAB_RUNS}
    elif topic == SHARE_TOPIC and alert_id:
        # ⭐ W32-T28: the sharer writes `key=<the ROUTE to open>` and, for a shared VIEW,
        # `row_id=<the view to select>`.
        #
        # ⛔ `key` IS ALREADY A ROUTE, NOT A RAW OBJECT ID, and the first version of this got it
        # wrong in a way worth recording: a shared VIEW put the VIEW's id in `alertId`, so the
        # target read `{module: "database", id: "view_42"}` — an instruction to open a database
        # called `view_42`. It looked right in the payload and would have opened nothing. The
        # producer resolves the object to its topic and hands over the route; this branch only
        # shapes what it is given.
        kind = NOTIF_KIND_SHARE
        row_id = str(item.get("rowId") or "").strip()
        target = {"module": TARGET_MODULE_DATABASE, "id": alert_id,
                  **({"tab": row_id} if row_id else {})}
    else:
        kind = NOTIF_KIND_ALERT
        route = route_for_topic(topic)
        view_id = str(item.get("viewId") or "").strip()
        target = None if route is None else (
            {"module": TARGET_MODULE_DATABASE, "id": route,
             **({"tab": view_id} if view_id else {})})

    # The email split: `subject` is the HEADER (what this is about — the alert, the automation,
    # the database), `label` stays the BODY (what happened — the record that entered, the run
    # summary). They were one field, which is why a notification read as a sentence with no
    # sender and the pane could not be laid out like mail.
    subject = str(item.get("alertLabel") or "").strip() or str(item.get("label") or "").strip()
    # ⚠ `kind` is OVERWRITTEN, not merged. There was one stored value (`automation_review`) and it
    # is D-101's dead one; leaving it through would give the client two vocabularies for one
    # question, which is the defect this wave's item 6 is about in a different file.
    out = {**item, "read": bool(item.get("read")), "kind": kind,
           "subject": subject or "Notification"}
    if target is not None:
        out["target"] = target
    return out


def inbox_view(box):
    """`core.alerts.inbox()`'s answer, with every item put through {@link notification_view}.

    ⚠ `unread` IS NOT RECOUNTED. It is the ACCOUNT's number and `items` is one page of it; a
    recount here would make the badge a function of whatever this page happened to include, which
    is the exact defect `alertsModel.parseInbox`'s own header records from the other side.
    """
    if not isinstance(box, dict):
        return box
    items = box.get("items")
    if not isinstance(items, list):
        return box
    return {**box, "items": [notification_view(n) for n in items]}


def _topic_or_400(raw):
    topic = str(raw or "").strip().lower()
    if topic.startswith("ut_") or topic in _TOPICS:
        return topic
    raise err(400, "bad_topic", f"topic must be one of {', '.join(_TOPICS)} or a ut_ table")


def _owner_session(session: Session, owner: str):
    """A `Session` for the alert's OWNER (see the module note on why the owner, not the caller).

    ⚠ `Session` exposes `uname`/`admin` as PROPERTIES derived from `user`, not as fields — so an
    owner session is built by swapping the `user` RECORD and letting both derive themselves. An
    earlier version passed `uname=`/`admin=` to the constructor, which would have raised on the
    first write hook of the wave; the properties are the single definition of who a session is,
    and going around them is how a session with an admin flag and a non-admin record exists.

    Returns None when the owner is gone or deactivated — their alerts then stop evaluating rather
    than evaluating as somebody else, which is the fail-closed direction.
    """
    import core.users as users

    if str(owner) == str(session.uname):
        return session
    rec = (users.registry() or {}).get(str(owner))
    if not isinstance(rec, dict) or not rec.get("active", True):
        return None
    # `_public` is THE definition of what a session may know about its own account (never a hash
    # or a salt) — the same one `routes_auth` uses. Building the dict by hand here would be a
    # second definition, and the one that leaks is always the copy.
    return Session(tenant=session.tenant, user=users._public(str(owner), rec),
                   claims=session.claims, runtime=session.runtime)


def _evaluate(session: Session, rec: dict, assemblies=None):
    """Resolve `rec`'s view over its topic AS THE ALERT'S OWNER, then fold the result in.

    ⭐⭐ W31-T24 — `assemblies` IS A PER-REQUEST MEMO, KEYED `(topic, owner)`, and it is the whole
    of this ticket's server half. `/notifications` re-evaluates EVERY alert inline on read and each
    one built a FULL assembly — the pool, the workspace, `rows_from_pool` over every row. Two
    alerts on one view built that table twice; ten built it ten times. Nothing dedupes them,
    because each `_evaluate` was a closed call.
    ⚠ `(topic, owner)` and not `topic`: the assembly is built as the alert's OWNER (see the module
    note — evaluating a BU-scoped user's alert on a full-access admin's pool is a permission leak
    wearing a notification's clothes), so two owners on one topic are two DIFFERENT tables and
    must never share an entry. Getting that key wrong is the one way this optimisation could leak.
    ⚠ Passing nothing keeps the old behaviour exactly, which is what the create/run doors want:
    they evaluate ONE alert and a memo for a single call is pure overhead.
    """
    import aios_grid
    from harness import filter_eval

    owner_sess = _owner_session(session, rec.get("owner"))
    if owner_sess is None:
        return {"skipped": "owner_unavailable"}
    topic = str(rec.get("topic") or "")
    memo_key = (topic, str(owner_sess.uname))
    g = assemblies.get(memo_key) if isinstance(assemblies, dict) else None
    if g is None:
        try:
            if topic.startswith("ut_"):
                from routes_tables import ut_assembly
                # ⛔ `consume_corrections=False`, and the default was a REAL BUG, not a tidy-up.
                # `ut_assembly` defaults it True, so every `/notifications` read CONSUMED the
                # one-shot field-name correction acks for every `ut_` topic that has an alert —
                # taking them from the `/workspace` refresh that exists to show them to the person
                # who made the edit. The customer branch below has always passed False; this one
                # inherited a default nobody re-read. An inbox poll must never consume a one-shot.
                g = ut_assembly(owner_sess, topic,
                                storage_key=f"{owner_sess.tenant}:{topic}:{owner_sess.uname}",
                                consume_corrections=False)
            else:
                from routes_customers import grid_assembly
                g = grid_assembly(owner_sess, scope=topic, consume_corrections=False)
        except Exception as e:                              # noqa: BLE001
            # ⭐ W32-T22 — SKIPPING IS FINE HERE; SKIPPING ANONYMOUSLY IS NOT. This one must not
            # raise (one bad alert cannot empty an inbox), so unlike `_require_filtered_view` it
            # keeps a blanket catch — but it now reports the refusal's OWN code where there is
            # one. `type(e).__name__` said `HTTPException` for four different causes, and
            # `lastError` is the only place a user ever learns why an alert stopped firing.
            #
            # ⚠ `with_rows=True` STAYS on this path, deliberately: unlike the create door, an
            # evaluation genuinely needs the rows to run the filter over. So an alert on a
            # read-through grid is created (T22) and then skips at evaluation with
            # `window_required` naming why — which is D-184's remaining half, and it is a
            # SENTENCE now rather than silence.
            return {"skipped": _refusal_code(e) or "unavailable", "detail": type(e).__name__}
        if isinstance(assemblies, dict):
            assemblies[memo_key] = g

    view = (g.get("views") or {}).get(str(rec.get("viewId")))
    if not isinstance(view, dict):
        # Deleted, or un-shared out from under the alert. Say so on the RECORD rather than
        # deleting the alert: an alert that silently vanishes is indistinguishable from one that
        # never fires, and the user cannot debug what is not there.
        return {"skipped": "view_missing"}

    # The SAME row build the grid and `/customers` use — `rows_from_pool` is what puts derived
    # and overlay values on a row. Evaluating a filter against raw pool dicts would silently
    # never match any condition on a user-created or measure column.
    rows = aios_grid.rows_from_pool(g["rows_src"], g["fields"], g["ws"].get("overlays"),
                                    derived=g.get("derived"))
    config = view.get("config") or view
    ctx = filter_eval.EvalCtx(
        cohort_sets={str(k): {str(p) for p in (v.get("memberPids") or ())}
                     for k, v in (g.get("lists") or {}).items() if isinstance(v, dict)},
        measure_sets=g.get("measure_sets") or {},
        today=g.get("today"))
    pids = filter_eval.visible_pids(config.get("filters") or [], rows, g["fields"], ctx,
                                    member_pids=config.get("memberPids"))
    labels = {str(r.get("pid")): str(r.get("name") or r.get("pid")) for r in rows}
    return alerts.evaluate(rec.get("id"), [str(p) for p in pids],
                           labels=labels, partial=False, st=session.runtime)


@router.get("/alerts")
def list_alerts(session: Session = Depends(require_session)):
    return {"alerts": alerts.list_alerts(user=session.uname, is_admin=session.admin,
                                         st=session.runtime)}


@router.post("/alerts")
def create_alert(body: dict = Body(default=None), session: Session = Depends(require_session)):
    body = body or {}
    view_id = str(body.get("viewId") or "").strip()
    if not view_id:
        raise err(400, "bad_view", "an alert needs the id of the view it watches")
    topic = _topic_or_400(body.get("topic"))
    _require_filtered_view(session, topic, view_id)
    import uuid
    aid = f"al_{uuid.uuid4().hex[:12]}"
    rec = alerts.create(aid, view_id=view_id, topic=topic, owner=session.uname,
                        label=body.get("label") or "", st=session.runtime)
    # SEED IMMEDIATELY, so the alert starts from "everything currently matching is old news".
    # Deferring this to the first write hook would mean the next edit announces the whole view.
    outcome = _evaluate(session, rec)
    return {"alert": {**rec, "seeded": True}, "first": outcome}


def _require_filtered_view(session: Session, topic: str, view_id: str):
    """400 unless `view_id` exists on `topic` AND actually narrows something.

    ⛔ AN ALERT ON AN UNFILTERED VIEW IS SILENTLY INCAPABLE OF ALERTING, which is worse than one
    that is refused. `filter_eval` treats an inactive tree as "no narrowing, every row shows"
    (`visible_pids`'s own rule), so such an alert seeds with the entire table and can never see an
    entrant again — there is nothing left to enter. The owner's words are *"when a Record gets
    into that Filter's criteria"*: no criteria, no alert, and said at creation rather than
    discovered by never being notified.

    `is_rule_active` is the SAME activeness predicate the engine and the column tints use — a
    half-typed rule is not a filter, and this must agree with what actually narrows or it would
    accept a view whose one rule the engine then ignores.

    ⭐⭐ WAVE 32 · T22 (owner item 17) — THIS FUNCTION WAS THE ERROR. Two defects, stacked, and
    the second one hid the first.

    (1) **IT ASKED FOR EVERY ROW OF A TABLE IT NEVER LOOKS AT.** The only thing read below is
        `g["views"]`. `ut_assembly` defaults `with_rows=True`, so creating an alert on a
        read-through grid built the whole pool — and `scoped_pool` refuses that with
        `409 window_required` over 963,783 rows, exactly as it is supposed to. `with_rows=False`
        (W31-T20's flag, built for precisely this) answers the same question with `scoped_pids`,
        runs the SAME `_defn_or_refuse` wall, and does not refuse. **That is D-184's create half,
        closed** — an alert on a read-through grid can now be made at all.
    (2) **A BLANKET `except Exception` TURNED EVERY NAMED REFUSAL INTO A 503.** `HTTPException`
        is an `Exception`, so `404 unknown_table`, `403 forbidden`, `409 window_required` and
        `503 store_not_ready` — four refusals that each say what is wrong — were all replaced by
        *"the table is unavailable — try again in a moment"*. ⛔ AND THAT SENTENCE NEVER REACHED
        A USER EITHER: `alertsApi.errorMessage` discards the text of any status ≥ 500 by design
        (a 5xx body is the server's internals), substituting *"Something went wrong on our
        side."* — which is the owner's screenshot, word for word. A knowable cause returned as a
        5xx is invisible by construction, so re-wording the 503 could never have fixed this.
    ⚠ The except is narrowed, not deleted: an UNEXPECTED failure is still a 503, because that is
    honest. What it may no longer do is catch a refusal that already knows its own name.
    """
    from fastapi import HTTPException

    from harness import filter_eval

    try:
        if topic.startswith("ut_"):
            from routes_tables import ut_assembly
            # ⚠ `consume_corrections=False` — the customer branch has always passed it and this
            # one inherited a default nobody re-read. Creating an alert must not eat the one-shot
            # field-name correction acks belonging to the `/workspace` refresh that exists to show
            # them to the person who made the edit. Same defect `_evaluate`'s header records.
            g = ut_assembly(session, topic,
                            storage_key=f"{session.tenant}:{topic}:{session.uname}",
                            consume_corrections=False, with_rows=False)
        else:
            from routes_customers import grid_assembly
            g = grid_assembly(session, scope=topic, consume_corrections=False)
    except HTTPException:
        raise                                               # it already names its own cause
    except Exception as e:                                  # noqa: BLE001
        # Genuinely unexpected. Still a 503, and now it carries the exception TYPE — without it,
        # the one path that reaches this branch is also the one path with nothing to debug from.
        raise err(503, "unavailable",
                  f"the table could not be read ({type(e).__name__}) — try again in a moment")
    view = (g.get("views") or {}).get(str(view_id))
    if not isinstance(view, dict):
        raise err(404, "no_view", "that view does not exist on this table")
    nodes, _conj = filter_eval.tree_parts((view.get("config") or view).get("filters") or [])

    # ⛔⛔ WAVE 32 · T22 — **THE CALL BELOW WAS MISSING AN ARGUMENT, AND THAT IS OWNER ITEM 17.**
    #
    # `is_rule_active(rule, columns)` takes TWO parameters (`harness/filter_sql.py`; every other
    # caller in the repo passes both). This one passed ONE, so the moment the walk reached a LEAF
    # rule it raised `TypeError: is_rule_active() missing 1 required positional argument`.
    #
    # ⚠ READ WHAT THAT MEANS BEFORE FIXING ANYTHING ELSE: the walk only reaches a leaf when the
    # view HAS a condition — and a view with a condition is the only kind an alert is allowed on.
    # A view with no filters yields an empty `nodes`, so `_any_active` returns False without ever
    # calling this, and the reader gets the honest 400 `no_filter`. **So the only path that
    # worked was the refusal path: "Alert me about new records" had never once created an alert
    # on a filtered view.** ⛔ And the raise lands OUTSIDE the `try` above, so it was not even the
    # 503 — it was a bare FastAPI 500, which `alertsApi.errorMessage` renders as *"Something went
    # wrong on our side. Try again in a moment."*, the owner's screenshot word for word.
    #
    # ⚠ THREE THINGS HID IT, and they are worth more than the fix. (1) Python does not check
    # arity until the line RUNS, and this line runs only on the success path of a feature whose
    # every test exercised its refusals. (2) The `no_filter` 400 above it is a real, correct,
    # well-tested refusal, so the door looked alive. (3) `verify_alerts.py` asserts the refusal
    # (`no_filter` reaches the user) and the transport — never a creation. A gate can be green,
    # thorough and honest about everything except the one path the feature exists for.
    #
    # `_columns_map` is the DEFINITION of fields -> the membership set `is_rule_active` looks a
    # column up in; building a second dict here would be a second answer to one question, which
    # is this wave's other headline defect in a different file. Its leading underscore is a real
    # smell and is BOOKED (PENDING, mailbox/C.md) rather than worked around.
    columns = filter_eval._columns_map(g.get("fields") or [])

    def _any_active(ns):
        for n in ns or ():
            if isinstance(n, dict) and isinstance(n.get("children"), list):
                if _any_active(n["children"]):
                    return True
            elif filter_eval.is_rule_active(n, columns):
                return True
        return False

    if not _any_active(nodes):
        raise err(400, "no_filter",
                  "this view has no active filter, so no record can ever ENTER it — add a "
                  "condition to the view first, then create the alert")


@router.delete("/alerts/{alert_id}")
def delete_alert(alert_id: str, session: Session = Depends(require_session)):
    rec = next((r for r in alerts.list_alerts(st=session.runtime)
                if str(r.get("id")) == str(alert_id)), None)
    if rec is None:
        raise err(404, "no_alert", "that alert does not exist")
    if str(rec.get("owner")) != str(session.uname) and not session.admin:
        raise err(403, "not_yours", "only the alert's owner (or an administrator) can delete it")
    alerts.delete(alert_id, st=session.runtime)
    return {"ok": True}


@router.post("/alerts/{alert_id}/run")
def run_alert(alert_id: str, session: Session = Depends(require_session)):
    rec = next((r for r in alerts.list_alerts(user=session.uname, is_admin=session.admin,
                                              st=session.runtime)
                if str(r.get("id")) == str(alert_id)), None)
    if rec is None:
        raise err(404, "no_alert", "that alert does not exist")
    return _evaluate(session, rec)


@router.get("/notifications")
def notifications(session: Session = Depends(require_session)):
    """The inbox — RE-EVALUATED on read, which is a deliberate design choice.

    ⭐ A-S1-2 RESOLVED THE OTHER WAY, and the reason is structural rather than a shortcut. The
    plan was a push hook: the automation engine calls `after_write` when it lands rows. But
    `run_async` runs on a BACKGROUND THREAD with no `Session` in scope, and an alert must be
    evaluated as its OWNER (see `_evaluate`) — so a push hook would have to mint a session inside
    a worker thread from a tenant runtime, which is exactly the kind of ad-hoc identity
    construction that leaks scope.

    Pulling on read has none of that: the caller IS a session, the assemblies are already
    scope-cached, and the user cannot observe the difference — an inbox is only ever read by
    someone opening it. The cost is that a notification is minted when you LOOK rather than when
    the row landed, so the `at` stamp is detection time, not arrival time.

    `after_write` stays exported for the day the engine can hand over a real identity.

    ⭐⭐ W31-T24 — ONE ASSEMBLY PER (TOPIC, OWNER), NOT ONE PER ALERT.
    ⛔ MEASURED FIRST, AND THE MEASUREMENT CORRECTS AN EARLIER READING OF IT. This route is
    **20 ms in-process and 3,280 ms live** on tenant #0 — but tenant #0 has **ZERO alerts**
    (censused 2026-08-12), so the 20 ms is an EMPTY LOOP and says nothing at all about what the
    re-evaluation costs. The live 3,280 ms is the two store reads either side of that loop. So the
    body below is not slow today; it is UNEXERCISED, and every alert a tenant creates adds a whole
    grid assembly to an inbox poll. The memo turns O(alerts) into O(distinct topic × owner), which
    is the difference between "fine" and "three seconds per alert" the day somebody uses the
    feature. ⚠ Making the read cheap by evaluating LESS is the obvious wrong fix and is not what
    this does: every alert is still evaluated, against the same rows, in the same order.
    """
    assemblies = {}
    for rec in alerts.list_alerts(user=session.uname, is_admin=False, st=session.runtime):
        try:
            _evaluate(session, rec, assemblies=assemblies)
        except Exception:                                   # noqa: BLE001
            continue                                        # one bad alert must not empty the pane
    # ⭐ W32-T20 (C3): every item leaves through `inbox_view`, so a notification queued before
    # this wave carries a `target` too. See `notification_view`'s header for why it is derived.
    return inbox_view(alerts.inbox(session.uname, st=session.runtime))


@router.post("/notifications/read")
def read_notifications(body: dict = Body(default=None),
                       session: Session = Depends(require_session)):
    body = body or {}
    ids = body.get("ids")
    if ids is not None and not isinstance(ids, list):
        raise err(400, "bad_ids", "ids must be a list, or null to mark every notification")
    # ⚠ THE SAME ENRICHMENT ON BOTH DOORS. `mark_read` returns a fresh inbox, and the Inbox
    # module re-renders from it — an un-enriched answer here would strip `target` off every row
    # the moment somebody marked one read, i.e. the feature would work until first use.
    return inbox_view(alerts.mark_read(session.uname, ids, read=bool(body.get("read", True)),
                                       st=session.runtime))


def after_write(session: Session, topic_key: str):
    """THE WRITE HOOK — call after a write that could change what a view matches.

    Exported as a plain function (not a route) so `core.grid_events`' callers and S2's automation
    upserts reach it the same way. It never raises: an alert evaluation failing must not fail the
    edit that triggered it.

    ⭐ W31-T24 — it shares `/notifications`' memo shape for the same reason: a write that changes
    one view can trip several alerts on the SAME topic, and each would otherwise rebuild the table.
    ⚠ STILL ZERO PRODUCTION CALLERS (W31-T24 confirmed it; the route docstring above says why the
    push hook was resolved the other way). Booked rather than wired: minting a session inside the
    engine's worker thread is the ad-hoc identity construction this file exists to avoid.
    """
    try:
        assemblies = {}
        return alerts.after_write(topic_key, st=session.runtime,
                                  runner=lambda rec: _evaluate(session, rec,
                                                               assemblies=assemblies))
    except Exception:                                       # noqa: BLE001
        return {"evaluated": 0}