"""routes_admin.py — Y4: user administration + the session's own settings (W2-3).

These routes are the standalone shell's replacement for `app.py`'s `users_dialog` / `settings_dialog`
modals, over the SAME `core/users` account store — so both front-ends administer one set of
accounts and the eventual OIDC migration (D-3) swaps the CREDENTIAL check, not the user model.

ADMIN-ONLY, FAIL-CLOSED. `admin_gate` is a role check (`perms.is_admin`), mirroring the Streamlit
dialog's `if not is_admin()`. A default record has `role: 'user'`, so an unreadable or partial record
is denied rather than admitted. `verify_api.py` proves it by having a viewer TRY every route.

⛔ THIS IS THE FIRST WRITER OF `modules` THAT HAS EVER EXISTED. `core.users.set_access(modules=…)`
has no caller anywhere in the shipped app — module grants are only settable by editing the store
JSON out of band. That matters because the record format carries TWO documented fail-OPENS, both
asserted in `verify_api.py` section A:

  * `modules: []` is FALSY, so `perms.allowed_modules` reads it as 'all' = UNRESTRICTED. An admin
    clearing every checkbox to lock an account down would grant it everything.
  * `bus: [<unknown id>]` falls through `allowed_bus_labels`'s "no recognisable label" branch to
    `['All','Fisch','Royal']` — so one typo'd BU id is FULL cross-BU access, in the model whose
    whole point is strict isolation.

Y4 says do not "fix" `modules: []` in this wave, and that is right: the READER is mirrored in
`ui/session.py` and `core/perms.py` and diverging one of them mid-wave breaks lock-step. But the
WRITER is new, and it can simply refuse to create either footgun. So both are 400s here and both
read semantics are untouched — the seam is write-strict / read-unchanged.

⚠ A STORE OUTAGE IS A 503, NEVER AN EMPTY LIST. `users.registry()` swallows a failed read into `{}`
one level down, and serving that as `{"users": []}` would tell an administrator their tenant has no
accounts. Same rule as the write path: an empty 200 is never how this API says "something is wrong".
"""
import os
import re

from fastapi import APIRouter, Body, Depends, Response

import core.platform_admin as platform_admin   # wave 19 R3 — the /settings chrome flag
import core.registry as registry
import core.store as store

from deps import Session, err, perms, require_session, users
from routes_auth import _public_user

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

#: A username is a STORE KEY (`users.json`) and also the key the table workspace is filed under
#: (`data[username]`), so it is constrained rather than trusted: lowercase, no separators, no
#: whitespace, nothing that could traverse or collide once it becomes part of a path or a filename.
_UNAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{1,31}$")
#: Passwords are PBKDF2-200k, which is only as strong as what it is given. The Streamlit dialog
#: enforces nothing; this does, and the two are allowed to differ because only one of them is
#: reachable from the internet.
_MIN_PW = 8
_ROLES = ("user", "admin")


def admin_gate(session: Session = Depends(require_session)) -> Session:
    """401 without a session, 403 without the admin role. Role, not a module grant — mirroring
    `app.py`'s `users_dialog`, whose only wall is `is_admin()`."""
    if not perms.is_admin(session.user):
        raise err(403, "forbidden", "administrators only")
    return session


def _require_store():
    """A 503 the moment the store cannot serve, so no route below can report emptiness as truth."""
    try:
        ok = store.available()
    except Exception:
        ok = False
    if not ok:
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — accounts cannot be read or changed")


def _registry():
    _require_store()
    try:
        reg = users.registry() or {}
    except Exception:
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    return reg


def _view(uname, rec):
    """What an administrator may see about an account. NEVER `salt` or `hash` — this function is
    the only projection these routes use, so there is one place that can leak and it does not."""
    return {"username": uname,
            "name": rec.get("name") or uname,
            "role": rec.get("role", "user"),
            "bus": rec.get("bus", "all"),
            "bu_labels": perms.allowed_bu_labels(rec),
            "modules": rec.get("modules", "all"),
            "agent": rec.get("agent") or None,
            "email": rec.get("email") or None,
            "tenant": _tenant_of(rec),
            "active": bool(rec.get("active", True)),
            # The session-revocation handle. An admin needs it: it is the only visible evidence
            # that "sign them out everywhere" actually happened.
            "epoch": int(rec.get("epoch") or 0),
            # S3 ask 4 — one sentence for the roster's Access column, so the list does not need
            # a /perms round trip per row to fill one cell. Computed from the same record the
            # editor will open, so the two cannot disagree.
            "accessSummary": _access_summary(rec),
            "perms_v": int(rec.get("perms_v") or 0)}


