AI Flow

#!/usr/bin/env python3
"""Google Antigravity usage, read locally at no quota cost. Prints one JSON object:
  quota:  [{label, pct, resetsAt}] from the running agy hub's RetrieveUserQuotaSummary
          (pct = percent of the window used; labels week, 5h, 3p week, 3p 5h)
  tokens: cumulative model-call counters summed over every conversation in
          ~/.gemini/antigravity/conversations/*.db (gen_metadata field 1.4:
          .2 fresh input, .3 output, .5 cached input), plus per-model totals.
Counters only grow, so the difference between two snapshots is what ran in between; byConversation
lets a report keep only the run's own conversation when other Antigravity work shares the machine.
Undocumented formats (decoded 2026-10-16/17); anything unreadable is skipped, never fatal."""
import glob, json, os, re, sqlite3, urllib.request


def varint(b, i):
    r = s = 0
    while True:
        c = b[i]; i += 1; r |= (c & 0x7F) << s; s += 7
        if c < 0x80:
            return r, i


def fields(b):
    i, out = 0, []
    while i < len(b):
        k, i = varint(b, i); f, t = k >> 3, k & 7
        if t == 0: v, i = varint(b, i)
        elif t == 2: n, i = varint(b, i); v = b[i:i + n]; i += n
        elif t == 1: v = b[i:i + 8]; i += 8
        elif t == 5: v = b[i:i + 4]; i += 4
        else: break
        out.append((f, t, v))
    return out


def tokens():
    tot = {"calls": 0, "freshIn": 0, "cachedIn": 0, "out": 0, "byModel": {}, "byConversation": {}}
    for db in glob.glob(os.path.expanduser("~/.gemini/antigravity/conversations/*.db")):
        conv = tot["byConversation"].setdefault(os.path.basename(db)[:-3], {"calls": 0, "freshIn": 0, "cachedIn": 0, "out": 0})
        try:
            con = sqlite3.connect(f"file:{db}?mode=ro", uri=True, timeout=2)
            rows = con.execute("select data from gen_metadata").fetchall()
            con.close()
        except Exception:
            continue
        for (blob,) in rows:
            if not isinstance(blob, (bytes, bytearray)):
                continue
            try:
                top = fields(blob)
                for f, t, v in top:
                    if f != 1 or t != 2: continue
                    sub = fields(v)
                    model = next((x.decode(errors="ignore") for f2, t2, x in sub if f2 == 19 and t2 == 2), "")
                    for f2, t2, x in sub:
                        if f2 == 4 and t2 == 2:
                            d = {a: c for a, b2, c in fields(x) if b2 == 0}
                            tot["calls"] += 1; tot["freshIn"] += d.get(2, 0); tot["out"] += d.get(3, 0); tot["cachedIn"] += d.get(5, 0)
                            conv["calls"] += 1; conv["freshIn"] += d.get(2, 0); conv["out"] += d.get(3, 0); conv["cachedIn"] += d.get(5, 0)
                            m = tot["byModel"].setdefault(model or "unknown", {"calls": 0, "freshIn": 0, "cachedIn": 0, "out": 0})
                            m["calls"] += 1; m["freshIn"] += d.get(2, 0); m["out"] += d.get(3, 0); m["cachedIn"] += d.get(5, 0)
            except Exception:
                continue
    return tot


def hub():
    for p in glob.glob("/proc/[0-9]*/cmdline"):
        try:
            a = open(p, "rb").read().split(b"\0")
        except Exception:
            continue
        s = b" ".join(a).decode(errors="ignore")
        if "--hub" in s and "agy" in s:
            port = re.search(r"--hub-port=(\d+)", s); tok = re.search(r"--csrf_token=(\S+)", s)
            if port and tok:
                return port.group(1), tok.group(1)
    return None


def quota():
    h = hub()
    if not h:
        return []
    port, tok = h
    req = urllib.request.Request(f"http://127.0.0.1:{port}/exa.language_server_pb.LanguageServerService/RetrieveUserQuotaSummary",
                                 data=b"{}", headers={"content-type": "application/json", "x-codeium-csrf-token": tok})
    try:
        r = json.load(urllib.request.urlopen(req, timeout=8)).get("response", {})
    except Exception:
        return []
    out = []
    for g in r.get("groups", []):
        third = "Claude" in g.get("displayName", "")
        for b in g.get("buckets", []):
            w = {"weekly": "week"}.get(b.get("window"), b.get("window", "limit"))
            out.append({"label": ("3p " if third else "") + w, "pct": round((1 - float(b.get("remainingFraction", 1))) * 100, 4), "resetsAt": b.get("resetTime")})
    return out


if __name__ == "__main__":
    print(json.dumps({"quota": quota(), "tokens": tokens()}))