api/routes_customers.py

"""routes_customers.py — X2's read + write of the customer table, BU-SCOPED (EXIT-3b / EXIT-2a).

Two things happen here that did not happen in the pre-wave `main.py`:

  1. **ROWS ARE SCOPED TO THE SESSION.** The old `/api/customers` served `cl.pool()` — the whole
     book, to anyone holding the shared APP_PASSWORD. Now the pool is built with
     `(team_id, agent_name)` derived from the USER RECORD, so a Royal-only user never receives a
     Fisch row and an agent-linked login never receives another rep's book. The scope is applied
     at the QUERY, not as a post-filter, so there is no moment at which the other BU's rows exist
     in this response.

  2. **THE OVERLAY FORK IS GONE.** `aios-web/api/data/overlay.json` was a SECOND writable home
     for the same user-owned fields the Streamlit app keeps in the tenant store — two truths, and
     whichever process you asked last was right. Reads and writes now both go through
     `modules.customer_data`'s table-workspace functions: ONE store (C1c, ARCHITECTURE §1a rule 3).

⚠ ONE STORE IS NOT YET ONE CACHE (strangler-period, booked honestly). `core.store.get()` is
cache-first per PROCESS, so a write from the Streamlit container is invisible to a running API
container until its cache is refreshed, and vice versa. Deleting the fork removes the second
SOURCE OF TRUTH; it does not make the two runtimes coherent. The fix is X4/Postgres (task C-4,
owner-blocked on B-3), and no test here may claim read-your-writes ACROSS runtimes — a TestClient
proof is single-process and would report green on exactly the thing that is still broken.
"""
import time

from fastapi import APIRouter, Body, Depends

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

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

#: The surface these routes serve. Both legs are gated on it, so a user without the grant gets a
#: 403 rather than an empty table that looks like "you have no customers".
MODULE = "customer_data"

_CACHE_TTL = 900       # the pool build is slow (Odoo + reconciliation); 15 min, as before


def _pool_rows(session: Session):
    """The reconciled Odoo pool for this session's SCOPE — the slow part, and the only part that
    may be shared between users.

    ⛔ WHAT MAY BE CACHED HERE, AND WHY THE LINE IS EXACTLY HERE. The cache key is
    `(team_id, agent)` and the cached value is the raw pool: Odoo-source columns only. That is
    genuinely scope-shaped — two users with the same BU and the same book are asking the same
    question, and `cl.pool()` is expensive (an Odoo pull plus reconciliation).

    The FULL PAYLOAD is NOT cacheable on this key, and caching it here was a real defect I shipped
    and then removed. `fields` comes from `fields_from_workspace(ws)` — that user's own `custom_`
    and `measure_` columns — and every overlay cell comes from `ws['overlays']`; the table
    workspace is read as `data[username]` and written as `patch_table_overlay(uname, …)`, i.e.
    PER USER. So two Royal-only users with no agent link share `(6, None)` and the second one
    would have been served the FIRST one's private notes and private columns. The scope key was
    right for the pool and wrong for everything wrapped around it.

    It survived a green 129-check battery because every fixture user had a DISTINCT
    `(team_id, agent)` pair, so no two of them ever collided on the key — the test set could not
    express the bug. `verify_api.py` now carries a same-scope second user for exactly this.
    """
    rt = session.runtime
    team_id, agent = _team_agent(session)
    return _pool_for(rt, team_id, agent)


def _pool_for(rt, team_id, agent):
    """The cached pool for an explicit scope — session-free so the prewarm thread and the
    stale-refresh path can call it. STALE-WHILE-REFRESH (scope_cache): once a copy exists no
    request blocks on the 10–30s Odoo rebuild again; only a scope's FIRST-ever build does.

    DEBT-2 (2026-08-04): while the tenant's RESOLVED Odoo source is PAUSED this path never
    goes live — it serves the in-process copy at any age, else the persisted pause-time
    snapshot (restart-safe), else answers 503. Never a WIDER scope's snapshot: handing a
    scoped user the consolidated rows would widen their book, which is worse than an error."""
    import modules.customer_data as cl
    import routes_keychain

    key = ("pool", team_id, agent)

    if routes_keychain.odoo_paused(rt):
        hit = rt.pool_cache.get(key)
        if hit:
            return hit[1]
        snap = routes_keychain.load_pool_snapshot(rt, team_id, agent)
        if snap is not None:
            rt.pool_cache[key] = snap          # seed, so the memo stamps stay coherent
            return snap[1]
        raise err(503, "connector_paused",
                  "this data source is paused and no snapshot exists for your scope — "
                  "an admin can resume it under Settings → Connectors")

    def _build():
        # ⚠ THE SCOPE GOES INTO THE BUILDER. `pool(agent_name, team_id)` is the same reconciled
        # builder the Streamlit page uses — passing the scope here is what makes the isolation a
        # property of the QUERY instead of a filter someone can forget to apply downstream.
        return cl.pool(agent, team_id)

    def _evict():
        # Bounded: a scope cache that only ever grows is a memory leak in a shared process.
        if len(rt.pool_cache) > 16:
            for stale in sorted(rt.pool_cache, key=lambda k: rt.pool_cache[k][0])[:8]:
                rt.pool_cache.pop(stale, None)

    return scope_cache.get(rt.pool_cache, key, _CACHE_TTL, _build, _evict)