def _access_summary(rec):
    """One sentence describing what this account may reach. Deliberately says LESS than the
    editor: it is a signpost, not a rule listing, and a summary that tried to spell out filters
    would be wrong the moment a filter got interesting."""
    import core.perm_scope as perm_scope

    if perms.is_admin(rec):
        # amendment 4: an admin bypasses `perms` entirely, so rendering their stored rules as if
        # they applied is the one misreading of that clause that could actually hurt.
        return "Everything (admin)"
    if not perm_scope.is_migrated(rec):
        mods = perms.allowed_modules(rec)
        return "All modules (legacy rules)" if mods is None else \
            f"{len(mods)} module{'' if len(mods) == 1 else 's'} (legacy rules)"
    entries = (rec.get("perms") or {})
    opened = [k for k, e in entries.items() if isinstance(e, dict) and e.get("access", True)]
    restricted = sum(1 for k in opened
                     if (entries[k].get("filter") or entries[k].get("hiddenFields")))
    if not opened:
        return "No modules"
    base = f"{len(opened)} module{'' if len(opened) == 1 else 's'}"
    return f"{base}, {restricted} restricted" if restricted else base


# ── request validation: every rule below is fail-closed ──────────────────────────────────────────
def _clean_username(v):
    uname = str(v or "").strip().lower()
    if not _UNAME_RE.match(uname):
        raise err(400, "bad_username",
                  "a username is 2-32 characters: lowercase letters, digits, dot, dash or "
                  "underscore, starting with a letter or digit")
    return uname


def _clean_password(v):
    pw = str(v or "")
    if len(pw) < _MIN_PW:
        raise err(400, "weak_password", f"a password must be at least {_MIN_PW} characters")
    return pw


def _clean_role(v):
    role = str(v or "").strip().lower()
    if role not in _ROLES:
        raise err(400, "bad_role", f"role must be one of {list(_ROLES)}")
    return role


def _clean_bus(v):
    """'all', or a non-empty list of KNOWN business-unit ids.

    Refusing an unknown id is the whole point: `allowed_bus_labels` treats a list with no
    recognisable label as full access, so `bus: [7]` would be a cross-BU grant created by a typo.
    """
    if isinstance(v, str):
        if v.strip().lower() == "all":
            return "all"
        raise err(400, "bad_bus", "bus must be 'all' or a list of business-unit ids")
    if not isinstance(v, (list, tuple)) or not v:
        raise err(400, "bad_bus",
                  "bus must be 'all' or a NON-EMPTY list of business-unit ids (an empty list "
                  "would read as unrestricted)")
    out = []
    for b in v:
        try:
            b = int(b)
        except (TypeError, ValueError):
            raise err(400, "bad_bus", "a business-unit id must be a number")
        if b not in users.BU_LABELS:
            raise err(400, "bad_bus",
                      f"{b} is not a known business unit {sorted(users.BU_LABELS)} — an "
                      "unrecognised id would widen access to every BU")
        out.append(b)
    return sorted(set(out))


def _clean_modules(v):
    """'all', or a non-empty list of KNOWN module keys.

    `[]` is refused because it READS as unrestricted (see the module docstring). An unknown key is
    refused because `may_open` is fail-closed on it: an admin who typed `sale` for `sales` would
    silently lock the account out of the page they meant to grant.
    """
    if isinstance(v, str):
        if v.strip().lower() == "all":
            return "all"
        raise err(400, "bad_modules", "modules must be 'all' or a list of module keys")
    if not isinstance(v, (list, tuple)) or not v:
        raise err(400, "bad_modules",
                  "modules must be 'all' or a NON-EMPTY list of module keys — an empty list "
                  "reads as UNRESTRICTED, which is the opposite of locking an account down")
    known = set(registry.BY_KEY) | set(perms._LEGACY_KEYS)
    out, bad = [], []
    for k in v:
        k = str(k or "").strip()
        (out if k in known else bad).append(k)
    if bad:
        raise err(400, "bad_modules", f"unknown module keys: {sorted(bad)}")
    return sorted(set(out))


def _tenant_of(rec):
    """The account's company, with the pre-wave default. ONE spelling of the default, used by
    every admin route — two spellings is how a tenant wall grows a gap."""
    return str((rec or {}).get("tenant") or "royal-imports").strip().lower()


# ── the routes ───────────────────────────────────────────────────────────────────────────────────
@router.get("/admin/users")
def list_users(session: Session = Depends(admin_gate)):
    """Wave 18 (C1-TENANT): scoped to the CALLER'S tenant. The user registry is a global
    control-plane bucket, so without this filter a Nurilab admin would read Royal's whole
    roster — names, emails, agents — from one URL."""
    reg = _registry()
    mine = _tenant_of(session.user)
    return {"users": [_view(u, reg[u]) for u in sorted(reg)
                      if _tenant_of(reg[u]) == mine]}


