"""routes_grid.py — X2's write seam over HTTP: the SECOND adapter on `core.grid_events` (EXIT-1c).

`POST /api/v1/grid/events` takes the component's event objects VERBATIM — the same objects the
Streamlit host receives through the component value slot, unchanged — and runs them through the
same `core.grid_events.handle_events` the Streamlit adapter runs. That is the whole point of
EXIT-1a: one implementation of every permission wall, two transports. A route that re-validated
anything here would be a second wall to keep in step, and the two would drift on the first change.

DEDUP IS PER REQUEST, and that is a deliberate limit, not an oversight. The Streamlit adapter's
`seen_ids` lives in `st.session_state` — genuinely per user session — because the client resends
its recent 24-event window on every emit inside one page session. A stateless API has no such
dict, and inventing a per-user server-side one would be exactly the resident per-tenant state
EXIT-4a exists to remove. So: ids are deduped WITHIN a request body (the resend window's whole
purpose — a batch that repeats an id processes it once), and a genuinely replayed request is
handled by the operations being idempotent. The one event where a replay is observable is
`add_to_list` (it would add the same pids to the same cohort twice — a set union, so the
membership is unchanged, but the toast count repeats). Noted rather than papered over; a
server-side idempotency key belongs with D2's session mirror (C-2).

STORE DOWN = 503. `fallback_ws` is None on this adapter, so `core.grid_events` raises
`StoreUnavailable` rather than writing to an in-memory workspace no API request could ever read
back. A 200 over a write that evaporated is the failure this rule exists to prevent.
"""
import datetime as dt
import time

from fastapi import APIRouter, Body, Depends

from deps import Session, err, module_gate, perms, require_session

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

MODULE = "customer_data"

#: The client's own resend window (`app.py`'s `[-24:]`). A body larger than this is not a
#: legitimate client — refuse it rather than doing 500 store writes on one request.
_MAX_EVENTS = 24


def _ctx(session: Session, fields, pids, **kw):
    from core import grid_events
    import aios_grid
    import core.perm_scope as perm_scope

    # Wave 16 C-TOPIC: the ctx is TOPIC-SHAPED. The product scope swaps all three of the
    # things a write is validated against — the module the permission wall reads, the canonical
    # contract the hidden-field closure runs over, and the TABLE OPS the write lands in.
    # Getting any one of them from the other topic is the "validated against the wrong field
    # contract" near-miss the wave-15 routes_products header warned about.
    scope_key = str(kw.get("scope_key") or "")
    if scope_key == "product":
        import modules.product_data as pd
        from routes_products import MODULE as _PMOD, pd_fields

        module, canonical = _PMOD, pd_fields(consolidated=True)
        kw.setdefault("table", pd.TABLE_OPS)
        hidden = perm_scope.hidden_keys(session.user, module, canonical)
    elif scope_key.startswith("ut_"):
        # Wave 18 C3-UT: a user table has no module in the permission wall (its wall is
        # `user_tables.may_open`, already applied by the assembly this route ran first), so
        # the hidden-field closure is EMPTY rather than borrowed from another topic's contract.
        import core.table_store as table_store

        kw.setdefault("table",
                      table_store.make(f"{scope_key}_table_workspace", st=session.runtime))
        hidden = frozenset()
    else:
        module, canonical = MODULE, aios_grid.FIELDS
        hidden = perm_scope.hidden_keys(session.user, module, canonical)
    return grid_events.EventCtx(
        uname=session.uname, allowed_pids=pids, fields=fields, admin=session.admin,
        # C-PERM: the write wall's field half, on the EVENTS transport too. This route takes the
        # component's event objects verbatim, so a hidden key would otherwise arrive here with
        # nothing between it and the store.
        hidden_keys=hidden,
        # ⭐ WAVE 25 (R6b, closes D-16) — THE TENANT HANDLE, on EVERY scope. This is the line
        # D-16's exit condition names, and without it the rest of the fix is inert: the handler
        # falls back to the module-global `core.store`, which is tenant #0's repo, so a Nurilab
        # user's documents, cohorts and — through `_tops` — their whole customer/product table
        # workspace were written into Royal Imports' dataset. Note it is set for the CUSTOMER
        # and PRODUCT branches too, not only `ut_`: those two never passed a scoped `table`, so
        # they were the ones actually resolving to tenant #0 on every request.
        st=session.runtime,
        fallback_ws=None, seen_ids={}, **kw)


#: The surfaces the ONE grid serves. `cohort` is the same table over hand-curated SETS rather
#: than over the whole scoped pool — `app.py:page_cohort` calls the same `_table_grid` with
#: `cohort_mode=True, scope_key='cohort'`. `product` (wave 16 C-TOPIC) is the SKU table:
#: same engine, its own pool, field contract and workspace BUCKET (routes_products).
_SCOPES = ("customer", "cohort", "product")


def _scope_or_400(raw):
    """⛔ REFUSE AN UNKNOWN SCOPE, never default it. A typo silently served as `customer` would
    hand a user the whole book on a page they opened to see one cohort — the widening direction,
    which is the one that must fail closed.

    Wave 18 (C3-UT): a `ut_`-prefixed scope names a USER TABLE and passes through here —
    existence and the per-table wall are enforced by `routes_tables.ut_assembly` (404/403),
    which every consumer of such a scope goes through. Passing an unknown ut key therefore
    still fails closed, just one layer down where the store can actually be consulted."""
    scope = (raw or "customer").strip().lower()
    if scope.startswith("ut_"):
        return scope
    if scope not in _SCOPES:
        raise err(400, "bad_scope",
                  f"scope must be one of {', '.join(_SCOPES)} — refusing to guess")
    return scope


