"""routes_platform_admin.py — THE LOOPABLE ADMIN PLANE (wave 19, owner item 13 / R3+R4). Every other admin surface in this product answers questions about ONE tenant. This one answers questions about the PLATFORM: who our customers are, what they are running, whether their data sources are alive, and what the automation fleet costs. It is the first cross-tenant reader that has ever existed here, which is why it is also the most carefully walled. ⛔ THE WALL. `padmin_gate` is `core.platform_admin.is_platform_admin` — a DOUBLE lock (the record flag AND the `loopable` tenant), fail-closed, applied to every single route in this file through one dependency. A tenant admin is NOT admitted: `role: 'admin'` is Royal's or Nurilab's authority over their own workspace and it must never widen into a view of each other. `verify_api.py`'s W19-ADMIN section proves that by ENUMERATING this router and having a tenant admin try every path it declares, so a route added later cannot quietly ship without the wall. ⛔ CROSS-TENANT READS GO THROUGH EACH TENANT'S OWN RUNTIME. `runtime.get_runtime(slug)` per tenant, then `rt.get(...)` — never `core.store.get("")`. The runtime is what applies the store NAMESPACE (`t//…`) or binds the tenant's OWN dataset repo (R2), so reading raw keys would silently return tenant #0's data labelled as somebody else's — a wrong answer that looks right, which is the worst failure this plane could have. HONEST DEGRADATION IS THE WHOLE DESIGN, NOT AN ERROR PATH. This plane reads six subsystems across N tenants; on any given day one of them can be unreachable (a suspended tenant record, a locked keychain, an HF repo hiccup, no AWS credentials in this container). A 500 would take the entire dashboard down because one cell could not be filled. So every collector catches, and every row can carry an `error` string that the pane RENDERS — "unknown" is a real answer and it is never rendered as a zero. ([[gate-can-report-green-on-nothing]]: a fabricated 0 and a true 0 must not look alike.) EVERY COUNT DRILLS TO ROWS. `/overview`'s per-tenant counts are computed by the SAME collector functions the `/users`, `/databases`, `/connectors` and `/automations` routes serve rows from, so a number and its drill-down cannot disagree — they are one computation, projected twice ([[no-unverifiable-aggregates]]). """ import json import os import subprocess import sys import time from pathlib import Path from fastapi import APIRouter, Depends import core.platform_admin as platform_admin import core.store as store from deps import Session, err, require_session, users router = APIRouter(prefix="/api/v1/platform-admin") def _now_iso(): """UTC, offset-bearing — the one stamp format anything client-side may subtract from.""" import datetime as _dt return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds") def padmin_gate(session: Session = Depends(require_session)) -> Session: """401 without a session, 403 unless this account is a Loopable platform operator. ONE dependency for the whole router. A per-route check is a per-route chance to forget, and the thing being forgotten here would be every customer's data at once. The message deliberately does not confirm that a platform plane exists for somebody else — a tenant admin who pokes at this URL learns only that their account cannot open it. """ if not platform_admin.is_platform_admin(session.user): raise err(403, "forbidden", "your account does not have access to this surface") return session # ══════════════════════════════════════════════════════════════════════════ tenants def _tenant_bucket(): """The control-plane `tenants` records, or {} — a platform fact, in the DEFAULT store.""" from harness import runtime try: recs = store.get(runtime.TENANTS_KEY) or {} return {str(k).strip().lower(): v for k, v in recs.items() if isinstance(v, dict)} except Exception: return {} def _slugs(): """Every tenant this deployment knows: compiled builders + control-plane records. `runtime.known_tenants()` is the union and it is the same list login resolves against, so this plane cannot show a customer the door does not recognise (or miss one it does). """ from harness import runtime try: return list(runtime.known_tenants()) except Exception: return sorted(_tenant_bucket()) def _runtime_for(slug): """(runtime, error). A tenant whose runtime will not build is a ROW WITH A NOTE, never a 500. Two real cases produce one: a SUSPENDED record (`get_runtime` raises KeyError by design — "which tenants exist" is not a question the login path answers), and tenant #0's builder, which makes a LIVE Odoo call (`harness.tenants.royal_imports` → `excluded_customer_ids`) and therefore fails on a box that cannot reach the ERP. Neither may take the dashboard down. """ from harness import runtime try: return runtime.get_runtime(slug), "" except KeyError: return None, "not resolvable (suspended, or no builder and no active record)" except Exception as e: # noqa: BLE001 return None, f"runtime unavailable ({type(e).__name__})" def _scan(want=""): """[(row, runtime)] — every tenant (or one), with its runtime RESOLVED EXACTLY ONCE. The one-resolution rule is not tidiness. `get_runtime` is LRU-cached on success, but a tenant that FAILS to build is not cached, and tenant #0's builder makes a live Odoo call — so a route that asked twice would pay the timeout twice, and `/overview` (which asks about six subsystems) would pay it six times. Resolve once, pass the handle down. """ bucket = _tenant_bucket() want = str(want or "").strip().lower() out = [] for slug in _slugs(): if want and slug != want: continue rec = bucket.get(slug) or {} rt, error = _runtime_for(slug) out.append(({ "slug": slug, # The record's name, else the built runtime's, else the slug. Tenant #0 is compiled # and has no record, so without the runtime fallback it would render as "royal-imports". "name": str(rec.get("name") or (getattr(rt, "name", "") if rt else "") or slug), "source": "record" if rec else "compiled", "status": str(rec.get("status") or ("active" if rt else "unknown")), "domains": list(rec.get("domains") or []), "modules": rec.get("modules", "all" if not rec else []), # R2: the isolation shape. "own repo" and "shared repo + prefix" are genuinely # different blast radii and an operator should be able to see which is which. "storeRepo": rec.get("store_repo") or ("shared (tenant #0 repo)" if not rec else ""), "storePrefix": getattr(rt, "store_namespace", "") if rt else "", "error": error, }, rt)) return out # ══════════════════════════════════════════════════════════════════════════ users def _user_rows(tenant=None): """Accounts across every tenant, from the GLOBAL registry (`users.json` is control-plane). ⛔ NEVER `salt` OR `hash`. This projection is the only one these routes use, mirroring `routes_admin._view`'s discipline: one function that can leak, and it does not. R4's two new fields ride here — `lastLogin` / `lastActive`, absent on every pre-wave record, rendered as "never" rather than as a fabricated date. """ try: reg = users.registry() or {} except Exception: return [] want = str(tenant or "").strip().lower() rows = [] for uname, rec in sorted(reg.items()): if not isinstance(rec, dict): continue slug = str(rec.get("tenant") or "royal-imports").strip().lower() if want and slug != want: continue rows.append({ "username": uname, "name": rec.get("name") or uname, "email": rec.get("email") or "", "tenant": slug, "role": rec.get("role", "user"), "active": bool(rec.get("active", True)), "lastLogin": rec.get("last_login") or "", "lastActive": rec.get("last_active") or "", "platformAdmin": rec.get("platform_admin") is True, }) return rows # ══════════════════════════════════════════════════════════════════════════ databases def _database_rows(slug, rt): """A tenant's user-created databases (`ut_*`) with their row counts. Via `core.user_tables.all_tables(st=rt)` — the TenantRuntime, never the module-global, which the engine's own header (`automation_engine.py:49-52`) flags as a cross-tenant defect for every R2 tenant. That booked defect is exactly the mistake a cross-tenant reader would make most easily, so it is stated at the call site too. """ try: import core.user_tables as ut tables = ut.all_tables(st=rt) or {} except Exception as e: # noqa: BLE001 return [], f"databases unreadable ({type(e).__name__})" rows = [] for key, t in sorted(tables.items(), key=lambda kv: (kv[1].get("label") or "").lower()): if not isinstance(t, dict): continue rows.append({ "tenant": slug, "key": key, "label": t.get("label") or key, "source": t.get("source") or "Blank", "createdBy": t.get("createdBy") or "", "created": t.get("created") or "", "fields": len(t.get("fields") or []), "rowCount": len(t.get("rows") or {}), }) return rows, "" # ══════════════════════════════════════════════════════════════════════════ connectors def _connector_rows(slug, rt): """A tenant's data sources and which one is actually RESOLVED (i.e. would serve a query). `_resolved_odoo_key` is imported from `routes_keychain` rather than re-derived. It is the route-level mirror of `TenantRuntime.odoo_source()`, it takes a runtime (not a session), and a second copy of that resolution here would be a copy that drifts — at which point this plane would confidently name the wrong live connector. Reused, not restated. """ try: import core.keychain as keychain from routes_keychain import _CONNECTOR_FLAGS_KEY, _resolved_odoo_key entries = keychain.list_entries(rt) locked = not keychain.unlocked() resolved, _flag = _resolved_odoo_key(rt) flags = rt.get(_CONNECTOR_FLAGS_KEY) or {} except Exception as e: # noqa: BLE001 return [], f"connectors unreadable ({type(e).__name__})", False rows = [] if slug == "royal-imports" and os.environ.get("ODOO_URL"): rows.append({"tenant": slug, "key": "odoo-env", "label": "Odoo (environment)", "type": "odoo", "source": "env", "active": resolved == "env", "paused": bool((flags.get("odoo-env") or {}).get("paused"))}) for e in entries: rows.append({"tenant": slug, "key": e["id"], "label": e["label"], "type": e["type"], "source": "keychain", "active": e["type"] == "odoo" and resolved == f"keychain:{e['id']}", "paused": bool((flags.get(e["id"]) or {}).get("paused"))}) return rows, "", locked # ══════════════════════════════════════════════════════════════════════════ automations + cost #: THE UNIT MODEL, from `ops/provision_automation_cron.py` (its cost note at :25-28 and the #: function it actually provisions at :211 / :269). Restated as constants rather than as prose so #: the arithmetic below is re-checkable against the thing that was really deployed: #: EventBridge Scheduler `rate(15 minutes)` → a 128 MB, 30 s-timeout Lambda that POSTs the tick. TICK_CADENCE = "rate(15 minutes)" TICK_PER_DAY = 96 # 1440 / 15 LAMBDA_MB = 128 FREE_LAMBDA_REQUESTS = 1_000_000 # AWS always-free, per month FREE_SCHEDULER_INVOCATIONS = 14_000_000 DAYS_PER_MONTH = 30.4 def _runs_per_day(cron): """Scheduled runs/day for the five-field crons this product offers, or None. ⚠ DELIBERATELY NARROW. It answers exactly the shapes `automation_engine.CRON_PRESETS` can produce (every-N-minutes, hourly, daily, weekly, monthly) and returns None for anything else — an unparsed cadence is reported as "custom", never as a guessed number that would then be multiplied into a cost. A wrong denominator is worse than an absent one. ⚠ AND IT IS NOT MEASURED FROM HISTORY, on purpose: `automation_engine.MAX_RUNS` trims run history to 20 entries, so a 15-minute automation retains ~5 hours of it. Deriving runs/day from that window would understate by ~5x. History is reported as what it is — the last N runs. """ parts = str(cron or "").split() if len(parts) != 5: return None minute, hour, dom, _mon, dow = parts if minute.startswith("*/") and hour == "*": try: step = int(minute[2:]) except ValueError: return None return (1440.0 / step) if step > 0 else None if minute.isdigit() and hour == "*": return 24.0 if minute.isdigit() and hour.isdigit(): if dom.isdigit(): return 1.0 / DAYS_PER_MONTH # monthly if dow.isdigit(): return 1.0 / 7.0 # weekly return 1.0 # daily return None def _automation_rows(slug, rt): """A tenant's automations, their schedules, and their retained run history.""" try: import automation_engine as engine defs = engine.all_definitions(rt) or {} max_runs = int(getattr(engine, "MAX_RUNS", 20)) except Exception as e: # noqa: BLE001 return [], f"automations unreadable ({type(e).__name__})", 20 rows = [] for auto_id, d in sorted(defs.items(), key=lambda kv: (kv[1].get("name") or "").lower()): if not isinstance(d, dict): continue sched = d.get("schedule") or {} status = d.get("status") or {} runs = list(d.get("runs") or []) cron = sched.get("cron") or "" per_day = _runs_per_day(cron) if sched.get("enabled") else 0.0 rows.append({ "tenant": slug, "id": auto_id, "name": d.get("name") or auto_id, "kind": d.get("kind") or "", "enabled": bool(sched.get("enabled")), "cron": cron, "runsPerDay": round(per_day, 2) if per_day is not None else None, "state": status.get("state") or "idle", "lastRunAt": status.get("lastRunAt") or "", "lastSummary": status.get("lastSummary") or "", # The retained window, named as such — see `_runs_per_day`'s second warning. "runsRetained": len(runs), "failedRetained": sum(1 for r in runs if isinstance(r, dict) and not r.get("ok")), "createdBy": d.get("createdBy") or "", "created": d.get("created") or "", }) return rows, "", max_runs def _automation_cost(auto_rows): """The estimated monthly cost of running the automation fleet — and the honest shape of it. THE POINT THIS BLOCK EXISTS TO MAKE, which is not intuitive: **the external cron does not scale with automations or tenants.** One EventBridge schedule fires one Lambda, which POSTs one tick, and that tick runs every DUE automation for every tenant (`provision_automation_cron` states this as the reason it stays $0 "at ten tenants"). So the AWS bill is a function of the CADENCE alone — the fleet below adds work inside the API container, which is already paid for. WHAT IS NOT COMPUTED HERE, deliberately: Lambda GB-seconds. That needs the real average duration of the function, which only CloudWatch knows; assuming one would be inventing the larger half of the free-tier calculation. `/aws` reports the measured figure when credentials are available, and this block says so rather than filling the gap with a plausible number. """ invocations = TICK_PER_DAY * DAYS_PER_MONTH fleet_runs = sum(r["runsPerDay"] or 0 for r in auto_rows if r.get("enabled")) unknown_cadence = sum(1 for r in auto_rows if r.get("enabled") and r.get("runsPerDay") is None) return { "cadence": TICK_CADENCE, "invocationsPerMonth": int(round(invocations)), "freeRequestsPct": round(100.0 * invocations / FREE_LAMBDA_REQUESTS, 3), "freeSchedulerPct": round(100.0 * invocations / FREE_SCHEDULER_INVOCATIONS, 4), "lambdaMb": LAMBDA_MB, "usd": 0.0, "fleetRunsPerDay": round(fleet_runs, 2), "unknownCadence": unknown_cadence, "basis": ( f"One EventBridge schedule ({TICK_CADENCE}) fires one {LAMBDA_MB} MB Lambda that " f"POSTs the tick; that ONE tick runs every due automation for every tenant, so the " f"AWS cost is set by the cadence and does not grow with the fleet. " f"{int(round(invocations)):,} invocations/month is " f"{round(100.0 * invocations / FREE_LAMBDA_REQUESTS, 3)}% of the 1,000,000-request " f"always-free tier, so the marginal cost of the automations below is $0.00. " f"Compute (GB-seconds) depends on measured durations and is NOT estimated here — " f"the AWS report reads the real figure when credentials are available." ), } # ══════════════════════════════════════════════════════════════════════════ the AWS report #: `ops/aws_usage_report.py`, relative to this file: aios-web/api/ -> repo root -> ops/. _AWS_REPORT = Path(__file__).resolve().parents[2] / "ops" / "aws_usage_report.py" _AWS_TIMEOUT = 60 def _aws_report(days=7): """The AWS usage report as text, or an honest block saying why there is none. ⛔ SUBPROCESS, NEVER `import`. Two concrete reasons, both found by reading that file rather than by running it: * it reassigns `sys.stdout` to a UTF-8 wrapper AT MODULE IMPORT (a cp1252 fix for this box) — importing it would mutate the API process's stdout as a side effect; * its `main()` calls `argparse.parse_args()` with no argv, so inside a server it would parse UVICORN's arguments and can `sys.exit(2)` — and `SystemExit` is a `BaseException`, which `except Exception` does not catch. A route that "cannot crash" would crash. ⚠ ON THE SPACE THIS IS THE NORMAL PATH, NOT THE ERROR PATH. `deploy_web.py` ships `api/*.py` and `web/`; `ops/` is not in the image, and AWS credentials live in a local `.env` that is never deployed. So the honest block below is what the deployed plane shows, and it is written to be read by an operator as information ("run it here") rather than as a fault. """ if not _AWS_REPORT.is_file(): return {"available": False, "text": "", "note": ("The AWS usage report is not part of this deployment — `ops/` ships " "with the repository, not with the container image. Run " "`python ops/aws_usage_report.py` locally for the live figures.")} try: proc = subprocess.run( [sys.executable, str(_AWS_REPORT), "--days", str(int(days))], capture_output=True, text=True, timeout=_AWS_TIMEOUT, cwd=str(_AWS_REPORT.parent.parent)) except subprocess.TimeoutExpired: return {"available": False, "text": "", "note": (f"The AWS report did not answer within {_AWS_TIMEOUT}s — CloudWatch may " f"be unreachable from here. Nothing was assumed about usage.")} except Exception as e: # noqa: BLE001 return {"available": False, "text": "", "note": f"The AWS report could not be run here ({type(e).__name__})."} out = (proc.stdout or "").strip() if proc.returncode != 0 or not out: detail = (proc.stderr or "").strip().splitlines() return {"available": False, "text": out, "note": ("AWS credentials are not available here, or boto3 is not installed — " "no usage could be read, and none is guessed. " + (detail[-1][:200] if detail else ""))} return {"available": True, "text": out, "note": ""} # ══════════════════════════════════════════════════════════════════════════ routes @router.get("/overview") def overview(session: Session = Depends(padmin_gate)): """THE PLANE'S ONE TABLE: every tenant, with the counts that drill to the rows below. Each count is produced by the same collector its `/…` route serves, so the number and its drill-down are one computation projected twice. A subsystem that cannot be read contributes an `errors` entry on that tenant's row and a NULL count — never a zero, which would read as "this customer has no databases" when the truth is "we could not look". """ t0 = time.time() all_users = _user_rows() users_by_tenant = {} for u in all_users: users_by_tenant.setdefault(u["tenant"], []).append(u) rows = [] for t, rt in _scan(): slug = t["slug"] row = dict(t) row["users"] = len(users_by_tenant.get(slug, [])) row["admins"] = sum(1 for u in users_by_tenant.get(slug, []) if u["role"] == "admin") errors = [t["error"]] if t["error"] else [] if rt is None: # An unresolvable tenant still shows its ACCOUNTS (they live in the global registry, # which is readable regardless) — everything tenant-store-shaped is honestly unknown. row.update({"databases": None, "rows": None, "connectors": None, "automations": None, "keychainLocked": None, "errors": errors}) rows.append(row) continue dbs, db_err = _database_rows(slug, rt) conns, conn_err, locked = _connector_rows(slug, rt) autos, auto_err, _mr = _automation_rows(slug, rt) errors += [e for e in (db_err, conn_err, auto_err) if e] row.update({ "databases": None if db_err else len(dbs), "rows": None if db_err else sum(d["rowCount"] for d in dbs), "connectors": None if conn_err else len(conns), "connectorsPaused": None if conn_err else sum(1 for c in conns if c["paused"]), "automations": None if auto_err else len(autos), "automationsEnabled": None if auto_err else sum(1 for a in autos if a["enabled"]), "keychainLocked": None if conn_err else locked, "errors": errors, }) rows.append(row) return { "tenants": rows, "totals": { "tenants": len(rows), # EVERY account, not the sum of the rows: an account whose `tenant` names a slug this # deployment no longer knows belongs in the platform total and would vanish from a # per-row sum. `orphanUsers` names that gap instead of hiding it. "users": len(all_users), "orphanUsers": len(all_users) - sum(r["users"] for r in rows), # Sums SKIP unknowns rather than treating them as 0, and say how many were skipped — # a total that silently absorbs an unreadable tenant is a fabricated total. "databases": sum(r["databases"] or 0 for r in rows), "rows": sum(r["rows"] or 0 for r in rows), "automations": sum(r["automations"] or 0 for r in rows), "unknownTenants": sum(1 for r in rows if r["databases"] is None), }, "storeAvailable": bool(_store_ok()), # ⚠ OFFSET-BEARING, and that is not pedantry. The client renders this as "read 2 minutes # ago" via `Date.parse`, which reads a NAIVE stamp as browser-local — so a UTC container # and a US viewer would turn "just now" into "5 hours ago", or into a future date that # renders as "just now" forever. The stamps written by `core.users` carry an offset for # the same reason; anything a browser subtracts from `Date.now()` must say what zone it # is in. `verify_api` asserts the offset so this cannot regress to `strftime`. "generatedAt": _now_iso(), "tookMs": int((time.time() - t0) * 1000), } def _store_ok(): try: return store.available() except Exception: return False @router.get("/users") def platform_users(tenant: str = "", session: Session = Depends(padmin_gate)): """Every account on the platform, or one tenant's — the drill behind the Users count. R4's stamps are the columns that did not exist before this wave: `lastLogin` is written by `routes_auth.login`, `lastActive` by `deps.require_session` (throttled to once an hour per account per process). Absent means never seen, and the pane renders it as "never". """ rows = _user_rows(tenant) return {"users": rows, "count": len(rows), "stampsNote": ("Login and activity stamps started with this release — accounts that " "have not signed in since show no date rather than an invented one. " "Activity is recorded at most once an hour per account.")} @router.get("/databases") def platform_databases(tenant: str = "", session: Session = Depends(padmin_gate)): """Every tenant's user-created databases and row counts — the drill behind Databases/Rows.""" out, errors = [], {} for t, rt in _scan(tenant): slug = t["slug"] if rt is None: errors[slug] = t["error"] continue rows, err = _database_rows(slug, rt) if err: errors[slug] = err continue out += rows return {"databases": out, "count": len(out), "rows": sum(d["rowCount"] for d in out), "errors": errors} @router.get("/connectors") def platform_connectors(tenant: str = "", session: Session = Depends(padmin_gate)): """Every tenant's data sources, which one is live, and which are paused. METADATA ONLY — `keychain.list_entries` never decrypts, so no credential, and no masked preview either: a platform operator needs to know a source EXISTS and whether it serves, not what the secret looks like. The tenant's own admin surface is where previews belong. """ out, errors, locked_any = [], {}, {} for t, rt in _scan(tenant): slug = t["slug"] if rt is None: errors[slug] = t["error"] continue rows, err, locked = _connector_rows(slug, rt) if err: errors[slug] = err continue locked_any[slug] = locked out += rows return {"connectors": out, "count": len(out), "keychainLocked": locked_any, "errors": errors, "note": ("A locked keychain means this container has no `AIOS_KEYCHAIN_KEY` — stored " "credentials cannot be read, so a tenant's sources fail closed rather than " "falling back to anyone else's.")} @router.get("/automations") def platform_automations(tenant: str = "", session: Session = Depends(padmin_gate)): """Every tenant's automations, their schedules and run history, plus the fleet cost model.""" out, errors, retained = [], {}, 20 for t, rt in _scan(tenant): slug = t["slug"] if rt is None: errors[slug] = t["error"] continue rows, err, max_runs = _automation_rows(slug, rt) if err: errors[slug] = err continue retained = max_runs out += rows return {"automations": out, "count": len(out), "enabled": sum(1 for a in out if a["enabled"]), "historyRetained": retained, "cost": _automation_cost(out), "errors": errors, "tickEnabled": os.environ.get("AIOS_AUTOMATIONS") == "1"} @router.get("/aws") def platform_aws(days: int = 7, session: Session = Depends(padmin_gate)): """The AWS cron's usage report, verbatim, or an honest note explaining its absence.""" days = max(1, min(int(days or 7), 90)) return {"days": days, "report": _aws_report(days)} @router.get("/releases") def platform_releases(session: Session = Depends(padmin_gate)): """⭐ WAVE 20 (owner item 12, ruling R6) — WHAT IS RUNNING WHERE, and what else could be. Owner: *"Make this part of the deploy skill. Also ability to revert to any version we want. Make sure App versioning is something that our company admin (loopable, non-tenant) can easily see."* This is the SEEING half; promoting stays a CLI command by R6, so nothing here writes and no web session can move production. ⛔ THE HUB READS LIVE IN `core/releases.py`, NOT HERE. `ops/verify_portability.py` B2 forbids the HuggingFace SDK anywhere under `aios-web/api/` — the API process is host-agnostic by design — and it caught this endpoint's first draft doing exactly that. Same delegation shape as `routes_assets` -> `core/assets.py`. """ import core.releases as releases return {"here": os.environ.get("AIOS_VERSION") or "unknown", "environments": releases.environments(), "releases": releases.history(), # The runbook, in the payload, because the panel is read-only BY DESIGN and a user # looking at it is exactly the person who needs to know how to move a version. "promote": "python aios-web/deploy_web.py --promote=vN (or --promote=staging)"}