metrics: break out the co-op's own (operator) spend in the daily series

metrics_daily.spend_usd is the co-op-wide total and deliberately spans every
team, operators included — the numbers were already right. What was missing
was a way to tell how much of that total is the co-op's OWN usage, so a
member-facing chart could not be misread as member-only activity.

Adds metrics_daily.operator_spend_usd: the per-day portion of spend_usd that
belongs to the accounts in OPERATOR_EMAILS (teams aliased member:<email>, like
anyone else). NULL = not measured (never a confident 0), mirroring spend_usd.

- metrics_snapshot.py: operator_spend_by_day() + write it on snapshot/backfill
- app/main.py: expose it in metrics_series (so /broker/metrics and
  /admin/metrics/series carry it), and add operator_spend /
  operator_all_time_spend to /admin/overview totals as the explicit breakout
  of total_spend / total_all_time_spend.

spend_usd and every existing column are untouched.

app/main.py

@@ -694,6 +694,7 @@ def get_db() -> sqlite3.Connection:
         "newsletter_subs INTEGER, "
         "matrix_space_members INTEGER, "
         "spend_usd REAL, "
+        "operator_spend_usd REAL, "
         "requests INTEGER, "
         "out_tokens INTEGER, "
         "energy_wh REAL, "
@@ -709,6 +710,11 @@ def get_db() -> sqlite3.Connection:
     # measured-vs-estimated provenance columns after those. All additive: existing
     # rows keep NULL, and a NULL split reports as unknown — never as "measured".
     for _col, _decl in (
+        # The co-op's OWN accounts' share of spend_usd (OPERATOR_EMAILS). The
+        # total keeps spanning every team; this is the breakout, so a
+        # member-facing surface can say how much of it is the co-op's own
+        # usage. NULL = not measured (never a confident 0).
+        ("operator_spend_usd", "REAL"),
         ("out_tokens", "INTEGER"),
         ("energy_wh", "REAL"),
         ("co2e_g", "REAL"),
@@ -3097,6 +3103,8 @@ async def admin_overview(request: Request):
     admins = []
     total_spend = 0.0
     total_all_time = 0.0
+    operator_spend = 0.0
+    operator_all_time = 0.0
     for email, active, balance, user_id, org_slug, matrix_mxid in rows:
         team = teams.get(f"member:{email}", {})
         spend = float(team.get("spend") or 0.0)
@@ -3140,6 +3148,8 @@ async def admin_overview(request: Request):
             admins.append(entry)
             total_spend += spend
             total_all_time += member_all_time
+            operator_spend += spend
+            operator_all_time += member_all_time
         else:
             members.append(entry)
             total_spend += spend
@@ -3177,8 +3187,13 @@ async def admin_overview(request: Request):
         "totals": {
             "members": len(members),
             "active": sum(1 for m in members if m["active"]),
+            # total_spend spans members AND the co-op's own operator accounts.
+            # operator_spend is the breakout of that portion, so a consumer
+            # never has to guess whether a figure includes the co-op's own use.
             "total_spend": round(total_spend, 2),
             "total_all_time_spend": round(total_all_time, 2),
+            "operator_spend": round(operator_spend, 2),
+            "operator_all_time_spend": round(operator_all_time, 2),
         },
     }
 