def warm_default(rt):
    """Boot prewarm: the consolidated pool `(None, None)` — the scope every admin and every
    all-BU account lands on. Called from main.py's prewarm thread only."""
    _pool_for(rt, None, None)


def _team_agent(session: Session):
    """The `(team_id, agent)` this session's POOL is built with — ONE derivation point, used by
    `_pool_rows` and `grid_assembly` alike, so the cache key and the query can never disagree.

    ⛔ WAVE 15 R1: THIS NOW COMES FROM THE PERMANENT FILTER (C-PERM amendment 3). `team_id` is
    not a row filter — `customer_data._pool_build` passes it into `cust._cust_rev` three times
    and into `_cadence_bulk`, so it decides what `rev`/`ly`/`ltm`/`aov`/`status` MEAN. Enforcing
    a BU purely as a post-filter would keep the row list right and silently consolidate every
    number. So the pushdown survives as a DERIVATION OF the declared filter rather than a second
    wall beside it, and `perm_scope.derive_pool_scope` falls back to the legacy `bus`/`agent`
    derivation for any record the migration has not reached yet.
    """
    import core.perm_scope as perm_scope
    return perm_scope.derive_pool_scope(session.user, MODULE)


def _pool_stamp(rt, team_id, agent):
    """The cached pool's build timestamp — the DATA STAMP in every measure-memo key, so a pool
    refresh invalidates the memoised answers exactly when the underlying rows changed."""
    entry = rt.pool_cache.get(("pool", team_id, agent))
    return entry[0] if isinstance(entry, tuple) and entry else 0


def _measure_err(tag, e):
    try:
        import harness.telemetry as _tel
        _tel.error(f"api:{tag}", e)
    except Exception:
        pass


def grid_assembly(session: Session, scope: str = "customer", storage_key: str = "",
                  consume_corrections: bool = True):
    """ONE assembly of this session's grid state, shared by `/customers`, `/workspace` and the
    events route (2026-07-31 — the standalone measure gap, owner item 1).

    What it adds over the pre-wave hand-rolled `_payload` loop, and why it replaced it:

      * rows go through `aios_grid.rows_from_pool` — the SAME builder the embedded host uses,
        so `lat`/`lon` (the Map view's data), `_created` and every derived column ride each row
        by construction instead of by a second loop that drifts. The hand loop was written to
        mirror the pre-Map contract and silently dropped the coordinates: the shell's Map view
        had nothing to plot ("the Map no longer works").
      * `derived` carries the cohort column's cells AND the measure columns' values, resolved
        through `core.measure_resolve` (EXIT-5's extraction of the `_cl_measure_*` family).
        Without them every measure column the owner built rendered BLANK in the shell.
      * `measures` (the offer) and `measure_sets` (condition answers) are computed here so the
        events route can finally validate measure fields/conditions instead of refusing them
        (an empty `measure_offer` made `clean_measure_field` reject every create over HTTP).

    Memos live on the TENANT RUNTIME (`rt.measure_memo` / `rt.mset_memo`) — bounded by the
    module's own clear-past-cap rule, keyed on (stamp, scope, pool identity, question), nothing
    user-shaped in them.
    """
    import aios_grid
    from core import grid_events, measure_resolve

    import core.perm_scope as perm_scope

    rt = session.runtime
    team_id, agent = _team_agent(session)
    rows_src = _pool_for(rt, team_id, agent)
    # ⛔ THE ROW WALL, APPLIED BEFORE `pids` IS TAKEN. Everything downstream is bounded by that
    # frozenset — `allowed_pids` for the workspace, cohort membership, measure resolution — so
    # scoping here means a row this account may not see never enters ANY of them, rather than
    # being filtered out of one payload and surviving in another.
    #
    # Evaluated against the CANONICAL field list, not the per-user assembled one, for two
    # reasons: the assembled list is not built yet (it needs `pids`), and a permanent filter may
    # only ever name a canonical field anyway — `routes_admin._clean_perms` validates it against
    # exactly this schema and 400s otherwise. `permits()` denies on anything it cannot answer.
    rows_src = perm_scope.apply_row_scope(rows_src, session.user, MODULE, aios_grid.FIELDS)
    pids = frozenset(r["pid"] for r in rows_src if r.get("pid") is not None)
    ws = grid_events.table_workspace(
        _ctx_for(session, pids), allowed_pids=pids,
        consume_corrections=consume_corrections)
    workspace, fields, views, lists = aios_grid.workspace_wire(
        ws, session.uname, set(pids), defs={}, scope_key=scope, storage_key=storage_key)
    # THE FIELD WALL — a TRANSITIVE closure (C-PERM amendment 5), so hiding a field also hides
    # every formula computed FROM it. Formulas evaluate in the browser from `{ref}`s, so
    # shipping a dependent formula while withholding its input either leaks the input through
    # the formula's value or silently computes a wrong one; only removing both is coherent.
    # Applied AFTER workspace_wire because custom + measure columns are what it must cover.
    hidden = perm_scope.hidden_keys(session.user, MODULE, fields)
    if hidden:
        fields = [f for f in fields if f.get("key") not in hidden]
        # The field LIST and the ROW payload are two different wires. Narrowing only the first
        # would leave the value sitting in the second, where anything can read it.
        rows_src = [perm_scope.strip_row(r, hidden) for r in rows_src]

    today = time.strftime("%Y-%m-%d")
    stamp = _pool_stamp(rt, team_id, agent)
    measures = measure_resolve.offer(team_id, on_error=_measure_err)
    measure_sets = measure_resolve.condition_sets(
        [v.get("config") or {} for v in (views or [])], None, team_id, pids, today, stamp,
        rt.mset_memo, on_error=_measure_err)
    # The derived channel: cohort membership cells + measure column values, ONE dict — the
    # same read-only channel the embed host hands to rows_from_pool.
    derived = aios_grid.cohort_cells(lists)
    for pid, cells in measure_resolve.column_values(
            fields, team_id, pids, today, stamp, rt.measure_memo,
            on_error=_measure_err).items():
        derived.setdefault(pid, {}).update(cells)

    return {"rows_src": rows_src, "pids": pids, "ws": ws, "workspace": workspace,
            "fields": fields, "views": views, "lists": lists, "derived": derived,
            "measures": measures, "measure_sets": measure_sets, "today": today,
            "team_id": team_id}


