diff --git a/bin/omarchy-agent-usage-antigravity b/bin/omarchy-agent-usage-antigravity
new file mode 100755
index 00000000000..f66826a7b2e
--- /dev/null
+++ b/bin/omarchy-agent-usage-antigravity
@@ -0,0 +1,375 @@
+#!/usr/bin/python3
+# omarchy:summary=Print the Antigravity usage record as JSON
+# omarchy:args=[--force] [--limits-only]
+# omarchy:hidden=true
+"""Read Antigravity CLI quotas and local prompt/session history.
+
+agy 1.1.11 introduced read-only /usage JSON output. Let the CLI handle
+credentials, refresh and quota requests; older versions must never receive
+this prompt because they may send it to a model instead. Transcripts do not
+record token counts or model IDs, so those fields are deliberately empty.
+"""
+
+import argparse
+import fcntl
+import hashlib
+import json
+import math
+import os
+import re
+import shutil
+import signal
+import sqlite3
+import subprocess
+import tempfile
+import time
+from collections import Counter, defaultdict
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+SCAN_REUSE_SECONDS = 20
+LIMITS_ONLY_REUSE_SECONDS = 900
+CACHE_VERSION = 3
+AUTH_HELP = "Run `agy` to check your sign-in, then refresh the panel."
+
+
+def app_root():
+ return Path(os.environ.get("AGY_DIR") or Path.home() / ".gemini/antigravity-cli").expanduser().resolve()
+
+
+def cache_root():
+ return Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") / "omarchy/agent-usage"
+
+
+def read_json(path):
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ return data if isinstance(data, dict) else {}
+ except (OSError, ValueError):
+ return {}
+
+
+def write_json(path, payload):
+ path.parent.mkdir(parents=True, exist_ok=True)
+ fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8") as handle:
+ json.dump(payload, handle, separators=(",", ":"), allow_nan=False)
+ handle.write("\n")
+ os.replace(name, path)
+ finally:
+ Path(name).unlink(missing_ok=True)
+
+
+def parse_time(value):
+ try:
+ if isinstance(value, (int, float)) and not isinstance(value, bool):
+ return datetime.fromtimestamp(value / 1000, timezone.utc)
+ parsed = datetime.fromisoformat(str(value or "").replace("Z", "+00:00"))
+ return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+ except (ValueError, TypeError, OverflowError, OSError):
+ return None
+
+
+def json_lines(path):
+ try:
+ with path.open(encoding="utf-8", errors="replace") as handle:
+ for line in handle:
+ try:
+ entry = json.loads(line)
+ if isinstance(entry, dict):
+ yield entry
+ except ValueError:
+ continue
+ except OSError:
+ return
+
+
+def prompt_matches(text, content):
+ if not isinstance(text, str) or not text.strip() or not isinstance(content, str):
+ return False
+ text, content = text.strip(), content.strip()
+ if text == content:
+ return True
+ # Match the complete candidate literally between the wrapper delimiters.
+ # A prompt can itself contain closing-tag text; splitting at the first
+ # closing tag would truncate it. Metadata may quote unrelated prompts.
+ return re.fullmatch(r"\s*" + re.escape(text)
+ + r"\s*(?:\s*.*)?"
+ + r"(?:\s*.*)?",
+ content, re.S) is not None
+
+
+def collect_local_stats(root, now):
+ today = now.astimezone().date().isoformat()
+ days = [(now.astimezone().date() - timedelta(days=i)).isoformat() for i in range(6, -1, -1)]
+ history_prompts = Counter()
+ transcript_prompts = Counter()
+ history_events = defaultdict(list)
+ active_dates = set()
+ sessions = set()
+ today_sessions = set()
+
+ def activity(value, session=None):
+ parsed = parse_time(value)
+ if parsed is None or parsed > now:
+ return None
+ day = parsed.astimezone().date().isoformat()
+ active_dates.add(day)
+ if isinstance(session, str) and session:
+ sessions.add(session)
+ if day == today:
+ today_sessions.add(session)
+ return day
+
+ for entry in json_lines(root / "history.jsonl"):
+ # History includes slash commands; those are not prompts to an agent.
+ display = entry.get("display", entry.get("prompt", ""))
+ if isinstance(display, str) and display.lstrip().startswith("/"):
+ continue
+ day = activity(entry.get("timestamp"), entry.get("conversationId") or entry.get("conversation_id"))
+ if day:
+ history_prompts[day] += 1
+ stamp = parse_time(entry["timestamp"]).timestamp()
+ history_events[int(stamp)].append({"time": stamp, "text": display, "matched": False,
+ "session": entry.get("conversationId") or entry.get("conversation_id")})
+
+ for transcript in (root / "brain").glob("*/.system_generated/logs/transcript.jsonl"):
+ session = transcript.parents[2].name
+ # An interrupted append can leave repeated steps. Count each dated user
+ # input once, without examining or estimating the text's token length.
+ seen = set()
+ for entry in json_lines(transcript):
+ if not isinstance(entry.get("created_at"), str) or not isinstance(entry.get("step_index"), int):
+ continue
+ key = (entry.get("step_index"), entry.get("created_at"))
+ if key in seen:
+ continue
+ seen.add(key)
+ day = activity(entry.get("created_at"), session)
+ if day and entry.get("type") == "USER_INPUT":
+ stamp = parse_time(entry["created_at"]).timestamp()
+ content = entry.get("content")
+ candidates = []
+ # CLI history timestamps are milliseconds; transcript creation is
+ # slightly later and may wrap the prompt in context.
+ # Match one event at a time so distinct prompts from either source
+ # survive, including repeated text and partially retained logs.
+ for second in range(int(stamp) - 1, int(stamp) + 2):
+ for event in history_events.get(second, []):
+ text = event["text"]
+ if (not event["matched"] and abs(event["time"] - stamp) <= 1
+ and (not event["session"] or event["session"] == session)
+ and prompt_matches(text, content)):
+ candidates.append(event)
+ if candidates:
+ min(candidates, key=lambda event: abs(event["time"] - stamp))["matched"] = True
+ else:
+ transcript_prompts[day] += 1
+
+ # Summaries also know about conversations whose transcript is unavailable.
+ database = root / "conversation_summaries.db"
+ if database.exists():
+ try:
+ connection = sqlite3.connect(database.as_uri() + "?mode=ro", uri=True)
+ try:
+ for session, modified in connection.execute("SELECT conversation_id, last_modified_time FROM conversation_summaries"):
+ if session:
+ sessions.add(session)
+ activity(modified, session)
+ finally:
+ connection.close()
+ except sqlite3.Error:
+ pass
+
+ prompts = history_prompts + transcript_prompts
+ return {
+ "todayPrompts": prompts.get(today, 0),
+ "todaySessions": len(today_sessions),
+ "todayTotalTokens": 0,
+ "todayTokensByModel": {},
+ "recentDays": [{"date": day, "messageCount": 0} for day in days],
+ "totalPrompts": sum(prompts.values()),
+ "totalSessions": len(sessions),
+ "activeDays": len(active_dates),
+ "activeDates": sorted(active_dates),
+ "modelUsage": {},
+ }
+
+
+def installed_cli():
+ path = shutil.which("agy")
+ if path:
+ executable = Path(path)
+ # Omarchy's stock wrapper installs agy on first invocation. A passive
+ # usage scan must not install a CLI, including through a mise shim.
+ try:
+ wrapper = b"mise use" in executable.read_bytes()[:1024] if executable.stat().st_size < 8192 else False
+ if executable.resolve().name != "mise" and not wrapper:
+ return str(executable)
+ except OSError:
+ pass
+ mise = shutil.which("mise")
+ if mise:
+ try:
+ found = subprocess.run([mise, "which", "agy"], stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=3)
+ executable = Path(found.stdout.strip())
+ if found.returncode == 0 and executable.is_file() and os.access(executable, os.X_OK):
+ return str(executable)
+ except (OSError, subprocess.TimeoutExpired):
+ pass
+ return ""
+
+
+def run_cli(cli, *args, timeout=20):
+ # A timed-out CLI must not leave descendants holding our output pipes.
+ with subprocess.Popen([cli, *args], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, text=True, start_new_session=True) as process:
+ try:
+ stdout, stderr = process.communicate(timeout=timeout)
+ except subprocess.TimeoutExpired:
+ os.killpg(process.pid, signal.SIGKILL)
+ process.communicate()
+ raise
+ return process.returncode, stdout, stderr
+
+
+def supported_version(value):
+ match = re.fullmatch(r"(?:agy\s+|Antigravity CLI\s+)?v?(\d+)\.(\d+)\.(\d+)", value.strip(), re.I)
+ return bool(match and tuple(map(int, match.groups())) >= (1, 1, 11))
+
+
+def extract_limits(data):
+ groups = data.get("groups")
+ if not isinstance(groups, list):
+ return []
+ groups = [group for group in groups if isinstance(group, dict)]
+ primary = next((group for group in groups if "gemini" in str(group.get("name", "")).lower()), None)
+ ordered = ([primary] if primary is not None else []) + [group for group in groups if group is not primary]
+ limits = []
+ for group in ordered:
+ name = str(group.get("name") or "Models")
+ scoped = "Claude/GPT" if "claude" in name.lower() else name
+ buckets = group.get("buckets")
+ if not isinstance(buckets, list):
+ continue
+ for window, title, label in (("5h", "Session", "Session (5-hour)"), ("weekly", "Weekly", "Weekly (7-day)")):
+ for bucket in buckets:
+ if not isinstance(bucket, dict) or bucket.get("window") != window:
+ continue
+ remaining = bucket.get("remaining_fraction")
+ if isinstance(remaining, bool) or not isinstance(remaining, (int, float)):
+ continue
+ if not math.isfinite(remaining) or not 0 <= remaining <= 1:
+ continue
+ used = round(1 - remaining, 4)
+ reset = parse_time(bucket.get("reset_time"))
+ limits.append({
+ "label": label if group is primary else f"{scoped} {label}",
+ "title": title if group is primary else f"{scoped} {title}",
+ "percent": used,
+ "resetsAt": reset.isoformat() if reset else "",
+ })
+ return limits
+
+
+def probe_failure(message):
+ # Do not publish raw stderr: it can contain account or credential details.
+ message = message.lower()
+ status = "Antigravity limits unavailable"
+ help_text = "Antigravity could not read quotas. Run `agy` and check `/usage`, then refresh."
+ retry = False
+ if re.search(r"\b429\b|resource.?exhausted|rate.?limit", message):
+ status = "Antigravity limits rate limited"
+ help_text = "Antigravity is rate limiting quota checks. Try refreshing later."
+ elif re.search(r"\b401\b|\b403\b|unauthenticated|permission.?denied", message):
+ status = "Antigravity sign-in required"
+ help_text = AUTH_HELP
+ elif re.search(r"network|connection|dial tcp|no such host|timed? ?out|timeout|deadline exceeded|unreachable", message):
+ help_text = "Could not reach Antigravity. Retrying shortly; local history is still shown."
+ retry = True
+ elif re.search(r"sign.?in|log.?in|auth|credential", message):
+ status = "Antigravity sign-in required"
+ help_text = AUTH_HELP
+ return {"usageStatusText": status, "authHelpText": help_text, "retryAdvised": retry}
+
+
+def probe_limits(cli):
+ try:
+ code, version, _ = run_cli(cli, "--version", timeout=5)
+ if code != 0 or not supported_version(version):
+ return {"usageStatusText": "Antigravity update required", "authHelpText": "Quota checks require Antigravity CLI 1.1.11 or newer. Update `agy`, then refresh."}
+ code, stdout, stderr = run_cli(cli, "-p", "/usage", "--output-format", "json")
+ if code != 0:
+ return probe_failure(stderr + "\n" + stdout)
+ payload = json.loads(stdout)
+ command = payload.get("command") if isinstance(payload, dict) else None
+ if not isinstance(command, dict) or command.get("name") not in ("usage", "quota") or payload.get("status") != "SUCCESS":
+ return probe_failure("invalid quota response")
+ data = command.get("data")
+ limits = extract_limits(data) if isinstance(data, dict) else []
+ if not limits:
+ return {"usageStatusText": "Antigravity limits unavailable", "authHelpText": "Antigravity returned no quota windows for this sign-in. Check `/usage` in `agy`."}
+ return {"limits": limits, "usageStatusText": "", "authHelpText": ""}
+ except subprocess.TimeoutExpired:
+ return probe_failure("timeout")
+ except (OSError, ValueError):
+ return probe_failure("could not read CLI output")
+
+
+def collect_limits(cli):
+ # /usage provides no account identifier. Reusing quotas or errors across
+ # calls could attribute a previous sign-in's result to the current account.
+ # Always ask the CLI and leave the meters empty when it cannot answer.
+ result = {"limits": [], "limitsStale": False, "limitsFetchedAt": 0,
+ "usageStatusText": "", "authHelpText": "", "retryAdvised": False}
+ if not cli:
+ result.update(usageStatusText="Antigravity CLI unavailable", authHelpText="Install Antigravity CLI and run `agy` to sign in.")
+ else:
+ result.update(probe_limits(cli))
+ if result["limits"]:
+ result["limitsFetchedAt"] = round(time.time() * 1000)
+ return result
+
+
+def tier_label():
+ # /usage does not report a plan. Only show a label the user supplied.
+ config_home = Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config")
+ value = os.environ.get("AGY_TIER") or read_json(config_home / "omarchy/agents/antigravity.json").get("tier")
+ return value.strip() if isinstance(value, str) else ""
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--force", action="store_true", help="Bypass the history cache; quotas are always read fresh")
+ parser.add_argument("--limits-only", action="store_true", help="Reuse a recent history scan")
+ args = parser.parse_args()
+ root = app_root()
+ now = datetime.now().astimezone()
+ # Isolate custom app roots and invalidate the old estimated-token cache.
+ key = hashlib.sha256(str(root).encode()).hexdigest()[:16]
+ prefix = cache_root() / f"antigravity-v{CACHE_VERSION}-{key}"
+ prefix.parent.mkdir(parents=True, exist_ok=True)
+ with prefix.with_suffix(".lock").open("w") as lock:
+ fcntl.flock(lock, fcntl.LOCK_EX)
+ stats_path = prefix.with_suffix(".stats.json")
+ cached = read_json(stats_path)
+ age = LIMITS_ONLY_REUSE_SECONDS if args.limits_only else SCAN_REUSE_SECONDS
+ fetched = cached.get("fetchedAt", 0)
+ stats = cached.get("stats")
+ if (args.force or not isinstance(stats, dict)
+ or not all(isinstance(stats.get(key), int) for key in ("totalPrompts", "totalSessions", "todayPrompts", "todaySessions"))
+ or cached.get("scanDate") != now.date().isoformat()
+ or not isinstance(fetched, (int, float)) or not 0 <= time.time() - fetched < age):
+ stats = collect_local_stats(root, now)
+ write_json(stats_path, {"scanDate": now.date().isoformat(), "fetchedAt": time.time(), "stats": stats})
+ limits = collect_limits(installed_cli())
+ has_stats = stats["totalPrompts"] > 0 or stats["totalSessions"] > 0
+ record = {"schemaVersion": 1, "id": "antigravity", "name": "Antigravity",
+ "updatedAt": datetime.now(timezone.utc).isoformat(), "ready": has_stats or bool(limits["limits"]),
+ "hasLocalStats": has_stats, "tierLabel": tier_label(), **stats, **limits}
+ print(json.dumps(record, separators=(",", ":"), allow_nan=False))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/bin/omarchy-agent-usage-update b/bin/omarchy-agent-usage-update
index e759dc72459..4bd1ed3a6ea 100755
--- a/bin/omarchy-agent-usage-update
+++ b/bin/omarchy-agent-usage-update
@@ -39,17 +39,55 @@ wanted() {
return 1
}
-collect() {
+published_sequence() {
+ local value
+ value=$(jq -r '._collectionSequence // 0' "$1" 2>/dev/null) || value=0
+ if [[ $value =~ ^[0-9]{1,15}$ ]]; then
+ printf '%s\n' "$((10#$value))"
+ else
+ printf '0\n'
+ fi
+}
+
+collect() (
local collector="$1" agent="$2"
- local record tmp
- if ! record=$("$collector" "${flags[@]}") || [[ -z $record ]] || ! jq -e . >/dev/null 2>&1 <<<"$record"; then
+ local record tmp sequence_tmp="" generation=0 published
+ local sequence="$USAGE_DIR/.$agent.sequence" lock="$USAGE_DIR/.$agent.lock"
+ tmp=$(mktemp "$USAGE_DIR/.$agent.XXXXXX") || return 1
+ trap 'rm -f "$tmp" "$sequence_tmp"' EXIT
+ sequence_tmp=$(mktemp "$USAGE_DIR/.$agent.sequence.XXXXXX") || return 1
+
+ # Overlapping refreshes may finish in either order, including across
+ # account changes. Assign request order before probing, then publish only
+ # if no later request has already published a valid record. A collector
+ # that crashes cannot prevent another valid result from landing. Hold the
+ # lock only for file operations, never through the network calls.
+ {
+ flock 9 || return 1
+ [[ -f $sequence ]] && read -r generation <"$sequence"
+ [[ $generation =~ ^[0-9]{1,15}$ ]] || generation=0
+ generation=$((10#$generation))
+ published=$(published_sequence "$USAGE_DIR/$agent.json")
+ (( published > generation )) && generation=$published
+ (( ++generation ))
+ printf '%s\n' "$generation" >"$sequence_tmp" && mv "$sequence_tmp" "$sequence"
+ } 9>"$lock" || return 1
+
+ if ! record=$("$collector" "${flags[@]}") || [[ -z $record ]] || ! jq -es 'length == 1 and (.[0] | type == "object")' >/dev/null 2>&1 <<<"$record"; then
echo "omarchy-agent-usage-update: $agent collector failed" >&2
return 1
fi
- tmp=$(mktemp "$USAGE_DIR/.$agent.XXXXXX")
- printf '%s\n' "$record" >"$tmp"
- mv "$tmp" "$USAGE_DIR/$agent.json"
-}
+ # Stamp the sequence into the same file as the data so the final rename
+ # publishes both atomically, even if the updater exits just after it.
+ jq --argjson sequence "$generation" '._collectionSequence = $sequence' <<<"$record" >"$tmp" || return 1
+ {
+ flock 9 || return 1
+ published=$(published_sequence "$USAGE_DIR/$agent.json")
+ if (( generation > published )); then
+ mv "$tmp" "$USAGE_DIR/$agent.json" || return 1
+ fi
+ } 9>"$lock"
+)
# Accounts can be signed in to someone else since they were added; the
# records name them from the registry, so bring it up to date first.
diff --git a/manual/17-ai.md b/manual/17-ai.md
index ea8a8d65eaf..30a5fc53982 100644
--- a/manual/17-ai.md
+++ b/manual/17-ai.md
@@ -38,7 +38,7 @@ There are terminal shortcuts too: `a` runs the default agent inline in the curre
### The agents panel
-The top bar has an agents icon. Before you've set up an agent, its panel walks you through signing in to Claude, Codex, or Grok, and the first one becomes your default agent. After that, the panel puts every subscription on one page: each agent and account with its plan and the percentage used of its 5-hour session and weekly limits (or the remaining prepaid balance), under a line that rotates through what you've used across all of them, like tokens this week and your most used model. Under _Make something_ it offers starter prompts for a new theme, plugin, or app that hand the task to your default agent. Apps are built the way Omarchy's own Hype, Monologue, and Omacut are, in C++ and Qt Quick, following the omarchy-app agent skill, and the compiler and Qt pieces they need come installed. Claude Code, Codex, Grok, and Fireworks are covered out of the box.
+The top bar has an agents icon. Before you've set up an agent, its panel walks you through signing in to Claude, Codex, or Grok, and the first one becomes your default agent. After that, the panel puts every subscription on one page: each agent and account with its plan and the percentage used of its 5-hour session and weekly limits (or the remaining prepaid balance), under a line that rotates through what you've used across all of them, like tokens this week and your most used model. Under _Make something_ it offers starter prompts for a new theme, plugin, or app that hand the task to your default agent. Apps are built the way Omarchy's own Hype, Monologue, and Omacut are, in C++ and Qt Quick, following the omarchy-app agent skill, and the compiler and Qt pieces they need come installed. Antigravity, Claude Code, Codex, Grok, and Fireworks are covered out of the box. Antigravity needs CLI 1.1.11 or newer and a sign-in through `agy`; its section shows subscription limits and contributes local prompt/session counts, but its saved transcripts do not provide token counts.
Left-click the bar icon for the panel, right-click to launch your default agent. The usage records behind it are regenerated every 15 minutes by `omarchy agent usage-update`, and the panel can even merge usage from your other machines via a synced folder. See the README under `$OMARCHY_PATH/shell/plugins/agents/` for the full settings.
diff --git a/shell/plugins/agents/README.md b/shell/plugins/agents/README.md
index c9daebba45e..e2ef7efad38 100644
--- a/shell/plugins/agents/README.md
+++ b/shell/plugins/agents/README.md
@@ -60,16 +60,23 @@ Each agent is one JSON record in `~/.local/state/omarchy/agents/usage/`,
written by `omarchy-agent-usage-update`. That command runs one
`omarchy-agent-usage-` collector per agent; the widget invokes it
on its refresh timer and whenever you ask for a refresh, and picks up any
-record that lands in the directory regardless of who wrote it.
+record that lands in the directory regardless of who wrote it. Overlapping
+updates reserve publication before collecting: a sequence assigned before collection
+prevents a slow older result from replacing a newer published record. A crashed
+collector or malformed response does not cancel a valid in-flight result; a
+newer valid error record, such as a sign-out, does. The sequence and record
+are published together atomically. The publication lock is released during
+collection.
Adding an agent therefore never touches this plugin: ship a collector that
prints the record contract (see the `claude` and `codex` collectors in
-`bin/`), and the panel gains a tab. An `assets/.svg` mark is optional —
+`bin/`), and the panel gains a section. An `assets/.svg` mark is optional —
with an `assets/-light.svg` twin if the mark needs a dark variant for
light surfaces — and the bar glyph stands in when there is none.
| Collector | Limits | Local stats |
|---|---|---|
+| `antigravity` | The Antigravity CLI’s read-only `/usage` command (1.1.11 or newer) | Prompt/session counts from transcripts, history and conversation summaries under `~/.gemini/antigravity-cli/` (`AGY_DIR`); token counts are unavailable |
| `claude` | Anthropic's OAuth usage endpoint (5-hour session + 7-day weekly) | `~/.claude/projects` transcripts, opencode sessions on an Anthropic provider, plus `stats-cache.json` and `history.jsonl` as fallback |
| `codex` | The Codex app-server RPC | native Codex CLI session files (plus pi and opencode sessions) |
| `grok` | The credits endpoint behind Grok's `/usage` view (the billing period's included usage) | Each session's `usage.json` (the ledger `grok usage` prints: tokens by model per finished turn), plus `summary.json` for sessions |
@@ -101,6 +108,14 @@ again. Fireworks reads
opencode stores in `~/.local/share/opencode/auth.json` when Fireworks is
signed in there.
+### Antigravity quotas and history
+
+The collector runs `agy -p "/usage" --output-format json` after checking that the installed CLI is at least 1.1.11. That version introduced read-only slash-command output without an agent turn or quota spend; an older or unrecognized version is never sent `/usage`. It resolves an existing mise installation without invoking Omarchy’s installer wrapper. Antigravity itself handles credential storage, token refresh and the account’s quota request.
+
+Session and weekly meters use the returned quota windows, including unused Claude/GPT allowances. Missing percentages are unknown, not zero usage. Failed checks show a status and help text; network failures request an early retry. Every refresh asks the CLI again. Because the response provides no account identity, quota results and failures are never cached: switching accounts or signing in again takes effect on the next check, and a failed check leaves the meters empty. `AGY_DIR` selects another app data directory with its own history cache.
+
+The CLI’s transcripts do not contain token usage or model IDs. Local history contributes prompt and session counts on local calendar dates, without character-based token estimates or invented model rows. History and transcript prompt events are combined individually, matching the complete prompt text and timestamps within one second (and conversation IDs when present) to avoid counting the same prompt twice. Transcript `` wrappers are recognized and surrounding metadata is excluded; unmatched events from either source and repeated prompts at different times are retained. The `/usage` response does not include a subscription tier, so the plan stays unknown unless you supply `AGY_TIER` or `{ "tier": "Ultra" }` in `~/.config/omarchy/agents/antigravity.json` (honoring `XDG_CONFIG_HOME`). Sign in through `agy`; the panel’s built-in account switching remains for Claude, Codex and Grok.
+
### Fireworks balance
The collector first asks the account's `:getBalance` endpoint for the real
@@ -177,6 +192,7 @@ edit `shell.json` directly):
```bash
omarchy bar set omarchy.agents providers '{
+ "antigravity": { "enabled": true },
"claude": { "enabled": true },
"codex": { "enabled": false },
"fireworks": { "enabled": true }
diff --git a/shell/plugins/agents/assets/antigravity-light.svg b/shell/plugins/agents/assets/antigravity-light.svg
new file mode 100644
index 00000000000..67ced694468
--- /dev/null
+++ b/shell/plugins/agents/assets/antigravity-light.svg
@@ -0,0 +1 @@
+
diff --git a/shell/plugins/agents/assets/antigravity.svg b/shell/plugins/agents/assets/antigravity.svg
new file mode 100644
index 00000000000..3743be8edf4
--- /dev/null
+++ b/shell/plugins/agents/assets/antigravity.svg
@@ -0,0 +1 @@
+
diff --git a/shell/plugins/agents/manifest.json b/shell/plugins/agents/manifest.json
index 7ca52484b12..51b32caae2a 100644
--- a/shell/plugins/agents/manifest.json
+++ b/shell/plugins/agents/manifest.json
@@ -5,7 +5,7 @@
"version": "1.0.0",
"author": "Omarchy",
"license": "MIT",
- "description": "Claude Code, Codex, Grok, and Fireworks usage, limits, and pace in a native Omarchy bar panel.",
+ "description": "Antigravity, Claude Code, Codex, Grok, and Fireworks usage, limits, and pace in a native Omarchy bar panel.",
"kinds": ["bar-widget"],
"activation": "on-demand",
"entryPoints": {
@@ -19,6 +19,7 @@
"allowMultiple": false,
"defaults": {
"providers": {
+ "antigravity": { "enabled": true },
"claude": { "enabled": true },
"codex": { "enabled": true },
"grok": { "enabled": true },
diff --git a/test/shell.d/agent-usage-antigravity-limits-test.sh b/test/shell.d/agent-usage-antigravity-limits-test.sh
new file mode 100755
index 00000000000..871457f6632
--- /dev/null
+++ b/test/shell.d/agent-usage-antigravity-limits-test.sh
@@ -0,0 +1,102 @@
+#!/bin/bash
+
+set -euo pipefail
+source "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)/base-test.sh"
+require_command python3
+
+python3 - <<'PY'
+import importlib.machinery
+import importlib.util
+import json
+import os
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+root = Path(os.environ['ROOT'])
+loader = importlib.machinery.SourceFileLoader('collector', str(root / 'bin/omarchy-agent-usage-antigravity'))
+spec = importlib.util.spec_from_loader(loader.name, loader)
+collector = importlib.util.module_from_spec(spec)
+loader.exec_module(collector)
+fixture = json.loads((root / 'test/shell.d/fixtures/antigravity/usage.json').read_text())
+
+class Limits(unittest.TestCase):
+ def test_real_cli_shape_and_reversed_groups(self):
+ data = fixture['command']['data']
+ limits = collector.extract_limits(data)
+ self.assertEqual([v['title'] for v in limits], ['Session', 'Weekly', 'Claude/GPT Session', 'Claude/GPT Weekly'])
+ self.assertEqual([v['percent'] for v in limits], [0, .27, 0, .03])
+ reversed_data = {'groups': list(reversed(data['groups']))}
+ self.assertEqual(limits, collector.extract_limits(reversed_data))
+ self.assertEqual(limits[0]['resetsAt'], '2099-10-02T07:42:57+00:00')
+
+ def test_missing_invalid_and_exhausted_buckets(self):
+ for remaining in [None, 'bad', True, float('nan'), float('inf'), -1, 2]:
+ data = {'groups': [{'name': 'Gemini Models', 'buckets': [{'window': '5h', 'remaining_fraction': remaining}]}]}
+ self.assertEqual(collector.extract_limits(data), [], remaining)
+ data['groups'][0]['buckets'][0]['remaining_fraction'] = 0
+ self.assertEqual(collector.extract_limits(data)[0]['percent'], 1)
+ del data['groups'][0]['buckets'][0]['remaining_fraction']
+ self.assertEqual(collector.extract_limits(data), [])
+ for data in [{}, {'groups': None}, {'groups': [None, {'buckets': 7}]}]:
+ self.assertEqual(collector.extract_limits(data), [])
+
+ def test_only_scoped_group_keeps_its_identity(self):
+ data = {'groups': [fixture['command']['data']['groups'][0]]}
+ self.assertEqual([limit['title'] for limit in collector.extract_limits(data)], ['Claude/GPT Session', 'Claude/GPT Weekly'])
+ self.assertEqual(collector.extract_limits(data)[0]['percent'], 0)
+
+ def test_version_gate_never_prompts_an_old_or_unknown_cli(self):
+ for version in ['1.1.10', '1.0.99', 'unknown', '1.1.11-preview', '']:
+ with patch.object(collector, 'run_cli', return_value=(0, version, '')) as run:
+ result = collector.probe_limits('agy')
+ self.assertEqual(run.call_count, 1)
+ self.assertEqual(result['usageStatusText'], 'Antigravity update required')
+ for version in ['1.1.11', '1.2.14', '2.0.0', 'agy 1.2.14']:
+ self.assertTrue(collector.supported_version(version))
+
+ def test_failures_and_malformed_success(self):
+ cases = [('401 unauthenticated', 'Antigravity sign-in required', False),
+ ('403 PermissionDenied', 'Antigravity sign-in required', False),
+ ('429 RESOURCE_EXHAUSTED', 'Antigravity limits rate limited', False),
+ ('dial tcp: no such host', 'Antigravity limits unavailable', True),
+ ('refresh credentials: dial tcp: connection refused', 'Antigravity limits unavailable', True),
+ ('500 internal server error', 'Antigravity limits unavailable', False)]
+ for error, status, retry in cases:
+ with patch.object(collector, 'run_cli', side_effect=[(0, '1.2.14', ''), (1, '', error)]):
+ result = collector.probe_limits('agy')
+ self.assertEqual(result['usageStatusText'], status)
+ self.assertEqual(result['retryAdvised'], retry)
+ self.assertNotIn(error, result['authHelpText'])
+ for output in ['invalid', '[]', '{}', '{"status":"SUCCESS","response":"pretend usage"}',
+ '{"status":"SUCCESS","command":{"name":"usage","data":{}}}']:
+ with patch.object(collector, 'run_cli', side_effect=[(0, '1.2.14', ''), (0, output, '')]):
+ self.assertTrue(collector.probe_limits('agy')['usageStatusText'])
+ with patch.object(collector, 'run_cli', side_effect=collector.subprocess.TimeoutExpired('agy', 20)):
+ self.assertTrue(collector.probe_limits('agy')['retryAdvised'])
+
+ def test_every_check_reads_the_current_sign_in(self):
+ good = {'limits': collector.extract_limits(fixture['command']['data']), 'usageStatusText': '', 'authHelpText': ''}
+ other = {'limits': [{'title': 'Session', 'percent': .8, 'resetsAt': '2099-01-01T00:00:00Z'}], 'usageStatusText': '', 'authHelpText': ''}
+ failure = collector.probe_failure('network unavailable')
+ auth = collector.probe_failure('401')
+ with patch.object(collector, 'probe_limits', side_effect=[good, other, failure, auth, good]) as probe:
+ first = collector.collect_limits('agy')
+ second = collector.collect_limits('agy')
+ self.assertEqual(first['limits'], good['limits'])
+ self.assertEqual(second['limits'], other['limits'])
+ self.assertGreater(second['limitsFetchedAt'], 0)
+ offline = collector.collect_limits('agy')
+ self.assertEqual(offline['limits'], [])
+ self.assertTrue(offline['retryAdvised'])
+ self.assertEqual(offline['limitsFetchedAt'], 0)
+ signed_out = collector.collect_limits('agy')
+ self.assertEqual(signed_out['limits'], [])
+ signed_in = collector.collect_limits('agy')
+ self.assertEqual(signed_in['usageStatusText'], '')
+ self.assertEqual(signed_in['limits'], good['limits'])
+ self.assertEqual(probe.call_count, 5)
+
+unittest.main(verbosity=2)
+PY
+pass "Antigravity quota parsing, version safety, failures and account changes"
diff --git a/test/shell.d/agent-usage-antigravity-scanner-test.sh b/test/shell.d/agent-usage-antigravity-scanner-test.sh
new file mode 100755
index 00000000000..a0e6d8ffec9
--- /dev/null
+++ b/test/shell.d/agent-usage-antigravity-scanner-test.sh
@@ -0,0 +1,251 @@
+#!/bin/bash
+
+set -euo pipefail
+source "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)/base-test.sh"
+require_command python3
+
+python3 - <<'PY'
+import fcntl
+import importlib.machinery
+import importlib.util
+import json
+import os
+import sqlite3
+import subprocess
+import tempfile
+import time
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+root = Path(os.environ['ROOT'])
+script = root / 'bin/omarchy-agent-usage-antigravity'
+loader = importlib.machinery.SourceFileLoader('collector', str(script))
+spec = importlib.util.spec_from_loader(loader.name, loader)
+collector = importlib.util.module_from_spec(spec)
+loader.exec_module(collector)
+
+class Scanner(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ self.home = Path(self.tmp.name)
+ self.app = self.home / 'app'
+ self.app.mkdir()
+ self.bin = self.home / 'bin'
+ self.bin.mkdir()
+ self.log = self.home / 'calls'
+ self.cli = self.bin / 'agy'
+ self.cli.write_text('''#!/usr/bin/python3
+import json, os, sys, time
+from pathlib import Path
+with open(os.environ['AGY_TEST_LOG'], 'a') as f:
+ f.write(json.dumps(sys.argv[1:]) + '\\n')
+if sys.argv[1:] == ['--version']:
+ print(os.environ.get('AGY_TEST_VERSION', '1.2.14'))
+ sys.exit(0)
+assert sys.argv[1:] == ['-p', '/usage', '--output-format', 'json']
+if os.environ.get('AGY_TEST_STARTED'):
+ Path(os.environ['AGY_TEST_STARTED']).touch()
+ time.sleep(2)
+error = os.environ.get('AGY_TEST_ERROR')
+if error:
+ print(error, file=sys.stderr)
+ sys.exit(1)
+print(Path(os.environ['AGY_TEST_FIXTURE']).read_text())
+''')
+ self.cli.chmod(0o755)
+ self.env = {'HOME': str(self.home), 'PATH': str(self.bin) + ':/usr/bin', 'TZ': 'Etc/GMT-14',
+ 'AGY_DIR': str(self.app), 'XDG_CACHE_HOME': str(self.home / 'cache'),
+ 'AGY_TEST_LOG': str(self.log), 'AGY_TEST_FIXTURE': str(root / 'test/shell.d/fixtures/antigravity/usage.json')}
+
+ def run_collector(self, *args, **extra):
+ result = subprocess.run([str(script), *args], env={**self.env, **extra}, text=True, capture_output=True, check=True)
+ self.assertEqual(result.stderr, '')
+ return json.loads(result.stdout)
+
+ def transcript(self, name, entries):
+ path = self.app / 'brain' / name / '.system_generated/logs/transcript.jsonl'
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text('\n'.join(json.dumps(e) for e in entries) + '\n{partial')
+
+ def test_local_dates_real_shapes_and_no_invented_tokens(self):
+ now = datetime(2026, 9, 28, 1, tzinfo=timezone.utc)
+ prompt = {'step_index': 0, 'source': 'USER', 'type': 'USER_INPUT', 'status': 'DONE',
+ 'created_at': '2026-09-27T23:00:00Z', 'content': 'Hello'}
+ self.transcript('session', [prompt, prompt,
+ {'step_index': 1, 'source': 'MODEL', 'type': 'PLANNER_RESPONSE', 'created_at': '2026-09-27T23:01:00Z', 'content': 'Answer', 'thinking': 'Thinking'},
+ {'step_index': 2, 'source': 'USER', 'type': 'USER_INPUT', 'content': 'Missing date'},
+ {'step_index': 3, 'type': 'USER_INPUT', 'created_at': '2099-01-01T00:00:00Z'},
+ {'step_index': 4, 'created_at': [], 'type': 'USER_INPUT'}])
+ history = [
+ {'display': 'Hello', 'timestamp': 1759014000000, 'workspace': '/project'},
+ {'display': 'Hello', 'timestamp': '2026-09-27T23:00:00Z', 'workspace': '/project'},
+ {'display': '/usage', 'timestamp': '2026-09-27T23:00:00Z'},
+ {'display': 'old', 'timestamp': '2026-09-27T01:00:00Z'},
+ {'timestamp': []}, {'timestamp': 10**30}, []]
+ # The first record exercises epoch milliseconds for the same instant.
+ history[0]['timestamp'] = int(datetime(2026, 9, 27, 23, tzinfo=timezone.utc).timestamp() * 1000)
+ (self.app / 'history.jsonl').write_text('\n'.join(json.dumps(e) for e in history))
+ with sqlite3.connect(self.app / 'conversation_summaries.db') as db:
+ db.execute('CREATE TABLE conversation_summaries (conversation_id TEXT, last_modified_time TEXT)')
+ db.executemany('INSERT INTO conversation_summaries VALUES (?, ?)', [('session', '2026-09-27 23:02:00+00:00'), ('summary-only', '2026-09-27 23:03:00+00:00')])
+ with patch.dict(os.environ, {'TZ': 'Etc/GMT-14'}):
+ time.tzset()
+ stats = collector.collect_local_stats(self.app, now)
+ time.tzset()
+ self.assertEqual(stats['todayPrompts'], 2)
+ self.assertEqual(stats['totalPrompts'], 3)
+ self.assertEqual(stats['todaySessions'], 2)
+ self.assertEqual(stats['totalSessions'], 2)
+ self.assertEqual(stats['activeDates'], ['2026-09-27', '2026-09-28'])
+ self.assertEqual(stats['recentDays'][-1], {'date': '2026-09-28', 'messageCount': 0})
+ self.assertEqual(stats['todayTotalTokens'], 0)
+ self.assertEqual(stats['modelUsage'], {})
+ self.assertEqual(stats['todayTokensByModel'], {})
+
+ def test_negative_timezone(self):
+ self.transcript('session', [{'step_index': 0, 'type': 'USER_INPUT', 'created_at': '2026-09-28T01:00:00Z'}])
+ with patch.dict(os.environ, {'TZ': 'America/New_York'}):
+ time.tzset()
+ stats = collector.collect_local_stats(self.app, datetime(2026, 9, 28, 2, tzinfo=timezone.utc))
+ time.tzset()
+ self.assertEqual(stats['activeDates'], ['2026-09-27'])
+ self.assertEqual(stats['recentDays'][-1]['date'], '2026-09-27')
+ self.assertEqual(stats['todayPrompts'], 1)
+
+ def test_executable_record_caches_custom_root_and_tier(self):
+ stamp = datetime.now(timezone.utc).isoformat()
+ (self.app / 'history.jsonl').write_text(json.dumps({'display': 'hello', 'timestamp': stamp}) + '\n')
+ record = self.run_collector('--force')
+ self.assertEqual(record['id'], 'antigravity')
+ self.assertTrue(record['ready'])
+ self.assertTrue(record['hasLocalStats'])
+ self.assertEqual(record['tierLabel'], '')
+ self.assertEqual(record['usageStatusText'], '')
+ self.assertEqual([e['percent'] for e in record['limits']], [0, .27, 0, .03])
+ self.assertEqual(record['todayPrompts'], 1)
+ self.assertEqual(record['todayTotalTokens'], 0)
+ self.assertEqual(record['modelUsage'], {})
+ calls = self.log.read_text()
+ (self.app / 'history.jsonl').write_text((json.dumps({'display': 'hello', 'timestamp': stamp}) + '\n') * 3)
+ self.assertEqual(self.run_collector('--limits-only')['todayPrompts'], 1)
+ self.assertEqual(len(self.log.read_text().splitlines()), len(calls.splitlines()) + 2)
+ self.assertEqual(self.run_collector('--force')['todayPrompts'], 3)
+ stats_path = next((self.home / 'cache/omarchy/agent-usage').glob('*.stats.json'))
+ cached = json.loads(stats_path.read_text())
+ cached['stats'] = {}
+ stats_path.write_text(json.dumps(cached))
+ self.assertEqual(self.run_collector('--limits-only')['todayPrompts'], 3)
+ other = self.home / 'other'
+ other.mkdir()
+ self.assertEqual(self.run_collector(AGY_DIR=str(other))['totalPrompts'], 0)
+ self.assertEqual(self.run_collector(AGY_TIER='Ultra')['tierLabel'], 'Ultra')
+ config = self.home / 'config/omarchy/agents'
+ config.mkdir(parents=True)
+ (config / 'antigravity.json').write_text('{"tier":"Enterprise"}')
+ self.assertEqual(self.run_collector(XDG_CONFIG_HOME=str(self.home / 'config'))['tierLabel'], 'Enterprise')
+
+ def test_real_command_errors_and_safety_gate(self):
+ for error, status, retry in [('401 expired token', 'Antigravity sign-in required', False),
+ ('403 permission denied', 'Antigravity sign-in required', False),
+ ('429 rate limit', 'Antigravity limits rate limited', False),
+ ('network unreachable', 'Antigravity limits unavailable', True),
+ ('500 server failure', 'Antigravity limits unavailable', False)]:
+ record = self.run_collector('--force', AGY_TEST_ERROR=error)
+ self.assertEqual(record['limits'], [])
+ self.assertEqual(record['tierLabel'], '')
+ self.assertEqual(record['usageStatusText'], status)
+ self.assertEqual(record['retryAdvised'], retry)
+ self.assertTrue(record['authHelpText'])
+ self.log.write_text('')
+ record = self.run_collector('--force', AGY_TEST_VERSION='1.1.10')
+ self.assertEqual(record['usageStatusText'], 'Antigravity update required')
+ self.assertEqual(self.log.read_text().splitlines(), ['["--version"]'])
+ self.run_collector('--force')
+ stale = self.run_collector('--force', AGY_TEST_ERROR='network unavailable')
+ self.assertFalse(stale['limitsStale'])
+ self.assertEqual(stale['limits'], [])
+ self.assertTrue(stale['retryAdvised'])
+
+ def test_partial_and_disjoint_history_preserve_distinct_prompts(self):
+ history = [{'display': 'shared', 'timestamp': '2026-09-27T12:00:00Z'},
+ {'display': 'history only', 'timestamp': '2026-09-27T12:01:00Z'}]
+ (self.app / 'history.jsonl').write_text('\n'.join(json.dumps(e) for e in history))
+ for shared, expected in [('shared', 4), ('different prompt', 5)]:
+ self.transcript('session', [
+ {'step_index': 0, 'type': 'USER_INPUT', 'created_at': '2026-09-27T12:00:00.400Z', 'content': '\n' + shared + '\n'},
+ {'step_index': 1, 'type': 'USER_INPUT', 'created_at': '2026-09-27T12:02:00Z', 'content': 'transcript only'},
+ {'step_index': 2, 'type': 'USER_INPUT', 'created_at': '2026-09-27T12:03:00Z', 'content': 'shared'}])
+ with patch.dict(os.environ, {'TZ': 'UTC'}):
+ time.tzset()
+ stats = collector.collect_local_stats(self.app, datetime(2026, 9, 27, 14, tzinfo=timezone.utc))
+ time.tzset()
+ self.assertEqual(stats['todayPrompts'], expected)
+ self.assertEqual(stats['totalPrompts'], expected)
+
+ def test_account_switch_and_new_sign_in_are_visible_immediately(self):
+ first = self.run_collector()
+ other = json.loads(Path(self.env['AGY_TEST_FIXTURE']).read_text())
+ other['command']['data']['groups'][1]['buckets'][1]['remaining_fraction'] = .2
+ other_file = self.home / 'other-account.json'
+ other_file.write_text(json.dumps(other))
+ second = self.run_collector(AGY_TEST_FIXTURE=str(other_file))
+ self.assertEqual(first['limits'][0]['percent'], 0)
+ self.assertEqual(second['limits'][0]['percent'], .8)
+ self.assertEqual(self.run_collector(AGY_TEST_ERROR='401 signed out')['limits'], [])
+ self.assertEqual(self.run_collector()['usageStatusText'], '')
+ self.assertFalse(list((self.home / 'cache').rglob('*.limits.json')))
+
+ def test_prompt_matching_ignores_substrings_and_metadata(self):
+ (self.app / 'history.jsonl').write_text(json.dumps({'display': 'hello', 'timestamp': '2026-09-27T12:00:00Z'}))
+ for content, expected in [
+ ('hellocontext', 1),
+ ('hellocontext\nmodel changed', 1),
+ ('hello worldcontext', 2),
+ ('differenthello', 2)]:
+ self.transcript('session', [{'step_index': 0, 'type': 'USER_INPUT', 'created_at': '2026-09-27T12:00:00.400Z', 'content': content}])
+ stats = collector.collect_local_stats(self.app, datetime(2026, 9, 27, 14, tzinfo=timezone.utc))
+ self.assertEqual(stats['totalPrompts'], expected)
+
+ def test_slow_quota_query_does_not_hold_the_history_lock(self):
+ started = self.home / 'quota-started'
+ process = subprocess.Popen([str(script)], env={**self.env, 'AGY_TEST_STARTED': str(started)},
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ try:
+ deadline = time.monotonic() + 5
+ while not started.exists() and process.poll() is None and time.monotonic() < deadline:
+ time.sleep(.02)
+ self.assertTrue(started.exists(), 'quota query started')
+ lock_path = next((self.home / 'cache').rglob('*.lock'))
+ with lock_path.open('w') as lock:
+ fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ fcntl.flock(lock, fcntl.LOCK_UN)
+ finally:
+ stdout, stderr = process.communicate(timeout=5)
+ self.assertEqual(process.returncode, 0, stderr)
+ self.assertEqual(json.loads(stdout)['usageStatusText'], '')
+
+ def test_literal_wrapper_tags_in_prompt_count_once(self):
+ for text in ['Explain please', 'nested',
+ 'Document literally']:
+ (self.app / 'history.jsonl').write_text(json.dumps({'display': text, 'timestamp': '2026-09-27T12:00:00Z'}))
+ wrapped = '\n' + text + '\n\ncontext'
+ self.transcript('session', [{'step_index': 0, 'type': 'USER_INPUT', 'created_at': '2026-09-27T12:00:00.400Z', 'content': wrapped}])
+ stats = collector.collect_local_stats(self.app, datetime(2026, 9, 27, 14, tzinfo=timezone.utc))
+ self.assertEqual(stats['totalPrompts'], 1)
+
+ def test_stock_wrapper_does_not_install(self):
+ self.cli.write_text('#!/bin/bash\nmise use -g antigravity-cli\nexit 99\n')
+ mise = self.bin / 'mise'
+ mise.write_text('#!/bin/bash\n[[ $* == "which agy" ]] || exit 99\nexit 1\n')
+ mise.chmod(0o755)
+ record = self.run_collector('--force')
+ self.assertEqual(record['usageStatusText'], 'Antigravity CLI unavailable')
+ self.assertFalse(record['ready'])
+ self.assertFalse(self.log.exists())
+
+unittest.main(verbosity=2)
+PY
+pass "Antigravity real CLI integration, local history, timezones and cache isolation"
diff --git a/test/shell.d/agent-usage-update-test.sh b/test/shell.d/agent-usage-update-test.sh
index ee3aaf0112e..db9f598c8f9 100755
--- a/test/shell.d/agent-usage-update-test.sh
+++ b/test/shell.d/agent-usage-update-test.sh
@@ -63,3 +63,84 @@ pass "update succeeds when the requested collectors all pass"
[[ -e $usage_dir/skipped.json && ! -e $usage_dir/noisy.json ]] ||
fail "update with agent arguments only runs the named collectors"
pass "update with agent arguments only runs the named collectors"
+
+require_command python3
+require_command flock
+
+TEST_HOME="$TEST_HOME" FAKE_OMARCHY="$FAKE_OMARCHY" python3 - <<'PY'
+import json
+import os
+import subprocess
+import time
+from pathlib import Path
+
+home = Path(os.environ['TEST_HOME'])
+fake = Path(os.environ['FAKE_OMARCHY'])
+updater = Path(os.environ['ROOT']) / 'bin/omarchy-agent-usage-update'
+collector = fake / 'bin/omarchy-agent-usage-racy'
+collector.write_text('''#!/bin/bash
+touch "$STARTED"
+if [[ -n $WAIT_FOR ]]; then
+ while [[ ! -e $WAIT_FOR ]]; do sleep .02; done
+fi
+printf '%s\\n' "$RECORD"
+exit "${COLLECT_EXIT:-0}"
+''')
+collector.chmod(0o755)
+env = {**os.environ, 'HOME': str(home), 'OMARCHY_PATH': str(fake), 'XDG_STATE_HOME': ''}
+usage_dir = home / '.local/state/omarchy/agents/usage'
+def published_record():
+ record = json.loads((usage_dir / 'racy.json').read_text())
+ sequence = record.pop('_collectionSequence')
+ assert isinstance(sequence, int) and sequence > 0
+ return record
+
+# A published service error is authoritative too: an older success must not
+# undo a newer sign-out. Only a hard failure to emit a record permits fallback.
+good_old = {'id': 'racy', 'limits': [{'percent': .1}]}
+good_new = {'id': 'racy', 'limits': [{'percent': .8}], 'usageStatusText': ''}
+error = {'id': 'racy', 'limits': [], 'usageStatusText': 'sign-in required'}
+for index, (old, new) in enumerate([(good_old, good_new), (error, good_new), (good_old, error)]):
+ started = home / f'started-{index}'
+ release = home / f'release-{index}'
+ older = subprocess.Popen([str(updater), 'racy'], env={**env, 'STARTED': str(started), 'WAIT_FOR': str(release), 'RECORD': json.dumps(old)},
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ try:
+ deadline = time.monotonic() + 5
+ while not started.exists() and older.poll() is None and time.monotonic() < deadline:
+ time.sleep(.02)
+ assert started.exists(), 'older collector started'
+ subprocess.run([str(updater), 'racy'], env={**env, 'STARTED': str(home / 'newer-started'), 'WAIT_FOR': '', 'RECORD': json.dumps(new)},
+ capture_output=True, text=True, check=True, timeout=5)
+ assert published_record() == new
+ finally:
+ release.touch()
+ stdout, stderr = older.communicate(timeout=5)
+ assert older.returncode == 0, stderr
+ assert published_record() == new, 'older completion replaced newer data'
+ assert sorted(p.name for p in usage_dir.glob('.racy.*')) == ['.racy.lock', '.racy.sequence'], 'temporary records are cleaned up'
+print('ok - older updates cannot overwrite a newer valid record, including sign-out errors')
+
+for index, failed in enumerate([{'RECORD': 'not json', 'COLLECT_EXIT': '0'},
+ {'RECORD': '{}\n{}', 'COLLECT_EXIT': '0'},
+ {'RECORD': '{}', 'COLLECT_EXIT': '1'}]):
+ started = home / f'failure-started-{index}'
+ release = home / f'failure-release-{index}'
+ good = {'id': 'racy', 'limits': [{'percent': .4}]}
+ older = subprocess.Popen([str(updater), 'racy'], env={**env, 'STARTED': str(started), 'WAIT_FOR': str(release), 'RECORD': json.dumps(good)},
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ try:
+ deadline = time.monotonic() + 5
+ while not started.exists() and older.poll() is None and time.monotonic() < deadline:
+ time.sleep(.02)
+ assert started.exists()
+ result = subprocess.run([str(updater), 'racy'], env={**env, 'STARTED': str(home / 'failure-newer'), 'WAIT_FOR': '', **failed},
+ capture_output=True, text=True, timeout=5)
+ assert result.returncode != 0, 'hard collector failures are reported'
+ finally:
+ release.touch()
+ stdout, stderr = older.communicate(timeout=5)
+ assert older.returncode == 0, stderr
+ assert published_record() == good, 'a failed newer request suppressed a valid in-flight result'
+print('ok - failed collectors do not cancel valid in-flight publications')
+PY
diff --git a/test/shell.d/fixtures/antigravity/README.md b/test/shell.d/fixtures/antigravity/README.md
new file mode 100644
index 00000000000..d26d0550f28
--- /dev/null
+++ b/test/shell.d/fixtures/antigravity/README.md
@@ -0,0 +1,9 @@
+# Antigravity fixtures
+
+`usage.json` preserves the structure of `agy -p "/usage" --output-format json` from Antigravity CLI 1.2.14, checked on October 1, 2026. Descriptions and redundant text output are omitted, percentages are rounded, reset dates are moved into the future, and the groups are reversed to exercise order independence. The command reports zero agent turns. Google's [1.1.11 release notes](https://github.com/google-antigravity/antigravity-cli/blob/main/CHANGELOG.md#1111) introduce this read-only interface.
+
+Local scanner fixtures use the observed 1.2.x transcript fields (`step_index`, `source`, `type`, `status`, `created_at`, `content`) and history fields (`display`, integer-millisecond `timestamp`, `workspace`). The installed CLI's transcripts have no model IDs or token usage fields. Tests must not assume they do or estimate tokens from content. SQLite fixtures reproduce only the summary columns the collector reads; conversation databases contain protobuf blobs and are not decoded.
+
+Prompt matching also covers `` wrappers observed in local transcripts. Matched history/transcript events in the inspected logs were less than one second apart; prompt text is only compared in memory and is never written to the usage cache.
+
+After the closing request tag, transcripts can append both an `` block and a separate `` block. Both are outside the user prompt and must be excluded from the comparison.
diff --git a/test/shell.d/fixtures/antigravity/usage.json b/test/shell.d/fixtures/antigravity/usage.json
new file mode 100644
index 00000000000..4e9c294d911
--- /dev/null
+++ b/test/shell.d/fixtures/antigravity/usage.json
@@ -0,0 +1,25 @@
+{
+ "status": "SUCCESS",
+ "num_turns": 0,
+ "command": {
+ "name": "usage",
+ "data": {
+ "groups": [
+ {
+ "name": "Claude and GPT models",
+ "buckets": [
+ {"id": "3p-weekly", "name": "Weekly Limit Remaining", "window": "weekly", "remaining_fraction": 0.97, "reset_time": "2099-10-06T02:43:46Z"},
+ {"id": "3p-5h", "name": "Five Hour Limit Remaining", "window": "5h", "remaining_fraction": 1, "reset_time": "2099-10-02T07:42:57Z"}
+ ]
+ },
+ {
+ "name": "Gemini Models",
+ "buckets": [
+ {"id": "gemini-weekly", "name": "Weekly Limit Remaining", "window": "weekly", "remaining_fraction": 0.73, "reset_time": "2099-10-02T15:41:56Z"},
+ {"id": "gemini-5h", "name": "Five Hour Limit Remaining", "window": "5h", "remaining_fraction": 1, "reset_time": "2099-10-02T07:42:57Z"}
+ ]
+ }
+ ]
+ }
+ }
+}