@router.get("/workspace")
def workspace(scope: str = "customer",
              session: Session = Depends(require_session)):
    """The durable table workspace for this session: views, fields, overlays, folders.

    ⛔ WAVE 21 (C4): the wall is TOPIC-SHAPED, so the DEPENDENCY is session-only and each scope
    asserts its own gate below. The old `module_gate("customer_data")` dependency 403'd a
    `ut_*` workspace for any tenant whose catalogue omits the customer module (loopable/
    nurilab/gtmlab ship modules w/o it) — and the client then fell back to the shared
    localStorage bucket + demo data, which is exactly the "new database shows RI's customer
    fields" defect. A user table's real wall is `user_tables.may_open`, enforced inside
    `ut_assembly` (404/403, one layer down where the store can be consulted).

    ⚠ WHY THIS EXISTS (S1↔S2, 2026-07-30 — an X2 AMENDMENT, see the split doc). X2 fixes
    `/customers` at the exact shape `verify_fields_contract.py` referees, so the workspace cannot
    ride along in it without forking the thing that gate keeps single-sourced. Without a
    workspace route the standalone shell's write path would be WRITE-ONLY: events persist
    server-side and nothing reads them back on reload, so a saved view looks lost to the user
    even though the store has it. This route closes that read-back gap at its own URL.

    `allowed_pids` is passed so a SHARED view's `memberPids` are re-scoped to THIS reader — the
    wave-9 leak rule. Omitting it would hand a Fisch-scoped user a member list built by a
    full-access user.
    """
    from core import grid_events
    from routes_customers import grid_assembly

    scope = _scope_or_400(scope)
    # Wave 21 C4 — the per-scope gate the dependency no longer asserts: customer/cohort need
    # the customer module; product asserts PRODUCT_MODULE in its branch; ut_* needs only a
    # session (its wall is the table's own, inside the assembly).
    if not (scope == "product" or scope.startswith("ut_")):
        session.require(MODULE)
    # ── Wave 18 C3-UT: a USER TABLE'S workspace — the third topic through the one wire. The
    # storage key has no (bu, agent) because a user table has no Odoo scope; per-user by the
    # table's own wall (creator or admin, `user_tables.may_open`).
    if scope.startswith("ut_"):
        from routes_tables import ut_assembly

        storage_key = f"{session.tenant}:{scope}:{session.uname}"
        try:
            # ⭐⭐ W31-T20 / D-174 — `with_rows=False`. THIS ROUTE RENDERS NO ROW: the grid fetches
            # them from `/tables/{key}/rows` or `/odoo-tables/{key}/rows` in the same paint, and
            # `useCustomerData.load` fires both calls in one `Promise.all`. Building the table here
            # bought nothing and cost everything — it is why `?scope=ut_odoo_gl_lines` answered
            # `409 window_required` six times out of six on the live deploy while the ROWS door for
            # the same table served a page in ~150 ms, i.e. a grid that could not be opened at all.
            g = ut_assembly(session, scope, storage_key=storage_key, with_rows=False)
        except grid_events.StoreUnavailable:
            raise err(503, "store_unavailable", "the tenant store is unavailable")
        workspace = g["workspace"]
        workspace["overlays"] = g["ws"].get("overlays") or {}
        # The field contract rides the cheap re-read on EVERY topic — see the customer branch
        # below for why (a user table's first cohort changes its contract too: `_cohorts(ctx)`
        # is scope-parameterized, so `ut_` surfaces have their own sets).
        workspace["fields"] = g["fields"]
        workspace["measures"] = g["measures"]
        workspace["measureSets"] = g["measure_sets"]
        # ⚠ `g["derived"]`, NOT `{}` (2026-08-04). R9 gave every topic its own cohort sets, and
        # `ut_assembly` has been building this topic's cohort CELLS since — but this route threw
        # them away, so the Locked-views column arrived on the cheap re-read with permanently
        # empty values. A column that exists and can never have one is worse than an absent
        # column: it reads as "this record is in no locked view", which is a claim.
        workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()}
        workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
        try:
            from core import users as _users
            workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
            workspace["userAvatars"] = {k: v for k, v in
                                        _users.avatar_map(tenant=session.tenant).items()
                                        if str(v).startswith("data:image/")}
        except Exception:
            workspace["userOptions"] = []
            workspace["userAvatars"] = {}
        workspace["scopeKey"] = scope
        # ⭐ W31-T20 — R6's SECOND SENTENCE REACHES THE BROWSER. On a read-through grid too large
        # for one window the pid set is empty, so cohort membership and a shared view's
        # `memberPids` are UNRESOLVED here (both fail closed, and `workspace_wire` already stamps
        # `missing` on a shortened cohort). An unannounced empty scope is the silent limit R6
        # forbids; this is the announcement, and W31-T22 is the ticket that renders it.
        workspace["limits"] = g.get("limits") or []
        return {"workspace": workspace}

    # ── Wave 16 C-TOPIC: the PRODUCT surface has its own assembly (own pool, own field
    # contract, own workspace BUCKET) and the contract's own storage key. It branches before
    # the customer derivation because the customer storage key embeds (bu, agent) while the
    # product one is deliberately `all:all` — one product workspace per user (the wave doc's
    # C-TOPIC line), since BU narrows the product VALUES, not which workspace you own.
    if scope == "product":
        from routes_products import MODULE as PRODUCT_MODULE, product_assembly

        # ⛔ THE GATE CHANGES WITH THE SCOPE (wave 21 C4: the dependency is session-only now,
        # so this line IS the product wall — not a second one on top of customer_data).
        session.require(PRODUCT_MODULE)
        storage_key = f"{session.tenant}:product-list:{session.uname}:all:all"
        try:
            g = product_assembly(session, scope=scope, storage_key=storage_key)
        except grid_events.StoreUnavailable:
            raise err(503, "store_unavailable", "the tenant store is unavailable")
        workspace = g["workspace"]
        workspace["overlays"] = g["ws"].get("overlays") or {}
        # R9 made cohorts per-topic, so the PRODUCT surface has its own sets and its own
        # first-cohort contract change. Same reason as the customer branch below.
        workspace["fields"] = g["fields"]
        # The measure channel is EMPTY on this topic (customer-grain descope) — stated
        # explicitly so the client's pickers grey rather than guess.
        workspace["measures"] = g["measures"]
        workspace["measureSets"] = g["measure_sets"]
        # `g["derived"]` — the same correction as the user-table branch above, for the same
        # reason: `product_assembly` builds this topic's cohort cells and this route discarded
        # them. The measure half stays empty on this topic by descope, which is a different
        # statement and one the offer already makes.
        workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()}
        workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
        try:
            from core import users as _users
            workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
            workspace["userAvatars"] = {k: v for k, v in
                                        _users.avatar_map(tenant=session.tenant).items()
                                        if str(v).startswith("data:image/")}
        except Exception:
            workspace["userOptions"] = []
            workspace["userAvatars"] = {}
        workspace["scopeKey"] = scope
        return {"workspace": workspace}

    # The storage key mirrors the host's own convention byte-for-byte (app.py's `_table_grid`
    # call sites) with the session's TENANT in the host's hardcoded slot, so localStorage state
    # carries across embed ⇄ standalone on the same browser.
    # Wave 15 R1 — the SAME derivation the pool uses (`routes_customers._team_agent`), so the
    # storage key cannot drift from the scope it names. Value-identical for every migrated
    # record (verify_perm_scope section D proves the derivation reproduces the legacy scope
    # exactly), so nobody's saved localStorage state moves when the migration runs.
    import core.perm_scope as perm_scope
    team_id, agent = perm_scope.derive_pool_scope(session.user, MODULE)
    bu = team_id if team_id is not None else "all"
    if scope == "cohort":
        storage_key = f"{session.tenant}:cohort:{session.uname}:{bu}"
    else:
        storage_key = f"{session.tenant}:customer-list:{session.uname}:{bu}:{agent or 'all'}"

    # ⛔ THE WIRE SHAPE IS THE SHARED PROJECTION (`aios_grid.workspace_wire`), not the store
    # shape. The first version returned the store dict with no `storageKey` — and the client
    # validator requires one, so the standalone shell DISCARDED the whole workspace: saved views
    # never rendered, `cohortMode` never arrived, and the Cohort route silently drew the Customer
    # surface. One projection for both servers is the fix that cannot drift.
    try:
        g = grid_assembly(session, scope=scope, storage_key=storage_key)
    except grid_events.StoreUnavailable:
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    workspace = g["workspace"]
    # ADDITIVE to the wire: the caller's own overlay cells. The grid reads overlays from the
    # /customers rows, but this route is the cheap read-back a write can be verified against
    # (verify_api's overlay probe) without paying the pool call. Per-user by construction —
    # `table_workspace` is this session's workspace.
    workspace["overlays"] = g["ws"].get("overlays") or {}
    # ⭐ THE FIELD CONTRACT, 2026-08-04 — and it is NOT decoration.
    #
    # `fields_from_workspace(ws, cohorts=bool(cohort_lists))` appends the derived "Locked
    # views" column ONLY when the caller owns at least one cohort. So a user's FIRST cohort
    # CHANGES THE FIELD CONTRACT — and the rows call that used to be the only carrier of
    # `fields` is deliberately never re-fetched on a write (it is a 15-minute-cached Odoo
    # pull; this route is the cheap re-read). The client therefore had no way to learn about
    # that column short of a remount, which is the second half of the owner's "it only shows
    # up when I switch modules and come back".
    #
    # ⚠ `g["fields"]` is the SAME list `_payload` serves on `/customers` — same assembly,
    # same permission wall (`hidden_keys` already applied) — so the two wires cannot disagree
    # about what a column is. Taking it from anywhere else would fork the contract that
    # verify_fields_contract.py exists to keep single-sourced.
    workspace["fields"] = g["fields"]
    # ── the STANDALONE measure channel (owner item 1, 2026-07-31) ────────────────────────────
    # The embed receives these as top-level render args; standalone lifts them off THIS route
    # (useCustomerData merges them into the payload slots the grid already reads). They ride
    # the workspace rather than /customers because a durable write re-reads exactly this route
    # (WORKSPACE_STALE), so a new measure column populates without refetching the heavy pool.
    workspace["measures"] = g["measures"]
    workspace["measureSets"] = g["measure_sets"]
    # Derived cells (cohort column + measure columns), keyed by pid. JSON object keys are
    # strings; the client indexes with String(pid).
    workspace["derived"] = {str(pid): cells for pid, cells in g["derived"].items()}
    # Who is looking (permissions verdicts) + the assignee choices for `user`-typed columns —
    # the two other host-only render args the shell was missing (fail-closed without them:
    # restricted fields uneditable, assignee picker empty).
    workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)}
    try:
        from core import users as _users
        workspace["userOptions"] = _users.assignable_people(tenant=session.tenant)
        # Wave 14 C-AVATAR — the options vocabulary's companion: display name -> data URL.
        # Absent entries fall back to the client's initials disc; a non-data: value is
        # dropped (defence in depth beside the write-side wall in routes_auth).
        workspace["userAvatars"] = {k: v for k, v in
                                    _users.avatar_map(tenant=session.tenant).items()
                                    if str(v).startswith("data:image/")}
    except Exception:
        workspace["userOptions"] = []
        workspace["userAvatars"] = {}

    # THE SURFACE STAMP — a MIRROR of the host's own (`_table_grid`'s hide/cohort stamps).
    # ⚠ Deliberately NOT `hideViews`: `cohortMode` is what swaps the Views panel for the cohort
    # list, and two switches for one behaviour would drift.
    workspace["scopeKey"] = scope
    if scope == "cohort":
        workspace["cohortMode"] = True
        # The create pane offers "this cohort table only" (default) vs "all customer tables".
        workspace["scopeChoice"] = True
    return {"workspace": workspace}