@router.post("/admin/users", status_code=201)
def create_user(body: dict = Body(default=None), session: Session = Depends(admin_gate)):
    """Create an account. **409 if the username already exists — `PATCH` is the update path.**

    ⛔ WHY 409 AND NOT AN UPSERT. `core.users.create_user` writes a fresh `_record()`, and a fresh
    record has NO `epoch` — so overwriting an existing account used to drop its epoch back to 0 and
    RESURRECT every cookie minted before its last password change. `core/users.py` now
    preserves-and-bumps on overwrite (which also closes it for `app.py`'s dialog, where "Add /
    update a user" calls the same function), but this route still refuses: an upsert that silently
    replaces an account's password, role and BU access because someone reused a username is not an
    update anybody asked for.
    """
    body = body or {}
    _require_store()
    uname = _clean_username(body.get("username"))
    pw = _clean_password(body.get("password"))
    role = _clean_role(body.get("role") or "user")
    bus = _clean_bus(body.get("bus") if body.get("bus") is not None else "all")
    modules = _clean_modules(body.get("modules") if body.get("modules") is not None else "all")
    name = str(body.get("name") or "").strip() or uname
    agent = str(body.get("agent") or "").strip() or None
    email = str(body.get("email") or "").strip() or None

    if uname in _registry():
        raise err(409, "user_exists", f"{uname} already exists — PATCH it to change it")
    # Wave 18 (C1-TENANT): an admin creates accounts in their OWN company, never another's —
    # the tenant is stamped from the session, not taken from the body.
    users.create_user(uname, pw, name, role=role, bus=bus, modules=modules,
                      agent=agent, email=email, tenant=_tenant_of(session.user))
    reg = _registry()
    if uname not in reg:
        # The store accepted the write and does not have it. Reporting 201 here would be the
        # "200 over a write that evaporated" failure the whole store-outage rule exists to prevent.
        raise err(503, "store_unavailable", "the account was not saved — try again")
    return {"user": _view(uname, reg[uname])}


@router.patch("/admin/users/{username}")
def update_user(username: str, body: dict = Body(default=None),
                session: Session = Depends(admin_gate)):
    """Change access on an existing account. Only the keys PRESENT in the body change.

    ⚠ NO EPOCH BUMP, and that is not an omission. `deps._user_for` re-reads the record on every
    request, so a narrowed role / BU / module grant applies to the target's LIVE session on its very
    next call — bumping the epoch would only add a forced re-login on top. `active: false` is the
    exception and it bumps, inside `core.users.set_active`, because a disabled account must lose its
    session rather than keep working until the cookie expires.
    """
    body = body if isinstance(body, dict) else {}
    reg = _registry()
    uname = str(username or "").strip().lower()
    if uname not in reg or _tenant_of(reg[uname]) != _tenant_of(session.user):
        # Wave 18 (C1-TENANT): a cross-tenant username answers exactly like a missing one —
        # 404, never 403, because "that account exists in another company" is itself a leak.
        raise err(404, "no_such_user", f"no account named {uname!r}")
    if not body:
        raise err(400, "empty_patch", "no fields to update")

    unknown = sorted(set(body) - {"name", "role", "bus", "modules", "active", "agent", "email"})
    if unknown:
        # A silently-ignored field is how a UI ends up believing it saved something it did not.
        raise err(400, "unknown_fields", f"cannot update: {unknown}")

    is_self = uname == session.uname
    role = _clean_role(body["role"]) if "role" in body else None
    active = bool(body["active"]) if "active" in body else None
    # THE ONE-CLICK LOCKOUT GUARD. APP_PASSWORD remains the real backstop for 'admin', so this is
    # not the thing that keeps the product reachable — it just stops an administrator removing
    # their own access with a toggle and having to go find the master password.
    if is_self and role is not None and role != "admin":
        raise err(400, "self_demote",
                  "you cannot remove your own administrator role — ask another admin")
    if is_self and active is False:
        raise err(400, "self_deactivate", "you cannot deactivate your own account")

    kw = {}
    if role is not None:
        kw["role"] = role
    if "name" in body:
        kw["name"] = str(body["name"] or "").strip() or uname
    if "bus" in body:
        kw["bus"] = _clean_bus(body["bus"])
    if "modules" in body:
        kw["modules"] = _clean_modules(body["modules"])
    if "agent" in body:
        kw["agent"] = str(body["agent"] or "").strip()      # '' clears the link
    if "email" in body:
        kw["email"] = str(body["email"] or "").strip()
    if kw:
        users.set_access(uname, **kw)
    if active is not None and active != bool(reg[uname].get("active", True)):
        users.set_active(uname, active)                     # bumps the epoch: kills live sessions

    fresh = _registry()
    return {"user": _view(uname, fresh[uname])}


@router.post("/admin/users/{username}/password", status_code=204)
def set_password(username: str, body: dict = Body(default=None),
                 session: Session = Depends(admin_gate)):
    """Rotate a password. **Revokes every outstanding session for that account** — the epoch bump
    happens inside the same read-modify-write as the hash (`core.users.set_password`), so the two
    can never disagree.

    204 with no body: there is nothing to say that the caller did not already know, and echoing the
    account back would invite a client to diff a response for a password change.
    """
    reg = _registry()
    uname = str(username or "").strip().lower()
    if uname not in reg or _tenant_of(reg[uname]) != _tenant_of(session.user):
        # Wave 18 (C1-TENANT): a cross-tenant username answers exactly like a missing one —
        # 404, never 403, because "that account exists in another company" is itself a leak.
        raise err(404, "no_such_user", f"no account named {uname!r}")
    pw = _clean_password((body or {}).get("password"))
    before = int(reg[uname].get("epoch") or 0)
    users.set_password(uname, pw)
    after = int((_registry().get(uname) or {}).get("epoch") or 0)
    if after <= before:
        # The whole value of this route is the revocation. If the epoch did not move, the sessions
        # were not revoked, and answering 204 would say they were.
        raise err(503, "store_unavailable",
                  "the password change did not persist — outstanding sessions were NOT revoked")
    return Response(status_code=204)


