diff --git a/bin/omarchy-agent-usage-x b/bin/omarchy-agent-usage-x
new file mode 100755
index 00000000000..b09636ee058
--- /dev/null
+++ b/bin/omarchy-agent-usage-x
@@ -0,0 +1,658 @@
+#!/usr/bin/python3
+# omarchy:summary=Print The X Subscription usage record as JSON
+# omarchy:args=[--force] [--limits-only]
+# omarchy:hidden=true
+"""Collect The X Subscription usage into one display-ready JSON record.
+
+The shared first-party pool (Cursor + Grok) and plan come from Cursor's
+dashboard API using the sign-in in ~/.config/cursor/auth.json. Tokens by day
+and by model come from the same dashboard's GetAggregatedUsageEvents. Local
+prompt and session counts come from Cursor agent transcripts under
+~/.cursor/projects. The agents panel only ever reads the JSON this prints.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from collections import defaultdict
+from datetime import date, datetime, time as dt_time, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+AGENT_ID = "x"
+AGENT_NAME = "X Subscription"
+AUTH_HELP = "Run `cursor-agent login`, or sign in to Cursor, to refresh X Subscription usage."
+API_BASE = "https://api2.cursor.sh"
+PROBE_REUSE_SECONDS = 60
+LIMITS_ONLY_SCAN_REUSE_SECONDS = 900
+# Cursor agent transcripts stamp user turns as
+# <timestamp>Monday, Oct 12, 2026, 6:27 PM (UTC-5)</timestamp>.
+TIMESTAMP_RE = re.compile(
+  r"<timestamp>\s*([A-Za-z]+,\s+[A-Za-z]+\s+\d{1,2},\s+\d{4},\s+\d{1,2}:\d{2}\s*[AP]M)\s*(?:\(([^)]+)\))?\s*</timestamp>",
+  re.IGNORECASE,
+)
+TZ_OFFSET_RE = re.compile(r"^(?:UTC|GMT)([+-])(\d{1,2})(?::?(\d{2}))?$", re.IGNORECASE)
+
+
+def read_json(path: Path) -> Any:
+  try:
+    return json.loads(path.read_text(encoding="utf-8"))
+  except Exception:
+    return None
+
+
+def write_json(path: Path, payload: Any) -> None:
+  path.parent.mkdir(parents=True, exist_ok=True)
+  fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=str(path.parent))
+  try:
+    with os.fdopen(fd, "w", encoding="utf-8") as handle:
+      json.dump(payload, handle)
+    os.replace(tmp, path)
+  except Exception:
+    Path(tmp).unlink(missing_ok=True)
+    raise
+
+
+def cache_root() -> Path:
+  return Path(os.environ.get("XDG_CACHE_HOME") or (Path.home() / ".cache")) / "omarchy" / "agent-usage"
+
+
+def auth_path() -> Path:
+  config = Path(os.environ.get("XDG_CONFIG_HOME") or (Path.home() / ".config"))
+  return config / "cursor" / "auth.json"
+
+
+def load_token() -> str | None:
+  data = read_json(auth_path())
+  if not isinstance(data, dict):
+    return None
+  token = data.get("accessToken")
+  return token if isinstance(token, str) and token else None
+
+
+def token_fingerprint(token: str) -> str:
+  return hashlib.sha1(token.encode("utf-8")).hexdigest()[:16]
+
+
+def limits_cache_path(token: str | None) -> Path:
+  if token:
+    return cache_root() / f"x-limits-{token_fingerprint(token)}.json"
+  return cache_root() / "x-limits.json"
+
+
+def rpc(token: str, method: str, body: dict | None = None) -> dict[str, Any]:
+  request = urllib.request.Request(
+    f"{API_BASE}/{method}",
+    data=json.dumps(body or {}).encode(),
+    headers={
+      "Authorization": f"Bearer {token}",
+      "Content-Type": "application/json",
+      "Connect-Protocol-Version": "1",
+      "Accept": "application/json",
+      "User-Agent": "omarchy-agent-usage-x/1.0",
+      "x-cursor-client-type": "cli",
+    },
+    method="POST",
+  )
+  with urllib.request.urlopen(request, timeout=20) as response:
+    payload = json.loads(response.read().decode())
+  return payload if isinstance(payload, dict) else {}
+
+
+def ms_to_iso(value: Any) -> str:
+  try:
+    ms = int(str(value))
+  except (TypeError, ValueError):
+    return ""
+  if ms <= 0:
+    return ""
+  return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def pct_to_unit(value: Any) -> float:
+  """Convert Cursor's 0–100 `*PercentUsed` fields to a 0–1 panel unit."""
+  try:
+    n = float(value)
+  except (TypeError, ValueError):
+    return -1.0
+  if n < 0:
+    return -1.0
+  return min(n / 100.0, 1.0)
+
+
+def empty_stats() -> dict[str, Any]:
+  today = date.today()
+  # recentDays.messageCount is token totals across agents; keep prompts in
+  # todayPrompts/totalPrompts so the hero does not call them tokens.
+  recent = [{"date": (today - timedelta(days=i)).isoformat(), "messageCount": 0} for i in range(6, -1, -1)]
+  return {
+    "hasLocalStats": False,
+    "hasPromptStats": True,
+    "usageUnit": "prompts",
+    "todayPrompts": 0,
+    "todaySessions": 0,
+    "todayTotalTokens": 0,
+    "todayTokensByModel": {},
+    "recentDays": recent,
+    "totalPrompts": 0,
+    "totalSessions": 0,
+    "activeDays": 0,
+    "activeDates": [],
+    "modelUsage": {},
+  }
+
+
+def local_tzinfo() -> timezone:
+  return datetime.now().astimezone().tzinfo or timezone.utc
+
+
+def parse_tz_label(label: str) -> timezone | None:
+  match = TZ_OFFSET_RE.fullmatch(label.strip())
+  if not match:
+    return None
+  sign = 1 if match.group(1) == "+" else -1
+  hours = int(match.group(2))
+  minutes = int(match.group(3) or 0)
+  return timezone(timedelta(hours=sign * hours, minutes=sign * minutes))
+
+
+def to_local_date(value: datetime) -> date:
+  """Calendar day in the machine's local zone (same basis as date.today())."""
+  if value.tzinfo is None:
+    value = value.replace(tzinfo=local_tzinfo())
+  return value.astimezone(local_tzinfo()).date()
+
+
+def empty_token_bucket() -> dict[str, int]:
+  return {
+    "inputTokens": 0,
+    "outputTokens": 0,
+    "cacheReadInputTokens": 0,
+    "cacheCreationInputTokens": 0,
+  }
+
+
+def day_windows(today: date, count: int = 7) -> list[tuple[str, int, int]]:
+  """Local calendar days as (date, startMs, endMs), oldest first."""
+  windows = []
+  for offset in range(count - 1, -1, -1):
+    day = today - timedelta(days=offset)
+    start = datetime.combine(day, dt_time.min).astimezone()
+    end = datetime.combine(day + timedelta(days=1), dt_time.min).astimezone()
+    windows.append((day.isoformat(), int(start.timestamp() * 1000), int(end.timestamp() * 1000)))
+  return windows
+
+
+def summarize_day_rows(rows: list[Any]) -> dict[str, dict[str, int]]:
+  usage: dict[str, dict[str, int]] = {}
+  for row in rows:
+    if not isinstance(row, dict):
+      continue
+    model = str(row.get("modelIntent") or "cursor")
+    bucket = usage.setdefault(model, empty_token_bucket())
+    bucket["inputTokens"] += int(float(row.get("inputTokens") or 0))
+    bucket["outputTokens"] += int(float(row.get("outputTokens") or 0))
+    bucket["cacheReadInputTokens"] += int(float(row.get("cacheReadTokens") or 0))
+    bucket["cacheCreationInputTokens"] += int(float(row.get("cacheWriteTokens") or 0))
+  return usage
+
+
+def remote_token_stats(token: str, previous: dict[str, Any] | None = None) -> dict[str, Any]:
+  """Fill recentDays.messageCount with real token totals from Cursor's dashboard.
+
+  A failed day keeps the last known count for that date instead of writing zero,
+  so a timeout cannot wipe the panel's week of tokens.
+  """
+  previous = previous if isinstance(previous, dict) else {}
+  prev_by_day = {
+    str(row.get("date")): row
+    for row in (previous.get("recentDays") or [])
+    if isinstance(row, dict) and row.get("date")
+  }
+  today = date.today()
+  recent: list[dict[str, Any]] = []
+  model_usage: dict[str, dict[str, int]] = {}
+  today_by_model: dict[str, int] = {}
+  active_dates: list[str] = []
+  any_ok = False
+  any_fail = False
+
+  for index, (day, start_ms, end_ms) in enumerate(day_windows(today, 7)):
+    usage: dict[str, dict[str, int]] | None = None
+    try:
+      payload = rpc(
+        token,
+        "aiserver.v1.DashboardService/GetAggregatedUsageEvents",
+        {"startDate": str(start_ms), "endDate": str(end_ms)},
+      )
+      rows = payload.get("aggregations") if isinstance(payload, dict) else None
+      usage = summarize_day_rows(rows if isinstance(rows, list) else [])
+      any_ok = True
+    except Exception:
+      any_fail = True
+      usage = None
+
+    if usage is None:
+      kept = prev_by_day.get(day) or {"date": day, "messageCount": 0}
+      tokens = int(kept.get("messageCount") or 0)
+      recent.append({"date": day, "messageCount": tokens})
+    else:
+      tokens = 0
+      for model, bucket in usage.items():
+        carried = model_usage.setdefault(model, empty_token_bucket())
+        for key, value in bucket.items():
+          carried[key] += value
+        tokens += sum(bucket.values())
+      recent.append({"date": day, "messageCount": tokens})
+      if index == 6:
+        today_by_model = {
+          model: sum(bucket.values()) for model, bucket in usage.items() if sum(bucket.values()) > 0
+        }
+
+    if tokens > 0:
+      active_dates.append(day)
+
+  if not any_ok and previous.get("recentDays"):
+    carried = dict(previous)
+    carried["tokensStale"] = True
+    return carried
+
+  # A partial week isn't a trustworthy model ledger; keep the last full one.
+  if any_fail and isinstance(previous.get("modelUsage"), dict):
+    model_usage = previous.get("modelUsage") or model_usage
+    if not today_by_model:
+      today_by_model = previous.get("todayTokensByModel") or {}
+
+  today_total = sum(today_by_model.values())
+  if not today_total and recent:
+    today_total = int(recent[-1].get("messageCount") or 0)
+
+  return {
+    "recentDays": recent,
+    "todayTotalTokens": int(today_total),
+    "todayTokensByModel": today_by_model,
+    "modelUsage": model_usage,
+    "tokenActiveDates": active_dates,
+    "tokensStale": any_fail,
+  }
+
+
+def cached_remote_token_stats(token: str, reuse_seconds: int) -> dict[str, Any]:
+  cache = cache_root() / f"x-tokens-{token_fingerprint(token)}.json"
+  today = date.today().isoformat()
+  cached = read_json(cache) if cache.exists() else None
+  cached = cached if isinstance(cached, dict) else {}
+  if reuse_seconds > 0 and cached.get("scanDate") == today and "recentDays" in cached:
+    age = time.time() - cache.stat().st_mtime
+    if age <= reuse_seconds:
+      return cached
+  stats = remote_token_stats(token, cached)
+  # Never replace a good cache with a total failure.
+  if stats.get("recentDays") or not cached.get("recentDays"):
+    stats["scanDate"] = today
+    write_json(cache, stats)
+    return stats
+  cached = dict(cached)
+  cached["tokensStale"] = True
+  return cached
+
+
+def prompt_day(obj: dict[str, Any]) -> date | None:
+  """Return the local calendar day of a transcript turn when the turn names it."""
+  chunks: list[str] = []
+  message = obj.get("message")
+  if isinstance(message, dict):
+    content = message.get("content")
+    if isinstance(content, str):
+      chunks.append(content)
+    elif isinstance(content, list):
+      for part in content:
+        if isinstance(part, dict) and isinstance(part.get("text"), str):
+          chunks.append(part["text"])
+        elif isinstance(part, str):
+          chunks.append(part)
+  for key in ("content", "text"):
+    value = obj.get(key)
+    if isinstance(value, str):
+      chunks.append(value)
+
+  for chunk in chunks:
+    match = TIMESTAMP_RE.search(chunk)
+    if not match:
+      continue
+    stamp = match.group(1)
+    zone = parse_tz_label(match.group(2) or "") or local_tzinfo()
+    for fmt in ("%A, %b %d, %Y, %I:%M %p", "%A, %B %d, %Y, %I:%M %p"):
+      try:
+        return to_local_date(datetime.strptime(stamp, fmt).replace(tzinfo=zone))
+      except ValueError:
+        continue
+
+  for key in ("createdAt", "timestamp", "time"):
+    raw = obj.get(key)
+    if raw is None:
+      continue
+    if isinstance(raw, (int, float)):
+      ms = float(raw)
+      if ms > 10_000_000_000:
+        ms /= 1000.0
+      try:
+        return to_local_date(datetime.fromtimestamp(ms, tz=timezone.utc))
+      except (OverflowError, OSError, ValueError):
+        continue
+    if isinstance(raw, str) and raw:
+      try:
+        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+      except ValueError:
+        continue
+      return to_local_date(parsed)
+  return None
+
+
+def local_stats() -> dict[str, Any]:
+  today = date.today()
+  by_day: dict[str, int] = defaultdict(int)
+  sessions: set[str] = set()
+  today_sessions: set[str] = set()
+  today_prompts = 0
+  total_prompts = 0
+  roots = [Path.home() / ".cursor" / "projects"]
+  files: list[Path] = []
+  for root in roots:
+    if root.exists():
+      files.extend(root.glob("**/agent-transcripts/**/*.jsonl"))
+
+  for path in files:
+    try:
+      session = path.stem
+      session_prompts = 0
+      with path.open(encoding="utf-8", errors="replace") as handle:
+        for line in handle:
+          line = line.strip()
+          if not line:
+            continue
+          try:
+            obj = json.loads(line)
+          except Exception:
+            continue
+          if not isinstance(obj, dict):
+            continue
+          role = obj.get("role") or obj.get("type")
+          if role not in ("user", "human"):
+            continue
+          session_prompts += 1
+          total_prompts += 1
+          day = prompt_day(obj)
+          if day is None:
+            # Undated turns still count toward totals, but must not land on
+            # today just because the file was touched later.
+            continue
+          by_day[day.isoformat()] += 1
+          if day == today:
+            today_prompts += 1
+            today_sessions.add(session)
+      if session_prompts:
+        sessions.add(session)
+    except Exception:
+      continue
+
+  recent = []
+  for i in range(6, -1, -1):
+    day = (today - timedelta(days=i)).isoformat()
+    # Token field; Cursor transcripts do not expose token ledgers here.
+    recent.append({"date": day, "messageCount": 0})
+
+  active_dates = sorted(by_day.keys())
+  stats = empty_stats()
+  stats.update(
+    {
+      "hasLocalStats": total_prompts > 0 or bool(sessions),
+      "todayPrompts": today_prompts,
+      "todaySessions": len(today_sessions),
+      "recentDays": recent,
+      "totalPrompts": total_prompts,
+      "totalSessions": len(sessions),
+      "activeDays": len(active_dates),
+      "activeDates": active_dates[-30:],
+    }
+  )
+  return stats
+
+
+def cached_local_stats(reuse_seconds: int) -> dict[str, Any]:
+  cache = cache_root() / "x-scan.json"
+  if reuse_seconds > 0 and cache.exists():
+    age = time.time() - cache.stat().st_mtime
+    if age <= reuse_seconds:
+      cached = read_json(cache)
+      if isinstance(cached, dict) and "totalPrompts" in cached:
+        return cached
+  stats = local_stats()
+  write_json(cache, stats)
+  return stats
+
+
+def plan_label(token: str) -> str:
+  try:
+    plan = rpc(token, "aiserver.v1.DashboardService/GetPlanInfo")
+  except Exception:
+    return ""
+  info = plan.get("planInfo") or {}
+  if not isinstance(info, dict):
+    return ""
+  name = str(info.get("planName") or "").strip()
+  price = str(info.get("price") or "").strip()
+  period = str(info.get("includedUsagePeriod") or "")
+  period_label = "monthly" if "MONTHLY" in period else ("weekly" if "WEEKLY" in period else "")
+  label = " · ".join(part for part in (name, price) if part)
+  if label and period_label:
+    label = f"{label} · shared {period_label} pool"
+  return label
+
+
+def usable_cached_limits(cached: dict[str, Any], token: str | None) -> list[dict[str, Any]]:
+  if token:
+    cached_fp = str(cached.get("tokenFingerprint") or "")
+    if cached_fp and cached_fp != token_fingerprint(token):
+      return []
+  elif cached.get("tokenFingerprint"):
+    # No sign-in: never surface another account's numbers as current.
+    return []
+
+  entries = cached.get("limits")
+  if not isinstance(entries, list):
+    return []
+  now = datetime.now(timezone.utc)
+  out = []
+  for entry in entries:
+    if not isinstance(entry, dict):
+      continue
+    raw = str(entry.get("resetsAt") or "")
+    if raw:
+      try:
+        resets_at = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+      except ValueError:
+        continue
+      if resets_at.tzinfo is None:
+        resets_at = resets_at.replace(tzinfo=timezone.utc)
+      if resets_at <= now:
+        continue
+    out.append(entry)
+  return out
+
+
+def probe_limits(token: str) -> dict[str, Any]:
+  result: dict[str, Any] = {
+    "limits": [],
+    "tierLabel": "",
+    "usageStatusText": "",
+    "authHelpText": AUTH_HELP,
+    "live": False,
+    "fetchedAtMs": 0,
+  }
+  result["tierLabel"] = plan_label(token) or AGENT_NAME
+  try:
+    usage = rpc(token, "aiserver.v1.DashboardService/GetCurrentPeriodUsage", {})
+  except urllib.error.HTTPError as error:
+    result["usageStatusText"] = "Waiting for auth" if error.code in (401, 403) else "X Subscription limits unavailable"
+    return result
+  except Exception:
+    result["usageStatusText"] = "X Subscription limits unavailable"
+    return result
+
+  plan_usage = usage.get("planUsage") or {}
+  if not isinstance(plan_usage, dict):
+    plan_usage = {}
+  resets_at = ms_to_iso(usage.get("billingCycleEnd"))
+  auto = pct_to_unit(plan_usage.get("autoPercentUsed"))
+  api = pct_to_unit(plan_usage.get("apiPercentUsed"))
+  total = pct_to_unit(plan_usage.get("totalPercentUsed"))
+  limits: list[dict[str, Any]] = []
+  if auto >= 0:
+    limits.append(
+      {
+        "label": "Included pool (Cursor + Grok)",
+        "percent": auto,
+        "resetsAt": resets_at,
+        "title": "First-party",
+      }
+    )
+  if api > 0:
+    limits.append(
+      {
+        "label": "API / named models",
+        "percent": api,
+        "resetsAt": resets_at,
+        "title": "API",
+      }
+    )
+  if total >= 0 and (auto < 0 or abs(total - auto) > 0.01):
+    limits.append(
+      {
+        "label": "Total included",
+        "percent": total,
+        "resetsAt": resets_at,
+        "title": "Total",
+      }
+    )
+
+  if not limits:
+    result["usageStatusText"] = "X Subscription limits unavailable"
+    return result
+
+  result.update(
+    limits=limits,
+    live=True,
+    fetchedAtMs=round(time.time() * 1000),
+    authHelpText="",
+    usageStatusText="",
+  )
+  return result
+
+
+def collect_limits(token: str | None, force: bool) -> dict[str, Any]:
+  cache = limits_cache_path(token)
+  raw_cached = read_json(cache)
+  cached = raw_cached if isinstance(raw_cached, dict) else {}
+  fallback = usable_cached_limits(cached, token)
+  result: dict[str, Any] = {
+    "limits": [],
+    "tierLabel": str(cached.get("tierLabel") or ""),
+    "usageStatusText": "",
+    "authHelpText": AUTH_HELP,
+    "live": False,
+    "fetchedAtMs": int(cached.get("fetchedAtMs") or 0),
+  }
+
+  if not token:
+    result["limits"] = fallback
+    result["usageStatusText"] = "Cursor auth missing" if not fallback else "Sign-in expired"
+    return result
+
+  fetched_at = int(cached.get("fetchedAtMs") or 0)
+  age_ms = round(time.time() * 1000) - fetched_at
+  if not force and fallback and fetched_at and age_ms <= PROBE_REUSE_SECONDS * 1000:
+    result.update(limits=fallback, live=True, authHelpText="", tierLabel=result["tierLabel"] or AGENT_NAME)
+    return result
+
+  probe = probe_limits(token)
+  if probe["live"]:
+    write_json(
+      cache,
+      {
+        "fetchedAtMs": probe["fetchedAtMs"],
+        "tierLabel": probe["tierLabel"],
+        "limits": probe["limits"],
+        "tokenFingerprint": token_fingerprint(token),
+      },
+    )
+    return probe
+
+  result["limits"] = fallback
+  result["usageStatusText"] = probe["usageStatusText"] or result["usageStatusText"]
+  result["authHelpText"] = probe["authHelpText"]
+  if fallback:
+    result["usageStatusText"] = result["usageStatusText"] or "Showing last known limits"
+  return result
+
+
+def main() -> None:
+  parser = argparse.ArgumentParser()
+  parser.add_argument("--force", action="store_true")
+  parser.add_argument("--limits-only", action="store_true")
+  args = parser.parse_args()
+
+  token = load_token()
+  limits = collect_limits(token, args.force)
+  reuse = LIMITS_ONLY_SCAN_REUSE_SECONDS if args.limits_only and not args.force else 0
+  stats = cached_local_stats(reuse)
+  tokens: dict[str, Any] = {}
+  if token:
+    tokens = cached_remote_token_stats(token, reuse)
+    stats["recentDays"] = tokens.get("recentDays") or stats.get("recentDays")
+    stats["todayTotalTokens"] = int(tokens.get("todayTotalTokens") or 0)
+    stats["todayTokensByModel"] = tokens.get("todayTokensByModel") or {}
+    stats["modelUsage"] = tokens.get("modelUsage") or {}
+    token_days = tokens.get("tokenActiveDates") or []
+    prompt_active_days = int(stats.get("activeDays") or 0)
+    prompt_dates = set(stats.get("activeDates") or [])
+    merged_dates = sorted(prompt_dates | set(token_days))
+    stats["activeDates"] = merged_dates[-30:]
+    # Keep the full prompt-day count; only the date list is capped at 30.
+    stats["activeDays"] = max(prompt_active_days, len(merged_dates))
+    if token_days or stats.get("todayTotalTokens"):
+      stats["hasLocalStats"] = True
+    if tokens.get("tokensStale") and not limits["usageStatusText"]:
+      limits["usageStatusText"] = "Showing last known token totals"
+  ready = bool(limits["limits"]) or bool(
+    stats.get("totalPrompts") or stats.get("totalSessions") or stats.get("todayTotalTokens")
+  )
+  record = {
+    "schemaVersion": 1,
+    "id": AGENT_ID,
+    "name": AGENT_NAME,
+    "updatedAt": datetime.now(timezone.utc).isoformat(),
+    "ready": ready,
+    # Dashboard token totals are account-global; synced machines must not sum them.
+    "scope": "account",
+    "tierLabel": limits["tierLabel"],
+    "usageStatusText": limits["usageStatusText"],
+    "authHelpText": limits["authHelpText"] if not limits["limits"] else "",
+    "limits": limits["limits"],
+    "limitsStale": not limits["live"] or bool(tokens.get("tokensStale") if token else False),
+    "limitsFetchedAt": limits["fetchedAtMs"],
+  }
+  record.update(stats)
+  print(json.dumps(record, separators=(",", ":")))
+
+
+if __name__ == "__main__":
+  main()
diff --git a/shell/plugins/agents/README.md b/shell/plugins/agents/README.md
index 26af2f72c6d..d97dae3a8d3 100644
--- a/shell/plugins/agents/README.md
+++ b/shell/plugins/agents/README.md
@@ -75,6 +75,7 @@ light surfaces — and the bar glyph stands in when there is none.
 | `codex` | The Codex app-server RPC | native Codex CLI session files on the built-in `openai` provider (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 |
 | `fireworks` | Estimated prepaid balance: configured funding minus rated account costs | Fireworks billing API, grouped by day and model for the last 30 days |
+| `x` | Shared first-party pool for The X Subscription (Cursor + Grok) via Cursor's private `DashboardService` RPCs (`GetCurrentPeriodUsage` / `GetPlanInfo` on `api2.cursor.sh`); tokens by day from `GetAggregatedUsageEvents` | Cursor agent transcripts under `~/.cursor/projects/**/agent-transcripts` |
 
 When `~/.local/state/omarchy/agents/accounts/<claude|codex|grok>.json`
 registers more than one account, the `claude`, `codex`, and `grok` records
@@ -105,6 +106,16 @@ pi stores in `~/.pi/agent/auth.json` when Fireworks is signed in there
 (honoring `PI_CODING_AGENT_DIR`, and pi's literal and `$ENV_VAR` key forms),
 and finally the key opencode stores in `~/.local/share/opencode/auth.json`.
 
+X Subscription reads the Cursor sign-in in `~/.config/cursor/auth.json`
+(the same token `cursor-agent login` writes). Its limits describe the shared
+included pool for Cursor and Grok on that plan; tokens by day come from
+`GetAggregatedUsageEvents`; local prompt counts come from Cursor agent
+transcripts. The product is branded under X, but the meter the
+Cursor UI itself uses for that shared pool still comes from Cursor's private
+dashboard RPCs — there is no documented public xAI usage API for the merged
+subscription yet, so this collector will break if those RPCs move. Machines
+with no Cursor sign-in still get an empty record the panel skips.
+
 ### Fireworks balance
 
 The collector first asks the account's `:getBalance` endpoint for the real
diff --git a/shell/plugins/agents/assets/x-light.svg b/shell/plugins/agents/assets/x-light.svg
new file mode 100644
index 00000000000..ace5bb685a4
--- /dev/null
+++ b/shell/plugins/agents/assets/x-light.svg
@@ -0,0 +1,3 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none">
+  <path d="M4.2 3.5h4.1L12 9.2l3.7-5.7h4.1l-5.8 8.4 6.2 8.9h-4.2L12 14.8l-3.9 5.5H3.9l6.2-8.9-5.9-8.4Z" fill="#0f1419"/>
+</svg>
diff --git a/shell/plugins/agents/assets/x.svg b/shell/plugins/agents/assets/x.svg
new file mode 100644
index 00000000000..435314cc49c
--- /dev/null
+++ b/shell/plugins/agents/assets/x.svg
@@ -0,0 +1,3 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none">
+  <path d="M4.2 3.5h4.1L12 9.2l3.7-5.7h4.1l-5.8 8.4 6.2 8.9h-4.2L12 14.8l-3.9 5.5H3.9l6.2-8.9-5.9-8.4Z" fill="white"/>
+</svg>
diff --git a/shell/plugins/agents/manifest.json b/shell/plugins/agents/manifest.json
index 7ca52484b12..cceef2041f2 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": "Claude Code, Codex, Grok, Fireworks, and X Subscription usage, limits, and pace in a native Omarchy bar panel.",
   "kinds": ["bar-widget"],
   "activation": "on-demand",
   "entryPoints": {
@@ -22,7 +22,8 @@
         "claude": { "enabled": true },
         "codex": { "enabled": true },
         "grok": { "enabled": true },
-        "fireworks": { "enabled": true }
+        "fireworks": { "enabled": true },
+        "x": { "enabled": true }
       },
       "refreshIntervalSec": 900,
       "syncMode": "Off",
diff --git a/test/shell.d/agent-usage-x-test.sh b/test/shell.d/agent-usage-x-test.sh
new file mode 100755
index 00000000000..230937b7d69
--- /dev/null
+++ b/test/shell.d/agent-usage-x-test.sh
@@ -0,0 +1,257 @@
+#!/bin/bash
+
+set -euo pipefail
+
+source "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)/base-test.sh"
+
+require_command jq
+require_command python3
+
+test_tmp=$(mktemp -d)
+trap 'rm -rf "$test_tmp"' EXIT
+
+export HOME="$test_tmp/home"
+export XDG_CONFIG_HOME="$test_tmp/config"
+export XDG_CACHE_HOME="$test_tmp/cache"
+export XDG_STATE_HOME="$test_tmp/state"
+
+mkdir -p "$HOME" "$XDG_CONFIG_HOME/cursor" "$XDG_CACHE_HOME" "$XDG_STATE_HOME"
+
+billing_end_ms=$(python3 -c 'import datetime as dt; print(int((dt.datetime.now(dt.timezone.utc) + dt.timedelta(days=12)).timestamp() * 1000))')
+
+signed_in() {
+  jq -nc --arg token "$1" '{accessToken: $token}' >"$XDG_CONFIG_HOME/cursor/auth.json"
+}
+
+# COLLECT_ARGS unset => --force. COLLECT_ARGS="" => no flags (reuse allowed).
+collect() {
+  COLLECTOR="$ROOT/bin/omarchy-agent-usage-x" BILLING_END_MS="$billing_end_ms" \
+    AUTO_PCT="${AUTO_PCT:-37.5}" python3 - "$@" <<'PY'
+import importlib.machinery, importlib.util, io, json, os, sys, urllib.error
+
+loader = importlib.machinery.SourceFileLoader("collector", os.environ["COLLECTOR"])
+spec = importlib.util.spec_from_loader(loader.name, loader)
+collector = importlib.util.module_from_spec(spec)
+loader.exec_module(collector)
+
+end = int(os.environ["BILLING_END_MS"])
+auto = float(os.environ["AUTO_PCT"])
+responses = {
+  "aiserver.v1.DashboardService/GetPlanInfo": {
+    "planInfo": {
+      "planName": "Pro",
+      "price": "$20/mo",
+      "includedUsagePeriod": "INCLUDED_USAGE_PERIOD_MONTHLY",
+    }
+  },
+  "aiserver.v1.DashboardService/GetCurrentPeriodUsage": {
+    "billingCycleEnd": end,
+    "planUsage": {
+      "autoPercentUsed": auto,
+      "apiPercentUsed": 0,
+      "totalPercentUsed": auto,
+    },
+  },
+}
+
+calls = {"n": 0}
+
+def urlopen(request, timeout=None):
+  calls["n"] += 1
+  if os.environ.get("X_FAIL_USAGE") == "1":
+    raise urllib.error.HTTPError(request.full_url, 401, "Unauthorized", hdrs=None, fp=None)
+  method = request.full_url.split("://", 1)[-1].split("/", 1)[-1]
+  if method.endswith("GetAggregatedUsageEvents"):
+    if os.environ.get("X_FAIL_AGG") == "1":
+      raise TimeoutError("dashboard day aggregate timed out")
+    # Optional day-token fixture for the tokens-by-day test.
+    rows = json.loads(os.environ.get("X_DAY_AGG", "[]"))
+    class Resp(io.BytesIO):
+      def __enter__(self): return self
+      def __exit__(self, *a): return False
+    return Resp(json.dumps({"aggregations": rows}).encode())
+  # Limits-reuse tests only reject plan/usage probes, not day-token scans.
+  if os.environ.get("X_REJECT_API") == "1":
+    raise AssertionError(f"unexpected API call: {request.full_url}")
+  if method not in responses:
+    raise AssertionError(f"unexpected method {method}")
+  auth = request.get_header("Authorization") or request.get_header("authorization")
+  expected = os.environ.get("X_TOKEN", "token-main")
+  assert auth.split(" ", 1)[1] == expected
+
+  class Resp(io.BytesIO):
+    def __enter__(self): return self
+    def __exit__(self, *a): return False
+
+  return Resp(json.dumps(responses[method]).encode())
+
+collector.urllib.request.urlopen = urlopen
+if "COLLECT_ARGS" in os.environ:
+  args = os.environ["COLLECT_ARGS"].split() if os.environ["COLLECT_ARGS"] else []
+else:
+  args = ["--force"]
+sys.argv = ["omarchy-agent-usage-x"] + args
+collector.main()
+print(f"\n__calls__={calls['n']}", file=sys.stderr)
+PY
+}
+
+record=$(collect)
+[[ $(jq -c '{ready, tierLabel, limits, usageStatusText}' <<<"$record") == '{"ready":false,"tierLabel":"","limits":[],"usageStatusText":"Cursor auth missing"}' ]] ||
+  fail "without Cursor auth the X Subscription record stays empty" "$record"
+pass "without Cursor auth the X Subscription record stays empty"
+
+signed_in token-main
+record=$(collect)
+[[ $(jq -c '{ready, id, name, tierLabel, stale: .limitsStale, label: .limits[0].label, percent: .limits[0].percent}' <<<"$record") == '{"ready":true,"id":"x","name":"X Subscription","tierLabel":"Pro · $20/mo · shared monthly pool","stale":false,"label":"Included pool (Cursor + Grok)","percent":0.375}' ]] ||
+  fail "X Subscription plan and shared pool come from Cursor dashboard APIs" "$record"
+[[ -n $(jq -r '.limits[0].resetsAt' <<<"$record") ]] || fail "the billing window says when the period ends" "$record"
+[[ $(jq -r '.authHelpText' <<<"$record") == "" ]] || fail "a successful probe drops the login hint" "$record"
+pass "X Subscription plan and shared pool come from Cursor dashboard APIs"
+
+# 0–100 API percents at and below 1 stay 1% / 0.5%, not a full bar.
+record=$(AUTO_PCT=1 COLLECT_ARGS="--force" collect)
+[[ $(jq -r '.limits[0].percent' <<<"$record") == "0.01" ]] ||
+  fail "1% used from the API is 0.01, not a full bar" "$record"
+record=$(AUTO_PCT=0.5 COLLECT_ARGS="--force" collect)
+[[ $(jq -r '.limits[0].percent' <<<"$record") == "0.005" ]] ||
+  fail "0.5% used from the API is 0.005" "$record"
+pass "small API percentages stay small"
+
+# Reused limits must not hit the network again.
+record=$(X_REJECT_API=1 COLLECT_ARGS="" collect 2>"$test_tmp/reuse.err")
+[[ $(jq -c '{percent: .limits[0].percent, help: .authHelpText, stale: .limitsStale}' <<<"$record") == '{"percent":0.005,"help":"","stale":false}' ]] ||
+  fail "reused X Subscription limits drop the login hint" "$record"
+grep -q 'unexpected API call' "$test_tmp/reuse.err" &&
+  fail "cache reuse still called the API" "$(cat "$test_tmp/reuse.err")"
+pass "reused X Subscription limits drop the login hint"
+
+# A failed probe keeps the last numbers for this login and says so.
+record=$(collect)
+record=$(X_FAIL_USAGE=1 COLLECT_ARGS="--force" collect)
+[[ $(jq -c '{usageStatusText, first: .limits[0].percent, stale: .limitsStale}' <<<"$record") == '{"usageStatusText":"Waiting for auth","first":0.375,"stale":true}' ]] ||
+  fail "a failed probe keeps the last shared-pool numbers" "$record"
+pass "a failed probe keeps the last shared-pool numbers"
+
+# A different Cursor login never inherits the previous account's cache.
+signed_in token-side
+record=$(X_TOKEN=token-side AUTO_PCT=12 COLLECT_ARGS="--force" collect)
+[[ $(jq -c '{percent: .limits[0].percent, stale: .limitsStale}' <<<"$record") == '{"percent":0.12,"stale":false}' ]] ||
+  fail "a new Cursor login probes its own shared pool" "$record"
+caches=$(find "$XDG_CACHE_HOME/omarchy/agent-usage" -name 'x-limits-*.json' | wc -l)
+((caches >= 2)) || fail "each Cursor login keeps its own limits cache" "caches=$caches"
+pass "a new Cursor login probes its own shared pool"
+
+# Local Cursor transcripts become prompt/session stats from each turn's stamp.
+projects="$HOME/.cursor/projects/demo/agent-transcripts"
+mkdir -p "$projects"
+today_stamp=$(python3 -c 'import datetime as dt; print(dt.date.today().strftime("%A, %b %-d, %Y"))')
+old_stamp="Monday, Jan 5, 2026"
+cat >"$projects/session-a.jsonl" <<JSONL
+{"role":"user","message":{"content":[{"type":"text","text":"<timestamp>${old_stamp}, 9:00 AM (UTC-5)</timestamp>\\n<user_query>old</user_query>"}]}}
+{"role":"assistant","content":"hi"}
+{"role":"user","message":{"content":[{"type":"text","text":"<timestamp>${today_stamp}, 10:00 AM (UTC-5)</timestamp>\\n<user_query>today</user_query>"}]}}
+{"role":"user","message":{"content":[{"type":"text","text":"<timestamp>${today_stamp}, 11:00 AM (UTC-5)</timestamp>\\n<user_query>again</user_query>"}]}}
+JSONL
+cat >"$projects/session-b.jsonl" <<JSONL
+{"type":"human","message":{"content":[{"type":"text","text":"<timestamp>${today_stamp}, 1:00 PM (UTC-5)</timestamp>\\n<user_query>other</user_query>"}]}}
+JSONL
+# Continuing an old file today must not move its undated/old prompts onto today.
+cat >"$projects/session-old.jsonl" <<'JSONL'
+{"role":"user","content":"no stamp here"}
+{"role":"assistant","content":"ok"}
+JSONL
+touch -d "2026-01-02 12:00:00" "$projects/session-old.jsonl"
+touch -d "$(date '+%Y-%m-%d %H:%M:%S')" "$projects/session-a.jsonl" "$projects/session-b.jsonl"
+rm -f "$XDG_CACHE_HOME"/omarchy/agent-usage/x-scan.json
+record=$(X_TOKEN=token-side COLLECT_ARGS="--force" collect)
+[[ $(jq -c '{hasLocalStats, todayPrompts, todaySessions, totalPrompts, totalSessions, weekTokens: ([.recentDays[].messageCount] | add)}' <<<"$record") == '{"hasLocalStats":true,"todayPrompts":3,"todaySessions":2,"totalPrompts":5,"totalSessions":3,"weekTokens":0}' ]] ||
+  fail "X Subscription counts dated prompts and keeps recentDays in tokens" "$record"
+pass "X Subscription counts dated prompts and keeps recentDays in tokens"
+
+# Empty / assistant-only transcripts do not invent activity.
+mkdir -p "$projects/empty"
+: >"$projects/empty/blank.jsonl"
+printf '%s\n' '{"role":"assistant","content":"only reply"}' >"$projects/empty/assistant-only.jsonl"
+rm -f "$XDG_CACHE_HOME"/omarchy/agent-usage/x-scan.json
+record=$(X_TOKEN=token-side COLLECT_ARGS="--force" collect)
+[[ $(jq -c '{totalPrompts, totalSessions, activeDays}' <<<"$record") == '{"totalPrompts":5,"totalSessions":3,"activeDays":2}' ]] ||
+  fail "empty transcripts do not invent prompts or active days" "$record"
+pass "empty transcripts do not invent prompts or active days"
+
+# Stamped offsets convert into the machine's local calendar day.
+python3 - "$ROOT/bin/omarchy-agent-usage-x" <<'PY'
+import importlib.machinery, importlib.util, os, sys
+loader = importlib.machinery.SourceFileLoader("collector", sys.argv[1])
+spec = importlib.util.spec_from_loader(loader.name, loader)
+collector = importlib.util.module_from_spec(spec)
+loader.exec_module(collector)
+obj = {
+  "role": "user",
+  "message": {
+    "content": [{
+      "type": "text",
+      "text": "<timestamp>Monday, Oct 12, 2026, 11:30 PM (UTC-5)</timestamp>\n<user_query>x</user_query>",
+    }]
+  },
+}
+os.environ["TZ"] = "UTC"
+# Reload local clock helpers after TZ change.
+import time
+time.tzset()
+day = collector.prompt_day(obj)
+assert str(day) == "2026-10-13", day
+os.environ["TZ"] = "America/Chicago"
+time.tzset()
+day = collector.prompt_day(obj)
+assert str(day) == "2026-10-12", day
+iso_day = collector.prompt_day({"role": "user", "createdAt": "2026-10-13T04:30:00Z"})
+assert str(iso_day) == "2026-10-12", iso_day
+print("ok")
+PY
+[[ $? -eq 0 ]] || fail "prompt stamps convert into the local calendar day"
+pass "prompt stamps convert into the local calendar day"
+
+# Dashboard day aggregates fill recentDays.messageCount with tokens.
+agg='[{"modelIntent":"grok-4.7-high-fast","inputTokens":100,"outputTokens":20,"cacheReadTokens":5,"cacheWriteTokens":0}]'
+rm -f "$XDG_CACHE_HOME"/omarchy/agent-usage/x-tokens-*.json
+record=$(X_TOKEN=token-side X_DAY_AGG="$agg" COLLECT_ARGS="--force" collect)
+[[ $(jq -c '{todayTotalTokens, weekTokens: ([.recentDays[].messageCount] | add), model: .todayTokensByModel["grok-4.7-high-fast"]}' <<<"$record") == '{"todayTotalTokens":125,"weekTokens":875,"model":125}' ]] ||
+  fail "dashboard day aggregates fill token counts for the week" "$record"
+[[ $(jq -r '.scope' <<<"$record") == "account" ]] ||
+  fail "dashboard token totals are account-scoped so sync does not double them" "$record"
+pass "dashboard day aggregates fill token counts for the week"
+
+# A timed-out day aggregate keeps the last known week totals.
+record=$(X_TOKEN=token-side X_FAIL_AGG=1 COLLECT_ARGS="--force" collect)
+[[ $(jq -c '{weekTokens: ([.recentDays[].messageCount] | add), today: .todayTotalTokens, stale: .limitsStale, status: .usageStatusText}' <<<"$record") == '{"weekTokens":875,"today":125,"stale":true,"status":"Showing last known token totals"}' ]] ||
+  fail "a failed day aggregate keeps the last known token totals" "$record"
+pass "a failed day aggregate keeps the last known token totals"
+
+# Merging token dates must not shrink a prompt activeDays count past 30.
+python3 - "$ROOT/bin/omarchy-agent-usage-x" <<'PY'
+import importlib.machinery, importlib.util, sys
+loader = importlib.machinery.SourceFileLoader("collector", sys.argv[1])
+spec = importlib.util.spec_from_loader(loader.name, loader)
+collector = importlib.util.module_from_spec(spec)
+loader.exec_module(collector)
+# Simulate main()'s merge: 80 prompt days truncated to 30 dates, plus token days.
+prompt_active_days = 80
+prompt_dates = {f"2026-01-{i:02d}" for i in range(1, 31)}
+token_days = ["2026-10-01", "2026-10-02"]
+merged_dates = sorted(prompt_dates | set(token_days))
+active_days = max(prompt_active_days, len(merged_dates))
+assert active_days == 80, active_days
+assert len(merged_dates[-30:]) == 30
+print("ok")
+PY
+[[ $? -eq 0 ]] || fail "activeDays keeps the full prompt-day count when merging token dates"
+pass "activeDays keeps the full prompt-day count when merging token dates"
+
+# limits-only reuses the transcript scan.
+
+mkdir -p "$projects/extra"
+printf '%s\n' "{\"role\":\"user\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"<timestamp>${today_stamp}, 2:00 PM (UTC-5)</timestamp>\\n<user_query>later</user_query>\"}]}}" >"$projects/extra/session-c.jsonl"
+record=$(X_TOKEN=token-side COLLECT_ARGS="--limits-only" collect)
+[[ $(jq -r '.totalSessions' <<<"$record") == 3 ]] || fail "a limits-only refresh reuses the transcript scan" "$record"
+pass "a limits-only refresh reuses the transcript scan"