# ─────────────────────────────────────────────── the TIME-SERIES channel (C-TS, 2026-08-02)
_TS_BUCKETS = ("week", "month", "quarter", "year")
_TS_MAX_BUCKETS = 120
_TS_MAX_FIELDS = 12
_TS_MAX_PIDS = 5000
_TS_MAX_LAST_N = 120
#: C-TSWIN (wave 14): sliding windows cost ONE aggregate query per (metric, bucket) — the
#: price of "bucket N == the grid cell at bucket N's end". The product cap keeps a legal
#: 120-bucket × 12-metric ask from becoming 1,440 queries in one request; typical panels
#: (12 × 3) sit two orders of magnitude under it. Dated amendment in the split doc.
_TS_MAX_CELLS = 720


def _ts_start_of(bucket, d):
    """The calendar START of the bucket holding `d` (week = Monday, the vocabulary rule)."""
    if bucket == "week":
        return d - dt.timedelta(days=d.weekday())
    if bucket == "month":
        return d.replace(day=1)
    if bucket == "quarter":
        return d.replace(month=((d.month - 1) // 3) * 3 + 1, day=1)
    return d.replace(month=1, day=1)


def _ts_next(bucket, d):
    if bucket == "week":
        return d + dt.timedelta(days=7)
    if bucket == "year":
        return d.replace(year=d.year + 1)
    step = 1 if bucket == "month" else 3
    m = d.month + step
    return dt.date(d.year + (m - 1) // 12, (m - 1) % 12 + 1, 1)


def _ts_prev(bucket, d):
    if bucket == "week":
        return d - dt.timedelta(days=7)
    if bucket == "year":
        return d.replace(year=d.year - 1)
    step = 1 if bucket == "month" else 3
    y, m = d.year, d.month - step
    while m < 1:
        m += 12
        y -= 1
    return dt.date(y, m, 1)


def _ts_label(bucket, start):
    if bucket == "week":
        return f"Wk of {start.strftime('%b %d')}"
    if bucket == "month":
        return start.strftime("%b %Y")
    if bucket == "quarter":
        return f"Q{(start.month - 1) // 3 + 1} {start.year}"
    return str(start.year)


@router.post("/grid/timeseries")
def grid_timeseries(body: dict = Body(default=None), scope: str = "customer",
                    session: Session = Depends(module_gate(MODULE))):
    """Pooled measure values per calendar bucket — the time-series view's data channel.

    C-TSWIN (wave 14, ruling R1): AS-OF semantics. Each metric's OWN stored window is
    re-resolved per bucket with `today := min(bucket end, real today)` — `ytd` is cumulative
    from Jan 1, `last_90_days` trailing, `all_time` cumulative ever; bucket N equals what the
    grid's measure column would show if today were bucket N's end. Fixed-range (`custom`)
    windows cannot slide and are dropped per field as `window_fixed`. Rows carry
    `window: {kind, label}` so a cumulative row cannot be misread as periodic, and there is
    NO `total` column — sliding windows overlap, so a sum of columns would double-count.

    The client sends the pids its view currently matches (the filter IS the scope); the server
    intersects them with the session's book, so the request can only ever NARROW. Values are
    POOLED aggregates computed by the semantic layer's own expression (the additivity law: an
    average is computed at pool grain, never averaged over per-customer answers). A bucket
    with no rows is 0 for a sum/count and null for anything else; a bucket that has not
    STARTED yet is null for every kind — an unstarted period's YTD is unanswered, not 0.
    """
    from core import measure_resolve
    from harness import windows as _wn
    from routes_customers import _pool_stamp, _team_agent, allowed_pids

    # Wave 16 C-TOPIC: the channel is CUSTOMER-GRAIN (measure_resolve's own grain), so the
    # product surface is refused in words rather than answered with an id-space accident —
    # product pids are CRC32 hashes and would mostly fall outside the customer book anyway,
    # but "mostly" is not a wall. The client hides the mode for this topic; this is the
    # server's half of the same refusal.
    _sc = _scope_or_400(scope)
    if _sc == "product" or _sc.startswith("ut_"):
        raise err(400, "bad_timeseries",
                  "this surface has no time-series channel — measures are customer-grain")

    body = body or {}
    bucket = body.get("bucket")
    if bucket not in _TS_BUCKETS:
        raise err(400, "bad_timeseries", "bucket must be one of week, month, quarter, year")
    raw_fields = body.get("fields")
    if not isinstance(raw_fields, list) or not raw_fields:
        raise err(400, "bad_timeseries", "fields must be a non-empty list of field keys")
    if len(raw_fields) > _TS_MAX_FIELDS:
        raise err(400, "bad_timeseries", f"at most {_TS_MAX_FIELDS} fields per request")
    raw_pids = body.get("pids")
    if not isinstance(raw_pids, list) or not raw_pids:
        raise err(400, "bad_timeseries", "pids must be a non-empty list")
    if len(raw_pids) > _TS_MAX_PIDS:
        raise err(400, "bad_timeseries", f"at most {_TS_MAX_PIDS} pids per request")
    try:
        wanted = {int(p) for p in raw_pids}
    except (TypeError, ValueError):
        raise err(400, "bad_timeseries", "pids must be integers")
    pool = wanted & {int(p) for p in allowed_pids(session)}
    if not pool:
        # 403, not an empty 200: the caller asked about customers outside their book, and an
        # all-zero series over nobody would read as "no activity", not "not yours".
        raise err(403, "out_of_scope", "none of those customers are in your book")

    span = body.get("span")
    if not isinstance(span, dict):
        raise err(400, "bad_timeseries", "span must be {'lastN': n} or {'from': .., 'to': ..}")
    today = time.strftime("%Y-%m-%d")
    t = dt.date.fromisoformat(today)
    n = span.get("lastN")
    starts = []
    if n is not None:
        if not isinstance(n, int) or isinstance(n, bool) or not (1 <= n <= _TS_MAX_LAST_N):
            raise err(400, "bad_timeseries", f"lastN must be 1..{_TS_MAX_LAST_N}")
        cur = _ts_start_of(bucket, t)
        starts = [cur]
        for _ in range(n - 1):
            cur = _ts_prev(bucket, cur)
            starts.append(cur)
        starts.reverse()
    else:
        try:
            d_from = dt.date.fromisoformat(str(span.get("from")))
            d_to = dt.date.fromisoformat(str(span.get("to")))
        except (TypeError, ValueError):
            raise err(400, "bad_timeseries", "span.from/to must be ISO dates (YYYY-MM-DD)")
        if d_from > d_to:
            d_from, d_to = d_to, d_from
        cur = _ts_start_of(bucket, d_from)
        while cur <= d_to:
            starts.append(cur)
            if len(starts) > _TS_MAX_BUCKETS:
                raise err(400, "bad_timeseries",
                          f"that span is more than {_TS_MAX_BUCKETS} {bucket} buckets - "
                          f"narrow it")
            cur = _ts_next(bucket, cur)
        if not starts:
            raise err(400, "bad_timeseries", "the span holds no buckets")
    ends = [_ts_next(bucket, s) - dt.timedelta(days=1) for s in starts]

    rt = session.runtime
    if not rt.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    import modules.customer_data as cl_mod
    # Read-only route: it must not consume a one-shot label-correction ack the browser has
    # not seen yet (the same rule event validation follows).
    ws = cl_mod.table_workspace(session.uname, consume_corrections=False)
    fdefs = ws.get("fields") or {}
    keys, dropped, seen = [], [], set()
    for k in raw_fields:
        k = str(k or "")
        if not k or k in seen:
            continue
        seen.add(k)
        fd = fdefs.get(k)
        if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict):
            # Named, never silent: v1 eligibility is measure-backed fields only (C-TS).
            dropped.append({"field": k, "reason": "not_a_measure_field"})
            continue
        keys.append(k)
    # ⚡ WAVE 17 R9 (amendment 2026-08-03, GRID's cross-fence ask) — A SHEET WITH NO MEASURE
    # FIELDS IS NO LONGER A 400. It answers with the BUCKET GRID and no rows.
    #
    # Why this is the honest direction and not a loosening: the bucket starts and their labels
    # are SERVER math (`_ts_start_of` / `_ts_next` / `_ts_label`), and R9 has the client
    # synthesizing snapshot rows for preset columns that have no window to slide. Those rows
    # must be painted under the SAME headings as everything else, so the client needs the
    # columns even when the server has no series to put in them. Refusing the whole request
    # meant the panel could offer nothing at all on this tenant — the shipped contract has zero
    # measure-backed presets (`aios_grid.py`: "this branch is currently MEMBERLESS"), which is
    # what item 5 is actually about.
    #
    # Nothing is invented by this: `rows` is empty and `meta.dropped` NAMES every field and why.
    # A 400 is still returned for a request that is malformed (bad bucket, bad span, no fields
    # at all) — this only stops treating "I asked about columns you cannot serve" as an error.
    mfields, slid_keys = [], []
    for k in keys:
        m = dict(fdefs[k]["measure"])
        if (m.get("window") or {}).get("kind") == "custom":
            # C-TSWIN: a fixed date range cannot slide across buckets — named, never silent,
            # the same posture as not_a_measure_field.
            dropped.append({"field": k, "reason": "window_fixed"})
            continue
        mfields.append({"key": k, "measure": m})
        slid_keys.append(k)
    # Same amendment as above: every metric being fixed-range is a sheet with no SERIES, not a
    # broken request. The columns still stand, and `window_fixed` still names each refusal.
    if len(starts) * len(mfields) > _TS_MAX_CELLS:
        raise err(400, "bad_timeseries",
                  "that ask is too wide - narrow the span or pick fewer metrics")

    team_id, agent = _team_agent(session)
    stamp = _pool_stamp(rt, team_id, agent)
    problems = []
    bucket_bounds = [(s.isoformat(), e.isoformat()) for s, e in zip(starts, ends)]
    # No slidable metrics = nothing to resolve. Skipping the call rather than asking the
    # resolver about an empty list keeps the memo free of a meaningless key.
    answers = measure_resolve.series_values(
        mfields, bucket, bucket_bounds,
        team_id, frozenset(pool), today, stamp, rt.series_memo,
        on_error=lambda tag, e: problems.append(str(e)[:200])) if mfields else {}

    columns = []
    for s, e in zip(starts, ends):
        col = {"key": s.isoformat(), "label": _ts_label(bucket, s),
               "from": s.isoformat(), "to": e.isoformat()}
        if e > t:
            col["partial"] = True
        columns.append(col)
    rows = []
    for k in slid_keys:
        ans = answers.get(k)
        if ans is None:
            dropped.append({"field": k, "reason": "unresolvable"})
            continue
        vals_by = ans.get("values") or {}
        agg_kind = str(ans.get("agg") or "sum")
        zero_fill = agg_kind in ("sum", "count")
        vals = []
        for s in starts:
            if s > t:
                # C-TSWIN: an unstarted period is unanswered for EVERY agg kind — a future
                # month's YTD zero-filled to 0 would read as "the year reset".
                vals.append(None)
            else:
                vals.append(vals_by.get(s.isoformat(), 0 if zero_fill else None))
        wspec = (fdefs[k].get("measure") or {}).get("window")
        wnorm = _wn.normalize(wspec) or {}
        rows.append({"field": k, "label": str(fdefs[k].get("label") or k)[:120],
                     "agg": agg_kind, "values": vals,
                     "window": {"kind": str(wnorm.get("kind") or ""),
                                "label": _wn.label(wspec)}})
    meta = {"pool": len(pool), "today": today, "bucket": bucket}
    if dropped:
        meta["dropped"] = dropped
    if problems:
        meta["problems"] = problems[:5]
    return {"columns": columns, "rows": rows, "meta": meta}


#: C-CAL caps (wave 17, owner item 9 / ruling R4). A month of days, a handful of metrics.
#: Every one of these is a 400 WITH ITS REASON, never a silent trim: a calendar that quietly
#: answered 20 of the 31 days asked about would paint eleven blank cells that look like days
#: with no activity.
_CAL_MAX_GROUPS = 31
_CAL_MAX_FIELDS = 6
_CAL_MAX_CELLS = 186


@router.post("/grid/calendar_metrics")
def grid_calendar_metrics(body: dict = Body(default=None), scope: str = "customer",
                          session: Session = Depends(module_gate(MODULE))):
    """Per-DAY measure values with each metric's own window slid to that day (C-CAL / R4).

    ⛔ WHY THIS EXISTS AT ALL. The calendar's summary cells used to aggregate the row VALUES the
    grid already held — which for a measure column means "every member's YTD **as of today**",
    summed and printed under a date in March. The number was arithmetically fine and semantically
    a lie: it answered a question about today while sitting in a cell labelled with another day.
    R4: a metric in a day cell is computed AS OF THAT DAY.

    The shape differs from the time-series channel in exactly one way, and it is the reason this
    is a separate route rather than a parameter: **every group carries its OWN pid set**. A
    calendar day holds the records the date field placed there, so day-to-day the subject
    changes. One `allowed_pids` for the whole request — the TS channel's shape — would compute
    each day over everybody, which is a different question again.

    Static (non-measure) fields are NOT served here. They have no window to slide, so the
    client's own per-day aggregation over row values stays correct for them; sending them would
    invite a second implementation of arithmetic that already works.
    """
    from core import measure_resolve
    from harness import windows as _wn
    from routes_customers import _pool_stamp, _team_agent, allowed_pids

    # The same refusal the TS channel makes, for the same reason: measures are customer-grain
    # and product pids are CRC32 hashes of SKU codes. "Mostly outside the book" is not a wall.
    _sc_cal = _scope_or_400(scope)
    if _sc_cal == "product" or _sc_cal.startswith("ut_"):
        raise err(400, "bad_calendar_metrics",
                  "the product surface has no measure channel yet — measures are customer-grain")

    body = body or {}
    raw_groups = body.get("groups")
    if not isinstance(raw_groups, list) or not raw_groups:
        raise err(400, "bad_calendar_metrics",
                  "groups must be a non-empty list of {key: 'YYYY-MM-DD', pids: [...]}")
    if len(raw_groups) > _CAL_MAX_GROUPS:
        raise err(400, "bad_calendar_metrics",
                  f"at most {_CAL_MAX_GROUPS} days per request (one month)")
    raw_fields = body.get("fields")
    if not isinstance(raw_fields, list) or not raw_fields:
        raise err(400, "bad_calendar_metrics", "fields must be a non-empty list of field keys")
    if len(raw_fields) > _CAL_MAX_FIELDS:
        raise err(400, "bad_calendar_metrics", f"at most {_CAL_MAX_FIELDS} metrics per request")
    if len(raw_groups) * len(raw_fields) > _CAL_MAX_CELLS:
        raise err(400, "bad_calendar_metrics",
                  "that ask is too wide - fewer days or fewer metrics")

    book = {int(p) for p in allowed_pids(session)}
    groups, seen_days = [], set()
    for g in raw_groups:
        if not isinstance(g, dict):
            raise err(400, "bad_calendar_metrics", "every group must be an object")
        day = str(g.get("key") or "")
        try:
            d = dt.date.fromisoformat(day)
        except (TypeError, ValueError):
            raise err(400, "bad_calendar_metrics",
                      "every group key must be an ISO date (YYYY-MM-DD)")
        if day in seen_days:
            raise err(400, "bad_calendar_metrics", f"day {day} appears twice")
        seen_days.add(day)
        raw_pids = g.get("pids")
        if not isinstance(raw_pids, list):
            raise err(400, "bad_calendar_metrics", "every group needs a pids list")
        try:
            wanted = {int(p) for p in raw_pids}
        except (TypeError, ValueError):
            raise err(400, "bad_calendar_metrics", "pids must be integers")
        # NARROW-ONLY, per group. The client sends what its calendar placed; the server can
        # only ever remove from that, never add.
        groups.append((day, d, frozenset(wanted & book)))
    if sum(len(p) for _, _, p in groups) > _TS_MAX_PIDS:
        raise err(400, "bad_calendar_metrics",
                  f"at most {_TS_MAX_PIDS} customer references per request")

    rt = session.runtime
    if not rt.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    import modules.customer_data as cl_mod
    ws = cl_mod.table_workspace(session.uname, consume_corrections=False)
    fdefs = ws.get("fields") or {}
    keys, dropped, seen = [], [], set()
    for k in raw_fields:
        k = str(k or "")
        if not k or k in seen:
            continue
        seen.add(k)
        fd = fdefs.get(k)
        if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict):
            # The TS channel's vocabulary, deliberately reused rather than re-coined: the client
            # already knows how to say these three words to a reader.
            dropped.append({"field": k, "reason": "not_a_measure_field"})
            continue
        if ((fd["measure"].get("window") or {}).get("kind") == "custom"):
            dropped.append({"field": k, "reason": "window_fixed"})
            continue
        keys.append(k)
    if not keys:
        raise err(400, "bad_calendar_metrics",
                  "none of the requested fields are measure fields with a window that can "
                  "slide to a day")

    today = time.strftime("%Y-%m-%d")
    t = dt.date.fromisoformat(today)
    team_id, agent = _team_agent(session)
    stamp = _pool_stamp(rt, team_id, agent)
    problems = []
    values = {k: {} for k in keys}
    for day, d, pids in groups:
        # A day that has not happened is unanswered for EVERY aggregate kind — the unstarted
        # bucket law at day grain. Answering 0 would say "we sold nothing", which is a claim
        # about a day nobody has lived through yet.
        if d > t or not pids:
            for k in keys:
                values[k][day] = None
            continue
        answers = measure_resolve.series_values(
            [{"key": k, "measure": dict(fdefs[k]["measure"])} for k in keys],
            "day", [(day, day)], team_id, pids, today, stamp, rt.series_memo,
            on_error=lambda tag, e: problems.append(str(e)[:200]))
        for k in keys:
            ans = answers.get(k)
            if ans is None:
                values[k][day] = None
                continue
            vals_by = ans.get("values") or {}
            zero_fill = str(ans.get("agg") or "sum") in ("sum", "count")
            values[k][day] = vals_by.get(day, 0 if zero_fill else None)

    for k in keys:
        if all(v is None for v in values[k].values()):
            # Every day unanswerable is a FIELD-level failure, and saying so is the difference
            # between "no activity that month" and "this metric could not be computed".
            dropped.append({"field": k, "reason": "unresolvable"})
    out = {"values": values, "today": today,
           "windows": {k: {"kind": str((_wn.normalize(
               (fdefs[k].get("measure") or {}).get("window")) or {}).get("kind") or ""),
               "label": _wn.label((fdefs[k].get("measure") or {}).get("window"))}
               for k in keys}}
    if dropped:
        out["dropped"] = dropped
    if problems:
        out["problems"] = problems[:5]
    return out


@router.post("/grid/events")
def grid_events_route(body: dict = Body(default=None),
                      session: Session = Depends(require_session)):
    """`{events: [<component event objects, verbatim>]}` → `{results, doc?, toast?}`.

    Wave 21 C4: session-only dependency, per-scope gate below — the write door must admit the
    same sessions the read door (`/workspace`) admits, or a tenant without the customer module
    can SEE its own user tables and not write to them."""
    from core import grid_events
    from routes_customers import grid_assembly

    events = (body or {}).get("events")
    if events is None and isinstance(body, dict) and body.get("type"):
        events = [body]                    # a single event object, the legacy shape
    if not isinstance(events, list):
        raise err(400, "bad_events", "expected {events: [...]}")
    if len(events) > _MAX_EVENTS:
        raise err(400, "too_many_events",
                  f"at most {_MAX_EVENTS} events per request (the client's resend window)")

    # Validated with the SAME predicate as the read route. An unrecognised scopeKey used to be
    # passed through verbatim, and `core.grid_events` only ever compares it to 'cohort' — so a
    # typo degraded silently to customer-scope behaviour on a WRITE. Read and write must agree on
    # what a scope is, or the surface you read is not the surface you wrote.
    scope = _scope_or_400((body or {}).get("scopeKey"))
    # Wave 21 C4 — same per-scope gate as /workspace (read and write doors must agree).
    if not (scope == "product" or scope.startswith("ut_")):
        session.require(MODULE)
    # This assembly validates the write; it is not a payload the browser will render. Leave
    # one-shot field-name correction acks queued for the subsequent /workspace refresh.
    # Wave 16 C-TOPIC: the PRODUCT topic gets the product assembly — product field contract,
    # product pids, and (below) the product TABLE OPS, so a product event is validated against
    # and lands in the product bucket. The measure/cohort context is honestly EMPTY there:
    # `clean_measure_field` refuses measure creates on this surface by construction (the
    # customer-grain descope), which is the fail-closed shape, not an accident.
    if scope == "product":
        from routes_products import MODULE as PRODUCT_MODULE, product_assembly

        session.require(PRODUCT_MODULE)   # the product wall (dependency is session-only, C4)
        g = product_assembly(session, consume_corrections=False)
    elif scope.startswith("ut_"):
        # Wave 18 C3-UT — the user-table wall (creator/admin) is inside the assembly; the
        # measure/cohort context is honestly EMPTY (customer-grain machinery, no meaning here).
        # ⭐⭐ WAVE 30 / W30-T30 (owner item 2: *"when I click hide fields it crash… no matter the
        # size"*). This used to be `ut_assembly(...)`, which builds the whole table to validate a
        # write it then throws away: `scoped_pool` allocates a dict per row and sorts them, so ONE
        # hide-fields checkbox rebuilt ~33k order rows before the event was even dispatched.
        # ⛔ NOTHING IS VALIDATED LESS. The comment above still holds — this assembly is the
        # permission wall and the admission context — and `ut_write_ctx` returns the SAME six keys
        # this route reads, with a pid set derived from exactly the row ids `scoped_pool` would
        # have kept. What it does not do is materialise the rows nobody here looks at.
        from routes_tables import ut_write_ctx

        g = ut_write_ctx(session, scope)
        # ⭐⭐ W31-T20 / D-174 — A WRITE THAT NEEDS THE ROW SET REFUSES OUT LOUD, and this is the
        # half that is easy to skip because the code already "fails closed" without it. When a
        # read-through grid's population exceeds one window the pid set is EMPTY, and every
        # pid-bearing handler in `grid_events` then answers `False` — `overlay_patch` and
        # `add_to_list` both `return False` for a pid not in `allowed_pids`. On the wire that is
        # `rerender: false` and HTTP 200: a write the user watched succeed, that did nothing
        # ([[lost-write-looks-like-failed-read]]). Schema-only events — hide a field, save a view,
        # rename a column — name no pid and are untouched, which is D-170.
        # ⚠ THE TEST IS THE EVENT'S OWN KEYS, not a kind list: a new pid-bearing event type would
        # otherwise inherit the silent no-op the day it is added.
        if g.get("limits"):
            named = [e for e in events
                     if isinstance(e, dict) and (e.get("pid") is not None or e.get("pids"))]
            if named:
                lim = g["limits"][0]
                raise err(409, "pid_scope_unresolved",
                          f"this database is served through the connector mirror and its rows "
                          f"cannot be listed in one window, so a change addressed to particular "
                          f"records ({len(named)} of {len(events)} here) cannot be admitted — "
                          f"{lim.get('cause') or 'the row set is unresolved'}. "
                          f"{lim.get('recommendation') or ''}".strip())
    else:
        g = grid_assembly(session, scope=scope, consume_corrections=False)
    # ⚠ THE MEASURE CONTEXT IS NOT OPTIONAL (2026-07-31). Without `measure_offer`,
    # `clean_measure_field` had an empty admission list and every measure-column create over
    # HTTP was silently refused; without `measure_keys`, `clean_filter_tree` stripped every
    # measure CONDITION out of a saved view. The embed always passed these; the API adapter
    # simply had not been given them — the standalone shell could read measures it could
    # never write.
    ctx = _ctx(session, g["fields"], g["pids"], scope_key=scope,
               measure_keys=frozenset(m["key"] for m in g["measures"]),
               resolved_ids=frozenset(g["measure_sets"]),
               cohort_ids=frozenset(c["id"] for c in g["lists"]),
               measure_offer=tuple(g["measures"]),
               visible_views=tuple(g["views"]))

    # Per-event results so the client can tell which of a batch landed — the component's own
    # bridge has no response channel at all, so this is strictly more than the embed gets.
    results = []
    try:
        for one in events:
            eid = str(one.get("id") or "") if isinstance(one, dict) else ""
            rerender = grid_events.handle_one(one, ctx)
            results.append({"id": eid, "rerender": bool(rerender)})
    except grid_events.StoreUnavailable:
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — none of your changes were saved")

    out = {"results": results, "rerender": any(r["rerender"] for r in results)}
    if ctx.out.doc is not None:
        # ⭐ C4 / W30-T27 — `docPayload` IS THE NAME THE CLIENT ALREADY DECLARES. `types.ts` has
        # carried `docPayload?: {pid, docId, name, mime, data_b64}` since C5, and `Documents.tsx`
        # matches it against the fetch it is waiting on — while this route has been answering
        # `doc`, which `apiBridge.ts` deliberately drops. One object, emitted under the name the
        # consumer looks for, so F's wiring needs no translation step to get wrong.
        # ⚠ `doc` stays for one wave: nothing in the client reads it, but a gate might, and
        # removing a key to save six bytes is not worth a red nobody predicted.
        out["doc"] = out["docPayload"] = ctx.out.doc
    if ctx.out.toast is not None:
        out["toast"] = ctx.out.toast

    # ── ⭐ owner item 2 (2026-08-03): THE NEW MEASURE COLUMN'S VALUES, ONE ROUND TRIP SOONER ──
    #
    # Creating a measure column cost the browser TWO sequential trips before a single number
    # appeared: this one to persist the field, then a whole `/workspace` to compute it. The
    # second cannot start until the first lands (the resolver reads the PERSISTED field), so the
    # wait was structural, not slow code — the owner's "it takes some time for the data to
    # populate". The values are computed here instead, immediately after the write, and ride
    # this response.
    #
    # ⚠ IT COSTS NOTHING EXTRA TO COMPUTE. The expensive part is one DuckDB aggregate over the
    # book, and `rt.measure_memo` is keyed on (pool stamp, scope, pool, measure, window) — so
    # the `/workspace` re-read that still follows HITS the memo instead of doing this work. The
    # query happens once either way; only its position moved.
    #
    # ⚠ NARROW ON PURPOSE. Gated to an actual measure-column write, so an overlay edit or a
    # cohort add — the overwhelming majority of events — never pays for a second assembly.
    #
    # ⚠ AND IT IS A SHORTCUT, NOT A PATH. Any failure is swallowed: `WORKSPACE_STALE` still
    # fires from `rerender`, and the re-read still delivers these values exactly as it does
    # today. Nothing depends on this having worked.
    if out["rerender"] and scope in ("customer", "cohort") and any(
            isinstance(e, dict) and e.get("type") == "field_upsert"
            and str(((e.get("field") or {}) if isinstance(e.get("field"), dict) else {})
                    .get("key") or "").startswith("measure_")
            for e in events):
        try:
            fresh = grid_assembly(session, scope=scope, consume_corrections=False)
            out["derived"] = {str(pid): cells for pid, cells in fresh["derived"].items()}
        except Exception:
            pass
    # ⚠ NOTHING IS INVALIDATED HERE, on purpose. The runtime cache holds ONLY the scope-shaped
    # Odoo pool (see `routes_customers._pool_rows`), and no event on this route can change an
    # Odoo column — Odoo is read-only. Everything an event DOES change (overlays, fields, views,
    # folders, cohorts) is re-read from the store on the next request. An earlier version cleared
    # `pool_cache` after an `overlay_patch`, which threw away an expensive Odoo pull to refresh
    # data that was never in it.
    return out