# ── C-PERM (wave 15): per-module access + permanent filters + hidden fields ──────────────────
#: The modules the permission editor governs — the GRID surfaces, the ones with a field schema
#: to filter and hide. Amendment 6 shipped the wall on `customer_data` alone and said this list
#: is the one place that changes when C-TOPIC lands. WAVE 16: it landed, so `product_data`
#: joins — and it is not cosmetic. `perms_v: 1` means an UNDECLARED module DENIES (amendment 4),
#: so a migrated account would be locked out of Product with no control anywhere to grant it:
#: fail-closed, but unadministrable. The editor grows its section with zero client edits
#: (S3 built it off this route's `modules` + `fields_by_module`).
_PERM_MODULES = ("customer_data", "product_data")


def _module_fields(key):
    """The field schema the editor builds its pickers from — the SAME vocabulary the grid uses,
    so an admin filters on exactly what the user sees.

    ⛔ Returned WITHOUT any per-user hiding applied, deliberately: this is the admin choosing
    what to hide, so hiding the choices would make a field unhideable the moment it was hidden
    once for somebody. The route is admin-gated; the projection a MEMBER receives is the one
    `perm_scope.visible_fields` narrows.

    ⚠ PER TOPIC (wave 16). `product_data` has its OWN canonical contract, and handing the
    editor the customer schema for it would let an admin build a wall out of columns that
    module does not have — `_clean_perms` would then 400 on the admin's own choices, or worse,
    a filter naming a field the product rows lack would DENY every row via `permits()`.
    The product contract is served CONSOLIDATED (every column) on purpose: a BU-scoped reader
    is served fewer columns, but the ADMIN is choosing what may ever be hidden.
    """
    import aios_grid
    src = aios_grid.product_fields() if key == "product_data" else aios_grid.FIELDS
    out = []
    for f in src:
        if not isinstance(f, dict) or not f.get("key"):
            continue
        out.append({"key": f["key"], "label": f.get("label") or f["key"],
                    "type": f.get("type") or "text",
                    **({"options": f["options"]} if isinstance(f.get("options"), list) else {}),
                    **({"pinned": True} if f.get("pinned") else {})})
    return out


def _clean_perms(v):
    """Validate a whole `perms` block. FAIL-CLOSED, and LOUD rather than lenient.

    ⛔ WHY THIS REFUSES WHERE `clean_filter_tree` DROPS. The view sanitiser is deliberately
    drop-per-node: one bad rule must never cost a user their whole saved view. A PERMISSION
    filter is the opposite situation — an admin types "agent is Tara", a leaf is silently
    dropped, and the stored wall is EMPTY. The admin sees "Saved", the account sees the whole
    book, and nothing anywhere says so. So anything that would be dropped is a 400 instead:
    the filter is re-validated with `clean_filter_tree` and the result must come back with the
    SAME leaf count it went in with.

    Shape (C-PERM amendment 2 — a `FilterTree` is a PAIR, `{conj?, nodes}`; `clean_filter_tree`
    validates the NODE LIST alone and never sees the root conjunction, so the two are checked
    separately and a bare node list is refused rather than silently read as `and`).
    """
    import aios_grid

    if v is None:
        return None
    if not isinstance(v, dict):
        raise err(400, "bad_perms", "perms must be an object keyed by module")
    unknown = sorted(set(v) - set(_PERM_MODULES))
    if unknown:
        raise err(400, "bad_perms",
                  f"unknown module keys: {unknown} — the permission editor governs "
                  f"{list(_PERM_MODULES)}")
    out = {}
    for key, raw in v.items():
        if not isinstance(raw, dict):
            raise err(400, "bad_perms", f"{key}: each entry must be an object")
        valid_keys = {f["key"] for f in _module_fields(key)}
        hidden = raw.get("hiddenFields") or []
        if not isinstance(hidden, list):
            raise err(400, "bad_perms", f"{key}: hiddenFields must be a list")
        bad = sorted({str(h) for h in hidden} - valid_keys)
        if bad:
            # A hiddenFields entry naming nothing hides nothing — and reads as a restriction
            # that is not there.
            raise err(400, "bad_perms",
                      f"{key}: hiddenFields names unknown fields {bad}")
        tree = raw.get("filter")
        clean_tree = None
        if tree not in (None, {}, []):
            if not isinstance(tree, dict) or not isinstance(tree.get("nodes"), list):
                raise err(400, "bad_filter",
                          f"{key}: filter must be {{conj?, nodes:[…]}} — a bare list would lose "
                          f"the root conjunction, and an 'or' wall read as 'and' restricts "
                          f"nothing it was meant to")
            conj = tree.get("conj", "and")
            if conj not in ("and", "or"):
                raise err(400, "bad_filter", f"{key}: conj must be 'and' or 'or'")
            nodes = tree["nodes"]
            cleaned = aios_grid.clean_filter_tree(nodes, valid_keys, cohort_ids=None)
            if _leaf_count(cleaned) != _leaf_count(nodes):
                raise err(400, "bad_filter",
                          f"{key}: the filter contains conditions this module cannot evaluate "
                          f"(an unknown field, an unknown operator, or a cohort leaf — cohorts "
                          f"are per-user and cannot be a permanent rule). Refused rather than "
                          f"saved with the bad conditions silently removed, which would store a "
                          f"weaker wall than the one on screen.")
            clean_tree = {"conj": conj, "nodes": cleaned}
        out[key] = {"access": bool(raw.get("access", True)),
                    "filter": clean_tree,
                    "hiddenFields": sorted({str(h) for h in hidden})}
    _refuse_unshapeable_bu(out)
    return out


