"""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