@@ -3232,7 +3247,7 @@ def metrics_series(days: int = 90) -> list[dict]:
     try:
         rows = conn.execute(
             "SELECT day, members_active, members_total, members_new, newsletter_subs, "
-            "matrix_space_members, spend_usd, requests, out_tokens, energy_wh, co2e_g, "
+            "matrix_space_members, spend_usd, operator_spend_usd, requests, out_tokens, energy_wh, co2e_g, "
             "tokens_measured, tokens_estimated, factor_revision, recorded_at "
             "FROM metrics_daily ORDER BY day DESC LIMIT ?",
             (limit,),
@@ -3247,6 +3262,7 @@ def metrics_series(days: int = 90) -> list[dict]:
         "newsletter_subs",
         "matrix_space_members",
         "spend_usd",
+        "operator_spend_usd",
         "requests",
         "out_tokens",
         "energy_wh",

metrics_snapshot.py

@@ -57,6 +57,7 @@ def connect() -> sqlite3.Connection:
                newsletter_subs INTEGER,
                matrix_space_members INTEGER,
                spend_usd REAL,
+               operator_spend_usd REAL,
                requests INTEGER,
                out_tokens INTEGER,
                energy_wh REAL,
@@ -68,6 +69,12 @@ def connect() -> sqlite3.Connection:
     )
     # CREATE IF NOT EXISTS never adds columns to an existing table, so migrate.
     for col, decl in (
+        # Per-day share of spend_usd that belongs to the co-op's OWN operator
+        # accounts (OPERATOR_EMAILS). spend_usd stays the co-op-wide total
+        # (it deliberately spans every team, operators included) — this column
+        # is the breakout of it, so member-facing surfaces can say how much of
+        # the co-op total is the co-op's own usage. NULL = not measured.
+        ("operator_spend_usd", "REAL"),
         ("out_tokens", "INTEGER"),
         ("energy_wh", "REAL"),
         ("co2e_g", "REAL"),
@@ -312,6 +319,61 @@ def spend_by_day(since: dt.date | None = None) -> dict[str, tuple[float, int]]:
     return out
 
 
+def operator_spend_by_day(since: dt.date | None = None) -> dict[str, float] | None:
+    """{day: operator_spend_usd} — the co-op's OWN accounts' share of spend.
+
+    spend_usd (spend_by_day) is the co-op-wide total and deliberately spans
+    every team, operators included. This is the breakout: the portion of it
+    that belongs to the operator accounts named by OPERATOR_EMAILS, whose
+    teams are aliased `member:<email>` like anyone else. The member dashboard
+    reads this through /broker/metrics so a member can see how much of the
+    co-op total is the co-op's own usage.
+
+    Returns None when the LiteLLM DB could not be read (the caller then leaves
+    the column NULL — "not measured", never a confident 0), mirroring
+    spend_by_day's fail-closed behaviour.
+    """
+    try:
+        import psycopg2
+    except Exception as exc:  # noqa: BLE001
+        print(f"metrics: psycopg2 unavailable ({exc})")
+        return None
+    aliases = [f"member:{e}" for e in OPERATOR_EMAILS]
+    if not aliases:
+        return {}
+    try:
+        conn = psycopg2.connect(
+            host=os.environ.get("LITELLM_DB_HOST"),
+            port=os.environ.get("LITELLM_DB_PORT"),
+            user=os.environ.get("LITELLM_DB_USER"),
+            password=os.environ.get("LITELLM_DB_PASSWORD"),
+            dbname=os.environ.get("LITELLM_DB_NAME"),
+            connect_timeout=10,
+            sslmode="prefer",
+        )
+    except Exception as exc:  # noqa: BLE001
+        print(f"metrics: LiteLLM DB connect failed ({exc}) — operator spend left NULL")
+        return None
+    try:
+        where, params = "", [aliases]
+        if since:
+            where, params = 'AND s."startTime" >= %s', [aliases, since.isoformat()]
+        with conn.cursor() as cur:
+            cur.execute(
+                f"""SELECT to_char(s."startTime", 'YYYY-MM-DD') AS day,
+                           COALESCE(SUM(s.spend), 0)
+                    FROM "LiteLLM_SpendLogs" s
+                    JOIN "LiteLLM_TeamTable" t ON t.team_id = s."team_id"
+                    WHERE t.team_alias = ANY(%s) {where}
+                    GROUP BY 1 ORDER BY 1""",
+                params,
+            )
+            out = {row[0]: round(float(row[1]), 6) for row in cur.fetchall()}
+    finally:
+        conn.close()
+    return out
+
+
 def tokens_by_day_model(since: dt.date | None = None) -> dict[str, dict[str, int]]:
     """{day: {model: output_tokens}} from LiteLLM's spend logs.
 
@@ -459,9 +521,10 @@ def upsert(day: str, row: dict) -> None:
         conn.execute(
             """INSERT INTO metrics_daily
                    (day, members_active, members_total, members_new, newsletter_subs,
-                    matrix_space_members, spend_usd, requests, out_tokens, energy_wh,
+                    matrix_space_members, spend_usd, operator_spend_usd, requests,
+                    out_tokens, energy_wh,
                     co2e_g, tokens_measured, tokens_estimated, factor_revision, recorded_at)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(day) DO UPDATE SET
                    members_active=excluded.members_active,
                    members_total=excluded.members_total,
@@ -469,6 +532,7 @@ def upsert(day: str, row: dict) -> None:
                    newsletter_subs=excluded.newsletter_subs,
                    matrix_space_members=excluded.matrix_space_members,
                    spend_usd=excluded.spend_usd,
+                   operator_spend_usd=excluded.operator_spend_usd,
                    requests=excluded.requests,
                    out_tokens=excluded.out_tokens,
                    energy_wh=excluded.energy_wh,
@@ -485,6 +549,7 @@ def upsert(day: str, row: dict) -> None:
                 row.get("newsletter_subs"),
                 row.get("matrix_space_members"),
                 row.get("spend_usd"),
+                row.get("operator_spend_usd"),
                 row.get("requests"),
                 row.get("out_tokens"),
                 row.get("energy_wh"),
@@ -509,6 +574,15 @@ def snapshot(day: str | None = None) -> dict:
     since = dt.date.fromisoformat(day) - dt.timedelta(days=1)
     spend = spend_by_day(since=since)
     today_spend = spend.get(day, (None, None))
+    op_spend = operator_spend_by_day(since=since)
+    # The co-op's own share of today's spend. If the LiteLLM DB could not be
+    # read, both spend and operator_spend are NULL (not measured). If it was
+    # read and operators simply used nothing, that is a measured 0.0 — but only
+    # when there WAS spend to attribute that day.
+    if op_spend is None or today_spend[0] is None:
+        op_today = None
+    else:
+        op_today = op_spend.get(day, 0.0)
     day_models = (tokens_by_day_model(since=since).get(day)) or {}
     tot = sum(day_models.values())
     out_tokens = tot or None
@@ -520,6 +594,7 @@ def snapshot(day: str | None = None) -> dict:
         "newsletter_subs": subs,
         "matrix_space_members": space,
         "spend_usd": today_spend[0],
+        "operator_spend_usd": op_today,
         "requests": today_spend[1],
         "out_tokens": out_tokens,
         "energy_wh": impact["wh"],
@@ -545,6 +620,7 @@ def backfill() -> None:
     m = from_sqlite()
     joins = sorted(listmonk_joins() or [])
     spend = spend_by_day()
+    op_spend = operator_spend_by_day()
     tokens = tokens_by_day_model()
     if not joins:
         print("metrics: no membership history (listmonk unavailable) — past days left NULL for members")
@@ -571,6 +647,10 @@ def backfill() -> None:
             "members_total": members_total,
             "members_new": sum(1 for j in joins if j == key),
             "spend_usd": spend.get(key, (None, None))[0],
+            "operator_spend_usd": (
+                None if op_spend is None
+                else (op_spend.get(key, 0.0) if key in spend else None)
+            ),
             "requests": spend.get(key, (None, None))[1],
         }
         day_models = tokens.get(key) or {}