def _refuse_unshapeable_bu(perms_out):
    """⛔ A BU CONDITION THE PUSHDOWN CANNOT READ IS A VALUE LEAK WEARING A CORRECT ROW LIST.

    Amendment 3 in one more place. `perm_scope.derive_pool_scope` recognises exactly the shapes
    that PIN a business unit: a top-level `dba eq` leaf, or a top-level OR-group of them — which
    is what `perm_migrate` emits and what the condition builder produces for "is any of". An
    admin composing the same intent a slightly different way (a NESTED group, `dba neq Royal`, a
    `dba` leaf one level down) produces a filter that still narrows the ROWS correctly through
    `permits()` — and leaves the pool built CONSOLIDATED, so every surviving row carries
    Fisch+Royal numbers. The row list looks right. The revenue is another BU's.

    That is precisely the defect amendment 3 exists for, re-entering through the editor R9 just
    shipped, and it cannot be caught downstream: by then the numbers are simply wrong, with
    nothing anomalous about them.

    So it is refused at the WRITE, where a person is present to fix it. The discriminator is
    narrow on purpose — the filter must MENTION `dba` and must fail to pin a team. A wall like
    `dba is Fisch OR revenue > 1000` genuinely does not confine anyone to one BU, and
    consolidated values are the correct answer for it, so it is not caught here (`derive` also
    refuses it) and must not be.
    """
    import core.perm_scope as perm_scope

    for key, e in perms_out.items():
        tree = e.get("filter")
        if not tree or not _mentions_dba(tree.get("nodes")):
            continue
        probe = {"role": "user", "perms_v": perm_scope.PERMS_VERSION,
                 "perms": {key: {"access": True, "filter": tree, "hiddenFields": []}}}
        team_id, _ = perm_scope.derive_pool_scope(probe, key)
        if team_id is None and _confines_to_one_bu(tree):
            raise err(400, "bu_condition_shape",
                      f"{key}: this filter restricts the business unit in a way the data layer "
                      f"cannot push into the query, so the rows would be correct while the "
                      f"revenue figures on them stayed consolidated across both units. Express "
                      f"the business-unit condition as a TOP-LEVEL condition — 'DBA is Fisch', "
                      f"or an 'any of' over the brands — and keep any other rules alongside it.")


def _guard_bu_widening(rec, cleaned, confirmed):
    """⛔ THE TWO-CLICK LEAK: a SECOND save that silently drops the business-unit condition.

    `_fold_legacy_scope` runs only while a record is un-migrated, which is correct — after the
    first save the filter IS the whole wall and the admin owns it. But the first save also
    CHANGES the tree (it folds the BU condition in), so a client holding the tree it submitted
    rather than the one the server returned will, on its next save, PUT a payload with no BU
    condition — and a whole-record replace deletes it. Measured, not theorised: the same client
    payload saved twice takes the pool scope from `(6, 'Ann')` to `(None, 'Ann')`, which is both
    business units.

    So a save that REMOVES the brand confinement is refused unless it says it means to. Widening
    BU access is a legitimate thing for an administrator to do — it is just never a thing to do
    by accident, and the difference between the two is one explicit flag.

    ⚠ Deliberately NOT solved by always folding the legacy scope: that would make a BU
    restriction permanent and unremovable, which contradicts R1 (the filter is the wall, and the
    admin edits it). The right shape is "you may widen, say so".
    """
    import core.perm_scope as perm_scope

    if confirmed:
        return
    stored = (rec.get("perms") or {}) if isinstance(rec.get("perms"), dict) else {}
    for key, new_entry in cleaned.items():
        if not new_entry.get("access", True):
            # Revoking the module entirely is the OPPOSITE of widening, and the filter on a
            # denied entry governs nothing. Guarding it would make "lock this account out"
            # require a confirm_widen flag, which reads as nonsense to whoever hits it.
            continue
        old = stored.get(key)
        if not isinstance(old, dict):
            continue                        # nothing stored yet: the fold already handled it
        old_tree, new_tree = old.get("filter"), new_entry.get("filter")
        if not old_tree or not _confines_to_one_bu(old_tree):
            continue                        # was not confined -> this save cannot widen it
        if new_tree and _confines_to_one_bu(new_tree):
            continue                        # still confined -> not a widening
        raise err(400, "bu_widening",
                  f"{key}: this would REMOVE the business-unit restriction and give the account "
                  f"both units. If that is intended, resend with confirm_widen: true. If it is "
                  f"not, reload the account's permissions first — saving a filter loaded before "
                  f"the last change drops the conditions it did not know about.")
    # A guard is only as good as its trigger: `perm_scope` is imported so a future reader sees
    # the connection to derive_pool_scope, which is what the removed condition was feeding.
    _ = perm_scope


