api/routes_platform_admin.py

"""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("<raw key>")`. The runtime is what applies the store
NAMESPACE (`t/<slug>/…`) 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)"}