def _payload(session: Session):
    """`{fields, rows, today, docs, pulled_at}` — X2's shape, which `verify_fields_contract.py`
    referees. Rows are now built by `aios_grid.rows_from_pool` (embed == standalone by
    construction); see `grid_assembly` for what that fixed.

    ⭐ `docs` joined the shape in wave 30 (W30-T37 / contract C4). Named here rather than left to
    the reader because a docstring that still lists the OLD shape is a stale comment on correct
    code — this repo's D-73 — and it is the first thing anyone greps to learn the payload.

    ⚠ `rows_src` is the SHARED cached list — `rows_from_pool` reads it and builds NEW dicts,
    never mutating a cached row (the same-scope-second-user leak rule).
    """
    import aios_grid

    g = grid_assembly(session)
    rows = aios_grid.rows_from_pool(
        g["rows_src"], g["fields"], g["ws"].get("overlays"), derived=g["derived"])
    # ⭐ C4 / D-138 (W30-T37) — THE DOCUMENTS PRODUCER FOR THE CUSTOMER SCOPE. The write door
    # (`doc_add`/`doc_fetch`/`doc_delete`) never stopped working and every client half is
    # complete; what vanished with `app.py` at EXIT-6 was the only thing that ever set this key.
    # All six `onDoc*` handlers in `CustomerGrid.tsx` read `payload?.docs ? … : undefined`, so an
    # ABSENT key — not a broken one — is what has been switching the whole feature off.
    #
    # ⛔ IMPORTED, NEVER RE-SERIALISED. `core.grid_events.docs_for` is the ONE serialiser and
    # `routes_tables` (the `ut_*` scope) calls the SAME function with the same argument order.
    # A matching pair here is precisely how the wave-29 close-out reintroduced its own defect in
    # the opposite direction inside a single commit ([[one-question-two-normalizers]]).
    # ⚠ `g["pids"]` is the row set this session is ALREADY scoped to — `docs_for` has no
    # "every document in the tenant" mode to reach for, deliberately.
    from core import grid_events as _ge
    return {"fields": g["fields"], "rows": rows,
            # `today` rides the payload because every relative date condition must resolve against
            # the TENANT's day, never the browser's — a client that falls back to its own clock
            # disagrees with the server for everyone west of it.
            "today": g["today"],
            "docs": _ge.docs_for(g["pids"], scope_key="customer", uname=session.uname,
                                 admin=session.admin, st=session.runtime),
            "pulled_at": time.strftime("%Y-%m-%d %H:%M")}