def _mentions_dba(nodes):
    for n in nodes or ():
        if not isinstance(n, dict):
            continue
        if isinstance(n.get("children"), list):
            if _mentions_dba(n["children"]):
                return True
        elif n.get("colId") == "dba":
            return True
    return False


def _confines_to_one_bu(tree):
    """Does the BRAND STRUCTURE ALONE confine the reader to one business unit?

    ⚠ "Alone" is the whole precision of this function, and getting it wrong makes the guard
    refuse legitimate walls. `dba is Fisch OR ar_open > 1000` does NOT confine anybody — a
    high-balance Royal customer satisfies it — so consolidated values are the correct answer and
    refusing it would be a false positive. My first attempt evaluated synthetic rows carrying
    only `dba`, which made `ar_open > 1000` read false on every probe row and turned that exact
    wall into a "confinement" it is not.

    So every NON-`dba` leaf is treated as TRUE — the most generous reading, i.e. "suppose the
    reader satisfies everything else; can they still see both brands?" If yes, the wall does not
    confine and the pool is correctly consolidated. If no, the brand structure is doing the
    confining and `derive_pool_scope` must be able to see it, or the numbers are wrong.
    """
    def admits(node, brand):
        if isinstance(node.get("children"), list):
            kids = [admits(k, brand) for k in node["children"] if isinstance(k, dict)]
            if not kids:
                return True
            return any(kids) if node.get("conj") == "or" else all(kids)
        if node.get("colId") != "dba":
            return True                     # every other condition: assume satisfied
        want = str(node.get("value") or "").strip().lower()
        op = node.get("op")
        if op == "eq":
            return brand.lower() == want
        if op == "neq":
            return brand.lower() != want
        if op == "contains":
            return want in brand.lower()
        if op == "doesNotContain":
            return want not in brand.lower()
        if op == "isEmpty":
            return False                    # a probe brand is never blank
        if op == "isNotEmpty":
            return True
        return True                         # an op that cannot judge a brand does not confine

    nodes, conj = (tree.get("nodes") or []), tree.get("conj", "and")
    root = {"conj": conj, "children": nodes}
    admitted = {b for b in ("Fisch", "Royal", "Both") if admits(root, b)}
    if not admitted:
        return False                        # admits nothing at all — a different problem
    # 'Both' belongs to either unit, so it cannot by itself widen the answer.
    return not ({"Fisch", "Royal"} <= admitted)


def _leaf_count(nodes):
    n = 0
    for node in nodes or ():
        if isinstance(node, dict) and isinstance(node.get("children"), list):
            n += _leaf_count(node["children"])
        else:
            n += 1
    return n


def _fold_legacy_scope(rec, cleaned):
    """⛔ MIGRATING A RECORD MUST NEVER WIDEN IT. The invariant this function exists to hold.

    Writing perms stamps `perms_v: 1`, and that marker SWITCHES OFF the legacy `bus`/`agent`
    fallback in `perm_scope` (amendment 4 — absence must start meaning DENY). So a first write
    whose filter carries no BU condition silently promotes a Royal-only account to BOTH business
    units: the wall that used to come from `bus` is gone and nothing replaced it.

    Caught by the end-to-end leg in `verify_api` and by nothing else — every model-level check
    passed, because the model was doing exactly what it was told. The account simply received
    the other BU's customers on the wire.

    So on the FIRST perms write we fold the legacy scope in, as ordinary conditions, exactly as
    `perm_migrate` would: this IS R1's migration, performed at the moment the record acquires a
    perms block. From then on the admin sees those leaves in the editor (GET returns the folded
    filter) and can change them like any other condition — which is R1's whole point, and why
    this is a one-time fold rather than a permanent floor AND-ed on every write.

    Idempotent by construction: it runs only while the record is un-migrated, and the write that
    calls it is the write that migrates it.

    ⛔ THE FOLD IS PER MODULE, BECAUSE THE LEGACY FILTER IS CUSTOMER-SHAPED AND THE MODULES ARE
    NOT. `perm_migrate.filter_for` speaks `dba` and `agent` — both CUSTOMER columns. `product_data`
    has neither (19 fields, no brand column at all), and `apply_row_scope` evaluates the permanent
    filter with `permits()`, which DENIES anything unanswerable. So folding the customer wall into
    the product entry does not narrow the product grid, it EMPTIES it: measured at 0 rows of 2 for
    a `bus:[5]` account the moment an admin pressed Save.

    Nothing was wrong upstream — `_clean_perms` already refuses a `dba` condition typed against
    `product_data` (its leaf keys are validated per topic). This fold was the ONLY door a
    cross-topic leaf could come through, which is why it is the only place that needs the rule.

    ⚠ WHAT IS LOST HERE IS RECOVERED IN `perm_scope.derive_pool_scope`, NOT DISCARDED. Dropping the
    BU condition from an AND root WIDENS that module's row scope, and for the product grid the BU
    is a VALUES question anyway (`team_id` shapes `rev_ytd`/`qty_ytd`/`inv_value` — amendment 3).
    The pushdown falls back to the record's own `bus` for exactly the modules whose field
    vocabulary cannot express a BU, so the product pool is still BUILT as Fisch. The two halves
    ship together or the second one is a leak.
    """
    import core.perm_migrate as perm_migrate
    import core.perm_scope as perm_scope

    if perm_scope.is_migrated(rec):
        return cleaned
    legacy = perm_migrate.filter_for(rec)
    if not legacy:
        return cleaned                     # nothing to preserve: the account was unrestricted
    out = {}
    for key, e in cleaned.items():
        legacy_here = _prune_to_module(legacy, {f["key"] for f in _module_fields(key)})
        tree = e.get("filter")
        if not legacy_here:
            # This module cannot evaluate ANY of the legacy conditions. Fold nothing rather than
            # storing a wall it will read as "deny every row".
            out[key] = e
            continue
        if not tree or not tree.get("nodes"):
            folded = legacy_here
        else:
            # AND the two roots together. The submitted tree keeps its own conjunction by being
            # nested as a GROUP — flattening an `or` tree into an `and` root would turn the
            # admin's "A or B" into "A and B", which is a different and much narrower wall.
            folded = {"conj": "and",
                      "nodes": list(legacy_here["nodes"]) + [{"conj": tree.get("conj", "and"),
                                                              "children": list(tree["nodes"])}]}
        out[key] = dict(e, filter=folded)
    return out


def _prune_to_module(tree, valid_keys):
    """`tree` with every leaf this module cannot evaluate removed, or None if nothing survives.

    Used ONLY on the legacy fold above, where the alternative to pruning is a filter that denies
    every row. It is not a general sanitiser: `_clean_perms` REFUSES rather than drops, for the
    reason its own docstring gives (a silently-weakened wall reads as "Saved").

    ⚠ A GROUP THAT LOSES ANY CHILD IS DROPPED WHOLE. Half of an `or` group is a narrower rule than
    the admin's intent and half of an `and` group is a wider one; neither is the thing that was
    written, and a fold has no standing to invent a third meaning. All-or-nothing per group keeps
    the surviving conditions ones somebody actually declared.
    """
    if not isinstance(tree, dict):
        return None

    def keep(node):
        if not isinstance(node, dict):
            return None
        kids = node.get("children")
        if isinstance(kids, list):
            surviving = [keep(k) for k in kids]
            if not kids or any(s is None for s in surviving):
                return None
            return dict(node, children=surviving)
        return node if node.get("colId") in valid_keys else None

    nodes = [n for n in (keep(n) for n in tree.get("nodes") or ()) if n is not None]
    return {"conj": tree.get("conj", "and"), "nodes": nodes} if nodes else None


@router.get("/admin/users/{username}/perms")
def get_perms(username: str, session: Session = Depends(admin_gate)):
    """This account's permission block plus the field schema the editor needs to render it."""
    reg = _registry()
    uname = str(username or "").strip().lower()
    if uname not in reg or _tenant_of(reg[uname]) != _tenant_of(session.user):
        # Wave 18 (C1-TENANT): a cross-tenant username answers exactly like a missing one —
        # 404, never 403, because "that account exists in another company" is itself a leak.
        raise err(404, "no_such_user", f"no account named {uname!r}")
    import core.perm_scope as perm_scope

    rec = reg[uname]
    stored = rec.get("perms") or {}
    # S3 ask 3 — AN ENTRY FOR EVERY DECLARED MODULE, never a gap the client has to default.
    # The PUT is a whole-record replace over this same module list, so a key declared here and
    # omitted from `perms` would become an explicit DENY the moment anyone pressed Save. The
    # default sent is the EFFECTIVE one (`may_access`), so an un-migrated record shows the
    # access its legacy grant currently gives rather than a guess in either direction.
    perms_out = {}
    for k in _PERM_MODULES:
        e = stored.get(k)
        perms_out[k] = e if isinstance(e, dict) else {
            "access": bool(perm_scope.may_access(rec, k)), "filter": None, "hiddenFields": []}
    return {"username": uname,
            "perms": perms_out,
            # S3 ask 2 — the marker rides the GET. An absent module entry means DENY on a
            # migrated record and LEGACY on an un-migrated one: opposite meanings for identical
            # JSON, so the client cannot render honestly without knowing which world it is in.
            "perms_v": int(rec.get("perms_v") or 0),
            # `role == 'admin'` bypasses perms entirely (amendment 4) — sent so the editor can
            # say "Everything (admin)" instead of rendering stored rules that do not apply.
            "is_admin": perms.is_admin(rec),
            "modules": [{"key": k, "label": _MODULE_LABELS.get(k, k)} for k in _PERM_MODULES],
            # S3 ask 1 — CANONICAL SHAPE: a BARE ARRAY of fields per module key. Not the nav
            # schema envelope; the editor needs the field list and nothing else, and one shape
            # beats two readings of a sentence.
            "fields_by_module": {k: _module_fields(k) for k in _PERM_MODULES}}