def _ctx_for(session: Session, pids):
    """An EventCtx for the READ path — no fallback workspace, so a store outage is a 503 rather
    than a phantom in-memory workspace an API request cannot persist."""
    from core import grid_events
    return grid_events.EventCtx(
        uname=session.uname, allowed_pids=frozenset(pids or ()), fields=[],
        # C-PERM: the write wall's field half. Computed from the CANONICAL contract because
        # `fields=[]` here — the closure only needs the schema, not this user's column list.
        hidden_keys=_hidden_for(session),
        admin=session.admin, fallback_ws=None, seen_ids={})



def _hidden_for(session: Session):
    """The fields this session's permissions hide — the write wall's half of C-PERM.

    Read paths strip these from both wires so they cannot be SEEN; this is what stops them
    being WRITTEN by a caller who knows the key. Evaluated against the canonical contract, the
    same schema `routes_admin._clean_perms` validates a hiddenFields entry against.
    """
    import aios_grid
    import core.perm_scope as perm_scope

    return perm_scope.hidden_keys(session.user, MODULE, aios_grid.FIELDS)

def allowed_pids(session: Session):
    """The pids this session may touch — the POOL's own ids, so the write wall and the read scope
    can never disagree.

    Reads `_pool_rows` rather than `_payload`: the wall only needs identities, and going through
    the full payload would pay for a workspace read and a row assembly on every write.

    ⛔ THE PERMANENT FILTER APPLIES HERE TOO, AND FORGETTING IT IS A WRITE-WITHOUT-READ HOLE.
    `_pool_rows` is built with the DERIVED pushdown, which expresses only what a `(team_id,
    agent)` pair can express. Any part of the wall the pushdown cannot carry — `revenue > 1000`,
    a nested group, a condition on any other column — leaves the pool WIDER than the filter. Read
    paths close that gap with `apply_row_scope`; without the same call here the write wall would
    be the wider set, and a restricted user could PATCH a row this API will not show them.
    Same function, same order as `grid_assembly`, so the two walls cannot drift.
    """
    import core.perm_scope as perm_scope
    import aios_grid

    rows = perm_scope.apply_row_scope(_pool_rows(session), session.user, MODULE,
                                      aios_grid.FIELDS)
    return frozenset(r["pid"] for r in rows if r.get("pid") is not None)


@router.get("/customers")
def customers(session: Session = Depends(module_gate(MODULE))):
    return _payload(session)


@router.patch("/customers/{pid}")
def patch_customer(pid: int, body: dict = Body(default=None),
                   session: Session = Depends(module_gate(MODULE))):
    """Write the EDITABLE overlay stratum only — Odoo stays read-only, forever.

    Routed through `core.grid_events.handle_one` as an `overlay_patch` event rather than writing
    the store directly: that handler is where the per-key `permissions.edit` wall, the pid wall
    and the truncation rules live, and a second implementation of those would be a second set of
    them to keep in step. The response reports what was ACCEPTED, which is not always what was
    asked for.
    """
    from core import grid_events

    updates = dict(body or {})
    if not updates:
        raise err(400, "empty_patch", "no fields to update")
    pool = allowed_pids(session)
    if pid not in pool:
        # 403, not 404: the pid may well exist — it is simply not in this session's book, and
        # saying "no such customer" would confirm the opposite to anyone who guessed right.
        raise err(403, "out_of_scope", "that customer is not in your book")
    payload = _payload(session)
    ctx = grid_events.EventCtx(
        uname=session.uname, allowed_pids=pool, fields=payload["fields"],
        admin=session.admin, fallback_ws=None, seen_ids={},
        hidden_keys=_hidden_for(session))
    try:
        grid_events.handle_one(
            {"id": f"patch:{pid}:{time.time_ns()}", "type": "overlay_patch",
             "pid": pid, "updates": updates}, ctx)
    except grid_events.StoreUnavailable:
        raise err(503, "store_unavailable",
                  "the tenant store is unavailable — your change was not saved")

    # What actually landed, read back from the store rather than echoed from the request: a
    # refused key or a truncated value must not be reported as accepted.
    stored = (grid_events.table_workspace(_ctx_for(session, pool), allowed_pids=None)
              .get("overlays") or {}).get(str(pid)) or {}
    accepted = {k: stored.get(k) for k in updates if k in stored}
    refused = sorted(k for k in updates if k not in accepted or stored.get(k) != str(updates[k]))
    # No cache to patch: the overlay stratum is re-read from the store on every `_payload`, so
    # read-your-writes within this runtime is a property of the design rather than of a
    # write-through step somebody has to remember. (It was a write-through step while the whole
    # payload was cached on a scope key — the arrangement that leaked one user's notes to
    # another. See `_pool_rows`.) Cross-RUNTIME coherence is still not claimed: the module
    # docstring says why, and Postgres is the fix.
    out = {"ok": True, "pid": pid, "updates": accepted}
    if refused:
        out["refused"] = refused
    return out