@router.put("/admin/users/{username}/perms")
def put_perms(username: str, body: dict = Body(default=None),
              session: Session = Depends(admin_gate)):
    """Replace this account's permission block wholesale.

    WHOLESALE, not a merge: "remove this restriction" has to be expressible, and a merge cannot
    express a deletion without a second vocabulary for it.

    ⚠ NO EPOCH BUMP, for the same reason `PATCH /admin/users/{u}` does not bump one:
    `deps._user_for` re-reads the record on every request, so a narrowed wall applies to the
    target's LIVE session on its very next call. Bumping would only add a forced re-login on top
    of a change that has already taken effect.
    """
    body = body if isinstance(body, dict) else {}
    reg = _registry()
    uname = str(username or "").strip().lower()
    if uname not in reg or _tenant_of(reg[uname]) != _tenant_of(session.user):
        # Wave 18 (C1-TENANT): a cross-tenant username answers exactly like a missing one —
        # 404, never 403, because "that account exists in another company" is itself a leak.
        raise err(404, "no_such_user", f"no account named {uname!r}")
    if "perms" not in body:
        raise err(400, "empty_patch", "no perms to save")

    # THE SELF-LOCKOUT GUARD, the twin of `self_demote` on the PATCH route. An admin bypasses
    # perm_scope entirely, so this cannot brick them today — but it stops an administrator
    # writing a wall for themselves that would bite the moment their role changed.
    if uname == session.uname:
        raise err(400, "self_perms",
                  "you cannot set permissions on your own account — ask another admin")

    cleaned = _clean_perms(body.get("perms")) or {}
    cleaned = _fold_legacy_scope(reg[uname], cleaned)
    _guard_bu_widening(reg[uname], cleaned, bool(body.get("confirm_widen")))
    users.set_access(uname, perms=cleaned)
    fresh = _registry()
    rec = fresh.get(uname) or {}
    if int(rec.get("perms_v") or 0) < users.PERMS_VERSION:
        # The store took the write and did not record it. A 200 here would tell an administrator
        # the wall is up when it is not — the one direction this must never fail in.
        raise err(503, "store_unavailable",
                  "the permissions were not saved — the account is UNCHANGED")
    return {"username": uname, "perms": rec.get("perms") or {},
            "perms_v": int(rec.get("perms_v") or 0)}


_MODULE_LABELS = {"customer_data": "Customer", "product_data": "Product"}


@router.get("/settings")
def settings(session: Session = Depends(require_session)):
    """The session's OWN settings — any authenticated user, not admin-only.

    `user` comes from `routes_auth._public_user`, deliberately reusing the ONE definition of what a
    client may know about itself (never a hash, never the epoch). `scope` restates it as the two
    things the Settings UI shows, and `tenant` names only THIS tenant — enumerating the others
    would answer a question about our customer list.
    """
    modules = perms.allowed_modules(session.user)
    return {
        # The build this Space is running. Behind the session on purpose — see `main.py::health`,
        # which refuses it for being the first URL a scanner finds. `deploy_web.py` stamps it as a
        # Space secret at every deploy; absent means "somebody started this container by hand".
        "version": os.environ.get("AIOS_VERSION") or "unknown",
        "user": _public_user(session.user),
        "scope": {
            "bus": perms.allowed_bu_labels(session.user),
            "team_id": perms.scope_team_id(session.user),
            "agent": perms.scope_agent(session.user),
            "modules": sorted(modules) if modules is not None else "all",
            # C-PERM: the caller's OWN effective wall, read-only. The grid uses it to grey
            # pickers rather than to enforce — hidden fields are stripped from every data wire
            # regardless, so a client that ignores this is narrowed anyway, never widened.
            "perms": session.user.get("perms") or {},
        },
        "tenant": {"key": session.tenant, "name": getattr(session.runtime, "name",
                                                          session.tenant)},
        # CHROME ONLY. Every /admin route re-checks the role server-side; this exists so the client
        # does not paint a "Manage users" button that 403s.
        "admin": perms.is_admin(session.user),
        # Wave 19 (R3, contract C2): the same courtesy for the LOOPABLE plane — the Settings rail
        # paints its entry iff this is true, and `routes_platform_admin` refuses every request
        # that is not, so a client that ignored this would gain exactly nothing. Derived from the
        # ONE predicate rather than restated: a second copy of a two-condition wall is a second
        # place for one of the conditions to go missing.
        "platformAdmin": platform_admin.is_platform_admin(session.user),
    }